Actions, jobs & schedules
Long-running work (AI streaming, email, webhooks), durable retryable background jobs (with backoff + dead-letter), and cron-triggered work that writes back to live tables — all without a new framework. Sync's mutations already work like Convex actions; pair them with defineSchedule for cron and @absolutejs/queue for durable jobs and the story is complete.
#Mutations ARE actions
Convex splits the surface into mutations (transactional, no side effects) and actions (can call external APIs, not transactional). Sync's defineMutation handler is async and can do anything — call HTTP, stream from an AI provider, send an email — and emit changes back to live tables via actions.change/insert/update/delete. The engine buffers those emitted changes and commits them as one applyChangeBatch only AFTER the handler resolves; if the handler throws, the buffered changes don't fan out. One concept, both behaviours:
engine.registerMutation(
defineMutation({
name: 'summariseTicket',
handler: async (args: { id: string }, ctx, actions) => {
const ticket = await db.tickets.findUnique({ where: { id: args.id } });
if (!ticket) return null;
// Call out to an external service — no transactional constraint.
const stream = anthropic({ apiKey }).stream({
messages: [{ role: 'user', content: ticket.body }]
});
let summary = '';
for await (const chunk of stream) {
if (chunk.type === 'text') summary += chunk.content;
}
// Write the summary back to the live tickets table; subscribers see
// it via the existing reactive subscription.
await actions.update('tickets', { id: args.id, summary });
return { id: args.id, summary };
}
})
);
// Client-side: typed end-to-end via treaty<typeof app> + syncStore.mutate.
const result = await tickets.mutate('summariseTicket', { id });#defineJobs — typed, durable, retryable
@absolutejs/queue is the durable-job piece: typed registry, exponential backoff, dead-letter, delayed one-shots, concurrency control, leases, and a standalone worker. In-memory store for dev; @absolutejs/queue-postgres for production (jobs survive restarts, multiple workers claim safely).
import {
createInMemoryJobStore,
createJobRegistry,
defineJobs,
exponentialBackoff,
queue,
t
} from '@absolutejs/queue';
import { Elysia } from 'elysia';
// One typed registry: each kind has a schema and a handler. Payload types
// are inferred end-to-end — no hand-written job map, no generics.
const jobs = defineJobs({
'email.send': t.Object({ to: t.String(), subject: t.String(), body: t.String() }),
'webhook.deliver': t.Object({ url: t.String(), body: t.Unknown() }),
'ticket.summarise': t.Object({ ticketId: t.String() })
});
const registry = createJobRegistry({
jobs,
handlers: {
'email.send': async ({ payload }) => sendEmail(payload),
'webhook.deliver': async ({ payload }) => fetch(payload.url, { method: 'POST', body: JSON.stringify(payload.body) }),
'ticket.summarise': async ({ payload }) => {
// Job handlers can run sync mutations to write back to live tables.
await engine.runMutation('summariseTicket', { id: payload.ticketId }, {});
}
}
});
const store = createInMemoryJobStore();
// Production: createPostgresJobStore({ db }) from @absolutejs/queue-postgres.
const app = new Elysia().use(queue({
store,
registry,
maxAttempts: 5,
backoff: exponentialBackoff({ baseMs: 500, factor: 2, maxMs: 30_000 }),
// Delayed one-shots, retries, dead-letter, concurrency are all built in.
}));#Enqueueing — from mutations, schedules, or anywhere
A mutation can enqueue durable follow-up work, a scheduled cron can fan out a batch of jobs, and any HTTP route can enqueue too. Job handlers can run sync mutations to write back to live tables — so the subscriber a user has open sees the result the moment the worker finishes.
// Inside a mutation handler: queue durable async work that survives restarts.
engine.registerMutation(
defineMutation({
name: 'createTicket',
handler: async (args: { title: string; body: string }, ctx, actions) => {
const ticket = await actions.insert('tickets', args);
// Fire-and-queue: the queue persists this job and a worker runs it.
await app.queue.enqueue('ticket.summarise', { ticketId: ticket.id });
return ticket;
}
})
);
// From a schedule: cron decides WHEN, the queue guarantees the work HAPPENS.
engine.registerSchedule(
defineSchedule({
name: 'dailyDigest',
pattern: '0 8 * * *', // 8 AM every day
run: async ({ db }) => {
const users = await db.all('users');
for (const user of users) {
await app.queue.enqueue('email.send', {
to: user.email,
subject: 'Your daily digest',
body: await renderDigest(user)
});
}
}
})
);
// One-shot delayed: 'run this 30 minutes from now.'
await app.queue.enqueue(
'webhook.deliver',
{ url: 'https://example.com/hook', body: { event: 'order.placed' } },
{ runAt: new Date(Date.now() + 30 * 60_000) }
);#Map to Convex's primitives
A direct translation table for teams comparing the two stacks. Different machinery, equivalent capability:
| Capability | Convex | AbsoluteJS Sync |
|---|---|---|
| Define an action Emitted changes buffer via actions.change and commit as one applyChangeBatch on resolve. | defineAction(handler) | defineMutation — the async handler IS the action |
| Call from the client | useAction(api.foo, args) | syncStore.mutate('foo', args) via treaty<typeof app> |
| Immediate follow-up work | ctx.scheduler.runAfter(0, internal.x) | app.queue.enqueue('x', args) from the mutation handler |
| Delayed one-shots | ctx.scheduler.runAfter(ms, internal.x) | app.queue.enqueue('x', args, { runAt }) |
| Cron triggers | Cron definitions | engine.registerSchedule + @elysiajs/cron (via the scheduled plugin) |
| Durable, retried background jobs | Workpool (durable, retried, deduped) | @absolutejs/queue — typed registry, exponentialBackoff, dead-letter, in-memory + Postgres stores |
| Action calls a mutation | ctx.runMutation from an action | Job handler calls engine.runMutation(...) directly |