Keep R2 response size advisory for compatibility with worker responses
This commit is contained in:
parent
5884d6cb30
commit
7ed8af5b86
|
|
@ -13,7 +13,6 @@ class R2UploadResult {
|
|||
final int? statusCode;
|
||||
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);
|
||||
const R2UploadResult.authRequired(String message, {int statusCode = 401}) : this(success: false, errorMessage: message, statusCode: statusCode, isRetryable: true, requiresAuth: true);
|
||||
|
|
@ -21,74 +20,46 @@ class R2UploadResult {
|
|||
const R2UploadResult.permanentFailure(String message, {int? statusCode}) : this(success: false, errorMessage: message, statusCode: statusCode, isRetryable: false);
|
||||
}
|
||||
|
||||
/// 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');
|
||||
final http.Client _client;
|
||||
final String _workerBaseUrl;
|
||||
|
||||
R2UploadService({http.Client? client, String? workerBaseUrl})
|
||||
: _client = client ?? http.Client(),
|
||||
_workerBaseUrl = (workerBaseUrl ?? defaultWorkerUrl).replaceAll(RegExp(r'/+$'), '');
|
||||
|
||||
R2UploadService({http.Client? client, String? workerBaseUrl}) : _client = client ?? http.Client(), _workerBaseUrl = (workerBaseUrl ?? defaultWorkerUrl).replaceAll(RegExp(r'/+$'), '');
|
||||
String get workerBaseUrl => _workerBaseUrl;
|
||||
|
||||
Future<R2UploadResult> uploadTelemetryBundle({
|
||||
required String idToken,
|
||||
required String remoteRideId,
|
||||
required Uint8List zipBytes,
|
||||
required String sha256Hash,
|
||||
required String appVersion,
|
||||
Duration timeout = const Duration(seconds: 45),
|
||||
}) async {
|
||||
Future<R2UploadResult> uploadTelemetryBundle({required String idToken, required String remoteRideId, required Uint8List zipBytes, required String sha256Hash, 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.trim()}',
|
||||
'Content-Type': 'application/zip',
|
||||
'X-Remote-Ride-Id': remoteRideId,
|
||||
'X-Bundle-Sha256': sha256Hash.toLowerCase(),
|
||||
'X-App-Version': appVersion,
|
||||
},
|
||||
body: zipBytes,
|
||||
).timeout(timeout);
|
||||
|
||||
final response = await _client.put(Uri.parse('$_workerBaseUrl/r2/upload'), 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);
|
||||
final status = response.statusCode;
|
||||
if (status >= 200 && status < 300) {
|
||||
Map<String, dynamic> data;
|
||||
try {
|
||||
final decoded = jsonDecode(response.body);
|
||||
if (decoded is! Map) throw const FormatException('Invalid worker response');
|
||||
data = Map<String, dynamic>.from(decoded);
|
||||
final data = Map<String, dynamic>.from(decoded);
|
||||
final objectKey = data['objectKey'];
|
||||
if (objectKey is! String || !objectKey.endsWith('/rides/$remoteRideId/telemetry.zip') || !objectKey.startsWith('users/')) return const R2UploadResult.permanentFailure('R2 worker returned an invalid object key.');
|
||||
final size = data['size'];
|
||||
final bytes = size is num && size >= 0 ? size.toInt() : zipBytes.length;
|
||||
// The worker may report an encoded/storage size independently; the upload
|
||||
// is already cryptographically identified by X-Bundle-Sha256.
|
||||
return R2UploadResult.success(objectKey: objectKey, bytesWritten: bytes);
|
||||
} 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 (status == 401 || status == 403) {
|
||||
return R2UploadResult.authRequired('R2 Worker authorization denied ($status).', statusCode: status);
|
||||
}
|
||||
if (status == 400 || status == 404 || status == 409 || status == 422) {
|
||||
return R2UploadResult.permanentFailure('R2 Worker rejected the bundle ($status).', statusCode: status);
|
||||
}
|
||||
// 408, 425, 429 and all 5xx are transient; the SyncManager owns backoff.
|
||||
if (status == 401 || status == 403) return R2UploadResult.authRequired('R2 Worker authorization denied ($status).', statusCode: status);
|
||||
if (status == 400 || status == 404 || status == 409 || status == 422) return R2UploadResult.permanentFailure('R2 Worker rejected the bundle ($status).', statusCode: status);
|
||||
return R2UploadResult.temporaryFailure('R2 Worker temporary error ($status).', statusCode: status);
|
||||
} on TimeoutException {
|
||||
return R2UploadResult.temporaryFailure('Upload to Cloudflare R2 timed out after ${timeout.inSeconds} seconds.');
|
||||
|
|
|
|||
Loading…
Reference in New Issue