pollForJobs function

Future<void> pollForJobs({ required ApiClient apiClient, required String workerId, })

Implementation

Future<void> pollForJobs({
  required ApiClient apiClient,
  required String workerId,
}) async {
  _log.info('Starting job poller...');

  final state = WorkerState();

  // Send initial heartbeat
  await _sendHeartbeat(apiClient, workerId, state);

  // Set up periodic heartbeat timer
  final heartbeatTimer = Timer.periodic(_heartbeatInterval, (_) {
    _sendHeartbeat(apiClient, workerId, state);
  });

  Timer? spinnerTimer;
  var spinnerIndex = 0;
  var lastUpdateCheck = DateTime.now();

  void startSpinner() {
    spinnerTimer?.cancel();
    spinnerTimer = Timer.periodic(const Duration(milliseconds: 100), (_) {
      final now = DateTime.now();
      final time =
          '${now.hour.toString().padLeft(2, '0')}:${now.minute.toString().padLeft(2, '0')}:${now.second.toString().padLeft(2, '0')}';
      final frame = _spinnerFrames[spinnerIndex % _spinnerFrames.length];
      stderr.write('\r$time $frame [Poller] Waiting for jobs...  ');
      spinnerIndex++;
    });
  }

  void stopSpinner() {
    if (spinnerTimer != null) {
      spinnerTimer!.cancel();
      spinnerTimer = null;
      stderr.writeln('');
    }
  }

  Future<void> tryAutoUpdate() async {
    lastUpdateCheck = DateTime.now();
    final updated = await checkAndUpdate();
    if (updated) {
      _log.info('Update installed. Exiting for restart...');
      state.status = 'stopping';
      await _sendHeartbeat(apiClient, workerId, state);
      heartbeatTimer.cancel();
      exit(exitCodeUpdateRequested);
    }
  }

  state.status = 'idle';
  await _sendHeartbeat(apiClient, workerId, state);

  try {
    while (true) {
      try {
        final bool jobFound;
        if (Platform.isLinux) {
          jobFound = await processDockerJob(
            apiClient,
            workerId,
            onJobFound: () {
              stopSpinner();
              state.status = 'busy';
              _sendHeartbeat(apiClient, workerId, state);
            },
          );
        } else {
          jobFound = await processJob(
            apiClient,
            workerId,
            onJobFound: () {
              stopSpinner();
              state.status = 'busy';
              _sendHeartbeat(apiClient, workerId, state);
            },
          );
        }

        if (jobFound) {
          _log.info('Job completed, checking for next...');
          state.status = 'idle';
          await _sendHeartbeat(apiClient, workerId, state);
          await tryAutoUpdate();
        } else {
          if (state.status == 'error') {
            state.status = 'idle';
            await _sendHeartbeat(apiClient, workerId, state);
          }
          final now = DateTime.now();
          if (now.difference(lastUpdateCheck) >= _updateCheckInterval) {
            stopSpinner();
            await tryAutoUpdate();
          }
          if (spinnerTimer == null) startSpinner();
          await Future.delayed(const Duration(seconds: 10));
        }
      } catch (e, s) {
        stopSpinner();
        _log.severe('Error in poll loop: $e');
        state.status = 'error';
        await _sendHeartbeat(apiClient, workerId, state);
        await Sentry.captureException(e, stackTrace: s);
        await Future.delayed(const Duration(seconds: 10));
      }
    }
  } finally {
    heartbeatTimer.cancel();
    state.status = 'stopping';
    await _sendHeartbeat(apiClient, workerId, state);
  }
}