// Linux commissioning receiver. SO_REUSEPORT allows multiple binders but
// ordinary flow hashing can still send the entire feed to one process.
package main

import (
    "bytes"
    "context"
    "encoding/binary"
    "fmt"
    "log"
    "net"
    "os"
    "strconv"
    "syscall"
    "time"
)

const sourceIP = "203.23.178.228"
const maxPayload = 1228
const commonHeader = 83
const soReusePort = 15 // Linux SOL_SOCKET ABI; use x/sys/unix for other targets.

func shredKind(variant byte) string {
    switch variant {
    case 0xa5: return "legacy-data"
    case 0x5a: return "legacy-code"
    }
    switch variant & 0xf0 {
    case 0x80, 0x90, 0xb0: return "merkle-data"
    case 0x40, 0x60, 0x70: return "merkle-code"
    default: return "unknown"
    }
}

func main() {
    port := 9000
    if len(os.Args) > 1 {
        var err error
        port, err = strconv.Atoi(os.Args[1])
        if err != nil || port < 1 || port > 65535 { log.Fatal("usage: receiver [udp-port]") }
    }

    cfg := net.ListenConfig{Control: func(_, _ string, raw syscall.RawConn) error {
        var sockErr error
        if err := raw.Control(func(fd uintptr) {
            sockErr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, soReusePort, 1)
        }); err != nil { return err }
        return sockErr
    }}
    listener, err := cfg.ListenPacket(context.Background(), "udp4", fmt.Sprintf("0.0.0.0:%d", port))
    if err != nil { log.Fatal(err) }
    defer listener.Close()
    conn := listener.(*net.UDPConn)
    if err := conn.SetReadBuffer(8 * 1024 * 1024); err != nil { log.Fatal(err) }
    log.Printf("listening on %s", conn.LocalAddr())

    packet := make([]byte, 2048)
    var packets, totalBytes uint64
    last := time.Now()
    for {
        n, peer, err := conn.ReadFromUDPAddrPort(packet)
        if err != nil { log.Fatal(err) }
        if peer.Addr().String() != sourceIP || n > maxPayload { continue }
        data := packet[:n]
        if bytes.HasPrefix(data, []byte("SHREDCAST-VERIFY")) {
            fmt.Printf("verification=%s\n", data)
            continue
        }
        packets++
        totalBytes += uint64(n)
        if n >= commonHeader && packets%10000 == 1 {
            fmt.Printf("sample kind=%s slot=%d index=%d version=%d fec=%d len=%d\n",
                shredKind(data[64]), binary.LittleEndian.Uint64(data[65:73]),
                binary.LittleEndian.Uint32(data[73:77]),
                binary.LittleEndian.Uint16(data[77:79]),
                binary.LittleEndian.Uint32(data[79:83]), n)
        }
        if time.Since(last) >= time.Second {
            fmt.Printf("stream_packets=%d stream_bytes=%d\n", packets, totalBytes)
            packets, totalBytes = 0, 0
            last = time.Now()
        }
    }
}