package hostbox

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"path/filepath"
	"sort"
	"strings"

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

// RetireArtifacts moves a predecessor installation out of the live namespace.
// It is called only when the durable local generation differs from the selected
// generation, and while Work still vouches for the installation being replaced.
type RetireArtifacts func(context.Context, string) error

// ColdRestore installs a selected immutable generation. It never accepts an
// unselected current object and never reuses a local database.
type ColdRestore struct {
	Reader          genselect.Reader
	Bucket          string
	Generation      string // newly admitted installation, distinct from baseline plan.Generation
	Work            hostwrite.Work
	Identity        hostwrite.Identity
	RetireArtifacts RetireArtifacts
	// BeforeLegacyInstall parks an unmarked source outside Work.With. The
	// caller must bind it for serving, including a genuinely empty creator.
	BeforeLegacyInstall func(context.Context, string, string) error
	// BeforeInstallLocked validates enrollment after the marker recheck. An
	// error prevents purge, temporary database creation and marker writes.
	BeforeInstallLocked func(context.Context, string, string) error
}

// Install purges the old local state and restores plan into dbPath. All local
// mutation stays inside the work selected after admission.
func (r ColdRestore) Install(ctx context.Context, dbPath string, plan *genselect.Plan) error {
	if ctx == nil || r.Reader == nil || r.Bucket == "" || r.Work == nil || plan == nil ||
		plan.Generation == "" || !safeGeneration(r.Generation) || len(plan.Files) == 0 || plan.MaxTXID == 0 || dbPath == "" {
		return ErrUnsupported
	}
	var capturedGeneration string
	if err := r.Work.With(ctx, r.Identity, func() error {
		var err error
		capturedGeneration, err = readGeneration(dbPath + ".gen")
		return err
	}); err != nil {
		return fmt.Errorf("capture local generation: %w", err)
	}
	files, finalChecksum, err := loadPinned(ctx, r.Reader, r.Bucket, plan)
	if err != nil {
		return err
	}
	// Retirement takes park.Local's own flock. Calling it inside Work.With
	// would recursively acquire the same lock and deadlock.
	if capturedGeneration != "" && capturedGeneration != r.Generation {
		if r.RetireArtifacts == nil {
			return fmt.Errorf("retire generation %q: %w", capturedGeneration, ErrUnsupported)
		}
		if err := r.RetireArtifacts(ctx, capturedGeneration); err != nil {
			return fmt.Errorf("retire generation %q: %w", capturedGeneration, err)
		}
	}
	if capturedGeneration == "" && r.BeforeLegacyInstall != nil {
		if err := r.BeforeLegacyInstall(ctx, dbPath, r.Generation); err != nil {
			return err
		}
	}
	return r.Work.With(ctx, r.Identity, func() error {
		currentGeneration, err := readGeneration(dbPath + ".gen")
		if err != nil {
			return fmt.Errorf("recheck local generation: %w", err)
		}
		if capturedGeneration != "" && capturedGeneration != r.Generation {
			// Successful predecessor retirement normally moves the enrolled
			// marker with the rest of the installation. Absence is therefore the
			// expected receipt disposition. A different marker means a successor
			// appeared after our capture and must never be purged.
			if currentGeneration != "" {
				return ErrChanged
			}
		} else if currentGeneration != capturedGeneration {
			return ErrChanged
		}
		if r.BeforeInstallLocked != nil {
			if err := r.BeforeInstallLocked(ctx, dbPath, r.Generation); err != nil {
				return err
			}
		}
		if err := purgeDB(dbPath); err != nil {
			return err
		}
		tmp, err := os.CreateTemp(filepath.Dir(dbPath), ".restore-*.db")
		if err != nil {
			return fmt.Errorf("create restore temp: %w", err)
		}
		tmpPath := tmp.Name()
		if err := tmp.Close(); err != nil {
			_ = os.Remove(tmpPath)
			return err
		}
		_ = os.Remove(tmpPath) // Litestream requires an absent output.
		// The check below opens the temp database read-only; a restored database
		// is in WAL mode, so that open attaches a -wal and a -shm it cannot
		// delete on close. Remove exactly the names derived from this restore's
		// own temp path, never a glob: another restore's .restore-* files may be
		// in flight beside it.
		defer removeRestoreTemp(tmpPath)
		client := &pinnedClient{files: files}
		replica := litestream.NewReplicaWithClient(nil, client)
		opt := litestream.NewRestoreOptions()
		opt.OutputPath = tmpPath
		opt.TXID = ltx.TXID(plan.MaxTXID)
		if err := replica.Restore(ctx, opt); err != nil {
			return fmt.Errorf("restore selected generation: %w", err)
		}
		f, err := os.Open(tmpPath)
		if err != nil {
			return err
		}
		checksum, checkErr := ltx.ChecksumReader(f, files[0].infoPageSize)
		closeErr := f.Close()
		if checkErr != nil {
			return fmt.Errorf("checksum restored database: %w", checkErr)
		}
		if closeErr != nil {
			return closeErr
		}
		// HeaderFlagNoChecksum is Litestream's pinned on-disk format. Its LTX
		// trailer still detects corruption across the complete file, but is not
		// an authenticity proof and intentionally has no database checksum to
		// compare. A moves-owned object digest must supply authenticity. When
		// the selected chain carries a database checksum, require the restored
		// database to match it.
		if finalChecksum != 0 && checksum != finalChecksum {
			return fmt.Errorf("restored database checksum %s, want %s", checksum, finalChecksum)
		}
		if err := lease.CheckDBContext(ctx, tmpPath); err != nil {
			return fmt.Errorf("check restored database: %w", err)
		}
		// The check may attach a WAL and a shared-memory file; neither is part
		// of the restore's result and neither may follow the rename.
		removeRestoreSidecars(tmpPath)
		if err := syncFile(tmpPath); err != nil {
			return err
		}
		if err := os.Rename(tmpPath, dbPath); err != nil {
			return fmt.Errorf("install restored database: %w", err)
		}
		if err := writeGeneration(dbPath+".gen", r.Generation); err != nil {
			return fmt.Errorf("install generation marker: %w", err)
		}
		return syncDir(filepath.Dir(dbPath))
	})
}

// InstallEmpty installs a separately proved initial-empty admission. It writes
// no database and no zero-TXID LTX. Authority must prove both the complete
// empty remote baseline and first-creator status before local mutation.
func (r ColdRestore) InstallEmpty(ctx context.Context, dbPath string, authority func(context.Context) error) error {
	if ctx == nil || r.Work == nil || !safeGeneration(r.Generation) || dbPath == "" || authority == nil {
		return ErrUnsupported
	}
	if err := authority(ctx); err != nil {
		return err
	}
	var captured string
	if err := r.Work.With(ctx, r.Identity, func() error { var e error; captured, e = readGeneration(dbPath + ".gen"); return e }); err != nil {
		return err
	}
	if captured != "" && captured != r.Generation {
		if r.RetireArtifacts == nil {
			return ErrUnsupported
		}
		if err := r.RetireArtifacts(ctx, captured); err != nil {
			return err
		}
	}
	if captured == "" && r.BeforeLegacyInstall != nil {
		if err := r.BeforeLegacyInstall(ctx, dbPath, r.Generation); err != nil {
			return err
		}
	}
	return r.Work.With(ctx, r.Identity, func() error {
		current, err := readGeneration(dbPath + ".gen")
		if err != nil {
			return err
		}
		if captured != "" && captured != r.Generation {
			if current != "" {
				return ErrChanged
			}
		} else if current != captured {
			return ErrChanged
		}
		if r.BeforeInstallLocked != nil {
			if err := r.BeforeInstallLocked(ctx, dbPath, r.Generation); err != nil {
				return err
			}
		}
		if err := purgeDB(dbPath); err != nil {
			return err
		}
		if err := writeGeneration(dbPath+".gen", r.Generation); err != nil {
			return err
		}
		return syncDir(filepath.Dir(dbPath))
	})
}

func safeGeneration(s string) bool {
	return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00\r\n")
}

// removeRestoreSidecars removes the -wal and -shm files a check of the temp
// database at tmpPath may have attached. Best effort, as in
// moves.BuildGeneration: a name that is already absent is the normal case.
func removeRestoreSidecars(tmpPath string) {
	for _, p := range []string{tmpPath + "-wal", tmpPath + "-shm"} {
		_ = os.Remove(p)
	}
}

// removeRestoreTemp is the failure-path cleanup: the temp database and its
// sidecars, by the names this restore derived and no others.
func removeRestoreTemp(tmpPath string) {
	_ = os.Remove(tmpPath)
	removeRestoreSidecars(tmpPath)
}

func purgeDB(path string) error {
	// ".seed" is the activator's seed marker (activator.SeedMarkerSuffix),
	// spelled out because the activator imports this package: a marker
	// must not outlive the database it named (re-review finding 8).
	for _, p := range []string{path, path + "-wal", path + "-shm", path + ".gen", path + ".seed"} {
		if err := os.Remove(p); err != nil && !errors.Is(err, os.ErrNotExist) {
			return fmt.Errorf("purge %s: %w", p, err)
		}
	}
	for _, p := range []string{
		path + "-litestream",
		filepath.Join(filepath.Dir(path), ".activator-"+filepath.Base(path)+"-litestream"),
	} {
		if err := os.RemoveAll(p); err != nil {
			return fmt.Errorf("purge %s: %w", p, err)
		}
	}
	return nil
}

// WriteGenerationMarker writes the <dbPath>.gen marker beside a database:
// the prepared generation's name as plain text, the one format the cold
// restore's readGeneration accepts. It is the shared writer the moves
// landing uses for the database it installs (moves Plan 2 ruling G2; the
// marker the moves spec §5.1 draws as JSON is written in this form instead,
// and the incarnation and location it would have carried are vouched for by
// the host guard record and the claim). A name readGeneration would refuse
// is refused here, with nothing written.
func WriteGenerationMarker(path, generation string) error {
	if !safeGeneration(generation) {
		return fmt.Errorf("hostbox: %q is not a generation marker: %w", generation, ErrUnsupported)
	}
	return writeGeneration(path, generation)
}

// ReadGenerationMarker reads a marker WriteGenerationMarker wrote. An
// absent marker is "" with no error (the origin).
func ReadGenerationMarker(path string) (string, error) { return readGeneration(path) }

func readGeneration(path string) (string, error) {
	b, err := os.ReadFile(path)
	if errors.Is(err, os.ErrNotExist) {
		return "", nil
	}
	if err != nil {
		return "", err
	}
	g := strings.TrimSpace(string(b))
	if g == "" || strings.ContainsAny(g, "/\\\x00\r\n") {
		return "", errors.New("invalid generation marker")
	}
	return g, nil
}

func writeGeneration(path, generation string) error {
	tmp, err := os.CreateTemp(filepath.Dir(path), ".generation-*")
	if err != nil {
		return err
	}
	name := tmp.Name()
	defer os.Remove(name)
	if _, err = io.WriteString(tmp, generation+"\n"); err == nil {
		err = tmp.Sync()
	}
	if closeErr := tmp.Close(); err == nil {
		err = closeErr
	}
	if err != nil {
		return err
	}
	return os.Rename(name, path)
}

func syncFile(path string) error {
	f, err := os.Open(path)
	if err != nil {
		return err
	}
	defer f.Close()
	return f.Sync()
}
func syncDir(path string) error {
	f, err := os.Open(path)
	if err != nil {
		return err
	}
	defer f.Close()
	return f.Sync()
}

type pinnedFile struct {
	plan         genselect.File
	data         []byte
	info         *ltx.FileInfo
	infoPageSize int
}

func loadPinned(ctx context.Context, reader genselect.Reader, bucket string, plan *genselect.Plan) ([]pinnedFile, ltx.Checksum, error) {
	files := make([]pinnedFile, 0, len(plan.Files))
	var last ltx.Checksum
	var previous uint64
	for i, selected := range plan.Files {
		if i != 0 && (previous == ^uint64(0) || !ltx.IsContiguous(ltx.TXID(previous), ltx.TXID(selected.MinTXID), ltx.TXID(selected.MaxTXID))) {
			return nil, 0, genselect.ErrGap
		}
		rc, err := reader.OpenVersion(ctx, bucket, selected.Key, selected.VersionID)
		if err != nil {
			return nil, 0, fmt.Errorf("open selected %s@%s: %w", selected.Key, selected.VersionID, err)
		}
		data, readErr := io.ReadAll(rc)
		closeErr := rc.Close()
		if readErr != nil {
			return nil, 0, fmt.Errorf("read selected %s: %w", selected.Key, readErr)
		}
		if closeErr != nil {
			return nil, 0, fmt.Errorf("close selected %s: %w", selected.Key, closeErr)
		}
		if int64(len(data)) != selected.Size {
			return nil, 0, fmt.Errorf("selected %s size %d, want %d", selected.Key, len(data), selected.Size)
		}
		dec := ltx.NewDecoder(bytes.NewReader(data))
		if err := dec.DecodeHeader(); err != nil {
			return nil, 0, fmt.Errorf("decode %s header: %w", selected.Key, err)
		}
		h := dec.Header()
		if uint64(h.MinTXID) != selected.MinTXID || uint64(h.MaxTXID) != selected.MaxTXID {
			return nil, 0, fmt.Errorf("selected %s range mismatch", selected.Key)
		}
		// Litestream emits HeaderFlagNoChecksum but copies the last input's
		// database checksum. Re-encode one page at a time for that format,
		// retaining the original checksum for the final restored database.
		// Nothing uses the converted bytes until the original proof passes.
		var encoded bytes.Buffer
		var enc *ltx.Encoder
		if !h.NoChecksum() {
			enc, err = ltx.NewEncoder(&encoded)
			if err != nil {
				return nil, 0, err
			}
			normalized := h
			normalized.Flags |= ltx.HeaderFlagNoChecksum
			normalized.PreApplyChecksum = 0
			if err = enc.EncodeHeader(normalized); err != nil {
				return nil, 0, err
			}
		}
		buf := make([]byte, h.PageSize)
		var ph ltx.PageHeader
		for {
			err = dec.DecodePage(&ph, buf)
			if errors.Is(err, io.EOF) {
				break
			}
			if err != nil {
				return nil, 0, fmt.Errorf("decode %s: %w", selected.Key, err)
			}
			if enc != nil {
				if err := enc.EncodePage(ph, buf); err != nil {
					return nil, 0, err
				}
			}
		}
		if err := dec.Close(); err != nil {
			return nil, 0, fmt.Errorf("verify %s trailer: %w", selected.Key, err)
		}
		t := dec.Trailer()
		if i == 0 && plan.Pin != nil && plan.Pin.SnapshotCRC64 != 0 && uint64(t.FileChecksum) != plan.Pin.SnapshotCRC64 {
			return nil, 0, genselect.ErrPinnedMismatch
		}
		if enc != nil {
			if err := enc.Close(); err != nil {
				return nil, 0, err
			}
			data = encoded.Bytes()
		}
		info := &ltx.FileInfo{Level: selected.Level, MinTXID: h.MinTXID, MaxTXID: h.MaxTXID, PreApplyChecksum: h.PreApplyChecksum, PostApplyChecksum: t.PostApplyChecksum, Size: int64(len(data))}
		files = append(files, pinnedFile{plan: selected, data: data, info: info, infoPageSize: int(h.PageSize)})
		last, previous = t.PostApplyChecksum, selected.MaxTXID
	}
	if previous != plan.MaxTXID {
		return nil, 0, fmt.Errorf("selected final TXID %d, want %d", previous, plan.MaxTXID)
	}
	return files, last, nil
}

type pinnedClient struct{ files []pinnedFile }

func (*pinnedClient) Type() string               { return "selected-versions" }
func (*pinnedClient) Init(context.Context) error { return nil }
func (*pinnedClient) SetLogger(*slog.Logger)     {}
func (c *pinnedClient) LTXFiles(_ context.Context, level int, seek ltx.TXID, _ bool) (ltx.FileIterator, error) {
	var a []*ltx.FileInfo
	for _, f := range c.files {
		if f.info.Level == level && f.info.MaxTXID >= seek {
			cp := *f.info
			a = append(a, &cp)
		}
	}
	sort.Slice(a, func(i, j int) bool { return a[i].MinTXID < a[j].MinTXID })
	return ltx.NewFileInfoSliceIterator(a), nil
}
func (c *pinnedClient) OpenLTXFile(_ context.Context, level int, min, max ltx.TXID, offset, size int64) (io.ReadCloser, error) {
	for _, f := range c.files {
		if f.info.Level == level && f.info.MinTXID == min && f.info.MaxTXID == max {
			if offset < 0 || size < 0 || offset > int64(len(f.data)) {
				return nil, os.ErrInvalid
			}
			end := int64(len(f.data))
			if size > 0 && offset+size < end {
				end = offset + size
			}
			return io.NopCloser(bytes.NewReader(f.data[offset:end])), nil
		}
	}
	return nil, os.ErrNotExist
}
func (*pinnedClient) WriteLTXFile(context.Context, int, ltx.TXID, ltx.TXID, io.Reader) (*ltx.FileInfo, error) {
	return nil, ErrUnsupported
}
func (*pinnedClient) DeleteLTXFiles(context.Context, []*ltx.FileInfo) error { return ErrUnsupported }
func (*pinnedClient) DeleteAll(context.Context) error                       { return ErrUnsupported }