@ossy/event-store

Event sourcing primitives for the Ossy platform. Provides Aggregate (ADR 0008 entity streams), EventStore (resource event I/O), ProjectionRebuild / getProjection (derived read models), PushInvalidation (SSE cache invalidation), and AggregateRebuild (startup snapshot rebuilding) on top of MongoDB.

Exports

| Export | Module | Purpose | |--------|--------|---------| | Mongo | mongodb.js | Connect to MongoDB | | Aggregate | aggregate.js | Entity stream facade (delegates to SchemaStream) | | SchemaStream | schema-stream.js | Low-level ADR 0008 entity stream I/O | | EventStore | event-store.js | Append and query resource events | | AggregateRebuild | aggregate-rebuild.js | Startup snapshot rebuild for entities and resources | | ProjectionRebuild | projection-rebuild.js | Incremental and full projection rebuild | | getProjection | projection-queries.js | Read a projection snapshot by scope | | PushInvalidation | push-invalidation.js | Workspace-scoped SSE fan-out | | buildInvalidationKeys, buildPushMessage | push-invalidation.js | Derive client cache keys from events |

Core concepts

The event store keeps two MongoDB collections:

• eventstore — every resource/entity event ever appended. Immutable. ADR 0008 documents use type (schema id), resourceId, event (lifecycle name), version, payload, created, and createdBy.
• aggregates — denormalized state snapshots: entity/resource folds keyed by id, and projection snapshots keyed by { id: scopeId, type: projectionId, kind: 'projection' }. Rebuilt at startup and updated on the changestream.

Aggregate

Facade over SchemaStream for entity aggregates that declare static SchemaId. All methods return Promises.

Reading the current state

import { Aggregate } from '@ossy/event-store'
import { User } from '@ossy/users/server'

const state = await Aggregate.Of(User, userId).then(Aggregate.View())

Creating a new entity

Pass a creation event as the identifier. The event is appended and the stream is returned.

const createdEvent = UsersEvents.Created({ email, firstName, lastName })
const userView = await Aggregate.Of(User, createdEvent).then(Aggregate.View())

resourceId in the event is the entity id; if absent, the event factory generates one.

Appending to an existing entity

await Aggregate.Of(User, userId)
  .then(Aggregate.Add(UsersEvents.NameUpdated({ firstName, lastName, createdBy: userId })))
  .then(Aggregate.Save())

API reference

| Method | Signature | Description | |---|---|---| | Aggregate.Of(Root, id) | (class, string) => Promise<Stream> | Load an entity stream by resourceId. | | Aggregate.Of(Root, event) | (class, object) => Promise<Stream> | Create a new entity by appending the first event. | | Aggregate.Add(event) | (event) => (stream) => Promise<Stream> | Append an event. Chainable. | | Aggregate.Save() | () => (stream) => Promise<void> | Persist the folded snapshot to aggregates. | | Aggregate.View(fn?) | (fn?) => (stream) => state | Fold events through the aggregate View reducer. | | Aggregate.Validate(fn) | (fn) => (stream) => Promise<Stream> | Run validation against (events, savedState). | | Aggregate.Find(id) | (string) => Promise<object \| null> | Find a snapshot document in aggregates by id. | | Aggregate.Collection | Collection | Direct access to the MongoDB aggregates collection. |

Entity aggregate classes require static SchemaId (ADR 0008). Resource documents use @ossy/resources commitResource / mutateResource instead.

EventStore

Low-level access to the eventstore collection.

import { EventStore } from '@ossy/event-store'

| Method | Signature | Description | |---|---|---| | EventStore.AppendResourceEvent(event) | (object) => Promise<object> | Insert an ADR 0008 event document. | | EventStore.GetResourceStream({ resourceId, fromVersion? }) | => Promise<object[]> | Events for one resourceId, optionally after a version. | | EventStore.GetResourceStreams() | => Promise<{ resourceId, type }[]> | All known (resourceId, schemaId) pairs. | | EventStore.FindEvent(query) | (MongoQuery) => Promise<object> | Find a single event. Rejects when not found. | | EventStore.FindEvents(query) | (MongoQuery) => Promise<object[]> | Find multiple events. Rejects when none found. | | EventStore.Aggregate(pipeline) | (Pipeline) => Promise<object[]> | Run a MongoDB aggregation pipeline. | | EventStore.Collection | Collection | Direct access to the eventstore collection. |

Projections

Projection aggregates (kind: 'projection') fold resource events into scope-keyed read models (e.g. workspace booking list).

import { getProjection } from '@ossy/event-store'

const list = await getProjection(workspaceId, '@ossy/booking/data/booking-list')

| API | Description | |-----|-------------| | ProjectionRebuild.registerProjection({ id, Aggregate }) | Register a projection class (also called via AggregateRebuild.registerAggregate when kind === 'projection') | | ProjectionRebuild.dispatch(event) | Apply one changestream event to matching projections | | ProjectionRebuild.rebuildAll() | Replay source events at startup (called from AggregateRebuild.BuildAndSaveAll()) |

Projection classes declare static sources, static scopeFromEvent(event), static Apply(event, state), and optionally static cacheKeys(scopeId, event).

Push invalidation (ADR 0008 §9)

After changestream rebuild, PushInvalidation.publish(event) fans out SSE messages to workspace subscribers.

• Server: GET /events (workspace-scoped via workspaceId header or user settings cookie)
• Client: sdk.subscribePush({ onMessage }) — opt in via PushInvalidationSubscriber or enablePushInvalidation in app config (off by default; see @ossy/sdk-react README)
• Keys: resource:{id}, location:{path}, projection:{id}:{scopeId}, action:{actionId}

Use buildInvalidationKeys(event) to derive keys server-side; buildPushMessage(event) wraps them for SSE payloads. Projection aggregates may define static cacheKeys(scopeId, event) for bespoke list-action keys.

AggregateRebuild

Rebuilds entity/resource snapshots from scratch at startup. Called automatically by @ossy/platform.

import { AggregateRebuild } from '@ossy/event-store'

AggregateRebuild.registerAggregate({ id: 'User', Aggregate: User })
await AggregateRebuild.BuildAndSaveAll()

registerAggregate routes projection modules (Aggregate.kind === 'projection') to ProjectionRebuild. Entity classes are indexed by Aggregate.SchemaId for resource rebuilds.

*.aggregate.js primitive

The platform auto-discovers aggregate files from installed packages. Entity aggregates export { id, Aggregate } with Aggregate.SchemaId; projection aggregates set Aggregate.kind = 'projection'. See PRIMITIVES.md.

MongoDB setup

import { Mongo } from '@ossy/event-store'

await Mongo.connect(process.env.DB_URL)

Recommended indexes (created at startup by ensureEventStoreIndexes):

// eventstore
{ resourceId: 1, version: 1 }  // stream queries

// aggregates
{ id: 1 }                                      // snapshot get-by-id
{ type: 1, 'state.email': 1 }                  // user lookup (sparse)
{ 'state.belongsTo': 1, 'state.location': 1 }  // folder list/search (sparse)

{ type: 1, event: 1 } on eventstore is still useful for projection replay / task triggers but is not created automatically.

Testing

npm test -w @ossy/event-store

Runs unit tests for push invalidation key derivation (push-invalidation.spec.js).