# 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

```ts
// 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

```ts
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

```ts
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

```ts
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

```ts
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

```ts
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

```text
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).