Skip to content

Task Triggers

Tasks can be triggered in multiple ways:

  • Invocable - Triggered manually via conductor.invoke()
  • Cron - Triggered on a schedule
  • Custom Events - Triggered when you emit custom application events

Invocable Tasks

Invocable tasks are triggered manually via conductor.invoke():

const processOrder = conductor.createTask(
  { name: "process-order" },
  { invocable: true },
  async (event, ctx) => {
    ctx.logger.info("Processing order:", event.payload);
  }
);

Invoke it:

await conductor.invoke(
  { name: "process-order" },
  { orderId: "123", items: ["widget"] }
);

Cron Tasks

Cron tasks run on a schedule defined with cron syntax:

const dailyReport = conductor.createTask(
  { name: "daily-report" },
  { cron: "0 9 * * *" }, // Every day at 9 AM
  async (event, ctx) => {
    ctx.logger.info("Generating daily report");
    // Generate and send report
  }
);

Custom Event Triggers

React to custom application events with type-safe payloads.

Defining Events

First, define your custom events:

import { defineEvent } from "pgconductor-js";
import { z } from "zod";

export const userCreated = defineEvent({
  name: "user.created",
  payload: z.object({
    userId: z.string(),
    email: z.string(),
    name: z.string(),
  }),
});

Creating Event Handlers

Register the event in your conductor and create a task to handle it:

const conductor = Conductor.create({
  connectionString: "postgres://localhost/mydb",
  tasks: TaskSchemas.fromSchema([taskDef]),
  events: [userCreated],
  context: {},
});

const onUserCreated = conductor.createTask(
  { name: "send-welcome-email" },
  { event: "user.created" },
  async (event, ctx) => {
    // event.name === "user.created"
    // event.payload is fully typed!
    const { userId, email, name } = event.payload;

    await sendEmail({
      to: email,
      subject: `Welcome ${name}!`,
    });
  }
);

Emitting Events

Emit events from anywhere in your application:

await conductor.emit("user.created", {
  userId: "user-123",
  email: "alice@example.com",
  name: "Alice",
});

Deduplicating Emits

Webhook providers redeliver, so ingress may emit the same delivery twice. Pass a dedupe_key to make the emit idempotent:

const eventId = await conductor.emit("slack.mentioned", payload, {
  dedupe_key: `slack:${delivery.eventId}`,
});

emit() always returns the event ID that Postgres Conductor assigned; the dedupe_key is your own key and is never returned. Repeating a dedupe_key for the same event name stores nothing and returns the original event ID, even when two emits race. Keys are scoped per event name: the same dedupe_key under two event names emits two events. A key is remembered while its dispatch execution exists. Completed dispatches are kept for one day and removed by the daily maintenance run of the internal queue, so a key deduplicates for 1–2 days after its event was dispatched. Outside that window, the same dedupe_key emits a new event.

From SQL, pass the key as p_dedupe_key: pgconductor.emit_event('slack.mentioned', payload, p_dedupe_key := 'slack:' || event_id).

Each event is stored as an internal dispatch execution. Destination inserts are atomic within one database statement. If dispatch retries before the source execution completes, it re-evaluates current subscriptions: existing deliveries are deduplicated, but newly registered subscriptions may receive additional deliveries. Handlers should be prepared for at-least-once delivery. Only orchestrators whose conductor has events configured dispatch events.

Emitting from Database Triggers

Postgres Conductor does not create or manage triggers on application tables. If a database change should emit an event, define the trigger in your own migrations and call pgconductor.emit_event():

create function app.emit_user_created()
returns trigger
language plpgsql
as $$
begin
  perform pgconductor.emit_event('user.created', to_jsonb(new));
  return new;
end;
$$;

create trigger emit_user_created
after insert on app.users
for each row execute function app.emit_user_created();

Because emit_event() inserts the dispatch execution in the current transaction, emission is committed or rolled back with the database change.

Event Filters

Declare top-level scalar fields as filterable, then register typed predicates on handlers:

const orderChanged = defineEvent({
  name: "order.changed",
  payload: z.object({
    status: z.string(),
    region: z.string(),
    amount: z.number(),
    note: z.string().optional(),
  }),
  filterable: ["status", "region", "amount", "note"],
});

conductor.createTask(
  { name: "handle-paid-us-orders" },
  {
    event: "order.changed",
    filter: {
      status: ["paid", "trial"],
      region: [{ prefix: "us-" }],
      amount: [{ numeric: [">=", 10, "<", 100] }],
      note: [{ exists: false }],
    },
  },
  async (event, ctx) => {
    // Both fields matched; values within one field are alternatives.
  }
);

Fields are combined with AND and alternatives within a field with OR. Supported alternatives are:

  • a string, number, boolean, or null for type-sensitive equality;
  • { prefix: "literal" } for a literal string prefix (%, _, and \\ have no special meaning);
  • { numeric: [">=", 10, "<", 100] } for a one- or two-bound numeric range;
  • { exists: true } or { exists: false } for field presence;
  • { "anything-but": value } for one atomic scalar exclusion.

Missing fields differ from JSON null. An anything-but predicate must be the field's only alternative. Filters are compiled into flat typed terms: alternatives for one field are ORed, while SQL requires every distinct filter field to match. Exact, prefix, and numeric-range terms use operator-specific indexes; presence and exclusions are evaluated set-wise in the same query. A filter may contain up to 8 fields and 4 alternatives per field; literal prefixes are limited to 64 characters.

Field Selection

For large events, you can select only specific fields to reduce payload size:

conductor.createTask(
  { name: "log-user-id" },
  { event: "user.created", fields: "userId" },
  async (event, ctx) => {
    // event.payload only contains { userId: string }
    ctx.logger.info("User created:", event.payload.userId);
  }
);

Waiting for Events in a Task

A running task can also wait for a custom event. ctx.waitForEvent() subscribes and suspends in one call, so it only sees events emitted after it has been registered:

const payment = await ctx.waitForEvent("payment", {
  event: paymentReceived,
  filter: { orderId: [orderId] },
  timeout: "1h",
});

When the event is a response to something the task does itself, such as posting an approval request, subscribe before the side effect and wait after it. Otherwise a fast response could be emitted before the wait is registered and be missed:

const subscription = await ctx.subscribe("approval", {
  event: approvalDecided,
  filter: { approvalId: [approvalId] },
});
await ctx.step("post-card", () => postApprovalCard(approvalId));
const decision = await subscription.wait({ timeout: "24h" });

A matching event emitted after subscribe() resolves is kept for wait(), even if the task is still running. The timeout starts at wait(). See ctx.subscribe().

To wait for whichever of several events comes first, subscribe to each and pass the subscriptions to ctx.waitForAny(). It returns the one that received an event first, or { key: "timeout" }:

const reply = await ctx.subscribe("reply", {
  event: threadReplied,
  filter: { interactionId: [interactionId] },
});
const winner = await ctx.waitForAny(
  "approval-or-reply",
  { decision: subscription, reply },
  { timeout: "24h" },
);

The other subscriptions are removed when one wins. See ctx.waitForAny().

Multiple Triggers

Tasks can respond to multiple trigger types:

const flexibleTask = conductor.createTask(
  { name: "flexible-task" },
  [
    { invocable: true },
    { cron: "0 * * * *", name: "hourly" },
    { event: "user.created" },
  ],
  async (event, ctx) => {
    // Discriminate based on event.name
    if (event.name === "pgconductor.invoke") {
      // Manually invoked
      const payload = event.payload;
    } else if (event.name === "hourly") {
      // Cron-triggered by "hourly" schedule
    } else if (event.name === "user.created") {
      // Custom event
      const { userId, email, name } = event.payload;
    }
  }
);

What's Next?