bugabinga.net

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)
}