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/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/reconcile.go b/reconcile.go index 8e546ec..4e45a52 100644 --- a/reconcile.go +++ b/reconcile.go @@ -17,7 +17,7 @@ func (s *Server) reconcile(ctx context.Context, runID int64, rep *User, dept str } 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]) + runID, len(reps), rep.ChapterID, dept, reps[0]) } agent := reps[0] @@ -26,7 +26,7 @@ func (s *Server) reconcile(ctx context.Context, runID int64, rep *User, dept str return err } -// loop + // loop assigned, skipped, failed := 0, 0, 0 for _, m := range members { @@ -38,7 +38,7 @@ func (s *Server) reconcile(ctx context.Context, runID int64, rep *User, dept str default: } - existing, err := client.ActiveAgent(ctx, m) + existing, err := s.st.ActiveAgent(ctx, m) if err != nil { log.Printf("run %d: checking agent for %d failed: %v", runID, m, err) failed++ @@ -54,7 +54,7 @@ func (s *Server) reconcile(ctx context.Context, runID int64, rep *User, dept str continue } - if err := client.AssignAgent(ctx, m, agent); err != nil { + 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 @@ -67,9 +67,9 @@ func (s *Server) reconcile(ctx context.Context, runID int64, rep *User, dept str } if assigned == 0 && skipped > 0 { log.Printf("WARNING: run %d: assigned nobody; all %d members already have agents", - runID, skipped) + runID, skipped) } log.Printf("run %d: %d assigned, %d skipped, dept %q chapter %d", - runID, assigned, skipped, dept, rep.ChapterID) + runID, assigned, skipped, dept, rep.ChapterID) return nil } diff --git a/server.go b/server.go index 3d9f5fa..c635528 100644 --- a/server.go +++ b/server.go @@ -25,6 +25,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. @@ -37,6 +39,8 @@ type Store interface { 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 diff --git a/store.go b/store.go index 2462922..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 ( @@ -61,14 +70,14 @@ func OpenStore(path string) (*SQLStore, error) { // 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) + 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) + 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) { @@ -179,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 +}