// ---------------------------------------------------------------------------
// Fork message-copy — off-event-loop bulk copy of a conversation's messages.
// ---------------------------------------------------------------------------
//
// The fork-based memory retrospective copies the visible tail of the source
// conversation's message rows (rows at-or-after the inherited compaction
// boundary) into a throwaway background conversation. Done in-process via
// `bun:sqlite` (synchronous), that copy is the single longest uninterruptible
// block on the daemon's event loop — on a multi-GB database it pegs the CPU
// for minutes, stalling `/healthz` and the gateway's IPC calls (see the
// runtime-freeze investigation).
//
// This module moves the message-row copy off the event loop via
// {@link runAsyncSqlite}, which runs the statements in a `sqlite3` CLI
// subprocess (its own connection; SQLite file-locking arbitrates with the
// in-process handle). The content/metadata never round-trip through JS — they
// are copied SELECT→INSERT inside SQLite — so there is no string-escaping
// hazard and the heavy bytes stay in the database.
//
// Two properties matter for correctness:
//
//   1. **Lock-friendly batching.** The main connection runs `busy_timeout=5000`
//      (`db-connection.ts`). A single `INSERT … SELECT` over the whole history
//      is one implicit transaction that holds the write lock for its full
//      duration — a concurrent in-process write (a live user turn persisting a
//      message) would then block up to 5 s and throw `SQLITE_BUSY`. So the copy
//      is split into batches that each auto-commit, releasing the lock between
//      batches so in-process writers slip in.
//
//   2. **`cloneForkMessageMetadata` parity.** The in-process fork stamps
//      `forkSourceMessageId` onto each copied row's metadata. The SQL `CASE`
//      below is a faithful translation of that helper's three branches
//      (object → preserve-or-stamp; null/non-object/invalid → fresh object).
//      `client_message_id` is intentionally NOT copied, matching the
//      in-process fork (a forked row is not a client-submitted message).

import { setTimeout as sleep } from "node:timers/promises";

import { BULK_BATCH_TIMEOUT_MS, withBulkWriteGate } from "./bulk-write-gate.js";
import {
  type AsyncSqliteBackend,
  type AsyncSqliteResult,
  runAsyncSqlite,
} from "./db-async-query.js";

/**
 * Default batch size for the chunked copy. Each batch is one `INSERT … SELECT`
 * that auto-commits, so this bounds how long the subprocess holds the write
 * lock before releasing it to in-process writers. Kept small so the worst-case
 * wait for a contending foreground write is one short batch, even on a bloated
 * database; per-batch statement overhead stays negligible against the row copy
 * and fork latency doesn't matter (nobody waits on a background fork).
 */
export const DEFAULT_FORK_COPY_BATCH_SIZE = 50;

/**
 * Pause inserted between batch subprocess calls. Without it the copy releases
 * the write lock on each batch's auto-commit but greedily re-acquires it
 * microseconds later, so a concurrent in-process writer (a live user turn
 * persisting a message) can lose every race. A brief yield lets foreground
 * writes reliably slip in between batches. The extra fork latency is free —
 * nothing waits on a background fork — so we trade copy speed for foreground
 * fairness.
 */
export const DEFAULT_FORK_COPY_INTER_BATCH_DELAY_MS = 25;

/**
 * A single source→fork message-id mapping. `oldId` is an existing source
 * message id; `newId` is the freshly-generated id the fork row will carry.
 * Generated by the caller (in JS) so the same map drives both this copy and
 * the in-process attachment relink that follows.
 */
export interface ForkIdPair {
  oldId: string;
  newId: string;
}

export interface CopyForkMessagesOptions {
  /** The fork conversation id every copied row is assigned to. */
  forkConversationId: string;
  /** Ordered source→fork id pairs for every message to copy. */
  idPairs: readonly ForkIdPair[];
  /** Override the per-batch row count (see {@link DEFAULT_FORK_COPY_BATCH_SIZE}). */
  batchSize?: number;
  /** Test-only passthrough to force the in-process backend. */
  forceInProcess?: boolean;
}

// Server-generated ids only ever use this charset. We interpolate ids (never
// message content) into the SQL script, so reject anything outside it as a
// defense-in-depth guard against a malformed id breaking out of the literal.
const SAFE_ID = /^[A-Za-z0-9_-]{1,64}$/;

function assertSafeId(id: string): void {
  if (!SAFE_ID.test(id)) {
    throw new Error(
      `fork message-copy: unsafe id literal: ${JSON.stringify(id)}`,
    );
  }
}

/**
 * SQL translation of `cloneForkMessageMetadata`. `m` is the source `messages`
 * row alias; the stamped value is the SOURCE message id (`m.id`), preserving an
 * existing `forkSourceMessageId` when the metadata is already an object.
 */
const METADATA_CLONE_EXPR = `CASE
  WHEN json_valid(m.metadata) AND json_type(m.metadata) = 'object'
    THEN json_set(m.metadata, '$.forkSourceMessageId',
      coalesce(json_extract(m.metadata, '$.forkSourceMessageId'), m.id))
  ELSE json_object('forkSourceMessageId', m.id)
END`;

/**
 * Build the multi-statement SQL script that copies the given id pairs into the
 * fork conversation in lock-friendly batches. Exported for unit testing the
 * generated SQL without spawning a subprocess.
 */
export function buildForkCopyScript(options: CopyForkMessagesOptions): string {
  const { forkConversationId, idPairs } = options;
  assertSafeId(forkConversationId);
  const batchSize = Math.max(
    1,
    options.batchSize ?? DEFAULT_FORK_COPY_BATCH_SIZE,
  );

  const statements: string[] = [
    // Connection-scoped staging table holding the current batch's id map. The
    // unqualified `messages` reference in the SELECT resolves to the main DB.
    `CREATE TEMP TABLE IF NOT EXISTS _fork_id_map (old_id TEXT PRIMARY KEY, new_id TEXT NOT NULL);`,
  ];

  for (let i = 0; i < idPairs.length; i += batchSize) {
    const batch = idPairs.slice(i, i + batchSize);
    const values = batch
      .map(({ oldId, newId }) => {
        assertSafeId(oldId);
        assertSafeId(newId);
        return `('${oldId}','${newId}')`;
      })
      .join(",");

    // Each batch auto-commits (no surrounding BEGIN), so the write lock is
    // released between batches for in-process writers.
    statements.push(`DELETE FROM _fork_id_map;`);
    statements.push(
      `INSERT INTO _fork_id_map (old_id, new_id) VALUES ${values};`,
    );
    statements.push(
      `INSERT INTO messages (id, conversation_id, role, content, created_at, metadata)
SELECT map.new_id, '${forkConversationId}', m.role, m.content, m.created_at, ${METADATA_CLONE_EXPR}
FROM messages m JOIN _fork_id_map map ON map.old_id = m.id;`,
    );
  }

  // Drop the staging table so the in-process fallback (which runs on the shared
  // daemon connection) does not leave it lingering; harmless for the subprocess
  // backend, whose connection is discarded on exit.
  statements.push(`DROP TABLE IF EXISTS _fork_id_map;`);

  return statements.join("\n");
}

/**
 * Copy the message rows for a fork off the event loop, in lock-friendly
 * batches. Each batch runs as its own subprocess call with a brief yield in
 * between (see {@link DEFAULT_FORK_COPY_INTER_BATCH_DELAY_MS}) so foreground
 * writers reliably acquire the write lock between batches instead of losing
 * every race to the copy's greedy lock re-acquisition. The whole batch stream
 * runs under the process-wide {@link withBulkWriteGate} so two copies (or a
 * copy and a batched delete) never convoy on the write lock — and the yields
 * go to foreground writers, not a sibling bulk stream. Resolves once every
 * batch has committed (or returns the failing batch's `ok: false` result on
 * subprocess failure — the caller is responsible for cleaning up the
 * partially-built fork).
 *
 * No-op (returns a synthetic ok result) when there is nothing to copy.
 */
export async function copyForkMessagesViaSubprocess(
  options: CopyForkMessagesOptions,
): Promise<AsyncSqliteResult> {
  if (options.idPairs.length === 0) {
    return {
      ok: true,
      backend: "in-process-blocking",
      error: null,
      elapsedMs: 0,
    };
  }

  return await withBulkWriteGate(
    `fork-message-copy:${options.forkConversationId}`,
    () => copyBatches(options),
  );
}

async function copyBatches(
  options: CopyForkMessagesOptions,
): Promise<AsyncSqliteResult> {
  const batchSize = Math.max(
    1,
    options.batchSize ?? DEFAULT_FORK_COPY_BATCH_SIZE,
  );
  const runOptions = options.forceInProcess
    ? { forceBackend: "in-process-blocking" as const }
    : {};

  let totalElapsedMs = 0;
  let backend: AsyncSqliteBackend = "in-process-blocking";

  for (let i = 0; i < options.idPairs.length; i += batchSize) {
    const batch = options.idPairs.slice(i, i + batchSize);
    // One subprocess call per batch. Each generated script is self-contained
    // (it creates and drops its own staging table), so splitting execution
    // across calls is safe — and it's what lets us yield between batches below.
    const sql = buildForkCopyScript({
      forkConversationId: options.forkConversationId,
      idPairs: batch,
      batchSize,
    });
    const result = await runAsyncSqlite(
      sql,
      `fork-message-copy:copy-batch:${options.forkConversationId}`,
      { ...runOptions, timeoutMs: BULK_BATCH_TIMEOUT_MS },
    );
    totalElapsedMs += result.elapsedMs;
    backend = result.backend;
    if (!result.ok) {
      return { ...result, elapsedMs: totalElapsedMs };
    }

    const isLastBatch = i + batchSize >= options.idPairs.length;
    if (!isLastBatch) {
      await sleep(DEFAULT_FORK_COPY_INTER_BATCH_DELAY_MS);
    }
  }

  return { ok: true, backend, error: null, elapsedMs: totalElapsedMs };
}