npm package: @robertkirk/jessie
Description: Simple, Type-safe, Observable, & Concurrent pipeline framework. Best for IO bottlenecked workflows.
Author: Robert Kirk
Latest version: 0.6.0

# jessie

[![npm version](https://badge.fury.io/js/@robertkirk%2Fjessie.svg)](https://badge.fury.io/js/@robertkirk%2Fjessie)
[![TypeScript](https://img.shields.io/badge/%3C%2F%3E-TypeScript-%230074c1.svg)](http://www.typescriptlang.org/)

A type-safe, observable pipeline framework with dependency injection for building robust data processing workflows. Best for io bottlenecked workflows like calling external APIs.

## Features

- 🔒 **Type-safe**: Full TypeScript support with context inference across pipeline steps
- ⚡ **Observable**: Comprehensive logging hooks for monitoring and debugging
- ⏱️ **Timeout Support**: Per-step timeout configuration with automatic enforcement
- 🚀 **Enhanced Queue Processing**: Full-featured queue management with priority, pause/resume, and statistics
- 🧩 **Composable**: Support for sub-pipelines with parent-child relationships
- 💉 **Dependency Injected**: No hard dependencies - bring your own logger, database, etc.
- 📊 **Metrics Ready**: Built-in timing, failure reporting, and queue statistics

## When to use Jessie

- You have a workflow that is IO bottlenecked (Jessie is not multithreaded)
- You want a simple way to build a pipeline
- You want logging and observability
- You want typesafety
- You want a simple solution with no dependencies (no Redis or other external services)

## Installation

```bash
npm install @robertkirk/jessie
# or
yarn add @robertkirk/jessie
# or
bun add @robertkirk/jessie
```

## Quick Start

```typescript
import { 
  Pipeline, 
  Queue,
  PipelineOptions,
  createConsoleLogger
} from '@robertkirk/jessie';

// Define your context type (only your data, no framework boilerplate)
interface MyContext {
  userId: string;
}

// Create pipeline options
const opts: PipelineOptions = {
  idGen: () => crypto.randomUUID(),
  logger: createConsoleLogger()
};

// Build your pipeline
const pipeline = new Pipeline<MyContext>('data-processing', opts)
  .step('fetchUser', async (ctx) => {
    // Your async logic here
    return { id: ctx.userId, name: 'John Doe' };
  }, { timeoutMs: 5000 })
  .step('processData', async (ctx) => {
    // Access previous step result via ctx.fetchUser
    return `Processed: ${ctx.fetchUser.name}`;
  })
  .step('saveResult', async (ctx) => {
    // Save the processed data
    console.log('Saving:', ctx.processData);
    return 'saved-id-123';
  });

// Execute the pipeline - only provide your data
const result = await pipeline.run({ userId: 'user-123' });
console.log('Final result:', result.saveResult);

// Or process multiple items with enhanced queue management
const queue = new Queue(pipeline, { concurrency: 2 });
queue.add({ userId: 'user-1' }, 5); // High priority
queue.add({ userId: 'user-2' }, 1); // Low priority
await queue.start();
```

## Core Concepts

### Pipeline

A `Pipeline` is a sequence of typed steps that build up a context object. Each step receives the accumulated context from previous steps and adds its result to the context.

```typescript
const opts: PipelineOptions = {
  idGen: () => crypto.randomUUID(),
  logger: createConsoleLogger()
};

const pipeline = new Pipeline<{ value: number }>('my-pipeline', opts)
  .step('step1', async (ctx) => 'result1')
  .step('step2', async (ctx) => ctx.step1.toUpperCase()) // TypeScript knows ctx.step1 exists
  .step('step3', async (ctx) => `${ctx.step1}-${ctx.step2}`); // And ctx.step2
```

#### Pipeline Visualization and Introspection

Pipelines support detailed introspection for visualization and debugging:

```typescript
// Add steps with descriptions for better visualization
const pipeline = new Pipeline<{ userId: string }>('user-pipeline', opts)
  .step('authenticate', async (ctx) => ({ valid: true, role: 'user' }), 
        { 
          timeoutMs: 3000, 
          description: 'Authenticates the user and determines their role'
        })
  .step('fetchData', async (ctx) => getUserData(ctx.userId), 
        { description: 'Retrieves user data from the database' })
  .step('transform', async (ctx) => processData((ctx as any).fetchData))
  .step('validate', async (ctx) => (ctx as any).transform.length > 0, 
        { 
          concurrency: 1, 
          description: 'Ensures we have valid data to process'
        });

// Get detailed step information for visualization
const stepOrder = pipeline.getStepOrder();
console.log(`Pipeline: ${stepOrder.pipelineName}`);
console.log(`Total steps: ${stepOrder.totalSteps}`);

stepOrder.steps.forEach(step => {
  console.log(`Step ${step.index}: ${step.key}`);
  console.log(`  Input context: ${step.inputContext.join(', ')}`);
  console.log(`  Output type: ${step.outputType}`);
  if (step.description) {
    console.log(`  Description: ${step.description}`);
  }
  if (step.options) {
    console.log(`  Options: ${JSON.stringify(step.options)}`);
  }
});

// Get enhanced type information with sample data
const typedOrder = pipeline.getStepOrderWithTypes({ userId: 'user-123' });
console.log('Initial context keys:', typedOrder.initialContextKeys);
console.log('Estimated types:', typedOrder.estimatedTypes);
```

#### Automatic Pipeline Logging

Pipelines automatically provide detailed logging for all steps using the standard pipeline logger methods. This comprehensive logging helps with debugging, monitoring, and understanding pipeline execution:

```typescript
const pipeline = new Pipeline<{ data: string }>('debug-pipeline', opts)
  .step('process', async (ctx) => {
    // Some complex processing
    return processData(ctx.data);
  }, { 
    description: 'Processes the input data with complex logic'
  });
```

Automatic pipeline logging provides:
- **Complete step visibility**: Every step logs start, completion, and context information
- **Performance insights**: Execution times for all steps
- **Context tracking**: See what data is available at each step
- **Error details**: Comprehensive error information when steps fail
- **Timeout monitoring**: Automatic logging when timeouts are configured

#### User Logging Methods

For custom logging within your pipeline steps, use the `verbose` and `debug` methods available on the logger:

```typescript
const pipeline = new Pipeline<{ userId: string }>('user-pipeline', opts)
  .step('fetchUser', async (ctx) => {
    ctx.log.verbose?.('Fetching user details', ctx.runId);
    const user = await getUserById(ctx.userId);
    ctx.log.debug?.('User data retrieved', ctx.runId);
    return user;
  }, { description: 'Retrieves user from database' });
```

### PipelineOptions

Every pipeline requires options that specify:
- `idGen`: Function to generate unique run IDs
- `logger`: Logger instance for the pipeline

### PipelineRunOptions

When calling `pipeline.run()`, you can optionally provide:
- `runId`: Override the auto-generated run ID
- `parentName`: Specify parent pipeline name for sub-pipeline tracking

```typescript
const result = await pipeline.run(context, {
  runId: 'custom-id',
  parentName: 'parent-pipeline'
});
```

### Context

Every pipeline step receives a readonly context that includes:
- Your custom data
- `runId`: Unique identifier for this pipeline execution (added automatically)
- `log`: Logger instance for the pipeline (added automatically)
- Results from all previous steps (automatically typed)

### Logging & Observability

Implement the `PipelineLogger` interface to capture pipeline events:

```typescript
interface PipelineLogger {
  logi(step: string, msg: string): void;
  logw(step: string, msg: string): void;
  loge(step: string, msg: string): void;
  onStepStart?(step: string, runId: string, meta?: PipelineMeta): void;
  onStepEnd?(step: string, runId: string, ms: number, error?: Error, meta?: PipelineMeta): void;
  onFailure?(step: string, runId: string, error: Error, meta?: PipelineMeta): void;
}
```

### Sub-pipelines

Create reusable sub-pipelines and execute them within main pipelines:

```typescript
const subOptions: PipelineOptions = {
  idGen: () => crypto.randomUUID(),
  logger: createConsoleLogger()
};

const subPipeline = new Pipeline<MyContext & { data: any }>('transform', subOptions)
  .step('validate', async (ctx) => validateData(ctx.data))
  .step('clean', async (ctx) => cleanData(ctx.validate));

const mainOptions: PipelineOptions = {
  idGen: () => crypto.randomUUID(),
  logger: createConsoleLogger()
};

const mainPipeline = new Pipeline<MyContext>('main', mainOptions)
  .step('fetchData', async (ctx) => getData(ctx.userId))
  .step('transform', async (ctx) => {
    const result = await subPipeline.run({
      ...ctx,
      data: ctx.fetchData
    }, { parentName: 'main' }); // Specify parent for logging
    return result.clean;
  });
```

### Queue Processing

Jessie provides two approaches to queue processing:

#### Simple Queue Processing

For basic use cases, use the `runQueue` function:

```typescript
import { runQueue, createArrayQueue } from '@robertkirk/jessie';

const items = [
  { userId: 'user1' },
  { userId: 'user2' },
  { userId: 'user3' }
];

const queue = createArrayQueue(items);
await runQueue(queue, pipeline, { concurrency: 2 });
```

#### Enhanced Queue Management

For advanced use cases, use the `Queue` class with full queue management capabilities:

```typescript
import { Queue, Pipeline } from '@robertkirk/jessie';

// Create a queue with your pipeline
const queue = new Queue(pipeline, { concurrency: 3 });

// Add items with optional priority (higher numbers = higher priority)
const taskId1 = queue.add({ userId: 'user1' }, 1);
const taskId2 = queue.add({ userId: 'user2' }, 5); // High priority
const taskId3 = queue.add({ userId: 'user3' }, 3);

// Add multiple items at once
const taskIds = queue.addMany([
  { userId: 'user4' },
  { userId: 'user5' }
], 2);

// Manage queue items
queue.remove(taskId1);                    // Remove specific item
queue.reorder(taskId2, 0);               // Move item to front
queue.setPriority(taskId3, 10);          // Change priority
queue.clear();                           // Remove all pending items

// Monitor queue status
const status = queue.getStatus();
console.log(`${status.pending} pending, ${status.processing} processing`);

// Get detailed statistics
const stats = queue.getStats();
console.log(`Processed: ${stats.totalProcessed}, Failed: ${stats.totalFailed}`);
console.log(`Avg time: ${stats.averageProcessingTime}ms`);
console.log(`Throughput: ${stats.throughputPerSecond} items/sec`);

// Control processing
await queue.start();                     // Start processing
queue.pause();                          // Pause processing
queue.resume();                         // Resume processing
await queue.stop();                     // Stop gracefully

// Dynamic concurrency control
queue.setConcurrency(5);                // Change concurrency on the fly
const currentConcurrency = queue.getConcurrency(); // Get current setting

// Pipeline timeout control
queue.setTimeoutMs(30000);              // Set overall pipeline timeout
const timeout = queue.getTimeoutMs();   // Get current timeout setting

// Wait for completion
await queue.waitUntilEmpty();           // Wait until all items processed
```

#### Queue Features

- **Priority Processing**: Items with higher priority values are processed first
- **Dynamic Management**: Add, remove, and reorder items while processing
- **Dynamic Concurrency**: Change concurrency limits on the fly with graceful scaling
- **Pipeline Timeout**: Set overall timeout limits for the entire queue processing
- **Pause/Resume**: Temporarily halt processing without losing progress
- **Real-time Statistics**: Monitor processing speed, success/failure rates
- **Graceful Shutdown**: Stop processing while allowing current items to complete
- **Status Monitoring**: Track pending, processing, completed, and failed items

### Error Handling & Timeouts

```typescript
const opts: PipelineOptions = {
  idGen: () => crypto.randomUUID(),
  logger: createConsoleLogger()
};

const pipeline = new Pipeline<MyContext>('error-handling', opts)
  .step('riskyOperation', async (ctx) => {
    // This step will timeout after 1 second
    await longRunningOperation();
    return 'success';
  }, { timeoutMs: 1000 })
  .step('cleanup', async (ctx) => {
    // This step only runs if the previous step succeeded
    return 'cleaned';
  });

try {
  await pipeline.run({ userId: 'user-123' });
} catch (error) {
  // Handle TimeoutError or other failures
  console.error('Pipeline failed:', error.message);
}
```

## API Reference

### Classes

#### `Pipeline<C extends Record<string, any>>`

Main pipeline class for building and executing step sequences.

**Constructor:**
```typescript
new Pipeline<C>(name: string, options: PipelineOptions, parentName?: string)
```

**Methods:**
- `step<K, V>(key: K, fn: StepFunction<C, V>, opts?: StepOptions): Pipeline<C & { [K]: V }>`
- `run(ctx: C, options?: PipelineRunOptions): Promise<C>`
- `getMeta(): PipelineMeta`
- `getStepCount(): number`
- `getStepNames(): string[]`
- `getStepOrder(): PipelineStepOrder` - Get detailed step information for visualization
- `getStepOrderWithTypes<T>(sampleContext?: T): PipelineStepOrder & { initialContextKeys: string[]; estimatedTypes: Record<string, string>; }` - Get enhanced step information with type details

#### `Queue<T extends Record<string, any>>`

Enhanced queue management with full control over processing.

**Constructor:**
```typescript
new Queue<T>(pipeline: Pipeline<T>, options?: QueueOptions)
```

**Methods:**
- `add(data: T, priority?: number): string` - Add item with optional priority
- `addMany(items: T[], priority?: number): string[]` - Add multiple items
- `remove(id: string): boolean` - Remove item by ID
- `clear(): void` - Remove all pending items
- `reorder(id: string, newIndex: number): boolean` - Reorder item position
- `setPriority(id: string, priority: number): boolean` - Update item priority
- `getStatus(): QueueStatus` - Get current queue status
- `getStats(): QueueStats` - Get processing statistics
- `getPendingItems(): QueueItem<T>[]` - Get all pending items
- `getItem(id: string): QueueItem<T> | null` - Get item by ID
- `start(): Promise<void>` - Start processing
- `stop(): Promise<void>` - Stop processing gracefully
- `pause(): void` - Pause processing
- `resume(): void` - Resume processing
- `isRunning(): boolean` - Check if processing
- `isPaused(): boolean` - Check if paused
- `isStopped(): boolean` - Check if stopped
- `waitUntilEmpty(): Promise<void>` - Wait for completion
- `getConcurrency(): number` - Get current concurrency setting
- `setConcurrency(concurrency: number): void` - Change concurrency dynamically
- `getTimeoutMs(): number | undefined` - Get pipeline timeout setting
- `setTimeoutMs(timeoutMs?: number): void` - Set pipeline timeout

#### `QueueProcessor<T extends Record<string, any>>`

Legacy queue processor with basic start/stop controls.

### Functions

- `runQueue<T>(getNext: () => Promise<T | null>, pipeline: Pipeline<T>, options?: QueueOptions): Promise<void>`
- `createArrayQueue<T>(items: T[]): () => Promise<T | null>`
- `createConsoleLogger(): PipelineLogger`
- `createNoOpLogger(): PipelineLogger`
- `createTimer(): StepTimer`
- `withTimeout<T>(promise: Promise<T>, timeoutMs: number, stepName: string): Promise<T>`

### Types

- `PipelineOptions`: Configuration for creating pipelines
- `PipelineLogger`: Interface for logging implementations
- `StepOptions`: Configuration for individual steps (timeoutMs, concurrency, description)
- `StepFunction<C, V>`: Function signature for pipeline steps
- `QueueOptions`: Configuration for queue processing (concurrency, timeoutMs)
- `QueueItem<T>`: Queue item with metadata (id, data, priority, addedAt, attempts, lastError)
- `QueueStatus`: Current queue status (total, pending, processing, completed, failed, isRunning, isPaused)
- `QueueStats`: Processing statistics (totalProcessed, totalFailed, averageProcessingTime, throughputPerSecond, startedAt, lastProcessedAt)
- `StepInfo`: Individual step information for visualization (key, index, inputContext, outputType, options, description)
- `PipelineStepOrder`: Complete pipeline step order information (pipelineName, totalSteps, steps, contextFlow)
- `TimeoutError`: Error thrown when steps exceed timeout

## License

MIT

## Contributing

Contributions are welcome! Please feel free to submit a Pull Request.