Workflows

notify() is one shot. A workflow can sleep for three days, wait for a user to act, and resume exactly where it left off after a deploy, because each step's outcome is saved before the next runs.

A worked example

Invite someone, wait three days, and if they haven't accepted, nudge them by SMS. In ordinary TypeScript:

import { workflow } from "notifkit";

workflow("invite-nudge", async ({ step, event }) => {
  await step.notify({
    template: "invite",
    channels: ["email"],
    data: { inviterName: event.inviterName },
  });

  await step.wait("3d");

  const accepted = await step.waitForEvent("invite.accepted", { timeout: "24h" });

  if (!accepted) {
    await step.notify({
      template: "invite-reminder",
      channels: ["sms"],
      data: { inviterName: event.inviterName },
    });
  }
});

Register it in the process that runs the workflow service, then trigger it:

import "./workflows/invite.js";   // registration happens on import

const { instanceId } = await notifkit.triggerWorkflow({
  name: "invite-nudge",
  input: {
    user: { id: "usr_123" },      // required — step.notify() targets this user
    inviterName: "Alice",
  },
});

curl -X POST http://localhost:3000/v1/workflows/trigger \
  -H "Authorization: Bearer $NK_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "invite-nudge",
    "input": {
      "user": { "id": "usr_123" },
      "inviterName": "Alice"
    }
  }'
# → 202 { "messageId": "...", "instanceId": "..." }

Triggering for a user the project has not seen

A top-level user is shorthand for the same, and accepts the full inline user shape notify() takes. Given an object, the trigger upserts that user (contacts, segments, preferences and all) before starting the instance, so signup can register the person and start onboarding in one call:

await notifkit.triggerWorkflow({
  name: "invite-nudge",
  user: {
    id: "usr_999",
    email: "[email protected]",
    timezone: "Europe/Berlin",
  },
  input: { inviterName: "Alice" },
});

// Or, when the user already exists, just the id:
await notifkit.triggerWorkflow({
  name: "invite-nudge",
  user: "usr_123",
  input: { inviterName: "Alice" },
});

curl -X POST http://localhost:3000/v1/workflows/trigger \
  -H "Authorization: Bearer $NK_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "invite-nudge",
    "user": { "id": "usr_999", "email": "[email protected]" },
    "input": { "inviterName": "Alice" }
  }'

Either form arrives as input.user, so event.user.id reads the same. They combine: fields you set on input.user are kept and take precedence, so don't set input.user.id to one person and user to another.

How it survives the gap

The step API

step.run for anything that is not a notification

workflow("trial-ending", async ({ step, event }) => {
  // Charged once, even if the workflow replays a dozen times.
  const invoice = await step.run("create-invoice", () =>
    stripe.invoices.create({ customer: event.stripeCustomerId }),
  );

  await step.notify({
    template: "invoice-ready",
    channels: ["email"],
    data: { invoiceUrl: invoice.hosted_invoice_url },
  });
});

Waking a workflow with an event

waitForEvent parks the instance until your app reports that something happened:

// In your invite-acceptance handler:
await notifkit.ingestEvent({
  name: "invite.accepted",
  properties: { userId: "usr_123", inviteId: "inv_88" },
});

# In your invite-acceptance handler:
curl -X POST http://localhost:3000/v1/events \
  -H "Authorization: Bearer $NK_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "invite.accepted",
    "properties": { "userId": "usr_123", "inviteId": "inv_88" }
  }'

The events worker finds instances waiting on that name whose match criteria are met, records the payload as the step's output, and republishes the instance to resume. Use match so one user's event doesn't wake another's workflow:

const accepted = await step.waitForEvent("invite.accepted", {
  timeout: "72h",
  match: { userId: event.user.id },
});

if (accepted === null) {
  // timed out — nobody accepted within 72 hours
}

The timeout is durable like any step outcome: once it fires, the step is recorded as timed out and every replay sees null. An event arriving just after doesn't fill the step in retroactively, so a branch on accepted === null takes the same path every replay.

Workflows without deploying code

The same three actions are available as JSON, stored per project, for sequences that should be editable without shipping a release.

await notifkit.createWorkflow({
  name: "onboarding",
  steps: [
    { action: "notify", payload: { template: "welcome", channels: ["email"] } },
    { action: "wait", duration: "1d" },
    { action: "notify", payload: { template: "getting-started", channels: ["email"] } },
    { action: "waitForEvent", event: "project.created", options: { timeout: "72h" } },
    { action: "notify", payload: { template: "need-a-hand", channels: ["email"] } },
  ],
});

curl -X POST http://localhost:3000/v1/workflows \
  -H "Authorization: Bearer $NK_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "onboarding",
    "steps": [
      { "action": "notify",
        "payload": { "template": "welcome", "channels": ["email"] } },
      { "action": "wait", "duration": "1d" },
      { "action": "notify",
        "payload": { "template": "getting-started", "channels": ["email"] } },
      { "action": "waitForEvent", "event": "project.created",
        "options": { "timeout": "72h" } },
      { "action": "notify",
        "payload": { "template": "need-a-hand", "channels": ["email"] } }
    ]
  }'

Triggering is identical. A registered code workflow wins over a JSON definition with the same name.

workflow("payment-failed", async ({ step, event }) => {
  await step.notify({ template: "card-declined", channels: ["email"] });   // the customer
  await step.notify({ template: "billing-alert", user: "usr_ops_oncall" }); // someone else
  await step.wait("3d");
  await step.notify({ template: "card-declined-final", channels: ["email", "sms"] });
});

Recurring sends and digests

A workflow can't start itself. notifkit has no cron: no looping step, no schedule on a definition. Every instance starts with an outside trigger. You supply the clock; everything after the tick (fan-out, preferences, quiet hours, retries) is notifkit's.

A weekly digest is not a workflow

The common case needs no workflow. Point whatever already ticks in your infrastructure (cron, a systemd timer, a GitHub Action, Cloud Scheduler) at /v1/notify with a segment, and one request becomes one message per subscriber.

# Mondays, 09:00 UTC. Note the escaped %% — crontab treats a bare % as a newline.
0 9 * * 1  curl -X POST http://localhost:3000/v1/notify \
  -H "Authorization: Bearer $NK_KEY" \
  -H "x-idempotency-key: weekly-digest-$(date -u +\%G-W\%V)" \
  -H "Content-Type: application/json" \
  -d '{"segment":"digest-subscribers","template":"weekly-digest","channels":["email"]}'

// Whatever runs this on a schedule is outside notifkit.
const week = isoWeek(new Date());          // e.g. "2026-W33"

// The idempotency key is a header, and headers are set per client, not per
// call — so build one scoped to this run.
const client = new NotifkitClient({
  baseUrl: process.env.NK_URL!,
  apiKey: process.env.NK_KEY!,
  headers: { "x-idempotency-key": `weekly-digest-${week}` },
});

await client.notify({
  segment: "digest-subscribers",
  template: "weekly-digest",
  channels: ["email"],
  data: { week, topStories: await stories.forWeek(week) },
});

The key matters most here. A scheduler that fires twice (an overlapping run, a retried Action, two hosts with the same crontab) would otherwise send everyone the digest twice. Derive it from the period, not the clock, so both firings produce the same string. See not sending the same thing twice.

When every recipient's digest differs

Loop, with a key per message. listUsers() paginates but has no segment filter, so drive the loop from your own app's list:

const week = isoWeek(new Date());

for (const userId of await db.digestSubscribers()) {
  const items = await db.unreadFor(userId, week);
  if (items.length === 0) continue;        // no digest beats an empty digest

  const client = new NotifkitClient({
    baseUrl: process.env.NK_URL!,
    apiKey: process.env.NK_KEY!,
    headers: { "x-idempotency-key": `digest-${week}-${userId}` },
  });

  await client.notify({
    user: userId,
    template: "weekly-digest",
    channels: ["email"],
    data: { week, items },
  });
}

With per-user keys, a crash mid-loop is safe to re-run: users already sent to collapse on their keys, and the rest go out.

A fixed number of repeats

For finite, known repetition (a six-week onboarding drip, a three-day countdown), a workflow fits directly. Alternate notify and wait:

workflow("onboarding-drip", async ({ step }) => {
  const weeks = ["week-1", "week-2", "week-3"];

  for (const [i, week] of weeks.entries()) {
    if (i > 0) await step.wait("7d");
    await step.notify({ template: `onboarding-${week}`, channels: ["email"] });
  }
});

// Nothing above names a recipient. This does — and all three emails go to them:
await notifkit.triggerWorkflow({
  name: "onboarding-drip",
  input: { user: { id: "usr_123" } },
});

No step names a recipient, so all three emails go to the trigger user: one instance, one person. A hundred users means a hundred triggerWorkflow calls, one instance each, so every subscriber is at their own point in the sequence. Give a step an explicit user or segment only when that message is for someone else.

The loop is safe because it's deterministic: the same three iterations make the same five step calls on every replay, as replay matching requires. A loop over something that can change between replays isn't. Waiting before each notify also avoids a trailing wait that leaves the instance pending for a week after the last email.

Endless recurrence without an external clock

If you truly have nowhere for a cron entry, a workflow can re-trigger itself from a step.run, starting a fresh instance as the current one finishes:

workflow("weekly-summary", async ({ step, event }) => {
  await step.notify({ template: "weekly-summary", channels: ["email"] });
  await step.wait("7d");

  // step.run so the re-trigger is recorded and cannot fire twice on replay.
  await step.run("chain-next", () =>
    notifkit.triggerWorkflow({ name: "weekly-summary", input: event }),
  );
});

Each cycle is a separate instance, so the chain survives restarts like any instance. Give the notify step a segment and one chain drives the whole list: one instance to track, not one per subscriber.

The costs are real: drift compounds forever with no calendar to reset it, and with no definition-level off switch, the only way to stop a chain is to cancel its pending instance before the last step. Prefer a real scheduler when you have one.

Watching and cancelling

const instance = await notifkit.getWorkflow(instanceId);
// { id, name, status: "pending" | "running" | "completed" | "failed", input, createdAt, updatedAt }

await notifkit.listWorkflows();          // definitions registered for the project
await notifkit.cancelWorkflow(instanceId);

curl -s http://localhost:3000/v1/workflows/instances/$INSTANCE_ID \
  -H "Authorization: Bearer $NK_KEY"
# { id, name, status: "pending" | "running" | "completed" | "failed",
#   input, createdAt, updatedAt }

# definitions registered for the project
curl -s http://localhost:3000/v1/workflows -H "Authorization: Bearer $NK_KEY"

curl -X DELETE http://localhost:3000/v1/workflows/instances/$INSTANCE_ID \
  -H "Authorization: Bearer $NK_KEY"

Trace every message a workflow sent by filtering the delivery log on its instance id:

curl -s "http://localhost:3000/v1/notifications/logs?workflowInstanceId={instanceId}" \
  -H "Authorization: Bearer $NK_KEY"

Operational notes

• Each instance has a Redis lock with a 60-second TTL, renewed while the handler runs, so two workers can't run it at once.
• A sweep every minute finds instances marked running with an expired lock (a hard-killed process) and returns them to pending to resume.
• The wake-up loop polls every five seconds, so wait("30s") is accurate to about that. Workflows aren't a precision scheduler.
• workflow() writes to a per-process in-memory registry. If you split services across deployments, the process running services: ["workflow"] must import your handler files.