nearle_pos/test/unit/transport_test.dart

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

import 'package:flutter_test/flutter_test.dart';
import 'package:http/http.dart' as http;
import 'package:nearle_pos/core/config/sync_config.dart';
import 'package:nearle_pos/data/remote/http_order_transport.dart';
import 'package:nearle_pos/data/remote/mqtt_order_transport.dart';
import 'package:nearle_pos/data/remote/order_transport.dart';

/// Two bills, enough to tell "all accepted" from "some accepted".
final _orders = [
  {'id': 'order-a', 'invoice_number': 'INV-1', 'total': 100.0},
  {'id': 'order-b', 'invoice_number': 'INV-2', 'total': 250.0},
];

void main() {
  group('MQTT ack correlation', () {
    late MqttOrderTransport transport;
    const config = SyncConfig(
      transport: TransportKind.mqtt,
      storeId: 'store-9',
      terminalId: 'TERM-04',
      brokerHost: 'broker.invalid',
    );

    setUp(() => transport = MqttOrderTransport(config: config));
    tearDown(() => transport.dispose());

    test('topics are namespaced per store and per terminal', () {
      // Two stores sharing one broker must never see each other's bills.
      expect(config.orderTopic, 'nearle/pos/store-9/TERM-04/order');
      expect(config.ackTopic, 'nearle/pos/store-9/TERM-04/ack');
      expect(config.statusTopic, 'nearle/pos/store-9/TERM-04/status');
      expect(config.healthTopic, 'nearle/pos/store-9/TERM-04/health');
      expect(config.catalogueTopic, 'nearle/pos/store-9/catalogue');
    });

    test('an ack naming only some ids accepts only those', () async {
      // The heart of it. A partial ack must not be read as "the batch went up":
      // order-b stays pending and is sent again.
      final receipt = _receiptFor(transport, {
        'accepted': ['order-a'],
        'rejected': {'order-b': 'unknown product'},
      });

      final result = await receipt;
      expect(result.accepted, ['order-a']);
      expect(result.rejected, {'order-b': 'unknown product'});
    });

    test('an ack for a different batch does not release this one', () async {
      final pending = transport.pushOrders(_orders).timeout(
            const Duration(milliseconds: 300),
            onTimeout: () => throw TimeoutException('not released'),
          );

      // Give the publish a turn to register its correlation id, then answer
      // with someone else's.
      await Future<void>.delayed(Duration.zero);
      transport.handleInbound(
        config.ackTopic,
        jsonEncode({'batch_id': 'a-different-batch', 'accepted': ['order-a']}),
      );

      await expectLater(pending, throwsA(isA<Exception>()));
    });

    test('a malformed ack is discarded rather than taken down the connection',
        () {
      // The next message may be a perfectly good ack releasing a day's bills.
      expect(
        () => transport.handleInbound(config.ackTopic, 'not json at all'),
        returnsNormally,
      );
      expect(
        () => transport.handleInbound(config.ackTopic, '{"no":"batch id"}'),
        returnsNormally,
      );
    });

    test('an ack with no accepted list releases nothing', () async {
      // Silence is not acceptance. A back office that answers `{}` must not
      // cause a single bill to be marked synced.
      final result = await _receiptFor(transport, {'accepted': <String>[]});
      expect(result.accepted, isEmpty);
    });

    test('a bare list of rejected ids is tolerated', () async {
      final result = await _receiptFor(transport, {
        'accepted': ['order-a'],
        'rejected': ['order-b'],
      });
      expect(result.rejected.keys, ['order-b']);
    });

    test('catalogue pushes arrive on the downlink', () async {
      final received = <DownlinkMessage>[];
      final sub = transport.downlink.listen(received.add);

      transport
        ..handleInbound(config.catalogueTopic, jsonEncode({'revision': 'r9'}))
        ..handleInbound(config.commandTopic, jsonEncode({'command': 'sync'}))
        ..handleInbound(
          config.commandTopic,
          jsonEncode({'command': 'self-destruct'}),
        );

      await Future<void>.delayed(Duration.zero);
      await sub.cancel();

      expect(
        received.map((m) => m.kind),
        [
          DownlinkKind.catalogueChanged,
          DownlinkKind.syncRequested,
          // A newer server talking to an older terminal stays visible instead
          // of being silently dropped.
          DownlinkKind.unknown,
        ],
      );
      expect(received.first.payload['revision'], 'r9');
    });
  });

  group('HTTP transport', () {
    const SyncConfig config = SyncConfig(
      transport: TransportKind.http,
      httpBaseUrl: 'https://back.office.test',
      apiKey: 'k',
    );

    test('accepts only the ids the endpoint names', () async {
      final transport = HttpOrderTransport(
        config: config,
        client: _FakeClient((_) => http.Response(
              jsonEncode({
                'accepted': ['order-a'],
                'rejected': {'order-b': 'stale price list'},
              }),
              200,
            ),),
      );

      final receipt = await transport.pushOrders(_orders);
      expect(receipt.accepted, ['order-a']);
      expect(receipt.rejected['order-b'], 'stale price list');
      await transport.dispose();
    });

    test('a bare 200 with no body marks nothing synced', () async {
      // Guessing here would retire a day's takings on an empty response.
      final transport = HttpOrderTransport(
        config: config,
        client: _FakeClient((_) => http.Response('', 200)),
      );

      await expectLater(
        transport.pushOrders(_orders),
        throwsA(isA<TransportException>()),
      );
      await transport.dispose();
    });

    test('a 200 that names nothing accepts nothing, without throwing',
        () async {
      final transport = HttpOrderTransport(
        config: config,
        client: _FakeClient((_) => http.Response('{"accepted":[]}', 200)),
      );

      final receipt = await transport.pushOrders(_orders);
      expect(receipt.accepted, isEmpty);
      await transport.dispose();
    });

    test('a bad credential is not retryable, so the engine can halt', () async {
      // Hammering the endpoint would only bury the one message a person needs.
      final transport = HttpOrderTransport(
        config: config,
        client: _FakeClient((_) => http.Response('nope', 401)),
      );

      await expectLater(
        transport.pushOrders(_orders),
        throwsA(isA<TransportException>()
            .having((e) => e.retryable, 'retryable', isFalse),),
      );
      await transport.dispose();
    });

    test('a server error is retryable', () async {
      final transport = HttpOrderTransport(
        config: config,
        client: _FakeClient((_) => http.Response('boom', 503)),
      );

      await expectLater(
        transport.pushOrders(_orders),
        throwsA(isA<TransportException>()
            .having((e) => e.retryable, 'retryable', isTrue),),
      );
      await transport.dispose();
    });

    test('a retry of the same bills carries the same idempotency key',
        () async {
      // So the endpoint can collapse a duplicate batch server-side rather than
      // relying on every order id being checked one at a time.
      final keys = <String>[];
      final transport = HttpOrderTransport(
        config: config,
        client: _FakeClient((request) {
          keys.add(request.headers['idempotency-key'] ?? '');
          return http.Response('{"accepted":["order-a","order-b"]}', 200);
        }),
      );

      await transport.pushOrders(_orders);
      await transport.pushOrders(_orders);
      expect(keys.first, keys.last);
      expect(keys.first, isNotEmpty);

      await transport.pushOrders([_orders.first]);
      expect(keys.last, isNot(keys.first));

      await transport.dispose();
    });

    test('an unconfigured endpoint fails fast rather than retrying', () async {
      final transport = HttpOrderTransport(
        config: const SyncConfig(transport: TransportKind.http),
        client: _FakeClient((_) => http.Response('', 200)),
      );

      await expectLater(
        transport.pushOrders(_orders),
        throwsA(isA<TransportException>()
            .having((e) => e.retryable, 'retryable', isFalse),),
      );
      await transport.dispose();
    });
  });
}

/// Waits on a known batch id, then feeds it [body] as if the back office had
/// answered — exercising the real parse and correlation path with no broker.
Future<PushReceipt> _receiptFor(
  MqttOrderTransport transport,
  Map<String, Object?> body,
) {
  const batchId = 'test-batch';
  final receipt = transport.awaitAck(batchId);

  transport.handleInbound(
    transport.config.ackTopic,
    jsonEncode({...body, 'batch_id': batchId}),
  );

  return receipt;
}

class _FakeClient extends http.BaseClient {
  _FakeClient(this.respond);

  final http.Response Function(http.BaseRequest) respond;

  @override
  Future<http.StreamedResponse> send(http.BaseRequest request) async {
    final response = respond(request);
    return http.StreamedResponse(
      Stream.value(utf8.encode(response.body)),
      response.statusCode,
      request: request,
    );
  }
}