import * as plugins from '../plugins.js';
import { logger } from '../logger.js';
import { CachedEmail } from '../db/index.js';
import type {
  ICachedEmailSmtpTransaction,
  TCachedEmailDirection,
} from '../db/documents/classes.cached.email.js';
import { AcceptEnvelopeRejectionError } from '@push.rocks/smartmta';
import type {
  Email,
  IAcceptedEnvelopeDispatchMetadata,
  IAcceptedEnvelopeRecipientPlan,
  IAcceptEnvelopeContext,
  IExtendedSmtpSession,
  IMessageAcceptanceContext,
  IMessageAcceptanceDecision,
  IResolvedRecipientRoute,
  UnifiedEmailServer,
} from '@push.rocks/smartmta';
import type { DcRouter } from '../classes.dcrouter.js';
import { evaluateInboundAcceptance } from './helpers.inbound-security.js';
export {
  evaluateInboundAcceptance,
  type IInboundAcceptanceEvaluation,
} from './helpers.inbound-security.js';

export const DCROUTER_CACHE_ID_HEADER = 'X-Dcrouter-Cached-Email-Id';
const ACCEPTED_EMAIL_SPOOL_INTERVAL_MS = 60_000;
const ACCEPTED_EMAIL_RETRY_DELAY_MS = 5 * 60_000;
const ACCEPTED_EMAIL_QUEUE_LEASE_MS = 30 * 60_000;
const ACCEPTED_EMAIL_SPOOL_BATCH_SIZE = 25;
const ACCEPTED_EMAIL_STOP_DRAIN_TIMEOUT_MS = 30_000;
/** Retention for catch-all stored inbound mail (30 days). */
const INBOUND_STORE_RETENTION_MS = 30 * 24 * 60 * 60 * 1000;
/** Redispatch attempts for the non-store recipients of a durable envelope. */
const ENVELOPE_DISPATCH_MAX_ATTEMPTS = 10;
const ENVELOPE_DISPATCH_RETRY_DELAY_MS = 5 * 60_000;

/**
 * Permanent per-email storage failure: the raw RFC822 payload of an accepted
 * email is unrecoverable, so delivery can never succeed. The spool marks such
 * rows failed instead of retrying them, so one lost blob cannot wedge the
 * whole queue.
 */
export class AcceptedEmailRawMessageMissingError extends Error {
  constructor(messageArg: string) {
    super(messageArg);
    this.name = 'AcceptedEmailRawMessageMissingError';
  }
}

export type TSmartMtaQueueItemLike = {
  id?: string;
  processingResult?: {
    headers?: Record<string, string>;
    email?: { headers?: Record<string, string> };
  };
  status?: 'pending' | 'processing' | 'queued' | 'delivered' | 'failed' | 'deferred';
  attempts?: number;
  nextAttempt?: Date;
  lastError?: string;
  smtpTransactions?: ICachedEmailSmtpTransaction[];
};

export interface IEmailQueuePersistedEvent {
  cachedEmailId: string;
  status: string;
  reason: string;
}

export interface ISmtpTransactionPersistedEvent {
  cachedEmailId: string;
  transaction: ICachedEmailSmtpTransaction;
}

type TSmartMtaQueueReader = UnifiedEmailServer & {
  getQueueItems?: () => TSmartMtaQueueItemLike[];
};

type TStoredCachedEmailEnvelopeAddress = {
  address: string;
  args?: Record<string, string>;
};

type TStoredCachedEmailSession = {
  id?: string;
  clientHostname?: string;
  remoteAddress?: string;
  secure?: boolean;
  authenticated?: boolean;
  user?: IExtendedSmtpSession['user'];
  envelope?: {
    mailFrom?: TStoredCachedEmailEnvelopeAddress;
    rcptTo?: TStoredCachedEmailEnvelopeAddress[];
  };
};

/**
 * Persisted per-recipient dispatch state for a durably accepted envelope.
 * Holds everything needed to replay `dispatchAcceptedEnvelope` byte-identically:
 * upstream fingerprints each recipient over the raw message, the metadata and
 * the plan entry, so a replay reconstructed from anything else would be refused
 * as an idempotency-key reuse instead of retrying.
 */
type TStoredEnvelopeDispatch = {
  plan: IAcceptedEnvelopeRecipientPlan[];
  metadata: IAcceptedEnvelopeDispatchMetadata;
  attempts: number;
  results?: Array<{
    recipient: string;
    status: string;
    message?: string;
    smtpCode?: number;
  }>;
  pendingRecipients?: string[];
  /** Set once retries are exhausted; the row stops being redispatched. */
  abandonedAt?: string;
};

type TStoredCachedEmailRouteData = {
  acceptance?: string;
  session?: TStoredCachedEmailSession;
  envelopeDispatch?: TStoredEnvelopeDispatch;
  smartMta?: {
    status?: 'queued' | 'deferred' | 'delivered' | 'failed';
    nextAttempt?: string;
    lastError?: string;
    updatedAt?: string;
  };
};

export interface IAcceptRawEmailOptions {
  rawMessage: string | plugins.buffer.Buffer;
  envelope: {
    mailFrom: string;
    rcptTo: string[];
  };
  session: IExtendedSmtpSession;
  messageId?: string;
  subject?: string;
  processAfterAccept?: boolean;
  /** Explicit direction override for programmatic submitters. Defaults to the session-authenticated rule. */
  direction?: TCachedEmailDirection;
  /** Exact authenticated outbound identity and replay metadata. */
  submissionCredentialId?: string;
  submissionIdempotencyKey?: string;
  submissionDigest?: string;
}

/**
 * Accept-then-spool pipeline for inbound SMTP messages: persists accepted
 * messages as CachedEmail docs, replays them through SmartMTA on an interval,
 * and mirrors SmartMTA delivery-queue outcomes back onto the cached docs.
 */
export class AcceptedEmailSpool {
  private spoolTimer?: ReturnType<typeof setInterval> & { unref?: () => void };
  private spoolRun?: Promise<void>;
  private processing = false;
  private stopping = false;
  private queueUpdatePromises = new Set<Promise<void>>();
  private cachedEmailUpdateChains = new Map<string, Promise<void>>();
  private smtpTransactionListeners = new Set<(
    eventArg: ISmtpTransactionPersistedEvent,
  ) => void | Promise<void>>();
  private emailQueuePersistedListeners = new Set<(
    eventArg: IEmailQueuePersistedEvent,
  ) => void | Promise<void>>();

  constructor(private dcRouterRef: DcRouter) {}

  /**
   * Direction is decided by the RECIPIENT, never by whether the session
   * authenticated. Authentication grants permission to relay; it does not make
   * a message addressed to a mailbox we host into outbound mail. A message with
   * at least one locally hosted recipient is inbound — a local mailbox receives
   * it — and only an envelope addressed exclusively to remote recipients is
   * outbound.
   *
   * `IResolvedRecipientRoute.localDomain` is smartmta's own per-recipient
   * verdict (non-null exactly when the domain registry hosts the domain), so
   * the classification uses the same truth the routing decision used. When an
   * acceptance context carries no resolution the envelope recipients are
   * classified against the live registry instead — never against the session,
   * which is the mistake being fixed.
   */
  private deriveDirectionFromResolvedRoutes(
    resolvedRecipientRoutes: readonly IResolvedRecipientRoute[] | undefined,
    envelopeRecipientsArg: readonly string[],
  ): TCachedEmailDirection {
    if (resolvedRecipientRoutes?.length) {
      return resolvedRecipientRoutes.some((resolution) => !!resolution.localDomain)
        ? 'inbound'
        : 'outbound';
    }
    return this.deriveDirectionFromRecipients(envelopeRecipientsArg);
  }

  /**
   * Recipient-derived direction for programmatic submitters, which have no SMTP
   * recipient resolution to consult. Falls back to the live domain registry.
   */
  private deriveDirectionFromRecipients(recipientsArg: readonly string[]): TCachedEmailDirection {
    const domainRegistry = this.dcRouterRef.emailServer?.domainRegistry;
    if (!domainRegistry) {
      // Without the registry there is no recipient truth to classify against.
      // Programmatic submission is relay by construction, so outbound is the
      // honest label rather than guessing from the session.
      return 'outbound';
    }
    for (const recipient of recipientsArg) {
      const domain = recipient.split('@')[1]?.trim().toLowerCase();
      if (domain && domainRegistry.isDomainRegistered(domain)) {
        return 'inbound';
      }
    }
    return 'outbound';
  }

  public async acceptMessage(
    context: IMessageAcceptanceContext,
    processAfterAccept = true,
  ): Promise<IMessageAcceptanceDecision> {
    if (!this.dcRouterRef.dcRouterDb?.isReady()) {
      throw new Error('DcRouterDb is not available for email acceptance');
    }
    this.throwIfMessageAcceptanceAborted(context.abortSignal);

    const rawMessage = context.rawMessage;
    const session = context.session;
    const envelope = session.envelope;
    const envelopeRecipients = Array.isArray(envelope.rcptTo)
      ? envelope.rcptTo.map((recipient) => recipient.address).filter(Boolean)
      : [];
    const email = context.email;
    const headers = email.headers;

    const cachedEmail = CachedEmail.createNew();
    this.removeHeader(email.headers, DCROUTER_CACHE_ID_HEADER);
    email.headers[DCROUTER_CACHE_ID_HEADER] = cachedEmail.id;
    cachedEmail.messageId = headers['Message-ID'] || headers['message-id'] || cachedEmail.id;
    cachedEmail.from = envelope.mailFrom?.address || email.from || '';
    cachedEmail.to = envelopeRecipients.length > 0
      ? envelopeRecipients
      : Array.isArray(email.to) ? email.to : [];
    cachedEmail.cc = Array.isArray(email.cc) ? email.cc : [];
    cachedEmail.bcc = Array.isArray(email.bcc) ? email.bcc : [];
    cachedEmail.subject = email.subject || '';
    const persistedRawMessage = this.setDcRouterCacheIdHeader(rawMessage.toString('utf8'), cachedEmail.id);
    await this.persistRawMessage(cachedEmail, persistedRawMessage);
    cachedEmail.status = 'pending';
    cachedEmail.nextAttempt = new Date();
    cachedEmail.direction = this.deriveDirectionFromResolvedRoutes(
      context.resolvedRecipientRoutes,
      envelopeRecipients.length > 0 ? envelopeRecipients : cachedEmail.to,
    );
    cachedEmail.acceptedAt = Date.now();
    cachedEmail.routeData = JSON.stringify({
      acceptedAt: new Date().toISOString(),
      session: {
        id: session.id,
        remoteAddress: session.remoteAddress,
        clientHostname: session.clientHostname,
        secure: !!session.secure,
        authenticated: !!session.authenticated,
        user: session.user,
        envelope,
      },
    });
    cachedEmail.updateSenderDomain();
    cachedEmail.updateRecipientDomains();
    try {
      await cachedEmail.save();
      await this.notifyEmailQueuePersisted(cachedEmail, 'email-accepted');
    } catch (error) {
      await this.cleanupRawMessageAfterSaveFailure(cachedEmail, error);
      throw error;
    }
    if (context.abortSignal?.aborted) {
      cachedEmail.markFailed('Message acceptance aborted before SMTP success');
      await cachedEmail.save();
      await this.notifyEmailQueuePersisted(cachedEmail, 'email-acceptance-aborted');
      throw new Error('Message acceptance aborted before SMTP success');
    }

    if (processAfterAccept) {
      this.run();
    } else {
      cachedEmail.markDelivered();
      await cachedEmail.save();
      await this.notifyEmailQueuePersisted(cachedEmail, 'email-accepted-without-queue');
    }

    this.trackAcceptedInboundEmail(cachedEmail);
    return {
      accepted: true,
      smtpCode: 250,
      smtpMessage: '2.0.0 Message accepted for delivery',
      continueProcessing: false,
    };
  }

  public async acceptRawMessage(optionsArg: IAcceptRawEmailOptions): Promise<IMessageAcceptanceDecision & { spoolItemId: string }> {
    if (!this.dcRouterRef.dcRouterDb?.isReady()) {
      throw new Error('DcRouterDb is not available for email acceptance');
    }

    const rawMessage = plugins.buffer.Buffer.isBuffer(optionsArg.rawMessage)
      ? optionsArg.rawMessage.toString('utf8')
      : optionsArg.rawMessage;
    const cachedEmail = CachedEmail.createNew();
    cachedEmail.messageId = optionsArg.messageId || this.extractHeader(rawMessage, 'message-id') || cachedEmail.id;
    cachedEmail.from = optionsArg.envelope.mailFrom;
    cachedEmail.to = optionsArg.envelope.rcptTo;
    cachedEmail.cc = [];
    cachedEmail.bcc = [];
    cachedEmail.subject = optionsArg.subject || this.extractHeader(rawMessage, 'subject') || '';
    cachedEmail.submissionCredentialId = optionsArg.submissionCredentialId;
    cachedEmail.submissionIdempotencyKey = optionsArg.submissionIdempotencyKey;
    cachedEmail.submissionDigest = optionsArg.submissionDigest;
    const persistedRawMessage = this.setDcRouterCacheIdHeader(rawMessage, cachedEmail.id);
    await this.persistRawMessage(cachedEmail, persistedRawMessage);
    cachedEmail.status = 'pending';
    cachedEmail.nextAttempt = new Date();
    cachedEmail.direction = optionsArg.direction
      ?? this.deriveDirectionFromRecipients(optionsArg.envelope.rcptTo);
    cachedEmail.acceptedAt = Date.now();
    cachedEmail.routeData = JSON.stringify({
      acceptedAt: new Date().toISOString(),
      session: {
        id: optionsArg.session.id,
        remoteAddress: optionsArg.session.remoteAddress,
        clientHostname: optionsArg.session.clientHostname,
        secure: !!optionsArg.session.secure,
        authenticated: !!optionsArg.session.authenticated,
        user: optionsArg.session.user,
        envelope: optionsArg.session.envelope,
      },
    });
    cachedEmail.updateSenderDomain();
    cachedEmail.updateRecipientDomains();
    try {
      await cachedEmail.save();
      await this.notifyEmailQueuePersisted(cachedEmail, 'raw-email-accepted');
    } catch (error) {
      await this.cleanupRawMessageAfterSaveFailure(cachedEmail, error);
      throw error;
    }

    if (optionsArg.processAfterAccept !== false) {
      this.run();
    } else {
      cachedEmail.markDelivered();
      await cachedEmail.save();
      await this.notifyEmailQueuePersisted(cachedEmail, 'raw-email-accepted-without-queue');
    }

    this.trackAcceptedInboundEmail(cachedEmail);
    return {
      accepted: true,
      spoolItemId: cachedEmail.id,
      smtpCode: 250,
      smtpMessage: '2.0.0 Message accepted for delivery',
      continueProcessing: false,
    };
  }

  /**
   * Durable acceptance for envelopes containing at least one catch-all store
   * route: anchors the exact raw message + metadata as a CachedEmail before
   * SMTP 250, then dispatches any non-store recipients through SmartMTA's
   * exact-plan executor. Store recipients are fulfilled by the anchor itself.
   */
  public async acceptEnvelope(
    context: IAcceptEnvelopeContext,
    emailServer: UnifiedEmailServer,
  ): Promise<void> {
    if (!this.dcRouterRef.dcRouterDb?.isReady()) {
      throw new Error('DcRouterDb is not available for durable envelope acceptance');
    }
    this.throwIfMessageAcceptanceAborted(context.abortSignal);

    const email = context.email;
    const session = context.session;
    const cachedEmail = CachedEmail.createNew();
    this.removeHeader(email.headers, DCROUTER_CACHE_ID_HEADER);
    email.headers[DCROUTER_CACHE_ID_HEADER] = cachedEmail.id;
    cachedEmail.messageId = email.headers['Message-ID'] || email.headers['message-id'] || cachedEmail.id;
    cachedEmail.from = context.envelope.mailFrom || email.from || '';
    cachedEmail.to = context.envelope.rcptTo.length > 0
      ? [...context.envelope.rcptTo]
      : Array.isArray(email.to) ? email.to : [];
    cachedEmail.cc = Array.isArray(email.cc) ? email.cc : [];
    cachedEmail.bcc = Array.isArray(email.bcc) ? email.bcc : [];
    cachedEmail.subject = email.subject || '';
    cachedEmail.attachmentCount = context.attachmentCount;
    if (context.securityResults) {
      cachedEmail.inboundSecurityResults = JSON.stringify(context.securityResults);
    }

    // Build the exact per-recipient dispatch plan before durably committing,
    // so an unplannable envelope is refused instead of half-anchored.
    const idempotencyKeys = context.resolvedRecipientRoutes.map(
      (resolution) => `${cachedEmail.id}:${resolution.recipient}`,
    );
    const plan = emailServer.createAcceptedEnvelopeDispatchPlan(context, idempotencyKeys);
    const nonStoreEntries = plan.filter((entry) => entry.action.type !== 'store');

    cachedEmail.direction = this.deriveDirectionFromResolvedRoutes(
      context.resolvedRecipientRoutes,
      context.envelope.rcptTo,
    );
    // Submission/relay authentication is not a receiver verdict. Only locally
    // received mail is classified or enforced here; outbound mail gets its
    // post-signing, actual-egress self-check from the delivery transaction.
    const acceptance = evaluateInboundAcceptance(
      cachedEmail.direction === 'inbound' ? context.securityResults : null,
    );
    if (cachedEmail.direction === 'inbound' && context.securityResults) {
      cachedEmail.inboundSecurityDisposition = acceptance.dmarcReject
        ? 'rejected'
        : acceptance.status;
    }
    const startedAtMs = Date.now();

    const persistedRawMessage = this.setDcRouterCacheIdHeader(
      context.rawMessage.toString('utf8'),
      cachedEmail.id,
    );
    await this.persistRawMessage(cachedEmail, persistedRawMessage);
    cachedEmail.acceptedAt = startedAtMs;

    if (acceptance.dmarcReject) {
      // Enforce the sender domain's published DMARC reject policy: persist the
      // rejected attempt for the email log, then refuse the envelope with a
      // permanent SMTP failure.
      const rejectionMessage = `5.7.1 Rejected by DMARC policy of ${acceptance.dmarcDomain || 'the sender domain'}`;
      cachedEmail.status = 'rejected';
      cachedEmail.lastError = rejectionMessage;
      cachedEmail.setTTL(INBOUND_STORE_RETENTION_MS);
      cachedEmail.appendSmtpTransaction(this.buildInboundTransaction(context, cachedEmail.id, {
        outcome: 'failed',
        smtpCode: 550,
        finalLine: rejectionMessage,
        doubts: acceptance.doubts,
        startedAtMs,
      }));
      cachedEmail.routeData = JSON.stringify({
        acceptedAt: new Date(startedAtMs).toISOString(),
        acceptance: 'dmarc-policy-reject',
        doubts: acceptance.doubts,
      });
      cachedEmail.updateSenderDomain();
      cachedEmail.updateRecipientDomains();
      await cachedEmail.save().catch(() => undefined);
      await this.notifyEmailQueuePersisted(cachedEmail, 'envelope-rejected-dmarc').catch(() => undefined);
      throw new AcceptEnvelopeRejectionError(550, rejectionMessage);
    }

    // Security disposition is orthogonal to the delivery lifecycle. The Ops
    // API synthesizes the visible flagged badge from inboundSecurityDisposition.
    cachedEmail.status = 'accepted';
    if (nonStoreEntries.length > 0) {
      // Relay recipients are dispatched after this row is committed. Park the
      // row in a status the spool actually scans so a crash between the commit
      // and the dispatch leaves those recipients recoverable instead of stranded
      // in a terminal-looking 'accepted' row. The dispatch outcome settles it
      // straight back to the acceptance status.
      cachedEmail.status = 'deferred';
      cachedEmail.nextAttempt = new Date(Date.now() + ENVELOPE_DISPATCH_RETRY_DELAY_MS);
    }
    cachedEmail.deliveredAt = new Date();
    cachedEmail.setTTL(INBOUND_STORE_RETENTION_MS);
    cachedEmail.appendSmtpTransaction(this.buildInboundTransaction(context, cachedEmail.id, {
      outcome: 'succeeded',
      smtpCode: 250,
      finalLine: '250 2.0.0 Message accepted for delivery',
      doubts: acceptance.doubts,
      startedAtMs,
    }));
    const dispatchMetadata = this.buildEnvelopeDispatchMetadata(context);
    cachedEmail.routeData = JSON.stringify({
      acceptedAt: new Date(startedAtMs).toISOString(),
      acceptance: 'durable-envelope',
      verdictDoubts: acceptance.doubts,
      recipientPlans: plan.map((entry) => ({
        recipient: entry.recipient,
        routeName: entry.routeName,
        source: entry.source,
        actionType: entry.action.type,
      })),
      // Persisted BEFORE the SMTP 250, so a dispatch failure for the non-store
      // recipients is recoverable instead of being lost with the process.
      ...(nonStoreEntries.length > 0
        ? {
          envelopeDispatch: {
            plan: [...plan],
            metadata: dispatchMetadata,
            attempts: 0,
            pendingRecipients: nonStoreEntries.map((entry) => entry.recipient),
          } satisfies TStoredEnvelopeDispatch,
        }
        : {}),
      session: {
        id: session.id,
        remoteAddress: session.remoteAddress,
        clientHostname: session.clientHostname,
        secure: !!session.secure,
        authenticated: !!session.authenticated,
        user: session.user,
        envelope: context.envelope,
      },
    });
    cachedEmail.updateSenderDomain();
    cachedEmail.updateRecipientDomains();
    try {
      await cachedEmail.save();
      await this.notifyEmailQueuePersisted(cachedEmail, 'envelope-stored');
    } catch (error) {
      await this.deleteRawMessage(cachedEmail).catch(() => undefined);
      throw error;
    }
    if (context.abortSignal?.aborted) {
      cachedEmail.markFailed('Envelope acceptance aborted before SMTP success');
      await cachedEmail.save();
      await this.notifyEmailQueuePersisted(cachedEmail, 'envelope-acceptance-aborted');
      throw new Error('Envelope acceptance aborted before SMTP success');
    }

    if (nonStoreEntries.length > 0) {
      await this.dispatchEnvelopeRecipients(cachedEmail, persistedRawMessage, emailServer);
    }
    this.trackAcceptedInboundEmail(cachedEmail);
  }

  /**
   * Session metadata for `dispatchAcceptedEnvelope`. It feeds the per-recipient
   * idempotency fingerprint, so it must be JSON-round-trip stable: a replay
   * reconstructed from persisted state has to produce byte-identical metadata
   * or upstream refuses it as an idempotency-key reuse.
   */
  private buildEnvelopeDispatchMetadata(
    context: IAcceptEnvelopeContext,
  ): IAcceptedEnvelopeDispatchMetadata {
    const session = context.session;
    return {
      mailFrom: context.envelope.mailFrom,
      session: {
        id: session.id || '',
        remoteAddress: session.remoteAddress || '',
        clientHostname: session.clientHostname || '',
        secure: !!session.secure,
        authenticated: !!session.authenticated,
      },
    };
  }

  /**
   * Dispatch (or redispatch) the non-store recipients of a durably accepted
   * envelope.
   *
   * The raw bytes handed to upstream are the exact bytes persisted for this row,
   * because upstream fingerprints every recipient over the raw message: a later
   * replay with different bytes would be rejected as an idempotency-key reuse
   * rather than retried. Recipients that already succeeded are short-circuited
   * by upstream's checkpoints, and `failed` results are deliberately not
   * checkpointed upstream, so an identical replay retries exactly those.
   *
   * A `failed` recipient leaves the row non-terminal so the spool retries it;
   * a `rejected` recipient is a permanent per-recipient refusal and is recorded
   * durably instead of being retried. Either way the outcome is persisted — a
   * relay recipient is never silently dropped after the SMTP 250.
   */
  private async dispatchEnvelopeRecipients(
    cachedEmailArg: CachedEmail,
    rawMessageArg: string,
    emailServerArg: UnifiedEmailServer,
  ): Promise<void> {
    const routeData = this.parseCachedEmailRouteData(cachedEmailArg);
    const envelopeDispatch = routeData.envelopeDispatch;
    if (!envelopeDispatch || envelopeDispatch.abandonedAt) return;

    const attempts = (envelopeDispatch.attempts || 0) + 1;
    let dispatchResults: Array<{
      recipient: string;
      status: string;
      message?: string;
      smtpCode?: number;
    }>;
    try {
      const dispatchResult = await emailServerArg.dispatchAcceptedEnvelope(
        plugins.buffer.Buffer.from(rawMessageArg, 'utf8'),
        envelopeDispatch.plan,
        envelopeDispatch.metadata,
      );
      dispatchResults = dispatchResult.results.map((result) => ({
        recipient: result.recipient,
        status: result.status,
        ...(result.message ? { message: result.message } : {}),
        ...(result.smtpCode !== undefined ? { smtpCode: result.smtpCode } : {}),
      }));
    } catch (error: unknown) {
      // A whole-call failure (missing managed queue storage, malformed
      // checkpoint) is retried the same way a per-recipient failure is.
      dispatchResults = envelopeDispatch.plan
        .filter((entry) => entry.action.type !== 'store' && entry.action.type !== 'deliver')
        .map((entry) => ({
          recipient: entry.recipient,
          status: 'failed',
          message: (error as Error).message,
        }));
    }

    const retryable = dispatchResults.filter((result) => result.status === 'failed');
    const rejected = dispatchResults.filter((result) => result.status === 'rejected');
    const exhausted = retryable.length > 0 && attempts >= ENVELOPE_DISPATCH_MAX_ATTEMPTS;

    await this.persistEnvelopeDispatchOutcome(cachedEmailArg.id, {
      attempts,
      results: dispatchResults,
      pendingRecipients: exhausted ? [] : retryable.map((result) => result.recipient),
      abandoned: exhausted,
    });

    if (rejected.length > 0) {
      logger.log('error', `Durable envelope ${cachedEmailArg.id}: ${rejected.length} relay recipient(s) permanently refused: ${rejected.map((result) => `${result.recipient}=${result.smtpCode || 550} ${result.message || 'rejected'}`).join('; ')}`);
    }
    if (exhausted) {
      logger.log('error', `Durable envelope ${cachedEmailArg.id}: giving up on ${retryable.length} relay recipient(s) after ${attempts} dispatch attempts: ${retryable.map((result) => `${result.recipient}=${result.message || 'failed'}`).join('; ')}`);
    } else if (retryable.length > 0) {
      logger.log('warn', `Durable envelope ${cachedEmailArg.id}: ${retryable.length} relay recipient(s) failed dispatch (attempt ${attempts}/${ENVELOPE_DISPATCH_MAX_ATTEMPTS}), scheduled for redispatch: ${retryable.map((result) => `${result.recipient}=${result.message || 'failed'}`).join('; ')}`);
    }
  }

  /**
   * Persist a dispatch attempt's outcome on the durable-envelope row.
   *
   * The row's status tracks the stored envelope, not the relay: it goes
   * `deferred` only while relay recipients still need a redispatch, and returns
   * to its acceptance status once none do. Relay progress itself lives in
   * `routeData.envelopeDispatch`, so a relay failure can never mark a row whose
   * local copy stored successfully as failed.
   */
  private async persistEnvelopeDispatchOutcome(
    cachedEmailIdArg: string,
    outcomeArg: {
      attempts: number;
      results: Array<{ recipient: string; status: string; message?: string; smtpCode?: number }>;
      pendingRecipients: string[];
      abandoned: boolean;
    },
  ): Promise<void> {
    await this.runCachedEmailUpdate(cachedEmailIdArg, async () => {
      const cachedEmail = await CachedEmail.findById(cachedEmailIdArg);
      if (!cachedEmail) return;
      const routeData = this.parseCachedEmailRouteData(cachedEmail);
      if (!routeData.envelopeDispatch) return;
      routeData.envelopeDispatch = {
        ...routeData.envelopeDispatch,
        attempts: outcomeArg.attempts,
        results: outcomeArg.results,
        pendingRecipients: outcomeArg.pendingRecipients,
        ...(outcomeArg.abandoned ? { abandonedAt: new Date().toISOString() } : {}),
      };
      cachedEmail.routeData = JSON.stringify(routeData);

      const stillPending = outcomeArg.pendingRecipients.length > 0;
      if (stillPending) {
        cachedEmail.status = 'deferred';
        cachedEmail.nextAttempt = new Date(Date.now() + ENVELOPE_DISPATCH_RETRY_DELAY_MS);
      } else if (cachedEmail.status === 'deferred') {
        cachedEmail.status = 'accepted';
        cachedEmail.nextAttempt = new Date();
      }
      await cachedEmail.save();
      await this.notifyEmailQueuePersisted(
        cachedEmail,
        stillPending ? 'envelope-dispatch-deferred' : 'envelope-dispatch-settled',
      );
    });
  }

  /**
   * Take exclusive ownership of a durably accepted envelope row in the spool.
   *
   * Runs before live-queue postponement and before the normal spool handoff: a
   * relay sibling that did enqueue would otherwise postpone this row forever,
   * and the normal handoff would re-run route evaluation and duplicate a
   * delivery the stored copy already fulfilled.
   */
  private async handleDurableEnvelopeRow(
    cachedEmailArg: CachedEmail,
    emailServerArg: UnifiedEmailServer,
  ): Promise<boolean> {
    if (!this.isDurableEnvelopeRow(cachedEmailArg)) return false;
    const envelopeDispatch = this.parseCachedEmailRouteData(cachedEmailArg).envelopeDispatch;

    if (envelopeDispatch?.pendingRecipients?.length && !envelopeDispatch.abandonedAt) {
      const rawMessage = await this.readRawMessage(cachedEmailArg);
      await this.dispatchEnvelopeRecipients(
        cachedEmailArg,
        rawMessage.toString('utf8'),
        emailServerArg,
      );
      return true;
    }

    // Nothing left to dispatch: settle the row back onto its acceptance status
    // so a fully dispatched envelope does not linger as deferred.
    if (cachedEmailArg.status === 'deferred') {
      cachedEmailArg.status = 'accepted';
      cachedEmailArg.nextAttempt = new Date();
      await cachedEmailArg.save();
      await this.notifyEmailQueuePersisted(cachedEmailArg, 'envelope-dispatch-settled');
    }
    return true;
  }

  private trackAcceptedInboundEmail(cachedEmailArg: CachedEmail): void {
    if (cachedEmailArg.direction !== 'inbound') return;
    this.dcRouterRef.metricsManager?.trackEmailReceived(cachedEmailArg.from);
  }

  /**
   * Synthesizes the inbound SMTP transaction for the email log from the
   * acceptance context: the receiving session where WE are the server. The
   * Rust frontend does not export a raw line transcript yet, so entries are
   * reconstructed from the envelope, session, and verdict data.
   */
  private buildInboundTransaction(
    context: IAcceptEnvelopeContext,
    cachedEmailId: string,
    resultArg: {
      outcome: 'succeeded' | 'failed';
      smtpCode: number;
      finalLine: string;
      doubts: string[];
      startedAtMs: number;
    },
  ): ICachedEmailSmtpTransaction {
    const session = context.session;
    const nowMs = Date.now();
    const hostname = this.dcRouterRef.options?.emailConfig?.hostname || 'mail-gateway';
    const transcript: Array<{
      timestampMs: number;
      direction: 'client' | 'server' | 'system';
      phase: string;
      text: string;
      responseCode?: number;
    }> = [];
    let timestampMs = resultArg.startedAtMs;
    const push = (direction: 'client' | 'server' | 'system', phase: string, text: string, responseCode?: number) => {
      transcript.push({ timestampMs, direction, phase, text, ...(responseCode !== undefined ? { responseCode } : {}) });
      timestampMs += 1;
    };

    push('system', 'connect', `Connection from ${session.remoteAddress || 'unknown'}${session.clientHostname ? ` (${session.clientHostname})` : ''}${session.secure ? ', TLS' : ', plaintext'}${session.authenticated ? ', authenticated' : ''}`);
    push('client', 'mail_from', `MAIL FROM:<${context.envelope.mailFrom}>`);
    push('server', 'mail_from', '250 OK', 250);
    for (const recipient of context.envelope.rcptTo) {
      push('client', 'rcpt_to', `RCPT TO:<${recipient}>`);
      push('server', 'rcpt_to', '250 OK', 250);
    }
    push('client', 'data_command', 'DATA');
    push('server', 'data_command', '354 Start mail input', 354);
    push('system', 'message_body', `Message received: ${context.rawMessage.length} bytes, ${context.attachmentCount} attachment(s)`);
    const spf = (context.securityResults as any)?.spf;
    const dkimSignatures = Array.isArray((context.securityResults as any)?.dkim) ? (context.securityResults as any).dkim : [];
    const dmarc = (context.securityResults as any)?.dmarc;
    if (spf) push('system', 'message_body', `SPF: ${spf.result || 'unknown'} (${spf.domain || 'unknown domain'})`);
    if (dkimSignatures.length > 0) {
      const valid = dkimSignatures.filter((signature: any) => signature?.is_valid).length;
      push('system', 'message_body', `DKIM: ${valid}/${dkimSignatures.length} signature(s) verified${dkimSignatures[0]?.domain ? ` (${dkimSignatures[0].domain})` : ''}`);
    }
    if (dmarc) push('system', 'message_body', `DMARC: ${dmarc.passed ? 'pass' : 'fail'} (policy ${dmarc.policy || 'unknown'})`);
    for (const doubt of resultArg.doubts) {
      push('system', 'message_body', `Verdict doubt: ${doubt}`);
    }
    push('server', 'final_response', resultArg.finalLine, resultArg.smtpCode);

    return {
      id: `inbound-${cachedEmailId}`,
      queueItemId: cachedEmailId,
      queueAttempt: 1,
      targetHost: hostname,
      targetPort: 25,
      recipients: [...context.envelope.rcptTo],
      startedAt: new Date(resultArg.startedAtMs).toISOString(),
      completedAt: new Date(nowMs).toISOString(),
      durationMs: Math.max(0, nowMs - resultArg.startedAtMs),
      outcome: resultArg.outcome,
      ...(resultArg.outcome === 'failed' ? { retryable: false, error: resultArg.finalLine, errorType: 'policy' } : {}),
      smtpCode: resultArg.smtpCode,
      transcript,
    } as unknown as ICachedEmailSmtpTransaction;
  }

  /** Start the interval-driven spool processor and trigger an immediate run. */
  public start(): void {
    this.clearSpoolTimer();
    this.stopping = false;
    const runProcessor = () => {
      this.run();
    };
    this.spoolTimer = setInterval(
      runProcessor,
      ACCEPTED_EMAIL_SPOOL_INTERVAL_MS,
    ) as ReturnType<typeof setInterval> & { unref?: () => void };
    this.spoolTimer.unref?.();
    runProcessor();
  }

  /** Mark the spool as stopping and clear the interval without awaiting in-flight work. */
  public beginStop(): void {
    this.stopping = true;
    this.clearSpoolTimer();
  }

  /** Stop the spool and wait (bounded) for an in-flight run to settle. */
  public async stop(): Promise<void> {
    this.beginStop();
    const spoolRun = this.spoolRun;
    if (spoolRun) {
      const settled = await this.waitForPromiseToSettleWithTimeout(
        spoolRun,
        ACCEPTED_EMAIL_STOP_DRAIN_TIMEOUT_MS,
      );
      if (!settled) {
        logger.log('warn', 'Timed out waiting for accepted email spool processing to stop');
      }
    }
  }

  /** Kick off a spool run unless one is already in flight. */
  public run(): void {
    if (this.spoolRun) {
      return;
    }
    const run = this.processSpool().catch((error) => {
      logger.log('warn', `Accepted email spool processing failed: ${(error as Error).message}`);
    });
    this.spoolRun = run;
    void run.finally(() => {
      if (this.spoolRun === run) {
        this.spoolRun = undefined;
      }
    });
  }

  public trackQueueUpdate(
    item: TSmartMtaQueueItemLike,
    status: 'queued' | 'deferred' | 'delivered' | 'failed',
    failureMessage: string,
  ): Promise<void> {
    const cachedEmailId = this.getCachedEmailIdFromQueueItem(item);
    const updatePromise = this.runCachedEmailUpdate(cachedEmailId, async () => {
      await this.updateAcceptedEmailFromQueueItem(item, status);
    }).catch((error) => {
      logger.log('warn', `${failureMessage}: ${(error as Error).message}`);
    });
    this.trackUpdatePromise(updatePromise);
    return updatePromise;
  }

  public trackSmtpTransaction(
    transactionArg: ICachedEmailSmtpTransaction,
    emailServerArg: UnifiedEmailServer,
  ): Promise<void> {
    const queueItem = emailServerArg.getQueueItem(
      transactionArg.queueItemId,
    ) as TSmartMtaQueueItemLike | undefined;
    const cachedEmailId = queueItem ? this.getCachedEmailIdFromQueueItem(queueItem) : undefined;
    const updatePromise = this.runCachedEmailUpdate(cachedEmailId, async () => {
      if (!cachedEmailId || !this.dcRouterRef.dcRouterDb?.isReady()) return;
      const cachedEmail = await CachedEmail.findById(cachedEmailId);
      if (!cachedEmail) return;
      this.appendSmtpTransaction(cachedEmail, transactionArg);
      await cachedEmail.save();
      const event: ISmtpTransactionPersistedEvent = {
        cachedEmailId,
        transaction: structuredClone(transactionArg),
      };
      await Promise.allSettled(
        [...this.smtpTransactionListeners].map(async (listener) => await listener(event)),
      );
    }).catch((error) => {
      logger.log('warn', `Unable to persist SMTP transaction: ${(error as Error).message}`);
    });
    this.trackUpdatePromise(updatePromise);
    return updatePromise;
  }

  /** Subscribe to updates only after the SMTP transaction is durably stored. */
  public onSmtpTransactionPersisted(
    listenerArg: (eventArg: ISmtpTransactionPersistedEvent) => void | Promise<void>,
  ): () => void {
    this.smtpTransactionListeners.add(listenerArg);
    return () => this.smtpTransactionListeners.delete(listenerArg);
  }
  /** Subscribe to queue state changes only after the CachedEmail row is durable. */
  public onEmailQueuePersisted(
    listenerArg: (eventArg: IEmailQueuePersistedEvent) => void | Promise<void>,
  ): () => void {
    this.emailQueuePersistedListeners.add(listenerArg);
    return () => this.emailQueuePersistedListeners.delete(listenerArg);
  }

  private async notifyEmailQueuePersisted(
    cachedEmailArg: CachedEmail,
    reasonArg: string,
  ): Promise<void> {
    if (this.emailQueuePersistedListeners.size === 0) return;
    const event: IEmailQueuePersistedEvent = {
      cachedEmailId: cachedEmailArg.id,
      status: cachedEmailArg.status,
      reason: reasonArg,
    };
    await Promise.allSettled(
      [...this.emailQueuePersistedListeners].map(async (listener) => await listener(event)),
    );
  }

  public async drainQueueUpdates(): Promise<void> {
    const queueUpdates = [...this.queueUpdatePromises];
    if (queueUpdates.length === 0) {
      return;
    }
    const settled = await this.waitForPromiseToSettleWithTimeout(
      Promise.allSettled(queueUpdates).then(() => undefined),
      ACCEPTED_EMAIL_STOP_DRAIN_TIMEOUT_MS,
    );
    if (!settled) {
      for (const queueUpdate of queueUpdates) {
        this.queueUpdatePromises.delete(queueUpdate);
      }
      logger.log('warn', `Timed out waiting for ${queueUpdates.length} accepted email queue update(s) to settle`);
    }
  }

  /** Reconcile persisted queue traces after restart before queue recovery runs. */
  public async recoverSmtpTransactionHistory(emailServerArg: UnifiedEmailServer): Promise<void> {
    for (const item of (emailServerArg as TSmartMtaQueueReader).getQueueItems?.() || []) {
      const cachedEmailId = this.getCachedEmailIdFromQueueItem(item);
      if (!cachedEmailId || !item.smtpTransactions?.length) continue;
      try {
        await this.runCachedEmailUpdate(cachedEmailId, async () => {
          const cachedEmail = await CachedEmail.findById(cachedEmailId);
          if (!cachedEmail) return;
          for (const transaction of item.smtpTransactions || []) {
            this.appendSmtpTransaction(cachedEmail, transaction);
          }
          await cachedEmail.save();
        });
      } catch (error: unknown) {
        // History reconciliation runs during email-server startup; one row that
        // cannot be persisted must not abort reconciliation or startup.
        logger.log('warn', `Unable to reconcile SMTP transaction history for accepted email ${cachedEmailId}: ${(error as Error).message}`);
      }
    }
  }

  /** Requeue emails left in 'queued' state by a previous process as pending. */
  public async recoverQueuedEmails(): Promise<void> {
    // Rows whose recovery save failed stay 'queued' and are re-served by
    // findQueuedForRecovery; track them so they are neither retried in a loop
    // within this run nor able to wedge recovery when they fill a whole batch.
    const unrecoverableIds = new Set<string>();
    while (true) {
      const queuedEmails = await CachedEmail.findQueuedForRecovery(ACCEPTED_EMAIL_SPOOL_BATCH_SIZE);
      const recoverableEmails = queuedEmails.filter((queuedEmail) => !unrecoverableIds.has(queuedEmail.id));
      if (recoverableEmails.length === 0) {
        return;
      }
      for (const queuedEmail of recoverableEmails) {
        if (this.isCachedEmailTerminal(queuedEmail)) {
          continue;
        }
        const previousStatus = queuedEmail.status;
        const previousNextAttempt = queuedEmail.nextAttempt;
        try {
          queuedEmail.status = 'pending';
          queuedEmail.nextAttempt = new Date();
          await queuedEmail.save();
          await this.notifyEmailQueuePersisted(queuedEmail, 'email-queue-recovered');
        } catch (error: unknown) {
          // Recovery runs during email-server startup; a row that cannot be
          // persisted stays 'queued' for the next startup and must not abort
          // recovery of the remaining rows.
          queuedEmail.status = previousStatus;
          queuedEmail.nextAttempt = previousNextAttempt;
          unrecoverableIds.add(queuedEmail.id);
          logger.log('warn', `Unable to recover queued accepted email ${queuedEmail.id}: ${(error as Error).message}`);
        }
      }
      if (queuedEmails.length < ACCEPTED_EMAIL_SPOOL_BATCH_SIZE) {
        return;
      }
    }
  }

  private async processSpool(): Promise<void> {
    const emailServer = this.dcRouterRef.emailServer;
    if (this.processing || !emailServer || !this.dcRouterRef.dcRouterDb?.isReady()) {
      return;
    }
    this.processing = true;
    try {
      const cachedEmails = await CachedEmail.findPendingForDelivery(ACCEPTED_EMAIL_SPOOL_BATCH_SIZE);
      for (const cachedEmail of cachedEmails) {
        if (this.stopping || this.dcRouterRef.emailServer !== emailServer) {
          break;
        }
        try {
          // Durable envelopes are owned by the accepted-envelope dispatch path,
          // never by the route-evaluating handoff below.
          if (await this.handleDurableEnvelopeRow(cachedEmail, emailServer)) {
            continue;
          }
          if (await this.postponeLiveSmartMtaOwnedEmail(cachedEmail, emailServer)) {
            continue;
          }
          const session = this.buildCachedEmailSession(cachedEmail);
          const rawMessage = await this.readRawMessage(cachedEmail);
          await this.processAcceptedCachedEmail(cachedEmail, rawMessage, session, emailServer);
        } catch (error: unknown) {
          // One poisoned row must never stall the rest of the batch: record the
          // failure on that row alone and keep processing the remaining emails.
          await this.handleSpoolItemFailure(cachedEmail, error);
        }
      }
    } finally {
      this.processing = false;
    }
  }

  /**
   * Record a per-email spool failure without aborting the batch: permanent
   * raw-message losses are marked failed, everything else is deferred through
   * the regular retry mechanism (which terminates at maxAttempts).
   */
  private async handleSpoolItemFailure(cachedEmail: CachedEmail, errorArg: unknown): Promise<void> {
    const message = (errorArg as Error).message;
    try {
      const currentCachedEmail = await CachedEmail.findById(cachedEmail.id) || cachedEmail;
      if (this.isCachedEmailTerminal(currentCachedEmail)) {
        return;
      }
      if (errorArg instanceof AcceptedEmailRawMessageMissingError) {
        currentCachedEmail.markFailed(message);
        await currentCachedEmail.save();
        await this.notifyEmailQueuePersisted(currentCachedEmail, 'email-raw-message-missing');
        logger.log('error', `Accepted email ${currentCachedEmail.id} failed permanently: ${message}`);
        return;
      }
      currentCachedEmail.scheduleRetry(ACCEPTED_EMAIL_RETRY_DELAY_MS);
      currentCachedEmail.lastError = message;
      await currentCachedEmail.save();
      await this.notifyEmailQueuePersisted(currentCachedEmail, 'email-spool-deferred');
      logger.log('warn', `Accepted email ${currentCachedEmail.id} deferred after spool processing failure: ${message}`);
    } catch (persistError: unknown) {
      // Even failing to persist the failure state must not abort the batch.
      logger.log('error', `Unable to persist spool failure state for accepted email ${cachedEmail.id}: ${(persistError as Error).message} (original error: ${message})`);
    }
  }

  private async processAcceptedCachedEmail(
    cachedEmail: CachedEmail,
    emailData: Email | plugins.buffer.Buffer,
    session: IExtendedSmtpSession,
    emailServer: UnifiedEmailServer,
  ): Promise<void> {
    cachedEmail.status = 'processing';
    cachedEmail.nextAttempt = new Date(Date.now() + ACCEPTED_EMAIL_QUEUE_LEASE_MS);
    await cachedEmail.save();
    await this.notifyEmailQueuePersisted(cachedEmail, 'email-processing');

    try {
      if (plugins.buffer.Buffer.isBuffer(emailData)) {
        const handledByTypedEndpoint = await this.dcRouterRef.workAppMailManager?.deliverCachedEmailToTypedEndpoint(
          cachedEmail,
          emailData,
          session,
        );
        if (handledByTypedEndpoint) {
          cachedEmail.markDelivered();
          await cachedEmail.save();
          await this.notifyEmailQueuePersisted(cachedEmail, 'email-delivered-to-typed-endpoint');
          return;
        }
      }

      await emailServer.processEmailByMode(emailData, session);
      if (session.matchedRoute?.action.type === 'forward') {
        const currentCachedEmail = await CachedEmail.findById(cachedEmail.id) || cachedEmail;
        currentCachedEmail.markDelivered();
        await currentCachedEmail.save();
        await this.notifyEmailQueuePersisted(currentCachedEmail, 'email-forwarded');
        return;
      } else {
        const currentCachedEmail = await CachedEmail.findById(cachedEmail.id) || cachedEmail;
        if (this.isCachedEmailTerminal(currentCachedEmail) || currentCachedEmail.status === 'deferred') {
          return;
        }
        currentCachedEmail.status = 'queued';
        currentCachedEmail.nextAttempt = new Date(Date.now() + ACCEPTED_EMAIL_QUEUE_LEASE_MS);
        await currentCachedEmail.save();
        await this.notifyEmailQueuePersisted(currentCachedEmail, 'email-queued');
        return;
      }
    } catch (error: unknown) {
      cachedEmail.scheduleRetry(ACCEPTED_EMAIL_RETRY_DELAY_MS);
      cachedEmail.lastError = (error as Error).message;
      await cachedEmail.save();
      await this.notifyEmailQueuePersisted(cachedEmail, 'email-spool-deferred');
      logger.log('warn', `Accepted email ${cachedEmail.id} deferred after SmartMTA handoff failure: ${(error as Error).message}`);
    }
  }

  private buildCachedEmailSession(cachedEmail: CachedEmail): IExtendedSmtpSession {
    const routeData = this.parseCachedEmailRouteData(cachedEmail);
    const storedSession = routeData.session || {};
    const storedEnvelope = storedSession.envelope || {};
    const storedRcptTo = Array.isArray(storedEnvelope.rcptTo) && storedEnvelope.rcptTo.length > 0
      ? storedEnvelope.rcptTo
      : cachedEmail.to.map((address) => ({ address, args: {} }));
    const rcptTo = storedRcptTo
      .filter((recipient) => !!recipient.address)
      .map((recipient) => ({ address: recipient.address, args: recipient.args || {} }));
    const mailFrom = storedEnvelope.mailFrom
      ? { address: storedEnvelope.mailFrom.address, args: storedEnvelope.mailFrom.args || {} }
      : { address: cachedEmail.from || '', args: {} };
    const session = {
      id: `${storedSession.id || cachedEmail.id}-replay-${Date.now()}`,
      state: 'DATA' as unknown as IExtendedSmtpSession['state'],
      clientHostname: storedSession.clientHostname || '',
      mailFrom: mailFrom.address,
      rcptTo: rcptTo.map((recipient) => recipient.address),
      emailData: '',
      useTLS: !!storedSession.secure,
      connectionEnded: false,
      remoteAddress: storedSession.remoteAddress || '127.0.0.1',
      secure: !!storedSession.secure,
      authenticated: !!storedSession.authenticated,
      envelope: {
        mailFrom,
        rcptTo,
      },
    } as IExtendedSmtpSession;
    if (storedSession.user) {
      session.user = storedSession.user;
    }
    return session;
  }

  private parseCachedEmailRouteData(cachedEmail: CachedEmail): TStoredCachedEmailRouteData {
    try {
      return cachedEmail.routeData ? JSON.parse(cachedEmail.routeData) : {};
    } catch {
      return {};
    }
  }

  private extractHeader(rawMessageArg: string, headerNameArg: string): string | undefined {
    const lowerName = `${headerNameArg.toLowerCase()}:`;
    for (const line of rawMessageArg.split(/\r?\n/)) {
      if (!line) return undefined;
      if (line.toLowerCase().startsWith(lowerName)) {
        return line.slice(lowerName.length).trim();
      }
    }
  }

  private async updateAcceptedEmailFromQueueItem(
    item: TSmartMtaQueueItemLike,
    status: 'queued' | 'deferred' | 'delivered' | 'failed',
  ): Promise<void> {
    const cachedEmailId = this.getCachedEmailIdFromQueueItem(item);
    if (!cachedEmailId || !this.dcRouterRef.dcRouterDb?.isReady()) {
      return;
    }
    const cachedEmail = await CachedEmail.findById(cachedEmailId);
    if (!cachedEmail) {
      return;
    }
    if (this.isCachedEmailTerminal(cachedEmail) && status !== 'delivered') {
      return;
    }
    cachedEmail.attempts = Math.max(cachedEmail.attempts || 0, item.attempts || 0);
    for (const transaction of item.smtpTransactions || []) {
      this.appendSmtpTransaction(cachedEmail, transaction);
    }
    this.updateSmartMtaRouteData(cachedEmail, item, status);
    if (this.isDurableEnvelopeRow(cachedEmail)) {
      // The row represents the durably accepted and stored envelope; the queue
      // item only covers its relay recipients. Record the relay telemetry but
      // never let a relay outcome overwrite the stored copy's status — a failed
      // relay must not mark a successfully stored envelope as failed.
      await cachedEmail.save();
      await this.notifyEmailQueuePersisted(cachedEmail, `envelope-relay-${status}`);
      return;
    }
    if (status === 'delivered') {
      cachedEmail.markDelivered();
    } else if (status === 'failed') {
      cachedEmail.markFailed(item.lastError || 'SmartMTA delivery failed');
    } else if (status === 'deferred') {
      cachedEmail.status = 'deferred';
      cachedEmail.lastError = item.lastError || 'SmartMTA delivery deferred';
      const smartMtaNextAttempt = item.nextAttempt || new Date(Date.now() + ACCEPTED_EMAIL_RETRY_DELAY_MS);
      cachedEmail.smartMtaNextAttempt = smartMtaNextAttempt;
      cachedEmail.nextAttempt = new Date(smartMtaNextAttempt.getTime() + ACCEPTED_EMAIL_QUEUE_LEASE_MS);
    } else {
      cachedEmail.status = 'queued';
      cachedEmail.smartMtaNextAttempt = undefined;
      cachedEmail.nextAttempt = new Date(Date.now() + ACCEPTED_EMAIL_QUEUE_LEASE_MS);
    }
    await cachedEmail.save();
    await this.notifyEmailQueuePersisted(cachedEmail, `email-queue-${status}`);
  }

  private async waitForPromiseToSettleWithTimeout(
    promise: Promise<unknown>,
    timeoutMs: number,
  ): Promise<boolean> {
    let timeout: (ReturnType<typeof setTimeout> & { unref?: () => void }) | undefined;
    return await new Promise<boolean>((resolve) => {
      let settled = false;
      const settle = (didSettle: boolean) => {
        if (settled) {
          return;
        }
        settled = true;
        if (timeout) {
          clearTimeout(timeout);
        }
        resolve(didSettle);
      };
      timeout = setTimeout(() => settle(false), timeoutMs) as ReturnType<typeof setTimeout> & { unref?: () => void };
      timeout.unref?.();
      promise.then(
        () => settle(true),
        () => settle(true),
      );
    });
  }

  private clearSpoolTimer(): void {
    if (this.spoolTimer) {
      clearInterval(this.spoolTimer);
      this.spoolTimer = undefined;
    }
  }

  private setDcRouterCacheIdHeader(rawContent: string, cachedEmailId: string): string {
    const headerRegex = new RegExp(`^${DCROUTER_CACHE_ID_HEADER}:.*(?:\r?\n[\t ].*)*\r?\n?`, 'gim');
    const sanitizedContent = rawContent.replace(headerRegex, '');
    return `${DCROUTER_CACHE_ID_HEADER}: ${cachedEmailId}\r\n${sanitizedContent}`;
  }

  private isCachedEmailTerminal(cachedEmail: CachedEmail): boolean {
    return cachedEmail.status === 'delivered' || cachedEmail.status === 'failed';
  }

  /** Whether this row was durably accepted through the envelope acceptance path. */
  private isDurableEnvelopeRow(cachedEmail: CachedEmail): boolean {
    return this.parseCachedEmailRouteData(cachedEmail).acceptance === 'durable-envelope';
  }

  /** The CachedEmail row a live queue item was spooled from (via the cache-id header). */
  public getCachedEmailIdFromQueueItem(item: TSmartMtaQueueItemLike): string | undefined {
    return this.getHeaderValue(item.processingResult?.headers, DCROUTER_CACHE_ID_HEADER)
      || this.getHeaderValue(item.processingResult?.email?.headers, DCROUTER_CACHE_ID_HEADER);
  }

  private async postponeLiveSmartMtaOwnedEmail(
    cachedEmail: CachedEmail,
    emailServer: UnifiedEmailServer,
  ): Promise<boolean> {
    if (cachedEmail.status !== 'deferred') {
      return false;
    }
    const queueItem = this.getLiveQueueItemForCachedEmail(cachedEmail.id, emailServer);
    if (!queueItem) {
      return false;
    }
    const status = queueItem.status === 'deferred' ? 'deferred' : 'queued';
    cachedEmail.attempts = Math.max(cachedEmail.attempts || 0, queueItem.attempts || 0);
    if (queueItem.lastError) {
      cachedEmail.lastError = queueItem.lastError;
    }
    for (const transaction of queueItem.smtpTransactions || []) {
      this.appendSmtpTransaction(cachedEmail, transaction);
    }
    this.updateSmartMtaRouteData(cachedEmail, queueItem, status);
    const smartMtaNextAttempt = queueItem.nextAttempt || new Date(Date.now() + ACCEPTED_EMAIL_RETRY_DELAY_MS);
    cachedEmail.smartMtaNextAttempt = status === 'deferred' ? smartMtaNextAttempt : undefined;
    cachedEmail.nextAttempt = new Date(smartMtaNextAttempt.getTime() + ACCEPTED_EMAIL_QUEUE_LEASE_MS);
    await cachedEmail.save();
    await this.notifyEmailQueuePersisted(cachedEmail, 'email-live-queue-postponed');
    return true;
  }

  private getLiveQueueItemForCachedEmail(
    cachedEmailId: string,
    emailServer: UnifiedEmailServer,
  ): TSmartMtaQueueItemLike | undefined {
    const queueItems = (emailServer as TSmartMtaQueueReader).getQueueItems?.() || [];
    return queueItems.find((item) => this.getCachedEmailIdFromQueueItem(item) === cachedEmailId);
  }

  private getHeaderValue(headers: Record<string, string> | undefined, headerName: string): string | undefined {
    if (!headers) {
      return undefined;
    }
    const normalizedHeaderName = headerName.toLowerCase();
    const matchingHeaderName = Object.keys(headers).find((key) => key.toLowerCase() === normalizedHeaderName);
    return matchingHeaderName ? headers[matchingHeaderName] : undefined;
  }

  private updateSmartMtaRouteData(
    cachedEmail: CachedEmail,
    item: TSmartMtaQueueItemLike,
    status: 'queued' | 'deferred' | 'delivered' | 'failed',
  ): void {
    const routeData = this.parseCachedEmailRouteData(cachedEmail);
    routeData.smartMta = {
      status,
      nextAttempt: item.nextAttempt?.toISOString(),
      lastError: item.lastError,
      updatedAt: new Date().toISOString(),
    };
    cachedEmail.routeData = JSON.stringify(routeData);
  }

  public async readRawMessage(cachedEmailArg: CachedEmail): Promise<plugins.buffer.Buffer> {
    if (cachedEmailArg.rawContentObjectKey) {
      const blobStorage = this.dcRouterRef.getSmartMtaBlobStorageManager();
      if (!blobStorage) {
        throw new Error(`SmartBucket storage is unavailable for accepted email ${cachedEmailArg.id}`);
      }
      const rawMessage = await blobStorage.get(cachedEmailArg.rawContentObjectKey);
      if (!rawMessage) {
        throw new AcceptedEmailRawMessageMissingError(`Raw RFC822 object is missing for accepted email ${cachedEmailArg.id}`);
      }
      return plugins.buffer.Buffer.from(rawMessage);
    }
    if (cachedEmailArg.rawContent) {
      // Legacy rows persisted before SmartBucket-backed storage keep their
      // payload inline; serve it rather than declaring the payload lost.
      return plugins.buffer.Buffer.from(cachedEmailArg.rawContent, 'utf8');
    }
    throw new AcceptedEmailRawMessageMissingError(`Accepted email ${cachedEmailArg.id} has no raw RFC822 payload`);
  }

  public async deleteRawMessage(cachedEmailArg: CachedEmail): Promise<void> {
    if (!cachedEmailArg.rawContentObjectKey) return;
    const blobStorage = this.dcRouterRef.getSmartMtaBlobStorageManager();
    if (!blobStorage) {
      throw new Error(`SmartBucket storage is unavailable for accepted email ${cachedEmailArg.id}`);
    }
    await blobStorage.delete(cachedEmailArg.rawContentObjectKey);
  }

  private async cleanupRawMessageAfterSaveFailure(
    cachedEmailArg: CachedEmail,
    saveErrorArg: unknown,
  ): Promise<void> {
    try {
      await this.deleteRawMessage(cachedEmailArg);
      return;
    } catch (deleteError: unknown) {
      // Keep a durable, immediately-expired reference. CacheCleaner retries the
      // SmartBucket deletion before removing this recovery document.
      cachedEmailArg.submissionCredentialId = undefined;
      cachedEmailArg.submissionIdempotencyKey = undefined;
      cachedEmailArg.submissionDigest = undefined;
      cachedEmailArg.markFailed(
        `Accepted-email persistence failed; raw-message cleanup pending: ${(saveErrorArg as Error).message}`,
      );
      cachedEmailArg.setTTL(0);
      try {
        await cachedEmailArg.save();
      } catch (recoveryError: unknown) {
        throw new AggregateError(
          [saveErrorArg, deleteError, recoveryError],
          `Accepted-email persistence and durable raw-message cleanup failed for ${cachedEmailArg.id}`,
        );
      }
      logger.log(
        'warn',
        `Persisted raw-message cleanup recovery for ${cachedEmailArg.id}: ${(deleteError as Error).message}`,
      );
    }
  }

  private async persistRawMessage(cachedEmailArg: CachedEmail, rawMessageArg: string): Promise<void> {
    const blobStorage = this.dcRouterRef.getSmartMtaBlobStorageManager();
    if (!blobStorage) {
      throw new Error('SmartBucket storage is required before accepting SMTP messages');
    }
    const objectKey = `/email/messages/${cachedEmailArg.id}.eml`;
    const rawMessage = plugins.buffer.Buffer.from(rawMessageArg, 'utf8');
    await blobStorage.set(objectKey, rawMessage);
    cachedEmailArg.rawContent = undefined;
    cachedEmailArg.rawContentObjectKey = objectKey;
    cachedEmailArg.rawContentSize = rawMessage.length;
  }

  private runCachedEmailUpdate(
    cachedEmailIdArg: string | undefined,
    operationArg: () => Promise<void>,
  ): Promise<void> {
    if (!cachedEmailIdArg) return operationArg();
    const previous = this.cachedEmailUpdateChains.get(cachedEmailIdArg) || Promise.resolve();
    const current = previous.catch(() => undefined).then(operationArg);
    this.cachedEmailUpdateChains.set(cachedEmailIdArg, current);
    const cleanup = () => {
      if (this.cachedEmailUpdateChains.get(cachedEmailIdArg) === current) {
        this.cachedEmailUpdateChains.delete(cachedEmailIdArg);
      }
    };
    void current.then(cleanup, cleanup);
    return current;
  }

  private trackUpdatePromise(updatePromiseArg: Promise<void>): void {
    this.queueUpdatePromises.add(updatePromiseArg);
    const cleanup = () => this.queueUpdatePromises.delete(updatePromiseArg);
    void updatePromiseArg.then(cleanup, cleanup);
  }

  private appendSmtpTransaction(
    cachedEmailArg: CachedEmail,
    transactionArg: ICachedEmailSmtpTransaction,
  ): void {
    if (typeof cachedEmailArg.appendSmtpTransaction === 'function') {
      cachedEmailArg.appendSmtpTransaction(transactionArg);
      return;
    }
    const transactions = (cachedEmailArg.smtpTransactions || []).filter(
      (transaction) => transaction.id !== transactionArg.id,
    );
    transactions.push(structuredClone(transactionArg));
    cachedEmailArg.smtpTransactions = transactions.slice(-20);
  }

  private removeHeader(headers: Record<string, string>, headerName: string): void {
    const normalizedHeaderName = headerName.toLowerCase();
    for (const key of Object.keys(headers)) {
      if (key.toLowerCase() === normalizedHeaderName) {
        delete headers[key];
      }
    }
  }

  private throwIfMessageAcceptanceAborted(abortSignal: AbortSignal | undefined): void {
    if (abortSignal?.aborted) {
      throw new Error('Message acceptance aborted before SMTP success');
    }
  }
}