package moves

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/benbjohnson/litestream"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/superfly/ltx"
)

// Errors a build refuses with. Both are internal: Task 10's landing decides
// what an operator reads.
var (
	// ErrBuildIncomplete: the pinned set does not restore to exactly N (a
	// gap between files, or a set that stops short). Nothing was written.
	ErrBuildIncomplete = errors.New("moves: the sealed replica does not restore to the required position")
	// ErrBuildDiverged: an object already under one of the generation's
	// keys is not what this build would write (another intent, other
	// snapshot or anchor bytes, another manifest). The generation is new
	// and the build is deterministic, so this is never a replay.
	ErrBuildDiverged = errors.New("moves: the prepared generation holds an object this build did not make")
	// ErrBuildUnpinned: a version id the build would record or read back
	// pins nothing ("" is no version; "null" is what a bucket whose
	// versioning is suspended answers, and the next write to that key
	// replaces it). Every artifact identity the manifest names, and every
	// current version a replay chooses, must be immutable, so the build
	// fails before reading or recording such an id and no install.json is
	// written (G27 on the output side).
	ErrBuildUnpinned = errors.New("moves: a prepared generation object has no usable version id")
)

// pinned is the check every recorded or chosen version id passes.
func pinned(key, versionID string) error {
	if !lease.UsableVersionID(versionID) {
		return fmt.Errorf("%w: %s as %q", ErrBuildUnpinned, key, versionID)
	}
	return nil
}

// buildContractVersion is the V of the intent and the manifest.
const buildContractVersion = 1

// BuildIntent is written before any other PUT with the encoding inputs,
// so a replay encodes identical bytes (§7.6 building, R7-4). Timestamp is
// the LTX header timestamp in milliseconds since the epoch.
type BuildIntent struct {
	V         int    `json:"v"`
	Op        string `json:"op"`
	Host      string `json:"host"`
	N         uint64 `json:"snapshot_txid"`
	PageSize  uint32 `json:"page_size"`
	Commit    uint32 `json:"commit"`
	Timestamp int64  `json:"timestamp"`
}

// GenerationPrefix is `<location>/g-<op>/`: the prepared generation a move
// (or a relocation) installs under a holder's location. The location is
// normalised with TrimSuffix(location, "/") + "/" through ReplicaSource,
// whose rule this shares with hostbox.GenerationPrefix and genselect's
// prefix; the four change together (G24, I73).
func GenerationPrefix(location, op string) string {
	prefix, _ := ReplicaSource(hostclaim.Holder{Location: location, Generation: op})
	return prefix
}

// FilesGenerationBase is `<location>/files/g-`, the start of every
// generation's files prefix in a location (GenerationFilesPrefix): what
// follows it up to the next "/" is the generation.
func FilesGenerationBase(location string) string {
	return strings.TrimSuffix(location, "/") + "/files/g-"
}

// GenerationFilesPrefix is `<location>/files/g-<generation>/` (spec §7.6):
// a generation's files manifest and the content objects it published. It
// is the one files prefix rule, GenerationPrefix's twin: the C8 files
// provider (host.movesFilesPrefix), the park view's files live set and
// classifier, and region HA's relocation (its pinner, build, install,
// release flush and probe) all name a generation's files through it.
func GenerationFilesPrefix(location, generation string) string {
	return FilesGenerationBase(location) + generation + "/"
}

// SnapshotKey is the generation's level-9 snapshot [1, n].
func SnapshotKey(prefix string, n uint64) string {
	return prefix + fmt.Sprintf("%04x/", litestream.SnapshotLevel) + ltx.FormatFilename(1, ltx.TXID(n))
}

// AnchorKey is the generation's level-0 anchor [n, n].
func AnchorKey(prefix string, n uint64) string {
	return prefix + "0000/" + ltx.FormatFilename(ltx.TXID(n), ltx.TXID(n))
}

// BuildIntentKey is where the intent lives in the generation.
func BuildIntentKey(prefix string) string { return prefix + "build-intent.json" }

// InstallManifestKey is where the completed install manifest lives.
func InstallManifestKey(prefix string) string { return prefix + "install.json" }

// BuildInputs are one generation build. Source is the pinned version set
// (a move: the seal's replica/ objects; region HA's relocation: the
// versions a genselect plan or a current listing pins, L16); the
// SealedClient is built over it, so nothing here names a staging lane.
// WorkDir is the build's scratch, `<JournalDir>/build/` (G34): the
// restored database is left there as `<WorkDir>/<Host>.db` for the
// landing's `restored` to rename into place on the same filesystem.
// Verify is the database check; nil means lease.CheckDB. Now is the clock
// for the intent's timestamp and the manifest's BuiltAt; nil means
// time.Now.
type BuildInputs struct {
	Store            lease.VersionedStore
	Bucket           string
	Source           []SealedObject
	N                uint64
	GenerationPrefix string
	Op, Host         string
	WorkDir          string
	Now              func() time.Time
	Verify           func(dbPath string) error
}

// buildName refuses a host or op that is empty or could leave WorkDir:
// the output path is built from the host and both name the generation.
func buildName(kind, s string) error {
	if s == "" || s == "." || s == ".." || strings.ContainsAny(s, "/\\\x00") {
		return fmt.Errorf("moves: build has an unusable %s %q", kind, s)
	}
	return nil
}

func (in BuildInputs) validate() error {
	if in.Store == nil || in.Bucket == "" {
		return errors.New("moves: build needs a store and a bucket")
	}
	if in.N == 0 {
		return errors.New("moves: build needs a position")
	}
	if in.GenerationPrefix == "" || !strings.HasSuffix(in.GenerationPrefix, "/") {
		return fmt.Errorf("moves: build needs a generation prefix ending in /, not %q", in.GenerationPrefix)
	}
	if err := buildName("op", in.Op); err != nil {
		return err
	}
	if err := buildName("host", in.Host); err != nil {
		return err
	}
	if in.WorkDir == "" {
		return errors.New("moves: build needs a work directory")
	}
	if len(in.Source) == 0 {
		return fmt.Errorf("%w: the pinned set is empty", ErrBuildIncomplete)
	}
	return nil
}

// BuildGeneration builds the prepared generation from the pinned set
// (§5.1 steps 2 and 7, §7.6 building): litestream's own planner and
// restore over a SealedClient into `<WorkDir>/<Host>.db`, refused unless
// the plan ends exactly at N and every planned file verifies; the database
// check; then, in order, the intent, the snapshot [1, N], the anchor
// [N, N] and the completed InstallManifest, each create-only. A replay at
// any boundary restores again from the same versions (the scratch file is
// removed first: litestream refuses an existing output and the file is
// reproducible), reads the intent it finds and so encodes identical bytes,
// and reuses an object it finds under its key only when the bytes are
// equal. The build never deletes an object.
func BuildGeneration(ctx context.Context, in BuildInputs) (InstallManifest, error) {
	if err := in.validate(); err != nil {
		return InstallManifest{}, err
	}
	now := in.Now
	if now == nil {
		now = time.Now
	}
	verify := in.Verify
	if verify == nil {
		verify = lease.CheckDB
	}
	logger := slog.New(slog.NewTextHandler(io.Discard, nil))

	client, err := NewSealedClient(in.Store, in.Bucket, in.Source)
	if err != nil {
		return InstallManifest{}, err
	}
	target := ltx.TXID(in.N)

	// The plan: litestream's own, rooted at the newest snapshot at or
	// below N and extended file by file. A gap under a TXID target is
	// ErrTxNotAvailable (the contiguity error is only raised for a
	// latest-state plan), as is a set that stops short.
	plan, err := litestream.CalcRestorePlan(ctx, client, target, time.Time{}, logger)
	if err != nil {
		if errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots) {
			return InstallManifest{}, fmt.Errorf("%w: planning to %s: %v", ErrBuildIncomplete, target, err)
		}
		return InstallManifest{}, fmt.Errorf("moves: plan the restore: %w", err)
	}
	if len(plan) == 0 || plan[len(plan)-1].MaxTXID != target {
		return InstallManifest{}, fmt.Errorf("%w: the plan ends at %s, not %s", ErrBuildIncomplete, plan[len(plan)-1].MaxTXID, target)
	}
	for _, info := range plan {
		if err := verifySealedFile(ctx, client, info); err != nil {
			return InstallManifest{}, err
		}
	}

	// A fresh scratch file. Litestream refuses an existing output path
	// and writes through `<path>.tmp`; a crash can leave either, plus the
	// sidecars the database check attaches.
	if err := os.MkdirAll(in.WorkDir, 0o700); err != nil {
		return InstallManifest{}, fmt.Errorf("moves: build scratch: %w", err)
	}
	dbPath := filepath.Join(in.WorkDir, in.Host+".db")
	for _, p := range []string{dbPath, dbPath + ".tmp", dbPath + "-wal", dbPath + "-shm", dbPath + "-txid"} {
		if err := os.Remove(p); err != nil && !errors.Is(err, os.ErrNotExist) {
			return InstallManifest{}, fmt.Errorf("moves: clear build scratch: %w", err)
		}
	}
	opt := litestream.NewRestoreOptions()
	opt.OutputPath = dbPath
	opt.TXID = target
	if err := litestream.NewReplicaWithClient(nil, client).Restore(ctx, opt); err != nil {
		if errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots) {
			return InstallManifest{}, fmt.Errorf("%w: restoring to %s: %v", ErrBuildIncomplete, target, err)
		}
		return InstallManifest{}, fmt.Errorf("moves: restore to %s: %w", target, err)
	}

	pageSize, pageCount, err := ReadDBHeader(dbPath)
	if err != nil {
		return InstallManifest{}, fmt.Errorf("moves: restored database: %w", err)
	}
	st, err := os.Stat(dbPath)
	if err != nil {
		return InstallManifest{}, err
	}
	if st.Size() != int64(pageSize)*int64(pageCount) {
		return InstallManifest{}, fmt.Errorf("moves: restored database is %d bytes, its header says %d pages of %d", st.Size(), pageCount, pageSize)
	}
	if err := verify(dbPath); err != nil {
		return InstallManifest{}, fmt.Errorf("moves: restored database check: %w", err)
	}
	// The check may attach a WAL and a shared-memory file; neither is part
	// of the build's result.
	for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} {
		_ = os.Remove(p)
	}

	// The intent, before any other PUT. On a replay the existing intent's
	// timestamp is the one that gives identical bytes.
	intent := BuildIntent{V: buildContractVersion, Op: in.Op, Host: in.Host, N: in.N, PageSize: pageSize, Commit: pageCount, Timestamp: now().UnixMilli()}
	intent, err = putOrReadIntent(ctx, in, intent)
	if err != nil {
		return InstallManifest{}, err
	}

	f, err := os.Open(dbPath)
	if err != nil {
		return InstallManifest{}, err
	}
	defer f.Close()

	snapshot, snapshotCRC, err := EncodeFull(f, pageSize, pageCount, 1, target, intent.Timestamp, 0)
	if err != nil {
		return InstallManifest{}, err
	}
	snapshotKey := SnapshotKey(in.GenerationPrefix, in.N)
	snapshotVersion, err := putOrReuse(ctx, in, snapshotKey, snapshot)
	if err != nil {
		return InstallManifest{}, err
	}
	snapshot = nil

	anchor, anchorCRC, err := EncodeFull(f, pageSize, pageCount, target, target, intent.Timestamp, litestream.WALHeaderSize)
	if err != nil {
		return InstallManifest{}, err
	}
	anchorKey := AnchorKey(in.GenerationPrefix, in.N)
	anchorVersion, err := putOrReuse(ctx, in, anchorKey, anchor)
	if err != nil {
		return InstallManifest{}, err
	}
	anchor = nil

	h := sha256.New()
	if _, err := io.Copy(h, io.NewSectionReader(f, 0, st.Size())); err != nil {
		return InstallManifest{}, fmt.Errorf("moves: hash the restored database: %w", err)
	}

	m := InstallManifest{
		V: buildContractVersion, Op: in.Op, Host: in.Host, N: in.N,
		SnapshotKey: snapshotKey, SnapshotVersion: snapshotVersion, SnapshotCRC64: snapshotCRC,
		AnchorKey: anchorKey, AnchorVersion: anchorVersion, AnchorCRC64: anchorCRC,
		IntegrityCheck: "ok",
		DBSHA256:       hex.EncodeToString(h.Sum(nil)),
		BuiltAt:        now().UTC(),
	}
	return putOrReadManifest(ctx, in, m)
}

// verifySealedFile reads one planned file whole through the sealed client
// and checks its CRC-64 trailer and that its header is the position the
// seal named it by.
func verifySealedFile(ctx context.Context, client *SealedClient, info *ltx.FileInfo) error {
	rc, err := client.OpenLTXFile(ctx, info.Level, info.MinTXID, info.MaxTXID, 0, 0)
	if err != nil {
		return fmt.Errorf("moves: open sealed %s: %w", ltx.FormatFilename(info.MinTXID, info.MaxTXID), err)
	}
	defer rc.Close()
	dec := ltx.NewDecoder(rc)
	if err := dec.Verify(); err != nil {
		return fmt.Errorf("moves: sealed %s at level %d does not verify: %w", ltx.FormatFilename(info.MinTXID, info.MaxTXID), info.Level, err)
	}
	if h := dec.Header(); h.MinTXID != info.MinTXID || h.MaxTXID != info.MaxTXID {
		return fmt.Errorf("moves: sealed %s at level %d carries [%s,%s]", ltx.FormatFilename(info.MinTXID, info.MaxTXID), info.Level, h.MinTXID, h.MaxTXID)
	}
	return nil
}

// currentVersion is the object version currently under key, found by an
// exact-key listing (a store names versions, never a bare key). Its id
// must pin an immutable object before anything is read by it.
func currentVersion(ctx context.Context, in BuildInputs, key string) (lease.ObjectVersion, error) {
	vs, err := in.Store.ListVersions(ctx, in.Bucket, key)
	if err != nil {
		return lease.ObjectVersion{}, fmt.Errorf("moves: list %s: %w", key, err)
	}
	for _, v := range vs {
		if v.Key == key && v.IsLatest && !v.DeleteMarker {
			if err := pinned(key, v.VersionID); err != nil {
				return lease.ObjectVersion{}, err
			}
			return v, nil
		}
	}
	return lease.ObjectVersion{}, fmt.Errorf("moves: %s exists but has no current version", key)
}

// putOrReuse writes body create-only under key and returns the version id.
// When the key already holds an object, the current version's bytes must
// equal body (identical by construction from the intent) and its id is the
// answer; anything else is ErrBuildDiverged.
func putOrReuse(ctx context.Context, in BuildInputs, key string, body []byte) (string, error) {
	v, err := in.Store.PutIfAbsent(ctx, in.Bucket, key, body)
	if err == nil {
		if err := pinned(key, v); err != nil {
			return "", err
		}
		return v, nil
	}
	if !errors.Is(err, lease.ErrExists) {
		return "", fmt.Errorf("moves: put %s: %w", key, err)
	}
	cur, err := currentVersion(ctx, in, key)
	if err != nil {
		return "", err
	}
	have, err := in.Store.GetVersion(ctx, in.Bucket, key, cur.VersionID)
	if err != nil {
		return "", fmt.Errorf("moves: read %s@%s: %w", key, cur.VersionID, err)
	}
	if !bytes.Equal(have, body) {
		return "", fmt.Errorf("%w: %s@%s", ErrBuildDiverged, key, cur.VersionID)
	}
	return cur.VersionID, nil
}

// putOrReadIntent writes the intent create-only, or reads the one already
// there: it must name the same op, host, position, page size and commit,
// and its timestamp is the one this build encodes with.
func putOrReadIntent(ctx context.Context, in BuildInputs, want BuildIntent) (BuildIntent, error) {
	key := BuildIntentKey(in.GenerationPrefix)
	body, err := json.Marshal(want)
	if err != nil {
		return BuildIntent{}, err
	}
	if v, err := in.Store.PutIfAbsent(ctx, in.Bucket, key, body); err == nil {
		if err := pinned(key, v); err != nil {
			return BuildIntent{}, err
		}
		return want, nil
	} else if !errors.Is(err, lease.ErrExists) {
		return BuildIntent{}, fmt.Errorf("moves: put %s: %w", key, err)
	}
	cur, err := currentVersion(ctx, in, key)
	if err != nil {
		return BuildIntent{}, err
	}
	have, err := in.Store.GetVersion(ctx, in.Bucket, key, cur.VersionID)
	if err != nil {
		return BuildIntent{}, fmt.Errorf("moves: read %s@%s: %w", key, cur.VersionID, err)
	}
	var got BuildIntent
	if err := json.Unmarshal(have, &got); err != nil {
		return BuildIntent{}, fmt.Errorf("%w: %s@%s is not an intent: %v", ErrBuildDiverged, key, cur.VersionID, err)
	}
	if got.V != want.V || got.Op != want.Op || got.Host != want.Host || got.N != want.N || got.PageSize != want.PageSize || got.Commit != want.Commit {
		return BuildIntent{}, fmt.Errorf("%w: %s@%s names other inputs", ErrBuildDiverged, key, cur.VersionID)
	}
	return got, nil
}

// putOrReadManifest writes the completed manifest create-only, or reads
// the one already there, which must name the same position, keys,
// versions, checksums and database digest; the stored one (its BuiltAt
// included) is the answer.
func putOrReadManifest(ctx context.Context, in BuildInputs, want InstallManifest) (InstallManifest, error) {
	key := InstallManifestKey(in.GenerationPrefix)
	body, err := json.Marshal(want)
	if err != nil {
		return InstallManifest{}, err
	}
	if v, err := in.Store.PutIfAbsent(ctx, in.Bucket, key, body); err == nil {
		if err := pinned(key, v); err != nil {
			return InstallManifest{}, err
		}
		return want, nil
	} else if !errors.Is(err, lease.ErrExists) {
		return InstallManifest{}, fmt.Errorf("moves: put %s: %w", key, err)
	}
	cur, err := currentVersion(ctx, in, key)
	if err != nil {
		return InstallManifest{}, err
	}
	have, err := in.Store.GetVersion(ctx, in.Bucket, key, cur.VersionID)
	if err != nil {
		return InstallManifest{}, fmt.Errorf("moves: read %s@%s: %w", key, cur.VersionID, err)
	}
	var got InstallManifest
	if err := json.Unmarshal(have, &got); err != nil {
		return InstallManifest{}, fmt.Errorf("%w: %s@%s is not a manifest: %v", ErrBuildDiverged, key, cur.VersionID, err)
	}
	same := got.V == want.V && got.Op == want.Op && got.Host == want.Host && got.N == want.N &&
		got.SnapshotKey == want.SnapshotKey && got.SnapshotVersion == want.SnapshotVersion && got.SnapshotCRC64 == want.SnapshotCRC64 &&
		got.AnchorKey == want.AnchorKey && got.AnchorVersion == want.AnchorVersion && got.AnchorCRC64 == want.AnchorCRC64 &&
		got.IntegrityCheck == want.IntegrityCheck && got.DBSHA256 == want.DBSHA256
	if !same {
		return InstallManifest{}, fmt.Errorf("%w: %s@%s records another build", ErrBuildDiverged, key, cur.VersionID)
	}
	return got, nil
}