Notifying Clients When Processing Finishes #

Make the asset row the single source of truth — every pipeline step updates status in the same transaction that records its outputs — publish each change to a per-user channel (Postgres NOTIFY, Redis pub/sub or a managed bus), stream it to open browser tabs over Server-Sent Events, let clients that cannot hold a stream poll GET /assets/:id with If-None-Match, and deliver signed webhooks to API integrations.

The upload bar reaching 100% is not the end of the user’s wait: the video still has to transcode, the photo still needs variants, the audio needs captions. Users who see nothing after 100% re-upload, refresh or leave. A notification path that reliably says “processing”, then “ready” or “failed, here is why”, removes the most common support ticket in media products. This page belongs to media job orchestration in media processing and delivery pipelines. The transport side of streaming events to the browser is covered in depth in streaming upload progress with Server-Sent Events.

When to use this approach #

• Processing takes seconds to minutes after the upload completes, and the UI needs to change when it finishes.
• The user may be on another page or tab when it finishes, or may have closed the laptop and come back later.
• Third parties upload through your API and need to be told when their asset is ready without polling.

Prerequisites #

1. An assets table with status, updated_at and a monotonically increasing version column.
2. PostgreSQL 13+ (for LISTEN/NOTIFY) or Redis 6+ for pub/sub between workers and the web tier.
3. Node 20+ HTTP server able to hold long-lived responses (SSE), behind a proxy with buffering disabled for the stream route.
4. For webhooks: an HMAC secret per integration and a queue for delivery retries.

One status, three ways to hear about it #

Implementation #

The write side — every worker calls this at the end of a step:

import pg from "pg";

export const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });

export type Status = "uploaded" | "processing" | "ready" | "failed";

export async function setStatus(
  assetId: string,
  status: Status,
  detail: Record<string, unknown> = {},
): Promise<number> {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const { rows } = await client.query(
      `UPDATE assets
          SET status = $2, detail = detail || $3::jsonb, version = version + 1, updated_at = now()
        WHERE id = $1
          AND NOT (status IN ('ready','failed') AND $2 = 'processing')   -- never go backwards
        RETURNING owner_id, version`,
      [assetId, status, JSON.stringify(detail)],
    );
    if (rows.length === 0) { await client.query("ROLLBACK"); return -1; }
    const { owner_id: ownerId, version } = rows[0];
    // NOTIFY is delivered only if the transaction commits — no phantom "ready" events.
    await client.query(`SELECT pg_notify('asset_status', $1)`,
      [JSON.stringify({ assetId, ownerId, status, version })]);
    await client.query("COMMIT");
    return version;
  } catch (err) {
    await client.query("ROLLBACK");
    throw err;
  } finally {
    client.release();
  }
}

The read side — an SSE endpoint per user, fed by one shared LISTEN connection, plus the polling endpoint:

import { createServer, type ServerResponse } from "node:http";
import pg from "pg";
import { pool } from "./status.ts";

const subscribers = new Map<string, Set<ServerResponse>>();   // ownerId → open streams

// One dedicated connection for LISTEN; pool connections must not be used for it.
const listener = new pg.Client({ connectionString: process.env.DATABASE_URL });
await listener.connect();
await listener.query("LISTEN asset_status");
listener.on("notification", (n) => {
  const evt = JSON.parse(n.payload ?? "{}") as { ownerId: string; version: number };
  for (const res of subscribers.get(evt.ownerId) ?? []) {
    res.write(`id: ${evt.version}\nevent: asset\ndata: ${n.payload}\n\n`);
  }
});

function userFrom(req: { headers: Record<string, string | string[] | undefined> }): string | null {
  // Replace with your session lookup; never trust a user ID from the query string.
  const h = req.headers["x-user-id"];
  return typeof h === "string" ? h : null;
}

createServer(async (req, res) => {
  const url = new URL(req.url ?? "/", "http://localhost");
  const user = userFrom(req);
  if (!user) { res.writeHead(401).end(); return; }

  if (url.pathname === "/events") {
    res.writeHead(200, {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-store",
      "X-Accel-Buffering": "no",          // nginx: do not buffer the stream
    });
    res.write("retry: 5000\n\n");         // reconnect delay for EventSource
    const set = subscribers.get(user) ?? new Set();
    set.add(res);
    subscribers.set(user, set);
    const ping = setInterval(() => res.write(": ping\n\n"), 25_000);   // keep proxies from idling out
    req.on("close", () => { clearInterval(ping); set.delete(res); });
    return;
  }

  const m = url.pathname.match(/^\/assets\/([0-9a-f-]{36})$/);
  if (m) {
    const { rows } = await pool.query(
      `SELECT id, status, detail, version FROM assets WHERE id = $1 AND owner_id = $2`, [m[1], user]);
    if (!rows[0]) { res.writeHead(404).end(); return; }
    const etag = `"v${rows[0].version}"`;
    if (req.headers["if-none-match"] === etag) { res.writeHead(304, { ETag: etag }).end(); return; }
    res.writeHead(200, { "Content-Type": "application/json", ETag: etag, "Cache-Control": "no-cache" });
    res.end(JSON.stringify(rows[0]));
    return;
  }
  res.writeHead(404).end();
}).listen(8080);

And the browser, which listens while the tab is open and re-reads the row on reconnect so it never misses a transition:

export function watchAsset(assetId: string, onChange: (a: { status: string; detail: unknown }) => void): () => void {
  let last = 0;
  const refresh = async () => {
    const res = await fetch(`/assets/${assetId}`, { headers: last ? { "If-None-Match": `"v${last}"` } : {} });
    if (res.status === 200) { const a = await res.json(); last = a.version; onChange(a); }
  };
  const es = new EventSource("/events");
  es.addEventListener("asset", (e) => {
    const evt = JSON.parse((e as MessageEvent).data);
    if (evt.assetId === assetId && evt.version > last) void refresh();
  });
  es.addEventListener("open", () => void refresh());    // covers anything missed while disconnected
  return () => es.close();
}

Line-by-line on the decisions that matter #

• pg_notify inside the transaction. PostgreSQL delivers notifications only on commit. A worker that crashes between UPDATE and COMMIT sends nothing, and a client can never see “ready” for a row that is not ready. Publishing to Redis after commit has a gap where the process can die; notifying inside the transaction does not.
• version as the ETag and the SSE id. Every change increments it. Polling with If-None-Match returns a body-less 304 while nothing has changed, and the browser can discard out-of-order events by comparing versions.
• “Never go backwards” in the WHERE. A late retry of an early step must not flip a ready asset back to processing. Guarding in SQL is simpler than coordinating workers.
• Re-read on open. SSE drops events while disconnected (a phone switching networks, a laptop lid). Re-reading the row whenever the stream opens turns a lossy push into a reliable one.
• Heartbeat comments every 25 s. Load balancers and proxies close idle connections after 30–60 s. A comment line keeps the stream alive without waking the client.

Timeline of what the user sees #

Webhooks for API clients #

Integrations that upload through your API want a callback, not a stream. Deliver them from a queue fed by the same status changes, sign every payload, and retry with backoff.

import { createHmac } from "node:crypto";

export async function deliverWebhook(url: string, secret: string, event: object, attempt: number): Promise<boolean> {
  const body = JSON.stringify(event);
  const ts = Math.floor(Date.now() / 1000);
  const sig = createHmac("sha256", secret).update(`${ts}.${body}`).digest("hex");
  const res = await fetch(url, {
    method: "POST",
    headers: { "Content-Type": "application/json", "X-Signature": `t=${ts},v1=${sig}`, "X-Attempt": String(attempt) },
    body,
    signal: AbortSignal.timeout(10_000),
  }).catch(() => null);
  return Boolean(res && res.status >= 200 && res.status < 300);
}

Include the timestamp in the signed string so receivers can reject replays older than a few minutes, and include the asset version in the payload so receivers can ignore a late retry of an older event.

Configuration gotchas #

SSE events arrive in bursts, minutes late. A proxy is buffering the response. nginx needs proxy_buffering off (or the X-Accel-Buffering: no header), and compression middleware must skip text/event-stream, or it waits for a full buffer before flushing.

ERR_HTTP2_PROTOCOL_ERROR or six-connection limits. Over HTTP/1.1, browsers allow six connections per origin, and every tab’s EventSource holds one. Serve the stream over HTTP/2, or use a SharedWorker or BroadcastChannel so only one tab per browser holds the connection.

Notifications lost when the web tier restarts. LISTEN only receives notifications sent while connected. That is why clients re-read on open; for webhooks, drive delivery from a durable outbox table, not from the notification itself.

pg_notify payload too long. The limit is 8000 bytes. Send identifiers and the new status, never the asset’s full detail; clients fetch the rest.

Verification #

# 1. Open a stream as user u1 and leave it running.
curl -N -H 'X-User-Id: u1' http://localhost:8080/events

# 2. In another shell, simulate a worker finishing.
node -e 'import("./status.ts").then(m => m.setStatus("3f0a2b6c-9d41-4a77-8d02-5c1b7e9a6f30", "ready", { hls: "v1/master.m3u8" }))'
# The first shell prints:
# id: 7
# event: asset
# data: {"assetId":"3f0a…","ownerId":"u1","status":"ready","version":7}

# 3. Poll with the current ETag and expect 304 until the next change.
curl -s -o /dev/null -w '%{http_code}\n' -H 'X-User-Id: u1' -H 'If-None-Match: "v7"' \
  http://localhost:8080/assets/3f0a2b6c-9d41-4a77-8d02-5c1b7e9a6f30
# 304

Frequently Asked Questions