package hostops

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"strconv"
	"sync/atomic"
	"time"

	"github.com/carlosframework/platform/internal/release"
)

type issuedMutationKey struct{}

// MarkAppMutationIssued is called immediately before an app effect reaches
// a raw sink. It never grants authority; it records whether failure may be
// ambiguous so a validation refusal can release an unissued operation.
func MarkAppMutationIssued(ctx context.Context) {
	// Every enclosing operation is marked, not only the innermost: a nested
	// scope's raw write is also a possible late effect of each outer operation,
	// so none of them may release as unapplied.
	marks, _ := ctx.Value(issuedMutationKey{}).([]*atomic.Bool)
	for _, p := range marks {
		if p != nil {
			p.Store(true)
		}
	}
}

// ObserveAppMutationEffects records raw effects in this scope and keeps all
// enclosing observers. It grants no authority and does not settle anything.
func ObserveAppMutationEffects(ctx context.Context) (context.Context, func() bool) {
	issued := new(atomic.Bool)
	outer, _ := ctx.Value(issuedMutationKey{}).([]*atomic.Bool)
	marks := append(append([]*atomic.Bool(nil), outer...), issued)
	return context.WithValue(ctx, issuedMutationKey{}, marks), issued.Load
}

// BindMutationRunner runs once at bootstrap, before publishing the boundary.
func BindMutationRunner(s release.Store, w *ControlWriter, owner RegistrationOwner, localAppID func(context.Context, string, string) (string, error)) error {
	b, ok := s.(*mutationBoundary)
	if !ok || w == nil || !validRegistrationOwner(owner) || b.control != nil || localAppID == nil {
		return ErrContract
	}
	b.localAppID = localAppID
	b.control = w
	b.owner = owner
	return nil
}

// unissuedRecoveryTimeout bounds the issuer's own uncancelled attempt to end
// an operation whose effect it never issued.
const unissuedRecoveryTimeout = 20 * time.Second

func (b *mutationBoundary) RunAppMutation(ctx context.Context, account, app, action string, f func(context.Context) error) error {
	if b.control == nil || b.localAppID == nil || f == nil || action == "" {
		return ErrContract
	}
	// Resolve the requesting console's identity before reading shared state.
	// A current bucket name is not authority to adopt a recreated app.
	appID, err := b.localAppID(ctx, account, app)
	if err != nil {
		return err
	}
	if appID == "" {
		return ErrIdentity
	}
	a, err := ReadAdmission(ctx, b.raw, account, app, appID)
	if err != nil {
		return err
	}
	if RequireAppMutation(ctx, b.raw, account, app, a.Value.AppID) == nil {
		return f(ctx)
	}
	// A callback carrying retired authority cannot acquire a successor's
	// incarnation merely by retrying through a high-level adapter.
	inherited, _ := ctx.Value(appMutationPermitKey{}).([]*AppMutationPermit)
	for _, permit := range inherited {
		if permit != nil && permit.account == account && permit.app == app {
			return ErrPending
		}
	}

	var nonce [24]byte
	if _, err = rand.Read(nonce[:]); err != nil {
		return err
	}
	id := hex.EncodeToString(nonce[:])
	owner := b.owner
	owner.Controller = id
	intent := map[string]string{"action": action, MutationActionStarted: strconv.FormatInt(b.control.now().Unix(), 10)}
	h, err := b.control.BeginExclusive(ctx, account, app, a.Value.AppID, AppOperation{ID: id, Kind: OpAppMutation, AppID: a.Value.AppID, Phase: "issued", Intent: intent}, owner)
	if err != nil {
		if !RefusedBeforeEffect(err) {
			// The begin CAS may have landed with its response lost. Nothing
			// was issued under this fresh ID, so end it if it exists.
			// Its own refusal is the normal outcome when the CAS never landed,
			// so the caller sees only the begin error.
			_ = recoverUnissued(ctx, err, func(rctx context.Context) error {
				return b.control.abandonUnissuedMutation(rctx, account, app, a.Value.AppID, id, 0)
			})
		}
		return err
	}
	pctx, settle, err := h.Runner(time.Now).Issue(ctx, Effect{ID: id, Resource: OpAppMutation, Action: action})
	if err != nil {
		// f was never called. Absent or merely prepared effects are
		// superseded and the app reopens with a new epoch; an issued one (a
		// lost response on the issued CAS) keeps the operation closed.
		return recoverUnissued(ctx, err, h.abandonUnissued)
	}
	issued := new(atomic.Bool)
	outer, _ := ctx.Value(issuedMutationKey{}).([]*atomic.Bool)
	marks := append(append([]*atomic.Bool(nil), outer...), issued)
	pctx = context.WithValue(pctx, issuedMutationKey{}, marks)
	callErr := f(pctx)
	settleErr := callErr
	if callErr != nil && !issued.Load() {
		settleErr = errors.Join(ErrNotApplied, callErr)
	}
	if err = settle("joined", settleErr); err != nil && !errors.Is(err, ErrNotApplied) {
		return errors.Join(callErr, err)
	}
	settled, cancel := context.WithTimeout(context.WithoutCancel(ctx), unissuedRecoveryTimeout)
	defer cancel()
	if err = h.FinishMutation(settled); err != nil {
		return errors.Join(callErr, err)
	}
	return callErr
}

// recoverUnissued runs one bounded, uncancelled recovery attempt after cause
// and returns cause with its classification intact: a recovery failure is
// reported as text only, so it cannot turn, say, context.Canceled into
// ErrPending for the caller. The sweep (RecoverMutation) retries later.
func recoverUnissued(ctx context.Context, cause error, attempt func(context.Context) error) error {
	rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), unissuedRecoveryTimeout)
	defer cancel()
	if err := attempt(rctx); err != nil {
		return fmt.Errorf("%w (unissued operation left for recovery: %v)", cause, err)
	}
	return cause
}