diff --git a/client.go b/client.go index 6c2a0e9..9bb91d2 100644 --- a/client.go +++ b/client.go @@ -104,6 +104,7 @@ func (a apiUser) toUser() *User { ChapterID: a.ChapterID, Department: departmentOf(a.CustomProps), AppointmentTypes: optionLabels(a.CustomProps, "appointment-type"), + IsRep: isRep(a.CustomProps), } } diff --git a/server.go b/server.go index 2adfe4c..073fc42 100644 --- a/server.go +++ b/server.go @@ -3,8 +3,8 @@ package main import ( "context" "crypto/subtle" - "fmt" "errors" + "fmt" "log" "net/http" "strconv" @@ -14,11 +14,12 @@ import ( // User is the subset of an ST user record this worker cares about. type User struct { - ID int64 - Email string - ChapterID int64 - Department string + ID int64 + Email string + ChapterID int64 + Department string AppointmentTypes []string + IsRep bool } // STClient is what the handler needs from the Solidarity Tech API. @@ -30,6 +31,10 @@ type STClient interface { 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 @@ -37,9 +42,9 @@ type Store interface { var ErrLocked = errors.New("department already running") type Server struct { - st STClient - store Store - secret string + st STClient + store Store + secret string eligible map[string]bool // runTimeout caps a single reconcile. runTimeout time.Duration @@ -86,8 +91,7 @@ func (s *Server) Routes() *http.ServeMux { http.Error(w, err.Error(), 500) return } - fmt.Fprintf(w, "%d users\n", len(users)) - counts := map[string]int{} + counts := map[string]int{} eligible := 0 for _, u := range users { if !u.Eligible(s.eligible) { @@ -102,6 +106,17 @@ func (s *Server) Routes() *http.ServeMux { } }) + // 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 } diff --git a/store.go b/store.go index 0f6d4e3..8409b93 100644 --- a/store.go +++ b/store.go @@ -3,41 +3,56 @@ package main import ( "database/sql" "errors" + "fmt" + "strconv" + "strings" "modernc.org/sqlite" sqlite3 "modernc.org/sqlite/lib" ) -const schema = ` -CREATE TABLE IF NOT EXISTS people ( - id INTEGER PRIMARY KEY, - email TEXT, - chapter_id INTEGER, - department TEXT, - appointment_type TEXT, - is_rep INTEGER NOT NULL DEFAULT 0, - synced_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP -); - -CREATE INDEX IF NOT EXISTS people_dept ON people(chapter_id, department); - -CREATE TABLE IF NOT EXISTS sync_state ( - key TEXT PRIMARY KEY, - value TEXT NOT NULL -); -` +var schemaStatements = []string{ + `CREATE TABLE IF NOT EXISTS runs ( + id INTEGER PRIMARY KEY, + chapter_id INTEGER NOT NULL, + department TEXT NOT NULL, + rep_user_id INTEGER NOT NULL, + status TEXT NOT NULL, + started_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + ended_at TEXT + )`, + `CREATE UNIQUE INDEX IF NOT EXISTS one_run_per_dept + ON runs(chapter_id, department) WHERE status = 'running'`, + `CREATE TABLE IF NOT EXISTS people ( + id INTEGER PRIMARY KEY, + email TEXT, + chapter_id INTEGER, + department TEXT, + appointment_type TEXT, + is_rep INTEGER NOT NULL DEFAULT 0, + is_eligible INTEGER NOT NULL DEFAULT 0, + synced_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + )`, + `CREATE INDEX IF NOT EXISTS people_dept ON people(chapter_id, department)`, + `CREATE TABLE IF NOT EXISTS sync_state ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + )`, +} type SQLStore struct{ db *sql.DB } func OpenStore(path string) (*SQLStore, error) { db, err := sql.Open("sqlite", - "file:"+path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)") + "file:"+path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)") if err != nil { return nil, err } db.SetMaxOpenConns(1) - if _, err := db.Exec(schema); err != nil { - return nil, err + for i, stmt := range schemaStatements { + if _, err := db.Exec(stmt); err != nil { + return nil, fmt.Errorf("schema statement %d: %w", i, err) + } } return &SQLStore{db: db}, nil } @@ -45,84 +60,90 @@ func OpenStore(path string) (*SQLStore, error) { 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) - VALUES (?, ?, ?, 'running')`, chapterID, dept, repID) - if err != nil { - var serr *sqlite.Error - if errors.As(err, &serr) && serr.Code() == sqlite3.SQLITE_CONSTRAINT_UNIQUE { + VALUES (?, ?, ?, 'running')`, chapterID, dept, repID) + if err != nil { + var serr *sqlite.Error + if errors.As(err, &serr) && serr.Code() == sqlite3.SQLITE_CONSTRAINT_UNIQUE { + return 0, ErrLocked + } return 0, ErrLocked } - return 0, ErrLocked + return res.LastInsertId() } - return res.LastInsertId() -} -func (s *SQLStore) ReleaseLock(runID int64) error { - _, err := s.db.Exec( - `UPDATE runs SET status='done', ended_at=CURRENT_TIMESTAMP WHERE id=?`, runID) - return err -} - -func (s *SQLStore) SweepStaleLocks() error { - _, err := s.db.Exec( - `UPDATE runs SET status='crashed', ended_at=CURRENT_TIMESTAMP - WHERE status='running'`) - return err -} - -func (s *SQLStore) UpsertPeople(users []User, eligible map[string]bool) error { - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - - stmt, err := tx.Prepare(` - INSERT INTO people (id, email, chapter_id, department, appointment_type, is_rep, synced_at) - VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) - ON CONFLICT(id) DO UPDATE SET - email=excluded.email, - chapter_id=excluded.chapter_id, - department=excluded.department, - appointment_type=excluded.appointment_type, - is_rep=excluded.is_rep, - synced_at=CURRENT_TIMESTAMP`) - if err != nil { - return err - } - defer stmt.Close() - - for _, u := range users { - rep := 0 - if u.IsRep { - rep = 1 + func (s *SQLStore) ReleaseLock(runID int64) error { + _, err := s.db.Exec( + `UPDATE runs SET status='done', ended_at=CURRENT_TIMESTAMP WHERE id=?`, runID) + return err } - _, err := stmt.Exec(u.ID, u.Email, u.ChapterID, - normalizeDept(u.Department), - strings.Join(u.AppointmentTypes, ","), - rep) - if err != nil { - return fmt.Errorf("upserting user %d: %w", u.ID, err) - } - } - return tx.Commit() -} -func (s *SQLStore) LastSync() (int64, error) { - var v string - err := s.db.QueryRow(`SELECT value FROM sync_state WHERE key='users_synced_at'`).Scan(&v) - if err == sql.ErrNoRows { - return 0, nil - } - if err != nil { - return 0, err - } - return strconv.ParseInt(v, 10, 64) -} + func (s *SQLStore) SweepStaleLocks() error { + _, err := s.db.Exec( + `UPDATE runs SET status='crashed', ended_at=CURRENT_TIMESTAMP + WHERE status='running'`) + return err + } -func (s *SQLStore) SetLastSync(ts int64) error { - _, err := s.db.Exec(` - INSERT INTO sync_state (key, value) VALUES ('users_synced_at', ?) - ON CONFLICT(key) DO UPDATE SET value=excluded.value`, - strconv.FormatInt(ts, 10)) - return err -} + func (s *SQLStore) UpsertPeople(users []User, eligible map[string]bool) error { + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + + stmt, err := tx.Prepare(` + INSERT INTO people (id, email, chapter_id, department, appointment_type, is_rep, is_eligible, synced_at) + VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) + ON CONFLICT(id) DO UPDATE SET + email=excluded.email, + chapter_id=excluded.chapter_id, + department=excluded.department, + appointment_type=excluded.appointment_type, + is_rep=excluded.is_rep, + is_rep=excluded.is_rep, + synced_at=CURRENT_TIMESTAMP`) + if err != nil { + return err + } + defer stmt.Close() + + //loop body + + for _, u := range users { + rep, elig := 0, 0 + if u.IsRep { + rep = 1 + } + if u.Eligible(eligible) { + elig = 1 + } + _, err := stmt.Exec(u.ID, u.Email, u.ChapterID, + normalizeDept(u.Department), + strings.Join(u.AppointmentTypes, ","), + rep, elig) + if err != nil { + return fmt.Errorf("upserting user %d: %w", u.ID, err) + } + } + return tx.Commit() + } + + func (s *SQLStore) LastSync() (int64, error) { + var v string + err := s.db.QueryRow(`SELECT value FROM sync_state WHERE key='users_synced_at'`).Scan(&v) + if err == sql.ErrNoRows { + return 0, nil + } + if err != nil { + return 0, err + } + return strconv.ParseInt(v, 10, 64) + } + + func (s *SQLStore) SetLastSync(ts int64) error { + _, err := s.db.Exec(` + INSERT INTO sync_state (key, value) VALUES ('users_synced_at', ?) + ON CONFLICT(key) DO UPDATE SET value=excluded.value`, + strconv.FormatInt(ts, 10)) + return err + }