client.go
1package appview
2
3import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "fmt"
8 "io"
9 "net/http"
10 "net/url"
11 "strings"
12 "time"
13
14 "git.kilimanjaro.io/ci/internal/lex"
15 "git.kilimanjaro.io/ci/internal/runner"
16)
17
18// Runner is a client of a runner's API (internal/runner Handler).
19type Runner struct {
20 URL string // e.g. http://127.0.0.1:8476
21 Token string
22 HTTP *http.Client
23}
24
25func (c *Runner) do(ctx context.Context, method, path string, body any, out any) error {
26 var rd io.Reader
27 if body != nil {
28 b, err := json.Marshal(body)
29 if err != nil {
30 return err
31 }
32 rd = bytes.NewReader(b)
33 }
34 req, err := http.NewRequestWithContext(ctx, method, strings.TrimSuffix(c.URL, "/")+path, rd)
35 if err != nil {
36 return err
37 }
38 if c.Token != "" {
39 req.Header.Set("Authorization", "Bearer "+c.Token)
40 }
41 if body != nil {
42 req.Header.Set("Content-Type", "application/json")
43 }
44 hc := c.HTTP
45 if hc == nil {
46 hc = &http.Client{Timeout: 30 * time.Second}
47 }
48 resp, err := hc.Do(req)
49 if err != nil {
50 return err
51 }
52 defer resp.Body.Close()
53 if resp.StatusCode >= 300 {
54 var e struct {
55 Error string `json:"error"`
56 }
57 b, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
58 if json.Unmarshal(b, &e) == nil && e.Error != "" {
59 return fmt.Errorf("%s", e.Error)
60 }
61 return fmt.Errorf("runner: %s: %s", resp.Status, strings.TrimSpace(string(b)))
62 }
63 if out == nil {
64 return nil
65 }
66 if w, ok := out.(io.Writer); ok {
67 _, err := io.Copy(w, resp.Body)
68 return err
69 }
70 return json.NewDecoder(resp.Body).Decode(out)
71}
72
73// Runs lists runs, newest first.
74func (c *Runner) Runs(ctx context.Context, repo string, limit int) ([]runner.RunView, error) {
75 q := url.Values{"limit": {fmt.Sprint(limit)}}
76 if repo != "" {
77 q.Set("repo", repo)
78 }
79 var out []runner.RunView
80 return out, c.do(ctx, "GET", "/api/runs?"+q.Encode(), nil, &out)
81}
82
83// Run returns one run.
84func (c *Runner) Run(ctx context.Context, id string) (runner.RunView, error) {
85 var out runner.RunView
86 return out, c.do(ctx, "GET", "/api/runs/"+url.PathEscape(id), nil, &out)
87}
88
89// Events returns a run's events after seq.
90func (c *Runner) Events(ctx context.Context, id string, after int) ([]lex.RunEvent, error) {
91 var buf bytes.Buffer
92 if err := c.do(ctx, "GET", fmt.Sprintf("/api/runs/%s/events?after=%d", url.PathEscape(id), after), nil, &buf); err != nil {
93 return nil, err
94 }
95 var out []lex.RunEvent
96 for _, line := range bytes.Split(buf.Bytes(), []byte("\n")) {
97 if len(bytes.TrimSpace(line)) == 0 {
98 continue
99 }
100 var ev lex.RunEvent
101 if err := json.Unmarshal(line, &ev); err != nil {
102 return nil, err
103 }
104 out = append(out, ev)
105 }
106 return out, nil
107}
108
109// Text copies a run's plain-text log to w.
110func (c *Runner) Text(ctx context.Context, id string, w io.Writer) error {
111 return c.do(ctx, "GET", "/api/runs/"+url.PathEscape(id)+"/log.txt", nil, w)
112}
113
114// File copies a run file (results/…, artifacts/…) to w.
115func (c *Runner) File(ctx context.Context, id, path string, w io.Writer) error {
116 return c.do(ctx, "GET", "/api/runs/"+url.PathEscape(id)+"/files/"+path, nil, w)
117}
118
119// Gate answers a run's gate: option is 1-based; dismiss ends the run.
120func (c *Runner) Gate(ctx context.Context, id string, option int, dismiss bool, actor string) error {
121 return c.do(ctx, "POST", "/api/runs/"+url.PathEscape(id)+"/gate",
122 map[string]any{"option": option, "dismiss": dismiss, "actor": actor}, nil)
123}
124
125// Action starts a manual action run and returns its id.
126func (c *Runner) Action(ctx context.Context, id, action, actor string) (string, error) {
127 var out struct {
128 Run string `json:"run"`
129 }
130 err := c.do(ctx, "POST", "/api/runs/"+url.PathEscape(id)+"/actions/"+url.PathEscape(action), map[string]any{"actor": actor}, &out)
131 return out.Run, err
132}
133
134// Cancel stops a run.
135func (c *Runner) Cancel(ctx context.Context, id string) error {
136 return c.do(ctx, "POST", "/api/runs/"+url.PathEscape(id)+"/cancel", nil, nil)
137}
138
139// Trigger starts a manual pipeline of repo (a DID) at rev (a commit or a
140// branch, resolved by the runner) for the named workflows (all if none).
141func (c *Runner) Trigger(ctx context.Context, repoDid, rev string, workflows []string) (string, error) {
142 in := lex.TriggerPipelineInput{Repo: repoDid, Workflows: workflows,
143 Trigger: &lex.ManualTrigger{LexiconTypeID: "sh.tangled.ci.trigger#manual", Sha: rev}}
144 if !isSha(rev) {
145 ref := "refs/heads/" + rev
146 in.Trigger.Ref = &ref
147 }
148 var out lex.TriggerPipelineOutput
149 return out.Pipeline, c.do(ctx, "POST", "/xrpc/sh.tangled.ci.triggerPipeline", in, &out)
150}
151
152func isSha(s string) bool {
153 if len(s) != 40 {
154 return false
155 }
156 for _, c := range s {
157 if !strings.ContainsRune("0123456789abcdef", c) {
158 return false
159 }
160 }
161 return true
162}
163
164// Forward sends a request through to the runner unchanged (method, path
165// with query, body) and copies the answer back: imports and lock leases from
166// local runs.
167func (c *Runner) Forward(w http.ResponseWriter, r *http.Request, path string) {
168 req, err := http.NewRequestWithContext(r.Context(), r.Method, strings.TrimSuffix(c.URL, "/")+path, r.Body)
169 if err != nil {
170 http.Error(w, err.Error(), http.StatusBadGateway)
171 return
172 }
173 if c.Token != "" {
174 req.Header.Set("Authorization", "Bearer "+c.Token)
175 }
176 if ct := r.Header.Get("Content-Type"); ct != "" {
177 req.Header.Set("Content-Type", ct)
178 }
179 hc := c.HTTP
180 if hc == nil {
181 hc = &http.Client{Timeout: 2 * time.Minute}
182 }
183 resp, err := hc.Do(req)
184 if err != nil {
185 http.Error(w, err.Error(), http.StatusBadGateway)
186 return
187 }
188 defer resp.Body.Close()
189 w.Header().Set("Content-Type", resp.Header.Get("Content-Type"))
190 w.WriteHeader(resp.StatusCode)
191 _, _ = io.Copy(w, resp.Body)
192}
