ApiCore/ApiCore-Backend/src/common/outbox/outbox-claims.ts

import { Prisma } from '../../../generated/prisma';
import { DbClient } from '../../prisma/db-client';
import { PrismaService } from '../../prisma/prisma.service';
import { MAX_ATTEMPTS, retryDelayMs, STALE_CLAIM_MS } from '../retry-policy';

/**
 * Claim, settle and stale release for every outbox table. One worker owns
 * an entry from the claim (status PROCESSING, claimedAt = now()) until it
 * settles it; settling is a compare-and-set on id + claimedAt, so a worker
 * whose claim was released in the meantime changes nothing. Every time
 * comparison and calculation runs in SQL against now() - the database
 * clock - never against a JS Date.
 */
export type OutboxTable =
  'DomainEvent' | 'WebhookDelivery' | 'NotificationDelivery';

export type OutboxOrderColumn = 'occurredAt' | 'createdAt';

export interface ClaimOptions {
  limit: number;
  /** Extra condition, ANDed to the claim (e.g. a correlation id filter). */
  filter?: Prisma.Sql;
  orderBy: OutboxOrderColumn;
}

interface OutboxTableSql {
  table: Prisma.Sql;
  statusType: Prisma.Sql;
  settledAtColumn: Prisma.Sql;
  successStatus: 'DONE' | 'DELIVERED';
}

/** Outbox tables whose entries can be put back or skipped by the sender. */
export type DeliveryOutboxTable = 'NotificationDelivery';

/**
 * Table, enum and column names cannot be bound as parameters. They come
 * only from these closed maps - never from caller input - which is what
 * makes Prisma.raw safe here.
 */
const OUTBOX_TABLES: Readonly<Record<OutboxTable, OutboxTableSql>> = {
  DomainEvent: {
    table: Prisma.raw('"DomainEvent"'),
    statusType: Prisma.raw('"DomainEventStatus"'),
    settledAtColumn: Prisma.raw('"processedAt"'),
    successStatus: 'DONE',
  },
  WebhookDelivery: {
    table: Prisma.raw('"WebhookDelivery"'),
    statusType: Prisma.raw('"WebhookDeliveryStatus"'),
    settledAtColumn: Prisma.raw('"deliveredAt"'),
    successStatus: 'DELIVERED',
  },
  NotificationDelivery: {
    table: Prisma.raw('"NotificationDelivery"'),
    statusType: Prisma.raw('"NotificationDeliveryStatus"'),
    settledAtColumn: Prisma.raw('"deliveredAt"'),
    successStatus: 'DELIVERED',
  },
};

const ORDER_COLUMNS: Readonly<Record<OutboxOrderColumn, Prisma.Sql>> = {
  occurredAt: Prisma.raw('"occurredAt"'),
  createdAt: Prisma.raw('"createdAt"'),
};

/** Longest wait deferClaim allows: a day. */
const MAX_DEFER_MS = 24 * 60 * 60 * 1000;
const MAX_SKIP_REASON_CHARS = 200;

const STALE_CLAIM_ERROR = 'claim expired (worker restarted?)';

function outboxTable(table: OutboxTable): OutboxTableSql {
  if (!Object.hasOwn(OUTBOX_TABLES, table)) {
    throw new Error(`Unknown outbox table "${String(table)}"`);
  }
  return OUTBOX_TABLES[table];
}

function orderColumn(column: OutboxOrderColumn): Prisma.Sql {
  if (!Object.hasOwn(ORDER_COLUMNS, column)) {
    throw new Error(`Unknown outbox order column "${String(column)}"`);
  }
  return ORDER_COLUMNS[column];
}

/**
 * Claims up to `limit` due PENDING entries, oldest first. "availableAt" is
 * TIMESTAMP(3), so a value written from now() is rounded to the nearest
 * millisecond - up to 0.5 ms ahead of the clock. Comparing against now()
 * rounded the same way keeps such an entry due as soon as it is written;
 * a plain now() would miss it for the first fraction of a millisecond.
 * SKIP LOCKED lets several workers claim side by side without waiting for
 * or overlapping each other. Returns the claimed rows (all columns) in claim order.
 * Uses the root client: a claim must commit on its own.
 */
export async function claimBatch<Row>(
  db: PrismaService,
  table: OutboxTable,
  options: ClaimOptions,
): Promise<Row[]> {
  if (!Number.isInteger(options.limit) || options.limit < 1) {
    throw new RangeError(
      `Claim limit must be a positive integer, got ${options.limit}`,
    );
  }
  const { table: name } = outboxTable(table);
  const orderBy = orderColumn(options.orderBy);
  const filter = options.filter
    ? Prisma.sql`AND ${options.filter}`
    : Prisma.empty;
  return db.$queryRaw<Row[]>(Prisma.sql`
    WITH claimed AS (
      UPDATE ${name} SET "status" = 'PROCESSING', "claimedAt" = now()
      WHERE "id" IN (
        SELECT "id" FROM ${name}
        WHERE "status" = 'PENDING' AND "availableAt" <= CAST(now() AS timestamp(3)) ${filter}
        ORDER BY ${orderBy}
        LIMIT ${options.limit}
        FOR UPDATE SKIP LOCKED
      )
      RETURNING *
    )
    SELECT * FROM claimed ORDER BY ${orderBy}`);
}

/** An entry a stale release put back (PENDING) or gave up on (FAILED). */
export interface ReleasedClaim {
  id: string;
  status: 'PENDING' | 'FAILED';
}

/**
 * Puts entries whose claim is older than STALE_CLAIM_MS (their worker died)
 * back to PENDING right away, with attempts + 1 - or to FAILED once that
 * reaches MAX_ATTEMPTS, so an entry that kills its worker cannot loop
 * forever. Returns the released entries, for follow-up work on FAILED ones.
 */
export async function releaseStaleClaimRows(
  db: PrismaService,
  table: OutboxTable,
): Promise<ReleasedClaim[]> {
  const { table: name, statusType } = outboxTable(table);
  return db.$queryRaw<ReleasedClaim[]>(Prisma.sql`
    UPDATE ${name}
    SET "status" = CAST(CASE WHEN "attempts" + 1 >= ${MAX_ATTEMPTS} THEN 'FAILED' ELSE 'PENDING' END AS ${statusType}),
        "attempts" = "attempts" + 1,
        "claimedAt" = NULL,
        "availableAt" = now(),
        "lastError" = ${STALE_CLAIM_ERROR}
    WHERE "status" = 'PROCESSING'
      AND "claimedAt" < now() - (${STALE_CLAIM_MS} * interval '1 millisecond')
    RETURNING "id", "status"`);
}

/** releaseStaleClaimRows for callers that only need the count. */
export async function releaseStaleClaims(
  db: PrismaService,
  table: OutboxTable,
): Promise<number> {
  return (await releaseStaleClaimRows(db, table)).length;
}

/**
 * Settles a claimed entry as done. Returns false if the claim was lost
 * (released as stale, maybe claimed again) - the current owner settles it.
 * lastError stays as history of earlier failures.
 */
export async function settleSuccess(
  db: DbClient,
  table: OutboxTable,
  id: string,
  claimedAt: Date,
  status: 'DONE' | 'DELIVERED',
): Promise<boolean> {
  const sql = outboxTable(table);
  if (status !== sql.successStatus) {
    throw new Error(`${table} settles as ${sql.successStatus}, not ${status}`);
  }
  const count = await db.$executeRaw(Prisma.sql`
    UPDATE ${sql.table}
    SET "status" = CAST(${status} AS ${sql.statusType}),
        "claimedAt" = NULL,
        ${sql.settledAtColumn} = now()
    WHERE "id" = ${id} AND "status" = 'PROCESSING' AND "claimedAt" = ${claimedAt}`);
  return count === 1;
}

/**
 * Settles a claimed entry after a failed attempt: back to PENDING after
 * retryDelayMs(attempts) by the database clock, or FAILED once attempts
 * reaches MAX_ATTEMPTS. `lastError` must come from loggableError().
 *
 * The outcome also holds if the claim was lost: only a stale release
 * changes a claimed row, and it applies the same attempts + 1 >=
 * MAX_ATTEMPTS rule to the same attempt count.
 */
export async function settleFailure(
  db: DbClient,
  table: OutboxTable,
  id: string,
  claimedAt: Date,
  attemptsBefore: number,
  lastError: string,
): Promise<'retry' | 'failed'> {
  const { table: name, statusType } = outboxTable(table);
  const attempts = attemptsBefore + 1;
  const outcome = attempts >= MAX_ATTEMPTS ? 'failed' : 'retry';
  const nextAttempt =
    outcome === 'retry'
      ? Prisma.sql`, "availableAt" = now() + (${retryDelayMs(attempts)} * interval '1 millisecond')`
      : Prisma.empty;
  await db.$executeRaw(Prisma.sql`
    UPDATE ${name}
    SET "status" = CAST(${outcome === 'retry' ? 'PENDING' : 'FAILED'} AS ${statusType}),
        "attempts" = "attempts" + 1,
        "claimedAt" = NULL,
        "lastError" = ${lastError}
        ${nextAttempt}
    WHERE "id" = ${id} AND "status" = 'PROCESSING' AND "claimedAt" = ${claimedAt}`);
  return outcome;
}

/**
 * Puts a claimed entry back to PENDING, due after `delayMs` by the database
 * clock, without counting an attempt: the entry did not fail, it has to wait
 * (rate limit from the receiver, quiet period, paused bot); the delay is
 * capped at 24 hours. Compare-and-set
 * on id + claimedAt; false if the claim was lost.
 */
export async function deferClaim(
  db: DbClient,
  table: DeliveryOutboxTable,
  id: string,
  claimedAt: Date,
  delayMs: number,
): Promise<boolean> {
  if (!Number.isInteger(delayMs) || delayMs < 0) {
    throw new RangeError(
      `Defer delay must be a non-negative integer, got ${delayMs}`,
    );
  }
  const { table: name } = outboxTable(table);
  const cappedDelayMs = Math.min(delayMs, MAX_DEFER_MS);
  const count = await db.$executeRaw(Prisma.sql`
    UPDATE ${name}
    SET "status" = 'PENDING',
        "claimedAt" = NULL,
        "availableAt" = now() + (${cappedDelayMs} * interval '1 millisecond')
    WHERE "id" = ${id} AND "status" = 'PROCESSING' AND "claimedAt" = ${claimedAt}`);
  return count === 1;
}

/**
 * Settles a claimed entry as SKIPPED: it will never be sent (the recipient
 * unlinked, the message went stale). `reason` lands in lastError and must
 * not contain personal data or secrets; it is cut to 200 characters. Compare-and-set on id + claimedAt;
 * false if the claim was lost.
 */
export async function settleSkipped(
  db: DbClient,
  table: DeliveryOutboxTable,
  id: string,
  claimedAt: Date,
  reason: string,
): Promise<boolean> {
  const { table: name, statusType } = outboxTable(table);
  const count = await db.$executeRaw(Prisma.sql`
    UPDATE ${name}
    SET "status" = CAST(${'SKIPPED'} AS ${statusType}),
        "claimedAt" = NULL,
        "lastError" = ${reason.slice(0, MAX_SKIP_REASON_CHARS)}
    WHERE "id" = ${id} AND "status" = 'PROCESSING' AND "claimedAt" = ${claimedAt}`);
  return count === 1;
}