@vibedout/automations

Time-based automation rules for Cloudflare Workers, on a D1 table and a one-minute cron. A rule says "when this event happens, do this action at this offset from the event, or from a time the event carries". The planner turns an event into runs with deterministic ids; the scheduler stores them in the same D1 batch as your own write, a sweep claims each due run once, checks it against the world as it is now, runs the action and settles it.

• No timer per booking. The table is the schedule. A moved booking is one UPDATE plus the new plan in the same batch; a missed tick is caught on the next one (a run that is by then past its lateness window is skipped, not sent late).
• Once, however often. The id of a run is a hash of the rule, the step, the subject, the subject's version and the contact, so a redelivered event plans the same rows and storing them again changes nothing. Two sweeps at once claim disjoint rows.
• Workers-first, no runtime dependencies. WebCrypto, Intl and D1 only. The shared kits are bundled by the build; drizzle-orm is an optional peer, needed only by @vibedout/automations/drizzle.

| Import | What | |---|---| | @vibedout/automations | the types, planRuns, planEvent, runId, parseRule, evaluateConditions, sweepDue, the Scheduler interface, statuses and the message-status rank, durations, time-zone helpers | | @vibedout/automations/d1 | D1Scheduler, D1RuleStore, automationsMigrationSql() and AUTOMATIONS_MIGRATION_STATEMENTS | | @vibedout/automations/drizzle | automationRule, automationRun, automationSubject (Drizzle tables for your schema barrel) | | @vibedout/automations/types | the types alone |

Setup

1. Copy the migration into your Worker's migrations/ folder:

   import { automationsMigrationSql } from "@vibedout/automations/d1";
   console.log(automationsMigrationSql()); // CREATE TABLE IF NOT EXISTS automation_rule / automation_run / automation_subject ...

   (AUTOMATIONS_MIGRATION_STATEMENTS is the same, one statement per string, for a caller that applies it with db.batch(); D1's exec splits on newlines and breaks a multi-line statement.)

2. Add a Cron Trigger ("triggers": { "crons": ["* * * * *"] }) and call the sweep from scheduled(). Cron Triggers are limited to 5 per account on the free plan.

3. Export the tables from your Drizzle schema if you query them: export { automationRule, automationRun, automationSubject } from "@vibedout/automations/drizzle";

A rule

import type { Rule } from "@vibedout/automations";

const reminder: Rule = {
  id: "booking-reminder-24h",
  enabled: true,
  version: 1,
  // The events that plan this rule. `booking.rescheduled` plans again for the new version of the booking.
  trigger: { events: ["booking.created", "booking.rescheduled"] },
  // Optional, flat. Every `all` holds and (if any) one `any` holds. Paths start at payload., contact. or event.
  conditions: { all: [{ field: "contact.tags", op: "not_in", value: ["do-not-message"] }] },
  steps: [
    {
      id: "email",                 // stable; part of every run's id (default: the step's position)
      action: "email.send",        // the name you register a handler under
      offset: "-24h",              // a day before the anchor; "2h" is two hours after; or milliseconds
      anchor: "startsAt",          // "event" (when it happened) or a path into the payload that holds a time
      params: { template: "booking.reminder" },
      lateness: "1h",              // default 1 hour; null: never too late
    },
  ],
  quietHours: { start: "21:00", end: "08:00" }, // on the contact's clock; mode "defer" (default) or "skip"
  frequencyCap: { max: 1, per: "1d" },          // per contact, checked at send time
  timezone: "America/Mexico_City",              // used when neither the contact nor the quiet hours name a zone
};

parseRule(input) reads a rule from anywhere (a form, a stored row, a seed file), reports every problem at once with a path and a stable code, and returns a rule whose durations, zones and clock times are known good. A rule is off (enabled: false) until someone turns it on.

Conditions are { field, op, value } with eq, ne, gt, gte, lt, lte, in, not_in, contains and exists. A path that leads nowhere is missing: it fails eq, in, contains and the comparisons, and passes ne, not_in and exists: false. in tests a value against a list, or any element when the field is itself a list (a contact's tags).

Anchors are epoch milliseconds, or ISO 8601 text with its zone (2026-10-12T10:00:00-06:00). A time without a zone is refused, not guessed: the step is written as skipped with the reason no_anchor.

Planning

import { planEvent } from "@vibedout/automations";
import { D1Scheduler } from "@vibedout/automations/d1";

const scheduler = new D1Scheduler(env.DB);

const event = {
  id: "evt_01J...",                                     // the outbox row, kept on the run
  type: "booking.rescheduled",
  occurredAt: Date.now(),
  // `version` must grow on EVERY change that should supersede a plan: a row version bumped by each write (a move, a cancellation).
  subject: { type: "booking", id: booking.id, version: booking.version },
  contactId: booking.contactId,
  payload: { startsAt: booking.startsAt, endsAt: booking.endsAt },
};

const plan = await planEvent(rules, event, Date.now(), {
  contact: { id: contact.id, timezone: contact.tz, fields: { tags: contact.tags } },
  supersedes: true, // this event type replaces what was planned before (a move or a cancellation)
});

// One transaction: the booking, the old plan going, the new plan arriving.
await env.DB.batch([updateBookingStatement, ...scheduler.apply(plan)]);

What a plan decides for each step:

1. Due time: anchor + offset.
2. Lateness: a step already further past due than its lateness window when the event arrives is written as skipped (late), so an outage cannot turn a 24-hour reminder into an hour-before one. The skipped run is stored too, so the admin's list says why nothing went out. The end of the window is kept on the run (expiresAt), and the sweep skips a run past it as late: when it is the sweep that was late (it did not run for hours), the same rule applies.
3. Quiet hours, on the contact's clock (the contact's zone, else the quiet hours' zone, else the rule's, else America/Mexico_City): a due time inside the window moves to its end, or is dropped in skip mode. A reminder that would be pushed to or past what it reminds of is dropped (quiet_hours). The window is judged at the due time: a sweep that runs late, within the lateness window, can still send at a time the window would have refused. On the night the contact's clocks change, a window end they skip (02:30 when 02:00 becomes 03:00) is 03:30, and one they show twice is its first occurrence, in any zone (zonedWallToInstant, the bundled wallTimeToInstant of @vibedout/locale-core; until 0.1.2 a zone east of Greenwich got both the other way round).

A rule that does not apply (switched off, another event type, conditions false) writes nothing.

Supersede, cancel, and events that arrive late

planEvent(..., { supersedes: true }) returns plan.cancel, and scheduler.apply(plan) turns it into statements that cancel the subject's pending runs of an older version (reason superseded) and record the version the subject has reached. So:

• a move (booking.rescheduled, version 1) cancels the version-0 plan and plans version 1, in one batch;
• a cancellation (booking.cancelled, version 2) cancels what is pending and plans nothing;
• a redelivered event changes nothing (same ids; the cancellation only touches older versions, never the plan it came with);
• an event that arrives late is refused: a booking.created delivered after the booking.rescheduled or the cancellation that followed it plans nothing, because the subject has already reached a newer version.

The version is the contract. A superseding event cancels only runs of a version below its own, which is what makes a redelivery safe (it cannot cancel the plan it came with). So every change that should cancel or replace a plan must carry a version above it: a cancellation that reuses the version the plan was made for finds nothing older to cancel and the reminders stay. Use a version that every write bumps (a row version, or the row's updatedAt in milliseconds), not only the reschedule count.

A cancel key names one subject (default "<type>:<id>"; event.cancelKey overrides it): the versions under one key must be comparable, so give runs that share a key one subject's version scale. Runs already claimed when the cancellation lands are not touched; the send-time guards (below) catch those.

The scheduler

Scheduler<Write> is the interface; D1Scheduler is the implementation. schedule, cancel and apply return statements for your own db.batch(); the rest run by themselves.

| Method | What it does | |---|---| | schedule(runs) / cancel(key, opts) / apply(plan) | statements for your batch (insert with ON CONFLICT(id) DO NOTHING, conditional cancel, version record) | | claimDue(now, limit) | one UPDATE ... WHERE id IN (SELECT ... LIMIT n) RETURNING *: pending runs that are due, and running runs whose lease ran out; each gets a token and a lease; attempts goes up | | complete / skip / fail / retry / defer | conditional on the claim token, so a second call, or a holder whose lease passed to another sweep, changes nothing and gets false ("lost" for retry) | | countRecent({ ruleId, contactId, since }) | sends to a contact in a window (the frequency cap) | | get(id), purgeFinished(before) | read one run; delete finished runs and old version records (run it daily) |

Options: maxAttempts (5), leaseMs (10 minutes: a sweep must settle its batch within it), backoff (1 minute doubling to an hour). retry waits the longer of the curve and the error's retryAfterMs, never more than a day, and fails the run when the attempts are used up. A run whose holders keep dying (the lease runs out on every attempt) is marked failed, not claimed forever.

Run statuses: pending, running, done, failed, cancelled, skipped; reason says why for the last two (superseded, late, no_anchor, quiet_hours, rule_disabled, frequency_cap, stale, no_consent, or your own).

The sweep and its guards

import { anchorUnchanged, sweepDue } from "@vibedout/automations";
import { D1RuleStore, D1Scheduler } from "@vibedout/automations/d1";

export default {
  async scheduled(_controller: ScheduledController, env: Env, ctx: ExecutionContext) {
    const rules = new D1RuleStore(env.DB);
    ctx.waitUntil(
      sweepDue({
        scheduler: new D1Scheduler(env.DB),
        loadRule: (id) => rules.get(id),
        actions: {
          // The run id is the idempotency key to hand the provider.
          "email.send": async (run) => ({ messageId: await sendEmail(env, run.params, run.id) }),
        },
        guards: [
          // The plan was made for a moment; is the booking still then? (null: cancelled or gone)
          anchorUnchanged(async (run) => (await findBooking(env, run.subjectId))?.startsAt ?? null),
          async (run) => ((await hasConsent(env, run)) ? undefined : "no_consent"),
        ],
      }),
    );
  },
};

A plan is a promise made at event time and the world moves, so nothing is sent on the strength of the plan alone. Each claimed run passes, in order: the rule still exists and is enabled; its frequency cap; your guards. The first objection ends the run as skipped with that reason. An action then answers by returning (done, with an optional messageId), { skip: reason }, { defer: untilMs } (back for later, no attempt used), or by throwing: an error with retryable: false fails the run, anything else is retried with backoff.

Guarantees and limits, stated plainly:

• Each run is claimed by one sweep at a time. If an action outlives its lease the run is claimed again and the action runs twice; make it safe to repeat, with the run id as the provider's idempotency key. A provider without one may rarely send twice.
• A sweep does not start a run whose lease ran out while it was busy with the ones before it (it counts it as lost and leaves it to the next claim), reading options.clock (default Date.now) for the time that has passed. Keep batches small enough, or the lease long enough, for a slow provider.
• The frequency cap counts runs already sent to the contact in the window, and runs another sweep holds right now that sort before this one (by due time, then id). Of two runs in flight at once exactly one passes, and an earlier run that then fails leaves the later one skipped with the cap unspent: the cap errs towards sending less. It does not reserve a slot for a run that is not yet claimed.
• Transient retries ignore quiet hours (a retry is minutes away, not a new plan).
• A rule edited after its runs were planned changes what the sweep checks (enabled, cap) and what future events plan; the runs already stored keep the offset, the params and the lateness they were planned with. There is no re-plan: cancel a rule's pending runs by deleting or disabling the rule, or by a superseding event.
• Cloudflare Workflows is a better home for sequences that wait for a reply; the Scheduler interface is the seam for it.

Message status

The timeline shares one status vocabulary for email and WhatsApp. Providers deliver updates out of order, so the status moves by rank:

import { advanceMessageStatus } from "@vibedout/automations";

advanceMessageStatus("read", "delivered"); // "read": a late `delivered` does not undo `read`
advanceMessageStatus("delivered", "bounced"); // "bounced": a failure outranks every success

queued < sent < deferred < delivered < read, then the failures (rejected, failed, bounced, complained), which end a message. received (an inbound message) never changes.

Message events

Every channel puts its messages on the app's event bus in one shape, so a timeline can record any channel without importing it. A channel declares its own event type, <channel>.message, and emits one event each time a message is created, received or changes status:

import { messageEventType, messagePreview, parseMessageEventPayload, type MessageEventPayload } from "@vibedout/automations";

const payload: MessageEventPayload = {
  channel: "email",
  direction: "outbound",
  messageId: "msg_01J…", // the channel's own id, the same on every event about this message
  status: "delivered",
  at: new Date().toISOString(),
  address: "[email protected]",
  preview: messagePreview(text), // at most 160 characters
  runId, // the automation run that sent it, if any
};
messageEventType("email"); // "email.message"

// A subscriber reads it back untrusted:
const message = parseMessageEventPayload(event.payload);

An audience, across two features

A broadcast needs to know who may be messaged, and the feature that knows (the people, their consent and their tags) must not be imported by the channel that sends. The contract between them is one action id and two shapes. The owner of the people registers an action under AUDIENCE_ACTION_ID (contacts.audience); a channel runs it with an AudienceRequest; the action reads one page of the audience with consent already applied and hands it, as { ...deliverParams, page }, to the action the request names in deliverTo. An action answers with a status only, so the people travel as the input of a second action and not as a return value.

import { AUDIENCE_ACTION_ID, parseAudienceRequest, parseAudiencePage, type AudienceRequest } from "@vibedout/automations";

const request: AudienceRequest = {
  channel: "email", purpose: "marketing", tag: "vip", after: null, limit: 100,
  deliverTo: "email.broadcast_page", deliverParams: { broadcastId },
};
await findAction(AUDIENCE_ACTION_ID)?.action.run(ctx, request);
// ...and in the receiving action: const { page, broadcastId } = input; page = parseAudiencePage(input.page)

A page is a keyset slice: after is the id of the last contact of the previous page and next the id to continue from (null at the end), so the requester stores its own cursor and a page applied twice is the same page. excluded counts the contacts a page looked at and left out, by reason (no_consent, revoked, blocked, unreachable, no_address), so a report can say why the audience is smaller than the tag.

Testing

Test against a real D1: @cloudflare/vitest-pool-workers with a d1_databases binding, apply AUTOMATIONS_MIGRATION_STATEMENTS in a beforeEach, and pass explicit now values (every time-dependent method takes one) instead of faking the clock.