From c990ad79f69bdcf1a8f9ab7d552222e10517d385 Mon Sep 17 00:00:00 2001 From: Marcelo Lazaroni Date: Mon, 8 Sep 2025 12:00:56 +0100 Subject: [PATCH 1/6] Move server start to main function --- src/index.ts | 128 +++++++++++++++++++++++++++------------------------ 1 file changed, 67 insertions(+), 61 deletions(-) diff --git a/src/index.ts b/src/index.ts index 2872b50..07f54c2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -13,76 +13,82 @@ import { MembersByCuisineQueryController } from '@/domain/cookingClub/membership import { EvaluateApplicationReactionController } from '@/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionController'; import { MembersByCuisineProjectionController } from '@/domain/cookingClub/membership/projection/membersByCuisine/MembersByCuisineProjectionController'; -// Configure dependency injection -configureDependencies(); +function main() { + // Configure dependency injection + configureDependencies(); -// Create express app -const app = express(); -app.use(express.json()); + // Create express app + const app = express(); + app.use(express.json()); -// Add scoped container middleware -app.use(scopedContainer); + // Add scoped container middleware + app.use(scopedContainer); -// Add routes -app.use('/api/v1/cooking-club/membership/command', (req, res, next) => { - const controller = req.container.resolve(SubmitApplicationCommandController); - return controller.router(req, res, next); -}); -app.use( - '/api/v1/cooking-club/membership/projection', - AmbarAuthMiddleware, - (req, res, next) => { + // Add routes + app.use('/api/v1/cooking-club/membership/command', (req, res, next) => { const controller = req.container.resolve( - MembersByCuisineProjectionController, + SubmitApplicationCommandController, ); return controller.router(req, res, next); - }, -); -app.use('/api/v1/cooking-club/membership/query', (req, res, next) => { - const controller = req.container.resolve(MembersByCuisineQueryController); - return controller.router(req, res, next); -}); -app.use( - '/api/v1/cooking-club/membership/reaction', - AmbarAuthMiddleware, - (req, res, next) => { - const controller = req.container.resolve( - EvaluateApplicationReactionController, - ); + }); + app.use( + '/api/v1/cooking-club/membership/projection', + AmbarAuthMiddleware, + (req, res, next) => { + const controller = req.container.resolve( + MembersByCuisineProjectionController, + ); + return controller.router(req, res, next); + }, + ); + app.use('/api/v1/cooking-club/membership/query', (req, res, next) => { + const controller = req.container.resolve(MembersByCuisineQueryController); return controller.router(req, res, next); - }, -); -app.get('/docker_healthcheck', (_req, res) => res.send('OK')); -app.get('/', (_req, res) => res.send('OK')); + }); + app.use( + '/api/v1/cooking-club/membership/reaction', + AmbarAuthMiddleware, + (req, res, next) => { + const controller = req.container.resolve( + EvaluateApplicationReactionController, + ); + return controller.router(req, res, next); + }, + ); + app.get('/docker_healthcheck', (_req, res) => res.send('OK')); + app.get('/', (_req, res) => res.send('OK')); -// Error handling middleware -app.use( - ( - err: Error, - _req: express.Request, - res: express.Response, - _next: express.NextFunction, - ) => { - log.error('Unhandled error:', err); - res.status(500).json({ - error: err.message, - stack: 'Available in logs', - }); - }, -); + // Error handling middleware + app.use( + ( + err: Error, + _req: express.Request, + res: express.Response, + _next: express.NextFunction, + ) => { + log.error('Unhandled error:', err); + res.status(500).json({ + error: err.message, + stack: 'Available in logs', + }); + }, + ); -// Initialize databases and start server + // Initialize databases and start server -const mongoInitializer = container.resolve(MongoInitializer); -const postgresInitializer = container.resolve(PostgresInitializer); + const mongoInitializer = container.resolve(MongoInitializer); + const postgresInitializer = container.resolve(PostgresInitializer); -Promise.all([postgresInitializer.initialize(), mongoInitializer.initialize()]) - .then(() => { - app.listen(8080, () => { - console.log('Server is running on port 8080'); + Promise.all([postgresInitializer.initialize(), mongoInitializer.initialize()]) + .then(() => { + app.listen(8080, () => { + console.log('Server is running on port 8080'); + }); + }) + .catch((error) => { + console.error('Failed to initialize databases:', error); + process.exit(1); }); - }) - .catch((error) => { - console.error('Failed to initialize databases:', error); - process.exit(1); - }); +} + +main(); From 48b84ac39ff3339fbc5b637ef5790a6a6d5092ca Mon Sep 17 00:00:00 2001 From: Marcelo Lazaroni Date: Wed, 10 Sep 2025 12:50:43 +0100 Subject: [PATCH 2/6] Create EventStore interface in event sourcing lib --- src/lib/eventSourcing/eventStore.ts | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) create mode 100644 src/lib/eventSourcing/eventStore.ts diff --git a/src/lib/eventSourcing/eventStore.ts b/src/lib/eventSourcing/eventStore.ts new file mode 100644 index 0000000..ecb7a90 --- /dev/null +++ b/src/lib/eventSourcing/eventStore.ts @@ -0,0 +1,29 @@ +export { type EventStore }; +import { Event } from '@/common/event/Event'; +import { Aggregate } from '@/common/aggregate/Aggregate'; +import { AggregateAndEventIdsInLastEvent } from '@/common/eventStore/AggregateAndEventIdsInLastEvent'; + +/* Note [Event Store] + + The Event Store is responsible for saving new Events and fetching existing + Events to hydrate / reconstitute Aggregates. + + The Event Store saves Events, but it does not save them directly, it first + converts them to a SerializedEvent. The SerializedEvent is a representation of + the Event that can be stored in a database. + + Additionally, the Event Store does not simply return aggregates, but it + returns an Aggregate plus Event Ids, that would be necessary to append more + events to the Aggregate (event_id and correlation_id in the last event of that Aggregate). + +*/ + +interface EventStore { + findAggregate( + aggregateId: string, + ): Promise>; + + saveEvent(event: Event): Promise; + + doesEventAlreadyExist(eventId: string): Promise; +} From 438d64a97ea1903eafdff053d28037811168bd9e Mon Sep 17 00:00:00 2001 From: Marcelo Lazaroni Date: Wed, 10 Sep 2025 12:50:56 +0100 Subject: [PATCH 3/6] Create generic Postgres connection --- src/lib/postgres.ts | 108 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) create mode 100644 src/lib/postgres.ts diff --git a/src/lib/postgres.ts b/src/lib/postgres.ts new file mode 100644 index 0000000..ac973c0 --- /dev/null +++ b/src/lib/postgres.ts @@ -0,0 +1,108 @@ +export { + Postgres, + PostgresTransaction, + type PoolSettings, + defaultPoolSettings, +}; + +import { Pool, PoolConfig, PoolClient, QueryConfig, QueryResult } from 'pg'; + +class PostgresTransaction { + public closed: boolean = false; + + constructor(private connection: PoolClient) {} + + async commit() { + if (this.closed) { + throw new Error('Committing a closed transaction'); + } + try { + await this.connection.query('COMMIT'); + this.closed = true; + } catch (error) { + this.closed = true; + throw new Error(`Failed to commit transaction: ${error}`); + } + } + + async abort() { + if (this.closed) { + throw new Error('Aborting a closed transaction'); + } + + try { + await this.connection.query('ROLLBACK'); + } catch (error) { + console.error('Failed to rollback PG transaction', error as Error); + } + this.closed = true; + + try { + this.connection.release(); + } catch (error) { + console.error('Failed to release PG connection', error as Error); + } + } + + async query( + query: string | QueryConfig, + values?: string[] | undefined, + ): Promise> { + if (this.closed) { + throw new Error('Querying a closed connection'); + } + + return this.connection.query(query, values); + } +} + +type PoolSettings = { + maxConnections: number; + minConnections: number; + idleTimeoutMillis: number; + connectionTimeoutMillis: number; +}; + +const defaultPoolSettings: PoolSettings = { + maxConnections: 10, + minConnections: 5, + idleTimeoutMillis: 300000, // 5 minutes + connectionTimeoutMillis: 20000, // 20 seconds +}; + +class Postgres { + private readonly pool: Pool; + + constructor(values: { + user: string; + password: string; + host: string; + port: number; + database: string; + poolSettings: PoolSettings; + }) { + const connectionString = `postgresql://${values.user}:${values.password}@${values.host}:${values.port}/${values.database}`; + const config: PoolConfig = { + connectionString, + max: 10, + min: 5, + idleTimeoutMillis: 300000, // 5 minutes + connectionTimeoutMillis: 20000, // 20 seconds + }; + this.pool = new Pool(config); + this.pool.on('error', (err) => { + console.error('Unexpected error on idle client', err); + }); + } + + // Execute an action with a transaction that will be automatically committed at the end. + async withTransaction( + f: (t: PostgresTransaction) => Promise, + ): Promise { + const connection = await this.pool.connect(); + const transaction = new PostgresTransaction(connection); + const result = await f(transaction); + if (!transaction.closed) await transaction.commit(); + return result; + } +} From dc3be94d66760b440c635b4da6fec4bc97a7d45c Mon Sep 17 00:00:00 2001 From: Marcelo Lazaroni Date: Wed, 10 Sep 2025 12:51:24 +0100 Subject: [PATCH 4/6] Implement postgresEventStore --- src/app/postgresEventStore.ts | 168 ++++++++++++++++++++++++++++++++++ 1 file changed, 168 insertions(+) create mode 100644 src/app/postgresEventStore.ts diff --git a/src/app/postgresEventStore.ts b/src/app/postgresEventStore.ts new file mode 100644 index 0000000..3e8c7cf --- /dev/null +++ b/src/app/postgresEventStore.ts @@ -0,0 +1,168 @@ +export { PostgresEventStore }; + +import { Serializer } from '@/common/serializedEvent/Serializer'; +import { Deserializer } from '@/common/serializedEvent/Deserializer'; +import { SerializedEvent } from '@/common/serializedEvent/SerializedEvent'; +import { Event } from '@/common/event/Event'; +import { CreationEvent } from '@/common/event/CreationEvent'; +import { TransformationEvent } from '@/common/event/TransformationEvent'; +import { Aggregate } from '@/common/aggregate/Aggregate'; +import { AggregateAndEventIdsInLastEvent } from '@/common/eventStore/AggregateAndEventIdsInLastEvent'; +import { EventStore } from '@/lib/eventSourcing/eventStore'; +import { PostgresTransaction } from '@/lib/postgres'; + +class PostgresEventStore implements EventStore { + constructor( + private transaction: PostgresTransaction, + private readonly serializer: Serializer, + private readonly deserializer: Deserializer, + private readonly eventStoreTable: string, + ) {} + + async findAggregate( + aggregateId: string, + ): Promise> { + const serializedEvents = + await this.findAllSerializedEventsByAggregateId(aggregateId); + const events = serializedEvents.map((e) => + this.deserializer.deserialize(e), + ); + + const firstEvent: Event | undefined = events[0]; + if (firstEvent == undefined) { + throw new Error(`No events found for aggregateId: ${aggregateId}`); + } + const creationEvent: Event = firstEvent; + if (!this.isCreationEventForAggregate(creationEvent)) { + throw new Error('First event is not a creation event'); + } + const transformationEvents = events.slice(1); + + let aggregate = creationEvent.createAggregate(); + let eventIdOfLastEvent = creationEvent.eventId; + let correlationIdOfLastEvent = creationEvent.correlationId; + + for (const transformationEvent of transformationEvents) { + if (!this.isTransformationEventForAggregate(transformationEvent)) { + throw new Error('Event is not a transformation event'); + } + aggregate = transformationEvent.transformAggregate(aggregate); + eventIdOfLastEvent = transformationEvent.eventId; + correlationIdOfLastEvent = transformationEvent.correlationId; + } + + return { + aggregate, + eventIdOfLastEvent, + correlationIdOfLastEvent, + }; + } + + async saveEvent(event: Event): Promise { + await this.saveSerializedEvent(this.serializer.serialize(event)); + } + + async doesEventAlreadyExist(eventId: string): Promise { + const event = await this.findSerializedEventByEventId(eventId); + return event !== null; + } + + private async findAllSerializedEventsByAggregateId( + aggregateId: string, + ): Promise { + const sql = ` + SELECT id, event_id, aggregate_id, causation_id, correlation_id, + aggregate_version, json_payload, json_metadata, recorded_on, event_name + FROM ${this.eventStoreTable} + WHERE aggregate_id = $1 + ORDER BY aggregate_version ASC + `; + + try { + const result = await this.transaction.query(sql, [aggregateId]); + return result.rows.map(this.mapRowToSerializedEvent); + } catch (error) { + throw new Error( + `Failed to fetch events for aggregate: ${aggregateId}: ${error}`, + ); + } + } + + private async saveSerializedEvent( + serializedEvent: SerializedEvent, + ): Promise { + const sql = ` + INSERT INTO ${this.eventStoreTable} ( + event_id, aggregate_id, causation_id, correlation_id, + aggregate_version, json_payload, json_metadata, recorded_on, event_name + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + `; + + const values = [ + serializedEvent.event_id, + serializedEvent.aggregate_id, + serializedEvent.causation_id, + serializedEvent.correlation_id, + serializedEvent.aggregate_version.toString(), + serializedEvent.json_payload, + serializedEvent.json_metadata, + serializedEvent.recorded_on, + serializedEvent.event_name, + ]; + + try { + await this.transaction.query(sql, values); + } catch (error) { + throw new Error( + `Failed to save event: ${serializedEvent.event_id}: ${error}`, + ); + } + } + + private async findSerializedEventByEventId( + eventId: string, + ): Promise { + const sql = ` + SELECT id, event_id, aggregate_id, causation_id, correlation_id, + aggregate_version, json_payload, json_metadata, recorded_on, event_name + FROM ${this.eventStoreTable} + WHERE event_id = $1 + `; + + try { + const result = await this.transaction.query(sql, [eventId]); + return result.rows.length > 0 + ? this.mapRowToSerializedEvent(result.rows[0]) + : null; + } catch (error) { + throw new Error(`Failed to fetch event: ${eventId}: ${error}`); + } + } + + private mapRowToSerializedEvent(row: any): SerializedEvent { + return { + id: row.id, + event_id: row.event_id, + aggregate_id: row.aggregate_id, + causation_id: row.causation_id, + correlation_id: row.correlation_id, + aggregate_version: row.aggregate_version, + json_payload: row.json_payload, + json_metadata: row.json_metadata, + recorded_on: row.recorded_on, + event_name: row.event_name, + }; + } + + private isCreationEventForAggregate( + event: Event, + ): event is CreationEvent { + return event instanceof CreationEvent; + } + + private isTransformationEventForAggregate( + event: Event, + ): event is TransformationEvent { + return event instanceof TransformationEvent; + } +} From 32342522b502ca1219426552952d50424adfccbc Mon Sep 17 00:00:00 2001 From: Marcelo Lazaroni Date: Wed, 10 Sep 2025 13:05:30 +0100 Subject: [PATCH 5/6] Create initialisation script for PostgreSQL --- src/app/postgresEventStore.ts | 95 ++++++++++++++++++++++++++++++++++- 1 file changed, 94 insertions(+), 1 deletion(-) diff --git a/src/app/postgresEventStore.ts b/src/app/postgresEventStore.ts index 3e8c7cf..6b3f81c 100644 --- a/src/app/postgresEventStore.ts +++ b/src/app/postgresEventStore.ts @@ -1,4 +1,4 @@ -export { PostgresEventStore }; +export { initialize, PostgresEventStore }; import { Serializer } from '@/common/serializedEvent/Serializer'; import { Deserializer } from '@/common/serializedEvent/Deserializer'; @@ -10,6 +10,7 @@ import { Aggregate } from '@/common/aggregate/Aggregate'; import { AggregateAndEventIdsInLastEvent } from '@/common/eventStore/AggregateAndEventIdsInLastEvent'; import { EventStore } from '@/lib/eventSourcing/eventStore'; import { PostgresTransaction } from '@/lib/postgres'; +import { log } from '@/common/util/Logger'; class PostgresEventStore implements EventStore { constructor( @@ -166,3 +167,95 @@ class PostgresEventStore implements EventStore { return event instanceof TransformationEvent; } } + +// Prepare the database to be used as an event store. +async function initialize({ + transaction, + database, + table, + replicationUserName, + replicationUserPass, + replicationPublication, +}: { + transaction: PostgresTransaction; + database: string; + table: string; + replicationUserName: string; + replicationUserPass: string; + replicationPublication: string; +}): Promise { + function run(description: string, query: string) { + log.info(description); + log.info(query); + return transaction.query(query); + } + + await run( + `Creating table ${table}`, + `CREATE TABLE IF NOT EXISTS ${table} ( + id BIGSERIAL NOT NULL, + event_id TEXT NOT NULL UNIQUE, + aggregate_id TEXT NOT NULL, + aggregate_version BIGINT NOT NULL, + causation_id TEXT NOT NULL, + correlation_id TEXT NOT NULL, + recorded_on TEXT NOT NULL, + event_name TEXT NOT NULL, + json_payload TEXT NOT NULL, + json_metadata TEXT NOT NULL, + PRIMARY KEY (id) + );`, + ); + + await run( + 'Creating replication user', + `CREATE USER ${replicationUserName} REPLICATION LOGIN PASSWORD '${replicationUserPass}';`, + ); + + await run( + 'Granting permissions to replication user', + `GRANT CONNECT ON DATABASE "${database}" TO ${replicationUserName};`, + ); + + await run( + 'Granting select to replication user', + `GRANT SELECT ON TABLE ${table} TO ${replicationUserName};`, + ); + + // Create publication + await run( + 'Creating publication for table', + `CREATE PUBLICATION ${replicationPublication} FOR TABLE ${table};`, + ); + + // Create indexes + await run( + 'Creating aggregate id, aggregate version index', + `CREATE UNIQUE INDEX event_store_idx_event_aggregate_id_version ON ${table}(aggregate_id, aggregate_version);`, + ); + + await run( + 'Creating id index', + `CREATE UNIQUE INDEX event_store_idx_event_id ON ${table}(event_id);`, + ); + + await run( + 'Creating causation index', + `CREATE INDEX event_store_idx_event_causation_id ON ${table}(causation_id);`, + ); + + await run( + 'Creating correlation index', + `CREATE INDEX event_store_idx_event_correlation_id ON ${table}(correlation_id);`, + ); + + await run( + 'Creating recording index', + `CREATE INDEX event_store_idx_occurred_on ON ${table}(recorded_on);`, + ); + + await run( + 'Creating event name index', + `CREATE INDEX event_store_idx_event_name ON ${table}(event_name);`, + ); +} From 1b07ba6e9387a6dfeba25ef8e4a0b113a7db3a78 Mon Sep 17 00:00:00 2001 From: Marcelo Lazaroni Date: Wed, 10 Sep 2025 13:13:57 +0100 Subject: [PATCH 6/6] Make configureDependencies return instance of Postgres --- src/di/container.ts | 32 +++++++++++++++++++++++++++++++- src/index.ts | 6 +++--- 2 files changed, 34 insertions(+), 4 deletions(-) diff --git a/src/di/container.ts b/src/di/container.ts index 281511b..7d597fb 100644 --- a/src/di/container.ts +++ b/src/di/container.ts @@ -20,6 +20,8 @@ import { MembersByCuisineProjectionHandler } from '@/domain/cookingClub/membersh import { MembershipApplicationRepository } from '@/domain/cookingClub/membership/projection/membersByCuisine/MembershipApplicationRepository'; import { CuisineRepository } from '@/domain/cookingClub/membership/projection/membersByCuisine/CuisineRepository'; import env from '@/app/environment'; +import { Postgres, defaultPoolSettings } from '@/lib/postgres'; +import * as postgresEventStore from '@/app/postgresEventStore'; function registerEnvironmentVariables() { const postgresConnectionString = @@ -98,8 +100,36 @@ function registerScopedServices() { registerScoped(EvaluateApplicationReactionHandler); } -export function configureDependencies() { +type Dependencies = { + postgres: Postgres; +}; + +export async function configureDependencies(): Promise { registerEnvironmentVariables(); registerSingletons(); registerScopedServices(); + + const postgres = new Postgres({ + user: env.EVENT_STORE_USER, + password: env.EVENT_STORE_PASSWORD, + host: env.EVENT_STORE_HOST, + port: env.EVENT_STORE_PORT, + database: env.EVENT_STORE_DATABASE_NAME, + poolSettings: defaultPoolSettings, + }); + + await postgres.withTransaction((transaction) => + postgresEventStore.initialize({ + transaction, + database: env.EVENT_STORE_DATABASE_NAME, + table: env.EVENT_STORE_CREATE_TABLE_WITH_NAME, + replicationUserName: + env.EVENT_STORE_CREATE_REPLICATION_USER_WITH_USERNAME, + replicationUserPass: + env.EVENT_STORE_CREATE_REPLICATION_USER_WITH_PASSWORD, + replicationPublication: env.EVENT_STORE_CREATE_REPLICATION_PUBLICATION, + }), + ); + + return { postgres }; } diff --git a/src/index.ts b/src/index.ts index 07f54c2..5397ce9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -13,9 +13,9 @@ import { MembersByCuisineQueryController } from '@/domain/cookingClub/membership import { EvaluateApplicationReactionController } from '@/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionController'; import { MembersByCuisineProjectionController } from '@/domain/cookingClub/membership/projection/membersByCuisine/MembersByCuisineProjectionController'; -function main() { +async function main() { // Configure dependency injection - configureDependencies(); + await configureDependencies(); // Create express app const app = express(); @@ -91,4 +91,4 @@ function main() { }); } -main(); +await main();