+1

knotserver/internal.go

Reviewed

··· 259 259 } 260 260 261 261 for _, line := range lines { 262 262 + // TODO: pass pushOptions to refUpdate 262 263 err := h.insertRefUpdate(line, gitUserDid, ownerDid, repoDid) 263 264 if err != nil { 264 265 l.Error("failed to insert op", "err", err, "line", line, "did", gitUserDid, "repo", gitRelativeDir)

+14

spindle/db/events.go

Reviewed

··· 19 19 return eventstream.List(d, cursor, limit) 20 20 } 21 21 22 22 + func (d *DB) CreatePipelineEvent(rkey string, pipeline tangled.Pipeline, n *notifier.Notifier) error { 23 23 + eventJson, err := json.Marshal(pipeline) 24 24 + if err != nil { 25 25 + return err 26 26 + } 27 27 + event := eventstream.Event{ 28 28 + Rkey: rkey, 29 29 + Nsid: tangled.PipelineNSID, 30 30 + Created: time.Now().UnixNano(), 31 31 + EventJson: eventJson, 32 32 + } 33 33 + return d.insertEvent(event, n) 34 34 + } 35 35 + 22 36 func (d *DB) createStatusEvent( 23 37 workflowId models.WorkflowId, 24 38 statusKind models.StatusKind,

+2 -2

spindle/embedtap.go

Reviewed

··· 67 67 RepoFetchTimeout: 5 * time.Minute, 68 68 IdentityCacheSize: 50_000, 69 69 EventCacheSize: 10_000, 70 70 - SignalCollection: tangled.RepoNSID, 71 71 - CollectionFilters: []string{tangled.RepoNSID, tangled.RepoCollaboratorNSID}, 70 70 + SignalCollection: tangled.RepoPullNSID, // HACK: to ingest PRs from any users 71 71 + CollectionFilters: []string{tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.RepoPullNSID}, 72 72 AdminPassword: cfg.Server.Tap.AdminPassword, 73 73 RetryTimeout: 60 * time.Second, 74 74 }

+207 -82

spindle/server.go

Reviewed

··· 4 4 "context" 5 5 _ "embed" 6 6 "encoding/json" 7 7 + "errors" 7 8 "fmt" 8 9 "log/slog" 9 10 "maps" ··· 12 13 "sync" 13 14 14 15 "github.com/bluesky-social/indigo/atproto/syntax" 16 16 + indigoxrpc "github.com/bluesky-social/indigo/xrpc" 15 17 "github.com/go-chi/chi/v5" 18 18 + "github.com/go-git/go-git/v5/plumbing/object" 16 19 "github.com/hashicorp/go-version" 17 20 "tangled.org/core/api/tangled" 18 21 "tangled.org/core/eventconsumer" ··· 20 23 "tangled.org/core/eventstream" 21 24 "tangled.org/core/idresolver" 22 25 "tangled.org/core/jetstream" 26 26 + kgit "tangled.org/core/knotserver/git" 23 27 "tangled.org/core/log" 24 28 "tangled.org/core/notifier" 25 29 "tangled.org/core/rbac" ··· 32 36 "tangled.org/core/spindle/queue" 33 37 "tangled.org/core/spindle/secrets" 34 38 "tangled.org/core/spindle/xrpc" 39 39 + "tangled.org/core/tid" 40 40 + "tangled.org/core/workflow" 35 41 "tangled.org/core/xrpc/serviceauth" 36 42 ) 37 43 ··· 186 192 return nil, fmt.Errorf("failed to start jetstream consumer: %w", err) 187 193 } 188 194 189 189 - // for each incoming sh.tangled.pipeline, we execute 190 190 - // spindle.processPipeline, which in turn enqueues the pipeline 191 191 - // job in the above registered queue. 195 195 + // spindle listen to knot stream for sh.tangled.git.refUpdate 196 196 + // which will sync the local workflow files in spindle and enqueues the 197 197 + // pipeline job for on-push workflows 192 198 ccfg := eventconsumer.NewConsumerConfig() 193 199 ccfg.Logger = log.SubLogger(logger, "eventconsumer") 194 194 - ccfg.ProcessFunc = spindle.processPipeline 200 200 + ccfg.ProcessFunc = spindle.processKnotStream 195 201 ccfg.CursorStore = cursorStore 196 202 knownKnots, err := d.Knots() 197 203 if err != nil { ··· 381 387 return x.Router() 382 388 } 383 389 384 384 - func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, msg eventstream.Event) error { 390 390 + func (s *Spindle) processKnotStream(ctx context.Context, src eventconsumer.Source, msg eventstream.Event) error { 385 391 l := log.FromContext(ctx).With("handler", "processKnotStream") 386 392 l = l.With("src", src.Key(), "msg.Nsid", msg.Nsid, "msg.Rkey", msg.Rkey) 387 393 if msg.Nsid == tangled.PipelineNSID { ··· 415 421 Rkey: msg.Rkey, 416 422 } 417 423 418 418 - workflows := make(map[models.Engine][]models.Workflow) 424 424 + err = s.processPipeline(ctx, repoDid, tpl, pipelineId) 425 425 + if err != nil { 426 426 + return err 427 427 + } 428 428 + } else if msg.Nsid == tangled.GitRefUpdateNSID { 429 429 + event := tangled.GitRefUpdate{} 430 430 + if err := json.Unmarshal(msg.EventJson, &event); err != nil { 431 431 + l.Error("error unmarshalling", "err", err) 432 432 + return err 433 433 + } 434 434 + l = l.With("repo", event.Repo, "ref", event.Ref, "newSha", event.NewSha) 435 435 + l.Debug("debug") 436 436 + 437 437 + repoDid := syntax.DID(event.Repo) 438 438 + repo, err := s.db.GetRepoByDid(repoDid) 439 439 + if err != nil { 440 440 + return fmt.Errorf("unknown repoDid %s: %w", repoDid, err) 441 441 + } 442 442 + 443 443 + // NOTE: we are blindly trusting the knot that it will return only repos it own 444 444 + repoCloneUri := s.newRepoCloneUrl(src.Key(), repoDid) 445 445 + repoPath := s.newRepoPath(repoDid) 446 446 + if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, event.NewSha); err != nil { 447 447 + return fmt.Errorf("sync git repo: %w", err) 448 448 + } 449 449 + l.Info("synced git repo") 419 450 420 420 - // Build pipeline environment variables once for all workflows 421 421 - pipelineEnv := models.PipelineEnvVars(tpl.TriggerMetadata, pipelineId, s.cfg.Server.Dev) 451 451 + scheme := "https" 452 452 + if s.cfg.Server.Dev { 453 453 + scheme = "http" 454 454 + } 455 455 + client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, repo.Knot)} 422 456 423 423 - for _, w := range tpl.Workflows { 424 424 - if w != nil { 425 425 - if _, ok := s.engs[w.Engine]; !ok { 426 426 - err = s.db.StatusFailed(models.WorkflowId{ 427 427 - PipelineId: pipelineId, 428 428 - Name: w.Name, 429 429 - }, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n) 430 430 - if err != nil { 431 431 - return fmt.Errorf("db.StatusFailed: %w", err) 432 432 - } 457 457 + // fetch current default branch 458 458 + defaultBranch, _ := func(repo syntax.DID) (string, error) { 459 459 + defaultBranchOut, err := tangled.RepoGetDefaultBranch(ctx, client, repo.String()) 460 460 + if err != nil { 461 461 + return "", err 462 462 + } 463 463 + return defaultBranchOut.Name, nil 464 464 + }(repoDid) 433 465 434 434 - continue 435 435 - } 466 466 + compiler := workflow.Compiler{ 467 467 + Trigger: tangled.Pipeline_TriggerMetadata{ 468 468 + Kind: string(workflow.TriggerKindPush), 469 469 + Push: &tangled.Pipeline_PushTriggerData{ 470 470 + Ref: event.Ref, 471 471 + OldSha: event.OldSha, 472 472 + NewSha: event.NewSha, 473 473 + }, 474 474 + Repo: &tangled.Pipeline_TriggerRepo{ 475 475 + Did: repo.Owner.String(), 476 476 + Knot: repo.Knot, 477 477 + Repo: (*string)(&repo.Rkey), 478 478 + RepoDid: (*string)(&repoDid), 479 479 + DefaultBranch: defaultBranch, 480 480 + }, 481 481 + }, 482 482 + } 436 483 437 437 - eng := s.engs[w.Engine] 484 484 + // load workflow definitions from rev (without spindle context) 485 485 + rawPipeline, err := s.loadPipeline(ctx, repoCloneUri, repoPath, event.NewSha) 486 486 + if err != nil { 487 487 + return fmt.Errorf("loading pipeline: %w", err) 488 488 + } 489 489 + if len(rawPipeline) == 0 { 490 490 + l.Info("no workflow definition find for the repo. skipping the event") 491 491 + return nil 492 492 + } 493 493 + tpl := compiler.Compile(compiler.Parse(rawPipeline)) 494 494 + // TODO: pass compile error to workflow log 495 495 + for _, w := range compiler.Diagnostics.Errors { 496 496 + l.Error(w.String()) 497 497 + } 498 498 + for _, w := range compiler.Diagnostics.Warnings { 499 499 + l.Warn(w.String()) 500 500 + } 501 501 + if len(tpl.Workflows) == 0 { 502 502 + l.Info("no workflow matching trigger 'push'. skipping the event") 503 503 + return nil 504 504 + } 438 505 439 439 - if _, ok := workflows[eng]; !ok { 440 440 - workflows[eng] = []models.Workflow{} 441 441 - } 506 506 + pipelineId := models.PipelineId{ 507 507 + Knot: tpl.TriggerMetadata.Repo.Knot, 508 508 + Rkey: tid.TID(), 509 509 + } 510 510 + if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { 511 511 + l.Error("failed to create pipeline event", "err", err) 512 512 + return nil 513 513 + } 514 514 + err = s.processPipeline(ctx, repoDid, tpl, pipelineId) 515 515 + if err != nil { 516 516 + return err 517 517 + } 518 518 + } 442 519 443 443 - ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl) 444 444 - if err != nil { 445 445 - err = s.db.StatusFailed(models.WorkflowId{ 446 446 - PipelineId: pipelineId, 447 447 - Name: w.Name, 448 448 - }, fmt.Sprintf("init workflow: %s", err), -1, s.n) 449 449 - if err != nil { 450 450 - return fmt.Errorf("db.StatusFailed: %w", err) 451 451 - } 520 520 + return nil 521 521 + } 452 522 453 453 - continue 454 454 - } 523 523 + func (s *Spindle) loadPipeline(ctx context.Context, repoUri, repoPath, rev string) (workflow.RawPipeline, error) { 524 524 + if err := git.SparseSyncGitRepo(ctx, repoUri, repoPath, rev); err != nil { 525 525 + return nil, fmt.Errorf("syncing git repo: %w", err) 526 526 + } 527 527 + gr, err := kgit.Open(repoPath, rev) 528 528 + if err != nil { 529 529 + return nil, fmt.Errorf("opening git repo: %w", err) 530 530 + } 455 531 456 456 - // inject TANGLED_* env vars after InitWorkflow 457 457 - // This prevents user-defined env vars from overriding them 458 458 - if ewf.Environment == nil { 459 459 - ewf.Environment = make(map[string]string) 460 460 - } 461 461 - maps.Copy(ewf.Environment, pipelineEnv) 532 532 + workflowDir, err := gr.FileTree(ctx, workflow.WorkflowDir) 533 533 + if errors.Is(err, object.ErrDirectoryNotFound) { 534 534 + // return empty RawPipeline when directory doesn't exist 535 535 + return nil, nil 536 536 + } else if err != nil { 537 537 + return nil, fmt.Errorf("loading file tree: %w", err) 538 538 + } 462 539 463 463 - workflows[eng] = append(workflows[eng], *ewf) 540 540 + var rawPipeline workflow.RawPipeline 541 541 + for _, e := range workflowDir { 542 542 + if !e.IsFile() { 543 543 + continue 544 544 + } 464 545 465 465 - err = s.db.StatusPending(models.WorkflowId{ 466 466 - PipelineId: pipelineId, 467 467 - Name: w.Name, 468 468 - }, s.n) 469 469 - if err != nil { 470 470 - return fmt.Errorf("db.StatusPending: %w", err) 471 471 - } 472 472 - } 546 546 + fpath := filepath.Join(workflow.WorkflowDir, e.Name) 547 547 + contents, err := gr.RawContent(fpath) 548 548 + if err != nil { 549 549 + return nil, fmt.Errorf("reading raw content of '%s': %w", fpath, err) 473 550 } 474 551 475 475 - ok := s.jq.Enqueue(queue.Job{ 476 476 - Run: func() error { 477 477 - engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.workflowSem, ctx, &models.Pipeline{ 478 478 - RepoDid: repoDid, 479 479 - Workflows: workflows, 480 480 - }, pipelineId) 481 481 - return nil 482 482 - }, 483 483 - OnFail: func(jobError error) { 484 484 - s.l.Error("pipeline run failed", "error", jobError) 485 485 - }, 552 552 + rawPipeline = append(rawPipeline, workflow.RawWorkflow{ 553 553 + Name: e.Name, 554 554 + Contents: contents, 486 555 }) 487 487 - if ok { 488 488 - s.l.Info("pipeline enqueued successfully", "id", msg.Rkey) 489 489 - } else { 490 490 - s.l.Error("failed to enqueue pipeline: queue is full") 556 556 + } 557 557 + 558 558 + return rawPipeline, nil 559 559 + } 560 560 + 561 561 + func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl tangled.Pipeline, pipelineId models.PipelineId) error { 562 562 + // Build pipeline environment variables once for all workflows 563 563 + pipelineEnv := models.PipelineEnvVars(tpl.TriggerMetadata, pipelineId, s.cfg.Server.Dev) 564 564 + 565 565 + // filter & init workflows 566 566 + workflows := make(map[models.Engine][]models.Workflow) 567 567 + for _, w := range tpl.Workflows { 568 568 + if w == nil { 569 569 + continue 491 570 } 492 492 - } else if msg.Nsid == tangled.GitRefUpdateNSID { 493 493 - event := tangled.GitRefUpdate{} 494 494 - if err := json.Unmarshal(msg.EventJson, &event); err != nil { 495 495 - l.Error("error unmarshalling", "err", err) 496 496 - return err 571 571 + if _, ok := s.engs[w.Engine]; !ok { 572 572 + err := s.db.StatusFailed(models.WorkflowId{ 573 573 + PipelineId: pipelineId, 574 574 + Name: w.Name, 575 575 + }, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n) 576 576 + if err != nil { 577 577 + return fmt.Errorf("db.StatusFailed: %w", err) 578 578 + } 579 579 + 580 580 + continue 497 581 } 498 498 - l = l.With("repo", event.Repo, "ref", event.Ref, "newSha", event.NewSha) 499 499 - l.Debug("debug") 500 582 501 501 - repoDid := syntax.DID(event.Repo) 502 502 - if _, err := s.db.GetRepoByDid(repoDid); err != nil { 503 503 - return fmt.Errorf("unknown repoDid %s: %w", repoDid, err) 583 583 + eng := s.engs[w.Engine] 584 584 + 585 585 + if _, ok := workflows[eng]; !ok { 586 586 + workflows[eng] = []models.Workflow{} 504 587 } 505 588 506 506 - // NOTE: we are blindly trusting the knot that it will return only repos it own 507 507 - repoCloneUri := s.newRepoCloneUrl(src.Key(), syntax.DID(event.Repo)) 508 508 - repoPath := s.newRepoPath(syntax.DID(event.Repo)) 509 509 - if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, event.NewSha); err != nil { 510 510 - return fmt.Errorf("sync git repo: %w", err) 589 589 + ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl) 590 590 + if err != nil { 591 591 + err = s.db.StatusFailed(models.WorkflowId{ 592 592 + PipelineId: pipelineId, 593 593 + Name: w.Name, 594 594 + }, fmt.Sprintf("init workflow: %s", err), -1, s.n) 595 595 + if err != nil { 596 596 + return fmt.Errorf("db.StatusFailed: %w", err) 597 597 + } 598 598 + 599 599 + continue 511 600 } 512 512 - l.Info("synced git repo") 513 601 514 514 - // TODO: plan the pipeline 602 602 + // inject TANGLED_* env vars after InitWorkflow 603 603 + // This prevents user-defined env vars from overriding them 604 604 + if ewf.Environment == nil { 605 605 + ewf.Environment = make(map[string]string) 606 606 + } 607 607 + maps.Copy(ewf.Environment, pipelineEnv) 608 608 + 609 609 + workflows[eng] = append(workflows[eng], *ewf) 515 610 } 516 611 612 612 + // enqueue pipeline 613 613 + ok := s.jq.Enqueue(queue.Job{ 614 614 + Run: func() error { 615 615 + engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.workflowSem, ctx, &models.Pipeline{ 616 616 + RepoDid: repoDid, 617 617 + Workflows: workflows, 618 618 + }, pipelineId) 619 619 + return nil 620 620 + }, 621 621 + OnFail: func(jobError error) { 622 622 + s.l.Error("pipeline run failed", "error", jobError) 623 623 + }, 624 624 + }) 625 625 + if !ok { 626 626 + return fmt.Errorf("failed to enqueue pipeline: queue is full") 627 627 + } 628 628 + s.l.Info("pipeline enqueued successfully", "id", pipelineId) 629 629 + 630 630 + // after successful enqueue, emit StatusPending for all workflows 631 631 + for _, ewfs := range workflows { 632 632 + for _, ewf := range ewfs { 633 633 + err := s.db.StatusPending(models.WorkflowId{ 634 634 + PipelineId: pipelineId, 635 635 + Name: ewf.Name, 636 636 + }, s.n) 637 637 + if err != nil { 638 638 + return fmt.Errorf("db.StatusPending: %w", err) 639 639 + } 640 640 + } 641 641 + } 517 642 return nil 518 643 } 519 644

+185

spindle/tapclient.go

Reviewed

··· 6 6 "encoding/json" 7 7 "errors" 8 8 "fmt" 9 9 + "io" 9 10 "log/slog" 11 11 + "net/http" 12 12 + "net/url" 10 13 "sync" 11 14 "time" 12 15 13 16 "github.com/bluesky-social/indigo/atproto/syntax" 17 17 + indigoxrpc "github.com/bluesky-social/indigo/xrpc" 14 18 "tangled.org/core/api/tangled" 19 19 + avmodels "tangled.org/core/appview/models" 15 20 "tangled.org/core/eventconsumer" 16 21 "tangled.org/core/log" 17 22 "tangled.org/core/rbac" 18 23 "tangled.org/core/spindle/db" 19 24 "tangled.org/core/spindle/git" 25 25 + "tangled.org/core/spindle/models" 20 26 "tangled.org/core/tapc" 27 27 + "tangled.org/core/tid" 28 28 + "tangled.org/core/workflow" 21 29 ) 22 30 23 31 const ( ··· 75 83 return t.processRepo(ctx, evt.Record) 76 84 case tangled.RepoCollaboratorNSID: 77 85 return t.processCollaborator(ctx, evt.Record) 86 86 + case tangled.RepoPullNSID: 87 87 + return t.processPull(ctx, evt.Record) 78 88 } 79 89 return nil 80 90 } ··· 306 316 return nil 307 317 } 308 318 319 319 + func (t *Tap) processPull(ctx context.Context, evt *tapc.RecordEventData) error { 320 320 + l := t.logger.With("collection", evt.Collection, "did", evt.Did, "rkey", evt.Rkey) 321 321 + 322 322 + // only listen to live events 323 323 + if !evt.Live { 324 324 + l.Info("skipping backfill event", "event", evt.AtUri()) 325 325 + return nil 326 326 + } 327 327 + 328 328 + switch evt.Action { 329 329 + case tapc.RecordCreateAction, tapc.RecordUpdateAction: 330 330 + record := tangled.RepoPull{} 331 331 + if err := json.Unmarshal(evt.Record, &record); err != nil { 332 332 + l.Error("invalid record", "err", err) 333 333 + return fmt.Errorf("parsing record: %w", err) 334 334 + } 335 335 + 336 336 + // ignore legacy records 337 337 + if record.Target == nil { 338 338 + l.Info("ignoring pull record: target repo is nil") 339 339 + return nil 340 340 + } 341 341 + 342 342 + // ignore patch-based and fork-based PRs 343 343 + if record.Source == nil || record.Source.Repo != nil { 344 344 + l.Info("ignoring pull record: not a branch-based pull request") 345 345 + return nil 346 346 + } 347 347 + 348 348 + // skip if target repo is unknown 349 349 + repo, err := t.spindle.db.GetRepoByDid(syntax.DID(record.Target.Repo)) 350 350 + if err != nil { 351 351 + l.Warn("target repo is not ingested yet", "repo", record.Target.Repo, "err", err) 352 352 + return fmt.Errorf("target repo is unknown") 353 353 + } 354 354 + 355 355 + // only accept branch-based PR (excluding patch-based and fork-based) 356 356 + if record.Source == nil || record.Source.Repo != nil { 357 357 + l.Warn("skipping non-branch-based PR") 358 358 + return nil 359 359 + } 360 360 + 361 361 + latestSubmission, err := t.fetchLatestSubmission(ctx, evt.Did.String(), evt.Rkey.String(), &record) 362 362 + if err != nil { 363 363 + return err 364 364 + } 365 365 + sourceSha := latestSubmission.SourceRev 366 366 + 367 367 + scheme := "https" 368 368 + if t.spindle.cfg.Server.Dev { 369 369 + scheme = "http" 370 370 + } 371 371 + client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, repo.Knot)} 372 372 + 373 373 + // fetch current default branch 374 374 + defaultBranch, _ := func(repo syntax.DID) (string, error) { 375 375 + defaultBranchOut, err := tangled.RepoGetDefaultBranch(ctx, client, repo.String()) 376 376 + if err != nil { 377 377 + return "", err 378 378 + } 379 379 + return defaultBranchOut.Name, nil 380 380 + }(repo.RepoDid) 381 381 + 382 382 + compiler := workflow.Compiler{ 383 383 + Trigger: tangled.Pipeline_TriggerMetadata{ 384 384 + Kind: string(workflow.TriggerKindPullRequest), 385 385 + PullRequest: &tangled.Pipeline_PullRequestTriggerData{ 386 386 + Action: "create", 387 387 + SourceBranch: record.Source.Branch, 388 388 + SourceSha: sourceSha, 389 389 + TargetBranch: record.Target.Branch, 390 390 + }, 391 391 + Repo: &tangled.Pipeline_TriggerRepo{ 392 392 + Did: repo.Owner.String(), 393 393 + Knot: repo.Knot, 394 394 + Repo: (*string)(&repo.Rkey), 395 395 + RepoDid: (*string)(&repo.RepoDid), 396 396 + DefaultBranch: defaultBranch, 397 397 + }, 398 398 + }, 399 399 + } 400 400 + 401 401 + repoUri := t.spindle.newRepoCloneUrl(repo.Knot, repo.RepoDid) 402 402 + repoPath := t.spindle.newRepoPath(repo.RepoDid) 403 403 + 404 404 + // load workflow definitions from rev (without spindle context) 405 405 + rawPipeline, err := t.spindle.loadPipeline(ctx, repoUri, repoPath, sourceSha) 406 406 + if err != nil { 407 407 + // don't retry 408 408 + l.Error("failed loading pipeline", "err", err) 409 409 + return nil 410 410 + } 411 411 + if len(rawPipeline) == 0 { 412 412 + l.Info("no workflow definition find for the repo. skipping the event") 413 413 + return nil 414 414 + } 415 415 + tpl := compiler.Compile(compiler.Parse(rawPipeline)) 416 416 + // TODO: pass compile error to workflow log 417 417 + for _, w := range compiler.Diagnostics.Errors { 418 418 + l.Error(w.String()) 419 419 + } 420 420 + for _, w := range compiler.Diagnostics.Warnings { 421 421 + l.Warn(w.String()) 422 422 + } 423 423 + if len(tpl.Workflows) == 0 { 424 424 + l.Info("no workflow matching trigger 'pull_request'. skipping the event") 425 425 + return nil 426 426 + } 427 427 + 428 428 + pipelineId := models.PipelineId{ 429 429 + Knot: tpl.TriggerMetadata.Repo.Knot, 430 430 + Rkey: tid.TID(), 431 431 + } 432 432 + if err := t.spindle.db.CreatePipelineEvent(pipelineId.Rkey, tpl, t.spindle.n); err != nil { 433 433 + l.Error("failed to create pipeline event", "err", err) 434 434 + return nil 435 435 + } 436 436 + err = t.spindle.processPipeline(ctx, repo.RepoDid, tpl, pipelineId) 437 437 + if err != nil { 438 438 + // don't retry 439 439 + l.Error("failed processing pipeline", "err", err) 440 440 + return nil 441 441 + } 442 442 + case tapc.RecordDeleteAction: 443 443 + // no-op 444 444 + } 445 445 + return nil 446 446 + } 447 447 + 309 448 func (t *Tap) bufferCollab(repoDid syntax.DID, evt *tapc.RecordEventData) { 310 449 t.pendingMu.Lock() 311 450 defer t.pendingMu.Unlock() ··· 373 512 t.logger.Warn("expired buffered collaborator events without matching repo arrival", "count", expired, "ttl", pendingCollabTTL) 374 513 } 375 514 } 515 515 + 516 516 + func (t *Tap) fetchLatestSubmission(ctx context.Context, did, rkey string, record *tangled.RepoPull) (*avmodels.PullSubmission, error) { 517 517 + // resolve the PR owner's identity to fetch the blob from their PDS 518 518 + prOwnerIdent, err := t.spindle.res.ResolveIdent(ctx, did) 519 519 + if err != nil || prOwnerIdent.Handle.IsInvalidHandle() { 520 520 + return nil, fmt.Errorf("failed to resolve PR owner handle: %w", err) 521 521 + } 522 522 + 523 523 + if len(record.Rounds) == 0 { 524 524 + return nil, fmt.Errorf("failed to fetch latest submission, no rounds in record") 525 525 + } 526 526 + 527 527 + roundNumber := len(record.Rounds) - 1 528 528 + round := record.Rounds[roundNumber] 529 529 + 530 530 + // fetch the blob from the PR owner's PDS 531 531 + prOwnerPds := prOwnerIdent.PDSEndpoint() 532 532 + blobUrl, err := url.Parse(fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob", prOwnerPds)) 533 533 + if err != nil { 534 534 + return nil, fmt.Errorf("failed to construct blob URL: %w", err) 535 535 + } 536 536 + q := blobUrl.Query() 537 537 + q.Set("cid", round.PatchBlob.Ref.String()) 538 538 + q.Set("did", did) 539 539 + blobUrl.RawQuery = q.Encode() 540 540 + 541 541 + req, err := http.NewRequestWithContext(ctx, http.MethodGet, blobUrl.String(), nil) 542 542 + if err != nil { 543 543 + return nil, fmt.Errorf("failed to create blob request: %w", err) 544 544 + } 545 545 + req.Header.Set("Content-Type", "application/json") 546 546 + 547 547 + blobResp, err := http.DefaultClient.Do(req) 548 548 + if err != nil { 549 549 + return nil, fmt.Errorf("failed to fetch blob: %w", err) 550 550 + } 551 551 + defer blobResp.Body.Close() 552 552 + 553 553 + blob := io.ReadCloser(blobResp.Body) 554 554 + latestSubmission, err := avmodels.PullSubmissionFromRecord(did, rkey, roundNumber, round, &blob) 555 555 + if err != nil { 556 556 + return nil, fmt.Errorf("failed to parse submission: %w", err) 557 557 + } 558 558 + 559 559 + return latestSubmission, nil 560 560 + }