nix,spindle: sync workflow files on `sh.tangled.git.refUpdate` #984

+165 -7

8 changed files

Diff round #3

expand all

unified split

go.mod

go.sum

nix

gomod2nix.toml

modules

spindle.nix

spindle

config

config.go

git

git.go

server.go

tap.go

go.mod

··· 29 29 github.com/gorilla/feeds v1.2.0 30 30 github.com/gorilla/sessions v1.4.0 31 31 github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 32 + github.com/hashicorp/go-version v1.8.0 32 33 github.com/hiddeco/sshsig v0.2.0 33 34 github.com/hpcloud/tail v1.0.0 34 35 github.com/ipfs/go-cid v0.5.0

go.sum

··· 265 265 github.com/hashicorp/go-secure-stdlib/strutil v0.1.2/go.mod h1:Gou2R9+il93BqX25LAKCLuM+y9U2T4hlwvT1yprcna4= 266 266 github.com/hashicorp/go-sockaddr v1.0.7 h1:G+pTkSO01HpR5qCxg7lxfsFEZaG+C0VssTy/9dbT+Fw= 267 267 github.com/hashicorp/go-sockaddr v1.0.7/go.mod h1:FZQbEYa1pxkQ7WLpyXJ6cbjpT8q0YgQaK/JakXqGyWw= 268 + github.com/hashicorp/go-version v1.8.0 h1:KAkNb1HAiZd1ukkxDFGmokVZe1Xy9HG6NUp+bPle2i4= 269 + github.com/hashicorp/go-version v1.8.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= 268 270 github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c= 269 271 github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= 270 272 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=

nix/gomod2nix.toml

··· 304 304 [mod."github.com/hashicorp/go-sockaddr"] 305 305 version = "v1.0.7" 306 306 hash = "sha256-p6eDOrGzN1jMmT/F/f/VJMq0cKNFhUcEuVVwTE6vSrs=" 307 + [mod."github.com/hashicorp/go-version"] 308 + version = "v1.8.0" 309 + hash = "sha256-KXtqERmYrWdpqPCViWcHbe6jnuH7k16bvBIcuJuevj8=" 307 310 [mod."github.com/hashicorp/golang-lru"] 308 311 version = "v1.0.2" 309 312 hash = "sha256-yy+5botc6T5wXgOe2mfNXJP3wr+MkVlUZ2JBkmmrA48="

nix/modules/spindle.nix

··· 1 1 { 2 2 config, 3 + pkgs, 3 4 lib, 4 5 ... 5 6 }: let ··· 145 146 description = "spindle service"; 146 147 after = ["network.target" "docker.service" "spindle-tap.service"]; 147 148 wantedBy = ["multi-user.target"]; 149 + path = [ 150 + pkgs.git 151 + ]; 148 152 serviceConfig = { 149 153 LogsDirectory = "spindle"; 150 154 StateDirectory = "spindle";

spindle/config/config.go

··· 27 27 return syntax.DID(fmt.Sprintf("did:web:%s", s.Hostname)) 28 28 } 29 29 30 + func (s Server) RepoDir() string { 31 + return filepath.Join(s.DataDir, "repos") 32 + } 33 + 30 34 func (s Server) DBPath() string { 31 35 return filepath.Join(s.DataDir, "spindle.db") 32 36 }

+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 + }

+66 -5

spindle/server.go

··· 8 8 "log/slog" 9 9 "maps" 10 10 "net/http" 11 + "path/filepath" 11 12 "sync" 12 13 13 14 "github.com/bluesky-social/indigo/atproto/syntax" 14 15 "github.com/go-chi/chi/v5" 16 + "github.com/hashicorp/go-version" 15 17 "tangled.org/core/api/tangled" 16 18 "tangled.org/core/eventconsumer" 17 19 "tangled.org/core/eventconsumer/cursor" ··· 23 25 "tangled.org/core/spindle/db" 24 26 "tangled.org/core/spindle/engine" 25 27 "tangled.org/core/spindle/engines/nixery" 28 + "tangled.org/core/spindle/git" 26 29 "tangled.org/core/spindle/models" 27 30 "tangled.org/core/spindle/queue" 28 31 "tangled.org/core/spindle/secrets" ··· 54 57 func New(ctx context.Context, cfg *config.Config, engines map[string]models.Engine) (*Spindle, error) { 55 58 logger := log.FromContext(ctx) 56 59 57 - d, err := db.Make(ctx, cfg.Server.DBPath) 60 + if err := ensureGitVersion(); err != nil { 61 + return nil, fmt.Errorf("ensuring git version: %w", err) 62 + } 63 + 64 + d, err := db.Make(ctx, cfg.Server.DBPath()) 58 65 if err != nil { 59 66 return nil, fmt.Errorf("failed to setup db: %w", err) 60 67 } 61 68 62 - e, err := rbac2.NewEnforcer(cfg.Server.DBPath) 69 + e, err := rbac2.NewEnforcer(cfg.Server.DBPath()) 63 70 if err != nil { 64 71 return nil, fmt.Errorf("failed to setup rbac enforcer: %w", err) 65 72 } ··· 82 89 } 83 90 logger.Info("using openbao secrets provider", "proxy_address", cfg.Server.Secrets.OpenBao.ProxyAddr, "mount", cfg.Server.Secrets.OpenBao.Mount) 84 91 case "sqlite", "": 85 - vault, err = secrets.NewSQLiteManager(cfg.Server.DBPath, secrets.WithTableName("secrets")) 92 + vault, err = secrets.NewSQLiteManager(cfg.Server.DBPath(), secrets.WithTableName("secrets")) 86 93 if err != nil { 87 94 return nil, fmt.Errorf("failed to setup sqlite secrets provider: %w", err) 88 95 } 89 - logger.Info("using sqlite secrets provider", "path", cfg.Server.DBPath) 96 + logger.Info("using sqlite secrets provider", "path", cfg.Server.DBPath()) 90 97 default: 91 98 return nil, fmt.Errorf("unknown secrets provider: %s", cfg.Server.Secrets.Provider) 92 99 } ··· 118 125 } 119 126 logger.Info("owner set", "did", cfg.Server.Owner) 120 127 121 - cursorStore, err := cursor.NewSQLiteStore(cfg.Server.DBPath) 128 + cursorStore, err := cursor.NewSQLiteStore(cfg.Server.DBPath()) 122 129 if err != nil { 123 130 return nil, fmt.Errorf("failed to setup sqlite3 cursor store: %w", err) 124 131 } ··· 275 282 } 276 283 277 284 func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, msg eventconsumer.Message) error { 285 + l := log.FromContext(ctx).With("handler", "processKnotStream") 286 + l = l.With("src", src.Key(), "msg.Nsid", msg.Nsid, "msg.Rkey", msg.Rkey) 278 287 if msg.Nsid == tangled.PipelineNSID { 288 + return nil 279 289 tpl := tangled.Pipeline{} 280 290 err := json.Unmarshal(msg.EventJson, &tpl) 281 291 if err != nil { ··· 376 386 } else { 377 387 s.l.Error("failed to enqueue pipeline: queue is full") 378 388 } 389 + } else if msg.Nsid == tangled.GitRefUpdateNSID { 390 + event := tangled.GitRefUpdate{} 391 + if err := json.Unmarshal(msg.EventJson, &event); err != nil { 392 + l.Error("error unmarshalling", "err", err) 393 + return err 394 + } 395 + l = l.With("repoDid", event.RepoDid, "repoName", event.RepoName) 396 + 397 + // resolve repo name to rkey 398 + // TODO: git.refUpdate should respond with rkey instead of repo name 399 + repo, err := s.db.GetRepoWithName(syntax.DID(event.RepoDid), event.RepoName) 400 + if err != nil { 401 + return fmt.Errorf("get repo with did and name (%s/%s): %w", event.RepoDid, event.RepoName, err) 402 + } 403 + 404 + // NOTE: we are blindly trusting the knot that it will return only repos it own 405 + repoCloneUri := s.newRepoCloneUrl(src.Key(), event.RepoDid, event.RepoName) 406 + repoPath := s.newRepoPath(repo.Did, repo.Rkey) 407 + if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, event.NewSha); err != nil { 408 + return fmt.Errorf("sync git repo: %w", err) 409 + } 410 + l.Info("synced git repo") 411 + 412 + // TODO: plan the pipeline 413 + } 414 + 415 + return nil 416 + } 417 + 418 + // newRepoPath creates a path to store repository by its did and rkey. 419 + // The path format would be: `/data/repos/did:plc:foo/sh.tangled.repo/repo-rkey 420 + func (s *Spindle) newRepoPath(did syntax.DID, rkey syntax.RecordKey) string { 421 + return filepath.Join(s.cfg.Server.RepoDir(), did.String(), tangled.RepoNSID, rkey.String()) 422 + } 423 + 424 + func (s *Spindle) newRepoCloneUrl(knot, did, name string) string { 425 + scheme := "https://" 426 + if s.cfg.Server.Dev { 427 + scheme = "http://" 379 428 } 429 + return fmt.Sprintf("%s%s/%s/%s", scheme, knot, did, name) 430 + } 431 + 432 + const RequiredVersion = "2.49.0" 380 433 434 + func ensureGitVersion() error { 435 + v, err := git.Version() 436 + if err != nil { 437 + return fmt.Errorf("fetching git version: %w", err) 438 + } 439 + if v.LessThan(version.Must(version.NewVersion(RequiredVersion))) { 440 + return fmt.Errorf("installed git version %q is not supported, Spindle requires git version >= %q", v, RequiredVersion) 441 + } 381 442 return nil 382 443 }

+12 -2

spindle/tap.go

··· 10 10 "tangled.org/core/api/tangled" 11 11 "tangled.org/core/eventconsumer" 12 12 "tangled.org/core/spindle/db" 13 + "tangled.org/core/spindle/git" 13 14 "tangled.org/core/tap" 14 15 ) 15 16 ··· 225 226 return nil 226 227 } 227 228 228 - if err := s.db.PutRepo(&db.Repo{ 229 + repo := &db.Repo{ 229 230 Did: evt.Record.Did, 230 231 Rkey: evt.Record.Rkey, 231 232 Name: record.Name, 232 233 Knot: record.Knot, 233 - }); err != nil { 234 + } 235 + 236 + if err := s.db.PutRepo(repo); err != nil { 234 237 return fmt.Errorf("adding repo to db: %w", err) 235 238 } 236 239 ··· 242 245 src := eventconsumer.NewKnotSource(record.Knot) 243 246 s.ks.AddSource(context.Background(), src) 244 247 248 + // setup sparse sync 249 + repoCloneUri := s.newRepoCloneUrl(repo.Knot, repo.Did.String(), repo.Name) 250 + repoPath := s.newRepoPath(repo.Did, repo.Rkey) 251 + if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, ""); err != nil { 252 + return fmt.Errorf("setting up sparse-clone git repo: %w", err) 253 + } 254 + 245 255 l.Info("added repo", "repo", evt.Record.AtUri()) 246 256 return nil 247 257