commit baacf30c6d37bec2e2156e0af02493a827555d8b Author: Miloš Jovanović Date: Tue Sep 8 09:45:23 2026 +0200 First commit, basic SQLite store, worker, and server diff --git a/fakes.go b/fakes.go new file mode 100644 index 0000000..832f86d --- /dev/null +++ b/fakes.go @@ -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 +} + diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..dbec5e2 --- /dev/null +++ b/go.mod @@ -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 +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..5162069 --- /dev/null +++ b/go.sum @@ -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= diff --git a/main.go b/main.go new file mode 100644 index 0000000..b2e32b9 --- /dev/null +++ b/main.go @@ -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())) +} diff --git a/server.go b/server.go new file mode 100644 index 0000000..e5e6f75 --- /dev/null +++ b/server.go @@ -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) +} diff --git a/store.go b/store.go new file mode 100644 index 0000000..e6441a4 --- /dev/null +++ b/store.go @@ -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 +} diff --git a/test.sh b/test.sh new file mode 100755 index 0000000..e971e06 --- /dev/null +++ b/test.sh @@ -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"