From 0a49323858fcefa49645c88d8d5d18304b0d072a Mon Sep 17 00:00:00 2001 From: Suriya Date: Sat, 1 Aug 2026 10:57:29 +0530 Subject: [PATCH] Drain bills to the back office automatically, over MQTT or HTTP Turns the orders table into a queue that empties itself. Bills were only uploaded when a cashier pressed Sync at end of day; a till that was never pressed held a day's takings indefinitely. Drain engine (lib/data/sync/sync_engine.dart) - Triggers on sale committed, network regained, 5-minute poll, head-office request, and the manual button. - Single flight: a busy till firing a trigger per sale would otherwise have several passes reading the same pending rows and send every bill twice. A trigger arriving mid-drain is queued and replayed, so nothing is dropped. - Exponential backoff with +/-20% jitter to a 5-minute ceiling. The jitter matters: a store's terminals all fail at the same instant when the line drops, and would retry in lockstep without it. - Halts rather than loops on a failure retrying cannot fix (bad credential, refused batch). Pressing Sync clears the halt. Transports (lib/data/remote/) - OrderTransport interface; MQTT, HTTP and simulated implementations. The repository does not know which is in use. - MQTT: QoS 1 uplink, application-level ACK correlated by batch_id on a return topic, retained Last Will for terminal-offline detection, downlink for catalogue pushes and remote sync requests. - A broker PUBACK is never treated as acceptance. It means the broker holds the bytes, not that the ledger took the sale. Only ids the back office names are marked synced; silence leaves a bill pending. - HTTP carries a stable idempotency key across retries of the same bills. Retention - Accepted bills are kept 7 days instead of deleted, so a batch the back office later loses can be re-sent in full. Purged after that; archived totals stay forever. - forBusinessDate now reads pending rows only. A retained bill exists in both the orders table and day_archive, and summing both would overstate the day. Fixes found while building this - SyncEngine._refreshPending wrote state.copyWith(pending: await ...). Dart evaluates the receiver before the awaited argument, so a connectivity drop during the wait was silently overwritten by the stale snapshot. Caught by the first run of the new engine tests. - PrinterSettingsController wrote state after four awaits with no mounted check, throwing "used after dispose" when Settings was left mid-load. This was pre-existing and reached the cashier as a red screen. Also - Header pill now reports real sync state: LIVE / n QUEUED / SYNCING / SYNC HALTED, with an explanation of where the bills are. - Settings shows the route, last upload, next retry and retention window. - docs/sync-contract.md states what the back office must implement, including the idempotency requirement that at-least-once delivery makes mandatory. Tests: 90 -> 129 passing. New coverage for backoff shape and jitter band, single flight, halting, ACK correlation and partial acceptance, at-least-once duplicate handling, retention and purge, and no double-counting after a sync. Suite run six times clean. Not addressed: bills already synced by an older build went up overstated and still need server-side reconciliation. Broker credentials have no Settings editor yet. Co-Authored-By: Claude Opus 5 --- docs/sync-contract.md | 125 +++++ lib/app/app.dart | 5 + lib/app/providers.dart | 71 ++- lib/core/config/sync_config.dart | 123 +++++ lib/core/services/connectivity_service.dart | 76 +++ .../datasources/remote_catalogue_source.dart | 26 -- lib/data/local/order_dao.dart | 67 ++- lib/data/remote/http_order_transport.dart | 154 +++++++ lib/data/remote/mqtt_order_transport.dart | 332 +++++++++++++ lib/data/remote/order_transport.dart | 99 ++++ .../remote/simulated_order_transport.dart | 62 +++ .../repositories/sync_repository_impl.dart | 116 +++-- lib/data/sync/sync_engine.dart | 360 +++++++++++++++ lib/domain/repositories/sync_repository.dart | 29 +- .../modules/providers/printer_settings.dart | 23 +- .../modules/screens/settings_view.dart | 42 +- .../payment/providers/payment_controller.dart | 6 + lib/presentation/pos/widgets/page_header.dart | 73 ++- .../sync/providers/sync_controller.dart | 35 +- macos/Flutter/GeneratedPluginRegistrant.swift | 2 + pubspec.lock | 50 +- pubspec.yaml | 3 + test/unit/persistence_test.dart | 11 +- test/unit/retention_test.dart | 356 ++++++++++++++ test/unit/sync_engine_test.dart | 436 ++++++++++++++++++ test/unit/transport_test.dart | 276 +++++++++++ test/widget/app_smoke_test.dart | 5 + .../flutter/generated_plugin_registrant.cc | 3 + windows/flutter/generated_plugins.cmake | 1 + 29 files changed, 2855 insertions(+), 112 deletions(-) create mode 100644 docs/sync-contract.md create mode 100644 lib/core/config/sync_config.dart create mode 100644 lib/core/services/connectivity_service.dart create mode 100644 lib/data/remote/http_order_transport.dart create mode 100644 lib/data/remote/mqtt_order_transport.dart create mode 100644 lib/data/remote/order_transport.dart create mode 100644 lib/data/remote/simulated_order_transport.dart create mode 100644 lib/data/sync/sync_engine.dart create mode 100644 test/unit/retention_test.dart create mode 100644 test/unit/sync_engine_test.dart create mode 100644 test/unit/transport_test.dart diff --git a/docs/sync-contract.md b/docs/sync-contract.md new file mode 100644 index 0000000..cfbfa2e --- /dev/null +++ b/docs/sync-contract.md @@ -0,0 +1,125 @@ +# Terminal ↔ back office sync contract + +What the till guarantees, and what the back office must do to hold up its end. +Everything here is enforced by tests in `test/unit/retention_test.dart`, +`test/unit/transport_test.dart` and `test/unit/sync_engine_test.dart`. + +## The shape + +``` +Customer pays + │ + ▼ +One SQLite transaction: bill + stock + loyalty ← never blocked on network + │ + ▼ +orders row lands at sync_status = 0 ← this table IS the outbox + │ + ▼ +SyncEngine drains on: sale committed · network back · 5-min poll · head office + │ asked · cashier pressed Sync + ▼ +Transport publishes a batch + │ + ▼ +Back office commits and names the ids it took + │ + ▼ +Those rows → sync_status = 1, folded into day_archive, kept 7 days, then purged +``` + +Anything the back office does not name stays at 0 and goes again. + +## Non-negotiables + +**1. Only an application acknowledgement counts.** +A broker PUBACK means "I hold these bytes". It is not evidence the ledger +accepted anything, and the terminal never treats it as such. The back office +must answer on the ack topic naming the order ids it committed. + +**2. Silence is not acceptance.** +A `200 OK` with an empty body, or an ack with no `accepted` array, marks *zero* +bills synced. The terminal will send them again rather than guess. + +**3. Delivery is at-least-once, so the back office must be idempotent.** +QoS 1 re-delivers, and a lost ack makes the terminal re-send the whole batch. +Every `order.id` is a UUID minted at the till. Put a unique index on it and +upsert. Without this you will double-count a day's takings the first time a +shop's line wobbles. + +**4. A refusal is final, a failure is not.** +Naming an id in `rejected` halts the terminal's drain — it will not retry the +same bytes, and a person has to press Sync. Use it for "this bill is wrong" +(unknown product, duplicate invoice). For "I am having a bad minute", drop the +connection or return 5xx instead, and the terminal will back off and retry. + +## MQTT topics + +| Topic | Direction | QoS | Retained | +|---|---|---|---| +| `pos/{store}/{terminal}/order` | 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** | + +`status` is also the Last Will. If a till loses power the broker publishes +`{"state":"offline"}` on its behalf — that is what makes a "which tills are +dark" board possible, and it is the only way to tell *closed for the night* +from *unplugged*. + +## Payloads + +**Uplink** — `pos/{store}/{terminal}/order` + +```json +{ + "schema": 1, + "batch_id": "9f1c…", + "store_id": "store-01", + "terminal_id": "TERM-01", + "sent_at": "2026-08-01T14:22:05.123Z", + "orders": [ { "id": "…", "invoice_number": "…", "items": [ … ] } ] +} +``` + +**Ack** — `pos/{store}/{terminal}/ack`. Must echo `batch_id`; anything else is +ignored as belonging to a batch the terminal is no longer waiting on. + +```json +{ + "batch_id": "9f1c…", + "accepted": ["order-uuid-a", "order-uuid-b"], + "rejected": { "order-uuid-c": "duplicate invoice number" } +} +``` + +No ack within `SyncConfig.ackTimeout` (20s default) → the outcome is unknown, +nothing is marked synced, and the batch goes again. + +**HTTP equivalent** — `POST {base}/orders`, same body, ack shape as the 200 +response. Carries an `idempotency-key` header that is stable across retries of +the same bills. + +## Retention on the terminal + +Accepted bills stay for 7 days (`OrderDao.retentionWindow`) so a batch the +back office later loses can be re-sent in full. After that only the archived +day totals survive, and a lost bill's line items are gone for good. + +While a bill is retained it exists in two places — its own row and +`day_archive`. `forBusinessDate` therefore reads **pending rows only**; without +that filter every synced bill would be counted twice and the shift report would +overstate the day. + +## What is deliberately not built + +- **Downlink beyond catalogue-changed and sync-requested.** The plumbing routes + unknown commands to the events log rather than dropping them, so adding one + is a server change plus a case arm. +- **Broker credentials in Settings.** `SyncConfig` carries them and Settings + displays the route, but there is no editor yet — a store is pointed at a + broker in code or by overriding `syncConfigProvider`. +- **Historical correction.** Bills already synced by an older build went up + with an overstated total. Nothing here fixes that; it needs a server-side + reconciliation against `bill_discount`. diff --git a/lib/app/app.dart b/lib/app/app.dart index 932201e..4c7eb3a 100644 --- a/lib/app/app.dart +++ b/lib/app/app.dart @@ -4,12 +4,17 @@ import 'package:flutter_riverpod/flutter_riverpod.dart'; import '../core/constants/app_constants.dart'; import '../core/router/app_router.dart'; import '../core/theme/app_theme.dart'; +import '../presentation/sync/providers/sync_controller.dart'; class NearlePosApp extends ConsumerWidget { const NearlePosApp({super.key}); @override Widget build(BuildContext context, WidgetRef ref) { + // Watched, not awaited: the till opens immediately and the queue drains + // behind it. Nothing on screen depends on this having finished. + ref.watch(syncBootstrapProvider); + return MaterialApp.router( title: AppConstants.appName, debugShowCheckedModeBanner: false, diff --git a/lib/app/providers.dart b/lib/app/providers.dart index 503d6d0..ca60194 100644 --- a/lib/app/providers.dart +++ b/lib/app/providers.dart @@ -1,13 +1,20 @@ import 'package:flutter_riverpod/flutter_riverpod.dart'; +import '../core/config/sync_config.dart'; +import '../core/services/connectivity_service.dart'; import '../core/services/receipt_service.dart'; import '../core/services/sound_service.dart'; import '../data/datasources/local_store.dart'; import '../data/repositories/customer_repository_impl.dart'; import '../data/repositories/product_repository_impl.dart'; import '../data/datasources/remote_catalogue_source.dart'; +import '../data/remote/http_order_transport.dart'; +import '../data/remote/mqtt_order_transport.dart'; +import '../data/remote/order_transport.dart'; +import '../data/remote/simulated_order_transport.dart'; import '../data/repositories/sync_repository_impl.dart'; import '../data/repositories/transaction_repository_impl.dart'; +import '../data/sync/sync_engine.dart'; import '../domain/repositories/customer_repository.dart'; import '../domain/repositories/product_repository.dart'; import '../domain/repositories/sync_repository.dart'; @@ -44,20 +51,72 @@ final remoteCatalogueProvider = Provider( ), ); -final remoteOrderSinkProvider = Provider( - (ref) => RemoteOrderSink( - isOffline: () => ref.read(simulateOfflineProvider), - ), -); +// ------------------------------------------------------------------- Sync +/// How this terminal reaches the back office. +/// +/// Defaults to the simulated route so a fresh install is usable with no broker +/// and no endpoint; Settings re-points it. +final syncConfigProvider = StateProvider((ref) => const SyncConfig()); + +/// Real network state, folded with the Settings offline switch. +final connectivityServiceProvider = Provider((ref) { + final service = ConnectivityService( + isSimulatedOffline: () => ref.read(simulateOfflineProvider), + ); + ref.onDispose(service.dispose); + return service; +}); + +/// The wire itself. Rebuilt when the configuration changes, and the old one is +/// closed so a re-pointed terminal does not keep a stale broker session open. +final orderTransportProvider = Provider((ref) { + final config = ref.watch(syncConfigProvider); + + final transport = switch (config.transport) { + TransportKind.mqtt => MqttOrderTransport(config: config), + TransportKind.http => HttpOrderTransport(config: config), + TransportKind.simulated => SimulatedOrderTransport( + isOffline: () => ref.read(simulateOfflineProvider), + ), + }; + + ref.onDispose(transport.dispose); + return transport; +}); final syncRepositoryProvider = Provider( (ref) => SyncRepositoryImpl( ref.watch(localStoreProvider), ref.watch(remoteCatalogueProvider), - ref.watch(remoteOrderSinkProvider), + ref.watch(orderTransportProvider), + batchSize: ref.watch(syncConfigProvider).batchSize, ), ); +/// Decides when bills are uploaded. Started once, by the app shell. +final syncEngineProvider = Provider((ref) { + final transport = ref.watch(orderTransportProvider); + + final engine = SyncEngine( + repository: ref.watch(syncRepositoryProvider), + connectivity: ref.watch(connectivityServiceProvider).onlineChanges, + downlink: transport.downlink, + onCatalogueChanged: () async { + await ref.read(syncRepositoryProvider).importCatalogue(); + ref.invalidate(localStoreProvider); + }, + ); + + ref.onDispose(engine.dispose); + return engine; +}); + +/// Live engine state for the header pill and the events screen. +final syncEngineStateProvider = StreamProvider((ref) { + final engine = ref.watch(syncEngineProvider); + return engine.states.map((s) => s); +}); + // ------------------------------------------------------------- Use cases final checkoutSaleProvider = Provider( (ref) => CheckoutSale( diff --git a/lib/core/config/sync_config.dart b/lib/core/config/sync_config.dart new file mode 100644 index 0000000..f95196c --- /dev/null +++ b/lib/core/config/sync_config.dart @@ -0,0 +1,123 @@ +/// Which route completed bills take to the back office. +enum TransportKind { + /// No back office wired up — bills queue and drain against a local stub. + simulated, + + /// Plain request/response upload. Easiest to debug: curl reproduces it and + /// failures come back as status codes. + http, + + /// Persistent connection. Costs a broker, and buys the downlink: head office + /// can push a price change or ask a terminal to sync without waiting for it + /// to ask first. + mqtt, +} + +/// Everything the terminal needs to reach the back office. +/// +/// One object rather than scattered constants so a store can be re-pointed at +/// a different broker without a rebuild, and so tests can construct a whole +/// configuration inline. +class SyncConfig { + const SyncConfig({ + this.transport = TransportKind.simulated, + this.storeId = 'store-01', + this.terminalId = 'TERM-01', + this.brokerHost = '', + this.brokerPort = 8883, + this.useTls = true, + this.username, + this.password, + this.httpBaseUrl = '', + this.apiKey, + this.ackTimeout = const Duration(seconds: 20), + this.batchSize = 50, + }); + + final TransportKind transport; + + /// Namespaces every topic. Two stores on one broker must never collide. + final String storeId; + final String terminalId; + + final String brokerHost; + final int brokerPort; + final bool useTls; + final String? username; + final String? password; + + final String httpBaseUrl; + final String? apiKey; + + /// How long to wait for the back office to confirm a batch before treating + /// the outcome as unknown and leaving every row pending. + /// + /// Generous on purpose: a timeout that fires while the server is committing + /// produces a duplicate send, which is safe only because the back office + /// keys on order id — but it is still wasted traffic on a slow shop line. + final Duration ackTimeout; + + /// Bills per publish. Small enough to stay well inside broker message limits + /// on a day that has built up a long backlog. + final int batchSize; + + bool get isConfigured => switch (transport) { + TransportKind.simulated => true, + TransportKind.http => httpBaseUrl.isNotEmpty, + TransportKind.mqtt => brokerHost.isNotEmpty, + }; + + // ------------------------------------------------------------------ Topics + String get _base => 'pos/$storeId/$terminalId'; + + /// Uplink. Completed bills, QoS 1. + String get orderTopic => '$_base/order'; + + /// The back office's answer, naming the ids it committed. Subscribed at + /// QoS 1: losing an ack means re-sending bills that are already banked. + String get ackTopic => '$_base/ack'; + + /// Retained, and set as the will message. A terminal that loses power stops + /// refreshing it and the broker publishes `offline` on its behalf, which is + /// what makes a head-office "which tills are dark" board possible. + String get statusTopic => '$_base/status'; + + /// Store-wide downlink: catalogue changes land here for every terminal. + String get catalogueTopic => 'pos/$storeId/catalogue'; + + /// Addressed to this terminal alone. + String get commandTopic => '$_base/command'; + + /// Stable across restarts so the broker can resume a session and redeliver + /// anything in flight, rather than treating each launch as a new client. + String get clientId => 'pos-$storeId-$terminalId'; + + SyncConfig copyWith({ + TransportKind? transport, + String? storeId, + String? terminalId, + String? brokerHost, + int? brokerPort, + bool? useTls, + String? username, + String? password, + String? httpBaseUrl, + String? apiKey, + Duration? ackTimeout, + int? batchSize, + }) => + SyncConfig( + transport: transport ?? this.transport, + storeId: storeId ?? this.storeId, + terminalId: terminalId ?? this.terminalId, + brokerHost: brokerHost ?? this.brokerHost, + brokerPort: brokerPort ?? this.brokerPort, + useTls: useTls ?? this.useTls, + username: username ?? this.username, + password: password ?? this.password, + httpBaseUrl: httpBaseUrl ?? this.httpBaseUrl, + apiKey: apiKey ?? this.apiKey, + ackTimeout: ackTimeout ?? this.ackTimeout, + batchSize: batchSize ?? this.batchSize, + ); +} diff --git a/lib/core/services/connectivity_service.dart b/lib/core/services/connectivity_service.dart new file mode 100644 index 0000000..712f294 --- /dev/null +++ b/lib/core/services/connectivity_service.dart @@ -0,0 +1,76 @@ +import 'dart:async'; + +import 'package:connectivity_plus/connectivity_plus.dart'; + +/// Tells the terminal when it is worth trying the network. +/// +/// This is a *hint*, not proof. The platform reports whether an interface is +/// up, which on shop wifi is routinely true while the line itself is dead. So +/// nothing here decides that a sync succeeded — only the transport's answer +/// does. What this buys is the moment to try: the difference between a bill +/// going up the second the router comes back and it waiting for the next poll. +/// +/// The Settings "Simulate offline" switch is folded in here so there is one +/// answer to "are we online", rather than a real state and a demo state that +/// can disagree. +class ConnectivityService { + ConnectivityService({ + Connectivity? connectivity, + bool Function()? isSimulatedOffline, + }) : _connectivity = connectivity ?? Connectivity(), + _isSimulatedOffline = isSimulatedOffline ?? (() => false); + + final Connectivity _connectivity; + final bool Function() _isSimulatedOffline; + + final _controller = StreamController.broadcast(); + StreamSubscription>? _subscription; + + bool _hasInterface = true; + bool _started = false; + + /// Fires only when the answer changes, so a subscriber can treat every event + /// as an edge. + Stream get onlineChanges => _controller.stream; + + bool get isOnline => _hasInterface && !_isSimulatedOffline(); + + Future start() async { + if (_started) return; + _started = true; + + try { + _hasInterface = _hasAny(await _connectivity.checkConnectivity()); + } on Exception { + // No platform channel — a test host, or a desktop build without the + // plugin registered. Assume online and let the transport be the judge; + // refusing to try would be worse than trying and failing. + _hasInterface = true; + } + + try { + _subscription = _connectivity.onConnectivityChanged.listen((results) { + _update(_hasAny(results)); + }); + } on Exception { + // As above: without the stream the periodic poll still drains the queue. + } + } + + /// Re-evaluates after the Settings switch is flipped. + void refresh() => _update(_hasInterface); + + void _update(bool hasInterface) { + final was = isOnline; + _hasInterface = hasInterface; + if (isOnline != was) _controller.add(isOnline); + } + + static bool _hasAny(List results) => + results.any((r) => r != ConnectivityResult.none); + + Future dispose() async { + await _subscription?.cancel(); + await _controller.close(); + } +} diff --git a/lib/data/datasources/remote_catalogue_source.dart b/lib/data/datasources/remote_catalogue_source.dart index 54a1b44..c6ff5d2 100644 --- a/lib/data/datasources/remote_catalogue_source.dart +++ b/lib/data/datasources/remote_catalogue_source.dart @@ -84,29 +84,3 @@ class RemoteCatalogueSource { ); } } - -/// Stands in for the back-office order intake API. -class RemoteOrderSink { - RemoteOrderSink({required this.isOffline}); - - final bool Function() isOffline; - - /// Uploads a batch of orders and returns the ids the server accepted. - /// - /// Throws on transport failure so the caller leaves every row at - /// `sync_status = 0` rather than marking anything sent. - Future> pushOrders(List> orders) async { - await Future.delayed( - Duration(milliseconds: 400 + orders.length * 60), - ); - - if (isOffline()) { - throw const CatalogueSyncException( - 'Simulate offline is ON in Settings, so the upload was failed on ' - 'purpose. Every bill is still stored on this terminal.', - ); - } - - return orders.map((o) => o['id']! as String).toList(); - } -} diff --git a/lib/data/local/order_dao.dart b/lib/data/local/order_dao.dart index d9643a2..1890f73 100644 --- a/lib/data/local/order_dao.dart +++ b/lib/data/local/order_dao.dart @@ -11,9 +11,13 @@ import 'catalogue_dao.dart'; /// Persists bills. /// -/// Every completed sale lands here with `sync_status = 0`. The end-of-day -/// upload selects those rows, sends them, and flips the accepted ones to 1. -/// Nothing is ever deleted as part of syncing. +/// Every completed sale lands here with `sync_status = 0`, which makes this +/// table the terminal's outbox: the drain engine selects those rows, sends +/// them, and flips the ones the back office confirmed to 1. +/// +/// Confirmed rows are kept for [retentionWindow] so a batch the server later +/// loses can be re-sent in full, then retired by [purgeSyncedBefore]. Syncing +/// itself never deletes anything. class OrderDao { const OrderDao(this._db); @@ -22,6 +26,9 @@ class OrderDao { static const int pending = 0; static const int synced = 1; + /// How long an accepted bill stays re-sendable on the terminal. + static const Duration retentionWindow = Duration(days: 7); + static String businessDateOf(DateTime dt) => '${dt.year.toString().padLeft(4, '0')}-' '${dt.month.toString().padLeft(2, '0')}-' @@ -132,20 +139,26 @@ class OrderDao { Future> recent({int limit = 100}) => _query(orderBy: 'created_at DESC', limit: limit); - /// Bills for a day, optionally narrowed to one operator. + /// Bills for a day that have *not* yet been accepted by the server. /// /// A shift report that is settled against a till has to cover exactly the /// bills that cashier rang, not everything the terminal did that day. + /// + /// Restricted to pending rows on purpose. The moment a bill is accepted its + /// figures are folded into [Tables.dayArchive], and the report adds the two + /// together — so an accepted row still sitting here during its retention + /// window would be counted twice and inflate the day's takings. Future> forBusinessDate( DateTime day, { String? cashierName, }) => _query( where: cashierName == null - ? 'business_date = ?' - : 'business_date = ? AND cashier_name = ?', + ? 'business_date = ? AND sync_status = ?' + : 'business_date = ? AND sync_status = ? AND cashier_name = ?', whereArgs: [ businessDateOf(day), + pending, if (cashierName != null) cashierName, ], ); @@ -241,14 +254,19 @@ class OrderDao { ); // ----------------------------------------------------------------- Sync - /// Folds accepted orders into the day archive, then deletes them. + /// Folds accepted orders into the day archive and marks them synced. /// - /// Once the server holds a bill the terminal has no reason to keep it, so - /// the rows go. Their figures are added to [Tables.dayArchive] first, so the - /// shift totals a cashier sees do not collapse after a mid-shift sync. - /// Both steps run in one transaction: if the delete fails the archive is - /// rolled back with it, and nothing is counted twice. - Future archiveAndDelete(List orders) async { + /// Their figures are added to [Tables.dayArchive] so the shift totals a + /// cashier sees do not collapse after a mid-shift sync, and the rows + /// themselves are kept — at `sync_status = 1` — until [purgeSyncedBefore] + /// retires them. Keeping them buys a recovery window: if the back office + /// loses a batch, the full bills are still on the terminal and can be sent + /// again. Once deleted only the archived totals survive, and a lost bill's + /// line items are gone for good. + /// + /// Both steps run in one transaction: if the status flip fails the archive + /// rolls back with it, and nothing is counted twice. + Future archiveAccepted(List orders) async { if (orders.isEmpty) return; await _db.transaction((txn) async { @@ -331,17 +349,30 @@ class OrderDao { ); } - // order_items goes with it via ON DELETE CASCADE. final ids = orders.map((o) => o.id).toList(); final placeholders = List.filled(ids.length, '?').join(','); - await txn.delete( - Tables.orders, - where: 'id IN ($placeholders)', - whereArgs: ids, + await txn.rawUpdate( + 'UPDATE ${Tables.orders} ' + 'SET sync_status = ?, synced_at = ?, sync_error = NULL ' + 'WHERE id IN ($placeholders)', + [synced, DateTime.now().millisecondsSinceEpoch, ...ids], ); }); } + /// Retires bills the server took delivery of more than [retention] ago. + /// + /// Their archived totals are untouched and stay forever — this drops only the + /// re-sendable copy once the recovery window has closed, so a terminal that + /// trades for years does not carry every bill it has ever rung. + /// + /// Returns how many rows went. `order_items` follows via ON DELETE CASCADE. + Future purgeSyncedBefore(DateTime cutoff) => _db.delete( + Tables.orders, + where: 'sync_status = ? AND synced_at IS NOT NULL AND synced_at < ?', + whereArgs: [synced, cutoff.millisecondsSinceEpoch], + ); + /// Archived figures for a business day, one row per cashier. /// /// Empty when nothing has synced yet. Pass [cashierName] to scope it to a diff --git a/lib/data/remote/http_order_transport.dart b/lib/data/remote/http_order_transport.dart new file mode 100644 index 0000000..cbc1943 --- /dev/null +++ b/lib/data/remote/http_order_transport.dart @@ -0,0 +1,154 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:http/http.dart' as http; + +import '../../core/config/sync_config.dart'; +import 'order_transport.dart'; + +/// Ships bills over plain HTTP. +/// +/// No downlink and no persistent connection — but a failure is a status code, +/// a request is reproducible with curl, and there is no broker to run. Worth +/// having as the route to bring up first, and as the fallback when a broker is +/// unreachable but the internet is not. +/// +/// The endpoint must answer with the ids it committed: +/// +/// ```json +/// { "accepted": ["", ...], "rejected": { "": "reason" } } +/// ``` +/// +/// A bare `200 OK` is not treated as success for any bill. Silence about which +/// rows landed is not the same as landing them, and guessing here would retire +/// a day's takings on an empty response. +class HttpOrderTransport implements OrderTransport { + HttpOrderTransport({required this.config, http.Client? client}) + : _client = client ?? http.Client(); + + final SyncConfig config; + final http.Client _client; + + final _connection = StreamController.broadcast(); + + bool _reachable = true; + + @override + String get label => 'HTTP ${config.httpBaseUrl}'; + + @override + bool get isConnected => _reachable; + + /// Nothing to receive: HTTP is a route out, not in. + @override + Stream get downlink => const Stream.empty(); + + @override + Stream get connectionState => _connection.stream; + + @override + Future connect() async {} + + @override + Future pushOrders(List> orders) async { + if (orders.isEmpty) return const PushReceipt(accepted: []); + + if (config.httpBaseUrl.isEmpty) { + throw const TransportException( + 'No back-office URL configured for this terminal.', + retryable: false, + ); + } + + final uri = Uri.parse('${config.httpBaseUrl}/orders'); + + http.Response response; + try { + response = await _client + .post( + uri, + headers: { + 'content-type': 'application/json', + if (config.apiKey != null) + 'authorization': 'Bearer ${config.apiKey}', + // Lets the endpoint collapse a retried batch server-side rather + // than relying on every order id being checked individually. + 'idempotency-key': _batchKey(orders), + }, + body: jsonEncode({ + 'schema': 1, + 'store_id': config.storeId, + 'terminal_id': config.terminalId, + 'sent_at': DateTime.now().toIso8601String(), + 'orders': orders, + }), + ) + .timeout(config.ackTimeout); + } on Exception catch (e) { + _setReachable(false); + throw TransportException('Upload failed: $e'); + } + + if (response.statusCode == 401 || response.statusCode == 403) { + _setReachable(false); + throw TransportException( + 'The back office rejected this terminal\'s credentials ' + '(${response.statusCode}). Bills are safe locally, but sync will keep ' + 'failing until the terminal is re-authorised.', + retryable: false, + ); + } + + if (response.statusCode >= 300) { + _setReachable(false); + throw TransportException( + 'Back office returned ${response.statusCode}: ' + '${_trim(response.body)}', + ); + } + + _setReachable(true); + + Map body; + try { + body = jsonDecode(response.body) as Map; + } on Exception { + throw TransportException( + 'Back office answered 200 with a body this terminal could not read, ' + 'so no bill was marked synced: ${_trim(response.body)}', + ); + } + + final accepted = (body['accepted'] as List? ?? const []) + .whereType() + .toList(); + + final rejected = {}; + final raw = body['rejected']; + if (raw is Map) { + raw.forEach((k, v) => rejected['$k'] = '$v'); + } + + return PushReceipt(accepted: accepted, rejected: rejected); + } + + /// Stable for a given set of bills, so a retry after a timeout carries the + /// same key as the attempt that may already have landed. + String _batchKey(List> orders) => + orders.map((o) => o['id']).join('|').hashCode.toRadixString(16); + + void _setReachable(bool value) { + if (_reachable == value) return; + _reachable = value; + _connection.add(value); + } + + static String _trim(String body) => + body.length <= 200 ? body : '${body.substring(0, 200)}…'; + + @override + Future dispose() async { + _client.close(); + await _connection.close(); + } +} diff --git a/lib/data/remote/mqtt_order_transport.dart b/lib/data/remote/mqtt_order_transport.dart new file mode 100644 index 0000000..98f8fd6 --- /dev/null +++ b/lib/data/remote/mqtt_order_transport.dart @@ -0,0 +1,332 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:flutter/foundation.dart'; +import 'package:mqtt_client/mqtt_client.dart'; +import 'package:mqtt_client/mqtt_server_client.dart'; +import 'package:uuid/uuid.dart'; + +import '../../core/config/sync_config.dart'; +import 'order_transport.dart'; + +/// Ships bills over MQTT and listens for what head office pushes back. +/// +/// ### Why the broker's acknowledgement is not enough +/// +/// QoS 1 gives a PUBACK from the *broker*, meaning "I hold these bytes". It +/// says nothing about whether the back office parsed the batch, or whether the +/// ledger accepted it. Marking bills synced on PUBACK would retire a day's +/// takings on the word of a message queue. +/// +/// So every publish carries a `batch_id` and this waits for the back office to +/// answer on [SyncConfig.ackTopic] naming the ids it actually committed. No +/// answer means no sync, and the bills are sent again. +/// +/// ### Duplicates are expected +/// +/// QoS 1 is at-least-once, and a batch whose ack is lost will be re-sent in +/// full. The back office must key on `order.id` and upsert. Every id is a UUID +/// minted at the till, so this costs the server one unique index. +class MqttOrderTransport implements OrderTransport { + MqttOrderTransport({ + required this.config, + MqttClient Function(SyncConfig)? clientFactory, + }) : _clientFactory = clientFactory ?? _defaultClient; + + final SyncConfig config; + final MqttClient Function(SyncConfig) _clientFactory; + + static const _uuid = Uuid(); + + MqttClient? _client; + + /// Batches published but not yet answered, keyed by `batch_id`. + final _awaitingAck = >{}; + + final _downlink = StreamController.broadcast(); + final _connection = StreamController.broadcast(); + + /// Guards against two drains racing to open the same connection. + Future? _connecting; + + StreamSubscription>>? _updates; + + static MqttClient _defaultClient(SyncConfig config) { + final client = MqttServerClient.withPort( + config.brokerHost, + config.clientId, + config.brokerPort, + ); + client.secure = config.useTls; + // Well under the shortest NAT timeout a shop router is likely to impose; + // a connection that dies silently is worse than one that pings. + client.keepAlivePeriod = 20; + client.autoReconnect = true; + client.resubscribeOnAutoReconnect = true; + client.logging(on: false); + return client; + } + + @override + String get label => 'MQTT ${config.brokerHost}:${config.brokerPort}'; + + @override + bool get isConnected => + _client?.connectionStatus?.state == MqttConnectionState.connected; + + @override + Stream get downlink => _downlink.stream; + + @override + Stream get connectionState => _connection.stream; + + // ------------------------------------------------------------- Connection + @override + Future connect() { + if (isConnected) return Future.value(); + return _connecting ??= _doConnect().whenComplete(() => _connecting = null); + } + + Future _doConnect() async { + if (config.brokerHost.isEmpty) { + throw const TransportException( + 'No MQTT broker configured for this terminal.', + retryable: false, + ); + } + + final client = _client ??= _clientFactory(config); + + client.onDisconnected = () { + _connection.add(false); + // Nobody is coming to answer these. Failing them now returns the bills + // to pending immediately instead of holding the drain for the full ack + // timeout on a connection that is already gone. + _failAllAwaiting('Connection to the broker dropped.'); + }; + client.onConnected = () => _connection.add(true); + + // Retained, so head office sees this terminal's last known state even if + // its dashboard connects hours later. The broker publishes it on our + // behalf if the till loses power mid-shift — which is the only way to tell + // "closed for the night" from "unplugged". + client.connectionMessage = MqttConnectMessage() + .withClientIdentifier(config.clientId) + .withWillTopic(config.statusTopic) + .withWillMessage(jsonEncode({'state': 'offline'})) + .withWillQos(MqttQos.atLeastOnce) + .withWillRetain(); + + try { + await client.connect(config.username, config.password); + } on Exception catch (e) { + client.disconnect(); + throw TransportException('Could not reach the broker: $e'); + } + + if (client.connectionStatus?.state != MqttConnectionState.connected) { + final status = client.connectionStatus; + client.disconnect(); + throw TransportException( + 'Broker refused the connection: ' + '${status?.returnCode?.name ?? 'unknown'}', + // Bad credentials or an unauthorised client id will be refused just as + // firmly on the next attempt; backing off forever would only hide it. + retryable: status?.returnCode != MqttConnectReturnCode.notAuthorized, + ); + } + + client + ..subscribe(config.ackTopic, MqttQos.atLeastOnce) + ..subscribe(config.catalogueTopic, MqttQos.atLeastOnce) + ..subscribe(config.commandTopic, MqttQos.atLeastOnce); + + await _updates?.cancel(); + _updates = client.updates?.listen(_onUpdates); + + _publish( + config.statusTopic, + jsonEncode({ + 'state': 'online', + 'terminal_id': config.terminalId, + 'at': DateTime.now().toIso8601String(), + }), + retain: true, + ); + + _connection.add(true); + } + + void _onUpdates(List> events) { + for (final event in events) { + final message = event.payload; + if (message is! MqttPublishMessage) continue; + handleInbound( + event.topic, + MqttPublishPayload.bytesToStringAsString(message.payload.message), + ); + } + } + + // ----------------------------------------------------------------- Uplink + @override + Future pushOrders(List> orders) async { + if (orders.isEmpty) return const PushReceipt(accepted: []); + + await connect(); + + final batchId = _uuid.v4(); + final completer = Completer(); + _awaitingAck[batchId] = completer; + + try { + _publish( + config.orderTopic, + jsonEncode({ + 'schema': 1, + 'batch_id': batchId, + 'store_id': config.storeId, + 'terminal_id': config.terminalId, + 'sent_at': DateTime.now().toIso8601String(), + 'orders': orders, + }), + ); + + return await completer.future.timeout( + config.ackTimeout, + onTimeout: () => throw TransportException( + 'The back office did not confirm the batch within ' + '${config.ackTimeout.inSeconds}s. The bills are still on this ' + 'terminal and will be sent again.', + ), + ); + } finally { + _awaitingAck.remove(batchId); + } + } + + void _publish(String topic, String payload, {bool retain = false}) { + final builder = MqttClientPayloadBuilder()..addString(payload); + try { + _client!.publishMessage( + topic, + MqttQos.atLeastOnce, + builder.payload!, + retain: retain, + ); + } on Exception catch (e) { + throw TransportException('Publish to $topic failed: $e'); + } + } + + // --------------------------------------------------------------- Downlink + /// Registers a batch as awaiting its ack, without publishing one. + /// + /// Lets a test drive the correlation rules — which is where the logic that + /// decides whether a bill counts as banked actually lives — without standing + /// up a broker. + @visibleForTesting + Future awaitAck(String batchId) { + final completer = Completer(); + _awaitingAck[batchId] = completer; + return completer.future; + } + + /// Routes one inbound message. Separated from the client so the correlation + /// and parsing rules can be tested without a broker. + @visibleForTesting + void handleInbound(String topic, String payload) { + Map body; + try { + body = jsonDecode(payload) as Map; + } on FormatException { + // A malformed message must not take down the connection: the next one + // may be a perfectly good ack releasing a day's bills. + debugPrint('Discarded unparseable MQTT message on $topic'); + return; + } + + if (topic == config.ackTopic) { + _resolveAck(body); + return; + } + + if (topic == config.catalogueTopic) { + _downlink.add( + DownlinkMessage(kind: DownlinkKind.catalogueChanged, payload: body), + ); + return; + } + + if (topic == config.commandTopic) { + final command = body['command'] as String?; + _downlink.add( + DownlinkMessage( + kind: command == 'sync' + ? DownlinkKind.syncRequested + : DownlinkKind.unknown, + payload: body, + ), + ); + } + } + + void _resolveAck(Map body) { + final batchId = body['batch_id'] as String?; + if (batchId == null) return; + + // An ack for a batch we are no longer waiting on — we timed out, or the + // terminal restarted. Harmless: those bills are still pending and will go + // up again, and the back office keys on order id. + final completer = _awaitingAck.remove(batchId); + if (completer == null || completer.isCompleted) return; + + final accepted = (body['accepted'] as List? ?? const []) + .whereType() + .toList(); + + final rejected = {}; + final raw = body['rejected']; + if (raw is Map) { + raw.forEach((k, v) => rejected['$k'] = '$v'); + } else if (raw is List) { + // Tolerates a back office that sends bare ids with no reason. + for (final id in raw.whereType()) { + rejected[id] = 'Rejected by the back office'; + } + } + + completer.complete(PushReceipt(accepted: accepted, rejected: rejected)); + } + + void _failAllAwaiting(String reason) { + final waiting = List.of(_awaitingAck.values); + _awaitingAck.clear(); + for (final c in waiting) { + if (!c.isCompleted) c.completeError(TransportException(reason)); + } + } + + @override + Future dispose() async { + _failAllAwaiting('The terminal is shutting down.'); + await _updates?.cancel(); + + if (isConnected) { + // A clean goodbye, so head office does not see a till it thinks crashed. + try { + _publish( + config.statusTopic, + jsonEncode({'state': 'offline', 'clean': true}), + retain: true, + ); + } on TransportException { + // Already going down; nothing useful to do about it. + } + } + + _client?.disconnect(); + await _downlink.close(); + await _connection.close(); + } +} diff --git a/lib/data/remote/order_transport.dart b/lib/data/remote/order_transport.dart new file mode 100644 index 0000000..87a1566 --- /dev/null +++ b/lib/data/remote/order_transport.dart @@ -0,0 +1,99 @@ +import 'dart:async'; + +/// Raised when a batch could not be handed to the back office. +/// +/// Distinct from *rejection*: a transport failure means nobody knows whether +/// the bills arrived, so every row stays pending and is tried again. A +/// rejection means the back office looked at a bill and refused it, which +/// retrying will not fix. +class TransportException implements Exception { + const TransportException(this.message, {this.retryable = true}); + + final String message; + + /// Whether trying again could plausibly succeed. A dropped connection is + /// retryable; a rejected certificate or a bad credential is not, and the + /// engine should stop rather than hammer the broker. + final bool retryable; + + @override + String toString() => message; +} + +/// What the back office said about one batch. +/// +/// [accepted] is the contract that matters: only ids named here are marked +/// synced. Anything absent stays pending, whatever the transport reported at +/// its own layer. +class PushReceipt { + const PushReceipt({required this.accepted, this.rejected = const {}}); + + final List accepted; + + /// Order id → why the back office refused it. Retrying these unchanged will + /// fail again, so the engine surfaces them instead of looping. + final Map rejected; + + bool get isEmpty => accepted.isEmpty && rejected.isEmpty; +} + +/// Something the cloud pushed down to this terminal. +/// +/// Only a transport with a live connection can deliver these; request/response +/// transports expose an empty stream. +class DownlinkMessage { + const DownlinkMessage({required this.kind, this.payload = const {}}); + + final DownlinkKind kind; + final Map payload; +} + +enum DownlinkKind { + /// The catalogue changed at head office — re-import rather than wait for + /// tomorrow morning's pull. + catalogueChanged, + + /// Head office is asking this terminal to upload now. + syncRequested, + + /// Anything this build does not recognise. Kept rather than dropped so a + /// newer server talking to an older terminal is visible in the events log + /// instead of silently ignored. + unknown, +} + +/// How completed bills leave the terminal. +/// +/// The drain engine owns *when* to send and what to do when sending fails; +/// this owns only the wire. Swapping HTTP for MQTT is a change of +/// implementation here and nothing else. +abstract class OrderTransport { + /// Shown in the events log so a cashier reporting a problem can say which + /// route the terminal was using. + String get label; + + /// Opens the connection. Safe to call when already open. + /// + /// A request/response transport has nothing to open and returns at once. + Future connect(); + + /// Hands a batch over and reports what the back office committed. + /// + /// Implementations must not report an id as accepted until the *application* + /// has confirmed it. A broker acknowledging receipt of the bytes is not the + /// back office confirming the sale. + /// + /// Throws [TransportException] when the outcome is unknown. + Future pushOrders(List> orders); + + /// Cloud-initiated messages. Empty for transports that cannot receive. + Stream get downlink; + + /// Whether the route is currently usable. Drives the header's live pill and + /// wakes the drain engine when it flips to true. + Stream get connectionState; + + bool get isConnected; + + Future dispose(); +} diff --git a/lib/data/remote/simulated_order_transport.dart b/lib/data/remote/simulated_order_transport.dart new file mode 100644 index 0000000..612db40 --- /dev/null +++ b/lib/data/remote/simulated_order_transport.dart @@ -0,0 +1,62 @@ +import 'dart:async'; + +import 'order_transport.dart'; + +/// Stands in for the back office when no broker or endpoint is configured. +/// +/// Keeps the terminal demonstrable on a bare laptop: bills queue, drain, and +/// respect the Settings offline switch exactly as they would against a real +/// server, so the drain engine can be exercised without one. +class SimulatedOrderTransport implements OrderTransport { + SimulatedOrderTransport({required this.isOffline}); + + /// Read on every call rather than copied, so the Settings switch takes effect + /// immediately instead of at the next restart. + final bool Function() isOffline; + + final _downlink = StreamController.broadcast(); + final _connection = StreamController.broadcast(); + + @override + String get label => 'Simulated'; + + @override + bool get isConnected => !isOffline(); + + @override + Stream get downlink => _downlink.stream; + + @override + Stream get connectionState => _connection.stream; + + @override + Future connect() async {} + + @override + Future pushOrders(List> orders) async { + await Future.delayed( + Duration(milliseconds: 400 + orders.length * 60), + ); + + if (isOffline()) { + throw const TransportException( + 'Simulate offline is ON in Settings, so the upload was failed on ' + 'purpose. Every bill is still stored on this terminal.', + ); + } + + return PushReceipt( + accepted: orders.map((o) => o['id']! as String).toList(), + ); + } + + /// Lets the events screen prove the downlink path end to end without a + /// broker. + void emit(DownlinkMessage message) => _downlink.add(message); + + @override + Future dispose() async { + await _downlink.close(); + await _connection.close(); + } +} diff --git a/lib/data/repositories/sync_repository_impl.dart b/lib/data/repositories/sync_repository_impl.dart index 3a2d312..91aae00 100644 --- a/lib/data/repositories/sync_repository_impl.dart +++ b/lib/data/repositories/sync_repository_impl.dart @@ -10,13 +10,24 @@ import '../../domain/repositories/sync_repository.dart'; import '../datasources/local_store.dart'; import '../datasources/remote_catalogue_source.dart'; import '../local/order_dao.dart'; +import '../remote/order_transport.dart'; class SyncRepositoryImpl implements SyncRepository { - SyncRepositoryImpl(this._store, this._catalogue, this._orderSink); + SyncRepositoryImpl( + this._store, + this._catalogue, + this._transport, { + this.batchSize = 50, + }); final LocalStore _store; final RemoteCatalogueSource _catalogue; - final RemoteOrderSink _orderSink; + final OrderTransport _transport; + + /// Bills per publish. Kept modest because an MQTT broker will refuse an + /// oversized message outright, and a terminal that has been offline for a + /// day can easily hold hundreds of bills. + final int batchSize; static const _uuid = Uuid(); @@ -176,13 +187,14 @@ class SyncRepositoryImpl implements SyncRepository { var attempted = 0; var uploaded = 0; + var refused = 0; final syncedInvoices = []; // `unsynced()` returns a bounded page. Draining it in a loop means a day // with more bills than one page still uploads completely, instead of // reporting success with the remainder silently left behind. while (true) { - final batch = await _store.orders.unsynced(); + final batch = await _store.orders.unsynced(limit: batchSize); if (batch.isEmpty) break; final ids = batch.map((o) => o.id).toList(); @@ -193,36 +205,19 @@ class SyncRepositoryImpl implements SyncRepository { 'Uploading $attempted of $total bills…', ); + PushReceipt receipt; try { - final accepted = await _orderSink.pushOrders( + receipt = await _transport.pushOrders( batch.map(_orderToPayload).toList(), ); - - // Only what the server confirmed is archived and removed. Anything it - // did not acknowledge stays on disk. - final acceptedOrders = - batch.where((o) => accepted.contains(o.id)).toList(); - await _store.orders.archiveAndDelete(acceptedOrders); - await _store.refreshUnsyncedCount(); - - uploaded += acceptedOrders.length; - syncedInvoices.addAll(acceptedOrders.map((o) => o.invoiceNumber)); - - final rejected = ids.where((id) => !accepted.contains(id)).toList(); - if (rejected.isNotEmpty) { - await _store.orders.markFailed(rejected, 'Rejected by server'); - // Rejected rows stay pending, so the next page would return the same - // bills forever. Stop and let the cashier retry. - break; - } - } catch (e) { - // Transport failed: record the attempt but leave every row at 0. + } on Object catch (e) { + // The outcome is unknown, so nothing is marked sent. The attempt is + // recorded against the rows and every one of them stays at 0. await _store.orders.markFailed(ids, e.toString()); await _store.refreshUnsyncedCount(); final remaining = await _store.orders.unsyncedCount(); - final pendingValue = - batch.fold(0, (s, o) => s + o.total); + final pendingValue = batch.fold(0, (s, o) => s + o.total); await _log(SyncEvent( id: _uuid.v4(), @@ -238,29 +233,90 @@ class SyncRepositoryImpl implements SyncRepository { return SyncOutcome( attempted: attempted, uploaded: uploaded, + rejected: refused, error: e.toString(), + isRetryable: e is! TransportException || e.retryable, + ); + } + + // Only what the back office named is archived and marked synced. + // Anything it stayed silent about is left pending — silence is not + // acceptance, whatever the transport reported at its own layer. + final acceptedIds = receipt.accepted.toSet(); + final acceptedOrders = + batch.where((o) => acceptedIds.contains(o.id)).toList(); + await _store.orders.archiveAccepted(acceptedOrders); + await _store.refreshUnsyncedCount(); + + uploaded += acceptedOrders.length; + syncedInvoices.addAll(acceptedOrders.map((o) => o.invoiceNumber)); + + final unconfirmed = ids.where((id) => !acceptedIds.contains(id)).toList(); + if (unconfirmed.isNotEmpty) { + for (final id in unconfirmed) { + await _store.orders.markFailed( + [id], + receipt.rejected[id] ?? 'Not confirmed by the back office', + ); + } + refused += unconfirmed.length; + + // These rows are still pending, so the next page would hand back the + // same bills forever. Stop, and let a person look at why. + final reasons = receipt.rejected.values.toSet().join('; '); + await _log(SyncEvent( + id: _uuid.v4(), + type: SyncEventType.shiftReport, + status: SyncStatus.failed, + createdAt: started, + summary: '$uploaded uploaded, ${unconfirmed.length} refused', + error: reasons.isEmpty ? 'Not confirmed by the back office' : reasons, + attempts: 1, + ),); + + return SyncOutcome( + attempted: attempted, + uploaded: uploaded, + rejected: refused, + error: '${unconfirmed.length} bill(s) were not accepted' + '${reasons.isEmpty ? '' : ': $reasons'}', + // A refusal is a decision, not a fault. Retrying the same bytes gets + // the same answer, so the engine halts instead of looping. + isRetryable: false, ); } } + onProgress?.call(0.95, 'Tidying up…'); + await purgeExpired(); onProgress?.call(1, 'Done'); - // Synced bills are deleted from the terminal, so this log line is the only - // remaining record on the device that they went up. + // Once the retention window closes this log line is the only remaining + // record on the device that these bills went up. await _log(SyncEvent( id: _uuid.v4(), type: SyncEventType.shiftReport, status: SyncStatus.synced, createdAt: started, syncedAt: DateTime.now(), - summary: '$uploaded of $attempted bills uploaded', + summary: '$uploaded of $attempted bills uploaded ' + 'via ${_transport.label}', payload: {'invoices': syncedInvoices}, attempts: 1, ),); - return SyncOutcome(attempted: attempted, uploaded: uploaded); + return SyncOutcome( + attempted: attempted, + uploaded: uploaded, + rejected: refused, + ); } + @override + Future purgeExpired() => _store.orders.purgeSyncedBefore( + DateTime.now().subtract(OrderDao.retentionWindow), + ); + /// The JSON body sent per order. Map _orderToPayload(SaleTransaction t) => { 'id': t.id, diff --git a/lib/data/sync/sync_engine.dart b/lib/data/sync/sync_engine.dart new file mode 100644 index 0000000..2fddf3d --- /dev/null +++ b/lib/data/sync/sync_engine.dart @@ -0,0 +1,360 @@ +import 'dart:async'; +import 'dart:math'; + +import 'package:flutter/foundation.dart'; + +import '../../domain/repositories/sync_repository.dart'; +import '../remote/order_transport.dart'; + +/// Why a drain was attempted. Shown in the events log, because "why did it try +/// then" is the first question when a sync misbehaves. +enum SyncTrigger { + startup, + saleCommitted, + connectivityRegained, + periodic, + retry, + headOfficeRequest, + manual, +} + +/// What the engine is doing, for the header and the events screen. +@immutable +class SyncEngineState { + const SyncEngineState({ + this.isSyncing = false, + this.pending = 0, + this.online = true, + this.consecutiveFailures = 0, + this.lastSuccessAt, + this.lastAttemptAt, + this.nextAttemptAt, + this.lastError, + this.isHalted = false, + this.lastTrigger, + }); + + final bool isSyncing; + final int pending; + final bool online; + final int consecutiveFailures; + final DateTime? lastSuccessAt; + final DateTime? lastAttemptAt; + + /// When the backoff timer will fire. Null when nothing is scheduled. + final DateTime? nextAttemptAt; + + final String? lastError; + + /// Set after a failure that retrying cannot fix — a bad credential, an + /// unconfigured endpoint. The engine stops its own scheduling so it does not + /// hammer a door that is locked; a manual sync or a config change resumes it. + final bool isHalted; + + final SyncTrigger? lastTrigger; + + bool get isHealthy => !isHalted && consecutiveFailures == 0; + + SyncEngineState copyWith({ + bool? isSyncing, + int? pending, + bool? online, + int? consecutiveFailures, + DateTime? lastSuccessAt, + DateTime? lastAttemptAt, + DateTime? nextAttemptAt, + String? lastError, + bool? isHalted, + SyncTrigger? lastTrigger, + bool clearNextAttempt = false, + bool clearError = false, + }) => + SyncEngineState( + isSyncing: isSyncing ?? this.isSyncing, + pending: pending ?? this.pending, + online: online ?? this.online, + consecutiveFailures: consecutiveFailures ?? this.consecutiveFailures, + lastSuccessAt: lastSuccessAt ?? this.lastSuccessAt, + lastAttemptAt: lastAttemptAt ?? this.lastAttemptAt, + nextAttemptAt: + clearNextAttempt ? null : nextAttemptAt ?? this.nextAttemptAt, + lastError: clearError ? null : lastError ?? this.lastError, + isHalted: isHalted ?? this.isHalted, + lastTrigger: lastTrigger ?? this.lastTrigger, + ); +} + +/// Decides *when* bills are uploaded. +/// +/// The repository knows how to send one batch; this knows when to ask, and what +/// to do when the answer is no. Together they turn the orders table into a +/// queue that empties itself: +/// +/// * a sale is committed — try immediately, so a bill is usually up within +/// seconds of the drawer closing; +/// * the network returns — try at once rather than waiting out a poll; +/// * nothing happened for a while — poll, because interface state lies; +/// * head office asked — the MQTT downlink can pull a shift up on demand; +/// * the cashier pressed sync — always allowed, even while halted. +/// +/// ### Two guarantees worth stating +/// +/// **Single flight.** Only one drain runs at a time. Without this, a busy till +/// firing a trigger per sale would have several passes reading the same pending +/// rows and publishing them concurrently — every bill sent two or three times. +/// A trigger arriving mid-drain sets a flag and is honoured once the current +/// pass finishes, so nothing is dropped either. +/// +/// **Backoff with jitter.** A failed attempt waits, and waits longer each time, +/// to a ceiling. The jitter matters more than it looks: when a shop's line +/// drops, every terminal in the store fails at the same instant, and without it +/// they would all retry in lockstep and keep colliding on the way back up. +class SyncEngine { + SyncEngine({ + required SyncRepository repository, + Stream? connectivity, + Stream? downlink, + Future Function()? onCatalogueChanged, + Duration idlePoll = const Duration(minutes: 5), + Duration baseBackoff = const Duration(seconds: 2), + Duration maxBackoff = const Duration(minutes: 5), + Random? random, + DateTime Function()? clock, + Timer Function(Duration, void Function())? scheduleTimer, + }) : _repository = repository, + _connectivity = connectivity, + _downlink = downlink, + _onCatalogueChanged = onCatalogueChanged, + _idlePoll = idlePoll, + _baseBackoff = baseBackoff, + _maxBackoff = maxBackoff, + _random = random ?? Random(), + _now = clock ?? DateTime.now, + _schedule = scheduleTimer ?? Timer.new; + + final SyncRepository _repository; + final Stream? _connectivity; + final Stream? _downlink; + final Future Function()? _onCatalogueChanged; + + final Duration _idlePoll; + final Duration _baseBackoff; + final Duration _maxBackoff; + final Random _random; + final DateTime Function() _now; + final Timer Function(Duration, void Function()) _schedule; + + final _states = StreamController.broadcast(); + + SyncEngineState _state = const SyncEngineState(); + SyncEngineState get state => _state; + Stream get states => _states.stream; + + /// Held for the whole of a drain. The single-flight guarantee rests on this + /// being checked and set without an `await` in between. + bool _draining = false; + + /// A trigger that arrived while a drain was already running. + SyncTrigger? _queuedTrigger; + + Timer? _backoffTimer; + Timer? _pollTimer; + StreamSubscription? _connectivitySub; + StreamSubscription? _downlinkSub; + bool _stopped = false; + + // ------------------------------------------------------------------ Life + Future start() async { + if (_stopped) throw StateError('This SyncEngine has been disposed.'); + + _connectivitySub = _connectivity?.listen((online) { + _emit(_state.copyWith(online: online)); + // Coming back is the single best moment to try; going offline is not + // worth an attempt that is certain to fail. + if (online) nudge(SyncTrigger.connectivityRegained); + }); + + _downlinkSub = _downlink?.listen(_onDownlink); + + _pollTimer = _startPoll(); + + await _refreshPending(); + nudge(SyncTrigger.startup); + } + + Timer _startPoll() => _schedule(_idlePoll, () { + if (_stopped) return; + _pollTimer = _startPoll(); + nudge(SyncTrigger.periodic); + }); + + void _onDownlink(DownlinkMessage message) { + switch (message.kind) { + case DownlinkKind.syncRequested: + nudge(SyncTrigger.headOfficeRequest); + case DownlinkKind.catalogueChanged: + // Sending what we owe before pulling new prices keeps the bills we + // already rang priced as they were rung. + nudge(SyncTrigger.headOfficeRequest); + unawaited(_onCatalogueChanged?.call() ?? Future.value()); + case DownlinkKind.unknown: + break; + } + } + + // -------------------------------------------------------------- Triggers + /// Asks for a drain. Cheap, non-blocking, and safe to call on every sale. + void nudge(SyncTrigger trigger) { + if (_stopped) return; + + if (_draining) { + // Remember it rather than dropping it: bills committed during this pass + // were not in the set it read, and would otherwise wait for the poll. + _queuedTrigger = trigger; + return; + } + + if (_state.isHalted && trigger != SyncTrigger.manual) return; + if (!_state.online && trigger != SyncTrigger.manual) return; + + unawaited(_drain(trigger)); + } + + /// The cashier pressed sync. Runs even when halted or believed offline — + /// they may know something the engine does not, and being told why it failed + /// beats a button that does nothing. + Future syncNow({ + void Function(double progress, String stage)? onProgress, + }) async { + _backoffTimer?.cancel(); + _emit(_state.copyWith(isHalted: false, clearNextAttempt: true)); + return _drain(SyncTrigger.manual, onProgress: onProgress); + } + + // ----------------------------------------------------------------- Drain + Future _drain( + SyncTrigger trigger, { + void Function(double progress, String stage)? onProgress, + }) async { + if (_draining) return const SyncOutcome(attempted: 0, uploaded: 0); + _draining = true; + + _backoffTimer?.cancel(); + _emit(_state.copyWith( + isSyncing: true, + lastTrigger: trigger, + lastAttemptAt: _now(), + clearNextAttempt: true, + ),); + + SyncOutcome outcome; + try { + try { + outcome = await _repository.syncOrders(onProgress: onProgress); + } on Object catch (e) { + // The repository is meant to fold failures into the outcome; anything + // escaping it is a defect, not a network fault. Treated as a retryable + // failure so a bad build still empties its queue once fixed. + outcome = SyncOutcome(attempted: 0, uploaded: 0, error: e.toString()); + } + + await _refreshPending(); + + if (outcome.isSuccess) { + _emit(_state.copyWith( + isSyncing: false, + consecutiveFailures: 0, + lastSuccessAt: _now(), + clearError: true, + clearNextAttempt: true, + ),); + } else { + _onFailure(outcome); + } + } finally { + // Held until the bookkeeping is done, not just until the send is. Freed + // any earlier, a queued trigger could start a second drain whose + // `isSyncing: true` this one would then overwrite with `false`, leaving + // the header claiming idle while an upload is in flight. + _draining = false; + } + + // Honour anything that arrived while we were busy. Bills committed during + // the pass were not in the set it read. + final queued = _queuedTrigger; + _queuedTrigger = null; + if (queued != null && outcome.isSuccess && _state.pending > 0) { + nudge(queued); + } + + return outcome; + } + + void _onFailure(SyncOutcome outcome) { + final failures = _state.consecutiveFailures + 1; + + if (!outcome.isRetryable) { + _emit(_state.copyWith( + isSyncing: false, + consecutiveFailures: failures, + lastError: outcome.error, + isHalted: true, + clearNextAttempt: true, + ),); + return; + } + + final delay = backoffFor(failures); + _emit(_state.copyWith( + isSyncing: false, + consecutiveFailures: failures, + lastError: outcome.error, + nextAttemptAt: _now().add(delay), + ),); + + _backoffTimer = _schedule(delay, () { + if (_stopped) return; + nudge(SyncTrigger.retry); + }); + } + + /// Doubles per failure to a ceiling, then ±20% so a store's terminals do not + /// come back in lockstep. + @visibleForTesting + Duration backoffFor(int failures) { + final exponent = (failures - 1).clamp(0, 30); + final raw = _baseBackoff * pow(2, exponent).toDouble(); + final capped = raw > _maxBackoff ? _maxBackoff : raw; + final jitter = 0.8 + _random.nextDouble() * 0.4; + return Duration( + milliseconds: (capped.inMilliseconds * jitter).round(), + ); + } + + Future _refreshPending() async { + try { + // Read first, then emit. Written as `copyWith(pending: await …)` the + // receiver `_state` is evaluated before the await completes, so anything + // that changed during the wait — connectivity dropping, most of all — + // would be overwritten by the stale snapshot. + final count = await _repository.unsyncedCount(); + _emit(_state.copyWith(pending: count)); + } on Object { + // A count is decoration; failing to read it must not fail the drain. + } + } + + void _emit(SyncEngineState next) { + _state = next; + if (!_states.isClosed) _states.add(next); + } + + Future dispose() async { + _stopped = true; + _backoffTimer?.cancel(); + _pollTimer?.cancel(); + await _connectivitySub?.cancel(); + await _downlinkSub?.cancel(); + await _states.close(); + } +} diff --git a/lib/domain/repositories/sync_repository.dart b/lib/domain/repositories/sync_repository.dart index e5872b0..72fb782 100644 --- a/lib/domain/repositories/sync_repository.dart +++ b/lib/domain/repositories/sync_repository.dart @@ -2,18 +2,30 @@ import '../entities/shift_report.dart'; import '../entities/sync_event.dart'; import '../entities/transaction.dart'; -/// Result of one end-of-day upload. +/// Result of one upload pass. class SyncOutcome { const SyncOutcome({ required this.attempted, required this.uploaded, + this.rejected = 0, this.error, + this.isRetryable = true, }); final int attempted; final int uploaded; + + /// Bills the back office looked at and refused. These stay on the terminal + /// but sending them again unchanged will fail again, so they need a person. + final int rejected; + final String? error; + /// Whether trying again could plausibly work. False for a bad credential or + /// an unconfigured endpoint — the drain engine halts rather than retrying + /// something that cannot succeed. + final bool isRetryable; + bool get isSuccess => error == null; bool get hadNothingToDo => attempted == 0; int get remaining => attempted - uploaded; @@ -42,10 +54,11 @@ class OrderSyncRow { final String? error; } -/// The terminal's two network touchpoints. +/// The terminal's network touchpoints. /// -/// Morning: pull the catalogue. End of day: upload every order still at -/// `sync_status = 0`. Nothing else leaves the device. +/// Pull the catalogue; upload every order still at `sync_status = 0`. Nothing +/// else leaves the device. *When* the upload runs is not decided here — see +/// `SyncEngine`, which owns triggers and retry policy. abstract class SyncRepository { bool get hasCatalogue; DateTime? get lastImportAt; @@ -71,12 +84,16 @@ abstract class SyncRepository { bool scopeToCashier = false, }); - /// End-of-day step — uploads pending orders and flips the accepted ones to - /// `sync_status = 1`. Failures leave every row untouched at 0. + /// One upload pass — sends pending orders and flips the ones the back office + /// confirmed to `sync_status = 1`. Failures leave every row untouched at 0. Future syncOrders({ void Function(double progress, String stage)? onProgress, }); + /// Retires confirmed bills past their retention window. Archived totals are + /// untouched. + Future purgeExpired(); + Future> orderSyncRows({int limit = 200}); List get events; diff --git a/lib/presentation/modules/providers/printer_settings.dart b/lib/presentation/modules/providers/printer_settings.dart index 7eff66a..c4be353 100644 --- a/lib/presentation/modules/providers/printer_settings.dart +++ b/lib/presentation/modules/providers/printer_settings.dart @@ -61,11 +61,22 @@ class PrinterSettingsController extends StateNotifier { if (!store.isReady) return; final dao = store.catalogue; + + // Read every value first, then assign. Written inline the four awaits still + // run before the assignment, so leaving Settings mid-read threw + // "used after dispose" — which reaches the cashier as a red screen. + final url = await dao.meta(MetaKeys.printerUrl); + final name = await dao.meta(MetaKeys.printerName); + final autoPrint = await dao.meta(MetaKeys.autoPrint); + final openDrawer = await dao.meta(MetaKeys.openDrawer); + + if (!mounted) return; + state = PrinterSettings( - printerUrl: await dao.meta(MetaKeys.printerUrl), - printerName: await dao.meta(MetaKeys.printerName), - autoPrint: (await dao.meta(MetaKeys.autoPrint)) == '1', - openDrawer: (await dao.meta(MetaKeys.openDrawer)) != '0', + printerUrl: url, + printerName: name, + autoPrint: autoPrint == '1', + openDrawer: openDrawer != '0', ); } @@ -75,12 +86,14 @@ class PrinterSettingsController extends StateNotifier { if (printer == null) { await dao.setMeta(MetaKeys.printerUrl, ''); await dao.setMeta(MetaKeys.printerName, ''); + if (!mounted) return; state = state.copyWith(clearPrinter: true, autoPrint: false); return; } await dao.setMeta(MetaKeys.printerUrl, printer.url); await dao.setMeta(MetaKeys.printerName, printer.name); + if (!mounted) return; state = state.copyWith( printerUrl: printer.url, printerName: printer.name, @@ -95,6 +108,7 @@ class PrinterSettingsController extends StateNotifier { .read(localStoreProvider) .catalogue .setMeta(MetaKeys.autoPrint, value ? '1' : '0'); + if (!mounted) return; state = state.copyWith(autoPrint: value); } @@ -103,6 +117,7 @@ class PrinterSettingsController extends StateNotifier { .read(localStoreProvider) .catalogue .setMeta(MetaKeys.openDrawer, value ? '1' : '0'); + if (!mounted) return; state = state.copyWith(openDrawer: value); } } diff --git a/lib/presentation/modules/screens/settings_view.dart b/lib/presentation/modules/screens/settings_view.dart index 363fe95..8253065 100644 --- a/lib/presentation/modules/screens/settings_view.dart +++ b/lib/presentation/modules/screens/settings_view.dart @@ -2,10 +2,12 @@ import 'package:flutter/material.dart'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import '../../../app/providers.dart'; +import '../../../core/config/sync_config.dart'; import '../../../core/constants/app_constants.dart'; import '../../../core/theme/app_colors.dart'; import '../../../core/theme/app_dimens.dart'; import '../../../core/utils/formatters.dart'; +import '../../../data/local/order_dao.dart'; import '../../../domain/entities/store_account.dart'; import '../../auth/providers/auth_controller.dart'; import '../providers/printer_settings.dart'; @@ -318,11 +320,14 @@ class _SettingsViewState extends ConsumerState { final ready = ref.watch(catalogueReadyProvider); final lastImport = ref.watch(lastImportAtProvider); final outstanding = ref.watch(unsyncedCountProvider).value ?? 0; + final config = ref.watch(syncConfigProvider); + final sync = ref.watch(syncEngineStateProvider).value ?? + ref.watch(syncEngineProvider).state; return PanelCard( title: 'Connectivity & sync', - subtitle: 'This terminal only needs a connection to import the ' - 'catalogue and to push the shift report.', + subtitle: 'Bills are written to this terminal first and uploaded in the ' + 'background. Nothing is ever held up waiting for the network.', child: Column( mainAxisSize: MainAxisSize.min, children: [ @@ -331,7 +336,26 @@ class _SettingsViewState extends ConsumerState { 'Last import', lastImport == null ? 'Never' : Formatters.dateTime(lastImport), ), - _row('Unsynced bills', '$outstanding'), + _row('Route', _transportLabel(config)), + _row('Waiting to upload', '$outstanding bill(s)'), + _row( + 'Last upload', + sync.lastSuccessAt == null + ? 'Never' + : Formatters.dateTime(sync.lastSuccessAt!), + ), + if (sync.isHalted) + _row('Status', 'Halted — ${sync.lastError ?? 'refused'}') + else if (sync.nextAttemptAt != null) + _row( + 'Next attempt', + '${Formatters.time(sync.nextAttemptAt!)} ' + '(attempt ${sync.consecutiveFailures + 1})', + ), + _row( + 'Bills kept on device', + '${OrderDao.retentionWindow.inDays} days after upload', + ), _toggle( 'Simulate offline', 'Forces import and sync to fail, so you can confirm nothing is ' @@ -340,6 +364,9 @@ class _SettingsViewState extends ConsumerState { ref.watch(simulateOfflineProvider), (v) { ref.read(simulateOfflineProvider.notifier).state = v; + // The pill and the drain both read connectivity, so they have to + // be told the switch moved. + ref.read(connectivityServiceProvider).refresh(); // Clear any stale failure banner left by the previous setting. ref.read(catalogueImportProvider.notifier).reset(); ref.read(orderSyncProvider.notifier).reset(); @@ -350,6 +377,15 @@ class _SettingsViewState extends ConsumerState { ); } + String _transportLabel(SyncConfig config) => switch (config.transport) { + TransportKind.simulated => + 'Simulated — no back office configured for this terminal', + TransportKind.http => 'HTTP · ${config.httpBaseUrl}', + TransportKind.mqtt => + 'MQTT · ${config.brokerHost}:${config.brokerPort}' + '${config.useTls ? ' (TLS)' : ''}', + }; + Widget _aboutCard() => PanelCard( title: 'About', child: Column( diff --git a/lib/presentation/payment/providers/payment_controller.dart b/lib/presentation/payment/providers/payment_controller.dart index 2a0193e..fdc6680 100644 --- a/lib/presentation/payment/providers/payment_controller.dart +++ b/lib/presentation/payment/providers/payment_controller.dart @@ -4,6 +4,7 @@ import 'package:flutter_riverpod/flutter_riverpod.dart'; import '../../../app/providers.dart'; import '../../../core/utils/extensions.dart'; +import '../../../data/sync/sync_engine.dart'; import '../../../domain/entities/transaction.dart'; import '../../../domain/usecases/checkout_sale.dart'; import '../../auth/providers/auth_controller.dart'; @@ -199,6 +200,11 @@ class PaymentController extends StateNotifier { _ref.invalidate(visibleProductsProvider); _ref.read(orderVersionProvider.notifier).state++; + // The bill is safely on disk; getting it to the back office is the + // engine's problem now. Deliberately not awaited — the cashier must + // reach the receipt screen at network speed of zero. + _ref.read(syncEngineProvider).nudge(SyncTrigger.saleCommitted); + return result; } on CheckoutFailure catch (e) { state = state.copyWith( diff --git a/lib/presentation/pos/widgets/page_header.dart b/lib/presentation/pos/widgets/page_header.dart index 9ac28c9..01719c1 100644 --- a/lib/presentation/pos/widgets/page_header.dart +++ b/lib/presentation/pos/widgets/page_header.dart @@ -154,16 +154,19 @@ class _Breadcrumb extends StatelessWidget { } } -class _LivePill extends StatefulWidget { +/// What the pill is saying, in the order it takes precedence. +enum _Liveness { offlineSim, halted, syncing, queued, live } + +class _LivePill extends ConsumerStatefulWidget { const _LivePill({required this.offline}); final bool offline; @override - State<_LivePill> createState() => _LivePillState(); + ConsumerState<_LivePill> createState() => _LivePillState(); } -class _LivePillState extends State<_LivePill> +class _LivePillState extends ConsumerState<_LivePill> with SingleTickerProviderStateMixin { late final AnimationController _c = AnimationController( vsync: this, @@ -178,16 +181,62 @@ class _LivePillState extends State<_LivePill> @override Widget build(BuildContext context) { - final offline = widget.offline; - final tone = offline ? AppColors.warning : AppColors.success; - final surface = - offline ? AppColors.warningSurface : AppColors.successSurface; + // The engine may not have emitted yet on a cold start, so fall back to its + // current value rather than showing nothing. + final sync = ref.watch(syncEngineStateProvider).value ?? + ref.watch(syncEngineProvider).state; + + final liveness = switch (0) { + _ when widget.offline => _Liveness.offlineSim, + _ when sync.isHalted => _Liveness.halted, + _ when sync.isSyncing => _Liveness.syncing, + // Bills waiting is normal for a few seconds after a sale; it is only + // worth flagging once they are visibly piling up. + _ when sync.pending > 0 => _Liveness.queued, + _ => _Liveness.live, + }; + + final (label, tone, surface, message) = switch (liveness) { + _Liveness.offlineSim => ( + 'OFFLINE (SIM)', + AppColors.warning, + AppColors.warningSurface, + 'Simulate offline is ON in Settings — imports and syncs are being ' + 'failed deliberately.', + ), + _Liveness.halted => ( + 'SYNC HALTED', + AppColors.danger, + AppColors.dangerSurface, + 'Uploading stopped because retrying will not help: ' + '${sync.lastError ?? 'the back office refused the batch'}. ' + 'Every bill is still safe on this terminal. Press Sync to try ' + 'again once it is sorted.', + ), + _Liveness.syncing => ( + 'SYNCING', + AppColors.primary, + AppColors.primarySurface, + 'Uploading bills to the back office.', + ), + _Liveness.queued => ( + '${sync.pending} QUEUED', + AppColors.warning, + AppColors.warningSurface, + '${sync.pending} bill(s) are stored on this terminal and waiting to ' + 'upload. They are safe; nothing is lost while the line is down.', + ), + _Liveness.live => ( + 'LIVE', + AppColors.success, + AppColors.successSurface, + 'Terminal is operating normally and everything rung has been ' + 'uploaded.', + ), + }; return Tooltip( - message: offline - ? 'Simulate offline is ON in Settings — imports and syncs are being ' - 'failed deliberately.' - : 'Terminal is operating normally.', + message: message, child: Container( padding: const EdgeInsets.symmetric( horizontal: AppSpacing.md, @@ -210,7 +259,7 @@ class _LivePillState extends State<_LivePill> ), const SizedBox(width: AppSpacing.xs + 2), Text( - offline ? 'OFFLINE (SIM)' : 'LIVE', + label, style: TextStyle( color: tone, fontSize: 10.5, diff --git a/lib/presentation/sync/providers/sync_controller.dart b/lib/presentation/sync/providers/sync_controller.dart index fa49bd3..2f1afc1 100644 --- a/lib/presentation/sync/providers/sync_controller.dart +++ b/lib/presentation/sync/providers/sync_controller.dart @@ -167,6 +167,12 @@ class OrderSyncController extends StateNotifier { bool get isRunning => state is SyncRunning; /// Uploads every bill at sync_status = 0 and flips the accepted ones to 1. + /// + /// Goes through the engine rather than straight to the repository, so a + /// cashier pressing sync while a background drain is already mid-flight + /// joins it instead of starting a second pass over the same rows. It also + /// clears a halt: pressing the button is how you retry after the back office + /// has been fixed. Future run() async { if (isRunning) { return const SyncOutcome(attempted: 0, uploaded: 0); @@ -174,7 +180,7 @@ class OrderSyncController extends StateNotifier { state = const SyncRunning(0, 'Starting…'); - final outcome = await _ref.read(syncRepositoryProvider).syncOrders( + final outcome = await _ref.read(syncEngineProvider).syncNow( onProgress: (progress, stage) { if (mounted) state = SyncRunning(progress, stage); }, @@ -192,3 +198,30 @@ final orderSyncProvider = StateNotifierProvider( (ref) => OrderSyncController(ref), ); + +// ------------------------------------------------------- Background drain +/// Brings the queue-and-drain machinery up, once, when the shell mounts. +/// +/// Deliberately not gated on sign-in: a terminal that boots holding yesterday's +/// bills should be emptying its queue before anyone reaches the till. +/// +/// Overridden to a no-op in widget tests, which have no network stack and +/// cannot drive real disk I/O on a fake clock. +final syncBootstrapProvider = FutureProvider((ref) async { + await ref.read(connectivityServiceProvider).start(); + + final engine = ref.read(syncEngineProvider); + + // A background drain moves bills out of the pending set, so the tallies and + // shift totals on screen are stale the moment one finishes. + var wasSyncing = false; + final subscription = engine.states.listen((state) { + if (wasSyncing && !state.isSyncing) { + ref.read(orderVersionProvider.notifier).state++; + } + wasSyncing = state.isSyncing; + }); + ref.onDispose(subscription.cancel); + + await engine.start(); +}); diff --git a/macos/Flutter/GeneratedPluginRegistrant.swift b/macos/Flutter/GeneratedPluginRegistrant.swift index 8f62591..1b0aef8 100644 --- a/macos/Flutter/GeneratedPluginRegistrant.swift +++ b/macos/Flutter/GeneratedPluginRegistrant.swift @@ -6,12 +6,14 @@ import FlutterMacOS import Foundation import audioplayers_darwin +import connectivity_plus import printing import sqflite_darwin import url_launcher_macos func RegisterGeneratedPlugins(registry: FlutterPluginRegistry) { AudioplayersDarwinPlugin.register(with: registry.registrar(forPlugin: "AudioplayersDarwinPlugin")) + ConnectivityPlusPlugin.register(with: registry.registrar(forPlugin: "ConnectivityPlusPlugin")) PrintingPlugin.register(with: registry.registrar(forPlugin: "PrintingPlugin")) SqflitePlugin.register(with: registry.registrar(forPlugin: "SqflitePlugin")) UrlLauncherPlugin.register(with: registry.registrar(forPlugin: "UrlLauncherPlugin")) diff --git a/pubspec.lock b/pubspec.lock index 8b86183..4ebad3e 100644 --- a/pubspec.lock +++ b/pubspec.lock @@ -137,6 +137,22 @@ packages: url: "https://pub.dev" source: hosted version: "1.19.1" + connectivity_plus: + dependency: "direct main" + description: + name: connectivity_plus + sha256: "762c99f890ca8bf87f7337236f99edd42793843bc6c3631da294a76653a54bd0" + url: "https://pub.dev" + source: hosted + version: "7.3.1" + connectivity_plus_platform_interface: + dependency: transitive + description: + name: connectivity_plus_platform_interface + sha256: "3c09627c536d22fd24691a905cdd8b14520de69da52c7a97499c8be5284a32ed" + url: "https://pub.dev" + source: hosted + version: "2.1.0" crypto: dependency: transitive description: @@ -145,6 +161,14 @@ packages: url: "https://pub.dev" source: hosted version: "3.0.7" + dbus: + dependency: transitive + description: + name: dbus + sha256: "792974a4007974fbc5c1b5433eb2330a9db3e368c3f906253af4c007d0f49a91" + url: "https://pub.dev" + source: hosted + version: "0.7.13" equatable: dependency: "direct main" description: @@ -153,6 +177,14 @@ packages: url: "https://pub.dev" source: hosted version: "2.1.0" + event_bus: + dependency: transitive + description: + name: event_bus + sha256: "1a55e97923769c286d295240048fc180e7b0768902c3c2e869fe059aafa15304" + url: "https://pub.dev" + source: hosted + version: "2.0.1" fake_async: dependency: transitive description: @@ -265,7 +297,7 @@ packages: source: hosted version: "2.0.2" http: - dependency: transitive + dependency: "direct main" description: name: http sha256: "87721a4a50b19c7f1d49001e51409bddc46303966ce89a65af4f4e6004896412" @@ -384,6 +416,14 @@ packages: url: "https://pub.dev" source: hosted version: "1.18.0" + mqtt_client: + dependency: "direct main" + description: + name: mqtt_client + sha256: "41c8edd3bc8efc80c1c8ebfb40081c24d12d13085faca96b9280a624eca2d893" + url: "https://pub.dev" + source: hosted + version: "10.11.11" native_toolchain_c: dependency: transitive description: @@ -392,6 +432,14 @@ packages: url: "https://pub.dev" source: hosted version: "0.19.2" + nm: + dependency: transitive + description: + name: nm + sha256: "2c9aae4127bdc8993206464fcc063611e0e36e72018696cd9631023a31b24254" + url: "https://pub.dev" + source: hosted + version: "0.5.0" objective_c: dependency: transitive description: diff --git a/pubspec.yaml b/pubspec.yaml index 8de34f3..40432f1 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -39,6 +39,9 @@ dependencies: audioplayers: ^6.0.0 pdf: ^3.11.0 printing: ^5.13.0 + mqtt_client: ^10.11.11 + connectivity_plus: ^7.3.1 + http: ^1.6.0 dev_dependencies: flutter_test: diff --git a/test/unit/persistence_test.dart b/test/unit/persistence_test.dart index 3dd602e..e149428 100644 --- a/test/unit/persistence_test.dart +++ b/test/unit/persistence_test.dart @@ -2,6 +2,7 @@ import 'package:flutter_test/flutter_test.dart'; import 'package:nearle_pos/data/datasources/local_store.dart'; import 'package:nearle_pos/data/datasources/remote_catalogue_source.dart'; import 'package:nearle_pos/data/datasources/seed_data.dart'; +import 'package:nearle_pos/data/remote/simulated_order_transport.dart'; import 'package:nearle_pos/data/repositories/customer_repository_impl.dart'; import 'package:nearle_pos/data/repositories/product_repository_impl.dart'; import 'package:nearle_pos/data/repositories/sync_repository_impl.dart'; @@ -270,7 +271,7 @@ void main() { final sync = SyncRepositoryImpl( store, RemoteCatalogueSource(isOffline: () => false), - RemoteOrderSink(isOffline: () => false), + SimulatedOrderTransport(isOffline: () => false), ); final milk = (await products.findByBarcode('8901234500011'))!; @@ -314,7 +315,7 @@ void main() { final sync = SyncRepositoryImpl( store, RemoteCatalogueSource(isOffline: () => false), - RemoteOrderSink(isOffline: () => false), + SimulatedOrderTransport(isOffline: () => false), ); final milk = (await products.findByBarcode('8901234500011'))!; @@ -343,7 +344,7 @@ void main() { final sync = SyncRepositoryImpl( store, RemoteCatalogueSource(isOffline: () => true), - RemoteOrderSink(isOffline: () => true), + SimulatedOrderTransport(isOffline: () => true), ); await sync.importCatalogue(); @@ -371,7 +372,7 @@ void main() { final sync = SyncRepositoryImpl( store, RemoteCatalogueSource(isOffline: () => false), - RemoteOrderSink(isOffline: () => false), + SimulatedOrderTransport(isOffline: () => false), ); await sell('Divya', 2); // 124 @@ -397,7 +398,7 @@ void main() { final sync = SyncRepositoryImpl( store, RemoteCatalogueSource(isOffline: () => false), - RemoteOrderSink(isOffline: () => false), + SimulatedOrderTransport(isOffline: () => false), ); await sell('Divya', 2); diff --git a/test/unit/retention_test.dart b/test/unit/retention_test.dart new file mode 100644 index 0000000..7b25bc6 --- /dev/null +++ b/test/unit/retention_test.dart @@ -0,0 +1,356 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:nearle_pos/data/datasources/local_store.dart'; +import 'package:nearle_pos/data/datasources/remote_catalogue_source.dart'; +import 'package:nearle_pos/data/datasources/seed_data.dart'; +import 'package:nearle_pos/data/local/app_database.dart'; +import 'package:nearle_pos/data/local/order_dao.dart'; +import 'package:nearle_pos/data/remote/order_transport.dart'; +import 'package:nearle_pos/data/remote/simulated_order_transport.dart'; +import 'package:nearle_pos/data/repositories/customer_repository_impl.dart'; +import 'package:nearle_pos/data/repositories/product_repository_impl.dart'; +import 'package:nearle_pos/data/repositories/sync_repository_impl.dart'; +import 'package:nearle_pos/data/repositories/transaction_repository_impl.dart'; +import 'package:nearle_pos/domain/entities/cart.dart'; +import 'package:nearle_pos/domain/entities/transaction.dart'; +import 'package:nearle_pos/domain/usecases/checkout_sale.dart'; + +/// A transport whose answer each call is dictated by the test. +class _ScriptedTransport implements OrderTransport { + _ScriptedTransport(this.answer); + + /// Given the ids in a batch, returns what the back office says about them. + PushReceipt Function(List ids) answer; + + int batches = 0; + + @override + String get label => 'Scripted'; + + @override + bool get isConnected => true; + + @override + Stream get downlink => const Stream.empty(); + + @override + Stream get connectionState => const Stream.empty(); + + @override + Future connect() async {} + + @override + Future pushOrders(List> orders) async { + batches++; + return answer(orders.map((o) => o['id']! as String).toList()); + } + + @override + Future dispose() async {} +} + +/// Bills the back office has taken delivery of stay on the terminal for a +/// week, so a batch the server later loses can still be re-sent in full. +/// +/// The risk this buys is double counting: an accepted bill is now in two +/// places at once — its own row, and the archived day totals. Most of what +/// follows is about that. +void main() { + late LocalStore store; + late ProductRepositoryImpl products; + late CustomerRepositoryImpl customers; + late TransactionRepositoryImpl transactions; + late CheckoutSale checkout; + + setUpAll(() { + LocalStore.registerSeed( + products: SeedData.products, + customers: SeedData.customers, + ); + }); + + setUp(() async { + store = LocalStore.instance; + await store.reset(withCatalogue: true); + + products = ProductRepositoryImpl(store); + customers = CustomerRepositoryImpl(store); + transactions = TransactionRepositoryImpl(store); + checkout = CheckoutSale( + productRepository: products, + customerRepository: customers, + transactionRepository: transactions, + ); + }); + + SyncRepositoryImpl syncWith(OrderTransport transport) => SyncRepositoryImpl( + store, + RemoteCatalogueSource(isOffline: () => false), + transport, + ); + + /// Rings one bill for [quantity] litres of milk at 62.00 each. + Future ringSale({ + double quantity = 2, + String cashier = 'Divya', + }) async { + final milk = (await products.findByBarcode('8901234500011'))!; + final cart = Cart(lines: [CartLine(product: milk, quantity: quantity)]); + final due = cart.grandTotal; + + await checkout( + cart: cart, + payments: [ + PaymentSplit( + method: PaymentMethod.cash, + amount: due, + tendered: due, + ), + ], + cashierName: cashier, + ); + return due; + } + + group('accepted bills are kept, not deleted', () { + test('a synced bill is still on the terminal and still re-sendable', + () async { + await ringSale(); + final sync = syncWith(SimulatedOrderTransport(isOffline: () => false)); + + expect(await sync.unsyncedCount(), 1); + await sync.syncOrders(); + expect(await sync.unsyncedCount(), 0); + + // The row survives, carrying its line items, so the full bill can go up + // again if the back office loses it. + final rows = await sync.orderSyncRows(); + expect(rows, hasLength(1)); + expect(rows.single.isSynced, isTrue); + expect(rows.single.syncedAt, isNotNull); + + final stored = await store.orders.recent(); + expect(stored, hasLength(1)); + expect(stored.single.cart.lines, isNotEmpty, + reason: 'a kept bill with no lines could not be re-sent',); + }); + + test("today's takings are not counted twice while the bill is retained", + () async { + // The bug this exists to catch: an accepted bill is in the archive *and* + // still in the orders table. Summing both would inflate the day. + final due = await ringSale(quantity: 3); + final sync = syncWith(SimulatedOrderTransport(isOffline: () => false)); + + final before = await sync.todayReport( + terminalId: 'TERM-01', + cashierName: 'Divya', + ); + expect(before.grossSales, closeTo(due, 0.01)); + + await sync.syncOrders(); + + final after = await sync.todayReport( + terminalId: 'TERM-01', + cashierName: 'Divya', + ); + expect(after.grossSales, closeTo(due, 0.01), + reason: 'syncing must not change what the shop took',); + expect(after.billCount, 1); + }); + + test('a second sync does not send an already-accepted bill again', + () async { + await ringSale(); + final transport = _ScriptedTransport((ids) => PushReceipt(accepted: ids)); + final sync = syncWith(transport); + + await sync.syncOrders(); + expect(transport.batches, 1); + + final outcome = await sync.syncOrders(); + expect(outcome.hadNothingToDo, isTrue); + expect(transport.batches, 1, + reason: 'a retained bill must not be re-uploaded',); + }); + }); + + group('purging', () { + test('a bill past its window goes, and its archived totals stay', () async { + final due = await ringSale(quantity: 4); + final sync = syncWith(SimulatedOrderTransport(isOffline: () => false)); + await sync.syncOrders(); + + // Backdate the acceptance past the retention window. + await AppDatabase.instance.db.rawUpdate( + 'UPDATE orders SET synced_at = ?', + [ + DateTime.now() + .subtract(OrderDao.retentionWindow + const Duration(days: 1)) + .millisecondsSinceEpoch, + ], + ); + + expect(await sync.purgeExpired(), 1); + expect(await store.orders.recent(), isEmpty); + + // What the shop was paid is unchanged — only the re-sendable copy went. + final report = await sync.todayReport( + terminalId: 'TERM-01', + cashierName: 'Divya', + ); + expect(report.grossSales, closeTo(due, 0.01)); + expect(report.billCount, 1); + }); + + test('a bill inside its window is left alone', () async { + await ringSale(); + final sync = syncWith(SimulatedOrderTransport(isOffline: () => false)); + await sync.syncOrders(); + + expect(await sync.purgeExpired(), 0); + expect(await store.orders.recent(), hasLength(1)); + }); + + test('an unsynced bill is never purged, however old', () async { + // The one thing that must never happen: a bill the back office has not + // taken delivery of being deleted from the only place it exists. + await ringSale(); + await AppDatabase.instance.db.rawUpdate( + 'UPDATE orders SET created_at = ?, synced_at = ?', + [0, 0], + ); + + final sync = syncWith(SimulatedOrderTransport(isOffline: () => false)); + expect(await sync.purgeExpired(), 0); + expect(await sync.unsyncedCount(), 1); + }); + }); + + group('partial acceptance', () { + test('a bill the back office stayed silent about stays pending', () async { + // Silence is not acceptance. + await ringSale(quantity: 1); + await ringSale(quantity: 2); + + final transport = _ScriptedTransport( + (ids) => PushReceipt(accepted: [ids.first]), + ); + final sync = syncWith(transport); + + final outcome = await sync.syncOrders(); + expect(outcome.uploaded, 1); + expect(outcome.rejected, 1); + expect(await sync.unsyncedCount(), 1, + reason: 'the unconfirmed bill must still be owed',); + }); + + test('a refusal stops the drain instead of looping on the same rows', + () async { + await ringSale(); + final transport = _ScriptedTransport( + (ids) => PushReceipt( + accepted: const [], + rejected: {for (final id in ids) id: 'duplicate invoice'}, + ), + ); + final sync = syncWith(transport); + + final outcome = await sync.syncOrders(); + + expect(outcome.uploaded, 0); + expect(outcome.isSuccess, isFalse); + expect(outcome.isRetryable, isFalse, + reason: 'the same bytes will be refused again',); + expect(transport.batches, 1, + reason: 'the refused page must not be fetched and sent forever',); + expect(outcome.error, contains('duplicate invoice')); + }); + + test('the refusal reason is recorded against the bill for a person to read', + () async { + await ringSale(); + final sync = syncWith(_ScriptedTransport( + (ids) => PushReceipt( + accepted: const [], + rejected: {for (final id in ids) id: 'unknown product code'}, + ), + ),); + + await sync.syncOrders(); + + final row = (await sync.orderSyncRows()).single; + expect(row.isSynced, isFalse); + expect(row.error, 'unknown product code'); + expect(row.attempts, 1); + }); + }); + + group('at-least-once delivery', () { + test('a batch accepted twice is banked once', () async { + // MQTT will re-deliver, and a lost ack means the terminal sends again. + // The second acceptance must not double the archived takings. + final due = await ringSale(quantity: 5); + final transport = _ScriptedTransport((ids) => PushReceipt(accepted: ids)); + final sync = syncWith(transport); + + await sync.syncOrders(); + // A duplicate ack for bills already marked synced — the drain finds + // nothing pending and does nothing. + await sync.syncOrders(); + + final report = await sync.todayReport( + terminalId: 'TERM-01', + cashierName: 'Divya', + ); + expect(report.grossSales, closeTo(due, 0.01)); + expect(report.billCount, 1); + }); + }); + + group('transport failure', () { + test('an unreachable back office leaves every bill exactly where it was', + () async { + final due = await ringSale(quantity: 6); + final sync = syncWith(SimulatedOrderTransport(isOffline: () => true)); + + final outcome = await sync.syncOrders(); + + expect(outcome.isSuccess, isFalse); + expect(outcome.uploaded, 0); + expect(await sync.unsyncedCount(), 1); + + final report = await sync.todayReport( + terminalId: 'TERM-01', + cashierName: 'Divya', + ); + expect(report.grossSales, closeTo(due, 0.01), + reason: 'a failed upload must not change the shift total',); + }); + + test('a batch is bounded so a backlog cannot exceed a broker message', + () async { + for (var i = 0; i < 5; i++) { + await ringSale(quantity: 1); + } + + final sizes = []; + final transport = _ScriptedTransport((ids) { + sizes.add(ids.length); + return PushReceipt(accepted: ids); + }); + + final sync = SyncRepositoryImpl( + store, + RemoteCatalogueSource(isOffline: () => false), + transport, + batchSize: 2, + ); + + final outcome = await sync.syncOrders(); + + expect(outcome.uploaded, 5); + expect(sizes, [2, 2, 1], reason: 'the backlog must drain in pages'); + expect(await sync.unsyncedCount(), 0); + }); + }); +} diff --git a/test/unit/sync_engine_test.dart b/test/unit/sync_engine_test.dart new file mode 100644 index 0000000..1e83d0b --- /dev/null +++ b/test/unit/sync_engine_test.dart @@ -0,0 +1,436 @@ +import 'dart:async'; +import 'dart:math'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:nearle_pos/data/remote/order_transport.dart'; +import 'package:nearle_pos/data/sync/sync_engine.dart'; +import 'package:nearle_pos/domain/entities/shift_report.dart'; +import 'package:nearle_pos/domain/entities/sync_event.dart'; +import 'package:nearle_pos/domain/entities/transaction.dart'; +import 'package:nearle_pos/domain/repositories/sync_repository.dart'; + +/// A repository whose every answer the test dictates. +/// +/// The engine is a scheduler; what it schedules is irrelevant here. Driving it +/// with a stub is what makes "did it retry, and when" answerable without a +/// database or a network. +class _StubRepository implements SyncRepository { + _StubRepository(); + + /// Answers handed out in order; the last one repeats once exhausted. + final List scripted = []; + int calls = 0; + int pending = 0; + + /// Completed by the test to hold a drain open, so overlapping triggers can + /// be observed. + Completer? gate; + + @override + Future syncOrders({ + void Function(double progress, String stage)? onProgress, + }) async { + calls++; + if (gate != null) await gate!.future; + if (scripted.isEmpty) return const SyncOutcome(attempted: 0, uploaded: 0); + return scripted[(calls - 1).clamp(0, scripted.length - 1)]; + } + + @override + Future unsyncedCount() async => pending; + + @override + Future purgeExpired() async => 0; + + @override + bool get hasCatalogue => true; + + @override + DateTime? get lastImportAt => null; + + @override + String? get catalogueRevision => null; + + @override + List get events => const []; + + @override + Future importCatalogue({ + void Function(double progress, String stage)? onProgress, + }) => + throw UnimplementedError(); + + @override + Future> unsyncedOrders() async => const []; + + @override + Future> orderSyncRows({int limit = 200}) async => const []; + + @override + Future todayReport({ + required String terminalId, + required String cashierName, + bool scopeToCashier = false, + }) => + throw UnimplementedError(); +} + +/// Captures what the engine asked to be scheduled instead of really waiting. +class _FakeScheduler { + final List delays = []; + final List callbacks = []; + + Timer schedule(Duration d, void Function() cb) { + delays.add(d); + callbacks.add(cb); + return Timer(Duration.zero, () {})..cancel(); + } + + /// Runs the most recently scheduled callback — the backoff retry. + void fireLast() => callbacks.last(); +} + +void main() { + late _StubRepository repo; + late _FakeScheduler scheduler; + + SyncEngine build({ + Stream? connectivity, + Stream? downlink, + Random? random, + }) => + SyncEngine( + repository: repo, + connectivity: connectivity, + downlink: downlink, + // Fixed seed so the jitter band is assertable rather than flaky. + random: random ?? Random(7), + scheduleTimer: scheduler.schedule, + ); + + setUp(() { + repo = _StubRepository(); + scheduler = _FakeScheduler(); + }); + + group('single flight', () { + test('overlapping triggers do not start a second pass over the same bills', + () async { + // Without this guarantee a busy till firing a trigger per sale would have + // several drains reading the same pending rows at once, and every bill + // would go up two or three times. + repo.gate = Completer(); + final engine = build(); + + engine + ..nudge(SyncTrigger.saleCommitted) + ..nudge(SyncTrigger.saleCommitted) + ..nudge(SyncTrigger.saleCommitted); + + await Future.delayed(Duration.zero); + expect(repo.calls, 1, reason: 'three triggers must yield one drain'); + + repo.gate!.complete(); + repo.gate = null; + await Future.delayed(Duration.zero); + + await engine.dispose(); + }); + + test('a trigger arriving mid-drain is honoured once that drain finishes', + () async { + // Bills committed during a pass were not in the set it read. Dropping + // the trigger would leave them waiting for the next poll. + repo + ..gate = Completer() + ..pending = 3; + final engine = build(); + + engine.nudge(SyncTrigger.startup); + await Future.delayed(Duration.zero); + expect(repo.calls, 1); + + engine.nudge(SyncTrigger.saleCommitted); + repo.gate!.complete(); + repo.gate = null; + + await Future.delayed(Duration.zero); + await Future.delayed(Duration.zero); + + expect(repo.calls, 2, reason: 'the queued trigger must be replayed'); + await engine.dispose(); + }); + + test('a queued trigger with nothing left owing does not run a second pass', + () async { + // The counterpart to the test above. The drain that just finished emptied + // the queue, so replaying the trigger would send nothing and only churn + // the connection. + repo + ..gate = Completer() + ..pending = 0; + final engine = build(); + + engine.nudge(SyncTrigger.startup); + await Future.delayed(Duration.zero); + + engine.nudge(SyncTrigger.saleCommitted); + repo.gate!.complete(); + repo.gate = null; + await _settle(); + + expect(repo.calls, 1); + await engine.dispose(); + }); + }); + + group('backoff', () { + test('doubles per failure and stops at the ceiling', () { + final engine = SyncEngine( + repository: repo, + baseBackoff: const Duration(seconds: 2), + maxBackoff: const Duration(minutes: 5), + // No jitter, so the shape of the curve is what is being asserted. + random: _ZeroJitter(), + scheduleTimer: scheduler.schedule, + ); + + // 0.8 is the bottom of the jitter band, which _ZeroJitter pins. + expect(engine.backoffFor(1).inMilliseconds, 1600); // 2s + expect(engine.backoffFor(2).inMilliseconds, 3200); // 4s + expect(engine.backoffFor(3).inMilliseconds, 6400); // 8s + expect(engine.backoffFor(9).inSeconds, 240); // 512s → capped + expect(engine.backoffFor(30).inSeconds, 240); // still capped + }); + + test('jitter keeps every delay inside ±20% of the nominal wait', () { + // A shop's terminals all fail at the same instant when the line drops. + // Without jitter they would retry in lockstep and keep colliding. + final engine = build(random: Random(1)); + + final seen = {}; + for (var i = 0; i < 200; i++) { + final ms = engine.backoffFor(4).inMilliseconds; + expect(ms, greaterThanOrEqualTo((16000 * 0.8).round())); + expect(ms, lessThanOrEqualTo((16000 * 1.2).round())); + seen.add(ms); + } + expect(seen.length, greaterThan(50), reason: 'delays must actually vary'); + }); + + test('a failed drain schedules a retry and a success clears it', () async { + repo.scripted.addAll([ + const SyncOutcome(attempted: 2, uploaded: 0, error: 'line dropped'), + const SyncOutcome(attempted: 2, uploaded: 2), + ]); + + final engine = build(); + engine.nudge(SyncTrigger.saleCommitted); + await _settle(); + + expect(engine.state.consecutiveFailures, 1); + expect(engine.state.nextAttemptAt, isNotNull); + expect(scheduler.delays, hasLength(1)); + + scheduler.fireLast(); + await _settle(); + + expect(engine.state.consecutiveFailures, 0); + expect(engine.state.lastError, isNull); + expect(engine.state.nextAttemptAt, isNull); + expect(engine.state.lastSuccessAt, isNotNull); + + await engine.dispose(); + }); + }); + + group('halting', () { + test('a refused batch halts instead of retrying the same bytes forever', + () async { + // A rejection is a decision, not a fault. Re-sending gets the same + // answer, and a loop would bury the one message a person needs to see. + repo.scripted.add(const SyncOutcome( + attempted: 1, + uploaded: 0, + rejected: 1, + error: 'duplicate invoice number', + isRetryable: false, + ),); + + final engine = build(); + engine.nudge(SyncTrigger.saleCommitted); + await _settle(); + + expect(engine.state.isHalted, isTrue); + expect(scheduler.delays, isEmpty, reason: 'no retry may be scheduled'); + + // Further background triggers are ignored while halted. + engine + ..nudge(SyncTrigger.periodic) + ..nudge(SyncTrigger.saleCommitted); + await _settle(); + expect(repo.calls, 1); + + await engine.dispose(); + }); + + test('pressing sync clears a halt and tries again', () async { + // The button is how a cashier retries once the back office is fixed. + repo.scripted.add(const SyncOutcome( + attempted: 1, + uploaded: 0, + error: 'bad credential', + isRetryable: false, + ),); + + final engine = build(); + engine.nudge(SyncTrigger.saleCommitted); + await _settle(); + expect(engine.state.isHalted, isTrue); + + repo.scripted + ..clear() + ..add(const SyncOutcome(attempted: 1, uploaded: 1)); + repo.calls = 0; + + await engine.syncNow(); + expect(repo.calls, 1); + expect(engine.state.isHalted, isFalse); + + await engine.dispose(); + }); + }); + + group('triggers', () { + test('the network coming back starts a drain; going away does not', + () async { + final connectivity = StreamController(); + final engine = build(connectivity: connectivity.stream); + await engine.start(); + + final atStart = repo.calls; + + connectivity.add(false); + await _settle(); + expect(repo.calls, atStart, + reason: 'an attempt certain to fail is not worth making',); + expect(engine.state.online, isFalse); + + connectivity.add(true); + await _settle(); + expect(repo.calls, atStart + 1); + + await connectivity.close(); + await engine.dispose(); + }); + + test('background triggers are ignored while offline, manual is not', + () async { + final connectivity = StreamController(); + final engine = build(connectivity: connectivity.stream); + await engine.start(); + + connectivity.add(false); + await _settle(); + final atStart = repo.calls; + + engine.nudge(SyncTrigger.saleCommitted); + await _settle(); + expect(repo.calls, atStart); + + // The cashier may know something the engine does not, and being told why + // it failed beats a button that does nothing. + await engine.syncNow(); + expect(repo.calls, atStart + 1); + + await connectivity.close(); + await engine.dispose(); + }); + + test('head office can pull a shift up over the downlink', () async { + final downlink = StreamController(); + // Bills owed, otherwise a request to sync correctly does nothing. + repo.pending = 2; + final engine = build(downlink: downlink.stream); + await engine.start(); + // Let the startup drain finish, so this measures the downlink and not a + // race with it. + await _settle(); + + final atStart = repo.calls; + downlink.add(const DownlinkMessage(kind: DownlinkKind.syncRequested)); + await _settle(); + + expect(repo.calls, atStart + 1); + + await downlink.close(); + await engine.dispose(); + }); + + test('an unrecognised downlink message is ignored, not acted on', () async { + final downlink = StreamController(); + repo.pending = 2; + final engine = build(downlink: downlink.stream); + await engine.start(); + await _settle(); + + final atStart = repo.calls; + downlink.add(const DownlinkMessage(kind: DownlinkKind.unknown)); + await _settle(); + + expect(repo.calls, atStart); + + await downlink.close(); + await engine.dispose(); + }); + }); + + test('a repository that throws is treated as a retryable failure, not a crash', + () async { + // Anything escaping the repository is a defect. The engine must still empty + // its queue once the defect is fixed, rather than dying on the first sale. + final engine = SyncEngine( + repository: _ThrowingRepository(), + random: Random(3), + scheduleTimer: scheduler.schedule, + ); + + engine.nudge(SyncTrigger.saleCommitted); + await _settle(); + + expect(engine.state.consecutiveFailures, 1); + expect(engine.state.isHalted, isFalse); + expect(scheduler.delays, hasLength(1)); + + await engine.dispose(); + }); +} + +/// Pins the jitter multiplier at its lower bound so the backoff curve itself +/// can be asserted. +class _ZeroJitter implements Random { + @override + double nextDouble() => 0; + + @override + bool nextBool() => false; + + @override + int nextInt(int max) => 0; +} + +class _ThrowingRepository extends _StubRepository { + @override + Future syncOrders({ + void Function(double progress, String stage)? onProgress, + }) async => + throw StateError('boom'); +} + +/// Lets queued microtasks run. The engine never really waits, so a handful of +/// turns is enough for everything it schedules to settle. +Future _settle() async { + for (var i = 0; i < 6; i++) { + await Future.delayed(Duration.zero); + } +} diff --git a/test/unit/transport_test.dart b/test/unit/transport_test.dart new file mode 100644 index 0000000..17c636a --- /dev/null +++ b/test/unit/transport_test.dart @@ -0,0 +1,276 @@ +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, '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'); + }); + + 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.delayed(Duration.zero); + transport.handleInbound( + config.ackTopic, + jsonEncode({'batch_id': 'a-different-batch', 'accepted': ['order-a']}), + ); + + await expectLater(pending, throwsA(isA())); + }); + + 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': []}); + 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 = []; + 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.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()), + ); + 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() + .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() + .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 = []; + 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() + .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 _receiptFor( + MqttOrderTransport transport, + Map 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 send(http.BaseRequest request) async { + final response = respond(request); + return http.StreamedResponse( + Stream.value(utf8.encode(response.body)), + response.statusCode, + request: request, + ); + } +} diff --git a/test/widget/app_smoke_test.dart b/test/widget/app_smoke_test.dart index 346d2bd..8d38e72 100644 --- a/test/widget/app_smoke_test.dart +++ b/test/widget/app_smoke_test.dart @@ -40,6 +40,11 @@ void main() { await tester.pumpWidget( ProviderScope( overrides: [ + // The background drain would open a broker connection and hit the + // disk on a clock this test controls. Neither is what these tests + // measure, and a half-driven timer would leak into the next one. + syncBootstrapProvider.overrideWith((ref) async {}), + // Catalogue reads come from the in-memory cache and resolve on the // spot, but these four go to SQLite. Real disk I/O cannot be driven // by the fake clock a widget test runs on: sqflite's own lock-warning diff --git a/windows/flutter/generated_plugin_registrant.cc b/windows/flutter/generated_plugin_registrant.cc index 87d9de7..d50e7d3 100644 --- a/windows/flutter/generated_plugin_registrant.cc +++ b/windows/flutter/generated_plugin_registrant.cc @@ -7,12 +7,15 @@ #include "generated_plugin_registrant.h" #include +#include #include #include void RegisterPlugins(flutter::PluginRegistry* registry) { AudioplayersWindowsPluginRegisterWithRegistrar( registry->GetRegistrarForPlugin("AudioplayersWindowsPlugin")); + ConnectivityPlusWindowsPluginRegisterWithRegistrar( + registry->GetRegistrarForPlugin("ConnectivityPlusWindowsPlugin")); PrintingPluginRegisterWithRegistrar( registry->GetRegistrarForPlugin("PrintingPlugin")); UrlLauncherWindowsRegisterWithRegistrar( diff --git a/windows/flutter/generated_plugins.cmake b/windows/flutter/generated_plugins.cmake index 27a28a3..11ad347 100644 --- a/windows/flutter/generated_plugins.cmake +++ b/windows/flutter/generated_plugins.cmake @@ -4,6 +4,7 @@ list(APPEND FLUTTER_PLUGIN_LIST audioplayers_windows + connectivity_plus printing url_launcher_windows )