Add Worker/Queue to process expired DNS records

.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.