import type {
  AcquireRuntimeReceiptExecutionLockInput,
  ClaimRuntimeStepReceiptInput,
  CompleteRuntimeStepReceiptInput,
  ContextOptions,
  FailRuntimeStepReceiptInput,
  ReleaseRuntimeReceiptExecutionLockInput,
  ReleaseRuntimeStepReceiptInput,
  RuntimeStepReceipt,
  SkipRuntimeStepReceiptInput,
} from './ctx-types';
import {
  RUNTIME_RECEIPT_GATEWAY_BATCH_MAX_BYTES,
  RUNTIME_RECEIPT_WRITER_TARGET_BATCH_BYTES,
} from './output-size-limits';
import {
  acquireRuntimeReceiptExecutionLockViaAppRuntime,
  AppRuntimeApiResponseError,
  AppRuntimeApiTransportError,
  claimRuntimeStepReceiptsViaAppRuntime,
  completeRuntimeStepReceiptsViaAppRuntime,
  failRuntimeStepReceiptsViaAppRuntime,
  getRuntimeStepReceiptsViaAppRuntime,
  heartbeatRuntimeStepReceiptsViaAppRuntime,
  isAppRuntimeApiCapacityError,
  releaseRuntimeReceiptExecutionLockViaAppRuntime,
  releaseRuntimeStepReceiptViaAppRuntime,
  skipRuntimeStepReceiptViaAppRuntime,
  type WorkerRuntimeApiContext,
} from './app-runtime-api';
import {
  RuntimeReceiptWriter,
  RuntimeReceiptWriterResultCountError,
} from './runtime-receipt-writer';

type ReceiptOutput =
  | RuntimeStepReceipt
  | null
  | boolean
  | { ownerExecutionId: string; expiresAt: string };

type ReceiptCommand =
  | {
      kind: 'get';
      input: { key: string; runId: string };
    }
  | {
      kind: 'claim';
      input: ClaimRuntimeStepReceiptInput;
    }
  | {
      kind: 'settle';
      outcome: 'complete';
      input: CompleteRuntimeStepReceiptInput;
    }
  | {
      kind: 'settle';
      outcome: 'fail';
      input: FailRuntimeStepReceiptInput;
    }
  | {
      kind: 'heartbeat';
      input: {
        key: string;
        runId: string;
        runAttempt?: number | null;
        leaseId: string;
      };
    }
  | {
      kind: 'settle';
      outcome: 'release';
      input: ReleaseRuntimeStepReceiptInput;
      serial: number;
    }
  | {
      kind: 'settle';
      outcome: 'skip';
      input: SkipRuntimeStepReceiptInput;
      serial: number;
    }
  | {
      kind: 'acquire_execution_lock';
      input: AcquireRuntimeReceiptExecutionLockInput;
      serial: number;
    }
  | {
      kind: 'release_execution_lock';
      input: ReleaseRuntimeReceiptExecutionLockInput;
      serial: number;
    };

type ReceiptHandlerName =
  | 'acquireRuntimeReceiptExecutionLock'
  | 'releaseRuntimeReceiptExecutionLock'
  | 'getRuntimeStepReceipt'
  | 'getRuntimeStepReceipts'
  | 'claimRuntimeStepReceipt'
  | 'claimRuntimeStepReceipts'
  | 'completeRuntimeStepReceipt'
  | 'completeRuntimeStepReceipts'
  | 'releaseRuntimeStepReceipt'
  | 'failRuntimeStepReceipt'
  | 'failRuntimeStepReceipts'
  | 'heartbeatRuntimeStepReceipts'
  | 'skipRuntimeStepReceipt';

export type RuntimeReceiptStoreHandlers = Required<
  Pick<ContextOptions, ReceiptHandlerName>
>;

// Four ambiguous 30-second receipt requests previously consumed the writer's
// entire two-minute budget. Receipt operations are idempotent and buffered, so
// keep retry ownership here and allow independent transports enough attempts
// to recover from a short gateway/network impairment.
export const RUNTIME_RECEIPT_STORE_RETRY_BUDGET_MS = 5 * 60_000;

/**
 * The runner's single Work Receipt Adapter.
 *
 * Receipt-domain batching and gateway transport stay behind this seam. Every
 * command shares one writer, so heterogeneous operations cannot produce
 * concurrent gateway requests. The low-level runtime client is explicitly
 * one-attempt; this Adapter is the sole retry owner.
 */
export class RuntimeReceiptStoreAdapter {
  readonly handlers: RuntimeReceiptStoreHandlers;
  readonly claimsEstablishExecutionFence = true;
  readonly #writer: RuntimeReceiptWriter<ReceiptCommand, ReceiptOutput>;
  #serial = 0;

  constructor(options: {
    context: WorkerRuntimeApiContext;
    playName: string;
    runId: string;
    maxBatchSize?: number;
    maxBatchBytes?: number;
    targetBatchBytes?: number;
    maxFlushMs?: number;
    maxBufferedBytes?: number;
    onRetryTelemetry?: (line: string) => void;
  }) {
    const context: WorkerRuntimeApiContext = {
      ...options.context,
      retryPolicy: 'none',
    };
    this.#writer = new RuntimeReceiptWriter({
      batchKey: receiptCommandBatchKey,
      maxBatchSize: options.maxBatchSize,
      maxBatchBytes:
        options.maxBatchBytes ?? RUNTIME_RECEIPT_GATEWAY_BATCH_MAX_BYTES,
      targetBatchBytes:
        options.targetBatchBytes ?? RUNTIME_RECEIPT_WRITER_TARGET_BATCH_BYTES,
      maxFlushMs: options.maxFlushMs,
      maxBufferedBytes: options.maxBufferedBytes,
      maxRetryElapsedMs: RUNTIME_RECEIPT_STORE_RETRY_BUDGET_MS,
      classifyRetryableError: classifyReceiptRetry,
      onRetryEvent: (event) =>
        emitReceiptWriterRetryTelemetry({
          event,
          playName: options.playName,
          runId: options.runId,
          gatewayHost: runtimeGatewayHost(context.baseUrl),
          onTelemetry: options.onRetryTelemetry,
        }),
      send: async (commands, { signal }) =>
        await sendReceiptCommands({
          context: { ...context, signal },
          playName: options.playName,
          commands,
        }),
    });

    const write = <Output extends ReceiptOutput>(
      command: ReceiptCommand,
    ): Promise<Output> =>
      this.#writer.write(command, {
        signal: options.context.signal,
      }) as Promise<Output>;
    const writeMany = <Output extends ReceiptOutput>(
      commands: readonly ReceiptCommand[],
    ): Promise<Output[]> =>
      Promise.all(commands.map((command) => write<Output>(command)));

    this.handlers = {
      acquireRuntimeReceiptExecutionLock: (input) =>
        write({
          kind: 'acquire_execution_lock',
          input,
          serial: this.#nextSerial(),
        }),
      releaseRuntimeReceiptExecutionLock: (input) =>
        write({
          kind: 'release_execution_lock',
          input,
          serial: this.#nextSerial(),
        }),
      getRuntimeStepReceipt: ({ key }) =>
        write({
          kind: 'get',
          input: { key, runId: options.runId },
        }),
      getRuntimeStepReceipts: ({ keys }) =>
        writeMany(
          keys.map((key) => ({
            kind: 'get' as const,
            input: { key, runId: options.runId },
          })),
        ),
      claimRuntimeStepReceipt: (input) => write({ kind: 'claim', input }),
      claimRuntimeStepReceipts: (input) =>
        writeMany(
          input.keys.map((key, index) => ({
            kind: 'claim' as const,
            input: {
              key,
              runId: input.runId,
              runAttempt: input.runAttempt,
              leaseAware: input.leaseAware,
              reclaimRunning: input.reclaimRunning,
              forceRefresh: input.forceRefresh,
              forceFailedRefresh: input.forceFailedRefresh,
              ...(input.leaseIds ? { leaseId: input.leaseIds[index] } : {}),
            },
          })),
        ),
      completeRuntimeStepReceipt: (input) =>
        write({ kind: 'settle', outcome: 'complete', input }),
      completeRuntimeStepReceipts: ({ receipts }) =>
        writeMany(
          receipts.map((input) => ({
            kind: 'settle' as const,
            outcome: 'complete' as const,
            input,
          })),
        ),
      releaseRuntimeStepReceipt: (input) =>
        write({
          kind: 'settle',
          outcome: 'release',
          input,
          serial: this.#nextSerial(),
        }),
      failRuntimeStepReceipt: (input) =>
        write({ kind: 'settle', outcome: 'fail', input }),
      failRuntimeStepReceipts: ({ receipts }) =>
        writeMany(
          receipts.map((input) => ({
            kind: 'settle' as const,
            outcome: 'fail' as const,
            input,
          })),
        ),
      heartbeatRuntimeStepReceipts: (input) =>
        writeMany(
          input.keys.map((key, index) => ({
            kind: 'heartbeat' as const,
            input: {
              key,
              runId: input.runId,
              runAttempt: input.runAttempt,
              leaseId: input.leaseIds?.[index] ?? input.leaseId ?? '',
            },
          })),
        ),
      skipRuntimeStepReceipt: (input) =>
        write({
          kind: 'settle',
          outcome: 'skip',
          input,
          serial: this.#nextSerial(),
        }),
    };
  }

  flush(): Promise<void> {
    return this.#writer.flush();
  }

  close(reason?: unknown): Promise<void> {
    return this.#writer.close(reason);
  }

  #nextSerial(): number {
    return ++this.#serial;
  }
}

const RECEIPT_WRITER_RETRY_TELEMETRY_TAG =
  '[perf][worker.receipt_writer.retry]';

function isPowerOfTwo(value: number): boolean {
  return value > 0 && (value & (value - 1)) === 0;
}

function runtimeGatewayHost(baseUrl: string): string | null {
  try {
    return new URL(baseUrl).host || null;
  } catch {
    return null;
  }
}

function receiptCommandTelemetryAction(command: ReceiptCommand): string {
  return command.kind === 'settle'
    ? `${command.kind}_${command.outcome}`
    : command.kind;
}

function receiptRetryErrorTelemetry(error: unknown): Record<string, unknown> {
  if (isAppRuntimeApiCapacityError(error)) {
    return {
      failureKind: 'capacity',
      errorName: error.name,
      status: error.status,
      code: error.code,
      requestId: error.requestId,
    };
  }
  if (error instanceof AppRuntimeApiTransportError) {
    return {
      failureKind: 'transport',
      errorName: error.name,
      transportAttempts: error.attempts,
      transportAction: error.action,
      transportAttemptId: error.transportAttemptId,
    };
  }
  if (error instanceof AppRuntimeApiResponseError) {
    return {
      failureKind: 'http',
      errorName: error.name,
      status: error.status,
      code: error.code,
      requestId: error.requestId,
    };
  }
  return {
    failureKind: 'unknown',
    errorName: error instanceof Error ? error.name : typeof error,
  };
}

function emitReceiptWriterRetryTelemetry(input: {
  event: import('./runtime-receipt-writer').RuntimeReceiptWriterRetryEvent<ReceiptCommand>;
  playName: string;
  runId: string;
  gatewayHost: string | null;
  onTelemetry?: (line: string) => void;
}): void {
  const { event } = input;
  if (event.phase === 'retry' && !isPowerOfTwo(event.attempt)) return;
  const payload = {
    phase: event.phase,
    playName: input.playName,
    runId: input.runId,
    gatewayHost: input.gatewayHost,
    action: receiptCommandTelemetryAction(event.firstInput),
    batchId: event.batchId,
    batchSize: event.batchSize,
    batchBytes: event.batchBytes,
    attempt: event.attempt,
    elapsedMs: event.elapsedMs,
    retryAfterMs: event.retryAfterMs,
    queued: event.queued,
    blocked: event.blocked,
    bufferedBytes: event.bufferedBytes,
    ...(event.error === undefined
      ? { failureKind: null }
      : receiptRetryErrorTelemetry(event.error)),
  };
  try {
    const serialized = JSON.stringify(payload);
    input.onTelemetry?.(`${RECEIPT_WRITER_RETRY_TELEMETRY_TAG} ${serialized}`);
    if (event.phase === 'retry') {
      console.warn(RECEIPT_WRITER_RETRY_TELEMETRY_TAG, serialized);
    } else {
      console.info(RECEIPT_WRITER_RETRY_TELEMETRY_TAG, serialized);
    }
  } catch {
    // Receipt telemetry must never affect durable delivery.
  }
}

function receiptCommandBatchKey(command: ReceiptCommand): string {
  switch (command.kind) {
    case 'get':
      return JSON.stringify([command.kind, command.input.runId]);
    case 'claim':
      return JSON.stringify([
        command.kind,
        command.input.runId,
        command.input.runAttempt ?? null,
        command.input.leaseAware === true,
        command.input.reclaimRunning === true,
        command.input.forceRefresh === true,
        command.input.forceFailedRefresh === true,
        command.input.leaseId !== undefined,
      ]);
    case 'settle':
      return command.outcome === 'complete' || command.outcome === 'fail'
        ? JSON.stringify([command.kind, command.outcome, command.input.runId])
        : `${command.kind}:${command.outcome}:${command.serial}`;
    case 'heartbeat':
      return JSON.stringify([
        command.kind,
        command.input.runId,
        command.input.runAttempt ?? null,
      ]);
    case 'acquire_execution_lock':
    case 'release_execution_lock':
      return `${command.kind}:${command.serial}`;
  }
}

function classifyReceiptRetry(error: unknown): { retryAfterMs: number } | null {
  if (isAppRuntimeApiCapacityError(error)) {
    return { retryAfterMs: error.retryAfterMs };
  }
  if (error instanceof AppRuntimeApiTransportError) {
    return { retryAfterMs: 250 };
  }
  if (error instanceof AppRuntimeApiResponseError && error.retryable) {
    return { retryAfterMs: 250 };
  }
  return null;
}

async function sendReceiptCommands(input: {
  context: WorkerRuntimeApiContext;
  playName: string;
  commands: readonly ReceiptCommand[];
}): Promise<readonly ReceiptOutput[]> {
  const first = input.commands[0];
  if (!first) return [];
  switch (first.kind) {
    case 'get': {
      const commands = input.commands as readonly Extract<
        ReceiptCommand,
        { kind: 'get' }
      >[];
      return await getRuntimeStepReceiptsViaAppRuntime(input.context, {
        playName: input.playName,
        runId: first.input.runId,
        keys: commands.map((command) => command.input.key),
      });
    }
    case 'claim': {
      const commands = input.commands as readonly Extract<
        ReceiptCommand,
        { kind: 'claim' }
      >[];
      return await claimRuntimeStepReceiptsViaAppRuntime(input.context, {
        playName: input.playName,
        runId: first.input.runId,
        runAttempt: first.input.runAttempt,
        keys: commands.map((command) => command.input.key),
        leaseIds:
          first.input.leaseId === undefined
            ? undefined
            : commands.map((command) => command.input.leaseId!),
        leaseAware: first.input.leaseAware,
        reclaimRunning: first.input.reclaimRunning,
        forceRefresh: first.input.forceRefresh,
        forceFailedRefresh: first.input.forceFailedRefresh,
      });
    }
    case 'settle':
      switch (first.outcome) {
        case 'complete': {
          const commands = input.commands as readonly Extract<
            ReceiptCommand,
            { kind: 'settle'; outcome: 'complete' }
          >[];
          return await completeRuntimeStepReceiptsViaAppRuntime(input.context, {
            playName: input.playName,
            runId: first.input.runId,
            receipts: commands.map((command) => command.input),
          });
        }
        case 'fail': {
          const commands = input.commands as readonly Extract<
            ReceiptCommand,
            { kind: 'settle'; outcome: 'fail' }
          >[];
          return await failRuntimeStepReceiptsViaAppRuntime(input.context, {
            playName: input.playName,
            runId: first.input.runId,
            receipts: commands.map((command) => command.input),
          });
        }
        case 'release':
          return [
            await releaseRuntimeStepReceiptViaAppRuntime(input.context, {
              playName: input.playName,
              ...first.input,
            }),
          ];
        case 'skip':
          return [
            await skipRuntimeStepReceiptViaAppRuntime(input.context, {
              playName: input.playName,
              ...first.input,
            }),
          ];
      }
    case 'heartbeat': {
      const commands = input.commands as readonly Extract<
        ReceiptCommand,
        { kind: 'heartbeat' }
      >[];
      const uniqueCommands: (typeof commands)[number][] = [];
      const uniqueIndexByIdentity = new Map<string, number>();
      const originalToUniqueIndex = commands.map((command) => {
        const identity = JSON.stringify([
          command.input.key,
          command.input.leaseId,
        ]);
        const existingIndex = uniqueIndexByIdentity.get(identity);
        if (existingIndex !== undefined) return existingIndex;
        const uniqueIndex = uniqueCommands.length;
        uniqueCommands.push(command);
        uniqueIndexByIdentity.set(identity, uniqueIndex);
        return uniqueIndex;
      });
      const leaseIds = uniqueCommands.map((command) => command.input.leaseId);
      const sharedLeaseId = leaseIds.every((leaseId) => leaseId === leaseIds[0])
        ? leaseIds[0]
        : undefined;
      const uniqueResults = await heartbeatRuntimeStepReceiptsViaAppRuntime(
        input.context,
        {
          playName: input.playName,
          runId: first.input.runId,
          runAttempt: first.input.runAttempt,
          ...(sharedLeaseId ? { leaseId: sharedLeaseId } : { leaseIds }),
          keys: uniqueCommands.map((command) => command.input.key),
        },
      );
      if (uniqueResults.length !== uniqueCommands.length) {
        throw new RuntimeReceiptWriterResultCountError(
          uniqueCommands.length,
          uniqueResults.length,
        );
      }
      return originalToUniqueIndex.map(
        (uniqueIndex) => uniqueResults[uniqueIndex]!,
      );
    }
    case 'acquire_execution_lock':
      return [
        await acquireRuntimeReceiptExecutionLockViaAppRuntime(input.context, {
          playName: input.playName,
          ...first.input,
        }),
      ];
    case 'release_execution_lock':
      return [
        await releaseRuntimeReceiptExecutionLockViaAppRuntime(input.context, {
          playName: input.playName,
          ...first.input,
        }),
      ];
  }
}