diff --git a/.gitignore b/.gitignore index f46a516..7181134 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ node_modules/ dist/ *.log +tags diff --git a/src/app/commandHandler.ts b/src/app/commandHandler.ts deleted file mode 100644 index 3d00628..0000000 --- a/src/app/commandHandler.ts +++ /dev/null @@ -1,59 +0,0 @@ -export { handleCommand }; - -import { Response } from '@/lib/router'; -import { EventStore } from '@/lib/eventSourcing/eventStore'; -import { Decoder, decode } from '@/lib/json/decoder'; -import * as express from 'express'; -import * as router from '@/lib/router'; -import { Future } from '@/lib/Future'; -import { Result, Failure } from '@/lib/Result'; - -type Projections = {}; -type Services = {}; - -type CommandController = { - decoder: Decoder; - handler: (v: { - command: Command; - store: EventStore; - projections: Projections; - services: Services; - }) => Future; -}; - -function handleCommand( - withEventStore: (f: (store: EventStore) => T) => T, - services: Services, - projections: Projections, - { decoder, handler }: CommandController, -): express.Handler { - return router.route((req) => - decodeCommand(decoder, req).chain((command) => - withEventStore((store) => - handler({ - command, - store, - projections, - services, - }), - ), - ), - ); -} - -function decodeCommand( - decoder: Decoder, - req: express.Request, -): Future { - const decoded: Result = decode(decoder, req.body); - if (decoded instanceof Failure) { - return Future.reject( - router.json({ - status: 400, - content: { message: `Unable to decode command: ${decoded.error}` }, - }), - ); - } - - return Future.resolve(decoded.value); -} diff --git a/src/app/event.ts b/src/app/event.ts deleted file mode 100644 index 5c078c3..0000000 --- a/src/app/event.ts +++ /dev/null @@ -1,87 +0,0 @@ -import { - Aggregate, - Id, - CreationEvent, - TransformationEvent, -} from '@/lib/eventSourcing/event'; - -import * as s from '@/lib/json/schema'; - -class User implements Aggregate { - constructor( - readonly aggregateId: Id, - readonly aggregateVersion: number, - readonly name: string, - ) {} -} - -export class CreateUser implements CreationEvent { - static type: 'CreateUserr' = 'CreateUserr'; - static schemaArgs = s.object({ - type: s.stringLiteral(CreateUser.type), - aggregateId: Id.schema(), - name: s.string, - }); - static schema = CreateUser.schemaArgs.dimap( - (v) => new CreateUser(v), - (v) => v.values, - ); - - schema = CreateUser.schema; - constructor(readonly values: s.Infer) {} - createAggregate() { - return new User(new Id('wat'), 0, this.values.name); - } -} - -export class AddName implements TransformationEvent { - static type: 'AddName' = 'AddName'; - constructor(readonly values: s.Infer) {} - - static schemaArgs = s.object({ - type: s.stringLiteral(AddName.type), - aggregateId: Id.schema(), - name: s.string, - }); - - static schema = AddName.schemaArgs.dimap( - (v) => new AddName(v), - (v) => v.values, - ); - readonly schema = AddName.schema; - - transformAggregate(agg: User): User { - const u = new User( - agg.aggregateId, - agg.aggregateVersion + 1, - this.values.name, - ); - return u; - } -} - -export class RemoveName implements TransformationEvent { - static type: 'RemoveName' = 'RemoveName'; - constructor(readonly values: s.Infer) {} - - static schemaArgs = s.object({ - type: s.stringLiteral(RemoveName.type), - aggregateId: Id.schema(), - name: s.string, - }); - - static schema = RemoveName.schemaArgs.dimap( - (v) => new RemoveName(v), - (v) => v.values, - ); - readonly schema = RemoveName.schema; - - transformAggregate(agg: User): User { - const u = new User( - agg.aggregateId, - agg.aggregateVersion + 1, - this.values.name, - ); - return u; - } -} diff --git a/src/app/postgresEventStore.ts b/src/app/eventStore.ts similarity index 90% rename from src/app/postgresEventStore.ts rename to src/app/eventStore.ts index 6d42f1f..0c56f2a 100644 --- a/src/app/postgresEventStore.ts +++ b/src/app/eventStore.ts @@ -1,4 +1,4 @@ -export { initialize, PostgresEventStore }; +export { initialize, PostgresEventStore, type WithEventStore }; import { Json } from '@/lib/json/types'; import { @@ -7,26 +7,28 @@ import { Aggregate, CreationEvent, TransformationEvent, - EventClass, EventInfo, } from '@/lib/eventSourcing/event'; import { EventStore, - Hydrator, + Schemas, Constructor, EventData, - schema_EventData, } from '@/lib/eventSourcing/eventStore'; import { PostgresTransaction } from '@/lib/postgres'; import { log } from '@/common/util/Logger'; -import { IdGenerator } from '@/common/util/IdGenerator'; -import { encode } from '@/lib/json/schema'; import { POSIX } from '@/lib/time'; +import { Future } from '@/lib/Future'; + +type WithEventStore = ( + onError: (e: Error) => E, + f: (s: EventStore) => Future, +) => Future; class PostgresEventStore implements EventStore { constructor( private transaction: PostgresTransaction, - private readonly hydrator: Hydrator, + private readonly schemas: Schemas, private readonly eventStoreTable: string, ) {} @@ -35,22 +37,22 @@ class PostgresEventStore implements EventStore { aggregateId: Id, ): Promise<{ aggregate: T; lastEvent: EventInfo }> { const events = await this.findAll(aggregateId); - const { lastEvent, aggregate } = this.hydrator + const { lastEvent, aggregate } = this.schemas .hydrate(cls, events) .unwrap((e) => e); return { aggregate, lastEvent }; } - async save, T extends Aggregate>(args: { + async emit>(args: { aggregate: Constructor; - event: CreationEvent | TransformationEvent; + event: CreationEvent | TransformationEvent; event_id?: Id>; correlation_id?: Id>; causation_id?: Id>; }): Promise { const event = args.event; - const event_id = args.event_id || new Id(IdGenerator.generateRandomId()); + const event_id = args.event_id || Id.random(); let info: EventInfo; switch (true) { case event instanceof CreationEvent: { @@ -121,14 +123,14 @@ class PostgresEventStore implements EventStore { } } - private async insert>(edata: EventData) { + private async insert>(edata: EventData) { 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 serialized = encode(schema_EventData(edata.event.schema), edata); + const serialized = this.schemas.encode(edata); const values = [ // @ts-ignore diff --git a/src/app/events.ts b/src/app/events.ts new file mode 100644 index 0000000..49bf0ed --- /dev/null +++ b/src/app/events.ts @@ -0,0 +1,21 @@ +/* + Schemas for all application events +*/ +export { schemas }; + +import { Schemas, CSchema, TSchema } from '@/lib/eventSourcing/eventStore'; +import { ApplicationSubmitted } from '@/domain/cookingClub/membership2/events/membership/applicationSubmitted'; +import { ApplicationEvaluated } from '@/domain/cookingClub/membership2/events/membership/applicationEvaluated'; + +const schemas = new Schemas([ + new CSchema( + ApplicationSubmitted.aggregate, + ApplicationSubmitted.schema, + ApplicationSubmitted.type, + ), + new TSchema( + ApplicationEvaluated.aggregate, + ApplicationEvaluated.schema, + ApplicationEvaluated.type, + ), +]); diff --git a/src/app/handleCommand.ts b/src/app/handleCommand.ts new file mode 100644 index 0000000..593e8c2 --- /dev/null +++ b/src/app/handleCommand.ts @@ -0,0 +1,71 @@ +export { handleCommand, type CommandController, type CommandHandler }; + +import { Response } from '@/lib/router'; +import { EventStore } from '@/lib/eventSourcing/eventStore'; +import { Decoder, decode } from '@/lib/json/decoder'; +import * as express from 'express'; +import * as router from '@/lib/router'; +import { Future } from '@/lib/Future'; +import { Result, Failure } from '@/lib/Result'; +import { Repositories, Projections, allProjections } from '@/app/projections'; +import { Services } from '@/app/services'; +import { WithProjectionStore } from '@/app/projectionStore'; +import { WithEventStore } from '@/app/eventStore'; + +type CommandHandler = (v: { + command: Command; + store: EventStore; + projections: Projections; + services: Services; +}) => Future; + +type CommandController = { + decoder: Decoder; + handler: CommandHandler; +}; + +const onStoreError = (_: Error): Response => + router.json({ + status: 500, + content: { message: 'Internal Server Error' }, + }); + +function handleCommand( + withEventStore: WithEventStore, + withProjectionStore: WithProjectionStore, + services: Services, + repositories: Repositories, + { decoder, handler }: CommandController, +): express.Handler { + return router.route((req) => + decodeCommand(decoder, req).chain((command) => + withProjectionStore(onStoreError, (projectionStore) => + withEventStore(onStoreError, (store) => + handler({ + command, + store, + projections: allProjections(repositories, projectionStore), + services, + }), + ), + ), + ), + ); +} + +function decodeCommand( + decoder: Decoder, + req: express.Request, +): Future { + const decoded: Result = decode(decoder, req.body); + if (decoded instanceof Failure) { + return Future.reject( + router.json({ + status: 400, + content: { message: `Unable to decode command: ${decoded.error}` }, + }), + ); + } + + return Future.resolve(decoded.value); +} diff --git a/src/app/handleProjection.ts b/src/app/handleProjection.ts new file mode 100644 index 0000000..ef7884c --- /dev/null +++ b/src/app/handleProjection.ts @@ -0,0 +1,88 @@ +export { + type ProjectionHandler, + type ProjectionController, + handleProjection, + decodeEvent, +}; + +import { Event, EventInfo } from '@/lib/eventSourcing/event'; +import { EventData } from '@/lib/eventSourcing/eventStore'; +import { Decoder, decode } from '@/lib/json/decoder'; +import * as express from 'express'; +import * as router from '@/lib/router'; +import * as Ambar from '@/lib/ambar'; +import { AmbarResponse } from '@/lib/ambar'; +import { Future } from '@/lib/Future'; +import { Result, Failure } from '@/lib/Result'; +import { Maybe, Nothing } from '@/lib/Maybe'; +import { Projections, Repositories, allProjections } from '@/app/projections'; +import { + MongoProjectionStore, + WithProjectionStore, +} from '@/app/projectionStore'; + +type ProjectionHandler = (v: { + event: E; + info: EventInfo; + projections: Projections; + store: MongoProjectionStore; +}) => Future; + +type ProjectionController> = { + decoder: Decoder>; + handler: ProjectionHandler; +}; + +const onProjectionStoreError = (err: Error) => + new Ambar.ErrorMustRetry(err.message); + +function handleProjection>( + withProjectionStore: WithProjectionStore, + repositories: Repositories, + { decoder, handler }: ProjectionController, +): express.Handler { + return router.route((req) => + decodeEvent(decoder, req) + .chain(({ event, info }) => + withProjectionStore(onProjectionStoreError, (store) => + handler({ + event, + info, + projections: allProjections(repositories, store), + store, + }), + ), + ) + .map((_) => new Ambar.Success()) + .bimap(Ambar.toResponse, Ambar.toResponse), + ); +} + +function decodeEvent( + decoder: Decoder>, + req: express.Request, +): Future> { + const bodyDecoder: Decoder>> = + Ambar.payloadDecoder(decoder); + + const decoded: Result>> = decode( + bodyDecoder, + req.body, + ); + + if (decoded instanceof Failure) { + return Future.reject( + new Ambar.ErrorMustRetry(`Unable to decode command: ${decoded.error}`), + ); + } + + if (decoded.value.event instanceof Nothing) { + // ignored + return Future.reject(new Ambar.Success()); + } + + return Future.resolve({ + info: decoded.value.info, + event: decoded.value.event.value, + }); +} diff --git a/src/app/queryHandler.ts b/src/app/handleQuery.ts similarity index 51% rename from src/app/queryHandler.ts rename to src/app/handleQuery.ts index 057db14..670392c 100644 --- a/src/app/queryHandler.ts +++ b/src/app/handleQuery.ts @@ -1,37 +1,40 @@ -export { handleQuery }; +export { handleQuery, type QueryController, type QueryHandler }; import { Response } from '@/lib/router'; -import { Event } from '@/lib/eventSourcing/event'; import { Decoder, decode } from '@/lib/json/decoder'; import * as express from 'express'; import * as router from '@/lib/router'; import { Future } from '@/lib/Future'; import { Result, Failure } from '@/lib/Result'; +import { Projections, Repositories, allProjections } from '@/app/projections'; +import { WithProjectionStore } from '@/app/projectionStore'; +import { internalServerError } from '@/app/responses'; -type Projections = {}; -type Services = {}; +type QueryHandler = (v: { + query: Query; + projections: Projections; +}) => Future; type QueryController = { decoder: Decoder; - handler: (v: { - query: Query; - projections: Projections; - services: Services; - }) => Future; + handler: QueryHandler; }; -function handleQuery>( - projections: Projections, - services: Services, - { decoder, handler }: QueryController, +const onProjectionStoreError = (_: Error) => internalServerError; + +function handleQuery( + withProjectionStore: WithProjectionStore, + repositories: Repositories, + { decoder, handler }: QueryController, ): express.Handler { return router.route((req) => decodeQuery(decoder, req).chain((query) => - handler({ - query, - projections, - services, - }), + withProjectionStore(onProjectionStoreError, (store) => + handler({ + query, + projections: allProjections(repositories, store), + }), + ), ), ); } diff --git a/src/app/handleReaction.ts b/src/app/handleReaction.ts new file mode 100644 index 0000000..d96049c --- /dev/null +++ b/src/app/handleReaction.ts @@ -0,0 +1,80 @@ +export { + wrapWithEventStore, + handleReaction, + type ReactionHandler, + type ReactionController, +}; + +import { EventStore } from '@/lib/eventSourcing/eventStore'; +import { Event, EventInfo } from '@/lib/eventSourcing/event'; +import { Decoder } from '@/lib/json/decoder'; +import * as express from 'express'; +import * as router from '@/lib/router'; +import * as Ambar from '@/lib/ambar'; +import { AmbarResponse, ErrorMustRetry } from '@/lib/ambar'; +import { Future } from '@/lib/Future'; +import { Maybe } from '@/lib/Maybe'; +import { decodeEvent } from '@/app/handleProjection'; + +type Projections = {}; +type Services = {}; + +type ReactionController> = { + decoder: Decoder>; + handler: ReactionHandler; +}; + +type ReactionHandler = (v: { + event: E; + info: EventInfo; + projections: Projections; + services: Services; + store: EventStore; +}) => Future; + +const onEventStoreError = (err: Error) => new Ambar.ErrorMustRetry(err.message); + +type WithStoreGeneric = ( + onError: (e: Error) => E, + f: (store: EventStore) => Future, +) => Future; + +type WithStoreConcrete = ( + f: (store: EventStore) => Future, +) => Future; + +const wrapWithEventStore = ( + withEventStore: WithStoreGeneric, +): WithStoreConcrete => + function (f) { + return withEventStore(onEventStoreError, (store) => f(store)); + }; + +function handleReaction>( + withEventStore: WithStoreConcrete, + projections: Projections, + services: Services, + { decoder, handler }: ReactionController, +): express.Handler { + return router.route((req) => + decodeEvent(decoder, req) + .chain(({ event, info }) => + withEventStore((store) => + handler({ + event, + info, + projections, + services, + store, + }).chainRej((r) => + r instanceof Ambar.Success + ? Future.resolve(undefined) + : r instanceof Ambar.ErrorMustRetry + ? Future.reject(r) + : (r satisfies never), + ), + ), + ) + .bimap(Ambar.toResponse, (_) => Ambar.toResponse(new Ambar.Success())), + ); +} diff --git a/src/app/projectionHandler.ts b/src/app/projectionHandler.ts deleted file mode 100644 index ec95c6f..0000000 --- a/src/app/projectionHandler.ts +++ /dev/null @@ -1,99 +0,0 @@ -export { handleProjection, decodeEvent, accept }; - -import { Response } from '@/lib/router'; -import { Event } from '@/lib/eventSourcing/event'; -import { Decoder, decode } from '@/lib/json/decoder'; -import * as express from 'express'; -import * as router from '@/lib/router'; -import * as d from '@/lib/json/decoder'; -import { Future } from '@/lib/Future'; -import { Result, Failure } from '@/lib/Result'; -import { Maybe, Nothing, Just } from '@/lib/Maybe'; -import { Schema } from '@/lib/json/schema'; -import * as s from '@/lib/json/schema'; - -type Projections = {}; -type ProjectionStore = {}; -type Mongo = {}; - -type ProjectionController> = { - decoder: Decoder>; - handler: (v: { - event: E; - projections: Projections; - store: ProjectionStore; - }) => Future; -}; - -function handleProjection>( - projections: Projections, - mongo: Mongo, - { decoder, handler }: ProjectionController, -): express.Handler { - return router.route((req) => - decodeEvent(decoder, req).chain((event) => - withProjectionStore(mongo, (store) => - handler({ - event, - projections, - store, - }), - ), - ), - ); -} - -function decodeEvent( - decoder: Decoder>, - req: express.Request, -): Future { - const bodyDecoder: Decoder> = d - .object({ payload: decoder }) - .map((r) => r.payload); - - const decoded: Result> = decode(bodyDecoder, req.body); - - if (decoded instanceof Failure) { - return Future.reject( - router.json({ - status: 400, - content: { message: `Unable to decode command: ${decoded.error}` }, - }), - ); - } - - if (decoded.value instanceof Nothing) { - return Future.reject( - router.json({ - status: 200, - content: { message: 'Ignored' }, - }), - ); - } - - return Future.resolve(decoded.value.value); -} - -function withProjectionStore( - _mongo: Mongo, - _f: (s: ProjectionStore) => Future, -): Future { - throw new Error('TODO'); -} - -type EventClass = { type: string; schema: Schema }; - -// Given some event classes, creates a decoder for those classes. -// Makes sure to error if decoding those class object fail, but -// succeeds if the encoded event was of another class. -function accept( - ts: T, -): Decoder>> { - type Ty = s.Infer; - return d.object({ type: d.string }).then(({ type: ty }) => { - const c: undefined | EventClass = ts.find((t) => t.type === ty); - return c - ? (c.schema.decoder.map(Just) as Decoder>) - : d.succeed(Nothing()); - }); -} diff --git a/src/app/projectionStore.ts b/src/app/projectionStore.ts new file mode 100644 index 0000000..bb53ad7 --- /dev/null +++ b/src/app/projectionStore.ts @@ -0,0 +1,149 @@ +export { + type Repository, // export only type here to prevent instantiation outside of module. + MongoProjectionStore, + Collection as Collection, + type JsonDoc, + type RepositoryArgs, + type WithProjectionStore, +}; + +import { Collection } from 'mongodb'; +import { Json } from '@/lib/json/types'; +import { Schema } from '@/lib/json/schema'; +import * as s from '@/lib/json/schema'; +import { MongoTransaction } from '@/lib/mongo'; +import { + Filter, + FindOptions, + Document, + InsertOneOptions, + WithId, +} from 'mongodb'; +import { Success, Failure } from '@/lib/Result'; +import { Future } from '@/lib/Future'; + +type JsonDoc = Exclude; + +type RepositoryArgs = { + collectionName: string; + createIndexes: (collection: Collection) => Promise; + schema: Schema; + toId: (v: T) => string; +}; + +// Represents an initialized collection. You can only get an +// instance of this class if the collection has been initialized. +class Repository { + constructor(public values: RepositoryArgs) {} +} + +type IdAndDoc = { _id: string; document: T }; + +const schemaIdAndValue = (schema: Schema): Schema> => { + const idSchema = s.object({ _id: s.string }).dimap( + ({ _id }) => _id, + (_id) => ({ _id }), + ); + + return s.both(idSchema, schema).dimap( + ([_id, document]) => ({ _id, document }), + ({ _id, document }) => [_id, document], + ); +}; + +type WithProjectionStore = ( + onError: (e: Error) => E, + f: (s: MongoProjectionStore) => Future, +) => Future; + +class MongoProjectionStore { + constructor(private transaction: MongoTransaction) {} + + // Initialize a repository, creating the collection and indexes if needed. + async createRepository(args: RepositoryArgs): Promise> { + console.log(`Initializing '${args.collectionName}'`); + + const db = this.transaction.database; + const collections = await db.listCollections().toArray(); + const names = collections.map((c) => c.name); + + if (names.includes(args.collectionName)) { + console.log(`Collection '${args.collectionName}' already exists`); + } else { + await db.createCollection(args.collectionName); + } + + const collection = db.collection(args.collectionName); + await args.createIndexes(collection); + console.log(`Indexes for '${args.collectionName}' created`); + + return new Repository(args); + } + + async findAny( + repository: Repository, + filter: Filter, + options?: FindOptions, + ): Promise[]> { + return this.transaction.find( + repository.values.collectionName, + filter, + options, + ); + } + + async find( + repository: Repository, + filter: Filter, + options?: FindOptions, + ): Promise { + const found = (await this.transaction.find( + repository.values.collectionName, + filter, + options, + )) as JsonDoc[]; + const schema = s.array(repository.values.schema); + const r = s.decode(schema, found); + switch (true) { + case r instanceof Success: + return r.value; + case r instanceof Failure: + throw new Error( + `Unable to decode: ${r.error}.\nIn ${JSON.stringify(found)}`, + ); + default: + return r satisfies never; + } + } + + // fails on _id clashes. + async insert( + repository: Repository, + document: T, + options?: InsertOneOptions, + ): Promise { + const schema = schemaIdAndValue(repository.values.schema); + const _id = repository.values.toId(document); + await this.transaction.insertOne( + repository.values.collectionName, + s.encode(schema, { _id, document }) as JsonDoc, + options, + ); + } + + // Upsert one value. Overwrites on _id clashes + async upsert( + repository: Repository, + document: T, + options: InsertOneOptions = {}, + ): Promise { + const schema = schemaIdAndValue(repository.values.schema); + const _id = repository.values.toId(document); + await this.transaction.replaceOne( + repository.values.collectionName, + { _id }, + s.encode(schema, { _id, document }) as JsonDoc, + Object.assign({ upsert: true }, options), + ); + } +} diff --git a/src/app/projections.ts b/src/app/projections.ts new file mode 100644 index 0000000..125a97f --- /dev/null +++ b/src/app/projections.ts @@ -0,0 +1,49 @@ +/* + List of all projections and repositories in the application. +*/ +export { + type Repositories, + type Projections, + initializeRepositories, + allProjections, +}; + +import { + RepoCuisine, + RepoMembershipApplication, +} from '@/domain/cookingClub/membership2/projection/membersByCuisine'; +import { MongoProjectionStore } from '@/app/projectionStore'; + +// An object containing all initialized repositories. +// Repository instances are used for writing into collections. +// An instance of this type can live for the lifetime of the application +// as it does not hold an active transaction. +type Repositories = Unwrap>; + +type Unwrap> = + A extends Promise ? B : never; + +async function initializeRepositories(mongo: MongoProjectionStore) { + return { + [RepoCuisine.collectionName]: await mongo.createRepository(RepoCuisine), + [RepoMembershipApplication.collectionName]: await mongo.createRepository( + RepoMembershipApplication, + ), + }; +} + +// An object containing all initialized projections. +type Projections = ReturnType; + +function allProjections(repos: Repositories, mongo: MongoProjectionStore) { + return { + [RepoCuisine.collectionName]: new RepoCuisine( + repos[RepoCuisine.collectionName], + mongo, + ), + [RepoMembershipApplication.collectionName]: new RepoMembershipApplication( + repos[RepoMembershipApplication.collectionName], + mongo, + ), + }; +} diff --git a/src/app/reactionHandler.ts b/src/app/reactionHandler.ts deleted file mode 100644 index 1791196..0000000 --- a/src/app/reactionHandler.ts +++ /dev/null @@ -1,44 +0,0 @@ -export { handleReaction }; - -import { Response } from '@/lib/router'; -import { EventStore } from '@/lib/eventSourcing/eventStore'; -import { Event } from '@/lib/eventSourcing/event'; -import { Decoder } from '@/lib/json/decoder'; -import * as express from 'express'; -import * as router from '@/lib/router'; -import { Future } from '@/lib/Future'; -import { Maybe } from '@/lib/Maybe'; -import { decodeEvent } from '@/app/projectionHandler'; - -type Projections = {}; -type Services = {}; - -type ReactionController> = { - decoder: Decoder>; - handler: (v: { - event: E; - projections: Projections; - services: Services; - store: EventStore; - }) => Future; -}; - -function handleReaction>( - withEventStore: (f: (s: EventStore) => T) => T, - projections: Projections, - services: Services, - { decoder, handler }: ReactionController, -): express.Handler { - return router.route((req) => - decodeEvent(decoder, req).chain((event) => - withEventStore((store) => - handler({ - event, - projections, - services, - store, - }), - ), - ), - ); -} diff --git a/src/app/responses.ts b/src/app/responses.ts new file mode 100644 index 0000000..2bd7203 --- /dev/null +++ b/src/app/responses.ts @@ -0,0 +1,25 @@ +export { internalServerError, forbidden, unauthorized, badRequest }; + +import { json } from '@/lib/router'; +import { Json } from '@/lib/json/types'; + +const internalServerError = json({ + status: 500, + content: { error: { message: 'Internal Server Error' } }, +}); + +const forbidden = json({ + status: 403, + content: { error: { message: 'Forbidden' } }, +}); + +const unauthorized = json({ + status: 401, + content: { error: { message: 'Unauthorized' } }, +}); + +const badRequest = (details: Json) => + json({ + status: 400, + content: { error: { message: 'Bad Request', details } }, + }); diff --git a/src/app/services.ts b/src/app/services.ts new file mode 100644 index 0000000..64fcff1 --- /dev/null +++ b/src/app/services.ts @@ -0,0 +1,6 @@ +/* + All application services +*/ +export { type Services }; + +type Services = {}; diff --git a/src/common/aggregate/Aggregate.md b/src/common/aggregate/Aggregate.md deleted file mode 100644 index c5290fd..0000000 --- a/src/common/aggregate/Aggregate.md +++ /dev/null @@ -1,5 +0,0 @@ -# Aggregate - -In Event Sourcing system, the Aggregate is an in-memory representation of the current state of the system based on past events. The process of taking events from the Event Store and instantiating an Aggregate from them is called Aggregate hydration or Aggregate reconstitution. - -An Aggregate is typically hydrated in a command handler, or a reaction handler, when appending new events to the system. Why? Because we want to check the current state of the system from Aggregates in an immediately consistent fashion. The Aggregate should be implemented in an immediately consistent fashion through the use of optimistic or pessimistic locking when reconstituting the Aggregate. diff --git a/src/common/aggregate/Aggregate.ts b/src/common/aggregate/Aggregate.ts deleted file mode 100644 index 2bbc811..0000000 --- a/src/common/aggregate/Aggregate.ts +++ /dev/null @@ -1,6 +0,0 @@ -export abstract class Aggregate { - protected constructor( - public readonly aggregateId: string, - public readonly aggregateVersion: number, - ) {} -} diff --git a/src/common/aggregate/index.ts b/src/common/aggregate/index.ts deleted file mode 100644 index 107ce18..0000000 --- a/src/common/aggregate/index.ts +++ /dev/null @@ -1 +0,0 @@ -export * from './Aggregate'; diff --git a/src/common/ambar/Ambar.md b/src/common/ambar/Ambar.md deleted file mode 100644 index 9b89f65..0000000 --- a/src/common/ambar/Ambar.md +++ /dev/null @@ -1,7 +0,0 @@ -# Ambar - -Tracking events in an EventStore for new events, filtering and forwarding them to downstream consumers while maintaining ordering and delivery guarantees can be complex and error-prone. Event buses such as RabbitMQ and Apache Kafka are often used to transmit and deliver events but can be complex to configure, manage, and scale. - -Ambar is a data streaming service that empowers you to build mission-critical real-time applications in minutes. Instead of producing to and consuming from message brokers, Ambar pulls records from databases, such as Event Stores and pushes records to application endpoints like your projection and reaction endpoints. - -Find out more about how to use Ambar in your production applications by visiting https://ambar.cloud/es diff --git a/src/common/ambar/AmbarAuthMiddleware.ts b/src/common/ambar/AmbarAuthMiddleware.ts deleted file mode 100644 index 0329d6b..0000000 --- a/src/common/ambar/AmbarAuthMiddleware.ts +++ /dev/null @@ -1,43 +0,0 @@ -import { Request, Response, NextFunction } from 'express'; -import env from '@/app/environment'; - -const VALID_USERNAME = env.AMBAR_HTTP_USERNAME; -const VALID_PASSWORD = env.AMBAR_HTTP_PASSWORD; - -if (!VALID_USERNAME || !VALID_PASSWORD) { - throw new Error( - 'Environment variables AUTH_USERNAME and AUTH_PASSWORD must be set', - ); -} - -export const AmbarAuthMiddleware = ( - req: Request, - res: Response, - next: NextFunction, -) => { - const authHeader = req.headers.authorization; - - if (!authHeader) { - return res.status(401).json({ error: 'Authentication required' }); - } - - if (!authHeader.startsWith('Basic ')) { - return res.status(401).json({ error: 'Basic authentication required' }); - } - - try { - const base64Credentials = authHeader.split(' ')[1] || ''; - const credentials = Buffer.from(base64Credentials, 'base64').toString( - 'utf8', - ); - const [username, password] = credentials.split(':'); - - if (username === VALID_USERNAME && password === VALID_PASSWORD) { - return next(); - } else { - return res.status(401).json({ error: 'Invalid credentials' }); - } - } catch (error) { - return res.status(401).json({ error: 'Invalid authentication format' }); - } -}; diff --git a/src/common/ambar/AmbarHttpRequest.ts b/src/common/ambar/AmbarHttpRequest.ts deleted file mode 100644 index 4d1c324..0000000 --- a/src/common/ambar/AmbarHttpRequest.ts +++ /dev/null @@ -1,9 +0,0 @@ -import { SerializedEvent } from '@/common/serializedEvent/SerializedEvent'; - -export interface AmbarHttpRequest { - data_source_id: string; - data_source_description: string; - data_destination_id: string; - data_destination_description: string; - payload: SerializedEvent; -} diff --git a/src/common/ambar/AmbarResponseFactory.ts b/src/common/ambar/AmbarResponseFactory.ts deleted file mode 100644 index 574365f..0000000 --- a/src/common/ambar/AmbarResponseFactory.ts +++ /dev/null @@ -1,10 +0,0 @@ -export class AmbarResponseFactory { - static retryResponse(exception: Error): string { - const message = exception.message.replace(/"/g, '\\"'); - return `{"result":{"error":{"policy":"must_retry","class":"${exception.constructor.name}","description":"message:${message}"}}}`; - } - - static successResponse(): string { - return '{"result":{"success":{}}}'; - } -} diff --git a/src/common/ambar/index.ts b/src/common/ambar/index.ts deleted file mode 100644 index 1f7c697..0000000 --- a/src/common/ambar/index.ts +++ /dev/null @@ -1,3 +0,0 @@ -export * from './AmbarAuthMiddleware'; -export * from './AmbarHttpRequest'; -export * from './AmbarResponseFactory'; diff --git a/src/common/command/Command.md b/src/common/command/Command.md deleted file mode 100644 index 54be5dd..0000000 --- a/src/common/command/Command.md +++ /dev/null @@ -1,7 +0,0 @@ -# Command Handler - -A Command Handler in an EventSourcing system is responsible for taking statements of intent (commands) from end users or other systems (both internal and external), performing validation, and upon valid conditions, adding new Events to the Event Store. - -To do this, the Command Handler reads past events from the Event store to hydrate / reconstitute an Aggregate. Once the aggregate is hydrated, the Command Handler checks for any business rules or constraints (e.g., ensuring an order hasn’t already been completed or that an account has sufficient balance). - -If all validations succeed, the Command Handler generates a new Event reflecting the state change requested by the command. This Event is then written back to the Event store, allowing the system to evolve while maintaining a full history of all changes. diff --git a/src/common/command/Command.ts b/src/common/command/Command.ts deleted file mode 100644 index 9074445..0000000 --- a/src/common/command/Command.ts +++ /dev/null @@ -1 +0,0 @@ -export abstract class Command {} diff --git a/src/common/command/CommandController.ts b/src/common/command/CommandController.ts deleted file mode 100644 index 5ba31e1..0000000 --- a/src/common/command/CommandController.ts +++ /dev/null @@ -1,35 +0,0 @@ -import { PostgresTransactionalEventStore } from '@/common/eventStore/PostgresTransactionalEventStore'; -import { MongoTransactionalProjectionOperator } from '@/common/projection/MongoTransactionalProjectionOperator'; -import { log } from '@/common/util/Logger'; -import { Command } from '@/common/command/Command'; -import { CommandHandler } from '@/common/command/CommandHandler'; - -export class CommandController { - constructor( - private readonly postgresTransactionalEventStore: PostgresTransactionalEventStore, - private readonly mongoTransactionalProjectionOperator: MongoTransactionalProjectionOperator, - ) {} - - protected async processCommand( - command: Command, - commandHandler: CommandHandler, - ): Promise { - try { - log.debug(`Starting to process command: ${command.constructor.name}`); - await this.postgresTransactionalEventStore.beginTransaction(); - await this.mongoTransactionalProjectionOperator.startTransaction(); - await commandHandler.handleCommand(command); - await this.postgresTransactionalEventStore.commitTransaction(); - await this.mongoTransactionalProjectionOperator.commitTransaction(); - - await this.postgresTransactionalEventStore.abortDanglingTransactionsAndReturnConnectionToPool(); - await this.mongoTransactionalProjectionOperator.abortDanglingTransactionsAndReturnSessionToPool(); - log.debug(`Successfully processed command: ${command.constructor.name}`); - } catch (error) { - await this.postgresTransactionalEventStore.abortDanglingTransactionsAndReturnConnectionToPool(); - await this.mongoTransactionalProjectionOperator.abortDanglingTransactionsAndReturnSessionToPool(); - log.error(`Exception in ProcessCommand: ${error}`, error as Error); - throw new Error(`Failed to process query: ${error}`); - } - } -} diff --git a/src/common/command/CommandHandler.ts b/src/common/command/CommandHandler.ts deleted file mode 100644 index 712c73d..0000000 --- a/src/common/command/CommandHandler.ts +++ /dev/null @@ -1,10 +0,0 @@ -import { Command } from '@/common/command/Command'; -import { PostgresTransactionalEventStore } from '@/common/eventStore/PostgresTransactionalEventStore'; - -export abstract class CommandHandler { - constructor( - protected readonly postgresTransactionalEventStore: PostgresTransactionalEventStore, - ) {} - - abstract handleCommand(command: Command): Promise; -} diff --git a/src/common/command/index.ts b/src/common/command/index.ts deleted file mode 100644 index fb64ee8..0000000 --- a/src/common/command/index.ts +++ /dev/null @@ -1,3 +0,0 @@ -export * from './Command'; -export * from './CommandController'; -export * from './CommandHandler'; diff --git a/src/common/event/CreationEvent.ts b/src/common/event/CreationEvent.ts deleted file mode 100644 index 552e517..0000000 --- a/src/common/event/CreationEvent.ts +++ /dev/null @@ -1,6 +0,0 @@ -import { Event } from '@/common/event/Event'; -import { Aggregate } from '@/common/aggregate/Aggregate'; - -export abstract class CreationEvent extends Event { - abstract createAggregate(): T; -} diff --git a/src/common/event/Event.md b/src/common/event/Event.md deleted file mode 100644 index 49d4f83..0000000 --- a/src/common/event/Event.md +++ /dev/null @@ -1,42 +0,0 @@ -# Event - -## What are Events? - -Events represent state changes that have occurred in the system. Instead of storing state, our system stores a series of events. Current state is derived by replaying these events in the order they occurred. - -Events are immutable, meaning once they are created and stored, they cannot be modified. They are a record of what happened in the system. - -An event typically contains: - -- Event Name: A description of the specific action that occurred (e.g., OrderPlaced, AccountDebited, UserSignedUp). -- Aggregate Identifier: The unique ID of the aggregate the event belongs in. -- Timestamp: The time when the Event occurred. -- Payload: Data describing the state change (the properties of the aggregate that have been changed). -- Metadata (optional): Information such as the user agent or IP of the end user. - -## Why use Events? - -Events are used to: - -- Rebuild the current state of an aggregate by replaying the series of events. -- Trigger side effects (reactions) such as sending notifications. -- Asynchronously update read models (projections). -- Provide an audit trail, capturing the full history of changes in the system for compliance and debugging. - -By relying on events as the source of truth, Event Sourcing allows for greater traceability, flexibility in replaying or restoring state, and the ability to respond to changes in a distributed, asynchronous manner. - -## Abstractions - -This directory contains our base definition for an Event. That is, `event_id`, `aggregate_id`, `aggregate_version`, `causation_id`, `correlation_id`, `recorded_on`. The event_name column, which is the name of the event, is not included because it's based on a mapping of the event class name to the event name. The `payload` column and `metadata` column are also not included because they are based on the event class properties. We use an abstraction called Serialized Event (see `src/main/java/cloud/ambar/common/serializedevent/SerializedEvent.java`) to store the `event_name`, `payload`, and `metadata`. - -**Why are there two extra abstract classes for creation events and transformation events?** - -Creation events are events that are used to create an aggregate. They are used to create the initial state of an aggregate. Transformation events are events that are used to transform an aggregate. They are used to change the state of an aggregate. - -When reconstituting / hydrating an aggregate, it's better not to have a default state of the aggregate which contains invalid state. Instead, it's better to codify into our type system which events can create a valid aggregate state on their own and which events can transform an aggregate. This way, we can ensure that the aggregate is always in a valid state. - -# Serialized Event - -A Serialized Event is a representation in which additional properties, not encoded in the abstract Event are converted into fields that go into the payload or metadata fields. Additionally, the Serialized Event contains an `event_name` which can be used to figure out which class the SerializedEvent should be deserialized into. - -Serialized Events are used when communicating with the database (Postgres) or event bus (Ambar). diff --git a/src/common/event/Event.ts b/src/common/event/Event.ts deleted file mode 100644 index ba0f5b3..0000000 --- a/src/common/event/Event.ts +++ /dev/null @@ -1,10 +0,0 @@ -export abstract class Event { - constructor( - public readonly eventId: string, - public readonly aggregateId: string, - public readonly aggregateVersion: number, - public readonly correlationId: string, - public readonly causationId: string, - public readonly recordedOn: Date, - ) {} -} diff --git a/src/common/event/TransformationEvent.ts b/src/common/event/TransformationEvent.ts deleted file mode 100644 index c789c26..0000000 --- a/src/common/event/TransformationEvent.ts +++ /dev/null @@ -1,6 +0,0 @@ -import { Event } from '@/common/event/Event'; -import { Aggregate } from '@/common/aggregate/Aggregate'; - -export abstract class TransformationEvent extends Event { - abstract transformAggregate(aggregate: T): T; -} diff --git a/src/common/event/index.ts b/src/common/event/index.ts deleted file mode 100644 index 6405b97..0000000 --- a/src/common/event/index.ts +++ /dev/null @@ -1,3 +0,0 @@ -export * from './CreationEvent'; -export * from './Event'; -export * from './TransformationEvent'; diff --git a/src/common/eventStore/AggregateAndEventIdsInLastEvent.ts b/src/common/eventStore/AggregateAndEventIdsInLastEvent.ts deleted file mode 100644 index 9ffb3a6..0000000 --- a/src/common/eventStore/AggregateAndEventIdsInLastEvent.ts +++ /dev/null @@ -1,7 +0,0 @@ -import { Aggregate } from '@/common/aggregate/Aggregate'; - -export interface AggregateAndEventIdsInLastEvent { - aggregate: T; - eventIdOfLastEvent: string; - correlationIdOfLastEvent: string; -} diff --git a/src/common/eventStore/EventStore.md b/src/common/eventStore/EventStore.md deleted file mode 100644 index 5a61cb3..0000000 --- a/src/common/eventStore/EventStore.md +++ /dev/null @@ -1,7 +0,0 @@ -# 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). diff --git a/src/common/eventStore/PostgresTransactionalEventStore.ts b/src/common/eventStore/PostgresTransactionalEventStore.ts deleted file mode 100644 index 98c239e..0000000 --- a/src/common/eventStore/PostgresTransactionalEventStore.ts +++ /dev/null @@ -1,247 +0,0 @@ -import { PoolClient } from 'pg'; -import { PostgresConnectionPool } from '@/common/util/PostgresConnectionPool'; -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 { log } from '@/common/util/Logger'; -import { AggregateAndEventIdsInLastEvent } from '@/common/eventStore/AggregateAndEventIdsInLastEvent'; -import { inject, injectable } from 'tsyringe'; - -@injectable() -export class PostgresTransactionalEventStore { - private connection: PoolClient | null = null; - private activeTransaction = false; - - constructor( - @inject(PostgresConnectionPool) - private readonly connectionPool: PostgresConnectionPool, - @inject(Serializer) private readonly serializer: Serializer, - @inject(Deserializer) private readonly deserializer: Deserializer, - @inject('eventStoreTable') private readonly eventStoreTable: string, - ) {} - - async beginTransaction(): Promise { - if (this.connection || this.activeTransaction) { - throw new Error('Connection or transaction already active!'); - } - - try { - this.connection = await this.connectionPool.openConnection(); - await this.connection.query('BEGIN ISOLATION LEVEL SERIALIZABLE'); - - this.activeTransaction = true; - } catch (error) { - const maxLen = 500; - const errorMessage = - error instanceof Error ? error.message : String(error); - throw new Error( - 'Failed to start transaction with ' + - (errorMessage.length > maxLen - ? errorMessage.substring(0, maxLen) - : errorMessage), - ); - } - } - - async findAggregate( - aggregateId: string, - ): Promise> { - if (!this.activeTransaction) { - throw new Error('Transaction must be active to perform operations!'); - } - - 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 { - if (!this.activeTransaction) { - throw new Error('Transaction must be active to perform operations!'); - } - - await this.saveSerializedEvent(this.serializer.serialize(event)); - } - - async doesEventAlreadyExist(eventId: string): Promise { - if (!this.activeTransaction) { - throw new Error('Transaction must be active to perform operations!'); - } - - const event = await this.findSerializedEventByEventId(eventId); - return event !== null; - } - - async commitTransaction(): Promise { - if (!this.activeTransaction) { - throw new Error('Transaction must be active to commit!'); - } - - try { - await this.connection?.query('COMMIT'); - this.activeTransaction = false; - } catch (error) { - throw new Error(`Failed to commit transaction: ${error}`); - } - } - - async abortDanglingTransactionsAndReturnConnectionToPool(): Promise { - if (this.activeTransaction) { - try { - await this.connection?.query('ROLLBACK'); - this.activeTransaction = false; - } catch (error) { - log.error('Failed to rollback PG transaction', error as Error); - } - } - - if (this.connection) { - try { - this.connection.release(); - this.connection = null; - } catch (error) { - log.error('Failed to release PG connection', error as Error); - } - } - } - - private async findAllSerializedEventsByAggregateId( - aggregateId: string, - ): Promise { - if (!this.connection) throw new Error('No active connection'); - - 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.connection.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 { - if (!this.connection) throw new Error('No active connection'); - - 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, - serializedEvent.json_payload, - serializedEvent.json_metadata, - serializedEvent.recorded_on, - serializedEvent.event_name, - ]; - - try { - await this.connection.query(sql, values); - } catch (error) { - throw new Error( - `Failed to save event: ${serializedEvent.event_id}: ${error}`, - ); - } - } - - private async findSerializedEventByEventId( - eventId: string, - ): Promise { - if (!this.connection) throw new Error('No active connection'); - - 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.connection.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; - } -} diff --git a/src/common/eventStore/index.ts b/src/common/eventStore/index.ts deleted file mode 100644 index 2268e67..0000000 --- a/src/common/eventStore/index.ts +++ /dev/null @@ -1,2 +0,0 @@ -export * from './AggregateAndEventIdsInLastEvent'; -export * from './PostgresTransactionalEventStore'; diff --git a/src/common/index.ts b/src/common/index.ts deleted file mode 100644 index 63b1582..0000000 --- a/src/common/index.ts +++ /dev/null @@ -1,12 +0,0 @@ -export * from './aggregate'; -export * from './ambar'; -export * from './command'; -export * from './event'; -export * from './eventStore'; -export * from './middleware'; -export * from './projection'; -export * from './query'; -export * from './reaction'; -export * from './serializedEvent'; -export * from './services'; -export * from './util'; diff --git a/src/common/middleware/ValidationPipe.ts b/src/common/middleware/ValidationPipe.ts deleted file mode 100644 index eba5dad..0000000 --- a/src/common/middleware/ValidationPipe.ts +++ /dev/null @@ -1,41 +0,0 @@ -import { Request } from 'express'; -import { plainToClass, ClassConstructor } from 'class-transformer'; -import { validate, ValidatorOptions } from 'class-validator'; -import { ValidationPipeException } from '@/common/middleware/ValidationPipeException'; - -export async function ValidationPipe( - targetClass: ClassConstructor, - req: Request, -): Promise { - try { - const dto = plainToClass(targetClass, req.body); - - const errors = await validate(dto, { - forbidNonWhitelisted: true, - forbidUnknownValues: true, - whitelist: true, - transform: true, - transformOptions: { - exposeDefaultValues: true, - }, - } as ValidatorOptions); - - if (errors.length > 0) { - const validationErrors = errors.map((error) => ({ - field: error.property, - constraints: Object.values(error.constraints || {}), - value: error.value, - })); - - throw ValidationPipeException.validationFailed(validationErrors); - } - - return dto; - } catch (error) { - if (error instanceof ValidationPipeException) { - throw error; - } - - throw ValidationPipeException.internalError(error as Error); - } -} diff --git a/src/common/middleware/ValidationPipeException.ts b/src/common/middleware/ValidationPipeException.ts deleted file mode 100644 index 3faa77a..0000000 --- a/src/common/middleware/ValidationPipeException.ts +++ /dev/null @@ -1,42 +0,0 @@ -export interface ValidationPipeError { - field: string; - constraints: string[]; - value: any; -} - -export class ValidationPipeException extends Error { - public readonly statusCode: number; - public readonly details: ValidationPipeError[]; - - constructor( - message: string, - statusCode: number, - details: ValidationPipeError[] = [], - ) { - super(message); - this.name = 'ValidationPipeException'; - this.statusCode = statusCode; - this.details = details; - - if (Error.captureStackTrace) { - Error.captureStackTrace(this, ValidationPipeException); - } - } - - static validationFailed( - errors: ValidationPipeError[], - ): ValidationPipeException { - return new ValidationPipeException('Validation failed', 400, errors); - } - - static internalError(originalError?: Error): ValidationPipeException { - const message = originalError?.message || 'Internal validation error'; - const exception = new ValidationPipeException(message, 500); - - if (originalError?.stack) { - exception.stack = originalError.stack; - } - - return exception; - } -} diff --git a/src/common/middleware/index.ts b/src/common/middleware/index.ts deleted file mode 100644 index d4b9aee..0000000 --- a/src/common/middleware/index.ts +++ /dev/null @@ -1,2 +0,0 @@ -export * from './ValidationPipe'; -export * from './ValidationPipeException'; diff --git a/src/common/projection/MongoTransactionalProjectionOperator.ts b/src/common/projection/MongoTransactionalProjectionOperator.ts deleted file mode 100644 index aca94d3..0000000 --- a/src/common/projection/MongoTransactionalProjectionOperator.ts +++ /dev/null @@ -1,160 +0,0 @@ -import { - ClientSession, - Filter, - FindOptions, - Document, - ReplaceOptions, - InsertOneOptions, - CountOptions, - Db, - ReadConcern, - WriteConcern, - ReadPreference, - TransactionOptions, - OptionalUnlessRequiredId, - WithId, -} from 'mongodb'; -import { MongoSessionPool } from '@/common/util/MongoSessionPool'; -import { log } from '@/common/util/Logger'; -import { inject, injectable } from 'tsyringe'; - -@injectable() -export class MongoTransactionalProjectionOperator { - private session: ClientSession | null = null; - private db: Db | null = null; - - constructor( - @inject(MongoSessionPool) private readonly sessionPool: MongoSessionPool, - @inject('mongoDatabaseName') private readonly databaseName: string, - ) {} - - async startTransaction(): Promise { - if (this.session) { - throw new Error('Session to MongoDB already active!'); - } - - if (this.db) { - throw new Error('Database already initialized in the current session.'); - } - - try { - this.session = await this.sessionPool.startSession(); - - const client = this.sessionPool.getClient(); - this.db = client.db(this.databaseName); - - const transactionOptions: TransactionOptions = { - readConcern: new ReadConcern('snapshot'), - writeConcern: new WriteConcern('majority'), - readPreference: ReadPreference.primary, - }; - - this.session.startTransaction(transactionOptions); - } catch (error) { - throw new Error(`Failed to start MongoDB transaction: ${error}`); - } - } - - async commitTransaction(): Promise { - if (!this.session) { - throw new Error( - 'Session must be active to commit transaction to MongoDB!', - ); - } - - if (!this.session.inTransaction()) { - throw new Error( - 'Transaction must be active to commit transaction to MongoDB!', - ); - } - - try { - await this.session.commitTransaction(); - } catch (error) { - throw new Error(`Failed to commit MongoDB transaction: ${error}`); - } - } - - async abortDanglingTransactionsAndReturnSessionToPool(): Promise { - if (!this.session) { - this.db = null; - return; - } - - try { - if (this.session.inTransaction()) { - await this.session.abortTransaction(); - } - } catch (error) { - log.error('Failed to abort Mongo transaction', error as Error); - } - - try { - await this.session.endSession(); - } catch (error) { - log.error('Failed to release Mongo session', error as Error); - } - - this.session = null; - this.db = null; - } - - async find( - collectionName: string, - filter: Filter, - options?: FindOptions, - ): Promise[]> { - const { session, db } = await this.operate(); - const collection = db.collection(collectionName); - return collection.find(filter, { ...options, session }).toArray(); - } - - async replaceOne( - collectionName: string, - filter: Filter, - replacement: T, - options?: ReplaceOptions, - ): Promise { - const { session, db } = await this.operate(); - const collection = db.collection(collectionName); - return collection.replaceOne(filter, replacement, { ...options, session }); - } - - async insertOne( - collectionName: string, - document: T & OptionalUnlessRequiredId, - options?: InsertOneOptions, - ): Promise { - const { session, db } = await this.operate(); - const collection = db.collection(collectionName); - await collection.insertOne(document, { ...options, session }); - } - - async countDocuments( - collectionName: string, - filter: Filter, - options?: CountOptions, - ): Promise { - const { session, db } = await this.operate(); - const collection = db.collection(collectionName); - return collection.countDocuments(filter, { ...options, session }); - } - - private async operate() { - if (!this.session) { - throw new Error('Session must be active to read or write to MongoDB!'); - } - - if (!this.session.inTransaction()) { - throw new Error( - 'Transaction must be active to read or write to MongoDB!', - ); - } - - if (!this.db) { - throw new Error('Database must be initialized in the current session.'); - } - - return { session: this.session, db: this.db }; - } -} diff --git a/src/common/projection/Projection.md b/src/common/projection/Projection.md deleted file mode 100644 index 6ef6d7e..0000000 --- a/src/common/projection/Projection.md +++ /dev/null @@ -1,29 +0,0 @@ -# Projection - -A projection is a read model that is derived from the events in the system. Projections are used to query the current state of the system. For example, in an ecommerce website users will need to know which items are available, before they add an item to their cart. This allows the read side of a system to often be decoupled from the write side. - -When an Event is emitted (e.g., OrderPlaced, ProductUpdated), a projection listens to the stream of events and filters the relevant events it needs to process. For example, a projection that builds a list of user orders would only listen for OrderPlaced and OrderCanceled events. It updates the read model by applying the Event data, ensuring the model reflects the latest state. - -Projections continuously update the projection database as new events arrive, keeping the read model in sync with the most recent state changes. This enables high-performance queries and ensures that the read side remains highly available and scalable. - -You can use projections for sharing state with your end users, but also to do validation in command handlers. But note that projections are built asynchronously, so they are eventually consistent. If you need to enforce business rules in an immediately consistent manner, you should do so by loading aggregates as opposed to reading projections. - -## How Projections Work - -Projections are built by listening to events and updating the read model accordingly. When an event is received, we update a projection database (MongoDB), based on the contents of the event and any existing data in the projection database. This behavior is captured by extending a `ProjectionHandler`. - -### How do events get sent from the Event Store to the Projection Handlers? - -We use Ambar to read events from the Event Store and send them to the Projection Handlers, via an HTTP endpoint. The HTTP endpoint is defined through extending a `ProjectionController`, which will receive the events and send them to the corresponding `ProjectionHandler`. - -#### How do we make sure that events are sent at least once, and in order per aggregate, to a Projection Handler? - -Ambar takes care of this out of the box. All you have to take care of is making every `ProjectionController` idempotent. To make sure projections endpoint only process events once, the `ProjectionController` uses an abstraction called `ProjectedEvent` which keeps track of every event that has already been processed. - -### Where can I find the Ambar configuration? - -The Ambar configuration is located in the `local-development/ambar-config.yml`. - -### In ambar-config.yml, why are events ordered per correlation id, instead of aggregate id? - -Ordering events per correlation id retains the order of events per aggregate, but also retains the order of events across related aggregates. E.g., if you have an aggregate for November, and an aggregate for December, and the aggregate for December directly follows the aggregate for November (using the same correlation id), Ambar will give you the events in order per aggregate, but will also retain order across aggregates (Ambar will project November first, and December second). diff --git a/src/common/projection/ProjectionController.ts b/src/common/projection/ProjectionController.ts deleted file mode 100644 index 63ed3f9..0000000 --- a/src/common/projection/ProjectionController.ts +++ /dev/null @@ -1,80 +0,0 @@ -import { MongoTransactionalProjectionOperator } from '@/common/projection/MongoTransactionalProjectionOperator'; -import { Deserializer } from '@/common/serializedEvent/Deserializer'; -import { AmbarHttpRequest } from '@/common/ambar/AmbarHttpRequest'; -import { AmbarResponseFactory } from '@/common/ambar/AmbarResponseFactory'; -import { ProjectionHandler } from '@/common/projection/ProjectionHandler'; -import { log } from '@/common/util/Logger'; - -export abstract class ProjectionController { - protected constructor( - private readonly mongoOperator: MongoTransactionalProjectionOperator, - private readonly deserializer: Deserializer, - ) {} - - protected async processProjectionHttpRequest( - ambarHttpRequest: AmbarHttpRequest, - projectionHandler: ProjectionHandler, - projectionName: string, - ): Promise { - try { - log.debug( - `Starting to process projection for event name: ${ambarHttpRequest.payload.event_name} using handler: ${projectionHandler.constructor.name}`, - ); - - const event = this.deserializer.deserialize(ambarHttpRequest.payload); - - await this.mongoOperator.startTransaction(); - - const isAlreadyProjected = - (await this.mongoOperator.countDocuments( - 'ProjectionIdempotency_ProjectedEvent', - { - eventId: event.eventId, - projectionName: projectionName, - }, - )) !== 0; - - if (isAlreadyProjected) { - await this.mongoOperator.abortDanglingTransactionsAndReturnSessionToPool(); - log.debug( - `Duplication projection ignored for event name: ${ambarHttpRequest.payload.event_name} using handler: ${projectionHandler.constructor.name}`, - ); - return AmbarResponseFactory.successResponse(); - } - - // Record projected event - await this.mongoOperator.insertOne( - 'ProjectionIdempotency_ProjectedEvent', - { - eventId: event.eventId, - projectionName: projectionName, - }, - ); - - await projectionHandler.project(event); - - await this.mongoOperator.commitTransaction(); - await this.mongoOperator.abortDanglingTransactionsAndReturnSessionToPool(); - - log.debug( - `Projection successfully processed for event name: ${ambarHttpRequest.payload.event_name} using handler: ${projectionHandler.constructor.name}`, - ); - return AmbarResponseFactory.successResponse(); - } catch (ex) { - if (ex instanceof Error && ex.message.startsWith('Unknown event type')) { - await this.mongoOperator.abortDanglingTransactionsAndReturnSessionToPool(); - - log.debug( - `Unknown event in projection ignored for event name: ${ambarHttpRequest.payload.event_name} using handler: ${projectionHandler.constructor.name}`, - ); - return AmbarResponseFactory.successResponse(); - } - - await this.mongoOperator.abortDanglingTransactionsAndReturnSessionToPool(); - log.error( - `Exception in ProcessProjectionHttpRequest: ${ex}. For event name: ${ambarHttpRequest.payload.event_name} using handler: ${projectionHandler.constructor.name}`, - ); - return AmbarResponseFactory.retryResponse(ex as Error); - } - } -} diff --git a/src/common/projection/ProjectionHandler.ts b/src/common/projection/ProjectionHandler.ts deleted file mode 100644 index 2fa5a9a..0000000 --- a/src/common/projection/ProjectionHandler.ts +++ /dev/null @@ -1,5 +0,0 @@ -import { Event } from '@/common/event/Event'; - -export abstract class ProjectionHandler { - public abstract project(event: Event): Promise; -} diff --git a/src/common/projection/index.ts b/src/common/projection/index.ts deleted file mode 100644 index bcb4fda..0000000 --- a/src/common/projection/index.ts +++ /dev/null @@ -1,3 +0,0 @@ -export * from './MongoTransactionalProjectionOperator'; -export * from './ProjectionController'; -export * from './ProjectionHandler'; diff --git a/src/common/query/Query.md b/src/common/query/Query.md deleted file mode 100644 index ffb2b22..0000000 --- a/src/common/query/Query.md +++ /dev/null @@ -1,11 +0,0 @@ -# Query Handler - -A Query Handler in an EventSourcing system is responsible for taking requests for information (queries) from end users or other systems (both internal and external), validating the query (e.g., checking if a user has the right permissions), and returning said information. - -To do this, Query Handlers will read state from read model / projection databases. Those databases are _filled_ up by Projections (see Projection directory). - -### Advantages: - -- Performance: Since queries access a read-optimized database, response times are faster and more efficient. -- Scalability: The query data storage (projection/read model databases) can be scaled separately from the Event Store used in Command Handlers. -- Flexibility: Different read models can be tailored for various use cases, offering specialized views for reporting, analytics, or specific user interfaces. diff --git a/src/common/query/Query.ts b/src/common/query/Query.ts deleted file mode 100644 index 1acca03..0000000 --- a/src/common/query/Query.ts +++ /dev/null @@ -1 +0,0 @@ -export abstract class Query {} diff --git a/src/common/query/QueryController.ts b/src/common/query/QueryController.ts deleted file mode 100644 index 4e2b82a..0000000 --- a/src/common/query/QueryController.ts +++ /dev/null @@ -1,30 +0,0 @@ -import { MongoTransactionalProjectionOperator } from '@/common/projection/MongoTransactionalProjectionOperator'; -import { log } from '@/common/util/Logger'; -import { QueryHandler } from '@/common/query/QueryHandler'; -import { Query } from '@/common/query/Query'; - -export class QueryController { - constructor( - private readonly mongoTransactionalProjectionOperator: MongoTransactionalProjectionOperator, - ) {} - - protected async processQuery( - query: Query, - queryHandler: QueryHandler, - ): Promise { - try { - log.debug(`Starting to process query: ${query.constructor.name}`); - await this.mongoTransactionalProjectionOperator.startTransaction(); - const result = await queryHandler.handleQuery(query); - await this.mongoTransactionalProjectionOperator.commitTransaction(); - await this.mongoTransactionalProjectionOperator.abortDanglingTransactionsAndReturnSessionToPool(); - - log.debug(`Successfully processed query: ${query.constructor.name}`); - return result; - } catch (error) { - await this.mongoTransactionalProjectionOperator.abortDanglingTransactionsAndReturnSessionToPool(); - log.error(`Exception in ProcessQuery: ${error}`, error as Error); - throw new Error(`Failed to process query: ${error}`); - } - } -} diff --git a/src/common/query/QueryHandler.ts b/src/common/query/QueryHandler.ts deleted file mode 100644 index de40e2c..0000000 --- a/src/common/query/QueryHandler.ts +++ /dev/null @@ -1,10 +0,0 @@ -import { MongoTransactionalProjectionOperator } from '@/common/projection/MongoTransactionalProjectionOperator'; -import { Query } from '@/common/query/Query'; - -export abstract class QueryHandler { - constructor( - protected readonly mongoTransactionalProjectionOperator: MongoTransactionalProjectionOperator, - ) {} - - abstract handleQuery(query: Query): Promise; -} diff --git a/src/common/query/index.ts b/src/common/query/index.ts deleted file mode 100644 index 21d5d1e..0000000 --- a/src/common/query/index.ts +++ /dev/null @@ -1,3 +0,0 @@ -export * from './Query'; -export * from './QueryController'; -export * from './QueryHandler'; diff --git a/src/common/reaction/Reaction.md b/src/common/reaction/Reaction.md deleted file mode 100644 index 0c7d615..0000000 --- a/src/common/reaction/Reaction.md +++ /dev/null @@ -1,23 +0,0 @@ -# Reaction - -A reaction performs a side effect in response to an Event. While projections update state, reactions may trigger actions like sending notifications, updating external systems, or initiating new workflows. Reactions also filter relevant events, allowing for targeted responses to specific state changes. Reactions are not only responsible for performing the side effect, but also for ensuring that the side effect is idempotent by writing the result of the side effect into the Event Store as an Event. - -## How Reactions Work - -Reactions are built by listening to events, triggering side effects, and recording the result of those side effects to the Event Store. This behavior is captured by extending a `ReactionHandler`. - -### How do events get sent from the Event Store to the Reaction Handlers? - -We use Ambar to read events from the Event Store and send them to the Reaction Handlers, via an HTTP endpoint. The HTTP endpoint is defined through extending a `ReactionController`, which will receive the events and send them to the corresponding `ReactionHandler`. - -#### How do we make sure that events are sent at least once, and in order per aggregate, to a Reaction Handler? - -Ambar takes care of this out of the box. All you have to take care of is making every `ReactionController` idempotent. To make sure reaction endpoint only process Events once, the `ReactionHandler` has to commit the results of its side effect into the Event Store with a new Event. This way, if the reaction is triggered again, it will be able to find an existing event in the Event Store. Note that the Reaction Event has to have a deterministic event id, so that we can check if the event has already been processed with the `ReactionHandler`. - -### Where can I find the Ambar configuration? - -The Ambar configuration is located in the `local-development/ambar-config.yml`. - -### In ambar-config.yml, why are events ordered per correlation id, instead of aggregate id? - -Ordering events per correlation id retains the order of events per aggregate, but also retains the order of events across related aggregates. E.g., if you have an aggregate for November, and an aggregate for December, and the aggregate for December directly follows the aggregate for November (using the same correlation id), Ambar will give you the events in order per aggregate, but will also retain order across aggregates (Ambar will send November first, and December second, so you can react in order). diff --git a/src/common/reaction/ReactionController.ts b/src/common/reaction/ReactionController.ts deleted file mode 100644 index cb17c1c..0000000 --- a/src/common/reaction/ReactionController.ts +++ /dev/null @@ -1,58 +0,0 @@ -import { AmbarHttpRequest } from '@/common/ambar/AmbarHttpRequest'; -import { AmbarResponseFactory } from '@/common/ambar/AmbarResponseFactory'; -import { PostgresTransactionalEventStore } from '@/common/eventStore/PostgresTransactionalEventStore'; -import { MongoTransactionalProjectionOperator } from '@/common/projection/MongoTransactionalProjectionOperator'; -import { Deserializer } from '@/common/serializedEvent/Deserializer'; -import { log } from '@/common/util/Logger'; -import { ReactionHandler } from '@/common/reaction/ReactionHandler'; - -export abstract class ReactionController { - constructor( - private readonly postgresTransactionalEventStore: PostgresTransactionalEventStore, - private readonly mongoTransactionalProjectionOperator: MongoTransactionalProjectionOperator, - private readonly deserializer: Deserializer, - ) {} - - protected async processReactionHttpRequest( - ambarHttpRequest: AmbarHttpRequest, - reactionHandler: ReactionHandler, - ): Promise { - try { - log.debug( - `Starting to process reaction for event name: ${ambarHttpRequest.payload.event_name} using handler: ${reactionHandler.constructor.name}`, - ); - await this.postgresTransactionalEventStore.beginTransaction(); - await this.mongoTransactionalProjectionOperator.startTransaction(); - await reactionHandler.react( - this.deserializer.deserialize(ambarHttpRequest.payload), - ); - await this.postgresTransactionalEventStore.commitTransaction(); - await this.mongoTransactionalProjectionOperator.commitTransaction(); - - await this.postgresTransactionalEventStore.abortDanglingTransactionsAndReturnConnectionToPool(); - await this.mongoTransactionalProjectionOperator.abortDanglingTransactionsAndReturnSessionToPool(); - - log.debug( - `Reaction successfully processed for event name: ${ambarHttpRequest.payload.event_name} using handler: ${reactionHandler.constructor.name}`, - ); - return AmbarResponseFactory.successResponse(); - } catch (error) { - if ( - error instanceof Error && - error.message.startsWith('Unknown event type') - ) { - await this.postgresTransactionalEventStore.abortDanglingTransactionsAndReturnConnectionToPool(); - await this.mongoTransactionalProjectionOperator.abortDanglingTransactionsAndReturnSessionToPool(); - log.debug( - `Unknown event in reaction ignored for event name: ${ambarHttpRequest.payload.event_name} using handler: ${reactionHandler.constructor.name}`, - ); - return AmbarResponseFactory.successResponse(); - } - - await this.postgresTransactionalEventStore.abortDanglingTransactionsAndReturnConnectionToPool(); - await this.mongoTransactionalProjectionOperator.abortDanglingTransactionsAndReturnSessionToPool(); - log.error('Exception in ProcessReactionHttpRequest:', error as Error); - return AmbarResponseFactory.retryResponse(error as Error); - } - } -} diff --git a/src/common/reaction/ReactionHandler.ts b/src/common/reaction/ReactionHandler.ts deleted file mode 100644 index 522f841..0000000 --- a/src/common/reaction/ReactionHandler.ts +++ /dev/null @@ -1,10 +0,0 @@ -import { Event } from '@/common/event/Event'; -import { PostgresTransactionalEventStore } from '@/common/eventStore/PostgresTransactionalEventStore'; - -export abstract class ReactionHandler { - constructor( - protected readonly postgresTransactionalEventStore: PostgresTransactionalEventStore, - ) {} - - abstract react(event: Event): Promise; -} diff --git a/src/common/reaction/index.ts b/src/common/reaction/index.ts deleted file mode 100644 index fc0b15c..0000000 --- a/src/common/reaction/index.ts +++ /dev/null @@ -1,2 +0,0 @@ -export * from './ReactionController'; -export * from './ReactionHandler'; diff --git a/src/common/serializedEvent/Deserializer.ts b/src/common/serializedEvent/Deserializer.ts deleted file mode 100644 index 151e99a..0000000 --- a/src/common/serializedEvent/Deserializer.ts +++ /dev/null @@ -1,95 +0,0 @@ -import { Event } from '@/common/event/Event'; -import { SerializedEvent } from '@/common/serializedEvent/SerializedEvent'; -import { injectable } from 'tsyringe'; -import { ApplicationEvaluated } from '@/domain/cookingClub/membership/event/ApplicationEvaluated'; -import { ApplicationSubmitted } from '@/domain/cookingClub/membership/event/ApplicationSubmitted'; -import { MembershipStatus } from '@/domain/cookingClub/membership/aggregate/membership'; - -@injectable() -export class Deserializer { - deserialize(serializedEvent: SerializedEvent): Event { - const recordedOn = this.parseDateTime(serializedEvent.recorded_on); - const payload = JSON.parse(serializedEvent.json_payload); - - switch (serializedEvent.event_name) { - case 'CookingClub_Membership_ApplicationSubmitted': - return new ApplicationSubmitted( - this.parseString(serializedEvent.event_id), - this.parseString(serializedEvent.aggregate_id), - this.parseNumber(serializedEvent.aggregate_version), - this.parseString(serializedEvent.correlation_id), - this.parseString(serializedEvent.causation_id), - recordedOn, - this.parseString(payload.firstName), - this.parseString(payload.lastName), - this.parseString(payload.favoriteCuisine), - this.parseNumber(payload.yearsOfProfessionalExperience), - this.parseNumber(payload.numberOfCookingBooksRead), - ); - - case 'CookingClub_Membership_ApplicationEvaluated': - return new ApplicationEvaluated( - this.parseString(serializedEvent.event_id), - this.parseString(serializedEvent.aggregate_id), - this.parseNumber(serializedEvent.aggregate_version), - this.parseString(serializedEvent.correlation_id), - this.parseString(serializedEvent.causation_id), - recordedOn, - this.parseEnum( - payload.evaluationOutcome, - MembershipStatus, - 'evaluationOutcome', - ), - ); - - default: - throw new Error(`Unknown event type: ${serializedEvent.event_name}`); - } - } - - private parseDateTime(dateStr: string): Date { - if (!dateStr.endsWith(' UTC')) { - throw new Error(`Invalid date format: ${dateStr}`); - } - const parsed = new Date(dateStr.slice(0, -4) + 'Z'); - if (isNaN(parsed.getTime())) { - throw new Error(`Invalid date format: ${dateStr}`); - } - return parsed; - } - - private parseString(value: any): string { - if (typeof value !== 'string') { - throw new Error(`Expected string but got ${typeof value}`); - } - return value; - } - - private parseNumber(value: any): number { - const parsed = Number(value); - if (isNaN(parsed)) { - throw new Error(`Expected number but got ${typeof value}`); - } - return parsed; - } - - private parseEnum( - value: any, - enumType: T, - fieldName: string, - ): T[keyof T] { - if (typeof value !== 'string') { - throw new Error( - `Expected string for ${fieldName} but got ${typeof value}`, - ); - } - - if (!Object.values(enumType).includes(value)) { - throw new Error( - `Invalid ${fieldName}: ${value}. Expected one of: ${Object.values(enumType).join(', ')}`, - ); - } - - return value as T[keyof T]; - } -} diff --git a/src/common/serializedEvent/SerializedEvent.ts b/src/common/serializedEvent/SerializedEvent.ts deleted file mode 100644 index 3f6bf9e..0000000 --- a/src/common/serializedEvent/SerializedEvent.ts +++ /dev/null @@ -1,12 +0,0 @@ -export interface SerializedEvent { - id?: number; - event_id: string; - aggregate_id: string; - causation_id: string; - correlation_id: string; - aggregate_version: number; - json_payload: string; - json_metadata: string; - recorded_on: string; - event_name: string; -} diff --git a/src/common/serializedEvent/Serializer.ts b/src/common/serializedEvent/Serializer.ts deleted file mode 100644 index aef2d60..0000000 --- a/src/common/serializedEvent/Serializer.ts +++ /dev/null @@ -1,53 +0,0 @@ -import { Event } from '@/common/event/Event'; -import { SerializedEvent } from '@/common/serializedEvent/SerializedEvent'; -import { injectable } from 'tsyringe'; -import { ApplicationSubmitted } from '@/domain/cookingClub/membership/event/ApplicationSubmitted'; -import { ApplicationEvaluated } from '@/domain/cookingClub/membership/event/ApplicationEvaluated'; - -@injectable() -export class Serializer { - serialize(event: Event): SerializedEvent { - return { - event_id: event.eventId, - aggregate_id: event.aggregateId, - aggregate_version: event.aggregateVersion, - correlation_id: event.correlationId, - causation_id: event.causationId, - recorded_on: this.formatDateTime(event.recordedOn), - event_name: this.determineEventName(event), - json_payload: this.createJsonPayload(event), - json_metadata: '{}', - }; - } - - private determineEventName(event: Event): string { - if (event instanceof ApplicationSubmitted) { - return 'CookingClub_Membership_ApplicationSubmitted'; - } - if (event instanceof ApplicationEvaluated) { - return 'CookingClub_Membership_ApplicationEvaluated'; - } - throw new Error(`Unknown event type: ${event.constructor.name}`); - } - - private createJsonPayload(event: Event): string { - const payload: Record = {}; - - if (event instanceof ApplicationSubmitted) { - payload['firstName'] = event.firstName; - payload['lastName'] = event.lastName; - payload['favoriteCuisine'] = event.favoriteCuisine; - payload['yearsOfProfessionalExperience'] = - event.yearsOfProfessionalExperience; - payload['numberOfCookingBooksRead'] = event.numberOfCookingBooksRead; - } else if (event instanceof ApplicationEvaluated) { - payload['evaluationOutcome'] = event.evaluationOutcome; - } - - return JSON.stringify(payload); - } - - private formatDateTime(date: Date): string { - return date.toISOString().replace('Z', ' UTC'); - } -} diff --git a/src/common/serializedEvent/index.ts b/src/common/serializedEvent/index.ts deleted file mode 100644 index 718fff1..0000000 --- a/src/common/serializedEvent/index.ts +++ /dev/null @@ -1,3 +0,0 @@ -export * from './Deserializer'; -export * from './SerializedEvent'; -export * from './Serializer'; diff --git a/src/common/services/email/EmailService.ts b/src/common/services/email.ts similarity index 100% rename from src/common/services/email/EmailService.ts rename to src/common/services/email.ts diff --git a/src/common/services/email/index.ts b/src/common/services/email/index.ts deleted file mode 100644 index 9843ada..0000000 --- a/src/common/services/email/index.ts +++ /dev/null @@ -1,2 +0,0 @@ -export { EmailService } from './EmailService'; -export type { EmailOptions, EmailServiceConfig } from './EmailService'; diff --git a/src/common/services/file-storage/FileStorageService.ts b/src/common/services/file-storage.ts similarity index 99% rename from src/common/services/file-storage/FileStorageService.ts rename to src/common/services/file-storage.ts index d427fb4..2f018a9 100644 --- a/src/common/services/file-storage/FileStorageService.ts +++ b/src/common/services/file-storage.ts @@ -1,6 +1,6 @@ import * as Minio from 'minio'; import { Readable } from 'stream'; -import { log } from '@/common/util'; +import { log } from '@/common/util/Logger'; import env from '@/app/environment'; export interface FileStorageOptions { diff --git a/src/common/services/file-storage/index.ts b/src/common/services/file-storage/index.ts deleted file mode 100644 index 8e58500..0000000 --- a/src/common/services/file-storage/index.ts +++ /dev/null @@ -1,7 +0,0 @@ -export { FileStorageService } from './FileStorageService'; -export type { - FileStorageOptions, - FileStorageServiceConfig, - FileStorageResult, - FileDownloadResult, -} from './FileStorageService'; diff --git a/src/common/services/index.ts b/src/common/services/index.ts deleted file mode 100644 index 98520d1..0000000 --- a/src/common/services/index.ts +++ /dev/null @@ -1,2 +0,0 @@ -export * from './email'; -export * from './file-storage'; diff --git a/src/common/util/IdGenerator.ts b/src/common/util/IdGenerator.ts deleted file mode 100644 index d0587ec..0000000 --- a/src/common/util/IdGenerator.ts +++ /dev/null @@ -1,45 +0,0 @@ -import { createHash, randomBytes } from 'crypto'; - -export class IdGenerator { - private static readonly ALPHANUMERIC_CHARACTERS = - '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'; - private static readonly ID_LENGTH = 56; - - static generateDeterministicId(seed: string): string { - if (!seed) { - throw new Error('Input string cannot be null or empty'); - } - - const firstHash = createHash('sha256').update(seed).digest(); - - const secondHash = createHash('sha256').update(firstHash).digest(); - - const combinedHash = Buffer.concat([firstHash, secondHash]); - - const base64Encoded = combinedHash.toString('base64'); - const cleanId = base64Encoded.replace(/[^A-Za-z0-9]/g, ''); - - return cleanId.substring(0, this.ID_LENGTH); - } - - static generateRandomId(): string { - const chars = new Array(this.ID_LENGTH); - - for (let i = 0; i < this.ID_LENGTH; i++) { - const byte = randomByte(); - chars[i] = this.ALPHANUMERIC_CHARACTERS.charAt( - byte % this.ALPHANUMERIC_CHARACTERS.length, - ); - } - - return chars.join(''); - } -} - -function randomByte(): number { - const byte = randomBytes(1)[0]; - if (byte == undefined) { - throw new Error('No byte returned by randomBytes'); - } - return byte; -} diff --git a/src/common/util/MongoInitializer.ts b/src/common/util/MongoInitializer.ts deleted file mode 100644 index a121d2e..0000000 --- a/src/common/util/MongoInitializer.ts +++ /dev/null @@ -1,99 +0,0 @@ -import { injectable, inject } from 'tsyringe'; -import { MongoClient } from 'mongodb'; -import { MongoSessionPool } from '@/common/util/MongoSessionPool'; -import { log } from '@/common/util/Logger'; - -@injectable() -export class MongoInitializer { - private readonly client: MongoClient; - - constructor( - @inject(MongoSessionPool) private readonly sessionPool: MongoSessionPool, - @inject('mongoDatabaseName') private readonly databaseName: string, - ) { - this.client = this.sessionPool.getClient(); - } - - async initialize(): Promise { - log.info('Initializing MongoDB collections and indexes...'); - - try { - await this.client.connect(); - const db = this.client.db(this.databaseName); - - // Create collections - log.info('Creating collections...'); - await Promise.all([ - this.ensureCollection( - db, - 'CookingClub_MembersByCuisine_MembershipApplication', - ), - this.ensureCollection(db, 'CookingClub_MembersByCuisine_Cuisine'), - this.ensureCollection(db, 'ProjectionIdempotency_ProjectedEvent'), - ]); - log.info('Collections created successfully'); - - // Create indexes - log.info('Creating indexes...'); - await this.createIndexes(db); - log.info('Indexes created successfully'); - } catch (error) { - log.error('Error initializing MongoDB:', error as Error); - throw error; - } - } - - private async ensureCollection( - db: any, - collectionName: string, - ): Promise { - try { - const collections = await db - .listCollections({ name: collectionName }) - .toArray(); - if (collections.length === 0) { - await db.createCollection(collectionName); - log.debug(`Collection ${collectionName} created`); - } else { - log.debug(`Collection ${collectionName} already exists`); - } - } catch (error) { - log.error(`Error ensuring collection ${collectionName}:`, error as Error); - throw error; - } - } - - private async createIndexes(db: any): Promise { - try { - const membershipApplicationCollection = db.collection( - 'CookingClub_MembersByCuisine_MembershipApplication', - ); - - await membershipApplicationCollection.createIndex( - { favoriteCuisine: 1 }, - { - background: true, - name: 'favoriteCuisine_asc', - }, - ); - - const projectionIdempotencyCollection = db.collection( - 'ProjectionIdempotency_ProjectedEvent', - ); - - await projectionIdempotencyCollection.createIndex( - { eventId: 1, projectionName: 1 }, - { - unique: true, - background: true, - name: 'eventId_ProjectionName_unique', - }, - ); - - log.debug('Indexes created'); - } catch (error) { - log.error('Error creating indexes:', error as Error); - throw error; - } - } -} diff --git a/src/common/util/MongoSessionPool.ts b/src/common/util/MongoSessionPool.ts deleted file mode 100644 index c81efc9..0000000 --- a/src/common/util/MongoSessionPool.ts +++ /dev/null @@ -1,34 +0,0 @@ -import { MongoClient, ClientSession, ServerApiVersion } from 'mongodb'; -import { inject, injectable } from 'tsyringe'; - -@injectable() -export class MongoSessionPool { - private readonly transactionalClient: MongoClient; - - constructor(@inject('mongoConnectionString') connectionString: string) { - const settings = { - maxPoolSize: 20, - minPoolSize: 5, - maxIdleTimeMS: 10 * 60 * 1000, // 10 minutes - maxConnecting: 30, - waitQueueTimeoutMS: 2000, - replicaSet: 'rs0', - serverApi: { - version: ServerApiVersion.v1, - strict: true, - deprecationErrors: true, - }, - }; - - this.transactionalClient = new MongoClient(connectionString, settings); - } - - async startSession(): Promise { - await this.transactionalClient.connect(); - return this.transactionalClient.startSession(); - } - - getClient(): MongoClient { - return this.transactionalClient; - } -} diff --git a/src/common/util/PostgresConnectionPool.ts b/src/common/util/PostgresConnectionPool.ts deleted file mode 100644 index 15ed362..0000000 --- a/src/common/util/PostgresConnectionPool.ts +++ /dev/null @@ -1,35 +0,0 @@ -import { Pool, PoolConfig, PoolClient } from 'pg'; -import { inject, injectable } from 'tsyringe'; - -@injectable() -export class PostgresConnectionPool { - private readonly pool: Pool; - - constructor(@inject('postgresConnectionString') connectionString: string) { - 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); - }); - } - - async openConnection(): Promise { - try { - return await this.pool.connect(); - } catch (error) { - throw new Error(`Failed to open database connection: ${error}`); - } - } - - async close(): Promise { - await this.pool.end(); - } -} diff --git a/src/common/util/PostgresInitializer.ts b/src/common/util/PostgresInitializer.ts deleted file mode 100644 index 9f5eb37..0000000 --- a/src/common/util/PostgresInitializer.ts +++ /dev/null @@ -1,128 +0,0 @@ -import { injectable, inject } from 'tsyringe'; -import { PostgresConnectionPool } from '@/common/util/PostgresConnectionPool'; -import { log } from '@/common/util/Logger'; - -@injectable() -export class PostgresInitializer { - constructor( - @inject(PostgresConnectionPool) - private readonly connectionPool: PostgresConnectionPool, - @inject('eventStoreDatabaseName') - private readonly eventStoreDatabaseName: string, - @inject('eventStoreTable') private readonly eventStoreTable: string, - @inject('eventStoreCreateReplicationUserWithUsername') - private readonly replicationUsername: string, - @inject('eventStoreCreateReplicationUserWithPassword') - private readonly replicationPassword: string, - @inject('eventStoreCreateReplicationPublication') - private readonly replicationPublication: string, - ) {} - - async initialize(): Promise { - const client = await this.connectionPool.openConnection(); - - try { - // Create table - log.info(`Creating table ${this.eventStoreTable}`); - await this.executeStatementIgnoreErrors( - client, - ` - CREATE TABLE IF NOT EXISTS ${this.eventStoreTable} ( - 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) - ); - `, - ); - - // Create replication user - log.info('Creating replication user'); - await this.executeStatementIgnoreErrors( - client, - `CREATE USER ${this.replicationUsername} REPLICATION LOGIN PASSWORD '${this.replicationPassword}';`, - ); - - // Grant permissions to user - log.info('Granting permissions to replication user'); - await this.executeStatementIgnoreErrors( - client, - `GRANT CONNECT ON DATABASE "${this.eventStoreDatabaseName}" TO ${this.replicationUsername};`, - ); - - log.info('Granting select to replication user'); - await this.executeStatementIgnoreErrors( - client, - `GRANT SELECT ON TABLE ${this.eventStoreTable} TO ${this.replicationUsername};`, - ); - - // Create publication - log.info('Creating publication for table'); - await this.executeStatementIgnoreErrors( - client, - `CREATE PUBLICATION ${this.replicationPublication} FOR TABLE ${this.eventStoreTable};`, - ); - - // Create indexes - log.info('Creating aggregate id, aggregate version index'); - await this.executeStatementIgnoreErrors( - client, - `CREATE UNIQUE INDEX event_store_idx_event_aggregate_id_version ON ${this.eventStoreTable}(aggregate_id, aggregate_version);`, - ); - - log.info('Creating id index'); - await this.executeStatementIgnoreErrors( - client, - `CREATE UNIQUE INDEX event_store_idx_event_id ON ${this.eventStoreTable}(event_id);`, - ); - - log.info('Creating causation index'); - await this.executeStatementIgnoreErrors( - client, - `CREATE INDEX event_store_idx_event_causation_id ON ${this.eventStoreTable}(causation_id);`, - ); - - log.info('Creating correlation index'); - await this.executeStatementIgnoreErrors( - client, - `CREATE INDEX event_store_idx_event_correlation_id ON ${this.eventStoreTable}(correlation_id);`, - ); - - log.info('Creating recording index'); - await this.executeStatementIgnoreErrors( - client, - `CREATE INDEX event_store_idx_occurred_on ON ${this.eventStoreTable}(recorded_on);`, - ); - - log.info('Creating event name index'); - await this.executeStatementIgnoreErrors( - client, - `CREATE INDEX event_store_idx_event_name ON ${this.eventStoreTable}(event_name);`, - ); - } finally { - client.release(); - } - } - - private async executeStatementIgnoreErrors( - client: any, - sqlStatement: string, - ): Promise { - try { - log.info(`Executing SQL: ${sqlStatement}`); - await client.query(sqlStatement); - } catch (error) { - log.warn( - 'Caught exception when executing SQL statement.', - error as Error, - ); - } - } -} diff --git a/src/common/util/index.ts b/src/common/util/index.ts deleted file mode 100644 index f46136e..0000000 --- a/src/common/util/index.ts +++ /dev/null @@ -1,6 +0,0 @@ -export * from './IdGenerator'; -export * from './Logger'; -export * from './MongoInitializer'; -export * from './MongoSessionPool'; -export * from './PostgresConnectionPool'; -export * from './PostgresInitializer'; diff --git a/src/di/container.ts b/src/di/container.ts index e8f90b2..9bd6bec 100644 --- a/src/di/container.ts +++ b/src/di/container.ts @@ -1,29 +1,19 @@ -import { container, Lifecycle } from 'tsyringe'; -import { - Serializer, - Deserializer, - PostgresConnectionPool, - MongoSessionPool, - PostgresTransactionalEventStore, - MongoTransactionalProjectionOperator, - MongoInitializer, - PostgresInitializer, - EmailService, - FileStorageService, -} from '@/common'; -import { constructor } from 'tsyringe/dist/typings/types'; -import { SubmitApplicationCommandController } from '@/domain/cookingClub/membership/command/submitApplication'; -import { SubmitApplicationCommandHandler } from '@/domain/cookingClub/membership/command/submitApplication'; -import { EvaluateApplicationReactionHandler } from '@/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionHandler'; -import { EvaluateApplicationReactionController } from '@/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionController'; -import { MembersByCuisineProjectionHandler } from '@/domain/cookingClub/membership/projection/membersByCuisine/MembersByCuisineProjectionHandler'; -import { MembershipApplicationRepository } from '@/domain/cookingClub/membership/projection/membersByCuisine/MembershipApplicationRepository'; -import { CuisineRepository } from '@/domain/cookingClub/membership/projection/membersByCuisine/CuisineRepository'; +import { container } from 'tsyringe'; +import { EmailService } from '@/common/services/email'; +import { FileStorageService } from '@/common/services/file-storage'; import env from '@/app/environment'; import { Postgres, defaultPoolSettings } from '@/lib/postgres'; import { Mongo } from '@/lib/mongo'; +import { + MongoProjectionStore, + WithProjectionStore, +} from '@/app/projectionStore'; import { ServerApiVersion } from 'mongodb'; -import * as postgresEventStore from '@/app/postgresEventStore'; +import * as eventStore from '@/app/eventStore'; +import { PostgresEventStore, WithEventStore } from '@/app/eventStore'; +import { schemas } from '@/app/events'; +import { Services } from '@/app/services'; +import { Repositories, initializeRepositories } from '@/app/projections'; function registerEnvironmentVariables() { const postgresConnectionString = @@ -62,49 +52,18 @@ function registerEnvironmentVariables() { } function registerSingletons() { - // common/serializedEvent - container.registerSingleton(Serializer); - container.registerSingleton(Deserializer); - - // common/util - container.registerSingleton(PostgresConnectionPool); - container.registerSingleton(MongoSessionPool); - container.registerSingleton(MongoInitializer); - container.registerSingleton(PostgresInitializer); - // common/services container.registerSingleton(EmailService); container.registerSingleton(FileStorageService); } -function registerScoped(token: constructor) { - container.register(token, token, { lifecycle: Lifecycle.ContainerScoped }); -} - -function registerScopedServices() { - // common/eventStore - registerScoped(PostgresTransactionalEventStore); - - // common/projection - registerScoped(MongoTransactionalProjectionOperator); - - // domain/cookingClub/command/submitApplication - registerScoped(SubmitApplicationCommandController); - registerScoped(SubmitApplicationCommandHandler); - - // domain/cookingClub/projection/membersByCuisine - registerScoped(CuisineRepository); - registerScoped(MembersByCuisineProjectionHandler); - registerScoped(MembershipApplicationRepository); - - // domain/cookingClub/reaction/evaluateApplication - registerScoped(EvaluateApplicationReactionController); - registerScoped(EvaluateApplicationReactionHandler); -} +function registerScopedServices() {} type Dependencies = { - postgres: Postgres; - mongo: Mongo; + withEventStore: WithEventStore; + withProjectionStore: WithProjectionStore; + services: Services; + repositories: Repositories; }; export async function configureDependencies(): Promise { @@ -142,11 +101,12 @@ export async function configureDependencies(): Promise { }, }); + const table = env.EVENT_STORE_CREATE_TABLE_WITH_NAME; await postgres.withTransactionP((transaction) => - postgresEventStore.initialize({ + eventStore.initialize({ transaction, database: env.EVENT_STORE_DATABASE_NAME, - table: env.EVENT_STORE_CREATE_TABLE_WITH_NAME, + table, replicationUserName: env.EVENT_STORE_CREATE_REPLICATION_USER_WITH_USERNAME, replicationUserPass: @@ -155,5 +115,22 @@ export async function configureDependencies(): Promise { }), ); - return { postgres, mongo }; + const withEventStore: WithEventStore = (onError, f) => + postgres.withTransaction(onError, (t) => + f(new PostgresEventStore(t, schemas, table)), + ); + + const withProjectionStore: WithProjectionStore = (onError, f) => + mongo.withTransaction(onError, (t) => f(new MongoProjectionStore(t))); + + const repositories = await mongo.withTransactionP(async (t) => + initializeRepositories(new MongoProjectionStore(t)), + ); + + return { + withEventStore, + withProjectionStore, + services: {}, + repositories, + }; } diff --git a/src/domain/cookingClub/membership/aggregate/membership.ts b/src/domain/cookingClub/membership/aggregate/membership.ts deleted file mode 100644 index f8978dd..0000000 --- a/src/domain/cookingClub/membership/aggregate/membership.ts +++ /dev/null @@ -1,19 +0,0 @@ -import { Aggregate } from '@/common'; - -export enum MembershipStatus { - Requested = 'Requested', - Approved = 'Approved', - Rejected = 'Rejected', -} - -export class Membership extends Aggregate { - constructor( - aggregateId: string, - aggregateVersion: number, - public readonly firstName: string, - public readonly lastName: string, - public readonly status: MembershipStatus, - ) { - super(aggregateId, aggregateVersion); - } -} diff --git a/src/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommand.ts b/src/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommand.ts deleted file mode 100644 index aea7c30..0000000 --- a/src/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommand.ts +++ /dev/null @@ -1,34 +0,0 @@ -import { IsNumber, IsString } from 'class-validator'; -import { Command } from '@/common'; - -export class SubmitApplicationCommand extends Command { - @IsString() - public readonly firstName: string; - - @IsString() - public readonly lastName: string; - - @IsString() - public readonly favoriteCuisine: string; - - @IsNumber() - public readonly yearsOfProfessionalExperience: number; - - @IsNumber() - public readonly numberOfCookingBooksRead: number; - - constructor( - firstName: string, - lastName: string, - favoriteCuisine: string, - yearsOfProfessionalExperience: number, - numberOfCookingBooksRead: number, - ) { - super(); - this.firstName = firstName; - this.lastName = lastName; - this.favoriteCuisine = favoriteCuisine; - this.yearsOfProfessionalExperience = yearsOfProfessionalExperience; - this.numberOfCookingBooksRead = numberOfCookingBooksRead; - } -} diff --git a/src/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommandController.ts b/src/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommandController.ts deleted file mode 100644 index fd5a5ab..0000000 --- a/src/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommandController.ts +++ /dev/null @@ -1,43 +0,0 @@ -import { Router, Request, Response } from 'express'; -import { - CommandController, - PostgresTransactionalEventStore, - MongoTransactionalProjectionOperator, - ValidationPipe, -} from '@/common'; -import { inject, injectable } from 'tsyringe'; -import { SubmitApplicationCommandHandler } from '@/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommandHandler'; -import { SubmitApplicationCommand } from '@/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommand'; - -@injectable() -export class SubmitApplicationCommandController extends CommandController { - public readonly router: Router; - - private readonly submitApplicationCommandHandler: SubmitApplicationCommandHandler; - - constructor( - @inject(PostgresTransactionalEventStore) - postgresTransactionalEventStore: PostgresTransactionalEventStore, - @inject(MongoTransactionalProjectionOperator) - mongoTransactionalProjectionOperator: MongoTransactionalProjectionOperator, - @inject(SubmitApplicationCommandHandler) - submitApplicationCommandHandler: SubmitApplicationCommandHandler, - ) { - super( - postgresTransactionalEventStore, - mongoTransactionalProjectionOperator, - ); - this.submitApplicationCommandHandler = submitApplicationCommandHandler; - this.router = Router(); - - //TODO: abstract this next - this.router.post('/submit-application', this.submitApplication.bind(this)); - } - - async submitApplication(req: Request, res: Response): Promise { - const command = await ValidationPipe(SubmitApplicationCommand, req); - - await this.processCommand(command, this.submitApplicationCommandHandler); - res.status(200).json({}); - } -} diff --git a/src/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommandHandler.ts b/src/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommandHandler.ts deleted file mode 100644 index 2eaf417..0000000 --- a/src/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommandHandler.ts +++ /dev/null @@ -1,39 +0,0 @@ -import { - CommandHandler, - PostgresTransactionalEventStore, - IdGenerator, -} from '@/common'; -import { inject, injectable } from 'tsyringe'; -import { ApplicationSubmitted } from '@/domain/cookingClub/membership/event/ApplicationSubmitted'; -import { SubmitApplicationCommand } from '@/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommand'; - -@injectable() -export class SubmitApplicationCommandHandler extends CommandHandler { - constructor( - @inject(PostgresTransactionalEventStore) - postgresTransactionalEventStore: PostgresTransactionalEventStore, - ) { - super(postgresTransactionalEventStore); - } - - async handleCommand(command: SubmitApplicationCommand): Promise { - const eventId = IdGenerator.generateRandomId(); - const aggregateId = IdGenerator.generateRandomId(); - - const applicationSubmitted = new ApplicationSubmitted( - eventId, - aggregateId, - 1, - eventId, - eventId, - new Date(), - command.firstName, - command.lastName, - command.favoriteCuisine, - command.yearsOfProfessionalExperience, - command.numberOfCookingBooksRead, - ); - - await this.postgresTransactionalEventStore.saveEvent(applicationSubmitted); - } -} diff --git a/src/domain/cookingClub/membership/command/submitApplication/index.ts b/src/domain/cookingClub/membership/command/submitApplication/index.ts deleted file mode 100644 index ccbbc83..0000000 --- a/src/domain/cookingClub/membership/command/submitApplication/index.ts +++ /dev/null @@ -1,3 +0,0 @@ -export * from './SubmitApplicationCommand'; -export * from './SubmitApplicationCommandHandler'; -export * from './SubmitApplicationCommandController'; diff --git a/src/domain/cookingClub/membership/event/ApplicationEvaluated.ts b/src/domain/cookingClub/membership/event/ApplicationEvaluated.ts deleted file mode 100644 index 74d20ba..0000000 --- a/src/domain/cookingClub/membership/event/ApplicationEvaluated.ts +++ /dev/null @@ -1,36 +0,0 @@ -import { TransformationEvent } from '@/common'; -import { - Membership, - MembershipStatus, -} from '@/domain/cookingClub/membership/aggregate/membership'; - -export class ApplicationEvaluated extends TransformationEvent { - constructor( - eventId: string, - aggregateId: string, - aggregateVersion: number, - correlationId: string, - causationId: string, - recordedOn: Date, - public readonly evaluationOutcome: MembershipStatus, - ) { - super( - eventId, - aggregateId, - aggregateVersion, - correlationId, - causationId, - recordedOn, - ); - } - - transformAggregate(aggregate: Membership): Membership { - return new Membership( - this.aggregateId, - this.aggregateVersion, - aggregate.firstName, - aggregate.lastName, - this.evaluationOutcome, - ); - } -} diff --git a/src/domain/cookingClub/membership/event/ApplicationSubmitted.ts b/src/domain/cookingClub/membership/event/ApplicationSubmitted.ts deleted file mode 100644 index 1a78ab1..0000000 --- a/src/domain/cookingClub/membership/event/ApplicationSubmitted.ts +++ /dev/null @@ -1,40 +0,0 @@ -import { CreationEvent } from '@/common'; -import { - Membership, - MembershipStatus, -} from '@/domain/cookingClub/membership/aggregate/membership'; - -export class ApplicationSubmitted extends CreationEvent { - constructor( - eventId: string, - aggregateId: string, - aggregateVersion: number, - correlationId: string, - causationId: string, - recordedOn: Date, - public readonly firstName: string, - public readonly lastName: string, - public readonly favoriteCuisine: string, - public readonly yearsOfProfessionalExperience: number, - public readonly numberOfCookingBooksRead: number, - ) { - super( - eventId, - aggregateId, - aggregateVersion, - correlationId, - causationId, - recordedOn, - ); - } - - createAggregate(): Membership { - return new Membership( - this.aggregateId, - this.aggregateVersion, - this.firstName, - this.lastName, - MembershipStatus.Requested, - ); - } -} diff --git a/src/domain/cookingClub/membership/projection/membersByCuisine/Cuisine.ts b/src/domain/cookingClub/membership/projection/membersByCuisine/Cuisine.ts deleted file mode 100644 index 491ac3d..0000000 --- a/src/domain/cookingClub/membership/projection/membersByCuisine/Cuisine.ts +++ /dev/null @@ -1,6 +0,0 @@ -export class Cuisine { - constructor( - public readonly _id: string, // needs to be _id to be recognized as an _id field by MongoDB - public readonly memberNames: string[], - ) {} -} diff --git a/src/domain/cookingClub/membership/projection/membersByCuisine/CuisineRepository.ts b/src/domain/cookingClub/membership/projection/membersByCuisine/CuisineRepository.ts deleted file mode 100644 index 2364bbb..0000000 --- a/src/domain/cookingClub/membership/projection/membersByCuisine/CuisineRepository.ts +++ /dev/null @@ -1,34 +0,0 @@ -import { MongoTransactionalProjectionOperator } from '@/common'; -import { inject, injectable } from 'tsyringe'; -import { Cuisine } from '@/domain/cookingClub/membership/projection/membersByCuisine/Cuisine'; - -@injectable() -export class CuisineRepository { - private readonly collectionName = 'CookingClub_MembersByCuisine_Cuisine'; - - constructor( - @inject(MongoTransactionalProjectionOperator) - private readonly mongoOperator: MongoTransactionalProjectionOperator, - ) {} - - async save(cuisine: Cuisine): Promise { - await this.mongoOperator.replaceOne( - this.collectionName, - { _id: cuisine._id }, - cuisine, - { upsert: true }, - ); - } - - async findOneById(_id: string): Promise { - const results = await this.mongoOperator.find( - this.collectionName, - { _id }, - ); - return results[0] || null; - } - - async findAll(): Promise { - return this.mongoOperator.find(this.collectionName, {}); - } -} diff --git a/src/domain/cookingClub/membership/projection/membersByCuisine/MembersByCuisineProjectionController.ts b/src/domain/cookingClub/membership/projection/membersByCuisine/MembersByCuisineProjectionController.ts deleted file mode 100644 index 4759faa..0000000 --- a/src/domain/cookingClub/membership/projection/membersByCuisine/MembersByCuisineProjectionController.ts +++ /dev/null @@ -1,41 +0,0 @@ -import { Request, Response, Router } from 'express'; -import { - MongoTransactionalProjectionOperator, - Deserializer, - AmbarHttpRequest, - ProjectionController, -} from '@/common'; -import { inject, injectable } from 'tsyringe'; -import { MembersByCuisineProjectionHandler } from '@/domain/cookingClub/membership/projection/membersByCuisine/MembersByCuisineProjectionHandler'; - -@injectable() -export class MembersByCuisineProjectionController extends ProjectionController { - public readonly router: Router; - - constructor( - @inject(MongoTransactionalProjectionOperator) - mongoOperator: MongoTransactionalProjectionOperator, - @inject(Deserializer) deserializer: Deserializer, - @inject(MembersByCuisineProjectionHandler) - private readonly membersByCuisineProjectionHandler: MembersByCuisineProjectionHandler, - ) { - super(mongoOperator, deserializer); - this.router = Router(); - this.router.post( - '/members-by-cuisine', - this.projectIsCardProductActive.bind(this), - ); - } - - private async projectIsCardProductActive( - req: Request, - res: Response, - ): Promise { - const response = await this.processProjectionHttpRequest( - req.body as AmbarHttpRequest, - this.membersByCuisineProjectionHandler, - 'CookingClub_MembersByCuisine', - ); - res.status(200).contentType('application/json').send(response); - } -} diff --git a/src/domain/cookingClub/membership/projection/membersByCuisine/MembersByCuisineProjectionHandler.ts b/src/domain/cookingClub/membership/projection/membersByCuisine/MembersByCuisineProjectionHandler.ts deleted file mode 100644 index 88bca3a..0000000 --- a/src/domain/cookingClub/membership/projection/membersByCuisine/MembersByCuisineProjectionHandler.ts +++ /dev/null @@ -1,60 +0,0 @@ -import { ProjectionHandler } from '@/common'; -import { inject, injectable } from 'tsyringe'; -import { CuisineRepository } from '@/domain/cookingClub/membership/projection/membersByCuisine/CuisineRepository'; -import { ApplicationSubmitted } from '@/domain/cookingClub/membership/event/ApplicationSubmitted'; -import { MembershipApplicationRepository } from '@/domain/cookingClub/membership/projection/membersByCuisine/MembershipApplicationRepository'; -import { MembershipApplication } from '@/domain/cookingClub/membership/projection/membersByCuisine/MembershipApplication'; -import { ApplicationEvaluated } from '@/domain/cookingClub/membership/event/ApplicationEvaluated'; -import { MembershipStatus } from '@/domain/cookingClub/membership/aggregate/membership'; -import { Cuisine } from '@/domain/cookingClub/membership/projection/membersByCuisine/Cuisine'; - -@injectable() -export class MembersByCuisineProjectionHandler extends ProjectionHandler { - constructor( - @inject(CuisineRepository) - private readonly cuisineRepository: CuisineRepository, - @inject(MembershipApplicationRepository) - private readonly membershipApplicationRepository: MembershipApplicationRepository, - ) { - super(); - } - - async project(event: any): Promise { - if (event instanceof ApplicationSubmitted) { - await this.membershipApplicationRepository.save( - new MembershipApplication( - event.aggregateId, - event.firstName, - event.lastName, - event.favoriteCuisine, - ), - ); - } - if ( - event instanceof ApplicationEvaluated && - event.evaluationOutcome === MembershipStatus.Approved - ) { - const membershipApplication = - await this.membershipApplicationRepository.findOneById( - event.aggregateId, - ); - - if (!membershipApplication) - throw new Error('Membership application not found'); - - let cuisine = await this.cuisineRepository.findOneById( - membershipApplication.favoriteCuisine, - ); - - if (!cuisine) { - cuisine = new Cuisine(membershipApplication.favoriteCuisine, []); - } - - cuisine.memberNames.push( - `${membershipApplication.firstName} ${membershipApplication.lastName}`, - ); - - await this.cuisineRepository.save(cuisine); - } - } -} diff --git a/src/domain/cookingClub/membership/projection/membersByCuisine/MembershipApplication.ts b/src/domain/cookingClub/membership/projection/membersByCuisine/MembershipApplication.ts deleted file mode 100644 index 7b30bb8..0000000 --- a/src/domain/cookingClub/membership/projection/membersByCuisine/MembershipApplication.ts +++ /dev/null @@ -1,8 +0,0 @@ -export class MembershipApplication { - constructor( - public readonly _id: string, // needs to be _id to be recognized as an _id field by MongoDB - public readonly firstName: string, - public readonly lastName: string, - public readonly favoriteCuisine: string, - ) {} -} diff --git a/src/domain/cookingClub/membership/projection/membersByCuisine/MembershipApplicationRepository.ts b/src/domain/cookingClub/membership/projection/membersByCuisine/MembershipApplicationRepository.ts deleted file mode 100644 index 848ddfe..0000000 --- a/src/domain/cookingClub/membership/projection/membersByCuisine/MembershipApplicationRepository.ts +++ /dev/null @@ -1,31 +0,0 @@ -import { MongoTransactionalProjectionOperator } from '@/common'; -import { inject, injectable } from 'tsyringe'; -import { MembershipApplication } from '@/domain/cookingClub/membership/projection/membersByCuisine/MembershipApplication'; - -@injectable() -export class MembershipApplicationRepository { - private readonly collectionName = - 'CookingClub_MembersByCuisine_MembershipApplication'; - - constructor( - @inject(MongoTransactionalProjectionOperator) - private readonly mongoOperator: MongoTransactionalProjectionOperator, - ) {} - - async save(membershipApplication: MembershipApplication): Promise { - await this.mongoOperator.replaceOne( - this.collectionName, - { _id: membershipApplication._id }, - membershipApplication, - { upsert: true }, - ); - } - - async findOneById(_id: string): Promise { - const results = await this.mongoOperator.find( - this.collectionName, - { _id }, - ); - return results[0] || null; - } -} diff --git a/src/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQuery.ts b/src/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQuery.ts deleted file mode 100644 index e5805b9..0000000 --- a/src/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQuery.ts +++ /dev/null @@ -1,3 +0,0 @@ -import { Query } from '@/common'; - -export class MembersByCuisineQuery extends Query {} diff --git a/src/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQueryController.ts b/src/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQueryController.ts deleted file mode 100644 index 2cdbf15..0000000 --- a/src/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQueryController.ts +++ /dev/null @@ -1,37 +0,0 @@ -import { Request, Response, Router } from 'express'; -import { - QueryController, - MongoTransactionalProjectionOperator, -} from '@/common'; -import { inject, injectable } from 'tsyringe'; -import { MembersByCuisineQueryHandler } from '@/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQueryHandler'; -import { MembersByCuisineQuery } from '@/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQuery'; - -@injectable() -export class MembersByCuisineQueryController extends QueryController { - public readonly router: Router; - - private readonly membersByCuisineQueryHandler: MembersByCuisineQueryHandler; - - constructor( - @inject(MongoTransactionalProjectionOperator) - mongoTransactionalProjectionOperator: MongoTransactionalProjectionOperator, - @inject(MembersByCuisineQueryHandler) - membersByCuisineQueryHandler: MembersByCuisineQueryHandler, - ) { - super(mongoTransactionalProjectionOperator); - this.membersByCuisineQueryHandler = membersByCuisineQueryHandler; - this.router = Router(); - this.router.post('/members-by-cuisine', this.membersByCuisine.bind(this)); - } - - async membersByCuisine(_req: Request, res: Response): Promise { - const query = new MembersByCuisineQuery(); - - const result = await this.processQuery( - query, - this.membersByCuisineQueryHandler, - ); - res.status(200).json(result); - } -} diff --git a/src/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQueryHandler.ts b/src/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQueryHandler.ts deleted file mode 100644 index f3dab26..0000000 --- a/src/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQueryHandler.ts +++ /dev/null @@ -1,23 +0,0 @@ -import { QueryHandler, MongoTransactionalProjectionOperator } from '@/common'; -import { inject, injectable } from 'tsyringe'; -import { CuisineRepository } from '@/domain/cookingClub/membership/projection/membersByCuisine/CuisineRepository'; -import { MembersByCuisineQuery } from '@/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQuery'; -import { Cuisine } from '@/domain/cookingClub/membership/projection/membersByCuisine/Cuisine'; - -@injectable() -export class MembersByCuisineQueryHandler extends QueryHandler { - private readonly cuisineRepository: CuisineRepository; - - constructor( - @inject(MongoTransactionalProjectionOperator) - mongoTransactionalProjectionOperator: MongoTransactionalProjectionOperator, - @inject(CuisineRepository) cuisineRepository: CuisineRepository, - ) { - super(mongoTransactionalProjectionOperator); - this.cuisineRepository = cuisineRepository; - } - - async handleQuery(_query: MembersByCuisineQuery): Promise { - return await this.cuisineRepository.findAll(); - } -} diff --git a/src/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionController.ts b/src/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionController.ts deleted file mode 100644 index 3d855f1..0000000 --- a/src/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionController.ts +++ /dev/null @@ -1,43 +0,0 @@ -import { - ReactionController, - AmbarHttpRequest, - PostgresTransactionalEventStore, - MongoTransactionalProjectionOperator, - Deserializer, -} from '@/common'; -import { Request, Response, Router } from 'express'; -import { inject, injectable } from 'tsyringe'; -import { EvaluateApplicationReactionHandler } from '@/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionHandler'; - -@injectable() -export class EvaluateApplicationReactionController extends ReactionController { - public readonly router: Router; - - constructor( - @inject(PostgresTransactionalEventStore) - eventStore: PostgresTransactionalEventStore, - @inject(MongoTransactionalProjectionOperator) - mongoOperator: MongoTransactionalProjectionOperator, - @inject(Deserializer) deserializer: Deserializer, - @inject(EvaluateApplicationReactionHandler) - private readonly evaluateApplicationReactionHandler: EvaluateApplicationReactionHandler, - ) { - super(eventStore, mongoOperator, deserializer); - this.router = Router(); - this.router.post( - '/evaluate-application', - this.reactWithEvaluateApplication.bind(this), - ); - } - - async reactWithEvaluateApplication( - req: Request, - res: Response, - ): Promise { - const response = await this.processReactionHttpRequest( - req.body as AmbarHttpRequest, - this.evaluateApplicationReactionHandler, - ); - res.status(200).contentType('application/json').send(response); - } -} diff --git a/src/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionHandler.ts b/src/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionHandler.ts deleted file mode 100644 index c8f0317..0000000 --- a/src/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionHandler.ts +++ /dev/null @@ -1,80 +0,0 @@ -import { - ReactionHandler, - PostgresTransactionalEventStore, - IdGenerator, - Event, -} from '@/common'; -import { inject, injectable } from 'tsyringe'; -import { ApplicationSubmitted } from '@/domain/cookingClub/membership/event/ApplicationSubmitted'; -import { - Membership, - MembershipStatus, -} from '@/domain/cookingClub/membership/aggregate/membership'; -import { ApplicationEvaluated } from '@/domain/cookingClub/membership/event/ApplicationEvaluated'; - -@injectable() -export class EvaluateApplicationReactionHandler extends ReactionHandler { - constructor( - @inject(PostgresTransactionalEventStore) - eventStore: PostgresTransactionalEventStore, - ) { - super(eventStore); - } - - async react(event: Event): Promise { - if (!(event instanceof ApplicationSubmitted)) { - return; - } - - const aggregateData = - await this.postgresTransactionalEventStore.findAggregate( - event.aggregateId, - ); - const membership = aggregateData.aggregate; - - if (membership.status !== MembershipStatus.Requested) { - return; - } - - const reactionEventId = IdGenerator.generateDeterministicId( - `CookingClub_Membership_ReviewedApplication:${event.eventId}`, - ); - if ( - await this.postgresTransactionalEventStore.doesEventAlreadyExist( - reactionEventId, - ) - ) { - return; - } - - const shouldApprove = - event.yearsOfProfessionalExperience == 0 && - event.numberOfCookingBooksRead > 0; - - if (shouldApprove) { - const reactionEvent = new ApplicationEvaluated( - reactionEventId, - membership.aggregateId, - membership.aggregateVersion + 1, - aggregateData.correlationIdOfLastEvent, - aggregateData.eventIdOfLastEvent, - new Date(), - MembershipStatus.Approved, - ); - - await this.postgresTransactionalEventStore.saveEvent(reactionEvent); - } else { - const reactionEvent = new ApplicationEvaluated( - reactionEventId, - membership.aggregateId, - membership.aggregateVersion + 1, - aggregateData.correlationIdOfLastEvent, - aggregateData.eventIdOfLastEvent, - new Date(), - MembershipStatus.Rejected, - ); - - await this.postgresTransactionalEventStore.saveEvent(reactionEvent); - } - } -} diff --git a/src/domain/cookingClub/membership2/aggregate/membership.ts b/src/domain/cookingClub/membership2/aggregate/membership.ts new file mode 100644 index 0000000..4980307 --- /dev/null +++ b/src/domain/cookingClub/membership2/aggregate/membership.ts @@ -0,0 +1,37 @@ +export { type MembershipStatus, Membership, schema_MembershipStatus }; + +import { Aggregate, Id } from '@/lib/eventSourcing/event'; +import { Schema } from '@/lib/json/schema'; +import * as s from '@/lib/json/schema'; + +type MembershipStatus = 'Requested' | 'Approved' | 'Rejected'; + +const schema_MembershipStatus = s.oneOf( + (str) => { + switch (str) { + case 'Requested': + return s.stringLiteral('Requested') as Schema; + case 'Approved': + return s.stringLiteral('Approved') as Schema; + case 'Rejected': + return s.stringLiteral('Rejected') as Schema; + default: + return str satisfies never; + } + }, + [ + s.stringLiteral('Requested') as Schema, + s.stringLiteral('Approved') as Schema, + s.stringLiteral('Rejected') as Schema, + ], +); + +class Membership implements Aggregate { + constructor( + readonly aggregateId: Id, + readonly aggregateVersion: number, + public firstName: string, + public lastName: string, + public status: MembershipStatus, + ) {} +} diff --git a/src/domain/cookingClub/membership2/command/submitApplication.ts b/src/domain/cookingClub/membership2/command/submitApplication.ts new file mode 100644 index 0000000..3fbbdb2 --- /dev/null +++ b/src/domain/cookingClub/membership2/command/submitApplication.ts @@ -0,0 +1,44 @@ +export { controller }; + +import * as d from '@/lib/json/decoder'; +import { CommandController, CommandHandler } from '@/app/handleCommand'; +import { Future } from '@/lib/Future'; +import { Response, json } from '@/lib/router'; +import { Id } from '@/lib/eventSourcing/event'; +import { ApplicationSubmitted } from '@/domain/cookingClub/membership2/events/membership/applicationSubmitted'; +import { Membership } from '@/domain/cookingClub/membership2/aggregate/membership'; + +type Command = d.Infer; +const decoder = d.object({ + firstName: d.string, + lastName: d.string, + favouriteCousine: d.string, + yearsOfProfessionalExperience: d.number, + numberOfCookingBooksRead: d.number, +}); + +const handler: CommandHandler = ({ + command, + store, +}): Future => { + store.emit({ + aggregate: ApplicationSubmitted.aggregate, + event: new ApplicationSubmitted({ + type: ApplicationSubmitted.type, + aggregateId: Id.random(), + firstName: command.firstName, + lastName: command.lastName, + favouriteCousine: command.favouriteCousine, + yearsOfProfessionalExperience: command.yearsOfProfessionalExperience, + numberOfCookingBooksRead: command.numberOfCookingBooksRead, + }), + }); + + return Future.resolve( + json({ + content: { message: 'success' }, + }), + ); +}; + +const controller: CommandController = { decoder, handler }; diff --git a/src/domain/cookingClub/membership2/events/membership/applicationEvaluated.ts b/src/domain/cookingClub/membership2/events/membership/applicationEvaluated.ts new file mode 100644 index 0000000..d66c41d --- /dev/null +++ b/src/domain/cookingClub/membership2/events/membership/applicationEvaluated.ts @@ -0,0 +1,27 @@ +export { ApplicationEvaluated }; + +import { Id, TransformationEvent, toSchema } from '@/lib/eventSourcing/event'; +import * as s from '@/lib/json/schema'; +import { + Membership, + schema_MembershipStatus, +} from '@/domain/cookingClub/membership2/aggregate/membership'; + +const type = 'ApplicationEvaluated' as const; +const args = s.object({ + type: s.stringLiteral(type), + aggregateId: Id.schema(), + evaluationOutcome: schema_MembershipStatus, +}); + +class ApplicationEvaluated implements TransformationEvent { + static readonly aggregate = Membership; + static readonly type = type; + static readonly schema = toSchema(this, args); + constructor(readonly values: s.Infer) {} + + transformAggregate(aggregate: Membership): Membership { + aggregate.status = this.values.evaluationOutcome; + return aggregate; + } +} diff --git a/src/domain/cookingClub/membership2/events/membership/applicationSubmitted.ts b/src/domain/cookingClub/membership2/events/membership/applicationSubmitted.ts new file mode 100644 index 0000000..7494efe --- /dev/null +++ b/src/domain/cookingClub/membership2/events/membership/applicationSubmitted.ts @@ -0,0 +1,33 @@ +export { ApplicationSubmitted }; + +import { Id, CreationEvent, toSchema } from '@/lib/eventSourcing/event'; +import * as s from '@/lib/json/schema'; +import { Membership } from '@/domain/cookingClub/membership2/aggregate/membership'; + +const type = 'ApplicationSubmitted' as const; +const args = s.object({ + type: s.stringLiteral(type), + aggregateId: Id.schema(), + firstName: s.string, + lastName: s.string, + favouriteCousine: s.string, + yearsOfProfessionalExperience: s.number, + numberOfCookingBooksRead: s.number, +}); + +class ApplicationSubmitted implements CreationEvent { + static readonly aggregate = Membership; + static readonly type = type; + static readonly schema = toSchema(this, args); + constructor(readonly values: s.Infer) {} + + createAggregate(): Membership { + return new Membership( + this.values.aggregateId, + 0, + this.values.firstName, + this.values.lastName, + 'Requested', + ); + } +} diff --git a/src/domain/cookingClub/membership2/projection/membersByCuisine.ts b/src/domain/cookingClub/membership2/projection/membersByCuisine.ts new file mode 100644 index 0000000..5d4459a --- /dev/null +++ b/src/domain/cookingClub/membership2/projection/membersByCuisine.ts @@ -0,0 +1,166 @@ +export { + controller, + RepoCuisine, + type Cuisine, + RepoMembershipApplication, + type MembershipApplication, +}; + +import * as d from '@/lib/json/decoder'; +import * as s from '@/lib/json/schema'; +import { Id } from '@/lib/eventSourcing/event'; +import { accept } from '@/lib/eventSourcing/projection'; +import { + ProjectionHandler, + ProjectionController, +} from '@/app/handleProjection'; +import { Future } from '@/lib/Future'; +import { ApplicationSubmitted } from '@/domain/cookingClub/membership2/events/membership/applicationSubmitted'; +import { ApplicationEvaluated } from '@/domain/cookingClub/membership2/events/membership/applicationEvaluated'; +import { AmbarResponse, ErrorMustRetry } from '@/lib/ambar'; +import * as m from '@/lib/Maybe'; +import { + Repository, + Collection, + MongoProjectionStore, +} from '@/app/projectionStore'; + +// ------------------------------------------------ +// Cuisine +// ------------------------------------------------ + +type Cuisine = s.Infer; + +const schema_Cuisine = s.object({ + name: s.string, // unique + memberNames: s.array(s.string), +}); + +class RepoCuisine { + static collectionName = 'CookingClub_MembersByCuisine_Cuisine' as const; + static schema = schema_Cuisine; + static async createIndexes(_collection: Collection) { + return; + } + static toId(c: Cuisine): string { + return c.name; + } + + constructor( + private repo: Repository, + private store: MongoProjectionStore, + ) {} + + async save(cuisine: Cuisine): Promise { + await this.store.upsert(this.repo, cuisine); + } + + async findOneById(_id: string): Promise { + const results = await this.store.find(this.repo, { _id }); + return results[0] || null; + } + + async findAll(): Promise { + return this.store.find(this.repo, {}); + } +} + +// ------------------------------------------------ +// Membership Application +// ------------------------------------------------ + +type MembershipApplication = s.Infer; + +const schema_MembershipApplication = s.object({ + id: Id.schema(), + firstName: s.string, + lastName: s.string, + favouriteCuisine: s.string, +}); + +class RepoMembershipApplication { + static collectionName = + 'CookingClub_MembersByCuisine_MembershipApplication' as const; + static schema = schema_MembershipApplication; + static async createIndexes(_collection: Collection) { + return; + } + static toId(c: MembershipApplication): string { + return c.id.value; + } + + constructor( + private repo: Repository, + private store: MongoProjectionStore, + ) {} + + async save(cuisine: MembershipApplication): Promise { + await this.store.upsert(this.repo, cuisine); + } + + async getById(_id: Id): Promise { + const results = await this.store.find(this.repo, { + _id, + }); + const found = results[0] || null; + if (found === null) { + throw new Error(`Unknown membership application ID: ${_id.value}`); + } + return found; + } +} + +// ------------------------------------------------ +// Projection +// ------------------------------------------------ + +type Events = m.Infer>; + +const decoder = accept([ApplicationSubmitted, ApplicationEvaluated]); + +const handler: ProjectionHandler = ({ + event, + projections, +}): Future => + Future.attemptP(async () => { + const repoCuisine = projections[RepoCuisine.collectionName]; + const repoMembershipApplication = + projections[RepoMembershipApplication.collectionName]; + + switch (true) { + case event instanceof ApplicationSubmitted: { + await repoMembershipApplication.save({ + id: event.values.aggregateId, + firstName: event.values.firstName, + lastName: event.values.lastName, + favouriteCuisine: event.values.favouriteCousine, + }); + return; + } + case event instanceof ApplicationEvaluated: { + if (event.values.evaluationOutcome != 'Approved') return; + const application = await repoMembershipApplication.getById( + event.values.aggregateId, + ); + + const newCuisine = { + name: application.favouriteCuisine, + memberNames: [], + }; + const cuisine = + (await repoCuisine.findOneById(application.favouriteCuisine)) || + newCuisine; + + cuisine.memberNames.push( + `${application.firstName} ${application.lastName}`, + ); + await repoCuisine.save(cuisine); + return; + } + default: { + return event satisfies never; + } + } + }).mapRej((err) => new ErrorMustRetry(err.message)); + +const controller: ProjectionController = { decoder, handler }; diff --git a/src/domain/cookingClub/membership2/query/membersByCuisine.ts b/src/domain/cookingClub/membership2/query/membersByCuisine.ts new file mode 100644 index 0000000..f669d6c --- /dev/null +++ b/src/domain/cookingClub/membership2/query/membersByCuisine.ts @@ -0,0 +1,23 @@ +export { controller }; + +import * as d from '@/lib/json/decoder'; +import { QueryHandler, QueryController } from '@/app/handleQuery'; +import { Future } from '@/lib/Future'; +import { internalServerError } from '@/app/responses'; +import { RepoCuisine } from '@/domain/cookingClub/membership2/projection/membersByCuisine'; +import * as router from '@/lib/router'; + +type Query = d.Infer; + +const decoder = d.object({}); + +const handler: QueryHandler = ({ + projections, +}): Future => + Future.attemptP(async () => { + const repoCuisine = projections[RepoCuisine.collectionName]; + const cuisines = await repoCuisine.findAll(); + return router.json({ content: cuisines }); + }).mapRej((_) => internalServerError); + +const controller: QueryController = { decoder, handler }; diff --git a/src/domain/cookingClub/membership2/reaction/evaluateApplication.ts b/src/domain/cookingClub/membership2/reaction/evaluateApplication.ts new file mode 100644 index 0000000..48bd533 --- /dev/null +++ b/src/domain/cookingClub/membership2/reaction/evaluateApplication.ts @@ -0,0 +1,45 @@ +export { controller }; + +import * as d from '@/lib/json/decoder'; +import { accept } from '@/lib/eventSourcing/projection'; +import { ReactionHandler, ReactionController } from '@/app/handleReaction'; +import { Future } from '@/lib/Future'; +import { ApplicationSubmitted } from '@/domain/cookingClub/membership2/events/membership/applicationSubmitted'; +import { ApplicationEvaluated } from '@/domain/cookingClub/membership2/events/membership/applicationEvaluated'; +import { Membership } from '@/domain/cookingClub/membership2/aggregate/membership'; +import { AmbarResponse, ErrorMustRetry } from '@/lib/ambar'; +import * as m from '@/lib/Maybe'; + +type Events = m.Infer>; + +const decoder = accept([ApplicationSubmitted]); + +const handler: ReactionHandler = ({ + event, + store, +}): Future => + Future.attemptP(async () => { + const { aggregate: membership } = await store.find( + Membership, + event.values.aggregateId, + ); + + if (membership.status !== 'Requested') { + return; + } + + const shouldApprove = + event.values.yearsOfProfessionalExperience == 0 && + event.values.numberOfCookingBooksRead > 0; + + await store.emit({ + aggregate: Membership, + event: new ApplicationEvaluated({ + type: 'ApplicationEvaluated', + aggregateId: membership.aggregateId, + evaluationOutcome: shouldApprove ? 'Approved' : 'Rejected', + }), + }); + }).mapRej((err) => new ErrorMustRetry(err.message)); + +const controller: ReactionController = { decoder, handler }; diff --git a/src/index.ts b/src/index.ts index 5397ce9..3a702c2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,21 +1,27 @@ import 'tsconfig-paths/register'; // enable absolute paths import 'reflect-metadata'; import express from 'express'; -import { container } from 'tsyringe'; import { configureDependencies } from '@/di/container'; import { scopedContainer } from '@/di/scopedContainer'; -import { MongoInitializer } from '@/common/util/MongoInitializer'; -import { PostgresInitializer } from '@/common/util/PostgresInitializer'; import { log } from '@/common/util/Logger'; -import { AmbarAuthMiddleware } from '@/common/ambar/AmbarAuthMiddleware'; -import { SubmitApplicationCommandController } from '@/domain/cookingClub/membership/command/submitApplication/SubmitApplicationCommandController'; -import { MembersByCuisineQueryController } from '@/domain/cookingClub/membership/query/membersByCuisine/MembersByCuisineQueryController'; -import { EvaluateApplicationReactionController } from '@/domain/cookingClub/membership/reaction/evaluateApplication/EvaluateApplicationReactionController'; -import { MembersByCuisineProjectionController } from '@/domain/cookingClub/membership/projection/membersByCuisine/MembersByCuisineProjectionController'; +import { handleCommand, CommandController } from '@/app/handleCommand'; +import { Event } from '@/lib/eventSourcing/event'; +import { + handleReaction, + wrapWithEventStore, + ReactionController, +} from '@/app/handleReaction'; +import { handleProjection, ProjectionController } from '@/app/handleProjection'; +import { handleQuery, QueryController } from '@/app/handleQuery'; +import * as membership_command_submitApplication from '@/domain/cookingClub/membership2/command/submitApplication'; +import * as membership_reaction_evaluateApplication from '@/domain/cookingClub/membership2/reaction/evaluateApplication'; +import * as membership_projection_membersByCuisine from '@/domain/cookingClub/membership2/projection/membersByCuisine'; +import * as membership_query_membersByCuisine from '@/domain/cookingClub/membership2/query/membersByCuisine'; async function main() { // Configure dependency injection - await configureDependencies(); + const { withEventStore, withProjectionStore, services, repositories } = + await configureDependencies(); // Create express app const app = express(); @@ -24,37 +30,71 @@ async function main() { // 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, + const command = (endpoint: string, controller: CommandController) => + app.post( + endpoint, + handleCommand( + withEventStore, + withProjectionStore, + services, + repositories, + controller, + ), ); - return controller.router(req, res, next); - }); - app.use( - '/api/v1/cooking-club/membership/projection', - AmbarAuthMiddleware, - (req, res, next) => { - const controller = req.container.resolve( - MembersByCuisineProjectionController, - ); - return controller.router(req, res, next); - }, + + const reaction = >( + endpoint: string, + controller: ReactionController, + ) => + app.post( + endpoint, + handleReaction( + wrapWithEventStore(withEventStore), + services, + repositories, + controller, + ), + ); + + const projection = >( + endpoint: string, + controller: ProjectionController, + ) => + app.post( + endpoint, + handleProjection(withProjectionStore, repositories, controller), + ); + + const query = (endpoint: string, controller: QueryController) => + app.get( + endpoint, + handleQuery(withProjectionStore, repositories, controller), + ); + + ////////////////////////////////////////////////////////////////////// + + command( + '/api/v1/cooking-club/membership/command/submit-application', + membership_command_submitApplication.controller, ); - 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, - ); - return controller.router(req, res, next); - }, + + reaction( + '/api/v1/cooking-club/membership/reaction/evaluateApplication', + membership_reaction_evaluateApplication.controller, + ); + + projection( + '/api/v1/cooking-club/membership/projection/membersByCuisine', + membership_projection_membersByCuisine.controller, + ); + + query( + '/api/v1/cooking-club/membership/projection/membersByCuisine', + membership_query_membersByCuisine.controller, ); + + ////////////////////////////////////////////////////////////////////// + app.get('/docker_healthcheck', (_req, res) => res.send('OK')); app.get('/', (_req, res) => res.send('OK')); @@ -75,20 +115,9 @@ async function main() { ); // Initialize databases and start server - - 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'); - }); - }) - .catch((error) => { - console.error('Failed to initialize databases:', error); - process.exit(1); - }); + app.listen(8080, () => { + console.log('Server is running on port 8080'); + }); } await main(); diff --git a/src/lib/ambar.ts b/src/lib/ambar.ts new file mode 100644 index 0000000..76cfe25 --- /dev/null +++ b/src/lib/ambar.ts @@ -0,0 +1,131 @@ +// Interacting with Ambar's infra +export { + type AmbarResponse, + toResponse, + Success, + ErrorMustRetry, + payloadDecoder, +}; + +import * as router from '@/lib/router'; +import * as e from '@/lib/json/encoder'; +import * as d from '@/lib/json/decoder'; +import { EventData, schema_EventData } from '@/lib/eventSourcing/eventStore'; +import { Decoder } from '@/lib/json/decoder'; +import { Encoder } from '@/lib/json/encoder'; +import { Schema } from '@/lib/json/schema'; + +// Success response from data destination +class Success { + // @ts-expect-error _tag's existence prevents structural comparison + private readonly _tag: null = null; + constructor() {} +} + +// Error response from data destination +class ErrorMustRetry { + // @ts-expect-error _tag's existence prevents structural comparison + private readonly _tag: null = null; + constructor(public readonly description: string) {} +} + +// Response to an Ambar request sent to a Reaction or a Projection. +type AmbarResponse = Success | ErrorMustRetry; + +function toResponse(r: AmbarResponse): router.Response { + switch (true) { + case r instanceof Success: + return router.json({ + status: 200, + content: { result: { success: {} } }, + }); + case r instanceof ErrorMustRetry: + return router.json({ + status: 200, + content: { + result: { + error: { + policy: 'must_retry', + description: r.description, + }, + }, + }, + }); + default: + return r satisfies never; + } +} + +// The request that Ambar sends to Reactions and Projections +type AmbarHttpRequest = { + data_source_id: string; + data_source_description: string; + data_destination_id: string; + data_destination_description: string; + payload: T; +}; + +// Create a decoder that operates on an AmbarHttpRequest +function payloadDecoder(decoder: Decoder): Decoder> { + const dummy: Encoder = new e.Encoder((_) => null); + const eschema: Schema> = schema_EventData( + new Schema(decoder, dummy), + ); + const reqDecoder: Decoder>> = d.object({ + data_source_id: d.string, + data_source_description: d.string, + data_destination_id: d.string, + data_destination_description: d.string, + payload: eschema.decoder, + }); + + return reqDecoder.map((v) => v.payload); +} + +// ================================================================================ +// AUTH +// ================================================================================ + +import { Request, Response, NextFunction } from 'express'; +import env from '@/app/environment'; + +const VALID_USERNAME = env.AMBAR_HTTP_USERNAME; +const VALID_PASSWORD = env.AMBAR_HTTP_PASSWORD; + +if (!VALID_USERNAME || !VALID_PASSWORD) { + throw new Error( + 'Environment variables AUTH_USERNAME and AUTH_PASSWORD must be set', + ); +} + +export const AmbarAuthMiddleware = ( + req: Request, + res: Response, + next: NextFunction, +) => { + const authHeader = req.headers.authorization; + + if (!authHeader) { + return res.status(401).json({ error: 'Authentication required' }); + } + + if (!authHeader.startsWith('Basic ')) { + return res.status(401).json({ error: 'Basic authentication required' }); + } + + try { + const base64Credentials = authHeader.split(' ')[1] || ''; + const credentials = Buffer.from(base64Credentials, 'base64').toString( + 'utf8', + ); + const [username, password] = credentials.split(':'); + + if (username === VALID_USERNAME && password === VALID_PASSWORD) { + return next(); + } else { + return res.status(401).json({ error: 'Invalid credentials' }); + } + } catch (error) { + return res.status(401).json({ error: 'Invalid authentication format' }); + } +}; diff --git a/src/lib/eventSourcing/event.ts b/src/lib/eventSourcing/event.ts index e014725..1776b76 100644 --- a/src/lib/eventSourcing/event.ts +++ b/src/lib/eventSourcing/event.ts @@ -1,17 +1,22 @@ export { type Event, type Aggregate, - EventClass, TransformationEvent, CreationEvent, type EventInfo, EventInfo_schema, Id, + toSchema, }; import * as s from '@/lib/json/schema'; import { Schema } from '@/lib/json/schema'; import { POSIX } from '@/lib/time'; +import { createHash, randomBytes } from 'crypto'; + +const ALPHANUMERIC_CHARACTERS = + '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'; +const ID_LENGTH = 56; // @ts-ignore class Id { @@ -26,6 +31,26 @@ class Id { (id) => id.value, ); } + + static random(): Id { + const toChar = (byte: number) => + ALPHANUMERIC_CHARACTERS.charAt(byte % ALPHANUMERIC_CHARACTERS.length); + const str = Array.from(randomBytes(ID_LENGTH)).map(toChar).join(''); + return new Id(str); + } + + static deterministic(seed: string): string { + if (seed.trim() == '') { + throw new Error('Input string cannot be null or empty'); + } + + const first = createHash('sha256').update(seed).digest(); + const second = createHash('sha256').update(first).digest(); + const combined = Buffer.concat([first, second]); + const base64Encoded = combined.toString('base64'); + const cleanId = base64Encoded.replace(/[^A-Za-z0-9]/g, ''); + return cleanId.substring(0, ID_LENGTH); + } } // Class which all events derive from. Used for type constraints. @@ -34,30 +59,21 @@ interface Aggregate { aggregateVersion: number; } -type Event> = EventClass; - // Class which all events derive from. Used for type constraints. -abstract class EventClass> { +abstract class Event> { abstract values: { type: string; aggregateId: Id; }; - abstract schema: Schema; } // The first event for an aggregate. -abstract class CreationEvent> extends EventClass< - Self, - T -> { +abstract class CreationEvent> extends Event { abstract createAggregate(): T; } // Any event that is not the first one for an aggregate. -abstract class TransformationEvent< - Self, - T extends Aggregate, -> extends EventClass { +abstract class TransformationEvent> extends Event { abstract transformAggregate(aggregate: T): T; } @@ -72,3 +88,17 @@ const EventInfo_schema = s.object({ causation_id: Id.schema>>(), recorded_on: POSIX.schema, }); + +// Create an event's schema. +// Enforces that the class' `type` property has +// the same type as the instance's `value.type` property. +function toSchema< + T extends string, + W extends { type: T }, + E extends { values: W }, +>(ctr: (new (values: W) => E) & { type: T }, schemaArgs: Schema): Schema { + return schemaArgs.dimap( + (v) => new ctr(v), + (v) => v.values, + ); +} diff --git a/src/lib/eventSourcing/eventStore.ts b/src/lib/eventSourcing/eventStore.ts index f4dba1b..e242184 100644 --- a/src/lib/eventSourcing/eventStore.ts +++ b/src/lib/eventSourcing/eventStore.ts @@ -1,11 +1,15 @@ export { type EventStore, type AggregateAndEventIdsInLastEvent, - Hydrator, + Schemas, type Constructor, type EventData, schema_EventData, makeSchema, + makeDecoder, + CSchema, + TSchema, + type Serialized, }; import { @@ -53,9 +57,9 @@ interface EventStore { aggregateId: Id, ): Promise<{ aggregate: T; lastEvent: EventInfo }>; - save, T extends Aggregate>(args: { + emit>(args: { aggregate: Constructor; - event: CreationEvent | TransformationEvent; + event: CreationEvent | TransformationEvent; event_id?: Id>; correlation_id?: Id>; causation_id?: Id>; @@ -130,12 +134,41 @@ const schema_EventData = (s: Schema): Schema> => type Constructor = new (...args: any[]) => T; -type Schemas> = { - creation: Schema>>; - transformation: Schema>>; +class CSchema< + A extends Aggregate, + E extends CreationEvent, + T extends E['values']['type'], +> { + constructor( + public aggregate: Constructor, + public schema: Schema, + public type: T, + ) {} +} + +class TSchema< + A extends Aggregate, + E extends TransformationEvent, + T extends E['values']['type'], +> { + constructor( + public aggregate: Constructor, + public schema: Schema, + public type: T, + ) {} +} + +type SomeSchema> = + | CSchema, any> + | TSchema, any>; + +// Efficient decoders for all creation and transformation events for an aggregate. +type Decoders> = { + creation: Decoder>>; + transformation: Decoder>>; }; -/* Note [Hydrator] +/* Note [Schemas] We need some type-safe way to decode events for an aggregate. That is, without casting. We perform type-directed decoding, where we specify the type of the aggregate, @@ -144,25 +177,85 @@ type Schemas> = { This ensures that we will never apply an incorrect aggregate transformation or create an aggregate of the incorrect type. */ -class Hydrator { - private tmap = new Map, Schemas>(); - - constructor() {} - - // add support for deserializing an aggregate's events. - add>({ - aggregate, - creation, - transformation, - }: { - aggregate: Constructor; - creation: Schema>; - transformation: Schema>; - }): void { - this.tmap.set(aggregate, { - creation: schema_EventData(creation), - transformation: schema_EventData(transformation), +class Schemas { + private cmap = new Map>, Decoders>(); + private tmap = new Map>>(); + + constructor( + arr: Array<{ + type: string; + schema: Schema; + aggregate: Constructor>; + }>, + ) { + const entries: Array>> = arr.map((entry) => { + if (entry instanceof CSchema || entry instanceof TSchema) { + return entry; + } + throw new Error(`Value should be an instance of SomeSchema`); }); + + // Set encoders + for (const entry of entries) { + if (this.tmap.has(entry.type)) { + throw new Error(`Duplicate entry for ${entry.type}`); + } + + if (entry instanceof CSchema) { + this.tmap.set(entry.type, schema_EventData(entry.schema).encoder); + } else if (entry instanceof TSchema) { + this.tmap.set(entry.type, schema_EventData(entry.schema).encoder); + } else { + entry satisfies never; + } + } + + type Events> = { + creation: Array<{ + type: string; + schema: Schema>; + }>; + transformation: Array<{ + type: string; + schema: Schema>; + }>; + }; + + const emap: Map>, Events> = new Map(); + + for (const entry of entries) { + const aggregate = entry.aggregate; + const found: Events = emap.get(aggregate) || { + creation: [], + transformation: [], + }; + + if (entry instanceof CSchema) { + found.creation.push({ schema: entry.schema, type: entry.type }); + } else if (entry instanceof TSchema) { + found.transformation.push({ schema: entry.schema, type: entry.type }); + } else { + entry satisfies never; + } + } + + for (const [aggregate, events] of emap.entries()) { + this.cmap.set(aggregate, { + creation: schema_EventData(makeSchema(events.creation)).decoder, + transformation: schema_EventData(makeSchema(events.transformation)) + .decoder, + }); + } + } + + encode>(edata: EventData): Json { + const ty = edata.event.values.type; + const found = this.tmap.get(ty) as undefined | Encoder>; + if (found == undefined) { + throw new Error(`Unknown event type ${ty}`); + } + + return found.run(edata); } // Build an aggregate from all its serialized events. @@ -170,7 +263,7 @@ class Hydrator { cls: Constructor, serialized: Json[], ): Result { - const schemas = this.tmap.get(cls) as undefined | Schemas; + const schemas = this.cmap.get(cls) as undefined | Decoders; if (schemas == undefined) { throw new Error(`Unknown aggregate ${cls.name}`); } @@ -180,10 +273,10 @@ class Hydrator { } return d - .decode(serialized[0], schemas.creation.decoder) + .decode(serialized[0], schemas.creation) .then(({ event: first, info }) => d - .decode(serialized.slice(1), d.array(schemas.transformation.decoder)) + .decode(serialized.slice(1), d.array(schemas.transformation)) .map((es) => { let aggregate = first.createAggregate(); let lastEvent = info; @@ -205,7 +298,7 @@ type EventConstructor = { type: string; schema: Schema }; // Create an efficient schema given a list of event classes // -// To be used when joining schemas for the Hydrator +// To be used when joining schemas for the Schemas function makeSchema( ts: T, ): Schema> { @@ -232,3 +325,9 @@ function makeSchema( return new Schema(decoder, encoder); } + +function makeDecoder( + ts: T, +): Decoder> { + return makeSchema(ts).decoder; +} diff --git a/src/lib/json/decoder.ts b/src/lib/json/decoder.ts index fd375c3..81f05ff 100644 --- a/src/lib/json/decoder.ts +++ b/src/lib/json/decoder.ts @@ -52,6 +52,7 @@ export { failure, optional, succeed, + both, }; import { Result, Success, Failure, traverse } from '@/lib/Result'; @@ -105,6 +106,22 @@ const succeed = always; const any: Decoder = new Decoder((v) => Success(v)); +// Use two decoders on the same input +const both = (left: Decoder, right: Decoder): Decoder<[T, U]> => + new Decoder((u) => { + const l = left.run(u); + if (l instanceof Failure) { + return new Failure(l.error); + } + + const r = right.run(u); + if (r instanceof Failure) { + return new Failure(r.error); + } + + return Success([l.value, r.value]); + }); + const string: Decoder = new Decoder((v) => typeof v === 'string' ? Success(v) diff --git a/src/lib/json/encoder.ts b/src/lib/json/encoder.ts index 1a57e78..b13b932 100644 --- a/src/lib/json/encoder.ts +++ b/src/lib/json/encoder.ts @@ -18,6 +18,7 @@ export { triple, optional, oneOf, + both, }; import { Maybe, Nothing, Just, Nullable } from '@/lib/Maybe'; @@ -43,6 +44,14 @@ type EncoderDef = { }; const toAny = (): Encoder => new Encoder((v) => v); + +// Encode two values into a single one. +// Conflicts result in properties being ovewritten. +const both = (left: Encoder, right: Encoder): Encoder<[T, U]> => + new Encoder(([l, r]) => { + return Object.assign({}, left.run(l), right.run(r)); + }); + const json: Encoder = toAny(); const boolean: Encoder = toAny(); const number: Encoder = toAny(); diff --git a/src/lib/json/schema.ts b/src/lib/json/schema.ts index 90b9d12..c443e3c 100644 --- a/src/lib/json/schema.ts +++ b/src/lib/json/schema.ts @@ -14,6 +14,7 @@ export { string, array, json, + both, maybe, nullable, optional, @@ -83,6 +84,12 @@ const string: Schema = new Schema(D.string, E.string); const array = (schema: Schema): Schema> => new Schema(D.array(schema.decoder), E.array(schema.encoder)); +const both = (left: Schema, right: Schema): Schema<[T, U]> => + new Schema( + D.both(left.decoder, right.decoder), + E.both(left.encoder, right.encoder), + ); + function object(def: SchemaDef): Schema { const pdef = {} as DecoderDef; const sdef = {} as EncoderDef; diff --git a/src/lib/mongo.ts b/src/lib/mongo.ts index e2c3f97..f0faabe 100644 --- a/src/lib/mongo.ts +++ b/src/lib/mongo.ts @@ -136,8 +136,18 @@ class Mongo { this.client = new MongoClient(connectionString, values.settings); } + async withTransactionP( + f: (t: MongoTransaction) => Promise, + ): Promise { + return this.withTransaction( + (err) => err, + (t) => Future.attemptP(() => f(t)), + ).promise((err) => err); + } + // Execute an action with a transaction that will be automatically committed at the end. withTransaction( + onError: (e: Error) => E, f: (t: MongoTransaction) => Future, ): Future { const session = this.client.startSession(); @@ -146,11 +156,10 @@ class Mongo { const database = this.client.db(this.values.database); const transaction = new MongoTransaction(session, database); return f(transaction).finally( - Future.create((_, res) => { - if (!transaction.closed) transaction.abort(); - session.endSession(); - return res(); - }), + Future.attemptP(async () => { + if (!transaction.closed) await transaction.abort(); + await session.endSession(); + }).mapRej(onError), ); } } diff --git a/src/lib/postgres.ts b/src/lib/postgres.ts index 4f1ec0b..5a0921c 100644 --- a/src/lib/postgres.ts +++ b/src/lib/postgres.ts @@ -127,14 +127,20 @@ class Postgres { .mapRej(onConnectionError) .chain((connection) => { const transaction = new PostgresTransaction(connection); - return f(transaction).chain((res) => { - if (!transaction.closed) { - Future.attemptP(transaction.commit) - .mapRej(onConnectionError) - .map(() => res); - } - return Future.resolve(res); - }); + return f(transaction).bichain( + (err) => + transaction.closed + ? Future.reject(err) + : Future.attemptP(transaction.abort) + .mapRej(onConnectionError) + .chain((_) => Future.reject(err)), + (res) => + transaction.closed + ? Future.resolve(res) + : Future.attemptP(transaction.commit) + .mapRej(onConnectionError) + .map(() => res), + ); }); } }