nearle_daily/nearle_pos
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Upload shopper registrations, and charge GST to the lines that earned it

Browse Source 
Two defects that share a shape: a figure landing on the wrong record.

Bill-level discounts were apportioned across every line by a single
factor, so "20% off Beverages" pulled tax out of the atta line as well.
The bill total was right either way, which is what made it easy to ship
— only the slab split on a filed return was wrong. Targeted campaigns
now reduce the lines they name, and bill-wide reductions still spread
pro rata, so the arithmetic is unchanged wherever it was already right.

Shoppers registered at a till only ever reached the back office as three
fields riding along on a bill. Somebody who signed up and bought nothing
existed on one terminal and nowhere else, and two tills registering the
same mobile each minted their own row. Customers are now an outbox of
their own on pos/{store}/{terminal}/customer, and the id is a UUIDv5
over the normalised mobile number — so a hundred terminals agree on who
a shopper is without talking to each other.

Registrations go up before bills, and a failure there cannot strand a
day's takings. No loyalty figures are sent: they belong to the bill
stream, which is idempotent and knows about every counter.

Two things found while building it. Numbers were keyed on raw digits, so
a cashier typing +91 forked a shopper as effectively as a random id
would. And the sale path wrote the customer with ConflictAlgorithm
.replace, which is a DELETE and an INSERT — every column absent from the
row reverts to its schema default, so the new sync flag would have been
cleared by the shopper's next purchase.

Schema v8. Existing customers are queued rather than assumed sent: the
terminal cannot tell an imported row from a locally registered one, and
only one of those mistakes loses somebody.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Suriya
2026-08-03 11:34:27 +05:30
parent 467d5eee75
commit fe428931ec
24 changed files with 1274 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
lib/data/datasources/local_store.dart
View File

@@ -39,6 +39,7 @@ class LocalStore {
DateTime? _lastImportAt;
String? _catalogueRevision;
int _unsyncedOrders = 0;
int _unsyncedCustomers = 0;
bool _ready = false;
@@ -92,6 +93,7 @@ class LocalStore {
_catalogueRevision = await catalogue.meta(MetaKeys.catalogueRevision);
terminal = await identityStore.load();
_unsyncedOrders = await orders.unsyncedCount();
_unsyncedCustomers = await catalogue.unsyncedCustomerCount();
_syncEvents
..clear()
@@ -239,11 +241,20 @@ class LocalStore {
Future<void> putCustomer(Customer c) async {
await catalogue.upsertCustomer(c);
_customers[c.id] = c;
await refreshUnsyncedCustomerCount();
}
/// Mirrors a customer already written to disk into the memory cache.
void cacheCustomer(Customer c) => _customers[c.id] = c;
/// Shoppers registered here and not yet uploaded.
int get unsyncedCustomers => _unsyncedCustomers;
Future<int> refreshUnsyncedCustomerCount() async {
_unsyncedCustomers = await catalogue.unsyncedCustomerCount();
return _unsyncedCustomers;
}
// ----------------------------------------------------------------- Orders
/// Refreshes the cached unsynced tally after a write or a sync.
Future<int> refreshUnsyncedCount() async {

54
lib/data/local/app_database.dart
View File

@@ -13,7 +13,7 @@ class AppDatabase {
static final AppDatabase instance = AppDatabase._();
static const String _fileName = 'nearle_pos.db';
static const int _version = 7;
static const int _version = 8;
Database? _db;
@@ -77,6 +77,7 @@ class AppDatabase {
'ALTER TABLE ${Tables.orders} ADD COLUMN promos_json TEXT',
);
}
if (from < 8) await _upgradeToV8(db);
},
),
);
@@ -188,12 +189,24 @@ class AppDatabase {
lifetime_spend REAL NOT NULL DEFAULT 0,
visit_count INTEGER NOT NULL DEFAULT 0,
created_at INTEGER,
last_visit_at INTEGER
last_visit_at INTEGER,
-- Registration outbox. A shopper signed up at the till has to reach
-- the back office even if they never buy anything, so this table
-- carries the same pending/synced flag the orders table does.
--
-- Defaults to 1: rows that arrived in a catalogue pull came *from*
-- the back office and must not be posted straight back.
sync_status INTEGER NOT NULL DEFAULT 1,
synced_at INTEGER
)
''');
await db.execute(
'CREATE UNIQUE INDEX idx_customers_mobile ON ${Tables.customers}(mobile)',
);
await db.execute(
'CREATE INDEX idx_customers_sync ON ${Tables.customers}(sync_status)',
);
// -------------------------------------------------------------- orders
await db.execute('''
@@ -322,6 +335,43 @@ Future<void> _upgradeToV4(Database db, {required int from}) async {
await db.execute('DROP TABLE _day_archive_v3');
}
/// Moves the schema to v8 — customers become an outbox.
///
/// Before this, a shopper registered at the till only ever reached the back
/// office as three fields riding along on a bill. Someone who signed up and
/// then didn't buy anything, or whose bill was still queued, existed on one
/// terminal and nowhere else.
///
/// Every existing row is marked pending rather than synced. The terminal
/// cannot tell which of them came down in a catalogue pull and which were rung
/// up locally, and of the two possible mistakes only one loses a shopper. This
/// is safe precisely because the customer uplink is specified as
/// insert-if-absent on id — re-sending one the back office already holds is a
/// no-op, never an overwrite of a profile edited at head office.
///
/// Ids are deliberately *not* rewritten. New customers are keyed on their
/// mobile number (see `Customer.idForMobile`) so terminals agree without
/// coordinating, but rows created before this version carry random ids that
/// bills already in the back office refer to. Re-keying them here would break
/// that link. They stay as they are, and the back office merges them on
/// mobile — a one-off for stores that were already trading.
Future<void> _upgradeToV8(Database db) async {
await db.execute(
'ALTER TABLE ${Tables.customers} '
'ADD COLUMN sync_status INTEGER NOT NULL DEFAULT 1',
);
await db.execute(
'ALTER TABLE ${Tables.customers} ADD COLUMN synced_at INTEGER',
);
await db.execute(
'CREATE INDEX idx_customers_sync ON ${Tables.customers}(sync_status)',
);
await db.rawUpdate(
'UPDATE ${Tables.customers} SET sync_status = 0, synced_at = NULL',
);
}
const String _createSyncLog = '''
CREATE TABLE sync_log (
id TEXT PRIMARY KEY,

66
lib/data/local/catalogue_dao.dart
View File

@@ -164,7 +164,7 @@ class CatalogueDao {
for (final c in customers) {
batch.insert(
Tables.customers,
customerToRow(c),
_importedCustomerRow(c),
conflictAlgorithm: ConflictAlgorithm.ignore,
);
}
@@ -173,6 +173,18 @@ class CatalogueDao {
});
}
/// A customer row that arrived from the back office, marked as already sent.
///
/// Stated outright rather than left to the column default, because the
/// default existing to protect imports is not obvious from this call site —
/// and getting it wrong would post the whole customer book straight back to
/// the server that just sent it.
static Map<String, Object?> _importedCustomerRow(Customer c) => {
...customerToRow(c),
'sync_status': syncedCustomer,
'synced_at': DateTime.now().millisecondsSinceEpoch,
};
/// Applies a change set, leaving everything it does not mention alone.
///
/// The counterpart to [replaceCatalogue], and the difference matters: a full
@@ -218,7 +230,7 @@ class CatalogueDao {
for (final c in customers) {
batch.insert(
Tables.customers,
customerToRow(c),
_importedCustomerRow(c),
conflictAlgorithm: ConflictAlgorithm.ignore,
);
}
@@ -254,11 +266,10 @@ class CatalogueDao {
}
Future<Customer?> customerByMobile(String mobile) async {
final digits = mobile.replaceAll(RegExp(r'\D'), '');
final rows = await _db.query(
Tables.customers,
where: 'mobile = ?',
whereArgs: [digits],
whereArgs: [Customer.normaliseMobile(mobile)],
limit: 1,
);
return rows.isEmpty ? null : customerFromRow(rows.first);
@@ -274,14 +285,59 @@ class CatalogueDao {
return rows.isEmpty ? null : customerFromRow(rows.first);
}
/// Writes a customer created or edited at the till, and queues them.
///
/// Everything registered on this terminal has to reach the back office in its
/// own right — a shopper who signs up for the loyalty scheme and then buys
/// nothing used to exist here and nowhere else.
Future<void> upsertCustomer(Customer c) async {
await _db.insert(
Tables.customers,
customerToRow(c),
{...customerToRow(c), 'sync_status': pendingCustomer, 'synced_at': null},
conflictAlgorithm: ConflictAlgorithm.replace,
);
}
// ---------------------------------------------------- Customer outbox
static const int pendingCustomer = 0;
static const int syncedCustomer = 1;
Future<int> unsyncedCustomerCount() async {
final rows = await _db.rawQuery(
'SELECT COUNT(*) AS n FROM ${Tables.customers} WHERE sync_status = ?',
[pendingCustomer],
);
return (rows.first['n']! as num).toInt();
}
/// A bounded page of shoppers waiting to go up, oldest first.
Future<List<Customer>> unsyncedCustomers({int limit = 100}) async {
final rows = await _db.query(
Tables.customers,
where: 'sync_status = ?',
whereArgs: [pendingCustomer],
orderBy: 'created_at ASC',
limit: limit,
);
return rows.map(customerFromRow).toList();
}
/// Flips only the ids the back office named. Anything it stayed silent about
/// is left pending — the same rule the orders outbox follows.
Future<void> markCustomersSynced(List<String> ids, {DateTime? at}) async {
if (ids.isEmpty) return;
final marks = List.filled(ids.length, '?').join(',');
await _db.rawUpdate(
'UPDATE ${Tables.customers} SET sync_status = ?, synced_at = ? '
'WHERE id IN ($marks)',
[
syncedCustomer,
(at ?? DateTime.now()).millisecondsSinceEpoch,
...ids,
],
);
}
// ------------------------------------------------------------------ Meta
Future<String?> meta(String key) async {
final rows = await _db.query(

20
lib/data/local/order_dao.dart
View File

@@ -61,10 +61,24 @@ class OrderDao {
);
});
if (customerRow != null) {
batch.insert(
// A targeted UPDATE, not an upsert. `ConflictAlgorithm.replace` is a
// DELETE followed by an INSERT, so every column absent from the row
// silently reverts to its schema default — which would reset
// `sync_status` to 1 and strand a shopper who had never been uploaded.
//
// Restricting it to the four figures a sale actually moves also stops
// a bill overwriting a name or number corrected at head office between
// the shopper being added to the cart and the cashier taking payment.
batch.update(
Tables.customers,
customerRow,
conflictAlgorithm: ConflictAlgorithm.replace,
{
'loyalty_points': customerRow['loyalty_points'],
'lifetime_spend': customerRow['lifetime_spend'],
'visit_count': customerRow['visit_count'],
'last_visit_at': customerRow['last_visit_at'],
},
where: 'id = ?',
whereArgs: [customerRow['id']],
);
}
await batch.commit(noResult: true);

27
lib/data/remote/http_order_transport.dart
View File

@@ -50,8 +50,19 @@ class HttpOrderTransport implements OrderTransport {
Future<void> connect() async {}
@override
Future<PushReceipt> pushOrders(List<Map<String, Object?>> orders) async {
if (orders.isEmpty) return const PushReceipt(accepted: []);
Future<PushReceipt> pushOrders(List<Map<String, Object?>> orders) =>
_post(path: 'orders', key: 'orders', items: orders);
@override
Future<PushReceipt> pushCustomers(List<Map<String, Object?>> customers) =>
_post(path: 'customers', key: 'customers', items: customers);
Future<PushReceipt> _post({
required String path,
required String key,
required List<Map<String, Object?>> items,
}) async {
if (items.isEmpty) return const PushReceipt(accepted: []);
if (config.httpBaseUrl.isEmpty) {
throw const TransportException(
@@ -60,7 +71,7 @@ class HttpOrderTransport implements OrderTransport {
);
}
final uri = Uri.parse('${config.httpBaseUrl}/orders');
final uri = Uri.parse('${config.httpBaseUrl}/$path');
http.Response response;
try {
@@ -73,14 +84,14 @@ class HttpOrderTransport implements OrderTransport {
'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),
'idempotency-key': _batchKey(items),
},
body: jsonEncode({
'schema': 1,
'store_id': config.storeId,
'terminal_id': config.terminalId,
'sent_at': DateTime.now().toIso8601String(),
'orders': orders,
key: items,
}),
)
.timeout(config.ackTimeout);
@@ -132,10 +143,10 @@ class HttpOrderTransport implements OrderTransport {
return PushReceipt(accepted: accepted, rejected: rejected);
}
/// Stable for a given set of bills, so a retry after a timeout carries the
/// Stable for a given set of records, so a retry after a timeout carries the
/// same key as the attempt that may already have landed.
String _batchKey(List<Map<String, Object?>> orders) =>
orders.map((o) => o['id']).join('|').hashCode.toRadixString(16);
String _batchKey(List<Map<String, Object?>> items) =>
items.map((o) => o['id']).join('|').hashCode.toRadixString(16);
void _setReachable(bool value) {
if (_reachable == value) return;

38
lib/data/remote/mqtt_order_transport.dart
View File

@@ -170,8 +170,36 @@ class MqttOrderTransport implements OrderTransport {
// ----------------------------------------------------------------- Uplink
@override
Future<PushReceipt> pushOrders(List<Map<String, Object?>> orders) async {
if (orders.isEmpty) return const PushReceipt(accepted: []);
Future<PushReceipt> pushOrders(List<Map<String, Object?>> orders) =>
_publishBatch(
topic: config.orderTopic,
key: 'orders',
items: orders,
noun: 'bills',
);
@override
Future<PushReceipt> pushCustomers(List<Map<String, Object?>> customers) =>
_publishBatch(
topic: config.customerTopic,
key: 'customers',
items: customers,
noun: 'registrations',
);
/// Publishes one correlated batch and waits for the back office to answer it.
///
/// Shared by both uplinks because the rule they must obey is the same one,
/// and it is the rule the whole design rests on: a batch counts as delivered
/// only when the *application* names its ids, never when the broker
/// acknowledges the bytes.
Future<PushReceipt> _publishBatch({
required String topic,
required String key,
required List<Map<String, Object?>> items,
required String noun,
}) async {
if (items.isEmpty) return const PushReceipt(accepted: []);
await connect();
@@ -181,14 +209,14 @@ class MqttOrderTransport implements OrderTransport {
try {
_publish(
config.orderTopic,
topic,
jsonEncode({
'schema': 1,
'batch_id': batchId,
'store_id': config.storeId,
'terminal_id': config.terminalId,
'sent_at': DateTime.now().toIso8601String(),
'orders': orders,
key: items,
}),
);
@@ -196,7 +224,7 @@ class MqttOrderTransport implements OrderTransport {
config.ackTimeout,
onTimeout: () => throw TransportException(
'The back office did not confirm the batch within '
'${config.ackTimeout.inSeconds}s. The bills are still on this '
'${config.ackTimeout.inSeconds}s. The $noun are still on this '
'terminal and will be sent again.',
),
);

11
lib/data/remote/order_transport.dart
View File

@@ -86,6 +86,17 @@ abstract class OrderTransport {
/// Throws [TransportException] when the outcome is unknown.
Future<PushReceipt> pushOrders(List<Map<String, Object?>> orders);
/// Hands over shoppers registered at this till.
///
/// Same acceptance contract as [pushOrders] — only ids the back office names
/// are marked sent — but the payload is a registration rather than a
/// financial record, so the back office is expected to treat it as
/// insert-if-absent on id. Replaying one it already holds must be a no-op,
/// never an overwrite of a profile corrected at head office.
///
/// Throws [TransportException] when the outcome is unknown.
Future<PushReceipt> pushCustomers(List<Map<String, Object?>> customers);
/// Cloud-initiated messages. Empty for transports that cannot receive.
Stream<DownlinkMessage> get downlink;

20
lib/data/remote/simulated_order_transport.dart
View File

@@ -33,20 +33,30 @@ class SimulatedOrderTransport implements OrderTransport {
Future<void> connect() async {}
@override
Future<PushReceipt> pushOrders(List<Map<String, Object?>> orders) async {
Future<PushReceipt> pushOrders(List<Map<String, Object?>> orders) =>
_accept(orders, 'bill');
@override
Future<PushReceipt> pushCustomers(List<Map<String, Object?>> customers) =>
_accept(customers, 'registration');
Future<PushReceipt> _accept(
List<Map<String, Object?>> items,
String noun,
) async {
await Future<void>.delayed(
Duration(milliseconds: 400 + orders.length * 60),
Duration(milliseconds: 400 + items.length * 60),
);
if (isOffline()) {
throw const TransportException(
throw TransportException(
'Simulate offline is ON in Settings, so the upload was failed on '
'purpose. Every bill is still stored on this terminal.',
'purpose. Every $noun is still stored on this terminal.',
);
}
return PushReceipt(
accepted: orders.map((o) => o['id']! as String).toList(),
accepted: items.map((o) => o['id']! as String).toList(),
);
}

31
lib/data/repositories/customer_repository_impl.dart
View File

@@ -1,5 +1,3 @@
import 'package:uuid/uuid.dart';
import '../../core/utils/extensions.dart';
import '../../domain/entities/customer.dart';
import '../../domain/repositories/customer_repository.dart';
@@ -9,15 +7,15 @@ class CustomerRepositoryImpl implements CustomerRepository {
CustomerRepositoryImpl(this._store);
final LocalStore _store;
static const _uuid = Uuid();
String _digits(String v) => v.replaceAll(RegExp(r'\D'), '');
@override
Future<Customer?> findByMobile(String mobile) async {
final needle = _digits(mobile);
return _store.customers
.firstWhereOrNull((c) => _digits(c.mobile) == needle);
// Normalised on both sides, so a shopper stored from `9840012345` is still
// found when a cashier at the next till types `+91 98400 12345`.
final needle = Customer.normaliseMobile(mobile);
return _store.customers.firstWhereOrNull(
(c) => Customer.normaliseMobile(c.mobile) == needle,
);
}
@override
@@ -30,9 +28,14 @@ class CustomerRepositoryImpl implements CustomerRepository {
throw StateError('A customer with this mobile number already exists.');
}
final created = Customer(
id: _uuid.v4(),
// Derived from the number, not random — see [Customer.idForMobile].
// Two tills registering the same shopper independently produce the same
// row rather than a duplicate the back office has to reconcile.
id: Customer.idForMobile(customer.mobile),
name: customer.name.trim(),
mobile: _digits(customer.mobile),
// Stored normalised, so the unique index on `mobile` actually catches a
// second attempt to register the same shopper.
mobile: Customer.normaliseMobile(customer.mobile),
email: customer.email?.trim().isEmpty ?? true
? null
: customer.email!.trim(),
@@ -60,11 +63,15 @@ class CustomerRepositoryImpl implements CustomerRepository {
// Stored numbers are digits only, so the query has to be reduced the same
// way — otherwise a cashier typing "98765 43210" or "98-76" matches nothing.
final digits = _digits(q);
//
// Raw digits rather than the normalised form on purpose: this is a partial
// match on whatever has been typed so far, and a half-entered number is not
// a number to be normalised.
final digits = Customer.digitsOf(q);
return _store.customers.where((c) {
if (c.name.toLowerCase().contains(q)) return true;
return digits.isNotEmpty && _digits(c.mobile).contains(digits);
return digits.isNotEmpty && Customer.digitsOf(c.mobile).contains(digits);
}).toList();
}

104
lib/data/repositories/sync_repository_impl.dart
View File

@@ -3,6 +3,7 @@ import 'dart:convert';
import 'package:uuid/uuid.dart';
import '../../core/utils/formatters.dart';
import '../../domain/entities/customer.dart';
import '../../domain/entities/shift_report.dart';
import '../../domain/entities/sync_event.dart';
import '../../domain/entities/transaction.dart';
@@ -333,6 +334,109 @@ class SyncRepositoryImpl implements SyncRepository {
DateTime.now().subtract(OrderDao.retentionWindow),
);
// ------------------------------------------------- Registrations: uplink
@override
Future<int> unsyncedCustomerCount() =>
_store.catalogue.unsyncedCustomerCount();
@override
Future<SyncOutcome> syncCustomers() async {
final started = DateTime.now();
var attempted = 0;
var uploaded = 0;
while (true) {
final batch = await _store.catalogue.unsyncedCustomers(limit: batchSize);
if (batch.isEmpty) break;
attempted += batch.length;
PushReceipt receipt;
try {
receipt = await _transport.pushCustomers(
batch.map(_customerToPayload).toList(),
);
} on Object catch (e) {
// Nothing is marked sent when the outcome is unknown. Unlike a bill,
// a registration is safe to send twice, so this simply waits for the
// next pass rather than needing a per-row attempt counter.
if (attempted > batch.length || uploaded > 0) {
await _store.refreshUnsyncedCustomerCount();
}
return SyncOutcome(
attempted: attempted,
uploaded: uploaded,
error: e.toString(),
isRetryable: e is! TransportException || e.retryable,
);
}
final acceptedIds = receipt.accepted.toSet();
await _store.catalogue.markCustomersSynced(acceptedIds.toList());
uploaded += acceptedIds.length;
// Nothing moved, so the next page would hand back the same rows for
// ever. Stop and let the events log show why.
if (acceptedIds.isEmpty) {
final reasons = receipt.rejected.values.toSet().join('; ');
await _log(SyncEvent(
id: _uuid.v4(),
type: SyncEventType.catalogueImport,
status: SyncStatus.failed,
createdAt: started,
summary: '${batch.length} registrations were not accepted',
error:
reasons.isEmpty ? 'Not confirmed by the back office' : reasons,
attempts: 1,
),);
await _store.refreshUnsyncedCustomerCount();
return SyncOutcome(
attempted: attempted,
uploaded: uploaded,
rejected: batch.length,
error: 'No registration in this batch was accepted'
'${reasons.isEmpty ? '' : ': $reasons'}',
isRetryable: false,
);
}
}
await _store.refreshUnsyncedCustomerCount();
if (uploaded > 0) {
await _log(SyncEvent(
id: _uuid.v4(),
type: SyncEventType.catalogueImport,
status: SyncStatus.synced,
createdAt: started,
syncedAt: DateTime.now(),
summary: '$uploaded shopper registrations uploaded '
'via ${_transport.label}',
attempts: 1,
),);
}
return SyncOutcome(attempted: attempted, uploaded: uploaded);
}
/// The JSON body sent per registration.
///
/// Identity and profile only. Points, spend and visit counts are deliberately
/// left out: they are derived from the bill stream, which is authoritative and
/// idempotent. Uploading a terminal's local balance would make the last till
/// to sync win, and a shopper who bought something at two counters on the same
/// day would end up with whichever figure arrived second.
Map<String, Object?> _customerToPayload(Customer c) => {
'id': c.id,
'mobile': c.mobile,
'name': c.name,
'email': c.email,
'gender': c.gender.name,
'date_of_birth': c.dateOfBirth?.toIso8601String(),
'registered_at': c.createdAt?.toIso8601String(),
'registered_by_terminal': _store.terminal.code,
};
/// The JSON body sent per order.
Map<String, Object?> _orderToPayload(SaleTransaction t) => {
'id': t.id,

11
lib/data/sync/sync_engine.dart
View File

@@ -250,6 +250,17 @@ class SyncEngine {
SyncOutcome outcome;
try {
try {
// Registrations first, so a bill naming a shopper the back office has
// never heard of arrives after the shopper does. A failure here is
// logged and swallowed: shoppers waiting to go up must never be the
// reason a day's takings stay on the terminal.
try {
await _repository.syncCustomers();
} on Object {
// Deliberately ignored — the next pass tries again, and the events
// log already carries the reason.
}
outcome = await _repository.syncOrders(onProgress: onProgress);
} on Object catch (e) {
// The repository is meant to fold failures into the outcome; anything