fix(sync): recover desktop syncing after a session goes stale

client/auth/data/impl/build.gradle.kts

@@ -13,6 +13,7 @@ kotlin {
             implementation(libs.supabase.storage)
             implementation(libs.supabase.functions)
         }
+        commonTest.dependencies { implementation(libs.ktor.client.mock) }
         jvmMain.dependencies { implementation(libs.ktor.client.cio) }
         androidMain.dependencies { implementation(libs.ktor.client.cio) }
         appleMain.dependencies { implementation(libs.ktor.client.darwin) }

client/auth/data/impl/src/commonMain/kotlin/com/plusmobileapps/chefmate/auth/data/impl/ExpiredTokenRetry.kt

@@ -0,0 +1,95 @@
+package com.plusmobileapps.chefmate.auth.data.impl
+
+import co.touchlab.kermit.Logger
+import io.ktor.client.plugins.api.Send
+import io.ktor.client.plugins.api.createClientPlugin
+import io.ktor.http.HttpHeaders
+import io.ktor.http.HttpStatusCode
+import io.ktor.http.encodedPath
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+
+/**
+ * Renews the access token and replays the request whenever Supabase rejects one as unauthorized.
+ *
+ * The SDK's auto-refresh is a single in-process timer, and outside Android nothing re-arms it when
+ * the machine sleeps through its window. A 401 is the only signal that doesn't depend on that timer
+ * — or on the client clock being right — so it's the one worth acting on.
+ *
+ * This sits below every repository, which matters: the sync paths swallow per-item failures to keep
+ * one bad row from stopping a whole reconcile, so a 401 surfaced any higher would be caught and
+ * dropped before anything could react to it. Down here the request simply succeeds on the retry.
+ */
+internal class ExpiredTokenRetryConfig {
+    /**
+     * Renews the session and returns a usable token, or null if it couldn't be renewed — in which
+     * case the original 401 stands rather than a second doomed request being sent.
+     *
+     * Receives the token the rejected request carried, so an implementation can tell "nobody has
+     * refreshed yet" from "someone already did". See [ConcurrentRefreshGuard].
+     */
+    var refreshToken: suspend (usedToken: String?) -> String? = { null }
+}
+
+internal val ExpiredTokenRetry =
+    createClientPlugin("ExpiredTokenRetry", ::ExpiredTokenRetryConfig) {
+        val refreshToken = pluginConfig.refreshToken
+        on(Send) { request ->
+            val call = proceed(request)
+            if (call.response.status != HttpStatusCode.Unauthorized) return@on call
+            // The token endpoint answers a dead refresh token with a 401 of its own. Retrying that
+            // is how one expired session turns into an infinite refresh loop.
+            if (request.url.encodedPath.startsWith(AUTH_PATH)) return@on call
+
+            val usedToken = request.headers[HttpHeaders.Authorization]?.removePrefix(BEARER_PREFIX)
+            val freshToken = refreshToken(usedToken) ?: return@on call
+            // Replace rather than append: bearerAuth() would add a second Authorization header.
+            request.headers[HttpHeaders.Authorization] = BEARER_PREFIX + freshToken
+            proceed(request)
+        }
+    }
+
+/**
+ * Collapses a burst of 401s into a single refresh.
+ *
+ * A full sync fires many requests at once, so a dead token produces many simultaneous 401s. Letting
+ * each one refresh independently is not merely wasteful: Supabase rotates the refresh token on
+ * every use, so parallel refreshes look like token replay and can invalidate the session outright —
+ * taking out the sign-in this code is trying to rescue.
+ *
+ * Callers hand over the token their request actually used. If the current token has moved on, some
+ * other caller already refreshed and that result is reused; otherwise this one refreshes, under a
+ * lock, while the rest wait.
+ */
+internal class ConcurrentRefreshGuard(
+    private val currentToken: () -> String?,
+    private val refresh: suspend () -> Unit,
+) {
+    private val mutex = Mutex()
+
+    suspend fun tokenAfterRefresh(usedToken: String?): String? = mutex.withLock {
+        val current = currentToken()
+        if (current != null && current != usedToken) return@withLock current
+        val refreshed =
+            try {
+                refresh()
+                true
+            } catch (e: CancellationException) {
+                throw e
+            } catch (t: Throwable) {
+                Logger.w(throwable = t, tag = TAG) {
+                    "Refresh after a 401 failed; leaving the request to fail"
+                }
+                false
+            }
+        if (refreshed) currentToken() else null
+    }
+
+    private companion object {
+        const val TAG = "ConcurrentRefreshGuard"
+    }
+}
+
+private const val AUTH_PATH = "/auth/v1"
+private const val BEARER_PREFIX = "Bearer "

client/auth/data/impl/src/commonMain/kotlin/com/plusmobileapps/chefmate/auth/data/impl/SupabaseAuthenticationRepository.kt

@@ -22,9 +22,18 @@ import io.github.jan.supabase.functions.functions
 import kotlin.coroutines.CoroutineContext
 import kotlin.io.encoding.Base64
 import kotlin.io.encoding.ExperimentalEncodingApi
+import kotlin.time.Clock
+import kotlin.time.Duration.Companion.minutes
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.launch
 import kotlinx.serialization.json.Json
@@ -37,18 +46,30 @@ import kotlinx.serialization.json.jsonObject
 import kotlinx.serialization.json.jsonPrimitive
 import kotlinx.serialization.json.put
 
+@OptIn(ExperimentalCoroutinesApi::class)
 @Inject
 @SingleIn(AppScope::class)
 @ContributesBinding(AppScope::class)
 class SupabaseAuthenticationRepository(
     private val supabaseClient: SupabaseClient,
     @Main private val mainContext: CoroutineContext,
 ) : AuthenticationRepository {
-    private val scope = CoroutineScope(mainContext)
+    // SupervisorJob so a throw inside the session collector below can't tear the scope down and
+    // freeze the auth state for the rest of the process' life.
+    private val scope = CoroutineScope(mainContext + SupervisorJob())
 
     private val _state = MutableStateFlow<AuthState>(AuthState.Unauthenticated)
     override val state: StateFlow<AuthState> = _state.asStateFlow()
 
+    private val _authenticatedSessions =
+        MutableSharedFlow<ChefMateUser>(
+            replay = 1,
+            extraBufferCapacity = 1,
+            onBufferOverflow = BufferOverflow.DROP_OLDEST,
+        )
+    override val authenticatedSessions: SharedFlow<ChefMateUser> =
+        _authenticatedSessions.asSharedFlow()
+
     init {
         // Mirror the Supabase session state into our AuthState. The NotAuthenticated branch no
         // longer auto-bootstraps an anonymous session — callers that need a session call
@@ -62,18 +83,34 @@ class SupabaseAuthenticationRepository(
                         val user = sessionStatus.session.user
                         val isAnon = isAnonymousJwt(sessionStatus.session.accessToken)
                         user?.let {
-                            _state.value =
-                                AuthState.Authenticated(it.toChefMateUser(isAnonymous = isAnon))
+                            val chefMateUser = it.toChefMateUser(isAnonymous = isAnon)
+                            _state.value = AuthState.Authenticated(chefMateUser)
+                            // Fires on every refresh, not only the first sign-in — this is the
+                            // signal sync triggers hang off, since [state] dedupes the equal
+                            // Authenticated value a refresh produces.
+                            _authenticatedSessions.tryEmit(chefMateUser)
                         }
                     }
                     is SessionStatus.NotAuthenticated -> {
+                        // Drop the replayed session so a repository constructed after sign-out
+                        // doesn't immediately sync as the signed-out user.
+                        _authenticatedSessions.resetReplayCache()
                         if (_state.value !is AuthState.AwaitingEmailVerification) {
                             _state.value = AuthState.Unauthenticated
                         }
                     }
-                    is SessionStatus.Initializing,
+                    is SessionStatus.Initializing -> {
+                        // Keep current state during initialization
+                    }
                     is SessionStatus.RefreshFailure -> {
-                        // Keep current state during initialization or refresh failures
+                        // Keep the last known state: the SDK retries on its own, and
+                        // [ExpiredTokenRetry] repairs the token on the next 401. Logged
+                        // because it is otherwise invisible — the app keeps looking signed in
+                        // while every request fails on an expired JWT.
+                        Logger.w(tag = TAG) {
+                            "Supabase session refresh failed; keeping the last known auth state. " +
+                                "Requests will fail until the token recovers."
+                        }
                     }
                 }
             }
@@ -92,6 +129,22 @@ class SupabaseAuthenticationRepository(
         }
     }
 
+    override suspend fun refreshSessionIfNeeded(): Boolean {
+        val session = supabaseClient.auth.currentSessionOrNull() ?: return false
+        if (Clock.System.now() < session.expiresAt - EXPIRY_MARGIN) return true
+        return try {
+            // Besides handing back a live token, this re-arms the SDK's auto-refresh job — which
+            // is the thing that actually died when the machine slept through its timer.
+            supabaseClient.auth.refreshCurrentSession()
+            true
+        } catch (e: CancellationException) {
+            throw e
+        } catch (t: Throwable) {
+            Logger.w(throwable = t, tag = TAG) { "Forced session refresh failed" }
+            false
+        }
+    }
+
     override suspend fun signInWithEmailAndPassword(email: String, password: String): Result<Unit> =
         try {
             supabaseClient.auth.signInWith(Email) {
@@ -288,5 +341,11 @@ class SupabaseAuthenticationRepository(
 
     private companion object {
         const val TAG = "SupabaseAuthenticationRepository"
+
+        /**
+         * Refresh this far ahead of expiry so a sync that starts with a barely-valid token doesn't
+         * have it expire mid-run.
+         */
+        val EXPIRY_MARGIN = 5.minutes
     }
 }

client/auth/data/impl/src/commonMain/kotlin/com/plusmobileapps/chefmate/auth/data/impl/SupabaseModule.kt

@@ -8,7 +8,9 @@ import dev.zacsweers.metro.ContributesTo
 import dev.zacsweers.metro.Provides
 import dev.zacsweers.metro.SingleIn
 import io.github.jan.supabase.SupabaseClient
+import io.github.jan.supabase.annotations.SupabaseInternal
 import io.github.jan.supabase.auth.Auth
+import io.github.jan.supabase.auth.auth
 import io.github.jan.supabase.createSupabaseClient
 import io.github.jan.supabase.functions.Functions
 import io.github.jan.supabase.postgrest.Postgrest
@@ -18,6 +20,10 @@ import io.github.jan.supabase.storage.Storage
 @SingleIn(AppScope::class)
 @ContributesTo(AppScope::class)
 interface SupabaseModule {
+    // httpConfig is marked internal by supabase-kt, but it is the only hook that reaches the
+    // shared HttpClient every plugin sends through — which is exactly the layer a 401 retry belongs
+    // at.
+    @OptIn(SupabaseInternal::class)
     @SingleIn(AppScope::class)
     @Provides
     fun provideSupabaseClient(environmentProvider: EnvironmentProvider): SupabaseClient {
@@ -33,12 +39,29 @@ interface SupabaseModule {
                 Environment.PROD,
                 Environment.FAKE -> BuildConfig.SUPABASE_PROD_URL to BuildConfig.SUPABASE_PROD_KEY
             }
+
+        // [ExpiredTokenRetry] has to refresh the session of the very client it is being installed
+        // into, so it reads the client back out of this holder, which is filled the moment
+        // createSupabaseClient returns — always before anything can issue a request through it.
+        var client: SupabaseClient? = null
+        val refreshGuard =
+            ConcurrentRefreshGuard(
+                currentToken = { client?.auth?.currentAccessTokenOrNull() },
+                refresh = { client?.auth?.refreshCurrentSession() },
+            )
+
         return createSupabaseClient(supabaseUrl = url, supabaseKey = key) {
-            install(Auth)
-            install(Postgrest)
-            install(Realtime)
-            install(Storage)
-            install(Functions)
-        }
+                install(Auth)
+                install(Postgrest)
+                install(Realtime)
+                install(Storage)
+                install(Functions)
+                httpConfig {
+                    install(ExpiredTokenRetry) {
+                        refreshToken = { usedToken -> refreshGuard.tokenAfterRefresh(usedToken) }
+                    }
+                }
+            }
+            .also { client = it }
     }
 }

client/auth/data/impl/src/commonTest/kotlin/com/plusmobileapps/chefmate/auth/data/impl/ConcurrentRefreshGuardTest.kt

@@ -0,0 +1,67 @@
+@file:Suppress("FunctionName")
+
+package com.plusmobileapps.chefmate.auth.data.impl
+
+import io.kotest.matchers.shouldBe
+import kotlin.test.Test
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.test.runTest
+
+class ConcurrentRefreshGuardTest {
+
+    private var token: String? = "stale"
+    private var refreshCalls = 0
+
+    private val guard =
+        ConcurrentRefreshGuard(
+            currentToken = { token },
+            refresh = {
+                refreshCalls += 1
+                // A real refresh is a network round trip. Suspending here is what makes the
+                // burst test meaningful: without the lock, every waiting caller would sail past
+                // the "has someone already refreshed?" check while this one is still in flight.
+                delay(10)
+                token = "fresh-$refreshCalls"
+            },
+        )
+
+    @Test
+    fun refreshes_when_the_caller_used_the_current_token() = runTest {
+        guard.tokenAfterRefresh(usedToken = "stale") shouldBe "fresh-1"
+
+        refreshCalls shouldBe 1
+    }
+
+    @Test
+    fun reuses_a_refresh_another_caller_already_did() = runTest {
+        token = "already-fresh"
+
+        guard.tokenAfterRefresh(usedToken = "stale") shouldBe "already-fresh"
+
+        refreshCalls shouldBe 0
+    }
+
+    @Test
+    fun a_burst_of_401s_on_the_same_token_causes_one_refresh() = runTest {
+        // Supabase rotates the refresh token on every use, so parallel refreshes read as replay
+        // and can invalidate the session — the sign-in this code exists to rescue.
+        val results = List(8) { async { guard.tokenAfterRefresh(usedToken = "stale") } }.awaitAll()
+
+        refreshCalls shouldBe 1
+        results.distinct() shouldBe listOf("fresh-1")
+    }
+
+    @Test
+    fun returns_null_when_the_refresh_fails() = runTest {
+        val failing =
+            ConcurrentRefreshGuard(
+                currentToken = { token },
+                refresh = { throw RuntimeException("offline") },
+            )
+
+        // Not the stale token: handing that back would only buy a second 401.
+        failing.tokenAfterRefresh(usedToken = "stale") shouldBe null
+    }
+}