// Outbox drain shared by the CDC dispatch executor (inline, right after the plan
// commits) and the insurance re-drain cron (scan for stranded rows). Delivery
// runs each adapter call outside any transaction; per-row errors are isolated so
// one stuck row never aborts the wider drain.

import { logger } from "@tailor-platform/sdk/runtime";
import type { DB } from "../generated/kysely-tailordb";
import type { DispatchAdapters } from "../lib/dispatchAdapters";
import { errorAttrs } from "../lib/logAttrs";
import { deliverDestinationLog } from "./deliverDestination";
import { deliverNotification } from "./deliverNotification";

export interface PlannedDeliveryIds {
  notificationIds: string[];
  destinationLogIds: string[];
}

/**
 * Which drain path a delivery failure came from: the CDC executor's inline
 * delivery, or the insurance cron sweeping stranded rows. Logged as an
 * attribute so the two can be told apart in the telemetry backend.
 */
type DrainPhase = "delivery" | "redrain";

async function drainDestinationLogs(
  db: DB,
  ids: string[],
  destination: NonNullable<DispatchAdapters["destination"]>,
  now: Date,
  phase: DrainPhase,
): Promise<number> {
  let failureCount = 0;
  for (const id of ids) {
    try {
      await deliverDestinationLog(db, id, destination, now);
    } catch (error) {
      failureCount += 1;
      logger.error("failed to deliver destination log", {
        phase,
        destinationLogId: id,
        ...errorAttrs(error),
      });
    }
  }
  return failureCount;
}

/** Deliver the outbox rows a single plan produced (CDC inline path). */
export async function deliverPlanned(
  db: DB,
  planned: PlannedDeliveryIds,
  adapters: DispatchAdapters,
  now: Date,
): Promise<void> {
  for (const id of planned.notificationIds) {
    try {
      await deliverNotification(db, id, adapters, now);
    } catch (error) {
      logger.error("failed to deliver notification", {
        phase: "delivery",
        notificationId: id,
        ...errorAttrs(error),
      });
    }
  }
  if (adapters.destination) {
    const destination = adapters.destination;
    await drainDestinationLogs(db, planned.destinationLogIds, destination, now, "delivery");
  }
}

/**
 * Drain stranded outbox rows: QUEUED Notifications and PENDING
 * DestinationDeliveryLogs whose plan committed but whose delivery never
 * completed (a crashed CDC executor). Returns the count of rows that errored.
 *
 * `staleBefore` is a grace cutoff: only rows created before it are swept. This
 * keeps the insurance cron from racing the CDC executor's inline delivery of a
 * row it *just* planned (both would call the provider for the same row, and the
 * conditional status update only serializes the write, not the external send).
 * A genuinely stranded row ages past the cutoff and is then re-driven; the
 * delivery worker passes the per-row idempotency key, but destination delivery
 * remains at-least-once for a provider that does not honor it (see
 * internal/deliverDestination).
 */
export async function drainStuckDeliveries(
  db: DB,
  adapters: DispatchAdapters,
  now: Date,
  limit: number,
  staleBefore: Date,
): Promise<number> {
  let failureCount = 0;

  const queued = await db
    .selectFrom("Notification")
    .select("id")
    .where("deliveryStatus", "=", "QUEUED")
    .where("createdAt", "<", staleBefore)
    .orderBy("createdAt", "asc")
    .limit(limit)
    .execute();
  for (const { id } of queued) {
    try {
      await deliverNotification(db, id, adapters, now);
    } catch (error) {
      failureCount += 1;
      logger.error("failed to deliver notification", {
        phase: "redrain",
        notificationId: id,
        ...errorAttrs(error),
      });
    }
  }

  if (adapters.destination) {
    const destination = adapters.destination;
    const pending = await db
      .selectFrom("DestinationDeliveryLog")
      .select("id")
      .where("status", "=", "PENDING")
      .where("createdAt", "<", staleBefore)
      .orderBy("createdAt", "asc")
      .limit(limit)
      .execute();
    failureCount += await drainDestinationLogs(
      db,
      pending.map((p) => p.id),
      destination,
      now,
      "redrain",
    );
  }

  return failureCount;
}