Move to reporting worker stats through logs

Author: ericallam

apps/webapp/app/env.server.ts

@@ -40,7 +40,6 @@ const EnvironmentSchema = z.object({
   EXECUTION_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
   WORKER_ENABLED: z.string().default("true"),
   EXECUTION_WORKER_ENABLED: z.string().default("true"),
-  WORKER_REPORTER_EMAIL: z.string().email().optional(),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/platform/zodWorker.server.ts

@@ -91,6 +91,8 @@ export type ZodWorkerCleanupOptions = {
   taskOptions?: CronItemOptions;
 };
 
+type ZodWorkerReporter = (event: string, properties: Record<string, any>) => Promise<void>;
+
 export type ZodWorkerOptions<TMessageCatalog extends MessageCatalogSchema> = {
   name: string;
   runnerOptions: RunnerOptions;
@@ -99,7 +101,7 @@ export type ZodWorkerOptions<TMessageCatalog extends MessageCatalogSchema> = {
   tasks: ZodTasks<TMessageCatalog>;
   recurringTasks?: ZodRecurringTasks;
   cleanup?: ZodWorkerCleanupOptions;
-  reporter?: (subject: string, message: string) => Promise<void>;
+  reporter?: ZodWorkerReporter;
 };
 
 export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
@@ -111,7 +113,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
   #recurringTasks?: ZodRecurringTasks;
   #runner?: GraphileRunner;
   #cleanup: ZodWorkerCleanupOptions | undefined;
-  #reporter?: (subject: string, message: string) => Promise<void>;
+  #reporter?: ZodWorkerReporter;
 
   constructor(options: ZodWorkerOptions<TMessageCatalog>) {
     this.#name = options.name;
@@ -528,10 +530,11 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
     });
 
     if (this.#reporter) {
-      await this.#reporter(
-        "Worker Queue Cleanup",
-        `Cleaned up ${results.length} jobs older than ${expirationDate.toISOString()}`
-      );
+      await this.#reporter("cleanup_stats", {
+        count: results.length,
+        expirationDate,
+        ts: payload._cron.ts,
+      });
     }
   }
 
@@ -581,10 +584,11 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       payload,
     });
 
-    await this.#reporter(
-      "Worker Queue Metrics",
-      `Added ${addedCountResults.count} jobs in the last hour, total jobs: ${totalCountResults.count}`
-    );
+    await this.#reporter("queue_metrics", {
+      addedCount: addedCountResults.count,
+      totalCount: totalCountResults.count,
+      ts: payload._cron.ts,
+    });
   }
 
   #logDebug(message: string, args?: any) {

apps/webapp/app/services/worker.server.ts

@@ -134,16 +134,8 @@ function getWorkerQueue() {
       ttl: 7 * 24 * 60 * 60 * 1000, // 7 days
       maxCount: 1000,
     },
-    reporter: async (subject, message) => {
-      logger.info("workerQueue reporter", { workerMessage: message, subject });
-
-      if (env.WORKER_REPORTER_EMAIL) {
-        await sendPlainTextEmail({
-          to: env.WORKER_REPORTER_EMAIL,
-          subject: `[${env.APP_ENV}] workerQueue Report: ${subject}`,
-          text: message,
-        });
-      }
+    reporter: async (event, properties) => {
+      logger.info("workerQueue report", { event, properties });
     },
     runnerOptions: {
       connectionString: env.DATABASE_URL,