processSyncPayload method

Future<({int newMessages, String peerNodeId})> processSyncPayload(Uint8List compressedBytes)

Decompresses, parses, and merges an incoming sync payload into the local database.

Heavy JSON parsing is offloaded to a background isolate if the decompressed payload exceeds kIsolateParsingThresholdBytes.

Returns a record with the number of new messages ingested and the peer's full node ID.

Implementation

Future<({int newMessages, String peerNodeId})> processSyncPayload(
  Uint8List compressedBytes,
) async {
  // 1. Decompress
  final decompressed = gzip.decode(compressedBytes);
  final jsonString = utf8.decode(decompressed);

  // 2. Parse — use isolate for large payloads
  SyncPayload syncPayload;
  if (decompressed.length > kIsolateParsingThresholdBytes) {
    syncPayload = await Isolate.run(() {
      final json = jsonDecode(jsonString) as Map<String, dynamic>;
      return SyncPayload.fromJson(json);
    });
  } else {
    final json = jsonDecode(jsonString) as Map<String, dynamic>;
    syncPayload = SyncPayload.fromJson(json);
  }

  // 3. Merge nodes into routing table
  await _mergeNodes(syncPayload.nodes, syncPayload.senderNodeId);

  // 4. Active Group Discovery — extract new groups from the routing table
  await _discoverGroupsFromNodes(syncPayload.nodes);

  // 5. Merge messages into store-and-forward table
  final newMessageCount = await _mergeMessages(syncPayload.messages);

  // 6. Update group member counts from the full routing table
  await _updateGroupMemberCounts();

  // 7. Cleanup
  await _db.pruneExpiredMessages(ttlHours: kAirpassMessageTtlHours);
  await _db.pruneStaleNodes(kStaleNodeThreshold);
  await _db.pruneOrphanedDeliveries();

  return (newMessages: newMessageCount, peerNodeId: syncPayload.senderNodeId);
}