Publish under nearle/pos, add a health heartbeat, send the GST slab split

Three changes, all driven by what the back office turned out to need.

The broker is shared with the rider fleet on nearle/riders/…, so topics
move under nearle/pos/{locationid}/{terminal}/… — one ACL rule per
system, and it is obvious from a topic which one owns it. Store ID now
carries the back office's numeric location id; the tenant is resolved
from it server-side and never taken from the wire.

A till publishes a heartbeat every 30 seconds on its own topic. The Last
Will already answers "is it dead", which is not enough to run a hundred
shops on: the failure that costs money is a terminal that is connected,
selling, and quietly holding two hundred bills it has never uploaded. So
the beat carries queue depth, the age of the oldest thing waiting,
today's trading, and printer reachability. Not retained — the back
office holds it under a TTL, and a retained beat would leave an
unplugged till looking alive until something overwrote it.

Bills now carry tax_breakdown, the GST slab split the cart already
computes. A tax return is filed per slab, and recomputing the split
server-side would mean redoing the discount apportionment and getting
exactly the same answer — or else the filed figure stops matching the
paper the shopper was handed.

Docs rewritten against the real deployment: Eclipse Mosquitto 2.1.2, no
NATS anywhere reachable, no TLS, and a broker whose queue and autosave
defaults mean it must not be treated as durable storage.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

lib/core/config/sync_config.dart

@@ -73,7 +73,14 @@ class SyncConfig {
       };
 
   // ------------------------------------------------------------------ Topics
-  String get _base => 'pos/$storeId/$terminalId';
+  /// Namespaced under `nearle/` alongside the rider fleet's
+  /// `nearle/riders/{riderId}/…`, so one broker ACL rule covers each system and
+  /// a topic says at a glance which one it belongs to.
+  ///
+  /// [storeId] carries the back office's numeric location id. The tenant is
+  /// resolved from it server-side and never taken from the wire — a till that
+  /// could name its own tenant could post sales into another shop's books.
+  String get _base => 'nearle/pos/$storeId/$terminalId';
 
   /// Uplink. Completed bills, QoS 1.
   String get orderTopic => '$_base/order';
@@ -99,8 +106,16 @@ class SyncConfig {
   /// what makes a head-office "which tills are dark" board possible.
   String get statusTopic => '$_base/status';
 
+  /// Liveness, published on a timer rather than on an event.
+  ///
+  /// Separate from [statusTopic]: that one is retained and doubles as the Last
+  /// Will, so it must stay small and rarely written. This carries queue depth,
+  /// today's trading and device state — the things a head-office board needs to
+  /// tell a till that is merely quiet from one that is in trouble.
+  String get healthTopic => '$_base/health';
+
   /// Store-wide downlink: catalogue changes land here for every terminal.
-  String get catalogueTopic => 'pos/$storeId/catalogue';
+  String get catalogueTopic => 'nearle/pos/$storeId/catalogue';
 
   /// Addressed to this terminal alone.
   String get commandTopic => '$_base/command';
@@ -118,7 +133,8 @@ class SyncConfig {
   /// NATS' MQTT gateway maps `/` to `.`, so this is what a JetStream stream or
   /// consumer is configured against. Provided so the wildcard a back-office
   /// consumer needs can be read off the terminal rather than guessed:
-  /// `pos.*.*.order` for every till's bills, `pos.*.*.status` for presence.
+  /// `nearle.pos.*.*.order` for every till's bills, `nearle.pos.*.*.health`
+  /// for presence.
   static String asNatsSubject(String topic) => topic.replaceAll('/', '.');
 
   SyncConfig copyWith({

lib/data/remote/mqtt_order_transport.dart

@@ -258,6 +258,18 @@ class MqttOrderTransport implements OrderTransport {
     _publish(config.statusTopic, payload, retain: true);
   }
 
+  /// Publishes a heartbeat, deliberately *not* retained.
+  ///
+  /// The back office holds these in Redis under a TTL, so a terminal that
+  /// stops beating ages off the board by itself. A retained heartbeat would
+  /// survive on the broker after the till was unplugged and keep it looking
+  /// alive until something happened to overwrite it — which is exactly the
+  /// failure a health board exists to catch.
+  Future<void> publishHealth(String payload) async {
+    if (!isConnected) return;
+    _publish(config.healthTopic, payload);
+  }
+
   /// Registers a batch as awaiting its ack, without publishing one.
   ///
   /// Lets a test drive the correlation rules — which is where the logic that

lib/data/repositories/sync_repository_impl.dart

@@ -463,6 +463,15 @@ class SyncRepositoryImpl implements SyncRepository {
             },
         ],
         'tax': t.cart.taxAmount,
+        // GST per slab, as printed on the invoice. Sent as well as the total
+        // because a compliant tax return is filed per slab, and recomputing the
+        // split server-side from line items would have to redo the discount
+        // apportionment — and get exactly the same answer, or the filed figure
+        // stops matching the paper the shopper was handed.
+        'tax_breakdown': {
+          for (final entry in t.cart.taxBreakdown.entries)
+            entry.key.toString(): entry.value,
+        },
         'round_off': t.cart.roundOff,
         'total': t.total,
         'points_earned': t.pointsEarned,

lib/data/sync/health_reporter.dart

@@ -0,0 +1,216 @@
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+
+import '../../core/config/sync_config.dart';
+import '../../domain/repositories/sync_repository.dart';
+import '../local/terminal_identity.dart';
+import '../remote/mqtt_order_transport.dart';
+import 'sync_engine.dart';
+
+/// What a till reports about itself, every 30 seconds.
+///
+/// The Last Will already answers *is it dead* — the broker publishes `offline`
+/// on a terminal's behalf when it stops responding. That is not enough to run a
+/// hundred shops on, because the failure that actually costs money looks
+/// completely healthy from outside: a till that is connected, selling, and
+/// quietly accumulating two hundred bills it has never managed to upload.
+///
+/// So this carries the numbers that separate *reachable* from *well*: how deep
+/// the queue is, how long the oldest thing in it has been waiting, whether the
+/// till has rung anything today, and whether the hardware is in the way.
+///
+/// Not retained, and never acknowledged. A heartbeat is a fact with an expiry
+/// date — the back office holds it in Redis under a TTL, so a terminal that
+/// loses power ages off the board by itself. Retaining it would leave a dead
+/// till looking alive until something overwrote it.
+class HealthReporter {
+  HealthReporter({
+    required MqttOrderTransport transport,
+    required TerminalIdentity terminal,
+    required SyncConfig config,
+    required SyncEngine engine,
+    required SyncRepository repository,
+    required this.appVersion,
+    this.deviceState,
+    this.printerEndpoint,
+    Duration interval = const Duration(seconds: 30),
+    DateTime Function()? clock,
+    Timer Function(Duration, void Function())? scheduleTimer,
+  })  : _transport = transport,
+        _terminal = terminal,
+        _config = config,
+        _engine = engine,
+        _repository = repository,
+        _interval = interval,
+        _now = clock ?? DateTime.now,
+        _schedule = scheduleTimer ?? Timer.new;
+
+  final MqttOrderTransport _transport;
+  final TerminalIdentity _terminal;
+  final SyncConfig _config;
+  final SyncEngine _engine;
+  final SyncRepository _repository;
+  final Duration _interval;
+  final DateTime Function() _now;
+  final Timer Function(Duration, void Function()) _schedule;
+
+  final String appVersion;
+
+  /// Hardware readings, if this build collects any.
+  ///
+  /// A hook rather than a hard dependency: battery level and free storage need
+  /// platform packages that a desktop build has no use for, and a health board
+  /// is not a good enough reason to make the whole app depend on them. What is
+  /// not collected is *omitted* rather than sent as zero — a dashboard showing
+  /// every till at 0% battery is worse than one showing nothing.
+  final Future<Map<String, Object?>> Function()? deviceState;
+
+  /// Where the receipt printer lives, if one is configured.
+  ///
+  /// Read fresh on each beat rather than captured once, because a shop can
+  /// re-point its printer in Settings without restarting the till.
+  final ({String host, int port})? Function()? printerEndpoint;
+
+  Timer? _timer;
+  bool _stopped = false;
+
+  Future<void> start() async {
+    if (_stopped) throw StateError('This HealthReporter has been disposed.');
+    await publish();
+    _tick();
+  }
+
+  void _tick() {
+    _timer = _schedule(_interval, () {
+      if (_stopped) return;
+      unawaited(publish());
+      _tick();
+    });
+  }
+
+  /// One heartbeat.
+  ///
+  /// Every failure is swallowed. A terminal that cannot say how it is must
+  /// still sell — losing a heartbeat is a monitoring gap, and stopping a till
+  /// because a dashboard is unreachable would be a self-inflicted outage.
+  Future<void> publish() async {
+    if (!_transport.isConnected) return;
+
+    try {
+      final state = _engine.state;
+
+      final pendingBills = state.pending;
+      final pendingRegistrations = await _repository.unsyncedCustomerCount();
+
+      // Terminal-wide rather than scoped to whoever is signed in: the board
+      // watches a till, not a shift.
+      final today = await _repository.todayReport(
+        terminalId: _terminal.code,
+        cashierName: '',
+      );
+
+      final payload = <String, Object?>{
+        'schema': 1,
+        'status': 'online',
+
+        'terminal_id': _terminal.code,
+        'device_id': _terminal.deviceId,
+        'terminal_name': _terminal.name,
+        'location_id': _config.storeId,
+        'store_name': _terminal.name,
+        'app_version': appVersion,
+        'transport': _config.transport.name,
+
+        // Queue depth — the number that makes a silent failure visible.
+        'pending_bills': pendingBills,
+        'pending_registrations': pendingRegistrations,
+        'oldest_pending_at': (await _oldestPendingAt())?.toIso8601String(),
+        'sync_halted': state.isHalted,
+        'sync_error': state.lastError,
+        'last_upload_at': state.lastSuccessAt?.toIso8601String(),
+
+        // Today's trading. A till that is connected but has rung nothing in
+        // three hours is usually a jammed printer or an absent cashier, and
+        // neither shows up on an online/offline board.
+        'today_bills': today.billCount,
+        'today_amount': today.grossSales,
+        'last_bill_at': today.lastBillAt?.toIso8601String(),
+
+        'reported_at': _now().toIso8601String(),
+      };
+
+      final device = await _collectDeviceState();
+      payload.addAll(device);
+
+      await _transport.publishHealth(jsonEncode(payload));
+    } on Object {
+      // Deliberately silent — see above.
+    }
+  }
+
+  /// When the oldest unsent bill was rung.
+  ///
+  /// More useful than the count on its own: fifty bills queued in the last ten
+  /// minutes is a broker hiccup, while three queued since Tuesday is a till
+  /// nobody has looked at.
+  Future<DateTime?> _oldestPendingAt() async {
+    try {
+      final rows = await _repository.orderSyncRows(limit: 500);
+      DateTime? oldest;
+      for (final row in rows) {
+        if (row.isSynced) continue;
+        if (oldest == null || row.createdAt.isBefore(oldest)) {
+          oldest = row.createdAt;
+        }
+      }
+      return oldest;
+    } on Object {
+      return null;
+    }
+  }
+
+  /// Hardware readings, plus whatever this build can work out for itself.
+  Future<Map<String, Object?>> _collectDeviceState() async {
+    final out = <String, Object?>{};
+
+    if (deviceState != null) {
+      try {
+        out.addAll(await deviceState!());
+      } on Object {
+        // A missing battery reading must not cost the rest of the heartbeat.
+      }
+    }
+
+    final printer = printerEndpoint?.call();
+    if (printer != null && printer.host.isNotEmpty) {
+      out['printer_reachable'] = await _canReach(printer.host, printer.port);
+    }
+
+    return out;
+  }
+
+  /// Opens and immediately closes a socket to the till's printer.
+  ///
+  /// Cheap enough to run every 30 seconds, and it answers the question a shop
+  /// actually phones about — a printer that is switched off looks identical to
+  /// a working one until someone tries to print a bill.
+  Future<bool> _canReach(String host, int port) async {
+    try {
+      final socket = await Socket.connect(
+        host,
+        port,
+        timeout: const Duration(seconds: 2),
+      );
+      socket.destroy();
+      return true;
+    } on Object {
+      return false;
+    }
+  }
+
+  Future<void> dispose() async {
+    _stopped = true;
+    _timer?.cancel();
+  }
+}

lib/presentation/sync/providers/sync_controller.dart

@@ -3,7 +3,9 @@ import 'package:flutter_riverpod/flutter_riverpod.dart';
 import '../../../app/providers.dart';
 import '../../../core/constants/app_constants.dart';
 import '../../../data/remote/mqtt_order_transport.dart';
+import '../../../data/sync/health_reporter.dart';
 import '../../../data/sync/presence_reporter.dart';
+import '../../modules/providers/printer_settings.dart';
 import '../../../domain/entities/shift_report.dart';
 import '../../../domain/entities/sync_event.dart';
 import '../../../domain/repositories/sync_repository.dart';
@@ -251,5 +253,29 @@ final syncBootstrapProvider = FutureProvider<void>((ref) async {
     );
     ref.onDispose(reporter.dispose);
     await reporter.start();
+
+    // The 30-second heartbeat the head-office board reads. Separate from the
+    // retained presence record above: that one is paired with the Last Will and
+    // answers "is this till alive", while this carries queue depth, today's
+    // trading and hardware state — what tells a till that is merely quiet from
+    // one that has stopped uploading.
+    final health = HealthReporter(
+      transport: transport,
+      terminal: ref.read(terminalIdentityProvider),
+      config: ref.read(syncConfigProvider),
+      engine: engine,
+      repository: ref.read(syncRepositoryProvider),
+      appVersion: AppConstants.appVersion,
+      // Read on each beat, so re-pointing the printer in Settings takes effect
+      // without a restart.
+      printerEndpoint: () {
+        final printer = ref.read(printerSettingsProvider);
+        final host = printer.drawerHost;
+        if (host == null || host.isEmpty) return null;
+        return (host: host, port: printer.drawerPort);
+      },
+    );
+    ref.onDispose(health.dispose);
+    await health.start();
   }
 });