runtime.ts

import { WorkflowAPIError, WorkflowRuntimeError } from '@workflow/errors';
import { parseWorkflowName } from '@workflow/utils/parse-name';
import {
  type Event,
  SPEC_VERSION_CURRENT,
  WorkflowInvokePayloadSchema,
  type WorkflowRun,
} from '@workflow/world';
import { importKey } from './encryption.js';
import { WorkflowSuspension } from './global.js';
import { runtimeLogger } from './logger.js';
import {
  getAllWorkflowRunEvents,
  getQueueOverhead,
  handleHealthCheckMessage,
  parseHealthCheckPayload,
  withHealthCheck,
} from './runtime/helpers.js';
import { handleSuspension } from './runtime/suspension-handler.js';
import { getWorld, getWorldHandlers } from './runtime/world.js';
import { remapErrorStack } from './source-map.js';
import * as Attribute from './telemetry/semantic-conventions.js';
import {
  linkToCurrentContext,
  trace,
  withTraceContext,
  withWorkflowBaggage,
} from './telemetry.js';
import { getErrorName, getErrorStack, normalizeUnknownError } from './types.js';
import { buildWorkflowSuspensionMessage } from './util.js';
import { runWorkflow } from './workflow.js';

export type { Event, WorkflowRun };
export { WorkflowSuspension } from './global.js';
export {
  type HealthCheckEndpoint,
  type HealthCheckOptions,
  type HealthCheckResult,
  healthCheck,
} from './runtime/helpers.js';
export {
  getHookByToken,
  resumeHook,
  resumeWebhook,
} from './runtime/resume-hook.js';
export {
  getRun,
  Run,
  type WorkflowReadableStreamOptions,
} from './runtime/run.js';
export {
  cancelRun,
  listStreams,
  type ReadStreamOptions,
  type RecreateRunOptions,
  readStream,
  recreateRunFromExisting,
  reenqueueRun,
  type StopSleepOptions,
  type StopSleepResult,
  wakeUpRun,
} from './runtime/runs.js';
export { type StartOptions, start } from './runtime/start.js';
export { stepEntrypoint } from './runtime/step-handler.js';
export {
  createWorld,
  getWorld,
  getWorldHandlers,
  setWorld,
} from './runtime/world.js';

/**
 * Function that creates a single route which handles any workflow execution
 * request and routes to the appropriate workflow function.
 *
 * @param workflowCode - The workflow bundle code containing all the workflow
 * functions at the top level.
 * @returns A function that can be used as a Vercel API route.
 */
export function workflowEntrypoint(
  workflowCode: string
): (req: Request) => Promise<Response> {
  const handler = getWorldHandlers().createQueueHandler(
    '__wkf_workflow_',
    async (message_, metadata) => {
      // Check if this is a health check message
      // NOTE: Health check messages are intentionally unauthenticated for monitoring purposes.
      // They only write a simple status response to a stream and do not expose sensitive data.
      // The stream name includes a unique correlationId that must be known by the caller.
      const healthCheck = parseHealthCheckPayload(message_);
      if (healthCheck) {
        await handleHealthCheckMessage(healthCheck, 'workflow');
        return;
      }

      const {
        runId,
        traceCarrier: traceContext,
        requestedAt,
      } = WorkflowInvokePayloadSchema.parse(message_);
      // Extract the workflow name from the topic name
      const workflowName = metadata.queueName.slice('__wkf_workflow_'.length);
      const spanLinks = await linkToCurrentContext();

      // Invoke user workflow within the propagated trace context and baggage
      return await withTraceContext(traceContext, async () => {
        // Set workflow context as baggage for automatic propagation
        return await withWorkflowBaggage(
          { workflowRunId: runId, workflowName },
          async () => {
            const world = getWorld();
            return trace(
              `WORKFLOW ${workflowName}`,
              { links: spanLinks },
              async (span) => {
                span?.setAttributes({
                  ...Attribute.WorkflowName(workflowName),
                  ...Attribute.WorkflowOperation('execute'),
                  // Standard OTEL messaging conventions
                  ...Attribute.MessagingSystem('vercel-queue'),
                  ...Attribute.MessagingDestinationName(metadata.queueName),
                  ...Attribute.MessagingMessageId(metadata.messageId),
                  ...Attribute.MessagingOperationType('process'),
                  ...getQueueOverhead({ requestedAt }),
                });

                // TODO: validate `workflowName` exists before consuming message?

                span?.setAttributes({
                  ...Attribute.WorkflowRunId(runId),
                  ...Attribute.WorkflowTracePropagated(!!traceContext),
                });

                let workflowStartedAt = -1;
                let workflowRun = await world.runs.get(runId);

                // --- Infrastructure: prepare the run state ---
                // Network/server errors propagate to the queue handler for retry.
                // WorkflowRuntimeError (data integrity issues) are fatal and
                // produce run_failed since retrying won't fix them.
                try {
                  if (workflowRun.status === 'pending') {
                    // Transition run to 'running' via event (event-sourced architecture)
                    const result = await world.events.create(runId, {
                      eventType: 'run_started',
                      specVersion: SPEC_VERSION_CURRENT,
                    });
                    // Use the run entity from the event response (no extra get call needed)
                    if (!result.run) {
                      throw new WorkflowRuntimeError(
                        `Event creation for 'run_started' did not return the run entity for run "${runId}"`
                      );
                    }
                    workflowRun = result.run;
                  }

                  // At this point, the workflow is "running" and `startedAt` should
                  // definitely be set.
                  if (!workflowRun.startedAt) {
                    throw new WorkflowRuntimeError(
                      `Workflow run "${runId}" has no "startedAt" timestamp`
                    );
                  }
                } catch (err) {
                  if (err instanceof WorkflowRuntimeError) {
                    runtimeLogger.error(
                      'Fatal runtime error during workflow setup',
                      { workflowRunId: runId, error: err.message }
                    );
                    try {
                      await world.events.create(runId, {
                        eventType: 'run_failed',
                        specVersion: SPEC_VERSION_CURRENT,
                        eventData: {
                          error: {
                            message: err.message,
                            stack: err.stack,
                          },
                        },
                      });
                    } catch (failErr) {
                      if (
                        WorkflowAPIError.is(failErr) &&
                        (failErr.status === 409 || failErr.status === 410)
                      ) {
                        return;
                      }
                      throw failErr;
                    }
                    return;
                  }
                  throw err;
                }
                workflowStartedAt = +workflowRun.startedAt;

                span?.setAttributes({
                  ...Attribute.WorkflowRunStatus(workflowRun.status),
                  ...Attribute.WorkflowStartedAt(workflowStartedAt),
                });

                if (workflowRun.status !== 'running') {
                  // Workflow has already completed or failed, so we can skip it
                  runtimeLogger.info(
                    'Workflow already completed or failed, skipping',
                    {
                      workflowRunId: runId,
                      status: workflowRun.status,
                    }
                  );

                  // TODO: for `cancel`, we actually want to propagate a WorkflowCancelled event
                  // inside the workflow context so the user can gracefully exit. this is SIGTERM
                  // TODO: furthermore, there should be a timeout or a way to force cancel SIGKILL
                  // so that we actually exit here without replaying the workflow at all, in the case
                  // the replaying the workflow is itself failing.

                  return;
                }

                // Load all events into memory before running
                let events = await getAllWorkflowRunEvents(workflowRun.runId);

                // Check for any elapsed waits and create wait_completed events
                const now = Date.now();

                // Pre-compute completed correlation IDs for O(n) lookup instead of O(n²)
                const completedWaitIds = new Set(
                  events
                    .filter((e) => e.eventType === 'wait_completed')
                    .map((e) => e.correlationId)
                );

                // Collect all waits that need completion
                const waitsToComplete = events
                  .filter(
                    (e): e is typeof e & { correlationId: string } =>
                      e.eventType === 'wait_created' &&
                      e.correlationId !== undefined &&
                      !completedWaitIds.has(e.correlationId) &&
                      now >= (e.eventData.resumeAt as Date).getTime()
                  )
                  .map((e) => ({
                    eventType: 'wait_completed' as const,
                    specVersion: SPEC_VERSION_CURRENT,
                    correlationId: e.correlationId,
                  }));

                // Create all wait_completed events.
                // If any creation returns 409, a concurrent invocation already
                // created the event. Re-fetch the full event log to get the
                // authoritative ordering rather than appending locally.
                let needsRefetch = false;
                for (const waitEvent of waitsToComplete) {
                  try {
                    const result = await world.events.create(runId, waitEvent);
                    // Add the event to the events array so the workflow can see it
                    events.push(result.event!);
                  } catch (err) {
                    if (WorkflowAPIError.is(err) && err.status === 409) {
                      runtimeLogger.info('Wait already completed, skipping', {
                        workflowRunId: runId,
                        correlationId: waitEvent.correlationId,
                      });
                      needsRefetch = true;
                      continue;
                    }
                    throw err;
                  }
                }

                // Re-fetch the event log if a concurrent invocation created
                // events we don't have locally, ensuring correct ordering.
                if (needsRefetch) {
                  events = await getAllWorkflowRunEvents(workflowRun.runId);
                }

                // Resolve the encryption key for this run's deployment
                const rawKey =
                  await world.getEncryptionKeyForRun?.(workflowRun);
                const encryptionKey = rawKey
                  ? await importKey(rawKey)
                  : undefined;

                // --- User code execution ---
                // Only errors from runWorkflow() (user workflow code) should
                // produce run_failed. Infrastructure errors (network, server)
                // must propagate to the queue handler for automatic retry.
                let workflowResult: unknown;
                try {
                  workflowResult = await trace(
                    'workflow.replay',
                    {},
                    async (replaySpan) => {
                      replaySpan?.setAttributes({
                        ...Attribute.WorkflowEventsCount(events.length),
                      });
                      return await runWorkflow(
                        workflowCode,
                        workflowRun,
                        events,
                        encryptionKey
                      );
                    }
                  );
                } catch (err) {
                  // WorkflowSuspension is normal control flow — not an error
                  if (WorkflowSuspension.is(err)) {
                    const suspensionMessage = buildWorkflowSuspensionMessage(
                      runId,
                      err.stepCount,
                      err.hookCount,
                      err.waitCount
                    );
                    if (suspensionMessage) {
                      runtimeLogger.debug(suspensionMessage);
                    }

                    const result = await handleSuspension({
                      suspension: err,
                      world,
                      run: workflowRun,
                      span,
                    });

                    if (result.timeoutSeconds !== undefined) {
                      return { timeoutSeconds: result.timeoutSeconds };
                    }

                    // Suspension handled, no further work needed
                    return;
                  }

                  // This is a user code error or a WorkflowRuntimeError
                  // (e.g., corrupted event log). Fail the workflow run.

                  // Record exception for OTEL error tracking
                  if (err instanceof Error) {
                    span?.recordException?.(err);
                  }

                  const normalizedError = await normalizeUnknownError(err);
                  const errorName = normalizedError.name || getErrorName(err);
                  const errorMessage = normalizedError.message;
                  let errorStack = normalizedError.stack || getErrorStack(err);

                  // Remap error stack using source maps to show original source locations
                  if (errorStack) {
                    const parsedName = parseWorkflowName(workflowName);
                    const filename =
                      parsedName?.moduleSpecifier || workflowName;
                    errorStack = remapErrorStack(
                      errorStack,
                      filename,
                      workflowCode
                    );
                  }

                  runtimeLogger.error('Error while running workflow', {
                    workflowRunId: runId,
                    errorName,
                    errorStack,
                  });

                  // Fail the workflow run via event (event-sourced architecture)
                  try {
                    await world.events.create(runId, {
                      eventType: 'run_failed',
                      specVersion: SPEC_VERSION_CURRENT,
                      eventData: {
                        error: {
                          message: errorMessage,
                          stack: errorStack,
                        },
                        // TODO: include error codes when we define them
                      },
                    });
                  } catch (failErr) {
                    if (
                      WorkflowAPIError.is(failErr) &&
                      (failErr.status === 409 || failErr.status === 410)
                    ) {
                      runtimeLogger.warn(
                        'Tried failing workflow run, but run has already finished.',
                        {
                          workflowRunId: runId,
                          message: failErr.message,
                        }
                      );
                      span?.setAttributes({
                        ...Attribute.WorkflowErrorName(errorName),
                        ...Attribute.WorkflowErrorMessage(errorMessage),
                        ...Attribute.ErrorType(errorName),
                      });
                      return;
                    } else {
                      throw failErr;
                    }
                  }

                  span?.setAttributes({
                    ...Attribute.WorkflowRunStatus('failed'),
                    ...Attribute.WorkflowErrorName(errorName),
                    ...Attribute.WorkflowErrorMessage(errorMessage),
                    ...Attribute.ErrorType(errorName),
                  });
                  return;
                }

                // --- Infrastructure: complete the run ---
                // This is outside the user-code try/catch so that failures
                // here (e.g., network errors) propagate to the queue handler.
                try {
                  await world.events.create(runId, {
                    eventType: 'run_completed',
                    specVersion: SPEC_VERSION_CURRENT,
                    eventData: {
                      output: workflowResult,
                    },
                  });
                } catch (err) {
                  if (
                    WorkflowAPIError.is(err) &&
                    (err.status === 409 || err.status === 410)
                  ) {
                    runtimeLogger.warn(
                      'Tried completing workflow run, but run has already finished.',
                      {
                        workflowRunId: runId,
                        message: err.message,
                      }
                    );
                    return;
                  } else {
                    throw err;
                  }
                }

                span?.setAttributes({
                  ...Attribute.WorkflowRunStatus('completed'),
                  ...Attribute.WorkflowEventsCount(events.length),
                });
              }
            ); // End trace
          }
        ); // End withWorkflowBaggage
      }); // End withTraceContext
    }
  );

  return withHealthCheck(handler);
}

// this is a no-op placeholder as the client is
// expecting this to be present but we aren't actually using it
export function runStep() {}