128 lines
3.1 KiB
Go
128 lines
3.1 KiB
Go
package main
|
|
|
|
import (
|
|
"database/sql"
|
|
"errors"
|
|
|
|
"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
|
|
);
|
|
`
|
|
|
|
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)")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
db.SetMaxOpenConns(1)
|
|
if _, err := db.Exec(schema); err != nil {
|
|
return nil, err
|
|
}
|
|
return &SQLStore{db: db}, nil
|
|
}
|
|
|
|
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 {
|
|
return 0, ErrLocked
|
|
}
|
|
return 0, ErrLocked
|
|
}
|
|
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
|
|
}
|
|
_, 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) 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
|
|
}
|