Worker

Workflows, queues, and cron schedules for the worker preset.

When you set preset: "worker", Oxide targets Cloudflare Workers semantics (and the same-worker runtime shape on celld).

This page documents the *.server.ts exports that wire into Cloudflare Workflows, Queues, and Cron triggers.

Workflows

Durable multi-step jobs on Cloudflare Workflows. Export workflow() from a *.server.ts file:

import { import SchemaSchema } from "effect";
import { const workflow: <P, R = unknown>(def: WorkflowDefinition<P, R>) => WorkflowHandle<P, R>
Define a durable Cloudflare Workers Workflow in a `*.server.ts` file. On the
worker, oxide emits a `WorkflowEntrypoint` class and wrangler binding. Call
`start` / `status` / `send` like actions (RPC stubs on the client).

```ts
// src/invoice.server.ts
export const invoice = workflow({
  name: "invoice",
  payload: Schema.Struct({ orderId: Schema.String }),
  run: async ({ payload }, step) => {
    await step.do("charge", () => charge(payload.orderId));
  },
});

const { id } = await invoice.start({ orderId: "…" });
```workflow } from "oxidejs";

const 
const Params: Schema.Struct<{
    readonly orderId: Schema.String;
}>Params = import SchemaSchema.
function Struct<{
    readonly orderId: Schema.String;
}>(fields: {
    readonly orderId: Schema.String;
}): Schema.Struct<{
    readonly orderId: Schema.String;
}>
Defines a struct schema from a map of field schemas.

**Details**

Each field value is a schema. Use 
{@link 
optionalKey
}
 or 
{@link 
optional
}
 to
mark fields as optional, and 
{@link 
mutableKey
}
 to mark them as mutable.

The resulting schema's `Type` is a readonly object type with the fields'
decoded types. The `Encoded` form mirrors the field schemas' encoded types.

**Example** (Defining a basic struct)

```ts import.meta.vitest
import { Schema } from "effect"

const Person = Schema.Struct({
  name: Schema.String,
  age: Schema.Number,
  email: Schema.optionalKey(Schema.String)
})

// { readonly name: string; readonly age: number; readonly email?: string }
type Person = typeof Person.Type

Schema.decodeUnknownSync(Person)({ name: "Alice", age: 30 }) // => { name: "Alice", age: 30 }
```
@categoryconstructors@since3.10.0Struct({ orderId: Schema.StringorderId: import SchemaSchema.const String: Schema.String
Type-level representation of 
{@link 
String
}
.


Schema for `string` values. Validates that the input is `typeof` `"string"`.
@categorymodels@since4.0.0@categoryschemas@since4.0.0String });

declare const 
const charge: (orderId: string) => Promise<{
    ok: true;
    orderId: string;
}>charge: (
  orderId: stringorderId: string
) => interface Promise<T>
Represents the completion of an asynchronous operationPromise<{ ok: trueok: true; orderId: stringorderId: string }>;

export const 
const invoice: WorkflowHandle<{
    readonly orderId: string;
}, {
    orderId: string;
}>invoice = 
workflow<{
    readonly orderId: string;
}, {
    orderId: string;
}>(def: WorkflowDefinition<{
    readonly orderId: string;
}, {
    orderId: string;
}>): WorkflowHandle<{
    readonly orderId: string;
}, {
    orderId: string;
}>
Define a durable Cloudflare Workers Workflow in a `*.server.ts` file. On the
worker, oxide emits a `WorkflowEntrypoint` class and wrangler binding. Call
`start` / `status` / `send` like actions (RPC stubs on the client).

```ts
// src/invoice.server.ts
export const invoice = workflow({
  name: "invoice",
  payload: Schema.Struct({ orderId: Schema.String }),
  run: async ({ payload }, step) => {
    await step.do("charge", () => charge(payload.orderId));
  },
});

const { id } = await invoice.start({ orderId: "…" });
```workflow({
  WorkflowDefinition<P, R = unknown>.name: string
Workflow name (wrangler `name` + Rpc prefix).name: "invoice",
  
WorkflowDefinition<{ readonly orderId: string; }, { orderId: string; }>.payload?: Schema.Codec<{
    readonly orderId: string;
}, unknown, never, never> | undefined
Decode `start` params (Encoded in, Type out).payload: 
const Params: Schema.Struct<{
    readonly orderId: Schema.String;
}>Params,
  
WorkflowDefinition<{ readonly orderId: string; }, { orderId: string; }>.run: (event: WorkflowRunEvent<{
    readonly orderId: string;
}>, step: WorkflowStep) => {
    orderId: string;
} | Promise<{
    orderId: string;
}>run: async ({ 
payload: {
    readonly orderId: string;
}payload }, step: WorkflowStepstep) => {
    await step: WorkflowStepstep.
WorkflowStep.do: <{
    ok: true;
    orderId: string;
}>(name: string, callback: () => {
    ok: true;
    orderId: string;
} | Promise<{
    ok: true;
    orderId: string;
}>) => Promise<{
    ok: true;
    orderId: string;
}> (+1 overload)do("charge", () => 
const charge: (orderId: string) => Promise<{
    ok: true;
    orderId: string;
}>charge(
payload: {
    readonly orderId: string;
}payload.orderId: stringorderId));
    await step: WorkflowStepstep.WorkflowStep.sleep: (name: string, duration: string | number) => Promise<void>sleep("settle", "1 day");
    return await step: WorkflowStepstep.
WorkflowStep.do: <{
    orderId: string;
}>(name: string, callback: () => {
    orderId: string;
} | Promise<{
    orderId: string;
}>) => Promise<{
    orderId: string;
}> (+1 overload)do("done", () => ({ orderId: stringorderId: 
payload: {
    readonly orderId: string;
}payload.orderId: stringorderId }));
  },
});

Defaults: binding INVOICE, class InvoiceWorkflow. Override with binding / className.

From the client or server:

// @filename: client.ts
import { 
const invoice: WorkflowHandle<{
    readonly orderId: string;
}, void>invoice } from "./invoice.server";

const { const id: stringid } = await 
const invoice: WorkflowHandle<{
    readonly orderId: string;
}, void>invoice.
WorkflowHandle<{ readonly orderId: string; }, void>.start: (params: {
    readonly orderId: string;
}, opts?: CallOptions) => Promise<WorkflowStartResult>start({ orderId: stringorderId: "ord_1" });
const const status: WorkflowInstanceStatusstatus = await 
const invoice: WorkflowHandle<{
    readonly orderId: string;
}, void>invoice.WorkflowHandle<{ readonly orderId: string; }, void>.status: (id: string, opts?: CallOptions) => Promise<WorkflowInstanceStatus>status(const id: stringid);

Keep side effects inside step.do. The runtime replays run() from the start for deterministic recovery. Pass { idempotencyKey } on start for a stable instance id.

Queues

Buffer work, then start a workflow per message (durable / resumable). Export queue() from a *.server.ts file next to the workflow it drives:

import { import SchemaSchema } from "effect";
import { const queue: <P>(def: QueueDefinition<P>) => QueueHandle<P>
Define a Cloudflare Workers Queue in a `*.server.ts` file. Oxide emits
producer + consumer wrangler entries and a same-worker `queue` handler that
starts `workflow` once per message.

```ts
export const invoice = workflow({ name: "invoice", payload: Params, run });
export const invoices = queue({
  name: "invoices",
  workflow: invoice,
});

const { id } = await invoices.send({ orderId: "…" });
await invoice.status(id);
```

Cloudflare does not return message ids from `send`, so oxide wraps each
body in an envelope with a client-chosen id (`idempotencyKey` / UUID) and
returns that id for workflow polling. `send` / `sendBatch` decode with the
workflow's `payload` schema.

celld does not deliver to a consumer that shares a Worker with `fetch()`.
Set `producerStart: true` so `send` / `sendBatch` also start the workflow
from the producer (same id). Default is off so Cloudflare queue semantics
control execution. Delayed messages (`delaySeconds`) still need a consumer.queue, const workflow: <P, R = unknown>(def: WorkflowDefinition<P, R>) => WorkflowHandle<P, R>
Define a durable Cloudflare Workers Workflow in a `*.server.ts` file. On the
worker, oxide emits a `WorkflowEntrypoint` class and wrangler binding. Call
`start` / `status` / `send` like actions (RPC stubs on the client).

```ts
// src/invoice.server.ts
export const invoice = workflow({
  name: "invoice",
  payload: Schema.Struct({ orderId: Schema.String }),
  run: async ({ payload }, step) => {
    await step.do("charge", () => charge(payload.orderId));
  },
});

const { id } = await invoice.start({ orderId: "…" });
```workflow } from "oxidejs";

const 
const Params: Schema.Struct<{
    readonly orderId: Schema.String;
}>Params = import SchemaSchema.
function Struct<{
    readonly orderId: Schema.String;
}>(fields: {
    readonly orderId: Schema.String;
}): Schema.Struct<{
    readonly orderId: Schema.String;
}>
Defines a struct schema from a map of field schemas.

**Details**

Each field value is a schema. Use 
{@link 
optionalKey
}
 or 
{@link 
optional
}
 to
mark fields as optional, and 
{@link 
mutableKey
}
 to mark them as mutable.

The resulting schema's `Type` is a readonly object type with the fields'
decoded types. The `Encoded` form mirrors the field schemas' encoded types.

**Example** (Defining a basic struct)

```ts import.meta.vitest
import { Schema } from "effect"

const Person = Schema.Struct({
  name: Schema.String,
  age: Schema.Number,
  email: Schema.optionalKey(Schema.String)
})

// { readonly name: string; readonly age: number; readonly email?: string }
type Person = typeof Person.Type

Schema.decodeUnknownSync(Person)({ name: "Alice", age: 30 }) // => { name: "Alice", age: 30 }
```
@categoryconstructors@since3.10.0Struct({ orderId: Schema.StringorderId: import SchemaSchema.const String: Schema.String
Type-level representation of 
{@link 
String
}
.


Schema for `string` values. Validates that the input is `typeof` `"string"`.
@categorymodels@since4.0.0@categoryschemas@since4.0.0String });

declare const const charge: (orderId: string) => Promise<void>charge: (orderId: stringorderId: string) => interface Promise<T>
Represents the completion of an asynchronous operationPromise<void>;

export const 
const invoice: WorkflowHandle<{
    readonly orderId: string;
}, void>invoice = 
workflow<{
    readonly orderId: string;
}, void>(def: WorkflowDefinition<{
    readonly orderId: string;
}, void>): WorkflowHandle<{
    readonly orderId: string;
}, void>
Define a durable Cloudflare Workers Workflow in a `*.server.ts` file. On the
worker, oxide emits a `WorkflowEntrypoint` class and wrangler binding. Call
`start` / `status` / `send` like actions (RPC stubs on the client).

```ts
// src/invoice.server.ts
export const invoice = workflow({
  name: "invoice",
  payload: Schema.Struct({ orderId: Schema.String }),
  run: async ({ payload }, step) => {
    await step.do("charge", () => charge(payload.orderId));
  },
});

const { id } = await invoice.start({ orderId: "…" });
```workflow({
  WorkflowDefinition<P, R = unknown>.name: string
Workflow name (wrangler `name` + Rpc prefix).name: "invoice",
  
WorkflowDefinition<{ readonly orderId: string; }, void>.payload?: Schema.Codec<{
    readonly orderId: string;
}, unknown, never, never> | undefined
Decode `start` params (Encoded in, Type out).payload: 
const Params: Schema.Struct<{
    readonly orderId: Schema.String;
}>Params,
  
WorkflowDefinition<{ readonly orderId: string; }, void>.run: (event: WorkflowRunEvent<{
    readonly orderId: string;
}>, step: WorkflowStep) => void | Promise<void>run: async ({ 
payload: {
    readonly orderId: string;
}payload }, step: WorkflowStepstep) => {
    await step: WorkflowStepstep.WorkflowStep.do: <void>(name: string, callback: () => void | Promise<void>) => Promise<void> (+1 overload)do("charge", () => const charge: (orderId: string) => Promise<void>charge(
payload: {
    readonly orderId: string;
}payload.orderId: stringorderId));
  },
});

export const 
const invoices: QueueHandle<{
    readonly orderId: string;
}>invoices = 
queue<{
    readonly orderId: string;
}>(def: QueueDefinition<{
    readonly orderId: string;
}>): QueueHandle<{
    readonly orderId: string;
}>
Define a Cloudflare Workers Queue in a `*.server.ts` file. Oxide emits
producer + consumer wrangler entries and a same-worker `queue` handler that
starts `workflow` once per message.

```ts
export const invoice = workflow({ name: "invoice", payload: Params, run });
export const invoices = queue({
  name: "invoices",
  workflow: invoice,
});

const { id } = await invoices.send({ orderId: "…" });
await invoice.status(id);
```

Cloudflare does not return message ids from `send`, so oxide wraps each
body in an envelope with a client-chosen id (`idempotencyKey` / UUID) and
returns that id for workflow polling. `send` / `sendBatch` decode with the
workflow's `payload` schema.

celld does not deliver to a consumer that shares a Worker with `fetch()`.
Set `producerStart: true` so `send` / `sendBatch` also start the workflow
from the producer (same id). Default is off so Cloudflare queue semantics
control execution. Delayed messages (`delaySeconds`) still need a consumer.queue({
  QueueDefinition<P>.name: string
Queue name (wrangler `queue` + Rpc prefix).name: "invoices",
  
QueueDefinition<{ readonly orderId: string; }>.workflow: string | WorkflowHandle<{
    readonly orderId: string;
}, unknown>
Workflow to start per message. Instance id is the client-chosen envelope
id (from `{ idempotencyKey }` / request header / UUID) — CF does not
return message ids from `send`. Payload schema is taken from the
workflow handle (pass a `workflow()` handle, not only a name string).workflow: 
const invoice: WorkflowHandle<{
    readonly orderId: string;
}, void>invoice,
});

const { const id: stringid } = await 
const invoices: QueueHandle<{
    readonly orderId: string;
}>invoices.
QueueHandle<{ readonly orderId: string; }>.send: (body: {
    readonly orderId: string;
}, opts?: QueueSendOptions & CallOptions) => Promise<QueueSendResult>send({ orderId: stringorderId: "ord_1" });
await 
const invoices: QueueHandle<{
    readonly orderId: string;
}>invoices.
QueueHandle<{ readonly orderId: string; }>.sendBatch: (messages: Iterable<QueueMessageSendRequest<{
    readonly orderId: string;
}>>, opts?: QueueSendOptions & CallOptions) => Promise<QueueSendBatchResult>sendBatch([{ 
QueueMessageSendRequest<{ readonly orderId: string; }>.body: {
    readonly orderId: string;
}body: { orderId: stringorderId: "ord_2" } }]);
const const status: WorkflowInstanceStatusstatus = await 
const invoice: WorkflowHandle<{
    readonly orderId: string;
}, void>invoice.WorkflowHandle<{ readonly orderId: string; }, void>.status: (id: string, opts?: CallOptions) => Promise<WorkflowInstanceStatus>status(const id: stringid);

Oxide merges queues.producers / queues.consumers into wrangler.jsonc and attaches a same-worker queue handler that starts the workflow from each message.

Cloudflare does not return message ids from send, so oxide wraps bodies in an envelope with a client-chosen id and returns { id } from send (and { ids } from sendBatch) for workflow.status polling.

Optional handle replaces auto-start (unwrap with readQueueEnvelope). Optional maxBatchSize / maxBatchTimeout / maxRetries stamp the consumer entry. Optional producerStart: true also starts the workflow from send / sendBatch for celld (same-worker consumers do not run with fetch()). Default is off so Cloudflare queue semantics control execution.

Queue name must not match a workflow name (RPC tags collide on .send) or an action module key.

celld caveat: a queue consumer cannot share a Worker with fetch(). Keep the same-worker consumer for Cloudflare; set producerStart: true when you need enqueue to progress on celld.

Schedules

Cron ticks that start a workflow (or enqueue / run a custom handle). Export schedule() from a *.server.ts file:

import { import SchemaSchema } from "effect";
import { const schedule: <P = unknown>(def: ScheduleDefinition<P>) => ScheduleHandle<P>
Define a Cloudflare Workers Cron trigger in a `*.server.ts` file. Oxide
merges `triggers.crons` into wrangler and attaches a same-worker
`scheduled` handler that starts `workflow` (or enqueues / runs `handle`).

```ts
export const invoice = workflow({ name: "invoice", payload: Params, run });
export const nightly = schedule({
  name: "nightly",
  cron: "0 3 * * *",
  workflow: invoice,
  params: { orderId: "batch" },
});
```

Exactly one of `workflow` / `queue` / `handle` is required. Workflow
instance id is `${name}:${scheduledTime}`.schedule, const workflow: <P, R = unknown>(def: WorkflowDefinition<P, R>) => WorkflowHandle<P, R>
Define a durable Cloudflare Workers Workflow in a `*.server.ts` file. On the
worker, oxide emits a `WorkflowEntrypoint` class and wrangler binding. Call
`start` / `status` / `send` like actions (RPC stubs on the client).

```ts
// src/invoice.server.ts
export const invoice = workflow({
  name: "invoice",
  payload: Schema.Struct({ orderId: Schema.String }),
  run: async ({ payload }, step) => {
    await step.do("charge", () => charge(payload.orderId));
  },
});

const { id } = await invoice.start({ orderId: "…" });
```workflow } from "oxidejs";

const 
const Params: Schema.Struct<{
    readonly orderId: Schema.String;
}>Params = import SchemaSchema.
function Struct<{
    readonly orderId: Schema.String;
}>(fields: {
    readonly orderId: Schema.String;
}): Schema.Struct<{
    readonly orderId: Schema.String;
}>
Defines a struct schema from a map of field schemas.

**Details**

Each field value is a schema. Use 
{@link 
optionalKey
}
 or 
{@link 
optional
}
 to
mark fields as optional, and 
{@link 
mutableKey
}
 to mark them as mutable.

The resulting schema's `Type` is a readonly object type with the fields'
decoded types. The `Encoded` form mirrors the field schemas' encoded types.

**Example** (Defining a basic struct)

```ts import.meta.vitest
import { Schema } from "effect"

const Person = Schema.Struct({
  name: Schema.String,
  age: Schema.Number,
  email: Schema.optionalKey(Schema.String)
})

// { readonly name: string; readonly age: number; readonly email?: string }
type Person = typeof Person.Type

Schema.decodeUnknownSync(Person)({ name: "Alice", age: 30 }) // => { name: "Alice", age: 30 }
```
@categoryconstructors@since3.10.0Struct({ orderId: Schema.StringorderId: import SchemaSchema.const String: Schema.String
Type-level representation of 
{@link 
String
}
.


Schema for `string` values. Validates that the input is `typeof` `"string"`.
@categorymodels@since4.0.0@categoryschemas@since4.0.0String });

declare const const charge: (orderId: string) => Promise<void>charge: (orderId: stringorderId: string) => interface Promise<T>
Represents the completion of an asynchronous operationPromise<void>;

export const 
const invoice: WorkflowHandle<{
    readonly orderId: string;
}, void>invoice = 
workflow<{
    readonly orderId: string;
}, void>(def: WorkflowDefinition<{
    readonly orderId: string;
}, void>): WorkflowHandle<{
    readonly orderId: string;
}, void>
Define a durable Cloudflare Workers Workflow in a `*.server.ts` file. On the
worker, oxide emits a `WorkflowEntrypoint` class and wrangler binding. Call
`start` / `status` / `send` like actions (RPC stubs on the client).

```ts
// src/invoice.server.ts
export const invoice = workflow({
  name: "invoice",
  payload: Schema.Struct({ orderId: Schema.String }),
  run: async ({ payload }, step) => {
    await step.do("charge", () => charge(payload.orderId));
  },
});

const { id } = await invoice.start({ orderId: "…" });
```workflow({
  WorkflowDefinition<P, R = unknown>.name: string
Workflow name (wrangler `name` + Rpc prefix).name: "invoice",
  
WorkflowDefinition<{ readonly orderId: string; }, void>.payload?: Schema.Codec<{
    readonly orderId: string;
}, unknown, never, never> | undefined
Decode `start` params (Encoded in, Type out).payload: 
const Params: Schema.Struct<{
    readonly orderId: Schema.String;
}>Params,
  
WorkflowDefinition<{ readonly orderId: string; }, void>.run: (event: WorkflowRunEvent<{
    readonly orderId: string;
}>, step: WorkflowStep) => void | Promise<void>run: async ({ 
payload: {
    readonly orderId: string;
}payload }, step: WorkflowStepstep) => {
    await step: WorkflowStepstep.WorkflowStep.do: <void>(name: string, callback: () => void | Promise<void>) => Promise<void> (+1 overload)do("charge", () => const charge: (orderId: string) => Promise<void>charge(
payload: {
    readonly orderId: string;
}payload.orderId: stringorderId));
  },
});

export const 
const nightly: ScheduleHandle<{
    readonly orderId: string;
}>nightly = 
schedule<{
    readonly orderId: string;
}>(def: ScheduleDefinition<{
    readonly orderId: string;
}>): ScheduleHandle<{
    readonly orderId: string;
}>
Define a Cloudflare Workers Cron trigger in a `*.server.ts` file. Oxide
merges `triggers.crons` into wrangler and attaches a same-worker
`scheduled` handler that starts `workflow` (or enqueues / runs `handle`).

```ts
export const invoice = workflow({ name: "invoice", payload: Params, run });
export const nightly = schedule({
  name: "nightly",
  cron: "0 3 * * *",
  workflow: invoice,
  params: { orderId: "batch" },
});
```

Exactly one of `workflow` / `queue` / `handle` is required. Workflow
instance id is `${name}:${scheduledTime}`.schedule({
  ScheduleDefinition<P = unknown>.cron: string
Cron expression (UTC). Emitted into wrangler `triggers.crons`.cron: "0 3 * * *",
  ScheduleDefinition<P = unknown>.name: string
Schedule name — used in the idempotent workflow / queue id.name: "nightly",
  
ScheduleDefinition<{ readonly orderId: string; }>.params?: ScheduleParams<{
    readonly orderId: string;
}> | undefined
Params for `workflow.start` / `queue.send`. Taken as-is or from a
function of the schedule event. Decoded with the workflow/queue payload
schema when present.params: { orderId: stringorderId: "batch" },
  
ScheduleDefinition<{ readonly orderId: string; }>.workflow?: string | WorkflowHandle<{
    readonly orderId: string;
}, unknown> | undefined
Start this workflow once per tick.workflow: 
const invoice: WorkflowHandle<{
    readonly orderId: string;
}, void>invoice,
});

Exactly one of workflow / queue / handle. Oxide merges unique cron expressions into triggers.crons and attaches a same-worker scheduled handler. Each tick starts the workflow with id `${name}:${scheduledTime}` (idempotent retries).

params may be a value or (event) => value; payload schema comes from the workflow/queue handle.