diff --git a/src/app/handleCommand.ts b/src/app/handleCommand.ts index d76f44c..ce88d06 100644 --- a/src/app/handleCommand.ts +++ b/src/app/handleCommand.ts @@ -9,7 +9,7 @@ import { Future } from '@/lib/Future'; import { Result, Failure } from '@/lib/Result'; import { Repositories, Projections, allProjections } from '@/app/projections'; import { Services } from '@/app/integrations'; -import { WithProjectionStore } from '@/app/projectionStore'; +import { WithProjectionStore, Mode } from '@/app/projectionStore'; import { WithEventStore } from '@/app/eventStore'; type CommandHandler = (v: { @@ -39,7 +39,7 @@ function handleCommand( ): express.Handler { return router.route((req) => decodeCommand(decoder, req).chain((command) => - withProjectionStore(onStoreError, (projectionStore) => + withProjectionStore(onStoreError, Mode.ReadOnly, (projectionStore) => withEventStore(onStoreError, (store) => handler({ command, diff --git a/src/app/handleProjection.ts b/src/app/handleProjection.ts index ef7884c..4ac001c 100644 --- a/src/app/handleProjection.ts +++ b/src/app/handleProjection.ts @@ -19,6 +19,7 @@ import { Projections, Repositories, allProjections } from '@/app/projections'; import { MongoProjectionStore, WithProjectionStore, + Mode, } from '@/app/projectionStore'; type ProjectionHandler = (v: { @@ -44,7 +45,7 @@ function handleProjection>( return router.route((req) => decodeEvent(decoder, req) .chain(({ event, info }) => - withProjectionStore(onProjectionStoreError, (store) => + withProjectionStore(onProjectionStoreError, Mode.ReadWrite, (store) => handler({ event, info, diff --git a/src/app/handleQuery.ts b/src/app/handleQuery.ts index 670392c..3e145cb 100644 --- a/src/app/handleQuery.ts +++ b/src/app/handleQuery.ts @@ -7,7 +7,7 @@ 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 { WithProjectionStore, Mode } from '@/app/projectionStore'; import { internalServerError } from '@/app/responses'; type QueryHandler = (v: { @@ -29,7 +29,7 @@ function handleQuery( ): express.Handler { return router.route((req) => decodeQuery(decoder, req).chain((query) => - withProjectionStore(onProjectionStoreError, (store) => + withProjectionStore(onProjectionStoreError, Mode.ReadOnly, (store) => handler({ query, projections: allProjections(repositories, store), diff --git a/src/app/handleReaction.ts b/src/app/handleReaction.ts index d96049c..9aac529 100644 --- a/src/app/handleReaction.ts +++ b/src/app/handleReaction.ts @@ -15,9 +15,9 @@ 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 = {}; +import { Services } from '@/app/integrations'; +import { Repositories, Projections, allProjections } from '@/app/projections'; +import { WithProjectionStore, Mode } from '@/app/projectionStore'; type ReactionController> = { decoder: Decoder>; @@ -32,7 +32,7 @@ type ReactionHandler = (v: { store: EventStore; }) => Future; -const onEventStoreError = (err: Error) => new Ambar.ErrorMustRetry(err.message); +const onStoreError = (err: Error) => new Ambar.ErrorMustRetry(err.message); type WithStoreGeneric = ( onError: (e: Error) => E, @@ -47,31 +47,34 @@ const wrapWithEventStore = ( withEventStore: WithStoreGeneric, ): WithStoreConcrete => function (f) { - return withEventStore(onEventStoreError, (store) => f(store)); + return withEventStore(onStoreError, (store) => f(store)); }; function handleReaction>( withEventStore: WithStoreConcrete, - projections: Projections, + withProjectionStore: WithProjectionStore, services: Services, + repositories: Repositories, { 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), + withProjectionStore(onStoreError, Mode.ReadOnly, (projectionStore) => + withEventStore((store) => + handler({ + event, + info, + projections: allProjections(repositories, projectionStore), + services, + store, + }).chainRej((r) => + r instanceof Ambar.Success + ? Future.resolve(undefined) + : r instanceof Ambar.ErrorMustRetry + ? Future.reject(r) + : (r satisfies never), + ), ), ), ) diff --git a/src/app/integrations.ts b/src/app/integrations.ts index bf802c2..3ec8eb3 100644 --- a/src/app/integrations.ts +++ b/src/app/integrations.ts @@ -6,6 +6,7 @@ import { Mongo } from '@/lib/mongo'; import { MongoProjectionStore, WithProjectionStore, + Mode, } from '@/app/projectionStore'; import { EmailService } from '@/app/services/email'; import { FileStorageService } from '@/app/services/file-storage'; @@ -77,11 +78,11 @@ async function configureDependencies(): Promise { f(new PostgresEventStore(t, schemas, table)), ); - const withProjectionStore: WithProjectionStore = (onError, f) => - mongo.withTransaction(onError, (t) => f(new MongoProjectionStore(t))); + const withProjectionStore: WithProjectionStore = (onError, mode, f) => + mongo.withTransaction(onError, (t) => f(new MongoProjectionStore(t, mode))); const repositories = await mongo.withTransactionP(async (t) => - initializeRepositories(new MongoProjectionStore(t)), + initializeRepositories(new MongoProjectionStore(t, Mode.ReadOnly)), ); const services = initializeServices(); diff --git a/src/app/projectionStore.ts b/src/app/projectionStore.ts index bb53ad7..d5cb36b 100644 --- a/src/app/projectionStore.ts +++ b/src/app/projectionStore.ts @@ -5,6 +5,7 @@ export { type JsonDoc, type RepositoryArgs, type WithProjectionStore, + Mode, }; import { Collection } from 'mongodb'; @@ -53,11 +54,20 @@ const schemaIdAndValue = (schema: Schema): Schema> => { type WithProjectionStore = ( onError: (e: Error) => E, + mode: Mode, f: (s: MongoProjectionStore) => Future, ) => Future; +enum Mode { + ReadOnly = 'ReadOnly', + ReadWrite = 'ReadWrite', +} + class MongoProjectionStore { - constructor(private transaction: MongoTransaction) {} + constructor( + private transaction: MongoTransaction, + private readonly readOnly: Mode, + ) {} // Initialize a repository, creating the collection and indexes if needed. async createRepository(args: RepositoryArgs): Promise> { @@ -122,6 +132,9 @@ class MongoProjectionStore { document: T, options?: InsertOneOptions, ): Promise { + if (this.readOnly == Mode.ReadOnly) { + throw new Error('Trying to write to read-only projection store'); + } const schema = schemaIdAndValue(repository.values.schema); const _id = repository.values.toId(document); await this.transaction.insertOne( @@ -137,6 +150,9 @@ class MongoProjectionStore { document: T, options: InsertOneOptions = {}, ): Promise { + if (this.readOnly == Mode.ReadOnly) { + throw new Error('Trying to write to read-only projection store'); + } const schema = schemaIdAndValue(repository.values.schema); const _id = repository.values.toId(document); await this.transaction.replaceOne( diff --git a/src/index.ts b/src/index.ts index a8362fa..8c0a86f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -46,6 +46,7 @@ async function main() { endpoint, handleReaction( wrapWithEventStore(withEventStore), + withProjectionStore, services, repositories, controller,