package hostclaim

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"io"
	"os"
	"sort"
	"strings"
	"sync"

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

const originNodeLimit = 4096
const originCountLimit = 1 << 20

// OriginHead authenticates only nodes reachable from this claim's committed head.
type OriginHead struct {
	Digest string `json:"digest"`
	Count  uint64 `json:"count"`
}

type OriginTuple struct {
	Incarnation string `json:"incarnation"`
	Location    string `json:"location"`
	Region      string `json:"region"`
	Generation  string `json:"generation"`
}

type originTuple = OriginTuple

type originNode struct {
	V        int         `json:"v"`
	Host     string      `json:"host"`
	Count    uint64      `json:"count"`
	Previous string      `json:"previous,omitempty"`
	Origin   originTuple `json:"origin"`
}

func originEnabled() bool {
	return hostwrite.Enabled(os.Getenv) && hostwrite.Enabled(func(string) string {
		return os.Getenv("CARLOS_HOST_CLAIM_ORIGINS")
	})
}

func originDigest(b []byte) string {
	h := sha256.New()
	h.Write([]byte("carlos-admitted-origin-v1\x00"))
	h.Write(b)
	return hex.EncodeToString(h.Sum(nil))
}

func originKey(host, digest string) string {
	return "control/host-origins/" + host + "/" + digest + ".json"
}

func originHeadValid(h *OriginHead) bool {
	return h == nil || h.Count > 0 && h.Count <= originCountLimit && isHex64(h.Digest) && h.Digest == strings.ToLower(h.Digest)
}

func (o originTuple) valid(host string) bool {
	if !isHex64(o.Incarnation) || o.Incarnation != strings.ToLower(o.Incarnation) ||
		o.Region == "" || len(o.Region) > 64 || strings.ContainsAny(o.Region, "/\\\x00") ||
		o.Generation == "" || len(o.Generation) > 256 || o.Generation == RootGeneration ||
		o.Generation == "." || o.Generation == ".." || strings.ContainsAny(o.Generation, "/\\\x00") {
		return false
	}
	return o.Location == "instances/"+host || o.Location == "instances/"+host+"/"+o.Incarnation
}

// ValidateOriginObject classifies inert, content-addressed candidates. Creating a
// candidate grants nothing; only the claim CAS can make one reachable.
func ValidateOriginObject(key string, body []byte) error {
	if len(body) == 0 || len(body) > originNodeLimit {
		return ErrMalformed
	}
	d := json.NewDecoder(bytes.NewReader(body))
	d.DisallowUnknownFields()
	var n originNode
	if err := d.Decode(&n); err != nil {
		return ErrMalformed
	}
	if _, err := d.Token(); err != io.EOF {
		return ErrMalformed
	}
	if _, err := Key(n.Host); err != nil {
		return ErrMalformed
	}
	if n.V != 1 || n.Count == 0 || n.Count > originCountLimit || !n.Origin.valid(n.Host) ||
		(n.Count == 1) != (n.Previous == "") ||
		n.Previous != "" && (!isHex64(n.Previous) || n.Previous != strings.ToLower(n.Previous)) ||
		key != originKey(n.Host, originDigest(body)) {
		return ErrMalformed
	}
	return nil
}

func readOrigin(ctx context.Context, s release.Store, host string, h OriginHead) (originNode, error) {
	var n originNode
	if err := ctx.Err(); err != nil {
		return n, err
	}
	key := originKey(host, h.Digest)
	b, err := readOriginBytes(ctx, s, key)
	if err != nil {
		return n, err
	}
	if err = ValidateOriginObject(key, b); err != nil {
		return n, err
	}
	if err = json.Unmarshal(b, &n); err != nil {
		return n, err
	}
	if n.Host != host || n.Count != h.Count {
		return n, ErrMalformed
	}
	return n, nil
}

func pendingOrigin(c *Claim, entry AdmissionRecord) error {
	if !originEnabled() {
		// Once written, provenance cannot silently stop collecting admissions.
		if c.Origins != nil {
			return ErrWrongState
		}
		return nil
	}
	if c.originAdmission != nil || c.Holder == nil {
		return ErrWrongState
	}
	o := originTuple{c.Holder.Incarnation, c.Holder.Location, entry.Region, entry.Generation}
	// The empty legacy root has no files/g-<generation>/ origin.
	if entry.Generation == "" {
		return nil
	}
	if !o.valid(c.Host) {
		return ErrWrongState
	}
	c.originAdmission = &o
	return nil
}

// commitOrigins stages immutable candidates before Update's existing claim CAS.
// A failed or uncertain CAS leaves candidates without granting them authority.
func commitOrigins(ctx context.Context, s release.Store, obs Observed, next *Claim) error {
	var old *OriginHead
	if obs.Claim != nil {
		old = obs.Claim.Origins
	}
	if !sameOriginHead(old, next.Origins) {
		return ErrWrongState
	}
	if next.originAdmission == nil {
		return nil
	}
	if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil || next.Holder == nil {
		return ErrWrongState
	}
	h := next.Holder
	want := *next.originAdmission
	if want.Incarnation != h.Incarnation || want.Location != h.Location || want.Generation != h.Generation ||
		h.Session != next.Gen || len(h.Admissions) == 0 {
		return ErrWrongState
	}
	a := h.Admissions[len(h.Admissions)-1]
	if a.Session != h.Session || a.Generation != want.Generation || a.Region != want.Region {
		return ErrWrongState
	}
	// Bind admission to the stored observation, never a caller-edited admission list.
	fresh, err := Read(ctx, s, obs.Host)
	if err != nil {
		return err
	}
	before, _ := json.Marshal(obs.Claim)
	stored, _ := json.Marshal(fresh.Claim)
	if fresh.Absent || fresh.ETag != obs.ETag || !bytes.Equal(before, stored) {
		return ErrConflict
	}
	// The chain starts empty at the independent writer boundary. Retained
	// admissions may have been forged before that boundary; never seed them.
	// Activation must re-copy every live origin through the #250 files backfill
	// into a prefix admitted after the boundary.
	head := OriginHead{}
	if old != nil {
		head = *old
		last, err := readOrigin(ctx, s, obs.Host, head)
		if err != nil {
			return err
		}
		if last.Origin == want {
			next.originAdmission = nil
			return nil
		}
	}
	if !want.valid(obs.Host) || head.Count == originCountLimit {
		return ErrWrongState
	}
	node := originNode{V: 1, Host: obs.Host, Count: head.Count + 1, Previous: head.Digest, Origin: want}
	body, err := json.Marshal(node)
	if err != nil {
		return err
	}
	key := originKey(obs.Host, originDigest(body))
	if err = ValidateOriginObject(key, body); err != nil {
		return err
	}
	if _, err = s.Put(ctx, key, body, ""); errors.Is(err, release.ErrCASConflict) {
		existing, readErr := readOriginBytes(ctx, s, key)
		if readErr != nil {
			return readErr
		}
		if !bytes.Equal(existing, body) {
			return ErrMalformed
		}
	} else if err != nil {
		return err
	}
	head = OriginHead{Digest: originDigest(body), Count: node.Count}
	next.Origins = &head
	next.originAdmission = nil
	return nil
}

func sameOriginHead(a, b *OriginHead) bool {
	if a == nil || b == nil {
		return a == nil && b == nil
	}
	return *a == *b
}

// OriginProofCache retains one exact batch, after authenticating its observation.
// Construct it with NewOriginProofCache; its store cannot be replaced.
type OriginProofCache struct {
	mu     sync.Mutex
	store  release.Store
	key    string
	result map[OriginTuple]bool
}

// NewOriginProofCache binds proof and reuse to one authoritative store. Queries
// accept no store argument, so a cache cannot carry answers to another store.
func NewOriginProofCache(s release.Store) *OriginProofCache {
	return &OriginProofCache{store: s}
}

func readOriginBytes(ctx context.Context, s release.Store, key string) ([]byte, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	// Require streaming through every OpenObject wrapper in the chain.
	reader, _, err := release.OpenObjectStreaming(ctx, s, key)
	if errors.Is(err, errors.ErrUnsupported) {
		return nil, ErrMalformed
	}
	if err != nil {
		return nil, err
	}
	b, readErr := io.ReadAll(io.LimitReader(reader, originNodeLimit+1))
	closeErr := reader.Close()
	if readErr != nil {
		return nil, readErr
	}
	if len(b) > originNodeLimit {
		return nil, ErrMalformed
	}
	if closeErr != nil {
		return nil, closeErr
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	return b, nil
}

// AdmittedOrigins authenticates once and proves the complete requested set in
// one walk, stopping once every tuple is proven. A malformed or unreadable node
// needed by the walk returns an error without partial results.
func AdmittedOrigins(ctx context.Context, s release.Store, obs Observed, set []OriginTuple) (map[OriginTuple]bool, error) {
	return NewOriginProofCache(s).Proven(ctx, obs, set)
}

func (cache *OriginProofCache) Proven(ctx context.Context, obs Observed, set []OriginTuple) (map[OriginTuple]bool, error) {
	if cache == nil {
		return nil, ErrMalformed
	}
	s := cache.store
	cache.mu.Lock()
	defer cache.mu.Unlock()
	invalidate := func() { cache.key, cache.result = "", nil }
	if err := ctx.Err(); err != nil {
		invalidate()
		return nil, err
	}
	if s == nil || obs.Absent || obs.Claim == nil || obs.ETag == "" || obs.Host != obs.Claim.Host {
		invalidate()
		return nil, ErrMalformed
	}
	wanted := make(map[OriginTuple]bool, len(set))
	for _, tuple := range set {
		if !tuple.valid(obs.Host) {
			invalidate()
			return nil, ErrMalformed
		}
		wanted[tuple] = false
	}
	fresh, err := Read(ctx, s, obs.Host)
	if err != nil {
		invalidate()
		return nil, err
	}
	before, marshalErr := json.Marshal(obs.Claim)
	stored, storedErr := json.Marshal(fresh.Claim)
	if fresh.Absent || fresh.ETag != obs.ETag || marshalErr != nil || storedErr != nil || !bytes.Equal(before, stored) {
		invalidate()
		return nil, ErrConflict
	}
	if err := ctx.Err(); err != nil {
		invalidate()
		return nil, err
	}
	// The store is fixed at construction. The key binds host, head, exact tuple
	// set, ETag and full claim body, including legacy admissions.
	tuples := make([]string, 0, len(wanted))
	for tuple := range wanted {
		b, _ := json.Marshal(tuple)
		tuples = append(tuples, string(b))
	}
	sort.Strings(tuples)
	keyBytes, _ := json.Marshal(struct {
		Host, ETag, Claim string
		Set               []string
	}{obs.Host, obs.ETag, string(stored), tuples})
	key := string(keyBytes)
	if cache.key == key {
		for tuple := range wanted {
			wanted[tuple] = cache.result[tuple]
		}
		return wanted, nil
	}
	invalidate()
	c := fresh.Claim
	// The chain starts empty: a claim with no recorded head proves nothing.
	// Retained admissions are never evidence (a box can write them), so live
	// origins are proven only through admissions recorded after the boundary
	// or the #250 files backfill into an admitted prefix.
	if c.Origins != nil {
		remaining := len(wanted)
		for head := *c.Origins; head.Count > 0 && remaining > 0; {
			node, err := readOrigin(ctx, s, obs.Host, head)
			if err != nil {
				return nil, err
			}
			if proven, ok := wanted[node.Origin]; ok && !proven {
				remaining--
				wanted[node.Origin] = true
			}
			head = OriginHead{Digest: node.Previous, Count: head.Count - 1}
		}
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	cache.key = key
	cache.result = make(map[OriginTuple]bool, len(wanted))
	for tuple, proven := range wanted {
		cache.result[tuple] = proven
	}
	return wanted, nil
}

// AdmittedOrigin is the singleton convenience form; multi-origin callers use
// AdmittedOrigins or one OriginProofCache.Proven call for their entire set.
func AdmittedOrigin(ctx context.Context, s release.Store, obs Observed, incarnation, location, region, generation string) (bool, error) {
	tuple := OriginTuple{incarnation, location, region, generation}
	proven, err := AdmittedOrigins(ctx, s, obs, []OriginTuple{tuple})
	return proven[tuple], err
}