diff --git a/docs/configuration.md b/docs/configuration.md index f85933e..92c0422 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -264,16 +264,19 @@ Owner-scoped GitHub ingestion may emit: - `github.public_repo_count` Notes: -- owner-scoped rows should use `scope_key = owner:` -- aggregate metrics are computed across the repositories included by that owner-scoped binding +- owner-scoped rows should use `scope_key = "owner:"` +- aggregate metrics are computed across the public repositories included by that owner-scoped binding +- the runtime resolves owner scope through the GitHub owner listing endpoint selected by `owner_type` +- when `GITHUB_TOKEN` is present in the runtime environment, GitHub requests use it to improve rate-limit headroom; unauthenticated mode remains supported but is more exposed to lower anonymous rate limits +- private repositories remain out of scope for owner aggregation even when authenticated +- public archived repositories and public forks are currently included in the owner aggregate because the binding contract does not yet define archive/fork filters - owner-scoped `github.contributors` is intentionally deferred for `v0.3.0` and must not be implied by the presence of `scope.owner` Examples: ```json { "owner": "Blockstream", - "owner_type": "organization", - "include_owner_repos": true + "owner_type": "organization" } ``` @@ -298,14 +301,16 @@ The binding contract remains query-driven and provider-agnostic at the config le For the first live `v0.3.0` runtime, the documented query-based provider is the Google News RSS search endpoint. This means: -- the runtime should construct requests from the configured query +- the runtime constructs requests from the configured query +- the runtime issues Google News RSS search requests under `/rss/search` +- accepted response entries normalize into `content_items` with `item_type = "news_article"` +- each successful binding execution also persists one `news.article_count` snapshot using the post-dedupe accepted item count (after `max_items`, when configured) - alternative query-based providers such as NewsAPI are intentionally deferred - the current docs do **not** imply runtime provider selection from config Example: ```json -{ - "query": "Blockstream Jade", +{ "query": "Blockstream Jade", "max_items": 15 } ``` diff --git a/docs/reference-dataset.md b/docs/reference-dataset.md index 776c853..9f0a057 100644 --- a/docs/reference-dataset.md +++ b/docs/reference-dataset.md @@ -92,6 +92,12 @@ Important `v0.3.0` limitation: - owner-scoped `github.contributors` is intentionally deferred - contributor counts are only expected from explicit repo-scoped GitHub bindings +Operational note for owner-scoped bindings in `v0.3.0`: +- the runtime aggregates across the public repositories returned by the configured owner endpoint +- `GITHUB_TOKEN` is optional and only improves rate-limit headroom; unauthenticated mode remains supported +- public archived repositories and public forks are currently included in that aggregate +- private repositories are excluded from the owner aggregate + ### 2. News RSS Every tracked entity in the demo pack is assumed to have a keyword-based news binding. diff --git a/internal/config/validator.go b/internal/config/validator.go index 11f0f23..e7d743c 100644 --- a/internal/config/validator.go +++ b/internal/config/validator.go @@ -483,6 +483,10 @@ func (v *packValidator) validateGitHubScope(file, path string, scope JSONMap) { } } + if owner != "" && len(repos) > 0 { + v.addIssue(file, path, "must define either a non-empty repos list or an owner/owner_type pair, not both") + } + if owner == "" && len(repos) == 0 { v.addIssue( file, diff --git a/internal/config/validator_test.go b/internal/config/validator_test.go index 10c095e..4daae3d 100644 --- a/internal/config/validator_test.go +++ b/internal/config/validator_test.go @@ -209,6 +209,77 @@ func TestValidatePackAggregatesCrossFileIssues(t *testing.T) { } } +func TestValidatePackRejectsMixedGitHubRepoAndOwnerScopes(t *testing.T) { + t.Parallel() + + pack := Pack{ + Entities: EntitiesFile{ + Version: 1, + DatasetID: "demo", + DatasetName: "Demo", + Entities: []Entity{ + { + ID: "org-1", + Slug: "org-1", + Kind: "organization", + Name: "Org 1", + }, + }, + Relationships: []Relationship{}, + }, + Sources: SourcesFile{ + Version: 1, + DatasetID: "demo", + Sources: []Source{ + { + ID: "github", + Kind: "github", + Enabled: true, + }, + }, + Bindings: []Binding{ + { + ID: "github-binding-1", + EntityID: "org-1", + SourceID: "github", + Enabled: true, + Scope: JSONMap{ + "owner": "example", + "owner_type": "organization", + "repos": []string{"example/repo-1"}, + }, + }, + }, + }, + Alerts: AlertsFile{ + Version: 1, + DatasetID: "demo", + Rules: []AlertRule{}, + }, + Schedules: SchedulesFile{ + Version: 1, + DatasetID: "demo", + Jobs: []ScheduleJob{}, + }, + } + + err := ValidatePack(pack) + if err == nil { + t.Fatal("ValidatePack() error = nil, want validation error") + } + + var validationErr *ValidationError + if !errors.As(err, &validationErr) { + t.Fatalf("ValidatePack() error = %T, want *ValidationError", err) + } + + got := joinIssues(validationErr.Issues) + want := `sources.json: bindings[0].scope: must define either a non-empty repos list or an owner/owner_type pair, not both` + if !strings.Contains(got, want) { + t.Fatalf("ValidatePack() issues missing %q\nGot:\n%s", want, got) + } +} + func joinIssues(issues []ValidationIssue) string { lines := make([]string, 0, len(issues)) for _, issue := range issues { diff --git a/internal/fetch/github.go b/internal/fetch/github.go index 4be00b9..2b7121a 100644 --- a/internal/fetch/github.go +++ b/internal/fetch/github.go @@ -8,6 +8,8 @@ import ( "io" "net/http" "net/url" + "os" + "sort" "strconv" "strings" "time" @@ -25,15 +27,19 @@ const ( MetricKeyGitHubOpenIssues = "github.open_issues" MetricKeyGitHubCommits30d = "github.commits_30d" MetricKeyGitHubCommits90d = "github.commits_90d" + MetricKeyGitHubPublicRepoCount = "github.public_repo_count" MetricKeyGitHubLatestReleasePublishedAt = "github.latest_release_published_at" defaultGitHubAPIBaseURL = "https://api.github.com/" defaultGitHubUserAgent = "signalscope-fetch/0.3.0" + githubTokenEnvVar = "GITHUB_TOKEN" ) type GitHubFetcherOptions struct { - Client *GitHubClient - Now func() time.Time + Client *GitHubClient + ClientOptions GitHubClientOptions + TokenLookup func() string + Now func() time.Time } type GitHubFetcher struct { @@ -44,7 +50,20 @@ type GitHubFetcher struct { func NewGitHubFetcher(options GitHubFetcherOptions) *GitHubFetcher { client := options.Client if client == nil { - client = NewGitHubClient(GitHubClientOptions{}) + clientOptions := options.ClientOptions + + tokenLookup := options.TokenLookup + if tokenLookup == nil { + tokenLookup = func() string { + return strings.TrimSpace(os.Getenv(githubTokenEnvVar)) + } + } + + if strings.TrimSpace(clientOptions.Token) == "" { + clientOptions.Token = strings.TrimSpace(tokenLookup()) + } + + client = NewGitHubClient(clientOptions) } now := options.Now @@ -58,6 +77,20 @@ func NewGitHubFetcher(options GitHubFetcherOptions) *GitHubFetcher { } } +type githubBindingScopeKind string + +const ( + githubBindingScopeRepos githubBindingScopeKind = "repos" + githubBindingScopeOwner githubBindingScopeKind = "owner" +) + +type githubBindingScope struct { + Kind githubBindingScopeKind + Repositories []githubRepositoryRef + Owner string + OwnerType string +} + type githubRepositoryRef struct { Owner string Name string @@ -67,17 +100,46 @@ func (repository githubRepositoryRef) FullName() string { return repository.Owner + "/" + repository.Name } +type githubRepositoryOwner struct { + Login string `json:"login"` +} + type githubRepository struct { - FullName string `json:"full_name"` - StargazersCount int `json:"stargazers_count"` - ForksCount int `json:"forks_count"` - OpenIssuesCount int `json:"open_issues_count"` + Name string `json:"name"` + FullName string `json:"full_name"` + StargazersCount int `json:"stargazers_count"` + ForksCount int `json:"forks_count"` + OpenIssuesCount int `json:"open_issues_count"` + Private bool `json:"private"` + Fork bool `json:"fork"` + Archived bool `json:"archived"` + Owner githubRepositoryOwner `json:"owner"` +} + +func (repository githubRepository) RepositoryRef() (githubRepositoryRef, error) { + if owner := strings.TrimSpace(repository.Owner.Login); owner != "" && strings.TrimSpace(repository.Name) != "" { + return githubRepositoryRef{ + Owner: owner, + Name: strings.TrimSpace(repository.Name), + }, nil + } + + if strings.TrimSpace(repository.FullName) != "" { + return parseGitHubRepositoryRef(repository.FullName) + } + + return githubRepositoryRef{}, fmt.Errorf("repository reference is missing owner/name fields") } type githubRelease struct { PublishedAt string `json:"published_at"` } +type githubRepositoryFetchOptions struct { + IncludeContributors bool + RequireLatestRelease bool +} + type githubRepositoryMetrics struct { ScopeKey string Stars int @@ -86,6 +148,7 @@ type githubRepositoryMetrics struct { OpenIssues int Commits30d int Commits90d int + PublicRepoCount int LatestReleasePublishedAt string } @@ -118,7 +181,7 @@ func (fetcher *GitHubFetcher) Fetch(ctx context.Context, request Request) (Resul return Result{}, fmt.Errorf("source ID is required") } - repositories, err := parseGitHubRepositoryRefs(request.Binding) + scope, err := parseGitHubBindingScope(request.Binding) if err != nil { return Result{}, err } @@ -128,46 +191,81 @@ func (fetcher *GitHubFetcher) Fetch(ctx context.Context, request Request) (Resul capturedAt = fetcher.now().UTC() } - snapshots := make([]githubSnapshotRecord, 0, len(repositories)*7) + switch scope.Kind { + case githubBindingScopeRepos: + snapshots := make([]githubSnapshotRecord, 0, len(scope.Repositories)*7) + + for _, repositoryRef := range scope.Repositories { + metrics, err := fetcher.fetchRepositoryMetrics( + ctx, + repositoryRef, + capturedAt, + githubRepositoryFetchOptions{ + IncludeContributors: true, + RequireLatestRelease: true, + }, + ) + if err != nil { + return Result{}, fmt.Errorf( + "binding %q repo %q: %w", + request.Binding.ID, + repositoryRef.FullName(), + err, + ) + } - for _, repositoryRef := range repositories { - metrics, err := fetcher.fetchRepositoryMetrics(ctx, repositoryRef, capturedAt) + snapshots = append(snapshots, buildGitHubRepositorySnapshotRecords(metrics)...) + } + + if err := persistGitHubMetricSnapshots(ctx, request.DB, request, capturedAt, snapshots); err != nil { + return Result{}, fmt.Errorf("persist metric snapshots: %w", err) + } + + return Result{ + RecordsWritten: len(scope.Repositories), + MetricsWritten: len(snapshots), + ContentItemsWritten: 0, + }, nil + + case githubBindingScopeOwner: + metrics, err := fetcher.fetchOwnerMetrics(ctx, scope, capturedAt) if err != nil { - return Result{}, fmt.Errorf( - "binding %q repo %q: %w", - request.Binding.ID, - repositoryRef.FullName(), - err, - ) + return Result{}, fmt.Errorf("binding %q owner %q: %w", request.Binding.ID, scope.Owner, err) } - snapshots = append(snapshots, buildGitHubSnapshotRecords(metrics)...) - } + snapshots := buildGitHubOwnerSnapshotRecords(metrics) + if err := persistGitHubMetricSnapshots(ctx, request.DB, request, capturedAt, snapshots); err != nil { + return Result{}, fmt.Errorf("persist metric snapshots: %w", err) + } - if err := persistGitHubMetricSnapshots(ctx, request.DB, request, capturedAt, snapshots); err != nil { - return Result{}, fmt.Errorf("persist metric snapshots: %w", err) - } + return Result{ + RecordsWritten: metrics.PublicRepoCount, + MetricsWritten: len(snapshots), + ContentItemsWritten: 0, + }, nil - return Result{ - RecordsWritten: len(repositories), - MetricsWritten: len(snapshots), - ContentItemsWritten: 0, - }, nil + default: + return Result{}, fmt.Errorf("binding %q has unsupported GitHub scope", request.Binding.ID) + } } func (fetcher *GitHubFetcher) fetchRepositoryMetrics( ctx context.Context, repositoryRef githubRepositoryRef, capturedAt time.Time, + options githubRepositoryFetchOptions, ) (githubRepositoryMetrics, error) { repository, err := fetcher.client.Repository(ctx, repositoryRef) if err != nil { return githubRepositoryMetrics{}, err } - contributors, err := fetcher.client.CountContributors(ctx, repositoryRef) - if err != nil { - return githubRepositoryMetrics{}, err + contributors := 0 + if options.IncludeContributors { + contributors, err = fetcher.client.CountContributors(ctx, repositoryRef) + if err != nil { + return githubRepositoryMetrics{}, err + } } commits30d, err := fetcher.client.CountCommitsSince(ctx, repositoryRef, capturedAt.Add(-30*24*time.Hour)) @@ -184,7 +282,7 @@ func (fetcher *GitHubFetcher) fetchRepositoryMetrics( if err != nil { return githubRepositoryMetrics{}, err } - if strings.TrimSpace(latestReleasePublishedAt) == "" { + if options.RequireLatestRelease && strings.TrimSpace(latestReleasePublishedAt) == "" { return githubRepositoryMetrics{}, fmt.Errorf( "repository %q has no latest release for metric %q", repositoryRef.FullName(), @@ -211,7 +309,58 @@ func (fetcher *GitHubFetcher) fetchRepositoryMetrics( }, nil } -func buildGitHubSnapshotRecords(metrics githubRepositoryMetrics) []githubSnapshotRecord { +func (fetcher *GitHubFetcher) fetchOwnerMetrics( + ctx context.Context, + scope githubBindingScope, + capturedAt time.Time, +) (githubRepositoryMetrics, error) { + repositories, err := fetcher.client.ListOwnerRepositories(ctx, scope.Owner, scope.OwnerType) + if err != nil { + return githubRepositoryMetrics{}, err + } + + aggregated := githubRepositoryMetrics{ + ScopeKey: "owner:" + strings.TrimSpace(scope.Owner), + PublicRepoCount: len(repositories), + } + + for _, repositoryRef := range repositories { + metrics, err := fetcher.fetchRepositoryMetrics( + ctx, + repositoryRef, + capturedAt, + githubRepositoryFetchOptions{ + IncludeContributors: false, + RequireLatestRelease: false, + }, + ) + if err != nil { + return githubRepositoryMetrics{}, fmt.Errorf("repo %q: %w", repositoryRef.FullName(), err) + } + + aggregated.Stars += metrics.Stars + aggregated.Forks += metrics.Forks + aggregated.OpenIssues += metrics.OpenIssues + aggregated.Commits30d += metrics.Commits30d + aggregated.Commits90d += metrics.Commits90d + aggregated.LatestReleasePublishedAt = maxRFC3339Timestamp( + aggregated.LatestReleasePublishedAt, + metrics.LatestReleasePublishedAt, + ) + } + + if strings.TrimSpace(aggregated.LatestReleasePublishedAt) == "" { + return githubRepositoryMetrics{}, fmt.Errorf( + "owner %q has no latest release across public repositories for metric %q", + scope.Owner, + MetricKeyGitHubLatestReleasePublishedAt, + ) + } + + return aggregated, nil +} + +func buildGitHubRepositorySnapshotRecords(metrics githubRepositoryMetrics) []githubSnapshotRecord { return []githubSnapshotRecord{ newNumericGitHubSnapshot(metrics.ScopeKey, MetricKeyGitHubStars, metrics.Stars), newNumericGitHubSnapshot(metrics.ScopeKey, MetricKeyGitHubForks, metrics.Forks), @@ -223,6 +372,18 @@ func buildGitHubSnapshotRecords(metrics githubRepositoryMetrics) []githubSnapsho } } +func buildGitHubOwnerSnapshotRecords(metrics githubRepositoryMetrics) []githubSnapshotRecord { + return []githubSnapshotRecord{ + newNumericGitHubSnapshot(metrics.ScopeKey, MetricKeyGitHubPublicRepoCount, metrics.PublicRepoCount), + newNumericGitHubSnapshot(metrics.ScopeKey, MetricKeyGitHubStars, metrics.Stars), + newNumericGitHubSnapshot(metrics.ScopeKey, MetricKeyGitHubForks, metrics.Forks), + newNumericGitHubSnapshot(metrics.ScopeKey, MetricKeyGitHubOpenIssues, metrics.OpenIssues), + newNumericGitHubSnapshot(metrics.ScopeKey, MetricKeyGitHubCommits30d, metrics.Commits30d), + newNumericGitHubSnapshot(metrics.ScopeKey, MetricKeyGitHubCommits90d, metrics.Commits90d), + newTextGitHubSnapshot(metrics.ScopeKey, MetricKeyGitHubLatestReleasePublishedAt, metrics.LatestReleasePublishedAt), + } +} + func newNumericGitHubSnapshot(scopeKey, metricKey string, value int) githubSnapshotRecord { numericValue := float64(value) @@ -282,29 +443,80 @@ func persistGitHubMetricSnapshots( return nil } -func parseGitHubRepositoryRefs(binding config.Binding) ([]githubRepositoryRef, error) { +func parseGitHubBindingScope(binding config.Binding) (githubBindingScope, error) { + repositories, hasRepos, err := parseOptionalGitHubRepositoryRefs(binding) + if err != nil { + return githubBindingScope{}, err + } + + owner, hasOwner, err := parseOptionalGitHubOwner(binding) + if err != nil { + return githubBindingScope{}, err + } + + ownerType, hasOwnerType, err := parseOptionalGitHubOwnerType(binding) + if err != nil { + return githubBindingScope{}, err + } + + switch { + case hasRepos && (hasOwner || hasOwnerType): + return githubBindingScope{}, fmt.Errorf("binding %q must not combine scope.repos with scope.owner/owner_type", binding.ID) + case hasRepos: + return githubBindingScope{ + Kind: githubBindingScopeRepos, + Repositories: repositories, + }, nil + case hasOwner || hasOwnerType: + if !hasOwner { + return githubBindingScope{}, fmt.Errorf("binding %q scope.owner is required when scope.owner_type is set", binding.ID) + } + if !hasOwnerType { + return githubBindingScope{}, fmt.Errorf("binding %q scope.owner_type is required when scope.owner is set", binding.ID) + } + + return githubBindingScope{ + Kind: githubBindingScopeOwner, + Owner: owner, + OwnerType: ownerType, + }, nil + default: + return githubBindingScope{}, fmt.Errorf("binding %q must define either scope.repos or scope.owner/owner_type", binding.ID) + } +} + +func parseOptionalGitHubRepositoryRefs(binding config.Binding) ([]githubRepositoryRef, bool, error) { rawRepos, ok := binding.Scope["repos"] if !ok { - return nil, fmt.Errorf("binding %q scope.repos is required", binding.ID) + return nil, false, nil + } + + repositories, err := parseGitHubRepositoryRefsValue(binding.ID, rawRepos) + if err != nil { + return nil, false, err } + return repositories, true, nil +} + +func parseGitHubRepositoryRefsValue(bindingID string, rawRepos any) ([]githubRepositoryRef, error) { switch typed := rawRepos.(type) { case []string: - return parseGitHubRepositoryRefStrings(binding.ID, typed) + return parseGitHubRepositoryRefStrings(bindingID, typed) case []any: values := make([]string, 0, len(typed)) for index, rawValue := range typed { value, ok := rawValue.(string) if !ok { - return nil, fmt.Errorf("binding %q scope.repos[%d] must be a string", binding.ID, index) + return nil, fmt.Errorf("binding %q scope.repos[%d] must be a string", bindingID, index) } values = append(values, value) } - return parseGitHubRepositoryRefStrings(binding.ID, values) + return parseGitHubRepositoryRefStrings(bindingID, values) default: - return nil, fmt.Errorf("binding %q scope.repos must be an array of repository strings", binding.ID) + return nil, fmt.Errorf("binding %q scope.repos must be an array of repository strings", bindingID) } } @@ -326,6 +538,47 @@ func parseGitHubRepositoryRefStrings(bindingID string, values []string) ([]githu return repositories, nil } +func parseOptionalGitHubOwner(binding config.Binding) (string, bool, error) { + rawOwner, ok := binding.Scope["owner"] + if !ok { + return "", false, nil + } + + owner, ok := rawOwner.(string) + if !ok { + return "", false, fmt.Errorf("binding %q scope.owner must be a string", binding.ID) + } + + owner = strings.TrimSpace(owner) + if owner == "" { + return "", false, fmt.Errorf("binding %q scope.owner must not be empty", binding.ID) + } + + return owner, true, nil +} + +func parseOptionalGitHubOwnerType(binding config.Binding) (string, bool, error) { + rawOwnerType, ok := binding.Scope["owner_type"] + if !ok { + return "", false, nil + } + + ownerType, ok := rawOwnerType.(string) + if !ok { + return "", false, fmt.Errorf("binding %q scope.owner_type must be a string", binding.ID) + } + + ownerType = strings.TrimSpace(strings.ToLower(ownerType)) + switch ownerType { + case "organization", "user": + return ownerType, true, nil + case "": + return "", false, fmt.Errorf("binding %q scope.owner_type must not be empty", binding.ID) + default: + return "", false, fmt.Errorf("binding %q scope.owner_type %q is unsupported", binding.ID, ownerType) + } +} + func parseGitHubRepositoryRef(value string) (githubRepositoryRef, error) { trimmed := strings.TrimSpace(value) if trimmed == "" { @@ -354,16 +607,19 @@ type GitHubClientOptions struct { BaseURL string HTTPClient *http.Client UserAgent string + Token string } type GitHubClient struct { baseURL *url.URL httpClient *http.Client userAgent string + token string } func NewGitHubClient(options GitHubClientOptions) *GitHubClient { baseURL := normalizeGitHubBaseURL(options.BaseURL) + httpClient := options.HTTPClient if httpClient == nil { httpClient = http.DefaultClient @@ -374,11 +630,77 @@ func NewGitHubClient(options GitHubClientOptions) *GitHubClient { userAgent = defaultGitHubUserAgent } + token := strings.TrimSpace(options.Token) + return &GitHubClient{ baseURL: mustParseGitHubBaseURL(baseURL), httpClient: httpClient, userAgent: userAgent, + token: token, + } +} + +func (client *GitHubClient) ListOwnerRepositories(ctx context.Context, owner, ownerType string) ([]githubRepositoryRef, error) { + requestURL, err := client.ownerRepositoriesURL(owner, ownerType) + if err != nil { + return nil, err + } + + repositories := make([]githubRepositoryRef, 0) + seen := make(map[string]struct{}) + + for { + response, err := client.doRequestURL(ctx, http.MethodGet, requestURL) + if err != nil { + return nil, err + } + + if response.StatusCode != http.StatusOK { + responseErr := client.responseError("list owner repositories", response) + _ = response.Body.Close() + return nil, responseErr + } + + var page []githubRepository + if err := json.NewDecoder(response.Body).Decode(&page); err != nil { + _ = response.Body.Close() + return nil, fmt.Errorf("decode owner repositories response: %w", err) + } + + nextRequestURL, hasNext := parseLinkRelationURL(response.Header.Get("Link"), "next") + _ = response.Body.Close() + + for _, repository := range page { + if repository.Private { + continue + } + + repositoryRef, err := repository.RepositoryRef() + if err != nil { + return nil, fmt.Errorf("parse listed repository: %w", err) + } + + fullName := repositoryRef.FullName() + if _, exists := seen[fullName]; exists { + continue + } + + seen[fullName] = struct{}{} + repositories = append(repositories, repositoryRef) + } + + if !hasNext { + break + } + + requestURL = nextRequestURL } + + sort.Slice(repositories, func(i, j int) bool { + return repositories[i].FullName() < repositories[j].FullName() + }) + + return repositories, nil } func (client *GitHubClient) Repository(ctx context.Context, repositoryRef githubRepositoryRef) (githubRepository, error) { @@ -511,50 +833,57 @@ func countFromSingleItemPage(linkHeader string, itemsOnPage int) (int, error) { return itemsOnPage, nil } -func parseLastPage(linkHeader string) (int, bool, error) { +func parseLinkRelationURL(linkHeader, relation string) (string, bool) { linkHeader = strings.TrimSpace(linkHeader) if linkHeader == "" { - return 0, false, nil + return "", false } - parts := strings.Split(linkHeader, ",") - for _, part := range parts { + for _, part := range strings.Split(linkHeader, ",") { sections := strings.Split(strings.TrimSpace(part), ";") if len(sections) < 2 { continue } - rel := strings.TrimSpace(sections[1]) - if !strings.Contains(rel, `rel="last"`) { - continue - } - rawURL := strings.TrimSpace(sections[0]) rawURL = strings.TrimPrefix(rawURL, "<") rawURL = strings.TrimSuffix(rawURL, ">") - parsed, err := url.Parse(rawURL) - if err != nil { - return 0, false, fmt.Errorf("parse Link header URL %q: %w", rawURL, err) + for _, section := range sections[1:] { + if strings.Contains(strings.TrimSpace(section), fmt.Sprintf(`rel="%s"`, relation)) { + return rawURL, true + } } + } - pageValue := strings.TrimSpace(parsed.Query().Get("page")) - if pageValue == "" { - return 0, false, fmt.Errorf("Link header last page is missing page query parameter") - } + return "", false +} - page, err := strconv.Atoi(pageValue) - if err != nil { - return 0, false, fmt.Errorf("parse Link header page value %q: %w", pageValue, err) - } - if page < 1 { - return 0, false, fmt.Errorf("Link header last page must be greater than zero") - } +func parseLastPage(linkHeader string) (int, bool, error) { + rawURL, ok := parseLinkRelationURL(linkHeader, "last") + if !ok { + return 0, false, nil + } + + parsed, err := url.Parse(rawURL) + if err != nil { + return 0, false, fmt.Errorf("parse Link header URL %q: %w", rawURL, err) + } + + pageValue := strings.TrimSpace(parsed.Query().Get("page")) + if pageValue == "" { + return 0, false, fmt.Errorf("Link header last page is missing page query parameter") + } - return page, true, nil + page, err := strconv.Atoi(pageValue) + if err != nil { + return 0, false, fmt.Errorf("parse Link header page value %q: %w", pageValue, err) + } + if page < 1 { + return 0, false, fmt.Errorf("Link header last page must be greater than zero") } - return 0, false, nil + return page, true, nil } func (client *GitHubClient) doRequest( @@ -568,18 +897,30 @@ func (client *GitHubClient) doRequest( RawQuery: query.Encode(), }) - request, err := http.NewRequestWithContext(ctx, method, requestURL.String(), nil) + return client.doRequestURL(ctx, method, requestURL.String()) +} + +func (client *GitHubClient) doRequestURL(ctx context.Context, method, rawURL string) (*http.Response, error) { + requestURL, err := client.resolveURL(rawURL) if err != nil { - return nil, fmt.Errorf("build GitHub request %s %s: %w", method, requestURL.String(), err) + return nil, fmt.Errorf("resolve GitHub request URL %q: %w", rawURL, err) + } + + request, err := http.NewRequestWithContext(ctx, method, requestURL, nil) + if err != nil { + return nil, fmt.Errorf("build GitHub request %s %s: %w", method, requestURL, err) } request.Header.Set("Accept", "application/vnd.github+json") request.Header.Set("User-Agent", client.userAgent) request.Header.Set("X-GitHub-Api-Version", "2022-11-28") + if client.token != "" { + request.Header.Set("Authorization", "Bearer "+client.token) + } response, err := client.httpClient.Do(request) if err != nil { - return nil, fmt.Errorf("perform GitHub request %s %s: %w", method, requestURL.String(), err) + return nil, fmt.Errorf("perform GitHub request %s %s: %w", method, requestURL, err) } return response, nil @@ -604,15 +945,19 @@ func (client *GitHubClient) responseError(action string, response *http.Response message = "unknown GitHub API error" } + requestMethod := "" requestPath := "" - if response.Request != nil && response.Request.URL != nil { - requestPath = response.Request.URL.Path + if response.Request != nil { + requestMethod = response.Request.Method + if response.Request.URL != nil { + requestPath = response.Request.URL.Path + } } return fmt.Errorf( "%s: GitHub API %s %s returned %d: %s", action, - response.Request.Method, + requestMethod, requestPath, response.StatusCode, message, @@ -627,6 +972,53 @@ func repositoryPath(repositoryRef githubRepositoryRef) string { ) } +func (client *GitHubClient) ownerRepositoriesURL(owner, ownerType string) (string, error) { + path, query, err := ownerRepositoriesPath(owner, ownerType) + if err != nil { + return "", err + } + + query.Set("per_page", "100") + + return client.baseURL.ResolveReference(&url.URL{ + Path: strings.TrimPrefix(path, "/"), + RawQuery: query.Encode(), + }).String(), nil +} + +func ownerRepositoriesPath(owner, ownerType string) (string, url.Values, error) { + owner = strings.TrimSpace(owner) + if owner == "" { + return "", nil, fmt.Errorf("owner is required") + } + + ownerType = strings.TrimSpace(strings.ToLower(ownerType)) + query := url.Values{} + + switch ownerType { + case "organization": + query.Set("type", "public") + return fmt.Sprintf("orgs/%s/repos", url.PathEscape(owner)), query, nil + case "user": + query.Set("type", "owner") + return fmt.Sprintf("users/%s/repos", url.PathEscape(owner)), query, nil + default: + return "", nil, fmt.Errorf("unsupported owner_type %q", ownerType) + } +} + +func (client *GitHubClient) resolveURL(rawURL string) (string, error) { + parsed, err := url.Parse(strings.TrimSpace(rawURL)) + if err != nil { + return "", err + } + if parsed.IsAbs() { + return parsed.String(), nil + } + + return client.baseURL.ResolveReference(parsed).String(), nil +} + func normalizeGitHubBaseURL(value string) string { trimmed := strings.TrimSpace(value) if trimmed == "" { @@ -647,3 +1039,14 @@ func mustParseGitHubBaseURL(value string) *url.URL { return parsed } + +func maxRFC3339Timestamp(current, candidate string) string { + current = strings.TrimSpace(current) + candidate = strings.TrimSpace(candidate) + + if candidate > current { + return candidate + } + + return current +} diff --git a/internal/fetch/github_test.go b/internal/fetch/github_test.go index 898d4a4..65ef2eb 100644 --- a/internal/fetch/github_test.go +++ b/internal/fetch/github_test.go @@ -10,6 +10,7 @@ import ( "testing" "time" + "github.com/CrisSTEM/signalscope/internal/config" "github.com/CrisSTEM/signalscope/internal/storage" ) @@ -22,8 +23,10 @@ type metricSnapshotRow struct { } type githubTestServerOptions struct { - Repo2Status int - Repo1ReleaseStatus int + Repo2Status int + Repo1ReleaseStatus int + Repo2ReleaseStatus int + RequireAuthorization string } func TestGitHubFetcherPersistsRepositorySnapshots(t *testing.T) { @@ -314,6 +317,202 @@ func TestGitHubFetcherFailsWhenRepositoryHasNoLatestRelease(t *testing.T) { } } +func TestGitHubFetcherPersistsOwnerScopedSnapshots(t *testing.T) { + t.Parallel() + + ctx := context.Background() + capturedAt := time.Date(2026, 4, 20, 15, 0, 0, 0, time.UTC) + + server := newGitHubTestServer(t, capturedAt, githubTestServerOptions{ + Repo2ReleaseStatus: http.StatusNotFound, + }) + defer server.Close() + + pack := ownerRuntimeFixturePack("github-owner-fetch-success") + db := openRuntimeTestDB(t, pack) + defer db.Close() + + registry := NewRegistry() + registry.MustRegister(SourceKindGitHub, newGitHubTestFetcher(server)) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: sequenceClock( + capturedAt, + capturedAt.Add(1*time.Second), + ), + } + + summary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindGitHub, + BindingID: "github-owner-binding", + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + + if summary.Attempted != 1 { + t.Fatalf("summary.Attempted = %d, want %d", summary.Attempted, 1) + } + if summary.Succeeded != 1 { + t.Fatalf("summary.Succeeded = %d, want %d", summary.Succeeded, 1) + } + if summary.Failed != 0 { + t.Fatalf("summary.Failed = %d, want %d", summary.Failed, 0) + } + if summary.RecordsWritten != 3 { + t.Fatalf("summary.RecordsWritten = %d, want %d", summary.RecordsWritten, 3) + } + if summary.MetricsWritten != 7 { + t.Fatalf("summary.MetricsWritten = %d, want %d", summary.MetricsWritten, 7) + } + + rows, err := readMetricSnapshotsForBinding(ctx, db, "github-owner-binding") + if err != nil { + t.Fatalf("readMetricSnapshotsForBinding() error = %v", err) + } + if len(rows) != 7 { + t.Fatalf("len(rows) = %d, want %d", len(rows), 7) + } + + expectedNumeric := map[string]int{ + MetricKeyGitHubPublicRepoCount: 3, + MetricKeyGitHubStars: 18, + MetricKeyGitHubForks: 7, + MetricKeyGitHubOpenIssues: 10, + MetricKeyGitHubCommits30d: 10, + MetricKeyGitHubCommits90d: 23, + } + expectedText := map[string]string{ + MetricKeyGitHubLatestReleasePublishedAt: "2026-04-10T00:00:00Z", + } + + for _, row := range rows { + if row.ScopeKey != "owner:example" { + t.Fatalf("row.ScopeKey = %q, want %q", row.ScopeKey, "owner:example") + } + if row.MetricKey == MetricKeyGitHubContributors { + t.Fatalf("unexpected owner-scoped metric key %q", row.MetricKey) + } + + if expectedValue, ok := expectedNumeric[row.MetricKey]; ok { + if !row.MetricValueNum.Valid { + t.Fatalf("row %q numeric value is invalid, want %d", row.MetricKey, expectedValue) + } + if int(row.MetricValueNum.Float64) != expectedValue { + t.Fatalf("row %q numeric value = %v, want %d", row.MetricKey, row.MetricValueNum.Float64, expectedValue) + } + continue + } + + if expectedValue, ok := expectedText[row.MetricKey]; ok { + if !row.MetricValueText.Valid { + t.Fatalf("row %q text value is invalid, want %q", row.MetricKey, expectedValue) + } + if row.MetricValueText.String != expectedValue { + t.Fatalf("row %q text value = %q, want %q", row.MetricKey, row.MetricValueText.String, expectedValue) + } + continue + } + + t.Fatalf("unexpected metric key %q", row.MetricKey) + } +} + +func TestGitHubFetcherUsesOptionalTokenAuthWhenConfigured(t *testing.T) { + t.Parallel() + + ctx := context.Background() + capturedAt := time.Date(2026, 4, 20, 16, 0, 0, 0, time.UTC) + + server := newGitHubTestServer(t, capturedAt, githubTestServerOptions{ + RequireAuthorization: "Bearer owner-token", + }) + defer server.Close() + + pack := ownerRuntimeFixturePack("github-owner-fetch-token") + db := openRuntimeTestDB(t, pack) + defer db.Close() + + registry := NewRegistry() + registry.MustRegister(SourceKindGitHub, NewGitHubFetcher(GitHubFetcherOptions{ + ClientOptions: GitHubClientOptions{ + BaseURL: server.URL + "/", + HTTPClient: server.Client(), + UserAgent: "signalscope-test", + }, + TokenLookup: func() string { + // This simulates the default GITHUB_TOKEN lookup path deterministically. + return "owner-token" + }, + })) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: sequenceClock( + capturedAt, + capturedAt.Add(1*time.Second), + ), + } + + summary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindGitHub, + BindingID: "github-owner-binding", + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + if summary.Failed != 0 { + t.Fatalf("summary.Failed = %d, want %d", summary.Failed, 0) + } + if summary.Succeeded != 1 { + t.Fatalf("summary.Succeeded = %d, want %d", summary.Succeeded, 1) + } +} + +func TestGitHubClientListOwnerRepositoriesSupportsUserOwners(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + w.Header().Set("Content-Type", "application/json") + + if request.URL.Path != "/users/example-user/repos" { + http.Error(w, `{"message":"unexpected path"}`, http.StatusNotFound) + return + } + if request.URL.Query().Get("type") != "owner" { + http.Error(w, `{"message":"expected type=owner"}`, http.StatusBadRequest) + return + } + if request.URL.Query().Get("per_page") != "100" { + http.Error(w, `{"message":"expected per_page=100"}`, http.StatusBadRequest) + return + } + + fmt.Fprint(w, `[{"name":"repo-1","full_name":"example-user/repo-1","owner":{"login":"example-user"},"private":false}]`) + })) + defer server.Close() + + client := NewGitHubClient(GitHubClientOptions{ + BaseURL: server.URL + "/", + HTTPClient: server.Client(), + UserAgent: "signalscope-test", + }) + + repositories, err := client.ListOwnerRepositories(context.Background(), "example-user", "user") + if err != nil { + t.Fatalf("ListOwnerRepositories() error = %v", err) + } + if len(repositories) != 1 { + t.Fatalf("len(repositories) = %d, want %d", len(repositories), 1) + } + if repositories[0].FullName() != "example-user/repo-1" { + t.Fatalf("repositories[0].FullName() = %q, want %q", repositories[0].FullName(), "example-user/repo-1") + } +} + func newGitHubTestFetcher(server *httptest.Server) *GitHubFetcher { return NewGitHubFetcher(GitHubFetcherOptions{ Client: NewGitHubClient(GitHubClientOptions{ @@ -340,10 +539,42 @@ func newGitHubTestServer(t *testing.T, capturedAt time.Time, options githubTestS repo1ReleaseStatus = http.StatusOK } + repo2ReleaseStatus := options.Repo2ReleaseStatus + if repo2ReleaseStatus == 0 { + repo2ReleaseStatus = http.StatusOK + } + + requiredAuthorization := strings.TrimSpace(options.RequireAuthorization) + handler := http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + if requiredAuthorization != "" && request.Header.Get("Authorization") != requiredAuthorization { + http.Error(w, `{"message":"missing or invalid authorization"}`, http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "application/json") switch request.URL.Path { + case "/orgs/example/repos": + if request.URL.Query().Get("per_page") != "100" { + http.Error(w, `{"message":"expected per_page=100"}`, http.StatusBadRequest) + return + } + if request.URL.Query().Get("type") != "public" { + http.Error(w, `{"message":"expected type=public"}`, http.StatusBadRequest) + return + } + + switch request.URL.Query().Get("page") { + case "", "1": + w.Header().Set("Link", `; rel="next", ; rel="last"`) + fmt.Fprint(w, `[{"name":"repo-1","full_name":"example/repo-1","owner":{"login":"example"},"archived":false,"fork":false,"private":false},{"name":"repo-2","full_name":"example/repo-2","owner":{"login":"example"},"archived":true,"fork":false,"private":false}]`) + case "2": + fmt.Fprint(w, `[{"name":"repo-3","full_name":"example/repo-3","owner":{"login":"example"},"archived":false,"fork":true,"private":false}]`) + default: + http.Error(w, `{"message":"unexpected page"}`, http.StatusBadRequest) + } + case "/repos/example/repo-1": fmt.Fprint(w, `{"full_name":"example/repo-1","stargazers_count":10,"forks_count":4,"open_issues_count":7}`) @@ -412,8 +643,36 @@ func newGitHubTestServer(t *testing.T, capturedAt time.Time, options githubTestS } case "/repos/example/repo-2/releases/latest": + if repo2ReleaseStatus == http.StatusNotFound { + w.WriteHeader(http.StatusNotFound) + fmt.Fprint(w, `{"message":"Not Found"}`) + return + } + fmt.Fprint(w, `{"published_at":"2026-03-15T00:00:00Z"}`) + case "/repos/example/repo-3": + fmt.Fprint(w, `{"full_name":"example/repo-3","stargazers_count":2,"forks_count":1,"open_issues_count":0}`) + + case "/repos/example/repo-3/commits": + if request.URL.Query().Get("per_page") != "1" { + http.Error(w, `{"message":"expected per_page=1"}`, http.StatusBadRequest) + return + } + + switch request.URL.Query().Get("since") { + case capturedAt.Add(-30 * 24 * time.Hour).Format(time.RFC3339): + fmt.Fprint(w, `[{"sha":"repo-3-30d"}]`) + case capturedAt.Add(-90 * 24 * time.Hour).Format(time.RFC3339): + w.Header().Set("Link", `; rel="last"`) + fmt.Fprint(w, `[{"sha":"repo-3-90d"}]`) + default: + http.Error(w, `{"message":"unexpected since query"}`, http.StatusBadRequest) + } + + case "/repos/example/repo-3/releases/latest": + fmt.Fprint(w, `{"published_at":"2026-04-10T00:00:00Z"}`) + default: http.Error(w, `{"message":"unexpected path"}`, http.StatusNotFound) } @@ -422,6 +681,60 @@ func newGitHubTestServer(t *testing.T, capturedAt time.Time, options githubTestS return httptest.NewServer(handler) } +func ownerRuntimeFixturePack(datasetID string) config.Pack { + return config.Pack{ + Entities: config.EntitiesFile{ + Version: 1, + DatasetID: datasetID, + DatasetName: "GitHub Owner Fixture Dataset", + Entities: []config.Entity{ + { + ID: "org-1", + Slug: "org-1", + Kind: "organization", + Name: "Org 1", + }, + }, + Relationships: []config.Relationship{}, + }, + Sources: config.SourcesFile{ + Version: 1, + DatasetID: datasetID, + Sources: []config.Source{ + { + ID: "github", + Kind: "github", + Enabled: true, + Defaults: config.JSONMap{}, + }, + }, + Bindings: []config.Binding{ + { + ID: "github-owner-binding", + EntityID: "org-1", + SourceID: "github", + Enabled: true, + Scope: config.JSONMap{ + "owner": "example", + "owner_type": "organization", + }, + Notes: "owner-scoped github binding", + }, + }, + }, + Alerts: config.AlertsFile{ + Version: 1, + DatasetID: datasetID, + Rules: []config.AlertRule{}, + }, + Schedules: config.SchedulesFile{ + Version: 1, + DatasetID: datasetID, + Jobs: []config.ScheduleJob{}, + }, + } +} + func readMetricSnapshotsForBinding(ctx context.Context, db *sql.DB, bindingID string) ([]metricSnapshotRow, error) { rows, err := db.QueryContext( ctx, @@ -437,7 +750,6 @@ func readMetricSnapshotsForBinding(ctx context.Context, db *sql.DB, bindingID st defer rows.Close() var result []metricSnapshotRow - for rows.Next() { var row metricSnapshotRow if err := rows.Scan( diff --git a/internal/fetch/news.go b/internal/fetch/news.go new file mode 100644 index 0000000..7b82cb4 --- /dev/null +++ b/internal/fetch/news.go @@ -0,0 +1,480 @@ +package fetch + +import ( + "context" + "encoding/json" + "encoding/xml" + "fmt" + "io" + "math" + "net/http" + "net/url" + "strings" + "time" + + "github.com/CrisSTEM/signalscope/internal/config" + "github.com/CrisSTEM/signalscope/internal/storage" +) + +const ( + SourceKindNewsRSS = "news_rss" + MetricKeyNewsArticleCount = "news.article_count" + ContentItemTypeNewsArticle = "news_article" + + newsProviderGoogleNewsRSS = "google_news_rss" + + defaultNewsRSSBaseURL = "https://news.google.com/" + defaultNewsRSSUserAgent = "signalscope-fetch/0.3.0" +) + +type NewsRSSFetcherOptions struct { + BaseURL string + HTTPClient *http.Client + UserAgent string + Now func() time.Time +} + +type NewsRSSFetcher struct { + baseURL *url.URL + httpClient *http.Client + userAgent string + now func() time.Time +} + +type newsRSSBindingScope struct { + Query string + MaxItems int +} + +type newsRSSFeed struct { + XMLName xml.Name `xml:"rss"` + Channel newsRSSChannel `xml:"channel"` +} + +type newsRSSChannel struct { + Items []newsRSSFeedItem `xml:"item"` +} + +type newsRSSFeedItem struct { + Title string `xml:"title"` + Link string `xml:"link"` + Description string `xml:"description"` + GUID newsRSSFeedGUID `xml:"guid"` + PubDate string `xml:"pubDate"` + Source newsRSSFeedSource `xml:"source"` +} + +type newsRSSFeedGUID struct { + Value string `xml:",chardata"` +} + +type newsRSSFeedSource struct { + URL string `xml:"url,attr"` + Value string `xml:",chardata"` +} + +type newsArticleMetadata struct { + Provider string `json:"provider"` + Query string `json:"query,omitempty"` + GUID string `json:"guid,omitempty"` + Publisher string `json:"publisher,omitempty"` + PublisherURL string `json:"publisher_url,omitempty"` + RawPubDate string `json:"raw_pub_date,omitempty"` +} + +func NewNewsRSSFetcher(options NewsRSSFetcherOptions) *NewsRSSFetcher { + httpClient := options.HTTPClient + if httpClient == nil { + httpClient = http.DefaultClient + } + + userAgent := strings.TrimSpace(options.UserAgent) + if userAgent == "" { + userAgent = defaultNewsRSSUserAgent + } + + now := options.Now + if now == nil { + now = time.Now + } + + return &NewsRSSFetcher{ + baseURL: mustParseNewsRSSBaseURL(normalizeNewsRSSBaseURL(options.BaseURL)), + httpClient: httpClient, + userAgent: userAgent, + now: now, + } +} + +func (fetcher *NewsRSSFetcher) Fetch(ctx context.Context, request Request) (Result, error) { + ctx = normalizeContext(ctx) + + if request.DB == nil { + return Result{}, fmt.Errorf("database handle is required") + } + if request.FetchRun.ID <= 0 { + return Result{}, fmt.Errorf("fetch run is required") + } + if strings.TrimSpace(request.DatasetID) == "" { + return Result{}, fmt.Errorf("dataset ID is required") + } + if strings.TrimSpace(request.Binding.ID) == "" { + return Result{}, fmt.Errorf("binding ID is required") + } + if strings.TrimSpace(request.Binding.EntityID) == "" { + return Result{}, fmt.Errorf("binding %q entity_id is required", request.Binding.ID) + } + if strings.TrimSpace(request.Source.ID) == "" { + return Result{}, fmt.Errorf("source ID is required") + } + + scope, err := parseNewsRSSBindingScope(request.Binding) + if err != nil { + return Result{}, err + } + + capturedAt := request.FetchRun.StartedAt.UTC() + if capturedAt.IsZero() { + capturedAt = fetcher.now().UTC() + } + + feedItems, err := fetcher.fetchFeedItems(ctx, scope.Query) + if err != nil { + return Result{}, fmt.Errorf("binding %q query %q: %w", request.Binding.ID, scope.Query, err) + } + + contentItems, err := fetcher.normalizeItems(scope, feedItems, capturedAt) + if err != nil { + return Result{}, fmt.Errorf("normalize news RSS items: %w", err) + } + + articleCount := float64(len(contentItems)) + metricSnapshots := []storage.MetricSnapshotInput{ + { + MetricKey: MetricKeyNewsArticleCount, + MetricValueNum: &articleCount, + Unit: "count", + WindowKey: "point_in_time", + CapturedAt: capturedAt, + }, + } + + metricsInserted, contentItemsInserted, err := storage.InsertMetricSnapshotsAndContentItems( + ctx, + request.DB, + request.FetchRun.ID, + metricSnapshots, + contentItems, + ) + if err != nil { + return Result{}, fmt.Errorf("persist news observations: %w", err) + } + if metricsInserted != len(metricSnapshots) { + return Result{}, fmt.Errorf("inserted %d metric snapshots, want %d", metricsInserted, len(metricSnapshots)) + } + + return Result{ + RecordsWritten: len(contentItems), + MetricsWritten: metricsInserted, + ContentItemsWritten: contentItemsInserted, + }, nil +} + +func (fetcher *NewsRSSFetcher) fetchFeedItems(ctx context.Context, query string) ([]newsRSSFeedItem, error) { + response, err := fetcher.doRequest(ctx, query) + if err != nil { + return nil, err + } + defer response.Body.Close() + + if response.StatusCode != http.StatusOK { + return nil, fetcher.responseError(response) + } + + var feed newsRSSFeed + if err := xml.NewDecoder(response.Body).Decode(&feed); err != nil { + return nil, fmt.Errorf("decode news RSS response: %w", err) + } + + return feed.Channel.Items, nil +} + +func (fetcher *NewsRSSFetcher) normalizeItems( + scope newsRSSBindingScope, + feedItems []newsRSSFeedItem, + discoveredAt time.Time, +) ([]storage.ContentItemInput, error) { + normalized := make([]storage.ContentItemInput, 0, len(feedItems)) + seen := make(map[string]struct{}, len(feedItems)) + + for _, item := range feedItems { + normalizedItem, skip, err := fetcher.normalizeItem(scope, item, discoveredAt) + if err != nil { + return nil, err + } + if skip { + continue + } + if _, exists := seen[normalizedItem.DedupeKey]; exists { + continue + } + + seen[normalizedItem.DedupeKey] = struct{}{} + normalized = append(normalized, normalizedItem) + + if scope.MaxItems > 0 && len(normalized) >= scope.MaxItems { + break + } + } + + return normalized, nil +} + +func (fetcher *NewsRSSFetcher) normalizeItem( + scope newsRSSBindingScope, + item newsRSSFeedItem, + discoveredAt time.Time, +) (storage.ContentItemInput, bool, error) { + resolvedURL := resolveNewsRSSLink(fetcher.baseURL, item.Link) + publishedAt, rawPubDate := parseNewsRSSPublishedAt(item.PubDate) + + metadataJSON, err := buildNewsArticleMetadata(scope.Query, item, rawPubDate) + if err != nil { + return storage.ContentItemInput{}, false, err + } + + input := storage.ContentItemInput{ + ItemType: ContentItemTypeNewsArticle, + Title: strings.TrimSpace(item.Title), + Summary: strings.TrimSpace(item.Description), + URL: resolvedURL, + ExternalID: strings.TrimSpace(item.GUID.Value), + PublishedAt: publishedAt, + DiscoveredAt: discoveredAt, + MetadataJSON: metadataJSON, + } + + dedupeKey, err := storage.BuildContentItemDedupeKey(input) + if err != nil { + return storage.ContentItemInput{}, true, nil + } + input.DedupeKey = dedupeKey + + return input, false, nil +} + +func buildNewsArticleMetadata(query string, item newsRSSFeedItem, rawPubDate string) (string, error) { + metadata := newsArticleMetadata{ + Provider: newsProviderGoogleNewsRSS, + Query: strings.TrimSpace(query), + GUID: strings.TrimSpace(item.GUID.Value), + Publisher: strings.TrimSpace(item.Source.Value), + PublisherURL: strings.TrimSpace(item.Source.URL), + RawPubDate: strings.TrimSpace(rawPubDate), + } + + encoded, err := json.Marshal(metadata) + if err != nil { + return "", fmt.Errorf("marshal news article metadata: %w", err) + } + + return string(encoded), nil +} + +func parseNewsRSSBindingScope(binding config.Binding) (newsRSSBindingScope, error) { + rawQuery, exists := binding.Scope["query"] + if !exists { + return newsRSSBindingScope{}, fmt.Errorf("binding %q scope.query is required", binding.ID) + } + + query, ok := rawQuery.(string) + if !ok { + return newsRSSBindingScope{}, fmt.Errorf("binding %q scope.query must be a string", binding.ID) + } + query = strings.TrimSpace(query) + if query == "" { + return newsRSSBindingScope{}, fmt.Errorf("binding %q scope.query must not be empty", binding.ID) + } + + scope := newsRSSBindingScope{ + Query: query, + } + + if rawMaxItems, exists := binding.Scope["max_items"]; exists { + maxItems, err := parsePositiveWholeNumber(binding.ID, "max_items", rawMaxItems) + if err != nil { + return newsRSSBindingScope{}, err + } + scope.MaxItems = maxItems + } + + return scope, nil +} + +func parsePositiveWholeNumber(bindingID, field string, value any) (int, error) { + number, ok := positiveWholeNumberValue(value) + if !ok || number <= 0 { + return 0, fmt.Errorf("binding %q scope.%s must be a positive whole number", bindingID, field) + } + + return number, nil +} + +func positiveWholeNumberValue(value any) (int, bool) { + switch typed := value.(type) { + case int: + return typed, true + case int8: + return int(typed), true + case int16: + return int(typed), true + case int32: + return int(typed), true + case int64: + return int(typed), true + case uint: + return int(typed), true + case uint8: + return int(typed), true + case uint16: + return int(typed), true + case uint32: + return int(typed), true + case uint64: + return int(typed), true + case float32: + if math.Trunc(float64(typed)) != float64(typed) { + return 0, false + } + return int(typed), true + case float64: + if math.Trunc(typed) != typed { + return 0, false + } + return int(typed), true + default: + return 0, false + } +} + +func parseNewsRSSPublishedAt(value string) (*time.Time, string) { + raw := strings.TrimSpace(value) + if raw == "" { + return nil, "" + } + + layouts := []string{ + time.RFC1123Z, + time.RFC1123, + time.RFC822Z, + time.RFC822, + time.RFC850, + time.RFC3339, + } + + for _, layout := range layouts { + parsed, err := time.Parse(layout, raw) + if err == nil { + publishedAt := parsed.UTC() + return &publishedAt, raw + } + } + + return nil, raw +} + +func resolveNewsRSSLink(baseURL *url.URL, value string) string { + trimmed := strings.TrimSpace(value) + if trimmed == "" { + return "" + } + + parsed, err := url.Parse(trimmed) + if err != nil { + return trimmed + } + if parsed.IsAbs() || baseURL == nil { + return parsed.String() + } + + return baseURL.ResolveReference(parsed).String() +} + +func (fetcher *NewsRSSFetcher) searchURL(query string) string { + values := url.Values{} + values.Set("q", strings.TrimSpace(query)) + + return fetcher.baseURL.ResolveReference(&url.URL{ + Path: "rss/search", + RawQuery: values.Encode(), + }).String() +} + +func (fetcher *NewsRSSFetcher) doRequest(ctx context.Context, query string) (*http.Response, error) { + requestURL := fetcher.searchURL(query) + + request, err := http.NewRequestWithContext(ctx, http.MethodGet, requestURL, nil) + if err != nil { + return nil, fmt.Errorf("build news RSS request: %w", err) + } + + request.Header.Set("Accept", "application/rss+xml, application/xml;q=0.9, text/xml;q=0.8") + request.Header.Set("User-Agent", fetcher.userAgent) + + response, err := fetcher.httpClient.Do(request) + if err != nil { + return nil, fmt.Errorf("perform news RSS request %s: %w", requestURL, err) + } + + return response, nil +} + +func (fetcher *NewsRSSFetcher) responseError(response *http.Response) error { + body, _ := io.ReadAll(io.LimitReader(response.Body, 4096)) + + message := strings.TrimSpace(string(body)) + if message == "" { + message = http.StatusText(response.StatusCode) + } + if message == "" { + message = "unknown news RSS error" + } + + requestMethod := "" + requestPath := "" + if response.Request != nil { + requestMethod = response.Request.Method + if response.Request.URL != nil { + requestPath = response.Request.URL.Path + } + } + + return fmt.Errorf( + "Google News RSS %s %s returned %d: %s", + requestMethod, + requestPath, + response.StatusCode, + message, + ) +} + +func normalizeNewsRSSBaseURL(value string) string { + trimmed := strings.TrimSpace(value) + if trimmed == "" { + trimmed = defaultNewsRSSBaseURL + } + if !strings.HasSuffix(trimmed, "/") { + trimmed += "/" + } + + return trimmed +} + +func mustParseNewsRSSBaseURL(value string) *url.URL { + parsed, err := url.Parse(value) + if err != nil { + panic(fmt.Errorf("parse news RSS base URL %q: %w", value, err)) + } + + return parsed +} diff --git a/internal/fetch/news_test.go b/internal/fetch/news_test.go new file mode 100644 index 0000000..ad7d5ca --- /dev/null +++ b/internal/fetch/news_test.go @@ -0,0 +1,642 @@ +package fetch + +import ( + "context" + "database/sql" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/CrisSTEM/signalscope/internal/config" + "github.com/CrisSTEM/signalscope/internal/storage" +) + +type newsContentItemRow struct { + BindingID string + ItemType string + Title string + Summary string + URL string + ExternalID string + PublishedAt sql.NullString + DiscoveredAt string + MetadataJSON string +} + +type newsMetricSnapshotRow struct { + MetricKey string + MetricValueNum sql.NullFloat64 + Unit string + WindowKey string + CapturedAt string +} + +type newsRSSTestResponse struct { + Status int + Body string +} + +type newsRSSTestServerOptions struct { + WantQuery string + Responses []newsRSSTestResponse +} + +func TestNewsRSSFetcherPersistsNormalizedItemsAndArticleCountSnapshot(t *testing.T) { + t.Parallel() + + ctx := context.Background() + capturedAt := time.Date(2026, 4, 21, 10, 0, 0, 0, time.UTC) + + server := newNewsRSSTestServer(t, newsRSSTestServerOptions{ + WantQuery: "Example Org", + Responses: []newsRSSTestResponse{ + { + Status: http.StatusOK, + Body: ` + + + Example Org + + Example Org ships something + https://example.com/article-1?ref=rss + First summary + guid-1 + Tue, 21 Apr 2026 09:30:00 GMT + Example Publisher + + + Example Org ships something updated + https://example.com/article-1?ref=rss + Duplicate summary should be ignored + guid-1 + Tue, 21 Apr 2026 09:30:00 GMT + Example Publisher + + + Example Org opens new office + https://example.com/article-2 + Second summary + Tue, 21 Apr 2026 09:45:00 GMT + + + Example Org ignored because of max items + https://example.com/article-3 + Third summary + Tue, 21 Apr 2026 09:50:00 GMT + + +`, + }, + }, + }) + defer server.Close() + + pack := runtimeFixturePack("news-fetch-success") + setNewsBindingMaxItems(t, &pack, 2) + + db := openRuntimeTestDB(t, pack) + defer db.Close() + + registry := NewRegistry() + registry.MustRegister(SourceKindNewsRSS, newNewsRSSTestFetcher(server)) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: sequenceClock( + capturedAt, + capturedAt.Add(time.Second), + ), + } + + summary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindNewsRSS, + BindingID: "news-binding-1", + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + + if summary.Attempted != 1 { + t.Fatalf("summary.Attempted = %d, want %d", summary.Attempted, 1) + } + if summary.Succeeded != 1 { + t.Fatalf("summary.Succeeded = %d, want %d", summary.Succeeded, 1) + } + if summary.Failed != 0 { + t.Fatalf("summary.Failed = %d, want %d", summary.Failed, 0) + } + if summary.RecordsWritten != 2 { + t.Fatalf("summary.RecordsWritten = %d, want %d", summary.RecordsWritten, 2) + } + if summary.MetricsWritten != 1 { + t.Fatalf("summary.MetricsWritten = %d, want %d", summary.MetricsWritten, 1) + } + if summary.ContentItemsWritten != 2 { + t.Fatalf("summary.ContentItemsWritten = %d, want %d", summary.ContentItemsWritten, 2) + } + if len(summary.Results) != 1 { + t.Fatalf("len(summary.Results) = %d, want %d", len(summary.Results), 1) + } + if summary.Results[0].Status != storage.FetchRunStatusSucceeded { + t.Fatalf("summary.Results[0].Status = %q, want %q", summary.Results[0].Status, storage.FetchRunStatusSucceeded) + } + + contentRows, err := readNewsContentItemsForBinding(ctx, db, "news-binding-1") + if err != nil { + t.Fatalf("readNewsContentItemsForBinding() error = %v", err) + } + if len(contentRows) != 2 { + t.Fatalf("len(contentRows) = %d, want %d", len(contentRows), 2) + } + + if contentRows[0].BindingID != "news-binding-1" { + t.Fatalf("contentRows[0].BindingID = %q, want %q", contentRows[0].BindingID, "news-binding-1") + } + if contentRows[0].ItemType != ContentItemTypeNewsArticle { + t.Fatalf("contentRows[0].ItemType = %q, want %q", contentRows[0].ItemType, ContentItemTypeNewsArticle) + } + if contentRows[0].Title != "Example Org ships something" { + t.Fatalf("contentRows[0].Title = %q, want %q", contentRows[0].Title, "Example Org ships something") + } + if contentRows[0].Summary != "First summary" { + t.Fatalf("contentRows[0].Summary = %q, want %q", contentRows[0].Summary, "First summary") + } + if contentRows[0].URL != "https://example.com/article-1?ref=rss" { + t.Fatalf("contentRows[0].URL = %q, want %q", contentRows[0].URL, "https://example.com/article-1?ref=rss") + } + if contentRows[0].ExternalID != "guid-1" { + t.Fatalf("contentRows[0].ExternalID = %q, want %q", contentRows[0].ExternalID, "guid-1") + } + if !contentRows[0].PublishedAt.Valid || contentRows[0].PublishedAt.String != "2026-04-21T09:30:00Z" { + t.Fatalf("contentRows[0].PublishedAt = %+v, want %q", contentRows[0].PublishedAt, "2026-04-21T09:30:00Z") + } + if contentRows[0].DiscoveredAt != capturedAt.Format(time.RFC3339) { + t.Fatalf("contentRows[0].DiscoveredAt = %q, want %q", contentRows[0].DiscoveredAt, capturedAt.Format(time.RFC3339)) + } + if !strings.Contains(contentRows[0].MetadataJSON, `"provider":"`+newsProviderGoogleNewsRSS+`"`) { + t.Fatalf("contentRows[0].MetadataJSON = %q, want provider marker", contentRows[0].MetadataJSON) + } + if !strings.Contains(contentRows[0].MetadataJSON, `"query":"Example Org"`) { + t.Fatalf("contentRows[0].MetadataJSON = %q, want query marker", contentRows[0].MetadataJSON) + } + if !strings.Contains(contentRows[0].MetadataJSON, `"guid":"guid-1"`) { + t.Fatalf("contentRows[0].MetadataJSON = %q, want guid marker", contentRows[0].MetadataJSON) + } + + if contentRows[1].Title != "Example Org opens new office" { + t.Fatalf("contentRows[1].Title = %q, want %q", contentRows[1].Title, "Example Org opens new office") + } + if contentRows[1].Summary != "Second summary" { + t.Fatalf("contentRows[1].Summary = %q, want %q", contentRows[1].Summary, "Second summary") + } + + for _, row := range contentRows { + if row.Title == "Example Org ignored because of max items" { + t.Fatal("unexpected third article persisted despite max_items limit") + } + } + + snapshotRows, err := readNewsMetricSnapshotsForBinding(ctx, db, "news-binding-1") + if err != nil { + t.Fatalf("readNewsMetricSnapshotsForBinding() error = %v", err) + } + if len(snapshotRows) != 1 { + t.Fatalf("len(snapshotRows) = %d, want %d", len(snapshotRows), 1) + } + if snapshotRows[0].MetricKey != MetricKeyNewsArticleCount { + t.Fatalf("snapshotRows[0].MetricKey = %q, want %q", snapshotRows[0].MetricKey, MetricKeyNewsArticleCount) + } + if !snapshotRows[0].MetricValueNum.Valid || int(snapshotRows[0].MetricValueNum.Float64) != 2 { + t.Fatalf("snapshotRows[0].MetricValueNum = %+v, want %d", snapshotRows[0].MetricValueNum, 2) + } + if snapshotRows[0].Unit != "count" { + t.Fatalf("snapshotRows[0].Unit = %q, want %q", snapshotRows[0].Unit, "count") + } + if snapshotRows[0].WindowKey != "point_in_time" { + t.Fatalf("snapshotRows[0].WindowKey = %q, want %q", snapshotRows[0].WindowKey, "point_in_time") + } + if snapshotRows[0].CapturedAt != capturedAt.Format(time.RFC3339) { + t.Fatalf("snapshotRows[0].CapturedAt = %q, want %q", snapshotRows[0].CapturedAt, capturedAt.Format(time.RFC3339)) + } +} + +func TestNewsRSSFetcherRerunDedupesExistingContentItemsAndPreservesFirstWrite(t *testing.T) { + t.Parallel() + + ctx := context.Background() + firstCapturedAt := time.Date(2026, 4, 21, 11, 0, 0, 0, time.UTC) + secondCapturedAt := time.Date(2026, 4, 21, 13, 0, 0, 0, time.UTC) + + server := newNewsRSSTestServer(t, newsRSSTestServerOptions{ + WantQuery: "Example Org", + Responses: []newsRSSTestResponse{ + { + Status: http.StatusOK, + Body: ` + + + + Example Org launches satellite + https://example.com/article-1 + Original summary + guid-1 + Tue, 21 Apr 2026 10:30:00 GMT + + + Example Org hires new CTO + https://example.com/article-2 + Original second summary + Tue, 21 Apr 2026 10:45:00 GMT + + +`, + }, + { + Status: http.StatusOK, + Body: ` + + + + Example Org launches satellite UPDATED + https://example.com/article-1 + Updated summary should not overwrite the first row + guid-1 + Tue, 21 Apr 2026 10:30:00 GMT + + + Example Org hires new CTO + https://example.com/article-2 + Updated second summary should not overwrite the first row + Tue, 21 Apr 2026 10:45:00 GMT + + +`, + }, + }, + }) + defer server.Close() + + pack := runtimeFixturePack("news-fetch-rerun") + db := openRuntimeTestDB(t, pack) + defer db.Close() + + registry := NewRegistry() + registry.MustRegister(SourceKindNewsRSS, newNewsRSSTestFetcher(server)) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: sequenceClock( + firstCapturedAt, + firstCapturedAt.Add(time.Second), + secondCapturedAt, + secondCapturedAt.Add(time.Second), + ), + } + + firstSummary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindNewsRSS, + BindingID: "news-binding-1", + }) + if err != nil { + t.Fatalf("Execute(first run) error = %v", err) + } + if firstSummary.ContentItemsWritten != 2 { + t.Fatalf("firstSummary.ContentItemsWritten = %d, want %d", firstSummary.ContentItemsWritten, 2) + } + if firstSummary.MetricsWritten != 1 { + t.Fatalf("firstSummary.MetricsWritten = %d, want %d", firstSummary.MetricsWritten, 1) + } + if firstSummary.RecordsWritten != 2 { + t.Fatalf("firstSummary.RecordsWritten = %d, want %d", firstSummary.RecordsWritten, 2) + } + + secondSummary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindNewsRSS, + BindingID: "news-binding-1", + }) + if err != nil { + t.Fatalf("Execute(second run) error = %v", err) + } + if secondSummary.ContentItemsWritten != 0 { + t.Fatalf("secondSummary.ContentItemsWritten = %d, want %d", secondSummary.ContentItemsWritten, 0) + } + if secondSummary.MetricsWritten != 1 { + t.Fatalf("secondSummary.MetricsWritten = %d, want %d", secondSummary.MetricsWritten, 1) + } + if secondSummary.RecordsWritten != 2 { + t.Fatalf("secondSummary.RecordsWritten = %d, want %d", secondSummary.RecordsWritten, 2) + } + + contentCount, err := queryCount(ctx, db, `SELECT COUNT(*) FROM content_items WHERE binding_id = ?`, "news-binding-1") + if err != nil { + t.Fatalf("queryCount(content_items) error = %v", err) + } + if contentCount != 2 { + t.Fatalf("contentCount = %d, want %d", contentCount, 2) + } + + contentRows, err := readNewsContentItemsForBinding(ctx, db, "news-binding-1") + if err != nil { + t.Fatalf("readNewsContentItemsForBinding() error = %v", err) + } + if len(contentRows) != 2 { + t.Fatalf("len(contentRows) = %d, want %d", len(contentRows), 2) + } + if contentRows[0].Title != "Example Org launches satellite" { + t.Fatalf("contentRows[0].Title = %q, want first-write title", contentRows[0].Title) + } + if contentRows[0].Summary != "Original summary" { + t.Fatalf("contentRows[0].Summary = %q, want first-write summary", contentRows[0].Summary) + } + if contentRows[0].DiscoveredAt != firstCapturedAt.Format(time.RFC3339) { + t.Fatalf("contentRows[0].DiscoveredAt = %q, want %q", contentRows[0].DiscoveredAt, firstCapturedAt.Format(time.RFC3339)) + } + + snapshotRows, err := readNewsMetricSnapshotsForBinding(ctx, db, "news-binding-1") + if err != nil { + t.Fatalf("readNewsMetricSnapshotsForBinding() error = %v", err) + } + if len(snapshotRows) != 2 { + t.Fatalf("len(snapshotRows) = %d, want %d", len(snapshotRows), 2) + } + if !snapshotRows[0].MetricValueNum.Valid || int(snapshotRows[0].MetricValueNum.Float64) != 2 { + t.Fatalf("snapshotRows[0].MetricValueNum = %+v, want %d", snapshotRows[0].MetricValueNum, 2) + } + if !snapshotRows[1].MetricValueNum.Valid || int(snapshotRows[1].MetricValueNum.Float64) != 2 { + t.Fatalf("snapshotRows[1].MetricValueNum = %+v, want %d", snapshotRows[1].MetricValueNum, 2) + } + if snapshotRows[0].CapturedAt != firstCapturedAt.Format(time.RFC3339) { + t.Fatalf("snapshotRows[0].CapturedAt = %q, want %q", snapshotRows[0].CapturedAt, firstCapturedAt.Format(time.RFC3339)) + } + if snapshotRows[1].CapturedAt != secondCapturedAt.Format(time.RFC3339) { + t.Fatalf("snapshotRows[1].CapturedAt = %q, want %q", snapshotRows[1].CapturedAt, secondCapturedAt.Format(time.RFC3339)) + } +} + +func TestNewsRSSFetcherFailsOnMalformedRSS(t *testing.T) { + t.Parallel() + + ctx := context.Background() + capturedAt := time.Date(2026, 4, 21, 14, 0, 0, 0, time.UTC) + + server := newNewsRSSTestServer(t, newsRSSTestServerOptions{ + WantQuery: "Example Org", + Responses: []newsRSSTestResponse{ + { + Status: http.StatusOK, + Body: ``, + }, + }, + }) + defer server.Close() + + pack := runtimeFixturePack("news-fetch-malformed") + db := openRuntimeTestDB(t, pack) + defer db.Close() + + registry := NewRegistry() + registry.MustRegister(SourceKindNewsRSS, newNewsRSSTestFetcher(server)) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: sequenceClock( + capturedAt, + capturedAt.Add(time.Second), + ), + } + + summary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindNewsRSS, + BindingID: "news-binding-1", + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + + if summary.Attempted != 1 { + t.Fatalf("summary.Attempted = %d, want %d", summary.Attempted, 1) + } + if summary.Succeeded != 0 { + t.Fatalf("summary.Succeeded = %d, want %d", summary.Succeeded, 0) + } + if summary.Failed != 1 { + t.Fatalf("summary.Failed = %d, want %d", summary.Failed, 1) + } + if len(summary.Results) != 1 { + t.Fatalf("len(summary.Results) = %d, want %d", len(summary.Results), 1) + } + if summary.Results[0].Status != storage.FetchRunStatusFailed { + t.Fatalf("summary.Results[0].Status = %q, want %q", summary.Results[0].Status, storage.FetchRunStatusFailed) + } + if !strings.Contains(summary.Results[0].ErrorMessage, "decode news RSS response") { + t.Fatalf("summary.Results[0].ErrorMessage = %q, want decode error", summary.Results[0].ErrorMessage) + } + + metricCount, err := queryCount(ctx, db, `SELECT COUNT(*) FROM metric_snapshots WHERE binding_id = ?`, "news-binding-1") + if err != nil { + t.Fatalf("queryCount(metric_snapshots) error = %v", err) + } + if metricCount != 0 { + t.Fatalf("metricCount = %d, want %d", metricCount, 0) + } + + contentCount, err := queryCount(ctx, db, `SELECT COUNT(*) FROM content_items WHERE binding_id = ?`, "news-binding-1") + if err != nil { + t.Fatalf("queryCount(content_items) error = %v", err) + } + if contentCount != 0 { + t.Fatalf("contentCount = %d, want %d", contentCount, 0) + } + + fetchRuns, err := readFetchRuns(ctx, db) + if err != nil { + t.Fatalf("readFetchRuns() error = %v", err) + } + if len(fetchRuns) != 1 { + t.Fatalf("len(fetchRuns) = %d, want %d", len(fetchRuns), 1) + } + if fetchRuns[0].BindingID != "news-binding-1" { + t.Fatalf("fetchRuns[0].BindingID = %q, want %q", fetchRuns[0].BindingID, "news-binding-1") + } + if fetchRuns[0].Status != storage.FetchRunStatusFailed { + t.Fatalf("fetchRuns[0].Status = %q, want %q", fetchRuns[0].Status, storage.FetchRunStatusFailed) + } +} + +func newNewsRSSTestFetcher(server *httptest.Server) *NewsRSSFetcher { + return NewNewsRSSFetcher(NewsRSSFetcherOptions{ + BaseURL: server.URL + "/", + HTTPClient: server.Client(), + UserAgent: "signalscope-test", + Now: func() time.Time { + return time.Date(2026, 4, 21, 0, 0, 0, 0, time.UTC) + }, + }) +} + +func newNewsRSSTestServer(t *testing.T, options newsRSSTestServerOptions) *httptest.Server { + t.Helper() + + responses := append([]newsRSSTestResponse(nil), options.Responses...) + if len(responses) == 0 { + responses = []newsRSSTestResponse{ + { + Status: http.StatusOK, + Body: ``, + }, + } + } + + wantQuery := strings.TrimSpace(options.WantQuery) + + var mu sync.Mutex + requestCount := 0 + + handler := http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/rss/search" { + http.Error(w, "unexpected path", http.StatusNotFound) + return + } + if wantQuery != "" && request.URL.Query().Get("q") != wantQuery { + http.Error(w, "unexpected query", http.StatusBadRequest) + return + } + + mu.Lock() + index := requestCount + if index >= len(responses) { + index = len(responses) - 1 + } + requestCount++ + response := responses[index] + mu.Unlock() + + status := response.Status + if status == 0 { + status = http.StatusOK + } + + w.Header().Set("Content-Type", "application/rss+xml; charset=utf-8") + w.WriteHeader(status) + _, _ = io.WriteString(w, response.Body) + }) + + return httptest.NewServer(handler) +} + +func setNewsBindingMaxItems(t *testing.T, pack *config.Pack, maxItems int) { + t.Helper() + + for index := range pack.Sources.Bindings { + if pack.Sources.Bindings[index].ID != "news-binding-1" { + continue + } + + if pack.Sources.Bindings[index].Scope == nil { + pack.Sources.Bindings[index].Scope = config.JSONMap{} + } + pack.Sources.Bindings[index].Scope["max_items"] = maxItems + return + } + + t.Fatal("news-binding-1 not found in fixture pack") +} + +func readNewsContentItemsForBinding(ctx context.Context, db *sql.DB, bindingID string) ([]newsContentItemRow, error) { + rows, err := db.QueryContext( + ctx, + `SELECT + binding_id, + item_type, + title, + summary, + url, + external_id, + published_at, + discovered_at, + metadata_json + FROM content_items + WHERE binding_id = ? + ORDER BY id ASC`, + bindingID, + ) + if err != nil { + return nil, err + } + defer rows.Close() + + var result []newsContentItemRow + for rows.Next() { + var row newsContentItemRow + if err := rows.Scan( + &row.BindingID, + &row.ItemType, + &row.Title, + &row.Summary, + &row.URL, + &row.ExternalID, + &row.PublishedAt, + &row.DiscoveredAt, + &row.MetadataJSON, + ); err != nil { + return nil, err + } + + result = append(result, row) + } + + if err := rows.Err(); err != nil { + return nil, err + } + + return result, nil +} + +func readNewsMetricSnapshotsForBinding(ctx context.Context, db *sql.DB, bindingID string) ([]newsMetricSnapshotRow, error) { + rows, err := db.QueryContext( + ctx, + `SELECT + metric_key, + metric_value_num, + unit, + window_key, + captured_at + FROM metric_snapshots + WHERE binding_id = ? + ORDER BY captured_at ASC, id ASC`, + bindingID, + ) + if err != nil { + return nil, err + } + defer rows.Close() + + var result []newsMetricSnapshotRow + for rows.Next() { + var row newsMetricSnapshotRow + if err := rows.Scan( + &row.MetricKey, + &row.MetricValueNum, + &row.Unit, + &row.WindowKey, + &row.CapturedAt, + ); err != nil { + return nil, err + } + + result = append(result, row) + } + + if err := rows.Err(); err != nil { + return nil, err + } + + return result, nil +} diff --git a/internal/fetch/runtime.go b/internal/fetch/runtime.go index 227f6b3..8ccfbef 100644 --- a/internal/fetch/runtime.go +++ b/internal/fetch/runtime.go @@ -35,6 +35,7 @@ func NewRegistry() *Registry { func DefaultRegistry() *Registry { registry := NewRegistry() registry.MustRegister(SourceKindGitHub, NewGitHubFetcher(GitHubFetcherOptions{})) + registry.MustRegister(SourceKindNewsRSS, NewNewsRSSFetcher(NewsRSSFetcherOptions{})) return registry } diff --git a/internal/fetch/runtime_test.go b/internal/fetch/runtime_test.go index bca4ef0..db544a9 100644 --- a/internal/fetch/runtime_test.go +++ b/internal/fetch/runtime_test.go @@ -187,7 +187,7 @@ func TestRuntimeExecuteMarksBindingsFailedWhenNoFetcherIsRegistered(t *testing.T t.Parallel() ctx := context.Background() - pack := runtimeFixturePack("runtime-fetch-no-fetcher") + pack := runtimeFixturePackWithChangelog("runtime-fetch-no-fetcher") db := openRuntimeTestDB(t, pack) defer db.Close() @@ -200,7 +200,7 @@ func TestRuntimeExecuteMarksBindingsFailedWhenNoFetcherIsRegistered(t *testing.T ), } - summary, err := runtime.Execute(ctx, pack, Options{SourceKind: "news_rss"}) + summary, err := runtime.Execute(ctx, pack, Options{SourceKind: "changelog"}) if err != nil { t.Fatalf("Execute() error = %v", err) } @@ -219,7 +219,7 @@ func TestRuntimeExecuteMarksBindingsFailedWhenNoFetcherIsRegistered(t *testing.T if result.Status != storage.FetchRunStatusFailed { t.Fatalf("result.Status = %q, want %q", result.Status, storage.FetchRunStatusFailed) } - if !strings.Contains(result.ErrorMessage, `no fetcher registered for source kind "news_rss"`) { + if !strings.Contains(result.ErrorMessage, `no fetcher registered for source kind "changelog"`) { t.Fatalf("result.ErrorMessage = %q, want missing fetcher message", result.ErrorMessage) } } @@ -231,8 +231,8 @@ func TestRuntimeExecuteMarksBindingsFailedWhenNoFetcherIsRegistered(t *testing.T if len(rows) != 1 { t.Fatalf("len(rows) = %d, want %d", len(rows), 1) } - if rows[0].BindingID != "news-binding-1" { - t.Fatalf("rows[0].BindingID = %q, want %q", rows[0].BindingID, "news-binding-1") + if rows[0].BindingID != "changelog-binding-1" { + t.Fatalf("rows[0].BindingID = %q, want %q", rows[0].BindingID, "changelog-binding-1") } if rows[0].Status != storage.FetchRunStatusFailed { t.Fatalf("rows[0].Status = %q, want %q", rows[0].Status, storage.FetchRunStatusFailed) @@ -382,6 +382,31 @@ func runtimeFixturePack(datasetID string) config.Pack { } } +func runtimeFixturePackWithChangelog(datasetID string) config.Pack { + pack := runtimeFixturePack(datasetID) + + pack.Sources.Sources = append(pack.Sources.Sources, config.Source{ + ID: "changelog", + Kind: "changelog", + Enabled: true, + Defaults: config.JSONMap{}, + }) + + pack.Sources.Bindings = append(pack.Sources.Bindings, config.Binding{ + ID: "changelog-binding-1", + EntityID: "org-1", + SourceID: "changelog", + Enabled: true, + Scope: config.JSONMap{ + "mode": "rss", + "feed_url": "https://example.com/changelog.xml", + }, + Notes: "changelog binding 1", + }) + + return pack +} + func readFetchRuns(ctx context.Context, db *sql.DB) ([]fetchRunRow, error) { rows, err := db.QueryContext( ctx, diff --git a/internal/storage/ingestion.go b/internal/storage/ingestion.go index 93eef2b..32a4d65 100644 --- a/internal/storage/ingestion.go +++ b/internal/storage/ingestion.go @@ -201,34 +201,74 @@ func FinishFetchRunFailure(ctx context.Context, db *sql.DB, params FinalizeFetch } func InsertMetricSnapshots(ctx context.Context, db *sql.DB, fetchRunID int64, snapshots []MetricSnapshotInput) (int, error) { + metricsInserted, _, err := InsertMetricSnapshotsAndContentItems(ctx, db, fetchRunID, snapshots, nil) + return metricsInserted, err +} + +func InsertContentItems(ctx context.Context, db *sql.DB, fetchRunID int64, items []ContentItemInput) (int, error) { + _, contentItemsInserted, err := InsertMetricSnapshotsAndContentItems(ctx, db, fetchRunID, nil, items) + return contentItemsInserted, err +} + +func InsertMetricSnapshotsAndContentItems( + ctx context.Context, + db *sql.DB, + fetchRunID int64, + snapshots []MetricSnapshotInput, + items []ContentItemInput, +) (int, int, error) { if db == nil { - return 0, fmt.Errorf("database handle is required") + return 0, 0, fmt.Errorf("database handle is required") } if fetchRunID <= 0 { - return 0, fmt.Errorf("fetch_run_id must be greater than zero") + return 0, 0, fmt.Errorf("fetch_run_id must be greater than zero") } - if len(snapshots) == 0 { - return 0, nil + if len(snapshots) == 0 && len(items) == 0 { + return 0, 0, nil } ctx = normalizeContext(ctx) - runContext, err := lookupFetchRunContext(ctx, db, fetchRunID) + runContext, err := lookupRunningFetchRunContext(ctx, db, fetchRunID) if err != nil { - return 0, err - } - if runContext.Status != FetchRunStatusRunning { - return 0, fmt.Errorf("fetch run %d is not running", fetchRunID) + return 0, 0, err } tx, err := db.BeginTx(ctx, nil) if err != nil { - return 0, fmt.Errorf("begin metric snapshot transaction: %w", err) + return 0, 0, fmt.Errorf("begin observation batch transaction: %w", err) } defer func() { _ = tx.Rollback() }() + metricsInserted, err := insertMetricSnapshotsTx(ctx, tx, runContext, snapshots) + if err != nil { + return 0, 0, err + } + + contentItemsInserted, err := insertContentItemsTx(ctx, tx, runContext, items) + if err != nil { + return 0, 0, err + } + + if err := tx.Commit(); err != nil { + return 0, 0, fmt.Errorf("commit observation batch transaction: %w", err) + } + + return metricsInserted, contentItemsInserted, nil +} + +func insertMetricSnapshotsTx( + ctx context.Context, + tx *sql.Tx, + runContext fetchRunContext, + snapshots []MetricSnapshotInput, +) (int, error) { + if len(snapshots) == 0 { + return 0, nil + } + stmt, err := tx.PrepareContext( ctx, `INSERT INTO metric_snapshots ( @@ -283,42 +323,19 @@ func InsertMetricSnapshots(ctx context.Context, db *sql.DB, fetchRunID int64, sn inserted++ } - if err := tx.Commit(); err != nil { - return 0, fmt.Errorf("commit metric snapshot transaction: %w", err) - } - return inserted, nil } -func InsertContentItems(ctx context.Context, db *sql.DB, fetchRunID int64, items []ContentItemInput) (int, error) { - if db == nil { - return 0, fmt.Errorf("database handle is required") - } - if fetchRunID <= 0 { - return 0, fmt.Errorf("fetch_run_id must be greater than zero") - } +func insertContentItemsTx( + ctx context.Context, + tx *sql.Tx, + runContext fetchRunContext, + items []ContentItemInput, +) (int, error) { if len(items) == 0 { return 0, nil } - ctx = normalizeContext(ctx) - - runContext, err := lookupFetchRunContext(ctx, db, fetchRunID) - if err != nil { - return 0, err - } - if runContext.Status != FetchRunStatusRunning { - return 0, fmt.Errorf("fetch run %d is not running", fetchRunID) - } - - tx, err := db.BeginTx(ctx, nil) - if err != nil { - return 0, fmt.Errorf("begin content item transaction: %w", err) - } - defer func() { - _ = tx.Rollback() - }() - stmt, err := tx.PrepareContext( ctx, `INSERT OR IGNORE INTO content_items ( @@ -386,10 +403,6 @@ func InsertContentItems(ctx context.Context, db *sql.DB, fetchRunID int64, items inserted += int(rowsAffected) } - if err := tx.Commit(); err != nil { - return 0, fmt.Errorf("commit content item transaction: %w", err) - } - return inserted, nil } @@ -521,6 +534,18 @@ func lookupFetchRunContext(ctx context.Context, db *sql.DB, fetchRunID int64) (f return context, nil } +func lookupRunningFetchRunContext(ctx context.Context, db *sql.DB, fetchRunID int64) (fetchRunContext, error) { + runContext, err := lookupFetchRunContext(ctx, db, fetchRunID) + if err != nil { + return fetchRunContext{}, err + } + if runContext.Status != FetchRunStatusRunning { + return fetchRunContext{}, fmt.Errorf("fetch run %d is not running", fetchRunID) + } + + return runContext, nil +} + func validateWriteCounts(recordsWritten, metricsWritten, contentItemsWritten int) error { switch { case recordsWritten < 0: diff --git a/internal/storage/ingestion_test.go b/internal/storage/ingestion_test.go index 3188e37..9b51441 100644 --- a/internal/storage/ingestion_test.go +++ b/internal/storage/ingestion_test.go @@ -450,6 +450,68 @@ func TestInsertContentItemsDeduplicatesWithinRunAndAcrossReruns(t *testing.T) { } } +func TestInsertMetricSnapshotsAndContentItemsRollsBackOnContentError(t *testing.T) { + t.Parallel() + + ctx := context.Background() + db, datasetID := openSeededIngestionDB(t) + defer db.Close() + + startedAt := time.Date(2026, 4, 14, 15, 0, 0, 0, time.UTC) + run, err := StartFetchRun(ctx, db, StartFetchRunParams{ + DatasetID: datasetID, + BindingID: "news-binding-1", + StartedAt: startedAt, + }) + if err != nil { + t.Fatalf("StartFetchRun() error = %v", err) + } + + value := 2.0 + _, _, err = InsertMetricSnapshotsAndContentItems( + ctx, + db, + run.ID, + []MetricSnapshotInput{ + { + MetricKey: "news.article_count", + MetricValueNum: &value, + Unit: "count", + WindowKey: "point_in_time", + CapturedAt: startedAt.Add(time.Second), + }, + }, + []ContentItemInput{ + { + ItemType: "news_article", + DiscoveredAt: startedAt.Add(time.Second), + }, + }, + ) + if err == nil { + t.Fatal("InsertMetricSnapshotsAndContentItems() error = nil, want error") + } + if !strings.Contains(err.Error(), "cannot derive dedupe_key") { + t.Fatalf("InsertMetricSnapshotsAndContentItems() error = %q, want dedupe_key error", err) + } + + metricCount, err := queryCount(ctx, db, `SELECT COUNT(*) FROM metric_snapshots WHERE fetch_run_id = ?`, run.ID) + if err != nil { + t.Fatalf("queryCount(metric_snapshots) error = %v", err) + } + if metricCount != 0 { + t.Fatalf("metric snapshot count = %d, want %d", metricCount, 0) + } + + contentCount, err := queryCount(ctx, db, `SELECT COUNT(*) FROM content_items WHERE fetch_run_id = ?`, run.ID) + if err != nil { + t.Fatalf("queryCount(content_items) error = %v", err) + } + if contentCount != 0 { + t.Fatalf("content item count = %d, want %d", contentCount, 0) + } +} + func TestSeedPackClearsIngestionRowsOnReseed(t *testing.T) { t.Parallel()