diff --git a/go.mod b/go.mod --- a/go.mod +++ b/go.mod @@ -47,6 +47,7 @@ github.com/gorilla/feeds v1.2.0 github.com/gorilla/sessions v1.4.0 github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 + github.com/hashicorp/go-version v1.8.0 github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/hiddeco/sshsig v0.2.0 github.com/hpcloud/tail v1.0.0 diff --git a/go.sum b/go.sum --- a/go.sum +++ b/go.sum @@ -445,6 +445,8 @@ 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 --- a/nix/gomod2nix.toml +++ b/nix/gomod2nix.toml @@ -479,6 +479,9 @@ [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/spindle/server.go b/spindle/server.go --- a/spindle/server.go +++ b/spindle/server.go @@ -8,11 +8,13 @@ "log/slog" "maps" "net/http" + "path/filepath" "sync" "time" "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" @@ -28,6 +30,7 @@ "tangled.org/core/spindle/engines/dummy" "tangled.org/core/spindle/engines/microvm" "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" @@ -328,6 +331,10 @@ return fmt.Errorf("failed to load config: %w", err) } + if err := ensureGitVersion(); err != nil { + return fmt.Errorf("ensuring git version: %w", err) + } + d, err := db.Make(ctx, cfg.Server.DBPath) if err != nil { return fmt.Errorf("failed to setup db: %w", err) @@ -389,7 +396,10 @@ } 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 { @@ -497,8 +507,58 @@ } 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 } diff --git a/spindle/tapclient.go b/spindle/tapclient.go --- a/spindle/tapclient.go +++ b/spindle/tapclient.go @@ -16,6 +16,7 @@ "tangled.org/core/log" "tangled.org/core/rbac" "tangled.org/core/spindle/db" + "tangled.org/core/spindle/git" "tangled.org/core/tapc" ) @@ -122,15 +123,24 @@ 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 := "" @@ -192,6 +202,7 @@ 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 } diff --git a/nix/modules/spindle.nix b/nix/modules/spindle.nix --- a/nix/modules/spindle.nix +++ b/nix/modules/spindle.nix @@ -32,6 +32,12 @@ 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"; @@ -301,6 +307,7 @@ config = let deps = [ + pkgs.git pkgs.qemu pkgs.e2fsprogs pkgs.slirp4netns @@ -341,6 +348,7 @@ 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 --- a/spindle/config/config.go +++ b/spindle/config/config.go @@ -12,6 +12,7 @@ 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 --- /dev/null +++ b/spindle/git/git.go @@ -0,0 +1,105 @@ +package git + +import ( + "bytes" + "context" + "fmt" + "os" + "os/exec" + "strings" + "sync" + + "github.com/hashicorp/go-version" +) + +// repoLocks serializes git operations per repo directory. Concurrent triggers +// on the same repo (a push landing while a manual run is dispatched, two "Run +// CI" clicks, etc.) resolve to the same path with different revisions; running +// clone/fetch/checkout there in parallel collides on .git/index.lock and can +// corrupt the dir. Locking is keyed by path so unrelated repos don't serialize. +var repoLocks keyedMutex + +type keyedMutex struct { + mu sync.Mutex + m map[string]*sync.Mutex +} + +// lock acquires the mutex for key and returns its unlock func. +func (k *keyedMutex) lock(key string) func() { + k.mu.Lock() + if k.m == nil { + k.m = make(map[string]*sync.Mutex) + } + mu, ok := k.m[key] + if !ok { + mu = &sync.Mutex{} + k.m[key] = mu + } + k.mu.Unlock() + + mu.Lock() + return mu.Unlock +} + +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 { + defer repoLocks.lock(path)() + + 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 +}