+1

knotserver/internal.go

··· 176 176 } 177 177 178 178 for _, line := range lines { 179 179 + // TODO: pass pushOptions to refUpdate 179 180 err := h.insertRefUpdate(line, gitUserDid, repoDid, repoName) 180 181 if err != nil { 181 182 l.Error("failed to insert op", "err", err, "line", line, "did", gitUserDid, "repo", gitRelativeDir)

+14

spindle/db/events.go

··· 70 70 return evts, nil 71 71 } 72 72 73 73 + func (d *DB) CreatePipelineEvent(rkey string, pipeline tangled.Pipeline, n *notifier.Notifier) error { 74 74 + eventJson, err := json.Marshal(pipeline) 75 75 + if err != nil { 76 76 + return err 77 77 + } 78 78 + event := Event{ 79 79 + Rkey: rkey, 80 80 + Nsid: tangled.PipelineNSID, 81 81 + Created: time.Now().UnixNano(), 82 82 + EventJson: string(eventJson), 83 83 + } 84 84 + return d.insertEvent(event, n) 85 85 + } 86 86 + 73 87 func (d *DB) createStatusEvent( 74 88 workflowId models.WorkflowId, 75 89 statusKind models.StatusKind,

+176 -73

spindle/server.go

··· 4 4 "context" 5 5 _ "embed" 6 6 "encoding/json" 7 7 + "errors" 7 8 "fmt" 8 9 "log/slog" 9 10 "maps" ··· 15 16 "github.com/bluesky-social/indigo/atproto/syntax" 16 17 "github.com/bluesky-social/indigo/service/tap" 17 18 "github.com/go-chi/chi/v5" 19 19 + "github.com/go-git/go-git/v5/plumbing/object" 18 20 "github.com/hashicorp/go-version" 19 21 "tangled.org/core/api/tangled" 20 22 "tangled.org/core/eventconsumer" 21 23 "tangled.org/core/eventconsumer/cursor" 22 24 "tangled.org/core/idresolver" 25 25 + kgit "tangled.org/core/knotserver/git" 23 26 "tangled.org/core/log" 24 27 "tangled.org/core/notifier" 25 28 "tangled.org/core/rbac2" ··· 33 36 "tangled.org/core/spindle/secrets" 34 37 "tangled.org/core/spindle/xrpc" 35 38 "tangled.org/core/tapc" 39 39 + "tangled.org/core/tid" 40 40 + "tangled.org/core/workflow" 36 41 "tangled.org/core/xrpc/serviceauth" 37 42 ) 38 43 ··· 132 137 return nil, fmt.Errorf("failed to setup sqlite3 cursor store: %w", err) 133 138 } 134 139 135 135 - // for each incoming sh.tangled.pipeline, we execute 136 136 - // spindle.processPipeline, which in turn enqueues the pipeline 137 137 - // job in the above registered queue. 140 140 + // spindle listen to knot stream for sh.tangled.git.refUpdate 141 141 + // which will sync the local workflow files in spindle and enqueues the 142 142 + // pipeline job for on-push workflows 138 143 ccfg := eventconsumer.NewConsumerConfig() 139 144 ccfg.Logger = log.SubLogger(logger, "eventconsumer") 140 145 ccfg.Dev = cfg.Server.Dev 141 141 - ccfg.ProcessFunc = spindle.processPipeline 146 146 + ccfg.ProcessFunc = spindle.processKnotStream 142 147 ccfg.CursorStore = cursorStore 143 148 knownKnots, err := d.Knots() 144 149 if err != nil { ··· 227 232 RepoFetchTimeout: 300 * time.Second, 228 233 IdentityCacheSize: 2_000_000, 229 234 EventCacheSize: 100_000, 230 230 - CollectionFilters: []string{tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.SpindleMemberNSID}, 235 235 + SignalCollection: tangled.RepoPullNSID, // HACK: listen for repo.pull from non-tangled users 236 236 + CollectionFilters: []string{tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.RepoPullNSID, tangled.SpindleMemberNSID}, 231 237 RetryTimeout: 3 * time.Second, 232 238 }) 233 239 if err != nil { ··· 318 324 return x.Router() 319 325 } 320 326 321 321 - func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, msg eventconsumer.Message) error { 327 327 + func (s *Spindle) processKnotStream(ctx context.Context, src eventconsumer.Source, msg eventconsumer.Message) error { 322 328 l := log.FromContext(ctx).With("handler", "processKnotStream") 323 329 l = l.With("src", src.Key(), "msg.Nsid", msg.Nsid, "msg.Rkey", msg.Rkey) 324 330 if msg.Nsid == tangled.PipelineNSID { ··· 356 362 Rkey: msg.Rkey, 357 363 } 358 364 359 359 - workflows := make(map[models.Engine][]models.Workflow) 360 360 - 361 361 - // Build pipeline environment variables once for all workflows 362 362 - pipelineEnv := models.PipelineEnvVars(tpl.TriggerMetadata, pipelineId, s.cfg.Server.Dev) 363 363 - 364 364 - for _, w := range tpl.Workflows { 365 365 - if w != nil { 366 366 - if _, ok := s.engs[w.Engine]; !ok { 367 367 - err = s.db.StatusFailed(models.WorkflowId{ 368 368 - PipelineId: pipelineId, 369 369 - Name: w.Name, 370 370 - }, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n) 371 371 - if err != nil { 372 372 - return fmt.Errorf("db.StatusFailed: %w", err) 373 373 - } 374 374 - 375 375 - continue 376 376 - } 377 377 - 378 378 - eng := s.engs[w.Engine] 379 379 - 380 380 - if _, ok := workflows[eng]; !ok { 381 381 - workflows[eng] = []models.Workflow{} 382 382 - } 383 383 - 384 384 - ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl) 385 385 - if err != nil { 386 386 - return fmt.Errorf("init workflow: %w", err) 387 387 - } 388 388 - 389 389 - // inject TANGLED_* env vars after InitWorkflow 390 390 - // This prevents user-defined env vars from overriding them 391 391 - if ewf.Environment == nil { 392 392 - ewf.Environment = make(map[string]string) 393 393 - } 394 394 - maps.Copy(ewf.Environment, pipelineEnv) 395 395 - 396 396 - workflows[eng] = append(workflows[eng], *ewf) 397 397 - 398 398 - err = s.db.StatusPending(models.WorkflowId{ 399 399 - PipelineId: pipelineId, 400 400 - Name: w.Name, 401 401 - }, s.n) 402 402 - if err != nil { 403 403 - return fmt.Errorf("db.StatusPending: %w", err) 404 404 - } 405 405 - } 406 406 - } 407 407 - 408 408 - ok := s.jq.Enqueue(queue.Job{ 409 409 - Run: func() error { 410 410 - engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, ctx, &models.Pipeline{ 411 411 - RepoOwner: tpl.TriggerMetadata.Repo.Did, 412 412 - RepoName: tpl.TriggerMetadata.Repo.Repo, 413 413 - Workflows: workflows, 414 414 - }, pipelineId) 415 415 - return nil 416 416 - }, 417 417 - OnFail: func(jobError error) { 418 418 - s.l.Error("pipeline run failed", "error", jobError) 419 419 - }, 420 420 - }) 421 421 - if ok { 422 422 - s.l.Info("pipeline enqueued successfully", "id", msg.Rkey) 423 423 - } else { 424 424 - s.l.Error("failed to enqueue pipeline: queue is full") 365 365 + err = s.processPipeline(ctx, tpl, pipelineId) 366 366 + if err != nil { 367 367 + return err 425 368 } 426 369 } else if msg.Nsid == tangled.GitRefUpdateNSID { 427 370 event := tangled.GitRefUpdate{} ··· 446 389 } 447 390 l.Info("synced git repo") 448 391 449 449 - // TODO: plan the pipeline 392 392 + compiler := workflow.Compiler{ 393 393 + Trigger: tangled.Pipeline_TriggerMetadata{ 394 394 + Kind: string(workflow.TriggerKindPush), 395 395 + Push: &tangled.Pipeline_PushTriggerData{ 396 396 + Ref: event.Ref, 397 397 + OldSha: event.OldSha, 398 398 + NewSha: event.NewSha, 399 399 + }, 400 400 + Repo: &tangled.Pipeline_TriggerRepo{ 401 401 + Did: repo.Did.String(), 402 402 + Knot: repo.Knot, 403 403 + Repo: repo.Name, 404 404 + }, 405 405 + }, 406 406 + } 407 407 + 408 408 + // load workflow definitions from rev (without spindle context) 409 409 + rawPipeline, err := s.loadPipeline(ctx, repoCloneUri, repoPath, event.NewSha) 410 410 + if err != nil { 411 411 + return fmt.Errorf("loading pipeline: %w", err) 412 412 + } 413 413 + if len(rawPipeline) == 0 { 414 414 + l.Info("no workflow definition find for the repo. skipping the event") 415 415 + return nil 416 416 + } 417 417 + tpl := compiler.Compile(compiler.Parse(rawPipeline)) 418 418 + // TODO: pass compile error to workflow log 419 419 + for _, w := range compiler.Diagnostics.Errors { 420 420 + l.Error(w.String()) 421 421 + } 422 422 + for _, w := range compiler.Diagnostics.Warnings { 423 423 + l.Warn(w.String()) 424 424 + } 425 425 + 426 426 + pipelineId := models.PipelineId{ 427 427 + Knot: tpl.TriggerMetadata.Repo.Knot, 428 428 + Rkey: tid.TID(), 429 429 + } 430 430 + if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { 431 431 + l.Error("failed to create pipeline event", "err", err) 432 432 + return nil 433 433 + } 434 434 + err = s.processPipeline(ctx, tpl, pipelineId) 435 435 + if err != nil { 436 436 + return err 437 437 + } 438 438 + } 439 439 + 440 440 + return nil 441 441 + } 442 442 + 443 443 + func (s *Spindle) loadPipeline(ctx context.Context, repoUri, repoPath, rev string) (workflow.RawPipeline, error) { 444 444 + if err := git.SparseSyncGitRepo(ctx, repoUri, repoPath, rev); err != nil { 445 445 + return nil, fmt.Errorf("syncing git repo: %w", err) 446 446 + } 447 447 + gr, err := kgit.Open(repoPath, rev) 448 448 + if err != nil { 449 449 + return nil, fmt.Errorf("opening git repo: %w", err) 450 450 + } 451 451 + 452 452 + workflowDir, err := gr.FileTree(ctx, workflow.WorkflowDir) 453 453 + if errors.Is(err, object.ErrDirectoryNotFound) { 454 454 + // return empty RawPipeline when directory doesn't exist 455 455 + return nil, nil 456 456 + } else if err != nil { 457 457 + return nil, fmt.Errorf("loading file tree: %w", err) 450 458 } 451 459 460 460 + var rawPipeline workflow.RawPipeline 461 461 + for _, e := range workflowDir { 462 462 + if !e.IsFile() { 463 463 + continue 464 464 + } 465 465 + 466 466 + fpath := filepath.Join(workflow.WorkflowDir, e.Name) 467 467 + contents, err := gr.RawContent(fpath) 468 468 + if err != nil { 469 469 + return nil, fmt.Errorf("reading raw content of '%s': %w", fpath, err) 470 470 + } 471 471 + 472 472 + rawPipeline = append(rawPipeline, workflow.RawWorkflow{ 473 473 + Name: e.Name, 474 474 + Contents: contents, 475 475 + }) 476 476 + } 477 477 + 478 478 + return rawPipeline, nil 479 479 + } 480 480 + 481 481 + func (s *Spindle) processPipeline(ctx context.Context, tpl tangled.Pipeline, pipelineId models.PipelineId) error { 482 482 + // Build pipeline environment variables once for all workflows 483 483 + pipelineEnv := models.PipelineEnvVars(tpl.TriggerMetadata, pipelineId, s.cfg.Server.Dev) 484 484 + 485 485 + // filter & init workflows 486 486 + workflows := make(map[models.Engine][]models.Workflow) 487 487 + for _, w := range tpl.Workflows { 488 488 + if w == nil { 489 489 + continue 490 490 + } 491 491 + if _, ok := s.engs[w.Engine]; !ok { 492 492 + err := s.db.StatusFailed(models.WorkflowId{ 493 493 + PipelineId: pipelineId, 494 494 + Name: w.Name, 495 495 + }, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n) 496 496 + if err != nil { 497 497 + return fmt.Errorf("db.StatusFailed: %w", err) 498 498 + } 499 499 + 500 500 + continue 501 501 + } 502 502 + 503 503 + eng := s.engs[w.Engine] 504 504 + 505 505 + if _, ok := workflows[eng]; !ok { 506 506 + workflows[eng] = []models.Workflow{} 507 507 + } 508 508 + 509 509 + ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl) 510 510 + if err != nil { 511 511 + return fmt.Errorf("init workflow: %w", err) 512 512 + } 513 513 + 514 514 + // inject TANGLED_* env vars after InitWorkflow 515 515 + // This prevents user-defined env vars from overriding them 516 516 + if ewf.Environment == nil { 517 517 + ewf.Environment = make(map[string]string) 518 518 + } 519 519 + maps.Copy(ewf.Environment, pipelineEnv) 520 520 + 521 521 + workflows[eng] = append(workflows[eng], *ewf) 522 522 + } 523 523 + 524 524 + // enqueue pipeline 525 525 + ok := s.jq.Enqueue(queue.Job{ 526 526 + Run: func() error { 527 527 + engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, ctx, &models.Pipeline{ 528 528 + RepoOwner: tpl.TriggerMetadata.Repo.Did, 529 529 + RepoName: tpl.TriggerMetadata.Repo.Repo, 530 530 + Workflows: workflows, 531 531 + }, pipelineId) 532 532 + return nil 533 533 + }, 534 534 + OnFail: func(jobError error) { 535 535 + s.l.Error("pipeline run failed", "error", jobError) 536 536 + }, 537 537 + }) 538 538 + if !ok { 539 539 + return fmt.Errorf("failed to enqueue pipeline: queue is full") 540 540 + } 541 541 + s.l.Info("pipeline enqueued successfully", "id", pipelineId) 542 542 + 543 543 + // emit StatusPending for all workflows here (after successful enqueue) 544 544 + for _, ewfs := range workflows { 545 545 + for _, ewf := range ewfs { 546 546 + err := s.db.StatusPending(models.WorkflowId{ 547 547 + PipelineId: pipelineId, 548 548 + Name: ewf.Name, 549 549 + }, s.n) 550 550 + if err != nil { 551 551 + return fmt.Errorf("db.StatusPending: %w", err) 552 552 + } 553 553 + } 554 554 + } 452 555 return nil 453 556 } 454 557

+89 -2

spindle/tap.go

··· 11 11 "tangled.org/core/eventconsumer" 12 12 "tangled.org/core/spindle/db" 13 13 "tangled.org/core/spindle/git" 14 14 + "tangled.org/core/spindle/models" 14 15 "tangled.org/core/tapc" 16 16 + "tangled.org/core/tid" 17 17 + "tangled.org/core/workflow" 15 18 ) 16 19 17 20 func (s *Spindle) processEvent(ctx context.Context, evt tapc.Event) error { ··· 281 284 282 285 l.Info("processing pull record") 283 286 287 287 + // only listen to live events 288 288 + if !evt.Record.Live { 289 289 + l.Info("skipping backfill event", "event", evt.Record.AtUri()) 290 290 + return nil 291 291 + } 292 292 + 284 293 switch evt.Record.Action { 285 294 case tapc.RecordCreateAction, tapc.RecordUpdateAction: 286 286 - // TODO 295 295 + record := tangled.RepoPull{} 296 296 + if err := json.Unmarshal(evt.Record.Record, &record); err != nil { 297 297 + l.Error("invalid record", "err", err) 298 298 + return fmt.Errorf("parsing record: %w", err) 299 299 + } 300 300 + 301 301 + // ignore legacy records 302 302 + if record.Target == nil { 303 303 + l.Info("ignoring pull record: target repo is nil") 304 304 + return nil 305 305 + } 306 306 + 307 307 + // ignore patch-based and fork-based PRs 308 308 + if record.Source == nil || record.Source.Repo != nil { 309 309 + l.Info("ignoring pull record: not a branch-based pull request") 310 310 + return nil 311 311 + } 312 312 + 313 313 + // skip if target repo is unknown 314 314 + repo, err := s.db.GetRepo(syntax.ATURI(record.Target.Repo)) 315 315 + if err != nil { 316 316 + l.Warn("target repo is not ingested yet", "repo", record.Target.Repo, "err", err) 317 317 + return fmt.Errorf("target repo is unknown") 318 318 + } 319 319 + 320 320 + compiler := workflow.Compiler{ 321 321 + Trigger: tangled.Pipeline_TriggerMetadata{ 322 322 + Kind: string(workflow.TriggerKindPullRequest), 323 323 + PullRequest: &tangled.Pipeline_PullRequestTriggerData{ 324 324 + Action: "create", 325 325 + SourceBranch: record.Source.Branch, 326 326 + SourceSha: record.Source.Sha, 327 327 + TargetBranch: record.Target.Branch, 328 328 + }, 329 329 + Repo: &tangled.Pipeline_TriggerRepo{ 330 330 + Did: repo.Did.String(), 331 331 + Knot: repo.Knot, 332 332 + Repo: repo.Name, 333 333 + }, 334 334 + }, 335 335 + } 336 336 + 337 337 + repoUri := s.newRepoCloneUrl(repo.Knot, repo.Did.String(), repo.Name) 338 338 + repoPath := s.newRepoPath(repo.Did, repo.Rkey) 339 339 + 340 340 + // load workflow definitions from rev (without spindle context) 341 341 + rawPipeline, err := s.loadPipeline(ctx, repoUri, repoPath, record.Source.Sha) 342 342 + if err != nil { 343 343 + // don't retry 344 344 + l.Error("failed loading pipeline", "err", err) 345 345 + return nil 346 346 + } 347 347 + if len(rawPipeline) == 0 { 348 348 + l.Info("no workflow definition find for the repo. skipping the event") 349 349 + return nil 350 350 + } 351 351 + tpl := compiler.Compile(compiler.Parse(rawPipeline)) 352 352 + // TODO: pass compile error to workflow log 353 353 + for _, w := range compiler.Diagnostics.Errors { 354 354 + l.Error(w.String()) 355 355 + } 356 356 + for _, w := range compiler.Diagnostics.Warnings { 357 357 + l.Warn(w.String()) 358 358 + } 359 359 + 360 360 + pipelineId := models.PipelineId{ 361 361 + Knot: tpl.TriggerMetadata.Repo.Knot, 362 362 + Rkey: tid.TID(), 363 363 + } 364 364 + if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { 365 365 + l.Error("failed to create pipeline event", "err", err) 366 366 + return nil 367 367 + } 368 368 + err = s.processPipeline(ctx, tpl, pipelineId) 369 369 + if err != nil { 370 370 + // don't retry 371 371 + l.Error("failed processing pipeline", "err", err) 372 372 + return nil 373 373 + } 287 374 case tapc.RecordDeleteAction: 288 288 - // TODO 375 375 + // no-op 289 376 } 290 377 return nil 291 378 }