// Ambient release retention, the box half (issue #136, spec
// 2026-08-10-releases-retention-design.md §6): for each app this box has
// routes for, read the app's retention record from the bucket and delete
// what release.PruneEligible says is safe — protecting, on top of the
// bucket-derived safety list, the versions THIS box's registry still
// records as adopted, previous or good (the fallback). Boxes race each other harmlessly: every
// delete is idempotent by store contract.
//
// Hosted-apps-only is a ruling, not an accident: an app with routes on no
// box has nothing at stake, and whole-app trash→purge owns its end of life.
package host

import (
	"context"
	"crypto/ed25519"
	"errors"
	"fmt"
	"log"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
)

// retentionPass runs one enforcement sweep. Per-app isolation: one app's
// error is logged and skipped, never allowed to stop another app's
// enforcement; a partial delete leaves the remainder for the next tick.
//
// reports is where the box reports are read (settings spec rev 11 §4.4's
// build-state builds; retentionReports). A pass that cannot read them
// prunes nothing at all.
func retentionPass(ctx context.Context, store release.Store, reports *boxReports, pub ed25519.PublicKey, reg *registry.Registry) error {
	routes, err := reg.List()
	if err != nil {
		return err
	}
	if reports == nil {
		return errors.New("no box reports to read: pruning nothing")
	}
	boxes, complete, err := reports.load(ctx)
	if err != nil {
		return fmt.Errorf("reading the box reports: pruning nothing this pass: %w", err)
	}
	type appKey struct{ acct, app string }
	pins := map[appKey]map[string]bool{}
	for _, rt := range routes {
		if rt.App == "" {
			continue
		}
		k := appKey{release.RouteAccount(rt), rt.App}
		if pins[k] == nil {
			pins[k] = map[string]bool{}
		}
		if rt.Version != "" {
			pins[k][rt.Version] = true
		}
		if rt.PrevVersion != "" {
			pins[k][rt.PrevVersion] = true
		}
		// The good build is a planned wake's fallback (zero-downtime spec
		// §8.5), and it can be neither Version nor PrevVersion once two
		// releases have failed in a row. Pruned, the fallback has no artifact
		// exactly when a failed release needs it (zdd-readiness Task 13,
		// Astra R1 T12). Pinned whatever CARLOS_SWAP says, the plan's
		// authorized flag-off exception: keeping one more version is the
		// cheap side of the trade.
		if rt.GoodVersion != "" {
			pins[k][rt.GoodVersion] = true
		}
	}
	for k, protected := range pins {
		ret, err := release.GetRetention(ctx, store, k.acct, k.app)
		if err != nil {
			log.Printf("agent: retention %s/%s: %v", k.acct, k.app, err)
			continue
		}
		if ret == nil {
			continue
		}
		extra, err := settingsProtected(ctx, store, boxes, complete, time.Now(), k.acct, k.app)
		if err != nil {
			log.Printf("agent: retention %s/%s: pruning nothing this pass: %v", k.acct, k.app, err)
			continue
		}
		for v := range extra {
			protected[v] = true
		}
		eligible, err := release.PruneEligible(ctx, store, pub, k.acct, k.app, ret.Keep, time.Now(), protected)
		if err != nil {
			log.Printf("agent: retention %s/%s: %v", k.acct, k.app, err)
			continue
		}
		if len(eligible) == 0 {
			continue
		}
		deleted := 0
		for _, v := range eligible {
			if err := release.DeleteVersion(ctx, store, k.acct, k.app, v); err != nil {
				log.Printf("agent: retention %s/%s: delete %s: %v", k.acct, k.app, v, err)
				break
			}
			deleted++
		}
		log.Printf("agent: retention: pruned %d of %d eligible versions for %s/%s (keep %d)",
			deleted, len(eligible), k.acct, k.app, ret.Keep)
	}
	return nil
}

// buildStatePins is this box's contribution to retention (settings spec rev
// 11 §4.4): the build-state builds of every row keepsBuildState selects (the
// rows refreshAll keeps a copy of), from the registry copies the agent
// refreshes at reconcile and before every wake. claims is whether host
// claims are on, as keepsBuildState takes it. Every such row is reported,
// with or without builds, so a missing or unreadable copy shows as
// incomplete rather than as nothing to protect. A failed list is an error:
// the caller reports it (BoxMetrics.BuildStatesErr), never an empty map.
func buildStatePins(reg routeLister, claims bool) (map[string]status.BuildStatePins, error) {
	routes, err := reg.List()
	if err != nil {
		return nil, fmt.Errorf("listing routes for build states: %w", err)
	}
	out := map[string]status.BuildStatePins{}
	for _, rt := range routes {
		if !keepsBuildState(rt, claims) {
			continue
		}
		p := status.BuildStatePins{Account: release.RouteAccount(rt), App: rt.App, Builds: []string{}}
		c, err := release.ParseBuildStateCopy(rt.BuildState)
		p.Complete = err == nil && c.Known && c.ReadErr == ""
		if err == nil {
			p.Builds = buildStateBuilds(c.State)
		}
		out[rt.Host] = p
	}
	if len(out) == 0 {
		return nil, nil
	}
	return out, nil
}

// buildStateBuilds is the builds a build state names: last_executed, and a
// resolved request's target when it differs. A failed or completed request's
// T is named too: it may still be what the instance runs or recovers onto.
func buildStateBuilds(st *release.BuildState) []string {
	builds := []string{}
	if st == nil {
		return builds
	}
	if st.LastExecuted != "" {
		builds = append(builds, st.LastExecuted)
	}
	if r := st.Resolved; r != nil && r.Target != "" && r.Target != st.LastExecuted {
		builds = append(builds, r.Target)
	}
	return builds
}

// buildStateReportMaxAge is §4.4's bound: an app is not pruned while any box
// hosting one of its instances has been silent longer than this.
const buildStateReportMaxAge = 6 * time.Hour

// boxReports is where a retention pass reads the box reports (settings spec
// rev 11 §4.4). At least one of status and local is set.
type boxReports struct {
	// status is the store the boxes publish to (openStatusStore's bucket);
	// nil for a sole housekeeper with none.
	status release.Store
	// prefix rebases the platform's own reports in status, as
	// CARLOS_STATUS_PREFIX rebases what the status loop writes
	// (resolveStatusPrefix). Fleet reports are not rebased: a fleet box
	// writes at fleets/<fleet>/ from the bucket root.
	prefix string
	// deploy is the deployment store, read for fleets/ as well (ruling
	// R-T9-FLEETS): a fleet box with no bucket publishes through the store
	// door, which serves the deployment store. nil when it is the status
	// store's own bucket.
	deploy release.Store
	// local is this box's own collector, read as a fresh report at every
	// pass: set for a sole housekeeper (a directory-backed deployment, one
	// box by construction, which publishes no status; ruling R-T9-SOLE).
	local func() (map[string]status.BuildStatePins, error)
	// now stamps the local report; nil is time.Now.
	now func() time.Time
}

// soleReportLabel names a sole housekeeper's own report in refusals.
const soleReportLabel = "this box (sole housekeeper)"

// retentionReports is retentionLoop's report source, from the environment:
// the status store (rebased by CARLOS_STATUS_PREFIX) and, for fleets, the
// deployment store too when it is another bucket; or, on a sole
// housekeeper with no status store, this box's own collector and the
// deployment store's fleets. Anything else is an error: no app can be
// pruned without the reports.
func retentionReports(ctx context.Context, getenv func(string) string, deploy release.Store, local func() (map[string]status.BuildStatePins, error)) (*boxReports, error) {
	st, _, err := openStatusStore(ctx, getenv)
	switch {
	case err == nil:
		r := &boxReports{status: st, prefix: resolveStatusPrefix(getenv), deploy: deploy}
		if b := getenv("CARLOS_DEPLOYMENT_BUCKET"); b != "" && b == resolveStatusBucket(getenv) {
			r.deploy = nil // the same bucket: its fleets/ is read once
		}
		return r, nil
	case errors.Is(err, errNoStatusStore) && newHousekeeper(getenv, nil).mode == housekeeperSole && local != nil:
		return &boxReports{deploy: deploy, local: local}, nil
	}
	return nil, err
}

// load reads every box report: the platform's (status/boxes/ under prefix,
// or this box's own on a sole housekeeper), and every fleet's in the status
// store and in the deployment store, the union (a report in both is read
// twice; the newest-per-host rule is indifferent to that). complete is
// false if any report or fleets listing could not be read. Failing to list
// the platform's own reports is an error.
func (r *boxReports) load(ctx context.Context) ([]status.Box, bool, error) {
	if r == nil || (r.status == nil && r.local == nil) {
		return nil, false, errors.New("no box reports to read")
	}
	var boxes []status.Box
	complete := true
	if r.status != nil {
		root, ok, err := status.LoadBoxesComplete(ctx, release.NewPrefixStore(r.status, r.prefix))
		if err != nil {
			return nil, false, err
		}
		boxes, complete = root, ok
	}
	if r.local != nil {
		boxes = append(boxes, r.localReport())
	}
	for _, s := range []release.Store{r.status, r.deploy} {
		if s == nil {
			continue
		}
		fb, ok := loadFleetReports(ctx, s)
		complete = complete && ok
		boxes = append(boxes, fb...)
	}
	return boxes, complete, nil
}

// localReport is this box's own report, taken now, as the status loop
// would publish it: BuildStatesErr when the collector fails.
func (r *boxReports) localReport() status.Box {
	now := time.Now
	if r.now != nil {
		now = r.now
	}
	m := &status.BoxMetrics{}
	if pins, err := r.local(); err != nil {
		m.BuildStatesErr = err.Error()
	} else {
		m.BuildStates = pins
	}
	return status.Box{Label: soleReportLabel, GeneratedAt: now(), Metrics: m}
}

// loadFleetReports reads every fleet's reports in store
// (fleets/<name>/status/boxes/, enumerated with ListDirs). ok is false when
// the listing or any report could not be read.
func loadFleetReports(ctx context.Context, store release.Store) ([]status.Box, bool) {
	fleets, err := store.ListDirs(ctx, "fleets/")
	if err != nil {
		log.Printf("agent: retention: listing fleets: %v", err)
		return nil, false
	}
	var boxes []status.Box
	ok := true
	for _, f := range fleets {
		fb, complete, err := status.LoadBoxesComplete(ctx, release.NewPrefixStore(store, f))
		if err != nil {
			log.Printf("agent: retention: reading the reports of %s: %v", f, err)
			ok = false
			continue
		}
		ok = ok && complete
		boxes = append(boxes, fb...)
	}
	return boxes, ok
}

// appRecords reads every provision record of acct/app, or fails. It lists
// the keys itself (provision.ListRecords caps on max and skips a record it
// cannot read, ruling R-T9-STORE): a key that is a record by ListRecords'
// own rule (instances/<host>.json, one level) and does not read, parse or
// agree with its path is an error, so the caller prunes nothing.
func appRecords(ctx context.Context, store release.Store, acct, app string) ([]provision.Record, error) {
	prefix := provision.RecordsPrefix(acct, app)
	keys, err := store.List(ctx, prefix)
	if err != nil {
		return nil, fmt.Errorf("listing the records of %s/%s: %w", acct, app, err)
	}
	var recs []provision.Record
	for _, key := range keys {
		rest, ok := strings.CutPrefix(key, prefix)
		if !ok || rest == "" || strings.Contains(rest, "/") {
			continue
		}
		host, ok := strings.CutSuffix(rest, ".json")
		if !ok || host == "" {
			continue
		}
		rec, err := provision.GetRecord(ctx, store, acct, app, host)
		if err != nil {
			return nil, fmt.Errorf("reading the record of %s: %w", host, err)
		}
		recs = append(recs, *rec)
	}
	return recs, nil
}

// settingsProtected is settings spec rev 11 §4.4's addition to an app's
// protected set: every pin a provision record names, and every build any
// box report names for one of the app's instances. It refuses (the caller
// prunes nothing for the app) when a record is unreadable, when the reports
// could not all be read, when a fresh report says its box could not collect
// its build states, when a provisioned instance that keeps a build state is
// named by no report, or when the newest report naming one of the app's
// hosts is older than six hours or says that host's build state is unread.
//
// The newest report for a host decides freshness and completeness, so a box
// the instance moved off does not block pruning forever. The builds of
// every report naming the host are protected, older ones included: that can
// only keep more.
func settingsProtected(ctx context.Context, store release.Store, boxes []status.Box, complete bool, now time.Time, acct, app string) (map[string]bool, error) {
	recs, err := appRecords(ctx, store, acct, app)
	if err != nil {
		return nil, err
	}
	if !complete {
		return nil, errors.New("the box reports could not all be read")
	}
	out := map[string]bool{}
	for _, r := range recs {
		if r.Pin != "" {
			out[r.Pin] = true
		}
	}
	type newest struct {
		at   time.Time
		pins status.BuildStatePins
		box  string
	}
	byHost := map[string]newest{}
	for _, b := range boxes {
		if b.Metrics == nil {
			continue
		}
		// A box that could not collect its build states may host any of
		// this app's instances (Astra I2 #5).
		if b.Metrics.BuildStatesErr != "" && now.Sub(b.GeneratedAt) <= buildStateReportMaxAge {
			return nil, fmt.Errorf("%s could not collect its build states: %s", b.Label, b.Metrics.BuildStatesErr)
		}
		for host, p := range b.Metrics.BuildStates {
			if p.Account != acct || p.App != app {
				continue
			}
			for _, v := range p.Builds {
				if v != "" {
					out[v] = true
				}
			}
			if cur, ok := byHost[host]; !ok || b.GeneratedAt.After(cur.at) {
				byHost[host] = newest{at: b.GeneratedAt, pins: p, box: b.Label}
			}
		}
	}
	// Every provisioned instance of the app that keeps a build state must
	// be named by some report (Astra I2 #5): a missing entry is unknown,
	// never "nothing to protect". For a valid record this is
	// keepsBuildState's instance rule (activator.ActivatorManaged): an
	// instance with no unit that hibernates has no addr (the record door
	// refuses addr with hibernate), so reconcile mints it a socket.
	for _, r := range recs {
		if r.KindOrDefault() != "instance" || r.Unit != "" || !r.HibernateOrDefault() {
			continue
		}
		if _, ok := byHost[r.Host]; !ok {
			return nil, fmt.Errorf("no box report names the build state of %s", r.Host)
		}
	}
	for host, n := range byHost {
		if now.Sub(n.at) > buildStateReportMaxAge {
			return nil, fmt.Errorf("%s, which hosts %s, has not reported since %s", n.box, host, n.at.UTC().Format(time.RFC3339))
		}
		if !n.pins.Complete {
			return nil, fmt.Errorf("%s could not read the build state of %s", n.box, host)
		}
	}
	return out, nil
}