package provision

import (
	"context"
	"errors"
	"fmt"
	"log"
	"maps"
	"os"
	"slices"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

// RecordView keeps the record families of the deployment bucket in memory
// and follows the change index so a pass re-reads only the scopes that moved
// (S3 change-index design §4). It holds, per account, the accounts/<a>/
// listing and account.json; per app, the instances/ and sidecars/ listings
// with every record body, the instances and sidecars manifests and the trash
// marker. Consumers never read it directly: they take a Snapshot and run the
// unchanged walk code (Reconcile, ReconcileSidecars, ListAllRecords*) over
// Snapshot().Store(live).
//
// Two maps drive it (§4.2): the watch's observed generations and the view's
// own applied ones. A scope whose applied generation is behind the observed
// one, or that is new, or whose last re-read failed, is re-read on the next
// Refresh, whether or not the index itself moved. Poll's moved list is never
// used: several consumers share one watch, and whichever polls first takes it.
type RecordView struct {
	live  release.Store
	watch *release.ChangeWatch
	// pollTimeout bounds each index poll (release.ChangeIndexPollTimeout;
	// tests shorten it).
	pollTimeout time.Duration

	// mu serializes Refresh and Rebuild for their whole duration, the backstop
	// walk included, so a publication always reflects reads that began after
	// the previous publication's (§4.3). It is a leaf: nothing else is taken
	// while it is held (the watch's own locks are leaves below it). The fields
	// after it are guarded by it.
	mu         sync.Mutex
	applied    map[string]uint64 // scope -> the generation the view's contents reflect
	pending    map[string]bool   // scopes whose last read failed: re-read on every Refresh
	rebuildDue bool              // the last walk failed or could not list an account's apps
	epoch      uint64

	snap atomic.Pointer[ViewSnapshot]
}

// recordViewRebuildHook, when set, runs inside Rebuild after the walk has
// read everything and before it publishes, with mu held. Tests only.
var recordViewRebuildHook func()

// NewRecordView returns a view over live that follows watch. It has published
// nothing: Snapshot is nil until the first Refresh or Rebuild succeeds.
func NewRecordView(live release.Store, watch *release.ChangeWatch) *RecordView {
	return &RecordView{live: live, watch: watch, pollTimeout: release.ChangeIndexPollTimeout, applied: map[string]uint64{}, pending: map[string]bool{}}
}

// Snapshot is the latest publication, or nil before the first. It never
// blocks on a refresh in progress.
func (v *RecordView) Snapshot() *ViewSnapshot { return v.snap.Load() }

// ScopeApp parses an app change scope, apps/<a>/<b>.
func ScopeApp(scope string) (acct, app string, ok bool) {
	rest, ok := strings.CutPrefix(scope, "apps/")
	if !ok {
		return "", "", false
	}
	acct, app, ok = strings.Cut(rest, "/")
	if !ok || acct == "" || app == "" || strings.Contains(app, "/") {
		return "", "", false
	}
	return acct, app, true
}

// scopeAccount parses an account change scope, accounts/<a>.
func scopeAccount(scope string) (string, bool) {
	a, ok := strings.CutPrefix(scope, "accounts/")
	if !ok || a == "" || strings.Contains(a, "/") {
		return "", false
	}
	return a, true
}

// viewScope reports whether the view keeps scope's contents. The control/
// scopes belong to other readers.
func viewScope(scope string) bool {
	if _, _, ok := ScopeApp(scope); ok {
		return true
	}
	_, ok := scopeAccount(scope)
	return ok
}

// Refresh brings the view up to the change index (§4.2): one conditional GET
// of the index, then a re-read of every scope whose applied generation is
// behind the observed one, that is new, or whose last re-read failed. A view
// that was never built, whose last walk was partial, or whose watch is
// unusable (no index support, a denied or malformed index) runs Rebuild
// instead, which is the walk every pass ran before the index (§3.4).
//
// A scope that fails to re-read keeps its applied generation, is retried on
// the next Refresh even when the index answers 304, and is published with the
// failure cached, so the snapshot is not Complete and a strict reader over it
// fails as it would on the bucket. A refresh whose ctx ends mid-read still
// publishes, with the reads that died cached as failures, and returns the
// ctx error. A refresh that walks and fails publishes the walk's failure
// (see Rebuild). Either way, no snapshot stays published as Complete once the
// view knows it is behind: consumers decide on Complete, and a strict reader
// over the snapshot fails where it would fail on the bucket.
func (v *RecordView) Refresh(ctx context.Context) error {
	v.mu.Lock()
	defer v.mu.Unlock()
	prev := v.snap.Load()
	if prev == nil || v.rebuildDue {
		return v.rebuildLocked(ctx, true)
	}
	if err := v.poll(ctx); err != nil {
		log.Printf("provision: record view: change index unusable (%v): walking", err)
		return v.rebuildLocked(ctx, false)
	}
	obs, ok := v.watch.Observed()
	if !ok {
		return v.rebuildLocked(ctx, false)
	}
	var due []string
	for scope, gen := range obs {
		if have, ok := v.applied[scope]; viewScope(scope) && (!ok || have < gen) {
			due = append(due, scope)
		}
	}
	for scope := range v.pending {
		if !slices.Contains(due, scope) {
			due = append(due, scope)
		}
	}
	if len(due) == 0 {
		return nil
	}
	sort.Strings(due)

	var mu sync.Mutex
	results := make(map[string]scopeRead, len(due))
	forEach(due, func(scope string) {
		r := v.readScope(ctx, prev, scope)
		mu.Lock()
		results[scope] = r
		mu.Unlock()
	})
	next := prev.clone()
	for _, scope := range due {
		r := results[scope]
		if a, ok := scopeAccount(scope); ok {
			next.setAccount(a, r.acct)
		} else if a, b, ok := ScopeApp(scope); ok {
			next.setApp(a, b, r.app)
		}
		if r.ok {
			v.applied[scope] = obs[scope]
			delete(v.pending, scope)
		} else {
			log.Printf("provision: record view: re-reading %s failed: retrying on the next refresh", scope)
			v.pending[scope] = true
		}
	}
	// Reads that died with the caller are cached as failures and the scopes
	// stay pending: the published snapshot is incomplete, never stale-complete.
	v.publish(next)
	return ctx.Err()
}

// Rebuild is the backstop (§4.4): under mu it polls the index, takes the
// observed generations, walks every account and app, publishes, and sets
// applied to the generations read BEFORE the walk, so a bump that lands
// during the walk is re-read by the next Refresh, never skipped. Unchanged
// bodies already in the view are kept by ETag rather than read again.
//
// A walk whose top-level listing fails (accounts/, or apps/ on a bucket-shaped
// store) returns the error and publishes the previous snapshot again with that
// error cached: List("accounts/") and ListDirs("apps/") over it answer the
// walk's failure, so knownAccounts, Reconcile and the strict readers fail over
// it exactly as they fail on the bucket, and it is not Complete. The next
// Refresh walks again. A view that never published stays unpublished.
func (v *RecordView) Rebuild(ctx context.Context) error {
	v.mu.Lock()
	defer v.mu.Unlock()
	return v.rebuildLocked(ctx, true)
}

// poll is one index poll, bounded by pollTimeout inside ctx: it runs under
// mu, and the callers' contexts are long-lived, so a stalled index GET must
// not hold the view (and with it every reconcile pass) or the watch's shared
// poll slot. A poll that runs out marks the watch unknown, and the caller
// walks.
func (v *RecordView) poll(ctx context.Context) error {
	pctx, cancel := context.WithTimeout(ctx, v.pollTimeout)
	defer cancel()
	_, err := v.watch.Poll(pctx)
	return err
}

// rebuildLocked is Rebuild under mu. poll is false when the caller's own
// poll just failed: the watch is unknown, so a second poll could only spend
// another bound before the same walk.
func (v *RecordView) rebuildLocked(ctx context.Context, poll bool) error {
	if poll {
		if err := v.poll(ctx); err != nil {
			log.Printf("provision: record view: change index unusable (%v): the walk records no generations", err)
		}
	}
	obs, usable := v.watch.Observed()
	prev := v.snap.Load()
	sn, failed, err := v.walk(ctx, prev)
	if err != nil {
		v.rebuildDue = true
		err = fmt.Errorf("provision: record view: walk: %w", err)
		if prev != nil {
			failedSn := prev.clone()
			failedSn.walkErr = err
			v.publish(failedSn)
		}
		return err
	}
	if hook := recordViewRebuildHook; hook != nil {
		hook()
	}
	v.applied = map[string]uint64{}
	if usable {
		for scope, gen := range obs {
			if !viewScope(scope) || failed[scope] {
				continue
			}
			if a, _, ok := ScopeApp(scope); ok && sn.dirErrs[a] != nil {
				continue // its account's apps were never listed
			}
			v.applied[scope] = gen
		}
	}
	v.pending = failed
	v.rebuildDue = len(sn.dirErrs) > 0
	// A walk whose ctx ended mid-way has its dead reads cached as failures:
	// published incomplete, retried by the next Refresh.
	v.publish(sn)
	return ctx.Err()
}

// publish stamps sn with the next epoch and its completeness and swaps it in.
func (v *RecordView) publish(sn *ViewSnapshot) {
	sn.Complete = len(v.pending) == 0 && !v.rebuildDue && walkComplete(sn)
	v.epoch++
	sn.Epoch = v.epoch
	v.snap.Store(sn)
}

// walkComplete runs the record walk itself over the snapshot, answering from
// the cache alone, and reports whether it would be partial: an account
// manifest that will not read cleanly, or an app listing that failed. Those
// are exactly what a strict reader refuses on the bucket.
func walkComplete(sn *ViewSnapshot) bool {
	ctx := context.Background()
	s := sn.Store(uncachedStore{})
	known, readFailed, err := knownAccounts(ctx, s)
	if err != nil || readFailed {
		return false
	}
	for _, sub := range []string{"instances/", "sidecars/"} {
		if _, partial, err := listAppSubtree(ctx, s, known, sub); err != nil || partial {
			return false
		}
	}
	return true
}

// scopeRead is one scope's re-read: the new entry and whether every read in
// it succeeded.
type scopeRead struct {
	acct *acctEntry
	app  *appEntry
	ok   bool
}

func (v *RecordView) readScope(ctx context.Context, prev *ViewSnapshot, scope string) scopeRead {
	if a, ok := scopeAccount(scope); ok {
		e, ok := v.readAccount(ctx, a, prev.accounts[a])
		return scopeRead{acct: e, ok: ok}
	}
	a, b, _ := ScopeApp(scope)
	p := release.AppPrefix(a, b)
	inst, instErr := listETags(ctx, v.live, p+"instances/")
	side, sideErr := listETags(ctx, v.live, p+"sidecars/")
	e, ok := v.readApp(ctx, a, b, prev.apps[a][b], listing{inst, instErr}, listing{side, sideErr})
	return scopeRead{app: e, ok: ok}
}

// readAccount re-reads accounts/<a>/: its listing, and account.json only
// when its ETag moved.
func (v *RecordView) readAccount(ctx context.Context, a string, prev *acctEntry) (*acctEntry, bool) {
	key := accountManifestKey(a)
	kes, err := listETags(ctx, v.live, "accounts/"+a+"/")
	if err != nil {
		return &acctEntry{listErr: err, manifest: object{err: err}}, false
	}
	e := &acctEntry{manifest: absent(key)}
	i := slices.IndexFunc(kes, func(k release.KeyETag) bool { return k.Key == key })
	if i >= 0 {
		var was *object
		if prev != nil {
			was = &prev.manifest
		}
		e.manifest = v.fetch(ctx, key, kes[i].ETag, was)
		if errors.Is(e.manifest.err, os.ErrNotExist) {
			kes = slices.Delete(kes, i, i+1) // deleted since it was listed
		}
	}
	e.keys = kes
	return e, !e.manifest.failed()
}

// listing is one prefix's ETag listing, or why it failed.
type listing struct {
	kes []release.KeyETag
	err error
}

// readApp fills one app's entry from its two listings: every listed record
// body (kept from prev when the ETag is unchanged), then both manifests and
// the trash marker by conditional GET.
func (v *RecordView) readApp(ctx context.Context, a, b string, prev *appEntry, inst, side listing) (*appEntry, bool) {
	e := &appEntry{instErr: inst.err, sideErr: side.err, objs: map[string]object{}}
	ok := inst.err == nil && side.err == nil
	read := func(kes []release.KeyETag) []release.KeyETag {
		var kept []release.KeyETag
		for _, ke := range kes {
			o := v.fetch(ctx, ke.Key, ke.ETag, prev.object(ke.Key))
			if errors.Is(o.err, os.ErrNotExist) {
				continue // deleted since it was listed: the listing follows the bucket
			}
			if o.failed() {
				ok = false
			}
			e.objs[ke.Key] = o
			kept = append(kept, ke)
		}
		return kept
	}
	e.inst, e.side = read(inst.kes), read(side.kes)
	for _, key := range []string{ManifestKey(a, b), SidecarManifestKey(a, b), release.AppTombstoneKey(a, b)} {
		o := condGet(ctx, v.live, key, prev.object(key))
		if o.failed() {
			ok = false
		}
		e.objs[key] = o
	}
	return e, ok
}

// fetch reads key unless prev already holds the body the listing's ETag
// names.
func (v *RecordView) fetch(ctx context.Context, key, listed string, prev *object) object {
	if prev != nil && prev.err == nil && listed != "" && prev.etag == listed {
		return *prev
	}
	return condGet(ctx, v.live, key, prev)
}

// condGet reads key with a conditional GET against prev's ETag, so an
// unchanged object costs a 304. Any error from the extension other than
// not-found means "unknown" and falls back to a plain Get, as does a store
// without it.
func condGet(ctx context.Context, s release.Store, key string, prev *object) object {
	etag := ""
	if prev != nil && prev.err == nil {
		etag = prev.etag
	}
	if g, ok := s.(release.ConditionalGetter); ok {
		body, newETag, changed, err := g.GetIfChanged(ctx, key, etag)
		switch {
		case err == nil && !changed && etag != "":
			return *prev
		case err == nil && changed:
			return object{body: body, etag: newETag}
		case errors.Is(err, os.ErrNotExist):
			return object{err: err}
		}
	}
	body, newETag, err := s.Get(ctx, key)
	return object{body: body, etag: newETag, err: err}
}

// listETags lists prefix with each key's ETag, sorted. A store without
// ETagLister, or any error from it, falls back to List with no ETags, which
// only costs the body reads the ETags would have saved.
func listETags(ctx context.Context, s release.Store, prefix string) ([]release.KeyETag, error) {
	var kes []release.KeyETag
	l, ok := s.(release.ETagLister)
	var err error
	if ok {
		kes, err = l.ListETags(ctx, prefix)
	}
	if !ok || err != nil {
		keys, err := s.List(ctx, prefix)
		if err != nil {
			return nil, err
		}
		kes = make([]release.KeyETag, len(keys))
		for i, k := range keys {
			kes[i] = release.KeyETag{Key: k}
		}
	}
	slices.SortFunc(kes, func(x, y release.KeyETag) int { return strings.Compare(x.Key, y.Key) })
	return kes, nil
}

// walk reads the whole deployment with the record walk's own code, knownAccounts
// and listAppSubtree, over a store that keeps what they read. Apps are walked
// under every account directory, not only known ones, so an account that
// becomes known later (restored, or materialized after its apps) already has
// its apps in the view. failed holds every scope with a read that failed.
func (v *RecordView) walk(ctx context.Context, prev *ViewSnapshot) (*ViewSnapshot, map[string]bool, error) {
	w := &walkStore{Store: v.live, prev: prev, lists: map[string]listing{}, dirs: map[string]dirListing{}, gets: map[string]object{}}
	known, _, err := knownAccounts(ctx, w)
	if err != nil {
		return nil, nil, err
	}
	every := maps.Clone(known)
	bucket := release.HasBucketListDirs(v.live)
	if bucket {
		dirs, err := w.ListDirs(ctx, "apps/")
		if err != nil {
			return nil, nil, fmt.Errorf("list apps: %w", err)
		}
		for _, d := range dirs {
			every[strings.TrimSuffix(strings.TrimPrefix(d, "apps/"), "/")] = true
		}
	}
	for _, sub := range []string{"instances/", "sidecars/"} {
		if _, _, err := listAppSubtree(ctx, w, every, sub); err != nil {
			return nil, nil, fmt.Errorf("list apps: %w", err)
		}
	}

	sn := &ViewSnapshot{accounts: map[string]*acctEntry{}, apps: map[string]map[string]*appEntry{}, dirErrs: map[string]error{}}
	failed := map[string]bool{}
	for _, ke := range w.lists["accounts/"].kes {
		scope, ok := release.ChangeScope(ke.Key)
		a, isAcct := scopeAccount(scope)
		if !ok || !isAcct {
			sn.strays = append(sn.strays, ke.Key)
			continue
		}
		e := sn.accounts[a]
		if e == nil {
			e = &acctEntry{manifest: absent(accountManifestKey(a))}
			sn.accounts[a] = e
		}
		e.keys = append(e.keys, ke)
	}
	for a, e := range sn.accounts {
		key := accountManifestKey(a)
		if !slices.ContainsFunc(e.keys, func(k release.KeyETag) bool { return k.Key == key }) {
			continue
		}
		o, ok := w.gets[key]
		if !ok { // knownAccounts reads every manifest it lists; this is belt and braces
			o = condGet(ctx, v.live, key, nil)
		}
		e.manifest = o
		if o.failed() {
			failed["accounts/"+a] = true
		}
	}

	type appListings struct{ inst, side listing }
	found := map[[2]string]*appListings{}
	at := func(a, b string) *appListings {
		k := [2]string{a, b}
		if found[k] == nil {
			found[k] = &appListings{}
		}
		return found[k]
	}
	if bucket {
		for _, d := range w.dirs["apps/"].dirs {
			a := strings.TrimSuffix(strings.TrimPrefix(d, "apps/"), "/")
			ad, ok := w.dirs[d]
			if !ok {
				continue
			}
			if ad.err != nil {
				sn.dirErrs[a] = ad.err
				continue
			}
			for _, p := range ad.dirs {
				b := strings.TrimSuffix(strings.TrimPrefix(p, d), "/")
				l := at(a, b)
				l.inst, l.side = w.lists[p+"instances/"], w.lists[p+"sidecars/"]
			}
		}
	} else {
		for _, ke := range w.lists["apps/"].kes {
			parts := strings.SplitN(ke.Key, "/", 4)
			if len(parts) < 4 || parts[1] == "" || parts[2] == "" {
				continue
			}
			l := at(parts[1], parts[2])
			switch {
			case strings.HasPrefix(parts[3], "instances/"):
				l.inst.kes = append(l.inst.kes, ke)
			case strings.HasPrefix(parts[3], "sidecars/"):
				l.side.kes = append(l.side.kes, ke)
			}
		}
	}
	ids := make([]string, 0, len(found))
	for k := range found {
		ids = append(ids, "apps/"+k[0]+"/"+k[1])
	}
	var mu sync.Mutex
	forEach(ids, func(scope string) {
		a, b, _ := ScopeApp(scope)
		l := found[[2]string{a, b}]
		e, ok := v.readApp(ctx, a, b, prev.app(a, b), l.inst, l.side)
		mu.Lock()
		defer mu.Unlock()
		sn.setApp(a, b, e)
		if !ok {
			failed[scope] = true
		}
	})
	return sn, failed, nil
}

// walkStore is the live store as the backstop walk sees it: every listing
// carries ETags and every read is conditional on what the previous snapshot
// held, and both are kept for the snapshot the walk builds. Each listing is
// read once per walk, however many times the walk code asks.
type walkStore struct {
	release.Store
	prev *ViewSnapshot

	mu    sync.Mutex
	lists map[string]listing
	dirs  map[string]dirListing
	gets  map[string]object
}

type dirListing struct {
	dirs []string
	err  error
}

func (w *walkStore) BucketShapedListDirs() bool { return release.HasBucketListDirs(w.Store) }

func (w *walkStore) List(ctx context.Context, prefix string) ([]string, error) {
	w.mu.Lock()
	l, ok := w.lists[prefix]
	w.mu.Unlock()
	if !ok {
		kes, err := listETags(ctx, w.Store, prefix)
		l = listing{kes, err}
		w.mu.Lock()
		w.lists[prefix] = l
		w.mu.Unlock()
	}
	return keysOf(l.kes), l.err
}

func (w *walkStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	w.mu.Lock()
	d, ok := w.dirs[prefix]
	w.mu.Unlock()
	if !ok {
		dirs, err := w.Store.ListDirs(ctx, prefix)
		d = dirListing{dirs, err}
		w.mu.Lock()
		w.dirs[prefix] = d
		w.mu.Unlock()
	}
	return slices.Clone(d.dirs), d.err
}

func (w *walkStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	var prev *object
	if o, ok := w.prev.lookup(key); ok && o.err == nil {
		prev = &o
	}
	o := condGet(ctx, w.Store, key, prev)
	w.mu.Lock()
	w.gets[key] = o
	w.mu.Unlock()
	return o.result()
}

func keysOf(kes []release.KeyETag) []string {
	if kes == nil {
		return nil
	}
	out := make([]string, len(kes))
	for i, ke := range kes {
		out[i] = ke.Key
	}
	return out
}