package com.sptmobile.link

import org.junit.Assert.assertEquals
import org.junit.Test

class SpoolInboxTest {

    /** Field-for-field the `rust/link-proto` SpoolEntry shape, plus an
     *  unknown key to pin forward tolerance. */
    @Test
    fun parses_wire_shape_with_optional_fields_and_unknowns() {
        val entries = SpoolInbox.parse(
            """
            [
              {"msg_id":"m-1","ts_ms":100,"from":"flynn","kind":"msg","body":"hi",
               "json":"{\"origin\":\"agent\"}","future_field":true},
              {"ts_ms":200,"from":"doyle","kind":"user-msg","body":"no id"}
            ]
            """.trimIndent()
        )
        assertEquals(2, entries.size)
        assertEquals("m-1", entries[0].msg_id)
        assertEquals("flynn", entries[0].from)
        assertEquals(null, entries[1].msg_id)
        assertEquals("user-msg", entries[1].kind)
    }

    private fun entry(id: String?, ts: Long = 1) =
        SpoolEntry(msg_id = id, ts_ms = ts, from = "flynn", kind = "msg", body = "b$ts")

    // [unit->REQ-INBOUND-NOTIFS]
    // The host commits spool removal only after its reply is on the wire, so
    // a drop mid-drain re-delivers the same batch — msg-id dedup across
    // accept calls is what keeps that from double-notifying.
    @Test
    fun redelivered_batch_dedups_by_msg_id_across_drains() {
        val inbox = SpoolInbox()
        val batch = listOf(entry("m-1"), entry("m-2", ts = 2))
        assertEquals(batch, inbox.accept(batch))
        assertEquals(emptyList<SpoolEntry>(), inbox.accept(batch))
        assertEquals(listOf(entry("m-3", ts = 3)), inbox.accept(batch + entry("m-3", ts = 3)))
    }

    // Dedup is exact msg-id only — never fuzzy. Rows without a msg-id have
    // no exact axis, so they always pass (a duplicate notification beats a
    // silently dropped message).
    @Test
    fun rows_without_msg_id_always_pass() {
        val inbox = SpoolInbox()
        val row = entry(null)
        assertEquals(listOf(row), inbox.accept(listOf(row)))
        assertEquals(listOf(row), inbox.accept(listOf(row)))
    }

    @Test
    fun seen_set_is_bounded_evicting_oldest_first() {
        val inbox = SpoolInbox(capacity = 2)
        inbox.accept(listOf(entry("m-1"), entry("m-2"), entry("m-3")))
        // m-1 evicted (capacity 2), so it re-passes; m-3 is still seen.
        assertEquals(listOf(entry("m-1")), inbox.accept(listOf(entry("m-1"), entry("m-3"))))
    }
}