From 267afb46a4acee60581595e5bf3fd2ffc3a716e5 Mon Sep 17 00:00:00 2001 From: bponders12-cell Date: Tue, 1 Sep 2026 21:20:03 -0400 Subject: [PATCH] feat: wire authenticated R2 telemetry restore into cloud restore --- lib/services/sync/cloud_restore_service.dart | 185 ++++++++++++++++++- 1 file changed, 175 insertions(+), 10 deletions(-) diff --git a/lib/services/sync/cloud_restore_service.dart b/lib/services/sync/cloud_restore_service.dart index 43e8ed7..47a1350 100644 --- a/lib/services/sync/cloud_restore_service.dart +++ b/lib/services/sync/cloud_restore_service.dart @@ -1,32 +1,54 @@ +import 'dart:convert'; +import 'dart:typed_data'; + +import 'package:archive/archive.dart'; +import 'package:cloud_firestore/cloud_firestore.dart'; +import 'package:drift/drift.dart' as drift; + import 'cloud_reconciliation_service.dart'; import '../../database/database.dart'; import '../auth/auth_service.dart'; +import 'r2_download_service.dart'; class CloudRestoreResult { final CloudReconciliationResult metadata; - final int telemetryPending; + final int telemetryRestored; + final int telemetrySkipped; + final int telemetryFailed; const CloudRestoreResult({ required this.metadata, - this.telemetryPending = 0, + this.telemetryRestored = 0, + this.telemetrySkipped = 0, + this.telemetryFailed = 0, }); + + int get telemetryPending => telemetryFailed; } /// Additive, resumable MotorCloud restore coordinator. /// -/// Metadata is restored first. Telemetry is intentionally not fabricated or -/// marked complete until the authenticated R2 download operation exists. -/// Existing local rides are never deleted or overwritten. +/// Metadata is reconciled first. Missing telemetry is then downloaded through +/// the authenticated Worker, SHA-256 verified, unpacked, validated, and added +/// to the local ride without replacing existing telemetry. class CloudRestoreService { + final AppDatabase _db; final AuthService _authService; final CloudReconciliationService _metadata; + final R2DownloadService _r2DownloadService; + final FirebaseFirestore _firestore; CloudRestoreService({ required AppDatabase db, required AuthService authService, CloudReconciliationService? metadata, - }) : _authService = authService, - _metadata = metadata ?? CloudReconciliationService(db: db, authService: authService); + R2DownloadService? r2DownloadService, + FirebaseFirestore? firestore, + }) : _db = db, + _authService = authService, + _metadata = metadata ?? CloudReconciliationService(db: db, authService: authService), + _r2DownloadService = r2DownloadService ?? R2DownloadService(), + _firestore = firestore ?? FirebaseFirestore.instance; Future restore() async { final user = _authService.currentUser; @@ -34,12 +56,155 @@ class CloudRestoreService { throw StateError('Authentication is required to restore MotorCloud data.'); } + final idToken = await _authService.getIdToken(); + if (idToken == null || idToken.isEmpty) { + throw StateError('A valid Firebase authentication token is required to restore MotorCloud data.'); + } + final metadataResult = await _metadata.reconcileMetadata(); + final snapshot = await _firestore + .collection('users') + .doc(user.uid) + .collection('rides') + .get(); + + var telemetryRestored = 0; + var telemetrySkipped = 0; + var telemetryFailed = 0; + + for (final document in snapshot.docs) { + final data = document.data(); + final cloudRideId = document.id; + if (!_isValidCloudRideId(cloudRideId) || data['ownerId'] != user.uid) { + telemetrySkipped++; + continue; + } + + final localRide = await (_db.select(_db.rides) + ..where((ride) => ride.cloudRideId.equals(cloudRideId))) + .getSingleOrNull(); + if (localRide == null) { + telemetryFailed++; + continue; + } + + final telemetryCount = await (_db.select(_db.telemetrySamples) + ..where((sample) => sample.rideId.equals(localRide.id))) + .get() + .then((rows) => rows.length); + if (telemetryCount > 0) { + telemetrySkipped++; + continue; + } + + final result = await _r2DownloadService.downloadTelemetryBundle( + idToken: idToken, + cloudRideId: cloudRideId, + ); + if (!result.success || result.zipBytes == null) { + telemetryFailed++; + continue; + } + + try { + final restoredCount = await _restoreTelemetryCsv( + localRide.id, + result.zipBytes!, + ); + if (restoredCount > 0) { + telemetryRestored++; + } else { + telemetryFailed++; + } + } catch (_) { + telemetryFailed++; + } + } + return CloudRestoreResult( metadata: metadataResult, - // This remains zero until the authenticated R2 download Worker contract - // is implemented. Metadata restore is still fully useful on its own. - telemetryPending: 0, + telemetryRestored: telemetryRestored, + telemetrySkipped: telemetrySkipped, + telemetryFailed: telemetryFailed, ); } + + Future _restoreTelemetryCsv(int rideId, Uint8List zipBytes) async { + final archive = ZipDecoder().decodeBytes(zipBytes, verify: true); + final file = archive.findFile('telemetry.csv'); + if (file == null) throw const FormatException('Telemetry bundle does not contain telemetry.csv.'); + + final csv = utf8.decode(file.content as List, allowMalformed: false); + final lines = const LineSplitter().convert(csv); + if (lines.length < 2) return 0; + + final header = lines.first.split(','); + final expected = [ + 'timestamp', + 'latitude', + 'longitude', + 'altitude', + 'speed', + 'heading', + 'gpsAccuracy', + 'leanAngle', + 'pitchAngle', + 'accelX', + 'accelY', + 'accelZ', + 'longitudinalG', + 'lateralG', + ]; + if (header.length != expected.length || !List.generate(header.length, (i) => header[i]) + .every((value) => expected.contains(value))) { + throw const FormatException('Unsupported telemetry.csv schema.'); + } + + final companions = []; + for (final line in lines.skip(1)) { + if (line.trim().isEmpty) continue; + final values = line.split(','); + if (values.length != expected.length) throw const FormatException('Malformed telemetry.csv row.'); + + final timestamp = DateTime.tryParse(values[0]); + if (timestamp == null) throw const FormatException('Telemetry row has an invalid timestamp.'); + + companions.add(TelemetrySamplesCompanion.insert( + rideId: rideId, + timestamp: timestamp.toUtc(), + latitude: _nullableDouble(values[1]), + longitude: _nullableDouble(values[2]), + altitude: _nullableDouble(values[3]), + speed: _nullableDouble(values[4]), + heading: _nullableDouble(values[5]), + gpsAccuracy: _nullableDouble(values[6]), + leanAngle: _nullableDouble(values[7]), + pitchAngle: _nullableDouble(values[8]), + accelX: _nullableDouble(values[9]), + accelY: _nullableDouble(values[10]), + accelZ: _nullableDouble(values[11]), + longitudinalG: _nullableDouble(values[12]), + lateralG: _nullableDouble(values[13]), + )); + } + + if (companions.isEmpty) return 0; + + await _db.transaction(() async { + for (final companion in companions) { + await _db.into(_db.telemetrySamples).insert(companion); + } + }); + return companions.length; + } + + drift.Value _nullableDouble(String value) { + final trimmed = value.trim(); + if (trimmed.isEmpty) return const drift.Value.absent(); + final parsed = double.tryParse(trimmed); + if (parsed == null) throw const FormatException('Telemetry row contains an invalid numeric value.'); + return drift.Value(parsed); + } + + bool _isValidCloudRideId(String id) => RegExp(r'^ride_[A-Za-z0-9_-]+$').hasMatch(id); }