Add buffersyncer

libquassel-api/src/main/kotlin/de/justjanne/libquassel/protocol/api/dto/BufferSyncerDto.kt

+/*
+ * libquassel
+ * Copyright (c) 2025 Janne Mareike Koschinski
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public License,
+ * v. 2.0. If a copy of the MPL was not distributed with this file, You can
+ * obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+package de.justjanne.libquassel.protocol.api.dto
+
+import de.justjanne.libquassel.protocol.models.ids.BufferId
+import de.justjanne.libquassel.protocol.models.ids.MsgId
+import de.justjanne.libquassel.protocol.models.types.QtType
+import de.justjanne.libquassel.protocol.models.types.QuasselType
+import de.justjanne.libquassel.protocol.util.collections.pairs
+import de.justjanne.libquassel.protocol.variant.QVariantList
+import de.justjanne.libquassel.protocol.variant.QVariantMap
+import de.justjanne.libquassel.protocol.variant.QVariant_
+import de.justjanne.libquassel.protocol.variant.into
+import de.justjanne.libquassel.protocol.variant.qVariant
+
+private typealias BufferMap = Map<BufferId, QVariant_>
+
+private fun BufferMap.toVariantList(): QVariantList = entries.flatMap {
+  listOf(qVariant(it.key, QuasselType.BufferId), it.value)
+}
+
+private fun QVariantList.toBufferMap(): BufferMap = pairs { key, value ->
+  Pair(key.into<BufferId>() ?: return@pairs null, value)
+}.filterNotNull().toMap()
+
+data class BufferSyncerDto(
+  val activities: Map<BufferId, Int>,
+  val highlightCounts: Map<BufferId, Int>,
+  val lastSeenMsg: Map<BufferId, MsgId>,
+  val markerLines: Map<BufferId, MsgId>,
+) {
+
+  fun serialize(): QVariantMap = mapOf(
+    "Activities" to qVariant(
+      activities.flatMap {
+        listOf(
+          qVariant(it.key, QuasselType.BufferId),
+          qVariant(it.value, QtType.Int),
+        )
+      },
+      QtType.QVariantList
+    ),
+    "HighlightCounts" to qVariant(
+      highlightCounts.flatMap {
+        listOf(
+          qVariant(it.key, QuasselType.BufferId),
+          qVariant(it.value, QtType.Int),
+        )
+      },
+      QtType.QVariantList
+    ),
+    "LastSeenMsg" to qVariant(
+      lastSeenMsg.flatMap {
+        listOf(
+          qVariant(it.key, QuasselType.BufferId),
+          qVariant(it.value, QuasselType.MsgId),
+        )
+      },
+      QtType.QVariantList
+    ),
+    "MarkerLines" to qVariant(
+      markerLines.flatMap {
+        listOf(
+          qVariant(it.key, QuasselType.BufferId),
+          qVariant(it.value, QuasselType.MsgId),
+        )
+      },
+      QtType.QVariantList
+    ),
+  )
+
+  companion object {
+    fun deserialize(data: QVariantMap) = BufferSyncerDto(
+      activities = data["Activities"].into<QVariantList>().orEmpty().pairs { key, value ->
+        Pair(
+          key.into<BufferId>() ?: return@pairs null,
+          value.into<Int>() ?: return@pairs null
+        )
+      }.filterNotNull().toMap(),
+      highlightCounts = data["HighlightCounts"].into<QVariantList>().orEmpty().pairs { key, value ->
+        Pair(
+          key.into<BufferId>() ?: return@pairs null,
+          value.into<Int>() ?: return@pairs null
+        )
+      }.filterNotNull().toMap(),
+      lastSeenMsg = data["LastSeenMsg"].into<QVariantList>().orEmpty().pairs { key, value ->
+        Pair(
+          key.into<BufferId>() ?: return@pairs null,
+          value.into<MsgId>() ?: return@pairs null
+        )
+      }.filterNotNull().toMap(),
+      markerLines = data["MarkerLines"].into<QVariantList>().orEmpty().pairs { key, value ->
+        Pair(
+          key.into<BufferId>() ?: return@pairs null,
+          value.into<MsgId>() ?: return@pairs null
+        )
+      }.filterNotNull().toMap(),
+    )
+  }
+}

libquassel-api/src/main/kotlin/de/justjanne/libquassel/protocol/api/proxy/Proxy.kt

@@ -14,11 +14,11 @@ import de.justjanne.libquassel.protocol.variant.QVariantList
 import de.justjanne.libquassel.protocol.variant.QVariant_
 
 interface Proxy {
-  fun sync(className: String, objectName: ObjectName, function: String, params: QVariantList)
-  fun sync(className: String, objectName: ObjectName, function: String, vararg arg: QVariant_) =
+  suspend fun sync(className: String, objectName: ObjectName, function: String, params: QVariantList)
+  suspend fun sync(className: String, objectName: ObjectName, function: String, vararg arg: QVariant_) =
     sync(className, objectName, function, arg.toList())
 
-  fun rpc(function: String, params: QVariantList)
-  fun rpc(function: String, vararg arg: QVariant_) =
+  suspend fun rpc(function: String, params: QVariantList)
+  suspend fun rpc(function: String, vararg arg: QVariant_) =
     rpc(function, arg.toList())
 }

libquassel-client/src/main/kotlin/de/justjanne/libquassel/backend/BufferSyncerPersister.kt

@@ -9,20 +9,52 @@
 
 package de.justjanne.libquassel.backend
 
+import de.justjanne.libquassel.persistence.BufferRepository
 import de.justjanne.libquassel.protocol.api.client.BufferSyncerClientApi
+import de.justjanne.libquassel.protocol.api.dto.BufferSyncerDto
 import de.justjanne.libquassel.protocol.models.ids.BufferId
 import de.justjanne.libquassel.protocol.models.ids.MsgId
 import de.justjanne.libquassel.protocol.variant.QVariantMap
 import javax.inject.Inject
 
-class BufferSyncerPersister @Inject constructor() : BufferSyncerClientApi {
+class BufferSyncerPersister @Inject constructor(
+  private val repository: BufferRepository,
+) : BufferSyncerClientApi {
+  override suspend fun update(properties: QVariantMap) {
+    val data = BufferSyncerDto.deserialize(properties)
+    repository.syncActivites(data.activities)
+    repository.syncHighlightCounts(data.highlightCounts)
+    repository.syncLastSeenMsgs(data.lastSeenMsg)
+    repository.syncMarkerLines(data.markerLines)
+  }
+
   override suspend fun markBufferAsRead(buffer: BufferId)  = Unit
-  override suspend fun mergeBuffersPermanently(buffer: BufferId, buffer2: BufferId)  = Unit
-  override suspend fun removeBuffer(buffer: BufferId)  = Unit
-  override suspend fun renameBuffer(buffer: BufferId, newName: String)  = Unit
-  override suspend fun setMarkerLine(buffer: BufferId, msgId: MsgId)  = Unit
-  override suspend fun setLastSeenMsg(buffer: BufferId, msgId: MsgId)  = Unit
-  override suspend fun setBufferActivity(buffer: BufferId, types: Int)  = Unit
-  override suspend fun setHighlightCount(buffer: BufferId, count: Int)  = Unit
-  override suspend fun update(properties: QVariantMap)  = Unit
+
+  override suspend fun mergeBuffersPermanently(buffer: BufferId, buffer2: BufferId) {
+    repository.delete(buffer2)
+  }
+
+  override suspend fun removeBuffer(buffer: BufferId) {
+    repository.delete(buffer)
+  }
+
+  override suspend fun renameBuffer(buffer: BufferId, newName: String) {
+    repository.rename(buffer, newName)
+  }
+
+  override suspend fun setMarkerLine(buffer: BufferId, msgId: MsgId) {
+    repository.setMarkerLine(buffer, msgId)
+  }
+
+  override suspend fun setLastSeenMsg(buffer: BufferId, msgId: MsgId) {
+    repository.setLastSeenMsg(buffer, msgId)
+  }
+
+  override suspend fun setBufferActivity(buffer: BufferId, types: Int) {
+    repository.setBufferActivity(buffer, types)
+  }
+
+  override suspend fun setHighlightCount(buffer: BufferId, count: Int) {
+    repository.setHighlightCount(buffer, count)
+  }
 }

libquassel-client/src/main/kotlin/de/justjanne/libquassel/connection/ClientSessionHandler.kt

@@ -9,6 +9,7 @@
 
 package de.justjanne.libquassel.connection
 
+import de.justjanne.libquassel.persistence.BufferRepository
 import de.justjanne.libquassel.protocol.api.ObjectName
 import de.justjanne.libquassel.protocol.api.dispatcher.RpcDispatcher
 import de.justjanne.libquassel.protocol.api.dispatcher.SyncHandler
@@ -23,21 +24,32 @@ class ClientSessionHandler(
   private val handshake: HandshakeState,
   private val sync: SyncHandler,
   private val rpc: RpcDispatcher,
+  private val bufferRepository: BufferRepository,
 ) : ConnectionHandler<Unit> {
   val toInit = MutableStateFlow<List<Pair<String, String>>?>(null)
 
   override suspend fun handle(connection: Connection) = runCatching {
     suspend fun Connection.send(message: SignalProxyMessage) = send(true) {
+      println("Send: $message")
       SignalProxyMessageSerializer.serialize(it, message, handshake.clientInitAck.featureSet)
     }
 
-    toInit.value = listOf(
-      Pair("AliasManager", "")
+    val initRequests = listOf(
+      Pair("AliasManager", ""),
+      Pair("BufferSyncer", "")
     )
-    connection.send(SignalProxyMessage.InitRequest("AliasManager", ""))
+
+    bufferRepository.sync(handshake.sessionInit.bufferInfos)
+
+    toInit.value = initRequests
+    for ((className, objectName) in initRequests) {
+      connection.send(SignalProxyMessage.InitRequest(className, objectName))
+    }
 
     while (true) {
-      when (val message = connection.read { SignalProxyMessageSerializer.deserialize(it, handshake.clientInitAck.featureSet) }) {
+      val message = connection.read { SignalProxyMessageSerializer.deserialize(it, handshake.clientInitAck.featureSet) }
+      println("Receive: $message")
+      when (message) {
         is SignalProxyMessage.HeartBeat -> connection.send(SignalProxyMessage.HeartBeatReply(message.timestamp))
         is SignalProxyMessage.HeartBeatReply -> Unit
         is SignalProxyMessage.InitData -> {

libquassel-client/src/test/kotlin/de/justjanne/libquassel/client/QuasselApiTest.kt

@@ -12,6 +12,7 @@ package de.justjanne.libquassel.client
 import androidx.room.Room
 import androidx.sqlite.driver.bundled.BundledSQLiteDriver
 import dagger.Binds
+import dagger.BindsInstance
 import dagger.Component
 import dagger.Module
 import dagger.Provides
@@ -35,6 +36,7 @@ import de.justjanne.libquassel.backend.RpcPersister
 import de.justjanne.libquassel.connection.ChannelConnection
 import de.justjanne.libquassel.connection.ClientHandshakeHandler
 import de.justjanne.libquassel.connection.ClientSessionHandler
+import de.justjanne.libquassel.connection.Connection
 import de.justjanne.libquassel.connection.MagicHandler
 import de.justjanne.libquassel.persistence.AliasEntity
 import de.justjanne.libquassel.persistence.AppDatabase
@@ -72,6 +74,7 @@ import de.justjanne.libquassel.protocol.api.server.IgnoreListManagerServerApi
 import de.justjanne.libquassel.protocol.api.server.IrcListHelperServerApi
 import de.justjanne.libquassel.protocol.api.server.NetworkConfigServerApi
 import de.justjanne.libquassel.protocol.api.server.NetworkServerApi
+import de.justjanne.libquassel.protocol.api.server.RpcServerApi
 import de.justjanne.libquassel.protocol.connection.ClientHeader
 import de.justjanne.libquassel.protocol.connection.ProtocolFeature
 import de.justjanne.libquassel.protocol.connection.ProtocolMeta
@@ -79,10 +82,13 @@ import de.justjanne.libquassel.protocol.connection.ProtocolVersion
 import de.justjanne.libquassel.protocol.exceptions.RpcInvocationFailedException
 import de.justjanne.libquassel.protocol.features.FeatureSet
 import de.justjanne.libquassel.protocol.io.CoroutineChannel
+import de.justjanne.libquassel.protocol.models.BufferInfo
 import de.justjanne.libquassel.protocol.models.HandshakeMessage
+import de.justjanne.libquassel.protocol.models.SignalProxyMessage
 import de.justjanne.libquassel.protocol.models.ids.BufferId
 import de.justjanne.libquassel.protocol.models.ids.MsgId
 import de.justjanne.libquassel.protocol.models.types.QtType
+import de.justjanne.libquassel.protocol.serializers.SignalProxyMessageSerializer
 import de.justjanne.libquassel.protocol.variant.QVariantList
 import de.justjanne.libquassel.protocol.variant.QVariant_
 import de.justjanne.libquassel.protocol.variant.qVariant
@@ -108,13 +114,26 @@ import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds
 
 @Singleton
-class ProxyImpl @Inject constructor() : Proxy {
-  override fun sync(className: String, objectName: ObjectName, function: String, params: QVariantList) {
-    println("making sync call: $className $objectName $function $params")
+class ProxyImpl @Inject constructor(
+  private val connection: Connection,
+  private val featureSet: FeatureSet,
+) : Proxy {
+  override suspend fun sync(className: String, objectName: ObjectName, function: String, params: QVariantList) {
+    val message = SignalProxyMessage.Sync(className, objectName.objectName, function, params)
+    println("Send: $message")
+    connection.send(true) {
+      SignalProxyMessageSerializer.serialize(it, message, featureSet)
+    }
+    println("Sent: $message")
   }
 
-  override fun rpc(function: String, params: QVariantList) {
-    println("making rpc call: $function $params")
+  override suspend fun rpc(function: String, params: QVariantList) {
+    val message = SignalProxyMessage.Rpc(function, params)
+    println("Send: $message")
+    connection.send(true) {
+      SignalProxyMessageSerializer.serialize(it, message, featureSet)
+    }
+    println("Sent: $message")
   }
 }
 
@@ -152,6 +171,8 @@ class DatabaseModule {
   @Provides
   fun provideAliasRepository(database: AppDatabase) = database.alias()
   @Provides
+  fun provideBufferRepository(database: AppDatabase) = database.buffer()
+  @Provides
   fun provideMessageRepository(database: AppDatabase) = database.message()
 }
 
@@ -169,6 +190,7 @@ class QuasselApiClient @Inject constructor(
   val ircListHelper: IrcListHelperServerApi,
   val network: NetworkServerApi,
   val networkConfig: NetworkConfigServerApi,
+  val rpc: RpcServerApi,
 )
 
 @Singleton
@@ -178,6 +200,13 @@ interface ClientComponent {
   fun sync(): SyncHandler
   fun rpc(): RpcDispatcher
   fun api(): QuasselApiClient
+
+  @Component.Builder
+  interface Builder {
+    fun build(): ClientComponent
+    @BindsInstance fun connection(connection: Connection): Builder
+    @BindsInstance fun featureSet(featureSet: FeatureSet): Builder
+  }
 }
 
 class QuasselApiTest {
@@ -244,7 +273,6 @@ class QuasselApiTest {
 
   @Test
   fun testConnection() = runBlocking {
-    val di = DaggerClientComponent.builder().build()
     val channel = CoroutineChannel()
     withTimeout(500.milliseconds) {
       channel.connect(InetSocketAddress("decentralised.chat", 4242), keepAlive = true)
@@ -264,15 +292,23 @@ class QuasselApiTest {
         HandshakeMessage.ClientLogin(Const.user, Const.pass),
       ).handle(connection)
     }.getOrThrow()
-    val sessionHandler = ClientSessionHandler(handshake, di.sync(), di.rpc())
-    withTimeout(2.seconds) {
+    val di = DaggerClientComponent.builder()
+      .connection(connection)
+      .featureSet(handshake.clientInitAck.featureSet)
+      .build()
+    val sessionHandler = ClientSessionHandler(handshake, di.sync(), di.rpc(), di.db().buffer())
     launch {
       sessionHandler.handle(connection)
     }
+    withTimeout(2.seconds) {
       sessionHandler.toInit.first { it != null && it.isEmpty() }
-      channel.close()
     }
+    val bufferInfo = handshake.sessionInit.bufferInfos.first { it.bufferName == "#quassel-test" }
+    di.api().rpc.sendInput(bufferInfo, "/SAY Test from libquassel?")
+    delay(15.seconds)
+    channel.close()
     println(di.db().alias().getAll())
+    println(di.db().buffer().getAll())
     Unit
   }
 }

libquassel-persistence/src/main/kotlin/de/justjanne/libquassel/persistence/AliasRepository.kt

@@ -14,7 +14,6 @@ import androidx.room.Insert
 import androidx.room.OnConflictStrategy
 import androidx.room.Query
 import androidx.room.Transaction
-import de.justjanne.libquassel.protocol.models.ids.BufferId
 import kotlinx.coroutines.flow.Flow
 
 @Dao
@@ -39,36 +38,3 @@ interface AliasRepository {
     }
   }
 }
-
-@Dao
-interface MessageRepository {
-  @Query("SELECT * FROM MessageEntity WHERE buffer = :buffer ORDER BY messageId ASC")
-  suspend fun getAll(buffer: BufferId): List<MessageEntity>
-
-  @Query("SELECT * FROM MessageEntity WHERE buffer = :buffer ORDER BY messageId ASC")
-  fun collectAll(buffer: BufferId): Flow<List<MessageEntity>>
-
-  @Query("SELECT * FROM messageentity WHERE buffer = :buffer ORDER BY messageId DESC LIMIT 1")
-  suspend fun getLast(buffer: BufferId): MessageEntity?
-
-  @Query("SELECT * FROM messageentity WHERE buffer = :buffer ORDER BY messageId DESC LIMIT 1")
-  suspend fun collectLast(buffer: BufferId): MessageEntity?
-
-  @Insert(onConflict = OnConflictStrategy.REPLACE)
-  suspend fun insert(vararg item: MessageEntity)
-
-  @Query("DELETE FROM MessageEntity WHERE buffer = :buffer")
-  suspend fun clear(buffer: BufferId)
-
-  @Transaction
-  suspend fun sync(buffer: BufferId, data: List<MessageEntity>) {
-    val newMessages = data.map(MessageEntity::messageId).toSet()
-    val last = getLast(buffer)
-    if (last != null && !newMessages.contains(last.messageId)) {
-      clear(buffer)
-    }
-    for (item in data) {
-      insert(item)
-    }
-  }
-}

libquassel-persistence/src/main/kotlin/de/justjanne/libquassel/persistence/AppDatabase.kt

@@ -16,6 +16,7 @@ import androidx.room.TypeConverters
 @Database(
   entities = [
     AliasEntity::class,
+    BufferEntity::class,
     MessageEntity::class,
   ],
   version = 1,
@@ -23,5 +24,6 @@ import androidx.room.TypeConverters
 @TypeConverters(InstantConverter::class)
 abstract class AppDatabase : RoomDatabase() {
   abstract fun alias(): AliasRepository
+  abstract fun buffer(): BufferRepository
   abstract fun message(): MessageRepository
 }

libquassel-persistence/src/main/kotlin/de/justjanne/libquassel/persistence/BufferEntity.kt

+/*
+ * libquassel
+ * Copyright (c) 2025 Janne Mareike Koschinski
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public License,
+ * v. 2.0. If a copy of the MPL was not distributed with this file, You can
+ * obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+package de.justjanne.libquassel.persistence
+
+import androidx.room.Entity
+import androidx.room.PrimaryKey
+import de.justjanne.libquassel.protocol.models.ids.BufferId
+import de.justjanne.libquassel.protocol.models.ids.MsgId
+import de.justjanne.libquassel.protocol.models.ids.NetworkId
+
+@Entity
+data class BufferEntity(
+  @PrimaryKey
+  val id: BufferId,
+  val network: NetworkId,
+  val type: Int,
+  val group: Int,
+  val name: String?,
+  val activity: Int?,
+  val highlightCount: Int?,
+  val lastSeenMsg: MsgId?,
+  val markerLine: MsgId?
+)

libquassel-persistence/src/main/kotlin/de/justjanne/libquassel/persistence/BufferRepository.kt

+/*
+ * libquassel
+ * Copyright (c) 2025 Janne Mareike Koschinski
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public License,
+ * v. 2.0. If a copy of the MPL was not distributed with this file, You can
+ * obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+package de.justjanne.libquassel.persistence
+
+import androidx.room.Dao
+import androidx.room.Insert
+import androidx.room.OnConflictStrategy
+import androidx.room.Query
+import androidx.room.Transaction
+import de.justjanne.bitflags.toBits
+import de.justjanne.libquassel.protocol.models.BufferInfo
+import de.justjanne.libquassel.protocol.models.ids.BufferId
+import de.justjanne.libquassel.protocol.models.ids.MsgId
+import kotlinx.coroutines.flow.Flow
+
+@Dao
+interface BufferRepository {
+  @Query("SELECT * FROM BufferEntity")
+  suspend fun getAll(): List<BufferEntity>
+
+  @Query("SELECT * FROM BufferEntity")
+  fun collectAll(): Flow<List<BufferEntity>>
+
+  @Insert(onConflict = OnConflictStrategy.REPLACE)
+  suspend fun insert(vararg item: BufferEntity)
+
+  @Query("UPDATE BufferEntity SET markerLine = :msgId WHERE id = :bufferId")
+  suspend fun setMarkerLine(bufferId: BufferId, msgId: MsgId)
+
+  @Query("UPDATE BufferEntity SET lastSeenMsg = :msgId WHERE id = :bufferId")
+  suspend fun setLastSeenMsg(bufferId: BufferId, msgId: MsgId)
+
+  @Query("UPDATE BufferEntity SET activity = :activity WHERE id = :bufferId")
+  suspend fun setBufferActivity(bufferId: BufferId, activity: Int)
+
+  @Query("UPDATE BufferEntity SET highlightCount = :count WHERE id = :bufferId")
+  suspend fun setHighlightCount(bufferId: BufferId, count: Int)
+
+  @Query("UPDATE BufferEntity SET name = :name WHERE id = :bufferId")
+  suspend fun rename(bufferId: BufferId, name: String?)
+
+  @Query("DELETE FROM BufferEntity WHERE id = :bufferId")
+  suspend fun delete(bufferId: BufferId)
+
+  @Query("DELETE FROM BufferEntity")
+  suspend fun clear()
+
+  @Transaction
+  suspend fun sync(data: List<BufferInfo>) {
+    clear()
+    for (item in data) {
+      insert(BufferEntity(
+        id = item.bufferId,
+        network = item.networkId,
+        type = item.type.toBits().toInt(),
+        group = item.groupId,
+        name = item.bufferName,
+        activity = null,
+        highlightCount = null,
+        lastSeenMsg = null,
+        markerLine = null
+      ))
+    }
+  }
+
+  @Transaction
+  suspend fun syncActivites(data: Map<BufferId, Int>) {
+    for (item in data) {
+      setBufferActivity(item.key, item.value)
+    }
+  }
+
+  @Transaction
+  suspend fun syncHighlightCounts(data: Map<BufferId, Int>) {
+    for (item in data) {
+      setHighlightCount(item.key, item.value)
+    }
+  }
+
+  @Transaction
+  suspend fun syncLastSeenMsgs(data: Map<BufferId, MsgId>) {
+    for (item in data) {
+      setLastSeenMsg(item.key, item.value)
+    }
+  }
+
+  @Transaction
+  suspend fun syncMarkerLines(data: Map<BufferId, MsgId>) {
+    for (item in data) {
+      setMarkerLine(item.key, item.value)
+    }
+  }
+}

libquassel-persistence/src/main/kotlin/de/justjanne/libquassel/persistence/MessageRepository.kt

+/*
+ * libquassel
+ * Copyright (c) 2025 Janne Mareike Koschinski
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public License,
+ * v. 2.0. If a copy of the MPL was not distributed with this file, You can
+ * obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+package de.justjanne.libquassel.persistence
+
+import androidx.room.Dao
+import androidx.room.Insert
+import androidx.room.OnConflictStrategy
+import androidx.room.Query
+import androidx.room.Transaction
+import de.justjanne.libquassel.protocol.models.ids.BufferId
+import kotlinx.coroutines.flow.Flow
+
+@Dao
+interface MessageRepository {
+  @Query("SELECT * FROM MessageEntity WHERE buffer = :buffer ORDER BY messageId ASC")
+  suspend fun getAll(buffer: BufferId): List<MessageEntity>
+
+  @Query("SELECT * FROM MessageEntity WHERE buffer = :buffer ORDER BY messageId ASC")
+  fun collectAll(buffer: BufferId): Flow<List<MessageEntity>>
+
+  @Query("SELECT * FROM messageentity WHERE buffer = :buffer ORDER BY messageId DESC LIMIT 1")
+  suspend fun getLast(buffer: BufferId): MessageEntity?
+
+  @Query("SELECT * FROM messageentity WHERE buffer = :buffer ORDER BY messageId DESC LIMIT 1")
+  suspend fun collectLast(buffer: BufferId): MessageEntity?
+
+  @Insert(onConflict = OnConflictStrategy.REPLACE)
+  suspend fun insert(vararg item: MessageEntity)
+
+  @Query("DELETE FROM MessageEntity WHERE buffer = :buffer")
+  suspend fun clear(buffer: BufferId)
+
+  @Transaction
+  suspend fun sync(buffer: BufferId, data: List<MessageEntity>) {
+    val newMessages = data.map(MessageEntity::messageId).toSet()
+    val last = getLast(buffer)
+    if (last != null && !newMessages.contains(last.messageId)) {
+      clear(buffer)
+    }
+    for (item in data) {
+      insert(item)
+    }
+  }
+}