Transactionally start a Temporal workflow when data changes in your database

export async function cleanUpExternalSystems(userId: string): Promise<void> {
  console.log(`[*] Performing external cleanup for user ${userId}...`);
  // Simulated delay to represent real API calls
  await new Promise(resolve => setTimeout(resolve, 1000));
}

export async function sendDeletionConfirmation(email: string): Promise<void> {
  console.log(`[*] Sending account deletion confirmation to ${email}...`);
  // Simulated delay to represent sending a real notification
  await new Promise(resolve => setTimeout(resolve, 500));
}

import { defineSignal, condition, proxyActivities, setHandler } from '@temporalio/workflow';
import * as activities from './activities';

// Configure activities for use in the Workflow
const { cleanUpExternalSystems, sendDeletionConfirmation } = proxyActivities<typeof activities>({
  startToCloseTimeout: '1 minute',
  retry: { maximumAttempts: 5 }
});

export const userDeletedSignal = defineSignal<[string, string]>('userDeleted');

export async function userDeletionCleanupWorkflow(): Promise<void> {
  let done = false;

  // Signal handler runs inside workflow lock – fully deterministic
  setHandler(userDeletedSignal, async (userId: string, email: string) => {
    await cleanUpExternalSystems(userId);
    await sendDeletionConfirmation(email);
    done = true;
  });

  // Park the workflow until the signal arrives
  await condition(() => done);
}

export const TASK_QUEUE_NAME = "user-deletion-workflow-queue";

import { Worker } from '@temporalio/worker';
import * as activities from './activities';
import { TASK_QUEUE_NAME } from './shared';

async function runWorker() {
  // Create a worker that connects to the Temporal server
  const worker = await Worker.create({
    // Path to the module containing our workflow definitions
    workflowsPath: require.resolve('./workflows'),
    // Register all exported activities for execution
    activities,
    // The task queue this worker will poll for work
    taskQueue: TASK_QUEUE_NAME
  });

  console.log(`Worker started for task queue: ${TASK_QUEUE_NAME}`);
  // Start polling for and processing workflow tasks
  await worker.run();
}

// Start the worker and handle any errors
runWorker().catch((err) => {
  console.error('Worker failed: ', err);
  process.exit(1);
});

import { Connection, Client } from '@temporalio/client';
import { userDeletionCleanupWorkflow, userDeletedSignal } from './workflows';
import express, { Request, Response, RequestHandler } from 'express';
import { TASK_QUEUE_NAME } from './shared';

const app = express();
const PORT = process.env.PORT || 3333;

app.use(express.json()); // parse JSON request bodies

app.get('/', (_req, res) => {
  res.send('Sequin-Temporal Webhook Receiver is running');
});

let client: Client;

async function setupTemporal() {
  const connection = await Connection.connect();
  client = new Client({ connection });
}

const webhookHandler: RequestHandler = async (req: Request, res: Response): Promise<void> => {
  const { data } = req.body || {};
  if (!Array.isArray(data)) {
    res.status(400).send('Bad request');
    return;
  }

  try {
    for (const change of data) {
      if (
        change.action === 'delete' &&
        change.metadata?.table_name === 'users'
      ) {
        const { id: userId, email } = change.record;

        const workflowId = `user-delete-${userId}`;

        await client.workflow.signalWithStart(userDeletionCleanupWorkflow, {
          workflowId,
          taskQueue: TASK_QUEUE_NAME,
          args: [],                         // workflow takes no start args
          signal: userDeletedSignal,
          signalArgs: [userId, email]
        });

        console.log(`Signalled (or started) workflow ${workflowId}`);
      }
    }
    res.sendStatus(200);                    // acknowledge Sequin
  } catch (err) {
    console.error(err);
    res.sendStatus(500);                    // Sequin will retry
  }
};

app.post('/delete-user', webhookHandler);

setupTemporal().then(() => {
  app.listen(PORT, () => console.log(`Webhook listener started on port ${PORT}`));
});