npm package: @ricofritzsche/eventstore
Description: A TypeScript event sourcing library with Postgres persistence, real-time subscriptions, and projection support for building responsive event-sourced applications
Author: Rico Fritzsche
Latest version: 1.0.10

# EventStore

[![npm version](https://img.shields.io/npm/v/@ricofritzsche/eventstore.svg)](https://www.npmjs.com/package/@ricofritzsche/eventstore)

A comprehensive TypeScript implementation of event sourcing with real-time event subscriptions and projections. 
This system provides persistent event storage with automatic notification to subscribers for building responsive, event-sourced applications.

This package is a collaboration between [Ralf Westphal](https://github.com/ralfw) and [Rico Fritzsche](https://github.com/ricofritzsche).

## Installation

```bash
npm install @ricofritzsche/eventstore
```

**NPM Package:** https://www.npmjs.com/package/@ricofritzsche/eventstore

## Additional Store Docs

- Supabase support: [SupabaseEventStore guide](src/eventstore/stores/supabase/README.md)

## High-Level Architecture

The system is built around a core EventStore with pluggable notification system.

### **EventStore** - The Source of Truth
- **Persistent Storage**: Events are immutably stored in PostgreSQL
- **Query Engine**: Fast retrieval with filtering and payload-based queries
- **Optimistic Locking**: Ensures consistency without traditional database locks
- **Auto-Notification**: Automatically notifies subscribers when events are appended
- **Pluggable Notifiers**: Configurable notification systems (memory, database, etc.)

### **Event Notifiers** - Real-time Processing  
- **Subscription Management**: Multiple subscribers can listen to the same events
- **Concurrent Processing**: Events are processed by all subscribers simultaneously
- **Error Isolation**: If one subscriber fails, others continue processing
- **Lifecycle Management**: Clean subscription setup and teardown

## Core Modules

### **EventStore Module** (`src/eventstore/`)

**Purpose**: Persistent event storage with real-time notifications

**Key Components**:
- **`types.ts`** - Core interfaces (Event, EventStore, EventQuery, EventStreamNotifier)
- **`stores/postgres/`** - PostgreSQL implementation of EventStore with subscription support
- **`stores/memory/`** - In-memory implementation of EventStore with subscription support
- **`stores/supabase/`** - Supabase implementation for browser/SPA and tenant-scoped usage
- **`notifiers/memory/`** - In-memory notification system (default)
- **`filter/`** - Event filters and queries

**Responsibilities**:
- Store events immutably in storage medium, e.g. PostgreSQL database or in-memory
- Query events with complex filtering using EventQuery
- Provide atomic consistency through optimistic locking (with CTE-based approach (Postgres))
- Notify subscribers immediately when events are appended
- Manage subscription lifecycle

### **Examples** (`src/examples/banking/`)

**Purpose**: Feature-sliced banking application demonstrating real-world usage

**Key Components**:
- **`features/`** - Individual feature slices with projections
- **`cli.ts`** - Interactive command-line interface
- **Feature Structure**:
  - `core.ts` - Pure business logic
  - `shell.ts` - EventStore integration
  - `types.ts` - Domain types and interfaces
  - `projector.ts` - Database projection logic
  - `listener.ts` - Event subscription handlers

**Banking Features**:
- **Account Management**: Open accounts, deposits, withdrawals, transfers
- **Account Projections**: Real-time account balance updates
- **Analytics Projections**: Monthly account opening statistics
- **Rebuild Functionality**: Projection recovery from event history

## Event Flow Diagram

```
┌────────────────────────────────────────────────────────────────────────────────┐
│                                Event Flow                                      │
├────────────────────────────────────────────────────────────────────────────────┤
│                                                                                │
│  ┌─────────────┐  append()  ┌─────────────┐   notify()   ┌─────────────┐       │
│  │   Command   │ ─────────▶ │ EventStore  │ ───────────▶ │   Event     │       │
│  │  Handler    │            │             │              │  Notifier   │       │
│  └─────────────┘            └─────────────┘              └─────────────┘       │
│                                     │                            │             │
│                                     ▼                            ▼             │
│  ┌─────────────┐            ┌─────────────┐              ┌─────────────┐       │
│  │  PostgreSQL │            │   Events    │              │  Multiple   │       │
│  │  Database   │            │   Saved     │              │ Subscribers │       │
│  └─────────────┘            └─────────────┘              └─────────────┘       │
│                                                                   │            │
│                                                                   ▼            │
│                                                          ┌─────────────┐       │
│  ┌─────────────┐                                         │ Concurrent  │       │
│  │  Queries    │ ◀───────────────────────────────────────│ Processing  │       │
│  │             │                                         │             │       │
│  └─────────────┘                                         └─────────────┘       │
│                                                                   │            │
│                                                                   ▼            │
│                                                          ┌─────────────┐       │
│                                                          │ Projections │       │
│                                                          │   Updated   │       │
│                                                          └─────────────┘       │
│                                                                                │
│                          Real-time, concurrent event processing                │
└────────────────────────────────────────────────────────────────────────────────┘
```

## Subscription System

The subscription system enables real-time, concurrent processing:

```
┌────────────────────────────────────────────────────────────────────────────────┐
│                         Subscription Architecture                              │
├────────────────────────────────────────────────────────────────────────────────┤
│                                                                                │
│                              ┌─────────────┐                                   │
│                              │ EventStore  │                                   │
│                              │             │                                   │
│                              │ ┌─────────┐ │                                   │
│                              │ │New Event│ │                                   │
│                              │ │Appended │ │                                   │
│                              │ └─────────┘ │                                   │
│                              └──────┬──────┘                                   │
│                                     │                                          │
│                                     ▼                                          │
│                              ┌─────────────┐                                   │
│                              │Event        │                                   │
│                              │Notifier     │                                   │
│                              │(Memory)     │                                   │
│                              └──────┬──────┘                                   │
│                                     │                                          │
│                    ┌────────────────┼────────────────┐                         │
│                    │                │                │                         │
│                    ▼                ▼                ▼                         │
│            ┌─────────────┐  ┌─────────────┐  ┌─────────────┐                   │
│            │ Projection  │  │ Analytics   │  │   Business  │                   │
│            │ Subscriber  │  │ Subscriber  │  │ Logic       │                   │
│            │             │  │             │  │ Subscriber  │                   │
│            └─────────────┘  └─────────────┘  └─────────────┘                   │
│                    │                │                │                         │
│                    ▼                ▼                ▼                         │
│            ┌─────────────┐  ┌─────────────┐  ┌─────────────┐                   │
│            │ Read Model  │  │ Metrics &   │  │ Notifications│                  │
│            │ Database    │  │ Reports     │  │ & Workflows │                   │
│            │             │  │             │  │             │                   │
│            └─────────────┘  └─────────────┘  └─────────────┘                   │
│                                                                                │
│              Concurrent, independent processing of the same events             │
└────────────────────────────────────────────────────────────────────────────────┘
```

## Quick Start

### 1. Setup
```bash
# Install the package
npm install @ricofritzsche/eventstore

# Start Postgres
docker run --name eventstore-postgres -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=bank -p 5432:5432 -d postgres:15

# Set connection string
export DATABASE_URL="postgres://postgres:postgres@localhost:5432/bank"
```

### 2. EventQuery
```typescript
import { PostgresEventStore, MemoryEventStore, createQuery, createFilter } from '@ricofritzsche/eventstore';

// Postgres
const eventStore = new PostgresEventStore( {connectionstring: "..."} ); 
await eventStore.initializeDatabase();

// In-memory
// const eventStore = new MemoryEventStore();

// Create events
const events = [
  { eventType: 'UserRegistered', payload: { userId: '123', email: 'user@example.com' } },
  { eventType: 'UserEmailVerified', payload: { userId: '123', verifiedAt: new Date() } }
];

// Subscribe before appending to catch real-time events
const subscription = await eventStore.subscribe(async (events) => {
console.log(`Received ${events.length} new events`);
  // Process events immediately as they're appended
});

// Append events - subscribers will be notified automatically
await eventStore.append(events);

// Query historical events using EventQuery (supports complex OR conditions)
const userFilter = createFilter(['UserRegistered', 'UserEmailVerified'], [{ userId: '123' }]);
const adminFilter = createFilter(['AdminAction'], [{ action: 'user_management' }]);
const query = createQuery(userFilter, adminFilter); // OR between filters

const result = await eventStore.query(query);
console.log(`Found ${result.events.length} historical events`);

// Example: Query with payload conditions
const specificUserQuery = createQuery(
  createFilter(['UserRegistered'], [{ email: 'user@example.com' }]),
  createFilter(['UserEmailVerified'], [{ userId: '123' }])
);
const specificResult = await eventStore.query(specificUserQuery);

// Example: Query only events after a known sequence number (incremental loading)
const incrementalQuery = createQuery(
  { minSequenceNumber: 42 },          // only events with sequenceNumber > 42
  createFilter(['UserRegistered', 'UserEmailVerified'])
);
const incrementalResult = await eventStore.query(incrementalQuery);
// incrementalResult.maxSequenceNumber can be stored and used as minSequenceNumber next time
```

### 3. Atomic Consistency with Optimistic Locking

The EventStore provides atomic consistency through optimistic locking using Common Table Expressions (CTEs (Postgres)). 
This approach ensures that concurrent operations only conflict when they actually depend on the same event context, rather than using traditional aggregate-level locking.

```typescript
// Atomic append with consistency check
const accountEvents = [
  { eventType: 'MoneyDeposited', payload: { accountId: 'acc-123', amount: 100 } }
];

// Create a query for the specific context we want to protect
const accountQuery = createQuery(
  createFilter(['BankAccountOpened', 'MoneyDeposited', 'MoneyWithdrawn'], 
               [{ accountId: 'acc-123' }])
);

// Get current state to determine expected sequence number
const currentState = await eventStore.query(accountQuery);
const expectedMaxSeq = currentState.maxSequenceNumber;

try {
  // Atomic append using EventQuery - only succeeds if no conflicting events were added
  await eventStore.append(accountEvents, accountQuery, expectedMaxSeq);
  console.log('Deposit successful');
  
  // Alternative: Using EventFilter for backward compatibility
  const accountFilter = createFilter(['BankAccountOpened', 'MoneyDeposited', 'MoneyWithdrawn'], 
    [{ accountId: 'acc-123' }]);
  await eventStore.append(accountEvents, accountFilter, expectedMaxSeq);
  
} catch (error) {
  if (error.message.includes('optimistic locking')) {
    // Retry the operation with updated state
    console.log('Concurrent modification detected, retrying...');
  }
}
```

**How CTE-based Consistency Works (Postgres):**

1. **Context-Specific Protection**: Only events matching the query filter are considered for consistency
2. **Atomic Check-and-Insert**: Uses SQL CTE to check max sequence number and insert events in one transaction
3. **Reduced Conflicts**: Commands only conflict when they actually affect the same business context
4. **High Concurrency**: Multiple commands can run simultaneously if they don't share context

The underlying SQL implementation:
```sql
WITH context AS (
  SELECT MAX(sequence_number) AS max_seq
  FROM events
  WHERE [filter conditions]
)
INSERT INTO events (event_type, payload, sequence_number)
SELECT event_type, payload, (max_seq + row_number())
FROM context, unnest($1) AS new_events
WHERE COALESCE(max_seq, 0) = $2
```

### 4. Event Subscription
```typescript
import { PostgresEventStore, createQuery, createFilter } from '@ricofritzsche/eventstore';

// Create EventStore with default MemoryEventStreamNotifier
const eventStore = new PostgresEventStore({connectionstring: "..."});
await eventStore.initializeDatabase();

// Subscribe to events for real-time processing
const subscription = await eventStore.subscribe(async (events) => {
  for (const event of events) {
    console.log('Processing event:', event.eventType);
    
    // Update projections, analytics, send notifications, etc.
    switch (event.eventType) {
      case 'BankAccountOpened':
        await updateAccountProjection(event);
        await updateAnalytics(event);
        break;
      case 'MoneyDeposited':
        await updateAccountBalance(event);
        break;
    }
  }
});
```

### 5. Pluggable Notifiers
Replace the notification system with your own:

```typescript
import { EventStreamNotifier, PostgresEventStore } from '@ricofritzsche/eventstore';

class DatabaseEventStreamNotifier implements EventStreamNotifier {
  // Custom implementation using database triggers, message queues, etc.
}

const eventStore = new PostgresEventStore({
  notifier: new DatabaseEventStreamNotifier()
});
```


## API Reference

### PostgresEventStore

The main event store implementation with PostgreSQL persistence.

```typescript
class PostgresEventStore {
  constructor(options?: PostgresEventStoreOptions)
  
  // Initialize database schema
  async initializeDatabase(): Promise<void>
  
  // Query events with filtering using EventQuery or EventFilter
  async query(eventQuery: EventQuery): Promise<QueryResult>
  async query(eventFilter: EventFilter): Promise<QueryResult>
  
  // Append events with multiple overloads for flexibility
  async append(events: Event[]): Promise<void>
  async append(events: Event[], filterCriteria: EventQuery, expectedMaxSequenceNumber: number): Promise<void>
  async append(events: Event[], filterCriteria: EventFilter, expectedMaxSequenceNumber: number): Promise<void>
  
  // Subscribe to new events
  async subscribe(handle: HandleEvents): Promise<EventSubscription>
  
  // Clean up resources
  async close(): Promise<void>
}
```

### Query and Filter Functions

```typescript
// Create event filters (AND within filter, OR between payload predicates)
createFilter(eventTypes: string[], payloadPredicates?: Record<string, unknown>[]): EventFilter

// Create event queries (OR between filters)
createQuery(...filters: EventFilter[]): EventQuery
```

### Append Method Overloads

```typescript
// Simple append without consistency checks
await eventStore.append(events);

// Append with EventQuery and optimistic locking
await eventStore.append(events, eventQuery, expectedMaxSequenceNumber);

// Append with EventFilter and optimistic locking (backward compatible)
await eventStore.append(events, eventFilter, expectedMaxSequenceNumber);
```

### EventQuery Structure

```typescript
interface EventFilter {
  readonly eventTypes: string[]; // OR condition
  readonly payloadPredicates?: Record<string, unknown>[]; // OR condition
}

interface EventQuery {
  readonly filters: EventFilter[]; // OR condition between filters
}
```

**Query Logic:**
- Within an `EventFilter`: event types are OR'ed AND payload predicates are OR'ed
- Within an `EventQuery`: filters are OR'ed
- This provides flexible querying: `((eventType1 OR eventType2) AND (payload1 OR payload2)) OR (eventType3 AND payload3)`

### MemoryEventStore

The in-memory event store behaves just like the Postgres event store - except all events are just kept in memory.

```typescript
const es = new MemoryEventStore()
...
```

The in-memory event store is a great tool for quick spikes and especially testing: inject the MemoryEventStore into SUTs during tests, inject the PostgresEventStore for production. The MemoryEventStore can easily be populated with events for testing before calling the SUT.

#### Persistence

For small, local applications/proof-of-concepts/spikes it's handy to work with the MemoryEventStore and also be able to persist its events. That can be done at any time like this:

```typescript
es.storeToFile("events.json");
```

Events are written to a file in a JSON data structure.

To instantiate a MemoryEventStore from persisted events, call the factory method:

```typescript
const es1 = MemoryEventStore.createFromFile("events.json");
const es2 = MemoryEventStore.createFromFile("events.json", true); // to ignoer a missing file
```

The MemoryEventStore can also be run in write-through mode, meaning that no explicit calls to persistence methods are needed. Rather the event store persists all changes automatically. That's quite slow, of course, but it's convenient and sufficient for small scenarios.

```typescript
const es1 = new MemoryEventStore("events.json") // when creating the event store with a file it's set to write-thru mode

const es2 = MemoryEventStore.createFromFile("events.json", true, true) // the same happens when passing true as the last parameter to the factory method
```

### Backward Compatibility

The EventStore maintains full backward compatibility with existing code using `EventFilter`:

```typescript
// Legacy approach (still supported)
const filter = createFilter(['UserRegistered'], [{ userId: '123' }]);
const result = await eventStore.query(filter);

// With optimistic locking using EventFilter
const currentState = await eventStore.query(filter);
await eventStore.append(newEvents, filter, currentState.maxSequenceNumber);

// New approach with EventQuery
const query = createQuery(
  createFilter(['UserRegistered'], [{ userId: '123' }]),
  createFilter(['UserUpdated'], [{ userId: '123' }])
);
const result2 = await eventStore.query(query);

// EventFilter is automatically converted to EventQuery internally
// Both approaches provide the same functionality and performance
```


## License

MIT License - see [LICENSE](LICENSE) file for details