type RetryableRuntimeSheetWriteError = {
  retryAfterMs: number;
};

type Deferred = {
  promise: Promise<void>;
  resolve: () => void;
  reject: (error: unknown) => void;
};

type PendingTerminal<Row> = {
  kind: 'terminal';
  key: string;
  value: Row;
  bytes: number;
  committed: Deferred[];
};

type PendingCheckpoint<Update> = {
  kind: 'checkpoint';
  key: string;
  value: Update;
  bytes: number;
};

type PendingWrite<Row, Update> =
  | PendingTerminal<Row>
  | PendingCheckpoint<Update>;

type BlockedTerminal<Row> = {
  kind: 'terminal';
  key: string;
  value: Row;
  bytes: number;
  admitted: Deferred;
  committed: Deferred;
};

type BlockedCheckpoint<Update> = {
  kind: 'checkpoint';
  key: string;
  value: Update;
  bytes: number;
  admitted: Deferred;
};

type BlockedWrite<Row, Update> =
  | BlockedTerminal<Row>
  | BlockedCheckpoint<Update>;

export type RuntimeSheetRowWriterBatch<Row, Update> =
  | {
      kind: 'terminal';
      rows: readonly Row[];
    }
  | {
      kind: 'checkpoint';
      updates: readonly Update[];
    };

export type RuntimeSheetRowWriterBatchResult = {
  committed?: number;
  staleKeys?: readonly string[];
};

export type RuntimeSheetRowWriterSummary = {
  batches: number;
  checkpointRows: number;
  staleKeys: string[];
  terminalRows: number;
};

export type RuntimeSheetRowWriterDiagnostics = {
  activeKind: 'terminal' | 'checkpoint' | null;
  activeRows: number;
  activeStartedAt: number | null;
  blockedTerminalRows: number;
  queuedTerminalRows: number;
};

export type RuntimeSheetRowSettlement = {
  /**
   * Resolves once the bounded live buffer owns this row. Awaiting this promise
   * is the writer's admission backpressure seam; it does not wait for Neon.
   */
  admitted: Promise<void>;
  /** Resolves only after the batch containing the row commits durably. */
  committed: Promise<void>;
};

export type RuntimeSheetRowWriterOptions<Row, Update> = {
  terminalKey: (row: Row) => string;
  checkpointKey: (update: Update) => string;
  mergeCheckpointUpdates: (current: Update, incoming: Update) => Update;
  writeBatch: (
    batch: RuntimeSheetRowWriterBatch<Row, Update>,
    options: { signal: AbortSignal },
  ) => Promise<RuntimeSheetRowWriterBatchResult | void>;
  estimateTerminalBytes?: (row: Row) => number;
  estimateCheckpointBytes?: (update: Update) => number;
  maxBatchRows?: number;
  maxBatchBytes?: number;
  maxBufferedBytes?: number;
  maxFlushMs?: number;
  classifyRetryableError?: (
    error: unknown,
  ) => RetryableRuntimeSheetWriteError | null;
  sleep?: (ms: number, signal: AbortSignal) => Promise<void>;
  onFailure?: (error: unknown) => void;
};

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

export class RuntimeSheetRowWriterBufferLimitError extends Error {
  constructor(
    readonly inputBytes: number,
    readonly limitBytes: number,
  ) {
    super(
      `Runtime Sheet row write is ${inputBytes} bytes, exceeding the ` +
        `${limitBytes}-byte writer limit.`,
    );
    this.name = 'RuntimeSheetRowWriterBufferLimitError';
  }
}

/**
 * Per-map durable row writer.
 *
 * This Module has one live, keyed accumulator and at most one active Adapter
 * call. A timer, row limit, byte limit, checkpoint, or finish may start the
 * pump. Arrivals during an active write remain coalescible in the live
 * accumulator instead of becoming a chain of frozen future batches.
 */
export class RuntimeSheetRowWriter<Row, Update> {
  readonly #terminalKey: (row: Row) => string;
  readonly #checkpointKey: (update: Update) => string;
  readonly #mergeCheckpointUpdates: (
    current: Update,
    incoming: Update,
  ) => Update;
  readonly #writeBatch: RuntimeSheetRowWriterOptions<Row, Update>['writeBatch'];
  readonly #estimateTerminalBytes: (row: Row) => number;
  readonly #estimateCheckpointBytes: (update: Update) => number;
  readonly #maxBatchRows: number;
  readonly #maxBatchBytes: number;
  readonly #maxBufferedBytes: number;
  readonly #maxFlushMs: number;
  readonly #classifyRetryableError: NonNullable<
    RuntimeSheetRowWriterOptions<Row, Update>['classifyRetryableError']
  >;
  readonly #sleep: NonNullable<
    RuntimeSheetRowWriterOptions<Row, Update>['sleep']
  >;
  readonly #onFailure:
    | RuntimeSheetRowWriterOptions<Row, Update>['onFailure']
    | undefined;
  readonly #live = new Map<string, PendingWrite<Row, Update>>();
  readonly #blocked: Array<BlockedWrite<Row, Update>> = [];
  readonly #terminalAcceptedKeys = new Set<string>();
  readonly #staleKeys = new Set<string>();
  readonly #controller = new AbortController();

  #active: PendingWrite<Row, Update>[] | null = null;
  #activeStartedAt: number | null = null;
  #bufferedBytes = 0;
  #liveBytes = 0;
  #timer: ReturnType<typeof setTimeout> | null = null;
  #pump: Promise<void> | null = null;
  #checkpointPromise: Promise<RuntimeSheetRowWriterSummary> | null = null;
  #finishPromise: Promise<RuntimeSheetRowWriterSummary> | null = null;
  #stateWaiters: Deferred[] = [];
  #failure: unknown = null;
  #closed = false;
  #batches = 0;
  #checkpointRows = 0;
  #terminalRows = 0;

  constructor(options: RuntimeSheetRowWriterOptions<Row, Update>) {
    this.#terminalKey = options.terminalKey;
    this.#checkpointKey = options.checkpointKey;
    this.#mergeCheckpointUpdates = options.mergeCheckpointUpdates;
    this.#writeBatch = options.writeBatch;
    this.#estimateTerminalBytes =
      options.estimateTerminalBytes ?? estimateJsonBytes;
    this.#estimateCheckpointBytes =
      options.estimateCheckpointBytes ?? estimateJsonBytes;
    this.#maxBatchRows = positiveInteger(options.maxBatchRows, 100);
    this.#maxBufferedBytes = positiveInteger(
      options.maxBufferedBytes,
      32 * 1024 * 1024,
    );
    this.#maxBatchBytes = Math.min(
      positiveInteger(options.maxBatchBytes, 1024 * 1024),
      this.#maxBufferedBytes,
    );
    this.#maxFlushMs = Math.max(0, Math.floor(options.maxFlushMs ?? 500));
    this.#classifyRetryableError =
      options.classifyRetryableError ?? (() => null);
    this.#sleep = options.sleep ?? abortableSleep;
    this.#onFailure = options.onFailure;
  }

  settle(row: Row): RuntimeSheetRowSettlement {
    const unavailable = this.#unavailableError();
    if (unavailable !== null) return rejectedSettlement(unavailable);

    const key = normalizedKey(this.#terminalKey(row));
    const bytes = this.#inputBytes(this.#estimateTerminalBytes(row));
    const admitted = deferred();
    const committed = deferred();
    const blocked: BlockedTerminal<Row> = {
      kind: 'terminal',
      key,
      value: row,
      bytes,
      admitted,
      committed,
    };
    if (this.#canAdmit(blocked)) this.#admit(blocked);
    else this.#blocked.push(blocked);
    return {
      admitted: admitted.promise,
      committed: committed.promise,
    };
  }

  checkpoint(
    updates: readonly Update[] = [],
  ): Promise<RuntimeSheetRowWriterSummary> {
    const unavailable = this.#unavailableError();
    if (unavailable !== null) return Promise.reject(unavailable);
    if (this.#checkpointPromise && updates.length === 0) {
      return this.#checkpointPromise;
    }

    const admissions = updates.map((update) =>
      this.#enqueueCheckpointUpdate(update),
    );
    const previous = this.#checkpointPromise;
    const operation = (async () => {
      await Promise.all([...(previous ? [previous] : []), ...admissions]);
      await this.#drain();
      return this.#summary();
    })();
    this.#checkpointPromise = operation;
    const clearCheckpoint = () => {
      if (this.#checkpointPromise === operation) {
        this.#checkpointPromise = null;
      }
    };
    void operation.then(clearCheckpoint, clearCheckpoint);
    return operation;
  }

  finish(): Promise<RuntimeSheetRowWriterSummary> {
    if (this.#finishPromise) return this.#finishPromise;
    if (this.#failure !== null) return Promise.reject(this.#failure);

    // Close synchronously. No settlement can race in after the terminal
    // barrier has captured the writer's accepted and capacity-blocked work.
    this.#closed = true;
    this.#finishPromise = (async () => {
      if (this.#checkpointPromise) await this.#checkpointPromise;
      await this.#drain();
      return this.#summary();
    })();
    return this.#finishPromise;
  }

  diagnostics(): RuntimeSheetRowWriterDiagnostics {
    return {
      activeKind: this.#active?.[0]?.kind ?? null,
      activeRows: this.#active?.length ?? 0,
      activeStartedAt: this.#activeStartedAt,
      blockedTerminalRows: this.#blocked.filter(
        (entry) => entry.kind === 'terminal',
      ).length,
      queuedTerminalRows: [...this.#live.values()].filter(
        (entry) => entry.kind === 'terminal',
      ).length,
    };
  }

  #enqueueCheckpointUpdate(update: Update): Promise<void> {
    const key = normalizedKey(this.#checkpointKey(update));
    if (this.#terminalAcceptedKeys.has(key)) return Promise.resolve();
    const admitted = deferred();
    const blocked: BlockedCheckpoint<Update> = {
      kind: 'checkpoint',
      key,
      value: update,
      bytes: this.#inputBytes(this.#estimateCheckpointBytes(update)),
      admitted,
    };
    try {
      if (this.#canAdmit(blocked)) this.#admit(blocked);
      else this.#blocked.push(blocked);
    } catch (error) {
      // A merged checkpoint can exceed the item limit even when each patch
      // fits independently. Reject only the incoming checkpoint.
      admitted.reject(error);
    }
    return admitted.promise;
  }

  #canAdmit(write: BlockedWrite<Row, Update>): boolean {
    const existing = this.#live.get(write.key);
    const replacedBytes =
      existing && (existing.kind === write.kind || write.kind === 'terminal')
        ? existing.bytes
        : 0;
    const admittedBytes =
      write.kind === 'checkpoint' && existing?.kind === 'checkpoint'
        ? this.#inputBytes(
            this.#estimateCheckpointBytes(
              this.#mergeCheckpointUpdates(existing.value, write.value),
            ),
          )
        : write.bytes;
    return (
      this.#bufferedBytes - replacedBytes + admittedBytes <=
      this.#maxBufferedBytes
    );
  }

  #admit(write: BlockedWrite<Row, Update>): void {
    const existing = this.#live.get(write.key);
    if (write.kind === 'terminal') {
      this.#terminalAcceptedKeys.add(write.key);
      if (existing) {
        this.#liveBytes -= existing.bytes;
        this.#bufferedBytes -= existing.bytes;
      }
      const committed =
        existing?.kind === 'terminal'
          ? [...existing.committed, write.committed]
          : [write.committed];
      const terminal: PendingTerminal<Row> = {
        kind: 'terminal',
        key: write.key,
        value: write.value,
        bytes: write.bytes,
        committed,
      };
      this.#live.set(write.key, terminal);
    } else if (this.#terminalAcceptedKeys.has(write.key)) {
      write.admitted.resolve();
      return;
    } else if (existing?.kind === 'checkpoint') {
      const merged = this.#mergeCheckpointUpdates(existing.value, write.value);
      let mergedBytes: number;
      try {
        mergedBytes = this.#inputBytes(this.#estimateCheckpointBytes(merged));
      } catch (error) {
        write.admitted.reject(error);
        return;
      }
      this.#liveBytes -= existing.bytes;
      this.#bufferedBytes -= existing.bytes;
      const checkpoint: PendingCheckpoint<Update> = {
        kind: 'checkpoint',
        key: write.key,
        value: merged,
        bytes: mergedBytes,
      };
      this.#live.set(write.key, checkpoint);
      write = { ...write, bytes: checkpoint.bytes };
    } else {
      const checkpoint: PendingCheckpoint<Update> = {
        kind: 'checkpoint',
        key: write.key,
        value: write.value,
        bytes: write.bytes,
      };
      this.#live.set(write.key, checkpoint);
    }
    this.#liveBytes += write.bytes;
    this.#bufferedBytes += write.bytes;
    write.admitted.resolve();
    this.#scheduleOrStart();
    this.#notifyStateChanged();
  }

  #scheduleOrStart(): void {
    if (this.#pump || this.#live.size === 0 || this.#failure !== null) return;
    if (
      this.#live.size >= this.#maxBatchRows ||
      this.#liveBytes >= this.#maxBatchBytes
    ) {
      this.#clearTimer();
      this.#startPump();
      return;
    }
    if (this.#timer) return;
    this.#timer = setTimeout(() => {
      this.#timer = null;
      this.#startPump();
    }, this.#maxFlushMs);
  }

  #startPump(): void {
    if (this.#pump || this.#live.size === 0 || this.#failure !== null) {
      return;
    }
    this.#pump = this.#runPump().finally(() => {
      this.#pump = null;
      this.#notifyStateChanged();
      if (this.#failure === null && this.#live.size > 0) {
        // Rows accumulated behind an active write are real backlog. Drain the
        // largest available next batch immediately instead of starting a new
        // age window and manufacturing latency.
        this.#startPump();
      }
    });
  }

  async #runPump(): Promise<void> {
    while (this.#live.size > 0 && this.#failure === null) {
      const batch = this.#takeBatch();
      if (batch.length === 0) return;
      const batchKind = batch[0]!.kind;
      this.#active = batch;
      this.#activeStartedAt = Date.now();
      try {
        const result = await this.#writeIdenticalBatchWithRetry(batch);
        const resultCommitted = result ? result.committed : undefined;
        const resultStaleKeys = result ? result.staleKeys : undefined;
        this.#batches += 1;
        const committed = resultCommitted ?? batch.length;
        if (batchKind === 'terminal') this.#terminalRows += committed;
        else this.#checkpointRows += committed;
        for (const staleKey of resultStaleKeys ?? []) {
          this.#staleKeys.add(staleKey);
        }
        for (const entry of batch) {
          if (entry.kind === 'terminal') {
            entry.committed.forEach((waiter) => waiter.resolve());
          }
        }
      } catch (error) {
        this.#latchFailure(error);
        return;
      } finally {
        const releasedBytes = batch.reduce(
          (total, entry) => total + entry.bytes,
          0,
        );
        this.#bufferedBytes = Math.max(0, this.#bufferedBytes - releasedBytes);
        this.#active = null;
        this.#activeStartedAt = null;
        this.#admitBlocked();
        this.#notifyStateChanged();
      }
    }
  }

  #takeBatch(): PendingWrite<Row, Update>[] {
    const first = this.#live.values().next().value as
      | PendingWrite<Row, Update>
      | undefined;
    if (!first) return [];
    const batch: PendingWrite<Row, Update>[] = [];
    const batchKind = first.kind;
    let bytes = 0;
    for (const [key, entry] of this.#live) {
      if (entry.kind !== batchKind) continue;
      if (batch.length >= this.#maxBatchRows) break;
      if (batch.length > 0 && bytes + entry.bytes > this.#maxBatchBytes) break;
      this.#live.delete(key);
      this.#liveBytes -= entry.bytes;
      batch.push(entry);
      bytes += entry.bytes;
    }
    return batch;
  }

  async #writeIdenticalBatchWithRetry(
    entries: readonly PendingWrite<Row, Update>[],
  ): Promise<RuntimeSheetRowWriterBatchResult | void> {
    const batch: RuntimeSheetRowWriterBatch<Row, Update> =
      entries[0]?.kind === 'checkpoint'
        ? {
            kind: 'checkpoint',
            updates: Object.freeze(
              entries
                .filter(
                  (entry): entry is PendingCheckpoint<Update> =>
                    entry.kind === 'checkpoint',
                )
                .map((entry) => entry.value),
            ),
          }
        : {
            kind: 'terminal',
            rows: Object.freeze(
              entries
                .filter(
                  (entry): entry is PendingTerminal<Row> =>
                    entry.kind === 'terminal',
                )
                .map((entry) => entry.value),
            ),
          };
    for (;;) {
      try {
        return await this.#writeBatch(batch, {
          signal: this.#controller.signal,
        });
      } catch (error) {
        const retry = this.#classifyRetryableError(error);
        if (!retry) throw error;
        await this.#sleep(
          Math.max(0, Math.ceil(retry.retryAfterMs)),
          this.#controller.signal,
        );
      }
    }
  }

  #admitBlocked(): void {
    for (let index = 0; index < this.#blocked.length; ) {
      const write = this.#blocked[index]!;
      let canAdmit = false;
      try {
        canAdmit = this.#canAdmit(write);
      } catch (error) {
        this.#blocked.splice(index, 1);
        write.admitted.reject(error);
        continue;
      }
      if (!canAdmit) {
        index += 1;
        continue;
      }
      this.#blocked.splice(index, 1);
      this.#admit(write);
    }
  }

  async #drain(): Promise<void> {
    this.#clearTimer();
    for (;;) {
      if (this.#failure !== null) throw this.#failure;
      this.#admitBlocked();
      if (
        this.#active === null &&
        this.#live.size === 0 &&
        this.#blocked.length === 0
      ) {
        return;
      }
      this.#startPump();
      const stateChanged = deferred();
      this.#stateWaiters.push(stateChanged);
      await stateChanged.promise;
    }
  }

  #latchFailure(error: unknown): void {
    if (this.#failure !== null) return;
    this.#failure = error;
    this.#clearTimer();
    for (const entry of this.#active ?? []) {
      if (entry.kind === 'terminal') {
        entry.committed.forEach((waiter) => waiter.reject(error));
      }
    }
    for (const entry of this.#live.values()) {
      if (entry.kind === 'terminal') {
        entry.committed.forEach((waiter) => waiter.reject(error));
      }
    }
    this.#live.clear();
    this.#liveBytes = 0;
    for (const blocked of this.#blocked.splice(0)) {
      blocked.admitted.reject(error);
      if (blocked.kind === 'terminal') blocked.committed.reject(error);
    }
    this.#onFailure?.(error);
    this.#notifyStateChanged(error);
  }

  #unavailableError(): unknown | null {
    if (this.#failure !== null) return this.#failure;
    return this.#closed ? new RuntimeSheetRowWriterClosedError() : null;
  }

  #inputBytes(estimated: number): number {
    const bytes = Math.max(0, Math.ceil(estimated));
    if (bytes > this.#maxBatchBytes || bytes > this.#maxBufferedBytes) {
      throw new RuntimeSheetRowWriterBufferLimitError(
        bytes,
        Math.min(this.#maxBatchBytes, this.#maxBufferedBytes),
      );
    }
    return bytes;
  }

  #summary(): RuntimeSheetRowWriterSummary {
    return {
      batches: this.#batches,
      checkpointRows: this.#checkpointRows,
      staleKeys: [...this.#staleKeys],
      terminalRows: this.#terminalRows,
    };
  }

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

  #notifyStateChanged(error?: unknown): void {
    const waiters = this.#stateWaiters.splice(0);
    if (error === undefined) {
      for (const waiter of waiters) {
        waiter.resolve();
      }
      return;
    }
    for (const waiter of waiters) {
      waiter.reject(error);
    }
  }
}

function deferred(): Deferred {
  let resolve!: () => void;
  let reject!: (error: unknown) => void;
  const promise = new Promise<void>((resolvePromise, rejectPromise) => {
    resolve = resolvePromise;
    reject = rejectPromise;
  });
  return { promise, reject, resolve };
}

function rejectedSettlement(error: unknown): RuntimeSheetRowSettlement {
  const rejected = Promise.reject<void>(error);
  return { admitted: rejected, committed: rejected };
}

function normalizedKey(value: string): string {
  const key = value.trim();
  if (!key) throw new Error('Runtime Sheet row writer requires a row key.');
  return key;
}

function positiveInteger(value: number | undefined, fallback: number): number {
  if (typeof value !== 'number' || !Number.isFinite(value) || value <= 0) {
    return fallback;
  }
  return Math.max(1, Math.floor(value));
}

function estimateJsonBytes(value: unknown): number {
  return new TextEncoder().encode(JSON.stringify(value)).byteLength;
}

async function abortableSleep(ms: number, signal: AbortSignal): Promise<void> {
  if (ms <= 0) {
    await Promise.resolve();
    return;
  }
  await new Promise<void>((resolve, reject) => {
    let settled = false;
    const finish = (error?: unknown) => {
      if (settled) return;
      settled = true;
      clearTimeout(timer);
      signal.removeEventListener('abort', onAbort);
      if (error === undefined) resolve();
      else reject(error);
    };
    const onAbort = () =>
      finish(signal.reason ?? new Error('Runtime Sheet row writer aborted.'));
    const timer = setTimeout(() => finish(), ms);
    signal.addEventListener('abort', onAbort, { once: true });
    if (signal.aborted) onAbort();
  });
}