import type { CommandContext } from "@tailor-platform/erp-kit/core";
import type { MachineUserName } from "@tailor-platform/sdk";
import { createExecutor, scheduleTrigger } from "@tailor-platform/sdk";
import { logger } from "@tailor-platform/sdk/runtime";
import type { DB, Transaction } from "../generated/kysely-tailordb";
import { drainStuckDeliveries } from "../internal/drainDeliveries";
import { planPersistedEvent, type PersistedEventRow } from "../internal/planEvent";
import type { DispatchAdapters } from "../lib/dispatchAdapters";
import { errorAttrs } from "../lib/logAttrs";
import { permissions } from "../lib/permissions.generated";

type ScheduleTriggerOptions = Parameters<typeof scheduleTrigger>[0];
type Timezone = ScheduleTriggerOptions["timezone"];

/** Default cap on events re-drained per cycle, to bound a runaway backlog. */
const DEFAULT_REDRAIN_LIMIT = 100;

/**
 * Default grace window before a stranded outbox row is swept. A row younger than
 * this is assumed to still be in flight on the CDC executor's inline delivery
 * path, so sweeping it would race that delivery (a duplicate provider send).
 */
const DEFAULT_STALE_AFTER_MS = 60_000;

/**
 * Insurance trigger: a low-frequency cron with two duties, the safety net for a
 * dropped/missed CDC fire on the primary path (dispatch-notification-events):
 *
 *   1. Re-plan NotificationEvents still stuck in PENDING (the plan never
 *      committed). Each candidate is re-read `forUpdate` and re-checked for
 *      `status = PENDING` inside its own transaction, so it never races the CDC
 *      executor or another cron tick.
 *   2. Drain stranded outbox rows — QUEUED Notifications and PENDING
 *      DestinationDeliveryLogs whose plan committed but whose delivery never
 *      completed (a crashed CDC executor).
 *
 * Stranded rows are only swept once they age past a grace window
 * (`staleAfterMs`), so the cron never races the CDC executor's inline delivery
 * of a row it just planned. Delivery is at-least-once: the worker passes a
 * per-row idempotency key, but a provider that does not honor it (Slack
 * `chat.postMessage` has no idempotency parameter) can still receive a duplicate
 * if a process crashes after the provider post but before the row's terminal
 * status write. Per the module's delivery stance the CDC stream is not
 * separately verified; this cron is the backstop for a dropped CDC fire.
 */
export interface RedrainNotificationEventsParams<Cron extends string> {
  /**
   * Cron expression (standard 5-field). Must be a string literal so the SDK's
   * type-level CRON validator can check it. A low cadence (e.g. every 15
   * minutes) is appropriate — this is a backstop, not the primary path.
   */
  cron: Cron;
  /** IANA timezone name. */
  timezone?: Timezone;
  /** Returns the app-side Kysely instance for the shared DB. */
  getDB: () => DB;
  /** Composed channel adapters (same instances used by dispatchNotification). */
  adapters: DispatchAdapters;
  /** Max events re-drained per cycle (default 100). */
  limit?: number;
  /**
   * Grace window (ms) before a stranded outbox row is swept; rows younger than
   * this are left for the CDC executor's inline delivery (default 60s). Set
   * higher if inline delivery can legitimately take longer than a minute.
   */
  staleAfterMs?: number;
  /** Machine user identity used to invoke the executor. */
  invoker?: MachineUserName;
}

export function redrainNotificationEvents<const Cron extends string>(
  params: RedrainNotificationEventsParams<Cron>,
) {
  const limit = params.limit ?? DEFAULT_REDRAIN_LIMIT;
  const staleAfterMs = params.staleAfterMs ?? DEFAULT_STALE_AFTER_MS;
  return createExecutor({
    name: "notification-redrain-events",
    description:
      "Insurance cron: re-drains NotificationEvents stuck in PENDING (a missed CDC fire) by re-invoking dispatch; never touches DISPATCHED/NO_DELIVERY events",
    trigger: scheduleTrigger({
      // oxlint-disable-next-line typescript/no-unsafe-type-assertion
      cron: params.cron as never,
      timezone: params.timezone,
    }),
    operation: {
      kind: "jobFunction",
      invoker: params.invoker,
      body: async () => {
        const db = params.getDB();
        const now = new Date();
        const ctx: CommandContext = {
          actorId: "notification-redrain-runner",
          permissions: [permissions.ingest.logNotificationEvent],
        };

        const stuck = (await db
          .selectFrom("NotificationEvent")
          .select("id")
          .where("status", "=", "PENDING")
          .orderBy("createdAt", "asc")
          .limit(limit)
          .execute()) as Array<{ id: string }>;

        let failureCount = 0;
        // Duty 1: re-plan stuck events. The outbox rows this produces are picked
        // up by the drain below (which runs after, in the same cycle).
        for (const { id } of stuck) {
          try {
            const outcome = await db.transaction().execute(async (trx: Transaction) => {
              const event = (await trx
                .selectFrom("NotificationEvent")
                .selectAll()
                .where("id", "=", id)
                .forUpdate()
                .executeTakeFirst()) as PersistedEventRow | undefined;

              // The CDC executor (or a prior tick) may have planned it between
              // the scan and the lock: only act while still PENDING.
              if (!event || event.status !== "PENDING") return undefined;

              return planPersistedEvent(trx, event, params.adapters, ctx, now);
            });
            // Surface why nothing was produced instead of silently stamping it.
            if (outcome?.status === "NO_DELIVERY") {
              logger.warn("notification event stamped NO_DELIVERY", {
                executor: "notification-redrain-events",
                eventId: id,
                reason: outcome.reason ?? "no deliverable recipients or destinations",
              });
            }
          } catch (error) {
            failureCount += 1;
            logger.error("failed to re-plan notification event", {
              executor: "notification-redrain-events",
              eventId: id,
              ...errorAttrs(error),
            });
          }
        }

        // Duty 2: deliver stranded outbox rows outside any transaction, but only
        // those aged past the grace window so we never race the CDC executor's
        // inline delivery of a freshly planned row.
        const staleBefore = new Date(now.getTime() - staleAfterMs);
        failureCount += await drainStuckDeliveries(db, params.adapters, now, limit, staleBefore);
        if (failureCount > 0) {
          logger.warn("redrain cycle finished with failures", {
            executor: "notification-redrain-events",
            failureCount,
          });
          throw new Error(`notification-redrain-events: ${failureCount} re-drains failed`);
        }
      },
    },
  });
}