package lease

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"path"
	"strings"
	"time"

	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/benbjohnson/litestream"
)

// ErrLockLocation is LockState's answer for a bucket or location it will
// not read: empty, absolute, or not in clean form.
var ErrLockLocation = errors.New("lease: invalid lock location")

// lockBodyLimit bounds the lock.json read. The leaser writes a few dozen
// bytes; anything near this is not a lock.
const lockBodyLimit = 64 << 10

// LockState reads the single-writer lease at s3://bucket/<location>/lock.json
// as the litestream leaser writes it (Acquire, lease.go), through the same
// client Acquire builds, so the read signs for the region and reaches the
// endpoint the writer used. It is §5.3 step 4's observation: it holds no
// lease and writes nothing.
//
// location is the claim holder's Location from a fresh claim read (the
// location moves.ReplicaSource derives its prefix from, ruling G3), never
// activator.HibernatePrefix and never the g-<generation>/ prefix: the lock
// sits at the location and Held.BindGeneration only fixes the data under
// it. One trailing slash is accepted, so ReplicaSource's root prefix reads
// the same lock.
//
// held is true while the lock is present and its expiry has not passed by
// more than RenewMargin, so a holder whose clock runs slightly behind
// still counts as holding. An absent lock (never acquired, or released:
// the leaser deletes it) is held=false with a zero owner and expiry and a
// nil error; only S3's explicit NoSuchKey counts as absent (lockAbsent).
// An expired lock is held=false with its owner and expiry, for
// the record. Anything else (a missing bucket, an unreadable or garbled
// body, a lock with no expiry) is an error, never "not held".
func LockState(ctx context.Context, bucket, location, region string) (held bool, owner string, expires time.Time, err error) {
	key, err := lockKey(bucket, location)
	if err != nil {
		return false, "", time.Time{}, err
	}
	client, _, err := newLeaserClient(ctx, region)
	if err != nil {
		return false, "", time.Time{}, err
	}
	out, err := client.GetObject(ctx, &awss3.GetObjectInput{Bucket: &bucket, Key: &key})
	if err != nil {
		if lockAbsent(err) {
			return false, "", time.Time{}, nil
		}
		return false, "", time.Time{}, fmt.Errorf("lease: read %s: %w", key, err)
	}
	defer out.Body.Close()
	body, err := io.ReadAll(io.LimitReader(out.Body, lockBodyLimit+1))
	if err != nil {
		return false, "", time.Time{}, fmt.Errorf("lease: read %s: %w", key, err)
	}
	if len(body) > lockBodyLimit {
		return false, "", time.Time{}, fmt.Errorf("lease: %s is larger than a lock", key)
	}
	var lock litestream.Lease
	if err := json.Unmarshal(body, &lock); err != nil {
		return false, "", time.Time{}, fmt.Errorf("lease: decode %s: %w", key, err)
	}
	if lock.ExpiresAt.IsZero() {
		return false, "", time.Time{}, fmt.Errorf("lease: %s has no expiry", key)
	}
	return lockHeld(lock.ExpiresAt, time.Now()), lock.Owner, lock.ExpiresAt, nil
}

// lockAbsent reports an explicit S3 object-absence answer: HTTP 404 with an
// error body the SDK parsed to the code NoSuchKey. Absence is what lets a
// cut call the lease free, so nothing weaker counts: not a 404 with a
// malformed, empty or unrecognised body (an endpoint that is not the store,
// a proxy), not NotFound or NoSuchVersion, not NoSuchBucket, and not
// NoSuchKey on any other status. mapReadErr is deliberately not used here:
// its looser reading (any 404 but NoSuchBucket) suits its callers, which
// treat absence as "nothing to copy", not as authority.
func lockAbsent(err error) bool {
	code, status := apiCode(err)
	return code == "NoSuchKey" && status == http.StatusNotFound
}

// lockHeld is §5.3 step 4's rule: a lock is released only once its expiry
// has passed by more than the renewal margin.
func lockHeld(expires, now time.Time) bool {
	return !now.After(expires.Add(RenewMargin))
}

// lockKey is the leaser's key for location (lss3.Leaser.lockKey with Path
// set, as AcquireTTL sets it), after refusing what Acquire would never be
// given: an empty bucket or location, an absolute one, or one not in clean
// form.
func lockKey(bucket, location string) (string, error) {
	loc := strings.TrimSuffix(location, "/")
	if bucket == "" || loc == "" || strings.HasPrefix(loc, "/") || strings.Contains(loc, `\`) || path.Clean(loc) != loc || loc == "." || strings.HasPrefix(loc, "../") || loc == ".." {
		return "", ErrLockLocation
	}
	return loc + "/lock.json", nil
}