EventLog unstable

Runtime for writing typed events to an event journal.

EventLog combines event groups, handlers, a journal, local identity, optional remote replicas, and reactivity hooks. Writers send typed payloads through a client; the matching handler runs first, and the journal entry is committed only after the handler succeeds. This module also contains the layers and helpers needed to assemble that runtime.

Remote compaction

An encrypted remote whose server sets HelloResponse.compaction can store a compacted snapshot of its history, so a new device replays the snapshot and a recent tail instead of every write. Clients build the snapshots because only they can decrypt the entries.

Only a client with a remote: true compactor, on a remote not built with compaction: false, calls the compaction RPCs. It polls the status on connect and about hourly. When the server wants compaction and the journal has caught up, it starts a background run after a random delay of a few minutes. The run reads the input, taking entries from the journal where it can and fetching the rest, runs the opted-in compactors and uploads the entries they keep. The server lets one run per store hold a stage and answers the others Busy, so devices that start together download the input once. EventLog.compact starts a run immediately, for example from a worker holding the identity.

An operator can roll back a bad snapshot on the server. The server then re-appends the snapshot's input at the head, so every client, including ones from before remote compaction, receives the dropped entries through changes and skips the ones it already holds.

Known limitations:

A run holds every kept entry in memory. A device with a large store can opt out with compaction: false and leave compaction to a device or worker with more memory.
The library can't check that a compactor keeps the remote contract (see groupCompaction). The drop-only and idempotence checks catch common mistakes. A snapshot that drops too much can only be undone by a rollback before the server prunes the history.
Changes to a field's meaning between schema versions aren't detected. Strict decoding only catches unknown fields.
A new device still replays the newest day or so of writes, because the server only compacts data older than its lag.
A client that stops partway through a run holds the server's busy claim until its stage expires.

27 exports Since v4.0.0 Source

Compaction

groupBatchCompaction

unstable

Since v4.0.0

Registers a batch compaction handler for an event group.

When to use

Use when an event under one primary key affects entries under other keys.

Details

The compaction effect receives every entry for the group in batch order, with its decoded event or event: undefined if the payload cannot be decoded.

Only entries that the effect outputs are kept:

forward(entry) keeps an input entry unchanged, with the same id, payload, and primary key.
write(tag, payload) encodes a replacement entry anchored to the first entry in the batch, with a fresh id in that entry's millisecond.

Entries that are neither forwarded nor replaced are dropped, including entries that could not be decoded. During remote replay, kept entries are replayed in the id order of their anchors, then in write order, and a replacement's conflicts are the stored entries newer than its anchor.

Gotchas

Compaction runs before conflicts are read, so it never sees stored entries. Collapsing several entries into one is only safe when no stored entry for the same event and primary key has an id between theirs. Otherwise that stored entry should have applied between the collapsed ones, and replicas that received the entries in separate batches end up different.

See

groupCompaction for compaction grouped by primary key

Signature

declare function groupBatchCompaction<Events extends Any, R>(group: EventGroup<Events>, effect: (options: {
  readonly entries: readonly Array<{
    readonly entry: Entry;
    readonly event: TaggedPayload<Events> | undefined;
  }>;
  readonly forward: (entry: Entry) => Effect<void>;
  readonly write: <Tag extends string>(tag: Tag, payload: Type<PayloadSchema<Extract<Events, {
    readonly tag: Tag;
  }>>>) => Effect<void, never, PayloadSchemaWithTag<Events, Tag>["EncodingServices"]>;
}) => Effect<void, never, R>, options?: {
  readonly remote?: boolean;
}): Layer<never, never, Registry | R | PayloadSchema<Events>["DecodingServices"]>

groupCompaction

unstable

Since v4.0.0

Registers a compaction handler for an event group.

Details

During remote replay, decoded entries are grouped by primary key and passed to the compaction effect, which outputs the entries to keep:

forward(entry) keeps an input entry unchanged, with the same id, payload, and primary key.
write(tag, payload) encodes a replacement entry anchored to the first decoded entry for the primary key, with a fresh id in that entry's millisecond.

Entries that are neither forwarded nor replaced are dropped. Entries that cannot be decoded are not passed to the effect and are kept unchanged. During remote replay, kept entries are replayed in the id order of their anchors, then in write order, and a replacement's conflicts are the stored entries newer than its anchor.

Gotchas

See

groupBatchCompaction for compaction across primary keys

Signature

declare function groupCompaction<Events extends Any, R>(group: EventGroup<Events>, effect: (options: {
  readonly entries: readonly Array<Entry>;
  readonly events: readonly Array<TaggedPayload<Events>>;
  readonly forward: (entry: Entry) => Effect<void>;
  readonly primaryKey: string;
  readonly write: <Tag extends string>(tag: Tag, payload: Type<PayloadSchema<Extract<Events, {
    readonly tag: Tag;
  }>>>) => Effect<void, never, PayloadSchemaWithTag<Events, Tag>["EncodingServices"]>;
}) => Effect<void, never, R>, options?: {
  readonly remote?: boolean;
}): Layer<never, never, Registry | R | PayloadSchema<Events>["DecodingServices"]>

Constructors

decodeIdentityString

unstable

Since v4.0.0

Decodes a base64url identity string produced by encodeIdentityString.

Gotchas

Invalid input throws a schema decoding error.

Signature

declare function decodeIdentityString(value: string): {
  readonly privateKey: Redacted<Uint8Array<ArrayBuffer>>;
  readonly publicKey: string;
}

encodeIdentityString

unstable

Since v4.0.0

Encodes an event-log identity as a base64url string containing the public key and private key bytes.

Signature

declare function encodeIdentityString(identity: {
  readonly privateKey: Redacted<Uint8Array<ArrayBuffer>>;
  readonly publicKey: string;
}): string

makeClient

unstable

Since v4.0.0

Creates a typed client function for writing events defined by an EventLogSchema.

Details

The returned function takes one { event, payload } object, typed so the payload matches the event, as writeMany items are. It delegates to the EventLog service and preserves each event's success and error types.

Signature

declare function makeClient<Groups extends Any>(schema: EventLogSchema<Groups>): Effect<<Item extends {
  readonly event: Tag;
  readonly payload: Type<PayloadSchema<Extract<Events<Groups>, {
    readonly tag: Tag;
  }>>>;
}>(item: Item) => Effect<Type<SuccessSchema<Extract<Events<Groups>, {
  readonly tag: Item["event"];
}>>>, EventJournalError | Type<ErrorSchema<Extract<Events<Groups>, {
  readonly tag: Item["event"];
}>>>>, never, EventLog>

Example Writing events with a client

import { Effect, Schema } from "effect"
import { EventGroup, EventLog } from "effect/eventlog"

const Users = EventGroup.empty.add({
  tag: "UserCreated",
  primaryKey: (payload) => payload.id,
  payload: Schema.Struct({ id: Schema.String })
})

export const createUser = Effect.gen(function*() {
  const client = yield* EventLog.makeClient(EventLog.schema(Users))
  yield* client({ event: "UserCreated", payload: { id: "user-1" } })
})

makeIdentity

unstable

Since v4.0.0

Generates a new event-log identity using the configured EventLogEncryption service.

Signature

declare const makeIdentity: Effect.Effect<Identity["Service"], never, EventLogEncryption.EventLogEncryption>

Guards

isEventLogSchema

unstable

Since v4.0.0

Returns true when a value carries the EventLogSchema marker.

Signature

declare function isEventLogSchema(u: unknown): u is EventLogSchema<Any>

Handlers

group

unstable

Since v4.0.0

Creates a layer that registers handlers for every event in an event group.

Details

The callback receives a Handlers builder; its return type is checked so every event in the group is handled.

Signature

declare function group<Events extends Any, Return>(group: EventGroup<Events>, f: (handlers: Handlers<never, Events>) => ValidateReturn<Return>): Layer<ToService<Events>, Error<Return>, Registry | Exclude<Services<Return>, Scope | Identity>>

Handlers

interface unstable

Since v4.0.0

Builder for the handlers associated with an EventGroup.

Details

The Events type parameter tracks the event tags that still need handlers, and each call to handle records a handler while accumulating any required services.

Signature

interface Handlers<R, Events extends Event.Any = never> extends Pipeable {
  readonly "~effect/eventlog/EventLog/Handlers": {
    _Events: Covariant<Events>;
  };
  readonly context: Context<R>;
  readonly group: AnyWithProps;
  readonly handlers: Record.ReadonlyRecord<string, Handlers.Item<R>>;
  handle<Tag extends string, R1>(name: Tag, handler: (options: {
    readonly conflicts: readonly Array<{
      readonly entry: Entry;
      readonly payload: Type<PayloadSchema<Extract<Events, {
        readonly tag: Tag;
      }>>>;
    }>;
    readonly entry: Entry;
    readonly payload: Type<PayloadSchema<Extract<Events, {
      readonly tag: Tag;
    }>>>;
    readonly storeId: StoreId;
  }) => Effect<Type<SuccessSchema<Extract<Events, {
    readonly tag: Tag;
  }>>>, Type<ErrorSchema<Extract<Events, {
    readonly tag: Tag;
  }>>>, R1>): Handlers<R | R1, Exclude<Events, {
    readonly tag: Tag;
  }>>;
}

makeReplayFromRemote

unstable

Since v4.0.0

Builds the effect used to replay entries received from a remote event log.

Details

The returned handler decodes the entry and conflicts with the registered event schema, runs the matching handler with the supplied identity and store id, and invalidates configured reactivity keys. Decode and handler failures are logged and fail the effect. Invalidations are deferred until the end of an enclosing Reactivity.withBatch.

Signature

declare function makeReplayFromRemote(options: {
  readonly handlers: ReadonlyMap<string, Item<any>>;
  readonly identity: {
    readonly privateKey: Redacted<Uint8Array<ArrayBuffer>>;
    readonly publicKey: string;
  };
  readonly logAnnotations: {
    readonly effect: string;
    readonly service: string;
  };
  readonly reactivity: Reactivity;
  readonly reactivityKeys: Record<string, ReadonlyArray<string>>;
  readonly storeId: StoreId;
}): (input: ReplayInput) => Effect<void, unknown, never>

Layers

layer

unstable

Since v4.0.0

Combines event-group handler layers with the EventLog runtime for a schema.

When to use

Use when you need one layer that installs the shared EventLog runtime for an EventLogSchema and registers an event-group handler layer for typed writes.

Details

The supplied handler layer is provided with layerEventLog. The returned layer provides EventLog | Registry, preserves the handler layer's error type, and still requires its remaining services plus EventJournal and Identity.

Gotchas

The schema argument does not register handlers by itself. Handler registration comes from the supplied layer, and writing an event without a registered handler dies with Event handler not found for: "<tag>".

See

schema for creating the schema argument from event groups
group for building the handler layer consumed by this layer
layerEventLog for installing the runtime and registry without combining a handler layer

Signature

declare function layer<Groups extends Any, E, R>(_schema: EventLogSchema<Groups>, layer: Layer<ToService<Groups>, E, R>): Layer<EventLog | Registry, E, EventJournal | Identity | Exclude<R, EventLog | Registry>>

layerEventLog

unstable

Since v4.0.0

Provides EventLog and Registry using the configured EventJournal and Identity.

Signature

declare const layerEventLog: Layer.Layer<EventLog | Registry, never, EventJournal | Identity>

layerRegistry

unstable

Since v4.0.0

Provides an in-memory Registry for event handlers, compactors, remote replicas, and reactivity keys.

Signature

declare const layerRegistry: Layer<Registry, never, never>

Other

Handlers

unstable

Since v4.0.0

Namespace containing helper types for Handlers values and handler-producing layers.

Reactivity

groupReactivity

unstable

Since v4.0.0

Registers reactivity keys to invalidate when events from a group are written or replayed.

Details

Pass a single key list for all events or a mapping from event tag to key list.

Signature

declare function groupReactivity<Events extends Any>(group: EventGroup<Events>, keys: readonly Array<string> | { [Tag in string]: readonly Array<string> }): Layer<never, never, Registry>

Schemas

EventLogSchema

interface unstable

Since v4.0.0

Schema describing the event groups that can be written through an EventLog.

Signature

interface EventLogSchema<Groups extends EventGroup.Any> {
  readonly "~effect/eventlog/EventLog/Schema": "~effect/eventlog/EventLog/Schema";
  readonly groups: readonly Array<Groups>;
}

IdentitySchema

unstable

Since v4.0.0

Schema for an event-log identity with a string public key and redacted base64-encoded private key bytes.

Signature

declare const IdentitySchema: Struct<{
  readonly privateKey: decodeTo<Redacted<Uint8Array>, Uint8ArrayFromBase64, never, never>;
  readonly publicKey: String;
}>

schema

unstable

Since v4.0.0

Creates an EventLogSchema from one or more event groups.

Signature

declare function schema<Groups extends readonly Array<Any>>(...groups: Groups): EventLogSchema<Groups[number]>

Services

CurrentStoreId

unstable

Since v4.0.0

Context reference for the store id used by event-log writes and remote replication.

Details

Defaults to the branded store id "default".

Signature

declare class CurrentStoreId extends {} {
  constructor(_: never);
}

EventLog

unstable

Since v4.0.0

Service for writing typed event-log events through registered handlers.

Details

write encodes the event payload, runs the matching handler, commits the entry only when the handler succeeds, and exposes access to the underlying journal entries and destroy operation.

Signature

declare class EventLog extends Shape<"effect/eventlog/EventLog", {
  readonly compact: (options?: {
    readonly remoteId?: RemoteId;
  }) => Effect<readonly Array<{
    readonly remoteId: RemoteId;
    readonly result: "Busy" | "Committed" | "Superseded" | "NoOp" | "NotWanted" | "Aborted";
  }>, EventJournalError | EventLogRemoteError>;
  readonly destroy: Effect<void, EventJournalError>;
  readonly entries: Effect<readonly Array<Entry>, EventJournalError>;
  readonly write: <Groups extends Any, Tag extends string>(options: {
    readonly event: Tag;
    readonly payload: Type<PayloadSchema<Extract<Events<Groups>, {
      readonly tag: Tag;
    }>>>;
    readonly schema: EventLogSchema<Groups>;
  }) => Effect<Type<SuccessSchema<Extract<Events<Groups>, {
    readonly tag: Tag;
  }>>>, EventJournalError | Type<ErrorSchema<Extract<Events<Groups>, {
    readonly tag: Tag;
  }>>>>;
  readonly writeMany: <Groups extends Any, Item extends {
    readonly event: Tag;
    readonly payload: Type<PayloadSchema<Extract<Events<Groups>, {
      readonly tag: Tag;
    }>>>;
  }>(options: {
    readonly events: readonly [Item, Item];
    readonly schema: EventLogSchema<Groups>;
  }) => Effect<[Type<SuccessSchema<Extract<Events<Groups>, {
    readonly tag: ...[...];
  }>>>, ...Array<Type<SuccessSchema<Extract<Events<Groups>, {
    readonly tag: ...[...];
  }>>>>], EventJournalError | Type<ErrorSchema<Extract<Events<Groups>, {
    readonly tag: ...[...];
  }>>>>;
}, this> {
  constructor(_: never);
}

HandlerTransaction

unstable

Since v4.0.6

Context reference for a transaction that wraps the event handlers of each batch replayed from a remote and of each local write or writeMany.

When to use

Use when handlers project into a store with transactions, such as sql.withTransaction or an IndexedDB transaction over the projection tables, so a batch of N events commits once instead of N times.

Details

The journal performs no I/O while the transaction is open: it reads replay conflicts before the transaction starts and commits the batch after it completes. Defaults to running the handlers without a transaction.

If the transaction fails, the journal commits nothing from the batch. For a replayed batch, the remote is subscribed again after a backoff of up to 10 seconds and the batch is delivered again, so later batches from that remote wait until it succeeds; local writes and other remotes are not blocked. A local write fails with the error and is not retried.

Gotchas

A replayed entry that fails to decode, or whose handler fails or dies, fails the whole batch, which is retried as above. A failure that repeats on every attempt, such as a payload written by a newer schema, stops sync from that remote until the app is updated to handle it.
In a local write, an error returned by a handler fails the transaction and the write, and the journal commits none of its events.
Failures that abort the underlying transaction fail the whole batch, even when the handler recovers from them. In IndexedDB, any failed request that is not prevented aborts the whole transaction, and in PostgreSQL so does a failed statement outside a savepoint.
Failing the batch does not undo side effects of handlers that already ran. Only changes made inside a transaction that rolls back are undone.
Replay handlers must tolerate running more than once for the same entry.
Unless the transaction and the journal share a database connection, the transaction commits before the journal does, and the two commits are not atomic. A crash between them replays a remote batch on restart, but a local write is lost from the journal while its projection changes remain.
An IndexedDB transaction commits once it has no pending requests and control returns to the event loop. A custom wrapper must stop the fiber from yielding to the scheduler while it is open, by providing References.PreventSchedulerYield as withTransaction from @effect/platform-browser does. Otherwise it can commit partway through a batch, and the remaining handlers fail.
The transaction stays open for the whole batch, so other read-write transactions on the same tables wait for it.
The transaction must not change the error or success of the wrapped effect. Convert its own failures into defects.

Signature

declare class HandlerTransaction extends {} {
  constructor(_: never);
}

Identity

unstable

Since v4.0.0

Context service for an event-log identity containing a public key and redacted private key material.

Details

The identity is used by remote replication for authentication and by the encryption service to derive signing and encryption keys.

Gotchas

publicKey identifies the identity but is not a secret, and nothing ties it to the signing key derived from privateKey. A server binds the first signing key that authenticates for a publicKey, unless its authorization checks the presented signing key.

Signature

declare class Identity extends Shape<"effect/eventlog/EventLog/Identity", {
  readonly privateKey: Redacted<Uint8Array<ArrayBuffer>>;
  readonly publicKey: string;
}, this> {
  constructor(_: never);
}

Registry

unstable

Since v4.0.0

Service that collects event handlers, compaction handlers, remote replicas, and reactivity invalidation keys.

Signature

declare class Registry extends Shape<"effect/eventlog/EventLog/Registry", {
  readonly compactors: ReadonlyMap<string, {
    readonly effect: (options: {
      readonly entries: readonly Array<Entry>;
      readonly remote?: boolean;
      readonly write: (entry: Entry, source?: Entry) => Effect<void>;
    }) => Effect<void>;
    readonly events: ReadonlySet<string>;
    readonly remote: boolean;
  }>;
  readonly handleRemote: (handler: (remote: {
    readonly changes: (options: {
      readonly identity: {
        readonly privateKey: Redacted<Uint8Array<...>>;
        readonly publicKey: string;
      };
      readonly knownEntries?: (ids: readonly Array<...>) => Effect<..., ...>;
      readonly startSequence: number;
      readonly storeId: StoreId;
    }) => Effect<Dequeue<RemoteEntry, EventLogRemoteError>, never, Scope>;
    readonly compaction?: EventLogRemoteCompaction;
    readonly firstSequence?: number;
    readonly id: RemoteId;
    readonly whenAuthenticated: <A, E, R>(effect: Effect<A, E, R>) => Effect<A, EventLogRemoteError | E, Identity | R>;
    readonly write: (options: {
      readonly entries: readonly Array<Entry>;
      readonly identity: {
        readonly privateKey: Redacted<Uint8Array<...>>;
        readonly publicKey: string;
      };
      readonly storeId: StoreId;
    }) => Effect<void | readonly Array<{
      readonly entryId: ...;
      readonly sequence: ...;
    }>, EventLogRemoteError>;
  }) => Effect<void>) => Effect<void>;
  readonly handlers: ReadonlyMap<string, Item<any>>;
  readonly reactivityKeys: Record<string, ReadonlyArray<string>>;
  readonly registerCompaction: (options: {
    readonly effect: (options: {
      readonly entries: readonly Array<Entry>;
      readonly remote?: boolean;
      readonly write: (entry: Entry, source?: Entry) => Effect<void>;
    }) => Effect<void>;
    readonly events: readonly Array<string>;
    readonly remote?: boolean;
  }) => Effect<void, never, Scope>;
  readonly registerHandlerUnsafe: (options: {
    readonly event: string;
    readonly handler: Item<any>;
  }) => void;
  readonly registerReactivity: (keys: Record<string, ReadonlyArray<string>>) => Effect<void, never, Scope>;
  readonly registerRemote: (remote: {
    readonly changes: (options: {
      readonly identity: {
        readonly privateKey: Redacted<Uint8Array<ArrayBuffer>>;
        readonly publicKey: string;
      };
      readonly knownEntries?: (ids: readonly Array<EntryId>) => Effect<readonly Array<...>, EventJournalError>;
      readonly startSequence: number;
      readonly storeId: StoreId;
    }) => Effect<Dequeue<RemoteEntry, EventLogRemoteError>, never, Scope>;
    readonly compaction?: EventLogRemoteCompaction;
    readonly firstSequence?: number;
    readonly id: RemoteId;
    readonly whenAuthenticated: <A, E, R>(effect: Effect<A, E, R>) => Effect<A, EventLogRemoteError | E, Identity | R>;
    readonly write: (options: {
      readonly entries: readonly Array<Entry>;
      readonly identity: {
        readonly privateKey: Redacted<Uint8Array<ArrayBuffer>>;
        readonly publicKey: string;
      };
      readonly storeId: StoreId;
    }) => Effect<void | readonly Array<{
      readonly entryId: ... & ...;
      readonly sequence: number;
    }>, EventLogRemoteError>;
  }) => Effect<void, never, Scope>;
  readonly remotes: ReadonlyMap<RemoteId, {
    readonly changes: (options: {
      readonly identity: {
        readonly privateKey: Redacted<Uint8Array<ArrayBuffer>>;
        readonly publicKey: string;
      };
      readonly knownEntries?: (ids: readonly Array<EntryId>) => Effect<readonly Array<...>, EventJournalError>;
      readonly startSequence: number;
      readonly storeId: StoreId;
    }) => Effect<Dequeue<RemoteEntry, EventLogRemoteError>, never, Scope>;
    readonly compaction?: EventLogRemoteCompaction;
    readonly firstSequence?: number;
    readonly id: RemoteId;
    readonly whenAuthenticated: <A, E, R>(effect: Effect<A, E, R>) => Effect<A, EventLogRemoteError | E, Identity | R>;
    readonly write: (options: {
      readonly entries: readonly Array<Entry>;
      readonly identity: {
        readonly privateKey: Redacted<Uint8Array<ArrayBuffer>>;
        readonly publicKey: string;
      };
      readonly storeId: StoreId;
    }) => Effect<void | readonly Array<{
      readonly entryId: ... & ...;
      readonly sequence: number;
    }>, EventLogRemoteError>;
  }>;
}, this> {
  constructor(_: never);
}

Type IDs

HandlersTypeId

unstable

Since v4.0.0

Runtime property key used to identify Handlers values.

Signature

declare const HandlersTypeId: "~effect/eventlog/EventLog/Handlers"

HandlersTypeId

type unstable

Since v4.0.0

Type-level identifier used to brand Handlers values.

Signature

type HandlersTypeId = "~effect/eventlog/EventLog/Handlers"

SchemaTypeId

unstable

Since v4.0.0

Runtime property key used to identify EventLogSchema values.

Signature

declare const SchemaTypeId: "~effect/eventlog/EventLog/Schema"

SchemaTypeId

type unstable

Since v4.0.0

Type-level identifier used to brand EventLogSchema values.

Signature

type SchemaTypeId = "~effect/eventlog/EventLog/Schema"