dorfteich/apps/collab/src/persistence.db.test.ts

import { randomUUID } from 'node:crypto';

import { editorSchema } from '@dorfteich/shared';
import { Pool } from 'pg';
import { prosemirrorJSONToYXmlFragment } from 'y-prosemirror';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import * as Y from 'yjs';

import { PostgresPagePersistence } from './persistence.js';
import { deriveContentFromDoc } from './yjs-content.js';
import { collabTestDatabaseUrlOrUndefined } from './testing/test-db.js';

const url = collabTestDatabaseUrlOrUndefined;

/** A Y.Doc whose "default" XmlFragment holds a single paragraph of `text`. */
function makeDoc(text: string): Y.Doc {
  const doc = new Y.Doc();
  const paragraph = editorSchema.node('paragraph', null, text ? [editorSchema.text(text)] : []);
  const pmDoc = editorSchema.node('doc', null, [paragraph]);
  prosemirrorJSONToYXmlFragment(editorSchema, pmDoc.toJSON(), doc.getXmlFragment('default'));
  return doc;
}

/** A Y.Doc whose paragraph contains a `[[slug]]` wikilink per slug (issue #47). */
function makeDocWithWikilinks(...slugs: string[]): Y.Doc {
  const doc = new Y.Doc();
  const inline = [
    editorSchema.text('See '),
    ...slugs.map((slug) => editorSchema.node('wikilink', { targetSlug: slug, displayText: null })),
  ];
  const paragraph = editorSchema.node('paragraph', null, inline);
  const pmDoc = editorSchema.node('doc', null, [paragraph]);
  prosemirrorJSONToYXmlFragment(editorSchema, pmDoc.toJSON(), doc.getXmlFragment('default'));
  return doc;
}

/**
 * A minimal valid Yjs state with no document content. Seeding pages with an
 * empty state (rather than an empty paragraph) keeps these SQL round-trip tests
 * from merging two independent doc lineages — a test artifact, not something the
 * real lifecycle produces, where the editor always edits the loaded document.
 */
function emptyState(): Buffer {
  const doc = new Y.Doc();
  const update = Buffer.from(Y.encodeStateAsUpdate(doc));
  doc.destroy();
  return update;
}

describe.skipIf(!url)('PostgresPagePersistence (DB-backed)', () => {
  let pool: Pool;
  const userId = randomUUID();
  const pondId = randomUUID();
  const createdPageIds: string[] = [];

  async function createPage(): Promise<string> {
    const id = randomUUID();
    await pool.query(
      `INSERT INTO pages (id, pond_id, title, slug, ydoc_state, sort_key, created_by, updated_at)
       VALUES ($1, $2, 'Test', $3, $4, 'a0', $5, now())`,
      [id, pondId, `p-${id.slice(0, 8)}`, emptyState(), userId],
    );
    createdPageIds.push(id);
    return id;
  }

  beforeAll(async () => {
    pool = new Pool({ connectionString: url });
    await pool.query(
      'INSERT INTO users (id, username, email, display_name) VALUES ($1, $2, $3, $4)',
      [userId, `collab-${userId.slice(0, 8)}`, `${userId}@example.test`, 'Collab Tester'],
    );
    await pool.query(
      `INSERT INTO ponds (id, slug, name, type, owner_id, updated_at)
       VALUES ($1, $2, 'Collab Pond', 'PERSONAL', $3, now())`,
      [pondId, `collab-pond-${pondId.slice(0, 8)}`, userId],
    );
  });

  afterAll(async () => {
    if (createdPageIds.length > 0) {
      // page_updates and page_content_cache cascade on page delete.
      await pool.query('DELETE FROM pages WHERE id = ANY($1::text[])', [createdPageIds]);
    }
    await pool.query('DELETE FROM ponds WHERE id = $1', [pondId]);
    await pool.query('DELETE FROM users WHERE id = $1', [userId]);
    await pool.end();
  });

  it('stores a document and reloads identical content, refreshing the cache', async () => {
    const pageId = await createPage();
    const persistence = new PostgresPagePersistence(pool);

    const doc = makeDoc('Hello from the DB test');
    const result = await persistence.store(pageId, doc);
    expect(result.outcome).toBe('stored');

    const cache = await pool.query<{ markdown: string; plain_text: string }>(
      'SELECT markdown, plain_text FROM page_content_cache WHERE page_id = $1',
      [pageId],
    );
    expect(cache.rows[0]?.markdown).toContain('Hello from the DB test');
    expect(cache.rows[0]?.plain_text).toContain('Hello from the DB test');

    const reloaded = new Y.Doc();
    const loaded = await persistence.loadInto(pageId, reloaded);
    expect(loaded).toBe(true);
    expect(deriveContentFromDoc(reloaded).markdown).toBe(deriveContentFromDoc(doc).markdown);
  });

  it('reconstructs a document from a long update log (1000 entries)', async () => {
    const pageId = await createPage();

    // Build 1000 independent Yjs updates, each a single character inserted into
    // a scratch text type, then bulk-insert them as the page's update log.
    const master = new Y.Doc();
    let vector = Y.encodeStateVector(master);
    const seqs: number[] = [];
    const updates: Buffer[] = [];
    for (let i = 0; i < 1000; i += 1) {
      master.getText('log').insert(i, 'x');
      seqs.push(i);
      updates.push(Buffer.from(Y.encodeStateAsUpdate(master, vector)));
      vector = Y.encodeStateVector(master);
    }
    await pool.query(
      `INSERT INTO page_updates (id, page_id, seq, update)
       SELECT gen_random_uuid(), $1, s, u FROM unnest($2::int[], $3::bytea[]) AS t(s, u)`,
      [pageId, seqs, updates],
    );

    const persistence = new PostgresPagePersistence(pool);
    const reloaded = new Y.Doc();
    const loaded = await persistence.loadInto(pageId, reloaded);
    expect(loaded).toBe(true);
    expect(reloaded.getText('log').toString()).toBe('x'.repeat(1000));
  });

  it('rejects an oversize document without persisting anything', async () => {
    const pageId = await createPage();
    const persistence = new PostgresPagePersistence(pool);

    const doc = makeDoc('');
    doc.getText('bloat').insert(0, 'a'.repeat(6 * 1024 * 1024)); // > 5 MiB ceiling

    const result = await persistence.store(pageId, doc);
    expect(result.outcome).toBe('too_large');

    const updates = await pool.query('SELECT 1 FROM page_updates WHERE page_id = $1', [pageId]);
    expect(updates.rowCount).toBe(0);
    const cache = await pool.query('SELECT 1 FROM page_content_cache WHERE page_id = $1', [pageId]);
    expect(cache.rowCount).toBe(0);
  });

  it('reports not-found for a page that does not exist', async () => {
    const persistence = new PostgresPagePersistence(pool);
    const missingId = randomUUID();

    const stored = await persistence.store(missingId, makeDoc('orphan'));
    expect(stored.outcome).toBe('not_found');

    const loaded = await persistence.loadInto(missingId, new Y.Doc());
    expect(loaded).toBe(false);
  });

  it('maintains the page_links index on store — resolved and phantom (issue #47)', async () => {
    const target = await createPage();
    const targetSlug = `wl-target-${target.slice(0, 8)}`;
    await pool.query('UPDATE pages SET slug = $2 WHERE id = $1', [target, targetSlug]);
    const source = await createPage();
    const persistence = new PostgresPagePersistence(pool);

    // The doc links an existing page and a missing one.
    const doc = makeDocWithWikilinks(targetSlug, 'ghost-slug');
    const result = await persistence.store(source, doc);
    expect(result.outcome).toBe('stored');

    const rows = await pool.query<{ target_slug: string; to_page_id: string | null }>(
      'SELECT target_slug, to_page_id FROM page_links WHERE from_page_id = $1 ORDER BY target_slug',
      [source],
    );
    const bySlug = new Map(rows.rows.map((r) => [r.target_slug, r.to_page_id]));
    expect(bySlug.get(targetSlug)).toBe(target); // resolved
    expect(bySlug.has('ghost-slug')).toBe(true);
    expect(bySlug.get('ghost-slug')).toBeNull(); // phantom

    // Re-storing without the ghost link removes its row (index is rewritten).
    await persistence.store(source, makeDocWithWikilinks(targetSlug));
    const after = await pool.query('SELECT target_slug FROM page_links WHERE from_page_id = $1', [
      source,
    ]);
    expect(after.rows.map((r) => r.target_slug)).toEqual([targetSlug]);
  });
});