spindle: sync workflow files on `sh.tangled.git.refUpdate`

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.

Signed-off-by: Seongmin Lee <[e-mail]>

go.mod

···
 	github.com/hashicorp/go-secure-stdlib/parseutil v0.2.0 // indirect
 	github.com/hashicorp/go-secure-stdlib/strutil v0.1.2 // indirect
 	github.com/hashicorp/go-sockaddr v1.0.7 // indirect
+	github.com/hashicorp/go-version v1.8.0 // indirect
 	github.com/hashicorp/golang-lru v1.0.2 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/hashicorp/hcl v1.0.1-vault-7 // indirect

go.sum

···
 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=

nix/gomod2nix.toml

···
   [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="

nix/modules/spindle.nix

···
 {
   config,
+  pkgs,
   lib,
   ...
 }: let
···
         description = "spindle service";
         after = ["network.target" "docker.service"];
         wantedBy = ["multi-user.target"];
+        path = [
+          pkgs.git
+        ];
         serviceConfig = {
           LogsDirectory = "spindle";
           StateDirectory = "spindle";

spindle/config/config.go

···
 import (
 	"context"
 	"fmt"
+	"path"
 
 	"github.com/bluesky-social/indigo/atproto/syntax"
 	"github.com/sethvargo/go-envconfig"
···
 	Owner             syntax.DID `env:"OWNER, required"`
 	Secrets           Secrets    `env:",prefix=SECRETS_"`
 	LogDir            string     `env:"LOG_DIR, default=/var/log/spindle"`
+	DataDir           string     `env:"DATA_DIR, default=/var/lib/spindle"`
 	QueueSize         int        `env:"QUEUE_SIZE, default=100"`
 	MaxJobCount       int        `env:"MAX_JOB_COUNT, default=2"` // max number of jobs that run at a time
 }
 
 func (s Server) Did() syntax.DID {
 	return syntax.DID(fmt.Sprintf("did:web:%s", s.Hostname))
+}
+
+func (s Server) RepoDir() string {
+	return path.Join(s.DataDir, "repos")
 }
 
 type Secrets struct {

spindle/server.go

···
 package spindle
 
 import (
+	"bytes"
 	"context"
 	_ "embed"
 	"encoding/json"
···
 	"log/slog"
 	"maps"
 	"net/http"
+	"os"
+	"os/exec"
+	"path"
+	"strings"
 
 	"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"
···
 // New creates a new Spindle server with the provided configuration and engines.
 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 {
···
 }
 
 func (s *Spindle) processPipeline(ctx context.Context, src eventconsumer.Source, msg eventconsumer.Message) 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 {
···
 		} 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("repoDid", event.RepoDid, "repoName", event.RepoName)
+
+		// use event.RepoAt
+		// sync git repos in {data}/repos/{did}/sh.tangled.repo/{rkey}
+		// if it's nil, don't run pipeline. knot needs upgrade
+		// we will leave sh.tangled.pipeline.trigger for backward compatibility
+
+		// NOTE: we are blindly trusting the knot that it will return only repos it own
+		repoCloneUri := s.newRepoCloneUrl(src.Key(), event.RepoDid, event.RepoName)
+		repoPath := s.newRepoPath(event.RepoDid, event.RepoName)
+		err := sparseSyncGitRepo(ctx, repoCloneUri, repoPath, event.NewSha)
+		if err != nil {
+			l.Error("failed to sync git repo", "err", err)
+			return fmt.Errorf("sync git repo: %w", err)
+		}
+		l.Info("synced git repo")
+
+		// TODO: plan the pipeline
 	}
 
 	return nil
 }
+
+func (s *Spindle) newRepoPath(did, name string) string {
+	return path.Join(s.cfg.Server.RepoDir(), did, name)
+}
+
+func (s *Spindle) newRepoCloneUrl(knot, did, name string) string {
+	scheme := "https://"
+	if s.cfg.Server.Dev {
+		scheme = "http://"
+	}
+	return fmt.Sprintf("%s%s/%s/%s", scheme, knot, did, name)
+}
+
+const RequiredVersion = "2.49.0"
+
+func ensureGitVersion() error {
+	v, err := gitVersion()
+	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
+}
+
+// TODO: move to "git" module shared between knot, appview & spindle
+func gitVersion() (*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)
+	}
+
+	// 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)
+}
+
+func sparseSyncGitRepo(ctx context.Context, cloneUri, path, rev string) error {
+	exist, err := isDir(path)
+	if err != nil {
+		return err
+	}
+	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", `'/.tangled/workflows'`).Run(); err != nil {
+			return fmt.Errorf("git sparse-checkout set: %w", err)
+		}
+		if err := exec.Command("git", "-C", path, "checkout", rev).Run(); err != nil {
+			return fmt.Errorf("git checkout: %w", err)
+		}
+	} else {
+		if err := exec.Command("git", "-C", path, "pull", "origin", rev).Run(); err != nil {
+			return fmt.Errorf("git pull: %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
+}