fix(script-executor): schedule member email deduplication sweep

services/apps/script_executor_worker/src/main.ts

@@ -7,6 +7,7 @@ import {
   scheduleOrganizationSegmentAggCleanup,
   scheduleOrganizationsCleanup,
 } from './schedules/scheduleCleanup'
+import { scheduleMergeMembersWithSameVerifiedEmails } from './schedules/scheduleMemberDeduplication'
 
 const config: Config = {
   envvars: [
@@ -47,6 +48,7 @@ setImmediate(async () => {
   await scheduleOrganizationsCleanup()
   await scheduleMemberSegmentsAggCleanup()
   await scheduleOrganizationSegmentAggCleanup()
+  await scheduleMergeMembersWithSameVerifiedEmails()
 
   await svc.start()
 })

services/apps/script_executor_worker/src/schedules/scheduleMemberDeduplication.ts

@@ -0,0 +1,43 @@
+import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client'
+
+import { svc } from '../main'
+import { findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms } from '../workflows/findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms'
+
+export const scheduleMergeMembersWithSameVerifiedEmails = async () => {
+  try {
+    await svc.temporal.schedule.create({
+      scheduleId: 'mergeMembersWithSameVerifiedEmails',
+      spec: {
+        cronExpressions: ['0 3 * * 0'],
+      },
+      policies: {
+        // The workflow walks the whole memberIdentities table via continueAsNew, so a single
+        // sweep can outlast the interval. Skip an overdue run instead of buffering it, so
+        // sweeps never stack up on top of each other.
+        overlap: ScheduleOverlapPolicy.SKIP,
+        catchupWindow: '1 minute',
+      },
+      action: {
+        type: 'startWorkflow',
+        workflowType: findAndMergeMembersWithSameVerifiedEmailsInDifferentPlatforms,
+        taskQueue: 'script-executor',
+        retry: {
+          initialInterval: '15 seconds',
+          backoffCoefficient: 2,
+          maximumAttempts: 3,
+        },
+        // Start from the beginning of the hash ordering; the workflow pages itself from there.
+        args: [{}],
+      },
+    })
+    svc.log.info('Schedule for merging members with same verified emails created successfully!')
+  } catch (err) {
+    if (err instanceof ScheduleAlreadyRunning) {
+      svc.log.info('Schedule mergeMembersWithSameVerifiedEmails already registered in Temporal.')
+      svc.log.info('Configuration may have changed since. Please make sure they are in sync.')
+    } else {
+      svc.log.error({ err }, 'Error creating schedule for member email deduplication')
+      throw new Error(err)
+    }
+  }
+}