sync mechanism complete
This commit is contained in:
Miloš Jovanović
parent
40f16346f5
commit
d3803dead3
3 changed files with 142 additions and 105 deletions
|
|
@ -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),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
35
server.go
35
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
|
||||
}
|
||||
|
||||
|
|
|
|||
211
store.go
211
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
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue