Skip to main content

Progress streaming (SSE)

A workflow streams live progress over Server-Sent Events. streamEvents(workflowId) opens the stream and yields typed events you iterate as an async iterable. Reach for it when you want to handle the events yourself — driving a progress bar, or watching a long-running video operation.

run() is the managed alternative: it waits for you, streaming over SSE first and falling back to polling only if the stream cannot be opened or drops. It is not a polling API — see Forcing polling in run() if you want polling specifically.

The stream is on a second host​

Live progress is not served from the API host: the stream has its own public host, and the SDK reads it from the contract's declaration. It never derives the stream host from baseUrl. When you build the client with gisl.create() / Gisl::create(), the stream host resolves in this order:

#SourceNotes
1streamBaseUrl optionMoves the stream only; uploads, workflow creation and downloads stay on baseUrl.
2environment option'prod' / 'staging' (PHP: Environment::Prod / Environment::Staging).
3GISL_STREAM_BASE_URLProcess environment.
4GISL_ENVIRONMENTProcess environment, same names as the option.
5nothing setThe production stream host, only if the API host also defaulted.

A client constructed directly (new GislClient(...) / GislClientConfig) uses only its streamBaseUrl. The value must be an absolute http(s) URL with no query or fragment; anything else throws GislConfigError at construction rather than being treated as absent.

import { gisl } from '@giveitsmaller/sdk';

const client = await gisl.create({
  apiKey: 'REPLACE_ME_API_KEY',
  baseUrl: 'https://gisl-proxy.internal.example',
  streamBaseUrl: 'https://gisl-stream.internal.example',
});

When no stream host is declared — in practice, a custom baseUrl with none of the stream sources above — streamEvents() throws GislStreamHostNotDeclaredError (a GislConfigError) before any I/O, and its message names the ways to declare one. It does not fall back to baseUrl. run() behaves differently on purpose: it treats a missing stream host as "SSE is unavailable here" and polls instead, so it never raises this error.

⚠️ Setting baseUrl does not move the stream. A client built with only baseUrl (no environment, streamBaseUrl or stream environment variable) has no stream host, so every run() polls and onProgress receives no processing events. To stream through a custom API host, also pass environment or streamBaseUrl (or set GISL_STREAM_BASE_URL).

Seeing which transport a run used​

The fallback is not silent:

  • Every waited result says how it finished. transport on the result of run() and Handle wait() is 'sse' when the stream delivered the terminal status and 'polling' when a status poll did (PHP: RunTransport::Sse / RunTransport::Polling). It is the final transport, one value: a run whose stream opened, sent progress and then dropped reports 'polling'; the progress it streamed is what onProgress received. Handle result() does not wait, so its transport is absent (PHP: null). Both toJSON() and toArray() serialise it as the last key.
  • A client with no stream host warns once. The first run that wants SSE but has no stream host emits one warning per client, not per run, naming the fix. TypeScript uses Node's process.emitWarning (type GislWarning, code GISL_STREAM_HOST_NOT_DECLARED, exported as GISL_STREAM_HOST_NOT_DECLARED_WARNING), so --no-warnings and process.on('warning') apply; browsers get console.warn. PHP raises E_USER_WARNING through trigger_error(), its message prefixed [GISL_STREAM_HOST_NOT_DECLARED] for your error handler to match. If your handler turns warnings into exceptions (Laravel, Symfony debug), the SDK catches that and writes the message to error_log() instead: a warning never fails a run whose workflow already exists.
  • No warning when a stream host resolved, or when you pass useSSE: false — you chose polling, so there is nothing to tell you.
const result = await client.file('photo.jpg').compress().run();
if (result.transport === 'polling') {
  // No live progress for this run: declare a stream host to get it.
}

Consuming the stream​

for await (const event of await client.streamEvents(workflowId)) {
  switch (event.event) {
    case 'operation.progress':
      console.log(`Progress: ${event.data.progress}%`);
      break;
    case 'workflow.completed':
    case 'workflow.failed':
    case 'workflow.partially_failed':
      // Terminal — break out of the loop.
      return;
  }
}

The stream returns as soon as the server closes it — which it does on a terminal status. Event tags are dot-delimited; match them as plain strings. A tag the SDK does not know (including a frame with no event: line, tagged message) arrives with the event unknown and the raw tag in name: TypeScript { event: 'unknown', name, data }, PHP $event->event === 'unknown' and $event->name (which is null on the contract events).

Event payloads​

event.data ($event->data in PHP) is the frame's JSON, parsed and handed to you unchanged: the keys are the wire's snake_case, in both SDKs.

TagTerminal?data keys (wire)
operation.progressnojob_ref, operation_id, type, progress (0–100), optional status, stage, phase_input_index, phase_total_inputs
operation.completednojob_ref, operation_id, type, status, progress, optional result (result_kind is single or multi)
operation.failednojob_ref, operation_id, type, status, error_code, error_message, optional message_key, message_params
job.completed / job.failednojob_ref, job_id, status
workflow.completedyesworkflow_id, status, optional reason
workflow.partially_failedyesas above
workflow.failedyesas above

The three workflow.* events are the terminal signals — break out of the loop on any of them. Tags the SDK does not know yet still arrive; log and ignore them.

TypeScript: the types are the wire shape

Each GislSseEvent arm types data with a snake_case wire interface (SseOperationProgressWire, SseOperationCompletedWire, SseOperationFailedWire and the rest), which is exactly what the stream yields. Checking event narrows data, and result_kind picks the shape of an operation.completed result, with no casts:

if (event.event === 'operation.completed' && event.data.result?.result_kind === 'multi') {
  for (const out of event.data.result.outputs) console.log(out.page_index, out.download_url);
}
if (event.event === 'operation.failed') {
  console.warn(`${event.data.operation_id} failed: ${event.data.error_code}`);
}

result may be absent on operation.completed; fall back to the downloads endpoint then. If you prefer camelCase, the matching FromJSON helper from the SDK's @giveitsmaller/contracts dependency converts it (SseOperationFailedDataFromJSON(event.data)).

Malformed frames​

A frame whose data: body fails to parse is skipped — the stream stays resilient so a long-running consumer never breaks on one garbled server frame. The skip is silent by default; pass an onParseError callback to observe it. It receives a { raw, event, error } diagnostic identical in shape across both SDKs. This is a low-level hook — the ergonomic run() path always skips silently — and throwing from it aborts the stream.

for await (const event of await client.streamEvents(workflowId, {
  onParseError: ({ raw, event: name, error }) => {
    console.warn(`dropped malformed "${name}" frame: ${error}`, raw);
  },
})) {
  // ...
}

If a terminal frame arrives malformed it is skipped like any other, so the stream ends without a recognised terminal event and run() falls back to polling getWorkflowStatus — bounded latency, no hang.

Forcing polling in run() (useSSE)​

The ergonomic run() streams progress over SSE first and silently falls back to polling getWorkflowStatus if the stream can't be opened or drops. That default (useSSE: true) suits almost everyone. To skip SSE and poll from the start — e.g. behind a proxy that buffers or blocks SSE — opt out per call. It works the same on the operation-first builders and all five file-first run surfaces (file(...), files(...), and the merge / archive / watermark recipes).

// operation-first
await client.compress('./big.mp4', { crf: 28 }).run({ useSSE: false });

// file-first (same options bag)
await client.file('big-video.mp4').compress(OptimizeFor.Balanced).run({ useSSE: false });

When to use SSE vs polling​

⚠️ run() is not the polling option. By default it streams over SSE and falls back to polling; run({ useSSE: false }) is the polling option. The choice below is about who handles the events, not about which transport you get.

Use streamEvents() whenUse run() when
You want each event as it arrivesYou want the final result and nothing else
You're driving a progress UIYou're running a cron / batch job
You want to decide what a dropped stream meansYou want the fallback handled for you
transport
streamEvents()SSE only — raises if no stream host is declared
run()SSE first, automatic fallback to polling; result.transport says which finished it
run({ useSSE: false })polling only

A connection can drop for benign reasons (proxy timeouts, laptop sleep).

The SDK does not reconnect

The SDK opens one stream. There is no retry loop and no Last-Event-ID resumption — it ignores the id: and retry: fields entirely. Events published while you were disconnected are lost and cannot be replayed.

Any reconnect loop is yours to write, and after it exits you must cross-check the terminal state with a workflow-status call — the gap is unrecoverable from the stream alone.

Webhooks are a third option — they push terminal events to your server without holding a connection open, and they do not have this failure mode.

Reconnecting yourself​

A starting point to copy, not an API the SDK ships. It re-opens the stream after a transport failure or an early end, and asks the status endpoint whether the workflow finished while it was disconnected.

These examples authenticate with an API key. A workflow created anonymously (first-party only) also needs its read capability on every read: pass { capability } to both streamEvents() and getWorkflowStatus().

import { GislApiError, GislTransportError } from '@giveitsmaller/sdk';
import type { GislClient, GislSseEvent } from '@giveitsmaller/sdk';

const TERMINAL_EVENTS = ['workflow.completed', 'workflow.failed', 'workflow.partially_failed'];
const TERMINAL_STATUSES = ['completed', 'failed', 'partially_failed', 'cancelled', 'expired',
  'paused_insufficient_credits'];

async function streamUntilTerminal(
  client: GislClient,
  workflowId: string,
  onEvent: (event: GislSseEvent) => void,
  maxAttempts = 5,
): Promise<void> {
  for (let attempt = 1; ; attempt++) {
    try {
      for await (const event of await client.streamEvents(workflowId)) {
        onEvent(event);
        if (TERMINAL_EVENTS.includes(event.event)) return;
      }
    } catch (err) {
      // Retry only what the SDK marks transient. Anything else, including an
      // exception from your own onEvent and a cancellation, is rethrown.
      const retryable =
        (err instanceof GislApiError && err.retryable) || err instanceof GislTransportError;
      if (!retryable || attempt >= maxAttempts) throw err;
    }
    // Events published during the gap are gone; the status endpoint is the truth.
    const { status } = await client.getWorkflowStatus(workflowId);
    if (TERMINAL_STATUSES.includes(status)) return;
    if (attempt >= maxAttempts) throw new Error(`Stream for ${workflowId} kept dropping`);
    await new Promise((r) => setTimeout(r, Math.min(1000 * 2 ** attempt, 30_000)));
  }
}

Either way, finish with one getWorkflowStatus call before trusting the outcome.

Harvesting a partially failed workflow​

workflow.partially_failed still leaves the successful jobs' outputs to collect. Leave the stream on the terminal event, then match job statuses to downloads by job ref. (run() does this for you: successes land in the result and failures in failed[].)

const status = await client.getWorkflowStatus(workflowId);
const { downloads } = await client.getWorkflowDownloads(workflowId);

for (const job of status.jobs) {
  if (job.status !== 'completed') {
    console.log(`failed ${job.ref}: ${job.status}`);
    continue;
  }
  for (const file of downloads.find((d) => d.ref === job.ref)?.files ?? []) {
    console.log(`ok ${job.ref}: ${file.filename} ${file.downloadUrl}`);
  }
}

Parsing a response yourself (TypeScript)​

parseSseStream(response, { signal?, onParseError? }) turns any fetch Response with an SSE body into the same GislSseEvent stream — for when you open the request yourself (a custom agent, extra headers). It handles frames split across chunks, multi-line data:, and keep-alive comments; it ignores id: and retry:. It does not check the status, so check response.ok first, and send the request to the stream host, not the API host. PHP has no public equivalent; use streamEvents().

import { parseSseStream } from '@giveitsmaller/sdk';

const response = await fetch(`${streamBaseUrl}/api/workflows/${encodeURIComponent(workflowId)}/events`, {
  headers: { Authorization: `Bearer ${apiKey}` },
});
if (!response.ok) throw new Error(`stream refused: ${response.status}`);

for await (const event of parseSseStream(response)) {
  // same shape as streamEvents(): raw snake_case data
}

See also​

  • Workflows — what is being streamed.
  • Errors — failures thrown while opening the stream.