Add ability to disable metadata flushing and logging (#1587)

Author: ericallam

apps/webapp/app/env.server.ts

@@ -52,7 +52,7 @@ const EnvironmentSchema = z.object({
   SMTP_PORT: z.coerce.number().optional(),
   SMTP_SECURE: z.coerce.boolean().optional(),
   SMTP_USER: z.string().optional(),
-  SMTP_PASSWORD:  z.string().optional(),
+  SMTP_PASSWORD: z.string().optional(),
 
   PLAIN_API_KEY: z.string().optional(),
   RUNTIME_PLATFORM: z.enum(["docker-compose", "ecs", "local"]).default("local"),
@@ -210,8 +210,7 @@ const EnvironmentSchema = z.object({
   ALERT_SMTP_PORT: z.coerce.number().optional(),
   ALERT_SMTP_SECURE: z.coerce.boolean().optional(),
   ALERT_SMTP_USER: z.string().optional(),
-  ALERT_SMTP_PASSWORD:  z.string().optional(),
-
+  ALERT_SMTP_PASSWORD: z.string().optional(),
 
   MAX_SEQUENTIAL_INDEX_FAILURE_COUNT: z.coerce.number().default(96),
 
@@ -261,6 +260,8 @@ const EnvironmentSchema = z.object({
 
   REALTIME_STREAM_VERSION: z.enum(["v1", "v2"]).default("v1"),
   BATCH_METADATA_OPERATIONS_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
+  BATCH_METADATA_OPERATIONS_FLUSH_ENABLED: z.string().default("1"),
+  BATCH_METADATA_OPERATIONS_FLUSH_LOGGING_ENABLED: z.string().default("1"),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/services/metadata/updateMetadata.server.ts

@@ -30,7 +30,9 @@ export class UpdateMetadataService extends BaseService {
 
   constructor(
     protected readonly _prisma: PrismaClientOrTransaction = prisma,
-    private readonly flushIntervalMs: number = 5000
+    private readonly flushIntervalMs: number = 5000,
+    private readonly flushEnabled: boolean = true,
+    private readonly flushLoggingEnabled: boolean = true
   ) {
     super();
 
@@ -39,6 +41,14 @@ export class UpdateMetadataService extends BaseService {
 
   // Start a loop that periodically flushes buffered operations
   private _startFlushing() {
+    if (!this.flushEnabled) {
+      logger.info("[UpdateMetadataService] 🚽 Flushing disabled");
+
+      return;
+    }
+
+    logger.info("[UpdateMetadataService] 🚽 Flushing started");
+
     // Create a program that sleeps, then processes buffered ops
     const program = Effect.gen(this, function* (_) {
       while (true) {
@@ -50,9 +60,11 @@ export class UpdateMetadataService extends BaseService {
         this._bufferedOperations.clear();
 
         yield* Effect.sync(() => {
-          logger.debug(`[UpdateMetadataService] Flushing operations`, {
-            operations: Object.fromEntries(currentOperations),
-          });
+          if (this.flushLoggingEnabled) {
+            logger.debug(`[UpdateMetadataService] Flushing operations`, {
+              operations: Object.fromEntries(currentOperations),
+            });
+          }
         });
 
         // If we have operations, process them
@@ -87,10 +99,12 @@ export class UpdateMetadataService extends BaseService {
         }
 
         yield* Effect.sync(() => {
-          logger.debug(`[UpdateMetadataService] Processing operations for run`, {
-            runId,
-            operationsCount: processedOps.length,
-          });
+          if (this.flushLoggingEnabled) {
+            logger.debug(`[UpdateMetadataService] Processing operations for run`, {
+              runId,
+              operationsCount: processedOps.length,
+            });
+          }
         });
 
         // Update run with retry
@@ -410,5 +424,11 @@ export class UpdateMetadataService extends BaseService {
 
 export const updateMetadataService = singleton(
   "update-metadata-service",
-  () => new UpdateMetadataService(prisma, env.BATCH_METADATA_OPERATIONS_FLUSH_INTERVAL_MS)
+  () =>
+    new UpdateMetadataService(
+      prisma,
+      env.BATCH_METADATA_OPERATIONS_FLUSH_INTERVAL_MS,
+      env.BATCH_METADATA_OPERATIONS_FLUSH_ENABLED === "1",
+      env.BATCH_METADATA_OPERATIONS_FLUSH_LOGGING_ENABLED === "1"
+    )
 );