package host

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfiles"
	"github.com/carlosframework/platform/internal/hostrelocate"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// RelocationWakeInputs is the relocating kind's ClaimWakeInputs (claim
// Plan 11 R1). Field names are distinct (finding 14): FilesManifest is
// moves' C8 provider, FilesStore the data bucket as a release.Store.
type RelocationWakeInputs struct {
	Launch        func(context.Context, registry.Route) (ClaimServingLaunch, error)
	FilesManifest ClaimFilesManifest
	Markers       ClaimMarkerStore
	Claims        release.Store
	FilesStore    release.Store
	Data          genselect.Reader
	Bucket        string
	Region        string
	// HomeRegion is the box's verified home binding (hostbox.Actor's), or
	// "": the selection and the fresh publisher see the holder through it
	// (park.HolderRegionView), as main's serving owners do.
	HomeRegion string
	Box        string
	NewID      func() string
	Now        func() time.Time
	// Releases verifies a journaled release at its receipt (stage A's
	// StopReceiptReleases over main's Complete).
	Releases hostclaim.ReleaseAuthority

	mu        sync.Mutex
	fences    map[string]hostwrite.Work          // host\x00session: the admitted spawn selection Materialize ran under
	published map[string]ClaimPublicationReceipt // host\x00session: P's prepared marker
	fresh     *claimServingOwners                // a fresh generation's marker: main's writer (ruling 47)
}

var (
	_ ClaimWakeInputs        = (*RelocationWakeInputs)(nil)
	_ ClaimFencedPublication = (*RelocationWakeInputs)(nil)
)

func (r *RelocationWakeInputs) valid() bool {
	return r != nil && r.Launch != nil && r.FilesManifest != nil && r.Markers != nil && r.Claims != nil && r.FilesStore != nil && r.Data != nil &&
		r.Bucket != "" && r.Region != "" && r.Box != "" && r.NewID != nil && r.Now != nil && r.Releases != nil
}

func (r *RelocationWakeInputs) probe(o hostclaim.Observed, p activator.StartPolicy) error {
	if !r.valid() {
		return hostwrite.ErrUnavailable
	}
	c := o.Claim
	if o.Absent || c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindRelocating || c.Op.Phase != hostclaim.PhaseRelocatingProbing || c.Op.Exec != r.Box || !p.Probe || c.Op.Journal["stuck"] != "" {
		return fmt.Errorf("%w: not this box's relocation probe", hostrelocate.ErrRefused)
	}
	return nil
}

// freshOwners is main's serving publisher bound to this box and region,
// used only through its kind-free halves (publishFresh, freshEvidence).
func (r *RelocationWakeInputs) freshOwners() *claimServingOwners {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.fresh == nil {
		r.fresh = &claimServingOwners{HomeRegion: r.HomeRegion, Data: r.Data, Bucket: r.Bucket, Region: r.Region, Box: r.Box, NewID: r.NewID, Now: r.Now, Sources: ClaimServingSources{Markers: r.Markers}}
	}
	return r.fresh
}

func (r *RelocationWakeInputs) request(o hostclaim.Observed) genselect.Request {
	return genselect.Request{Bucket: r.Bucket, Region: r.Region, Host: o.Host, Holder: park.HolderRegionView(*o.Claim.Holder, r.HomeRegion, r.Region)}
}

func preparedAuthority(c *hostclaim.Claim) (hostclaim.PreparedAuthority, error) {
	var auth hostclaim.PreparedAuthority
	if json.Unmarshal([]byte(c.Op.Journal["prepared_authority"]), &auth) != nil || auth.Generation == "" || auth.Generation != c.Op.Journal["prepared"] {
		return hostclaim.PreparedAuthority{}, hostbox.ErrChanged
	}
	return auth, nil
}

// emptyPrepared reports whether P is the evidenced empty bootstrap: the
// journaled pin says so (RecoveryPin.Empty, which the pinner admits only
// for a release reporting final_txid 0 or a holder with no writer) and the
// prepared authority carries no install manifest. The two must agree: an
// unreadable pin, or a pin and an authority that disagree, is an error,
// never an empty P. A missing install manifest alone is not evidence of
// an empty database, and a later attempt's empty listing is legitimate
// only over an empty P (a pinned snapshot lives in the location, so an
// empty listing under a non-empty pin is a wrong bucket or a listing not
// yet visible).
func emptyPrepared(c *hostclaim.Claim, auth hostclaim.PreparedAuthority) (bool, error) {
	var ref hostclaim.RecoveryPin
	if json.Unmarshal([]byte(c.Op.Journal["pin"]), &ref) != nil {
		return false, fmt.Errorf("%w: the journaled pin is unreadable", hostbox.ErrChanged)
	}
	if ref.Empty != (auth.InstallManifest == "") {
		return false, fmt.Errorf("%w: the pin and the prepared authority disagree on an empty P", hostbox.ErrChanged)
	}
	return ref.Empty, nil
}

// Select (§5.8 "The probe's admission"): the first attempt opens P with
// KeepStamp, certified only by this operation's verified release (C4);
// it is a prepared admission (Prepared, main's pre-replication marker)
// when P has a snapshot, and an empty one otherwise (ruling 47). Every
// later attempt is cold into a fresh generation from the predecessor.
func (r *RelocationWakeInputs) Select(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (ClaimLaunchInputs, error) {
	if err := r.probe(o, p); err != nil {
		return ClaimLaunchInputs{}, err
	}
	c, h := o.Claim, o.Claim.Holder
	if rt.Host != o.Host || !strings.HasPrefix(rt.DBPath, "/") {
		return ClaimLaunchInputs{}, hostbox.ErrUnsupported
	}
	l, err := r.Launch(ctx, rt)
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	if l.Identity.Config == "" || l.Identity.Release == "" || l.Executable == "" || len(l.ExecutableSHA256) != 64 || l.Version == "" {
		return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
	}
	if l.Version != c.Op.Journal["install_release"] {
		return ClaimLaunchInputs{}, fmt.Errorf("%w: the staged release %s is not the pinned %s", hostrelocate.ErrRefused, l.Version, c.Op.Journal["install_release"])
	}
	prepared := c.Op.Journal["prepared"]
	in := ClaimLaunchInputs{Roles: []string{"main"}, Identity: l.Identity, Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: append([]string(nil), l.Env...),
		DBPath: rt.DBPath, Version: l.Version, Staged: l.Staged, Limit: l.Limit, ReadyPath: "/"}
	adm := hostclaim.Admission{ObservedOpID: c.Op.ID, ObservedPhase: hostclaim.PhaseRelocatingProbing, ObservedSession: h.Session, ObservedGeneration: h.Generation,
		Box: r.Box, Mode: "cold", Region: r.Region, PredecessorRegion: r.Region}
	if n := len(h.Admissions); n > 0 {
		adm.PredecessorRegion = h.Admissions[n-1].Region
	}
	if !hostclaim.PreparedOpened(c) {
		certified, err := hostrelocate.CertifiedFor(c)
		if err != nil {
			return ClaimLaunchInputs{}, err
		}
		if certified {
			var rel hostclaim.Release
			if err := json.Unmarshal([]byte(c.Op.Journal["release"]), &rel); err != nil {
				return ClaimLaunchInputs{}, err
			}
			if err := r.Releases.VerifyRelease(ctx, *c, rel); err != nil {
				return ClaimLaunchInputs{}, err
			}
		}
		auth, err := preparedAuthority(c)
		if err != nil {
			return ClaimLaunchInputs{}, err
		}
		empty, err := emptyPrepared(c, auth)
		if err != nil {
			return ClaimLaunchInputs{}, err
		}
		adm.NewGeneration, adm.KeepStamp, adm.PredecessorCertified = prepared, true, certified
		if empty {
			// An evidenced empty pin opened as P: no snapshot, so no root
			// until the first transaction; main's fenced publisher waits
			// for it (ErrClaimRootPending).
			in.Empty = true
			in.Admission = adm
			return in, nil
		}
		im, _, err := readInstallManifest(ctx, r.Data, r.Bucket, moves.GenerationPrefix(h.Location, prepared))
		if err != nil {
			return ClaimLaunchInputs{}, err
		}
		plan, err := genselect.SelectPrepared(ctx, r.Data, r.request(o), genselect.PreparedPin{Generation: prepared, SnapshotKey: im.SnapshotKey, SnapshotVersion: im.SnapshotVersion, SnapshotTXID: im.N, SnapshotCRC64: im.SnapshotCRC64})
		if err != nil {
			return ClaimLaunchInputs{}, err
		}
		in.Prepared = true
		adm.RemoteMaxTXID, in.Baseline = plan.MaxTXID, plan
		in.Admission = adm
		return in, nil
	}
	adm.NewGeneration = r.NewID()
	if !claimFreshGeneration(h, adm.NewGeneration) {
		return ClaimLaunchInputs{}, hostwrite.ErrUnavailable
	}
	plan, err := genselect.Select(ctx, r.Data, r.request(o))
	if errors.Is(err, genselect.ErrEmpty) {
		// A later attempt follows P's admitted session, which ran: a
		// listing that finds nothing is absence (nothing replicated, or
		// not yet visible), never an empty predecessor. Main's serving
		// rule admits an empty cold admission only for a holder that never
		// admitted a writer (claimEmptyBootstrap), which a later attempt
		// never is (fix round 1, I2).
		return ClaimLaunchInputs{}, fmt.Errorf("%w: the location lists nothing after P's admitted session", hostbox.ErrChanged)
	}
	if err != nil {
		return ClaimLaunchInputs{}, err
	}
	adm.RemoteMaxTXID, in.Baseline = plan.MaxTXID, plan
	in.Admission = adm
	return in, nil
}

// Materialize, under the admitted spawn selection: the held identities
// proved, C8 establishes the files manifest (Establish publishes
// <host>.env's exact bytes before it returns, O1), the selection is
// recorded as the session's fence, and on P's prepared opening P's marker
// is published under that fence (main's prepared-publication contract,
// ruling 47). The held render and release ran at the local install.
func (r *RelocationWakeInputs) Materialize(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error {
	if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
		return err
	}
	if work == nil || in.Identity.Config == "" || in.Identity.Release == "" {
		return hostwrite.ErrUnavailable
	}
	fence := writerFence(work, in.Identity)
	if err := fence(ctx); err != nil {
		return err
	}
	if err := r.FilesManifest.Establish(ctx, o, in, work); err != nil {
		return err
	}
	h := o.Claim.Holder
	r.mu.Lock()
	if r.fences == nil {
		r.fences = map[string]hostwrite.Work{}
	}
	r.fences[claimPublishedKey(o.Host, h.Session)] = work
	r.mu.Unlock()
	if !in.Prepared {
		return nil
	}
	receipt, err := r.preparedMarker(ctx, o, fence)
	if err != nil {
		return err
	}
	r.record(o.Host, h.Session, receipt)
	return nil
}

func (r *RelocationWakeInputs) record(host string, session uint64, receipt ClaimPublicationReceipt) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.published == nil {
		r.published = map[string]ClaimPublicationReceipt{}
	}
	r.published[claimPublishedKey(host, session)] = receipt
}

// preparedMarker establishes P's verified marker under fence: install.json
// is the journaled prepared authority's bytes, and the location's
// generation.json names P with the build's pin (left as it is when it
// already does) or supersedes the current marker by its exact ETag.
func (r *RelocationWakeInputs) preparedMarker(ctx context.Context, o hostclaim.Observed, fence func(context.Context) error) (ClaimPublicationReceipt, error) {
	c, h := o.Claim, o.Claim.Holder
	auth, err := preparedAuthority(c)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
	defer cancel()
	im, raw, err := readInstallManifest(bounded, r.Data, r.Bucket, moves.GenerationPrefix(h.Location, auth.Generation))
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	sum := sha256.Sum256(raw)
	if hex.EncodeToString(sum[:]) != auth.SHA256 || im.Host != o.Host || im.N != auth.SnapshotTXID || h.Generation != auth.Generation {
		return ClaimPublicationReceipt{}, fmt.Errorf("%w: install.json is not the prepared authority's", hostbox.ErrChanged)
	}
	root := genselect.File{Key: im.SnapshotKey, VersionID: im.SnapshotVersion, Level: 9, MinTXID: 1, MaxTXID: im.N}
	current, err := currentMarker(bounded, r.Data, r.Bucket, h) // moves' reader, one implementation
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	expected := ""
	if current != nil {
		got, err := r.Data.GetVersion(bounded, r.Bucket, markerKey(h), current.ID)
		if err != nil {
			return ClaimPublicationReceipt{}, err
		}
		var marker genselect.Marker
		if err := json.Unmarshal(got, &marker); err != nil {
			return ClaimPublicationReceipt{}, err
		}
		if marker.Op == auth.Generation {
			if marker.Incarnation != h.Incarnation || marker.Host != o.Host || marker.Box != r.Box || marker.ManifestSHA256 != auth.SHA256 {
				return ClaimPublicationReceipt{}, hostbox.ErrChanged
			}
			return ClaimPublicationReceipt{MarkerVersion: current.ID, Root: root}, nil
		}
		expected = current.ETag
	}
	body, err := json.Marshal(genselect.Marker{Op: auth.Generation, Incarnation: h.Incarnation, Host: o.Host, Box: r.Box, InstalledAt: r.Now().UTC(), ManifestSHA256: auth.SHA256, Pin: manifestPin(im)})
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if err := fence(bounded); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	version, err := r.Markers.PutMarker(bounded, r.Bucket, markerKey(h), body, expected, fence)
	if err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if version == "" {
		return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable
	}
	return ClaimPublicationReceipt{MarkerVersion: version, Root: root}, nil
}

// EmptyAuthority: P's opening is empty only with an evidenced empty pin
// (the pinner's RecoveryPin.Empty, which the prepared authority agrees
// with). A later attempt is never empty: it follows P's admitted session,
// and main's serving rule admits an empty cold admission only for a holder
// that never admitted a writer (fix round 1, I2).
func (r *RelocationWakeInputs) EmptyAuthority(_ context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
	if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
		return err
	}
	if !in.Empty || in.Baseline != nil || in.Prepared {
		return hostbox.ErrChanged
	}
	auth, err := preparedAuthority(o.Claim)
	if err != nil {
		return err
	}
	empty, err := emptyPrepared(o.Claim, auth)
	if err != nil {
		return err
	}
	if in.Admission.NewGeneration != o.Claim.Op.Journal["prepared"] || hostclaim.PreparedOpened(o.Claim) {
		return fmt.Errorf("%w: a later attempt follows P's admitted session and is never empty", hostbox.ErrChanged)
	}
	if !empty {
		return fmt.Errorf("%w: no evidenced empty pin", hostrelocate.ErrRefused)
	}
	return nil
}

// Publish is main's pre-replication call for a prepared admission
// (ClaimWakeAssembly s.Replicate when in.Prepared). It finds, or writes,
// P's marker under the fence Materialize recorded for the admitted session
// and records the receipt; without that fence it refuses. Any other call
// is refused: a fresh generation is published fenced (PublishFenced).
func (r *RelocationWakeInputs) Publish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error {
	if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
		return err
	}
	h := o.Claim.Holder
	if !in.Prepared || in.Admission.NewGeneration != o.Claim.Op.Journal["prepared"] || h.Generation != in.Admission.NewGeneration {
		return hostbox.ErrChanged
	}
	r.mu.Lock()
	work := r.fences[claimPublishedKey(o.Host, h.Session)]
	r.mu.Unlock()
	if work == nil {
		return fmt.Errorf("%w: no admitted writer's fence for session %d", hostbox.ErrChanged, h.Session)
	}
	receipt, err := r.preparedMarker(ctx, o, writerFence(work, in.Identity))
	if err != nil {
		return err
	}
	r.record(o.Host, h.Session, receipt)
	return nil
}

// PublishFenced is main's fresh-generation marker writer, for a later
// attempt's generation and an empty P's opening (ruling 47).
func (r *RelocationWakeInputs) PublishFenced(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication) (ClaimPublicationReceipt, error) {
	if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
		return ClaimPublicationReceipt{}, err
	}
	if in.Prepared {
		return ClaimPublicationReceipt{}, hostbox.ErrChanged
	}
	return r.freshOwners().publishFresh(ctx, o, in, pub)
}

// AdmissionEvidence binds a relocation probe admission to its observation
// and to the certified baseline, read again: on P's opening the exact
// prepared pin and P's files manifest by digest (prepared_authority); on a
// later attempt the predecessor's fresh baseline.
func (r *RelocationWakeInputs) AdmissionEvidence(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) error {
	if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
		return err
	}
	c, h := o.Claim, o.Claim.Holder
	if in.ObservedOpID != c.Op.ID || in.ObservedPhase != hostclaim.PhaseRelocatingProbing || in.ObservedSession != h.Session || in.ObservedGeneration != h.Generation || in.Box != r.Box || in.Mode != "cold" || in.Coverage != "" || len(in.AdditionalWritesLost) != 0 || in.Region != r.Region {
		return hostbox.ErrChanged
	}
	prepared := c.Op.Journal["prepared"]
	auth, err := preparedAuthority(c)
	if err != nil {
		return err
	}
	empty, err := emptyPrepared(c, auth)
	if err != nil {
		return err
	}
	if in.NewGeneration == prepared {
		certified, err := hostrelocate.CertifiedFor(c)
		if err != nil {
			return err
		}
		if !in.KeepStamp || hostclaim.PreparedOpened(c) || in.PredecessorCertified != certified {
			return hostbox.ErrChanged
		}
		raw, _, err := r.FilesStore.Get(ctx, hostfiles.ManifestKey(moves.GenerationFilesPrefix(h.Location, prepared)))
		if err != nil {
			return err
		}
		if sum := sha256.Sum256(raw); hex.EncodeToString(sum[:]) != auth.FilesManifestSHA256 {
			return fmt.Errorf("%w: P's files manifest is not the prepared authority's", hostbox.ErrChanged)
		}
		if baseline == nil {
			if !empty || in.RemoteMaxTXID != 0 {
				return hostbox.ErrChanged
			}
			return nil
		}
		if baseline.Generation != prepared || baseline.MaxTXID != in.RemoteMaxTXID || baseline.MaxTXID != auth.SnapshotTXID {
			return hostbox.ErrChanged
		}
		return nil
	}
	if in.KeepStamp || in.PredecessorCertified || !claimFreshGeneration(h, in.NewGeneration) {
		return hostbox.ErrChanged
	}
	if baseline == nil {
		// A later attempt follows P's admitted session and is never an
		// empty cold admission (fix round 1, I2).
		return hostbox.ErrChanged
	}
	fresh, err := genselect.Select(ctx, r.Data, r.request(o))
	if err != nil {
		return err
	}
	if fresh.Generation != baseline.Generation || fresh.MaxTXID != baseline.MaxTXID || in.RemoteMaxTXID != fresh.MaxTXID || !reflect.DeepEqual(fresh.Files, baseline.Files) {
		return hostbox.ErrChanged
	}
	return nil
}

// PublicationEvidence: on P's prepared opening, this process's receipt for
// the admitted session with the marker still exactly that version; on any
// fresh generation, main's freshEvidence.
func (r *RelocationWakeInputs) PublicationEvidence(ctx context.Context, o hostclaim.Observed) error {
	if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil {
		return err
	}
	h := o.Claim.Holder
	r.mu.Lock()
	receipt, prepared := r.published[claimPublishedKey(o.Host, h.Session)]
	r.mu.Unlock()
	if !prepared {
		return r.freshOwners().freshEvidence(ctx, o)
	}
	current, err := currentMarker(ctx, r.Data, r.Bucket, h)
	if err != nil {
		return err
	}
	if current == nil || current.ID != receipt.MarkerVersion || h.Generation != o.Claim.Op.Journal["prepared"] {
		return hostbox.ErrChanged
	}
	return nil
}