Job tracker

1// A backend job — video encoding, CSV parsing, an AI generation
2// call, image optimization — doesn't finish in the same request that
3// starts it. Something needs to watch it and report back.
4//
5// Flux supports two transports for that watching, unified behind
6// ONE interpreter function so the calling code never has to know or
7// care which one is active:
8 
9// Transport A — realtime. The job's row lives in a table Flux
10// already has a channel for. Every UPDATE is a normal realtime
11// event, interpreted the same way any other row change would be.
12 
13// Transport B — polling. No realtime table for this job at all —
14// common for a queue-backed worker with no live row to subscribe to.
15// A configured pollFn is called on an interval instead.
16 
17// Both transports funnel into the exact same interpreter:
18// processJobTrackerEvent(). It has no idea which transport produced
19// the event it's looking at, and doesn't need to.

1// The common case — a job type tied to a store you've already
2// registered. jobTracker piggybacks on that registration entirely;
3// no separate call needed to start tracking.
4 
5flux.register({
6  store: videoJobsStore,
7  channel: 'video_jobs',
8  event: 'UPDATE',
9  idbKey: 'video_jobs',
10  ttl: 'short',
11  scope: 'user',
12  jobTracker: {
13    trackBy: 'job_id',
14    scope: 'user', // inherits reg.scope if omitted — see Step 5
15    onProgress: (store, percent) => store.setEncodeProgress(percent),
16    onComplete: (store, payload) => store.markEncodeDone(payload),
17    onFailure:  (store, error)   => store.markEncodeFailed(error),
18  },
19})
20 
21// Because table is set on the registration, this uses the realtime
22// transport — every UPDATE on video_jobs runs through the
23// interpreter. If table were omitted and pollFn were provided
24// instead, the same config would use polling. table is ignored for
25// jobTracker purposes if pollFn is also set AND table is undefined —
26// see Step 3 for exactly when each transport wins.

1// The interpreter doesn't require a fixed row shape — it checks a
2// priority-ordered list of common field names, so it works against
3// whatever your backend already calls these columns.
4 
5// Status field — checked in this order, first match wins:
6//   status, job_status, state, jobState
7 
8// Progress field — checked in this order:
9//   progress, percent, percent_complete
10 
11// Terminal SUCCESS values (any of these fires onComplete, once):
12//   complete, completed, done, finished, success, succeeded
13 
14// Terminal FAILURE values (any of these fires onFailure, once):
15//   failed, failure, error, cancelled, canceled
16 
17// All of the above are Flux-native conventions — not project-
18// specific strings you invent. If your backend uses something
19// outside this list (e.g. a numeric status enum), map it to one of
20// these strings before Flux ever sees the row, rather than expecting
21// the interpreter to learn a new vocabulary.
22 
23// Every terminal value — success or failure — fires its callback
24// EXACTLY ONCE per job. Internal state cleans up immediately on that
25// transition, so a job can't accidentally fire onComplete twice from
26// two UPDATE events that both happen to carry the same terminal
27// status.

1// The polling path — for jobs with no realtime table to subscribe
2// to. Configured on the same jobTracker object, just swap table for
3// pollFn:
4 
5flux.register({
6  store: csvJobsStore,
7  channel: 'csv_jobs',
8  idbKey: 'csv_jobs',
9  ttl: 'short',
10  // no "table" field — this store never opens a realtime channel
11  jobTracker: {
12    trackBy: 'job_id',
13    pollFn: () => fetch('/api/csv_jobs/status').then(r => r.json()),
14    pollIntervalMs: 3000, // default — see the recommended values below
15    onProgress: (store, percent) => store.setParseProgress(percent),
16    onComplete: (store, payload) => store.markParseDone(payload),
17    onFailure:  (store, error)   => store.markParseFailed(error),
18  },
19})
20 
21// pollIntervalMs recommendations, by urgency:
22//   1000-1500ms   high urgency — payment verification, AI streaming fallback
23//   3000ms        default — image optimization, CSV parsing
24//   10000-30000ms long-running — video encoding, deep database audits
25 
26// The effective interval is pollIntervalMs PLUS however long the
27// fetch itself takes — the next poll only starts once the current
28// one fully resolves. This prevents request stacking if the network
29// is slow; it does NOT guarantee exactly-every-N-ms polling.
30 
31// IMPORTANT: pollFn is ignored entirely if reg.table IS set. Setting
32// both table and pollFn on the same jobTracker config doesn't merge
33// them — realtime wins, pollFn is silently unused. Pick one
34// transport per job type.

1// Polling doesn't run on every open tab — only the tab that holds
2// the realtime leader lock (Section 4.4) actually polls. Every other
3// tab's poller exists but stays dormant.
4 
5// registerStore() creates a JobPoller for every tab, but never calls
6// .start() itself. The poller only activates when THIS tab wins the
7// realtime leader Web Lock and RealtimeManager calls
8// registry.notifyLeaderElected(), which promotes every stored poller
9// to active.
10 
11// This means: 5 tabs open on a page with a pollFn-based job tracker
12// -> exactly 1 tab is actually calling pollFn on an interval, not 5.
13// If the leader tab closes, leadership transfers (Section 4.4) and
14// the newly-elected tab's poller picks up polling from there —
15// there's a brief gap during transfer, not a guaranteed
16// zero-interruption handoff.
17 
18// Polling also pauses automatically on visibilitychange to hidden
19// and the offline event, and only resumes on visible/online if this
20// tab is STILL the leader at that point — a tab that lost leadership
21// while backgrounded does not resume polling when it becomes visible
22// again.

1// Not every job exists at registration time. The offline-upload
2// pattern (offline/upload_pattern) is the canonical example: a file
3// is queued while offline, and there's no job_id to track until the
4// upload actually lands and the server hands one back.
5 
6// flux.trackJob() covers exactly this — starting to track a job
7// AFTER something else (usually onReplaySuccess) discovers its id:
8 
9queueConfig: {
10  storeName: 'uploads',
11  replayEndpoint: '/api/uploads',
12  onReplaySuccess: (responseBody, entry) => {
13    flux.trackJob(responseBody.job_id, uploadsStore, {
14      trackBy: 'job_id',
15      scope: 'user', // see the callout below — this is NOT optional
16                       // in practice for a private upload flow
17      onProgress: (store, percent) => store.setUploadProgress(percent),
18      onComplete: (store, payload) => store.markUploadDone(payload),
19      onFailure:  (store, error)   => store.markUploadFailed(error),
20    })
21  },
22}
23 
24// trackJob() returns a cleanup function, same contract as register():
25const stopTracking = flux.trackJob(jobId, store, config)
26// stopTracking() // unregisters this dynamic job early if ever needed

1// scope resolution differs between the two registration paths — one
2// inherits automatically, the other doesn't, and this is the single
3// most common mistake with job tracking on private data.
4 
5// STATIC registration (jobTracker on a StoreRegistration) — inherits
6// automatically if omitted:
7//   config.scope ?? parentRegistration.scope ?? 'global'
8 
9flux.register({
10  channel: 'dashboard_jobs',
11  scope: 'user', // <- jobTracker below inherits THIS
12  jobTracker: {
13    trackBy: 'job_id',
14    // no scope set here — resolves to 'user' automatically
15  },
16})
17 
18// DYNAMIC registration (flux.trackJob()) — has no parent
19// registration to inherit FROM. There is nothing for it to fall back
20// to except the hardcoded 'global' default:
21 
22flux.trackJob(jobId, store, {
23  trackBy: 'job_id',
24  // scope omitted here -> 'global', ALWAYS, regardless of whose
25  // upload this is. A private user upload tracked this way will
26  // NOT be cleared by clearUserSession() unless scope: 'user' is
27  // set explicitly, every single time.
28})

1// Two different teardown functions exist, and using the wrong one
2// either wipes too much or too little:
3 
4// clearJobTrackerState() — unconditional FULL wipe. Called only from
5// RealtimeManager.destroy() and engine.destroy() — genuine app
6// teardown, not logout. Calling this from a logout handler would
7// also destroy a public system-status job tracker that has nothing
8// to do with the user who's leaving.
9 
10// clearJobTrackerByScope('user') — selective purge. This is what
11// flux.clearUserSession() actually calls (Section 4.17) — removes
12// only 'user'-scoped jobStates and dynamicJobs entries, leaves every
13// 'global'-scoped job tracker untouched.
14 
15await flux.clearUserSession()
16// internally calls clearJobTrackerByScope('user') as one step among
17// several (queue purge, IDB sweep, anchor clearing — Section 4.17)
18 
19// This is exactly why the scope: 'user' step above matters so much
20// for trackJob() — a dynamic job tracked without an explicit scope
21// silently survives logout, tracking progress for a job that belongs
22// to a session that no longer exists.

1// Everything the quick-import path needs from @tsworldtech/flux_react:
2 
3import { FluxJobPanel } from '@tsworldtech/flux_react'
4import type { FluxJobPanelProps, JobStatus, PendingIntent } from '@tsworldtech/flux_react'
5 
6function UploadStatus() {
7  const { job, progress, status, isQueued, pendingIntent, isOnline } = useJobState()
8 
9  return (
10    <FluxJobPanel
11      job={job}
12      progress={progress}
13      status={status}
14      isQueued={isQueued}
15      pendingIntent={pendingIntent}
16      isOnline={isOnline}
17      onRequestReselect={() => openFilePicker()}
18    />
19  )
20}
21 
22// The panel covers seven states out of the box: idle, running,
23// complete, failed, cancelled (see the callout below), queued
24// offline, and pending re-select — the last two designed specifically
25// to slot in next to the FluxUploadGate pattern from
26// offline/upload_pattern.

1// One thing to know before relying on the "cancelled" state as
2// shipped: it is NOT a Flux mechanism. Section 4.7's terminal
3// FAILURE values are generic — failed, failure, error, cancelled,
4// canceled — and a job matching any of them fires onFailure. There
5// is no separate 'cancelled' status Flux itself distinguishes.
6 
7// FluxJobPanel layers its own convention on top: it inspects
8// job.error and looks for the EXACT string 'Cancelled by user' to
9// decide whether to show the amber "cancelled" state instead of the
10// red "failed" state:
11 
12function isCancelledJob(job) {
13  return job?.error === 'Cancelled by user'
14}
15 
16// This string has to come from YOUR OWN server-side cancel endpoint
17// setting job.error to exactly that value — it is not something
18// Flux's tracker sets automatically, and no such convention is
19// documented anywhere in the core engine. If your backend uses a
20// different string, a different field, or a numeric cancel reason
21// code, this check silently never matches and every cancellation
22// renders as a plain "Job failed" instead. Update the string (or the
23// whole check) to match your own API before relying on this
24// distinction.

1// Same reasoning as the upload gate and conflict panel: FluxJobPanel
2// is fully inline-styled with its own hardcoded COLORS object,
3// unrelated to any host app's design tokens, with no theming prop or
4// className pass-through into its internal elements.
5//
6// It's also the component most likely to need a REAL structural
7// edit, not just a recolor — the cancelled-state detection above is
8// baked to one specific backend convention that almost certainly
9// isn't yours. Importing this as a black box means either matching
10// your cancel endpoint's error string to Flux's default, or losing
11// the cancelled/failed distinction entirely. Copying it means fixing
12// isCancelledJob() once, in your own copy, to match however your own
13// backend actually reports a cancellation.

1// packages/flux_react/src/components/FluxJobPanel.tsx
2//
3// Copy into your own project. The two edits almost everyone will
4// want to make immediately: the theme CSS variable mappings, and
5// isCancelledJob() to match your own backend's cancel convention
6// instead of the 'Cancelled by user' default.
7 
8import React from 'react'
9 
10export type JobStatus = 'idle' | 'running' | 'complete' | 'failed'
11 
12export interface PendingIntent {
13  label: string
14  fileName: string
15  fileSize: number
16  fileType: string
17  lastModified: number
18}
19 
20export interface FluxJobPanelProps {
21  job: { id: string; label?: string; error?: string | null; [key: string]: any } | null
22  progress: number
23  status: JobStatus
24  isQueued: boolean
25  pendingIntent: PendingIntent | null
26  isOnline: boolean
27  onRequestReselect?: () => void
28  className?: string
29}
30 
31function formatBytes(bytes: number): string {
32  if (bytes === 0) return '0 B'
33  const k = 1024
34  const sizes = ['B', 'KB', 'MB', 'GB', 'TB']
35  const i = Math.floor(Math.log(bytes) / Math.log(k))
36  return `${parseFloat((bytes / Math.pow(k, i)).toFixed(1))} ${sizes[i]}`
37}
38 
39// ── EDIT THIS — match your own cancel endpoint convention ────────
40function isCancelledJob(job: FluxJobPanelProps['job']): boolean {
41  return job?.error === 'Cancelled by user' // <- your string here,
42                                              // or a different field
43                                              // entirely (job.reason,
44                                              // job.cancelledBy, etc)
45}
46 
47function ProgressBar({ value }: { value: number }) {
48  const clamped = Math.max(0, Math.min(100, value))
49  return (
50    <div style={styles.progressTrack}>
51      <div style={{ ...styles.progressFill, width: `${clamped}%` }} />
52    </div>
53  )
54}
55 
56function StatusDot({ color }: { color: string }) {
57  return (
58    <span
59      style={{
60        display: 'inline-block',
61        width: 8,
62        height: 8,
63        borderRadius: '50%',
64        backgroundColor: color,
65        flexShrink: 0,
66      }}
67    />
68  )
69}
70 
71export function FluxJobPanel({
72  job, progress, status, isQueued, pendingIntent, isOnline, onRequestReselect, className,
73}: FluxJobPanelProps) {
74 
75  if (isQueued && pendingIntent && !isOnline) {
76    return (
77      <div style={styles.panel} className={className}>
78        <div style={styles.row}>
79          <StatusDot color="var(--color-muted)" />
80          <span style={styles.label}>Saved offline</span>
81        </div>
82        <p style={styles.meta}>{pendingIntent.fileName} · {formatBytes(pendingIntent.fileSize)}</p>
83        <p style={styles.hint}>Reconnect to upload and start this job.</p>
84      </div>
85    )
86  }
87 
88  if (isQueued && pendingIntent && isOnline) {
89    return (
90      <div style={{ ...styles.panel, borderColor: 'var(--color-accent)' }} className={className}>
91        <div style={styles.row}>
92          <StatusDot color="var(--color-accent)" />
93          <span style={styles.label}>Ready to upload</span>
94        </div>
95        <p style={styles.meta}>{pendingIntent.fileName} · {formatBytes(pendingIntent.fileSize)}</p>
96        <p style={styles.hint}>You're back online. Re-select the file to continue.</p>
97        <button style={styles.button} onClick={onRequestReselect} type="button">Re-select File</button>
98      </div>
99    )
100  }
101 
102  if (status === 'complete') {
103    return (
104      <div style={styles.panel} className={className}>
105        <div style={styles.row}>
106          <StatusDot color="var(--color-primary)" />
107          <span style={styles.label}>Job complete</span>
108        </div>
109        {job && <p style={styles.meta}>ID: {job.id}</p>}
110      </div>
111    )
112  }
113 
114  if (status === 'failed' && isCancelledJob(job)) {
115    return (
116      <div style={{ ...styles.panel, borderColor: 'var(--color-muted)' }} className={className}>
117        <div style={styles.row}>
118          <StatusDot color="var(--color-muted)" />
119          <span style={styles.label}>Job cancelled</span>
120        </div>
121        {job && <p style={styles.meta}>ID: {job.id}</p>}
122      </div>
123    )
124  }
125 
126  if (status === 'failed') {
127    return (
128      <div style={{ ...styles.panel, borderColor: 'var(--color-faint)' }} className={className}>
129        <div style={styles.row}>
130          <StatusDot color="var(--color-faint)" />
131          <span style={styles.label}>Job failed</span>
132        </div>
133        {job?.error && <p style={styles.hint}>{job.error}</p>}
134        {job && <p style={styles.meta}>ID: {job.id}</p>}
135      </div>
136    )
137  }
138 
139  if (status === 'running' && job) {
140    return (
141      <div style={styles.panel} className={className}>
142        <div style={styles.row}>
143          <StatusDot color="var(--color-accent)" />
144          <span style={styles.label}>{job.label ?? 'Processing…'}</span>
145          <span style={styles.pct}>{Math.round(progress)}%</span>
146        </div>
147        <ProgressBar value={progress} />
148        <p style={styles.meta}>ID: {job.id}</p>
149      </div>
150    )
151  }
152 
153  return (
154    <div style={{ ...styles.panel, ...styles.idle }} className={className}>
155      <span style={styles.idleText}>No active job</span>
156    </div>
157  )
158}
159 
160const styles: Record<string, React.CSSProperties> = {
161  panel: {
162    display: 'flex',
163    flexDirection: 'column',
164    gap: 8,
165    padding: '14px 16px',
166    borderRadius: 10,
167    border: '1px solid color-mix(in srgb, var(--color-accent) 20%, transparent)',
168    backgroundColor: 'var(--color-surface)',
169    color: 'var(--color-primary)',
170    fontFamily: 'inherit',
171    fontSize: 14,
172    minWidth: 240,
173  },
174  row: { display: 'flex', alignItems: 'center', gap: 8 },
175  label: { fontWeight: 600, flex: 1, color: 'var(--color-primary)' },
176  pct: { fontSize: 13, color: 'var(--color-muted)', fontVariantNumeric: 'tabular-nums' },
177  meta: { margin: 0, fontSize: 12, color: 'var(--color-muted)' },
178  hint: { margin: 0, fontSize: 12, color: 'var(--color-muted)', lineHeight: 1.5 },
179  progressTrack: {
180    height: 6,
181    borderRadius: 3,
182    backgroundColor: 'color-mix(in srgb, var(--color-accent) 15%, transparent)',
183    overflow: 'hidden',
184  },
185  progressFill: {
186    height: '100%',
187    borderRadius: 3,
188    backgroundColor: 'var(--color-accent)',
189    transition: 'width 0.2s ease',
190  },
191  button: {
192    alignSelf: 'flex-start',
193    padding: '6px 12px',
194    borderRadius: 6,
195    border: '1px solid var(--color-accent)',
196    backgroundColor: 'transparent',
197    color: 'var(--color-accent)',
198    fontSize: 13,
199    fontWeight: 600,
200    cursor: 'pointer',
201    lineHeight: 1,
202  },
203  idle: { alignItems: 'center', justifyContent: 'center', padding: '20px 16px' },
204  idleText: { color: 'var(--color-muted)', fontSize: 13 },
205};

1// Job tracking (both static and dynamic, both transports) is a
2// Pro/Enterprise-tier capability (Section 4.12's tier matrix) — not
3// console-enforced against the component itself, but the underlying
4// engine methods (trackJob, and jobTracker on register()) are what's
5// actually gated. Free and Starter tiers don't have a meaningful
6// fallback here the way fire_and_forget covers handshake replay's
7// gate — there's no degraded version of job tracking below Pro.