package provision

import (
	"context"
	"errors"
	"fmt"
	"io"
	"maps"
	"os"
	"slices"
	"strings"
	"sync"

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

// ViewSnapshot is one immutable publication of a RecordView. Epoch increases
// with every publication. Complete is false while any scope has a failed or
// outstanding re-read, after a walk that could not list an account's apps,
// or when the record walk over the snapshot would itself be partial (an
// account manifest that will not read cleanly, an app listing that failed).
// It is never read and changed at once: a Refresh builds a new one.
type ViewSnapshot struct {
	Epoch    uint64
	Complete bool

	accounts map[string]*acctEntry           // sqid -> accounts/<sqid>/
	strays   []string                        // keys under accounts/ outside every account scope, from the last walk
	apps     map[string]map[string]*appEntry // acct -> app -> apps/<acct>/<app>/
	dirErrs  map[string]error                // acct -> why the last walk could not list apps/<acct>/
	walkErr  error                           // the last walk failed at the top: accounts/ and apps/ answer it
}

// acctEntry is one account scope: its complete key listing and account.json,
// or the listing failure (which the manifest then carries too, so a reader
// asking for the manifest learns the read failed rather than that it is
// absent).
type acctEntry struct {
	keys     []release.KeyETag
	listErr  error
	manifest object
}

// appEntry is one app scope: both record listings (or why each failed), and
// objs, every listed record body plus the two manifests and the trash marker.
type appEntry struct {
	inst, side       []release.KeyETag
	instErr, sideErr error
	objs             map[string]object
}

// object is one cached read: a body and its ETag, a not-found error wrapping
// os.ErrNotExist, or the read error itself.
type object struct {
	body []byte
	etag string
	err  error
}

func absent(key string) object {
	return object{err: fmt.Errorf("provision: record view: %s: %w", key, os.ErrNotExist)}
}

// failed reports a read that did not answer: anything but a body or absence.
func (o object) failed() bool { return o.err != nil && !errors.Is(o.err, os.ErrNotExist) }

// result is the object as Get returns it, with a body the caller may keep.
func (o object) result() ([]byte, string, error) {
	if o.err != nil {
		return nil, "", o.err
	}
	return slices.Clone(o.body), o.etag, nil
}

func accountManifestKey(a string) string { return "accounts/" + a + "/account.json" }

func (e *acctEntry) empty() bool {
	return e == nil || (len(e.keys) == 0 && e.listErr == nil)
}

// empty reports an app with no record-family state at all: nothing listed,
// nothing failed, and all three per-app objects absent. A purged app is
// dropped from the view, and from ListDirs over it.
func (e *appEntry) empty() bool {
	if e == nil {
		return true
	}
	if len(e.inst) > 0 || len(e.side) > 0 || e.instErr != nil || e.sideErr != nil {
		return false
	}
	for _, o := range e.objs {
		if !errors.Is(o.err, os.ErrNotExist) {
			return false
		}
	}
	return true
}

// object is the cached read of key, or nil when the entry holds none.
func (e *appEntry) object(key string) *object {
	if e == nil {
		return nil
	}
	o, ok := e.objs[key]
	if !ok {
		return nil
	}
	return &o
}

func (sn *ViewSnapshot) app(a, b string) *appEntry {
	if sn == nil {
		return nil
	}
	return sn.apps[a][b]
}

// clone copies the maps a Refresh changes; entries are shared, never mutated.
func (sn *ViewSnapshot) clone() *ViewSnapshot {
	out := &ViewSnapshot{
		accounts: maps.Clone(sn.accounts),
		strays:   sn.strays,
		apps:     make(map[string]map[string]*appEntry, len(sn.apps)),
		dirErrs:  sn.dirErrs,
		walkErr:  sn.walkErr,
	}
	for a, inner := range sn.apps {
		out.apps[a] = maps.Clone(inner)
	}
	return out
}

func (sn *ViewSnapshot) setAccount(a string, e *acctEntry) {
	if e.empty() {
		delete(sn.accounts, a)
		return
	}
	sn.accounts[a] = e
}

func (sn *ViewSnapshot) setApp(a, b string, e *appEntry) {
	if e.empty() {
		delete(sn.apps[a], b)
		if len(sn.apps[a]) == 0 {
			delete(sn.apps, a)
		}
		return
	}
	if sn.apps[a] == nil {
		sn.apps[a] = map[string]*appEntry{}
	}
	sn.apps[a][b] = e
}

// appKey splits apps/<a>/<b>/<rest> with all three parts non-empty.
func appKey(key string) (a, b, rest string, ok bool) {
	parts := strings.SplitN(key, "/", 4)
	if len(parts) < 4 || parts[0] != "apps" || parts[1] == "" || parts[2] == "" || parts[3] == "" {
		return "", "", "", false
	}
	return parts[1], parts[2], parts[3], true
}

// lookup answers Get of a key in a cached family; handled is false for every
// other key. A key the view does not hold is absent: its family's listing is
// complete. A record under a listing that failed answers that failure, and
// so does any key of an app the view never read because the walk could not
// list its account's apps.
func (sn *ViewSnapshot) lookup(key string) (o object, handled bool) {
	if sn == nil {
		return object{}, false
	}
	if a, ok := domains.AccountManifestSqid(key); ok {
		e := sn.accounts[a]
		if e == nil {
			return absent(key), true
		}
		return e.manifest, true
	}
	a, b, rest, ok := appKey(key)
	if !ok {
		return object{}, false
	}
	var listErr error
	switch {
	case rest == "instances.json" || rest == "sidecars.json" || rest == "deleted.json":
	case strings.HasPrefix(rest, "instances/") && rest != "instances/":
		listErr = sn.app(a, b).listErr(true)
	case strings.HasPrefix(rest, "sidecars/") && rest != "sidecars/":
		listErr = sn.app(a, b).listErr(false)
	default:
		return object{}, false
	}
	if listErr != nil {
		return object{err: listErr}, true
	}
	if err := sn.dirErrs[a]; err != nil && sn.app(a, b) == nil {
		// The last walk could not list this account's apps, and no refresh
		// has read this one since: the view does not know, so it must not
		// answer absent.
		return object{err: err}, true
	}
	if p := sn.app(a, b).object(key); p != nil {
		return *p, true
	}
	return absent(key), true
}

func (e *appEntry) listErr(inst bool) error {
	switch {
	case e == nil:
		return nil
	case inst:
		return e.instErr
	}
	return e.sideErr
}

// list answers List of a cached family, sorted as the stores list.
func (sn *ViewSnapshot) list(prefix string) (keys []string, handled bool, err error) {
	if prefix == "accounts/" {
		if sn.walkErr != nil {
			return nil, true, sn.walkErr
		}
		out := slices.Clone(sn.strays)
		for a, e := range sn.accounts {
			if e.listErr != nil {
				// Unknown keys; the manifest is listed so a reader asks for it
				// and learns the read failed (knownAccounts' readFailed).
				out = append(out, accountManifestKey(a))
				continue
			}
			out = append(out, keysOf(e.keys)...)
		}
		slices.Sort(out)
		return out, true, nil
	}
	if a, ok := strings.CutPrefix(prefix, "accounts/"); ok {
		a, ok = strings.CutSuffix(a, "/")
		if !ok || a == "" || strings.Contains(a, "/") {
			return nil, false, nil
		}
		e := sn.accounts[a]
		if e == nil {
			return nil, true, nil
		}
		return keysOf(e.keys), true, e.listErr
	}
	a, b, rest, ok := appKey(prefix)
	if !ok || (rest != "instances/" && rest != "sidecars/") {
		return nil, false, nil
	}
	e := sn.app(a, b)
	if e == nil {
		return nil, true, sn.dirErrs[a] // unknown, not empty, when the walk could not list the account's apps
	}
	if rest == "instances/" {
		return keysOf(e.inst), true, e.instErr
	}
	return keysOf(e.side), true, e.sideErr
}

// listDirs answers ListDirs("apps/") and ListDirs("apps/<a>/"): accounts and
// apps holding record-family state, as full prefixes, sorted. An account whose
// apps the last walk could not list is listed, and listing it fails, as it did.
func (sn *ViewSnapshot) listDirs(prefix string) (dirs []string, handled bool, err error) {
	if prefix == "apps/" {
		if sn.walkErr != nil {
			return nil, true, sn.walkErr
		}
		var out []string
		for a := range sn.apps {
			out = append(out, "apps/"+a+"/")
		}
		for a := range sn.dirErrs {
			if sn.apps[a] == nil {
				out = append(out, "apps/"+a+"/")
			}
		}
		slices.Sort(out)
		return out, true, nil
	}
	a, ok := strings.CutPrefix(prefix, "apps/")
	if !ok {
		return nil, false, nil
	}
	a, ok = strings.CutSuffix(a, "/")
	if !ok || a == "" || strings.Contains(a, "/") {
		return nil, false, nil
	}
	if err := sn.dirErrs[a]; err != nil {
		return nil, true, err
	}
	var out []string
	for b := range sn.apps[a] {
		out = append(out, prefix+b+"/")
	}
	slices.Sort(out)
	return out, true, nil
}

// Store is the snapshot as a release.Store over live: the cached families are
// answered from the snapshot, everything else from live. Every write goes to
// live, and from then on that key, and every listing it falls under, is read
// from live for the rest of this store's life, so a caller reads its own
// writes and a CAS retry re-reads the bucket. Each call returns a fresh store;
// the snapshot itself never changes.
func (sn *ViewSnapshot) Store(live release.Store) release.Store {
	return &snapStore{sn: sn, live: live}
}

// freshStore is the store a reader that must see the bucket as it is now
// reads through: live under a snapshot store, s itself otherwise. A guard
// that re-reads a record to check that nothing moved since the listing
// (holdExisting) would only read the listing's own snapshot back.
func freshStore(s release.Store) release.Store {
	if ss, ok := s.(*snapStore); ok {
		return ss.live
	}
	return s
}

// onSnapshot reports whether s is a record view snapshot store. A snapshot's
// absence is not the bucket's decision: a record written without its bump (a
// failed bump, an old binary, a direct-bucket writer) is missing from a
// Complete snapshot until the backstop rebuild. So Reconcile and
// ReconcileSidecars over a snapshot decide any removal from the bucket's own
// walk (scanRecords, scanSidecars over freshStore), and over any other store
// from the walk they already made.
func onSnapshot(s release.Store) bool {
	_, ok := s.(*snapStore)
	return ok
}

// walkErrSuffix renders the error of a bucket walk that could not be read
// cleanly for a log line: nothing when the walk returned but read partially
// (its own lines name what failed).
func walkErrSuffix(err error) string {
	if err == nil {
		return ""
	}
	return ": " + err.Error()
}

type snapStore struct {
	sn   *ViewSnapshot
	live release.Store

	mu    sync.Mutex
	wrote []string
}

var (
	_ release.Store              = (*snapStore)(nil)
	_ release.BucketListDirs     = (*snapStore)(nil)
	_ release.Opener             = (*snapStore)(nil)
	_ release.PutStreamer        = (*snapStore)(nil)
	_ release.ConditionalDeleter = (*snapStore)(nil)
	_ release.ModLister          = (*snapStore)(nil)
	_ release.SizeLister         = (*snapStore)(nil)
	_ release.PageLister         = (*snapStore)(nil)
	_ release.ConditionalGetter  = (*snapStore)(nil)
	_ release.ETagLister         = (*snapStore)(nil)
	_ release.AppMutationRunner  = (*snapStore)(nil)
)

func (s *snapStore) mark(key string) {
	s.mu.Lock()
	s.wrote = append(s.wrote, key)
	s.mu.Unlock()
}

// wroteUnder reports a write through this store to a key under prefix (or,
// with exact, to key itself).
func (s *snapStore) wroteUnder(prefix string, exact bool) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, k := range s.wrote {
		if k == prefix || (!exact && strings.HasPrefix(k, prefix)) {
			return true
		}
	}
	return false
}

// BucketShapedListDirs: the snapshot answers ListDirs("apps/") and
// ListDirs("apps/<a>/") in the bucket's shape whatever live's shape is (after
// a write it asks live only when live is bucket-shaped too), so the record
// walk lists per app instead of the whole of apps/.
func (s *snapStore) BucketShapedListDirs() bool { return true }

func (s *snapStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if !s.wroteUnder(key, true) {
		if o, ok := s.sn.lookup(key); ok {
			return o.result()
		}
	}
	return s.live.Get(ctx, key)
}

func (s *snapStore) List(ctx context.Context, prefix string) ([]string, error) {
	if !s.wroteUnder(prefix, false) {
		if keys, ok, err := s.sn.list(prefix); ok {
			return keys, err
		}
	}
	return s.live.List(ctx, prefix)
}

func (s *snapStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
	norm := prefix
	if norm != "" && !strings.HasSuffix(norm, "/") {
		norm += "/"
	}
	dirs, ok, err := s.sn.listDirs(norm)
	if !ok {
		return s.live.ListDirs(ctx, prefix)
	}
	if err != nil || !s.wroteUnder(norm, false) {
		return dirs, err
	}
	// A write under prefix: live knows the directories it made, but only a
	// bucket-shaped live answers in the shape this store promises. The fleet
	// door's ListDirs("apps/") is its account's apps, and walking that as
	// accounts found no records at all (listAppSubtree). Otherwise the
	// snapshot's directories plus those the writes imply.
	if release.HasBucketListDirs(s.live) {
		return s.live.ListDirs(ctx, prefix)
	}
	s.mu.Lock()
	for _, k := range s.wrote {
		rest, ok := strings.CutPrefix(k, norm)
		if !ok {
			continue
		}
		if child, _, ok := strings.Cut(rest, "/"); ok && child != "" {
			if d := norm + child + "/"; !slices.Contains(dirs, d) {
				dirs = append(dirs, d)
			}
		}
	}
	s.mu.Unlock()
	slices.Sort(dirs)
	return dirs, nil
}

func (s *snapStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	s.mark(key)
	return s.live.Put(ctx, key, body, ifMatch)
}

func (s *snapStore) PutFrom(ctx context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) {
	s.mark(key)
	return release.PutObjectFrom(ctx, s.live, key, body, size, ifMatch)
}

func (s *snapStore) Delete(ctx context.Context, key string) error {
	s.mark(key)
	return s.live.Delete(ctx, key)
}

func (s *snapStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
	s.mark(key)
	return release.DeleteIfMatch(ctx, s.live, key, etag)
}

// The remaining extensions are not cached families: forwarded to live, and
// errors.ErrUnsupported when live lacks one.

func (s *snapStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
	return release.OpenObject(ctx, s.live, key)
}

func (s *snapStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
	if g, ok := s.live.(release.ConditionalGetter); ok {
		return g.GetIfChanged(ctx, key, etag)
	}
	return nil, "", false, errors.ErrUnsupported
}

func (s *snapStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
	if l, ok := s.live.(release.ETagLister); ok {
		return l.ListETags(ctx, prefix)
	}
	return nil, errors.ErrUnsupported
}

func (s *snapStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) {
	if l, ok := s.live.(release.ModLister); ok {
		return l.ListMod(ctx, prefix)
	}
	return nil, errors.ErrUnsupported
}

func (s *snapStore) ListSizes(ctx context.Context, prefix string) ([]release.KeySize, error) {
	if l, ok := s.live.(release.SizeLister); ok {
		return l.ListSizes(ctx, prefix)
	}
	return nil, errors.ErrUnsupported
}

func (s *snapStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
	if l, ok := s.live.(release.PageLister); ok {
		return l.ListPage(ctx, prefix, startAfter, limit)
	}
	return nil, errors.ErrUnsupported
}

// RunAppMutation keeps live's claim boundary for writes made through the
// snapshot store; a live store without one runs f directly, as
// release.WithAppMutation would.
func (s *snapStore) RunAppMutation(ctx context.Context, acct, app, action string, f func(context.Context) error) error {
	if r, ok := s.live.(release.AppMutationRunner); ok {
		return r.RunAppMutation(ctx, acct, app, action, f)
	}
	return f(ctx)
}

// uncachedStore refuses everything. walkComplete runs the record walk over a
// snapshot store backed by it, so a read the cache does not answer shows up
// as a failure instead of a bucket request.
type uncachedStore struct{}

var errUncached = errors.New("provision: record view: not in the snapshot")

func (uncachedStore) Get(context.Context, string) ([]byte, string, error) {
	return nil, "", errUncached
}
func (uncachedStore) Put(context.Context, string, []byte, string) (string, error) {
	return "", errUncached
}
func (uncachedStore) List(context.Context, string) ([]string, error)     { return nil, errUncached }
func (uncachedStore) ListDirs(context.Context, string) ([]string, error) { return nil, errUncached }
func (uncachedStore) Delete(context.Context, string) error               { return errUncached }