+ "github.com/atmin/gitmote/internal/scaleway"

+ // The Scaleway Jobs client triggers a runner per CI job. It no-ops when

+ // SCW_CI_JOB_DEFINITION_ID is unset (local dev / tests / e2e), so the write

+ // path runs fully except for the external call. SCW_REGION falls back to the

+ // AWS_REGION already set for the object store.

+ jobDefID := os.Getenv("SCW_CI_JOB_DEFINITION_ID")

+ workerSecret := os.Getenv("WORKER_SECRET")

+ gitmoteURL := os.Getenv("GITMOTE_URL")

+ if jobDefID != "" && (workerSecret == "" || gitmoteURL == "") {

+ _ = writer.Close()

+ _ = md.Close()

+ return nil, nil, noop, fmt.Errorf("SCW_CI_JOB_DEFINITION_ID is set but WORKER_SECRET or GITMOTE_URL is missing")

+ }

+ jobs := scaleway.NewJobsClient(os.Getenv("SCW_SECRET_KEY"), envOr("SCW_REGION", os.Getenv("AWS_REGION")), jobDefID)

+ // A successful, branch-advancing push discovers its workflows, records a run,

+ // and triggers a runner per job — fire-and-forget, so dispatch never fails the

+ // push (tasks/16-ci.md).

+ dispatcher := ci.NewDispatcher(ci.Config{

+ Runs: md,

+ Materializer: materializer,

+ Trigger: jobs,

+ BaseURL: gitmoteURL,

+ WorkerSecret: workerSecret,

+ Logger: logger,

+ })

+| `GITMOTE_URL` | public base URL (`https://gitmote.atmin.net`) — injected into the CI runner's env so it clones and reports back here |

+| `SCW_CI_JOB_DEFINITION_ID` | the Scaleway Serverless Job definition for the CI runner; **unset disables the trigger** (runs/jobs still record). One-time `scw jobs definition create` lands with the runner (stage 4) |

+| `SCW_SECRET_KEY` | secret — Scaleway API secret key (the UUID) used to start the CI job; also the registry/deploy key |

+| `SCW_REGION` | Scaleway region for the CI job start (falls back to `AWS_REGION`) |

+| `WORKER_SECRET` | secret — shared runner-auth secret; injected into the runner env and compared on its report-back (stage 4) |

+> `SCW_CI_JOB_DEFINITION_ID` gates CI execution: unset (local dev, tests, e2e) the

+> dispatcher records runs/jobs but the Scaleway trigger no-ops. When it **is** set,

+> `WORKER_SECRET` and `GITMOTE_URL` are required — the server refuses to start

+> otherwise, so a half-configured CI can't silently trigger unreachable runners.

+

+ "strconv"

+ SetJobStatus(ctx context.Context, jobID int64, status meta.RunStatus) error

+// Trigger starts one CI job's runner with the given env. *scaleway.JobsClient

+// satisfies it; it no-ops when unconfigured (local dev / tests).

+type Trigger interface {

+ Trigger(ctx context.Context, env map[string]string) error

+}

+

+// Config wires the dispatcher's collaborators. BaseURL and WorkerSecret are

+// injected into each triggered runner's env (the runner clones from BaseURL and

+// authenticates its report-back with WorkerSecret — stage 4).

+type Config struct {

+ Runs Runs

+ Materializer Materializer

+ Trigger Trigger

+ BaseURL string // GITMOTE_URL

+ WorkerSecret string // WORKER_SECRET

+ Logger *slog.Logger

+}

+

+// Dispatcher enqueues CI runs for ref advances and triggers a runner per job.

+// Only the leader constructs and calls it (it is the only instance that

+// processes receive-pack).

+ runs Runs

+ mz Materializer

+ trigger Trigger

+ baseURL string

+ secret string

+ logger *slog.Logger

+// NewDispatcher returns a Dispatcher from cfg.

+func NewDispatcher(cfg Config) *Dispatcher {

+ logger := cfg.Logger

+ return &Dispatcher{

+ runs: cfg.Runs,

+ mz: cfg.Materializer,

+ trigger: cfg.Trigger,

+ baseURL: cfg.BaseURL,

+ secret: cfg.WorkerSecret,

+ logger: logger,

+ }

+// recordRun creates a queued run and one job per workflow file, then triggers a

+// runner per job. A file that fails to parse as YAML downgrades the run to failed

+// (the first parse error is logged), but the parsable files still get job rows,

+// so the failure is visible in the UI. A trigger error marks that job error and,

+// if every job failed to trigger, the run error — one job's failure never aborts

+// the others, and nothing here fails the push.

+ var (

+ parseErr error

+ jobs int

+ triggerErrors int

+ )

+ job, err := d.runs.CreateJob(ctx, run.ID, f.Name)

+ if err != nil {

+ continue

+ }

+ jobs++

+ if err := d.trigger.Trigger(ctx, d.jobEnv(ev, run.ID, job.ID)); err != nil {

+ d.logger.Error("ci: trigger failed",

+ "run", run.ID, "job", job.ID, "name", f.Name, "error", err)

+ if err := d.runs.SetJobStatus(ctx, job.ID, meta.RunError); err != nil {

+ d.logger.Error("ci: set job error failed", "job", job.ID, "error", err)

+ }

+ triggerErrors++

+ switch {

+ case parseErr != nil:

+ d.setRunStatus(ctx, run.ID, meta.RunFailed)

+ case jobs > 0 && triggerErrors == jobs:

+ // Every job failed to trigger — the run produced nothing.

+ d.setRunStatus(ctx, run.ID, meta.RunError)

+ }

+}

+

+// jobEnv is the per-job environment injected into the triggered runner. The

+// clone token is a placeholder until scoped tokens land (task 20).

+func (d *Dispatcher) jobEnv(ev Event, runID, jobID int64) map[string]string {

+ return map[string]string{

+ "GITMOTE_CI_RUN_ID": strconv.FormatInt(runID, 10),

+ "GITMOTE_CI_JOB_ID": strconv.FormatInt(jobID, 10),

+ "GITMOTE_URL": d.baseURL,

+ "WORKER_SECRET": d.secret,

+ "GITMOTE_REPO": ev.RepoName,

+ "GITMOTE_SHA": ev.NewSHA,

+ "GITMOTE_REF": ev.Ref,

+ "GITMOTE_CI_CLONE_TOKEN": "", // placeholder until task 20

+ }

+}

+

+// setRunStatus transitions a run, logging (not returning) any error.

+func (d *Dispatcher) setRunStatus(ctx context.Context, runID int64, status meta.RunStatus) {

+ if err := d.runs.SetRunStatus(ctx, runID, status); err != nil {

+ d.logger.Error("ci: set run status failed", "run", runID, "status", status, "error", err)

+ "errors"

+ "strconv"

+// stubTrigger records the env of each trigger and can be made to fail.

+type stubTrigger struct {

+ calls []map[string]string

+ err error

+}

+

+func (s *stubTrigger) Trigger(_ context.Context, env map[string]string) error {

+ s.calls = append(s.calls, env)

+ return s.err

+}

+

+func newDispatcher(md *meta.Metadata, mz *repo.Materializer, tr Trigger) *Dispatcher {

+ return NewDispatcher(Config{

+ Runs: md,

+ Materializer: mz,

+ Trigger: tr,

+ BaseURL: "https://gitmote.test",

+ WorkerSecret: "worker-secret",

+ })

+}

+

+ tr := &stubTrigger{}

+ newDispatcher(md, mz, tr).Dispatch(ctx, branchEvent(r, head))

+

+ // The job was triggered exactly once, with the runner env for that run/job.

+ if len(tr.calls) != 1 {

+ t.Fatalf("trigger calls = %d, want 1", len(tr.calls))

+ }

+ env := tr.calls[0]

+ if env["GITMOTE_CI_RUN_ID"] != strconv.FormatInt(runs[0].ID, 10) ||

+ env["GITMOTE_CI_JOB_ID"] != strconv.FormatInt(jobs[0].ID, 10) {

+ t.Errorf("env run/job ids = %q/%q, want %d/%d",

+ env["GITMOTE_CI_RUN_ID"], env["GITMOTE_CI_JOB_ID"], runs[0].ID, jobs[0].ID)

+ }

+ if env["GITMOTE_REPO"] != "atmin/app" || env["GITMOTE_SHA"] != head ||

+ env["GITMOTE_REF"] != "refs/heads/main" {

+ t.Errorf("env checkout coords = %+v, want repo/sha/ref set", env)

+ }

+ if env["GITMOTE_URL"] != "https://gitmote.test" || env["WORKER_SECRET"] != "worker-secret" {

+ t.Errorf("env url/secret = %q/%q, want configured values", env["GITMOTE_URL"], env["WORKER_SECRET"])

+ }

+ newDispatcher(md, mz, &stubTrigger{}).Dispatch(ctx, branchEvent(r, head))

+ newDispatcher(md, mz, &stubTrigger{}).Dispatch(ctx, branchEvent(r, head))

+ newDispatcher(md, mz, &stubTrigger{}).Dispatch(ctx, branchEvent(r, head))

+ d := newDispatcher(md, mz, &stubTrigger{})

+

+func TestDispatchAllTriggersFailMarksRunError(t *testing.T) {

+ ctx := context.Background()

+ md, s, mz := newFixture(t)

+ r, head := seedRepo(t, md, s, "atmin/app", map[string]string{

+ ".github/workflows/ci.yml": "name: CI\non: push\n",

+ })

+

+ tr := &stubTrigger{err: errors.New("scaleway down")}

+ newDispatcher(md, mz, tr).Dispatch(ctx, branchEvent(r, head))

+

+ runs, _ := md.ListRuns(ctx, r.ID, 0)

+ if len(runs) != 1 || runs[0].Status != meta.RunError {

+ t.Fatalf("run = %+v, want one run in error", runs)

+ }

+ jobs, _ := md.ListJobs(ctx, runs[0].ID)

+ if len(jobs) != 1 || jobs[0].Status != meta.RunError {

+ t.Errorf("jobs = %+v, want the job marked error", jobs)

+ }

+}

+

+func TestDispatchOneTriggerFailsOthersProceed(t *testing.T) {

+ ctx := context.Background()

+ md, s, mz := newFixture(t)

+ r, head := seedRepo(t, md, s, "atmin/app", map[string]string{

+ ".github/workflows/a.yml": "name: A\non: push\n",

+ ".github/workflows/b.yml": "name: B\non: push\n",

+ })

+

+ // Fail the first trigger, succeed the rest — one failure must not abort others.

+ tr := &failNthTrigger{failOn: 1}

+ newDispatcher(md, mz, tr).Dispatch(ctx, branchEvent(r, head))

+

+ runs, _ := md.ListRuns(ctx, r.ID, 0)

+ if len(runs) != 1 {

+ t.Fatalf("runs = %d, want 1", len(runs))

+ }

+ // Not every job failed, so the run is not marked error; it stays queued.

+ if runs[0].Status != meta.RunQueued {

+ t.Errorf("run status = %q, want queued (not all jobs failed)", runs[0].Status)

+ }

+ jobs, _ := md.ListJobs(ctx, runs[0].ID)

+ if len(jobs) != 2 {

+ t.Fatalf("jobs = %d, want 2 (both created despite one trigger failure)", len(jobs))

+ }

+ var errored, queued int

+ for _, j := range jobs {

+ switch j.Status {

+ case meta.RunError:

+ errored++

+ case meta.RunQueued:

+ queued++

+ }

+ }

+ if errored != 1 || queued != 1 {

+ t.Errorf("job statuses = %+v, want one error + one queued", jobs)

+ }

+}

+

+// failNthTrigger fails the Nth (1-based) trigger call and succeeds the others.

+type failNthTrigger struct {

+ failOn int

+ n int

+}

+

+func (f *failNthTrigger) Trigger(_ context.Context, _ map[string]string) error {

+ f.n++

+ if f.n == f.failOn {

+ return errors.New("trigger failed")

+ }

+ return nil

+}

+// noopTrigger is a ci.Trigger that does nothing — the write-path tests only care

+// that runs/jobs are recorded, not that a runner starts.

+type noopTrigger struct{}

+

+func (noopTrigger) Trigger(context.Context, map[string]string) error { return nil }

+

+ d := ci.NewDispatcher(ci.Config{

+ Runs: m,

+ Materializer: repo.New(m, s, t.TempDir()),

+ Trigger: noopTrigger{},

+ Logger: slog.New(slog.DiscardHandler),

+ })

+// Package scaleway is a thin client for Scaleway Serverless Jobs — the CI runner

+// substrate (tasks/16-ci.md §Runner substrate). A job definition is created once

+// out of band; the app injects per-run env at start, so one definition serves

+// all repos. The client is fire-and-forget: the caller triggers a run and never

+// waits on it — the runner reports back over the internal API (stage 4).

+package scaleway

+

+import (

+ "bytes"

+ "context"

+ "encoding/json"

+ "fmt"

+ "io"

+ "net/http"

+ "time"

+)

+

+// triggerTimeout bounds the single start POST; the job itself runs

+// asynchronously on Scaleway well beyond this.

+const triggerTimeout = 10 * time.Second

+

+// JobsClient starts runs of one Scaleway Serverless Job definition.

+type JobsClient struct {

+ secretKey string

+ region string

+ jobDefinitionID string

+ httpClient *http.Client

+}

+

+// NewJobsClient returns a client for the given job definition. When

+// jobDefinitionID is empty (local dev / tests), Trigger is a no-op.

+func NewJobsClient(secretKey, region, jobDefinitionID string) *JobsClient {

+ return &JobsClient{

+ secretKey: secretKey,

+ region: region,

+ jobDefinitionID: jobDefinitionID,

+ httpClient: &http.Client{Timeout: triggerTimeout},

+ }

+}

+

+// Trigger starts one job run with env injected as its environment variables. It

+// is a no-op returning nil when the client has no job definition configured. A

+// response status >= 300 is an error, with the response body attached.

+func (c *JobsClient) Trigger(ctx context.Context, env map[string]string) error {

+ if c.jobDefinitionID == "" {

+ return nil

+ }

+

+ url := fmt.Sprintf(

+ "https://api.scaleway.com/serverless-jobs/v1alpha2/regions/%s/job-definitions/%s/start",

+ c.region, c.jobDefinitionID,

+ )

+ body, err := json.Marshal(map[string]any{"environment_variables": env})

+ if err != nil {

+ return err

+ }

+

+ req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))

+ if err != nil {

+ return err

+ }

+ req.Header.Set("Content-Type", "application/json")

+ req.Header.Set("X-Auth-Token", c.secretKey)

+

+ resp, err := c.httpClient.Do(req)

+ if err != nil {

+ return err

+ }

+ defer func() { _ = resp.Body.Close() }()

+

+ if resp.StatusCode >= 300 {

+ msg, _ := io.ReadAll(resp.Body)

+ return fmt.Errorf("scaleway: start job definition %s returned %d: %s",

+ c.jobDefinitionID, resp.StatusCode, msg)

+ }

+ return nil

+}

+package scaleway

+

+import (

+ "context"

+ "encoding/json"

+ "io"

+ "net/http"

+ "net/http/httptest"

+ "strings"

+ "testing"

+)

+

+func TestTriggerPostsStartRequest(t *testing.T) {

+ var (

+ gotPath string

+ gotToken string

+ gotBody map[string]map[string]string

+ )

+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

+ gotPath = r.URL.Path

+ gotToken = r.Header.Get("X-Auth-Token")

+ body, _ := io.ReadAll(r.Body)

+ _ = json.Unmarshal(body, &gotBody)

+ w.WriteHeader(http.StatusOK)

+ }))

+ defer srv.Close()

+

+ // Point the client at the test server by overriding the base URL via a client

+ // that rewrites the host: simplest is to build the client and swap its

+ // transport to redirect to srv.

+ c := NewJobsClient("secret-key", "fr-par", "def-123")

+ c.httpClient = srv.Client()

+ c.httpClient.Transport = rewriteHost(srv.URL)

+

+ env := map[string]string{"GITMOTE_CI_RUN_ID": "7", "WORKER_SECRET": "s3cr3t"}

+ if err := c.Trigger(context.Background(), env); err != nil {

+ t.Fatalf("Trigger: %v", err)

+ }

+

+ wantPath := "/serverless-jobs/v1alpha2/regions/fr-par/job-definitions/def-123/start"

+ if gotPath != wantPath {

+ t.Errorf("path = %q, want %q", gotPath, wantPath)

+ }

+ if gotToken != "secret-key" {

+ t.Errorf("X-Auth-Token = %q, want secret-key", gotToken)

+ }

+ vars := gotBody["environment_variables"]

+ if vars["GITMOTE_CI_RUN_ID"] != "7" || vars["WORKER_SECRET"] != "s3cr3t" {

+ t.Errorf("environment_variables = %v, want the passed env", vars)

+ }

+}

+

+func TestTriggerErrorsOnBadStatus(t *testing.T) {

+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

+ w.WriteHeader(http.StatusForbidden)

+ _, _ = w.Write([]byte("denied"))

+ }))

+ defer srv.Close()

+

+ c := NewJobsClient("k", "fr-par", "def-123")

+ c.httpClient = srv.Client()

+ c.httpClient.Transport = rewriteHost(srv.URL)

+

+ err := c.Trigger(context.Background(), nil)

+ if err == nil {

+ t.Fatal("Trigger returned nil, want an error for a 403")

+ }

+ if !strings.Contains(err.Error(), "403") || !strings.Contains(err.Error(), "denied") {

+ t.Errorf("error = %v, want it to include the status and body", err)

+ }

+}

+

+func TestTriggerNoOpWhenUnconfigured(t *testing.T) {

+ called := false

+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

+ called = true

+ }))

+ defer srv.Close()

+

+ c := NewJobsClient("", "", "") // no job definition

+ c.httpClient = srv.Client()

+ c.httpClient.Transport = rewriteHost(srv.URL)

+

+ if err := c.Trigger(context.Background(), map[string]string{"x": "y"}); err != nil {

+ t.Fatalf("Trigger no-op returned %v, want nil", err)

+ }

+ if called {

+ t.Error("Trigger made an HTTP call when the job definition is unset")

+ }

+}

+

+// rewriteHost returns a transport that redirects every request to the test

+// server's host while preserving the request path, so the client's hardcoded

+// api.scaleway.com URL reaches httptest.

+func rewriteHost(serverURL string) http.RoundTripper {

+ base := strings.TrimPrefix(strings.TrimPrefix(serverURL, "http://"), "https://")

+ return roundTripFunc(func(r *http.Request) (*http.Response, error) {

+ r.URL.Scheme = "http"

+ r.URL.Host = base

+ return http.DefaultTransport.RoundTrip(r)

+ })

+}

+

+type roundTripFunc func(*http.Request) (*http.Response, error)

+

+func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }