refactor(webapp): move the postgres task-event write guard to the repository entry points

Guarding at TaskEventStore.createMany stopped the write but the repository still
buffered, reported success, and published Redis notifications for events that
were never persisted. Move the check to the insertMany/insertImmediate/
insertManyImmediate entry points so the postgres store is fully inert when
writes are disabled.

Author: ericallam

apps/webapp/app/v3/eventRepository/eventRepository.server.ts

@@ -120,11 +120,7 @@ export class EventRepository implements IEventRepository {
     this._tracer = _config.tracer ?? trace.getTracer("eventRepo", "0.0.1");
 
     // Instantiate the store using the partitioning flag.
-    this.taskEventStore = new TaskEventStore(
-      db,
-      readReplica,
-      env.EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED
-    );
+    this.taskEventStore = new TaskEventStore(db, readReplica);
   }
 
   #createableEventToPrismaEvent(event: CreateEventInput): Prisma.TaskEventCreateManyInput {
@@ -161,14 +157,23 @@ export class EventRepository implements IEventRepository {
   }
 
   private async insertImmediate(event: CreateEventInput) {
+    if (env.EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED) {
+      return;
+    }
     await this.#flushBatch(nanoid(), [this.#createableEventToPrismaEvent(event)]);
   }
 
   insertMany(events: CreateEventInput[]) {
+    if (env.EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED) {
+      return;
+    }
     this._flushScheduler.addToBatch(events.map(this.#createableEventToPrismaEvent));
   }
 
   async insertManyImmediate(events: CreateEventInput[]) {
+    if (env.EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED) {
+      return;
+    }
     await this.#flushBatchWithReturn(nanoid(), events.map(this.#createableEventToPrismaEvent));
   }
 

apps/webapp/app/v3/taskEventStore.server.ts

@@ -59,17 +59,13 @@ export function getTaskEventStore(): TaskEventStoreTable {
 export class TaskEventStore {
   constructor(
     private db: PrismaClient,
-    private readReplica: PrismaReplicaClient,
-    private writesDisabled: boolean = false
+    private readReplica: PrismaReplicaClient
   ) {}
 
   /**
    * Insert one record.
    */
   async create(table: TaskEventStoreTable, data: Prisma.TaskEventCreateInput) {
-    if (this.writesDisabled) {
-      return;
-    }
     if (table === "taskEventPartitioned") {
       return await this.db.taskEventPartitioned.create({ data });
     } else {
@@ -81,9 +77,6 @@ export class TaskEventStore {
    * Insert many records.
    */
   async createMany(table: TaskEventStoreTable, data: Prisma.TaskEventCreateManyInput[]) {
-    if (this.writesDisabled) {
-      return { count: 0 };
-    }
     if (table === "taskEventPartitioned") {
       return await this.db.taskEventPartitioned.createMany({ data });
     } else {