Storage adapter authoring

Streamsy storage packages implement StorageAdapter from @streamsy/core. The interface is flat and serializable: every per-stream method takes streamId as its first argument, while lifecycle plans carry their relevant ids. Adapters may use private per-stream handles internally, but no handle crosses the public storage seam.

This is a storage-author seam, not the public Durable Streams API. Applications wire an adapter into createStreamProtocol({ storage: { adapter } }); protocol-bound streams then expose append, read, readLive, metadata, and delete.

Responsibilities

Core owns protocol policy:

• content-type checks and result mapping;
• message framing, timestamping, offset allocation, and counters;
• sequence and producer-idempotency decisions;
• TTL/expires-at policy and append plans;
• fork payload materialization and fork idempotency checks;
• mapping typed storage unsupported errors to public not-supported results;
• running expiry scheduling and cancellation after durable success.

Adapters own atomic persistence and local runtime capabilities:

• read records, messages, and producer state by stream id;
• atomically apply AppendPlan preconditions and writes;
• create, fork, delete, and reclaim storage records according to an adapter lineage policy;
• implement level-triggered change waiting;
• schedule and cancel active expiry work.

Do not reimplement protocol policy in adapters. Apply the plans core gives you.

Interfaces to implement

import type { StorageAdapter } from "@streamsy/core";

export interface StorageAdapter
  extends StreamReader, StreamAppender, StreamLiveWaiter, StreamExpiryScheduler {
  create(plan: CreatePlan): Promise<StorageCreateResult>;
  fork?(plan: ForkPlan): Promise<StorageForkResult>;
  delete(plan: DeletePlan): Promise<StorageDeleteResult>;
}

Private record/message/producer stores are fine internally, but the public storage contract is the flat StorageAdapter.

append(streamId, AppendPlan) algorithm

append must be atomic for one stream. Use a SQL transaction/CAS, an actor or Durable Object turn, or a synchronous in-memory critical section.

Implement it as:

1. Open the backend's atomic boundary.
2. Re-read the current stream record inside that boundary.
3. If no record exists, return precondition-failed with record: null and an appropriate reason.
4. Enforce expectedOffset and expectedClosed inside the boundary.
5. If plan.preconditions.producer is present, compare the current producer state to expected. An absent expected means the producer must not exist.
6. Insert plan.messages exactly as supplied.
7. Apply the required recordPatch, merging partial config and lifecycle objects.
8. Return { status: "appended", record: freshRecord } after durable success.
9. On a failed precondition, write nothing and return { status: "precondition-failed", record, reason }. Attribute simultaneous failures in offset, closed, producer order.

Message inserts, record patches, and producer-state changes must commit or abort together. Never allocate offsets, check content type, validate sequence headers, or derive protocol responses in append.

Adapter lifecycle methods

create(plan)

Persist plan.record exactly as supplied, together with plan.initialMessages, in one atomic operation. The record already contains the complete lifecycle state, including a created-closed stream. Return created or exists, carrying the persisted record in either case.

fork?(plan)

fork is optional. If supported, the adapter should:

1. Re-check that plan.sourceId exists, is not soft-deleted, and is live at plan.precondition.sourceLiveAtOffset.
2. Create plan.child and plan.initialMessages atomically for the child.
3. Record any adapter-private parent-child lineage edge.
4. Return created, exists, or fork-source-gone.

Core builds the child record and materialized prefix. Cross-stream edge registration should be idempotent and repairable.

delete(plan)

plan.streamId identifies the stream and plan.reason is delete or expiry. Return:

• not-found when no record exists;
• gone when a user delete targets an already soft-deleted stream;
• purged when storage was physically removed;
• retained-soft-deleted when the record must remain because forks depend on it.

Deletion may purge messages, producer state, records, timers, and adapter-private lineage.

Lineage strategy choices

Core exports helpers for common reclaim policies:

The strategies are adapter-composed; core does not mandate one storage layout.

Change waiting and expiry

awaitChange is required. It receives a serializable baseline snapshot and must return only when the stream differs, disappears, or the deadline is reached. A backend without native notifications can poll its own durable reads using the exported runAwaitChangeLoop, buildChangeSnapshot, and changeSnapshotDiffers helpers.

scheduleExpiry and cancelExpiry are also required. Core invokes them after a successful durable mutation; they do not belong inside append or create transactions.

Unsupported features

fork may be omitted. For other unsupported storage behavior, throw the typed storage error:

import { unsupported } from "@streamsy/core";

throw unsupported("feature", "This adapter does not support the requested feature");

The protocol maps NotSupportedError to a structured public not-supported result. Avoid silent no-ops.

Backend sketches

SQL-like stores

• Use tables for streams, messages, and producer states.
• Wrap append in an immediate/write transaction.
• Use conditional writes for offset, closed-state, and producer preconditions.
• Insert messages in the same transaction as record and producer updates.
• Use a reverse index or child-edge rows for fork reclamation.

Reference: packages/storage-sqlite/src.

Actor or Durable Object stores

• Route one public stream id to one actor/object instance internally.
• Apply append in one actor turn using the actor's storage API.
• Keep the exported adapter flat even if it delegates to private per-stream RPC objects.
• Model fork/delete/GC as convergent, idempotent operations.
• Use actor alarms for active expiry and actor-local waiters for live reads.

Reference: packages/storage-durable-object/src.

Memory and test stores

• Keep a process-wide state registry for streams.
• Apply the same precondition rules as durable backends.
• Provide notifier and timeout scheduler implementations for live reads and expiry.
• Use memory as a conformance baseline, not as a weaker protocol.

Reference: packages/core/src/storage/memory.

Verification checklist

For an adapter:

1. Typecheck the package and public examples.
2. Test create, append, close, offset CAS, producer idempotency, reads, live reads, expiry, delete, and supported forks.
3. Run the exported runStorageAdapterContract kit.
4. Run the shared protocol conformance harness for the backend where possible.
5. Test concurrent append and producer-state races.
6. Test crash/retry/idempotency paths around lifecycle operations.

Useful repo commands:

bun run format:check
bun run typecheck
bun run test:unit
bun run test:conformance:memory
bun run test:conformance:sqlite
bun run test:conformance:do

Run the checks that match the adapter and document any skipped ones.

Common pitfalls

• Checking preconditions before the atomic boundary instead of inside it.
• Treating an absent producer expected value as "do not check" rather than "must be absent".
• Updating the record but forgetting producer CAS, or vice versa.
• Inserting messages outside the transaction that advances the record.
• Recomputing offsets, timestamps, framing, content type, sequence policy, or public results in the adapter.
• Returning per-stream handles across the flat adapter seam.
• Treating notification as the only correctness path instead of implementing level-triggered awaitChange.
• Implementing fork edge registration as a non-idempotent cross-stream operation.