tasq/lib/providers/stream_recovery.dart

import 'dart:async';
import 'dart:math' as math;
import 'package:flutter/foundation.dart';

/// Connection status for a single stream subscription.
enum StreamConnectionStatus {
  /// Connected and receiving live updates.
  connected,

  /// Attempting to recover the connection; data may be stale.
  recovering,

  /// Connection failed; attempting to fallback to polling.
  polling,

  /// Connection and polling both failed; data is stale.
  stale,

  /// 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.
class PollResult<T> {
  final List<T> data;
  final bool success;
  final String? error;

  PollResult({required this.data, required this.success, this.error});
}

/// Configuration for stream recovery behavior.
class StreamRecoveryConfig {
  /// Maximum number of automatic recovery attempts before giving up.
  final int maxRecoveryAttempts;

  /// Initial delay (in milliseconds) before first recovery attempt.
  final int initialDelayMs;

  /// Maximum delay (in milliseconds) for exponential backoff.
  final int maxDelayMs;

  /// Multiplier for exponential backoff (e.g., 2.0 = double each attempt).
  final double backoffMultiplier;

  /// Enable polling fallback when realtime fails.
  final bool enablePollingFallback;

  /// Polling interval (in milliseconds) when realtime is unavailable.
  final int pollingIntervalMs;

  const StreamRecoveryConfig({
    this.maxRecoveryAttempts = 4,
    this.initialDelayMs = 1000,
    this.maxDelayMs = 32000, // 32 seconds max
    this.backoffMultiplier = 2.0,
    this.enablePollingFallback = true,
    this.pollingIntervalMs = 5000, // Poll every 5 seconds
  });
}

/// Callback type for per-channel status change notifications.
///
/// Used by [StreamRecoveryWrapper] to notify [RealtimeController] about
/// individual channel recovery state so the UI can show per-channel skeletons.
typedef ChannelStatusCallback =
    void Function(String channel, StreamConnectionStatus status);

/// Wraps a Supabase realtime stream with automatic recovery, polling fallback,
/// 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
///   fresh so the UI shows shimmer/skeleton instead of an error.
/// - **Generic errors**: same recovery flow with standard backoff.
///
/// Errors are **never** forwarded to consumers; instead the wrapper emits
/// [StreamRecoveryResult] events with appropriate [connectionStatus] so the
/// UI can react per-channel (e.g. show skeleton shimmer).
///
/// Usage:
/// ```dart
/// final wrapper = StreamRecoveryWrapper<Task>(
///   stream: client.from('tasks').stream(primaryKey: ['id']),
///   onPollData: () => fetchTasksViaRest(),
///   fromMap: Task.fromMap,
///   channelName: 'tasks',
///   onStatusChanged: (channel, status) { ... },
/// );
/// ```
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;
  final StreamRecoveryConfig _config;

  /// Human-readable channel name for logging and per-channel status tracking.
  final String channelName;

  /// Optional callback invoked whenever this channel's connection status
  /// changes. Used to integrate with [RealtimeController] for per-channel
  /// 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;
  Timer? _recoveryTimer;
  Timer? _stabilityTimer;
  StreamSubscription<List<Map<String, dynamic>>>? _realtimeSub;
  StreamController<StreamRecoveryResult<T>>? _controller;
  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,
    required T Function(Map<String, dynamic>) fromMap,
    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,
       _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.
  ///
  /// Lazily initializes the internal controller and starts listening to the
  /// realtime stream on first access. Errors from the realtime channel are
  /// handled internally — consumers only see [StreamRecoveryResult] events.
  Stream<StreamRecoveryResult<T>> get stream {
    if (_controller == null) {
      _controller = StreamController<StreamRecoveryResult<T>>.broadcast();
      _startRealtimeSubscription();
    }
    return _controller!.stream;
  }

  /// Current connection status of this stream.
  StreamConnectionStatus get connectionStatus => _connectionStatus;

  // ── Realtime subscription ───────────────────────────────────────────────

  void _startRealtimeSubscription() {
    if (_disposed || _listening) return;
    _listening = true;
    _realtimeSub?.cancel();
    _realtimeSub = _realtimeStream.listen(
      _onRealtimeData,
      onError: _onRealtimeError,
      onDone: _onRealtimeDone,
      cancelOnError: false, // keep listening even after transient errors
    );
  }

  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
    // on that REST data creates an infinite loop (data → reset → subscribe
    // → fail → data → reset …).  Instead, start a stability timer — only
    // reset after staying connected without errors for 30 seconds.
    if (_recoveryAttempts > 0) {
      _stabilityTimer?.cancel();
      _stabilityTimer = Timer(const Duration(seconds: 30), () {
        if (!_disposed) {
          _recoveryAttempts = 0;
          debugPrint(
            'StreamRecoveryWrapper[$channelName]: '
            'connection stable for 30s — recovery counter reset',
          );
        }
      });
    } else {
      _recoveryAttempts = 0;
    }

    _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: converted,
        connectionStatus: StreamConnectionStatus.connected,
        isStale: false,
      ),
    );
  }

  void _onRealtimeError(Object error, [StackTrace? stack]) {
    if (_disposed) return;

    // 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
        ? ' (rate-limited)'
        : isTimeout
        ? ' (timeout)'
        : '';

    debugPrint('StreamRecoveryWrapper[$channelName]: stream error$tag: $error');

    _setStatus(StreamConnectionStatus.recovering);

    if (_recoveryAttempts >= _config.maxRecoveryAttempts) {
      if (_config.enablePollingFallback) {
        _startPollingFallback();
      } else {
        _setStatus(StreamConnectionStatus.failed);
        _emit(
          StreamRecoveryResult<T>(
            data: const [],
            connectionStatus: StreamConnectionStatus.failed,
            isStale: true,
            error: error.toString(),
          ),
        );
      }
      return;
    }

    _recoveryAttempts++;

    // Compute backoff delay. Rate-limit errors get a longer floor (5 s).
    final baseDelay =
        _config.initialDelayMs *
        math.pow(_config.backoffMultiplier, _recoveryAttempts - 1);
    final effectiveDelay = isRateLimit
        ? math.max(baseDelay.toInt(), 5000)
        : baseDelay.toInt();
    final cappedDelay = math.min(effectiveDelay, _config.maxDelayMs);

    debugPrint(
      'StreamRecoveryWrapper[$channelName]: recovery attempt '
      '$_recoveryAttempts/${_config.maxRecoveryAttempts}, '
      'delay=${cappedDelay}ms$tag',
    );

    // Fire a single REST poll immediately so the UI can show fresh data
    // under the skeleton shimmer while waiting for realtime to reconnect.
    _pollOnce();

    // Schedule re-subscription after backoff.
    _recoveryTimer?.cancel();
    _recoveryTimer = Timer(Duration(milliseconds: cappedDelay), () {
      if (_disposed) return;
      _listening = false;
      _startRealtimeSubscription();
    });
  }

  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'));
    }
  }

  // ── Polling fallback ──────────────────────────────────────────────────

  void _startPollingFallback() {
    _realtimeSub?.cancel();
    _listening = false;
    _setStatus(StreamConnectionStatus.polling);
    _pollingTimer?.cancel();
    _pollOnce(); // Immediate first poll
    _pollingTimer = Timer.periodic(
      Duration(milliseconds: _config.pollingIntervalMs),
      (_) => _pollOnce(),
    );
  }

  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,
          connectionStatus: _connectionStatus,
          isStale: true,
        ),
      );
    } catch (e) {
      debugPrint('StreamRecoveryWrapper[$channelName]: poll error: $e');
      if (_connectionStatus == StreamConnectionStatus.polling) {
        _setStatus(StreamConnectionStatus.stale);
        _emit(
          StreamRecoveryResult<T>(
            data: const [],
            connectionStatus: StreamConnectionStatus.stale,
            isStale: true,
            error: e.toString(),
          ),
        );
      }
    }
  }

  // ── Error classification ──────────────────────────────────────────────

  /// Whether [error] indicates a Supabase channel rate limit.
  static bool _isRateLimitError(Object error) {
    final msg = error.toString().toLowerCase();
    return msg.contains('rate limit') ||
        msg.contains('rate_limit') ||
        msg.contains('channelratelimitreached') ||
        msg.contains('too many') ||
        msg.contains('429');
  }

  /// Whether [error] indicates a subscription timeout.
  static bool _isTimeoutError(Object error) {
    if (error is TimeoutException) return true;
    final msg = error.toString().toLowerCase();
    return msg.contains('timeout') ||
        msg.contains('timed out') ||
        msg.contains('timed_out');
  }

  // ── 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);
    }
  }

  /// Update connection status and notify the per-channel callback.
  void _setStatus(StreamConnectionStatus status) {
    if (_connectionStatus != status) {
      _connectionStatus = status;
      _onStatusChanged?.call(channelName, status);
    }
  }

  /// Immediately fetch fresh data via REST without restarting the realtime
  /// subscription. Use this as a periodic safety net for missed realtime events
  /// (e.g., when the table is not yet in the supabase_realtime publication).
  Future<void> pollNow() async => _pollOnce();

  /// Manually trigger a recovery attempt.
  void retry() {
    _recoveryAttempts = 0;
    _pollingTimer?.cancel();
    _recoveryTimer?.cancel();
    _listening = false;
    _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.
  void dispose() {
    if (_disposed) return;
    _disposed = true;
    _pollingTimer?.cancel();
    _recoveryTimer?.cancel();
    _stabilityTimer?.cancel();
    _realtimeSub?.cancel();
    // Ensure the channel is removed from the recovering set when the
    // wrapper is torn down (e.g. provider disposed during navigation).
    // Without this, disposed wrappers that were mid-recovery leave
    // orphaned entries that keep the reconnection indicator spinning.
    if (_connectionStatus != StreamConnectionStatus.connected) {
      _onStatusChanged?.call(channelName, StreamConnectionStatus.connected);
    }
    _controller?.close();
  }
}

/// Result of a stream emission, including metadata about connection status.
class StreamRecoveryResult<T> {
  /// The data emitted by the stream.
  final List<T> data;

  /// Current connection status.
  final StreamConnectionStatus connectionStatus;

  /// Whether the data is stale (not live from realtime).
  final bool isStale;

  /// Error message, if any.
  final String? error;

  StreamRecoveryResult({
    required this.data,
    required this.connectionStatus,
    required this.isStale,
    this.error,
  });

  /// True if data is live and reliable.
  bool get isLive => connectionStatus == StreamConnectionStatus.connected;

  /// True if we should show a "data may be stale" indicator.
  bool get shouldIndicateStale =>
      isStale || connectionStatus == StreamConnectionStatus.polling;
}