From d711f270ddddf27d3588980c5f09cb3edd80c83a Mon Sep 17 00:00:00 2001 From: Rijad Zuzo Date: Sun, 15 Feb 2026 16:56:55 +0100 Subject: [PATCH] Add sync queue orchestration and local mutation envelopes --- docs/progress.md | 36 +- lib/features/local/local_repository.dart | 602 +++++++++++++++++- lib/features/sync/sync_coordinator.dart | 179 ++++++ lib/features/sync/sync_queue_repository.dart | 133 ++++ lib/features/sync/sync_state.dart | 98 +++ lib/features/sync/sync_view.dart | 246 ++++--- test/features/sync/sync_coordinator_test.dart | 115 ++++ .../sync/sync_queue_repository_test.dart | 103 +++ 8 files changed, 1411 insertions(+), 101 deletions(-) create mode 100644 lib/features/sync/sync_coordinator.dart create mode 100644 lib/features/sync/sync_queue_repository.dart create mode 100644 lib/features/sync/sync_state.dart create mode 100644 test/features/sync/sync_coordinator_test.dart create mode 100644 test/features/sync/sync_queue_repository_test.dart diff --git a/docs/progress.md b/docs/progress.md index ba99d82..40f53ca 100644 --- a/docs/progress.md +++ b/docs/progress.md @@ -2,6 +2,34 @@ Updated: 2026-02-15 +## Latest Milestone (2026-02-15): Sync Queue + Orchestrator + +Implemented the next sync phase so local mutations are now queued and synced through a real coordinator flow: + +- Added persisted sync queue state: + - `lib/features/sync/sync_state.dart` + - `lib/features/sync/sync_queue_repository.dart` + - stores pending `ChangeEnvelope`s, cursor, last attempt/sync, and rejection info +- Added sync orchestration service: + - `lib/features/sync/sync_coordinator.dart` + - supports `pushPending`, `pullRemote`, and `syncNow` + - keeps offline-first behavior by recording failures without blocking local usage +- Wired local CRUD to sync envelopes: + - `lib/features/local/local_repository.dart` + - all person/moment/idea/reminder mutations now enqueue typed change envelopes + - added `applyRemoteChanges(...)` to apply pulled backend envelopes into local state +- Upgraded Sync UI: + - `lib/features/sync/sync_view.dart` + - now shows queue/cursor/error status and provides real sync actions (sync/push/pull/clear) +- Added sync tests: + - `test/features/sync/sync_queue_repository_test.dart` + - `test/features/sync/sync_coordinator_test.dart` + +Validation for this milestone: + +- `flutter analyze` -> pass +- `flutter test` -> pass + ## Latest Milestone (2026-02-15): Auth Session UX Implemented app-level authentication/session flow so navigation is now session-aware: @@ -149,7 +177,7 @@ Validation status after UI milestone: - Local database persistence and repository wiring (currently demo in-memory data). - Full CRUD flows (create/edit/delete) for people, captures, reminder rules, gift/event ideas. - Auth UX flows (sign-in, session-expired recovery, sign-out confirmation). -- Real sync orchestration layer (queue, cursor lifecycle, conflict policy UI). +- Conflict policy UI and merge/resolution UX for rejected remote/local mutations. - Notifications/reminders delivery and scheduling integration. - End-to-end and integration tests for app flows beyond unit/widget basics. @@ -169,9 +197,9 @@ Validation status after UI milestone: ### Phase 3: Sync + Auth UX -1. Add sync queue service and cursor management use cases. -2. Add conflict handling strategy and UI states. -3. Implement sign-in/session lifecycle screens and provider state. +1. Add background sync scheduling/triggers (app start, manual retry, connectivity hooks). +2. Add conflict handling strategy and user-facing UI states for rejected mutations. +3. Add session-expired recovery UX and guarded routes for protected actions. ### Phase 4: Quality + Release Readiness diff --git a/lib/features/local/local_repository.dart b/lib/features/local/local_repository.dart index 298e397..80107b0 100644 --- a/lib/features/local/local_repository.dart +++ b/lib/features/local/local_repository.dart @@ -2,6 +2,8 @@ import 'dart:convert'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import 'package:relationship_saver/features/local/local_models.dart'; +import 'package:relationship_saver/features/sync/sync_queue_repository.dart'; +import 'package:relationship_saver/integrations/backend/models/backend_models.dart'; import 'package:shared_preferences/shared_preferences.dart'; import 'package:uuid/uuid.dart'; @@ -10,6 +12,7 @@ class LocalRepository extends AsyncNotifier { static const String _storageKey = 'local_data_state_v1'; static const String _schemaVersionKey = 'local_data_schema_version'; static const int _schemaVersion = 2; + static const int _syncSchemaVersion = 1; static const Uuid _uuid = Uuid(); @@ -64,6 +67,14 @@ class LocalRepository extends AsyncNotifier { ...current.people, ]; await _setState(current.copyWith(people: people)); + await _enqueueChange( + _buildEnvelope( + entityType: 'person', + entityId: person.id, + op: ChangeOperation.upsert, + payload: _personPayload(person), + ), + ); } Future updatePerson(PersonProfile person) async { @@ -76,10 +87,28 @@ class LocalRepository extends AsyncNotifier { .map((PersonProfile item) => item.id == person.id ? person : item) .toList(growable: false); await _setState(current.copyWith(people: updated)); + await _enqueueChange( + _buildEnvelope( + entityType: 'person', + entityId: person.id, + op: ChangeOperation.upsert, + payload: _personPayload(person), + ), + ); } Future deletePerson(String personId) async { final LocalDataState current = _requireState(); + final List removedMoments = current.moments + .where((RelationshipMoment moment) => moment.personId == personId) + .toList(growable: false); + final List removedIdeas = current.ideas + .where((RelationshipIdea idea) => idea.personId == personId) + .toList(growable: false); + final List removedReminders = current.reminders + .where((ReminderRule reminder) => reminder.personId == personId) + .toList(growable: false); + final List people = current.people .where((PersonProfile person) => person.id != personId) .toList(growable: false); @@ -101,6 +130,35 @@ class LocalRepository extends AsyncNotifier { reminders: reminders, ), ); + + await _enqueueChanges([ + _buildEnvelope( + entityType: 'person', + entityId: personId, + op: ChangeOperation.delete, + ), + ...removedMoments.map( + (RelationshipMoment moment) => _buildEnvelope( + entityType: 'capture', + entityId: moment.id, + op: ChangeOperation.delete, + ), + ), + ...removedIdeas.map( + (RelationshipIdea idea) => _buildEnvelope( + entityType: _ideaEntityType(idea.type), + entityId: idea.id, + op: ChangeOperation.delete, + ), + ), + ...removedReminders.map( + (ReminderRule reminder) => _buildEnvelope( + entityType: 'reminderRule', + entityId: reminder.id, + op: ChangeOperation.delete, + ), + ), + ]); } Future addMoment({ @@ -132,6 +190,14 @@ class LocalRepository extends AsyncNotifier { ...current.moments, ]; await _setState(current.copyWith(moments: moments)); + await _enqueueChange( + _buildEnvelope( + entityType: 'capture', + entityId: moment.id, + op: ChangeOperation.upsert, + payload: _momentPayload(moment), + ), + ); } Future updateMoment(RelationshipMoment moment) async { @@ -144,6 +210,14 @@ class LocalRepository extends AsyncNotifier { .map((RelationshipMoment item) => item.id == moment.id ? moment : item) .toList(growable: false); await _setState(current.copyWith(moments: updated)); + await _enqueueChange( + _buildEnvelope( + entityType: 'capture', + entityId: moment.id, + op: ChangeOperation.upsert, + payload: _momentPayload(moment), + ), + ); } Future deleteMoment(String momentId) async { @@ -152,6 +226,13 @@ class LocalRepository extends AsyncNotifier { .where((RelationshipMoment moment) => moment.id != momentId) .toList(growable: false); await _setState(current.copyWith(moments: moments)); + await _enqueueChange( + _buildEnvelope( + entityType: 'capture', + entityId: momentId, + op: ChangeOperation.delete, + ), + ); } Future addIdea({ @@ -180,6 +261,14 @@ class LocalRepository extends AsyncNotifier { ...current.ideas, ]; await _setState(current.copyWith(ideas: ideas)); + await _enqueueChange( + _buildEnvelope( + entityType: _ideaEntityType(idea.type), + entityId: idea.id, + op: ChangeOperation.upsert, + payload: _ideaPayload(idea), + ), + ); } Future updateIdea(RelationshipIdea idea) async { @@ -192,18 +281,42 @@ class LocalRepository extends AsyncNotifier { .map((RelationshipIdea item) => item.id == idea.id ? idea : item) .toList(growable: false); await _setState(current.copyWith(ideas: ideas)); + await _enqueueChange( + _buildEnvelope( + entityType: _ideaEntityType(idea.type), + entityId: idea.id, + op: ChangeOperation.upsert, + payload: _ideaPayload(idea), + ), + ); } Future toggleIdeaArchived(String ideaId) async { final LocalDataState current = _requireState(); + RelationshipIdea? updatedIdea; final List ideas = current.ideas - .map( - (RelationshipIdea item) => item.id == ideaId - ? item.copyWith(isArchived: !item.isArchived) - : item, - ) + .map((RelationshipIdea item) { + if (item.id != ideaId) { + return item; + } + final RelationshipIdea toggled = item.copyWith( + isArchived: !item.isArchived, + ); + updatedIdea = toggled; + return toggled; + }) .toList(growable: false); await _setState(current.copyWith(ideas: ideas)); + if (updatedIdea != null) { + await _enqueueChange( + _buildEnvelope( + entityType: _ideaEntityType(updatedIdea!.type), + entityId: updatedIdea!.id, + op: ChangeOperation.upsert, + payload: _ideaPayload(updatedIdea!), + ), + ); + } } Future deleteIdea(String ideaId) async { @@ -212,6 +325,13 @@ class LocalRepository extends AsyncNotifier { .where((RelationshipIdea idea) => idea.id != ideaId) .toList(growable: false); await _setState(current.copyWith(ideas: ideas)); + await _enqueueChange( + _buildEnvelope( + entityType: _ideaEntityTypeFromId(current.ideas, ideaId), + entityId: ideaId, + op: ChangeOperation.delete, + ), + ); } Future addReminder({ @@ -240,6 +360,14 @@ class LocalRepository extends AsyncNotifier { ...current.reminders, ]; await _setState(current.copyWith(reminders: reminders)); + await _enqueueChange( + _buildEnvelope( + entityType: 'reminderRule', + entityId: reminder.id, + op: ChangeOperation.upsert, + payload: _reminderPayload(reminder), + ), + ); } Future updateReminder(ReminderRule reminder) async { @@ -252,18 +380,40 @@ class LocalRepository extends AsyncNotifier { .map((ReminderRule item) => item.id == reminder.id ? reminder : item) .toList(growable: false); await _setState(current.copyWith(reminders: reminders)); + await _enqueueChange( + _buildEnvelope( + entityType: 'reminderRule', + entityId: reminder.id, + op: ChangeOperation.upsert, + payload: _reminderPayload(reminder), + ), + ); } Future toggleReminderEnabled(String reminderId) async { final LocalDataState current = _requireState(); + ReminderRule? updatedReminder; final List reminders = current.reminders - .map( - (ReminderRule item) => item.id == reminderId - ? item.copyWith(enabled: !item.enabled) - : item, - ) + .map((ReminderRule item) { + if (item.id != reminderId) { + return item; + } + final ReminderRule toggled = item.copyWith(enabled: !item.enabled); + updatedReminder = toggled; + return toggled; + }) .toList(growable: false); await _setState(current.copyWith(reminders: reminders)); + if (updatedReminder != null) { + await _enqueueChange( + _buildEnvelope( + entityType: 'reminderRule', + entityId: updatedReminder!.id, + op: ChangeOperation.upsert, + payload: _reminderPayload(updatedReminder!), + ), + ); + } } Future deleteReminder(String reminderId) async { @@ -272,6 +422,13 @@ class LocalRepository extends AsyncNotifier { .where((ReminderRule reminder) => reminder.id != reminderId) .toList(growable: false); await _setState(current.copyWith(reminders: reminders)); + await _enqueueChange( + _buildEnvelope( + entityType: 'reminderRule', + entityId: reminderId, + op: ChangeOperation.delete, + ), + ); } Future toggleTaskDone(String taskId) async { @@ -286,6 +443,431 @@ class LocalRepository extends AsyncNotifier { await _setState(current.copyWith(tasks: tasks)); } + Future applyRemoteChanges(List changes) async { + if (changes.isEmpty) { + return; + } + + LocalDataState next = _requireState(); + for (final ChangeEnvelope change in changes) { + next = _applyRemoteChange(next, change); + } + await _setState(next); + } + + LocalDataState _applyRemoteChange( + LocalDataState current, + ChangeEnvelope change, + ) { + switch (change.entityType) { + case 'person': + return _applyPersonChange(current, change); + case 'capture': + return _applyMomentChange(current, change); + case 'giftIdea': + case 'eventIdea': + return _applyIdeaChange(current, change); + case 'reminderRule': + return _applyReminderChange(current, change); + default: + return current; + } + } + + LocalDataState _applyPersonChange( + LocalDataState current, + ChangeEnvelope change, + ) { + if (change.op == ChangeOperation.delete) { + final String personId = change.entityId; + return current.copyWith( + people: current.people + .where((PersonProfile person) => person.id != personId) + .toList(growable: false), + moments: current.moments + .where((RelationshipMoment moment) => moment.personId != personId) + .toList(growable: false), + ideas: current.ideas + .where((RelationshipIdea idea) => idea.personId != personId) + .toList(growable: false), + reminders: current.reminders + .where((ReminderRule reminder) => reminder.personId != personId) + .toList(growable: false), + ); + } + + final int existingIndex = current.people.indexWhere( + (PersonProfile person) => person.id == change.entityId, + ); + final PersonProfile? existing = existingIndex >= 0 + ? current.people[existingIndex] + : null; + final Map payload = change.payload ?? {}; + final PersonProfile next = PersonProfile( + id: change.entityId, + name: _stringField(payload, 'name', existing?.name ?? 'Unknown'), + relationship: _stringField( + payload, + 'relationship', + existing?.relationship ?? 'Relationship', + ), + affinityScore: _intField( + payload, + 'affinityScore', + existing?.affinityScore ?? 70, + ), + nextMoment: _dateField( + payload, + 'nextMoment', + existing?.nextMoment ?? DateTime.now().add(const Duration(days: 3)), + ), + tags: _stringListField(payload, 'tags', existing?.tags ?? []), + notes: _stringField(payload, 'notes', existing?.notes ?? ''), + ); + final List people = _upsertItem( + items: current.people, + item: next, + idOf: (PersonProfile person) => person.id, + ); + return current.copyWith(people: people); + } + + LocalDataState _applyMomentChange( + LocalDataState current, + ChangeEnvelope change, + ) { + if (change.op == ChangeOperation.delete) { + return current.copyWith( + moments: current.moments + .where((RelationshipMoment moment) => moment.id != change.entityId) + .toList(growable: false), + ); + } + + final int existingIndex = current.moments.indexWhere( + (RelationshipMoment moment) => moment.id == change.entityId, + ); + final RelationshipMoment? existing = existingIndex >= 0 + ? current.moments[existingIndex] + : null; + final Map payload = change.payload ?? {}; + final String summary = _stringField( + payload, + 'summary', + existing?.summary ?? '', + ); + final RelationshipMoment next = RelationshipMoment( + id: change.entityId, + personId: _stringField( + payload, + 'personId', + existing?.personId ?? + (current.people.isEmpty + ? 'person-unknown' + : current.people.first.id), + ), + title: _stringField( + payload, + 'title', + existing?.title ?? _titleFromSummary(summary), + ), + summary: summary, + at: _dateField(payload, 'at', existing?.at ?? DateTime.now()), + type: _stringField(payload, 'type', existing?.type ?? 'capture'), + ); + final List moments = _upsertItem( + items: current.moments, + item: next, + idOf: (RelationshipMoment moment) => moment.id, + ); + return current.copyWith(moments: moments); + } + + LocalDataState _applyIdeaChange( + LocalDataState current, + ChangeEnvelope change, + ) { + if (change.op == ChangeOperation.delete) { + return current.copyWith( + ideas: current.ideas + .where((RelationshipIdea idea) => idea.id != change.entityId) + .toList(growable: false), + ); + } + + final int existingIndex = current.ideas.indexWhere( + (RelationshipIdea idea) => idea.id == change.entityId, + ); + final RelationshipIdea? existing = existingIndex >= 0 + ? current.ideas[existingIndex] + : null; + final Map payload = change.payload ?? {}; + final IdeaType fallbackType = _ideaTypeFromEntityType( + change.entityType, + fallback: existing?.type ?? IdeaType.gift, + ); + final IdeaType type = _ideaTypeFromRaw( + payload['type'], + fallback: fallbackType, + ); + final RelationshipIdea next = RelationshipIdea( + id: change.entityId, + personId: payload['personId'] as String? ?? existing?.personId, + type: type, + title: _stringField(payload, 'title', existing?.title ?? 'Untitled idea'), + details: _stringField(payload, 'details', existing?.details ?? ''), + createdAt: _dateField( + payload, + 'createdAt', + existing?.createdAt ?? DateTime.now(), + ), + isArchived: _boolField( + payload, + 'isArchived', + existing?.isArchived ?? false, + ), + ); + final List ideas = _upsertItem( + items: current.ideas, + item: next, + idOf: (RelationshipIdea idea) => idea.id, + ); + return current.copyWith(ideas: ideas); + } + + LocalDataState _applyReminderChange( + LocalDataState current, + ChangeEnvelope change, + ) { + if (change.op == ChangeOperation.delete) { + return current.copyWith( + reminders: current.reminders + .where((ReminderRule reminder) => reminder.id != change.entityId) + .toList(growable: false), + ); + } + + final int existingIndex = current.reminders.indexWhere( + (ReminderRule reminder) => reminder.id == change.entityId, + ); + final ReminderRule? existing = existingIndex >= 0 + ? current.reminders[existingIndex] + : null; + final Map payload = change.payload ?? {}; + final ReminderRule next = ReminderRule( + id: change.entityId, + personId: payload['personId'] as String? ?? existing?.personId, + title: _stringField(payload, 'title', existing?.title ?? 'Reminder'), + cadence: _cadenceFromRaw( + payload['cadence'], + fallback: existing?.cadence ?? ReminderCadence.weekly, + ), + nextAt: _dateField(payload, 'nextAt', existing?.nextAt ?? DateTime.now()), + enabled: _boolField(payload, 'enabled', existing?.enabled ?? true), + ); + final List reminders = _upsertItem( + items: current.reminders, + item: next, + idOf: (ReminderRule reminder) => reminder.id, + ); + return current.copyWith(reminders: reminders); + } + + Future _enqueueChange(ChangeEnvelope change) async { + await ref.read(syncQueueRepositoryProvider.notifier).enqueue(change); + } + + Future _enqueueChanges(List changes) async { + await ref.read(syncQueueRepositoryProvider.notifier).enqueueAll(changes); + } + + ChangeEnvelope _buildEnvelope({ + required String entityType, + required String entityId, + required ChangeOperation op, + Map? payload, + }) { + return ChangeEnvelope( + schemaVersion: _syncSchemaVersion, + entityType: entityType, + entityId: entityId, + op: op, + modifiedAt: DateTime.now().toUtc(), + clientMutationId: _uuid.v4(), + payload: payload, + ); + } + + Map _personPayload(PersonProfile person) { + return { + 'name': person.name, + 'relationship': person.relationship, + 'affinityScore': person.affinityScore, + 'nextMoment': person.nextMoment.toUtc().toIso8601String(), + 'tags': person.tags, + 'notes': person.notes, + }; + } + + Map _momentPayload(RelationshipMoment moment) { + return { + 'personId': moment.personId, + 'title': moment.title, + 'summary': moment.summary, + 'at': moment.at.toUtc().toIso8601String(), + 'type': moment.type, + }; + } + + Map _ideaPayload(RelationshipIdea idea) { + return { + 'personId': idea.personId, + 'type': idea.type.name, + 'title': idea.title, + 'details': idea.details, + 'createdAt': idea.createdAt.toUtc().toIso8601String(), + 'isArchived': idea.isArchived, + }; + } + + Map _reminderPayload(ReminderRule reminder) { + return { + 'personId': reminder.personId, + 'title': reminder.title, + 'cadence': reminder.cadence.name, + 'nextAt': reminder.nextAt.toUtc().toIso8601String(), + 'enabled': reminder.enabled, + }; + } + + String _ideaEntityType(IdeaType type) { + return type == IdeaType.event ? 'eventIdea' : 'giftIdea'; + } + + String _ideaEntityTypeFromId(List ideas, String ideaId) { + final int index = ideas.indexWhere( + (RelationshipIdea idea) => idea.id == ideaId, + ); + final RelationshipIdea? existing = index >= 0 ? ideas[index] : null; + if (existing == null) { + return 'giftIdea'; + } + return _ideaEntityType(existing.type); + } + + IdeaType _ideaTypeFromEntityType( + String entityType, { + required IdeaType fallback, + }) { + switch (entityType) { + case 'eventIdea': + return IdeaType.event; + case 'giftIdea': + return IdeaType.gift; + default: + return fallback; + } + } + + IdeaType _ideaTypeFromRaw(dynamic value, {required IdeaType fallback}) { + if (value is! String || value.isEmpty) { + return fallback; + } + + return IdeaType.values.firstWhere( + (IdeaType type) => type.name == value, + orElse: () => fallback, + ); + } + + ReminderCadence _cadenceFromRaw( + dynamic value, { + required ReminderCadence fallback, + }) { + if (value is! String || value.isEmpty) { + return fallback; + } + + return ReminderCadence.values.firstWhere( + (ReminderCadence cadence) => cadence.name == value, + orElse: () => fallback, + ); + } + + List _upsertItem({ + required List items, + required T item, + required String Function(T value) idOf, + }) { + final int index = items.indexWhere((T value) => idOf(value) == idOf(item)); + if (index < 0) { + return [item, ...items]; + } + + final List copy = items.toList(); + copy[index] = item; + return copy; + } + + String _stringField( + Map payload, + String key, + String fallback, + ) { + final dynamic value = payload[key]; + if (value is String && value.trim().isNotEmpty) { + return value.trim(); + } + return fallback; + } + + int _intField(Map payload, String key, int fallback) { + final dynamic value = payload[key]; + if (value is int) { + return value; + } + if (value is num) { + return value.toInt(); + } + return fallback; + } + + bool _boolField(Map payload, String key, bool fallback) { + final dynamic value = payload[key]; + if (value is bool) { + return value; + } + return fallback; + } + + List _stringListField( + Map payload, + String key, + List fallback, + ) { + final dynamic value = payload[key]; + if (value is! List) { + return fallback; + } + + return value.map((dynamic item) => '$item').toList(growable: false); + } + + DateTime _dateField( + Map payload, + String key, + DateTime fallback, + ) { + final dynamic value = payload[key]; + if (value is String) { + final DateTime? parsed = DateTime.tryParse(value); + if (parsed != null) { + return parsed.toLocal(); + } + } + return fallback; + } + Future _setState(LocalDataState next) async { state = AsyncData(next); await _persist(next); diff --git a/lib/features/sync/sync_coordinator.dart b/lib/features/sync/sync_coordinator.dart new file mode 100644 index 0000000..84b6f78 --- /dev/null +++ b/lib/features/sync/sync_coordinator.dart @@ -0,0 +1,179 @@ +import 'package:flutter/foundation.dart'; +import 'package:flutter_riverpod/flutter_riverpod.dart'; +import 'package:relationship_saver/features/local/local_repository.dart'; +import 'package:relationship_saver/features/sync/sync_queue_repository.dart'; +import 'package:relationship_saver/features/sync/sync_state.dart'; +import 'package:relationship_saver/integrations/backend/backend_gateway_provider.dart'; + +/// Result details for one sync action. +@immutable +class SyncRunResult { + const SyncRunResult({ + required this.success, + required this.operation, + required this.message, + required this.pendingAfter, + this.cursor, + this.accepted = 0, + this.rejected = 0, + this.pulled = 0, + this.error, + }); + + final bool success; + final String operation; + final String message; + final int pendingAfter; + final String? cursor; + final int accepted; + final int rejected; + final int pulled; + final Object? error; +} + +/// Coordinates push/pull sync operations across queue, gateway, and local store. +class SyncCoordinator { + const SyncCoordinator(this._ref); + + final Ref _ref; + + Future pushPending() async { + final SyncState queueState = await _ref.read( + syncQueueRepositoryProvider.future, + ); + if (queueState.pendingChanges.isEmpty) { + return SyncRunResult( + success: true, + operation: 'push', + message: 'No pending local changes to push.', + pendingAfter: 0, + cursor: queueState.cursor, + ); + } + + final DateTime now = DateTime.now(); + final syncQueue = _ref.read(syncQueueRepositoryProvider.notifier); + try { + final result = await _ref + .read(backendGatewayProvider) + .pushChanges( + changes: queueState.pendingChanges, + cursor: queueState.cursor, + ); + await syncQueue.applyPushResult(result, at: now); + final SyncState after = await _ref.read( + syncQueueRepositoryProvider.future, + ); + return SyncRunResult( + success: true, + operation: 'push', + message: + 'Push complete. accepted=${result.accepted.length}, rejected=${result.rejected.length}.', + accepted: result.accepted.length, + rejected: result.rejected.length, + pendingAfter: after.pendingChanges.length, + cursor: after.cursor, + ); + } catch (error) { + await syncQueue.markFailure(error, at: now); + final SyncState after = await _ref.read( + syncQueueRepositoryProvider.future, + ); + return SyncRunResult( + success: false, + operation: 'push', + message: 'Push failed. Local changes stay queued.', + pendingAfter: after.pendingChanges.length, + cursor: after.cursor, + error: error, + ); + } + } + + Future pullRemote() async { + final SyncState queueState = await _ref.read( + syncQueueRepositoryProvider.future, + ); + final DateTime now = DateTime.now(); + final syncQueue = _ref.read(syncQueueRepositoryProvider.notifier); + + try { + final result = await _ref + .read(backendGatewayProvider) + .pullChanges(cursor: queueState.cursor); + await _ref + .read(localRepositoryProvider.notifier) + .applyRemoteChanges(result.changes); + await syncQueue.applyPullResult(result, at: now); + final SyncState after = await _ref.read( + syncQueueRepositoryProvider.future, + ); + return SyncRunResult( + success: true, + operation: 'pull', + message: 'Pulled ${result.changes.length} change(s) from backend.', + pulled: result.changes.length, + pendingAfter: after.pendingChanges.length, + cursor: after.cursor, + ); + } catch (error) { + await syncQueue.markFailure(error, at: now); + final SyncState after = await _ref.read( + syncQueueRepositoryProvider.future, + ); + return SyncRunResult( + success: false, + operation: 'pull', + message: 'Pull failed. Staying in local-first mode.', + pendingAfter: after.pendingChanges.length, + cursor: after.cursor, + error: error, + ); + } + } + + Future syncNow() async { + final SyncRunResult pushResult = await pushPending(); + if (!pushResult.success) { + return SyncRunResult( + success: false, + operation: 'sync', + message: 'Sync paused at push step. ${pushResult.message}', + pendingAfter: pushResult.pendingAfter, + cursor: pushResult.cursor, + accepted: pushResult.accepted, + rejected: pushResult.rejected, + error: pushResult.error, + ); + } + + final SyncRunResult pullResult = await pullRemote(); + if (!pullResult.success) { + return SyncRunResult( + success: false, + operation: 'sync', + message: 'Sync paused at pull step. ${pullResult.message}', + pendingAfter: pullResult.pendingAfter, + cursor: pullResult.cursor, + accepted: pushResult.accepted, + rejected: pushResult.rejected, + error: pullResult.error, + ); + } + + return SyncRunResult( + success: true, + operation: 'sync', + message: + 'Sync complete. accepted=${pushResult.accepted}, rejected=${pushResult.rejected}, pulled=${pullResult.pulled}.', + pendingAfter: pullResult.pendingAfter, + cursor: pullResult.cursor, + accepted: pushResult.accepted, + rejected: pushResult.rejected, + pulled: pullResult.pulled, + ); + } +} + +final Provider syncCoordinatorProvider = + Provider(SyncCoordinator.new); diff --git a/lib/features/sync/sync_queue_repository.dart b/lib/features/sync/sync_queue_repository.dart new file mode 100644 index 0000000..21864dc --- /dev/null +++ b/lib/features/sync/sync_queue_repository.dart @@ -0,0 +1,133 @@ +import 'dart:convert'; + +import 'package:flutter_riverpod/flutter_riverpod.dart'; +import 'package:relationship_saver/features/sync/sync_state.dart'; +import 'package:relationship_saver/integrations/backend/models/backend_models.dart'; +import 'package:shared_preferences/shared_preferences.dart'; + +/// Persists queued local mutations and sync cursor metadata. +class SyncQueueRepository extends AsyncNotifier { + static const String _storageKey = 'sync_queue_state_v1'; + + @override + Future build() async { + final SharedPreferences prefs = await SharedPreferences.getInstance(); + final String? raw = prefs.getString(_storageKey); + if (raw == null || raw.isEmpty) { + return SyncState.empty; + } + + try { + final Map json = jsonDecode(raw) as Map; + return SyncState.fromJson(json); + } on FormatException { + await prefs.remove(_storageKey); + return SyncState.empty; + } + } + + Future enqueue(ChangeEnvelope change) async { + final SyncState current = await _currentState(); + final List pending = [ + change, + ...current.pendingChanges, + ]; + await _setState(current.copyWith(pendingChanges: pending)); + } + + Future enqueueAll(Iterable changes) async { + final List values = changes.toList(growable: false); + if (values.isEmpty) { + return; + } + + final SyncState current = await _currentState(); + final List pending = [ + ...values, + ...current.pendingChanges, + ]; + await _setState(current.copyWith(pendingChanges: pending)); + } + + Future applyPushResult(SyncPushResult result, {DateTime? at}) async { + final SyncState current = await _currentState(); + final DateTime now = at ?? DateTime.now(); + final Set completedMutationIds = { + ...result.accepted.map((MutationAck ack) => ack.clientMutationId), + ...result.rejected.map( + (MutationRejection rejection) => rejection.clientMutationId, + ), + }; + final List pending = current.pendingChanges + .where( + (ChangeEnvelope change) => + !completedMutationIds.contains(change.clientMutationId), + ) + .toList(growable: false); + + final String? rejectionMessage = result.rejected.isEmpty + ? null + : 'Push rejected ${result.rejected.length} change(s).'; + + await _setState( + current.copyWith( + cursor: result.cursor, + pendingChanges: pending, + lastAttemptAt: now, + lastSyncAt: now, + lastError: rejectionMessage, + lastRejected: result.rejected, + ), + ); + } + + Future applyPullResult(SyncPullResult result, {DateTime? at}) async { + final SyncState current = await _currentState(); + final DateTime now = at ?? DateTime.now(); + await _setState( + current.copyWith( + cursor: result.cursor, + lastAttemptAt: now, + lastSyncAt: now, + lastError: null, + lastRejected: const [], + ), + ); + } + + Future markFailure(Object error, {DateTime? at}) async { + final SyncState current = await _currentState(); + final DateTime now = at ?? DateTime.now(); + await _setState(current.copyWith(lastAttemptAt: now, lastError: '$error')); + } + + Future clearQueue() async { + final SyncState current = await _currentState(); + await _setState( + current.copyWith( + pendingChanges: const [], + lastRejected: const [], + ), + ); + } + + Future _currentState() async { + final SyncState? value = state.asData?.value; + if (value != null) { + return value; + } + return future; + } + + Future _setState(SyncState next) async { + state = AsyncData(next); + final SharedPreferences prefs = await SharedPreferences.getInstance(); + await prefs.setString(_storageKey, jsonEncode(next.toJson())); + } +} + +final AsyncNotifierProvider +syncQueueRepositoryProvider = + AsyncNotifierProvider( + SyncQueueRepository.new, + ); diff --git a/lib/features/sync/sync_state.dart b/lib/features/sync/sync_state.dart new file mode 100644 index 0000000..37f1227 --- /dev/null +++ b/lib/features/sync/sync_state.dart @@ -0,0 +1,98 @@ +import 'package:flutter/foundation.dart'; +import 'package:relationship_saver/integrations/backend/models/backend_models.dart'; + +/// Persisted state for sync queue and cursor progression. +@immutable +class SyncState { + const SyncState({ + this.cursor, + required this.pendingChanges, + this.lastSyncAt, + this.lastAttemptAt, + this.lastError, + this.lastRejected = const [], + }); + + factory SyncState.fromJson(Map json) { + return SyncState( + cursor: json['cursor'] as String?, + pendingChanges: (json['pendingChanges'] as List? ?? []) + .map( + (dynamic item) => + ChangeEnvelope.fromJson(item as Map), + ) + .toList(growable: false), + lastSyncAt: _tryParseDate(json['lastSyncAt'] as String?), + lastAttemptAt: _tryParseDate(json['lastAttemptAt'] as String?), + lastError: json['lastError'] as String?, + lastRejected: (json['lastRejected'] as List? ?? []) + .map( + (dynamic item) => + MutationRejection.fromJson(item as Map), + ) + .toList(growable: false), + ); + } + + final String? cursor; + final List pendingChanges; + final DateTime? lastSyncAt; + final DateTime? lastAttemptAt; + final String? lastError; + final List lastRejected; + + static const Object _unset = Object(); + + static const SyncState empty = SyncState( + pendingChanges: [], + lastRejected: [], + ); + + SyncState copyWith({ + Object? cursor = _unset, + List? pendingChanges, + Object? lastSyncAt = _unset, + Object? lastAttemptAt = _unset, + Object? lastError = _unset, + List? lastRejected, + }) { + return SyncState( + cursor: identical(cursor, _unset) ? this.cursor : cursor as String?, + pendingChanges: pendingChanges ?? this.pendingChanges, + lastSyncAt: identical(lastSyncAt, _unset) + ? this.lastSyncAt + : lastSyncAt as DateTime?, + lastAttemptAt: identical(lastAttemptAt, _unset) + ? this.lastAttemptAt + : lastAttemptAt as DateTime?, + lastError: identical(lastError, _unset) + ? this.lastError + : lastError as String?, + lastRejected: lastRejected ?? this.lastRejected, + ); + } + + Map toJson() { + return { + 'cursor': cursor, + 'pendingChanges': pendingChanges + .map((ChangeEnvelope change) => change.toJson()) + .toList(growable: false), + 'lastSyncAt': lastSyncAt?.toUtc().toIso8601String(), + 'lastAttemptAt': lastAttemptAt?.toUtc().toIso8601String(), + 'lastError': lastError, + 'lastRejected': lastRejected + .map((MutationRejection rejection) => rejection.toJson()) + .toList(growable: false), + }; + } + + static DateTime? _tryParseDate(String? raw) { + if (raw == null || raw.isEmpty) { + return null; + } + + final DateTime? parsed = DateTime.tryParse(raw); + return parsed?.toLocal(); + } +} diff --git a/lib/features/sync/sync_view.dart b/lib/features/sync/sync_view.dart index 8883beb..84f1a69 100644 --- a/lib/features/sync/sync_view.dart +++ b/lib/features/sync/sync_view.dart @@ -2,9 +2,9 @@ import 'package:flutter/material.dart'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import 'package:relationship_saver/core/config/app_theme.dart'; import 'package:relationship_saver/features/shared/frosted_card.dart'; -import 'package:relationship_saver/integrations/backend/backend_gateway_provider.dart'; -import 'package:relationship_saver/integrations/backend/models/backend_models.dart'; -import 'package:uuid/uuid.dart'; +import 'package:relationship_saver/features/sync/sync_coordinator.dart'; +import 'package:relationship_saver/features/sync/sync_queue_repository.dart'; +import 'package:relationship_saver/features/sync/sync_state.dart'; class SyncView extends ConsumerStatefulWidget { const SyncView({super.key}); @@ -14,14 +14,15 @@ class SyncView extends ConsumerStatefulWidget { } class _SyncViewState extends ConsumerState { - static const Uuid _uuid = Uuid(); - bool _busy = false; String _status = 'Ready. Local-first mode keeps app usable without sync.'; - String? _cursor; @override Widget build(BuildContext context) { + final AsyncValue syncState = ref.watch( + syncQueueRepositoryProvider, + ); + return Padding( padding: const EdgeInsets.fromLTRB(28, 24, 28, 28), child: Column( @@ -30,52 +31,26 @@ class _SyncViewState extends ConsumerState { Text('Sync', style: Theme.of(context).textTheme.headlineMedium), const SizedBox(height: 6), Text( - 'Push local changes and pull remote updates when connectivity is available.', + 'Queue local changes, then push and pull when network is available.', style: Theme.of( context, ).textTheme.bodyLarge?.copyWith(color: AppTheme.textSecondary), ), const SizedBox(height: 16), - FrostedCard( - child: Column( - crossAxisAlignment: CrossAxisAlignment.start, - children: [ - Wrap( - spacing: 10, - runSpacing: 10, - children: [ - FilledButton.icon( - onPressed: _busy ? null : _pushSample, - icon: const Icon(Icons.upload_rounded), - label: const Text('Push Sample Changes'), + Expanded( + child: syncState.when( + data: (SyncState state) => _buildBody(context, state), + loading: () => const Center(child: CircularProgressIndicator()), + error: (Object error, StackTrace stackTrace) { + return Center( + child: Text( + 'Unable to load sync state. $error', + style: Theme.of(context).textTheme.bodyMedium?.copyWith( + color: AppTheme.textSecondary, ), - OutlinedButton.icon( - onPressed: _busy ? null : _pullChanges, - icon: const Icon(Icons.download_rounded), - label: const Text('Pull Changes'), - ), - ], - ), - const SizedBox(height: 16), - Text( - 'Cursor: ${_cursor ?? 'not set'}', - style: Theme.of(context).textTheme.bodyMedium, - ), - const SizedBox(height: 8), - Text( - _status, - style: Theme.of(context).textTheme.bodyLarge?.copyWith( - color: AppTheme.textSecondary, ), - ), - ], - ), - ), - const SizedBox(height: 16), - FrostedCard( - child: Text( - 'Sync policy\n• Core usage never blocks on backend\n• Retries only for transient failures\n• Non-idempotent requests require idempotency key', - style: Theme.of(context).textTheme.bodyLarge, + ); + }, ), ), ], @@ -83,71 +58,168 @@ class _SyncViewState extends ConsumerState { ); } + Widget _buildBody(BuildContext context, SyncState state) { + return ListView( + children: [ + FrostedCard( + child: Column( + crossAxisAlignment: CrossAxisAlignment.start, + children: [ + Wrap( + spacing: 10, + runSpacing: 10, + children: [ + FilledButton.icon( + onPressed: _busy ? null : _syncNow, + icon: const Icon(Icons.sync_rounded), + label: const Text('Sync Now'), + ), + OutlinedButton.icon( + onPressed: _busy ? null : _pushPending, + icon: const Icon(Icons.upload_rounded), + label: const Text('Push Pending'), + ), + OutlinedButton.icon( + onPressed: _busy ? null : _pullChanges, + icon: const Icon(Icons.download_rounded), + label: const Text('Pull Changes'), + ), + TextButton.icon( + onPressed: _busy ? null : _clearQueue, + icon: const Icon(Icons.clear_all_rounded), + label: const Text('Clear Queue'), + ), + ], + ), + const SizedBox(height: 16), + Text( + _status, + style: Theme.of( + context, + ).textTheme.bodyLarge?.copyWith(color: AppTheme.textSecondary), + ), + ], + ), + ), + const SizedBox(height: 16), + FrostedCard( + child: Column( + crossAxisAlignment: CrossAxisAlignment.start, + children: [ + Text( + 'Queue Status', + style: Theme.of(context).textTheme.titleMedium, + ), + const SizedBox(height: 10), + _infoRow('Pending changes', '${state.pendingChanges.length}'), + _infoRow('Cursor', state.cursor ?? 'not set'), + _infoRow('Last sync', _formatDateTime(state.lastSyncAt)), + _infoRow('Last attempt', _formatDateTime(state.lastAttemptAt)), + _infoRow('Last rejected count', '${state.lastRejected.length}'), + _infoRow('Last error', state.lastError ?? 'none'), + ], + ), + ), + const SizedBox(height: 16), + FrostedCard( + child: Text( + 'Sync policy\n• Core usage never blocks on backend\n• Local changes queue first and remain safe offline\n• Pull applies backend envelopes into local state', + style: Theme.of(context).textTheme.bodyLarge, + ), + ), + ], + ); + } + + Widget _infoRow(String label, String value) { + return Padding( + padding: const EdgeInsets.symmetric(vertical: 2), + child: Row( + crossAxisAlignment: CrossAxisAlignment.start, + children: [ + SizedBox(width: 150, child: Text(label)), + Expanded( + child: Text( + value, + style: Theme.of( + context, + ).textTheme.bodyMedium?.copyWith(color: AppTheme.textSecondary), + ), + ), + ], + ), + ); + } + + String _formatDateTime(DateTime? value) { + if (value == null) { + return 'never'; + } + + final DateTime local = value.toLocal(); + final String month = local.month.toString().padLeft(2, '0'); + final String day = local.day.toString().padLeft(2, '0'); + final String hour = local.hour.toString().padLeft(2, '0'); + final String minute = local.minute.toString().padLeft(2, '0'); + return '${local.year}-$month-$day $hour:$minute'; + } + + Future _syncNow() async { + await _run((SyncCoordinator coordinator) => coordinator.syncNow()); + } + + Future _pushPending() async { + await _run((SyncCoordinator coordinator) => coordinator.pushPending()); + } + Future _pullChanges() async { + await _run((SyncCoordinator coordinator) => coordinator.pullRemote()); + } + + Future _clearQueue() async { setState(() { _busy = true; - _status = 'Pulling changes...'; }); try { - final SyncPullResult result = await ref - .read(backendGatewayProvider) - .pullChanges(cursor: _cursor); - + await ref.read(syncQueueRepositoryProvider.notifier).clearQueue(); setState(() { - _cursor = result.cursor; - _status = 'Pulled ${result.changes.length} changes successfully.'; - }); - } catch (error) { - setState(() { - _status = 'Pull failed. Staying offline-safe. ($error)'; + _status = 'Cleared pending sync queue.'; }); } finally { - setState(() { - _busy = false; - }); + if (mounted) { + setState(() { + _busy = false; + }); + } } } - Future _pushSample() async { + Future _run( + Future Function(SyncCoordinator coordinator) action, + ) async { setState(() { _busy = true; - _status = 'Pushing local sample changes...'; + _status = 'Running sync operation...'; }); - final List changes = [ - ChangeEnvelope( - schemaVersion: 1, - entityType: 'capture', - entityId: 'capture-${_uuid.v4()}', - op: ChangeOperation.upsert, - modifiedAt: DateTime.now().toUtc(), - clientMutationId: _uuid.v4(), - payload: const { - 'summary': 'Shared a thoughtful check-in after work.', - 'sentiment': 'positive', - }, - ), - ]; - try { - final SyncPushResult result = await ref - .read(backendGatewayProvider) - .pushChanges(changes: changes, cursor: _cursor); - + final SyncRunResult result = await action( + ref.read(syncCoordinatorProvider), + ); setState(() { - _cursor = result.cursor; - _status = - 'Push complete. accepted=${result.accepted.length}, rejected=${result.rejected.length}'; + _status = result.message; }); } catch (error) { setState(() { - _status = 'Push failed. Local data remains safe. ($error)'; + _status = 'Sync operation failed unexpectedly. ($error)'; }); } finally { - setState(() { - _busy = false; - }); + if (mounted) { + setState(() { + _busy = false; + }); + } } } } diff --git a/test/features/sync/sync_coordinator_test.dart b/test/features/sync/sync_coordinator_test.dart new file mode 100644 index 0000000..281555d --- /dev/null +++ b/test/features/sync/sync_coordinator_test.dart @@ -0,0 +1,115 @@ +import 'package:flutter_riverpod/flutter_riverpod.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:relationship_saver/features/local/local_repository.dart'; +import 'package:relationship_saver/features/sync/sync_coordinator.dart'; +import 'package:relationship_saver/features/sync/sync_queue_repository.dart'; +import 'package:relationship_saver/features/sync/sync_state.dart'; +import 'package:relationship_saver/integrations/backend/backend_gateway_fake.dart'; +import 'package:relationship_saver/integrations/backend/backend_gateway_provider.dart'; +import 'package:relationship_saver/integrations/backend/models/backend_models.dart'; +import 'package:shared_preferences/shared_preferences.dart'; + +void main() { + setUp(() { + SharedPreferences.setMockInitialValues({}); + }); + + test('pushPending flushes queued local mutations', () async { + final ProviderContainer container = ProviderContainer( + overrides: [backendGatewayProvider.overrideWithValue(_TestGateway())], + ); + addTearDown(container.dispose); + + await container.read(localRepositoryProvider.future); + await container + .read(localRepositoryProvider.notifier) + .addPerson( + name: 'Queued Person', + relationship: 'Friend', + notes: 'sync me', + tags: ['test'], + ); + + final SyncState before = await container.read( + syncQueueRepositoryProvider.future, + ); + expect(before.pendingChanges.length, 1); + + final SyncRunResult result = await container + .read(syncCoordinatorProvider) + .pushPending(); + + expect(result.success, isTrue); + expect(result.accepted, 1); + + final SyncState after = await container.read( + syncQueueRepositoryProvider.future, + ); + expect(after.pendingChanges, isEmpty); + expect(after.cursor, isNotNull); + }); + + test('pullRemote applies backend changes into local repository', () async { + final SyncPullResult pullResult = SyncPullResult( + cursor: 'cursor-remote-1', + changes: [ + ChangeEnvelope( + schemaVersion: 1, + entityType: 'person', + entityId: 'p-remote-1', + op: ChangeOperation.upsert, + modifiedAt: DateTime(2026, 2, 15, 8).toUtc(), + clientMutationId: 'remote-1', + payload: const { + 'name': 'Remote Person', + 'relationship': 'Sibling', + 'affinityScore': 88, + 'nextMoment': '2026-02-20T12:00:00Z', + 'tags': ['remote'], + 'notes': 'arrived from backend', + }, + ), + ], + ); + + final ProviderContainer container = ProviderContainer( + overrides: [ + backendGatewayProvider.overrideWithValue( + _TestGateway(pullResult: pullResult), + ), + ], + ); + addTearDown(container.dispose); + + await container.read(localRepositoryProvider.future); + + final SyncRunResult result = await container + .read(syncCoordinatorProvider) + .pullRemote(); + + expect(result.success, isTrue); + expect(result.pulled, 1); + + final localState = await container.read(localRepositoryProvider.future); + final bool hasRemote = localState.people.any( + (person) => person.id == 'p-remote-1' && person.name == 'Remote Person', + ); + expect(hasRemote, isTrue); + + final SyncState queueState = await container.read( + syncQueueRepositoryProvider.future, + ); + expect(queueState.cursor, 'cursor-remote-1'); + }); +} + +class _TestGateway extends BackendGatewayFake { + _TestGateway({this.pullResult}); + + final SyncPullResult? pullResult; + + @override + Future pullChanges({String? cursor, int limit = 200}) async { + return pullResult ?? super.pullChanges(cursor: cursor, limit: limit); + } +} diff --git a/test/features/sync/sync_queue_repository_test.dart b/test/features/sync/sync_queue_repository_test.dart new file mode 100644 index 0000000..5d7c091 --- /dev/null +++ b/test/features/sync/sync_queue_repository_test.dart @@ -0,0 +1,103 @@ +import 'package:flutter_riverpod/flutter_riverpod.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:relationship_saver/features/sync/sync_queue_repository.dart'; +import 'package:relationship_saver/integrations/backend/models/backend_models.dart'; +import 'package:shared_preferences/shared_preferences.dart'; + +void main() { + setUp(() { + SharedPreferences.setMockInitialValues({}); + }); + + test('enqueue persists and can be reloaded', () async { + final ProviderContainer container = ProviderContainer(); + + await container + .read(syncQueueRepositoryProvider.notifier) + .enqueue(_change(id: 'p-1', mutationId: 'cm-1')); + final SyncStateData firstState = await _readState(container); + expect(firstState.pending.length, 1); + + container.dispose(); + + final ProviderContainer reloaded = ProviderContainer(); + addTearDown(reloaded.dispose); + final SyncStateData reloadedState = await _readState(reloaded); + expect(reloadedState.pending.length, 1); + expect(reloadedState.pending.first.clientMutationId, 'cm-1'); + }); + + test( + 'applyPushResult removes completed mutation ids and tracks rejections', + () async { + final ProviderContainer container = ProviderContainer(); + addTearDown(container.dispose); + + final SyncQueueRepository notifier = container.read( + syncQueueRepositoryProvider.notifier, + ); + + await notifier.enqueueAll([ + _change(id: 'p-1', mutationId: 'cm-a'), + _change(id: 'p-2', mutationId: 'cm-b'), + ]); + + await notifier.applyPushResult( + SyncPushResult( + cursor: 'cursor-2', + accepted: const [MutationAck(clientMutationId: 'cm-a')], + rejected: const [ + MutationRejection( + clientMutationId: 'cm-b', + code: 'VALIDATION_FAILED', + message: 'invalid payload', + ), + ], + ), + at: DateTime(2026, 2, 15, 12), + ); + + final SyncStateData state = await _readState(container); + expect(state.pending, isEmpty); + expect(state.cursor, 'cursor-2'); + expect(state.rejected.length, 1); + expect(state.lastError, contains('Push rejected')); + }, + ); +} + +class SyncStateData { + const SyncStateData({ + required this.cursor, + required this.pending, + required this.rejected, + required this.lastError, + }); + + final String? cursor; + final List pending; + final List rejected; + final String? lastError; +} + +Future _readState(ProviderContainer container) async { + final state = await container.read(syncQueueRepositoryProvider.future); + return SyncStateData( + cursor: state.cursor, + pending: state.pendingChanges, + rejected: state.lastRejected, + lastError: state.lastError, + ); +} + +ChangeEnvelope _change({required String id, required String mutationId}) { + return ChangeEnvelope( + schemaVersion: 1, + entityType: 'person', + entityId: id, + op: ChangeOperation.upsert, + modifiedAt: DateTime(2026, 2, 15, 12).toUtc(), + clientMutationId: mutationId, + payload: const {'name': 'Alex', 'relationship': 'Friend'}, + ); +}