medplum / medplum / 28812604563

// SPDX-FileCopyrightText: Copyright Orangebot, Inc. and Medplum contributors

// SPDX-License-Identifier: Apache-2.0

import type { ILogger } from '@medplum/core';

import { normalizeErrorString } from '@medplum/core';

import { randomUUID } from 'node:crypto';

import { chmodSync, existsSync } from 'node:fs';

import type { DatabaseSync, SQLInputValue, StatementSync } from 'node:sqlite';

import {

  CHANNEL_DEPTH,

  CHECKPOINT_WAL,

  CLAIM_NEXT,

  COMMIT_SEQ_NO,

  COUNT_BY_STATE,

  DB_SIZE_BYTES,

  ENQUEUE,

  ENQUEUE_REJECTED,

  FIND_BY_CALLBACK,

  FIND_BY_ID,

  FIND_SEEN_BY_CONTROL_ID,

  GET_LEASE,

  HEARTBEAT_LEASE,

  LIST_QUEUED_IDS_FOR_CHANNEL,

  MARK_BOT_FAILED,

  MARK_PROCESSED,

  MARK_SENT,

  PEEK_LAST_SEQ_NO,

  RECORD_SERVER_RESPONSE,

  RECOVER_CLAIMED,

  RECOVER_INFLIGHT,

  RECOVER_INFLIGHT_GUARANTEED,

  RELEASE_LEASE,

  REQUEUE,

  SCHEDULE_RETRY,

  SET_ACK_OUTCOME,

  TRY_ACQUIRE_LEASE,

} from './queries';

import { runMigrations } from './schema';

import type {

  AckOutcome,

  ClaimedRow,

  EnqueueInput,

  EnqueueRejectedInput,

  EnqueueResult,

  InboundRow,

  MessageState,

  NackedRow,

  QueueErrorCode,

} from './types';

import {

  AckOutcome as AckOutcomeValues,

  assertRowState,

  MessageState as MessageStateValues,

  QueueLeaseError,

} from './types';

/**

 * Default lease lifetime. Picked to be:

 * - Long enough that ordinary GC pauses or slow disks don't cost us the lease

 *   under load.

 * - Short enough that, when an agent dies un-gracefully, a peer can take over

 *   within tens of seconds rather than minutes.

 */

/**

 * How often the leaseholder re-extends its lease. Must be comfortably less than

 * {@link DEFAULT_LEASE_TTL_MS} so transient delays don't let the lease expire.

 */

/**

 * How often a non-leader retries acquisition. Shorter is fine — the underlying

 * SQL is a single conditional UPSERT and is cheap.

 */

/**

 * How often the queue folds its WAL back into the main DB file. SQLite only

 * checkpoints piggybacked on commits, so once traffic stops nothing else would

 * drain the WAL until the next retention sweep or close(); this loop bounds how

 * far the main DB file can lag behind the last write. Matches the agent

 * heartbeat cadence — the loop's former home, before the queue took ownership.

 */

export interface DurableQueueOptions {

  /** Filesystem path to the SQLite DB file. */

  path: string;

  /** Logger for migration / lifecycle messages. */

  log: ILogger;

  /** Override the per-process dispatch-lease holder ID. Defaults to a fresh UUID. */

  leaseHolder?: string;

  /** Override the dispatch-lease TTL in ms. */

  leaseTtlMs?: number;

  /** Override the dispatch-lease heartbeat interval in ms. */

  leaseHeartbeatMs?: number;

  /** Override the dispatch-lease acquire retry interval in ms. */

  leaseAcquireRetryMs?: number;

  /** Override the WAL-checkpoint loop interval in ms. */

  checkpointIntervalMs?: number;

}

/** Counts by lifecycle state — used by stats and the retention sweeper. */

export type StateCounts = Record<MessageState, number>;

/** Per-channel queue depth snapshot returned by {@link DurableQueue.getChannelDepths}. */

export interface ChannelDepth {

  queued: number;

  /** Claimed by a worker but not yet written to the socket. */

  claimed: number;

  /** Written to the socket, awaiting the server response. */

  inflight: number;

  oldestQueuedAgeMs: number | null;

}

/**

 * SQLite-backed durable FIFO for inbound HL7 messages.

 *

 * One instance owns one synchronous `node:sqlite` connection plus a bag of

 * prepared statements covering the hot path. All channel intake and worker

 * dispatch goes through this object. See DURABLE_QUEUE_ARCHITECTURE.md §3, §5.

 *

 * The class is designed to be opened once at agent startup, used for the

 * lifetime of the process, and closed exactly once during {@link DurableQueue.close}.

 */

export class DurableQueue {

  private readonly db: DatabaseSync;

  private readonly log: ILogger;

  private readonly path: string;

  // The dispatch-lease holder ID of THIS process, set when startDispatchLease()

  // begins the acquire/heartbeat loop (or bound directly via setLeaseHolder in

  // tests). Dispatch-class mutations are gated on the live lease still belonging

  // to this holder; when a peer takes over, the gate throws QueueLeaseError so the

  // demoted worker stops driving the queue. Undefined when the dispatch lease was

  // never started (single-process use and most unit tests), in which case the gate

  // is inert because no lease row exists.

  private leaseHolderId: string | undefined;

  // ── Dispatch-lease orchestration (formerly DispatchLeaseManager) ──

  // Lightweight leader election for the DISPATCH path. The lease gates *dispatch

  // only* — claiming rows and driving them to the server (the worker +

  // recoverOnStartup). It deliberately does NOT gate intake (followers still

  // persist inbound messages via enqueue), maintenance (WAL checkpoint, retention

  // sweep), or diagnostics (stats): those are valid on any process with the queue

  // file open. Exactly one process may dispatch at a time, which is the

  // load-bearing primitive that makes zero-downtime upgrades safe: during the

  // overlap window where the old and new agent processes both have the SQLite file

  // open, only the leaseholder claims/sends rows; the non-leader keeps accepting

  // and persisting inbound traffic but does not dispatch.

  //

  // There is deliberately NO lost-leadership callback: loss is enforced at the

  // data layer instead — the dispatch ops throw QueueLeaseError once a peer holds

  // the lease (see assertNotDemoted), so the worker self-detects and drains

  // without pushing an event down a chain of callbacks.

  private readonly leaseTtlMs: number;

  private readonly leaseHeartbeatMs: number;

  private readonly leaseAcquireRetryMs: number;

  private leaseAcquireTimer: NodeJS.Timeout | undefined;

  private leaseHeartbeatTimer: NodeJS.Timeout | undefined;

  private onBecameLeader: (() => void) | undefined;

  // True when the WAL may contain frames not yet checkpointed into the main DB

  // file. SQLite only attempts checkpoints piggybacked on commits, so once

  // traffic stops nothing would ever drain the WAL — the queue's own checkpoint

  // loop (startCheckpointLoop) polls this via checkpointWalIfDirty(). Starts true

  // because open() itself writes (pragmas, migrations, lease).

  // WAL-checkpoint loop. Runs while the queue is open, INDEPENDENT of the

  // dispatch lease: a follower still accepts intake writes (enqueue runs on any

  // process with the file open), so WAL draining can't be gated on leadership the

  // way the lease timers are. Started in open(), cleared in close().

  private readonly checkpointIntervalMs: number;

  private checkpointTimer: NodeJS.Timeout | undefined;

  // Prepared statements — created once at open(), reused for every call.

  // Names mirror the public methods that use them.

  private readonly enqueueStmt: StatementSync;

  private readonly enqueueRejectedStmt: StatementSync;

  private readonly peekLastSeqNoStmt: StatementSync;

  private readonly commitSeqNoStmt: StatementSync;

  private readonly findSeenByControlIdStmt: StatementSync;

  private readonly claimNextStmt: StatementSync;

  private readonly markSentStmt: StatementSync;

  private readonly findByCallbackStmt: StatementSync;

  private readonly findByIdStmt: StatementSync;

  private readonly recordServerResponseStmt: StatementSync;

  private readonly markProcessedStmt: StatementSync;

  private readonly markBotFailedStmt: StatementSync;

  private readonly setAckOutcomeStmt: StatementSync;

  private readonly scheduleRetryStmt: StatementSync;

  private readonly requeueStmt: StatementSync;

  private readonly recoverInflightStmt: StatementSync;

  private readonly recoverInflightGuaranteedStmt: StatementSync;

  private readonly recoverClaimedStmt: StatementSync;

  private readonly listQueuedIdsForChannelStmt: StatementSync;

  private readonly countByStateStmt: StatementSync;

  private readonly channelDepthStmt: StatementSync;

  private readonly tryAcquireLeaseStmt: StatementSync;

  private readonly heartbeatLeaseStmt: StatementSync;

  private readonly releaseLeaseStmt: StatementSync;

  private readonly getLeaseStmt: StatementSync;

  private readonly checkpointStmt: StatementSync;

  private readonly dbSizeBytesStmt: StatementSync;

  constructor(db: DatabaseSync, options: DurableQueueOptions) {

    // Single-writer, durable-across-crash, WAL-friendly settings (§5).

    // Picked for HL7 inbound rates and the durability contract we owe the sender:

    //   synchronous=NORMAL gives crash safety without per-write fsync;

    //   WAL lets workers read while intake writes;

    //   mmap_size keeps the read path off the kernel I/O hot path.

      PRAGMA journal_mode = WAL;

      PRAGMA synchronous  = NORMAL;

      PRAGMA temp_store   = MEMORY;

      PRAGMA cache_size   = -65536;

      PRAGMA mmap_size    = 268435456;

      PRAGMA wal_autocheckpoint = 1000;

      PRAGMA busy_timeout = 5000;

      PRAGMA foreign_keys = ON;

    `);

    // Per-channel monotonic sequence counter (assignSeqNo). Single-process +

    // synchronous, so no locking is needed between peek and commit.

    // Canonical prior row for a (channel, control_id): any state except `nacked`

    // (nacked rows are rejected-intake audit records and intentionally reuse

    // control IDs, so they're not a "real" prior delivery to dedupe against).

    // id DESC returns the most recent in the unlikely event legacy duplicates

    // predate the dedupe-on-intake logic.

    // FIFO claim: take the lowest-id queued row for this channel, flip it to

    // `claimed` in the same statement so concurrent workers can't double-claim.

    // node:sqlite is synchronous and the agent is single-process, so RETURNING is

    // enough — no advisory locking needed.

    // Phase A → B: the App's send path calls this the moment the transmit request

    // is written to the socket, flipping `claimed` → `inflight` and stamping sent_at.

    // Auto-retry transition: a claimed/inflight row goes back to `queued` with a

    // future next_attempt_at so the channel's FIFO head blocks (and the row is

    // not re-claimed) until the backoff elapses. See SCHEDULE_RETRY.

    // Crash recovery splits three ways on whether the request reached the wire

    // and the channel's guaranteed-delivery setting:

    //   recoverClaimed — rows left `claimed` provably never reached the server,

    //     so they always return to `queued` for a clean re-dispatch with no

    //     duplicate risk (regardless of guaranteed delivery).

    //   recoverInflight — rows left `inflight` on a NORMAL channel are ambiguous

    //     (the server may or may not have processed them), so they land in

    //     `failed` for operator review, never silently retried. The source leg's

    //     ack_outcome is left as-is (`pending`): we don't know whether an ACK was owed.

    //   recoverInflightGuaranteed — rows left `inflight` on a GUARANTEED channel

    //     return to `queued` (duplication risk accepted) so the channel keeps

    //     trying until upstream gives a definitive answer (§4.1).

    // Prepared once like every other statement — checkpointWal() runs on every

    // checkpoint-loop tick and every retention sweep, so re-compiling it per call

    // would leak GC-finalised statement objects proportional to that frequency.

    // Prepared once like the rest — the retention sweeper calls getDbSizeBytes()

    // in a tight loop while purging under size pressure, so re-compiling it per

    // call would leak GC-finalised statement objects proportional to sweep work.

  }

  /**

   * Opens (or creates) the DB file at `options.path`, runs migrations, and

   * returns a ready DurableQueue.

   *

   * Constructs synchronously — `node:sqlite` is itself synchronous so there's

   * no benefit to making this `async`.

   *

   * @param options - Path + logger.

   * @returns The opened DurableQueue.

   */

  static open(options: DurableQueueOptions): DurableQueue {

    // Lazy require so this module can be imported in environments where

    // node:sqlite isn't available (e.g. type-only tooling).

    // eslint-disable-next-line @typescript-eslint/no-require-imports, @typescript-eslint/consistent-type-imports

    // Lock the DB file down to the agent's user. Best-effort — fails silently on

    // platforms (Windows) where chmod is a no-op.

      } catch {

        // ignore

      }

    }

    // Start the WAL-checkpoint loop here (not in the constructor) so a bare

    // `new DurableQueue(db, opts)` — used by unit tests — stays timer-free; the

    // production lifecycle is open() … close(), and close() clears it.

  }

  /** Closes the underlying SQLite handle. Idempotent. */

  close(): void {

    }

    // Stop the WAL-checkpoint loop — the final flush below covers shutdown, and a

    // checkpointWalIfDirty() tick is a no-op once `closed` is set anyway.

    // Release the lease BEFORE closing the DB so a waiting peer can take over

    // immediately rather than waiting for our TTL to expire. Idempotent and a

    // no-op when the dispatch lease was never started.

    // SQLite only checkpoints + deletes the WAL on close when this is the last

    // connection to the file. An upgrade-overlap peer or an operator's sqlite3

    // shell defeats that, so flush explicitly — a clean shutdown should always

    // leave a self-contained main DB file (e.g. for file-level backups).

    } catch (err) {

    }

  }

  /**

   * Unconditionally runs a TRUNCATE checkpoint, folding the WAL into the main

   * DB file and truncating the WAL to zero bytes. Best-effort: failures are

   * logged, and an incomplete checkpoint (a peer connection pinning part of the

   * WAL) leaves the dirty flag set so the next attempt retries.

   * @returns True when the checkpoint fully completed.

   */

  checkpointWal(): boolean {

      }

    } catch (err) {

    }

  }

  /**

   * Runs {@link DurableQueue.checkpointWal} only when writes have landed since

   * the last completed checkpoint. Driven by the queue's own checkpoint loop

   * (see {@link DurableQueue.startCheckpointLoop}) every `checkpointIntervalMs`,

   * so the WAL drains shortly after traffic stops instead of waiting for the next

   * retention sweep or for close(). A no-op on an idle queue.

   * @returns True when a checkpoint ran and fully completed.

   */

  checkpointWalIfDirty(): boolean {

    }

  }

  /**

   * Starts the periodic WAL-checkpoint loop. Idempotent. Called once by

   * {@link DurableQueue.open}; runs regardless of dispatch leadership (see the

   * `checkpointTimer` field) and is torn down by {@link DurableQueue.close}.

   */

  private startCheckpointLoop(): void {

    }

    // Don't keep the event loop alive solely for housekeeping — same as the

    // dispatch-lease timers.

    }

  }

  /** Stops the WAL-checkpoint loop. Idempotent and a no-op when never started. */

  private stopCheckpointLoop(): void {

    }

  }

  /** @returns Filesystem path of the underlying SQLite file. */

  getPath(): string {

  }

  /** @returns Direct access to the underlying handle. Used by retention sweeper for PRAGMA queries. */

  getDb(): DatabaseSync {

  }

  /**

   * Runs `fn` inside a single SQLite transaction, committing if it returns and

   * rolling back if it throws. Used to make a row state transition atomic with an

   * external side effect — e.g. flipping a row `claimed` → `inflight` only if the

   * socket write that puts it on the wire also succeeds (see `App.sendToWebSocket`):

   * if the write throws, the marker rolls back and the row stays `claimed`

   * (provably unsent, safe to requeue) rather than a phantom `inflight`.

   *

   * `node:sqlite` is synchronous, so `fn` must be synchronous too — do any `await`

   * (e.g. token refresh) before calling, not inside.

   * @param fn - The work to run transactionally.

   * @returns Whatever `fn` returns.

   */

  runInTransaction<T>(fn: () => T): T {

    } catch (err) {

    }

  }

  /**

   * Inserts a new `queued` row.

   *

   * If any prior non-`nacked` row already owns this `(channel_name,

   * msg_control_id)` — in `queued`, `claimed`, `inflight`, `processed`,

   * `rejected`, or `failed` — we don't insert; we surface a `duplicate` result

   * carrying that row so the caller can compare bodies and decide between

   * replaying the prior ACK and rejecting the collision (§8). This SELECT is the

   * single dedup authority on the intake path: the partial unique index only

   * covers the live `queued`/`claimed`/`inflight` window, so it can't recognize a

   * retransmit of an already-`processed`/`rejected`/`failed` row —

   * it remains only as a last-resort guard against an active-window race (not

   * expected in the single-process agent).

   * @param input - Fields to persist on the new row.

   * @param options - Optional behavior.

   * @param options.assignSeqNo - When provided, enqueue assigns the channel's next

   *   sequence number — but only after the duplicate check passes, so a retransmit

   *   never consumes one. It peeks the persisted counter (a non-consuming read),

   *   calls this callback with the candidate so the caller can stamp it into MSH.13

   *   and return the finalized bytes to persist, then advances the counter in the

   *   SAME transaction as the insert (so a crash can't store the row while leaving

   *   the counter behind, which would reuse the number on restart). On a duplicate

   *   the callback is never invoked.

   * @returns Either the inserted row, or the prior row that owns the MSH.10.

   */

  enqueue(input: EnqueueInput, options?: { assignSeqNo?: (candidate: number) => Buffer }): EnqueueResult {

      }

    }

    // Not a duplicate — only now assign the sequence number (if configured). Peek

    // is non-consuming; the counter is advanced (commitSeqNo) inside the insert

    // transaction below, so a failed insert burns no number and a duplicate (above)

    // never reaches here.

    let seqNoToCommit: number | undefined;

    }

          input.channelName,

          input.remote,

          input.msgControlId,

          input.msgType,

          toBlob(input.originalMessage),

          toBlob(finalizedMessage),

          input.encoding,

          input.enhancedMode,

          input.callbackId,

          seqNo,

          input.receivedAt,

        );

      };

      let id: number;

        // Atomically insert the row and advance the sequence counter.

        } catch (txErr) {

        }

      } else {

      }

      }

      // The insert (just committed above) is what durably accepted the message —

      // this re-read only recovers the row for the caller. Don't assert it's

      // still 'queued': a peer process sharing this DB file may have already

      // claimed (or, in principle, settled) it between the commit and this

      // read. That's not a failed intake, so we return whatever state it's in

      // rather than throwing and having the caller NACK an already-committed message.

    } catch (err) {

        }

      }

    }

  }

  /**

   * Inserts an audit row in the `nacked` terminal state — used when intake

   * rejected the message (duplicate, malformed) but we still want a forensics

   * record. Failure to write this is non-fatal; the caller is expected to log.

   * @param input - Fields to persist plus the `lastError`/`errorCode` describing why we rejected.

   * @returns The newly inserted audit row, or null if even the audit insert failed.

   */

  enqueueRejected(input: EnqueueRejectedInput): NackedRow | null {

        input.channelName,

        input.remote,

        input.msgControlId,

        input.msgType,

        toBlob(input.originalMessage),

        toBlob(input.finalizedMessage),

        input.encoding,

        input.enhancedMode,

        input.callbackId,

        input.lastError,

        input.errorCode,

        input.seqNo,

        input.receivedAt

      );

      // ENQUEUE_REJECTED always inserts in the 'nacked' state.

    } catch (err) {

    }

  }

  /**

   * Atomically claims the next `queued` row for `channelName`, flipping it to

   * `claimed` and bumping `attempt_count`. The row stays `claimed` until

   * {@link markSent} flips it to `inflight` once the request hits the socket.

   * Honors a retry backoff: a `queued` row whose `next_attempt_at` is still in

   * the future is not handed out, and because the FIFO head-of-line is preserved

   * (the predicate is on the outer update, see {@link CLAIM_NEXT}) the whole

   * channel waits behind it rather than skipping ahead.

   * @param channelName - The channel to claim from.

   * @param now - Override the timestamp written to `processing_started_at` and

   *   compared against `next_attempt_at` (for tests).

   * @returns The claimed row, or `null` if the channel queue is empty or its head is backing off.

   */

    // Bind `now` twice: once for processing_started_at, once for the

    // next_attempt_at backoff predicate on the outer update.

    }

    // CLAIM_NEXT sets state='claimed' in the same statement (RETURNING *), so the

    // decoded row is always a ClaimedRow.

  }

  /**

   * Phase A → B transition: records that the transmit request for `callbackId`

   * was written to the WebSocket, flipping the row from `claimed` to `inflight`

   * and stamping `sent_at`. Called from the App's send path the instant the

   * request leaves the process — the durable marker that distinguishes a

   * provably-unsent row (safe to requeue on crash) from an ambiguous in-flight

   * one (failed for review). Guarded on `state = 'claimed'`, so it's a no-op for

   * legacy (non-durable) sends and for any row already past `claimed`.

   * @param callbackId - The callback ID of the transmit request just sent.

   * @param now - Override for `sent_at` (for tests).

   * @returns True if a `claimed` row was flipped to `inflight`.

   */

    }

  }

  /**

   * Looks up a row by its server-callback ID. Returns `null` if not found.

   * @param callbackId - The callback ID echoed back by the Medplum server.

   * @returns The matching row, or null.

   */

  findByCallback(callbackId: string): InboundRow | null {

  }

  /**

   * Looks up a row by its primary key.

   * @param id - The row's primary key.

   * @returns The matching row, or null.

   */

  getById(id: number): InboundRow | null {

  }

  /**

   * Records the server's `agent:transmit:response` body + status against the row.

   * @param id - Row primary key.

   * @param statusCode - HTTP-style status code returned by the server.

   * @param body - Response payload, or null when the server omitted one.

   */

  recordServerResponse(id: number, statusCode: number | null, body: Buffer | string | null): void {

  }

  /**

   * Terminal Bot-leg transition: the server accepted the message (2xx). The

   * source leg is recorded independently via `ackOutcome` — `delivered` when the

   * app-level ACK reached the source, `undelivered` when it couldn't (e.g. the

   * source connection had closed). An `undelivered` row is fully processed

   * upstream and must never be re-dispatched; it recovers when the source

   * retransmits and the stored ACK is replayed (see {@link setAckOutcome}).

   *

   * Attempt-scoped (see {@link MARK_PROCESSED}): a no-op if `attemptCount`

   * doesn't match the row's current attempt.

   * @param id - Row primary key.

   * @param attemptCount - The attempt this response answers; must match the

   *   row's current `attempt_count` for the write to apply.

   * @param ackOutcome - Source-leg result: `delivered` or `undelivered`.

   * @param now - Override for `processed_at` (for tests).

   * @returns True if the row was settled; false if this attempt was superseded.

   */

    }

  }

  /**

   * Terminal Bot-leg transition: the server **rejected** the message itself

   * (permanent 4xx). Retrying can never help — the content must be triaged.

   *

   * Attempt-scoped (see {@link MARK_PROCESSED}): a no-op if `attemptCount`

   * doesn't match the row's current attempt.

   * @param id - Row primary key.

   * @param attemptCount - The attempt this failure answers; must match the

   *   row's current `attempt_count` for the write to apply.

   * @param error - Human-readable error string, written to `last_error`.

   * @param errorCode - Machine-readable classification, written to `error_code`.

   * @param now - Override for `errored_at` (for tests).

   * @returns True if the row was settled; false if this attempt was superseded.

   */

  markRejected(

    id: number,

    attemptCount: number,

    error: string,

    errorCode: QueueErrorCode,

  ): boolean {

    }

  }

  /**

   * Terminal-for-now Bot-leg transition: a **transient/ambiguous** failure

   * (5xx, 429, response timeout, dispatch error, interrupted). The retry/

   * operator-review candidate — distinct from a `rejected` message so a future

   * retry policy can re-dispatch `failed` rows without ever touching `rejected`

   * ones (or `processed` + `undelivered` ones, whose Bot leg already succeeded).

   *

   * Attempt-scoped (see {@link MARK_PROCESSED}): a no-op if `attemptCount`

   * doesn't match the row's current attempt.

   * @param id - Row primary key.

   * @param attemptCount - The attempt this failure answers; must match the

   *   row's current `attempt_count` for the write to apply.

   * @param error - Human-readable error string, written to `last_error`.

   * @param errorCode - Machine-readable classification, written to `error_code`.

   * @param now - Override for `errored_at` (for tests).

   * @returns True if the row was settled; false if this attempt was superseded.

   */

  markFailed(

    id: number,

    attemptCount: number,

    error: string,

    errorCode: QueueErrorCode,

  ): boolean {

    }

  }

  /**

   * Updates only the source-leg {@link AckOutcome}, leaving the Bot-leg `state`

   * untouched. Used when a duplicate retransmit replays a previously

   * `undelivered` ACK and it lands — flipping the row to `delivered` so it no

   * longer reads as awaiting source delivery.

   * @param id - Row primary key.

   * @param ackOutcome - The new source-leg outcome.

   */

  setAckOutcome(id: number, ackOutcome: AckOutcome): void {

  }

  /**

   * Auto-retry transition: returns a `claimed`/`inflight`/`queued` row to

   * `queued`, scheduled to become claimable at `nextAttemptAt`. Because the row

   * keeps its id and claims are ordered by id, it sits at the head of its

   * channel's FIFO and blocks younger rows until it either succeeds or exhausts

   * its attempts — preserving per-channel ordering across retries.

   *

   * Distinct from {@link markFailed}: a retry stays `queued` (still in flight),

   * whereas `markFailed` is the terminal landing for a failure the policy is not

   * retrying. The worker decides which to call by gating the row's

   * {@link QueueErrorCode} against the retry policy (see worker.ts). Gated by the

   * dispatch lease like the other settle ops: a demoted process must not

   * reschedule. Attempt-scoped (see {@link MARK_PROCESSED}): also a no-op if

   * `attemptCount` doesn't match the row's current attempt — this is what lets

   * the late-response settle path (`onServerResponse` in worker.ts) call this on

   * an already-`queued` row without risking clobbering a newer attempt.

   * @param id - Row primary key.

   * @param attemptCount - The attempt this decision was made for; must match the

   *   row's current `attempt_count` for the write to apply.

   * @param error - Human-readable error string, written to `last_error`.

   * @param errorCode - Machine-readable classification, written to `error_code`.

   * @param nextAttemptAt - Earliest timestamp (ms) at which the row may be claimed again.

   * @returns True if the row was rescheduled; false if this attempt was superseded.

   */

  scheduleRetry(

    id: number,

    attemptCount: number,

    error: string,

    errorCode: QueueErrorCode,

    nextAttemptAt: number

  ): boolean {

    }

  }

  /**

   * Returns a `claimed` row to `queued` — used when the WS connection drops

   * before the transmit request was ever written to the socket, so retrying on

   * reconnect carries no duplicate-delivery risk. (A row that already reached the

   * socket is `inflight`, not `claimed`, and this is a no-op for it — its outcome

   * is ambiguous and owned by the response timeout.) Because the row keeps its

   * original id and claims are ordered by id, it goes back to the front of its

   * channel's FIFO.

   * @param id - Row primary key.

   * @returns True if the row was requeued; false if it was not in `claimed`.

   */

  requeue(id: number): boolean {

    }

  }

  /**

   * Recovers rows left mid-flight by a previous process, splitting on whether the

   * request reached the wire and the channel's guaranteed-delivery setting. Runs

   * once at startup, after acquiring the dispatch lease (§10):

   * - `claimed` rows provably never left the process (`sent_at` is NULL) → always

   *   returned to `queued` for a clean re-dispatch with no duplicate-delivery risk.

   * - `inflight` rows on a NORMAL channel are ambiguous (the server may have

   *   processed them) → `failed` with error code `interrupted`, surfaced for

   *   operator review, never silently retried.

   * - `inflight` rows on a GUARANTEED-delivery channel → returned to `queued`

   *   (duplication risk accepted) so the channel keeps trying until upstream gives

   *   a definitive answer (§4.1).

   * @param now - Override for `errored_at` (for tests).

   * @returns Counts of rows promoted to `failed` and returned to `queued`, respectively.

   */

    // Both legs return a row to `queued`: provably-unsent `claimed` rows (always

    // safe) and guaranteed-delivery `inflight` rows (duplication risk accepted).

    const requeued =

    }

  }

  /**

   * Lists all `queued` row IDs for a channel, in FIFO order. Used by the

   * worker on startup to repopulate its in-memory wake signal.

   * @param channelName - The channel to query.

   * @returns Row IDs in FIFO order.

   */

  listQueuedIdsForChannel(channelName: string): number[] {

  }

  /** @returns Counts of rows by state. Missing states are reported as 0. */

  countByState(): StateCounts {

      queued: 0,

      claimed: 0,

      inflight: 0,

      processed: 0,

      rejected: 0,

      failed: 0,

      nacked: 0,

    };

    }

  }

  /**

   * @param channelName - The channel to query.

   * @param now - Override for the "now" timestamp used in `oldestQueuedAgeMs`.

   * @returns Depth snapshot for `channelName` (queued/claimed/inflight counts + oldest queued age).

   */

      | { queued: number | null; claimed: number | null; inflight: number | null; oldest_queued_at: number | null }

      | undefined;

    };

  }

  /**

   * Attempts to acquire (or re-acquire) the queue lease as `holder`.

   *

   * Succeeds when there is no current lease, the current lease is already held

   * by us (refresh case), or the prior holder's lease has expired. Fails when a

   * different holder still has a valid lease — caller should wait and retry.

   *

   * The lease is the cross-process coordination primitive that makes zero-downtime

   * upgrades safe: only the holder runs workers and `recoverOnStartup`, so two

   * processes sharing the DB during the upgrade overlap don't fight over rows.

   * @param holder - Stable identifier for this process (a per-process UUID).

   * @param ttlMs - How long the new lease should remain valid before another

   *                process can take over (also drives the heartbeat cadence).

   * @param now - Override the "now" timestamp (for tests).

   * @returns True if the lease is now held by `holder`; false if a foreign holder still owns it.

   */

    }

  }

  /**

   * Extends the existing lease held by `holder`. Fails (returns false) when the

   * lease is no longer ours — most commonly because a peer took over after our

   * TTL elapsed. A leader that sees `false` here must stop driving the queue.

   * @param holder - The same identifier passed to {@link DurableQueue.tryAcquireLease}.

   * @param ttlMs - New TTL for the lease (added to `now`).

   * @param now - Override the "now" timestamp (for tests).

   * @returns True if the heartbeat extended our lease; false if we lost it.

   */

    }

  }

  /**

   * Releases the lease if (and only if) `holder` still owns it. Idempotent. The

   * holder check means a process that already lost its lease won't accidentally

   * delete a newer leader's row.

   * @param holder - The identifier of the lease to release.

   */

  releaseLease(holder: string): void {

    }

  }

  /**

   * @returns The current lease, or null if no row exists. Used for diagnostics

   *          and stats; the acquire/heartbeat methods do the actual coordination.

   */

  getCurrentLease(): { holder: string; expiresAt: number } | null {