wip: Migrate to Trixnity
This commit is contained in:
@@ -1,242 +1,4 @@
|
||||
/*
|
||||
* Created by sweetbread
|
||||
* Copyright (c) 2025. All rights reserved.
|
||||
*/
|
||||
|
||||
package ru.risdeveau.pixeldragon.api
|
||||
|
||||
import android.util.Log
|
||||
import io.ktor.client.request.bearerAuth
|
||||
import io.ktor.client.request.get
|
||||
import io.ktor.client.request.header
|
||||
import io.ktor.client.request.parameter
|
||||
import io.ktor.client.request.post
|
||||
import io.ktor.client.request.setBody
|
||||
import io.ktor.client.statement.bodyAsText
|
||||
import io.ktor.http.ContentType
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.http.contentLength
|
||||
import io.ktor.http.contentType
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.flow.asStateFlow
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.launch
|
||||
import org.json.JSONObject
|
||||
import ru.risdeveau.pixeldragon.AccountData
|
||||
import ru.risdeveau.pixeldragon.AccountData.syncLastBatch
|
||||
import ru.risdeveau.pixeldragon.baseUrl
|
||||
import ru.risdeveau.pixeldragon.client
|
||||
import ru.risdeveau.pixeldragon.db.isConnected
|
||||
import ru.risdeveau.pixeldragon.token
|
||||
import ru.risdeveau.pixeldragon.ui.activity.ME
|
||||
import splitties.init.appCtx
|
||||
import java.io.File
|
||||
|
||||
|
||||
class MatrixSyncService {
|
||||
private val _syncState = MutableStateFlow<SyncState>(SyncState.Idle)
|
||||
val syncState: StateFlow<SyncState> = _syncState.asStateFlow()
|
||||
|
||||
private var syncJob: Job? = null
|
||||
// private var isInitialized = false
|
||||
//
|
||||
// fun initialize() {
|
||||
// isInitialized = true
|
||||
// }
|
||||
|
||||
fun startSync() {
|
||||
if (AccountData.token == null)
|
||||
Log.wtf("MatrixSyncService", "Token is null")
|
||||
if (AccountData.userId == null)
|
||||
Log.wtf("MatrixSyncService", "User ID is null")
|
||||
if (AccountData.homeserver == null)
|
||||
Log.wtf("MatrixSyncService", "Homeserver is null")
|
||||
|
||||
Log.i("MatrixSyncService", "Start syncing")
|
||||
// if (!isInitialized)
|
||||
// throw IllegalStateException("Sync service not initialized")
|
||||
|
||||
if (syncJob?.isActive == true) return
|
||||
|
||||
syncJob = CoroutineScope(Dispatchers.IO).launch {
|
||||
if (syncLastBatch == null) {
|
||||
Log.i("MatrixSyncService", "Init syncing")
|
||||
_syncState.value = SyncState.Syncing
|
||||
try {
|
||||
initialSync()
|
||||
_syncState.value = SyncState.Idle
|
||||
} catch (e: Exception) {
|
||||
Log.e("MatrixSyncService", "Initial sync error", e)
|
||||
_syncState.value = SyncState.Error(e.message)
|
||||
}
|
||||
}
|
||||
|
||||
// while (isActive) {
|
||||
// try {
|
||||
// val response = sync (
|
||||
// timeout = 30000,
|
||||
// filter = getFilter()
|
||||
// )
|
||||
//
|
||||
// processSyncResponse(response)
|
||||
//
|
||||
// } catch (e: Exception) {
|
||||
// Log.w("sync", e.message.toString())
|
||||
// delay(5000) // Wait before retry
|
||||
// }
|
||||
// }
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
fun stopSync() {
|
||||
syncJob?.cancel()
|
||||
_syncState.value = SyncState.Idle
|
||||
}
|
||||
|
||||
fun pauseSync() {
|
||||
// Called when app goes to background
|
||||
syncJob?.cancel()
|
||||
}
|
||||
|
||||
fun resumeSync() {
|
||||
// Called when app comes to foreground
|
||||
if (!isSyncActive()) {
|
||||
startSync()
|
||||
}
|
||||
}
|
||||
|
||||
fun isSyncActive(): Boolean = syncJob?.isActive == true
|
||||
|
||||
private suspend fun parseSyncResponse(response: JSONObject) {
|
||||
Log.v("syncResponse", response.toString(2))
|
||||
val newMessages = mutableListOf<Message>()
|
||||
val roomUpdates = mutableListOf<RoomUpdate>()
|
||||
|
||||
response.rooms?.join?.forEach { (roomId, roomData) ->
|
||||
// Process timeline events (messages)
|
||||
roomData.timeline?.events?.forEach { event ->
|
||||
val message = event.toMessage(roomId)
|
||||
database.messageDao().insertMessage(message)
|
||||
newMessages.add(message)
|
||||
}
|
||||
|
||||
// Process room state updates
|
||||
roomData.state?.events?.forEach { event ->
|
||||
when (event.type) {
|
||||
"m.room.name", "m.room.avatar" -> {
|
||||
roomUpdates.add(RoomUpdate(roomId, event.type, event.content))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Process ephemeral events (typing, receipts)
|
||||
roomData.ephemeral?.events?.forEach { event ->
|
||||
when (event.type) {
|
||||
"m.typing" -> handleTypingEvent(roomId, event)
|
||||
"m.receipt" -> handleReceiptEvent(roomId, event)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Notify the app about new data
|
||||
if (newMessages.isNotEmpty()) {
|
||||
_newMessages.emit(newMessages)
|
||||
}
|
||||
if (roomUpdates.isNotEmpty()) {
|
||||
_roomUpdates.emit(roomUpdates)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
sealed class SyncState {
|
||||
object Idle : SyncState()
|
||||
object Syncing : SyncState()
|
||||
object Success : SyncState()
|
||||
data class Error(val message: String?) : SyncState()
|
||||
}
|
||||
|
||||
suspend fun sync(): JSONObject {
|
||||
|
||||
return JSONObject()
|
||||
|
||||
}
|
||||
|
||||
suspend fun initialSync(): JSONObject {
|
||||
fetchRoomMeta()
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch rooms metadata during initial sync
|
||||
*/
|
||||
suspend fun fetchRoomMeta() {
|
||||
val filterId = getFilterId("""
|
||||
{
|
||||
"room": {
|
||||
"state": {
|
||||
"types": [
|
||||
"m.room.name",
|
||||
"m.room.avatar",
|
||||
"m.room.canonical_alias",
|
||||
"m.room.member"
|
||||
],
|
||||
"lazy_load_members": true
|
||||
},
|
||||
"timeline": {
|
||||
"limit": 0,
|
||||
"types": []
|
||||
},
|
||||
"ephemeral": {
|
||||
"types": []
|
||||
},
|
||||
"include_leave": false
|
||||
},
|
||||
"presence": {
|
||||
"types": []
|
||||
}
|
||||
}
|
||||
""".trimIndent())
|
||||
// val filterId = "vmNk"
|
||||
|
||||
val r = client.get("$baseUrl/sync") {
|
||||
bearerAuth(token)
|
||||
url {
|
||||
parameter("filter", filterId)
|
||||
}
|
||||
}
|
||||
|
||||
if (r.status != HttpStatusCode.OK)
|
||||
throw IllegalStateException("Failed to sync")
|
||||
Log.v("initialSync", "Response size: ${r.contentLength()}")
|
||||
|
||||
r.contentLength()?.let {
|
||||
if (it >= 50*1024*1024)
|
||||
Log.w("initialSync", "Response size is too large")
|
||||
}
|
||||
val json = JSONObject(r.bodyAsText())
|
||||
syncLastBatch = json.getString("next_batch")
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a filter to the server and get its id
|
||||
* @param filter JSON Filter
|
||||
* @return Filter ID
|
||||
*/
|
||||
suspend fun getFilterId(filter: String): String {
|
||||
val userId = AccountData.userId
|
||||
if (userId == null)
|
||||
Log.wtf("getFilter", "user_id is not defined")
|
||||
|
||||
val r = client.post("$baseUrl/user/$userId/filter") {
|
||||
setBody(filter)
|
||||
contentType(ContentType.Application.Json)
|
||||
bearerAuth(token)
|
||||
}
|
||||
|
||||
return JSONObject(r.bodyAsText()).getString("filter_id")
|
||||
}
|
||||
* Copyright (c) 2026. All rights reserved.
|
||||
*/
|
||||
Reference in New Issue
Block a user