import * as plugins from '../plugins.js';

type TCoreMailOperation =
  plugins.servezoneInterfaces.data.TCoreMailWorkloadOperation;
type TCoreMailSubmission =
  plugins.servezoneInterfaces.data.ICoreMailSubmission;
type TCoreMailDelivery =
  plugins.servezoneInterfaces.data.ICoreMailInboundDelivery;
type TCoreMailUploadGrant =
  plugins.servezoneInterfaces.data.ICoreMailUploadGrant;
type TCoreMailDownloadGrant =
  plugins.servezoneInterfaces.data.ICoreMailDownloadGrant;

export interface ISzCoreMailConnectorOptions {
  endpointUrl: string;
  bindingId: string;
  credentialId: string;
  credentialVersion: number;
  credentialSecret: string;
}

export interface ISzCoreMailUploadPartOptions {
  submissionId: string;
  part: plugins.servezoneInterfaces.data.ICoreMailContentDescriptor;
  bytes: Uint8Array;
}

export interface ISzCoreMailInboundFetchResult {
  delivery: TCoreMailDelivery;
  bytes: Uint8Array;
}

interface ICoreMailAuthenticatedSession {
  bindingState: 'active' | 'draining';
  capabilities: Array<'outbound' | 'inbound'>;
  allowedOperations: TCoreMailOperation[];
  transferOrigin: string;
}

interface ICoreMailConnection {
  generation: number;
  socket: plugins.typedsocket.TypedSocket;
  statusSubscription: { unsubscribe(): void };
  session: ICoreMailAuthenticatedSession;
}

interface ICoreMailPendingConnection {
  generation: number;
  promise: Promise<ICoreMailConnection>;
}

class CoreMailTransferError extends Error {
  public constructor(
    messageArg: string,
    public readonly ambiguous: boolean,
  ) {
    super(messageArg);
    this.name = 'CoreMailTransferError';
  }
}

const identifierRegex = /^[A-Za-z0-9](?:[A-Za-z0-9._:-]{0,126}[A-Za-z0-9])?$/;

export class SzCoreMailConnector {
  private options?: Readonly<ISzCoreMailConnectorOptions>;
  private stopped = false;
  private generation = 0;
  private connection?: ICoreMailConnection;
  private pendingConnection?: ICoreMailPendingConnection;
  private readonly stopController = new AbortController();
  private readonly activeOperations = new Set<Promise<unknown>>();
  private stopPromise?: Promise<void>;

  public get isConfigured(): boolean {
    return Boolean(this.options);
  }

  public configure(optionsArg: ISzCoreMailConnectorOptions): void {
    if (this.stopped) {
      throw new Error('CoreMail connector has stopped.');
    }
    const options = this.normalizeOptions(optionsArg);
    if (this.options) {
      if (!this.optionsEqual(this.options, options)) {
        throw new Error('CoreMail connector configuration is one-shot.');
      }
      return;
    }
    this.options = Object.freeze(options);
  }

  public async init(): Promise<void> {
    await this.getAuthenticatedConnection();
  }

  public async prepareOutboundSubmission(
    requestArg:
      plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailPrepareOutboundSubmission['request'],
  ): Promise<
    plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailPrepareOutboundSubmission['response']
  > {
    const response = await this.fire<
      plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailPrepareOutboundSubmission
    >(
      'coreMailPrepareOutboundSubmission',
      {
        idempotencyKey: requestArg.idempotencyKey,
        message: plugins.servezoneInterfaces.data
          .normalizeCoreMailOutboundMessageDescriptor(requestArg.message),
      },
    );
    if (typeof response.replayed !== 'boolean') {
      throw new Error('CoreMail returned an invalid replay state.');
    }
    return {
      submission: plugins.servezoneInterfaces.data
        .normalizeCoreMailSubmission(response.submission),
      replayed: response.replayed,
    };
  }

  public async uploadOutboundPart(
    optionsArg: ISzCoreMailUploadPartOptions,
  ): Promise<TCoreMailSubmission> {
    return await this.trackOperation(this.uploadOutboundPartInner(optionsArg));
  }

  public async finalizeOutboundSubmission(
    submissionIdArg: string,
  ): Promise<TCoreMailSubmission> {
    const response = await this.fire<
      plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailFinalizeOutboundSubmission
    >('coreMailFinalizeOutboundSubmission', { submissionId: submissionIdArg });
    return plugins.servezoneInterfaces.data
      .normalizeCoreMailSubmission(response.submission);
  }

  public async getOutboundSubmission(
    submissionIdArg: string,
  ): Promise<TCoreMailSubmission> {
    const response = await this.fire<
      plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailGetOutboundSubmission
    >('coreMailGetOutboundSubmission', { submissionId: submissionIdArg });
    return plugins.servezoneInterfaces.data
      .normalizeCoreMailSubmission(response.submission);
  }

  public async listInboundDeliveries(
    requestArg:
      plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailListInboundDeliveries['request'],
  ): Promise<plugins.servezoneInterfaces.data.ICoreMailInboundDeliveryPage> {
    const response = await this.fire<
      plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailListInboundDeliveries
    >('coreMailListInboundDeliveries', requestArg);
    return plugins.servezoneInterfaces.data
      .normalizeCoreMailInboundDeliveryPage(response);
  }

  public async fetchInboundDelivery(
    deliveryArg: TCoreMailDelivery,
  ): Promise<ISzCoreMailInboundFetchResult> {
    return await this.trackOperation(this.fetchInboundDeliveryInner(deliveryArg));
  }

  public async acknowledgeInboundDelivery(
    requestArg:
      plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailAcknowledgeInboundDelivery['request'],
  ): Promise<
    plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailAcknowledgeInboundDelivery['response']
  > {
    const response = await this.fire<
      plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailAcknowledgeInboundDelivery
    >('coreMailAcknowledgeInboundDelivery', requestArg);
    if (typeof response.replayed !== 'boolean') {
      throw new Error('CoreMail returned an invalid acknowledgement replay state.');
    }
    return {
      delivery: plugins.servezoneInterfaces.data
        .normalizeCoreMailInboundDelivery(response.delivery),
      replayed: response.replayed,
    };
  }

  public stop(): Promise<void> {
    this.stopPromise ??= this.stopInner();
    return this.stopPromise;
  }

  private normalizeOptions(
    optionsArg: ISzCoreMailConnectorOptions,
  ): ISzCoreMailConnectorOptions {
    const endpointUrl = new URL(optionsArg.endpointUrl);
    if (
      optionsArg.endpointUrl.trim() !== optionsArg.endpointUrl
      || endpointUrl.protocol !== 'https:'
      || endpointUrl.username
      || endpointUrl.password
      || endpointUrl.search
      || endpointUrl.hash
      || endpointUrl.pathname !== '/socket'
      || endpointUrl.toString() !== optionsArg.endpointUrl
    ) {
      throw new Error('CoreMail endpointUrl must be one canonical HTTPS /socket URL.');
    }
    for (const [name, value] of [
      ['bindingId', optionsArg.bindingId],
      ['credentialId', optionsArg.credentialId],
    ] as const) {
      if (typeof value !== 'string' || !identifierRegex.test(value)) {
        throw new Error(`CoreMail ${name} must be canonical.`);
      }
    }
    if (
      !Number.isSafeInteger(optionsArg.credentialVersion)
      || optionsArg.credentialVersion < 1
    ) {
      throw new Error('CoreMail credentialVersion must be a positive integer.');
    }
    if (
      typeof optionsArg.credentialSecret !== 'string'
      || optionsArg.credentialSecret.length < 32
      || optionsArg.credentialSecret.length > 512
    ) {
      throw new Error('CoreMail credentialSecret must contain 32 to 512 characters.');
    }
    return {
      endpointUrl: optionsArg.endpointUrl,
      bindingId: optionsArg.bindingId,
      credentialId: optionsArg.credentialId,
      credentialVersion: optionsArg.credentialVersion,
      credentialSecret: optionsArg.credentialSecret,
    };
  }

  private optionsEqual(
    leftArg: Readonly<ISzCoreMailConnectorOptions>,
    rightArg: Readonly<ISzCoreMailConnectorOptions>,
  ): boolean {
    return leftArg.endpointUrl === rightArg.endpointUrl
      && leftArg.bindingId === rightArg.bindingId
      && leftArg.credentialId === rightArg.credentialId
      && leftArg.credentialVersion === rightArg.credentialVersion
      && leftArg.credentialSecret === rightArg.credentialSecret;
  }

  private requireOptions(): Readonly<ISzCoreMailConnectorOptions> {
    if (this.stopped) {
      throw this.createAbortError();
    }
    if (!this.options) {
      throw new Error('CoreMail connector is not configured.');
    }
    return this.options;
  }

  private async getAuthenticatedConnection(): Promise<ICoreMailConnection> {
    this.requireOptions();
    if (
      this.connection
      && this.connection.socket.getStatus() === 'connected'
    ) {
      return this.connection;
    }
    if (this.pendingConnection) {
      return await this.pendingConnection.promise;
    }
    const generation = ++this.generation;
    const promise = this.connectAndAuthenticate(generation);
    this.pendingConnection = { generation, promise };
    try {
      return await promise;
    } finally {
      if (this.pendingConnection?.promise === promise) {
        this.pendingConnection = undefined;
      }
    }
  }

  private async connectAndAuthenticate(
    generationArg: number,
  ): Promise<ICoreMailConnection> {
    const options = this.requireOptions();
    let socket: plugins.typedsocket.TypedSocket | undefined;
    let statusSubscription: { unsubscribe(): void } | undefined;
    let socketInvalidated = false;
    try {
      socket = await this.createSocket(options.endpointUrl);
      this.assertGeneration(generationArg);
      statusSubscription = socket.statusSubject.subscribe((statusArg) => {
        if (statusArg !== 'connected') {
          socketInvalidated = true;
          this.invalidateConnection(socket!, generationArg, statusSubscription!);
        }
      });
      const response = await this.trackOperation(socket.createTypedRequest<
        plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailAuthenticateWorkload
      >('coreMailAuthenticateWorkload', undefined, {
        timeoutMs: 4_500,
        abortSignal: this.stopController.signal,
      }).fire({
        bindingId: options.bindingId,
        credentialId: options.credentialId,
        credentialVersion: options.credentialVersion,
        credentialSecret: options.credentialSecret,
      }));
      this.assertGeneration(generationArg);
      if (socket.getStatus() !== 'connected') {
        throw new Error('CoreMail disconnected during authentication.');
      }
      const connection: ICoreMailConnection = {
        generation: generationArg,
        socket,
        statusSubscription,
        session: this.normalizeAuthenticationResponse(response, options),
      };
      this.connection = connection;
      return connection;
    } catch (error) {
      statusSubscription?.unsubscribe();
      if (!socketInvalidated) {
        await socket?.stop().catch(() => undefined);
      }
      throw error;
    }
  }

  private async createSocket(
    endpointUrlArg: string,
  ): Promise<plugins.typedsocket.TypedSocket> {
    return await plugins.typedsocket.TypedSocket.createClient(
      new plugins.typedrequest.TypedRouter(),
      endpointUrlArg,
      {
        autoReconnect: false,
        maxRetries: 0,
        abortSignal: this.stopController.signal,
      },
    );
  }

  private normalizeAuthenticationResponse(
    responseArg:
      plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailAuthenticateWorkload['response'],
    optionsArg: Readonly<ISzCoreMailConnectorOptions>,
  ): ICoreMailAuthenticatedSession {
    if (
      !responseArg
      || typeof responseArg !== 'object'
      || Array.isArray(responseArg)
      || JSON.stringify(Object.keys(responseArg).sort()) !== JSON.stringify([
        'allowedOperations',
        'authenticated',
        'bindingState',
        'capabilities',
        'coreMailTransferOrigin',
        'credentialId',
        'credentialVersion',
      ])
      || responseArg.authenticated !== true
      || responseArg.credentialId !== optionsArg.credentialId
      || responseArg.credentialVersion !== optionsArg.credentialVersion
      || (responseArg.bindingState !== 'active' && responseArg.bindingState !== 'draining')
      || !Array.isArray(responseArg.capabilities)
      || responseArg.capabilities.some((entryArg) =>
        entryArg !== 'outbound' && entryArg !== 'inbound'
      )
      || new Set(responseArg.capabilities).size !== responseArg.capabilities.length
      || !Array.isArray(responseArg.allowedOperations)
    ) {
      throw new Error('CoreMail returned invalid workload authentication authority.');
    }
    const expectedOperations =
      plugins.servezoneInterfaces.data.resolveCoreMailWorkloadOperations(
        responseArg.bindingState,
        responseArg.capabilities,
      );
    if (
      JSON.stringify(responseArg.allowedOperations)
      !== JSON.stringify(expectedOperations)
    ) {
      throw new Error('CoreMail returned inconsistent workload operations.');
    }
    return Object.freeze({
      bindingState: responseArg.bindingState,
      capabilities: Object.freeze([...responseArg.capabilities]) as
        Array<'outbound' | 'inbound'>,
      allowedOperations: Object.freeze([...responseArg.allowedOperations]) as
        TCoreMailOperation[],
      transferOrigin: this.normalizeTransferOrigin(
        responseArg.coreMailTransferOrigin,
      ),
    });
  }

  private normalizeTransferOrigin(valueArg: unknown): string {
    if (typeof valueArg !== 'string') {
      throw new Error('CoreMail returned an invalid transfer origin.');
    }
    const url = new URL(valueArg);
    if (
      url.protocol !== 'https:'
      || url.username
      || url.password
      || url.search
      || url.hash
      || url.pathname !== '/'
      || url.origin !== valueArg
    ) {
      throw new Error('CoreMail returned a noncanonical transfer origin.');
    }
    return valueArg;
  }

  private async fire<
    TRequest extends plugins.typedrequestInterfaces.ITypedRequest,
  >(
    methodArg: TRequest['method'] & TCoreMailOperation,
    requestArg: TRequest['request'],
  ): Promise<TRequest['response']> {
    const connection = await this.getAuthenticatedConnection();
    if (!connection.session.allowedOperations.includes(methodArg)) {
      throw new Error(`CoreMail operation ${methodArg} is unavailable.`);
    }
    const response = await this.trackOperation(
      connection.socket.createTypedRequest<TRequest>(methodArg, undefined, {
        timeoutMs: 30_000,
        abortSignal: this.stopController.signal,
      }).fire(requestArg),
    );
    this.assertGeneration(connection.generation);
    return response;
  }

  private async prepareUploadGrant(
    submissionIdArg: string,
    partIdArg: string,
  ): Promise<{ grant: TCoreMailUploadGrant; connection: ICoreMailConnection }> {
    const connection = await this.getAuthenticatedConnection();
    const method: TCoreMailOperation = 'coreMailPrepareOutboundPartUpload';
    if (!connection.session.allowedOperations.includes(method)) {
      throw new Error(`CoreMail operation ${method} is unavailable.`);
    }
    const response = await this.trackOperation(connection.socket.createTypedRequest<
      plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailPrepareOutboundPartUpload
    >(method, undefined, {
      timeoutMs: 30_000,
      abortSignal: this.stopController.signal,
    }).fire({ submissionId: submissionIdArg, partId: partIdArg }));
    this.assertGeneration(connection.generation);
    return {
      grant: plugins.servezoneInterfaces.data.normalizeCoreMailUploadGrant(
        response.grant,
      ),
      connection,
    };
  }

  private async uploadOutboundPartInner(
    optionsArg: ISzCoreMailUploadPartOptions,
  ): Promise<TCoreMailSubmission> {
    if (!(optionsArg.bytes instanceof Uint8Array)) {
      throw new Error('CoreMail outbound part bytes must be a Uint8Array.');
    }
    const expectedSha256 = plugins.servezoneInterfaces.data
      .normalizeCoreMailSha256(optionsArg.part.sha256);
    const actualSha256 = await this.createSha256(optionsArg.bytes);
    if (
      optionsArg.bytes.byteLength !== optionsArg.part.lengthBytes
      || actualSha256 !== expectedSha256
    ) {
      throw new Error('CoreMail outbound part bytes do not match their descriptor.');
    }

    let lastError: unknown;
    for (let grantAttempt = 0; grantAttempt < 2; grantAttempt++) {
      const { grant, connection } = await this.prepareUploadGrant(
        optionsArg.submissionId,
        optionsArg.part.partId,
      );
      this.assertUploadGrantContext(grant, optionsArg.part, connection);
      let uploadError: unknown;
      try {
        await this.putTransfer(grant, connection.session.transferOrigin, optionsArg.bytes);
      } catch (errorArg) {
        uploadError = errorArg;
      }

      let completionError: unknown;
      for (let completionAttempt = 0; completionAttempt < 2; completionAttempt++) {
        try {
          const response = await this.fire<
            plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailCompleteOutboundPartUpload
          >('coreMailCompleteOutboundPartUpload', {
            submissionId: optionsArg.submissionId,
            partId: optionsArg.part.partId,
            grantId: grant.grantId,
            sha256: expectedSha256,
            lengthBytes: optionsArg.bytes.byteLength,
          });
          return plugins.servezoneInterfaces.data
            .normalizeCoreMailSubmission(response.submission);
        } catch (errorArg) {
          completionError = errorArg;
          if (!this.hasCoreMailErrorCode(errorArg, 'INVALID_REQUEST')) {
            throw uploadError ?? errorArg;
          }
          if (completionAttempt === 0) {
            await this.waitForDelay(25);
          }
        }
      }

      const current = await this.getOutboundSubmission(optionsArg.submissionId);
      const part = current.parts.find((entryArg) =>
        entryArg.partId === optionsArg.part.partId
      );
      if (part?.state === 'complete') {
        return current;
      }
      lastError = uploadError ?? completionError;
      const uploadWasAmbiguous = uploadError instanceof CoreMailTransferError
        && uploadError.ambiguous;
      if (
        grantAttempt === 0
        && (uploadWasAmbiguous || !uploadError)
        && this.hasCoreMailErrorCode(completionError, 'INVALID_REQUEST')
      ) {
        continue;
      }
      throw lastError;
    }
    throw lastError ?? new Error('CoreMail outbound part upload failed.');
  }

  private assertUploadGrantContext(
    grantArg: TCoreMailUploadGrant,
    partArg: plugins.servezoneInterfaces.data.ICoreMailContentDescriptor,
    connectionArg: ICoreMailConnection,
  ): void {
    this.assertGeneration(connectionArg.generation);
    this.resolveTransferUrl(grantArg, connectionArg.session.transferOrigin);
    if (
      grantArg.method !== 'PUT'
      || grantArg.sha256 !== partArg.sha256
      || grantArg.lengthBytes !== partArg.lengthBytes
      || grantArg.contentType !== partArg.contentType
    ) {
      throw new Error('CoreMail upload grant does not match the prepared part.');
    }
  }

  private async putTransfer(
    grantArg: TCoreMailUploadGrant,
    transferOriginArg: string,
    bytesArg: Uint8Array,
  ): Promise<void> {
    await this.trackOperation(this.putTransferInner(
      grantArg,
      transferOriginArg,
      bytesArg,
    ));
  }

  private async putTransferInner(
    grantArg: TCoreMailUploadGrant,
    transferOriginArg: string,
    bytesArg: Uint8Array,
  ): Promise<void> {
    const deadline = this.createTransferDeadline(
      plugins.servezoneInterfaces.data.coreMailLimits.transferOverallTimeoutMs,
    );
    try {
      let response: Response;
      try {
        response = await this.fetchTransfer(
          this.resolveTransferUrl(grantArg, transferOriginArg),
          {
            method: 'PUT',
            headers: {
              Authorization: `Bearer ${grantArg.bearerToken}`,
              'Content-Type': grantArg.contentType,
            },
            body: this.copyBytesToArrayBuffer(bytesArg),
            cache: 'no-store',
            credentials: 'omit',
            redirect: 'error',
            signal: deadline.controller.signal,
          },
        );
      } catch (error) {
        if (deadline.controller.signal.aborted) {
          throw deadline.controller.signal.reason || error;
        }
        throw new CoreMailTransferError('CoreMail upload response is ambiguous.', true);
      }
      if (response.status !== 204) {
        await response.body?.cancel().catch(() => undefined);
        throw new CoreMailTransferError(
          `CoreMail upload failed with HTTP ${response.status}.`,
          false,
        );
      }
    } finally {
      deadline.cleanup();
    }
  }

  private async fetchInboundDeliveryInner(
    deliveryArg: TCoreMailDelivery,
  ): Promise<ISzCoreMailInboundFetchResult> {
    const delivery = plugins.servezoneInterfaces.data
      .normalizeCoreMailInboundDelivery(deliveryArg);
    const connection = await this.getAuthenticatedConnection();
    const method: TCoreMailOperation = 'coreMailPrepareInboundFetch';
    if (!connection.session.allowedOperations.includes(method)) {
      throw new Error(`CoreMail operation ${method} is unavailable.`);
    }
    const response = await this.trackOperation(connection.socket.createTypedRequest<
      plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailPrepareInboundFetch
    >(method, undefined, {
      timeoutMs: 30_000,
      abortSignal: this.stopController.signal,
    }).fire({ deliveryId: delivery.deliveryId }));
    this.assertGeneration(connection.generation);
    const grant = plugins.servezoneInterfaces.data
      .normalizeCoreMailDownloadGrant(response.grant);
    this.assertDownloadGrantContext(grant, delivery, connection);
    const bytes = await this.getTransfer(
      grant,
      connection.session.transferOrigin,
    );
    const completed = await this.completeInboundFetchWithRetry(delivery, grant);
    return { delivery: completed, bytes };
  }

  private assertDownloadGrantContext(
    grantArg: TCoreMailDownloadGrant,
    deliveryArg: TCoreMailDelivery,
    connectionArg: ICoreMailConnection,
  ): void {
    this.assertGeneration(connectionArg.generation);
    this.resolveTransferUrl(grantArg, connectionArg.session.transferOrigin);
    if (
      grantArg.method !== 'GET'
      || grantArg.sha256 !== deliveryArg.rawMime.sha256
      || grantArg.lengthBytes !== deliveryArg.rawMime.lengthBytes
      || grantArg.contentType !== deliveryArg.rawMime.contentType
    ) {
      throw new Error('CoreMail download grant does not match the selected delivery.');
    }
  }

  private async getTransfer(
    grantArg: TCoreMailDownloadGrant,
    transferOriginArg: string,
  ): Promise<Uint8Array> {
    return await this.trackOperation(this.getTransferInner(grantArg, transferOriginArg));
  }

  private async getTransferInner(
    grantArg: TCoreMailDownloadGrant,
    transferOriginArg: string,
  ): Promise<Uint8Array> {
    const deadline = this.createTransferDeadline(
      plugins.servezoneInterfaces.data.coreMailLimits.transferOverallTimeoutMs,
    );
    let headerTimeout: ReturnType<typeof setTimeout> | undefined = setTimeout(() => {
      deadline.controller.abort(new Error('CoreMail transfer header timeout.'));
    }, plugins.servezoneInterfaces.data.coreMailLimits.transferHeaderTimeoutMs);
    let responseBody: ReadableStream<Uint8Array> | undefined;
    let reader: ReadableStreamDefaultReader<Uint8Array> | undefined;
    let idleTimeout: ReturnType<typeof setTimeout> | undefined;
    try {
      const response = await this.fetchTransfer(
        this.resolveTransferUrl(grantArg, transferOriginArg),
        {
          method: 'GET',
          headers: { Authorization: `Bearer ${grantArg.bearerToken}` },
          cache: 'no-store',
          credentials: 'omit',
          redirect: 'error',
          signal: deadline.controller.signal,
        },
      );
      clearTimeout(headerTimeout);
      headerTimeout = undefined;
      responseBody = response.body || undefined;
      if (
        response.status !== 200
        || response.headers.get('content-type') !== grantArg.contentType
        || response.headers.get('content-length') !== String(grantArg.lengthBytes)
        || response.headers.get('cache-control') !== 'no-store'
        || response.headers.get('x-content-type-options') !== 'nosniff'
        || !responseBody
      ) {
        throw new Error('CoreMail download response metadata is invalid.');
      }
      reader = responseBody.getReader();
      const chunks: Uint8Array[] = [];
      let totalLength = 0;
      const resetIdleTimeout = () => {
        if (idleTimeout) clearTimeout(idleTimeout);
        idleTimeout = setTimeout(() => {
          deadline.controller.abort(new Error('CoreMail transfer idle timeout.'));
        }, plugins.servezoneInterfaces.data.coreMailLimits.transferIdleTimeoutMs);
      };
      resetIdleTimeout();
      while (true) {
        const result = await reader.read();
        if (result.done) break;
        resetIdleTimeout();
        totalLength += result.value.byteLength;
        if (totalLength > grantArg.lengthBytes) {
          throw new Error('CoreMail download exceeds its declared length.');
        }
        chunks.push(result.value);
      }
      if (idleTimeout) clearTimeout(idleTimeout);
      if (totalLength !== grantArg.lengthBytes) {
        throw new Error('CoreMail download length does not match its grant.');
      }
      const bytes = new Uint8Array(totalLength);
      let offset = 0;
      for (const chunk of chunks) {
        bytes.set(chunk, offset);
        offset += chunk.byteLength;
      }
      if (await this.createSha256(bytes) !== grantArg.sha256) {
        throw new Error('CoreMail download digest does not match its grant.');
      }
      return bytes;
    } catch (error) {
      if (reader) {
        await reader.cancel().catch(() => undefined);
      } else {
        await responseBody?.cancel().catch(() => undefined);
      }
      throw error;
    } finally {
      clearTimeout(headerTimeout);
      if (idleTimeout) clearTimeout(idleTimeout);
      deadline.cleanup();
    }
  }

  private async completeInboundFetchWithRetry(
    deliveryArg: TCoreMailDelivery,
    grantArg: TCoreMailDownloadGrant,
  ): Promise<TCoreMailDelivery> {
    let lastError: unknown;
    for (let attempt = 0; attempt < 4; attempt++) {
      try {
        const response = await this.fire<
          plugins.servezoneInterfaces.requests.coremail.IReq_CoreMailCompleteInboundFetch
        >('coreMailCompleteInboundFetch', {
          deliveryId: deliveryArg.deliveryId,
          grantId: grantArg.grantId,
          sha256: grantArg.sha256,
          lengthBytes: grantArg.lengthBytes,
        });
        const delivery = plugins.servezoneInterfaces.data
          .normalizeCoreMailInboundDelivery(response.delivery);
        if (delivery.deliveryId !== deliveryArg.deliveryId) {
          throw new Error('CoreMail completed another inbound delivery.');
        }
        return delivery;
      } catch (errorArg) {
        lastError = errorArg;
        if (
          attempt === 3
          || !this.hasCoreMailErrorCode(errorArg, 'INVALID_REQUEST')
        ) {
          throw errorArg;
        }
        await this.waitForDelay(25 * (2 ** attempt));
      }
    }
    throw lastError;
  }

  private resolveTransferUrl(
    grantArg: TCoreMailUploadGrant | TCoreMailDownloadGrant,
    transferOriginArg: string,
  ): string {
    const url = new URL(grantArg.path, transferOriginArg);
    if (
      url.origin !== transferOriginArg
      || url.pathname !== grantArg.path
      || url.search
      || url.hash
    ) {
      throw new Error('CoreMail transfer grant escapes its authenticated origin.');
    }
    return url.toString();
  }

  private async createSha256(
    bytesArg: Uint8Array,
  ): Promise<plugins.servezoneInterfaces.data.TCoreMailSha256> {
    const digest = new Uint8Array(
      await globalThis.crypto.subtle.digest(
        'SHA-256',
        this.copyBytesToArrayBuffer(bytesArg),
      ),
    );
    return `sha256:${Array.from(digest)
      .map((byteArg) => byteArg.toString(16).padStart(2, '0'))
      .join('')}`;
  }

  private copyBytesToArrayBuffer(bytesArg: Uint8Array): ArrayBuffer {
    const bytes = new Uint8Array(bytesArg.byteLength);
    bytes.set(bytesArg);
    return bytes.buffer;
  }

  private async fetchTransfer(
    inputArg: string,
    initArg: RequestInit,
  ): Promise<Response> {
    return await globalThis.fetch(inputArg, initArg);
  }

  private createTransferDeadline(timeoutMsArg: number): {
    controller: AbortController;
    cleanup: () => void;
  } {
    const controller = new AbortController();
    const onStop = () => controller.abort(this.stopController.signal.reason);
    if (this.stopController.signal.aborted) {
      onStop();
    } else {
      this.stopController.signal.addEventListener('abort', onStop, { once: true });
    }
    const timeout = setTimeout(() => {
      controller.abort(new Error('CoreMail transfer overall timeout.'));
    }, timeoutMsArg);
    return {
      controller,
      cleanup: () => {
        clearTimeout(timeout);
        this.stopController.signal.removeEventListener('abort', onStop);
      },
    };
  }

  private hasCoreMailErrorCode(
    errorArg: unknown,
    codeArg: plugins.servezoneInterfaces.data.TCoreMailErrorCode,
  ): boolean {
    if (!(errorArg instanceof plugins.typedrequest.TypedResponseError)) {
      return false;
    }
    try {
      return plugins.servezoneInterfaces.data
        .normalizeCoreMailErrorData(errorArg.errorData).code === codeArg;
    } catch {
      return false;
    }
  }

  private async waitForDelay(delayMsArg: number): Promise<void> {
    if (this.stopController.signal.aborted) throw this.createAbortError();
    await this.trackOperation(new Promise<void>((resolveArg, rejectArg) => {
      const timeout = setTimeout(() => {
        this.stopController.signal.removeEventListener('abort', onAbort);
        resolveArg();
      }, delayMsArg);
      const onAbort = () => {
        clearTimeout(timeout);
        rejectArg(this.createAbortError());
      };
      this.stopController.signal.addEventListener('abort', onAbort, { once: true });
    }));
  }

  private invalidateConnection(
    socketArg: plugins.typedsocket.TypedSocket,
    generationArg: number,
    subscriptionArg: { unsubscribe(): void },
  ): void {
    const ownsConnection = this.connection?.socket === socketArg
      && this.connection.generation === generationArg;
    const ownsPending = this.pendingConnection?.generation === generationArg;
    if (!ownsConnection && !ownsPending) return;
    this.generation++;
    if (ownsConnection) this.connection = undefined;
    if (ownsPending) this.pendingConnection = undefined;
    subscriptionArg.unsubscribe();
    void socketArg.stop().catch(() => undefined);
  }

  private assertGeneration(generationArg: number): void {
    if (this.stopped || generationArg !== this.generation) {
      throw this.createAbortError();
    }
  }

  private trackOperation<T>(promiseArg: Promise<T>): Promise<T> {
    this.activeOperations.add(promiseArg);
    return promiseArg.finally(() => {
      this.activeOperations.delete(promiseArg);
    });
  }

  private async stopInner(): Promise<void> {
    this.stopped = true;
    this.generation++;
    this.stopController.abort(this.createAbortError());
    const connection = this.connection;
    const pendingConnection = this.pendingConnection;
    this.connection = undefined;
    this.pendingConnection = undefined;
    connection?.statusSubscription.unsubscribe();
    await Promise.allSettled([
      connection?.socket.stop(),
      pendingConnection?.promise.then(async (resultArg) => {
        resultArg.statusSubscription.unsubscribe();
        await resultArg.socket.stop();
      }),
      ...this.activeOperations,
    ].filter(Boolean) as Promise<unknown>[]);
    this.options = undefined;
  }

  private createAbortError(): Error {
    const error = new Error('CoreMail connector operation aborted.');
    error.name = 'AbortError';
    return error;
  }
}