Monorepo for Tangled
spindle: implement queryPipelines and getPipeline XRPC endpoints
Signed-off-by: dawn <dawn@tangled.org>
+301
+222
spindle/db/pipelines.go
+222
spindle/db/pipelines.go
···
1
+
package db
2
+
3
+
import (
4
+
"context"
5
+
"encoding/json"
6
+
"strconv"
7
+
"strings"
8
+
"time"
9
+
10
+
"tangled.org/core/api/tangled"
11
+
"tangled.org/core/spindle/models"
12
+
)
13
+
14
+
func (d *DB) QueryPipelines(ctx context.Context, repoDid string, commits []string, cursor string, limit int) ([]*tangled.CiDefs_Pipeline, string, int64, error) {
15
+
if limit <= 0 {
16
+
limit = 30
17
+
}
18
+
19
+
var query string
20
+
var args []interface{}
21
+
query = `
22
+
select
23
+
rkey, event, created from events
24
+
where
25
+
nsid = 'sh.tangled.pipeline'
26
+
and coalesce(json_extract(event, '$.triggerMetadata.repo.repoDid'), json_extract(event, '$.triggerMetadata.repo.did')) = ?
27
+
`
28
+
args = append(args, repoDid)
29
+
30
+
if len(commits) > 0 {
31
+
placeholders := make([]string, len(commits))
32
+
for i := range commits {
33
+
placeholders[i] = "?"
34
+
args = append(args, commits[i])
35
+
}
36
+
query += " and json_extract(event, '$.triggerMetadata.push.newSha') in (" + strings.Join(placeholders, ",") + ")"
37
+
}
38
+
39
+
if cursor != "" {
40
+
if cVal, err := strconv.ParseInt(cursor, 10, 64); err == nil {
41
+
query += " and created < ?"
42
+
args = append(args, cVal)
43
+
}
44
+
}
45
+
46
+
// First get total count
47
+
var total int64
48
+
countQuery := "select count(*) from (" + query + ")"
49
+
if err := d.QueryRowContext(ctx, countQuery, args...).Scan(&total); err != nil {
50
+
return nil, "", 0, err
51
+
}
52
+
53
+
query += " order by created desc limit ?"
54
+
args = append(args, limit)
55
+
56
+
rows, err := d.QueryContext(ctx, query, args...)
57
+
if err != nil {
58
+
return nil, "", 0, err
59
+
}
60
+
defer rows.Close()
61
+
62
+
var pipelines []*tangled.CiDefs_Pipeline
63
+
var lastCreated int64
64
+
65
+
for rows.Next() {
66
+
var rkey, eventJson string
67
+
var created int64
68
+
if err := rows.Scan(&rkey, &eventJson, &created); err != nil {
69
+
return nil, "", 0, err
70
+
}
71
+
lastCreated = created
72
+
73
+
var rawPipeline tangled.Pipeline
74
+
if err := json.Unmarshal([]byte(eventJson), &rawPipeline); err != nil {
75
+
continue
76
+
}
77
+
78
+
p, err := d.mapToCiDefsPipeline(ctx, rkey, created, rawPipeline)
79
+
if err != nil {
80
+
return nil, "", 0, err
81
+
}
82
+
pipelines = append(pipelines, p)
83
+
}
84
+
85
+
nextCursor := ""
86
+
if len(pipelines) == limit {
87
+
nextCursor = strconv.FormatInt(lastCreated, 10)
88
+
}
89
+
90
+
return pipelines, nextCursor, total, nil
91
+
}
92
+
93
+
func (d *DB) GetPipeline(ctx context.Context, rkey string) (*tangled.CiDefs_Pipeline, error) {
94
+
var eventJson string
95
+
var created int64
96
+
err := d.QueryRowContext(ctx,
97
+
`
98
+
select
99
+
event, created from events
100
+
where
101
+
nsid = 'sh.tangled.pipeline'
102
+
and rkey = ?
103
+
`,
104
+
rkey,
105
+
).Scan(&eventJson, &created)
106
+
107
+
if err != nil {
108
+
return nil, err
109
+
}
110
+
111
+
var rawPipeline tangled.Pipeline
112
+
if err := json.Unmarshal([]byte(eventJson), &rawPipeline); err != nil {
113
+
return nil, err
114
+
}
115
+
116
+
return d.mapToCiDefsPipeline(ctx, rkey, created, rawPipeline)
117
+
}
118
+
119
+
func (d *DB) mapToCiDefsPipeline(ctx context.Context, rkey string, created int64, raw tangled.Pipeline) (*tangled.CiDefs_Pipeline, error) {
120
+
createdAtStr := time.Unix(0, created).Format(time.RFC3339)
121
+
122
+
var repoDidStr string
123
+
if raw.TriggerMetadata != nil && raw.TriggerMetadata.Repo != nil {
124
+
if raw.TriggerMetadata.Repo.RepoDid != nil {
125
+
repoDidStr = *raw.TriggerMetadata.Repo.RepoDid
126
+
} else {
127
+
repoDidStr = raw.TriggerMetadata.Repo.Did
128
+
}
129
+
}
130
+
131
+
commitSha := ""
132
+
var trigger tangled.CiDefs_Pipeline_Trigger
133
+
134
+
if raw.TriggerMetadata != nil {
135
+
switch raw.TriggerMetadata.Kind {
136
+
case "push":
137
+
if raw.TriggerMetadata.Push != nil {
138
+
commitSha = raw.TriggerMetadata.Push.NewSha
139
+
trigger.CiTrigger_Push = &tangled.CiTrigger_Push{
140
+
NewSha: raw.TriggerMetadata.Push.NewSha,
141
+
OldSha: raw.TriggerMetadata.Push.OldSha,
142
+
Ref: raw.TriggerMetadata.Push.Ref,
143
+
}
144
+
}
145
+
case "pullRequest":
146
+
if raw.TriggerMetadata.PullRequest != nil {
147
+
commitSha = raw.TriggerMetadata.PullRequest.SourceSha
148
+
trigger.CiTrigger_PullRequest = &tangled.CiTrigger_PullRequest{
149
+
Action: raw.TriggerMetadata.PullRequest.Action,
150
+
SourceBranch: &raw.TriggerMetadata.PullRequest.SourceBranch,
151
+
SourceSha: raw.TriggerMetadata.PullRequest.SourceSha,
152
+
TargetBranch: raw.TriggerMetadata.PullRequest.TargetBranch,
153
+
}
154
+
}
155
+
case "manual":
156
+
if raw.TriggerMetadata.Manual != nil {
157
+
trigger.CiTrigger_Manual = &tangled.CiTrigger_Manual{}
158
+
}
159
+
}
160
+
}
161
+
162
+
var workflows []*tangled.CiDefs_Workflow
163
+
for _, wf := range raw.Workflows {
164
+
status := "pending"
165
+
var startedAt, finishedAt *string
166
+
167
+
if raw.TriggerMetadata != nil && raw.TriggerMetadata.Repo != nil {
168
+
wfId := models.WorkflowId{
169
+
PipelineId: models.PipelineId{
170
+
Knot: raw.TriggerMetadata.Repo.Knot,
171
+
Rkey: rkey,
172
+
},
173
+
Name: wf.Name,
174
+
}
175
+
176
+
wfStatus, err := d.GetStatus(wfId)
177
+
if err == nil && wfStatus != nil {
178
+
status = wfStatus.Status
179
+
startedAt, finishedAt = d.GetWorkflowTimes(wfId)
180
+
}
181
+
}
182
+
183
+
workflows = append(workflows, &tangled.CiDefs_Workflow{
184
+
Id: wf.Name,
185
+
Name: wf.Name,
186
+
Status: status,
187
+
StartedAt: startedAt,
188
+
FinishedAt: finishedAt,
189
+
})
190
+
}
191
+
192
+
return &tangled.CiDefs_Pipeline{
193
+
Id: rkey,
194
+
Commit: commitSha,
195
+
Repo: &repoDidStr,
196
+
CreatedAt: &createdAtStr,
197
+
Trigger: &trigger,
198
+
Workflows: workflows,
199
+
}, nil
200
+
}
201
+
202
+
func (d *DB) GetWorkflowTimes(workflowId models.WorkflowId) (startedAt, finishedAt *string) {
203
+
pipelineAtUri := workflowId.PipelineId.AtUri()
204
+
205
+
_ = d.QueryRow(
206
+
`
207
+
select
208
+
min(case when json_extract(event, '$.status') = 'running' then json_extract(event, '$.createdAt') end),
209
+
max(case when json_extract(event, '$.status') in ('success', 'failed', 'timeout', 'cancelled') then json_extract(event, '$.createdAt') end)
210
+
from events
211
+
where
212
+
nsid = ?
213
+
and json_extract(event, '$.pipeline') = ?
214
+
and json_extract(event, '$.workflow') = ?
215
+
`,
216
+
tangled.PipelineStatusNSID,
217
+
string(pipelineAtUri),
218
+
workflowId.Name,
219
+
).Scan(&startedAt, &finishedAt)
220
+
221
+
return
222
+
}
+77
spindle/xrpc/ci_query_pipelines.go
+77
spindle/xrpc/ci_query_pipelines.go
···
1
+
package xrpc
2
+
3
+
import (
4
+
"fmt"
5
+
"net/http"
6
+
"strconv"
7
+
8
+
"tangled.org/core/api/tangled"
9
+
xrpcerr "tangled.org/core/xrpc/errors"
10
+
)
11
+
12
+
func (x *Xrpc) HandleCiQueryPipelines(w http.ResponseWriter, r *http.Request) {
13
+
l := x.Logger
14
+
fail := func(e xrpcerr.XrpcError, status int) {
15
+
l.Error("failed", "kind", e.Tag, "error", e.Message)
16
+
writeError(w, e, status)
17
+
}
18
+
19
+
repo := r.URL.Query().Get("repo")
20
+
if repo == "" {
21
+
fail(xrpcerr.GenericError(fmt.Errorf("missing repo parameter")), http.StatusBadRequest)
22
+
return
23
+
}
24
+
25
+
commits := r.URL.Query()["commits"]
26
+
cursor := r.URL.Query().Get("cursor")
27
+
limitStr := r.URL.Query().Get("limit")
28
+
29
+
limit := 30
30
+
if limitStr != "" {
31
+
if val, err := strconv.Atoi(limitStr); err == nil && val > 0 {
32
+
limit = val
33
+
}
34
+
}
35
+
36
+
pipelines, nextCursor, total, err := x.Db.QueryPipelines(r.Context(), repo, commits, cursor, limit)
37
+
if err != nil {
38
+
fail(xrpcerr.GenericError(err), http.StatusInternalServerError)
39
+
return
40
+
}
41
+
42
+
output := tangled.CiQueryPipelines_Output{
43
+
Pipelines: pipelines,
44
+
Total: total,
45
+
}
46
+
if nextCursor != "" {
47
+
output.Cursor = &nextCursor
48
+
}
49
+
50
+
if err := writeJson(w, http.StatusOK, output); err != nil {
51
+
fail(xrpcerr.GenericError(err), http.StatusInternalServerError)
52
+
}
53
+
}
54
+
55
+
func (x *Xrpc) HandleCiGetPipeline(w http.ResponseWriter, r *http.Request) {
56
+
l := x.Logger
57
+
fail := func(e xrpcerr.XrpcError, status int) {
58
+
l.Error("failed", "kind", e.Tag, "error", e.Message)
59
+
writeError(w, e, status)
60
+
}
61
+
62
+
pipeline := r.URL.Query().Get("pipeline")
63
+
if pipeline == "" {
64
+
fail(xrpcerr.GenericError(fmt.Errorf("missing pipeline parameter")), http.StatusBadRequest)
65
+
return
66
+
}
67
+
68
+
p, err := x.Db.GetPipeline(r.Context(), pipeline)
69
+
if err != nil {
70
+
fail(xrpcerr.GenericError(err), http.StatusInternalServerError)
71
+
return
72
+
}
73
+
74
+
if err := writeJson(w, http.StatusOK, p); err != nil {
75
+
fail(xrpcerr.GenericError(err), http.StatusInternalServerError)
76
+
}
77
+
}
+2
spindle/xrpc/xrpc.go
+2
spindle/xrpc/xrpc.go
···
49
49
// service query endpoints (no auth required)
50
50
r.Get("/"+tangled.OwnerNSID, x.Owner)
51
51
r.Get("/"+tangled.CiPipelineSubscribeLogsNSID, x.HandleCiPipelineSubscribeLogs)
52
+
r.Get("/"+tangled.CiQueryPipelinesNSID, x.HandleCiQueryPipelines)
53
+
r.Get("/"+tangled.CiGetPipelineNSID, x.HandleCiGetPipeline)
52
54
53
55
return r
54
56
}