// Package listener spins up one listener per registered device, hands each
// inbound ASTM session to the queue, and stops cleanly on shutdown.
package listener

import (
	"context"
	"log"
	"sync"
	"time"

	"tps.local/labbridge-agent/internal/queue"
	syncpkg "tps.local/labbridge-agent/internal/sync"
)

// Sink is the callback invoked when a listener has assembled a complete
// device message. The manager wires this to the queue store + sync trigger.
type Sink func(deviceID *int64, vendor, raw, peer string, receivedAt time.Time)

// Manager owns all per-device listeners. Devices are reconfigured by passing
// a fresh list to Start; the next call would replace the set (not implemented
// here — V1 expects a restart on config change).
type Manager struct {
	store  *queue.Store
	worker *syncpkg.Worker

	mu        sync.Mutex
	listeners []closer
}

type closer interface{ Close() error }

func NewManager(store *queue.Store, worker *syncpkg.Worker) *Manager {
	return &Manager{store: store, worker: worker}
}

// Start opens listeners for every supported device. Devices with unsupported
// connection types are skipped with a warning.
func (m *Manager) Start(ctx context.Context, devices []syncpkg.Device) {
	sink := m.makeSink()
	for _, d := range devices {
		d := d
		switch d.ConnectionType {
		case "tcp", "serial2eth":
			l, err := startTCP(ctx, d, sink)
			if err != nil {
				log.Printf("listener: device %d %q tcp: %v", d.ID, d.Name, err)
				continue
			}
			m.add(l)
			log.Printf("listener: device %d %q TCP on :%d (%s)", d.ID, d.Name, d.Port, d.Vendor)
		case "serial":
			l, err := startSerial(ctx, d, sink)
			if err != nil {
				log.Printf("listener: device %d %q serial: %v", d.ID, d.Name, err)
				continue
			}
			m.add(l)
			log.Printf("listener: device %d %q serial %s @ %d (%s)", d.ID, d.Name, d.ComPort, d.Baud, d.Vendor)
		default:
			log.Printf("listener: device %d %q unknown connection_type %q — skipped", d.ID, d.Name, d.ConnectionType)
		}
	}
}

// Stop closes every listener. Safe to call multiple times.
func (m *Manager) Stop() {
	m.mu.Lock()
	defer m.mu.Unlock()
	for _, l := range m.listeners {
		_ = l.Close()
	}
	m.listeners = nil
}

func (m *Manager) add(l closer) {
	m.mu.Lock()
	m.listeners = append(m.listeners, l)
	m.mu.Unlock()
}

// makeSink writes an inbound device message into the queue and kicks the
// sync worker so the cloud sees it without waiting for the next tick.
func (m *Manager) makeSink() Sink {
	return func(deviceID *int64, vendor, raw, peer string, receivedAt time.Time) {
		uuid, err := m.store.EnqueueResult(vendor, deviceID, raw, receivedAt, "")
		if err != nil {
			log.Printf("listener: enqueue: %v", err)
			return
		}
		log.Printf("listener: queued %s from %s (vendor=%s, %d bytes)", uuid, peer, vendor, len(raw))
		// Record the connection event so the cloud's device monitor sees it
		// even when the sync sweep is delayed.
		_ = m.store.EnqueueEvent(queue.Event{
			Type: "log", Level: "info", Category: "result",
			Vendor: vendor, DeviceID: deviceID, DeviceIP: peer,
			Message: "Result message received from analyzer",
		})
		m.worker.SweepNow()
	}
}
