Skip to content

Commit b86c7a7

Browse files
wataruoguchiwataruoguchi
committed
docs(pencil): README
1 parent 12413b2 commit b86c7a7

2 files changed

Lines changed: 345 additions & 0 deletions

File tree

‎package/1758758113676_event_sourcing_migration_example.ts‎

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
/**
2+
* References:
3+
* https://github.com/event-driven-io/emmett/blob/main/src/packages/emmett-postgresql/src/eventStore/schema/tables.ts
4+
*/
5+
import { sql, type Kysely } from "kysely";
6+
7+
// `any` is required here since migrations should be frozen in time. alternatively, keep a "snapshot" db interface.
8+
export async function up(db: Kysely<any>): Promise<void> {
9+
/**
10+
* ================================================
11+
* Streams
12+
* ================================================
13+
*/
14+
await db.schema
15+
.createTable("streams")
16+
.ifNotExists()
17+
.addColumn("stream_id", "text", (col) => col.notNull())
18+
.addColumn("stream_position", "bigint", (col) => col.notNull())
19+
.addColumn("partition", "text", (col) => col.notNull())
20+
.addColumn("stream_type", "text", (col) => col.notNull())
21+
.addColumn("stream_metadata", "jsonb", (col) => col.notNull())
22+
.addColumn("is_archived", "boolean", (col) =>
23+
col.notNull().defaultTo(false),
24+
)
25+
.addPrimaryKeyConstraint("pk_streams", [
26+
"stream_id",
27+
"partition",
28+
"is_archived",
29+
])
30+
.addUniqueConstraint("uq_streams_stream_id_partition_is_archived", [
31+
"stream_id",
32+
"stream_position",
33+
"partition",
34+
"is_archived",
35+
])
36+
.modifyEnd(sql` PARTITION BY LIST (partition);`)
37+
.execute();
38+
// Unique index for streams
39+
await sql`CREATE UNIQUE INDEX idx_streams_stream_id_partition_is_archived
40+
ON streams (stream_id, partition, is_archived)
41+
INCLUDE (stream_position);`.execute(db);
42+
// DEFAULT partition for streams
43+
await sql`CREATE TABLE IF NOT EXISTS streams_default PARTITION OF streams DEFAULT;`.execute(
44+
db,
45+
);
46+
47+
/**
48+
* ================================================
49+
* Messages
50+
* ================================================
51+
*/
52+
const sequenceName = "emt_global_message_position";
53+
await sql`CREATE SEQUENCE IF NOT EXISTS ${sql.raw(sequenceName)};`.execute(
54+
db,
55+
);
56+
57+
await db.schema
58+
.createTable("messages")
59+
.ifNotExists()
60+
.addColumn("stream_id", "text", (col) => col.notNull())
61+
.addColumn("stream_position", "bigint", (col) => col.notNull())
62+
.addColumn("partition", "text", (col) => col.notNull())
63+
.addColumn("message_kind", "char(1)", (col) => col.notNull().defaultTo("E"))
64+
.addColumn("message_data", "jsonb", (col) => col.notNull())
65+
.addColumn("message_metadata", "jsonb", (col) => col.notNull())
66+
.addColumn("message_schema_version", "text", (col) => col.notNull())
67+
.addColumn("message_type", "text", (col) => col.notNull())
68+
.addColumn("message_id", "text", (col) => col.notNull())
69+
.addColumn("is_archived", "boolean", (col) =>
70+
col.notNull().defaultTo(false),
71+
)
72+
.addColumn("global_position", "bigint", (col) =>
73+
col.defaultTo(sql`nextval('${sql.raw(sequenceName)}')`),
74+
)
75+
// .addColumn("transaction_id", "xid8", (col) => col.notNull()) // TODO: Let's ignore this for now. I don't know how to handle this.
76+
.addColumn("created", "timestamptz(3)", (col) =>
77+
col.notNull().defaultTo(sql`now()`),
78+
)
79+
.addPrimaryKeyConstraint("pk_messages", [
80+
"stream_id",
81+
"stream_position",
82+
"partition",
83+
"is_archived",
84+
])
85+
.modifyEnd(sql` PARTITION BY LIST (partition);`)
86+
.execute();
87+
// DEFAULT partition for messages
88+
await sql`CREATE TABLE IF NOT EXISTS messages_default PARTITION OF messages DEFAULT;`.execute(
89+
db,
90+
);
91+
92+
/**
93+
* ================================================
94+
* Subscriptions
95+
* ================================================
96+
*/
97+
await db.schema
98+
.createTable("subscriptions")
99+
.ifNotExists()
100+
.addColumn("subscription_id", "text", (col) => col.notNull())
101+
.addColumn("version", "integer", (col) => col.notNull().defaultTo(1))
102+
.addColumn("partition", "text", (col) => col.notNull())
103+
.addColumn("last_processed_position", "bigint", (col) => col.notNull())
104+
// .addColumn('last_processed_transaction_id', 'xid8', (col) => col.notNull())
105+
.addPrimaryKeyConstraint("pk_subscriptions", [
106+
"subscription_id",
107+
"partition",
108+
"version",
109+
])
110+
.modifyEnd(sql` PARTITION BY LIST (partition);`)
111+
.execute();
112+
// DEFAULT partition for subscriptions
113+
await sql`CREATE TABLE IF NOT EXISTS subscriptions_default PARTITION OF subscriptions DEFAULT;`.execute(
114+
db,
115+
);
116+
}
117+
118+
// `any` is required here since migrations should be frozen in time. alternatively, keep a "snapshot" db interface.
119+
export async function down(db: Kysely<any>): Promise<void> {
120+
await sql`DROP TABLE IF EXISTS streams_default;`.execute(db);
121+
await sql`DROP TABLE IF EXISTS messages_default;`.execute(db);
122+
await sql`DROP TABLE IF EXISTS subscriptions_default;`.execute(db);
123+
await sql`DROP TABLE IF EXISTS streams;`.execute(db);
124+
await sql`DROP TABLE IF EXISTS messages;`.execute(db);
125+
await sql`DROP TABLE IF EXISTS subscriptions;`.execute(db);
126+
await sql`DROP SEQUENCE IF EXISTS emt_global_message_position;`.execute(db);
127+
}

‎package/README.md‎

Lines changed: 218 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,218 @@
1+
# @wataruoguchi/emmett-event-store-kysely
2+
3+
A Kysely-based event store implementation for [Emmett](https://github.com/event-driven-io/emmett), providing event sourcing capabilities with PostgreSQL.
4+
5+
## Features
6+
7+
- **Event Store**: Emmett event store implementation with Kysely
8+
- **Projections**: Read model projections with automatic event processing
9+
10+
## Installation
11+
12+
```bash
13+
npm install @wataruoguchi/emmett-event-store-kysely @event-driven-io/emmett kysely
14+
```
15+
16+
## Database Setup
17+
18+
First, you need to set up the event store tables in your PostgreSQL database. You can achieve this with [the Kysely migration file](https://github.com/wataruoguchi/poc-emmett/blob/main/package/1758758113676_event_sourcing_migration_example.ts)
19+
20+
## Basic Usage
21+
22+
You can find [the complete working example](https://github.com/wataruoguchi/poc-emmett/tree/main/example). The following are some snippets.
23+
24+
### 1. Setting up the Event Store
25+
26+
```typescript
27+
import { createEventStore } from "@wataruoguchi/emmett-event-store-kysely";
28+
import { Kysely, PostgresDialect } from "kysely";
29+
import { Pool } from "pg";
30+
31+
// Set up your Kysely database connection
32+
const db = new Kysely<YourDatabaseSchema>({
33+
dialect: new PostgresDialect({
34+
pool: new Pool({
35+
connectionString: process.env.DATABASE_URL,
36+
}),
37+
}),
38+
});
39+
40+
// Create the event store
41+
const eventStore = createEventStore({ db, logger });
42+
```
43+
44+
### 2. Using the Event Store with Emmett
45+
46+
```typescript
47+
import { DeciderCommandHandler } from "@event-driven-io/emmett";
48+
import type { EventStore } from "@wataruoguchi/emmett-event-store-kysely";
49+
50+
// Define your domain events and commands
51+
type CreateCartCommand = {
52+
type: "CreateCart";
53+
data: { tenantId: string; cartId: string; currency: string };
54+
};
55+
56+
type CartCreatedEvent = {
57+
type: "CartCreated";
58+
data: { tenantId: string; cartId: string; currency: string };
59+
};
60+
61+
// Create your event handler
62+
export function cartEventHandler({
63+
eventStore,
64+
getContext,
65+
}: {
66+
eventStore: EventStore;
67+
getContext: () => AppContext;
68+
}) {
69+
const handler = DeciderCommandHandler({
70+
decide: createDecide(getContext),
71+
evolve: createEvolve(),
72+
initialState,
73+
});
74+
75+
return {
76+
create: (cartId: string, data: CreateCartCommand["data"]) =>
77+
handler(
78+
eventStore,
79+
cartId,
80+
{ type: "CreateCart", data },
81+
{ partition: data.tenantId, streamType: "cart" }
82+
),
83+
};
84+
}
85+
86+
// Use in your service
87+
const cartService = createCartService({
88+
eventStore,
89+
getContext,
90+
});
91+
```
92+
93+
## Projections (Read Models)
94+
95+
### 1. Creating a Projection
96+
97+
```typescript
98+
import type {
99+
ProjectionEvent,
100+
ProjectionRegistry,
101+
} from "@wataruoguchi/emmett-event-store-kysely/projections";
102+
103+
export function cartsProjection(): ProjectionRegistry<DatabaseExecutor> {
104+
return {
105+
CartCreated: async (db, event) => {
106+
await db
107+
.insertInto("carts")
108+
.values({
109+
stream_id: event.metadata.streamId,
110+
tenant_id: event.data.tenantId,
111+
cart_id: event.data.cartId,
112+
currency: event.data.currency,
113+
items: JSON.stringify([]),
114+
total: 0,
115+
last_stream_position: event.metadata.streamPosition,
116+
})
117+
.execute();
118+
},
119+
ItemAddedToCart: async (db, event) => {
120+
// Update cart with new item
121+
await db
122+
.updateTable("carts")
123+
.set({
124+
items: JSON.stringify([...existingItems, event.data.item]),
125+
total: newTotal,
126+
last_stream_position: event.metadata.streamPosition,
127+
})
128+
.where("stream_id", "=", event.metadata.streamId)
129+
.execute();
130+
},
131+
};
132+
}
133+
```
134+
135+
### 2. Running Projections
136+
137+
```typescript
138+
import {
139+
createProjectionRegistry,
140+
createProjectionRunner,
141+
} from "@wataruoguchi/emmett-event-store-kysely/projections";
142+
import { createReadStream } from "@wataruoguchi/emmett-event-store-kysely";
143+
144+
// Set up projection runner
145+
const readStream = createReadStream({ db, logger });
146+
const registry = createProjectionRegistry(cartsProjection());
147+
const runner = createProjectionRunner({
148+
db,
149+
readStream,
150+
registry,
151+
});
152+
153+
// Project events for a specific stream
154+
await runner.projectEvents("carts-read-model", "cart-123", {
155+
partition: "tenant-456",
156+
batchSize: 100,
157+
});
158+
```
159+
160+
### 3. Projection Worker
161+
162+
Create a worker to continuously process projections:
163+
164+
```typescript
165+
#!/usr/bin/env node
166+
import { createReadStream } from "@wataruoguchi/emmett-event-store-kysely";
167+
import {
168+
createProjectionRegistry,
169+
createProjectionRunner,
170+
} from "@wataruoguchi/emmett-event-store-kysely/projections";
171+
172+
async function main(partition: string) {
173+
const db = getDb();
174+
const readStream = createReadStream({ db, logger });
175+
const registry = createProjectionRegistry(cartsProjection());
176+
const runner = createProjectionRunner({
177+
db,
178+
readStream,
179+
registry,
180+
});
181+
182+
const subscriptionId = "carts-read-model";
183+
const batchSize = 200;
184+
const pollIntervalMs = 1000;
185+
186+
while (true) {
187+
// Get streams for this partition
188+
const streams = await db
189+
.selectFrom("streams")
190+
.select(["stream_id"])
191+
.where("is_archived", "=", false)
192+
.where("partition", "=", partition)
193+
.where("stream_type", "=", "cart")
194+
.execute();
195+
196+
// Process each stream
197+
for (const stream of streams) {
198+
await runner.projectEvents(subscriptionId, stream.stream_id, {
199+
partition,
200+
batchSize,
201+
});
202+
}
203+
204+
await new Promise((r) => setTimeout(r, pollIntervalMs));
205+
}
206+
}
207+
208+
// Run with: node projection-worker.js tenant-123
209+
main(process.argv[2]);
210+
```
211+
212+
## License
213+
214+
MIT
215+
216+
## Contributing
217+
218+
Contributions are welcome! Please feel free to submit a Pull Request.

0 commit comments

Comments
 (0)