From e8ca5931643d0791982bee44248f9a489ec34588 Mon Sep 17 00:00:00 2001 From: Pluto Date: Sun, 8 Mar 2026 23:17:33 -0500 Subject: [PATCH] Add missing FCM support classes - ClippyResponseBus: SharedFlow event bus for FCM + sync responses - ResponseDedup: SHA-256 dedup with sliding window - Updated ClippyFCMService and FCMTokenManager to use MatrixRestClient - Added putAccountData() to MatrixRestClient for FCM token storage Co-authored-by: Spirit --- .../clippyauto/matrix/MatrixRestClient.kt | 21 ++++++++ .../hobbs/clippyauto/push/ClippyFCMService.kt | 38 ++----------- .../clippyauto/push/ClippyResponseBus.kt | 40 ++++++++++++++ .../hobbs/clippyauto/push/FCMTokenManager.kt | 31 +++++------ .../hobbs/clippyauto/push/ResponseDedup.kt | 53 +++++++++++++++++++ 5 files changed, 130 insertions(+), 53 deletions(-) create mode 100644 app/src/main/java/farm/hobbs/clippyauto/push/ClippyResponseBus.kt create mode 100644 app/src/main/java/farm/hobbs/clippyauto/push/ResponseDedup.kt diff --git a/app/src/main/java/farm/hobbs/clippyauto/matrix/MatrixRestClient.kt b/app/src/main/java/farm/hobbs/clippyauto/matrix/MatrixRestClient.kt index fe9d179..2f5e2d1 100644 --- a/app/src/main/java/farm/hobbs/clippyauto/matrix/MatrixRestClient.kt +++ b/app/src/main/java/farm/hobbs/clippyauto/matrix/MatrixRestClient.kt @@ -164,6 +164,27 @@ class MatrixRestClient( } } + + /** + * Store account data in Matrix. + * Used for FCM token registration so n8n can read it. + */ + suspend fun putAccountData(userId: String, type: String, data: Map): Boolean { + return try { + val response = client.put("$homeserverUrl/_matrix/client/v3/user/$userId/account_data/$type") { + header("Authorization", "Bearer $accessToken") + contentType(ContentType.Application.Json) + setBody(buildJsonObject { + data.forEach { (key, value) -> put(key, value) } + }.toString()) + } + response.status.isSuccess() + } catch (e: Exception) { + Log.e("MatrixREST", "putAccountData failed", e) + false + } + } + fun shutdown() { httpClient.close() } diff --git a/app/src/main/java/farm/hobbs/clippyauto/push/ClippyFCMService.kt b/app/src/main/java/farm/hobbs/clippyauto/push/ClippyFCMService.kt index 567911f..301070f 100644 --- a/app/src/main/java/farm/hobbs/clippyauto/push/ClippyFCMService.kt +++ b/app/src/main/java/farm/hobbs/clippyauto/push/ClippyFCMService.kt @@ -8,10 +8,7 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.launch -import io.ktor.client.* -import io.ktor.client.engine.cio.* -import io.ktor.client.request.* -import io.ktor.http.* +import farm.hobbs.clippyauto.ClippyAutoApp /** * Firebase Cloud Messaging service for ClippyAuto. @@ -85,40 +82,11 @@ class ClippyFCMService : FirebaseMessagingService() { /** * Register the FCM token with Matrix via account data. - * - * Stores at: PUT /_matrix/client/v3/user/{userId}/account_data/{type} - * n8n reads this to know where to send FCM pushes. + * Delegates to FCMTokenManager to avoid duplicate Ktor client setup. */ private fun registerTokenWithMatrix(token: String) { scope.launch { - try { - val accessToken = AppConfig.matrixAccessToken ?: run { - Log.w(TAG, "No Matrix access token — can't register FCM token") - return@launch - } - - val userId = "@${AppConfig.matrixUser}:hobbs.farm" - val client = HttpClient(CIO) - - val response = client.put( - "${AppConfig.homeserverUrl}/_matrix/client/v3/user/" + - "${userId}/account_data/${ACCOUNT_DATA_TYPE}" - ) { - header("Authorization", "Bearer $accessToken") - contentType(ContentType.Application.Json) - setBody("""{"fcm_token":"$token","updated_at":"${System.currentTimeMillis()}"}""") - } - - if (response.status.isSuccess()) { - Log.d(TAG, "FCM token registered in Matrix account data") - } else { - Log.e(TAG, "Failed to register FCM token: ${response.status}") - } - - client.close() - } catch (e: Exception) { - Log.e(TAG, "Error registering FCM token", e) - } + FCMTokenManager.registerToken(token) } } } diff --git a/app/src/main/java/farm/hobbs/clippyauto/push/ClippyResponseBus.kt b/app/src/main/java/farm/hobbs/clippyauto/push/ClippyResponseBus.kt new file mode 100644 index 0000000..9ba08aa --- /dev/null +++ b/app/src/main/java/farm/hobbs/clippyauto/push/ClippyResponseBus.kt @@ -0,0 +1,40 @@ +package farm.hobbs.clippyauto.push + +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.coroutines.flow.asSharedFlow + +/** + * Singleton event bus for Clippy responses. + * + * Both delivery paths (FCM fast path + Matrix sync) emit here. + * VoiceLoop collects from [responses] and routes to TTS. + * ResponseDedup ensures only the first delivery of each message is processed. + * + * Flow: + * ClippyFCMService.onMessageReceived() ──emit──┐ + * ├──► VoiceLoop.onClippyResponse() + * MatrixRestClient.sync() ──────────emit──────┘ + */ +object ClippyResponseBus { + + data class Response( + val text: String, + val path: String, // "fcm" or "sync" + val timestamp: Long = System.currentTimeMillis() + ) + + private val _responses = MutableSharedFlow( + replay = 0, + extraBufferCapacity = 16 + ) + val responses: SharedFlow = _responses.asSharedFlow() + + /** + * Emit a Clippy response from either delivery path. + * Dedup is handled by the consumer (VoiceLoop) via ResponseDedup. + */ + fun emit(text: String, path: String) { + _responses.tryEmit(Response(text, path)) + } +} diff --git a/app/src/main/java/farm/hobbs/clippyauto/push/FCMTokenManager.kt b/app/src/main/java/farm/hobbs/clippyauto/push/FCMTokenManager.kt index 25aac57..5dd13cf 100644 --- a/app/src/main/java/farm/hobbs/clippyauto/push/FCMTokenManager.kt +++ b/app/src/main/java/farm/hobbs/clippyauto/push/FCMTokenManager.kt @@ -1,13 +1,9 @@ package farm.hobbs.clippyauto.push -import android.content.Context import android.util.Log import com.google.firebase.messaging.FirebaseMessaging +import farm.hobbs.clippyauto.ClippyAutoApp import farm.hobbs.clippyauto.config.AppConfig -import io.ktor.client.* -import io.ktor.client.engine.cio.* -import io.ktor.client.request.* -import io.ktor.http.* import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.launch @@ -45,6 +41,7 @@ object FCMTokenManager { /** * Register a token with Matrix account data. + * Uses the app's existing MatrixRestClient to avoid extra HTTP dependencies. */ suspend fun registerToken(token: String) { val accessToken = AppConfig.matrixAccessToken ?: run { @@ -53,25 +50,23 @@ object FCMTokenManager { } try { + val matrixClient = ClippyAutoApp.instance.matrixClient val userId = "@${AppConfig.matrixUser}:hobbs.farm" - val client = HttpClient(CIO) - val response = client.put( - "${AppConfig.homeserverUrl}/_matrix/client/v3/user/" + - "${userId}/account_data/${ClippyFCMService.ACCOUNT_DATA_TYPE}" - ) { - header("Authorization", "Bearer $accessToken") - contentType(ContentType.Application.Json) - setBody("""{"fcm_token":"$token","updated_at":"${System.currentTimeMillis()}"}""") - } + val success = matrixClient.putAccountData( + userId = userId, + type = ClippyFCMService.ACCOUNT_DATA_TYPE, + data = mapOf( + "fcm_token" to token, + "updated_at" to System.currentTimeMillis().toString() + ) + ) - if (response.status.isSuccess()) { + if (success) { Log.d(TAG, "FCM token registered successfully") } else { - Log.e(TAG, "Failed to register FCM token: ${response.status}") + Log.e(TAG, "Failed to register FCM token") } - - client.close() } catch (e: Exception) { Log.e(TAG, "Error registering FCM token with Matrix", e) } diff --git a/app/src/main/java/farm/hobbs/clippyauto/push/ResponseDedup.kt b/app/src/main/java/farm/hobbs/clippyauto/push/ResponseDedup.kt new file mode 100644 index 0000000..d4f0724 --- /dev/null +++ b/app/src/main/java/farm/hobbs/clippyauto/push/ResponseDedup.kt @@ -0,0 +1,53 @@ +package farm.hobbs.clippyauto.push + +import java.security.MessageDigest +import java.util.concurrent.ConcurrentLinkedQueue + +/** + * Deduplicates Clippy responses across FCM and sync delivery paths. + * + * Both paths deliver the same message — this ensures only the first + * arrival is processed (routed to TTS). The second is silently dropped. + * + * Uses SHA-256 hash of message text for comparison. + * Keeps a sliding window of recent hashes to bound memory. + */ +object ResponseDedup { + + private const val MAX_HISTORY = 100 + + private val seenHashes = ConcurrentLinkedQueue() + + /** + * Generate a hash for dedup comparison. + */ + fun hash(text: String): String { + val digest = MessageDigest.getInstance("SHA-256") + val bytes = digest.digest(text.toByteArray()) + return bytes.take(8).joinToString("") { "%02x".format(it) } + } + + /** + * Check if this hash has been seen before. + */ + fun seen(hash: String): Boolean { + return seenHashes.contains(hash) + } + + /** + * Mark a hash as seen. Trims history if over MAX_HISTORY. + */ + fun mark(hash: String) { + seenHashes.add(hash) + while (seenHashes.size > MAX_HISTORY) { + seenHashes.poll() + } + } + + /** + * Clear all history (e.g., on new session). + */ + fun clear() { + seenHashes.clear() + } +}