Inngest pipeline

Inngest turns each bundle call into a durable step. If enrich fails, Inngest retries just that step — upsert doesn’t re-run. Cron triggers give you the safety-net loop; events give you webhook fast-paths. You serve the functions from your existing Node app.

See Running the enrich pipeline durably for the model. This page wires it to Inngest.

Bundle setup

lib/search.ts

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),
});

Inngest client and functions

Each stage is a separate step.run — durable, auto-retried (Inngest’s default is four retries per step — five attempts total).

inngest/client.ts

import { Inngest } from "inngest";

export const inngest = new Inngest({ id: "samesake-pipeline" });

inngest/functions.ts

import { inngest } from "./client.ts";
import { app } from "../lib/search.ts";
import { fetchChangedRows, rowFromWebhook } from "../catalog-source.ts";

// Scheduled full-loop — every 5 minutes, one at a time
export const syncCatalog = inngest.createFunction(
  {
    id: "sync-catalog",
    triggers: [{ cron: "*/5 * * * *" }],
    concurrency: 1,
  },
  async ({ step }) => {
    await step.run("upsert", async () => {
      const rows = await fetchChangedRows();
      await app.enrich.upsert(rows);
    });
    await step.run("enrich", () => app.enrich.enrich());
  },
);

// Hourly maintenance
export const catalogMaintenance = inngest.createFunction(
  {
    id: "catalog-maintenance",
    triggers: [{ cron: "0 * * * *" }],
  },
  async ({ step }) => {
    // Image revalidation is a caller-owned @samesake/server workflow; retry failed rows here.
    await step.run("retry-failed", () => app.enrich.retryFailed({ limit: 500 }));
  },
);

// Webhook fast-path — push one row on product update
export const onProductUpdated = inngest.createFunction(
  {
    id: "product-updated",
    triggers: [{ event: "store/product.updated" }],
  },
  async ({ event, step }) => {
    await step.run("upsert", async () => {
      const row = rowFromWebhook(event.data);
      await app.enrich.upsert([row]);
    });
    // Optional: run enrich inline for instant search
    // await step.run("enrich", () => app.enrich.enrich());
  },
);

Tip

Cron strings support timezones: { cron: "TZ=Europe/Paris 0 12 * * 5" }. A function can also have multiple triggers — same handler, cron plus event — if you want one function for both scheduled and manual syncs.

Serve the handler

Expose your functions through Inngest’s serve handler. On Next.js App Router:

app/api/inngest/route.ts

import { serve } from "inngest/next";
import { inngest } from "@/inngest/client";
import { syncCatalog, catalogMaintenance, onProductUpdated } from "@/inngest/functions";

export const maxDuration = 300;

export const { GET, POST, PUT } = serve({
  client: inngest,
  functions: [syncCatalog, catalogMaintenance, onProductUpdated],
});

Other frameworks: inngest/express, inngest/hono, etc. — same shape, different import path.

Fire events from your store webhook

app/api/webhooks/store/route.ts

import { inngest } from "@/inngest/client";

export async function POST(request: Request) {
  const payload = await request.json();
  await inngest.send({
    name: "store/product.updated",
    data: payload,
  });
  return Response.json({ ok: true });
}

Use step.sendEvent("name", events[]) inside a function when you need to fan out — not needed for the catalog loop, because samesake batches internally.

Why this fits samesake

Inngest’s job is durability and observability, not catalog fan-out. Your 50,000-product marketplace still runs two bundle calls per sync; samesake skips unchanged rows inside each call. Separate steps mean a failed embed retry doesn’t re-fetch your Shopify feed.

Note

Row statuses (pending, ready, failed, …) are in Postgres, not Inngest. See Pipeline lifecycle.

Where to go next

• Concept and platform picker — Running the enrich pipeline durably
• Inngest docs — inngest.com/docs

Run it end to end

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