Add a background job

Anything slow, anything that calls the network, and every AI call runs as a background job, never inside a page request or server action. This page shows how to add one: write the handler, register it, enqueue it from the UI, optionally schedule it, and show its result live. No migration is needed.

How the queue works

• Queue: the jobs table, managed by packages/db/src/queue.ts. enqueue() inserts a queued row; claim() atomically takes the oldest ready job on a queue and marks it running with a lease; complete() deletes it; fail() either schedules a retry or marks it failed for good.
• Runner: startRunner() in packages/core/src/jobs/runner.ts, started once per server process. It runs one loop per registered queue, so jobs on different queues run in parallel and jobs on the same queue run one at a time. An idle loop checks again every 1.5 seconds.
• Registry: packages/core/src/jobs/registry.ts maps each queue name to its payload schema, handler, lease and attempt limit. The runner only processes queues listed there.
• History: every attempt writes a job_runs row (queue, attempt, payload, status, result or error). The System page lists them live.

Jobs run only while the app is open. If the app quits mid-job, the job’s lease expires and it’s claimed again on the next launch.

Step 1: Write the handler

Put the handler next to the feature it belongs to in packages/core/src/<area>/. A job is a Zod payload schema plus an async function that takes an AiContext and the parsed payload. This example suggests listening keywords for a product:

import { z } from "zod";
import { transaction } from "@startica/db/client";
import { keywords } from "@startica/db/schema";
import { runStructured, type AiContext } from "../ai/client.ts";
import { loadProductContext } from "../distribution/context.ts";

export const KEYWORD_IDEAS_PROMPT_VERSION = "keyword-ideas@1";

export const keywordIdeasPayload = z.object({
  workspaceId: z.guid(),
  productId: z.guid(),
});

const keywordIdeasOutput = z.object({
  keywords: z.array(z.string()).describe("Short phrases people use when they have the problem"),
});

export async function runKeywordIdeas(ctx: AiContext, payload: z.infer<typeof keywordIdeasPayload>) {
  const product = await loadProductContext(payload.productId);

  // The AI call happens outside any transaction.
  const result = await runStructured(ctx, {
    task: "strategy",
    promptVersion: KEYWORD_IDEAS_PROMPT_VERSION,
    system: `Suggest listening keywords for this product.\n\n<product>\n${product.text}\n</product>`,
    messages: [{ role: "user", content: "Suggest up to 10 keywords." }],
    schema: keywordIdeasOutput,
  });

  await transaction(async (db) => {
    for (const keyword of result.keywords.slice(0, 10)) {
      await db
        .insert(keywords)
        .values({ workspaceId: payload.workspaceId, productId: payload.productId, keyword, useFor: "listening", source: "ai" })
        .onConflictDoNothing();
    }
  });

  // Whatever you return is stored as the job run's result.
  return { suggested: result.keywords.length };
}

Payload rules

• Every payload has workspaceId. enqueue() adds it for you, but declare it in the schema (z.guid()).
• The runner validates the stored payload with your schema before calling the handler. A payload that fails validation fails the attempt like any other error.
• Scheduled jobs are enqueued with an empty payload (only workspaceId). If your job runs on a schedule, make the other fields optional and treat “missing” as “every product”, as listenScanPayload does with productId: z.guid().optional().
• Keep payloads small: ids, not rows. Read the current data in the handler.
• Use .default() for optional flags so the handler always gets a value.

AI calls and logging

Call the AI only through runStructured() or runResearch() from packages/core/src/ai/client.ts, and pass the ctx the runner gave you. The context carries workspaceId and jobRunId, so every call is logged to ai_runs (task, model, tokens, cost, latency, prompt version, status) and linked to this job run. The client also enforces the monthly budget: when the cap is reached it logs a blocked run and throws an AiError with kind "blocked".

Pick the task that matches the work (on Claude, strategy and research route to Opus, draft, vision and answer to Sonnet, score and ping to Haiku by default; on OpenAI, score and ping route to GPT-6 Luna and the rest to GPT-6.1 Sol; image always uses an OpenAI image model). Users can switch provider and re-route tasks in Settings. Adding a new AiTask means updating the AiTask union and both providers’ DEFAULT_MODELS in packages/core/src/ai/models.ts and DEFAULT_AI_MODELS in packages/db/src/bootstrap.ts; existing workspaces fall back to DEFAULT_MODELS until they have a row.

Give the prompt a version string and bump it whenever you change the prompt. See Prompts and evals.

Transactions

Keep network and AI calls outside transaction(). The transaction holds a process-wide lock, and every other query in the app waits while it’s open. Fetch first, then write in one short transaction.

Step 2: Register it

Add an entry to jobs in packages/core/src/jobs/registry.ts:

import { keywordIdeasPayload, runKeywordIdeas } from "../intel/keyword-ideas.ts";

export const jobs = {
  // …existing jobs…
  keyword_ideas: defineJob({
    payload: keywordIdeasPayload,
    run: runKeywordIdeas,
    visibilityTimeoutSeconds: 300,
    maxAttempts: 1,
  }),
} satisfies Record<string, Job>;

The key (keyword_ideas) is the queue name, and it becomes part of the QueueName type. The runner starts a loop for it on the next server start.

• visibilityTimeoutSeconds is the lease. It must be longer than the job can ever run: once a lease expires, the job counts as abandoned and can be claimed and run again. Slow, rate-limited work gets long leases: listen_scan and engagement_sync use 2400 seconds because Reddit allows about one request per minute.
• maxAttempts is the total number of tries. Failed attempts are retried after 1, 4, 9… minutes (attempts squared). Use 1 for anything that calls an expensive model or costs real money, so a failure shows in the UI with a button to retry instead of silently spending more. Cheap, idempotent jobs can retry (ping uses 3, opportunity_score uses 2).

Step 3: Enqueue it from a server action

Pages and server actions never do the work themselves. They enqueue and return:

"use server";

import { enqueue } from "@startica/db/queue";
import type { ActionResult } from "@/components/app/form-dialog";
import { mutate, productScope } from "@/lib/actions";

export async function requestKeywordIdeas(productId: string): Promise<ActionResult> {
  return mutate(async () => {
    const scope = await productScope(productId);
    await enqueue(scope.workspaceId, "keyword_ideas", { productId: scope.productId });
  });
}

enqueue(workspaceId, queue, payload, opts?) takes the queue name as a plain string, so a typo isn’t caught by the compiler and the job would wait forever. Double-check the name against the registry. To delay a job, pass { runAfter: date }.

Jobs can enqueue other jobs the same way: listen_scan enqueues opportunity_score in batches of 20.

Step 4: Schedule it (optional)

Recurring jobs are listed in SCHEDULES in runner.ts. Cron expressions use croner syntax in the computer’s local time:

export const SCHEDULES: { name: string; cron: string; queue: QueueName }[] = [
  { name: "listen", cron: "7 * * * *", queue: "listen_scan" }, // hourly
  { name: "engagement", cron: "17 */6 * * *", queue: "engagement_sync" }, // every 6 hours
  { name: "metrics", cron: "15 8 * * *", queue: "metrics_sync" }, // daily, 08:15
  { name: "replan", cron: "0 7 * * 1", queue: "roadmap_plan" }, // weekly, Monday 07:00
];

Pick an odd minute so jobs don’t all start on the hour. Remember that a schedule only fires while the app is running, and that its payload is empty apart from workspaceId.

Step 5: Show progress and results live

When a job writes to a table, the database layer emits a change event and /api/events streams it to the window. Add the tables your page reads to its LiveRefresh:

<LiveRefresh tables={["keywords", "jobs"]} />

LiveRefresh takes a fixed list of table names (the Table type in src/components/app/live-refresh.tsx); add your table there if it’s new.

To show that a job is queued or running, either keep a status column on the row the job works on (products.intelStatus is set to "queued" when an Intel build is requested and updated by the job), or query the jobs table for the queue, as the Roadmap page does for roadmap_plan. Show failures with the error and a retry button rather than retrying automatically.

Step 6: Test it

packages/core/src/jobs/runner.test.ts shows the pattern: open a real SQLite database in a temp folder, run migrations, enqueue, start the runner, wait for the job_runs row, stop the runner. Tests must not call an AI provider or the network, so test your handler’s pure parts directly, and use the runner test only for jobs that can run without AI (or with no work to do).

yarn test
yarn typecheck && yarn lint && yarn smoke:desktop

To try it in the app, run an isolated dev copy (see Development setup), trigger the job, and watch it on the System page. STARTICA_DISABLE_JOBS=1 starts the server without the runner, which lets you inspect queued payloads before they run.

Finally, add the job to the jobs reference.