Package Exports

chronos-db

Readme

Unified Data Manager

S3-agnostic, cost-first & stability-first unified persistence layer for MongoDB + S3-compatible storage

📖 Overview

unified-data-manager provides a production-ready persistence layer that combines:

MongoDB for indexed metadata, head pointers, and bounded recent version index
S3-compatible storage for authoritative payloads, full JSON per version
Automatic versioning with explicit restore capabilities
Multi-backend routing with connection pooling
Cheap analytics with conditional counters
Enrichment API for incremental updates
Fallback queues for guaranteed durability
Write optimization for high-throughput scenarios

Key Principles

✅ No Environment Variables - All configuration via JSON
✅ Cost-First - Minimize storage and compute costs
✅ Stability-First - Immutable versioning, transactions, optimistic locking
✅ Portable - Works with any S3-compatible provider
✅ Type-Safe - Full TypeScript support with Zod validation

🚀 Quick Start

Installation

npm install unified-data-manager

Basic Usage

import { initUnifiedDataManager } from 'unified-data-manager';

const udm = initUnifiedDataManager({
  mongoUris: ['mongodb://localhost:27017'],
  spacesConns: [{
    endpoint: 'https://nyc3.digitaloceanspaces.com',
    region: 'nyc3',
    accessKey: 'YOUR_ACCESS_KEY',
    secretKey: 'YOUR_SECRET_KEY',
    backupsBucket: 'udm-backups',
    jsonBucket: 'udm-json',
    contentBucket: 'udm-content',
  }],
  counters: {
    mongoUri: 'mongodb://localhost:27017',
    dbName: 'udm_counters',
  },
  routing: {
    hashAlgo: 'rendezvous',
  },
  retention: {},
  rollup: {},
  collectionMaps: {
    users: {
      indexedProps: ['email', 'status'],
      validation: {
        requiredIndexed: ['email'],
      },
    },
  },
});

// Context-bound operations
const ops = udm.with({
  dbName: 'myapp',
  collection: 'users',
  tenantId: 'tenant123',
});

// Create
const result = await ops.create({
  email: 'user@example.com',
  status: 'active',
}, 'system', 'user signup');

// Update
await ops.update(result.id, {
  status: 'verified',
}, result.ov, 'system', 'email verified');

// Read latest
const user = await ops.getLatest(result.id);

// Restore to previous version
await ops.restoreObject(result.id, { ov: 0 });

// Enrich incrementally
await ops.enrich(result.id, {
  tags: ['vip'],
  metadata: { score: 100 },
}, { functionId: 'scorer@v1' });

// Shutdown
await udm.admin.shutdown();

🎯 Core Features

1. CRUD Operations

Full transaction support with optimistic locking:

// Create with automatic versioning (ov=0)
const created = await ops.create(data, 'actor', 'reason');
// Returns: { id, ov: 0, cv: 0, createdAt }

// Update with optimistic lock
const updated = await ops.update(id, newData, expectedOv, 'actor', 'reason');
// Returns: { id, ov: 1, cv: 1, updatedAt }

// Logical delete (default)
const deleted = await ops.delete(id, expectedOv, 'actor', 'reason');
// Returns: { id, ov: 2, cv: 2, deletedAt }

2. Enrichment API

Incrementally augment records without full rewrite:

// Deep merge with array union
await ops.enrich(id, {
  tags: ['premium'],              // Arrays unioned
  metadata: { newField: 'value' }, // Objects deep merged
}, {
  functionId: 'enricher@v1',       // Provenance tracking
  actor: 'system',
  reason: 'automated enrichment',
});

// Batch enrichment
await ops.enrich(id, [
  { tags: ['vip'] },
  { metadata: { score: 100 } },
  { tags: ['verified'] },
]);

3. Read Operations

Multiple read strategies with presigned URL support:

// Get latest version
const latest = await ops.getLatest(id, { 
  presign: true,
  ttlSeconds: 3600,
  projection: ['email', 'status'],
});

// Get specific version
const v1 = await ops.getVersion(id, 1);

// Get as of time
const historical = await ops.getAsOf(id, '2025-09-01T00:00:00Z');

// List by metadata with pagination
const results = await ops.listByMeta({
  filter: { status: 'active' },
  limit: 50,
  afterId: lastId,
  sort: { updatedAt: -1 },
}, { presign: true });

4. Restore Operations

Explicit, append-only restore:

// Restore object to specific version
await ops.restoreObject(id, { ov: 5 });
// or by time
await ops.restoreObject(id, { at: '2025-09-01T00:00:00Z' });

// Restore entire collection
await ops.restoreCollection({ cv: 100 });
// or by time
await ops.restoreCollection({ at: '2025-09-01T00:00:00Z' });

5. Counters & Analytics

Cheap, always-on totals:

// Configure conditional counters
const config = {
  // ... other config
  counterRules: {
    rules: [
      {
        name: 'activeUsers',
        when: { status: 'active' },
        on: ['CREATE', 'UPDATE'],
        scope: 'meta',
      },
    ],
  },
};

// Query totals
const totals = await udm.counters.getTotals({
  dbName: 'myapp',
  collection: 'users',
});

// Returns:
// {
//   created: 1000,
//   updated: 500,
//   deleted: 50,
//   activeUsers: 750,
// }

6. Fallback Queues

Guaranteed durability with automatic retry:

// Enable fallback queues
const config = {
  // ... other config
  fallback: {
    enabled: true,
    maxAttempts: 10,
    baseDelayMs: 2000,
    maxDelayMs: 60000,
    deadLetterCollection: 'udm_fallback_dead',
  },
};

// Start worker for automatic retries
await udm.fallback?.startWorker();

// Monitor queue
const stats = await udm.fallback?.getQueueStats();
console.log('Pending ops:', stats.queueSize);
console.log('Dead letters:', stats.deadLetterSize);

// Retry dead letter operation
const deadLetters = await udm.fallback?.getDeadLetterOps({}, 10);
for (const op of deadLetters) {
  await udm.fallback?.retryDeadLetter(op._id.toString());
}

// Stop worker
await udm.fallback?.stopWorker();

7. Write Optimization

Reduce I/O overhead under load:

const config = {
  // ... other config
  writeOptimization: {
    batchS3: true,              // Batch S3 uploads
    batchWindowMs: 100,         // 100ms window
    debounceCountersMs: 1000,   // Update counters every 1s
    allowShadowSkip: true,      // Skip shadows for heavy ops
  },
};

// Monitor optimizer
const stats = udm.fallback?.getOptimizerStats();
console.log('S3 queue:', stats.s3QueueSize);
console.log('Counter queue:', stats.counterQueueSize);

🏗️ Architecture

Data Flow

┌─────────────┐
│   Client    │
└──────┬──────┘
       │
       ▼
┌─────────────────────────────────┐
│  Unified Data Manager (UDM)     │
│  ┌───────────────────────────┐  │
│  │  Router (HRW Hashing)     │  │
│  └───────────────────────────┘  │
│          │           │           │
│          ▼           ▼           │
│  ┌──────────┐  ┌──────────┐     │
│  │  Mongo   │  │    S3    │     │
│  │ (Indexed)│  │(Payloads)│     │
│  └──────────┘  └──────────┘     │
│                                  │
│  ┌───────────────────────────┐  │
│  │  Fallback Queue (Optional)│  │
│  └───────────────────────────┘  │
└─────────────────────────────────┘

MongoDB Collections

<collection>_head - Latest state pointers
<collection>_ver - Immutable version index
<collection>_counter - Collection version counter
cnt_total - Counter totals (in separate DB)
udm_fallback_ops - Fallback queue (if enabled)
udm_fallback_dead - Dead letter queue (if enabled)

S3 Storage Layout

<jsonBucket>/
  <collection>/
    <itemId>/
      v0/item.json
      v1/item.json
      v2/item.json

<contentBucket>/
  <collection>/
    <itemId>/
      v0/
        <property>/blob.bin
        <property>/text.txt
      v1/
        <property>/blob.bin

🔐 Production Deployment

MongoDB Replica Set (REQUIRED)

⚠️ MongoDB MUST run as a 3-node replica set in production

# Example docker-compose.yml
services:
  mongo1:
    image: mongo:6
    command: mongod --replSet rs0
    
  mongo2:
    image: mongo:6
    command: mongod --replSet rs0
    
  mongo3:
    image: mongo:6
    command: mongod --replSet rs0

Connection string:

mongodb://mongo1:27017,mongo2:27017,mongo3:27017/dbname?replicaSet=rs0

S3-Compatible Providers

Tested with:

✅ AWS S3
✅ DigitalOcean Spaces
✅ MinIO
✅ Cloudflare R2

📚 Documentation

Implementation Status - Complete feature matrix
Configuration Guide - Detailed configuration
Architecture Plan - Master workplan
Extensions - System fields & shadows
Enrichment API - Deep merge semantics

🧪 Testing

# Build
npm run build

# Run tests
npm test

# Unit tests only
npm run test:unit

# Integration tests only
npm run test:integration

# Type check
npm run type-check

🤝 Contributing

Contributions welcome! Please ensure:

TypeScript compilation passes
Tests are added for new features
Documentation is updated

📄 License

MIT © Sagente

🙏 Credits

Built with:

MongoDB - Document database
AWS SDK - S3-compatible storage
TypeScript - Type safety
Zod - Schema validation
tsup - Build system

📞 Support

For issues, questions, or feature requests, please open an issue on GitHub.

Made with ❤️ for production-grade data management

chronos-db