From eb67c70fa0de4f8d4321c56ef4a6f5c87edcc8c4 Mon Sep 17 00:00:00 2001
From: Inex Code <inex.code@selfprivacy.org>
Date: Tue, 6 Aug 2024 19:46:52 +0300
Subject: [PATCH] feat: When Jobs websocket subscription disconnects, fall back
 to usual queries for two minutes

Resolves #542
---
 .../graphql_maps/graphql_api_map.dart         |  5 +++-
 .../graphql_maps/server_api/jobs_api.dart     |  8 ++++--
 .../get_it/api_connection_repository.dart     | 28 ++++++++++++++++---
 3 files changed, 34 insertions(+), 7 deletions(-)

diff --git a/lib/logic/api_maps/graphql_maps/graphql_api_map.dart b/lib/logic/api_maps/graphql_maps/graphql_api_map.dart
index 80d345b8..511ae78f 100644
--- a/lib/logic/api_maps/graphql_maps/graphql_api_map.dart
+++ b/lib/logic/api_maps/graphql_maps/graphql_api_map.dart
@@ -97,12 +97,15 @@ abstract class GraphQLApiMap {
     );
   }
 
-  Future<GraphQLClient> getSubscriptionClient() async {
+  Future<GraphQLClient> getSubscriptionClient({
+    final Future<Duration?>? Function(int?, String?)? onConnectionLost,
+  }) async {
     final WebSocketLink webSocketLink = WebSocketLink(
       'ws://api.$rootAddress/graphql',
       // Only [GraphQLProtocol.graphqlTransportWs] supports automatic pings, so we don't disconnect when nothing happens.
       subProtocol: GraphQLProtocol.graphqlTransportWs,
       config: SocketClientConfig(
+        onConnectionLost: onConnectionLost,
         autoReconnect: true,
         initialPayload: _token.isEmpty
             ? null
diff --git a/lib/logic/api_maps/graphql_maps/server_api/jobs_api.dart b/lib/logic/api_maps/graphql_maps/server_api/jobs_api.dart
index 8f667465..50149b7e 100644
--- a/lib/logic/api_maps/graphql_maps/server_api/jobs_api.dart
+++ b/lib/logic/api_maps/graphql_maps/server_api/jobs_api.dart
@@ -22,8 +22,12 @@ mixin JobsApi on GraphQLApiMap {
     return jobsList;
   }
 
-  Stream<List<ServerJob>> getServerJobsStream() async* {
-    final GraphQLClient client = await getSubscriptionClient();
+  Stream<List<ServerJob>> getServerJobsStream({
+    final Future<Duration?>? Function(int?, String?)? onConnectionLost,
+  }) async* {
+    final GraphQLClient client = await getSubscriptionClient(
+      onConnectionLost: onConnectionLost,
+    );
     final subscription = client.subscribe$JobUpdates();
     await for (final response in subscription) {
       final jobsList = response.parsedData?.jobUpdates
diff --git a/lib/logic/get_it/api_connection_repository.dart b/lib/logic/get_it/api_connection_repository.dart
index 99f5d9b8..885c90f0 100644
--- a/lib/logic/get_it/api_connection_repository.dart
+++ b/lib/logic/get_it/api_connection_repository.dart
@@ -45,6 +45,7 @@ class ApiConnectionRepository {
   Timer? _timer;
 
   StreamSubscription<List<ServerJob>>? _serverJobsStreamSubscription;
+  DateTime? _jobsStreamDisconnectTime;
 
   Future<void> removeServerJob(final String uid) async {
     await api.removeApiJob(uid);
@@ -291,10 +292,12 @@ class ApiConnectionRepository {
 
     if (VersionConstraint.parse(wsJobsUpdatesSupportedVersion)
         .allows(Version.parse(apiVersion))) {
-      _serverJobsStreamSubscription =
-          api.getServerJobsStream().listen((final List<ServerJob> jobs) {
+      _serverJobsStreamSubscription = api
+          .getServerJobsStream(onConnectionLost: _handleWebsocketDisconnect)
+          .listen((final List<ServerJob> jobs) {
         _apiData.serverJobs.data = jobs;
         _dataStream.add(_apiData);
+        _jobsStreamDisconnectTime = null;
       });
     }
 
@@ -314,9 +317,26 @@ class ApiConnectionRepository {
 
   static const String wsJobsUpdatesSupportedVersion = '>=3.3.0';
 
+  bool _isForceServerJobsRefetchRequired() {
+    if (_serverJobsStreamSubscription == null) {
+      return true;
+    }
+    return _apiData.serverJobs.data == null ||
+        (_jobsStreamDisconnectTime != null &&
+            DateTime.now().difference(_jobsStreamDisconnectTime!) <
+                const Duration(seconds: 120));
+  }
+
+  Future<Duration?>? _handleWebsocketDisconnect(
+    final int? code,
+    final String? reason,
+  ) {
+    _jobsStreamDisconnectTime = DateTime.now();
+    return null;
+  }
+
   Future<void> _refetchEverything(final Version version) async {
-    if (_serverJobsStreamSubscription == null ||
-        _apiData.serverJobs.data == null) {
+    if (_isForceServerJobsRefetchRequired()) {
       await _apiData.serverJobs
           .refetchData(version, () => _dataStream.add(_apiData));
     }