Add Worker/Queue to process expired DNS records (#357)

.env.example

@@ -30,3 +30,9 @@ SMTP_PORT=1025
 APP_URL=http://host.docker.internal:8080
 # The SimpleSAML IDP's XML metadata
 SAML_IDP_METADATA_PATH=config/idp-metadata-dev.xml
+
+# Background jobs
+# 24 * 60 * 60 * 1000 = 604800000 (24 hours)
+EXPIRATION_REPEAT_FREQUENCY_MS=86400000
+# 7 * 24 * 60 * 60 * 1000 = 604800000 (7 days)
+JOB_REMOVAL_FREQUENCY_MS=604800000

app/models/record.server.ts

@@ -112,3 +112,16 @@ export function renewDnsRecordById(id: Record['id']) {
     },
   });
 }
+
+export function getExpiredRecords() {
+  return prisma.record.findMany({
+    where: {
+      expiresAt: {
+        lt: new Date(),
+      },
+    },
+    include: {
+      user: true,
+    },
+  });
+}

app/queues/common/expiration-request.server.ts

@@ -0,0 +1,60 @@
+import { Worker, Queue, UnrecoverableError } from 'bullmq';
+
+import { redis } from '~/lib/redis.server';
+import logger from '~/lib/logger.server';
+
+import { getExpiredRecords } from '~/models/record.server';
+import { addNotification } from '../notifications/notifications.server';
+import { deleteDnsRequest } from '../dns/delete-record-flow.server';
+
+const { EXPIRATION_REPEAT_FREQUENCY_MS, JOB_REMOVAL_FREQUENCY_MS } = process.env;
+
+// constant  for removing job on completion/failure (in milliseconds)
+const JOB_REMOVAL_INTERVAL_MS = 7 * 24 * 60 * 60 * 1000; // 7 days
+declare global {
+  var __expiration_request_init__: boolean;
+}
+// queue name
+const expirationRequestQueueName = 'expiration-request';
+
+// queue initialization
+const expirationRequestQueue = new Queue(expirationRequestQueueName, {
+  connection: redis,
+});
+
+export function addExpirationRequest() {
+  return expirationRequestQueue.add(expirationRequestQueueName, {
+    repeat: { every: Number(EXPIRATION_REPEAT_FREQUENCY_MS) || 24 * 60 * 60 * 1000 },
+    removeOnComplete: { age: Number(JOB_REMOVAL_FREQUENCY_MS) || JOB_REMOVAL_INTERVAL_MS },
+    removeOnFail: { age: Number(JOB_REMOVAL_FREQUENCY_MS) || JOB_REMOVAL_INTERVAL_MS },
+  });
+}
+
+// worker definition
+const expirationRequestWorker = new Worker(
+  expirationRequestQueueName,
+  async (job) => {
+    try {
+      logger.info('process DNS record expiration');
+      let dnsRecords = await getExpiredRecords();
+      Promise.all(
+        dnsRecords.map(async ({ id, username, type, subdomain, value, user }) => {
+          // delete records from Route53 and DB
+          await deleteDnsRequest({ id, username, type, subdomain, value });
+          // add notification jobs (assuming deletion went successfully)
+          await addNotification({
+            emailAddress: user.email,
+            subject: 'DNS record expiration subject',
+            message: 'DNS record expiration message',
+          });
+        })
+      );
+    } catch (err) {
+      throw new UnrecoverableError(`Unable to process DNS record expiration: ${err}`);
+    }
+    logger.info('TODO: process certificate expiration');
+  },
+  { connection: redis }
+);
+
+process.on('SIGINT', () => expirationRequestWorker.close());

app/queues/notifications/expiration-notification.server.ts

@@ -14,12 +14,11 @@ import type { NotificationData } from './notifications.server';
 declare global {
   var __expiration_init__: boolean;
 }
-
-enum RecordType {
+export enum RecordType {
   Certificate = 'certificate',
   DnsRecord = 'record',
 }
-interface ExpirationStatusPayload {
+export interface ExpirationStatusPayload {
   type: RecordType;
 }
 // constant for notification frequency in days

app/queues/notifications/notifications.server.ts

@@ -3,13 +3,32 @@ import { Queue, Worker } from 'bullmq';
 import { redis } from '~/lib/redis.server';
 import logger from '~/lib/logger.server';
 import sendNotification from '~/lib/notifications.server';
+import { addExpirationRequest } from '../common/expiration-request.server';
 
 export type NotificationData = {
   emailAddress: string;
   subject: string;
   message: string;
 };
 
+async function init() {
+  try {
+    logger.debug('Expiration Requests init: adding jobs for certificate/record expiration');
+    await addExpirationRequest();
+  } catch (err) {
+    logger.error(`Unable to start expiration notification workers: ${err}`);
+  }
+}
+if (process.env.NODE_ENV === 'production') {
+  init();
+} else {
+  // Only do this setup once in dev
+  if (!global.__expiration_request_init__) {
+    init();
+    global.__expiration_request_init__ = true;
+  }
+}
+
 /**
  * This is the main way callers interact with the notifications
  * queue. It takes care of creating a unique job name.