processCloudflareJobBatch function

Future<void> processCloudflareJobBatch(CloudflareQueueBatch batch, JobConsumer consumer)

Runs a Routed consumer against one Cloudflare Queue delivery batch.

Terminal outcomes are acknowledged. A retry outcome uses Cloudflare's native per-message retry operation, preserving the platform's delivery attempt counter. Unexpected consumer errors are retried as well, so a transient Worker or storage failure cannot be mistaken for successful processing.

Implementation

Future<void> processCloudflareJobBatch(
  CloudflareQueueBatch batch,
  JobConsumer consumer,
) async {
  for (final delivery in batch.messages) {
    try {
      final body = delivery.body;
      if (body is! Map) {
        delivery.ack();
        continue;
      }
      final payload = <String, Object?>{};
      for (final entry in body.entries) {
        if (entry.key is! String) {
          throw const FormatException(
            'Cloudflare job payload keys must be strings',
          );
        }
        payload[entry.key as String] = entry.value;
      }
      final message = JobMessage.fromJson(payload);
      final result = await consumer.process(
        message,
        // Cloudflare's attempts counter starts at one, while the portable
        // jobs contract (and Stem's internal envelope) is zero-based.
        deliveryAttempt: delivery.attempts < 1 ? 0 : delivery.attempts - 1,
      );
      if (result.state == JobProcessState.retry) {
        delivery.retry(delay: result.retryAfter);
      } else {
        delivery.ack();
      }
    } on Object {
      delivery.retry();
    }
  }
}