From b7f48f172b3eb29510cebf2daf6211f62cef83fa Mon Sep 17 00:00:00 2001 From: Dinesh Pandiyan Date: Sun, 25 Oct 2026 15:15:14 +1100 Subject: [PATCH] feat: GitHub pull request fetcher with leased, repo-scoped fetches Server leases coalesced refresh intent on an internal listener with a shared secret and a contents:read installation token; diffx-fetcher mirrors with partial clones and collects offline; the server authors the observation and admits it to every verified follower atomically with lease completion. --- Dockerfile | 16 + cmd/diffx-fetcher/main.go | 58 ++++ cmd/diffx-server/main.go | 7 +- compose.yaml | 45 +++ docs/architecture.md | 7 +- docs/development.md | 7 + docs/system.md | 24 ++ internal/collector/branch_test.go | 21 ++ internal/collector/collector.go | 7 + internal/config/fetcher.go | 52 +++ internal/config/fetcher_test.go | 59 ++++ internal/config/server.go | 32 +- internal/config/server_test.go | 34 ++ internal/diffsource/stable_identity.go | 3 +- internal/fetcher/disk.go | 165 +++++++++ internal/fetcher/fetcher.go | 182 ++++++++++ internal/fetcher/fetcher_test.go | 455 +++++++++++++++++++++++++ internal/fetcher/git.go | 114 +++++++ internal/fetcher/integration_test.go | 132 +++++++ internal/fetcher/mirror.go | 264 ++++++++++++++ internal/fetcher/process_unix.go | 24 ++ internal/fetcher/process_windows.go | 12 + internal/fetchprotocol/protocol.go | 154 +++++++++ internal/githubsync/fetch.go | 244 +++++++++++++ internal/githubsync/fetch_test.go | 310 +++++++++++++++++ internal/githubsync/service.go | 4 +- internal/githubsync/service_test.go | 23 ++ internal/githubsync/store.go | 33 ++ internal/remoteserver/fetch.go | 66 ++++ internal/remoteserver/github.go | 3 + internal/remoteserver/github_test.go | 48 +++ internal/remoteserver/server.go | 31 +- internal/review/model_test.go | 8 + internal/review/repository_name.go | 10 + internal/reviewstore/github.go | 77 ++++- internal/reviewstore/queue.go | 24 +- tools/check-boundaries/main.go | 12 +- 37 files changed, 2741 insertions(+), 26 deletions(-) create mode 100644 cmd/diffx-fetcher/main.go create mode 100644 internal/config/fetcher.go create mode 100644 internal/config/fetcher_test.go create mode 100644 internal/fetcher/disk.go create mode 100644 internal/fetcher/fetcher.go create mode 100644 internal/fetcher/fetcher_test.go create mode 100644 internal/fetcher/git.go create mode 100644 internal/fetcher/integration_test.go create mode 100644 internal/fetcher/mirror.go create mode 100644 internal/fetcher/process_unix.go create mode 100644 internal/fetcher/process_windows.go create mode 100644 internal/fetchprotocol/protocol.go create mode 100644 internal/githubsync/fetch.go create mode 100644 internal/githubsync/fetch_test.go create mode 100644 internal/remoteserver/fetch.go diff --git a/Dockerfile b/Dockerfile index 100fdff2..5c8f5b89 100644 --- a/Dockerfile +++ b/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 diff --git a/cmd/diffx-fetcher/main.go b/cmd/diffx-fetcher/main.go new file mode 100644 index 00000000..998752f8 --- /dev/null +++ b/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(), + }) +} diff --git a/cmd/diffx-server/main.go b/cmd/diffx-server/main.go index 4cdb303b..93f73393 100644 --- a/cmd/diffx-server/main.go +++ b/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 } diff --git a/compose.yaml b/compose.yaml index f7982a74..589b9e1a 100644 --- a/compose.yaml +++ b/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: diff --git a/docs/architecture.md b/docs/architecture.md index 322a09fd..898f8be4 100644 --- a/docs/architecture.md +++ b/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 diff --git a/docs/development.md b/docs/development.md index 470e99c1..26e95f92 100644 --- a/docs/development.md +++ b/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. diff --git a/docs/system.md b/docs/system.md index aeaa7c78..913ea280 100644 --- a/docs/system.md +++ b/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 diff --git a/internal/collector/branch_test.go b/internal/collector/branch_test.go index 2ebdabfe..28a7c2aa 100644 --- a/internal/collector/branch_test.go +++ b/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) + } +} diff --git a/internal/collector/collector.go b/internal/collector/collector.go index 3dae9b3a..60fb90e3 100644 --- a/internal/collector/collector.go +++ b/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 diff --git a/internal/config/fetcher.go b/internal/config/fetcher.go new file mode 100644 index 00000000..8dd0d8a2 --- /dev/null +++ b/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 +} diff --git a/internal/config/fetcher_test.go b/internal/config/fetcher_test.go new file mode 100644 index 00000000..17747972 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/internal/config/server.go b/internal/config/server.go index c33f4d44..3feb801e 100644 --- a/internal/config/server.go +++ b/internal/config/server.go @@ -18,6 +18,9 @@ import ( // so both modes can run on one machine. const DefaultServerPort = 7980 +// DefaultInternalPort serves the GitHub fetcher protocol; never publish it. +const DefaultInternalPort = 7979 + // ServerValues is deployment configuration, independent of personal CLI settings. type ServerValues struct { Host string `json:"host"` @@ -27,7 +30,9 @@ type ServerValues struct { SignupEnabled bool `json:"signupEnabled"` InsecureCookies bool `json:"insecureCookies"` RetentionDays int `json:"retentionDays"` - GitHub GitHub `json:"github"` + // InternalPort listens beside Port, on the same host, when GitHub is enabled. + InternalPort int `json:"internalPort"` + GitHub GitHub `json:"github"` } // GitHub configures the optional GitHub App integration. Secrets come only from @@ -40,6 +45,8 @@ type GitHub struct { ClientSecret string `json:"-"` WebhookSecret string `json:"-"` TokenKey string `json:"-"` + // FetcherSecret authenticates diffx-fetcher on the internal listener. + FetcherSecret string `json:"-"` } // Enabled reports whether any GitHub setting is present; Validate requires all. @@ -65,6 +72,7 @@ func (github GitHub) Validate(insecureCookies bool) error { {"DIFFX_SERVER_GITHUB_CLIENT_SECRET", github.ClientSecret != ""}, {"DIFFX_SERVER_GITHUB_WEBHOOK_SECRET", github.WebhookSecret != ""}, {"DIFFX_SERVER_GITHUB_TOKEN_KEY", github.TokenKey != ""}, + {"DIFFX_SERVER_FETCHER_SECRET", github.FetcherSecret != ""}, } { if !field.present { missing = append(missing, field.name) @@ -82,6 +90,9 @@ func (github GitHub) Validate(insecureCookies bool) error { if err != nil || len(key) != 32 { return errors.New("DIFFX_SERVER_GITHUB_TOKEN_KEY must be 32 random bytes in standard base64, such as the output of: openssl rand -base64 32") } + if len(github.FetcherSecret) < 32 { + return errors.New("DIFFX_SERVER_FETCHER_SECRET must be at least 32 characters, such as the output of: openssl rand -base64 32") + } return nil } @@ -89,7 +100,7 @@ func (github GitHub) Validate(insecureCookies bool) error { // It never discovers personal config, creates files, or acquires a config lock. // Variables use the DIFFX_SERVER_ prefix so they never configure the local CLI. func LoadServerFile(path string) (ServerValues, error) { - values := ServerValues{Host: "127.0.0.1", Port: DefaultServerPort, State: "data/state.db", RetentionDays: 7} + values := ServerValues{Host: "127.0.0.1", Port: DefaultServerPort, InternalPort: DefaultInternalPort, State: "data/state.db", RetentionDays: 7} if path == "" { path = os.Getenv("DIFFX_SERVER_CONFIG") } @@ -124,6 +135,7 @@ func LoadServerFile(path string) (ServerValues, error) { {"DIFFX_SERVER_GITHUB_CLIENT_SECRET", &values.GitHub.ClientSecret}, {"DIFFX_SERVER_GITHUB_WEBHOOK_SECRET", &values.GitHub.WebhookSecret}, {"DIFFX_SERVER_GITHUB_TOKEN_KEY", &values.GitHub.TokenKey}, + {"DIFFX_SERVER_FETCHER_SECRET", &values.GitHub.FetcherSecret}, } { if value, exists := os.LookupEnv(field.name); exists { *field.target = value @@ -134,6 +146,7 @@ func LoadServerFile(path string) (ServerValues, error) { target *int }{ {"DIFFX_SERVER_PORT", &values.Port}, + {"DIFFX_SERVER_INTERNAL_PORT", &values.InternalPort}, {"DIFFX_SERVER_RETENTION_DAYS", &values.RetentionDays}, } { if raw, exists := os.LookupEnv(field.name); exists { @@ -144,7 +157,8 @@ func LoadServerFile(path string) (ServerValues, error) { *field.target = value } } - if raw, exists := os.LookupEnv("DIFFX_SERVER_GITHUB_APP_ID"); exists { + // Compose passes unset integration variables through as empty. + if raw := os.Getenv("DIFFX_SERVER_GITHUB_APP_ID"); raw != "" { value, err := strconv.ParseInt(raw, 10, 64) if err != nil { return values, fmt.Errorf("invalid DIFFX_SERVER_GITHUB_APP_ID: %w", err) @@ -179,3 +193,15 @@ func (values ServerValues) ListenAddress() (string, error) { } return net.JoinHostPort(values.Host, strconv.Itoa(values.Port)), nil } + +// InternalAddress places the internal listener on the public listener's host. +func (values ServerValues) InternalAddress(listen string) (string, error) { + host, port, err := net.SplitHostPort(listen) + if err != nil { + return "", fmt.Errorf("invalid listen address: %w", err) + } + if values.InternalPort <= 0 || values.InternalPort > 65535 || strconv.Itoa(values.InternalPort) == port { + return "", errors.New("server internal port must be between 1 and 65535 and differ from the server port") + } + return net.JoinHostPort(host, strconv.Itoa(values.InternalPort)), nil +} diff --git a/internal/config/server_test.go b/internal/config/server_test.go index 650fe5f0..d5951153 100644 --- a/internal/config/server_test.go +++ b/internal/config/server_test.go @@ -95,6 +95,7 @@ func TestGitHubConfigIsAllOrNothingWithSecretsFromEnvironment(t *testing.T) { "DIFFX_SERVER_GITHUB_APP_ID": "12", "DIFFX_SERVER_GITHUB_CLIENT_ID": "client", "DIFFX_SERVER_GITHUB_PRIVATE_KEY_FILE": "/keys/app.pem", "DIFFX_SERVER_GITHUB_PUBLIC_URL": "https://diffx.example.com", "DIFFX_SERVER_GITHUB_CLIENT_SECRET": "secret", "DIFFX_SERVER_GITHUB_WEBHOOK_SECRET": "hook", "DIFFX_SERVER_GITHUB_TOKEN_KEY": key, + "DIFFX_SERVER_FETCHER_SECRET": strings.Repeat("f", 32), } for name, value := range complete { t.Setenv(name, value) @@ -108,6 +109,8 @@ func TestGitHubConfigIsAllOrNothingWithSecretsFromEnvironment(t *testing.T) { {"DIFFX_SERVER_GITHUB_PUBLIC_URL", "http://diffx.example.com", "https origin"}, {"DIFFX_SERVER_GITHUB_PUBLIC_URL", "https://diffx.example.com/app", "https origin"}, {"DIFFX_SERVER_GITHUB_TOKEN_KEY", "short", "32 random bytes"}, + {"DIFFX_SERVER_FETCHER_SECRET", "", "DIFFX_SERVER_FETCHER_SECRET"}, + {"DIFFX_SERVER_FETCHER_SECRET", "short", "at least 32 characters"}, } { t.Run(tc.name+"="+tc.value, func(t *testing.T) { t.Setenv(tc.name, tc.value) @@ -136,3 +139,34 @@ func TestGitHubConfigIsAllOrNothingWithSecretsFromEnvironment(t *testing.T) { t.Fatal("absent integration must be valid and disabled") } } + +func TestInternalListenerSharesHostOnSeparatePort(t *testing.T) { + t.Setenv("DIFFX_SERVER_CONFIG", "") + values, err := LoadServerFile("") + if err != nil || values.InternalPort != DefaultInternalPort { + t.Fatalf("default internal port: %+v %v", values, err) + } + if address, err := values.InternalAddress("0.0.0.0:7980"); err != nil || address != "0.0.0.0:7979" { + t.Fatalf("internal address %q %v", address, err) + } + t.Setenv("DIFFX_SERVER_INTERNAL_PORT", "7980") + values, err = LoadServerFile("") + if err != nil { + t.Fatal(err) + } + if _, err := values.InternalAddress("0.0.0.0:7980"); err == nil { + t.Fatal("internal listener accepted the public port") + } +} + +func TestEmptyGitHubEnvironmentDisablesIntegration(t *testing.T) { + t.Setenv("DIFFX_SERVER_CONFIG", "") + for _, name := range []string{"APP_ID", "CLIENT_ID", "PRIVATE_KEY_FILE", "PUBLIC_URL", "CLIENT_SECRET", "WEBHOOK_SECRET", "TOKEN_KEY"} { + t.Setenv("DIFFX_SERVER_GITHUB_"+name, "") + } + t.Setenv("DIFFX_SERVER_FETCHER_SECRET", "") + values, err := LoadServerFile("") + if err != nil || values.GitHub.Enabled() || values.GitHub.Validate(false) != nil { + t.Fatalf("empty passthrough must disable GitHub: %+v %v", values.GitHub, err) + } +} diff --git a/internal/diffsource/stable_identity.go b/internal/diffsource/stable_identity.go index 34d4e475..20e17e79 100644 --- a/internal/diffsource/stable_identity.go +++ b/internal/diffsource/stable_identity.go @@ -14,6 +14,7 @@ import ( "time" "github.com/flexdinesh/diffx/internal/processlock" + "github.com/flexdinesh/diffx/internal/review" "github.com/google/uuid" ) @@ -48,7 +49,7 @@ func StableIdentity(ctx context.Context, root, sourceID, branch string) (StableR defer lock.Close() repositoryKey := identityHash("repository", sourceID, facts.CommonDir) if facts.RemoteURL != "" { - repositoryKey = identityHash("remote", facts.RemoteURL) + repositoryKey = review.RemoteRepositoryKey(facts.RemoteURL) } repositoryPath := filepath.Join(directory, "repository.json") repository, err := readSavedIdentity(repositoryPath, sourceID, repositoryKey) diff --git a/internal/fetcher/disk.go b/internal/fetcher/disk.go new file mode 100644 index 00000000..26618027 --- /dev/null +++ b/internal/fetcher/disk.go @@ -0,0 +1,165 @@ +package fetcher + +import ( + "errors" + "io/fs" + "log" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "sync" + "time" +) + +const ( + stagingPrefix = ".init-" + evictingPrefix = ".evict-" +) + +// repositoryLocks serializes work per repository ID; eviction skips held IDs. +type repositoryLocks struct { + mu sync.Mutex + held map[int64]*repositoryLock +} + +type repositoryLock struct { + sync.Mutex + users int +} + +func (locks *repositoryLocks) lock(id int64) func() { + locks.mu.Lock() + entry := locks.held[id] + if entry == nil { + entry = &repositoryLock{} + locks.held[id] = entry + } + entry.users++ + locks.mu.Unlock() + entry.Lock() + return func() { locks.release(id, entry) } +} + +func (locks *repositoryLocks) tryLock(id int64) (func(), bool) { + locks.mu.Lock() + defer locks.mu.Unlock() + if locks.held[id] != nil { + return nil, false + } + entry := &repositoryLock{users: 1} + entry.Lock() + locks.held[id] = entry + return func() { locks.release(id, entry) }, true +} + +func (locks *repositoryLocks) release(id int64, entry *repositoryLock) { + entry.Unlock() + locks.mu.Lock() + defer locks.mu.Unlock() + if entry.users--; entry.users == 0 { + delete(locks.held, id) + } +} + +// prepareMirrors removes staging and eviction leftovers from an interrupted run. +func prepareMirrors(root string) error { + if err := os.MkdirAll(root, 0o700); err != nil { + return err + } + entries, err := os.ReadDir(root) + if err != nil { + return err + } + for _, entry := range entries { + if strings.HasPrefix(entry.Name(), stagingPrefix) || strings.HasPrefix(entry.Name(), evictingPrefix) { + if err := os.RemoveAll(filepath.Join(root, entry.Name())); err != nil { + return err + } + } + } + return nil +} + +type mirrorUsage struct { + id int64 + size int64 + used time.Time +} + +// enforceDiskLimit removes least recently used mirrors until the directory +// fits its limit. Mirrors in use are kept; they are reconsidered after their job. +func (fetcher *fetcher) enforceDiskLimit() { + fetcher.evicting.Lock() + defer fetcher.evicting.Unlock() + root := fetcher.settings.Mirrors + entries, err := os.ReadDir(root) + if err != nil { + log.Printf("diffx-fetcher: read mirrors: %v", err) + return + } + var total int64 + var mirrors []mirrorUsage + for _, entry := range entries { + size := directorySize(filepath.Join(root, entry.Name())) + total += size + id, err := strconv.ParseInt(entry.Name(), 10, 64) + if err != nil || !entry.IsDir() { + continue + } + usage := mirrorUsage{id: id, size: size} + if info, err := os.Stat(filepath.Join(root, entry.Name(), ".git", usedMarker)); err == nil { + usage.used = info.ModTime() + } + mirrors = append(mirrors, usage) + } + sort.Slice(mirrors, func(left, right int) bool { return mirrors[left].used.Before(mirrors[right].used) }) + for _, mirror := range mirrors { + if total <= fetcher.settings.DiskLimit { + return + } + release, ok := fetcher.locks.tryLock(mirror.id) + if !ok { + continue + } + err := evict(root, mirror.id) + release() + if err != nil { + log.Printf("diffx-fetcher: evict mirror %d: %v", mirror.id, err) + continue + } + total -= mirror.size + } + if total > fetcher.settings.DiskLimit { + log.Printf("diffx-fetcher: mirrors use %d bytes, over the %d byte limit, while in use", total, fetcher.settings.DiskLimit) + } +} + +// evict renames before deleting so a partial removal is never reused. +func evict(root string, id int64) error { + name := strconv.FormatInt(id, 10) + target := filepath.Join(root, evictingPrefix+name+"-"+strconv.FormatInt(time.Now().UnixNano(), 10)) + if err := os.Rename(filepath.Join(root, name), target); err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return err + } + return os.RemoveAll(target) +} + +// directorySize skips entries that vanish while a concurrent job writes. +func directorySize(path string) int64 { + var size int64 + _ = filepath.WalkDir(path, func(_ string, entry fs.DirEntry, err error) error { + if err != nil || !entry.Type().IsRegular() { + return nil + } + if info, err := entry.Info(); err == nil { + size += info.Size() + } + return nil + }) + return size +} diff --git a/internal/fetcher/fetcher.go b/internal/fetcher/fetcher.go new file mode 100644 index 00000000..25b3d18a --- /dev/null +++ b/internal/fetcher/fetcher.go @@ -0,0 +1,182 @@ +// Package fetcher mirrors leased GitHub pull requests and collects them offline +// with the shared collector. It never claims an account or pull request +// metadata; the server overwrites identity and validates the captured commits. +package fetcher + +import ( + "context" + "errors" + "fmt" + "log" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/flexdinesh/diffx/internal/fetchprotocol" + "github.com/flexdinesh/diffx/internal/ingestion" +) + +type Settings struct { + Mirrors string + DiskLimit int64 + Concurrency int + // PollInterval is the idle wait between empty leases; zero means 5s. + PollInterval time.Duration + CollectorVersion string +} + +// Server is the fetcher's view of the server's internal fetch listener. +type Server interface { + Lease(context.Context) (*fetchprotocol.Job, error) + Complete(context.Context, fetchprotocol.Job, ingestion.Request) error + Fail(context.Context, fetchprotocol.Job, string) error +} + +const maxLeaseBackoff = time.Minute + +// Run leases and collects pull requests on Concurrency workers until ctx is +// cancelled, then joins every worker and returns nil. Collection inherits the +// process environment, so the caller must export GIT_NO_LAZY_FETCH=1; Run +// refuses to start otherwise rather than mutating process-wide state. +func Run(ctx context.Context, server Server, settings Settings) error { + return run(ctx, server, settings, github) +} + +func run(ctx context.Context, server Server, settings Settings, remote remote) error { + if os.Getenv("GIT_NO_LAZY_FETCH") != "1" { + return errors.New("GIT_NO_LAZY_FETCH=1 must be set so collection never fetches missing objects") + } + if settings.Mirrors == "" || settings.DiskLimit <= 0 || settings.Concurrency < 1 { + return errors.New("fetcher requires a mirror directory, a positive disk limit and at least one worker") + } + if settings.PollInterval <= 0 { + settings.PollInterval = 5 * time.Second + } + root, err := filepath.Abs(settings.Mirrors) + if err != nil { + return err + } + if err := requireGit(ctx); err != nil { + return err + } + if err := prepareMirrors(root); err != nil { + return err + } + settings.Mirrors = root + fetcher := &fetcher{server: server, settings: settings, remote: remote, locks: repositoryLocks{held: make(map[int64]*repositoryLock)}} + var workers sync.WaitGroup + for range settings.Concurrency { + workers.Go(func() { fetcher.work(ctx) }) + } + workers.Wait() + return nil +} + +type fetcher struct { + server Server + settings Settings + remote remote + locks repositoryLocks + evicting sync.Mutex +} + +func (fetcher *fetcher) work(ctx context.Context) { + delay := fetcher.settings.PollInterval + for ctx.Err() == nil { + job, err := fetcher.server.Lease(ctx) + if err != nil { + if ctx.Err() != nil { + return + } + log.Printf("diffx-fetcher: lease failed; retrying in %s: %v", delay, err) + wait(ctx, delay) + delay = min(2*delay, max(maxLeaseBackoff, fetcher.settings.PollInterval)) + continue + } + delay = fetcher.settings.PollInterval + if job == nil { + wait(ctx, delay) + continue + } + fetcher.handle(ctx, *job) + } +} + +func wait(ctx context.Context, delay time.Duration) { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + case <-timer.C: + } +} + +// handle reports exactly one result per lease unless shutdown abandons it; +// an abandoned lease expires and the server leases it again. +func (fetcher *fetcher) handle(ctx context.Context, job fetchprotocol.Job) { + request, err := fetcher.collect(ctx, job) + defer fetcher.enforceDiskLimit() + if ctx.Err() != nil { + log.Printf("diffx-fetcher: abandoned lease %s on shutdown", job.ID) + return + } + if err != nil { + detail := redact(err.Error(), job.Token) + log.Printf("diffx-fetcher: repository %d pull %d failed: %s", job.Repository.ID, job.Number, detail) + fetcher.reported(job, fetcher.server.Fail(ctx, job, detail)) + return + } + fetcher.reported(job, fetcher.server.Complete(ctx, job, request)) +} + +func (fetcher *fetcher) reported(job fetchprotocol.Job, err error) { + var status *fetchprotocol.StatusError + switch { + case err == nil: + case errors.As(err, &status) && status.Status == 409: + log.Printf("diffx-fetcher: lease %s lost; dropped result", job.ID) + default: + log.Printf("diffx-fetcher: report lease %s: %s", job.ID, redact(err.Error(), job.Token)) + } +} + +// errMoved means GitHub no longer serves the leased commits; the server reads +// the pull request again before its next lease. +var errMoved = errors.New("pull request moved; refetch") + +func (fetcher *fetcher) collect(ctx context.Context, job fetchprotocol.Job) (ingestion.Request, error) { + if err := validateJob(job); err != nil { + return ingestion.Request{}, err + } + url, err := fetcher.remote.url(job.Repository) + if err != nil { + return ingestion.Request{}, err + } + release := fetcher.locks.lock(job.Repository.ID) + defer release() + mirror, err := fetcher.open(ctx, job.Repository.ID, url, job.Token) + if err != nil { + return ingestion.Request{}, fmt.Errorf("prepare mirror: %w", err) + } + if err := mirror.fetch(ctx, job); err != nil { + return ingestion.Request{}, err + } + if err := mirror.verify(ctx, job); err != nil { + return ingestion.Request{}, err + } + if err := mirror.prefetch(ctx, job); err != nil { + return ingestion.Request{}, err + } + return mirror.collect(ctx, job, fetcher.settings.CollectorVersion) +} + +// redact removes the installation token and its encoded header from text +// that may reach logs or the server. +func redact(text, token string) string { + if token == "" { + return text + } + return strings.NewReplacer(token, "[redacted]", basicCredential(token), "[redacted]").Replace(text) +} diff --git a/internal/fetcher/fetcher_test.go b/internal/fetcher/fetcher_test.go new file mode 100644 index 00000000..6f883660 --- /dev/null +++ b/internal/fetcher/fetcher_test.go @@ -0,0 +1,455 @@ +package fetcher + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/flexdinesh/diffx/internal/fetchprotocol" + "github.com/flexdinesh/diffx/internal/ingestion" + "github.com/flexdinesh/diffx/internal/review" +) + +const testToken = "ghs_installation-token-value" + +func requireTestGit(t *testing.T) { + t.Helper() + raw, err := exec.Command("git", "version").Output() + if err != nil || !gitAtLeast(string(raw), 2, 44) { + t.Skipf("fetcher requires git 2.44 or later: %s %v", raw, err) + } + t.Setenv("GIT_NO_LAZY_FETCH", "1") +} + +func git(t *testing.T, directory string, args ...string) string { + t.Helper() + command := exec.Command("git", args...) + command.Dir = directory + command.Env = append(os.Environ(), "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL="+os.DevNull, "GIT_NO_LAZY_FETCH=1") + raw, err := command.CombinedOutput() + if err != nil { + t.Fatalf("git %v: %v: %s", args, err, raw) + } + return strings.TrimSpace(string(raw)) +} + +func write(t *testing.T, root, name, value string) { + t.Helper() + if err := os.WriteFile(filepath.Join(root, name), []byte(value), 0o600); err != nil { + t.Fatal(err) + } +} + +type origin struct { + bare, work string + base, head string + // unchanged is a blob outside the pull request diff. + unchanged string +} + +// newOrigin publishes main and pull request 7, whose base advanced after the +// pull request branched so merge-base and base tip differ. +func newOrigin(t *testing.T) origin { + t.Helper() + work := t.TempDir() + git(t, work, "init", "--quiet", "--initial-branch=main") + git(t, work, "config", "user.name", "Test") + git(t, work, "config", "user.email", "test@example.com") + write(t, work, "stable.txt", "unchanged\n") + write(t, work, "edited.txt", "before\n") + write(t, work, "removed.txt", "gone soon\n") + git(t, work, "add", ".") + git(t, work, "commit", "--quiet", "-m", "initial") + git(t, work, "switch", "--quiet", "-c", "feature") + write(t, work, "edited.txt", "after\n") + write(t, work, "added.txt", "new file\n") + git(t, work, "rm", "--quiet", "removed.txt") + git(t, work, "add", ".") + git(t, work, "commit", "--quiet", "-m", "feature") + git(t, work, "switch", "--quiet", "main") + write(t, work, "base-only.txt", "not in the pull request\n") + git(t, work, "add", ".") + git(t, work, "commit", "--quiet", "-m", "base advances") + bare := filepath.Join(t.TempDir(), "origin.git") + git(t, work, "init", "--quiet", "--bare", bare) + git(t, bare, "config", "uploadpack.allowFilter", "true") + git(t, bare, "config", "uploadpack.allowAnySHA1InWant", "true") + git(t, work, "push", "--quiet", bare, "main:refs/heads/main", "feature:refs/pull/7/head") + return origin{bare: bare, work: work, base: git(t, work, "rev-parse", "main"), head: git(t, work, "rev-parse", "feature"), unchanged: git(t, work, "rev-parse", "main:stable.txt")} +} + +func (origin origin) job(repository int64) fetchprotocol.Job { + return fetchprotocol.Job{ID: "job", LeaseToken: "lease", Repository: fetchprotocol.Repository{ID: repository, FullName: "octo/repo", CloneURL: "https://github.com/octo/repo.git"}, + Number: 7, HeadSHA: origin.head, BaseRef: "main", BaseSHA: origin.base, Token: testToken} +} + +// local substitutes a file origin; production permits only GitHub HTTPS. +func (origin origin) remote() remote { + return remote{protocol: "file", url: func(fetchprotocol.Repository) (string, error) { return "file://" + origin.bare, nil }} +} + +type fakeServer struct { + mu sync.Mutex + jobs []fetchprotocol.Job + completed []ingestion.Request + failed []string + reported func() +} + +func (server *fakeServer) Lease(context.Context) (*fetchprotocol.Job, error) { + server.mu.Lock() + defer server.mu.Unlock() + if len(server.jobs) == 0 { + return nil, nil + } + job := server.jobs[0] + server.jobs = server.jobs[1:] + return &job, nil +} + +func (server *fakeServer) Complete(_ context.Context, _ fetchprotocol.Job, request ingestion.Request) error { + server.mu.Lock() + defer server.mu.Unlock() + server.completed = append(server.completed, request) + if server.reported != nil { + server.reported() + } + return nil +} + +func (server *fakeServer) Fail(_ context.Context, _ fetchprotocol.Job, detail string) error { + server.mu.Lock() + defer server.mu.Unlock() + server.failed = append(server.failed, detail) + if server.reported != nil { + server.reported() + } + return nil +} + +func newTestFetcher(t *testing.T, server Server, origin origin, limit int64) *fetcher { + t.Helper() + root := t.TempDir() + return &fetcher{server: server, settings: Settings{Mirrors: root, DiskLimit: limit, Concurrency: 1, PollInterval: time.Millisecond, CollectorVersion: "test"}, + remote: origin.remote(), locks: repositoryLocks{held: make(map[int64]*repositoryLock)}} +} + +func TestRunCollectsPullRequestDiffFromMergeBaseWithContents(t *testing.T) { + requireTestGit(t) + origin := newOrigin(t) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + server := &fakeServer{jobs: []fetchprotocol.Job{origin.job(42)}, reported: cancel} + mirrors := t.TempDir() + if err := run(ctx, server, Settings{Mirrors: mirrors, DiskLimit: 1 << 30, Concurrency: 2, PollInterval: time.Millisecond, CollectorVersion: "test"}, origin.remote()); err != nil { + t.Fatal(err) + } + if len(server.failed) != 0 || len(server.completed) != 1 { + t.Fatalf("completed %d failed %v", len(server.completed), server.failed) + } + request := server.completed[0] + if len(request.Scopes) != 1 || request.Scopes[0].Snapshot.Mode != review.DiffAll { + t.Fatalf("objects-only collection must capture only the all scope: %+v", request.Scopes) + } + if head := request.Metadata.Head; head == nil || *head != origin.head || request.Metadata.Comparison == nil || request.Metadata.Comparison.BaseCommit != origin.base { + t.Fatalf("captured commits: %+v %+v", request.Metadata.Head, request.Metadata.Comparison) + } + scope := request.Scopes[0] + contents := map[string]review.FileContents{} + for _, file := range scope.Snapshot.Files { + patch := scope.Patches[file.ID] + if patch.Contents == nil { + t.Fatalf("%s has no contents: %+v", file.Path, patch) + } + contents[file.Path+":"+file.Status] = *patch.Contents + } + want := map[string]review.FileContents{ + "edited.txt:M": {Before: "before\n", After: "after\n"}, "added.txt:A": {After: "new file\n"}, "removed.txt:D": {Before: "gone soon\n"}, + } + if len(contents) != len(want) { + t.Fatalf("pull request files %v, want %v (base-only changes excluded)", contents, want) + } + for key, value := range want { + if contents[key] != value { + t.Fatalf("%s contents %+v, want %+v", key, contents[key], value) + } + } + mirror := filepath.Join(mirrors, "42") + if missing := catFileMissing(t, mirror, origin.unchanged); !missing { + t.Fatal("mirror downloaded a blob outside the pull request diff") + } + if git(t, mirror, "rev-parse", "--is-bare-repository") != "false" { + t.Fatal("mirror must be a non-bare repository for collection") + } + assertNoToken(t, mirrors) +} + +func catFileMissing(t *testing.T, repository, object string) bool { + t.Helper() + command := exec.Command("git", "cat-file", "--batch-check") + command.Dir = repository + command.Env = append(os.Environ(), "GIT_NO_LAZY_FETCH=1") + command.Stdin = strings.NewReader(object + "\n") + raw, err := command.Output() + if err != nil { + t.Fatal(err) + } + return strings.HasSuffix(strings.TrimSpace(string(raw)), " missing") +} + +func assertNoToken(t *testing.T, root string) { + t.Helper() + err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { + if err != nil || !entry.Type().IsRegular() { + return err + } + raw, err := os.ReadFile(path) + if err != nil { + return err + } + if strings.Contains(string(raw), testToken) || strings.Contains(string(raw), basicCredential(testToken)) { + t.Errorf("token persisted in %s", path) + } + return nil + }) + if err != nil { + t.Fatal(err) + } +} + +func TestMovedHeadFailsForRefetch(t *testing.T) { + requireTestGit(t) + origin := newOrigin(t) + server := &fakeServer{} + fetcher := newTestFetcher(t, server, origin, 1<<30) + job := origin.job(1) + job.HeadSHA = origin.base + fetcher.handle(t.Context(), job) + if len(server.completed) != 0 || len(server.failed) != 1 || server.failed[0] != errMoved.Error() { + t.Fatalf("completed %d failed %v", len(server.completed), server.failed) + } +} + +func TestFailuresNeverExposeToken(t *testing.T) { + requireTestGit(t) + origin := newOrigin(t) + server := &fakeServer{} + fetcher := newTestFetcher(t, server, origin, 1<<30) + job := origin.job(1) + job.Number = 8 + fetcher.handle(t.Context(), job) + if len(server.failed) != 1 || !strings.Contains(server.failed[0], "fetch pull request") { + t.Fatalf("missing pull request ref: %v", server.failed) + } + assertNoToken(t, fetcher.settings.Mirrors) + if got := redact("header Authorization: Basic "+basicCredential(testToken)+" token "+testToken, testToken); strings.Contains(got, testToken) || strings.Contains(got, basicCredential(testToken)) { + t.Fatalf("redaction leaked credentials: %s", got) + } +} + +func TestSecondJobReusesMirror(t *testing.T) { + requireTestGit(t) + origin := newOrigin(t) + server := &fakeServer{} + fetcher := newTestFetcher(t, server, origin, 1<<30) + fetcher.handle(t.Context(), origin.job(5)) + sentinel := filepath.Join(fetcher.settings.Mirrors, "5", ".git", "sentinel") + write(t, filepath.Dir(sentinel), "sentinel", "kept") + git(t, origin.work, "switch", "--quiet", "feature") + write(t, origin.work, "edited.txt", "after again\n") + git(t, origin.work, "commit", "--quiet", "-am", "follow-up") + git(t, origin.work, "push", "--quiet", "--force", origin.bare, "feature:refs/pull/7/head") + job := origin.job(5) + job.HeadSHA = git(t, origin.work, "rev-parse", "feature") + fetcher.handle(t.Context(), job) + if len(server.failed) != 0 || len(server.completed) != 2 { + t.Fatalf("completed %d failed %v", len(server.completed), server.failed) + } + if _, err := os.Stat(sentinel); err != nil { + t.Fatalf("second job recreated the mirror: %v", err) + } + if head := server.completed[1].Metadata.Head; head == nil || *head != job.HeadSHA { + t.Fatalf("second job captured %v, want %s", head, job.HeadSHA) + } +} + +func TestCollectionNeverFetchesMissingObjects(t *testing.T) { + requireTestGit(t) + origin := newOrigin(t) + fetcher := newTestFetcher(t, &fakeServer{}, origin, 1<<30) + job := origin.job(3) + mirror, err := fetcher.open(t.Context(), 3, "file://"+origin.bare, job.Token) + if err != nil { + t.Fatal(err) + } + if err := mirror.fetch(t.Context(), job); err != nil { + t.Fatal(err) + } + // Without prefetch, changed blobs are absent; collection must fail rather + // than fetch them from the reachable origin. + if _, err := mirror.collect(t.Context(), job, "test"); err == nil || !strings.Contains(err.Error(), "lazy fetching disabled") { + t.Fatalf("collection without the pull request's blobs: %v", err) + } + if !catFileMissing(t, mirror.dir, git(t, origin.work, "rev-parse", "feature:added.txt")) { + t.Fatal("collection fetched a missing blob") + } +} + +func TestDiskLimitEvictsLeastRecentlyUsedIdleMirrors(t *testing.T) { + requireTestGit(t) + origin := newOrigin(t) + server := &fakeServer{} + fetcher := newTestFetcher(t, server, origin, 1<<40) + for _, id := range []int64{1, 2, 3} { + fetcher.handle(t.Context(), origin.job(id)) + } + if len(server.completed) != 3 { + t.Fatalf("setup failed: %v", server.failed) + } + root := fetcher.settings.Mirrors + now := time.Now() + for id, age := range map[string]time.Duration{"1": 3 * time.Hour, "2": time.Hour, "3": 2 * time.Hour} { + marker := filepath.Join(root, id, ".git", usedMarker) + if err := os.Chtimes(marker, now.Add(-age), now.Add(-age)); err != nil { + t.Fatal(err) + } + } + fetcher.settings.DiskLimit = directorySize(root) - directorySize(filepath.Join(root, "1")) + fetcher.enforceDiskLimit() + assertMirrors(t, root, "2", "3") + release := fetcher.locks.lock(2) + fetcher.settings.DiskLimit = 1 + fetcher.enforceDiskLimit() + release() + assertMirrors(t, root, "2") +} + +func assertMirrors(t *testing.T, root string, want ...string) { + t.Helper() + entries, err := os.ReadDir(root) + if err != nil { + t.Fatal(err) + } + var names []string + for _, entry := range entries { + names = append(names, entry.Name()) + } + if strings.Join(names, ",") != strings.Join(want, ",") { + t.Fatalf("mirrors %v, want %v", names, want) + } +} + +type blockingServer struct { + fakeServer + mu sync.Mutex + active int + calls int +} + +func (server *blockingServer) Lease(ctx context.Context) (*fetchprotocol.Job, error) { + server.mu.Lock() + server.active++ + server.calls++ + server.mu.Unlock() + <-ctx.Done() + time.Sleep(10 * time.Millisecond) + server.mu.Lock() + server.active-- + server.mu.Unlock() + return nil, ctx.Err() +} + +func TestRunJoinsWorkersOnCancel(t *testing.T) { + requireTestGit(t) + server := &blockingServer{} + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { + done <- run(ctx, server, Settings{Mirrors: t.TempDir(), DiskLimit: 1, Concurrency: 3}, github) + }() + deadline := time.Now().Add(5 * time.Second) + for { + server.mu.Lock() + calls := server.calls + server.mu.Unlock() + if calls == 3 { + break + } + if time.Now().After(deadline) { + t.Fatalf("%d of 3 workers leased", calls) + } + time.Sleep(time.Millisecond) + } + cancel() + if err := <-done; err != nil { + t.Fatal(err) + } + server.mu.Lock() + defer server.mu.Unlock() + if server.active != 0 { + t.Fatalf("Run returned with %d workers still leasing", server.active) + } +} + +func TestRunRequiresLazyFetchDisabled(t *testing.T) { + t.Setenv("GIT_NO_LAZY_FETCH", "") + err := Run(t.Context(), &fakeServer{}, Settings{Mirrors: t.TempDir(), DiskLimit: 1, Concurrency: 1}) + if err == nil || !strings.Contains(err.Error(), "GIT_NO_LAZY_FETCH") { + t.Fatalf("Run started with lazy fetch enabled: %v", err) + } +} + +func TestGitHubRemoteAcceptsOnlyLeasedHTTPSRepository(t *testing.T) { + valid := fetchprotocol.Repository{ID: 1, FullName: "octo-org/repo.name_1", CloneURL: "https://github.com/octo-org/repo.name_1.git"} + if url, err := githubURL(valid); err != nil || url != valid.CloneURL { + t.Fatalf("valid repository: %q %v", url, err) + } + for _, clone := range []string{ + "http://github.com/octo-org/repo.name_1.git", "https://github.com/other/repo.git", "https://x@github.com/octo-org/repo.name_1.git", + "https://github.com.evil/octo-org/repo.name_1.git", "file:///octo-org/repo.name_1.git", "https://github.com/octo-org/repo.name_1", + "ext::sh -c touch% /tmp/pwned", "https://github.com/octo-org/repo.name_1.git?x=1", + } { + repository := valid + repository.CloneURL = clone + if _, err := githubURL(repository); err == nil { + t.Errorf("accepted clone URL %q", clone) + } + } + for _, name := range []string{"octo/..", "octo/.", "../repo", "octo/repo/extra", "-octo/repo", "octo/re po", "octo/repo.git"} { + repository := fetchprotocol.Repository{ID: 1, FullName: name, CloneURL: "https://github.com/" + name + ".git"} + if _, err := githubURL(repository); err == nil { + t.Errorf("accepted repository name %q", name) + } + } +} + +func TestBaseRefRejectsRefspecAndOptionSyntax(t *testing.T) { + for _, name := range []string{"main", "release/1.2", "feature/ΓΌ-name", "v1.0-rc"} { + if !validBranch(name) { + t.Errorf("rejected branch %q", name) + } + } + for _, name := range []string{"", "-main", "a..b", "a:b", "main*", "a/.hidden", "main.lock", "main/", "a//b", "a@{1}", "@", "a b", "a\nb", "a~1", "a^", "a\\b", "main."} { + if validBranch(name) { + t.Errorf("accepted branch %q", name) + } + } +} + +func TestChangedBlobsSkipsSubmodulesAndAbsentSides(t *testing.T) { + zero := strings.Repeat("0", 40) + a, b, c, d := strings.Repeat("a", 40), strings.Repeat("b", 40), strings.Repeat("c", 40), strings.Repeat("d", 40) + raw := ":100644 100644 " + a + " " + b + " M\x00edited\x00:000000 100644 " + zero + " " + a + " A\x00copy\x00:160000 160000 " + c + " " + d + " M\x00sub\x00" + blobs, err := changedBlobs(raw) + if err != nil || strings.Join(blobs, ",") != a+","+b { + t.Fatalf("blobs %v %v", blobs, err) + } + if _, err := changedBlobs("garbage\x00path\x00"); err == nil { + t.Fatal("accepted malformed diff-tree output") + } +} diff --git a/internal/fetcher/git.go b/internal/fetcher/git.go new file mode 100644 index 00000000..14cd501f --- /dev/null +++ b/internal/fetcher/git.go @@ -0,0 +1,114 @@ +package fetcher + +import ( + "bytes" + "context" + "encoding/base64" + "errors" + "fmt" + "os" + "os/exec" + "regexp" + "strconv" + "strings" + "time" +) + +const ( + networkTimeout = 5 * time.Minute + localTimeout = time.Minute + maxOutput = 64 << 20 + maxErrorOutput = 16 << 10 +) + +func basicCredential(token string) string { + return base64.StdEncoding.EncodeToString([]byte("x-access-token:" + token)) +} + +// git runs one command in the mirror with a minimal environment that ignores +// system and user configuration, prompts and hooks, and allows only the +// remote's protocol. The token reaches only network commands, through +// environment configuration; never argv or on-disk config. +func (repository mirror) git(ctx context.Context, network bool, stdin string, arguments ...string) (string, error) { + timeout := localTimeout + environment := []string{ + "PATH=" + os.Getenv("PATH"), "HOME=" + repository.dir, + "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL=" + os.DevNull, "GIT_TERMINAL_PROMPT=0", "GIT_ASKPASS=", "SSH_ASKPASS=", + "GIT_ALLOW_PROTOCOL=" + repository.protocol, "GIT_NO_LAZY_FETCH=1", + } + if network { + timeout = networkTimeout + environment = append(environment, "GIT_CONFIG_COUNT=1", "GIT_CONFIG_KEY_0=http.extraHeader", + "GIT_CONFIG_VALUE_0=Authorization: Basic "+basicCredential(repository.token)) + } + name := arguments[0] + commandContext, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + arguments = append([]string{ + "-c", "protocol.allow=never", "-c", "protocol." + repository.protocol + ".allow=always", "-c", "core.hooksPath=" + os.DevNull, + // Credentials must not follow redirects; maintenance must finish within the timeout. + "-c", "http.followRedirects=false", "-c", "gc.autoDetach=false", "-c", "maintenance.autoDetach=false", + }, arguments...) + command := exec.CommandContext(commandContext, "git", arguments...) + command.Dir = repository.dir + command.Env = environment + command.Stdin = strings.NewReader(stdin) + output := &limitedBuffer{maximum: maxOutput} + errorOutput := &limitedBuffer{maximum: maxErrorOutput} + command.Stdout, command.Stderr = output, errorOutput + isolate(command) + err := command.Run() + reap(command) + if commandContext.Err() != nil { + return "", fmt.Errorf("git %s: %w", name, commandContext.Err()) + } + if err != nil { + if detail := strings.TrimSpace(errorOutput.String()); detail != "" { + err = fmt.Errorf("%w: %s", err, detail) + } + return "", errors.New(redact(fmt.Sprintf("git %s: %v", name, err), repository.token)) + } + if output.exceeded { + return "", fmt.Errorf("git %s: output exceeds %d bytes", name, maxOutput) + } + return output.String(), nil +} + +type limitedBuffer struct { + bytes.Buffer + maximum int + exceeded bool +} + +func (buffer *limitedBuffer) Write(value []byte) (int, error) { + if remaining := buffer.maximum - buffer.Len(); len(value) > remaining { + buffer.exceeded = true + buffer.Buffer.Write(value[:max(remaining, 0)]) + return len(value), nil + } + return buffer.Buffer.Write(value) +} + +var gitVersion = regexp.MustCompile(`^git version (\d+)\.(\d+)`) + +// requireGit needs 2.44 for GIT_NO_LAZY_FETCH, which keeps collection offline. +func requireGit(ctx context.Context) error { + raw, err := (mirror{dir: os.TempDir(), protocol: "https"}).git(ctx, false, "", "version") + if err != nil { + return fmt.Errorf("diffx-fetcher requires git 2.44 or later: %w", err) + } + if !gitAtLeast(raw, 2, 44) { + return fmt.Errorf("diffx-fetcher requires git 2.44 or later; found %q", strings.TrimSpace(raw)) + } + return nil +} + +func gitAtLeast(raw string, major, minor int) bool { + match := gitVersion.FindStringSubmatch(strings.TrimSpace(raw)) + if match == nil { + return false + } + foundMajor, _ := strconv.Atoi(match[1]) + foundMinor, _ := strconv.Atoi(match[2]) + return foundMajor > major || foundMajor == major && foundMinor >= minor +} diff --git a/internal/fetcher/integration_test.go b/internal/fetcher/integration_test.go new file mode 100644 index 00000000..ca93e1e8 --- /dev/null +++ b/internal/fetcher/integration_test.go @@ -0,0 +1,132 @@ +package fetcher + +import ( + "context" + "net/url" + "testing" + "time" + + "github.com/flexdinesh/diffx/internal/fetchprotocol" + gh "github.com/flexdinesh/diffx/internal/github" + "github.com/flexdinesh/diffx/internal/githubsync" + "github.com/flexdinesh/diffx/internal/ingestion" + "github.com/flexdinesh/diffx/internal/review" + "github.com/flexdinesh/diffx/internal/reviewstore" +) + +// stubGitHub serves one repository with pull request 7 of an origin. +type stubGitHub struct{ pull gh.PullRequest } + +func (stubGitHub) AppSlug(context.Context) (string, error) { return "diffx-test", nil } +func (stubGitHub) VerifyWebhook([]byte, string) bool { return false } +func (stubGitHub) AuthorizeURL(state, redirect string) string { + return "https://github.test/?state=" + url.QueryEscape(state) +} +func (stubGitHub) Exchange(context.Context, string, string) (gh.UserToken, error) { + return gh.UserToken{AccessToken: "user", ExpiresAt: time.Now().Add(time.Hour), RefreshToken: "refresh"}, nil +} +func (stubGitHub) Refresh(context.Context, string) (gh.UserToken, error) { + return gh.UserToken{}, gh.ErrUnauthorized +} +func (stubGitHub) User(context.Context, string) (gh.Account, error) { + return gh.Account{ID: 1, Login: "alice"}, nil +} +func (stubGitHub) UserRepositories(context.Context, string) ([]gh.Repository, error) { + return []gh.Repository{{ID: 42, FullName: "octo/repo", InstallationID: 5}}, nil +} +func (s stubGitHub) OpenPullRequests(context.Context, int64, string) ([]gh.PullRequest, error) { + return []gh.PullRequest{s.pull}, nil +} +func (s stubGitHub) PullRequest(context.Context, int64, string, int) (gh.PullRequest, error) { + return s.pull, nil +} +func (stubGitHub) InstallationToken(context.Context, int64, []int64, map[string]string) (gh.Token, error) { + return gh.Token{Value: testToken, ExpiresAt: time.Now().Add(time.Hour)}, nil +} + +// leaseServer is the fetcher's view of the server, wired straight to the service. +type leaseServer struct { + service *githubsync.Service + reported func() + err error +} + +func (server *leaseServer) Lease(ctx context.Context) (*fetchprotocol.Job, error) { + return server.service.Lease(ctx) +} +func (server *leaseServer) Complete(ctx context.Context, job fetchprotocol.Job, request ingestion.Request) error { + defer server.reported() + server.err = server.service.Finish(ctx, fetchprotocol.Result{ID: job.ID, LeaseToken: job.LeaseToken, Request: &request}) + return server.err +} +func (server *leaseServer) Fail(ctx context.Context, job fetchprotocol.Job, detail string) error { + defer server.reported() + server.err = &fetchprotocol.StatusError{Detail: detail} + return server.service.Finish(ctx, fetchprotocol.Result{ID: job.ID, LeaseToken: job.LeaseToken, Error: detail}) +} + +func TestFetchedPullRequestBecomesFollowerReview(t *testing.T) { + requireTestGit(t) + origin := newOrigin(t) + store, err := reviewstore.Open("") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + owner, err := store.User("alice", "alice") + if err != nil { + t.Fatal(err) + } + created := time.UnixMilli(1_700_000_000_000) + repository := gh.Repository{ID: 42, FullName: "octo/repo", CloneURL: "https://github.com/octo/repo.git"} + pull := gh.PullRequest{Number: 7, State: "open", Title: "Edit files", Body: "Body", Author: gh.Account{Login: "mona"}, + Base: gh.Ref{Ref: "main", SHA: origin.base, Repository: repository}, Head: gh.Ref{Ref: "feature", SHA: origin.head, Repository: repository}, + CreatedAt: created, UpdatedAt: created.Add(time.Minute)} + service, err := githubsync.New(store.GitHub(), stubGitHub{pull: pull}, githubsync.Options{PublicURL: "https://diffx.test", TokenKey: make([]byte, 32)}) + if err != nil { + t.Fatal(err) + } + authorize, err := service.Connect(t.Context(), owner.ID) + if err != nil { + t.Fatal(err) + } + parsed, err := url.Parse(authorize) + if err != nil { + t.Fatal(err) + } + if err := service.Complete(t.Context(), owner.ID, parsed.Query().Get("state"), "code"); err != nil { + t.Fatal(err) + } + if err := service.Follow(t.Context(), owner.ID, 42); err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + server := &leaseServer{service: service, reported: cancel} + if err := run(ctx, server, Settings{Mirrors: t.TempDir(), DiskLimit: 1 << 30, Concurrency: 1, PollInterval: time.Millisecond, CollectorVersion: "test"}, origin.remote()); err != nil { + t.Fatal(err) + } + if server.err != nil { + t.Fatalf("server refused the fetched snapshot: %v", server.err) + } + delivery, err := store.Queue().Claim(t.Context(), time.Minute) + if err != nil || delivery.OwnerID != owner.ID { + t.Fatalf("follower admission: %+v %v", delivery, err) + } + published, err := store.Ingest(owner.ID, delivery.Request) + if err != nil { + t.Fatal(err) + } + item, err := store.Context(t.Context(), owner.ID, published.ContextID, time.Now()) + if err != nil { + t.Fatal(err) + } + metadata := item.Metadata + if metadata == nil || metadata.PullRequest == nil || metadata.PullRequest.Number != 7 || metadata.PullRequest.Author.Login != "mona" { + t.Fatalf("stored review lacks pull request metadata: %+v", metadata) + } + if metadata.RepositoryKey != review.RemoteRepositoryKey("https://github.com/octo/repo.git") || metadata.Branch != "feature" || metadata.Root != "https://github.com/octo/repo/pull/7" { + t.Fatalf("stored identity %+v", metadata) + } +} diff --git a/internal/fetcher/mirror.go b/internal/fetcher/mirror.go new file mode 100644 index 00000000..c22c928e --- /dev/null +++ b/internal/fetcher/mirror.go @@ -0,0 +1,264 @@ +package fetcher + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "regexp" + "strconv" + "strings" + "time" + + "github.com/flexdinesh/diffx/internal/collector" + "github.com/flexdinesh/diffx/internal/diffsource" + "github.com/flexdinesh/diffx/internal/fetchprotocol" + "github.com/flexdinesh/diffx/internal/ingestion" +) + +// remote resolves a leased repository's fetch URL and the only Git transport +// allowed to reach it. Production accepts GitHub HTTPS alone. +type remote struct { + protocol string + url func(fetchprotocol.Repository) (string, error) +} + +var github = remote{protocol: "https", url: githubURL} + +var repositoryName = regexp.MustCompile(`^[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?/[A-Za-z0-9._-]+$`) + +func githubURL(repository fetchprotocol.Repository) (string, error) { + name := repository.FullName[strings.IndexByte(repository.FullName, '/')+1:] + if !repositoryName.MatchString(repository.FullName) || name == "." || name == ".." || strings.HasSuffix(name, ".git") || + repository.CloneURL != "https://github.com/"+repository.FullName+".git" { + return "", errors.New("clone URL must be https://github.com/OWNER/NAME.git for the leased repository") + } + return repository.CloneURL, nil +} + +var objectID = regexp.MustCompile(`^(?:[0-9a-f]{40}|[0-9a-f]{64})$`) + +func validateJob(job fetchprotocol.Job) error { + switch { + case job.Repository.ID <= 0 || job.Number <= 0: + return errors.New("lease requires a repository ID and pull request number") + case !objectID.MatchString(job.HeadSHA) || !objectID.MatchString(job.BaseSHA): + return errors.New("lease head and base must be full commit IDs") + case job.Token == "" || strings.ContainsAny(job.Token, "\r\n"): + return errors.New("lease requires an installation token") + case !validBranch(job.BaseRef): + return errors.New("lease base ref is not a valid branch name") + } + return nil +} + +// validBranch applies `git check-ref-format --branch` rules for a refs/heads +// name, and also excludes refspec and option syntax. +func validBranch(name string) bool { + if name == "" || len(name) > 255 || strings.HasPrefix(name, "-") || strings.HasSuffix(name, "/") || strings.HasSuffix(name, ".") || + strings.Contains(name, "..") || strings.Contains(name, "@{") || strings.Contains(name, "//") || name == "@" || + strings.ContainsAny(name, " ~^:?*[\\\x7f") { + return false + } + for _, character := range name { + if character < 0x20 { + return false + } + } + for _, component := range strings.Split(name, "/") { + if strings.HasPrefix(component, ".") || strings.HasSuffix(component, ".lock") { + return false + } + } + return true +} + +type mirror struct { + dir string + protocol string + token string +} + +const usedMarker = "diffx-fetcher-used" + +// open reuses /, or atomically publishes a new partial +// clone there. Non-bare layout lets the collector resolve a repository root; +// nothing is ever checked out. +func (fetcher *fetcher) open(ctx context.Context, id int64, url, token string) (mirror, error) { + repository := mirror{dir: filepath.Join(fetcher.settings.Mirrors, strconv.FormatInt(id, 10)), protocol: fetcher.remote.protocol, token: token} + if _, err := os.Stat(filepath.Join(repository.dir, ".git", usedMarker)); errors.Is(err, os.ErrNotExist) { + // A mirror without its marker was never published; start over. + if err := os.RemoveAll(repository.dir); err != nil { + return mirror{}, err + } + if err := fetcher.create(ctx, repository); err != nil { + return mirror{}, err + } + } else if err != nil { + return mirror{}, err + } + // The URL follows repository renames by ID; it never carries credentials. + if _, err := repository.git(ctx, false, "", "config", "remote.origin.url", url); err != nil { + return mirror{}, err + } + now := time.Now() + return repository, os.Chtimes(filepath.Join(repository.dir, ".git", usedMarker), now, now) +} + +func (fetcher *fetcher) create(ctx context.Context, target mirror) error { + staging, err := os.MkdirTemp(fetcher.settings.Mirrors, stagingPrefix) + if err != nil { + return err + } + defer os.RemoveAll(staging) + repository := mirror{dir: staging, protocol: target.protocol} + if _, err := repository.git(ctx, false, "", "init", "--quiet"); err != nil { + return err + } + for _, setting := range [][2]string{ + {"core.repositoryformatversion", "1"}, + {"extensions.partialClone", "origin"}, + {"remote.origin.promisor", "true"}, + {"remote.origin.partialclonefilter", "blob:none"}, + } { + if _, err := repository.git(ctx, false, "", "config", setting[0], setting[1]); err != nil { + return err + } + } + if err := os.WriteFile(filepath.Join(staging, ".git", usedMarker), nil, 0o600); err != nil { + return err + } + return os.Rename(staging, target.dir) +} + +func headRef(job fetchprotocol.Job) string { return "refs/heads/diffx/pr/" + strconv.Itoa(job.Number) } +func baseRef(job fetchprotocol.Job) string { return "refs/remotes/origin/" + job.BaseRef } + +func (repository mirror) fetch(ctx context.Context, job fetchprotocol.Job) error { + // Only the leased base is kept, so renamed or nested base branches never + // conflict with an earlier lease's ref. + refs, err := repository.git(ctx, false, "", "for-each-ref", "--format=%(refname)", "refs/remotes/origin/") + if err != nil { + return err + } + var stale strings.Builder + for _, ref := range strings.Fields(refs) { + if ref != baseRef(job) { + stale.WriteString("delete " + ref + "\n") + } + } + if stale.Len() > 0 { + if _, err := repository.git(ctx, false, stale.String(), "update-ref", "--stdin"); err != nil { + return err + } + } + _, err = repository.git(ctx, true, "", "fetch", "--quiet", "--no-tags", "--recurse-submodules=no", "--filter=blob:none", "origin", + "+refs/pull/"+strconv.Itoa(job.Number)+"/head:"+headRef(job), "+refs/heads/"+job.BaseRef+":"+baseRef(job)) + if err != nil { + return fmt.Errorf("fetch pull request: %w", err) + } + return nil +} + +func (repository mirror) verify(ctx context.Context, job fetchprotocol.Job) error { + for _, expected := range [][2]string{{headRef(job), job.HeadSHA}, {baseRef(job), job.BaseSHA}} { + current, err := repository.git(ctx, false, "", "rev-parse", "--verify", "--end-of-options", expected[0]+"^{commit}") + if err != nil { + return err + } + if strings.TrimSpace(current) != expected[1] { + return errMoved + } + } + return nil +} + +// prefetch downloads only blobs on either side of the pull request diff, so +// offline collection can read every patch and full file contents. +func (repository mirror) prefetch(ctx context.Context, job fetchprotocol.Job) error { + raw, err := repository.git(ctx, false, "", "merge-base", job.BaseSHA, job.HeadSHA) + if err != nil { + return fmt.Errorf("find merge base: %w", err) + } + // Without rename detection both sides of every change are listed. + raw, err = repository.git(ctx, false, "", "diff-tree", "-r", "--raw", "--no-abbrev", "--no-renames", "-z", strings.TrimSpace(raw), job.HeadSHA) + if err != nil { + return err + } + blobs, err := changedBlobs(raw) + if err != nil { + return err + } + missing, err := repository.missing(ctx, blobs) + if err != nil || len(missing) == 0 { + return err + } + _, err = repository.git(ctx, true, strings.Join(missing, "\n")+"\n", "fetch", "--quiet", "--no-tags", "--no-write-fetch-head", + "--recurse-submodules=no", "--filter=blob:none", "--stdin", "origin") + if err != nil { + return fmt.Errorf("fetch changed files: %w", err) + } + missing, err = repository.missing(ctx, missing) + if err != nil { + return err + } + if len(missing) > 0 { + return fmt.Errorf("%d changed file objects unavailable after fetch, including %s", len(missing), missing[0]) + } + return nil +} + +func changedBlobs(raw string) ([]string, error) { + fields := strings.Split(raw, "\x00") + seen := make(map[string]bool) + var blobs []string + for index := 0; index+1 < len(fields); index += 2 { + header, found := strings.CutPrefix(fields[index], ":") + parts := strings.Fields(header) + if !found || len(parts) != 5 { + return nil, fmt.Errorf("unexpected diff-tree record %q", fields[index]) + } + for _, side := range [][2]string{{parts[0], parts[2]}, {parts[1], parts[3]}} { + mode, oid := side[0], side[1] + // Submodule commits and absent sides have no blob to read. + if mode == "160000" || strings.Trim(oid, "0") == "" || seen[oid] { + continue + } + seen[oid] = true + blobs = append(blobs, oid) + } + } + return blobs, nil +} + +func (repository mirror) missing(ctx context.Context, objects []string) ([]string, error) { + if len(objects) == 0 { + return nil, nil + } + raw, err := repository.git(ctx, false, strings.Join(objects, "\n")+"\n", "cat-file", "--batch-check") + if err != nil { + return nil, err + } + var missing []string + for _, line := range strings.Split(strings.TrimSpace(raw), "\n") { + if oid, absent := strings.CutSuffix(line, " missing"); absent { + missing = append(missing, oid) + } + } + return missing, nil +} + +// collect reads only local objects: the collector's Git processes inherit +// GIT_NO_LAZY_FETCH=1, which Run requires. +func (repository mirror) collect(ctx context.Context, job fetchprotocol.Job, collectorVersion string) (ingestion.Request, error) { + request, _, err := collector.CollectChanged(ctx, repository.dir, collector.Options{ + SourceID: "github", Hostname: "github.com", Trigger: "github", CollectorVersion: collectorVersion, + Branch: strings.TrimPrefix(headRef(job), "refs/heads/"), Base: baseRef(job), + Identity: &diffsource.StableRepositoryIdentity{RepositoryKey: "github", CheckoutKey: "github", BranchID: "github"}, + }, "") + if err != nil { + return ingestion.Request{}, fmt.Errorf("collect pull request: %w", err) + } + return request, nil +} diff --git a/internal/fetcher/process_unix.go b/internal/fetcher/process_unix.go new file mode 100644 index 00000000..61fbdf46 --- /dev/null +++ b/internal/fetcher/process_unix.go @@ -0,0 +1,24 @@ +//go:build !windows + +package fetcher + +import ( + "os/exec" + "syscall" + "time" +) + +// isolate gives each Git invocation its own process group so cancellation +// also stops remote helpers and index-pack. +func isolate(command *exec.Cmd) { + command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + command.WaitDelay = time.Second + command.Cancel = func() error { return syscall.Kill(-command.Process.Pid, syscall.SIGKILL) } +} + +// reap stops helpers that outlive their leader, before the mirror is reused or evicted. +func reap(command *exec.Cmd) { + if command.Process != nil { + _ = syscall.Kill(-command.Process.Pid, syscall.SIGKILL) + } +} diff --git a/internal/fetcher/process_windows.go b/internal/fetcher/process_windows.go new file mode 100644 index 00000000..5e2ea43d --- /dev/null +++ b/internal/fetcher/process_windows.go @@ -0,0 +1,12 @@ +//go:build windows + +package fetcher + +import ( + "os/exec" + "time" +) + +func isolate(command *exec.Cmd) { command.WaitDelay = time.Second } + +func reap(*exec.Cmd) {} diff --git a/internal/fetchprotocol/protocol.go b/internal/fetchprotocol/protocol.go new file mode 100644 index 00000000..1c842720 --- /dev/null +++ b/internal/fetchprotocol/protocol.go @@ -0,0 +1,154 @@ +// Package fetchprotocol is the internal contract between diffx-server and +// diffx-fetcher. The server owns GitHub credentials and identity; the fetcher +// leases one pull request at a time, mirrors it and returns an owner-free +// snapshot. It never claims an account or pull request metadata. +package fetchprotocol + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "github.com/flexdinesh/diffx/internal/ingestion" +) + +const ( + LeasePath = "/internal/v1/fetch-jobs/lease" + ResultPath = "/internal/v1/fetch-jobs/result" +) + +// Job leases one pull request head. Token is a short-lived installation token +// limited to Repository with contents:read; it is valid only for fetching. +type Job struct { + ID string `json:"id"` + LeaseToken string `json:"leaseToken"` + LeaseExpiresAt int64 `json:"leaseExpiresAt"` + Repository Repository `json:"repository"` + Number int `json:"number"` + HeadSHA string `json:"headSha"` + BaseRef string `json:"baseRef"` + BaseSHA string `json:"baseSha"` + Token string `json:"token"` + TokenExpiresAt int64 `json:"tokenExpiresAt"` +} + +type Repository struct { + ID int64 `json:"id"` + FullName string `json:"fullName"` + CloneURL string `json:"cloneUrl"` +} + +// Result completes a lease with a snapshot, or fails it with a detail for retry. +type Result struct { + ID string `json:"id"` + LeaseToken string `json:"leaseToken"` + Request *ingestion.Request `json:"request,omitempty"` + Error string `json:"error,omitempty"` +} + +// MaxResultBytes bounds a result: one snapshot plus envelope. +const MaxResultBytes = ingestion.MaxRequestBytes + 1<<20 + +// Client is the fetcher's view of the server's internal listener. +type Client struct { + baseURL string + secret string + http *http.Client +} + +func NewClient(baseURL, secret string) (*Client, error) { + parsed, err := url.Parse(baseURL) + if err != nil || parsed.Host == "" || parsed.Scheme != "http" && parsed.Scheme != "https" || parsed.User != nil { + return nil, errors.New("fetcher server URL must be an http(s) URL without credentials") + } + if secret == "" { + return nil, errors.New("fetcher secret required") + } + return &Client{baseURL: strings.TrimSuffix(baseURL, "/"), secret: secret, http: &http.Client{Timeout: 2 * time.Minute}}, nil +} + +// Lease returns nil when no pull request needs fetching. +func (c *Client) Lease(ctx context.Context) (*Job, error) { + response, err := c.post(ctx, LeasePath, nil) + if err != nil { + return nil, err + } + defer response.Body.Close() + if response.StatusCode == http.StatusNoContent { + return nil, nil + } + if response.StatusCode != http.StatusOK { + return nil, statusError(response) + } + var job Job + if err := json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(&job); err != nil { + return nil, fmt.Errorf("decode fetch lease: %w", err) + } + return &job, nil +} + +func (c *Client) Complete(ctx context.Context, job Job, request ingestion.Request) error { + return c.result(ctx, Result{ID: job.ID, LeaseToken: job.LeaseToken, Request: &request}) +} + +func (c *Client) Fail(ctx context.Context, job Job, detail string) error { + if len(detail) > 4096 { + detail = detail[:4096] + } + return c.result(ctx, Result{ID: job.ID, LeaseToken: job.LeaseToken, Error: detail}) +} + +func (c *Client) result(ctx context.Context, result Result) error { + raw, err := json.Marshal(result) + if err != nil { + return err + } + response, err := c.post(ctx, ResultPath, raw) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode != http.StatusNoContent { + return statusError(response) + } + return nil +} + +func (c *Client) post(ctx context.Context, path string, body []byte) (*http.Response, error) { + request, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+path, bytes.NewReader(body)) + if err != nil { + return nil, err + } + request.Header.Set("Authorization", "Bearer "+c.secret) + request.Header.Set("Content-Type", "application/json") + return c.http.Do(request) +} + +// StatusError reports a non-success response; 409 means the lease was lost. +type StatusError struct { + Status int + Detail string +} + +func (err *StatusError) Error() string { + return fmt.Sprintf("fetch server responded %d: %s", err.Status, err.Detail) +} + +func statusError(response *http.Response) error { + raw, _ := io.ReadAll(io.LimitReader(response.Body, 4096)) + var problem struct { + Detail string `json:"detail"` + } + detail := strings.TrimSpace(string(raw)) + if json.Unmarshal(raw, &problem) == nil && problem.Detail != "" { + detail = problem.Detail + } + return &StatusError{Status: response.StatusCode, Detail: detail} +} diff --git a/internal/githubsync/fetch.go b/internal/githubsync/fetch.go new file mode 100644 index 00000000..c682a74a --- /dev/null +++ b/internal/githubsync/fetch.go @@ -0,0 +1,244 @@ +package githubsync + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "github.com/flexdinesh/diffx/internal/fetchprotocol" + "github.com/flexdinesh/diffx/internal/github" + "github.com/flexdinesh/diffx/internal/ingestion" + "github.com/flexdinesh/diffx/internal/review" +) + +// leaseDuration bounds one fetch; installation tokens outlive it. +const leaseDuration = 10 * time.Minute + +// maxSkips bounds pending refreshes resolved without a fetch in one lease call. +const maxSkips = 32 + +// ErrInvalidResult reports a fetcher snapshot that does not match its lease. +var ErrInvalidResult = errors.New("fetch result does not match the leased pull request") + +// Lease claims the next pending pull request refresh for the fetcher. It +// resolves refreshes that need no fetch (deleted pull requests, repositories +// without followers) itself. Nil means nothing is pending. +func (s *Service) Lease(ctx context.Context) (*fetchprotocol.Job, error) { + for range maxSkips { + token, err := randomToken() + if err != nil { + return nil, err + } + now := s.now() + lease, err := s.store.ClaimRefresh(ctx, token, now.UnixMilli(), now.Add(leaseDuration).UnixMilli()) + if errors.Is(err, ErrNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + job, err := s.prepare(ctx, lease, now) + if err != nil { + return nil, errors.Join(err, s.release(ctx, lease, err.Error())) + } + if job != nil { + return job, nil + } + } + return nil, nil +} + +// prepare reads the pull request and mints the fetch credential. Nil without +// error means the refresh completed without a fetch. +func (s *Service) prepare(ctx context.Context, lease Lease, now time.Time) (*fetchprotocol.Job, error) { + ref := lease.Ref + pull, err := s.github.PullRequest(ctx, ref.InstallationID, ref.FullName, ref.Number) + var status *github.Error + if errors.As(err, &status) && status.Status == http.StatusNotFound { + return nil, s.store.CompleteRefresh(ctx, lease, "closed", nil, now.UnixMilli()) + } + if err != nil { + return nil, fmt.Errorf("read pull request %s#%d: %w", ref.FullName, ref.Number, err) + } + follows, err := s.store.RepositoryFollows(ctx, ref.RepositoryID) + if err != nil { + return nil, err + } + if !hasActive(follows) { + return nil, s.store.CompleteRefresh(ctx, lease, pullState(pull), nil, now.UnixMilli()) + } + credential, err := s.github.InstallationToken(ctx, ref.InstallationID, []int64{ref.RepositoryID}, map[string]string{"contents": "read"}) + if err != nil { + return nil, fmt.Errorf("mint fetch token for %s: %w", ref.FullName, err) + } + raw, err := json.Marshal(pull) + if err != nil { + return nil, err + } + if err := s.store.RecordLease(ctx, lease, raw, now.UnixMilli()); err != nil { + return nil, err + } + // GitHub's current name follows renames; the fetcher refuses redirects. + name := ref.FullName + if pull.Base.Repository.FullName != "" { + name = pull.Base.Repository.FullName + } + return &fetchprotocol.Job{ + ID: jobID(ref), LeaseToken: lease.Token, LeaseExpiresAt: now.Add(leaseDuration).UnixMilli(), + Repository: fetchprotocol.Repository{ID: ref.RepositoryID, FullName: name, CloneURL: "https://github.com/" + name + ".git"}, + Number: ref.Number, HeadSHA: pull.Head.SHA, BaseRef: pull.Base.Ref, BaseSHA: pull.Base.SHA, + Token: credential.Value, TokenExpiresAt: credential.ExpiresAt.UnixMilli(), + }, nil +} + +// Finish applies a fetcher result. A snapshot is published to every account +// with verified access in one transaction with the refresh; a failure retries +// with backoff. ErrLeaseLost means another lease superseded this one. +func (s *Service) Finish(ctx context.Context, result fetchprotocol.Result) error { + repositoryID, number, ok := parseJobID(result.ID) + if !ok { + return ErrLeaseLost + } + lease, raw, err := s.store.Leased(ctx, repositoryID, number, result.LeaseToken, s.millis()) + if err != nil { + return err + } + if result.Request == nil { + detail := result.Error + if detail == "" { + detail = "fetch failed" + } + return s.release(ctx, lease, detail) + } + var pull github.PullRequest + if err := json.Unmarshal(raw, &pull); err != nil { + return err + } + request, err := publication(lease, pull, *result.Request) + if err != nil { + return errors.Join(err, s.release(ctx, lease, err.Error())) + } + owners, err := s.Followers(ctx, repositoryID) + if err != nil { + return err + } + admissions := make([]Admission, 0, len(owners)) + for _, owner := range owners { + admissions = append(admissions, Admission{OwnerID: owner, Request: request}) + } + return s.store.CompleteRefresh(ctx, lease, pullState(pull), admissions, s.millis()) +} + +// release ends a lease with exponential backoff capped at an hour. +func (s *Service) release(ctx context.Context, lease Lease, detail string) error { + delay := time.Minute << min(lease.Attempt, 6) + if delay > time.Hour { + delay = time.Hour + } + if len(detail) > 4096 { + detail = detail[:4096] + } + return s.store.ReleaseRefresh(ctx, lease, s.now().Add(delay).UnixMilli(), detail) +} + +// publication turns an owner-free fetcher snapshot into the server-authored +// observation. Identity, labels and pull request metadata come from the lease, +// never the fetcher; the snapshot must be of the leased head and base. +func publication(lease Lease, pull github.PullRequest, fetched ingestion.Request) (ingestion.Request, error) { + metadata := fetched.Metadata + comparison := metadata.Comparison + if metadata.Head == nil || *metadata.Head != pull.Head.SHA || comparison == nil || comparison.Kind != "branch" || comparison.BaseCommit != pull.Base.SHA { + return ingestion.Request{}, ErrInvalidResult + } + ref := lease.Ref + remote := "https://github.com/" + ref.FullName + ".git" + repositoryKey := review.RemoteRepositoryKey(remote) + pullURL := "https://github.com/" + ref.FullName + "/pull/" + strconv.Itoa(ref.Number) + _, name, _ := strings.Cut(ref.FullName, "/") + identity := strconv.FormatInt(ref.RepositoryID, 10) + "\x00" + strconv.Itoa(ref.Number) + head := pull.Head.SHA + request := ingestion.Request{ + ProtocolVersion: ingestion.ProtocolVersion, + ContentHash: fetched.ContentHash, + Metadata: ingestion.Metadata{ + SourceID: ingestion.GitHubSourceID, Hostname: "github.com", Trigger: ingestion.GitHubTrigger, + RepositoryKey: repositoryKey, RepositoryName: name, RemoteURL: remote, + CheckoutKey: digest("github-pull-request-checkout", identity), BranchID: digest("github-pull-request-branch", identity), + Root: pullURL, WorktreeName: "PR #" + strconv.Itoa(ref.Number), Branch: pull.Head.Ref, Head: &head, + CollectedAt: pull.UpdatedAt.UnixMilli(), ActivityAt: pull.UpdatedAt.UnixMilli(), CollectorVersion: metadata.CollectorVersion, + Comparison: &ingestion.Comparison{Kind: "branch", BaseRef: "refs/heads/" + pull.Base.Ref, BaseCommit: comparison.BaseCommit, MergeBase: comparison.MergeBase}, + PullRequest: &ingestion.PullRequest{ + Provider: "github", RepositoryID: ref.RepositoryID, FullName: ref.FullName, Number: ref.Number, URL: pullURL, + Title: pull.Title, Body: pull.Body, State: pullState(pull), Draft: pull.Draft, + Author: ingestion.PullRequestAuthor{Login: pull.Author.Login, URL: "https://github.com/" + pull.Author.Login}, + BaseRef: pull.Base.Ref, HeadRef: pull.Head.Ref, HeadRepository: pull.Head.Repository.FullName, + CreatedAt: pull.CreatedAt.UnixMilli(), UpdatedAt: pull.UpdatedAt.UnixMilli(), + }, + }, + Scopes: make([]ingestion.Scope, 0, len(fetched.Scopes)), + } + for _, scope := range fetched.Scopes { + scope.Snapshot.Root, scope.Snapshot.Name, scope.Snapshot.Branch = pullURL, ref.FullName+"#"+strconv.Itoa(ref.Number), pull.Head.Ref + request.Scopes = append(request.Scopes, scope) + } + // Identical pull request state yields an identical payload, so refetches + // replay; a retention run makes the submission fresh. + raw, err := json.Marshal(request) + if err != nil { + return ingestion.Request{}, err + } + request.SubmissionID = "gh-" + digest(string(raw), lease.RetentionRun) + if err := ingestion.Validate(request); err != nil { + return ingestion.Request{}, errors.Join(ErrInvalidResult, err) + } + return request, nil +} + +func pullState(pull github.PullRequest) string { + switch { + case pull.Merged: + return "merged" + case pull.State == "closed": + return "closed" + default: + return "open" + } +} + +func hasActive(follows []Follow) bool { + for _, follow := range follows { + if follow.Status == "active" { + return true + } + } + return false +} + +func digest(parts ...string) string { + sum := sha256.Sum256([]byte(strings.Join(parts, "\x00"))) + return hex.EncodeToString(sum[:]) +} + +func jobID(ref PullRequestRef) string { + return strconv.FormatInt(ref.RepositoryID, 10) + "-" + strconv.Itoa(ref.Number) +} + +func parseJobID(id string) (int64, int, bool) { + repository, number, ok := strings.Cut(id, "-") + if !ok { + return 0, 0, false + } + repositoryID, err := strconv.ParseInt(repository, 10, 64) + if err != nil { + return 0, 0, false + } + pullNumber, err := strconv.Atoi(number) + return repositoryID, pullNumber, err == nil +} diff --git a/internal/githubsync/fetch_test.go b/internal/githubsync/fetch_test.go new file mode 100644 index 00000000..37623cf8 --- /dev/null +++ b/internal/githubsync/fetch_test.go @@ -0,0 +1,310 @@ +package githubsync_test + +import ( + "errors" + "slices" + "strings" + "testing" + "time" + + "github.com/flexdinesh/diffx/internal/fetchprotocol" + "github.com/flexdinesh/diffx/internal/github" + "github.com/flexdinesh/diffx/internal/githubsync" + "github.com/flexdinesh/diffx/internal/ingestion" + "github.com/flexdinesh/diffx/internal/ingestionqueue" + "github.com/flexdinesh/diffx/internal/review" +) + +const ( + headOne = "1111111111111111111111111111111111111111" + headTwo = "2222222222222222222222222222222222222222" + baseTip = "3333333333333333333333333333333333333333" +) + +func pullRequest(head string) github.PullRequest { + created := time.UnixMilli(1_700_000_000_000) + return github.PullRequest{ + Number: 1, State: "open", Title: "Add search", Body: "Adds **search**.", HTMLURL: "https://github.com/octo/app/pull/1", + Author: github.Account{Login: "mona"}, + Base: github.Ref{Ref: "main", SHA: baseTip, Repository: github.Repository{ID: appRepo.ID, FullName: "octo/app", CloneURL: "https://github.com/octo/app.git"}}, + Head: github.Ref{Ref: "search", SHA: head, Repository: github.Repository{ID: 77, FullName: "mona/app"}}, + CreatedAt: created, UpdatedAt: created.Add(time.Hour), + } +} + +// fetched is what a fetcher returns: an owner-free snapshot of its mirror with +// its own labels, which the server must replace. +func fetched(job *fetchprotocol.Job) ingestion.Request { + head := job.HeadSHA + return ingestion.Request{ + ProtocolVersion: ingestion.ProtocolVersion, SubmissionID: "fetcher-chosen", + Metadata: ingestion.Metadata{ + SourceID: "fetcher", Hostname: "fetcher-host", Trigger: "manual", RepositoryKey: "mirror", CheckoutKey: "mirror", + Root: "/mirrors/10", Branch: "diffx/pr/1", Head: &head, CollectedAt: 1, CollectorVersion: "test", + Comparison: &ingestion.Comparison{Kind: "branch", BaseRef: "refs/remotes/origin/main", BaseCommit: job.BaseSHA, MergeBase: job.BaseSHA}, + }, + Scopes: []ingestion.Scope{{ + Snapshot: review.RepositoryDiff{ + Source: "local", Root: "/mirrors/10", Name: "10", Branch: "diffx/pr/1", Head: &head, Mode: review.DiffAll, + ComparisonKind: review.ComparisonCommitted, Revision: "revision-" + head, + Files: []review.ChangedFile{{ID: "f1", Path: "search.go", Status: "M", Fingerprint: "fp-" + head}}, + }, + Patches: map[string]review.FilePatch{"f1": {Patch: "@@ -1 +1 @@\n-a\n+b\n"}}, + }}, + } +} + +func (f *fixture) followPull(t *testing.T, owners ...string) { + t.Helper() + f.github.open = map[string][]github.PullRequest{"octo/app": {{Number: 1}}} + f.github.pulls = map[int]github.PullRequest{1: pullRequest(headOne)} + for _, owner := range owners { + f.connect(t, owner, map[string]string{f.alice: "alice", f.bob: "bob"}[owner]) + if err := f.service.Follow(t.Context(), owner, appRepo.ID); err != nil { + t.Fatal(err) + } + } +} + +func (f *fixture) lease(t *testing.T) *fetchprotocol.Job { + t.Helper() + job, err := f.service.Lease(t.Context()) + if err != nil { + t.Fatal(err) + } + return job +} + +func (f *fixture) publish(t *testing.T) *fetchprotocol.Job { + t.Helper() + job := f.lease(t) + if job == nil { + t.Fatal("no pending pull request") + } + request := fetched(job) + if err := f.service.Finish(t.Context(), fetchprotocol.Result{ID: job.ID, LeaseToken: job.LeaseToken, Request: &request}); err != nil { + t.Fatal(err) + } + return job +} + +// admitted claims every newly admitted submission. +func (f *fixture) admitted(t *testing.T) []ingestionqueue.Delivery { + t.Helper() + var deliveries []ingestionqueue.Delivery + for { + delivery, err := f.store.Queue().Claim(t.Context(), time.Hour) + if errors.Is(err, ingestionqueue.ErrEmpty) { + return deliveries + } + if err != nil { + t.Fatal(err) + } + deliveries = append(deliveries, delivery) + } +} + +func TestLeasePublishesServerAuthoredSnapshotToFollowers(t *testing.T) { + f := newFixture(t) + f.followPull(t, f.alice, f.bob) + job := f.lease(t) + if job == nil || job.HeadSHA != headOne || job.BaseSHA != baseTip || job.BaseRef != "main" || job.Repository.CloneURL != "https://github.com/octo/app.git" || job.Token != "installation-token" { + t.Fatalf("job %+v", job) + } + grant := f.github.tokens[len(f.github.tokens)-1] + if grant.installation != appRepo.InstallationID || !slices.Equal(grant.repositories, []int64{appRepo.ID}) || len(grant.permissions) != 1 || grant.permissions["contents"] != "read" { + t.Fatalf("fetch token not narrowed to the repository's contents: %+v", grant) + } + if again := f.lease(t); again != nil { + t.Fatalf("leased pull request leased twice: %+v", again) + } + request := fetched(job) + if err := f.service.Finish(t.Context(), fetchprotocol.Result{ID: job.ID, LeaseToken: job.LeaseToken, Request: &request}); err != nil { + t.Fatal(err) + } + deliveries := f.admitted(t) + if len(deliveries) != 2 { + t.Fatalf("admitted %d submissions, want one per follower", len(deliveries)) + } + owners := []string{deliveries[0].OwnerID, deliveries[1].OwnerID} + slices.Sort(owners) + want := []string{f.alice, f.bob} + slices.Sort(want) + if !slices.Equal(owners, want) { + t.Fatalf("admitted to %v, want both followers", owners) + } + got := deliveries[0].Request + metadata := got.Metadata + pull := metadata.PullRequest + if metadata.SourceID != "github" || metadata.Trigger != "github" || metadata.Hostname != "github.com" || pull == nil { + t.Fatalf("fetcher provenance survived: %+v", metadata) + } + if metadata.RepositoryKey != review.RemoteRepositoryKey("https://github.com/octo/app.git") || metadata.Root != "https://github.com/octo/app/pull/1" || metadata.Branch != "search" || metadata.WorktreeName != "PR #1" { + t.Fatalf("identity %+v", metadata) + } + if got.Scopes[0].Snapshot.Root != metadata.Root || got.Scopes[0].Snapshot.Branch != "search" || !strings.HasPrefix(got.SubmissionID, "gh-") { + t.Fatalf("scope or submission not rewritten: %+v %q", got.Scopes[0].Snapshot, got.SubmissionID) + } + if pull.Title != "Add search" || pull.Author.Login != "mona" || pull.Author.URL != "https://github.com/mona" || pull.State != "open" || pull.HeadRepository != "mona/app" || metadata.CollectedAt != pull.UpdatedAt { + t.Fatalf("pull request %+v", pull) + } + if deliveries[1].Request.SubmissionID != got.SubmissionID { + t.Fatal("followers received different submissions of one snapshot") + } + if f.lease(t) != nil { + t.Fatal("completed refresh still pending") + } +} + +func TestRefetchReplaysAndNewHeadIsFresh(t *testing.T) { + f := newFixture(t) + f.followPull(t, f.alice) + f.publish(t) + first := f.admitted(t) + if len(first) != 1 { + t.Fatalf("admitted %d", len(first)) + } + f.webhook(t, "again", "synchronize") + f.publish(t) + if replayed := f.admitted(t); len(replayed) != 0 { + t.Fatalf("unchanged pull request admitted again: %+v", replayed) + } + f.github.pulls[1] = pullRequest(headTwo) + f.webhook(t, "pushed", "synchronize") + f.publish(t) + pushed := f.admitted(t) + if len(pushed) != 1 || pushed[0].SubmissionID == first[0].SubmissionID || *pushed[0].Request.Metadata.Head != headTwo { + t.Fatalf("force push not a fresh submission: %+v", pushed) + } + if pushed[0].Request.Metadata.CheckoutKey != first[0].Request.Metadata.CheckoutKey || pushed[0].Request.Metadata.BranchID != first[0].Request.Metadata.BranchID { + t.Fatal("new head left the pull request's stream") + } +} + +func TestResyncMakesSubmissionFresh(t *testing.T) { + f := newFixture(t) + f.followPull(t, f.alice) + f.publish(t) + first := f.admitted(t) + if err := f.service.Resync(t.Context(), f.alice); err != nil { + t.Fatal(err) + } + f.publish(t) + resynced := f.admitted(t) + if len(resynced) != 1 || resynced[0].SubmissionID == first[0].SubmissionID { + t.Fatalf("resync replayed instead of extending retention: %+v", resynced) + } + f.webhook(t, "after", "edited") + f.publish(t) + if later := f.admitted(t); len(later) != 0 { + t.Fatalf("retention run outlived its refresh: %+v", later) + } +} + +func TestFinishIsFencedByLease(t *testing.T) { + f := newFixture(t) + f.followPull(t, f.alice) + job := f.lease(t) + request := fetched(job) + if err := f.service.Finish(t.Context(), fetchprotocol.Result{ID: job.ID, LeaseToken: "forged", Request: &request}); !errors.Is(err, githubsync.ErrLeaseLost) { + t.Fatalf("forged lease: %v", err) + } + *f.clock = f.clock.Add(11 * time.Minute) + if err := f.service.Finish(t.Context(), fetchprotocol.Result{ID: job.ID, LeaseToken: job.LeaseToken, Request: &request}); !errors.Is(err, githubsync.ErrLeaseLost) { + t.Fatalf("expired lease: %v", err) + } + next := f.lease(t) + if next == nil || next.LeaseToken == job.LeaseToken { + t.Fatal("expired lease not reclaimable") + } + if err := f.service.Finish(t.Context(), fetchprotocol.Result{ID: job.ID, LeaseToken: job.LeaseToken, Request: &request}); !errors.Is(err, githubsync.ErrLeaseLost) { + t.Fatalf("superseded lease: %v", err) + } + if deliveries := f.admitted(t); len(deliveries) != 0 { + t.Fatalf("fenced result admitted: %+v", deliveries) + } +} + +func TestMismatchedSnapshotRetriesWithBackoff(t *testing.T) { + f := newFixture(t) + f.followPull(t, f.alice) + job := f.lease(t) + staleBase := *job + staleBase.BaseSHA = headTwo + request := fetched(&staleBase) + if err := f.service.Finish(t.Context(), fetchprotocol.Result{ID: job.ID, LeaseToken: job.LeaseToken, Request: &request}); !errors.Is(err, githubsync.ErrInvalidResult) { + t.Fatalf("snapshot against another base: %v", err) + } + *f.clock = f.clock.Add(time.Minute) + job = f.lease(t) + staleHead := *job + staleHead.HeadSHA = headTwo + request = fetched(&staleHead) + if err := f.service.Finish(t.Context(), fetchprotocol.Result{ID: job.ID, LeaseToken: job.LeaseToken, Request: &request}); !errors.Is(err, githubsync.ErrInvalidResult) { + t.Fatalf("snapshot of another head: %v", err) + } + if deliveries := f.admitted(t); len(deliveries) != 0 { + t.Fatalf("mismatched snapshot admitted: %+v", deliveries) + } + if f.lease(t) != nil { + t.Fatal("failed refresh retried without backoff") + } + *f.clock = f.clock.Add(2 * time.Minute) + retry := f.lease(t) + if retry == nil { + t.Fatal("failed refresh not retried") + } + if err := f.service.Finish(t.Context(), fetchprotocol.Result{ID: retry.ID, LeaseToken: retry.LeaseToken, Error: "mirror unavailable"}); err != nil { + t.Fatal(err) + } + *f.clock = f.clock.Add(2 * time.Minute) + if f.lease(t) != nil { + t.Fatal("third failure backed off no longer than the second") + } +} + +func TestRefreshDuringLeaseStaysPending(t *testing.T) { + f := newFixture(t) + f.followPull(t, f.alice) + job := f.lease(t) + f.github.pulls[1] = pullRequest(headTwo) + f.webhook(t, "during", "synchronize") + request := fetched(job) + if err := f.service.Finish(t.Context(), fetchprotocol.Result{ID: job.ID, LeaseToken: job.LeaseToken, Request: &request}); err != nil { + t.Fatal(err) + } + next := f.lease(t) + if next == nil || next.HeadSHA != headTwo { + t.Fatalf("push during the fetch was lost: %+v", next) + } +} + +func TestDeletedOrUnfollowedPullRequestsNeedNoFetch(t *testing.T) { + f := newFixture(t) + f.followPull(t, f.alice) + delete(f.github.pulls, 1) + if job := f.lease(t); job != nil { + t.Fatalf("deleted pull request leased: %+v", job) + } + if err := f.service.Reconcile(t.Context()); err != nil { + t.Fatal(err) + } + f.github.pulls[1] = pullRequest(headOne) + if err := f.service.Unfollow(t.Context(), f.alice, appRepo.ID); err != nil { + t.Fatal(err) + } + if job := f.lease(t); job != nil { + t.Fatalf("unfollowed pull request leased: %+v", job) + } + if deliveries := f.admitted(t); len(deliveries) != 0 { + t.Fatalf("admitted without followers: %+v", deliveries) + } +} + +func (f *fixture) webhook(t *testing.T, delivery, action string) { + t.Helper() + body := []byte(`{"action":"` + action + `","number":1,"installation":{"id":5},"repository":{"id":10,"full_name":"octo/app"}}`) + if err := f.service.Webhook(t.Context(), "pull_request", delivery, "valid", body); err != nil { + t.Fatal(err) + } +} diff --git a/internal/githubsync/service.go b/internal/githubsync/service.go index 04e1873c..cdc1c003 100644 --- a/internal/githubsync/service.go +++ b/internal/githubsync/service.go @@ -29,6 +29,8 @@ type GitHub interface { User(ctx context.Context, accessToken string) (github.Account, error) UserRepositories(ctx context.Context, accessToken string) ([]github.Repository, error) OpenPullRequests(ctx context.Context, installationID int64, fullName string) ([]github.PullRequest, error) + PullRequest(ctx context.Context, installationID int64, fullName string, number int) (github.PullRequest, error) + InstallationToken(ctx context.Context, installationID int64, repositoryIDs []int64, permissions map[string]string) (github.Token, error) } var ( @@ -476,7 +478,7 @@ func (s *Service) apply(ctx context.Context, event any) error { return nil } follows, err := s.store.RepositoryFollows(ctx, event.Repository.ID) - if err != nil || !slices.ContainsFunc(follows, func(f Follow) bool { return f.Status == "active" }) { + if err != nil || !hasActive(follows) { return err } ref := PullRequestRef{RepositoryID: event.Repository.ID, InstallationID: event.InstallationID, FullName: event.Repository.FullName, Number: event.Number} diff --git a/internal/githubsync/service_test.go b/internal/githubsync/service_test.go index 25361d5f..aed78eb8 100644 --- a/internal/githubsync/service_test.go +++ b/internal/githubsync/service_test.go @@ -23,6 +23,14 @@ type fakeGitHub struct { refreshes int refreshError error listings int + pulls map[int]github.PullRequest + tokens []tokenGrant +} + +type tokenGrant struct { + installation int64 + repositories []int64 + permissions map[string]string } func (f *fakeGitHub) AppSlug(context.Context) (string, error) { return "diffx-test", nil } @@ -56,6 +64,21 @@ func (f *fakeGitHub) UserRepositories(_ context.Context, token string) ([]github func (f *fakeGitHub) OpenPullRequests(_ context.Context, _ int64, fullName string) ([]github.PullRequest, error) { return f.open[fullName], nil } +func (f *fakeGitHub) PullRequest(_ context.Context, _ int64, _ string, number int) (github.PullRequest, error) { + f.mu.Lock() + defer f.mu.Unlock() + pull, ok := f.pulls[number] + if !ok { + return github.PullRequest{}, &github.Error{Status: 404, Message: "Not Found"} + } + return pull, nil +} +func (f *fakeGitHub) InstallationToken(_ context.Context, installation int64, repositories []int64, permissions map[string]string) (github.Token, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.tokens = append(f.tokens, tokenGrant{installation, repositories, permissions}) + return github.Token{Value: "installation-token", ExpiresAt: time.Now().Add(time.Hour)}, nil +} // refreshLog records refresh intent while delegating to the real store. type refreshLog struct { diff --git a/internal/githubsync/store.go b/internal/githubsync/store.go index d86beda9..fd5e9cf0 100644 --- a/internal/githubsync/store.go +++ b/internal/githubsync/store.go @@ -3,6 +3,8 @@ package githubsync import ( "context" "errors" + + "github.com/flexdinesh/diffx/internal/ingestion" ) // ErrNotFound reports an absent connection, follow or OAuth state. @@ -68,4 +70,35 @@ type Store interface { RequestRefresh(ctx context.Context, ref PullRequestRef, retentionRun string, now int64) error // KnownOpen lists pull request numbers last observed open. KnownOpen(ctx context.Context, repositoryID int64) ([]int, error) + + // ClaimRefresh leases the oldest pending, available pull request until + // leaseUntil, fencing earlier leases. ErrNotFound means none is pending. + ClaimRefresh(ctx context.Context, token string, now, leaseUntil int64) (Lease, error) + // RecordLease stores what the lease will publish. ErrLeaseLost when fenced. + RecordLease(ctx context.Context, lease Lease, leased []byte, now int64) error + // Leased returns the recorded lease for a current token. ErrLeaseLost when fenced or expired. + Leased(ctx context.Context, repositoryID int64, number int, token string, now int64) (Lease, []byte, error) + // ReleaseRefresh ends a failed lease; the pull request stays pending until retryAt. + ReleaseRefresh(ctx context.Context, lease Lease, retryAt int64, detail string) error + // CompleteRefresh atomically admits every submission, marks the leased + // generation refreshed and records the observed state. A generation + // requested during the lease stays pending. + CompleteRefresh(ctx context.Context, lease Lease, state string, admissions []Admission, now int64) error +} + +var ErrLeaseLost = errors.New("fetch lease lost") + +// Lease is one claimed refresh of a pull request at a generation. +type Lease struct { + Ref PullRequestRef + Token string + Generation int64 + RetentionRun string + Attempt int +} + +// Admission is one account's submission of a fetched snapshot. +type Admission struct { + OwnerID string + Request ingestion.Request } diff --git a/internal/remoteserver/fetch.go b/internal/remoteserver/fetch.go new file mode 100644 index 00000000..a563b31c --- /dev/null +++ b/internal/remoteserver/fetch.go @@ -0,0 +1,66 @@ +package remoteserver + +import ( + "crypto/sha256" + "crypto/subtle" + "encoding/json" + "errors" + "log" + "net/http" + "strings" + + "github.com/flexdinesh/diffx/internal/fetchprotocol" + "github.com/flexdinesh/diffx/internal/githubsync" +) + +// fetchHandler serves the fetcher's lease protocol on the internal listener +// only. The shared secret authenticates the fetcher; it never carries an account. +func fetchHandler(service *githubsync.Service, secret string) http.Handler { + want := sha256.Sum256([]byte(secret)) + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + bearer, ok := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ") + got := sha256.Sum256([]byte(bearer)) + if !ok || subtle.ConstantTimeCompare(got[:], want[:]) != 1 { + problem(w, 401, "Fetcher authentication required") + return + } + if r.Method != http.MethodPost { + problem(w, 405, "Method not allowed") + return + } + switch r.URL.Path { + case fetchprotocol.LeasePath: + job, err := service.Lease(r.Context()) + if err != nil { + log.Printf("GitHub fetch lease: %v", err) + problem(w, 500, "Fetch lease failed") + return + } + if job == nil { + w.WriteHeader(http.StatusNoContent) + return + } + writeJSON(w, 200, job) + case fetchprotocol.ResultPath: + var result fetchprotocol.Result + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, fetchprotocol.MaxResultBytes)).Decode(&result); err != nil { + problem(w, 400, "Invalid fetch result") + return + } + err := service.Finish(r.Context(), result) + switch { + case err == nil: + w.WriteHeader(http.StatusNoContent) + case errors.Is(err, githubsync.ErrLeaseLost): + problem(w, 409, err.Error()) + case errors.Is(err, githubsync.ErrInvalidResult): + problem(w, 400, err.Error()) + default: + log.Printf("GitHub fetch result: %v", err) + problem(w, 500, "Fetch result not applied") + } + default: + problem(w, 404, "Not found") + } + }) +} diff --git a/internal/remoteserver/github.go b/internal/remoteserver/github.go index f56a7da2..50aad7b7 100644 --- a/internal/remoteserver/github.go +++ b/internal/remoteserver/github.go @@ -26,6 +26,9 @@ type GitHubSettings struct { WebhookSecret string TokenKey string PublicURL string + // InternalListen serves the fetcher protocol, authenticated by FetcherSecret. + InternalListen string + FetcherSecret string // apiURL and webURL replace github.com in tests. apiURL, webURL string } diff --git a/internal/remoteserver/github_test.go b/internal/remoteserver/github_test.go index 1e518110..578cafad 100644 --- a/internal/remoteserver/github_test.go +++ b/internal/remoteserver/github_test.go @@ -17,8 +17,11 @@ import ( "path/filepath" "strings" "testing" + "testing/fstest" + "github.com/flexdinesh/diffx/internal/fetchprotocol" "github.com/flexdinesh/diffx/internal/identityservice" + "github.com/flexdinesh/diffx/internal/reviewstore" ) // fakeGitHubAPI serves the API and OAuth endpoints the integration calls. @@ -196,3 +199,48 @@ func TestGitHubCallbackRefusesForeignAndMissingSessions(t *testing.T) { t.Fatalf("bob completed alice's authorization: %s", w.Body.String()) } } + +func TestFetchProtocolOnlyOnInternalListenerWithSecret(t *testing.T) { + store, err := reviewstore.Open("") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { store.Close() }) + settings := githubSettings(t, fakeGitHubAPI(t).URL) + settings.FetcherSecret = strings.Repeat("s", 32) + public, internal, closeServices, _, err := compose(t.Context(), store, fstest.MapFS{"index.html": {Data: []byte("app shell")}}, Settings{SignupEnabled: true, GitHub: settings}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { closeServices() }) + lease := func(handler http.Handler, method, authorization string) *httptest.ResponseRecorder { + r := httptest.NewRequest(method, fetchprotocol.LeasePath, nil) + if authorization != "" { + r.Header.Set("Authorization", authorization) + } + w := httptest.NewRecorder() + handler.ServeHTTP(w, r) + return w + } + if w := lease(public, "POST", "Bearer "+settings.FetcherSecret); w.Code == 200 || w.Code == 204 { + t.Fatalf("public listener served the fetch protocol: %d", w.Code) + } + for _, authorization := range []string{"", "Bearer wrong", settings.FetcherSecret} { + if w := lease(internal, "POST", authorization); w.Code != 401 { + t.Fatalf("lease with %q: %d", authorization, w.Code) + } + } + if w := lease(internal, "GET", "Bearer "+settings.FetcherSecret); w.Code != 405 { + t.Fatalf("GET lease: %d", w.Code) + } + if w := lease(internal, "POST", "Bearer "+settings.FetcherSecret); w.Code != 204 { + t.Fatalf("idle lease: %d %s", w.Code, w.Body.String()) + } + r := httptest.NewRequest("POST", fetchprotocol.ResultPath, strings.NewReader(`{"id":"10-1","leaseToken":"stale","error":"x"}`)) + r.Header.Set("Authorization", "Bearer "+settings.FetcherSecret) + w := httptest.NewRecorder() + internal.ServeHTTP(w, r) + if w.Code != 409 { + t.Fatalf("result without a lease: %d", w.Code) + } +} diff --git a/internal/remoteserver/server.go b/internal/remoteserver/server.go index ede873b5..766ee867 100644 --- a/internal/remoteserver/server.go +++ b/internal/remoteserver/server.go @@ -79,7 +79,7 @@ func MultiHandler(ctx context.Context, store *reviewstore.Store, assets fs.FS, o settings = options[0] } ctx, cancel := context.WithCancel(ctx) - handler, closeServices, tasks, err := compose(ctx, store, assets, settings) + handler, _, closeServices, tasks, err := compose(ctx, store, assets, settings) if err != nil { cancel() return nil, nil, err @@ -101,14 +101,17 @@ func MultiHandler(ctx context.Context, store *reviewstore.Store, assets fs.FS, o return closeErr }, nil } -func compose(ctx context.Context, store *reviewstore.Store, assets fs.FS, settings Settings) (http.Handler, func() error, []serverapp.Task, error) { + +// compose returns the public handler and, when GitHub is configured, the +// internal fetcher handler. +func compose(ctx context.Context, store *reviewstore.Store, assets fs.FS, settings Settings) (http.Handler, http.Handler, func() error, []serverapp.Task, error) { identity := identityservice.New(store) if err := bootstrap(ctx, identity, settings); err != nil { - return nil, nil, nil, err + return nil, nil, nil, nil, err } github, err := composeGitHub(store, settings.GitHub) if err != nil { - return nil, nil, nil, err + return nil, nil, nil, nil, err } applications := ®istry{ctx: ctx, store: store, assets: assets, applications: make(map[string]application)} remote := &transport{identity: identity, applications: applications, settings: settings, assets: assets, limiter: newLimiter(), authSlots: make(chan struct{}, 4), github: github} @@ -132,10 +135,12 @@ func compose(ctx context.Context, store *reviewstore.Store, assets fs.FS, settin }) } tasks := []serverapp.Task{worker} + var internal http.Handler if github != nil { tasks = append(tasks, reconcileAtStartup(github)) + internal = fetchHandler(github, settings.GitHub.FetcherSecret) } - return remote, applications.close, tasks, nil + return remote, internal, applications.close, tasks, nil } func bootstrap(ctx context.Context, identity *identityservice.Service, settings Settings) error { if (settings.Username == "") != (settings.Password == "") { @@ -174,7 +179,7 @@ func Run(ctx context.Context, settings Settings, ready func(string)) error { if err := store.PruneExpired(time.Now()); err != nil { return err } - handler, closeServices, tasks, err := compose(ctx, store, webui.Assets(), settings) + handler, internal, closeServices, tasks, err := compose(ctx, store, webui.Assets(), settings) if err != nil { return err } @@ -186,7 +191,19 @@ func Run(ctx context.Context, settings Settings, ready func(string)) error { defer listener.Close() server := &http.Server{Handler: handler, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 2 * time.Minute, IdleTimeout: 60 * time.Second} defer server.Close() - return serverapp.Run(ctx, []serverapp.Endpoint{{Server: server, Listener: listener}}, append(tasks, serverapp.Prune(store.PruneExpired)), func() { + endpoints := []serverapp.Endpoint{{Server: server, Listener: listener}} + if internal != nil { + // The fetcher protocol never shares the public listener. + internalListener, err := net.Listen("tcp", settings.GitHub.InternalListen) + if err != nil { + return err + } + defer internalListener.Close() + internalServer := &http.Server{Handler: internal, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 2 * time.Minute, IdleTimeout: 60 * time.Second} + defer internalServer.Close() + endpoints = append(endpoints, serverapp.Endpoint{Server: internalServer, Listener: internalListener}) + } + return serverapp.Run(ctx, endpoints, append(tasks, serverapp.Prune(store.PruneExpired)), func() { if ready != nil { ready(listener.Addr().String()) } diff --git a/internal/review/model_test.go b/internal/review/model_test.go index de852de1..e18d37b1 100644 --- a/internal/review/model_test.go +++ b/internal/review/model_test.go @@ -97,3 +97,11 @@ func TestMovedCommentExportsCurrentLines(t *testing.T) { t.Fatalf("moved comment must report current lines: %s", output) } } + +func TestRemoteRepositoryKeyMatchesEnrolledIdentity(t *testing.T) { + // Collectors enrolled this key before the rule was shared; it must not drift. + const enrolled = "b11f5fec12c6f0510b7b34d163a1a59584a49a42a6dc76dcf6e3d7f5de56c799" + if got := RemoteRepositoryKey("https://github.com/octo/app.git"); got != enrolled { + t.Fatalf("key %q, want %q", got, enrolled) + } +} diff --git a/internal/review/repository_name.go b/internal/review/repository_name.go index 65c0e7d4..b11331ae 100644 --- a/internal/review/repository_name.go +++ b/internal/review/repository_name.go @@ -1,11 +1,21 @@ package review import ( + "crypto/sha256" + "encoding/hex" "net/url" "path" "strings" ) +// RemoteRepositoryKey is the initial repository identity enrolled for a +// sanitized remote URL. Collectors and the server's GitHub integration share +// it, so pull requests group with checkouts cloned from the same URL. +func RemoteRepositoryKey(remote string) string { + digest := sha256.Sum256([]byte("remote\x00" + remote)) + return hex.EncodeToString(digest[:]) +} + func RemoteRepositoryName(remote, fallback string) string { if remote == "" { return fallback diff --git a/internal/reviewstore/github.go b/internal/reviewstore/github.go index 356fb4d4..d6c87df6 100644 --- a/internal/reviewstore/github.go +++ b/internal/reviewstore/github.go @@ -22,7 +22,10 @@ func initializeGitHub(transaction *sql.Tx) error { // One row per pull request coalesces refresh intent: webhooks and // reconciliation advance generation; the refresh task records the last // generation it processed and the state it observed. - `CREATE TABLE IF NOT EXISTS github_pull_requests (repository_id INTEGER NOT NULL, number INTEGER NOT NULL, installation_id INTEGER NOT NULL, full_name TEXT NOT NULL, generation INTEGER NOT NULL, refreshed_generation INTEGER NOT NULL DEFAULT 0, state TEXT NOT NULL DEFAULT 'open', retention_run TEXT NOT NULL DEFAULT '', requested_at INTEGER NOT NULL, PRIMARY KEY(repository_id, number))`, + // A lease fences the fetcher by token; leased holds what it will publish. + `CREATE TABLE IF NOT EXISTS github_pull_requests (repository_id INTEGER NOT NULL, number INTEGER NOT NULL, installation_id INTEGER NOT NULL, full_name TEXT NOT NULL, generation INTEGER NOT NULL, refreshed_generation INTEGER NOT NULL DEFAULT 0, state TEXT NOT NULL DEFAULT 'open', retention_run TEXT NOT NULL DEFAULT '', requested_at INTEGER NOT NULL, + lease_token TEXT NOT NULL DEFAULT '', lease_until INTEGER NOT NULL DEFAULT 0, leased_generation INTEGER NOT NULL DEFAULT 0, leased_run TEXT NOT NULL DEFAULT '', leased TEXT NOT NULL DEFAULT '', + attempt INTEGER NOT NULL DEFAULT 0, available_at INTEGER NOT NULL DEFAULT 0, detail TEXT NOT NULL DEFAULT '', PRIMARY KEY(repository_id, number))`, `CREATE INDEX IF NOT EXISTS github_pull_requests_pending ON github_pull_requests(requested_at) WHERE generation>refreshed_generation`, } for _, statement := range statements { @@ -202,6 +205,78 @@ func (s *GitHubStore) KnownOpen(ctx context.Context, repositoryID int64) ([]int, return result, rows.Err() } +func (s *GitHubStore) ClaimRefresh(ctx context.Context, token string, now, leaseUntil int64) (githubsync.Lease, error) { + var lease githubsync.Lease + err := s.store.db.QueryRowContext(ctx, `UPDATE github_pull_requests SET lease_token=?,lease_until=?,leased_generation=generation,leased_run=retention_run,leased='' + WHERE rowid=(SELECT rowid FROM github_pull_requests WHERE generation>refreshed_generation AND lease_until<=? AND available_at<=? ORDER BY requested_at,repository_id,number LIMIT 1) + RETURNING repository_id,installation_id,full_name,number,leased_generation,leased_run,attempt`, token, leaseUntil, now, now). + Scan(&lease.Ref.RepositoryID, &lease.Ref.InstallationID, &lease.Ref.FullName, &lease.Ref.Number, &lease.Generation, &lease.RetentionRun, &lease.Attempt) + if errors.Is(err, sql.ErrNoRows) { + return lease, githubsync.ErrNotFound + } + lease.Token = token + return lease, err +} + +func (s *GitHubStore) RecordLease(ctx context.Context, lease githubsync.Lease, leased []byte, now int64) error { + result, err := s.store.db.ExecContext(ctx, `UPDATE github_pull_requests SET leased=? WHERE repository_id=? AND number=? AND lease_token=? AND lease_until>?`, + string(leased), lease.Ref.RepositoryID, lease.Ref.Number, lease.Token, now) + return fenced(result, err) +} + +func (s *GitHubStore) Leased(ctx context.Context, repositoryID int64, number int, token string, now int64) (githubsync.Lease, []byte, error) { + lease := githubsync.Lease{Token: token} + var leased string + err := s.store.db.QueryRowContext(ctx, `SELECT repository_id,installation_id,full_name,number,leased_generation,leased_run,attempt,leased FROM github_pull_requests + WHERE repository_id=? AND number=? AND lease_token=? AND lease_until>? AND leased<>''`, repositoryID, number, token, now). + Scan(&lease.Ref.RepositoryID, &lease.Ref.InstallationID, &lease.Ref.FullName, &lease.Ref.Number, &lease.Generation, &lease.RetentionRun, &lease.Attempt, &leased) + if errors.Is(err, sql.ErrNoRows) { + return lease, nil, githubsync.ErrLeaseLost + } + return lease, []byte(leased), err +} + +func (s *GitHubStore) ReleaseRefresh(ctx context.Context, lease githubsync.Lease, retryAt int64, detail string) error { + result, err := s.store.db.ExecContext(ctx, `UPDATE github_pull_requests SET lease_token='',lease_until=0,leased='',attempt=attempt+1,available_at=?,detail=? WHERE repository_id=? AND number=? AND lease_token=?`, + retryAt, detail, lease.Ref.RepositoryID, lease.Ref.Number, lease.Token) + return fenced(result, err) +} + +func (s *GitHubStore) CompleteRefresh(ctx context.Context, lease githubsync.Lease, state string, admissions []githubsync.Admission, now int64) error { + transaction, err := s.store.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer transaction.Rollback() + // Fence first, so a lost lease admits nothing. + result, err := transaction.ExecContext(ctx, `UPDATE github_pull_requests SET refreshed_generation=MAX(refreshed_generation,leased_generation),state=?, + retention_run=CASE WHEN retention_run=leased_run THEN '' ELSE retention_run END,lease_token='',lease_until=0,leased='',attempt=0,available_at=0,detail='' + WHERE repository_id=? AND number=? AND lease_token=? AND lease_until>?`, state, lease.Ref.RepositoryID, lease.Ref.Number, lease.Token, now) + if err := fenced(result, err); err != nil { + return err + } + for _, admission := range admissions { + if _, err := acceptJob(transaction, admission.OwnerID, admission.Request); err != nil { + return err + } + } + return transaction.Commit() +} + +func fenced(result sql.Result, err error) error { + if err != nil { + return err + } + rows, err := result.RowsAffected() + if err != nil { + return err + } + if rows != 1 { + return githubsync.ErrLeaseLost + } + return nil +} + // pruneGitHub removes expired OAuth states and delivery identities older than // GitHub's redelivery window with margin. func pruneGitHub(transaction *sql.Tx, now int64) error { diff --git a/internal/reviewstore/queue.go b/internal/reviewstore/queue.go index 4bc1c6bf..1176207e 100644 --- a/internal/reviewstore/queue.go +++ b/internal/reviewstore/queue.go @@ -58,6 +58,21 @@ type SQLiteQueue struct{ store *Store } func (store *Store) Queue() *SQLiteQueue { return &SQLiteQueue{store: store} } func (queue *SQLiteQueue) Accept(ctx context.Context, owner string, request ingestion.Request) (ingestionqueue.Job, error) { + tx, err := queue.store.db.BeginTx(ctx, nil) + if err != nil { + return ingestionqueue.Job{}, err + } + defer tx.Rollback() + job, err := acceptJob(tx, owner, request) + if err != nil { + return job, err + } + return job, tx.Commit() +} + +// acceptJob admits within the caller's transaction, so other state can commit +// atomically with durable admission. +func acceptJob(tx *sql.Tx, owner string, request ingestion.Request) (ingestionqueue.Job, error) { if err := ingestion.Validate(request); err != nil { return ingestionqueue.Job{}, err } @@ -68,18 +83,13 @@ func (queue *SQLiteQueue) Accept(ctx context.Context, owner string, request inge if len(raw) > ingestion.MaxRequestBytes { return ingestionqueue.Job{}, ingestionqueue.ErrFull } - tx, err := queue.store.db.BeginTx(ctx, nil) - if err != nil { - return ingestionqueue.Job{}, err - } - defer tx.Rollback() sequence, contextID, err := recordSubmission(tx, owner, request, hash) if err != nil { return ingestionqueue.Job{}, err } job, err := scanJob(tx.QueryRow(jobSelect+` WHERE j.sequence=?`, sequence)) if err == nil { - return job, tx.Commit() + return job, nil } if !errors.Is(err, sql.ErrNoRows) { return job, err @@ -102,7 +112,7 @@ func (queue *SQLiteQueue) Accept(ctx context.Context, owner string, request inge return job, err } job = ingestionqueue.Job{ID: id, SubmissionID: request.SubmissionID, State: state, ContextID: contextID} - return job, tx.Commit() + return job, nil } const jobSelect = `SELECT j.id,r.submission_id,j.state,j.attempt,j.context_id,j.detail FROM ingestion_jobs j JOIN ingestion_records r ON r.sequence=j.sequence` diff --git a/tools/check-boundaries/main.go b/tools/check-boundaries/main.go index a637657f..6b3a38cb 100644 --- a/tools/check-boundaries/main.go +++ b/tools/check-boundaries/main.go @@ -22,6 +22,7 @@ var allowed = map[string][]string{ "internal/diffsource", "internal/hooks", "internal/ingestion", "internal/localserver", "internal/review", "internal/reviewstore", "internal/submission", "internal/version", }, + "cmd/diffx-fetcher": {"internal/config", "internal/fetcher", "internal/fetchprotocol", "internal/version"}, "cmd/diffx-server": {"internal/config", "internal/identityservice", "internal/remoteserver", "internal/reviewstore", "internal/version"}, "internal/browser": {}, "internal/collector": {"internal/diffsource", "internal/ingestion", "internal/review"}, @@ -31,8 +32,11 @@ var allowed = map[string][]string{ "internal/diffsource": {"internal/processlock", "internal/review"}, // GitHub wire adapter; integration policy lives in githubsync. "internal/github": {}, - "internal/githubsync": {"internal/github"}, - "internal/hooks": {"internal/ingestion", "internal/processlock", "internal/submission"}, + "internal/githubsync": {"internal/fetchprotocol", "internal/github", "internal/ingestion", "internal/review"}, + // The fetcher mirrors GitHub repositories and collects them offline. + "internal/fetcher": {"internal/collector", "internal/diffsource", "internal/fetchprotocol", "internal/ingestion"}, + "internal/fetchprotocol": {"internal/ingestion"}, + "internal/hooks": {"internal/ingestion", "internal/processlock", "internal/submission"}, "internal/httpapi": { "internal/review", "internal/reviewdata", "internal/reviewsession", "internal/contextservice", "internal/reviewservice", "internal/ingestion", "internal/processmetrics", "packages/api", @@ -48,7 +52,7 @@ var allowed = map[string][]string{ "internal/processlock": {}, "internal/processmetrics": {}, "internal/remoteserver": { - "internal/contextservice", "internal/github", "internal/githubsync", "internal/identityservice", "internal/ingestion", "internal/ingestionqueue", "internal/review", + "internal/contextservice", "internal/fetchprotocol", "internal/github", "internal/githubsync", "internal/identityservice", "internal/ingestion", "internal/ingestionqueue", "internal/review", "internal/reviewstore", "internal/serverapp", "internal/webui", "packages/api", }, "internal/review": {}, @@ -84,7 +88,7 @@ func checkPackage(pkg listedPackage) []string { var violations []string for _, dependency := range pkg.Imports { if dependency == "os/exec" && !slices.Contains([]string{ - "internal/browser", "internal/diffsource", "internal/hooks", "internal/localserver", + "internal/browser", "internal/diffsource", "internal/fetcher", "internal/hooks", "internal/localserver", }, name) { violations = append(violations, fmt.Sprintf("%s must not execute processes", name)) }