import 'dart:convert'; import 'package:drift/drift.dart'; import '../models/remote_command.dart'; import '../models/tab_event.dart'; import 'app_database.dart'; class ProductNotFoundException implements Exception { final String productId; const ProductNotFoundException(this.productId); @override String toString() => 'Product not found: $productId'; } /// Atomic stock updates shared by the product and tab services, so they can /// run inside either one's transaction. /// /// Stock may go below zero: the bar keeps serving when the count was off. extension StockQueries on AppDatabase { /// Products with infinite stock are only checked for existence. Future decreaseStock(String productId, int amount) async { _validateStockAmount(amount); if (await changeTrackedStock(productId, -amount) == 0) { await _findProduct(productId); } } Future increaseStock(String productId, int amount) async { _validateStockAmount(amount); if (await changeTrackedStock(productId, amount) == 0) { await _findProduct(productId); } } /// Adds [delta] to the stock of [productId] when its stock is tracked. /// Returns the number of products changed: 0 when the product has infinite /// stock or no longer exists, which is not an error. Future changeTrackedStock(String productId, int delta) { return (update(products) ..where((row) => row.id.equals(productId) & row.infiniteStock.not())) .write( ProductsCompanion.custom( stockQuantity: products.stockQuantity + Variable(delta), ), ); } /// Queues an event that is not about a tab (a stock count, a product /// change) in the outbox. Call it inside the transaction that makes the /// change, so the change and its event are stored together. Future queueEvent(TabEvent event) async { await into(tabEvents).insert( TabEventsCompanion.insert( id: event.id, payload: jsonEncode(event.payload), createdAt: DateTime.now(), ), ); } /// Replaces the local stock of the products in [serverStock] with the /// server's, then replays the outbox events the server hasn't applied yet /// on top, the way the server will. The events the server acknowledged /// must have left the outbox first: they are in [serverStock] already. /// /// Products the server doesn't know keep their local stock. Future rebaseStock(List serverStock) { return transaction(() async { final stock = { for (final entry in serverStock) entry.productId: (stock: entry.stock, countedAt: entry.countedAt), }; // Read here, so changes made while the sync was in flight are included final outbox = await (select(tabEvents)..orderBy([ (row) => OrderingTerm.asc(row.createdAt), (row) => OrderingTerm.asc(const CustomExpression('rowid')), ])) .get(); for (final row in outbox) { final event = jsonDecode(row.payload) as Map; final productId = event['productId'] as String?; final current = stock[productId]; if (productId == null || current == null) continue; final occurredAt = DateTime.tryParse( event['occurredAt'] as String? ?? '', ); if (occurredAt == null) continue; final countedAt = current.countedAt; switch (TabEventType.fromValue(event['type'] as String? ?? '')) { // An older count than the server's is outdated case TabEventType.stockSet: if (countedAt != null && occurredAt.isBefore(countedAt)) continue; stock[productId] = ( stock: event['quantity'] as int?, countedAt: occurredAt, ); // A sale before the count is already in the count case TabEventType.lineAdded || TabEventType.lineRemoved when current.stock != null && (countedAt == null || occurredAt.isAfter(countedAt)): final quantity = event['quantity'] as int? ?? 0; final sign = event['type'] == TabEventType.lineAdded.value ? -1 : 1; stock[productId] = ( stock: current.stock! + sign * quantity, countedAt: countedAt, ); default: continue; } } for (final product in await select(products).get()) { final rebased = stock[product.id]; if (rebased == null) continue; final infinite = rebased.stock == null; // An untracked product keeps its last count for when tracking resumes final quantity = rebased.stock ?? product.stockQuantity; if (product.infiniteStock == infinite && product.stockQuantity == quantity) { continue; } await (update( products, )..where((row) => row.id.equals(product.id))).write( ProductsCompanion( stockQuantity: Value(quantity), infiniteStock: Value(infinite), ), ); } }); } Future _findProduct(String productId) async { final product = await (select( products, )..where((row) => row.id.equals(productId))).getSingleOrNull(); if (product == null) throw ProductNotFoundException(productId); return product; } void _validateStockAmount(int amount) { if (amount <= 0) { throw ArgumentError.value(amount, 'amount', 'Must be greater than zero'); } } }