import 'dart:async';
import 'dart:convert';

import 'package:archive/archive.dart';
import 'package:dart_mq/dart_mq.dart';
import 'package:flutter_test/flutter_test.dart';
import 'package:mqtt5_client/mqtt5_client.dart';
import 'package:mqtt5_client/mqtt5_server_client.dart';
import 'package:shared_preferences/shared_preferences.dart';
import 'package:stadtbote/main.dart';
import 'package:stadtbote/model/enqueued_message.dart';
import 'package:stadtbote/model/job.dart';
import 'package:stadtbote/model/log_entry.dart';
import 'package:stadtbote/objectbox.g.dart';
import 'package:stadtbote/tools/constants.dart';
import 'package:stadtbote/tools/message_header.dart';
import 'package:stadtbote/tools/messages/publish_payload_message.dart';
import 'package:stadtbote/tools/mqtt_core.dart';
import 'package:stadtbote/tools/util.dart';
// ignore: depend_on_referenced_packages
import 'package:typed_data/typed_buffers.dart';

void main() {
  TestWidgetsFlutterBinding.ensureInitialized();

  late _MessageStore store;
  late _MqttClient client;

  setUp(() async {
    SharedPreferences.setMockInitialValues({});
    await Util.init();
    MQClient.initialize();
    for (final topic in [
      Constants.messageTopicPublishPayload,
      Constants.messageTopicDisconnectClient,
      Constants.messageTopicConnectClient,
      Constants.messageTopicRequestConnectionStatus,
      Constants.messageTopicConnectionStatusChanged,
    ]) {
      MQClient.instance.declareQueue(topic);
    }
    store = _MessageStore();
    MyApp.objectBoxStore = store;
    client = _MqttClient();
    MqttCore.withClient(
      client: client,
      clientId: 'test-device',
      fromServerTopic: 'server/a/b/c/d/e/',
      toServerTopic: 'client/a/b/c/d/e/',
    );
  });

  tearDown(() async {
    await client.incoming.close();
    MQClient.instance.close();
  });

  Future<void> publish(WidgetTester tester, {bool requiresAck = true}) async {
    MQClient.instance.sendMessage(
      routingKey: Constants.messageTopicPublishPayload,
      message: Message(
        payload: PublishPayloadMessage(
          topic: 'mssg',
          message: {'text': 'Auftrag erledigt'},
          acknowledgeRequired: requiresAck,
        ),
      ),
    );
    await tester.pump();
  }

  Future<void> retry(WidgetTester tester) async {
    client.receive('sync_request', {});
    await tester.pump();
  }

  testWidgets('behält ID, Inhalt und Bestätigungspflicht bei Wiederholungen', (
    tester,
  ) async {
    // Auch beim ersten Sendeversuch muss die Nachricht bereits gespeichert sein.
    List<EnqueuedMessage>? queuedAtFirstSend;
    client.onPublish = () => queuedAtFirstSend = store.messages.getAll();
    await publish(tester);
    client.onPublish = null;
    final original = client.chatMessages.single;
    final queued = store.messages.rows.single;
    expect(queuedAtFirstSend, hasLength(1));
    expect(queuedAtFirstSend!.single.messageId, original.messageId);
    expect(queued.messageId, original.messageId);
    expect(jsonDecode(queued.payload), original.payload);

    await retry(tester);
    await retry(tester);

    expect(client.chatMessages, hasLength(3));
    for (final message in client.chatMessages) {
      expect(message.toJson(), original.toJson());
      expect(message.requiresAck, isTrue);
    }
    expect(store.messages.rows.single.id, queued.id);
    expect(store.messages.putCount, 1);
  });

  testWidgets('sendet bestätigte Nachrichten nicht erneut', (tester) async {
    await publish(tester);
    final messageId = client.chatMessages.single.messageId;
    await retry(tester);
    expect(store.messages.rows, hasLength(1));

    client.receive('acknowledge', {'messageId': messageId});
    await tester.pump();
    expect(store.messages.rows, isEmpty);

    await retry(tester);
    expect(client.chatMessages, hasLength(2));
  });

  testWidgets('behält unbestätigte Nachrichten nach fehlgeschlagenem Versand', (
    tester,
  ) async {
    client.failPublish = true;
    await publish(tester);
    final messageId = store.messages.rows.single.messageId;
    await retry(tester);
    expect(store.messages.rows.single.messageId, messageId);
    expect(client.chatMessages, isEmpty);

    client.failPublish = false;
    await retry(tester);
    expect(client.chatMessages.single.messageId, messageId);
    expect(client.chatMessages.single.requiresAck, isTrue);
    expect(store.messages.rows, hasLength(1));
  });

  testWidgets('verwendet die ID bereits gespeicherter Nachrichten', (
    tester,
  ) async {
    store.messages.put(
      EnqueuedMessage()
        ..messageId = 'persisted-message-id'
        ..topic = 'mssg'
        ..payload = '{"text":"Vor dem Neustart gespeichert"}',
    );

    await retry(tester);

    final message = client.chatMessages.single;
    expect(message.messageId, 'persisted-message-id');
    expect(message.requiresAck, isTrue);
    expect(message.payload['text'], 'Vor dem Neustart gespeichert');
    expect(store.messages.rows, hasLength(1));
    expect(store.messages.putCount, 1);
  });

  testWidgets('wiederholt Nachrichten ohne Bestätigungspflicht nicht', (
    tester,
  ) async {
    await publish(tester, requiresAck: false);
    await retry(tester);

    expect(client.chatMessages, hasLength(1));
    expect(client.chatMessages.single.requiresAck, isFalse);
    expect(store.messages.rows, isEmpty);
  });

  testWidgets(
    'vergibt für neue Nachrichten auch bei gleichem Inhalt eigene IDs',
    (tester) async {
      await publish(tester);
      await publish(tester);

      final ids = client.chatMessages
          .map((message) => message.messageId)
          .toList();
      expect(ids.toSet(), hasLength(2));
      await retry(tester);
      expect(client.chatMessages.map((message) => message.messageId), [
        ...ids,
        ...ids,
      ]);
      expect(store.messages.rows, hasLength(2));
    },
  );

  testWidgets('sendet bei fehlgeschlagener Speicherung keine neue Nachricht', (
    tester,
  ) async {
    store.messages.failPut = true;
    await publish(tester);

    expect(client.chatMessages, isEmpty);
    expect(store.messages.rows, isEmpty);
  });
}

class _MqttClient extends Fake implements MqttServerClient {
  final incoming =
      StreamController<List<MqttReceivedMessage<MqttMessage>>>.broadcast();
  final chatMessages = <MessageHeader>[];
  bool failPublish = false;
  void Function()? onPublish;

  @override
  Stream<List<MqttReceivedMessage<MqttMessage>>> get updates => incoming.stream;

  @override
  MqttSubscription? subscribe(String topic, MqttQos qosLevel) => null;

  @override
  int publishMessage(
    String topic,
    MqttQos qualityOfService,
    Uint8Buffer data, {
    bool retain = false,
    List<MqttUserProperty>? userProperties,
  }) {
    onPublish?.call();
    if (failPublish) throw StateError('Keine Verbindung');
    if (topic.endsWith('/mssg')) {
      chatMessages.add(
        MessageHeader.fromJson(
          jsonDecode(utf8.decode(GZipDecoder().decodeBytes(data))),
        ),
      );
    }
    return 1;
  }

  void receive(String topic, Map<String, dynamic> payload) {
    final data = Uint8Buffer()
      ..addAll(
        GZipEncoder().encode(
          utf8.encode(
            jsonEncode({
              'messageId': 'server-message-id',
              'requiresAck': false,
              'payload': payload,
            }),
          ),
        ),
      );
    final message = MqttPublishMessage()
        .toTopic('server/a/b/c/d/e/test-device/$topic')
        .publishData(data);
    incoming.add([MqttReceivedMessage(topic, message)]);
  }
}

class _MessageStore extends Fake implements Store {
  final messages = _MessageBox();
  final logs = _MemoryBox<LogEntry>();
  final jobs = _MemoryBox<Job>();

  @override
  Box<T> box<T>() {
    if (T == EnqueuedMessage) return messages as Box<T>;
    if (T == LogEntry) return logs as Box<T>;
    if (T == Job) return jobs as Box<T>;
    throw StateError('Unerwarteter Entity-Typ: $T');
  }
}

class _MemoryBox<T> extends Fake implements Box<T> {
  final rows = <T>[];

  @override
  List<T> getAll() => List.of(rows);

  @override
  int put(T object, {PutMode mode = PutMode.put}) => 0;

  @override
  QueryBuilder<T> query([Condition<T>? qc]) => _MemoryQueryBuilder(rows);
}

class _MessageBox extends _MemoryBox<EnqueuedMessage> {
  int putCount = 0;
  bool failPut = false;

  @override
  int put(EnqueuedMessage object, {PutMode mode = PutMode.put}) {
    if (failPut) throw StateError('Speicherung fehlgeschlagen');
    object.id = ++putCount;
    rows.add(object);
    return object.id;
  }

  @override
  int removeMany(List<int> ids) {
    final count = rows.length;
    rows.removeWhere((row) => ids.contains(row.id));
    return count - rows.length;
  }
}

// Die nativen Datenbankzugriffe sind ersetzt. Im Bestätigungstest enthält
// die Warteschlange genau die eine zum ACK passende Nachricht.
class _MemoryQueryBuilder<T> extends Fake implements QueryBuilder<T> {
  final List<T> rows;

  _MemoryQueryBuilder(this.rows);

  @override
  Query<T> build() => _MemoryQuery(rows);
}

class _MemoryQuery<T> extends Fake implements Query<T> {
  final List<T> rows;

  _MemoryQuery(this.rows);

  @override
  List<T> find() => List.of(rows);

  @override
  void close() {}
}