Drizzle (PostgreSQL) Driver

The Drizzle driver persists events to a PostgreSQL database using Drizzle ORM. It is a recommended choice for production.

import { createEventStore, createAggregateRoot } from '@requence/event-sourcing/drizzle'
import { drizzle } from 'drizzle-orm/node-postgres'

const db = drizzle(process.env.DATABASE_URL!)

const eventStore = createEventStore({
  database: db,
  aggregateRoots: [userAggregateRoot],
})

Database Schema

The driver uses a dedicated event_sourcing PostgreSQL schema with the following tables:

Table	Purpose
`events`	Append-only event log with position, stream, and metadata
`checkpoints`	Last-processed position (and optimistic-concurrency `version`) per projection/process manager
`aggregate_root_snapshots`	Cached aggregate state for fast rehydration
`projection_snapshots`	Cached projection state for snapshot-based projections
`projection_applied_events`	Counter for snapshot frequency tracking

The table definitions are exported from @requence/event-sourcing/drizzle/postgres for use in Drizzle migrations:

import {
  events,
  checkpoints,
  aggregateRootSnapshots,
  projectionSnapshots,
  projectionAppliedEvents,
} from '@requence/event-sourcing/drizzle/postgres'

Configuration

import { redisLock } from '@requence/event-sourcing/redis'
import Redis from 'ioredis'

const redis = new Redis(process.env.REDIS_URL)

const eventStore = createEventStore({
  // Required — Drizzle database instance or factory
  database: db,

  // Required — one or more aggregate roots
  aggregateRoots: [user, project],

  // Optional — distributed lock for multi-instance deployments
  lock: redisLock(redis),

  // Optional — disable automatic initialization
  autoInit: false,

  // Optional — callback invoked after events are appended
  onEventsAppended(events) {
    // e.g. broadcast via WebSocket
  },

  // Optional — replay and refresh progress callbacks
  onProjectionReplay: { /* ... */ },
  onProcessManagerRefresh: { /* ... */ },

  // Optional — post-process events before persistence
  postProcessEvent(event, aggregateRoot) {
    return event
  },
})