Report driver Reverb connection status

app/src/main/java/pl/firmatpp/kierowca/data/DriverRepository.kt

@@ -18,6 +18,7 @@ import pl.firmatpp.kierowca.data.model.DriverDto
 import pl.firmatpp.kierowca.data.model.OtpResponse
 import pl.firmatpp.kierowca.data.model.PhotoUploadResponse
 import pl.firmatpp.kierowca.data.model.PushTokenBody
+import pl.firmatpp.kierowca.data.model.RealtimeStatusBody
 import pl.firmatpp.kierowca.data.model.RequestOtpBody
 import pl.firmatpp.kierowca.data.model.RouteResponse
 import pl.firmatpp.kierowca.data.model.SyncStateResponse
@@ -59,6 +60,10 @@ class DriverRepository(
         api.storePushToken(authHeader(requireToken()), PushTokenBody(token))
     }
 
+    suspend fun storeRealtimeStatus(status: String, socketId: String? = null, error: String? = null) {
+        api.storeRealtimeStatus(authHeader(requireToken()), RealtimeStatusBody(status, socketId, error))
+    }
+
     suspend fun deletePushToken() {
         val token = tokenStore.read() ?: return
         runCatching { api.deletePushToken(authHeader(token)) }

app/src/main/java/pl/firmatpp/kierowca/data/api/MobileDriverApi.kt

@@ -9,6 +9,7 @@ import pl.firmatpp.kierowca.data.model.CompleteRouteResponse
 import pl.firmatpp.kierowca.data.model.OtpResponse
 import pl.firmatpp.kierowca.data.model.PhotoUploadResponse
 import pl.firmatpp.kierowca.data.model.PushTokenBody
+import pl.firmatpp.kierowca.data.model.RealtimeStatusBody
 import pl.firmatpp.kierowca.data.model.RequestOtpBody
 import pl.firmatpp.kierowca.data.model.RouteResponse
 import pl.firmatpp.kierowca.data.model.SyncStateResponse
@@ -72,6 +73,12 @@ interface MobileDriverApi {
         @Body body: PushTokenBody,
     ): Map<String, Boolean>
 
+    @PUT("mobile/driver/realtime-status")
+    suspend fun storeRealtimeStatus(
+        @Header("Authorization") authorization: String,
+        @Body body: RealtimeStatusBody,
+    ): Map<String, Boolean>
+
     @DELETE("mobile/driver/push-token")
     suspend fun deletePushToken(
         @Header("Authorization") authorization: String,

app/src/main/java/pl/firmatpp/kierowca/data/model/DriverModels.kt

@@ -110,6 +110,12 @@ data class PushTokenBody(
     val platform: String = "android",
 )
 
+data class RealtimeStatusBody(
+    val reverbStatus: String,
+    val socketId: String? = null,
+    val error: String? = null,
+)
+
 data class DriverRouteDto(
     val id: String,
     val startsAt: String,

app/src/main/java/pl/firmatpp/kierowca/sync/DriverLiveSyncClient.kt

@@ -4,6 +4,7 @@ import com.google.gson.Gson
 import com.google.gson.JsonObject
 import com.google.gson.JsonParser
 import java.util.concurrent.atomic.AtomicBoolean
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
@@ -28,6 +29,8 @@ class DriverLiveSyncClient(
     private val started = AtomicBoolean(false)
     private var webSocket: WebSocket? = null
     private var driverId: String? = null
+    private var socketId: String? = null
+    private var heartbeatJob: Job? = null
 
     fun start(driverId: String) {
         if (BuildConfig.REVERB_APP_KEY.isBlank()) return
@@ -46,11 +49,15 @@ class DriverLiveSyncClient(
                 }
 
                 override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
+                    stopHeartbeat()
+                    reportRealtimeStatus("disconnected", reason.takeIf { it.isNotBlank() })
                     started.set(false)
                     scheduleReconnect()
                 }
 
                 override fun onFailure(webSocket: WebSocket, t: Throwable, response: okhttp3.Response?) {
+                    stopHeartbeat()
+                    reportRealtimeStatus("error", t.message ?: response?.message)
                     started.set(false)
                     scheduleReconnect()
                 }
@@ -60,9 +67,12 @@ class DriverLiveSyncClient(
 
     fun stop() {
         started.set(false)
+        stopHeartbeat()
+        reportRealtimeStatus("disconnected", "client_stop")
         webSocket?.close(1000, "logout")
         webSocket = null
         driverId = null
+        socketId = null
     }
 
     fun close() {
@@ -76,9 +86,17 @@ class DriverLiveSyncClient(
 
         when (event) {
             "pusher:connection_established" -> {
-                val socketId = root.dataObject()?.string("socket_id") ?: return
+                socketId = root.dataObject()?.string("socket_id") ?: return
-                subscribe(socket, socketId)
+                subscribe(socket, socketId ?: return)
-                onConnected()
+            }
+            "pusher_internal:subscription_succeeded" -> {
+                val channel = root.string("channel")
+                val id = driverId ?: return
+                if (channel == "private-driver-mobile.$id") {
+                    reportRealtimeStatus("connected")
+                    startHeartbeat()
+                    onConnected()
+                }
             }
             "DriverMobileSyncHint" -> parseHint(root.dataObject())?.let(onHint)
         }
@@ -132,4 +150,27 @@ class DriverLiveSyncClient(
             }
         }
     }
+
+    private fun startHeartbeat() {
+        heartbeatJob?.cancel()
+        heartbeatJob = scope.launch {
+            while (started.get()) {
+                reportRealtimeStatus("connected")
+                delay(30_000)
+            }
+        }
+    }
+
+    private fun stopHeartbeat() {
+        heartbeatJob?.cancel()
+        heartbeatJob = null
+    }
+
+    private fun reportRealtimeStatus(status: String, error: String? = null) {
+        scope.launch {
+            runCatching {
+                repository.storeRealtimeStatus(status, socketId, error)
+            }
+        }
+    }
 }