package activator

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"strings"

	"github.com/carlosframework/platform/internal/instancefs"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/seeds"
)

// SeedObjects reads a seed's snapshot, seeds/<host>/<id>/snapshot.db, from
// this box's own region data bucket.
type SeedObjects interface {
	Open(ctx context.Context, host, id string) (io.ReadCloser, error)
}

// SeedMarkerSuffix names the installation evidence beside a placed
// database: <db>.seed holds the seed's id (spec §3.4 step 3, codex R2-1).
const SeedMarkerSuffix = ".seed"

// seedMarkerMax bounds the marker read: it holds a 29-byte id and a newline,
// and it sits in tenant-writable ground.
const seedMarkerMax = 128

// hasOwnReplica and checkSeedDB are seams for the tests.
var (
	hasOwnReplica = lease.HasReplica
	checkSeedDB   = lease.CheckDBContext
)

// SeedPlacementError is a wake the seed fallback failed or refused. Its
// Sentence is the member's (SD-N14 to SD-N19): memberSafeWakeReason passes
// it to carlos logs whole. Err is the cause, for the box log.
type SeedPlacementError struct {
	Sentence string
	Err      error
}

func (e *SeedPlacementError) Error() string { return e.Sentence }
func (e *SeedPlacementError) Unwrap() error { return e.Err }

func seedMismatch(host string) error {
	return &SeedPlacementError{Sentence: fmt.Sprintf("%s could not start from its seed: the snapshot's length or sha256 does not match its record.", host)}
}

func seedNotDB(host string, err error) error {
	return &SeedPlacementError{Sentence: fmt.Sprintf("%s could not start from its seed: the snapshot is not a database that passes an integrity check.", host), Err: err}
}

func seedUnread(host string, err error) error {
	return &SeedPlacementError{Sentence: fmt.Sprintf("%s could not start from its seed: the snapshot could not be read. The next wake tries again.", host), Err: err}
}

func seedRefusedFlat(host string) error {
	return &SeedPlacementError{Sentence: fmt.Sprintf("%s could not start from its seed: this box does not run instances nested.", host)}
}

func seedRefusedClaim(host string) error {
	return &SeedPlacementError{Sentence: fmt.Sprintf("%s could not start from its seed: claim control is on for this box.", host)}
}

func seedPlaceFailed(host string, err error) error {
	return &SeedPlacementError{Sentence: fmt.Sprintf("%s could not start from its seed: it could not be put in place. The next wake tries again.", host), Err: err}
}

// ownsReplica reports a seed whose host already has a replica of its own:
// placed is set only after the first upload, and committed only from
// placed (spec §5.4). Such a row restores its own lineage and never boots
// fresh: after commit the app retires its old process, so an empty boot
// would read as the import's data lost (preflight M11).
func ownsReplica(src *seeds.RowSeed) bool {
	return src.State == seeds.StatePlaced || src.State == seeds.StateCommitted
}

// seedRestores reports a row Restore sends through seededRestore: a seed
// still to place, or one whose own replica exists.
func seedRestores(src *seeds.RowSeed) bool {
	return src != nil && (src.Placeable() || ownsReplica(src))
}

// seededRestore is the plain-path seed fallback (spec §3.4). It never
// reports fresh: a seeded row never runs the manifest's seed command, and
// every failure fails the wake instead of falling through to an empty
// database. Every file it touches in the instance directory goes through
// instancefs, so no symlink there is followed.
func (b *ExecBackend) seededRestore(ctx context.Context, rt registry.Route, src *seeds.RowSeed) (bool, error) {
	host := rt.Host
	fail := func(err error) (bool, error) {
		var spe *SeedPlacementError
		if errors.As(err, &spe) && spe.Err != nil {
			log.Printf("activator: %s: seed %s: %v: %v", host, src.ID, err, spe.Err)
		} else {
			log.Printf("activator: %s: seed %s: %v", host, src.ID, err)
		}
		return false, err
	}
	if err := seeds.Check(host, src.ID); err != nil {
		return fail(fmt.Errorf("seed fallback for %s: %w", host, err))
	}
	// restoreOwn restores the host's own lineage, the staging fallback
	// included, and never reports fresh: a host that has (or was listed as
	// having) a replica of its own and finds none must not boot empty. The
	// next wake asks again.
	//
	// The database it leaves is this seed's lineage, so its <db>.seed
	// marker is written to name the seed (final review carry item): the
	// rollback's missing-row purge and an aborted seed's local purge decide
	// by the marker, and a database with none is left in place.
	restoreOwn := func() (bool, error) {
		fresh, err := b.restorePlain(ctx, rt)
		if err != nil {
			return false, err
		}
		if fresh {
			return fail(fmt.Errorf("seed fallback for %s: found no replica of its own to restore", host))
		}
		db := b.pathsFor(rt).dbPath
		if err := instancefs.WriteAtomic(filepath.Dir(db), filepath.Base(db)+SeedMarkerSuffix, []byte(src.ID+"\n"), 0o600); err != nil {
			return fail(seedPlaceFailed(host, err))
		}
		return false, nil
	}
	// A placed seed stays on the row for life, and placed is set only once
	// the host's own replica exists. So a placed row's wake is a plain
	// restore on any box: no layout or claim refusal, no listing on every
	// wake, and the staging fallback still promotes a moved host's replica
	// (the snapshot was deleted at placement; it is never read again). A
	// committed import was placed first, so it is restored the same way.
	if ownsReplica(src) {
		return restoreOwn()
	}
	if b.ClaimArmed {
		return fail(seedRefusedClaim(host))
	}
	paths := b.pathsFor(rt)
	if paths.layout != LayoutNested {
		return fail(seedRefusedFlat(host))
	}
	dir, base := filepath.Dir(paths.dbPath), filepath.Base(paths.dbPath)
	// The instance directory may not exist yet on a first wake. Created as
	// prepare and lease.Restore create it.
	if err := os.MkdirAll(dir, 0o700); err != nil {
		return fail(seedPlaceFailed(host, err))
	}

	// Step 1: a replica of its own means this host was placed already. A
	// listing that cannot tell fails the wake: read as "no replica", it
	// would place the snapshot over a lineage that exists.
	has, err := hasOwnReplica(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region)
	if err != nil {
		return fail(fmt.Errorf("seed fallback for %s: cannot tell whether it has a replica of its own: %w", host, err))
	}
	if has {
		if _, err := restoreOwn(); err != nil {
			return false, err
		}
		b.finishPlacement(ctx, rt, src)
		return false, nil
	}

	// An earlier first-boot seed command at this path whose processes were
	// not proven gone left an obligation beside the database
	// (settleSeedObligation, as restorePlain answers it): it is answered
	// before the file there is kept as this seed's, purged or replaced, so
	// nothing is placed under a process that may still write the path. A
	// seeded row runs no seed command itself, so this is another row's
	// leftover. Unproven, the wake is refused and the lease left to its
	// TTL, as on the plain path.
	if err := b.settleSeedObligation(ctx, rt, paths.dbPath); err != nil {
		return fail(err)
	}

	// This seed, placed by an interrupted first wake: keep it. Replication
	// starts in doWake, as generation one of the host's own lineage.
	placed, err := seedPlacedAt(dir, base, src.ID)
	if err != nil {
		return fail(seedPlaceFailed(host, err))
	}
	if placed {
		log.Printf("activator: %s: keeping seed %s placed by an earlier wake", host, src.ID)
		return false, nil
	}
	if b.Seeds == nil {
		return fail(seedUnread(host, errors.New("no seed reader on this box")))
	}
	// Anything else is purged before the fetch: another seed's database,
	// one with no marker, their WAL and SHM, the marker, litestream's
	// metadata, and an earlier fetch's leftovers.
	if err := purgeSeedTarget(dir, base); err != nil {
		return fail(seedPlaceFailed(host, err))
	}
	if err := b.placeSeed(ctx, host, src, dir, base); err != nil {
		return fail(err)
	}
	log.Printf("activator: %s: placed seed %s (%d bytes)", host, src.ID, src.Bytes)
	return false, nil
}

// reconcileSeedMarker runs first on the plain restore path, whether or not
// a database is at dbPath (final review F2b, re-review finding 8). A
// <db>.seed marker beside the path that names a seed other than the row's
// own current seed (the row has none, or another) is a leftover of that
// seed: an aborted --from seed whose purge ran on no box, a rolled-back
// import, or a placement that crashed between its marker and its rename
// and left the marker alone. Whatever is there goes, through
// purgeSeedTarget (the database and its sidecars, the marker, any
// .seed-sd_*.tmp), before anything is restored or created, so the file is
// never served as this row's warm cache and no new database is ever born
// beside a foreign marker that a later wake would read as the old seed's.
// A marker naming the row's own seed, or none, changes nothing. A marker
// that cannot be read fails the wake, with nothing served or purged.
//
// A relative path is no seeded path (seeded rows are always given the
// nested, absolute one), so it is left to the restore as before.
func reconcileSeedMarker(rt registry.Route, dbPath string) error {
	if !filepath.IsAbs(dbPath) {
		return nil
	}
	dbPath = filepath.Clean(dbPath)
	marker, err := ReadSeedMarker(dbPath)
	if err != nil {
		return fmt.Errorf("restore %s: the seed marker beside %s cannot be read: %w", rt.Host, dbPath, err)
	}
	own := rowSeedID(rt.SeedSource)
	if marker == "" || marker == own {
		return nil
	}
	dir, base, err := splitDBPath(dbPath)
	if err != nil {
		return err
	}
	if err := purgeSeedTarget(dir, base); err != nil {
		return fmt.Errorf("restore %s: purge what seed %s left at %s: %w", rt.Host, marker, dbPath, err)
	}
	log.Printf("activator: %s: %s held seed %s's marker, not this route's (seed %q): purged before the restore", rt.Host, dbPath, marker, own)
	return nil
}

// seedPlacedAt reports whether dir/base is a regular file whose marker names
// id. A missing, symlinked or oversized marker or database is "not placed",
// and is purged; any other error is returned.
func seedPlacedAt(dir, base, id string) (bool, error) {
	m, err := instancefs.ReadBeneathLimit(dir, base+SeedMarkerSuffix, seedMarkerMax)
	if err != nil {
		if notAPlacedFile(err) {
			return false, nil
		}
		return false, err
	}
	if strings.TrimSpace(string(m)) != id {
		return false, nil
	}
	f, err := instancefs.OpenBeneath(dir, base, os.O_RDONLY, 0)
	if err != nil {
		if notAPlacedFile(err) {
			return false, nil
		}
		return false, err
	}
	f.Close()
	return true, nil
}

func notAPlacedFile(err error) bool {
	return errors.Is(err, os.ErrNotExist) || errors.Is(err, instancefs.ErrSymlink) ||
		errors.Is(err, instancefs.ErrNotRegular) || errors.Is(err, instancefs.ErrTooLarge)
}

// seedTempName is where a seed's snapshot is fetched to, in the instance
// directory. The id is a seed id (seeds.Check), one clean path element.
func seedTempName(id string) string { return ".seed-" + id + ".tmp" }

// PurgeSeedLocal removes an instance's local database at dbPath and
// everything beside it that could stand in for one (purgeSeedTarget): the
// database, -wal, -shm, the <db>.seed marker, litestream's metadata and
// any seed fetch leftovers, through instancefs. It is the import
// rollback's local removal (seeded-instances spec §5.6 step 3), run after
// the host is fenced and stopped. A directory that is not there holds
// nothing, so a second run, or one for an instance never woken, is done.
func PurgeSeedLocal(dbPath string) error {
	dir, base, err := splitDBPath(dbPath)
	if err != nil {
		return err
	}
	if _, err := os.Lstat(dir); errors.Is(err, os.ErrNotExist) {
		return nil
	}
	return purgeSeedTarget(dir, base)
}

// ReadSeedMarker is the seed id the <db>.seed marker beside dbPath names,
// read through instancefs: the evidence that the database there was placed
// for that seed (spec §3.4 step 3). No marker, or one that is a symlink,
// not a regular file or oversized, is "" (it names no seed); so is a
// directory that is not there. Any other failure to read it is an error:
// the caller cannot tell.
func ReadSeedMarker(dbPath string) (string, error) {
	dir, base, err := splitDBPath(dbPath)
	if err != nil {
		return "", err
	}
	if _, err := os.Lstat(dir); errors.Is(err, os.ErrNotExist) {
		return "", nil
	}
	m, err := instancefs.ReadBeneathLimit(dir, base+SeedMarkerSuffix, seedMarkerMax)
	if err != nil {
		if notAPlacedFile(err) {
			return "", nil
		}
		return "", err
	}
	return strings.TrimSpace(string(m)), nil
}

// splitDBPath is a clean absolute database path's directory and file name.
func splitDBPath(dbPath string) (dir, base string, err error) {
	if !filepath.IsAbs(dbPath) || filepath.Clean(dbPath) != dbPath {
		return "", "", fmt.Errorf("activator: %q is not a clean absolute database path", dbPath)
	}
	dir, base = filepath.Split(dbPath)
	if base == "" {
		return "", "", fmt.Errorf("activator: %q names no database", dbPath)
	}
	return filepath.Clean(dir), base, nil
}

// purgeSeedTarget removes everything at and beside dir/base that could
// stand in for a seed's database: what PurgeLocalDBArtifacts removes in its
// full form, the marker, and any seed temp file, through instancefs.
func purgeSeedTarget(dir, base string) error {
	names := []string{base, base + "-wal", base + "-shm", base + SeedMarkerSuffix,
		base + "-litestream", ".activator-" + base + "-litestream"}
	ents, err := os.ReadDir(dir)
	if err != nil {
		return err
	}
	for _, e := range ents {
		if n := e.Name(); strings.HasPrefix(n, ".seed-sd_") && strings.Contains(n, ".tmp") {
			names = append(names, n)
		}
	}
	for _, n := range names {
		if err := instancefs.RemoveAllBeneath(dir, n); err != nil {
			return err
		}
	}
	return nil
}

// placeSeed fetches and checks the snapshot, then writes the marker and
// renames the snapshot onto dir/base. The marker goes first, so a database
// at that path always has one. The temp file and any sidecar the check left
// are removed on every path.
//
// The instance directory is the tenant's uid's to write, so the file is
// pinned by identity: the one fetchSeed wrote is reopened beneath dir and
// must be the same file, the integrity check reads it through that
// descriptor, it must still be the entry at tmp after the check, and the
// entry renamed onto dir/base must still be it. Anything else is refused,
// and a refusal after the rename removes the marker and the database, so no
// unchecked file is ever kept as this seed.
func (b *ExecBackend) placeSeed(ctx context.Context, host string, src *seeds.RowSeed, dir, base string) error {
	tmp := seedTempName(src.ID)
	defer func() {
		for _, n := range []string{tmp, tmp + "-wal", tmp + "-shm"} {
			_ = instancefs.RemoveAllBeneath(dir, n)
		}
	}()
	wrote, err := b.fetchSeed(ctx, host, src, dir, tmp)
	if err != nil {
		return err
	}
	f, err := openSameFile(dir, tmp, wrote)
	if err != nil {
		return seedPlaceFailed(host, err)
	}
	cerr := checkSeedDB(ctx, "/proc/self/fd/"+strconv.FormatUint(uint64(f.Fd()), 10))
	f.Close()
	if cerr != nil {
		return seedNotDB(host, cerr)
	}
	if err := sameFileAt(dir, tmp, wrote); err != nil {
		return seedPlaceFailed(host, err)
	}
	if err := instancefs.WriteAtomic(dir, base+SeedMarkerSuffix, []byte(src.ID+"\n"), 0o600); err != nil {
		return seedPlaceFailed(host, err)
	}
	if err := instancefs.RenameBetween(dir, tmp, dir, base); err != nil {
		return seedPlaceFailed(host, err)
	}
	if err := sameFileAt(dir, base, wrote); err != nil {
		_ = instancefs.RemoveAllBeneath(dir, base+SeedMarkerSuffix)
		_ = instancefs.RemoveAllBeneath(dir, base)
		return seedPlaceFailed(host, err)
	}
	return nil
}

// errSwapped is a seed file that is no longer the one fetchSeed wrote.
var errSwapped = errors.New("the fetched snapshot was replaced in the instance directory")

// openSameFile opens dir/name beneath dir, read-only, and refuses it unless
// it is the file want describes.
func openSameFile(dir, name string, want os.FileInfo) (*os.File, error) {
	f, err := instancefs.OpenBeneath(dir, name, os.O_RDONLY, 0)
	if err != nil {
		return nil, err
	}
	fi, err := f.Stat()
	if err != nil {
		f.Close()
		return nil, err
	}
	if !os.SameFile(fi, want) {
		f.Close()
		return nil, errSwapped
	}
	return f, nil
}

// sameFileAt reports errSwapped unless dir/name is the file want describes.
func sameFileAt(dir, name string, want os.FileInfo) error {
	f, err := openSameFile(dir, name, want)
	if err != nil {
		return err
	}
	return f.Close()
}

// fetchSeed copies the snapshot into dir/tmp, checking its length and
// sha256 against the row as it writes, and returns the identity of the file
// it wrote. It reads at most one byte more than the row records, so an
// oversized object is refused without being read.
func (b *ExecBackend) fetchSeed(ctx context.Context, host string, src *seeds.RowSeed, dir, tmp string) (os.FileInfo, error) {
	rc, err := b.Seeds.Open(ctx, host, src.ID)
	if err != nil {
		return nil, seedUnread(host, err)
	}
	defer rc.Close()
	f, err := instancefs.OpenBeneath(dir, tmp, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
	if err != nil {
		return nil, seedPlaceFailed(host, err)
	}
	h := sha256.New()
	r := &readErrs{r: io.LimitReader(rc, src.Bytes+1)}
	n, cerr := io.Copy(io.MultiWriter(f, h), r)
	serr := f.Sync()
	fi, iderr := f.Stat()
	ferr := f.Close()
	if r.err != nil {
		return nil, seedUnread(host, r.err)
	}
	if cerr != nil || serr != nil || iderr != nil || ferr != nil {
		return nil, seedPlaceFailed(host, errors.Join(cerr, serr, iderr, ferr))
	}
	if n != src.Bytes || hex.EncodeToString(h.Sum(nil)) != src.SHA256 {
		return nil, seedMismatch(host)
	}
	return fi, nil
}

// readErrs keeps the read side's error apart from the write side's, which
// io.Copy returns alike: a bucket that fails is SD-N16, a disk that fails is
// SD-N19.
type readErrs struct {
	r   io.Reader
	err error
}

func (e *readErrs) Read(p []byte) (int, error) {
	n, err := e.r.Read(p)
	if err != nil && err != io.EOF {
		e.err = err
	}
	return n, err
}

// SeedPlaced records a seed placed once the host's own replica holds it
// (spec §3.4 steps 4 and 5): it waits for the first upload, then
// OnSeedPlaced CASes the record to placed and deletes seeds/<host>/<id>/.
// The snapshot goes only after the replica exists, so a crash at any point
// retries from the placed local file or the snapshot. A replicator that
// cannot be asked for its first upload records nothing.
func (b *ExecBackend) SeedPlaced(ctx context.Context, rt registry.Route, rep Rep) error {
	src, err := seeds.ParseRowSeed(rt.SeedSource)
	if err != nil || src == nil || !src.Placeable() || src.State == seeds.StatePlaced {
		return err
	}
	s, ok := rep.(interface{ SyncAndWait(context.Context) error })
	if !ok {
		return fmt.Errorf("first upload of %s: the replicator cannot confirm it", rt.Host)
	}
	if err := s.SyncAndWait(ctx); err != nil {
		return fmt.Errorf("first upload of %s: %w", rt.Host, err)
	}
	if b.OnSeedPlaced == nil {
		return errors.New("no seed recorder on this box")
	}
	return b.OnSeedPlaced(ctx, rt.Account, rt.App, rt.Host, src.ID)
}

// finishPlacement is SeedPlaced for a host whose own replica already
// exists (a placed CAS lost on an earlier wake). Best effort: a failure is
// logged, never the wake's, and the next wake asks again.
func (b *ExecBackend) finishPlacement(ctx context.Context, rt registry.Route, src *seeds.RowSeed) {
	if b.OnSeedPlaced == nil {
		return
	}
	if err := b.OnSeedPlaced(ctx, rt.Account, rt.App, rt.Host, src.ID); err != nil {
		log.Printf("activator: %s: seed %s not recorded as placed yet: %v", rt.Host, src.ID, err)
	}
}