import * as plugins from '../plugins.js';
import { logger } from '../logger.js';
import {
  DcRouterDb,
  WebPushAdmissionDoc,
  WebPushBindingDoc,
  WebPushSpoolDoc,
  type IWebPushCredentialRotationRecovery,
  type IWebPushStoredControllerFence,
  type IWebPushStoredCredential,
  type IWebPushStoredVapidKey,
  type TWebPushControllerIntent,
} from '../db/index.js';
import {
  WebPushCrypto,
  canonicalJson,
  parseWebPushKeyRing,
  type IWebPushAadContext,
} from './classes.webpush-crypto.js';
import {
  WebPushHttpTransport,
  WebPushTransportError,
  type IWebPushTransport,
} from './classes.webpush-transport.js';
import { WebPushEndpointPolicyError } from './helpers.webpush-endpoint.js';
import {
  getWebPushIndexSpecifications,
  type IWebPushIndexSpecification,
  type TWebPushIndexCollection,
} from './webpush-indexes.js';

type IWebPushBinding = plugins.servezoneInterfaces.data.IWebPushBinding;
type IWebPushCredentialOneTimeSecret =
  plugins.servezoneInterfaces.data.IWebPushCredentialOneTimeSecret;
type IWebPushCredentialPublic =
  plugins.servezoneInterfaces.data.IWebPushCredentialPublic;
type IWebPushDeliveryStatus = plugins.servezoneInterfaces.data.IWebPushDeliveryStatus;
type IWebPushNotificationPayload =
  plugins.servezoneInterfaces.data.IWebPushNotificationPayload;
type IWebPushResourceOwner = plugins.servezoneInterfaces.data.IWebPushResourceOwner;
type IWebPushServiceStatus = plugins.servezoneInterfaces.data.IWebPushServiceStatus;
type IWebPushSubscription = plugins.servezoneInterfaces.data.IWebPushSubscription;
type TWebPushBindingSync =
  plugins.servezoneInterfaces.requests.webpush.TWebPushBindingSync;
type IWebPushAppCredentialAuth =
  plugins.servezoneInterfaces.requests.webpush.IWebPushAppCredentialAuth;
type TWebPushCancellationTarget =
  plugins.servezoneInterfaces.data.TWebPushCancellationTarget;
type TWebPushDeliveryState = plugins.servezoneInterfaces.data.TWebPushDeliveryState;
type TWebPushUrgency = plugins.servezoneInterfaces.data.TWebPushUrgency;

interface IWebPushQenv {
  getEnvVarOnDemand(nameArg: string): Promise<string | undefined>;
}

export interface IWebPushManagerOptions {
  qenv?: IWebPushQenv;
  transport?: IWebPushTransport;
  startWorker?: boolean;
  now?: () => number;
  admissionPolicy?: Partial<IWebPushAdmissionPolicy>;
}

export interface IWebPushAdmissionPolicy {
  shortWindowMs: number;
  shortLimit: number;
  longWindowMs: number;
  longLimit: number;
}

export interface IWebPushAuthenticatedBinding {
  binding: WebPushBindingDoc;
  credential: IWebPushStoredCredential;
}

export interface IWebPushBindingSyncResult {
  binding: IWebPushBinding;
  credential?: IWebPushCredentialOneTimeSecret;
}

/**
 * Durable controller ordering token. The controller id and epoch remain stable
 * for an owner, while generation increases for every desired-state mutation.
 */
export interface IWebPushControllerMutation {
  id: string;
  epoch: string;
  generation: number;
  operationId: string;
  intent: TWebPushControllerIntent;
}

export interface IWebPushBindingDelete {
  id?: string;
  owner: IWebPushResourceOwner;
  controller: IWebPushControllerMutation;
}

export interface IWebPushEnqueueResult {
  accepted: boolean;
  spoolItemId: string;
}

export interface IWebPushCancelResult {
  cancelledCount: number;
  alreadyTerminalCount: number;
}

const MAX_PAYLOAD_BYTES = 3500;
const MAX_TTL_SECONDS = 86_400;
const DEFAULT_TTL_SECONDS = 86_400;
const MAX_ID_LENGTH = 256;
const MAX_IDEMPOTENCY_KEY_LENGTH = 256;
const MAX_COLLAPSE_KEY_LENGTH = 256;
const MAX_ROUTE_LENGTH = 2048;
const MAX_EVENT_ID_LENGTH = 256;
const MAX_NONTERMINAL_ITEMS_PER_BINDING = 10_000;
const DEFAULT_ADMISSION_POLICY: IWebPushAdmissionPolicy = {
  shortWindowMs: 60_000,
  shortLimit: 120,
  longWindowMs: 60 * 60_000,
  longLimit: 5_000,
};
const MAX_ADMISSION_CAS_ATTEMPTS = 512;
const DELIVERY_BATCH_SIZE = 25;
const DELIVERY_LEASE_MS = 60_000;
const DELIVERY_WORKER_INTERVAL_MS = 1_000;
const DELIVERY_MAX_ATTEMPTS = 8;
const TERMINAL_RETENTION_MS = 7 * 24 * 60 * 60 * 1000;
const CREDENTIAL_OVERLAP_MS = 15 * 60 * 1000;
const VAPID_RETIREMENT_MS = 90 * 24 * 60 * 60 * 1000;
const VAPID_KEY_RING_LIMIT = 4;
const LIFECYCLE_DRAIN_TIMEOUT_MS = DELIVERY_LEASE_MS + 5_000;
const LIFECYCLE_DRAIN_POLL_MS = 250;
const MAINTENANCE_BATCH_SIZE = 100;
const OPAQUE_ID_PATTERN = /^[A-Za-z0-9._:-]+$/;
const BASE64URL_PATTERN = /^[A-Za-z0-9_-]+$/;
const TERMINAL_STATES: TWebPushDeliveryState[] = [
  'pushServiceAccepted',
  'invalidSubscription',
  'failed',
  'expired',
  'cancelled',
];
const NONTERMINAL_STATES: TWebPushDeliveryState[] = [
  'accepted',
  'queued',
  'sending',
  'deferred',
];

function requireNonEmptyString(
  valueArg: unknown,
  labelArg: string,
  maxLengthArg = MAX_ID_LENGTH,
): string {
  if (
    typeof valueArg !== 'string'
    || valueArg.trim().length === 0
    || valueArg.length > maxLengthArg
  ) {
    throw new plugins.typedrequest.TypedResponseError(`${labelArg} is missing or malformed`);
  }
  return valueArg.trim();
}

function requireOpaqueId(valueArg: unknown, labelArg: string): string {
  const value = requireNonEmptyString(valueArg, labelArg);
  if (!OPAQUE_ID_PATTERN.test(value) || value.includes('://')) {
    throw new plugins.typedrequest.TypedResponseError(`${labelArg} is malformed`);
  }
  return value;
}

function requireBase64Url(
  valueArg: unknown,
  labelArg: string,
  expectedLengthArg: number,
): string {
  const value = requireNonEmptyString(valueArg, labelArg, 512);
  if (!BASE64URL_PATTERN.test(value)) {
    throw new plugins.typedrequest.TypedResponseError(`${labelArg} is malformed`);
  }
  const decoded = Buffer.from(value, 'base64url');
  if (
    decoded.length !== expectedLengthArg
    || decoded.toString('base64url') !== value
  ) {
    throw new plugins.typedrequest.TypedResponseError(`${labelArg} is malformed`);
  }
  return value;
}

function normalizeOwner(ownerArg: IWebPushResourceOwner): IWebPushResourceOwner {
  if (!['onebox', 'cloudly', 'custom'].includes(ownerArg?.gatewayClientType)) {
    throw new plugins.typedrequest.TypedResponseError('Web Push owner gatewayClientType is malformed');
  }
  return {
    gatewayClientType: ownerArg.gatewayClientType,
    gatewayClientId: requireOpaqueId(ownerArg.gatewayClientId, 'Web Push owner gatewayClientId'),
    appInstanceId: requireOpaqueId(ownerArg.appInstanceId, 'Web Push owner appInstanceId'),
  };
}

function normalizeSubscription(subscriptionArg: IWebPushSubscription): IWebPushSubscription {
  if (!subscriptionArg || typeof subscriptionArg !== 'object') {
    throw new plugins.typedrequest.TypedResponseError('Web Push subscription is malformed');
  }
  const endpoint = requireNonEmptyString(subscriptionArg.endpoint, 'Web Push subscription endpoint', 4096);
  const expirationTime = subscriptionArg.expirationTime;
  if (
    expirationTime !== null
    && (!Number.isSafeInteger(expirationTime) || expirationTime < 0)
  ) {
    throw new plugins.typedrequest.TypedResponseError('Web Push subscription expirationTime is malformed');
  }
  return {
    endpoint,
    expirationTime,
    keys: {
      p256dh: requireBase64Url(
        subscriptionArg.keys?.p256dh,
        'Web Push subscription p256dh key',
        65,
      ),
      auth: requireBase64Url(
        subscriptionArg.keys?.auth,
        'Web Push subscription auth key',
        16,
      ),
    },
  };
}

function normalizePayload(payloadArg: IWebPushNotificationPayload): IWebPushNotificationPayload {
  if (!payloadArg || typeof payloadArg !== 'object' || Array.isArray(payloadArg)) {
    throw new plugins.typedrequest.TypedResponseError('Web Push payload is malformed');
  }
  const keys = Object.keys(payloadArg).sort();
  if (keys.join(',') !== 'event,eventId,route,schemaVersion') {
    throw new plugins.typedrequest.TypedResponseError('Web Push payload contains unsupported fields');
  }
  if (payloadArg.schemaVersion !== 1 || payloadArg.event !== 'notificationAvailable') {
    throw new plugins.typedrequest.TypedResponseError('Web Push payload schema is unsupported');
  }
  const eventId = requireOpaqueId(payloadArg.eventId, 'Web Push payload eventId');
  if (eventId.length > MAX_EVENT_ID_LENGTH) {
    throw new plugins.typedrequest.TypedResponseError('Web Push payload eventId is too long');
  }
  const route = requireNonEmptyString(payloadArg.route, 'Web Push payload route', MAX_ROUTE_LENGTH);
  if (!route.startsWith('/') || route.startsWith('//') || route.includes('\\')) {
    throw new plugins.typedrequest.TypedResponseError('Web Push payload route must be same-origin');
  }
  let parsedRoute: URL;
  try {
    parsedRoute = new URL(route, 'https://webpush.invalid');
  } catch {
    throw new plugins.typedrequest.TypedResponseError('Web Push payload route is malformed');
  }
  if (parsedRoute.origin !== 'https://webpush.invalid') {
    throw new plugins.typedrequest.TypedResponseError('Web Push payload route must be same-origin');
  }
  const normalized = {
    schemaVersion: 1 as const,
    event: 'notificationAvailable' as const,
    eventId,
    route,
  };
  if (Buffer.byteLength(canonicalJson(normalized), 'utf8') > MAX_PAYLOAD_BYTES) {
    throw new plugins.typedrequest.TypedResponseError('Web Push payload exceeds the maximum size');
  }
  return normalized;
}

function normalizeTtl(ttlArg: number | undefined): number {
  const ttl = ttlArg ?? DEFAULT_TTL_SECONDS;
  if (!Number.isSafeInteger(ttl) || ttl <= 0 || ttl > MAX_TTL_SECONDS) {
    throw new plugins.typedrequest.TypedResponseError('Web Push ttlSeconds is malformed');
  }
  return ttl;
}

function normalizeUrgency(urgencyArg: TWebPushUrgency | undefined): TWebPushUrgency {
  const urgency = urgencyArg ?? 'normal';
  if (!['very-low', 'low', 'normal', 'high'].includes(urgency)) {
    throw new plugins.typedrequest.TypedResponseError('Web Push urgency is malformed');
  }
  return urgency;
}

function normalizeFeatureFlag(valueArg: string | undefined): boolean {
  if (valueArg === undefined || !valueArg.trim()) return false;
  const value = valueArg.trim().toLowerCase();
  if (['false', '0', 'off', 'no'].includes(value)) return false;
  if (['true', '1', 'on', 'yes'].includes(value)) return true;
  throw new Error('DCROUTER_WEB_PUSH_ENABLED must be an explicit boolean');
}

function normalizeAdmissionPolicy(
  policyArg: Partial<IWebPushAdmissionPolicy> | undefined,
): IWebPushAdmissionPolicy {
  const policy = {
    ...DEFAULT_ADMISSION_POLICY,
    ...(policyArg || {}),
  };
  for (const [name, value] of Object.entries(policy)) {
    if (!Number.isSafeInteger(value) || value <= 0) {
      throw new Error(`Web Push admission policy ${name} must be a positive integer`);
    }
  }
  if (
    policy.shortWindowMs > policy.longWindowMs
    || policy.shortLimit > policy.longLimit
  ) {
    throw new Error('Web Push admission policy short window must fit inside the long window');
  }
  return policy;
}

function requireLifecycleGeneration(valueArg: unknown): number {
  if (!Number.isSafeInteger(valueArg) || (valueArg as number) < 1) {
    throw new Error('Web Push binding lifecycle generation is malformed');
  }
  return valueArg as number;
}

function normalizeControllerMutation(
  controllerArg: IWebPushControllerMutation,
  expectedIntentArg: TWebPushControllerIntent,
): IWebPushControllerMutation {
  if (!controllerArg || typeof controllerArg !== 'object') {
    throw new plugins.typedrequest.TypedResponseError(
      'Web Push controller mutation is missing or malformed',
    );
  }
  if (
    !Number.isSafeInteger(controllerArg.generation)
    || controllerArg.generation < 1
  ) {
    throw new plugins.typedrequest.TypedResponseError(
      'Web Push controller generation is malformed',
    );
  }
  if (controllerArg.intent !== expectedIntentArg) {
    throw new plugins.typedrequest.TypedResponseError(
      'Web Push controller intent does not match the requested mutation',
    );
  }
  return {
    id: requireOpaqueId(controllerArg.id, 'Web Push controller id'),
    epoch: requireOpaqueId(controllerArg.epoch, 'Web Push controller epoch'),
    generation: controllerArg.generation,
    operationId: requireOpaqueId(
      controllerArg.operationId,
      'Web Push controller operation id',
    ),
    intent: expectedIntentArg,
  };
}

function normalizeDeleteControllerMutation(
  controllerArg: IWebPushControllerMutation,
): IWebPushControllerMutation {
  if (!controllerArg || !['disabled', 'deleted'].includes(controllerArg.intent)) {
    throw new plugins.typedrequest.TypedResponseError(
      'Web Push delete controller intent must be disabled or deleted',
    );
  }
  return normalizeControllerMutation(controllerArg, controllerArg.intent);
}

function controllerRequestDigest(valueArg: unknown): string {
  return plugins.crypto
    .createHash('sha256')
    .update('dcrouter:webpush:controller:v1\0', 'utf8')
    .update(canonicalJson(valueArg), 'utf8')
    .digest('base64url');
}

function assertStoredControllerFence(
  fenceArg: IWebPushStoredControllerFence,
): void {
  const hasValidOpaqueId = (valueArg: unknown): valueArg is string => (
    typeof valueArg === 'string'
    && valueArg.length > 0
    && valueArg.length <= MAX_ID_LENGTH
    && valueArg === valueArg.trim()
    && OPAQUE_ID_PATTERN.test(valueArg)
    && !valueArg.includes('://')
  );
  const digest = typeof fenceArg?.requestDigest === 'string'
    ? fenceArg.requestDigest
    : '';
  const decodedDigest = BASE64URL_PATTERN.test(digest)
    ? Buffer.from(digest, 'base64url')
    : Buffer.alloc(0);
  if (
    !fenceArg
    || typeof fenceArg !== 'object'
    || !hasValidOpaqueId(fenceArg.id)
    || !hasValidOpaqueId(fenceArg.epoch)
    || !Number.isSafeInteger(fenceArg.generation)
    || fenceArg.generation < 1
    || !hasValidOpaqueId(fenceArg.operationId)
    || !['enabled', 'disabled', 'deleted'].includes(fenceArg.intent)
    || decodedDigest.length !== 32
    || decodedDigest.toString('base64url') !== digest
    || !Number.isSafeInteger(fenceArg.appliedAt)
    || fenceArg.appliedAt < 0
  ) {
    throw new Error('Stored Web Push controller fence is malformed');
  }
}

function controllerFenceDecision(
  fenceArg: IWebPushStoredControllerFence | undefined,
  controllerArg: IWebPushControllerMutation,
  requestDigestArg: string,
): 'apply' | 'replay' {
  if (fenceArg === undefined) return 'apply';
  assertStoredControllerFence(fenceArg);
  if (fenceArg.id !== controllerArg.id || fenceArg.epoch !== controllerArg.epoch) {
    throw new plugins.typedrequest.TypedResponseError(
      'Web Push binding is owned by a different controller epoch',
    );
  }
  if (controllerArg.generation < fenceArg.generation) {
    throw new plugins.typedrequest.TypedResponseError(
      'Web Push controller generation is stale',
    );
  }
  if (controllerArg.generation > fenceArg.generation) return 'apply';
  if (
    fenceArg.operationId !== controllerArg.operationId
    || fenceArg.intent !== controllerArg.intent
    || fenceArg.requestDigest !== requestDigestArg
  ) {
    throw new plugins.typedrequest.TypedResponseError(
      'Web Push controller generation conflicts with an applied mutation',
    );
  }
  return 'replay';
}

function createControllerFence(
  controllerArg: IWebPushControllerMutation,
  requestDigestArg: string,
  nowArg: number,
): IWebPushStoredControllerFence {
  return {
    ...controllerArg,
    requestDigest: requestDigestArg,
    appliedAt: nowArg,
  };
}

function validateVapidSubject(subjectArg: string): string {
  const subject = requireNonEmptyString(subjectArg, 'Web Push VAPID subject', 2048);
  let parsed: URL;
  try {
    parsed = new URL(subject);
  } catch {
    throw new Error('DCROUTER_WEB_PUSH_VAPID_SUBJECT must be an https: or mailto: URI');
  }
  if (!['https:', 'mailto:'].includes(parsed.protocol)) {
    throw new Error('DCROUTER_WEB_PUSH_VAPID_SUBJECT must be an https: or mailto: URI');
  }
  if (parsed.protocol === 'https:' && parsed.hostname.toLowerCase() === 'localhost') {
    throw new Error('DCROUTER_WEB_PUSH_VAPID_SUBJECT must not use localhost');
  }
  return subject;
}

function constantTimeDigestEqual(leftArg: string, rightArg: string): boolean {
  if (!BASE64URL_PATTERN.test(leftArg) || !BASE64URL_PATTERN.test(rightArg)) return false;
  const left = Buffer.from(leftArg, 'base64url');
  const right = Buffer.from(rightArg, 'base64url');
  return left.length === right.length && plugins.crypto.timingSafeEqual(left, right);
}

function isDuplicateKeyError(errorArg: unknown): boolean {
  return (errorArg as { code?: unknown })?.code === 11000;
}

function asBindingDoc(rawArg: unknown): WebPushBindingDoc {
  return rawArg as WebPushBindingDoc;
}

function asAdmissionDoc(rawArg: unknown): WebPushAdmissionDoc {
  return rawArg as WebPushAdmissionDoc;
}

function asSpoolDoc(rawArg: unknown): WebPushSpoolDoc {
  return rawArg as WebPushSpoolDoc;
}

function toCredentialPublic(
  credentialArg: IWebPushStoredCredential,
): IWebPushCredentialPublic {
  return {
    id: credentialArg.id,
    status: credentialArg.status,
    createdAt: credentialArg.createdAt,
    updatedAt: credentialArg.updatedAt,
    ...(credentialArg.lastRotatedAt !== null
      ? { lastRotatedAt: credentialArg.lastRotatedAt }
      : {}),
  };
}

function toVapidKeyPublic(
  keyArg: IWebPushStoredVapidKey,
): plugins.servezoneInterfaces.data.IWebPushVapidKeyPublic {
  return {
    id: keyArg.id,
    publicKey: keyArg.publicKey,
    status: keyArg.status,
    createdAt: keyArg.createdAt,
    ...(keyArg.activatedAt !== null ? { activatedAt: keyArg.activatedAt } : {}),
    ...(keyArg.retireAfter !== null ? { retireAfter: keyArg.retireAfter } : {}),
    ...(keyArg.retiredAt !== null ? { retiredAt: keyArg.retiredAt } : {}),
  };
}

export function toPublicWebPushBinding(bindingArg: WebPushBindingDoc): IWebPushBinding {
  return {
    id: bindingArg.id,
    owner: {
      gatewayClientType: bindingArg.ownerGatewayClientType,
      gatewayClientId: bindingArg.ownerGatewayClientId,
      appInstanceId: bindingArg.ownerAppInstanceId,
    },
    enabled: bindingArg.enabled,
    status: bindingArg.status,
    credential: toCredentialPublic(bindingArg.credential),
    vapidKeys: (bindingArg.vapidKeys || []).map((keyArg) => toVapidKeyPublic(keyArg)),
    createdAt: bindingArg.createdAt,
    updatedAt: bindingArg.updatedAt,
    createdBy: bindingArg.createdBy,
  };
}

export function toPublicWebPushDelivery(spoolArg: WebPushSpoolDoc): IWebPushDeliveryStatus {
  return {
    spoolItemId: spoolArg.id,
    state: spoolArg.state,
    attempts: spoolArg.attempts,
    acceptedAt: spoolArg.acceptedAt,
    updatedAt: spoolArg.updatedAt,
    ...(!TERMINAL_STATES.includes(spoolArg.state)
      ? { nextAttemptAt: spoolArg.nextAttemptAt }
      : {}),
    ...(spoolArg.terminalAt !== undefined ? { terminalAt: spoolArg.terminalAt } : {}),
    ...(spoolArg.pushServiceStatusCode !== null
      ? { pushServiceStatusCode: spoolArg.pushServiceStatusCode }
      : {}),
    ...(spoolArg.errorCode ? { errorCode: spoolArg.errorCode } : {}),
  };
}

export function buildWebPushTerminalUpdate(optionsArg: {
  state: Extract<
    TWebPushDeliveryState,
    'pushServiceAccepted' | 'invalidSubscription' | 'failed' | 'expired' | 'cancelled'
  >;
  now: number;
  pushServiceStatusCode?: number;
  errorCode?: string;
}): Record<string, unknown> {
  return {
    $set: {
      state: optionsArg.state,
      updatedAt: optionsArg.now,
      terminalAt: optionsArg.now,
      purgeAt: new Date(optionsArg.now + TERMINAL_RETENTION_MS),
      pushServiceStatusCode: optionsArg.pushServiceStatusCode ?? null,
      errorCode: optionsArg.errorCode ?? null,
      _updatedAt: new Date(optionsArg.now).toISOString(),
    },
    $unset: {
      subscriptionEnvelope: '',
      payloadEnvelope: '',
      leaseToken: '',
      leaseExpiresAt: '',
      cancelRequestedAt: '',
      cancelErrorCode: '',
    },
  };
}

interface IWebPushNativeIndexDescription {
  name?: string;
  key?: Record<string, unknown>;
  unique?: boolean;
  expireAfterSeconds?: number;
  partialFilterExpression?: Record<string, unknown>;
  sparse?: boolean;
  collation?: Record<string, unknown>;
  hidden?: boolean;
}

function canonicalIndexValue(valueArg: unknown): string {
  if (valueArg === undefined) return 'undefined';
  return canonicalJson(valueArg);
}

function assertWebPushIndexMatches(
  collectionArg: TWebPushIndexCollection,
  actualArg: IWebPushNativeIndexDescription | undefined,
  expectedArg: IWebPushIndexSpecification,
): void {
  const malformed = (): never => {
    throw new Error(
      `Web Push persistence index ${collectionArg}.${expectedArg.name} is missing or malformed`,
    );
  };
  if (!actualArg) {
    throw new Error(
      `Web Push persistence index ${collectionArg}.${expectedArg.name} is missing or malformed`,
    );
  }
  if (actualArg.name !== expectedArg.name) malformed();
  if (JSON.stringify(actualArg.key) !== JSON.stringify(expectedArg.key)) malformed();
  if (Boolean(actualArg.unique) !== Boolean(expectedArg.unique)) malformed();
  if (actualArg.expireAfterSeconds !== expectedArg.expireAfterSeconds) malformed();
  if (
    canonicalIndexValue(actualArg.partialFilterExpression)
    !== canonicalIndexValue(expectedArg.partialFilterExpression)
  ) {
    malformed();
  }
  if (
    actualArg.sparse === true
    || actualArg.collation !== undefined
    || actualArg.hidden === true
  ) {
    malformed();
  }
}

function assertWebPushIndexes(
  collectionArg: TWebPushIndexCollection,
  actualIndexesArg: IWebPushNativeIndexDescription[],
): void {
  const byName = new Map(
    actualIndexesArg.map((indexArg) => [indexArg.name, indexArg]),
  );
  for (const expected of getWebPushIndexSpecifications(collectionArg)) {
    assertWebPushIndexMatches(collectionArg, byName.get(expected.name), expected);
  }
}

export class WebPushManager {
  private readonly qenv: IWebPushQenv;
  private readonly transport: IWebPushTransport;
  private readonly startWorker: boolean;
  private readonly now: () => number;
  private readonly admissionPolicy: IWebPushAdmissionPolicy;
  private enabled = false;
  private started = false;
  private ready = false;
  private crypto?: WebPushCrypto;
  private vapidSubject?: string;
  private workerTimer?: NodeJS.Timeout;
  private activeWorkerCycle?: Promise<void>;
  private expiredCredentialOverlapCursor?: string;
  private expiredCredentialRecoveryCursor?: string;
  private retiredVapidCursor?: string;
  private inactiveLifecycleCursor?: string;

  public constructor(optionsArg: IWebPushManagerOptions = {}) {
    this.qenv = optionsArg.qenv ?? new plugins.qenv.Qenv('./', '.nogit/', false);
    this.transport = optionsArg.transport ?? new WebPushHttpTransport();
    this.startWorker = optionsArg.startWorker !== false;
    this.now = optionsArg.now ?? (() => Date.now());
    this.admissionPolicy = normalizeAdmissionPolicy(optionsArg.admissionPolicy);
  }

  public get isEnabled(): boolean {
    return this.enabled;
  }

  public get isReady(): boolean {
    return this.ready;
  }

  public async start(): Promise<void> {
    if (this.started) return;
    const enabledValue = await this.qenv.getEnvVarOnDemand('DCROUTER_WEB_PUSH_ENABLED');
    this.enabled = normalizeFeatureFlag(enabledValue);
    if (!this.enabled) {
      this.started = true;
      this.ready = false;
      logger.log('info', 'Web Push provider is disabled');
      return;
    }
    if (!DcRouterDb.getInstance().isReady()) {
      throw new Error('Web Push persistence is unavailable');
    }
    const encryptionKeyRingRaw = await this.qenv.getEnvVarOnDemand(
      'DCROUTER_WEB_PUSH_MASTER_KEY_RING',
    );
    const hmacKeyRingRaw = await this.qenv.getEnvVarOnDemand(
      'DCROUTER_WEB_PUSH_HMAC_KEY_RING',
    );
    const vapidSubjectRaw = await this.qenv.getEnvVarOnDemand(
      'DCROUTER_WEB_PUSH_VAPID_SUBJECT',
    );
    if (!encryptionKeyRingRaw || !hmacKeyRingRaw || !vapidSubjectRaw) {
      throw new Error(
        'Enabled Web Push requires master key ring, HMAC key ring, and VAPID subject',
      );
    }
    this.crypto = new WebPushCrypto(
      parseWebPushKeyRing(encryptionKeyRingRaw, 'DCROUTER_WEB_PUSH_MASTER_KEY_RING'),
      parseWebPushKeyRing(hmacKeyRingRaw, 'DCROUTER_WEB_PUSH_HMAC_KEY_RING'),
    );
    this.vapidSubject = validateVapidSubject(vapidSubjectRaw);
    const bindingCollection = WebPushBindingDoc.getNativeCollection();
    const admissionCollection = WebPushAdmissionDoc.getNativeCollection();
    const spoolCollection = WebPushSpoolDoc.getNativeCollection();
    await bindingCollection.findOne({}, { projection: { _id: 1 } });
    await admissionCollection.findOne({}, { projection: { _id: 1 } });
    await spoolCollection.findOne({}, { projection: { _id: 1 } });
    assertWebPushIndexes(
      'WebPushBindingDoc',
      await bindingCollection.listIndexes().toArray(),
    );
    assertWebPushIndexes(
      'WebPushAdmissionDoc',
      await admissionCollection.listIndexes().toArray(),
    );
    assertWebPushIndexes(
      'WebPushSpoolDoc',
      await spoolCollection.listIndexes().toArray(),
    );
    await this.runMaintenanceOnce();
    this.started = true;
    this.ready = true;
    logger.log('info', 'Web Push provider is ready');
    if (this.startWorker) this.scheduleWorker(0);
  }

  public async stop(): Promise<void> {
    this.ready = false;
    if (this.workerTimer) {
      clearTimeout(this.workerTimer);
      this.workerTimer = undefined;
    }
    await this.activeWorkerCycle;
    this.started = false;
  }

  public getUnavailableServiceStatus(): IWebPushServiceStatus {
    return {
      ready: false,
      bindingId: 'unavailable',
      bindingStatus: 'disabled',
      retiringVapidKeys: [],
      maxPayloadBytes: MAX_PAYLOAD_BYTES,
      maxTtlSeconds: MAX_TTL_SECONDS,
      contentEncoding: 'aes128gcm',
      message: this.enabled
        ? 'Web Push provider is not ready'
        : 'Web Push provider is disabled',
    };
  }

  public async listBindings(
    ownerFilterArg: Partial<IWebPushResourceOwner> = {},
  ): Promise<IWebPushBinding[]> {
    this.requirePersistenceReady();
    const selector: Record<string, unknown> = {};
    if (ownerFilterArg.gatewayClientType !== undefined) {
      if (!['onebox', 'cloudly', 'custom'].includes(ownerFilterArg.gatewayClientType)) {
        throw new plugins.typedrequest.TypedResponseError('Web Push owner filter is malformed');
      }
      selector.ownerGatewayClientType = ownerFilterArg.gatewayClientType;
    }
    if (ownerFilterArg.gatewayClientId !== undefined) {
      selector.ownerGatewayClientId = requireOpaqueId(
        ownerFilterArg.gatewayClientId,
        'Web Push owner gatewayClientId',
      );
    }
    if (ownerFilterArg.appInstanceId !== undefined) {
      selector.ownerAppInstanceId = requireOpaqueId(
        ownerFilterArg.appInstanceId,
        'Web Push owner appInstanceId',
      );
    }
    const rows = await WebPushBindingDoc.getNativeCollection()
      .find(selector)
      .sort({ createdAt: 1, id: 1 })
      .limit(1000)
      .toArray();
    return rows.map((rowArg) => toPublicWebPushBinding(asBindingDoc(rowArg)));
  }

  public async getBindingById(idArg: string): Promise<WebPushBindingDoc | null> {
    this.requirePersistenceReady();
    const id = requireOpaqueId(idArg, 'Web Push binding id');
    const row = await WebPushBindingDoc.getNativeCollection().findOne({ id });
    return row ? asBindingDoc(row) : null;
  }

  public async getBindingByCredentialId(
    credentialIdArg: string,
  ): Promise<WebPushBindingDoc | null> {
    this.requirePersistenceReady();
    const credentialId = requireOpaqueId(credentialIdArg, 'Web Push credential id');
    const row = await WebPushBindingDoc.getNativeCollection().findOne({
      $or: [
        { 'credential.id': credentialId },
        { 'previousCredentials.id': credentialId },
      ],
    });
    return row ? asBindingDoc(row) : null;
  }

  public async syncBinding(
    bindingArg: TWebPushBindingSync,
    controllerArg: IWebPushControllerMutation,
    createdByArg: string,
  ): Promise<IWebPushBindingSyncResult> {
    this.requireReady();
    const owner = normalizeOwner(bindingArg.owner);
    const createdBy = requireNonEmptyString(createdByArg, 'Web Push binding creator');
    if (bindingArg.enabled !== true) {
      throw new plugins.typedrequest.TypedResponseError(
        'Web Push binding synchronization only accepts enabled bindings',
      );
    }
    const controller = normalizeControllerMutation(
      controllerArg,
      'enabled',
    );
    const requestedId = bindingArg.id !== undefined
      ? requireOpaqueId(bindingArg.id, 'Web Push binding id')
      : undefined;
    const requestDigest = controllerRequestDigest({
      action: 'sync',
      id: requestedId ?? null,
      owner,
      enabled: bindingArg.enabled,
    });
    const collection = WebPushBindingDoc.getNativeCollection();
    const byId = requestedId ? await collection.findOne({ id: requestedId }) : null;
    const byOwner = await collection.findOne({
      ownerGatewayClientType: owner.gatewayClientType,
      ownerGatewayClientId: owner.gatewayClientId,
      ownerAppInstanceId: owner.appInstanceId,
    });
    if (byId && byOwner && String(byId.id) !== String(byOwner.id)) {
      throw new plugins.typedrequest.TypedResponseError('Web Push binding identity conflicts with owner');
    }
    if (requestedId && byOwner && String(byOwner.id) !== requestedId) {
      throw new plugins.typedrequest.TypedResponseError('Web Push binding identity conflicts with owner');
    }
    const existingRaw = byId || byOwner;
    if (existingRaw) {
      const existing = asBindingDoc(existingRaw);
      this.assertBindingOwner(existing, owner);
      if (
        controllerFenceDecision(existing.controllerFence, controller, requestDigest)
        === 'replay'
      ) {
        return await this.resumeControllerSyncReplay(existing, controller, requestDigest);
      }
      const lifecycleGeneration = requireLifecycleGeneration(existing.lifecycleGeneration);
      if (typeof existing.deletedAt === 'number') {
        await this.scrubDeletedBindingLifecycle(existing);
        const scrubbedRaw = await collection.findOne({
          id: existing.id,
          lifecycleGeneration,
          deletedAt: existing.deletedAt,
        });
        if (!scrubbedRaw) {
          throw new plugins.typedrequest.TypedResponseError(
            'Web Push binding changed concurrently; retry synchronization',
          );
        }
        const scrubbed = asBindingDoc(scrubbedRaw);
        if (
          controllerFenceDecision(scrubbed.controllerFence, controller, requestDigest)
          === 'replay'
        ) {
          return await this.resumeControllerSyncReplay(
            scrubbed,
            controller,
            requestDigest,
          );
        }
        const now = this.now();
        const createdCredential = this.createCredential(scrubbed.id, now);
        const nextLifecycleGeneration = lifecycleGeneration + 1;
        const reactivated = await collection.updateOne(
          {
            id: scrubbed.id,
            mutationRevision: scrubbed.mutationRevision,
            deletedAt: scrubbed.deletedAt,
            lifecycleGeneration,
          },
          {
            $set: {
              enabled: true,
              status: 'active',
              credential: createdCredential.stored,
              previousCredentials: [],
              vapidKeys: [this.createVapidKey(owner, scrubbed.id, now)],
              updatedAt: now,
              lifecycleGeneration: nextLifecycleGeneration,
              controllerFence: createControllerFence(controller, requestDigest, now),
              mutationRevision: scrubbed.mutationRevision + 1,
              _updatedAt: new Date(now).toISOString(),
            },
            $unset: {
              deletedAt: '',
              credentialRotationRecovery: '',
            },
          },
        );
        if (reactivated.matchedCount !== 1) {
          throw new plugins.typedrequest.TypedResponseError(
            'Web Push binding changed concurrently; retry synchronization',
          );
        }
        await WebPushAdmissionDoc.getNativeCollection().deleteMany({
          bindingId: scrubbed.id,
          lifecycleGeneration: { $lt: nextLifecycleGeneration },
        });
        const refreshed = await collection.findOne({
          id: scrubbed.id,
          lifecycleGeneration: nextLifecycleGeneration,
          deletedAt: { $exists: false },
        });
        if (!refreshed) throw new Error('Reactivated Web Push binding disappeared');
        return {
          binding: toPublicWebPushBinding(asBindingDoc(refreshed)),
          credential: createdCredential.oneTime,
        };
      }
      if (bindingArg.enabled && !existing.enabled) {
        await this.drainBindingLifecyclesBefore(
          existing.id,
          lifecycleGeneration,
          'BINDING_DISABLED',
        );
        const stillDisabled = await collection.findOne({
          id: existing.id,
          mutationRevision: existing.mutationRevision,
          lifecycleGeneration,
          enabled: false,
          deletedAt: { $exists: false },
        }, { projection: { _id: 1 } });
        if (!stillDisabled) {
          throw new plugins.typedrequest.TypedResponseError(
            'Web Push binding changed concurrently; retry synchronization',
          );
        }
      }
      let credential: IWebPushCredentialOneTimeSecret | undefined;
      const now = this.now();
      const lifecycleChanged = existing.enabled !== bindingArg.enabled;
      const nextLifecycleGeneration = lifecycleChanged
        ? lifecycleGeneration + 1
        : lifecycleGeneration;
      const nextCredential = { ...existing.credential };
      if (bindingArg.enabled && nextCredential.status === 'revoked') {
        const created = this.createCredential(existing.id, now);
        Object.assign(nextCredential, created.stored);
        credential = created.oneTime;
      } else {
        nextCredential.status = bindingArg.enabled ? 'active' : 'disabled';
        nextCredential.updatedAt = now;
      }
      const update: {
        $set: Record<string, unknown>;
        $unset?: Record<string, ''>;
      } = {
        $set: {
          enabled: bindingArg.enabled,
          status: bindingArg.enabled ? 'active' : 'disabled',
          credential: nextCredential,
          updatedAt: now,
          lifecycleGeneration: nextLifecycleGeneration,
          controllerFence: createControllerFence(controller, requestDigest, now),
          mutationRevision: existing.mutationRevision + 1,
          _updatedAt: new Date(now).toISOString(),
          ...(lifecycleChanged ? { previousCredentials: [] } : {}),
        },
      };
      if (lifecycleChanged) {
        update.$unset = {
          credentialRotationRecovery: '',
        };
      }
      const updated = await collection.updateOne(
        {
          id: existing.id,
          mutationRevision: existing.mutationRevision,
          lifecycleGeneration,
          enabled: existing.enabled,
          deletedAt: { $exists: false },
        },
        update,
      );
      if (updated.matchedCount !== 1) {
        throw new plugins.typedrequest.TypedResponseError(
          'Web Push binding changed concurrently; retry synchronization',
        );
      }
      if (lifecycleChanged) {
        await WebPushAdmissionDoc.getNativeCollection().deleteMany({
          bindingId: existing.id,
          lifecycleGeneration: { $lt: nextLifecycleGeneration },
        });
      }
      if (!bindingArg.enabled || lifecycleChanged) {
        await this.drainBindingLifecyclesBefore(
          existing.id,
          nextLifecycleGeneration,
          bindingArg.enabled ? 'BINDING_LIFECYCLE_CHANGED' : 'BINDING_DISABLED',
        );
      }
      const refreshed = await collection.findOne({ id: existing.id });
      if (!refreshed) throw new Error('Synchronized Web Push binding disappeared');
      return {
        binding: toPublicWebPushBinding(asBindingDoc(refreshed)),
        credential,
      };
    }

    const now = this.now();
    const id = requestedId ?? plugins.uuid.v4();
    const createdCredential = this.createCredential(id, now);
    const vapidKey = this.createVapidKey(owner, id, now);
    const doc = new WebPushBindingDoc();
    doc.id = id;
    doc.ownerGatewayClientType = owner.gatewayClientType;
    doc.ownerGatewayClientId = owner.gatewayClientId;
    doc.ownerAppInstanceId = owner.appInstanceId;
    doc.enabled = bindingArg.enabled;
    doc.status = bindingArg.enabled ? 'active' : 'disabled';
    doc.credential = {
      ...createdCredential.stored,
      status: bindingArg.enabled ? 'active' : 'disabled',
    };
    doc.previousCredentials = [];
    doc.vapidKeys = [vapidKey];
    doc.createdAt = now;
    doc.updatedAt = now;
    doc.createdBy = createdBy;
    doc.mutationRevision = 1;
    doc.lifecycleGeneration = 1;
    doc.controllerFence = createControllerFence(controller, requestDigest, now);
    try {
      await doc.save();
    } catch (error: unknown) {
      if (!isDuplicateKeyError(error)) throw error;
      const raced = await collection.findOne({
        ownerGatewayClientType: owner.gatewayClientType,
        ownerGatewayClientId: owner.gatewayClientId,
        ownerAppInstanceId: owner.appInstanceId,
      });
      if (!raced) throw error;
      const racedBinding = asBindingDoc(raced);
      this.assertBindingOwner(racedBinding, owner);
      if (
        controllerFenceDecision(racedBinding.controllerFence, controller, requestDigest)
        === 'replay'
      ) {
        return await this.resumeControllerSyncReplay(racedBinding, controller, requestDigest);
      }
      throw new plugins.typedrequest.TypedResponseError(
        'Web Push binding changed concurrently; retry synchronization',
      );
    }
    return {
      binding: toPublicWebPushBinding(doc),
      credential: createdCredential.oneTime,
    };
  }

  private async resumeControllerSyncReplay(
    bindingArg: WebPushBindingDoc,
    controllerArg: IWebPushControllerMutation,
    requestDigestArg: string,
  ): Promise<IWebPushBindingSyncResult> {
    const lifecycleGeneration = requireLifecycleGeneration(bindingArg.lifecycleGeneration);
    if (controllerArg.intent === 'enabled') {
      if (
        !bindingArg.enabled
        || bindingArg.status !== 'active'
        || typeof bindingArg.deletedAt === 'number'
      ) {
        throw new Error('Applied Web Push enable controller mutation is inconsistent');
      }
      await this.drainBindingLifecyclesBefore(
        bindingArg.id,
        lifecycleGeneration,
        'BINDING_LIFECYCLE_CHANGED',
      );
    } else if (controllerArg.intent === 'disabled') {
      if (bindingArg.enabled) {
        throw new Error('Applied Web Push disable controller mutation is inconsistent');
      }
      if (typeof bindingArg.deletedAt === 'number') {
        await this.scrubDeletedBindingLifecycle(bindingArg);
      } else {
        await this.drainBindingLifecyclesBefore(
          bindingArg.id,
          lifecycleGeneration,
          'BINDING_DISABLED',
        );
      }
    } else {
      throw new Error('Delete controller mutation cannot be replayed as synchronization');
    }
    const refreshedRaw = await WebPushBindingDoc.getNativeCollection().findOne({
      id: bindingArg.id,
    });
    if (!refreshedRaw) throw new Error('Synchronized Web Push binding disappeared');
    const refreshed = asBindingDoc(refreshedRaw);
    controllerFenceDecision(refreshed.controllerFence, controllerArg, requestDigestArg);
    return {
      binding: toPublicWebPushBinding(refreshed),
    };
  }

  public async deleteBinding(
    deleteArg: IWebPushBindingDelete,
    deletedByArg: string,
  ): Promise<boolean> {
    this.requireReady();
    const owner = normalizeOwner(deleteArg?.owner);
    const deletedBy = requireNonEmptyString(deletedByArg, 'Web Push binding deletion actor');
    const requestedId = deleteArg.id !== undefined
      ? requireOpaqueId(deleteArg.id, 'Web Push binding id')
      : undefined;
    const controller = normalizeDeleteControllerMutation(deleteArg.controller);
    const requestDigest = controllerRequestDigest({
      action: 'delete',
      id: requestedId ?? null,
      owner,
    });
    const collection = WebPushBindingDoc.getNativeCollection();
    const byId = requestedId ? await collection.findOne({ id: requestedId }) : null;
    const byOwner = await collection.findOne({
      ownerGatewayClientType: owner.gatewayClientType,
      ownerGatewayClientId: owner.gatewayClientId,
      ownerAppInstanceId: owner.appInstanceId,
    });
    if (byId && byOwner && String(byId.id) !== String(byOwner.id)) {
      throw new plugins.typedrequest.TypedResponseError(
        'Web Push binding identity conflicts with owner',
      );
    }
    if (requestedId && byOwner && String(byOwner.id) !== requestedId) {
      throw new plugins.typedrequest.TypedResponseError(
        'Web Push binding identity conflicts with owner',
      );
    }
    const existingRaw = byId || byOwner;
    if (!existingRaw) {
      const now = this.now();
      const id = requestedId ?? plugins.uuid.v4();
      const tombstone = new WebPushBindingDoc();
      tombstone.id = id;
      tombstone.ownerGatewayClientType = owner.gatewayClientType;
      tombstone.ownerGatewayClientId = owner.gatewayClientId;
      tombstone.ownerAppInstanceId = owner.appInstanceId;
      tombstone.enabled = false;
      tombstone.status = 'disabled';
      tombstone.credential = {
        id: plugins.uuid.v4(),
        status: 'revoked',
        hmacKeyId: this.crypto!.currentHmacKeyId,
        createdAt: now,
        updatedAt: now,
        lastRotatedAt: null,
        validUntil: null,
      };
      tombstone.previousCredentials = [];
      tombstone.vapidKeys = [];
      tombstone.createdAt = now;
      tombstone.updatedAt = now;
      tombstone.createdBy = deletedBy;
      tombstone.mutationRevision = 1;
      tombstone.lifecycleGeneration = 1;
      tombstone.controllerFence = createControllerFence(controller, requestDigest, now);
      tombstone.deletedAt = now;
      try {
        await tombstone.save();
      } catch (error: unknown) {
        if (!isDuplicateKeyError(error)) throw error;
        return await this.deleteBinding(deleteArg, deletedBy);
      }
      await this.scrubDeletedBindingLifecycle(tombstone);
      return true;
    }
    const binding = asBindingDoc(existingRaw);
    this.assertBindingOwner(binding, owner);
    if (
      controllerFenceDecision(binding.controllerFence, controller, requestDigest)
      === 'replay'
    ) {
      return await this.resumeControllerDeleteReplay(binding, controller, requestDigest);
    }
    const lifecycleGeneration = requireLifecycleGeneration(binding.lifecycleGeneration);
    if (typeof binding.deletedAt === 'number') {
      await this.scrubDeletedBindingLifecycle(binding);
      const scrubbedRaw = await collection.findOne({
        id: binding.id,
        lifecycleGeneration,
        deletedAt: binding.deletedAt,
        enabled: false,
      });
      if (!scrubbedRaw) {
        throw new plugins.typedrequest.TypedResponseError(
          'Web Push binding changed concurrently; retry deletion',
        );
      }
      const scrubbed = asBindingDoc(scrubbedRaw);
      if (
        controllerFenceDecision(scrubbed.controllerFence, controller, requestDigest)
        === 'replay'
      ) {
        return await this.resumeControllerDeleteReplay(
          scrubbed,
          controller,
          requestDigest,
        );
      }
      const now = this.now();
      const recorded = await collection.updateOne(
        {
          id: scrubbed.id,
          mutationRevision: scrubbed.mutationRevision,
          lifecycleGeneration,
          deletedAt: scrubbed.deletedAt,
        },
        {
          $set: {
            controllerFence: createControllerFence(controller, requestDigest, now),
            updatedAt: now,
            mutationRevision: scrubbed.mutationRevision + 1,
            _updatedAt: new Date(now).toISOString(),
          },
        },
      );
      if (recorded.matchedCount !== 1) {
        throw new plugins.typedrequest.TypedResponseError(
          'Web Push binding changed concurrently; retry deletion',
        );
      }
      const refreshedRaw = await collection.findOne({ id: scrubbed.id });
      if (!refreshedRaw) throw new Error('Deleted Web Push binding tombstone disappeared');
      return await this.resumeControllerDeleteReplay(
        asBindingDoc(refreshedRaw),
        controller,
        requestDigest,
      );
    }
    const now = this.now();
    const revokedCredential: IWebPushStoredCredential = {
      ...binding.credential,
      status: 'revoked',
      updatedAt: now,
      validUntil: null,
    };
    delete revokedCredential.secretHash;
    const scrubbedVapidKeys = (binding.vapidKeys || []).map((keyArg) => {
      const scrubbed: IWebPushStoredVapidKey = {
        ...keyArg,
        status: 'retired',
        retireAfter: null,
        retiredAt: now,
      };
      delete scrubbed.privateKeyEnvelope;
      return scrubbed;
    });
    const deletedLifecycleGeneration = lifecycleGeneration + 1;
    const result = await WebPushBindingDoc.getNativeCollection().updateOne(
      {
        id: binding.id,
        mutationRevision: binding.mutationRevision,
        lifecycleGeneration,
        deletedAt: { $exists: false },
      },
      {
        $set: {
          enabled: false,
          status: 'disabled',
          credential: revokedCredential,
          previousCredentials: [],
          vapidKeys: scrubbedVapidKeys,
          deletedAt: now,
          lifecycleGeneration: deletedLifecycleGeneration,
          controllerFence: createControllerFence(controller, requestDigest, now),
          updatedAt: now,
          mutationRevision: binding.mutationRevision + 1,
          _updatedAt: new Date(now).toISOString(),
        },
        $unset: {
          credentialRotationRecovery: '',
        },
      },
    );
    if (result.matchedCount !== 1) {
      throw new plugins.typedrequest.TypedResponseError(
        'Web Push binding changed concurrently; retry deletion',
      );
    }
    const tombstone = await WebPushBindingDoc.getNativeCollection().findOne({
      id: binding.id,
      lifecycleGeneration: deletedLifecycleGeneration,
      deletedAt: now,
    });
    if (!tombstone) {
      throw new Error('Deleted Web Push binding tombstone disappeared');
    }
    await this.scrubDeletedBindingLifecycle(asBindingDoc(tombstone));
    return true;
  }

  private async resumeControllerDeleteReplay(
    bindingArg: WebPushBindingDoc,
    controllerArg: IWebPushControllerMutation,
    requestDigestArg: string,
  ): Promise<boolean> {
    if (bindingArg.enabled || typeof bindingArg.deletedAt !== 'number') {
      throw new Error('Applied Web Push delete controller mutation is inconsistent');
    }
    await this.scrubDeletedBindingLifecycle(bindingArg);
    const refreshedRaw = await WebPushBindingDoc.getNativeCollection().findOne({
      id: bindingArg.id,
    });
    if (!refreshedRaw) throw new Error('Deleted Web Push binding tombstone disappeared');
    const refreshed = asBindingDoc(refreshedRaw);
    controllerFenceDecision(refreshed.controllerFence, controllerArg, requestDigestArg);
    if (refreshed.enabled || typeof refreshed.deletedAt !== 'number') {
      throw new Error('Applied Web Push delete controller mutation is inconsistent');
    }
    return true;
  }

  public async rotateCredential(
    credentialIdArg: string,
  ): Promise<IWebPushCredentialOneTimeSecret> {
    this.requireReady();
    const requestedCredentialId = requireOpaqueId(
      credentialIdArg,
      'Web Push credential id',
    );
    for (let attempt = 0; attempt < 4; attempt++) {
      const binding = await this.getBindingByCredentialId(requestedCredentialId);
      if (
        !binding
        || !binding.enabled
        || binding.status !== 'active'
        || typeof binding.deletedAt === 'number'
      ) {
        throw new plugins.typedrequest.TypedResponseError(
          'Active Web Push credential not found',
        );
      }
      const now = this.now();
      const replay = this.replayCredentialRotation(
        binding,
        requestedCredentialId,
        now,
      );
      if (replay) return replay;
      if (binding.credential.id !== requestedCredentialId) {
        throw new plugins.typedrequest.TypedResponseError(
          'Active Web Push credential not found',
        );
      }
      const unexpiredPrevious = (binding.previousCredentials || []).filter(
        (credentialArg) => (
          credentialArg.secretHash
          && credentialArg.validUntil !== null
          && credentialArg.validUntil > now
        ),
      );
      if (unexpiredPrevious.length > 0) {
        throw new plugins.typedrequest.TypedResponseError(
          'Web Push credential rotation overlap is still active',
        );
      }
      const lifecycleGeneration = requireLifecycleGeneration(
        binding.lifecycleGeneration,
      );
      const created = this.createCredential(binding.id, now);
      const previous: IWebPushStoredCredential = {
        ...binding.credential,
        status: 'rotated',
        updatedAt: now,
        lastRotatedAt: now,
        validUntil: now + CREDENTIAL_OVERLAP_MS,
      };
      const recoveryIdentity = {
        requestedCredentialId,
        resultingCredentialId: created.stored.id,
        lifecycleGeneration,
      };
      const recovery: IWebPushCredentialRotationRecovery = {
        ...recoveryIdentity,
        secretEnvelope: this.crypto!.encryptJson(
          { secret: created.oneTime.secret },
          this.credentialRotationRecoveryAad(binding, recoveryIdentity),
        ),
        createdAt: now,
        expiresAt: now + CREDENTIAL_OVERLAP_MS,
      };
      const result = await WebPushBindingDoc.getNativeCollection().updateOne(
        {
          id: binding.id,
          mutationRevision: binding.mutationRevision,
          lifecycleGeneration,
          enabled: true,
          status: 'active',
          'credential.id': requestedCredentialId,
          deletedAt: { $exists: false },
        },
        {
          $set: {
            credential: created.stored,
            previousCredentials: [previous],
            credentialRotationRecovery: recovery,
            updatedAt: now,
            mutationRevision: binding.mutationRevision + 1,
            _updatedAt: new Date(now).toISOString(),
          },
        },
      );
      if (result.matchedCount === 1) return created.oneTime;
      const raced = await this.getBindingByCredentialId(requestedCredentialId);
      if (raced) {
        const racedReplay = this.replayCredentialRotation(
          raced,
          requestedCredentialId,
          this.now(),
        );
        if (racedReplay) return racedReplay;
      }
    }
    throw new plugins.typedrequest.TypedResponseError(
      'Web Push binding changed concurrently; retry credential rotation',
    );
  }

  public async rotateVapidKey(bindingIdArg: string): Promise<IWebPushBinding> {
    this.requireReady();
    await this.cleanupRetiredVapidKeys(bindingIdArg);
    const binding = await this.getBindingById(bindingIdArg);
    if (
      !binding
      || !binding.enabled
      || binding.status !== 'active'
      || typeof binding.deletedAt === 'number'
    ) {
      throw new plugins.typedrequest.TypedResponseError('Active Web Push binding not found');
    }
    if ((binding.vapidKeys || []).length >= VAPID_KEY_RING_LIMIT) {
      throw new plugins.typedrequest.TypedResponseError(
        'Web Push VAPID key ring is full; retiring keys are still in use',
      );
    }
    const now = this.now();
    const owner = this.bindingOwner(binding);
    const nextKeys = (binding.vapidKeys || []).map((keyArg) => (
      keyArg.status === 'active'
        ? {
            ...keyArg,
            status: 'retiring' as const,
            retireAfter: now + VAPID_RETIREMENT_MS,
          }
        : keyArg
    ));
    nextKeys.push(this.createVapidKey(owner, binding.id, now));
    const result = await WebPushBindingDoc.getNativeCollection().updateOne(
      { id: binding.id, mutationRevision: binding.mutationRevision },
      {
        $set: {
          vapidKeys: nextKeys,
          updatedAt: now,
          mutationRevision: binding.mutationRevision + 1,
          _updatedAt: new Date(now).toISOString(),
        },
      },
    );
    if (result.matchedCount !== 1) {
      throw new plugins.typedrequest.TypedResponseError(
        'Web Push binding changed concurrently; retry VAPID rotation',
      );
    }
    const refreshed = await this.getBindingById(binding.id);
    if (!refreshed) throw new Error('Rotated Web Push binding disappeared');
    return toPublicWebPushBinding(refreshed);
  }

  public async authenticateAppCredential(
    authArg: IWebPushAppCredentialAuth,
  ): Promise<IWebPushAuthenticatedBinding> {
    this.requireReady();
    const credentialId = requireOpaqueId(authArg?.credentialId, 'Web Push credential id');
    const credentialSecret = requireNonEmptyString(
      authArg?.credentialSecret,
      'Web Push credential secret',
      512,
    );
    const binding = await this.getBindingByCredentialId(credentialId);
    if (
      !binding
      || !binding.enabled
      || binding.status !== 'active'
      || typeof binding.deletedAt === 'number'
    ) {
      throw new plugins.typedrequest.TypedResponseError('Web Push credential is invalid');
    }
    requireLifecycleGeneration(binding.lifecycleGeneration);
    const now = this.now();
    const credential = binding.credential.id === credentialId
      ? binding.credential
      : (binding.previousCredentials || []).find((entryArg) => entryArg.id === credentialId);
    const credentialUsable = credential
      && typeof credential.secretHash === 'string'
      && (
        credential.status === 'active'
        || (
          credential.status === 'rotated'
          && credential.validUntil !== null
          && credential.validUntil > now
        )
      );
    if (
      !credentialUsable
      || !this.crypto!.verifyHmac(
        'credential',
        this.credentialHmacValue(binding.id, credentialId, credentialSecret),
        credential.hmacKeyId,
        credential.secretHash!,
      )
    ) {
      throw new plugins.typedrequest.TypedResponseError('Web Push credential is invalid');
    }
    return {
      binding,
      credential,
    };
  }

  public async getServiceStatus(
    authArg: IWebPushAppCredentialAuth,
  ): Promise<IWebPushServiceStatus> {
    const context = await this.authenticateAppCredential(authArg);
    const activeVapidKey = context.binding.vapidKeys.find((keyArg) => keyArg.status === 'active');
    return {
      ready: Boolean(activeVapidKey?.privateKeyEnvelope),
      bindingId: context.binding.id,
      bindingStatus: context.binding.status,
      ...(activeVapidKey ? { activeVapidKey: toVapidKeyPublic(activeVapidKey) } : {}),
      retiringVapidKeys: context.binding.vapidKeys
        .filter((keyArg) => keyArg.status === 'retiring')
        .map((keyArg) => toVapidKeyPublic(keyArg)),
      maxPayloadBytes: MAX_PAYLOAD_BYTES,
      maxTtlSeconds: MAX_TTL_SECONDS,
      contentEncoding: 'aes128gcm',
      ...(!activeVapidKey ? { message: 'Web Push binding has no active VAPID key' } : {}),
    };
  }

  public async enqueue(
    authArg: IWebPushAppCredentialAuth,
    requestArg: {
      idempotencyKey: string;
      subscriptionId: string;
      subscription: IWebPushSubscription;
      vapidKeyId: string;
      payload: IWebPushNotificationPayload;
      ttlSeconds?: number;
      urgency?: TWebPushUrgency;
      collapseKey?: string;
    },
  ): Promise<IWebPushEnqueueResult> {
    const context = await this.authenticateAppCredential(authArg);
    const idempotencyKey = requireNonEmptyString(
      requestArg.idempotencyKey,
      'Web Push idempotencyKey',
      MAX_IDEMPOTENCY_KEY_LENGTH,
    );
    const subscriptionId = requireOpaqueId(
      requestArg.subscriptionId,
      'Web Push subscriptionId',
    );
    const subscription = normalizeSubscription(requestArg.subscription);
    const vapidKeyId = requireOpaqueId(requestArg.vapidKeyId, 'Web Push vapidKeyId');
    const payload = normalizePayload(requestArg.payload);
    const ttlSeconds = normalizeTtl(requestArg.ttlSeconds);
    const urgency = normalizeUrgency(requestArg.urgency);
    const collapseKey = requestArg.collapseKey === undefined
      ? undefined
      : requireNonEmptyString(
          requestArg.collapseKey,
          'Web Push collapseKey',
          MAX_COLLAPSE_KEY_LENGTH,
        );
    const now = this.now();
    const lifecycleGeneration = requireLifecycleGeneration(
      context.binding.lifecycleGeneration,
    );
    const vapidKey = context.binding.vapidKeys.find((keyArg) => (
      keyArg.id === vapidKeyId
      && Boolean(keyArg.privateKeyEnvelope)
      && (
        keyArg.status === 'active'
        || (
          keyArg.status === 'retiring'
          && keyArg.retireAfter !== null
          && keyArg.retireAfter > now
        )
      )
    ));
    if (!vapidKey) {
      throw new plugins.typedrequest.TypedResponseError(
        'Web Push VAPID key is not active for new delivery',
      );
    }
    const hmacKeyId = context.credential.hmacKeyId;
    const idempotencyKeyDigest = this.crypto!.hmac(
      'idempotency-key',
      `${context.credential.id}\0${idempotencyKey}`,
      hmacKeyId,
    ).digest;
    const canonicalRequest = canonicalJson({
      subscriptionId,
      subscription,
      vapidKeyId,
      payload,
      ttlSeconds,
      urgency,
      collapseKey: collapseKey ?? null,
    });
    const requestDigest = this.crypto!.hmac(
      'request',
      canonicalRequest,
      hmacKeyId,
    ).digest;
    const collection = WebPushSpoolDoc.getNativeCollection();
    const existing = await collection.findOne({
      bindingId: context.binding.id,
      lifecycleGeneration,
      credentialId: context.credential.id,
      idempotencyKeyDigest,
    });
    if (existing) {
      await this.assertBindingLifecycleActive(context.binding);
      const spool = asSpoolDoc(existing);
      if (!constantTimeDigestEqual(spool.requestDigest, requestDigest)) {
        throw new plugins.typedrequest.TypedResponseError(
          'Web Push idempotency key conflicts with a different request',
        );
      }
      return {
        accepted: true,
        spoolItemId: spool.id,
      };
    }
    await this.assertBindingLifecycleActive(context.binding);
    await this.reserveAdmission(context.binding);
    const nonterminalCount = await collection.countDocuments({
      bindingId: context.binding.id,
      lifecycleGeneration,
      state: { $in: NONTERMINAL_STATES },
    });
    if (nonterminalCount >= MAX_NONTERMINAL_ITEMS_PER_BINDING) {
      throw new plugins.typedrequest.TypedResponseError('Web Push binding queue quota exceeded');
    }
    const spoolId = plugins.uuid.v4();
    const aadBase = this.spoolAadBase(context.binding, spoolId);
    const payloadText = canonicalJson(payload);
    const endpointDigest = this.crypto!.hmac(
      'endpoint',
      `${context.binding.id}\0${subscription.endpoint}`,
      hmacKeyId,
    ).digest;
    const topic = collapseKey
      ? Buffer.from(
          this.crypto!.hmac(
            'topic',
            `${context.binding.id}\0${collapseKey}`,
            hmacKeyId,
          ).digest,
          'base64url',
        ).subarray(0, 24).toString('base64url')
      : null;
    const spool = new WebPushSpoolDoc();
    spool.id = spoolId;
    spool.bindingId = context.binding.id;
    spool.lifecycleGeneration = lifecycleGeneration;
    spool.credentialId = context.credential.id;
    spool.hmacKeyId = hmacKeyId;
    spool.idempotencyKeyDigest = idempotencyKeyDigest;
    spool.requestDigest = requestDigest;
    spool.subscriptionId = subscriptionId;
    spool.endpointDigest = endpointDigest;
    spool.vapidKeyId = vapidKeyId;
    spool.subscriptionEnvelope = this.crypto!.encryptJson(subscription, {
      ...aadBase,
      documentType: 'spool-subscription',
    });
    spool.payloadEnvelope = this.crypto!.encryptJson(payload, {
      ...aadBase,
      documentType: 'spool-payload',
    });
    spool.payloadBytes = Buffer.byteLength(payloadText, 'utf8');
    spool.ttlSeconds = ttlSeconds;
    spool.urgency = urgency;
    spool.topic = topic;
    spool.state = 'accepted';
    spool.attempts = 0;
    spool.acceptedAt = now;
    spool.updatedAt = now;
    spool.nextAttemptAt = now;
    spool.expiresAt = now + ttlSeconds * 1000;
    spool.pushServiceStatusCode = null;
    spool.errorCode = null;
    try {
      await spool.save();
    } catch (error: unknown) {
      if (!isDuplicateKeyError(error)) throw error;
      const raced = await collection.findOne({
        bindingId: context.binding.id,
        lifecycleGeneration,
        credentialId: context.credential.id,
        idempotencyKeyDigest,
      });
      if (!raced) throw error;
      const racedSpool = asSpoolDoc(raced);
      if (!constantTimeDigestEqual(racedSpool.requestDigest, requestDigest)) {
        throw new plugins.typedrequest.TypedResponseError(
          'Web Push idempotency key conflicts with a different request',
        );
      }
      await this.cancelSpoolIfLifecycleChanged(
        racedSpool.id,
        context.binding.id,
        lifecycleGeneration,
      );
      return {
        accepted: true,
        spoolItemId: racedSpool.id,
      };
    }
    await this.cancelSpoolIfLifecycleChanged(
      spool.id,
      context.binding.id,
      lifecycleGeneration,
    );
    return {
      accepted: true,
      spoolItemId: spool.id,
    };
  }

  public async getDeliveryStatus(
    authArg: IWebPushAppCredentialAuth,
    spoolItemIdArg: string,
  ): Promise<IWebPushDeliveryStatus | undefined> {
    const context = await this.authenticateAppCredential(authArg);
    const spoolItemId = requireOpaqueId(spoolItemIdArg, 'Web Push spoolItemId');
    const lifecycleGeneration = requireLifecycleGeneration(
      context.binding.lifecycleGeneration,
    );
    const row = await WebPushSpoolDoc.getNativeCollection().findOne({
      id: spoolItemId,
      bindingId: context.binding.id,
      lifecycleGeneration,
    });
    return row ? toPublicWebPushDelivery(asSpoolDoc(row)) : undefined;
  }

  public async cancel(
    authArg: IWebPushAppCredentialAuth,
    targetArg: TWebPushCancellationTarget,
  ): Promise<IWebPushCancelResult> {
    const context = await this.authenticateAppCredential(authArg);
    return await this.cancelWithinBinding(
      context.binding.id,
      targetArg,
      false,
      context.binding.lifecycleGeneration,
    );
  }

  public async processDueDeliveriesOnce(limitArg = DELIVERY_BATCH_SIZE): Promise<number> {
    this.requireReady();
    const limit = Math.min(Math.max(Math.floor(limitArg), 1), 100);
    await this.runMaintenanceOnce();
    await this.expireDueItems(limit);
    let processed = 0;
    for (let index = 0; index < limit; index++) {
      // stop() flips readiness before waiting for the active cycle. Check the
      // boundary between deliveries so shutdown waits for at most one send.
      if (!this.ready) break;
      const claimed = await this.claimNextItem();
      if (!claimed) break;
      await this.processClaimedItem(claimed);
      processed++;
    }
    return processed;
  }

  private requirePersistenceReady(): void {
    if (!this.started || !this.enabled || !this.crypto) {
      throw new plugins.typedrequest.TypedResponseError('Web Push provider is unavailable');
    }
  }

  private requireReady(): void {
    if (!this.ready || !this.crypto || !this.vapidSubject) {
      throw new plugins.typedrequest.TypedResponseError('Web Push provider is unavailable');
    }
  }

  private bindingOwner(bindingArg: WebPushBindingDoc): IWebPushResourceOwner {
    return {
      gatewayClientType: bindingArg.ownerGatewayClientType,
      gatewayClientId: bindingArg.ownerGatewayClientId,
      appInstanceId: bindingArg.ownerAppInstanceId,
    };
  }

  private assertBindingOwner(
    bindingArg: WebPushBindingDoc,
    ownerArg: IWebPushResourceOwner,
  ): void {
    if (
      bindingArg.ownerGatewayClientType !== ownerArg.gatewayClientType
      || bindingArg.ownerGatewayClientId !== ownerArg.gatewayClientId
      || bindingArg.ownerAppInstanceId !== ownerArg.appInstanceId
    ) {
      throw new plugins.typedrequest.TypedResponseError(
        'Web Push binding owner cannot be changed',
      );
    }
  }

  private credentialHmacValue(
    bindingIdArg: string,
    credentialIdArg: string,
    secretArg: string,
  ): string {
    return `${bindingIdArg}\0${credentialIdArg}\0${secretArg}`;
  }

  private createCredential(bindingIdArg: string, nowArg: number): {
    stored: IWebPushStoredCredential;
    oneTime: IWebPushCredentialOneTimeSecret;
  } {
    const id = plugins.uuid.v4();
    const secret = plugins.crypto.randomBytes(32).toString('base64url');
    const hmac = this.crypto!.hmac(
      'credential',
      this.credentialHmacValue(bindingIdArg, id, secret),
    );
    const stored: IWebPushStoredCredential = {
      id,
      status: 'active',
      hmacKeyId: hmac.keyId,
      secretHash: hmac.digest,
      createdAt: nowArg,
      updatedAt: nowArg,
      lastRotatedAt: null,
      validUntil: null,
    };
    return {
      stored,
      oneTime: {
        credential: toCredentialPublic(stored),
        secret,
        secretShownOnce: true,
      },
    };
  }

  private credentialRotationRecoveryAad(
    bindingArg: WebPushBindingDoc,
    identityArg: Pick<
      IWebPushCredentialRotationRecovery,
      'requestedCredentialId' | 'resultingCredentialId' | 'lifecycleGeneration'
    >,
  ): IWebPushAadContext {
    return {
      gatewayClientId: bindingArg.ownerGatewayClientId,
      appInstanceId: bindingArg.ownerAppInstanceId,
      bindingId: bindingArg.id,
      documentType: 'credential-rotation-recovery',
      documentId: canonicalJson({
        lifecycleGeneration: identityArg.lifecycleGeneration,
        requestedCredentialId: identityArg.requestedCredentialId,
        resultingCredentialId: identityArg.resultingCredentialId,
      }),
    };
  }

  private replayCredentialRotation(
    bindingArg: WebPushBindingDoc,
    requestedCredentialIdArg: string,
    nowArg: number,
  ): IWebPushCredentialOneTimeSecret | undefined {
    const recovery = bindingArg.credentialRotationRecovery;
    if (
      !recovery
      || recovery.expiresAt <= nowArg
      || (
        recovery.requestedCredentialId !== requestedCredentialIdArg
        && recovery.resultingCredentialId !== requestedCredentialIdArg
      )
    ) {
      return undefined;
    }
    const lifecycleGeneration = requireLifecycleGeneration(
      bindingArg.lifecycleGeneration,
    );
    if (
      !bindingArg.enabled
      || bindingArg.status !== 'active'
      || typeof bindingArg.deletedAt === 'number'
      || recovery.lifecycleGeneration !== lifecycleGeneration
      || recovery.resultingCredentialId !== bindingArg.credential.id
      || bindingArg.credential.status !== 'active'
      || typeof bindingArg.credential.secretHash !== 'string'
    ) {
      throw new Error('Web Push credential rotation recovery state is inconsistent');
    }
    let secret: string;
    try {
      secret = this.crypto!.decryptJson<{ secret: string }>(
        recovery.secretEnvelope,
        this.credentialRotationRecoveryAad(bindingArg, recovery),
      ).secret;
      secret = requireNonEmptyString(
        secret,
        'Web Push credential rotation recovery secret',
        512,
      );
    } catch (error: unknown) {
      throw new Error('Web Push credential rotation recovery is unreadable', {
        cause: error,
      });
    }
    if (
      !this.crypto!.verifyHmac(
        'credential',
        this.credentialHmacValue(
          bindingArg.id,
          bindingArg.credential.id,
          secret,
        ),
        bindingArg.credential.hmacKeyId,
        bindingArg.credential.secretHash,
      )
    ) {
      throw new Error('Web Push credential rotation recovery failed verification');
    }
    return {
      credential: toCredentialPublic(bindingArg.credential),
      secret,
      secretShownOnce: true,
    };
  }

  private createVapidKey(
    ownerArg: IWebPushResourceOwner,
    bindingIdArg: string,
    nowArg: number,
  ): IWebPushStoredVapidKey {
    const id = plugins.uuid.v4();
    const keyPair = plugins.webpush.generateVAPIDKeys();
    const aad: IWebPushAadContext = {
      gatewayClientId: ownerArg.gatewayClientId,
      appInstanceId: ownerArg.appInstanceId,
      bindingId: bindingIdArg,
      documentType: 'vapid-private-key',
      documentId: id,
    };
    return {
      id,
      publicKey: keyPair.publicKey,
      status: 'active',
      createdAt: nowArg,
      activatedAt: nowArg,
      retireAfter: null,
      retiredAt: null,
      privateKeyEnvelope: this.crypto!.encryptJson(
        { privateKey: keyPair.privateKey },
        aad,
      ),
    };
  }

  private spoolAadBase(
    bindingArg: WebPushBindingDoc,
    spoolIdArg: string,
  ): Omit<IWebPushAadContext, 'documentType'> {
    return {
      gatewayClientId: bindingArg.ownerGatewayClientId,
      appInstanceId: bindingArg.ownerAppInstanceId,
      bindingId: bindingArg.id,
      documentId: spoolIdArg,
    };
  }

  private async assertBindingLifecycleActive(bindingArg: WebPushBindingDoc): Promise<void> {
    const lifecycleGeneration = requireLifecycleGeneration(bindingArg.lifecycleGeneration);
    const active = await WebPushBindingDoc.getNativeCollection().findOne({
      id: bindingArg.id,
      lifecycleGeneration,
      enabled: true,
      status: 'active',
      deletedAt: { $exists: false },
    }, { projection: { _id: 1 } });
    if (!active) {
      throw new plugins.typedrequest.TypedResponseError(
        'Web Push binding lifecycle changed; retry with current credentials',
      );
    }
  }

  private async reserveAdmission(bindingArg: WebPushBindingDoc): Promise<void> {
    const bindingId = requireOpaqueId(bindingArg.id, 'Web Push binding id');
    const lifecycleGeneration = requireLifecycleGeneration(bindingArg.lifecycleGeneration);
    await this.assertBindingLifecycleActive(bindingArg);
    const collection = WebPushAdmissionDoc.getNativeCollection();
    const policy = this.admissionPolicy;

    for (let attempt = 0; attempt < MAX_ADMISSION_CAS_ATTEMPTS; attempt++) {
      const now = this.now();
      const purgeAt = new Date(now + policy.longWindowMs * 2);
      const raw = await collection.findOne({
        bindingId,
        lifecycleGeneration,
      });
      if (!raw) {
        const admission = new WebPushAdmissionDoc();
        admission.bindingId = bindingId;
        admission.lifecycleGeneration = lifecycleGeneration;
        admission.shortWindowStartedAt = now;
        admission.shortWindowCount = 1;
        admission.longWindowStartedAt = now;
        admission.longWindowCount = 1;
        admission.revision = 1;
        admission.updatedAt = now;
        admission.purgeAt = purgeAt;
        try {
          await admission.save();
          return;
        } catch (error: unknown) {
          if (isDuplicateKeyError(error)) continue;
          throw error;
        }
      }

      const admission = asAdmissionDoc(raw);
      const admissionGeneration = requireLifecycleGeneration(admission.lifecycleGeneration);
      if (admissionGeneration !== lifecycleGeneration) {
        throw new Error('Web Push admission lifecycle generation is inconsistent');
      }
      for (const [name, value] of Object.entries({
        shortWindowStartedAt: admission.shortWindowStartedAt,
        shortWindowCount: admission.shortWindowCount,
        longWindowStartedAt: admission.longWindowStartedAt,
        longWindowCount: admission.longWindowCount,
        revision: admission.revision,
      })) {
        if (!Number.isSafeInteger(value) || value < 0 || (name === 'revision' && value < 1)) {
          throw new Error(`Web Push admission document ${name} is malformed`);
        }
      }

      const shortWindowExpired = now - admission.shortWindowStartedAt >= policy.shortWindowMs;
      const longWindowExpired = now - admission.longWindowStartedAt >= policy.longWindowMs;
      const shortWindowStartedAt = shortWindowExpired
        ? now
        : admission.shortWindowStartedAt;
      const longWindowStartedAt = longWindowExpired
        ? now
        : admission.longWindowStartedAt;
      const shortWindowCount = shortWindowExpired ? 0 : admission.shortWindowCount;
      const longWindowCount = longWindowExpired ? 0 : admission.longWindowCount;
      if (shortWindowCount >= policy.shortLimit || longWindowCount >= policy.longLimit) {
        throw new plugins.typedrequest.TypedResponseError(
          'Web Push request rate limit exceeded',
        );
      }

      const updated = await collection.updateOne(
        {
          bindingId,
          lifecycleGeneration,
          revision: admission.revision,
        },
        {
          $set: {
            shortWindowStartedAt,
            shortWindowCount: shortWindowCount + 1,
            longWindowStartedAt,
            longWindowCount: longWindowCount + 1,
            revision: admission.revision + 1,
            updatedAt: now,
            purgeAt,
            _updatedAt: new Date(now).toISOString(),
          },
        },
      );
      if (updated.matchedCount === 1) return;
    }

    throw new plugins.typedrequest.TypedResponseError(
      'Web Push admission changed concurrently; retry request',
    );
  }

  private async cancelSpoolIfLifecycleChanged(
    spoolItemIdArg: string,
    bindingIdArg: string,
    lifecycleGenerationArg: number,
  ): Promise<void> {
    const lifecycleGeneration = requireLifecycleGeneration(lifecycleGenerationArg);
    const active = await WebPushBindingDoc.getNativeCollection().findOne({
      id: bindingIdArg,
      lifecycleGeneration,
      enabled: true,
      status: 'active',
      deletedAt: { $exists: false },
    }, { projection: { _id: 1 } });
    if (active) return;
    await this.cancelWithinBinding(
      bindingIdArg,
      {
        type: 'spoolItem',
        spoolItemId: spoolItemIdArg,
      },
      false,
      lifecycleGeneration,
      'BINDING_LIFECYCLE_CHANGED',
    );
    await WebPushAdmissionDoc.getNativeCollection().deleteOne({
      bindingId: bindingIdArg,
      lifecycleGeneration,
    });
  }

  private async scrubDeletedBindingLifecycle(
    bindingArg: WebPushBindingDoc,
    waitForDrainArg = true,
  ): Promise<void> {
    const tombstoneGeneration = requireLifecycleGeneration(bindingArg.lifecycleGeneration);
    if (typeof bindingArg.deletedAt !== 'number') return;
    const cancellationErrorCode = bindingArg.controllerFence?.intent === 'disabled'
      ? 'BINDING_DISABLED'
      : 'BINDING_DELETED';
    await this.requestBindingLifecyclesBeforeCancellation(
      bindingArg.id,
      tombstoneGeneration,
      cancellationErrorCode,
    );
    const collection = WebPushBindingDoc.getNativeCollection();
    let scrubbed = false;
    for (let attempt = 0; attempt < 4; attempt++) {
      const currentRaw = await collection.findOne({
        id: bindingArg.id,
        lifecycleGeneration: tombstoneGeneration,
        deletedAt: bindingArg.deletedAt,
        enabled: false,
      });
      if (!currentRaw) return;
      const current = asBindingDoc(currentRaw);
      const needsScrub = (
        typeof current.credential?.secretHash === 'string'
        || (current.previousCredentials || []).length > 0
        || current.credentialRotationRecovery !== undefined
        || (current.vapidKeys || []).some((keyArg) => (
          keyArg.privateKeyEnvelope !== undefined
          || keyArg.status !== 'retired'
          || keyArg.retireAfter !== null
        ))
      );
      if (!needsScrub) {
        scrubbed = true;
        break;
      }
      const now = this.now();
      const credential: IWebPushStoredCredential = {
        ...current.credential,
        status: 'revoked',
        updatedAt: now,
        validUntil: null,
      };
      delete credential.secretHash;
      const vapidKeys = (current.vapidKeys || []).map((keyArg) => {
        const scrubbedKey: IWebPushStoredVapidKey = {
          ...keyArg,
          status: 'retired',
          retireAfter: null,
          retiredAt: keyArg.retiredAt ?? now,
        };
        delete scrubbedKey.privateKeyEnvelope;
        return scrubbedKey;
      });
      const result = await collection.updateOne(
        {
          id: current.id,
          mutationRevision: current.mutationRevision,
          lifecycleGeneration: tombstoneGeneration,
          deletedAt: bindingArg.deletedAt,
          enabled: false,
        },
        {
          $set: {
            credential,
            previousCredentials: [],
            vapidKeys,
            updatedAt: now,
            mutationRevision: current.mutationRevision + 1,
            _updatedAt: new Date(now).toISOString(),
          },
          $unset: {
            credentialRotationRecovery: '',
          },
        },
      );
      if (result.matchedCount === 1) {
        scrubbed = true;
        break;
      }
    }
    if (!scrubbed) {
      throw new plugins.typedrequest.TypedResponseError(
        'Web Push binding cleanup changed concurrently; retry deletion',
      );
    }
    await WebPushAdmissionDoc.getNativeCollection().deleteMany({
      bindingId: bindingArg.id,
      lifecycleGeneration: { $lte: tombstoneGeneration },
    });
    await this.finishStaleCancelledSendingBefore(
      bindingArg.id,
      tombstoneGeneration,
      cancellationErrorCode,
    );
    if (waitForDrainArg) {
      await this.waitForBindingLifecyclesBeforeDrain(
        bindingArg.id,
        tombstoneGeneration,
        cancellationErrorCode,
      );
    }
  }

  private async cleanupExpiredCredentialOverlaps(
    limitArg = MAINTENANCE_BATCH_SIZE,
  ): Promise<void> {
    const now = this.now();
    const collection = WebPushBindingDoc.getNativeCollection();
    const batch = await this.selectBindingMaintenanceBatch(
      { 'previousCredentials.validUntil': { $lte: now } },
      this.expiredCredentialOverlapCursor,
      limitArg,
    );
    this.expiredCredentialOverlapCursor = batch.nextCursor;
    for (const binding of batch.rows) {
      const remaining = (binding.previousCredentials || []).filter(
        (credentialArg) => (
          typeof credentialArg.secretHash === 'string'
          && credentialArg.validUntil !== null
          && credentialArg.validUntil > now
        ),
      );
      if (remaining.length === (binding.previousCredentials || []).length) continue;
      await collection.updateOne(
        {
          id: binding.id,
          mutationRevision: binding.mutationRevision,
          lifecycleGeneration: binding.lifecycleGeneration,
        },
        {
          $set: {
            previousCredentials: remaining,
            updatedAt: now,
            mutationRevision: binding.mutationRevision + 1,
            _updatedAt: new Date(now).toISOString(),
          },
        },
      );
    }
  }

  private async cleanupExpiredCredentialRotationRecoveries(
    limitArg = MAINTENANCE_BATCH_SIZE,
  ): Promise<void> {
    const now = this.now();
    const collection = WebPushBindingDoc.getNativeCollection();
    const batch = await this.selectBindingMaintenanceBatch(
      { 'credentialRotationRecovery.expiresAt': { $lte: now } },
      this.expiredCredentialRecoveryCursor,
      limitArg,
    );
    this.expiredCredentialRecoveryCursor = batch.nextCursor;
    for (const binding of batch.rows) {
      const recovery = binding.credentialRotationRecovery;
      if (!recovery || recovery.expiresAt > now) continue;
      await collection.updateOne(
        {
          id: binding.id,
          mutationRevision: binding.mutationRevision,
          lifecycleGeneration: binding.lifecycleGeneration,
          'credentialRotationRecovery.requestedCredentialId':
            recovery.requestedCredentialId,
          'credentialRotationRecovery.resultingCredentialId':
            recovery.resultingCredentialId,
          'credentialRotationRecovery.expiresAt': recovery.expiresAt,
        },
        {
          $set: {
            updatedAt: now,
            mutationRevision: binding.mutationRevision + 1,
            _updatedAt: new Date(now).toISOString(),
          },
          $unset: {
            credentialRotationRecovery: '',
          },
        },
      );
    }
  }

  private async cleanupRetiredVapidKeys(
    bindingIdArg?: string,
    limitArg = MAINTENANCE_BATCH_SIZE,
  ): Promise<void> {
    const now = this.now();
    const bindingCollection = WebPushBindingDoc.getNativeCollection();
    const spoolCollection = WebPushSpoolDoc.getNativeCollection();
    const selector: Record<string, unknown> = bindingIdArg
      ? { id: requireOpaqueId(bindingIdArg, 'Web Push binding id') }
      : {
          $or: [
            { 'vapidKeys.status': 'retired' },
            { 'vapidKeys.retireAfter': { $lte: now } },
          ],
        };
    const rows = bindingIdArg
      ? (await bindingCollection
          .find(selector)
          .limit(1)
          .toArray())
          .map((rawArg) => asBindingDoc(rawArg))
      : await (async () => {
          const batch = await this.selectBindingMaintenanceBatch(
            selector,
            this.retiredVapidCursor,
            limitArg,
          );
          this.retiredVapidCursor = batch.nextCursor;
          return batch.rows;
        })();
    for (const binding of rows) {
      const retained: IWebPushStoredVapidKey[] = [];
      for (const key of binding.vapidKeys || []) {
        if (key.status === 'active') {
          retained.push(key);
          continue;
        }
        if (
          key.status === 'retiring'
          && (key.retireAfter === null || key.retireAfter > now)
        ) {
          retained.push(key);
          continue;
        }
        if (key.status === 'retired') continue;
        const pending = await spoolCollection.countDocuments({
          bindingId: binding.id,
          lifecycleGeneration: requireLifecycleGeneration(binding.lifecycleGeneration),
          vapidKeyId: key.id,
          state: { $in: NONTERMINAL_STATES },
        }, { limit: 1 });
        if (pending > 0) {
          retained.push(key);
        }
      }
      if (retained.length === binding.vapidKeys.length) continue;
      await bindingCollection.updateOne(
        {
          id: binding.id,
          mutationRevision: binding.mutationRevision,
          lifecycleGeneration: binding.lifecycleGeneration,
        },
        {
          $set: {
            vapidKeys: retained,
            updatedAt: now,
            mutationRevision: binding.mutationRevision + 1,
            _updatedAt: new Date(now).toISOString(),
          },
        },
      );
    }
  }

  private async runMaintenanceOnce(): Promise<void> {
    await this.cleanupExpiredCredentialOverlaps();
    await this.cleanupExpiredCredentialRotationRecoveries();
    await this.cleanupRetiredVapidKeys();
    await this.finishStaleCancelledSending();
    await this.cleanupInactiveBindingLifecycles();
  }

  private async cleanupInactiveBindingLifecycles(
    limitArg = Math.min(MAINTENANCE_BATCH_SIZE, 20),
  ): Promise<void> {
    const batch = await this.selectBindingMaintenanceBatch(
      {
        $or: [
          { deletedAt: { $type: 'number' } },
          { enabled: false },
        ],
      },
      this.inactiveLifecycleCursor,
      limitArg,
    );
    this.inactiveLifecycleCursor = batch.nextCursor;
    for (const binding of batch.rows) {
      try {
        const lifecycleGeneration = requireLifecycleGeneration(
          binding.lifecycleGeneration,
        );
        if (typeof binding.deletedAt === 'number') {
          await this.scrubDeletedBindingLifecycle(binding, false);
        } else {
          await this.requestBindingLifecyclesBeforeCancellation(
            binding.id,
            lifecycleGeneration,
            'BINDING_DISABLED',
          );
          await WebPushAdmissionDoc.getNativeCollection().deleteMany({
            bindingId: binding.id,
            lifecycleGeneration: { $lt: lifecycleGeneration },
          });
          await this.finishStaleCancelledSendingBefore(
            binding.id,
            lifecycleGeneration,
            'BINDING_DISABLED',
          );
        }
      } catch (error: unknown) {
        logger.log(
          'warn',
          `Web Push lifecycle maintenance will retry binding ${binding.id}: ${(error as Error).name}`,
        );
      }
    }
  }

  private async selectBindingMaintenanceBatch(
    selectorArg: Record<string, unknown>,
    cursorArg: string | undefined,
    limitArg: number,
  ): Promise<{
    rows: WebPushBindingDoc[];
    nextCursor?: string;
  }> {
    const selector = cursorArg === undefined
      ? selectorArg
      : {
          $and: [
            selectorArg,
            { id: { $gt: cursorArg } },
          ],
        };
    const rows = (await WebPushBindingDoc.getNativeCollection()
      .find(selector)
      .sort({ id: 1 })
      .limit(Math.min(Math.max(Math.floor(limitArg), 1), MAINTENANCE_BATCH_SIZE))
      .toArray())
      .map((rawArg) => asBindingDoc(rawArg));
    return {
      rows,
      ...(rows.length > 0 ? { nextCursor: rows[rows.length - 1].id } : {}),
    };
  }

  private async requestBindingLifecyclesBeforeCancellation(
    bindingIdArg: string,
    beforeLifecycleGenerationArg: number,
    errorCodeArg: string,
  ): Promise<void> {
    const beforeLifecycleGeneration = requireLifecycleGeneration(
      beforeLifecycleGenerationArg,
    );
    await this.cancelWithinBinding(
      bindingIdArg,
      {
        type: 'subscription',
        subscriptionId: '*',
      },
      true,
      { $lt: beforeLifecycleGeneration },
      errorCodeArg,
    );
  }

  private async drainBindingLifecyclesBefore(
    bindingIdArg: string,
    beforeLifecycleGenerationArg: number,
    errorCodeArg: string,
  ): Promise<void> {
    const beforeLifecycleGeneration = requireLifecycleGeneration(
      beforeLifecycleGenerationArg,
    );
    await this.requestBindingLifecyclesBeforeCancellation(
      bindingIdArg,
      beforeLifecycleGeneration,
      errorCodeArg,
    );
    await WebPushAdmissionDoc.getNativeCollection().deleteMany({
      bindingId: bindingIdArg,
      lifecycleGeneration: { $lt: beforeLifecycleGeneration },
    });
    await this.waitForBindingLifecyclesBeforeDrain(
      bindingIdArg,
      beforeLifecycleGeneration,
      errorCodeArg,
    );
  }

  private async waitForBindingLifecyclesBeforeDrain(
    bindingIdArg: string,
    beforeLifecycleGenerationArg: number,
    errorCodeArg: string,
  ): Promise<void> {
    const bindingId = requireOpaqueId(bindingIdArg, 'Web Push binding id');
    const beforeLifecycleGeneration = requireLifecycleGeneration(
      beforeLifecycleGenerationArg,
    );
    const deadline = Date.now() + LIFECYCLE_DRAIN_TIMEOUT_MS;
    const collection = WebPushSpoolDoc.getNativeCollection();
    while (true) {
      await this.requestBindingLifecyclesBeforeCancellation(
        bindingId,
        beforeLifecycleGeneration,
        errorCodeArg,
      );
      while (
        await this.finishStaleCancelledSendingBefore(
          bindingId,
          beforeLifecycleGeneration,
          errorCodeArg,
        ) > 0
      ) {
        // Drain every bounded stale-lease batch before checking completion.
      }
      const remaining = await collection.countDocuments({
        bindingId,
        lifecycleGeneration: { $lt: beforeLifecycleGeneration },
        state: { $in: NONTERMINAL_STATES },
      }, { limit: 1 });
      if (remaining === 0) return;
      if (Date.now() >= deadline) {
        throw new plugins.typedrequest.TypedResponseError(
          'Web Push binding cleanup is still draining in-flight delivery; retry',
        );
      }
      await new Promise<void>((resolveArg) => {
        setTimeout(resolveArg, LIFECYCLE_DRAIN_POLL_MS);
      });
    }
  }

  private async finishStaleCancelledSending(
    limitArg = MAINTENANCE_BATCH_SIZE,
  ): Promise<number> {
    const now = this.now();
    const collection = WebPushSpoolDoc.getNativeCollection();
    const rows = await collection.find({
      state: 'sending',
      cancelRequestedAt: { $exists: true },
      $or: [
        { leaseToken: { $exists: false } },
        { leaseExpiresAt: { $exists: false } },
        { leaseExpiresAt: { $lte: now } },
      ],
    }, {
      projection: {
        id: 1,
        leaseToken: 1,
        leaseExpiresAt: 1,
        cancelRequestedAt: 1,
      },
    })
      .sort({ cancelRequestedAt: 1, id: 1 })
      .limit(Math.min(Math.max(Math.floor(limitArg), 1), MAINTENANCE_BATCH_SIZE))
      .toArray();
    let finished = 0;
    for (const raw of rows) {
      const spool = asSpoolDoc(raw);
      const result = await collection.updateOne(
        {
          id: spool.id,
          state: 'sending',
          cancelRequestedAt: spool.cancelRequestedAt,
          ...(spool.leaseToken === undefined
            ? { leaseToken: { $exists: false } }
            : { leaseToken: spool.leaseToken }),
          ...(spool.leaseExpiresAt === undefined
            ? { leaseExpiresAt: { $exists: false } }
            : { leaseExpiresAt: spool.leaseExpiresAt }),
        },
        buildWebPushTerminalUpdate({
          state: 'failed',
          now,
          errorCode: 'DELIVERY_OUTCOME_UNKNOWN_AFTER_CANCELLATION',
        }),
      );
      finished += result.modifiedCount;
    }
    return finished;
  }

  private async finishStaleCancelledSendingBefore(
    bindingIdArg: string,
    beforeLifecycleGenerationArg: number,
    _fallbackErrorCodeArg: string,
  ): Promise<number> {
    const bindingId = requireOpaqueId(bindingIdArg, 'Web Push binding id');
    const beforeLifecycleGeneration = requireLifecycleGeneration(
      beforeLifecycleGenerationArg,
    );
    const now = this.now();
    const collection = WebPushSpoolDoc.getNativeCollection();
    const rows = await collection.find({
      bindingId,
      lifecycleGeneration: { $lt: beforeLifecycleGeneration },
      state: 'sending',
      cancelRequestedAt: { $exists: true },
      $or: [
        { leaseToken: { $exists: false } },
        { leaseExpiresAt: { $exists: false } },
        { leaseExpiresAt: { $lte: now } },
      ],
    }, {
      projection: {
        id: 1,
        leaseToken: 1,
        leaseExpiresAt: 1,
        cancelRequestedAt: 1,
        cancelErrorCode: 1,
      },
    }).limit(MAINTENANCE_BATCH_SIZE).toArray();
    let finished = 0;
    for (const raw of rows) {
      const spool = asSpoolDoc(raw);
      const result = await collection.updateOne(
        {
          id: spool.id,
          bindingId,
          state: 'sending',
          cancelRequestedAt: spool.cancelRequestedAt,
          ...(spool.leaseToken === undefined
            ? { leaseToken: { $exists: false } }
            : { leaseToken: spool.leaseToken }),
          ...(spool.leaseExpiresAt === undefined
            ? { leaseExpiresAt: { $exists: false } }
            : { leaseExpiresAt: spool.leaseExpiresAt }),
        },
        buildWebPushTerminalUpdate({
          state: 'failed',
          now,
          errorCode: 'DELIVERY_OUTCOME_UNKNOWN_AFTER_CANCELLATION',
        }),
      );
      finished += result.modifiedCount;
    }
    return finished;
  }

  private async cancelWithinBinding(
    bindingIdArg: string,
    targetArg: TWebPushCancellationTarget,
    allSubscriptionsArg = false,
    lifecycleGenerationArg?: number | { $lt: number },
    errorCodeArg = 'CANCELLED_BY_APPLICATION',
  ): Promise<IWebPushCancelResult> {
    const bindingId = requireOpaqueId(bindingIdArg, 'Web Push binding id');
    const targetSelector: Record<string, unknown> = {};
    if (targetArg.type === 'spoolItem') {
      targetSelector.id = requireOpaqueId(targetArg.spoolItemId, 'Web Push spoolItemId');
    } else if (allSubscriptionsArg) {
      targetSelector.subscriptionId = { $type: 'string' };
    } else {
      targetSelector.subscriptionId = requireOpaqueId(
        targetArg.subscriptionId,
        'Web Push subscriptionId',
      );
    }
    const collection = WebPushSpoolDoc.getNativeCollection();
    const lifecycleGenerationSelector = typeof lifecycleGenerationArg === 'number'
      ? requireLifecycleGeneration(lifecycleGenerationArg)
      : lifecycleGenerationArg === undefined
        ? undefined
        : { $lt: requireLifecycleGeneration(lifecycleGenerationArg.$lt) };
    const selector = {
      bindingId,
      ...(lifecycleGenerationSelector === undefined
        ? {}
        : { lifecycleGeneration: lifecycleGenerationSelector }),
      ...targetSelector,
    };
    const alreadyTerminalCount = await collection.countDocuments({
      ...selector,
      state: { $in: TERMINAL_STATES },
    });
    let cancelledCount = 0;
    for (let attempt = 0; attempt < 16; attempt++) {
      const now = this.now();
      const terminalized = await collection.updateMany(
        {
          ...selector,
          state: { $in: ['accepted', 'queued', 'deferred'] },
        },
        buildWebPushTerminalUpdate({
          state: 'cancelled',
          now,
          errorCode: errorCodeArg,
        }),
      );
      const markedSending = await collection.updateMany(
        {
          ...selector,
          state: 'sending',
          cancelRequestedAt: { $exists: false },
        },
        {
          $set: {
            cancelRequestedAt: now,
            cancelErrorCode: errorCodeArg,
            updatedAt: now,
            _updatedAt: new Date(now).toISOString(),
          },
        },
      );
      const modified = terminalized.modifiedCount + markedSending.modifiedCount;
      cancelledCount += modified;
      if (modified === 0) break;
    }
    return {
      cancelledCount,
      alreadyTerminalCount,
    };
  }

  private async expireDueItems(limitArg: number): Promise<void> {
    const now = this.now();
    const collection = WebPushSpoolDoc.getNativeCollection();
    const rows = await collection
      .find({
        state: { $in: NONTERMINAL_STATES },
        expiresAt: { $lte: now },
      }, {
        projection: { id: 1, state: 1, leaseToken: 1, leaseExpiresAt: 1 },
      })
      .limit(limitArg)
      .toArray();
    for (const raw of rows) {
      const spool = asSpoolDoc(raw);
      if (
        spool.state === 'sending'
        && typeof spool.leaseExpiresAt === 'number'
        && spool.leaseExpiresAt > now
      ) {
        continue;
      }
      const leaseSelector = spool.state === 'sending'
        ? {
            ...(spool.leaseToken === undefined
              ? { leaseToken: { $exists: false } }
              : { leaseToken: spool.leaseToken }),
            ...(spool.leaseExpiresAt === undefined
              ? { leaseExpiresAt: { $exists: false } }
              : { leaseExpiresAt: spool.leaseExpiresAt }),
          }
        : {};
      await collection.updateOne(
        {
          id: spool.id,
          state: spool.state,
          ...leaseSelector,
        },
        buildWebPushTerminalUpdate({
          state: 'expired',
          now,
          errorCode: 'DELIVERY_TTL_EXPIRED',
        }),
      );
    }
  }

  private async claimNextItem(): Promise<WebPushSpoolDoc | null> {
    const now = this.now();
    const leaseToken = plugins.uuid.v4();
    const row = await WebPushSpoolDoc.getNativeCollection().findOneAndUpdate(
      {
        expiresAt: { $gt: now },
        $or: [
          {
            state: { $in: ['accepted', 'queued', 'deferred'] },
            nextAttemptAt: { $lte: now },
            cancelRequestedAt: { $exists: false },
            $or: [
              { leaseToken: { $exists: false } },
              { leaseExpiresAt: { $lte: now } },
            ],
          },
          {
            state: 'sending',
            leaseExpiresAt: { $lte: now },
            cancelRequestedAt: { $exists: false },
          },
        ],
      },
      {
        $set: {
          state: 'sending',
          leaseToken,
          leaseExpiresAt: now + DELIVERY_LEASE_MS,
          updatedAt: now,
          _updatedAt: new Date(now).toISOString(),
        },
        $inc: { attempts: 1 },
      },
      {
        sort: { nextAttemptAt: 1, acceptedAt: 1, id: 1 },
        returnDocument: 'after',
      },
    );
    return row ? asSpoolDoc(row) : null;
  }

  private async processClaimedItem(spoolArg: WebPushSpoolDoc): Promise<void> {
    const collection = WebPushSpoolDoc.getNativeCollection();
    const leaseToken = spoolArg.leaseToken;
    if (!leaseToken) return;
    const now = this.now();
    const current = await collection.findOne({
      id: spoolArg.id,
      state: 'sending',
      leaseToken,
    });
    if (!current) return;
    const currentSpool = asSpoolDoc(current);
    if (currentSpool.cancelRequestedAt !== undefined) {
      await this.finishTerminal(currentSpool, 'cancelled', {
        errorCode: currentSpool.cancelErrorCode || 'CANCELLED_BY_APPLICATION',
      });
      return;
    }
    if (currentSpool.expiresAt <= now) {
      await this.finishTerminal(currentSpool, 'expired', {
        errorCode: 'DELIVERY_TTL_EXPIRED',
      });
      return;
    }
    let lifecycleGeneration: number;
    try {
      lifecycleGeneration = requireLifecycleGeneration(currentSpool.lifecycleGeneration);
    } catch {
      await this.finishTerminal(currentSpool, 'failed', {
        errorCode: 'DELIVERY_CONFIGURATION_UNAVAILABLE',
      });
      return;
    }
    const bindingRaw = await WebPushBindingDoc.getNativeCollection().findOne({
      id: currentSpool.bindingId,
      lifecycleGeneration,
      enabled: true,
      status: 'active',
      deletedAt: { $exists: false },
    });
    if (!bindingRaw) {
      await this.finishTerminal(currentSpool, 'cancelled', {
        errorCode: 'BINDING_LIFECYCLE_CHANGED',
      });
      return;
    }
    const binding = asBindingDoc(bindingRaw);
    const vapidKey = binding.vapidKeys.find((keyArg) => (
      keyArg.id === currentSpool.vapidKeyId
      && Boolean(keyArg.privateKeyEnvelope)
      && (keyArg.status === 'active' || keyArg.status === 'retiring')
    ));
    if (
      !vapidKey
      || !vapidKey.privateKeyEnvelope
      || !currentSpool.subscriptionEnvelope
      || !currentSpool.payloadEnvelope
    ) {
      await this.finishTerminal(currentSpool, 'failed', {
        errorCode: 'DELIVERY_CONFIGURATION_UNAVAILABLE',
      });
      return;
    }
    const aadBase = this.spoolAadBase(binding, currentSpool.id);
    let subscription: IWebPushSubscription;
    let payload: IWebPushNotificationPayload;
    let privateKey: string;
    try {
      subscription = this.crypto!.decryptJson<IWebPushSubscription>(
        currentSpool.subscriptionEnvelope,
        { ...aadBase, documentType: 'spool-subscription' },
      );
      payload = this.crypto!.decryptJson<IWebPushNotificationPayload>(
        currentSpool.payloadEnvelope,
        { ...aadBase, documentType: 'spool-payload' },
      );
      privateKey = this.crypto!.decryptJson<{ privateKey: string }>(
        vapidKey.privateKeyEnvelope,
        {
          gatewayClientId: binding.ownerGatewayClientId,
          appInstanceId: binding.ownerAppInstanceId,
          bindingId: binding.id,
          documentType: 'vapid-private-key',
          documentId: vapidKey.id,
        },
      ).privateKey;
      subscription = normalizeSubscription(subscription);
      payload = normalizePayload(payload);
      requireBase64Url(privateKey, 'Web Push VAPID private key', 32);
    } catch {
      await this.finishTerminal(currentSpool, 'failed', {
        errorCode: 'ENCRYPTED_DELIVERY_DATA_INVALID',
      });
      return;
    }
    const immediatelyBeforeSend = await collection.findOne({
      id: currentSpool.id,
      state: 'sending',
      leaseToken,
    });
    if (!immediatelyBeforeSend) return;
    const immediatelyBeforeSendSpool = asSpoolDoc(immediatelyBeforeSend);
    if (immediatelyBeforeSendSpool.cancelRequestedAt !== undefined) {
      await this.finishTerminal(immediatelyBeforeSendSpool, 'cancelled', {
        errorCode: immediatelyBeforeSendSpool.cancelErrorCode || 'CANCELLED_BY_APPLICATION',
      });
      return;
    }
    const bindingImmediatelyBeforeSend = await WebPushBindingDoc.getNativeCollection().findOne({
      id: currentSpool.bindingId,
      lifecycleGeneration,
      enabled: true,
      status: 'active',
      deletedAt: { $exists: false },
    }, { projection: { _id: 1 } });
    if (!bindingImmediatelyBeforeSend) {
      await this.finishTerminal(currentSpool, 'cancelled', {
        errorCode: 'BINDING_LIFECYCLE_CHANGED',
      });
      return;
    }
    const sendStartedAt = this.now();
    if (currentSpool.expiresAt <= sendStartedAt) {
      await this.finishTerminal(currentSpool, 'expired', {
        errorCode: 'DELIVERY_TTL_EXPIRED',
      });
      return;
    }
    const remainingTtlSeconds = Math.max(
      0,
      Math.floor((currentSpool.expiresAt - sendStartedAt) / 1000),
    );
    try {
      const response = await this.transport.send({
        subscription,
        payload: canonicalJson(payload),
        ttlSeconds: remainingTtlSeconds,
        urgency: currentSpool.urgency,
        ...(currentSpool.topic ? { topic: currentSpool.topic } : {}),
        vapid: {
          subject: this.vapidSubject!,
          publicKey: vapidKey.publicKey,
          privateKey,
        },
      });
      if (response.statusCode >= 200 && response.statusCode <= 299) {
        await this.finishTerminal(currentSpool, 'pushServiceAccepted', {
          pushServiceStatusCode: response.statusCode,
        });
        return;
      }
      if (response.statusCode === 404 || response.statusCode === 410) {
        await this.finishTerminal(currentSpool, 'invalidSubscription', {
          pushServiceStatusCode: response.statusCode,
          errorCode: 'PUSH_SUBSCRIPTION_INVALID',
        });
        return;
      }
      if (response.statusCode === 429) {
        await this.deferOrFinish(currentSpool, {
          pushServiceStatusCode: response.statusCode,
          errorCode: 'PUSH_SERVICE_RATE_LIMITED',
          retryAfter: response.retryAfter,
        });
        return;
      }
      if (response.statusCode >= 500 && response.statusCode <= 599) {
        await this.deferOrFinish(currentSpool, {
          pushServiceStatusCode: response.statusCode,
          errorCode: 'PUSH_SERVICE_UNAVAILABLE',
        });
        return;
      }
      await this.finishTerminal(currentSpool, 'failed', {
        pushServiceStatusCode: response.statusCode,
        errorCode: 'PUSH_SERVICE_REJECTED',
      });
    } catch (error: unknown) {
      if (error instanceof WebPushEndpointPolicyError) {
        await this.finishTerminal(currentSpool, 'invalidSubscription', {
          errorCode: error.code,
        });
        return;
      }
      if (error instanceof WebPushTransportError) {
        await this.deferOrFinish(currentSpool, {
          errorCode: error.code,
        });
        return;
      }
      await this.finishTerminal(currentSpool, 'failed', {
        errorCode: 'VAPID_OR_PAYLOAD_CONFIGURATION_ERROR',
      });
    }
  }

  private async finishTerminal(
    spoolArg: WebPushSpoolDoc,
    stateArg: Extract<
      TWebPushDeliveryState,
      'pushServiceAccepted' | 'invalidSubscription' | 'failed' | 'expired' | 'cancelled'
    >,
    metadataArg: {
      pushServiceStatusCode?: number;
      errorCode?: string;
    } = {},
  ): Promise<void> {
    if (!spoolArg.leaseToken) return;
    const now = this.now();
    await WebPushSpoolDoc.getNativeCollection().updateOne(
      {
        id: spoolArg.id,
        state: 'sending',
        leaseToken: spoolArg.leaseToken,
      },
      buildWebPushTerminalUpdate({
        state: stateArg,
        now,
        ...metadataArg,
      }),
    );
  }

  private async deferOrFinish(
    spoolArg: WebPushSpoolDoc,
    metadataArg: {
      pushServiceStatusCode?: number;
      errorCode: string;
      retryAfter?: string;
    },
  ): Promise<void> {
    if (!spoolArg.leaseToken) return;
    const collection = WebPushSpoolDoc.getNativeCollection();
    const currentRaw = await collection.findOne({
      id: spoolArg.id,
      state: 'sending',
      leaseToken: spoolArg.leaseToken,
    });
    if (!currentRaw) return;
    const current = asSpoolDoc(currentRaw);
    if (current.cancelRequestedAt !== undefined) {
      await this.finishTerminal(current, 'cancelled', {
        errorCode: current.cancelErrorCode || 'CANCELLED_BY_APPLICATION',
      });
      return;
    }
    const now = this.now();
    const retryDelay = this.retryDelayMs(current.attempts, metadataArg.retryAfter, now);
    const nextAttemptAt = now + retryDelay;
    if (current.attempts >= DELIVERY_MAX_ATTEMPTS || nextAttemptAt >= current.expiresAt) {
      await this.finishTerminal(
        current,
        nextAttemptAt >= current.expiresAt ? 'expired' : 'failed',
        {
          pushServiceStatusCode: metadataArg.pushServiceStatusCode,
          errorCode: nextAttemptAt >= current.expiresAt
            ? 'DELIVERY_TTL_EXPIRED'
            : 'DELIVERY_ATTEMPTS_EXHAUSTED',
        },
      );
      return;
    }
    await collection.updateOne(
      {
        id: current.id,
        state: 'sending',
        leaseToken: current.leaseToken,
        cancelRequestedAt: { $exists: false },
      },
      {
        $set: {
          state: 'deferred',
          nextAttemptAt,
          updatedAt: now,
          pushServiceStatusCode: metadataArg.pushServiceStatusCode ?? null,
          errorCode: metadataArg.errorCode,
          _updatedAt: new Date(now).toISOString(),
        },
        $unset: {
          leaseToken: '',
          leaseExpiresAt: '',
        },
      },
    );
  }

  private retryDelayMs(
    attemptsArg: number,
    retryAfterArg: string | undefined,
    nowArg: number,
  ): number {
    if (retryAfterArg) {
      const seconds = Number.parseInt(retryAfterArg, 10);
      if (Number.isSafeInteger(seconds) && seconds >= 0) {
        return Math.min(Math.max(seconds * 1000, 1000), 60 * 60 * 1000);
      }
      const date = Date.parse(retryAfterArg);
      if (Number.isFinite(date) && date > nowArg) {
        return Math.min(Math.max(date - nowArg, 1000), 60 * 60 * 1000);
      }
    }
    return Math.min(30_000 * 2 ** Math.max(0, attemptsArg - 1), 60 * 60 * 1000);
  }

  private scheduleWorker(delayArg: number): void {
    if (!this.ready || !this.startWorker) return;
    this.workerTimer = setTimeout(() => {
      this.workerTimer = undefined;
      this.activeWorkerCycle = this.processDueDeliveriesOnce()
        .then(() => undefined)
        .catch((error: unknown) => {
          logger.log('error', `Web Push delivery cycle failed: ${(error as Error).name}`);
        })
        .finally(() => {
          this.activeWorkerCycle = undefined;
          this.scheduleWorker(DELIVERY_WORKER_INTERVAL_MS);
        });
    }, delayArg);
    this.workerTimer.unref();
  }
}