package console

import (
	"context"
	"encoding/json"
	"errors"
	"os"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/provision"
)

type ClaimPlacementCoordinator struct {
	Server     *Server
	Controller hostops.Controller
	Authority  *hostops.ProductionAuthority
	Owner      hostops.RegistrationOwner
}

type metadataPlacementHost struct {
	Host        string `json:"host"`
	Incarnation string `json:"incarnation"`
	Executor    string `json:"executor"`
	Stamp       uint64 `json:"stamp"`
	RecordETag  string `json:"record_etag"`
	ClaimETag   string `json:"claim_etag"`
	ClaimGen    uint64 `json:"claim_gen"`
}

func (p *ClaimPlacementCoordinator) ready() error {
	if p.Authority == nil || p.Controller.Control == nil || p.Owner == (hostops.RegistrationOwner{}) {
		return hostops.ErrContract
	}
	return nil
}

func (p *ClaimPlacementCoordinator) Start(ctx context.Context, account, app, target, by string) (string, error) {
	if err := p.ready(); err != nil {
		return "", err
	}
	acct, err := p.Server.store.AccountBySqid(account)
	if err != nil {
		return "", err
	}
	if acct == nil {
		return "", hostops.ErrIdentity
	}
	appID, err := p.Server.store.ClaimAppID(acct.ID, app)
	if err != nil {
		return "", err
	}
	a, err := hostops.ReadAdmission(ctx, p.Controller.Store, account, app, appID)
	if err != nil {
		return "", err
	}
	if a.Value.Op != nil {
		if a.Value.Op.Kind != hostops.OpPlacing || a.Value.Op.Intent["target"] != target {
			return "", hostops.ErrPending
		}
		return a.Value.Op.ID, p.Resume(ctx, account, app, appID, a.Value.Op.ID)
	}
	source, err := p.Server.store.AppPlacement(acct.ID, app)
	if err != nil {
		return "", err
	}
	id := p.Controller.NewID()
	op := hostops.AppOperation{ID: id, Kind: hostops.OpPlacing, AppID: appID, Phase: "draining", Intent: map[string]string{"source": source, "target": target, "by": by}}
	if _, err = p.Controller.Control.BeginExclusive(ctx, account, app, appID, op, p.Owner); err != nil {
		return "", err
	}
	// Resume binds the SQL operation fence as its first effect.
	return id, p.Resume(ctx, account, app, appID, id)
}

// sqlEffect runs one app SQL write as an effect of the placing operation;
// the SQL gate admits it on that effect's permit.
func (p *ClaimPlacementCoordinator) sqlEffect(ctx context.Context, h *hostops.OperationHandle, accountID int64, action string, fn func(*Store) error) error {
	id, err := claimEffectID()
	if err != nil {
		return err
	}
	scope := claimSQLScope{AccountID: accountID, App: h.App(), AppID: h.AppID()}
	var localErr error
	_, err = h.Runner(p.Controller.Now).Run(ctx, hostops.Effect{ID: id, Resource: "app-sql:" + action, Action: action}, func(ctx context.Context) (string, error) {
		// SQL has joined before it returns. Preserve its error for the caller,
		// but do not strand a remote effect for a local, replayable projection.
		localErr = p.Server.store.WithAppMutation(ctx, []sqlGrant{permitGrant(ctx, p.Controller.Store, h.Account(), scope)}, fn)
		return "joined", nil
	})
	return errors.Join(localErr, err)
}

func (p *ClaimPlacementCoordinator) Resume(ctx context.Context, account, app, appID, id string) error {
	if err := p.ready(); err != nil {
		return err
	}
	h, err := p.Controller.Control.ResumeExclusive(ctx, account, app, appID, id, hostops.OpPlacing, p.Owner)
	if err != nil {
		return err
	}
	ctx = h.Lifecycle(ctx)
	a, err := hostops.ReadAdmission(ctx, p.Controller.Store, account, app, appID)
	if err != nil {
		return err
	}
	acct, err := p.Server.store.AccountBySqid(account)
	if err != nil {
		return err
	}
	if acct == nil {
		return hostops.ErrIdentity
	}
	op := a.Value.Op
	if op == nil || op.ID != id {
		return hostops.ErrPending
	}
	if op.Phase != "releasing" {
		revision := a.Value.Revision
		if err = p.sqlEffect(ctx, h, acct.ID, "bind-operation", func(st *Store) error {
			return st.BindClaimAppOperation(acct.ID, app, appID, id, revision)
		}); err != nil {
			return err
		}
	}
	var hosts []metadataPlacementHost
	switch op.Phase {
	case "draining":
		if err = h.Drain(ctx, p.Controller.Now(), p.Controller.RecoverCreate); err != nil {
			return err
		}
		// The census is keyed by this operation's reservation: every roster
		// box publishes its report for it. Until each one has, the owned
		// operation is waiting, not failed (an empty app included: a box
		// may still serve a stale installation of it).
		census, err := p.Authority.Census(ctx, account, app, id)
		if errors.Is(err, os.ErrNotExist) {
			return errors.Join(hostops.ErrPending, err)
		}
		if err != nil {
			return err
		}
		byHost := map[string]metadataPlacementHost{}
		for _, record := range census.Records {
			if record.Holder.Family != "instance" {
				return p.cancelUnsupported(ctx, h, acct.ID)
			}
			wire, err := provision.DecodeRecord(record.Body)
			if err != nil {
				return err
			}
			if len(wire.Replicas) != 0 || wire.Region != op.Intent["target"] {
				return p.cancelUnsupported(ctx, h, acct.ID)
			}
			obs, err := hostclaim.Read(ctx, p.Controller.Store, record.Holder.Host)
			if err != nil {
				return err
			}
			incarnation, err := record.Incarnation()
			if err != nil {
				return err
			}
			if obs.Absent || obs.Claim.Holder == nil || obs.Claim.Holder.Incarnation != incarnation || obs.Claim.Op.Kind != hostclaim.KindServing || !obs.Claim.Touched.Complete {
				return hostops.ErrPending
			}
			if len(obs.Claim.Op.Participants) != 1 || obs.Claim.Op.Exec != obs.Claim.Op.Participants[0] {
				return p.cancelUnsupported(ctx, h, acct.ID)
			}
			byHost[obs.Host] = metadataPlacementHost{Host: obs.Host, Incarnation: incarnation, Executor: obs.Claim.Op.Exec, Stamp: obs.Claim.Holder.Stamp, RecordETag: record.ETag, ClaimETag: obs.ETag, ClaimGen: obs.Claim.Gen}
		}
		seen := map[string]bool{}
		for _, report := range census.Reports {
			for _, installation := range report.Installations {
				if installation.Account != account || installation.App != app {
					continue
				}
				ph, ok := byHost[installation.Host]
				if !ok || installation.Incarnation != ph.Incarnation || installation.Stamp != ph.Stamp || report.Box != ph.Executor || installation.Placement != op.Intent["target"] || seen[ph.Host] {
					return p.cancelUnsupported(ctx, h, acct.ID)
				}
				seen[ph.Host] = true
			}
		}
		for host, ph := range byHost {
			if !seen[host] {
				return hostops.ErrPending
			}
			hosts = append(hosts, ph)
		}
		raw, err := json.Marshal(hosts)
		if err != nil {
			return err
		}
		return h.SetPhase(ctx, "draining", "claimed", map[string]string{"hosts": string(raw)})
	case "claimed":
		if err = json.Unmarshal([]byte(op.Intent["hosts"]), &hosts); err != nil {
			return err
		}
		for _, ph := range hosts {
			obs, err := hostclaim.Read(ctx, p.Controller.Store, ph.Host)
			if err != nil {
				return err
			}
			if obs.Absent || obs.Claim.Holder == nil || obs.Claim.Holder.Incarnation != ph.Incarnation || obs.Claim.Holder.Stamp != ph.Stamp || obs.Claim.Op.Exec != ph.Executor {
				return hostops.ErrPending
			}
			if obs.Claim.Op.Kind == hostclaim.KindPlacing && obs.Claim.Op.ID == id {
				continue
			}
			if obs.ETag != ph.ClaimETag || obs.Claim.Gen != ph.ClaimGen {
				// A cancelled unused hold invalidated the persisted observation:
				// re-save it for this same incarnation only.
				if obs.Claim.Op.Kind != hostclaim.KindServing {
					return hostops.ErrPending
				}
				for i := range hosts {
					if hosts[i].Host == ph.Host {
						hosts[i].ClaimETag, hosts[i].ClaimGen = obs.ETag, obs.Claim.Gen
					}
				}
				raw, err := json.Marshal(hosts)
				if err != nil {
					return err
				}
				return h.SetPhase(ctx, "claimed", "claimed", map[string]string{"hosts": string(raw)})
			}
			if _, err = p.Authority.StartPlacementClaim(ctx, obs, account, app, appID, id, op.Intent["by"], op.Intent["target"]); err != nil {
				return err
			}
		}
		for _, ph := range hosts {
			obs, err := hostclaim.Read(ctx, p.Controller.Store, ph.Host)
			if err != nil {
				return err
			}
			if obs.Absent || obs.Claim.Op.Kind != hostclaim.KindPlacing || obs.Claim.Op.ID != id {
				return hostops.ErrPending
			}
			if err = p.Authority.FirstPhase(ctx, obs); err != nil {
				return err
			}
			record, err := hostops.ReadRecord(ctx, p.Controller.Store, obs.Claim.Holder.Key)
			if err != nil {
				return err
			}
			if record.ETag != ph.RecordETag {
				return hostops.ErrPending
			}
			wire, err := provision.DecodeRecord(record.Body)
			if err != nil {
				return err
			}
			if wire.Region != op.Intent["target"] || len(wire.Replicas) != 0 {
				return hostops.ErrPending
			}
		}
		// The app-level projection correction is an app SQL write: it runs
		// as an effect so the SQL gate admits it on that permit.
		target := op.Intent["target"]
		if err = p.sqlEffect(ctx, h, acct.ID, "placement", func(st *Store) error {
			return st.SetClaimAppPlacement(acct.ID, app, appID, id, target)
		}); err != nil {
			return err
		}
		return h.SetPhase(ctx, "claimed", "releasing", nil)
	case "releasing":
		if err = json.Unmarshal([]byte(op.Intent["hosts"]), &hosts); err != nil {
			return err
		}
		for _, ph := range hosts {
			obs, err := hostclaim.Read(ctx, p.Controller.Store, ph.Host)
			if err != nil {
				return err
			}
			if obs.Absent || obs.Claim.Holder == nil || obs.Claim.Holder.Incarnation != ph.Incarnation || obs.Claim.Holder.Stamp != ph.Stamp || obs.Claim.Op.Exec != ph.Executor {
				return hostops.ErrPending
			}
			if obs.Claim.Op.Kind == hostclaim.KindServing {
				continue
			}
			if _, err = hostclaim.Update(ctx, p.Controller.Store, obs, hostops.CompleteMetadataPlacement(id, appID)); err != nil {
				return err
			}
		}
		return p.finish(ctx, h, acct.ID)
	}
	return hostops.ErrPending
}

func (p *ClaimPlacementCoordinator) finish(ctx context.Context, h *hostops.OperationHandle, accountID int64) error {
	a, err := hostops.ReadAdmission(ctx, p.Controller.Store, h.Account(), h.App(), h.AppID())
	if err != nil {
		return err
	}
	if a.Value.Op == nil || a.Value.Op.ID != h.ID() {
		return hostops.ErrPending
	}
	// Cancellation during draining has no host claims. Persist releasing
	// before touching SQL, so a crash cannot re-enter the older bind.
	if a.Value.Op.Phase == "draining" {
		if err = h.SetPhase(ctx, "draining", "releasing", map[string]string{"hosts": "[]"}); err != nil {
			return err
		}
	}
	// Advance the SQL fence BEFORE reopening admission. FinishPlacement adds
	// exactly one revision, so old workers cannot bind again afterwards.
	// Advance the SQL fence BEFORE reopening. Inside this effect prepare and
	// issue have already written; its settle and FinishPlacement add one CAS
	// each, so the reopened revision is the current one plus two. Stale
	// workers carry lower revisions and cannot bind again.
	if err = p.sqlEffect(ctx, h, accountID, "release-fence", func(st *Store) error {
		fresh, err := hostops.ReadAdmission(ctx, p.Controller.Store, h.Account(), h.App(), h.AppID())
		if err != nil {
			return err
		}
		return st.BindClaimAppOperation(accountID, h.App(), h.AppID(), "", fresh.Value.Revision+2)
	}); err != nil {
		return err
	}
	return h.FinishPlacement(ctx)
}

// cancelUnsupported is reached only during draining, before any host claim
// or projection changed.
func (p *ClaimPlacementCoordinator) cancelUnsupported(ctx context.Context, h *hostops.OperationHandle, accountID int64) error {
	if err := p.finish(ctx, h, accountID); err != nil {
		return err
	}
	return errors.Join(hostops.ErrContract, errors.New("console: placement shape unsupported"))
}