diff --git a/README.md b/README.md index 9fd7ae0c..7488818e 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,7 @@ Octopool owns pooled live `gh` reads. Gitcrawl keeps local mirror, search, clust | Check archive health | `gitcrawl status` / `gitcrawl doctor` | [Configuration](docs/configuration.md) | | Refresh a portable subscriber | `gitcrawl portable refresh --expected-remote URL` | [Portable stores](docs/portable-stores.md#routine-subscriber-refresh) | | Mirror GitHub threads | `gitcrawl sync owner/repo` | [Sync](docs/sync.md) | +| Collect repository counters and releases | `gitcrawl metrics collect\|import\|status --config metrics.json` | [Repository metrics](docs/metrics.md) | | Search threads or indexed code | `gitcrawl search ...` | [Search](docs/search.md) | | Build and inspect clusters | `gitcrawl refresh`, `clusters`, `tui` | [Clustering](docs/clustering.md) | | Export a code-free conversation snapshot | `gitcrawl capture owner/repo` | [Capture](docs/capture.md) | diff --git a/docs/commands.md b/docs/commands.md index aa4f11ef..3d3b740e 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -38,6 +38,17 @@ These work on every command. | `gitcrawl configure [--summary-model --embed-model --embedding-basis --json]` | Update model fields in `config.toml` | [Configuration](/configuration/#gitcrawl-configure) | | `gitcrawl version` | Print version | — | +## Repository metrics + +| Command | Purpose | Docs | +| --- | --- | --- | +| `gitcrawl metrics collect --config metrics.json [--json]` | Observe stars, forks, actual watchers, open PRs/issues, optional daily clones, and stable releases in a separate database | [Repository metrics](/metrics/) | +| `gitcrawl metrics import --config metrics.json [--json]` | Atomically import scoped NDJSON history from stdin, preserving NULLs and IDs | [Repository metrics](/metrics/#storage-imports-and-failures) | +| `gitcrawl metrics status --config metrics.json [--json]` | Inspect the metrics database without writes or network calls | [Repository metrics](/metrics/) | + +For `metrics`, `--config` selects an independent JSON config; it never selects or +initializes the normal thread archive. No embedding or model calls are made. + ## Sync | Command | Purpose | Docs | diff --git a/docs/installation.md b/docs/installation.md index c5bf7f04..23bb9b75 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -76,6 +76,11 @@ go build \ Symlink or copy `bin/gitcrawl` somewhere on your `PATH` (`~/bin`, `/usr/local/bin`, `~/.local/bin`). +For a metrics-only source deployment alongside an existing portable subscriber, +use an [isolated versioned metrics runtime](/metrics/#isolated-source-built-installation) +and verify it with `metrics status` and its separate config. Keep the subscriber's +binary selection and refresh job unchanged. + Source builds use scalar vector scoring by default. Set `GOEXPERIMENT=simd` on `go build` to enable Go 1.27's experimental portable SIMD kernels for clustering and exact neighbors. The API may change in Go 1.28; unsupported hardware and diff --git a/docs/metrics.md b/docs/metrics.md new file mode 100644 index 00000000..9b803c12 --- /dev/null +++ b/docs/metrics.md @@ -0,0 +1,303 @@ +--- +title: Repository metrics +nav_order: 16 +permalink: /metrics/ +--- + +# Repository metrics + +`gitcrawl metrics` collects repository headline counters and stable release events +into a separate, private SQLite database. It does not sync the thread archive, +refresh a portable store, generate embeddings, call a model, or start a scheduler. + +## Configuration and commands + +Create a metrics JSON config. `database` must be an absolute filesystem path to a +new file or an existing Gitcrawl metrics database, outside your archive and portable +store. The example path is illustrative; choose a private directory on your host. + +```json +{ + "database": "/private/metrics/gitcrawl/metrics.sqlite", + "targets": [ + {"entity": "OpenClaw", "target": "openclaw/openclaw"}, + {"entity": "Example", "target": "example/project"} + ] +} +``` + +```sh +gitcrawl metrics collect --config /private/metrics/github.json --json +gitcrawl metrics import --config /private/metrics/github.json --json < history.ndjson +gitcrawl metrics status --config /private/metrics/github.json --json +gitcrawl help metrics +``` + +The config is independent of `config.toml`, `GITCRAWL_CONFIG`, and `GITCRAWL_DB_PATH`. For these commands, +`--config` selects the metrics JSON file; it can appear before `metrics` or after +the subcommand. Global output flags and command-local `--json` work normally. + +Public counters and releases can be read without authentication. Authenticated +collection uses `GITHUB_TOKEN`, then the normal native `gh auth token` resolver. +An optional `tokenEnv` selects another environment variable. A global +`--github-token-command /absolute/executable` selects the normal managed credential +provider exclusively on supported platforms. Tokens never appear in results. +The metrics config contains no token values; cookie authentication is not used. + +## Isolated source-built installation + +A source-built metrics runtime can coexist with an official Gitcrawl installation +that refreshes a portable mirror. Give the metrics binary a private, versioned +directory and invoke that exact executable. The installed metrics CLI path is: + +```text +$HOME/.local/share/gitcrawl/metrics-runtimes//gitcrawl +``` + +Copy an already validated artifact into a new directory; do not overwrite an +existing version. Record its full source commit and expected SHA-256 from the +validation handoff, verify the hash before and after copying, and use directory +mode `0700` and executable mode `0500`. This installation does not replace a +`current` symlink, the command on `PATH`, an archive config, or a refresh job. + +Local source builds do not require official release signing credentials under the +[installation policy](/installation/#install-from-source). On macOS, verify the +source artifact's signature with `codesign --verify --strict` before and after +copying. This is source-build verification, not official release notarization; +official release signing and notarization remain the [release workflow's](/releasing/) +responsibility. Do not change signing policies or remove quarantine to force an +untrusted artifact to run. + +The installation check is read-only and uses the separately provided metrics config: + +```sh +metrics_revision=SOURCE_COMMIT +metrics_binary="$HOME/.local/share/gitcrawl/metrics-runtimes/$metrics_revision/gitcrawl" +"$metrics_binary" --version +"$metrics_binary" metrics status \ + --config "$HOME/.local/share/gitcrawl/metrics.json" --json +``` + +Keep a machine-local installation receipt at the path returned by +`git rev-parse --git-path metrics-runtime-installation.json`. Record the exact +resolved CLI/config/database paths, source commit, artifact hash, signature result, +read-only status, and preservation checks there. Git metadata keeps these private +host details out of the public documentation and PR. The coordinator's handoff +should contain the same exact CLI path. Installation alone does not authorize +collection, imports, scheduling, or a final cutover. + +### Local signing for background collection + +A directly launched macOS job has its own file-access identity; permission granted +to a terminal does not prove the background executable has access. When the owner +authorizes an existing local signing certificate, sign a copy of the verified +runtime with a stable identifier and an explicit requirement bound to that +certificate. Use the already selected public SHA-1 fingerprint, without searching +for another identity or changing keychain trust or access controls. + +```sh +metrics_revision=SOURCE_COMMIT +metrics_identity=SELECTED_PUBLIC_CERTIFICATE_SHA1 +metrics_identifier=com.example.gitcrawl.metrics +metrics_source="$HOME/.local/share/gitcrawl/metrics-runtimes/$metrics_revision/gitcrawl" +metrics_install="$HOME/.local/libexec/gitcrawl-metrics/$metrics_revision" +metrics_requirement="identifier \"$metrics_identifier\" and certificate leaf = H\"$metrics_identity\"" +codesign --verify --strict "$metrics_source" +# Check the source SHA-256 against the validation receipt before copying. +umask 077 +mkdir -p "$(dirname "$metrics_install")" +mkdir "$metrics_install" +cp "$metrics_source" "$metrics_install/gitcrawl" +chmod 0700 "$metrics_install/gitcrawl" +codesign --force --sign "$metrics_identity" --identifier "$metrics_identifier" \ + --requirements "=designated => $metrics_requirement" --timestamp=none \ + "$metrics_install/gitcrawl" +codesign --verify --strict --test-requirement "=$metrics_requirement" \ + "$metrics_install/gitcrawl" +codesign --display --requirements - "$metrics_install/gitcrawl" +chmod 0500 "$metrics_install/gitcrawl" +shasum -a 256 "$metrics_source" "$metrics_install/gitcrawl" +``` + +Record both hashes, the source commit, certificate fingerprint, identifier and +requirement in the private receipt. Preserve the original runtime. Point only the +metrics LaunchAgent's first argument at the signed copy, preserving its other +settings. If this job was disabled, enable its exact label before bootstrapping. +Request one collection and let the user approve normal macOS file-access prompts. +Follow the specific permission request through its prompt and decision: a default +`SystemPolicyAllFiles` probe denial can precede a normal +`SystemPolicyRemovableVolumes` prompt. That probe alone does not establish a Full +Disk Access requirement. Keep the same authorized attempt running while the user +answers its normal prompt, then verify the terminal result and metrics database. +If permission remains blocked, stop and disable that job and report the evidence; +do not retry repeatedly, grant Full Disk Access, export keys, add privileged +wrappers, or relocate data. Local signing is not official release notarization. + +## Hourly collection on macOS + +After authorizing local collection, create a separate user LaunchAgent that calls +the pinned native binary directly. Use absolute paths; launchd does not expand +`~`, `$HOME`, or shell variables in a plist. Keep the config and logs outside Git +and shared publication. Give their directories mode `0700` and files mode `0600`. +Create both log files before bootstrapping the job. + +The following is a template for `~/Library/LaunchAgents/org.openclaw.gitcrawl.metrics.plist`. +Replace `/Users/you` and `SOURCE_COMMIT` with your actual installation paths: + +```xml + + + + + Labelorg.openclaw.gitcrawl.metrics + ProgramArguments + + /Users/you/.local/share/gitcrawl/metrics-runtimes/SOURCE_COMMIT/gitcrawl + metricscollect + --config/Users/you/.local/share/gitcrawl/metrics.json + --json + + WorkingDirectory/Users/you/.local/share/gitcrawl + EnvironmentVariables + + HOME/Users/you + PATH/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin + GITCRAWL_NO_UPDATE_CHECK1 + + StartCalendarIntervalMinute6 + KeepAlive + Umask63 + StandardOutPath/Users/you/.local/share/gitcrawl/metrics-logs/stdout.log + StandardErrorPath/Users/you/.local/share/gitcrawl/metrics-logs/stderr.log + + +``` + +The job runs hourly at minute 6 in the host's local time, while observations use +UTC. `63` is the decimal representation of umask `0077`. The explicit minimal +`PATH` supports the existing native `gh` credential resolver without interactive +shell initialization. Do not put credentials in the plist. The versioned runtime +is not auto-updated; `GITCRAWL_NO_UPDATE_CHECK` also disables release notices. + +Validate and bootstrap this new job once, then request one immediate collection: + +```sh +metrics_plist="$HOME/Library/LaunchAgents/org.openclaw.gitcrawl.metrics.plist" +plutil -lint "$metrics_plist" +launchctl bootstrap "gui/$(id -u)" "$metrics_plist" +launchctl kickstart "gui/$(id -u)/org.openclaw.gitcrawl.metrics" +launchctl print "gui/$(id -u)/org.openclaw.gitcrawl.metrics" +``` + +Wait for the job to exit and check its last exit code, private logs, and the latest +`metric_runs` row using read-only SQLite. Confirm all five required counters for +every configured target have non-NULL values at that run's timestamp. A successful +bootstrap or a PID alone is not collection evidence. An unavailable optional clone +report is not a required-counter failure. On a provider failure, inspect the +recorded result before retrying; `KeepAlive` is disabled to avoid rapid restarts. +Existing archive refresh jobs are independent and need no changes. + +`collect` and `import` hold a nonblocking native OS lock on +`.writer.lock` from before database initialization through database +close. A second writer exits nonzero before collection or import; `status` remains +read-only and available. Ownership is released even if the process is killed. +The private lock file remains in place: never delete or replace it while writers +can run. Use the locking runtime for every writer; older binaries do not honor +this lock. + +## What is collected + +Each invocation observes all configured targets at a single UTC timestamp. Run +`collect` hourly with an external scheduler when hourly history is required. + +| Metric | Meaning | +| --- | --- | +| `stars` | Repository `stargazers_count` | +| `forks` | Repository `forks_count`; changes between snapshots are net changes | +| `watchers` | Actual subscribers, `subscribers_count`; **not** `watchers_count`, which aliases stars | +| `open_prs` | Open pull-request search count, only when results are complete | +| `open_issues` | Combined repository issue/PR count minus the valid open-PR count | +| `clones` | Optional daily clone counts from the authenticated traffic endpoint | + +Missing, invalid, or incomplete required counts remain SQL `NULL`; they never +become zero. If the PR count is unavailable or exceeds the combined count, +`open_issues` remains unknown. Real zeroes and decreases are retained. Individual +fork creation events are not crawled. + +Clone traffic requires repository push access (or a fine-grained token with repository +administration read permission). Permission-denied HTTP 403/404 means optional +unavailability and does not fail otherwise healthy collection. A rate-limit +response is a collection failure, including on the traffic endpoint. Only completed UTC +days are recorded: `ts` is that day's final millisecond and `observed_at` is the +actual read time. Unchanged daily values are not re-appended; corrected values +receive a new sequence. Imported daily timestamps retain their original spelling; +group them by UTC day. Sum only the latest observation for each day, never all +revisions. GitHub's traffic window limits how far a missed day can be backfilled. + +Stable releases exclude drafts and prereleases. All release pages are read; +events use stable GitHub release IDs so repeated collection is idempotent. +`published_at` is preferred, with `created_at` as the fallback for older records. + +Each target costs one repository request, one search request for the open-PR +count, optionally one traffic request, and one request per 100-release page +(including a final empty page when the total is a multiple of 100). Every +collection rereads the release history. Search has its own GitHub quota; there +is no metrics-specific quota reserve or incremental release checkpoint. The +shared client retries a rate-limited request once with a wait capped at five +minutes. An exhausted rate limit stops collection, retaining completed reads and +leaving later targets unattempted. Choose the schedule and target count accordingly. + +## Storage, imports, and failures + +New database files are private (`0600`). Existing files must identify themselves +with `metric_meta.owner = gitcrawl` and `metric_meta.version = 1`. Databases with +foreign tables, another owner/version, database symlinks or hard-link aliases, and +pre-existing empty files are rejected before a writable open. Existing databases +are inspected read-only for this check; no archive runtime or config is loaded. +Before applying the metrics schema or ownership metadata, the writable connection +rechecks ownership under a write transaction. Newly created databases must still +be empty, and changed file identities are rejected. Schema and ownership metadata +commit together on that same connection. +A failed first initialization removes only the newly created file so it can be +retried; pre-existing files are never removed. A process killed during that first +initialization can still leave an unowned file requiring operator inspection. `status` checks identity read-only and +does not create a missing database. Never point this config at the thread archive. + +The delivery tables are: + +- `metric_observations(sequence, id, entity, target, metric, kind, ts, value, + observed_at, provenance)` — counters and daily observations; `value` is nullable. +- `metric_events(sequence, id, entity, target, kind, ts, label, url, observed_at, + provenance)` — release history. +- `metric_runs(sequence, ts, status, rows_written)` — completed collection attempts. + +`status` reports total observations and events. Its `last_observed` is the latest +observation instant, comparing parsed timestamps even when imported offsets or +fractional precision differ; it is absent until an observation exists. + +Observation and event sequences advance independently. Read each table using its +own delivery cursor. Daily revisions supersede by latest sequence. Counter values +are snapshots, not increments; derive net change from consecutive observations. + +Import accepts one JSON object per line on stdin: + +```json +{"type":"metric","id":"history:github:watchers:1","entity":"OpenClaw","target":"openclaw/openclaw","metric":"watchers","kind":"counter","ts":"2026-09-14T00:00:00Z","value":null,"observed_at":"2026-09-15T00:00:00Z","provenance":"historical-import"} +{"type":"event","id":"history:github:release:1","entity":"Example","target":"example/project","kind":"release","ts":"2026-09-14T00:00:00Z","label":"v1","url":"https://github.com/example/project/releases/tag/v1","observed_at":"2026-09-15T00:00:00Z","provenance":"historical-import"} +``` + +IDs are required and idempotent within each destination table. Imported explicit +IDs preserve distinct observations even when values match. `entity` and `target` +must match a configured pair. Config files are limited to 1 MiB. Import validates every row and commits the whole +input atomically; a malformed or out-of-scope late row rolls everything back. +Daily imports must identify a day completed before `observed_at`'s UTC day. + +On a partial source failure, successful reads and explicit unknown counter values +are committed together with a `partial` run; stdout contains the result and the +command exits nonzero. A canceled collection gets a bounded opportunity to retain +already completed reads, without starting another network request. Diagnostics +never include GitHub response bodies or credential output. + +This command does not install schedules, migrate another application's history, +change existing refresh jobs, or publish the private database. Those are explicit +operator/integration responsibilities. diff --git a/internal/cli/app.go b/internal/cli/app.go index 0a390fc7..df486939 100644 --- a/internal/cli/app.go +++ b/internal/cli/app.go @@ -30,6 +30,7 @@ type App struct { observedGitHubToken string Stdout io.Writer Stderr io.Writer + Stdin io.Reader configPath string format OutputFormat @@ -53,6 +54,7 @@ func New() *App { return &App{ Stdout: os.Stdout, Stderr: os.Stderr, + Stdin: os.Stdin, format: FormatText, getWorkingDirectory: os.Getwd, } @@ -126,6 +128,8 @@ func (a *App) Run(ctx context.Context, args []string) error { return a.runDoctor(ctx, rest[1:]) case "status": return a.runStatus(ctx, rest[1:]) + case "metrics": + return a.runMetrics(ctx, rest[1:]) case "sync": return a.runSync(ctx, rest[1:]) case "fill-pr-details": diff --git a/internal/cli/control.go b/internal/cli/control.go index 90863dad..af94af7a 100644 --- a/internal/cli/control.go +++ b/internal/cli/control.go @@ -37,9 +37,12 @@ func (a *App) runMetadata(args []string) error { DefaultCache: cfg.CacheDir, DefaultLogs: cfg.LogDir, } - manifest.Capabilities = []string{"metadata", "status", "doctor", "sync", "capture", "coverage", "search", "code-index", "tui", "portable", "remote", "cloud-publish", "clusters", "summaries", "embeddings"} + manifest.Capabilities = []string{"metadata", "status", "metrics", "doctor", "sync", "capture", "coverage", "search", "code-index", "tui", "portable", "remote", "cloud-publish", "clusters", "summaries", "embeddings"} manifest.Privacy = control.Privacy{ContainsPrivateMessages: true, ExportsSecrets: false, LocalOnlyScopes: []string{"github", "git", "sqlite", "portable"}} manifest.Commands = map[string]control.Command{ + "metrics-collect": {Title: "Collect repository metrics", Argv: []string{"gitcrawl", "metrics", "collect", "--config", "METRICS_CONFIG", "--json"}, JSON: true, Mutates: true}, + "metrics-import": {Title: "Import metric history from stdin", Argv: []string{"gitcrawl", "metrics", "import", "--config", "METRICS_CONFIG", "--json"}, JSON: true, Mutates: true}, + "metrics-status": {Title: "Metrics database status", Argv: []string{"gitcrawl", "metrics", "status", "--config", "METRICS_CONFIG", "--json"}, JSON: true}, "status": {Title: "Status", Argv: []string{"gitcrawl", "status", "--json"}, JSON: true}, "remote-status": {Title: "Remote archive status", Argv: []string{"gitcrawl", "remote", "status", "--json"}, JSON: true}, "remote-archives": {Title: "Remote archive list", Argv: []string{"gitcrawl", "remote", "archives", "--json"}, JSON: true}, diff --git a/internal/cli/help.go b/internal/cli/help.go index a9d9c4d1..95464122 100644 --- a/internal/cli/help.go +++ b/internal/cli/help.go @@ -45,6 +45,7 @@ Core commands: metadata print crawlkit control metadata check-update check for a newer gitcrawl release status print fast read-only archive status + metrics collect, import, or inspect an independent metrics database remote status print remote archive status remote archives list remote archives visible to the current identity remote login authenticate with GitHub org access for a remote archive @@ -90,6 +91,19 @@ No API server is provided. There is intentionally no serve command. ` var commandUsageTexts = map[string]string{ + "metrics": `gitcrawl metrics collects repository headline metrics in a separate SQLite database. + +Usage: + gitcrawl metrics collect --config /absolute/metrics.json [--json] + gitcrawl metrics import --config /absolute/metrics.json [--json] < history.ndjson + gitcrawl metrics status --config /absolute/metrics.json [--json] + +The JSON config requires database (absolute path) and targets [{entity,target}]. +Optional tokenEnv overrides GITHUB_TOKEN; native gh auth and --github-token-command +are supported. No archive, portable-store, embedding, or model operations occur. +Import validates target scope and IDs atomically. Unknown values remain SQL NULL. +Status is read-only. Clone traffic 403/404 is optional unavailability. +`, "metadata": `gitcrawl metadata prints crawlkit control metadata. Usage: diff --git a/internal/cli/metrics.go b/internal/cli/metrics.go new file mode 100644 index 00000000..bc940913 --- /dev/null +++ b/internal/cli/metrics.go @@ -0,0 +1,84 @@ +package cli + +import ( + "context" + "errors" + "flag" + "fmt" + "io" + "net/http" + "time" + + "github.com/openclaw/gitcrawl/internal/config" + "github.com/openclaw/gitcrawl/internal/github" + "github.com/openclaw/gitcrawl/internal/headlinemetrics" +) + +func (a *App) runMetrics(ctx context.Context, args []string) error { + if len(args) == 0 || args[0] == "help" || args[0] == "--help" || args[0] == "-h" { + return a.printCommandUsage("metrics") + } + command := args[0] + if command != "collect" && command != "import" && command != "status" { + return usageErr(fmt.Errorf("unknown metrics command %q", command)) + } + fs := flag.NewFlagSet("metrics "+command, flag.ContinueOnError) + fs.SetOutput(io.Discard) + path := fs.String("config", a.configPath, "metrics JSON config path") + jsonOut := fs.Bool("json", false, "write JSON output") + if err := fs.Parse(args[1:]); err != nil { + if errors.Is(err, flag.ErrHelp) { + return a.printCommandUsage("metrics") + } + return usageErr(err) + } + if *path == "" || fs.NArg() != 0 { + return usageErr(errors.New("metrics requires --config and no positional arguments")) + } + c, err := headlinemetrics.ReadConfig(*path) + if err != nil { + return usageErr(err) + } + a.applyCommandJSON(*jsonOut) + var collect headlinemetrics.Collector + if command == "collect" { + cfg := config.Default() + if c.TokenEnv != "" { + cfg.GitHub.TokenEnv = c.TokenEnv + } + token := a.resolveGitHubToken(ctx, cfg) + var provider func(context.Context) (string, error) + if a.githubTokenCommand != nil { + provider, err = githubTokenProvider(*a.githubTokenCommand) + if err != nil { + return usageErr(err) + } + } + client := github.New(github.Options{Token: token.Value, TokenProvider: provider, BaseURL: githubBaseURL(), HTTPClient: &http.Client{ + Timeout: 30 * time.Second, + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, + }}) + collect = headlinemetrics.GitHubCollector(client, token.Value != "" || provider != nil) + } + in := a.Stdin + if in == nil { + in = http.NoBody + } + result, err := headlinemetrics.Execute(ctx, command, c, collect, in) + // Partial collection is still a useful structured result; diagnostics stay + // on stderr and a nonzero exit communicates unavailable required metrics. + if err == nil || errors.Is(err, headlinemetrics.ErrPartialCollection) { + var output any = result + if command == "status" { + output = struct { + headlinemetrics.Result + Observations int `json:"observations"` + Events int `json:"events"` + }{result, result.Observations, result.Events} + } + if writeErr := a.writeOutput("metrics "+command, output, false); writeErr != nil { + return writeErr + } + } + return err +} diff --git a/internal/cli/metrics_test.go b/internal/cli/metrics_test.go new file mode 100644 index 00000000..4892996a --- /dev/null +++ b/internal/cli/metrics_test.go @@ -0,0 +1,224 @@ +package cli + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + + "github.com/openclaw/gitcrawl/internal/headlinemetrics" +) + +func metricsConfigFixture(t *testing.T) (string, headlinemetrics.Config) { + t.Helper() + dir := t.TempDir() + c := headlinemetrics.Config{Database: filepath.Join(dir, "metrics.sqlite"), Targets: []headlinemetrics.Target{{Entity: "OpenClaw", Target: "openclaw/openclaw"}}} + raw, err := json.Marshal(c) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(dir, "metrics.json") + if err := os.WriteFile(path, raw, 0600); err != nil { + t.Fatal(err) + } + return path, c +} +func metricsApp(t *testing.T) (*App, *bytes.Buffer) { + t.Helper() + app := New() + out := new(bytes.Buffer) + app.Stdout = out + app.Stderr = new(bytes.Buffer) + app.githubAuthTokenLookup = func(context.Context) (string, error) { + t.Error("unexpected credential lookup") + return "", errors.New("no credential") + } + return app, out +} + +func TestMetricsNativeHelpMetadataAndUsage(t *testing.T) { + for _, args := range [][]string{{"--help"}, {"help", "metrics"}, {"metrics"}, {"metrics", "help"}, {"metrics", "-h"}, {"metrics", "collect", "--help"}, {"--json", "metrics", "status", "-h"}, {"metrics", "import", "--help"}} { + app, out := metricsApp(t) + if err := app.Run(context.Background(), args); err != nil || !strings.Contains(out.String(), "metrics") { + t.Fatalf("%v: %v %s", args, err, out) + } + } + app, out := metricsApp(t) + if err := app.Run(context.Background(), []string{"metadata", "--json"}); err != nil { + t.Fatal(err) + } + for _, name := range []string{"metrics-collect", "metrics-import", "metrics-status"} { + if !strings.Contains(out.String(), name) { + t.Fatalf("metadata missing %s", name) + } + } + for _, args := range [][]string{{"metrics", "bad"}, {"metrics", "collect"}, {"metrics", "status", "--unknown"}, {"metrics", "import", "--config", "/missing", "extra"}, {"metrics", "collect", "--config", "/missing"}} { + app, _ := metricsApp(t) + if err := app.Run(context.Background(), args); err == nil || ExitCode(err) != 2 { + t.Fatalf("%v: usage error %v", args, err) + } + } + if releaseNotificationAllowed([]string{"metrics", "status"}) { + t.Fatal("metrics triggers release side effects") + } +} + +func TestMetricsImportStatusNativeJSONAndArchiveIsolation(t *testing.T) { + path, c := metricsConfigFixture(t) + archive := filepath.Join(t.TempDir(), "archive.db") + if err := os.WriteFile(archive, []byte("untouched archive"), 0600); err != nil { + t.Fatal(err) + } + t.Setenv("GITCRAWL_DB_PATH", archive) + t.Setenv("GITCRAWL_CONFIG", filepath.Join(t.TempDir(), "does-not-exist.toml")) + row := `{"type":"metric","id":"history:1","entity":"OpenClaw","target":"openclaw/openclaw","metric":"watchers","kind":"counter","ts":"2026-09-14T00:00:00Z","value":null,"observed_at":"2026-09-15T00:00:00Z","provenance":"claw-track"}` + for i, args := range [][]string{{"--json", "metrics", "import", "--config", path}, {"metrics", "import", "--config", path, "--json"}} { + app, out := metricsApp(t) + app.Stdin = strings.NewReader(row + "\n") + if err := app.Run(context.Background(), args); err != nil { + t.Fatal(err) + } + var r headlinemetrics.Result + if err := json.Unmarshal(out.Bytes(), &r); err != nil { + t.Fatal(err) + } + if r.RowsWritten != 1-i || !r.OK { + t.Fatalf("import=%+v", r) + } + } + for _, args := range [][]string{{"metrics", "status", "--config", path, "--json"}, {"--config", path, "--json", "metrics", "status"}, {"--format", "json", "metrics", "status", "--config", path}} { + app, out := metricsApp(t) + if err := app.Run(context.Background(), args); err != nil { + t.Fatal(err) + } + var r headlinemetrics.Result + if err := json.Unmarshal(out.Bytes(), &r); err != nil || r.Observations != 1 { + t.Fatalf("status=%s %v", out, err) + } + } + if b, _ := os.ReadFile(archive); string(b) != "untouched archive" { + t.Fatal("archive changed") + } + info, err := os.Stat(c.Database) + if err != nil || info.Size() == 0 { + t.Fatalf("metrics DB absent: %v", err) + } + app, _ := metricsApp(t) + app.Stdin = strings.NewReader("bad") + if err := app.Run(context.Background(), []string{"metrics", "import", "--config", path}); err == nil || ExitCode(err) != 1 { + t.Fatal("invalid history accepted") + } +} + +func TestMetricsCollectUsesNativeCredentialsAndKeepsPartialOutput(t *testing.T) { + for _, mode := range []string{"environment", "gh-fallback", "managed", "partial", "quota"} { + t.Run(mode, func(t *testing.T) { + if mode == "managed" && runtime.GOOS == "windows" { + t.Skip("managed helper is Unix-only") + } + path, c := metricsConfigFixture(t) + c.TokenEnv = "GITCRAWL_METRICS_TEST_TOKEN" + b, _ := json.Marshal(c) + if err := os.WriteFile(path, b, 0600); err != nil { + t.Fatal(err) + } + t.Setenv("GITHUB_TOKEN", "ambient-must-not-win") + t.Setenv(c.TokenEnv, "") + token := "fixture-token" + if mode == "environment" || mode == "partial" || mode == "quota" { + t.Setenv(c.TokenEnv, token) + } + requests := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests++ + if r.Header.Get("Authorization") != "Bearer "+token { + t.Error("wrong credential selected") + } + switch r.URL.Path { + case "/repos/openclaw/openclaw": + if mode == "quota" { + w.WriteHeader(http.StatusTooManyRequests) + return + } + fmt.Fprint(w, `{"stargazers_count":4,"forks_count":2,"subscribers_count":1,"open_issues_count":8}`) + case "/search/issues": + if mode == "partial" { + fmt.Fprint(w, `{"incomplete_results":true,"total_count":3}`) + } else { + fmt.Fprint(w, `{"total_count":3}`) + } + case "/repos/openclaw/openclaw/traffic/clones": + w.WriteHeader(403) + case "/repos/openclaw/openclaw/releases": + fmt.Fprint(w, `[]`) + default: + t.Errorf("unexpected API/model/archive request %s", r.URL.Path) + w.WriteHeader(404) + } + })) + defer server.Close() + t.Setenv("GITCRAWL_GITHUB_BASE_URL", server.URL) + t.Setenv("GITCRAWL_OPENAI_BASE_URL", server.URL) + app, out := metricsApp(t) + args := []string{"metrics", "collect", "--config", path, "--json"} + if mode == "gh-fallback" { + app.githubAuthTokenLookup = func(context.Context) (string, error) { return token, nil } + } + if mode == "managed" { + helper := filepath.Join(t.TempDir(), "credential-helper") + if err := os.WriteFile(helper, []byte("#!/bin/sh\nprintf '%s\\n' 'fixture-token'\n"), 0700); err != nil { + t.Fatal(err) + } + args = append([]string{"--github-token-command", helper}, args...) + } + err := app.Run(context.Background(), args) + if (err != nil) != (mode == "partial" || mode == "quota") { + t.Fatalf("error=%v", err) + } + var result headlinemetrics.Result + if e := json.Unmarshal(out.Bytes(), &result); e != nil { + t.Fatal(e) + } + wantRows, wantRequests := 5, 4 + if mode == "quota" { + wantRows, wantRequests = 0, 1 + } + if result.RowsWritten != wantRows || result.OK == (mode == "partial" || mode == "quota") || requests != wantRequests { + t.Fatalf("result=%+v requests=%d", result, requests) + } + if strings.Contains(out.String(), token) { + t.Fatal("token printed") + } + }) + } +} + +func TestMetricsEmptyStatusReportsZeroTotals(t *testing.T) { + path, _ := metricsConfigFixture(t) + app, _ := metricsApp(t) + app.Stdin = strings.NewReader("") + if err := app.Run(context.Background(), []string{"metrics", "import", "--config", path, "--json"}); err != nil { + t.Fatal(err) + } + app, out := metricsApp(t) + if err := app.Run(context.Background(), []string{"metrics", "status", "--config", path, "--json"}); err != nil { + t.Fatal(err) + } + var result map[string]any + if err := json.Unmarshal(out.Bytes(), &result); err != nil { + t.Fatal(err) + } + for _, key := range []string{"observations", "events"} { + if value, present := result[key]; !present || value != float64(0) { + t.Fatalf("status must report %s=0: %s", key, out) + } + } +} diff --git a/internal/cli/releasecheck.go b/internal/cli/releasecheck.go index 67d285a1..73b93d15 100644 --- a/internal/cli/releasecheck.go +++ b/internal/cli/releasecheck.go @@ -37,6 +37,11 @@ func (a *App) maybeNotifyRelease(ctx context.Context, args []string) { } func releaseNotificationAllowed(args []string) bool { + // Metrics commands must not touch archive/runtime state or make unrelated + // network requests, including during read-only imports and status checks. + if len(args) > 0 && args[0] == "metrics" { + return false + } if len(args) == 0 || args[0] != "fill-pr-details" { return true } diff --git a/internal/github/headline_metrics.go b/internal/github/headline_metrics.go new file mode 100644 index 00000000..626dd569 --- /dev/null +++ b/internal/github/headline_metrics.go @@ -0,0 +1,70 @@ +package github + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/url" +) + +// Pointer counts distinguish an unavailable field from an actual zero. +type RepositoryHeadline struct { + Stars *float64 `json:"stargazers_count"` + Forks *float64 `json:"forks_count"` + Watchers *float64 `json:"subscribers_count"` + OpenIssuesAndPulls *float64 `json:"open_issues_count"` +} + +type PullCount struct { + Count *float64 `json:"total_count"` + Incomplete bool `json:"incomplete_results"` +} + +type CloneTraffic struct { + Clones []struct { + Timestamp string `json:"timestamp"` + Count *float64 `json:"count"` + } `json:"clones"` +} + +type Release struct { + ID int64 `json:"id"` + Draft bool `json:"draft"` + Prerelease bool `json:"prerelease"` + Published string `json:"published_at"` + Created string `json:"created_at"` + Name string `json:"name"` + Tag string `json:"tag_name"` + URL string `json:"html_url"` +} + +func (c *Client) RepositoryHeadline(ctx context.Context, repository string) (RepositoryHeadline, error) { + var out RepositoryHeadline + err := c.doJSON(ctx, http.MethodGet, "/repos/"+repository, nil, nil, &out) + return out, err +} + +func (c *Client) OpenPullCount(ctx context.Context, repository string) (PullCount, error) { + var out PullCount + err := c.doJSON(ctx, http.MethodGet, "/search/issues?q="+url.QueryEscape("repo:"+repository+" is:pr is:open")+"&per_page=1", nil, nil, &out) + return out, err +} + +func (c *Client) CloneTraffic(ctx context.Context, repository string) (CloneTraffic, error) { + var out CloneTraffic + err := c.doJSON(ctx, http.MethodGet, "/repos/"+repository+"/traffic/clones?per=day", nil, nil, &out) + if err == nil && out.Clones == nil { + err = errors.New("missing GitHub clone traffic list") + } + return out, err +} + +func (c *Client) ReleasePage(ctx context.Context, repository string, page int) ([]Release, error) { + var out []Release + err := c.doJSON(ctx, http.MethodGet, fmt.Sprintf("/repos/%s/releases?per_page=100&page=%d", repository, page), nil, nil, &out) + if err == nil && out == nil { + err = errors.New("missing GitHub release list") + } + return out, err +} diff --git a/internal/headlinemetrics/github.go b/internal/headlinemetrics/github.go new file mode 100644 index 00000000..6c2c67a8 --- /dev/null +++ b/internal/headlinemetrics/github.go @@ -0,0 +1,171 @@ +package headlinemetrics + +import ( + "context" + "errors" + "fmt" + "math" + "net/http" + "strings" + "time" + + "github.com/openclaw/gitcrawl/internal/github" +) + +// GitHubCollector uses the same HTTP client and credential provider as native +// Gitcrawl commands. Clone traffic is optional and requires authentication. +func GitHubCollector(client *github.Client, trafficEnabled bool) Collector { + return func(ctx context.Context, c Config, ts string) ([]Row, error) { + if err := c.Validate(); err != nil { + return nil, err + } + observed, err := time.Parse(time.RFC3339Nano, ts) + if err != nil { + return nil, errors.New("invalid collection timestamp") + } + rows := []Row{} + failed := false + for _, t := range c.Targets { + if err := ctx.Err(); err != nil { + return rows, err + } + repo, err := client.RepositoryHeadline(ctx, t.Target) + if stopCollection(ctx, err) { + return rows, errors.Join(err, ctx.Err()) + } + if err != nil { + failed = true + repo = github.RepositoryHeadline{} + } + prs, err := client.OpenPullCount(ctx, t.Target) + if err != nil || prs.Incomplete { + failed = true + prs = github.PullCount{} + } + var issues *float64 + if validCount(repo.OpenIssuesAndPulls) != nil && validCount(prs.Count) != nil { + issues = Value(*repo.OpenIssuesAndPulls - *prs.Count) + } + for _, m := range []struct { + Name string + Value *float64 + }{ + {"stars", repo.Stars}, {"forks", repo.Forks}, {"watchers", repo.Watchers}, {"open_prs", prs.Count}, {"open_issues", issues}, + } { + value := validCount(m.Value) + if value == nil { + failed = true + } + rows = append(rows, Counter(t, m.Name, value, ts, "github_rest")) + } + if stopCollection(ctx, err) { + return rows, errors.Join(err, ctx.Err()) + } + if trafficEnabled { + traffic, err := client.CloneTraffic(ctx, t.Target) + if stopCollection(ctx, err) { + return rows, errors.Join(err, ctx.Err()) + } + var requestErr *github.RequestError + optional := errors.As(err, &requestErr) && (requestErr.Status == 403 || requestErr.Status == 404) + if err != nil && !optional { + failed = true + } + if err == nil { + for _, v := range traffic.Clones { + day, err := time.Parse(time.RFC3339Nano, v.Timestamp) + if err != nil { + failed = true + continue + } + day = day.UTC().Truncate(24 * time.Hour) + if !day.Before(observed.UTC().Truncate(24 * time.Hour)) { + continue + } + value := validCount(v.Count) + if value == nil { + failed = true + } + r := Counter(t, "clones", value, day.Add(24*time.Hour-time.Millisecond).Format(time.RFC3339Nano), "github_traffic") + r.Kind = "daily" + r.ObservedAt = ts + rows = append(rows, r) + } + } + } + // Follow all release pages. Do not silently truncate stable release history. + for page := 1; ; page++ { + releases, err := client.ReleasePage(ctx, t.Target, page) + if stopCollection(ctx, err) { + return rows, errors.Join(err, ctx.Err()) + } + if err != nil { + failed = true + break + } + for _, v := range releases { + if v.Draft || v.Prerelease { + continue + } + at := v.Published + if at == "" { + at = v.Created + } + when, err := time.Parse(time.RFC3339Nano, at) + label := v.Name + if label == "" { + label = v.Tag + } + if err != nil || v.ID <= 0 || label == "" { + failed = true + continue + } + rows = append(rows, Row{Type: "event", ID: fmt.Sprintf("github-release:%d", v.ID), Entity: t.Entity, Target: t.Target, Kind: "release", TS: when.UTC().Format(time.RFC3339Nano), ObservedAt: ts, Provenance: "github_releases", Label: label, URL: v.URL}) + } + if len(releases) < 100 { + break + } + } + } + if err := ctx.Err(); err != nil { + return rows, err + } + if failed { + return rows, errors.New("one or more GitHub metrics unavailable") + } + return rows, nil + } +} + +func validCount(value *float64) *float64 { + if value == nil || math.Trunc(*value) != *value { + return nil + } + return Value(*value) +} + +// Permission-denied traffic is optional; exhausted quota and cancellation stop +// all acquisition after the shared HTTP client's bounded retry. +func stopCollection(ctx context.Context, err error) bool { + // Retain a successfully decoded response even if cancellation arrived as + // it finished; the next request uses the canceled context. + if err == nil { + return false + } + if ctx.Err() != nil { + return true + } + var reserve *github.RateLimitReserveError + if errors.As(err, &reserve) { + return true + } + var response *github.RequestError + if !errors.As(err, &response) { + return false + } + return response.Status == http.StatusTooManyRequests || + (response.Status == http.StatusForbidden && + (response.Headers.Get("X-RateLimit-Remaining") == "0" || + response.Headers.Get("Retry-After") != "" || + strings.Contains(strings.ToLower(response.Body), "rate limit"))) +} diff --git a/internal/headlinemetrics/github_test.go b/internal/headlinemetrics/github_test.go new file mode 100644 index 00000000..60920f13 --- /dev/null +++ b/internal/headlinemetrics/github_test.go @@ -0,0 +1,310 @@ +package headlinemetrics + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/openclaw/gitcrawl/internal/github" +) + +func fixtureCollector(t *testing.T, override func(http.ResponseWriter, *http.Request) bool, traffic bool) Collector { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("unexpected mutation %s", r.Method) + } + if r.Header.Get("Authorization") != "Bearer fixture-token" { + t.Error("missing normal client authorization") + } + if override != nil && override(w, r) { + return + } + w.Header().Set("Content-Type", "application/json") + switch { + case r.URL.Path == "/search/issues": + if !strings.Contains(r.URL.Query().Get("q"), "is:pr is:open") { + t.Error("incorrect pull query") + } + fmt.Fprint(w, `{"total_count":3,"incomplete_results":false}`) + case strings.HasSuffix(r.URL.Path, "/traffic/clones"): + fmt.Fprint(w, `{"clones":[{"timestamp":"2026-09-14T00:00:00Z","count":0},{"timestamp":"2026-09-15T00:00:00Z","count":7}]}`) + case strings.HasSuffix(r.URL.Path, "/releases"): + fmt.Fprint(w, `[{"id":42,"published_at":"2026-09-14T15:00:00Z","name":"Stable","tag_name":"v1","html_url":"https://example.test/v1"},{"id":43,"draft":true},{"id":44,"prerelease":true}]`) + case strings.HasPrefix(r.URL.Path, "/repos/"): + fmt.Fprint(w, `{"stargazers_count":100,"forks_count":5,"watchers_count":100,"subscribers_count":7,"open_issues_count":11}`) + default: + t.Errorf("unexpected API %s", r.URL.String()) + http.NotFound(w, r) + } + })) + t.Cleanup(server.Close) + return GitHubCollector(github.New(github.Options{BaseURL: server.URL, Token: "fixture-token", HTTPClient: server.Client()}), traffic) +} + +func TestGitHubMetricsUseActualWatchersSeparateIssuesAndCompletedDays(t *testing.T) { + c := testConfig(t) + rows, err := fixtureCollector(t, nil, true)(context.Background(), c, "2026-09-15T01:00:00Z") + if err != nil { + t.Fatal(err) + } + if len(rows) != 14 { + t.Fatalf("rows=%d", len(rows)) + } + for _, target := range c.Targets { + values := map[string]float64{} + events := 0 + for _, r := range rows { + if r.Target != target.Target { + continue + } + if r.Type == "event" { + events++ + if r.Label != "Stable" { + t.Fatal(r) + } + continue + } + if r.Value == nil { + t.Fatal("unexpected unknown", r) + } + values[r.Metric] = *r.Value + if r.Kind == "daily" && (r.TS != "2026-09-14T23:59:59.999Z" || r.ObservedAt != "2026-09-15T01:00:00Z") { + t.Fatal("incorrect UTC day", r) + } + } + if values["stars"] != 100 || values["forks"] != 5 || values["watchers"] != 7 || values["open_prs"] != 3 || values["open_issues"] != 8 || values["clones"] != 0 || events != 1 { + t.Fatalf("%s values=%v events=%d", target.Target, values, events) + } + } +} + +func TestMissingOrIncompletePRCountNeverFallsBackToCombinedIssues(t *testing.T) { + for _, body := range []string{`{}`, `{"total_count":3,"incomplete_results":true}`, `{"total_count":-1}`, `{"total_count":30}`, `{"total_count":3,"incomplete_results":"invalid"}`} { + t.Run(body, func(t *testing.T) { + collect := fixtureCollector(t, func(w http.ResponseWriter, r *http.Request) bool { + if r.URL.Path == "/search/issues" { + fmt.Fprint(w, body) + return true + } + return false + }, false) + rows, err := collect(context.Background(), testConfig(t), "2026-09-15T01:00:00Z") + if err == nil { + t.Fatal("missing count reported healthy") + } + for _, r := range rows { + if r.Metric == "open_issues" && r.Value != nil { + t.Fatal("combined count used", r) + } + } + }) + } +} + +func TestOptionalCloneTrafficAndOtherPartialFailures(t *testing.T) { + for _, status := range []int{403, 404} { + t.Run(fmt.Sprint(status), func(t *testing.T) { + collect := fixtureCollector(t, func(w http.ResponseWriter, r *http.Request) bool { + if strings.HasSuffix(r.URL.Path, "/traffic/clones") { + w.WriteHeader(status) + fmt.Fprint(w, "private response") + return true + } + return false + }, true) + rows, err := collect(context.Background(), testConfig(t), "2026-09-15T01:00:00Z") + if err != nil { + t.Fatal(err) + } + if len(rows) != 12 { + t.Fatalf("unexpected optional traffic rows %d", len(rows)) + } + }) + } + cases := []struct { + name string + match func(*http.Request) bool + body string + missing string + }{ + {"missing watchers", func(r *http.Request) bool { return r.URL.Path == "/repos/openclaw/openclaw" }, `{"stargazers_count":1,"forks_count":1,"watchers_count":900,"open_issues_count":10}`, "watchers"}, + {"malformed repo", func(r *http.Request) bool { return r.URL.Path == "/repos/openclaw/openclaw" }, `{"stargazers_count":12,`, "stars"}, + {"malformed traffic", func(r *http.Request) bool { return strings.HasSuffix(r.URL.Path, "/traffic/clones") }, `{"clones":[{"timestamp":"bad","count":3}]}`, ""}, + {"missing clone count", func(r *http.Request) bool { return strings.HasSuffix(r.URL.Path, "/traffic/clones") }, `{"clones":[{"timestamp":"2026-09-14T00:00:00Z"}]}`, "clones"}, + {"invalid stable release", func(r *http.Request) bool { return strings.HasSuffix(r.URL.Path, "/releases") }, `[{"id":9,"published_at":"bad","name":"v1"}]`, ""}, + {"malformed releases", func(r *http.Request) bool { return strings.HasSuffix(r.URL.Path, "/releases") }, `{`, ""}, + {"null releases", func(r *http.Request) bool { return strings.HasSuffix(r.URL.Path, "/releases") }, `null`, ""}, + {"missing traffic", func(r *http.Request) bool { return strings.HasSuffix(r.URL.Path, "/traffic/clones") }, `{}`, ""}, + {"null traffic", func(r *http.Request) bool { return strings.HasSuffix(r.URL.Path, "/traffic/clones") }, `{"clones":null}`, ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + collect := fixtureCollector(t, func(w http.ResponseWriter, r *http.Request) bool { + if tc.match(r) { + fmt.Fprint(w, tc.body) + return true + } + return false + }, true) + rows, err := collect(context.Background(), testConfig(t), "2026-09-15T01:00:00Z") + if err == nil || len(rows) < 10 { + t.Fatalf("partial=%d,%v", len(rows), err) + } + if tc.missing != "" { + found := false + for _, r := range rows { + if r.Target == "openclaw/openclaw" && r.Metric == tc.missing { + found = true + if r.Value != nil { + t.Fatal("invalid value retained", r) + } + } + } + if !found { + t.Fatal("missing unknown observation") + } + } + }) + } +} + +func TestReleasePaginationBeyondFivePagesAndNoUnauthenticatedTraffic(t *testing.T) { + pages := 0 + collect := fixtureCollector(t, func(w http.ResponseWriter, r *http.Request) bool { + if strings.HasSuffix(r.URL.Path, "/traffic/clones") { + t.Error("unauthenticated traffic request") + return true + } + if !strings.HasSuffix(r.URL.Path, "/releases") { + return false + } + pages++ + size := 100 + if pages == 6 { + size = 1 + } + releases := make([]github.Release, size) + for i := range releases { + releases[i] = github.Release{ID: int64(pages*100 + i), Created: "2026-09-14T00:00:00Z", Tag: "v1"} + } + if err := json.NewEncoder(w).Encode(releases); err != nil { + t.Error(err) + } + return true + }, false) + c := testConfig(t) + c.Targets = c.Targets[:1] + rows, err := collect(context.Background(), c, "2026-09-15T01:00:00Z") + if err != nil { + t.Fatal(err) + } + if pages != 6 || len(rows) != 506 { + t.Fatalf("pages=%d rows=%d", pages, len(rows)) + } + if _, err := collect(context.Background(), c, "bad"); err == nil { + t.Fatal("invalid clock accepted") + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := collect(ctx, c, "2026-09-15T01:00:00Z"); err == nil { + t.Fatal("cancellation accepted") + } +} + +func TestQuotaFailureStopsCollectionAndRetainsCompletedReads(t *testing.T) { + for _, endpoint := range []string{"/repos/openclaw/openclaw", "/search/issues", "/repos/openclaw/openclaw/traffic/clones", "/repos/openclaw/openclaw/releases"} { + t.Run(endpoint, func(t *testing.T) { + limited := false + requestsAfter := 0 + collect := fixtureCollector(t, func(w http.ResponseWriter, r *http.Request) bool { + if limited { + requestsAfter++ + } + if r.URL.Path == endpoint { + limited = true + w.Header().Set("X-RateLimit-Remaining", "0") + w.WriteHeader(http.StatusForbidden) + return true + } + return false + }, true) + c := testConfig(t) + result, err := Execute(context.Background(), "collect", c, func(ctx context.Context, cfg Config, _ string) ([]Row, error) { + return collect(ctx, cfg, "2026-09-15T01:00:00Z") + }, nil) + if err == nil || result.OK || requestsAfter != 0 { + t.Fatalf("quota stop = %+v, %v; extra requests = %d", result, err, requestsAfter) + } + want := 5 + if endpoint == "/repos/openclaw/openclaw" { + want = 0 + } else if strings.HasSuffix(endpoint, "/releases") { + want = 6 + } + if result.RowsWritten != want { + t.Fatalf("retained %d rows; want %d", result.RowsWritten, want) + } + }) + } +} + +type metricsRoundTripper func(*http.Request) (*http.Response, error) + +func (f metricsRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +type cancelMetricsBody struct { + io.ReadCloser + cancel context.CancelFunc +} + +func (b cancelMetricsBody) Close() error { + err := b.ReadCloser.Close() + b.cancel() + return err +} + +func TestCancellationRetainsTheResponseJustRead(t *testing.T) { + for _, endpoint := range []string{"/repos/openclaw/openclaw", "/repos/openclaw/openclaw/traffic/clones", "/repos/openclaw/openclaw/releases"} { + t.Run(endpoint, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + client := github.New(github.Options{BaseURL: "https://example.test", HTTPClient: &http.Client{Transport: metricsRoundTripper(func(r *http.Request) (*http.Response, error) { + if err := r.Context().Err(); err != nil { + return nil, err + } + payload := `{"stargazers_count":4,"forks_count":0,"subscribers_count":2,"open_issues_count":1}` + switch r.URL.Path { + case "/search/issues": + payload = `{"total_count":0}` + case "/repos/openclaw/openclaw/traffic/clones": + payload = `{"clones":[{"timestamp":"2026-09-14T00:00:00Z","count":3}]}` + case "/repos/openclaw/openclaw/releases": + payload = `[{"id":1,"published_at":"2026-09-14T00:00:00Z","tag_name":"v1"}]` + } + var body io.ReadCloser = io.NopCloser(strings.NewReader(payload)) + if r.URL.Path == endpoint { + body = cancelMetricsBody{body, cancel} + } + return &http.Response{StatusCode: 200, Header: make(http.Header), Body: body, Request: r}, nil + })}}) + c := testConfig(t) + c.Targets = c.Targets[:1] + rows, err := GitHubCollector(client, true)(ctx, c, "2026-09-15T01:00:00Z") + want := 5 + if strings.HasSuffix(endpoint, "/clones") { + want = 6 + } else if strings.HasSuffix(endpoint, "/releases") { + want = 7 + } + if err == nil || len(rows) != want || rows[0].Value == nil || *rows[0].Value != 4 { + t.Fatalf("completed reads = %+v, %v; want %d rows", rows, err, want) + } + }) + } +} diff --git a/internal/headlinemetrics/lock.go b/internal/headlinemetrics/lock.go new file mode 100644 index 00000000..4a7ea3af --- /dev/null +++ b/internal/headlinemetrics/lock.go @@ -0,0 +1,55 @@ +package headlinemetrics + +import ( + "errors" + "fmt" + "os" + "path/filepath" +) + +var errWriterBusy = errors.New("another metrics writer is running") + +// The persistent sidecar must never be unlinked: all writers lock the same inode. +// Closing the file (including process exit) releases ownership without stale PIDs. +func acquireWriter(database string) (*os.File, error) { + if err := os.MkdirAll(filepath.Dir(database), 0700); err != nil { + return nil, err + } + path := filepath.Clean(database) + ".writer.lock" + f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_RDWR, 0600) + if errors.Is(err, os.ErrExist) { + info, statErr := os.Lstat(path) + if statErr != nil { + return nil, statErr + } + if !info.Mode().IsRegular() { + return nil, errors.New("metrics writer lock must be a regular file") + } + f, err = os.OpenFile(path, os.O_RDWR, 0600) + } + if err != nil { + return nil, err + } + fail := func(err error) (*os.File, error) { + _ = f.Close() + return nil, err + } + info, err := f.Stat() + if err != nil { + return fail(err) + } + entry, err := os.Lstat(path) + if err != nil { + return fail(err) + } + if !info.Mode().IsRegular() || !entry.Mode().IsRegular() || !os.SameFile(info, entry) { + return fail(errors.New("metrics writer lock path changed or is not regular")) + } + if err := lockWriterFile(f); err != nil { + return fail(fmt.Errorf("lock metrics database: %w", err)) + } + if err := f.Chmod(0600); err != nil { + return fail(err) + } + return f, nil +} diff --git a/internal/headlinemetrics/lock_other.go b/internal/headlinemetrics/lock_other.go new file mode 100644 index 00000000..3f6ca899 --- /dev/null +++ b/internal/headlinemetrics/lock_other.go @@ -0,0 +1,16 @@ +//go:build !darwin && !dragonfly && !freebsd && !linux && !netbsd && !openbsd && !solaris && !windows + +package headlinemetrics + +import ( + "errors" + "os" +) + +func lockWriterFile(*os.File) error { + return errors.New("metrics writer locking is unsupported on this platform") +} + +func checkSingleLink(*os.File) error { + return errors.New("metrics file identity checks are unsupported on this platform") +} diff --git a/internal/headlinemetrics/lock_test.go b/internal/headlinemetrics/lock_test.go new file mode 100644 index 00000000..caf87cd0 --- /dev/null +++ b/internal/headlinemetrics/lock_test.go @@ -0,0 +1,152 @@ +package headlinemetrics + +import ( + "bufio" + "bytes" + "context" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "slices" + "strings" + "testing" + "time" +) + +func TestExecuteSerializesWriterProcesses(t *testing.T) { + for _, mode := range []string{"complete", "kill"} { + t.Run(mode, func(t *testing.T) { + c := testConfig(t) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestMetricsWriterProcessHelper$", "--", c.Database) + out, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + in, err := cmd.StdinPipe() + if err != nil { + t.Fatal(err) + } + var stderr bytes.Buffer + cmd.Stderr = &stderr + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + waited := false + defer func() { + if !waited { + cancel() + _ = cmd.Wait() + } + }() + line, err := bufio.NewReader(out).ReadString('\n') + if err != nil || line != "collecting\n" { + t.Fatalf("child = %q, %v, %s", line, err, stderr.String()) + } + first, err := os.Stat(c.Database + ".writer.lock") + if err != nil { + t.Fatal(err) + } + collector := func(context.Context, Config, string) ([]Row, error) { + t.Fatal("overlapping collector reached provider") + return nil, nil + } + for _, command := range []string{"collect", "import"} { + result, err := Execute(ctx, command, c, collector, strings.NewReader("invalid input")) + if !errors.Is(err, errWriterBusy) || result.RowsWritten != 0 { + t.Fatalf("overlap %s = %+v, %v", command, result, err) + } + } + if result, err := Execute(ctx, "status", c, nil, nil); err != nil || !result.OK || result.Observations != 0 { + t.Fatalf("concurrent reader = %+v, %v", result, err) + } + // A different metrics database has independent ownership. + if _, err := Execute(ctx, "import", testConfig(t), nil, strings.NewReader("")); err != nil { + t.Fatal(err) + } + if mode == "kill" { + if err := cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + } else if _, err := in.Write([]byte("finish\n")); err != nil { + t.Fatal(err) + } + _ = in.Close() + err = cmd.Wait() + waited = true + if (mode == "complete") != (err == nil) { + t.Fatalf("child exit = %v: %s", err, stderr.String()) + } + if _, err := Execute(ctx, "import", c, nil, strings.NewReader("")); err != nil { + t.Fatalf("writer after %s = %v", mode, err) + } + last, err := os.Stat(c.Database + ".writer.lock") + if err != nil || !os.SameFile(first, last) || last.Mode().Perm() != 0600 { + t.Fatalf("persistent private lock = %v, %v", last, err) + } + result, err := Execute(ctx, "status", c, nil, nil) + want := 0 + if mode == "complete" { + want = 1 + } + if err != nil || result.Observations != want { + t.Fatalf("retained observations = %+v, %v", result, err) + } + }) + } +} + +func TestMetricsWriterProcessHelper(t *testing.T) { + i := slices.Index(os.Args, "--") + if i < 0 { + return + } + c := Config{Database: os.Args[i+1], Targets: []Target{{Entity: "OpenClaw", Target: "openclaw/openclaw"}}} + _, err := Execute(context.Background(), "collect", c, func(context.Context, Config, string) ([]Row, error) { + fmt.Println("collecting") + if _, err := bufio.NewReader(os.Stdin).ReadString('\n'); err != nil { + return nil, err + } + return []Row{testRow()}, nil + }, nil) + if err != nil { + t.Fatal(err) + } +} + +func TestMetricsWriterRejectsUnsafeLockPaths(t *testing.T) { + for _, kind := range []string{"directory", "symlink", "hardlink"} { + t.Run(kind, func(t *testing.T) { + c := testConfig(t) + path := c.Database + ".writer.lock" + target := filepath.Join(filepath.Dir(c.Database), "unrelated") + if err := os.WriteFile(target, []byte("retained"), 0644); err != nil { + t.Fatal(err) + } + var err error + switch kind { + case "directory": + err = os.Mkdir(path, 0700) + case "symlink": + err = os.Symlink(target, path) + case "hardlink": + err = os.Link(target, path) + } + if err != nil { + t.Fatal(err) + } + if _, err := Execute(context.Background(), "import", c, nil, strings.NewReader("")); err == nil { + t.Fatal("unsafe lock accepted") + } + if _, err := os.Stat(c.Database); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("database initialized before lock: %v", err) + } + if got, err := os.ReadFile(target); err != nil || string(got) != "retained" { + t.Fatalf("unrelated file = %q, %v", got, err) + } + }) + } +} diff --git a/internal/headlinemetrics/lock_unix.go b/internal/headlinemetrics/lock_unix.go new file mode 100644 index 00000000..b5d7d848 --- /dev/null +++ b/internal/headlinemetrics/lock_unix.go @@ -0,0 +1,33 @@ +//go:build darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris + +package headlinemetrics + +import ( + "errors" + "os" + + "golang.org/x/sys/unix" +) + +func checkSingleLink(f *os.File) error { + var info unix.Stat_t + if err := unix.Fstat(int(f.Fd()), &info); err != nil { + return err + } + if info.Nlink != 1 { + return errors.New("metrics files must not have hardlink aliases") + } + return nil +} + +func lockWriterFile(f *os.File) error { + if err := checkSingleLink(f); err != nil { + return err + } + + err := unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB) + if errors.Is(err, unix.EWOULDBLOCK) || errors.Is(err, unix.EAGAIN) { + return errWriterBusy + } + return err +} diff --git a/internal/headlinemetrics/lock_windows.go b/internal/headlinemetrics/lock_windows.go new file mode 100644 index 00000000..4168dac8 --- /dev/null +++ b/internal/headlinemetrics/lock_windows.go @@ -0,0 +1,33 @@ +package headlinemetrics + +import ( + "errors" + "os" + + "golang.org/x/sys/windows" +) + +func checkSingleLink(f *os.File) error { + handle := windows.Handle(f.Fd()) + var info windows.ByHandleFileInformation + if err := windows.GetFileInformationByHandle(handle, &info); err != nil { + return err + } + if info.NumberOfLinks != 1 { + return errors.New("metrics files must not have hardlink aliases") + } + return nil +} + +func lockWriterFile(f *os.File) error { + if err := checkSingleLink(f); err != nil { + return err + } + handle := windows.Handle(f.Fd()) + + err := windows.LockFileEx(handle, windows.LOCKFILE_EXCLUSIVE_LOCK|windows.LOCKFILE_FAIL_IMMEDIATELY, 0, 1, 0, &windows.Overlapped{}) + if errors.Is(err, windows.ERROR_LOCK_VIOLATION) { + return errWriterBusy + } + return err +} diff --git a/internal/headlinemetrics/metrics.go b/internal/headlinemetrics/metrics.go new file mode 100644 index 00000000..815b7156 --- /dev/null +++ b/internal/headlinemetrics/metrics.go @@ -0,0 +1,562 @@ +// Package headlinemetrics owns a separate, append-only repository metrics store. +// It rejects thread archives before writes and never starts embedding/model work. +package headlinemetrics + +import ( + "bufio" + "bytes" + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "math" + "os" + "path/filepath" + "regexp" + "strings" + "time" + + "github.com/openclaw/crawlkit/store" +) + +const Owner = "gitcrawl" + +var ErrPartialCollection = errors.New("one or more GitHub metrics unavailable; successful values and unknown observations were retained") + +type Target struct { + Entity string `json:"entity"` + Target string `json:"target"` +} +type Config struct { + Database string `json:"database"` + Targets []Target `json:"targets"` + CookieJar string `json:"cookieJar,omitempty"` + TokenEnv string `json:"tokenEnv,omitempty"` +} +type Row struct { + Type string `json:"type"` + ID string `json:"id,omitempty"` + Entity string `json:"entity"` + Target string `json:"target"` + Metric string `json:"metric,omitempty"` + Kind string `json:"kind"` + TS string `json:"ts"` + Value *float64 `json:"value"` + ObservedAt string `json:"observed_at"` + Provenance string `json:"provenance"` + Label string `json:"label,omitempty"` + URL string `json:"url,omitempty"` +} +type Collector func(context.Context, Config, string) ([]Row, error) + +type Result struct { + Source string `json:"source"` + Command string `json:"command"` + RowsWritten int `json:"rows_written"` + OK bool `json:"ok"` + Observations int `json:"observations,omitempty"` + Events int `json:"events,omitempty"` + LastObserved *string `json:"last_observed,omitempty"` +} + +var repositoryPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9-]*/[A-Za-z0-9_.-]+$`) +var envPattern = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`) + +func ReadConfig(path string) (Config, error) { + var c Config + f, err := os.Open(path) + if err != nil { + return c, fmt.Errorf("read metrics config: %w", err) + } + defer f.Close() + data, err := io.ReadAll(io.LimitReader(f, 1024*1024+1)) + if err != nil { + return c, fmt.Errorf("read metrics config: %w", err) + } + if len(data) > 1024*1024 { + return c, errors.New("metrics config exceeds 1 MiB") + } + d := json.NewDecoder(bytes.NewReader(data)) + d.DisallowUnknownFields() + if err := d.Decode(&c); err != nil { + return c, errors.New("invalid metrics config JSON") + } + if err := d.Decode(new(any)); err != io.EOF { + return c, errors.New("metrics config must contain one JSON object") + } + return c, c.Validate() +} + +func (c Config) Validate() error { + if !filepath.IsAbs(c.Database) || strings.TrimSpace(c.Database) != c.Database || strings.ContainsAny(c.Database, "\x00?") { + return errors.New("metrics database must be an absolute filesystem path") + } + if len(c.Targets) == 0 { + return errors.New("metrics config requires targets") + } + if c.TokenEnv != "" && !envPattern.MatchString(c.TokenEnv) { + return errors.New("invalid tokenEnv name") + } + if c.CookieJar != "" { + return errors.New("GitHub metrics do not use cookieJar") + } + seen := map[string]bool{} + for _, t := range c.Targets { + if strings.TrimSpace(t.Entity) == "" || len(t.Entity) > 200 || len(t.Target) > 300 || !validRepository(t.Target) { + return errors.New("metrics target requires an entity and owner/repo") + } + key := strings.ToLower(t.Target) + if seen[key] { + return errors.New("duplicate metrics repository target") + } + seen[key] = true + } + return nil +} + +func validRepository(target string) bool { + if !repositoryPattern.MatchString(target) { + return false + } + _, name, _ := strings.Cut(target, "/") + return name != "." && name != ".." +} + +func Value(n float64) *float64 { + if math.IsNaN(n) || math.IsInf(n, 0) || n < 0 { + return nil + } + return &n +} +func Counter(t Target, metric string, value *float64, ts, basis string) Row { + return Row{Type: "metric", Entity: t.Entity, Target: t.Target, Metric: metric, Kind: "counter", TS: ts, Value: value, ObservedAt: ts, Provenance: basis} +} + +const Schema = ` +CREATE TABLE IF NOT EXISTS metric_meta(key TEXT PRIMARY KEY,value TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS metric_observations(sequence INTEGER PRIMARY KEY AUTOINCREMENT,id TEXT NOT NULL UNIQUE,entity TEXT NOT NULL,target TEXT NOT NULL,metric TEXT NOT NULL,kind TEXT NOT NULL CHECK(kind IN ('counter','daily')),ts TEXT NOT NULL,value REAL,observed_at TEXT NOT NULL,provenance TEXT NOT NULL); +CREATE INDEX IF NOT EXISTS metric_series ON metric_observations(target,metric,ts,sequence); +CREATE TABLE IF NOT EXISTS metric_events(sequence INTEGER PRIMARY KEY AUTOINCREMENT,id TEXT NOT NULL UNIQUE,entity TEXT NOT NULL,target TEXT NOT NULL,kind TEXT NOT NULL,ts TEXT NOT NULL,label TEXT NOT NULL,url TEXT NOT NULL,observed_at TEXT NOT NULL,provenance TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS metric_runs(sequence INTEGER PRIMARY KEY AUTOINCREMENT,ts TEXT NOT NULL,status TEXT NOT NULL,rows_written INTEGER NOT NULL); +` + +// ownedReadOnly checks identity and schema version before any writable SQLite open. +func ownedReadOnly(ctx context.Context, path string) (*store.Store, error) { + if !filepath.IsAbs(path) || strings.TrimSpace(path) != path { + return nil, errors.New("metrics database must be an absolute path without surrounding whitespace") + } + info, err := os.Lstat(path) + if err != nil { + return nil, err + } + if !info.Mode().IsRegular() || info.Size() == 0 { + return nil, errors.New("metrics database must be a nonempty regular file") + } + f, err := os.Open(path) + if err != nil { + return nil, err + } + err = checkSingleLink(f) + closeErr := f.Close() + if err != nil || closeErr != nil { + return nil, errors.Join(err, closeErr) + } + s, err := store.OpenReadOnly(ctx, path) + if err != nil { + return nil, err + } + if err := validateOwnership(ctx, s.DB()); err != nil { + s.Close() + return nil, err + } + return s, nil +} + +type ownershipReader interface { + QueryRowContext(context.Context, string, ...any) *sql.Row +} + +func validateOwnership(ctx context.Context, db ownershipReader) error { + var owner, version string + err := db.QueryRowContext(ctx, "SELECT value FROM metric_meta WHERE key='owner'").Scan(&owner) + if err == nil { + err = db.QueryRowContext(ctx, "SELECT value FROM metric_meta WHERE key='version'").Scan(&version) + } + var foreign int + if err == nil { + err = db.QueryRowContext(ctx, `SELECT count(*) FROM sqlite_master WHERE type='table' AND name NOT IN ('metric_meta','metric_observations','metric_events','metric_runs','sqlite_sequence') AND name NOT LIKE 'sqlite_%'`).Scan(&foreign) + } + if err != nil || owner != Owner || version != "1" || foreign != 0 { + return errors.New("refusing a database not exclusively owned by gitcrawl metrics schema version 1") + } + return nil +} + +// beforeWritableOpen is nil except in tests that replace a checked path. +var beforeWritableOpen func() + +func Open(ctx context.Context, path string) (_ *store.Store, err error) { + if !filepath.IsAbs(path) || strings.TrimSpace(path) != path { + return nil, errors.New("metrics database path must be absolute") + } + info, err := os.Lstat(path) + newFile := errors.Is(err, os.ErrNotExist) + if err == nil { + if !info.Mode().IsRegular() { + return nil, errors.New("refusing a non-regular metrics database") + } + read, err := ownedReadOnly(ctx, path) + if err != nil { + return nil, err + } + if err = read.Close(); err != nil { + return nil, err + } + } else if !errors.Is(err, os.ErrNotExist) { + return nil, err + } else { + if err = os.MkdirAll(filepath.Dir(path), 0700); err != nil { + return nil, err + } + // Never initialize an existing empty archive or follow a database symlink. + var f *os.File + f, err = os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0600) + if err != nil { + return nil, err + } + created, statErr := f.Stat() + if statErr != nil { + _ = f.Close() + return nil, statErr + } + info = created + // Remove only our newly created file on failure, never an existing + // database or a replacement installed at the same path. + defer func() { + if err != nil { + if current, e := os.Lstat(path); e == nil && os.SameFile(created, current) { + err = errors.Join(err, os.Remove(path)) + } + } + }() + if err = f.Close(); err != nil { + return nil, err + } + } + if beforeWritableOpen != nil { + beforeWritableOpen() + } + // Connecting must not apply schema before the opened database is checked. + s, err := store.Open(ctx, store.Options{Path: path, MaxOpenConns: 1, MaxIdleConns: 1}) + if err != nil { + return nil, err + } + if err = initialize(ctx, s.DB(), path, info, newFile); err != nil { + return nil, errors.Join(err, s.Close()) + } + return s, nil +} + +func initialize(ctx context.Context, db *sql.DB, path string, expected os.FileInfo, newFile bool) (err error) { + conn, err := db.Conn(ctx) + if err != nil { + return err + } + defer func() { err = errors.Join(err, conn.Close()) }() + if _, err = conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil { + return err + } + committed := false + defer func() { + if !committed { + rollbackCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, rollbackErr := conn.ExecContext(rollbackCtx, "ROLLBACK") + err = errors.Join(err, rollbackErr) + } + }() + // Validate through the pinned connection under the write lock, not through + // another pathname lookup that could inspect a different database. + if newFile { + var objects int + if err = conn.QueryRowContext(ctx, "SELECT count(*) FROM sqlite_master").Scan(&objects); err != nil { + return err + } + if objects != 0 { + return errors.New("refusing to initialize a nonempty metrics database") + } + } else if err = validateOwnership(ctx, conn); err != nil { + return err + } + current, err := os.Lstat(path) + if err != nil { + return err + } + if !current.Mode().IsRegular() || !os.SameFile(expected, current) { + return errors.New("metrics database path changed before initialization") + } + if _, err = conn.ExecContext(ctx, Schema); err != nil { + return err + } + if _, err = conn.ExecContext(ctx, "INSERT OR IGNORE INTO metric_meta VALUES('owner',?),('version','1')", Owner); err != nil { + return err + } + _, err = conn.ExecContext(ctx, "COMMIT") + committed = err == nil + return err +} + +func Validate(r Row) error { + if r.Type != "metric" && r.Type != "event" { + return errors.New("invalid observation type") + } + if strings.TrimSpace(r.Entity) == "" || !validRepository(r.Target) || len(r.Entity) > 200 || len(r.Target) > 300 || strings.TrimSpace(r.Provenance) == "" { + return errors.New("invalid observation identity") + } + for _, v := range []string{r.TS, r.ObservedAt} { + if _, err := time.Parse(time.RFC3339Nano, v); err != nil { + return errors.New("invalid observation time") + } + } + if r.Type == "metric" { + if r.Metric == "" || (r.Kind != "counter" && r.Kind != "daily") || (r.Value != nil && Value(*r.Value) == nil) { + return errors.New("invalid metric") + } + if r.Kind == "daily" { + at, _ := time.Parse(time.RFC3339Nano, r.TS) + observed, _ := time.Parse(time.RFC3339Nano, r.ObservedAt) + if !at.UTC().Truncate(24 * time.Hour).Before(observed.UTC().Truncate(24 * time.Hour)) { + return errors.New("daily observation must describe a completed UTC day") + } + } + } else if r.Kind == "" || r.Label == "" { + return errors.New("event kind and label are required") + } + return nil +} + +func insert(ctx context.Context, tx *sql.Tx, r Row) (int, error) { + if err := Validate(r); err != nil { + return 0, err + } + // Imported IDs preserve their exact history. Only freshly collected daily + // values suppress unchanged re-reads; later corrections append a new sequence. + if r.ID == "" && r.Type == "metric" && r.Kind == "daily" { + previous, provenance, err := latestDaily(ctx, tx, r) + if err == nil && provenance == r.Provenance && ((r.Value == nil && !previous.Valid) || (r.Value != nil && previous.Valid && previous.Float64 == *r.Value)) { + return 0, nil + } + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return 0, err + } + } + if r.ID == "" { + b, _ := json.Marshal(r) + h := sha256.Sum256(b) + r.ID = hex.EncodeToString(h[:]) + } + var result sql.Result + var err error + if r.Type == "metric" { + result, err = tx.ExecContext(ctx, "INSERT INTO metric_observations(id,entity,target,metric,kind,ts,value,observed_at,provenance) VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO NOTHING", r.ID, r.Entity, r.Target, r.Metric, r.Kind, r.TS, r.Value, r.ObservedAt, r.Provenance) + } else { + result, err = tx.ExecContext(ctx, "INSERT INTO metric_events(id,entity,target,kind,ts,label,url,observed_at,provenance) VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO NOTHING", r.ID, r.Entity, r.Target, r.Kind, r.TS, r.Label, r.URL, r.ObservedAt, r.Provenance) + } + if err != nil { + return 0, err + } + n, err := result.RowsAffected() + return int(n), err +} + +// A daily import can spell the same UTC day with another offset or time of day. +// Compare parsed days without rewriting imported timestamps or delivery IDs. +func latestDaily(ctx context.Context, tx *sql.Tx, r Row) (sql.NullFloat64, string, error) { + rows, err := tx.QueryContext(ctx, `SELECT ts,value,provenance FROM metric_observations WHERE entity=? AND target=? AND metric=? AND kind='daily' ORDER BY sequence DESC`, r.Entity, r.Target, r.Metric) + if err != nil { + return sql.NullFloat64{}, "", err + } + defer rows.Close() + at, _ := time.Parse(time.RFC3339Nano, r.TS) + day := at.UTC().Truncate(24 * time.Hour) + for rows.Next() { + var raw, provenance string + var value sql.NullFloat64 + if err := rows.Scan(&raw, &value, &provenance); err != nil { + return value, "", err + } + previous, err := time.Parse(time.RFC3339Nano, raw) + if err != nil { + return value, "", errors.New("invalid stored daily observation time") + } + if previous.UTC().Truncate(24 * time.Hour).Equal(day) { + return value, provenance, nil + } + } + if err := rows.Err(); err != nil { + return sql.NullFloat64{}, "", err + } + return sql.NullFloat64{}, "", sql.ErrNoRows +} + +func Write(ctx context.Context, s *store.Store, rows []Row) (int, error) { + written := 0 + err := s.WithTx(ctx, func(tx *sql.Tx) error { + for _, r := range rows { + n, err := insert(ctx, tx, r) + if err != nil { + return err + } + written += n + } + return nil + }) + if err != nil { + return 0, err + } + return written, nil +} + +func inScope(c Config, r Row) bool { + for _, t := range c.Targets { + if t.Target == r.Target && t.Entity == r.Entity { + return true + } + } + return false +} + +// Import validates scope and streams the whole input in one transaction. A bad +// row (including one past a batch boundary) never leaves a partial history. +func Import(ctx context.Context, s *store.Store, c Config, in io.Reader) (int, error) { + written := 0 + err := s.WithTx(ctx, func(tx *sql.Tx) error { + scanner := bufio.NewScanner(in) + scanner.Buffer(make([]byte, 65536), 4*1024*1024) + line := 0 + for scanner.Scan() { + line++ + var r Row + d := json.NewDecoder(strings.NewReader(scanner.Text())) + d.DisallowUnknownFields() + if d.Decode(&r) != nil || d.Decode(new(any)) != io.EOF || !inScope(c, r) { + return fmt.Errorf("invalid import scope or JSON at line %d", line) + } + if r.ID == "" { + return fmt.Errorf("import ID required at line %d", line) + } + n, err := insert(ctx, tx, r) + if err != nil { + return fmt.Errorf("import line %d: %w", line, err) + } + written += n + } + return scanner.Err() + }) + if err != nil { + return 0, err + } + return written, nil +} + +func Execute(ctx context.Context, command string, c Config, collect Collector, in io.Reader) (Result, error) { + result := Result{Source: Owner, Command: command} + if err := c.Validate(); err != nil { + return result, err + } + if command != "status" && command != "import" && command != "collect" { + return result, errors.New("unknown metrics command") + } + if command == "status" { + s, err := ownedReadOnly(ctx, c.Database) + if err != nil { + return result, err + } + defer s.Close() + rows, err := s.DB().QueryContext(ctx, "SELECT observed_at FROM metric_observations") + if err != nil { + return result, err + } + var latest time.Time + for rows.Next() { + var raw string + if err = rows.Scan(&raw); err != nil { + break + } + var at time.Time + at, err = time.Parse(time.RFC3339Nano, raw) + if err != nil { + err = errors.New("invalid stored observation time") + break + } + result.Observations++ + // Imported timestamps retain their original offsets and precision. + if result.LastObserved == nil || at.After(latest) { + latest = at + result.LastObserved = &raw + } + } + err = errors.Join(err, rows.Err(), rows.Close()) + if err == nil { + err = s.DB().QueryRowContext(ctx, "SELECT count(*) FROM metric_events").Scan(&result.Events) + } + result.OK = err == nil + return result, err + } + // Own the database from before initialization through provider reads, commit, + // and close. SQLite transactions alone do not serialize collection attempts. + lock, err := acquireWriter(c.Database) + if err != nil { + return result, err + } + defer lock.Close() + s, err := Open(ctx, c.Database) + if err != nil { + return result, err + } + defer s.Close() + if command == "import" { + result.RowsWritten, err = Import(ctx, s, c, in) + result.OK = err == nil + return result, err + } + ts := time.Now().UTC().Format(time.RFC3339Nano) + rows, collectionErr := collect(ctx, c, ts) + for _, r := range rows { + if !inScope(c, r) { + return result, errors.New("collector returned invalid target") + } + } + // Preserve completed reads even when collection is canceled. This bounded, + // independent write does not start another request or retry collection. + writeCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + err = s.WithTx(writeCtx, func(tx *sql.Tx) error { + for _, r := range rows { + n, err := insert(writeCtx, tx, r) + if err != nil { + return err + } + result.RowsWritten += n + } + status := "ok" + if collectionErr != nil { + status = "partial" + } + _, err := tx.ExecContext(writeCtx, "INSERT INTO metric_runs(ts,status,rows_written) VALUES(?,?,?)", ts, status, result.RowsWritten) + return err + }) + if err != nil { + result.RowsWritten = 0 + return result, err + } + result.OK = collectionErr == nil + if collectionErr != nil { + return result, ErrPartialCollection + } + return result, nil +} diff --git a/internal/headlinemetrics/metrics_test.go b/internal/headlinemetrics/metrics_test.go new file mode 100644 index 00000000..c4748835 --- /dev/null +++ b/internal/headlinemetrics/metrics_test.go @@ -0,0 +1,646 @@ +package headlinemetrics + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "math" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/openclaw/crawlkit/store" +) + +func testConfig(t *testing.T) Config { + t.Helper() + return Config{Database: filepath.Join(t.TempDir(), "metrics.sqlite"), Targets: []Target{{Entity: "OpenClaw", Target: "openclaw/openclaw"}, {Entity: "Example", Target: "example/project"}}} +} +func testRow() Row { + return Counter(Target{"OpenClaw", "openclaw/openclaw"}, "stars", Value(12), "2026-09-15T01:00:00Z", "github_rest") +} +func openTestStore(t *testing.T, c Config) *store.Store { + t.Helper() + s, err := Open(context.Background(), c.Database) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { s.Close() }) + return s +} +func ndjson(t *testing.T, rows ...Row) string { + t.Helper() + var out strings.Builder + for _, r := range rows { + b, err := json.Marshal(r) + if err != nil { + t.Fatal(err) + } + out.Write(b) + out.WriteByte('\n') + } + return out.String() +} + +func TestStorePreservesZerosNullsDecreasesAndIdempotentImports(t *testing.T) { + ctx := context.Background() + c := testConfig(t) + s := openTestStore(t, c) + rows := []Row{} + for i, v := range []*float64{Value(12), Value(8), Value(0), nil} { + r := testRow() + r.ID = fmt.Sprint(i) + r.Value = v + rows = append(rows, r) + } + event := Row{Type: "event", ID: "import-release", Entity: "Example", Target: "example/project", Kind: "release", TS: "2026-09-14T00:00:00Z", ObservedAt: "2026-09-15T00:00:00Z", Provenance: "claw-track", Label: "v1", URL: "https://github.com/example/project/releases/tag/v1"} + rows = append(rows, event) + for _, want := range []int{5, 0} { + n, err := Import(ctx, s, c, strings.NewReader(ndjson(t, rows...))) + if err != nil || n != want { + t.Fatalf("import = %d,%v want %d", n, err, want) + } + } + cursor, err := s.DB().Query("SELECT value FROM metric_observations ORDER BY sequence") + if err != nil { + t.Fatal(err) + } + defer cursor.Close() + for _, want := range []*float64{Value(12), Value(8), Value(0), nil} { + if !cursor.Next() { + t.Fatal("missing observation") + } + var got sql.NullFloat64 + if err := cursor.Scan(&got); err != nil { + t.Fatal(err) + } + if got.Valid != (want != nil) || (want != nil && got.Float64 != *want) { + t.Fatalf("value = %+v want %v", got, want) + } + } + if cursor.Next() { + t.Fatal("duplicate observations") + } + result, err := Execute(ctx, "status", c, nil, nil) + if err != nil || result.Observations != 4 || result.Events != 1 || result.LastObserved == nil { + t.Fatalf("status = %+v %v", result, err) + } + info, err := os.Stat(c.Database) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm()&0077 != 0 { + t.Fatalf("database permissions: %v", info.Mode()) + } +} + +func TestImportRollbackAfterBatchBoundaryAndInvalidInput(t *testing.T) { + for _, bad := range []string{"not json", `{"type":"metric"}`, "", strings.Repeat("x", 4*1024*1024+1)} { + t.Run(fmt.Sprint(len(bad)), func(t *testing.T) { + c := testConfig(t) + s := openTestStore(t, c) + var input strings.Builder + for i := 0; i < 501; i++ { + r := testRow() + r.ID = fmt.Sprint(i) + input.WriteString(ndjson(t, r)) + } + input.WriteString(bad + "\n") + n, err := Import(context.Background(), s, c, strings.NewReader(input.String())) + if err == nil || n != 0 { + t.Fatalf("import=%d,%v", n, err) + } + var count int + if err := s.DB().QueryRow("SELECT count(*) FROM metric_observations").Scan(&count); err != nil || count != 0 { + t.Fatalf("partial import committed: %d,%v", count, err) + } + }) + } + c := testConfig(t) + s := openTestStore(t, c) + for _, mutate := range []func(*Row){func(r *Row) { r.Entity = "wrong" }, func(r *Row) { r.Target = "other/repo" }, func(r *Row) { r.ID = "" }, func(r *Row) { r.Value = Value(-1); r.TS = "invalid" }} { + r := testRow() + r.ID = "id" + mutate(&r) + if _, err := Import(context.Background(), s, c, strings.NewReader(ndjson(t, r))); err == nil { + t.Fatalf("accepted invalid row %+v", r) + } + } +} + +func TestDailyRevisionsAppendAndImportedIDsRemainIndependent(t *testing.T) { + c := testConfig(t) + s := openTestStore(t, c) + ctx := context.Background() + r := testRow() + r.Kind = "daily" + r.Metric = "clones" + r.TS = "2026-09-14T23:59:59.999Z" + r.Provenance = "github_traffic" + for i, v := range []float64{10, 10, 9, 10} { + r.Value = Value(v) + r.ObservedAt = fmt.Sprintf("2026-09-15T%02d:00:00Z", i) + n, err := Write(ctx, s, []Row{r}) + want := 1 + if i == 1 { + want = 0 + } + if err != nil || n != want { + t.Fatalf("daily revision %d: %d,%v", i, n, err) + } + } + r.ID = "historical-1" + r.Provenance = "claw-track" + n, err := Import(ctx, s, c, strings.NewReader(ndjson(t, r))) + if err != nil || n != 1 { + t.Fatalf("history: %d,%v", n, err) + } + r.ID = "historical-2" + n, err = Import(ctx, s, c, strings.NewReader(ndjson(t, r))) + if err != nil || n != 1 { + t.Fatalf("distinct history ID: %d,%v", n, err) + } + var count int + var latest float64 + if err = s.DB().QueryRow("SELECT count(*) FROM metric_observations").Scan(&count); err != nil || count != 5 { + t.Fatalf("count %d %v", count, err) + } + if err = s.DB().QueryRow("SELECT value FROM metric_observations ORDER BY sequence DESC LIMIT 1").Scan(&latest); err != nil || latest != 10 { + t.Fatalf("latest %f %v", latest, err) + } +} + +func TestRefuseArchiveWrongOwnerVersionAndLinksWithoutChangingBytes(t *testing.T) { + for _, schema := range []string{ + "CREATE TABLE threads(id INTEGER PRIMARY KEY)", + "CREATE TABLE metric_meta(key TEXT PRIMARY KEY,value TEXT);INSERT INTO metric_meta VALUES('owner','redcrawl'),('version','1')", + "CREATE TABLE metric_meta(key TEXT PRIMARY KEY,value TEXT);INSERT INTO metric_meta VALUES('owner','gitcrawl'),('version','2')", + "CREATE TABLE metric_meta(key TEXT PRIMARY KEY,value TEXT);INSERT INTO metric_meta VALUES('owner','gitcrawl'),('version','1');CREATE TABLE threads(id INTEGER)", + } { + t.Run(schema, func(t *testing.T) { + c := testConfig(t) + s, err := store.Open(context.Background(), store.Options{Path: c.Database, Schema: schema}) + if err != nil { + t.Fatal(err) + } + s.Close() + before, err := os.ReadFile(c.Database) + if err != nil { + t.Fatal(err) + } + if s, err := Open(context.Background(), c.Database); err == nil { + s.Close() + t.Fatal("archive accepted") + } + if _, err := Execute(context.Background(), "status", c, nil, nil); err == nil { + t.Fatal("wrong schema status accepted") + } + after, _ := os.ReadFile(c.Database) + if string(before) != string(after) { + t.Fatal("archive bytes changed") + } + }) + } + c := testConfig(t) + s := openTestStore(t, c) + s.Close() + link := filepath.Join(t.TempDir(), "linked.db") + if err := os.Symlink(c.Database, link); err != nil { + t.Skip(err) + } + if s, err := Open(context.Background(), link); err == nil { + s.Close() + t.Fatal("database symlink accepted") + } + empty := filepath.Join(t.TempDir(), "empty.db") + if err := os.WriteFile(empty, nil, 0600); err != nil { + t.Fatal(err) + } + if s, err := Open(context.Background(), empty); err == nil { + s.Close() + t.Fatal("empty existing database accepted") + } +} + +func TestValidationAndReadOnlyStatus(t *testing.T) { + c := testConfig(t) + if _, err := Execute(context.Background(), "status", c, nil, nil); err == nil { + t.Fatal("missing status accepted") + } + if _, err := os.Stat(c.Database); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("status created database: %v", err) + } + for _, mutate := range []func(*Config){func(c *Config) { c.Database = "relative.db" }, func(c *Config) { c.Targets = nil }, func(c *Config) { c.Targets = append(c.Targets, c.Targets[0]) }, func(c *Config) { c.Targets[0].Target = "../archive" }, func(c *Config) { c.TokenEnv = "bad-name" }, func(c *Config) { c.CookieJar = "/unused" }} { + cfg := testConfig(t) + mutate(&cfg) + if err := cfg.Validate(); err == nil { + t.Fatalf("config accepted: %+v", cfg) + } + } + for _, mutate := range []func(*Row){func(r *Row) { r.Type = "unknown" }, func(r *Row) { r.Entity = "" }, func(r *Row) { r.Provenance = "" }, func(r *Row) { r.TS = "bad" }, func(r *Row) { r.Kind = "gauge" }, func(r *Row) { r.Value = new(float64); *r.Value = -1 }, func(r *Row) { r.Value = new(float64); *r.Value = math.Inf(1) }, func(r *Row) { r.Kind = "daily" }, func(r *Row) { r.Type = "event"; r.Label = "" }} { + r := testRow() + mutate(&r) + if err := Validate(r); err == nil { + t.Fatalf("row accepted: %+v", r) + } + } + for _, value := range []float64{-1, math.Inf(1), math.NaN()} { + if Value(value) != nil { + t.Fatal("invalid value accepted") + } + } + for _, body := range []string{`{"database":"/tmp/test","targets":[]} invalid`, `{"database":"/tmp/test","targets":[],"unknown":true}`} { + path := filepath.Join(t.TempDir(), "config.json") + if err := os.WriteFile(path, []byte(body), 0600); err != nil { + t.Fatal(err) + } + if _, err := ReadConfig(path); err == nil { + t.Fatal("invalid JSON config accepted") + } + } +} + +func TestDatabaseWhitespaceCannotBypassArchiveOwnership(t *testing.T) { + c := testConfig(t) + if err := os.WriteFile(c.Database, []byte("archive bytes"), 0600); err != nil { + t.Fatal(err) + } + for _, suffix := range []string{" ", "\t", "\n"} { + path := c.Database + suffix + if s, err := Open(context.Background(), path); err == nil { + s.Close() + t.Fatal("whitespace path accepted") + } + if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("created alias path: %v", err) + } + cfg := c + cfg.Database = path + if err := cfg.Validate(); err == nil { + t.Fatal("config accepted whitespace alias") + } + } + if b, _ := os.ReadFile(c.Database); string(b) != "archive bytes" { + t.Fatal("archive changed") + } + c.Targets[0].Target = "openclaw/.github" + if err := c.Validate(); err != nil { + t.Fatal(err) + } + c.Targets[0].Target = "openclaw/.." + if err := c.Validate(); err == nil { + t.Fatal("traversal target accepted") + } +} + +func TestCollectRetainsPartialAndCancelledResultsAtomically(t *testing.T) { + for _, cancelled := range []bool{false, true} { + t.Run(fmt.Sprint(cancelled), func(t *testing.T) { + c := testConfig(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + collect := func(_ context.Context, _ Config, ts string) ([]Row, error) { + if cancelled { + cancel() + } + r := testRow() + r.TS = ts + r.ObservedAt = ts + missing := r + missing.Metric = "watchers" + missing.Value = nil + return []Row{r, missing}, errors.New("do not expose source credentials") + } + result, err := Execute(ctx, "collect", c, collect, nil) + if err == nil || result.OK || result.RowsWritten != 2 || strings.Contains(err.Error(), "credentials") { + t.Fatalf("result=%+v error=%v", result, err) + } + s, err := ownedReadOnly(context.Background(), c.Database) + if err != nil { + t.Fatal(err) + } + defer s.Close() + var status string + var n int + if err := s.DB().QueryRow("SELECT status,rows_written FROM metric_runs").Scan(&status, &n); err != nil || status != "partial" || n != 2 { + t.Fatalf("run=%s,%d %v", status, n, err) + } + }) + } + c := testConfig(t) + result, err := Execute(context.Background(), "collect", c, func(context.Context, Config, string) ([]Row, error) { + r := testRow() + r.Target = "outside/scope" + return []Row{r}, nil + }, nil) + if err == nil || result.RowsWritten != 0 { + t.Fatalf("scope leak: %+v %v", result, err) + } + s := openTestStore(t, c) + r := testRow() + bad := r + bad.Type = "invalid" + n, err := Write(context.Background(), s, []Row{r, bad}) + if err == nil || n != 0 { + t.Fatalf("write rollback=%d %v", n, err) + } +} + +func TestStatusOrdersExistingTimestampsChronologically(t *testing.T) { + for _, times := range [][]string{ + {"2026-09-15T01:00:00+02:00", "2026-09-15T00:00:00Z"}, + {"2026-09-15T00:00:00Z", "2026-09-15T00:00:00.000000001Z"}, + } { + c := testConfig(t) + s := openTestStore(t, c) + for i, at := range times { + r := testRow() + r.ID = fmt.Sprint(i) + r.ObservedAt = at + if _, err := Import(context.Background(), s, c, strings.NewReader(ndjson(t, r))); err != nil { + t.Fatal(err) + } + } + result, err := Execute(context.Background(), "status", c, nil, nil) + if err != nil || result.LastObserved == nil || *result.LastObserved != times[1] { + t.Fatalf("status = %+v, %v; latest instant = %s", result, err, times[1]) + } + } +} + +func TestFailedInitializationCanRetryWithoutAdoptingForeignFiles(t *testing.T) { + c := testConfig(t) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if s, err := Open(ctx, c.Database); err == nil { + s.Close() + t.Fatal("canceled initialization succeeded") + } + if _, err := os.Stat(c.Database); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("failed initialization stranded a file: %v", err) + } + s := openTestStore(t, c) + if _, err := Write(context.Background(), s, []Row{testRow()}); err != nil { + t.Fatal(err) + } + s.Close() + if s, err := Open(ctx, c.Database); err == nil { + s.Close() + t.Fatal("canceled reopen succeeded") + } + result, err := Execute(context.Background(), "status", c, nil, nil) + if err != nil || result.Observations != 1 { + t.Fatalf("existing history lost after failed reopen: %+v, %v", result, err) + } +} + +func TestRefuseMetricsDatabaseHardlinkAliases(t *testing.T) { + c := testConfig(t) + s := openTestStore(t, c) + s.Close() + alias := filepath.Join(t.TempDir(), "alias.sqlite") + if err := os.Link(c.Database, alias); err != nil { + t.Skip(err) + } + before, err := os.ReadFile(c.Database) + if err != nil { + t.Fatal(err) + } + for _, path := range []string{c.Database, alias} { + cfg := c + cfg.Database = path + if _, err := Execute(context.Background(), "import", cfg, nil, strings.NewReader("")); err == nil { + t.Fatal("hardlinked database bypasses writer serialization") + } + } + after, err := os.ReadFile(c.Database) + if err != nil || string(after) != string(before) { + t.Fatal("hardlinked database changed") + } +} + +func TestDailyCollectionUsesLatestImportedUTCDay(t *testing.T) { + c := testConfig(t) + s := openTestStore(t, c) + ctx := context.Background() + r := testRow() + r.ID, r.Kind, r.Metric = "imported-day", "daily", "clones" + r.TS, r.Provenance = "2026-09-15T01:00:00+02:00", "github_traffic" + if _, err := Import(ctx, s, c, strings.NewReader(ndjson(t, r))); err != nil { + t.Fatal(err) + } + r.ID, r.TS = "", "2026-09-14T23:59:59.999Z" + for i, value := range []*float64{Value(12), nil, nil, Value(0), Value(12)} { + r.Value = value + r.ObservedAt = fmt.Sprintf("2026-09-15T%02d:00:00Z", i+2) + n, err := Write(ctx, s, []Row{r}) + want := 1 + if i == 0 || i == 2 { + want = 0 + } + if err != nil || n != want { + t.Fatalf("daily revision %d = %d, %v; want %d", i, n, err, want) + } + } +} + +func TestConfigRejectsContentBeyondSizeLimit(t *testing.T) { + c := testConfig(t) + data, err := json.Marshal(c) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "metrics.json") + data = append(data, []byte(strings.Repeat(" ", 1024*1024)+`{"ignored":true}`)...) + if err := os.WriteFile(path, data, 0600); err != nil { + t.Fatal(err) + } + if _, err := ReadConfig(path); err == nil { + t.Fatal("oversized config with trailing JSON was accepted") + } +} + +func TestOpenRejectsReplacedDatabaseBeforeSchemaWrites(t *testing.T) { + ctx := context.Background() + for _, existing := range []bool{false, true} { + for _, parent := range []bool{false, true} { + for _, metrics := range []bool{false, true} { + t.Run(fmt.Sprintf("existing=%t/parent=%t/metrics=%t", existing, parent, metrics), func(t *testing.T) { + root := t.TempDir() + active := filepath.Join(root, "active") + if err := os.Mkdir(active, 0700); err != nil { + t.Fatal(err) + } + path := filepath.Join(active, "metrics.sqlite") + if existing { + s, err := Open(ctx, path) + if err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + } + replacementDir := filepath.Join(root, "replacement") + replacement := filepath.Join(replacementDir, "metrics.sqlite") + schema := "CREATE TABLE threads(id INTEGER PRIMARY KEY,body TEXT); INSERT INTO threads VALUES(1,'archive retained')" + if metrics { + schema = Schema + "INSERT INTO metric_meta VALUES('owner','gitcrawl'),('version','1'); INSERT INTO metric_runs(ts,status,rows_written) VALUES('retained','ok',7)" + } + s, err := store.Open(ctx, store.Options{Path: replacement, Schema: schema}) + if err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + replacementInfo, err := os.Stat(replacement) + if err != nil { + t.Fatal(err) + } + called := false + beforeWritableOpen = func() { + called = true + from, to := replacement, path + if parent { + from, to = replacementDir, active + } + if err := os.Rename(to, to+".original"); err != nil { + t.Fatal(err) + } + if err := os.Rename(from, to); err != nil { + t.Fatal(err) + } + } + t.Cleanup(func() { beforeWritableOpen = nil }) + opened, err := Open(ctx, path) + if opened != nil { + opened.Close() + } + if !called || err == nil || opened != nil { + t.Fatalf("replaced database accepted: called=%t store=%v error=%v", called, opened, err) + } + current, err := os.Stat(path) + if err != nil || !os.SameFile(replacementInfo, current) { + t.Fatalf("replacement removed or changed: %v", err) + } + read, err := store.OpenReadOnly(ctx, path) + if err != nil { + t.Fatal(err) + } + defer read.Close() + if metrics { + var retained int + if err := read.DB().QueryRow("SELECT rows_written FROM metric_runs WHERE ts='retained'").Scan(&retained); err != nil || retained != 7 { + t.Fatalf("replacement metrics history changed: %d, %v", retained, err) + } + } else { + var objects int + if err := read.DB().QueryRow("SELECT count(*) FROM sqlite_master WHERE name GLOB 'metric_*'").Scan(&objects); err != nil || objects != 0 { + t.Fatalf("metrics schema landed in archive: %d, %v", objects, err) + } + var body string + if err := read.DB().QueryRow("SELECT body FROM threads WHERE id=1").Scan(&body); err != nil || body != "archive retained" { + t.Fatalf("archive contents changed: %q, %v", body, err) + } + } + }) + } + } + } +} + +func TestOpenRevalidatesDatabaseChangedInPlace(t *testing.T) { + ctx := context.Background() + for _, existing := range []bool{false, true} { + t.Run(fmt.Sprint(existing), func(t *testing.T) { + c := testConfig(t) + if existing { + s := openTestStore(t, c) + if err := s.Close(); err != nil { + t.Fatal(err) + } + } + beforeWritableOpen = func() { + schema := "CREATE VIEW unrelated AS SELECT 1" + if existing { + schema = "UPDATE metric_meta SET value='another-owner' WHERE key='owner'" + } + s, err := store.Open(ctx, store.Options{Path: c.Database, Schema: schema}) + if err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + } + t.Cleanup(func() { beforeWritableOpen = nil }) + s, err := Open(ctx, c.Database) + if s != nil { + s.Close() + } + if err == nil || s != nil { + t.Fatalf("changed database accepted: %v, %v", s, err) + } + if existing { + read, err := store.OpenReadOnly(ctx, c.Database) + if err != nil { + t.Fatal(err) + } + defer read.Close() + var owner string + if err := read.DB().QueryRow("SELECT value FROM metric_meta WHERE key='owner'").Scan(&owner); err != nil || owner != "another-owner" { + t.Fatalf("replaced ownership overwritten: %q, %v", owner, err) + } + } + }) + } +} + +func TestOpenRollsBackSchemaWhenOwnershipInsertFails(t *testing.T) { + ctx := context.Background() + c := testConfig(t) + s, err := store.Open(ctx, store.Options{Path: c.Database, Schema: ` +CREATE TABLE metric_meta(key TEXT PRIMARY KEY,value TEXT NOT NULL); +INSERT INTO metric_meta VALUES('owner','gitcrawl'),('version','1'); +CREATE TRIGGER reject_ownership BEFORE INSERT ON metric_meta BEGIN SELECT RAISE(ABORT,'fixture rejection'); END; +`}) + if err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + if s, err := Open(ctx, c.Database); err == nil { + s.Close() + t.Fatal("ownership insert unexpectedly succeeded") + } + read, err := store.OpenReadOnly(ctx, c.Database) + if err != nil { + t.Fatal(err) + } + var tables int + err = read.DB().QueryRow("SELECT count(*) FROM sqlite_master WHERE type='table' AND name IN ('metric_observations','metric_events','metric_runs')").Scan(&tables) + closeErr := read.Close() + if err != nil || closeErr != nil || tables != 0 { + t.Fatalf("schema was not rolled back: tables=%d error=%v close=%v", tables, err, closeErr) + } + // The failed open must release its transaction and connection for a retry. + s, err = store.Open(ctx, store.Options{Path: c.Database, Schema: "DROP TRIGGER reject_ownership"}) + if err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + s, err = Open(ctx, c.Database) + if err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } +} diff --git a/scripts/build-docs-site.mjs b/scripts/build-docs-site.mjs index bf09cb3f..12086d8a 100644 --- a/scripts/build-docs-site.mjs +++ b/scripts/build-docs-site.mjs @@ -16,7 +16,7 @@ const sections = [ ["Start", ["index.md", "installation.md", "quickstart.md", "concepts.md"]], ["Configure", ["configuration.md", "sync.md", "refresh-and-embed.md"]], ["Use", ["search.md", "clustering.md", "governance.md", "tui.md", "gh-shim.md"]], - ["Operate", ["portable-stores.md", "maintainer-archive.md", "automation.md", "releasing.md"]], + ["Operate", ["portable-stores.md", "maintainer-archive.md", "metrics.md", "automation.md", "releasing.md"]], ["Reference", ["commands.md", "reference.md"]], ];