package client import ( "bytes" "cmp" "context" "encoding/json" "errors" "io/fs" "maps" "net/http" "os" "os/exec" "path/filepath" "slices" "strings" "syscall" "unicode/utf8" "lodge/internal/gitrepo" "lodge/internal/protocol" ) const ( maxPasses = 4 tempSuffix = ".lodge-tmp" ) type worktreeState struct { Branch string `json:"branch"` Base map[string]protocol.Manifest `json:"base"` Stash map[string]protocol.Manifest `json:"stash"` } type Result struct { Pushed int `json:"pushed"` Pulled int `json:"pulled"` Skipped []string `json:"skipped"` } type diskFiles struct { shared protocol.Manifest branch protocol.Manifest skipped []string tracked map[string]bool } func worktreePath(dataDir string, repo *gitrepo.Repo) string { return filepath.Join(dataDir, "repos", repo.Key, hashOf([]byte(repo.Root))[:16]) } func (c *Client) statePath(repo *gitrepo.Repo) string { return worktreePath(c.DataDir, repo) + "-" + hashOf([]byte(c.URL))[:16] + ".json" } func (c *Client) loadState(repo *gitrepo.Repo) *worktreeState { st := &worktreeState{} if data, err := os.ReadFile(c.statePath(repo)); err == nil { _ = json.Unmarshal(data, st) } if st.Base == nil { st.Base = map[string]protocol.Manifest{} } if st.Stash == nil { st.Stash = map[string]protocol.Manifest{} } return st } func flock(path string) (func(), error) { if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return nil, err } file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) if err != nil { return nil, err } if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX); err != nil { file.Close() return nil, err } return func() { file.Close() }, nil } func (cfg *Config) Hook(ctx context.Context, dir string) error { repo, err := gitrepo.Open(dir) if err != nil { return err } firstSight := !slices.Contains(cfg.Worktrees(), repo.Root) _, err = cfg.Sync(ctx, repo.Root, firstSight) return err } func (cfg *Config) Sync(ctx context.Context, dir string, online bool) (Result, error) { repo, err := gitrepo.Open(dir) if err != nil { return Result{}, err } worktree := worktreePath(cfg.DataDir, repo) unlock, err := flock(worktree + ".lock") if err != nil { return Result{}, err } defer unlock() current, err := cfg.reload() if err != nil { return Result{}, err } c, err := current.Server(repo.Key) if err != nil { return Result{}, err } if err := cfg.register(repo.Root); err != nil { return Result{}, err } if c.URL == current.URL { _ = os.Rename(worktree+".json", c.statePath(repo)) } return c.sync(ctx, repo, online) } func (c *Client) sync(ctx context.Context, repo *gitrepo.Repo, online bool) (Result, error) { var res Result st := c.loadState(repo) for range maxPasses { again, err := c.pass(ctx, repo, st, online, &res) saveErr := writeJSONFile(c.statePath(repo), st) if err != nil || saveErr != nil || !again { return res, errors.Join(err, saveErr) } } return res, errors.New("gave up: server state keeps changing") } func (c *Client) scan(repo *gitrepo.Repo, rules Rules) diskFiles { found := diskFiles{shared: protocol.Manifest{}, branch: protocol.Manifest{}, tracked: repo.Tracked(rules.paths())} for _, rule := range rules.synced() { start := filepath.Join(repo.Root, filepath.FromSlash(rule.Path)) _ = filepath.WalkDir(start, func(full string, entry fs.DirEntry, err error) error { if err != nil { return nil } rel, _ := filepath.Rel(repo.Root, full) rel = filepath.ToSlash(rel) if entry.IsDir() && (entry.Name() == ".git" || rules.ignored(rel)) { return filepath.SkipDir } if !entry.Type().IsRegular() || strings.HasSuffix(full, tempSuffix) { return nil } if found.tracked[rel] || !rules.covers(rel, rule.Shared) { return nil } info, err := entry.Info() if err != nil { return nil } if info.Size() > protocol.MaxBlobSize { found.skipped = append(found.skipped, rel) return nil } data, err := os.ReadFile(full) if err != nil { return nil } hash, err := c.cacheBlob(data) if err != nil { return nil } manifest := found.branch if rule.Shared { manifest = found.shared } manifest[rel] = protocol.File{Hash: hash, Time: info.ModTime().UnixMilli()} return nil }) } return found } type mergeFunc func(local, base, remote protocol.File) (protocol.File, bool) func reconcile(space string, local, base, remote protocol.Manifest, relevant func(string) bool, merge mergeFunc) (first, second []protocol.Change, pull, next protocol.Manifest) { pull, next = protocol.Manifest{}, maps.Clone(base) if next == nil { next = protocol.Manifest{} } paths := map[string]bool{} for _, manifest := range []protocol.Manifest{local, base, remote} { for path := range manifest { paths[path] = true } } for path := range paths { if !relevant(path) { continue } l, b, r := local[path], base[path], remote[path] if l.Hash != b.Hash && r.Hash != b.Hash && l.Hash != r.Hash { if merged, ok := merge(l, b, r); ok { if merged.Hash != l.Hash { pull[path] = merged } l, b = merged, r } } mine := protocol.Change{Space: space, Path: path, Base: r.Hash, Hash: l.Hash} settled := r switch { case l.Hash == r.Hash: case l.Hash == b.Hash, l.Hash == "" && r.Hash != b.Hash: pull[path] = r case r.Hash == b.Hash, r.Hash == "", l.Time >= r.Time: first = append(first, mine) settled = l default: first = append(first, mine) second = append(second, protocol.Change{Space: space, Path: path, Base: l.Hash, Hash: r.Hash}) pull[path] = r } if settled.Hash == "" { delete(next, path) } else { next[path] = settled } } return first, second, pull, next } func (c *Client) merge(ctx context.Context, local, base, remote protocol.File) (protocol.File, bool) { names := []string{"local", "base", "remote"} versions := make([][]byte, len(names)) for i, file := range []protocol.File{local, base, remote} { if file.Hash == "" { return protocol.File{}, false } data, err := c.Blob(ctx, file.Hash) if err != nil || !isText(data) { return protocol.File{}, false } versions[i] = data } dir, err := os.MkdirTemp(c.DataDir, "merge-") if err != nil { return protocol.File{}, false } defer os.RemoveAll(dir) for i, name := range names { if err := os.WriteFile(filepath.Join(dir, name), versions[i], 0o600); err != nil { return protocol.File{}, false } } cmd := exec.CommandContext(ctx, "git", append([]string{"merge-file", "-p"}, names...)...) cmd.Dir = dir merged, err := cmd.Output() clean := err == nil if !clean { merged, clean = appended(versions[0], versions[1], versions[2]) } if !clean || len(merged) > protocol.MaxBlobSize { return protocol.File{}, false } hash, err := c.cacheBlob(merged) return protocol.File{Hash: hash, Time: max(local.Time, remote.Time)}, err == nil } func appended(local, base, remote []byte) ([]byte, bool) { if !bytes.HasPrefix(local, base) || !bytes.HasPrefix(remote, base) { return nil, false } tail, glue := local[len(base):], []byte{} if !bytes.HasSuffix(remote, []byte("\n")) && !bytes.HasPrefix(tail, []byte("\n")) { glue = []byte("\n") } return slices.Concat(remote, glue, tail), true } func isText(data []byte) bool { return len(data) <= protocol.MaxBlobSize && utf8.Valid(data) && bytes.IndexByte(data, 0) == -1 } func (c *Client) pass(ctx context.Context, repo *gitrepo.Repo, st *worktreeState, online bool, res *Result) (bool, error) { rules := LoadRules(repo.Root) if err := writeExclude(repo, rules); err != nil { return false, err } var remote *protocol.State var offlineErr error if online { state, err := c.State(ctx, repo.Key, "") remote, offlineErr = &state, err if err != nil { remote = nil } } found := c.scan(repo, rules) branch := cmp.Or(repo.Branch(), st.Branch) if st.Branch != "" && branch != st.Branch { if err := c.switchBranch(ctx, repo, st, rules, branch, found, remote); err != nil { return false, err } found = c.scan(repo, rules) } st.Branch = branch res.Skipped = found.skipped if remote == nil { return false, offlineErr } if remote.Seq == 0 { st.Base = map[string]protocol.Manifest{} } locals := map[string]protocol.Manifest{protocol.SharedSpace: found.shared} if branch != "" { locals[protocol.BranchSpace(branch)] = found.branch } active := maps.Clone(locals) for space, stashed := range st.Stash { if active[space] == nil { locals[space] = stashed } } var first, second []protocol.Change pulls, bases := map[string]protocol.Manifest{}, map[string]protocol.Manifest{} spaces := map[string]bool{} merge := func(l, b, r protocol.File) (protocol.File, bool) { return c.merge(ctx, l, b, r) } for _, group := range []map[string]protocol.Manifest{locals, st.Base, remote.Spaces} { for space := range group { spaces[space] = true } } for space := range spaces { local, syncing := locals[space] if !syncing { st.Base[space] = remote.Spaces[space] if st.Base[space] == nil { delete(st.Base, space) } continue } relevant := func(path string) bool { return protocol.SafePath(path) && !rules.ignored(path) } if active[space] != nil { relevant = func(path string) bool { return protocol.SafePath(path) && rules.covers(path, space == protocol.SharedSpace) && !found.tracked[path] } } mine, theirs, pull, next := reconcile(space, local, st.Base[space], remote.Spaces[space], relevant, merge) first, second, pulls[space], bases[space] = append(first, mine...), append(second, theirs...), pull, next } for _, change := range first { if change.Hash == "" { continue } if err := c.uploadBlob(ctx, change.Hash); err != nil { return false, err } } for _, changes := range [][]protocol.Change{first, second} { if len(changes) == 0 { continue } push := protocol.Push{Name: repo.Name, MainSHA: repo.Head(), MainBranch: branch, Changes: changes} err := c.call(ctx, http.MethodPost, "/api/repos/"+repo.Key+"/revisions", push, nil) if errors.Is(err, errConflict) { return true, nil } if err != nil { return false, err } } res.Pushed += len(first) again := false for space, pull := range pulls { for path, file := range pull { if active[space] != nil { if err := c.materialize(ctx, repo.Root, path, file); err != nil { return false, err } } else if file.Hash == "" { delete(st.Stash[space], path) } else if _, err := c.Blob(ctx, file.Hash); err != nil { return false, err } else { st.Stash[space][path] = file } res.Pulled++ again = again || path == RulesFile } } maps.Copy(st.Base, bases) return again, nil } func (c *Client) switchBranch(ctx context.Context, repo *gitrepo.Repo, st *worktreeState, rules Rules, branch string, found diskFiles, remote *protocol.State) error { space := protocol.BranchSpace(branch) st.Stash[protocol.BranchSpace(st.Branch)] = found.branch known, exists := st.Stash[space] if !exists { known, exists = st.Base[space] } if !exists && remote != nil { known, exists = remote.Spaces[space] } if !exists { return nil } incoming := protocol.Manifest{} for path, file := range known { if !protocol.SafePath(path) || !rules.covers(path, false) || found.tracked[path] { continue } if _, err := c.Blob(ctx, file.Hash); err != nil { return err } incoming[path] = file } for path := range found.branch { if err := c.materialize(ctx, repo.Root, path, protocol.File{}); err != nil { return err } } for path, file := range incoming { if err := c.materialize(ctx, repo.Root, path, file); err != nil { return err } } delete(st.Stash, space) return nil } func (c *Client) materialize(ctx context.Context, root, path string, file protocol.File) error { full := filepath.Join(root, filepath.FromSlash(path)) if file.Hash != "" { data, err := c.Blob(ctx, file.Hash) if err != nil { return err } return writeFile(full, data, 0o644) } if err := os.Remove(full); err != nil && !errors.Is(err, fs.ErrNotExist) { return err } for dir := filepath.Dir(full); dir != root; dir = filepath.Dir(dir) { if os.Remove(dir) != nil { break } } return nil }