npm package: @rebarxyz/pipeline
Description: Simplified data pipeline framework for Rebar
Author: Rebar
Latest version: 0.0.28

# @rebarxyz/pipeline

Simplified data pipeline framework for Rebar. Build data sources with minimal code.

## Features

- ✅ **Automatic batching** - All updates bundled into single transaction
- ✅ **Change detection** - Only submits when values actually change
- ✅ **Multiple schedules** - Timer-based, block-based, or event-based
- ✅ **Multi-asset support** - Track multiple assets/pools/tokens with one definition
- ✅ **Type-safe** - Full TypeScript support
- ✅ **Minimal boilerplate** - Focus on your data logic, not infrastructure

## Installation

```bash
npm install @rebarxyz/pipeline
```

## Quick Start

### Simple Timer-Based Data Source

```typescript
import { RebarClient } from '@rebarxyz/sdk';
import { SimplePipeline, defineSource } from '@rebarxyz/pipeline';

const businessMetrics = defineSource({
  name: 'Daily Business Metrics',

  outputs: {
    daily_revenue: 'float',
    active_users: 'uint256'
  },

  schedule: 'every 1 hour',

  async fetch() {
    const db = await connectToDatabase();

    return {
      daily_revenue: await getRevenue(db),
      active_users: await getActiveUsers(db)
    };
  }
});

const client = new RebarClient(url, wallet, owner);
const pipeline = new SimplePipeline(client, owner, [businessMetrics]);

await pipeline.start();
```

### Block-Based Data Source

```typescript
const ethGas = defineSource({
  name: 'Ethereum Gas Metrics',

  outputs: {
    gas_limit: 'uint256',
    gas_used: 'uint256',
    base_fee: 'uint256'
  },

  schedule: 'every block',

  async fetch({ trigger }) {
    if (trigger.type !== 'block') return null;

    return {
      gas_limit: trigger.block.gasLimit.toString(),
      gas_used: trigger.block.gasUsed.toString(),
      base_fee: (trigger.block.baseFeePerGas ?? 0n).toString()
    };
  }
});

const pipeline = new SimplePipeline(client, owner, [ethGas], {
  ethRpcUrl: process.env.ETH_RPC_URL, // HTTP endpoint
  ethWsUrl: process.env.ETH_WS_URL    // WebSocket endpoint (optional but recommended)
});

await pipeline.start();
```

### Multi-Asset Protocol Tracking

```typescript
import { defineMultiAssetSource } from '@rebarxyz/pipeline';

const TOKENS = {
  weth: { address: '0x...', name: 'WETH' },
  usdc: { address: '0x...', name: 'USDC' }
};

const aavePools = defineMultiAssetSource({
  name: 'Aave V3 Pools',

  assets: TOKENS,

  outputsPerAsset: (token, key) => ({
    [`${key}_supply_apr`]: {
      type: 'float',
      name: `${token.name} Supply APR`,
      unit: 'percent'
    }
  }),

  schedule: 'every block',

  async fetchAsset(token, key, { clients }) {
    const apr = await getAaveAPR(clients.ethereum, token.address);
    return { [`${key}_supply_apr`]: apr };
  }
});
```

## What the Framework Does Automatically

1. **Creates data nodes** - On first run, all outputs are created on Rebar
2. **Reuses existing nodes** - Loads inventory to avoid duplicate creation
3. **Schedules updates** - Sets up timers, block watchers, or event listeners
4. **Detects changes** - Compares new values to current state
5. **Batches transactions** - Combines all updates into single transaction
6. **Handles errors** - Continues running even if individual updates fail

## Inventory Management

The framework uses inventory files to track created data nodes, making deployments idempotent and preventing duplicate node creation.

### Basic Usage

```typescript
import { SimplePipeline, loadInventory, saveInventory } from '@rebarxyz/pipeline';

// Load existing inventory (returns null if doesn't exist)
const inventory = await loadInventory('./data/inventory.json');

// Create pipeline with existing inventory
const pipeline = new SimplePipeline(client, owner, sources, {
  existingInventory: inventory ?? undefined,
  ethRpcUrl: process.env.ETH_RPC_URL,
  ethWsUrl: process.env.ETH_WS_URL
});

await pipeline.start();

// Save inventory after first run
if (!inventory) {
  await saveInventory('./data/inventory.json', pipeline.getInventory());
}
```

### How It Works

- **First run**: No inventory exists → creates all nodes → saves inventory
- **Subsequent runs**: Loads inventory → reuses existing nodes → no creation costs
- **Adding sources**: Missing nodes detected → creates only new nodes → updates inventory

The inventory is a simple JSON file mapping output keys to addresses:

```json
{
  "daily_revenue": {
    "address": "rebar1...",
    "owner": "rebar1..."
  },
  "active_users": {
    "address": "rebar1...",
    "owner": "rebar1..."
  }
}
```

## API Reference

### `defineSource(config)`

Create a simple data source.

**Config:**
- `name` - Human-readable name
- `outputs` - Map of output keys to types or detailed configs
- `schedule` - When to update (`'every 1 hour'`, `'every block'`, or event config)
- `fetch` - Function that returns data
- `onError` - Optional error handler

### `defineMultiAssetSource(config)`

Create a multi-asset data source (for tracking protocols with multiple assets/pools).

**Config:**
- `name` - Human-readable name
- `assets` - Map of assets to track
- `outputsPerAsset` - Function that generates outputs for each asset
- `schedule` - When to update
- `fetchAsset` - Function that fetches data for one asset
- `onAssetError` - Optional per-asset error handler

### `SimplePipeline`

Main orchestrator that manages sources.

**Methods:**
- `start()` - Create nodes and begin scheduling
- `stop()` - Stop all timers and watchers
- `addSource(source)` - Add a data source
- `removeSource(id)` - Remove a data source
- `getInventory()` - Get all created data nodes
- `getSourceState(id)` - Get state of a specific source

### `SetupHelper`

Utility for one-time setup of functions and calculations.

**Methods:**
- `createFunction(config)` - Create a function node
- `createCalculation(config)` - Create a calculation node
- `saveAddresses(path, addresses)` - Save addresses to JSON file
- `loadAddresses(path)` - Load addresses from JSON file

## Examples

See the `/examples` directory for complete examples:

- `fed-macro.ts` - Fetching economic data from FRED API
- `eth-gas.ts` - Block-based Ethereum gas metrics
- `aave-pools.ts` - Multi-asset protocol tracking

## Advanced Usage

### Using Functions in Calculations

```typescript
import { SetupHelper } from '@rebarxyz/pipeline';

const setup = new SetupHelper(client, owner);

// Create utility function
const aprToApy = await setup.createFunction({
  name: 'APR to APY',
  calculationType: 'starlark',
  calculation: `
def apr_to_apy(apr, compounds_per_year):
  return pow((1 + apr / compounds_per_year), compounds_per_year) - 1
  `
});

// Reference in data source
const source = defineSource({
  name: 'Staking Metrics',
  functions: {
    apr_to_apy: aprToApy.address
  },
  // ... rest of config
});
```

### Conditional Updates

Return `null` from `fetch()` to skip an update:

```typescript
async fetch({ trigger, currentValues }) {
  // Only update if significant change
  const newValue = await getValue();
  if (Math.abs(newValue - currentValues.myMetric) < 0.01) {
    return null; // Skip update
  }

  return { myMetric: newValue };
}
```

### Error Handling

```typescript
const source = defineSource({
  // ... config

  onError(error) {
    console.error('Failed:', error);
    // Send alert, log to monitoring service, etc.
  }
});
```

## License

MIT