Transactionally start a Temporal workflow when data changes in your database

-- Create a users table
create table users (
  id uuid primary key default gen_random_uuid (),
  first_name text not null,
  last_name text not null,
  email text not null unique,
  created_at timestamptz default now(),
  updated_at timestamptz default now()
);

-- Insert some mock data
insert into
  users (first_name, last_name, email)
values
  ('John', 'Doe', 'john.doe@example.com'),
  ('Jane', 'Smith', 'jane.smith@example.com'),
  ('Alice', 'Johnson', 'alice.johnson@example.com'),
  ('Bob', 'Brown', 'bob.brown@example.com'),
  ('Charlie', 'Davis', 'charlie.davis@example.com'),
  ('Emily', 'Wilson', 'emily.wilson@example.com'),
  ('Frank', 'Miller', 'frank.miller@example.com'),
  ('Grace', 'Moore', 'grace.moore@example.com'),
  ('Hannah', 'Taylor', 'hannah.taylor@example.com'),
  ('Isaac', 'Anderson', 'isaac.anderson@example.com');

-- Capture changes to the users table
alter table users replica identity full;

npx @temporalio/create --sample empty sequin-cdc-to-temporal

├── README.md
├── node_modules
├── package-lock.json
├── package.json
├── src
│   ├── activities.ts
│   ├── client.ts
│   ├── mocha
│   ├── shared.ts
│   ├── worker.ts
│   └── workflows.ts
└── tsconfig.json

touch src/server.ts
npm install express --save

temporal server start-dev --db-filename your_temporal.db

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}`));
});

temporal server start-dev --db-filename your_temporal.db

npm start

npx ts-node src/server.ts

delete from users where first_name = 'John';