// Timer/on-demand web->local sync loop. This keeps the local mirror warm by running the
// existing bootstrap -> inbox -> CRM pull path on a cadence, with per-source cursors and health rows
// the Swift Settings UI can read from SQLite.

import { DatabaseSync } from "node:sqlite";
import { execSync } from "node:child_process";
import fs from "node:fs";
import path from "node:path";
import * as websync from "./webClient.ts";
import { storeConversations, type WebConversation } from "./pull.ts";
import { storeCrm, type WebContact, type WebCompany, type WebDeal } from "./crm.ts";
import {
  markOutboundDraftAcked,
  pendingOutboundDraftAcks,
  storeOutboundDrafts,
  type WebOutboundDraft,
} from "./outboundDrafts.ts";
import { selectedWorkspaceId, storeBootstrapWorkspaces } from "./workspace.ts";

export type SyncSource = "bootstrap" | "inbox" | "outbound_drafts" | "crm";
type SourceStatus = "ok" | "skipped" | "waiting_for_web" | "error";

export interface SyncDeps {
  companionToken(): string | null;
  bootstrap(): Promise<websync.BootstrapTree>;
  fetchConversations(opts?: { integrationId?: string; limit?: number; since?: number; workspaceId?: string }): Promise<WebConversation[]>;
  fetchOutboundDrafts(opts?: { limit?: number; since?: number }): Promise<WebOutboundDraft[]>;
  ackOutboundDraft(id: string, body: { status: string; providerMsgId?: string; error?: string | null }): Promise<{ id: string; status: string }>;
  fetchCrm(opts?: { since?: number }): Promise<{ contacts: WebContact[]; companies: WebCompany[]; deals: WebDeal[] }>;
  now(): number;
}

export interface SourceResult {
  source: SyncSource;
  status: SourceStatus;
  cursor: number;
  counts: Record<string, unknown>;
  error?: string;
}

export interface SyncRunResult {
  ok: boolean;
  skipped?: string;
  sources: SourceResult[];
}

const SOURCES: SyncSource[] = ["bootstrap", "inbox", "outbound_drafts", "crm"];
const INBOX_PAGE = 25;
const INBOX_LEGACY_MAX_PAGES = 8;
const INBOX_SYNC_MAX_PAGES = 40;
const STALE_RUNNING_MS = 10 * 60_000;
const SOURCE_META: Record<SyncSource, Record<string, unknown>> = {
  bootstrap: {
    direction: "web_to_local",
    capability: "Companion workspace and integration bootstrap",
    target: "web-companion-auth",
    path: websync.endpointManifest.bootstrap[0],
  },
  inbox: {
    direction: "web_to_local",
    capability: "Unipile/web conversations and recent messages",
    target: "web-companion-conversations",
    paths: websync.endpointManifest.conversations,
  },
  outbound_drafts: {
    direction: "web_to_local",
    capability: "DewLink local-device outbound drafts for approval",
    target: "web-companion-conversations",
    executionTarget: "local-device",
    requiresOnlineDevice: true,
    paths: websync.endpointManifest.outboundDrafts,
    ackPaths: websync.endpointManifest.outboundDraftStatus,
  },
  crm: {
    direction: "web_to_local",
    capability: "CRM contacts, companies, and deals",
    target: "web-crm-companion",
    paths: [
      websync.endpointManifest.crm.contacts(),
      websync.endpointManifest.crm.companies(),
      websync.endpointManifest.crm.deals(),
    ],
  },
};

const defaultDeps: SyncDeps = {
  companionToken: websync.companionToken,
  bootstrap: websync.bootstrap,
  fetchConversations: websync.fetchConversations,
  fetchOutboundDrafts: websync.fetchOutboundDrafts,
  ackOutboundDraft: websync.ackOutboundDraft,
  fetchCrm: websync.fetchCrm,
  now: () => Date.now(),
};

let backgroundRunning = false;
let backgroundStarted = false;
const engineRoot = path.resolve(path.dirname(new URL(import.meta.url).pathname), "../../..");

function packageVersion(): string {
  try {
    const pkg = JSON.parse(fs.readFileSync(path.join(engineRoot, "package.json"), "utf8"));
    return typeof pkg.version === "string" && pkg.version.trim() ? pkg.version.trim() : "unknown";
  } catch {
    return "unknown";
  }
}

function gitSha(): string | null {
  try {
    return execSync("git rev-parse --short HEAD", {
      cwd: engineRoot,
      encoding: "utf8",
      stdio: ["ignore", "pipe", "ignore"],
    }).trim() || null;
  } catch {
    return null;
  }
}

export function ensureSyncLoopSchema(db: DatabaseSync) {
  db.exec(`
    CREATE TABLE IF NOT EXISTS _sync_state (
      table_name  TEXT PRIMARY KEY,
      last_pulled INTEGER NOT NULL DEFAULT 0,
      last_pushed INTEGER NOT NULL DEFAULT 0
    );
    CREATE TABLE IF NOT EXISTS sync_health (
      source           TEXT PRIMARY KEY,
      status           TEXT NOT NULL,
      last_started_at  INTEGER,
      last_finished_at INTEGER,
      cursor           INTEGER NOT NULL DEFAULT 0,
      counts_json      TEXT NOT NULL DEFAULT '{}',
      last_error       TEXT,
      updated_at       INTEGER NOT NULL
    );
    CREATE TABLE IF NOT EXISTS sync_command (
      id           TEXT PRIMARY KEY,
      workspace_id TEXT NOT NULL DEFAULT 'local',
      kind         TEXT NOT NULL DEFAULT 'web_pull',
      status       TEXT NOT NULL DEFAULT 'pending',
      source       TEXT NOT NULL DEFAULT 'ui',
      requested_at INTEGER NOT NULL,
      started_at   INTEGER,
      finished_at  INTEGER,
      error        TEXT
    );
    CREATE INDEX IF NOT EXISTS idx_sync_command_status ON sync_command(status, requested_at);
    CREATE TABLE IF NOT EXISTS engine_info (
      id         TEXT PRIMARY KEY DEFAULT 'engine',
      version    TEXT NOT NULL,
      git_sha    TEXT,
      started_at INTEGER NOT NULL,
      updated_at INTEGER NOT NULL
    );
  `);
}

export function recordEngineInfo(
  db: DatabaseSync,
  info: { version?: string; gitSha?: string | null; startedAt?: number; now?: number } = {},
) {
  ensureSyncLoopSchema(db);
  const now = Math.floor(info.now ?? Date.now());
  const startedAt = Math.floor(info.startedAt ?? now);
  db.prepare(
    `INSERT INTO engine_info(id, version, git_sha, started_at, updated_at)
     VALUES ('engine', ?, ?, ?, ?)
     ON CONFLICT(id) DO UPDATE SET
       version=excluded.version,
       git_sha=excluded.git_sha,
       started_at=excluded.started_at,
       updated_at=excluded.updated_at`,
  ).run(info.version ?? packageVersion(), info.gitSha ?? gitSha(), startedAt, now);
}

export function syncHealth(db: DatabaseSync) {
  ensureSyncLoopSchema(db);
  return db.prepare(
    `SELECT source, status, last_started_at, last_finished_at, cursor, counts_json, last_error, updated_at
     FROM sync_health
     ORDER BY CASE source WHEN 'bootstrap' THEN 0 WHEN 'inbox' THEN 1 WHEN 'outbound_drafts' THEN 2 WHEN 'crm' THEN 3 WHEN 'dewlink_ingest' THEN 4 ELSE 9 END`,
  ).all();
}

export function resetStaleRunning(db: DatabaseSync, now: number) {
  ensureSyncLoopSchema(db);
  db.prepare(
    `UPDATE sync_health
     SET status='error',
         last_error='interrupted (engine restart or stale run)',
         updated_at=?
     WHERE status='running' AND last_started_at < ?`,
  ).run(now, now - STALE_RUNNING_MS);
}

export function resetRunningOnBoot(db: DatabaseSync, now: number = Date.now()) {
  ensureSyncLoopSchema(db);
  db.prepare(
    `UPDATE sync_health
     SET status='error',
         last_error='interrupted (engine restart or stale run)',
         updated_at=?
     WHERE status='running'`,
  ).run(now);
}

function withSourceMeta(source: SyncSource, counts: Record<string, unknown> = {}): Record<string, unknown> {
  return { ...SOURCE_META[source], ...counts };
}

export function queueSyncNow(db: DatabaseSync, source = "mcp") {
  ensureSyncLoopSchema(db);
  const now = Date.now();
  const id = `sync:${now}:${Math.random().toString(36).slice(2, 8)}`;
  const ws = selectedWorkspaceId(db);
  db.prepare(
    `INSERT INTO sync_command (id, workspace_id, kind, status, source, requested_at)
     VALUES (?, ?, 'web_pull', 'pending', ?, ?)`,
  ).run(id, ws, source, now);
  return { queued: true, command_id: id, requested_at: now };
}

function cursorFor(db: DatabaseSync, source: SyncSource): number {
  ensureSyncLoopSchema(db);
  const row = db.prepare("SELECT last_pulled FROM _sync_state WHERE table_name=?").get(`sync:${source}`) as any;
  return Number(row?.last_pulled) || 0;
}

function setCursor(db: DatabaseSync, source: SyncSource, cursor: number) {
  db.prepare(
    `INSERT INTO _sync_state(table_name, last_pulled, last_pushed)
     VALUES (?, ?, 0)
     ON CONFLICT(table_name) DO UPDATE SET last_pulled=excluded.last_pulled`,
  ).run(`sync:${source}`, Math.max(0, Math.floor(cursor)));
}

function startSource(db: DatabaseSync, source: SyncSource, now: number) {
  db.prepare(
    `INSERT INTO sync_health(source, status, last_started_at, cursor, counts_json, updated_at)
     VALUES (?, 'running', ?, COALESCE((SELECT last_pulled FROM _sync_state WHERE table_name=?), 0), '{}', ?)
     ON CONFLICT(source) DO UPDATE SET
       status='running',
       last_started_at=excluded.last_started_at,
       last_error=NULL,
       updated_at=excluded.updated_at`,
  ).run(source, now, `sync:${source}`, now);
}

function finishSource(
  db: DatabaseSync,
  result: SourceResult,
  finishedAt: number,
) {
  db.prepare(
    `INSERT INTO sync_health
       (source, status, last_started_at, last_finished_at, cursor, counts_json, last_error, updated_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?)
     ON CONFLICT(source) DO UPDATE SET
       status=excluded.status,
       last_finished_at=excluded.last_finished_at,
       cursor=excluded.cursor,
       counts_json=excluded.counts_json,
       last_error=excluded.last_error,
       updated_at=excluded.updated_at`,
  ).run(
    result.source,
    result.status,
    finishedAt,
    finishedAt,
    result.cursor,
    JSON.stringify(result.counts),
    result.error ?? null,
    finishedAt,
  );
}

export function classifyError(e: unknown): { status: SourceStatus; error: string } {
  const error = e instanceof Error ? e.message : String(e);
  if (/\bweb (404|501)\b/.test(error) || /not live|not deployed/i.test(error)) {
    return { status: "waiting_for_web", error };
  }
  return { status: "error", error };
}

function maxConversationCursor(convs: WebConversation[], fallback: number): number {
  let max = fallback;
  for (const c of convs) {
    max = Math.max(max, Number(c.lastTs) || 0);
    for (const m of c.messages ?? []) max = Math.max(max, Number(m.ts) || 0);
  }
  return max;
}

function maxOutboundDraftCursor(drafts: WebOutboundDraft[], fallback: number, now: number): number {
  let max = fallback;
  for (const draft of drafts) {
    max = Math.max(max, Number(draft.updatedAt) || 0, Number(draft.createdAt) || 0);
  }
  return drafts.length > 0 && max === fallback ? now : max;
}

function maxCrmCursor(
  crm: { contacts: WebContact[]; companies: WebCompany[]; deals: WebDeal[] },
  fallback: number,
  now: number,
): number {
  let max = fallback;
  let count = 0;
  for (const row of [...crm.contacts, ...crm.companies, ...crm.deals] as Array<{ updatedAt?: number | string | null }>) {
    count++;
    max = Math.max(max, Number(row.updatedAt) || 0);
  }
  return count > 0 && max === fallback ? now : max;
}

export async function runIncrementalSync(
  db: DatabaseSync,
  deps: SyncDeps | undefined = defaultDeps,
  opts: { inboxLimit?: number } = {},
): Promise<SyncRunResult> {
  const activeDeps = deps ?? defaultDeps;
  ensureSyncLoopSchema(db);
  const startedAt = activeDeps.now();
  resetStaleRunning(db, startedAt);
  const token = activeDeps.companionToken();
  if (!token) {
    const sources = SOURCES.map((source): SourceResult => ({
      source,
      status: "skipped",
      cursor: cursorFor(db, source),
      counts: withSourceMeta(source),
      error: "not signed in",
    }));
    for (const r of sources) finishSource(db, r, startedAt);
    return { ok: true, skipped: "not signed in", sources };
  }

  const sources: SourceResult[] = [];
  let workspaceId = selectedWorkspaceId(db);

  for (const source of SOURCES) {
    const now = activeDeps.now();
    startSource(db, source, now);
    const cursor = cursorFor(db, source);
    try {
      if (source === "bootstrap") {
        const tree = await activeDeps.bootstrap();
        const integrations = websync.flattenIntegrations(tree);
        workspaceId = storeBootstrapWorkspaces(db, tree).selectedWorkspaceId;
        const next = activeDeps.now();
        setCursor(db, source, next);
        const result: SourceResult = {
          source,
          status: "ok",
          cursor: next,
          counts: withSourceMeta(source, { organizations: tree.organizations.length, integrations: integrations.length }),
        };
        finishSource(db, result, next);
        sources.push(result);
      } else if (source === "inbox") {
        const pageSize = opts.inboxLimit ?? INBOX_PAGE;
        let pageCursor = cursor;
        let committedCursor = cursor;
        let pages = 0;
        let maxPages = INBOX_LEGACY_MAX_PAGES;
        const totalCounts: Record<string, number> = {};
        try {
          for (let page = 0; page < maxPages; page++) {
            workspaceId = selectedWorkspaceId(db);
            const convs = await activeDeps.fetchConversations({
              limit: pageSize,
              since: pageCursor || undefined,
              workspaceId,
            });
            const routeKind = websync.conversationPageRouteKind(convs);
            if (routeKind === "sync") {
              maxPages = INBOX_SYNC_MAX_PAGES;
              totalCounts.syncEndpointPages = (totalCounts.syncEndpointPages ?? 0) + 1;
            } else {
              totalCounts.legacyEndpointPages = (totalCounts.legacyEndpointPages ?? 0) + 1;
            }
            const counts = storeConversations(db, convs, { workspaceId }) as Record<string, number>;
            for (const [key, value] of Object.entries(counts)) totalCounts[key] = (totalCounts[key] ?? 0) + Number(value || 0);
            const next = maxConversationCursor(convs, pageCursor);
            setCursor(db, source, next);
            committedCursor = next;
            pages++;
            if (next === pageCursor) break;
            pageCursor = next;
            if (convs.length < pageSize) break;
          }
        } catch (e) {
          if (pages > 0) {
            const classified = classifyError(e);
            const result: SourceResult = {
              source,
              status: classified.status,
              cursor: committedCursor,
              counts: withSourceMeta(source, { ...totalCounts, pages, partial: true }),
              error: classified.error,
            };
            finishSource(db, result, activeDeps.now());
            sources.push(result);
            continue;
          }
          throw e;
        }
        const result: SourceResult = {
          source,
          status: "ok",
          cursor: committedCursor,
          counts: withSourceMeta(source, { ...totalCounts, pages }),
        };
        finishSource(db, result, activeDeps.now());
        sources.push(result);
      } else if (source === "outbound_drafts") {
        const drafts = await activeDeps.fetchOutboundDrafts({ limit: 50, since: cursor || undefined });
        workspaceId = selectedWorkspaceId(db);
        const counts = storeOutboundDrafts(db, drafts, { workspaceId });
        let acked = 0;
        for (const ack of pendingOutboundDraftAcks(db)) {
          await activeDeps.ackOutboundDraft(ack.webMessageId, {
            status: ack.status,
            providerMsgId: ack.providerMsgId ?? undefined,
            error: ack.error ?? undefined,
          });
          markOutboundDraftAcked(db, ack.webMessageId, ack.status, activeDeps.now());
          acked++;
        }
        const next = maxOutboundDraftCursor(drafts, cursor, activeDeps.now());
        setCursor(db, source, next);
        const result: SourceResult = { source, status: "ok", cursor: next, counts: withSourceMeta(source, { ...counts, acked }) };
        finishSource(db, result, activeDeps.now());
        sources.push(result);
      } else {
        const crm = await activeDeps.fetchCrm({ since: cursor || undefined });
        workspaceId = selectedWorkspaceId(db);
        const counts = storeCrm(db, crm, { workspaceId });
        const next = maxCrmCursor(crm, cursor, activeDeps.now());
        setCursor(db, source, next);
        const result: SourceResult = { source, status: "ok", cursor: next, counts: withSourceMeta(source, counts) };
        finishSource(db, result, activeDeps.now());
        sources.push(result);
      }
    } catch (e) {
      const classified = classifyError(e);
      const result: SourceResult = {
        source,
        status: classified.status,
        cursor,
        counts: withSourceMeta(source),
        error: classified.error,
      };
      finishSource(db, result, activeDeps.now());
      sources.push(result);
    }
  }
  return { ok: sources.every((s) => s.status === "ok" || s.status === "skipped" || s.status === "waiting_for_web"), sources };
}

function pendingCommands(db: DatabaseSync): string[] {
  ensureSyncLoopSchema(db);
  return (db.prepare(
    `SELECT id FROM sync_command WHERE kind='web_pull' AND status='pending' ORDER BY requested_at ASC LIMIT 20`,
  ).all() as Array<{ id: string }>).map((r) => r.id);
}

function markCommands(db: DatabaseSync, ids: string[], status: string, a: { startedAt?: number; finishedAt?: number; error?: string }) {
  if (!ids.length) return;
  const ph = ids.map(() => "?").join(",");
  const sets: string[] = ["status=?"];
  const params: any[] = [status];
  if (a.startedAt != null) { sets.push("started_at=?"); params.push(a.startedAt); }
  if (a.finishedAt != null) { sets.push("finished_at=?"); params.push(a.finishedAt); }
  if (a.error != null) { sets.push("error=?"); params.push(a.error); }
  db.prepare(`UPDATE sync_command SET ${sets.join(", ")} WHERE id IN (${ph})`).run(...params, ...ids);
}

async function runWithDb(dbPath: string, fn: (db: DatabaseSync) => Promise<void> | void) {
  const db = new DatabaseSync(dbPath);
  try {
    db.exec("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;");
    await fn(db);
  } finally {
    db.close();
  }
}

async function runBackgroundOnce(dbPath: string, reason: "timer" | "command") {
  if (backgroundRunning) return;
  backgroundRunning = true;
  try {
    await runWithDb(dbPath, async (db) => {
      let commandIds: string[] = [];
      if (reason === "command") {
        commandIds = pendingCommands(db);
        if (!commandIds.length) return;
        markCommands(db, commandIds, "running", { startedAt: Date.now() });
      }
      const result = await runIncrementalSync(db);
      if (commandIds.length) {
        const errors = result.sources.filter((s) => s.status === "error").map((s) => `${s.source}: ${s.error}`).join("; ");
        markCommands(db, commandIds, errors ? "error" : "done", { finishedAt: Date.now(), error: errors || undefined });
      }
    });
  } finally {
    backgroundRunning = false;
  }
}

export function startBackgroundSync(
  dbPath: string,
  opts: { intervalMs?: number; commandPollMs?: number; initialDelayMs?: number } = {},
) {
  if (backgroundStarted) return;
  backgroundStarted = true;
  const startedAt = Date.now();
  try {
    const db = new DatabaseSync(dbPath);
    try {
      db.exec("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;");
      resetRunningOnBoot(db, startedAt);
      recordEngineInfo(db, { startedAt, now: startedAt });
    } finally {
      db.close();
    }
  } catch {}
  const intervalMs = opts.intervalMs ?? 5 * 60_000;
  const commandPollMs = opts.commandPollMs ?? 5_000;
  const initialDelayMs = opts.initialDelayMs ?? 15_000;
  setTimeout(() => { void runBackgroundOnce(dbPath, "timer"); }, initialDelayMs).unref();
  setInterval(() => { void runBackgroundOnce(dbPath, "timer"); }, intervalMs).unref();
  setInterval(() => { void runBackgroundOnce(dbPath, "command"); }, commandPollMs).unref();
}
