Method pump

backend/src/workflows/workflows.controller.ts:942–1021

()

    };

    const pump = async () => {
      if (!active) {
        return;
      }

      try {
        const { events, cursor } = await this.traceService.listSince(runId, lastSequence, auth);
        if (events.length > 0) {
          const lastId = events[events.length - 1]?.id;
          if (lastId) {
            const parsed = Number.parseInt(lastId, 10);
            if (!Number.isNaN(parsed)) {
              lastSequence = parsed;
            }
          }
          send('trace', { events, cursor: cursor ?? lastSequence.toString() });

          const timestamps = events
            .map((event) => Date.parse(event.timestamp))
            .filter((value) => !Number.isNaN(value));
          if (timestamps.length > 0) {
            const first = Math.min(...timestamps);
            const last = Math.max(...timestamps);
            if (earliestEventTimestamp === null || first < earliestEventTimestamp) {
              earliestEventTimestamp = first;
            }
            if (latestEventTimestamp === null || last > latestEventTimestamp) {
              latestEventTimestamp = last;
            }

            const packets = await this.workflowsService.buildDataFlows(runId, events, {
              baseTimestamp: earliestEventTimestamp ?? first,
              latestTimestamp: latestEventTimestamp ?? last,
            });

            if (packets.length > 0) {
              send('dataflow', { packets });
            }
          }
        }

        const terminal = await this.terminalStreamService.fetchChunks(runId, {
          cursor: terminalCursor,
        });
        if (terminal.chunks.length > 0) {
          terminalCursor = terminal.cursor;
          send('terminal', { runId, ...terminal });
        }

        const { logs: newLogs, cursor: nextCursor } = await this.logStreamService.fetchRecentLogs(
          runId,
          auth?.organizationId ?? null,
          lastLogCursor,
        );
        if (newLogs.length > 0) {
          lastLogCursor = nextCursor ?? lastLogCursor;
          send('logs', { logs: newLogs, cursor: lastLogCursor });
        }