Files
proxy/internal/app/app.go
T

699 lines
24 KiB
Go

package app
import (
"context"
"errors"
"fmt"
"strings"
"sync"
"time"
"proxyctl/internal/profile"
"proxyctl/internal/target"
)
type IntentKind int
const (
CreateProfile IntentKind = iota + 1
UpdateProfile
DeleteProfile
ActivateProfile
SetTargetDesired
RefreshTargets
)
type Intent struct {
Kind IntentKind
Name string
Profile profile.Profile
Confirm bool
TargetID string
Enabled bool
}
type TargetDetectionPhase uint8
const (
// TargetReady is the zero value so snapshots supplied by non-progressive
// Modules remain compatible with the established dashboard contract.
TargetReady TargetDetectionPhase = iota
TargetChecking
TargetDetectionFailed
)
type TargetSnapshot struct {
ID string
Description target.Description
DetectionPhase TargetDetectionPhase
DetectionErr error
Available bool
AvailabilityEvidence string
State target.State
Evidence string
}
// TargetDetectionUpdate carries the complete authoritative Application state
// after one Target changed. TargetID identifies that change without duplicating
// availability, inspection, or failure fields already present in Snapshot.
type TargetDetectionUpdate struct {
Snapshot Snapshot
TargetID string
Complete bool
}
type Snapshot struct {
Profiles []profile.Profile
ActiveProfile string
Targets []TargetSnapshot
}
type TargetProgress struct {
TargetID string
// Applying is retained for synchronous DispatchProgress callers. New
// interactive callers use the structured Phase field.
Applying bool
Phase TargetProgressPhase
State target.State
Evidence string
Err error
}
type Result struct {
Snapshot Snapshot
Notice string
// Confirmation remains for existing synchronous callers.
Confirmation string
ConfirmationRequest *Confirmation
Status OperationStatus
Progress []TargetProgress
FirstFailedTargetID string
Complete bool
Err error
}
type Module interface {
Snapshot() Snapshot
Dispatch(Intent) Result
}
// ProgressiveModule is implemented by Applications that can expose each
// coordinated Target update while an Active Profile transition is running.
// The final Result has Complete set and closes the stream.
type ProgressiveModule interface {
DispatchProgress(Intent) <-chan Result
}
// ProgressiveDetectionModule exposes a useful all-Checking snapshot
// synchronously, followed by one structured Target result at a time. Startup
// and explicit refresh use this same presentation-independent contract.
type ProgressiveDetectionModule interface {
BeginTargetDetection() (Snapshot, <-chan TargetDetectionUpdate)
}
type Application struct {
store profile.Store
state profile.State
targets []target.Target
targetIDs []string
targetDescriptions []target.Description
availabilities map[string]target.Availability
inspections map[string]target.Inspection
detectionPhase map[string]TargetDetectionPhase
detectionErr map[string]error
detectionGeneration uint64
stateMu sync.RWMutex
dispatchMu sync.Mutex
interactiveMu sync.Mutex
interactiveOperation bool
mutationLock MutationLock
logger OperationLogger
}
func New(store profile.Store) (*Application, error) {
return NewWithTargets(store, nil)
}
func NewWithTargets(store profile.Store, targets []target.Target) (*Application, error) {
return NewWithTargetsAndLock(store, targets, nil)
}
// NewWithTargetsAndLock configures the cross-process lock used by every
// Profile or Target mutation. A nil lock is intended only for isolated tests.
func NewWithTargetsAndLock(store profile.Store, targets []target.Target, mutationLock MutationLock) (*Application, error) {
return NewWithTargetsAndRuntime(store, targets, mutationLock, nil)
}
// NewWithTargetsAndRuntime configures shared mutation serialization and local
// operation diagnostics for production callers.
func NewWithTargetsAndRuntime(store profile.Store, targets []target.Target, mutationLock MutationLock, logger OperationLogger) (*Application, error) {
state, err := store.Load()
if err != nil {
return nil, err
}
a := &Application{
store: store,
state: state,
targets: append([]target.Target(nil), targets...),
availabilities: make(map[string]target.Availability),
inspections: make(map[string]target.Inspection),
detectionPhase: make(map[string]TargetDetectionPhase),
detectionErr: make(map[string]error),
mutationLock: mutationLock,
logger: logger,
}
for _, adapter := range a.targets {
id := adapter.ID()
a.targetIDs = append(a.targetIDs, id)
a.targetDescriptions = append(a.targetDescriptions, adapter.Description())
a.detectionPhase[id] = TargetChecking
}
return a, nil
}
func (a *Application) Snapshot() Snapshot {
a.stateMu.RLock()
defer a.stateMu.RUnlock()
return a.snapshotLocked()
}
func (a *Application) snapshotLocked() Snapshot {
state := cloneState(a.state)
snapshot := Snapshot{Profiles: state.Profiles, ActiveProfile: state.ActiveProfile}
for i, id := range a.targetIDs {
availability := a.availabilities[id]
inspection := a.inspections[id]
snapshot.Targets = append(snapshot.Targets, TargetSnapshot{
ID: id, Description: a.targetDescriptions[i], DetectionPhase: a.detectionPhase[id], DetectionErr: a.detectionErr[id],
Available: availability.Available, AvailabilityEvidence: availability.Evidence, State: inspection.State, Evidence: inspection.Evidence,
})
}
return snapshot
}
type targetDetectionOutcome struct {
targetID string
detection target.Detection
}
func (a *Application) BeginTargetDetection() (Snapshot, <-chan TargetDetectionUpdate) {
a.stateMu.Lock()
a.detectionGeneration++
generation := a.detectionGeneration
for _, id := range a.targetIDs {
a.detectionPhase[id] = TargetChecking
a.detectionErr[id] = nil
}
active := activeProfile(a.state)
checking := a.snapshotLocked()
a.stateMu.Unlock()
updates := make(chan TargetDetectionUpdate, len(a.targets)+1)
outcomes := make(chan targetDetectionOutcome, len(a.targets))
for i, adapter := range a.targets {
adapter, id := adapter, a.targetIDs[i]
go func() {
outcomes <- targetDetectionOutcome{targetID: id, detection: detectTarget(adapter, active)}
}()
}
go func() {
defer close(updates)
for range a.targets {
outcome := <-outcomes
a.stateMu.Lock()
if generation != a.detectionGeneration {
a.stateMu.Unlock()
continue
}
phase := TargetReady
if outcome.detection.Err != nil {
phase = TargetDetectionFailed
}
a.availabilities[outcome.targetID] = outcome.detection.Availability
a.inspections[outcome.targetID] = outcome.detection.Inspection
a.detectionPhase[outcome.targetID] = phase
a.detectionErr[outcome.targetID] = outcome.detection.Err
update := TargetDetectionUpdate{Snapshot: a.snapshotLocked(), TargetID: outcome.targetID}
// The buffered send cannot wait on a consumer and happens before a
// newer Begin call can make this generation stale.
updates <- update
a.stateMu.Unlock()
}
a.stateMu.Lock()
if generation == a.detectionGeneration {
updates <- TargetDetectionUpdate{Snapshot: a.snapshotLocked(), Complete: true}
}
a.stateMu.Unlock()
}()
return checking, updates
}
func detectTarget(adapter target.Target, active *profile.Profile) target.Detection {
if detector, ok := adapter.(target.Detector); ok {
return detector.Detect(context.Background(), active)
}
return target.Detection{
Availability: adapter.Availability(context.Background()),
Inspection: adapter.Inspect(context.Background(), active),
}
}
func (a *Application) Dispatch(intent Intent) Result {
return a.dispatch(intent, nil)
}
func (a *Application) DispatchProgress(intent Intent) <-chan Result {
results := make(chan Result, 2*len(a.targets)+1)
go func() {
defer close(results)
final := a.dispatch(intent, func(update Result) { results <- update })
final.Complete = true
results <- final
}()
return results
}
func (a *Application) dispatch(intent Intent, publish func(Result)) Result {
// Serialize local mutation workflows even when tests intentionally omit the
// cross-process MutationLock.
a.dispatchMu.Lock()
defer a.dispatchMu.Unlock()
if a.mutationLock != nil {
unlock, err := a.mutationLock.Lock(context.Background())
if err != nil {
return Result{Snapshot: a.Snapshot(), Err: fmt.Errorf("lock proxyctl mutation: %w", err)}
}
defer unlock()
}
// A different process may have changed the persisted Active Profile since
// this Application was constructed. Reload only after the global mutation
// lock is held, before selecting a Profile or invoking a Target. Load is
// deliberately separate from Store.Mutate so no Profile-store lock is nested.
fresh, err := a.store.Load()
if err != nil {
return Result{Snapshot: a.Snapshot(), Err: fmt.Errorf("reload Profile state for mutation: %w", err)}
}
a.replaceState(fresh)
if intent.Kind == SetTargetDesired || intent.Kind == ActivateProfile {
a.refreshInspections()
}
if intent.Kind == SetTargetDesired {
return a.setTargetDesired(intent, publish)
}
if intent.Kind == RefreshTargets {
a.refreshTargets()
return Result{Snapshot: a.Snapshot(), Notice: "Refreshed Target availability and state"}
}
if intent.Kind == ActivateProfile {
return a.activateProfile(intent, publish)
}
var notice string
next, err := a.store.Mutate(func(current profile.State) (profile.State, error) {
var mutationErr error
switch intent.Kind {
case CreateProfile:
current, notice, mutationErr = create(current, intent.Profile)
case UpdateProfile:
current, notice, mutationErr = update(current, intent.Name, intent.Profile)
case DeleteProfile:
current, notice, mutationErr = deleteProfile(current, intent.Name, intent.Confirm)
default:
mutationErr = errors.New("unsupported Profile action")
}
return current, mutationErr
})
if err == nil {
a.replaceState(next)
a.refreshInspections()
} else {
var committed *profile.CommitError
if errors.As(err, &committed) {
// Rename committed next before directory durability reporting failed.
// Prefer that state even if the required verification reload also fails.
a.replaceState(next)
reloaded, loadErr := a.store.Load()
if loadErr != nil {
err = errors.Join(err, fmt.Errorf("reload committed Profile configuration: %w", loadErr))
} else {
a.replaceState(reloaded)
}
}
}
return Result{Snapshot: a.Snapshot(), Notice: notice, Err: err}
}
func (a *Application) activateProfile(intent Intent, publish func(Result)) Result {
state := a.stateCopy()
if index(state, intent.Name) < 0 {
return Result{Snapshot: a.Snapshot(), Err: fmt.Errorf("Profile %q does not exist", intent.Name)}
}
if state.ActiveProfile == intent.Name {
return Result{Snapshot: a.Snapshot(), Notice: fmt.Sprintf("Profile %q is already Active", intent.Name)}
}
inspections := a.inspectionsCopy()
var enabled []int
var warnings []string
for i, adapter := range a.targets {
id := a.targetIDs[i]
if inspections[id].State != target.Enabled {
continue
}
enabled = append(enabled, i)
if warning := adapter.Confirmation(true); warning != "" {
warnings = append(warnings, a.targetDescriptions[i].Name+": "+warning)
}
}
if len(warnings) > 0 && !intent.Confirm {
message := "Changing the Active Profile requires confirmation:\n" + strings.Join(warnings, "\n")
confirmation := &Confirmation{
Title: "Confirm Active Profile change", Summary: "Changing the Active Profile has high-impact Target effects.",
Details: warnings, ConfirmLabel: "Change Profile", CancelLabel: "Cancel", HighImpact: true,
}
return Result{Snapshot: a.Snapshot(), Confirmation: message, ConfirmationRequest: confirmation,
Status: OperationStatus{Kind: StatusConfirmation, Summary: confirmation.Title, Detail: confirmation.Summary}}
}
next, err := a.store.Mutate(func(current profile.State) (profile.State, error) {
next, _, mutationErr := activate(current, intent.Name)
return next, mutationErr
})
if err != nil {
var committed *profile.CommitError
if errors.As(err, &committed) {
a.replaceState(next)
if reloaded, loadErr := a.store.Load(); loadErr != nil {
err = errors.Join(err, fmt.Errorf("reload committed Profile configuration: %w", loadErr))
} else {
a.replaceState(reloaded)
}
}
return Result{Snapshot: a.Snapshot(), Err: err}
}
a.replaceState(next)
active := a.activeProfile()
// The Active Profile has changed before any Target write begins. Re-inspect
// without the state lock, then publish all observations atomically.
observed := make(map[string]target.Inspection, len(a.targets))
for i, adapter := range a.targets {
observed[a.targetIDs[i]] = adapter.Inspect(context.Background(), active)
}
a.replaceInspections(observed)
var progress []TargetProgress
var failures []error
updated := make(map[string]bool, len(enabled))
for _, targetIndex := range enabled {
adapter := a.targets[targetIndex]
id := a.targetIDs[targetIndex]
name := a.targetDescriptions[targetIndex].Name
updated[id] = true
if publish != nil {
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{{TargetID: id, Applying: true, Phase: TargetProgressRunning, State: a.inspection(id).State}}, Notice: "Updating " + name + " Target",
Status: OperationStatus{Kind: StatusProgress, Summary: "Updating " + name + " Target", TargetID: id}})
}
operation := adapter.SetDesired(context.Background(), true, active)
inspection := operation.Inspection
var safeErr error
if operation.Err != nil {
// A failure-returned inspection may predate a partial write. Always
// inspect explicitly against the new Active Profile before publishing.
inspection = adapter.Inspect(context.Background(), active)
safeErr = fmt.Errorf("%s (fresh inspection: %s — %s)", profile.RedactText(operation.Err.Error()), inspection.State, profile.RedactText(inspection.Evidence))
}
a.replaceInspection(id, inspection)
progressPhase := TargetProgressSucceeded
if safeErr != nil {
progressPhase = TargetProgressFailed
}
item := TargetProgress{TargetID: id, Phase: progressPhase, State: inspection.State, Evidence: inspection.Evidence, Err: safeErr}
progress = append(progress, item)
if publish != nil {
status := OperationStatus{Kind: StatusSuccess, Summary: name + " Target update completed", TargetID: id}
if safeErr != nil {
status = OperationStatus{Kind: StatusError, Summary: name + " Target update failed", Detail: safeErr.Error(), TargetID: id}
}
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{item}, Notice: name + " Target update completed", Status: status, Err: safeErr})
}
if a.logger != nil {
outcome, detail := "success", inspection.Evidence
if safeErr != nil {
outcome, detail = "failure", safeErr.Error()
}
_ = a.logger.Write(LogEntry{Time: time.Now(), Operation: "switch-Active-Profile", Target: id, Result: outcome, Detail: RedactLogDetail(detail)})
}
if safeErr != nil {
failures = append(failures, fmt.Errorf("%s Target: %w", name, safeErr))
}
}
// Re-inspect Targets which were not part of the transition so any managed
// configuration is interpreted against the new Active Profile as well.
remaining := make(map[string]target.Inspection)
for i, adapter := range a.targets {
id := a.targetIDs[i]
if !updated[id] {
remaining[id] = adapter.Inspect(context.Background(), active)
}
}
a.mergeInspections(remaining)
result := Result{Snapshot: a.Snapshot(), Progress: progress, Notice: fmt.Sprintf("Activated Profile %q", intent.Name),
Status: OperationStatus{Kind: StatusSuccess, Summary: fmt.Sprintf("Activated Profile %q", intent.Name)}}
if len(failures) > 0 {
result.Err = errors.Join(failures...)
result.Notice = fmt.Sprintf("Activated Profile %q with %d Target update failure(s); successful updates were kept and no global rollback was attempted", intent.Name, len(failures))
result.Status = OperationStatus{Kind: StatusError, Summary: fmt.Sprintf("Activated Profile %q with %d Target update failure(s)", intent.Name, len(failures)), Detail: result.Err.Error()}
for _, item := range progress {
if item.Err != nil {
result.FirstFailedTargetID = item.TargetID
result.Status.TargetID = item.TargetID
break
}
}
}
return result
}
func activeProfile(state profile.State) *profile.Profile {
for i := range state.Profiles {
if state.Profiles[i].Name == state.ActiveProfile {
value := state.Profiles[i]
value.Bypass = append([]string(nil), value.Bypass...)
return &value
}
}
return nil
}
func (a *Application) activeProfile() *profile.Profile {
a.stateMu.RLock()
defer a.stateMu.RUnlock()
return activeProfile(a.state)
}
func (a *Application) stateCopy() profile.State {
a.stateMu.RLock()
defer a.stateMu.RUnlock()
return cloneState(a.state)
}
func (a *Application) replaceState(state profile.State) {
a.stateMu.Lock()
defer a.stateMu.Unlock()
a.detectionGeneration++
a.state = cloneState(state)
}
func (a *Application) inspectionsCopy() map[string]target.Inspection {
a.stateMu.RLock()
defer a.stateMu.RUnlock()
result := make(map[string]target.Inspection, len(a.inspections))
for id, inspection := range a.inspections {
result[id] = inspection
}
return result
}
func (a *Application) inspection(id string) target.Inspection {
a.stateMu.RLock()
defer a.stateMu.RUnlock()
return a.inspections[id]
}
func (a *Application) replaceInspection(id string, inspection target.Inspection) {
a.stateMu.Lock()
defer a.stateMu.Unlock()
a.detectionGeneration++
a.inspections[id] = inspection
}
func (a *Application) replaceInspections(inspections map[string]target.Inspection) {
a.stateMu.Lock()
defer a.stateMu.Unlock()
a.detectionGeneration++
for _, id := range a.targetIDs {
a.inspections[id] = inspections[id]
}
}
func (a *Application) mergeInspections(inspections map[string]target.Inspection) {
if len(inspections) == 0 {
return
}
a.stateMu.Lock()
defer a.stateMu.Unlock()
a.detectionGeneration++
for id, inspection := range inspections {
a.inspections[id] = inspection
}
}
func (a *Application) refreshTargets() {
_, updates := a.BeginTargetDetection()
for range updates {
}
}
func (a *Application) refreshInspections() {
active := a.activeProfile()
inspections := make(map[string]target.Inspection, len(a.targets))
for i, adapter := range a.targets {
inspections[a.targetIDs[i]] = adapter.Inspect(context.Background(), active)
}
a.replaceInspections(inspections)
}
func (a *Application) setTargetDesired(intent Intent, publish func(Result)) Result {
selectedIndex := -1
for i, id := range a.targetIDs {
if id == intent.TargetID {
selectedIndex = i
break
}
}
if selectedIndex < 0 {
return Result{Snapshot: a.Snapshot(), Err: fmt.Errorf("Target %q does not exist", intent.TargetID)}
}
selected := a.targets[selectedIndex]
active := a.activeProfile()
if intent.Enabled && active == nil {
return Result{Snapshot: a.Snapshot(), Err: errors.New("no Active Profile; open Profile management before enabling a Target")}
}
if warning := selected.Confirmation(intent.Enabled); warning != "" && !intent.Confirm {
confirmation := &Confirmation{Title: "Confirm " + a.targetDescriptions[selectedIndex].Name + " Target change", Summary: warning,
ConfirmLabel: "Continue", CancelLabel: "Cancel", HighImpact: true}
return Result{Snapshot: a.Snapshot(), Confirmation: warning, ConfirmationRequest: confirmation,
Status: OperationStatus{Kind: StatusConfirmation, Summary: confirmation.Title, Detail: confirmation.Summary, TargetID: intent.TargetID}}
}
if publish != nil {
publish(Result{Snapshot: a.Snapshot(), Progress: []TargetProgress{{TargetID: intent.TargetID, Applying: true, Phase: TargetProgressRunning, State: a.inspection(intent.TargetID).State}},
Status: OperationStatus{Kind: StatusProgress, Summary: "Updating " + a.targetDescriptions[selectedIndex].Name + " Target", TargetID: intent.TargetID}})
}
result := selected.SetDesired(context.Background(), intent.Enabled, active)
inspection := result.Inspection
if result.Err != nil {
inspection = selected.Inspect(context.Background(), active)
}
a.replaceInspection(intent.TargetID, inspection)
outcome, detail := "success", inspection.Evidence
if result.Err != nil {
outcome, detail = "failure", result.Err.Error()
}
if a.logger != nil {
operation := "disable"
if intent.Enabled {
operation = "enable"
}
_ = a.logger.Write(LogEntry{Time: time.Now(), Operation: operation, Target: intent.TargetID, Result: outcome, Detail: RedactLogDetail(detail)})
}
notice := ""
progressPhase := TargetProgressSucceeded
status := OperationStatus{Kind: StatusSuccess, TargetID: intent.TargetID}
if result.Err == nil {
notice = fmt.Sprintf("%s Target is now %s", a.targetDescriptions[selectedIndex].Name, inspection.State)
status.Summary = notice
} else {
progressPhase = TargetProgressFailed
status.Kind = StatusError
status.Summary = a.targetDescriptions[selectedIndex].Name + " Target update failed"
status.Detail = result.Err.Error()
}
progress := TargetProgress{TargetID: intent.TargetID, Phase: progressPhase, State: inspection.State, Evidence: inspection.Evidence, Err: result.Err}
return Result{Snapshot: a.Snapshot(), Notice: notice, Status: status, Progress: []TargetProgress{progress}, FirstFailedTargetID: map[bool]string{true: intent.TargetID}[result.Err != nil], Err: result.Err}
}
func create(state profile.State, input profile.Profile) (profile.State, string, error) {
p, err := profile.ValidateAndNormalize(input)
if err != nil {
return state, "", err
}
if index(state, p.Name) >= 0 {
return state, "", fmt.Errorf("a Profile named %q already exists", p.Name)
}
next := cloneState(state)
next.Profiles = append(next.Profiles, p)
return next, fmt.Sprintf("Created Profile %q", p.Name), nil
}
func activate(state profile.State, name string) (profile.State, string, error) {
if index(state, name) < 0 {
return state, "", fmt.Errorf("Profile %q does not exist", name)
}
next := cloneState(state)
next.ActiveProfile = name
return next, fmt.Sprintf("Activated Profile %q", name), nil
}
func update(state profile.State, originalName string, input profile.Profile) (profile.State, string, error) {
profileIndex := index(state, originalName)
if profileIndex < 0 {
return state, "", fmt.Errorf("Profile %q does not exist", originalName)
}
p, err := profile.ValidateAndNormalize(input)
if err != nil {
return state, "", err
}
if other := index(state, p.Name); other >= 0 && other != profileIndex {
return state, "", fmt.Errorf("a Profile named %q already exists", p.Name)
}
next := cloneState(state)
next.Profiles[profileIndex] = p
if next.ActiveProfile == originalName {
next.ActiveProfile = p.Name
}
return next, fmt.Sprintf("Updated Profile %q", p.Name), nil
}
func deleteProfile(state profile.State, name string, confirm bool) (profile.State, string, error) {
profileIndex := index(state, name)
if profileIndex < 0 {
return state, "", fmt.Errorf("Profile %q does not exist", name)
}
if len(state.Profiles) == 1 && !confirm {
return state, "", errors.New("deleting the final Profile requires confirmation")
}
next := cloneState(state)
next.Profiles = append(next.Profiles[:profileIndex], next.Profiles[profileIndex+1:]...)
if next.ActiveProfile == name {
next.ActiveProfile = ""
}
return next, fmt.Sprintf("Deleted Profile %q", name), nil
}
func index(state profile.State, name string) int {
for i := range state.Profiles {
if state.Profiles[i].Name == name {
return i
}
}
return -1
}
func cloneState(state profile.State) profile.State {
state.Profiles = append([]profile.Profile(nil), state.Profiles...)
for i := range state.Profiles {
state.Profiles[i].Bypass = append([]string(nil), state.Profiles[i].Bypass...)
}
return state
}