package console

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

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

// retireClaimHostRecord runs one delete as an exclusive "deleting"
// operation. The order is fixed:
//
//  1. Begin (or resume) the operation and drain its registrations.
//  2. Under the operation's own lifecycle grant, and with no effect in
//     flight, CAS the host claim from serving to deleting. The production
//     authority refuses that CAS while any effect of the app is unsettled,
//     so it never runs inside an effect.
//  3. Delete the record and park the claim inside one record effect: the
//     typed record delete needs that effect's permit.
//  4. Free the claim once parking has no obligations left, release the
//     wake and ownership markers (one effect) and reopen (FinishDelete).
//
// From step 2 on the delete intent is durable and owned by this
// operation: a step that is only waiting (parking, a box acknowledgement)
// answers claimRetireAccepted, and recovery or a retry carries it on.
func (s *Server) retireClaimHostRecord(ctx context.Context, ref provision.RetireRef) error {
	c := s.cfg.HostClaimOps
	if c == nil || c.Control == nil || s.claimOwner == (hostops.RegistrationOwner{}) {
		return provision.ErrClaimRetirer
	}
	acct, err := s.store.AccountBySqid(ref.Account)
	if err != nil {
		return err
	}
	if acct == nil {
		return hostops.ErrIdentity
	}
	appID, err := s.store.ClaimAppID(acct.ID, ref.App)
	if err != nil {
		return err
	}
	admission, err := hostops.ReadAdmission(ctx, s.cfg.Releases, ref.Account, ref.App, appID)
	if err != nil {
		return err
	}
	var h *hostops.OperationHandle
	if op := admission.Value.Op; op == nil {
		key, err := claimRetireRecordKey(ref)
		if err != nil {
			return err
		}
		obs, err := hostclaim.Read(ctx, s.cfg.Releases, ref.Host)
		if err != nil {
			return err
		}
		if obs.Absent {
			// Nothing holds the host: "not there" is reached when no record
			// exists either. A record without a claim waits for backfill.
			if _, _, gerr := s.cfg.Releases.Get(ctx, key); errors.Is(gerr, os.ErrNotExist) {
				return nil
			} else if gerr != nil {
				return gerr
			}
			return hostops.ErrPending
		}
		holder := obs.Claim.Holder
		if holder == nil {
			if obs.Claim.Free() {
				return nil
			}
			return hostops.ErrPending
		}
		if !claimRetireHolderIs(holder, ref, key) {
			return hostops.ErrIdentity
		}
		id, err := claimEffectID()
		if err != nil {
			return err
		}
		begin := hostops.AppOperation{ID: id, Kind: hostops.OpDeleting, AppID: appID, Phase: "draining", Intent: map[string]string{"host": ref.Host, "incarnation": holder.Incarnation, "acquisition": holder.Acquisition, "family": ref.Family, "name": ref.Name}}
		if h, err = c.Control.BeginExclusive(ctx, ref.Account, ref.App, appID, begin, s.claimOwner); err != nil {
			return err
		}
		if admission, err = hostops.ReadAdmission(ctx, s.cfg.Releases, ref.Account, ref.App, appID); err != nil {
			return err
		}
	} else {
		// Only this host's own delete is resumed. Any other exclusive
		// operation (a foreign mutation, another host's delete, a purge or a
		// placement) refuses here, before any effect.
		if op.Kind != hostops.OpDeleting || op.Intent["host"] != ref.Host || (ref.Incarnation != "" && op.Intent["incarnation"] != ref.Incarnation) {
			return hostops.ErrPending
		}
		if h, err = c.Control.ResumeExclusive(ctx, ref.Account, ref.App, appID, op.ID, hostops.OpDeleting, s.claimOwner); err != nil {
			return err
		}
	}
	op := admission.Value.Op
	if op == nil || op.ID != h.ID() || op.Intent["host"] != ref.Host || (ref.Incarnation != "" && op.Intent["incarnation"] != ref.Incarnation) {
		return hostops.ErrPending
	}
	ref.Incarnation = op.Intent["incarnation"]
	if ref.Family == "" {
		ref.Family, ref.Name = op.Intent["family"], op.Intent["name"]
	}
	key, err := claimRetireRecordKey(ref)
	if err != nil {
		return err
	}
	// Every claim write below carries this operation's lifecycle grant.
	ctx = h.Lifecycle(ctx)
	// The operation is durable from here: a step that is only waiting
	// answers claimRetireAccepted with this operation's ID and phase.
	if err = h.Drain(ctx, c.Now(), c.RecoverCreate); err != nil {
		return claimRetireWaiting(h, "draining", ref.Host, err)
	}
	if err = s.advanceClaimRetire(ctx, c, h, ref, key); err != nil {
		return err
	}
	// The claim is free: the incarnation is retired, and its move log
	// outbox rows go with it (Plan 6 Task 3; the drainer's drop of a row
	// whose claim is gone is the backstop).
	if err = s.store.PurgeMoveLogIncarnation(ref.Host, ref.Incarnation); err != nil {
		return claimRetireWaiting(h, "releasing", ref.Host, err)
	}
	if err = s.releaseClaimRetireMarkers(ctx, h, ref, appID, op.Intent["acquisition"]); err != nil {
		return claimRetireWaiting(h, "releasing", ref.Host, err)
	}
	return claimRetireWaiting(h, "releasing", ref.Host, h.FinishDelete(ctx))
}

// advanceClaimRetire moves the host claim from wherever this operation
// left it to free. nil means free; claimRetireAccepted means the owned
// delete is waiting on boxes or parking.
func (s *Server) advanceClaimRetire(ctx context.Context, c *hostops.Controller, h *hostops.OperationHandle, ref provision.RetireRef, key string) error {
	runner := h.Runner(time.Now)
	owned := false
	// Each pass makes at most one transition, so four passes cover
	// serving, deleting, parking and free.
	for range 4 {
		obs, err := hostclaim.Read(ctx, s.cfg.Releases, ref.Host)
		if err != nil {
			return err
		}
		if obs.Absent {
			return hostops.ErrPending
		}
		if obs.Claim.Free() {
			return nil
		}
		holder := obs.Claim.Holder
		if holder == nil || !claimRetireHolderIs(holder, ref, key) || holder.Incarnation != ref.Incarnation {
			return hostops.ErrIdentity
		}
		claimOp := obs.Claim.Op
		switch claimOp.Kind {
		case hostclaim.KindServing:
			if owned {
				return hostops.ErrPending
			}
			// Step 2, the durable delete intent. No effect is issued here.
			if _, err = c.BeginRetire(ctx, key, ref.By, ref.Incarnation); err != nil {
				return claimRetireWaiting(h, "starting", ref.Host, err)
			}
		case hostclaim.KindDeleting:
			owned = true
			refreshed, err := s.refreshClaimRetireRecord(ctx, c, obs, key)
			if err != nil {
				return claimRetireWaiting(h, claimOp.Kind, ref.Host, err)
			}
			if refreshed {
				// Re-read the re-bound claim before deleting.
				continue
			}
			effectID, err := claimEffectID()
			if err != nil {
				return err
			}
			// Step 3: the typed record delete runs under this effect's permit.
			_, err = runner.Run(ctx, hostops.Effect{ID: effectID, Resource: "record:" + ref.Host, Action: "retire"}, func(ctx context.Context) (string, error) {
				rerr := c.Retire(ctx, ref.Host, claimOp.ID)
				if rerr == nil {
					return "retired", nil
				}
				if errors.Is(rerr, hostops.ErrPending) {
					fresh, ferr := hostclaim.Read(ctx, s.cfg.Releases, ref.Host)
					if ferr == nil && !fresh.Absent && fresh.Claim.Op.ID == claimOp.ID && fresh.Claim.Op.Kind == hostclaim.KindParking {
						// Retire answers pending after a successful park.
						return "parked", nil
					}
					// The record was not deleted (boxes not ready yet).
					return "", errors.Join(hostops.ErrNotApplied, rerr)
				}
				return "", rerr
			})
			if err != nil {
				return claimRetireWaiting(h, claimOp.Kind, ref.Host, err)
			}
		case hostclaim.KindParking:
			owned = true
			if len(obs.Claim.Obligations) != 0 {
				return &claimRetireAccepted{operation: h.ID(), phase: hostclaim.KindParking, host: ref.Host}
			}
			// A claim-only transition under the lifecycle grant.
			if err = c.Free(ctx, ref.Host, claimOp.ID); err != nil {
				return claimRetireWaiting(h, claimOp.Kind, ref.Host, err)
			}
		default:
			// Purging, placing and moving claims belong to their own operations.
			return hostops.ErrPending
		}
	}
	return &claimRetireAccepted{operation: h.ID(), phase: hostclaim.KindDeleting, host: ref.Host}
}

// refreshClaimRetireRecord re-binds the deleting claim to a record whose
// ETag moved since the claim was written. It runs outside any effect: the
// production refresh refuses while an effect is unsettled.
func (s *Server) refreshClaimRetireRecord(ctx context.Context, c *hostops.Controller, obs hostclaim.Observed, key string) (bool, error) {
	rec, err := hostops.ReadRecord(ctx, s.cfg.Releases, key)
	if errors.Is(err, os.ErrNotExist) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	if rec.ETag == obs.Claim.Op.Intent["etag"] {
		return false, nil
	}
	refresher, ok := c.Authority.(interface {
		RefreshDeleteRecord(context.Context, hostclaim.Observed, hostops.Record) error
	})
	if !ok {
		return true, hostops.ErrPending
	}
	return true, refresher.RefreshDeleteRecord(ctx, obs, rec)
}

// releaseClaimRetireMarkers removes the sidecar wake object and the
// ownership marker, both bound to this incarnation, as one effect. It runs
// only after the claim is free.
func (s *Server) releaseClaimRetireMarkers(ctx context.Context, h *hostops.OperationHandle, ref provision.RetireRef, appID, acquisition string) error {
	cleanupID, err := claimEffectID()
	if err != nil {
		return err
	}
	_, err = h.Runner(time.Now).Run(ctx, hostops.Effect{ID: cleanupID, Resource: "ownership:" + ref.Host, Action: "cleanup"}, func(ctx context.Context) (string, error) {
		if ref.Family == "sidecar" {
			key := release.SidecarWakeKey(ref.Account, ref.App, ref.Name)
			raw, etag, err := s.cfg.Releases.Get(ctx, key)
			if err != nil && !errors.Is(err, os.ErrNotExist) {
				return "", err
			}
			if err == nil {
				var wake struct {
					AppID       string `json:"app_id"`
					Incarnation string `json:"incarnation"`
					RequestID   string `json:"request_id"`
				}
				if json.Unmarshal(raw, &wake) != nil {
					return "", hostops.ErrIdentity
				}
				if (wake.AppID != "" || wake.Incarnation != "" || wake.RequestID != "") && (wake.AppID != appID || wake.Incarnation != ref.Incarnation || wake.RequestID == "") {
					return "", hostops.ErrIdentity
				}
				if err = release.DeleteIfMatch(ctx, s.cfg.Releases, key, etag); err != nil {
					return "", err
				}
			}
		}
		key := domains.ClaimKey(ref.Host)
		raw, etag, err := s.cfg.Releases.Get(ctx, key)
		if err != nil && !errors.Is(err, os.ErrNotExist) {
			return "", err
		}
		if err == nil {
			var owner domains.Claim
			if json.Unmarshal(raw, &owner) != nil {
				return "", hostops.ErrIdentity
			}
			if owner.Account != ref.Account || owner.App != ref.App || owner.Acquisition != acquisition || owner.RecordIncarnation != ref.Incarnation {
				return "", hostops.ErrIdentity
			}
			if err = release.DeleteIfMatch(ctx, s.cfg.Releases, key, etag); err != nil {
				return "", err
			}
		}
		return "joined", nil
	})
	return err
}

func (s *Server) retireInstanceRecord(ctx context.Context, account, app, host, by, incarnation string) error {
	return s.retireHostRecord(ctx, provision.RetireRef{Family: "instance", Account: account, App: app, Host: host, By: by, Incarnation: incarnation})
}

func mustCreateIncarnation(account, app, host string, created time.Time) string {
	inc, err := hostclaim.Incarnation(account, app, host, created)
	if err != nil {
		return "invalid-incarnation"
	}
	return inc
}

func (s *Server) deleteLegacySidecarWake(ctx context.Context, account, app, name string) error {
	if s.cfg.HostClaim {
		return nil
	}
	return s.cfg.Releases.Delete(ctx, release.SidecarWakeKey(account, app, name))
}