// Name claims: one store per bucket name and per IAM user name (#218).
//
// MintedBucketIn and UserNameIn join account, app and environment with "-",
// and app and environment names may both contain "-" — so app "shop" with
// environment "eu" and app "shop-eu" resolve to ONE bucket and ONE user.
// Nothing downstream notices: EnsureBucket takes BucketAlreadyOwnedByYou as
// success, EnsureUser adopts an existing user, the grant walk's reap deletes
// the other store's live key, and a later Teardown empties and deletes the
// other store's bucket. An operator adopting a bucket by name reaches the
// same place by a typo.
//
// The names themselves cannot change — production buckets and users carry
// them — so the fix is to make a name belong to one store. Same discipline as
// internal/mail/domainclaim.go: a create-only bucket key per name, GLOBAL
// rather than per account (an adopted bucket name is not scoped to the
// account that adopts it), whose write is the uniqueness check. Two walks
// racing for one name cannot both win, in one console or in two, whichever
// store was declared first.
//
// Stores granted before claims existed hold none. The walk backfills a claim
// on its next run, and until then two checks stand in for it: the stores in
// the same account whose names collide by construction are few and
// computable (collidingStores), so they are read directly; and every first
// grant scans every store record, which is the only way to find a name an
// unclaimed record holds in some other account (an operator's adopt can be
// shaped exactly like another account's minted name).
package objstore

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path"
	"strings"
	"time"

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

// ErrStoreNameTaken refuses a grant whose bucket or IAM user name already
// belongs to another store. It is raised before anything is created,
// recorded or reaped.
var ErrStoreNameTaken = errors.New("objstore: bucket or IAM user name belongs to another store")

const nameClaimsPrefix = "objstore/name-claims/"

func bucketClaimKey(bucket string) string { return nameClaimsPrefix + "buckets/" + bucket + ".json" }
func userClaimKey(user string) string     { return nameClaimsPrefix + "users/" + user + ".json" }

// NameClaim is the wire shape at a claim key: which store holds the name.
type NameClaim struct {
	Schema      int       `json:"schema"` // 1
	Name        string    `json:"name"`
	Account     string    `json:"account"`
	App         string    `json:"app"`
	Environment string    `json:"environment,omitempty"`
	CreatedAt   time.Time `json:"created_at"`
}

func (c *NameClaim) heldBy(acct, app, env string) bool {
	return c.Account == acct && c.App == app && c.Environment == env
}

// getNameClaim reads a claim; absent is (nil, nil).
func getNameClaim(ctx context.Context, s release.Store, key string) (*NameClaim, error) {
	raw, _, err := s.Get(ctx, key)
	if errors.Is(err, os.ErrNotExist) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	var c NameClaim
	if err := json.Unmarshal(raw, &c); err != nil {
		return nil, fmt.Errorf("objstore: name claim %s: %w", key, err)
	}
	if c.Schema > 1 {
		return nil, fmt.Errorf("objstore: name claim %s: schema %d is newer than this binary understands — upgrade carlos", key, c.Schema)
	}
	return &c, nil
}

// claimName takes key for the store, or confirms the store already holds it.
// The create-only Put is the check; a lost create race re-reads and judges
// the winner, as claimDomain does.
func claimName(ctx context.Context, s release.Store, key, name, acct, app, env string, now time.Time) error {
	body, err := json.Marshal(NameClaim{Schema: 1, Name: name, Account: acct, App: app, Environment: env, CreatedAt: now})
	if err != nil {
		return err
	}
	for attempt := 0; ; attempt++ {
		cur, err := getNameClaim(ctx, s, key)
		if err != nil {
			return err
		}
		if cur != nil {
			if !cur.heldBy(acct, app, env) {
				return fmt.Errorf("%w: %s", ErrStoreNameTaken, name)
			}
			return nil
		}
		_, err = s.Put(ctx, key, body, "")
		if err == nil {
			return nil
		}
		if !errors.Is(err, release.ErrCASConflict) || attempt > 0 {
			return err
		}
	}
}

// releaseName deletes key only when this store holds it. A claim held by
// another store is theirs, and an absent one is already released.
func releaseName(ctx context.Context, s release.Store, key, acct, app, env string) error {
	cur, err := getNameClaim(ctx, s, key)
	if err != nil {
		return err
	}
	if cur == nil || !cur.heldBy(acct, app, env) {
		return nil
	}
	return s.Delete(ctx, key)
}

type storeRef struct{ app, env string }

// collidingStores lists every OTHER store in the account whose minted bucket
// and IAM user names are the same as this one's: every way of splitting
// app[-env] at a hyphen into a valid app and an optional valid environment.
func collidingStores(app, env string) []storeRef {
	joined := app
	if env != "" {
		joined += "-" + env
	}
	var out []storeRef
	for i := 0; i <= len(joined); i++ {
		var ref storeRef
		switch {
		case i == len(joined):
			ref = storeRef{joined, ""}
		case joined[i] == '-':
			ref = storeRef{joined[:i], joined[i+1:]}
		default:
			continue
		}
		if ref.app == app && ref.env == env {
			continue
		}
		if !release.ValidApp(ref.app) || (ref.env != "" && !appconfig.ValidEnvironmentName(ref.env)) {
			continue
		}
		out = append(out, ref)
	}
	return out
}

// grantedColliders reads the records of collidingStores and returns those a
// grant has acted on — the ones holding the shared names, claimed or not.
func grantedColliders(ctx context.Context, s release.Store, acct, app, env string) ([]*Record, error) {
	var out []*Record
	for _, ref := range collidingStores(app, env) {
		rec, _, err := GetIn(ctx, s, acct, ref.app, ref.env)
		if err != nil {
			return nil, err
		}
		if rec != nil && (rec.Grant != nil || rec.Provisioned != nil) {
			rec.App, rec.Environment = ref.app, ref.env // named by the key read, not the body
			out = append(out, rec)
		}
	}
	return out, nil
}

// recordNames is the bucket and user a record's estate is named by, read the
// way Teardown reads them.
func recordNames(acct string, rec *Record) (bucket, user string) {
	user = UserNameIn(acct, rec.App, rec.Environment)
	if rec.Grant != nil {
		bucket = rec.Grant.Bucket
	}
	if rec.Provisioned != nil {
		bucket, user = rec.Provisioned.Bucket, rec.Provisioned.IAMUser
	}
	return bucket, user
}

// storeNamingBucket scans every store record in the deployment for one other
// than (acct, app, env) that names bucket. Every FIRST grant pays for it, not
// only an operator's adopt: an adopt made before claims existed can hold a
// name shaped exactly like another account's minted one (#191), and nothing
// but the record says so. It walks with ListDirs, as mail.SweepLegacyRecords
// does, so it never lists a release artifact; a re-run skips it, because by
// then the walk holds the claim.
func storeNamingBucket(ctx context.Context, s release.Store, bucket, acct, app, env string) (bool, error) {
	accts, err := s.ListDirs(ctx, "apps/")
	if err != nil {
		return false, err
	}
	for _, acctPrefix := range accts {
		a := path.Base(strings.TrimSuffix(acctPrefix, "/"))
		if !release.ValidAccount(a) {
			continue
		}
		apps, err := s.ListDirs(ctx, acctPrefix)
		if err != nil {
			return false, err
		}
		for _, appPrefix := range apps {
			p := path.Base(strings.TrimSuffix(appPrefix, "/"))
			if !release.ValidApp(p) {
				continue
			}
			envs, err := StoreEnvironments(ctx, s, a, p)
			if err != nil {
				return false, err
			}
			for _, e := range append([]string{""}, envs...) {
				if a == acct && p == app && e == env {
					continue
				}
				rec, _, err := GetIn(ctx, s, a, p, e)
				if err != nil {
					return false, err
				}
				if rec == nil {
					continue
				}
				if b, _ := recordNames(a, rec); b == bucket {
					return true, nil
				}
			}
		}
	}
	return false, nil
}

// claimStoreNames is the grant walk's gate, run before the grant block is
// written and before any backend call. firstGrant is a grant with no block
// yet — the only run that can be choosing a name another record holds.
func claimStoreNames(ctx context.Context, s release.Store, acct, app, env, bucket, user string, firstGrant bool, now time.Time) error {
	others, err := grantedColliders(ctx, s, acct, app, env)
	if err != nil {
		return err
	}
	if len(others) > 0 {
		return fmt.Errorf("%w: %s is already the IAM user of %s/%s environment %q",
			ErrStoreNameTaken, user, acct, others[0].App, others[0].Environment)
	}
	if firstGrant {
		taken, err := storeNamingBucket(ctx, s, bucket, acct, app, env)
		if err != nil {
			return err
		}
		if taken {
			return fmt.Errorf("%w: %s", ErrStoreNameTaken, bucket)
		}
	}
	// Both are read before either is written, so a refusal on the second
	// does not leave the first behind.
	for _, key := range []string{userClaimKey(user), bucketClaimKey(bucket)} {
		cur, err := getNameClaim(ctx, s, key)
		if err != nil {
			return err
		}
		if cur != nil && !cur.heldBy(acct, app, env) {
			return fmt.Errorf("%w: %s", ErrStoreNameTaken, cur.Name)
		}
	}
	if err := claimName(ctx, s, userClaimKey(user), user, acct, app, env, now); err != nil {
		return err
	}
	return claimName(ctx, s, bucketClaimKey(bucket), bucket, acct, app, env, now)
}

// sharedNames reports, for Teardown, whether the bucket or the user it is
// about to delete belongs to another store: by that store's claim, or — for
// a pair granted before claims existed — by a colliding record that names it.
// Either way the other store's estate is not this teardown's to destroy.
func sharedNames(ctx context.Context, s release.Store, acct, app, env, bucket, user string) (bucketShared, userShared bool, err error) {
	if bucket != "" {
		c, err := getNameClaim(ctx, s, bucketClaimKey(bucket))
		if err != nil {
			return false, false, err
		}
		bucketShared = c != nil && !c.heldBy(acct, app, env)
	}
	c, err := getNameClaim(ctx, s, userClaimKey(user))
	if err != nil {
		return false, false, err
	}
	userShared = c != nil && !c.heldBy(acct, app, env)
	others, err := grantedColliders(ctx, s, acct, app, env)
	if err != nil {
		return false, false, err
	}
	for _, o := range others {
		b, u := recordNames(acct, o)
		bucketShared = bucketShared || (bucket != "" && b == bucket)
		userShared = userShared || u == user
	}
	return bucketShared, userShared, nil
}

// claimedBucket finds the bucket claim this store holds, for the one teardown
// that has no record of the name: a grant that took its claims and died
// before writing its block. Only a declared-only teardown pays for the
// listing. "" means none.
func claimedBucket(ctx context.Context, s release.Store, acct, app, env string) (string, error) {
	keys, err := s.List(ctx, nameClaimsPrefix+"buckets/")
	if err != nil {
		return "", err
	}
	for _, key := range keys {
		c, err := getNameClaim(ctx, s, key)
		if err != nil {
			return "", err
		}
		if c != nil && c.heldBy(acct, app, env) {
			return c.Name, nil
		}
	}
	return "", nil
}

// releaseStoreNames gives back whatever claims this store holds on bucket and
// user. Owner-gated, so a name another store holds is never touched.
func releaseStoreNames(ctx context.Context, s release.Store, acct, app, env, bucket, user string) error {
	keys := []string{userClaimKey(user)}
	if bucket != "" {
		keys = append(keys, bucketClaimKey(bucket))
	}
	for _, key := range keys {
		if err := releaseName(ctx, s, key, acct, app, env); err != nil {
			return fmt.Errorf("objstore: release name claim %s: %w", key, err)
		}
	}
	return nil
}