// The pin set (zero-downtime deploys spec §4.2, §8.5): every staged copy
// something on this box still refers to. A staged copy in it is never
// overwritten, because staged paths are keyed by version name with no digest
// and another instance's good build or running generation would change
// underneath it.
package release

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"

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

// StagedPaths is a set of staged artifact paths (VersionPath values, held
// cleaned) that something on this box refers to.
type StagedPaths map[string]bool

func (sp StagedPaths) add(p string) { sp[filepath.Clean(p)] = true }

// Has reports whether p is referred to.
func (sp StagedPaths) Has(p string) bool { return sp[filepath.Clean(p)] }

// addReal adds p and, when p exists, the fully resolved real path it names
// (filepath.EvalSymlinks), so a copy reached through a symlinked directory is
// pinned under the name the OS gives it too (Astra r3 N4, ruling R19 item 5).
// A p that does not exist adds p alone; any other resolution error is
// returned, and the caller's pin set is incomplete.
func (sp StagedPaths) addReal(p string) error {
	sp.add(p)
	real, err := filepath.EvalSymlinks(p)
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("resolve %s: %w", p, err)
	}
	sp.add(real)
	return nil
}

// Refers is the membership check a restage asks: whether p, or the fully
// resolved real path of p when p exists, is referred to. Comparing the real
// path as well means a symlinked opt dir cannot make a pinned file look
// unpinned under its other name (Astra r3 N4). A resolution error other than
// not existing is returned, and never permits a restage.
func (sp StagedPaths) Refers(p string) (bool, error) {
	if sp.Has(p) {
		return true, nil
	}
	real, err := filepath.EvalSymlinks(p)
	if errors.Is(err, os.ErrNotExist) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("resolve %s: %w", p, err)
	}
	return sp.Has(real), nil
}

// PinSource supplies references beyond the live links and the registry rows.
// GenerationPins is the one this package ships. A source that cannot answer
// returns an error, and an error never permits a restage.
type PinSource interface {
	StagedPins(ctx context.Context, optDir string) (StagedPaths, error)
}

// PinSourceFunc adapts a function to PinSource.
type PinSourceFunc func(ctx context.Context, optDir string) (StagedPaths, error)

func (f PinSourceFunc) StagedPins(ctx context.Context, optDir string) (StagedPaths, error) {
	return f(ctx, optDir)
}

// BoxPins is the part of the pin set this package reads itself: every hop of
// the symlink chain of every entry under <opt>/live (linkChain; a crashed
// swap's <host>.tmp included, as LivePins includes it), so a live link that
// reaches a staged copy through another link still pins that copy (a
// relative hop pinned at the real path the OS reaches, and at its lexical
// join too when the two differ), plus the fully resolved real path of the
// chain's end when it exists (StagedPaths.addReal), and the staged path of
// every row's Version, PrevVersion and GoodVersion (the good build a failed
// first wake falls back to, spec §8.5). routes is every registry row, unfiltered: a hibernating route,
// an expired canary and a mid-provision row all still refer to their builds.
//
// It also holds every build a row's update settings and build-state copy
// name (SettingsRefs: a pin, last_executed, a resolved request's T), under
// the row's own account and app (settings spec §4.1 retained identity;
// Astra round 1 finding 2): a sleeper whose build state alone names a build
// keeps that copy, executable and sibling, when another instance adopts a
// reuse of its name.
//
// An entry that cannot be read is an error, never a skipped entry: a pin set
// that quietly narrowed is how a running build's bytes get replaced. A row
// whose settings references cannot be read completely is reported apart, as
// an *IncompleteRefsError returned WITH the pins: it refuses a restage of its
// own app only, so one app's unread build state does not stop another's.
func BoxPins(optDir string, routes []registry.Route) (StagedPaths, error) {
	pins := StagedPaths{}
	dir := liveDir(optDir)
	entries, err := os.ReadDir(dir)
	if err != nil && !os.IsNotExist(err) {
		return nil, fmt.Errorf("read %s: %w", dir, err)
	}
	for _, e := range entries {
		_, hops, aliases, err := linkChain(filepath.Join(dir, e.Name()))
		if err != nil {
			return nil, err
		}
		for _, h := range hops {
			pins.add(h)
		}
		for _, h := range aliases {
			pins.add(h)
		}
		// And the fully resolved real path of where the chain ends (Astra r3
		// N4, ruling R19 item 5): a hop written through a symlinked directory
		// names the copy under another spelling than VersionPath's.
		if err := pins.addReal(hops[len(hops)-1]); err != nil {
			return nil, fmt.Errorf("live link %s: %w", filepath.Join(dir, e.Name()), err)
		}
	}
	var incomplete *IncompleteRefsError
	for _, rt := range routes {
		if rt.App == "" {
			continue
		}
		acct := RouteAccount(rt)
		refs, rerr := SettingsRefs(rt)
		for _, v := range append([]string{rt.Version, rt.PrevVersion, rt.GoodVersion}, refs...) {
			if ValidVersion(v) {
				pins.add(VersionPath(optDir, acct, v, rt.App))
			}
		}
		if rerr != nil {
			if incomplete == nil {
				incomplete = &IncompleteRefsError{apps: map[string]string{}}
			}
			incomplete.apps[acct+"/"+rt.App] = rerr.Error()
		}
	}
	if incomplete != nil {
		return pins, incomplete
	}
	return pins, nil
}

// IncompleteRefsError is BoxPins' report of apps whose settings references
// (SettingsRefs) could not be read completely. BoxPins returns it together
// with the pins it did read: a restage of a copy of a named app is refused,
// because a held instance's last_executed may be the very name being reused;
// any other app restages on the pins as usual.
type IncompleteRefsError struct {
	apps map[string]string // "<account>/<app>" -> why
}

func (e *IncompleteRefsError) Error() string {
	keys := make([]string, 0, len(e.apps))
	for k := range e.apps {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	parts := make([]string, 0, len(keys))
	for _, k := range keys {
		parts = append(parts, k+": "+e.apps[k])
	}
	return "settings references incomplete for " + strings.Join(parts, "; ")
}

// Covers reports whether acct/app's references are incomplete.
func (e *IncompleteRefsError) Covers(acct, app string) bool {
	_, ok := e.apps[acct+"/"+app]
	return ok
}

// SettingsRefs is every build a row's update settings and build-state copy
// name (settings spec §4.4, §4.1): its pin, last_executed and a resolved
// request's T, read as root reads the columns (ParseUpdateSettingsLenient,
// ParseBuildStateCopy). A copy that kept its last known state after a failed
// refresh still names it.
//
// The error says the list is incomplete, and comes with what could be read:
// a column that does not parse; or a row settings hold (non-default settings
// on a row whose build state the agent refreshes, BuildStateKept) whose copy
// is not read yet, or whose latest refresh failed (ReadErr; Astra round 1
// finding 6): its last_executed may name another build by now. A row the
// agent never refreshes (a named unit, the platform's own row, a worker)
// stays unread forever, and must not make every set incomplete for good.
func SettingsRefs(rt registry.Route) ([]string, error) {
	var out []string
	u, uerr := ParseUpdateSettingsLenient(rt.UpdateSettings)
	if uerr == nil && u.Pin != "" {
		out = append(out, u.Pin)
	}
	c, cerr := ParseBuildStateCopy(rt.BuildState)
	if cerr == nil && c.State != nil {
		if c.State.LastExecuted != "" {
			out = append(out, c.State.LastExecuted)
		}
		if r := c.State.Resolved; r != nil && r.Target != "" {
			out = append(out, r.Target)
		}
	}
	switch {
	case uerr != nil:
		return out, fmt.Errorf("update settings of %s: %w", rt.Host, uerr)
	case cerr != nil:
		return out, fmt.Errorf("build state of %s: %w", rt.Host, cerr)
	case u.Default() || !BuildStateKept(rt):
		return out, nil
	case !c.Known:
		return out, fmt.Errorf("the build state of held %s is not read yet", rt.Host)
	case c.ReadErr != "":
		return out, fmt.Errorf("the build state of held %s could not be read at its last refresh (%s)", rt.Host, c.ReadErr)
	}
	return out, nil
}

// BuildStateKept is a row update settings can hold whose build state the
// agent keeps and refreshes at every reconcile: an activator-managed exec
// instance (Kind instance, hibernating, a socket, no Backing, no Tunnel), not
// a named unit and not the platform's own row. It is the host's
// keepsBuildState less workers, which settings never apply to; this package
// cannot import the activator, so the condition is spelled out here and the
// host's sweep uses this one.
func BuildStateKept(rt registry.Route) bool {
	return rt.Kind == "instance" && rt.Unit == "" && !rt.Platform &&
		rt.Hibernate && rt.Socket != "" && rt.Backing == "" && rt.Tunnel == ""
}

// Generations is the generations file: for each host with a live or pending
// generation (current, successor, retiring, spawning: spec §8.5), the build
// each one runs. Live generations exist only in the agent's memory, while the
// pin set is also needed by root's pass, a separate process; this file is how
// root learns them (index amendments, 2026-10-07). The agent's activator
// writes it with WriteGenerations, listing every worker's captured serving
// build and every build a spawn is about to exec, before that exec and
// whenever a spawn or stop changes the set (rulings R17, R19); plan 5 extends
// the same writer to its operation set.
//
// A build is named by version (Hosts, placed under the host's row by
// GenerationPins) and by the absolute staged path it is exec'd from (Paths,
// pinned as written, no row needed). Paths covers a build whose Version is
// empty, such as one reached through an operator's chained link, which a
// version alone cannot name (ruling R19 item 2).
//
// On disk: {"schema":1,"seq":N,"complete":true,"hosts":{"<host>":["<version>",...]},
// "paths":{"<host>":["/abs/path",...]}}. A file with only hosts, only paths,
// or no seq (as first written, ruling R17) still reads.
//
// Publication is two atomic writes (ruling R19 item 4): first the intent
// file, GenerationsSeqPath, holding the publication's sequence N, then this
// file carrying seq N and complete:true. A publication that fails between
// them leaves the intent ahead of the file, and the reader then refuses the
// older file instead of trusting it.
type Generations struct {
	Schema   int                 `json:"schema"` // 1
	Seq      uint64              `json:"seq,omitempty"`
	Complete bool                `json:"complete,omitempty"`
	Hosts    map[string][]string `json:"hosts"`
	Paths    map[string][]string `json:"paths,omitempty"`
}

// GenerationsPath is the generations file under a data dir:
// /data/zdd/generations.json in production. The data dir is what every
// production pass already passes as BakeDir, and the one place both the
// sandboxed agent and root can write and read.
func GenerationsPath(dataDir string) string {
	return filepath.Join(dataDir, "zdd", "generations.json")
}

// GenerationsSeqPath is the publication intent file beside the generations
// file at path: /data/zdd/generations.seq in production. It holds the
// sequence of the latest publication begun, as a decimal number.
func GenerationsSeqPath(path string) string {
	return filepath.Join(filepath.Dir(path), "generations.seq")
}

// ReadGenerations reads the generations file and its intent file, the one
// reader of both (GenerationPins and the version sweep's GenerationVersions
// use it). Neither file present is (nil, nil): no generations, which is every
// box before its agent first published. With the intent present, the file
// must exist, read, parse, and carry complete:true and the intent's sequence;
// anything else is a publication that did not finish, and an error. A file
// with no intent is read as first written (ruling R17) only if it carries no
// sequence. Any error means the caller's pin set is incomplete.
func ReadGenerations(path string) (*Generations, error) {
	seqPath := GenerationsSeqPath(path)
	intent, hasIntent, err := readGenerationsSeq(seqPath)
	if err != nil {
		return nil, err
	}
	raw, err := safefile.ReadFile(path)
	if errors.Is(err, os.ErrNotExist) {
		if !hasIntent {
			return nil, nil
		}
		return nil, fmt.Errorf("generations file %s: absent while %s names publication %d: incomplete", path, seqPath, intent)
	}
	if err != nil {
		return nil, fmt.Errorf("generations file %s: %w", path, err)
	}
	var g Generations
	if err := json.Unmarshal(raw, &g); err != nil {
		return nil, fmt.Errorf("generations file %s: %v", path, err)
	}
	if g.Schema != 1 {
		return nil, fmt.Errorf("generations file %s: unsupported schema %d", path, g.Schema)
	}
	if err := validGenerations(g); err != nil {
		return nil, fmt.Errorf("generations file %s: %w", path, err)
	}
	switch {
	case hasIntent && (!g.Complete || g.Seq != intent):
		return nil, fmt.Errorf("generations file %s: publication %d (complete %t) while %s names publication %d: incomplete", path, g.Seq, g.Complete, seqPath, intent)
	case !hasIntent && (g.Seq != 0 || g.Complete):
		return nil, fmt.Errorf("generations file %s: publication %d with no intent file %s: incomplete", path, g.Seq, seqPath)
	}
	return &g, nil
}

// readGenerationsSeq reads the intent file: (0, false, nil) when it is
// absent, an error when it cannot be read or does not hold a sequence of 1 or
// more.
func readGenerationsSeq(seqPath string) (uint64, bool, error) {
	raw, err := safefile.ReadFile(seqPath)
	if errors.Is(err, os.ErrNotExist) {
		return 0, false, nil
	}
	if err != nil {
		return 0, false, fmt.Errorf("generations intent %s: %w", seqPath, err)
	}
	n, err := strconv.ParseUint(strings.TrimSpace(string(raw)), 10, 64)
	if err != nil || n == 0 {
		return 0, false, fmt.Errorf("generations intent %s: %q is not a publication sequence: incomplete", seqPath, raw)
	}
	return n, true, nil
}

// nextGenerationsSeq is the sequence for the next publication: one more than
// the larger of the intent on disk and the file's own seq, so it is
// monotonic across agent restarts. An intent that cannot be read is an error
// (no publication); one that reads but does not parse, or a file that does
// not read, counts as 0, because the file on disk can never carry the new
// sequence either way.
func nextGenerationsSeq(path string) (uint64, error) {
	var n uint64
	seqPath := GenerationsSeqPath(path)
	raw, err := safefile.ReadFile(seqPath)
	switch {
	case err == nil:
		if v, perr := strconv.ParseUint(strings.TrimSpace(string(raw)), 10, 64); perr == nil {
			n = v
		}
	case !errors.Is(err, os.ErrNotExist):
		return 0, fmt.Errorf("generations intent %s: %w", seqPath, err)
	}
	if body, err := safefile.ReadFile(path); err == nil {
		var g Generations
		if json.Unmarshal(body, &g) == nil && g.Seq > n {
			n = g.Seq
		}
	}
	return n + 1, nil
}

// WriteGenerations publishes g (ruling R19 item 4). It validates g first, so
// a refused g touches neither file. Then it writes the intent file holding
// the next sequence N, and then the generations file carrying seq N and
// complete:true. Each write is atomic: a temp file in the same directory,
// fsynced, mode 0644 (root and the agent both read them), then renamed over
// the old one, so a reader sees the old file or the new one and never a
// partial write. A failure of either write is returned; after a failed second
// write the intent is ahead of the file, which the reader refuses. g's own
// Schema, Seq and Complete are ignored.
func WriteGenerations(path string, g Generations) error {
	g.Schema = 1
	if g.Hosts == nil {
		g.Hosts = map[string][]string{}
	}
	if err := validGenerations(g); err != nil {
		return err
	}
	if err := safefile.MkdirAll(filepath.Dir(path), 0o755); err != nil {
		return err
	}
	seq, err := nextGenerationsSeq(path)
	if err != nil {
		return err
	}
	if err := writeFileAtomic(GenerationsSeqPath(path), []byte(strconv.FormatUint(seq, 10)+"\n")); err != nil {
		return err
	}
	g.Seq, g.Complete = seq, true
	body, err := json.MarshalIndent(g, "", "  ")
	if err != nil {
		return err
	}
	return writeFileAtomic(path, append(body, '\n'))
}

// writeFileAtomic replaces path with body: a temp file in path's directory,
// written, mode 0644, fsynced, closed, then renamed over path.
func writeFileAtomic(path string, body []byte) error {
	return safefile.WriteFile(path, body, 0o644)
}

func validGenerations(g Generations) error {
	for host, vers := range g.Hosts {
		for _, v := range vers {
			if !ValidVersion(v) {
				return fmt.Errorf("host %s names invalid version %q", host, v)
			}
		}
	}
	for host, paths := range g.Paths {
		for _, p := range paths {
			if p == "" || !filepath.IsAbs(p) {
				return fmt.Errorf("host %s names staged path %q, which is not absolute", host, p)
			}
		}
	}
	return nil
}

// GenerationPins is the PinSource for the generations file at path. Each
// listed version is placed under its host's own account and app, from rows,
// and each listed path is pinned as written (cleaned), with the real path it
// resolves to when it exists, and needs no row. A file ReadGenerations
// refuses, a listed version whose host has no row to place it, or a path
// whose resolution fails other than by not existing, is an error: those
// bytes are running somewhere this pass cannot name, so nothing may be
// restaged.
func GenerationPins(path string, rows func() ([]registry.Route, error)) PinSource {
	return PinSourceFunc(func(_ context.Context, optDir string) (StagedPaths, error) {
		g, err := ReadGenerations(path)
		if err != nil {
			return nil, err
		}
		pins := StagedPaths{}
		if g == nil {
			return pins, nil
		}
		for _, paths := range g.Paths {
			for _, p := range paths {
				if err := pins.addReal(p); err != nil {
					return nil, fmt.Errorf("generations file %s: %w", path, err)
				}
			}
		}
		if len(g.Hosts) == 0 {
			return pins, nil
		}
		rs, err := rows()
		if err != nil {
			return nil, fmt.Errorf("list routes: %w", err)
		}
		byHost := make(map[string]registry.Route, len(rs))
		for _, rt := range rs {
			byHost[rt.Host] = rt
		}
		for host, vers := range g.Hosts {
			rt, ok := byHost[host]
			if !ok || rt.App == "" {
				return nil, fmt.Errorf("generations file %s: %s has running generations and no registry row to place them", path, host)
			}
			for _, v := range vers {
				pins.add(VersionPath(optDir, RouteAccount(rt), v, rt.App))
			}
		}
		return pins, nil
	})
}

// GenerationVersions is the generations file as the version sweep keys its
// pins (spec §8.5, ruling R19 item 1): the name of every version directory a
// listed build lives in. Each listed version is one. Each listed path inside
// optDir's version store names its version directory, whether it is written
// under the opt dir's own name or reaches the store by its real path (a
// symlinked opt dir), so a build named only by path still protects its
// directory. A path outside the store names none: the sweep removes nothing
// there. It reads the file with ReadGenerations, so neither file is no
// versions, and an incomplete file, or a path or the store that cannot be
// resolved other than by not existing, is an error: the sweep's pin set is
// then incomplete.
func GenerationVersions(path, optDir string) (map[string]bool, error) {
	g, err := ReadGenerations(path)
	if err != nil {
		return nil, err
	}
	vers := map[string]bool{}
	if g == nil {
		return vers, nil
	}
	for _, vs := range g.Hosts {
		for _, v := range vs {
			vers[v] = true
		}
	}
	if len(g.Paths) == 0 {
		return vers, nil
	}
	bases := []string{versionsDir(optDir)}
	realBase, err := filepath.EvalSymlinks(versionsDir(optDir))
	switch {
	case err == nil:
		bases = append(bases, realBase)
	case !errors.Is(err, os.ErrNotExist):
		return nil, fmt.Errorf("resolve %s: %w", versionsDir(optDir), err)
	}
	for _, ps := range g.Paths {
		for _, p := range ps {
			names := []string{filepath.Clean(p)}
			real, err := filepath.EvalSymlinks(p)
			switch {
			case err == nil:
				names = append(names, real)
			case !errors.Is(err, os.ErrNotExist):
				return nil, fmt.Errorf("generations file %s: resolve %s: %w", path, p, err)
			}
			for _, n := range names {
				for _, b := range bases {
					if v := versionUnder(b, n); v != "" {
						vers[v] = true
					}
				}
			}
		}
	}
	return vers, nil
}