feat: add remote pull and push
This commit is contained in:
@@ -87,6 +87,7 @@ class DriftBarTabService implements BarTabService {
|
||||
status: row.status,
|
||||
openedAt: row.openedAt,
|
||||
closedAt: row.closedAt,
|
||||
version: row.version,
|
||||
items: items,
|
||||
);
|
||||
}
|
||||
@@ -292,6 +293,7 @@ class DriftBarTabService implements BarTabService {
|
||||
@override
|
||||
Future<BarTab> createTab({required String customerName}) async {
|
||||
final id = _uuid.v4();
|
||||
final now = DateTime.now();
|
||||
|
||||
await database
|
||||
.into(database.barTabs)
|
||||
@@ -300,7 +302,8 @@ class DriftBarTabService implements BarTabService {
|
||||
id: id,
|
||||
customerName: customerName,
|
||||
status: const Value('open'),
|
||||
openedAt: DateTime.now(),
|
||||
openedAt: now,
|
||||
version: const Value(1),
|
||||
),
|
||||
);
|
||||
|
||||
@@ -335,6 +338,8 @@ class DriftBarTabService implements BarTabService {
|
||||
if (updatedRows != 1) {
|
||||
throw StateError('Tab not found.');
|
||||
}
|
||||
|
||||
await _incrementTabVersion(tabId);
|
||||
}
|
||||
|
||||
@override
|
||||
@@ -368,6 +373,14 @@ class DriftBarTabService implements BarTabService {
|
||||
});
|
||||
}
|
||||
|
||||
Future<void> _incrementTabVersion(String tabId) async {
|
||||
await database.customUpdate(
|
||||
'UPDATE bar_tabs SET version = version + 1 WHERE id = ?',
|
||||
variables: [Variable.withString(tabId)],
|
||||
updates: {database.barTabs},
|
||||
);
|
||||
}
|
||||
|
||||
@override
|
||||
Future<void> addProductToTab({
|
||||
required String tabId,
|
||||
@@ -414,6 +427,8 @@ class DriftBarTabService implements BarTabService {
|
||||
|
||||
await _recordPurchase(tabItemId: tabItemId);
|
||||
});
|
||||
|
||||
await _incrementTabVersion(tabId);
|
||||
}
|
||||
|
||||
@override
|
||||
@@ -421,7 +436,9 @@ class DriftBarTabService implements BarTabService {
|
||||
required String tabItemId,
|
||||
required int delta,
|
||||
}) async {
|
||||
return database.transaction(() async {
|
||||
String? tabId;
|
||||
|
||||
final actualDelta = await database.transaction(() async {
|
||||
final item = await (database.select(
|
||||
database.tabItems,
|
||||
)..where((row) => row.id.equals(tabItemId))).getSingleOrNull();
|
||||
@@ -430,6 +447,8 @@ class DriftBarTabService implements BarTabService {
|
||||
return 0;
|
||||
}
|
||||
|
||||
tabId = item.tabId;
|
||||
|
||||
final requestedQuantity = item.quantity + delta;
|
||||
final actualDelta = requestedQuantity <= 0 ? -item.quantity : delta;
|
||||
|
||||
@@ -483,6 +502,12 @@ class DriftBarTabService implements BarTabService {
|
||||
|
||||
return actualDelta;
|
||||
});
|
||||
|
||||
if (tabId != null && actualDelta != 0) {
|
||||
await _incrementTabVersion(tabId!);
|
||||
}
|
||||
|
||||
return actualDelta;
|
||||
}
|
||||
|
||||
@override
|
||||
@@ -544,6 +569,8 @@ class DriftBarTabService implements BarTabService {
|
||||
|
||||
await deleteQuery.go();
|
||||
});
|
||||
|
||||
await _incrementTabVersion(tabId);
|
||||
}
|
||||
|
||||
@override
|
||||
|
||||
@@ -0,0 +1,418 @@
|
||||
import 'dart:async';
|
||||
import 'dart:convert';
|
||||
|
||||
import 'package:drift/drift.dart';
|
||||
import 'package:flutter/foundation.dart';
|
||||
import 'package:flutter_secure_storage/flutter_secure_storage.dart';
|
||||
import 'package:http/http.dart' as http;
|
||||
import 'package:sentry_flutter/sentry_flutter.dart';
|
||||
|
||||
import '../database/app_database.dart';
|
||||
import '../models/payment_method.dart';
|
||||
import '../models/remote_command.dart';
|
||||
import '../utils/app_config.dart';
|
||||
import 'bar_tab_service.dart';
|
||||
|
||||
class RemoteSyncService {
|
||||
final AppDatabase database;
|
||||
final BarTabService barTabService;
|
||||
final FlutterSecureStorage secureStorage;
|
||||
|
||||
RemoteSyncService({
|
||||
required this.database,
|
||||
required this.barTabService,
|
||||
FlutterSecureStorage? secureStorage,
|
||||
}) : secureStorage = secureStorage ?? const FlutterSecureStorage();
|
||||
|
||||
static const _deviceIdKey = 'remote_device_id';
|
||||
static const _authTokenKey = 'remote_auth_token';
|
||||
static const _serverUrlKey = 'remote_server_url';
|
||||
static const _venueIdKey = 'remote_venue_id';
|
||||
|
||||
final _eventController = StreamController<RemoteSyncEvent>.broadcast();
|
||||
|
||||
Stream<RemoteSyncEvent> get events => _eventController.stream;
|
||||
|
||||
bool _isSyncing = false;
|
||||
bool get isSyncing => _isSyncing;
|
||||
|
||||
Timer? _syncTimer;
|
||||
int _consecutiveFailures = 0;
|
||||
bool _paused = false;
|
||||
|
||||
Future<String?> get deviceId async =>
|
||||
await secureStorage.read(key: _deviceIdKey);
|
||||
|
||||
Future<String?> get authToken async =>
|
||||
await secureStorage.read(key: _authTokenKey);
|
||||
|
||||
Future<String?> get venueId async =>
|
||||
await secureStorage.read(key: _venueIdKey);
|
||||
|
||||
Future<bool> get isConfigured async {
|
||||
final token = await authToken;
|
||||
final vid = await venueId;
|
||||
return token != null && token.isNotEmpty && vid != null && vid.isNotEmpty;
|
||||
}
|
||||
|
||||
String get _baseUrl => kRemoteServerUrl;
|
||||
|
||||
void startPeriodicSync({int intervalSeconds = 90}) {
|
||||
_syncTimer?.cancel();
|
||||
_syncTimer = Timer.periodic(
|
||||
Duration(seconds: intervalSeconds),
|
||||
(_) => _onTick(),
|
||||
);
|
||||
}
|
||||
|
||||
void stopPeriodicSync() {
|
||||
_syncTimer?.cancel();
|
||||
_syncTimer = null;
|
||||
}
|
||||
|
||||
void pause() {
|
||||
_paused = true;
|
||||
}
|
||||
|
||||
void resume() {
|
||||
_paused = false;
|
||||
}
|
||||
|
||||
Future<void> _onTick() async {
|
||||
if (_isSyncing || _paused) return;
|
||||
await sync();
|
||||
}
|
||||
|
||||
Future<void> sync() async {
|
||||
if (_isSyncing) return;
|
||||
|
||||
final configured = await isConfigured;
|
||||
if (!configured) return;
|
||||
|
||||
_isSyncing = true;
|
||||
|
||||
try {
|
||||
await _doSync();
|
||||
_consecutiveFailures = 0;
|
||||
_eventController.add(const RemoteSyncEvent(
|
||||
type: RemoteSyncEventType.syncCompleted,
|
||||
));
|
||||
} catch (e, stack) {
|
||||
_consecutiveFailures++;
|
||||
debugPrint('RemoteSyncService: sync failed: $e');
|
||||
Sentry.captureException(e, stackTrace: stack);
|
||||
_eventController.add(RemoteSyncEvent(
|
||||
type: RemoteSyncEventType.syncFailed,
|
||||
message: e.toString(),
|
||||
));
|
||||
_applyBackoff();
|
||||
} finally {
|
||||
_isSyncing = false;
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> _doSync() async {
|
||||
final token = await authToken;
|
||||
if (token == null) return;
|
||||
|
||||
final deviceId = await this.deviceId;
|
||||
if (deviceId == null) return;
|
||||
|
||||
final state = await _getSyncState();
|
||||
final openTabs = await _buildSnapshot();
|
||||
|
||||
final request = SyncRequest(
|
||||
deviceId: deviceId,
|
||||
cursor: state?.serverCursor,
|
||||
openTabs: openTabs,
|
||||
);
|
||||
|
||||
final response = await _httpPost(
|
||||
'$_baseUrl/devices/$deviceId/sync',
|
||||
body: request.toJson(),
|
||||
token: token,
|
||||
);
|
||||
|
||||
if (response == null) return;
|
||||
|
||||
final syncResponse = SyncResponse.fromJson(response);
|
||||
|
||||
for (final command in syncResponse.commands) {
|
||||
await _processCommand(command);
|
||||
}
|
||||
|
||||
for (final name in syncResponse.plannedTabNames) {
|
||||
final hasOpenTab = openTabs.any((t) => t.customerName == name);
|
||||
if (!hasOpenTab) {
|
||||
await barTabService.createTab(customerName: name);
|
||||
}
|
||||
}
|
||||
|
||||
if (syncResponse.cursor != null) {
|
||||
await _saveCursor(syncResponse.cursor!);
|
||||
}
|
||||
|
||||
await _updateSyncState(success: true);
|
||||
}
|
||||
|
||||
Future<List<SnapshotTab>> _buildSnapshot() async {
|
||||
final openTabs = await barTabService.getOpenTabs();
|
||||
return openTabs.map((tab) => SnapshotTab(
|
||||
id: tab.id,
|
||||
customerName: tab.customerName,
|
||||
version: tab.version,
|
||||
totalInCents: tab.totalInCents,
|
||||
itemCount: tab.itemCount,
|
||||
)).toList();
|
||||
}
|
||||
|
||||
Future<void> _processCommand(RemoteCommand command) async {
|
||||
final alreadyProcessed = await _isCommandProcessed(command.id);
|
||||
if (alreadyProcessed) return;
|
||||
|
||||
if (command.isExpired) {
|
||||
await _ackCommand(command.id, 'expired', 'Command expired');
|
||||
await _recordProcessed(command.id, command.type, 'expired');
|
||||
return;
|
||||
}
|
||||
|
||||
if (command.type != 'close_tab') {
|
||||
await _ackCommand(command.id, 'failed', 'Unknown command type: ${command.type}');
|
||||
await _recordProcessed(command.id, command.type, 'failed');
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
final tab = await barTabService.getTabById(command.tabId);
|
||||
|
||||
if (tab == null) {
|
||||
await _ackCommand(command.id, 'conflict', 'Tab not found');
|
||||
await _recordProcessed(command.id, command.type, 'conflict');
|
||||
_eventController.add(const RemoteSyncEvent(
|
||||
type: RemoteSyncEventType.conflict,
|
||||
message: 'Tab not found',
|
||||
));
|
||||
return;
|
||||
}
|
||||
|
||||
if (!tab.isOpen) {
|
||||
await _ackCommand(command.id, 'conflict', 'Tab already closed');
|
||||
await _recordProcessed(command.id, command.type, 'conflict');
|
||||
_eventController.add(const RemoteSyncEvent(
|
||||
type: RemoteSyncEventType.conflict,
|
||||
message: 'Tab already closed',
|
||||
));
|
||||
return;
|
||||
}
|
||||
|
||||
if (tab.version != command.expectedVersion) {
|
||||
await _ackCommand(command.id, 'conflict', 'Version mismatch');
|
||||
await _recordProcessed(command.id, command.type, 'conflict');
|
||||
_eventController.add(const RemoteSyncEvent(
|
||||
type: RemoteSyncEventType.conflict,
|
||||
message: 'Tab was modified since the request',
|
||||
));
|
||||
return;
|
||||
}
|
||||
|
||||
final method = PaymentMethod.fromValue(command.paymentMethod);
|
||||
await barTabService.closeTab(command.tabId, paymentMethod: method);
|
||||
|
||||
await _ackCommand(command.id, 'completed');
|
||||
await _recordProcessed(command.id, command.type, 'completed');
|
||||
|
||||
_eventController.add(RemoteSyncEvent(
|
||||
type: RemoteSyncEventType.tabClosedRemotely,
|
||||
tabCustomerName: tab.customerName,
|
||||
requestedBy: command.requestedBy,
|
||||
));
|
||||
} catch (e, stack) {
|
||||
debugPrint('RemoteSyncService: command processing failed: $e');
|
||||
Sentry.captureException(e, stackTrace: stack);
|
||||
await _ackCommand(command.id, 'failed', e.toString());
|
||||
await _recordProcessed(command.id, command.type, 'failed');
|
||||
}
|
||||
}
|
||||
|
||||
Future<Map<String, dynamic>?> _httpPost(
|
||||
String url, {
|
||||
required Map<String, dynamic> body,
|
||||
required String token,
|
||||
}) async {
|
||||
final response = await http.post(
|
||||
Uri.parse(url),
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': 'Bearer $token',
|
||||
},
|
||||
body: jsonEncode(body),
|
||||
).timeout(const Duration(seconds: 15));
|
||||
|
||||
if (response.statusCode == 401) {
|
||||
await secureStorage.delete(key: _authTokenKey);
|
||||
debugPrint('RemoteSyncService: token revoked');
|
||||
return null;
|
||||
}
|
||||
|
||||
if (response.statusCode == 429) {
|
||||
final retryAfter = response.headers['retry-after'];
|
||||
if (retryAfter != null) {
|
||||
final seconds = int.tryParse(retryAfter) ?? 60;
|
||||
_applyBackoff(forcedDelay: seconds);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
if (response.statusCode < 200 || response.statusCode >= 300) {
|
||||
debugPrint('RemoteSyncService: HTTP ${response.statusCode}: ${response.body}');
|
||||
return null;
|
||||
}
|
||||
|
||||
if (response.body.isEmpty) return {};
|
||||
|
||||
return jsonDecode(response.body) as Map<String, dynamic>;
|
||||
}
|
||||
|
||||
Future<void> _ackCommand(
|
||||
String commandId,
|
||||
String status, [
|
||||
String? result,
|
||||
]) async {
|
||||
final token = await authToken;
|
||||
final deviceId = await this.deviceId;
|
||||
if (token == null || deviceId == null) return;
|
||||
|
||||
try {
|
||||
final ack = CommandAck(commandId: commandId, status: status, result: result);
|
||||
await _httpPost(
|
||||
'$_baseUrl/devices/$deviceId/commands/$commandId/ack',
|
||||
body: ack.toJson(),
|
||||
token: token,
|
||||
);
|
||||
} catch (e) {
|
||||
debugPrint('RemoteSyncService: ack failed: $e');
|
||||
}
|
||||
}
|
||||
|
||||
Future<RemoteSyncStateRow?> _getSyncState() async {
|
||||
final query = database.select(database.remoteSyncState)
|
||||
..where((row) => row.deviceId.equals('default'));
|
||||
return query.getSingleOrNull();
|
||||
}
|
||||
|
||||
Future<void> _saveCursor(String cursor) async {
|
||||
await database
|
||||
.into(database.remoteSyncState)
|
||||
.insertOnConflictUpdate(
|
||||
RemoteSyncStateCompanion(
|
||||
deviceId: const Value('default'),
|
||||
serverCursor: Value(cursor),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> _updateSyncState({required bool success}) async {
|
||||
final now = DateTime.now();
|
||||
final existing = await _getSyncState();
|
||||
|
||||
if (existing == null) {
|
||||
await database.into(database.remoteSyncState).insert(
|
||||
RemoteSyncStateCompanion.insert(
|
||||
deviceId: 'default',
|
||||
lastSuccessfulSyncAt: Value(success ? now : null),
|
||||
lastAttemptedSyncAt: Value(now),
|
||||
consecutiveFailureCount: Value(success ? 0 : 1),
|
||||
),
|
||||
);
|
||||
} else {
|
||||
await (database.update(database.remoteSyncState)
|
||||
..where((row) => row.deviceId.equals('default')))
|
||||
.write(RemoteSyncStateCompanion(
|
||||
lastSuccessfulSyncAt: Value(success ? now : existing.lastSuccessfulSyncAt),
|
||||
lastAttemptedSyncAt: Value(now),
|
||||
consecutiveFailureCount: Value(
|
||||
success ? 0 : existing.consecutiveFailureCount + 1,
|
||||
),
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
Future<bool> _isCommandProcessed(String commandId) async {
|
||||
final query = database.select(database.processedRemoteCommands)
|
||||
..where((row) => row.commandId.equals(commandId));
|
||||
final result = await query.getSingleOrNull();
|
||||
return result != null;
|
||||
}
|
||||
|
||||
Future<void> _recordProcessed(
|
||||
String commandId,
|
||||
String commandType,
|
||||
String result,
|
||||
) async {
|
||||
await database
|
||||
.into(database.processedRemoteCommands)
|
||||
.insertOnConflictUpdate(
|
||||
ProcessedRemoteCommandsCompanion.insert(
|
||||
commandId: commandId,
|
||||
commandType: commandType,
|
||||
processedAt: DateTime.now(),
|
||||
result: result,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
void _applyBackoff({int? forcedDelay}) {
|
||||
final delay = forcedDelay ?? _calculateBackoff();
|
||||
_syncTimer?.cancel();
|
||||
_syncTimer = Timer(Duration(seconds: delay), () => _onTick());
|
||||
}
|
||||
|
||||
int _calculateBackoff() {
|
||||
if (_consecutiveFailures <= 1) return 90;
|
||||
final backoff = 90 * (1 << (_consecutiveFailures - 1).clamp(0, 3));
|
||||
return backoff.clamp(90, 600);
|
||||
}
|
||||
|
||||
Future<void> setCredentials({
|
||||
required String deviceId,
|
||||
required String token,
|
||||
required String venueId,
|
||||
String? serverUrl,
|
||||
}) async {
|
||||
await secureStorage.write(key: _deviceIdKey, value: deviceId);
|
||||
await secureStorage.write(key: _authTokenKey, value: token);
|
||||
await secureStorage.write(key: _venueIdKey, value: venueId);
|
||||
if (serverUrl != null) {
|
||||
await secureStorage.write(key: _serverUrlKey, value: serverUrl);
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> clearCredentials() async {
|
||||
await secureStorage.delete(key: _deviceIdKey);
|
||||
await secureStorage.delete(key: _authTokenKey);
|
||||
await secureStorage.delete(key: _serverUrlKey);
|
||||
await secureStorage.delete(key: _venueIdKey);
|
||||
}
|
||||
|
||||
Future<DateTime?> get lastSuccessfulSyncAt async {
|
||||
final state = await _getSyncState();
|
||||
return state?.lastSuccessfulSyncAt;
|
||||
}
|
||||
|
||||
Future<int> get lastFailureCount async {
|
||||
final state = await _getSyncState();
|
||||
return state?.consecutiveFailureCount ?? 0;
|
||||
}
|
||||
|
||||
Future<List<ProcessedRemoteCommandRow>> getRecentActivity() async {
|
||||
final query = database.select(database.processedRemoteCommands)
|
||||
..orderBy([(row) => OrderingTerm.desc(row.processedAt)])
|
||||
..limit(50);
|
||||
return query.get();
|
||||
}
|
||||
|
||||
void dispose() {
|
||||
_syncTimer?.cancel();
|
||||
_eventController.close();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user