Write

AppendCopy Anchor Link

import { Effect } from "effect";
import { Notes } from "./streams.ts";

const write = Effect.gen(function* () {
  const notes = yield* Notes;
  const result = yield* notes.append({}, [{ id: "n1", text: "hello" }]);
  return result.offset;
});

append(params, items, options?) encodes the items and writes them as one atomic append.

A JSON stream keeps one message for each item. A bytes stream joins the items of one append into one message.

Durable State changesCopy Anchor Link

DurableState.changes makes the change events for one address of a Streamsy.state stream.

import { Effect } from "effect";
import { DurableState } from "@streamsy/core";
import { Workspace } from "./streams.ts";

const createIssue = (offset: string, issue: { id: string; projectId: string; title: string }) =>
  Effect.gen(function* () {
    const workspace = yield* Workspace;
    const params = { workspaceId: "main" };
    const facts = DurableState.changes(Workspace.at(params), { offset }, [
      DurableState.upsert("issue", issue),
    ]);
    yield* workspace.append(params, facts);
  });

DurableState.remove("issue", id) deletes by key. Add { oldValue } to also write old_value. If you give no offset, the events have no offset and no default txid. An upsert or delete can also set timestamp or from. A change to an undeclared collection, or a value that does not match its schema, does not compile.

Control messagesCopy Anchor Link

DurableState.changes also takes the State control messages. Make them with StreamAddress.reset(), StreamAddress.snapshotStart() and StreamAddress.snapshotEnd(). Each takes an optional { offset }.

import { Effect } from "effect";
import { DurableState, StreamAddress } from "@streamsy/core";
import { Workspace } from "./streams.ts";

const replaceIssues = (issues: ReadonlyArray<{ id: string; projectId: string; title: string }>) =>
  Effect.gen(function* () {
    const workspace = yield* Workspace;
    const params = { workspaceId: "main" };
    const facts = DurableState.changes(Workspace.at(params), {}, [
      StreamAddress.reset(),
      ...issues.map((issue) => DurableState.upsert("issue", issue)),
    ]);
    yield* workspace.append(params, facts);
  });

A control passes through changes unchanged. With an offset, a control that has no offset of its own gets that offset. Controls do not take a txid, but they count in the index of the default txid of the changes after them.

CloseCopy Anchor Link

import { Effect } from "effect";
import { Notes } from "./streams.ts";

const close = Effect.flatMap(Notes, (notes) => notes.append({}, [], { close: true }));

Later appends fail with StreamClosed.

Producer retriesCopy Anchor Link

A producer position is { producerId, producerEpoch, producerSeq }. Producer.next advances it.

import { Effect } from "effect";
import { Producer } from "@streamsy/core";
import { Notes } from "./streams.ts";

const importOne = (position: Producer.Position) =>
  Effect.gen(function* () {
    const notes = yield* Notes;
    const result = yield* notes.append({}, [{ id: "n3", text: "retried" }], {
      producer: position,
    });
    // Appended and Duplicate both acknowledge this position.
    return { result, next: Producer.next(position) };
  });

Advance the position only after Appended or Duplicate. Send the same bytes on a retry. The server compares positions, not payloads.

producerSeq starts at 0 for each new epoch. For restarts, see the core producer restart contract and PROTOCOL.md section 5.2.1.

Conditional writesCopy Anchor Link

expectedOffset writes only if the tail still equals it. ZERO_OFFSET means the stream must be empty. A miss writes nothing and fails with OffsetMismatch, with expected and actual.

import { Effect, Predicate } from "effect";
import { Orders, apply, initial } from "./orders.ts";

const reserve = (sku: string) =>
  Effect.gen(function* () {
    const orders = yield* Orders;
    const { state, offset } = yield* orders.fold({}, () => initial, apply);
    if (state.reserved.has(sku)) return { _tag: "Rejected" as const };
    const appended = yield* orders.append({}, [{ type: "reserved", sku }], {
      expectedOffset: offset,
    });
    return { _tag: "Committed" as const, offset: appended.offset };
  }).pipe(Effect.retry({ while: Predicate.isTagged("OffsetMismatch"), times: 5 }));

fold returns the folded state and the tail it read up to. That tail is the expectedOffset. A competing writer moves the tail, so the append fails and the retry reads again. To give one writer ownership of a stream, use a producer epoch.

The append's AppendFailure<O> type follows its options. Only an append with expectedOffset can fail with OffsetMismatch, and only an append with producer can fail with StaleEpoch or ProducerGap. An append needs at least one item, unless it sets close: true.

expectedOffset is a Streamsy extension to the protocol. See Extensions.

Read, fold, appendCopy Anchor Link

decide does the loop above in one call. It folds the stream, calls decide with the state and the tail it read, and appends with that tail as expectedOffset:

import { Effect } from "effect";
import { Orders, apply, initial } from "./orders.ts";

const reserve = (sku: string) =>
  Effect.gen(function* () {
    const orders = yield* Orders;
    return yield* orders.decide(
      {},
      {
        initial: () => initial,
        fold: apply,
        decide: (state) =>
          Effect.succeed(
            state.reserved.has(sku)
              ? { value: "rejected" as const, items: [] }
              : { value: "reserved" as const, items: [{ type: "reserved" as const, sku }] },
          ),
      },
    );
  });

What decide does:

• It folds the stream from the start with initial and fold.
• It calls decide with the state and the tail it read. decide must not write and must not change the state.
• Empty items returns at once. There is no append and no tail check.
• Otherwise it appends the items with expectedOffset set to that tail.
• When its own append fails with OffsetMismatch, it folds only the new items onto the same state and decides again. retry sets the schedule. The default is Schedule.recurs(3). When the schedule ends, it fails with the last OffsetMismatch.
• It does not retry other failures. An OffsetMismatch that decide raises is not retried. A lost reply can hide an append that landed, so a caller that retries the whole call must make decide safe against its own write.

It returns { value, offset, appended }. offset is the new tail after a write, or the tail that was read. Streams.decide(address, spec) is the same call on an address.

On a Durable State stream, pass the offset to DurableState.changes, so each change records the offset that the append checks. The issue tracker does this in IssueTracker.mutate.

To project a stream into state and then append, see Read, decide, append.

ErrorsCopy Anchor Link

AppendFailure<O> includes protocol rejections, StreamsFault, and EncodeFault. create and fork use CreateFailure; remove uses RemoveFailure. Import these types from @streamsy/core.

Caller bugs are defects in toolkit append: InvalidAppendRequest, InvalidEpochSeq, and rejections ruled out by its options. An empty encoded bytes item dies with a RangeError before the write. The protocol writer and HTTP edge keep all append rejections typed.

Over HTTPCopy Anchor Link

POST on the stream path appends. It needs append or full access.

The remote Layer sends these headers only when its capabilities allow them. See Remote client.