diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..a577a9b --- /dev/null +++ b/.dockerignore @@ -0,0 +1,7 @@ +.git +.env +*.db +*.db-wal +*.db-shm +bin/ +st-agent-worker diff --git a/.eng.example b/.eng.example new file mode 100644 index 0000000..fd20250 --- /dev/null +++ b/.eng.example @@ -0,0 +1,16 @@ +# generate this in Settings/API/Create API Key +ST_API_KEY= + +# secret for webhook, generate with openssl rand -hex 32 +WEBHOOK_SECRET= + +# set eligible appointment types for the action, comma-separated (for example, senate, recall) +ELIGIBLE_APPOINTMENTS=s + +# location of SQLite database file +DB_PATH=worker.db + + +# set this to 0 to write to Solidarity Tech's API, default is to keep it read_only +DRY_RUN=1 + diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..d8cf456 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,13 @@ +FROM golang:1.27-alpine AS build +WORKDIR /src +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +RUN CGO_ENABLED=0 go build -o /worker . + +FROM alpine:3.20 +RUN apk add --no-cache ca-certificates tzdata +COPY --from=build /worker /worker +RUN mkdir -p /data +EXPOSE 8080 +ENTRYPOINT ["/worker"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..99a17f6 --- /dev/null +++ b/README.md @@ -0,0 +1,133 @@ +# st-agent-worker + +Assigns Solidarity Tech agents to department members automatically. + +When someone is given the Department Rep role in Solidarity Tech, an ST +automation calls this service. The service looks up that person's chapter +(campus) and department, finds every eligible member of the same +chapter/department pair, and assigns the rep as their agent. + +## Rules + +- Only members whose appointment type is Senate or Recall are assigned. + Set `ELIGIBLE_APPOINTMENTS` to change this. +- Matching is on primary chapteri (UCLAFA, BFA, SCFA...) plus Department 1. + A rep at one campus is never assigned members at another. +- Reps are excluded from assignment. Organizer leads (OC members, organizing + staff) assign their agents by hand in the ST web interface. +- **Fill-only**: a member who already has an agent is left alone and + recorded in the `skips` table. No agent is ever reassigned automatically. + See below for consequences of this. + +### Known consequence of fill-only + +Some departments currently have two reps. Whichever rep runs first takes +the whole department; the second gets nobody. The reconciler logs a warning +in both cases that can be seen in the logs and reconciled manually. + +_Potential fix_ In the future, round-robin splitting could be implemented. `RepsFor` already returns all reps for a department sorted by ID, so adding it is a +change to how the agent is picked, and wouldn't require a code restructuring. + +## Architecture + +Four components do the work: + +- `server.go` — HTTP handler. Verifies the shared secret, reads the rep's + ST user ID from the query string, looks up their chapter and department + live, takes a per-department lock, returns 200, and runs the + reconciliation in the background. The 200 is returned before any work + happens, because ST times out and retries otherwise. +- `client.go` — Solidarity Tech API client. Holds the rate limiter and + handles 429 responses. Nothing above it builds a URL. +- `store.go` — SQLite database handle. Four tables: `people` (cached roster), + `runs` (locks and progress), `skips` (members left alone), `sync_state` (last + sync timestamp). +- `reconcile.go` — the loop. Reads department members from the cache, + checks each one's current agent live against the API, assigns or skips. + +Roster data is cached because it changes more slowly, and API rate limits would make a read/write of the whole roster take two hours. Instead, agent assignments are read live on every check, so an assignment made by hand in the dashboard +during a run can be respected right away. See _Rate limits_ below. + +### Locking + +`runs` has a partial unique index on `(chapter_id, department)` where +status is `running`. A second webhook for the same department fails the +insert and exits. On startup, any row still marked `running` belongs to a +dead process and is marked `crashed`. + +### Rate limits + +The ST API allows 60 requests per 30 seconds per API key, which averages +2 per second and permits bursts. The client is configured to match this. + +## Configuration + +| Variable | Meaning | +| --- | --- | +| `ST_API_KEY` | Solidarity Tech API key | +| `WEBHOOK_SECRET` | Shared secret, passed as `?s=` on the webhook URL | +| `DB_PATH` | SQLite file. `/data/worker.db` in the container | +| `DRY_RUN` | Any non-empty value logs writes instead of sending them | +| `ELIGIBLE_APPOINTMENTS` | Comma-separated labels, defaults to `senate,recall` | + +The ST webhook action has no support for custom headers, so the secret +travels in the query string. Do not log raw query strings. + +## Endpoints + +- `POST /webhook?id={user_id}&s={secret}` — triggered by the ST automation + when someone is assigned the Department Rep role. +- `POST /debug/sync?s={secret}` — forces a roster sync. Useful after a bulk + import rather than waiting for the hourly update. +- `GET /healthz` — returns 200 once the server is listening. + +## Health checks + +The startup sync runs before the server binds, and a cold start with an +empty volume pulls the full roster, which takes about 90 seconds. Give the +container a `start_period` of at least 120 seconds or it will be marked +unhealthy before it comes up. + +After deploying, check: + + curl https://$HOST/healthz + +and confirm the log shows a completed sync. Then: + + sqlite3 /data/worker.db " + SELECT COUNT(*) total, SUM(is_eligible) eligible, SUM(is_rep) reps + FROM people;" + +Should give roughly 17,500 total, 14,700 eligible. + +Log lines to watch for: + +- `WARNING: run N: assigned nobody` — every member already had an agent. + Expected for the second rep in a co-repped department, unexpected + otherwise. +- `N eligible users synced with no department` — those members cannot be + assigned to anyone. These are likely Senate or Recall duplicates, or + people we don't have information for. + +## Initial backfill + +The webhook only fires when a rep is newly assigned the role, so reps who +already had it when this was deployed never trigger it. To assign their +departments, call the endpoint for each: + + sqlite3 worker.db "SELECT id FROM people WHERE is_rep = 1;" | while read id; do + curl -s -o /dev/null -X POST "https://$HOST/webhook?id=$id&s=$WEBHOOK_SECRET" + sleep 1 + done + +Run it with `DRY_RUN` set first and count the assignments in the log. + +## Not built yet + +- A nightly sweep to reassign reps. Without it, a rep who steps down keeps + their assignments and their replacement inherits nobody. Every member added + after their department's rep signed up also goes unassigned until + something triggers that department again. **FIRST PRIORITY** +- Schema migrations. The schema is created with `CREATE TABLE IF NOT + EXISTS`, which does not add columns to an existing table. Changing the + schema currently means deleting the database and re-syncing. diff --git a/client.go b/client.go index 9bb91d2..1c3641d 100644 --- a/client.go +++ b/client.go @@ -10,6 +10,7 @@ import ( "net/url" "strconv" "time" + "bytes" "golang.org/x/time/rate" ) @@ -123,7 +124,7 @@ func departmentOf(props map[string]json.RawMessage) string { //optionLabels returne the labels of an option-valued custom-property // ST serialises select and checkbox fields as [{label, value}], and the -// same label can appear more than once. +// same label appears more than one time func optionLabels(props map[string]json.RawMessage, key string) []string { raw, ok := props[key] @@ -163,6 +164,61 @@ func (c *Client) GetUser(ctx context.Context, id int64) (*User, error) { return u.toUser(), nil } +// ActiveAgent returns the user ID of the person's current agent, or 0 if +// they have none. The is_active query param is ignored by the API, so +// rows are filtered here. +func (c *Client) ActiveAgent(ctx context.Context, userID int64) (int64, error) { + q := url.Values{} + q.Set("user_id", strconv.FormatInt(userID, 10)) + q.Set("_limit", "100") + + raw, err := c.do(ctx, http.MethodGet, "/agent_assignments?"+q.Encode(), nil) + if err != nil { + return 0, err + } + + var env struct { + Data []struct { + AgentUserID int64 `json:"agent_user_id"` + IsActive bool `json:"is_active"` + } `json:"data"` + } + if err := json.Unmarshal(raw, &env); err != nil { + return 0, err + } + + for _, a := range env.Data { + if a.IsActive { + return a.AgentUserID, nil + } + } + return 0, nil +} + +// AssignAgent creates an agent assignment. ST deactivates any prior active +// assignment automatically. Not idempotent: a repeat call creates a new row. +func (c *Client) AssignAgent(ctx context.Context, userID, agentID int64) error { + if c.dryRun { + log.Printf("DRY RUN: would assign user %d to agent %d", userID, agentID) + return nil + } + + body, err := json.Marshal(map[string]int64{ + "user_id": userID, + "agent_user_id": agentID, + }) + if err != nil { + return err + } + + _, err = c.do(ctx, http.MethodPost, "/agent_assignments", bytes.NewReader(body)) + if err != nil { + return err + } + log.Printf("assigned user %d to agent %d", userID, agentID) + return nil +} + // ListUsers pages the full user list. If since is non-zero, only users // updated after that Unix timestamp are returned. func (c *Client) ListUsers(ctx context.Context, since int64) ([]User, error) { diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..37e6a7c --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,21 @@ +services: + worker: + build: . + restart: unless-stopped + environment: + ST_API_KEY: ${ST_API_KEY} + WEBHOOK_SECRET: ${WEBHOOK_SECRET} + ELIGIBLE_APPOINTMENTS: ${ELIGIBLE_APPOINTMENTS} + DB_PATH: /data/worker.db + DRY_RUN: ${DRY_RUN:-} + volumes: + - worker-data:/data + healthcheck: + test: ["CMD", "wget", "-qO-", "http://localhost:8080/healthz"] + interval: 30s + timeout: 5s + retries: 3 + start_period: 120s + +volumes: + worker-data: diff --git a/fakes.go b/fakes.go index cbac13c..93e389f 100644 --- a/fakes.go +++ b/fakes.go @@ -6,8 +6,23 @@ import ( type fakeST struct { dept string - err error chapterID int64 + err error + + agents map[int64]int64 // userID → existing agent, 0 or absent = none + assigned map[int64]int64 // recorded writes +} + +func (f *fakeST) ActiveAgent(ctx context.Context, userID int64) (int64, error) { + return f.agents[userID], nil +} + +func (f *fakeST) AssignAgent(ctx context.Context, userID, agentID int64) error { + if f.assigned == nil { + f.assigned = map[int64]int64{} + } + f.assigned[userID] = agentID + return nil } func (f *fakeST) GetUser(ctx context.Context, id int64) (*User, error) { diff --git a/go.mod b/go.mod index 42c2a6d..4f378ed 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module ucfa/st-agent-worker -go 1.27.0 +go 1.26.0 require ( golang.org/x/time v0.16.0 diff --git a/main.go b/main.go index 20287cd..36d5989 100644 --- a/main.go +++ b/main.go @@ -5,6 +5,8 @@ import ( "net/http" "strings" "os" + "context" + "time" ) // ELIGIBLE_APPOINTMENTS is a comma-separated list of appointment type @@ -57,6 +59,26 @@ func main() { } srv := NewServer(NewClient(apiKey, dryRun), store, secret, loadEligible()) + // Prime the cache: a fresh container starts empty. + { + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) + if err := srv.SyncRoster(ctx); err != nil { + log.Printf("initial sync failed: %v", err) + } + cancel() + } + + go func() { + t := time.NewTicker(time.Hour) + defer t.Stop() + for range t.C { + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) + if err := srv.SyncRoster(ctx); err != nil { + log.Printf("scheduled sync failed: %v", err) + } + cancel() + } + }() log.Println("listening on :8080") log.Fatal(http.ListenAndServe(":8080", srv.Routes())) } diff --git a/reconcile.go b/reconcile.go new file mode 100644 index 0000000..4e45a52 --- /dev/null +++ b/reconcile.go @@ -0,0 +1,75 @@ +package main + +import ( + "context" + "log" +) + +func (s *Server) reconcile(ctx context.Context, runID int64, rep *User, dept string) error { + reps, err := s.store.RepsFor(rep.ChapterID, dept) + if err != nil { + return err + } + if len(reps) == 0 { + // Shouldn't happen: the webhook fired for a rep in this department. + log.Printf("run %d: no reps found for chapter %d dept %q", runID, rep.ChapterID, dept) + return nil + } + if len(reps) > 1 { + log.Printf("WARNING: run %d: %d reps for chapter %d dept %q; assigning all to %d (fill-only mode)", + runID, len(reps), rep.ChapterID, dept, reps[0]) + } + agent := reps[0] + + members, err := s.store.EligibleMembers(rep.ChapterID, dept) + if err != nil { + return err + } + + // loop + assigned, skipped, failed := 0, 0, 0 + + for _, m := range members { + + select { + case <-ctx.Done(): + log.Printf("run %d: cancelled after %d assigned, %d skipped", runID, assigned, skipped) + return ctx.Err() + default: + } + + existing, err := s.st.ActiveAgent(ctx, m) + if err != nil { + log.Printf("run %d: checking agent for %d failed: %v", runID, m, err) + failed++ + continue + } + + if existing != 0 { + // Fill-only: never displace an existing agent. + skipped++ + if err := s.store.LogSkip(runID, m, existing, agent); err != nil { + log.Printf("run %d: logging skip for %d failed: %v", runID, m, err) + } + continue + } + + if err := s.st.AssignAgent(ctx, m, agent); err != nil { + log.Printf("run %d: assigning %d to %d failed: %v", runID, m, agent, err) + failed++ + continue + } + assigned++ + + if err := s.store.SetProgress(runID, m); err != nil { + log.Printf("run %d: recording progress failed: %v", runID, err) + } + } + if assigned == 0 && skipped > 0 { + log.Printf("WARNING: run %d: assigned nobody; all %d members already have agents", + runID, skipped) + } + log.Printf("run %d: %d assigned, %d skipped, dept %q chapter %d", + runID, assigned, skipped, dept, rep.ChapterID) + return nil +} diff --git a/server.go b/server.go index 073fc42..cebbb4a 100644 --- a/server.go +++ b/server.go @@ -4,7 +4,6 @@ import ( "context" "crypto/subtle" "errors" - "fmt" "log" "net/http" "strconv" @@ -25,6 +24,8 @@ type User struct { // STClient is what the handler needs from the Solidarity Tech API. type STClient interface { GetUser(ctx context.Context, id int64) (*User, error) + ActiveAgent(ctx context.Context, userID int64) (int64, error) + AssignAgent(ctx context.Context, userID, agentID int64) error } // Store is what the handler needs from persistence. @@ -35,6 +36,10 @@ type Store interface { UpsertPeople(users []User, eligible map[string]bool) error LastSync() (int64, error) SetLastSync(ts int64) error + RepsFor(chapterID int64, dept string) ([]int64, error) + EligibleMembers(chapterID int64, dept string) ([]int64, error) + LogSkip(runID, userID, existingAgent, wouldHaveAssigned int64) error + SetProgress(runID, lastUserID int64) error } // ErrLocked is returned by AcquireLock when a run is already in flight @@ -78,36 +83,12 @@ func (s *Server) Routes() *http.ServeMux { 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) { + if !s.validSecret(r) { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) defer cancel() if err := s.SyncRoster(ctx); err != nil { @@ -189,12 +170,6 @@ func (s *Server) handleWebhook(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } -// reconcile is stubbed until piece 3 lands. -func (s *Server) reconcile(ctx context.Context, runID int64, rep *User, dept string) error { - log.Printf("run %d: would reconcile dept %q for rep %d", runID, dept, rep.ID) - return nil -} - // ST doesn't support setting headers so we use the URL func (s *Server) validSecret(r *http.Request) bool { diff --git a/store.go b/store.go index 8409b93..99ddd73 100644 --- a/store.go +++ b/store.go @@ -18,9 +18,18 @@ var schemaStatements = []string{ department TEXT NOT NULL, rep_user_id INTEGER NOT NULL, status TEXT NOT NULL, + last_user_id INTEGER, started_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, ended_at TEXT )`, + `CREATE TABLE IF NOT EXISTS skips ( + id INTEGER PRIMARY KEY, + run_id INTEGER NOT NULL, + user_id INTEGER NOT NULL, + existing_agent INTEGER NOT NULL, + would_have_assigned INTEGER NOT NULL, + at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + )`, `CREATE UNIQUE INDEX IF NOT EXISTS one_run_per_dept ON runs(chapter_id, department) WHERE status = 'running'`, `CREATE TABLE IF NOT EXISTS people ( @@ -57,6 +66,38 @@ func OpenStore(path string) (*SQLStore, error) { return &SQLStore{db: db}, nil } +// RepsFor returns the user IDs of every rep for a department, ordered by +// ID so assignment is deterministic across runs. +func (s *SQLStore) RepsFor(chapterID int64, dept string) ([]int64, error) { + return s.userIDs(`SELECT id FROM people + WHERE is_rep = 1 AND chapter_id = ? AND department = ? + ORDER BY id`, chapterID, dept) +} + +func (s *SQLStore) EligibleMembers(chapterID int64, dept string) ([]int64, error) { + return s.userIDs(`SELECT id FROM people + WHERE is_eligible = 1 AND is_rep = 0 AND chapter_id = ? AND department = ? + ORDER BY id`, chapterID, dept) +} + +func (s *SQLStore) userIDs(query string, args ...any) ([]int64, error) { + rows, err := s.db.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + out = append(out, id) + } + return out, rows.Err() +} + func (s *SQLStore) AcquireLock(chapterID int64, dept string, repID int64) (int64, error) { res, err := s.db.Exec( `INSERT INTO runs (chapter_id, department, rep_user_id, status) @@ -147,3 +188,19 @@ func (s *SQLStore) AcquireLock(chapterID int64, dept string, repID int64) (int64 strconv.FormatInt(ts, 10)) return err } + + // LogSkip records a member left alone because they already had an agent. +// This is the manual-fix queue: nothing else records what fill-only cost. +func (s *SQLStore) LogSkip(runID, userID, existingAgent, wouldHaveAssigned int64) error { + _, err := s.db.Exec( + `INSERT INTO skips (run_id, user_id, existing_agent, would_have_assigned) + VALUES (?, ?, ?, ?)`, + runID, userID, existingAgent, wouldHaveAssigned) + return err +} + +func (s *SQLStore) SetProgress(runID, lastUserID int64) error { + _, err := s.db.Exec( + `UPDATE runs SET last_user_id = ? WHERE id = ?`, lastUserID, runID) + return err +}