suaveplan/openspec/changes/etl-completion/proposal.md

ETL completion

Package: @suaveplan/etl Status: PROPOSED Type: spec-conformance + feature

Why

The genesis port of @suaveplan/etl ships the core stage primitives (arraySource, mapTransform, BatchSink, etc.) and a Pipeline builder, but the surface around it is materially thinner than the upstream package at ../suaveplan/packages/universal/etl. Five concrete gaps make consumers either rebuild missing pieces in-app or abandon the package altogether:

1. No PipelineContext — upstream exposes a mutable execution context (data, errors, stepResults, hooks, metadata K/V) that travels through every step. Genesis stages get isolated TransformContext snapshots, so cross-step state requires bolting on a separate map.
2. No batch concurrency / timeout config — upstream's BatchConfig ships size, concurrency, and (via pipeline) per-batch timeouts. Genesis only honours size via BatchSink, so consumers can't opt into parallel batches or per-item timeout enforcement.
3. No EtlHookMap lifecycle bus — upstream emits pipeline:start, pipeline:end, step:start, step:end, step:error, batch:start, batch:end, batch:error through a typed hook map. Genesis has no observability surface beyond raw error returns.
4. Missing utility primitives — applyMappings, getNestedValue, setNestedValue, parallelMap, toBatches, withTimeout are all absent. These are the shared building blocks every non-trivial ETL pipeline needs and they belong in the package, not the call site.
5. Missing FieldMapping type — upstream ships FieldMapping ({ from, to, transform? }) plus applyMappings(record, mappings). Genesis has no canonical mapping shape, so each consumer hand-rolls one.

This change closes the five gaps in one coordinated pass so downstream services (@suaveplan/packer import pipelines, future analytics ETLs) can rely on a single coherent ETL surface.

What changes

• ADDED: core/context/context module — exports class PipelineContext and function createContext(options: PipelineContextOptions) returning a typed shared-state object with data, errors, stepResults, metadata map, hook registration, and an injectable LoggerLike / Clock / AbortSignal. The context threads through every step.
• ADDED: BatchConfig interface — { size: number; concurrency?: number; perItemTimeoutMs?: number; batchTimeoutMs?: number; continueOnError?: boolean }. Pipeline config grows batch?: BatchConfig. The pipeline runner SHALL honour all four fields.
• ADDED: types/etl-hooks module — exports EtlHookMap extending HookMap from @suaveplan/events::hooks.types, with keys "pipeline:started", "pipeline:completed", "stage:started", "stage:completed", "stage:failed", "batch:started", "batch:completed", "batch:failed". Plus EtlTransformHookData and EtlTransformingHandler aliases for the most common subscriber shape.
• ADDED: utils/utils module — exports applyMappings(record, mappings), getNestedValue(obj, path), setNestedValue(obj, path, value), parallelMap(items, fn, concurrency), toBatches(items, size), withTimeout(promise, ms).
• ADDED: FieldMapping type — { from: string; to: string; transform?: (value: unknown) => unknown }. mapTransform and any TransformConfig SHALL accept mappings?: readonly FieldMapping[] applied before the transform function.
• MODIFIED: Pipeline and RunnablePipeline — accept and emit EtlHookMap events through an injectable HookBus (defaults to a fresh internal bus when omitted). Existing errorHandler API preserved.
• ADDED: per-module sibling .md for every new file.

Out of scope

• Connectors for specific databases (e.g. SurrealDB source, Postgres sink, S3 source) — separate adapter packages so each can pull in its native client and integration test fixture independently. This proposal only stabilises the runner contract.
• CDC stream support — change-data-capture / event-source ingestion belongs to a streaming-etl change paired with @suaveplan/events interceptors; this proposal stays batch-pipeline-only.
• Schema inference / auto-mapping — synthesising FieldMapping[] from a sample record is a separate research item.
• Web Streams runner — the existing pipeline is in-memory and uses async iterators; a ReadableStream runner is its own change.
• Distributed pipeline orchestration — multi-worker pipelines (queue-backed work items) require an @suaveplan/idempotency integration that isn't in scope here.

Source mapping

This change derives from the upstream package at ../suaveplan/packages/universal/etl/src and the upstream enterprise audit at ../suaveplan/enterprise/etl.md.

Requirement	Source file	Source lines	Notes
PipelineContext class with mutable shared state	packages/universal/etl/src/core/context/context.ts	L1-L120	Verbatim port; thread through Clock / LoggerLike from @suaveplan/types per genesis convention
createContext(options) factory	packages/universal/etl/src/core/context/context.ts	(export at file end)	Verbatim
BatchConfig interface (size, concurrency, continueOnError)	packages/universal/etl/src/types/etl.types.ts	L98-L106	Verbatim port; extend with perItemTimeoutMs and batchTimeoutMs (genesis additions, see "no source line" below)
EtlHookMap typed event bus	packages/universal/etl/src/types/etl-hooks.ts	L1-L29	Verbatim port; expand from one upstream key (etl:transforming) to the eight lifecycle keys upstream emits via PipelineEventPayload
EtlTransformHookData / EtlTransformingHandler	packages/universal/etl/src/types/etl-hooks.ts	L9-L29	Verbatim
applyMappings(record, mappings) utility	packages/universal/etl/src/utils/utils.ts	L48-L80	Verbatim
getNestedValue(obj, path) utility	packages/universal/etl/src/utils/utils.ts	L7-L23	Verbatim
setNestedValue(obj, path, value) utility	packages/universal/etl/src/utils/utils.ts	L29-L47	Verbatim
parallelMap(items, fn, concurrency) utility	packages/universal/etl/src/utils/utils.ts	L82-L120	Verbatim
toBatches(items, size) utility	packages/universal/etl/src/utils/utils.ts	L122-L140	Verbatim
withTimeout(promise, ms) utility	packages/universal/etl/src/utils/utils.ts	L142-L151	Verbatim
FieldMapping type	packages/universal/etl/src/types/etl.types.ts	L64-L72	Verbatim
pipeline:started / pipeline:completed / stage:* / batch:* lifecycle events	packages/universal/etl/src/types/etl.types.ts	L106-L130	Renamed from upstream pipeline:start / step:start etc. to lifecycle-named (started/completed/failed) per genesis convention; payload shapes preserved

Requirements with no source line

Requirement	Justification
BatchConfig.perItemTimeoutMs and batchTimeoutMs fields	Upstream models pipeline-wide timeoutMs only. Genesis enforces granular timeouts because @suaveplan/retry::withTimeout is already in the dep graph and per-item enforcement is a common production requirement (slow rows shouldn't stall the whole batch).
Hook keys renamed to lifecycle-named pattern (*:started / *:completed / *:failed)	Genesis-wide naming convention applied across @suaveplan/migration, @suaveplan/network, @suaveplan/cache (see lifecycle keys in those packages). Upstream's *:start / *:end / *:error style was itself inconsistent across packages.
EtlHookMap built on HookMap from @suaveplan/events::hooks.types rather than upstream's bespoke listener registry	Genesis-wide event bus convention (CLAUDE.md @suaveplan/events listing).
ETL_CODES registered idempotently via registerEtlErrorCodes()	Genesis-wide error registry rule (CLAUDE.md §Workflow rule 2).

Verification approach

Each spec delta requirement carries a **Verified by:** line. The test files that will exist after this change ships:

• packages/platform/etl/src/core/context/context.test.ts — set / get / delete metadata, hook register / off, errors / stepResults appended, AbortSignal propagation
• packages/platform/etl/src/core/context/context.spec-conformance.test.ts — pins type-level PipelineContextOptions shape; injected Clock deterministic
• packages/platform/etl/src/core/pipeline/pipeline.test.ts (extended) — BatchConfig honoured: size chunks input, concurrency runs N batches in parallel, perItemTimeoutMs rejects slow items with EtlError(BATCH_ITEM_TIMEOUT), batchTimeoutMs rejects slow batches, continueOnError skips failed batch
• packages/platform/etl/src/types/etl-hooks/etl-hooks.test.ts — every key in EtlHookMap fires once per lifecycle transition with the correct payload
• packages/platform/etl/src/utils/utils.test.ts — applyMappings / getNestedValue / setNestedValue / parallelMap / toBatches / withTimeout happy-path + edge cases (deep path missing, concurrency 0, batch size 0, timeout 0)
• packages/platform/etl/src/utils/utils.spec-conformance.test.ts — parallelMap honours concurrency limit empirically; withTimeout rejects with EtlError(TIMEOUT)
• packages/platform/etl/src/types/etl-hooks/etl-hooks.spec-conformance.test.ts — pins lifecycle key strings via expectTypeOf; pins payload shapes
• packages/platform/etl/src/errors/etl-error.spec-conformance.test.ts — ETL_CODES registered after import; registerEtlErrorCodes idempotent

Impact

Affected dependents

Run bunx turbo ls --filter=...@suaveplan/etl to enumerate. Today grep -rl "@suaveplan/etl" packages/ --include="package.json" returns zero matches outside packages/platform/etl/package.json itself, so this change is purely additive.

• @suaveplan/etl (self) — [x] tests run / [x] docs reviewed / [ ] no action needed
• (no in-repo dependents) — verified by bunx turbo ls --filter=...@suaveplan/etl returning a single package.

If tests run is unchecked for any dependent, the change MUST NOT be folded into the canonical spec — finish that work first or split the proposal.

Migration impact

• Pipeline consumers — no break. Existing constructors continue to work; PipelineContext, BatchConfig, and the hook bus are all opt-in additive.
• mapTransform consumers — no break. New mappings?: readonly FieldMapping[] argument is optional.
• PipelineOptions consumers — no break. New batch, hooks, and context fields are optional and default to no-op behavior.
• SKIP consumers — no break. The SKIP symbol export is preserved.