Upstash Workflow pipeline

Upstash Workflow wraps each bundle call in a durable, auto-retried step via context.run. QStash delivers HTTP requests to your workflow endpoint — on a cron schedule for the safety-net loop, or ad hoc for webhooks.

See Running the enrich pipeline durably for the model.

Bundle setup

import { samesake } from "@samesake/postgres";
import { llmRerank } from "@samesake/server";
import { products } from "../catalog.ts";
import { geminiEmbed, geminiGenerate } from "../gemini.ts";

export const app = samesake({
  url: process.env.SAMESAKE_DATABASE_URL!,
  collection: products,
  models: { embed: geminiEmbed, generate: geminiGenerate },
  rerank: llmRerank(geminiGenerate),
});

Workflow endpoints

Each loop gets its own route. Steps inside a route are separate durable units.

import { serve } from "@upstash/workflow/nextjs";
import { app } from "@/lib/search";
import { fetchChangedRows } from "@/catalog-source";

export const { POST } = serve(async (context) => {
  await context.run("upsert", async () => {
    const rows = await fetchChangedRows();
    await app.enrich.upsert(rows);
  });
  await context.run("enrich", () => app.enrich.enrich());
});

import { serve } from "@upstash/workflow/nextjs";
import { app } from "@/lib/search";

export const { POST } = serve(async (context) => {
  // Image revalidation is a caller-owned @samesake/server workflow; retry failed rows here.
  await context.run("retry-failed", () =>
    app.enrich.retryFailed({ limit: 500 }),
  );
});

For Hono, Fastify, or other frameworks, swap the import: @upstash/workflow/hono, etc.

Schedule the recurring loop with QStash

Create cron schedules in the QStash dashboard (Schedules → Create Schedule), or programmatically:

import { Client } from "@upstash/qstash";

const client = new Client({ token: process.env.QSTASH_TOKEN! });
const baseUrl = process.env.APP_URL!; // your deployed app

await client.schedules.create({
  scheduleId: "catalog-sync",
  destination: `${baseUrl}/api/workflows/sync-catalog`,
  cron: "*/5 * * * *",
});

await client.schedules.create({
  scheduleId: "catalog-maintenance",
  destination: `${baseUrl}/api/workflows/maintenance`,
  cron: "0 * * * *",
});

Trigger manually or from a webhook

Use the Workflow Client to start a run with retries, delay, and flow control:

import { Client } from "@upstash/workflow";
import { app } from "@/lib/search";

const workflow = new Client({ token: process.env.QSTASH_TOKEN! });

export async function POST(request: Request) {
  const product = await request.json();

  // Fast-path: push one row directly (no workflow needed for a single upsert)
  await app.enrich.upsert([product]);

  // Or trigger the full workflow with flow control so bursts don't pile up:
  await workflow.trigger({
    url: `${process.env.APP_URL}/api/workflows/sync-catalog`,
    retries: 3,
    retryDelay: "(1 + retried) * 1000",
    flowControl: { key: "catalog-sync", parallelism: 1, rate: 10 },
  });

  return Response.json({ ok: true });
}

Non-retryable failures

When a row is permanently bad (bad config, not a transient model error), stop retrying:

import { WorkflowNonRetryableError } from "@upstash/workflow";

if (!process.env.SAMESAKE_DATABASE_URL) {
  throw new WorkflowNonRetryableError("SAMESAKE_DATABASE_URL is not set");
}

Why this fits samesake

QStash is HTTP-native — your workflow runs wherever you host the route. samesake still owns batching and incrementality inside each context.run step. Separate steps mean a failed enrich retries without re-running upsert.

Where to go next

Concept and platform picker — Running the enrich pipeline durably
Row statuses — Pipeline lifecycle
Upstash Workflow docs — upstash.com/docs/workflow

Run it end to end

The runnable, tested version of this guide lives in examples/pipeline-upstash — deterministic stubs, no API key. Run bun test (Postgres-backed examples need SAMESAKE_DATABASE_URL).