package xds

import (
	"errors"
	"sync"
)

// Subscriber is one Envoy's stream: it receives pushed snapshots.
type Subscriber struct {
	node string
	ch   chan *Snapshot
}

// Updates is the channel the Envoy reads pushed snapshots from.
func (s *Subscriber) Updates() <-chan *Snapshot { return s.ch }

// SnapshotCache holds the latest snapshot per node and tracks, per node,
// the version each Envoy has ACKed (or the last NACK error). It is the
// control plane's source of truth about the fleet's deployed config.
type SnapshotCache struct {
	mu      sync.Mutex
	snaps   map[string]*Snapshot
	subs    map[string][]*Subscriber
	applied map[string]string // node -> last ACKed version (last-known-good)
	nacks   map[string]string // node -> last NACK reason
}

func NewSnapshotCache() *SnapshotCache {
	return &SnapshotCache{
		snaps:   map[string]*Snapshot{},
		subs:    map[string][]*Subscriber{},
		applied: map[string]string{},
		nacks:   map[string]string{},
	}
}

// Subscribe registers a node's stream and immediately pushes the current
// snapshot if one exists.
func (c *SnapshotCache) Subscribe(node string) *Subscriber {
	c.mu.Lock()
	defer c.mu.Unlock()
	sub := &Subscriber{node: node, ch: make(chan *Snapshot, 8)}
	c.subs[node] = append(c.subs[node], sub)
	if s, ok := c.snaps[node]; ok {
		sub.ch <- s
	}
	return sub
}

// SetSnapshot validates and stores a snapshot for a node, then pushes it
// to that node's subscribers. An inconsistent snapshot is REJECTED and
// the previous snapshot remains in force (never push a half-broken set).
func (c *SnapshotCache) SetSnapshot(node string, s *Snapshot) error {
	if err := s.Consistent(); err != nil {
		return err
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	c.snaps[node] = s
	for _, sub := range c.subs[node] {
		select {
		case sub.ch <- s:
		default: // slow subscriber; in real xDS the stream applies backpressure
		}
	}
	return nil
}

// Ack records that node successfully applied a version (Envoy echoing the
// version in its next DiscoveryRequest). This is the rollout signal:
// the control plane knows exactly which version each node runs.
func (c *SnapshotCache) Ack(node, version string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.applied[node] = version
	delete(c.nacks, node)
}

// Nack records that node REJECTED a version (DiscoveryRequest with an
// error_detail). The applied version stays at the last good one — Envoy
// keeps serving last-known-good. Unmonitored NACKs are a silent "my
// change didn't take" failure; alert on them.
func (c *SnapshotCache) Nack(node, rejectedVersion, reason string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.nacks[node] = reason
}

// AppliedVersion returns the last version node ACKed ("" if none).
func (c *SnapshotCache) AppliedVersion(node string) string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.applied[node]
}

// LastNack returns node's last NACK reason ("" if none / cleared by ACK).
func (c *SnapshotCache) LastNack(node string) string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.nacks[node]
}

// ErrNoSnapshot is returned when querying a node with no snapshot.
var ErrNoSnapshot = errors.New("xds: no snapshot for node")

// CurrentVersion returns the version currently set for node.
func (c *SnapshotCache) CurrentVersion(node string) (string, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	s, ok := c.snaps[node]
	if !ok {
		return "", ErrNoSnapshot
	}
	return s.Version, nil
}