package applog

import (
	"cmp"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"
)

// AppKey names the (account, app) a buffered record belongs to — the unit
// both budgets and bucket keys are scoped by.
type AppKey struct{ Acct, App string }

// Sink buffers tagged records between ship-loop flushes: a bounded ring per
// app plus a token-bucket rate cap per app, both with drop honesty — when
// either binds, oldest lines go first (edge records before any app or
// platform line; see edgeRequest) and the next Drain carries one synthetic
// record naming the count, never silent thinning (spec §§4–5). Safe for concurrent use; producers (exec tees, event emit
// sites) and the ship loop share one Sink per agent process.
type Sink struct {
	// Budgets are fields, not constructor params, so tests and future
	// per-app overrides set them directly; the zero Sink is unusable —
	// NewSink applies the spec defaults.
	MaxBytesPerApp int     // ring cap per app (default 4 MiB)
	RatePerMin     float64 // sustained app-line budget per app (default 1 MiB/min)
	Burst          float64 // token bucket depth (default 4 MiB)

	// The edge stream has its own bucket, so a flood of request records
	// (a scanner hammering one host) can never spend the tokens the app's
	// own lines need. Same unit as RatePerMin: recCost plus field bytes.
	EdgeRatePerMin float64 // sustained edge-record budget per app (default 256 KiB/min)
	EdgeBurst      float64 // edge bucket depth (default 1 MiB)

	box string
	now func() time.Time

	mu   sync.Mutex
	apps map[AppKey]*appBuf
	seq  uint64 // monotonic across apps AND restarts — the follow cursor's spine

	notify chan struct{} // poked (non-blocking) on Event; the ship loop's fast-flush signal
}

type appBuf struct {
	// recs holds app and platform records and edge holds edge records by
	// eviction class, each list in seq order; Drain merges them back into
	// one. Keeping them apart is what lets the ring shed edge records first
	// without a scan.
	recs      []Record
	edge      [edgeClasses][]Record
	edgeBytes [edgeClasses]int
	bytes     int // every list's records, by costOf
	tokens    float64
	lastRef   time.Time
	dropped   int

	// Edge bucket state. edgeInit is false until the app's first edge
	// record: the bucket is made lazily, starting full at whatever
	// EdgeBurst is by then, as buf() does for the app bucket.
	edgeInit    bool
	edgeTokens  float64
	edgeLastRef time.Time
	edgeDropped int
}

// Edge records are evicted from the ring in class order: request records
// first, then blocks, then summaries. A record may only displace edge records
// of its own class or a lower one, and never an app or platform record, so an
// edge flood cannot evict the app's own lines (spec §7), a not-found or watch
// flood cannot evict the record of a block, and nothing but other summaries
// can evict the minute counts that say how much was sampled away.
const (
	edgeRequest = iota // not-found and watch records, and any other kind
	edgeBlock          // block records (kind EdgeKindBlock)
	edgeSummary        // minute counts (EdgeSummary) and the edge drop notice
	edgeClasses
)

// EdgeKindBlock is the kind field the edge's request recorder (internal/reqrec,
// KindBlock) writes on a block record. applog imports nothing from reqrec, so
// the word is repeated here; internal/host tests that the two agree.
const EdgeKindBlock = "block"

// NewSink returns a Sink stamping records with box and timestamps from now
// (a seam so tests control the clock). The sequence counter starts at the
// construction instant's UnixMilli rather than zero: chunk keys embed the
// first record's seq, and per-box seq order is a promise the cursor relies
// on — a restart that re-based to zero would write keys that sort BEFORE
// last boot's and break exact per-box ordering. Milliseconds always outrun
// the admitted line rate (the rate cap is ~17 KiB/s), so a rebase lands
// above everything the previous process numbered.
func NewSink(box string, now func() time.Time) *Sink {
	return &Sink{
		MaxBytesPerApp: 4 << 20,
		RatePerMin:     1 << 20,
		Burst:          4 << 20,
		EdgeRatePerMin: 256 << 10,
		EdgeBurst:      1 << 20,
		box:            box,
		now:            now,
		apps:           make(map[AppKey]*appBuf),
		seq:            uint64(now().UnixMilli()),
		notify:         make(chan struct{}, 1),
	}
}

// Wake is the ship loop's fast-flush signal: it receives after an Event has
// been buffered, so a wake failure reaches the bucket in seconds rather
// than at the next tick (spec §5). Level-triggered and lossy by design —
// one pending poke summarizes any number of events.
func (s *Sink) Wake() <-chan struct{} { return s.notify }

// Known lists every app this sink has ever buffered for, in no particular
// order — the ship loop's retention worklist.
func (s *Sink) Known() []AppKey {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]AppKey, 0, len(s.apps))
	for k := range s.apps {
		out = append(out, k)
	}
	return out
}

// App buffers one line of app output for (acct, app), parsing structured
// JSON lines at the door (ParseAppLine) so the bucket holds fields and
// honored levels rather than opaque text. Subject to both budgets.
func (s *Sink) App(acct, app, host, line string) {
	msg, level, fields := ParseAppLine(line)
	s.mu.Lock()
	defer s.mu.Unlock()
	b := s.buf(AppKey{acct, app})
	cost := float64(recCost(msg))
	b.refill(s.now(), s.RatePerMin, s.Burst)
	if b.tokens < cost {
		b.dropped++
		return
	}
	b.tokens -= cost
	s.append(b, Record{
		Host:   host,
		Stream: StreamApp,
		Level:  level,
		Msg:    msg,
		Fields: fields,
	}, -1)
}

// Edge buffers one edge request record for (acct, app). It runs on the
// request goroutine through reqrec, so it takes s.mu briefly and does
// nothing else: no I/O, no blocking send, and it does not poke notify
// (edge records ride the ship tick; only wake failures need the fast
// flush). The edge bucket is separate from App's, and the fields map is
// copied because callers may reuse it. In the ring the record may displace
// only edge records (see edgeRequest); with none to displace it is refused
// and counted in the edge drop notice.
func (s *Sink) Edge(acct, app, host, level, msg string, fields map[string]string) {
	r := Record{
		Host:   host,
		Stream: StreamEdge,
		Level:  level,
		Msg:    msg,
		Fields: copyFields(fields),
	}
	cost := float64(costOf(r))
	s.mu.Lock()
	defer s.mu.Unlock()
	b := s.buf(AppKey{acct, app})
	now := s.now()
	if !b.edgeInit {
		b.edgeInit, b.edgeTokens, b.edgeLastRef = true, s.EdgeBurst, now
	}
	elapsed := now.Sub(b.edgeLastRef)
	b.edgeLastRef = now
	b.edgeTokens += s.EdgeRatePerMin * elapsed.Minutes()
	if b.edgeTokens > s.EdgeBurst {
		b.edgeTokens = s.EdgeBurst
	}
	if b.edgeTokens < cost {
		b.edgeDropped++
		return
	}
	b.edgeTokens -= cost
	class := edgeRequest
	if r.Fields["kind"] == EdgeKindBlock {
		class = edgeBlock
	}
	s.append(b, r, class)
}

// EdgeSummary buffers one edge summary record (the per-minute counts) for
// (acct, app). It bypasses the edge rate bucket, as Event bypasses the app
// one: the counts are what explain the sampling, and a flood that has spent
// the edge budget is exactly when they matter. The ring still bounds it, and
// it is the last edge record evicted.
func (s *Sink) EdgeSummary(acct, app, host, level, msg string, fields map[string]string) {
	r := Record{
		Host:   host,
		Stream: StreamEdge,
		Level:  level,
		Msg:    msg,
		Fields: copyFields(fields),
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	s.append(s.buf(AppKey{acct, app}), r, edgeSummary)
}

func copyFields(f map[string]string) map[string]string {
	if len(f) == 0 {
		return nil
	}
	out := make(map[string]string, len(f))
	for k, v := range f {
		out[k] = v
	}
	return out
}

// Event buffers one platform event — member-safe text composed at the emit
// site (spec §5), never a relayed journal line. Events bypass the rate cap:
// they are the diagnosis, and a chatty app must not crowd out its own
// wake-failure line. The ring cap still applies (a platform-event storm is
// a bug, not a budget).
func (s *Sink) Event(acct, app, host, level, msg string) {
	s.mu.Lock()
	s.append(s.buf(AppKey{acct, app}), Record{
		Host:   host,
		Stream: StreamPlatform,
		Level:  level,
		Msg:    msg,
	}, -1)
	s.mu.Unlock()
	select { // non-blocking: a pending poke already says everything
	case s.notify <- struct{}{}:
	default:
	}
}

// dropNoticeSuffix ends the synthetic line Drain appends for an app whose
// lines hit the rate cap.
const dropNoticeSuffix = " (rate cap)"

// IsDropNotice reports whether r is that synthetic line: a platform event
// about the logging budget, not about the instance, so Availability (which
// takes the rest of the platform stream) leaves it out.
func IsDropNotice(r Record) bool {
	return r.Stream == StreamPlatform && strings.HasPrefix(r.Msg, "dropped ") && strings.HasSuffix(r.Msg, dropNoticeSuffix)
}

// Drain returns everything buffered, keyed by app, and resets the buffers
// (sequence numbers and rate-bucket state persist — they are the process's,
// not the batch's). An app whose lines were dropped since the last drain
// gets one trailing synthetic record naming the count. Empty apps are
// omitted; an empty sink drains to an empty map.
func (s *Sink) Drain() map[AppKey][]Record {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make(map[AppKey][]Record)
	for k, b := range s.apps {
		if b.dropped > 0 {
			s.append(b, Record{
				Stream: StreamPlatform,
				Level:  LevelWarn,
				Msg:    fmt.Sprintf("dropped %d app log line(s)"+dropNoticeSuffix, b.dropped),
			}, -1)
			b.dropped = 0
		}
		if n := b.edgeDropped; n > 0 {
			// Not IsDropNotice: this is an edge-stream record about the
			// edge budget, so Availability (platform stream) never sees it.
			// It is a summary for eviction. Records it evicts to fit are
			// counted afresh, for the next drain's notice; if it cannot fit
			// at all (the ring holds only app lines), the count waits.
			b.edgeDropped = 0
			if !s.append(b, Record{
				Stream: StreamEdge,
				Level:  LevelWarn,
				Msg:    fmt.Sprintf("dropped %d edge record(s)"+dropNoticeSuffix, n),
				Fields: map[string]string{"kind": "dropped", "n": strconv.Itoa(n)},
			}, edgeSummary) {
				b.edgeDropped = n
			}
		}
		recs := b.drainAll()
		if len(recs) == 0 {
			continue
		}
		out[k] = recs
	}
	return out
}

// drainAll empties the buffer and returns its records in seq order. Callers
// hold s.mu.
func (b *appBuf) drainAll() []Record {
	n := len(b.recs)
	for _, l := range b.edge {
		n += len(l)
	}
	if n == 0 {
		return nil
	}
	out := make([]Record, 0, n)
	out = append(out, b.recs...)
	for _, l := range b.edge {
		out = append(out, l...)
	}
	// Each list is already in seq order; one sort merges them.
	slices.SortFunc(out, func(x, y Record) int { return cmp.Compare(x.Seq, y.Seq) })
	b.recs = nil
	b.edge = [edgeClasses][]Record{}
	b.edgeBytes = [edgeClasses]int{}
	b.bytes = 0
	return out
}

// buf returns (creating if needed) the app's buffer. Callers hold s.mu.
func (s *Sink) buf(k AppKey) *appBuf {
	b := s.apps[k]
	if b == nil {
		b = &appBuf{tokens: s.Burst, lastRef: s.now()}
		s.apps[k] = b
	}
	return b
}

// append stamps and stores one record, evicting past the ring cap, and
// reports whether it was stored. class is the edge eviction class of an edge
// record, or -1 for an app or platform record.
//
// An edge record may displace only edge records of its class or a lower one,
// oldest first; if even all of those would not make room it is refused
// (counted in edgeDropped) and nothing is evicted. An app or platform record
// displaces edge records first, lowest class first, and only then the oldest
// app or platform record, never itself: a single oversized line still lands,
// as before. Callers hold s.mu.
func (s *Sink) append(b *appBuf, r Record, class int) bool {
	cost := costOf(r)
	if class >= 0 {
		room := s.MaxBytesPerApp - b.bytes
		for c := 0; c <= class; c++ {
			room += b.edgeBytes[c]
		}
		if cost > room {
			b.edgeDropped++
			return false
		}
	}
	s.seq++
	r.TS = s.now().UTC()
	r.Box = s.box
	r.Seq = s.seq
	if class >= 0 {
		b.edge[class] = append(b.edge[class], r)
		b.edgeBytes[class] += cost
	} else {
		b.recs = append(b.recs, r)
	}
	b.bytes += cost
	for b.bytes > s.MaxBytesPerApp {
		if !b.evictOne(class) {
			break
		}
	}
	return true
}

// evictOne drops the oldest record the incoming record of class may displace
// (see append) and reports whether there was one. The incoming record is the
// newest in its list and is never the one dropped. Callers hold s.mu.
func (b *appBuf) evictOne(class int) bool {
	top := class
	if class < 0 {
		top = edgeClasses - 1
	}
	for c := 0; c <= top; c++ {
		keep := 0
		if c == class {
			keep = 1 // the incoming record itself
		}
		if len(b.edge[c]) > keep {
			old := b.edge[c][0]
			b.edge[c] = b.edge[c][1:]
			b.edgeBytes[c] -= costOf(old)
			b.bytes -= costOf(old)
			b.edgeDropped++
			return true
		}
	}
	if class < 0 && len(b.recs) > 1 {
		old := b.recs[0]
		b.recs = b.recs[1:]
		b.bytes -= costOf(old)
		b.dropped++
		return true
	}
	return false
}

// refill advances the token bucket to now. Callers hold s.mu.
func (b *appBuf) refill(now time.Time, perMin, burst float64) {
	elapsed := now.Sub(b.lastRef)
	b.lastRef = now
	b.tokens += perMin * elapsed.Minutes()
	if b.tokens > burst {
		b.tokens = burst
	}
}

// recCost is the budget cost of one record: its message plus a flat
// overhead standing in for the envelope. Both budgets use the same measure
// so "4 MiB of ring" and "1 MiB/min of rate" describe the same bytes.
func recCost(msg string) int { return len(msg) + 64 }

// costOf is recCost for a whole record. Edge records carry their request
// detail in Fields, so those bytes count; app and platform records cost
// exactly what they did before, which keeps their budgets unchanged. The
// ring accounting in append (add and evict) and the edge bucket must all
// use this one measure.
func costOf(r Record) int {
	c := recCost(r.Msg)
	if r.Stream == StreamEdge {
		for k, v := range r.Fields {
			c += len(k) + len(v)
		}
	}
	return c
}