feat: add authenticated R2 telemetry download service
This commit is contained in:
parent
03932d573f
commit
827d15eb84
|
|
@ -0,0 +1,186 @@
|
||||||
|
import 'dart:async';
|
||||||
|
import 'dart:convert';
|
||||||
|
import 'dart:typed_data';
|
||||||
|
|
||||||
|
import 'package:crypto/crypto.dart';
|
||||||
|
import 'package:flutter/foundation.dart';
|
||||||
|
import 'package:http/http.dart' as http;
|
||||||
|
|
||||||
|
@immutable
|
||||||
|
class R2DownloadResult {
|
||||||
|
final bool success;
|
||||||
|
final String? cloudRideId;
|
||||||
|
final Uint8List? zipBytes;
|
||||||
|
final String? sha256Hash;
|
||||||
|
final String? etag;
|
||||||
|
final String? errorMessage;
|
||||||
|
final int? statusCode;
|
||||||
|
final bool isRetryable;
|
||||||
|
final bool requiresAuth;
|
||||||
|
|
||||||
|
const R2DownloadResult({
|
||||||
|
required this.success,
|
||||||
|
this.cloudRideId,
|
||||||
|
this.zipBytes,
|
||||||
|
this.sha256Hash,
|
||||||
|
this.etag,
|
||||||
|
this.errorMessage,
|
||||||
|
this.statusCode,
|
||||||
|
this.isRetryable = true,
|
||||||
|
this.requiresAuth = false,
|
||||||
|
});
|
||||||
|
|
||||||
|
const R2DownloadResult.success({
|
||||||
|
required String cloudRideId,
|
||||||
|
required Uint8List zipBytes,
|
||||||
|
required String sha256Hash,
|
||||||
|
String? etag,
|
||||||
|
}) : this(
|
||||||
|
success: true,
|
||||||
|
cloudRideId: cloudRideId,
|
||||||
|
zipBytes: zipBytes,
|
||||||
|
sha256Hash: sha256Hash,
|
||||||
|
etag: etag,
|
||||||
|
statusCode: 200,
|
||||||
|
isRetryable: false,
|
||||||
|
);
|
||||||
|
|
||||||
|
const R2DownloadResult.authRequired(String message, {int statusCode = 401})
|
||||||
|
: this(
|
||||||
|
success: false,
|
||||||
|
errorMessage: message,
|
||||||
|
statusCode: statusCode,
|
||||||
|
isRetryable: true,
|
||||||
|
requiresAuth: true,
|
||||||
|
);
|
||||||
|
|
||||||
|
const R2DownloadResult.temporaryFailure(String message, {int? statusCode})
|
||||||
|
: this(success: false, errorMessage: message, statusCode: statusCode, isRetryable: true);
|
||||||
|
|
||||||
|
const R2DownloadResult.permanentFailure(String message, {int? statusCode})
|
||||||
|
: this(success: false, errorMessage: message, statusCode: statusCode, isRetryable: false);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Downloads a verified telemetry bundle through the authenticated Worker gateway.
|
||||||
|
/// The Worker authorizes the ride, and this service verifies the returned SHA-256
|
||||||
|
/// before exposing the bytes to restore code.
|
||||||
|
class R2DownloadService {
|
||||||
|
static const String defaultWorkerUrl = String.fromEnvironment(
|
||||||
|
'R2_WORKER_URL',
|
||||||
|
defaultValue: 'https://motorsignal-r2-worker.bponders12.workers.dev',
|
||||||
|
);
|
||||||
|
|
||||||
|
final http.Client _client;
|
||||||
|
final String _workerBaseUrl;
|
||||||
|
|
||||||
|
R2DownloadService({http.Client? client, String? workerBaseUrl})
|
||||||
|
: _client = client ?? http.Client(),
|
||||||
|
_workerBaseUrl = (workerBaseUrl ?? defaultWorkerUrl).replaceAll(RegExp(r'/+$'), '');
|
||||||
|
|
||||||
|
String get workerBaseUrl => _workerBaseUrl;
|
||||||
|
|
||||||
|
Future<R2DownloadResult> downloadTelemetryBundle({
|
||||||
|
required String idToken,
|
||||||
|
required String cloudRideId,
|
||||||
|
Duration timeout = const Duration(seconds: 45),
|
||||||
|
}) async {
|
||||||
|
if (idToken.trim().isEmpty) {
|
||||||
|
return const R2DownloadResult.authRequired('Firebase ID token is required for R2 download.');
|
||||||
|
}
|
||||||
|
if (!RegExp(r'^ride_[A-Za-z0-9_-]+$').hasMatch(cloudRideId)) {
|
||||||
|
return const R2DownloadResult.permanentFailure('Invalid cloud ride identifier.', statusCode: 400);
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
final uri = Uri.parse('$_workerBaseUrl/r2/download').replace(
|
||||||
|
queryParameters: {'rideId': cloudRideId},
|
||||||
|
);
|
||||||
|
final response = await _client.get(
|
||||||
|
uri,
|
||||||
|
headers: {
|
||||||
|
'Authorization': 'Bearer ${idToken.trim()}',
|
||||||
|
'Accept': 'application/zip',
|
||||||
|
},
|
||||||
|
).timeout(timeout);
|
||||||
|
|
||||||
|
final status = response.statusCode;
|
||||||
|
if (status == 401 || status == 403) {
|
||||||
|
return R2DownloadResult.authRequired(
|
||||||
|
'R2 Worker authorization denied ($status).',
|
||||||
|
statusCode: status,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
if (status == 404) {
|
||||||
|
return R2DownloadResult.permanentFailure(
|
||||||
|
'Telemetry bundle is not available in MotorCloud.',
|
||||||
|
statusCode: status,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
if (status == 409 || status == 422) {
|
||||||
|
return R2DownloadResult.permanentFailure(
|
||||||
|
'MotorCloud rejected the telemetry bundle integrity or ownership metadata.',
|
||||||
|
statusCode: status,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
if (status < 200 || status >= 300) {
|
||||||
|
return R2DownloadResult.temporaryFailure(
|
||||||
|
'R2 Worker temporary error ($status).',
|
||||||
|
statusCode: status,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
final returnedRideId = response.headers['x-motorsignal-ride-id'];
|
||||||
|
if (returnedRideId != cloudRideId) {
|
||||||
|
return const R2DownloadResult.permanentFailure(
|
||||||
|
'R2 Worker returned an unexpected cloud ride identifier.',
|
||||||
|
statusCode: 422,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
final expectedSha256 = response.headers['x-motorsignal-sha256']?.trim().toLowerCase();
|
||||||
|
if (expectedSha256 == null || !RegExp(r'^[a-f0-9]{64}$').hasMatch(expectedSha256)) {
|
||||||
|
return const R2DownloadResult.permanentFailure(
|
||||||
|
'R2 Worker did not provide a valid telemetry SHA-256.',
|
||||||
|
statusCode: 422,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
final bytes = Uint8List.fromList(response.bodyBytes);
|
||||||
|
if (bytes.isEmpty) {
|
||||||
|
return const R2DownloadResult.permanentFailure('Downloaded telemetry bundle is empty.', statusCode: 422);
|
||||||
|
}
|
||||||
|
|
||||||
|
final actualSha256 = sha256.convert(bytes).toString().toLowerCase();
|
||||||
|
if (actualSha256 != expectedSha256) {
|
||||||
|
return const R2DownloadResult.permanentFailure(
|
||||||
|
'Telemetry bundle SHA-256 verification failed.',
|
||||||
|
statusCode: 422,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
return R2DownloadResult.success(
|
||||||
|
cloudRideId: cloudRideId,
|
||||||
|
zipBytes: bytes,
|
||||||
|
sha256Hash: actualSha256,
|
||||||
|
etag: response.headers['etag'],
|
||||||
|
);
|
||||||
|
} on TimeoutException {
|
||||||
|
return R2DownloadResult.temporaryFailure(
|
||||||
|
'Download from Cloudflare R2 timed out after ${timeout.inSeconds} seconds.',
|
||||||
|
);
|
||||||
|
} on FormatException catch (e) {
|
||||||
|
return R2DownloadResult.permanentFailure('Invalid R2 download response: $e', statusCode: 422);
|
||||||
|
} catch (e) {
|
||||||
|
return R2DownloadResult.temporaryFailure('Network error downloading from Cloudflare R2: $e');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Useful when the Worker returns a JSON error body for diagnostics.
|
||||||
|
static String errorBody(http.Response response) {
|
||||||
|
try {
|
||||||
|
final decoded = jsonDecode(response.body);
|
||||||
|
if (decoded is Map && decoded['error'] is String) return decoded['error'] as String;
|
||||||
|
} catch (_) {}
|
||||||
|
return 'R2 Worker request failed (${response.statusCode}).';
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Reference in New Issue