import type { Db } from "@sttracker/db/db";
import { newId, nowIso } from "@sttracker/db/ids";
import { fromJson, toJson } from "@sttracker/db/json";
import type { ConnectionStatus, RunMode, RunStatus } from "@sttracker/db/schema";
import { createHttpClient, type HttpClientOptions } from "@sttracker/plugin-sdk/http";
import type { HttpClient, PluginContext, PluginManifest } from "@sttracker/plugin-sdk/types";
import { decryptJson, encryptJson } from "./crypto.ts";
import { CryptoError, NotFoundError, ValidationError } from "./errors.ts";
import { DropJobError, NonRetryableError, type JobContext } from "./jobs.ts";
import type { PluginRegistry } from "./registry.ts";
import { runSync, type Connection, type RunSummary, type SyncMode } from "./sync/run.ts";

const WEEK_MS = 7 * 24 * 60 * 60 * 1000;

export interface ConnectionView {
  id: string;
  pluginId: string;
  name: string;
  config: Record<string, unknown>;
  scheduleMinutes: number | null;
  status: ConnectionStatus;
  lastRunAt: string | null;
  createdAt: string;
  hasCredentials: boolean;
}

export interface ConnectionInput {
  pluginId: string;
  name: string;
  config: Record<string, unknown>;
  credentials?: Record<string, string>;
  scheduleMinutes?: number | null;
}

type Creds = Record<string, string>;

/** Split incoming config: fields typed `"secret"` in the manifest go to the credentials blob, the rest stay config. */
function splitConfig(manifest: PluginManifest, config: Record<string, unknown>): { plain: Record<string, unknown>; secrets: Creds } {
  const secretKeys = new Set(manifest.config.filter((f) => f.type === "secret").map((f) => f.key));
  const plain: Record<string, unknown> = {};
  const secrets: Creds = {};
  for (const [k, v] of Object.entries(config)) {
    if (!secretKeys.has(k)) plain[k] = v;
    else if (typeof v === "string" && v !== "") secrets[k] = v;
    else if (v !== undefined && v !== null && v !== "") throw new ValidationError(k, "secret_must_be_string");
  }
  return { plain, secrets };
}

async function readCreds(secret: string, enc: string | null): Promise<Creds | null> {
  return enc === null ? null : decryptJson<Creds>(secret, enc);
}

const encOrNull = async (secret: string, c: Creds): Promise<string | null> => (Object.keys(c).length === 0 ? null : encryptJson(secret, c));

function checkSchedule(m: number | null | undefined): void {
  if (m !== undefined && m !== null && (!Number.isInteger(m) || m < 1)) throw new ValidationError("scheduleMinutes", "must_be_positive_integer");
}

export async function createConnection(db: Db, secret: string, userId: string, input: ConnectionInput, manifest: PluginManifest): Promise<string> {
  if (manifest.id !== input.pluginId) throw new ValidationError("pluginId", "manifest_mismatch");
  if (input.name.trim() === "") throw new ValidationError("name", "required");
  checkSchedule(input.scheduleMinutes);
  const { plain, secrets } = splitConfig(manifest, input.config);
  const id = newId();
  await db.k
    .insertInto("source_connections")
    .values({
      id,
      user_id: userId,
      plugin_id: input.pluginId,
      name: input.name.trim(),
      config: toJson(plain),
      credentials_enc: await encOrNull(secret, { ...secrets, ...input.credentials }),
      schedule_minutes: input.scheduleMinutes ?? null,
      cursor: null,
      status: "active",
      last_run_at: null,
      created_at: nowIso(),
      deleted_at: null,
    })
    .execute();
  return id;
}

export interface ConnectionPatch {
  name?: string;
  config?: Record<string, unknown>;
  credentials?: Record<string, string>;
  scheduleMinutes?: number | null;
  status?: ConnectionStatus;
}

/** Secrets omitted from `config` are kept; plain config is replaced wholesale. Read-modify-write runs in one transaction. */
export async function updateConnection(db: Db, secret: string, userId: string, id: string, patch: ConnectionPatch, manifest: PluginManifest): Promise<void> {
  checkSchedule(patch.scheduleMinutes);
  await db.k.transaction().execute(async (tx) => {
    let q = tx.selectFrom("source_connections").selectAll().where("id", "=", id).where("user_id", "=", userId).where("deleted_at", "is", null);
    if (db.dialect === "postgres") q = q.forUpdate();
    const row = await q.executeTakeFirst();
    if (!row) throw new NotFoundError("connection");
    if (row.plugin_id !== manifest.id) throw new ValidationError("pluginId", "manifest_mismatch");
    const set: Record<string, unknown> = {};
    if (patch.name !== undefined) {
      if (patch.name.trim() === "") throw new ValidationError("name", "required");
      set.name = patch.name.trim();
    }
    if (patch.scheduleMinutes !== undefined) set.schedule_minutes = patch.scheduleMinutes;
    if (patch.status !== undefined) set.status = patch.status;
    if (patch.config !== undefined || patch.credentials !== undefined) {
      const split = patch.config ? splitConfig(manifest, patch.config) : { plain: undefined, secrets: {} };
      if (split.plain) set.config = toJson(split.plain);
      const merged = { ...((await readCreds(secret, row.credentials_enc)) ?? {}), ...split.secrets, ...patch.credentials };
      set.credentials_enc = await encOrNull(secret, merged);
    }
    if (Object.keys(set).length > 0) await tx.updateTable("source_connections").set(set).where("id", "=", id).execute();
  });
}

/** Soft delete: `deleted_at` is set, the connection is paused and its credentials are wiped. Records and runs stay. */
export async function deleteConnection(db: Db, userId: string, id: string): Promise<void> {
  const r = await db.k
    .updateTable("source_connections")
    .set({ status: "paused", credentials_enc: null, deleted_at: nowIso() })
    .where("id", "=", id)
    .where("user_id", "=", userId)
    .where("deleted_at", "is", null)
    .executeTakeFirst();
  if (r.numUpdatedRows !== 1n) throw new NotFoundError("connection");
}

/** The user's live (not deleted) connections. Never returns secrets: plain config plus `hasCredentials`. */
export async function listConnections(db: Db, userId: string): Promise<ConnectionView[]> {
  const rows = await db.k
    .selectFrom("source_connections")
    .selectAll()
    .where("user_id", "=", userId)
    .where("deleted_at", "is", null)
    .orderBy("created_at")
    .execute();
  return rows.map((r) => ({
    id: r.id,
    pluginId: r.plugin_id,
    name: r.name,
    config: fromJson<Record<string, unknown>>(r.config, {}),
    scheduleMinutes: r.schedule_minutes,
    status: r.status,
    lastRunAt: r.last_run_at,
    createdAt: r.created_at,
    hasCredentials: r.credentials_enc !== null,
  }));
}

async function ownedRow(db: Db, userId: string, id: string) {
  const r = await db.k
    .selectFrom("source_connections")
    .selectAll()
    .where("id", "=", id)
    .where("user_id", "=", userId)
    .where("deleted_at", "is", null)
    .executeTakeFirst();
  if (!r) throw new NotFoundError("connection");
  return r;
}

/** One of the user's live connections, without secrets. Deleted or foreign connections throw `NotFoundError`. */
export async function getConnection(db: Db, userId: string, id: string): Promise<ConnectionView> {
  const r = await ownedRow(db, userId, id);
  return {
    id: r.id,
    pluginId: r.plugin_id,
    name: r.name,
    config: fromJson<Record<string, unknown>>(r.config, {}),
    scheduleMinutes: r.schedule_minutes,
    status: r.status,
    lastRunAt: r.last_run_at,
    createdAt: r.created_at,
    hasCredentials: r.credentials_enc !== null,
  };
}

export interface RunView {
  id: string;
  mode: RunMode;
  status: RunStatus;
  startedAt: string;
  finishedAt: string | null;
  counts: Record<string, number>;
  log: string;
  error: string | null;
}

/** Runs of one of the user's connections, newest first (at most `limit`, default 50). */
export async function listRuns(db: Db, userId: string, connectionId: string, limit = 50): Promise<RunView[]> {
  await ownedRow(db, userId, connectionId);
  const rows = await db.k
    .selectFrom("source_runs")
    .selectAll()
    .where("connection_id", "=", connectionId)
    .orderBy("started_at", "desc")
    .orderBy("id", "desc")
    .limit(Math.min(Math.max(limit, 1), 200))
    .execute();
  return rows.map((r) => ({
    id: r.id,
    mode: r.mode,
    status: r.status,
    startedAt: r.started_at,
    finishedAt: r.finished_at,
    counts: fromJson<Record<string, number>>(r.counts, {}),
    log: r.log,
    error: r.error,
  }));
}

async function deviceContext(db: Db, secret: string, registry: PluginRegistry, userId: string, id: string) {
  const r = await ownedRow(db, userId, id);
  const plugin = registry.get(r.plugin_id);
  if (!plugin) throw new ValidationError("pluginId", "not_installed");
  if (!plugin.deviceAuth) throw new ValidationError("pluginId", "no_device_auth");
  const conn: Connection = {
    id: r.id,
    userId: r.user_id,
    pluginId: r.plugin_id,
    name: r.name,
    config: fromJson<Record<string, unknown>>(r.config, {}),
    credentialsEnc: r.credentials_enc,
    scheduleMinutes: r.schedule_minutes,
    cursor: r.cursor,
    status: r.status,
    lastRunAt: r.last_run_at,
    createdAt: r.created_at,
  };
  const ctx = await makePluginContext(db, secret, conn, "device", plugin.manifest);
  return { deviceAuth: plugin.deviceAuth, ctx };
}

/** Start a device-code sign-in for one of the user's connections. */
export async function deviceStart(db: Db, secret: string, registry: PluginRegistry, userId: string, id: string) {
  const { deviceAuth, ctx } = await deviceContext(db, secret, registry, userId, id);
  return deviceAuth.start(ctx);
}

/** Poll a device-code sign-in. On success the returned credentials are merged into the encrypted blob. */
export async function devicePoll(db: Db, secret: string, registry: PluginRegistry, userId: string, id: string, deviceCode: string): Promise<"pending" | "connected"> {
  const { deviceAuth, ctx } = await deviceContext(db, secret, registry, userId, id);
  const res = await deviceAuth.poll(ctx, deviceCode);
  if (res === "pending") return "pending";
  await ctx.credentials.save(res);
  return "connected";
}

export const USER_AGENT = "STTracker/0.1 (+self-hosted)";

export interface ContextOpts {
  /** Replaces the rate-limited client entirely. */
  http?: HttpClient;
  /** Overrides for the built client (fetch, sleep, clock); used by tests. */
  httpOptions?: Pick<HttpClientOptions, "fetchImpl" | "sleep" | "now">;
  signal?: AbortSignal;
  /** Receives `ctx.log` messages. */
  onLog?: (msg: string) => void;
}

/** Synchronous context builder over already-decrypted credentials. */
function buildContext(db: Db, secret: string, conn: Connection, manifest: PluginManifest, initial: Creds | null, opts: ContextOpts): PluginContext {
  let creds = initial;
  let cursor = conn.cursor;
  return {
    http:
      opts.http ??
      createHttpClient({
        ratePerMinute: manifest.rateLimit?.perMinute ?? 60,
        burst: manifest.rateLimit?.burst ?? 5,
        userAgent: USER_AGENT,
        ...opts.httpOptions,
      }),
    log: (msg) => opts.onLog?.(msg),
    config: { ...conn.config, ...(initial ?? {}) },
    credentials: {
      get: async () => creds,
      async save(c) {
        // Read the current blob, merge, and write in one transaction so concurrent saves cannot lose keys.
        creds = await db.k.transaction().execute(async (tx) => {
          let q = tx.selectFrom("source_connections").select("credentials_enc").where("id", "=", conn.id);
          if (db.dialect === "postgres") q = q.forUpdate();
          const row = await q.executeTakeFirstOrThrow();
          const merged = { ...((await readCreds(secret, row.credentials_enc)) ?? {}), ...c };
          await tx.updateTable("source_connections").set({ credentials_enc: await encryptJson(secret, merged) }).where("id", "=", conn.id).execute();
          return merged;
        });
      },
    },
    cursor: {
      get: async () => cursor,
      async set(c) {
        await db.k.updateTable("source_connections").set({ cursor: c }).where("id", "=", conn.id).execute();
        cursor = c;
      },
    },
    now: () => new Date(),
    signal: opts.signal ?? new AbortController().signal,
  };
}

/**
 * Build the plugin context for one run. `ctx.http` is rate limited from `manifest.rateLimit` (default 60/min, burst 5)
 * unless `opts.http` is given. `credentials.save` merges into the current blob and persists it encrypted immediately;
 * `cursor.set` persists immediately.
 */
export async function makePluginContext(db: Db, secret: string, conn: Connection, _runId: string, manifest: PluginManifest, opts: ContextOpts = {}): Promise<PluginContext> {
  return buildContext(db, secret, conn, manifest, await readCreds(secret, conn.credentialsEnc), opts);
}

export interface SyncHandlerDeps {
  db: Db;
  secret: string;
  registry: PluginRegistry;
  http?: HttpClient;
}

const MSG_CRYPTO = "Cannot decrypt this source's credentials. Was STTRACKER_SECRET changed? Reconnect the source to fix it.";

/** Record a run that failed before `runSync` could start, flag the connection, and stop the job without retry. */
async function failBeforeRun(db: Db, connectionId: string, mode: RunMode, message: string): Promise<never> {
  const now = nowIso();
  await db.k
    .insertInto("source_runs")
    .values({ id: newId(), connection_id: connectionId, mode, started_at: now, finished_at: now, status: "failed", counts: "{}", log: message, error: message })
    .execute();
  await db.k.updateTable("source_connections").set({ status: "error", last_run_at: now }).where("id", "=", connectionId).execute();
  throw new NonRetryableError(message);
}

/**
 * The `"sync"` job handler. Payload `{ connectionId, mode? }`. Runs `full` when asked, when the plugin cannot
 * `pullChanges`, or when the last successful full run is over a week old; otherwise `changes`.
 * A missing or paused connection drops the job. A missing plugin or undecryptable credentials record a failed
 * run, mark the connection `error` and fail the job without retry. A failed run is rethrown so the worker retries.
 * `ctx.log` output is appended to the run's `source_runs.log`.
 */
export interface SyncPayload {
  connectionId: string;
  mode?: "full" | "changes";
  /** File mode: run the plugin's `importFile` over these bytes instead of pulling. */
  file?: { name: string; base64: string };
}

export function syncHandler(deps: SyncHandlerDeps): (payload: SyncPayload, jc?: JobContext) => Promise<RunSummary | void> {
  return async (payload, jc) => {
    const { db } = deps;
    const full = await db.k.selectFrom("source_connections").selectAll().where("id", "=", payload.connectionId).executeTakeFirst();
    if (!full) throw new DropJobError("connection not found");
    if (full.deleted_at !== null) throw new DropJobError("connection is deleted");
    if (full.status === "paused") throw new DropJobError("connection is paused");
    const plugin = deps.registry.get(full.plugin_id);
    if (!plugin) return failBeforeRun(db, full.id, payload.file ? "file" : (payload.mode ?? "full"), `Plugin '${full.plugin_id}' is not installed or failed to load.`);
    let mode: SyncMode = "full";
    if (payload.file) {
      mode = { file: { name: payload.file.name, bytes: new Uint8Array(Buffer.from(payload.file.base64, "base64")) } };
    } else if (plugin.pullChanges && payload.mode !== "full") {
      const cutoff = new Date(Date.now() - WEEK_MS).toISOString();
      const recentFull = await db.k
        .selectFrom("source_runs")
        .select("id")
        .where("connection_id", "=", full.id)
        .where("mode", "=", "full")
        .where("status", "=", "ok")
        .where("started_at", ">=", cutoff)
        .executeTakeFirst();
      if (recentFull) mode = "changes";
    }
    const conn: Connection = {
      id: full.id,
      userId: full.user_id,
      pluginId: full.plugin_id,
      name: full.name,
      config: fromJson<Record<string, unknown>>(full.config, {}),
      credentialsEnc: full.credentials_enc,
      scheduleMinutes: full.schedule_minutes,
      cursor: full.cursor,
      status: full.status,
      lastRunAt: full.last_run_at,
      createdAt: full.created_at,
    };
    let creds: Creds | null;
    try {
      creds = await readCreds(deps.secret, conn.credentialsEnc);
    } catch (e) {
      if (e instanceof CryptoError) return failBeforeRun(db, full.id, typeof mode === "string" ? mode : "file", MSG_CRYPTO);
      throw e;
    }
    const logs: string[] = [];
    let runId: string | null = null;
    const makeContext = (c: Connection, id: string) => {
      runId = id;
      return buildContext(db, deps.secret, c, plugin.manifest, creds, { http: deps.http, signal: jc?.signal, onLog: (m) => void logs.push(m) });
    };
    let summary: RunSummary;
    try {
      summary = await runSync({ db, makeContext }, plugin, full.id, mode);
    } finally {
      if (runId !== null && logs.length > 0) {
        const cur = await db.k.selectFrom("source_runs").select("log").where("id", "=", runId).executeTakeFirst();
        if (cur) {
          const log = cur.log === "" ? logs.join("\n") : `${logs.join("\n")}\n${cur.log}`;
          await db.k.updateTable("source_runs").set({ log }).where("id", "=", runId).execute();
        }
      }
    }
    if (summary.status === "failed") throw new Error(summary.error ?? "sync failed");
    return summary;
  };
}