tangled.org
Monorepo for Tangled
tangled.org
appview: listen for pipeline events from spindlestream
Signed-off-by: Seongmin Lee <git@boltless.me>
boltless.me
-86
appview/state/knotstream.go
-86
appview/state/knotstream.go
···
20
20
"tangled.org/core/log"
21
21
"tangled.org/core/orm"
22
22
"tangled.org/core/rbac"
23
-
"tangled.org/core/workflow"
24
23
25
-
"github.com/bluesky-social/indigo/atproto/syntax"
26
24
"github.com/go-git/go-git/v5/plumbing"
27
25
"github.com/posthog/posthog-go"
28
26
)
···
69
67
switch msg.Nsid {
70
68
case tangled.GitRefUpdateNSID:
71
69
return ingestRefUpdate(d, enforcer, posthog, notifier, dev, source, msg, ctx)
72
-
case tangled.PipelineNSID:
73
-
return ingestPipeline(d, source, msg)
74
70
}
75
71
76
72
return nil
···
219
215
220
216
return tx.Commit()
221
217
}
222
-
223
-
func ingestPipeline(d *db.DB, source ec.Source, msg ec.Message) error {
224
-
var record tangled.Pipeline
225
-
err := json.Unmarshal(msg.EventJson, &record)
226
-
if err != nil {
227
-
return err
228
-
}
229
-
230
-
if record.TriggerMetadata == nil {
231
-
return fmt.Errorf("empty trigger metadata: nsid %s, rkey %s", msg.Nsid, msg.Rkey)
232
-
}
233
-
234
-
if record.TriggerMetadata.Repo == nil {
235
-
return fmt.Errorf("empty repo: nsid %s, rkey %s", msg.Nsid, msg.Rkey)
236
-
}
237
-
238
-
// does this repo have a spindle configured?
239
-
repos, err := db.GetRepos(
240
-
d,
241
-
0,
242
-
orm.FilterEq("did", record.TriggerMetadata.Repo.Did),
243
-
orm.FilterEq("name", record.TriggerMetadata.Repo.Repo),
244
-
)
245
-
if err != nil {
246
-
return fmt.Errorf("failed to look for repo in DB: nsid %s, rkey %s, %w", msg.Nsid, msg.Rkey, err)
247
-
}
248
-
if len(repos) != 1 {
249
-
return fmt.Errorf("incorrect number of repos returned: %d (expected 1)", len(repos))
250
-
}
251
-
if repos[0].Spindle == "" {
252
-
return fmt.Errorf("repo does not have a spindle configured yet: nsid %s, rkey %s", msg.Nsid, msg.Rkey)
253
-
}
254
-
255
-
// trigger info
256
-
var trigger models.Trigger
257
-
var sha string
258
-
trigger.Kind = workflow.TriggerKind(record.TriggerMetadata.Kind)
259
-
switch trigger.Kind {
260
-
case workflow.TriggerKindPush:
261
-
trigger.PushRef = &record.TriggerMetadata.Push.Ref
262
-
trigger.PushNewSha = &record.TriggerMetadata.Push.NewSha
263
-
trigger.PushOldSha = &record.TriggerMetadata.Push.OldSha
264
-
sha = *trigger.PushNewSha
265
-
case workflow.TriggerKindPullRequest:
266
-
trigger.PRSourceBranch = &record.TriggerMetadata.PullRequest.SourceBranch
267
-
trigger.PRTargetBranch = &record.TriggerMetadata.PullRequest.TargetBranch
268
-
trigger.PRSourceSha = &record.TriggerMetadata.PullRequest.SourceSha
269
-
trigger.PRAction = &record.TriggerMetadata.PullRequest.Action
270
-
sha = *trigger.PRSourceSha
271
-
}
272
-
273
-
tx, err := d.Begin()
274
-
if err != nil {
275
-
return fmt.Errorf("failed to start txn: %w", err)
276
-
}
277
-
278
-
triggerId, err := db.AddTrigger(tx, trigger)
279
-
if err != nil {
280
-
return fmt.Errorf("failed to add trigger entry: %w", err)
281
-
}
282
-
283
-
pipeline := models.Pipeline{
284
-
Rkey: msg.Rkey,
285
-
Knot: source.Key(),
286
-
RepoOwner: syntax.DID(record.TriggerMetadata.Repo.Did),
287
-
RepoName: record.TriggerMetadata.Repo.Repo,
288
-
TriggerId: int(triggerId),
289
-
Sha: sha,
290
-
}
291
-
292
-
err = db.AddPipeline(tx, pipeline)
293
-
if err != nil {
294
-
return fmt.Errorf("failed to add pipeline: %w", err)
295
-
}
296
-
297
-
err = tx.Commit()
298
-
if err != nil {
299
-
return fmt.Errorf("failed to commit txn: %w", err)
300
-
}
301
-
302
-
return nil
303
-
}
+89
appview/state/spindlestream.go
+89
appview/state/spindlestream.go
···
20
20
"tangled.org/core/orm"
21
21
"tangled.org/core/rbac"
22
22
spindle "tangled.org/core/spindle/models"
23
+
"tangled.org/core/workflow"
23
24
)
24
25
25
26
func Spindlestream(ctx context.Context, c *config.Config, d *db.DB, enforcer *rbac.Enforcer) (*ec.Consumer, error) {
···
62
63
func spindleIngester(ctx context.Context, logger *slog.Logger, d *db.DB) ec.ProcessFunc {
63
64
return func(ctx context.Context, source ec.Source, msg ec.Message) error {
64
65
switch msg.Nsid {
66
+
case tangled.PipelineNSID:
67
+
return ingestPipeline(logger, d, source, msg)
65
68
case tangled.PipelineStatusNSID:
66
69
return ingestPipelineStatus(ctx, logger, d, source, msg)
67
70
}
68
71
69
72
return nil
70
73
}
74
+
}
75
+
76
+
func ingestPipeline(l *slog.Logger, d *db.DB, source ec.Source, msg ec.Message) error {
77
+
var record tangled.Pipeline
78
+
err := json.Unmarshal(msg.EventJson, &record)
79
+
if err != nil {
80
+
return err
81
+
}
82
+
83
+
if record.TriggerMetadata == nil {
84
+
return fmt.Errorf("empty trigger metadata: nsid %s, rkey %s", msg.Nsid, msg.Rkey)
85
+
}
86
+
87
+
if record.TriggerMetadata.Repo == nil {
88
+
return fmt.Errorf("empty repo: nsid %s, rkey %s", msg.Nsid, msg.Rkey)
89
+
}
90
+
91
+
// does this repo have a spindle configured?
92
+
repos, err := db.GetRepos(
93
+
d,
94
+
0,
95
+
orm.FilterEq("did", record.TriggerMetadata.Repo.Did),
96
+
orm.FilterEq("name", record.TriggerMetadata.Repo.Repo),
97
+
)
98
+
if err != nil {
99
+
return fmt.Errorf("failed to look for repo in DB: nsid %s, rkey %s, %w", msg.Nsid, msg.Rkey, err)
100
+
}
101
+
if len(repos) != 1 {
102
+
return fmt.Errorf("incorrect number of repos returned: %d (expected 1)", len(repos))
103
+
}
104
+
if repos[0].Spindle == "" {
105
+
return fmt.Errorf("repo does not have a spindle configured yet: nsid %s, rkey %s", msg.Nsid, msg.Rkey)
106
+
}
107
+
108
+
// trigger info
109
+
var trigger models.Trigger
110
+
var sha string
111
+
trigger.Kind = workflow.TriggerKind(record.TriggerMetadata.Kind)
112
+
switch trigger.Kind {
113
+
case workflow.TriggerKindPush:
114
+
trigger.PushRef = &record.TriggerMetadata.Push.Ref
115
+
trigger.PushNewSha = &record.TriggerMetadata.Push.NewSha
116
+
trigger.PushOldSha = &record.TriggerMetadata.Push.OldSha
117
+
sha = *trigger.PushNewSha
118
+
case workflow.TriggerKindPullRequest:
119
+
trigger.PRSourceBranch = &record.TriggerMetadata.PullRequest.SourceBranch
120
+
trigger.PRTargetBranch = &record.TriggerMetadata.PullRequest.TargetBranch
121
+
trigger.PRSourceSha = &record.TriggerMetadata.PullRequest.SourceSha
122
+
trigger.PRAction = &record.TriggerMetadata.PullRequest.Action
123
+
sha = *trigger.PRSourceSha
124
+
}
125
+
126
+
tx, err := d.Begin()
127
+
if err != nil {
128
+
return fmt.Errorf("failed to start txn: %w", err)
129
+
}
130
+
131
+
triggerId, err := db.AddTrigger(tx, trigger)
132
+
if err != nil {
133
+
return fmt.Errorf("failed to add trigger entry: %w", err)
134
+
}
135
+
136
+
// TODO: we shouldn't even use knot to identify pipelines
137
+
knot := record.TriggerMetadata.Repo.Knot
138
+
pipeline := models.Pipeline{
139
+
Rkey: msg.Rkey,
140
+
Knot: knot,
141
+
RepoOwner: syntax.DID(record.TriggerMetadata.Repo.Did),
142
+
RepoName: record.TriggerMetadata.Repo.Repo,
143
+
TriggerId: int(triggerId),
144
+
Sha: sha,
145
+
}
146
+
147
+
err = db.AddPipeline(tx, pipeline)
148
+
if err != nil {
149
+
return fmt.Errorf("failed to add pipeline: %w", err)
150
+
}
151
+
152
+
err = tx.Commit()
153
+
if err != nil {
154
+
return fmt.Errorf("failed to commit txn: %w", err)
155
+
}
156
+
157
+
l.Info("added pipeline", "pipeline", pipeline)
158
+
159
+
return nil
71
160
}
72
161
73
162
func ingestPipelineStatus(ctx context.Context, logger *slog.Logger, d *db.DB, source ec.Source, msg ec.Message) error {