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

import type { Db } from "@sttracker/db/db";
import { nowIso } from "@sttracker/db/ids";
import { fromJson, toJson } from "@sttracker/db/json";
import type { ChangeStatus, ChangeType, Database } from "@sttracker/db/schema";
import type { WorkHint } from "@sttracker/plugin-sdk/types";
import { type EngagementPatch, softDeleteEngagement, updateEngagement } from "../engagements.ts";
import { type EntryPatch, moveEntryToWork, softDeleteEntry, upsertEntry } from "../entries.ts";
import { NotFoundError, ValidationError } from "../errors.ts";
import type { Actor, Executor } from "../history.ts";
import { softDeleteList } from "../lists.ts";
import { matchWork } from "../matching.ts";
import { entryTitles } from "../titles.ts";
import { resolveAliasIn } from "../works.ts";
import { applyPatch, engagementValues, entryValues } from "./apply.ts";
import { decodeVersion } from "./snapshot.ts";

export interface ChangeEvent {
  id: string;
  type: ChangeType;
  status: ChangeStatus;
  connectionId: string | null;
  runId: string | null;
  recordId: string | null;
  targetEntity: string;
  targetId: string | null;
  payload: Record<string, unknown>;
  createdAt: string;
  resolvedAt: string | null;
}

/** An inbox item: the event plus the titles the Changes screen shows. */
export interface ChangeItem extends ChangeEvent {
  /** The target's title: an entry's (override-aware), an engagement's entry's, a list's name. Null if gone. */
  title: string | null;
  /** Merged events only: the title of the entry the merge target record maps to, else null. */
  intoTitle?: string | null;
}

function toChange(r: Database["change_events"]): ChangeEvent {
  return {
    id: r.id,
    type: r.type,
    status: r.status,
    connectionId: r.connection_id,
    runId: r.run_id,
    recordId: r.record_id,
    targetEntity: r.target_entity,
    targetId: r.target_id,
    payload: fromJson<Record<string, unknown>>(r.payload, {}),
    createdAt: r.created_at,
    resolvedAt: r.resolved_at,
  };
}

/**
 * The user's change events, newest first. `status` defaults to "pending" (the inbox); "all" includes
 * auto-applied and resolved ones. `cursor` is the `nextCursor` of the previous page.
 */
export async function listChanges(
  db: Db,
  userId: string,
  opts: { status?: "pending" | "all"; connectionId?: string; limit?: number; cursor?: string } = {},
): Promise<{ items: ChangeItem[]; nextCursor: string | null }> {
  const limit = Math.min(Math.max(opts.limit ?? 50, 1), 200);
  let q = db.k.selectFrom("change_events").selectAll().where("user_id", "=", userId);
  if ((opts.status ?? "pending") === "pending") q = q.where("status", "=", "pending");
  if (opts.connectionId) q = q.where("connection_id", "=", opts.connectionId);
  if (opts.cursor) q = q.where("id", "<", opts.cursor);
  const rows = await q.orderBy("id", "desc").limit(limit + 1).execute();
  const events = rows.slice(0, limit).map(toChange);
  return { items: await withTitles(db.k, userId, events), nextCursor: rows.length > limit ? events[events.length - 1]!.id : null };
}

/** Attach titles with a fixed number of batched queries, whatever the page size. */
async function withTitles(k: Executor, userId: string, events: ChangeEvent[]): Promise<ChangeItem[]> {
  const ids = (entity: string) => [...new Set(events.filter((e) => e.targetEntity === entity && e.targetId).map((e) => e.targetId!))];

  // Merged events: the record their intoKey names (same connection) → its target.
  const merges = events.filter((e) => e.type === "merged" && e.connectionId && typeof e.payload.intoKey === "string");
  const intoRecords = merges.length
    ? await k
        .selectFrom("source_records")
        .select(["connection_id", "key", "target_entity", "target_id"])
        .where("connection_id", "in", [...new Set(merges.map((e) => e.connectionId!))])
        .where("key", "in", [...new Set(merges.map((e) => String(e.payload.intoKey)))])
        .execute()
    : [];
  const intoTarget = new Map(intoRecords.map((r) => [`${r.connection_id}\u0000${r.key}`, r]));

  const engagementIds = [...ids("engagement"), ...intoRecords.filter((r) => r.target_entity === "engagement").map((r) => r.target_id!)];
  const engagementEntry = new Map<string, string>();
  if (engagementIds.length) {
    const rows = await k.selectFrom("engagements").select(["id", "entry_id"]).where("id", "in", [...new Set(engagementIds)]).execute();
    for (const r of rows) engagementEntry.set(r.id, r.entry_id);
  }
  const entryOf = (entity: string | null, id: string | null): string | null =>
    !id ? null : entity === "entry" ? id : entity === "engagement" ? (engagementEntry.get(id) ?? null) : null;

  const entryIds = [
    ...ids("entry"),
    ...engagementEntry.values(),
    ...intoRecords.map((r) => entryOf(r.target_entity, r.target_id)),
  ];
  const entryTitle = await entryTitles(k, userId, entryIds);
  const listIds = ids("list");
  const listName = new Map<string, string>();
  if (listIds.length) {
    const rows = await k.selectFrom("lists").select(["id", "name"]).where("user_id", "=", userId).where("id", "in", listIds).execute();
    for (const r of rows) listName.set(r.id, r.name);
  }

  return events.map((e) => {
    const title =
      e.targetEntity === "list" ? (listName.get(e.targetId ?? "") ?? null) : (entryTitle.get(entryOf(e.targetEntity, e.targetId) ?? "") ?? null);
    if (e.type !== "merged") return { ...e, title };
    const into = intoTarget.get(`${e.connectionId}\u0000${String(e.payload.intoKey)}`);
    const intoEntry = into ? entryOf(into.target_entity, into.target_id) : null;
    return { ...e, title, intoTitle: (intoEntry && entryTitle.get(intoEntry)) || null };
  });
}

const STATUS_FOR = { accept: "accepted", ignore: "ignored", keep_mine: "kept_mine" } as const;

/**
 * Resolve a pending event.
 * - `accept`: changed/conflict applies the incoming value as the event's plugin actor (which then owns
 *   the field); removed soft-deletes the target (an engagement other sources still report only loses
 *   this connection's `engagement_sources` row); merged moves the entry to the merge target's work.
 * - `keep_mine`: changed/conflict set `field_sources[field] = "user"`, so that source stops overriding it.
 * - `ignore`: data untouched.
 * All three set `status` and `resolved_at`. Accept/keep_mine on a target that is gone (deleted, or merged
 * away with no live survivor) resolves the event as `ignored` with `payload.reason = "target_gone"`.
 */
export async function resolveChange(
  db: Db,
  userId: string,
  changeId: string,
  action: "accept" | "ignore" | "keep_mine",
): Promise<void> {
  if (!Object.hasOwn(STATUS_FOR, action)) throw new ValidationError("action", "invalid");
  const row = await db.k.selectFrom("change_events").selectAll().where("id", "=", changeId).where("user_id", "=", userId).executeTakeFirst();
  if (!row) throw new NotFoundError("change");
  if (row.status !== "pending") throw new ValidationError("status", "not_pending");
  const ev = toChange(row);
  const actor = await actorFor(db, ev.connectionId);

  // matchWork runs its own transaction; resolve the merge target before opening ours.
  const mergeTarget = action === "accept" && ev.type === "merged" ? await mergeTargetWork(db, userId, ev, actor) : null;

  await db.k.transaction().execute(async (tx) => {
    // Claim the event first so a concurrent resolve of the same event becomes a no-op failure.
    const claimed = await tx
      .updateTable("change_events")
      .set({ status: STATUS_FOR[action], resolved_at: nowIso() })
      .where("id", "=", ev.id)
      .where("status", "=", "pending")
      .executeTakeFirst();
    if (Number(claimed.numUpdatedRows) === 0) throw new ValidationError("status", "not_pending");
    if (action === "ignore") return;
    const isField = ev.type === "changed" || ev.type === "conflict";
    if (action === "keep_mine" && !isField) return;

    const targetId = await liveTargetId(tx, userId, ev);
    if (!targetId) {
      await tx
        .updateTable("change_events")
        .set({ status: "ignored", payload: toJson({ ...ev.payload, reason: "target_gone" }) })
        .where("id", "=", ev.id)
        .execute();
      return;
    }
    if (isField) {
      const field = String(ev.payload.field);
      const value = action === "keep_mine" ? undefined : ev.type === "conflict" ? ev.payload.incoming : ev.payload.to;
      await setField(db, tx, userId, ev.targetEntity, targetId, field, value, action === "keep_mine" ? "user" : actor);
      return;
    }
    if (ev.type === "removed") return removeTarget(db, tx, userId, ev, targetId, actor);
    if (ev.type === "merged" && mergeTarget) return moveTo(db, tx, userId, targetId, mergeTarget, actor);
  });
}

async function actorFor(db: Db, connectionId: string | null): Promise<Actor> {
  if (!connectionId) return "user";
  const c = await db.k.selectFrom("source_connections").select(["id", "plugin_id"]).where("id", "=", connectionId).executeTakeFirst();
  return c ? `plugin:${c.plugin_id}:${c.id}` : "user";
}

/**
 * The event's target if it is still live. A soft-deleted entry that was merged into another (the user's
 * work alias leads to a work with a live entry) resolves to that survivor. Otherwise null.
 */
async function liveTargetId(k: Executor, userId: string, ev: ChangeEvent): Promise<string | null> {
  if (!ev.targetId) return null;
  if (ev.targetEntity === "entry") {
    const r = await k.selectFrom("entries").select(["id", "work_id", "deleted_at"]).where("id", "=", ev.targetId).where("user_id", "=", userId).executeTakeFirst();
    if (!r) return null;
    if (r.deleted_at === null) return r.id;
    const to = await resolveAliasIn(k, userId, r.work_id);
    if (to === r.work_id) return null;
    const survivor = await k
      .selectFrom("entries")
      .select("id")
      .where("user_id", "=", userId)
      .where("work_id", "=", to)
      .where("deleted_at", "is", null)
      .executeTakeFirst();
    return survivor?.id ?? null;
  }
  if (ev.targetEntity === "engagement") {
    const r = await k
      .selectFrom("engagements as g")
      .innerJoin("entries as e", "e.id", "g.entry_id")
      .select("g.id")
      .where("g.id", "=", ev.targetId)
      .where("e.user_id", "=", userId)
      .where("g.deleted_at", "is", null)
      .where("e.deleted_at", "is", null)
      .executeTakeFirst();
    return r?.id ?? null;
  }
  if (ev.targetEntity === "list") {
    const r = await k
      .selectFrom("lists")
      .select("id")
      .where("id", "=", ev.targetId)
      .where("user_id", "=", userId)
      .where("deleted_at", "is", null)
      .executeTakeFirst();
    return r?.id ?? null;
  }
  return null;
}

/** Write `value` to a live target's field as `actor`; with `value` undefined, re-stamp the current value. */
async function setField(db: Db, k: Executor, userId: string, entity: string, id: string, field: string, value: unknown, actor: Actor): Promise<void> {
  if (entity === "entry") {
    const r = await k.selectFrom("entries").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
    const v = value === undefined ? entryValues(r)[field] : value;
    await upsertEntry(db, userId, r.work_id, { [field]: v } as Partial<EntryPatch>, actor, k);
  } else if (entity === "engagement") {
    const r = await k.selectFrom("engagements").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
    const v = value === undefined ? engagementValues(r)[field] : value;
    await updateEngagement(db, userId, id, { [field]: v } as Partial<EngagementPatch>, actor, k);
  } else if (entity === "list") {
    // Lists have no field_sources: ownership comes from history, so keep_mine needs no write.
    if (value === undefined) return;
    await applyPatch(k, db, userId, { entity: "list", id }, { [field]: value }, actor);
  } else {
    throw new ValidationError("targetEntity", "unsupported");
  }
}

async function removeTarget(db: Db, k: Executor, userId: string, ev: ChangeEvent, targetId: string, actor: Actor): Promise<void> {
  if (ev.targetEntity === "entry") return softDeleteEntry(db, userId, targetId, actor, k);
  if (ev.targetEntity === "list") return softDeleteList(db, userId, targetId, actor, k);
  if (ev.targetEntity === "engagement") {
    // The same play may still be reported by other sources: detach this one, delete only when none remain.
    if (ev.connectionId) {
      await k.deleteFrom("engagement_sources").where("engagement_id", "=", targetId).where("connection_id", "=", ev.connectionId).execute();
    }
    const left = await k.selectFrom("engagement_sources").select("id").where("engagement_id", "=", targetId).executeTakeFirst();
    if (left) return;
    await softDeleteEngagement(db, userId, targetId, actor, k);
  }
}

/**
 * The work a merged event moves its entry to. An admin merge carries `toWorkId`. A source merge names
 * the target record (`intoKey`): its mapped entry's work when it has one, else the work matched for
 * the target record's latest hint.
 */
async function mergeTargetWork(db: Db, userId: string, ev: ChangeEvent, actor: Actor): Promise<string> {
  if (typeof ev.payload.toWorkId === "string") return ev.payload.toWorkId;
  const intoKey = ev.payload.intoKey;
  if (!ev.connectionId || typeof intoKey !== "string") throw new ValidationError("payload", "no_merge_target");
  const target = await db.k
    .selectFrom("source_records")
    .selectAll()
    .where("connection_id", "=", ev.connectionId)
    .where("key", "=", intoKey)
    .executeTakeFirst();
  if (!target) throw new ValidationError("intoKey", "unknown_record");
  // A record matched but not yet applied stores its work directly.
  if (target.target_entity === "work" && target.target_id) return target.target_id;
  if (target.target_id && (target.target_entity === "entry" || target.target_entity === "engagement")) {
    const entryId =
      target.target_entity === "entry"
        ? target.target_id
        : (await db.k.selectFrom("engagements").select("entry_id").where("id", "=", target.target_id).executeTakeFirst())?.entry_id;
    const entry = entryId
      ? await db.k.selectFrom("entries").select("work_id").where("id", "=", entryId).where("user_id", "=", userId).executeTakeFirst()
      : undefined;
    if (entry) return entry.work_id;
  }
  const version = await db.k
    .selectFrom("source_record_versions")
    .select("raw_gz")
    .where("record_id", "=", target.id)
    .where("hash", "=", target.current_hash)
    .orderBy("seen_at", "desc")
    .orderBy("id", "desc")
    .executeTakeFirst();
  const hint = version ? (decodeVersion(version.raw_gz).normalized.work as WorkHint | undefined) : undefined;
  if (!hint) throw new ValidationError("intoKey", "no_work_hint");
  return (await matchWork(db, userId, hint, actor)).workId;
}

/** Move the live entry to the work; `moveEntryToWork` re-points source records if it merges. */
async function moveTo(db: Db, k: Executor, userId: string, entryId: string, workId: string, actor: Actor): Promise<void> {
  const entry = await k.selectFrom("entries").select("work_id").where("id", "=", entryId).executeTakeFirstOrThrow();
  if (entry.work_id === workId) return;
  await moveEntryToWork(db, userId, entryId, workId, actor, k);
}