OpsLog/internal/audio/engine.go

//go:build windows

package audio

import (
	"fmt"
	"runtime"
	"sync"
	"time"
	"unsafe"

	"github.com/go-ole/go-ole"
	"github.com/moutend/go-wca/pkg/wca"
)

const (
	// AUDCLNT_BUFFERFLAGS_SILENT — the capture packet is silent; emit zeros.
	bufferFlagSilent uint32 = 0x1
	// 1-second WASAPI buffer (REFERENCE_TIME is in 100-ns units).
	bufferDuration100ns = 10_000_000
)

func coInit() error {
	if e := ole.CoInitializeEx(0, ole.COINIT_APARTMENTTHREADED); e != nil {
		if oe, ok := e.(*ole.OleError); !ok || oe.Code() != 0x00000001 { // S_FALSE ok
			return e
		}
	}
	return nil
}

// openDevice resolves an IMMDevice by endpoint id, falling back to the default
// endpoint for the flow when id is empty or not found. Caller must Release().
func openDevice(flow uint32, id string) (*wca.IMMDevice, error) {
	var mmde *wca.IMMDeviceEnumerator
	if err := wca.CoCreateInstance(wca.CLSID_MMDeviceEnumerator, 0, wca.CLSCTX_ALL,
		wca.IID_IMMDeviceEnumerator, &mmde); err != nil {
		return nil, fmt.Errorf("create enumerator: %w", err)
	}
	defer mmde.Release()

	if id != "" {
		var coll *wca.IMMDeviceCollection
		if err := mmde.EnumAudioEndpoints(flow, wca.DEVICE_STATE_ACTIVE, &coll); err == nil && coll != nil {
			defer coll.Release()
			var count uint32
			coll.GetCount(&count)
			for i := uint32(0); i < count; i++ {
				var dev *wca.IMMDevice
				if coll.Item(i, &dev) != nil || dev == nil {
					continue
				}
				var did string
				dev.GetId(&did)
				if did == id {
					return dev, nil // caller owns it
				}
				dev.Release()
			}
		}
	}
	var dev *wca.IMMDevice
	if err := mmde.GetDefaultAudioEndpoint(flow, wca.EConsole, &dev); err != nil {
		return nil, fmt.Errorf("no audio endpoint (id %q): %w", id, err)
	}
	return dev, nil
}

// pcmFormat is the fixed capture format (16 kHz mono 16-bit PCM). WASAPI's
// AUTOCONVERTPCM resamples from the device's native mix format for us.
func pcmFormat() *wca.WAVEFORMATEX {
	return &wca.WAVEFORMATEX{
		WFormatTag:      1, // WAVE_FORMAT_PCM
		NChannels:       channels,
		NSamplesPerSec:  sampleRate,
		NAvgBytesPerSec: bytesPerSec,
		NBlockAlign:     blockAlign,
		WBitsPerSample:  bitsPerSample,
		CbSize:          0,
	}
}

const autoConvert = wca.AUDCLNT_STREAMFLAGS_AUTOCONVERTPCM | wca.AUDCLNT_STREAMFLAGS_SRC_DEFAULT_QUALITY

// recordPCM captures from a device into 16 kHz mono 16-bit PCM bytes until the
// stop channel is closed.
func recordPCM(deviceID string, stop <-chan struct{}) ([]byte, error) {
	out := make([]byte, 0, bytesPerSec*4)
	err := captureStream(deviceID, stop, func(chunk []byte) { out = append(out, chunk...) })
	return out, err
}

// captureStream opens a device and calls onChunk with freshly-captured 16 kHz
// mono 16-bit PCM as it arrives, until stop closes. onChunk receives a private
// copy it may retain. Runs on a COM-initialised, OS-locked thread.
func captureStream(deviceID string, stop <-chan struct{}, onChunk func([]byte)) error {
	runtime.LockOSThread()
	defer runtime.UnlockOSThread()
	if err := coInit(); err != nil {
		return fmt.Errorf("CoInitialize: %w", err)
	}
	defer ole.CoUninitialize()

	dev, err := openDevice(wca.ECapture, deviceID)
	if err != nil {
		return err
	}
	defer dev.Release()

	var ac *wca.IAudioClient
	if err := dev.Activate(wca.IID_IAudioClient, wca.CLSCTX_ALL, nil, &ac); err != nil {
		return fmt.Errorf("activate capture: %w", err)
	}
	defer ac.Release()

	if err := ac.Initialize(wca.AUDCLNT_SHAREMODE_SHARED, autoConvert,
		wca.REFERENCE_TIME(bufferDuration100ns), 0, pcmFormat(), nil); err != nil {
		return fmt.Errorf("initialize capture: %w", err)
	}
	var acc *wca.IAudioCaptureClient
	if err := ac.GetService(wca.IID_IAudioCaptureClient, &acc); err != nil {
		return fmt.Errorf("get capture service: %w", err)
	}
	defer acc.Release()

	if err := ac.Start(); err != nil {
		return fmt.Errorf("start capture: %w", err)
	}
	defer ac.Stop()

	for {
		select {
		case <-stop:
			return nil
		default:
		}
		var packet uint32
		if err := acc.GetNextPacketSize(&packet); err != nil {
			return err
		}
		if packet == 0 {
			time.Sleep(10 * time.Millisecond)
			continue
		}
		for packet > 0 {
			var data *byte
			var frames, flags uint32
			var devpos, qpcpos uint64
			if err := acc.GetBuffer(&data, &frames, &flags, &devpos, &qpcpos); err != nil {
				return err
			}
			n := int(frames) * blockAlign
			if n > 0 {
				chunk := make([]byte, n)
				if flags&bufferFlagSilent == 0 && data != nil {
					copy(chunk, unsafe.Slice(data, n))
				}
				onChunk(chunk)
			}
			acc.ReleaseBuffer(frames)
			if err := acc.GetNextPacketSize(&packet); err != nil {
				return err
			}
		}
	}
}

// playPCM renders raw PCM (with the given format) to a device, stopping early
// if the stop channel closes. Runs on a COM-initialised, OS-locked thread.
func playPCM(deviceID string, pcm []byte, rate, ch, bits int, stop <-chan struct{}) error {
	if len(pcm) == 0 {
		return nil
	}
	runtime.LockOSThread()
	defer runtime.UnlockOSThread()
	if err := coInit(); err != nil {
		return fmt.Errorf("CoInitialize: %w", err)
	}
	defer ole.CoUninitialize()

	dev, err := openDevice(wca.ERender, deviceID)
	if err != nil {
		return err
	}
	defer dev.Release()

	var ac *wca.IAudioClient
	if err := dev.Activate(wca.IID_IAudioClient, wca.CLSCTX_ALL, nil, &ac); err != nil {
		return fmt.Errorf("activate render: %w", err)
	}
	defer ac.Release()

	frameBytes := ch * bits / 8
	if frameBytes <= 0 {
		return fmt.Errorf("bad audio format")
	}
	wfx := &wca.WAVEFORMATEX{
		WFormatTag: 1, NChannels: uint16(ch), NSamplesPerSec: uint32(rate),
		NAvgBytesPerSec: uint32(rate * frameBytes), NBlockAlign: uint16(frameBytes),
		WBitsPerSample: uint16(bits), CbSize: 0,
	}
	if err := ac.Initialize(wca.AUDCLNT_SHAREMODE_SHARED, autoConvert,
		wca.REFERENCE_TIME(bufferDuration100ns), 0, wfx, nil); err != nil {
		return fmt.Errorf("initialize render: %w", err)
	}
	var bufFrames uint32
	if err := ac.GetBufferSize(&bufFrames); err != nil {
		return err
	}
	var arc *wca.IAudioRenderClient
	if err := ac.GetService(wca.IID_IAudioRenderClient, &arc); err != nil {
		return fmt.Errorf("get render service: %w", err)
	}
	defer arc.Release()

	totalFrames := len(pcm) / frameBytes
	written := 0
	feed := func(maxFrames int) error {
		if maxFrames <= 0 || written >= totalFrames {
			return nil
		}
		n := totalFrames - written
		if n > maxFrames {
			n = maxFrames
		}
		var data *byte
		if err := arc.GetBuffer(uint32(n), &data); err != nil {
			return err
		}
		dst := unsafe.Slice(data, n*frameBytes)
		copy(dst, pcm[written*frameBytes:(written+n)*frameBytes])
		arc.ReleaseBuffer(uint32(n), 0)
		written += n
		return nil
	}

	// Pre-fill before starting to avoid an initial glitch.
	if err := feed(int(bufFrames)); err != nil {
		return err
	}
	if err := ac.Start(); err != nil {
		return fmt.Errorf("start render: %w", err)
	}
	defer ac.Stop()

	for written < totalFrames {
		select {
		case <-stop:
			return nil
		default:
		}
		var padding uint32
		ac.GetCurrentPadding(&padding)
		if err := feed(int(bufFrames - padding)); err != nil {
			return err
		}
		time.Sleep(8 * time.Millisecond)
	}
	// Drain the remaining buffered audio.
	for {
		select {
		case <-stop:
			return nil
		default:
		}
		var padding uint32
		if ac.GetCurrentPadding(&padding) != nil || padding == 0 {
			return nil
		}
		time.Sleep(10 * time.Millisecond)
	}
}

// pcmRing is a thread-safe, latency-bounded FIFO of PCM bytes feeding a live
// render stream. Producers (a USB-codec capture, or a decoded network audio
// stream) Push freshly-arrived samples; the render loop Pulls. It is the shared
// hand-off point between "where the audio comes from" (USB device / UDP 50003)
// and "where it's heard" (any WASAPI output) — so the transport can be swapped
// without touching the render side, mirroring the civTransport split on the CAT
// side. On overflow the oldest audio is dropped to keep latency bounded; on
// underrun Pull simply returns short and the render loop pads with silence.
type pcmRing struct {
	mu  sync.Mutex
	buf []byte
	max int // hard cap in bytes (drops oldest beyond this → bounded latency)
}

// newPCMRing makes a ring whose backlog is capped at maxBytes. Size it from the
// acceptable latency: bytesPerSec (=32000) worth ≈ 1 s.
func newPCMRing(maxBytes int) *pcmRing {
	if maxBytes <= 0 {
		maxBytes = bytesPerSec // 1 s default
	}
	return &pcmRing{max: maxBytes}
}

// Push appends samples, dropping the oldest audio if the backlog would exceed
// the cap (a slow/absent consumer never makes the producer block or grow without
// bound). A short glitch beats runaway latency for live monitoring.
func (r *pcmRing) Push(p []byte) {
	if len(p) == 0 {
		return
	}
	r.mu.Lock()
	r.buf = append(r.buf, p...)
	if len(r.buf) > r.max {
		drop := len(r.buf) - r.max
		r.buf = append(r.buf[:0], r.buf[drop:]...)
	}
	r.mu.Unlock()
}

// pull removes and returns up to maxBytes of queued PCM (a private copy), or nil
// when empty. The render loop pads any shortfall with silence.
func (r *pcmRing) pull(maxBytes int) []byte {
	r.mu.Lock()
	defer r.mu.Unlock()
	if len(r.buf) == 0 || maxBytes <= 0 {
		return nil
	}
	n := maxBytes
	if n > len(r.buf) {
		n = len(r.buf)
	}
	out := make([]byte, n)
	copy(out, r.buf[:n])
	r.buf = append(r.buf[:0], r.buf[n:]...)
	return out
}

// renderStream continuously renders PCM pulled from src to a device until stop
// closes — the streaming counterpart to playPCM's fixed buffer. On underrun it
// writes silence rather than glitching, keeping the WASAPI clock steady so live
// monitor audio flows smoothly even when the source stalls briefly. Runs on a
// COM-initialised, OS-locked thread.
func renderStream(deviceID string, rate, ch, bits int, stop <-chan struct{}, src *pcmRing) error {
	runtime.LockOSThread()
	defer runtime.UnlockOSThread()
	if err := coInit(); err != nil {
		return fmt.Errorf("CoInitialize: %w", err)
	}
	defer ole.CoUninitialize()

	dev, err := openDevice(wca.ERender, deviceID)
	if err != nil {
		return err
	}
	defer dev.Release()

	var ac *wca.IAudioClient
	if err := dev.Activate(wca.IID_IAudioClient, wca.CLSCTX_ALL, nil, &ac); err != nil {
		return fmt.Errorf("activate render: %w", err)
	}
	defer ac.Release()

	frameBytes := ch * bits / 8
	if frameBytes <= 0 {
		return fmt.Errorf("bad audio format")
	}
	wfx := &wca.WAVEFORMATEX{
		WFormatTag: 1, NChannels: uint16(ch), NSamplesPerSec: uint32(rate),
		NAvgBytesPerSec: uint32(rate * frameBytes), NBlockAlign: uint16(frameBytes),
		WBitsPerSample: uint16(bits), CbSize: 0,
	}
	if err := ac.Initialize(wca.AUDCLNT_SHAREMODE_SHARED, autoConvert,
		wca.REFERENCE_TIME(bufferDuration100ns), 0, wfx, nil); err != nil {
		return fmt.Errorf("initialize render: %w", err)
	}
	var bufFrames uint32
	if err := ac.GetBufferSize(&bufFrames); err != nil {
		return err
	}
	var arc *wca.IAudioRenderClient
	if err := ac.GetService(wca.IID_IAudioRenderClient, &arc); err != nil {
		return fmt.Errorf("get render service: %w", err)
	}
	defer arc.Release()

	// feed fills up to `frames` render frames: as much real audio as the ring
	// has, the remainder silence (so the buffer stays full and the clock steady).
	feed := func(frames int) error {
		if frames <= 0 {
			return nil
		}
		var data *byte
		if err := arc.GetBuffer(uint32(frames), &data); err != nil {
			return err
		}
		dst := unsafe.Slice(data, frames*frameBytes)
		got := src.pull(frames * frameBytes)
		n := copy(dst, got)
		for i := n; i < len(dst); i++ {
			dst[i] = 0 // silence-fill the shortfall
		}
		arc.ReleaseBuffer(uint32(frames), 0)
		return nil
	}

	if err := feed(int(bufFrames)); err != nil { // pre-fill to avoid a start glitch
		return err
	}
	if err := ac.Start(); err != nil {
		return fmt.Errorf("start render: %w", err)
	}
	defer ac.Stop()

	for {
		select {
		case <-stop:
			return nil
		default:
		}
		var padding uint32
		ac.GetCurrentPadding(&padding)
		if err := feed(int(bufFrames - padding)); err != nil {
			return err
		}
		time.Sleep(8 * time.Millisecond)
	}
}