Track online szinkronizáció refraktorálása

2026-07-08 23:29:22 +02:00
parent 1ab656585b
commit d23f736ead
7 changed files with 131 additions and 154 deletions
@@ -57,6 +57,14 @@ class AppDatabase {
  Future<void> upgradeSchema(Database db, int oldVersion, int newVersion) =>
      _onUpgrade(db, oldVersion, newVersion);

+  Future<void> _tryExec(Database db, String sql) async {
+    try {
+      await db.execute(sql);
+    } on DatabaseException catch (e) {
+      if (!e.toString().contains('duplicate column')) rethrow;
+    }
+  }
+
  Future<void> _onCreate(Database db, int _) async {
    // ── Projects ────────────────────────────────────────────────────
    // uuid: a projekt globális azonosítója. Közös projekthez csatlakozáskor
@@ -261,6 +269,7 @@ class AppDatabase {

    await _createStakeoutTable(db);
    await _createContactsOutbox(db);
+    await _addAppInstanceIdColumns(db);

    // Alap projekt létrehozása az első indításhoz
    final now = DateTime.now().toIso8601String();
@@ -296,6 +305,8 @@ class AppDatabase {
    if (oldVersion < 5) {
      _createContactsOutbox(db);
    }
+
+    await _addAppInstanceIdColumns(db);
  }

  Future<void> _migrateToV4(Database db) async {
@@ -499,6 +510,7 @@ class AppDatabase {
            DeviceIdentityService.to.isReady
        ? DeviceIdentityService.to.appInstanceId
        : null;
+    map['app_instance_id'] ??= DeviceIdentityService.to.appInstanceId;
    return db.insert('tracks', map);
  }

@@ -524,10 +536,14 @@ class AppDatabase {
    );
  }

-  Future<List<Track>> listTracks() async {
+  Future<List<Track>> listTracks({int? projectId}) async {
    final db = await database;
    final rows = await db.query('tracks',
-        where: 'deleted_at IS NULL', orderBy: 'start_time DESC');
+        where: projectId != null
+            ? 'deleted_at IS NULL AND project_id = ?'
+            : 'deleted_at IS NULL',
+        whereArgs: projectId != null ? [projectId] : null,
+        orderBy: 'start_time DESC');
    return rows.map(Track.fromMap).toList();
  }

@@ -554,9 +570,11 @@ class AppDatabase {
        await txn.rawUpdate('''
        UPDATE tracks
        SET distance_m = ?,
-            point_count     = point_count + 1
+            point_count = point_count + 1,
+            updated_at = ?,
+            sync_status = 'pending'
        WHERE id = ?
-      ''', [newDistance, point.trackId]);
+      ''', [newDistance, DateTime.now().toIso8601String(), point.trackId]);
      });
    } catch (e) {
      print(
@@ -879,6 +897,8 @@ class AppDatabase {
            DeviceIdentityService.to.isReady
        ? DeviceIdentityService.to.appInstanceId
        : null;
+    map['app_instance_id'] ??= DeviceIdentityService.to.appInstanceId;
+
    return db.insert('measured_points', map);
  }

@@ -974,7 +994,7 @@ class AppDatabase {
            DeviceIdentityService.to.isReady
        ? DeviceIdentityService.to.appInstanceId
        : null;
-    ;
+    map['app_instance_id'] ??= DeviceIdentityService.to.appInstanceId;
    return db.insert('stakeout_points', map);
  }

@@ -998,6 +1018,8 @@ class AppDatabase {
                DeviceIdentityService.to.isReady
            ? DeviceIdentityService.to.appInstanceId
            : null;
+        map['app_instance_id'] ??= DeviceIdentityService.to.appInstanceId;
+
        await txn.insert('stakeout_points', map);
        inserted++;
      }
@@ -1353,6 +1375,16 @@ class AppDatabase {
        0;
  }

+  /// A device_id (ANDROID_ID, TEXT) MARAD — helyes, mert az oszlop nem
+  /// uuid típusú. Az app_instance_id egy KIEGÉSZÍTŐ azonosító: az adott
+  /// app-telepítéshez tartozó, saját generált uuid — mindkettő tárolva,
+  /// hogy az "eszköz" és a "konkrét app-példány" is visszakereshető legyen.
+  Future<void> _addAppInstanceIdColumns(Database db) async {
+    for (final table in ['tracks', 'measured_points', 'stakeout_points']) {
+      await _tryExec(db, 'ALTER TABLE $table ADD COLUMN app_instance_id TEXT');
+    }
+  }
+
  // Future<void> testOnly() async {
  //   final db = await database;
  //   await db.execute(
@@ -94,6 +94,7 @@ class StakeoutSyncService extends GetxService {
              'offset_dist': r['offset_dist'],
              'offset_bearing': r['offset_bearing'],
              'device_id': r['device_id'],
+              'app_instance_id': r['app_instance_id'],
              'deleted_at': _toUtc(r['deleted_at'] as String?),
            })
        .toList();
@@ -164,6 +165,7 @@ class StakeoutSyncService extends GetxService {
          'offset_bearing': r['offset_bearing'],
          'created_by': r['created_by'],
          'device_id': r['device_id'],
+          'app_instance_id': r['app_instance_id'],
          'created_at': r['created_at'] ?? DateTime.now().toIso8601String(),
          'updated_at': r['updated_at'],
          'deleted_at': r['deleted_at'],
@@ -11,7 +11,8 @@ class TrackDatabase {
  Future<int> insertTrack(Track t) => AppDatabase.instance.insertTrack(t);
  Future<void> updateTrack(Track t) => AppDatabase.instance.updateTrack(t);
  Future<void> deleteTrack(int id) => AppDatabase.instance.deleteTrack(id);
-  Future<List<Track>> listTracks() => AppDatabase.instance.listTracks();
+  Future<List<Track>> listTracks({int? projectId}) =>
+      AppDatabase.instance.listTracks(projectId: projectId);
  Future<Track?> getTrack(int id) => AppDatabase.instance.getTrack(id);
  Future<void> addPoint(TrackPoint p, double d) =>
      AppDatabase.instance.addPoint(p, d);
@@ -1,6 +1,11 @@
-// Kétszintű Supabase szinkronizáció:
-//   1. Élő pozíció — minden 3 mp-ben UPSERT → device_positions
-//   2. Track pontok — batch INSERT (10 pont vagy 8 mp) → terepi_track_points
+// EGYETLEN track-szinkron útvonal:
+//   1. Élő pozíció — minden 3 mp-ben UPSERT → terepi_seged_device_positions
+//      (ez a "hol van most ez az eszköz" jelenlét, független a track-adattól)
+//   2. Track fejléc + pontok — a TsSyncService KÖZÖS, uuid-alapú, idempotens
+//      push-függvényeit hívja, csak SŰRŰBBEN (10 pont vagy 8 mp), amíg fut
+//      a rögzítés. Nincs külön insert/upsert-kód és külön szerveroldali
+//      azonosító — pontosan ugyanaz fut, mint amit a háttér-szinkron
+//      (TsSyncService) 3 percenként amúgy is lefuttatna.

 import 'dart:async';

@@ -9,7 +14,7 @@ import 'package:get/get.dart';
 import 'package:latlong2/latlong.dart';
 import 'package:supabase_flutter/supabase_flutter.dart';

-import 'app_database.dart';
+import 'ts_sync_service.dart';
 import 'device_identity_service.dart';
 import '../models/track.dart';

@@ -24,8 +29,8 @@ class TrackSyncService extends GetxService {
  static const _positionIntervalSec = 3;

  // ── Belső állapot ─────────────────────────────────────────────────
-  final _buffer = <TrackPoint>[];
-  Timer? _batchTimer;
+  int _pointsSinceLastPush = 0;
+  Timer? _pushTimer;
  Timer? _posTimer;

  Track? _track;
@@ -34,7 +39,6 @@ class TrackSyncService extends GetxService {

  // ── Publikus állapot ──────────────────────────────────────────────
  final isSyncing = false.obs;
-  final pendingCount = 0.obs;

  // ── Lifecycle ─────────────────────────────────────────────────────

@@ -55,14 +59,17 @@ class TrackSyncService extends GetxService {

  void startSession(Track track) {
    _track = track;
-    _buffer.clear();
-    pendingCount.value = 0;
+    _pointsSinceLastPush = 0;

    if (track.isLocalOnly) return;

-    _batchTimer = Timer.periodic(
+    // Azonnali első push — ne kelljen az első 8 mp-et kivárni ahhoz, hogy
+    // a track fejléce megjelenjen a szerveren más eszközök számára.
+    unawaited(_pushTrackData());
+
+    _pushTimer = Timer.periodic(
      const Duration(seconds: _batchIntervalSec),
-      (_) => _flush(),
+      (_) => _pushTrackData(),
    );
    _posTimer = Timer.periodic(
      const Duration(seconds: _positionIntervalSec),
@@ -72,20 +79,12 @@ class TrackSyncService extends GetxService {

  Future<void> stopSession(Track track) async {
    _stopTimers();
-    if (track.isLocalOnly) return;
-
-    await _flush(); // utolsó batch
-
-    // Track fejléc lezárása Supabase-ben
-    if (track.supabaseId != null) {
-      await _supabase.from('terepi_seged_tracks').update({
-        'end_time': track.endTime?.toIso8601String(),
-        'status': 'finished',
-        'distance_m': track.distanceMeters,
-        'point_count': track.pointCount,
-      }).eq('id', track.supabaseId!);
+    if (!track.isLocalOnly) {
+      // Záró push: a lezárt állapotot (vég idő, státusz, végleges táv/
+      // pontszám) a TrackingController.stopRecording() már beírta
+      // helyileg 'pending' státusszal — ez küldi fel + az utolsó pontokat.
+      await _pushTrackData();
    }
-
    await _setInactive();
    _track = null;
    _lastPos = null;
@@ -93,79 +92,42 @@ class TrackSyncService extends GetxService {

  // ── Pont pufferelés ───────────────────────────────────────────────

-  /// TrackingController._onPosition() hívja minden pontnál
+  /// TrackingController._onPosition() hívja minden pontnál. A pont már
+  /// ekkor a LOKÁLIS adatbázisban van (a controller írja be) — itt csak
+  /// azt számoljuk, mikor induljon köztes push a 8 mp-es időzítő előtt.
  void onNewPoint(TrackPoint point) {
    _lastPos = LatLng(point.latitude, point.longitude);

    if (_track == null || _track!.isLocalOnly) return;

-    _buffer.add(point);
-    pendingCount.value = _buffer.length;
-
-    if (_buffer.length >= _batchSize) _flush();
+    _pointsSinceLastPush++;
+    if (_pointsSinceLastPush >= _batchSize) _pushTrackData();
  }

-  // ── Supabase track létrehozása ────────────────────────────────────
-
-  /// startRecording()-ban hívandó, visszaadja a Supabase UUID-t
-  Future<String?> createRemoteTrack(Track track) async {
-    if (!_online) return null;
-    try {
-      final res = await _supabase
-          .from('terepi_seged_tracks')
-          .insert({
-            'device_id': DeviceIdentityService.to.deviceId,
-            'name': track.name,
-            'source': track.source,
-            'start_time': track.startTime.toIso8601String(),
-            'status': 'recording',
-          })
-          .select('id')
-          .single();
-      return res['id'] as String?;
-    } catch (e) {
-      return null;
-    }
-  }
-
-  // ── Batch feltöltés ───────────────────────────────────────────────
-
-  Future<void> _flush() async {
-    if (_buffer.isEmpty || !_online) return;
-
-    final supabaseId = _track?.supabaseId;
-    if (supabaseId == null) return;
-
-    final batch = List<TrackPoint>.from(_buffer);
-    _buffer.clear();
-    pendingCount.value = 0;
+  // ── Push a KÖZÖS, uuid-alapú TsSyncService-függvényekkel ──────────
+  // Nincs saját insert/upsert-kód, nincs külön szerveroldali azonosító —
+  // pontosan az fut, amit a háttér-szinkron is futtatna, csak sűrűbben.

+  Future<void> _pushTrackData() async {
+    if (!_online) return;
+    _pointsSinceLastPush = 0;
    try {
      isSyncing.value = true;
-      await _supabase.from('terepi_seged_track_points').insert(
-            batch
-                .map((p) => {
-                      'track_id': supabaseId,
-                      'latitude': p.latitude,
-                      'longitude': p.longitude,
-                      'altitude': p.altitude,
-                      'accuracy': p.accuracy,
-                      'speed': p.speed,
-                      'heading': p.heading,
-                      'timestamp': p.timestamp.toIso8601String(),
-                    })
-                .toList(),
-          );
+      // Fejléc ELŐBB (FK: track_points.track_id → tracks.id), utána a
+      // pontok — ugyanaz a sorrend, mint a háttér-szinkron ciklusában.
+      await TsSyncService.to.pushTracks();
+      await TsSyncService.to.pushTrackPoints();
    } catch (_) {
-      // Hiba → visszateszi a bufferbe
-      _buffer.insertAll(0, batch);
-      pendingCount.value = _buffer.length;
+      // Hiba esetén a sorok 'pending' állapotban maradnak — a
+      // következő ciklus (vagy a háttér-szinkron) újrapróbálja.
    } finally {
      isSyncing.value = false;
    }
  }

  // ── Élő pozíció broadcast ─────────────────────────────────────────
+  // Ez a "hol van most ez az eszköz" jelenlét — nem track-adat, ezért
+  // marad önálló, egyszerű upsert.

  Future<void> _broadcastPosition() async {
    final pos = _lastPos;
@@ -175,10 +137,11 @@ class TrackSyncService extends GetxService {
    try {
      await _supabase.from('terepi_seged_device_positions').upsert({
        'device_id': device.deviceId,
+        'app_instance_id': device.appInstanceId,
        'user_name': device.deviceLabel.value,
        'latitude': pos.latitude,
        'longitude': pos.longitude,
-        'track_id': _track?.supabaseId,
+        'track_id': _track?.uuid,
        'is_active': true,
        'updated_at': DateTime.now().toUtc().toIso8601String(),
      }, onConflict: 'device_id');
@@ -193,53 +156,13 @@ class TrackSyncService extends GetxService {
    } catch (_) {}
  }

-  // ── Offline → Online szinkron ─────────────────────────────────────
-
-  Future<void> syncTrack(Track track) async {
-    if (!_online) return;
-
-    String? supabaseId = track.supabaseId;
-    if (supabaseId == null) {
-      supabaseId = await createRemoteTrack(track);
-      if (supabaseId == null) return;
-      final updated = track.copyWith(supabaseId: supabaseId);
-      await AppDatabase.instance.updateTrack(updated);
-    }
-
-    final points = await AppDatabase.instance.getPoints(track.id!);
-    if (points.isEmpty) return;
-
-    const chunk = 100;
-    for (int i = 0; i < points.length; i += chunk) {
-      final slice = points.sublist(i, (i + chunk).clamp(0, points.length));
-      await _supabase.from('terepi_seged_track_points').insert(
-            slice
-                .map((p) => {
-                      'track_id': supabaseId,
-                      'latitude': p.latitude,
-                      'longitude': p.longitude,
-                      'altitude': p.altitude,
-                      'accuracy': p.accuracy,
-                      'speed': p.speed,
-                      'heading': p.heading,
-                      'timestamp': p.timestamp.toIso8601String(),
-                    })
-                .toList(),
-          );
-    }
-
-    await AppDatabase.instance.updateTrack(
-      track.copyWith(supabaseId: supabaseId),
-    );
-  }
-
  // ── Kapcsolat figyelés ────────────────────────────────────────────

  void _listenConnectivity() {
    Connectivity().onConnectivityChanged.listen((results) async {
      final wasOffline = !_online;
      _online = results.any((r) => r != ConnectivityResult.none);
-      if (wasOffline && _online) await _flush();
+      if (wasOffline && _online) await _pushTrackData();
    });
  }

@@ -249,9 +172,9 @@ class TrackSyncService extends GetxService {
  }

  void _stopTimers() {
-    _batchTimer?.cancel();
+    _pushTimer?.cancel();
    _posTimer?.cancel();
-    _batchTimer = null;
+    _pushTimer = null;
    _posTimer = null;
  }
 }
@@ -141,8 +141,8 @@ class TsSyncService extends GetxService {
  Future<void> _push() async {
    await _pushProjects();
    await _pushMeasuredPoints();
-    await _pushTracks();
-    await _pushTrackPoints();
+    await pushTracks();
+    await pushTrackPoints();
    await _pushNoteItems();
  }

@@ -188,6 +188,7 @@ class TsSyncService extends GetxService {
              'measured_at': _toUtc(r['timestamp'] as String?),
              'note': r['note'],
              'device_id': r['device_id'],
+              'app_instance_id': r['app_instance_id'],
              'deleted_at': _toUtc(r['deleted_at'] as String?),
            })
        .toList();
@@ -197,7 +198,7 @@ class TsSyncService extends GetxService {
        'measured_points', rows.map((r) => r['uuid'] as String).toList());
  }

-  Future<void> _pushTracks() async {
+  Future<void> pushTracks() async {
    final rows = await _db.pendingTracks();
    if (rows.isEmpty) return;

@@ -213,6 +214,7 @@ class TsSyncService extends GetxService {
              'distance_m': r['distance_m'],
              'point_count': r['point_count'],
              'device_id': r['device_id'],
+              'app_instance_id': r['app_instance_id'],
              'deleted_at': _toUtc(r['deleted_at'] as String?),
            })
        .toList();
@@ -225,7 +227,7 @@ class TsSyncService extends GetxService {
  /// Track-pontok: nagy mennyiség lehet, ezért 500-as batch-ekben, ciklusban,
  /// amíg el nem fogy. A szerveren nincs UPDATE policy (append-only), ezért
  /// ignoreDuplicates — az ismételt feltöltés csendben kimarad.
-  Future<void> _pushTrackPoints() async {
+  Future<void> pushTrackPoints() async {
    while (true) {
      final rows = await _db.pendingTrackPoints(limit: 500);
      if (rows.isEmpty) break;
@@ -328,6 +330,7 @@ class TsSyncService extends GetxService {
        'note': r['note'] ?? '',
        'created_by': r['created_by'],
        'device_id': r['device_id'],
+        'app_instance_id': r['app_instance_id'],
        'updated_at': r['updated_at'],
        'deleted_at': r['deleted_at'],
        'sync_status': 'synced',
@@ -358,6 +361,7 @@ class TsSyncService extends GetxService {
        'is_local_only': 0,
        'created_by': r['created_by'],
        'device_id': r['device_id'],
+        'app_instance_id': r['app_instance_id'],
        'updated_at': r['updated_at'],
        'deleted_at': r['deleted_at'],
        'sync_status': 'synced',