silo-server/internal/policy/system.go

package policy

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"time"

	"github.com/Silo-Server/silo-server/internal/cache"
)

const defaultSystemPollInterval = 60 * time.Second

// System owns the live policy engine, reloads it when policy documents change,
// and exposes a stable PDP pointer for request-path adapters.
type System struct {
	store          *PolicyStore
	eventBus       cache.EventBus
	logger         *slog.Logger
	decisionLogger *DecisionLogger

	pollInterval time.Duration
	evalTimeout  time.Duration

	mu       sync.RWMutex
	engine   *Engine
	pdp      *PDP
	cancel   context.CancelFunc
	reloadCh chan struct{}
	wg       sync.WaitGroup

	// bootDegradedReason is set when the initial engine could not load the
	// full stored policy (store unreachable or custom bundle failed) and is
	// cleared by the first successful reload from the store.
	bootDegradedReason string
}

// Degraded-state reasons reported by DegradedState.
const (
	DegradedReasonStoreUnavailable    = "store_unavailable"
	DegradedReasonCustomBundleFailed  = "custom_bundle_failed"
	DegradedReasonCustomSourceInvalid = "custom_source_invalid"
)

// DegradedState reports whether the live engine is serving with less than the
// full set of enabled custom policy sources. Because custom policy is
// tighten-only, a degraded engine is more permissive than administrator intent.
type DegradedState struct {
	Degraded bool
	Reason   string
	// Domains lists policy domains whose enabled custom source was dropped
	// from the loaded bundle (empty when the whole store was unavailable).
	Domains []string
}

// SystemOption configures a System.
type SystemOption func(*System)

// WithSystemPollInterval configures how often the poll fallback checks policy
// generation. Non-positive durations keep the default.
func WithSystemPollInterval(interval time.Duration) SystemOption {
	return func(system *System) {
		if interval > 0 {
			system.pollInterval = interval
		}
	}
}

// WithSystemEvalTimeout configures the per-decision policy evaluation timeout.
// Non-positive durations keep the default.
func WithSystemEvalTimeout(timeout time.Duration) SystemOption {
	return func(system *System) {
		if timeout > 0 {
			system.evalTimeout = timeout
		}
	}
}

// WithSystemDecisionLogger configures the async logger used by the PDP.
func WithSystemDecisionLogger(logger *DecisionLogger) SystemOption {
	return func(system *System) {
		system.decisionLogger = logger
	}
}

// NewSystem constructs a policy System. Call Start before using PDP.
func NewSystem(store *PolicyStore, eventBus cache.EventBus, logger *slog.Logger, opts ...SystemOption) *System {
	if logger == nil {
		logger = slog.Default()
	}
	system := &System{
		store:        store,
		eventBus:     eventBus,
		logger:       logger.With("component", "policy.system"),
		pollInterval: defaultSystemPollInterval,
		evalTimeout:  defaultEvalTimeout,
		reloadCh:     make(chan struct{}, 1),
	}
	for _, opt := range opts {
		opt(system)
	}
	return system
}

// Start loads the initial engine, subscribes to policy-change events, and
// starts the poll fallback loop.
func (s *System) Start(ctx context.Context) error {
	if s == nil {
		return nil
	}

	engine, bootDegradedReason, err := s.initialEngine(ctx)
	if err != nil {
		return err
	}

	runCtx, cancel := context.WithCancel(ctx)
	s.mu.Lock()
	if s.engine != nil {
		s.mu.Unlock()
		cancel()
		return errors.New("policy system already started")
	}
	s.engine = engine
	s.bootDegradedReason = bootDegradedReason
	s.pdp = NewPDP(engine, WithDecisionLogger(s.decisionLogger))
	s.cancel = cancel
	s.mu.Unlock()

	if s.decisionLogger != nil {
		s.decisionLogger.Start(runCtx)
	}

	if s.eventBus != nil {
		if err := s.eventBus.Subscribe(runCtx, cache.ChannelAdmin, func(event cache.Event) {
			if event.Type == cache.EventPolicyChanged {
				s.requestReload()
			}
		}); err != nil {
			s.logger.WarnContext(ctx, "policy system: subscribe to admin channel failed, using poll-only mode", "error", err)
		}
	}

	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		s.poll(runCtx)
	}()
	return nil
}

// Stop cancels background work and waits for it to exit.
func (s *System) Stop() {
	if s == nil {
		return
	}
	s.mu.RLock()
	cancel := s.cancel
	s.mu.RUnlock()
	if cancel != nil {
		cancel()
	}
	if s.decisionLogger != nil {
		s.decisionLogger.Stop()
	}
	s.Wait()
}

// Wait blocks until background reload loops exit.
func (s *System) Wait() {
	if s != nil {
		s.wg.Wait()
		if s.decisionLogger != nil {
			s.decisionLogger.Wait()
		}
	}
}

// PDP returns the live policy decision point. The returned PDP remains valid
// across engine reloads because the System mutates one Engine in place.
func (s *System) PDP() *PDP {
	if s == nil {
		return nil
	}
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.pdp
}

// DecisionLogger returns the async decision logger attached to the System.
func (s *System) DecisionLogger() *DecisionLogger {
	if s == nil {
		return nil
	}
	return s.decisionLogger
}

// Generation returns the policy generation currently loaded in the live engine.
func (s *System) Generation() int64 {
	if s == nil {
		return 0
	}
	s.mu.RLock()
	engine := s.engine
	s.mu.RUnlock()
	if engine == nil {
		return 0
	}
	return engine.Revision()
}

// DegradedState reports whether the live engine dropped any enabled custom
// policy (boot-time store outage, custom bundle failure, or per-source compile
// skips). A degraded engine serves more permissive decisions than the stored
// policy, so operators must be able to see it.
func (s *System) DegradedState() DegradedState {
	if s == nil {
		return DegradedState{}
	}
	s.mu.RLock()
	engine := s.engine
	bootReason := s.bootDegradedReason
	s.mu.RUnlock()

	if bootReason != "" {
		return DegradedState{Degraded: true, Reason: bootReason}
	}
	if engine == nil {
		return DegradedState{}
	}
	skipped := engine.SkippedSources()
	if len(skipped) == 0 {
		return DegradedState{}
	}
	domains := make([]string, 0, len(skipped))
	for _, skip := range skipped {
		domains = append(domains, skip.Domain)
	}
	return DegradedState{
		Degraded: true,
		Reason:   DegradedReasonCustomSourceInvalid,
		Domains:  domains,
	}
}

// EvalTimeout returns the configured per-decision evaluation budget, used by
// the activation-time cost guard so it measures against the live budget.
func (s *System) EvalTimeout() time.Duration {
	if s == nil {
		return defaultEvalTimeout
	}
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.evalTimeout
}

// EvalTimeouts returns how many evaluations exceeded the eval budget on the
// live engine since boot.
func (s *System) EvalTimeouts() int64 {
	if s == nil {
		return 0
	}
	s.mu.RLock()
	engine := s.engine
	s.mu.RUnlock()
	if engine == nil {
		return 0
	}
	return engine.EvalTimeouts()
}

// SetEvalTimeout hot-updates the per-decision policy evaluation timeout.
func (s *System) SetEvalTimeout(timeout time.Duration) {
	if s == nil || timeout <= 0 {
		return
	}
	s.mu.Lock()
	s.evalTimeout = timeout
	engine := s.engine
	s.mu.Unlock()
	if engine != nil {
		engine.SetEvalTimeout(timeout)
	}
}

// ApplyStatus describes how a policy change was applied: whether this node's
// engine reloaded to the stored generation and whether peers were notified.
type ApplyStatus struct {
	LocalReloadErr error
	PublishErr     error
	// Generation is the generation loaded in the live engine after the apply
	// attempt — it matches the stored generation only when Applied.
	Generation int64
}

// Applied reports whether the local engine now serves the stored policy.
func (st ApplyStatus) Applied() bool { return st.LocalReloadErr == nil }

// FailedStep names the first failed apply step ("local_reload" or
// "event_publish"), or "" when both steps succeeded.
func (st ApplyStatus) FailedStep() string {
	switch {
	case st.LocalReloadErr != nil:
		return "local_reload"
	case st.PublishErr != nil:
		return "event_publish"
	}
	return ""
}

// Err joins the per-step errors, nil when the change fully applied.
func (st ApplyStatus) Err() error { return errors.Join(st.LocalReloadErr, st.PublishErr) }

// NotifyChanged reloads this node synchronously and publishes a cross-node
// invalidation event. The last known-good engine remains active on reload
// failure.
func (s *System) NotifyChanged(ctx context.Context) error {
	if s == nil {
		return nil
	}
	return s.ApplyChanged(ctx).Err()
}

// ApplyChanged reloads this node synchronously, publishes a cross-node
// invalidation event, and reports per-step outcomes so callers can distinguish
// "persisted" from "live". The last known-good engine remains active on reload
// failure.
func (s *System) ApplyChanged(ctx context.Context) ApplyStatus {
	if s == nil {
		return ApplyStatus{}
	}

	var status ApplyStatus
	if err := s.reloadFromStore(ctx); err != nil {
		s.logger.ErrorContext(ctx, "policy reload after local change failed", "error", err)
		status.LocalReloadErr = err
	}
	if s.eventBus != nil {
		if err := s.eventBus.Publish(ctx, cache.ChannelAdmin, cache.Event{Type: cache.EventPolicyChanged}); err != nil {
			s.logger.ErrorContext(ctx, "policy change publish failed", "error", err)
			status.PublishErr = err
		}
	}
	status.Generation = s.Generation()
	return status
}

func (s *System) initialEngine(ctx context.Context) (*Engine, string, error) {
	sources, generation, err := s.loadSnapshot(ctx)
	if err != nil {
		s.logger.ErrorContext(ctx, "policy store unavailable; starting DEGRADED with vendor policy only", "error", err)
		engine, err := s.newVendorEngine(ctx)
		return engine, DegradedReasonStoreUnavailable, err
	}

	engine, err := NewEngineWithCustom(ctx, sources, s.engineOptions(WithRevision(generation))...)
	if err != nil {
		s.logger.ErrorContext(ctx, "policy custom bundle load failed; starting DEGRADED with vendor policy only", "error", err)
		engine, err := s.newVendorEngine(ctx)
		return engine, DegradedReasonCustomBundleFailed, err
	}
	return engine, "", nil
}

func (s *System) newVendorEngine(ctx context.Context) (*Engine, error) {
	engine, err := NewEngine(ctx, s.engineOptions()...)
	if err != nil {
		return nil, fmt.Errorf("compile vendor policy: %w", err)
	}
	return engine, nil
}

func (s *System) engineOptions(extra ...EngineOption) []EngineOption {
	opts := []EngineOption{
		WithLogger(s.logger),
		WithEvalTimeout(s.evalTimeout),
	}
	return append(opts, extra...)
}

func (s *System) reloadFromStore(ctx context.Context) error {
	sources, generation, err := s.loadSnapshot(ctx)
	if err != nil {
		return err
	}

	s.mu.RLock()
	engine := s.engine
	s.mu.RUnlock()
	if engine == nil {
		return errors.New("policy system is not started")
	}
	if err := engine.Reload(ctx, sources, generation); err != nil {
		return err
	}
	s.mu.Lock()
	s.bootDegradedReason = ""
	s.mu.Unlock()
	s.logger.InfoContext(ctx, "policy engine reloaded", "generation", generation)
	return nil
}

func (s *System) loadSnapshot(ctx context.Context) (map[string]ActiveSource, int64, error) {
	if s.store == nil {
		return nil, 0, errors.New("policy store is nil")
	}

	for {
		before, err := s.store.Generation(ctx)
		if err != nil {
			return nil, 0, err
		}
		sources, err := s.store.ActiveSources(ctx)
		if err != nil {
			return nil, 0, err
		}
		after, err := s.store.Generation(ctx)
		if err != nil {
			return nil, 0, err
		}
		if before == after {
			return sources, after, nil
		}
		if err := ctx.Err(); err != nil {
			return nil, 0, err
		}
	}
}

func (s *System) requestReload() {
	select {
	case s.reloadCh <- struct{}{}:
	default:
	}
}

func (s *System) poll(ctx context.Context) {
	ticker := time.NewTicker(s.pollInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			s.reloadIfGenerationChanged(ctx)
		case <-s.reloadCh:
			if err := s.reloadFromStore(ctx); err != nil {
				s.logger.ErrorContext(ctx, "policy event reload failed", "error", err)
			}
		}
	}
}

func (s *System) reloadIfGenerationChanged(ctx context.Context) {
	if s.store == nil {
		return
	}
	generation, err := s.store.Generation(ctx)
	if err != nil {
		s.logger.ErrorContext(ctx, "policy generation poll failed", "error", err)
		return
	}

	s.mu.RLock()
	engine := s.engine
	s.mu.RUnlock()
	if engine == nil || generation == engine.Revision() {
		return
	}
	if err := s.reloadFromStore(ctx); err != nil {
		s.logger.ErrorContext(ctx, "policy poll reload failed", "error", err)
	}
}