package release

import (
	"context"
	"errors"
	"fmt"
	"os"
	"sort"
	"sync"
	"time"
)

// ChangeIndexBackstop is the longest a gated reader goes without a successful
// full read of a scope, however quiet the change index is (R3). It lives here
// so provision, host, edge and console share one value without import cycles.
// Callers jitter it by +-10%.
const ChangeIndexBackstop = 30 * time.Minute

// ChangeIndexPollTimeout bounds one Poll from a caller with a deadline of its
// own to keep (the agent's fast tick, the edge's health-stamping loops): the
// wait for another caller's in-flight poll and the conditional GET together.
// A poll that runs out is "unknown" to its caller, never "unchanged".
const ChangeIndexPollTimeout = 5 * time.Second

// ChangeWatch follows the change index (ChangeIndexKey) with one conditional
// GET per Poll: an unchanged index costs a 304 and no body. It is safe for
// concurrent use.
//
// The watch is unusable until its first good poll, and again after any
// failed one: an access error, a network error, an unknown schema, a
// malformed body, a store without ConditionalGetter, or a caller's context
// ending, while waiting for another caller's poll included. "Unknown" is never
// read as "empty and unchanged"; callers walk as they did before the index
// existed. The one exception is an absent index, which is the ordinary state
// before the first bump and reads as usable and empty.
type ChangeWatch struct {
	store Store

	// pollSem serializes polls: a one-slot semaphore held across the
	// network call. A channel rather than a mutex so a waiter behind a hung
	// poll gives up when its context ends instead of inheriting the hang.
	pollSem chan struct{}

	mu      sync.Mutex // guards the fields below; never held across I/O
	usable  bool
	etag    string            // etag of the last good index body ("" when it was absent)
	entries map[string]uint64 // the last good observation
}

// NewChangeWatch returns a watch over s. It has not polled yet, so it is
// unusable.
func NewChangeWatch(s Store) *ChangeWatch {
	return &ChangeWatch{store: s, entries: map[string]uint64{}, pollSem: make(chan struct{}, 1)}
}

// Poll reads the index conditionally and returns, sorted, the scopes whose
// generation differs from the last good observation: every scope on the
// first good poll, then those that rose or appeared. On a 304 it returns
// nil, nil. Any error marks the watch unusable until the next good poll and
// is returned; that next poll reports every scope that differs from the last
// good observation, including ones that rose while the watch was unusable.
//
// moved is advisory. Several consumers in one process share a watch (a box
// agent's tick, the record view, the cert mirror), so whichever polls first
// consumes a moved list and the others see nil. A consumer that must act on
// movement compares Observed against the map it last acted on instead of
// relying on Poll's return.
//
// A Poll that waits behind another caller's in-flight poll stops waiting when
// ctx ends and returns ctx's error. That give-up is a failed poll like any
// other: the watch is unusable until the next good poll, so no gated reader
// trusts an observation a stalled index GET has frozen. The ETag and entries
// are kept, so the next good poll of an unchanged index is a 304 again.
func (w *ChangeWatch) Poll(ctx context.Context) (moved []string, err error) {
	if err := ctx.Err(); err != nil {
		return nil, w.fail(err)
	}
	select {
	case w.pollSem <- struct{}{}:
	case <-ctx.Done():
		return nil, w.fail(ctx.Err())
	}
	defer func() { <-w.pollSem }()

	g, ok := w.store.(ConditionalGetter)
	if !ok {
		return nil, w.fail(fmt.Errorf("release: change index unreadable: store has no conditional read: %w", errors.ErrUnsupported))
	}
	w.mu.Lock()
	etag := w.etag
	w.mu.Unlock()

	body, newETag, changed, err := g.GetIfChanged(ctx, ChangeIndexKey, etag)
	if errors.Is(err, os.ErrNotExist) {
		return w.adopt(ChangeIndex{Schema: 1, Entries: map[string]uint64{}}, ""), nil
	}
	if err != nil {
		return nil, w.fail(err)
	}
	if !changed {
		w.mu.Lock()
		w.usable = true // same bytes as the last good body, which are the last good observation
		w.mu.Unlock()
		return nil, nil
	}
	idx, _, err := readChangeIndex(ctx, func(context.Context, string) ([]byte, string, error) {
		return body, newETag, nil
	})
	if err != nil {
		return nil, w.fail(err)
	}
	return w.adopt(idx, newETag), nil
}

func (w *ChangeWatch) fail(err error) error {
	w.mu.Lock()
	w.usable = false
	w.mu.Unlock()
	return err
}

// adopt records idx as the last good observation and returns the scopes it
// moved.
func (w *ChangeWatch) adopt(idx ChangeIndex, etag string) []string {
	w.mu.Lock()
	defer w.mu.Unlock()
	var moved []string
	for scope, gen := range idx.Entries {
		if old, ok := w.entries[scope]; !ok || old != gen {
			moved = append(moved, scope)
		}
	}
	sort.Strings(moved)
	w.entries = idx.Entries
	w.etag = etag
	w.usable = true
	return moved
}

// Usable reports whether the last poll succeeded.
func (w *ChangeWatch) Usable() bool {
	w.mu.Lock()
	defer w.mu.Unlock()
	return w.usable
}

// Gen is the last observed generation of scope. A scope the index does not
// list has generation 0 with ok true; ok is false while the watch is
// unusable.
func (w *ChangeWatch) Gen(scope string) (gen uint64, ok bool) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if !w.usable {
		return 0, false
	}
	return w.entries[scope], true
}

// Observed returns a copy of the last good entries, and false while the watch
// is unusable. Poll's moved list is advisory when consumers share a watch, so
// a consumer that must act on movement diffs Observed against its own
// last-acted map.
func (w *ChangeWatch) Observed() (map[string]uint64, bool) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if !w.usable {
		return nil, false
	}
	out := make(map[string]uint64, len(w.entries))
	for k, v := range w.entries {
		out[k] = v
	}
	return out, true
}

// ReadGate decides, per scope, whether a full read is due: when the watch
// cannot vouch for the scope, the scope's generation has moved since the last
// successful read, or the last successful read is maxAge old. It is safe for
// concurrent use.
type ReadGate struct {
	w      *ChangeWatch
	maxAge time.Duration

	mu    sync.Mutex
	marks map[string]gateMark
}

type gateMark struct {
	gen uint64
	at  time.Time
	ok  bool
}

// NewReadGate gates reads on w. A nil w makes every read due. maxAge is the
// longest a successful read stands in for the next one (use
// ChangeIndexBackstop, jittered).
func NewReadGate(w *ChangeWatch, maxAge time.Duration) *ReadGate {
	return &ReadGate{w: w, maxAge: maxAge, marks: map[string]gateMark{}}
}

// MaxAge is the longest a successful read stands in for the next one. A
// caller keeping finer marks of its own under one scope (the agent's canary
// groups, one per channel) ages them by it too.
func (g *ReadGate) MaxAge() time.Duration { return g.maxAge }

// Due reports whether scope must be read now and returns the generation to
// pass to Mark once the read is done. Capture it BEFORE the read: a write that
// lands during the read then leaves the scope due on the next check instead
// of being hidden by it. The generation is 0 when the watch is nil or
// unusable.
func (g *ReadGate) Due(scope string, now time.Time) (gen uint64, due bool) {
	if g == nil || g.w == nil {
		return 0, true
	}
	gen, ok := g.w.Gen(scope)
	if !ok {
		return 0, true
	}
	g.mu.Lock()
	m, marked := g.marks[scope]
	g.mu.Unlock()
	switch {
	case !marked, !m.ok, m.gen != gen, now.Sub(m.at) >= g.maxAge:
		return gen, true
	}
	return gen, false
}

// Mark records the outcome of a read of scope that began when Due returned
// gen. A failed read (ok false) leaves the scope due.
func (g *ReadGate) Mark(scope string, gen uint64, now time.Time, ok bool) {
	if g == nil {
		return
	}
	g.mu.Lock()
	g.marks[scope] = gateMark{gen: gen, at: now, ok: ok}
	g.mu.Unlock()
}