proxy-pool/internal/controller/admin/service.go

package admin

import (
	"context"
	"errors"
	"sort"
	"strings"
	"time"

	"proxy-pool/internal/config"
	"proxy-pool/internal/domain/adminstate"
)

var ErrInvalidApplicationService = errors.New("invalid admin application service")

type StateRepository interface {
	adminstate.Mutator
	adminstate.SnapshotReader
}

type OperationalStatusReader interface {
	ReadOperationalStatus(context.Context) (OperationalStatus, error)
}

type ConfigurationLoader interface {
	LoadConfiguration(context.Context) (LoadedConfiguration, error)
}

// ConfigurationPublisher publishes only a newer authoritative configuration revision.
type ConfigurationPublisher interface {
	PublishRevision(*config.Config, uint64) bool
}

type RuntimeController interface {
	Notify()
	ValidateConfiguration(context.Context, *config.Config) error
	ValidateUpstream(context.Context, string) error
}

// SnapshotRefreshNotifier requests immediate full Worker snapshots after a
// committed management change. Implementations must coalesce notifications.
type SnapshotRefreshNotifier interface {
	NotifySnapshotRefresh()
}

var _ ConfigurationPublisher = (*config.Store)(nil)

type ApplicationDependencies struct {
	State           StateRepository
	Operations      OperationalStatusReader
	Configuration   ConfigurationLoader
	Publisher       ConfigurationPublisher
	Runtime         RuntimeController
	SnapshotRefresh SnapshotRefreshNotifier
}

type ApplicationOptions struct {
	Now            func() time.Time
	FingerprintKey []byte
}

type OperationalStatus struct {
	SnapshotVersion uint64
	Upstreams       []UpstreamActivity
	Workers         []WorkerStatus
}

type UpstreamActivity struct {
	Name                  string
	Available             int64
	Checking              int64
	Suspect               int64
	Draining              int64
	Extracted             int64
	ConsecutiveEmptyFetch int64
	FetchErrorCount       int64
}

type LoadedConfiguration struct {
	Value  *config.Config
	Source string
}

type ApplicationService struct {
	state           StateRepository
	operations      OperationalStatusReader
	configuration   ConfigurationLoader
	publisher       ConfigurationPublisher
	runtime         RuntimeController
	snapshotRefresh SnapshotRefreshNotifier
	now             func() time.Time
	fingerprintKey  []byte
}

var _ Service = (*ApplicationService)(nil)

func NewApplicationService(dependencies ApplicationDependencies, options ApplicationOptions) (*ApplicationService, error) {
	if nilInterface(dependencies.State) || nilInterface(dependencies.Operations) || nilInterface(dependencies.Configuration) ||
		nilInterface(dependencies.Publisher) || options.Now == nil ||
		len(options.FingerprintKey) < config.MinimumFingerprintKeyBytes {
		return nil, ErrInvalidApplicationService
	}
	return &ApplicationService{
		state:           dependencies.State,
		operations:      dependencies.Operations,
		configuration:   dependencies.Configuration,
		publisher:       dependencies.Publisher,
		runtime:         dependencies.Runtime,
		snapshotRefresh: dependencies.SnapshotRefresh,
		now:             options.Now,
		fingerprintKey:  append([]byte(nil), options.FingerprintKey...),
	}, nil
}

func (service *ApplicationService) SetUpstreamEnabled(ctx context.Context, command SetUpstreamCommand) (MutationResult, error) {
	if command.Enabled && service.runtime != nil {
		if err := service.runtime.ValidateUpstream(ctx, command.Name); err != nil {
			return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err)
		}
	}
	result, err := service.state.SetUpstreamEnabled(ctx, adminstate.SetUpstreamCommand{
		RequestID:  command.RequestID,
		Actor:      adminstate.Actor{ID: command.ActorID, SourceIP: command.SourceIP},
		OccurredAt: service.now().UTC(),
		Name:       command.Name,
		Enabled:    command.Enabled,
	})
	if err == nil && service.runtime != nil {
		service.runtime.Notify()
	}
	if err == nil && result.Changed && service.snapshotRefresh != nil {
		service.snapshotRefresh.NotifySnapshotRefresh()
	}
	return mutationResult(result), mapAdminStateError(err)
}

func (service *ApplicationService) SwitchRouting(ctx context.Context, command SwitchCommand) (MutationResult, error) {
	result, err := service.state.SwitchRouting(ctx, adminstate.SwitchRoutingCommand{
		RequestID:       command.RequestID,
		Actor:           adminstate.Actor{ID: command.ActorID, SourceIP: command.SourceIP},
		OccurredAt:      service.now().UTC(),
		Name:            command.Name,
		ExpectedCurrent: command.ExpectedCurrent,
		Target:          command.Target,
		Reason:          command.Reason,
	})
	if err == nil && result.Changed && service.snapshotRefresh != nil {
		service.snapshotRefresh.NotifySnapshotRefresh()
	}
	return mutationResult(result), mapAdminStateError(err)
}

func (service *ApplicationService) Status(ctx context.Context) (Status, error) {
	snapshot, err := service.state.Snapshot(ctx)
	if err != nil {
		return Status{}, mapAdminStateError(err)
	}
	operations, err := service.operations.ReadOperationalStatus(ctx)
	if err != nil {
		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
			return Status{}, err
		}
		return Status{}, errors.Join(ErrUnavailable, err)
	}

	activityByName := make(map[string]UpstreamActivity, len(operations.Upstreams))
	for _, activity := range operations.Upstreams {
		activityByName[activity.Name] = activity
	}
	status := Status{SnapshotVersion: operations.SnapshotVersion}
	if snapshot.Config != nil {
		status.ConfigVersion = snapshot.Config.ConfigVersion
	}
	status.Upstreams = make([]UpstreamStatus, 0, len(snapshot.Upstreams))
	for _, authoritative := range snapshot.Upstreams {
		activity := activityByName[authoritative.Name]
		status.Upstreams = append(status.Upstreams, UpstreamStatus{
			Name:                  authoritative.Name,
			Enabled:               authoritative.Enabled,
			Available:             activity.Available,
			Checking:              activity.Checking,
			Suspect:               activity.Suspect,
			Draining:              activity.Draining,
			Extracted:             activity.Extracted,
			ConsecutiveEmptyFetch: activity.ConsecutiveEmptyFetch,
			FetchErrorCount:       activity.FetchErrorCount,
		})
	}
	sort.Slice(status.Upstreams, func(left, right int) bool {
		return status.Upstreams[left].Name < status.Upstreams[right].Name
	})
	status.Workers = append([]WorkerStatus(nil), operations.Workers...)
	sort.Slice(status.Workers, func(left, right int) bool {
		if status.Workers[left].ID == status.Workers[right].ID {
			return status.Workers[left].Zone < status.Workers[right].Zone
		}
		return status.Workers[left].ID < status.Workers[right].ID
	})
	return status, nil
}

func (service *ApplicationService) ReloadConfiguration(ctx context.Context, command ReloadCommand) (MutationResult, error) {
	loaded, err := service.configuration.LoadConfiguration(ctx)
	if err != nil {
		switch {
		case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
			return MutationResult{RequestID: command.RequestID}, err
		case errors.Is(err, ErrUnavailable), errors.Is(err, ErrInvalidConfiguration):
			return MutationResult{RequestID: command.RequestID}, err
		default:
			return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err)
		}
	}
	return service.ApplyConfiguration(ctx, command, loaded)
}

// ApplyConfiguration persists and publishes one already loaded configuration snapshot.
// Startup and runtime reload paths share this method so storage connections and the
// committed management view cannot be built from different reads of the source file.
func (service *ApplicationService) ApplyConfiguration(
	ctx context.Context,
	command ReloadCommand,
	loaded LoadedConfiguration,
) (MutationResult, error) {
	if ctx == nil {
		return MutationResult{RequestID: command.RequestID}, ErrInvalidConfiguration
	}
	if err := ctx.Err(); err != nil {
		return MutationResult{RequestID: command.RequestID}, err
	}
	if loaded.Value == nil || strings.TrimSpace(loaded.Source) != loaded.Source || loaded.Source == "" ||
		len(loaded.Source) > adminstate.MaxSourceBytes {
		return MutationResult{RequestID: command.RequestID}, ErrInvalidConfiguration
	}
	if err := config.Validate(loaded.Value); err != nil {
		return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err)
	}
	if service.runtime != nil {
		if err := service.runtime.ValidateConfiguration(ctx, loaded.Value); err != nil {
			return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err)
		}
	}

	checksum, err := config.Fingerprint(loaded.Value, service.fingerprintKey)
	if err != nil {
		return MutationResult{RequestID: command.RequestID}, errors.Join(ErrInvalidConfiguration, err)
	}

	current, err := service.state.Snapshot(ctx)
	if err != nil {
		return MutationResult{RequestID: command.RequestID}, mapAdminStateError(err)
	}
	upstreams, routings := managementDefinitions(loaded.Value, current)
	result, err := service.state.CommitConfig(ctx, adminstate.CommitConfigCommand{
		RequestID:     command.RequestID,
		Actor:         adminstate.Actor{ID: command.ActorID, SourceIP: command.SourceIP},
		OccurredAt:    service.now().UTC(),
		ConfigVersion: "cfg-" + checksum,
		Checksum:      checksum,
		Source:        loaded.Source,
		Upstreams:     upstreams,
		Routings:      routings,
	})
	if err != nil {
		return mutationResult(result), mapAdminStateError(err)
	}
	if result.Revision == 0 {
		return mutationResult(result), errors.Join(ErrUnavailable, ErrInvalidApplicationService)
	}
	published := service.publisher.PublishRevision(loaded.Value, result.Revision)
	if published && service.runtime != nil {
		service.runtime.Notify()
	}
	if published && service.snapshotRefresh != nil {
		service.snapshotRefresh.NotifySnapshotRefresh()
	}
	return mutationResult(result), nil
}

func managementDefinitions(configuration *config.Config, current adminstate.Snapshot) ([]adminstate.UpstreamDefinition, []adminstate.RoutingDefinition) {
	upstreams := make([]adminstate.UpstreamDefinition, 0, len(configuration.Upstreams))
	for name, upstream := range configuration.Upstreams {
		upstreams = append(upstreams, adminstate.UpstreamDefinition{Name: name, Enabled: upstream.Enabled})
	}
	sort.Slice(upstreams, func(left, right int) bool { return upstreams[left].Name < upstreams[right].Name })

	currentByName := make(map[string]string, len(current.Routings))
	for _, routing := range current.Routings {
		currentByName[routing.Name] = routing.CurrentUpstream
	}
	routings := make([]adminstate.RoutingDefinition, 0, len(configuration.Routing))
	for _, routing := range configuration.Routing {
		selected := ""
		if existing := currentByName[routing.Name]; containsString(routing.Upstreams, existing) {
			selected = existing
		} else if len(routing.Upstreams) > 0 {
			selected = routing.Upstreams[0]
		}
		routings = append(routings, adminstate.RoutingDefinition{
			Name: routing.Name, Enabled: routing.Enabled,
			Upstreams: append([]string(nil), routing.Upstreams...), CurrentUpstream: selected,
		})
	}
	sort.Slice(routings, func(left, right int) bool { return routings[left].Name < routings[right].Name })
	return upstreams, routings
}

func containsString(values []string, target string) bool {
	for _, value := range values {
		if value == target {
			return true
		}
	}
	return false
}

func mutationResult(result adminstate.MutationResult) MutationResult {
	return MutationResult{
		RequestID: result.RequestID,
		Changed:   result.Changed,
		Version:   result.Revision,
		Message:   result.Message,
	}
}

func mapAdminStateError(err error) error {
	if err == nil {
		return nil
	}
	switch {
	case errors.Is(err, adminstate.ErrNotFound):
		return errors.Join(ErrNotFound, err)
	case errors.Is(err, adminstate.ErrConflict):
		return errors.Join(ErrConflict, err)
	case errors.Is(err, adminstate.ErrInvalidCommand):
		return errors.Join(ErrInvalidConfiguration, err)
	case errors.Is(err, adminstate.ErrUnavailable):
		return errors.Join(ErrUnavailable, err)
	default:
		return err
	}
}