package com.sptmobile.voice

import android.util.Base64
import androidx.test.ext.junit.runners.AndroidJUnit4
import androidx.test.platform.app.InstrumentationRegistry
import com.sptmobile.endpoint.Timeline
import com.sptmobile.link.HostLinkState
import com.sptmobile.link.LinkNative
import com.sptmobile.pairing.PairedHost
import java.net.InetAddress
import java.net.Socket
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.runBlocking
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertNull
import org.junit.Assert.assertTrue
import org.junit.Assume.assumeTrue
import org.junit.Test
import org.junit.runner.RunWith

// [int->REQ-VOICE-PIPE]
/**
 * The voice pipe end-to-end on-device (VOICE-PLAN V3): a Pebble-shaped POST
 * over loopback → real Room spool → [VoiceForwarder] → the stub host's star
 * over real UDP → the star's history carries the minted msg-id. Driven by
 * `ci/emulator-smoke.ps1` like the link smoke; SKIPS without the host args.
 */
@RunWith(AndroidJUnit4::class)
class VoicePipeSmokeTest {
    @Test
    fun webhook_to_spool_to_star_over_real_udp() {
        val args = InstrumentationRegistry.getArguments()
        val addrB64 = args.getString("sptHostAddrB64")
        val token = args.getString("sptToken")
        assumeTrue("no loopback host provided — skipping smoke", addrB64 != null && token != null)

        val context = InstrumentationRegistry.getInstrumentation().targetContext
        context.deleteDatabase(DB) // fresh spool per run
        val spool = RoomVoiceSpool(context, DB)
        val server = WebhookServer(0, WebhookHandler(spool, { "smoke-token" }))
        val port = server.start()
        try {
            // 1. Pebble posts a recording.
            val status = postLikePebble(port, "voice smoke dictation")
            assertEquals(200, status)

            // [int->REQ-HAZARD-DICTATION-LOSS] The 200 has landed; the row is
            // already durable in SQLite — before any forwarding exists.
            val row = spool.oldest()
            assertNotNull("row durable behind the 200", row)
            assertEquals("voice smoke dictation", row!!.transcription)
            assertEquals(1_751_791_234_567L, row.recordedAtMs)

            // 2. The forwarder drains it to the first reachable host's star.
            LinkNative.init(context)
            val addrJson = String(Base64.decode(addrB64, Base64.DEFAULT), Charsets.UTF_8)
            val handle = LinkNative.connectTo(addrJson, token!!)
            try {
                LinkNative.pair(handle, "emulator-smoke-voice")
                val host = PairedHost(
                    node = "stub-host",
                    endpoint = "mobile-gw",
                    token = token,
                    star = "star-1",
                )
                val forwarder = VoiceForwarder(
                    spool = spool,
                    hosts = flowOf(listOf(host)),
                    states = MutableStateFlow(
                        mapOf<String, HostLinkState>(
                            "stub-host" to HostLinkState.Connected(handle),
                        )
                    ),
                )
                runBlocking { forwarder.drainOnce() }

                assertNull("spool drained FIFO-empty", spool.oldest())
                assertTrue(
                    "delivery outcome recorded",
                    forwarder.lastOutcome.value is VoiceOutcome.Delivered,
                )

                // 3. The star's history carries the enqueue-minted msg-id —
                // the voice tag rode --json-payload end to end (ruling 6).
                val history = Timeline.parseHistory(
                    LinkNative.historyFetch(handle, "star-1", 50),
                )
                assertTrue(
                    "star history carries the dictation's msg-id",
                    history.any { it.msg_id == row.msgId },
                )
            } finally {
                LinkNative.close(handle)
            }
        } finally {
            server.stop()
        }
    }

    /**
     * The DESIGN.md §Pebble webhook contract shape, byte for byte — over a
     * raw loopback socket: Pebble is its own HTTP client, and the app's
     * cleartext policy (which blocks HttpURLConnection to 127.0.0.1 in this
     * test process on API 28+) governs neither it nor our inbound serving.
     */
    private fun postLikePebble(port: Int, transcription: String): Int {
        val boundary = "pebble-smoke"
        val body = ("--$boundary\r\n" +
            "Content-Disposition: form-data; name=\"transcription\"\r\n\r\n" +
            "$transcription\r\n" +
            "--$boundary\r\n" +
            "Content-Disposition: form-data; name=\"recordedAt\"\r\n\r\n" +
            "1751791234567\r\n" +
            "--$boundary\r\n" +
            "Content-Disposition: form-data; name=\"client\"\r\n\r\n" +
            "ring\r\n" +
            "--$boundary--\r\n").toByteArray(Charsets.UTF_8)
        Socket(InetAddress.getLoopbackAddress(), port).use { socket ->
            socket.soTimeout = 10_000
            socket.getOutputStream().apply {
                write(
                    ("POST / HTTP/1.1\r\n" +
                        "Host: 127.0.0.1:$port\r\n" +
                        "X-Widget-Token: smoke-token\r\n" +
                        "Content-Type: multipart/form-data; boundary=$boundary\r\n" +
                        "Content-Length: ${body.size}\r\n" +
                        "Connection: close\r\n\r\n").toByteArray(Charsets.ISO_8859_1)
                )
                write(body)
                flush()
            }
            val statusLine = socket.getInputStream().bufferedReader(Charsets.ISO_8859_1)
                .readLine() ?: throw AssertionError("no status line from receiver")
            return statusLine.split(" ")[1].toInt()
        }
    }

    private companion object {
        const val DB = "voice_smoke.db"
    }
}