package main

import (
	"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
	rxErrs   atomic.Uint64
	_        [16]byte
}

type rxWorker struct {
	fd      int
	batch   int
	spec    *frameSpec
	stats   *rxStats
	streams []lossWindow
	ready   *sync.WaitGroup
}

func (w *rxWorker) run(done *atomic.Bool) {
	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.stats.rxErrs.Add(1)
			}
			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) {
				w.streams[p.stream].observe(p.seq)
			}

			want, ok := w.spec.expectedCRC(p.patIdx, p.payLen)
			if !ok {
				w.stats.badLen.Add(1)
				continue
			}
			pay := buf[minFrame : minFrame+p.payLen]
			if crc32.Checksum(pay, crcTable) != want {
				w.stats.crcErr.Add(1)
			}
		}
	}
}