ihompadmin/tasq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Offline Support

Browse Source
This commit is contained in:
Marc Rejohn Castillano
2026-04-27 06:20:39 +08:00
parent 7d8851a94a
commit 48cd3bcd60
121 changed files with 14798 additions and 2214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/providers/stream_recovery.dart
+223 -2
View File
@@ -18,6 +18,12 @@ enum StreamConnectionStatus {
/// Fatal error; stream will not recover without manual intervention.
failed,
/// Device is offline. The realtime subscription is paused; Brick's local
/// SQLite cache is the active data source. No reconnection is attempted
/// until [notifyOnlineRestored] is called. The UI should show the offline
/// banner but NOT a skeleton/loading indicator.
offline,
}
/// Represents the result of a polling attempt.
@@ -70,7 +76,19 @@ typedef ChannelStatusCallback =
/// and connection status tracking. Provides graceful degradation when the
/// realtime connection fails.
///
/// ## Offline behaviour
///
/// Recovery is **suppressed** when the device is offline (as reported by
/// [setIsOnlineCallback]). In that state the wrapper marks the channel as
/// `recovering` but does **not** schedule timers or fire polls — Brick's
/// offline queue handles local data until connectivity returns.
///
/// When the device comes back online, call [notifyAllOnlineRestored] (done
/// automatically by `connectivity_provider._onReconnected`) to restart every
/// live wrapper whose stream is not already connected.
///
/// Error handling:
/// - **Offline**: recovery suppressed; restarts on [notifyAllOnlineRestored].
/// - **Timeout**: detected and handled internally with exponential backoff.
/// - **ChannelRateLimitReached**: detected and handled with a longer minimum
/// delay (5 s) before retrying. During recovery, a REST poll keeps data
@@ -92,6 +110,39 @@ typedef ChannelStatusCallback =
/// );
/// ```
class StreamRecoveryWrapper<T> {
// ── Static online check ───────────────────────────────────────────────────
/// Returns whether the device currently has a working internet connection.
///
/// Defaults to `() => true` (recovery enabled) until the connectivity
/// provider overrides it via [setIsOnlineCallback].
static bool Function() _globalIsOnline = () => true;
/// Configure the global online check. Call this once from
/// `connectivityMonitorProvider` so all wrappers suppress recovery while
/// offline without needing per-instance wiring.
static void setIsOnlineCallback(bool Function() fn) {
_globalIsOnline = fn;
}
// ── Static registry (WeakReference so GC can collect disposed wrappers) ──
static final List<WeakReference<StreamRecoveryWrapper>> _instances = [];
/// Notify every live wrapper that connectivity has been restored.
///
/// Wrappers whose stream is already [StreamConnectionStatus.connected] are
/// skipped. All others call [retry] to restart the Supabase realtime
/// subscription. Called by `connectivity_provider._onReconnected`.
static void notifyAllOnlineRestored() {
_instances.removeWhere((r) => r.target == null);
for (final ref in List.of(_instances)) {
ref.target?.notifyOnlineRestored();
}
}
// ─────────────────────────────────────────────────────────────────────────
final Stream<List<Map<String, dynamic>>> _realtimeStream;
final Future<List<T>> Function() _onPollData;
final T Function(Map<String, dynamic>) _fromMap;
@@ -105,6 +156,17 @@ class StreamRecoveryWrapper<T> {
/// skeleton indicators in the UI.
final ChannelStatusCallback? _onStatusChanged;
/// Optional callback to fetch data from the local offline cache (e.g. Brick
/// SQLite) when the device is offline. Called once when offline is first
/// detected to populate the stream so the UI shows cached data instead of
/// an empty list. If null, an empty offline result is emitted.
final Future<List<T>> Function()? _onOfflineData;
/// Optional callback invoked with each batch of live data so the provider
/// can mirror it into a local cache (e.g. Brick SQLite). Fires for every
/// realtime emission. Errors are caught and logged; mirroring is best-effort.
final void Function(List<T> rows)? _onCacheMirror;
StreamConnectionStatus _connectionStatus = StreamConnectionStatus.connected;
int _recoveryAttempts = 0;
Timer? _pollingTimer;
@@ -115,6 +177,11 @@ class StreamRecoveryWrapper<T> {
bool _disposed = false;
bool _listening = false;
/// True once live data has been emitted at least once. Used to distinguish
/// "offline at startup (never received data)" from "reconnecting after
/// having had data" for the race-condition fix.
bool _hasEmittedData = false;
StreamRecoveryWrapper({
required Stream<List<Map<String, dynamic>>> stream,
required Future<List<T>> Function() onPollData,
@@ -122,11 +189,21 @@ class StreamRecoveryWrapper<T> {
StreamRecoveryConfig config = const StreamRecoveryConfig(),
this.channelName = 'unknown',
ChannelStatusCallback? onStatusChanged,
Future<List<T>> Function()? onOfflineData,
void Function(List<T> rows)? onCacheMirror,
}) : _realtimeStream = stream,
_onPollData = onPollData,
_fromMap = fromMap,
_config = config,
_onStatusChanged = onStatusChanged;
_onStatusChanged = onStatusChanged,
_onOfflineData = onOfflineData,
_onCacheMirror = onCacheMirror {
// Register in the static registry so [notifyAllOnlineRestored] can reach
// this instance without requiring call-site changes.
_instances.add(WeakReference(this));
// Prune any dead references (disposed wrappers) to keep the list small.
_instances.removeWhere((r) => r.target == null);
}
/// The wrapped stream that emits recovery results with metadata.
///
@@ -161,6 +238,10 @@ class StreamRecoveryWrapper<T> {
void _onRealtimeData(List<Map<String, dynamic>> rows) {
if (_disposed) return;
// Mark that live data has been received at least once. This is used in
// the offline race-condition check in [_onRealtimeError].
_hasEmittedData = true;
// When recovering, don't reset _recoveryAttempts immediately.
// Supabase streams emit an initial REST fetch before the realtime
// channel is established. If the channel keeps failing, resetting
@@ -183,9 +264,22 @@ class StreamRecoveryWrapper<T> {
}
_setStatus(StreamConnectionStatus.connected);
final converted = rows.map(_fromMap).toList();
// Mirror live data into the local cache so cold-launching offline still
// shows recent state. Best-effort — never throws into the consumer.
final mirror = _onCacheMirror;
if (mirror != null) {
try {
mirror(converted);
} catch (e) {
debugPrint('StreamRecoveryWrapper[$channelName]: mirror failed: $e');
}
}
_emit(
StreamRecoveryResult<T>(
data: rows.map(_fromMap).toList(),
data: converted,
connectionStatus: StreamConnectionStatus.connected,
isStale: false,
),
@@ -198,6 +292,37 @@ class StreamRecoveryWrapper<T> {
// Cancel any stability timer — the connection is not stable.
_stabilityTimer?.cancel();
// Determine whether this error means the device is offline.
//
// Two cases handled:
// 1. Connectivity monitor already updated → _globalIsOnline() is false.
// 2. Race condition at startup: the realtime stream fails (SocketException)
// BEFORE ConnectivityMonitor.check() completes (it takes up to 5 s).
// In this window _globalIsOnline() still returns the default `true`
// but the device is actually offline. We detect this by requiring:
// (a) the error is a raw network connectivity failure, AND
// (b) no live data has been received yet on this wrapper.
// If data was already flowing and then dropped, we follow normal
// recovery so the user gets their data back when connectivity
// restores.
final isOffline = !_globalIsOnline() ||
(!_hasEmittedData && _isNetworkConnectivityError(error));
if (isOffline) {
debugPrint(
'StreamRecoveryWrapper[$channelName]: offline — suppressing recovery',
);
_setStatus(StreamConnectionStatus.offline);
// Emit offline data so the StreamProvider exits AsyncLoading state.
// Without this emission the provider stays in loading indefinitely when
// offline at startup, causing the skeleton shimmer to show even though
// Brick's SQLite cache has data.
if (!_hasEmittedData) {
unawaited(_emitOfflineData());
}
return;
}
final isRateLimit = _isRateLimitError(error);
final isTimeout = _isTimeoutError(error);
final tag = isRateLimit
@@ -260,6 +385,21 @@ class StreamRecoveryWrapper<T> {
void _onRealtimeDone() {
if (_disposed) return;
debugPrint('StreamRecoveryWrapper[$channelName]: stream completed');
// If offline, the channel closed because the network dropped. Suppress
// the restart — [notifyOnlineRestored] will handle it when we're back.
if (!_globalIsOnline()) {
debugPrint(
'StreamRecoveryWrapper[$channelName]: offline — suppressing stream restart',
);
_setStatus(StreamConnectionStatus.offline);
// Emit offline data to unblock any StreamProvider still in AsyncLoading.
if (!_hasEmittedData) {
unawaited(_emitOfflineData());
}
return;
}
// Attempt to reconnect once if the stream closes unexpectedly.
if (_recoveryAttempts < _config.maxRecoveryAttempts) {
_onRealtimeError(StateError('realtime stream completed unexpectedly'));
@@ -282,8 +422,23 @@ class StreamRecoveryWrapper<T> {
Future<void> _pollOnce() async {
if (_disposed) return;
// Don't poll when offline — it will fail and mislead the status tracker.
if (!_globalIsOnline()) return;
try {
final data = await _onPollData();
// Mirror polled data into the local cache too.
final mirror = _onCacheMirror;
if (mirror != null) {
try {
mirror(data);
} catch (e) {
debugPrint(
'StreamRecoveryWrapper[$channelName]: poll mirror failed: $e',
);
}
}
_emit(
StreamRecoveryResult<T>(
data: data,
@@ -330,6 +485,58 @@ class StreamRecoveryWrapper<T> {
// ── Helpers ───────────────────────────────────────────────────────────
/// Returns true when [error] is a raw network-connectivity failure
/// (socket closed, no route to host, DNS failure, etc.) rather than an
/// application-level error from the server.
///
/// Used in the startup race-condition fix: if the realtime stream errors
/// with a connectivity failure before the connectivity monitor has updated
/// [_globalIsOnline], we treat it as offline so the device shows cached
/// data instead of entering the recovery/shimmer loop.
static bool _isNetworkConnectivityError(Object error) {
final msg = error.toString().toLowerCase();
return msg.contains('socketexception') ||
msg.contains('connection refused') ||
msg.contains('network is unreachable') ||
msg.contains('no route to host') ||
msg.contains('failed host lookup') ||
msg.contains('connection reset') ||
msg.contains('broken pipe') ||
msg.contains('connection timed out') ||
(msg.contains('clientexception') && msg.contains('network')) ||
msg.contains('errno = 111') || // ECONNREFUSED
msg.contains('errno = 101'); // ENETUNREACH
}
/// Fetches data from the offline cache (via [_onOfflineData] callback when
/// provided, otherwise an empty list) and emits it as an
/// [StreamConnectionStatus.offline] result.
///
/// Called once when the device first goes offline so that consuming
/// [StreamProvider]s exit [AsyncLoading] and the UI shows cached content
/// (or an empty list) instead of the skeleton shimmer.
Future<void> _emitOfflineData() async {
if (_disposed) return;
List<T> cached = const [];
if (_onOfflineData != null) {
try {
cached = await _onOfflineData();
} catch (e) {
debugPrint(
'StreamRecoveryWrapper[$channelName]: offline cache read failed: $e',
);
cached = const [];
}
}
_emit(
StreamRecoveryResult<T>(
data: cached,
connectionStatus: StreamConnectionStatus.offline,
isStale: cached.isNotEmpty,
),
);
}
void _emit(StreamRecoveryResult<T> result) {
if (!_disposed && _controller != null && !_controller!.isClosed) {
_controller!.add(result);
@@ -358,6 +565,20 @@ class StreamRecoveryWrapper<T> {
_startRealtimeSubscription();
}
/// Called by [notifyAllOnlineRestored] when connectivity is restored.
///
/// Only restarts the subscription if the channel is not already connected.
/// Wrappers that were mid-backoff-timer while offline resume from scratch
/// rather than continuing an already-stale timer.
void notifyOnlineRestored() {
if (_disposed) return;
if (_connectionStatus == StreamConnectionStatus.connected) return;
debugPrint(
'StreamRecoveryWrapper[$channelName]: back online — restarting stream',
);
retry();
}
/// Clean up all resources and notify the status callback that this
/// channel is no longer active, preventing ghost entries in the
/// [RealtimeController]'s recovering-channels set.