suaveplan/openspec/archive/2026-04-24-enterprise-etl/proposal.md

OpenSpec Change: enterprise-etl

• Change ID: enterprise-etl
• Package: @suaveplan/etl
• Category: universal
• Target path: universal/data/etl
• Status: PROPOSED
• Date: 2026-04-24
• Reference spec: /mnt/s/web/suaveplan/enterprise/etl.md

---

Summary

Deliver @suaveplan/etl on the genesis branch from scratch at enterprise quality. The package provides extract-transform-load pipeline primitives: typed Source, Transform, and Sink stages composable into declarative pipelines with built-in backpressure, AbortSignal cancellation, structured errors, OTel instrumentation, retry/circuit-breaker on IO paths, and property-based correctness guarantees. Genesis builds on @suaveplan/streams for the underlying stream substrate and delegates serialization to @suaveplan/serialize.

---

Current-state audit (genesis)

The package does not exist on genesis yet — fresh build. No prior source, no test results, no coverage data.

---

Available workspace deps on genesis

Package	Role
@suaveplan/error (workspace:^)	Base class for EtlError; defineCodes for ETL.* codes
@suaveplan/types (workspace:^)	LoggerLike, Clock, EventBusLike injection interfaces
@suaveplan/streams (workspace:^)	WHATWG-Streams substrate for pipeline stages
@suaveplan/serialize (workspace:^)	Chunk serialization / deserialization for file and network sinks
@suaveplan/retry (workspace:^)	Retry + circuit-breaker wrappers for IO source/sink adapters
@suaveplan/events (workspace:^)	Progress and lifecycle event emission
zod (catalog:forms)	Zod schemas for all public option types and record contracts
@opentelemetry/api (catalog:observability)	Spans, counters, histograms on pipeline run, stage durations
tslib (catalog:build)	Runtime helpers

Dev deps: vitest, @vitest/coverage-istanbul, fast-check, mitata (all catalog:testing).

---

Public API surface

Core pipeline

// Pipeline orchestrator
function createPipeline<TIn, TOut>(options: PipelineOptions<TIn, TOut>): Pipeline<TIn, TOut>

interface Pipeline<TIn, TOut> {
  run(signal?: AbortSignal): Promise<PipelineResult<TOut>>
  pipe<TNext>(transform: Transform<TOut, TNext>): Pipeline<TIn, TNext>
}

interface PipelineResult<T> {
  records: T[]
  stats: PipelineStats
  errors: EtlError[]
}

interface PipelineStats {
  extracted: number
  transformed: number
  loaded: number
  skipped: number
  durationMs: number
}

interface PipelineOptions<TIn, TOut> {
  source: Source<TIn>
  transforms?: Transform<TIn, unknown>[]
  sink: Sink<TOut>
  batchSize?: number
  concurrency?: number
  onError?: ErrorHandler
  logger?: LoggerLike
  clock?: Clock
}

Source interface

interface Source<T> {
  readonly name: string
  open(signal?: AbortSignal): Promise<void>
  read(): AsyncGenerator<T>
  close(): Promise<void>
}

// Built-in sources
function arraySource<T>(items: T[]): Source<T>
function generatorSource<T>(gen: () => AsyncGenerator<T>, name?: string): Source<T>
function streamSource<T>(stream: ReadableStream<T>, name?: string): Source<T>

Transform interface

interface Transform<TIn, TOut> {
  readonly name: string
  transform(record: TIn, ctx: TransformContext): Promise<TOut | typeof SKIP>
}

const SKIP: unique symbol

interface TransformContext {
  index: number
  signal?: AbortSignal
  logger?: LoggerLike
}

// Built-in transforms
function mapTransform<TIn, TOut>(fn: (record: TIn) => TOut | Promise<TOut>, name?: string): Transform<TIn, TOut>
function filterTransform<T>(predicate: (record: T) => boolean | Promise<boolean>, name?: string): Transform<T, T>
function validateTransform<T>(schema: ZodSchema<T>, name?: string): Transform<unknown, T>
function batchTransform<T>(size: number): Transform<T, T[]>
function flatMapTransform<TIn, TOut>(fn: (record: TIn) => TOut[] | Promise<TOut[]>, name?: string): Transform<TIn, TOut>

Sink interface

interface Sink<T> {
  readonly name: string
  open(signal?: AbortSignal): Promise<void>
  write(record: T): Promise<void>
  flush(): Promise<void>
  close(): Promise<void>
}

// Built-in sinks
function arraySink<T>(): Sink<T> & { records: T[] }
function nullSink<T>(): Sink<T>
function transformSink<T>(fn: (record: T) => void | Promise<void>, name?: string): Sink<T>

Error handling

class EtlError extends SuaveplanError<EtlErrorCode>

type EtlErrorCode =
  | "ETL_SOURCE_FAILED"
  | "ETL_TRANSFORM_FAILED"
  | "ETL_SINK_FAILED"
  | "ETL_PIPELINE_ABORTED"
  | "ETL_VALIDATION_FAILED"
  | "ETL_BATCH_OVERFLOW"

type ErrorHandler = (error: EtlError, record: unknown, stage: string) => ErrorAction
type ErrorAction = "skip" | "abort" | "retry"

function registerEtlErrorCodes(): void

Zod option schemas

const PipelineOptionsSchema: ZodSchema
const SourceOptionsSchema: ZodSchema
const TransformOptionsSchema: ZodSchema
const SinkOptionsSchema: ZodSchema

---

New additions for genesis

1. Streams integration

Pipeline stages are backed by @suaveplan/streams ReadableStream / TransformStream / WritableStream internally. The public API remains stage-based but the execution engine uses WHATWG streams for backpressure propagation.

2. Retry/circuit-breaker on IO paths

Source.open, Source.read, Sink.write, and Sink.flush are wrapped by @suaveplan/retry strategies when a retryOptions field is provided in source/sink options.

3. OTel instrumentation

• pipeline.run — span etl.pipeline.run, attributes: etl.source, etl.sink, etl.batch_size
• Per-stage spans: etl.source.read, etl.transform.<name>, etl.sink.write
• Counters: etl.records_extracted, etl.records_transformed, etl.records_loaded, etl.records_skipped, etl.errors_total
• Histogram: etl.stage_duration_ms (attribute: etl.stage)

4. Progress events via @suaveplan/events

When a bus is provided in PipelineOptions, the pipeline emits typed events:

• etl:pipeline:started — { pipelineId, source, sink }
• etl:stage:completed — { stage, records, durationMs }
• etl:pipeline:completed — { stats }
• etl:pipeline:failed — { error }

5. Property-based tests (fast-check)

• mapTransform composition: pipe(f, g)(x) ≡ g(f(x)) for all pure functions
• filterTransform idempotency: applying the same filter twice yields the same result
• batchTransform completeness: total records across all batches equals input count
• Pipeline abort: any mid-flight AbortSignal cancellation stops processing within one batch

6. Benchmark suite (src/__benchmarks__/)

• Array source → map transform → array sink, 10k records
• Array source → validate transform → array sink, 10k records
• Batch transform with batch size 100 vs 1000

---

File layout

universal/data/etl/
  package.json
  tsconfig.json
  vitest.config.ts
  src/
    index.ts
    error-codes.ts
    constants/
      index.ts
    types/
      common.ts
      common.md
    core/
      pipeline/
        pipeline.ts
        pipeline.md
        pipeline.types.ts
        pipeline.test.ts
        index.ts
    stages/
      source/
        source.ts
        source.md
        source.types.ts
        source.test.ts
        index.ts
      transform/
        transform.ts
        transform.md
        transform.types.ts
        transform.test.ts
        index.ts
      sink/
        sink.ts
        sink.md
        sink.types.ts
        sink.test.ts
        index.ts
    adapters/
      array-source/
        array-source.ts
        array-source.md
        array-source.test.ts
        index.ts
      stream-source/
        stream-source.ts
        stream-source.md
        stream-source.test.ts
        index.ts
      array-sink/
        array-sink.ts
        array-sink.md
        array-sink.test.ts
        index.ts
    utils/
      retry-adapter.ts
      retry-adapter.md
      retry-adapter.test.ts
    __benchmarks__/
      pipeline.bench.ts
  tests/
    integration.test.ts
    property.test.ts

---

Impact

Downstream packages: @suaveplan/etl is consumed by any data-migration, import/export, or batch-processing feature. No existing genesis package depends on it yet.

Workspace dep prerequisites: @suaveplan/error, @suaveplan/types, @suaveplan/streams, @suaveplan/serialize, @suaveplan/retry, @suaveplan/events must all be published to Verdaccio before this package can be built.

Breaking changes: none (new package).