Turn cronjob into long-running interruptible task

This makes the breach-alerts cronjob explicitly interruptible via
SIGINT, and adds a script to turn it into a long-running task,
instead of a script meant for periodical execution. This should
enable scaling it based on the size of the PubSub queue.

package.json

@@ -13,6 +13,7 @@
     "dev:cron:monthly-activity-free": "tsx --tsconfig tsconfig.cronjobs.json src/scripts/cronjobs/monthlyActivityFree.tsx",
     "dev:cron:monthly-activity-plus": "tsx --tsconfig tsconfig.cronjobs.json src/scripts/cronjobs/monthlyActivityPlus.tsx",
     "dev:cron:breach-alerts": "tsx --tsconfig tsconfig.cronjobs.json src/scripts/cronjobs/emailBreachAlerts.tsx",
+    "dev:task:breach-alerts": "LOOP='true' tsx --tsconfig tsconfig.cronjobs.json src/scripts/cronjobs/emailBreachAlerts.tsx",
     "dev:cron:db-delete-unverified-subscribers": "tsx --tsconfig tsconfig.cronjobs.json src/scripts/cronjobs/deleteUnverifiedSubscribers.ts",
     "dev:cron:db-pull-breaches": "tsx --tsconfig tsconfig.cronjobs.json src/scripts/cronjobs/syncBreaches.ts",
     "dev:cron:db-pull-data-brokers": "tsx --tsconfig tsconfig.cronjobs.json src/scripts/cronjobs/syncOnerepDataBrokers.ts",
@@ -34,6 +35,7 @@
     "cron:monthly-activity-free": "node dist/scripts/cronjobs/monthlyActivityFree.js",
     "cron:monthly-activity-plus": "node dist/scripts/cronjobs/monthlyActivityPlus.js",
     "cron:breach-alerts": "node dist/scripts/cronjobs/emailBreachAlerts.js",
+    "task:breach-alerts": "LOOP='true' node dist/scripts/cronjobs/emailBreachAlerts.js",
     "cron:db-delete-unverified-subscribers": "node dist/scripts/cronjobs/deleteUnverifiedSubscribers.js",
     "cron:db-pull-breaches": "node dist/scripts/cronjobs/syncBreaches.js",
     "cron:db-pull-data-brokers": "node dist/scripts/cronjobs/syncOnerepDataBrokers.js",

src/scripts/cronjobs/emailBreachAlerts.tsx

@@ -4,6 +4,7 @@
 
 import Sentry from "@sentry/nextjs";
 
+import process from "node:process";
 import * as pubsub from "@google-cloud/pubsub";
 import * as grpc from "@grpc/grpc-js";
 import type { SubscriberClient } from "@google-cloud/pubsub/build/src/v1/subscriber_client.js";
@@ -392,11 +393,26 @@ async function pullMessages() {
 
   return [subClient, response.receivedMessages] as const;
 }
-async function init() {
-  await initEmail();
 
+async function run() {
   const [subClient, receivedMessages] = await pullMessages();
   await poll(subClient, receivedMessages ?? []);
+
+  // We're transitioning this into a long-running process,
+  // but to ensure a smooth transition, allow keeping the cronjob behaviour:
+  if (process.env.LOOP === "true") {
+    logger.info("Processed batch, starting next one.");
+    await run();
+  }
+}
+
+async function init() {
+  process.on("SIGINT", () => {
+    throw new Error("SIGINT received, exiting...");
+  });
+  await initEmail();
+
+  await run();
 }
 
 if (process.env.NODE_ENV !== "test") {
@@ -408,7 +424,7 @@ if (process.env.NODE_ENV !== "test") {
         );
       }
     })
-    .catch((err) => console.error(err))
+    .catch((err) => console.error("breach-alerts ended with an error:", err))
     .finally(async () => {
       // Tear down knex connection pools
       await knexSubscribers.destroy();
@@ -420,7 +436,7 @@ if (process.env.NODE_ENV !== "test") {
         monitorSlug: SENTRY_SLUG,
         status: "ok",
       });
-      setTimeout(process.exit, 1000);
+      setTimeout(() => process.exit(0), 1000);
     });
 }
 /* c8 ignore stop */