alice.pds.demo.boltless.dev
This repository has no description
1package db
2
3import (
4 "encoding/json"
5 "time"
6
7 "tangled.org/core/api/tangled"
8 "tangled.org/core/eventstream"
9 "tangled.org/core/notifier"
10 "tangled.org/core/spindle/models"
11 "tangled.org/core/tid"
12)
13
14func (d *DB) insertEvent(event eventstream.Event, n *notifier.Notifier) error {
15 return eventstream.Insert(d, event, n)
16}
17
18func (d *DB) GetEvents(cursor int64, limit int) ([]eventstream.Event, error) {
19 return eventstream.List(d, cursor, limit)
20}
21
22func (d *DB) EventHighWater() (int64, error) {
23 return eventstream.HighWater(d)
24}
25
26func (d *DB) CreatePipelineEvent(rkey string, pipeline tangled.Pipeline, n *notifier.Notifier) error {
27 eventJson, err := json.Marshal(pipeline)
28 if err != nil {
29 return err
30 }
31 event := eventstream.Event{
32 Rkey: rkey,
33 Nsid: tangled.PipelineNSID,
34 EventJson: eventJson,
35 }
36 return d.insertEvent(event, n)
37}
38
39// leaves created at zero so insertevent stamps the local clock
40func statusEvent(pipelineAtUri, workflow, status string, workflowError *string, exitCode *int64) (eventstream.Event, error) {
41 s := tangled.PipelineStatus{
42 CreatedAt: time.Now().Format(time.RFC3339),
43 Error: workflowError,
44 ExitCode: exitCode,
45 Pipeline: pipelineAtUri,
46 Workflow: workflow,
47 Status: status,
48 }
49
50 eventJson, err := json.Marshal(s)
51 if err != nil {
52 return eventstream.Event{}, err
53 }
54
55 return eventstream.Event{
56 Rkey: tid.TID(),
57 Nsid: tangled.PipelineStatusNSID,
58 EventJson: eventJson,
59 }, nil
60}
61
62func (d *DB) createStatusEvent(
63 workflowId models.WorkflowId,
64 statusKind models.StatusKind,
65 workflowError *string,
66 exitCode *int64,
67 n *notifier.Notifier,
68) error {
69 event, err := statusEvent(string(workflowId.PipelineId.AtUri()), workflowId.Name, string(statusKind), workflowError, exitCode)
70 if err != nil {
71 return err
72 }
73 return d.insertEvent(event, n)
74}
75
76// stamps the mill's own clock so it orders against the cursor like a local write
77func (d *DB) InsertEventStatus(
78 pipelineAtUri string,
79 workflow string,
80 status string,
81 workflowError *string,
82 exitCode *int64,
83 n *notifier.Notifier,
84) error {
85 event, err := statusEvent(pipelineAtUri, workflow, status, workflowError, exitCode)
86 if err != nil {
87 return err
88 }
89 return d.insertEvent(event, n)
90}
91
92// deleting the lease in the same transaction prevents the terminal event
93// from replaying
94func (d *DB) CompleteMillLease(
95 leaseID string,
96 pipelineAtUri string,
97 workflow string,
98 status string,
99 workflowError *string,
100 exitCode *int64,
101 n *notifier.Notifier,
102) error {
103 return d.ApplyEventBatch(n, func(tx *EventBatchTx) error {
104 if err := tx.InsertStatusEvent(pipelineAtUri, workflow, status, workflowError, exitCode); err != nil {
105 return err
106 }
107 return tx.DeleteLease(leaseID)
108 })
109}
110
111func (d *DB) GetStatus(workflowId models.WorkflowId) (*tangled.PipelineStatus, error) {
112 pipelineAtUri := workflowId.PipelineId.AtUri()
113
114 var eventJson string
115 err := d.QueryRow(
116 `
117 select
118 event from events
119 where
120 nsid = ?
121 and json_extract(event, '$.pipeline') = ?
122 and json_extract(event, '$.workflow') = ?
123 order by
124 created desc
125 limit
126 1
127 `,
128 tangled.PipelineStatusNSID,
129 string(pipelineAtUri),
130 workflowId.Name,
131 ).Scan(&eventJson)
132
133 if err != nil {
134 return nil, err
135 }
136
137 var status tangled.PipelineStatus
138 if err := json.Unmarshal([]byte(eventJson), &status); err != nil {
139 return nil, err
140 }
141
142 return &status, nil
143}
144
145func (d *DB) StatusPending(workflowId models.WorkflowId, n *notifier.Notifier) error {
146 return d.createStatusEvent(workflowId, models.StatusKindPending, nil, nil, n)
147}
148
149func (d *DB) StatusRunning(workflowId models.WorkflowId, n *notifier.Notifier) error {
150 return d.createStatusEvent(workflowId, models.StatusKindRunning, nil, nil, n)
151}
152
153func (d *DB) StatusFailed(workflowId models.WorkflowId, workflowError string, exitCode int64, n *notifier.Notifier) error {
154 return d.createStatusEvent(workflowId, models.StatusKindFailed, &workflowError, &exitCode, n)
155}
156
157func (d *DB) StatusCancelled(workflowId models.WorkflowId, workflowError string, exitCode int64, n *notifier.Notifier) error {
158 return d.createStatusEvent(workflowId, models.StatusKindCancelled, &workflowError, &exitCode, n)
159}
160
161func (d *DB) StatusSuccess(workflowId models.WorkflowId, n *notifier.Notifier) error {
162 return d.createStatusEvent(workflowId, models.StatusKindSuccess, nil, nil, n)
163}
164
165func (d *DB) StatusTimeout(workflowId models.WorkflowId, n *notifier.Notifier) error {
166 return d.createStatusEvent(workflowId, models.StatusKindTimeout, nil, nil, n)
167}