# @schemeless/event-store

The runtime core that orchestrates event flow execution, persistence, observer notification, and side-effect retries. It glues your `EventFlow` definitions together with an `IEventStoreRepo` implementation and emits every lifecycle outcome through an observable stream.

## Installation

```bash
yarn add @schemeless/event-store
```

You will also need one of the persistence adapters (for example `@schemeless/event-store-adapter-typeorm` or `@schemeless/event-store-adapter-dynamodb`).

## Define event flows

An event flow describes how a domain event is received, validated, applied, and enriched with consequent events or side effects. Each handler is optional, so you can start with a bare minimum and grow behaviour over time.

```ts
import { EventFlow } from '@schemeless/event-store';

export const userRegisteredFlow: EventFlow = {
  domain: 'user',
  type: 'registered',
  validate: async (event) => {
    if (!event.payload.email) {
      throw new Error('email is required');
    }
  },
  apply: async (event) => {
    // Update projections, write to read models, etc.
  },
};
```

You can attach additional hooks such as `sideEffect` (retryable asynchronous work) or `createConsequentEvents` (fan out new root events) by setting the corresponding properties on the flow object.

### Creating consequent events

When an event spawns additional events via `createConsequentEvents`, the framework automatically maintains the causal relationship:

```ts
export const orderPlacedFlow: EventFlow = {
  domain: 'order',
  type: 'placed',
  createConsequentEvents: (parentEvent) => [
    {
      domain: 'account',
      type: 'transfer',
      payload: {
        fromAccountId: parentEvent.payload.buyerAccountId,
        toAccountId: parentEvent.payload.sellerAccountId,
        amount: parentEvent.payload.total,
      },
      // No need to set causationId or correlationId
      // The framework handles this automatically:
      // - causationId = parentEvent.id
      // - correlationId = parentEvent.correlationId
    },
  ],
};
```

This ensures all derived events share the same `correlationId` (for grouping) while each maintains a `causationId` pointer to its immediate parent (for chain traversal).

## Aggregate Event Flows

If a flow manages aggregate state, declare it with `AggregateEventFlow`. The framework will load the current state automatically before `validate` and `apply`, and it will pass the computed state through replay as well.

```ts
import type { AggregateEventFlow, AggregateEventObserver } from '@schemeless/event-store';

type StockPayload = { amount: number };
type StockState = { count: number };

export const stockFlow: AggregateEventFlow<StockPayload, StockState> = {
  domain: 'stock',
  type: 'updated',
  aggregate: {
    initialState: { count: 0 },
    reducer: (state, event) => ({ count: state.count + event.payload.amount }),
  },
  validate: (event, state) => {
    if (state.count < 0) {
      throw new Error('stock cannot go below zero');
    }
  },
  apply: (_event, state) => state,
};

export const stockObserver: AggregateEventObserver<StockPayload, StockState> = {
  aggregate: true,
  filters: [{ domain: 'stock', type: 'updated' }],
  priority: 0,
  apply: async (event, state) => {
    console.log(event.id, state.count);
  },
};
```

- `aggregate.reducer` is the pure fold function used by `getAggregate()`.
- `validate(event, state)` receives the current aggregate state.
- `apply(event, state)` returns the next state and must not perform side effects.
- Aggregate observers opt in with `aggregate: true`; regular observers still receive `(event)` only.

Use the existing `getAggregate()` helper when you need to read aggregate state manually from application code. Inside aggregate flows, the framework does that work for you.

## Optional Runtime Validation (Recommended)

`@schemeless/event-store` does not require any specific validation library. The framework only calls your `validate` hook, so you can choose the tool that fits your stack.

For TypeScript-first projects, we recommend **Zod** because it gives clear runtime errors and strong type inference.

```ts
import { z } from 'zod';
import type { EventFlow } from '@schemeless/event-store';

const userRegisteredPayload = z.object({
  id: z.string().uuid(),
  email: z.string().email(),
  name: z.string().min(1),
});

type UserRegisteredPayload = z.infer<typeof userRegisteredPayload>;

export const userRegisteredFlow: EventFlow<UserRegisteredPayload> = {
  domain: 'user',
  type: 'registered',
  validate: async (event) => {
    const result = userRegisteredPayload.safeParse(event.payload);
    if (!result.success) {
      throw new Error(result.error.issues.map((i) => `${i.path.join('.')}: ${i.message}`).join('; '));
    }
  },
  apply: async (event) => {
    // event.payload is now validated at runtime
  },
};
```

Alternative options:

- **Valibot**: similar DX to Zod, smaller footprint.
- **TypeBox + Ajv**: best when you need JSON Schema output and ecosystem tooling.
- **class-validator**: useful in decorator/class-based DTO stacks.

Recommendation:

- Keep validation in event flows (`validate` / `preApply`).
- Keep the core event-store package library-agnostic.

## Build the store

`makeEventStore` wires your repository and flows together, returning a `submit` helper, a `replay` function, and lifecycle utilities.

```ts
import { makeEventStore } from '@schemeless/event-store';
import { EventStoreRepo as TypeOrmRepo } from '@schemeless/event-store-adapter-typeorm';

const repo = new TypeOrmRepo({ type: 'sqlite', database: ':memory:' });
const buildStore = makeEventStore(repo);

const eventStore = await buildStore([userRegisteredFlow]);

const [created] = await eventStore.submit(userRegisteredFlow, {
  payload: { id: '123', email: 'user@example.com' },
});

// `submit` resolves after persistence and observer execution.
// Side effects run asynchronously; call `shutdown()` for a full flush before teardown.
```

## Performance & Concurrency

You can configure the concurrency level for the internal queues to optimize throughput. By default, all queues run sequentially (`concurrent: 1`) to guarantee strict ordering.

### Configurable Concurrency

Pass `EventStoreOptions` to `makeEventStore` to enable parallel processing:

```ts
const eventStore = await makeEventStore(repo, {
  mainQueueConcurrent: 5,        // 5 parallel partition workers
  sideEffectQueueConcurrent: 10, // 10 parallel side-effect workers
})([userRegisteredFlow]);
```

> **Note:** `mainQueueConcurrent` controls the number of partition workers. Events routed to the same partition (via `getShardKey`) are always processed sequentially; events on different partitions run in parallel. Set `getShardKey` on your flows to control routing.

### Key-Based Partitioning (Sharding)

To solve race conditions with global parallelism, we support **Key-Based Partitioning**. This ensures events for the same entity (e.g., `userId`) are processed sequentially, while different entities are processed in parallel.

1.  **Define Shard Key**: Implement `getShardKey` in your `EventFlow`:

    ```typescript
    const UserFlow: EventFlow = {
      // ...
      getShardKey: (event) => event.payload.userId,
    };
    ```

2.  **Enable Concurrency**:
    ```typescript
    const store = await makeEventStore(..., {
      mainQueueConcurrent: 10, // 10 parallel shards
    });
    ```

👉 **[Read the Full Migration Guide](MIGRATION.md)** for detailed implementation steps.

## Snapshot Support (Performance)

### When Do You NEED This?

| Your Use Case                                                             | Solution                   | Need `getAggregate`? |
| ------------------------------------------------------------------------- | -------------------------- | -------------------- |
| Prevent concurrent writes to same stream                                  | `expectedSequence` (OCC)   | ❌ No                |
| Update projections/read models                                            | `apply` hook (incremental) | ❌ No                |
| Validate based on **current aggregate state** (e.g., "balance >= amount") | `getAggregate`             | ✅ Yes               |

**OCC (Optimistic Concurrency Control) is O(1)** – it only checks a version number, NOT by replaying events. If your validation logic only needs the event payload itself, you don't need snapshots.

**Snapshots are for reconstructing aggregate state** – When your `validate` or `preApply` hook needs the _current state_ of an aggregate (calculated from all past events), `getAggregate` helps. Snapshots optimize this by caching intermediate state.

### Example: When You Need `getAggregate`

```typescript
// In your EventFlow's validate hook:
validate: async (event) => {
  // You need the current account balance to validate
  const { state } = await eventStore.getAggregate('account', event.payload.accountId, accountReducer, { balance: 0 });

  if (state.balance < event.payload.amount) {
    throw new Error('Insufficient funds');
  }
};
```

### Using `getAggregate`

```typescript
const { state, sequence } = await eventStore.getAggregate(
  'user', // domain
  '123', // identifier
  userReducer, // your reducer function
  { balance: 0 } // initial state
);
```

The helper automatically:

1.  Tries to load a snapshot (if adapter supports it).
2.  Fetches **only** events after the snapshot.
3.  Reduces them to get the final state.

### Adapter Requirements

> [!WARNING]
> As of v3.0.0, **most adapters do NOT implement these methods yet**. Check your adapter's documentation.

To use `getAggregate`, your `IEventStoreRepo` adapter must implement:

1.  **`getStreamEvents(domain, identifier, fromSequence)`** (Required)
2.  **`getSnapshot` / `saveSnapshot`** (Optional, for performance)

### Capability Detection

`eventStore` now exposes `capabilities.aggregate` so callers can branch safely before invoking `getAggregate`:

```typescript
if (!eventStore.capabilities.aggregate) {
  // Fallback to projection/CQRS read model validation
}
```

Adapters can also declare support explicitly with `repo.capabilities.aggregate`.
If omitted, support is inferred from the presence of `repo.getStreamEvents`.

### Adapter Capability Matrix (v3.x)

| Adapter                                    | `capabilities.aggregate`   | `getAggregate` Support                                               |
| ------------------------------------------ | -------------------------- | -------------------------------------------------------------------- |
| `@schemeless/event-store-adapter-dynamodb` | `false` (declared)         | ❌ No                                                                |
| Other adapters                             | inferred / adapter-defined | Check adapter docs or `eventStore.capabilities.aggregate` at runtime |

### `EventStoreOptions` Reference

| property                    | type     | default | description                                                                                                           |
| --------------------------- | -------- | ------- | --------------------------------------------------------------------------------------------------------------------- |
| `mainQueueConcurrent`       | `number` | `1`     | Number of partition workers for main event processing. Events sharing the same shard key are always sequential; events on different partitions run in parallel. |
| `sideEffectQueueConcurrent` | `number` | `1`     | Number of partition workers for side-effect execution. Safe to increase; side effects are retryable and independent of the main pipeline. |

### Fire-and-Forget Observers

For observers that perform non-critical, time-consuming tasks (like sending analytics or notifications) where you don't want to block the main event processing flow, use `fireAndForget: true`.

```ts
const analyticsObserver: SuccessEventObserver = {
  filters: [{ domain: 'user', type: 'registered' }],
  priority: 1,
  fireAndForget: true, // Run asynchronously, do not wait
  apply: async (event) => {
    await sendAnalytics(event);
  },
};
```

- **Non-blocking**: `submit()` resolves without waiting for this observer.
- **Error Isolation**: If this observer throws an error, it is logged but does **not** fail the main event flow.

The returned object exposes:

- `submit(flow, input)` – validates, applies, persists, runs side effects, and notifies observers. Resolves with all created events.
- `on('processed', handler)` – subscribe to processed event notifications without RxJS. Returns an unsubscribe function.
- `replay` – streams stored events back through the flows and success observers, enabling projection rebuilds.
- `shutdown(timeout?)` – drains all in-flight work and closes the repository connection.

## Observers and replay

Register success observers when constructing the store to react to committed events. Observers run inline after each event is persisted, in priority order. During replay, the same lifecycle is used for `upcast` / `validate` / `preApply` / `apply`, then observers are called with the resulting event and aggregate state.

```ts
const logObserver = {
  filters: [{ domain: 'user', type: 'registered' }],
  priority: 100,
  apply: async (event) => {
    console.log('User registered:', event.payload.email);
  },
};

const eventStore = await buildStore([userRegisteredFlow], [logObserver]);
await eventStore.replay();
```

`replay` batches historical records and re-applies each event in chronological order. Observers are called inline after each event, so read models stay consistent after deployments or migrations. Replay uses the same lifecycle as live processing: `upcast`, `validate`, `preApply`, `apply`, then observers. Aggregate observers marked with `aggregate: true` receive `(event, state)` during replay; regular observers receive only `(event)`.

## Graceful Shutdown

To prevent data loss or resource leaks during application shutdown (or testing), use the `shutdown` method. This stops accepting new events, drains existing queues, and closes repository connections.

```ts
// 30 second timeout
await eventStore.shutdown(30000);
```

This is particularly useful for:

- **Jest Tests**: Preventing "Store is not a constructor" or "file after teardown" errors.
- **Kubernetes**: Handling `SIGTERM` signals gracefully.
- **Updates**: Ensuring queues are empty before deploying new versions.

## Export and Import Events

The core package ships built-in utilities to snapshot the entire repository into a plain array and restore it later. This is adapter-agnostic — it works with every `IEventStoreRepo` implementation.

```ts
import { exportEventsToArray, importEventsFromArray, createSnapshot, parseSnapshot } from '@schemeless/event-store';

// ── Export ──────────────────────────────────────────────────────────────────
const events = await exportEventsToArray(eventStore.eventStoreRepo, {
  pageSize: 200, // optional, default 200
  onProgress: (n) => console.log(`exported ${n}`),
});
const json = JSON.stringify(createSnapshot(events));

// ── Import (restore) ────────────────────────────────────────────────────────
const { events } = parseSnapshot(json); // Date fields automatically restored
await importEventsFromArray(eventStore.eventStoreRepo, events, {
  replace: true, // call resetStore() before writing (optional)
  batchSize: 100, // optional, default 100
  onProgress: (n) => console.log(`imported ${n}`),
});
```

**React Native / Expo integration example:**

```ts
import * as FileSystem from 'expo-file-system';
import * as Sharing from 'expo-sharing';

// Backup
const json = JSON.stringify(createSnapshot(await exportEventsToArray(store.eventStoreRepo)));
const path = FileSystem.documentDirectory + 'backup.json';
await FileSystem.writeAsStringAsync(path, json);
await Sharing.shareAsync(path, { mimeType: 'application/json' });

// Restore
const json = await FileSystem.readAsStringAsync(backupPath);
await importEventsFromArray(store.eventStoreRepo, parseSnapshot(json).events, { replace: true });
```

| Function                                     | Description                                          |
| -------------------------------------------- | ---------------------------------------------------- |
| `exportEventsToArray(repo, opts?)`           | Pages through `getAllEvents` → `IEventStoreEntity[]` |
| `importEventsFromArray(repo, events, opts?)` | Writes in batches; coerces date strings to `Date`    |
| `createSnapshot(events)`                     | Adds `exportedAt` + `count` metadata                 |
| `parseSnapshot(json)`                        | Parses JSON; restores `created` as `Date`            |

## Cascading Revert

Sometimes you need to undo an event and all its derived effects. The framework provides a built-in **revert** mechanism that traverses the causal chain and generates compensating events.

### Define the compensate hook

Each event flow can define a `compensate` hook that returns one or more compensating events:

```ts
export const orderPlacedFlow: EventFlow = {
  domain: 'order',
  type: 'placed',
  compensate: (originalEvent) => ({
    domain: 'order',
    type: 'voided',
    payload: {
      orderId: originalEvent.payload.orderId,
      reason: 'Reverted via framework',
    },
  }),
};
```

The framework adds metadata to each compensating event (`meta.isCompensating = true`, `meta.compensatesEventId`).

### Check if an event can be reverted

Before attempting a revert, check if the event tree is fully covered:

```ts
const result = await eventStore.canRevert(eventId);

if (!result.canRevert) {
  console.log('Cannot revert:', result.reason);
  console.log('Missing hooks:', result.missingCompensateEvents);
}
```

> **Strict mode:** If any event in the tree (including descendants) lacks a `compensate` hook, the revert will fail. This ensures data consistency.

### Preview a revert

See what would be affected without making changes:

```ts
const preview = await eventStore.previewRevert(eventId);

console.log('Root event:', preview.rootEvent.id);
console.log('Descendants:', preview.descendantEvents.length);
console.log('Total affected:', preview.totalEventCount);
```

### Execute a revert

Revert the event and all its descendants:

```ts
const result = await eventStore.revert(eventId);

console.log('Reverted:', result.revertedEventId);
console.log('Compensation events:', result.compensatingEvents);
console.log('Child results:', result.childResults); // Nested results
```

The revert processes events depth-first, starting from the leaves and working up to the root. This ensures that dependent events are compensated before their parents.

> **Root events only:** Only events without a `causationId` (root events) can be reverted. Attempting to revert intermediate events will throw an error.

## Schema Versioning & Upcasting

To support evolving event schemas over time (e.g., changing a price field from a number to an object), you can define a `schemaVersion` and an `upcast` hook in your `EventFlow`.

### How it works

1.  **Tagging**: New events are automatically tagged with the flow's current `schemaVersion` in `event.meta.schemaVersion`.
2.  **Upcasting**: When an older event (lower version) is processed (during `receive` or `replay`), the `upcast` hook is called to migrate it to the current structure.
3.  **Pipeline**: Upcasting happens **before** validation and pre-application, so your `validate` and `apply` logic only ever needs to handle the _current_ schema version.

### Example

```typescript
interface OrderPlacedPayloadV1 {
  price: number;
}

interface OrderPlacedPayloadV2 {
  price: { amount: number; currency: string };
}

// The flow definition uses the LATEST payload type
export const orderPlacedFlow: EventFlow<OrderPlacedPayloadV2> = {
  domain: 'order',
  type: 'placed',

  // 1. Set the current version
  schemaVersion: 2,

  // 2. Define how to migrate from older versions
  upcast: (event, fromVersion) => {
    if (fromVersion < 2) {
      // Migrate V1 -> V2
      return {
        ...event,
        payload: {
          ...event.payload,
          price: { amount: event.payload.price, currency: 'USD' }, // Default to USD
        },
      };
    }
    // Return void if no changes needed
  },

  // 3. Validation and Application logic only sees V2
  validate: async (event) => {
    // event.payload.price is guaranteed to be { amount, currency }
    if (event.payload.price.amount < 0) throw new Error('Negative price');
  },
};
```