feat: wire authenticated R2 telemetry restore into cloud restore
This commit is contained in:
parent
827d15eb84
commit
267afb46a4
|
|
@ -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<CloudRestoreResult> 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<int> _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<int>, allowMalformed: false);
|
||||
final lines = const LineSplitter().convert(csv);
|
||||
if (lines.length < 2) return 0;
|
||||
|
||||
final header = lines.first.split(',');
|
||||
final expected = <String>[
|
||||
'timestamp',
|
||||
'latitude',
|
||||
'longitude',
|
||||
'altitude',
|
||||
'speed',
|
||||
'heading',
|
||||
'gpsAccuracy',
|
||||
'leanAngle',
|
||||
'pitchAngle',
|
||||
'accelX',
|
||||
'accelY',
|
||||
'accelZ',
|
||||
'longitudinalG',
|
||||
'lateralG',
|
||||
];
|
||||
if (header.length != expected.length || !List<String>.generate(header.length, (i) => header[i])
|
||||
.every((value) => expected.contains(value))) {
|
||||
throw const FormatException('Unsupported telemetry.csv schema.');
|
||||
}
|
||||
|
||||
final companions = <TelemetrySamplesCompanion>[];
|
||||
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<double> _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);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue