package main

import (
	"fmt"
	"hash/crc32"
	"sync"
	"sync/atomic"

	"golang.org/x/sys/unix"
)

type rxStats struct {
	frames   atomic.Uint64
	bytes    atomic.Uint64
	badMagic atomic.Uint64
	badLen   atomic.Uint64
	crcErr   atomic.Uint64
	_        [24]byte
}

type streamState struct {
	maxSeq atomic.Uint64
	count  atomic.Uint64
	_      [48]byte
}

type rxWorker struct {
	fd      int
	batch   int
	cpu     int
	verify  bool
	spec    *frameSpec
	stats   *rxStats
	streams []streamState
	reports chan string
	ready   *sync.WaitGroup
}

func (w *rxWorker) run(done *atomic.Bool) {
	pinTo(w.cpu)

	bufs := make([][]byte, w.batch)
	for i := range bufs {
		bufs[i] = make([]byte, maxFrame)
		for j := 0; j < maxFrame; j += 4096 {
			bufs[i][j] = 0
		}
	}
	hdrs, _ := newMmsghdrs(bufs)

	w.ready.Done()

	for !done.Load() {
		n, err := recvmmsg(w.fd, hdrs, unix.MSG_WAITFORONE)
		if n <= 0 {
			if err != nil && err != unix.EAGAIN && err != unix.EINTR {
				w.report(fmt.Sprintf("recvmmsg: %v", err))
			}
			continue
		}
		for i := 0; i < n; i++ {
			buf := bufs[i][:int(hdrs[i].len)]
			p, ok := parseHeader(buf)
			if !ok {
				w.stats.badMagic.Add(1)
				continue
			}
			w.stats.frames.Add(1)
			w.stats.bytes.Add(uint64(len(buf)))

			if int(p.stream) < len(w.streams) {
				st := &w.streams[p.stream]
				st.count.Add(1)
				for {
					old := st.maxSeq.Load()
					if p.seq <= old || st.maxSeq.CompareAndSwap(old, p.seq) {
						break
					}
				}
			}

			if !w.verify {
				continue
			}
			if p.payLen > len(w.spec.ref) {
				w.stats.badLen.Add(1)
				continue
			}
			pay := buf[minFrame : minFrame+p.payLen]
			if crc32.Checksum(pay, crcTable) == p.crc {
				continue
			}
			w.stats.crcErr.Add(1)
			off, bits := firstDiff(pay, w.spec.ref[:p.payLen])
			w.report(fmt.Sprintf("payload mismatch stream=%d seq=%d len=%d first-diff-offset=%d bits=%d",
				p.stream, p.seq, p.payLen, off, bits))
		}
	}
}

func (w *rxWorker) report(msg string) {
	select {
	case w.reports <- msg:
	default:
	}
}