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; 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 { 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 { if (!this.blocked) { await this.runPhase("postconditions"); } else { await this.writePhaseArtifact("postconditions", "skipped"); } } async runCleanup(): Promise { 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 { 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 { 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; 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; 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)[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 { return typeof value === "object" && value !== null && !Array.isArray(value); }