st-agent-worker/server.go

package main

import (
	"context"
	"crypto/subtle"
	"errors"
	"fmt"
	"log"
	"net/http"
	"strconv"
	"strings"
	"time"
)

// User is the subset of an ST user record this worker cares about.
type User struct {
	ID               int64
	Email            string
	ChapterID        int64
	Department       string
	AppointmentTypes []string
	IsRep            bool
}

// STClient is what the handler needs from the Solidarity Tech API.
type STClient interface {
	GetUser(ctx context.Context, id int64) (*User, error)
}

// Store is what the handler needs from persistence.
type Store interface {
	AcquireLock(chapterID int64, dept string, repID int64) (int64, error)
	ReleaseLock(runID int64) error

	UpsertPeople(users []User, eligible map[string]bool) error
	LastSync() (int64, error)
	SetLastSync(ts int64) error
}

// ErrLocked is returned by AcquireLock when a run is already in flight
// for that department.
var ErrLocked = errors.New("department already running")

type Server struct {
	st       STClient
	store    Store
	secret   string
	eligible map[string]bool
	// runTimeout caps a single reconcile.
	runTimeout time.Duration
}

func NewServer(st STClient, store Store, secret string, eligible map[string]bool) *Server {
	return &Server{
		st:         st,
		store:      store,
		secret:     secret,
		eligible:   eligible,
		runTimeout: 2 * time.Hour,
	}
}

// Eligible reports whether any of the user's appointment types is in set.
// Labels are matched, not option values, so a rename in the ST dashboard
// silently disqualifies everyone — log unknown labels during sync.
func (u User) Eligible(set map[string]bool) bool {
	for _, label := range u.AppointmentTypes {
		if set[strings.ToLower(strings.TrimSpace(label))] {
			return true
		}
	}
	return false
}

func (s *Server) Routes() *http.ServeMux {
	mux := http.NewServeMux()
	mux.HandleFunc("POST /webhook", s.handleWebhook)
	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
	})

	// TEMPORARY: remove before deploy.
	mux.HandleFunc("GET /debug/users", func(w http.ResponseWriter, r *http.Request) {
		client, ok := s.st.(*Client)
		if !ok {
			http.Error(w, "not a real client", 500)
			return
		}
		users, err := client.ListUsers(r.Context(), 0)
		if err != nil {
			http.Error(w, err.Error(), 500)
			return
		}
		counts := map[string]int{}
		eligible := 0
		for _, u := range users {
			if !u.Eligible(s.eligible) {
				continue
			}
			eligible++
			counts[fmt.Sprintf("%d|%s", u.ChapterID, normalizeDept(u.Department))]++
		}
		fmt.Fprintf(w, "%d users, %d eligible\n\n", len(users), eligible)
		for k, n := range counts {
			fmt.Fprintf(w, "%6d  %s\n", n, k)
		}
	})

	// TEMPORARY: remove before deploy.
	mux.HandleFunc("POST /debug/sync", func(w http.ResponseWriter, r *http.Request) {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
		defer cancel()
		if err := s.SyncRoster(ctx); err != nil {
			http.Error(w, err.Error(), 500)
			return
		}
		w.WriteHeader(http.StatusOK)
	})

	return mux
}

func (s *Server) handleWebhook(w http.ResponseWriter, r *http.Request) {
	if !s.validSecret(r) {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}

	repID, err := parseRepID(r)
	if err != nil {
		log.Printf("bad webhook payload: %v", err)
		http.Error(w, "bad request", http.StatusBadRequest)
		return
	}

	rep, err := s.st.GetUser(r.Context(), repID)
	if err != nil {
		log.Printf("lookup of rep %d failed: %v", repID, err)
		http.Error(w, "lookup failed", http.StatusInternalServerError)
		return
	}

	if rep.ChapterID == 0 {
		log.Printf("rel %d has no chapter; ignoring", repID)
		w.WriteHeader(http.StatusOK)
		return
	}

	dept := normalizeDept(rep.Department)
	if dept == "" {
		// An empty department would match every member whose department
		// is also blank. Never reconcile on it.
		log.Printf("rep %d has no department; ignoring", repID)
		w.WriteHeader(http.StatusOK)
		return
	}

	runID, err := s.store.AcquireLock(rep.ChapterID, dept, repID)
	if errors.Is(err, ErrLocked) {
		// 200, not 409: a non-2xx invites ST to retry, and a retried
		// duplicate is still a duplicate.
		log.Printf("dept %q already running; ignoring rep %d", dept, repID)
		w.WriteHeader(http.StatusOK)
		return
	}
	if err != nil {
		log.Printf("acquiring lock for %q failed: %v", dept, err)
		http.Error(w, "lock failed", http.StatusInternalServerError)
		return
	}

	// Detached from the request: r.Context() is cancelled the moment
	// this handler returns.
	go func() {
		defer func() {
			if err := s.store.ReleaseLock(runID); err != nil {
				log.Printf("releasing lock for run %d failed: %v", runID, err)
			}
		}()

		ctx, cancel := context.WithTimeout(context.Background(), s.runTimeout)
		defer cancel()

		if err := s.reconcile(ctx, runID, rep, dept); err != nil {
			log.Printf("run %d failed: %v", runID, err)
		}
	}()

	w.WriteHeader(http.StatusOK)
}

// ST doesn't support setting headers so we use the URL

func (s *Server) validSecret(r *http.Request) bool {
	got := r.Header.Get("X-Secret")
	if got == "" {
		got = r.URL.Query().Get("s")
	}
	return subtle.ConstantTimeCompare([]byte(got), []byte(s.secret)) == 1
}

// normalizeDept must be applied identically here and when filtering the
// cached roster. Department 1 is a free-text field.
func normalizeDept(s string) string {
	return strings.ToLower(strings.TrimSpace(s))
}

// parseRepID reads the rep's ST user ID from the webhook query string.
// ST sends it as {{ user.id }} in the webhook action's URL.
func parseRepID(r *http.Request) (int64, error) {
	raw := r.URL.Query().Get("id")
	if raw == "" {
		return 0, errors.New("missing id")
	}
	return strconv.ParseInt(raw, 10, 64)
}