Messaging Models & Ordering

Copy code@Entity(tableName = "messages")
data class MessageEntity(
    @PrimaryKey val id: String,           // server-assigned UUID
    val conversationId: String,
    val senderId: String,
    val body: String,
    val status: MessageStatus,
    val createdAt: Long,                  // server timestamp (for ordering)
    val localCreatedAt: Long,             // client timestamp (for optimistic ordering)
    val isOptimistic: Boolean = false,    // true = not yet confirmed by server
    val localId: String? = null           // temp ID for matching optimistic → server message
)

enum class MessageStatus {
    SENDING,    // optimistic — sent to server, waiting for ack
    SENT,       // server received
    DELIVERED,  // delivered to recipient's device
    READ,       // recipient opened
    FAILED      // send failed; can retry
}

Copy code@Dao
interface MessageDao {
    // Order by server timestamp for confirmed messages; local timestamp for optimistic
    @Query("""
        SELECT * FROM messages 
        WHERE conversation_id = :conversationId 
        ORDER BY 
            CASE WHEN is_optimistic = 1 THEN local_created_at ELSE created_at END ASC
    """)
    fun getMessages(conversationId: String): Flow<List<MessageEntity>>
}

Copy codeclass MessageRepository(
    private val api: MessagingApi,
    private val dao: MessageDao
) {
    suspend fun sendMessage(conversationId: String, body: String) {
        val localId = UUID.randomUUID().toString()

        // Step 1: Insert optimistic message immediately
        val optimistic = MessageEntity(
            id = localId,
            conversationId = conversationId,
            senderId = currentUserId,
            body = body,
            status = MessageStatus.SENDING,
            createdAt = 0,                          // unknown yet
            localCreatedAt = System.currentTimeMillis(),
            isOptimistic = true,
            localId = localId
        )
        dao.insert(optimistic)

        // Step 2: Send to server
        try {
            val response = api.sendMessage(conversationId, body, localId)

            // Step 3a: Replace optimistic message with server-confirmed version
            dao.delete(localId)
            dao.insert(
                MessageEntity(
                    id = response.messageId,
                    conversationId = conversationId,
                    senderId = currentUserId,
                    body = body,
                    status = MessageStatus.SENT,
                    createdAt = response.createdAt,
                    localCreatedAt = optimistic.localCreatedAt,
                    isOptimistic = false,
                    localId = localId
                )
            )
        } catch (e: IOException) {
            // Step 3b: Mark as failed — allow retry
            dao.updateStatus(localId, MessageStatus.FAILED)
        }
    }

    suspend fun retryMessage(localId: String) {
        val message = dao.getByLocalId(localId) ?: return
        dao.updateStatus(localId, MessageStatus.SENDING)
        sendMessage(message.conversationId, message.body)
    }
}

Copy code// Pass localId as idempotency key to the server
// If the same request is retried, server returns the same message ID
data class SendMessageRequest(
    val conversationId: String,
    val body: String,
    val idempotencyKey: String  // = localId; server deduplicates on this
)

Copy codeclass MessageSyncManager(private val socket: WebSocket, private val dao: MessageDao) {

    fun startSync(conversationId: String): Job = coroutineScope.launch {
        socket.events
            .filterIsInstance<SocketEvent.NewMessage>()
            .filter { it.conversationId == conversationId }
            .collect { event ->
                val incoming = event.message

                // Dedup: ignore if we already have this message (from optimistic or prior sync)
                if (dao.exists(incoming.id)) return@collect

                // Check if this matches one of our optimistic messages
                val optimisticMatch = incoming.idempotencyKey?.let { dao.getByLocalId(it) }
                if (optimisticMatch != null) {
                    // Replace optimistic with confirmed server message
                    dao.delete(optimisticMatch.id)
                }

                dao.insert(incoming.toEntity())
            }
    }
}