Refactor dns flows and workers

app/lib/record-db.server.ts

@@ -0,0 +1,55 @@
+import type { Record } from '@prisma/client';
+import {
+ createRecord,
+ deleteRecordById,
+ doesRecordExist,
+ getUserRecordCount,
+ updateRecordById,
+} from '~/models/record.server';
+
+export interface RecordCreation {
+ username: Record['username'];
+ type: Record['type'];
+ name: Record['name'];
+ value: Record['value'];
+}
+
+export interface RecordUpdate {
+ id: Record['id'];
+ type: Record['type'];
+ name: Record['name'];
+ value: Record['value'];
+ username?: Record['username'];
+ description?: Record['description'];
+ course?: Record['course'];
+ ports?: Record['ports'];
+}
+
+export const createUserRecord = async (data: RecordCreation) => {
+ if (await doesRecordExist(data)) {
+ throw new Error('Record already exists');
+ }
+
+ const result = await createRecord(data);
+
+ if (!result) {
+ throw new Error('Could not create a record in DB');
+ }
+ return result.id;
+};
+
+export const updateUserRecord = async (data: RecordUpdate) => {
+ const result = await updateRecordById(data);
+
+ if (!result) {
+ throw new Error('Could not update the record in DB');
+ }
+};
+
+export const deleteUserRecord = async (id: Record['id']) => {
+ const result = await deleteRecordById(id);
+
+ if (!result) {
+ throw new Error('Could not delete the record in DB');
+ }
+};

app/models/record.server.ts

@@ -1,59 +1,47 @@
 import { RecordStatus } from '@prisma/client';
-import { prisma } from '~/db.server';
 import dayjs from 'dayjs';
+import { prisma } from '~/db.server';
 
 import type { Record } from '@prisma/client';
+import type { RecordCreation, RecordUpdate } from '~/lib/record-db.server';
 export type { Record } from '@prisma/client';
 
-export async function getRecordsByUsername(username: Record['username']) {
+export function getRecordsByUsername(username: Record['username']) {
  return prisma.record.findMany({ where: { username } });
 }
 
-export async function getRecordById(id: Record['id']) {
+export function getRecordById(id: Record['id']) {
  return prisma.record.findUnique({ where: { id } });
 }
 
-export async function createRecord(data: Pick<Record, 'username' | 'type' | 'name' | 'value'>) {
+export function getUserRecordCount(username: Record['username']) {
+ return prisma.record.count({
+ where: {
+ username,
+ },
+ });
+}
+
+export function createRecord(data: Pick<Record, 'username' | 'type' | 'name' | 'value'>) {
  // Set expiration date 6 months from now
  const expiresAt = dayjs().set('month', 6).toDate();
  const status = RecordStatus.pending;
 
  return prisma.record.create({ data: { ...data, expiresAt, status } });
 }
 
-export async function updateRecordById(
- id: Record['id'],
- type?: Record['type'],
- name?: Record['name'],
- value?: Record['value'],
- status?: Record['status'],
- username?: Record['username'],
- description?: Record['description'],
- course?: Record['course'],
- ports?: Record['ports'],
- expiresAt?: Record['expiresAt'],
- lastNotified?: Record['lastNotified']
-) {
+export function updateRecordById(data: RecordUpdate) {
  return prisma.record.update({
- where: { id },
+ where: { id: data.id },
  data: {
- type,
- name,
- value,
- status,
- username,
- description,
- course,
- ports,
- expiresAt,
- lastNotified,
+ ...data,
+ expiresAt: dayjs().set('month', 6).toDate(),
  },
  });
 }
 
-export async function updateRecordStatusById(id: Record['id'], status: Record['status']) {
+export function updateRecordStatusById(id: Record['id'], status: Record['status']) {
  const expireToSet = dayjs().set('month', 6).toDate();
-
  return prisma.record.update({
  where: {
  id,
@@ -65,7 +53,18 @@ export async function updateRecordStatusById(id: Record['id'], status: Record['s
  });
 }
 
-export async function doesRecordExist(data: Pick<Record, 'username' | 'type' | 'name' | 'value'>) {
+export function renewRecordById(id: Record['id']) {
+ return prisma.record.update({
+ where: {
+ id,
+ },
+ data: {
+ expiresAt: dayjs().set('month', 6).toDate(),
+ },
+ });
+}
+
+export async function doesRecordExist(data: RecordCreation) {
  const { username, type, name, value } = data;
  const count = await prisma.record.count({
  where: {
@@ -79,6 +78,6 @@ export async function doesRecordExist(data: Pick<Record, 'username' | 'type' | '
  return count > 0;
 }
 
-export async function deleteRecordById(id: Record['id']) {
+export function deleteRecordById(id: Record['id']) {
  return prisma.record.delete({ where: { id } });
 }

app/queues/certificate/order-creator-worker.server.ts

@@ -4,7 +4,7 @@ import logger from '~/lib/logger.server';
 import LetsEncrypt from '~/lib/lets-encrypt.server';
 import * as certificateModel from '~/models/certificate.server';
 import * as challengeModel from '~/models/challenge.server';
-import { addDnsRequest } from '~/queues/dns/dns-flow.server';
+import { addDnsRequest } from '~/queues/dns/add-record-flow.server';
 
 import type { ChallengeBundle } from '~/lib/lets-encrypt.server';
 
@@ -40,8 +40,9 @@ const handleChallenges = ({
  await addDnsRequest({
  username,
  type: 'TXT',
- name: domain,
+ subdomain: domain,
  value: challengeKey,
+ id: 1, //This ID should to be picked up
  });
 
  return challengeModel.createChallenge({

app/queues/dns/add-record-flow.server.ts

@@ -0,0 +1,91 @@
+import { FlowProducer } from 'bullmq';
+import { redis } from '~/lib/redis.server';
+import { buildDomain } from '~/utils';
+import { dnsRecordUpdateQueueName } from './workers/dns-record-worker.server';
+import { checkDnsStatusQueueName } from './workers/poll-dns-status-worker.server';
+import { syncDbStatusQueueName } from './workers/sync-db-status-worker.server';
+
+import type { FlowJob } from 'bullmq';
+import type { Record } from '@prisma/client';
+import type { DnsRecordUpdaterData } from './workers/dns-record-worker.server';
+import type { DbRecordSynchronizerData } from './workers/sync-db-status-worker.server';
+
+export interface DnsRecordUpdateRequest {
+ username: Record['username'];
+ type: Record['type'];
+ subdomain: Record['name'];
+ value: Record['value'];
+ id: Record['id'];
+}
+
+export enum WorkType {
+ create = 'create',
+ update = 'update',
+ delete = 'delete',
+}
+
+const flowProducer = new FlowProducer({ connection: redis });
+
+export const addDnsRequest = async ({
+ username,
+ type,
+ subdomain,
+ value,
+ id,
+}: DnsRecordUpdateRequest) => {
+ const fullRecordName = buildDomain(username, subdomain);
+
+ // Step 1. Request Route53 to create a record
+ const createDnsRecord: FlowJob = {
+ name: `createDnsRecord:${subdomain}-${username}`,
+ queueName: dnsRecordUpdateQueueName,
+ data: {
+ workType: WorkType.create,
+ username,
+ type,
+ fqdn: fullRecordName,
+ value,
+ } as DnsRecordUpdaterData,
+ opts: {
+ failParentOnFailure: true,
+ attempts: 5,
+ backoff: {
+ type: 'exponential',
+ delay: 15_000,
+ },
+ },
+ };
+
+ // Step 2. Poll Route53 to check connection status of the domain until it's ready
+ const checkDnsStatus: FlowJob = {
+ name: `checkDnsStatus:${subdomain}-${username}`,
+ queueName: checkDnsStatusQueueName,
+ children: [createDnsRecord],
+ opts: {
+ failParentOnFailure: true,
+ attempts: 5,
+ backoff: {
+ type: 'exponential',
+ delay: 60_000,
+ },
+ },
+ };
+
+ // Step 3. Update the MySQL record with the active or error status
+ const syncDbStatus: FlowJob = {
+ name: `syncDbStatus:${subdomain}-${username}`,
+ queueName: syncDbStatusQueueName,
+ children: [checkDnsStatus],
+ data: { workType: WorkType.create, id } as DbRecordSynchronizerData,
+ opts: {
+ failParentOnFailure: true,
+ attempts: 5,
+ backoff: {
+ type: 'exponential',
+ delay: 30_000,
+ },
+ },
+ };
+
+ return flowProducer.add(syncDbStatus);
+};

app/queues/dns/delete-record-flow.server.ts

@@ -0,0 +1,78 @@
+import { FlowProducer } from 'bullmq';
+import { redis } from '~/lib/redis.server';
+import { buildDomain } from '~/utils';
+import { dnsRecordUpdateQueueName } from './workers/dns-record-worker.server';
+import { checkDnsStatusQueueName } from './workers/poll-dns-status-worker.server';
+import { syncDbStatusQueueName } from './workers/sync-db-status-worker.server';
+import { WorkType } from './add-record-flow.server';
+
+import type { FlowJob } from 'bullmq';
+import type { DnsRecordUpdateRequest } from './add-record-flow.server';
+import type { DnsRecordUpdaterData } from './workers/dns-record-worker.server';
+import type { DbRecordSynchronizerData } from './workers/sync-db-status-worker.server';
+
+const flowProducer = new FlowProducer({ connection: redis });
+
+export const deleteDnsRequest = async ({
+ username,
+ type,
+ subdomain,
+ value,
+ id,
+}: DnsRecordUpdateRequest) => {
+ const fullRecordName = buildDomain(username, subdomain);
+
+ // Step 1. Request Route53 to delete the record
+ const updateDnsRecord: FlowJob = {
+ name: `deleteDnsRecord:${subdomain}-${username}`,
+ queueName: dnsRecordUpdateQueueName,
+ data: {
+ workType: WorkType.delete,
+ username,
+ type,
+ fqdn: fullRecordName,
+ value,
+ } as DnsRecordUpdaterData,
+ opts: {
+ failParentOnFailure: true,
+ attempts: 5,
+ backoff: {
+ type: 'exponential',
+ delay: 15_000,
+ },
+ },
+ };
+
+ // Step 2. Poll Route53 to check connection status of the domain until it's ready
+ const checkDnsStatus: FlowJob = {
+ name: `checkDnsStatus:${subdomain}-${username}`,
+ queueName: checkDnsStatusQueueName,
+ children: [updateDnsRecord],
+ opts: {
+ failParentOnFailure: true,
+ attempts: 5,
+ backoff: {
+ type: 'exponential',
+ delay: 60_000,
+ },
+ },
+ };
+
+ // Step 3. Delete the record in MySQL
+ const updateDbRecord: FlowJob = {
+ name: `deleteDbRecord:${subdomain}-${username}`,
+ queueName: syncDbStatusQueueName,
+ data: { workType: WorkType.delete, id } as DbRecordSynchronizerData,
+ children: [checkDnsStatus],
+ opts: {
+ failParentOnFailure: true,
+ attempts: 5,
+ backoff: {
+ type: 'exponential',
+ delay: 15_000,
+ },
+ },
+ };
+
+ return flowProducer.add(updateDbRecord);
+};