Function useJobStream

ui/src/job/useJobStream.ts:68–280

(jobService: JobService)

};

export function useJobStream(jobService: JobService) {
  const [state, setState] = useState<JobState>(INITIAL_STATE);
  const streamRef = useRef<JobStream | null>(null);

  /** Consume a JobStream: drive the React state machine from its
   *  events. Shared by `start` (new submissions) and `attach`
   *  (Fix #14 resume-after-reload). */
  const consumeStream = useCallback((stream: JobStream) => {
    streamRef.current = stream;
    (async () => {
      try {
        for await (const event of stream) {
          switch (event.kind) {
            case JobEventKind.JOB_EVENT_KIND_PROGRESS: {
              const d = event.detail ?? EMPTY_DETAIL;
              setState((s) => {
                // Don't reopen a completed stage with new progress events
                // (e.g. enrichment batches fire "submitting" after parse already completed it)
                const existing = s.stages[event.phase];
                const stageUpdate =
                  existing?.status === 'completed'
                    ? {
                        ...existing,
                        current: d.current,
                        total: d.total,
                        message: event.message,
                      }
                    : {
                        status: 'active' as StageStatus,
                        current: d.current,
                        total: d.total,
                        message: event.message,
                        fileName: d.fileName || undefined,
                        ...(event.phase === JobPhase.JOB_PHASE_FETCHING
                          ? { format: 'bytes' as const }
                          : {}),
                      };
                return {
                  ...s,
                  // Keep "enriching" status during enrichment progress updates
                  status: s.status === 'enriching' ? 'enriching' : s.status,
                  phase: event.phase,
                  message: event.message,
                  detail: d,
                  nodesCreated: d.nodesCreated || s.nodesCreated,
                  relationshipsCreated:
                    d.relationshipsCreated || s.relationshipsCreated,
                  stages: { ...s.stages, [event.phase]: stageUpdate },
                };
              });
              break;
            }
            case JobEventKind.JOB_EVENT_KIND_STAGE_COMPLETE:
              setState((s) => ({
                ...s,
                stages: {
                  ...s.stages,
                  [event.phase]: {