/**
 * `createWorkflow({...})` - the public entry point for the workflow
 * runtime. Validates the supplied configuration and returns the
 * {@link Workflow} handle that exposes `execute / resume / getState /
 * listCheckpoints / fork`.
 *
 * @packageDocumentation
 */

import type { Checkpoint, CheckpointId, Directive, WorkflowEvent } from '@graphorin/core';
import { DEFAULT_APPROVAL_TIMEOUT_DECISION } from '@graphorin/core';

import {
  CheckpointNotFoundError,
  InvalidWorkflowConfigError,
  ThreadNotFoundError,
  UnknownNodeError,
  WorkflowError,
} from './errors/index.js';
import {
  forkThread,
  namespaceFor,
  readFrontier,
  resumeEngine,
  runEngine,
  unwrapPersistedState as unwrapPersistedStateForRead,
} from './internal/engine.js';
import { newId } from './internal/ids.js';
import {
  END_NODE,
  START_NODE,
  type StreamMode,
  TASKS_CHANNEL,
  type Workflow,
  type WorkflowConfig,
  type WorkflowExecuteOptions,
  type WorkflowResumeOptions,
  type WorkflowState,
} from './types.js';

const DEFAULT_STREAM_MODE: StreamMode = 'values';

/**
 * Build a {@link Workflow} from the supplied configuration. The
 * factory performs eager validation so misuse is caught at build time
 * rather than mid-execution.
 *
 * @stable
 */
export function createWorkflow<
  TState extends object = Record<string, unknown>,
  TInput extends Partial<TState> = Partial<TState>,
>(config: WorkflowConfig<TState>): Workflow<TState, TInput> {
  validateConfig(config);

  const nodeNames = Object.freeze(Object.keys(config.nodes));
  const namespace = namespaceFor(config);
  // Per-instance re-entrancy guard for resume/retry; cross-instance and
  // cross-process races are handled by the store-level checkpoint CAS.
  const resumeLock = new Set<string>();

  const workflow: Workflow<TState, TInput> = Object.freeze({
    name: config.name,
    nodeNames,
    execute(input: TInput, opts?: WorkflowExecuteOptions): AsyncIterable<WorkflowEvent<TState>> {
      const threadId = opts?.threadId ?? newId('thread');
      const streamMode = opts?.stream ?? DEFAULT_STREAM_MODE;
      return runEngine<TState, TInput>({
        config,
        threadId,
        input,
        streamMode,
        ...(opts?.signal !== undefined ? { signal: opts.signal } : {}),
        ...(opts?.durability !== undefined ? { durability: opts.durability } : {}),
      });
    },
    resume(
      threadId: string,
      directive?: Directive,
      opts?: WorkflowResumeOptions,
    ): AsyncIterable<WorkflowEvent<TState>> {
      const streamMode = opts?.stream ?? DEFAULT_STREAM_MODE;
      return resumeEngine<TState>({
        config,
        threadId,
        ...(directive !== undefined ? { directive } : {}),
        streamMode,
        ...(opts?.signal !== undefined ? { signal: opts.signal } : {}),
        ...(opts?.durability !== undefined ? { durability: opts.durability } : {}),
        ...(opts?.allowVersionMismatch !== undefined
          ? { allowVersionMismatch: opts.allowVersionMismatch }
          : {}),
        resumeLock,
      });
    },
    retry(threadId: string, opts?: WorkflowResumeOptions): AsyncIterable<WorkflowEvent<TState>> {
      const streamMode = opts?.stream ?? DEFAULT_STREAM_MODE;
      return resumeEngine<TState>({
        config,
        threadId,
        streamMode,
        ...(opts?.signal !== undefined ? { signal: opts.signal } : {}),
        ...(opts?.durability !== undefined ? { durability: opts.durability } : {}),
        ...(opts?.allowVersionMismatch !== undefined
          ? { allowVersionMismatch: opts.allowVersionMismatch }
          : {}),
        resumeLock,
        mode: 'retry',
      });
    },
    async tick(
      threadId: string,
      opts?: { readonly now?: number },
    ): Promise<{ readonly fired: boolean; readonly nextWakeAt: number | null }> {
      const tuple = await config.checkpointStore.getTuple(threadId, namespace);
      if (!tuple) throw new ThreadNotFoundError(threadId);
      const now = opts?.now ?? Date.now();
      const timers = readFrontier(tuple.metadata).pauses.filter(
        (p): p is typeof p & { wakeAt: number } => typeof p.wakeAt === 'number',
      );
      const due = timers.filter((p) => p.wakeAt <= now);
      const pendingAfter = timers.filter((p) => p.wakeAt > now).map((p) => p.wakeAt);
      if (due.length === 0 || tuple.metadata.status !== 'suspended') {
        return {
          fired: false,
          nextWakeAt: pendingAfter.length > 0 ? Math.min(...pendingAfter) : null,
        };
      }
      // Resume targeting the earliest due timer; drain the resulting
      // events internally, surfacing the first workflow.error as a throw.
      const earliest = due.reduce((min, p) => (p.wakeAt < min.wakeAt ? p : min));
      // E1 defer-timeout: a due pause that carries a NAME is a pending
      // approval/awakeable with a deadline, not a bare `sleepUntil` -
      // resolve it with its timeout decision (auto-deny by default) so
      // an unattended deferred permission fails closed.
      const timeoutValue =
        earliest.name !== undefined
          ? ((earliest.value as { readonly timeoutDecision?: unknown } | null | undefined)
              ?.timeoutDecision ?? DEFAULT_APPROVAL_TIMEOUT_DECISION)
          : undefined;
      const events = resumeEngine<TState>({
        config,
        threadId,
        streamMode: 'values',
        resumeLock,
        ...(earliest.name !== undefined ? { directive: { resume: timeoutValue } } : {}),
        selectPause: (p) => p.wakeAt === earliest.wakeAt && p.nodeName === earliest.nodeName,
        selectPauseLabel: `timer@${earliest.wakeAt}`,
      });
      for await (const event of events) {
        if (event.type === 'workflow.error') {
          throw new WorkflowError(
            event.error.code as ConstructorParameters<typeof WorkflowError>[0],
            event.error.message,
          );
        }
      }
      const after = await config.checkpointStore.getTuple(threadId, namespace);
      const remaining =
        after === null
          ? []
          : readFrontier(after.metadata)
              .pauses.filter(
                (p): p is typeof p & { wakeAt: number } => typeof p.wakeAt === 'number',
              )
              .map((p) => p.wakeAt);
      return {
        fired: true,
        nextWakeAt: remaining.length > 0 ? Math.min(...remaining) : null,
      };
    },
    resolveAwakeable(
      threadId: string,
      name: string,
      value?: unknown,
      opts?: WorkflowResumeOptions,
    ): AsyncIterable<WorkflowEvent<TState>> {
      const streamMode = opts?.stream ?? DEFAULT_STREAM_MODE;
      return resumeEngine<TState>({
        config,
        threadId,
        directive: { resume: value },
        streamMode,
        ...(opts?.signal !== undefined ? { signal: opts.signal } : {}),
        ...(opts?.durability !== undefined ? { durability: opts.durability } : {}),
        ...(opts?.allowVersionMismatch !== undefined
          ? { allowVersionMismatch: opts.allowVersionMismatch }
          : {}),
        resumeLock,
        selectPause: (p) => p.name === name,
        selectPauseLabel: name,
      });
    },
    approve(
      threadId: string,
      name: string,
      decision: unknown,
      opts?: WorkflowResumeOptions,
    ): AsyncIterable<WorkflowEvent<TState>> {
      return workflow.resolveAwakeable(threadId, name, decision, opts);
    },
    async getState(threadId: string): Promise<WorkflowState<TState>> {
      const tuple = await config.checkpointStore.getTuple(threadId, namespace);
      if (!tuple) throw new ThreadNotFoundError(threadId);
      const status =
        tuple.metadata.status === 'running' ||
        tuple.metadata.status === 'suspended' ||
        tuple.metadata.status === 'completed' ||
        tuple.metadata.status === 'failed' ||
        tuple.metadata.status === 'aborted'
          ? tuple.metadata.status
          : 'running';
      const unwrapped = unwrapPersistedStateForRead(tuple.checkpoint.state);
      const stateRecord = (unwrapped as TState) ?? ({} as TState);
      const pendingPause = readPauseTag(tuple.metadata.tags);
      const frontierPauses = readFrontier(tuple.metadata).pauses;
      return {
        threadId,
        stepNumber: tuple.checkpoint.stepNumber,
        status,
        state: { ...(stateRecord as object) } as TState,
        checkpointId: tuple.checkpoint.id,
        ...(pendingPause !== undefined ? { pendingPause } : {}),
        // D1: the FULL pause set - timers (wakeAt), awakeables /
        // approvals (name), parallel pausers.
        ...(frontierPauses.length > 0 ? { pendingPauses: frontierPauses } : {}),
      };
    },
    async listCheckpoints(threadId: string): Promise<ReadonlyArray<Checkpoint>> {
      const out: Checkpoint[] = [];
      for await (const tuple of config.checkpointStore.list(threadId, namespace)) {
        out.push(tuple.checkpoint);
      }
      return Object.freeze(out);
    },
    async deleteThread(threadId: string): Promise<void> {
      await config.checkpointStore.deleteThread(threadId);
    },
    async fork(
      threadId: string,
      fromCheckpointId: CheckpointId,
      opts?: { readonly patch?: Readonly<Record<string, unknown>> },
    ): Promise<{ readonly newThreadId: string }> {
      const probe = await config.checkpointStore.getTuple(threadId, namespace, fromCheckpointId);
      if (!probe) throw new CheckpointNotFoundError(threadId, fromCheckpointId);
      return forkThread<TState>({
        config,
        threadId,
        fromCheckpointId,
        ...(opts?.patch !== undefined ? { patch: opts.patch } : {}),
      });
    },
  });

  return workflow;
}

function validateConfig<TState extends object>(config: WorkflowConfig<TState>): void {
  if (typeof config.name !== 'string' || config.name.length === 0) {
    throw new InvalidWorkflowConfigError('createWorkflow({ name }) must be a non-empty string');
  }
  if (typeof config.nodes !== 'object' || config.nodes === null) {
    throw new InvalidWorkflowConfigError(
      'createWorkflow({ nodes }) must be a record of name -> WorkflowNode',
    );
  }
  if (Object.keys(config.nodes).length === 0) {
    throw new InvalidWorkflowConfigError(
      'createWorkflow({ nodes }) must declare at least one node',
    );
  }
  for (const [name, node] of Object.entries(config.nodes)) {
    if (name === START_NODE || name === END_NODE) {
      throw new InvalidWorkflowConfigError(
        `node name "${name}" is reserved for the implicit ${name} sentinel`,
      );
    }
    if (typeof node?.run !== 'function') {
      throw new InvalidWorkflowConfigError(
        `node "${name}" must expose a callable run(state, ctx) function`,
      );
    }
  }
  if (!Array.isArray(config.edges)) {
    throw new InvalidWorkflowConfigError('createWorkflow({ edges }) must be an array');
  }
  let hasStartEdge = false;
  for (const edge of config.edges) {
    if (typeof edge.from !== 'string' || typeof edge.to !== 'string') {
      throw new InvalidWorkflowConfigError('every edge must declare string `from` and `to` fields');
    }
    if (edge.from === START_NODE) hasStartEdge = true;
    if (edge.from !== START_NODE && !(edge.from in config.nodes)) {
      throw new UnknownNodeError(edge.from, `edge.from`);
    }
    if (edge.to !== END_NODE && !(edge.to in config.nodes)) {
      throw new UnknownNodeError(edge.to, `edge.to`);
    }
  }
  if (!hasStartEdge) {
    throw new InvalidWorkflowConfigError(
      `the edges list must contain at least one edge from "${START_NODE}" - workflows always start there`,
    );
  }
  if (typeof config.channels !== 'object' || config.channels === null) {
    throw new InvalidWorkflowConfigError(
      'createWorkflow({ channels }) must be a record of stateKey -> Channel descriptor',
    );
  }
  for (const channelName of Object.keys(config.channels)) {
    if (channelName === TASKS_CHANNEL) {
      throw new InvalidWorkflowConfigError(
        `channel name "${channelName}" is reserved for internal task scheduling`,
      );
    }
  }
  if (!config.checkpointStore) {
    throw new InvalidWorkflowConfigError(
      'createWorkflow({ checkpointStore }) is required - supply an InMemoryCheckpointStore for tests or a SqliteCheckpointStore for production',
    );
  }
  if (
    config.durability !== undefined &&
    config.durability !== 'sync' &&
    config.durability !== 'exit' &&
    // WF-7: legacy value - the engine coerces it to 'sync' with a warn.
    (config.durability as string) !== 'async'
  ) {
    throw new InvalidWorkflowConfigError(
      `durability mode "${config.durability}" is invalid - accepted values are "sync" | "exit"`,
    );
  }
  if (
    config.maxSteps !== undefined &&
    (config.maxSteps < 1 || !Number.isInteger(config.maxSteps))
  ) {
    throw new InvalidWorkflowConfigError('maxSteps must be a positive integer');
  }
  if (
    config.maxTotalSteps !== undefined &&
    (config.maxTotalSteps < 1 || !Number.isInteger(config.maxTotalSteps))
  ) {
    throw new InvalidWorkflowConfigError('maxTotalSteps must be a positive integer');
  }
  if (
    config.cancelGraceMs !== undefined &&
    (config.cancelGraceMs < 0 || !Number.isFinite(config.cancelGraceMs))
  ) {
    throw new InvalidWorkflowConfigError('cancelGraceMs must be a non-negative finite number');
  }
}

function readPauseTag(
  tags: ReadonlyArray<string> | undefined,
): WorkflowState<object>['pendingPause'] {
  const tag = tags?.find((t) => t.startsWith('pause:'));
  if (!tag) return undefined;
  try {
    const parsed = JSON.parse(tag.slice('pause:'.length)) as {
      nodeName?: string;
      value?: unknown;
      dispatchArgs?: unknown;
      staticBefore?: boolean;
      staticAfter?: boolean;
    };
    if (typeof parsed.nodeName !== 'string') return undefined;
    return {
      nodeName: parsed.nodeName,
      value: parsed.value,
      ...(parsed.dispatchArgs !== undefined ? { dispatchArgs: parsed.dispatchArgs } : {}),
      ...(parsed.staticBefore ? { staticBefore: true } : {}),
      ...(parsed.staticAfter ? { staticAfter: true } : {}),
    };
  } catch {
    return undefined;
  }
}