alice.pds.demo.boltless.dev
This repository has no description
1package db
2
3import (
4 "context"
5 "fmt"
6 "slices"
7 "strings"
8 "time"
9
10 "github.com/bluesky-social/indigo/atproto/syntax"
11 "tangled.org/core/appview/models"
12 "tangled.org/core/orm"
13)
14
15func GetPipelines(e Execer, filters ...orm.Filter) ([]models.Pipeline, error) {
16 var pipelines []models.Pipeline
17
18 var conditions []string
19 var args []any
20 for _, filter := range filters {
21 conditions = append(conditions, filter.Condition())
22 args = append(args, filter.Arg()...)
23 }
24
25 whereClause := ""
26 if conditions != nil {
27 whereClause = " where " + strings.Join(conditions, " and ")
28 }
29
30 query := fmt.Sprintf(`select id, rkey, knot, repo_owner, repo_name, sha, created from pipelines %s`, whereClause)
31
32 rows, err := e.Query(query, args...)
33
34 if err != nil {
35 return nil, err
36 }
37 defer rows.Close()
38
39 for rows.Next() {
40 var pipeline models.Pipeline
41 var createdAt string
42 err = rows.Scan(
43 &pipeline.Id,
44 &pipeline.Rkey,
45 &pipeline.Knot,
46 &pipeline.RepoOwner,
47 &pipeline.RepoName,
48 &pipeline.Sha,
49 &createdAt,
50 )
51 if err != nil {
52 return nil, err
53 }
54
55 if t, err := time.Parse(time.RFC3339, createdAt); err == nil {
56 pipeline.Created = t
57 }
58
59 pipelines = append(pipelines, pipeline)
60 }
61
62 if err = rows.Err(); err != nil {
63 return nil, err
64 }
65
66 return pipelines, nil
67}
68
69func AddPipeline(e Execer, pipeline models.Pipeline) error {
70 args := []any{
71 pipeline.Rkey,
72 pipeline.Knot,
73 pipeline.RepoOwner,
74 pipeline.RepoName,
75 pipeline.TriggerId,
76 pipeline.Sha,
77 }
78
79 placeholders := make([]string, len(args))
80 for i := range placeholders {
81 placeholders[i] = "?"
82 }
83
84 query := fmt.Sprintf(`
85 insert or ignore into pipelines (
86 rkey,
87 knot,
88 repo_owner,
89 repo_name,
90 trigger_id,
91 sha
92 ) values (%s)
93 `, strings.Join(placeholders, ","))
94
95 _, err := e.Exec(query, args...)
96
97 return err
98}
99
100func AddTrigger(e Execer, trigger models.Trigger) (int64, error) {
101 args := []any{
102 trigger.Kind,
103 trigger.PushRef,
104 trigger.PushNewSha,
105 trigger.PushOldSha,
106 trigger.PRSourceBranch,
107 trigger.PRTargetBranch,
108 trigger.PRSourceSha,
109 trigger.PRAction,
110 }
111
112 placeholders := make([]string, len(args))
113 for i := range placeholders {
114 placeholders[i] = "?"
115 }
116
117 query := fmt.Sprintf(`insert or ignore into triggers (
118 kind,
119 push_ref,
120 push_new_sha,
121 push_old_sha,
122 pr_source_branch,
123 pr_target_branch,
124 pr_source_sha,
125 pr_action
126 ) values (%s)`, strings.Join(placeholders, ","))
127
128 res, err := e.Exec(query, args...)
129 if err != nil {
130 return 0, err
131 }
132
133 return res.LastInsertId()
134}
135
136func AddPipelineStatus(ctx context.Context, e Execer, status models.PipelineStatus) error {
137 args := []any{
138 status.Spindle,
139 status.Rkey,
140 status.PipelineKnot,
141 status.PipelineRkey,
142 status.Workflow,
143 status.Status,
144 status.Error,
145 status.ExitCode,
146 status.Created.Format(time.RFC3339),
147 }
148
149 placeholders := make([]string, len(args))
150 for i := range placeholders {
151 placeholders[i] = "?"
152 }
153
154 query := fmt.Sprintf(`
155 insert or ignore into pipeline_statuses (
156 spindle,
157 rkey,
158 pipeline_knot,
159 pipeline_rkey,
160 workflow,
161 status,
162 error,
163 exit_code,
164 created
165 ) values (%s)
166 `, strings.Join(placeholders, ","))
167
168 _, err := e.ExecContext(ctx, query, args...)
169 return err
170}
171
172// this is a mega query, but the most useful one:
173// get N pipelines, for each one get the latest status of its N workflows
174//
175// the pipelines table is aliased to `p`
176// the triggers table is aliased to `t`
177func GetPipelineStatuses(e Execer, limit int, filters ...orm.Filter) ([]models.Pipeline, error) {
178 var conditions []string
179 var args []any
180 for _, filter := range filters {
181 conditions = append(conditions, filter.Condition())
182 args = append(args, filter.Arg()...)
183 }
184
185 whereClause := ""
186 if conditions != nil {
187 whereClause = " where " + strings.Join(conditions, " and ")
188 }
189
190 query := fmt.Sprintf(`
191 select
192 p.id,
193 p.knot,
194 p.rkey,
195 p.repo_owner,
196 p.repo_name,
197 p.sha,
198 p.created,
199 t.id,
200 t.kind,
201 t.push_ref,
202 t.push_new_sha,
203 t.push_old_sha,
204 t.pr_source_branch,
205 t.pr_target_branch,
206 t.pr_source_sha,
207 t.pr_action
208 from
209 pipelines p
210 join
211 triggers t ON p.trigger_id = t.id
212 %s
213 order by p.created desc
214 limit %d
215 `, whereClause, limit)
216
217 rows, err := e.Query(query, args...)
218 if err != nil {
219 return nil, err
220 }
221 defer rows.Close()
222
223 pipelines := make(map[syntax.ATURI]models.Pipeline)
224 for rows.Next() {
225 var p models.Pipeline
226 var t models.Trigger
227 var created string
228
229 err := rows.Scan(
230 &p.Id,
231 &p.Knot,
232 &p.Rkey,
233 &p.RepoOwner,
234 &p.RepoName,
235 &p.Sha,
236 &created,
237 &p.TriggerId,
238 &t.Kind,
239 &t.PushRef,
240 &t.PushNewSha,
241 &t.PushOldSha,
242 &t.PRSourceBranch,
243 &t.PRTargetBranch,
244 &t.PRSourceSha,
245 &t.PRAction,
246 )
247 if err != nil {
248 return nil, err
249 }
250
251 p.Created, err = time.Parse(time.RFC3339, created)
252 if err != nil {
253 return nil, fmt.Errorf("invalid pipeline created timestamp %q: %w", created, err)
254 }
255
256 t.Id = p.TriggerId
257 p.Trigger = &t
258 p.Statuses = make(map[string]models.WorkflowStatus)
259
260 pipelines[p.AtUri()] = p
261 }
262
263 // get all statuses
264 // the where clause here is of the form:
265 //
266 // where (pipeline_knot = k1 and pipeline_rkey = r1)
267 // or (pipeline_knot = k2 and pipeline_rkey = r2)
268 conditions = nil
269 args = nil
270 for _, p := range pipelines {
271 knotFilter := orm.FilterEq("pipeline_knot", p.Knot)
272 rkeyFilter := orm.FilterEq("pipeline_rkey", p.Rkey)
273 conditions = append(conditions, fmt.Sprintf("(%s and %s)", knotFilter.Condition(), rkeyFilter.Condition()))
274 args = append(args, p.Knot)
275 args = append(args, p.Rkey)
276 }
277 whereClause = ""
278 if conditions != nil {
279 whereClause = "where " + strings.Join(conditions, " or ")
280 }
281 query = fmt.Sprintf(`
282 select
283 id, spindle, rkey, pipeline_knot, pipeline_rkey, created, workflow, status, error, exit_code
284 from
285 pipeline_statuses
286 %s
287 `, whereClause)
288
289 rows, err = e.Query(query, args...)
290 if err != nil {
291 return nil, err
292 }
293 defer rows.Close()
294
295 for rows.Next() {
296 var ps models.PipelineStatus
297 var created string
298
299 err := rows.Scan(
300 &ps.ID,
301 &ps.Spindle,
302 &ps.Rkey,
303 &ps.PipelineKnot,
304 &ps.PipelineRkey,
305 &created,
306 &ps.Workflow,
307 &ps.Status,
308 &ps.Error,
309 &ps.ExitCode,
310 )
311 if err != nil {
312 return nil, err
313 }
314
315 ps.Created, err = time.Parse(time.RFC3339, created)
316 if err != nil {
317 return nil, fmt.Errorf("invalid status created timestamp %q: %w", created, err)
318 }
319
320 pipelineAt := ps.PipelineAt()
321
322 // extract
323 pipeline, ok := pipelines[pipelineAt]
324 if !ok {
325 continue
326 }
327 statuses, _ := pipeline.Statuses[ps.Workflow]
328 if !ok {
329 pipeline.Statuses[ps.Workflow] = models.WorkflowStatus{}
330 }
331
332 // append
333 statuses.Data = append(statuses.Data, ps)
334
335 // reassign
336 pipeline.Statuses[ps.Workflow] = statuses
337 pipelines[pipelineAt] = pipeline
338 }
339
340 var all []models.Pipeline
341 for _, p := range pipelines {
342 for _, s := range p.Statuses {
343 slices.SortFunc(s.Data, func(a, b models.PipelineStatus) int {
344 if a.Created.After(b.Created) {
345 return 1
346 }
347 if a.Created.Before(b.Created) {
348 return -1
349 }
350 if a.ID > b.ID {
351 return 1
352 }
353 if a.ID < b.ID {
354 return -1
355 }
356 return 0
357 })
358 }
359 all = append(all, p)
360 }
361
362 // sort pipelines by date
363 slices.SortFunc(all, func(a, b models.Pipeline) int {
364 if a.Created.After(b.Created) {
365 return -1
366 }
367 return 1
368 })
369
370 return all, nil
371}
372
373// the pipelines table is aliased to `p`
374// the triggers table is aliased to `t`
375func GetTotalPipelineStatuses(e Execer, filters ...orm.Filter) (int64, error) {
376 var conditions []string
377 var args []any
378 for _, filter := range filters {
379 conditions = append(conditions, filter.Condition())
380 args = append(args, filter.Arg()...)
381 }
382
383 whereClause := ""
384 if conditions != nil {
385 whereClause = " where " + strings.Join(conditions, " and ")
386 }
387
388 query := fmt.Sprintf(`
389 select
390 count(1)
391 from
392 pipelines p
393 join
394 triggers t ON p.trigger_id = t.id
395 %s
396 `, whereClause)
397
398 rows, err := e.Query(query, args...)
399 if err != nil {
400 return 0, err
401 }
402 defer rows.Close()
403
404 for rows.Next() {
405 var count int64
406 err := rows.Scan(&count)
407 if err != nil {
408 return 0, err
409 }
410
411 return count, nil
412 }
413
414 // unreachable
415 return 0, nil
416}