import { test } from "node:test";
import assert from "node:assert/strict";
import { DatabaseSync } from "node:sqlite";
import { classifyError, queueSyncNow, recordEngineInfo, resetRunningOnBoot, resetStaleRunning, runIncrementalSync, syncHealth, type SyncDeps } from "./loop.ts";
import { markConversationPageRoute } from "./webClient.ts";

function freshDb(): DatabaseSync {
  const db = new DatabaseSync(":memory:");
  db.exec("PRAGMA foreign_keys=ON;");
  db.exec(`CREATE TABLE unified_conversation (
    id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, channel TEXT NOT NULL, account_id TEXT,
    external_id TEXT, title TEXT, contact_id TEXT, last_ts INTEGER, unread INTEGER NOT NULL DEFAULT 0,
    archived_at INTEGER, muted INTEGER NOT NULL DEFAULT 0, updated_at INTEGER NOT NULL, deleted_at INTEGER);`);
  db.exec(`CREATE TABLE unified_message (
    id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, conversation_id TEXT NOT NULL, direction TEXT NOT NULL,
    body TEXT, sender TEXT, ts INTEGER NOT NULL, provider_msg_id TEXT, attachments TEXT, media_type TEXT,
    media_path TEXT, reaction TEXT, source TEXT, updated_at INTEGER NOT NULL, deleted_at INTEGER,
    FOREIGN KEY(conversation_id) REFERENCES unified_conversation(id));`);
  db.exec(`CREATE UNIQUE INDEX ux_msg_provider ON unified_message(provider_msg_id) WHERE provider_msg_id IS NOT NULL;`);
  db.exec(`CREATE TABLE contact (id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, display_name TEXT, avatar_url TEXT, avatar_path TEXT, updated_at INTEGER NOT NULL, deleted_at INTEGER);`);
  db.exec(`CREATE TABLE company (id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, name TEXT NOT NULL, industry TEXT, domain TEXT, updated_at INTEGER NOT NULL, deleted_at INTEGER);`);
  db.exec(`CREATE TABLE crm_deal (id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, name TEXT NOT NULL, company TEXT, stage TEXT, value TEXT, close_date TEXT, updated_at INTEGER NOT NULL, deleted_at INTEGER);`);
  return db;
}

function deps(nowValues: number[]): SyncDeps {
  const nextNow = () => nowValues.shift() ?? 9_999;
  return {
    companionToken: () => "token",
    now: nextNow,
    bootstrap: async () => ({
      activeOrganizationId: "org1",
      organizations: [{
        id: "org1",
        name: "Dewx",
        slug: "dewx",
        role: "OWNER",
        integrations: [],
        workspaces: [{
          id: "ws1",
          name: "Main",
          slug: "main",
          isDefault: true,
          status: "ACTIVE",
          integrations: [{
            id: "int1",
            type: "LINKEDIN",
            provider: "unipile",
            displayName: "Roki LinkedIn",
            username: "roki",
            status: "CONNECTED",
            avatarUrl: null,
            externalAccountId: "acct1",
            workspaceId: "ws1",
          }],
        }],
      }],
    }),
    fetchConversations: async () => [{
      externalId: "chat1",
      channel: "LINKEDIN",
      accountId: "int1",
      title: "Ada",
      lastTs: 2000,
      unread: 1,
      messages: [{ providerMsgId: "m1", direction: "IN", body: "hello", sender: "ada", ts: 2100 }],
    }],
    fetchOutboundDrafts: async () => [{
      id: "draft-web-1",
      conversationId: "web-conv-1",
      providerConversationId: "wa-chat-1",
      channel: "WHATSAPP",
      accountId: "wa1",
      externalId: "wa-chat-1",
      title: "Client WA",
      body: "Reply from web",
      status: "pending",
      source: "web",
      executionTarget: "local-device",
      requiresOnlineDevice: true,
      createdAt: 2400,
      updatedAt: 2500,
    }],
    ackOutboundDraft: async (id, body) => ({ id, status: body.status }),
    fetchCrm: async () => ({
      contacts: [{ id: "c1", name: "Ada", updatedAt: 3000 }],
      companies: [{ id: "co1", name: "Engines", updatedAt: 3100 }],
      deals: [{ id: "d1", title: "Pilot", stage: "OPEN", updatedAt: 3200, contactId: "c1" }],
    }),
  };
}

function conversations(prefix: string, startTs: number, count: number) {
  return Array.from({ length: count }, (_, i) => ({
    externalId: `${prefix}-${i}`,
    channel: "LINKEDIN",
    accountId: "int1",
    title: `${prefix} ${i}`,
    lastTs: startTs + i,
    unread: 0,
    messages: [],
  }));
}

test("runIncrementalSync skips cleanly when there is no companion token", async () => {
  const db = freshDb();
  const result = await runIncrementalSync(db, {
    ...deps([1000]),
    companionToken: () => null,
  });

  assert.equal(result.ok, true);
  assert.equal(result.skipped, "not signed in");
  assert.deepEqual(result.sources.map((s) => s.status), ["skipped", "skipped", "skipped", "skipped"]);
  const rows = syncHealth(db) as any[];
  assert.equal(rows.length, 4);
  assert.equal(rows[0].last_error, "not signed in");
});

test("recordEngineInfo writes package version, git sha, and start timestamp", () => {
  const db = freshDb();

  recordEngineInfo(db, { version: "9.8.7", gitSha: "abc1234", startedAt: 1234, now: 1300 });

  const row = { ...(db.prepare("SELECT version, git_sha, started_at, updated_at FROM engine_info WHERE id='engine'").get() as any) };
  assert.deepEqual(row, {
    version: "9.8.7",
    git_sha: "abc1234",
    started_at: 1234,
    updated_at: 1300,
  });
});

test("runIncrementalSync stores bootstrap/inbox/crm and advances per-source cursors", async () => {
  const db = freshDb();
  const result = await runIncrementalSync(db, deps([1000, 1100, 1200, 1300, 1400, 1500, 1600, 1700, 1800, 1900, 2000]));

  assert.equal(result.ok, true);
  assert.deepEqual(result.sources.map((s) => s.source), ["bootstrap", "inbox", "outbound_drafts", "crm"]);
  assert.equal((db.prepare("SELECT COUNT(*) c FROM unified_conversation").get() as any).c, 2);
  assert.equal((db.prepare("SELECT COUNT(*) c FROM unified_message").get() as any).c, 1);
  assert.equal((db.prepare("SELECT COUNT(*) c FROM outbound_draft").get() as any).c, 1);
  assert.equal((db.prepare("SELECT COUNT(*) c FROM contact").get() as any).c, 1);

  const cursors = db.prepare("SELECT table_name, last_pulled FROM _sync_state ORDER BY table_name").all() as any[];
  assert.deepEqual(cursors.map((r) => [r.table_name, r.last_pulled]), [
    ["sync:bootstrap", 1200],
    ["sync:crm", 3200],
    ["sync:inbox", 2100],
    ["sync:outbound_drafts", 2500],
  ]);

  const health = syncHealth(db) as any[];
  assert.deepEqual(health.map((r) => r.status), ["ok", "ok", "ok", "ok"]);
  assert.match(health.find((r) => r.source === "inbox").counts_json, /"messages":1/);
  assert.match(health.find((r) => r.source === "outbound_drafts").counts_json, /"stored":1/);
});

test("queueSyncNow records an on-demand pull request", () => {
  const db = freshDb();
  const queued = queueSyncNow(db, "test") as any;
  assert.equal(queued.queued, true);
  const row = db.prepare("SELECT kind, status, source FROM sync_command WHERE id=?").get(queued.command_id) as any;
  assert.equal(row.kind, "web_pull");
  assert.equal(row.status, "pending");
  assert.equal(row.source, "test");
});

test("runIncrementalSync acks locally completed web drafts with provider ids", async () => {
  const db = freshDb();
  db.exec(`CREATE TABLE outbound_draft (
    id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, conversation_id TEXT, channel TEXT,
    account_id TEXT, external_id TEXT, title TEXT, body TEXT, status TEXT, source TEXT,
    created_at INTEGER, updated_at INTEGER, web_message_id TEXT, web_ack_status TEXT,
    provider_msg_id TEXT, error TEXT);`);
  db.prepare(`INSERT INTO outbound_draft
    (id, workspace_id, conversation_id, channel, account_id, external_id, title, body, status, source, created_at, updated_at, web_message_id, provider_msg_id)
    VALUES ('draft-web-acked', 'local', 'wa:wa1:wa-chat-1', 'WHATSAPP', 'wa1', 'wa-chat-1', 'Client WA', 'Approved', 'sent', 'web', 1000, 1100, 'web-msg-1', 'wa-provider-1')`).run();
  const calls: Array<{ id: string; body: { status: string; providerMsgId?: string; error?: string | null } }> = [];
  const testDeps: SyncDeps = {
    ...deps([1000, 1100, 1200, 1300, 1400, 1500, 1600, 1700]),
    fetchOutboundDrafts: async () => [],
    ackOutboundDraft: async (id, body) => {
      calls.push({ id, body });
      return { id, status: body.status };
    },
  };

  const result = await runIncrementalSync(db, testDeps);

  assert.equal(result.ok, true);
  assert.deepEqual(calls, [{
    id: "web-msg-1",
    body: { status: "sent", providerMsgId: "wa-provider-1", error: undefined },
  }]);
  const row = db.prepare("SELECT web_ack_status FROM outbound_draft WHERE id='draft-web-acked'").get() as any;
  assert.equal(row.web_ack_status, "sent");
});

test("runIncrementalSync acks LinkedIn browser placements distinctly", async () => {
  const db = freshDb();
  db.exec(`CREATE TABLE outbound_draft (
    id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, conversation_id TEXT, channel TEXT,
    account_id TEXT, external_id TEXT, title TEXT, body TEXT, status TEXT, source TEXT,
    created_at INTEGER, updated_at INTEGER, web_message_id TEXT, web_ack_status TEXT,
    provider_msg_id TEXT, error TEXT);`);
  db.prepare(`INSERT INTO outbound_draft
    (id, workspace_id, conversation_id, channel, account_id, external_id, title, body, status, source, created_at, updated_at, web_message_id, provider_msg_id)
    VALUES ('draft-li-placed', 'local', 'web:LINKEDIN:thread-1', 'LINKEDIN', 'li1', 'thread-1', 'Ada', 'Approved', 'placed', 'web', 1000, 1100, 'web-li-1', 'li:placed:thread-1')`).run();
  const calls: Array<{ id: string; body: { status: string; providerMsgId?: string; error?: string | null } }> = [];
  const testDeps: SyncDeps = {
    ...deps([1000, 1100, 1200, 1300, 1400, 1500, 1600, 1700]),
    fetchOutboundDrafts: async () => [],
    ackOutboundDraft: async (id, body) => {
      calls.push({ id, body });
      return { id, status: body.status };
    },
  };

  const result = await runIncrementalSync(db, testDeps);

  assert.equal(result.ok, true);
  assert.deepEqual(calls, [{
    id: "web-li-1",
    body: { status: "placed", providerMsgId: "li:placed:thread-1", error: undefined },
  }]);
  const row = db.prepare("SELECT web_ack_status FROM outbound_draft WHERE id='draft-li-placed'").get() as any;
  assert.equal(row.web_ack_status, "placed");
});

test("runIncrementalSync paginates inbox and advances cursor per page", async () => {
  const db = freshDb();
  const sinceValues: Array<number | undefined> = [];
  const pages = [
    conversations("p1", 1000, 25),
    conversations("p2", 2000, 25),
    conversations("p3", 3000, 3),
  ];
  const testDeps: SyncDeps = {
    ...deps([1000, 1100, 1200, 1300, 1400, 1500, 1600, 1700, 1800]),
    fetchConversations: async (opts) => {
      sinceValues.push(opts?.since);
      return pages.shift() ?? [];
    },
    fetchOutboundDrafts: async () => [],
    fetchCrm: async () => ({ contacts: [], companies: [], deals: [] }),
  };

  const result = await runIncrementalSync(db, testDeps);

  assert.equal(result.ok, true);
  assert.deepEqual(sinceValues, [undefined, 1024, 2024]);
  assert.equal((db.prepare("SELECT last_pulled FROM _sync_state WHERE table_name='sync:inbox'").get() as any).last_pulled, 3002);
  const inbox = result.sources.find((s) => s.source === "inbox")!;
  assert.equal(inbox.cursor, 3002);
  assert.equal(inbox.counts.pages, 3);
  assert.equal((db.prepare("SELECT COUNT(*) c FROM unified_conversation WHERE channel='LINKEDIN'").get() as any).c, 53);
});

test("runIncrementalSync raises inbox page budget when sync conversations endpoint serves pages", async () => {
  const db = freshDb();
  const sinceValues: Array<number | undefined> = [];
  let calls = 0;
  const testDeps: SyncDeps = {
    ...deps([1000, 1100, 1200, 1300, 1400, 1500, 1600, 1700, 1800]),
    fetchConversations: async (opts) => {
      calls++;
      sinceValues.push(opts?.since);
      assert.equal(opts?.workspaceId, "ws1");
      return markConversationPageRoute(conversations(`sync${calls}`, calls * 1000, 25), "sync");
    },
    fetchOutboundDrafts: async () => [],
    fetchCrm: async () => ({ contacts: [], companies: [], deals: [] }),
  };

  const result = await runIncrementalSync(db, testDeps);

  assert.equal(result.ok, true);
  assert.equal(calls, 40);
  assert.equal(sinceValues[0], undefined);
  assert.equal(sinceValues.at(-1), 39024);
  assert.equal((db.prepare("SELECT COUNT(*) c FROM unified_conversation WHERE channel='LINKEDIN'").get() as any).c, 1000);
  const inbox = result.sources.find((s) => s.source === "inbox")!;
  assert.equal(inbox.cursor, 40024);
  assert.equal(inbox.counts.pages, 40);
  assert.equal(inbox.counts.syncEndpointPages, 40);
});

test("runIncrementalSync records partial inbox progress when a later page fails", async () => {
  const db = freshDb();
  let calls = 0;
  const testDeps: SyncDeps = {
    ...deps([1000, 1100, 1200, 1300, 1400, 1500, 1600, 1700]),
    fetchConversations: async () => {
      calls++;
      if (calls === 1) return conversations("ok", 2000, 25);
      throw new Error("socket reset");
    },
    fetchOutboundDrafts: async () => [],
    fetchCrm: async () => ({ contacts: [], companies: [], deals: [] }),
  };

  const result = await runIncrementalSync(db, testDeps);

  const inbox = result.sources.find((s) => s.source === "inbox")!;
  assert.equal(result.ok, false);
  assert.equal(inbox.status, "error");
  assert.equal(inbox.cursor, 2024);
  assert.equal(inbox.counts.partial, true);
  assert.equal(inbox.counts.pages, 1);
  assert.equal((db.prepare("SELECT last_pulled FROM _sync_state WHERE table_name='sync:inbox'").get() as any).last_pulled, 2024);
  assert.equal((db.prepare("SELECT COUNT(*) c FROM unified_conversation WHERE external_id LIKE 'ok-%'").get() as any).c, 25);
});

test("runIncrementalSync stops inbox pagination on timestamp tie", async () => {
  const db = freshDb();
  recordEngineInfo(db, { version: "1", gitSha: null, startedAt: 1, now: 1 });
  db.exec(`INSERT INTO _sync_state(table_name, last_pulled, last_pushed) VALUES ('sync:inbox', 1000, 0);`);
  let calls = 0;
  const testDeps: SyncDeps = {
    ...deps([1000, 1100, 1200, 1300, 1400, 1500, 1600, 1700]),
    fetchConversations: async () => {
      calls++;
      return conversations("tie", 1000, 25).map((c) => ({ ...c, lastTs: 1000 }));
    },
    fetchOutboundDrafts: async () => [],
    fetchCrm: async () => ({ contacts: [], companies: [], deals: [] }),
  };

  const result = await runIncrementalSync(db, testDeps);

  assert.equal(result.ok, true);
  assert.equal(calls, 1);
  assert.equal((db.prepare("SELECT last_pulled FROM _sync_state WHERE table_name='sync:inbox'").get() as any).last_pulled, 1000);
});

test("stale running rows are reset by age and boot reset catches all running rows", () => {
  const db = freshDb();
  recordEngineInfo(db, { version: "1", gitSha: null, startedAt: 1, now: 1 });
  db.prepare(`INSERT INTO sync_health(source, status, last_started_at, cursor, counts_json, updated_at)
    VALUES ('inbox', 'running', 1000, 0, '{}', 1000)`).run();
  db.prepare(`INSERT INTO sync_health(source, status, last_started_at, cursor, counts_json, updated_at)
    VALUES ('crm', 'running', 590000, 0, '{}', 590000)`).run();

  resetStaleRunning(db, 700000);
  assert.equal((db.prepare("SELECT status FROM sync_health WHERE source='inbox'").get() as any).status, "error");
  assert.equal((db.prepare("SELECT status FROM sync_health WHERE source='crm'").get() as any).status, "running");

  resetRunningOnBoot(db, 800000);
  assert.equal((db.prepare("SELECT status FROM sync_health WHERE source='crm'").get() as any).status, "error");
  assert.equal((db.prepare("SELECT last_error FROM sync_health WHERE source='crm'").get() as any).last_error, "interrupted (engine restart or stale run)");
});

test("web timeout errors classify as retryable source errors", () => {
  assert.deepEqual(classifyError(new Error("web timeout after 20000ms on /api/companion/conversations")), {
    status: "error",
    error: "web timeout after 20000ms on /api/companion/conversations",
  });
});
