// Package relay: Meter accumulates per-app byte and packet counts and
// flushes them as an hourly time series to the deployment bucket (spec §3
// metering).
package relay

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"strconv"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/metrics"
	"github.com/carlosframework/platform/internal/release"
)

// Meter satisfies ServerConfig.Meter (internal/relay/server.go).
var _ interface {
	Add(id Identity, in, out, rejected int64)
} = (*Meter)(nil)

// MeterKey is the bucket key for a box's metering chunk covering the given
// hour (metrics.HourOf's format, e.g. 2026-09-07-12).
func MeterKey(box, hour string) string { return "metrics/relay/" + box + "/" + hour + ".json" }

// MeterApp is one app's counters for one sample interval. In and Out are
// bytes forwarded on the two relay legs, counted at the socket writes — Out
// includes the STUN Data-indication framing (about 36 bytes per packet), so
// it is larger than the raw media payload. Rejected is every peer packet
// that arrived at a relay socket and was not forwarded (no permission, over
// a limit, etc). Allocs is a gauge: the number of open allocations at
// sample time, not a delta.
type MeterApp struct {
	In       int64 `json:"in"`
	Out      int64 `json:"out"`
	Rejected int64 `json:"rejected"`
	Allocs   int   `json:"allocs"`
}

// MeterSample is one flush's worth of per-app counters, keyed
// "<account>/<app>".
type MeterSample struct {
	At   time.Time           `json:"at"`
	Apps map[string]MeterApp `json:"apps"`
}

// MeterChunk is the hourly object at MeterKey: every sample flushed for
// that box and hour, in order.
type MeterChunk struct {
	V       int           `json:"v"`
	Box     string        `json:"box"`
	Hour    string        `json:"hour"`
	Samples []MeterSample `json:"samples"`
}

// Meter accumulates per-app counters between flushes and appends them as a
// sample to the box's hourly object. Safe for concurrent use.
type Meter struct {
	mu  sync.Mutex
	box string
	now func() time.Time
	// Log receives diagnostics (a corrupt chunk found, etc); nil discards.
	Log func(string, ...any)
	cur map[Identity]*MeterApp
}

// NewMeter returns a Meter for box, using now for both sample timestamps
// and hour selection (a func so tests control it), and logf for
// diagnostics (nil discards them).
func NewMeter(box string, now func() time.Time, logf func(string, ...any)) *Meter {
	return &Meter{box: box, now: now, Log: logf, cur: map[Identity]*MeterApp{}}
}

func (m *Meter) log(format string, args ...any) {
	if m.Log != nil {
		m.Log(format, args...)
	}
}

func (m *Meter) get(id Identity) *MeterApp {
	a, ok := m.cur[id]
	if !ok {
		a = &MeterApp{}
		m.cur[id] = a
	}
	return a
}

// Add accumulates in/out bytes and rejected packets for id since the last
// flush.
func (m *Meter) Add(id Identity, in, out, rejected int64) {
	m.mu.Lock()
	a := m.get(id)
	a.In += in
	a.Out += out
	a.Rejected += rejected
	m.mu.Unlock()
}

// SetAllocs sets id's current open-allocation gauge.
func (m *Meter) SetAllocs(id Identity, n int) {
	m.mu.Lock()
	m.get(id).Allocs = n
	m.mu.Unlock()
}

// SetAllocsAll sets the gauge from a complete census of live allocations:
// every id in live gets its count, and every id the meter is carrying that
// live does NOT name gets zero.
//
// The zeroing half is the point. Flush keeps an app whose gauge is above
// zero and evicts one whose four counters are all zero, so an app that is
// only ever set and never cleared reports a phantom allocation in every
// sample from then on and is never evicted — a slow leak in a process the
// unit caps at 256 MiB, and a lie in the numbers the bill is made of.
// Setting the gauge one id at a time cannot do this, because the ids that
// need zeroing are precisely the ones the caller no longer sees.
func (m *Meter) SetAllocsAll(live map[Identity]int) {
	m.mu.Lock()
	defer m.mu.Unlock()
	for id, a := range m.cur {
		a.Allocs = live[id] // absent from live: zero
	}
	for id, n := range live {
		m.get(id).Allocs = n
	}
}

// Flush writes one sample of deltas since the previous flush, appended to
// this box's hourly object by read-append-CAS. The relay is the only
// writer of these objects, so the CAS only ever races with itself.
//
// An app with nothing to report — every delta zero and no open
// allocations — is dropped from the sample and from the meter's live set,
// so a burst of short-lived apps does not grow m.cur without bound. An app
// with open allocations is kept (with zero deltas) even while idle, so its
// allocation gauge keeps being reported. And a sample left naming no app at
// all is not written: an idle relay writes nothing rather than a minute-by-
// minute record of having nothing to say.
func (m *Meter) Flush(ctx context.Context, s release.Store) error {
	m.mu.Lock()
	now := m.now()
	sample := MeterSample{At: now, Apps: map[string]MeterApp{}}
	for id, a := range m.cur {
		if a.In == 0 && a.Out == 0 && a.Rejected == 0 && a.Allocs == 0 {
			delete(m.cur, id)
			continue
		}
		sample.Apps[id.Account+"/"+id.App] = *a
		m.cur[id] = &MeterApp{Allocs: a.Allocs} // deltas reset, gauge kept
	}
	m.mu.Unlock()
	// Nothing to say, so say nothing. Almost every relay box is idle almost
	// all of the time, and a sample with no apps in it is a read, a write and
	// a row of JSON that carries no information — 1,440 of them a day per
	// box, growing the hourly object a reader has to walk and the CAS chain
	// that guards it. An hour in which nothing happened is best represented
	// by an hour with no object.
	if len(sample.Apps) == 0 {
		return nil
	}
	hour := metrics.HourOf(now)
	key := MeterKey(m.box, hour)
	var lastErr error
	for attempt := 0; attempt < 3; attempt++ {
		body, etag, err := s.Get(ctx, key)
		var c MeterChunk
		switch {
		case errors.Is(err, os.ErrNotExist):
			c = MeterChunk{V: 1, Box: m.box, Hour: hour}
			etag = ""
		case err != nil:
			return err
		default:
			if uerr := json.Unmarshal(body, &c); uerr != nil {
				m.log("relay: meter chunk %s unreadable (%v); starting a fresh chunk, %d bytes discarded", key, uerr, len(body))
				corruptKey := key + ".corrupt-" + strconv.FormatInt(now.Unix(), 10)
				if _, perr := s.Put(ctx, corruptKey, body, ""); perr != nil {
					m.log("relay: could not preserve corrupt meter chunk %s as %s: %v", key, corruptKey, perr)
				}
				c = MeterChunk{V: 1, Box: m.box, Hour: hour}
			}
		}
		c.Samples = append(c.Samples, sample)
		out, _ := json.Marshal(c)
		if _, err = s.Put(ctx, key, out, etag); err == nil {
			return nil
		} else if !errors.Is(err, release.ErrCASConflict) {
			return err
		} else {
			lastErr = err
		}
	}
	return lastErr
}

// LoadWindow returns every sample from the previous and current hour's
// objects, for rebuilding rolling-hour budgets after a restart. An hour
// whose chunk is corrupt is logged via logf (nil discards) and skipped
// rather than failing the whole load.
func LoadWindow(ctx context.Context, s release.Store, box string, now time.Time, logf func(string, ...any)) ([]MeterSample, error) {
	var out []MeterSample
	for _, h := range []string{metrics.HourOf(now.Add(-time.Hour)), metrics.HourOf(now)} {
		key := MeterKey(box, h)
		body, _, err := s.Get(ctx, key)
		if errors.Is(err, os.ErrNotExist) {
			continue
		}
		if err != nil {
			return nil, err
		}
		var c MeterChunk
		if uerr := json.Unmarshal(body, &c); uerr != nil {
			if logf != nil {
				logf("relay: meter chunk %s unreadable (%v); budgets rebuilt without it", key, uerr)
			}
			continue
		}
		out = append(out, c.Samples...)
	}
	return out, nil
}