/**
 * Cross-process render lanes: one fair, visible queue per transport.
 *
 * WHY THIS EXISTS
 * ---------------
 * Many agents share a few rendering engines and nothing coordinates them. On
 * 2026-08-10 two drivers on the SAME Higgsfield lane produced 88
 * `free slot stayed busy` events and 0 NSFW rejections over ~7 hours: scene 5
 * exhausted all 6 attempts, scene 13 burned 3 attempts across 70 minutes, and
 * after one driver was killed scene 13 landed in 45 SECONDS. Contention made
 * throughput worse than serialising, and it was misdiagnosed as moderation for
 * hours because the two look identical at the scene level and have OPPOSITE
 * fixes (serialise vs. redraw).
 *
 * `execute-pool.ts` already caps concurrency, but only WITHIN one process — one
 * driver's `--max-concurrent 1` tells another process nothing. That is the gap.
 *
 * SHAPE
 * -----
 * A lease, not a task queue. Each agent still runs its own render (it needs its
 * own project context, references and the shared browser profile); it only waits
 * its TURN. Worker-daemon queues (huey et al.) were rejected because they assume
 * a central process executes the work.
 *
 * One lane per transport, so different engines run genuinely in parallel — a
 * global queue would serialise Flow behind Higgsfield for no reason (observed:
 * `baby-hanuman-ep5` rendered on veo-useapi concurrently with a Higgsfield film,
 * with zero contention).
 *
 * STORAGE: the sqlite3 CLI, not `node:sqlite`. package.json declares
 * `engines.node >=20.10`; `node:sqlite` needs 22.5+ and prints an experimental
 * warning on import, which would corrupt this CLI's machine-readable JSON.
 * Shelling out to a binary is already the norm across `src/video/*.ts`.
 */
import { execFileSync } from 'node:child_process';
import { existsSync, mkdirSync } from 'node:fs';
import { dirname, join } from 'node:path';
import { createHash, randomUUID } from 'node:crypto';

/** A lane is keyed `route:account` — limits are per ACCOUNT, not per route. */
export type LaneId = string;

export type TicketState = 'queued' | 'held' | 'done';

export interface LaneTicket {
  id: string;
  lane: LaneId;
  project: string;
  scene: number | null;
  promptHash: string | null;
  state: TicketState;
  createdAt: number;
  grantedAt: number | null;
  expiresAt: number | null;
  holder: string | null;
  /** The driver's claim (remote coordinator only): two sessions on one machine
   *  share `holder`; this is what tells them apart. Absent on the local queue. */
  claimId?: string;
  /** The render log (remote coordinator): what the driver reported on release,
   *  `expired` from the coordinator's reap, null before the log existed. */
  outcome?: string | null;
  providerJobId?: string | null;
  note?: string | null;
}

export interface AcquireResult {
  status: 'granted' | 'queued';
  ticketId: string;
  /** 1-based place in the fair ordering. 0 when granted. */
  position: number;
  /** Rough wait from observed job durations on this lane; null when unknown. */
  etaSeconds: number | null;
  lane: LaneId;
  /** True when an identical in-flight request already existed (dedupe hit). */
  deduped: boolean;
  coordinator?: 'local-sqlite' | 'cloudflare-durable-object';
}

export interface LaneStatus {
  lane: LaneId;
  limit: number | null;
  held: LaneTicket[];
  queued: Array<LaneTicket & { position: number; reason: string }>;
  medianJobSeconds: number | null;
  /** Bounded terminal rows retained for ETA/history diagnostics. */
  terminalRetained: number;
  coordinator?: 'local-sqlite' | 'cloudflare-durable-object';
}

/** Injectable so tests can drive the clock and target a temp database. */
export interface LaneQueueOptions {
  dbPath?: string;
  now?: () => number;
  /** Escape hatch for tests; defaults to the sqlite3 CLI. */
  runSql?: (sql: string, dbPath: string) => string;
  /** Retained terminal rows per lane; injectable only for compact regression tests. */
  maxTerminalPerLane?: number;
}

const DEFAULT_TTL_SEC = 1800;
const DEFAULT_QUEUE_TTL_SEC = 7200;
const MAX_TERMINAL_PER_LANE = 500;

export function defaultLaneDbPath(env: NodeJS.ProcessEnv = process.env): string {
  const home = env.VCLAW_WORKSPACE || env.VIDEOCLAW_WORKSPACE
    || join(env.HOME || '.', 'videoclaw');
  return join(home, 'lanes.db');
}

/**
 * `busy_timeout` is PER CONNECTION, and every call here spawns a fresh sqlite3
 * process — so setting it once at init does nothing for later invocations.
 * Without it on every script, concurrent callers get an instant
 * `database is locked (5)` and CRASH instead of queueing. Measured: 8 racing
 * processes → 1 holder and 7 hard failures. Mutual exclusion was correct; the
 * losers just died. Every unit test passed, because they share one process.
 * The pragma is prepended here so it cannot be forgotten at a call site.
 */
function sqlite(sql: string, dbPath: string, attempt = 0): string {
  try {
    // `.timeout` (dot-command), NOT `PRAGMA busy_timeout` — the pragma RETURNS A
    // ROW, so under `-json` it emits `[{"timeout":15000}]`. That row was then read
    // as the result of the dedupe SELECT, so every caller returned early claiming
    // a dedupe hit and never inserted a ticket: 8 racers, 0 rows in the database.
    // Dot-commands produce no result rows and cannot corrupt a query's output.
    return execFileSync('sqlite3', ['-json', '-bail', '-cmd', '.timeout 15000', dbPath], {
      input: sql,
      encoding: 'utf-8',
      timeout: 30_000,
    });
  } catch (err) {
    const e = err as NodeJS.ErrnoException & { stderr?: string };
    if (e.code === 'ENOENT') {
      throw new Error(
        'lane queue needs the `sqlite3` CLI on PATH (standard on macOS and most Linux). '
        + 'Install it, or pass a custom runSql.',
      );
    }
    // Belt and braces: a writer can still lose the race at the moment WAL is
    // being checkpointed. Bounded retry with backoff, then surface it.
    const msg = String(e.stderr ?? e.message ?? '');
    if (/database is locked|database table is locked/i.test(msg) && attempt < 5) {
      const until = Date.now() + 150 * (attempt + 1);
      while (Date.now() < until) { /* brief spin; calls are seconds apart */ }
      return sqlite(sql, dbPath, attempt + 1);
    }
    throw err;
  }
}

/** Single-quote escaping; every value written here is a slug, hash or number. */
function q(v: string | number | null): string {
  if (v === null || v === undefined) return 'NULL';
  if (typeof v === 'number') return Number.isFinite(v) ? String(v) : 'NULL';
  return `'${String(v).replace(/'/g, "''")}'`;
}

function rows<T = Record<string, unknown>>(out: string): T[] {
  const trimmed = out.trim();
  if (!trimmed) return [];
  // sqlite3 -json emits one JSON array per statement that returns rows.
  const chunks = trimmed.split(/\n(?=\[)/);
  const last = chunks[chunks.length - 1];
  try {
    return JSON.parse(last) as T[];
  } catch {
    return [];
  }
}

export function promptHashOf(text: string): string {
  return createHash('sha256').update(text).digest('hex').slice(0, 16);
}

export class LaneQueue {
  readonly coordinator = 'local-sqlite' as const;
  private readonly dbPath: string;
  private readonly now: () => number;
  private readonly exec: (sql: string, dbPath: string) => string;
  private readonly maxTerminalPerLane: number;
  private ready = false;

  constructor(opts: LaneQueueOptions = {}) {
    this.dbPath = opts.dbPath ?? defaultLaneDbPath();
    this.now = opts.now ?? (() => Date.now() / 1000);
    this.exec = opts.runSql ?? sqlite;
    this.maxTerminalPerLane = opts.maxTerminalPerLane ?? MAX_TERMINAL_PER_LANE;
    if (!Number.isInteger(this.maxTerminalPerLane) || this.maxTerminalPerLane < 1) {
      throw new Error('maxTerminalPerLane must be a positive integer');
    }
  }

  private init(): void {
    if (this.ready) return;
    const dir = dirname(this.dbPath);
    if (!existsSync(dir)) mkdirSync(dir, { recursive: true });
    this.exec(
      `PRAGMA journal_mode=WAL;
       PRAGMA busy_timeout=15000;
       CREATE TABLE IF NOT EXISTS tickets (
         id TEXT PRIMARY KEY,
         lane TEXT NOT NULL,
         project TEXT NOT NULL,
         scene INTEGER,
         prompt_hash TEXT,
         state TEXT NOT NULL,
         created_at REAL NOT NULL,
         granted_at REAL,
         expires_at REAL,
         released_at REAL,
         holder TEXT
       );
       CREATE INDEX IF NOT EXISTS tickets_lane_state ON tickets(lane, state);
       -- Repair any duplicate active requests written by pre-atomic versions
       -- before installing the invariant. The oldest ticket is canonical.
       UPDATE tickets SET state='done', released_at=COALESCE(released_at, CAST(strftime('%s','now') AS REAL))
       WHERE id IN (
         SELECT id FROM (
           SELECT id, ROW_NUMBER() OVER (
             PARTITION BY lane, project, IFNULL(scene, -1), prompt_hash
             ORDER BY created_at ASC, id ASC
           ) AS duplicate_rank
           FROM tickets
           WHERE state IN ('queued','held') AND prompt_hash IS NOT NULL
         ) WHERE duplicate_rank > 1
       );
       CREATE UNIQUE INDEX IF NOT EXISTS tickets_active_request
         ON tickets(lane, project, IFNULL(scene, -1), prompt_hash)
         WHERE state IN ('queued','held') AND prompt_hash IS NOT NULL;
       -- Fairness ledger: when each project last got the lane. Round-robin reads
       -- this, so one 15-scene film cannot starve another agent's single clip.
       CREATE TABLE IF NOT EXISTS lane_service (
         lane TEXT NOT NULL,
         project TEXT NOT NULL,
         last_served REAL NOT NULL,
         PRIMARY KEY (lane, project)
       );`,
      this.dbPath,
    );
    this.ready = true;
  }

  /** Expire leases whose holder died without releasing. Never a PID check:
   *  `kill -0 0` signals the process group and returns success, which hung a
   *  runner for 20 minutes on 2026-08-11. A TTL cannot lie that way. */
  private reapSql(now: number): string {
    return `UPDATE tickets SET state='done', released_at=${q(now)}
            WHERE state IN ('queued','held') AND expires_at IS NOT NULL AND expires_at < ${q(now)};`;
  }

  /** Keep queue history useful but bounded under feature-scale throughput. */
  private pruneSql(keepPerLane = MAX_TERMINAL_PER_LANE): string {
    return `DELETE FROM tickets WHERE id IN (
              SELECT id FROM (
                SELECT id, ROW_NUMBER() OVER (
                  PARTITION BY lane ORDER BY released_at DESC, id DESC
                ) AS terminal_rank
                FROM tickets WHERE state='done'
              ) WHERE terminal_rank > ${q(keepPerLane)}
            );`;
  }

  /** Fill every free slot while rotating projects before taking their next task. */
  private promoteSql(laneExpression: string, limit: number, now: number, ttl: number, table: string): string {
    return `CREATE TEMP TABLE IF NOT EXISTS ${table}(id TEXT);
            DELETE FROM ${table};
            INSERT INTO ${table}
              WITH ranked AS (
                SELECT t.id, t.created_at,
                       COALESCE(s.last_served, 0) AS last_served,
                       ROW_NUMBER() OVER (
                         PARTITION BY t.project ORDER BY t.created_at ASC, t.id ASC
                       ) AS project_rank
                FROM tickets t
                LEFT JOIN lane_service s ON s.lane=t.lane AND s.project=t.project
                WHERE t.lane=${laneExpression} AND t.state='queued'
              )
              SELECT id FROM ranked
              ORDER BY project_rank ASC, last_served ASC, created_at ASC, id ASC
              LIMIT (SELECT MAX(0, ${q(limit)} - COUNT(*))
                     FROM tickets WHERE lane=${laneExpression} AND state='held');
            UPDATE tickets SET state='held', granted_at=${q(now)}, expires_at=${q(now + ttl)}
              WHERE id IN (SELECT id FROM ${table});
            INSERT INTO lane_service (lane, project, last_served)
              SELECT lane, project, ${q(now)} FROM tickets WHERE id IN (SELECT id FROM ${table})
              ON CONFLICT(lane, project) DO UPDATE SET last_served=${q(now)};`;
  }

  acquire(params: {
    lane: LaneId;
    project: string;
    scene?: number | null;
    promptHash?: string | null;
    limit: number | null;
    ttlSec?: number;
    holder?: string;
  }): AcquireResult {
    this.init();
    const now = this.now();
    const ttl = params.ttlSec ?? DEFAULT_TTL_SEC;
    const queueTtl = Math.max(ttl, DEFAULT_QUEUE_TTL_SEC);
    const scene = params.scene ?? null;
    const hash = params.promptHash?.trim() || null;

    // An unlimited lane is a no-op: hand back a granted ticket without ever
    // contending. This is what keeps enforcement genuinely opt-in per lane.
    if (params.limit === null || params.limit === undefined) {
      return {
        status: 'granted', ticketId: `unlimited-${now}`, position: 0,
        etaSeconds: 0, lane: params.lane, deduped: false,
      };
    }

    if (!Number.isInteger(params.limit) || params.limit < 1) {
      throw new Error(`lane ${params.lane} requires a positive integer limit`);
    }
    if (hash === null) {
      throw new Error(
        `lane ${params.lane} requires a stable promptHash/request identity; `
        + 'retries without one cannot be deduplicated safely',
      );
    }

    const id = `${params.lane}-${params.project}-${scene ?? 'x'}-${randomUUID()}`
      .replace(/[^A-Za-z0-9:_-]/g, '_');

    // Dedupe, insert and capacity grant are ONE write transaction. The partial
    // unique index is the cross-process invariant; INSERT OR IGNORE makes every
    // racing caller converge on the same canonical active ticket.
    const granted = this.exec(
      `BEGIN IMMEDIATE;
       ${this.reapSql(now)}
       INSERT OR IGNORE INTO tickets
         (id, lane, project, scene, prompt_hash, state, created_at, expires_at, holder)
         VALUES (${q(id)}, ${q(params.lane)}, ${q(params.project)},
                 ${scene === null ? 'NULL' : q(scene)}, ${q(hash)}, 'queued', ${q(now)}, ${q(now + queueTtl)},
                 ${q(params.holder ?? null)});
       -- A live waiter proves this request is still wanted. Refresh only queued
       -- tickets; a held lease is renewed explicitly by heartbeat.
       UPDATE tickets SET expires_at=${q(now + queueTtl)}
         WHERE lane=${q(params.lane)} AND project=${q(params.project)}
           AND scene IS ${scene === null ? 'NULL' : q(scene)} AND prompt_hash=${q(hash)}
           AND state='queued';
       ${this.promoteSql(q(params.lane), params.limit, now, ttl, 'acquire_pick')}
       SELECT id, state, CASE WHEN id=${q(id)} THEN 0 ELSE 1 END AS deduped
         FROM tickets WHERE lane=${q(params.lane)} AND project=${q(params.project)}
           AND scene IS ${scene === null ? 'NULL' : q(scene)} AND prompt_hash=${q(hash)}
           AND state IN ('queued','held') LIMIT 1;
       COMMIT;`,
      this.dbPath,
    );
    const active = rows<{ id: string; state: string; deduped: number }>(granted)[0];
    if (!active) throw new Error(`lane ${params.lane} failed to persist request ${hash}`);
    const state = active.state;
    if (state === 'held') {
      return {
        status: 'granted', ticketId: active.id, position: 0, etaSeconds: 0,
        lane: params.lane, deduped: active.deduped === 1,
      };
    }
    const st = this.status(params.lane, params.limit);
    const mine = st.queued.find((t) => t.id === active.id);
    const position = mine ? mine.position : st.queued.length;
    return {
      status: 'queued', ticketId: active.id, position,
      etaSeconds: this.etaFor(st, position), lane: params.lane, deduped: active.deduped === 1,
    };
  }

  /** Refresh a lease. A holder that stops heartbeating loses the slot at TTL. */
  heartbeat(ticketId: string, ttlSec = DEFAULT_TTL_SEC): boolean {
    this.init();
    const now = this.now();
    const out = this.exec(
      `BEGIN IMMEDIATE;
       UPDATE tickets SET expires_at=${q(now + ttlSec)}
         WHERE id=${q(ticketId)} AND state='held';
       COMMIT;
       SELECT changes() AS n;`,
      this.dbPath,
    );
    return (rows<{ n: number }>(out)[0]?.n ?? 0) > 0;
  }

  /** Release the slot and immediately promote the fair next ticket. */
  release(ticketId: string, lane?: LaneId, limit: number | null = null): boolean {
    this.init();
    const now = this.now();
    const laneExpr = lane ? q(lane) : `(SELECT lane FROM tickets WHERE id=${q(ticketId)})`;
    const out = this.exec(
      `BEGIN IMMEDIATE;
       CREATE TEMP TABLE IF NOT EXISTS release_change(n INTEGER);
       DELETE FROM release_change;
       UPDATE tickets SET state='done', released_at=${q(now)}
         WHERE id=${q(ticketId)} AND state IN ('queued','held');
       INSERT INTO release_change VALUES (changes());
       ${this.reapSql(now)}
       ${limit === null ? '' : `
       ${this.promoteSql(laneExpr, limit, now, DEFAULT_TTL_SEC, 'release_pick')}`}
       ${this.pruneSql(this.maxTerminalPerLane)}
       SELECT n FROM release_change;
       COMMIT;`,
      this.dbPath,
    );
    return (rows<{ n: number }>(out)[0]?.n ?? 0) > 0;
  }

  /**
   * Backward-compatible escape hatch for old callers that did not persist a
   * ticket. It is intentionally safe only when exactly one lease matches;
   * ambiguity is refused so parallel attempts can never release one another.
   * New execution paths must call `release(ticketId, ...)` directly.
   */
  releaseByProject(lane: LaneId, project: string, limit: number | null = null): number {
    this.init();
    const out = this.exec(
      `SELECT id FROM tickets
       WHERE lane=${q(lane)} AND project=${q(project)} AND state='held'
       ORDER BY granted_at ASC, id ASC LIMIT 2;`,
      this.dbPath,
    );
    const active = rows<{ id: string }>(out);
    if (active.length > 1) {
      throw new Error(
        `refusing broad release for ${project} on ${lane}: ${active.length}+ held tickets; carry the exact ticket id`,
      );
    }
    return active[0] && this.release(active[0].id, lane, limit) ? 1 : 0;
  }

  status(lane: LaneId, limit: number | null = null): LaneStatus {
    this.init();
    const now = this.now();
    const out = this.exec(
      `BEGIN IMMEDIATE;
       ${this.reapSql(now)}
       ${limit === null ? '' : this.promoteSql(q(lane), limit, now, DEFAULT_TTL_SEC, 'status_pick')}
       ${this.pruneSql(this.maxTerminalPerLane)}
       COMMIT;
       SELECT id, lane, project, scene, prompt_hash, state, created_at, granted_at,
              expires_at, holder,
              COALESCE((SELECT s.last_served FROM lane_service s
                        WHERE s.lane=t.lane AND s.project=t.project), 0) AS last_served
       FROM tickets t
       WHERE lane=${q(lane)} AND state IN ('queued','held')
       ORDER BY CASE state WHEN 'held' THEN 0 ELSE 1 END,
                last_served ASC, created_at ASC;`,
      this.dbPath,
    );
    const all = rows<Record<string, never>>(out) as unknown as Array<Record<string, unknown>>;
    const map = (r: Record<string, unknown>): LaneTicket => ({
      id: String(r.id), lane: String(r.lane), project: String(r.project),
      scene: r.scene === null ? null : Number(r.scene),
      promptHash: r.prompt_hash === null ? null : String(r.prompt_hash),
      state: String(r.state) as TicketState,
      createdAt: Number(r.created_at),
      grantedAt: r.granted_at === null ? null : Number(r.granted_at),
      expiresAt: r.expires_at === null ? null : Number(r.expires_at),
      holder: r.holder === null ? null : String(r.holder),
    });
    const held = all.filter((r) => r.state === 'held').map(map);
    const queued = all.filter((r) => r.state === 'queued').map((r, i) => ({
      ...map(r),
      position: i + 1,
      reason: Number(r.last_served) === 0
        ? 'never served on this lane'
        : `project last served ${Math.round(now - Number(r.last_served))}s ago`,
    }));
    const terminalRetained = rows<{ n: number }>(this.exec(
      `SELECT COUNT(*) AS n FROM tickets WHERE lane=${q(lane)} AND state='done';`,
      this.dbPath,
    ))[0]?.n ?? 0;
    return { lane, limit, held, queued, medianJobSeconds: this.medianJobSeconds(lane), terminalRetained };
  }

  /** Median of completed job durations — the basis for a real ETA. */
  medianJobSeconds(lane: LaneId): number | null {
    const out = this.exec(
      `SELECT (released_at - granted_at) AS d FROM tickets
       WHERE lane=${q(lane)} AND state='done' AND granted_at IS NOT NULL
         AND released_at IS NOT NULL AND released_at > granted_at
       ORDER BY released_at DESC LIMIT 25;`,
      this.dbPath,
    );
    const ds = rows<{ d: number }>(out).map((r) => Number(r.d)).filter((n) => Number.isFinite(n));
    if (ds.length === 0) return null;
    ds.sort((a, b) => a - b);
    return Math.round(ds[Math.floor(ds.length / 2)]);
  }

  private etaFor(st: LaneStatus, position: number): number | null {
    if (st.medianJobSeconds === null) return null;
    const lim = st.limit && st.limit > 0 ? st.limit : 1;
    return Math.round((position / lim) * st.medianJobSeconds);
  }
}