Jaypore CI

server.go (22496B)

---

 package jci
 
 import (
 	"bytes"
 	"crypto/hmac"
 	"crypto/rand"
 	"crypto/sha256"
 	"database/sql"
 	"encoding/base64"
 	"encoding/hex"
 	"encoding/json"
 	_ "embed"
 	"flag"
 	"fmt"
 	"io"
 	"log"
 	"net/http"
 	"os"
 	"strconv"
 	"strings"
 	"time"
 
 	_ "modernc.org/sqlite"
 )
 
 //go:embed default_runner_script.sh
 var defaultRunnerScript []byte
 
 // Server runs the coordination server: accepts Gitea webhooks and serves runner poll requests.
 //
 // Configuration env vars:
 //
 //	GITEA_HOST                        e.g. gitea.example.com
 //	GITEA_USER                        service account username
 //	GITEA_TOKEN                       token with read/write access to the target repos
 //	GITEA_WEBHOOK_SECRET              HMAC-SHA256 secret shared with Gitea
 //	JCI_RUNNER_SECRET                  shared secret that runners authenticate with
 //	JCI_MAX_JOBS                      max concurrent jobs per runner (default 4)
 //	JCI_JOB_TIMEOUT                   server-wide job timeout (default 60m)
 //	JCI_LISTEN                        listen address (default :8080)
 //	JCI_DB                            SQLite DB path (default /var/lib/jci/server.db)
 //	JCI_RUNNER_SCRIPT                 path to a shell script to send to runners (default: embedded default_runner_script.sh)
 //	JCI_RUNNER_IMAGE                  container image runners should use (default: python:3.11)
 //	JCI_JOB_ALLOW_DOCKER_FOR_REPOS    comma-separated list of owner/repo that get the Docker socket mounted (e.g. myorg/myrepo,myorg/other)
 //
 // Flags (override env vars):
 //
 //	--ip    bind IP  (default "" = all interfaces)
 //	--port  bind port (default 8080)
 func Server(args []string) error {
 	fs := flag.NewFlagSet("server", flag.ContinueOnError)
 	flagIP := fs.String("ip", "", "IP address to listen on (default: all interfaces)")
 	flagPort := fs.String("port", "", "port to listen on (default: 8080)")
 	if err := fs.Parse(args); err != nil {
 		return err
 	}
 
 	cfg, err := loadServerConfig()
 	if err != nil {
 		return err
 	}
 
 	// Flags override env vars
 	if *flagIP != "" || *flagPort != "" {
 		ip := *flagIP
 		port := *flagPort
 		if port == "" {
 			port = "8080"
 		}
 		cfg.listen = ip + ":" + port
 	}
 
 	db, err := openServerDB(cfg.dbPath)
 	if err != nil {
 		return fmt.Errorf("open db: %w", err)
 	}
 	defer db.Close()
 
 	log.Printf("verifying Gitea token against %s …", cfg.giteaHost)
 	if err := verifyGiteaToken(cfg); err != nil {
 		return fmt.Errorf("Gitea token check failed: %w", err)
 	}
 	log.Printf("Gitea token OK")
 
 	srv := &coordinationServer{cfg: cfg, db: db}
 
 	mux := http.NewServeMux()
 	mux.HandleFunc("/webhook", srv.handleWebhook)
 	mux.HandleFunc("/poll", srv.handlePoll)
 
 	log.Printf("jci server listening on %s", cfg.listen)
 	return http.ListenAndServe(cfg.listen, mux)
 }
 
 // ---------------------------------------------------------------------------
 // Config
 // ---------------------------------------------------------------------------
 
 type serverConfig struct {
 	giteaHost            string
 	giteaUser            string
 	giteaToken           string
 	webhookSecret        string
 	runnerSecret         string
 	maxJobsPerRunner     int
 	jobTimeout           time.Duration
 	listen               string
 	dbPath               string
 	runnerScript         string // base64-encoded script to send to runners
 	runnerImage          string // container image runners should use
 	allowDockerForRepos  map[string]bool // set of "owner/repo" keys
 }
 
 func loadServerConfig() (serverConfig, error) {
 	cfg := serverConfig{
 		maxJobsPerRunner:    4,
 		jobTimeout:          60 * time.Minute,
 		listen:              ":8080",
 		dbPath:              "/var/lib/jci/server.db",
 		runnerImage:         "python:3.11",
 		allowDockerForRepos: map[string]bool{},
 	}
 	cfg.giteaHost = os.Getenv("GITEA_HOST")
 	cfg.giteaUser = os.Getenv("GITEA_USER")
 	cfg.giteaToken = os.Getenv("GITEA_TOKEN")
 	cfg.webhookSecret = os.Getenv("GITEA_WEBHOOK_SECRET")
 	cfg.runnerSecret = os.Getenv("JCI_RUNNER_SECRET")
 
 	missing := []string{}
 	for _, pair := range [][2]string{
 		{"GITEA_HOST", cfg.giteaHost}, {"GITEA_USER", cfg.giteaUser},
 		{"GITEA_TOKEN", cfg.giteaToken}, {"GITEA_WEBHOOK_SECRET", cfg.webhookSecret},
 		{"JCI_RUNNER_SECRET", cfg.runnerSecret},
 	} {
 		if pair[1] == "" {
 			missing = append(missing, pair[0])
 		}
 	}
 	if len(missing) > 0 {
 		return cfg, fmt.Errorf("missing required env vars: %s", strings.Join(missing, ", "))
 	}
 
 	if v := os.Getenv("JCI_MAX_JOBS"); v != "" {
 		n, err := strconv.Atoi(v)
 		if err != nil {
 			return cfg, fmt.Errorf("JCI_MAX_JOBS must be an integer: %w", err)
 		}
 		cfg.maxJobsPerRunner = n
 	}
 	if v := os.Getenv("JCI_JOB_TIMEOUT"); v != "" {
 		d, err := time.ParseDuration(v)
 		if err != nil {
 			return cfg, fmt.Errorf("JCI_JOB_TIMEOUT: %w", err)
 		}
 		cfg.jobTimeout = d
 	}
 	if v := os.Getenv("JCI_LISTEN"); v != "" {
 		cfg.listen = v
 	}
 	if v := os.Getenv("JCI_DB"); v != "" {
 		cfg.dbPath = v
 	}
 	if v := os.Getenv("JCI_RUNNER_IMAGE"); v != "" {
 		cfg.runnerImage = v
 	}
 
 	// Load runner script: use file from JCI_RUNNER_SCRIPT if set, else embedded default.
 	scriptBytes := defaultRunnerScript
 	if v := os.Getenv("JCI_RUNNER_SCRIPT"); v != "" {
 		b, err := os.ReadFile(v)
 		if err != nil {
 			return cfg, fmt.Errorf("JCI_RUNNER_SCRIPT: %w", err)
 		}
 		scriptBytes = b
 	}
 	cfg.runnerScript = base64.StdEncoding.EncodeToString(scriptBytes)
 
 	// Parse allow-docker repos: "owner/repo,owner2/repo2" → set.
 	if v := os.Getenv("JCI_JOB_ALLOW_DOCKER_FOR_REPOS"); v != "" {
 		for _, entry := range strings.Split(v, ",") {
 			entry = strings.TrimSpace(entry)
 			if entry != "" {
 				cfg.allowDockerForRepos[entry] = true
 			}
 		}
 	}
 
 	return cfg, nil
 }
 
 // ---------------------------------------------------------------------------
 // SQLite schema
 // ---------------------------------------------------------------------------
 
 func openServerDB(path string) (*sql.DB, error) {
 	if err := os.MkdirAll(parentDir(path), 0755); err != nil {
 		return nil, err
 	}
 	db, err := sql.Open("sqlite", path)
 	if err != nil {
 		return nil, err
 	}
 	_, err = db.Exec(`
 		CREATE TABLE IF NOT EXISTS runners (
 			runner_id  TEXT PRIMARY KEY,
 			last_seen  DATETIME NOT NULL
 		);
 		CREATE TABLE IF NOT EXISTS jobs (
 			job_id             TEXT PRIMARY KEY,
 			repo_owner         TEXT NOT NULL,
 			repo_name          TEXT NOT NULL,
 			commit_sha         TEXT NOT NULL,
 			runner_id          TEXT,
 			assigned_at        DATETIME,
 			expires_at         DATETIME,
 			gitea_token        TEXT NOT NULL DEFAULT '',
 			status_cache       TEXT,
 			cache_until        DATETIME,
 			allow_docker_socket INTEGER NOT NULL DEFAULT 0
 		);
 	`)
 	if err != nil {
 		return db, err
 	}
 	// Migrate existing DB: add allow_docker_socket column if absent.
 	_, err = db.Exec(`ALTER TABLE jobs ADD COLUMN allow_docker_socket INTEGER NOT NULL DEFAULT 0`)
 	if err != nil && !strings.Contains(err.Error(), "duplicate column") {
 		return db, fmt.Errorf("migrate jobs table: %w", err)
 	}
 	return db, nil
 }
 
 // ---------------------------------------------------------------------------
 // Server
 // ---------------------------------------------------------------------------
 
 type coordinationServer struct {
 	cfg serverConfig
 	db  *sql.DB
 }
 
 // ---------------------------------------------------------------------------
 // Webhook handler
 // ---------------------------------------------------------------------------
 
 func (s *coordinationServer) handleWebhook(w http.ResponseWriter, r *http.Request) {
 	log.Printf("webhook: incoming request from %s", r.RemoteAddr)
 	if r.Method != http.MethodPost {
 		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
 		return
 	}
 	body, err := io.ReadAll(r.Body)
 	if err != nil {
 		http.Error(w, "read error", http.StatusBadRequest)
 		return
 	}
 
 	sig := r.Header.Get("X-Gitea-Signature")
 	if !checkHMAC(body, sig, s.cfg.webhookSecret) {
 		log.Printf("webhook: HMAC check failed from %s (sig=%q)", r.RemoteAddr, sig)
 		http.Error(w, "invalid signature", http.StatusBadRequest)
 		return
 	}
 
 	var payload struct {
 		After string `json:"after"`
 		Repo  struct {
 			Owner struct{ Login string } `json:"owner"`
 			Name  string                 `json:"name"`
 		} `json:"repository"`
 	}
 	if err := json.Unmarshal(body, &payload); err != nil || payload.After == "" {
 		log.Printf("webhook: ignoring non-push event from %s (no 'after' field)", r.RemoteAddr)
 		w.WriteHeader(http.StatusOK) // not a push event; ignore
 		return
 	}
 
 	owner, name, commit := payload.Repo.Owner.Login, payload.Repo.Name, payload.After
 
 	// Deduplicate on commit SHA
 	var count int
 	_ = s.db.QueryRow(`SELECT COUNT(*) FROM jobs WHERE commit_sha = ?`, commit).Scan(&count)
 	if count > 0 {
 		log.Printf("webhook: duplicate commit %s/%s@%.12s — already queued, ignoring", owner, name, commit)
 		w.WriteHeader(http.StatusOK)
 		return
 	}
 
 	allowDocker := 0
 	if s.cfg.allowDockerForRepos[owner+"/"+name] {
 		allowDocker = 1
 	}
 
 	_, err = s.db.Exec(
 		`INSERT INTO jobs (job_id, repo_owner, repo_name, commit_sha, status_cache, allow_docker_socket) VALUES (?,?,?,?,'pending',?)`,
 		randomID(), owner, name, commit, allowDocker,
 	)
 	if err != nil {
 		log.Printf("webhook: insert job: %v", err)
 		http.Error(w, "db error", http.StatusInternalServerError)
 		return
 	}
 	log.Printf("webhook: queued %s/%s@%.12s", owner, name, commit)
 	w.WriteHeader(http.StatusOK)
 }
 
 // ---------------------------------------------------------------------------
 // Poll handler
 // ---------------------------------------------------------------------------
 
 func (s *coordinationServer) handlePoll(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
 		return
 	}
 	var req struct {
 		RunnerID string `json:"runner_id"`
 		Secret   string `json:"secret"`
 	}
 	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
 		http.Error(w, "bad request", http.StatusBadRequest)
 		return
 	}
 	if req.Secret != s.cfg.runnerSecret {
 		http.Error(w, "forbidden", http.StatusForbidden)
 		return
 	}
 
 	// Auto-register / heartbeat
 	var knownRunner int
 	_ = s.db.QueryRow(`SELECT COUNT(*) FROM runners WHERE runner_id = ?`, req.RunnerID).Scan(&knownRunner)
 	_, _ = s.db.Exec(
 		`INSERT INTO runners (runner_id, last_seen) VALUES (?,?)
 		 ON CONFLICT(runner_id) DO UPDATE SET last_seen=excluded.last_seen`,
 		req.RunnerID, time.Now().UTC().Format(time.RFC3339),
 	)
 	if knownRunner == 0 {
 		log.Printf("new worker registered: %s", req.RunnerID)
 	} else {
 		log.Printf("worker heartbeat: %s", req.RunnerID)
 	}
 
 	// Clean expired jobs; finalise completed jobs for this runner
 	s.reapExpiredJobs()
 	s.syncCompletedJobsForRunner(req.RunnerID)
 
 	// Capacity check
 	var activeCount int
 	_ = s.db.QueryRow(`SELECT COUNT(*) FROM jobs WHERE runner_id = ?`, req.RunnerID).Scan(&activeCount)
 	if activeCount >= s.cfg.maxJobsPerRunner {
 		log.Printf("worker %s at capacity (%d/%d jobs); asking it to back off", req.RunnerID, activeCount, s.cfg.maxJobsPerRunner)
 		w.Header().Set("Retry-After", "5")
 		w.WriteHeader(http.StatusTooManyRequests)
 		return
 	}
 
 	// Pick one unassigned job using round-robin across repos:
 	// choose the repo/owner pair with the fewest currently-assigned jobs
 	// (across all runners), then take its oldest (lowest rowid) unassigned job.
 	// This prevents any single repo from monopolising the queue.
 	var jobID, repoOwner, repoName, commitSHA string
 	var allowDockerSocket int
 	err := s.db.QueryRow(`
 		SELECT j.job_id, j.repo_owner, j.repo_name, j.commit_sha, j.allow_docker_socket
 		FROM jobs j
 		WHERE j.runner_id IS NULL
 		ORDER BY (
 			SELECT COUNT(*) FROM jobs a
 			WHERE a.runner_id IS NOT NULL
 			  AND a.repo_owner = j.repo_owner
 			  AND a.repo_name  = j.repo_name
 		) ASC, j.rowid ASC
 		LIMIT 1
 	`).Scan(&jobID, &repoOwner, &repoName, &commitSHA, &allowDockerSocket)
 	if err == sql.ErrNoRows {
 		log.Printf("worker %s polled: no pending jobs", req.RunnerID)
 		w.Header().Set("Content-Type", "application/json")
 		json.NewEncoder(w).Encode(map[string]any{"job": nil})
 		return
 	} else if err != nil {
 		http.Error(w, "db error", http.StatusInternalServerError)
 		return
 	}
 
 	now := time.Now().UTC()
 	expiresAt := now.Add(s.cfg.jobTimeout)
 	_, _ = s.db.Exec(
 		`UPDATE jobs SET runner_id=?, assigned_at=?, expires_at=?, status_cache='running' WHERE job_id=?`,
 		req.RunnerID, now.Format(time.RFC3339), expiresAt.Format(time.RFC3339), jobID,
 	)
 
 	s.setGiteaCommitStatus(repoOwner, repoName, commitSHA, "running", "jci: job assigned")
 	log.Printf("job %s assigned to worker %s (%s/%s@%.12s) allowDocker=%v", jobID, req.RunnerID, repoOwner, repoName, commitSHA, allowDockerSocket == 1)
 
 	cloneURL := fmt.Sprintf("https://%s:%s@%s/%s/%s", s.cfg.giteaUser, s.cfg.giteaToken, s.cfg.giteaHost, repoOwner, repoName)
 	w.Header().Set("Content-Type", "application/json")
 	json.NewEncoder(w).Encode(map[string]any{
 		"job": map[string]any{
 			"job_id":              jobID,
 			"clone_url":           cloneURL,
 			"commit_sha":          commitSHA,
 			"repo_owner":          repoOwner,
 			"repo_name":           repoName,
 			"timeout_seconds":     int(s.cfg.jobTimeout.Seconds()),
 			"image":               s.cfg.runnerImage,
 			"script":              s.cfg.runnerScript,
 			"allow_docker_socket": allowDockerSocket == 1,
 		},
 	})
 }
 
 // ---------------------------------------------------------------------------
 // Gitea API helpers
 // ---------------------------------------------------------------------------
 
 // verifyGiteaToken confirms the token is valid by listing repos accessible to it.
 func verifyGiteaToken(cfg serverConfig) error {
 	url := fmt.Sprintf("https://%s/api/v1/repos/search?limit=1", cfg.giteaHost)
 	log.Printf("gitea: verifying token for host %s", cfg.giteaHost)
 	resp, err := giteaCall("GET", url, cfg.giteaToken, nil)
 	if err != nil {
 		return err
 	}
 	if resp.status != 200 {
 		return fmt.Errorf("GET %s returned %d — check GITEA_TOKEN", url, resp.status)
 	}
 	return nil
 }
 
 func (s *coordinationServer) setGiteaCommitStatus(owner, repo, commitSHA, state, description string) {
 	url := fmt.Sprintf("https://%s/api/v1/repos/%s/%s/statuses/%s", s.cfg.giteaHost, owner, repo, commitSHA)
 	body, _ := json.Marshal(map[string]string{
 		"state":       state,
 		"description": description,
 		"context":     "jci",
 	})
 	resp, err := giteaCall("POST", url, s.cfg.giteaToken, body)
 	if err != nil {
 		log.Printf("set commit status %s/%s@%.12s → %s: error: %v", owner, repo, commitSHA, state, err)
 		return
 	}
 	log.Printf("set commit status %s/%s@%.12s → %s: HTTP %d: %s", owner, repo, commitSHA, state, resp.status, resp.body)
 }
 
 // checkJobStatusOnGitea reads status.txt from the latest refs/jci-runs/<commit>/* on Gitea.
 // Returns "ok", "err", "running", or "" (unknown/not found).
 func (s *coordinationServer) checkJobStatusOnGitea(owner, repo, commitSHA string) string {
 	// List matching refs
 	url := fmt.Sprintf("https://%s/api/v1/repos/%s/%s/git/refs/jci-runs/%s", s.cfg.giteaHost, owner, repo, commitSHA)
 	resp, err := giteaCall("GET", url, s.cfg.giteaToken, nil)
 	if err != nil {
 		log.Printf("checkJobStatus %s/%s@%.12s: list refs error: %v", owner, repo, commitSHA, err)
 		return ""
 	}
 	if resp.status != 200 {
 		log.Printf("checkJobStatus %s/%s@%.12s: list refs HTTP %d", owner, repo, commitSHA, resp.status)
 		return ""
 	}
 	var refs []struct {
 		Object struct{ SHA string } `json:"object"`
 	}
 	if err := json.Unmarshal(resp.body, &refs); err != nil {
 		log.Printf("checkJobStatus %s/%s@%.12s: decode refs: %v", owner, repo, commitSHA, err)
 		return ""
 	}
 	if len(refs) == 0 {
 		log.Printf("checkJobStatus %s/%s@%.12s: no jci-runs refs found yet", owner, repo, commitSHA)
 		return ""
 	}
 	runCommitSHA := refs[len(refs)-1].Object.SHA
 
 	// Get tree
 	treeURL := fmt.Sprintf("https://%s/api/v1/repos/%s/%s/git/trees/%s", s.cfg.giteaHost, owner, repo, runCommitSHA)
 	treeresp, err := giteaCall("GET", treeURL, s.cfg.giteaToken, nil)
 	if err != nil {
 		log.Printf("checkJobStatus %s/%s@%.12s: get tree error: %v", owner, repo, commitSHA, err)
 		return ""
 	}
 	if treeresp.status != 200 {
 		log.Printf("checkJobStatus %s/%s@%.12s: get tree HTTP %d", owner, repo, commitSHA, treeresp.status)
 		return ""
 	}
 	var tree struct {
 		Tree []struct {
 			Path string `json:"path"`
 			SHA  string `json:"sha"`
 			Type string `json:"type"`
 		} `json:"tree"`
 	}
 	if err := json.Unmarshal(treeresp.body, &tree); err != nil {
 		return ""
 	}
 	var blobSHA string
 	for _, e := range tree.Tree {
 		if e.Path == "status.txt" && e.Type == "blob" {
 			blobSHA = e.SHA
 			break
 		}
 	}
 	if blobSHA == "" {
 		log.Printf("checkJobStatus %s/%s@%.12s: status.txt not found in tree %s", owner, repo, commitSHA, runCommitSHA)
 		return ""
 	}
 
 	blobURL := fmt.Sprintf("https://%s/api/v1/repos/%s/%s/git/blobs/%s", s.cfg.giteaHost, owner, repo, blobSHA)
 	blobResp, err := giteaCall("GET", blobURL, s.cfg.giteaToken, nil)
 	if err != nil {
 		log.Printf("checkJobStatus %s/%s@%.12s: get blob error: %v", owner, repo, commitSHA, err)
 		return ""
 	}
 	if blobResp.status != 200 {
 		log.Printf("checkJobStatus %s/%s@%.12s: get blob HTTP %d", owner, repo, commitSHA, blobResp.status)
 		return ""
 	}
 	var blob struct {
 		Content  string `json:"content"`
 		Encoding string `json:"encoding"`
 	}
 	if err := json.Unmarshal(blobResp.body, &blob); err != nil {
 		return ""
 	}
 	content := blob.Content
 	if blob.Encoding == "base64" {
 		decoded, err := base64.StdEncoding.DecodeString(strings.ReplaceAll(content, "\n", ""))
 		if err != nil {
 			return ""
 		}
 		content = string(decoded)
 	}
 	content = strings.TrimSpace(content)
 	if content == "ok" || content == "err" || content == "running" {
 		log.Printf("checkJobStatus %s/%s@%.12s: status.txt = %q", owner, repo, commitSHA, content)
 		return content
 	}
 	log.Printf("checkJobStatus %s/%s@%.12s: unrecognised status.txt content %q", owner, repo, commitSHA, content)
 	return ""
 }
 
 // ---------------------------------------------------------------------------
 // Background maintenance
 // ---------------------------------------------------------------------------
 
 func (s *coordinationServer) reapExpiredJobs() {
 	rows, err := s.db.Query(
 		`SELECT job_id, repo_owner, repo_name, commit_sha FROM jobs WHERE expires_at IS NOT NULL AND expires_at < ?`,
 		time.Now().UTC().Format(time.RFC3339),
 	)
 	if err != nil {
 		return
 	}
 	type expiredJob struct{ jobID, owner, repo, commit string }
 	var expired []expiredJob
 	for rows.Next() {
 		var j expiredJob
 		if rows.Scan(&j.jobID, &j.owner, &j.repo, &j.commit) == nil {
 			expired = append(expired, j)
 		}
 	}
 	rows.Close()
 
 	for _, j := range expired {
 		log.Printf("job %s timed out", j.jobID)
 		s.setGiteaCommitStatus(j.owner, j.repo, j.commit, "failure", "jci: job timed out")
 		_, _ = s.db.Exec(`DELETE FROM jobs WHERE job_id = ?`, j.jobID)
 	}
 }
 
 func (s *coordinationServer) syncCompletedJobsForRunner(runnerID string) {
 	rows, err := s.db.Query(
 		`SELECT job_id, repo_owner, repo_name, commit_sha, COALESCE(status_cache,''), COALESCE(cache_until,'')
 		 FROM jobs WHERE runner_id = ?`, runnerID,
 	)
 	if err != nil {
 		return
 	}
 	type jobRow struct {
 		jobID, owner, repo, commit, statusCache, cacheUntilStr string
 	}
 	var jobs []jobRow
 	for rows.Next() {
 		var j jobRow
 		if rows.Scan(&j.jobID, &j.owner, &j.repo, &j.commit, &j.statusCache, &j.cacheUntilStr) == nil {
 			jobs = append(jobs, j)
 		}
 	}
 	rows.Close()
 
 	for _, j := range jobs {
 		status := j.statusCache
 		cacheUntil, _ := time.Parse(time.RFC3339, j.cacheUntilStr)
 		if time.Now().UTC().After(cacheUntil) {
 			log.Printf("sync job %s (%s/%s@%.12s): cache expired, checking Gitea", j.jobID, j.owner, j.repo, j.commit)
 			if fresh := s.checkJobStatusOnGitea(j.owner, j.repo, j.commit); fresh != "" {
 				log.Printf("sync job %s: status %q → %q", j.jobID, status, fresh)
 				status = fresh
 			} else {
 				log.Printf("sync job %s: no fresh status from Gitea, keeping %q", j.jobID, status)
 			}
 			newCacheUntil := time.Now().UTC().Add(15 * time.Second).Format(time.RFC3339)
 			_, _ = s.db.Exec(`UPDATE jobs SET status_cache=?, cache_until=? WHERE job_id=?`, status, newCacheUntil, j.jobID)
 		} else {
 			log.Printf("sync job %s (%s/%s@%.12s): status cache still valid (%q until %s)", j.jobID, j.owner, j.repo, j.commit, status, j.cacheUntilStr)
 		}
 
 		if status == "ok" || status == "err" {
 			giteaState := "success"
 			if status == "err" {
 				giteaState = "failure"
 			}
 			s.setGiteaCommitStatus(j.owner, j.repo, j.commit, giteaState, "jci: job completed")
 			_, _ = s.db.Exec(`DELETE FROM jobs WHERE job_id = ?`, j.jobID)
 			log.Printf("job %s completed: %s", j.jobID, status)
 		}
 	}
 }
 
 // ---------------------------------------------------------------------------
 // Generic Gitea HTTP helper
 // ---------------------------------------------------------------------------
 
 type giteaResponse struct {
 	status int
 	body   []byte
 }
 
 func giteaCall(method, url, token string, body []byte) (giteaResponse, error) {
 	log.Printf("gitea: %s %s", method, url)
 	var bodyReader io.Reader
 	if body != nil {
 		bodyReader = bytes.NewReader(body)
 	}
 	req, err := http.NewRequest(method, url, bodyReader)
 	if err != nil {
 		return giteaResponse{}, err
 	}
 	req.Header.Set("Authorization", "token "+token)
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Accept", "application/json")
 
 	client := &http.Client{Timeout: 15 * time.Second}
 	resp, err := client.Do(req)
 	if err != nil {
 		log.Printf("gitea: %s %s → error: %v", method, url, err)
 		return giteaResponse{}, err
 	}
 	defer resp.Body.Close()
 	data, _ := io.ReadAll(resp.Body)
 	log.Printf("gitea: %s %s → HTTP %d: %s", method, url, resp.StatusCode, data)
 	return giteaResponse{status: resp.StatusCode, body: data}, nil
 }
 
 // ---------------------------------------------------------------------------
 // Tiny utilities
 // ---------------------------------------------------------------------------
 
 func checkHMAC(body []byte, sig, secret string) bool {
 	mac := hmac.New(sha256.New, []byte(secret))
 	mac.Write(body)
 	expected := hex.EncodeToString(mac.Sum(nil))
 	return hmac.Equal([]byte(sig), []byte(expected))
 }
 
 func randomID() string {
 	b := make([]byte, 16)
 	rand.Read(b)
 	return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:])
 }
 
 // parentDir returns the directory component of a file path without importing filepath
 // (avoids a collision with the filepath import in other files in the package).
 func parentDir(path string) string {
 	for i := len(path) - 1; i >= 0; i-- {
 		if path[i] == '/' || path[i] == '\\' {
 			return path[:i]
 		}
 	}
 	return "."
 }