Week 3 Day 17: Implement peer state sync with sync_request/response and late-join consistency

Author: Anbu-00001

PROJECT_TIMELINE.txt

@@ -939,3 +939,82 @@ Verification:
   Flutter analyze → PASS (only pre-existing info-level naming warnings)
 
 ------------------------------------------------------------
+
+DAY 17 — Initial State Synchronization Layer
+
+Summary:
+Implemented the initial state synchronization layer so that when a new device joins the mesh network, it automatically fetches existing incidents from connected peers. This ensures late-joining devices have a complete view of the current incident landscape without relying on a central server.
+
+Architecture Components Added:
+
+• Peer Connection Hook (Go daemon)
+  - When a new peer is discovered and connected via mDNS, the daemon
+    automatically broadcasts a sync_request message to the network.
+  - PubSubManager is now initialized before mDNS discovery to enable this.
+  - sync_request is triggered once per peer connection event.
+
+• Sync Request Handling (Flutter)
+  - P2PService listens for sync_request messages via dedicated stream.
+  - IncidentRepository responds by fetching all local incidents from the
+    Drift database and sending them as batched sync_response messages.
+
+• Sync Response Format
+  - Incidents are sent in batches of 50 per message to avoid network flooding.
+  - Each sync_response envelope contains:
+    msg_id, msg_type: "sync_response", origin_peer, payload: { incidents: [...] }
+
+• Sync Response Handling (Flutter)
+  - P2PService routes sync_response messages to a dedicated stream.
+  - IncidentRepository parses incident payloads and inserts each one
+    through the existing _handleIncomingP2PIncident deduplication logic.
+  - DB-level dedup prevents duplicate insertions.
+
+• Peer Sync State Map
+  - P2PService maintains a _peerSyncState map (peer_id → sync_completed).
+  - Prevents responding to duplicate sync_request messages from the same peer.
+  - Prevents infinite sync loops across the network.
+
+• Deduplication Compatibility
+  - All sync messages pass through MessageCache (Flutter) and dedupCache (Go).
+  - sync_response msg_ids are added to the dedup cache before sending.
+
+Message Flow (Late-Join Sync):
+  Device B connects to mesh via mDNS
+    ↓
+  Go daemon broadcasts sync_request (SYNC_REQUEST_SENT)
+    ↓
+  Device A Flutter receives sync_request
+    ↓
+  IncidentRepository fetches all local incidents
+    ↓
+  P2PService sends batched sync_response (SYNC_RESPONSE_SENT)
+    ↓
+  Device B Flutter receives sync_response (SYNC_RESPONSE_RECEIVED)
+    ↓
+  IncidentRepository merges incidents via dedup logic (INCIDENT_MERGED)
+    ↓
+  Incidents appear on Device B map
+
+Logging Added:
+  Go daemon:
+    [Discovery] SYNC_REQUEST_SENT
+  Flutter:
+    [P2P] SYNC_REQUEST_RECEIVED
+    [P2P] SYNC_RESPONSE_RECEIVED
+    [P2P] SYNC_RESPONSE_SENT
+    [IncidentRepo] SYNC_RESPONSE_SENT
+    [IncidentRepo] SYNC_RESPONSE_RECEIVED
+    [IncidentRepo] INCIDENT_MERGED
+
+Files Modified:
+- backend/p2p-node/main.go (reordered PubSub init before Discovery)
+- backend/p2p-node/discovery.go (sync_request broadcast on peer connect)
+- mobile_app/lib/services/p2p_service.dart (sync streams, peer state map, batch response)
+- mobile_app/lib/data/repositories/incident_repository.dart (sync handlers, getAllIncidents)
+- PROJECT_TIMELINE.txt (this update)
+
+Verification:
+  Go build → PASS
+  Flutter analyze → PASS
+
+------------------------------------------------------------

backend/p2p-node/discovery.go

@@ -15,7 +15,8 @@ const discoveryServiceTag = "openrescue.p2p"
 
 // discoveryNotifee gets notified when we find a new peer via mDNS discovery
 type discoveryNotifee struct {
-	h host.Host
+	h   host.Host
+	psm *PubSubManager
 }
 
 // HandlePeerFound connects to peers discovered via mDNS
@@ -36,12 +37,23 @@ func (n *discoveryNotifee) HandlePeerFound(pi peer.AddrInfo) {
 		log.Printf("[Discovery] Failed to connect to peer %s: %s", pi.ID.String(), err)
 	} else {
 		log.Printf("[Discovery] Peer connected: %s (addrs: %v)", pi.ID.String(), pi.Addrs)
+		
+		// Broadcast sync_request to the network
+		syncReq := NetworkEnvelope{
+			MsgType: "sync_request",
+			Payload: map[string]interface{}{},
+		}
+		
+		log.Printf("[Discovery] SYNC_REQUEST_SENT: Requesting state sync after connecting to peer %s", pi.ID.String())
+		if err := n.psm.Broadcast(syncReq); err != nil {
+			log.Printf("[Discovery] Failed to broadcast sync_request: %v", err)
+		}
 	}
 }
 
 // setupDiscovery creates an mDNS discovery service
-func setupDiscovery(h host.Host) error {
-	s := mdns.NewMdnsService(h, discoveryServiceTag, &discoveryNotifee{h: h})
+func setupDiscovery(h host.Host, psm *PubSubManager) error {
+	s := mdns.NewMdnsService(h, discoveryServiceTag, &discoveryNotifee{h: h, psm: psm})
 	if s == nil {
 		return fmt.Errorf("failed creating mDNS service")
 	}

backend/p2p-node/main.go

@@ -31,20 +31,20 @@ func main() {
 		fmt.Printf("Listening on: %s/p2p/%s\n", addr.String(), host.ID().String())
 	}
 
-	// 2. Setup mDNS discovery
-	if err := setupDiscovery(host); err != nil {
-		log.Fatalf("Failed to setup mDNS discovery: %v", err)
-	}
-
 	// Channel for messages received from GossipSub to be sent to WebSocket clients
 	msgChan := make(chan NetworkEnvelope, 100)
 
-	// 3. Setup PubSub (GossipSub)
+	// 2. Setup PubSub (GossipSub) first so we can pass it to discovery
 	pubSubManager, err := setupPubSub(ctx, host, msgChan)
 	if err != nil {
 		log.Fatalf("Failed to setup PubSub: %v", err)
 	}
 
+	// 3. Setup mDNS discovery
+	if err := setupDiscovery(host, pubSubManager); err != nil {
+		log.Fatalf("Failed to setup mDNS discovery: %v", err)
+	}
+
 	// 4. Start HTTP/WS API server on port 7000
 	apiServer := NewAPIServer(7000, pubSubManager, msgChan)
 	if err := apiServer.Start(); err != nil {

mobile_app/lib/data/repositories/incident_repository.dart

@@ -13,9 +13,20 @@ class IncidentRepository {
   final P2PService _p2pService;
 
   IncidentRepository(this._db, this._apiClient, this._p2pService) {
+    // Day-16: Listen for incoming incident_create messages
     _p2pService.incomingIncidents.listen((dto) {
       _handleIncomingP2PIncident(dto);
     });
+
+    // Day-17: Listen for sync_request messages — respond with local incidents
+    _p2pService.syncRequests.listen((envelope) {
+      _handleSyncRequest(envelope);
+    });
+
+    // Day-17: Listen for sync_response messages — merge received incidents
+    _p2pService.syncResponses.listen((envelope) {
+      _handleSyncResponse(envelope);
+    });
   }
 
   Stream<List<domain.Incident>> watchIncidents() {
@@ -29,6 +40,14 @@ class IncidentRepository {
     return incidentFromDb(row);
   }
 
+  /// Fetches all non-deleted incidents from the local database.
+  Future<List<domain.Incident>> getAllIncidents() async {
+    final rows = await (_db.select(_db.incidents)
+          ..where((t) => t.deleted_flag.equals(false)))
+        .get();
+    return rows.map((r) => incidentFromDb(r)).toList();
+  }
+
   Future<void> createIncident(domain.IncidentCreateDto dto) async {
     final localDocId = 'temp_${DateTime.now().millisecondsSinceEpoch}';
 
@@ -159,4 +178,70 @@ class IncidentRepository {
     debugPrint(
         '[IncidentRepo] P2P incident inserted: $incidentId');
   }
+
+  // ─── Day-17: State Synchronization Handlers ────────────────────────────
+
+  /// Handles a sync_request: fetches all local incidents and sends them back
+  /// as batched sync_response messages via P2PService.
+  Future<void> _handleSyncRequest(dynamic envelope) async {
+    debugPrint('[IncidentRepo] Processing sync_request — fetching local incidents');
+
+    try {
+      final incidents = await getAllIncidents();
+      debugPrint(
+          '[IncidentRepo] Found ${incidents.length} local incidents to sync');
+
+      // Send the incidents back as a sync_response via P2PService
+      await _p2pService.sendSyncResponse(incidents);
+
+      debugPrint(
+          '[IncidentRepo] SYNC_RESPONSE_SENT: ${incidents.length} incidents sent');
+    } catch (e) {
+      debugPrint('[IncidentRepo] Failed to handle sync_request: $e');
+    }
+  }
+
+  /// Handles a sync_response: extracts incidents from the payload and inserts
+  /// them using the existing deduplication logic.
+  Future<void> _handleSyncResponse(dynamic envelope) async {
+    try {
+      final payload = (envelope as dynamic).payload as Map<String, dynamic>;
+      final incidentsList = payload['incidents'] as List<dynamic>? ?? [];
+
+      debugPrint(
+          '[IncidentRepo] SYNC_RESPONSE_RECEIVED: ${incidentsList.length} incidents in batch');
+
+      int mergedCount = 0;
+      for (final incidentData in incidentsList) {
+        final Map<String, dynamic> incMap =
+            incidentData is Map<String, dynamic>
+                ? incidentData
+                : Map<String, dynamic>.from(incidentData as Map);
+
+        final incidentId =
+            incMap['incident_id'] as String? ??
+            'sync_${DateTime.now().millisecondsSinceEpoch}_$mergedCount';
+
+        // Use IncidentCreateDto and pass through existing dedup logic
+        final dto = domain.IncidentCreateDto(
+          type: incMap['type'] as String? ?? 'unknown',
+          lat: (incMap['lat'] as num?)?.toDouble() ?? 0.0,
+          lon: (incMap['lon'] as num?)?.toDouble() ?? 0.0,
+          priority: incMap['priority'] as String? ?? 'medium',
+          status: incMap['status'] as String? ?? 'new',
+          client_id: incMap['reporter_id'] as String? ?? 'synced_peer',
+          sequence_num: 1,
+          data: {'incident_id': incidentId},
+        );
+
+        await _handleIncomingP2PIncident(dto);
+        mergedCount++;
+      }
+
+      debugPrint(
+          '[IncidentRepo] INCIDENT_MERGED: processed $mergedCount incidents from sync_response');
+    } catch (e) {
+      debugPrint('[IncidentRepo] Failed to handle sync_response: $e');
+    }
+  }
 }