From a6a3372015864fcccc20339328cf1789f02ebb53 Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Tue, 16 Dec 2025 15:03:55 +0900 Subject: [PATCH] nix,spindle: sync workflow files on `sh.tangled.git.refUpdate` Spindle will sync git repo when new repo is registered Spindle will listen to `sh.tangled.git.refUpdate` event from knot stream and sync its local git repo instead. Spindle's git repo will sparse-checkout only `/.tangled/workflows` directory. Spindle now requires git version >=2.49 for `--revision` flag in `git clone` command. References: - - Signed-off-by: Seongmin Lee --- go.mod | 1 + go.sum | 2 ++ nix/gomod2nix.toml | 3 ++ nix/modules/spindle.nix | 11 ++++++ spindle/config/config.go | 1 + spindle/git/git.go | 73 ++++++++++++++++++++++++++++++++++++++++ spindle/server.go | 60 +++++++++++++++++++++++++++++++++ spindle/tapclient.go | 15 +++++++-- 8 files changed, 164 insertions(+), 2 deletions(-) create mode 100644 spindle/git/git.go diff --git a/go.mod b/go.mod index 2d45e87d..b6e0aa71 100644 --- a/go.mod +++ b/go.mod @@ -43,6 +43,7 @@ require ( github.com/gorilla/sessions v1.4.0 github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 github.com/hashicorp/golang-lru/v2 v2.0.7 + github.com/hashicorp/go-version v1.8.0 github.com/hiddeco/sshsig v0.2.0 github.com/hpcloud/tail v1.0.0 github.com/ipfs/go-cid v0.6.0 diff --git a/go.sum b/go.sum index 5aa7a19d..12aa21d3 100644 --- a/go.sum +++ b/go.sum @@ -375,6 +375,8 @@ github.com/hashicorp/go-secure-stdlib/strutil v0.1.2 h1:kes8mmyCpxJsI7FTwtzRqEy9 github.com/hashicorp/go-secure-stdlib/strutil v0.1.2/go.mod h1:Gou2R9+il93BqX25LAKCLuM+y9U2T4hlwvT1yprcna4= github.com/hashicorp/go-sockaddr v1.0.7 h1:G+pTkSO01HpR5qCxg7lxfsFEZaG+C0VssTy/9dbT+Fw= github.com/hashicorp/go-sockaddr v1.0.7/go.mod h1:FZQbEYa1pxkQ7WLpyXJ6cbjpT8q0YgQaK/JakXqGyWw= +github.com/hashicorp/go-version v1.8.0 h1:KAkNb1HAiZd1ukkxDFGmokVZe1Xy9HG6NUp+bPle2i4= +github.com/hashicorp/go-version v1.8.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c= github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= diff --git a/nix/gomod2nix.toml b/nix/gomod2nix.toml index 316303a4..7757b7af 100644 --- a/nix/gomod2nix.toml +++ b/nix/gomod2nix.toml @@ -431,6 +431,9 @@ schema = 3 [mod."github.com/hashicorp/go-sockaddr"] version = "v1.0.7" hash = "sha256-p6eDOrGzN1jMmT/F/f/VJMq0cKNFhUcEuVVwTE6vSrs=" + [mod."github.com/hashicorp/go-version"] + version = "v1.8.0" + hash = "sha256-KXtqERmYrWdpqPCViWcHbe6jnuH7k16bvBIcuJuevj8=" [mod."github.com/hashicorp/golang-lru"] version = "v1.0.2" hash = "sha256-yy+5botc6T5wXgOe2mfNXJP3wr+MkVlUZ2JBkmmrA48=" diff --git a/nix/modules/spindle.nix b/nix/modules/spindle.nix index 502a681b..b0097f54 100644 --- a/nix/modules/spindle.nix +++ b/nix/modules/spindle.nix @@ -1,5 +1,6 @@ { config, + pkgs, lib, ... }: let @@ -31,6 +32,12 @@ in description = "Path to the database file"; }; + repoDir = mkOption { + type = types.path; + default = "/var/lib/spindle/repos"; + description = "Path where synced git repositories live"; + }; + hostname = mkOption { type = types.str; example = "my.spindle.com"; @@ -184,6 +191,9 @@ in description = "spindle service"; after = ["network.target" "docker.service"]; wantedBy = ["multi-user.target"]; + path = [ + pkgs.git + ]; serviceConfig = { LogsDirectory = "spindle"; StateDirectory = "spindle"; @@ -192,6 +202,7 @@ in Environment = [ "SPINDLE_SERVER_LISTEN_ADDR=${cfg.server.listenAddr}" "SPINDLE_SERVER_DB_PATH=${cfg.server.dbPath}" + "SPINDLE_SERVER_REPO_DIR=${cfg.server.repoDir}" "SPINDLE_SERVER_HOSTNAME=${cfg.server.hostname}" "SPINDLE_SERVER_PLC_URL=${cfg.server.plcUrl}" "SPINDLE_SERVER_JETSTREAM_ENDPOINT=${cfg.server.jetstreamEndpoint}" diff --git a/spindle/config/config.go b/spindle/config/config.go index 3d264cc3..913caad1 100644 --- a/spindle/config/config.go +++ b/spindle/config/config.go @@ -11,6 +11,7 @@ import ( type Server struct { ListenAddr string `env:"LISTEN_ADDR, default=0.0.0.0:6555"` DBPath string `env:"DB_PATH, default=spindle.db"` + RepoDir string `env:"REPO_DIR, default=repos"` Hostname string `env:"HOSTNAME, required"` JetstreamEndpoint string `env:"JETSTREAM_ENDPOINT, default=wss://jetstream1.us-west.bsky.network/subscribe"` Tap Tap `env:",prefix=TAP_"` diff --git a/spindle/git/git.go b/spindle/git/git.go new file mode 100644 index 00000000..f3e549c1 --- /dev/null +++ b/spindle/git/git.go @@ -0,0 +1,73 @@ +package git + +import ( + "bytes" + "context" + "fmt" + "os" + "os/exec" + "strings" + + "github.com/hashicorp/go-version" +) + +func Version() (*version.Version, error) { + var buf bytes.Buffer + cmd := exec.Command("git", "version") + cmd.Stdout = &buf + cmd.Stderr = os.Stderr + err := cmd.Run() + if err != nil { + return nil, err + } + fields := strings.Fields(buf.String()) + if len(fields) < 3 { + return nil, fmt.Errorf("invalid git version: %s", buf.String()) + } + + // version string is like: "git version 2.29.3" or "git version 2.29.3.windows.1" + versionString := fields[2] + if pos := strings.Index(versionString, "windows"); pos >= 1 { + versionString = versionString[:pos-1] + } + return version.NewVersion(versionString) +} + +const WorkflowDir = `/.tangled/workflows` + +func SparseSyncGitRepo(ctx context.Context, cloneUri, path, rev string) error { + exist, err := isDir(path) + if err != nil { + return err + } + if rev == "" { + rev = "HEAD" + } + if !exist { + if err := exec.Command("git", "clone", "--no-checkout", "--depth=1", "--filter=tree:0", "--revision="+rev, cloneUri, path).Run(); err != nil { + return fmt.Errorf("git clone: %w", err) + } + if err := exec.Command("git", "-C", path, "sparse-checkout", "set", "--no-cone", WorkflowDir).Run(); err != nil { + return fmt.Errorf("git sparse-checkout set: %w", err) + } + } else { + if err := exec.Command("git", "-C", path, "fetch", "--depth=1", "--filter=tree:0", "origin", rev).Run(); err != nil { + return fmt.Errorf("git pull: %w", err) + } + } + if err := exec.Command("git", "-C", path, "checkout", rev).Run(); err != nil { + return fmt.Errorf("git checkout: %w", err) + } + return nil +} + +func isDir(path string) (bool, error) { + info, err := os.Stat(path) + if err == nil && info.IsDir() { + return true, nil + } + if os.IsNotExist(err) { + return false, nil + } + return false, err +} diff --git a/spindle/server.go b/spindle/server.go index a661754d..0952f502 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -8,10 +8,12 @@ import ( "log/slog" "maps" "net/http" + "path/filepath" "sync" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/go-chi/chi/v5" + "github.com/hashicorp/go-version" "tangled.org/core/api/tangled" "tangled.org/core/eventconsumer" "tangled.org/core/eventconsumer/cursor" @@ -26,6 +28,7 @@ import ( "tangled.org/core/spindle/engine" "tangled.org/core/spindle/engines/dummy" "tangled.org/core/spindle/engines/nixery" + "tangled.org/core/spindle/git" "tangled.org/core/spindle/models" "tangled.org/core/spindle/queue" "tangled.org/core/spindle/secrets" @@ -64,6 +67,10 @@ type Spindle struct { func New(ctx context.Context, cfg *config.Config, engines map[string]models.Engine) (*Spindle, error) { logger := log.FromContext(ctx) + if err := ensureGitVersion(); err != nil { + return nil, fmt.Errorf("ensuring git version: %w", err) + } + d, err := db.Make(ctx, cfg.Server.DBPath) if err != nil { return nil, fmt.Errorf("failed to setup db: %w", err) @@ -379,7 +386,10 @@ func (s *Spindle) XrpcRouter() http.Handler { } func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, msg eventstream.Event) error { + l := log.FromContext(ctx).With("handler", "processKnotStream") + l = l.With("src", src.Key(), "msg.Nsid", msg.Nsid, "msg.Rkey", msg.Rkey) if msg.Nsid == tangled.PipelineNSID { + return nil tpl := tangled.Pipeline{} err := json.Unmarshal(msg.EventJson, &tpl) if err != nil { @@ -483,11 +493,61 @@ func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, } else { s.l.Error("failed to enqueue pipeline: queue is full") } + } else if msg.Nsid == tangled.GitRefUpdateNSID { + event := tangled.GitRefUpdate{} + if err := json.Unmarshal(msg.EventJson, &event); err != nil { + l.Error("error unmarshalling", "err", err) + return err + } + l = l.With("repo", event.Repo, "ref", event.Ref, "newSha", event.NewSha) + l.Debug("debug") + + repoDid := syntax.DID(event.Repo) + if _, err := s.db.GetRepoByDid(repoDid); err != nil { + return fmt.Errorf("unknown repoDid %s: %w", repoDid, err) + } + + // NOTE: we are blindly trusting the knot that it will return only repos it own + repoCloneUri := s.newRepoCloneUrl(src.Key(), syntax.DID(event.Repo)) + repoPath := s.newRepoPath(syntax.DID(event.Repo)) + if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, event.NewSha); err != nil { + return fmt.Errorf("sync git repo: %w", err) + } + l.Info("synced git repo") + + // TODO: plan the pipeline } return nil } +// newRepoPath creates a path to store repository by its did and rkey. +// The path format would be: `/data/repos/did:plc:foo/sh.tangled.repo/repo-rkey +func (s *Spindle) newRepoPath(repo syntax.DID) string { + return filepath.Join(s.cfg.Server.RepoDir, repo.String()) +} + +func (s *Spindle) newRepoCloneUrl(knot string, did syntax.DID) string { + scheme := "https://" + if s.cfg.Server.Dev { + scheme = "http://" + } + return fmt.Sprintf("%s%s/%s", scheme, knot, did) +} + +const RequiredVersion = "2.49.0" + +func ensureGitVersion() error { + v, err := git.Version() + if err != nil { + return fmt.Errorf("fetching git version: %w", err) + } + if v.LessThan(version.Must(version.NewVersion(RequiredVersion))) { + return fmt.Errorf("installed git version %q is not supported, Spindle requires git version >= %q", v, RequiredVersion) + } + return nil +} + func (s *Spindle) resolvePipelineRepoDid(repo *tangled.Pipeline_TriggerRepo) (syntax.DID, error) { if repo.RepoDid == nil || *repo.RepoDid == "" { return "", fmt.Errorf("pipeline trigger missing repoDid") diff --git a/spindle/tapclient.go b/spindle/tapclient.go index 7299fbde..19434db6 100644 --- a/spindle/tapclient.go +++ b/spindle/tapclient.go @@ -16,6 +16,7 @@ import ( "tangled.org/core/log" "tangled.org/core/rbac" "tangled.org/core/spindle/db" + "tangled.org/core/spindle/git" "tangled.org/core/tapc" ) @@ -122,17 +123,26 @@ func (t *Tap) processRepo(ctx context.Context, evt *tapc.RecordEventData) error src := eventconsumer.NewKnotSource(record.Knot) t.spindle.ks.AddSource(t.spindle.rootCtx, src) - if err := t.spindle.db.AddRepo(db.Repo{ + repo := db.Repo{ Knot: record.Knot, Owner: ownerDid, Rkey: rkey, RepoDid: repoDid, CreatedAt: record.CreatedAt, - }); err != nil { + } + + if err := t.spindle.db.AddRepo(repo); err != nil { l.Error("failed to add repo row", "err", err) return fmt.Errorf("add repo: %w", err) } + // setup sparse sync + repoCloneUri := t.spindle.newRepoCloneUrl(repo.Knot, repo.RepoDid) + repoPath := t.spindle.newRepoPath(repo.RepoDid) + if err := git.SparseSyncGitRepo(ctx, repoCloneUri, repoPath, ""); err != nil { + return fmt.Errorf("setting up sparse-clone git repo: %w", err) + } + legacyName := "" if record.Name != nil { legacyName = *record.Name @@ -192,6 +202,7 @@ func (t *Tap) teardownRepo(l *slog.Logger, repo *db.Repo, ownerDid syntax.DID, r l.Error("failed to delete repo row", "err", err) return fmt.Errorf("delete repo row: %w", err) } + // TODO: clear sparse-synced git repo return nil } -- 2.51.2