First commit, basic SQLite store, worker, and server
This commit is contained in:
Miloš Jovanović
commit
baacf30c6d
7 changed files with 319 additions and 0 deletions
20
fakes.go
Normal file
20
fakes.go
Normal file
|
|
@ -0,0 +1,20 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type fakeST struct {
|
||||
dept string
|
||||
err error
|
||||
chapterID int64
|
||||
}
|
||||
|
||||
func (f *fakeST) GetUser(ctx context.Context, id int64) (*User, error) {
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return &User{ID: id, ChapterID: f.chapterID, Department: f.dept}, nil
|
||||
}
|
||||
|
||||
17
go.mod
Normal file
17
go.mod
Normal file
|
|
@ -0,0 +1,17 @@
|
|||
module ucfa/st-agent-worker
|
||||
|
||||
go 1.27.0
|
||||
|
||||
require (
|
||||
github.com/dustin/go-humanize v1.0.1 // indirect
|
||||
github.com/google/uuid v1.6.0 // indirect
|
||||
github.com/mattn/go-isatty v0.0.24 // indirect
|
||||
github.com/ncruces/go-strftime v1.0.0 // indirect
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
|
||||
golang.org/x/sys v0.47.0 // indirect
|
||||
golang.org/x/time v0.15.0 // indirect
|
||||
modernc.org/libc v1.75.6 // indirect
|
||||
modernc.org/mathutil v1.7.1 // indirect
|
||||
modernc.org/memory v1.12.1 // indirect
|
||||
modernc.org/sqlite v1.58.0 // indirect
|
||||
)
|
||||
22
go.sum
Normal file
22
go.sum
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
|
||||
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
|
||||
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI=
|
||||
github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A=
|
||||
github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
|
||||
github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
|
||||
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
|
||||
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
|
||||
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
||||
modernc.org/libc v1.75.6 h1:yKk8qo+Di4gkmvRboK8ocCqH22FiUCR6jRy2OwtCRus=
|
||||
modernc.org/libc v1.75.6/go.mod h1:bO5o2ztHxBb2rjz0PgdHN0sSMw57CgxGFLZ3Qd/QpVQ=
|
||||
modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
|
||||
modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
|
||||
modernc.org/memory v1.12.1 h1:nFMiWrpStgZczNl6XI9GnIk/rWhYIyHGUaR04pGbp9g=
|
||||
modernc.org/memory v1.12.1/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
|
||||
modernc.org/sqlite v1.58.0 h1:38u40/bwkfM7f0Myhosl+SEMltSDxnGdQf8o6Kjmys0=
|
||||
modernc.org/sqlite v1.58.0/go.mod h1:rsD2CckafgObKC4DhBlGBf+RiHxkc3hINGt1Xw32tVY=
|
||||
34
main.go
Normal file
34
main.go
Normal file
|
|
@ -0,0 +1,34 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
)
|
||||
|
||||
func main() {
|
||||
secret := os.Getenv("WEBHOOK_SECRET")
|
||||
if secret == "" {
|
||||
log.Fatal("WEBHOOK_SECRET not set")
|
||||
}
|
||||
|
||||
dbPath := os.Getenv("DB_PATH")
|
||||
if dbPath == "" {
|
||||
dbPath = "worker.db"
|
||||
}
|
||||
|
||||
store, err := OpenStore(dbPath)
|
||||
if err != nil {
|
||||
log.Fatalf("opening store: %v", err)
|
||||
}
|
||||
|
||||
// Any run still marked running belongs to a dead process.
|
||||
if err := store.SweepStaleLocks(); err != nil {
|
||||
log.Fatalf("sweeping stale locks: %v", err)
|
||||
}
|
||||
|
||||
srv := NewServer(&fakeST{dept: "History", chapterID: 602}, store, secret)
|
||||
|
||||
log.Println("listening on :8080")
|
||||
log.Fatal(http.ListenAndServe(":8080", srv.Routes()))
|
||||
}
|
||||
164
server.go
Normal file
164
server.go
Normal file
|
|
@ -0,0 +1,164 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/subtle"
|
||||
"errors"
|
||||
"log"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// User is the subset of an ST user record this worker cares about.
|
||||
type User struct {
|
||||
ID int64
|
||||
Email string
|
||||
ChapterID int64
|
||||
Department string
|
||||
}
|
||||
|
||||
// STClient is what the handler needs from the Solidarity Tech API.
|
||||
type STClient interface {
|
||||
GetUser(ctx context.Context, id int64) (*User, error)
|
||||
}
|
||||
|
||||
// Store is what the handler needs from persistence.
|
||||
type Store interface {
|
||||
AcquireLock(chapterID int64, dept string, repID int64) (int64, error)
|
||||
ReleaseLock(runID int64) error
|
||||
}
|
||||
|
||||
// ErrLocked is returned by AcquireLock when a run is already in flight
|
||||
// for that department.
|
||||
var ErrLocked = errors.New("department already running")
|
||||
|
||||
type Server struct {
|
||||
st STClient
|
||||
store Store
|
||||
secret string
|
||||
|
||||
// runTimeout caps a single reconcile. At 2 req/sec, size this
|
||||
// against your largest department.
|
||||
runTimeout time.Duration
|
||||
}
|
||||
|
||||
func NewServer(st STClient, store Store, secret string) *Server {
|
||||
return &Server{
|
||||
st: st,
|
||||
store: store,
|
||||
secret: secret,
|
||||
runTimeout: 2 * time.Hour,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) Routes() *http.ServeMux {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("POST /webhook", s.handleWebhook)
|
||||
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
return mux
|
||||
}
|
||||
|
||||
func (s *Server) handleWebhook(w http.ResponseWriter, r *http.Request) {
|
||||
if !s.validSecret(r) {
|
||||
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
|
||||
repID, err := parseRepID(r)
|
||||
if err != nil {
|
||||
log.Printf("bad webhook payload: %v", err)
|
||||
http.Error(w, "bad request", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
rep, err := s.st.GetUser(r.Context(), repID)
|
||||
if err != nil {
|
||||
log.Printf("lookup of rep %d failed: %v", repID, err)
|
||||
http.Error(w, "lookup failed", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
if rep.ChapterID == 0 {
|
||||
log.Printf("rel %d has no chapter; ignoring", repID)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
return
|
||||
}
|
||||
|
||||
dept := normalizeDept(rep.Department)
|
||||
if dept == "" {
|
||||
// An empty department would match every member whose department
|
||||
// is also blank. Never reconcile on it.
|
||||
log.Printf("rep %d has no department; ignoring", repID)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
return
|
||||
}
|
||||
|
||||
runID, err := s.store.AcquireLock(rep.ChapterID, dept, repID)
|
||||
if errors.Is(err, ErrLocked) {
|
||||
// 200, not 409: a non-2xx invites ST to retry, and a retried
|
||||
// duplicate is still a duplicate.
|
||||
log.Printf("dept %q already running; ignoring rep %d", dept, repID)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
log.Printf("acquiring lock for %q failed: %v", dept, err)
|
||||
http.Error(w, "lock failed", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
// Detached from the request: r.Context() is cancelled the moment
|
||||
// this handler returns.
|
||||
go func() {
|
||||
defer func() {
|
||||
if err := s.store.ReleaseLock(runID); err != nil {
|
||||
log.Printf("releasing lock for run %d failed: %v", runID, err)
|
||||
}
|
||||
}()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), s.runTimeout)
|
||||
defer cancel()
|
||||
|
||||
if err := s.reconcile(ctx, runID, rep, dept); err != nil {
|
||||
log.Printf("run %d failed: %v", runID, err)
|
||||
}
|
||||
}()
|
||||
|
||||
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 {
|
||||
got := r.Header.Get("X-Secret")
|
||||
if got == "" {
|
||||
got = r.URL.Query().Get("s")
|
||||
}
|
||||
return subtle.ConstantTimeCompare([]byte(got), []byte(s.secret)) == 1
|
||||
}
|
||||
|
||||
// normalizeDept must be applied identically here and when filtering the
|
||||
// cached roster. Department 1 is a free-text field.
|
||||
func normalizeDept(s string) string {
|
||||
return strings.ToLower(strings.TrimSpace(s))
|
||||
}
|
||||
|
||||
// parseRepID reads the rep's ST user ID from the webhook query string.
|
||||
// ST sends it as {{ user.id }} in the webhook action's URL.
|
||||
func parseRepID(r *http.Request) (int64, error) {
|
||||
raw := r.URL.Query().Get("id")
|
||||
if raw == "" {
|
||||
return 0, errors.New("missing id")
|
||||
}
|
||||
return strconv.ParseInt(raw, 10, 64)
|
||||
}
|
||||
59
store.go
Normal file
59
store.go
Normal file
|
|
@ -0,0 +1,59 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
const schema = `
|
||||
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';
|
||||
`
|
||||
|
||||
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 {
|
||||
return 0, ErrLocked // refine: only on constraint violation
|
||||
}
|
||||
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
|
||||
}
|
||||
3
test.sh
Executable file
3
test.sh
Executable file
|
|
@ -0,0 +1,3 @@
|
|||
ST_KEY='5dc47e1aaa92c7cc8b9a627dd405ea8426475cc268623df0ad20112c47859c85ad4eca5a9cf3441ea1129c114d9126aab17ef220a15f12227611ade0a85b3167'
|
||||
ST=https://api.solidarity.tech/v1
|
||||
curl -sS -g -H "Authorization: Bearer $ST_KEY" -H "Accept: application/json" "$ST/$1"
|
||||
Loading…
Add table
Add a link
Reference in a new issue