Durable AI Email Workflows with Temporal and MultiMail

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

const { sendEmailForApproval, recordApprovalOutcome } = proxyActivities<EmailActivities>({
  startToCloseTimeout: '30 seconds',
});

export const approvalSignal = defineSignal<[{ approved: boolean; approver: string }]>('approval');

export async function emailApprovalWorkflow(params: {
  to: string;
  subject: string;
  body: string;
  mailboxId: string;
}): Promise<{ sent: boolean; messageId: string | null }> {
  const { pendingId, messageId } = await sendEmailForApproval(params);

  let approved = false;
  let approver = '';
  let signalReceived = false;

  setHandler(approvalSignal, ({ approved: a, approver: ap }) => {
    approved = a;
    approver = ap;
    signalReceived = true;
  });

  "cm">// Wait up to 72 hours for human approval
  await condition(() => signalReceived, '72 hours');

  await recordApprovalOutcome({ pendingId, approved, approver, messageId });

  return { sent: approved, messageId: approved ? messageId : null };
}

import { Context } from '@temporalio/activity';

const MULTIMAIL_API = 'https://api.multimail.dev';
const MM_TOKEN = process.env.MULTIMAIL_API_KEY!;

export interface EmailActivities {
  sendEmailForApproval(params: {
    to: string;
    subject: string;
    body: string;
    mailboxId: string;
  }): Promise<{ pendingId: string; messageId: string }>;

  recordApprovalOutcome(params: {
    pendingId: string;
    approved: boolean;
    approver: string;
    messageId: string;
  }): Promise<void>;
}

export const emailActivities: EmailActivities = {
  async sendEmailForApproval({ to, subject, body, mailboxId }) {
    const info = Context.current().info;
    "cm">// Deterministic idempotency key: workflow + activity attempt
    const idempotencyKey = `${info.workflowExecution.workflowId}-${info.attempt}`;

    const res = await fetch(`${MULTIMAIL_API}/send_email`, {
      method: 'POST',
      headers: {
        Authorization: `Bearer ${MM_TOKEN}`,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({
        mailbox_id: mailboxId,
        to,
        subject,
        body,
        idempotency_key: idempotencyKey,
      }),
    });

    if (!res.ok) throw new Error(`MultiMail send failed: ${res.status}`);
    const data = await res.json() as { pending_id: string; message_id: string };
    return { pendingId: data.pending_id, messageId: data.message_id };
  },

  async recordApprovalOutcome({ pendingId, approved, approver, messageId }) {
    "cm">// Fetch the final state for audit logging
    const res = await fetch(`${MULTIMAIL_API}/read_email?message_id=${messageId}`, {
      headers: { Authorization: `Bearer ${MM_TOKEN}` },
    });
    const email = await res.json();
    console.log({ pendingId, approved, approver, status: email.status });
  },
};

import { Client, Connection } from '@temporalio/client';
import { approvalSignal } from './workflows';
import { createHmac } from 'crypto';

const connection = await Connection.connect({ address: process.env.TEMPORAL_ADDRESS });
const client = new Client({ connection, namespace: process.env.TEMPORAL_NAMESPACE });

export async function handleMultimailWebhook(req: Request): Promise<Response> {
  "cm">// Verify MultiMail webhook signature
  const sig = req.headers.get('x-multimail-signature') ?? '';
  const body = await req.text();
  const expected = createHmac('sha256', process.env.MULTIMAIL_WEBHOOK_SECRET!)
    .update(body)
    .digest('hex');
  if (sig !== expected) return new Response('Unauthorized', { status: 401 });

  const event = JSON.parse(body) as {
    type: 'approval.approved' | 'approval.rejected';
    workflow_id: string;
    approver_email: string;
    pending_id: string;
  };

  if (event.type !== 'approval.approved' && event.type !== 'approval.rejected') {
    return new Response('ignored', { status: 200 });
  }

  "cm">// workflow_id is set as a custom metadata field when the Temporal workflow starts
  const handle = client.workflow.getHandle(event.workflow_id);
  await handle.signal(approvalSignal, {
    approved: event.type === 'approval.approved',
    approver: event.approver_email,
  });

  return new Response('ok', { status: 200 });
}

import { Client } from '@temporalio/client';
import { emailApprovalWorkflow } from './workflows';

const MM_TOKEN = process.env.MULTIMAIL_API_KEY!;

export async function pollInboxAndDispatch(
  client: Client,
  mailboxId: string,
): Promise<void> {
  const res = await fetch(
    `https:"cm">//api.multimail.dev/check_inbox?mailbox_id=${mailboxId}&unread=true`,
    { headers: { Authorization: `Bearer ${MM_TOKEN}` } },
  );
  const { emails } = await res.json() as { emails: Array<{ id: string; from: string; subject: string }> };

  for (const email of emails) {
    const workflowId = `reply-${email.id}`;

    "cm">// Idempotent start — skip if workflow already running
    await client.workflow.start(emailApprovalWorkflow, {
      taskQueue: 'email-agent',
      workflowId,
      args: [{
        to: email.from,
        subject: `Re: ${email.subject}`,
        body: await generateReply(email.id),
        mailboxId,
      }],
    });
  }
}

async function generateReply(emailId: string): Promise<string> {
  const res = await fetch(`https:"cm">//api.multimail.dev/read_email?message_id=${emailId}`, {
    headers: { Authorization: `Bearer ${MM_TOKEN}` },
  });
  const { body } = await res.json() as { body: string };
  "cm">// Replace with your LLM call
  return `Thank you for your message. We will respond shortly.\n\nOriginal: ${body.slice(0, 200)}`;
}