dorfteich/apps/api/src/compaction/compaction.service.db.test.ts

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<string> {
  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<string> {
    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<ArrayBuffer>, 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<ArrayBuffer> }[] = [];
    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<number> {
    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);
  });
});