syncIP method

Future<void> syncIP(String ip)

Syncs events with the given client identified by the given IP.

Implementation

Future<void> syncIP(String ip) async {
  if (ip == await client.getCurrentIp()) {
    _logger.log('Skipping self-sync');
    return;
  }

  if (_syncRequestIndex[ip] case Future<void> request) {
    _logger.log('De-duping already in progress sync for $ip');
    await request;
    return;
  }

  await (_syncRequestIndex[ip] = (() async {
    try {
      var hasMore = true;
      while (hasMore) {
        final mySequenceIndex = await db.events.latestDeviceSequenceIndex
            .get();

        final stopwatch = Stopwatch()..start();
        final response = await client.postJson(
          ip,
          '/api/v1/events/sync',
          headers: {'Content-Type': 'application/json'},
          body: mySequenceIndex,
        );
        stopwatch.stop();

        if (response.statusCode != 200) {
          hasMore = false;
          _logger.log(
            'Sync failed with $ip: ${response.statusCode} ${response.body}',
          );
          continue;
        }

        final json = jsonDecode(response.body);
        if (json is! Map<String, dynamic>) {
          hasMore = false;
          _logger.log('Sync failed with $ip: invalid response payload');
          continue;
        }

        hasMore = json['has_more'] == true;
        final rawEvents = json['events'];
        final events = switch (rawEvents) {
          final List values =>
            values
                .whereType<Map>()
                .map(
                  (value) =>
                      EventModel.fromJson(Map<String, dynamic>.from(value)),
                )
                .toList(),
          _ => <EventModel>[],
        };

        await EventIngestor.instance.ingestAll(events);

        if (events.isNotEmpty) {
          final device = await db.devices.ip(ip).get();
          if (device != null) {
            await db.logs.insert(
              EventLogModel(
                type: EventLogTypes.syncReceived,
                deviceKey: device.signingKey,
                body: SyncReceivedLogBodyModel(
                  deviceKey: device.signingKey,
                  count: events.length,
                  duration: stopwatch.elapsed,
                ),
              ),
            );
          }
        }
      }
    } catch (error) {
      _logger.log('Sync failed with $ip: $error');
    } finally {
      _syncRequestIndex.remove(ip);
    }
  })());
}