package lease

import (
	"context"
	"errors"
	"sync"
	"time"

	"github.com/benbjohnson/litestream"
	lss3 "github.com/benbjohnson/litestream/s3"
)

var (
	ErrLost  = errors.New("lease safety deadline elapsed")
	ErrSpent = errors.New("lease handle is spent")
	ErrBusy  = errors.New("lease operation already running")
)

const terminationMargin = 5 * time.Second

// RenewMargin is the margin Held.SafeUntil subtracts from the lease's
// expiry at production TTLs: SafeUntil is at most TTL-RenewMargin ahead of
// a fresh acquire. The moves staging guard reasons from it.
const RenewMargin = terminationMargin

// Production uses the fixed margin. Short TTLs exist only through AcquireTTL
// (tests; production callers pass LeaseTTL or the moves staging term).
func marginForTTL(ttl time.Duration) time.Duration {
	if ttl < 3*terminationMargin {
		return ttl / 10
	}
	return terminationMargin
}

type renewer struct {
	ctx       context.Context
	leaser    *lss3.Leaser
	mu        sync.Mutex
	lease     *litestream.Lease
	err       error
	until     time.Time
	margin    time.Duration
	stopped   bool
	cancel    context.CancelFunc
	done      chan struct{}
	watchDone chan struct{}
	changed   chan struct{}
	lost      chan struct{}
}

func startRenewer(leaser *lss3.Leaser, lease *litestream.Lease, started time.Time, margin time.Duration) *renewer {
	ctx, cancel := context.WithCancel(context.Background())
	rn := &renewer{ctx: ctx, leaser: leaser, lease: lease, until: started.Add(leaser.TTL - margin), margin: margin, cancel: cancel, done: make(chan struct{}), watchDone: make(chan struct{}), changed: make(chan struct{}, 1), lost: make(chan struct{})}
	go rn.watch(ctx)
	go rn.renew(ctx)
	return rn
}

func (rn *renewer) loseLocked(err error) {
	if rn.err == nil {
		rn.err = err
		close(rn.lost)
	}
}

// checkLocked latches loss permanently; an acknowledgement arriving after a
// blocked renewal cannot resurrect ownership, even before the timer runs.
func (rn *renewer) checkLocked() error {
	if rn.err != nil {
		return rn.err
	}
	if rn.stopped {
		return ErrSpent
	}
	if !time.Now().Before(rn.until) {
		rn.loseLocked(ErrLost)
	}
	return rn.err
}
func (rn *renewer) Err() error {
	rn.mu.Lock()
	defer rn.mu.Unlock()
	return rn.checkLocked()
}
func (rn *renewer) SafeUntil() (time.Time, error) {
	rn.mu.Lock()
	defer rn.mu.Unlock()
	if err := rn.checkLocked(); err != nil {
		return time.Time{}, err
	}
	return rn.until, nil
}
func (rn *renewer) watch(ctx context.Context) {
	defer close(rn.watchDone)
	for {
		until, err := rn.SafeUntil()
		if err != nil {
			rn.cancel()
			return
		}
		timer := time.NewTimer(time.Until(until))
		select {
		case <-ctx.Done():
			timer.Stop()
			return
		case <-rn.changed:
			timer.Stop()
		case <-timer.C:
		}
	}
}
func (rn *renewer) renew(ctx context.Context) {
	defer close(rn.done)
	every := rn.leaser.TTL / 3
	if every <= 0 {
		every = time.Nanosecond
	}
	ticker := time.NewTicker(every)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			rn.mu.Lock()
			err := rn.checkLocked()
			cur := rn.lease
			rn.mu.Unlock()
			if err != nil {
				rn.cancel()
				return
			}
			started := time.Now()
			renewed, err := rn.leaser.RenewLease(ctx, cur)
			rn.mu.Lock()
			health := rn.checkLocked()
			// Retain the newest ETag solely for conditional cleanup, including a late
			// success. It never repairs a latched loss or extends a stopped handle.
			if err == nil {
				rn.lease = renewed
			}
			if health == nil {
				if err != nil {
					rn.loseLocked(err)
				} else {
					rn.until = started.Add(rn.leaser.TTL - rn.margin)
					rn.checkLocked()
				}
			}
			failed := rn.err != nil || rn.stopped
			rn.mu.Unlock()
			if failed {
				rn.cancel()
				return
			}
			select {
			case rn.changed <- struct{}{}:
			default:
			}
		}
	}
}

func (rn *renewer) stop() *litestream.Lease {
	rn.mu.Lock()
	rn.checkLocked()
	rn.stopped = true
	rn.mu.Unlock()
	rn.cancel()
	<-rn.done
	<-rn.watchDone
	rn.mu.Lock()
	defer rn.mu.Unlock()
	return rn.lease
}