ci/main.go

// ci — the dashboard behind ci.hanzo.ai.
//
// It owns no build state. Run truth lives in Hanzo Git (git.hanzo.ai), which
// schedules the jobs and holds every log; this reads that and presents it. The
// alternative — a CI service with its own run database — would put two answers
// to "did the build pass" in the fleet, and the one users look at would be the
// one that can drift. So: git.hanzo.ai is the store, ci.hanzo.ai is the view.
//
// Build and deploy are one event on this fleet, and the page says so. A push
// runs the pipeline this repo publishes; that pipeline's `image` job produces the
// artifact and its `rollout` job writes the tag and digest into hanzoai/universe,
// which cd.hanzo.ai reconciles onto the cluster. So the last jobs of a run ARE
// the deploy, and showing a run without showing whether its pin took effect is
// reporting on the first half of an event.
//
// The fleet view (fleet.go) closes it: for each service it reads what we wrote,
// what was proved, what is declared and what is running, from the four systems
// that own those answers. cd.hanzo.ai remains the place to drive a sync; this is
// the place to see whether one was needed.
//
// Tenancy is the same value everywhere: an org slug. Hanzo Git namespaces repos
// by org, IAM issues that slug in the `owner` claim, and Hanzo CD fences
// projects by it. Filtering here by `org` is therefore the same boundary those
// enforce, not a parallel notion of who-sees-what.
package ci

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"os"
	"os/signal"
	"sort"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"
)

// Serve builds the dashboard's handler and starts the pollers that feed it.
//
// Exported so this surface can be MOUNTED as well as run: hanzoai/cloud carries
// /v1/deploy behind the gateway's IAM identity, and this is the CI half of that
// same delivery plane. A package main cannot be mounted by anything, which is
// why it was the one surface a valid hanzo.id token could not read.
//
// The caller owns the lifetime: cancelling ctx stops the pollers. The returned
// handler is safe to serve immediately — both caches answer "no snapshot yet"
// rather than blocking, so a mount never waits on an upstream fetch.
func Serve(ctx context.Context, logger *slog.Logger, cfg config) http.Handler {
	src := &gitSource{base: cfg.gitBase, token: cfg.gitToken, http: &http.Client{Timeout: 20 * time.Second}}
	cache := &runCache{}
	board := &fleetCache{}

	// The cluster is read through the pod's own ServiceAccount. Failing to dial
	// is not fatal: the run view is unaffected, and the fleet view says it cannot
	// read running state — which is true and visible, where exiting here would
	// take the whole dashboard down over half of one page.
	cl, err := dial(serviceAccount, apiHost)
	if err != nil {
		logger.Warn("cluster identity", "err", err)
	}

	// One poller, one cache. Every viewer reads the same snapshot, so N open
	// dashboards cost Hanzo Git exactly as much as one — a dashboard that
	// fanned each page load into upstream calls is how a status page takes the
	// system it reports on down.
	go poll(ctx, logger, src, cache, cfg)
	go watch(ctx, logger, src, cl, board, cfg)

	return routes(cfg, cache, board)
}

// Main is the standalone server: the same handler, with a listener and shutdown
// around it. cmd/ci is a thin wrapper over this. Named Main rather than Run
// because a Run in this package is a CI run — the domain object — and the
// binary's body is not one.
func Main() {
	logger := slog.New(slog.NewJSONHandler(os.Stderr, nil))

	cfg, err := loadConfig()
	if err != nil {
		logger.Error("config", "err", err)
		os.Exit(1)
	}

	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	srv := &http.Server{
		Addr:              cfg.listen,
		Handler:           Serve(ctx, logger, cfg),
		ReadHeaderTimeout: 10 * time.Second,
	}
	go func() {
		<-ctx.Done()
		sh, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		_ = srv.Shutdown(sh)
	}()

	logger.Info("ci dashboard listening", "addr", cfg.listen, "source", cfg.gitBase, "refresh", cfg.refresh.String())
	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
		logger.Error("serve", "err", err)
		os.Exit(1)
	}
}

// routes is the whole surface, in one place, so that what the scope tests drive
// is what the binary serves. A test that rebuilt a handler of its own would be
// asserting a copy, and the copy is free to stop matching the route it stands in
// for — which for the org boundary is the one thing that must not happen.
//
// Every handler here begins by resolving the viewer, and every list it answers
// with passes through that viewer first. /healthz is the single exception, and
// only because it answers a constant.
// Runs and Fleet are what the two API operations answer.
//
// They were anonymous maps, which is fine for a page this repo also renders and
// wrong for an operation a client dispatches: a caller needs a shape it can
// generate against, and a mounted copy of this surface would otherwise have to
// declare a second one — the two-schema drift HIP-0106 §4.2 names.
//
// Both carry the SAME four facts about the fetch itself, because a board whose
// staleness a reader cannot see is a board that lies quietly.
type Executions struct {
	Executions []Execution `json:"runs"`
	Repos      int         `json:"repos"`
	Orgs       []string    `json:"orgs"`
	FetchedAt  time.Time   `json:"fetchedAt"`
	Stale      bool        `json:"stale"`
	SourceErr  string      `json:"sourceErr"`
}

// Fleet is one row per service: what was written against what is running.
type Pipelines struct {
	Services  []Pipeline `json:"services"`
	Orgs      []string   `json:"orgs"`
	FetchedAt time.Time  `json:"fetchedAt"`
	Stale     bool       `json:"stale"`
	SourceErr string     `json:"sourceErr"`
}

func routes(cfg config, cache *runCache, board *fleetCache) *http.ServeMux {
	mux := http.NewServeMux()
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
		// Liveness only: up means "serving". Readiness deliberately does NOT
		// gate on having a snapshot — a Hanzo Git outage must show as a stale
		// dashboard saying so, not as ci.hanzo.ai disappearing from the LB too.
		writeJSON(w, http.StatusOK, map[string]any{"status": "ok"})
	})
	// Namespaced under /v1/ci. An API path names the app that answers it: /v1/runs
	// and /v1/fleet name neither, and two apps on one host cannot both have them.
	// api.hanzo.ai already carries /v1/deploy for the CD half of this same plane,
	// behind the gateway's IAM identity; this is the shape that lets the CI half
	// join it rather than needing a second front door and a second way to prove
	// who you are.
	mux.HandleFunc("/v1/ci/runs", func(w http.ResponseWriter, r *http.Request) {
		v, ok := requireViewer(w, r, cfg.adminOrg)
		if !ok {
			return
		}
		snap := cache.get()
		writeJSON(w, http.StatusOK, Executions{
			Executions: v.visible(snap.Executions, r.URL.Query().Get("org")),
			Repos:      snap.Repos,
			Orgs:       v.orgs(snap.Executions),
			FetchedAt:  snap.FetchedAt,
			Stale:      snap.stale(cfg.staleAfter),
			SourceErr:  snap.errString(),
		})
	})
	mux.HandleFunc("/v1/ci/fleet", func(w http.ResponseWriter, r *http.Request) {
		v, ok := requireViewer(w, r, cfg.adminOrg)
		if !ok {
			return
		}
		snap := board.get()
		services := v.services(snap.Services, r.URL.Query().Get("org"))
		writeJSON(w, http.StatusOK, Pipelines{
			Services:  services,
			Orgs:      orgsOfServices(services),
			FetchedAt: snap.FetchedAt,
			Stale:     snap.stale(4 * cfg.fleetRefresh),
			SourceErr: snap.errString(),
		})
	})
	mux.HandleFunc("/runs", func(w http.ResponseWriter, r *http.Request) {
		v, ok := requireViewer(w, r, cfg.adminOrg)
		if !ok {
			return
		}
		renderRuns(w, cache.get(), v, r.URL.Query().Get("org"), cfg)
	})
	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/" {
			http.NotFound(w, r)
			return
		}
		v, ok := requireViewer(w, r, cfg.adminOrg)
		if !ok {
			return
		}
		renderFleet(w, board.get(), v, r.URL.Query().Get("org"), cfg)
	})
	return mux
}

// ───────────────────────────── config ─────────────────────────────

type config struct {
	listen  string
	gitBase string
	// adminOrg is the ONE org whose members see across tenants. It must match
	// admin-guard's IAM_ADMIN_ORG — the guard decides who gets in, this decides
	// who sees everything, and a mismatch would silently demote the fleet view to
	// a single-org view (or, if set too wide, promote a tenant to it).
	adminOrg   string
	gitToken   string
	refresh    time.Duration
	staleAfter time.Duration
	scanRepos  int
	runsPer    int

	// universe is the repo holding the declared state of the fleet, and
	// fleetRefresh is how often the four values are re-read. The fleet interval
	// is longer than the run interval because these values move on deploy
	// cadence, not push cadence, and each cycle costs Hanzo Git a read per
	// service.
	universe     string
	fleetRefresh time.Duration
}

func loadConfig() (config, error) {
	c := config{
		listen:    env("CI_LISTEN", ":8080"),
		gitBase:   strings.TrimRight(env("CI_GIT_BASE", "https://git.hanzo.ai"), "/"),
		adminOrg:  env("CI_ADMIN_ORG", "admin"),
		gitToken:  os.Getenv("CI_GIT_TOKEN"),
		scanRepos: envInt("CI_SCAN_REPOS", 60),
		runsPer:   envInt("CI_RUNS_PER_REPO", 8),
		universe:  env("CI_UNIVERSE", "hanzo/universe"),
	}
	c.refresh = time.Duration(envInt("CI_REFRESH_SECONDS", 45)) * time.Second
	c.fleetRefresh = time.Duration(envInt("CI_FLEET_SECONDS", 300)) * time.Second
	// Stale is a multiple of refresh, not its own knob: the only meaningful
	// definition of stale is "we have missed several refreshes", and deriving
	// it means the two can never be configured into contradiction.
	c.staleAfter = 4 * c.refresh
	if c.gitToken == "" {
		return c, errors.New("CI_GIT_TOKEN required (Hanzo Git API token)")
	}
	return c, nil
}

func env(k, def string) string {
	if v := os.Getenv(k); v != "" {
		return v
	}
	return def
}

func envInt(k string, def int) int {
	if v, err := strconv.Atoi(os.Getenv(k)); err == nil && v > 0 {
		return v
	}
	return def
}

// ───────────────────────────── model ─────────────────────────────

// Run is the projection of a Hanzo Git workflow run this dashboard shows. It is
// deliberately a SUBSET: the upstream object carries a dozen more fields, and
// copying them all would make this a second schema to maintain against theirs.
type Execution struct {
	ID       int64  `json:"id"`
	Org      string `json:"org"`
	Repo     string `json:"repo"`
	Workflow string `json:"workflow"`
	Title    string `json:"title"`

	// Status and Conclusion are BOTH required to know how a run went, and
	// reading only one is wrong in a way that looks fine. Status answers
	// "is it over" (queued | in_progress | completed); Conclusion answers
	// "how did it end" and is empty until it is over. A view that buckets on
	// Status alone sees `completed` and cannot tell a pass from a failure, so
	// it draws every finished run — successes and cancellations included — the
	// same way.
	Status     string `json:"status"`
	Conclusion string `json:"conclusion"`

	Event     string    `json:"event"`
	Branch    string    `json:"branch"`
	SHA       string    `json:"sha"`
	Actor     string    `json:"actor"`
	Number    int       `json:"number"`
	URL       string    `json:"url"`
	StartedAt time.Time `json:"startedAt"`
	EndedAt   time.Time `json:"endedAt"`
}

// Duration is zero-valued rather than negative when a run has not finished —
// callers render "running", and a negative duration would print as one.
func (r Execution) Duration() time.Duration {
	if r.StartedAt.IsZero() || r.EndedAt.IsZero() || r.EndedAt.Before(r.StartedAt) {
		return 0
	}
	return r.EndedAt.Sub(r.StartedAt)
}

type snapshot struct {
	Executions []Execution `json:"runs"`
	Repos      int         `json:"repos"`
	FetchedAt  time.Time   `json:"fetchedAt"`
	Err        error       `json:"-"`
}

func (s snapshot) stale(after time.Duration) bool {
	return s.FetchedAt.IsZero() || time.Since(s.FetchedAt) > after
}

func (s snapshot) errString() string {
	if s.Err == nil {
		return ""
	}
	return s.Err.Error()
}

type runCache struct {
	mu   sync.RWMutex
	snap snapshot
}

func (c *runCache) get() snapshot {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.snap
}

// put keeps the LAST GOOD run list when a refresh fails, recording the error
// alongside it. A failed poll must not blank the dashboard: "Hanzo Git is
// unreachable, here is what we last saw" is strictly more useful than an empty
// page, which reads as "nothing is building".
func (c *runCache) put(s snapshot) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if s.Err != nil && len(s.Executions) == 0 && len(c.snap.Executions) > 0 {
		prev := c.snap
		prev.Err = s.Err
		c.snap = prev
		return
	}
	c.snap = s
}

// ───────────────────────────── source ─────────────────────────────

type gitSource struct {
	base  string
	token string
	http  *http.Client

	// Answers that outlive a refresh. The declared fleet is re-read only when the
	// universe head commit moves, and which repo publishes an image is settled
	// once — neither changes on this service's clock, and asking again every
	// cycle would spend the whole budget confirming that.
	mu      sync.Mutex
	pinsAt  string
	pinned  map[string]pin
	located map[string]place
}

func (g *gitSource) getJSON(ctx context.Context, path string, out any) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, g.base+path, nil)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "token "+g.token)
	req.Header.Set("Accept", "application/json")
	resp, err := g.http.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("%s: %s", path, resp.Status)
	}
	return json.NewDecoder(resp.Body).Decode(out)
}

// getText fetches a path whose body is a file rather than a document.
func (g *gitSource) getText(ctx context.Context, path string) (string, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, g.base+path, nil)
	if err != nil {
		return "", err
	}
	req.Header.Set("Authorization", "token "+g.token)
	resp, err := g.http.Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("%s: %s", path, resp.Status)
	}
	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	return string(body), err
}

type repoRef struct {
	FullName string `json:"full_name"`
}

// repos returns the most recently ACTIVE repositories. Sorting by activity and
// taking a window is the whole scan strategy: the instance mirrors ~1400 repos
// and almost none of them built in the last hour, so walking all of them would
// spend the entire refresh budget confirming silence.
func (g *gitSource) repos(ctx context.Context, limit int) ([]string, error) {
	var body struct {
		Data []repoRef `json:"data"`
	}
	q := url.Values{}
	q.Set("sort", "updated")
	q.Set("order", "desc")
	q.Set("limit", strconv.Itoa(limit))
	if err := g.getJSON(ctx, "/v1/repos/search?"+q.Encode(), &body); err != nil {
		return nil, err
	}
	names := make([]string, 0, len(body.Data))
	for _, r := range body.Data {
		if r.FullName != "" {
			names = append(names, r.FullName)
		}
	}
	return names, nil
}

type apiRun struct {
	ID           int64  `json:"id"`
	DisplayTitle string `json:"display_title"`
	Path         string `json:"path"`
	Event        string `json:"event"`
	Status       string `json:"status"`
	Conclusion   string `json:"conclusion"`
	HeadBranch   string `json:"head_branch"`
	HeadSHA      string `json:"head_sha"`
	RunNumber    int    `json:"run_number"`
	HTMLURL      string `json:"html_url"`
	StartedAt    string `json:"started_at"`
	CompletedAt  string `json:"completed_at"`
	Actor        struct {
		Login string `json:"login"`
	} `json:"actor"`
}

func (g *gitSource) runs(ctx context.Context, fullName string, limit int) ([]Execution, error) {
	var body struct {
		WorkflowRuns []apiRun `json:"workflow_runs"`
	}
	path := fmt.Sprintf("/v1/repos/%s/actions/runs?limit=%d", fullName, limit)
	if err := g.getJSON(ctx, path, &body); err != nil {
		return nil, err
	}
	org, repo := splitFullName(fullName)
	out := make([]Execution, 0, len(body.WorkflowRuns))
	for _, r := range body.WorkflowRuns {
		out = append(out, Execution{
			ID:         r.ID,
			Org:        org,
			Repo:       repo,
			Workflow:   workflowOf(r.Path),
			Title:      r.DisplayTitle,
			Status:     r.Status,
			Conclusion: r.Conclusion,
			Event:      r.Event,
			Branch:     r.HeadBranch,
			SHA:        shortSHA(r.HeadSHA),
			Actor:      r.Actor.Login,
			Number:     r.RunNumber,
			URL:        r.HTMLURL,
			StartedAt:  parseTime(r.StartedAt),
			EndedAt:    parseTime(r.CompletedAt),
		})
	}
	return out, nil
}

// poll refreshes the snapshot on an interval, forever.
func poll(ctx context.Context, logger *slog.Logger, src *gitSource, cache *runCache, cfg config) {
	refresh := func() {
		rctx, cancel := context.WithTimeout(ctx, 90*time.Second)
		defer cancel()

		names, err := src.repos(rctx, cfg.scanRepos)
		if err != nil {
			logger.Warn("repo scan failed", "err", err)
			cache.put(snapshot{FetchedAt: time.Now().UTC(), Err: err})
			return
		}

		// Fan out, bounded. The cap is small on purpose: this is a read against
		// the server that schedules every build in the fleet, and a dashboard is
		// never worth degrading it.
		const workers = 6
		var (
			mu   sync.Mutex
			all  []Execution
			errs []string
			wg   sync.WaitGroup
		)
		jobs := make(chan string)
		for i := 0; i < workers; i++ {
			wg.Add(1)
			go func() {
				defer wg.Done()
				for name := range jobs {
					rs, err := src.runs(rctx, name, cfg.runsPer)
					mu.Lock()
					if err != nil {
						// A repo with Actions disabled 404s. That is normal and
						// not worth surfacing as a dashboard-level failure, so
						// it is counted, not shown.
						errs = append(errs, name)
					} else {
						all = append(all, rs...)
					}
					mu.Unlock()
				}
			}()
		}
		for _, n := range names {
			select {
			case jobs <- n:
			case <-rctx.Done():
			}
		}
		close(jobs)
		wg.Wait()

		sort.Slice(all, func(i, j int) bool { return all[i].StartedAt.After(all[j].StartedAt) })
		cache.put(snapshot{Executions: all, Repos: len(names) - len(errs), FetchedAt: time.Now().UTC()})
		logger.Info("refreshed", "repos", len(names), "withRuns", len(names)-len(errs), "runs", len(all))
	}

	refresh()
	t := time.NewTicker(cfg.refresh)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			refresh()
		}
	}
}

// ───────────────────────────── helpers ─────────────────────────────

func splitFullName(s string) (org, repo string) {
	if i := strings.IndexByte(s, '/'); i > 0 {
		return s[:i], s[i+1:]
	}
	return "", s
}

// workflowOf reduces "e2e.yml@refs/heads/main" to "e2e.yml".
func workflowOf(path string) string {
	if i := strings.IndexByte(path, '@'); i > 0 {
		return path[:i]
	}
	return path
}

func shortSHA(s string) string {
	if len(s) > 7 {
		return s[:7]
	}
	return s
}

func parseTime(s string) time.Time {
	if s == "" {
		return time.Time{}
	}
	t, err := time.Parse(time.RFC3339, s)
	if err != nil {
		return time.Time{}
	}
	// Hanzo Git reports an unset timestamp as the Unix epoch rather than null;
	// treated as absent so the UI shows "—" instead of 1970.
	if t.Year() < 2000 {
		return time.Time{}
	}
	return t.UTC()
}

func filterByOrg(runs []Execution, org string) []Execution {
	if org == "" {
		return runs
	}
	out := make([]Execution, 0, len(runs))
	for _, r := range runs {
		if r.Org == org {
			out = append(out, r)
		}
	}
	return out
}

func orgsOf(runs []Execution) []string {
	seen := map[string]bool{}
	for _, r := range runs {
		if r.Org != "" {
			seen[r.Org] = true
		}
	}
	out := make([]string, 0, len(seen))
	for o := range seen {
		out = append(out, o)
	}
	sort.Strings(out)
	return out
}

func writeJSON(w http.ResponseWriter, status int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(v)
}

// ServeFromEnv is Serve with the configuration read the way the standalone
// binary reads it, so a host that mounts this surface configures it exactly as
// the deployment does — one set of names, one meaning, whichever way it runs.
//
// It exists because config is unexported and should stay so: a caller that had
// to construct one would be a second place that decides what the defaults are.
func ServeFromEnv(ctx context.Context, logger *slog.Logger) (http.Handler, error) {
	cfg, err := loadConfig()
	if err != nil {
		return nil, err
	}
	return Serve(ctx, logger, cfg), nil
}