src/lifecycle/execute.ts

import { mkdir, writeFile } from "node:fs/promises";
import { join } from "node:path";
import type {
  DataEnvironment,
  DataLifecycleCheck,
  DataLifecycleDefinition,
  DataLifecycleExecution,
  DataLifecycleOperation,
  DataLifecycleOperationResult,
  DataLifecyclePhase,
  TesterProfile,
  UserJourney,
} from "../core/schemas.js";
import {
  DataLifecycleBlockedError,
  type DataLifecycleProvider,
  DataLifecycleProviderError,
} from "./types.js";
import { redactSensitiveText, redactSensitiveValue } from "../utils/redaction.js";

const phaseOrder: DataLifecyclePhase[] = [
  "setup",
  "preflight",
  "postconditions",
  "cleanup",
];

export interface DataLifecycleControllerOptions {
  definition: DataLifecycleDefinition;
  environments: Record<string, DataEnvironment>;
  provider: DataLifecycleProvider;
  scope: "suite" | "journey";
  runId: string;
  runDir: string;
  suiteRunId?: string;
  journeyRunId?: string;
  journey?: UserJourney;
  profile?: TesterProfile;
  suiteManifest?: unknown;
  keepData?: boolean;
}

export class DataLifecycleController {
  readonly result: DataLifecycleExecution;
  manifest: unknown;

  private readonly environment: DataEnvironment;
  private readonly startedAt: Date;
  private manifestRedactValues: string[] = [];
  private blocked = false;

  constructor(private readonly options: DataLifecycleControllerOptions) {
    const environment = options.environments[options.definition.environment];
    if (!environment) {
      throw new Error(
        `No data lifecycle environment named "${options.definition.environment}" was configured.`,
      );
    }

    this.environment = environment;
    this.startedAt = new Date();
    const namespace =
      options.definition.namespace ??
      `jt_${sanitizeNamespace(options.journey?.id ?? options.runId)}_${shortRunId(options.runId)}`;
    const artifacts = Object.fromEntries(
      phaseOrder.map((phase) => [phase, join(options.runDir, `${phase}.json`)]),
    ) as DataLifecycleExecution["artifacts"];

    this.result = {
      schemaVersion: "0.1",
      scope: options.scope,
      environment: options.definition.environment,
      namespace,
      status: "passed",
      startedAt: this.startedAt.toISOString(),
      ...(options.suiteManifest === undefined
        ? {}
        : { suiteManifest: redactSensitiveValue(options.suiteManifest) }),
      artifacts,
      setup: [],
      preflight: [],
      postconditions: [],
      cleanup: [],
    };
  }

  get artifactPaths(): DataLifecycleExecution["artifacts"] {
    return this.result.artifacts;
  }

  async runSetupAndPreflight(): Promise<void> {
    await this.runPhase("setup");
    await this.runPhase("preflight");

    const blockingFailures = [...this.result.setup, ...this.result.preflight].filter(
      (operation) => operation.status === "failed",
    );
    if (blockingFailures.length > 0) {
      this.blocked = true;
      this.finish("blocked");
      throw new DataLifecycleBlockedError(
        `Data lifecycle blocked the run with ${blockingFailures.length} setup/preflight failure(s).`,
        blockingFailures,
      );
    }
  }

  async runPostconditions(): Promise<void> {
    if (!this.blocked) {
      await this.runPhase("postconditions");
    } else {
      await this.writePhaseArtifact("postconditions", "skipped");
    }
  }

  async runCleanup(): Promise<void> {
    if (this.options.keepData) {
      await this.writePhaseArtifact("cleanup", "skipped");
      this.finish(this.result.status);
      return;
    }

    await this.runPhase("cleanup");
  }

  finish(status = this.result.status): void {
    const endedAt = new Date();
    this.result.status = status;
    this.result.endedAt = endedAt.toISOString();
    this.result.durationMs = Math.max(0, endedAt.getTime() - this.startedAt.getTime());
    if (this.manifest !== undefined) {
      this.result.manifest = redactSensitiveValue(this.manifest, {
        extraValues: this.manifestRedactValues,
      });
    }
  }

  private async runPhase(phase: DataLifecyclePhase): Promise<void> {
    const operations = operationsForPhase(this.options.definition, phase);
    if (operations.length === 0) {
      await this.writePhaseArtifact(phase, "skipped");
      return;
    }

    for (const [index, operation] of operations.entries()) {
      const { operationResult, rawResult, redactValues } = await this.runOperation(phase, operation, index);
      this.result[phase].push(operationResult);

      if (phase === "setup" && operationResult.status === "passed" && rawResult !== undefined) {
        this.captureManifest(operation, rawResult, redactValues);
      }
    }

    const failed = this.result[phase].some(
      (operation) => operation.status === "failed" && !operationAllowFailure(operations, operation),
    );
    if (failed) {
      this.result.status =
        phase === "setup" || phase === "preflight" ? "blocked" : "failed";
    }

    await this.writePhaseArtifact(phase, failed ? this.result.status : "passed");
  }

  private async runOperation(
    phase: DataLifecyclePhase,
    operation: DataLifecycleOperation,
    index: number,
  ): Promise<{
    operationResult: DataLifecycleOperationResult;
    rawResult?: unknown;
    redactValues?: string[];
  }> {
    const startedAt = new Date();
    const operationId = operation.id ?? `${phase}-${index + 1}`;
    const inheritedRedactValues = this.manifestRedactValues;
    const resolvedArgs = resolveTemplate(operation.args, {
      context: {
        scope: this.options.scope,
        runId: this.options.runId,
        suiteRunId: this.options.suiteRunId,
        journeyRunId: this.options.journeyRunId,
        journeyId: this.options.journey?.id,
        testerProfileId: this.options.profile?.id,
        namespace: this.result.namespace,
        environment: this.result.environment,
      },
      manifest: this.manifest,
      suiteManifest: this.options.suiteManifest,
      env: process.env,
    });

    try {
      assertCapabilities(this.environment, operation);
      const providerResult = await this.options.provider.runOperation({
        scope: this.options.scope,
        runId: this.options.runId,
        suiteRunId: this.options.suiteRunId,
        journeyRunId: this.options.journeyRunId,
        journey: this.options.journey,
        profile: this.options.profile,
        namespace: this.result.namespace,
        manifest: this.manifest,
        suiteManifest: this.options.suiteManifest,
        environmentName: this.result.environment,
        environment: this.environment,
        phase,
        operation,
        args: resolvedArgs,
      });
      const endedAt = new Date();
      const checks = extractChecks(providerResult.result);
      const assertionFailed = checks?.some((check) => check.status === "fail") ?? false;
      const status = assertionFailed && !operation.allowFailure ? "failed" : "passed";
      const redactValues = combineRedactValues(inheritedRedactValues, providerResult.redactValues);
      const assertionError =
        status === "failed" && assertionFailed
          ? {
              message: redactSensitiveText(assertionFailureMessage(checks), { extraValues: redactValues }),
              classification: "assertion" as const,
            }
          : undefined;

      return {
        operationResult: {
          id: operationId,
          phase,
          environment: this.result.environment,
          provider: this.environment.provider,
          transport: "transport" in this.environment ? this.environment.transport : undefined,
          kind: operation.kind,
          function: operation.function,
          status,
          startedAt: startedAt.toISOString(),
          endedAt: endedAt.toISOString(),
          durationMs: Math.max(0, endedAt.getTime() - startedAt.getTime()),
          ...(resolvedArgs === undefined
            ? {}
            : {
                args: redactSensitiveValue(resolvedArgs, {
                  extraKeys: operation.redactKeys,
                  extraValues: redactValues,
                }),
              }),
          ...(providerResult.result === undefined
            ? {}
            : {
                result: redactSensitiveValue(providerResult.result, {
                  extraKeys: operation.redactKeys,
                  extraValues: redactValues,
                }),
              }),
          ...(checks
            ? {
                checks: redactSensitiveValue(checks, {
                  extraKeys: operation.redactKeys,
                  extraValues: redactValues,
                }) as DataLifecycleCheck[],
              }
            : {}),
          ...(providerResult.stdout
            ? { stdout: redactSensitiveText(providerResult.stdout, { extraValues: redactValues }) }
            : {}),
          ...(providerResult.stderr
            ? { stderr: redactSensitiveText(providerResult.stderr, { extraValues: redactValues }) }
            : {}),
          ...(providerResult.exitCode === undefined ? {} : { exitCode: providerResult.exitCode }),
          ...(providerResult.signal ? { signal: providerResult.signal } : {}),
          ...(providerResult.httpStatus === undefined ? {} : { httpStatus: providerResult.httpStatus }),
          ...(assertionError ? { error: assertionError } : {}),
        },
        rawResult: providerResult.result,
        redactValues,
      };
    } catch (caught) {
      const endedAt = new Date();
      const error = caught instanceof Error ? caught : new Error(String(caught));
      const providerError = error instanceof DataLifecycleProviderError ? error : undefined;
      const redactValues = combineRedactValues(inheritedRedactValues, providerError?.redactValues);
      return {
        operationResult: {
          id: operationId,
          phase,
          environment: this.result.environment,
          provider: this.environment.provider,
          transport: "transport" in this.environment ? this.environment.transport : undefined,
          kind: operation.kind,
          function: operation.function,
          status: "failed",
          startedAt: startedAt.toISOString(),
          endedAt: endedAt.toISOString(),
          durationMs: Math.max(0, endedAt.getTime() - startedAt.getTime()),
          ...(resolvedArgs === undefined
            ? {}
            : {
                args: redactSensitiveValue(resolvedArgs, {
                  extraKeys: operation.redactKeys,
                  extraValues: redactValues,
                }),
              }),
          ...(providerError?.result === undefined
            ? {}
            : {
                result: redactSensitiveValue(providerError.result, {
                  extraKeys: operation.redactKeys,
                  extraValues: redactValues,
                }),
              }),
          ...(providerError?.stdout
            ? { stdout: redactSensitiveText(providerError.stdout, { extraValues: redactValues }) }
            : {}),
          ...(providerError?.stderr
            ? { stderr: redactSensitiveText(providerError.stderr, { extraValues: redactValues }) }
            : {}),
          ...(providerError?.exitCode === undefined ? {} : { exitCode: providerError.exitCode }),
          ...(providerError?.signal ? { signal: providerError.signal } : {}),
          ...(providerError?.httpStatus === undefined ? {} : { httpStatus: providerError.httpStatus }),
          error: {
            message: redactSensitiveText(error.message, { extraValues: redactValues }),
            stack: error.stack
              ? redactSensitiveText(error.stack, { extraValues: redactValues })
              : undefined,
            classification: classifyError(error),
          },
        },
      };
    }
  }

  private captureManifest(
    operation: DataLifecycleOperation,
    result: unknown,
    redactValues: string[] = [],
  ): void {
    const selected =
      operation.manifestPath === undefined ? result : selectJsonPath(result, operation.manifestPath);
    this.manifest = selected;
    this.manifestRedactValues = redactValues;
    this.result.manifest = redactSensitiveValue(selected, {
      extraKeys: operation.redactKeys,
      extraValues: redactValues,
    });
  }

  private async writePhaseArtifact(
    phase: DataLifecyclePhase,
    status: DataLifecycleExecution["status"],
  ): Promise<void> {
    const artifactPath = this.result.artifacts[phase];
    if (!artifactPath) {
      throw new Error(`No data lifecycle artifact path configured for phase "${phase}".`);
    }

    await mkdir(this.options.runDir, { recursive: true });
    await writeFile(
      artifactPath,
      `${JSON.stringify(
        {
          schemaVersion: "0.1",
          scope: this.options.scope,
          environment: this.result.environment,
          namespace: this.result.namespace,
          phase,
          status,
          manifest: this.result.manifest,
          suiteManifest: this.result.suiteManifest,
          operations: this.result[phase],
        },
        null,
        2,
      )}\n`,
      "utf8",
    );
  }
}

function operationsForPhase(
  definition: DataLifecycleDefinition,
  phase: DataLifecyclePhase,
): DataLifecycleOperation[] {
  return definition[phase] ?? [];
}

function operationAllowFailure(
  operations: DataLifecycleOperation[],
  result: DataLifecycleOperationResult,
): boolean {
  const operation = operations.find(
    (candidate, index) => (candidate.id ?? `${result.phase}-${index + 1}`) === result.id,
  );
  return operation?.allowFailure ?? false;
}

function assertCapabilities(
  environment: DataEnvironment,
  operation: DataLifecycleOperation,
): void {
  const requested = operation.requiredCapabilities ?? [];
  if (requested.length === 0) {
    return;
  }

  const defaults =
    environment.provider === "convex" && environment.transport === "cli"
      ? { publicFunctions: true, internalFunctions: true }
      : { publicFunctions: true, internalFunctions: false };
  const capabilities = { ...defaults, ...(environment.capabilities ?? {}) };

  for (const capability of requested) {
    if (!capabilities[capability]) {
      throw new Error(
        `Data lifecycle operation "${operation.function}" requires capability "${capability}", but environment does not declare it.`,
      );
    }
  }
}

function extractChecks(result: unknown): DataLifecycleCheck[] | undefined {
  if (!isRecord(result) || !Array.isArray(result.checks)) {
    return undefined;
  }

  return result.checks.map((raw, index) => {
    if (!isRecord(raw)) {
      return {
        id: `check-${index + 1}`,
        status: "fail",
        message: "Check result was not an object.",
        data: raw,
      };
    }

    const status = raw.status === "pass" || raw.status === "warn" || raw.status === "fail"
      ? raw.status
      : "fail";

    return {
      id: typeof raw.id === "string" && raw.id ? raw.id : `check-${index + 1}`,
      status,
      ...(typeof raw.message === "string" && raw.message ? { message: raw.message } : {}),
      ...("data" in raw ? { data: raw.data } : {}),
    };
  });
}

function assertionFailureMessage(checks: DataLifecycleCheck[] | undefined): string {
  const failures = checks
    ?.filter((check) => check.status === "fail")
    .map((check) => `${check.id}${check.message ? `: ${check.message}` : ""}`);
  return failures && failures.length > 0
    ? `Data lifecycle checks failed: ${failures.join("; ")}`
    : "Data lifecycle checks failed.";
}

function combineRedactValues(
  inherited: string[] = [],
  current: string[] = [],
): string[] {
  return [...new Set([...inherited, ...current].filter((value) => value.length > 0))];
}

function resolveTemplate(
  value: unknown,
  sources: {
    context: Record<string, unknown>;
    manifest?: unknown;
    suiteManifest?: unknown;
    env: NodeJS.ProcessEnv;
  },
): unknown {
  if (typeof value === "string" && value.startsWith("$")) {
    return resolveReference(value, sources);
  }
  if (Array.isArray(value)) {
    return value.map((item) => resolveTemplate(item, sources));
  }
  if (isRecord(value)) {
    return Object.fromEntries(
      Object.entries(value).map(([key, entry]) => [key, resolveTemplate(entry, sources)]),
    );
  }
  return value;
}

function resolveReference(
  value: string,
  sources: {
    context: Record<string, unknown>;
    manifest?: unknown;
    suiteManifest?: unknown;
    env: NodeJS.ProcessEnv;
  },
): unknown {
  if (value === "$manifest") {
    return sources.manifest;
  }
  if (value.startsWith("$manifest.")) {
    return selectJsonPath(sources.manifest, value.replace("$manifest", "$"));
  }
  if (value === "$suiteManifest") {
    return sources.suiteManifest;
  }
  if (value.startsWith("$suiteManifest.")) {
    return selectJsonPath(sources.suiteManifest, value.replace("$suiteManifest", "$"));
  }
  if (value.startsWith("$context.")) {
    return selectJsonPath(sources.context, value.replace("$context", "$"));
  }
  if (value.startsWith("$env.")) {
    const key = value.slice("$env.".length);
    const resolved = sources.env[key];
    if (resolved === undefined) {
      throw new Error(`Environment variable "${key}" is not set.`);
    }
    return resolved;
  }

  throw new Error(`Unknown data lifecycle template reference "${value}".`);
}

export function selectJsonPath(value: unknown, path: string): unknown {
  const normalized = path.startsWith("$.") ? path.slice(2) : path.startsWith("$") ? path.slice(1) : path;
  if (!normalized) {
    return value;
  }

  let current = value;
  for (const segment of normalized.split(".")) {
    if (!segment) {
      continue;
    }

    const match = /^([^[\]]+)(?:\[(\d+)])?$/.exec(segment);
    if (!match) {
      throw new Error(`Unsupported JSON path segment "${segment}" in "${path}".`);
    }

    const key = match[1];
    if (!isRecord(current) && !Array.isArray(current)) {
      throw new Error(`Cannot resolve "${path}" because "${key}" is not reachable.`);
    }
    current = (current as Record<string, unknown>)[key];

    if (match[2] !== undefined) {
      if (!Array.isArray(current)) {
        throw new Error(`Cannot resolve "${path}" because "${key}" is not an array.`);
      }
      current = current[Number(match[2])];
    }
  }

  return current;
}

function classifyError(error: Error): "configuration" | "capability" | "execution" | "assertion" {
  if (error instanceof DataLifecycleProviderError && error.classification) {
    return error.classification;
  }
  if (/environment|configured|not set|JSON path|template/i.test(error.message)) {
    return "configuration";
  }
  if (/capability/i.test(error.message)) {
    return "capability";
  }
  return "execution";
}

function sanitizeNamespace(value: string): string {
  return value.replace(/[^a-zA-Z0-9_]+/g, "_").replace(/^_+|_+$/g, "") || "run";
}

function shortRunId(value: string): string {
  return sanitizeNamespace(value).slice(-12) || "run";
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}