import { createHash, randomUUID } from "node:crypto";
import postgres from "postgres";
import { createHourlyJsonlWriter, logRetentionBytesForService } from "../../../shared/src/rotating-jsonl";

type JsonValue = string | number | boolean | null | JsonValue[] | { [key: string]: JsonValue };
type JsonRecord = Record<string, JsonValue>;
type SqlClient = postgres.Sql;
type SqlExecutor = postgres.Sql | postgres.TransactionSql;

interface RuntimeConfig {
  host: string;
  port: number;
  databaseUrl: string;
  logFile: string;
  databasePoolMax: number;
  pipelineBridgeBaseUrl: string;
  pipelineBridgeIntervalMs: number;
  pipelineBridgeRunLimit: number;
  deployServiceId: string;
  deployRef: string;
  deployRepo: string;
  deployCommit: string;
  deployRequestedCommit: string;
}

interface OaEventInput {
  eventId?: unknown;
  type?: unknown;
  createdAt?: unknown;
  sourceKind?: unknown;
  sourceId?: unknown;
  aggregateType?: unknown;
  aggregateId?: unknown;
  correlationId?: unknown;
  causationId?: unknown;
  tags?: unknown;
  payload?: unknown;
}

interface OaEventRecord {
  eventId: string;
  sequence: number;
  type: string;
  createdAt: string;
  sourceKind: string;
  sourceId: string;
  aggregateType: string;
  aggregateId: string;
  correlationId: string;
  causationId: string | null;
  tags: string[];
  payload: JsonRecord;
}

interface OaEventRow {
  event_id: string;
  sequence: string | number;
  type: string;
  source_kind: string;
  source_id: string;
  aggregate_type: string;
  aggregate_id: string;
  correlation_id: string;
  causation_id: string | null;
  tags: JsonValue;
  payload: JsonValue;
  created_at: Date | string;
}

interface TraceStatsRow {
  scope_id: string;
  service_id: string;
  subject_kind: string;
  subject_id: string;
  stats_revision: string | number;
  step_count: string | number;
  llm_step_count: string | number;
  read_count: string | number;
  edit_count: string | number;
  run_count: string | number;
  error_count: string | number;
  trace_line_count: string | number;
  output_max_seq: string | number;
  attempt_stats_json: JsonValue;
  last_event_sequence: string | number | null;
  updated_at: Date | string;
}

interface TraceStatsRecord extends JsonRecord {
  scopeId: string;
  serviceId: string;
  subjectKind: string;
  subjectId: string;
  statsRevision: number;
  stepCount: number;
  llmStepCount: number;
  readCount: number;
  editCount: number;
  runCount: number;
  errorCount: number;
  traceLineCount: number;
  outputMaxSeq: number;
  attemptStats: JsonValue;
  lastEventSequence: number | null;
  updatedAt: string;
  taskId: string;
  attemptIndex: number | null;
  source: "oa-event-flow";
}

interface TraceScope {
  scopeId: string;
  serviceId: string;
  subjectKind: string;
  subjectId: string;
  taskId?: string;
  attemptIndex?: number | null;
}

interface EventClient {
  id: string;
  tags: string[];
  enqueue: (chunk: string) => void;
  close: () => void;
  connectedAt: string;
}

interface PipelineBridgeRunState extends JsonRecord {
  runId: string;
  pipelineId: string;
  lastObservedEventCount: number;
  lastSyncedAt: string;
  lastError: string;
  bridgeMode: string;
  insertedCount: number;
  duplicateCount: number;
}

interface PipelineBridgeState extends JsonRecord {
  enabled: boolean;
  mode: string;
  baseUrl: string;
  intervalMs: number;
  runLimit: number;
  inFlight: boolean;
  lastStartedAt: string;
  lastFinishedAt: string;
  lastError: string;
  syncCount: number;
  insertedCount: number;
  duplicateCount: number;
  snapshotSyncCount: number;
  runs: Record<string, PipelineBridgeRunState>;
}

const serviceStartedAt = new Date().toISOString();
const recentLogs: JsonRecord[] = [];
const eventClients = new Set<EventClient>();
const eventEncoder = new TextEncoder();
let databaseReady = false;
let databaseLastError: string | null = null;
let pipelineBridgeTimer: ReturnType<typeof setInterval> | null = null;

function envString(name: string, fallback: string): string {
  const value = process.env[name];
  return value === undefined || value.length === 0 ? fallback : value;
}

function envRequiredString(name: string): string {
  const value = process.env[name];
  if (value === undefined || value.trim().length === 0) throw new Error(`${name} is required`);
  return value;
}

function envNumber(name: string, fallback: number): number {
  const raw = process.env[name];
  if (raw === undefined || raw.length === 0) return fallback;
  const value = Number(raw);
  return Number.isFinite(value) && value > 0 ? Math.floor(value) : fallback;
}

function configFromEnv(): RuntimeConfig {
  return {
    host: envString("HOST", "0.0.0.0"),
    port: envNumber("PORT", 4255),
    databaseUrl: envRequiredString("DATABASE_URL"),
    logFile: envString("LOG_FILE", "/var/log/unidesk/oa-event-flow.jsonl"),
    databasePoolMax: Math.max(1, Math.min(8, envNumber("DATABASE_POOL_MAX", 2))),
    pipelineBridgeBaseUrl: envString("PIPELINE_OA_BRIDGE_BASE_URL", "").replace(/\/+$/u, ""),
    pipelineBridgeIntervalMs: envNumber("PIPELINE_OA_BRIDGE_INTERVAL_MS", 15_000),
    pipelineBridgeRunLimit: envNumber("PIPELINE_OA_BRIDGE_RUN_LIMIT", 50),
    deployServiceId: envString("UNIDESK_DEPLOY_SERVICE_ID", "oa-event-flow"),
    deployRef: envString("UNIDESK_DEPLOY_REF", ""),
    deployRepo: envString("UNIDESK_DEPLOY_REPO", ""),
    deployCommit: envString("UNIDESK_DEPLOY_COMMIT", ""),
    deployRequestedCommit: envString("UNIDESK_DEPLOY_REQUESTED_COMMIT", ""),
  };
}

const config = configFromEnv();
const sql: SqlClient = postgres(config.databaseUrl, {
  max: config.databasePoolMax,
  idle_timeout: 20,
  connect_timeout: 10,
  connection: { application_name: "unidesk-oa-event-flow" },
});
const pipelineBridgeState: PipelineBridgeState = {
  enabled: config.pipelineBridgeBaseUrl.length > 0,
  mode: "snapshot",
  baseUrl: config.pipelineBridgeBaseUrl,
  intervalMs: config.pipelineBridgeIntervalMs,
  runLimit: config.pipelineBridgeRunLimit,
  inFlight: false,
  lastStartedAt: "",
  lastFinishedAt: "",
  lastError: "",
  syncCount: 0,
  insertedCount: 0,
  duplicateCount: 0,
  snapshotSyncCount: 0,
  runs: {},
};
const logWriter = config.logFile
  ? createHourlyJsonlWriter({
      baseLogFile: config.logFile,
      service: "oa-event-flow",
      maxBytes: logRetentionBytesForService("oa-event-flow"),
    })
  : null;
logWriter?.prune();

function log(level: "debug" | "info" | "warn" | "error", event: string, detail: JsonRecord = {}): void {
  const record: JsonRecord = { at: new Date().toISOString(), service: "oa-event-flow", level, event, ...detail };
  recentLogs.push(record);
  while (recentLogs.length > 500) recentLogs.shift();
  try {
    logWriter?.appendJson(record, new Date(String(record.at)));
  } catch {
    // Logging must not break the event bus.
  }
  const line = JSON.stringify(record);
  const writer = level === "error" ? console.error : level === "warn" ? console.warn : console.log;
  writer(line);
}

function jsonResponse(body: unknown, status = 200, headers: Record<string, string> = {}): Response {
  return new Response(JSON.stringify(body), {
    status,
    headers: { "content-type": "application/json; charset=utf-8", ...headers },
  });
}

function errorToJson(error: unknown): JsonRecord {
  if (error instanceof Error) return { name: error.name, message: error.message, stack: error.stack || "" };
  return { message: String(error) };
}

function errorResponse(error: unknown, status = 500): Response {
  log("error", "request_failed", { status, error: errorToJson(error) });
  return jsonResponse({ ok: false, error: error instanceof Error ? error.message : String(error) }, status);
}

function asRecord(value: unknown): Record<string, unknown> | null {
  return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, unknown> : null;
}

function asJsonRecord(value: unknown): JsonRecord {
  if (typeof value === "object" && value !== null && !Array.isArray(value)) return value as JsonRecord;
  return {};
}

function stringValue(value: unknown, fallback = ""): string {
  if (typeof value === "string") return value.trim();
  if (typeof value === "number" || typeof value === "boolean" || typeof value === "bigint") return String(value).trim();
  return fallback;
}

function isoTimestamp(value: unknown, fallback = new Date().toISOString()): string {
  const raw = stringValue(value);
  if (raw.length === 0) return fallback;
  const time = Date.parse(raw);
  return Number.isFinite(time) ? new Date(time).toISOString() : fallback;
}

function normalizeTag(value: unknown): string {
  return stringValue(value).replace(/\s+/gu, " ").trim();
}

function normalizeTags(value: unknown): string[] {
  const source = Array.isArray(value) ? value : typeof value === "string" ? value.split(",") : [];
  const tags = source.map(normalizeTag).filter((tag) => tag.length > 0 && tag.length <= 160);
  return Array.from(new Set(tags));
}

function addTag(tags: string[], tag: string): void {
  const normalized = normalizeTag(tag);
  if (normalized.length > 0 && !tags.includes(normalized)) tags.push(normalized);
}

function stableEventHash(value: unknown): string {
  return createHash("sha256").update(JSON.stringify(value)).digest("hex").slice(0, 32);
}

function normalizeEvent(input: OaEventInput): OaEventRecord {
  const payload = asJsonRecord(input.payload);
  const type = stringValue(input.type, "event").slice(0, 160) || "event";
  const sourceKind = stringValue(input.sourceKind, "service").slice(0, 120) || "service";
  const sourceId = stringValue(input.sourceId, "unknown").slice(0, 160) || "unknown";
  const aggregateType = stringValue(input.aggregateType, stringValue(payload.aggregateType, "")).slice(0, 120);
  const aggregateId = stringValue(input.aggregateId, stringValue(payload.aggregateId, "")).slice(0, 240);
  const createdAt = isoTimestamp(input.createdAt, new Date().toISOString());
  const correlationId = stringValue(input.correlationId, stringValue(payload.correlationId, "")).slice(0, 240)
    || `${sourceId}:${aggregateId || type}`;
  const causationId = stringValue(input.causationId, "").slice(0, 240) || null;
  const tags = normalizeTags(input.tags);
  if (sourceId) addTag(tags, `service:${sourceId}`);
  if (aggregateType && aggregateId) addTag(tags, `${aggregateType}:${aggregateId}`);
  const eventId = stringValue(input.eventId).slice(0, 300)
    || `evt_${stableEventHash({ type, sourceKind, sourceId, aggregateType, aggregateId, correlationId, causationId, tags, payload, createdAt })}`
    || `evt_${randomUUID()}`;
  return { eventId, sequence: 0, type, createdAt, sourceKind, sourceId, aggregateType, aggregateId, correlationId, causationId, tags, payload };
}

function rowToEvent(row: OaEventRow): OaEventRecord {
  return {
    eventId: row.event_id,
    sequence: Number(row.sequence),
    type: row.type,
    createdAt: isoTimestamp(row.created_at),
    sourceKind: row.source_kind,
    sourceId: row.source_id,
    aggregateType: row.aggregate_type,
    aggregateId: row.aggregate_id,
    correlationId: row.correlation_id,
    causationId: row.causation_id,
    tags: Array.isArray(row.tags) ? row.tags.map((item) => String(item)) : [],
    payload: asJsonRecord(row.payload),
  };
}

function statsRowToRecord(row: TraceStatsRow): TraceStatsRecord {
  const taskAttempt = row.scope_id.match(/^task:([^:]+):attempt:(\d+)$/u);
  const taskOnly = row.scope_id.match(/^task:([^:]+)$/u);
  const taskId = taskAttempt?.[1] ?? taskOnly?.[1] ?? "";
  const attemptIndex = taskAttempt === null ? null : Number(taskAttempt[2]);
  return {
    scopeId: row.scope_id,
    serviceId: row.service_id,
    subjectKind: row.subject_kind,
    subjectId: row.subject_id,
    statsRevision: Number(row.stats_revision),
    stepCount: Number(row.step_count),
    llmStepCount: Number(row.llm_step_count),
    readCount: Number(row.read_count),
    editCount: Number(row.edit_count),
    runCount: Number(row.run_count),
    errorCount: Number(row.error_count),
    traceLineCount: Number(row.trace_line_count),
    outputMaxSeq: Number(row.output_max_seq),
    attemptStats: row.attempt_stats_json,
    lastEventSequence: row.last_event_sequence === null ? null : Number(row.last_event_sequence),
    updatedAt: isoTimestamp(row.updated_at),
    taskId,
    attemptIndex,
    source: "oa-event-flow",
  };
}

async function ensureSchema(): Promise<void> {
  await sql`
    CREATE TABLE IF NOT EXISTS oa_events (
      event_id TEXT PRIMARY KEY,
      sequence BIGSERIAL UNIQUE NOT NULL,
      type TEXT NOT NULL,
      source_kind TEXT NOT NULL,
      source_id TEXT NOT NULL,
      aggregate_type TEXT NOT NULL DEFAULT '',
      aggregate_id TEXT NOT NULL DEFAULT '',
      correlation_id TEXT NOT NULL DEFAULT '',
      causation_id TEXT,
      tags JSONB NOT NULL DEFAULT '[]'::jsonb,
      payload JSONB NOT NULL DEFAULT '{}'::jsonb,
      created_at TIMESTAMPTZ NOT NULL DEFAULT now()
    )
  `;
  await sql`CREATE INDEX IF NOT EXISTS idx_oa_events_sequence ON oa_events(sequence ASC)`;
  await sql`CREATE INDEX IF NOT EXISTS idx_oa_events_type_sequence ON oa_events(type, sequence ASC)`;
  await sql`CREATE INDEX IF NOT EXISTS idx_oa_events_source_sequence ON oa_events(source_id, sequence ASC)`;
  await sql`CREATE INDEX IF NOT EXISTS idx_oa_events_created ON oa_events(created_at DESC)`;
  await sql`CREATE INDEX IF NOT EXISTS idx_oa_events_tags_gin ON oa_events USING GIN(tags jsonb_path_ops)`;
  await sql`
    CREATE TABLE IF NOT EXISTS oa_projection_offsets (
      projection_name TEXT PRIMARY KEY,
      last_sequence BIGINT NOT NULL DEFAULT 0,
      updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
    )
  `;
  await sql`
    CREATE TABLE IF NOT EXISTS oa_trace_stats (
      scope_id TEXT PRIMARY KEY,
      service_id TEXT NOT NULL DEFAULT '',
      subject_kind TEXT NOT NULL DEFAULT '',
      subject_id TEXT NOT NULL DEFAULT '',
      stats_revision BIGINT NOT NULL DEFAULT 0,
      step_count INTEGER NOT NULL DEFAULT 0,
      llm_step_count INTEGER NOT NULL DEFAULT 0,
      read_count INTEGER NOT NULL DEFAULT 0,
      edit_count INTEGER NOT NULL DEFAULT 0,
      run_count INTEGER NOT NULL DEFAULT 0,
      error_count INTEGER NOT NULL DEFAULT 0,
      trace_line_count INTEGER NOT NULL DEFAULT 0,
      output_max_seq BIGINT NOT NULL DEFAULT 0,
      attempt_stats_json JSONB NOT NULL DEFAULT '[]'::jsonb,
      last_event_sequence BIGINT,
      updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
    )
  `;
  await sql`CREATE INDEX IF NOT EXISTS idx_oa_trace_stats_service_updated ON oa_trace_stats(service_id, updated_at DESC)`;
  await sql`CREATE INDEX IF NOT EXISTS idx_oa_trace_stats_subject ON oa_trace_stats(subject_kind, subject_id)`;
  await sql`
    CREATE TABLE IF NOT EXISTS oa_trace_steps (
      scope_id TEXT NOT NULL,
      step_seq BIGINT NOT NULL,
      kind TEXT NOT NULL DEFAULT '',
      title TEXT NOT NULL DEFAULT '',
      status TEXT NOT NULL DEFAULT '',
      summary_lines JSONB NOT NULL DEFAULT '[]'::jsonb,
      raw_seqs JSONB NOT NULL DEFAULT '[]'::jsonb,
      source_event_id TEXT NOT NULL DEFAULT '',
      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
      PRIMARY KEY(scope_id, step_seq)
    )
  `;
  await sql`CREATE INDEX IF NOT EXISTS idx_oa_trace_steps_scope_updated ON oa_trace_steps(scope_id, updated_at DESC)`;
}

async function waitForSchema(): Promise<void> {
  let attempt = 0;
  const started = Date.now();
  while (true) {
    attempt += 1;
    try {
      await ensureSchema();
      databaseReady = true;
      databaseLastError = null;
      log("info", "database_ready", { attempt, elapsedMs: Date.now() - started });
      return;
    } catch (error) {
      databaseReady = false;
      databaseLastError = error instanceof Error ? error.message : String(error);
      log("warn", "database_init_retry", { attempt, error: errorToJson(error) });
      await Bun.sleep(Math.min(30_000, 500 * attempt));
    }
  }
}

function parseLimit(url: URL, fallback = 100, max = 500): number {
  const value = Number(url.searchParams.get("limit") ?? fallback);
  return Number.isInteger(value) && value > 0 ? Math.min(max, value) : fallback;
}

function parseTagsFromUrl(url: URL): string[] {
  const values = [...url.searchParams.getAll("tag"), ...url.searchParams.getAll("tags")];
  return normalizeTags(values.flatMap((value) => value.split(",")));
}

function parseScopeIdsFromUrl(url: URL): string[] {
  const values = [...url.searchParams.getAll("scopeId"), ...url.searchParams.getAll("scopeIds")];
  return Array.from(new Set(values.flatMap((value) => value.split(",")).map((value) => value.trim()).filter(Boolean)));
}

function eventMatchesTags(event: OaEventRecord, tags: string[]): boolean {
  if (tags.length === 0) return true;
  return tags.every((tag) => event.tags.includes(tag));
}

function eventSseChunk(event: OaEventRecord): string {
  return [
    `id: ${event.sequence}`,
    `event: ${event.type}`,
    `data: ${JSON.stringify(event)}`,
    "",
    "",
  ].join("\n");
}

function emitEvent(event: OaEventRecord): void {
  const chunk = eventSseChunk(event);
  for (const client of Array.from(eventClients)) {
    if (!eventMatchesTags(event, client.tags)) continue;
    try {
      client.enqueue(chunk);
    } catch {
      client.close();
    }
  }
}

function tagsFromEvent(event: OaEventRecord): Set<string> {
  return new Set(event.tags);
}

function tagValue(tags: Set<string>, prefix: string): string | null {
  const tag = Array.from(tags).find((item) => item.startsWith(prefix));
  return tag === undefined ? null : tag.slice(prefix.length);
}

function traceScopeFromScopeId(scopeId: string, serviceId: string): TraceScope {
  const taskAttempt = scopeId.match(/^task:([^:]+):attempt:(\d+)$/u);
  if (taskAttempt !== null) {
    const taskId = taskAttempt[1] ?? "";
    const attemptIndex = Number(taskAttempt[2] ?? 0);
    return { scopeId, serviceId, subjectKind: "task-attempt", subjectId: `${taskId}:${attemptIndex}`, taskId, attemptIndex };
  }
  const taskOnly = scopeId.match(/^task:([^:]+)$/u);
  if (taskOnly !== null) {
    const taskId = taskOnly[1] ?? "";
    return { scopeId, serviceId, subjectKind: "task", subjectId: taskId, taskId };
  }
  const [kind = "trace", ...rest] = scopeId.split(":");
  const id = rest.join(":") || scopeId;
  return { scopeId, serviceId, subjectKind: kind || "trace", subjectId: id };
}

function positiveInteger(value: unknown): number | null {
  const number = Number(value);
  return Number.isInteger(number) && number > 0 ? Math.floor(number) : null;
}

function traceScope(event: OaEventRecord): TraceScope | null {
  const tags = tagsFromEvent(event);
  const payload = event.payload;
  const taskId = stringValue(payload.taskId) || tagValue(tags, "task:") || "";
  const pipelineId = stringValue(payload.pipelineId) || tagValue(tags, "pipeline:") || "";
  const epochId = stringValue(payload.runId) || stringValue(payload.epochId) || tagValue(tags, "epoch:") || "";
  const explicitScopeId = stringValue(payload.scopeId);
  const serviceId = tagValue(tags, "service:") || event.sourceId;
  if (explicitScopeId.length > 0) return traceScopeFromScopeId(explicitScopeId, serviceId);
  if (taskId.length > 0) return traceScopeFromScopeId(`task:${taskId}`, serviceId);
  if (pipelineId.length > 0 && epochId.length > 0) return traceScopeFromScopeId(`epoch:${epochId}`, serviceId);
  if (pipelineId.length > 0) return traceScopeFromScopeId(`pipeline:${pipelineId}`, serviceId);
  return null;
}

function traceAttemptScope(event: OaEventRecord, primary: TraceScope | null): TraceScope | null {
  const tags = tagsFromEvent(event);
  const payload = event.payload;
  const serviceId = primary?.serviceId ?? tagValue(tags, "service:") ?? event.sourceId;
  const explicit = stringValue(payload.attemptScopeId);
  if (explicit.length > 0) return traceScopeFromScopeId(explicit, serviceId);
  const attemptIndex = positiveInteger(payload.attemptIndex) ?? positiveInteger(tagValue(tags, "attempt:"));
  const taskId = stringValue(payload.taskId) || primary?.taskId || (primary?.subjectKind === "task" ? primary.subjectId : "") || tagValue(tags, "task:") || "";
  if (taskId.length === 0 || attemptIndex === null) return null;
  return traceScopeFromScopeId(`task:${taskId}:attempt:${attemptIndex}`, serviceId);
}

function traceProjectionScopes(event: OaEventRecord): TraceScope[] {
  const primary = traceScope(event);
  if (primary === null) return [];
  if (event.type !== "trace-step-created" && event.type !== "trace-error") return [primary];
  const scopes = [primary];
  const attempt = traceAttemptScope(event, primary);
  if (attempt !== null && !scopes.some((scope) => scope.scopeId === attempt.scopeId)) scopes.push(attempt);
  return scopes;
}

function numberField(record: JsonRecord, keys: string[], fallback = 0): number {
  for (const key of keys) {
    const value = Number(record[key]);
    if (Number.isFinite(value) && value >= 0) return Math.floor(value);
  }
  return fallback;
}

function nestedRecord(record: JsonRecord, key: string): JsonRecord {
  const value = record[key];
  return asJsonRecord(value);
}

function normalizeStepKind(value: unknown): "read" | "edit" | "run" | "error" | "message" | "system" {
  const kind = stringValue(value).toLowerCase();
  if (["read", "explored", "explore", "grep", "search", "list", "view"].includes(kind)) return "read";
  if (["edit", "edited", "write", "patch", "diff", "apply"].includes(kind)) return "edit";
  if (["run", "ran", "command", "tool"].includes(kind)) return "run";
  if (kind === "error" || kind === "failed") return "error";
  if (kind === "message" || kind === "assistant" || kind === "user") return "message";
  return "system";
}

async function fetchTraceStats(executor: SqlExecutor, scopeId: string): Promise<TraceStatsRecord | null> {
  const rows = await executor<TraceStatsRow[]>`
    SELECT *
    FROM oa_trace_stats
    WHERE scope_id = ${scopeId}
    LIMIT 1
  `;
  return rows[0] === undefined ? null : statsRowToRecord(rows[0]);
}

function traceStatsEvent(stats: TraceStatsRecord, sourceEvent: OaEventRecord): OaEventRecord {
  const tags = ["stats", "trace", `service:${stats.serviceId}`, stats.scopeId];
  if (typeof stats.taskId === "string" && stats.taskId.length > 0) addTag(tags, `task:${stats.taskId}`);
  if (Number(stats.attemptIndex) > 0) addTag(tags, `attempt:${Number(stats.attemptIndex)}`);
  const payload: JsonRecord = {
    scopeId: stats.scopeId,
    serviceId: stats.serviceId,
    subjectKind: stats.subjectKind,
    subjectId: stats.subjectId,
    taskId: typeof stats.taskId === "string" ? stats.taskId : stats.subjectKind === "task" ? stats.subjectId : "",
    attemptIndex: Number(stats.attemptIndex) > 0 ? Number(stats.attemptIndex) : null,
    stats,
  };
  const event = normalizeEvent({
    eventId: `oa-event-flow:trace-stats-updated:${stats.scopeId}:${stats.statsRevision}`,
    type: "trace-stats-updated",
    sourceKind: "projection",
    sourceId: "oa-event-flow",
    aggregateType: stats.subjectKind,
    aggregateId: stats.subjectId,
    correlationId: sourceEvent.correlationId,
    causationId: sourceEvent.eventId,
    tags,
    payload,
  });
  return { ...event, sequence: sourceEvent.sequence, createdAt: sourceEvent.createdAt };
}

async function applyTraceSnapshot(executor: SqlExecutor, event: OaEventRecord, scope: TraceScope): Promise<TraceStatsRecord | null> {
  await executor`SELECT pg_advisory_xact_lock(482755, hashtext(${scope.scopeId}))`;
  const stats = nestedRecord(event.payload, "stats");
  const stepCount = numberField(stats, ["stepCount", "llmStepCount"]);
  const llmStepCount = numberField(stats, ["llmStepCount", "stepCount"], stepCount);
  const readCount = numberField(stats, ["readCount"]);
  const editCount = numberField(stats, ["editCount"]);
  const runCount = numberField(stats, ["runCount"]);
  const errorCount = numberField(stats, ["errorCount"]);
  const traceLineCount = numberField(stats, ["traceLineCount"], stepCount);
  const outputMaxSeq = numberField(stats, ["outputMaxSeq", "transcriptMaxSeq"]);
  const attemptStats = stats.attemptStats ?? stats.attempts ?? [];
  await executor`
    INSERT INTO oa_trace_stats (
      scope_id, service_id, subject_kind, subject_id, stats_revision,
      step_count, llm_step_count, read_count, edit_count, run_count, error_count,
      trace_line_count, output_max_seq, attempt_stats_json, last_event_sequence, updated_at
    )
    VALUES (
      ${scope.scopeId}, ${scope.serviceId}, ${scope.subjectKind}, ${scope.subjectId}, 1,
      ${stepCount}, ${llmStepCount}, ${readCount}, ${editCount}, ${runCount}, ${errorCount},
      ${traceLineCount}, ${outputMaxSeq}, ${executor.json(attemptStats)}, ${event.sequence}, now()
    )
    ON CONFLICT (scope_id) DO UPDATE SET
      service_id = EXCLUDED.service_id,
      subject_kind = EXCLUDED.subject_kind,
      subject_id = EXCLUDED.subject_id,
      stats_revision = oa_trace_stats.stats_revision + 1,
      step_count = EXCLUDED.step_count,
      llm_step_count = EXCLUDED.llm_step_count,
      read_count = EXCLUDED.read_count,
      edit_count = EXCLUDED.edit_count,
      run_count = EXCLUDED.run_count,
      error_count = EXCLUDED.error_count,
      trace_line_count = EXCLUDED.trace_line_count,
      output_max_seq = GREATEST(oa_trace_stats.output_max_seq, EXCLUDED.output_max_seq),
      attempt_stats_json = EXCLUDED.attempt_stats_json,
      last_event_sequence = GREATEST(COALESCE(oa_trace_stats.last_event_sequence, 0), EXCLUDED.last_event_sequence),
      updated_at = now()
  `;
  return fetchTraceStats(executor, scope.scopeId);
}

async function applyTraceStep(executor: SqlExecutor, event: OaEventRecord, scope: TraceScope): Promise<TraceStatsRecord | null> {
  await executor`SELECT pg_advisory_xact_lock(482755, hashtext(${scope.scopeId}))`;
  const payload = event.payload;
  const seq = numberField(payload, ["seq", "stepSeq", "outputSeq"], event.sequence);
  const outputMaxSeq = numberField(payload, ["outputMaxSeq", "seq", "stepSeq", "outputSeq"], seq);
  const kind = normalizeStepKind(payload.kind ?? payload.stepKind ?? payload.channel);
  const title = stringValue(payload.title, stringValue(payload.method, kind)).slice(0, 300);
  const status = stringValue(payload.status).slice(0, 120);
  const summaryLines = Array.isArray(payload.summaryLines) ? payload.summaryLines : [];
  const rawSeqs = Array.isArray(payload.rawSeqs) ? payload.rawSeqs : [seq];
  await executor`
    INSERT INTO oa_trace_steps (scope_id, step_seq, kind, title, status, summary_lines, raw_seqs, source_event_id, created_at, updated_at)
    VALUES (${scope.scopeId}, ${seq}, ${kind}, ${title}, ${status}, ${executor.json(summaryLines)}, ${executor.json(rawSeqs)}, ${event.eventId}, ${event.createdAt}, now())
    ON CONFLICT (scope_id, step_seq) DO UPDATE SET
      kind = EXCLUDED.kind,
      title = EXCLUDED.title,
      status = EXCLUDED.status,
      summary_lines = EXCLUDED.summary_lines,
      raw_seqs = EXCLUDED.raw_seqs,
      source_event_id = EXCLUDED.source_event_id,
      updated_at = now()
  `;
  const countRows = await executor<{
    step_count: string | number;
    read_count: string | number;
    edit_count: string | number;
    run_count: string | number;
    error_count: string | number;
  }[]>`
    SELECT
      COUNT(*) FILTER (WHERE kind <> 'system') AS step_count,
      COUNT(*) FILTER (WHERE kind = 'read') AS read_count,
      COUNT(*) FILTER (WHERE kind = 'edit') AS edit_count,
      COUNT(*) FILTER (WHERE kind = 'run') AS run_count,
      COUNT(*) FILTER (WHERE kind = 'error') AS error_count
    FROM oa_trace_steps
    WHERE scope_id = ${scope.scopeId}
  `;
  const counts = countRows[0] ?? { step_count: 0, read_count: 0, edit_count: 0, run_count: 0, error_count: 0 };
  const stepCount = Number(counts.step_count);
  const readCount = Number(counts.read_count);
  const editCount = Number(counts.edit_count);
  const runCount = Number(counts.run_count);
  const errorCount = Number(counts.error_count);
  await executor`
    INSERT INTO oa_trace_stats (
      scope_id, service_id, subject_kind, subject_id, stats_revision,
      step_count, llm_step_count, read_count, edit_count, run_count, error_count,
      trace_line_count, output_max_seq, attempt_stats_json, last_event_sequence, updated_at
    )
    VALUES (
      ${scope.scopeId}, ${scope.serviceId}, ${scope.subjectKind}, ${scope.subjectId}, 1,
      ${stepCount}, ${stepCount}, ${readCount}, ${editCount}, ${runCount}, ${errorCount},
      ${stepCount}, ${outputMaxSeq}, '[]'::jsonb, ${event.sequence}, now()
    )
    ON CONFLICT (scope_id) DO UPDATE SET
      service_id = EXCLUDED.service_id,
      subject_kind = EXCLUDED.subject_kind,
      subject_id = EXCLUDED.subject_id,
      stats_revision = oa_trace_stats.stats_revision + 1,
      step_count = EXCLUDED.step_count,
      llm_step_count = EXCLUDED.llm_step_count,
      read_count = EXCLUDED.read_count,
      edit_count = EXCLUDED.edit_count,
      run_count = EXCLUDED.run_count,
      error_count = EXCLUDED.error_count,
      trace_line_count = EXCLUDED.trace_line_count,
      output_max_seq = GREATEST(oa_trace_stats.output_max_seq, EXCLUDED.output_max_seq),
      attempt_stats_json = oa_trace_stats.attempt_stats_json,
      last_event_sequence = GREATEST(COALESCE(oa_trace_stats.last_event_sequence, 0), EXCLUDED.last_event_sequence),
      updated_at = now()
  `;
  return fetchTraceStats(executor, scope.scopeId);
}

async function applyTraceProjection(executor: SqlExecutor, event: OaEventRecord): Promise<TraceStatsRecord[]> {
  if (event.type === "trace-stats-updated") return [];
  const tags = tagsFromEvent(event);
  if (!tags.has("trace") && !event.type.startsWith("trace-")) return [];
  const scopes = traceProjectionScopes(event);
  if (scopes.length === 0) return [];
  if (event.type === "trace-stats-snapshot") {
    const stats = await applyTraceSnapshot(executor, event, scopes[0]);
    return stats === null ? [] : [stats];
  }
  if (event.type === "trace-step-created") {
    const records: TraceStatsRecord[] = [];
    for (const scope of scopes) {
      const stats = await applyTraceStep(executor, event, scope);
      if (stats !== null) records.push(stats);
    }
    return records;
  }
  if (event.type === "trace-error") {
    const stepEvent = { ...event, payload: { ...event.payload, kind: "error" } };
    const records: TraceStatsRecord[] = [];
    for (const scope of scopes) {
      const stats = await applyTraceStep(executor, stepEvent, scope);
      if (stats !== null) records.push(stats);
    }
    return records;
  }
  return [];
}

async function storeInputEvent(executor: SqlExecutor, event: OaEventRecord): Promise<OaEventRecord | null> {
  const rows = await executor<OaEventRow[]>`
    INSERT INTO oa_events (event_id, type, source_kind, source_id, aggregate_type, aggregate_id, correlation_id, causation_id, tags, payload, created_at)
    VALUES (${event.eventId}, ${event.type}, ${event.sourceKind}, ${event.sourceId}, ${event.aggregateType}, ${event.aggregateId}, ${event.correlationId}, ${event.causationId}, ${executor.json(event.tags)}, ${executor.json(event.payload)}, ${event.createdAt})
    ON CONFLICT (event_id) DO NOTHING
    RETURNING *
  `;
  return rows[0] === undefined ? null : rowToEvent(rows[0]);
}

async function publishEvents(inputs: OaEventInput[]): Promise<{ inserted: OaEventRecord[]; duplicates: number; derived: OaEventRecord[] }> {
  const inserted: OaEventRecord[] = [];
  const derived: OaEventRecord[] = [];
  let duplicates = 0;
  await sql.begin(async (tx: postgres.TransactionSql) => {
    for (const input of inputs) {
      const normalized = normalizeEvent(input);
      const stored = await storeInputEvent(tx, normalized);
      if (stored === null) {
        duplicates += 1;
        continue;
      }
      inserted.push(stored);
      for (const stats of await applyTraceProjection(tx, stored)) {
        derived.push(traceStatsEvent(stats, stored));
      }
    }
    const lastSequence = Math.max(0, ...inserted.map((event) => event.sequence), ...derived.map((event) => event.sequence));
    if (lastSequence > 0) {
      await tx`
        INSERT INTO oa_projection_offsets (projection_name, last_sequence, updated_at)
        VALUES ('trace-stats', ${lastSequence}, now())
        ON CONFLICT (projection_name) DO UPDATE SET last_sequence = GREATEST(oa_projection_offsets.last_sequence, EXCLUDED.last_sequence), updated_at = now()
      `;
    }
  });
  for (const event of inserted) emitEvent(event);
  for (const event of derived) emitEvent(event);
  return { inserted, duplicates, derived };
}

function arrayValue(value: unknown): unknown[] {
  return Array.isArray(value) ? value : [];
}

function nonNegativeInteger(value: unknown, fallback = 0): number {
  const number = Number(value);
  return Number.isFinite(number) && number >= 0 ? Math.floor(number) : fallback;
}

async function bridgeFetchJson(url: URL): Promise<unknown> {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), 20_000);
  try {
    const response = await fetch(url, {
      headers: { "accept": "application/json" },
      signal: controller.signal,
    });
    const text = await response.text();
    if (!response.ok) throw new Error(`pipeline bridge fetch failed ${response.status} ${response.statusText}: ${text.slice(0, 500)}`);
    return text.trim().length > 0 ? JSON.parse(text) as unknown : {};
  } finally {
    clearTimeout(timeout);
  }
}

function pipelineRunBridgeState(runId: string, pipelineId: string): PipelineBridgeRunState {
  const current = pipelineBridgeState.runs[runId];
  if (current !== undefined) {
    if (pipelineId.length > 0) current.pipelineId = pipelineId;
    return current;
  }
  const state: PipelineBridgeRunState = {
    runId,
    pipelineId,
    lastObservedEventCount: 0,
    lastSyncedAt: "",
    lastError: "",
    bridgeMode: "snapshot",
    insertedCount: 0,
    duplicateCount: 0,
  };
  pipelineBridgeState.runs[runId] = state;
  return state;
}

function pipelineBridgeEventTags(record: JsonRecord, runId: string, pipelineId: string): string[] {
  const tags = normalizeTags(record.tags);
  addTag(tags, "service:pipeline");
  addTag(tags, "pipeline.fact");
  if (pipelineId.length > 0) addTag(tags, `pipeline:${pipelineId}`);
  addTag(tags, `epoch:${runId}`);
  const nodeId = stringValue(record.nodeId) || stringValue(record.targetNodeId) || stringValue(record.sourceNodeId);
  const procedureRunId = stringValue(record.procedureRunId);
  const attemptId = stringValue(record.attemptId) || stringValue(record.attempt);
  if (nodeId.length > 0) addTag(tags, `node:${nodeId}`);
  if (procedureRunId.length > 0) addTag(tags, `procedure:${procedureRunId}`);
  if (attemptId.length > 0) addTag(tags, `attempt:${attemptId}`);
  return tags;
}

function pipelineDiagnosticsByRun(diagnostics: JsonRecord): Map<string, JsonRecord> {
  const map = new Map<string, JsonRecord>();
  for (const item of arrayValue(diagnostics.runs)) {
    const record = asRecord(item);
    const runId = stringValue(record?.runId);
    if (record !== null && runId.length > 0) map.set(runId, record as JsonRecord);
  }
  return map;
}

function compactPipelineRunSnapshot(run: JsonRecord): JsonRecord {
  const nodes = arrayValue(run.nodes).flatMap((item) => {
    const record = asRecord(item);
    if (record === null) return [];
    return [{
      nodeId: stringValue(record.nodeId),
      kind: stringValue(record.kind),
      status: stringValue(record.status),
      currentProcedureRunId: stringValue(record.currentProcedureRunId),
      procedureRunCount: arrayValue(record.procedureRunIds).length,
    }];
  });
  return {
    status: stringValue(run.status),
    error: stringValue(run.error),
    startedAt: stringValue(run.startedAt),
    finishedAt: stringValue(run.finishedAt),
    updatedAt: stringValue(run.updatedAt),
    durationMs: nonNegativeInteger(run.durationMs),
    nodeCount: nodes.length,
    procedureRunCount: arrayValue(run.procedureRuns).length,
    scorerCount: arrayValue(run.scorers).length,
    submissionCount: arrayValue(run.submissions).length,
    nodes,
  };
}

function pipelineSnapshotBridgeEventInput(run: JsonRecord, diagnostics: JsonRecord | null): OaEventInput {
  const runId = stringValue(run.runId);
  const pipelineId = stringValue(run.pipelineId);
  const compactRun = compactPipelineRunSnapshot(run);
  const compactDiagnostics: JsonRecord = diagnostics === null ? {} : {
    eventCount: nonNegativeInteger(diagnostics.eventCount),
    nodeFinishedCount: nonNegativeInteger(diagnostics.nodeFinishedCount),
    nodeFinishedWithPolicyCount: nonNegativeInteger(diagnostics.nodeFinishedWithPolicyCount),
    monitorAuditNodeFinishedCount: nonNegativeInteger(diagnostics.monitorAuditNodeFinishedCount),
    noAuditPolicyCount: nonNegativeInteger(diagnostics.noAuditPolicyCount),
    controlQueuedCount: nonNegativeInteger(diagnostics.controlQueuedCount),
    controlAppliedCount: nonNegativeInteger(diagnostics.controlAppliedCount),
  };
  const payload: JsonRecord = {
    runId,
    epochId: runId,
    pipelineId,
    taskPreview: stringValue(run.task).slice(0, 600),
    bridgeSource: "pipeline-snapshot",
    bridgeService: "oa-event-flow",
    run: compactRun,
    diagnostics: compactDiagnostics,
  };
  const tags = pipelineBridgeEventTags({ ...payload, event: "pipeline-run-snapshot" }, runId, pipelineId);
  addTag(tags, "pipeline.run");
  addTag(tags, `status:${stringValue(run.status, "unknown") || "unknown"}`);
  return {
    eventId: `pipeline:${runId}:run-snapshot:${stableEventHash(payload)}`,
    type: "pipeline-run-snapshot",
    createdAt: run.updatedAt || run.finishedAt || run.startedAt,
    sourceKind: "service",
    sourceId: "pipeline",
    aggregateType: pipelineId.length > 0 ? "pipeline" : "epoch",
    aggregateId: pipelineId.length > 0 ? pipelineId : runId,
    correlationId: `pipeline:${runId}`,
    causationId: null,
    tags,
    payload,
  };
}

async function syncPipelineRunSnapshot(run: JsonRecord, diagnostics: JsonRecord | null): Promise<void> {
  const runId = stringValue(run.runId);
  if (runId.length === 0) return;
  const pipelineId = stringValue(run.pipelineId);
  const state = pipelineRunBridgeState(runId, pipelineId);
  const result = await publishEvents([pipelineSnapshotBridgeEventInput(run, diagnostics)]);
  state.insertedCount += result.inserted.length;
  state.duplicateCount += result.duplicates;
  state.bridgeMode = "snapshot";
  state.lastObservedEventCount = Math.max(state.lastObservedEventCount, nonNegativeInteger(diagnostics?.eventCount));
  state.lastSyncedAt = new Date().toISOString();
  state.lastError = "";
  pipelineBridgeState.insertedCount += result.inserted.length;
  pipelineBridgeState.duplicateCount += result.duplicates;
  pipelineBridgeState.snapshotSyncCount += 1;
}

async function syncPipelineBridge(reason: string): Promise<void> {
  if (!pipelineBridgeState.enabled || pipelineBridgeState.inFlight || !databaseReady) return;
  pipelineBridgeState.inFlight = true;
  pipelineBridgeState.lastStartedAt = new Date().toISOString();
  pipelineBridgeState.syncCount += 1;
  try {
    const snapshotUrl = new URL(`${pipelineBridgeState.baseUrl}/api/snapshot`);
    snapshotUrl.searchParams.set("__unideskArrayLimit", `runs:${pipelineBridgeState.runLimit}`);
    const snapshot = asRecord(await bridgeFetchJson(snapshotUrl)) ?? {};
    let diagnosticsByRun = new Map<string, JsonRecord>();
    try {
      const diagnosticsUrl = new URL(`${pipelineBridgeState.baseUrl}/api/oa-event-flow/diagnostics`);
      const diagnostics = asJsonRecord(await bridgeFetchJson(diagnosticsUrl));
      diagnosticsByRun = pipelineDiagnosticsByRun(diagnostics);
    } catch (error) {
      log("warn", "pipeline_bridge_diagnostics_unavailable", { reason, error: errorToJson(error) });
    }
    const runs = arrayValue(snapshot.runs).flatMap((item) => {
      const record = asRecord(item);
      const runId = stringValue(record?.runId);
      if (record === null || runId.length === 0) return [];
      return [record as JsonRecord];
    }).slice(0, pipelineBridgeState.runLimit);
    for (const run of runs) {
      const runId = stringValue(run.runId);
      try {
        await syncPipelineRunSnapshot(run, diagnosticsByRun.get(runId) ?? null);
      } catch (error) {
        const state = pipelineRunBridgeState(runId, stringValue(run.pipelineId));
        state.lastError = error instanceof Error ? error.message : String(error);
        log("warn", "pipeline_bridge_run_failed", { reason, runId, error: errorToJson(error) });
      }
    }
    pipelineBridgeState.lastError = "";
    pipelineBridgeState.lastFinishedAt = new Date().toISOString();
    log("info", "pipeline_bridge_synced", { reason, runCount: runs.length, insertedCount: pipelineBridgeState.insertedCount, duplicateCount: pipelineBridgeState.duplicateCount });
  } catch (error) {
    pipelineBridgeState.lastError = error instanceof Error ? error.message : String(error);
    pipelineBridgeState.lastFinishedAt = new Date().toISOString();
    log("warn", "pipeline_bridge_failed", { reason, error: errorToJson(error) });
  } finally {
    pipelineBridgeState.inFlight = false;
  }
}

function startPipelineBridge(): void {
  if (!pipelineBridgeState.enabled) {
    log("info", "pipeline_bridge_disabled");
    return;
  }
  const run = (reason: string) => {
    void syncPipelineBridge(reason).catch((error) => log("warn", "pipeline_bridge_unhandled_error", { reason, error: errorToJson(error) }));
  };
  setTimeout(() => run("startup"), 2_000).unref?.();
  pipelineBridgeTimer = setInterval(() => run("interval"), pipelineBridgeState.intervalMs);
  pipelineBridgeTimer.unref?.();
  log("info", "pipeline_bridge_started", {
    baseUrl: pipelineBridgeState.baseUrl,
    intervalMs: pipelineBridgeState.intervalMs,
    runLimit: pipelineBridgeState.runLimit,
    mode: "snapshot",
  });
}

async function readJson(req: Request): Promise<unknown> {
  const text = await req.text();
  if (text.trim().length === 0) return {};
  return JSON.parse(text) as unknown;
}

function eventsFromBody(body: unknown): OaEventInput[] {
  const record = asRecord(body);
  if (Array.isArray(body)) return body.map((item) => asRecord(item) ?? {}) as OaEventInput[];
  if (record !== null && Array.isArray(record.events)) return record.events.map((item) => asRecord(item) ?? {}) as OaEventInput[];
  return [record ?? {} as OaEventInput];
}

async function publishEventsRoute(req: Request): Promise<Response> {
  const inputs = eventsFromBody(await readJson(req));
  if (inputs.length === 0) return jsonResponse({ ok: false, error: "events must not be empty" }, 400);
  if (inputs.length > 200) return jsonResponse({ ok: false, error: "too many events", maxEvents: 200 }, 413);
  const result = await publishEvents(inputs);
  return jsonResponse({ ok: true, inserted: result.inserted, duplicates: result.duplicates, derived: result.derived });
}

function taskIdFromTraceScopeId(scopeId: string): string {
  const match = scopeId.match(/^task:([^:]+)(?::attempt:\d+)?$/u);
  return match?.[1] ?? "";
}

async function rebuildTraceProjectionRoute(url: URL): Promise<Response> {
  const scopeId = stringValue(url.searchParams.get("scopeId"));
  const taskId = taskIdFromTraceScopeId(scopeId);
  if (scopeId.length === 0 || taskId.length === 0) return jsonResponse({ ok: false, error: "scopeId must be a task scope, e.g. task:<taskId>" }, 400);
  const taskScopeId = `task:${taskId}`;
  const attemptPrefix = `${taskScopeId}:attempt:`;
  const replayedStats = new Map<string, TraceStatsRecord>();
  let eventCount = 0;
  await sql.begin(async (tx: postgres.TransactionSql) => {
    await tx`SELECT pg_advisory_xact_lock(482756, hashtext(${taskScopeId}))`;
    await tx`DELETE FROM oa_trace_steps WHERE scope_id = ${taskScopeId} OR left(scope_id, ${attemptPrefix.length}) = ${attemptPrefix}`;
    await tx`DELETE FROM oa_trace_stats WHERE scope_id = ${taskScopeId} OR left(scope_id, ${attemptPrefix.length}) = ${attemptPrefix}`;
    const rows = await tx<OaEventRow[]>`
      SELECT *
      FROM oa_events
      WHERE type <> 'trace-stats-updated'
        AND (tags @> ${tx.json([`task:${taskId}`])}::jsonb OR payload->>'taskId' = ${taskId})
        AND (tags @> ${tx.json(["trace"])}::jsonb OR type LIKE 'trace-%')
      ORDER BY sequence ASC
    `;
    eventCount = rows.length;
    for (const row of rows) {
      const event = rowToEvent(row);
      for (const stats of await applyTraceProjection(tx, event)) replayedStats.set(stats.scopeId, stats);
    }
  });
  const stats = Array.from(replayedStats.values()).sort((left, right) => left.scopeId.localeCompare(right.scopeId));
  log("info", "trace_projection_rebuilt", { scopeId: taskScopeId, taskId, eventCount, statsCount: stats.length });
  return jsonResponse({ ok: true, scopeId: taskScopeId, taskId, eventCount, statsCount: stats.length, stats });
}

async function queryEvents(url: URL): Promise<OaEventRecord[]> {
  const tags = parseTagsFromUrl(url);
  const type = stringValue(url.searchParams.get("type"));
  const sourceId = stringValue(url.searchParams.get("sourceId"));
  const afterSeq = Math.max(0, Number(url.searchParams.get("afterSeq") ?? 0) || 0);
  const limit = parseLimit(url);
  if (tags.length > 0 && type.length > 0 && sourceId.length > 0) {
    const rows = await sql<OaEventRow[]>`
      SELECT * FROM oa_events
      WHERE sequence > ${afterSeq} AND tags @> ${sql.json(tags)}::jsonb AND type = ${type} AND source_id = ${sourceId}
      ORDER BY sequence ASC
      LIMIT ${limit}
    `;
    return rows.map(rowToEvent);
  }
  if (tags.length > 0 && type.length > 0) {
    const rows = await sql<OaEventRow[]>`
      SELECT * FROM oa_events
      WHERE sequence > ${afterSeq} AND tags @> ${sql.json(tags)}::jsonb AND type = ${type}
      ORDER BY sequence ASC
      LIMIT ${limit}
    `;
    return rows.map(rowToEvent);
  }
  if (tags.length > 0 && sourceId.length > 0) {
    const rows = await sql<OaEventRow[]>`
      SELECT * FROM oa_events
      WHERE sequence > ${afterSeq} AND tags @> ${sql.json(tags)}::jsonb AND source_id = ${sourceId}
      ORDER BY sequence ASC
      LIMIT ${limit}
    `;
    return rows.map(rowToEvent);
  }
  if (tags.length > 0) {
    const rows = await sql<OaEventRow[]>`
      SELECT * FROM oa_events
      WHERE sequence > ${afterSeq} AND tags @> ${sql.json(tags)}::jsonb
      ORDER BY sequence ASC
      LIMIT ${limit}
    `;
    return rows.map(rowToEvent);
  }
  if (type.length > 0 && sourceId.length > 0) {
    const rows = await sql<OaEventRow[]>`
      SELECT * FROM oa_events
      WHERE sequence > ${afterSeq} AND type = ${type} AND source_id = ${sourceId}
      ORDER BY sequence ASC
      LIMIT ${limit}
    `;
    return rows.map(rowToEvent);
  }
  if (type.length > 0) {
    const rows = await sql<OaEventRow[]>`
      SELECT * FROM oa_events
      WHERE sequence > ${afterSeq} AND type = ${type}
      ORDER BY sequence ASC
      LIMIT ${limit}
    `;
    return rows.map(rowToEvent);
  }
  if (sourceId.length > 0) {
    const rows = await sql<OaEventRow[]>`
      SELECT * FROM oa_events
      WHERE sequence > ${afterSeq} AND source_id = ${sourceId}
      ORDER BY sequence ASC
      LIMIT ${limit}
    `;
    return rows.map(rowToEvent);
  }
  const rows = await sql<OaEventRow[]>`
    SELECT * FROM oa_events
    WHERE sequence > ${afterSeq}
    ORDER BY sequence ASC
    LIMIT ${limit}
  `;
  return rows.map(rowToEvent);
}

async function eventsRoute(url: URL): Promise<Response> {
  const events = await queryEvents(url);
  return jsonResponse({ ok: true, events, returned: events.length, nextAfterSeq: events.at(-1)?.sequence ?? Number(url.searchParams.get("afterSeq") ?? 0) });
}

async function eventStreamRoute(url: URL): Promise<Response> {
  const tags = parseTagsFromUrl(url);
  const afterSeq = Math.max(0, Number(url.searchParams.get("afterSeq") ?? 0) || 0);
  let heartbeat: ReturnType<typeof setInterval> | null = null;
  let client: EventClient | null = null;
  const stream = new ReadableStream<Uint8Array>({
    async start(controller) {
      const id = `client_${randomUUID()}`;
      client = {
        id,
        tags,
        connectedAt: new Date().toISOString(),
        enqueue(chunk: string) {
          controller.enqueue(eventEncoder.encode(chunk));
        },
        close() {
          if (heartbeat !== null) {
            clearInterval(heartbeat);
            heartbeat = null;
          }
          if (client !== null) eventClients.delete(client);
          client = null;
          try {
            controller.close();
          } catch {
            // The browser may have already closed the EventSource.
          }
        },
      };
      eventClients.add(client);
      controller.enqueue(eventEncoder.encode("retry: 1000\n\n"));
      client.enqueue([
        "event: hello",
        `data: ${JSON.stringify({ ok: true, service: "oa-event-flow", tags, afterSeq, at: new Date().toISOString() })}`,
        "",
        "",
      ].join("\n"));
      const backlogUrl = new URL("http://oa-event-flow.local/api/events");
      backlogUrl.searchParams.set("afterSeq", String(afterSeq));
      backlogUrl.searchParams.set("limit", "200");
      if (tags.length > 0) backlogUrl.searchParams.set("tags", tags.join(","));
      for (const event of await queryEvents(backlogUrl)) client.enqueue(eventSseChunk(event));
      heartbeat = setInterval(() => {
        try {
          controller.enqueue(eventEncoder.encode(`: heartbeat ${new Date().toISOString()}\n\n`));
        } catch {
          client?.close();
        }
      }, 25_000);
      heartbeat.unref?.();
    },
    cancel() {
      client?.close();
    },
  });
  return new Response(stream, {
    headers: {
      "content-type": "text/event-stream; charset=utf-8",
      "cache-control": "no-store, no-transform",
      "connection": "keep-alive",
      "x-accel-buffering": "no",
    },
  });
}

async function traceStatsRoute(url: URL): Promise<Response> {
  const scopeIds = parseScopeIdsFromUrl(url);
  const tags = parseTagsFromUrl(url);
  const limit = parseLimit(url, 100, 1000);
  if (scopeIds.length > 0) {
    const rows = await sql<TraceStatsRow[]>`
      SELECT *
      FROM oa_trace_stats
      WHERE scope_id IN ${sql(scopeIds)}
      ORDER BY updated_at DESC
      LIMIT ${limit}
    `;
    const stats = rows.map(statsRowToRecord);
    return jsonResponse({ ok: true, stats, stat: stats[0] ?? null, returned: stats.length });
  }
  const serviceTag = tags.find((tag) => tag.startsWith("service:")) ?? "";
  const serviceId = serviceTag.length > 0 ? serviceTag.slice("service:".length) : stringValue(url.searchParams.get("serviceId"));
  if (serviceId.length > 0) {
    const rows = await sql<TraceStatsRow[]>`
      SELECT *
      FROM oa_trace_stats
      WHERE service_id = ${serviceId}
      ORDER BY updated_at DESC
      LIMIT ${limit}
    `;
    const stats = rows.map(statsRowToRecord);
    return jsonResponse({ ok: true, stats, returned: stats.length });
  }
  const rows = await sql<TraceStatsRow[]>`
    SELECT *
    FROM oa_trace_stats
    ORDER BY updated_at DESC
    LIMIT ${limit}
  `;
  const stats = rows.map(statsRowToRecord);
  return jsonResponse({ ok: true, stats, returned: stats.length });
}

async function diagnosticsRoute(): Promise<Response> {
  const [eventCountRows, latestRows, statsCountRows, typeRows, recentStatsRows, offsetRows] = await Promise.all([
    sql<Array<{ count: string | number }>>`SELECT COUNT(*) AS count FROM oa_events`,
    sql<OaEventRow[]>`SELECT * FROM oa_events ORDER BY sequence DESC LIMIT 20`,
    sql<Array<{ count: string | number }>>`SELECT COUNT(*) AS count FROM oa_trace_stats`,
    sql<Array<{ type: string; count: string | number }>>`SELECT type, COUNT(*) AS count FROM oa_events GROUP BY type ORDER BY count DESC, type ASC LIMIT 20`,
    sql<TraceStatsRow[]>`SELECT * FROM oa_trace_stats ORDER BY updated_at DESC LIMIT 20`,
    sql<Array<{ projection_name: string; last_sequence: string | number; updated_at: Date | string }>>`SELECT * FROM oa_projection_offsets ORDER BY projection_name ASC`,
  ]);
  return jsonResponse({
    ok: true,
    service: "oa-event-flow",
    startedAt: serviceStartedAt,
    databaseReady,
    databaseLastError,
    eventCount: Number(eventCountRows[0]?.count ?? 0),
    traceStatsCount: Number(statsCountRows[0]?.count ?? 0),
    latestSequence: Number(latestRows[0]?.sequence ?? 0),
    eventTypes: (typeRows as Array<{ type: string; count: string | number }>).map((row) => ({ type: row.type, count: Number(row.count) })),
    projectionOffsets: (offsetRows as Array<{ projection_name: string; last_sequence: string | number; updated_at: Date | string }>).map((row) => ({ name: row.projection_name, lastSequence: Number(row.last_sequence), updatedAt: isoTimestamp(row.updated_at) })),
    sseClients: Array.from(eventClients).map((client) => ({ id: client.id, tags: client.tags, connectedAt: client.connectedAt })),
    recentEvents: latestRows.map(rowToEvent),
    recentTraceStats: recentStatsRows.map(statsRowToRecord),
    pipelineBridge: pipelineBridgeState,
  });
}

async function healthRoute(): Promise<Response> {
  const [eventCountRows, statsCountRows, latestRows] = databaseReady
    ? await Promise.all([
        sql<Array<{ count: string | number }>>`SELECT COUNT(*) AS count FROM oa_events`,
        sql<Array<{ count: string | number }>>`SELECT COUNT(*) AS count FROM oa_trace_stats`,
        sql<Array<{ sequence: string | number }>>`SELECT sequence FROM oa_events ORDER BY sequence DESC LIMIT 1`,
      ])
    : [[{ count: 0 }], [{ count: 0 }], [{ sequence: 0 }]];
  return jsonResponse({
    ok: databaseReady,
    service: "oa-event-flow",
    startedAt: serviceStartedAt,
    databaseReady,
    databaseLastError,
    eventCount: Number(eventCountRows[0]?.count ?? 0),
    traceStatsCount: Number(statsCountRows[0]?.count ?? 0),
    latestSequence: Number(latestRows[0]?.sequence ?? 0),
    sseClientCount: eventClients.size,
    pipelineBridge: {
      enabled: pipelineBridgeState.enabled,
      mode: pipelineBridgeState.mode,
      inFlight: pipelineBridgeState.inFlight,
      lastFinishedAt: pipelineBridgeState.lastFinishedAt,
      lastError: pipelineBridgeState.lastError,
      insertedCount: pipelineBridgeState.insertedCount,
      duplicateCount: pipelineBridgeState.duplicateCount,
    },
    deploy: {
      serviceId: config.deployServiceId,
      ref: config.deployRef,
      repo: config.deployRepo,
      commit: config.deployCommit,
      requestedCommit: config.deployRequestedCommit,
    },
  }, databaseReady ? 200 : 503);
}

async function route(req: Request): Promise<Response> {
  const url = new URL(req.url);
  if (req.method === "OPTIONS") return jsonResponse({ ok: true });
  try {
    if (url.pathname === "/" || url.pathname === "/health") return await healthRoute();
    if (url.pathname === "/logs" && req.method === "GET") return jsonResponse({ ok: true, logs: recentLogs.slice(-parseLimit(url, 100, 500)) });
    if (url.pathname === "/api/events" && req.method === "POST") return await publishEventsRoute(req);
    if (url.pathname === "/api/events" && req.method === "GET") return await eventsRoute(url);
    if (url.pathname === "/api/events/stream" && req.method === "GET") return await eventStreamRoute(url);
    if (url.pathname === "/api/stats/trace" && req.method === "GET") return await traceStatsRoute(url);
    if (url.pathname === "/api/projections/trace/rebuild" && (req.method === "GET" || req.method === "POST")) return await rebuildTraceProjectionRoute(url);
    if (url.pathname === "/api/diagnostics" && req.method === "GET") return await diagnosticsRoute();
    return jsonResponse({ ok: false, error: "not found", path: url.pathname }, 404);
  } catch (error) {
    return errorResponse(error);
  }
}

await waitForSchema();
Bun.serve({ hostname: config.host, port: config.port, idleTimeout: 120, fetch: route });
log("info", "service_started", { port: config.port });
startPipelineBridge();
