package host

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
)

// The installer's park and lease adapters (moves spec §6.4, §7.6), shared by
// the landing and the dirty return: the park engine over this box's own
// bucket for the residue parks, with the claim's view and authorizer
// (MoveParks, armMovesParks), the generation classifier the view shares, and
// the own-location lease the build holds from checking through published.
// The phases are in movesinstall.go, the landing's pass in movesland.go.
//
// Nothing stands in for the claim's view or authorizers (Plan 5 Task 9a,
// Q10): an arm without ParkView and ParkAuthorize is not armed, nor one that
// runs a park.Local without ParkLocalAuthorize, and the tests inject fakes.
// Each arm keeps its own operation check in front of the claim's authorizer
// (M17): movesArmAuthorize for the engine's chores, movesLocalAuthorize for
// park.Local's (M9).

// movesArmAuthorize is an arm's park authorizer (M17): the arm's own
// operation check, then the claim's authorizer, ParkAuthorize from the
// claim's MoveParks. A missing claim authorizer refuses; nothing stands in.
func movesArmAuthorize(claim, own func(context.Context, hostclaim.Observed, hostclaim.Obligation) error) func(context.Context, hostclaim.Observed, hostclaim.Obligation) error {
	return func(ctx context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
		if claim == nil || own == nil {
			return fmt.Errorf("%w: no park authorizer", park.ErrUnsafe)
		}
		if err := own(ctx, obs, o); err != nil {
			return err
		}
		return claim(ctx, obs, o)
	}
}

// movesLocalAuthorize is an arm's park.Local authorizer (M9): the arm's own
// operation check, then the claim's MoveParks local authorizer
// (ParkLocalAuthorize). A missing claim authorizer refuses; the arm's own
// rule never stands alone.
func movesLocalAuthorize(claim, own func(context.Context, *hostclaim.Claim, hostclaim.Obligation) error) func(context.Context, *hostclaim.Claim, hostclaim.Obligation) error {
	return func(ctx context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
		if claim == nil || own == nil {
			return fmt.Errorf("%w: no local park authorizer", park.ErrUnsafe)
		}
		if err := own(ctx, c, o); err != nil {
			return err
		}
		return claim(ctx, c, o)
	}
}

// authorizeMoveChore is an install's own operation check: a chore is run
// only when it is the op's own and the claim still names the op, at the
// install's kind, with this box executing.
func authorizeMoveChore(c *hostclaim.Claim, o hostclaim.Obligation, opID, kind, box string) error {
	if c == nil || o.Op != opID || c.Op.ID != opID || c.Op.Kind != kind || c.Op.Exec != box {
		return fmt.Errorf("%w: not this install's chore", park.ErrUnsafe)
	}
	return nil
}

// authorizeChore is the install's own check for its local chores, which
// park.Local's authorizer runs before the claim's (movesLocalAuthorize, M9).
func (ir *installRun) authorizeChore(_ context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
	return authorizeMoveChore(c, o, ir.g.ID, ir.src.kind(), ir.cfg().Box)
}

// authorizeObligation is the install's own operation check over an
// observation, which the engine's authorizer runs before the claim's.
func (ir *installRun) authorizeObligation(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
	if obs.Absent {
		return park.ErrUnsafe
	}
	return authorizeMoveChore(obs.Claim, o, ir.g.ID, ir.src.kind(), ir.cfg().Box)
}

// authorizeObligation is the landing's own operation check, the installer's
// rule for a landing.
func (l *landingRun) authorizeObligation(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
	if obs.Absent {
		return park.ErrUnsafe
	}
	return authorizeMoveChore(obs.Claim, o, l.g.ID, hostclaim.KindLanding, l.cfg().Box)
}

// landingMarkerGeneration is what landingGenerationOf answers for the
// location's generation.json. The marker is the one key every generation
// reuses (§5.1): the probe's admission supersedes an earlier stay's marker
// with If-Match rather than deleting it, and a park-generation chore whose
// deletion set named the marker key could never be acknowledged once a new
// marker stands there (the engine skips the marker key and then refuses a
// recorded delete at it). So the marker belongs to no generation's chore:
// parking never queues one for it and the engine passes it over. "@" keeps
// it apart from any generation name, as hostclaim.RootGeneration does.
const landingMarkerGeneration = "@marker"

// landingGenerationOf classifies an object under a holder location by
// generation, as the view does: `g-<gen>/...` and `files/g-<gen>/...` are
// gen's; a level directory (four hex digits) at the location is the
// origin, ""; generation.json is landingMarkerGeneration; anything else is
// unknown, and a caller fails closed on it. The naming rule is
// moves.GenerationPrefix's twin.
func landingGenerationOf(location, key string) (string, bool) {
	rel, ok := strings.CutPrefix(key, location+"/")
	if !ok || rel == "" {
		return "", false
	}
	for _, p := range []string{"g-", "files/g-"} {
		if rest, ok := strings.CutPrefix(rel, p); ok {
			if gen, _, ok := strings.Cut(rest, "/"); ok && gen != "" {
				return gen, true
			}
			return "", false
		}
	}
	if rel == "generation.json" {
		return landingMarkerGeneration, true
	}
	if first, _, ok := strings.Cut(rel, "/"); ok && len(first) == 4 && isHexString(first) {
		return "", true
	}
	return "", false
}

// movesHolderBuildState reports whether key is the build state beside the
// lease at the live holder location (release.BuildStateKey; ruling
// R-PARK): the marker's twin, live state that belongs to no generation.
// A caller passes it over exactly where it passes over the marker at that
// location, before classifying (landingGenerationOf stays strict). The
// same file name anywhere else (a retired location, a moves lane, under a
// generation) is not it. TestMovesHolderBuildStateMatchesTheEngine keeps
// it in step with the park engine.
func movesHolderBuildState(location, key string) bool {
	return key == release.BuildStateKey(strings.TrimSuffix(location, "/"))
}

func isHexString(s string) bool {
	for _, c := range s {
		if (c < '0' || c > '9') && (c < 'a' || c > 'f') && (c < 'A' || c > 'F') {
			return false
		}
	}
	return s != ""
}

// engine is the park engine over this box's own bucket with the claim's
// leases and view, and the claim's authorizer behind this install's own
// check (the arming requires all three).
func (ir *installRun) engine(vs lease.VersionedStore) park.Engine {
	cfg := ir.cfg()
	return park.Engine{HomeRegion: cfg.ParkHomeRegion, Store: vs, Claims: ir.s, Bucket: cfg.OwnBucket, Region: cfg.parkRegion(), Leases: cfg.ParkLeases,
		Now: cfg.Now, NewID: obligationID, View: cfg.ParkView, Authorize: movesArmAuthorize(cfg.ParkAuthorize, ir.authorizeObligation), After: cfg.parkFault}
}

// lease is the own-location lease the build holds (§7.6 checking: "hold it
// through published"), kept in the worker's per-host map across passes and
// re-acquired after a restart. waiting is true when another holder has it.
func (ir *installRun) lease(ctx context.Context, holder hostclaim.Holder) (held park.HeldLease, waiting bool, err error) {
	return acquireOwnLease(ctx, ir.w, ir.rec.Host, holder)
}

// acquireOwnLease takes host's own-location lease in this box's own bucket,
// or answers the one w already holds while it still checks. A landing takes
// it at checking (spec §7.6) and keeps it through published; its residue
// parks run under it (ParkRunHeld, M5). waiting is true when another holder
// has it.
func acquireOwnLease(ctx context.Context, w *movesWorker, host string, holder hostclaim.Holder) (held park.HeldLease, waiting bool, err error) {
	cfg := w.cfg
	w.mu.Lock()
	held = w.landLeases[host]
	w.mu.Unlock()
	if held != nil {
		if err := held.Check(ctx); err == nil {
			return held, false, nil
		}
		releaseOwnLease(w, host)
	}
	held, err = cfg.OwnLease(ctx, cfg.OwnBucket, strings.TrimSuffix(holder.Location, "/"))
	if err != nil {
		var he *lease.HeldError
		if errors.As(err, &he) {
			log.Printf("moves: %s: another holder has the location's lease; the build waits", host)
			return nil, true, nil
		}
		return nil, false, fmt.Errorf("the location's lease: %w", err)
	}
	w.mu.Lock()
	if w.landLeases == nil {
		w.landLeases = map[string]park.HeldLease{}
	}
	w.landLeases[host] = held
	w.mu.Unlock()
	return held, false, nil
}

// releaseLease releases the host's own-location lease, if held.
func (ir *installRun) releaseLease() { releaseOwnLease(ir.w, ir.rec.Host) }

// releaseOwnLease releases host's own-location lease on w, if held. The
// landing's pass settles it at its end (settleLease); a return's caller
// does the same once its install call returns; a rollback that abandons
// the landing releases it when the decision takes the record and again on
// every entry to the target's half (rollbackTarget).
func releaseOwnLease(w *movesWorker, host string) {
	w.mu.Lock()
	held := w.landLeases[host]
	delete(w.landLeases, host)
	w.mu.Unlock()
	if held == nil {
		return
	}
	rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := held.Release(rctx); err != nil {
		log.Printf("moves: %s: release the location's lease: %v", host, err)
	}
}

// settleLease releases the lease once the record is past published, marked
// for a rollback, or no longer landing: the lease is the build's alone.
func (l *landingRun) settleLease() {
	m := l.rec.Move
	if m == nil || m.State != provision.MoveLanding || m.Reason != "" || !landingBefore(m.LandingPhase, provision.LandLocal) {
		releaseOwnLease(l.w, l.rec.Host)
	}
}

// leaseContext derives the context every generation request runs under from
// the lease's own (fix round 2): hostbox's renewingHeld, the production
// park.HeldLease, cancels it when the lease is lost or released, so a
// request in flight, and the S3 client's retries of it, stop with the lease
// rather than landing after it. A lease without a Context method (a plain
// fake) leaves ctx as it is; leasedStore's check before each request still
// stands.
func leaseContext(ctx context.Context, held park.HeldLease) context.Context {
	if bound, ok := held.(interface {
		Context(context.Context) context.Context
	}); ok {
		return bound.Context(ctx)
	}
	return ctx
}

// leasedStore is the own bucket by version under live lease authority: check
// (the held lease's Check) runs before every request, so a generation write
// after the lease was lost is refused before it is sent, as the cut's
// guarded staging writes are (fix round 1, item 3); the requests themselves
// run under leaseContext, so one already in flight when the lease is lost is
// cancelled with its retries (fix round 2). The build's reads run under both
// too: a holder that has lost the lease should not read ahead either.
type leasedStore struct {
	lease.VersionedStore
	check func(context.Context) error
}

func (s leasedStore) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
	if err := s.check(ctx); err != nil {
		return nil, fmt.Errorf("the location's lease: %w", err)
	}
	return s.VersionedStore.ListVersions(ctx, bucket, prefix)
}

func (s leasedStore) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
	if err := s.check(ctx); err != nil {
		return nil, fmt.Errorf("the location's lease: %w", err)
	}
	return s.VersionedStore.OpenVersion(ctx, bucket, key, versionID)
}

func (s leasedStore) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) {
	if err := s.check(ctx); err != nil {
		return nil, fmt.Errorf("the location's lease: %w", err)
	}
	return s.VersionedStore.GetVersion(ctx, bucket, key, versionID)
}

func (s leasedStore) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
	if err := s.check(ctx); err != nil {
		return "", "", fmt.Errorf("the location's lease: %w", err)
	}
	return s.VersionedStore.CopyVersion(ctx, srcBucket, srcKey, srcVersionID, dstBucket, dstKey, createOnly)
}

func (s leasedStore) DeleteVersion(ctx context.Context, bucket, key, versionID string) error {
	if err := s.check(ctx); err != nil {
		return fmt.Errorf("the location's lease: %w", err)
	}
	return s.VersionedStore.DeleteVersion(ctx, bucket, key, versionID)
}

func (s leasedStore) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
	if err := s.check(ctx); err != nil {
		return "", fmt.Errorf("the location's lease: %w", err)
	}
	return s.VersionedStore.PutIfAbsent(ctx, bucket, key, body)
}