// 🤖 Package activator 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
// *litestream.LeaseExistsError (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 Phase 2 activator (wake/sleep,
// activator.go in the main package) builds on these primitives.
package activator

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

	awsconfig "github.com/aws/aws-sdk-go-v2/config"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	"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},
}

// 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

// renewer keeps a held lease alive on a ticker. A failed renewal is recorded
// and renewal stops — leadership is lost and the owner must stand down;
// retrying could fence with a new claimant who took over after expiry.
type renewer struct {
	leaser *lss3.Leaser

	mu    sync.Mutex
	lease *litestream.Lease
	err   error // set if a renewal ever fails (leadership lost)

	cancel context.CancelFunc
	done   chan struct{}
}

func startRenewer(leaser *lss3.Leaser, lease *litestream.Lease) *renewer {
	ctx, cancel := context.WithCancel(context.Background())
	rn := &renewer{leaser: leaser, lease: lease, cancel: cancel, done: make(chan struct{})}
	go func() {
		defer close(rn.done)
		t := time.NewTicker(LeaseTTL / 3)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				rn.mu.Lock()
				cur := rn.lease
				rn.mu.Unlock()
				renewed, err := rn.leaser.RenewLease(ctx, cur)
				rn.mu.Lock()
				if err != nil {
					rn.err = err
				} else {
					rn.lease = renewed
				}
				rn.mu.Unlock()
				if err != nil {
					return
				}
			}
		}
	}()
	return rn
}

// Err reports whether a renewal has failed (leadership lost).
func (rn *renewer) Err() error {
	rn.mu.Lock()
	defer rn.mu.Unlock()
	return rn.err
}

// stop halts renewal and returns the most recent lease (for release, or to
// hand renewal duty to a successor renewer).
func (rn *renewer) stop() *litestream.Lease {
	rn.cancel()
	<-rn.done
	rn.mu.Lock()
	defer rn.mu.Unlock()
	return rn.lease
}

// HeldLease 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 HeldLease struct {
	leaser *lss3.Leaser
	region string // the leaser's client keeps its region private; carry it for Replicate
	rn     *renewer
}

// AcquireLease 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 *litestream.LeaseExistsError (check with errors.As) — this is the
// signal the activator reacts to. Region may be empty; the AWS default chain
// (env/profile) resolves it.
func AcquireLease(ctx context.Context, bucket, prefix, region string) (*HeldLease, error) {
	// The leaser wants a ready-made S3 client (unlike the replica client,
	// which builds its own). Default AWS credential chain.
	var opts []func(*awsconfig.LoadOptions) error
	if region != "" {
		opts = append(opts, awsconfig.WithRegion(region))
	}
	cfg, err := awsconfig.LoadDefaultConfig(ctx, opts...)
	if err != nil {
		return nil, fmt.Errorf("load aws config: %w", err)
	}

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

	lease, err := leaser.AcquireLease(ctx)
	if err != nil {
		// Includes *litestream.LeaseExistsError when someone else holds it.
		return nil, fmt.Errorf("acquire lease: %w", err)
	}
	return &HeldLease{leaser: leaser, region: region, rn: startRenewer(leaser, lease)}, nil
}

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

// Release stops renewal and releases the lease — the bail-out when a wake
// fails between acquisition and replication.
func (h *HeldLease) Release(ctx context.Context) error {
	lease := h.rn.stop()
	if lease == nil {
		return nil
	}
	return h.leaser.ReleaseLease(ctx, lease)
}

// Replicate trades the held lease for running replication of dbPath to the
// lease's S3 prefix. On success the HeldLease is spent — renewal duty moves to
// the returned LeasedReplicator (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 *HeldLease) Replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (*LeasedReplicator, error) {
	if err := h.rn.Err(); err != nil {
		return nil, fmt.Errorf("lease already lost: %w", err)
	}

	db := litestream.NewDB(dbPath)
	// The activator's replication state lives in its OWN meta directory —
	// litestream derives ".<db>-litestream" beside the DB by default, which any
	// other litestream process over the same file (the box-wide backup
	// replicator, before or after a handover) also uses. Sharing corrupts LTX
	// state even SEQUENTIALLY: the fleet-wide flip left a dozen hosts in an
	// error loop until their meta dirs were cleared by hand. A distinct
	// namespace ends the class.
	db.SetMetaPath(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream"))
	db.MonitorInterval = dbMonitorInterval
	client := lss3.NewReplicaClient()
	client.Bucket = h.leaser.Bucket
	client.Path = h.leaser.Path
	client.Region = h.region
	db.Replica = litestream.NewReplicaWithClient(db, client)
	if syncInterval > 0 {
		db.Replica.SyncInterval = syncInterval
	}

	store := litestream.NewStore([]*litestream.DB{db}, compactionLevels)
	store.L0RetentionCheckInterval = l0RetentionCheckInterval
	if err := store.Open(ctx); err != nil {
		return nil, fmt.Errorf("open litestream store: %w", err)
	}

	// Hand renewal to the replicator. The handoff gap between stopping one
	// renewer and starting the next is microseconds against a 30s TTL.
	lease := h.rn.stop()
	return &LeasedReplicator{leaser: h.leaser, store: store, db: db, rn: startRenewer(h.leaser, lease)}, nil
}

// LeasedReplicator 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 LeasedReplicator struct {
	leaser *lss3.Leaser
	store  *litestream.Store
	db     *litestream.DB
	rn     *renewer
}

// 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
// *litestream.LeaseExistsError (check with errors.As) — this is the signal
// the activator reacts to.
func ReplicateWithLease(ctx context.Context, dbPath, bucket, prefix, region string) (*LeasedReplicator, error) {
	held, err := AcquireLease(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 *LeasedReplicator) SyncAndWait(ctx context.Context) error {
	return r.db.SyncAndWait(ctx)
}

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

// 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 *LeasedReplicator) 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 *LeasedReplicator) 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) {
	client := lss3.NewReplicaClient()
	client.Bucket = bucket
	client.Path = prefix
	client.Region = region

	// 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
}