import PgBoss from "pg-boss";
import { nodePostgresConnection, type TenantTx } from "@slackwsh/data";

export const JOB_QUEUES = [
  "notifications",
  "email",
  "sms",
  "unfurl",
  "search_index",
  "media",
  "webhooks",
  "retention",
  "exports",
  "analytics",
] as const;
export type JobQueue = (typeof JOB_QUEUES)[number];

export interface NotificationJob {
  messageId: number;
  workspaceId: number;
  channelId: number;
  authorId: number;
}

export interface SearchIndexJob {
  messageId: number;
  workspaceId: number;
}

let started: Promise<PgBoss> | null = null;

/**
 * Producer-only pg-boss. The worker process is what `work()`s the queues;
 * api/gateway only `send()` inside the write transaction (§4.3 / ADR-006).
 * `supervise: false` so N API nodes do not all run maintenance.
 */
export function getBoss(): Promise<PgBoss> {
  if (!started) {
    started = (async () => {
      const rawUrl =
        process.env.DATABASE_URL ?? "postgres://slackwsh_app:slackwsh_app@localhost:5432/slackwsh";
      const boss = new PgBoss({
        ...nodePostgresConnection(rawUrl),
        supervise: false,
        schedule: false,
      });
      await boss.start();
      for (const queue of JOB_QUEUES) {
        await boss.createQueue(queue);
      }
      return boss;
    })();
  }
  return started;
}

export async function stopBoss(): Promise<void> {
  if (!started) return;
  const boss = await started;
  started = null;
  await boss.stop({ graceful: false, timeout: 5000 });
}

/**
 * pg-boss `send({ db })` emits node-postgres SQL (`$1::uuid`, `$3::jsonb`,
 * nulls). Rewriting those into a Drizzle `sql` template drops null
 * placeholders and leaves a bare `::`, which postgres.js rejects.
 * Run the original text on the reserved transaction connection instead.
 *
 * node-postgres also binds `undefined` as NULL and JSON-encodes objects;
 * postgres.js `unsafe()` throws on both.
 */
function toPostgresJsParam(value: unknown): unknown {
  if (value === undefined) return null;
  if (value !== null && typeof value === "object" && !(value instanceof Date) && !Buffer.isBuffer(value)) {
    return JSON.stringify(value);
  }
  return value;
}

function postgresJsTxDb(tx: TenantTx): PgBoss.Db {
  const client = (
    tx as unknown as { session?: { client?: { unsafe: (text: string, values?: unknown[]) => Promise<unknown> } } }
  ).session?.client;
  if (!client?.unsafe) {
    throw new Error("enqueueInTx requires a postgres.js drizzle transaction");
  }
  return {
    async executeSql(text: string, values: unknown[] = []) {
      const result = await client.unsafe(text, values.map(toPostgresJsParam));
      const rows = Array.isArray(result) ? result : [];
      return { rows };
    },
  };
}

/** Insert a pg-boss job on the same connection as `tx` so commit/rollback is atomic. */
export async function enqueueInTx(tx: TenantTx, queue: JobQueue, data: object): Promise<string | null> {
  const boss = await getBoss();
  return boss.send(queue, data, { db: postgresJsTxDb(tx) });
}