package lease

// The tombstone lease: the offline operator move's linearization point.
//
// A move that has stopped a host on its source box writes a lease at the
// host's own lock.json under the owner "moved:<id>" and a long expiry. From
// then on lease.Acquire on that location fails with a *HeldError naming the
// move, so no box can wake the host and write to the replica behind the
// move's back. Three operations manage it, each a conditional write over the
// client AcquireTTL builds (IfNoneMatch "*" on create, IfMatch on replace
// or delete) and none of them starts a renewer: the move's sealer re-asserts
// the tombstone itself on every pass.
//
//   - Tombstone writes it. It never replaces a live holder's lease.
//   - Reassert pushes the expiry out again, only on a lease the same move
//     owns, with the ETag it just read.
//   - LiftTombstone lifts it, only when the owner is a moved:<id> lease the
//     caller claims as its own, with the ETag it just read. It lifts by a
//     conditional PUT that rewrites the lease as already expired, never by a
//     DELETE: Ceph RGW and MinIO are not known to enforce If-Match on a
//     DELETE, and the conditional PUT is the primitive the rest of this
//     package relies on. IsLiftedTombstone reads that state back.
//
// "Free" is exactly what Acquire means by it (leaseFree): the lease's expiry
// has passed. There is no RenewMargin here. A tombstone that is free to a
// wake is free to the next Tombstone, and the two cannot drift apart.

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

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

// TombstonePrefix starts the owner of every tombstone lease. The three
// functions here refuse any other owner.
const TombstonePrefix = "moved:"

// MaxTombstoneTTL is the longest a tombstone is ever written for. The moves
// package's TombstoneTTL equals it; this package cannot import that one.
const MaxTombstoneTTL = 7 * 24 * time.Hour

var (
	// ErrTombstoneOwner refuses an owner that is not moved:<id>.
	ErrTombstoneOwner = errors.New("lease: a tombstone owner is moved:<id>")
	// ErrTombstoneTTL refuses a TTL that is not positive or exceeds
	// MaxTombstoneTTL.
	ErrTombstoneTTL = errors.New("lease: a tombstone lives for at most 7 days")
	// ErrTombstoneChanged reports that the lease was rewritten or removed
	// between the read and the conditional write. Nothing was overwritten;
	// the caller's next pass reads it afresh.
	ErrTombstoneChanged = errors.New("lease: the lock changed under the tombstone write")
	// ErrTombstoneLost reports that the tombstone Reassert was asked to
	// keep is absent or expired: the stop was lost. Nothing was written.
	ErrTombstoneLost = errors.New("lease: the tombstone is absent or expired; the stop was lost")
)

// leaseFree is the predicate litestream's AcquireLease applies to the lease
// it reads (Lease.IsExpired: now is after the expiry), which is what decides
// whether lease.Acquire takes a lock over. Tombstone, Reassert and
// IsLiftedTombstone use it so a lease is free to them exactly when it is free
// to a wake. TestTombstoneAndAcquireAgreeOnFree pins the two together.
func leaseFree(expires, now time.Time) bool { return now.After(expires) }

// liftedExpiry is the expiry LiftTombstone writes: long past, so every
// reading (leaseFree, and LockState's RenewMargin) agrees the lease is free.
var liftedExpiry = time.Unix(0, 0).UTC()

// IsLiftedTombstone reports whether a lock read as owner and expires (from
// LockState) is a lifted tombstone: a moved:<id> lease that has expired. That
// is what LiftTombstone leaves behind, and it is free to every Acquire (which
// already treats any expired lease as free) and to the next Tombstone. It is
// false for an ordinary process's lease, expired or not, and for a tombstone
// that is still unexpired. An absent lock is not "lifted" here: callers read
// absence from LockState's zero owner and expiry and treat it the same way.
func IsLiftedTombstone(owner string, expires time.Time) bool {
	return tombstoneOwnerOK(owner) && !expires.IsZero() && leaseFree(expires, time.Now())
}

func tombstoneOwnerOK(owner string) bool {
	return strings.HasPrefix(owner, TombstonePrefix) && len(owner) > len(TombstonePrefix)
}

func tombstoneTTLOK(ttl time.Duration) bool { return ttl > 0 && ttl <= MaxTombstoneTTL }

// tombstoneLeaser is the leaser a tombstone call writes with, built over the
// client AcquireTTL builds, and the lock's key. ttl is zero for calls that
// write no expiry (LiftTombstone).
func tombstoneLeaser(ctx context.Context, bucket, location, region, owner string, ttl time.Duration) (*lss3.Leaser, string, error) {
	key, err := lockKey(bucket, location)
	if err != nil {
		return nil, "", err
	}
	client, _, err := newLeaserClient(ctx, region)
	if err != nil {
		return nil, "", err
	}
	leaser := lss3.NewLeaser()
	leaser.SetClient(client)
	leaser.Bucket = bucket
	leaser.Path = strings.TrimSuffix(location, "/")
	leaser.TTL = ttl
	leaser.Owner = owner
	return leaser, key, nil
}

// readLock reads the lock with the ETag it was read at. Only S3's explicit
// NoSuchKey is absence (lockAbsent); a garbled body, an oversized one or a
// lock with no expiry is an error, never "free".
func readLock(ctx context.Context, leaser *lss3.Leaser, key string) (lock *litestream.Lease, absent bool, err error) {
	out, err := leaser.Client().GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(leaser.Bucket), Key: aws.String(key)})
	if err != nil {
		if lockAbsent(err) {
			return nil, true, nil
		}
		return nil, false, 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 nil, false, fmt.Errorf("lease: read %s: %w", key, err)
	}
	if len(body) > lockBodyLimit {
		return nil, false, fmt.Errorf("lease: %s is larger than a lock", key)
	}
	var l litestream.Lease
	if err := json.Unmarshal(body, &l); err != nil {
		return nil, false, fmt.Errorf("lease: decode %s: %w", key, err)
	}
	if l.ExpiresAt.IsZero() {
		return nil, false, fmt.Errorf("lease: %s has no expiry", key)
	}
	if out.ETag == nil || *out.ETag == "" {
		return nil, false, fmt.Errorf("lease: %s came back without an ETag", key)
	}
	l.ETag = *out.ETag
	return &l, false, nil
}

func heldErr(l *litestream.Lease) error {
	return fmt.Errorf("acquire lease: %w", &HeldError{Owner: l.Owner, ExpiresAt: l.ExpiresAt})
}

// Tombstone writes the lease at s3://bucket/<location>/lock.json owned by
// owner (moved:<id>), expiring ttl from now (at most MaxTombstoneTTL).
//
//   - absent, or expired (as Acquire reads it): created (IfNoneMatch "*") or
//     replaced (IfMatch on the expired lease's ETag);
//   - live and held by owner: nil and no write; the caller re-asserts with
//     Reassert;
//   - live and held by anyone else: a *HeldError, nothing written;
//   - the lock changed under the write and cannot be read back:
//     ErrTombstoneChanged.
//
// A live holder is never replaced: the replace is conditional on the ETag of
// the lease that was read as expired, so a holder that renews in between
// wins and Tombstone reports it as held.
func Tombstone(ctx context.Context, bucket, location, region, owner string, ttl time.Duration) error {
	if !tombstoneOwnerOK(owner) {
		return ErrTombstoneOwner
	}
	if !tombstoneTTLOK(ttl) {
		return ErrTombstoneTTL
	}
	leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, owner, ttl)
	if err != nil {
		return err
	}
	cur, absent, err := readLock(ctx, leaser, key)
	if err != nil {
		return err
	}
	if !absent && !leaseFree(cur.ExpiresAt, time.Now()) {
		if cur.Owner == owner {
			return nil
		}
		return heldErr(cur)
	}
	// Absent or expired. AcquireLease reads again and writes conditionally:
	// IfNoneMatch "*" when absent, IfMatch on the ETag it reads otherwise.
	// If a holder appeared since our read it answers LeaseExistsError.
	if _, err := leaser.AcquireLease(ctx); err != nil {
		var lexists *litestream.LeaseExistsError
		if errors.As(err, &lexists) {
			switch lexists.Owner {
			case "":
				// The write lost its condition and AcquireLease could not
				// read the winner back (it was removed again, or the read
				// failed). There is no holder to name.
				return fmt.Errorf("tombstone lease: %w", ErrTombstoneChanged)
			case owner:
				return nil
			}
			return fmt.Errorf("acquire lease: %w", &HeldError{Owner: lexists.Owner, ExpiresAt: lexists.ExpiresAt})
		}
		return fmt.Errorf("tombstone lease: %w", err)
	}
	return nil
}

// Reassert moves the expiry of a tombstone owner already holds to now+ttl:
//
//   - live and held by owner: rewritten with IfMatch on the ETag just read.
//     If the lock changed since, ErrTombstoneChanged and nothing is written;
//   - absent, or expired (including lifted), whoever wrote it:
//     ErrTombstoneLost, and nothing is written. The stop was lost: in
//     between a wake may have acquired, written and released the lease, so
//     a tombstone is never taken again here;
//   - live and held by anyone else: a *HeldError, nothing written.
//
// Callers read LockState first and treat absent or expired as a lost stop;
// ErrTombstoneLost is the same verdict for a lock that went between that
// read and this one. Never Reassert after Lift.
func Reassert(ctx context.Context, bucket, location, region, owner string, ttl time.Duration) error {
	if !tombstoneOwnerOK(owner) {
		return ErrTombstoneOwner
	}
	if !tombstoneTTLOK(ttl) {
		return ErrTombstoneTTL
	}
	leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, owner, ttl)
	if err != nil {
		return err
	}
	cur, absent, err := readLock(ctx, leaser, key)
	if err != nil {
		return err
	}
	if absent {
		return fmt.Errorf("reassert tombstone: %w", ErrTombstoneLost)
	}
	if leaseFree(cur.ExpiresAt, time.Now()) {
		return fmt.Errorf("reassert tombstone: %w (%s, expired %s)", ErrTombstoneLost, cur.Owner, cur.ExpiresAt.UTC().Format(time.RFC3339))
	}
	if cur.Owner != owner {
		return heldErr(cur)
	}
	if _, err := leaser.RenewLease(ctx, cur); err != nil {
		if errors.Is(err, litestream.ErrLeaseNotHeld) {
			return fmt.Errorf("reassert tombstone: %w", ErrTombstoneChanged)
		}
		return fmt.Errorf("reassert tombstone: %w", err)
	}
	return nil
}

// LiftTombstone lifts the tombstone at s3://bucket/<location>/lock.json by
// rewriting it as an expired lease of the same owner and generation, with
// IfMatch on the ETag just read. It never deletes the object.
//
//   - absent: (false, nil), and mine is not consulted;
//   - owner is not a moved:<id> that mine accepts: (false, *HeldError),
//     nothing written. This holds for an expired lease too: only the move
//     that wrote a tombstone lifts it;
//   - mine and already lifted (expired, IsLiftedTombstone): (false, nil),
//     nothing written;
//   - mine and live: rewritten, (true, nil). Acquire takes the lease from
//     here, as it takes any expired lease, and so does Tombstone;
//   - the lock was rewritten since the read: (false, ErrTombstoneChanged),
//     and the rewrite stands. One that was removed since is (false, nil).
//
// After a lift, LockState reports held=false with the move's owner and a
// past expiry; IsLiftedTombstone recognises that. Never Reassert after it.
func LiftTombstone(ctx context.Context, bucket, location, region string, mine func(owner string) bool) (lifted bool, err error) {
	leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, "", 0)
	if err != nil {
		return false, err
	}
	cur, absent, err := readLock(ctx, leaser, key)
	if err != nil {
		return false, err
	}
	if absent {
		return false, nil
	}
	if !tombstoneOwnerOK(cur.Owner) || mine == nil || !mine(cur.Owner) {
		return false, heldErr(cur)
	}
	if IsLiftedTombstone(cur.Owner, cur.ExpiresAt) {
		return false, nil
	}
	body, err := json.Marshal(&litestream.Lease{Generation: cur.Generation, ExpiresAt: liftedExpiry, Owner: cur.Owner})
	if err != nil {
		return false, fmt.Errorf("lift tombstone: %w", err)
	}
	_, err = leaser.Client().PutObject(ctx, &awss3.PutObjectInput{
		Bucket:      aws.String(leaser.Bucket),
		Key:         aws.String(key),
		Body:        bytes.NewReader(body),
		ContentType: aws.String("application/json"),
		IfMatch:     aws.String(cur.ETag),
	})
	if err == nil {
		return true, nil
	}
	if code, status := apiCode(err); code == "PreconditionFailed" || status == http.StatusPreconditionFailed || code == "NoSuchKey" || status == http.StatusNotFound {
		// Rewritten, or gone (AWS answers a conditional PUT on a missing key
		// 404, some stores 412). Read it back to tell which.
		if _, gone, rerr := readLock(ctx, leaser, key); rerr == nil && gone {
			return false, nil
		}
		return false, fmt.Errorf("lift tombstone: %w", ErrTombstoneChanged)
	}
	return false, fmt.Errorf("lift tombstone: %w", err)
}