dorfteich/apps/collab/src/version-store.ts

import { PAGE_VERSION_CREATED_CHANNEL, type PageVersionCreatedEvent } from '@dorfteich/shared';
import type { Pool } from 'pg';
import type { Logger } from 'pino';
import * as Y from 'yjs';

/**
 * Automatic version snapshots created by the collab server (issue #41, ADR
 * 0013). Contributors are tracked per open document: {@link recordContributor}
 * collects the users who edit, {@link onStore} flushes them to the shared
 * `page_pending_contributors` accumulator (so the api's named-version endpoint
 * sees them too) and creates the periodic interval snapshot, and
 * {@link onSessionEnd} creates the end-of-session snapshot.
 *
 * A snapshot is only written when there were edits since the previous version
 * (the accumulator is non-empty), which both implements "skip no-op periods"
 * and prevents duplicate versions on a quick reconnect with no changes.
 */
export interface VersionStore {
  /** Attribute a change to a user (called from `onChange`). */
  recordContributor(pageId: string, userId: string): void;
  /** Initialise interval bookkeeping when a document opens. */
  noteOpened(pageId: string): void;
  /** Flush contributors and create an interval snapshot if one is due. */
  onStore(pageId: string, doc: Y.Doc): Promise<void>;
  /** Create the end-of-session snapshot when the last participant leaves. */
  onSessionEnd(pageId: string, doc: Y.Doc): Promise<void>;
  /** Drop per-document bookkeeping when the document unloads. */
  forget(pageId: string): void;
}

export interface VersionStoreDeps {
  pool: Pool;
  logger: Logger;
  /** Active-editing interval between automatic snapshots (default 30 min). */
  intervalMs?: number;
}

const DEFAULT_INTERVAL_MS = 30 * 60 * 1000;

export class PostgresVersionStore implements VersionStore {
  private readonly intervalMs: number;
  /** Users who have edited since the last flush, per open document. */
  private readonly dirty = new Map<string, Set<string>>();
  /** Timestamp (ms) of the last automatic snapshot, per open document. */
  private readonly lastVersionAt = new Map<string, number>();

  constructor(private readonly deps: VersionStoreDeps) {
    this.intervalMs = deps.intervalMs ?? DEFAULT_INTERVAL_MS;
  }

  recordContributor(pageId: string, userId: string): void {
    let set = this.dirty.get(pageId);
    if (!set) {
      set = new Set();
      this.dirty.set(pageId, set);
    }
    set.add(userId);
  }

  noteOpened(pageId: string): void {
    this.lastVersionAt.set(pageId, Date.now());
  }

  async onStore(pageId: string, doc: Y.Doc): Promise<void> {
    await this.flushContributors(pageId);
    const last = this.lastVersionAt.get(pageId) ?? Date.now();
    if (Date.now() - last >= this.intervalMs) {
      const created = await this.createVersion(pageId, doc);
      // Reset the interval clock only when a snapshot was actually written, so
      // an idle-but-open page doesn't churn empty checks into versions.
      if (created) this.lastVersionAt.set(pageId, Date.now());
    }
  }

  async onSessionEnd(pageId: string, doc: Y.Doc): Promise<void> {
    await this.flushContributors(pageId);
    await this.createVersion(pageId, doc);
  }

  forget(pageId: string): void {
    this.dirty.delete(pageId);
    this.lastVersionAt.delete(pageId);
  }

  /** Move the in-memory contributor set into the shared DB accumulator. */
  private async flushContributors(pageId: string): Promise<void> {
    const set = this.dirty.get(pageId);
    if (!set || set.size === 0) return;
    // Swap in a fresh set first so contributors arriving during the await are
    // not lost with the ones being flushed.
    this.dirty.set(pageId, new Set());
    const users = [...set];
    try {
      await this.deps.pool.query(
        `INSERT INTO page_pending_contributors (page_id, user_id)
         SELECT $1, u FROM unnest($2::text[]) AS u
         ON CONFLICT (page_id, user_id) DO NOTHING`,
        [pageId, users],
      );
    } catch (error) {
      // Put the users back so the next flush retries them; never throw out of a
      // hook (a page purged mid-session is the expected benign failure here).
      const current = this.dirty.get(pageId) ?? new Set<string>();
      for (const u of users) current.add(u);
      this.dirty.set(pageId, current);
      this.deps.logger.warn(
        { event: 'version.contributors.flush_failed', pageId, err: (error as Error).message },
        'could not flush contributors',
      );
    }
  }

  /**
   * Write an automatic snapshot if there are pending contributors, consuming
   * them atomically. Returns whether a version row was created.
   */
  private async createVersion(pageId: string, doc: Y.Doc): Promise<boolean> {
    const snapshot = Buffer.from(Y.encodeStateAsUpdate(doc));
    const client = await this.deps.pool.connect();
    try {
      await client.query('BEGIN');
      const pending = await client.query<{ user_id: string }>(
        'SELECT user_id FROM page_pending_contributors WHERE page_id = $1 FOR UPDATE',
        [pageId],
      );
      if (pending.rows.length === 0) {
        await client.query('ROLLBACK');
        return false; // nothing changed since the last version
      }
      const contributors = pending.rows.map((row) => row.user_id);
      // Guard the FK: skip (but still clear the accumulator) if the page was
      // trashed/purged mid-session.
      const inserted = await client.query(
        `INSERT INTO page_versions (id, page_id, ydoc_snapshot, trigger, label, contributor_ids, created_at)
         SELECT gen_random_uuid(), $1, $2, 'AUTO', NULL, $3::text[], now()
         WHERE EXISTS (SELECT 1 FROM pages WHERE id = $1 AND deleted_at IS NULL)`,
        [pageId, snapshot, contributors],
      );
      await client.query('DELETE FROM page_pending_contributors WHERE page_id = $1', [pageId]);
      await client.query('COMMIT');
      const created = (inserted.rowCount ?? 0) > 0;
      if (created) {
        this.deps.logger.debug(
          { event: 'version.auto.created', pageId, contributors: contributors.length },
          'automatic version snapshot created',
        );
        // Tell the api so it can notify watchers (issue #94) — it owns the
        // permission resolution. Fire-and-forget: a lost event costs a
        // notification, never the snapshot.
        const event: PageVersionCreatedEvent = { pageId, contributorIds: contributors };
        await client
          .query('SELECT pg_notify($1, $2)', [PAGE_VERSION_CREATED_CHANNEL, JSON.stringify(event)])
          .catch(() => undefined);
      }
      return created;
    } catch (error) {
      await client.query('ROLLBACK').catch(() => undefined);
      this.deps.logger.warn(
        { event: 'version.auto.failed', pageId, err: (error as Error).message },
        'could not create automatic version snapshot',
      );
      return false;
    } finally {
      client.release();
    }
  }
}