// Package relay: serve.go is the process. Everything else in this package
// binds a socket or parses a record; this is the loop that owns them for the
// life of `carlos relay serve` (spec §3). It reads the masters file as it
// changes, the authority record every minute and on every control/serial
// change, writes a heartbeat on each read, flushes the meter each minute,
// sweeps expiry, and sheds under the box's memory pressure.
package relay

import (
	"bufio"
	"context"
	"errors"
	"fmt"
	"net"
	"os"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

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

// The relay's fixed listeners (spec §3). Ports, not addresses: 3478 is the
// registered STUN/TURN port and the DNS names and security group are
// written against it, so it is not tunable.
const (
	relayUDPAddr = ":3478"
	relayTCPAddr = ":3478"

	// defaultMastersPath is where the operator stages the region masters
	// and what carlos-relay.service's ConditionPathExists names.
	defaultMastersPath = "/etc/carlos/relay-masters"

	// defaultPressurePath is the kernel's memory pressure-stall file. The
	// relay reads the `some avg10` value out of it (spec §3).
	defaultPressurePath = "/proc/pressure/memory"

	// shedAbove and shedUntil are the pressure hysteresis from spec §3: the
	// relay starts shedding above 20% and keeps going until pressure falls
	// under 10%, so a relay that has just freed a little memory does not
	// flap in and out of shedding on every tick.
	shedAbove = 20.0
	shedUntil = 10.0
)

// The serve loop's cadences. Vars, not consts, so a test can shorten them.
var (
	authInterval   = time.Minute
	serialInterval = 3 * time.Second
	meterInterval  = time.Minute
	expiryInterval = 10 * time.Second
	psiInterval    = 5 * time.Second

	// The store-open backoff (OpenStoreWithRetry).
	storeRetryInitial  = time.Second
	storeRetryMax      = 30 * time.Second
	storeRetryLogEvery = time.Minute
)

// OpenStoreWithRetry calls open until it succeeds or ctx is done, backing
// off from a second to half a minute and logging at most one line a minute.
//
// It exists because the alternative is worse than it looks. The relay's
// unit is Restart=always, and systemd's default start limit turns five
// restarts in ten seconds into a unit that sits in `failed` and stays
// there: a box whose deployment bucket was unreachable for the fifteen
// seconds around boot would then serve no STUN until a platform update or
// a human touched it — and STUN, the whole reason this stage exists, needs
// no store at all. A relay that waits is a relay that comes back on its
// own. Configuration errors are NOT retried here; they are fatal at the
// call site, because no amount of waiting fixes a missing public IP.
func OpenStoreWithRetry(ctx context.Context, open func(context.Context) (release.Store, error), logf func(string, ...any)) (release.Store, error) {
	if logf == nil {
		logf = func(string, ...any) {}
	}
	delay := storeRetryInitial
	var lastLog time.Time
	for attempt := 1; ; attempt++ {
		s, err := open(ctx)
		if err == nil {
			if attempt > 1 {
				logf("relay: deploy store opened on attempt %d", attempt)
			}
			return s, nil
		}
		if now := time.Now(); lastLog.IsZero() || now.Sub(lastLog) >= storeRetryLogEvery {
			logf("relay: deploy store unavailable after %d attempt(s): %v (retrying in %s; STUN is unaffected)", attempt, err, delay)
			lastLog = now
		}
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-time.After(delay):
		}
		delay *= 2
		if delay > storeRetryMax {
			delay = storeRetryMax
		}
	}
}

// Config is everything `carlos relay serve` needs from its box: who it is,
// where it listens from, what it may not forward to, and the store it reads
// its authority from and writes its heartbeat and metering to.
type Config struct {
	Box    string // the metrics box label; the key its heartbeat and meter objects live under
	Region string // the AWS region name, as the edges file spells it

	MastersPath string // /etc/carlos/relay-masters
	PublicIP    net.IP // the EIP advertised to clients as the relay address

	DenyCIDRs []*net.IPNet // peer ranges refused on top of the built-in private list
	Store     release.Store
	Limits    Limits

	// PressurePath is /proc/pressure/memory. A path that cannot be read
	// (an old kernel, a container without PSI) simply disables shedding —
	// the cgroup caps in the unit are still there — rather than failing.
	PressurePath string

	// udp and tcp are the listen addresses, unexported because they are a
	// test seam and nothing else: production is :3478 on both, per spec §3,
	// and no env may move them. Empty means the constants above.
	udp, tcp string

	// extraAllow is the same kind of seam, for the same reason. A relay on a
	// real box never permits a loopback peer, and an end-to-end test that
	// carries traffic between two sockets on this machine has nowhere else
	// to put them. Unexported so no env and no operator can widen the peer
	// policy; ServerConfig.ExtraAllow is the reviewed, production-facing one.
	extraAllow []*net.IPNet
}

// ConfigFromEnv builds a Config from the unit's environment. imdsIP and
// imdsRegion are the IMDS probes (host.ReadPublicIP, host.ReadPlacementRegion),
// consulted only when the matching override is unset, so a box off EC2 is
// configured entirely by host.env.
//
// Box is deliberately NOT read from IMDS or derived here: the label must be
// the one the metrics loop publishes under (host.BoxLabel), which lives in
// internal/host, so the caller fills it in when CARLOS_RELAY_BOX is unset.
func ConfigFromEnv(ctx context.Context, getenv func(string) string, imdsIP, imdsRegion func() string, store release.Store) (Config, error) {
	_ = ctx // the probes are the caller's; nothing here does I/O of its own
	if getenv == nil {
		getenv = func(string) string { return "" }
	}
	env := func(name string) string { return strings.TrimSpace(getenv(name)) }

	cfg := Config{
		Box:          env("CARLOS_RELAY_BOX"),
		MastersPath:  defaultMastersPath,
		PressurePath: defaultPressurePath,
		Store:        store,
		Limits:       LimitsFromEnv(getenv),
	}
	if p := env("CARLOS_RELAY_MASTERS_PATH"); p != "" {
		cfg.MastersPath = p
	}
	if p := env("CARLOS_RELAY_PRESSURE_PATH"); p != "" {
		cfg.PressurePath = p
	}

	// The region rule is edgesFileRegion's, spelled again here rather than
	// imported, because the two must agree: CARLOS_AWS_REGION wins outright,
	// otherwise the IMDS placement probe stands. Empty is not fatal — such a
	// relay matches no app's region list and serves STUN only, which is a
	// legitimate state, not a misconfiguration to die on.
	cfg.Region = env("CARLOS_AWS_REGION")
	if cfg.Region == "" && imdsRegion != nil {
		cfg.Region = strings.TrimSpace(imdsRegion())
	}

	ip := env("CARLOS_RELAY_PUBLIC_IP")
	if ip == "" && imdsIP != nil {
		ip = strings.TrimSpace(imdsIP())
	}
	if ip == "" {
		return Config{}, errors.New("relay: no public IP: set CARLOS_RELAY_PUBLIC_IP, or run on a box whose IMDS reports a public-ipv4")
	}
	cfg.PublicIP = net.ParseIP(ip)
	if cfg.PublicIP == nil {
		return Config{}, fmt.Errorf("relay: CARLOS_RELAY_PUBLIC_IP %q is not an IP address", ip)
	}

	// A typo in the deny list must stop the relay, not quietly open the VPC
	// to any browser that can reach 3478.
	for _, s := range strings.Split(env("CARLOS_RELAY_DENY_CIDRS"), ",") {
		s = strings.TrimSpace(s)
		if s == "" {
			continue
		}
		_, n, err := net.ParseCIDR(s)
		if err != nil {
			return Config{}, fmt.Errorf("relay: CARLOS_RELAY_DENY_CIDRS %q: %w", s, err)
		}
		cfg.DenyCIDRs = append(cfg.DenyCIDRs, n)
	}
	return cfg, nil
}

// Serve runs the relay until ctx is done, then closes every listener and
// allocation and returns nil. The only errors it returns are from starting:
// once it is listening, every failure is logged and retried on the next
// tick, because a relay that exits on a bad bucket read is a relay that
// stops answering STUN for a reason that has nothing to do with STUN.
func Serve(ctx context.Context, cfg Config, logf func(string, ...any)) error {
	if logf == nil {
		logf = func(string, ...any) {}
	}
	switch {
	case cfg.Box == "":
		return errors.New("relay: Config.Box is required (the metrics box label)")
	case cfg.Store == nil:
		return errors.New("relay: Config.Store is required")
	case cfg.PublicIP == nil:
		return errors.New("relay: Config.PublicIP is required")
	}
	if cfg.MastersPath == "" {
		cfg.MastersPath = defaultMastersPath
	}
	if cfg.Region == "" {
		logf("relay: no region: this box matches no app's region list and will serve STUN only")
	}

	// The four live views the server reads on every request. They are
	// written here, under mu, and read on pion's goroutines, so every
	// accessor takes the lock and hands back a value the server may hold.
	var (
		mu       sync.Mutex
		masters  *Masters
		mastersE string
		auth     *Authority
		lease    Lease
	)
	now := time.Now
	meter := NewMeter(cfg.Box, now, logf)

	// Each defaults on its own, so an empty tcp seam can never silently
	// take the UDP listener down with it (or the reverse): spec §3 says
	// both listeners, always, and a half-configured pair is a bug rather
	// than a supported shape. A test seams both or neither.
	udp, tcp := cfg.udp, cfg.tcp
	if udp == "" {
		udp = relayUDPAddr
	}
	if tcp == "" {
		tcp = relayTCPAddr
	}
	srv, err := NewServer(ServerConfig{
		Region:     cfg.Region,
		RelayIP:    cfg.PublicIP,
		UDP:        udp,
		TCP:        tcp,
		MinPort:    49152,
		MaxPort:    65535,
		DenyCIDRs:  cfg.DenyCIDRs,
		ExtraAllow: cfg.extraAllow,
		Limits:     cfg.Limits,
		Now:        now,
		Log:        logf,
		Meter:      meter,
		Masters:    func() *Masters { mu.Lock(); defer mu.Unlock(); return masters },
		Authority:  func() *Authority { mu.Lock(); defer mu.Unlock(); return auth },
		Lease:      func(t time.Time) bool { mu.Lock(); defer mu.Unlock(); return lease.Live(t) },
	})
	if err != nil {
		return err
	}
	defer srv.Close()
	logf("relay: listening on %v, relay address %s, region %q", srv.Listeners(), cfg.PublicIP, cfg.Region)

	// Rebuild the rolling-hour byte windows from what this box already
	// metered this hour and last (spec §3, "Limits"): a restart must not
	// hand an app that has spent its 20 GB a fresh budget.
	if samples, err := LoadWindow(ctx, cfg.Store, cfg.Box, now(), logf); err != nil {
		logf("relay: rolling-hour windows not rebuilt (%v); budgets start empty", err)
	} else {
		seedWindows(srv, samples)
	}

	go WatchMasters(ctx, cfg.MastersPath, func(m *Masters, err error) {
		mu.Lock()
		if err != nil {
			mastersE = err.Error()
		} else {
			masters, mastersE = m, ""
		}
		mu.Unlock()
		if err != nil {
			logf("relay: masters file %s: %v", cfg.MastersPath, err)
			return
		}
		// A reload can retire a master, and an allocation signed by it must
		// end now rather than at the next expiry sweep.
		srv.tick()
	})

	// readAuth is one authority read: it renews the lease on success only
	// (a failed read renews nothing, spec §3), ends what the new record no
	// longer allows, and writes the heartbeat that tells the console this
	// box is current.
	readAuth := func() {
		a, _, err := ReadAuthority(ctx, cfg.Store)
		t := now()
		listeners := srv.Listeners() // outside mu: it takes the server's own lock
		mu.Lock()
		if err == nil {
			auth = a
			lease.Renew(t)
		}
		cur, live := auth, lease.Live(t)
		hb := Heartbeat{
			Box:          cfg.Box,
			Region:       cfg.Region,
			ReadAt:       lease.Last(), // the last SUCCESSFUL read, never the wall clock
			Listeners:    listeners,
			Kids:         kidsOf(masters),
			Retiring:     retiringOf(masters),
			MastersError: mastersE,
		}
		if err != nil {
			hb.AuthorityError = err.Error()
		}
		if cur != nil {
			hb.WrittenAt = cur.WrittenAt
		}
		mu.Unlock()
		if err != nil {
			logf("relay: authority read: %v", err)
		}
		if !live {
			srv.EndAll("lease lapsed")
		}
		// cur is a snapshot, so the predicate runs off the lock without
		// racing the next read. A nil record means no app is enabled here
		// and every allocation goes, which is the "STUN only" state.
		srv.EndWhere(func(b *Binding) bool {
			x, ok := cur.Lookup(b.ID.Account, b.ID.App)
			return !ok || x.Incarnation != b.Incarnation || !slices.Contains(x.Regions, cfg.Region)
		}, "authority")
		if err := WriteHeartbeat(ctx, cfg.Store, hb); err != nil {
			logf("relay: heartbeat: %v", err)
		}
	}
	readAuth()

	flush := func(ctx context.Context) {
		// A complete census, not one id at a time: an app whose last
		// allocation ended has to be told its gauge is zero, or it reports
		// a phantom allocation forever and the meter never evicts it.
		meter.SetAllocsAll(srv.LiveByIdentity())
		if err := meter.Flush(ctx, cfg.Store); err != nil {
			logf("relay: meter flush: %v", err)
		}
	}

	authTick := time.NewTicker(authInterval)
	serialTick := time.NewTicker(serialInterval)
	meterTick := time.NewTicker(meterInterval)
	expiryTick := time.NewTicker(expiryInterval)
	psiTick := time.NewTicker(psiInterval)
	defer func() {
		for _, t := range []*time.Ticker{authTick, serialTick, meterTick, expiryTick, psiTick} {
			t.Stop()
		}
	}()

	var (
		lastSerial int64 = -1
		shedding   bool
	)
	for {
		select {
		case <-ctx.Done():
			// The last minute of metering is real money; give it its own
			// short deadline rather than losing it to the cancelled ctx.
			final, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
			flush(final)
			cancel()
			return nil
		case <-authTick.C:
			readAuth()
		case <-serialTick.C:
			// The relay polls control/serial itself: the agent's
			// fastConvergeLoop is a different process (spec §3).
			doc, _, err := release.ReadSerial(ctx, cfg.Store)
			switch {
			case err != nil:
				logf("relay: serial read: %v", err)
			case doc.Serial != lastSerial:
				lastSerial = doc.Serial
				readAuth()
			}
		case <-meterTick.C:
			flush(ctx)
		case <-expiryTick.C:
			srv.tick()
		case <-psiTick.C:
			some, err := readPSI(cfg.PressurePath)
			if err != nil {
				continue
			}
			// Spec §3 asks for shedding above 20% and a resumption of
			// admission under 10%. The Server has no admission pause to
			// lift — admission is governed by the limits, not by pressure —
			// so "resume" here is simply the end of shedding: once over 20%
			// the relay drops its oldest allocation every tick until
			// pressure falls under 10%, and then stops.
			switch {
			case some > shedAbove:
				shedding = true
			case some < shedUntil:
				shedding = false
			}
			if shedding {
				srv.ShedOldest(1)
			}
		}
	}
}

// seedWindows replays a window of metered samples into the admission
// counters. A sample's app key is "<account>/<app>" and an app name can
// never contain a slash (spec §2's grammar), so the first slash splits it.
func seedWindows(srv *Server, samples []MeterSample) {
	for _, s := range samples {
		for key, app := range s.Apps {
			account, name, ok := strings.Cut(key, "/")
			if !ok || account == "" || name == "" {
				continue
			}
			srv.Seed(Identity{Account: account, App: name}, s.At, app.In+app.Out)
		}
	}
}

// kidsOf is Masters.Kids with a nil guard: the heartbeat reports the kids
// currently loaded, and "none, because the file has never parsed" is a
// state the console wants to see rather than a panic.
func kidsOf(m *Masters) []string {
	if m == nil {
		return nil
	}
	return m.Kids()
}

// retiringOf is Masters.Retiring with the same nil guard kidsOf carries.
func retiringOf(m *Masters) map[string]time.Time {
	if m == nil {
		return nil
	}
	r := m.Retiring()
	if len(r) == 0 {
		return nil
	}
	return r
}

// readPSI reads the `some avg10` percentage out of a pressure-stall file
// (/proc/pressure/memory):
//
//	some avg10=0.00 avg60=0.00 avg300=0.00 total=0
//	full avg10=0.00 avg60=0.00 avg300=0.00 total=0
//
// Only the `some` line is used: it is the share of time ANY task stalled on
// memory, which is what "the box is under pressure" means for a unit that
// has to lose first. Plain file reading, so this compiles on the platforms
// the CLI cross-builds for and simply errors there.
func readPSI(path string) (float64, error) {
	f, err := os.Open(path)
	if err != nil {
		return 0, err
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	for sc.Scan() {
		fields := strings.Fields(sc.Text())
		if len(fields) == 0 || fields[0] != "some" {
			continue
		}
		for _, kv := range fields[1:] {
			v, ok := strings.CutPrefix(kv, "avg10=")
			if !ok {
				continue
			}
			n, err := strconv.ParseFloat(v, 64)
			if err != nil {
				return 0, fmt.Errorf("%s: some avg10=%q: %w", path, v, err)
			}
			return n, nil
		}
	}
	if err := sc.Err(); err != nil {
		return 0, err
	}
	return 0, fmt.Errorf("%s: no `some avg10=` field", path)
}