npm package: @neo-solution/langgraph-checkpoint-dynamodb
Description: A robust, production-ready DynamoDB checkpoint saver for LangGraph.js workflows
Author: Neo Solution
Latest version: 0.1.0

# @neo-solution/langgraph-checkpoint-dynamodb

[![npm version](https://badge.fury.io/js/@neo-solution%2Flanggraph-checkpoint-dynamodb.svg)](https://badge.fury.io/js/@neo-solution%2Flanggraph-checkpoint-dynamodb)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

A production-ready DynamoDB checkpoint saver for [LangGraph.js](https://github.com/langchain-ai/langgraphjs) workflows.

## Table of Contents

- [Overview](#overview)
- [Installation](#installation)
- [Quick Start](#quick-start)
- [Configuration](#configuration)
- [API Reference](#api-reference)
- [Development](#development)
- [Contributing](#contributing)

## Overview

Persist LangGraph workflow state in AWS DynamoDB with high performance and reliability.

**Key Features:**

- ✅ Production-ready with comprehensive error handling
- ✅ TypeScript support with full type safety
- ✅ Optimized batch operations and efficient key design
- ✅ Cross-namespace checkpoint support
- ✅ Configurable serialization and AWS client options
- ✅ Optional smart gzip compression with backwards compatibility

## Installation

```bash
npm install @neo-solution/langgraph-checkpoint-dynamodb
```

**Requirements:**

- Node.js 18+
- AWS credentials configured (via AWS CLI, environment variables, or IAM roles)
- DynamoDB table (created using the provided helper or manually)

**Required AWS Permissions:**

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:DescribeTable",
        "dynamodb:PutItem",
        "dynamodb:GetItem",
        "dynamodb:Query",
        "dynamodb:BatchWriteItem",
        "dynamodb:DeleteItem"
      ],
      "Resource": [
        "arn:aws:dynamodb:*:*:table/your-table-name",
        "arn:aws:dynamodb:*:*:table/your-table-name/index/*"
      ]
    }
  ]
}
```

Add `dynamodb:CreateTable` if you plan to use the `createTable` helper function.

## Quick Start

### 1. Create DynamoDB Table

```typescript
import { createTable } from '@neo-solution/langgraph-checkpoint-dynamodb';

// Create table with default settings
await createTable();

// Or with custom options
await createTable({
  tableName: 'my-checkpoints',
  billingMode: 'PAY_PER_REQUEST',
});
```

### 2. Basic Usage

```typescript
import { DynamoDBSaver } from '@neo-solution/langgraph-checkpoint-dynamodb';

// Using default settings
const checkpointer = new DynamoDBSaver();

// With custom configuration
const checkpointer = new DynamoDBSaver({
  tableName: 'my-checkpoints',
  client: new DynamoDBClient({ region: 'us-east-1' }),
});
```

### 3. LangGraph Integration

```typescript
import { StateGraph, MessagesAnnotation } from '@langchain/langgraph';
import { DynamoDBSaver } from '@neo-solution/langgraph-checkpoint-dynamodb';

const checkpointer = new DynamoDBSaver();

const workflow = new StateGraph(MessagesAnnotation)
  .addNode('model', callModel)
  .addEdge('__start__', 'model')
  .addEdge('model', '__end__')
  .compile({ checkpointer });

// Use with thread persistence
const result = await workflow.invoke(
  { messages: [{ role: 'user', content: 'Hello!' }] },
  { configurable: { thread_id: 'conversation-123' } },
);
```

## Configuration

### DynamoDBSaver Options

```typescript
interface DynamoDBSaverOptions {
  /** DynamoDB client instance (optional) */
  client?: DynamoDBClient;
  /** Table name (default: 'Checkpoints') */
  tableName?: string;
  /** Enable smart gzip compression for checkpoint, metadata, and value fields (default: false) */
  enableCompression?: boolean;
  /** Custom serializer (default: JsonPlusSerializer) */
  serde?: SerializerProtocol;
}
```

### Examples

```typescript
import { DynamoDBClient } from '@aws-sdk/client-dynamodb';

// Production setup with compression enabled (reduces storage costs)
const prodSaver = new DynamoDBSaver({
  client: new DynamoDBClient({ region: 'us-west-2' }),
  tableName: 'prod-workflow-checkpoints',
  enableCompression: true, // Enable smart gzip compression for all data
});

// Production setup without compression (default)
const compatSaver = new DynamoDBSaver({
  client: new DynamoDBClient({ region: 'us-west-2' }),
  tableName: 'prod-workflow-checkpoints',
  enableCompression: false, // Explicit disable (default behavior)
});

// Development with DynamoDB Local
const devSaver = new DynamoDBSaver({
  client: new DynamoDBClient({
    endpoint: 'http://localhost:8000',
    region: 'local',
  }),
  tableName: 'dev-checkpoints',
});
```

### Table Creation Options

```typescript
interface CreateTableOptions {
  client?: DynamoDBClient;
  tableName?: string;
  billingMode?: 'PAY_PER_REQUEST' | 'PROVISIONED';
  readCapacityUnits?: number;
  writeCapacityUnits?: number;
  gsiReadCapacityUnits?: number;
  gsiWriteCapacityUnits?: number;
  tags?: Tag[];
  waitForActive?: boolean;
  waiter?: {
    maxWaitTime?: number;
    minDelay?: number;
    maxDelay?: number;
  };
}
```

## Data Compression

The DynamoDBSaver supports optional smart gzip compression to reduce storage costs and improve performance for large checkpoints.

### Smart Compression Overview

When `enableCompression` is set to `true`, the library uses **smart compression** that automatically optimizes compression decisions:
- **Checkpoint data**: The serialized workflow state
- **Metadata**: Associated checkpoint metadata
- **Write values**: Pending write operation data

**Smart Compression Logic:**
- ✅ **Size Threshold**: Skips compression for data smaller than 100 bytes (overhead not worth it)
- ✅ **Compression Ratio**: Only compresses if it reduces size by at least 10% (90% ratio threshold)
- ✅ **Automatic Optimization**: Uses uncompressed data when compression doesn't provide meaningful benefits

**Key Benefits:**
- ✅ **Reduced Storage Costs**: Up to 70-90% reduction in DynamoDB storage usage for large data
- ✅ **Optimized Performance**: Smart decisions prevent compression overhead for small data
- ✅ **Backwards Compatible**: Uses version metadata to handle mixed compressed/uncompressed data
- ✅ **Transparent**: Compression/decompression happens automatically

### Configuration

```typescript
// Enable compression (recommended for production)
const compressedSaver = new DynamoDBSaver({
  tableName: 'my-checkpoints',
  enableCompression: true,
});

// Disable compression (default, maximum compatibility)
const uncompressedSaver = new DynamoDBSaver({
  tableName: 'my-checkpoints',
  enableCompression: false, // Default value
});
```

### Backwards Compatibility

The library uses **version metadata** to ensure complete backwards compatibility:

- **Mixed Data**: Tables can contain both compressed and uncompressed records
- **Safe Migration**: Changing `enableCompression` doesn't break existing data
- **Automatic Detection**: Each record includes metadata indicating its compression state

### Performance Considerations

| Scenario | Storage Size | Read Performance | Write Performance |
|----------|-------------|------------------|-------------------|
| **Very small data** (<100 bytes) | No compression applied | No overhead (uncompressed) | No overhead (uncompressed) |
| **Small checkpoints** (100B-1KB) | Compressed only if >10% savings | Variable (depends on compression decision) | Variable (depends on compression decision) |
| **Medium checkpoints** (1-10KB) | 50-70% reduction when compressed | Similar performance | Similar performance |
| **Large checkpoints** (>10KB) | 70-90% reduction | **Faster** (less network I/O) | **Faster** (less network I/O) |

**Smart Compression Logic**: The system automatically decides whether to compress based on data size and compression effectiveness, ensuring optimal performance without manual tuning.

**Recommendation**: Enable compression for production workloads with medium to large checkpoint data.

## API Reference

### DynamoDBSaver

Main class for checkpoint management.

#### Constructor

```typescript
new DynamoDBSaver(options?: DynamoDBSaverOptions)
```

#### Methods

**`getTuple(config: RunnableConfig): Promise<CheckpointTuple | undefined>`**

Retrieves the latest checkpoint for a thread, or a specific checkpoint if `checkpoint_id` is provided.

```typescript
// Get latest checkpoint
const latest = await saver.getTuple({
  configurable: { thread_id: 'thread-123' },
});

// Get specific checkpoint in root namespace
const specific = await saver.getTuple({
  configurable: {
    thread_id: 'thread-123',
    checkpoint_id: 'checkpoint-abc',
  },
});
```

**`list(config: RunnableConfig, options?: CheckpointListOptions): AsyncGenerator<CheckpointTuple>`**

Lists checkpoints for a thread with optional filtering and pagination.

```typescript
for await (const checkpoint of saver.list(
  { configurable: { thread_id: 'thread-123' } },
  { limit: 10 },
)) {
  console.log(checkpoint.config.configurable.checkpoint_id);
}
```

**`put(config, checkpoint, metadata, newVersions): Promise<RunnableConfig>`**

Saves a checkpoint with associated metadata and channel versions.

**`putWrites(config, writes, taskId): Promise<void>`**

Saves pending writes for a checkpoint.

### Utility Functions

**`createTable(options?: CreateTableOptions): Promise<void>`**

Creates a DynamoDB table with the required schema.

```typescript
await createTable({
  tableName: 'my-checkpoints',
  billingMode: 'PAY_PER_REQUEST',
  tags: [{ Key: 'Environment', Value: 'production' }],
});
```

### Error Classes

- **`DynamoDBSaverError`**: Base error class for all DynamoDB saver operations
- **`DynamoDBSaverConfigError`**: Configuration validation errors (missing thread_id, invalid config)
- **`DynamoDBSaverKeyError`**: Key validation errors (invalid characters, null values)
- **`DynamoDBSaverSchemaError`**: Table schema validation errors
- **`DynamoDBSaverSerializationError`**: Serialization/deserialization errors

## Development

### Local Development with DynamoDB Local

```bash
# Start DynamoDB Local
docker run -p 8000:8000 amazon/dynamodb-local
```

```typescript
const localSaver = new DynamoDBSaver({
  client: new DynamoDBClient({
    endpoint: 'http://localhost:8000',
    region: 'local',
    credentials: { accessKeyId: 'dummy', secretAccessKey: 'dummy' },
  }),
  tableName: 'dev-checkpoints',
});
```

### Testing

```bash
# Run all tests
pnpm nx test checkpoint-dynamodb

# Run integration tests with DynamoDB Local
RUN_INTEGRATION_TESTS=true \
DYNAMODB_ENDPOINT=http://localhost:8000 \
pnpm nx test checkpoint-dynamodb
```

See [tests/README.md](./tests/README.md) for detailed testing documentation.

### Common Issues

**Missing thread_id:**

```typescript
// ❌ Wrong
const result = await workflow.invoke(data, {});

// ✅ Correct
const result = await workflow.invoke(data, {
  configurable: { thread_id: 'thread-123' },
});
```

**Authentication errors:**
Ensure AWS credentials are configured via AWS CLI, environment variables, or IAM roles.

**Table schema mismatch:**
If you get schema errors, ensure the table was created using the `createTable` helper or matches the required schema.

## Contributing

1. Fork the repository
2. Create a feature branch: `git checkout -b feature/amazing-feature`
3. Make changes and add tests
4. Run tests: `pnpm test`
5. Check linting: `pnpm lint`
6. Commit: `git commit -m 'Add amazing feature'`
7. Push and create a Pull Request

**Code Standards:**

- TypeScript with proper type annotations
- Include both unit and integration tests
- Update documentation for new features
- Follow ESLint and Prettier formatting

---

## License

[MIT License](./LICENSE)