feat: add admin email domain management and sync job (#2538)

packages/lib/jobs/client.ts

@@ -16,6 +16,7 @@ import { EXECUTE_WEBHOOK_JOB_DEFINITION } from './definitions/internal/execute-w
 import { EXPIRE_RECIPIENTS_SWEEP_JOB_DEFINITION } from './definitions/internal/expire-recipients-sweep';
 import { PROCESS_RECIPIENT_EXPIRED_JOB_DEFINITION } from './definitions/internal/process-recipient-expired';
 import { SEAL_DOCUMENT_JOB_DEFINITION } from './definitions/internal/seal-document';
+import { SYNC_EMAIL_DOMAINS_JOB_DEFINITION } from './definitions/internal/sync-email-domains';
 
 /**
  * The `as const` assertion is load bearing as it provides the correct level of type inference for
@@ -39,6 +40,7 @@ export const jobsClient = new JobClient([
   EXPIRE_RECIPIENTS_SWEEP_JOB_DEFINITION,
   PROCESS_RECIPIENT_EXPIRED_JOB_DEFINITION,
   CLEANUP_RATE_LIMITS_JOB_DEFINITION,
+  SYNC_EMAIL_DOMAINS_JOB_DEFINITION,
 ] as const);
 
 export const jobs = jobsClient;

packages/lib/jobs/client/inngest.ts

@@ -26,6 +26,7 @@ export class InngestJobProvider extends BaseJobProvider {
       const client = new InngestClient({
         id: env('NEXT_PRIVATE_INNGEST_APP_ID') || 'documenso-app',
         eventKey: env('INNGEST_EVENT_KEY') || env('NEXT_PRIVATE_INNGEST_EVENT_KEY'),
+        logger: console,
       });
 
       this._instance = new InngestJobProvider({ client });
@@ -90,7 +91,10 @@ export class InngestJobProvider extends BaseJobProvider {
     return {
       wait: step.sleep,
       logger: {
-        ...ctx.logger,
+        info: ctx.logger.info,
+        debug: ctx.logger.debug,
+        error: ctx.logger.error,
+        warn: ctx.logger.warn,
         log: ctx.logger.info,
       },
       runTask: async (cacheKey, callback) => {

packages/lib/jobs/definitions/internal/sync-email-domains.handler.ts

@@ -0,0 +1,86 @@
+import { DateTime } from 'luxon';
+
+import { reregisterEmailDomain } from '@documenso/ee/server-only/lib/reregister-email-domain';
+import { verifyEmailDomain } from '@documenso/ee/server-only/lib/verify-email-domain';
+import { prisma } from '@documenso/prisma';
+
+import type { JobRunIO } from '../../client/_internal/job';
+import type { TSyncEmailDomainsJobDefinition } from './sync-email-domains';
+
+const BATCH_SIZE = 10;
+const AUTO_REREGISTER_AFTER_HOURS = 48;
+
+export const run = async ({ io }: { payload: TSyncEmailDomainsJobDefinition; io: JobRunIO }) => {
+  const pendingDomains = await prisma.emailDomain.findMany({
+    where: {
+      status: 'PENDING',
+    },
+    select: {
+      id: true,
+      domain: true,
+      createdAt: true,
+      lastVerifiedAt: true,
+    },
+    orderBy: {
+      lastVerifiedAt: { sort: 'asc', nulls: 'first' },
+    },
+  });
+
+  if (pendingDomains.length === 0) {
+    io.logger.info('No pending email domains to sync');
+    return;
+  }
+
+  io.logger.info(`Found ${pendingDomains.length} pending email domains to sync`);
+
+  let verifiedCount = 0;
+  let reregisteredCount = 0;
+  let errorCount = 0;
+
+  const reregisterCutoff = DateTime.now().minus({ hours: AUTO_REREGISTER_AFTER_HOURS }).toJSDate();
+
+  for (let i = 0; i < pendingDomains.length; i += BATCH_SIZE) {
+    const batch = pendingDomains.slice(i, i + BATCH_SIZE);
+
+    const results = await Promise.allSettled(
+      batch.map(async (domain) => {
+        const shouldReregister = domain.createdAt < reregisterCutoff;
+
+        if (shouldReregister) {
+          io.logger.info(
+            `Domain "${domain.domain}" has been pending since ${domain.createdAt.toISOString()}, attempting re-registration`,
+          );
+
+          await reregisterEmailDomain({ emailDomainId: domain.id });
+          return 'reregistered' as const;
+        }
+
+        const { isVerified } = await verifyEmailDomain(domain.id);
+
+        return isVerified ? ('verified' as const) : ('pending' as const);
+      }),
+    );
+
+    for (const result of results) {
+      if (result.status === 'rejected') {
+        errorCount++;
+        io.logger.error(`Failed to process email domain: ${String(result.reason)}`);
+      } else if (result.value === 'verified') {
+        verifiedCount++;
+      } else if (result.value === 'reregistered') {
+        reregisteredCount++;
+      }
+    }
+
+    // Small delay between batches to respect SES API rate limits.
+    if (i + BATCH_SIZE < pendingDomains.length) {
+      await new Promise<void>((resolve) => {
+        setTimeout(resolve, 1000);
+      });
+    }
+  }
+
+  io.logger.info(
+    `Sync complete: ${verifiedCount} verified, ${reregisteredCount} re-registered, ${errorCount} errors out of ${pendingDomains.length} pending domains`,
+  );
+};

packages/lib/jobs/definitions/internal/sync-email-domains.ts

@@ -0,0 +1,30 @@
+import { z } from 'zod';
+
+import { type JobDefinition } from '../../client/_internal/job';
+
+const SYNC_EMAIL_DOMAINS_JOB_DEFINITION_ID = 'internal.sync-email-domains';
+
+const SYNC_EMAIL_DOMAINS_JOB_DEFINITION_SCHEMA = z.object({});
+
+export type TSyncEmailDomainsJobDefinition = z.infer<
+  typeof SYNC_EMAIL_DOMAINS_JOB_DEFINITION_SCHEMA
+>;
+
+export const SYNC_EMAIL_DOMAINS_JOB_DEFINITION = {
+  id: SYNC_EMAIL_DOMAINS_JOB_DEFINITION_ID,
+  name: 'Sync Email Domains',
+  version: '1.0.0',
+  trigger: {
+    name: SYNC_EMAIL_DOMAINS_JOB_DEFINITION_ID,
+    schema: SYNC_EMAIL_DOMAINS_JOB_DEFINITION_SCHEMA,
+    cron: '0 * * * *', // Every hour, on the hour.
+  },
+  handler: async ({ payload, io }) => {
+    const handler = await import('./sync-email-domains.handler');
+
+    await handler.run({ payload, io });
+  },
+} as const satisfies JobDefinition<
+  typeof SYNC_EMAIL_DOMAINS_JOB_DEFINITION_ID,
+  TSyncEmailDomainsJobDefinition
+>;

packages/lib/types/email-domain.ts

@@ -18,6 +18,7 @@ export const ZEmailDomainSchema = EmailDomainSchema.pick({
   publicKey: true,
   createdAt: true,
   updatedAt: true,
+  lastVerifiedAt: true,
 }).extend({
   emails: ZOrganisationEmailLiteSchema.array(),
 });
@@ -35,6 +36,7 @@ export const ZEmailDomainManySchema = EmailDomainSchema.pick({
   selector: true,
   createdAt: true,
   updatedAt: true,
+  lastVerifiedAt: true,
 });
 
 export type TEmailDomainMany = z.infer<typeof ZEmailDomainManySchema>;