MotorSignal-App/lib/services/sync/cloud_restore_service.dart

202 lines
6.7 KiB
Dart

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 telemetryRestored;
final int telemetrySkipped;
final int telemetryFailed;
const CloudRestoreResult({
required this.metadata,
this.telemetryRestored = 0,
this.telemetrySkipped = 0,
this.telemetryFailed = 0,
});
int get telemetryPending => telemetryFailed;
}
/// Additive, resumable MotorCloud restore coordinator.
///
/// 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,
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;
if (user == null || user.isAnonymous) {
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,
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;
const expected = <String>[
'timestamp',
'latitude',
'longitude',
'altitude',
'speed',
'heading',
'gpsAccuracy',
'leanAngle',
'pitchAngle',
'accelX',
'accelY',
'accelZ',
'longitudinalG',
'lateralG',
];
final header = lines.first.split(',');
if (header.length != expected.length || !_sameStringList(header, expected)) {
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 _sameStringList(List<String> actual, List<String> expected) {
if (actual.length != expected.length) return false;
for (var i = 0; i < expected.length; i++) {
if (actual[i] != expected[i]) return false;
}
return true;
}
bool _isValidCloudRideId(String id) => RegExp(r'^ride_[A-Za-z0-9_-]+$').hasMatch(id);
}