package roster

import (
	"context"
	"crypto/rand"
	"encoding/base64"
	"io"
	"net/http"
	"net/url"
	"sync"
	"time"

	"amadan.net/oficina/home/handoff"
	"github.com/go-json-experiment/json"
)

const (
	RefreshAfter = 30 * time.Second
	MaximumAge   = 60 * time.Second
	FetchTimeout = 5 * time.Second
	RetryAfter   = 5 * time.Second
	// A backward wall-clock discrepancy beyond two seconds fences standing.
	// Comparing wall advancement to monotonic advancement tolerates ordinary slew.
	ClockTolerance = 2 * time.Second
	recordLimit    = 1024
)

// Clock returns wall time and monotonic elapsed time from one observation. The
// elapsed value must never go backwards. Nil Config.Clock uses Go's monotonic clock.
type Clock interface {
	Now() (time.Time, time.Duration)
}
type realClock struct{ start time.Time }

func (c realClock) Now() (time.Time, time.Duration) { n := time.Now(); return n, n.Sub(c.start) }

// Durable stores one opaque envelope. Save must atomically replace it, retaining
// the previous value on error. Load reports ErrNoEvidence only for absent data.
// Its implementation must honor context cancellation and use bounded local IO.
type Durable interface {
	Load(context.Context) ([]byte, error)
	Save(context.Context, []byte) error
}

type Config struct {
	Record    VerifiedRecord
	Token     string
	Transport http.RoundTripper
	Durable   Durable
	Sealer    handoff.Sealer
	Clock     Clock
}

type State uint8

const (
	Unknown State = iota
	NotMember
	Member
)

type Decision struct {
	State State
	Role  string
	Lease Lease
}

// Lease is an observation, not a stored authorization grant. Call Valid at the
// protected operation boundary; stream users must recheck and re-arm their timer
// from a new decision after renewal, or close when renewal cannot authorize it.
// Both accessors refuse immediately on cache contention: a caller may hold the
// database writer that an in-flight durable publication needs to acquire.
type Lease struct {
	cache    *Cache
	version  uint64
	deadline time.Duration
}

func (l Lease) Valid() bool {
	if l.cache == nil {
		return false
	}
	c := l.cache
	if !c.mu.TryLock() {
		return false
	}
	defer c.mu.Unlock()
	_, n := c.observe()
	return c.version == l.version && c.usable && n < l.deadline
}
func (l Lease) Remaining() time.Duration {
	if l.cache == nil {
		return 0
	}
	c := l.cache
	if !c.mu.TryLock() {
		return 0
	}
	defer c.mu.Unlock()
	_, n := c.observe()
	if c.version != l.version || !c.usable || n >= l.deadline {
		return 0
	}
	return l.deadline - n
}

type receipt struct {
	Operation string `json:"operation"`
	Digest    string `json:"digest"`
}
type evidence struct {
	Snapshot     bool          `json:"snapshot"`
	Blocked      bool          `json:"blocked"`
	Version      int           `json:"version"`
	Sequence     int64         `json:"sequence"`
	RecordDigest string        `json:"record_digest"`
	Generation   string        `json:"generation"`
	Epoch        int64         `json:"epoch"`
	Members      []MemberEntry `json:"members"`
	Digest       string        `json:"digest"`
	Receipts     []receipt     `json:"receipts"`
}

type flight struct {
	done   chan struct{}
	err    error
	cancel context.CancelFunc
}
type Cache struct {
	mu          sync.Mutex
	record      VerifiedRecord
	token       string
	client      *http.Client
	durable     Durable
	sealer      handoff.Sealer
	purpose     string
	clock       Clock
	seen        bool
	wall        time.Time
	mono        time.Duration
	version     uint64
	usable      bool
	start       time.Duration
	members     map[string]string
	evidence    evidence
	flight      *flight
	attempted   bool
	lastAttempt time.Duration
}

func New(ctx context.Context, cfg Config) (*Cache, error) {
	if !cfg.Record.valid || !ValidChallenge(cfg.Token) || cfg.Durable == nil || cfg.Sealer == nil {
		return nil, ErrRefused
	}
	if cfg.Clock == nil {
		cfg.Clock = realClock{time.Now()}
	}
	if cfg.Transport == nil {
		tr := http.DefaultTransport.(*http.Transport).Clone()
		tr.DisableCompression = true
		cfg.Transport = tr
	}
	b, _ := json.Marshal(cfg.Record.record.Bindings)
	c := &Cache{record: cfg.Record, token: cfg.Token, client: &http.Client{Transport: cfg.Transport, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}, durable: cfg.Durable, sealer: cfg.Sealer, purpose: "oficina/roster/evidence/v1/" + hash(b), clock: cfg.Clock, version: 1}
	raw, e := cfg.Durable.Load(ctx)
	if e != nil && e != ErrNoEvidence {
		return nil, ErrUnavailable
	}
	if e == nil {
		if len(raw) > EvidenceLimit+128 {
			return nil, ErrRefused
		}
		plain, err := cfg.Sealer.Open(c.purpose, raw)
		if err != nil || len(plain) > EvidenceLimit || strict(plain, &c.evidence) != nil || !validEvidence(c.evidence) {
			return nil, ErrRefused
		}
	}
	ev, e := nextEvidence(c.evidence, cfg.Record)
	if e != nil {
		return nil, e
	}
	if e = c.save(ctx, ev); e != nil {
		return nil, e
	}
	c.evidence = ev
	return c, nil
}
func validEvidence(e evidence) bool {
	if e.Version != 1 || e.Sequence < 1 || e.Epoch < 1 || !ValidChallenge(e.RecordDigest) || !ValidChallenge(e.Generation) || len(e.Receipts) == 0 || len(e.Receipts) > recordLimit || len(e.Members) > MemberLimit {
		return false
	}
	seen := map[string]bool{}
	for _, r := range e.Receipts {
		if !atom(r.Operation, 128) || !ValidChallenge(r.Digest) || seen[r.Operation] {
			return false
		}
		seen[r.Operation] = true
	}
	for i, m := range e.Members {
		if !validMember(m) || (i > 0 && e.Members[i-1].Subject >= m.Subject) {
			return false
		}
	}
	return e.Digest == memberDigest(e.Members)
}
func memberDigest(m []MemberEntry) string {
	if m == nil {
		m = []MemberEntry{}
	}
	b, _ := json.Marshal(m)
	return hash(b)
}
func nextEvidence(old evidence, r VerifiedRecord) (evidence, error) {
	p := r.record
	if old.Version != 0 {
		if p.Sequence < old.Sequence {
			return evidence{}, ErrRefused
		}
		if p.Sequence == old.Sequence {
			if r.digest != old.RecordDigest {
				return evidence{}, ErrRefused
			}
			return old, nil
		}
		if p.Generation != old.Generation && p.OwnerConfirmation == "" {
			return evidence{}, ErrRefused
		}
	}
	for _, a := range old.Receipts {
		if a.Operation == p.Operation {
			return evidence{}, ErrRefused
		}
	}
	if len(old.Receipts) >= recordLimit {
		return evidence{}, ErrRefused
	}
	receipts := append(append([]receipt(nil), old.Receipts...), receipt{p.Operation, r.digest})
	ev := evidence{Version: 1, Sequence: p.Sequence, RecordDigest: r.digest, Generation: p.Generation, Epoch: p.MinEpoch, Members: []MemberEntry{}, Digest: memberDigest(nil), Receipts: receipts}
	if old.Generation == p.Generation && old.Epoch >= p.MinEpoch {
		ev.Epoch = old.Epoch
		ev.Members = old.Members
		ev.Digest = old.Digest
		ev.Snapshot = old.Snapshot
	}
	if old.Generation == p.Generation {
		ev.Blocked = old.Blocked
	}
	return ev, nil
}
func (c *Cache) save(ctx context.Context, e evidence) error {
	raw, err := json.Marshal(e)
	if err != nil || len(raw) > EvidenceLimit {
		return ErrUnavailable
	}
	ct, err := c.sealer.Seal(c.purpose, raw)
	if err != nil {
		return ErrUnavailable
	}
	if err = c.durable.Save(ctx, ct); err != nil {
		return ErrUnavailable
	}
	return nil
}

// UpdateRecord accepts only independently operator-verified records for the same
// bindings. Durable failure changes neither the current floor nor usable lease.
func (c *Cache) UpdateRecord(ctx context.Context, r VerifiedRecord) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if !r.valid || r.record.Bindings != c.record.record.Bindings || r.record.Sequence <= c.record.record.Sequence {
		return ErrRefused
	}
	ev, e := nextEvidence(c.evidence, r)
	if e != nil {
		return e
	}
	if e = c.save(ctx, ev); e != nil {
		return e
	}
	c.record = r
	c.evidence = ev
	c.fence()
	c.attempted = false
	return nil
}
func (c *Cache) fence() {
	c.usable = false
	c.version++
	if c.flight != nil {
		c.flight.cancel()
	}
}
func (c *Cache) observe() (time.Time, time.Duration) {
	w, m := c.clock.Now()
	if c.seen && (m < c.mono || w.Round(0).Sub(c.wall.Round(0)).Nanoseconds()-(m-c.mono).Nanoseconds() < -ClockTolerance.Nanoseconds()) {
		c.fence()
	}
	c.wall = w
	c.mono = m
	c.seen = true
	return w, m
}

func (c *Cache) Check(ctx context.Context, issuer, teamID, subject string) Decision {
	c.mu.Lock()
	b := c.record.record.Bindings
	if issuer != b.Issuer || teamID != b.Team || !validMember(MemberEntry{subject, "member"}) {
		c.mu.Unlock()
		return Decision{State: Unknown}
	}
	_, now := c.observe()
	usable := c.usable && now < c.start+MaximumAge
	due := !c.usable || now >= c.start+RefreshAfter
	var f *flight
	if due {
		f = c.startRefresh(now)
	}
	if usable {
		d := c.decision(subject)
		c.mu.Unlock()
		return d
	}
	c.mu.Unlock()
	if f != nil {
		select {
		case <-ctx.Done():
			return Decision{State: Unknown}
		case <-f.done:
		}
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	_, now = c.observe()
	if !c.usable || now >= c.start+MaximumAge {
		return Decision{State: Unknown}
	}
	return c.decision(subject)
}
func (c *Cache) decision(subject string) Decision {
	role, ok := c.members[subject]
	if !ok {
		return Decision{State: NotMember}
	}
	return Decision{State: Member, Role: role, Lease: Lease{c, c.version, c.start + MaximumAge}}
}

// Refresh joins the same bounded flight as Check. Caller cancellation stops
// waiting, not another request's shared refresh. Backoff never changes lease age.
func (c *Cache) Refresh(ctx context.Context) error {
	c.mu.Lock()
	_, now := c.observe()
	f := c.startRefresh(now)
	c.mu.Unlock()
	if f == nil {
		return ErrUnavailable
	}
	select {
	case <-ctx.Done():
		return ErrUnavailable
	case <-f.done:
		return f.err
	}
}
func (c *Cache) startRefresh(now time.Duration) *flight {
	if c.flight != nil {
		return c.flight
	}
	if c.evidence.Blocked {
		return nil
	}
	if c.attempted && now-c.lastAttempt < RetryAfter {
		return nil
	}
	ctx, cancel := context.WithTimeout(context.Background(), FetchTimeout)
	f := &flight{done: make(chan struct{}), cancel: cancel}
	c.flight = f
	c.attempted = true
	c.lastAttempt = now
	r := c.record
	version := c.version
	go func() {
		defer cancel()
		err := c.fetch(ctx, r, version, now)
		c.mu.Lock()
		f.err = err
		c.flight = nil
		close(f.done)
		c.mu.Unlock()
	}()
	return f
}
func (c *Cache) fetch(ctx context.Context, r VerifiedRecord, version uint64, start time.Duration) error {
	random := make([]byte, 32)
	if _, e := rand.Read(random); e != nil {
		return ErrUnavailable
	}
	challenge := base64.RawURLEncoding.EncodeToString(random)
	var members []MemberEntry
	cursor := ""
	epoch := int64(0)
	seen := map[string]bool{}
	for index := 0; index < 100; index++ {
		q := url.Values{"team": {r.record.Bindings.Team}, "fetch": {challenge}}
		if cursor != "" {
			q.Set("cursor", cursor)
		}
		target := r.record.Bindings.Directory + "/roster?" + q.Encode()
		if len(target) > 4096 {
			return ErrRefused
		}
		req, e := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
		if e != nil {
			return ErrUnavailable
		}
		req.Header.Set("Authorization", "Bearer "+c.token)
		req.Header.Set("Accept", "application/json")
		req.Header.Set("Accept-Encoding", "identity")
		response, e := c.client.Do(req)
		if e != nil {
			return ErrUnavailable
		}
		raw, readErr := io.ReadAll(io.LimitReader(response.Body, BodyLimit+1))
		response.Body.Close()
		if readErr != nil || len(raw) > BodyLimit || response.Header.Get("Content-Type") != "application/json" || response.Header.Get("Content-Encoding") != "" || response.Uncompressed {
			return ErrRefused
		}
		p, e := verifyPage(raw, r, challenge, cursor, response.StatusCode)
		if e != nil {
			return ErrRefused
		}
		if p.Generation != r.record.Generation {
			c.mu.Lock()
			if c.record.digest == r.digest {
				ev := c.evidence
				ev.Blocked = true
				_ = c.save(ctx, ev)
				c.evidence.Blocked = true
				c.fence()
			}
			c.mu.Unlock()
			return ErrRefused
		}
		if p.Type != "page" || *p.Index != index || p.Epoch < r.record.MinEpoch || (epoch != 0 && p.Epoch != epoch) {
			return ErrRefused
		}
		epoch = p.Epoch
		if len(members) > 0 && len(*p.Members) > 0 && members[len(members)-1].Subject >= (*p.Members)[0].Subject {
			return ErrRefused
		}
		members = append(members, (*p.Members)...)
		if len(members) > MemberLimit {
			return ErrRefused
		}
		if *p.Complete {
			return c.publish(ctx, r, version, start, epoch, members)
		}
		cursor = *p.Cursor
		if seen[cursor] {
			return ErrRefused
		}
		seen[cursor] = true
	}
	return ErrRefused
}
func (c *Cache) publish(ctx context.Context, r VerifiedRecord, version uint64, start time.Duration, epoch int64, members []MemberEntry) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	_, now := c.observe()
	if ctx.Err() != nil || c.version != version || c.record.digest != r.digest || now-start >= FetchTimeout || epoch < c.evidence.Epoch {
		return ErrUnavailable
	}
	digest := memberDigest(members)
	// A provisioned floor without a fetched snapshot has no member digest yet.
	if epoch == c.evidence.Epoch && c.evidence.Digest != "" && c.evidenceSnapshot() && digest != c.evidence.Digest {
		return ErrRefused
	}
	ev := c.evidence
	ev.Epoch = epoch
	ev.Members = append([]MemberEntry{}, members...)
	ev.Digest = digest
	ev.Snapshot = true
	if e := c.save(ctx, ev); e != nil {
		return e
	}
	c.evidence = ev
	_, now = c.observe()
	if ctx.Err() != nil || c.version != version || now-start >= FetchTimeout {
		return ErrUnavailable
	}
	c.members = make(map[string]string, len(members))
	for _, m := range members {
		c.members[m.Subject] = m.Role
	}
	c.start = start
	c.usable = true
	c.version++
	return nil
}
func (c *Cache) evidenceSnapshot() bool {
	return c.evidence.Snapshot
}