feat: diffx-fetcher with leased, repo-scoped pull request fetches

Dockerfile

@@ -10,6 +10,22 @@ ARG TARGETARCH
 RUN CGO_ENABLED=0 GOOS=${TARGETOS} GOARCH=${TARGETARCH} go build -trimpath -o /diffx-server ./cmd/diffx-server
 RUN mkdir /data && chown 10001:10001 /data
 
+FROM build AS fetcher-build
+ARG TARGETOS=linux
+ARG TARGETARCH
+RUN CGO_ENABLED=0 GOOS=${TARGETOS} GOARCH=${TARGETARCH} go build -trimpath -o /diffx-fetcher ./cmd/diffx-fetcher
+
+# GitHub pull request fetcher: `docker build --target fetcher .`. Git 2.44+
+# keeps collection offline; the server image below stays the default target.
+FROM alpine:3.22 AS fetcher
+RUN apk add --no-cache git ca-certificates && mkdir /mirrors && chown 10001:10001 /mirrors
+COPY --from=fetcher-build /diffx-fetcher /diffx-fetcher
+USER 10001:10001
+ENV DIFFX_FETCHER_MIRRORS=/mirrors
+VOLUME ["/mirrors"]
+STOPSIGNAL SIGTERM
+ENTRYPOINT ["/diffx-fetcher"]
+
 FROM scratch
 COPY --from=build /diffx-server /diffx-server
 COPY --from=build --chown=10001:10001 /data /data

cmd/diffx-fetcher/main.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/flexdinesh/diffx/internal/config"
+	"github.com/flexdinesh/diffx/internal/fetcher"
+	"github.com/flexdinesh/diffx/internal/fetchprotocol"
+	"github.com/flexdinesh/diffx/internal/version"
+)
+
+func main() {
+	if err := run(os.Args[1:]); err != nil {
+		if err == flag.ErrHelp {
+			return
+		}
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
+}
+
+func run(arguments []string) error {
+	flags := flag.NewFlagSet("diffx-fetcher", flag.ContinueOnError)
+	showVersion := flags.Bool("version", false, "print version and exit")
+	if err := flags.Parse(arguments); err != nil {
+		return err
+	}
+	if *showVersion {
+		fmt.Println(version.String())
+		return nil
+	}
+	if flags.NArg() != 0 {
+		return fmt.Errorf("diffx-fetcher takes no positional arguments; configure it with DIFFX_FETCHER_* environment")
+	}
+	values, err := config.LoadFetcher()
+	if err != nil {
+		return err
+	}
+	client, err := fetchprotocol.NewClient(values.ServerURL, values.Secret)
+	if err != nil {
+		return err
+	}
+	// Collection inherits this process environment; it must never fetch.
+	if err := os.Setenv("GIT_NO_LAZY_FETCH", "1"); err != nil {
+		return err
+	}
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	fmt.Printf("diffx-fetcher polling %s\n", values.ServerURL)
+	return fetcher.Run(ctx, client, fetcher.Settings{
+		Mirrors: values.Mirrors, DiskLimit: values.DiskLimitBytes, Concurrency: values.Concurrency, CollectorVersion: version.String(),
+	})
+}

cmd/diffx-server/main.go

@@ -83,8 +83,13 @@ func parseSettings(arguments []string, stderr io.Writer) (remoteserver.Settings,
 		return settings, false, err
 	}
 	if github := values.GitHub; github.Enabled() {
+		internal, err := values.InternalAddress(settings.Listen)
+		if err != nil {
+			return settings, false, err
+		}
 		settings.GitHub = &remoteserver.GitHubSettings{AppID: github.AppID, ClientID: github.ClientID, ClientSecret: github.ClientSecret,
-			PrivateKeyFile: github.PrivateKeyFile, WebhookSecret: github.WebhookSecret, TokenKey: github.TokenKey, PublicURL: github.PublicURL}
+			PrivateKeyFile: github.PrivateKeyFile, WebhookSecret: github.WebhookSecret, TokenKey: github.TokenKey, PublicURL: github.PublicURL,
+			InternalListen: internal, FetcherSecret: github.FetcherSecret}
 	}
 	return settings, false, nil
 }

compose.yaml

@@ -16,14 +16,59 @@ services:
       DIFFX_SERVER_SIGNUP_ENABLED: "${DIFFX_SERVER_SIGNUP_ENABLED:-false}"
       # Loopback HTTP development only. Use false behind production HTTPS.
       DIFFX_SERVER_INSECURE_COOKIES: "${DIFFX_SERVER_INSECURE_COOKIES:-true}"
+      # GitHub App integration: set all or none (empty disables it), then start
+      # with `--profile github`. Mount the private key read-only in an override
+      # file and point PRIVATE_KEY_FILE at it.
+      DIFFX_SERVER_GITHUB_APP_ID: "${DIFFX_SERVER_GITHUB_APP_ID:-}"
+      DIFFX_SERVER_GITHUB_CLIENT_ID: "${DIFFX_SERVER_GITHUB_CLIENT_ID:-}"
+      DIFFX_SERVER_GITHUB_PRIVATE_KEY_FILE: "${DIFFX_SERVER_GITHUB_PRIVATE_KEY_FILE:-}"
+      DIFFX_SERVER_GITHUB_PUBLIC_URL: "${DIFFX_SERVER_GITHUB_PUBLIC_URL:-}"
+      DIFFX_SERVER_GITHUB_CLIENT_SECRET: "${DIFFX_SERVER_GITHUB_CLIENT_SECRET:-}"
+      DIFFX_SERVER_GITHUB_WEBHOOK_SECRET: "${DIFFX_SERVER_GITHUB_WEBHOOK_SECRET:-}"
+      DIFFX_SERVER_GITHUB_TOKEN_KEY: "${DIFFX_SERVER_GITHUB_TOKEN_KEY:-}"
+      # The fetcher listener (internal port 7979) is reachable only on the fetch
+      # network; never publish it.
+      DIFFX_SERVER_FETCHER_SECRET: "${DIFFX_SERVER_FETCHER_SECRET:-}"
     ports:
       # Only the published host port varies; local `diffx [PATH]` uses 7981-7990.
       - "127.0.0.1:${DIFFX_SERVER_PORT:-7980}:7980"
+    networks:
+      - default
+      - fetch
     volumes:
       - diffx-data:/data
     read_only: true
     restart: unless-stopped
     stop_grace_period: 30s
 
+  # Mirrors leased pull requests from GitHub; it has no published ports.
+  fetcher:
+    profiles: ["github"]
+    build:
+      context: .
+      target: fetcher
+    image: diffx-fetcher:local
+    environment:
+      DIFFX_FETCHER_SERVER_URL: http://server:7979
+      DIFFX_FETCHER_SECRET: "${DIFFX_SERVER_FETCHER_SECRET:-}"
+      DIFFX_FETCHER_DISK_LIMIT_GB: "${DIFFX_FETCHER_DISK_LIMIT_GB:-20}"
+      DIFFX_FETCHER_CONCURRENCY: "${DIFFX_FETCHER_CONCURRENCY:-2}"
+    networks:
+      - fetch
+    volumes:
+      - diffx-mirrors:/mirrors
+    tmpfs:
+      - /tmp
+    read_only: true
+    depends_on:
+      - server
+    restart: unless-stopped
+    stop_grace_period: 30s
+
+networks:
+  # Shared by server and fetcher only; the fetcher also reaches GitHub through it.
+  fetch:
+
 volumes:
   diffx-data:
+  diffx-mirrors:

docs/architecture.md

@@ -76,7 +76,12 @@ personal lifecycle discovery.
   REST reads, webhook signatures and payloads. No storage or diffx policy.
 - `githubsync`: GitHub integration rules: which accounts may follow which
   repositories, sealed user tokens, webhook deduplication, coalesced pull
-  request refresh intent and reconciliation. Remote composition only.
+  request refresh intent, reconciliation, fetch leases and server-authored
+  pull request observations. Remote composition only.
+- `fetchprotocol`: lease/result contract and client between `diffx-server`'s
+  internal listener and `diffx-fetcher`.
+- `fetcher`: `diffx-fetcher`'s partial-clone mirrors and offline collection of
+  leased pull request heads. It holds no account, GitHub App key or store.
 - `reviewstore`: SQLite transactions, identity persistence, stored reads,
   retention and queue adapter.
 - `ingestionqueue`: admission/lease contract and worker failure policy; jobs fail

docs/development.md

@@ -242,6 +242,13 @@ set `DIFFX_SERVER_INSECURE_COOKIES=false` behind production HTTPS. The fixed
 `diffx` project name keeps one
 volume regardless of the checkout directory.
 For other settings, use image env overrides or a Compose override file.
+The GitHub integration adds the `github` profile: set every
+`DIFFX_SERVER_GITHUB_*` variable and `DIFFX_SERVER_FETCHER_SECRET`, mount the
+app's private key in an override file, then run
+`docker compose --profile github up`. The `fetcher` service (`--target fetcher`,
+Alpine with Git 2.44+) keeps mirrors in the `diffx-mirrors` volume, capped by
+`DIFFX_FETCHER_DISK_LIMIT_GB` (default 20), and reaches the server's internal
+port 7979 only on the `fetch` network.
 Only one server may own a SQLite database; competing starts fail rather than
 share ownership. SIGTERM stops HTTP and workers before releasing storage;
 unfinished admitted jobs recover from the durable queue after leases expire.

docs/system.md

@@ -157,6 +157,30 @@ delivery. Resync's run identity makes the resulting submissions fresh, extending
 retention of pull requests idle since their last change; startup
 reconciliation and webhooks do not. There is no periodic poll.
 
+`diffx-fetcher` runs in its own container and turns refresh intent into
+snapshots. It leases one pull request at a time from the server's internal
+listener (`DIFFX_SERVER_INTERNAL_PORT`, default 7979, never published),
+authenticated by `DIFFX_SERVER_FETCHER_SECRET`; the public listener does not
+serve the protocol. Leasing reads the pull request with the app's installation
+credentials and hands out a token narrowed to that repository with
+`contents:read`; the fetcher never holds the app key, a user token or an
+account. A deleted pull request is recorded closed and a repository without
+active followers completes without a fetch. The fetcher keeps a partial-clone
+mirror per repository, fetches the leased head and base, and collects the
+branch comparison offline. A result is accepted only for the current lease
+token and only if it captures the leased head and base commits; failures retry
+with exponential backoff capped at an hour, and an expired lease is reclaimable.
+
+The server, not the fetcher, authors the observation: the `github` source and
+trigger, the repository key shared with HTTPS checkouts of the same remote, a
+checkout and branch identity per pull request, the pull request URL as root,
+`PR #N` as worktree, the head branch, GitHub's update time as collection and
+activity time, and the `pullRequest` metadata. Its submission ID hashes that
+payload and any Resync run, so refetching an unchanged pull request replays and
+a push, edit or state change is a fresh observation in the same stream. Every
+follower with verified access is admitted in one transaction with completing
+the lease; a refresh requested during the lease stays pending.
+
 ### Browse and review
 
 The comparison selector reads scopes of the open observation. It does not

internal/collector/branch_test.go

@@ -267,3 +267,24 @@ func TestStableFingerprintSurvivesCheckoutMoveAndBranchRename(t *testing.T) {
 		t.Fatalf("moved provenance: before %+v after %+v", request.Metadata, updated.Metadata)
 	}
 }
+
+func TestSuppliedIdentityCollectsWithoutEnrollment(t *testing.T) {
+	root := branchRepository(t)
+	git(t, root, "switch", "--detach", "HEAD~1")
+	options := branchOptions()
+	options.Branch, options.Base = "feature", "refs/heads/main"
+	options.Identity = &diffsource.StableRepositoryIdentity{RepositoryKey: "repository", CheckoutKey: "checkout", BranchID: "branch"}
+	request, err := Collect(t.Context(), root, options)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if request.Metadata.RepositoryKey != "repository" || request.Metadata.BranchID != "branch" || request.Metadata.CheckoutKey != diffsource.BranchCheckoutKey("repository", "branch") {
+		t.Fatalf("supplied identity ignored: %+v", request.Metadata)
+	}
+	if _, err := os.Stat(filepath.Join(root, ".git", "diffx")); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("collection enrolled identity in Git metadata: %v", err)
+	}
+	if config := git(t, root, "config", "--local", "--list"); strings.Contains(config, "diffx-id") {
+		t.Fatalf("collection wrote branch identity: %s", config)
+	}
+}

internal/collector/collector.go

@@ -36,6 +36,9 @@ type Options struct {
 	Base             string
 	Branch           string
 	OnSource         func(ingestion.Metadata, diffsource.Comparison)
+	// Identity supplies repository, checkout and branch identity instead of
+	// enrolling it in Git metadata, for producers that own identity elsewhere.
+	Identity *diffsource.StableRepositoryIdentity
 }
 
 func Collect(ctx context.Context, directory string, options Options) (ingestion.Request, error) {
@@ -392,6 +395,10 @@ func repositoryMetadata(ctx context.Context, root string, facts diffsource.Repos
 	metadata.RemoteURL = facts.RemoteURL
 	metadata.RepositoryName = filepath.Base(filepath.Dir(facts.CommonDir))
 	metadata.RepositoryName = diffsource.RemoteRepositoryName(facts.RemoteURL, metadata.RepositoryName)
+	if options.Identity != nil {
+		metadata.RepositoryKey, metadata.CheckoutKey, metadata.BranchID = options.Identity.RepositoryKey, options.Identity.CheckoutKey, options.Identity.BranchID
+		return metadata, nil
+	}
 	identity, err := diffsource.StableIdentity(ctx, root, options.SourceID, snapshot.Branch)
 	if err != nil {
 		return ingestion.Metadata{}, err

internal/config/fetcher.go

@@ -0,0 +1,52 @@
+package config
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"strconv"
+)
+
+// FetcherValues configures diffx-fetcher. DiskLimitBytes bounds its mirrors.
+type FetcherValues struct {
+	ServerURL      string
+	Secret         string
+	Mirrors        string
+	DiskLimitBytes int64
+	Concurrency    int
+}
+
+// LoadFetcher reads DIFFX_FETCHER_* environment only; the fetcher has no
+// config file. Diagnostics never echo the secret.
+func LoadFetcher() (FetcherValues, error) {
+	values := FetcherValues{ServerURL: os.Getenv("DIFFX_FETCHER_SERVER_URL"), Secret: os.Getenv("DIFFX_FETCHER_SECRET"), Mirrors: "data/mirrors", Concurrency: 2}
+	if values.ServerURL == "" {
+		return values, errors.New("DIFFX_FETCHER_SERVER_URL is required, such as http://server:7979")
+	}
+	if values.Secret == "" {
+		return values, errors.New("DIFFX_FETCHER_SECRET is required; use the server's DIFFX_SERVER_FETCHER_SECRET")
+	}
+	if mirrors, exists := os.LookupEnv("DIFFX_FETCHER_MIRRORS"); exists {
+		if mirrors == "" {
+			return values, errors.New("DIFFX_FETCHER_MIRRORS must be a directory path")
+		}
+		values.Mirrors = mirrors
+	}
+	gigabytes := int64(20)
+	if raw, exists := os.LookupEnv("DIFFX_FETCHER_DISK_LIMIT_GB"); exists {
+		value, err := strconv.ParseInt(raw, 10, 64)
+		if err != nil || value <= 0 || value > 1<<20 {
+			return values, fmt.Errorf("DIFFX_FETCHER_DISK_LIMIT_GB must be a positive whole number of GiB, such as 20")
+		}
+		gigabytes = value
+	}
+	values.DiskLimitBytes = gigabytes << 30
+	if raw, exists := os.LookupEnv("DIFFX_FETCHER_CONCURRENCY"); exists {
+		value, err := strconv.Atoi(raw)
+		if err != nil || value < 1 || value > 8 {
+			return values, fmt.Errorf("DIFFX_FETCHER_CONCURRENCY must be between 1 and 8")
+		}
+		values.Concurrency = value
+	}
+	return values, nil
+}

internal/config/fetcher_test.go

@@ -0,0 +1,59 @@
+package config
+
+import (
+	"os"
+	"strings"
+	"testing"
+)
+
+// setFetcherEnvironment leaves every fetcher variable absent except values.
+func setFetcherEnvironment(t *testing.T, values map[string]string) {
+	t.Helper()
+	for _, name := range []string{"DIFFX_FETCHER_SERVER_URL", "DIFFX_FETCHER_SECRET", "DIFFX_FETCHER_MIRRORS", "DIFFX_FETCHER_DISK_LIMIT_GB", "DIFFX_FETCHER_CONCURRENCY"} {
+		t.Setenv(name, "")
+		if err := os.Unsetenv(name); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for name, value := range values {
+		t.Setenv(name, value)
+	}
+}
+
+func TestFetcherConfigDefaultsAndOverrides(t *testing.T) {
+	setFetcherEnvironment(t, map[string]string{"DIFFX_FETCHER_SERVER_URL": "http://server:7979", "DIFFX_FETCHER_SECRET": "shh"})
+	values, err := LoadFetcher()
+	want := FetcherValues{ServerURL: "http://server:7979", Secret: "shh", Mirrors: "data/mirrors", DiskLimitBytes: 20 << 30, Concurrency: 2}
+	if err != nil || values != want {
+		t.Fatalf("defaults: %+v %v", values, err)
+	}
+	setFetcherEnvironment(t, map[string]string{"DIFFX_FETCHER_SERVER_URL": "http://server:7979", "DIFFX_FETCHER_SECRET": "shh",
+		"DIFFX_FETCHER_MIRRORS": "/mirrors", "DIFFX_FETCHER_DISK_LIMIT_GB": "3", "DIFFX_FETCHER_CONCURRENCY": "8"})
+	values, err = LoadFetcher()
+	want = FetcherValues{ServerURL: "http://server:7979", Secret: "shh", Mirrors: "/mirrors", DiskLimitBytes: 3 << 30, Concurrency: 8}
+	if err != nil || values != want {
+		t.Fatalf("overrides: %+v %v", values, err)
+	}
+}
+
+func TestFetcherConfigRejectsInvalidValuesWithoutEchoingSecret(t *testing.T) {
+	const secret = "fetcher-secret-value"
+	for _, tc := range []struct{ name, value string }{
+		{"DIFFX_FETCHER_SERVER_URL", ""},
+		{"DIFFX_FETCHER_SECRET", ""},
+		{"DIFFX_FETCHER_MIRRORS", ""},
+		{"DIFFX_FETCHER_DISK_LIMIT_GB", "0"},
+		{"DIFFX_FETCHER_DISK_LIMIT_GB", "1.5"},
+		{"DIFFX_FETCHER_DISK_LIMIT_GB", "9223372036854775807"},
+		{"DIFFX_FETCHER_CONCURRENCY", "0"},
+		{"DIFFX_FETCHER_CONCURRENCY", "9"},
+	} {
+		t.Run(tc.name+"="+tc.value, func(t *testing.T) {
+			setFetcherEnvironment(t, map[string]string{"DIFFX_FETCHER_SERVER_URL": "http://server:7979", "DIFFX_FETCHER_SECRET": secret, tc.name: tc.value})
+			_, err := LoadFetcher()
+			if err == nil || !strings.Contains(err.Error(), tc.name) || strings.Contains(err.Error(), secret) {
+				t.Fatalf("error %v, want mention of %s without secret", err, tc.name)
+			}
+		})
+	}
+}