durable-cf-streams

bounded SQL storage for durable streams on cloudflare. the package owns persistence, producer fencing, incarnation checks, bounded reads, and fold checkpoints while callers own transport and admission policy.

install

pnpm add durable-cf-streams

storage backends

import {
  SqliteStore,
  createStrictSqliteStore,
} from "durable-cf-streams/storage/sqlite";
import { D1Store, createStrictD1Store } from "durable-cf-streams/storage/d1";

// durable object sqlite with bounded chunk rows
const store = new SqliteStore(state.storage);
store.initialize(); // creates tables

// sqlite for canonical transcripts: require an acquired producer before data writes
const strictStore = createStrictSqliteStore(state.storage);
strictStore.initialize();

// d1 database with bounded chunk rows
const store = new D1Store(env.DB);
await store.initialize(); // creates tables

// d1 for canonical transcripts: require an acquired producer before data writes
const strictStore = createStrictD1Store(env.DB);
await strictStore.initialize();

SqliteStore and D1Store keep bounded append bytes in chunk rows while the authoritative stream row owns identity, incarnation, offsets, and producer state. They use a breaking schema and do not migrate old streams.data rows. Operational limits and invariants are documented beside the store options and interface methods that enforce them. SQLite and D1 fold checkpoints use bounded, surrogate-safe string chunks with independent maxCheckpointStringLength and maxCheckpointChunks ceilings. Both stores reject configurations whose aggregate ceiling exceeds the atomic transaction's row budget.

storage interface

D1Store and SqliteStore implement one strong stream contract:

interface StreamStore {
  readonly producerClaimsRequired: boolean;
  put(path: string, options: PutOptions): Promise<PutResult>;
  append(path: string, data: Uint8Array, options?: AppendOptions): Promise<AppendResult>;
  get(path: string, options?: GetOptions): Promise<GetResult>;
  head(path: string): Promise<HeadResult | null>;
  delete(path: string): Promise<void>;
  has(path: string): boolean | Promise<boolean>;
  waitForData(path: string, offset: string, timeoutMs: number, options?: WaitOptions): Promise<WaitResult>;
  formatResponse(path: string, messages: StreamMessage[]): Uint8Array;
  acquireProducer(path: string, producerId: string): Promise<ProducerClaim>;
  getStreamStorageState(path: string): Promise<StreamStorageState | null>;
  getAppendEndOffset(path: string, appendIndex: number, incarnation?: StreamIncarnation): Promise<Offset | null>;
  readAppendBatches(path: string, startAppendIndex: number, limit: number, incarnation?: StreamIncarnation, maxBytes?: number): Promise<readonly StreamAppendBatch[]>;
  putFoldCheckpoint(path: string, checkpoint: FoldCheckpoint): Promise<void>;
  getFoldCheckpoint(path: string, options?: GetFoldCheckpointOptions): Promise<FoldCheckpoint | null>;
}

protocol constants

compatible with the durable streams protocol:

import {
  // header constants
  STREAM_OFFSET_HEADER,         // "Stream-Next-Offset"
  STREAM_CURSOR_HEADER,         // "Stream-Cursor"
  STREAM_UP_TO_DATE_HEADER,     // "Stream-Up-To-Date"
  STREAM_INCARNATION_HEADER,    // "Stream-Incarnation"
  STREAM_IF_INCARNATION_HEADER, // "Stream-If-Incarnation"
  STREAM_SEQ_HEADER,            // "Stream-Seq"
  STREAM_TTL_HEADER,            // "Stream-TTL"
  STREAM_EXPIRES_AT_HEADER,     // "Stream-Expires-At"
  STREAM_SSE_DATA_ENCODING_HEADER, // "Stream-SSE-Data-Encoding"
  STREAM_CLOSED_HEADER,         // "Stream-Closed"
  STREAM_FORKED_FROM_HEADER,    // "Stream-Forked-From"
  STREAM_FORK_OFFSET_HEADER,    // "Stream-Fork-Offset"
  STREAM_FORK_SUB_OFFSET_HEADER, // "Stream-Fork-Sub-Offset"
  RESERVED_CONTROL_PATH_SEGMENT, // "__ds"
  PRODUCER_ID_HEADER,           // "Producer-Id"
  PRODUCER_EPOCH_HEADER,        // "Producer-Epoch"
  PRODUCER_SEQ_HEADER,          // "Producer-Seq"
  PRODUCER_RETRY_METADATA_HEADER, // "Producer-Retry-Metadata"
  PRODUCER_EXPECTED_SEQ_HEADER, // "Producer-Expected-Seq"
  PRODUCER_RECEIVED_SEQ_HEADER, // "Producer-Received-Seq"
  CACHE_CONTROL_HEADER,     // "Cache-Control"
  CONTENT_TYPE_OPTIONS_HEADER,        // "X-Content-Type-Options"
  CROSS_ORIGIN_RESOURCE_POLICY_HEADER, // "Cross-Origin-Resource-Policy"

  // response header values
  PROTOCOL_SECURITY_HEADERS,
  HEAD_CACHE_CONTROL_VALUE, // "no-store"
  SSE_CACHE_CONTROL_VALUE,  // "no-cache"
  DEFAULT_CONTENT_TYPE,     // "application/octet-stream"

  // query param constants
  OFFSET_QUERY_PARAM,       // "offset"
  TAIL_OFFSET_QUERY_VALUE,  // "now"
  LIVE_QUERY_PARAM,         // "live"
  CURSOR_QUERY_PARAM,       // "cursor"

  // sse
  SSE_OFFSET_FIELD, // "streamNextOffset"
  SSE_CURSOR_FIELD, // "streamCursor"
  SSE_CLOSED_FIELD, // "streamClosed"
  SSE_INCARNATION_FIELD, // "streamIncarnation"
  SSE_COMPATIBLE_CONTENT_TYPES,

  // path encoding
  encodeStreamPath,
  decodeStreamPath,

  // cursor utilities
  calculateCursor,
  generateResponseCursor,
  DEFAULT_CURSOR_EPOCH,
  DEFAULT_CURSOR_INTERVAL_SECONDS,
} from "durable-cf-streams";

branded protocol types

import {
  CursorSchema,
  ETagSchema,
  OffsetSchema,
  ProducerStateMapSchema,
  ProducerStateSchema,
  StreamIncarnationSchema,
  type Cursor,
  type ETag,
  type Offset,
  type ProducerState,
  type ProducerStateMap,
  type StreamIncarnation,
} from "durable-cf-streams";

utilities

import {
  // offsets
  parseOffset,
  formatOffset,
  compareOffsets,
  isValidOffset,
  initialOffset,
  isSentinelOffset,
  normalizeOffset,
  advanceOffset,
  incrementSeq,
  
  // protocol
  normalizeContentType,
  isJsonContentType,
  isSSETextCompatibleContentType,
  validateTTL,
  validateForkSubOffset,
  validateExpiresAt,
  generateETag,
  parseETag,
  processJsonAppend,
  formatJsonResponse,
  formatSSEControl,
  validateJsonCreate,
  encodeSSEData,
  encodeBase64Data,

  // producer idempotency
  parseProducerHeaders,
  evaluateProducerAppend,
  commitProducerAppend,
} from "durable-cf-streams";

errors

tagged errors for pattern matching:

import {
  ContentTypeMismatchError,
  InvalidJsonError,
  InvalidOffsetError,
  InvalidProducerError,
  PayloadTooLargeError,
  ProducerFencedError,
  ProducerSequenceConflictError,
  SequenceConflictError,
  StreamClosedError,
  StreamConflictError,
  type StreamErrorEventData,
  StreamGoneError,
  StreamNotFoundError,
  streamErrorEventData,
  streamErrorEventJson,
  isStreamError,
  streamErrorHeaders,
  streamErrorStatus,
} from "durable-cf-streams";

// check error type
if (error instanceof StreamNotFoundError) {
  return new Response("not found", { status: 404 });
}

// or map any known stream error to its protocol status
if (isStreamError(error)) {
  return new Response(error.message, {
    headers: streamErrorHeaders(error),
    status: streamErrorStatus(error),
  });
}

example

import { SqliteStore } from "durable-cf-streams/storage/sqlite";
import {
  normalizeContentType,
  STREAM_INCARNATION_HEADER,
  STREAM_OFFSET_HEADER,
} from "durable-cf-streams";

export class StreamDO extends DurableObject {
  private store: SqliteStore;

  constructor(state: DurableObjectState, env: Env) {
    super(state, env);
    this.store = new SqliteStore(state.storage);
    this.store.initialize();
  }

  async fetch(request: Request): Promise<Response> {
    const path = new URL(request.url).pathname;

    if (request.method === "PUT") {
      const contentType = request.headers.get("content-type");
      const body = new Uint8Array(await request.arrayBuffer());
      
      const result = await this.store.put(path, {
        contentType: contentType ? normalizeContentType(contentType) : undefined,
        data: body.length > 0 ? body : undefined,
      });

      return new Response(null, {
        status: result.created ? 201 : 200,
        headers: {
          [STREAM_INCARNATION_HEADER]: result.incarnation,
          [STREAM_OFFSET_HEADER]: result.nextOffset,
          "Content-Type": result.contentType,
        },
      });
    }

    // ...
  }
}

see examples for complete implementations.

license

mit