import { randomUUID } from 'node:crypto'; import { INestApplication } from '@nestjs/common'; import { PrismaClient } from '@prisma/client'; import { afterAll, afterEach, beforeAll, describe, expect, it } from 'vitest'; import * as Y from 'yjs'; import { createTestApp } from '../testing/test-app'; import { createTestPrisma, hasTestDb, uniqueSuffix } from '../testing/test-db'; import { COMPACTION_LOG_THRESHOLD, CompactionService } from './compaction.service'; /** Reconstruct a page's document text from its stored state plus update log. */ async function reconstructText(prisma: PrismaClient, pageId: string): Promise { const page = await prisma.page.findUniqueOrThrow({ where: { id: pageId } }); const updates = await prisma.pageUpdate.findMany({ where: { pageId }, orderBy: { seq: 'asc' }, }); const doc = new Y.Doc(); Y.applyUpdate(doc, new Uint8Array(page.ydocState)); for (const row of updates) Y.applyUpdate(doc, new Uint8Array(row.update)); const text = doc.getText('log').toString(); doc.destroy(); return text; } describe.skipIf(!hasTestDb)('CompactionService (db, issue #40)', () => { let app: INestApplication; let prisma: PrismaClient; let compaction: CompactionService; const suffix = uniqueSuffix(); let userId: string; let pondId: string; const pageIds: string[] = []; beforeAll(async () => { prisma = createTestPrisma(); app = await createTestApp(); compaction = app.get(CompactionService); const user = await prisma.user.create({ data: { username: `compaction-${suffix}`, email: `compaction-${suffix}@example.test`, displayName: 'Compaction Tester', }, }); userId = user.id; const pond = await prisma.pond.create({ data: { slug: `compaction-pond-${suffix}`, name: 'Compaction Pond', type: 'PERSONAL', ownerId: userId, }, }); pondId = pond.id; }); afterEach(async () => { await prisma.collabOpenSession.deleteMany({ where: { pageId: { in: pageIds } } }); if (pageIds.length > 0) { await prisma.page.deleteMany({ where: { id: { in: pageIds } } }); pageIds.length = 0; } }); afterAll(async () => { await prisma.pond.deleteMany({ where: { id: pondId } }); await prisma.user.deleteMany({ where: { id: userId } }); await prisma.$disconnect(); await app.close(); }); /** * Create a page whose update log has `logRows` entries, each a single * character appended to a scratch text type — a well-formed incremental log * over an empty base state, like a real editing session produces. */ async function createPageWithLog(logRows: number): Promise { const id = randomUUID(); await prisma.page.create({ data: { id, pondId, title: 'Test', slug: `p-${id.slice(0, 8)}`, // A fresh Uint8Array copy, not a Buffer: Prisma's Bytes input type is // Uint8Array, which a Buffer (ArrayBufferLike) does not // satisfy under strict types (handoff gotcha). ydocState: new Uint8Array(Y.encodeStateAsUpdate(new Y.Doc())), sortKey: 'a0', createdBy: userId, }, }); const master = new Y.Doc(); let vector = Y.encodeStateVector(master); const rows: { id: string; pageId: string; seq: number; update: Uint8Array }[] = []; for (let i = 0; i < logRows; i += 1) { master.getText('log').insert(i, 'x'); rows.push({ id: randomUUID(), pageId: id, seq: i, update: new Uint8Array(Y.encodeStateAsUpdate(master, vector)), }); vector = Y.encodeStateVector(master); } master.destroy(); await prisma.pageUpdate.createMany({ data: rows }); pageIds.push(id); return id; } async function logCount(pageId: string): Promise { return prisma.pageUpdate.count({ where: { pageId } }); } it('merges the log into ydoc_state without changing content (hash comparison)', async () => { const pageId = await createPageWithLog(COMPACTION_LOG_THRESHOLD + 10); const before = await reconstructText(prisma, pageId); const summary = await compaction.compactDuePages(); expect(summary.pagesCompacted).toBeGreaterThanOrEqual(1); expect(await logCount(pageId)).toBe(0); // Content is identical after compaction (reconstructed from ydoc_state alone). expect(await reconstructText(prisma, pageId)).toBe(before); expect(before).toBe('x'.repeat(COMPACTION_LOG_THRESHOLD + 10)); }); it('leaves a page below the threshold untouched', async () => { const pageId = await createPageWithLog(3); await compaction.compactDuePages(); expect(await logCount(pageId)).toBe(3); }); it('skips a page with an active session and compacts it on the next run', async () => { const pageId = await createPageWithLog(COMPACTION_LOG_THRESHOLD + 5); // A fresh heartbeat marks the page as being edited. await prisma.collabOpenSession.create({ data: { pageId, heartbeatAt: new Date() }, }); await compaction.compactDuePages(); expect(await logCount(pageId)).toBe(COMPACTION_LOG_THRESHOLD + 5); // skipped // Session ends: the row is removed (or ages out), and the next run compacts. await prisma.collabOpenSession.deleteMany({ where: { pageId } }); await compaction.compactDuePages(); expect(await logCount(pageId)).toBe(0); }); it('ignores a stale session heartbeat (crashed collab does not block forever)', async () => { const pageId = await createPageWithLog(COMPACTION_LOG_THRESHOLD + 5); // A row left behind by a crashed collab process: heartbeat far in the past. await prisma.collabOpenSession.create({ data: { pageId, heartbeatAt: new Date(Date.now() - 60 * 60 * 1000) }, }); await compaction.compactDuePages(); expect(await logCount(pageId)).toBe(0); }); it('is idempotent: a second run leaves the compacted page alone', async () => { const pageId = await createPageWithLog(COMPACTION_LOG_THRESHOLD + 7); const first = await compaction.compactDuePages(); expect(first.pagesCompacted).toBeGreaterThanOrEqual(1); const afterFirst = await reconstructText(prisma, pageId); // A second run must neither error nor touch the already-compacted page // (the summary itself is DB-global, so we assert on this page specifically). await compaction.compactDuePages(); expect(await logCount(pageId)).toBe(0); expect(await reconstructText(prisma, pageId)).toBe(afterFirst); }); });