package host

import (
	"context"
	"crypto/ed25519"
	"errors"
	"log"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/boxchoice"
	"github.com/carlosframework/platform/internal/boxvolumes"
	"github.com/carlosframework/platform/internal/edgesfile"
	"github.com/carlosframework/platform/internal/genselect"
	"github.com/carlosframework/platform/internal/hostactors"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfence"
	"github.com/carlosframework/platform/internal/hostfloor"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/hostrelocate"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/termfence"
)

// RelocationRuntime is one agent's relocation: its worker, participant and
// effects, built once the claim assembly is bound. It replays relocations;
// it starts none (Worker.Start is the operator's and the placer's, not
// this pass's).
type RelocationRuntime struct {
	Worker      relocationStepper
	Participant relocationStepper
	Box         string // this box: its alerts, and the completions it finishes
	// Convert is J3 for a landing claim (convertFencedLanding); Finish is
	// RelocationEffects.Finish for a completion this box has not yet
	// marked discharged (ruling 80).
	Convert func(context.Context, release.Store, hostclaim.Observed) error
	Finish  func(ctx context.Context, host, op string) error
	// Hosts, when set, is the hosts a pass reads instead of every claim
	// in the bucket. The completion-only runtime's (ruling 82) are this
	// box's registry rows and every host its own op journals name
	// (relocate/ops/*/{probe-fence,attempts}.json), so a completion whose
	// row is gone is still visited and its refusal said (Task 10 review
	// M1). An error beside a non-nil list is said, and the listed hosts
	// are still read.
	Hosts func() ([]string, error)
	// Self, when Box is empty, answers this box: the completion-only
	// runtime has no roster read of its own, so it asks the bound
	// assembly's verified stop identity. It is asked at most once a pass,
	// and only once a serving claim with an undischarged completion is
	// found (review M2), so a box with nothing to finish makes no identity
	// read. A pass that cannot answer it finishes nothing (carry 8: a
	// completion is finished only on the box it serves on).
	Self    func(context.Context) (string, error)
	DataDir string // the managed data root: relocate/ journals (N9)
	alerts  relocationAlerts
	// upgrade, when set, builds the full pass this completion-only runtime
	// stands in for while region HA is on. The first pass, bounded like
	// every pass, tries it (review M9); a pass whose build fails tries it
	// again, until it builds or answers that the owners are not bound. A
	// read that failed once is never taken as "no relocation on this box".
	upgrade func(context.Context) (*RelocationRuntime, error)
	// config is the assembly config the full pass was built with
	// (NewRelocationRuntime); zero for the completion-only runtime. No
	// production code reads it: it is the seam Task 10's review M7 test
	// asserts the held-stage evidence (ReleasePub, Ratchet) through
	// (relocatepass_carries_test.go), kept for that reading alone.
	config RelocationAssemblyConfig
}

// relocationStepper is hostrelocate.Worker's and Participant's Step.
type relocationStepper interface {
	Step(context.Context, string) (hostclaim.Observed, error)
}

// relocationStores is rulings 50 and 51: the data bucket as a release.Store
// of its own (production's lease.VersionedStore is not one), and the AWS
// signing region kept apart from the logical placement region.
type relocationStores struct {
	Files       release.Store
	Versions    lease.VersionedStore
	Bucket      string
	Region      string // logical: genselect, pins, the worker, the chooser
	LeaseRegion string // AWS: lease.Acquire, lease.NewVersionedS3
}

func openRelocationStores(ctx context.Context, self hostactors.Box, dataBucket string, files func(context.Context, string) (release.Store, error), versioned func(context.Context, string, string) (lease.VersionedStore, error)) (relocationStores, error) {
	if dataBucket == "" || self.Region == "" || self.Cloud.Region == "" || files == nil || versioned == nil {
		return relocationStores{}, hostwrite.ErrUnavailable
	}
	f, err := files(ctx, dataBucket)
	if err != nil {
		return relocationStores{}, err
	}
	v, err := versioned(ctx, dataBucket, self.Cloud.Region)
	if err != nil {
		return relocationStores{}, err
	}
	return relocationStores{Files: f, Versions: v, Bucket: dataBucket, Region: self.Region, LeaseRegion: self.Cloud.Region}, nil
}

// relocationLeases is the worker's lease provider (ruling 51): the replica
// lease on a holder's location, signed for the cloud region.
func relocationLeases(s relocationStores) func(context.Context, string) (hostrelocate.Lease, error) {
	return func(ctx context.Context, location string) (hostrelocate.Lease, error) {
		held, err := lease.Acquire(ctx, s.Bucket, location, s.LeaseRegion)
		if err != nil {
			return nil, err
		}
		return held, nil
	}
}

// relocationAssemblyConfig fills Task 8a's config from the bound assembly,
// the stores and main's participation discharge.
func relocationAssemblyConfig(a *ClaimAssembly, providers ClaimProviders, s relocationStores, reg *registry.Registry, act RelocationActivator, d hostactors.Directory, optDir string, discharge func(context.Context, hostclaim.Observed) error) (RelocationAssemblyConfig, error) {
	src, ok := providers.Selector.(hostbox.WorkSource)
	if !ok || a == nil || reg == nil || act == nil || discharge == nil || optDir == "" || a.Guard.DataDir == "" {
		return RelocationAssemblyConfig{}, hostwrite.ErrUnavailable
	}
	dd := a.Guard.DataDir
	return RelocationAssemblyConfig{Directory: d, Data: s.Files, Versions: s.Versions, Bucket: s.Bucket, Region: s.Region, LeaseRegion: s.LeaseRegion,
		Registry: reg, DataDir: dd, OptDir: optDir, WorkDir: filepath.Join(dd, "relocate", "build"), ReleaseDir: filepath.Join(dd, "relocate", "release"),
		Activator: act, Source: src, Discharge: discharge, Now: time.Now}, nil
}

// relocationRuntimeConfig is relocationAssemblyConfig with what the agent
// binds beyond it. ReleasePub and Ratchet are the held stage's evidence
// (carry 4): the agent's release key (CARLOS_RELEASE_PUBKEY, the key the
// held-stage pass verifies under) and the edge's pointer digest ratchet,
// pointerRatchet(false, ...), exactly as moves' held stage takes it in the
// agent (movespass.go, heldReleaseReady); never root's. The secrets key,
// when the box has one, renders the held config.
func relocationRuntimeConfig(getenv func(string) string, a *ClaimAssembly, providers ClaimProviders, s relocationStores, reg *registry.Registry, act RelocationActivator, d hostactors.Directory, in relocationRuntimeInputs, discharge func(context.Context, hostclaim.Observed) error) (RelocationAssemblyConfig, error) {
	if getenv == nil || len(in.Pub) != ed25519.PublicKeySize {
		return RelocationAssemblyConfig{}, hostwrite.ErrUnavailable
	}
	cfg, err := relocationAssemblyConfig(a, providers, s, reg, act, d, in.OptDir, discharge)
	if err != nil {
		return RelocationAssemblyConfig{}, err
	}
	cfg.ReleasePub = in.Pub
	cfg.Ratchet = pointerRatchet(false, cfg.OptDir, cfg.DataDir, getenv)
	if raw := getenv("CARLOS_SECRETS_KEY"); raw != "" {
		key, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY")
		if err != nil {
			return RelocationAssemblyConfig{}, err
		}
		cfg.SecretsKey = key
	}
	return cfg, nil
}

// channelRelease is ruling 30: the host's channel pointer in the control
// bucket, by the holder's account and app and the record's channel.
func channelRelease(store release.Store, pub ed25519.PublicKey) func(context.Context, string) (string, error) {
	return func(ctx context.Context, host string) (string, error) {
		obs, err := hostclaim.Read(ctx, store, host)
		if err != nil || obs.Absent || obs.Claim.Holder == nil {
			return "", errors.Join(err, hostrelocate.ErrRefused)
		}
		rec, err := provision.GetRecord(ctx, store, obs.Claim.Holder.Account, obs.Claim.Holder.App, host)
		if err != nil {
			return "", err
		}
		p, err := release.Channel(ctx, store, pub, obs.Claim.Holder.Account, obs.Claim.Holder.App, rec.Channel)
		if err != nil {
			return "", err
		}
		return p.Version, nil
	}
}

// volumeDischarge is ruling 3 with enrolled_at in the roster: unqualified
// without it.
func volumeDischarge(store release.Store) func(context.Context, hostactors.Box, termfence.Record) *hostclaim.LocalDischarge {
	return func(ctx context.Context, b hostactors.Box, f termfence.Record) *hostclaim.LocalDischarge {
		if b.EnrolledAt.IsZero() {
			return &hostclaim.LocalDischarge{Reason: "the roster records no enrollment time"}
		}
		at, err := time.Parse(time.RFC3339Nano, f.ObservedAt)
		if err != nil {
			return &hostclaim.LocalDischarge{Reason: err.Error()}
		}
		records, err := boxvolumes.List(ctx, store, b.Enrollment)
		if err != nil {
			return &hostclaim.LocalDischarge{Reason: err.Error()}
		}
		v := boxvolumes.Discharge(records, boxvolumes.Entry{Box: b.BoxID, Enrollment: b.Enrollment, Cloud: b.Cloud}, b.EnrolledAt, at)
		return &hostclaim.LocalDischarge{Qualified: v.Discharge, Reason: v.Reason}
	}
}

// relocationRelocatable is Worker.Relocatable over relocationRecord (K24):
// only a hibernating, path-deriving instance record that mints nested
// relocates (ruling 45).
func relocationRelocatable(dataDir string, store release.Store) func(context.Context, hostclaim.Observed) error {
	return func(ctx context.Context, obs hostclaim.Observed) error {
		_, err := relocationRecord(ctx, dataDir, store, obs)
		return err
	}
}

// errRelocationNotStarted is the pass's answer to Worker.Start's two
// providers: this runtime replays relocations and never starts one. The
// app admission and the census are the starter's (the operator's carlos
// ops host relocate and spec 3's placer), with their own bindings.
var errRelocationNotStarted = errors.New("host: the relocation pass starts no relocation")

type relocationStartRefused struct{}

func (relocationStartRefused) Register(context.Context, hostclaim.Observed, hostrelocate.Owner, string) (string, error) {
	return "", errRelocationNotStarted
}

func (relocationStartRefused) Recheck(context.Context, string) error { return errRelocationNotStarted }

// relocationRuntimeInputs is what NewRelocationRuntime binds beyond the
// claim's own: the box's enrollment, the release key, the bucket
// constructors (production's S3; tests' fakes).
type relocationRuntimeInputs struct {
	Enrollment hostfloor.Enrollment
	Pub        ed25519.PublicKey
	Files      func(context.Context, string) (release.Store, error)
	Versioned  func(context.Context, string, string) (lease.VersionedStore, error)
	OptDir     string
}

// productionRelocationRuntimeInputs reads the box's enrollment and the
// release key, and opens the data bucket with release.NewS3Store and
// lease.NewVersionedS3.
func productionRelocationRuntimeInputs(getenv func(string) string, optDir string) (relocationRuntimeInputs, error) {
	pub, err := release.ParsePublicKey(getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
	if err != nil {
		return relocationRuntimeInputs{}, err
	}
	enrollment, err := hostfloor.ReadEnrollment("/etc/carlos/host-actor.json")
	if err != nil {
		return relocationRuntimeInputs{}, err
	}
	return relocationRuntimeInputs{Enrollment: enrollment, Pub: pub, OptDir: optDir,
		Files: func(ctx context.Context, bucket string) (release.Store, error) {
			s, err := release.NewS3Store(ctx, bucket)
			if err != nil {
				return nil, err
			}
			return s, nil
		},
		Versioned: func(ctx context.Context, bucket, region string) (lease.VersionedStore, error) {
			return lease.NewVersionedS3(ctx, map[string]string{bucket: region})
		}}, nil
}

// NewRelocationRuntime builds the agent's relocation runtime from the bound
// claim assembly. It returns (nil, nil) unless CARLOS_HOST_CLAIM is on.
// With CARLOS_REGION_HA off, or the relocation owners not bound into the
// wake's dispatcher (ruling 53: they need moves' C8 provider), it is the
// completion-only runtime (ruling 82): a relocation that already completed
// on this box is still finished. Only with both on and the owners bound is
// it the full pass.
func NewRelocationRuntime(ctx context.Context, getenv func(string) string, store release.Store, local *productionClaimLocal, providers ClaimProviders, reg *registry.Registry, act RelocationActivator, in relocationRuntimeInputs) (*RelocationRuntime, error) {
	if !hostwrite.Enabled(getenv) || local == nil {
		return nil, nil
	}
	if !regionHAEnabled(getenv) {
		return newCompletionRuntime(local, reg, act)
	}
	a := local.boundAssembly()
	if a == nil {
		return nil, hostwrite.ErrUnavailable
	}
	if d, ok := wakeDispatch(a); !ok || d.Relocation == nil {
		return newCompletionRuntime(local, reg, act)
	}
	if a.RetirementAuthority == nil || store == nil || reg == nil || act == nil || len(in.Pub) != ed25519.PublicKeySize || in.Enrollment.Box == "" {
		return nil, hostwrite.ErrUnavailable
	}
	d := a.RetirementAuthority.Directory
	roster, err := d.Read(ctx)
	if err != nil {
		return nil, err
	}
	self, ok := roster.Roster.Boxes[in.Enrollment.Box]
	if !ok || self.Enrollment != in.Enrollment.Enrollment {
		return nil, hostbox.ErrChanged
	}
	stores, err := openRelocationStores(ctx, self, getenv("CARLOS_HIBERNATE_BUCKET"), in.Files, in.Versioned)
	if err != nil {
		return nil, err
	}
	cfg, err := relocationRuntimeConfig(getenv, a, providers, stores, reg, act, d, in, local.DischargeCompleted)
	if err != nil {
		return nil, err
	}
	effects := a.RelocationEffects(cfg, local.Acknowledged)
	chooser := hostrelocate.RegionChooser{Records: store, Namespace: self.Namespace, Region: stores.Region,
		Executors: hostops.ExecutorChooser{Directory: d, Namespace: self.Namespace, Pool: self.Pool, HomeName: stores.Region, Now: time.Now,
			Edges: func(ctx context.Context) (boxchoice.Snapshot, error) {
				return edgesfile.LiveSnapshot(ctx, store, time.Now())
			},
			Loads: func(ctx context.Context) (map[string]boxchoice.Load, error) { return hostops.ClaimLoads(ctx, store) },
			// The fingerprint this box's own edges file publishes: a target
			// must hold the same secrets key to render the held config.
			SecretsKeyFP: edgesFileSecretsFP(getenv)}}
	worker := hostrelocate.Worker{Claims: store, Directory: d,
		Fences:    hostfence.ClaimFenceAuthority{Directory: d, Now: time.Now},
		Releases:  StopReceiptReleases{Receipts: RelocationReleaseReceipts{Store: store, Directory: d}},
		Admission: relocationStartRefused{},
		Census:    func(context.Context, string, []string) error { return errRelocationNotStarted },
		Choose:    chooser.Choose, Eligible: chooser.Eligible,
		Relocatable: relocationRelocatable(cfg.DataDir, store),
		Leases:      relocationLeases(stores),
		Pins: hostrelocate.StorePinner{Claims: store, Data: stores.Files, Bucket: stores.Bucket, Region: stores.Region, HomeRegion: cfg.Source.HomeRegion,
			SelectDB: func(ctx context.Context, r genselect.Request) (*genselect.Plan, error) {
				return genselect.Select(ctx, stores.Versions, r)
			},
			Release: channelRelease(store, in.Pub)},
		Prepared:  PreparedFromInstallManifest{Claims: store, Files: stores.Files, Data: stores.Versions, Bucket: stores.Bucket},
		Discharge: volumeDischarge(store),
		Notify:    func(ctx context.Context) { _ = release.BumpSerial(ctx, store, time.Now()) },
		Self:      termfence.WrittenBy{Box: in.Enrollment.Box, Enrollment: in.Enrollment.Enrollment},
		Region:    stores.Region, HomeRegion: cfg.Source.HomeRegion, Controller: newClaimID(), NewID: newClaimID, Now: time.Now}
	participant := hostrelocate.Participant{Claims: store, Box: in.Enrollment.Box, Enrollment: in.Enrollment.Enrollment, Effects: effects, Now: time.Now}
	return &RelocationRuntime{Worker: worker, Participant: participant, Box: in.Enrollment.Box, Convert: convertFencedLanding(worker, store), Finish: effects.Finish, DataDir: cfg.DataDir, config: cfg}, nil
}

// newCompletionRuntime is ruling 82: the smallest runtime that finishes a
// relocation already completed on this box (discharge, hold release,
// owned-fence lift, mark; relocationReleaseHold) and does nothing else. It
// steps no relocation, converts no landing and starts nothing, and it
// reads only the claims of this box's own registry rows, so a box that
// never completed a relocation reads its own hosts' claims and writes
// nothing.
func newCompletionRuntime(local *productionClaimLocal, reg *registry.Registry, act RelocationActivator) (*RelocationRuntime, error) {
	a := local.boundAssembly()
	if a == nil || reg == nil || act == nil || a.Guard.DataDir == "" {
		return nil, hostwrite.ErrUnavailable
	}
	cfg := RelocationAssemblyConfig{Registry: reg, Activator: act, DataDir: a.Guard.DataDir, Discharge: local.DischargeCompleted, Now: time.Now}
	dataDir := a.Guard.DataDir
	hosts := func() ([]string, error) {
		rows, err := reg.List()
		if err != nil {
			return nil, err
		}
		seen := make(map[string]bool, len(rows))
		out := make([]string, 0, len(rows))
		for _, row := range rows {
			if !seen[row.Host] {
				seen[row.Host] = true
				out = append(out, row.Host)
			}
		}
		named, err := relocationJournalHosts(dataDir)
		for _, host := range named {
			if !seen[host] {
				seen[host] = true
				out = append(out, host)
			}
		}
		return out, err
	}
	// Carry 8: the completion-only runtime has no roster read, so this box
	// is the bound assembly's own verified stop identity, asked each pass.
	self := func(ctx context.Context) (string, error) {
		l, err := a.localStopIdentity(ctx)
		if err != nil {
			return "", err
		}
		return l.Actor.Box, nil
	}
	return &RelocationRuntime{Finish: a.relocationReleaseHold(cfg), Hosts: hosts, Self: self, DataDir: a.Guard.DataDir}, nil
}

// wakeDispatch is the wake's owner dispatcher, as Build bound it.
func wakeDispatch(a *ClaimAssembly) (ClaimOwnerDispatch, bool) {
	if a == nil || a.Wake == nil {
		return ClaimOwnerDispatch{}, false
	}
	d, ok := a.Wake.Inputs.(ClaimOwnerDispatch)
	return d, ok
}

// Run is one pass every 30 seconds, each bounded at two minutes, and one
// more whenever serial is signalled. The agent gives no serial channel
// (launchRelocationPass): the control serial's one wake channel is the
// regional loop's (agent.go, serialClaims), so the pass runs on its ticker
// alone and a BumpSerial wakes it no sooner than the next tick.
func (r *RelocationRuntime) Run(ctx context.Context, store release.Store, serial <-chan struct{}) {
	tick := time.NewTicker(30 * time.Second)
	defer tick.Stop()
	for {
		pass, cancel := context.WithTimeout(ctx, 2*time.Minute)
		r.once(pass, store)
		cancel()
		select {
		case <-ctx.Done():
			return
		case <-tick.C:
		case <-serial:
		}
	}
}

// once is one pass. Per host, in this order, so a passed probe completes
// in one pass (ruling 49): the participant (its probe stop), the worker
// (the serving CAS), then the completion this box has not yet finished,
// found from the claim's record (ruling 80). A landing whose target is
// fenced is converted when its move committed (J3). A claim is read once
// a pass, and once more only after a step that may have written it.
func (r *RelocationRuntime) once(ctx context.Context, store release.Store) {
	r.tryUpgrade(ctx)
	hosts, err := r.hosts(ctx, store)
	r.result("", "list hosts", err)
	if err != nil && hosts == nil {
		return
	}
	var self passSelf
	read := make(map[string]bool, len(hosts))
	for _, host := range hosts {
		read[host] = true
		obs, err := hostclaim.Read(ctx, store, host)
		// A claim that cannot be read is not "nothing to do": said once,
		// and read again next pass.
		if r.result(host, "read", err); err != nil || obs.Absent {
			continue
		}
		stepped := false
		switch obs.Claim.Op.Kind {
		case hostclaim.KindLanding:
			if r.Convert != nil {
				stepped = true
				r.result(host, "convert", r.Convert(ctx, store, obs))
			}
		case hostclaim.KindRelocating:
			if r.Participant != nil {
				stepped = true
				_, err := r.Participant.Step(ctx, host)
				r.result(host, "participant", notYetIsNoError(err))
			}
			if r.Worker != nil {
				stepped = true
				_, err := r.Worker.Step(ctx, host)
				r.result(host, "worker", notYetIsNoError(err))
			}
		}
		if stepped {
			// A step may have just made the serving CAS (ruling 49).
			obs, err = hostclaim.Read(ctx, store, host)
			if r.result(host, "read", err); err != nil {
				continue
			}
		}
		r.finishCompletion(ctx, host, obs, &self)
	}
	r.alerts.keep(read)
	// Ruling 61: journals of operations that are over go, once per pass,
	// after the pass's own steps and never beside a journal write (carry 3:
	// sweepRelocationJournals holds relocationJournals exclusively).
	if r.DataDir != "" {
		r.result("", "sweep journals", sweepRelocationJournals(ctx, r.DataDir, store))
	}
}

func notYetIsNoError(err error) error {
	if errors.Is(err, hostrelocate.ErrNotYet) {
		return nil
	}
	return err
}

// tryUpgrade replaces a completion-only stand-in with the full pass once
// it builds (upgrade). A build that answers no relocation owners ends the
// tries: completions are then all this box runs, as NewRelocationRuntime
// answers when the owners are not bound.
func (r *RelocationRuntime) tryUpgrade(ctx context.Context) {
	if r.upgrade == nil {
		return
	}
	full, err := r.upgrade(ctx)
	if r.result("", "region HA: relocation pass not built, finishing completions only", err); err != nil {
		return
	}
	r.upgrade = nil
	if full == nil || full.Worker == nil {
		log.Printf("agent: region HA: the relocation owners are not bound, so this box finishes completions only")
		return
	}
	r.Worker, r.Participant, r.Box, r.Convert, r.Finish, r.Hosts, r.Self, r.DataDir, r.config = full.Worker, full.Participant, full.Box, full.Convert, full.Finish, full.Hosts, full.Self, full.DataDir, full.config
	log.Printf("agent: region HA: relocation pass started")
}

// hosts is every claim in the bucket, or Hosts when it is set.
func (r *RelocationRuntime) hosts(ctx context.Context, store release.Store) ([]string, error) {
	if r.Hosts != nil {
		return r.Hosts()
	}
	keys, err := store.List(ctx, "control/hosts/")
	if err != nil {
		return nil, err
	}
	var out []string
	for _, key := range keys {
		host, ok := strings.CutSuffix(strings.TrimPrefix(key, "control/hosts/"), ".json")
		if ok && !strings.Contains(host, "/") {
			out = append(out, host)
		}
	}
	return out, nil
}

// passSelf is this box as one pass knows it: asked at most once.
type passSelf struct {
	asked bool
	box   string
}

// thisBox is Box, or Self's answer, asked once a pass and only by a
// pending completion (review M2). No answer is no box.
func (r *RelocationRuntime) thisBox(ctx context.Context, s *passSelf) string {
	if r.Box != "" {
		return r.Box
	}
	if !s.asked {
		s.asked = true
		err := errors.New("no identity to answer which box this is")
		if r.Self != nil {
			s.box, err = r.Self(ctx)
		}
		if r.result("", "this box", err); err != nil {
			s.box = ""
		}
	}
	return s.box
}

// finishCompletion is ruling 80: a claim serving on this box whose
// completed relocation names this box and is not yet marked discharged is
// handed to Finish (the hold release, relocationReleaseHold) on every
// pass, from the pass's latest read of it, so a serving CAS made this pass
// is finished in the same pass. The work is found from the claim's record,
// never from the row: a row already released, a lift that never ran or a
// mark whose CAS failed is retried until the mark is written.
//
// A claim with no undischarged completion asks nothing more, not even
// which box this is (review M2). A claim that does not serve on this box is
// skipped before Finish is called (carry 8). A record that cannot be read
// on a claim serving here is said and retried, never read as "no
// completion" (carry 1), and so is every refusal of Finish, a probe-fence
// owner it cannot find (carry 6) and an ErrChanged after the serving-here
// check (review M4) among them: the record stays unmarked.
func (r *RelocationRuntime) finishCompletion(ctx context.Context, host string, obs hostclaim.Observed, self *passSelf) {
	if r.Finish == nil || obs.Absent || obs.Claim == nil || obs.Claim.Op.Kind != hostclaim.KindServing {
		return
	}
	c := obs.Claim
	done, ok, err := hostclaim.ReadCompletedRelocation(c)
	if err == nil && (!ok || done.Discharged) {
		r.alerts.clear(host, "finish")
		return
	}
	box := r.thisBox(ctx, self)
	if box == "" || c.Op.Exec != box {
		return
	}
	if err != nil {
		r.alert(host, "finish", err)
		return
	}
	if done.Stop.Box != box {
		return
	}
	r.result(host, "finish", r.Finish(ctx, host, done.Op))
}

// convertFencedLanding is J3 through the worker's own authorization
// (hostrelocate.Worker.ConvertLanding): a landing whose op.exec box has a
// termination fence becomes a failover relocation when its move committed
// (CommittedLandingRecords). An uncommitted one is moves' case and answers
// nil (ErrWrongState); any other refusal, a caller that is not an active
// worker of the landing's region included, is returned for the pass to
// alert once.
func convertFencedLanding(w hostrelocate.Worker, store release.Store) func(context.Context, release.Store, hostclaim.Observed) error {
	return func(ctx context.Context, _ release.Store, obs hostclaim.Observed) error {
		_, err := w.ConvertLanding(ctx, obs, CommittedLandingRecords{Records: MovesRecordsFrom(store)})
		if errors.Is(err, hostclaim.ErrWrongState) {
			return nil
		}
		return err
	}
}

func (r *RelocationRuntime) alert(host, step string, err error) { r.alerts.say(host, step, err) }

// result says err for host's step, or, with none, forgets the step's last
// cause, so the cause is said again if it returns.
func (r *RelocationRuntime) result(host, step string, err error) {
	if err != nil {
		r.alerts.say(host, step, err)
		return
	}
	r.alerts.clear(host, step)
}

// relocationAlerts is rulings 35 and 36's one alert per host, step and
// cause (review M8): a cause is said when it is first seen for that host
// and step, and again only after that step has succeeded since, or after
// another cause. It holds one entry per host and step failing now; each
// pass forgets the hosts it no longer reads, so it is bounded by them.
type relocationAlerts struct {
	mu   sync.Mutex
	last map[[2]string]string // {host, step} -> the cause last said
}

func (a *relocationAlerts) say(host, step string, err error) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.last == nil {
		a.last = map[[2]string]string{}
	}
	k := [2]string{host, step}
	if cause, said := a.last[k]; said && cause == err.Error() {
		return
	}
	a.last[k] = err.Error()
	if host == "" {
		log.Printf("relocate: %s: %v", step, err)
		return
	}
	log.Printf("relocate %s: %s: %v", host, step, err)
}

func (a *relocationAlerts) clear(host, step string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	delete(a.last, [2]string{host, step})
}

// keep forgets every host's causes but those of hosts, and the box-wide
// ones (host "").
func (a *relocationAlerts) keep(hosts map[string]bool) {
	a.mu.Lock()
	defer a.mu.Unlock()
	for k := range a.last {
		if k[0] != "" && !hosts[k[0]] {
			delete(a.last, k)
		}
	}
}

// size is the number of causes held (tests).
func (a *relocationAlerts) size() int {
	a.mu.Lock()
	defer a.mu.Unlock()
	return len(a.last)
}

// registerRelocationSource registers region HA's owner source with the
// roster directory the claim's own owners read (claimproduction.go). The
// release key and the deployment bucket are read from the claim entry's
// environment, the one the flags and the claim store came from (final
// review minor 5: one reading of the environment).
func registerRelocationSource(claimCtx context.Context, dataDir string) error {
	store, getenv := claimStore(claimCtx), claimEntryEnv(claimCtx)
	pub, err := release.ParsePublicKey(getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
	if err != nil || store == nil {
		return errors.Join(hostwrite.ErrUnavailable, err)
	}
	return RegisterProductionClaimRelocationSource(&RelocationSource{DataDir: dataDir,
		Directory: hostactors.Directory{Store: store, PublicKey: pub, Bucket: getenv("CARLOS_DEPLOYMENT_BUCKET")},
		FilesStore: func(ctx context.Context, bucket string) (release.Store, error) {
			s, err := release.NewS3Store(ctx, bucket)
			if err != nil {
				return nil, err
			}
			return s, nil
		}})
}

// startRelocationPass builds the relocation runtime from the claim entry's
// assembly and starts its loop. With claim mode on it always starts one:
// the completion-only pass (ruling 82), which with region HA on builds
// the full pass on its first, bounded, pass (review M9) and becomes it
// once the owners are bound. Nothing with claim mode off; with it on,
// every way of not starting the pass is said.
func startRelocationPass(agentCtx, claimCtx context.Context, claims *ClaimWiring, reg *registry.Registry, svc *activator.Service) {
	getenv := claimEntryEnv(claimCtx)
	if !hostwrite.Enabled(getenv) {
		return
	}
	entry, _ := claimCtx.Value(claimEntryKey{}).(claimEntry)
	// Claim mode is on, so a completion recorded on this box must be
	// finished here (ruling 82).
	if claims == nil || svc == nil || entry.Assembly == nil {
		log.Printf("agent: relocation completion pass not started: claim mode is on but the claim wiring, the activator or the claim assembly is missing")
		return
	}
	local, ok := entry.Assembly.Local.(*productionClaimLocal)
	if !ok {
		log.Printf("agent: relocation completion pass not started: the assembly has no production participation owner")
		return
	}
	var full func(context.Context) (*RelocationRuntime, error)
	ha := regionHAEnabled(getenv)
	if ha {
		full = func(ctx context.Context) (*RelocationRuntime, error) {
			in, err := productionRelocationRuntimeInputs(getenv, release.OptDir())
			if err != nil {
				return nil, err
			}
			return NewRelocationRuntime(ctx, getenv, entry.Store, local, claims.Providers, reg, svc, in)
		}
	}
	runtime, err := relocationPassFor(local, reg, svc, full)
	if err != nil {
		log.Printf("agent: relocation completion pass not started: %v", err)
		return
	}
	launchRelocationPass(agentCtx, runtime, entry.Store, ha)
}

// relocationPassFor is the runtime the agent runs in claim mode: the
// completion-only one (ruling 82), standing in for the full pass while
// full (region HA on) builds it. Nothing is built here: the build reads
// the roster and the data bucket, so it runs inside the first pass, under
// the pass's bound, never on the agent's startup path (review M9). A build
// that fails is not "no relocation on this box": each pass tries it again.
func relocationPassFor(local *productionClaimLocal, reg *registry.Registry, act RelocationActivator, full func(context.Context) (*RelocationRuntime, error)) (*RelocationRuntime, error) {
	r, err := newCompletionRuntime(local, reg, act)
	if err != nil {
		return nil, err
	}
	r.upgrade = full
	return r, nil
}

// launchRelocationPass says the pass started and runs it. Everything it
// says is decided before the loop starts: the loop owns the runtime's
// fields from then on, and a pass may replace them (tryUpgrade; review
// M3). The channel closes when the loop has returned.
func launchRelocationPass(ctx context.Context, r *RelocationRuntime, store release.Store, regionHA bool) <-chan struct{} {
	log.Printf("agent: relocation pass started (completions only: %v)", !regionHA)
	done := make(chan struct{})
	go func() {
		defer close(done)
		r.Run(ctx, store, nil)
	}()
	return done
}