type RetryableReceiptError = {
  retryAfterMs: number;
};

const MAX_RETRY_JITTER_RATIO = 0.25;

type PendingWrite<Input, Output> = {
  input: Input;
  bytes: number;
  signal: AbortSignal | null;
  state: 'blocked' | 'queued' | 'active' | 'settled';
  aborted: unknown;
  onAbort: (() => void) | null;
  resolve: (output: Output) => void;
  reject: (error: unknown) => void;
};

export type RuntimeReceiptWriterStats = {
  active: boolean;
  queued: number;
  blocked: number;
  bufferedBytes: number;
};

export type RuntimeReceiptWriterRetryEvent<Input> = {
  phase: 'retry' | 'recovered';
  attempt: number;
  batchId: number;
  batchSize: number;
  batchBytes: number;
  firstInput: Input;
  elapsedMs: number;
  retryAfterMs: number;
  queued: number;
  blocked: number;
  bufferedBytes: number;
  error?: unknown;
};

export class RuntimeReceiptWriterBufferLimitError extends Error {
  constructor(
    readonly inputBytes: number,
    readonly maxBufferedBytes: number,
  ) {
    super(
      `Runtime receipt input is ${inputBytes} bytes, exceeding the ` +
        `${maxBufferedBytes}-byte writer limit.`,
    );
    this.name = 'RuntimeReceiptWriterBufferLimitError';
  }
}

export class RuntimeReceiptWriterResultCountError extends Error {
  constructor(
    readonly inputCount: number,
    readonly outputCount: number,
  ) {
    super(
      `Runtime receipt transport returned ${outputCount} results for ` +
        `${inputCount} inputs.`,
    );
    this.name = 'RuntimeReceiptWriterResultCountError';
  }
}

export class RuntimeReceiptWriterClosedError extends Error {
  constructor() {
    super('Runtime receipt writer is closed.');
    this.name = 'RuntimeReceiptWriterClosedError';
  }
}

export class RuntimeReceiptWriterRetryDeadlineError extends Error {
  constructor(
    readonly attempts: number,
    readonly elapsedMs: number,
    readonly maxRetryElapsedMs: number,
    options?: { cause?: unknown },
  ) {
    super(
      `Runtime receipt persistence did not recover within ${maxRetryElapsedMs}ms after ${attempts} attempts.`,
      options,
    );
    this.name = 'RuntimeReceiptWriterRetryDeadlineError';
  }
}

/**
 * Private batching implementation used by a Play Durability Store Adapter.
 *
 * The writer deliberately exposes no receipt-domain operations. The Adapter
 * translates claim, settle, read, and heartbeat calls into commands while this
 * Module owns serialization, bounded buffering, and transport redelivery.
 */
export class RuntimeReceiptWriter<Input, Output> {
  readonly #batchKey: (input: Input) => unknown;
  readonly #send: (
    inputs: readonly Input[],
    options: { signal: AbortSignal },
  ) => Promise<readonly Output[]>;
  readonly #classifyRetryableError: (
    error: unknown,
  ) => RetryableReceiptError | null;
  readonly #estimateBytes: (input: Input) => number;
  readonly #maxBatchSize: number;
  readonly #maxBatchBytes: number;
  readonly #targetBatchBytes: number;
  readonly #maxBufferedBytes: number;
  readonly #maxFlushMs: number;
  readonly #maxRetryElapsedMs: number;
  readonly #onRetryEvent:
    | ((event: RuntimeReceiptWriterRetryEvent<Input>) => void)
    | null;
  readonly #queued: Array<PendingWrite<Input, Output>> = [];
  readonly #blocked: Array<PendingWrite<Input, Output>> = [];
  readonly #closeController = new AbortController();

  #bufferedBytes = 0;
  #nextBatchId = 0;
  #activeBatch: Array<PendingWrite<Input, Output>> | null = null;
  #pump: Promise<void> | null = null;
  #flushTimer: ReturnType<typeof setTimeout> | null = null;
  #closed = false;
  #closeReason: unknown = null;
  #drainWaiters: Array<{
    resolve: () => void;
    reject: (error: unknown) => void;
  }> = [];

  constructor(options: {
    batchKey: (input: Input) => unknown;
    send: (
      inputs: readonly Input[],
      options: { signal: AbortSignal },
    ) => Promise<readonly Output[]>;
    classifyRetryableError?: (error: unknown) => RetryableReceiptError | null;
    estimateBytes?: (input: Input) => number;
    maxBatchSize?: number;
    maxBatchBytes?: number;
    targetBatchBytes?: number;
    maxBufferedBytes?: number;
    maxFlushMs?: number;
    maxRetryElapsedMs?: number;
    onRetryEvent?: (event: RuntimeReceiptWriterRetryEvent<Input>) => void;
  }) {
    this.#batchKey = options.batchKey;
    this.#send = options.send;
    this.#classifyRetryableError =
      options.classifyRetryableError ?? (() => null);
    this.#estimateBytes = options.estimateBytes ?? estimateJsonBytes;
    this.#maxBatchSize = normalizePositiveInteger(options.maxBatchSize, 100);
    this.#maxBufferedBytes = normalizePositiveInteger(
      options.maxBufferedBytes,
      32 * 1024 * 1024,
    );
    this.#maxBatchBytes = Math.min(
      normalizePositiveInteger(options.maxBatchBytes, this.#maxBufferedBytes),
      this.#maxBufferedBytes,
    );
    this.#targetBatchBytes = Math.min(
      normalizePositiveInteger(options.targetBatchBytes, this.#maxBatchBytes),
      this.#maxBatchBytes,
    );
    this.#maxFlushMs = Math.max(0, Math.floor(options.maxFlushMs ?? 5));
    this.#maxRetryElapsedMs = normalizePositiveInteger(
      options.maxRetryElapsedMs,
      2 * 60_000,
    );
    this.#onRetryEvent = options.onRetryEvent ?? null;
  }

  write(input: Input, options: { signal?: AbortSignal } = {}): Promise<Output> {
    if (this.#closed) {
      return Promise.reject(
        this.#closeReason ?? new RuntimeReceiptWriterClosedError(),
      );
    }

    const bytes = Math.max(0, Math.ceil(this.#estimateBytes(input)));
    if (bytes > this.#maxBufferedBytes || bytes > this.#maxBatchBytes) {
      return Promise.reject(
        new RuntimeReceiptWriterBufferLimitError(
          bytes,
          Math.min(this.#maxBufferedBytes, this.#maxBatchBytes),
        ),
      );
    }
    if (options.signal?.aborted) {
      return Promise.reject(abortReason(options.signal));
    }

    return new Promise<Output>((resolve, reject) => {
      const pending: PendingWrite<Input, Output> = {
        input,
        bytes,
        signal: options.signal ?? null,
        state: 'blocked',
        aborted: null,
        onAbort: null,
        resolve,
        reject,
      };
      pending.onAbort = () => this.#abortPending(pending);
      pending.signal?.addEventListener('abort', pending.onAbort, {
        once: true,
      });

      if (this.#canBuffer(bytes)) {
        this.#accept(pending);
      } else {
        this.#blocked.push(pending);
      }
    });
  }

  async flush(): Promise<void> {
    this.#clearFlushTimer();
    this.#startPump();
    if (this.#isDrained()) return;
    await new Promise<void>((resolve, reject) => {
      this.#drainWaiters.push({ resolve, reject });
    });
  }

  async close(reason?: unknown): Promise<void> {
    if (!this.#closed) {
      this.#closed = true;
      this.#closeReason =
        reason === undefined ? new RuntimeReceiptWriterClosedError() : reason;
      if (reason !== undefined) {
        this.#closeController.abort(reason);
        this.#rejectWaiting(reason);
      }
    }
    if (reason === undefined) {
      await this.flush();
      return;
    }
    if (this.#pump) await this.#pump;
  }

  getStats(): RuntimeReceiptWriterStats {
    return {
      active: this.#activeBatch !== null,
      queued: this.#queued.length,
      blocked: this.#blocked.length,
      bufferedBytes: this.#bufferedBytes,
    };
  }

  #canBuffer(bytes: number): boolean {
    return this.#bufferedBytes + bytes <= this.#maxBufferedBytes;
  }

  #accept(pending: PendingWrite<Input, Output>): void {
    pending.state = 'queued';
    this.#queued.push(pending);
    this.#bufferedBytes += pending.bytes;
    if (
      this.#queued.length >= this.#maxBatchSize ||
      this.#queuedBytesAtHead() >= this.#targetBatchBytes
    ) {
      this.#clearFlushTimer();
      this.#startPump();
      return;
    }
    this.#scheduleFlush();
  }

  #scheduleFlush(): void {
    if (this.#pump || this.#flushTimer || this.#queued.length === 0) return;
    this.#flushTimer = setTimeout(() => {
      this.#flushTimer = null;
      this.#startPump();
    }, this.#maxFlushMs);
  }

  #startPump(): void {
    if (this.#pump || this.#queued.length === 0) {
      this.#notifyDrained();
      return;
    }
    this.#pump = this.#runPump().finally(() => {
      this.#pump = null;
      if (this.#queued.length > 0) this.#startPump();
      else this.#notifyDrained();
    });
  }

  async #runPump(): Promise<void> {
    while (this.#queued.length > 0) {
      const batch = this.#takeBatch();
      if (batch.length === 0) continue;
      const batchId = ++this.#nextBatchId;
      this.#activeBatch = batch;
      try {
        const outputs = await this.#sendWithRetry(batch, batchId);
        if (outputs.length !== batch.length) {
          throw new RuntimeReceiptWriterResultCountError(
            batch.length,
            outputs.length,
          );
        }
        batch.forEach((pending, index) => {
          if (pending.aborted !== null) {
            this.#settleRejected(pending, pending.aborted);
          } else {
            this.#settleResolved(pending, outputs[index]!);
          }
        });
      } catch (error) {
        batch.forEach((pending) => {
          this.#settleRejected(
            pending,
            pending.aborted !== null ? pending.aborted : error,
          );
        });
      } finally {
        this.#activeBatch = null;
        for (const pending of batch) {
          this.#bufferedBytes -= pending.bytes;
        }
        this.#admitBlocked();
      }
    }
  }

  #takeBatch(): Array<PendingWrite<Input, Output>> {
    while (this.#queued[0]?.aborted !== null) {
      const aborted = this.#queued.shift()!;
      this.#bufferedBytes -= aborted.bytes;
      this.#settleRejected(aborted, aborted.aborted);
      this.#admitBlocked();
    }
    const first = this.#queued[0];
    if (!first) return [];

    const key = this.#batchKey(first.input);
    const batch: Array<PendingWrite<Input, Output>> = [];
    let batchBytes = 0;
    while (batch.length < this.#maxBatchSize) {
      const next = this.#queued[0];
      if (!next || !Object.is(this.#batchKey(next.input), key)) break;
      if (
        batch.length > 0 &&
        batchBytes + next.bytes > this.#targetBatchBytes
      ) {
        break;
      }
      this.#queued.shift();
      next.state = 'active';
      batch.push(next);
      batchBytes += next.bytes;
    }
    return batch;
  }

  async #sendWithRetry(
    batch: readonly PendingWrite<Input, Output>[],
    batchId: number,
  ): Promise<readonly Output[]> {
    const inputs = batch.map((pending) => pending.input);
    const startedAt = Date.now();
    let attempt = 0;
    while (true) {
      attempt += 1;
      if (this.#closeController.signal.aborted) {
        throw this.#closeReason;
      }
      if (batch.every((pending) => pending.aborted !== null)) {
        throw batch[0]?.aborted;
      }
      try {
        const outputs = await this.#send(inputs, {
          signal: this.#closeController.signal,
        });
        if (attempt > 1) {
          this.#emitRetryEvent({
            phase: 'recovered',
            attempt,
            batchId,
            batch,
            startedAt,
            retryAfterMs: 0,
          });
        }
        return outputs;
      } catch (error) {
        const retry = this.#classifyRetryableError(error);
        if (!retry) throw error;
        const elapsedMs = Date.now() - startedAt;
        if (elapsedMs >= this.#maxRetryElapsedMs) {
          throw new RuntimeReceiptWriterRetryDeadlineError(
            attempt,
            elapsedMs,
            this.#maxRetryElapsedMs,
            { cause: error },
          );
        }
        const retryAfterMs = jitteredRetryDelayMs(retry.retryAfterMs);
        if (elapsedMs + retryAfterMs > this.#maxRetryElapsedMs) {
          throw new RuntimeReceiptWriterRetryDeadlineError(
            attempt,
            elapsedMs,
            this.#maxRetryElapsedMs,
            { cause: error },
          );
        }
        this.#emitRetryEvent({
          phase: 'retry',
          attempt,
          batchId,
          batch,
          startedAt,
          retryAfterMs,
          error,
        });
        await this.#waitForRetry(retryAfterMs, batch);
      }
    }
  }

  #emitRetryEvent(input: {
    phase: 'retry' | 'recovered';
    attempt: number;
    batchId: number;
    batch: readonly PendingWrite<Input, Output>[];
    startedAt: number;
    retryAfterMs: number;
    error?: unknown;
  }): void {
    const firstInput = input.batch[0]?.input;
    if (!this.#onRetryEvent || firstInput === undefined) return;
    try {
      this.#onRetryEvent({
        phase: input.phase,
        attempt: input.attempt,
        batchId: input.batchId,
        batchSize: input.batch.length,
        batchBytes: input.batch.reduce(
          (total, pending) => total + pending.bytes,
          0,
        ),
        firstInput,
        elapsedMs: Date.now() - input.startedAt,
        retryAfterMs: input.retryAfterMs,
        queued: this.#queued.length,
        blocked: this.#blocked.length,
        bufferedBytes: this.#bufferedBytes,
        ...(input.error === undefined ? {} : { error: input.error }),
      });
    } catch {
      // Receipt telemetry must never affect durable delivery.
    }
  }

  async #waitForRetry(
    delayMs: number,
    batch: readonly PendingWrite<Input, Output>[],
  ): Promise<void> {
    if (delayMs === 0) {
      await Promise.resolve();
      return;
    }
    await new Promise<void>((resolve, reject) => {
      const signals = batch
        .map((pending) => pending.signal)
        .filter((signal): signal is AbortSignal => signal !== null);
      let settled = false;
      const finish = (error?: unknown): void => {
        if (settled) return;
        settled = true;
        clearTimeout(timer);
        this.#closeController.signal.removeEventListener('abort', onClose);
        for (const signal of signals) {
          signal.removeEventListener('abort', onCommandAbort);
        }
        if (error !== undefined) reject(error);
        else resolve();
      };
      const onClose = (): void => finish(this.#closeReason);
      const onCommandAbort = (): void => {
        if (batch.every((pending) => pending.aborted !== null)) {
          finish(batch[0]?.aborted);
        }
      };
      const timer = setTimeout(() => finish(), delayMs);
      this.#closeController.signal.addEventListener('abort', onClose, {
        once: true,
      });
      for (const signal of signals) {
        signal.addEventListener('abort', onCommandAbort, { once: true });
      }
      if (this.#closeController.signal.aborted) onClose();
      else onCommandAbort();
    });
  }

  #abortPending(pending: PendingWrite<Input, Output>): void {
    if (pending.state === 'settled') return;
    pending.aborted = abortReason(pending.signal);
    if (pending.state === 'active') return;

    const collection =
      pending.state === 'queued' ? this.#queued : this.#blocked;
    const index = collection.indexOf(pending);
    if (index >= 0) collection.splice(index, 1);
    if (pending.state === 'queued') {
      this.#bufferedBytes -= pending.bytes;
      this.#admitBlocked();
    }
    this.#settleRejected(pending, pending.aborted);
    this.#notifyDrained();
  }

  #admitBlocked(): void {
    while (this.#blocked.length > 0) {
      const pending = this.#blocked[0]!;
      if (pending.aborted !== null) {
        this.#blocked.shift();
        this.#settleRejected(pending, pending.aborted);
        continue;
      }
      if (!this.#canBuffer(pending.bytes)) break;
      this.#blocked.shift();
      this.#accept(pending);
    }
  }

  #settleResolved(pending: PendingWrite<Input, Output>, output: Output): void {
    if (pending.state === 'settled') return;
    this.#detachAbort(pending);
    pending.state = 'settled';
    pending.resolve(output);
  }

  #settleRejected(pending: PendingWrite<Input, Output>, error: unknown): void {
    if (pending.state === 'settled') return;
    this.#detachAbort(pending);
    pending.state = 'settled';
    pending.reject(error);
  }

  #detachAbort(pending: PendingWrite<Input, Output>): void {
    if (pending.signal && pending.onAbort) {
      pending.signal.removeEventListener('abort', pending.onAbort);
    }
  }

  #rejectWaiting(reason: unknown): void {
    this.#clearFlushTimer();
    for (const pending of this.#queued.splice(0)) {
      this.#bufferedBytes -= pending.bytes;
      this.#settleRejected(pending, reason);
    }
    for (const pending of this.#blocked.splice(0)) {
      this.#settleRejected(pending, reason);
    }
    this.#notifyDrained(reason);
  }

  #queuedBytesAtHead(): number {
    const first = this.#queued[0];
    if (!first) return 0;
    const key = this.#batchKey(first.input);
    let bytes = 0;
    let count = 0;
    for (const pending of this.#queued) {
      if (
        count >= this.#maxBatchSize ||
        !Object.is(this.#batchKey(pending.input), key)
      ) {
        break;
      }
      bytes += pending.bytes;
      count += 1;
    }
    return bytes;
  }

  #isDrained(): boolean {
    return (
      this.#activeBatch === null &&
      this.#queued.length === 0 &&
      this.#blocked.length === 0
    );
  }

  #notifyDrained(error?: unknown): void {
    if (!this.#isDrained() && error === undefined) return;
    const waiters = this.#drainWaiters.splice(0);
    for (const waiter of waiters) {
      if (error !== undefined) waiter.reject(error);
      else waiter.resolve();
    }
  }

  #clearFlushTimer(): void {
    if (!this.#flushTimer) return;
    clearTimeout(this.#flushTimer);
    this.#flushTimer = null;
  }
}

const textEncoder = new TextEncoder();

function estimateJsonBytes(input: unknown): number {
  const encoded = JSON.stringify(input);
  return textEncoder.encode(encoded ?? String(input)).byteLength;
}

function normalizePositiveInteger(
  value: number | undefined,
  fallback: number,
): number {
  return Math.max(1, Math.floor(value ?? fallback));
}

function jitteredRetryDelayMs(delayMs: number): number {
  const normalized = Math.max(0, Math.floor(delayMs));
  if (normalized === 0) return 0;
  return (
    normalized + Math.floor(normalized * MAX_RETRY_JITTER_RATIO * Math.random())
  );
}

function abortReason(signal: AbortSignal | null): unknown {
  return signal?.reason ?? new Error('Runtime receipt write was aborted.');
}