export type RuntimeSheetRowStatus =
  | 'pending'
  | 'running'
  | 'failed'
  | 'enriched'
  | 'stale';

export type RuntimeSheetAttemptIdentity = {
  attemptId: string | null;
  attemptOwnerRunId: string | null;
  attemptExpiresAt: string | null;
  attemptSeq?: number | null;
};

export type RuntimeSheetRowAttemptInput = RuntimeSheetAttemptIdentity & {
  status: RuntimeSheetRowStatus;
  runId: string | null;
};

export type RuntimeSheetAttemptRelation =
  | { kind: 'unowned' }
  | { kind: 'same_attempt' }
  | { kind: 'expired' }
  | { kind: 'same_owner_newer_attempt' }
  | { kind: 'same_owner_stale_attempt' }
  | { kind: 'other_active_owner' }
  | {
      kind: 'malformed';
      reason: 'invalid_row_expiry';
    };

export type RuntimeSheetPrepareDecision =
  | { kind: 'claim_executable' }
  | { kind: 'reuse_completed' }
  | { kind: 'blocked_by_active_owner' };

export type RuntimeSheetTerminalDecision =
  | { kind: 'write_terminal' }
  | { kind: 'reject_stale_attempt' };

export type RuntimeSheetReleaseDecision =
  | { kind: 'release' }
  | { kind: 'skip' };

function normalize(value: string | null | undefined): string | null {
  const trimmed = value?.trim();
  return trimmed ? trimmed : null;
}

function parseOptionalMs(value: string | null): number | null {
  if (!value) return null;
  const parsed = Date.parse(value);
  return Number.isFinite(parsed) ? parsed : Number.NaN;
}

function normalizeAttemptSeq(value: number | null | undefined): number {
  return Number.isFinite(value) ? Math.trunc(value as number) : 0;
}

function isSameOwnerOlderRequest(input: {
  row: RuntimeSheetRowAttemptInput;
  request: RuntimeSheetAttemptIdentity;
}): boolean {
  const rowOwner =
    normalize(input.row.attemptOwnerRunId) ?? normalize(input.row.runId);
  const requestOwner = normalize(input.request.attemptOwnerRunId);
  return (
    Boolean(rowOwner && requestOwner && rowOwner === requestOwner) &&
    normalizeAttemptSeq(input.request.attemptSeq) <
      normalizeAttemptSeq(input.row.attemptSeq)
  );
}

export function classifyRuntimeSheetAttemptRelation(input: {
  row: RuntimeSheetRowAttemptInput;
  request: RuntimeSheetAttemptIdentity;
  nowMs?: number;
}): RuntimeSheetAttemptRelation {
  const nowMs = input.nowMs ?? Date.now();
  const rowAttemptId = normalize(input.row.attemptId);
  const requestAttemptId = normalize(input.request.attemptId);
  if (!rowAttemptId) return { kind: 'unowned' };
  if (rowAttemptId === requestAttemptId) return { kind: 'same_attempt' };

  const rowExpiresAt = normalize(input.row.attemptExpiresAt);
  const rowExpiresAtMs = parseOptionalMs(rowExpiresAt);
  if (rowExpiresAtMs === null) {
    return { kind: 'malformed', reason: 'invalid_row_expiry' };
  }
  if (Number.isNaN(rowExpiresAtMs)) {
    return { kind: 'malformed', reason: 'invalid_row_expiry' };
  }
  if (rowExpiresAtMs <= nowMs) return { kind: 'expired' };

  const rowOwner =
    normalize(input.row.attemptOwnerRunId) ?? normalize(input.row.runId);
  const requestOwner = normalize(input.request.attemptOwnerRunId);
  const rowAttemptSeq = normalizeAttemptSeq(input.row.attemptSeq);
  const requestAttemptSeq = normalizeAttemptSeq(input.request.attemptSeq);
  if (rowOwner && requestOwner && rowOwner === requestOwner) {
    return requestAttemptSeq > rowAttemptSeq
      ? { kind: 'same_owner_newer_attempt' }
      : { kind: 'same_owner_stale_attempt' };
  }

  return { kind: 'other_active_owner' };
}

export function decideRuntimeSheetPrepare(input: {
  row: RuntimeSheetRowAttemptInput;
  request: RuntimeSheetAttemptIdentity;
  missingRequiredOutput?: boolean;
  nowMs?: number;
}): RuntimeSheetPrepareDecision {
  if (input.row.status === 'stale' || input.row.status === 'failed') {
    return { kind: 'claim_executable' };
  }
  const relation = classifyRuntimeSheetAttemptRelation(input);
  if (input.row.status === 'pending' || input.row.status === 'running') {
    return (relation.kind !== 'expired' && isSameOwnerOlderRequest(input)) ||
      relation.kind === 'malformed'
      ? { kind: 'blocked_by_active_owner' }
      : { kind: 'claim_executable' };
  }
  if (input.row.status === 'enriched') {
    if (!input.missingRequiredOutput) return { kind: 'reuse_completed' };
    return (relation.kind !== 'expired' && isSameOwnerOlderRequest(input)) ||
      relation.kind === 'malformed'
      ? { kind: 'blocked_by_active_owner' }
      : { kind: 'claim_executable' };
  }
  return assertNever(input.row.status);
}

export function decideRuntimeSheetTerminalWrite(input: {
  row: RuntimeSheetRowAttemptInput;
  request: RuntimeSheetAttemptIdentity;
  nowMs?: number;
}): RuntimeSheetTerminalDecision {
  const relation = classifyRuntimeSheetAttemptRelation(input);
  if (input.row.status !== 'enriched') {
    return (relation.kind !== 'expired' && isSameOwnerOlderRequest(input)) ||
      relation.kind === 'malformed'
      ? { kind: 'reject_stale_attempt' }
      : { kind: 'write_terminal' };
  }
  if (
    relation.kind === 'unowned' ||
    relation.kind === 'same_attempt' ||
    relation.kind === 'expired' ||
    relation.kind === 'same_owner_newer_attempt'
  ) {
    return { kind: 'write_terminal' };
  }
  if (input.row.status === 'enriched') {
    const rowAttemptSeq = normalizeAttemptSeq(input.row.attemptSeq);
    const requestAttemptSeq = normalizeAttemptSeq(input.request.attemptSeq);
    if (requestAttemptSeq > rowAttemptSeq) {
      return { kind: 'write_terminal' };
    }
  }
  return { kind: 'reject_stale_attempt' };
}

/**
 * Decide whether a controlled run-fatal teardown may release (clear the attempt
 * lease on) a row. Only rows the releasing attempt currently owns are cleared:
 * the row must carry an attempt lease whose owner run id and attempt generation
 * (seq) match the releasing run. When the request carries a concrete attempt id
 * the row's attempt id must match it too; when the request attempt id is null,
 * every attempt id the run holds at that generation is releasable (the workers
 * runner mints a distinct attempt id per map chunk under one owner + seq). Row
 * status is irrelevant — the lease strands work receipts and blocks the write
 * fence whether the row is pending, running, failed, or enriched — but an
 * unowned row is never touched.
 */
export function decideRuntimeSheetRelease(input: {
  row: RuntimeSheetRowAttemptInput;
  request: RuntimeSheetAttemptIdentity;
}): RuntimeSheetReleaseDecision {
  const rowAttemptId = normalize(input.row.attemptId);
  if (!rowAttemptId) return { kind: 'skip' };
  const rowOwner =
    normalize(input.row.attemptOwnerRunId) ?? normalize(input.row.runId);
  const requestOwner = normalize(input.request.attemptOwnerRunId);
  if (!rowOwner || !requestOwner || rowOwner !== requestOwner) {
    return { kind: 'skip' };
  }
  if (
    normalizeAttemptSeq(input.row.attemptSeq) !==
    normalizeAttemptSeq(input.request.attemptSeq)
  ) {
    return { kind: 'skip' };
  }
  const requestAttemptId = normalize(input.request.attemptId);
  if (requestAttemptId && requestAttemptId !== rowAttemptId) {
    return { kind: 'skip' };
  }
  return { kind: 'release' };
}

function assertNever(value: never): never {
  throw new Error(`Unhandled runtime sheet row status: ${String(value)}`);
}