- 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 <spirit@hobbs.farm>
This commit is contained in:
@@ -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<String, String>): 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()
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Response>(
|
||||
replay = 0,
|
||||
extraBufferCapacity = 16
|
||||
)
|
||||
val responses: SharedFlow<Response> = _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))
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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<String>()
|
||||
|
||||
/**
|
||||
* 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()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user