motorsignal
/
MotorSignal-App
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Harden R2 upload validation and HTTP retry classification

This commit is contained in:
bponders12-cell 2026-09-01 16:56:59 -04:00
parent d4ffddafb0
commit 693f5f8aec
1 changed files with 52 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

167
lib/services/sync/r2_upload_service.dart
View File

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