package server

import (
	"errors"
	"io"
	"sync"
)

const (
	// jobRingSize is the backfill buffer size for the aggregated job log stream.
	// Subscribers connecting mid-run receive up to this many bytes of recent
	// history before live tail begins.
	jobRingSize = 512 * 1024

	// subChanDepth is the per-subscriber channel depth. When full, writes are
	// dropped so a slow consumer never blocks the action output pipeline.
	subChanDepth = 64

	// maxSubscribersPerStream is the maximum concurrent subscribers per stream.
	// When exceeded, subscribe returns errTooManySubscribers so the HTTP handler
	// can surface a 503.
	maxSubscribersPerStream = 32
)

// errTooManySubscribers is returned by subscribe when the stream already has
// maxSubscribersPerStream active subscribers.
var errTooManySubscribers = errors.New("too many concurrent subscribers for job log stream")

// ringBuffer is a fixed-size byte ring that retains the last len(buf) bytes
// written.
//
// Snapshot returns the current contents in write order. ringBuffer is not safe
// for concurrent use; callers (logStream) provide their own mutex.
type ringBuffer struct {
	buf  []byte
	head int // next write position; oldest is head when full
	full bool
}

func newRingBuffer(size int) *ringBuffer {
	return &ringBuffer{buf: make([]byte, size)}
}

func (r *ringBuffer) Write(p []byte) (int, error) {
	n := len(p)
	if n == 0 {
		return 0, nil
	}
	// When p is larger than the ring, only the tail fits. Skip the prefix
	// rather than cycling through multiple wraps.
	if n >= len(r.buf) {
		copy(r.buf, p[n-len(r.buf):])
		r.head = 0
		r.full = true
		return n, nil
	}
	written := copy(r.buf[r.head:], p)
	if written < n {
		copy(r.buf, p[written:])
	}
	r.head = (r.head + n) % len(r.buf)
	if r.head == 0 {
		r.full = true
	}
	return n, nil
}

// Snapshot returns a copy of the current ring contents in write order.
//
// Snapshot returns an empty slice if nothing has been written.
func (r *ringBuffer) Snapshot() []byte {
	if !r.full && r.head == 0 {
		return []byte{}
	}
	if !r.full {
		out := make([]byte, r.head)
		copy(out, r.buf[:r.head])
		return out
	}
	out := make([]byte, len(r.buf))
	copy(out, r.buf[r.head:])
	copy(out[len(r.buf)-r.head:], r.buf[:r.head])
	return out
}

// subscriber is a per-connection listener.
//
// The subscriber goroutine reads from ch; full-channel writes are dropped by
// logStream.Write.
type subscriber struct {
	ch chan []byte
}

// logStream is the broadcast hub for one migration job's aggregated action
// output.
//
// logStream implements io.WriteCloser: the loop driver's ActionLogObserver writes
// every tenant's output here in execution order, and Close is called once when
// the job finishes. On Close, all subscriber channels are closed and the
// log-stream handler reads that as the "job done" signal.
type logStream struct {
	mu     sync.Mutex
	ring   *ringBuffer
	subs   map[*subscriber]struct{}
	closed bool
}

func newLogStream() *logStream {
	return &logStream{
		ring: newRingBuffer(jobRingSize),
		subs: map[*subscriber]struct{}{},
	}
}

// Write appends p to the ring and broadcasts to every subscriber.
//
// If a subscriber's channel is full, the write is dropped for that subscriber
// only and never blocks. Write returns io.ErrClosedPipe after Close.
func (s *logStream) Write(p []byte) (int, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed {
		return 0, io.ErrClosedPipe
	}
	s.ring.Write(p)
	if len(s.subs) == 0 {
		// No subscribers: skip the dup-alloc. The ring already retained the
		// bytes for future backfill.
		return len(p), nil
	}
	// Copy p so subscribers receive a stable slice (the caller may reuse
	// the underlying buffer).
	dup := append([]byte(nil), p...)
	for sub := range s.subs {
		select {
		case sub.ch <- dup:
		default:
			// Subscriber fell behind. Drop this write so the action
			// pipeline keeps moving.
		}
	}
	return len(p), nil
}

// Close marks the stream closed and closes every subscriber channel.
//
// Log-stream handlers read the channel close as the "job done" signal.
// Subsequent writes return io.ErrClosedPipe. Close is idempotent.
func (s *logStream) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed {
		return nil
	}
	s.closed = true
	for sub := range s.subs {
		close(sub.ch)
	}
	return nil
}

// subscribe registers a new subscriber.
//
// The returned bool is the stream's closed state at subscription time. When
// the stream is already closed, the subscriber's channel is also closed so
// handlers reading on it exit immediately (they still receive any ring
// backfill before reading). subscribe returns errTooManySubscribers when the
// stream already has maxSubscribersPerStream active subscribers. Close holds
// the same mutex, so subscribe-then-Close cannot interleave; the closed bool
// is consistent with the channel state.
func (s *logStream) subscribe() (*subscriber, bool, error) {
	sub := &subscriber{ch: make(chan []byte, subChanDepth)}
	s.mu.Lock()
	defer s.mu.Unlock()
	if len(s.subs) >= maxSubscribersPerStream {
		return nil, false, errTooManySubscribers
	}
	s.subs[sub] = struct{}{}
	if s.closed {
		close(sub.ch)
	}
	return sub, s.closed, nil
}

// unsubscribe removes sub from the subscriber set.
//
// unsubscribe is safe to call after Close because sub is already removed from
// the map during Close's iteration. unsubscribe does not close sub.ch; the
// handler owns the channel lifetime.
func (s *logStream) unsubscribe(sub *subscriber) {
	s.mu.Lock()
	delete(s.subs, sub)
	s.mu.Unlock()
}

// snapshot returns a copy of the current ring contents.
//
// snapshot holds the lock while copying.
func (s *logStream) snapshot() []byte {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.ring.Snapshot()
}

// logHub holds the single aggregated log stream for a Migration.
//
// logHub is injected as migration.ActionLogObserver: every tenant's
// action output flows into one stream, in execution order. The stream is
// created eagerly from Migration.start (or lazily on first write) and closed
// once when the job ends, then recreated fresh for the next job.
type logHub struct {
	mu  sync.Mutex
	cur *logStream
}

func newLogHub() *logHub {
	return &logHub{}
}

// Write appends to the current job stream, creating it lazily when needed.
// Migration.start normally creates the stream before job.Run begins.
func (h *logHub) Write(p []byte) (int, error) {
	return h.GetOrCreate().Write(p)
}

// Stream returns an io.WriteCloser that writes to the aggregated job stream.
//
// Stream matches the migration.ActionLogObserver signature; the name argument is
// ignored because all tenants share one stream. The returned Closer is a
// no-op so individual tenant completion does not terminate the job-wide stream.
// The stream is closed once by logHub.Close when the whole job ends.
func (h *logHub) Stream(_ string) io.WriteCloser {
	return &hubSink{stream: h.GetOrCreate()}
}

// GetOrCreate returns the current job stream, creating a fresh one when no
// stream exists or the previous one is closed.
//
// The recreate-on-closed behavior supports resumed runs: a later job's
// subscribers see only that job's output.
func (h *logHub) GetOrCreate() *logStream {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.cur != nil && !h.cur.isClosedLocked() {
		return h.cur
	}
	h.cur = newLogStream()
	return h.cur
}

// Get returns the current job stream, or nil if no job has produced output yet.
//
// Get returns even closed streams. Callers that need "active or new" behavior
// should use GetOrCreate.
func (h *logHub) Get() *logStream {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.cur
}

// Close closes the current job stream.
//
// Close is called by Migration's run goroutine when a job ends so subscribers
// receive the terminal done event.
func (h *logHub) Close() {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.cur != nil {
		h.cur.Close()
	}
}

// isClosedLocked reports the stream's closed state.
//
// The caller must hold s.mu. isClosedLocked is used by logHub.GetOrCreate to
// decide whether to reuse or recreate.
func (s *logStream) isClosedLocked() bool {
	return s.closed
}

// hubSink adapts the aggregated logStream to the per-tenant Closer contract.
//
// Write fans into the stream. Close is a no-op so one tenant finishing does
// not end the job-wide tail.
type hubSink struct {
	stream *logStream
}

func (s *hubSink) Write(p []byte) (int, error) { return s.stream.Write(p) }
func (s *hubSink) Close() error                { return nil }