@pipe/node-sdk

SDK for building ETL stages in Node.js/TypeScript, with built-in codegen CLI.

Installation

npm install @pipe/node-sdk

Usage

Writing Stages

import { StageRunner, StageContext, successResult } from '@pipe/node-sdk';

const myStage: StageRunner = async (context: StageContext) => {
  const { logger, params, inputs } = context;
  
  // Your stage logic here
  return successResult({ output: 'result' });
};

export default myStage;

Generating Types

The SDK includes a CLI tool for generating TypeScript types from proto contracts:

# Generate types once
npx pipe-codegen --proto-dir=../torv-proto --stage-dir=./stages/my_stage

# Watch mode (regenerates on changes)
npx pipe-codegen --proto-dir=../torv-proto --stage-dir=./stages/my_stage --watch

The CLI will:

1. Look for a proto service matching your stage name (e.g., MyStage service for my_stage stage)
2. Extract input/output contracts from the service
3. Generate TypeScript types in .etl/generated/

Using Generated Types

import { Inputs, Outputs } from './.etl/generated';
import { StageRunner, successResult } from '@pipe/node-sdk';

const myStage: StageRunner = async (context) => {
  // Full type safety!
  const input: Inputs['primary'] = context.inputs.primary;
  const output: Outputs['result'] = transform(input);
  
  return successResult({ result: output });
};

CLI Options

pipe-codegen [generate] --proto-dir=<dir> --stage-dir=<dir> [--watch]

Options:
  --proto-dir=<dir>    Directory containing .proto files
  --stage-dir=<dir>    Stage directory (where .etl/generated will be created)
  --watch, -w          Watch for changes and regenerate

Proto Contract Format

Define your stage contract using a proto service:

service MyStage {
    rpc Execute(MyStageInput) returns (MyStageOutput);
}

message MyStageInput {
    Address primary = 1;
}

message MyStageOutput {
    EnrichedAddress result = 1;
}

The service name must match your stage name in PascalCase (e.g., my_stage → MyStage).