Track Progress

Track job and workflow progress in the UI with reactive stores that subscribe to status updates via SSE.

The Code

import { trackExportUsers } from "$lib/forge/api";

const job = trackExportUsers({ userId: "abc-123" });

// Reactive state
$: if ($job.status === "completed") {
  downloadFile($job.output);
}

The store starts the job, subscribes to status updates, and exposes unsubscribe() so you can stop local updates when the component unmounts.

What Happens

Calling trackExportUsers() does three things:

1. Starts the job by calling the backend function
2. Registers for SSE updates with the gateway
3. Returns a reactive store that updates as the job progresses

The subscription uses the same SSE connection as query subscriptions. When your component unmounts, Svelte detaches the store subscription for that component automatically.

Job State

interface JobState<TOutput> {
  jobId: string;
  status: JobStatus;
  progress: number | null; // 0-100
  message: string | null; // "Processing row 500..."
  output: TOutput | null; // Result when completed
  error: string | null;
}

type JobStatus =
  | "pending"
  | "claimed"
  | "running"
  | "completed"
  | "retry"
  | "failed"
  | "dead_letter"
  | "cancel_requested"
  | "cancelled"
  | "not_found";

Field	Description
jobId	Unique job identifier
status	Current job status
progress	Percentage complete (0-100), reported by ctx.progress()
message	Status message from ctx.progress(percent, message)
output	Job return value when status is completed
error	Error message when status is failed, dead_letter, or cancelled

Workflow State

Workflows expose additional fields for step-by-step tracking:

interface WorkflowState<TOutput> {
  workflowId: string;
  status: WorkflowStatus;
  step: string | null; // Current step name
  waitingFor: string | null; // Reserved for future event details; currently null
  steps: WorkflowStepState[]; // All steps with status
  output: TOutput | null;
  error: string | null;
}

interface WorkflowStepState {
  name: string;
  status:
    | "pending"
    | "running"
    | "completed"
    | "failed"
    | "compensated"
    | "skipped";
  error: string | null;
}

type WorkflowStatus =
  | "created"
  | "running"
  | "waiting"
  | "completed"
  | "compensating"
  | "compensated"
  | "failed"
  | "not_found";

Patterns

Progress Bar

<script>
  import { trackExportUsers } from "$lib/forge/api";

  const job = trackExportUsers({ format: "csv" });
</script>

{#if $job.status === "running"}
  <div class="progress-bar">
    <div class="fill" style="width: {$job.progress ?? 0}%"></div>
    <span>{$job.message ?? "Processing..."}</span>
  </div>
{:else if $job.status === "completed"}
  <a href={$job.output.downloadUrl}>Download</a>
{:else if $job.status === "failed"}
  <p class="error">{$job.error}</p>
{/if}

Workflow Stepper

<script>
  import { trackAccountVerification } from "$lib/forge/api";

  const workflow = trackAccountVerification({ userId });
</script>

<ol class="steps">
  {#each $workflow.steps as step}
    <li class:current={step.name === $workflow.step} class:done={step.status === "completed"}>
      {step.name}
      {#if step.status === "failed"}
        <span class="error">{step.error}</span>
      {/if}
    </li>
  {/each}
</ol>

Handle All States

<script>
  import { trackSendReport } from "$lib/forge/api";

  let job = null;

  function startExport() {
    job = trackSendReport({ reportId: "q4-2024" });
  }
</script>

<button onclick={startExport} disabled={job && $job.status === "running"}>
  Generate Report
</button>

{#if job}
  {#if $job.status === "pending" || $job.status === "claimed"}
    <p>Waiting to start...</p>
  {:else if $job.status === "running"}
    <p>Processing: {$job.progress ?? 0}%</p>
  {:else if $job.status === "retry"}
    <p>Retrying after temporary failure...</p>
  {:else if $job.status === "completed"}
    <p>Done! Result: {JSON.stringify($job.output)}</p>
  {:else if $job.status === "failed"}
    <p>Failed: {$job.error}</p>
  {:else if $job.status === "dead_letter"}
    <p>Permanently failed after max retries</p>
  {:else if $job.status === "cancel_requested"}
    <p>Cancellation requested...</p>
  {:else if $job.status === "cancelled"}
    <p>Job cancelled</p>
  {/if}
{/if}

Manual Unsubscribe

For stores created outside component lifecycle:

const job = trackExportUsers({ userId });

// Later, clean up manually
job.unsubscribe();

Store Interface

JobStore

interface JobStore<TOutput> {
  subscribe: (
    run: (value: JobState<TOutput> & { loading: boolean }) => void,
  ) => () => void;
  unsubscribe: () => void;
}

WorkflowStore

interface WorkflowStore<TOutput> {
  subscribe: (
    run: (value: WorkflowState<TOutput> & { loading: boolean }) => void,
  ) => () => void;
  unsubscribe: () => void;
}

Both stores include a loading: boolean field that is true until the initial job/workflow is created.

Generated Functions

For each job:

#[forge::job]
pub async fn export_users(ctx: &JobContext, format: String) -> Result<ExportResult> {
    // ...
}

Forge generates:

export const trackExportUsers = (args: { format: string }) =>
  createJobStore<{ format: string }, ExportResult>("export_users", args);

For workflows, the pattern is identical:

#[forge::workflow]
pub async fn account_verification(ctx: &WorkflowContext, user_id: Uuid) -> Result<()> {
    // ...
}

Generates:

export const trackAccountVerification = (args: { userId: string }) =>
  createWorkflowStore<{ userId: string }, void>("account_verification", args);

Under the Hood

SSE transport: Job updates arrive over the same SSE connection used for query subscriptions. The client registers for job-specific updates via /_api/subscribe-job:

await client._registerJob(clientSubId, jobId);

Connection lifecycle: If the SSE connection drops, the client reconnects with exponential backoff. Query subscriptions are re-registered automatically; recreate job and workflow stores after reconnect if you need fresh tracking on the new session.

Automatic cleanup: Svelte's store contract means the unsubscribe callback runs when the last subscriber unsubscribes. For job and workflow stores, unsubscribe() stops local updates for that store instance.

Server-side updates: Progress updates are pushed via PostgreSQL LISTEN/NOTIFY. The gateway receives notifications and broadcasts to relevant SSE connections:

NOTIFY forge_job_updates, '{"job_id": "...", "status": "running", "progress": 50}'