package hub

import "sync/atomic"

const subBuffer = 64

// Sub is one stream's outbound channel. A subscriber that falls 64
// events behind is declared lost and told so (spec §4.4), as in Sheets.
type Sub struct {
	ch   chan Message
	lost atomic.Bool
	done chan struct{}
}

func newSub() *Sub { return &Sub{ch: make(chan Message, subBuffer), done: make(chan struct{})} }

func (s *Sub) C() <-chan Message { return s.ch }
func (s *Sub) Lost() bool        { return s.lost.Load() }

// send is non-blocking; a full buffer declares the subscriber lost,
// drains room for the lost event, and closes.
func (s *Sub) send(m Message) {
	select {
	case <-s.done:
		return
	default:
	}
	select {
	case s.ch <- m:
	default:
		s.terminate(Message{Type: TypeLost, Reason: ReasonBuffer})
	}
}

func (s *Sub) terminate(final Message) {
	if s.lost.Swap(true) {
		return
	}
	close(s.done)
	// The final event must land, but nothing already queued may be
	// dropped to make room for it: a peer ended with a real reason
	// still gets the events it had not read yet. Only a full buffer
	// costs one, and by then the subscriber is lost anyway.
	//
	// Every operation here is non-blocking, and that is the whole point
	// (Codex S2-4). This runs with h.mu held, and a failed send proves
	// the buffer was full at that instant only: the stream's own reader
	// is not excluded by the lock and can drain the channel between the
	// two statements. An unconditional receive would then block on an
	// empty channel, holding the hub's one lock against every other
	// room in the process, until the reader happened to send something
	// nobody is sending. So: try, evict one if there is one, try again.
	// Two attempts is the ceiling — the hub is the only sender and it
	// holds the lock, so a send that fails after a successful eviction
	// cannot happen, and dropping the final event is still better than
	// deadlocking the hub if it ever does.
	for i := 0; i < 2; i++ {
		select {
		case s.ch <- final:
			close(s.ch)
			return
		default:
		}
		select {
		case <-s.ch: // full a moment ago; one event makes room
		default: // the reader drained it: the retry above will fit
		}
	}
	close(s.ch)
}

func (s *Sub) Close() { s.terminate(Message{Type: TypeLost, Reason: ReasonBuffer}) }