Defining Tasks and Events¶

Define type-safe tasks and events that work across your entire application.

Why Define Schemas?¶

Schema definitions give you:

Type safety - TypeScript infers payload and return types automatically
Shared schemas - One source of truth for worker and client
Runtime validation - Workers validate payloads and return values with StandardSchema (Zod, Valibot, etc.)
Better DX - Autocomplete for payloads and return values

Task Definitions¶

With StandardSchema¶

Use defineTask() with Zod or other StandardSchema-compatible libraries:

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

export const sendEmailTask = defineTask({
  name: "send-email",
  payload: z.object({
    to: z.string().email(),
    subject: z.string(),
    body: z.string(),
  }),
  returns: z.object({
    messageId: z.string(),
    sent: z.boolean(),
  }),
});

With TypeScript Types¶

Define tasks using pure TypeScript types:

import type { DefineTask } from "pgconductor-js";

export type SendEmailTask = DefineTask<{
  name: "send-email";
  payload: {
    to: string;
    subject: string;
    body: string;
  };
  returns: {
    messageId: string;
    sent: boolean;
  };
}>;

Registering Tasks¶

Both approaches register similarly:

import { Conductor, TaskSchemas } from "pgconductor-js";

// With StandardSchema
const conductor = Conductor.create({
  connectionString: "postgres://localhost/mydb",
  tasks: TaskSchemas.fromSchema([sendEmailTask, processOrderTask]),
  context: {},
});

// With TypeScript types
const conductor = Conductor.create({
  connectionString: "postgres://localhost/mydb",
  tasks: TaskSchemas.fromUnion<SendEmailTask | ProcessOrderTask>(),
  context: {},
});

Runtime Validation¶

Workers validate every execution of a task defined with defineTask() against its schemas, whichever way it was invoked: conductor.invoke(), ctx.invoke(), signalWithStart(), pgconductor.invoke() from SQL, or a dead-letter delivery.

The payload is validated before middleware and the handler run. The handler receives the schema's output, so transforms, defaults and stripped keys apply.
The return value is validated before the execution completes. The schema's output is stored as the result, serialized as JSON.
A payload or return value that does not match fails the execution without retries, with the issues as its error, for example Invalid payload for task "send-email": to: Invalid email address. Like any permanent failure it fails waiting parents and is delivered to a dead-letter queue.
A validator that throws instead of returning issues fails the attempt, which is retried.
Async validators are awaited.
Batch tasks validate each execution on its own: members with an invalid payload fail without reaching the handler, and the others run as the batch. Return values are validated per item.
Invoking does not validate, so the stored payload is the one you passed. Cron and custom event triggers carry no task payload.

Type-only definitions (DefineTask with TaskSchemas.fromUnion()) have no runtime schema and are never validated.

Event Definitions¶

Events enable reactive, event-driven patterns.

With StandardSchema¶

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

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

export const orderPlaced = defineEvent({
  name: "order.placed",
  payload: z.object({
    orderId: z.string(),
    total: z.number().positive(),
  }),
});

Custom events require runtime Standard Schema definitions so pgconductor can validate event names, payload fields, and filterable fields when tasks are registered.

Registering Events¶

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

Return Types¶

Tasks can return typed results when invoked from other tasks using ctx.invokeAndWait():

export const processDataTask = defineTask({
  name: "process-data",
  payload: z.object({
    items: z.array(z.string()),
  }),
  returns: z.object({
    processed: z.number(),
    errors: z.array(z.string()),
  }),
});

// Create handler
const task = conductor.createTask(
  { name: "process-data" },
  { invocable: true },
  async (event, ctx) => {
    // The return value is validated against the schema
    return {
      processed: event.payload.items.length,
      errors: [],
    };
  }
);

// Invoke from another task and get typed result
const parentTask = conductor.createTask(
  { name: "parent-task" },
  { invocable: true },
  async (event, ctx) => {
    // ctx.invokeAndWait() waits for completion and returns the result
    const result = await ctx.invokeAndWait(
      "process-data",
      { name: "process-data" },
      { items: ["a", "b", "c"] }
    );

    // result is typed as { processed: number, errors: string[] }
    console.log(result.processed); // Type-safe!
  }
);

Note: conductor.invoke() (called from outside a task) does not wait for completion and does not return results. Return types type the results of ctx.invokeAndWait(), ctx.waitForResult(), conductor.getExecution() and conductor.waitForResult().

Store schemas in a shared package so both worker and client use the same definitions.

Monorepo Structure¶

packages/
├── schemas/              # Shared schema package
│   ├── package.json
│   ├── tasks.ts          # Task definitions
│   └── events.ts         # Event definitions
├── worker/               # Worker service
│   ├── package.json
│   └── src/
│       └── index.ts
└── api/                  # API/client service
    ├── package.json
    └── src/
        └── index.ts

packages/schemas/tasks.ts:

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

export const sendEmailTask = defineTask({
  name: "send-email",
  payload: z.object({
    to: z.string().email(),
    subject: z.string(),
  }),
  returns: z.object({
    sent: z.boolean(),
    messageId: z.string(),
  }),
});

export const processOrderTask = defineTask({
  name: "process-order",
  payload: z.object({
    orderId: z.string().uuid(),
    items: z.array(z.object({
      productId: z.string(),
      quantity: z.number().positive(),
    })),
  }),
  returns: z.object({
    success: z.boolean(),
    total: z.number(),
  }),
});

packages/schemas/events.ts:

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

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

worker/src/index.ts:

import { Conductor, Orchestrator, TaskSchemas } from "pgconductor-js";
import { sendEmailTask, processOrderTask } from "@myapp/schemas/tasks";
import { userCreated } from "@myapp/schemas/events";

const conductor = Conductor.create({
  connectionString: process.env.DATABASE_URL,
  tasks: TaskSchemas.fromSchema([sendEmailTask, processOrderTask]),
  events: [userCreated],
  context: {},
});

const sendEmail = conductor.createTask(
  { name: "send-email" },
  { invocable: true },
  async (event, ctx) => {
    // Implementation...
    return { sent: true, messageId: "msg-123" };
  }
);

const processOrder = conductor.createTask(
  { name: "process-order" },
  { invocable: true },
  async (event, ctx) => {
    // Implementation...
    return { success: true, total: 99.99 };
  }
);

const orchestrator = Orchestrator.create({
  conductor,
  tasks: [sendEmail, processOrder],
});

await orchestrator.start();

api/src/index.ts:

import { Conductor, TaskSchemas } from "pgconductor-js";
import { sendEmailTask, processOrderTask } from "@myapp/schemas/tasks";
import { userCreated } from "@myapp/schemas/events";

const conductor = Conductor.create({
  connectionString: process.env.DATABASE_URL,
  tasks: TaskSchemas.fromSchema([sendEmailTask, processOrderTask]),
  events: [userCreated],
  context: {},
});

// Invoke tasks with full type safety
await conductor.invoke(
  { name: "send-email" },
  { to: "user@example.com", subject: "Welcome" }
);

await conductor.invoke(
  { name: "process-order" },
  {
    orderId: "123e4567-e89b-12d3-a456-426614174000",
    items: [
      { productId: "prod-1", quantity: 2 },
      { productId: "prod-2", quantity: 1 },
    ],
  }
);

// Emit events with full type safety
await conductor.emit("user.created", {
  userId: "123e4567-e89b-12d3-a456-426614174000",
  email: "alice@example.com",
});

What's Next?¶

Task Triggers - Configure invocable, cron, and custom event triggers
Testing - Unit test tasks with type-safe mocks
Conductor API - Full API documentation

Defining Tasks and Events¶

Why Define Schemas?¶

Task Definitions¶

With StandardSchema¶

With TypeScript Types¶

Registering Tasks¶

Runtime Validation¶

Event Definitions¶

With StandardSchema¶

Registering Events¶

Return Types¶

Sharing Definitions¶

Monorepo Structure¶

What's Next?¶