import PgBoss from "pg-boss";
import db from "../db.server";
import { adminLog } from "./admin-logger.server";

/**
 * pg-boss queue + BackgroundJob bridge (CDC §XVI).
 *
 * Why a singleton:
 * - Remix HMR / per-request handlers re-import this file. We must hold a single
 *   pg-boss instance per Node process (one container on Railway).
 *
 * Neon-specific constraint:
 * - pg-boss relies on LISTEN/NOTIFY for low-latency job dispatch. Neon's PgBouncer
 *   pooler does NOT support LISTEN/NOTIFY in transaction-pooling mode. We therefore
 *   require a DIRECT (unpooled) connection string. If `DIRECT_DATABASE_URL` is not
 *   set we fall back to `DATABASE_URL`, which works only if the user provided the
 *   direct URL in `DATABASE_URL` already. See README of pg-boss + Neon docs.
 *
 * Railway co-location:
 * - Workers run inside the same Remix process. Acceptable for low/medium throughput;
 *   for higher scale, split into a dedicated worker dyno that imports this module
 *   and calls `bootWorkers()` only.
 */

export const JobType = {
  GENERATE_URL_MAPPINGS: "generate-url-mappings",
  EMERGENCY_CLEANUP: "emergency-cleanup",
  PROCESS_BULK_IMPORT: "process-bulk-import",
} as const;

export type JobTypeValue = (typeof JobType)[keyof typeof JobType];

export interface GenerateUrlMappingsData {
  shopId: string;
}

export interface EmergencyCleanupData {
  shopId: string;
}

export interface ProcessBulkImportData {
  shopId: string;
  ruleKey: string;
  csvText: string;
}

export type JobBusinessData =
  | GenerateUrlMappingsData
  | EmergencyCleanupData
  | ProcessBulkImportData;

interface JobEnvelope<T extends JobBusinessData> {
  jobRowId: string;
  businessData: T;
}

let bossInstance: PgBoss | null = null;
let bossStartPromise: Promise<PgBoss> | null = null;
let workersBooted = false;
const ensuredQueues = new Set<string>();

/**
 * Idempotently create a pg-boss queue. v10 requires this before send()/work().
 * Swallows "already exists" so it is safe to call from multiple paths.
 */
async function ensureQueue(boss: PgBoss, type: JobTypeValue): Promise<void> {
  if (ensuredQueues.has(type)) return;
  try {
    await boss.createQueue(type);
  } catch (err) {
    const msg = err instanceof Error ? err.message : String(err);
    if (!/already exists|duplicate/i.test(msg)) throw err;
  }
  ensuredQueues.add(type);
}

function resolveConnectionString(): string {
  const url = process.env.DIRECT_DATABASE_URL ?? process.env.DATABASE_URL;
  if (!url) {
    throw new Error(
      "DIRECT_DATABASE_URL or DATABASE_URL must be set for pg-boss",
    );
  }
  return url;
}

export async function getBoss(): Promise<PgBoss> {
  if (bossInstance) return bossInstance;
  if (bossStartPromise) return bossStartPromise;

  bossStartPromise = (async () => {
    const boss = new PgBoss({
      connectionString: resolveConnectionString(),
      // Defaults applied to every job. Per-job overrides happen via SendOptions.
      retryLimit: 3,
      retryDelay: 30, // seconds
      expireInHours: 24,
      // Archive completed jobs after 24h so the pg-boss tables stay small.
      archiveCompletedAfterSeconds: 86_400,
    });

    boss.on("error", (err) => {
      // Async, do not await — keep the listener synchronous.
      void adminLog("error", `pg-boss error: ${err.message}`);
    });

    await boss.start();
    bossInstance = boss;
    return boss;
  })();

  return bossStartPromise;
}

/**
 * Insert a BackgroundJob row, dispatch to pg-boss, and link the two together.
 * Returns the BackgroundJob row id (used for client-side polling).
 */
export async function enqueueJob<T extends JobBusinessData>(
  type: JobTypeValue,
  businessData: T,
  opts?: PgBoss.SendOptions,
): Promise<string> {
  const shopId = businessData.shopId;

  const row = await db.backgroundJob.create({
    data: {
      shopId,
      type,
      status: "pending",
      businessData: businessData as unknown as object,
    },
  });

  try {
    const boss = await getBoss();
    await ensureQueue(boss, type);
    const envelope: JobEnvelope<T> = { jobRowId: row.id, businessData };
    const pgBossJobId = await boss.send(type, envelope, opts ?? {});

    if (pgBossJobId) {
      await db.backgroundJob.update({
        where: { id: row.id },
        data: { pgBossJobId },
      });
    }
  } catch (error) {
    const msg = error instanceof Error ? error.message : String(error);
    await db.backgroundJob.update({
      where: { id: row.id },
      data: { status: "failed", lastError: `Enqueue failed: ${msg}` },
    });
    await adminLog("error", `enqueueJob failed for ${type}: ${msg}`, { shopId });
    throw error;
  }

  return row.id;
}

export type WorkerHandler<T extends JobBusinessData> = (input: {
  shopId: string;
  jobRowId: string;
  businessData: T;
}) => Promise<unknown>;

/**
 * Register a worker for a given JobType. The handler receives the unwrapped
 * envelope and is responsible for business logic only — status bookkeeping
 * (active / completed / failed / attempts / lastError) is handled here.
 *
 * pg-boss v10 calls handlers with `Job[]`. We set `batchSize: 1` so we always
 * deal with a single job per invocation.
 */
export async function registerWorker<T extends JobBusinessData>(
  type: JobTypeValue,
  handler: WorkerHandler<T>,
): Promise<void> {
  const boss = await getBoss();
  await ensureQueue(boss, type);

  await boss.work<JobEnvelope<T>>(
    type,
    { batchSize: 1 },
    async (jobs) => {
      const job = jobs[0];
      if (!job) return;
      const { jobRowId, businessData } = job.data;

      await db.backgroundJob.update({
        where: { id: jobRowId },
        data: {
          status: "active",
          attempts: { increment: 1 },
        },
      });

      try {
        const result = await handler({
          shopId: businessData.shopId,
          jobRowId,
          businessData,
        });

        await db.backgroundJob.update({
          where: { id: jobRowId },
          data: {
            status: "completed",
            completedAt: new Date(),
            businessResult: (result ?? null) as unknown as object,
            lastError: null,
          },
        });
      } catch (error) {
        const msg = error instanceof Error ? error.message : String(error);
        await db.backgroundJob.update({
          where: { id: jobRowId },
          data: {
            status: "failed",
            lastError: msg,
          },
        });
        await adminLog("error", `Worker ${type} failed: ${msg}`, {
          shopId: businessData.shopId,
          jobRowId,
        });
        // Re-throw so pg-boss applies retry policy.
        throw error;
      }
    },
  );
}

/**
 * Boot all workers. Call once at module load (gated by env). Idempotent.
 */
export async function bootWorkers(): Promise<void> {
  if (workersBooted) return;
  workersBooted = true;

  // Lazy import to avoid a circular dep with workers.server.ts.
  const { registerAllWorkers } = await import("./workers.server");
  try {
    await registerAllWorkers();
  } catch (error) {
    workersBooted = false; // allow a retry next time
    const msg = error instanceof Error ? error.message : String(error);
    await adminLog("error", `bootWorkers failed: ${msg}`);
    throw error;
  }
}

// --- Auto-boot on production / when explicitly enabled ---
// The Remix server only ever loads this module once per process, but we still
// guard with `workersBooted` because dev HMR re-evaluates server modules.
if (
  process.env.WORKERS_ENABLED === "true" ||
  process.env.NODE_ENV === "production"
) {
  void bootWorkers().catch((err) => {
    console.error("[jobs] bootWorkers crashed:", err);
  });
}