// Ported from lchat internal/activator/replica.go.
//
// Package lease holds the replication primitives for instance
// hibernation (issue #550 Part 4 v2): litestream embedded as a library — pure
// modernc.org/sqlite, no cgo — with its S3 lease (lock.json via conditional
// writes) wrapped around replication as a single-writer lock. The lease is
// the truth of ownership: a second claimant fails loudly with a typed
// *HeldError (the CLI, by contrast, silently absorbs a
// second replicator — proven on #550), and a failed renewal means leadership
// is lost and the caller must stand down. The activator (wake/sleep)
// builds on these primitives.
package lease

import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"sync"
	"time"

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

// LeaseTTL is deliberately long — 30× litestream's 30s default. Renewal runs
// at TTL/3 and each renewal is one S3 Tier-1 PUT, so at the old 30s TTL the
// lease heartbeat alone was 8,640 PUTs/day per awake instance (~$16/yr —
// ~70% of the fleet's entire S3 bill, and more than a €5/y space's revenue);
// at 15m it's 288/day (~$0.50/yr). The price is the failover window: a
// HARD-crashed holder (kill -9, box death) strands the lease until expiry,
// so a peer's takeover — or a re-wake on the same box — can stall up to
// ~15m. Every graceful path (sleep, deploy, shutdown) releases the lease
// explicitly and is unaffected. Shorten this only if fast crash-failover
// ever outranks per-instance cost.
const LeaseTTL = 15 * time.Minute

// compactionLevels replaces litestream.DefaultCompactionLevels (L1 30s /
// L2 5m / L3 1h) with a far gentler ladder. The defaults are tuned for a
// long-lived replicator where background compaction promptness matters; a
// hibernating instance instead checkpoints deterministically at sleep via
// SyncAndWait (which calls db.Sync + Replica.Sync directly, independent of
// every interval here), so aggressive background compaction buys nothing —
// it only burns S3 Tier-1 LISTs: each level's monitor tick costs up to two
// LISTs per DB even with zero writes — L1 at 30s dominated the fleet's idle
// request bill, and the first fix (5m/30m/3h) still left ~1.3k LISTs/day per
// awake instance once the lease PUTs were tamed.
// Correctness is untouched — compaction is purely a restore-time
// read-amplification optimisation (restore replays snapshot + whatever L0
// files exist), and each level's monitor still runs one immediate pass at
// store.Open, so every wake compacts anything a previous short awake window
// left pending. Same three-level geometric shape, so restore reads stay
// bounded: worst case a hard crash replays up to ~an hour of L0 files.
var compactionLevels = litestream.CompactionLevels{
	{Level: 0, Interval: 0},
	{Level: 1, Interval: 1 * time.Hour},
	{Level: 2, Interval: 6 * time.Hour},
	{Level: 3, Interval: 24 * time.Hour},
}

// SnapshotInterval and SnapshotRetention are the fleet's point-in-time
// recovery window, for every replica the platform writes: this store (a
// hibernating instance's) and the box-wide /etc/litestream.yml, which
// internal/ops writes from these same two constants so they cannot drift.
//
// litestream 0.5's DefaultSnapshotRetention is 24h. That covers a dead box;
// it does not cover LOGICAL corruption, where integrity_check says "ok" and
// the data is simply wrong. A bad migration or an errant DELETE is
// typically noticed days later, and by then the only copy is the
// faithfully-replicated bad one. Here the default was worse than a stated
// day: retention runs in the snapshot monitor, which runs only while the
// instance is awake, so each wake pruned every snapshot older than a day
// (#263), and `carlos db pull --at` reached back about a day.
//
// A week is the deliberate choice. It adds no requests: the snapshot
// monitor's cadence is SnapshotInterval either way, and only how far back
// its deletes reach changes. It does add storage, and more than the seven
// snapshots: retention keeps every L1-L3 file (compactionLevels: hourly,
// 6-hourly, daily) back to the oldest kept snapshot too. Each holds only
// the pages that changed in its window, so a typical small, lightly
// written database adds little. The worst case is a database rewriting
// most of its pages every hour: about 168 L1 + 28 L2 + 7 L3 files plus 7
// snapshots, some 200 database sizes (100 MiB -> ~20 GiB, ~$0.50/mo at
// ~$0.023/GB-mo), against about 30 under the old 24h. Lengthen it if a
// restore is ever wanted from further back than "we noticed within the
// week"; shorten it if the bucket's size says the worst case is common.
// (Astra's review of #263 corrected an estimate that counted only the
// snapshots.)
const (
	SnapshotInterval  = 24 * time.Hour
	SnapshotRetention = 7 * 24 * time.Hour
)

// dbMonitorInterval slows litestream's DB monitor (default 1s) — the local
// WAL poll that shovels new frames into shadow LTX. It makes no S3 calls,
// but 1s polling per awake instance is pointless churn when the sleep-time
// SyncAndWait ships everything regardless; 10s bounds an awake instance's
// crash-loss window to ~10s, plenty tight for chat.
// (Replica.SyncInterval stays at its 1s default on purpose: idle replica
// ticks make zero S3 calls — the upload loop no-ops when the local TXID
// hasn't moved — and while awake it bounds crash-loss to ~1s.)
const dbMonitorInterval = 10 * time.Second

// l0RetentionCheckInterval slows the store's L0 retention monitor (default
// 15s — tuned for litestream's VFS read replicas, which we don't run). Every
// tick LISTs L1 and usually L0 per DB even with zero writes — the single
// biggest idle-request source. L0 files only become reclaimable after an L1
// compaction, so checking more often than the L1 interval (1h above) is pure
// waste; matching it cuts the LISTs 240× vs the default with no change to
// what gets deleted.
const l0RetentionCheckInterval = 1 * time.Hour

// HeldError reports that another process holds the single-writer lease. It
// wraps litestream's *LeaseExistsError so callers of this package never need
// to import litestream to react to a contested lease (check with errors.As).
type HeldError struct {
	Owner     string
	ExpiresAt time.Time
}

func (e *HeldError) Error() string {
	if e.Owner != "" {
		return fmt.Sprintf("lease already held by %s until %s", e.Owner, e.ExpiresAt.Format(time.RFC3339))
	}
	return fmt.Sprintf("lease already held until %s", e.ExpiresAt.Format(time.RFC3339))
}

// Held is an acquired single-writer lease that is not yet coupled to
// replication. The wake path needs this split: the DB restore must happen
// *under* the lock (restoring without it could later shadow newer S3 state
// written by the true holder), but replication can only start once the DB
// file exists. Renewal runs from acquisition, so a slow restore can't silently
// outlive the TTL.
type Held struct {
	leaser           *lss3.Leaser
	region           string // the leaser's client keeps its region private; carry it for Replicate
	rn               *renewer
	mu               sync.Mutex
	busy             bool
	spent            bool
	poison           error
	replicaClient    litestream.ReplicaClient // private local-replica test seam
	generationPrefix string
	// owed is the lease record a failed Release could not delete (#292).
	// Release spends the handle either way; RetryRelease re-issues the
	// same conditional delete until it lands or the lease is no longer
	// ours. Guarded by mu.
	owed *litestream.Lease
}

// Acquire claims the single-writer lease at s3://bucket/prefix/lock.json
// and starts renewing it. If another process holds the lease it fails LOUDLY
// with a *HeldError (check with errors.As) — this is the
// signal the activator reacts to. Region may be empty; the AWS default chain
// (env/profile) resolves it, or — under a CARLOS_S3_ENDPOINT override, where
// there is no chain to resolve from — endpoint.go's non-AWS default.
func Acquire(ctx context.Context, bucket, prefix, region string) (*Held, error) {
	return AcquireTTL(ctx, bucket, prefix, region, LeaseTTL)
}

// AcquireTTL is Acquire with the lease TTL as a parameter. Production
// Acquire passes LeaseTTL; the moves staging-work lease passes its longer
// term (moves.StagingLeaseTTL, whose renewals have to leave a SigV4
// signature's whole life in front of SafeUntil); tests pass a short TTL so
// an expiry-and-takeover sequence (the failover race) fits inside a test
// run instead of 15 minutes. The TTL governs renewal spacing too:
// startRenewer ticks at leaser.TTL/3, so a short-TTL lease is renewed
// proportionally often and lapses proportionally fast, which is what
// makes "the holder went dark" observable.
func AcquireTTL(ctx context.Context, bucket, prefix, region string, ttl time.Duration) (*Held, error) {
	// The leaser wants a ready-made S3 client (unlike the replica client,
	// which builds its own). Default AWS credential chain, and the
	// CARLOS_S3_ENDPOINT override applied — see endpoint.go for why the lock
	// and the data it guards must resolve to the same store. The region is
	// resolved ONCE, here, and carried on the Held: Replicate's replica client
	// has to sign for whatever this client signed for.
	began := time.Now()
	client, region, err := newLeaserClient(ctx, region)
	if err != nil {
		return nil, err
	}
	built := time.Now()

	leaser := lss3.NewLeaser()
	leaser.SetClient(client)
	leaser.Bucket = bucket
	leaser.Path = prefix // lock lives at <prefix>/lock.json, colocated with the replica
	leaser.TTL = ttl

	started := time.Now()
	lease, err := createLease(ctx, client, leaser)
	if errors.Is(err, errLockPresent) {
		lease, err = leaser.AcquireLease(ctx)
	}
	if err != nil {
		// A *litestream.LeaseExistsError means someone else holds it —
		// re-surfaced as *HeldError so callers stay litestream-free.
		var lexists *litestream.LeaseExistsError
		if errors.As(err, &lexists) {
			return nil, fmt.Errorf("acquire lease: %w", &HeldError{Owner: lexists.Owner, ExpiresAt: lexists.ExpiresAt})
		}
		return nil, fmt.Errorf("acquire lease: %w", err)
	}
	rn := startRenewer(leaser, lease, started, marginForTTL(ttl))
	if err := rn.Err(); err != nil {
		cur := rn.stop()
		relCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = leaser.ReleaseLease(relCtx, cur)
		return nil, err
	}
	// Journal only: the wake's lease step, split into the client and the
	// lock's own read and conditional write.
	log.Printf("lease: %s: acquired in %s (client %s, lock read and write %s)", prefix,
		time.Since(began).Round(time.Millisecond), built.Sub(began).Round(time.Millisecond), time.Since(started).Round(time.Millisecond))
	return &Held{leaser: leaser, region: region, rn: rn}, nil
}

// Err refuses a spent or failed handle as well as lost ownership.
func (h *Held) Err() error {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.spent {
		return ErrSpent
	}
	if h.poison != nil {
		return h.poison
	}
	return h.rn.Err()
}

// Lost closes once on renewal failure or the safety deadline. Healthy transfer
// and explicit release do not close it. A worker must recheck Err/SafeUntil and
// prove its process tree stopped before the deadline; notification is not proof.
func (h *Held) Lost() <-chan struct{} { return h.rn.lost }

// SafeUntil retains Go's monotonic clock component. It must not be serialized
// and later treated as ownership evidence; callers must keep checking health.
func (h *Held) SafeUntil() (time.Time, error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.spent {
		return time.Time{}, ErrSpent
	}
	if h.poison != nil {
		return time.Time{}, h.poison
	}
	return h.rn.SafeUntil()
}

func (h *Held) begin(release bool) error {
	h.mu.Lock()
	defer h.mu.Unlock()
	if h.spent {
		return ErrSpent
	}
	if h.busy {
		return ErrBusy
	}
	if !release {
		if h.poison != nil {
			return h.poison
		}
		if err := h.rn.Err(); err != nil {
			return err
		}
	}
	h.busy = true
	return nil
}
func (h *Held) finish(spent bool, poison error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.busy = false
	h.spent = spent
	if poison != nil {
		h.poison = poison
	}
}

// Release is the only operation allowed after a failed checkpoint. A failed
// remote release still spends this handle; it cannot authorize more work.
func (h *Held) Release(ctx context.Context) error {
	if err := h.begin(true); err != nil {
		return err
	}
	defer h.finish(true, nil)
	lease := h.rn.stop()
	if lease == nil {
		return nil
	}
	err := h.leaser.ReleaseLease(ctx, lease)
	if err != nil && !releaseSettled(err) {
		// The handle is spent, but the lease record may still stand: keep
		// it so RetryRelease can delete it, on the same condition (#292).
		h.mu.Lock()
		h.owed = lease
		h.mu.Unlock()
	}
	return err
}

// releaseSettled reports whether a release error still leaves nothing of
// this lease to delete: the record is already gone, or its ETag no longer
// matches (the lease lapsed and is someone else's now).
func releaseSettled(err error) bool {
	return errors.Is(err, lss3.ErrLeaseAlreadyReleased) || errors.Is(err, litestream.ErrLeaseNotHeld)
}

// ReleaseOwed reports whether a failed Release left the lease record
// standing for RetryRelease.
func (h *Held) ReleaseOwed() bool {
	h.mu.Lock()
	defer h.mu.Unlock()
	return h.owed != nil
}

// ErrReleaseLapsed is RetryRelease's answer when the delete failed again
// but the owed record has passed its own expiry: the lease is free to any
// acquirer by then, so the obligation is retired rather than retried
// forever (#292). It wraps the delete's own error.
var ErrReleaseLapsed = errors.New("lease release retired: the lease record has passed its expiry")

// RetryRelease discharges what a failed Release left owed (#292): the same
// conditional delete (If-Match on the lease's ETag), so it can never delete
// a lease someone else has taken since. nil once nothing is owed, including
// when the record is already gone or is no longer ours. A delete that fails
// again once the record has passed its own expiry retires the obligation
// with ErrReleaseLapsed: under a persistent refusal (AccessDenied, say)
// obligations would otherwise pile up and be retried forever. Callers do
// not run it concurrently with itself.
func (h *Held) RetryRelease(ctx context.Context) error {
	h.mu.Lock()
	lease := h.owed
	h.mu.Unlock()
	if lease == nil {
		return nil
	}
	if err := h.leaser.ReleaseLease(ctx, lease); err != nil && !releaseSettled(err) {
		if lease.ExpiresAt.IsZero() || time.Now().Before(lease.ExpiresAt) {
			return err
		}
		h.mu.Lock()
		if h.owed == lease {
			h.owed = nil
		}
		h.mu.Unlock()
		return fmt.Errorf("%w (expired %s): %w", ErrReleaseLapsed, lease.ExpiresAt.UTC().Format(time.RFC3339), err)
	}
	h.mu.Lock()
	if h.owed == lease {
		h.owed = nil
	}
	h.mu.Unlock()
	return nil
}

// Replicate trades the held lease for running replication of dbPath to the
// lease's S3 prefix. On success the Held is spent — renewal duty moves to
// the returned Replicator (Close it to release). On failure the lease is
// STILL HELD; the caller decides whether to retry or Release.
//
// syncInterval sets litestream's replica sync tick — the per-plan durability
// ladder's one knob. It is a MINIMUM spacing between S3 syncs (the monitor's
// change-notify only short-circuits the idle wait, never this spacing), so 5m
// bounds an awake host's hard-crash loss to ~5m and 24h to ~a day, while the
// wake-time initial pass and the sleep-time SyncAndWait/Close flush in full
// regardless — the interval never weakens hibernation's parked copy. 0 keeps
// litestream's 1s default (the stream class): idle ticks make zero S3 calls,
// so stream's request bill is change-driven, not clock-driven.
func (h *Held) Replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (*Replicator, error) {
	if err := h.begin(false); err != nil {
		return nil, err
	}
	transferred := false
	defer func() { h.finish(transferred, nil) }()
	opCtx, cancel := context.WithCancel(ctx)
	stop := context.AfterFunc(h.rn.ctx, cancel)
	defer stop()
	defer cancel()
	store, db, err := h.newStore(dbPath, syncInterval)
	if err != nil {
		return nil, err
	}
	if err := store.Open(opCtx); err != nil {
		cancel()
		closeErr := store.Close(opCtx)
		return nil, errors.Join(fmt.Errorf("open litestream store: %w", err), closeErr)
	}
	if err := h.rn.Err(); err != nil {
		cancel()
		closeErr := store.Close(opCtx)
		return nil, errors.Join(err, closeErr)
	}
	transferred = true
	return &Replicator{leaser: h.leaser, store: store, db: db, rn: h.rn}, nil
}

func (h *Held) newStore(dbPath string, syncInterval time.Duration) (*litestream.Store, *litestream.DB, error) {
	db := litestream.NewDB(dbPath)
	// Preserve the activator namespace: sharing the default metadata directory
	// with another replicator corrupts shadow state even across sequential runs.
	db.SetMetaPath(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream"))
	db.MonitorInterval = dbMonitorInterval
	client := h.replicaClient
	if client == nil {
		var err error
		client, err = newReplicaClient(h.leaser.Bucket, h.leaser.Path, h.region)
		if err != nil {
			return nil, nil, err
		}
	}
	db.Replica = litestream.NewReplicaWithClient(db, client)
	if syncInterval > 0 {
		db.Replica.SyncInterval = syncInterval
	}
	store := litestream.NewStore([]*litestream.DB{db}, compactionLevels)
	store.SnapshotInterval = SnapshotInterval
	store.SnapshotRetention = SnapshotRetention
	if h.generationPrefix != "" {
		// NewStore resets the DB flag. Set all three store/DB/compactor flags
		// afterward, before Open starts any monitor or compaction work.
		store.SetRetentionEnabled(false)
	}
	store.L0RetentionCheckInterval = l0RetentionCheckInterval
	return store, db, nil
}

// Replicator couples an exclusive S3 lease with a running litestream
// replication of one SQLite database to the same S3 prefix. The lease is
// acquired *before* replication starts and renewed on a ticker; Close stops
// replication and releases the lease.
type Replicator struct {
	leaser        *lss3.Leaser
	store         *litestream.Store
	db            *litestream.DB
	rn            *renewer
	mu            sync.Mutex
	busy          bool
	spent         bool
	poison        error
	replicaClient litestream.ReplicaClient // private local-replica test seam
}

// ReplicateWithLease acquires the single-writer lease at
// s3://bucket/prefix/lock.json and, only if that succeeds, starts litestream
// replication of dbPath to s3://bucket/prefix at the default (stream-class)
// sync interval. If another process holds the lease it fails LOUDLY with a
// *HeldError (check with errors.As) — this is the signal
// the activator reacts to.
func ReplicateWithLease(ctx context.Context, dbPath, bucket, prefix, region string) (*Replicator, error) {
	held, err := Acquire(ctx, bucket, prefix, region)
	if err != nil {
		return nil, err
	}
	r, err := held.Replicate(ctx, dbPath, 0)
	if err != nil {
		_ = held.Release(ctx)
		return nil, err
	}
	return r, nil
}

// SyncAndWait forces a full sync of the database to the replica and blocks
// until the remote position catches up — used by tests to make replication
// observable without waiting on monitor intervals.
func (r *Replicator) SyncAndWait(ctx context.Context) error {
	return r.db.SyncAndWait(ctx)
}

// Err reports whether lease renewal has failed (leadership lost).
func (r *Replicator) Err() error { return r.rn.Err() }

func (r *Replicator) SafeUntil() (time.Time, error) { return r.rn.SafeUntil() }

// Lost is the same loss notification channel held before replication began.
// Normal Close does not signal loss; SafeUntil refuses the stopped handle.
func (r *Replicator) Lost() <-chan struct{} { return r.rn.lost }

// Abandon stops lease renewal and replication WITHOUT the final sync or the
// lease release — the wreckage a crashed box leaves behind: lock.json simply
// expires on its TTL, and the replica holds whatever was last synced. It
// exists so tests can simulate hard box death in-process; production code has
// no reason to call it (a real crash needs no help).
func (r *Replicator) Abandon(ctx context.Context) {
	r.rn.stop() // deliberately no ReleaseLease — the TTL is the point
	// litestream's only teardown is Close, which includes a sync a real crash
	// would skip; it ships nothing as long as the caller killed the writer
	// first, and reclaims the replication goroutines either way.
	_ = r.store.Close(ctx)
}

// Close stops replication (final sync included via litestream's DB.Close)
// and releases the lease so the next claimant can acquire immediately.
func (r *Replicator) Close(ctx context.Context) error {
	// Force one full sync before teardown. litestream's DB.Close only
	// final-syncs an *initialized* DB, and initialization happens on the
	// monitor's first pass — so a checkpoint within ~a second of Replicate
	// could silently ship nothing (a drain right after a wake, caught by the
	// mesh live test: the replica held no restorable snapshot). SyncAndWait
	// forces init + a replica sync; its error is deliberately dropped — with
	// the DB now initialized, store.Close's own final sync is the one that
	// counts, and its error is returned.
	_ = r.db.SyncAndWait(ctx)
	err := r.store.Close(ctx)

	lease := r.rn.stop()
	if lease != nil {
		// The release gets its OWN budget, never the caller's: a wedged final
		// sync exhausts ctx, and releasing under a dead context leaves the
		// lease held until TTL — the next wake then 503s for up to the full
		// TTL over nothing (hit live on the first real instance flip). The
		// release is
		// one small S3 DELETE; five seconds is generous.
		relCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if e := r.leaser.ReleaseLease(relCtx, lease); e != nil && err == nil {
			err = e
		}
	}
	return err
}

// Restore pulls the newest replicated state of s3://bucket/prefix into
// dbPath via litestream's restore API — the cold-wake path when the local DB
// is missing. Returns restored=false with a nil error when the prefix holds
// no backup at all (a brand-new instance that has never replicated boots
// fresh instead). The caller must hold the prefix's lease: restoring outside
// it could resurrect a stale copy that later shadows the true holder's newer
// state.
func Restore(ctx context.Context, dbPath, bucket, prefix, region string) (bool, error) {
	// Resolved the same way the leaser's client resolves it — a restore that
	// signed for a different region than the lease guarding it would be a
	// silent split against a store that checks.
	region, err := resolveRegion(ctx, region)
	if err != nil {
		return false, err
	}
	client, err := newReplicaClient(bucket, prefix, region)
	if err != nil {
		return false, err
	}

	// A nested row's instance directory may not exist yet the first time it
	// wakes: the directory is created by the spawn path (isolated-exec
	// design §3.1), which runs AFTER this restore. Without this, a fresh
	// nested row's very first wake fails with ENOENT before it ever gets to
	// create anything.
	if err := os.MkdirAll(filepath.Dir(dbPath), 0o700); err != nil {
		return false, fmt.Errorf("create %s: %w", filepath.Dir(dbPath), err)
	}

	// A nil *DB is litestream's own pattern for URL restores (the CLI does the
	// same): Restore only needs the replica client and the output path.
	r := litestream.NewReplicaWithClient(nil, client)
	opt := litestream.NewRestoreOptions()
	opt.OutputPath = dbPath
	if err := r.Restore(ctx, opt); err != nil {
		if errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}