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 adminstate.AuditReader } 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) } // ReadAudit returns one bounded, stable page from the authoritative management // audit log. It deliberately exposes only management-plane records and never // reads Proxy activity, credentials, or Provider payloads. func (service *ApplicationService) ReadAudit(ctx context.Context, query AuditQuery) (AuditPage, error) { if service == nil || nilInterface(service.state) || query.Validate() != nil { return AuditPage{}, ErrInvalidConfiguration } records, err := service.state.ReadAudit(ctx, adminstate.AuditQuery{AfterID: query.AfterID, Limit: query.Limit}) if err != nil { return AuditPage{}, mapAdminStateError(err) } page := AuditPage{Records: make([]AuditRecord, 0, len(records))} for _, record := range records { page.Records = append(page.Records, AuditRecord{ ID: record.ID, RequestID: record.RequestID, ActorID: record.Actor.ID, SourceIP: record.Actor.SourceIP, Action: string(record.Action), ResourceType: record.ResourceType, ResourceName: record.ResourceName, Changed: record.Changed, Version: record.Revision, Reason: record.Reason, OccurredAt: record.OccurredAt.UTC(), }) } return page, nil } 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 } }