nix,spindle: sync workflow files on `sh.tangled.git.refUpdate` · tangled.org/core@8bbface

+1

go.mod

··· 47 47 github.com/gorilla/sessions v1.4.0 48 48 github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 49 49 github.com/hashicorp/golang-lru/v2 v2.0.7 50 + github.com/hashicorp/go-version v1.8.0 50 51 github.com/hiddeco/sshsig v0.2.0 51 52 github.com/hpcloud/tail v1.0.0 52 53 github.com/ipfs/go-cid v0.6.0

+2

go.sum

··· 402 402 github.com/hashicorp/go-secure-stdlib/strutil v0.1.2/go.mod h1:Gou2R9+il93BqX25LAKCLuM+y9U2T4hlwvT1yprcna4= 403 403 github.com/hashicorp/go-sockaddr v1.0.7 h1:G+pTkSO01HpR5qCxg7lxfsFEZaG+C0VssTy/9dbT+Fw= 404 404 github.com/hashicorp/go-sockaddr v1.0.7/go.mod h1:FZQbEYa1pxkQ7WLpyXJ6cbjpT8q0YgQaK/JakXqGyWw= 405 + github.com/hashicorp/go-version v1.8.0 h1:KAkNb1HAiZd1ukkxDFGmokVZe1Xy9HG6NUp+bPle2i4= 406 + github.com/hashicorp/go-version v1.8.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= 405 407 github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c= 406 408 github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= 407 409 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=

+3

nix/gomod2nix.toml

··· 464 464 [mod."github.com/hashicorp/go-sockaddr"] 465 465 version = "v1.0.7" 466 466 hash = "sha256-p6eDOrGzN1jMmT/F/f/VJMq0cKNFhUcEuVVwTE6vSrs=" 467 + [mod."github.com/hashicorp/go-version"] 468 + version = "v1.8.0" 469 + hash = "sha256-KXtqERmYrWdpqPCViWcHbe6jnuH7k16bvBIcuJuevj8=" 467 470 [mod."github.com/hashicorp/golang-lru"] 468 471 version = "v1.0.2" 469 472 hash = "sha256-yy+5botc6T5wXgOe2mfNXJP3wr+MkVlUZ2JBkmmrA48="

+8

nix/modules/spindle.nix

··· 32 32 description = "Path to the database file"; 33 33 }; 34 34 35 + repoDir = mkOption { 36 + type = types.path; 37 + default = "/var/lib/spindle/repos"; 38 + description = "Path where synced git repositories live"; 39 + }; 40 + 35 41 hostname = mkOption { 36 42 type = types.str; 37 43 example = "my.spindle.com"; ··· 301 307 302 308 config = let 303 309 deps = [ 310 + pkgs.git 304 311 pkgs.qemu 305 312 pkgs.e2fsprogs 306 313 pkgs.slirp4netns ··· 341 348 Environment = [ 342 349 "SPINDLE_SERVER_LISTEN_ADDR=${cfg.server.listenAddr}" 343 350 "SPINDLE_SERVER_DB_PATH=${cfg.server.dbPath}" 351 + "SPINDLE_SERVER_REPO_DIR=${cfg.server.repoDir}" 344 352 "SPINDLE_SERVER_HOSTNAME=${cfg.server.hostname}" 345 353 "SPINDLE_SERVER_PLC_URL=${cfg.server.plcUrl}" 346 354 "SPINDLE_SERVER_JETSTREAM_ENDPOINT=${cfg.server.jetstreamEndpoint}"

+1

spindle/config/config.go

··· 12 12 type Server struct { 13 13 ListenAddr string `env:"LISTEN_ADDR, default=0.0.0.0:6555"` 14 14 DBPath string `env:"DB_PATH, default=spindle.db"` 15 + RepoDir string `env:"REPO_DIR, default=repos"` 15 16 Hostname string `env:"HOSTNAME, required"` 16 17 JetstreamEndpoint string `env:"JETSTREAM_ENDPOINT, default=wss://jetstream1.us-west.bsky.network/subscribe"` 17 18 Tap Tap `env:",prefix=TAP_"`

+73

spindle/git/git.go

··· 1 + package git 2 + 3 + import ( 4 + "bytes" 5 + "context" 6 + "fmt" 7 + "os" 8 + "os/exec" 9 + "strings" 10 + 11 + "github.com/hashicorp/go-version" 12 + ) 13 + 14 + func Version() (*version.Version, error) { 15 + var buf bytes.Buffer 16 + cmd := exec.Command("git", "version") 17 + cmd.Stdout = &buf 18 + cmd.Stderr = os.Stderr 19 + err := cmd.Run() 20 + if err != nil { 21 + return nil, err 22 + } 23 + fields := strings.Fields(buf.String()) 24 + if len(fields) < 3 { 25 + return nil, fmt.Errorf("invalid git version: %s", buf.String()) 26 + } 27 + 28 + // version string is like: "git version 2.29.3" or "git version 2.29.3.windows.1" 29 + versionString := fields[2] 30 + if pos := strings.Index(versionString, "windows"); pos >= 1 { 31 + versionString = versionString[:pos-1] 32 + } 33 + return version.NewVersion(versionString) 34 + } 35 + 36 + const WorkflowDir = `/.tangled/workflows` 37 + 38 + func SparseSyncGitRepo(ctx context.Context, cloneUri, path, rev string) error { 39 + exist, err := isDir(path) 40 + if err != nil { 41 + return err 42 + } 43 + if rev == "" { 44 + rev = "HEAD" 45 + } 46 + if !exist { 47 + if err := exec.Command("git", "clone", "--no-checkout", "--depth=1", "--filter=tree:0", "--revision="+rev, cloneUri, path).Run(); err != nil { 48 + return fmt.Errorf("git clone: %w", err) 49 + } 50 + if err := exec.Command("git", "-C", path, "sparse-checkout", "set", "--no-cone", WorkflowDir).Run(); err != nil { 51 + return fmt.Errorf("git sparse-checkout set: %w", err) 52 + } 53 + } else { 54 + if err := exec.Command("git", "-C", path, "fetch", "--depth=1", "--filter=tree:0", "origin", rev).Run(); err != nil { 55 + return fmt.Errorf("git pull: %w", err) 56 + } 57 + } 58 + if err := exec.Command("git", "-C", path, "checkout", rev).Run(); err != nil { 59 + return fmt.Errorf("git checkout: %w", err) 60 + } 61 + return nil 62 + } 63 + 64 + func isDir(path string) (bool, error) { 65 + info, err := os.Stat(path) 66 + if err == nil && info.IsDir() { 67 + return true, nil 68 + } 69 + if os.IsNotExist(err) { 70 + return false, nil 71 + } 72 + return false, err 73 + }

+60

spindle/server.go

··· 8 8 "log/slog" 9 9 "maps" 10 10 "net/http" 11 + "path/filepath" 11 12 "sync" 12 13 "time" 13 14 14 15 "github.com/bluesky-social/indigo/atproto/syntax" 15 16 "github.com/go-chi/chi/v5" 17 + "github.com/hashicorp/go-version" 16 18 "tangled.org/core/api/tangled" 17 19 "tangled.org/core/eventconsumer" 18 20 "tangled.org/core/eventconsumer/cursor" ··· 28 30 "tangled.org/core/spindle/engines/dummy" 29 31 "tangled.org/core/spindle/engines/microvm" 30 32 "tangled.org/core/spindle/engines/nixery" 33 + "tangled.org/core/spindle/git" 31 34 "tangled.org/core/spindle/models" 32 35 "tangled.org/core/spindle/queue" 33 36 "tangled.org/core/spindle/secrets" ··· 328 331 return fmt.Errorf("failed to load config: %w", err) 329 332 } 330 333 334 + if err := ensureGitVersion(); err != nil { 335 + return fmt.Errorf("ensuring git version: %w", err) 336 + } 337 + 331 338 d, err := db.Make(ctx, cfg.Server.DBPath) 332 339 if err != nil { 333 340 return fmt.Errorf("failed to setup db: %w", err) ··· 389 396 } 390 397 391 398 func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, msg eventstream.Event) error { 399 + l := log.FromContext(ctx).With("handler", "processKnotStream") 400 + l = l.With("src", src.Key(), "msg.Nsid", msg.Nsid, "msg.Rkey", msg.Rkey) 392 401 if msg.Nsid == tangled.PipelineNSID { 402 + return nil 393 403 tpl := tangled.Pipeline{} 394 404 err := json.Unmarshal(msg.EventJson, &tpl) 395 405 if err != nil { ··· 493 503 } else { 494 504 s.l.Error("failed to enqueue pipeline: queue is full") 495 505 } 506 + } else if msg.Nsid == tangled.GitRefUpdateNSID { 507 + event := tangled.GitRefUpdate{} 508 + if err := json.Unmarshal(msg.EventJson, &event); err != nil { 509 + l.Error("error unmarshalling", "err", err) 510 + return err 511 + } 512 + l = l.With("repo", event.Repo, "ref", event.Ref, "newSha", event.NewSha) 513 + l.Debug("debug") 514 + 515 + repoDid := syntax.DID(event.Repo) 516 + if _, err := s.db.GetRepoByDid(repoDid); err != nil { 517 + return fmt.Errorf("unknown repoDid %s: %w", repoDid, err) 518 + } 519 + 520 + // NOTE: we are blindly trusting the knot that it will return only repos it own 521 + repoCloneUri := s.newRepoCloneUrl(src.Key(), syntax.DID(event.Repo)) 522 + repoPath := s.newRepoPath(syntax.DID(event.Repo)) 523 + if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, event.NewSha); err != nil { 524 + return fmt.Errorf("sync git repo: %w", err) 525 + } 526 + l.Info("synced git repo") 527 + 528 + // TODO: plan the pipeline 496 529 } 497 530 531 + return nil 532 + } 533 + 534 + // newRepoPath creates a path to store repository by its did and rkey. 535 + // The path format would be: `/data/repos/did:plc:foo/sh.tangled.repo/repo-rkey 536 + func (s *Spindle) newRepoPath(repo syntax.DID) string { 537 + return filepath.Join(s.cfg.Server.RepoDir, repo.String()) 538 + } 539 + 540 + func (s *Spindle) newRepoCloneUrl(knot string, did syntax.DID) string { 541 + scheme := "https://" 542 + if s.cfg.Server.Dev { 543 + scheme = "http://" 544 + } 545 + return fmt.Sprintf("%s%s/%s", scheme, knot, did) 546 + } 547 + 548 + const RequiredVersion = "2.49.0" 549 + 550 + func ensureGitVersion() error { 551 + v, err := git.Version() 552 + if err != nil { 553 + return fmt.Errorf("fetching git version: %w", err) 554 + } 555 + if v.LessThan(version.Must(version.NewVersion(RequiredVersion))) { 556 + return fmt.Errorf("installed git version %q is not supported, Spindle requires git version >= %q", v, RequiredVersion) 557 + } 498 558 return nil 499 559 } 500 560

+13 -2

spindle/tapclient.go

··· 16 16 "tangled.org/core/log" 17 17 "tangled.org/core/rbac" 18 18 "tangled.org/core/spindle/db" 19 + "tangled.org/core/spindle/git" 19 20 "tangled.org/core/tapc" 20 21 ) 21 22 ··· 122 123 src := eventconsumer.NewKnotSource(record.Knot) 123 124 t.spindle.ks.AddSource(t.spindle.rootCtx, src) 124 125 125 - if err := t.spindle.db.AddRepo(db.Repo{ 126 + repo := db.Repo{ 126 127 Knot: record.Knot, 127 128 Owner: ownerDid, 128 129 Rkey: rkey, 129 130 RepoDid: repoDid, 130 131 CreatedAt: record.CreatedAt, 131 - }); err != nil { 132 + } 133 + 134 + if err := t.spindle.db.AddRepo(repo); err != nil { 132 135 l.Error("failed to add repo row", "err", err) 133 136 return fmt.Errorf("add repo: %w", err) 137 + } 138 + 139 + // setup sparse sync 140 + repoCloneUri := t.spindle.newRepoCloneUrl(repo.Knot, repo.RepoDid) 141 + repoPath := t.spindle.newRepoPath(repo.RepoDid) 142 + if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, ""); err != nil { 143 + return fmt.Errorf("setting up sparse-clone git repo: %w", err) 134 144 } 135 145 136 146 legacyName := "" ··· 192 202 l.Error("failed to delete repo row", "err", err) 193 203 return fmt.Errorf("delete repo row: %w", err) 194 204 } 205 + // TODO: clear sparse-synced git repo 195 206 return nil 196 207 } 197 208

Configure Feed

Configure Feed