diff --git a/docs/integration-guide.md b/docs/integration-guide.md index dc5a0b2..ac00beb 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -13,7 +13,7 @@ Settings. ## The five hops ``` -1. NATS running with the MQTT gateway on +1. Mosquitto reachable, with a scoped account for the tills 2. Terminal connects and shows LIVE 3. Terminal publishes a bill 4. Your consumer commits it and acks @@ -26,59 +26,63 @@ guessing. --- -## Hop 1 — NATS with MQTT enabled +## Hop 1 — Mosquitto, with an account for the tills -MQTT needs JetStream turned on, because the gateway stores session state in it. +The deployed broker is **Eclipse Mosquitto 2.1.2** at `66.116.225.226:1883`, +already carrying `nearle/riders/#` and `doormile/#`. There is no NATS in this +estate — the NATS servers that exist belong to other projects, on hosts whose +ports are closed, and their configs expose no MQTT gateway. -```conf -# nats.conf -jetstream { - store_dir: /var/lib/nats - max_file: 10Gi -} +Current config has `allow_anonymous false` and a password file, but **no +`acl_file`** — so every authenticated user, including the `admin` account +hardcoded in the rider APK, has full run of every topic. Add scoped accounts +before a hundred tills start publishing takings: -mqtt { - port: 1883 - # TLS in production. Bills carry customer names and mobile numbers. - # tls { - # cert_file: "/etc/nats/server.pem" - # key_file: "/etc/nats/server-key.pem" - # } -} - -authorization { - users = [ - { user: "till", password: "…", permissions: { - publish: ["pos.*.*.order", "pos.*.*.customer", "pos.*.*.status"] - subscribe: ["pos.*.*.ack", "pos.*.*.command", "pos.*.catalogue"] - }} - ] -} +```bash +mosquitto_passwd -b /mosquitto/config/passwd pos_terminal '' +mosquitto_passwd -b /mosquitto/config/passwd pos_ingest '' ``` -Note the permissions are deliberately narrow. A terminal has no business -publishing to another terminal's ack topic. +```conf +# /mosquitto/config/acl, then add `acl_file /mosquitto/config/acl` to mosquitto.conf +user pos_terminal +topic write nearle/pos/+/+/order +topic write nearle/pos/+/+/customer +topic write nearle/pos/+/+/status +topic write nearle/pos/+/+/health +topic read nearle/pos/+/+/ack +topic read nearle/pos/+/+/command +topic read nearle/pos/+/catalogue + +user pos_ingest +topic read nearle/pos/+/+/order +topic read nearle/pos/+/+/customer +topic read nearle/pos/+/+/health +topic write nearle/pos/+/+/ack +topic write nearle/pos/+/catalogue + +user admin +topic readwrite nearle/riders/# +topic readwrite doormile/# +``` Prove it: ```bash -nats sub 'pos.>' & -mosquitto_pub -h localhost -p 1883 -t pos/test/T0000/order -m 'hello' -# the nats sub should print it — that is the / → . mapping working +mosquitto_sub -h 66.116.225.226 -p 1883 -u pos_ingest -P '' \ + -t 'nearle/pos/#' -v & +mosquitto_pub -h 66.116.225.226 -p 1883 -u pos_terminal -P '' \ + -t nearle/pos/12/T0000/order -m 'hello' ``` -**The stream must be file-backed.** A memory stream loses a shop's bills on a -server restart, and the terminal has already been told they landed. +**The broker is a transport, not a ledger.** `max_queued_messages` defaults to +1000 and `autosave_interval` to 30 minutes, so a long outage or a hard kill can +drop queued messages. That costs nothing here — an undelivered batch is never +acked, so the till keeps it and sends again — but only while nobody +acknowledges on the broker's behalf. -```bash -nats stream add POS_ORDERS \ - --subjects 'pos.*.*.order' \ - --storage file \ - --retention limits \ - --max-age 720h -``` - ---- +**TLS is not configured** and 8883 is closed. Bills carry customer names and +mobile numbers; worth adding a listener before rollout rather than after. ## Hop 2 — Point the terminal at it @@ -87,11 +91,11 @@ On the terminal: **Settings → Connectivity & sync → Configure**. | Field | Value | |---|---| | Terminal name | What staff call this till, e.g. "Counter 2" | -| Store ID | Namespaces the shop on the broker. Must match across its tills | +| Store ID | **The numeric `locationid`.** The tenant is resolved from it server-side | | Transport | MQTT | -| Broker host / port | Your NATS host, `1883` plain or `8883` TLS | -| Username / password | From the `authorization` block above | -| Use TLS | On in production | +| Broker host / port | `66.116.225.226`, port `1883`, **TLS off** | +| Username / password | The `pos_terminal` account from hop 1 | +| Use TLS | **Off** — 8883 is not configured on this broker yet | The dialog also shows a **Device ID** and a terminal code like `T4A9`. Neither is editable. They are minted on first run and stay with the physical machine, which @@ -104,7 +108,7 @@ Keystore), not into the database alongside the bills. Prove it: the header pill switches from `OFFLINE (SIM)` to `LIVE`, and ```bash -nats sub 'pos.*.*.status' +mosquitto_sub -h 66.116.225.226 -u pos_ingest -P '' -t 'nearle/pos/+/+/status' -v ``` should immediately show a retained presence record for the terminal. If it @@ -116,7 +120,7 @@ rejection before looking anywhere else. ## Hop 3 — Ring a sale and watch it publish ```bash -nats sub 'pos.*.*.order' +mosquitto_sub -h 66.116.225.226 -u pos_ingest -P '' -t 'nearle/pos/+/+/order' -v ``` Ring a bill on the terminal. Within a couple of seconds you should see the @@ -130,117 +134,69 @@ acknowledged it yet. That is correct behaviour, not a fault. ## Hop 4 — Consume, commit, acknowledge -This is the hop that matters. Two rules, both non-negotiable: +**This is already built**, in the Fiesta backend. See +`backend_fiesta/POS_TERMINAL_INGEST.md` for how to turn it on; what follows is +what it guarantees, so you can check it still holds if anyone changes it. -**Acknowledge from the consumer, after the database commit.** Not from an ingest -handler that has merely queued the work. That ack is the terminal's only -evidence, and it deletes its own copy seven days later on the strength of it. +Set `MQTT_URL` and it subscribes to `nearle/pos/+/+/{order,customer,health}`, +commits, and acknowledges. Bills land in `pos_orders` / `pos_order_items`, and +the stock they consumed goes through the same `productstocks` ledger an app +order uses. -**Be idempotent on `order.id`.** QoS 1 is at-least-once and a lost ack makes the -terminal re-send the whole batch. Every id is a UUID minted at the till, so this -costs you one unique index. +Three rules the implementation is built around, and that any replacement must +also keep: -```sql -CREATE TABLE orders ( - id UUID PRIMARY KEY, -- the terminal's order id - invoice_number TEXT NOT NULL, - store_id TEXT NOT NULL, - terminal_id TEXT NOT NULL, - cashier TEXT NOT NULL, - created_at TIMESTAMPTZ NOT NULL, - subtotal NUMERIC(12,2) NOT NULL, - discount NUMERIC(12,2) NOT NULL, - tax NUMERIC(12,2) NOT NULL, - round_off NUMERIC(12,2) NOT NULL, - total NUMERIC(12,2) NOT NULL, - payload JSONB NOT NULL -- keep the raw envelope -); +**Acknowledge from the consumer, after the database commit.** Not from a handler +that has merely queued the work. That ack is the terminal's only evidence, and +it deletes its own copy seven days later on the strength of it. --- Invoice numbers are unique per terminal, not globally. A till that was --- replaced restarts its own series, so gaps are normal and do not mean --- missing bills. -CREATE UNIQUE INDEX orders_invoice_per_terminal - ON orders (terminal_id, invoice_number); -``` +**A duplicate is accepted, not rejected.** QoS 1 is at-least-once and a lost ack +makes the terminal re-send the whole batch. Reporting those as failures would +strand a day of takings on the till. Deduplication is a unique index on the +till's UUID plus a Postgres advisory lock. -Consumer shape: +**Read the store and terminal from the topic, never the body.** A till that +could name its own store in a payload could redirect another counter's +acknowledgements. -```python -for msg in subscribe("pos.*.*.order"): - batch = json.loads(msg.data) - accepted, rejected = [], {} - - with db.transaction(): # one transaction for the batch - for order in batch["orders"]: - try: - db.execute(""" - INSERT INTO orders (id, invoice_number, …, payload) - VALUES (%(id)s, %(invoice_number)s, …, %(payload)s) - ON CONFLICT (id) DO NOTHING - """, order) - accepted.append(order["id"]) - except BusinessRuleError as e: - rejected[order["id"]] = str(e) - - # Published only after the commit above has succeeded. - publish(f"pos.{batch['store_id']}.{batch['terminal_id']}.ack", json.dumps({ - "batch_id": batch["batch_id"], - "accepted": accepted, - "rejected": rejected, - })) -``` - -`ON CONFLICT DO NOTHING` still counts as accepted — a redelivery of a bill you -already hold is a success, not a rejection. - -### Shopper registrations - -A second uplink runs on `pos/{store}/{terminal}/customer`, acked on the same -topic by the same rules. Wire it the same way: - -```sql -CREATE TABLE customers ( - id UUID PRIMARY KEY, -- derived from the mobile number - mobile TEXT NOT NULL UNIQUE, - name TEXT NOT NULL, - email TEXT, - gender TEXT, - date_of_birth DATE, - registered_at TIMESTAMPTZ, - registered_by_terminal TEXT -); -``` - -Insert with `ON CONFLICT (id) DO NOTHING` — **never** an upsert. A registration -is replayed freely and must not overwrite a profile corrected at head office. - -The id is a UUIDv5 over the shopper's normalised ten-digit mobile, so two tills -registering the same person independently produce the same row. Don't reassign -it. And don't expect loyalty points in this payload: derive those from the bill -stream, which is idempotent and knows about every counter. - -```bash -nats sub 'pos.*.*.customer' -``` - -Add a shopper on the terminal — no sale needed — and it should appear. - -**Use `rejected` sparingly.** Naming an id there halts the terminal's drain: it -stops retrying and waits for a person to press Sync. That is right for "this bill -is malformed" and wrong for "my database is having a bad minute" — for the -latter, don't ack at all and let the terminal back off and retry. - -Prove it by hand before wiring the real consumer: +Prove the ack path by hand before trusting the consumer: ```bash # copy batch_id and the order id from the hop-3 output -nats pub 'pos.store-01.T4A9.ack' \ - '{"batch_id":"","accepted":[""]}' +mosquitto_pub -h 66.116.225.226 -u pos_ingest -P '' \ + -t 'nearle/pos/12/T4A9/ack' \ + -m '{"batch_id":"","accepted":[""]}' ``` The pill should flip to `LIVE` and the bill disappear from the queue. ---- +### Shopper registrations + +A second uplink runs on `nearle/pos/{loc}/{terminal}/customer`, acked on the +same topic by the same rules, and handled by the same consumer. + +The id is a UUIDv5 over the shopper's normalised ten-digit mobile, so two tills +registering the same person independently produce the same row. It is stored +insert-if-absent — never an update, so a profile corrected at head office is not +reverted by a terminal replaying an old capture. No loyalty figures travel +upward: those are derived from the bill stream, which is idempotent and sees +every counter. + +```bash +mosquitto_sub -h 66.116.225.226 -u pos_ingest -P '' -t 'nearle/pos/+/+/customer' -v +``` + +Add a shopper on the terminal — no sale needed — and it should appear. + +### Terminal health + +Every till publishes to `nearle/pos/{loc}/{terminal}/health` every 30 seconds. +The consumer writes it to Redis as `pos:terminal:{code}` under a 90-second TTL, +so a till that loses power ages off the board by itself. Read it back at +`GET /live/api/v1/pos/health/location?location_id=12`. + +Heartbeats are never acknowledged — a till that could be blocked by a busy +dashboard would be a self-inflicted outage. ## Hop 5 — Catalogue down @@ -268,7 +224,8 @@ easiest to get wrong: To push a change mid-day rather than waiting for the next pull: ```bash -nats pub 'pos.store-01.catalogue' '{"revision":"rev-8822"}' +mosquitto_pub -h 66.116.225.226 -u pos_ingest -P '' \ + -t 'nearle/pos/12/catalogue' -m '{"revision":"rev-8822"}' ``` Every terminal in that store pulls immediately. @@ -280,7 +237,7 @@ Every terminal in that store pulls immediately. | Symptom | Where to look | |---|---| | Pill stuck on `OFFLINE (SIM)` | Simulate offline is still on in Settings | -| Pill shows `LIVE`, no presence on `pos.*.*.status` | Terminal never connected — check broker auth logs | +| Pill shows `LIVE`, no presence on `nearle/pos/+/+/status` | Terminal never connected — check broker auth logs | | Bills publish, queue never empties | You are acking the wrong `batch_id`, or not acking at all | | Queue empties then refills with the same bills | Ack arriving after `ackTimeout` (20s default) — the terminal gave up and re-sent | | Two terminals fighting for the connection | They share a client id. Each device mints its own; check they have different terminal codes | @@ -295,6 +252,9 @@ and per-bill state — start there before the broker logs. ## Before a fleet rollout +- **Add the broker ACL first** (hop 1). Today every authenticated user is + unrestricted on every topic, including the `admin` account hardcoded in the + rider APK. - **Back up `nearle_pos.db` on any terminal already trading.** The schema goes to v8 on first launch and the migration is one-way. It also queues every shopper already on the terminal for upload, so expect one burst of registrations from diff --git a/docs/sync-contract.md b/docs/sync-contract.md index e33b210..cf1153f 100644 --- a/docs/sync-contract.md +++ b/docs/sync-contract.md @@ -62,15 +62,21 @@ connection or return 5xx instead, and the terminal will back off and retry. | Topic | Direction | QoS | Retained | |---|---|---|---| -| `pos/{store}/{terminal}/order` | till → cloud | 1 | no | -| `pos/{store}/{terminal}/customer` | till → cloud | 1 | no | -| `pos/{store}/{terminal}/ack` | cloud → till | 1 | no | -| `pos/{store}/{terminal}/status` | till → cloud | 1 | **yes** | -| `pos/{store}/{terminal}/command` | cloud → till | 1 | no | -| `pos/{store}/catalogue` | cloud → all tills | 1 | **yes** | +| `nearle/pos/{loc}/{terminal}/order` | till → cloud | 1 | no | +| `nearle/pos/{loc}/{terminal}/customer` | till → cloud | 1 | no | +| `nearle/pos/{loc}/{terminal}/health` | till → cloud | 1 | no | +| `nearle/pos/{loc}/{terminal}/ack` | cloud → till | 1 | no | +| `nearle/pos/{loc}/{terminal}/status` | till → cloud | 1 | **yes** | +| `nearle/pos/{loc}/{terminal}/command` | cloud → till | 1 | no | +| `nearle/pos/{loc}/catalogue` | cloud → all tills | 1 | **yes** | -`{store}` and `{terminal}` come from the device's own identity, minted on first -run and stored in its database. They are never literals — 100 tills sharing one +Namespaced under `nearle/` alongside the rider fleet's `nearle/riders/…`, so one +broker ACL rule covers each system. + +`{loc}` is the back office's numeric location id, entered once in Settings; the +tenant is resolved from it server-side and never taken from the wire. +`{terminal}` comes from the device's own identity, minted on first run and +stored in its database. They are never literals — 100 tills sharing one id would collide on every topic and evict each other from the broker, since a second connection with the same client id kicks the first off. @@ -79,28 +85,28 @@ second connection with the same client id kicks the first off. dark" board possible, and it is the only way to tell *closed for the night* from *unplugged*. -### Running this on NATS +### Running this on Mosquitto -The MQTT gateway maps `/` to `.`, so the topics above arrive as subjects and a -JetStream consumer binds to them directly: +The deployed broker is Eclipse Mosquitto 2.1.2. A consumer binds to the topics +above directly, using `+` as the single-level wildcard: -| Purpose | Subject | +| Purpose | Filter | |---|---| -| Every till's bills | `pos.*.*.order` | -| Every till's presence | `pos.*.*.status` | -| One store's bills | `pos.store-01.*.order` | -| Ack back to one till | `pos.store-01.T4A9.ack` | +| Every till's bills | `nearle/pos/+/+/order` | +| Every till's heartbeat | `nearle/pos/+/+/health` | +| One shop's bills | `nearle/pos/12/+/order` | +| Ack back to one till | `nearle/pos/12/T4A9/ack` | -`SyncConfig.asNatsSubject()` does the translation, so a consumer's subject can -be read off the terminal rather than guessed. +Two things to get right: -Two things to get right on the NATS side: - -- **The stream must be durable and file-backed.** A memory stream loses a shop's - bills on a server restart, and the till has already been told they landed. - **Publish the ack from the consumer, after the database commit** — not from an - ingest handler that has merely queued the work. The ack is the terminal's - only evidence, and it deletes its copy seven days later on the strength of it. + ingest handler that has merely queued the work. The ack is the terminal's only + evidence, and it deletes its copy seven days later on the strength of it. +- **Do not treat the broker as durable storage.** Mosquitto's default + `max_queued_messages` is 1000 and its `autosave_interval` is 30 minutes, so a + long outage or a hard kill can drop queued messages. Nothing is lost, because + an undelivered batch is simply never acked and the till sends it again — but + only as long as nobody acknowledges on the broker's behalf. ### Fleet presence @@ -127,7 +133,7 @@ so. ## Payloads -**Uplink** — `pos/{store}/{terminal}/order` +**Uplink** — `nearle/pos/{loc}/{terminal}/order` ```json { @@ -140,7 +146,7 @@ so. } ``` -**Ack** — `pos/{store}/{terminal}/ack`. Must echo `batch_id`; anything else is +**Ack** — `nearle/pos/{loc}/{terminal}/ack`. Must echo `batch_id`; anything else is ignored as belonging to a batch the terminal is no longer waiting on. ```json @@ -160,7 +166,7 @@ the same bills. ### Shopper registrations -**Uplink** — `pos/{store}/{terminal}/customer`, or `POST {base}/customers`. +**Uplink** — `nearle/pos/{loc}/{terminal}/customer`, or `POST {base}/customers`. Acked on the same topic and by the same rules: only ids you name are marked sent. @@ -269,7 +275,7 @@ Dates are ISO 8601 or epoch milliseconds; both are accepted. ### Pushing a change mid-day -Publish to `pos/{store}/catalogue` (retained) and every terminal in the shop +Publish to `nearle/pos/{loc}/catalogue` (retained) and every terminal in the shop pulls immediately instead of waiting for tomorrow morning. The message body is only a nudge — the catalogue itself still comes over HTTP, because a broker is the wrong shape for tens of thousands of rows. diff --git a/lib/core/config/sync_config.dart b/lib/core/config/sync_config.dart index e3dc676..fb13417 100644 --- a/lib/core/config/sync_config.dart +++ b/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({ diff --git a/lib/data/remote/mqtt_order_transport.dart b/lib/data/remote/mqtt_order_transport.dart index f7ae934..1be9935 100644 --- a/lib/data/remote/mqtt_order_transport.dart +++ b/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 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 diff --git a/lib/data/repositories/sync_repository_impl.dart b/lib/data/repositories/sync_repository_impl.dart index 3e569dd..df4772c 100644 --- a/lib/data/repositories/sync_repository_impl.dart +++ b/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, diff --git a/lib/data/sync/health_reporter.dart b/lib/data/sync/health_reporter.dart new file mode 100644 index 0000000..f4891b3 --- /dev/null +++ b/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> 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 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 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 = { + '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 _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> _collectDeviceState() async { + final out = {}; + + 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 _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 dispose() async { + _stopped = true; + _timer?.cancel(); + } +} diff --git a/lib/presentation/sync/providers/sync_controller.dart b/lib/presentation/sync/providers/sync_controller.dart index 042c907..4029777 100644 --- a/lib/presentation/sync/providers/sync_controller.dart +++ b/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((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(); } }); diff --git a/test/unit/fleet_identity_test.dart b/test/unit/fleet_identity_test.dart index 6721419..dc8014a 100644 --- a/test/unit/fleet_identity_test.dart +++ b/test/unit/fleet_identity_test.dart @@ -143,11 +143,15 @@ void main() { const config = SyncConfig(storeId: 'store-01', terminalId: 'TA1B2'); expect( SyncConfig.asNatsSubject(config.orderTopic), - 'pos.store-01.TA1B2.order', + 'nearle.pos.store-01.TA1B2.order', ); expect( SyncConfig.asNatsSubject(config.statusTopic), - 'pos.store-01.TA1B2.status', + 'nearle.pos.store-01.TA1B2.status', + ); + expect( + SyncConfig.asNatsSubject(config.healthTopic), + 'nearle.pos.store-01.TA1B2.health', ); }); }); diff --git a/test/unit/transport_test.dart b/test/unit/transport_test.dart index 17c636a..16dd3b5 100644 --- a/test/unit/transport_test.dart +++ b/test/unit/transport_test.dart @@ -29,10 +29,11 @@ void main() { test('topics are namespaced per store and per terminal', () { // Two stores sharing one broker must never see each other's bills. - expect(config.orderTopic, 'pos/store-9/TERM-04/order'); - expect(config.ackTopic, 'pos/store-9/TERM-04/ack'); - expect(config.statusTopic, 'pos/store-9/TERM-04/status'); - expect(config.catalogueTopic, 'pos/store-9/catalogue'); + 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 {