stadtbote-app/lib/tools/mqtt_core.dart

import 'dart:async';
import 'dart:convert';
import 'dart:math';
import 'dart:typed_data';

import 'package:archive/archive.dart';
import 'package:device_info_plus/device_info_plus.dart';
import 'package:flutter/foundation.dart';
import 'package:isar/isar.dart';
import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';
import 'package:stadtbote/model/enqueued_message.dart';
import 'package:stadtbote/model/log_entry.dart';
import 'package:stadtbote/model/ranking.dart';
import 'package:stadtbote/model/status.dart';
import 'package:stadtbote/protocoll_messages/acknowledge_job.dart';
import 'package:stadtbote/protocoll_messages/bootstrap_request.dart';
import 'package:stadtbote/protocoll_messages/bootstrap_response_payload.dart';
import 'package:stadtbote/protocoll_messages/device_log_request.dart';
import 'package:stadtbote/protocoll_messages/device_log_response.dart';
import 'package:stadtbote/protocoll_messages/device_object.dart';
import 'package:stadtbote/model/job.dart';
import 'package:stadtbote/model/message.dart';
import 'package:stadtbote/protocoll_messages/payload_container.dart';
import 'package:stadtbote/protocoll_messages/start_session_message.dart';
import 'package:stadtbote/protocoll_messages/sync_jobs_message.dart';
import 'package:stadtbote/protocoll_messages/tail.dart';
import 'package:stadtbote/protocoll_messages/track_message.dart';
import 'package:stadtbote/tools/connection_status.dart';
import 'package:stadtbote/tools/message_service.dart';
import 'package:stadtbote/tools/messages/job_deleted_message.dart';
import 'package:stadtbote/tools/messages/job_received_message.dart';
import 'package:stadtbote/tools/messages/message_message.dart';
import 'package:stadtbote/tools/messages/ranking_updated.dart';
import 'package:stadtbote/tools/messages/session_info_update.dart';
import 'package:stadtbote/tools/messages/status_info_update.dart';
// ignore: depend_on_referenced_packages
import 'package:typed_data/typed_buffers.dart';
import 'package:http/http.dart' as http;
import 'package:uuid/uuid.dart';

import '../main.dart';
import '../model/session_status.dart';
import 'constants.dart';
import 'util.dart';

enum ConnectionStateEnum {
  notConnected,
  connected
}

class MqttCore {
  static late MqttCore instance;

  late ConnectionStateEnum connectionState = ConnectionStateEnum.notConnected;

  late bool _disconnectInProgress = false;

  late String _brokerUri;
  late int _brokerPort;
  late String _fromServerTopic;
  late String _toServerTopic;

  late MqttServerClient _mqttClient;

  late int _reconnectDelay = 0;

  Timer? _reconnectTimer;

  Timer? _trackTimer;

  MqttCore() {
    _disconnectInProgress = false;

    _processBootstrap();
  }

  static setup() {
    instance = MqttCore();
  }

  diconnect() {
    _disconnectInProgress = true;

    _cancelReconnectTimer();

    _cancelTrackTimer();

    _mqttClient.disconnect();

    connectionState = ConnectionStateEnum.notConnected;
  }

  _connectToBroker() async {
    try {
      String clientId = await Util.getAppUuid();
      Util.logInfo("client id: $clientId");
      Util.logInfo("broker uri: $_brokerUri");
      Util.logInfo("broker port: $_brokerPort");

      _mqttClient = MqttServerClient.withPort(_brokerUri, clientId, _brokerPort);
      _mqttClient.logging(on: false);
      _mqttClient.keepAlivePeriod = 20;
      _mqttClient.autoReconnect = false;
      _mqttClient.onDisconnected = _clientDisconnected;
      _mqttClient.connectionMessage = MqttConnectMessage()
          .withClientIdentifier(await Util.getAppUuid())
          .withWillQos(MqttQos.atMostOnce)
          .withWillTopic('willTopic')
          .withWillMessage('willMessage')
          .authenticateAs("app", "apppwd");

      await _mqttClient.connect();

      if (_mqttClient.connectionStatus?.state == MqttConnectionState.connected) {
        Util.logInfo('Verbunden mit dem MQTT-Broker!');

        _reconnectDelay = 0;

        _setupSubscriptions();

        _initSession();
      } else {
        Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!');

        MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));

        _startReconnectTimer();
      }

    } catch (ex) {
      _startReconnectTimer();

      Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex');

      MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
    }
  }

  publishBinary(String topic, List<int> data) {
    var buffer = Uint8Buffer();
    buffer.addAll(data);

    _mqttClient.publishMessage(topic, MqttQos.exactlyOnce, buffer);

    Util.logInfo("binary message send to $topic");
  }

  publishText(String topic, String text) {
    var buffer = Uint8Buffer();
    buffer.addAll(utf8.encode(text));

    _mqttClient.publishMessage(topic, MqttQos.exactlyOnce, buffer);

    Util.logInfo("message\n$text\nsend to $topic");
  }

  publishPayloadMessage(String topic, Map<String, dynamic> payload) {
    var payloadContainer = PayloadContainer();
    payloadContainer.id = const Uuid().v4().toString();
    payloadContainer.recv = topic;
    payloadContainer.store = true;
    payloadContainer.payload = payload;

    var json = jsonEncode(payloadContainer.toJson());

    try {
      publishText(_toServerTopic, json);
    } catch (e) {
      _enqueueMessage(_toServerTopic, jsonEncode(payloadContainer.toJson()));
    }
  }

  void _setupSubscriptions() async {
    Util.logInfo('_setupSubscriptions called');

    _mqttClient.updates?.listen((List<MqttReceivedMessage<MqttMessage>> c) {
      try {
        final MqttPublishMessage message = c[0].payload as MqttPublishMessage;
        String payload = utf8.decode(MqttPublishPayload.bytesToStringAsString(message.payload.message).codeUnits);
        var topic = message.variableHeader?.topicName;

        Util.logInfo('message received\ntopic: $topic\nmessage: $payload');

        Map<String, dynamic> messageData = jsonDecode(payload);

        for (var entry in messageData.keys) {
          if (entry == 'payload') {
            var payload = messageData[entry];
            for (var payloadEntry in payload.keys) {
              var payloadData = payload[payloadEntry];

              switch(payloadEntry) {
                case 'session':
                  _sessionMessageReceived(payloadData);
                  break;

                case 'status':
                  _statusMessageReceived(payloadData);
                  break;

                case 'ranks':
                  _rankingMessageReceived(payloadData);
                  break;

                case 'jobs':
                  _jobMessageReceived(payloadData);
                  break;

                case 'job_del':
                  _jobDeleteReceived(payloadData);
                  break;

                case 'sync_request':
                  _syncRequestReceived(payloadData);
                  break;

                case "mssg":
                  _messageMessageReceived(payloadData);
                  break;

                case "upload_device_log":
                  _messageUploadDeviceLogReceived(payloadData);
                  break;

                case "tail":
                  _messageTailReceived(payloadData);
                  break;

                default:
                  Util.logInfo('Unknown topic: $payloadEntry');
                  break;
              }
            }
          }
        }
      } catch(ex) {
        Util.logException(ex.toString());
      }
    });

    _mqttClient.subscribe(_fromServerTopic, MqttQos.exactlyOnce);
    Util.logInfo('subscribed to topic: $_fromServerTopic');

    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
  }

  void _sessionMessageReceived(payloadData) async {
    connectionState = ConnectionStateEnum.connected;

    await _sendSyncMessage();

    await _sendEnqueuedMessages();

    var sessionStatusMessage = SessionStatus.fromJson(payloadData['status']);

    await MyApp.isarInstance.writeTxn(() async {
      await MyApp.isarInstance.sessionStatus.clear();
      await MyApp.isarInstance.sessionStatus.put(sessionStatusMessage);
    });

    MessageService().publish<SessionStatusInfoUpdate>(Constants.messageTopicSessionStatusInfoUpdate, arguments: SessionStatusInfoUpdate(status: sessionStatusMessage));

    _startTrackTimer();
  }

  void _statusMessageReceived(payloadData) async {
    var statusMessage = Status.fromJson(payloadData);

    await MyApp.isarInstance.writeTxn(() async {
      await MyApp.isarInstance.status.clear();
      await MyApp.isarInstance.status.put(statusMessage);
    });

    MessageService().publish<StatusInfoUpdate>(Constants.messageTopicStatusInfoUpdated, arguments: StatusInfoUpdate(status: statusMessage));
  }

  void _rankingMessageReceived(List<dynamic> payloadData) async {
    Map<String, dynamic> rankingMap = {};
    rankingMap["ranks"] = payloadData;
    var ranking = Ranking.fromJson(rankingMap);

    await MyApp.isarInstance.writeTxn(() async {
      await MyApp.isarInstance.rankings.put(ranking);
    });

    MessageService().publish<RankingUpdated>(Constants.messageTopicRankingUpdated, arguments: RankingUpdated());
  }

  void _jobMessageReceived(payloadData) async {
    for (var foundJobData in payloadData) {
      Job newJob = Job.fromJson(foundJobData);

      await MyApp.isarInstance.writeTxn(() async {
        await MyApp.isarInstance.jobs.put(newJob);
      });

      if (newJob.state == Constants.stateAssigned) {
        if (newJob.global) {
          await Util.playMp3('aaa.mp3');
        } else {
          await Util.playMp3('job_received.mp3');
        }
      }

      try {
        MessageService().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
      } catch (ex) {
        Util.logException("publish messageTopicJobReceived failed!");
      }

      _acknowlegdeJob(newJob);
    }
  }

  void _jobDeleteReceived(payloadData) async {
    for(var jobId in payloadData) {
      await MyApp.isarInstance.writeTxn(() async {
        await MyApp.isarInstance.jobs.delete(jobId);
      });

      MessageService().publish<JobDeletedMessage>(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(jobId: jobId));
    }
  }

  void _syncRequestReceived(payloadData) async {
    Util.logInfo('_syncRequestReceived');

    _sendSyncMessage();
  }

  void _messageMessageReceived(Map<String, dynamic> payloadData) async {
    var newMessage = await Message.mergeFromJson(payloadData);

    await MyApp.isarInstance.writeTxn(() async {
      await MyApp.isarInstance.messages.put(newMessage);
    });

    await Util.playMp3('asbeep.mp3');

    MessageService().publish<MessageReceivedMessage>(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage));
  }

  void _messageUploadDeviceLogReceived(payloadData) async {
    var deviceLogRequest = DeviceLogRequest.fromJson(payloadData);

    var logEintries = await MyApp.isarInstance.logEntrys.filter().timeGreaterThan(deviceLogRequest.startDateTime).findAll();

    var deviceLog = DeviceLogResponse();
    deviceLog.entries = logEintries;

    //MqttCore.instance.publishPayloadMessage("device_log", deviceLog.toJson());

    var json = jsonEncode(deviceLog.toJson());
    var stringBytes = utf8.encode(json);
    var gzipBytes = GZipEncoder().encode(stringBytes);

    var topic = "/log/${_toServerTopic.replaceAll("/app/", "").replaceAll("/to_server", "")}";
    publishBinary(topic, gzipBytes!);
  }

  void _messageTailReceived(payloadData) async {
    var tailMessage = Tail.fromJson(payloadData);

    Util.logInfo('message');
  }

  void _startReconnectTimer() {
    connectionState = ConnectionStateEnum.notConnected;

    if (!_disconnectInProgress) {
      Util.logInfo("_startReconnectTimer called");

      _reconnectDelay = min(_reconnectDelay + 1, 30);

      _reconnectTimer = Timer(Duration(seconds: _reconnectDelay), () {
        _processBootstrap();
      });
    }
  }

  void _cancelReconnectTimer() {
    if (_reconnectTimer != null && _reconnectTimer!.isActive) {
      _reconnectTimer!.cancel();
    }
  }

  void _clientDisconnected() {
    connectionState = ConnectionStateEnum.notConnected;

    _cancelTrackTimer();

    Util.logInfo("_clientDisconnected called");

    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));

    if (!_disconnectInProgress) {
      _startReconnectTimer();
    }
  }

  void _processBootstrap() async {
    try {
      var deviceInfoPlugin = DeviceInfoPlugin();

      AndroidDeviceInfo androidInfo = await deviceInfoPlugin.androidInfo;
      BaseDeviceInfo deviceInfo = await deviceInfoPlugin.deviceInfo;

      var deviceObject = DeviceObject();
      deviceObject.manufacturer = deviceInfo.data['manufacturer'];
      deviceObject.model = deviceInfo.data['model'];
      deviceObject.os = androidInfo.version.release;
      deviceObject.osVer = androidInfo.version.sdkInt.toString();

      var handshakeMessage = BootstrapRequest();
      handshakeMessage.app = "sb";
      handshakeMessage.mqttClientId = await Util.getAppUuid();
      handshakeMessage.deviceObject = deviceObject;
      handshakeMessage.ver = await Util.getAppVersion();

      var jsonMap = handshakeMessage.toJson();
      var json = jsonEncode(jsonMap);
      var uri = "https://init-x.assecutor.de/bootstrap?get_mqtt_broker=$json";
      Util.logInfo("Bootstrap request: $uri");

      var response = await http.get(Uri.parse(uri));
      var responseText = response.body;

      var bootstrapResponsePayload = BootstrapResponsePayload.fromJson(jsonDecode(responseText)["payload"]);

      var brokerUriAndPortArray = bootstrapResponsePayload.config.broker.replaceFirst("mqtt://", "").split(":");
      _brokerUri = brokerUriAndPortArray[0];
      _brokerPort = int.parse(brokerUriAndPortArray[1]);
      _fromServerTopic = bootstrapResponsePayload.config.topics.fromServer;
      _toServerTopic = bootstrapResponsePayload.config.topics.toServer;

      _connectToBroker();
    } catch (ex) {
      _startReconnectTimer();
    }
  }

  _initSession() {
    var playload = Payload();
    playload.startSession = "true";

    var startSessionMessage = StartSessionMessage();
    startSessionMessage.recv = "server";
    startSessionMessage.store = true;
    startSessionMessage.messageId = const Uuid().v4();
    startSessionMessage.payload = playload;

    var jsonMap = startSessionMessage.toJson();
    var json = jsonEncode(jsonMap);
    Util.logInfo("send start session json: $json");

    final builder = MqttClientPayloadBuilder();
    builder.addString(json);
    var payload = builder.payload;

    var topic = "$_toServerTopic/init";
    Util.logInfo("send start session json to topic: $topic");

    var messageId = _mqttClient.publishMessage(topic, MqttQos.exactlyOnce, payload!);
    Util.logInfo("send start session json with messageId: $messageId");
  }

  _sendSyncMessage() async
  {
    var jobs = await MyApp.isarInstance.jobs.filter().stateGreaterThan(-1).findAll();

    var answer = SyncJobsMessage();
    answer.utcUnixTime = DateTime.now().toUtc().millisecondsSinceEpoch ~/ 1000;
    answer.sync = SyncJobMessageSync();
    answer.sync.jobs = [];
    for (var job in jobs) {
      SyncJobMessageJob syncJob = SyncJobMessageJob();
      syncJob.state = job.state;
      syncJob.id = job.id;
      syncJob.ver = job.ver.toString();
      answer.sync.jobs.add(syncJob);
    }

    MqttCore.instance.publishPayloadMessage("server", answer.toJson());
  }

  void _startTrackTimer() {
    _trackTimer ??= Timer.periodic(const Duration(seconds: 60), (Timer t) {
        var position = MyApp.currentPosition;
        if (position != null) {
          var trackMessage = TrackMessage();
          trackMessage.pos = TrackPosition();
          trackMessage.pos!.time = position.timestamp.toIso8601String();
          trackMessage.pos!.lat = position.latitude;
          trackMessage.pos!.long = position.longitude;
          trackMessage.pos!.acc = position.accuracy.toInt();

          publishPayloadMessage("track", trackMessage.toJson());
        }
      });
  }

  void _cancelTrackTimer() {
    if (_trackTimer != null && _trackTimer!.isActive) {
      _trackTimer!.cancel();
      _trackTimer = null;
    }
  }

  void _enqueueMessage(String receiver, String payload) async {
    var enqueuedMessage = EnqueuedMessage();
    enqueuedMessage.topic = receiver;
    enqueuedMessage.payload = payload;

    await MyApp.isarInstance.writeTxn(() async {
      await MyApp.isarInstance.enqueuedMessages.put(enqueuedMessage);
    });
  }

  _sendEnqueuedMessages() async {
    var enqueuedMessages = await MyApp.isarInstance.enqueuedMessages.where().findAll();
    if (enqueuedMessages.isNotEmpty) {
      for (var enqueuedMessage in enqueuedMessages) {
        try {
          publishText(enqueuedMessage.topic, enqueuedMessage.payload);

          await MyApp.isarInstance.writeTxn(() async {
            await MyApp.isarInstance.enqueuedMessages.delete(enqueuedMessage.id);
          });
        } catch (ex) {
          Util.logInfo("resend enqueued message failed: $ex");
        }
      }
    }
  }

  void _acknowlegdeJob(Job job) {
    var message = AcknowledgeJob();
    message.received = AcknowledgeJobReceived();
    message.received.id = job.id;
    message.received.state = job.state;
    message.received.ver = job.ver;
    message.received.utc = DateTime.now().millisecondsSinceEpoch;

    MqttCore.instance.publishPayloadMessage('job', message.toJson());
  }
}