// SPDX-License-Identifier: Apache-2.0 // Copyright (c) 2024-2026 Quad4.io package rnsgit import ( "bufio" "context" "fmt" "io" "os" "path/filepath" "strings" "time" "quad4/reticulum-go/pkg/destination" "quad4/reticulum-go/pkg/identity" "quad4/reticulum-go/pkg/link" "quad4/reticulum-go/pkg/node" "quad4/reticulum-go/pkg/rnsutil" "quad4/reticulum-go/pkg/transport" ) // ClientOptions configures the git-remote-rns helper. type ClientOptions struct { ConfigDir string RNSConfigDir string DestHex string Group string Repo string Timeout time.Duration JSONProgress bool ProgressWriter io.Writer } // Client implements git-remote-rns over Reticulum links. type Client struct { cfg *ClientConfig destHex string repoPath string identity *identity.Identity tr *transport.Transport nodeStop func() link *link.Link refBatchSize int remoteRefs map[string]string progress bool jsonProgress bool progressWriter io.Writer tmpDir string } // NewClient creates a git-remote-rns client. func NewClient(opts ClientOptions) (*Client, error) { if err := EnsureClientConfig(opts.ConfigDir); err != nil { return nil, err } cfg, err := LoadClientConfig(opts.ConfigDir) if err != nil { return nil, err } if opts.RNSConfigDir != "" { cfg.RNSConfigDir = opts.RNSConfigDir } destHex := opts.DestHex if alias, ok := cfg.DestAliases[destHex]; ok { destHex = alias } id, err := PrepareGitIdentity(cfg.IdentityPath) if err != nil { return nil, fmt.Errorf("identity: %w", err) } tmp, err := os.MkdirTemp("", "rnsgit-client-") if err != nil { return nil, err } batch := cfg.RefBatchSize if batch <= 0 { batch = DefaultRefBatchSize } return &Client{ cfg: cfg, destHex: destHex, repoPath: RepoPath(opts.Group, opts.Repo), identity: id, refBatchSize: batch, remoteRefs: map[string]string{}, progressWriter: opts.ProgressWriter, jsonProgress: opts.JSONProgress, tmpDir: tmp, }, nil } // Close releases temporary resources. func (c *Client) Close() { if c.link != nil { c.link.Teardown() } if c.nodeStop != nil { c.nodeStop() } _ = os.RemoveAll(c.tmpDir) } // UseTransport attaches an existing Reticulum transport for Connect. func (c *Client) UseTransport(tr *transport.Transport, stop func()) { c.tr = tr c.nodeStop = stop } // Connect starts transport and opens a link to the git node. func (c *Client) Connect(ctx context.Context) error { status := c.progressWriter statusLine(status, "Requesting path...") c.emitProgress("path_wait", nil) if c.tr == nil { rnsCfg, err := rnsutil.LoadConfigDir(c.cfg.RNSConfigDir) if err != nil { return err } n, err := node.New(rnsCfg) if err != nil { return err } if err := n.Start(); err != nil { return err } c.nodeStop = func() { _ = n.Stop() } c.tr = n.Transport() } if err := waitTransportOutgoing(ctx, c.tr); err != nil { c.stopAttachedNode() return err } destHash, err := rnsutil.ParseDestHash(c.destHex) if err != nil { c.stopAttachedNode() return err } if err := rnsutil.WaitPathWindow(ctx, c.tr, destHash); err != nil { c.stopAttachedNode() return fmt.Errorf("path: %w", err) } statusLinef(status, "\rPath resolved \n") c.emitProgress("path_resolved", nil) statusLine(status, "Establishing link...") remote, err := identity.Recall(destHash) if err != nil { c.stopAttachedNode() return fmt.Errorf("recall: %w", err) } outDest, err := destination.New(remote, destination.Out, destination.Single, AppName, c.tr, Aspect) if err != nil { c.stopAttachedNode() return err } l := link.NewLink(outDest, c.tr, nil, nil, nil) if err := activateGitLink(ctx, l); err != nil { c.stopAttachedNode() return err } if err := l.Identify(c.identity); err != nil { l.Teardown() c.stopAttachedNode() return fmt.Errorf("identify: %w", err) } time.Sleep(100 * time.Millisecond) c.link = l c.refBatchSize = dynamicRefBatch(c.refBatchSize, l, c.tr) statusLinef(status, "Link established with remote\n") c.emitProgress("link_ready", map[string]any{"ref_batch_size": c.refBatchSize}) return nil } func (c *Client) stopAttachedNode() { if c.nodeStop != nil { c.nodeStop() c.nodeStop = nil } c.tr = nil } func activateGitLink(ctx context.Context, l *link.Link) error { if err := l.Establish(); err != nil { return err } wait, cancel := rnsutil.BoundWait(ctx, rnsutil.LinkEstablishmentWindow(l)) defer cancel() if err := rnsutil.WaitLinkActive(wait, l); err != nil { l.Teardown() return err } return nil } type batchLink interface { GetRTT() float64 } type batchTransport interface { SlowestOnlineBitrate() int64 } func dynamicRefBatch(base int, l batchLink, tr batchTransport) int { if base <= 0 { base = DefaultRefBatchSize } batch := base if l != nil { rtt := l.GetRTT() if rtt > 2.0 { n := max(batch/2, 4) batch = n } else if rtt > 0 && rtt < 0.2 { n := min(batch*2, 64) batch = n } } if tr != nil { br := tr.SlowestOnlineBitrate() switch { case br > 0 && br < 100_000: n := max(batch/2, 4) batch = n case br >= 10_000_000: n := min(batch*2, 64) batch = n } } return batch } // RunGitHelper executes the git remote-helper protocol on stdin/stdout. func (c *Client) RunGitHelper(ctx context.Context, stdin io.Reader, stdout, stderr io.Writer) error { if c.progressWriter == nil && stderr != nil { c.progressWriter = stderr } if err := c.Connect(ctx); err != nil { return err } in := bufio.NewReader(stdin) fetchQueue := make([][2]string, 0) pushQueue := make([][2]string, 0) for { line, err := in.ReadString('\n') if err == io.EOF { break } if err != nil { return err } line = strings.TrimRight(line, "\r\n") if line == "" { if err := c.flushQueues(ctx, fetchQueue, pushQueue, stdout, stderr); err != nil { return err } fetchQueue = fetchQueue[:0] pushQueue = pushQueue[:0] fmt.Fprintln(stdout) continue } switch { case line == "capabilities": fmt.Fprintln(stdout, "list") fmt.Fprintln(stdout, "fetch") fmt.Fprintln(stdout, "push") fmt.Fprintln(stdout, "option") fmt.Fprintln(stdout) case line == "list": if err := c.handleList(ctx, stdout, false); err != nil { return err } case strings.HasPrefix(line, "list "): if err := c.handleList(ctx, stdout, true); err != nil { return err } case strings.HasPrefix(line, "option"): parts := strings.Fields(line) if len(parts) >= 3 && parts[1] == "progress" { c.progress = parts[2] == "true" || parts[2] == "1" || parts[2] == "yes" fmt.Fprintln(stdout, "ok") } else { fmt.Fprintln(stdout, "unsupported") } case strings.HasPrefix(line, "fetch"): parts := strings.Fields(line) if len(parts) >= 3 { fetchQueue = append(fetchQueue, [2]string{parts[1], parts[2]}) pushQueue = pushQueue[:0] } case strings.HasPrefix(line, "push"): parts := strings.Fields(line) if len(parts) >= 2 { spec := parts[1] local, remote, ok := strings.Cut(spec, ":") if !ok { local, remote = spec, "" } pushQueue = append(pushQueue, [2]string{local, remote}) fetchQueue = fetchQueue[:0] } default: return fmt.Errorf("unknown git command: %s", line) } } return nil } func (c *Client) handleList(ctx context.Context, stdout io.Writer, forPush bool) error { req := map[any]any{IdxRepository: c.repoPath} if forPush { req["for_push"] = true } body, err := c.sendRequest(ctx, PathList, req, 120*time.Second) if err != nil { return err } if len(body) == 0 || body[0] != ResOK { msg := string(body[1:]) return fmt.Errorf("list failed: %s", msg) } text := string(body[1:]) c.remoteRefs = map[string]string{} for line := range strings.SplitSeq(text, "\n") { line = strings.TrimSpace(line) if line == "" { continue } parts := strings.SplitN(line, " ", 2) if len(parts) != 2 || parts[1] == "HEAD" { continue } c.remoteRefs[parts[1]] = parts[0] } fmt.Fprint(stdout, text) if !strings.HasSuffix(text, "\n") { fmt.Fprintln(stdout) } return nil } func (c *Client) flushQueues(ctx context.Context, fetch [][2]string, push [][2]string, stdout, stderr io.Writer) error { if len(fetch) > 0 { if err := c.processFetch(ctx, fetch, stderr); err != nil { return err } } for _, item := range push { if err := c.processPush(ctx, item[0], item[1], stdout, stderr); err != nil { return err } } return nil } func (c *Client) processFetch(ctx context.Context, queue [][2]string, stderr io.Writer) error { have := make([]string, 0) for _, sha := range c.remoteRefs { if out, err := localGitCmd("cat-file", "-t", sha).Output(); err == nil && strings.TrimSpace(string(out)) != "" { have = append(have, sha) } } for len(queue) > 0 { n := min(c.refBatchSize, len(queue)) batch := queue[:n] queue = queue[n:] refs := make([]map[string]string, 0, len(batch)) for _, item := range batch { entry := map[string]string{"sha": item[0], "ref": item[1]} if out, err := localGitCmd("rev-parse", item[1]).Output(); err == nil { local := strings.TrimSpace(string(out)) if local != "" && local != item[0] { entry["have"] = local } } refs = append(refs, entry) } reqMap := map[any]any{IdxRepository: c.repoPath, "refs": refs} if len(have) > 0 { reqMap["have"] = have } c.emitProgress("fetch_batch", map[string]any{"refs": len(batch), "remaining": len(queue)}) body, meta, err := c.sendRequestWithMeta(ctx, PathFetch, reqMap, 2*time.Hour) if err != nil { return err } if body == nil && meta == nil { return fmt.Errorf("empty fetch response") } if len(body) == 1 && body[0] == ResOK { continue } bundlePath, err := c.writeFetchBundle(body, meta) if err != nil { return err } if bundlePath == "" { continue } c.reportFetchTransfer(stderr, bundlePath) if err := localGitCmd("bundle", "verify", "-q", bundlePath).Run(); err != nil { return fmt.Errorf("bundle verify failed") } args := []string{"bundle", "unbundle", bundlePath} if c.progress { args = []string{"bundle", "unbundle", "--progress", bundlePath} } cmd := localGitCmd(args...) if c.progress { cmd.Stderr = stderr } if err := cmd.Run(); err != nil { return fmt.Errorf("bundle unbundle: %w", err) } c.emitProgress("fetch_unbundled", map[string]any{"refs": len(batch)}) } return nil } func (c *Client) writeFetchBundle(body []byte, meta map[string]any) (string, error) { if meta != nil { if code, ok := MetadataResultCode(meta); ok && code != ResOK { return "", fmt.Errorf("fetch failed code %d", code) } if len(body) == 0 { return "", nil } bundlePath := filepath.Join(c.tmpDir, "fetch.bundle") if err := os.WriteFile(bundlePath, body, 0o600); err != nil { return "", err } return bundlePath, nil } if len(body) == 0 { return "", nil } if body[0] != ResOK { if isGitBundle(body) { bundlePath := filepath.Join(c.tmpDir, "fetch.bundle") if err := os.WriteFile(bundlePath, body, 0o600); err != nil { return "", err } return bundlePath, nil } return "", fmt.Errorf("fetch failed: %s", string(body[1:])) } if len(body) == 1 { return "", nil } bundlePath := filepath.Join(c.tmpDir, "fetch.bundle") if err := os.WriteFile(bundlePath, body[1:], 0o600); err != nil { return "", err } return bundlePath, nil } func (c *Client) reportFetchTransfer(stderr io.Writer, bundlePath string) { st, err := os.Stat(bundlePath) if err != nil { return } c.emitProgress("fetch_transfer", map[string]any{"bytes": st.Size()}) if !c.progress || stderr == nil { return } fmt.Fprintf(stderr, "Transferring: 100%% (%d bytes).\n", st.Size()) } func (c *Client) processPush(ctx context.Context, localRef, remoteRef string, stdout, stderr io.Writer) error { if localRef == "" { body, err := c.sendRequest(ctx, PathDelete, map[any]any{IdxRepository: c.repoPath, "ref": remoteRef}, 120*time.Second) if err != nil || len(body) == 0 || body[0] != ResOK { fmt.Fprintf(stdout, "error %s %s\n", remoteRef, EscapeGitStdout("delete failed")) return nil } fmt.Fprintf(stdout, "ok %s\n", remoteRef) return nil } force := strings.HasPrefix(localRef, "+") if force { localRef = localRef[1:] } localSHA, err := localGitCmd("rev-parse", localRef).Output() if err != nil { fmt.Fprintf(stdout, "error %s %s\n", remoteRef, EscapeGitStdout("could not resolve local ref")) return nil } sha := strings.TrimSpace(string(localSHA)) bundlePath := filepath.Join(c.tmpDir, "push.bundle") createArgs := []string{"bundle", "create", bundlePath, localRef} for _, rsha := range c.remoteRefs { if out, err := localGitCmd("cat-file", "-t", rsha).Output(); err == nil && len(out) > 0 { createArgs = append(createArgs, "^"+rsha) } } create := localGitCmd(createArgs...) if c.progress { create.Stderr = stderr } bundleEmpty := false if err := create.Run(); err != nil { out, _ := create.CombinedOutput() if !strings.Contains(strings.ToLower(string(out)), "empty bundle") { fmt.Fprintf(stdout, "error %s %s\n", remoteRef, EscapeGitStdout("bundle creation failed")) return nil } bundleEmpty = true } if !bundleEmpty { data, err := os.ReadFile(bundlePath) // #nosec G304 -- temp bundle path if err != nil { return err } body, err := c.sendRequest(ctx, PathPush, map[any]any{ IdxRepository: c.repoPath, "local_ref": localRef, "remote_ref": remoteRef, "force": force, "bundle": data, }, 2*time.Hour) if err != nil || len(body) == 0 || body[0] != ResOK { msg := "push failed" if len(body) > 1 { msg = string(body[1:]) } fmt.Fprintf(stdout, "error %s %s\n", remoteRef, EscapeGitStdout(msg)) return nil } } else { body, err := c.sendRequest(ctx, PathPush, map[any]any{ IdxRepository: c.repoPath, "operations": []map[string]any{{ "action": "update_ref", "ref": remoteRef, "sha": sha, "force": force, }}, }, 120*time.Second) if err != nil || len(body) == 0 || body[0] != ResOK { fmt.Fprintf(stdout, "error %s %s\n", remoteRef, EscapeGitStdout("ref update failed")) return nil } } fmt.Fprintf(stdout, "ok %s\n", remoteRef) return nil } func (c *Client) sendRequest(ctx context.Context, path string, payload any, timeout time.Duration) ([]byte, error) { body, _, err := c.sendRequestWithMeta(ctx, path, payload, timeout) return body, err } func (c *Client) sendRequestWithMeta(ctx context.Context, path string, payload any, timeout time.Duration) ([]byte, map[string]any, error) { if c.link == nil { return nil, nil, fmt.Errorf("link not ready") } receipt, err := c.link.Request(path, payload, timeout) if err != nil { return nil, nil, err } if err := rnsutil.WaitRequest(ctx, receipt); err != nil { return nil, nil, err } if receipt.GetStatus() != link.StatusActive { return nil, nil, fmt.Errorf("request failed") } meta := receipt.GetMetadata() if meta != nil { return receipt.GetResponse(), meta, nil } return receipt.GetResponse(), nil, nil } // RunGitRemoteRNS is the entry point for git-remote-rns. func RunGitRemoteRNS(args []string, rnsConfig string) int { jsonOut := false configDir := "" rnsDir := rnsConfig pos := 0 for pos < len(args) && strings.HasPrefix(args[pos], "-") { switch args[pos] { case "-json", "--json": jsonOut = true case "-config": if pos+1 >= len(args) { fmt.Fprintln(os.Stderr, "missing value for -config") return 2 } configDir = args[pos+1] pos++ case "-rnsconfig": if pos+1 >= len(args) { fmt.Fprintln(os.Stderr, "missing value for -rnsconfig") return 2 } rnsDir = args[pos+1] pos++ default: fmt.Fprintf(os.Stderr, "unknown flag %q\n", args[pos]) return 2 } pos++ } args = args[pos:] if len(args) < 2 { fmt.Fprintln(os.Stderr, "Usage: git-remote-rns [-json] ") return 1 } url := args[1] if !strings.HasPrefix(strings.ToLower(url), ProtoRNS) { fmt.Fprintln(os.Stderr, "Invalid URL scheme. Must be rns://") return 1 } destHex, group, repo, err := ParseRNSURL(url) if err != nil { fmt.Fprintln(os.Stderr, err) return 1 } cfgDir := configDir if cfgDir == "" { cfgDir = os.Getenv("RNGIT_CONFIG") } client, err := NewClient(ClientOptions{ ConfigDir: cfgDir, RNSConfigDir: rnsDir, DestHex: destHex, Group: group, Repo: repo, JSONProgress: jsonOut, ProgressWriter: os.Stderr, }) if err != nil { fmt.Fprintf(os.Stderr, "git-remote-rns failed: %v\n", err) return 255 } defer client.Close() ctx, cancel := rnsutil.CLIWaitContext(0) defer cancel() if err := client.RunGitHelper(ctx, os.Stdin, os.Stdout, os.Stderr); err != nil { fmt.Fprintf(os.Stderr, "git-remote-rns failed: %v\n", err) return 255 } return 0 } func waitTransportOutgoing(ctx context.Context, tr *transport.Transport) error { if tr == nil { return fmt.Errorf("nil transport") } deadline := time.Now().Add(5 * time.Second) ticker := time.NewTicker(50 * time.Millisecond) defer ticker.Stop() for { if tr.SlowestOnlineBitrate() > 0 { return nil } if time.Now().After(deadline) { return nil } select { case <-ctx.Done(): return ctx.Err() case <-ticker.C: } } }