Sockets

Two WebSocket endpoints, both on the queue's own domain and both upgraded transparently by Caddy: the monitor socket for producers watching items, and the work socket for workers. Every message is a single JSON object. Because a browser can't set request headers on a WebSocket, a gated socket takes a user's token as ?token=<token> in the URL.

Monitor: GET /<queue>/monitor

Requires fetch access. A producer subscribes to item ids and the server pushes their status: once immediately on subscribe, then again whenever a subscribed item changes.

Client → server

MessageEffect
{ "subscribe": ["<id>", …] }Start watching these ids. Each valid id gets an immediate status reply with its current state.
{ "unsubscribe": ["<id>", …] }Stop watching these ids.

You can send either at any time on the open socket; unknown or malformed ids are ignored. A single socket watches one queue.

Server → client

{
  "type": "status",
  "id": "9f2c1a0b7d4e6f83",
  "status": "processing",
  "postedBy": "alice",
  "postedAt": 1751500000000,
  "updatedAt": 1751500012000,
  "finishedAt": null,
  "attempts": 1,
  "progress": { "percent": 40 },
  "result": null,
  "error": null
}

The payload is the same status view as GET /<queue>/items/<id>. A push that first delivers a terminal result to a watching producer also stamps the item's fetched clock, exactly like an HTTP fetch.

Example

const ws = new WebSocket("wss://task.example.com/resize/monitor?token=USER_TOKEN");
ws.onopen = () => ws.send(JSON.stringify({ subscribe: ["9f2c1a0b7d4e6f83"] }));
ws.onmessage = (ev) => {
  const m = JSON.parse(ev.data);   // { type: "status", ... }
  if (m.status === "finished") ws.send(JSON.stringify({ unsubscribe: [m.id] }));
};

Work: GET /<queue>/work

Requires work access. A worker claims, reports progress and posts results over one connection. Its defining extra: if the socket drops, every item still leased to it is re-queued immediately, the fast unlock path alongside the lease timeout.

Client → server

MessageEffect
{ "claim": { "criteria": {…}, "worker": "id?" } }Claim one matching pending item. Both fields optional.
{ "progress": { "id": "<id>", "value": {…} } }Post a progress record (validated against the progress schema).
{ "result": { "id": "<id>", "value": {…} } }Post the result (validated against the result schema); finishes the item if it wasn't terminal.

Server → client

MessageWhen
{ "type": "item", "id", "item", "attempts" }A claim matched: this item is yours, leased.
{ "type": "empty" }A claim matched nothing.
{ "type": "progress-ack", "id", "accepted", "status" }Reply to a progress.
{ "type": "result-ack", "id", "accepted", "status" }Reply to a result; accepted:false if the item was already terminal.
{ "type": "error", "id", "errors": [ … ] }A progress or result failed schema validation; nothing was stored.

Example

const ws = new WebSocket("wss://task.example.com/resize/work?token=USER_TOKEN");
const next = () => ws.send(JSON.stringify({ claim: { worker: "box-1", criteria: { format: "jpg" } } }));

ws.onopen = next;
ws.onmessage = async (ev) => {
  const m = JSON.parse(ev.data);
  if (m.type === "empty")       setTimeout(next, 2000);
  else if (m.type === "item") {
    const output = await doWork(m.item);
    ws.send(JSON.stringify({ result: { id: m.id, value: { output } } }));
  }
  else if (m.type === "result-ack") next();   // m.accepted tells you if it counted
};
On disconnect the server re-queues only the items this socket is still holding (those it claimed and hasn't had a result accepted for). A result already accepted has moved the item to done and is untouched.

Auth & connection notes