Harden driver app realtime release

2026-07-05 03:15:54 +02:00
parent 72db88b25c
commit b1a34105f1
17 changed files with 1226 additions and 385 deletions
@@ -3,10 +3,10 @@ package pl.firmatpp.kierowca.sync
 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 java.util.concurrent.TimeUnit
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
@@ -16,75 +16,155 @@ import okhttp3.Request
 import okhttp3.WebSocket
 import okhttp3.WebSocketListener
 import pl.firmatpp.kierowca.data.DriverRepository
+import pl.firmatpp.kierowca.data.model.BroadcastAuthResponse
 import pl.firmatpp.kierowca.data.model.RealtimeConfigDto
 import pl.firmatpp.kierowca.diagnostics.AppDiagnostics

-class DriverLiveSyncClient(
+interface DriverLiveSyncGateway {
+    suspend fun broadcastAuth(socketId: String, channelName: String): BroadcastAuthResponse
+    suspend fun storeRealtimeStatus(status: String, socketId: String?, error: String?)
+}
+
+class DriverRepositoryLiveSyncGateway(
    private val repository: DriverRepository,
+) : DriverLiveSyncGateway {
+    override suspend fun broadcastAuth(socketId: String, channelName: String): BroadcastAuthResponse =
+        repository.broadcastAuth(socketId, channelName)
+
+    override suspend fun storeRealtimeStatus(status: String, socketId: String?, error: String?) {
+        repository.storeRealtimeStatus(status, socketId, error)
+    }
+}
+
+interface LiveWebSocketFactory {
+    fun newWebSocket(url: String, listener: WebSocketListener): WebSocket
+}
+
+class OkHttpLiveWebSocketFactory(
+    private val client: OkHttpClient,
+) : LiveWebSocketFactory {
+    override fun newWebSocket(url: String, listener: WebSocketListener): WebSocket =
+        client.newWebSocket(Request.Builder().url(url).build(), listener)
+}
+
+private enum class LiveSyncConnectionState {
+    Stopped,
+    WaitingForNetwork,
+    Disconnected,
+    Connecting,
+    Subscribing,
+    Connected,
+}
+
+class DriverLiveSyncClient(
+    private val gateway: DriverLiveSyncGateway,
    private val onConnected: () -> Unit,
    private val onHint: (DriverSyncHint) -> Unit,
-    private val client: OkHttpClient = OkHttpClient(),
+    private val webSocketFactory: LiveWebSocketFactory = OkHttpLiveWebSocketFactory(defaultOkHttpClient()),
    private val gson: Gson = Gson(),
+    private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.IO),
+    private val reconnectDelaysMs: List<Long> = DEFAULT_RECONNECT_DELAYS_MS,
+    private val staleTimeoutMs: Long = STALE_TIMEOUT_MS,
 ) {
+    constructor(
+        repository: DriverRepository,
+        onConnected: () -> Unit,
+        onHint: (DriverSyncHint) -> Unit,
+        client: OkHttpClient = defaultOkHttpClient(),
+        gson: Gson = Gson(),
+    ) : this(
+        gateway = DriverRepositoryLiveSyncGateway(repository),
+        onConnected = onConnected,
+        onHint = onHint,
+        webSocketFactory = OkHttpLiveWebSocketFactory(client),
+        gson = gson,
+    )
+
    private companion object {
        const val HEARTBEAT_INTERVAL_MS = 10_000L
+        const val STALE_TIMEOUT_MS = 30_000L
+        val DEFAULT_RECONNECT_DELAYS_MS = listOf(1_000L, 2_000L, 5_000L, 10_000L, 30_000L)
+
+        fun defaultOkHttpClient(): OkHttpClient =
+            OkHttpClient.Builder()
+                .pingInterval(15, TimeUnit.SECONDS)
+                .build()
    }

-    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
-    private val started = AtomicBoolean(false)
+    private val lock = Any()
+    private var desiredActive: Boolean = false
+    private var networkAvailable: Boolean = true
+    private var state: LiveSyncConnectionState = LiveSyncConnectionState.Stopped
    private var webSocket: WebSocket? = null
    private var driverId: String? = null
    private var realtimeConfig: RealtimeConfigDto? = null
    private var socketId: String? = null
+    private var reconnectJob: Job? = null
    private var heartbeatJob: Job? = null
+    private var staleWatchdogJob: Job? = null
+    private var reconnectAttempt: Int = 0
+    private var messageVersion: Long = 0

    fun start(driverId: String, config: RealtimeConfigDto?) {
-        val appKey = config?.reverbAppKey?.takeIf { it.isNotBlank() } ?: return
-        val wsBaseUrl = config.reverbWsBaseUrl?.takeIf { it.isNotBlank() } ?: return
-        if (!config.reverbEnabled) return
+        if (!isConfigUsable(config)) return

-        realtimeConfig = config
-        this.driverId = driverId
-        if (!started.compareAndSet(false, true)) return
+        synchronized(lock) {
+            realtimeConfig = config
+            this.driverId = driverId
+            desiredActive = true
+        }
+        ensureConnected()
+    }

-        val wsUrl = wsBaseUrl.trimEnd('/') +
-            "/" + appKey +
-            "?protocol=7&client=android&version=1.0&flash=false"
+    fun ensureConnected() {
+        val shouldConnect = synchronized(lock) {
+            desiredActive &&
+                networkAvailable &&
+                state !in setOf(
+                    LiveSyncConnectionState.Connecting,
+                    LiveSyncConnectionState.Subscribing,
+                    LiveSyncConnectionState.Connected,
+                )
+        }
+        if (shouldConnect) connectNow()
+    }

-        webSocket = client.newWebSocket(
-            Request.Builder().url(wsUrl).build(),
-            object : WebSocketListener() {
-                override fun onMessage(webSocket: WebSocket, text: String) {
-                    handleMessage(webSocket, text)
-                }
+    fun onNetworkAvailable(available: Boolean) {
+        val shouldReconnect = synchronized(lock) {
+            networkAvailable = available
+            if (!available) {
+                reconnectJob?.cancel()
+                reconnectJob = null
+                closeSocketLocked("network_lost")
+                state = if (desiredActive) LiveSyncConnectionState.WaitingForNetwork else LiveSyncConnectionState.Stopped
+                false
+            } else {
+                desiredActive && state == LiveSyncConnectionState.WaitingForNetwork
+            }
+        }

-                override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
-                    stopHeartbeat()
-                    reportRealtimeStatus("disconnected", reason.takeIf { it.isNotBlank() })
-                    started.set(false)
-                    scheduleReconnect()
-                }
+        if (available) {
+            reportRealtimeStatus("reconnecting", "network_available")
+        } else {
+            reportRealtimeStatus("disconnected", "network_lost")
+        }

-                override fun onFailure(webSocket: WebSocket, t: Throwable, response: okhttp3.Response?) {
-                    stopHeartbeat()
-                    AppDiagnostics.log("realtime_error: ${t.message ?: response?.message ?: "unknown"}")
-                    reportRealtimeStatus("error", t.message ?: response?.message)
-                    started.set(false)
-                    scheduleReconnect()
-                }
-            },
-        )
+        if (shouldReconnect) connectNow(resetAttempt = true)
    }

    fun stop() {
-        started.set(false)
-        stopHeartbeat()
+        synchronized(lock) {
+            desiredActive = false
+            reconnectJob?.cancel()
+            reconnectJob = null
+            closeSocketLocked("client_stop")
+            driverId = null
+            realtimeConfig = null
+            socketId = null
+            reconnectAttempt = 0
+            state = LiveSyncConnectionState.Stopped
+        }
        reportRealtimeStatus("disconnected", "client_stop")
-        webSocket?.close(1000, "logout")
-        webSocket = null
-        driverId = null
-        realtimeConfig = null
-        socketId = null
    }

    fun close() {
@@ -92,35 +172,112 @@ class DriverLiveSyncClient(
        scope.cancel()
    }

+    private fun connectNow(resetAttempt: Boolean = false) {
+        val wsUrl = synchronized(lock) {
+            val config = realtimeConfig ?: return
+            val id = driverId ?: return
+            if (!desiredActive || !networkAvailable || !isConfigUsable(config)) return
+            if (state == LiveSyncConnectionState.Connecting || state == LiveSyncConnectionState.Subscribing || state == LiveSyncConnectionState.Connected) return
+
+            if (resetAttempt) reconnectAttempt = 0
+            reconnectJob?.cancel()
+            reconnectJob = null
+            socketId = null
+            state = LiveSyncConnectionState.Connecting
+
+            val appKey = config.reverbAppKey.orEmpty()
+            val wsBaseUrl = config.reverbWsBaseUrl.orEmpty()
+            wsBaseUrl.trimEnd('/') + "/" + appKey + "?protocol=7&client=android&version=1.0&flash=false"
+        }
+
+        val socket = webSocketFactory.newWebSocket(
+            wsUrl,
+            object : WebSocketListener() {
+                override fun onMessage(webSocket: WebSocket, text: String) {
+                    handleMessage(webSocket, text)
+                }
+
+                override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
+                    handleDisconnect(
+                        status = "disconnected",
+                        reason = reason.takeIf { it.isNotBlank() } ?: "closed",
+                        socket = webSocket,
+                    )
+                }
+
+                override fun onFailure(webSocket: WebSocket, t: Throwable, response: okhttp3.Response?) {
+                    AppDiagnostics.log("realtime_error: ${t.message ?: response?.message ?: "unknown"}")
+                    handleDisconnect(
+                        status = "error",
+                        reason = t.message ?: response?.message ?: "failure",
+                        socket = webSocket,
+                    )
+                }
+            },
+        )
+
+        synchronized(lock) {
+            webSocket = socket
+        }
+    }
+
    private fun handleMessage(socket: WebSocket, text: String) {
+        val isCurrentSocket = synchronized(lock) {
+            if (webSocket !== socket) {
+                false
+            } else {
+                messageVersion += 1
+                true
+            }
+        }
+        if (!isCurrentSocket) return
+
        val root = runCatching { JsonParser.parseString(text).asJsonObject }.getOrNull() ?: return
        val event = root.string("event") ?: return

        when (event) {
+            "pusher:ping" -> {
+                socket.send("""{"event":"pusher:pong","data":{}}""")
+                restartStaleWatchdogIfConnected()
+            }
            "pusher:connection_established" -> {
-                socketId = root.dataObject()?.string("socket_id") ?: return
-                subscribe(socket, socketId ?: return)
+                val nextSocketId = root.dataObject()?.string("socket_id") ?: return
+                synchronized(lock) {
+                    socketId = nextSocketId
+                    state = LiveSyncConnectionState.Subscribing
+                }
+                subscribe(socket, nextSocketId)
            }
            "pusher_internal:subscription_succeeded" -> {
                val channel = root.string("channel")
-                val id = driverId ?: return
+                val id = synchronized(lock) { driverId } ?: return
                if (channel == "private-driver-mobile.$id") {
+                    synchronized(lock) {
+                        reconnectAttempt = 0
+                        state = LiveSyncConnectionState.Connected
+                    }
                    reportRealtimeStatus("connected")
                    startHeartbeat()
+                    startStaleWatchdog()
                    onConnected()
                }
            }
-            "DriverMobileSyncHint" -> parseHint(root.dataObject())?.let(onHint)
+            "DriverMobileSyncHint" -> {
+                restartStaleWatchdogIfConnected()
+                parseHint(root.dataObject())?.let(onHint)
+            }
+            else -> restartStaleWatchdogIfConnected()
        }
    }

    private fun subscribe(socket: WebSocket, socketId: String) {
-        val id = driverId ?: return
+        val id = synchronized(lock) { driverId } ?: return
        val channel = "private-driver-mobile.$id"

        scope.launch {
            runCatching {
-                val auth = repository.broadcastAuth(socketId, channel).auth
+                val auth = gateway.broadcastAuth(socketId, channel).auth
+                if (!isCurrentSocket(socket)) return@launch
                val payload = mapOf(
                    "event" to "pusher:subscribe",
                    "data" to mapOf(
@@ -128,11 +285,129 @@ class DriverLiveSyncClient(
                        "auth" to auth,
                    ),
                )
-                socket.send(gson.toJson(payload))
+                check(socket.send(gson.toJson(payload))) { "WebSocket send returned false" }
+            }.onFailure { throwable ->
+                AppDiagnostics.log("realtime_subscription_error: ${throwable.message ?: throwable::class.java.simpleName}")
+                socket.close(1000, "subscription_failed")
+                handleDisconnect("error", "subscription_failed", socket)
            }
        }
    }

+    private fun handleDisconnect(status: String, reason: String?, socket: WebSocket? = null) {
+        val shouldSchedule = synchronized(lock) {
+            if (socket != null && webSocket !== socket) return
+
+            stopHeartbeatLocked()
+            stopStaleWatchdogLocked()
+            webSocket = null
+            socketId = null
+
+            if (!desiredActive) {
+                state = LiveSyncConnectionState.Stopped
+                false
+            } else if (!networkAvailable) {
+                state = LiveSyncConnectionState.WaitingForNetwork
+                false
+            } else {
+                state = LiveSyncConnectionState.Disconnected
+                true
+            }
+        }
+
+        reportRealtimeStatus(status, reason)
+        if (shouldSchedule) scheduleReconnect()
+    }
+
+    private fun scheduleReconnect() {
+        val delayMs = synchronized(lock) {
+            if (!desiredActive || !networkAvailable || state == LiveSyncConnectionState.Stopped) return
+            if (reconnectJob?.isActive == true) return
+            val delay = reconnectDelaysMs.getOrElse(reconnectAttempt) { reconnectDelaysMs.last() }
+            reconnectAttempt += 1
+            delay
+        }
+
+        reconnectJob = scope.launch {
+            delay(delayMs)
+            connectNow()
+        }
+    }
+
+    private fun startHeartbeat() {
+        synchronized(lock) {
+            heartbeatJob?.cancel()
+            heartbeatJob = scope.launch {
+                while (true) {
+                    reportRealtimeStatus("connected")
+                    delay(HEARTBEAT_INTERVAL_MS)
+                }
+            }
+        }
+    }
+
+    private fun startStaleWatchdog() {
+        synchronized(lock) {
+            stopStaleWatchdogLocked()
+            val observedVersion = messageVersion
+            staleWatchdogJob = scope.launch {
+                delay(staleTimeoutMs)
+                val socketToClose = synchronized(lock) {
+                    if (state == LiveSyncConnectionState.Connected && messageVersion == observedVersion) {
+                        webSocket
+                    } else {
+                        null
+                    }
+                }
+
+                if (socketToClose != null) {
+                    socketToClose.close(1001, "stale_connection")
+                    handleDisconnect("error", "stale_connection", socketToClose)
+                    ensureConnected()
+                }
+            }
+        }
+    }
+
+    private fun restartStaleWatchdogIfConnected() {
+        val connected = synchronized(lock) { state == LiveSyncConnectionState.Connected }
+        if (connected) startStaleWatchdog()
+    }
+
+    private fun isCurrentSocket(socket: WebSocket): Boolean =
+        synchronized(lock) { webSocket === socket }
+
+    private fun closeSocketLocked(reason: String) {
+        stopHeartbeatLocked()
+        stopStaleWatchdogLocked()
+        webSocket?.close(1000, reason)
+        webSocket = null
+        socketId = null
+    }
+
+    private fun stopHeartbeatLocked() {
+        heartbeatJob?.cancel()
+        heartbeatJob = null
+    }
+
+    private fun stopStaleWatchdogLocked() {
+        staleWatchdogJob?.cancel()
+        staleWatchdogJob = null
+    }
+
+    private fun reportRealtimeStatus(status: String, error: String? = null) {
+        scope.launch {
+            runCatching {
+                gateway.storeRealtimeStatus(status, socketId, error)
+            }
+        }
+    }
+
+    private fun isConfigUsable(config: RealtimeConfigDto?): Boolean =
+        config?.reverbEnabled == true &&
+            !config.reverbAppKey.isNullOrBlank() &&
+            !config.reverbWsBaseUrl.isNullOrBlank()
+
    private fun parseHint(data: JsonObject?): DriverSyncHint? {
        if (data == null || data.string("type") != "driver_sync_hint") return null

@@ -152,37 +427,4 @@ class DriverLiveSyncClient(

    private fun JsonObject.string(name: String): String? =
        get(name)?.takeIf { !it.isJsonNull }?.asString
-
-    private fun scheduleReconnect() {
-        val id = driverId ?: return
-        scope.launch {
-            delay(5_000)
-            if (!started.get() && driverId == id) {
-                start(id, realtimeConfig)
-            }
-        }
-    }
-
-    private fun startHeartbeat() {
-        heartbeatJob?.cancel()
-        heartbeatJob = scope.launch {
-            while (started.get()) {
-                reportRealtimeStatus("connected")
-                delay(HEARTBEAT_INTERVAL_MS)
-            }
-        }
-    }
-
-    private fun stopHeartbeat() {
-        heartbeatJob?.cancel()
-        heartbeatJob = null
-    }
-
-    private fun reportRealtimeStatus(status: String, error: String? = null) {
-        scope.launch {
-            runCatching {
-                repository.storeRealtimeStatus(status, socketId, error)
-            }
-        }
-    }
 }