Monorepo for Tangled
knotserver,spindle: remove all pipeline logics from knotserver
`sh.tangled.pipeline` events are now completely generated & streamed
from spindle
Signed-off-by: Seongmin Lee <git@boltless.me>
authored by
Seongmin Lee
and committed by
Tangled
(Jul 4, 2026, 8:12 AM +0300)
90e94861
6790668a
+12
-456
-283
knotserver/ingester.go
-283
knotserver/ingester.go
···
4
4
"context"
5
5
"encoding/json"
6
6
"fmt"
7
-
"io"
8
-
"net/http"
9
-
"net/url"
10
-
"path/filepath"
11
7
"strings"
12
8
13
-
comatproto "github.com/bluesky-social/indigo/api/atproto"
14
9
"github.com/bluesky-social/indigo/atproto/syntax"
15
-
"github.com/bluesky-social/indigo/xrpc"
16
10
jmodels "github.com/bluesky-social/jetstream/pkg/models"
17
11
"tangled.org/core/api/tangled"
18
-
"tangled.org/core/appview/models"
19
-
"tangled.org/core/eventstream"
20
12
"tangled.org/core/knotserver/db"
21
-
"tangled.org/core/knotserver/git"
22
13
knotxrpc "tangled.org/core/knotserver/xrpc"
23
14
"tangled.org/core/log"
24
-
"tangled.org/core/tid"
25
-
"tangled.org/core/workflow"
26
15
)
27
16
28
17
func (h *Knot) processPublicKey(ctx context.Context, event *jmodels.Event) error {
···
65
54
DefaultBranch string // default branch
66
55
}
67
56
68
-
func (h *Knot) validatePullRecord(ctx context.Context, record *tangled.RepoPull) (*targetRepo, error) {
69
-
if record.Target == nil {
70
-
return nil, fmt.Errorf("ignoring pull record: target repo is nil")
71
-
}
72
-
73
-
l := log.FromContext(ctx).With("handler", "validatePullRecord")
74
-
l = l.With("target_repo", record.Target.Repo)
75
-
l = l.With("target_branch", record.Target.Branch)
76
-
77
-
if record.Source == nil {
78
-
return nil, fmt.Errorf("ignoring pull record: not a branch-based pull request")
79
-
}
80
-
81
-
if record.Source.Repo != nil {
82
-
return nil, fmt.Errorf("ignoring pull record: fork based pull")
83
-
}
84
-
85
-
var repoPath, ownerDid, repoName, repoDid string
86
-
switch {
87
-
case strings.HasPrefix(record.Target.Repo, "did:"):
88
-
repoDid = record.Target.Repo
89
-
var lookupErr error
90
-
repoPath, ownerDid, repoName, lookupErr = h.db.ResolveRepoDIDOnDisk(h.c.Repo.ScanPath, repoDid)
91
-
if lookupErr != nil {
92
-
return nil, fmt.Errorf("unknown target repo DID %s: %w", repoDid, lookupErr)
93
-
}
94
-
95
-
case strings.Contains(record.Target.Repo, "/"):
96
-
// TODO: get rid of this PDS fetch once all repos have DIDs
97
-
repoAt, parseErr := syntax.ParseATURI(record.Target.Repo)
98
-
if parseErr != nil {
99
-
return nil, fmt.Errorf("failed to parse ATURI: %w", parseErr)
100
-
}
101
-
102
-
ident, resolveErr := h.resolver.ResolveIdent(ctx, repoAt.Authority().String())
103
-
if resolveErr != nil || ident.Handle.IsInvalidHandle() {
104
-
return nil, fmt.Errorf("failed to resolve handle: %w", resolveErr)
105
-
}
106
-
107
-
xrpcc := xrpc.Client{
108
-
Host: ident.PDSEndpoint(),
109
-
}
110
-
111
-
resp, getErr := comatproto.RepoGetRecord(ctx, &xrpcc, "", tangled.RepoNSID, repoAt.Authority().String(), repoAt.RecordKey().String())
112
-
if getErr != nil {
113
-
return nil, fmt.Errorf("failed to resolve repo: %w", getErr)
114
-
}
115
-
116
-
repo, ok := resp.Value.Val.(*tangled.Repo)
117
-
if !ok {
118
-
return nil, fmt.Errorf("record at %s is not a tangled.Repo", repoAt)
119
-
}
120
-
121
-
if repo.Knot != h.c.Server.Hostname {
122
-
return nil, fmt.Errorf("rejected pull record: not this knot, %s != %s", repo.Knot, h.c.Server.Hostname)
123
-
}
124
-
125
-
ownerDid = ident.DID.String()
126
-
repoName = repoAt.RecordKey().String()
127
-
128
-
repoDid, didErr := h.db.GetRepoDid(ownerDid, repoName)
129
-
if didErr != nil {
130
-
return nil, fmt.Errorf("failed to resolve repo DID for %s/%s: %w", ownerDid, repoName, didErr)
131
-
}
132
-
133
-
var lookupErr error
134
-
repoPath, _, _, lookupErr = h.db.ResolveRepoDIDOnDisk(h.c.Repo.ScanPath, repoDid)
135
-
if lookupErr != nil {
136
-
return nil, fmt.Errorf("failed to resolve repo on disk: %w", lookupErr)
137
-
}
138
-
139
-
default:
140
-
return nil, fmt.Errorf("ignoring pull record: target repo has unrecognized format: %s", record.Target.Repo)
141
-
}
142
-
143
-
gr, err := git.Open(repoPath, record.Source.Branch)
144
-
if err != nil {
145
-
return nil, fmt.Errorf("failed to open git repository: %w", err)
146
-
}
147
-
148
-
defaultBranch, _ := gr.FindMainBranch()
149
-
150
-
return &targetRepo{
151
-
RepoPath: repoPath,
152
-
OwnerDid: ownerDid,
153
-
RepoName: repoName,
154
-
RepoDid: repoDid,
155
-
DefaultBranch: defaultBranch,
156
-
}, nil
157
-
}
158
-
159
-
func (h *Knot) fetchLatestSubmission(ctx context.Context, did, rkey string, record *tangled.RepoPull) (*models.PullSubmission, error) {
160
-
// resolve the PR owner's identity to fetch the blob from their PDS
161
-
prOwnerIdent, err := h.resolver.ResolveIdent(ctx, did)
162
-
if err != nil || prOwnerIdent.Handle.IsInvalidHandle() {
163
-
return nil, fmt.Errorf("failed to resolve PR owner handle: %w", err)
164
-
}
165
-
166
-
if len(record.Rounds) == 0 {
167
-
return nil, fmt.Errorf("failed to fetch latest submission, no rounds in record")
168
-
}
169
-
170
-
roundNumber := len(record.Rounds) - 1
171
-
round := record.Rounds[roundNumber]
172
-
173
-
// fetch the blob from the PR owner's PDS
174
-
prOwnerPds := prOwnerIdent.PDSEndpoint()
175
-
blobUrl, err := url.Parse(fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob", prOwnerPds))
176
-
if err != nil {
177
-
return nil, fmt.Errorf("failed to construct blob URL: %w", err)
178
-
}
179
-
q := blobUrl.Query()
180
-
q.Set("cid", round.PatchBlob.Ref.String())
181
-
q.Set("did", did)
182
-
blobUrl.RawQuery = q.Encode()
183
-
184
-
req, err := http.NewRequestWithContext(ctx, http.MethodGet, blobUrl.String(), nil)
185
-
if err != nil {
186
-
return nil, fmt.Errorf("failed to create blob request: %w", err)
187
-
}
188
-
req.Header.Set("Content-Type", "application/json")
189
-
190
-
blobResp, err := http.DefaultClient.Do(req)
191
-
if err != nil {
192
-
return nil, fmt.Errorf("failed to fetch blob: %w", err)
193
-
}
194
-
defer blobResp.Body.Close()
195
-
196
-
blob := io.ReadCloser(blobResp.Body)
197
-
latestSubmission, err := models.PullSubmissionFromRecord(did, rkey, roundNumber, round, &blob)
198
-
if err != nil {
199
-
return nil, fmt.Errorf("failed to parse submission: %w", err)
200
-
}
201
-
202
-
return latestSubmission, nil
203
-
}
204
-
205
-
func (h *Knot) discoverWorkflows(ctx context.Context, repoPath, sha string) (workflow.RawPipeline, error) {
206
-
gr, err := git.Open(repoPath, sha)
207
-
if err != nil {
208
-
return nil, fmt.Errorf("failed to open git repository: %w", err)
209
-
}
210
-
211
-
workflowDir, err := gr.FileTree(ctx, workflow.WorkflowDir)
212
-
if err != nil {
213
-
return nil, fmt.Errorf("failed to open workflow directory: %w", err)
214
-
}
215
-
216
-
var pipeline workflow.RawPipeline
217
-
for _, e := range workflowDir {
218
-
if !e.IsFile() {
219
-
continue
220
-
}
221
-
222
-
fpath := filepath.Join(workflow.WorkflowDir, e.Name)
223
-
contents, err := gr.RawContent(fpath)
224
-
if err != nil {
225
-
continue
226
-
}
227
-
228
-
pipeline = append(pipeline, workflow.RawWorkflow{
229
-
Name: e.Name,
230
-
Contents: contents,
231
-
})
232
-
}
233
-
234
-
return pipeline, nil
235
-
}
236
-
237
-
func (h *Knot) compilePipeline(ctx context.Context, targetRepo *targetRepo, sourceBranch, sourceSha, targetBranch string, rawPipeline workflow.RawPipeline) tangled.Pipeline {
238
-
l := log.FromContext(ctx)
239
-
240
-
trigger := tangled.Pipeline_PullRequestTriggerData{
241
-
Action: "create",
242
-
SourceBranch: sourceBranch,
243
-
SourceSha: sourceSha,
244
-
TargetBranch: targetBranch,
245
-
}
246
-
247
-
compiler := workflow.Compiler{
248
-
Trigger: tangled.Pipeline_TriggerMetadata{
249
-
Kind: string(workflow.TriggerKindPullRequest),
250
-
PullRequest: &trigger,
251
-
Repo: &tangled.Pipeline_TriggerRepo{
252
-
Knot: h.c.Server.Hostname,
253
-
RepoDid: &targetRepo.RepoDid,
254
-
Did: targetRepo.OwnerDid,
255
-
Repo: &targetRepo.RepoName,
256
-
DefaultBranch: targetRepo.DefaultBranch,
257
-
},
258
-
},
259
-
}
260
-
261
-
l.Info("raw", "raw", rawPipeline)
262
-
parsed := compiler.Parse(rawPipeline)
263
-
l.Info("parsed", "parsed", parsed)
264
-
compiled := compiler.Compile(parsed)
265
-
266
-
l.Info("compiler diagnostics", "diagnostics", compiler.Diagnostics)
267
-
268
-
return compiled
269
-
}
270
-
271
-
func (h *Knot) processPull(ctx context.Context, event *jmodels.Event) error {
272
-
raw := json.RawMessage(event.Commit.Record)
273
-
rkey := event.Commit.RKey
274
-
did := event.Did
275
-
276
-
var record tangled.RepoPull
277
-
if err := json.Unmarshal(raw, &record); err != nil {
278
-
return fmt.Errorf("failed to unmarshal record: %w", err)
279
-
}
280
-
281
-
l := log.FromContext(ctx)
282
-
l = l.With("handler", "processPull")
283
-
l = l.With("did", did)
284
-
285
-
l.Info("validating pull record")
286
-
targetRepo, err := h.validatePullRecord(ctx, &record)
287
-
if err != nil {
288
-
l.Warn("pull record did not validate, skipping...")
289
-
return err
290
-
}
291
-
292
-
l = l.With("target_repo", record.Target.Repo)
293
-
l = l.With("target_branch", record.Target.Branch)
294
-
295
-
l.Info("fetching latest submission")
296
-
latestSubmission, err := h.fetchLatestSubmission(ctx, did, rkey, &record)
297
-
if err != nil {
298
-
return err
299
-
}
300
-
301
-
sha := latestSubmission.SourceRev
302
-
if sha == "" {
303
-
return fmt.Errorf("failed to extract source SHA from pull submission")
304
-
}
305
-
l = l.With("sha", sha)
306
-
307
-
l.Info("discovering workflows", "repo_path", targetRepo.RepoPath)
308
-
pipeline, err := h.discoverWorkflows(ctx, targetRepo.RepoPath, sha)
309
-
if err != nil {
310
-
return err
311
-
}
312
-
313
-
l.Info("compiling pipeline", "workflow_count", len(pipeline))
314
-
cp := h.compilePipeline(ctx, targetRepo, record.Source.Branch, sha, record.Target.Branch, pipeline)
315
-
316
-
// do not run empty pipelines
317
-
if cp.Workflows == nil {
318
-
l.Info("skipping empty pipeline")
319
-
return nil
320
-
}
321
-
322
-
l.Info("marshaling pipeline event")
323
-
eventJson, err := json.Marshal(cp)
324
-
if err != nil {
325
-
return fmt.Errorf("failed to marshal pipeline event: %w", err)
326
-
}
327
-
328
-
ev := eventstream.Event{
329
-
Rkey: tid.TID(),
330
-
Nsid: tangled.PipelineNSID,
331
-
EventJson: eventJson,
332
-
}
333
-
334
-
l.Info("inserting pipeline event")
335
-
return h.db.InsertEvent(ev, h.n)
336
-
}
337
-
338
57
func (h *Knot) processRepo(ctx context.Context, event *jmodels.Event) error {
339
58
l := log.FromContext(ctx).With("handler", "processRepo", "did", event.Did, "rkey", event.Commit.RKey)
340
59
···
407
126
err = h.processPublicKey(ctx, event)
408
127
case tangled.RepoNSID:
409
128
err = h.processRepo(ctx, event)
410
-
case tangled.RepoPullNSID:
411
-
err = h.processPull(ctx, event)
412
129
}
413
130
default:
414
131
return nil
+7
-136
knotserver/internal.go
+7
-136
knotserver/internal.go
···
28
28
"tangled.org/core/notifier"
29
29
"tangled.org/core/rbac"
30
30
"tangled.org/core/tid"
31
-
"tangled.org/core/workflow"
32
31
)
33
32
34
33
type InternalHandle struct {
···
188
187
fmt.Fprint(w, diskRelative)
189
188
}
190
189
191
-
type PushOptions struct {
192
-
skipCi bool
193
-
verboseCi bool
194
-
}
195
-
196
190
func (h *InternalHandle) PostReceiveHook(w http.ResponseWriter, r *http.Request) {
197
191
l := h.l.With("handler", "PostReceiveHook")
198
192
···
263
257
l.Error("failed to reply with pull request link", "err", err, "line", line, "did", gitUserDid, "repo", gitRelativeDir)
264
258
}
265
259
266
-
err = h.triggerPipeline(&resp.Messages, line, gitUserDid, ownerDid, repoName, repoDid, pushOptions)
267
-
if err != nil {
268
-
l.Error("failed to trigger pipeline", "err", err, "line", line, "did", gitUserDid, "repo", gitRelativeDir)
260
+
// emit pipeline logs link
261
+
if h.c.LogsAddr != "" {
262
+
host, port, err := net.SplitHostPort(h.c.LogsAddr)
263
+
if err == nil {
264
+
resp.Messages = append(resp.Messages, "→ Browse CI logs in your terminal:")
265
+
resp.Messages = append(resp.Messages, fmt.Sprintf(" ssh -t -p %s %s %s %s", port, host, repoDid, line.NewSha))
266
+
}
269
267
}
270
268
}
271
269
···
319
317
Rkey: tid.TID(),
320
318
Nsid: tangled.GitRefUpdateNSID,
321
319
EventJson: eventJson,
322
-
}
323
-
324
-
return h.db.InsertEvent(event, h.n)
325
-
}
326
-
327
-
func (h *InternalHandle) triggerPipeline(
328
-
clientMsgs *[]string,
329
-
line git.PostReceiveLine,
330
-
gitUserDid string,
331
-
ownerDid string,
332
-
repoName string,
333
-
repoDid string,
334
-
pushOptionsRaw []string,
335
-
) error {
336
-
var pushOptions PushOptions
337
-
for _, option := range pushOptionsRaw {
338
-
if option == "skip-ci" || option == "ci-skip" {
339
-
pushOptions.skipCi = true
340
-
}
341
-
if option == "verbose-ci" || option == "ci-verbose" {
342
-
pushOptions.verboseCi = true
343
-
}
344
-
}
345
-
if pushOptions.skipCi {
346
-
return nil
347
-
}
348
-
349
-
repoPath, _, _, resolveErr := h.db.ResolveRepoDIDOnDisk(h.c.Repo.ScanPath, repoDid)
350
-
if resolveErr != nil {
351
-
return fmt.Errorf("failed to resolve repo on disk: %w", resolveErr)
352
-
}
353
-
354
-
gr, err := git.Open(repoPath, line.Ref)
355
-
if err != nil {
356
-
return err
357
-
}
358
-
359
-
workflowDir, err := gr.FileTree(context.Background(), workflow.WorkflowDir)
360
-
if err != nil {
361
-
return err
362
-
}
363
-
364
-
var pipeline workflow.RawPipeline
365
-
for _, e := range workflowDir {
366
-
if !e.IsFile() {
367
-
continue
368
-
}
369
-
370
-
fpath := filepath.Join(workflow.WorkflowDir, e.Name)
371
-
contents, err := gr.RawContent(fpath)
372
-
if err != nil {
373
-
continue
374
-
}
375
-
376
-
pipeline = append(pipeline, workflow.RawWorkflow{
377
-
Name: e.Name,
378
-
Contents: contents,
379
-
})
380
-
}
381
-
382
-
defaultBranch, _ := gr.FindMainBranch()
383
-
384
-
trigger := tangled.Pipeline_PushTriggerData{
385
-
Ref: line.Ref,
386
-
OldSha: line.OldSha.String(),
387
-
NewSha: line.NewSha.String(),
388
-
}
389
-
390
-
triggerRepo := &tangled.Pipeline_TriggerRepo{
391
-
Did: ownerDid,
392
-
Knot: h.c.Server.Hostname,
393
-
Repo: &repoName,
394
-
RepoDid: &repoDid,
395
-
DefaultBranch: defaultBranch,
396
-
}
397
-
398
-
changedFiles, err := gr.ChangedFilesBetween(line.OldSha.String(), line.NewSha.String())
399
-
if err != nil {
400
-
return fmt.Errorf("getting changed files: %w", err)
401
-
}
402
-
403
-
compiler := workflow.Compiler{
404
-
Trigger: tangled.Pipeline_TriggerMetadata{
405
-
Kind: string(workflow.TriggerKindPush),
406
-
Push: &trigger,
407
-
Repo: triggerRepo,
408
-
},
409
-
ChangedFiles: changedFiles,
410
-
}
411
-
412
-
cp := compiler.Compile(compiler.Parse(pipeline))
413
-
eventJson, err := json.Marshal(cp)
414
-
if err != nil {
415
-
return err
416
-
}
417
-
418
-
for _, e := range compiler.Diagnostics.Errors {
419
-
*clientMsgs = append(*clientMsgs, e.String())
420
-
}
421
-
422
-
if pushOptions.verboseCi {
423
-
if compiler.Diagnostics.IsEmpty() {
424
-
*clientMsgs = append(*clientMsgs, "success: pipeline compiled with no diagnostics")
425
-
}
426
-
427
-
for _, w := range compiler.Diagnostics.Warnings {
428
-
*clientMsgs = append(*clientMsgs, w.String())
429
-
}
430
-
}
431
-
432
-
// do not run empty pipelines
433
-
if cp.Workflows == nil {
434
-
return nil
435
-
}
436
-
437
-
event := eventstream.Event{
438
-
Rkey: tid.TID(),
439
-
Nsid: tangled.PipelineNSID,
440
-
EventJson: eventJson,
441
-
}
442
-
443
-
if h.c.LogsAddr != "" {
444
-
host, port, err := net.SplitHostPort(h.c.LogsAddr)
445
-
if err == nil {
446
-
*clientMsgs = append(*clientMsgs, "→ Browse CI logs in your terminal:")
447
-
*clientMsgs = append(*clientMsgs, fmt.Sprintf(" ssh -t -p %s %s %s %s", port, host, repoDid, line.NewSha))
448
-
}
449
320
}
450
321
451
322
return h.db.InsertEvent(event, h.n)
-1
knotserver/server.go
-1
knotserver/server.go
···
96
96
jc, err := jetstream.NewJetstreamClient(c.Server.JetstreamEndpoint, "knotserver", []string{
97
97
tangled.PublicKeyNSID,
98
98
tangled.RepoNSID,
99
-
tangled.RepoPullNSID,
100
99
}, nil, log.SubLogger(logger, "jetstream"), db, true, c.Server.LogDids)
101
100
if err != nil {
102
101
logger.Error("failed to setup jetstream", "error", err)
+5
-36
spindle/server.go
+5
-36
spindle/server.go
···
404
404
func (s *Spindle) processKnotStream(ctx context.Context, src eventconsumer.Source, msg eventstream.Event) error {
405
405
l := log.FromContext(ctx).With("handler", "processKnotStream")
406
406
l = l.With("src", src.Key(), "msg.Nsid", msg.Nsid, "msg.Rkey", msg.Rkey)
407
-
if msg.Nsid == tangled.PipelineNSID {
408
-
return nil
409
-
tpl := tangled.Pipeline{}
410
-
err := json.Unmarshal(msg.EventJson, &tpl)
411
-
if err != nil {
412
-
s.l.Error("failed to unmarshal pipeline event", "err", err)
413
-
return err
414
-
}
415
-
416
-
if tpl.TriggerMetadata == nil {
417
-
return fmt.Errorf("no trigger metadata found")
418
-
}
419
-
420
-
if tpl.TriggerMetadata.Repo == nil {
421
-
return fmt.Errorf("no repo data found")
422
-
}
423
-
424
-
if src.Host != tpl.TriggerMetadata.Repo.Knot {
425
-
return fmt.Errorf("repo knot does not match event source: %s != %s", src.Host, tpl.TriggerMetadata.Repo.Knot)
426
-
}
427
-
428
-
repoDid, err := s.resolvePipelineRepoDid(tpl.TriggerMetadata.Repo)
429
-
if err != nil {
430
-
return err
431
-
}
432
-
433
-
pipelineId := models.PipelineId{
434
-
Knot: src.Host,
435
-
Rkey: msg.Rkey,
436
-
}
437
-
438
-
err = s.processPipeline(ctx, repoDid, tpl, pipelineId)
439
-
if err != nil {
440
-
return err
441
-
}
442
-
} else if msg.Nsid == tangled.GitRefUpdateNSID {
407
+
if msg.Nsid == tangled.GitRefUpdateNSID {
443
408
event := tangled.GitRefUpdate{}
444
409
if err := json.Unmarshal(msg.EventJson, &event); err != nil {
445
410
l.Error("error unmarshalling", "err", err)
···
452
417
repo, err := s.db.GetRepoByDid(repoDid)
453
418
if err != nil {
454
419
return fmt.Errorf("unknown repoDid %s: %w", repoDid, err)
420
+
}
421
+
422
+
if src.Host != repo.Knot {
423
+
return fmt.Errorf("repo knot does not match event source: %s != %s", src.Host, repo.Knot)
455
424
}
456
425
457
426
// NOTE: we are blindly trusting the knot that it will return only repos it own