// The usage meter (#857, docs/ai-provisioning.md phase 1). The sidecar is the
// only place model spend is visible, so it counts here and reports upward:
// token counters per (day, model) plus activity counters, flushed to the
// *instance* (POST /api/bot-usage, bot-token-gated like charter-next-due)
// rather than straight to the cloud. Via-instance keeps the sidecar's egress
// surface at its two destinations (the instance + the model API — docs/
// bots.md) and keeps the meter token in exactly one env file, the instance's;
// the instance relays to the cloud ledger and relays the budget state back.
//
// Counters are quantities only — token counts, request counts, run counts —
// never content, per the meter-only-what-ciphertext-exposes rule (#668). They
// live in memory: a sidecar restart loses at most one flush interval, which
// is acceptable for metering (billing-grade reconciliation is cloud-side).
package bot

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"sync"
	"time"
)

// UsageRow is one meter cell: a (day, model, dimension) quantity. Model is
// empty for activity dimensions (charter_runs). Day is UTC — the ledger's
// grain, deliberately coarse.
type UsageRow struct {
	Day       string `json:"day"`
	Model     string `json:"model,omitempty"`
	Dimension string `json:"dimension"`
	Quantity  int64  `json:"quantity"`
}

type meterKey struct{ day, model, dim string }

type usageMeter struct {
	mu   sync.Mutex
	rows map[meterKey]int64
}

func newUsageMeter() *usageMeter { return &usageMeter{rows: map[meterKey]int64{}} }

func meterDay() string { return time.Now().UTC().Format("2006-01-02") }

// addLLM records one Messages call's usage block, all four token counters
// kept distinct (cache reads/writes price differently from fresh input), plus
// the request count.
func (m *usageMeter) addLLM(model string, u anthUsage) {
	m.mu.Lock()
	defer m.mu.Unlock()
	day := meterDay()
	for dim, n := range map[string]int64{
		"llm_requests":           1,
		"llm_input_tokens":       u.InputTokens,
		"llm_output_tokens":      u.OutputTokens,
		"llm_cache_write_tokens": u.CacheCreationInputTokens,
		"llm_cache_read_tokens":  u.CacheReadInputTokens,
	} {
		if n > 0 {
			m.rows[meterKey{day, model, dim}] += n
		}
	}
}

// inc records a model-less activity dimension (e.g. charter_runs).
func (m *usageMeter) inc(dim string, n int64) {
	if n <= 0 {
		return
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	m.rows[meterKey{meterDay(), "", dim}] += n
}

// drain snapshots and clears the counters; the caller restores the snapshot
// if the flush fails, so a transient error loses nothing.
func (m *usageMeter) drain() []UsageRow {
	m.mu.Lock()
	defer m.mu.Unlock()
	if len(m.rows) == 0 {
		return nil
	}
	out := make([]UsageRow, 0, len(m.rows))
	for k, q := range m.rows {
		out = append(out, UsageRow{Day: k.day, Model: k.model, Dimension: k.dim, Quantity: q})
	}
	m.rows = map[meterKey]int64{}
	return out
}

// restore folds a failed flush's rows back into the live counters.
func (m *usageMeter) restore(rows []UsageRow) {
	m.mu.Lock()
	defer m.mu.Unlock()
	for _, r := range rows {
		m.rows[meterKey{r.Day, r.Model, r.Dimension}] += r.Quantity
	}
}

// recordUsage is the model client's accounting tap (its record hook).
func (r *Runtime) recordUsage(model string, u anthUsage) {
	if r.meter != nil {
		r.meter.addLLM(model, u)
	}
}

// meterInc bumps an activity dimension from anywhere in the runtime.
func (r *Runtime) meterInc(dim string, n int64) {
	if r.meter != nil {
		r.meter.inc(dim, n)
	}
}

// meterFlushEvery paces reporting. Coarse on purpose: the ledger's grain is a
// day, so nothing downstream needs freshness — this only bounds how much a
// crash can lose.
const meterFlushEvery = 5 * time.Minute

// usageReport is one flush in flight: drained rows plus the id the ledger
// dedupes replays on. A failed flush keeps the report VERBATIM and resends
// it — never re-drains into it — because "failed" includes "the cloud
// committed but the ack got lost", and only a byte-identical replay under
// the same id lets the ledger recognize and drop the duplicate.
type usageReport struct {
	ID   string     `json:"report_id"`
	Rows []UsageRow `json:"rows"`
}

func newReportID() string {
	raw := make([]byte, 16)
	if _, err := rand.Read(raw); err != nil {
		// Out of entropy is unrecoverable weirdness; a constant would break
		// dedupe silently, so fail the report instead (retried next tick).
		return ""
	}
	return hex.EncodeToString(raw)
}

func (r *Runtime) meterLoop(ctx context.Context) {
	t := time.NewTicker(meterFlushEvery)
	defer t.Stop()
	for {
		select {
		case <-t.C:
			r.flushUsage()
		case <-ctx.Done():
			// Best-effort final flush — the graceful paths (sleep co-spawn
			// SIGTERM, deploy restart) shouldn't drop their last interval.
			r.flushUsage()
			return
		}
	}
}

// flushUsage reports counters to the instance and remembers the budget state
// the response carries (inert today; phase 3 gates the model path on it).
// Runs only on the meterLoop goroutine, so pendingUsage needs no lock: a
// pending report is retried as-is before anything new is drained, and fresh
// counts keep accumulating in the meter for the report after it.
func (r *Runtime) flushUsage() {
	if r.meter == nil {
		return
	}
	if r.pendingUsage == nil {
		rows := r.meter.drain()
		if len(rows) == 0 {
			return
		}
		id := newReportID()
		if id == "" {
			r.meter.restore(rows)
			return
		}
		r.pendingUsage = &usageReport{ID: id, Rows: rows}
	}
	var resp struct {
		OK     bool `json:"ok"`
		Budget struct {
			State string `json:"state"`
		} `json:"budget"`
	}
	if err := r.post("/api/bot-usage", map[string]any{
		"token": r.Session.Token, "report_id": r.pendingUsage.ID, "rows": r.pendingUsage.Rows,
	}, &resp); err != nil {
		r.logf("usage: flush failed (%v) — report retried next tick", err)
		return
	}
	r.pendingUsage = nil
	if resp.Budget.State != "" {
		r.budgetState.Store(resp.Budget.State)
		if resp.Budget.State != "exhausted" {
			// Back within budget (a new month, a raised allowance): re-arm
			// the one-time notice for the next exhaustion.
			r.budgetNotified.Store(false)
		}
	}
}

// budgetExhausted is phase 3's enforcement gate (docs/ai-provisioning.md):
// when the cloud's last flush response said this space's AI budget is spent,
// every model path stands down and the deterministic command floor keeps
// answering — the same degrade shape as ai_enabled off, plus one honest
// notice. Unmetered, "ok", or never-flushed all read as within budget.
func (r *Runtime) budgetExhausted() bool {
	v, _ := r.budgetState.Load().(string)
	return v == "exhausted"
}

// noteBudgetExhausted returns true exactly once per exhaustion (re-armed when
// the budget recovers), so the bot explains itself without repeating. Process
// memory only: a restart repeats the notice at most once more.
func (r *Runtime) noteBudgetExhausted() bool {
	return r.budgetNotified.CompareAndSwap(false, true)
}