@delta-base/do-document-store

A MongoDB-like Document Store built on Cloudflare Durable Objects with SQLite storage.

Features

• MongoDB-like API - Familiar insertOne, find, updateOne, deleteOne operations
• IReadModelStore Support - Adapter for event sourcing projections with testable interface
• Type-safe - Full TypeScript support with generics for document types
• SQLite-backed - Leverages Cloudflare's SQLite storage in Durable Objects
• Deploy in your account - Export the DO class and deploy to your own Cloudflare account
• Optimistic concurrency - Built-in _version field for conflict detection
• Soft deletes - Optional soft delete support with _archived flag
• RPC-friendly - Flat methods for direct DO calls over Cloudflare RPC

Installation

npm install @delta-base/do-document-store
# or
pnpm add @delta-base/do-document-store

Quick Start

1. Configure your wrangler.jsonc

{
  "name": "my-worker",
  "main": "src/index.ts",
  "compatibility_date": "2024-12-01",
  "durable_objects": {
    "bindings": [
      {
        "name": "DOCUMENT_STORE",
        "class_name": "DocumentStoreDurableObject"
      }
    ]
  },
  "migrations": [
    {
      "tag": "v1",
      "new_sqlite_classes": ["DocumentStoreDurableObject"]
    }
  ]
}

2. Export the Durable Object class

// src/index.ts
export { DocumentStoreDurableObject } from '@delta-base/do-document-store';

interface Env {
  DOCUMENT_STORE: DurableObjectNamespace<DocumentStoreDurableObject>;
}

export default {
  async fetch(request: Request, env: Env) {
    const id = env.DOCUMENT_STORE.idFromName('my-store');
    const store = env.DOCUMENT_STORE.get(id);
    
    // Use the document store
    const users = store.collection<User>('users');
    // ...
  }
};

3. Define your document types

interface User {
  name: string;
  email: string;
  age: number;
  tags?: string[];
}

4. Use the MongoDB-like API

const users = store.collection<User>('users');

// Insert
const { insertedId } = users.insertOne({
  name: 'Alice',
  email: '[email protected]',
  age: 30,
  tags: ['developer']
});

// Find
const alice = users.findOne({ email: '[email protected]' });
const adults = users.find({ age: { $gte: 18 } });

// Update
users.updateOne(
  { _id: insertedId },
  { $set: { age: 31 }, $push: { tags: 'senior' } }
);

// Delete
users.deleteOne({ _id: insertedId });

API Reference

DocumentStoreDurableObject

The main Durable Object class that provides document store functionality.

Methods

| Method | Description | |--------|-------------| | collection<T>(name) | Get a typed collection | | createCollection(name) | Create a collection (called automatically) | | dropCollection(name) | Drop a collection and all its data | | listCollections() | List all collection names | | hasCollection(name) | Check if a collection exists |

Collection

A typed collection for document operations.

Insert Operations

// Insert one document
const result = users.insertOne({ name: 'Alice', email: '[email protected]', age: 30 });
// Returns: { acknowledged: true, insertedId: 'uuid-...' }

// Insert many documents
const result = users.insertMany([
  { name: 'Alice', email: '[email protected]', age: 30 },
  { name: 'Bob', email: '[email protected]', age: 25 }
]);
// Returns: { acknowledged: true, insertedCount: 2, insertedIds: ['...', '...'] }

Find Operations

// Find one document
const user = users.findOne({ email: '[email protected]' });

// Find many documents
const adults = users.find({ age: { $gte: 18 } });

// Find with options
const page = users.find(
  { status: 'active' },
  { limit: 10, offset: 20, sort: { createdAt: -1 } }
);

// Count documents
const count = users.countDocuments({ age: { $gt: 25 } });

Update Operations

// Update one document
users.updateOne(
  { _id: 'some-id' },
  { $set: { name: 'New Name' }, $inc: { loginCount: 1 } }
);

// Update many documents
users.updateMany(
  { status: 'pending' },
  { $set: { status: 'active' } }
);

// Replace entire document
users.replaceOne(
  { _id: 'some-id' },
  { name: 'Alice', email: '[email protected]', age: 31 }
);

// Upsert (insert if not exists)
users.updateOne(
  { email: '[email protected]' },
  { $set: { name: 'New User' } },
  { upsert: true }
);

Delete Operations

// Hard delete
users.deleteOne({ _id: 'some-id' });

// Soft delete (sets _archived = 1)
users.deleteOne({ _id: 'some-id' }, { softDelete: true });

// Delete many
users.deleteMany({ status: 'inactive' });

Collection Operations

// Rename collection
users.rename('members');

// Drop collection
users.drop();

IReadModelStore Interface

The package provides DOReadModelStore, an adapter that implements the IReadModelStore interface from @delta-base/toolkit. This enables:

• Testability: Projections can depend on IReadModelStore and use InMemoryReadModelStore for unit tests
• Flexibility: Same projection code works with different storage backends
• Event Sourcing: Clean integration with the command → projection → query lifecycle

Basic Usage

import { DOReadModelStore } from '@delta-base/do-document-store';

// Get the DO stub
const doId = env.DOCUMENT_STORE.idFromName('my-store');
const stub = env.DOCUMENT_STORE.get(doId);

// Create the adapter
const store = new DOReadModelStore(stub);

// Use as IReadModelStore
await store.put('user:123', { name: 'Alice', email: '[email protected]' });
const user = await store.get('user:123');
await store.delete('user:123');

Testable Projections

import type { IReadModelStore } from '@delta-base/toolkit';

// Projection depends on interface, not concrete implementation
class UserProjection {
  constructor(private store: IReadModelStore) {}

  async apply(event: UserCreatedEvent) {
    await this.store.put(`user:${event.userId}`, {
      id: event.userId,
      name: event.name,
      email: event.email,
    });
  }
}

// Unit test with InMemory store
import { InMemoryReadModelStore } from '@delta-base/toolkit';

const store = new InMemoryReadModelStore();
const projection = new UserProjection(store);
await projection.apply(mockEvent);
expect(await store.get('user:123')).toEqual({ ... });

// Production with DO-backed store
const stub = env.DOCUMENT_STORE.get(doId);
const store = new DOReadModelStore(stub);
const projection = new UserProjection(store);

Multiple Collections (Tables)

Use the tableName option to store data in different collections:

const store = new DOReadModelStore(stub);

// Store in different collections
await store.put('user:123', userData, { tableName: 'users' });
await store.put('order:456', orderData, { tableName: 'orders' });

// Retrieve from specific collection
const user = await store.get('user:123', { tableName: 'users' });

Rich Queries (Escape Hatch)

For complex queries that need the full MongoDB-like API, use getDocumentStore():

const store = new DOReadModelStore(stub);

// Simple operations via IReadModelStore
await store.put('user:123', userData);
const user = await store.get('user:123');

// Rich queries via DocumentStore escape hatch
const docStore = store.getDocumentStore();
const activeUsers = await docStore.find('users', 
  { status: 'active', age: { $gte: 18 } },
  { sort: { createdAt: -1 }, limit: 10 }
);

IReadModelStore Methods

| Method | Description | |--------|-------------| | put(key, value, options?) | Store a value by key (upserts) | | get(key, options?) | Retrieve a value by key | | delete(key, options?) | Delete a value by key | | getAll(options?) | Get all values, optionally filtered by prefix | | listKeys(options?) | List keys with pagination support | | batchGet(keys, options?) | Retrieve multiple values by keys | | batchPut(items, options?) | Store multiple key-value pairs | | batchDelete(keys, options?) | Delete multiple keys | | getCapabilities() | Get store capabilities | | getDocumentStore() | Access underlying DO stub for rich queries |

Middleware Integration

import { DOReadModelStore } from '@delta-base/do-document-store';
import { createMiddleware } from 'hono/factory';

export const readModelStoreMiddleware = createMiddleware(async (c, next) => {
  const orgId = c.get('orgId'); // From auth middleware
  
  const doId = c.env.DOCUMENT_STORE.idFromName(orgId);
  const stub = c.env.DOCUMENT_STORE.get(doId);
  
  c.set('readModelStore', new DOReadModelStore(stub));
  await next();
});

// Usage in routes
app.get('/users/:id', async (c) => {
  const store = c.get('readModelStore');
  const user = await store.get(`user:${c.req.param('id')}`);
  return c.json(user);
});

RPC-Friendly Methods

The DocumentStoreDurableObject also exposes flat RPC-friendly methods for direct access without going through the collection() method. These are useful when calling the DO over RPC from a Worker:

const stub = env.DOCUMENT_STORE.get(doId);

// Direct RPC calls (instead of stub.collection('users').findOne(...))
const user = await stub.findOne('users', { _id: 'user:123' });
const users = await stub.find('users', { status: 'active' }, { limit: 10 });
await stub.insertOne('users', { name: 'Alice', age: 30 });
await stub.updateOne('users', { _id: 'user:123' }, { $set: { age: 31 } });
await stub.deleteOne('users', { _id: 'user:123' });

| Method | Description | |--------|-------------| | findOne(collection, filter?) | Find a single document | | find(collection, filter?, options?) | Find multiple documents | | insertOne(collection, doc, options?) | Insert a document | | insertMany(collection, docs, options?) | Insert multiple documents | | updateOne(collection, filter, update, options?) | Update a single document | | updateMany(collection, filter, update, options?) | Update multiple documents | | replaceOne(collection, filter, doc, options?) | Replace a document | | deleteOne(collection, filter, options?) | Delete a single document | | deleteMany(collection, filter, options?) | Delete multiple documents | | countDocuments(collection, filter?) | Count matching documents |

Filter Operators

Supported MongoDB-style filter operators:

| Operator | Description | Example | |----------|-------------|---------| | $eq | Equal | { age: { $eq: 30 } } | | $ne | Not equal | { status: { $ne: 'deleted' } } | | $gt | Greater than | { age: { $gt: 18 } } | | $gte | Greater than or equal | { age: { $gte: 21 } } | | $lt | Less than | { age: { $lt: 65 } } | | $lte | Less than or equal | { age: { $lte: 100 } } | | $in | In array | { status: { $in: ['active', 'pending'] } } | | $nin | Not in array | { role: { $nin: ['admin'] } } | | $exists | Field exists | { email: { $exists: true } } | | $and | Logical AND | { $and: [{ age: { $gte: 18 } }, { age: { $lte: 65 } }] } | | $or | Logical OR | { $or: [{ status: 'active' }, { role: 'admin' }] } |

Nested Fields

Use dot notation for nested fields:

users.find({ 'address.city': 'New York' });
users.updateOne({ _id: 'id' }, { $set: { 'profile.bio': 'Hello' } });

Update Operators

Supported MongoDB-style update operators:

| Operator | Description | Example | |----------|-------------|---------| | $set | Set field value | { $set: { name: 'Bob' } } | | $unset | Remove field | { $unset: { oldField: '' } } | | $inc | Increment number | { $inc: { count: 1 } } | | $push | Add to array | { $push: { tags: 'new-tag' } } | | $pull | Remove from array | { $pull: { tags: 'old-tag' } } |

System Fields

Every document includes these system fields:

| Field | Type | Description | |-------|------|-------------| | _id | string | Primary key (auto-generated UUID if not provided) | | _version | number | Incremented on each update (starts at 1) | | _created | string | ISO timestamp of creation | | _updated | string | ISO timestamp of last update |

Optimistic Concurrency Control

The document store supports optimistic concurrency via the expectedVersion option. This allows you to prevent lost updates when multiple clients modify the same document.

Expected Version Values

| Value | Description | |-------|-------------| | number | Exact version the document must have | | 'DOCUMENT_DOES_NOT_EXIST' | Document must not exist (for inserts) | | 'DOCUMENT_EXISTS' | Document must exist (for updates/deletes, any version) | | 'NO_CONCURRENCY_CHECK' | Skip version checking (default) |

Insert with Concurrency Check

// Ensure document doesn't already exist
users.insertOne(
  { _id: 'user-123', name: 'Alice', email: '[email protected]', age: 30 },
  { expectedVersion: 'DOCUMENT_DOES_NOT_EXIST' }
);

// Throws VersionMismatchError if document with _id 'user-123' already exists

Update with Concurrency Check

// Read document
const user = users.findOne({ _id: 'user-123' });
// user._version === 1

// Update only if version matches
users.updateOne(
  { _id: 'user-123' },
  { $set: { age: 31 } },
  { expectedVersion: 1 }
);

// Throws VersionMismatchError if another client modified the document

Replace with Concurrency Check

users.replaceOne(
  { _id: 'user-123' },
  { name: 'Alice Updated', email: '[email protected]', age: 31 },
  { expectedVersion: 1 }
);

Delete with Concurrency Check

users.deleteOne(
  { _id: 'user-123' },
  { expectedVersion: 2 }
);

// Works with soft delete too
users.deleteOne(
  { _id: 'user-123' },
  { softDelete: true, expectedVersion: 2 }
);

Ensure Document Exists (Any Version)

Use 'DOCUMENT_EXISTS' when you want to ensure a document exists before updating or deleting, but don't care about the specific version:

// Update only if document exists (fails if not found)
users.updateOne(
  { _id: 'user-123' },
  { $set: { lastSeen: new Date().toISOString() } },
  { expectedVersion: 'DOCUMENT_EXISTS' }
);

// Delete only if document exists (fails if not found)
users.deleteOne(
  { _id: 'user-123' },
  { expectedVersion: 'DOCUMENT_EXISTS' }
);

// Throws VersionMismatchError if document doesn't exist

This is useful when you want to distinguish between "no document matched the filter" and "document was successfully modified" - without DOCUMENT_EXISTS, these operations silently return with matchedCount: 0 or deletedCount: 0.

Handling Version Mismatch Errors

import {
  VersionMismatchError,
  isVersionMismatchError
} from '@delta-base/do-document-store';

try {
  users.updateOne(
    { _id: 'user-123' },
    { $set: { age: 31 } },
    { expectedVersion: 1 }
  );
} catch (error) {
  if (isVersionMismatchError(error)) {
    console.log('Conflict detected!');
    console.log('Expected version:', error.expectedVersion);
    console.log('Actual version:', error.actualVersion);
    console.log('Document ID:', error.documentId);
    // Handle conflict: reload document and retry, or notify user
  }
}

Bulk Operations with Expected Version

For updateMany and deleteMany, the expectedVersion acts as an additional filter. Only documents matching both the filter AND the version will be affected:

// Only updates documents at version 1
const result = users.updateMany(
  { status: 'pending' },
  { $set: { status: 'active' } },
  { expectedVersion: 1 }
);

// result.matchedCount shows how many matched both filter AND version

If NO documents match (but some exist at different versions), a VersionMismatchError is thrown.

Schema

Each collection is stored as a SQLite table:

CREATE TABLE collection_name (
  _id       TEXT    PRIMARY KEY,
  data      JSON    NOT NULL,
  metadata  JSON    NOT NULL DEFAULT '{}',
  _version  INTEGER NOT NULL DEFAULT 1,
  _archived INTEGER NOT NULL DEFAULT 0,
  _created  TEXT    NOT NULL DEFAULT (datetime('now')),
  _updated  TEXT    NOT NULL DEFAULT (datetime('now'))
);

Error Handling

import {
  DuplicateKeyError,
  DocumentNotFoundError,
  InvalidFilterError,
  isDuplicateKeyError
} from '@delta-base/do-document-store';

try {
  users.insertOne({ _id: 'existing-id', ... });
} catch (error) {
  if (isDuplicateKeyError(error)) {
    console.log('Document already exists:', error.documentId);
  }
}

Testing

Tests use the @cloudflare/vitest-pool-workers package:

pnpm test

License

See LICENSE file in the repository root.