Transfer Engine

Warning: This is experimental, pre-production software. The API is subject to change without notice. Use at your own risk in production environments.

A flexible file transfer library for Node.js with support for chunked HTTP uploads, S3 multipart uploads, local file transfers, and streaming hash verification.

Features

• Chunked HTTP Uploads - Resume-capable uploads with configurable chunk sizes
• S3 Multipart Uploads - Native AWS S3 integration with automatic part management
• Local File Transfers - Efficient file-to-file transfers with streaming
• Hash Verification - Streaming BLAKE3 and CRC32 integrity checks
• Transfer Queue - Managed concurrency with configurable limits
• Cancellation Support - Cooperative cancellation for long-running transfers
• ESM Native - Modern ES modules with full tree-shaking support

Installation

npm install dpo-transfer-engine

Requirements

• Node.js >= 18.0.0

Quick Start

Server-Side (Express)

import express from 'express';
import { TransferEngine, TransferFactory, RunState } from 'dpo-transfer-engine/engine';

const app = express();
const engine = new TransferEngine();

// Create a transfer
app.post('/transfers', async (req, res) => {
  const transfer = await engine.createTransfer({
    type: 'http-upload-to-local',
    payload: {
      destinationPath: '/uploads',
      fileName: req.body.fileName,
      fileSize: req.body.fileSize,
    }
  });
  res.json(transfer);
});

// Upload a chunk
app.put('/transfers/:id/chunks/:index', async (req, res) => {
  const chunks: Buffer[] = [];
  req.on('data', chunk => chunks.push(chunk));
  req.on('end', async () => {
    const result = await engine.submitChunk(req.params.id, {
      index: parseInt(req.params.index),
      data: Buffer.concat(chunks),
    });
    res.json(result);
  });
});

// Finalize upload
app.post('/transfers/:id/finalize', async (req, res) => {
  const result = await engine.finalizeTransfer(req.params.id, req.body.hashes);
  res.json(result);
});

app.listen(3000);

Client-Side (Browser/Node)

import { ChunkUploader, executeFullUpload } from 'dpo-transfer-engine/client';

// Simple full upload
const result = await executeFullUpload(file, {
  onProgress: (progress) => {
    console.log(`${progress.percent}% complete`);
  }
});

// Or use ChunkUploader for more control
const uploader = new ChunkUploader({
  file,
  chunkSize: 5 * 1024 * 1024, // 5MB chunks
  onProgress: (progress) => console.log(progress),
  onComplete: (result) => console.log('Done!', result),
  onError: (error) => console.error(error),
});

await uploader.start();

Subpath Imports

The library is organized into subpaths for optimal tree-shaking:

// Full library (includes everything)
import { TransferEngine, ChunkUploader } from 'dpo-transfer-engine';

// Server-side only (smaller bundle)
import { TransferEngine, TransferFactory } from 'dpo-transfer-engine/engine';

// Client-side only (browser compatible)
import { ChunkUploader, executeFullUpload } from 'dpo-transfer-engine/client';

Transfer Types

HTTP Upload to Local

Receives chunked uploads via HTTP and writes to local filesystem.

const transfer = await engine.createTransfer({
  type: 'http-upload-to-local',
  payload: {
    destinationPath: '/uploads',
    fileName: 'document.pdf',
    fileSize: 1024000,
  }
});

Local to S3

Transfers a local file to an S3 bucket using multipart upload.

const transfer = await engine.createTransfer({
  type: 'local-to-s3',
  payload: {
    sourcePath: '/data/backup.zip',
    bucket: 'my-bucket',
    key: 'backups/backup.zip',
    credentials: {
      accessKeyId: process.env.AWS_ACCESS_KEY_ID,
      secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
      region: 'us-east-1',
    }
  }
});

Local File to File

Copies a file from one local path to another.

const transfer = await engine.createTransfer({
  type: 'local-file-to-file',
  payload: {
    sourcePath: '/data/original.bin',
    destinationPath: '/backup/copy.bin',
  }
});

API Reference

TransferEngine

The main orchestrator for managing transfers.

const engine = new TransferEngine({
  maxConcurrency: 2,        // Max concurrent transfers
  logger: new ConsoleLogger(), // Optional custom logger
  persistence: new InMemoryPersistenceStore(), // Optional persistence
});

// Create a transfer
const transfer = await engine.createTransfer(input);

// Get transfer status
const status = await engine.getTransfer(transferId);

// Cancel a transfer
await engine.cancelTransfer(transferId);

// List all transfers
const transfers = await engine.listTransfers();

ChunkUploader

Client-side chunked upload manager.

const uploader = new ChunkUploader({
  file: File | Blob,           // File to upload
  chunkSize: 5 * 1024 * 1024,  // Chunk size in bytes
  maxRetries: 3,               // Retries per chunk
  hashMethod: 'blake3',        // 'blake3' | 'crc32' | 'none'
  onProgress: (progress) => {},
  onComplete: (result) => {},
  onError: (error) => {},
});

await uploader.start();
uploader.pause();
uploader.resume();
uploader.cancel();

Custom Transfer Types

Extend TransferBase to create custom transfer types:

import { TransferBase, EndpointSourceBase, EndpointDestinationBase } from 'dpo-transfer-engine/engine';

class MyCustomTransfer extends TransferBase {
  protected onDefineEndpoints(): void {
    this.source = new MySourceEndpoint(this.context);
    this.destination = new MyDestinationEndpoint(this.context);
  }

  protected async onExecute(): Promise<void> {
    await this.source.start();
    await this.destination.start();
    await this.destination.waitForCompletion();
  }
}

// Register with factory
TransferFactory.registerTransfer('my-custom', MyCustomTransfer);

Hash Verification

The library supports streaming hash verification using BLAKE3 and CRC32:

import { Hasher } from 'dpo-transfer-engine/engine';

const hasher = new Hasher({ method: 'blake3' });
hasher.update(chunk1);
hasher.update(chunk2);
const hash = hasher.finalize(); // Returns hex string

Error Handling

Transfers provide detailed error information:

const status = await engine.getTransfer(transferId);

if (status.state === RunState.FAILED) {
  console.log(status.error?.code);    // Error code
  console.log(status.error?.message); // Human-readable message
  console.log(status.error?.scope);   // 'source' | 'destination' | 'transfer'
}

Documentation

For detailed usage and API reference, see the guides in the docs/ directory:

• User Guide - Getting started, quick start examples, and basic usage
• Integration Guide - Complete API reference, configuration options, error handling, and advanced topics

Examples

See the examples/ directory for complete working examples:

• examples/express-server/ - Express.js server with transfer routes
• examples/react-client/ - React UI with upload progress

License

MIT