alice.pds.demo.boltless.dev
This repository has no description
1package db
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7 "tangled.org/core/api/tangled"
8 "tangled.org/core/spindle/models"
9)
10
11type JobRow struct {
12 Id int64
13 RepoDid string
14 PipelineIdKnot string
15 PipelineIdRkey string
16 SourceRepo *tangled.Pipeline_TriggerRepo
17 Tpl tangled.Pipeline
18}
19
20func (d *DB) EnqueueJob(ctx context.Context, repoDid string, pipelineId models.PipelineId, sourceRepo *tangled.Pipeline_TriggerRepo, tpl tangled.Pipeline) error {
21 tplJson, err := json.Marshal(tpl)
22 if err != nil {
23 return err
24 }
25 _, err = d.ExecContext(ctx, `
26 insert into jobs (repo_did, pipeline_id_knot, pipeline_id_rkey, source_repo, tpl)
27 values (?, ?, ?, ?, ?)
28 `, repoDid, pipelineId.Knot, pipelineId.Rkey, string(sourceRepoJson(sourceRepo)), string(tplJson))
29 return err
30}
31func (d *DB) DequeueJob(ctx context.Context) (*JobRow, error) {
32 var row JobRow
33 var sourceRepoStr *string
34 var tplJson string
35 err := d.QueryRowContext(ctx, `
36 delete from jobs
37 where id = (
38 select id from jobs
39 order by id asc
40 limit 1
41 )
42 returning id, repo_did, pipeline_id_knot, pipeline_id_rkey, source_repo, tpl
43 `).Scan(&row.Id, &row.RepoDid, &row.PipelineIdKnot, &row.PipelineIdRkey, &sourceRepoStr, &tplJson)
44 if err != nil {
45 if err == sql.ErrNoRows {
46 return nil, nil
47 }
48 return nil, err
49 }
50 if err := json.Unmarshal([]byte(tplJson), &row.Tpl); err != nil {
51 return nil, err
52 }
53 if sourceRepoStr != nil {
54 row.SourceRepo = &tangled.Pipeline_TriggerRepo{}
55 if err := json.Unmarshal([]byte(*sourceRepoStr), row.SourceRepo); err != nil {
56 return nil, err
57 }
58 }
59 return &row, nil
60}
61
62func sourceRepoJson(sr *tangled.Pipeline_TriggerRepo) []byte {
63 if sr == nil {
64 return nil
65 }
66 b, _ := json.Marshal(sr)
67 return b
68}