package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math/rand/v2"
	"os"
	"path/filepath"
	"sort"
	"time"

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

// The cold-standby spec (2026-08-12 HA posture, Part I rung 1): the current
// snapshot at a per-box key a runbook can name (snapshotDBKey), a sidecar
// describing it, and a short tail of prior generations. All under control/
// beside control/serial; platform boxes write direct S3, so no store-door
// ACL is involved.
const (
	snapshotKeep  = 8
	snapshotStamp = "20060102T150405Z"
)

// Per-box keys (spec 2026-08-20 Part 1, closes #158): every platform box
// snapshots ITS OWN registry under its own name; boxes never contend.
// The pre-multi-region fixed keys ("control/registry-snapshot.db" etc.)
// are still present in production buckets until both deployments roll —
// they are abandoned, not rewritten; deletion is a rollout step, not code.
func snapshotDBKey(box string) string   { return "control/registry-snapshot/" + box + ".db" }
func snapshotMetaKey(box string) string { return "control/registry-snapshot/" + box + ".json" }
func snapshotGenPrefix(box string) string {
	return "control/registry-snapshots/" + box + "/"
}

// snapshotMeta describes the object at snapshotDBKey(box). `Box` is the writing
// box's label (per-instance hashed machine id) — a restore must check it
// names the DEAD box, not the standby, or it is restoring the standby's own
// empty registry. `Fingerprint` carries registry.SnapshotFingerprint (ALL
// columns), not registry.Fingerprint (edge-reload columns only): it is only
// ever compared against itself — the loop seeds lastFP from it and skips an
// unchanged registry — so no reader outside this file depends on which
// digest it is.
type snapshotMeta struct {
	V           int       `json:"v"`
	Box         string    `json:"box"`
	Fingerprint string    `json:"fingerprint"`
	CreatedAt   time.Time `json:"created_at"`
	Bytes       int       `json:"bytes"`
}

// registrySnapshotPass snapshots the registry into the bucket when its
// snapshot fingerprint moved since *lastFP. One pass; the loop owns cadence.
// *lastFP is updated only after every write (including pruning) succeeded, so
// a partial failure is retried whole next tick.
//
// The gate is SnapshotFingerprint, NOT Fingerprint: Fingerprint deliberately
// omits caps, unit, replication, environment, account, canary and
// provisioned, which are precisely the box-scoped columns only this snapshot
// can rebuild — gating on it meant a caps grant or a console unit stamp never
// produced a snapshot at all.
func registrySnapshotPass(ctx context.Context, s release.Store, reg *registry.Registry, box string, now time.Time, lastFP *string) error {
	fp, err := reg.SnapshotFingerprint()
	if err != nil {
		return fmt.Errorf("snapshot: fingerprint: %w", err)
	}
	if fp == *lastFP {
		return nil
	}

	dir, err := os.MkdirTemp("", "carlos-registry-snapshot")
	if err != nil {
		return fmt.Errorf("snapshot: %w", err)
	}
	defer os.RemoveAll(dir)
	file := filepath.Join(dir, "registry.db")
	if err := reg.SnapshotTo(file); err != nil {
		return fmt.Errorf("snapshot: vacuum: %w", err)
	}
	body, err := os.ReadFile(file)
	if err != nil {
		return fmt.Errorf("snapshot: %w", err)
	}

	// Generation first (create-only — a same-second re-run or a clock
	// step-back can collide on the stamp, in which case Put's ErrCASConflict
	// is deliberately swallowed and the retained generation is KEPT rather
	// than replaced). Be precise about what that costs: after a clock
	// step-back WITH a registry change in between, the retained generation
	// holds OLDER content while the current key and its sidecar hold the
	// newer — the two are not equivalent, and we are choosing to keep the
	// older one under that stamp. What is lost is one intermediate rung of
	// the generation tail, never the restore target itself: current + sidecar
	// are written unconditionally below, so a restore still gets the newest
	// registry. Retrying under a bumped stamp would buy a slightly denser
	// tail at the cost of a write loop against a lying clock.
	//
	// Then current + sidecar (replace with the CAS-retry recipe
	// usage.UpsertEdgeDay uses), then prune.
	genKey := snapshotGenPrefix(box) + now.UTC().Format(snapshotStamp) + ".db"
	if _, err := s.Put(ctx, genKey, body, ""); err != nil && !errors.Is(err, release.ErrCASConflict) {
		return fmt.Errorf("snapshot: put %s: %w", genKey, err)
	}
	// This guard predates per-box keys and is retained for the shared-key
	// era and for hand-placed sidecars only (a sidecar copied or edited by
	// an operator with a Box field that does not match this box). Under
	// per-box keys it reads ONLY this box's own key, and the only writer
	// that ever lands there stamps meta.Box = box — so prior.Box != box
	// cannot occur through normal operation, INCLUDING the same-name
	// standby-restore case the runbook worries about (a rebuilt standby
	// reusing the dead box's machine-id before restoring: same name means
	// prior.Box already equals box, so this stays silent). That clobber —
	// the freshly bootstrapped standby's agent starting and writing its own
	// empty registry over the dead box's last-good snapshot before the
	// operator restores it — is NOT detected here; the DR runbook's Step 0
	// ordering rule (restore before the agent is allowed to snapshot) owns
	// that hazard instead.
	if prior := readSnapshotMeta(ctx, s, box); prior != nil && prior.Box != "" && prior.Box != box {
		log.Printf("agent: registry snapshot: OVERWRITING a snapshot written by box %q (this box is %q, prior created_at %s) — if this box is a standby that has not been restored yet, STOP and preserve the prior snapshot",
			prior.Box, box, prior.CreatedAt.Format(time.RFC3339))
	}
	if err := putReplacing(ctx, s, snapshotDBKey(box), body); err != nil {
		return err
	}
	meta := snapshotMeta{V: 1, Box: box, Fingerprint: fp, CreatedAt: now.UTC(), Bytes: len(body)}
	metaRaw, err := json.Marshal(meta)
	if err != nil {
		return fmt.Errorf("snapshot: %w", err)
	}
	if err := putReplacing(ctx, s, snapshotMetaKey(box), metaRaw); err != nil {
		return err
	}

	gens, err := s.List(ctx, snapshotGenPrefix(box))
	if err != nil {
		return fmt.Errorf("snapshot: list generations: %w", err)
	}
	sort.Strings(gens) // stamps are lexically chronological
	for len(gens) > snapshotKeep {
		if err := s.Delete(ctx, gens[0]); err != nil {
			return fmt.Errorf("snapshot: prune %s: %w", gens[0], err)
		}
		gens = gens[1:]
	}

	*lastFP = fp
	return nil
}

// seedSnapshotFingerprint reads the sidecar a previous run (or a previous
// process on this same box) left at snapshotMetaKey(box) and returns its
// fingerprint, so a freshly started loop's in-memory lastFP starts where the
// bucket already is rather than empty. Without this, a crash-looping agent
// would re-snapshot an unchanged registry on every restart and burn through
// all snapshotKeep generations with identical copies. Any read or parse
// failure — no sidecar yet, a corrupt object, a store that is down — returns
// "": the loop then snapshots on its first tick, the same behavior a box
// that has never snapshotted before gets. This must never fail the loop.
func seedSnapshotFingerprint(ctx context.Context, s release.Store, box string) string {
	meta := readSnapshotMeta(ctx, s, box)
	if meta == nil {
		return ""
	}
	return meta.Fingerprint
}

// readSnapshotMeta returns the sidecar currently in the bucket at this box's
// own key, or nil when there is none, it cannot be read, or it does not
// parse. Advisory only: both callers treat nil as "no prior snapshot" and
// carry on, so a store outage must never fail a pass.
func readSnapshotMeta(ctx context.Context, s release.Store, box string) *snapshotMeta {
	raw, _, err := s.Get(ctx, snapshotMetaKey(box))
	if err != nil {
		return nil
	}
	var meta snapshotMeta
	if err := json.Unmarshal(raw, &meta); err != nil {
		return nil
	}
	return &meta
}

// registrySnapshotLoop snapshots the registry to the deployment bucket every
// 5 minutes + jitter when it changed — rung 1 of the HA ladder (cold
// standby restores this file). Platform-box job, retentionLoop's rules:
// skipped on fleet boxes, disabled with a log line when there is no bucket.
// Per-box keys (spec 2026-08-20 Part 1, closes #158): each platform box
// writes under its own name, so N boxes sharing one deployment bucket never
// interleave each other's current/sidecar writes.
func registrySnapshotLoop(reg *registry.Registry) {
	ctx := context.Background()
	if shouldSkipPublisherElection(os.Getenv) {
		log.Printf("agent: registry snapshot loop is a platform-box job — skipped on a fleet box")
		return
	}
	store, err := openDeployStore(ctx)
	if err != nil {
		log.Printf("agent: registry snapshot loop disabled: %v", err)
		return
	}
	box := metricsBoxLabel(os.Getenv, readMachineID())
	log.Printf("agent: registry snapshot loop armed (5min + jitter, %s)", snapshotDBKey(box))
	lastFP := seedSnapshotFingerprint(ctx, store, box)
	delay := time.Minute
	for {
		time.Sleep(delay)
		delay = 5*time.Minute + rand.N(time.Minute)
		if err := registrySnapshotPass(ctx, store, reg, box, time.Now(), &lastFP); err != nil {
			log.Printf("agent: registry snapshot pass: %v", err)
		}
	}
}

// putReplacing writes key regardless of prior state: create-only when
// absent, etag-replace when present, re-reading on a lost race. Three
// attempts, like usage.UpsertEdgeDay.
func putReplacing(ctx context.Context, s release.Store, key string, body []byte) error {
	for attempt := 0; attempt < 3; attempt++ {
		_, etag, err := s.Get(ctx, key)
		switch {
		case errors.Is(err, os.ErrNotExist):
			etag = ""
		case err != nil:
			return fmt.Errorf("snapshot: get %s: %w", key, err)
		}
		if _, err := s.Put(ctx, key, body, etag); err == nil {
			return nil
		} else if !errors.Is(err, release.ErrCASConflict) {
			return fmt.Errorf("snapshot: put %s: %w", key, err)
		}
	}
	return fmt.Errorf("snapshot: %s: lost CAS race three times", key)
}