STTracker/packages/core/src/sync/run.ts

import type { Db } from "@sttracker/db/db";
import { newId, nowIso } from "@sttracker/db/ids";
import { fromJson, toJson } from "@sttracker/db/json";
import type { ConnectionStatus, Database, RunMode } from "@sttracker/db/schema";
import type { PluginContext, SourcePlugin, SourceRecord } from "@sttracker/plugin-sdk/types";
import { NotFoundError, ValidationError } from "../errors.ts";
import type { Actor } from "../history.ts";
import { applyRecord, committed, emptyCounts, markRemoved, type RunCounts, type SyncState } from "./apply.ts";
import { ingestListens } from "./listens.ts";

/** A `source_connections` row with JSON parsed. Credentials stay encrypted; the caller's `makeContext` decrypts. */
export interface Connection {
  id: string;
  userId: string;
  pluginId: string;
  name: string;
  config: Record<string, unknown>;
  credentialsEnc: string | null;
  scheduleMinutes: number | null;
  cursor: string | null;
  status: ConnectionStatus;
  lastRunAt: string | null;
  createdAt: string;
}

export interface SyncDeps {
  db: Db;
  /** Builds the plugin context (http client, credentials, cursor, logging). Injected by the caller. */
  makeContext: (conn: Connection, runId: string) => PluginContext;
}

export type SyncMode = "full" | "changes" | { file: { name: string; bytes: Uint8Array } };

export interface RunSummary {
  runId: string;
  status: "ok" | "failed";
  counts: RunCounts;
  error?: string;
}

const LISTEN_BATCH = 500;

function toConnection(r: Database["source_connections"]): Connection {
  return {
    id: r.id,
    userId: r.user_id,
    pluginId: r.plugin_id,
    name: r.name,
    config: fromJson<Record<string, unknown>>(r.config, {}),
    credentialsEnc: r.credentials_enc,
    scheduleMinutes: r.schedule_minutes,
    cursor: r.cursor,
    status: r.status,
    lastRunAt: r.last_run_at,
    createdAt: r.created_at,
  };
}

function iterate(plugin: SourcePlugin, ctx: PluginContext, mode: SyncMode): AsyncIterable<SourceRecord> {
  const id = plugin.manifest.id;
  if (mode === "full") {
    if (!plugin.pull) throw new Error(`plugin "${id}" cannot pull`);
    return plugin.pull(ctx);
  }
  if (mode === "changes") {
    if (!plugin.pullChanges) throw new Error(`plugin "${id}" cannot pull changes`);
    return plugin.pullChanges(ctx);
  }
  if (!plugin.importFile) throw new Error(`plugin "${id}" cannot import files`);
  return plugin.importFile(ctx, mode.file);
}

const message = (e: unknown) => (e instanceof Error ? e.message : String(e));

/**
 * One sync run (spec §8.3): pull records, snapshot and diff each against its previous version, apply
 * or file change events, and append listens. Absent records are marked removed only after a `full`
 * pull whose iteration finished and whose records all applied. Any failure marks the run `failed` and
 * the connection `error`; the summary is returned, not thrown.
 */
export async function runSync(deps: SyncDeps, plugin: SourcePlugin, connectionId: string, mode: SyncMode): Promise<RunSummary> {
  const { db } = deps;
  const row = await db.k.selectFrom("source_connections").selectAll().where("id", "=", connectionId).executeTakeFirst();
  if (!row) throw new NotFoundError("connection");
  if (row.plugin_id !== plugin.manifest.id) throw new ValidationError("plugin", "connection_plugin_mismatch");
  const conn = toConnection(row);
  const runId = newId();
  const runMode: RunMode = typeof mode === "string" ? mode : "file";
  await db.k
    .insertInto("source_runs")
    .values({ id: runId, connection_id: conn.id, mode: runMode, started_at: nowIso(), finished_at: null, status: "running", counts: toJson(emptyCounts()), log: "", error: null })
    .execute();

  const actor: Actor = `plugin:${conn.pluginId}:${conn.id}`;
  const st: SyncState = { db, userId: conn.userId, connectionId: conn.id, pluginId: conn.pluginId, runId, actor, counts: emptyCounts(), log: [] };
  const seen = new Set<string>();
  const errors: string[] = [];
  let pullError: string | null = null;
  let listens: SourceRecord[] = [];
  const flushListens = async () => {
    if (listens.length === 0) return;
    const batch = listens;
    listens = [];
    try {
      st.counts.listens += (await ingestListens(db, conn.userId, conn.pluginId, batch, actor)).inserted;
    } catch (e) {
      errors.push(`listens: ${message(e)}`);
    }
  };

  try {
    const ctx = deps.makeContext(conn, runId);
    for await (const rec of iterate(plugin, ctx, mode)) {
      st.counts.seen++;
      if (rec.kind === "listen") {
        listens.push(rec);
        if (listens.length >= LISTEN_BATCH) await flushListens();
        continue;
      }
      seen.add(rec.key);
      try {
        await applyRecord(st, rec);
      } catch (e) {
        errors.push(`record ${rec.key}: ${message(e)}`);
      }
    }
  } catch (e) {
    pullError = message(e);
  }
  await flushListens();

  // Removal by absence: only a completed full pull with every record applied can prove a record is gone.
  if (mode === "full" && pullError === null && errors.length === 0) {
    try {
      const live = await db.k
        .selectFrom("source_records")
        .selectAll()
        .where("connection_id", "=", conn.id)
        .where("deleted_at", "is", null)
        .execute();
      for (const r of live) {
        if (seen.has(r.key)) continue;
        await committed(st, (local) => db.k.transaction().execute((tx) => markRemoved(tx, local, r)));
      }
    } catch (e) {
      errors.push(`removal: ${message(e)}`);
    }
  }

  const failed = pullError !== null || errors.length > 0;
  const error = pullError ?? (errors.length > 0 ? `${errors.length} error(s); first: ${errors[0]}` : undefined);
  const log = [...st.log, ...errors];
  if (pullError !== null) log.push(`pull failed: ${pullError}`);
  const now = nowIso();
  await db.k
    .updateTable("source_runs")
    .set({ finished_at: now, status: failed ? "failed" : "ok", counts: toJson(st.counts), log: log.join("\n"), error: error ?? null })
    .where("id", "=", runId)
    .execute();
  const nextStatus: ConnectionStatus = failed ? "error" : conn.status === "error" ? "active" : conn.status;
  await db.k.updateTable("source_connections").set({ last_run_at: now, status: nextStatus }).where("id", "=", conn.id).execute();

  return { runId, status: failed ? "failed" : "ok", counts: st.counts, ...(error !== undefined ? { error } : {}) };
}