package lease

import (
	"context"
	"database/sql"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"

	"github.com/benbjohnson/litestream"
	"github.com/superfly/ltx"
	_ "modernc.org/sqlite"
)

// restoreForVerify is Restore behind a package var so the drill's three
// outcomes (clean, damaged, never-replicated) are testable without S3.
var restoreForVerify = Restore

// ErrNoReplica reports that a prefix holds no backup at all — a brand-new
// instance that has never replicated, not a damaged one. Verification callers
// must keep the two apart: "nothing to restore yet" is not a failure, and
// reporting it as one would train operators to ignore the alarm.
var ErrNoReplica = errors.New("no replica in prefix")

// CheckDB opens path read-only and proves it is a sane SQLite database:
// integrity_check == ok and at least one table present. It is the one check a
// disaster recovery leans on, so it lives here — beside Restore — where both
// the wake path (internal/activator) and the verification pass (internal/ops)
// can reach it without either importing the other.
//
// The schema check is not pedantry: a restore that yields a valid but empty
// database passes integrity_check and would be useless in the recovery it was
// meant for.
func CheckDB(path string) error {
	return CheckDBContext(context.Background(), path)
}

// CheckDBContext is CheckDB that stops when ctx does: the integrity scan
// reads every page, and a caller that has been asked to stop (carlos db,
// on Ctrl-C) should not finish a full scan first.
func CheckDBContext(ctx context.Context, path string) error {
	db, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
	if err != nil {
		return err
	}
	defer db.Close()
	db.SetMaxOpenConns(1)
	// busy_timeout first, always (Global Constraints) — and on a garbage file
	// this first statement is what fails loudly instead of a later query.
	if _, err := db.ExecContext(ctx, "PRAGMA busy_timeout=5000"); err != nil {
		return fmt.Errorf("open: %w", err)
	}
	var res string
	if err := db.QueryRowContext(ctx, "PRAGMA integrity_check").Scan(&res); err != nil {
		return fmt.Errorf("integrity_check: %w", err)
	}
	if res != "ok" {
		return fmt.Errorf("integrity_check: %s", res)
	}
	var tables int
	if err := db.QueryRowContext(ctx, "SELECT count(*) FROM sqlite_master WHERE type='table'").Scan(&tables); err != nil {
		return fmt.Errorf("schema count: %w", err)
	}
	if tables == 0 {
		return errors.New("restored database has no tables")
	}
	return nil
}

// VerifyReplica restores s3://bucket/prefix into a throwaway file under dir
// and proves it opens clean, then deletes it. It returns ErrNoReplica when
// the prefix has never been replicated to, and a describing error when the
// restore or the check fails.
//
// IT DELIBERATELY DOES NOT TAKE THE LEASE, and that is safe in a way Restore
// itself is not. Restore's contract requires the lease because it materializes
// the LIVE database: an unleased restore there can resurrect a stale copy that
// later replicates over the true holder's newer state. Here the output path is
// a temp file that is never replicated from and is removed before return, so
// there is nothing that could shadow anything. What the lease forbids is a
// second WRITER; a reader costs the holder nothing — the box-wide
// `carlos ops restore-verify` has always restored databases that its own
// replicator was actively streaming, and litestream restores a committed
// position rather than a live tail.
func VerifyReplica(ctx context.Context, bucket, prefix, region, dir string) (uint32, error) {
	out := filepath.Join(dir, "verify-"+filepath.Base(prefix)+".db")
	// A retry must not trip over a previous attempt's file: litestream refuses
	// to restore over an existing output path.
	_ = os.Remove(out)
	// Sweep on EVERY path, including a failed check. The sidecars are swept
	// too: CheckDB's "mode=ro" does not stop the driver attaching a WAL, and a
	// drill that left a copy of every tenant's database on the box would be
	// both a disk bug and a data-exposure one.
	defer func() {
		_ = os.Remove(out)
		_ = os.Remove(out + "-wal")
		_ = os.Remove(out + "-shm")
	}()

	restored, err := restoreForVerify(ctx, out, bucket, prefix, region)
	if err != nil {
		return 0, fmt.Errorf("restore: %w", err)
	}
	if !restored {
		return 0, ErrNoReplica
	}
	if err := CheckDB(out); err != nil {
		return 0, err
	}
	// The counter is read BEFORE the deferred sweep removes the copy: it is
	// what lets the caller ask the second question, whether this replica is
	// still tracking its source (see ops.replicaFreshness).
	return ChangeCounter(out)
}

// ChangeCounter reads the SQLite file-change counter: bytes 24-27 of the
// database header, big-endian. SQLite increments it on every write transaction
// that modifies the main database file, which makes it the cheapest honest
// answer to "is this replica still tracking its source?" — compare the number
// on the live database with the number on a restored copy.
//
// It lives here, beside CheckDB and Restore, because both drills need it: the
// box-wide one (internal/ops) and the hibernation one, which restores through
// VerifyReplica.
//
// In WAL mode the counter in the main file only advances at a checkpoint, so a
// source with a hot WAL reads LOWER than reality. That skew is deliberately
// harmless: it can only make a stale replica look current (a missed alarm),
// never a current replica look stale (a false one).
func ChangeCounter(path string) (uint32, error) {
	f, err := os.Open(path)
	if err != nil {
		return 0, err
	}
	defer f.Close()

	var header [28]byte
	if _, err := io.ReadFull(f, header[:]); err != nil {
		return 0, fmt.Errorf("read header: %w", err)
	}
	if string(header[:15]) != "SQLite format 3" {
		return 0, fmt.Errorf("not a SQLite database")
	}
	return binary.BigEndian.Uint32(header[24:28]), nil
}

// HasReplica reports whether s3://bucket/prefix holds any LTX file at any
// level: the seed fallback's "has this host a replica of its own" (seeded
// instances spec §3.4 step 1). A listing error is returned, never read as
// "no replica".
func HasReplica(ctx context.Context, bucket, prefix, region string) (bool, error) {
	region, err := resolveRegion(ctx, region)
	if err != nil {
		return false, err
	}
	client, err := newReplicaClient(bucket, prefix, region)
	if err != nil {
		return false, err
	}
	return hasReplicaIn(ctx, client)
}

// ltxLister is the one replica-client method HasReplica needs.
type ltxLister interface {
	LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error)
}

func hasReplicaIn(ctx context.Context, client ltxLister) (bool, error) {
	for level := 0; level <= litestream.SnapshotLevel; level++ {
		itr, err := client.LTXFiles(ctx, level, 0, false)
		if err != nil {
			return false, err
		}
		found := itr.Next()
		ierr := itr.Err()
		cerr := itr.Close()
		if ierr != nil {
			return false, ierr
		}
		if cerr != nil {
			return false, cerr
		}
		if found {
			return true, nil
		}
	}
	return false, nil
}