bvandeusen/FabledApp
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 32 Wiki Activity

feat(offline): tier 2 phase 3 — write queue with optimistic UI

Browse Source 
Offline writes now queue + apply optimistically instead of throwing.
On reconnect, the queue drains automatically; conflicts and rejections
surface as one-time SnackBars so the user knows their edit didn't land.

- Drift schema v3: `pending_writes` table (verb + payload + baseline +
  tries + last_error). Negative ids serve as cache placeholders for
  offline-created rows until replay assigns the server id.
- Each write repository (notes, tasks, projects, milestones, events)
  now catches NetworkException, applies the change to cache (with a
  fresh `nextTempId()` for creates), and enqueues the API call.
- Edits to a still-queued offline-created row coalesce into the
  original create payload — only one server call per row, in order.
- Deletes of still-queued offline-created rows drop the queue entry
  and the cache row; no server call ever happens.
- New WriteQueue service drains the queue oldest-first.
  Server-wins on `updated_at` baseline check (notes/tasks/projects);
  last-writer-wins for milestones/events (no getOne available).
  4xx → drop + surface as `rejected`; conflict → drop + `overwritten`;
  404 on update → drop + `missing`; 5xx/network → keep + retry next
  online cycle.
- Replay fires on AuthStatus → authenticated transitions
  (cold-start, came-back-online, login).
- OfflineBanner shows "Retry (N)" with the pending-writes count.
- Distinct ServerException added so 4xx no longer masquerade as
  NetworkException — the queue can drop them instead of looping.

flutter analyze clean; 21 tests pass (3 new for QueueFailure messaging).
Phase 4 (read-only UI indicators on cached/temp rows) still ahead.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
bvandeusen
2026-04-28 22:08:40 -04:00
parent fe10067761
commit 7df7b5ff85
14 changed files with 2090 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/data/repositories/events_repository.dart
+111 -12
View File
@@ -3,10 +3,13 @@ import '../api/events_api.dart';
import '../local/database.dart';
import '../models/calendar_event.dart';
/// Calendar events repository with read-through caching for Tier 2 offline
/// support. Range-scoped: each `getEvents(from, to)` mirrors that window into
/// the cache (events outside the window are left alone) so successive
/// disjoint range fetches don't clobber each other.
/// Calendar events repository with read-through caching (Phase 1+2) and
/// offline-write queueing (Phase 3). Range-scoped reads: `getEvents(from, to)`
/// mirrors that window into the cache (events outside the window are left
/// alone) so successive disjoint range fetches don't clobber each other.
///
/// CalendarEvent has no `updatedAt` field, so write replay uses last-
/// writer-wins on update/delete (no server-side baseline check).
class EventsRepository {
final EventsApi _api;
final FabledDatabase _db;
@@ -26,20 +29,116 @@ class EventsRepository {
}
Future<CalendarEvent> createEvent(Map<String, dynamic> payload) async {
final event = await _api.createEvent(payload);
await _db.upsertEvent(event);
return event;
try {
final event = await _api.createEvent(payload);
await _db.upsertEvent(event);
return event;
} on NetworkException {
final tempId = await _db.nextTempId();
final optimistic = _eventFromPayload(tempId, payload);
await _db.upsertEvent(optimistic);
await _db.enqueuePending(
domain: kSyncDomainEvents,
verb: kWriteVerbCreate,
tempId: tempId,
payload: payload,
);
return optimistic;
}
}
Future<CalendarEvent> updateEvent(
int id, Map<String, dynamic> fields) async {
final updated = await _api.updateEvent(id, fields);
await _db.upsertEvent(updated);
return updated;
try {
final updated = await _api.updateEvent(id, fields);
await _db.upsertEvent(updated);
return updated;
} on NetworkException {
final cached = (await _db.getEventsInRange(
DateTime.fromMicrosecondsSinceEpoch(0),
DateTime.now().add(const Duration(days: 365 * 100)),
))
.where((e) => e.id == id)
.firstOrNull;
if (cached == null) rethrow;
final optimistic = _applyEventFields(cached, fields);
await _db.upsertEvent(optimistic);
if (id < 0) {
final queued = await _db.findQueuedCreate(kSyncDomainEvents, id);
if (queued != null) {
await _db.updatePendingPayload(queued.id, fields);
return optimistic;
}
}
await _db.enqueuePending(
domain: kSyncDomainEvents,
verb: kWriteVerbUpdate,
targetId: id,
payload: fields,
);
return optimistic;
}
}
Future<void> deleteEvent(int id) async {
await _api.deleteEvent(id);
await _db.deleteEvent(id);
try {
await _api.deleteEvent(id);
await _db.deleteEvent(id);
} on NetworkException {
if (id < 0) {
final queued = await _db.findQueuedCreate(kSyncDomainEvents, id);
if (queued != null) await _db.deletePending(queued.id);
await _db.deleteEvent(id);
return;
}
await _db.deleteEvent(id);
await _db.enqueuePending(
domain: kSyncDomainEvents,
verb: kWriteVerbDelete,
targetId: id,
);
}
}
}
CalendarEvent _eventFromPayload(int id, Map<String, dynamic> p) {
return CalendarEvent(
id: id,
title: p['title'] as String? ?? '',
startDt: _parseIso(p['start_dt'])!,
endDt: _parseIso(p['end_dt']),
allDay: p['all_day'] as bool? ?? false,
description: p['description'] as String? ?? '',
location: p['location'] as String? ?? '',
color: p['color'] as String? ?? '',
recurrence: p['recurrence'] as String?,
projectId: p['project_id'] as int?,
reminderMinutes: p['reminder_minutes'] as int?,
);
}
CalendarEvent _applyEventFields(CalendarEvent e, Map<String, dynamic> f) {
return CalendarEvent(
id: e.id,
title: f['title'] as String? ?? e.title,
startDt: _parseIso(f['start_dt']) ?? e.startDt,
endDt: f.containsKey('end_dt') ? _parseIso(f['end_dt']) : e.endDt,
allDay: f['all_day'] as bool? ?? e.allDay,
description: f['description'] as String? ?? e.description,
location: f['location'] as String? ?? e.location,
color: f['color'] as String? ?? e.color,
recurrence:
f.containsKey('recurrence') ? f['recurrence'] as String? : e.recurrence,
projectId: f.containsKey('project_id')
? f['project_id'] as int?
: e.projectId,
reminderMinutes: f.containsKey('reminder_minutes')
? f['reminder_minutes'] as int?
: e.reminderMinutes,
);
}
DateTime? _parseIso(dynamic raw) {
if (raw is String && raw.isNotEmpty) return DateTime.tryParse(raw)?.toLocal();
return null;
}
lib/data/repositories/milestones_repository.dart
+112 -15
View File
@@ -3,8 +3,9 @@ import '../api/milestones_api.dart';
import '../local/database.dart';
import '../models/milestone.dart';
/// Milestones repository with read-through caching for Tier 2 offline support.
/// Mirrors the pattern in NotesRepository — see that file for full notes.
/// Milestones repository with read-through caching (Phase 1+2) and
/// offline-write queueing (Phase 3). See `notes_repository.dart` for the
/// full pattern.
class MilestonesRepository {
final MilestonesApi _api;
final FabledDatabase _db;
@@ -40,14 +41,45 @@ class MilestonesRepository {
String? description,
int orderIndex = 0,
}) async {
final milestone = await _api.create(
projectId,
title: title,
description: description,
orderIndex: orderIndex,
);
await _db.upsertMilestone(milestone);
return milestone;
try {
final milestone = await _api.create(
projectId,
title: title,
description: description,
orderIndex: orderIndex,
);
await _db.upsertMilestone(milestone);
return milestone;
} on NetworkException {
final tempId = await _db.nextTempId();
final now = DateTime.now();
final optimistic = Milestone(
id: tempId,
projectId: projectId,
title: title,
description: description,
status: 'active',
orderIndex: orderIndex,
total: 0,
completed: 0,
pct: 0.0,
createdAt: now,
updatedAt: now,
);
await _db.upsertMilestone(optimistic);
await _db.enqueuePending(
domain: kSyncDomainMilestones,
verb: kWriteVerbCreate,
tempId: tempId,
payload: {
'project_id': projectId,
'title': title,
'description': description,
'order_index': orderIndex,
},
);
return optimistic;
}
}
Future<Milestone> update(
@@ -55,13 +87,78 @@ class MilestonesRepository {
int milestoneId,
Map<String, dynamic> fields,
) async {
final updated = await _api.update(projectId, milestoneId, fields);
await _db.upsertMilestone(updated);
return updated;
try {
final updated = await _api.update(projectId, milestoneId, fields);
await _db.upsertMilestone(updated);
return updated;
} on NetworkException {
final cached = (await _db.getMilestonesForProject(projectId))
.where((m) => m.id == milestoneId)
.firstOrNull;
if (cached == null) rethrow;
final optimistic = _applyMilestoneFields(cached, fields);
await _db.upsertMilestone(optimistic);
if (milestoneId < 0) {
final queued =
await _db.findQueuedCreate(kSyncDomainMilestones, milestoneId);
if (queued != null) {
// Carry project_id through coalesced payload — replay needs it.
final merged = <String, dynamic>{
'project_id': projectId,
...fields,
};
await _db.updatePendingPayload(queued.id, merged);
return optimistic;
}
}
await _db.enqueuePending(
domain: kSyncDomainMilestones,
verb: kWriteVerbUpdate,
targetId: milestoneId,
payload: <String, dynamic>{'project_id': projectId, ...fields},
baselineUpdatedAt: cached.updatedAt,
);
return optimistic;
}
}
Future<void> delete(int projectId, int milestoneId) async {
await _api.delete(projectId, milestoneId);
await _db.deleteMilestone(milestoneId);
try {
await _api.delete(projectId, milestoneId);
await _db.deleteMilestone(milestoneId);
} on NetworkException {
if (milestoneId < 0) {
final queued =
await _db.findQueuedCreate(kSyncDomainMilestones, milestoneId);
if (queued != null) await _db.deletePending(queued.id);
await _db.deleteMilestone(milestoneId);
return;
}
await _db.deleteMilestone(milestoneId);
await _db.enqueuePending(
domain: kSyncDomainMilestones,
verb: kWriteVerbDelete,
targetId: milestoneId,
payload: {'project_id': projectId},
);
}
}
}
Milestone _applyMilestoneFields(Milestone m, Map<String, dynamic> f) {
return Milestone(
id: m.id,
projectId: m.projectId,
title: f['title'] as String? ?? m.title,
description: f.containsKey('description')
? f['description'] as String?
: m.description,
status: f['status'] as String? ?? m.status,
orderIndex: f['order_index'] as int? ?? m.orderIndex,
total: m.total,
completed: m.completed,
pct: m.pct,
createdAt: m.createdAt,
updatedAt: m.updatedAt,
);
}
lib/data/repositories/notes_repository.dart
+115 -29
View File
@@ -3,19 +3,22 @@ import '../api/notes_api.dart';
import '../local/database.dart';
import '../models/note.dart';
/// Notes repository with read-through caching for Tier 2 offline support.
/// Notes repository with read-through caching for Tier 2 offline support
/// (Phase 1+2) and offline-write queueing (Phase 3).
///
/// Reads attempt the network first; on success the response is written to
/// the local Drift cache. On `NetworkException` the read falls back to the
/// cache (rethrowing if the cache is empty so the UI can show its
/// fresh-install empty state). Writes (create / update / delete) hit the
/// server then sync the cache; offline write queueing is Phase 3 work and
/// not yet wired here — write methods will throw on `NetworkException` in
/// the meantime.
/// fresh-install empty state).
///
/// Writes hit the server then sync the cache. On `NetworkException` the
/// write is queued in `pending_writes` and applied optimistically to the
/// cache (negative temp-id for creates, in-place update for edits, removal
/// for deletes). Subsequent edits to a temp-id row are coalesced into the
/// queued create — only one server call is made per offline-created row.
///
/// `AuthStatus.offline` is owned by `AuthNotifier.verify()` (a periodic
/// heartbeat). This repository deliberately does not poke that state — its
/// only job is to serve what's cached when the network is unreachable.
/// heartbeat). This repository deliberately does not poke that state.
class NotesRepository {
final NotesApi _api;
final FabledDatabase _db;
@@ -53,15 +56,44 @@ class NotesRepository {
int? projectId,
String noteType = 'note',
}) async {
final note = await _api.create(
title,
body,
tags: tags,
projectId: projectId,
noteType: noteType,
);
await _db.upsertNote(note);
return note;
try {
final note = await _api.create(
title,
body,
tags: tags,
projectId: projectId,
noteType: noteType,
);
await _db.upsertNote(note);
return note;
} on NetworkException {
final tempId = await _db.nextTempId();
final now = DateTime.now();
final optimistic = Note(
id: tempId,
title: title,
body: body,
tags: tags,
noteType: noteType,
projectId: projectId,
createdAt: now,
updatedAt: now,
);
await _db.upsertNote(optimistic);
await _db.enqueuePending(
domain: kSyncDomainNotes,
verb: kWriteVerbCreate,
tempId: tempId,
payload: {
'title': title,
'body': body,
'tags': tags,
'project_id': projectId,
'note_type': noteType,
},
);
return optimistic;
}
}
Future<Note> update(
@@ -73,21 +105,75 @@ class NotesRepository {
bool clearProject = false,
String noteType = 'note',
}) async {
final updated = await _api.update(
id,
title,
body,
tags: tags,
projectId: projectId,
clearProject: clearProject,
noteType: noteType,
);
await _db.upsertNote(updated);
return updated;
try {
final updated = await _api.update(
id,
title,
body,
tags: tags,
projectId: projectId,
clearProject: clearProject,
noteType: noteType,
);
await _db.upsertNote(updated);
return updated;
} on NetworkException {
final cached = await _db.getNote(id);
if (cached == null) rethrow;
final payload = <String, dynamic>{
'title': title,
'body': body,
'tags': tags,
'project_id': projectId,
'clear_project': clearProject,
'note_type': noteType,
};
final optimistic = cached.copyWith(
title: title,
body: body,
tags: tags,
noteType: noteType,
projectId: clearProject ? null : projectId,
);
await _db.upsertNote(optimistic);
// Edits to an offline-created row coalesce into the queued create.
if (id < 0) {
final queued = await _db.findQueuedCreate(kSyncDomainNotes, id);
if (queued != null) {
await _db.updatePendingPayload(queued.id, payload);
return optimistic;
}
}
await _db.enqueuePending(
domain: kSyncDomainNotes,
verb: kWriteVerbUpdate,
targetId: id,
payload: payload,
baselineUpdatedAt: cached.updatedAt,
);
return optimistic;
}
}
Future<void> delete(int id) async {
await _api.delete(id);
await _db.deleteNote(id);
try {
await _api.delete(id);
await _db.deleteNote(id);
} on NetworkException {
// Deleting an offline-created row that never reached the server
// just drops the queued create — no server call needed.
if (id < 0) {
final queued = await _db.findQueuedCreate(kSyncDomainNotes, id);
if (queued != null) await _db.deletePending(queued.id);
await _db.deleteNote(id);
return;
}
await _db.deleteNote(id);
await _db.enqueuePending(
domain: kSyncDomainNotes,
verb: kWriteVerbDelete,
targetId: id,
);
}
}
}
lib/data/repositories/projects_repository.dart
+97 -16
View File
@@ -3,8 +3,8 @@ import '../api/projects_api.dart';
import '../local/database.dart';
import '../models/project.dart';
/// Projects repository with read-through caching for Tier 2 offline support.
/// Mirrors the pattern in NotesRepository — see that file for full notes.
/// Projects repository with read-through caching (Phase 1+2) and offline-write
/// queueing (Phase 3). See `notes_repository.dart` for the full pattern.
///
/// `getAll`'s sort/order/status query parameters are server-side filters; the
/// cache stores the unfiltered list as last seen. Offline fallback returns
@@ -63,25 +63,106 @@ class ProjectsRepository {
String? color,
String status = 'active',
}) async {
final project = await _api.create(
title: title,
description: description,
goal: goal,
color: color,
status: status,
);
await _db.upsertProject(project);
return project;
try {
final project = await _api.create(
title: title,
description: description,
goal: goal,
color: color,
status: status,
);
await _db.upsertProject(project);
return project;
} on NetworkException {
final tempId = await _db.nextTempId();
final now = DateTime.now();
final optimistic = Project(
id: tempId,
title: title,
description: description,
goal: goal,
status: status,
color: color,
createdAt: now,
updatedAt: now,
);
await _db.upsertProject(optimistic);
await _db.enqueuePending(
domain: kSyncDomainProjects,
verb: kWriteVerbCreate,
tempId: tempId,
payload: {
'title': title,
'description': description,
'goal': goal,
'color': color,
'status': status,
},
);
return optimistic;
}
}
Future<Project> update(int id, Map<String, dynamic> fields) async {
final updated = await _api.update(id, fields);
await _db.upsertProject(updated);
return updated;
try {
final updated = await _api.update(id, fields);
await _db.upsertProject(updated);
return updated;
} on NetworkException {
final cached = await _db.getProject(id);
if (cached == null) rethrow;
final optimistic = _applyProjectFields(cached, fields);
await _db.upsertProject(optimistic);
if (id < 0) {
final queued = await _db.findQueuedCreate(kSyncDomainProjects, id);
if (queued != null) {
await _db.updatePendingPayload(queued.id, fields);
return optimistic;
}
}
await _db.enqueuePending(
domain: kSyncDomainProjects,
verb: kWriteVerbUpdate,
targetId: id,
payload: fields,
baselineUpdatedAt: cached.updatedAt,
);
return optimistic;
}
}
Future<void> delete(int id) async {
await _api.delete(id);
await _db.deleteProject(id);
try {
await _api.delete(id);
await _db.deleteProject(id);
} on NetworkException {
if (id < 0) {
final queued = await _db.findQueuedCreate(kSyncDomainProjects, id);
if (queued != null) await _db.deletePending(queued.id);
await _db.deleteProject(id);
return;
}
await _db.deleteProject(id);
await _db.enqueuePending(
domain: kSyncDomainProjects,
verb: kWriteVerbDelete,
targetId: id,
);
}
}
}
Project _applyProjectFields(Project p, Map<String, dynamic> f) {
return Project(
id: p.id,
title: f['title'] as String? ?? p.title,
description:
f.containsKey('description') ? f['description'] as String? : p.description,
goal: f.containsKey('goal') ? f['goal'] as String? : p.goal,
status: f['status'] as String? ?? p.status,
color: f.containsKey('color') ? f['color'] as String? : p.color,
autoSummary: p.autoSummary,
createdAt: p.createdAt,
updatedAt: p.updatedAt,
);
}
lib/data/repositories/tasks_repository.dart
+120 -18
View File
@@ -3,8 +3,8 @@ import '../api/tasks_api.dart';
import '../local/database.dart';
import '../models/task.dart';
/// Tasks repository with read-through caching for Tier 2 offline support.
/// Mirrors the pattern in NotesRepository — see that file for full notes.
/// Tasks repository with read-through caching (Phase 1+2) and offline-write
/// queueing (Phase 3). See `notes_repository.dart` for the full pattern.
class TasksRepository {
final TasksApi _api;
final FabledDatabase _db;
@@ -68,27 +68,129 @@ class TasksRepository {
int? projectId,
int? parentId,
}) async {
final task = await _api.create(
title: title,
description: description,
status: status,
priority: priority,
dueDate: dueDate,
projectId: projectId,
parentId: parentId,
);
await _db.upsertTask(task);
return task;
try {
final task = await _api.create(
title: title,
description: description,
status: status,
priority: priority,
dueDate: dueDate,
projectId: projectId,
parentId: parentId,
);
await _db.upsertTask(task);
return task;
} on NetworkException {
final tempId = await _db.nextTempId();
final now = DateTime.now();
final optimistic = Task(
id: tempId,
title: title,
description: description,
status: status,
priority: priority,
dueDate: dueDate,
projectId: projectId,
parentId: parentId,
createdAt: now,
updatedAt: now,
);
await _db.upsertTask(optimistic);
await _db.enqueuePending(
domain: kSyncDomainTasks,
verb: kWriteVerbCreate,
tempId: tempId,
payload: {
'title': title,
'description': description,
'status': status.value,
'priority': priority.value,
'due_date': dueDate?.toIso8601String(),
'project_id': projectId,
'parent_id': parentId,
},
);
return optimistic;
}
}
Future<Task> update(int id, Map<String, dynamic> fields) async {
final updated = await _api.update(id, fields);
await _db.upsertTask(updated);
return updated;
try {
final updated = await _api.update(id, fields);
await _db.upsertTask(updated);
return updated;
} on NetworkException {
final cached = await _db.getTask(id);
if (cached == null) rethrow;
final optimistic = _applyTaskFields(cached, fields);
await _db.upsertTask(optimistic);
if (id < 0) {
final queued = await _db.findQueuedCreate(kSyncDomainTasks, id);
if (queued != null) {
await _db.updatePendingPayload(queued.id, fields);
return optimistic;
}
}
await _db.enqueuePending(
domain: kSyncDomainTasks,
verb: kWriteVerbUpdate,
targetId: id,
payload: fields,
baselineUpdatedAt: cached.updatedAt,
);
return optimistic;
}
}
Future<void> delete(int id) async {
await _api.delete(id);
await _db.deleteTask(id);
try {
await _api.delete(id);
await _db.deleteTask(id);
} on NetworkException {
if (id < 0) {
final queued = await _db.findQueuedCreate(kSyncDomainTasks, id);
if (queued != null) await _db.deletePending(queued.id);
await _db.deleteTask(id);
return;
}
await _db.deleteTask(id);
await _db.enqueuePending(
domain: kSyncDomainTasks,
verb: kWriteVerbDelete,
targetId: id,
);
}
}
}
/// Applies a partial-fields map (the same shape sent to the PUT endpoint)
/// to a cached Task so the UI can show the edit immediately.
Task _applyTaskFields(Task t, Map<String, dynamic> f) {
return Task(
id: t.id,
title: f['title'] as String? ?? t.title,
description:
f.containsKey('body') ? f['body'] as String? : t.description,
status: f.containsKey('status')
? TaskStatusExtension.fromString(f['status'] as String?)
: t.status,
priority: f.containsKey('priority')
? TaskPriorityExtension.fromString(f['priority'] as String?)
: t.priority,
dueDate: f.containsKey('due_date')
? (f['due_date'] is String
? DateTime.tryParse(f['due_date'] as String)
: null)
: t.dueDate,
projectId: f.containsKey('project_id')
? f['project_id'] as int?
: t.projectId,
milestoneId: f.containsKey('milestone_id')
? f['milestone_id'] as int?
: t.milestoneId,
parentId:
f.containsKey('parent_id') ? f['parent_id'] as int? : t.parentId,
createdAt: t.createdAt,
updatedAt: t.updatedAt,
);
}
lib/data/repositories/write_queue.dart
+385
View File
@@ -0,0 +1,385 @@
import 'dart:async';
import 'dart:convert';
import '../../core/exceptions.dart';
import '../api/events_api.dart';
import '../api/milestones_api.dart';
import '../api/notes_api.dart';
import '../api/projects_api.dart';
import '../api/tasks_api.dart';
import '../local/database.dart';
import '../models/note.dart';
import '../models/project.dart';
import '../models/task.dart';
/// Reasons a queued write was dropped during replay. Surfaced to the UI as
/// a one-time SnackBar so the user knows their offline edit didn't land.
enum QueueFailureReason { overwritten, missing, rejected }
class QueueFailure {
final QueueFailureReason reason;
final String domain;
final String? title;
final String? detail;
const QueueFailure({
required this.reason,
required this.domain,
this.title,
this.detail,
});
String get message {
final label = title?.isNotEmpty == true ? '"$title"' : 'an offline edit';
return switch (reason) {
QueueFailureReason.overwritten =>
'$label was overwritten by a newer change on the server.',
QueueFailureReason.missing =>
'$label was deleted on the server before your offline edit could be saved.',
QueueFailureReason.rejected =>
'Failed to save $label: ${detail ?? 'rejected by server.'}',
};
}
}
/// Drains the offline write queue (Phase 3). Owns the network → server
/// dispatch logic for every queued op; repos write to `pending_writes`,
/// this service replays them.
///
/// Replay semantics:
/// - oldest-first, single in-flight `replay()` call
/// - on `NetworkException` (or 5xx): leave op in place, stop the loop
/// - on conflict (server `updated_at` newer than op baseline): drop +
/// surface as `overwritten`
/// - on 404: for delete → silent success; for update → drop + `missing`
/// - on other 4xx: drop + `rejected`
class WriteQueue {
final FabledDatabase _db;
final NotesApi _notesApi;
final TasksApi _tasksApi;
final ProjectsApi _projectsApi;
final MilestonesApi _milestonesApi;
final EventsApi _eventsApi;
final StreamController<QueueFailure> _failures =
StreamController<QueueFailure>.broadcast();
bool _replaying = false;
WriteQueue({
required FabledDatabase db,
required NotesApi notesApi,
required TasksApi tasksApi,
required ProjectsApi projectsApi,
required MilestonesApi milestonesApi,
required EventsApi eventsApi,
}) : _db = db,
_notesApi = notesApi,
_tasksApi = tasksApi,
_projectsApi = projectsApi,
_milestonesApi = milestonesApi,
_eventsApi = eventsApi;
Stream<QueueFailure> get failures => _failures.stream;
void dispose() => _failures.close();
/// Drain the queue. Reentrant calls are no-ops while a replay is in
/// flight — the in-flight one will see new entries on its next iteration.
Future<void> replay() async {
if (_replaying) return;
_replaying = true;
try {
while (true) {
final ops = await _db.listPending();
if (ops.isEmpty) break;
final op = ops.first;
final keepGoing = await _replayOne(op);
if (!keepGoing) break;
}
} finally {
_replaying = false;
}
}
Future<bool> _replayOne(PendingWrite op) async {
final payload = jsonDecode(op.payloadJson) as Map<String, dynamic>;
await _db.incrementPendingTries(op.id);
try {
switch (op.domain) {
case kSyncDomainNotes:
await _replayNote(op, payload);
case kSyncDomainTasks:
await _replayTask(op, payload);
case kSyncDomainProjects:
await _replayProject(op, payload);
case kSyncDomainMilestones:
await _replayMilestone(op, payload);
case kSyncDomainEvents:
await _replayEvent(op, payload);
default:
// Unknown domain — drop so we don't loop forever.
await _db.deletePending(op.id);
return true;
}
await _db.deletePending(op.id);
return true;
} on NetworkException catch (e) {
// Still offline — leave op for next replay trigger.
await _db.markPendingFailed(op.id, e.message);
return false;
} on ServerException catch (e) {
if (e.statusCode >= 500) {
// Likely transient — keep + stop the loop, retry next time.
await _db.markPendingFailed(op.id, e.message);
return false;
}
// 4xx — non-retryable. Drop + surface.
await _db.deletePending(op.id);
_failures.add(QueueFailure(
reason: QueueFailureReason.rejected,
domain: op.domain,
title: payload['title'] as String?,
detail: e.message,
));
return true;
} on NotFoundException {
// Target gone server-side. Update → tell user; delete → silent success.
await _db.deletePending(op.id);
if (op.verb != kWriteVerbDelete) {
_failures.add(QueueFailure(
reason: QueueFailureReason.missing,
domain: op.domain,
title: payload['title'] as String?,
));
}
// Clean the cache so the optimistic row doesn't linger.
await _evictCache(op.domain, op.targetId ?? op.tempId);
return true;
} on _OverwrittenSignal catch (s) {
await _db.deletePending(op.id);
_failures.add(QueueFailure(
reason: QueueFailureReason.overwritten,
domain: op.domain,
title: s.title,
));
return true;
} on AppException catch (e) {
// Auth or other — surface and drop so the queue can drain.
await _db.deletePending(op.id);
_failures.add(QueueFailure(
reason: QueueFailureReason.rejected,
domain: op.domain,
title: payload['title'] as String?,
detail: e.message,
));
return true;
}
}
Future<void> _evictCache(String domain, int? id) async {
if (id == null) return;
switch (domain) {
case kSyncDomainNotes:
await _db.deleteNote(id);
case kSyncDomainTasks:
await _db.deleteTask(id);
case kSyncDomainProjects:
await _db.deleteProject(id);
case kSyncDomainMilestones:
await _db.deleteMilestone(id);
case kSyncDomainEvents:
await _db.deleteEvent(id);
}
}
// ── Notes ──────────────────────────────────────────────────────────────────
Future<void> _replayNote(PendingWrite op, Map<String, dynamic> p) async {
switch (op.verb) {
case kWriteVerbCreate:
final note = await _notesApi.create(
p['title'] as String? ?? '',
p['body'] as String? ?? '',
tags: _stringList(p['tags']),
projectId: p['project_id'] as int?,
noteType: p['note_type'] as String? ?? 'note',
);
if (op.tempId != null) await _db.deleteNote(op.tempId!);
await _db.upsertNote(note);
case kWriteVerbUpdate:
await _conflictGuard<Note>(
baseline: op.baselineUpdatedAt,
fetch: () => _notesApi.getOne(op.targetId!),
updatedAt: (n) => n.updatedAt,
title: (n) => n.title,
syncCache: (n) => _db.upsertNote(n),
);
final note = await _notesApi.update(
op.targetId!,
p['title'] as String? ?? '',
p['body'] as String? ?? '',
tags: _stringList(p['tags']),
projectId: p['project_id'] as int?,
clearProject: p['clear_project'] as bool? ?? false,
noteType: p['note_type'] as String? ?? 'note',
);
await _db.upsertNote(note);
case kWriteVerbDelete:
try {
await _notesApi.delete(op.targetId!);
} on NotFoundException {/* already gone */}
await _db.deleteNote(op.targetId!);
}
}
// ── Tasks ──────────────────────────────────────────────────────────────────
Future<void> _replayTask(PendingWrite op, Map<String, dynamic> p) async {
switch (op.verb) {
case kWriteVerbCreate:
final task = await _tasksApi.create(
title: p['title'] as String? ?? '',
description: p['description'] as String?,
status: TaskStatusExtension.fromString(p['status'] as String?),
priority: TaskPriorityExtension.fromString(p['priority'] as String?),
dueDate: _parseDate(p['due_date']),
projectId: p['project_id'] as int?,
parentId: p['parent_id'] as int?,
);
if (op.tempId != null) await _db.deleteTask(op.tempId!);
await _db.upsertTask(task);
case kWriteVerbUpdate:
await _conflictGuard<Task>(
baseline: op.baselineUpdatedAt,
fetch: () => _tasksApi.getOne(op.targetId!),
updatedAt: (t) => t.updatedAt,
title: (t) => t.title,
syncCache: (t) => _db.upsertTask(t),
);
final task = await _tasksApi.update(op.targetId!, p);
await _db.upsertTask(task);
case kWriteVerbDelete:
try {
await _tasksApi.delete(op.targetId!);
} on NotFoundException {/* already gone */}
await _db.deleteTask(op.targetId!);
}
}
// ── Projects ───────────────────────────────────────────────────────────────
Future<void> _replayProject(PendingWrite op, Map<String, dynamic> p) async {
switch (op.verb) {
case kWriteVerbCreate:
final project = await _projectsApi.create(
title: p['title'] as String? ?? '',
description: p['description'] as String?,
goal: p['goal'] as String?,
color: p['color'] as String?,
status: p['status'] as String? ?? 'active',
);
if (op.tempId != null) await _db.deleteProject(op.tempId!);
await _db.upsertProject(project);
case kWriteVerbUpdate:
await _conflictGuard<Project>(
baseline: op.baselineUpdatedAt,
fetch: () => _projectsApi.getOne(op.targetId!),
updatedAt: (proj) => proj.updatedAt,
title: (proj) => proj.title,
syncCache: (proj) => _db.upsertProject(proj),
);
final project = await _projectsApi.update(op.targetId!, p);
await _db.upsertProject(project);
case kWriteVerbDelete:
try {
await _projectsApi.delete(op.targetId!);
} on NotFoundException {/* already gone */}
await _db.deleteProject(op.targetId!);
}
}
// ── Milestones ─────────────────────────────────────────────────────────────
Future<void> _replayMilestone(
PendingWrite op, Map<String, dynamic> p) async {
final projectId = p['project_id'] as int;
switch (op.verb) {
case kWriteVerbCreate:
final milestone = await _milestonesApi.create(
projectId,
title: p['title'] as String? ?? '',
description: p['description'] as String?,
orderIndex: p['order_index'] as int? ?? 0,
);
if (op.tempId != null) await _db.deleteMilestone(op.tempId!);
await _db.upsertMilestone(milestone);
case kWriteVerbUpdate:
// Milestones API has no getOne — skip the conflict pre-check and
// let the PUT apply unconditionally. If two clients fight over the
// same milestone, last-writer-wins is acceptable here.
final milestone = await _milestonesApi.update(
projectId,
op.targetId!,
Map<String, dynamic>.from(p)..remove('project_id'),
);
await _db.upsertMilestone(milestone);
case kWriteVerbDelete:
try {
await _milestonesApi.delete(projectId, op.targetId!);
} on NotFoundException {/* already gone */}
await _db.deleteMilestone(op.targetId!);
}
}
// ── Events ─────────────────────────────────────────────────────────────────
Future<void> _replayEvent(PendingWrite op, Map<String, dynamic> p) async {
switch (op.verb) {
case kWriteVerbCreate:
final event = await _eventsApi.createEvent(p);
if (op.tempId != null) await _db.deleteEvent(op.tempId!);
await _db.upsertEvent(event);
case kWriteVerbUpdate:
// Events API also has no getOne — skip the pre-check; last-writer-wins.
final event = await _eventsApi.updateEvent(op.targetId!, p);
await _db.upsertEvent(event);
case kWriteVerbDelete:
try {
await _eventsApi.deleteEvent(op.targetId!);
} on NotFoundException {/* already gone */}
await _db.deleteEvent(op.targetId!);
}
}
// ── Helpers ────────────────────────────────────────────────────────────────
/// Server-wins conflict guard. Fetch the target, compare its `updated_at`
/// against the baseline captured when the user made the offline edit;
/// if the server is newer, sync cache and signal the caller to drop.
Future<void> _conflictGuard<T>({
required DateTime? baseline,
required Future<T> Function() fetch,
required DateTime Function(T) updatedAt,
required String Function(T) title,
required Future<void> Function(T) syncCache,
}) async {
if (baseline == null) return;
final server = await fetch();
if (updatedAt(server).isAfter(baseline)) {
await syncCache(server);
throw _OverwrittenSignal(title(server));
}
}
List<String> _stringList(dynamic raw) =>
raw is List ? raw.map((e) => e.toString()).toList() : const [];
DateTime? _parseDate(dynamic raw) {
if (raw is String && raw.isNotEmpty) return DateTime.tryParse(raw);
return null;
}
}
class _OverwrittenSignal implements Exception {
final String title;
const _OverwrittenSignal(this.title);
}