Files
nearle_pos/lib/presentation/sync/providers/sync_controller.dart
Suriya 30d6d1080f Give every terminal its own identity, and report fleet presence
Answers "which of my 100 tills are alive and healthy", and fixes three things
that were fine on one device and broken on a hundred.

Terminal identity (lib/data/local/terminal_identity.dart)
- Every device mints a UUID on first run, stored in its own database, plus a
  short code (T4A9) derived from it. Renaming keeps the device id, so history
  keeps pointing at the same physical till.
- Replaces the literal 'TERM-01', which was hardcoded in five places. The whole
  fleet reported as one terminal: shift reports merged, MQTT topics collided,
  and a second connection with the same client id evicts the first from the
  broker — so two tills would have knocked each other offline in a loop.

Invoice numbers now carry the terminal code
- INV-2608-T4A9-00042. The sequence counter lives in each till's own database
  and starts at 1, so without this every terminal in the fleet minted
  INV-2608-00001 for its first sale of the month. The order UUID kept the data
  distinct; the number a customer quotes on a receipt was not.

SQLite pragmas
- WAL, so the product grid refreshing does not block the sale being written,
  and the file is never left mid-rewrite by a power cut.
- busy_timeout 5s, so a contended lock waits instead of throwing "database is
  locked" — which at checkout is a failed sale with a customer standing there.
- synchronous NORMAL, the right trade under WAL for a till.

Fleet presence (lib/data/sync/presence_reporter.dart)
- Retained status record on connect and once a minute: device id, code, name,
  app version, pending bill count, last upload, catalogue revision, sync halt
  state. Retained so a dashboard connecting at noon gets all 100 terminals
  immediately rather than a blank board.
- The Last Will already said "reachable". A till can be connected and still be
  holding 200 unsent bills or running last month's prices; only pending_bills
  and catalogue_revision say so.

NATS
- The MQTT gateway maps / to . so the existing transport works unchanged.
  SyncConfig.asNatsSubject() exposes the translation, and the contract doc
  gives the JetStream subjects (pos.*.*.order, pos.*.*.status) plus the two
  server-side requirements: a file-backed stream, and the ack published by the
  consumer after commit rather than by the ingest handler.

Tests: 129 -> 140. New coverage for identity minting and stability, per-device
invoice uniqueness, topic and client-id separation, NATS subject mapping, and
the two pragmas. Suite run three times clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-01 11:17:23 +05:30

247 lines
7.7 KiB
Dart

import 'package:flutter_riverpod/flutter_riverpod.dart';
import '../../../app/providers.dart';
import '../../../core/constants/app_constants.dart';
import '../../../data/remote/mqtt_order_transport.dart';
import '../../../data/sync/presence_reporter.dart';
import '../../../domain/entities/shift_report.dart';
import '../../../domain/entities/sync_event.dart';
import '../../../domain/repositories/sync_repository.dart';
import '../../auth/providers/auth_controller.dart';
import '../../pos/providers/catalog_providers.dart';
/// Bumped after every import so catalogue-backed providers refetch.
final catalogueVersionProvider = StateProvider<int>((ref) => 0);
/// Bumped after every sale or sync so order-backed providers refetch.
final orderVersionProvider = StateProvider<int>((ref) => 0);
/// Whether products exist on this terminal. Billing is gated on it.
final catalogueReadyProvider = Provider<bool>((ref) {
ref.watch(catalogueVersionProvider);
return ref.watch(syncRepositoryProvider).hasCatalogue;
});
final lastImportAtProvider = Provider<DateTime?>((ref) {
ref.watch(catalogueVersionProvider);
return ref.watch(syncRepositoryProvider).lastImportAt;
});
// ------------------------------------------------------- Morning: import
sealed class ImportState {
const ImportState();
}
class ImportIdle extends ImportState {
const ImportIdle();
}
class ImportRunning extends ImportState {
const ImportRunning(this.progress, this.stage);
final double progress;
final String stage;
}
class ImportDone extends ImportState {
const ImportDone(this.event);
final SyncEvent event;
}
class ImportFailed extends ImportState {
const ImportFailed(this.message);
final String message;
}
class CatalogueImportController extends StateNotifier<ImportState> {
CatalogueImportController(this._ref) : super(const ImportIdle());
final Ref _ref;
Future<bool> run() async {
if (state is ImportRunning) return false;
state = const ImportRunning(0, 'Starting…');
final event = await _ref.read(syncRepositoryProvider).importCatalogue(
onProgress: (progress, stage) {
if (mounted) state = ImportRunning(progress, stage);
},
);
if (event.status == SyncStatus.synced) {
_ref.read(catalogueVersionProvider.notifier).state++;
_ref.invalidate(allProductsProvider);
_ref.invalidate(visibleProductsProvider);
_ref.invalidate(categoryCountsProvider);
_ref.invalidate(lowStockProductsProvider);
state = ImportDone(event);
return true;
}
state = ImportFailed(event.error ?? 'Import failed.');
return false;
}
/// Drops a stale success or failure banner.
void reset() => state = const ImportIdle();
}
final catalogueImportProvider =
StateNotifierProvider<CatalogueImportController, ImportState>(
(ref) => CatalogueImportController(ref),
);
// ---------------------------------------------------- Business hours: read
/// Bills still held on this terminal at sync_status = 0.
final unsyncedCountProvider = FutureProvider<int>((ref) {
ref.watch(orderVersionProvider);
return ref.watch(syncRepositoryProvider).unsyncedCount();
});
/// Everything this terminal traded today, across every operator.
final todayReportProvider = FutureProvider<ShiftReport>((ref) {
ref.watch(orderVersionProvider);
final session = ref.watch(cashierSessionProvider);
final user = ref.watch(currentUserProvider);
return ref.watch(syncRepositoryProvider).todayReport(
terminalId: session.terminalId,
cashierName: user?.name ?? session.name,
);
});
/// Only the bills the signed-in operator rang.
///
/// This is the figure a cashier counts their drawer against at the end of a
/// shift, so it must not include anyone else's sales.
final myShiftReportProvider = FutureProvider<ShiftReport>((ref) {
ref.watch(orderVersionProvider);
final session = ref.watch(cashierSessionProvider);
final user = ref.watch(currentUserProvider);
return ref.watch(syncRepositoryProvider).todayReport(
terminalId: session.terminalId,
cashierName: user?.name ?? session.name,
scopeToCashier: true,
);
});
/// Per-order sync state for the events log.
final orderSyncRowsProvider = FutureProvider<List<OrderSyncRow>>((ref) {
ref.watch(orderVersionProvider);
return ref.watch(syncRepositoryProvider).orderSyncRows();
});
final syncEventsProvider = Provider<List<SyncEvent>>((ref) {
ref.watch(orderVersionProvider);
ref.watch(catalogueVersionProvider);
return ref.watch(syncRepositoryProvider).events;
});
// ------------------------------------------------------ End of day: upload
sealed class OrderSyncState {
const OrderSyncState();
}
class SyncIdle extends OrderSyncState {
const SyncIdle();
}
class SyncRunning extends OrderSyncState {
const SyncRunning(this.progress, this.stage);
final double progress;
final String stage;
}
class SyncFinished extends OrderSyncState {
const SyncFinished(this.outcome);
final SyncOutcome outcome;
}
class OrderSyncController extends StateNotifier<OrderSyncState> {
OrderSyncController(this._ref) : super(const SyncIdle());
final Ref _ref;
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<SyncOutcome> run() async {
if (isRunning) {
return const SyncOutcome(attempted: 0, uploaded: 0);
}
state = const SyncRunning(0, 'Starting…');
final outcome = await _ref.read(syncEngineProvider).syncNow(
onProgress: (progress, stage) {
if (mounted) state = SyncRunning(progress, stage);
},
);
_ref.read(orderVersionProvider.notifier).state++;
if (mounted) state = SyncFinished(outcome);
return outcome;
}
void reset() => state = const SyncIdle();
}
final orderSyncProvider =
StateNotifierProvider<OrderSyncController, OrderSyncState>(
(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<void>((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();
// Fleet presence only exists on a transport that can carry it.
final transport = ref.read(orderTransportProvider);
if (transport is MqttOrderTransport) {
final reporter = PresenceReporter(
transport: transport,
terminal: ref.read(terminalIdentityProvider),
config: ref.read(syncConfigProvider),
engine: engine,
appVersion: AppConstants.appVersion,
catalogueRevision: () async =>
ref.read(syncRepositoryProvider).catalogueRevision,
);
ref.onDispose(reporter.dispose);
await reporter.start();
}
});