Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 66 additions & 0 deletions app/lib/data/local_desk_repo.dart
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ import 'secure_key_store.dart';
import 'sift.dart';
import 'pull_walk.dart';
import 'sync_jobs.dart';
import 'sync_plan.dart';
import 'sync_socket.dart';
import 'supabase_auth.dart';

Expand Down Expand Up @@ -606,6 +607,34 @@ class LocalDeskRepo extends ChangeNotifier implements RelicRepo, BillingRepo {
/// WorkerRepo.emailUnverified.
final ValueNotifier<bool> emailUnverified = ValueNotifier(false);

/// The free plan's sync pause (403 sync_paused, data/sync_plan.dart): set
/// while the account syncs a different device. Capture and the local vault
/// keep working; queued changes wait and go up once this PC syncs again. The
/// shell watches [syncPaused] for its banner. Lockstep with WorkerRepo.
final SyncPauseGate _pause = SyncPauseGate();
ValueNotifier<SyncPause?> get syncPaused => _pause.paused;

/// Make this PC the one that keeps syncing (POST /account/sync-device). On
/// success the pause lifts and a sync runs straight away. Lockstep with
/// WorkerRepo.syncThisDevice.
Future<SyncPickResult> syncThisDevice() async {
final id = _deviceId;
if (_syncUrl == null || id == null || id.isEmpty) {
return const SyncPickResult(SyncPickOutcome.failed);
}
final r = await postSyncDevice(
url: Uri.parse(_u('/account/sync-device')),
headers: _h,
deviceId: id,
);
if (r.outcome == SyncPickOutcome.ok) {
_pause.clear();
notifyListeners();
unawaited(syncNow());
}
return r;
}

/// True when [body] is the worker's `{"error":"email_unverified",...}` payload.
/// Static + pure so the flag logic is unit-testable. Lockstep with WorkerRepo.
/// Set true when the account's session has been revoked: the worker answered
Expand Down Expand Up @@ -2692,6 +2721,9 @@ class LocalDeskRepo extends ChangeNotifier implements RelicRepo, BillingRepo {
SyncState get sync {
if (_mk == null) return const SyncState(SyncKind.synced); // local-only
final pending = _db?.pendingCount() ?? 0;
// Paused by the free plan: nothing is broken, and nothing will move until
// the person picks this PC or upgrades, so this outranks "pending".
if (_pause.holding) return SyncState(SyncKind.paused, pending: pending);
if (pending > 0) return SyncState(SyncKind.pending, pending: pending);
final rejected = _db?.rejectionCount() ?? 0;
if (rejected > 0) return SyncState(SyncKind.quotaFull, pending: rejected);
Expand Down Expand Up @@ -5375,6 +5407,7 @@ class LocalDeskRepo extends ChangeNotifier implements RelicRepo, BillingRepo {
_uploadProgress.clear();
_mk = null;
_remoteAccount = null;
_pause.clear();
_online = false;
_clearFailureStreak();
_supabaseMode = false;
Expand Down Expand Up @@ -5418,6 +5451,7 @@ class LocalDeskRepo extends ChangeNotifier implements RelicRepo, BillingRepo {
_startTimer(); // re-arm the poll at the new cadence
if (up) unawaited(_pullRemote()); // catch up on anything missed
},
holding: () => _pause.holding,
);
_syncSocket!.start();
}
Expand Down Expand Up @@ -5622,6 +5656,9 @@ class LocalDeskRepo extends ChangeNotifier implements RelicRepo, BillingRepo {
_flushAgain = true;
return;
}
// Paused by the free plan: the queue waits. The pull that follows in the
// same cycle asks the server again and lifts the pause when it may sync.
if (_pause.holding) return;
_flushing = true;
var changed = false;
var skipped = 0;
Expand All @@ -5636,6 +5673,7 @@ class LocalDeskRepo extends ChangeNotifier implements RelicRepo, BillingRepo {
stalled = true;
break;
}
if (_pause.holding) break; // this op was refused by the pause
if (result.kind == _OutboundKind.skip) {
// The server answered but wouldn't take THIS op right now — leave it
// queued and keep draining, or one bad item head-of-line blocks the
Expand Down Expand Up @@ -5756,6 +5794,12 @@ class LocalDeskRepo extends ChangeNotifier implements RelicRepo, BillingRepo {
emailUnverified.value = false;
return _sent;
}
if (_pause.absorb(resp.statusCode, resp.body)) {
// Free plan sync pause. Leave this op queued; the drain stops.
_online = true;
notifyListeners();
return _skip;
}
if (resp.statusCode == 403 && isEmailUnverifiedBody(resp.body)) {
// Email not confirmed yet (VERIFY_GATE). Reachable, so keep draining;
// leave this op queued to flush once they confirm. The banner explains.
Expand Down Expand Up @@ -5796,6 +5840,11 @@ class LocalDeskRepo extends ChangeNotifier implements RelicRepo, BillingRepo {
_online = true;
return _sent;
}
if (_pause.absorb(resp.statusCode, resp.body)) {
_online = true;
notifyListeners();
return _skip; // stays queued until this PC syncs again
}
if (_permanentSyncStatus(resp.statusCode)) {
return _rejected(resp.statusCode);
}
Expand All @@ -5820,6 +5869,9 @@ class LocalDeskRepo extends ChangeNotifier implements RelicRepo, BillingRepo {
Future<void> _pullRemote() async {
final db = _db;
if (_pulling || _mk == null || db == null) return;
// Paused by the free plan and asked a moment ago: the answer will not have
// changed. A tap on "Sync now" asks anyway.
if (_pause.skipPull(force: _manualSync)) return;
// First-ever pull of this bind: syncBusy is watching, so the chip has to
// repaint on both edges. Later cycles skip the churn.
final firstSync = _lastSyncAt == 0;
Expand Down Expand Up @@ -5868,9 +5920,22 @@ class LocalDeskRepo extends ChangeNotifier implements RelicRepo, BillingRepo {
_refreshToken = null;
sessionRevoked.value = true;
}
// The free plan syncs another device. Reachable, not offline.
if (_pause.absorb(resp.statusCode, resp.body)) {
_online = true;
notifyListeners();
return;
}
_online = false;
return;
}
if (_pause.holding) {
// Picked elsewhere, or the account paid: this PC syncs again, and
// the changes queued while it was paused can go up now.
_pause.clear();
notifyListeners();
_kickSync();
}
final body = jsonDecode(resp.body) as Map<String, dynamic>;
final items = (body['items'] as List).cast<Map<String, dynamic>>();
for (final env in items) {
Expand Down Expand Up @@ -6942,6 +7007,7 @@ class LocalDeskRepo extends ChangeNotifier implements RelicRepo, BillingRepo {
historyCap: (j['history_cap'] as num?)?.toInt(),
evictedCount: (j['evicted_count'] as num?)?.toInt() ?? 0,
devicesCap: (j['devices_cap'] as num?)?.toInt(),
syncUntil: (j['sync_until'] as num?)?.toInt(),
);
}
} catch (_) {}
Expand Down
6 changes: 6 additions & 0 deletions app/lib/data/repo.dart
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,11 @@ class AccountInfo {

/// How many devices the plan allows. Null on Max and on a self-hosted server.
final int? devicesCap;

/// Unix seconds the free week of sync on every device ends (the free plan
/// sync rule, data/sync_plan.dart). Null on a paid plan, a self-hosted
/// server, and any server with the rule switched off.
final int? syncUntil;
const AccountInfo({
required this.tier,
required this.usedBytes,
Expand All @@ -46,6 +51,7 @@ class AccountInfo {
this.historyCap,
this.evictedCount = 0,
this.devicesCap,
this.syncUntil,
});

/// True only when a ring is actually in force. A self-hosted server enrolls
Expand Down
225 changes: 225 additions & 0 deletions app/lib/data/sync_plan.dart
Original file line number Diff line number Diff line change
@@ -0,0 +1,225 @@
import 'dart:convert';

import 'package:flutter/foundation.dart';
import 'package:http/http.dart' as http;

/// The free plan's sync rule, as the apps see it (server: worker/src/syncplan.ts).
///
/// Free is one device. A new account syncs on every device for its first
/// week; after that one device keeps syncing and the others get 403
/// `sync_paused`. A paused device keeps everything it already has and keeps
/// capturing locally. Its outgoing changes stay queued, so they go up the
/// moment this device syncs again (it is picked, or the account pays).
///
/// Shared by LocalDeskRepo and WorkerRepo so the two hosts read the 403 and
/// make the pick the same way.
class SyncPause {
/// Unix seconds the free week ended. Null from a server that did not say.
final int? syncUntil;

/// The device that is still syncing. Null if the server has not settled one.
final String? syncDevice;

/// The account's devices, newest first, as the server listed them.
final List<SyncPlanDevice> devices;

const SyncPause({this.syncUntil, this.syncDevice, this.devices = const []});

/// The pause carried by a 403 body, or null when [body] is any other error.
static SyncPause? fromBody(String body) {
try {
final j = jsonDecode(body);
if (j is! Map || j['error'] != 'sync_paused') return null;
return SyncPause(
syncUntil: (j['sync_until'] as num?)?.toInt(),
syncDevice: j['sync_device'] as String?,
devices: [
for (final d in (j['devices'] as List? ?? const []))
if (d is Map) SyncPlanDevice.fromJson(d.cast<String, dynamic>()),
],
);
} catch (_) {
return null;
}
}

/// The device that keeps syncing, as a person would name it.
String? get syncDeviceName {
for (final d in devices) {
if (d.id == syncDevice) return d.name;
}
return null;
}

/// The banner line. Names the syncing device when the server listed it.
String get message {
final other = syncDeviceName;
return other == null
? 'Sync is paused here. The free plan syncs one device. '
'Everything on this device is still here.'
: 'Sync is paused here. The free plan syncs one device, and '
'yours is $other. Everything on this device is still here.';
}
}

class SyncPlanDevice {
final String id;
final String? label;
final String? platform;
final bool thisDevice;

const SyncPlanDevice({
required this.id,
this.label,
this.platform,
this.thisDevice = false,
});

factory SyncPlanDevice.fromJson(Map<String, dynamic> j) => SyncPlanDevice(
id: j['device_id'] as String? ?? '',
label: j['label'] as String?,
platform: j['platform'] as String?,
thisDevice: j['this_device'] == true,
);

/// "Jordan's PC", or "your Windows PC" when the device has no label.
String get name {
final l = label?.trim();
if (l != null && l.isNotEmpty) return l;
return switch (platform) {
'windows' => 'your Windows PC',
'macos' => 'your Mac',
'linux' => 'your Linux PC',
'android' => 'your Android phone',
'ios' => 'your iPhone',
_ => 'another device',
};
}
}

/// How often a paused device asks again whether it may sync. The answer only
/// changes when the person picks a device or pays, so a poll every few seconds
/// would be thousands of refused requests a day for nothing.
const kPausedRecheck = Duration(minutes: 5);

/// One host's view of the pause: the notifier the UI watches, plus the clock
/// that spaces out the re-checks. Each repo owns one.
class SyncPauseGate {
/// The current pause, or null while this device syncs.
final ValueNotifier<SyncPause?> paused = ValueNotifier(null);
DateTime? _checkedAt;

/// [kPausedRecheck], shortened by tests that need the re-check to happen.
@visibleForTesting
static Duration recheck = kPausedRecheck;

/// True while this device may not sync.
bool get holding => paused.value != null;

/// True when a pull should not go out yet: paused, and asked recently.
/// [force] is a person tapping "Sync now".
bool skipPull({bool force = false, DateTime? now}) {
final at = _checkedAt;
if (force || paused.value == null || at == null) return false;
return (now ?? DateTime.now()).difference(at) < recheck;
}

/// Reads a refused response. True when it was the pause, which is then
/// published; false for any other error, which the caller handles as before.
bool absorb(int status, String body, {DateTime? now}) {
if (status != 403) return false;
final p = SyncPause.fromBody(body);
if (p == null) return false;
paused.value = p;
_checkedAt = now ?? DateTime.now();
return true;
}

/// The server took a sync request from this device: not paused.
void clear() {
_checkedAt = null;
if (paused.value != null) paused.value = null;
}
}

/// What came of asking to sync this device instead.
enum SyncPickOutcome { ok, cooldown, failed }

class SyncPickResult {
final SyncPickOutcome outcome;

/// [SyncPickOutcome.cooldown] only: unix seconds a switch is allowed again.
final int? retryAt;
const SyncPickResult(this.outcome, {this.retryAt});

/// The line to show the person after the tap.
String message({DateTime? now}) {
switch (outcome) {
case SyncPickOutcome.ok:
return 'This device syncs now.';
case SyncPickOutcome.failed:
return 'Could not switch right now. Try again in a minute.';
case SyncPickOutcome.cooldown:
final at = retryAt;
if (at == null) return 'You can switch devices once a week.';
final days = daysUntil(at, now: now);
return days <= 1
? 'You can switch devices once a week. Try again tomorrow.'
: 'You can switch devices once a week. Try again in $days days.';
}
}
}

/// POST /account/sync-device: make [deviceId] the device that keeps syncing.
Future<SyncPickResult> postSyncDevice({
required Uri url,
required Map<String, String> headers,
required String deviceId,
Duration timeout = const Duration(seconds: 15),
}) async {
try {
final r = await http
.post(
url,
headers: {...headers, 'Content-Type': 'application/json'},
body: jsonEncode({'device_id': deviceId}),
)
.timeout(timeout);
if (r.statusCode == 200) return const SyncPickResult(SyncPickOutcome.ok);
if (r.statusCode == 429) {
int? retryAt;
try {
final j = jsonDecode(r.body);
if (j is Map && j['error'] == 'sync_switch_cooldown') {
retryAt = (j['retry_at'] as num?)?.toInt();
}
} catch (_) {}
// A 429 with no cooldown body is the route's rate limiter: plain failure.
if (retryAt != null) {
return SyncPickResult(SyncPickOutcome.cooldown, retryAt: retryAt);
}
}
return const SyncPickResult(SyncPickOutcome.failed);
} catch (_) {
return const SyncPickResult(SyncPickOutcome.failed);
}
}

/// Whole days from [now] until unix second [at], rounded up. 0 once it passed.
int daysUntil(int at, {DateTime? now}) {
final n = (now ?? DateTime.now()).millisecondsSinceEpoch ~/ 1000;
if (at <= n) return 0;
return ((at - n) + 86399) ~/ 86400;
}

/// The countdown line for the free week, or null outside it. [syncUntil] is
/// GET /account's `sync_until`, which the server only sends while the rule is
/// switched on, so this stays silent until launch day.
String? syncWeekLine(int? syncUntil, {DateTime? now}) {
if (syncUntil == null) return null;
final days = daysUntil(syncUntil, now: now);
if (days == 0) return null;
return days == 1
? 'Free sync on all your devices ends in 1 day.'
: 'Free sync on all your devices ends in $days days.';
}
Loading
Loading