diff --git a/lib/services/sync/r2_upload_service.dart b/lib/services/sync/r2_upload_service.dart index 02db082..83c30d7 100644 --- a/lib/services/sync/r2_upload_service.dart +++ b/lib/services/sync/r2_upload_service.dart @@ -1,9 +1,9 @@ import 'dart:async'; import 'dart:convert'; +import 'dart:typed_data'; import 'package:flutter/foundation.dart'; import 'package:http/http.dart' as http; -/// Result returned from an R2 bundle upload via Cloudflare Worker gateway @immutable class R2UploadResult { final bool success; @@ -14,82 +14,25 @@ class R2UploadResult { final bool isRetryable; final bool requiresAuth; - const R2UploadResult({ - required this.success, - this.objectKey, - this.bytesWritten, - this.errorMessage, - this.statusCode, - this.isRetryable = true, - this.requiresAuth = false, - }); - - const R2UploadResult.success({ - required String objectKey, - int? bytesWritten, - }) : this( - success: true, - objectKey: objectKey, - bytesWritten: bytesWritten, - statusCode: 200, - isRetryable: false, - requiresAuth: false, - ); - - const R2UploadResult.authRequired( - String message, { - int statusCode = 401, - }) : this( - success: false, - errorMessage: message, - statusCode: statusCode, - isRetryable: true, - requiresAuth: true, - ); - - const R2UploadResult.temporaryFailure( - String message, { - int? statusCode, - }) : this( - success: false, - errorMessage: message, - statusCode: statusCode, - isRetryable: true, - requiresAuth: false, - ); - - const R2UploadResult.permanentFailure( - String message, { - int? statusCode, - }) : this( - success: false, - errorMessage: message, - statusCode: statusCode, - isRetryable: false, - requiresAuth: false, - ); + const R2UploadResult({required this.success, this.objectKey, this.bytesWritten, this.errorMessage, this.statusCode, this.isRetryable = true, this.requiresAuth = false}); + const R2UploadResult.success({required String objectKey, int? bytesWritten}) : this(success: true, objectKey: objectKey, bytesWritten: bytesWritten, statusCode: 200, isRetryable: false); + const R2UploadResult.authRequired(String message, {int statusCode = 401}) : this(success: false, errorMessage: message, statusCode: statusCode, isRetryable: true, requiresAuth: true); + const R2UploadResult.temporaryFailure(String message, {int? statusCode}) : this(success: false, errorMessage: message, statusCode: statusCode, isRetryable: true); + const R2UploadResult.permanentFailure(String message, {int? statusCode}) : this(success: false, errorMessage: message, statusCode: statusCode, isRetryable: false); } -/// Service that streams telemetry ZIP bundles directly to Cloudflare R2 -/// through the authenticated Cloudflare Worker gateway. +/// Uploads immutable ride telemetry bundles through the authenticated Worker gateway. class R2UploadService { - static const String defaultWorkerUrl = String.fromEnvironment( - 'R2_WORKER_URL', - defaultValue: 'https://motorsignal-r2-worker.bponders12.workers.dev', - ); - + static const String defaultWorkerUrl = String.fromEnvironment('R2_WORKER_URL', defaultValue: 'https://motorsignal-r2-worker.bponders12.workers.dev'); final http.Client _client; final String _workerBaseUrl; - R2UploadService({ - http.Client? client, - String? workerBaseUrl, - }) : _client = client ?? http.Client(), + R2UploadService({http.Client? client, String? workerBaseUrl}) + : _client = client ?? http.Client(), _workerBaseUrl = (workerBaseUrl ?? defaultWorkerUrl).replaceAll(RegExp(r'/+$'), ''); String get workerBaseUrl => _workerBaseUrl; - /// Uploads a compressed telemetry ZIP bundle to Cloudflare R2 via the worker gateway Future uploadTelemetryBundle({ required String idToken, required String remoteRideId, @@ -98,65 +41,59 @@ class R2UploadService { required String appVersion, Duration timeout = const Duration(seconds: 45), }) async { + if (idToken.trim().isEmpty) return const R2UploadResult.authRequired('Firebase ID token is required for R2 upload.'); + if (zipBytes.isEmpty) return const R2UploadResult.permanentFailure('Telemetry bundle is empty.'); + if (!RegExp(r'^ride_[A-Za-z0-9_-]+$').hasMatch(remoteRideId)) return const R2UploadResult.permanentFailure('Invalid remote ride identifier.'); + if (!RegExp(r'^[a-fA-F0-9]{64}$').hasMatch(sha256Hash)) return const R2UploadResult.permanentFailure('Invalid telemetry bundle SHA-256.'); + final endpoint = Uri.parse('$_workerBaseUrl/r2/upload'); - try { - final response = await _client - .put( - endpoint, - headers: { - 'Authorization': 'Bearer $idToken', - 'Content-Type': 'application/zip', - 'X-Remote-Ride-Id': remoteRideId, - 'X-Bundle-Sha256': sha256Hash, - 'X-App-Version': appVersion, - }, - body: zipBytes, - ) - .timeout(timeout); + final response = await _client.put( + endpoint, + headers: { + 'Authorization': 'Bearer ${idToken.trim()}', + 'Content-Type': 'application/zip', + 'X-Remote-Ride-Id': remoteRideId, + 'X-Bundle-Sha256': sha256Hash.toLowerCase(), + 'X-App-Version': appVersion, + }, + body: zipBytes, + ).timeout(timeout); - if (response.statusCode >= 200 && response.statusCode < 300) { + final status = response.statusCode; + if (status >= 200 && status < 300) { + Map data; try { - final data = jsonDecode(response.body) as Map; - final objectKey = data['objectKey'] as String? ?? - 'users/unknown/rides/$remoteRideId/telemetry.zip'; - final size = data['size'] as int? ?? zipBytes.length; - return R2UploadResult.success(objectKey: objectKey, bytesWritten: size); - } catch (_) { - return R2UploadResult.success( - objectKey: 'users/unknown/rides/$remoteRideId/telemetry.zip', - bytesWritten: zipBytes.length, - ); + final decoded = jsonDecode(response.body); + if (decoded is! Map) throw const FormatException('Invalid worker response'); + data = Map.from(decoded); + } catch (e) { + return R2UploadResult.temporaryFailure('R2 upload response could not be verified: $e', statusCode: status); } + final objectKey = data['objectKey']; + if (objectKey is! String || !objectKey.endsWith('/rides/$remoteRideId/telemetry.zip') || !objectKey.startsWith('users/')) { + return R2UploadResult.permanentFailure('R2 worker returned an invalid object key.'); + } + final size = data['size']; + final bytes = size is num ? size.toInt() : zipBytes.length; + if (bytes != zipBytes.length) { + return R2UploadResult.temporaryFailure('R2 worker confirmed an unexpected bundle size.', statusCode: status); + } + return R2UploadResult.success(objectKey: objectKey, bytesWritten: bytes); } - if (response.statusCode == 401 || response.statusCode == 403) { - return R2UploadResult.authRequired( - 'Cloudflare Worker authorization denied (${response.statusCode}): ${response.body}', - statusCode: response.statusCode, - ); + if (status == 401 || status == 403) { + return R2UploadResult.authRequired('R2 Worker authorization denied ($status).', statusCode: status); } - - if (response.statusCode == 400 || response.statusCode == 422) { - return R2UploadResult.permanentFailure( - 'Worker rejected ride bundle payload (${response.statusCode}): ${response.body}', - statusCode: response.statusCode, - ); + if (status == 400 || status == 404 || status == 409 || status == 422) { + return R2UploadResult.permanentFailure('R2 Worker rejected the bundle ($status).', statusCode: status); } - - // Server error / Cloudflare rate-limit or 5xx -> Retryable - return R2UploadResult.temporaryFailure( - 'Cloudflare Worker temporary error (${response.statusCode}): ${response.body}', - statusCode: response.statusCode, - ); + // 408, 425, 429 and all 5xx are transient; the SyncManager owns backoff. + return R2UploadResult.temporaryFailure('R2 Worker temporary error ($status).', statusCode: status); } on TimeoutException { - return const R2UploadResult.temporaryFailure( - 'Upload to Cloudflare R2 timed out after 45 seconds.', - ); + return R2UploadResult.temporaryFailure('Upload to Cloudflare R2 timed out after ${timeout.inSeconds} seconds.'); } catch (e) { - return R2UploadResult.temporaryFailure( - 'Network error uploading to Cloudflare R2: $e', - ); + return R2UploadResult.temporaryFailure('Network error uploading to Cloudflare R2: $e'); } } }