package de.upapp

import android.util.Log
import java.io.Closeable
import java.io.IOException
import java.io.InputStream
import java.io.OutputStream
import java.net.SocketTimeoutException
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
import kotlin.concurrent.thread

internal const val TRANSMIT = 0
internal const val OPEN = 1
internal const val CLOSE = 2
internal const val HEARTBEAT = 3
private const val MAX_PAYLOAD = 0xFFFF
// Values from maps+more.
private const val HEARTBEAT_MS = 5_000L
private const val SILENCE_MS = 15_000L

internal class Segment(val port: Int, val type: Int, val data: ByteArray)

internal fun frame(port: Int, type: Int, extra: ByteArray = ByteArray(0)) =
    byteArrayOf(port.toByte(), (port shr 8).toByte(), type.toByte()) + extra

internal fun u16(v: Int) = byteArrayOf(v.toByte(), (v shr 8).toByte())

private fun u16(b: ByteArray, at: Int) = (b[at].toInt() and 0xFF) or ((b[at + 1].toInt() and 0xFF) shl 8)

/** Returns the first segment in [b] and its encoded size, or null if [b] holds no complete segment. */
internal fun decode(b: ByteArray): Pair<Segment, Int>? {
    if (b.size < 3) return null
    val port = u16(b, 0)
    return when (val type = b[2].toInt() and 0xFF) {
        CLOSE, HEARTBEAT -> Segment(port, type, ByteArray(0)) to 3
        OPEN -> if (b.size < 5) null else Segment(port, type, b.copyOfRange(3, 5)) to 5
        TRANSMIT -> {
            if (b.size < 5) return null
            val end = 5 + u16(b, 3)
            if (b.size < end) null else Segment(port, type, b.copyOfRange(5, end)) to end
        }
        else -> throw IOException("Unknown segment type $type")
    }
}

/** MIBBridge protocol: multiplexes virtual TCP connections to the car over one RFCOMM stream. */
class Tunnel(private val input: InputStream, private val output: OutputStream) : Closeable {
    private val channels = ConcurrentHashMap<Int, Channel>()
    private val nextPort = AtomicInteger()
    @Volatile var closed = false
        private set
    @Volatile private var lastReceived = System.currentTimeMillis()

    init {
        thread(isDaemon = true, name = "tunnel") { readLoop() }
        thread(isDaemon = true, name = "heartbeat") { heartbeatLoop() }
    }

    /** Opens a connection to [remotePort] on the car. Reads fail after [timeoutMs] without data. */
    fun open(remotePort: Int, timeoutMs: Long = 15_000): Channel {
        if (closed) throw IOException("Bluetooth connection closed")
        while (true) {
            val ch = Channel(0xC000 + (nextPort.getAndIncrement() and 0x3FFF), timeoutMs)
            if (channels.putIfAbsent(ch.port, ch) == null) {
                send(frame(ch.port, OPEN, u16(remotePort)))
                return ch
            }
        }
    }

    override fun close() {
        if (closed) return
        closed = true
        channels.values.forEach { it.pipe.end() }
        channels.clear()
        runCatching { input.close() }
        runCatching { output.close() }
    }

    private fun send(bytes: ByteArray) = synchronized(output) {
        output.write(bytes)
        output.flush()
    }

    private fun heartbeatLoop() {
        try {
            while (!closed) {
                send(frame(0, HEARTBEAT))
                Thread.sleep(HEARTBEAT_MS)
                if (System.currentTimeMillis() - lastReceived > SILENCE_MS) {
                    Log.w(TAG, "No data from the car for ${SILENCE_MS / 1000} s")
                    break
                }
            }
        } catch (e: IOException) {
            if (!closed) Log.w(TAG, "Heartbeat failed", e)
        } finally {
            close()
        }
    }

    private fun readLoop() {
        val chunk = ByteArray(4096)
        var buf = ByteArray(0)
        try {
            while (true) {
                val n = input.read(chunk)
                if (n < 0) break
                lastReceived = System.currentTimeMillis()
                buf += chunk.copyOf(n)
                while (true) {
                    val (segment, size) = decode(buf) ?: break
                    buf = buf.copyOfRange(size, buf.size)
                    handle(segment)
                }
            }
        } catch (e: IOException) {
            if (!closed) Log.w(TAG, "Tunnel read failed", e)
        } finally {
            close()
        }
    }

    private fun handle(s: Segment) {
        when (s.type) {
            HEARTBEAT -> {}
            TRANSMIT -> channels[s.port]?.pipe?.put(s.data) ?: Log.w(TAG, "Data for unknown port ${s.port}")
            CLOSE -> channels[s.port]?.onCarClose()
            else -> Log.w(TAG, "Unexpected segment type ${s.type}")
        }
    }

    inner class Channel(val port: Int, timeoutMs: Long) : Closeable {
        internal val pipe = Pipe(timeoutMs)
        @Volatile private var closedByCar = false
        @Volatile private var closedByUs = false

        val input: InputStream = pipe.input

        val output: OutputStream = object : OutputStream() {
            override fun write(b: Int) = write(byteArrayOf(b.toByte()), 0, 1)

            override fun write(b: ByteArray, off: Int, len: Int) {
                if (closedByCar || closed) throw IOException("Connection closed by the car")
                var at = off
                while (at < off + len) {
                    val n = minOf(MAX_PAYLOAD, off + len - at)
                    send(frame(port, TRANSMIT, u16(n) + b.copyOfRange(at, at + n)))
                    at += n
                }
            }
        }

        override fun close() {
            if (closedByUs) return
            closedByUs = true
            pipe.end()
            if (closedByCar) return
            // The car acknowledges with its own CLOSE. onCarClose then unregisters the port.
            runCatching { send(frame(port, CLOSE)) }.onFailure { channels.remove(port) }
        }

        internal fun onCarClose() {
            if (!closedByUs) {
                closedByCar = true
                pipe.end()
                runCatching { send(frame(port, CLOSE)) }
            }
            channels.remove(port)
        }
    }

    private companion object {
        const val TAG = "Tunnel"
    }
}

/** In-memory byte stream between threads. Reads fail after [timeoutMs] without data. */
internal class Pipe(private val timeoutMs: Long = Long.MAX_VALUE) {
    private val queue = LinkedBlockingQueue<ByteArray>()
    private var current = ByteArray(0)
    private var pos = 0
    private var ended = false

    fun put(bytes: ByteArray) {
        if (bytes.isNotEmpty()) queue.add(bytes)
    }

    fun end() {
        queue.add(END)
    }

    val input: InputStream = object : InputStream() {
        override fun read(): Int {
            val b = ByteArray(1)
            return if (read(b, 0, 1) < 0) -1 else b[0].toInt() and 0xFF
        }

        override fun read(b: ByteArray, off: Int, len: Int): Int {
            if (len == 0) return 0
            while (pos == current.size) {
                if (ended) return -1
                val next = queue.poll(timeoutMs, TimeUnit.MILLISECONDS)
                    ?: throw SocketTimeoutException("No answer from the car for ${timeoutMs / 1000} s")
                ended = next === END
                current = next
                pos = 0
            }
            val n = minOf(len, current.size - pos)
            current.copyInto(b, off, pos, pos + n)
            pos += n
            return n
        }

        override fun close() = end()
    }

    val output: OutputStream = object : OutputStream() {
        override fun write(b: Int) = put(byteArrayOf(b.toByte()))
        override fun write(b: ByteArray, off: Int, len: Int) = put(b.copyOfRange(off, off + len))
        override fun close() = end()
    }

    private companion object {
        val END = ByteArray(0)
    }
}