Files
ingtrakapp/lib/services/pb_sync_service.dart
T
2026-09-10 23:59:00 +02:00

1546 lines
51 KiB
Dart

/// PocketBase pull/push for labels, annotations, and bank transactions.
library;
import 'dart:async';
import 'package:flutter/foundation.dart';
import 'package:flutter/material.dart';
import 'package:pocketbase/pocketbase.dart';
import '../models/app_cloud_prefs.dart';
import '../models/bank.dart';
import '../models/home_built_in_sections.dart';
import '../models/home_filter_widget.dart';
import '../models/overview_favourite.dart';
import '../models/transaction_meta.dart';
import '../widgets/label_filter_dialog.dart';
import 'ing_csv_import.dart';
import 'storage_service.dart';
/// Union local + remote bank rows (no data loss). Newest date first.
List<BankTransaction> mergeBankTransactionLists(
List<BankTransaction> local,
List<BankTransaction> remote,
) {
return mergeTransactions(local, remote);
}
/// Union Overview favourites by client id (local wins on conflict).
List<OverviewFavourite> mergeOverviewFavourites(
List<OverviewFavourite> local,
List<OverviewFavourite> remote,
) {
final byId = <String, OverviewFavourite>{
for (final fav in remote) fav.id: fav,
};
for (final fav in local) {
byId[fav.id] = fav;
}
return byId.values.toList(growable: false);
}
/// Union-merge so neither device loses annotations on first sync.
TxMetaStoreData mergeTxMetaStores(TxMetaStoreData a, TxMetaStoreData b) {
final byId = <String, TxMeta>{...a.byId};
for (final e in b.byId.entries) {
final existing = byId[e.key];
byId[e.key] = existing == null ? e.value : _mergeMeta(existing, e.value);
}
final labelsByMerchant = <String, List<String>>{
for (final e in a.labelsByMerchant.entries) e.key: List<String>.of(e.value),
};
for (final e in b.labelsByMerchant.entries) {
labelsByMerchant[e.key] = _unionIds(labelsByMerchant[e.key], e.value);
}
final notesByMerchant = <String, String>{...a.notesByMerchant};
for (final e in b.notesByMerchant.entries) {
final cur = notesByMerchant[e.key];
if (cur == null || cur.trim().isEmpty) {
notesByMerchant[e.key] = e.value;
} else if (e.value.trim().length > cur.trim().length) {
notesByMerchant[e.key] = e.value;
}
}
final exactTemplates = <String, TxExactTemplate>{...a.exactTemplates};
for (final e in b.exactTemplates.entries) {
final existing = exactTemplates[e.key];
exactTemplates[e.key] =
existing == null ? e.value : _mergeExact(existing, e.value);
}
final recurringTemplates = <String, RecurringTemplate>{
...a.recurringTemplates,
};
for (final e in b.recurringTemplates.entries) {
recurringTemplates.putIfAbsent(e.key, () => e.value);
}
final labelsByClientId = <String, TxLabel>{
for (final label in a.customLabels) label.id: label,
};
for (final label in b.customLabels) {
labelsByClientId.putIfAbsent(label.id, () => label);
}
final labelParents = <String, String>{...a.labelParents, ...b.labelParents};
return TxMetaStoreData(
byId: byId,
labelsByMerchant: labelsByMerchant,
notesByMerchant: notesByMerchant,
exactTemplates: exactTemplates,
recurringTemplates: recurringTemplates,
customLabels: labelsByClientId.values.toList(growable: false),
labelParents: labelParents,
);
}
TxMeta _mergeMeta(TxMeta a, TxMeta b) {
return TxMeta(
labelIds: _unionIds(a.labelIds, b.labelIds),
recurring: a.recurring || b.recurring,
internalTransfer: a.internalTransfer || b.internalTransfer,
notes: _preferNotes(a.notes, b.notes),
);
}
TxExactTemplate _mergeExact(TxExactTemplate a, TxExactTemplate b) {
return TxExactTemplate(
notes: _preferNotes(a.notes, b.notes),
recurring: a.recurring || b.recurring,
internalTransfer: a.internalTransfer || b.internalTransfer,
);
}
List<String> _unionIds(List<String>? a, List<String>? b) {
final out = <String>[];
final seen = <String>{};
for (final id in [...?a, ...?b]) {
if (id.isEmpty || id == 'uncategorized' || id == 'other') continue;
if (seen.add(id)) out.add(id);
}
return out;
}
String _preferNotes(String a, String b) {
final at = a.trim();
final bt = b.trim();
if (at.isEmpty) return b;
if (bt.isEmpty) return a;
return bt.length > at.length ? b : a;
}
bool _sameIdList(List<String> a, List<String> b) {
if (a.length != b.length) return false;
if (listEquals(a, b)) return true;
final sa = [...a]..sort();
final sb = [...b]..sort();
return listEquals(sa, sb);
}
enum _BatchOpKind { create, update, delete }
class _BatchOp {
const _BatchOp.create(this.body)
: kind = _BatchOpKind.create,
id = null;
const _BatchOp.update(this.id, this.body) : kind = _BatchOpKind.update;
const _BatchOp.delete(this.id)
: kind = _BatchOpKind.delete,
body = null;
final _BatchOpKind kind;
final String? id;
final Map<String, dynamic>? body;
}
/// Talks to PocketBase collections created by the sync migration.
class PbSyncService {
PbSyncService(this._pb);
final PocketBase _pb;
/// Parallel HTTP writes — batch API often hangs/stalls on large sets.
static const _writeConcurrency = 12;
/// Per-request timeout so a single hung call can't freeze sync forever.
static const _requestTimeout = Duration(seconds: 45);
/// Page size for list fetches (progress updates after each page).
static const _listPageSize = 200;
String get _userId {
final id = _pb.authStore.record?.id;
if (id == null || id.isEmpty) {
throw StateError('Not authenticated');
}
return id;
}
Future<T> _withTimeout<T>(Future<T> future, {String? label}) {
return future.timeout(
_requestTimeout,
onTimeout: () => throw TimeoutException(
label == null ? 'PocketBase request timed out' : '$label timed out',
),
);
}
/// Fetches all matching records page-by-page so UI progress can move.
Future<List<RecordModel>> _fullList(
String collection,
String filter, {
void Function(int loaded)? onPage,
}) async {
final out = <RecordModel>[];
var page = 1;
var lastReported = -1;
while (true) {
final result = await _withTimeout(
_pb.collection(collection).getList(
page: page,
perPage: _listPageSize,
filter: filter,
skipTotal: true,
),
label: '$collection list page $page',
);
out.addAll(result.items);
final finished = result.items.length < _listPageSize;
if (onPage != null &&
(finished || out.length - lastReported >= 400)) {
lastReported = out.length;
onPage(out.length);
}
if (finished) break;
page++;
}
return out;
}
Future<TxMetaStoreData> pull({
void Function(double progress)? onProgress,
void Function(String message)? onStatus,
}) async {
final userId = _userId;
final filter = 'user="$userId"';
void report(double p) => onProgress?.call(p.clamp(0.0, 1.0));
report(0.05);
onStatus?.call('Pulling labels…');
final labelRecords = await _fullList('labels', filter);
report(0.12);
onStatus?.call('Pulling tx_meta…');
final metaRecords = await _fullList(
'tx_meta',
filter,
onPage: (n) => onStatus?.call('Pulling tx_meta ($n)…'),
);
report(0.28);
onStatus?.call('Pulling merchant rules…');
final merchantRecords = await _fullList('merchant_rules', filter);
report(0.36);
onStatus?.call('Pulling templates…');
final exactRecords = await _fullList('exact_templates', filter);
report(0.42);
onStatus?.call('Pulling recurring templates…');
List<RecordModel> recurringRecords = const [];
try {
recurringRecords = await _fullList('recurring_templates', filter);
} catch (error) {
debugPrint(
'PbSyncService pull: recurring_templates unavailable ($error)',
);
}
report(0.48);
final customLabels = <TxLabel>[];
final labelParents = <String, String>{};
for (final r in labelRecords) {
final clientId = r.getStringValue('client_id');
if (clientId.isEmpty) continue;
final parent = r.getStringValue('parent_client_id');
customLabels.add(
TxLabel(
id: clientId,
name: r.getStringValue('name').isEmpty
? 'Label'
: r.getStringValue('name'),
color: Color(r.getIntValue('color', 0xFF64748B)),
custom: true,
parentId: parent.isEmpty ? null : parent,
),
);
if (parent.isNotEmpty) labelParents[clientId] = parent;
}
final byId = <String, TxMeta>{};
for (final r in metaRecords) {
final txId = r.getStringValue('tx_id');
if (txId.isEmpty) continue;
byId[txId] = TxMeta(
labelIds: _jsonStringList(r.get<dynamic>('label_ids')),
recurring: r.getBoolValue('recurring'),
internalTransfer: r.getBoolValue('internal_transfer'),
notes: r.getStringValue('notes'),
);
}
final labelsByMerchant = <String, List<String>>{};
final notesByMerchant = <String, String>{};
for (final r in merchantRecords) {
final key = r.getStringValue('merchant_key');
if (key.isEmpty) continue;
final ids = _jsonStringList(r.get<dynamic>('label_ids'));
if (ids.isNotEmpty) labelsByMerchant[key] = ids;
final notes = r.getStringValue('notes');
if (notes.trim().isNotEmpty) notesByMerchant[key] = notes;
}
final exactTemplates = <String, TxExactTemplate>{};
for (final r in exactRecords) {
final key = r.getStringValue('exact_key');
if (key.isEmpty) continue;
exactTemplates[key] = TxExactTemplate(
notes: r.getStringValue('notes'),
recurring: r.getBoolValue('recurring'),
internalTransfer: r.getBoolValue('internal_transfer'),
);
}
final recurringTemplates = <String, RecurringTemplate>{};
for (final r in recurringRecords) {
final clientId = r.getStringValue('client_id');
if (clientId.isEmpty) continue;
final sourceTx = r.getStringValue('source_tx_id');
final labelId = r.getStringValue('label_id');
recurringTemplates[clientId] = RecurringTemplate(
id: clientId,
name: r.getStringValue('name').isEmpty
? 'Recurring'
: r.getStringValue('name'),
dayOfMonth: r.getIntValue('day_of_month', 1).clamp(1, 28),
currency: r.getStringValue('currency').isEmpty
? 'EUR'
: r.getStringValue('currency'),
source: _parseTemplateSource(r.getStringValue('source')),
amountRule: _parseAmountRule(r.getStringValue('amount_rule')),
merchantKey: r.getStringValue('merchant_key'),
sourceTxId: sourceTx.isEmpty ? null : sourceTx,
labelId: labelId.isEmpty ? null : labelId,
avgCount: r.getIntValue('avg_count', 5).clamp(1, 36),
autoFromRecurring: r.getBoolValue('auto_from_recurring'),
);
}
debugPrint(
'PbSyncService pull: labels=${labelRecords.length} '
'tx_meta=${metaRecords.length} merchants=${merchantRecords.length} '
'exact=${exactRecords.length} recurring=${recurringRecords.length}',
);
return TxMetaStoreData(
byId: byId,
labelsByMerchant: labelsByMerchant,
notesByMerchant: notesByMerchant,
exactTemplates: exactTemplates,
recurringTemplates: recurringTemplates,
customLabels: customLabels,
labelParents: labelParents,
);
}
/// Cheap remote count for stale detection (1 row, with totals).
Future<int> countTransactions() async {
final userId = _userId;
final filter = 'user="$userId"';
final result = await _withTimeout(
_pb.collection('transactions').getList(
page: 1,
perPage: 1,
filter: filter,
skipTotal: false,
),
label: 'transactions count',
);
return result.totalItems;
}
/// Pulls the full bank transaction list for the signed-in user.
Future<List<BankTransaction>> pullTransactions({
void Function(double progress)? onProgress,
void Function(String message)? onStatus,
}) async {
final userId = _userId;
final filter = 'user="$userId"';
onProgress?.call(0.05);
onStatus?.call('Pulling bank transactions…');
final records = await _fullList(
'transactions',
filter,
onPage: (n) => onStatus?.call('Pulling bank transactions ($n)…'),
);
onProgress?.call(0.95);
final out = <BankTransaction>[];
for (final r in records) {
final tx = _bankTxFromRecord(r);
if (tx != null) out.add(tx);
}
debugPrint('PbSyncService pull: transactions=${out.length}');
onProgress?.call(1.0);
return out;
}
/// Upserts [store] and removes remote rows that are no longer local.
///
/// Returns a short summary of tx_meta sync for the Account screen.
Future<String> push(
TxMetaStoreData store, {
void Function(double progress)? onProgress,
void Function(String message)? onStatus,
double progressStart = 0.5,
double progressEnd = 1.0,
}) async {
final userId = _userId;
final filter = 'user="$userId"';
final span = progressEnd - progressStart;
void report(double t) =>
onProgress?.call((progressStart + span * t).clamp(0.0, 1.0));
report(0.02);
onStatus?.call('Syncing labels…');
await _syncLabels(userId, filter, store);
report(0.08);
onStatus?.call('Syncing annotations…');
final txSummary = await _syncTxMeta(
userId,
filter,
store,
onProgress: (t) => report(0.08 + 0.82 * t),
onStatus: onStatus,
);
report(0.92);
onStatus?.call('Syncing merchant rules…');
await _syncMerchantRules(userId, filter, store);
report(0.96);
onStatus?.call('Syncing templates…');
await _syncExactTemplates(userId, filter, store);
report(0.98);
onStatus?.call('Syncing recurring templates…');
try {
await _syncRecurringTemplates(userId, filter, store);
} catch (error) {
debugPrint(
'PbSyncService push: recurring_templates unavailable ($error)',
);
}
report(1.0);
return txSummary;
}
/// Pull Overview favourites for the signed-in user.
///
/// Soft-fails at the call site if the collection is missing.
Future<List<OverviewFavourite>> pullOverviewFavourites({
void Function(String message)? onStatus,
}) async {
final userId = _userId;
final filter = 'user="$userId"';
onStatus?.call('Pulling overview favourites…');
final records = await _fullList('overview_favourites', filter);
final out = <OverviewFavourite>[];
for (final r in records) {
final clientId = r.getStringValue('client_id');
if (clientId.isEmpty) continue;
out.add(
OverviewFavourite(
id: clientId,
name: r.getStringValue('name').isEmpty
? 'Favourite'
: r.getStringValue('name'),
display: r.getBoolValue('display', true),
metric: OverviewMetricMode.values.firstWhere(
(e) => e.name == r.getStringValue('metric'),
orElse: () => OverviewMetricMode.flow,
),
period: OverviewPeriodMode.values.firstWhere(
(e) => e.name == r.getStringValue('period'),
orElse: () => OverviewPeriodMode.monthly,
),
labelIds: _jsonStringList(r.get<dynamic>('label_ids')),
flow: TxFlowFilter.values.firstWhere(
(e) => e.name == r.getStringValue('flow'),
orElse: () => TxFlowFilter.expense,
),
barStyle: LabelChartBarStyle.values.firstWhere(
(e) => e.name == r.getStringValue('bar_style'),
orElse: () => LabelChartBarStyle.parallel,
),
invertExpensePerspective:
r.getBoolValue('invert_expense_perspective'),
),
);
}
debugPrint('PbSyncService pull: overview_favourites=${out.length}');
return out;
}
/// Upsert Overview favourites; deletes remote rows no longer local.
Future<void> pushOverviewFavourites(
List<OverviewFavourite> favourites, {
void Function(String message)? onStatus,
}) async {
final userId = _userId;
final filter = 'user="$userId"';
onStatus?.call('Syncing overview favourites…');
await _syncOverviewFavourites(userId, filter, favourites);
}
/// Pull Home filter widgets for the signed-in user.
Future<List<HomeFilterWidget>> pullHomeFilterWidgets({
void Function(String message)? onStatus,
}) async {
final userId = _userId;
final filter = 'user="$userId"';
onStatus?.call('Pulling home widgets…');
final records = await _fullList('home_filter_widgets', filter);
final out = <HomeFilterWidget>[];
for (final r in records) {
final clientId = r.getStringValue('client_id');
if (clientId.isEmpty) continue;
final rangeRaw = r.get<dynamic>('amount_range');
out.add(
HomeFilterWidget.fromJson({
'id': clientId,
'name': r.getStringValue('name'),
'display': r.getBoolValue('display', true),
'sortOrder': r.getIntValue('sort_order', 0),
'halfWidth': r.getBoolValue('half_width'),
'type': r.getStringValue('widget_type'),
'iconId': r.getStringValue('icon_id'),
'chartStyle': r.getStringValue('chart_style'),
'budgetCapMode': r.getStringValue('budget_cap_mode'),
'budgetFixedAmount': switch (r.get<dynamic>('budget_fixed_amount')) {
num n => n.toDouble(),
String s => double.tryParse(s) ?? 0,
_ => 0.0,
},
'budgetAverageMonths': r.getIntValue('budget_average_months', 3),
'labelIds': _jsonStringList(r.get<dynamic>('label_ids')),
'labelMatch': r.getStringValue('label_match'),
'flow': r.getStringValue('flow'),
'amountRange': rangeRaw is Map
? Map<String, dynamic>.from(rangeRaw)
: const <String, dynamic>{},
'query': r.getStringValue('query'),
'monthOnly': r.getBoolValue('month_only'),
'showAmountIn': r.getBoolValue('show_amount_in', true),
'showAmountOut': r.getBoolValue('show_amount_out', true),
'showCount': r.getBoolValue('show_count', true),
}),
);
}
debugPrint('PbSyncService pull: home_filter_widgets=${out.length}');
return out;
}
Future<void> pushHomeFilterWidgets(
List<HomeFilterWidget> widgets, {
void Function(String message)? onStatus,
}) async {
final userId = _userId;
final filter = 'user="$userId"';
onStatus?.call('Syncing home widgets…');
await _syncHomeFilterWidgets(userId, filter, widgets);
}
/// Pull per-user app prefs (Home sections + Look ahead). Soft-fail if missing.
Future<AppCloudPrefs?> pullAppPrefs({
void Function(String message)? onStatus,
}) async {
final userId = _userId;
final filter = 'user="$userId"';
onStatus?.call('Pulling app prefs…');
final records = await _fullList('app_prefs', filter);
if (records.isEmpty) {
debugPrint('PbSyncService pull: app_prefs=0');
return null;
}
final r = records.first;
final prefs = AppCloudPrefs(
homeBuiltInSections: HomeBuiltInSections(
showTransactions: r.getBoolValue('home_show_transactions', true),
showCategorized: r.getBoolValue('home_show_categorized', true),
showRecent: r.getBoolValue('home_show_recent', true),
),
lookAheadMaxYears:
r.getIntValue('look_ahead_max_years', 5).clamp(1, 30),
);
debugPrint('PbSyncService pull: app_prefs=1');
return prefs;
}
/// Upsert the single app_prefs row for this user.
Future<void> pushAppPrefs(
AppCloudPrefs prefs, {
void Function(String message)? onStatus,
}) async {
final userId = _userId;
final filter = 'user="$userId"';
onStatus?.call('Syncing app prefs…');
await _syncAppPrefs(userId, filter, prefs);
}
/// Upserts local bank txs to PocketBase.
///
/// Also garbage-collects remote rows that are:
/// - admin test rows (`is_manual` / `·#…` stamp) no longer local, or
/// - listed in [deleteTombs] (local delete queue).
Future<String> pushTransactions(
List<BankTransaction> local, {
List<BankTransaction> deleteTombs = const [],
void Function(double progress)? onProgress,
void Function(String message)? onStatus,
double progressStart = 0.0,
double progressEnd = 1.0,
}) async {
final userId = _userId;
final filter = 'user="$userId"';
final span = progressEnd - progressStart;
void report(double t) =>
onProgress?.call((progressStart + span * t).clamp(0.0, 1.0));
report(0.02);
onStatus?.call('Loading remote bank transactions…');
final existing = {
for (final r in await _fullList(
'transactions',
filter,
onPage: (n) =>
onStatus?.call('Loading remote bank transactions ($n)…'),
))
r.getStringValue('tx_id'): r,
};
report(0.12);
final ops = <_BatchOp>[];
var skipped = 0;
final seen = <String>{};
final tombIds = {
for (final t in deleteTombs)
if (_clip(t.id, 2000).isNotEmpty) _clip(t.id, 2000),
};
for (final tx in local) {
final syncId = _clip(tx.id, 2000);
if (syncId.isEmpty || !seen.add(syncId)) continue;
// Never re-upsert a tombstoned row in the same pass.
if (tombIds.contains(syncId) ||
deleteTombs.any(tx.matchesForDelete)) {
continue;
}
final body = <String, dynamic>{
'user': userId,
'tx_id': syncId,
'date': _clip(tx.date, 32),
'description': _clip(
tx.description.isEmpty ? 'Unknown' : tx.description,
4000,
),
'amount': tx.amount,
'currency': _clip(
tx.currency.isEmpty ? 'EUR' : tx.currency.toUpperCase(),
8,
),
'pending': tx.pending,
// Only send when true so servers without the migration still accept
// normal bank/CSV upserts. Test rows are also detected via ·# stamp.
if (tx.isManual) 'is_manual': true,
};
final remote = existing[syncId];
if (remote == null) {
ops.add(_BatchOp.create(body));
} else if (!_bankTxUnchanged(remote, tx)) {
ops.add(_BatchOp.update(remote.id, body));
} else {
skipped++;
}
}
// Remove admin test / tombstoned rows that this device no longer has.
var deletes = 0;
for (final entry in existing.entries) {
if (seen.contains(entry.key)) continue;
final remoteTx = _bankTxFromRecord(entry.value);
final tombHit = tombIds.contains(entry.key) ||
(remoteTx != null && deleteTombs.any(remoteTx.matchesForDelete));
final orphanManual = _remoteLooksLikeManualTest(entry.value);
if (!tombHit && !orphanManual) continue;
ops.add(_BatchOp.delete(entry.value.id));
deletes++;
}
final creates = ops.where((o) => o.kind == _BatchOpKind.create).length;
final updates = ops.where((o) => o.kind == _BatchOpKind.update).length;
debugPrint(
'PbSyncService transactions: local=${local.length} '
'remote=${existing.length} create=$creates update=$updates '
'delete=$deletes skip=$skipped',
);
onStatus?.call(
ops.isEmpty
? 'Bank txs up to date (local ${local.length} · remote ${existing.length})'
: 'Writing bank txs (0/${ops.length})…',
);
await _runWriteOps(
'transactions',
ops,
onProgress: (t) => report(0.12 + 0.88 * t),
onStatus: onStatus,
);
return 'Bank txs: local ${local.length} → remote '
'${existing.length + creates - deletes} '
'(+$creates ~$updates -$deletes · $skipped unchanged)';
}
bool _remoteLooksLikeManualTest(RecordModel r) {
if (r.getBoolValue('is_manual')) return true;
return RegExp(r' ·#\d+$').hasMatch(r.getStringValue('description'));
}
BankTransaction? _bankTxFromRecord(RecordModel r) {
final date = r.getStringValue('date');
final description = r.getStringValue('description');
if (date.isEmpty && description.isEmpty) return null;
final rawAmount = r.get<dynamic>('amount');
final amount = switch (rawAmount) {
num n => n.toDouble(),
String s => double.tryParse(s) ?? 0,
_ => 0.0,
};
return BankTransaction(
date: date,
description: description.isEmpty ? 'Unknown' : description,
amount: amount,
currency: r.getStringValue('currency').isEmpty
? 'EUR'
: r.getStringValue('currency'),
pending: r.getBoolValue('pending'),
isManual: r.getBoolValue('is_manual') ||
RegExp(r' ·#\d+$').hasMatch(description),
);
}
bool _bankTxUnchanged(RecordModel remote, BankTransaction tx) {
final rawAmount = remote.get<dynamic>('amount');
final remoteAmount = switch (rawAmount) {
num n => n.toDouble(),
String s => double.tryParse(s) ?? 0,
_ => 0.0,
};
final remoteManual = remote.getBoolValue('is_manual') ||
RegExp(r' ·#\d+$').hasMatch(remote.getStringValue('description'));
return remote.getStringValue('date') == tx.date &&
remote.getStringValue('description') ==
_clip(tx.description.isEmpty ? 'Unknown' : tx.description, 4000) &&
(remoteAmount * 100).round() == (tx.amount * 100).round() &&
remote.getStringValue('currency').toUpperCase() ==
tx.currency.toUpperCase() &&
remote.getBoolValue('pending') == tx.pending &&
remoteManual == tx.isManual;
}
/// Deletes remote bank rows matching [tombstones] by tx_id or field fingerprint.
///
/// Returns the tombstones that matched at least one remote row (safe to clear).
Future<List<BankTransaction>> deleteTransactionsMatching(
List<BankTransaction> tombstones,
) async {
if (tombstones.isEmpty) return const [];
final userId = _userId;
final filter = 'user="$userId"';
final existing = await _fullList('transactions', filter);
final matchedTombs = <BankTransaction>[];
final ops = <_BatchOp>[];
final seenRecordIds = <String>{};
for (final r in existing) {
final remoteTx = _bankTxFromRecord(r);
if (remoteTx == null) continue;
for (final tomb in tombstones) {
final txId = r.getStringValue('tx_id');
final idHit = txId == _clip(tomb.id, 2000);
if (!idHit && !remoteTx.matchesForDelete(tomb)) continue;
if (!seenRecordIds.add(r.id)) break;
ops.add(_BatchOp.delete(r.id));
if (!matchedTombs.any(tomb.matchesForDelete)) {
matchedTombs.add(tomb);
}
break;
}
}
if (ops.isEmpty) {
debugPrint(
'PbSyncService deleteTransactionsMatching: 0 hits '
'for ${tombstones.length} tombstone(s)',
);
return const [];
}
await _runWriteOps('transactions', ops);
debugPrint(
'PbSyncService deleteTransactionsMatching: deleted ${ops.length} '
'for ${matchedTombs.length} tombstone(s)',
);
return matchedTombs;
}
Future<void> _runOne(String collection, _BatchOp op) {
final col = _pb.collection(collection);
final Future<dynamic> future;
switch (op.kind) {
case _BatchOpKind.create:
future = col.create(body: op.body!);
case _BatchOpKind.update:
future = col.update(op.id!, body: op.body!);
case _BatchOpKind.delete:
future = col.delete(op.id!);
}
return _withTimeout(future, label: '$collection ${op.kind.name}');
}
String _errorDetail(Object error) {
if (error is ClientException) {
final data = error.response['data'];
final message = error.response['message'];
if (data is Map && data.isNotEmpty) {
final parts = <String>[];
for (final e in data.entries) {
final v = e.value;
if (v is Map && v['message'] != null) {
parts.add('${e.key}: ${v['message']}');
} else {
parts.add('${e.key}: $v');
}
}
if (parts.isNotEmpty) return parts.join('; ');
}
if (message is String && message.isNotEmpty) {
return 'HTTP ${error.statusCode}: $message';
}
return 'HTTP ${error.statusCode}';
}
return error.toString();
}
/// Parallel chunked writes with live progress (avoids hung /api/batch).
Future<void> _runWriteOps(
String collection,
List<_BatchOp> ops, {
void Function(double progress)? onProgress,
void Function(String message)? onStatus,
}) async {
if (ops.isEmpty) {
onProgress?.call(1);
return;
}
final total = ops.length;
var done = 0;
final errors = <String>[];
for (var i = 0; i < ops.length; i += _writeConcurrency) {
final end = (i + _writeConcurrency > ops.length)
? ops.length
: i + _writeConcurrency;
final chunk = ops.sublist(i, end);
await Future.wait([
for (final op in chunk)
() async {
try {
await _runOne(collection, op);
} catch (error, stackTrace) {
final detail = _errorDetail(error);
// Unique conflict on create → try update if we can resolve the id.
if (op.kind == _BatchOpKind.create &&
error is ClientException &&
(error.statusCode == 400 || error.statusCode == 409)) {
try {
await _upsertAfterConflict(collection, op.body!);
return;
} catch (retryError, retryStack) {
errors.add(detail);
debugPrint(
'PbSyncService $collection create+upsert failed: '
'$detail / ${_errorDetail(retryError)}\n$retryStack',
);
return;
}
}
errors.add(detail);
debugPrint(
'PbSyncService $collection ${op.kind.name} failed: $detail\n'
'$stackTrace',
);
}
}(),
]);
done = end;
onProgress?.call(done / total);
// Status less often than progress — SyncController already throttles,
// but fewer string allocs/churn still help.
if (done == total || done == end && (done ~/ _writeConcurrency) % 4 == 0) {
onStatus?.call('Syncing $collection ($done/$total)…');
}
}
if (errors.isNotEmpty) {
throw StateError(
'Sync $collection: ${errors.length}/$total failed — ${errors.first}',
);
}
}
/// After a unique-key create conflict, load the existing row and PATCH it.
Future<void> _upsertAfterConflict(
String collection,
Map<String, dynamic> body,
) async {
final userId = body['user'] as String?;
String? filter;
switch (collection) {
case 'merchant_rules':
final key = body['merchant_key'];
if (userId != null && key is String) {
filter = 'user="$userId" && merchant_key="${key.replaceAll('"', '')}"';
}
case 'labels':
final key = body['client_id'];
if (userId != null && key is String) {
filter = 'user="$userId" && client_id="${key.replaceAll('"', '')}"';
}
case 'tx_meta':
final key = body['tx_id'];
if (userId != null && key is String) {
filter =
'user="$userId" && tx_id="${key.replaceAll('"', '\\"')}"';
}
case 'exact_templates':
final key = body['exact_key'];
if (userId != null && key is String) {
filter =
'user="$userId" && exact_key="${key.replaceAll('"', '\\"')}"';
}
case 'recurring_templates':
final key = body['client_id'];
if (userId != null && key is String) {
filter =
'user="$userId" && client_id="${key.replaceAll('"', '\\"')}"';
}
case 'overview_favourites':
final key = body['client_id'];
if (userId != null && key is String) {
filter =
'user="$userId" && client_id="${key.replaceAll('"', '\\"')}"';
}
case 'home_filter_widgets':
final key = body['client_id'];
if (userId != null && key is String) {
filter =
'user="$userId" && client_id="${key.replaceAll('"', '\\"')}"';
}
case 'app_prefs':
if (userId != null) {
filter = 'user="$userId"';
}
case 'transactions':
final key = body['tx_id'];
if (userId != null && key is String) {
filter =
'user="$userId" && tx_id="${key.replaceAll('"', '\\"')}"';
}
}
if (filter == null) {
throw StateError('Cannot upsert $collection without filter');
}
final existing = await _withTimeout(
_pb.collection(collection).getList(
page: 1,
perPage: 1,
filter: filter,
),
label: '$collection conflict lookup',
);
if (existing.items.isEmpty) {
throw StateError('Conflict on $collection but record not found');
}
await _withTimeout(
_pb.collection(collection).update(existing.items.first.id, body: body),
label: '$collection conflict update',
);
}
static String _clip(String value, int max) {
final t = value.trim();
if (t.length <= max) return t;
return t.substring(0, max);
}
Future<void> _syncLabels(
String userId,
String filter,
TxMetaStoreData store,
) async {
final existing = {
for (final r in await _fullList('labels', filter))
r.getStringValue('client_id'): r,
};
final keep = <String>{};
final ops = <_BatchOp>[];
for (final label in store.customLabels) {
keep.add(label.id);
final parent = store.labelParents[label.id] ?? label.parentId ?? '';
final body = <String, dynamic>{
'user': userId,
'client_id': _clip(label.id, 80),
'name': _clip(label.name.isEmpty ? 'Label' : label.name, 80),
'color': label.color.toARGB32(),
'parent_client_id': _clip(parent, 80),
};
final remote = existing[label.id];
if (remote == null) {
ops.add(_BatchOp.create(body));
} else if (!_labelUnchanged(remote, body)) {
ops.add(_BatchOp.update(remote.id, body));
}
}
for (final e in existing.entries) {
if (!keep.contains(e.key)) {
ops.add(_BatchOp.delete(e.value.id));
}
}
debugPrint('PbSyncService labels: remote=${existing.length} ops=${ops.length}');
await _runWriteOps('labels', ops);
}
Future<String> _syncTxMeta(
String userId,
String filter,
TxMetaStoreData store, {
void Function(double progress)? onProgress,
void Function(String message)? onStatus,
}) async {
onProgress?.call(0.02);
onStatus?.call('Loading remote tx_meta…');
final existing = {
for (final r in await _fullList(
'tx_meta',
filter,
onPage: (n) => onStatus?.call('Loading remote tx_meta ($n)…'),
))
r.getStringValue('tx_id'): r,
};
onProgress?.call(0.12);
final keep = <String>{};
final ops = <_BatchOp>[];
var skipped = 0;
var localSyncable = 0;
for (final e in store.byId.entries) {
final meta = e.value;
if (!meta.hasLabel &&
!meta.hasNotes &&
!meta.recurring &&
!meta.internalTransfer) {
continue;
}
localSyncable++;
final syncId = _clip(e.key, 2000);
keep.add(syncId);
final body = <String, dynamic>{
'user': userId,
'tx_id': syncId,
'label_ids': meta.labelIds,
'recurring': meta.recurring,
'internal_transfer': meta.internalTransfer,
'notes': _clip(meta.notes, 4000),
};
final remote = existing[syncId] ?? existing[e.key];
if (remote == null) {
ops.add(_BatchOp.create(body));
} else if (!_txMetaUnchanged(remote, meta)) {
ops.add(_BatchOp.update(remote.id, body));
} else {
skipped++;
}
}
for (final e in existing.entries) {
if (!keep.contains(e.key)) {
ops.add(_BatchOp.delete(e.value.id));
}
}
final creates = ops.where((o) => o.kind == _BatchOpKind.create).length;
final updates = ops.where((o) => o.kind == _BatchOpKind.update).length;
final deletes = ops.where((o) => o.kind == _BatchOpKind.delete).length;
debugPrint(
'PbSyncService tx_meta: local=$localSyncable remote=${existing.length} '
'create=$creates update=$updates delete=$deletes skip=$skipped',
);
onStatus?.call(
ops.isEmpty
? 'tx_meta up to date (local $localSyncable · remote ${existing.length})'
: 'Writing tx_meta (0/${ops.length})…',
);
await _runWriteOps(
'tx_meta',
ops,
onProgress: (t) => onProgress?.call(0.12 + 0.88 * t),
onStatus: onStatus,
);
return 'Annotations: local $localSyncable → remote '
'${existing.length + creates - deletes} '
'(+$creates ~$updates -$deletes · $skipped unchanged)';
}
Future<void> _syncMerchantRules(
String userId,
String filter,
TxMetaStoreData store,
) async {
final existing = {
for (final r in await _fullList('merchant_rules', filter))
r.getStringValue('merchant_key'): r,
};
final keep = <String>{};
final ops = <_BatchOp>[];
final keys = {
...store.labelsByMerchant.keys,
...store.notesByMerchant.keys,
};
for (final rawKey in keys) {
final key = _clip(rawKey, 200);
if (key.isEmpty) continue;
final ids = store.labelsByMerchant[rawKey] ?? const <String>[];
final notes = _clip(store.notesByMerchant[rawKey] ?? '', 4000);
if (ids.isEmpty && notes.isEmpty) continue;
keep.add(key);
final body = <String, dynamic>{
'user': userId,
'merchant_key': key,
'label_ids': ids,
'notes': notes,
};
final remote = existing[key] ?? existing[rawKey];
if (remote == null) {
ops.add(_BatchOp.create(body));
} else if (!_merchantUnchanged(remote, ids, notes)) {
ops.add(_BatchOp.update(remote.id, body));
}
}
for (final e in existing.entries) {
if (!keep.contains(e.key)) {
ops.add(_BatchOp.delete(e.value.id));
}
}
debugPrint(
'PbSyncService merchant_rules: remote=${existing.length} ops=${ops.length}',
);
await _runWriteOps('merchant_rules', ops);
}
Future<void> _syncExactTemplates(
String userId,
String filter,
TxMetaStoreData store,
) async {
final existing = {
for (final r in await _fullList('exact_templates', filter))
r.getStringValue('exact_key'): r,
};
final keep = <String>{};
final ops = <_BatchOp>[];
for (final e in store.exactTemplates.entries) {
final t = e.value;
if (t.notes.trim().isEmpty && !t.recurring && !t.internalTransfer) {
continue;
}
final key = _clip(e.key, 2000);
if (key.isEmpty) continue;
keep.add(key);
final body = <String, dynamic>{
'user': userId,
'exact_key': key,
'notes': _clip(t.notes, 4000),
'recurring': t.recurring,
'internal_transfer': t.internalTransfer,
};
final remote = existing[key] ?? existing[e.key];
if (remote == null) {
ops.add(_BatchOp.create(body));
} else if (!_exactUnchanged(remote, t)) {
ops.add(_BatchOp.update(remote.id, body));
}
}
for (final e in existing.entries) {
if (!keep.contains(e.key)) {
ops.add(_BatchOp.delete(e.value.id));
}
}
debugPrint(
'PbSyncService exact_templates: remote=${existing.length} ops=${ops.length}',
);
await _runWriteOps('exact_templates', ops);
}
Future<void> _syncRecurringTemplates(
String userId,
String filter,
TxMetaStoreData store,
) async {
final existing = {
for (final r in await _fullList('recurring_templates', filter))
r.getStringValue('client_id'): r,
};
final keep = <String>{};
final ops = <_BatchOp>[];
for (final e in store.recurringTemplates.entries) {
final t = e.value;
final clientId = _clip(t.id.isNotEmpty ? t.id : e.key, 120);
if (clientId.isEmpty) continue;
keep.add(clientId);
final body = <String, dynamic>{
'user': userId,
'client_id': clientId,
'name': _clip(t.name.isEmpty ? 'Recurring' : t.name, 400),
'day_of_month': t.dayOfMonth.clamp(1, 28),
'currency': _clip(
t.currency.isEmpty ? 'EUR' : t.currency.toUpperCase(),
8,
),
'source': t.source.name,
'amount_rule': t.amountRule.name,
'merchant_key': _clip(t.merchantKey, 500),
'source_tx_id': _clip(t.sourceTxId ?? '', 2000),
'label_id': _clip(t.labelId ?? '', 80),
'avg_count': t.avgCount.clamp(1, 36),
'auto_from_recurring': t.autoFromRecurring,
};
final remote = existing[clientId];
if (remote == null) {
ops.add(_BatchOp.create(body));
} else if (!_recurringUnchanged(remote, t)) {
ops.add(_BatchOp.update(remote.id, body));
}
}
for (final e in existing.entries) {
if (!keep.contains(e.key)) {
ops.add(_BatchOp.delete(e.value.id));
}
}
debugPrint(
'PbSyncService recurring_templates: remote=${existing.length} '
'ops=${ops.length}',
);
await _runWriteOps('recurring_templates', ops);
}
Future<void> _syncOverviewFavourites(
String userId,
String filter,
List<OverviewFavourite> favourites,
) async {
final existing = {
for (final r in await _fullList('overview_favourites', filter))
r.getStringValue('client_id'): r,
};
final keep = <String>{};
final ops = <_BatchOp>[];
for (final fav in favourites) {
final clientId = _clip(fav.id, 120);
if (clientId.isEmpty) continue;
keep.add(clientId);
final body = <String, dynamic>{
'user': userId,
'client_id': clientId,
'name': _clip(fav.name.isEmpty ? 'Favourite' : fav.name, 200),
'display': fav.display,
'metric': fav.metric.name,
'period': fav.period.name,
'label_ids': fav.labelIds,
'flow': fav.flow.name,
'bar_style': fav.barStyle.name,
'invert_expense_perspective': fav.invertExpensePerspective,
};
final remote = existing[clientId];
if (remote == null) {
ops.add(_BatchOp.create(body));
} else if (!_overviewFavouriteUnchanged(remote, fav)) {
ops.add(_BatchOp.update(remote.id, body));
}
}
for (final e in existing.entries) {
if (!keep.contains(e.key)) {
ops.add(_BatchOp.delete(e.value.id));
}
}
debugPrint(
'PbSyncService overview_favourites: remote=${existing.length} '
'ops=${ops.length}',
);
await _runWriteOps('overview_favourites', ops);
}
Future<void> _syncHomeFilterWidgets(
String userId,
String filter,
List<HomeFilterWidget> widgets,
) async {
final existing = {
for (final r in await _fullList('home_filter_widgets', filter))
r.getStringValue('client_id'): r,
};
final keep = <String>{};
final ops = <_BatchOp>[];
for (final w in widgets) {
final clientId = _clip(w.id, 120);
if (clientId.isEmpty) continue;
keep.add(clientId);
final body = <String, dynamic>{
'user': userId,
'client_id': clientId,
'name': _clip(w.name.isEmpty ? 'Filter' : w.name, 200),
'display': w.display,
'sort_order': w.sortOrder,
'half_width': w.halfWidth,
'widget_type': w.type.name,
'icon_id': w.iconId,
'chart_style': w.chartStyle.name,
'budget_cap_mode': w.budgetCapMode.name,
'budget_fixed_amount': w.budgetFixedAmount,
'budget_average_months': w.budgetAverageMonths,
'label_ids': w.labelIds,
'label_match': w.labelMatch.name,
'flow': w.flow.name,
'amount_range': {
if (w.amountRange.exact != null) 'exact': w.amountRange.exact,
if (w.amountRange.min != null) 'min': w.amountRange.min,
if (w.amountRange.max != null) 'max': w.amountRange.max,
},
'query': _clip(w.query, 400),
'month_only': w.monthOnly,
'show_amount_in': w.showAmountIn,
'show_amount_out': w.showAmountOut,
'show_count': w.showCount,
};
final remote = existing[clientId];
if (remote == null) {
ops.add(_BatchOp.create(body));
} else if (!_homeFilterWidgetUnchanged(remote, w)) {
ops.add(_BatchOp.update(remote.id, body));
}
}
for (final e in existing.entries) {
if (!keep.contains(e.key)) {
ops.add(_BatchOp.delete(e.value.id));
}
}
debugPrint(
'PbSyncService home_filter_widgets: remote=${existing.length} '
'ops=${ops.length}',
);
await _runWriteOps('home_filter_widgets', ops);
}
Future<void> _syncAppPrefs(
String userId,
String filter,
AppCloudPrefs prefs,
) async {
final existing = await _fullList('app_prefs', filter);
final body = <String, dynamic>{
'user': userId,
'home_show_transactions': prefs.homeBuiltInSections.showTransactions,
'home_show_categorized': prefs.homeBuiltInSections.showCategorized,
'home_show_recent': prefs.homeBuiltInSections.showRecent,
'look_ahead_max_years': prefs.lookAheadMaxYears.clamp(1, 30),
};
final ops = <_BatchOp>[];
if (existing.isEmpty) {
ops.add(_BatchOp.create(body));
} else {
final remote = existing.first;
if (!_appPrefsUnchanged(remote, prefs)) {
ops.add(_BatchOp.update(remote.id, body));
}
// Enforce singleton: drop extras if any.
for (var i = 1; i < existing.length; i++) {
ops.add(_BatchOp.delete(existing[i].id));
}
}
debugPrint('PbSyncService app_prefs: remote=${existing.length} ops=${ops.length}');
await _runWriteOps('app_prefs', ops);
}
bool _labelUnchanged(RecordModel remote, Map<String, dynamic> body) {
return remote.getStringValue('name') == body['name'] &&
remote.getIntValue('color') == body['color'] &&
remote.getStringValue('parent_client_id') == body['parent_client_id'];
}
bool _txMetaUnchanged(RecordModel remote, TxMeta meta) {
final remoteIds = _jsonStringList(remote.get<dynamic>('label_ids'));
return _sameIdList(remoteIds, meta.labelIds) &&
remote.getBoolValue('recurring') == meta.recurring &&
remote.getBoolValue('internal_transfer') == meta.internalTransfer &&
remote.getStringValue('notes') == meta.notes;
}
bool _merchantUnchanged(
RecordModel remote,
List<String> ids,
String notes,
) {
final remoteIds = _jsonStringList(remote.get<dynamic>('label_ids'));
return _sameIdList(remoteIds, ids) &&
remote.getStringValue('notes') == notes;
}
bool _exactUnchanged(RecordModel remote, TxExactTemplate t) {
return remote.getStringValue('notes') == t.notes &&
remote.getBoolValue('recurring') == t.recurring &&
remote.getBoolValue('internal_transfer') == t.internalTransfer;
}
bool _recurringUnchanged(RecordModel remote, RecurringTemplate t) {
return remote.getStringValue('name') == t.name &&
remote.getIntValue('day_of_month') == t.dayOfMonth &&
remote.getStringValue('currency') == t.currency &&
remote.getStringValue('source') == t.source.name &&
remote.getStringValue('amount_rule') == t.amountRule.name &&
remote.getStringValue('merchant_key') == t.merchantKey &&
remote.getStringValue('source_tx_id') == (t.sourceTxId ?? '') &&
remote.getStringValue('label_id') == (t.labelId ?? '') &&
remote.getIntValue('avg_count', 5) == t.avgCount &&
remote.getBoolValue('auto_from_recurring') == t.autoFromRecurring;
}
bool _overviewFavouriteUnchanged(RecordModel remote, OverviewFavourite fav) {
final remoteIds = _jsonStringList(remote.get<dynamic>('label_ids'));
return remote.getStringValue('name') == fav.name &&
remote.getBoolValue('display', true) == fav.display &&
remote.getStringValue('metric') == fav.metric.name &&
remote.getStringValue('period') == fav.period.name &&
_sameIdList(remoteIds, fav.labelIds) &&
remote.getStringValue('flow') == fav.flow.name &&
remote.getStringValue('bar_style') == fav.barStyle.name &&
remote.getBoolValue('invert_expense_perspective') ==
fav.invertExpensePerspective;
}
bool _homeFilterWidgetUnchanged(RecordModel remote, HomeFilterWidget w) {
final remoteIds = _jsonStringList(remote.get<dynamic>('label_ids'));
final rangeRaw = remote.get<dynamic>('amount_range');
final rangeMap = rangeRaw is Map
? Map<String, dynamic>.from(rangeRaw)
: const <String, dynamic>{};
double? asDouble(dynamic value) => value is num ? value.toDouble() : null;
final remoteFixed = switch (remote.get<dynamic>('budget_fixed_amount')) {
num n => n.toDouble(),
String s => double.tryParse(s) ?? 0,
_ => 0.0,
};
return remote.getStringValue('name') == w.name &&
remote.getBoolValue('display', true) == w.display &&
remote.getIntValue('sort_order', 0) == w.sortOrder &&
remote.getBoolValue('half_width') == w.halfWidth &&
remote.getStringValue('widget_type') == w.type.name &&
remote.getStringValue('icon_id') == w.iconId &&
remote.getStringValue('chart_style') == w.chartStyle.name &&
remote.getStringValue('budget_cap_mode') == w.budgetCapMode.name &&
(remoteFixed * 100).round() == (w.budgetFixedAmount * 100).round() &&
remote.getIntValue('budget_average_months', 3) ==
w.budgetAverageMonths &&
_sameIdList(remoteIds, w.labelIds) &&
(remote.getStringValue('label_match').isEmpty
? 'and'
: remote.getStringValue('label_match')) ==
w.labelMatch.name &&
remote.getStringValue('flow') == w.flow.name &&
asDouble(rangeMap['exact']) == w.amountRange.exact &&
asDouble(rangeMap['min']) == w.amountRange.min &&
asDouble(rangeMap['max']) == w.amountRange.max &&
remote.getStringValue('query') == w.query &&
remote.getBoolValue('month_only') == w.monthOnly &&
remote.getBoolValue('show_amount_in', true) == w.showAmountIn &&
remote.getBoolValue('show_amount_out', true) == w.showAmountOut &&
remote.getBoolValue('show_count', true) == w.showCount;
}
bool _appPrefsUnchanged(RecordModel remote, AppCloudPrefs prefs) {
final s = prefs.homeBuiltInSections;
return remote.getBoolValue('home_show_transactions', true) ==
s.showTransactions &&
remote.getBoolValue('home_show_categorized', true) ==
s.showCategorized &&
remote.getBoolValue('home_show_recent', true) == s.showRecent &&
remote.getIntValue('look_ahead_max_years', 5) ==
prefs.lookAheadMaxYears.clamp(1, 30);
}
}
RecurringTemplateSource _parseTemplateSource(String raw) {
return switch (raw) {
'label' => RecurringTemplateSource.label,
_ => RecurringTemplateSource.transaction,
};
}
RecurringAmountRule _parseAmountRule(String raw) {
return switch (raw) {
'avgLastTxs' => RecurringAmountRule.avgLastTxs,
'labelAvgMonths' => RecurringAmountRule.labelAvgMonths,
_ => RecurringAmountRule.lastValue,
};
}
List<String> _jsonStringList(dynamic raw) {
if (raw is! List) return const [];
final out = <String>[];
for (final item in raw) {
if (item == null) continue;
final id = item is String ? item : '$item';
if (id.isEmpty || id == 'uncategorized' || id == 'other') continue;
out.add(id);
}
return out;
}