/**
 * Scheduler-backend interface — owns play workflow lifecycle.
 *
 * Durable scheduling Interface for Play Runs.
 * Selected per-run via PlayExecutionProfile.
 *
 * Absurd is the only durable production scheduler. In-process exists only for
 * focused local execution.
 *
 * Customer plays are unaffected — this is purely the orchestration layer.
 */

import type {
  PlayCheckpoint,
  PlayExecutionEvent,
  PlayRowUpdate,
} from './ctx-types';
import type { ExecutionPlan } from './execution-plan';
import type { PreloadedRuntimeDbSession } from './db-session';
import type { RuntimeAuthorityDescriptor } from './execution-capabilities';
import type { PlayRunnerRuntimeTiming } from './protocol';
import type { RuntimeTestPolicyOverrides } from './test-runtime-seams';
import type { PlayRunInputPayload } from './play-input';
import type { RuntimeSandboxPlacementPolicyId } from './runtime-sandbox-placement-policy';

export const PLAY_SCHEDULER_BACKENDS = {
  /**
   * Postgres-native durable-execution scheduler (vendored Absurd engine). Runs
   * on the battle-tested run-state store without an external workflow engine.
   * Selected explicitly by the `absurd` profile.
   */
  absurd: 'absurd',
  inProcess: 'in-process',
} as const;

export type PlaySchedulerBackendId =
  (typeof PLAY_SCHEDULER_BACKENDS)[keyof typeof PLAY_SCHEDULER_BACKENDS];

export type { PlayRunInputPayload };

export type PlaySchedulerSubmitInput = {
  runId: string;
  playId: string;
  playName: string;
  workflowFamilyKey?: string | null;
  artifactStorageKey: string;
  /** Optional inline artifact for the Node scheduler. */
  runtimeArtifact?: unknown;
  artifactHash: string;
  graphHash: string;
  input: PlayRunInputPayload;
  /** Convex metadata for the exact input saved before scheduler submission. */
  inputFileId?: string;
  inputBytes?: number;
  inputSha256?: string;
  replayedFromRunId?: string | null;
  /** Start a fresh run graph and recompute runtime-sheet rows. */
  force?: boolean;
  /** Explicit cache bypass for completed ctx.tools.execute receipts. */
  forceToolRefresh?: boolean | null;
  /** Validated per-run ceiling for provider-tool executions and ctx.fetch. */
  maxConcurrentExternalCalls?: number | null;
  maxConcurrentRows?: number | null;
  inputFile?: {
    name?: string;
    path?: string;
    r2Key?: string;
    storageKey?: string;
    fileName?: string;
    logicalPath?: string;
    contentType?: string;
    bytes?: number;
  } | null;
  inlineCsv?: { name: string; rows: Record<string, unknown>[] } | null;
  inlineInputFile?: {
    logicalPath: string;
    fileName: string;
    contentBase64: string;
    contentType: string;
    bytes: number;
  } | null;
  packagedFiles?: Array<{
    playPath?: string;
    logicalPath?: string;
    fileName?: string;
    storageKey: string;
    contentType?: string;
    bytes?: number;
    inlineText?: string;
  }> | null;
  contractSnapshot?: unknown;
  executionPlan?: ExecutionPlan | null;
  preloadedDbSessions?: PreloadedRuntimeDbSession[] | null;
  /**
   * Secret that decrypts `preloadedDbSessions` postgres URLs. Equals the run's
   * original submit-time executor (authority) token that AES-GCM sealed the
   * sessions. Held separately from `executorToken` because durable schedulers
   * re-mint a per-attempt `executorToken` (adding `run_attempt`/capabilities
   * claims) for API auth; that re-minted string is NOT valid decryption key
   * material for sessions sealed under the original token. The worker/runner use
   * this field for DB-session unwrap and the re-minted `executorToken` for auth.
   */
  preloadedDbSessionUnwrapKey?: string | null;
  executorToken: string;
  /** Immutable facts used by durable schedulers to mint a fresh leg token. */
  runtimeAuthority: RuntimeAuthorityDescriptor;
  /**
   * Public app origin used by Workers to call Deepline runtime API routes.
   * When omitted, legacy schedulers fall back to baseUrl.
   */
  runtimeApiBaseUrl?: string | null;
  /**
   * Stable signing origin for executor JWTs. This is intentionally distinct
   * from runtimeApiBaseUrl: a staged Vercel deployment must receive the
   * canary's runtime API traffic without forcing a long-lived gateway fleet to
   * trust a one-off deployment hostname.
   */
  runtimeTokenOrigin?: string | null;
  baseUrl: string;
  /** Optional per-run provider execution mode. Defaults to server policy when omitted. */
  integrationMode?: 'live' | 'eval_stub' | 'fixture';
  orgId: string;
  userEmail: string;
  userId?: string | null;
  source?: 'published' | 'ad_hoc' | 'draft';
  /** Invocation origin, distinct from the artifact source above. */
  triggerSource?: 'api' | 'webhook' | 'cron' | 'sql_listener';
  /** Null lets a durable trigger wait for capacity; a number bounds queue delay. */
  queueMaxDelaySeconds?: number | null;
  /**
   * Durable relative queue ordering. Higher values win within an organization
   * after cross-org fairness has selected that organization.
   */
  queuePriority?: number | null;
  executionProfile?: string | null;
  /** runner backend to use for executing attempts */
  runtimeBackend: string;
  /**
   * Versioned managed-sandbox placement contract. Optional only so launches
   * queued before the policy field existed can drain using runtimeBackend's
   * historical meaning.
   */
  runtimeSandboxPlacementPolicyId?: RuntimeSandboxPlacementPolicyId | null;
  /** If known at submit time, total input rows (for partition decisions). */
  totalRows?: number;
  /** Internal scheduler/coordinator URL for Worker-side capabilities. */
  coordinatorUrl?: string | null;
  /** Request-scoped coordinator auth token for non-production preview/dev runs. */
  coordinatorInternalToken?: string | null;
  /** Runtime deploy generation/version that owns this run, when known. */
  runtimeDeployVersion?: string | null;
  /**
   * Absurd release lane that owns this run for its whole life (children, parks,
   * wakes, retries). Stamped at launch by the absurd scheduler backend and
   * persisted inside launch_json. The worker derives the run's queue from this,
   * never from process-local config. Absent on pre-release launch rows, in which
   * case the worker falls back to the dev-collapse id (today's queue).
   */
  absurdReleaseId?: string | null;
  absurdReleaseEnvironment?: 'preview' | 'production';
  /** Request-scoped Vercel Deployment Protection bypass for preview runtime callbacks. */
  vercelProtectionBypassToken?: string | null;
  /** Request-scoped, dev-only runtime fault injection header for black-box durability tests. */
  runtimeTestFaultHeader?: string | null;
  /** Request-scoped, dev-only runtime policy overrides for black-box durability tests. */
  testPolicyOverrides?: RuntimeTestPolicyOverrides | null;
  /** Millisecond epoch timestamp captured immediately before scheduler submit. */
  submittedAtMs?: number;
  /** Scheduler backend that owns orchestration for this run. */
  schedulerBackend?: PlaySchedulerBackendId | null;
};

export type PlaySchedulerProgressEvent =
  | {
      type: 'status';
      status: string;
      logs?: string[];
      ts: number;
      activeNodeId?: string | null;
      activeArtifactTableNamespace?: string | null;
      updatedAt?: number | null;
      liveNodeProgress?: unknown;
    }
  | { type: 'log'; line: string; ts: number }
  | { type: 'row'; update: PlayRowUpdate; ts: number }
  | { type: 'execution_event'; event: PlayExecutionEvent; ts: number }
  | { type: 'completed'; result: unknown; ts: number }
  | { type: 'failed'; error: string; ts: number }
  | { type: 'suspended'; reason: string; ts: number };

export type PlaySchedulerRunHandle = {
  runId: string;
  /**
   * Stream live progress events. Implementations may use SSE, polling, etc.
   * The contract: yields events in order until terminal status.
   */
  observe(options?: {
    signal?: AbortSignal;
  }): AsyncIterable<PlaySchedulerProgressEvent>;
  /** Cooperatively cancel the run. */
  cancel(): Promise<void>;
  /** Inject an external event (HITL, webhook). */
  signal(payload: PlaySchedulerSignalPayload): Promise<void>;
  /** Block until terminal state and return final envelope. */
  result(): Promise<PlaySchedulerResultEnvelope>;
};

export type PlaySchedulerSignalPayload = {
  kind: 'integration_event' | 'cancel' | 'custom';
  eventKey?: string;
  data?: unknown;
};

export type PlaySchedulerResultEnvelope = {
  runId: string;
  status: 'completed' | 'failed' | 'cancelled';
  output?: unknown;
  error?: string;
  finalCheckpoint?: PlayCheckpoint;
  totalRows?: number;
  durationMs?: number;
  runtimeTiming?: PlayRunnerRuntimeTiming;
};

export interface PlaySchedulerBackend {
  readonly id: PlaySchedulerBackendId;

  /** Submit a play run; returns a handle to observe / cancel / signal it. */
  submit(input: PlaySchedulerSubmitInput): Promise<PlaySchedulerRunHandle>;

  /** Open a handle to an already-submitted run (e.g. for tail-reconnect). */
  attach(
    runId: string,
    options?: {
      coordinatorUrl?: string | null;
      coordinatorInternalToken?: string | null;
      runtimeDeployVersion?: string | null;
      initialState?: Record<string, unknown> | null;
      orgId?: string | null;
    },
  ): Promise<PlaySchedulerRunHandle>;
}

export function normalizePlaySchedulerBackend(
  value?: string | null,
): PlaySchedulerBackendId {
  const normalized = value?.trim().toLowerCase();
  if (!normalized) {
    return PLAY_SCHEDULER_BACKENDS.absurd;
  }
  if (normalized === 'absurd' || normalized === 'absurd-scheduler') {
    return PLAY_SCHEDULER_BACKENDS.absurd;
  }
  if (normalized === 'in-process' || normalized === 'in_process') {
    return PLAY_SCHEDULER_BACKENDS.inProcess;
  }
  throw new Error(
    `Unsupported scheduler backend "${normalized}". Expected one of: ${PLAY_SCHEDULER_BACKENDS.absurd}, ${PLAY_SCHEDULER_BACKENDS.inProcess}.`,
  );
}