// Package deployment implements the app deployment engine: preflight,
// journaled execution, and terminal-result computation.
package deployment

import (
	"context"
	"encoding/hex"
	"errors"
	"fmt"
	"maps"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/bnema/zerowrap"
	"github.com/google/uuid"

	"github.com/bnema/gordon/internal/boundaries/out"
	"github.com/bnema/gordon/internal/domain"
	"github.com/bnema/gordon/pkg/validation"
)

// Deps are the driven adapters owned by the deployment engine.
type Deps struct {
	State   out.AppState
	Runtime out.ContainerRuntime
	Images  out.ImageResolver
	Secrets out.SecretProvider
	// Registry configures image pulls from the installation registry.
	// Domain scopes which refs pull with auth; empty disables pulling.
	Registry RegistryConfig
	// Networks is the installation network policy applied to every app
	// workload: the prefix used to derive Gordon-owned network names and
	// whether those networks block external egress.
	Networks NetworkConfig
	// Limits are the installation container resource limits applied on
	// every create and recovery path. Zero means unlimited.
	Limits ResourceLimits
	// ImagePolicy restricts which registries and references may be
	// resolved, pulled, or run. It is enforced on every path, including
	// boot, restart, and recovery.
	ImagePolicy domain.ImageSourcePolicy
	// Traffic is the serialized HTTP/L4 publish boundary. Rebuilt after
	// every activation (deploy/start/restart/stop/remove) so the proxy
	// host index follows ACTIVE; recovery also withdraws one service
	// fail-closed before touching a workload. Nil in tests that assert
	// state effects only. Failures are reported to the caller.
	Traffic out.AppTrafficRefresher
}

// RegistryConfig carries the installation registry identity for pulls.
type RegistryConfig struct {
	Domain      string
	PullAddress string
	Username    string
	Password    string
}

// NetworkConfig carries the installation network policy for app workloads.
type NetworkConfig struct {
	// Prefix names the Gordon-owned networks. Empty uses the domain default.
	Prefix string
	// Internal blocks external egress from containers attached to those
	// networks (Docker's Internal flag).
	Internal bool
}

// ResourceLimits are the installation container limits applied on every
// create and recovery path. Zero values mean unlimited.
type ResourceLimits struct {
	MemoryBytes int64
	NanoCPUs    int64
	PidsLimit   int64
}

// Service orchestrates captured-revision deploys.
type Service struct {
	deps   Deps
	log    zerowrap.Logger
	probes *ProbeDeps

	// coord serializes workload mutations per app: deploy/start/
	// restart/stop/remove/boot lock blocking, periodic reconciliation
	// skips busy apps.
	coord *appCoordinator
	// barrier is the process-wide GC barrier. A shared lease covers
	// every workload mutation from resource selection through durable
	// protection publication; prune holds the exclusive lease.
	barrier out.GCBarrier
	// backoff is the in-memory, generation-scoped recovery budget.
	backoff *recoveryBackoff
	// publication holds the in-memory publication inhibition: a
	// service whose fail-closed withdrawal could not be applied must be
	// withdrawn again before any later publication.
	publication publicationInhibition
	// executions remembers the last observed execution start per
	// generation, so a native restart is detected without restarting
	// the workload ourselves.
	executions executionTracker
	// bindMu guards bindPolicies. A config reload replaces the whole
	// map atomically; deploy paths read the current map without holding
	// the lock.
	bindMu       sync.RWMutex
	bindPolicies map[string]domain.AppBindPolicy
	// deviceMu guards devicePolicies. A config reload replaces the whole
	// map atomically; deploy paths read the current map without holding
	// the lock.
	deviceMu       sync.RWMutex
	devicePolicies map[string]domain.AppDevicePolicy
	// liveMu guards liveOps. liveOps holds the operations this process
	// owns between the StartDeploy claim and the end of ExecuteDeploy. It
	// is in-memory only: a fresh Service (boot) starts empty and may
	// reconcile every persisted non-terminal operation, while a
	// foreground mutation must never finalize an operation a live
	// goroutine still owns.
	liveMu  sync.Mutex
	liveOps map[string]struct{}
}

// NewService creates the deployment engine. All deps are required;
// Secrets may be nil only in tests that never reach secret preflight.
func NewService(deps Deps, log zerowrap.Logger) *Service {
	if deps.ImagePolicy.InstallationRegistry == "" {
		deps.ImagePolicy.InstallationRegistry = deps.Registry.Domain
	}
	return &Service{
		deps:    deps,
		log:     log,
		coord:   newAppCoordinator(),
		backoff: newRecoveryBackoff(time.Now),
	}
}

// markOperationLive records that this process owns one operation from the
// StartDeploy claim until ExecuteDeploy finishes. A live operation is the
// single owner's in-flight work: a foreground reconciliation must not
// finalize it, or the claim would be settled before the owner can execute it.
func (s *Service) markOperationLive(app, op string) {
	if app == "" || op == "" {
		return
	}
	s.liveMu.Lock()
	defer s.liveMu.Unlock()
	if s.liveOps == nil {
		s.liveOps = make(map[string]struct{})
	}
	s.liveOps[liveOperationKey(app, op)] = struct{}{}
}

// clearOperationLive drops the live marker. It is called once ExecuteDeploy
// has recorded its outcome or left the claim recoverable, so a later
// reconciliation can converge the operation instead of leaving it in flight.
func (s *Service) clearOperationLive(app, op string) {
	s.liveMu.Lock()
	defer s.liveMu.Unlock()
	delete(s.liveOps, liveOperationKey(app, op))
}

// operationLive reports whether this process still owns the operation.
func (s *Service) operationLive(app, op string) bool {
	s.liveMu.Lock()
	defer s.liveMu.Unlock()
	_, ok := s.liveOps[liveOperationKey(app, op)]
	return ok
}

func liveOperationKey(app, op string) string {
	return app + "\x00" + op
}

// WithGCBarrier wires the process-wide GC barrier. Every workload
// mutation holds a shared lease from resource selection through the
// durable publication of the resource's protection, so prune can never
// observe a half-acquired resource. Nil is valid for tests that never
// run prune.
func (s *Service) WithGCBarrier(barrier out.GCBarrier) *Service {
	s.barrier = barrier
	return s
}

// WithBindPolicies supplies the initial administrative bind policies.
func (s *Service) WithBindPolicies(policies map[string]domain.AppBindPolicy) *Service {
	s.SetBindPolicies(policies)
	return s
}

// SetBindPolicies atomically replaces the administrative bind policies used
// to authorize service binds. The map is copied so later caller mutation
// cannot race an in-flight deploy.
func (s *Service) SetBindPolicies(policies map[string]domain.AppBindPolicy) {
	copied := make(map[string]domain.AppBindPolicy, len(policies))
	for name, policy := range policies {
		copied[name] = policy
	}
	s.bindMu.Lock()
	s.bindPolicies = copied
	s.bindMu.Unlock()
}

// snapshotBindPolicies returns the current policy map. SetBindPolicies never
// mutates a published map, so the snapshot stays safe after the lock drops.
func (s *Service) snapshotBindPolicies() map[string]domain.AppBindPolicy {
	s.bindMu.RLock()
	defer s.bindMu.RUnlock()
	return s.bindPolicies
}

// WithDevicePolicies supplies the initial administrative device policies.
func (s *Service) WithDevicePolicies(policies map[string]domain.AppDevicePolicy) *Service {
	s.SetDevicePolicies(policies)
	return s
}

// SetDevicePolicies atomically replaces the administrative device policies
// used to authorize service devices. The map and its slices are copied so
// later caller mutation cannot race an in-flight deploy.
func (s *Service) SetDevicePolicies(policies map[string]domain.AppDevicePolicy) {
	copied := make(map[string]domain.AppDevicePolicy, len(policies))
	for name, policy := range policies {
		policy.CDI = append([]string(nil), policy.CDI...)
		policy.AllowedApps = append([]string(nil), policy.AllowedApps...)
		policy.AllowedServices = append([]string(nil), policy.AllowedServices...)
		copied[name] = policy
	}
	s.deviceMu.Lock()
	s.devicePolicies = copied
	s.deviceMu.Unlock()
}

// snapshotDevicePolicies returns the current policy map. SetDevicePolicies
// never mutates a published map, so the snapshot stays safe after the
// lock drops.
func (s *Service) snapshotDevicePolicies() map[string]domain.AppDevicePolicy {
	s.deviceMu.RLock()
	defer s.deviceMu.RUnlock()
	return s.devicePolicies
}

// resolveServiceDevices resolves every declared logical device against the
// current administrative policies into runtime CDI IDs. Errors name only
// the app, service, and device: host inventory never appears.
func (s *Service) resolveServiceDevices(app string, svc domain.AppService) ([]string, error) {
	if len(svc.Devices) == 0 {
		return nil, nil
	}
	ids, err := domain.ResolveAppDevices(app, svc.Name, svc.Devices, s.snapshotDevicePolicies())
	if err != nil {
		return nil, fmt.Errorf("deployment: %w", err)
	}
	return ids, nil
}

// resolveServiceBinds resolves every declared bind against the current
// administrative policies into ephemeral runtime binds. Errors name only the
// app, service, and mount: host source paths never appear.
func (s *Service) resolveServiceBinds(app string, svc domain.AppService) ([]domain.ContainerBind, error) {
	if len(svc.Binds) == 0 {
		return nil, nil
	}
	policies := s.snapshotBindPolicies()
	resolved := make([]domain.ContainerBind, 0, len(svc.Binds))
	for _, bind := range svc.Binds {
		policy, ok := policies[bind.Name]
		if !ok {
			return nil, fmt.Errorf("deployment: app %q service %q mount %q: no administrative mount policy configured: %w", app, svc.Name, bind.Name, domain.ErrBindPolicy)
		}
		r, err := policy.ResolveAppBind(app, svc.Name, bind)
		if err != nil {
			return nil, fmt.Errorf("deployment: app %q service %q mount %q: refused by administrative mount policy: %w", app, svc.Name, bind.Name, domain.ErrBindPolicy)
		}
		resolved = append(resolved, domain.ContainerBind(r))
	}
	return resolved, nil
}

// acquireAppContext takes the GC shared lease first and the per-app
// coordinator second, following the documented lock order
// (GC barrier → per-app coordinator → registry lock → bbolt
// transaction). The returned release drops both, coordinator first.
func (s *Service) acquireAppContext(ctx context.Context, app string) (func(), error) {
	lease, err := s.acquireSharedGC(ctx)
	if err != nil {
		return nil, err
	}
	release, err := s.coord.acquire(ctx, app)
	if err != nil {
		lease.Release()
		return nil, err
	}
	return func() {
		release()
		lease.Release()
	}, nil
}

// tryAcquireAppContext takes the GC shared lease, then tries the per-app
// coordinator without blocking. ok is false when the app is busy; the
// caller skips it. Periodic reconciliation still waits for an in-flight
// prune, because a half-planned snapshot is exactly what must not be
// raced.
func (s *Service) tryAcquireAppContext(ctx context.Context, app string) (func(), bool) {
	lease, err := s.acquireSharedGC(ctx)
	if err != nil {
		return nil, false
	}
	release, ok := s.coord.tryAcquire(app)
	if !ok {
		lease.Release()
		return nil, false
	}
	return func() {
		release()
		lease.Release()
	}, true
}

// acquireSharedGC takes a shared GC lease, or a no-op lease when no
// barrier is wired.
func (s *Service) acquireSharedGC(ctx context.Context) (out.GCLease, error) {
	if s.barrier == nil {
		return noopGCLease{}, nil
	}
	return s.barrier.AcquireShared(ctx)
}

// noopGCLease is the lease used when no barrier is wired.
type noopGCLease struct{}

func (noopGCLease) Release() {}

// WithProbeDeps overrides readiness probing (tests inject a mock-backed
// ProbeDeps via NewProbeDeps; production leaves it nil for the live
// runtime adapter).
func (s *Service) WithProbeDeps(probes ProbeDeps) *Service {
	s.probes = &probes
	return s
}

// DeployInput selects the revision and service scope.
type DeployInput struct {
	App      string
	Revision string
	Service  string
	Op       string
}

// DeployResult carries terminal per-service results and cleanup warnings.
type DeployResult struct {
	Op              string
	App             string
	Revision        string
	Outcome         string
	Services        map[string]ServiceResult
	CleanupWarnings []CleanupWarning
	// Removed names services retired because the new revision no longer
	// declares them.
	Removed []string
}

// ServiceResultUnchanged marks a deploy step that kept the running
// container because its image, spec, and environment were already current.
const ServiceResultUnchanged = domain.AppServiceUnchanged

// ServiceResult is one service's terminal deployment result.
type ServiceResult struct {
	Result            string
	EffectiveRevision string
	Before            string
	After             string
	RestartUnsafe     bool
	Error             string
	// BackendBinds carries the created container's TCP loopback publishes
	// (container port -> 127.0.0.1 host port) for the active record.
	BackendBinds map[int]int
	// UDPBackendBinds carries the created container's UDP loopback
	// publishes for the active record. Nil when the service declares
	// no UDP interface.
	UDPBackendBinds map[int]int
	// CleanupWarnings are bounded leftovers of this service's terminal
	// path: a candidate that could not be removed, or a backend claim
	// that could not be released.
	CleanupWarnings []CleanupWarning
	// Diagnostics is bounded, redacted failure output of the failed
	// candidate. It is never embedded in Error and is persisted to the
	// journal for logs-scoped retrieval only.
	Diagnostics []string
}

// CleanupWarning records post-publication leftovers for operator action.
type CleanupWarning struct {
	Service  string
	Leftover string
	Detail   string
}

// pinnedService binds a service spec to its resolved digest plus the
// captured revision's app-wide public env (injected into every service
// alongside that service's own resolved secrets). appEnv is a copy owned
// by this pin: never mutated, never persisted with secret values.
type pinnedService struct {
	name         string
	spec         domain.AppService
	digest       string
	runtimeImage string
	appEnv       map[string]string
	appNetworks  []domain.AppSharedNetwork
	// sharedNetworks are the declared shared-network memberships this
	// service joins. The private incarnation network is derived from the
	// app UUID at create time so recovery and deploy agree.
	sharedNetworks []domain.AppSharedNetwork
}

// newOpID allocates a time-ordered op- identifier (uuid v7).
func newOpID() string {
	id, err := uuid.NewV7()
	if err != nil {
		id = uuid.New()
	}
	var buf [32]byte
	hex.Encode(buf[:], id[:])
	return "op-" + string(buf[:])
}

// claimDeploymentLocked is the first phase of a deploy: it resolves the
// requested revision, enforces the targeted-deploy convergence rules, and
// claims the request key, persisting the non-terminal journal. It performs
// no image pull and no workload mutation, so a crash after it leaves a
// durable claim that reconciliation can converge. owned is true only for the
// single claimer of the key; a replay returns the stored journal instead.
// The caller holds the app lock and has already recovered the store.
func (s *Service) claimDeploymentLocked(ctx context.Context, input DeployInput) (domain.AppDesiredRevision, domain.AppOperation, bool, error) {
	rev, err := s.resolveRevision(ctx, input)
	if err != nil {
		return domain.AppDesiredRevision{}, domain.AppOperation{}, false, err
	}
	if input.Service != "" {
		if err := s.checkConverged(ctx, input.App, input.Service, rev); err != nil {
			return domain.AppDesiredRevision{}, domain.AppOperation{}, false, err
		}
	}

	op := domain.AppOperation{
		Kind:          "deploy",
		App:           input.App,
		InputRevision: rev.Revision,
		StartedAt:     time.Now().UTC(),
		Request:       domain.AppOperationRequestFor("deploy", input.App, input.Revision, input.Service),
	}
	op, owned, err := s.claimOperation(ctx, input.Op, op, []domain.AppOperationStep{{ID: "preflight", State: domain.AppStepPending}})
	if err != nil {
		return domain.AppDesiredRevision{}, domain.AppOperation{}, false, err
	}
	return rev, op, owned, nil
}

// pinPreflightLocked is the second preflight phase: it runs the image
// pull/pin and resource gates of an already claimed operation and records the
// pinned table (or the terminal preflight failure) in the journal. The
// caller holds the app lock and owns the claimed, non-terminal operation.
func (s *Service) pinPreflightLocked(ctx context.Context, app, onlyService string, rev domain.AppDesiredRevision, op *domain.AppOperation) ([]pinnedService, error) {
	log := zerowrap.FromCtx(ctx)

	pinned, err := s.preflightServices(ctx, app, rev, onlyService)
	if err != nil {
		op.Steps[0] = domain.AppOperationStep{ID: "preflight", State: domain.AppStepFailed, Error: err.Error()}
		op.Outcome = domain.AppOutcomeFailed
		if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
			log.Warn().Err(saveErr).Msg("deployment: failed to record preflight failure")
		}
		return nil, err
	}
	op.Steps[0] = domain.AppOperationStep{ID: "preflight", State: domain.AppStepSucceeded}
	// A resumed claim can carry the service-step plan of an earlier attempt,
	// including candidates it recorded. Converge those candidates before
	// replacing the plan: an unreconciled leftover may still be running, and
	// the replacement this execution creates must never overlap it.
	if err := s.convergeResumedCandidates(ctx, app, op, log); err != nil {
		return nil, err
	}
	// Replace the persisted plan instead of appending: the preflight step plus
	// one step per pinned service, so runServiceStep's index contract matches a
	// normalized layout and no earlier step is overwritten or left pending.
	plan := make([]domain.AppOperationStep, 1, len(pinned)+1)
	plan[0] = op.Steps[0]
	for _, p := range pinned {
		plan = append(plan, domain.AppOperationStep{
			ID:      "service." + p.name + ".replace",
			State:   domain.AppStepPending,
			Service: p.name,
			Digest:  p.digest,
			Image:   p.runtimeImage,
		})
	}
	op.Steps = plan
	if err := s.deps.State.SaveOperation(ctx, *op); err != nil {
		return nil, fmt.Errorf("deployment: persist pinned table: %w", err)
	}
	log.Info().Str("op", op.Op).Str("revision", rev.Revision).Int("services", len(pinned)).Msg("deployment: preflight passed")
	return pinned, nil
}

// convergeResumedCandidates converges any candidate an earlier attempt of a
// resumed claim recorded, so replacing its service-step plan cannot drop a
// leftover that may still be running. A fresh claim has only its preflight
// step and does no work here.
func (s *Service) convergeResumedCandidates(ctx context.Context, app string, op *domain.AppOperation, log zerowrap.Logger) error {
	needsActive := false
	for _, step := range op.Steps {
		if reconcileNeedsContainer(step, false) {
			needsActive = true
			break
		}
	}
	if !needsActive {
		return nil
	}
	active, _, err := s.deps.State.LoadActive(ctx, app)
	if err != nil {
		return fmt.Errorf("deployment: load active for resumed claim: %w", err)
	}
	if _, err := s.convergeInterruptedSteps(ctx, app, op, active, false, log); err != nil {
		return err
	}
	return nil
}

// claimOperation is the single claim point before any mutation effect.
// A keyed request is claimed atomically: the first claim persists the
// in-flight journal and reports owned=true; a repeat returns the stored
// journal and reports owned=false. A key that already answered a
// different request fails with ErrAppStateConflict, and a key for a name
// with no app identity fails with ErrAppNotFound without writing
// anything. An unkeyed request (internal recovery paths) starts a fresh
// journal under a generated id.
func (s *Service) claimOperation(ctx context.Context, key string, op domain.AppOperation, steps []domain.AppOperationStep) (domain.AppOperation, bool, error) {
	op.Steps = steps
	// A journal is never opened while a different operation of this app is
	// still unfinished: that predecessor must be reconciled first, or this
	// claim would mask it and its leftover generation could stay untracked.
	latest, found, err := s.deps.State.LoadLatestOperation(ctx, op.App)
	if err != nil {
		return domain.AppOperation{}, false, fmt.Errorf("deployment: load latest operation: %w", err)
	}
	if found && !latest.Terminal() && latest.Op != key {
		return domain.AppOperation{}, false, fmt.Errorf(
			"deployment: app %q has unfinished operation %s: %w", op.App, latest.Op, domain.ErrAppStateConflict)
	}
	if key == "" {
		op.Op = newOpID()
		if err := s.deps.State.SaveOperation(ctx, op); err != nil {
			return domain.AppOperation{}, false, fmt.Errorf("deployment: persist journal before effects: %w", err)
		}
		return op, true, nil
	}
	op.Op = key
	existing, claimed, err := s.deps.State.ClaimOperation(ctx, op)
	if err != nil {
		return domain.AppOperation{}, false, err
	}
	if !claimed {
		return existing, false, nil
	}
	return op, true, nil
}

// replayError reports a replayed key whose journal never reached a
// terminal outcome: the operation is still in flight or was interrupted,
// so its effects must not run again.
func replayError(op domain.AppOperation) error {
	if !op.Terminal() {
		return fmt.Errorf("deployment: operation %s has not reached a terminal outcome: %w", op.Op, domain.ErrAppStateConflict)
	}
	if op.Outcome != domain.AppOutcomeSuccess {
		return fmt.Errorf("deployment: operation %s previously ended with outcome %s: %w", op.Op, op.Outcome, domain.ErrAppStateConflict)
	}
	return nil
}

// resolveRevision loads the captured revision (default: current desired).
// A revision that cannot be resolved on a name with no app identity at
// all reports ErrAppNotFound: the failure is the missing app, not the
// missing revision.
func (s *Service) resolveRevision(ctx context.Context, input DeployInput) (domain.AppDesiredRevision, error) {
	if input.Revision != "" {
		rev, err := s.deps.State.LoadRevision(ctx, input.App, input.Revision)
		if err != nil && errors.Is(err, domain.ErrAppRevisionNotFound) {
			if knownErr := s.requireKnownApp(ctx, input.App); knownErr != nil {
				return domain.AppDesiredRevision{}, knownErr
			}
		}
		return rev, err
	}
	rev, ok, err := s.deps.State.LoadDesired(ctx, input.App)
	if err != nil {
		return domain.AppDesiredRevision{}, err
	}
	if !ok {
		if knownErr := s.requireKnownApp(ctx, input.App); knownErr != nil {
			return domain.AppDesiredRevision{}, knownErr
		}
		return domain.AppDesiredRevision{}, fmt.Errorf("deployment: app %q has no desired state: %w", input.App, domain.ErrAppRevisionNotFound)
	}
	return rev, nil
}

// requireKnownApp refuses a mutation of a name with no live app identity.
func (s *Service) requireKnownApp(ctx context.Context, app string) error {
	exists, err := s.deps.State.AppExists(ctx, app)
	if err != nil {
		return fmt.Errorf("deployment: app existence: %w", err)
	}
	if !exists {
		return fmt.Errorf("deployment: app %q does not exist: %w", app, domain.ErrAppNotFound)
	}
	return nil
}

// checkConverged permits a targeted deploy when every desired/effective
// difference belongs to that service. App-wide and other-service changes
// must be deployed together so ACTIVE never combines incompatible specs.
func (s *Service) checkConverged(ctx context.Context, app, service string, rev domain.AppDesiredRevision) error {
	active, ok, err := s.deps.State.LoadActive(ctx, app)
	if err != nil {
		return err
	}
	if !ok {
		return fmt.Errorf("deployment: app %q was never deployed, targeted deploy refused: %w", app, domain.ErrAppStateConflict)
	}
	if active.Converged && active.ConvergedRevision == rev.Revision {
		return nil
	}
	diff := domain.DiffAppSpec(rev.Spec, effectiveAppSpec(active))
	prefix := "service/" + service + "/"
	if len(diff.Added) > 0 {
		return targetedDivergenceError(rev.Revision, diff.Added[0]+" (added)")
	}
	if len(diff.Removed) > 0 {
		return targetedDivergenceError(rev.Revision, diff.Removed[0]+" (removed)")
	}
	for _, path := range diff.Changed {
		if !strings.HasPrefix(path, prefix) {
			return targetedDivergenceError(rev.Revision, path)
		}
	}
	return nil
}

func effectiveAppSpec(active domain.AppActive) domain.AppSpec {
	spec := domain.AppSpec{Name: active.App, Networks: append([]domain.AppSharedNetwork(nil), active.Networks...)}
	for name, service := range active.Services {
		effective := service.Spec
		effective.Name = name
		spec.Services = append(spec.Services, effective)
	}
	return spec
}

func targetedDivergenceError(revision, path string) error {
	return fmt.Errorf("%w: service-targeted deploy refused, desired %s also changes %s", domain.ErrAppStateConflict, revision, path)
}

// selectPreflightServices copies and optionally narrows a revision's services.
func selectPreflightServices(rev domain.AppDesiredRevision, onlyService string) ([]domain.AppService, error) {
	services := append([]domain.AppService(nil), rev.Spec.Services...)
	if onlyService == "" {
		return services, nil
	}
	for _, svc := range services {
		if svc.Name == onlyService {
			return []domain.AppService{svc}, nil
		}
	}
	return nil, fmt.Errorf("deployment: service %q not in revision %s: %w", onlyService, rev.Revision, domain.ErrAppStateConflict)
}

// preflightServices runs the preflight gates in order. No mutation.
func (s *Service) preflightServices(ctx context.Context, app string, rev domain.AppDesiredRevision, onlyService string) ([]pinnedService, error) {
	services, err := selectPreflightServices(rev, onlyService)
	if err != nil {
		return nil, err
	}
	sort.Slice(services, func(i, j int) bool { return services[i].Name < services[j].Name })

	var pinned []pinnedService
	ownership, err := s.deps.State.LoadOwnership(ctx, app)
	if err != nil {
		return nil, err
	}
	// Defensive re-check: stored revisions predate or bypass apply
	// validation; an overlap must fail preflight before any mutation.
	if err := rev.Spec.CheckEnvSecretCollisions(); err != nil {
		return nil, err
	}
	appEnv := maps.Clone(rev.Spec.Env)
	if appEnv == nil {
		appEnv = map[string]string{}
	}
	for _, svc := range services {
		digest, err := s.deps.Images.ResolveDigest(ctx, svc.Image)
		if err != nil {
			return nil, fmt.Errorf("deployment: image %q unresolvable: %w", svc.Image, domain.ErrAppImageUnresolvable)
		}
		if err := s.checkSecrets(ctx, app, ownership.ID, svc); err != nil {
			return nil, err
		}
		runtimeImage, err := s.preflightImage(ctx, svc.Image, digest)
		if err != nil {
			return nil, err
		}
		if err := s.checkImageVolumes(ctx, svc, runtimeImage); err != nil {
			return nil, err
		}
		// Bind resolution stays a preflight gate: an unresolvable bind
		// must fail before any workload mutation. The successful values
		// are discarded because createAndStart re-resolves against the
		// current policy immediately before the runtime mutation.
		if _, err := s.resolveServiceBinds(app, svc); err != nil {
			return nil, err
		}
		if err := s.preflightServiceDevices(ctx, app, svc, pinned); err != nil {
			return nil, err
		}
		pinned = append(pinned, pinnedService{
			name: svc.Name, spec: svc, digest: digest, runtimeImage: runtimeImage, appEnv: appEnv,
			appNetworks:    append([]domain.AppSharedNetwork(nil), rev.Spec.Networks...),
			sharedNetworks: domain.AppServiceSharedNetworks(rev.Spec, svc.Name),
		})
	}
	if err := s.recheckReservations(ctx, app, rev); err != nil {
		return nil, err
	}
	if err := s.checkResources(ctx, app, ownership, services); err != nil {
		return nil, err
	}
	return pinned, nil
}

// preflightServiceDevices runs the device authorization and engine
// capability gates for one service. Split from preflightServices to keep
// its complexity within budget.
func (s *Service) preflightServiceDevices(ctx context.Context, app string, svc domain.AppService, pinned []pinnedService) error {
	if _, err := s.resolveServiceDevices(app, svc); err != nil {
		return err
	}
	// Engine capability is a knowable-before-mutation failure: a
	// device-bearing revision on an unsupported engine must fail here,
	// before withdrawal retires the serving generation. The check runs
	// once per revision, on the first device-bearing service;
	// CreateContainer keeps the same gate as defense.
	if len(svc.Devices) > 0 && !devicesEngineChecked(pinned) {
		if s.deps.Runtime == nil {
			return fmt.Errorf("deployment: runtime unavailable: %w", domain.ErrRuntimeUnsupported)
		}
		// Sanitize the adapter cause: version-probe failures may embed
		// the daemon endpoint, which must not reach journals or API
		// responses. Cancellation and deadlines are the exception: they
		// stay recognizable so aborted deploys are not misread as
		// incapable engines.
		if err := s.deps.Runtime.SupportsCDIDevices(ctx); err != nil {
			return sanitizedEngineProbeError(svc.Name, err)
		}
	}
	return nil
}

// sanitizedEngineProbeError maps an engine device-capability probe failure
// onto the caller-visible error. Cancellation and deadlines survive so
// shutdown and timeout handling keep working; every other cause collapses
// to ErrRuntimeUnsupported alone, dropping the adapter text that may embed
// the daemon endpoint.
func sanitizedEngineProbeError(service string, err error) error {
	switch {
	case errors.Is(err, context.Canceled):
		return fmt.Errorf("deployment: engine device capability check for service %q: %w", service, context.Canceled)
	case errors.Is(err, context.DeadlineExceeded):
		return fmt.Errorf("deployment: engine device capability check for service %q: %w", service, context.DeadlineExceeded)
	default:
		return fmt.Errorf("deployment: engine device capability check for service %q: %w", service, domain.ErrRuntimeUnsupported)
	}
}

// devicesEngineChecked reports whether an earlier pinned service already
// triggered the engine capability probe for this revision.
func devicesEngineChecked(pinned []pinnedService) bool {
	for _, p := range pinned {
		if len(p.spec.Devices) > 0 {
			return true
		}
	}
	return false
}

// checkSecrets reads every required secret path once. Values stay in
// memory for container creation only; nothing is written anywhere.
// Paths are UUID-keyed so a removed app's secrets are never adopted
// by a new app reusing the name.
func (s *Service) checkSecrets(ctx context.Context, app, appID string, svc domain.AppService) error {
	if s.deps.Secrets == nil {
		return fmt.Errorf("deployment: secret provider unavailable: %w", domain.ErrAppSecretMissing)
	}
	for envKey, name := range svc.Secrets {
		path := domain.AppSecretPathForID(appID, app, svc.Name, name)
		if _, err := s.deps.Secrets.GetSecret(ctx, path); err != nil {
			return fmt.Errorf("deployment: secret %s (env %s) missing: %w", path, envKey, domain.ErrAppSecretMissing)
		}
	}
	return nil
}

// appSecretID returns the stable internal UUID for secret paths.
// Empty means legacy name-keyed paths (records predating UUIDs).
// Used on the execution path where preflight's ownership snapshot is
// not in scope; deploys are infrequent, one extra store read per
// service is acceptable.
func (s *Service) appSecretID(ctx context.Context, app string) (string, error) {
	ownership, err := s.deps.State.LoadOwnership(ctx, app)
	if err != nil {
		return "", err
	}
	return ownership.ID, nil
}

func (s *Service) preflightImage(ctx context.Context, image, digest string) (string, error) {
	if err := validation.ValidateImageDigest(digest); err != nil {
		return "", fmt.Errorf("deployment: image %q resolved to invalid digest: %w", image, domain.ErrAppImageUnresolvable)
	}
	if err := s.validateImageSource(image, digest); err != nil {
		return "", err
	}
	if s.deps.Registry.Domain == "" {
		return image, nil
	}
	return s.pullImage(ctx, stripImageTag(image)+"@"+digest)
}

// validateImageSource enforces the installation image policy on a
// reference before any pull. Both the manifest reference and the pinned
// digest are checked, so a digest-pinned ref cannot bypass the hostname
// and port allowlist on deploy or recovery.
func (s *Service) validateImageSource(image, digest string) error {
	ref := strings.TrimSpace(image)
	if digest != "" && !strings.Contains(ref, "@") {
		ref = stripImageTag(image) + "@" + digest
	}
	if err := s.deps.ImagePolicy.ValidateImageSource(ref); err != nil {
		return fmt.Errorf("deployment: image %q: %w", image, err)
	}
	return nil
}

// checkImageVolumes rejects images with unmapped VOLUME declarations.
func (s *Service) checkImageVolumes(ctx context.Context, svc domain.AppService, runtimeImage string) error {
	declared, err := s.deps.Runtime.InspectImageVolumes(ctx, runtimeImage)
	if err != nil {
		return fmt.Errorf("deployment: inspect image volumes: %w", err)
	}
	if len(declared) == 0 {
		return nil
	}
	mapped := map[string]struct{}{}
	for _, vol := range svc.Volumes {
		mapped[vol.Path] = struct{}{}
	}
	// Bind destinations also map image-declared volumes: a bind that
	// mounts over an image VOLUME is an explicit operator mapping.
	for _, bind := range svc.Binds {
		mapped[bind.Path] = struct{}{}
	}
	for _, path := range declared {
		if _, ok := mapped[path]; !ok {
			return fmt.Errorf(
				"deployment: image declares unmanaged volume %q, map it explicitly: %w",
				path, domain.ErrAppUnmanagedImageVolume,
			)
		}
	}
	return nil
}

// recheckReservations re-validates against the live global table.
func (s *Service) recheckReservations(ctx context.Context, app string, rev domain.AppDesiredRevision) error {
	checkpoint, err := s.deps.State.LoadCheckpoint(ctx)
	if err != nil {
		return err
	}
	reservations := domain.ReservationsFor(rev.Spec)
	if err := domain.CheckReservations(checkpoint.Reservations, reservations, app); err != nil {
		return fmt.Errorf("deployment: %w", err)
	}
	return nil
}

// checkResources verifies volumes exist-or-creatable and ownership is consistent.
// A generated runtime volume that already exists but is NOT recorded in
// this app's ownership is refused fail-closed: mounting it would
// implicitly adopt (and possibly modify) foreign data. Volumes recorded
// in ownership (attached or retained) were created by this app's own
// deploys and are safe to reuse.
func (s *Service) checkResources(ctx context.Context, app string, ownership domain.AppOwnership, services []domain.AppService) error {
	owned := map[string]string{}
	ownedRuntime := map[string]struct{}{}
	for _, vol := range ownership.Volumes {
		owned[vol.Name] = vol.Service
		ownedRuntime[vol.RuntimeName] = struct{}{}
	}
	for _, svc := range services {
		for _, vol := range svc.Volumes {
			if owner, ok := owned[vol.Name]; ok && owner != svc.Name {
				return fmt.Errorf(
					"deployment: volume %q owned by %q, not %q: %w",
					vol.Name, owner, svc.Name, domain.ErrAppStateConflict,
				)
			}
			runtimeName := domain.RuntimeVolumeName(app, svc.Name, vol.Name)
			exists, err := s.deps.Runtime.VolumeExists(ctx, runtimeName)
			if err != nil {
				return fmt.Errorf("deployment: check volume %q: %w", runtimeName, err)
			}
			if exists {
				if _, ok := ownedRuntime[runtimeName]; !ok {
					return fmt.Errorf(
						"deployment: runtime volume %q already exists but is not owned by app %q: refusing to adopt foreign data: %w",
						runtimeName, app, domain.ErrAppStateConflict,
					)
				}
			}
		}
	}
	return nil
}

// ComputeOutcome derives the op outcome over terminal per-service results.
func ComputeOutcome(results map[string]ServiceResult) string {
	deployed := 0
	failed := 0
	for _, result := range results {
		switch result.Result {
		case "deployed", ServiceResultUnchanged:
			deployed++
		case "failed":
			failed++
		}
	}
	switch {
	case deployed > 0 && failed > 0:
		return domain.AppOutcomePartial
	case deployed > 0:
		return domain.AppOutcomeSuccess
	default:
		return domain.AppOutcomeFailed
	}
}

// itoa formats integers without extra imports.
func itoa(n int) string {
	return fmt.Sprintf("%d", n)
}