package api import ( "context" "encoding/json" "errors" "fmt" "io" "net/http" "sort" "strings" "sync" "time" "bugabinga.net/toad/internal/credential" "bugabinga.net/toad/internal/enroll" "bugabinga.net/toad/internal/rollout" "bugabinga.net/toad/internal/store" ) var ( ErrConflict = errors.New("conflict") ErrRequest = errors.New("invalid request") ) type Manager struct { Store store.Store Enrollment enroll.Set Credentials *credential.Store Engine rollout.Engine Now func() time.Time ThemeCSS []byte mu sync.Mutex active map[string]*running wait sync.WaitGroup } type running struct { id string cancel context.CancelFunc } type DeployRequest struct { Digest string `json:"digest"` ExpectedGeneration uint64 `json:"expectedGeneration"` IdempotencyKey string `json:"idempotencyKey"` } // Submit durably accepts one deployment and starts its rollout in the background. func (m *Manager) Submit(service enroll.Service, request DeployRequest) (store.Operation, error) { if !enroll.ValidDigest(request.Digest) { return store.Operation{}, fmt.Errorf("%w: digest must be sha256:<64 hex>", ErrRequest) } if strings.TrimSpace(request.IdempotencyKey) == "" || len(request.IdempotencyKey) > 200 { return store.Operation{}, fmt.Errorf("%w: idempotency key required", ErrRequest) } m.mu.Lock() defer m.mu.Unlock() if m.active == nil { m.active = map[string]*running{} } if existing, ok, err := m.Store.LookupIdempotency(service.Name, request.IdempotencyKey); err != nil { return store.Operation{}, err } else if ok { return m.Store.ReadOperation(service.Name, existing) } if current, busy := m.active[service.Name]; busy { return store.Operation{}, fmt.Errorf("%w: rollout %s is active for %s", ErrConflict, current.id, service.Name) } release, err := m.Store.Lock(service.Name) if err != nil { return store.Operation{}, fmt.Errorf("%w: %s", ErrConflict, err) } state, err := m.Store.ReadState(service.Name) if err != nil { release() return store.Operation{}, err } if state.Generation != request.ExpectedGeneration { release() return store.Operation{}, fmt.Errorf("%w: expected generation %d, current generation %d", ErrConflict, request.ExpectedGeneration, state.Generation) } now := m.Now() reference := service.Reference(request.Digest) operation := store.Operation{ ID: store.NewOperationID(now), Service: service.Name, Phase: store.PhasePreparing, ExpectedGeneration: request.ExpectedGeneration, IdempotencyKey: request.IdempotencyKey, Candidate: store.Revision{ Digest: request.Digest, Reference: reference, Unit: service.RenderUnit(reference), }, Rollback: state.Current, StartedAt: now.UTC(), Deadline: now.Add(service.Probation.DeadlineDuration()).UTC(), } if err := m.Store.WriteOperation(operation); err != nil { release() return store.Operation{}, err } if err := m.Store.RecordIdempotency(service.Name, request.IdempotencyKey, operation.ID); err != nil { release() return store.Operation{}, err } m.launch(service, operation, release) return operation, nil } // launch owns the active marker and the on-disk lock for one running rollout. func (m *Manager) launch(service enroll.Service, operation store.Operation, release func()) { ctx, cancel := context.WithCancel(context.Background()) m.active[service.Name] = &running{id: operation.ID, cancel: cancel} m.wait.Add(1) go func() { defer m.wait.Done() defer release() defer cancel() defer func() { m.mu.Lock() delete(m.active, service.Name) m.mu.Unlock() }() m.Engine.Run(ctx, service, operation) }() } // Abort ends the active probation and lets the rollout restore the retained revision. func (m *Manager) Abort(service string) (string, error) { m.mu.Lock() defer m.mu.Unlock() current, busy := m.active[service] if !busy { return "", fmt.Errorf("%w: no active rollout for %s", ErrConflict, service) } _ = m.Store.AppendLog(service, current.id, "operator requested rollback") current.cancel() return current.id, nil } // Recover resumes or explicitly fails operations interrupted by a crash or host restart. // Deadlines are not extended, so an expired candidate rolls back instead of promoting late. func (m *Manager) Recover() error { m.mu.Lock() defer m.mu.Unlock() if m.active == nil { m.active = map[string]*running{} } for _, name := range m.Enrollment.Names() { service, ok := m.Enrollment.Lookup(name) if !ok { continue } pending, err := m.pendingOperations(name) if err != nil { return err } if len(pending) == 0 { continue } for _, abandoned := range pending[:len(pending)-1] { if err := m.abandon(abandoned, "abandoned by a newer interrupted operation"); err != nil { return err } } resumed := pending[len(pending)-1] release, err := m.Store.Lock(name) if err != nil { return fmt.Errorf("%w: %s", ErrConflict, err) } resume, err := m.reconcile(resumed) if err != nil { release() return err } if !resume { release() continue } _ = m.Store.AppendLog(name, resumed.ID, "resuming interrupted operation in phase "+string(resumed.Phase)) m.launch(service, resumed, release) } return nil } // reconcile verifies persistent state before recovery can mutate a host. // A committed generation names its operation, so a later desired state cannot be // mistaken for this interrupted rollout. func (m *Manager) reconcile(operation store.Operation) (bool, error) { state, err := m.Store.ReadState(operation.Service) if err != nil { return false, err } if state.Generation == operation.ExpectedGeneration && sameRevision(state.Current, operation.Rollback) { return true, nil } committed := state.Generation == operation.ExpectedGeneration+1 && state.OperationID == operation.ID switch operation.Phase { case store.PhasePromoting: if committed && sameRevision(state.Current, &operation.Candidate) && sameRevision(state.Rollback, operation.Rollback) { return false, m.completeRecovered(&operation, store.PhasePromoted, state.Generation, "") } case store.PhaseRollingBack: if committed && operation.Rollback != nil && sameRevision(state.Current, operation.Rollback) { return false, m.completeRecovered(&operation, store.PhaseRolledBack, state.Generation, operation.Reason) } } return false, m.abandon(operation, "recovery refused: persisted state does not match interrupted operation") } func (m *Manager) completeRecovered(operation *store.Operation, phase store.Phase, generation uint64, reason string) error { ended := m.Now().UTC() operation.Phase = phase operation.Generation = generation operation.Reason = reason operation.EndedAt = &ended if err := m.Store.WriteOperation(*operation); err != nil { return err } return m.Store.AppendLog(operation.Service, operation.ID, "recovered committed "+string(phase)) } func sameRevision(left, right *store.Revision) bool { if left == nil || right == nil { return left == right } return *left == *right } func (m *Manager) pendingOperations(service string) ([]store.Operation, error) { ids, err := m.Store.OperationIDs(service) if err != nil { return nil, err } sort.Strings(ids) var pending []store.Operation for _, id := range ids { operation, err := m.Store.ReadOperation(service, id) if err != nil { return nil, err } if !operation.Phase.Terminal() { pending = append(pending, operation) } } return pending, nil } func (m *Manager) abandon(operation store.Operation, reason string) error { ended := m.Now().UTC() operation.Phase = store.PhaseFailed operation.Reason = reason operation.EndedAt = &ended return m.Store.WriteOperation(operation) } // Drain waits for running rollouts, so shutdown never abandons an accepted operation silently. func (m *Manager) Drain() { m.wait.Wait() } // Handler serves the read-only dashboard and token-authenticated consumers. // Token administration is deliberately absent here. func (m *Manager) Handler() http.Handler { mux := http.NewServeMux() mux.HandleFunc("GET /{$}", m.serveDashboard) mux.HandleFunc("GET /assets/{asset}", func(writer http.ResponseWriter, request *http.Request) { if request.PathValue("asset") == "theme.css" { secureWebResponse(writer) writer.Header().Set("Content-Type", "text/css; charset=utf-8") writer.Header().Set("Cache-Control", "public, max-age=300") _, _ = writer.Write(m.ThemeCSS) return } serveWebAsset(writer, request) }) m.serviceRoutes(mux, m.tokenAuthorized) return mux } // LocalHandler serves the Unix socket, where the filesystem permission is the credential. func (m *Manager) LocalHandler() http.Handler { mux := http.NewServeMux() m.serviceRoutes(mux, m.local) mux.Handle("POST /v1/services/{service}/tokens", m.local(m.handleIssueToken)) mux.Handle("GET /v1/services/{service}/tokens", m.local(m.handleListTokens)) mux.Handle("DELETE /v1/services/{service}/tokens/{id}", m.local(m.handleRevokeToken)) return mux } type serviceHandler func(http.ResponseWriter, *http.Request, enroll.Service) func (m *Manager) serviceRoutes(mux *http.ServeMux, guard func(serviceHandler) http.Handler) { mux.Handle("POST /v1/services/{service}/deployments", guard(m.handleDeploy)) mux.Handle("POST /v1/services/{service}/rollback", guard(m.handleRollback)) mux.Handle("GET /v1/services/{service}", guard(m.handleService)) mux.Handle("GET /v1/services/{service}/operations/{operation}", guard(m.handleOperation)) mux.Handle("GET /v1/services/{service}/operations/{operation}/log", guard(m.handleLog)) } func (m *Manager) tokenAuthorized(next serviceHandler) http.Handler { return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { name := request.PathValue("service") service, ok := m.Enrollment.Lookup(name) presented, bearer := strings.CutPrefix(request.Header.Get("Authorization"), "Bearer ") if !ok || !bearer || !m.Credentials.Verify(name, strings.TrimSpace(presented)) { // One response for unknown service and wrong credential, so tokens cannot enumerate services. writeJSON(writer, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) return } next(writer, request, service) }) } func (m *Manager) local(next serviceHandler) http.Handler { return http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { service, ok := m.Enrollment.Lookup(request.PathValue("service")) if !ok { writeJSON(writer, http.StatusNotFound, map[string]string{"error": "unknown service"}) return } next(writer, request, service) }) } func (m *Manager) handleIssueToken(writer http.ResponseWriter, request *http.Request, service enroll.Service) { var body struct { Label string `json:"label"` } decoder := json.NewDecoder(http.MaxBytesReader(writer, request.Body, 4096)) decoder.DisallowUnknownFields() if err := decoder.Decode(&body); err != nil && !errors.Is(err, io.EOF) { writeJSON(writer, http.StatusBadRequest, map[string]string{"error": err.Error()}) return } value, issued, err := m.Credentials.Issue(service.Name, body.Label) if err != nil { writeJSON(writer, http.StatusInternalServerError, map[string]string{"error": err.Error()}) return } writeJSON(writer, http.StatusCreated, map[string]any{ "id": issued.ID, "label": issued.Label, "createdAt": issued.CreatedAt, "token": value, }) } func (m *Manager) handleListTokens(writer http.ResponseWriter, _ *http.Request, service enroll.Service) { listed, err := m.Credentials.List(service.Name) if err != nil { writeJSON(writer, http.StatusInternalServerError, map[string]string{"error": err.Error()}) return } writeJSON(writer, http.StatusOK, listed) } func (m *Manager) handleRevokeToken(writer http.ResponseWriter, request *http.Request, service enroll.Service) { err := m.Credentials.Revoke(service.Name, request.PathValue("id")) switch { case errors.Is(err, credential.ErrUnknownToken): writeJSON(writer, http.StatusNotFound, map[string]string{"error": "unknown token"}) case err != nil: writeJSON(writer, http.StatusInternalServerError, map[string]string{"error": err.Error()}) default: writeJSON(writer, http.StatusOK, map[string]string{"revoked": request.PathValue("id")}) } } func (m *Manager) handleDeploy(writer http.ResponseWriter, request *http.Request, service enroll.Service) { var body DeployRequest decoder := json.NewDecoder(http.MaxBytesReader(writer, request.Body, 4096)) decoder.DisallowUnknownFields() if err := decoder.Decode(&body); err != nil { writeJSON(writer, http.StatusBadRequest, map[string]string{"error": err.Error()}) return } operation, err := m.Submit(service, body) switch { case errors.Is(err, ErrRequest): writeJSON(writer, http.StatusBadRequest, map[string]string{"error": err.Error()}) case errors.Is(err, ErrConflict): state, _ := m.Store.ReadState(service.Name) writeJSON(writer, http.StatusConflict, map[string]any{"error": err.Error(), "generation": state.Generation}) case err != nil: writeJSON(writer, http.StatusInternalServerError, map[string]string{"error": err.Error()}) default: writeJSON(writer, http.StatusAccepted, operation) } } func (m *Manager) handleService(writer http.ResponseWriter, _ *http.Request, service enroll.Service) { state, err := m.Store.ReadState(service.Name) if err != nil { writeJSON(writer, http.StatusInternalServerError, map[string]string{"error": err.Error()}) return } m.mu.Lock() var active string if current, busy := m.active[service.Name]; busy { active = current.id } m.mu.Unlock() writeJSON(writer, http.StatusOK, map[string]any{ "service": service.Name, "generation": state.Generation, "current": state.Current, "rollback": state.Rollback, "activeOperation": active, "updatedAt": state.UpdatedAt, "probationSeconds": service.Probation.DeadlineDuration().Seconds(), }) } func (m *Manager) handleRollback(writer http.ResponseWriter, _ *http.Request, service enroll.Service) { id, err := m.Abort(service.Name) if err != nil { writeJSON(writer, http.StatusConflict, map[string]string{"error": err.Error()}) return } writeJSON(writer, http.StatusAccepted, map[string]string{"operation": id}) } func (m *Manager) handleOperation(writer http.ResponseWriter, request *http.Request, service enroll.Service) { operation, err := m.Store.ReadOperation(service.Name, request.PathValue("operation")) if err != nil { writeJSON(writer, http.StatusNotFound, map[string]string{"error": "unknown operation"}) return } writeJSON(writer, http.StatusOK, operation) } func (m *Manager) handleLog(writer http.ResponseWriter, request *http.Request, service enroll.Service) { data, err := m.Store.ReadLog(service.Name, request.PathValue("operation")) if err != nil { writeJSON(writer, http.StatusNotFound, map[string]string{"error": "unknown operation"}) return } writer.Header().Set("Content-Type", "text/plain; charset=utf-8") writer.Header().Set("X-Content-Type-Options", "nosniff") _, _ = writer.Write(data) } func writeJSON(writer http.ResponseWriter, status int, value any) { writer.Header().Set("Content-Type", "application/json") writer.Header().Set("X-Content-Type-Options", "nosniff") writer.WriteHeader(status) _ = json.NewEncoder(writer).Encode(value) }