diff --git a/PROJECTS/intermediate/security-news-scraper/.gitignore b/PROJECTS/intermediate/security-news-scraper/.gitignore index 38335067..d938e12c 100644 --- a/PROJECTS/intermediate/security-news-scraper/.gitignore +++ b/PROJECTS/intermediate/security-news-scraper/.gitignore @@ -18,6 +18,9 @@ docs/ *.sqlite /data/ +# large local KEV catalog snapshot (manual download; tests use kev-sample.json) +/testdata/kev/kev-full.json + # env / secrets / local overrides .env *.local.yaml diff --git a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/enrich.go b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/enrich.go index 7b041561..27568097 100644 --- a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/enrich.go +++ b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/enrich.go @@ -4,11 +4,9 @@ package main import ( - "context" "fmt" "net/http" "os" - "os/signal" "time" "github.com/spf13/cobra" @@ -43,10 +41,7 @@ func runEnrich(cmd *cobra.Command, args []string) error { } defer st.Close() - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) - defer stop() - - stats, err := enrich.Run(ctx, st, buildEnrichClients(cfg), time.Now(), cfg.Enrich.CacheTTLHours, cfg.Enrich.NegativeTTLHours) + stats, err := enrich.Run(cmd.Context(), st, buildEnrichClients(cfg), time.Now(), cfg.Enrich.CacheTTLHours, cfg.Enrich.NegativeTTLHours) if err != nil { return err } diff --git a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/main.go b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/main.go index 3c1d5bcc..63e5bdcc 100644 --- a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/main.go +++ b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/main.go @@ -7,10 +7,11 @@ import ( "context" "os" "os/signal" + "syscall" ) func main() { - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() if err := rootCmd.ExecuteContext(ctx); err != nil { os.Exit(1) diff --git a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/pipeline.go b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/pipeline.go new file mode 100644 index 00000000..bd8bdea3 --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/pipeline.go @@ -0,0 +1,30 @@ +// ©AngelaMos | 2026 +// pipeline.go + +package main + +import ( + "context" + "time" + + "github.com/CarterPerez-dev/nadezhda/internal/cluster" + "github.com/CarterPerez-dev/nadezhda/internal/config" + "github.com/CarterPerez-dev/nadezhda/internal/fetch" + "github.com/CarterPerez-dev/nadezhda/internal/ingest" + "github.com/CarterPerez-dev/nadezhda/internal/source" + "github.com/CarterPerez-dev/nadezhda/internal/store" +) + +func ingestAndCluster(ctx context.Context, fc *fetch.Client, st *store.Store, cfg config.Config, targets []source.Source, start time.Time) (ingest.Summary, cluster.Stats, error) { + summary, err := ingest.Run(ctx, fc, st, cfg, targets, start) + if err != nil { + return ingest.Summary{}, cluster.Stats{}, err + } + sinceUnix := start.Unix() - int64(cfg.Cluster.LookbackHours)*secondsPerHour + windowSeconds := int64(cfg.Cluster.WindowHours) * secondsPerHour + stats, err := cluster.Rebuild(st, cfg.Cluster.TitleJaccard, windowSeconds, sinceUnix) + if err != nil { + return summary, cluster.Stats{}, err + } + return summary, stats, nil +} diff --git a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/scrape.go b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/scrape.go index fbb1c0f8..e89da4e0 100644 --- a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/scrape.go +++ b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/scrape.go @@ -6,13 +6,10 @@ package main import ( "context" "fmt" - "os" - "os/signal" "time" "github.com/spf13/cobra" - "github.com/CarterPerez-dev/nadezhda/internal/cluster" "github.com/CarterPerez-dev/nadezhda/internal/config" "github.com/CarterPerez-dev/nadezhda/internal/enrich" "github.com/CarterPerez-dev/nadezhda/internal/fetch" @@ -78,24 +75,15 @@ func runScrape(cmd *cobra.Command, args []string) error { MaxRetries: cfg.Fetch.MaxRetries, }) - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) - defer stop() - + ctx := cmd.Context() now := time.Now() - summary, err := ingest.Run(ctx, fc, st, cfg, targets, now) + summary, cstats, err := ingestAndCluster(ctx, fc, st, cfg, targets, now) if err != nil { return err } printSummary(cmd, summary) - - sinceUnix := now.Unix() - int64(cfg.Cluster.LookbackHours)*secondsPerHour - windowSeconds := int64(cfg.Cluster.WindowHours) * secondsPerHour - stats, err := cluster.Rebuild(st, cfg.Cluster.TitleJaccard, windowSeconds, sinceUnix) - if err != nil { - return err - } fmt.Fprintf(cmd.OutOrStdout(), "%d clusters (%d multi-source, largest %d)\n", - stats.Total, stats.MultiSource, stats.LargestSize) + cstats.Total, cstats.MultiSource, cstats.LargestSize) if !scrapeNoEnrich { enrichAfterScrape(ctx, cmd, cfg, st) diff --git a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/stubs.go b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/stubs.go deleted file mode 100644 index 9a0cee9e..00000000 --- a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/stubs.go +++ /dev/null @@ -1,33 +0,0 @@ -// ©AngelaMos | 2026 -// stubs.go - -package main - -import ( - "fmt" - - "github.com/spf13/cobra" -) - -func notImplemented(milestone string) func(*cobra.Command, []string) error { - return func(cmd *cobra.Command, args []string) error { - return fmt.Errorf("%s is not implemented yet (%s)", cmd.Name(), milestone) - } -} - -func init() { - stubs := []struct { - use string - short string - milestone string - }{ - {"watch", "Run as a daemon, re-ingesting on an interval", "milestone M7"}, - } - for _, s := range stubs { - rootCmd.AddCommand(&cobra.Command{ - Use: s.use, - Short: s.short, - RunE: notImplemented(s.milestone), - }) - } -} diff --git a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/watch.go b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/watch.go new file mode 100644 index 00000000..52f63d60 --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/watch.go @@ -0,0 +1,258 @@ +// ©AngelaMos | 2026 +// watch.go + +package main + +import ( + "context" + "fmt" + "io" + "net/http" + "net/url" + "time" + + "github.com/spf13/cobra" + + "github.com/CarterPerez-dev/nadezhda/internal/config" + "github.com/CarterPerez-dev/nadezhda/internal/enrich" + "github.com/CarterPerez-dev/nadezhda/internal/fetch" + "github.com/CarterPerez-dev/nadezhda/internal/rank" + "github.com/CarterPerez-dev/nadezhda/internal/source" + "github.com/CarterPerez-dev/nadezhda/internal/store" + "github.com/CarterPerez-dev/nadezhda/internal/watch" +) + +const untitledCluster = "(untitled cluster)" + +var ( + watchInterval string + watchOnce bool + watchNoEnrich bool +) + +var watchCmd = &cobra.Command{ + Use: "watch", + Short: "Run as a daemon, re-ingesting on an interval with an optional webhook notify", + Long: "Run nadezhda as a long-lived daemon that re-ingests every enabled source on an interval " + + "(default from config, override with --interval). Each cycle scrapes, clusters, and enriches " + + "exactly like the scrape command. When watch.webhook_url is set, genuinely new high-signal " + + "stories are POSTed to that webhook (Slack, Discord, or any JSON endpoint).", + RunE: runWatch, +} + +func init() { + watchCmd.Flags().StringVar(&watchInterval, "interval", "", "re-ingest interval, e.g. 30m or 1h (overrides watch.interval in config)") + watchCmd.Flags().BoolVar(&watchOnce, "once", false, "run a single cycle and exit instead of looping") + watchCmd.Flags().BoolVar(&watchNoEnrich, "no-enrich", false, "skip the keyless CVE enrichment pass each cycle") + rootCmd.AddCommand(watchCmd) +} + +func runWatch(cmd *cobra.Command, args []string) error { + cfg, err := loadConfig() + if err != nil { + return err + } + interval, err := resolveWatchInterval(cfg) + if err != nil { + return err + } + if cfg.Watch.WebhookURL != "" { + if err := validateWebhookURL(cfg.Watch.WebhookURL); err != nil { + return err + } + } + + srcs, err := source.Load(cfg.SourcesPath) + if err != nil { + return err + } + targets := source.Enabled(srcs) + if len(targets) == 0 { + return fmt.Errorf("watch: no enabled sources to poll") + } + + st, err := store.Open(cfg.DBPath) + if err != nil { + return err + } + defer st.Close() + + fc := fetch.New(fetch.Options{ + UserAgent: cfg.Fetch.UserAgent, + PerHostRate: cfg.Fetch.PerHostRate, + PerHostBurst: cfg.Fetch.PerHostBurst, + Timeout: time.Duration(cfg.Fetch.TimeoutSeconds) * time.Second, + MaxRetries: cfg.Fetch.MaxRetries, + }) + + out := cmd.ErrOrStderr() + if watchNoEnrich && cfg.Watch.NotifyOnKEV { + fmt.Fprintln(out, "watch: --no-enrich with notify_on_kev set: KEV status is never computed, so KEV alerts will not fire") + } + + cycle := func(ctx context.Context) (watch.Report, error) { + start := time.Now() + summary, cstats, err := ingestAndCluster(ctx, fc, st, cfg, targets, start) + if err != nil { + return watch.Report{}, err + } + newArticles, duplicates, failed := summary.Totals() + + var enriched, kevHits int + if !watchNoEnrich { + enriched, kevHits = watchEnrich(ctx, out, st, cfg) + } + + notable, err := buildNotable(st, cfg, start) + if err != nil { + return watch.Report{}, err + } + return watch.Report{ + Start: start, + Duration: time.Since(start), + NewArticles: newArticles, + Duplicates: duplicates, + Clusters: cstats.Total, + Enriched: enriched, + KEVHits: kevHits, + Failed: failed, + Notable: notable, + }, nil + } + + opts := watch.Options{ + Interval: interval, + RunAtStart: true, + Cycle: cycle, + Out: out, + } + if cfg.Watch.WebhookURL != "" { + opts.Notifier = watch.WebhookNotifier{ + URL: cfg.Watch.WebhookURL, + Client: &http.Client{Timeout: time.Duration(cfg.Fetch.TimeoutSeconds) * time.Second}, + } + } + + if watchOnce { + return watch.Once(cmd.Context(), opts) + } + return watch.Run(cmd.Context(), opts) +} + +func resolveWatchInterval(cfg config.Config) (time.Duration, error) { + raw := cfg.Watch.Interval + if watchInterval != "" { + raw = watchInterval + } + d, err := time.ParseDuration(raw) + if err != nil { + return 0, fmt.Errorf("watch: invalid interval %q: %w", raw, err) + } + if d < config.MinWatchInterval { + return 0, fmt.Errorf("watch: interval %s is below the minimum %s", d, config.MinWatchInterval) + } + return d, nil +} + +func validateWebhookURL(raw string) error { + u, err := url.Parse(raw) + if err != nil { + return fmt.Errorf("watch: invalid webhook_url %q: %w", raw, err) + } + if u.Scheme != "http" && u.Scheme != "https" { + return fmt.Errorf("watch: webhook_url must be http or https, got %q", raw) + } + if u.Host == "" { + return fmt.Errorf("watch: webhook_url must include a host, got %q", raw) + } + return nil +} + +func watchEnrich(ctx context.Context, out io.Writer, st *store.Store, cfg config.Config) (enriched, kevHits int) { + ectx, cancel := context.WithTimeout(ctx, enrichBudget) + defer cancel() + stats, err := enrich.Run(ectx, st, buildEnrichClients(cfg), time.Now(), cfg.Enrich.CacheTTLHours, cfg.Enrich.NegativeTTLHours) + if err != nil && ctx.Err() == nil { + fmt.Fprintf(out, "watch: enrich degraded, KEV/CVE data may be stale: %v\n", err) + } + return stats.Enriched, stats.KEVHits +} + +func buildNotable(st *store.Store, cfg config.Config, cycleStart time.Time) ([]watch.NotableItem, error) { + fresh, err := st.NewlyFetchedClusters(cycleStart.Unix()) + if err != nil { + return nil, err + } + scored := rank.Rank(fresh, cfg.Rank, cfg.Watchlist, time.Now()) + items := make([]watch.NotableItem, 0, cfg.Watch.NotifyMaxItems) + for _, sc := range scored { + if len(items) >= cfg.Watch.NotifyMaxItems { + break + } + if isNotable(sc, cfg.Watch) { + items = append(items, toNotable(sc)) + } + } + return items, nil +} + +func isNotable(sc rank.Scored, w config.Watch) bool { + if sc.Score >= w.NotifyMinScore { + return true + } + if w.NotifyOnKEV { + for _, v := range sc.Cluster.CVEs { + if v.IsKEV { + return true + } + } + } + return false +} + +func toNotable(sc rank.Scored) watch.NotableItem { + c := sc.Cluster + title, link := representativeArticle(c.Articles) + var maxCVSS float64 + var isKEV bool + cves := make([]string, 0, len(c.CVEs)) + for _, v := range c.CVEs { + cves = append(cves, v.ID) + if v.CVSSScore != nil && *v.CVSSScore > maxCVSS { + maxCVSS = *v.CVSSScore + } + if v.IsKEV { + isKEV = true + } + } + return watch.NotableItem{ + Title: title, + URL: link, + Score: sc.Score, + MaxCVSS: maxCVSS, + IsKEV: isKEV, + CVEs: cves, + Sources: distinctSources(c.Articles), + } +} + +func representativeArticle(articles []store.DigestArticle) (title, link string) { + if len(articles) == 0 { + return untitledCluster, "" + } + best := articles[0] + for _, a := range articles[1:] { + if a.SourceWeight > best.SourceWeight { + best = a + } + } + return best.Title, best.CanonicalURL +} + +func distinctSources(articles []store.DigestArticle) int { + set := make(map[string]struct{}, len(articles)) + for _, a := range articles { + set[a.SourceName] = struct{}{} + } + return len(set) +} diff --git a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/watch_test.go b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/watch_test.go new file mode 100644 index 00000000..cfd88fa5 --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/watch_test.go @@ -0,0 +1,149 @@ +// ©AngelaMos | 2026 +// watch_test.go + +package main + +import ( + "fmt" + "path/filepath" + "strconv" + "testing" + "time" + + "github.com/CarterPerez-dev/nadezhda/internal/config" + "github.com/CarterPerez-dev/nadezhda/internal/rank" + "github.com/CarterPerez-dev/nadezhda/internal/store" +) + +func TestIsNotable(t *testing.T) { + kevCluster := store.DigestCluster{CVEs: []store.DigestCVE{{ID: "CVE-2026-1", IsKEV: true}}} + plainCluster := store.DigestCluster{} + tests := []struct { + name string + scored rank.Scored + watch config.Watch + want bool + }{ + {"score above threshold", rank.Scored{Cluster: plainCluster, Score: 0.9}, config.Watch{NotifyMinScore: 0.5}, true}, + {"score below threshold, not kev", rank.Scored{Cluster: plainCluster, Score: 0.2}, config.Watch{NotifyMinScore: 0.5}, false}, + {"below threshold but kev with notify_on_kev", rank.Scored{Cluster: kevCluster, Score: 0.1}, config.Watch{NotifyMinScore: 0.5, NotifyOnKEV: true}, true}, + {"kev but notify_on_kev off", rank.Scored{Cluster: kevCluster, Score: 0.1}, config.Watch{NotifyMinScore: 0.5, NotifyOnKEV: false}, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isNotable(tt.scored, tt.watch); got != tt.want { + t.Errorf("isNotable = %v, want %v", got, tt.want) + } + }) + } +} + +func TestToNotablePicksHighestAuthoritySource(t *testing.T) { + cvss := 9.8 + sc := rank.Scored{ + Score: 0.77, + Cluster: store.DigestCluster{ + Articles: []store.DigestArticle{ + {Title: "low authority", CanonicalURL: "https://low", SourceName: "low", SourceWeight: 0.5}, + {Title: "high authority", CanonicalURL: "https://high", SourceName: "high", SourceWeight: 1.0}, + }, + CVEs: []store.DigestCVE{{ID: "CVE-2026-1", CVSSScore: &cvss, IsKEV: true}}, + }, + } + n := toNotable(sc) + if n.Title != "high authority" || n.URL != "https://high" { + t.Errorf("representative should be the highest-source-weight article, got %q / %q", n.Title, n.URL) + } + if !n.IsKEV || n.MaxCVSS != 9.8 { + t.Errorf("cve signals wrong: kev=%v cvss=%v", n.IsKEV, n.MaxCVSS) + } + if n.Sources != 2 { + t.Errorf("distinct sources = %d, want 2", n.Sources) + } + if len(n.CVEs) != 1 || n.CVEs[0] != "CVE-2026-1" { + t.Errorf("cves wrong: %v", n.CVEs) + } + if n.Score != 0.77 { + t.Errorf("score = %v, want 0.77", n.Score) + } +} + +func TestResolveWatchInterval(t *testing.T) { + cfg := config.Default() + watchInterval = "" + defer func() { watchInterval = "" }() + + if d, err := resolveWatchInterval(cfg); err != nil || d != time.Hour { + t.Errorf("default interval = %v, %v; want 1h, nil", d, err) + } + + watchInterval = "30m" + if d, err := resolveWatchInterval(cfg); err != nil || d != 30*time.Minute { + t.Errorf("flag override = %v, %v; want 30m, nil", d, err) + } + + watchInterval = "5s" + if _, err := resolveWatchInterval(cfg); err == nil { + t.Error("expected an error for a sub-minimum interval") + } + + watchInterval = "nonsense" + if _, err := resolveWatchInterval(cfg); err == nil { + t.Error("expected an error for a garbage interval") + } +} + +func TestValidateWebhookURL(t *testing.T) { + good := []string{"https://hooks.slack.com/services/x", "http://127.0.0.1:39871/hook"} + for _, u := range good { + if err := validateWebhookURL(u); err != nil { + t.Errorf("validateWebhookURL(%q) = %v, want nil", u, err) + } + } + bad := []string{"example.com/hook", "ftp://x/y", "not a url", ""} + for _, u := range bad { + if err := validateWebhookURL(u); err == nil { + t.Errorf("validateWebhookURL(%q) = nil, want error", u) + } + } +} + +func TestBuildNotableCapsToNotifyMax(t *testing.T) { + st, err := store.Open(filepath.Join(t.TempDir(), "watch.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + + src, err := st.UpsertSource(store.SourceInput{Name: "s", URL: "https://s/f", Type: "rss", Weight: 1, Enabled: true}) + if err != nil { + t.Fatal(err) + } + var rows []store.ClusterRow + for i := 0; i < 5; i++ { + id, err := st.InsertArticle(store.Article{ + SourceID: src, CanonicalURL: fmt.Sprintf("https://s/%d", i), + ContentHash: "c" + strconv.Itoa(i), TitleHash: "t" + strconv.Itoa(i), + Title: "story " + strconv.Itoa(i), PublishedAt: 10000, FetchedAt: 10000, + }) + if err != nil { + t.Fatal(err) + } + rows = append(rows, store.ClusterRow{Key: strconv.Itoa(i), Members: []int64{id}, FirstSeen: 10000, LastSeen: 10000}) + } + if err := st.ReplaceClusters(rows); err != nil { + t.Fatal(err) + } + + cfg := config.Default() + cfg.Watch.NotifyMinScore = 0 + cfg.Watch.NotifyMaxItems = 2 + + items, err := buildNotable(st, cfg, time.Unix(9999, 0)) + if err != nil { + t.Fatal(err) + } + if len(items) != 2 { + t.Fatalf("buildNotable returned %d items, want the notify_max_items cap of 2", len(items)) + } +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/config/config.go b/PROJECTS/intermediate/security-news-scraper/internal/config/config.go index b26b0f96..b773d46a 100644 --- a/PROJECTS/intermediate/security-news-scraper/internal/config/config.go +++ b/PROJECTS/intermediate/security-news-scraper/internal/config/config.go @@ -6,6 +6,7 @@ package config import ( "fmt" "os" + "time" "gopkg.in/yaml.v3" ) @@ -50,6 +51,12 @@ const ( defaultGeminiModel = "gemini-2.5-flash" defaultAnthropicURL = "https://api.anthropic.com/v1" defaultClaudeModel = "claude-sonnet-4-6" + + defaultWatchInterval = "1h" + defaultWatchNotifyMinScore = 0.5 + defaultWatchNotifyMax = 10 + + MinWatchInterval = time.Minute ) var defaultTrackingParams = []string{ @@ -109,6 +116,14 @@ type AI struct { Anthropic Provider `yaml:"anthropic"` } +type Watch struct { + Interval string `yaml:"interval"` + WebhookURL string `yaml:"webhook_url"` + NotifyMinScore float64 `yaml:"notify_min_score"` + NotifyMaxItems int `yaml:"notify_max_items"` + NotifyOnKEV bool `yaml:"notify_on_kev"` +} + type Config struct { DBPath string `yaml:"db_path"` SourcesPath string `yaml:"sources_path"` @@ -118,6 +133,7 @@ type Config struct { Cluster Cluster `yaml:"cluster"` Rank Rank `yaml:"rank"` AI AI `yaml:"ai"` + Watch Watch `yaml:"watch"` } func Default() Config { @@ -163,6 +179,12 @@ func Default() Config { Gemini: Provider{BaseURL: defaultGeminiURL, Model: defaultGeminiModel}, Anthropic: Provider{BaseURL: defaultAnthropicURL, Model: defaultClaudeModel}, }, + Watch: Watch{ + Interval: defaultWatchInterval, + NotifyMinScore: defaultWatchNotifyMinScore, + NotifyMaxItems: defaultWatchNotifyMax, + NotifyOnKEV: true, + }, } } @@ -232,5 +254,18 @@ func (c Config) validate() error { default: return fmt.Errorf("config: ai.provider must be one of qwen|openai|gemini|anthropic, got %q", c.AI.Provider) } + d, err := time.ParseDuration(c.Watch.Interval) + if err != nil { + return fmt.Errorf("config: watch.interval %q is not a valid duration: %w", c.Watch.Interval, err) + } + if d < MinWatchInterval { + return fmt.Errorf("config: watch.interval must be >= %s, got %s", MinWatchInterval, c.Watch.Interval) + } + if c.Watch.NotifyMinScore < 0 || c.Watch.NotifyMinScore > 1 { + return fmt.Errorf("config: watch.notify_min_score must be in [0,1], got %v", c.Watch.NotifyMinScore) + } + if c.Watch.NotifyMaxItems < 1 { + return fmt.Errorf("config: watch.notify_max_items must be >= 1, got %d", c.Watch.NotifyMaxItems) + } return nil } diff --git a/PROJECTS/intermediate/security-news-scraper/internal/store/store.go b/PROJECTS/intermediate/security-news-scraper/internal/store/store.go index e28bb698..f689ceff 100644 --- a/PROJECTS/intermediate/security-news-scraper/internal/store/store.go +++ b/PROJECTS/intermediate/security-news-scraper/internal/store/store.go @@ -347,12 +347,16 @@ func (s *Store) digestClusterRows(since int64) (map[int64]*DigestCluster, []int6 return nil, nil, fmt.Errorf("digest clusters: %w", err) } defer rows.Close() + return scanClusterRows(rows) +} + +func scanClusterRows(rows *sql.Rows) (map[int64]*DigestCluster, []int64, error) { byID := make(map[int64]*DigestCluster) var order []int64 for rows.Next() { var dc DigestCluster if err := rows.Scan(&dc.ClusterID, &dc.Key, &dc.Size, &dc.FirstSeen, &dc.LastSeen); err != nil { - return nil, nil, fmt.Errorf("digest clusters: scan: %w", err) + return nil, nil, fmt.Errorf("scan cluster rows: %w", err) } clone := dc byID[dc.ClusterID] = &clone diff --git a/PROJECTS/intermediate/security-news-scraper/internal/store/watch.go b/PROJECTS/intermediate/security-news-scraper/internal/store/watch.go new file mode 100644 index 00000000..d5b89796 --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/internal/store/watch.go @@ -0,0 +1,76 @@ +// ©AngelaMos | 2026 +// watch.go + +package store + +import ( + "fmt" + "strings" +) + +func (s *Store) NewlyFetchedClusters(sinceFetched int64) ([]DigestCluster, error) { + ids, err := s.clusterIDsFetchedSince(sinceFetched) + if err != nil { + return nil, err + } + if len(ids) == 0 { + return nil, nil + } + byID, order, err := s.clusterRowsByID(ids) + if err != nil { + return nil, err + } + if err := s.digestAttachArticles(byID); err != nil { + return nil, err + } + if err := s.digestAttachCVEs(byID); err != nil { + return nil, err + } + out := make([]DigestCluster, 0, len(order)) + for _, id := range order { + out = append(out, *byID[id]) + } + return out, nil +} + +func (s *Store) clusterIDsFetchedSince(sinceFetched int64) ([]int64, error) { + rows, err := s.db.Query(` + SELECT DISTINCT cm.cluster_id + FROM cluster_members cm + JOIN articles a ON a.id = cm.article_id + WHERE a.fetched_at >= ? + ORDER BY cm.cluster_id`, sinceFetched) + if err != nil { + return nil, fmt.Errorf("newly fetched clusters: %w", err) + } + defer rows.Close() + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("newly fetched clusters: scan: %w", err) + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +func (s *Store) clusterRowsByID(ids []int64) (map[int64]*DigestCluster, []int64, error) { + if len(ids) == 0 { + return map[int64]*DigestCluster{}, nil, nil + } + placeholders := make([]string, len(ids)) + args := make([]any, len(ids)) + for i, id := range ids { + placeholders[i] = "?" + args[i] = id + } + query := `SELECT id, cluster_key, size, first_seen, last_seen FROM clusters WHERE id IN (` + + strings.Join(placeholders, ",") + `) ORDER BY id` + rows, err := s.db.Query(query, args...) + if err != nil { + return nil, nil, fmt.Errorf("cluster rows by id: %w", err) + } + defer rows.Close() + return scanClusterRows(rows) +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/store/watch_test.go b/PROJECTS/intermediate/security-news-scraper/internal/store/watch_test.go new file mode 100644 index 00000000..a6199e7f --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/internal/store/watch_test.go @@ -0,0 +1,99 @@ +// ©AngelaMos | 2026 +// watch_test.go + +package store + +import "testing" + +func TestNewlyFetchedClustersFiltersByFetchTimeNotPublishTime(t *testing.T) { + s := openTemp(t) + src, _ := s.UpsertSource(SourceInput{Name: "a", URL: "https://a/f", Type: "rss", Weight: 1, Enabled: true}) + + aOld, _ := s.InsertArticle(Article{SourceID: src, CanonicalURL: "https://a/old", ContentHash: "cold", TitleHash: "told", Title: "old", PublishedAt: 100, FetchedAt: 100}) + aRecent, _ := s.InsertArticle(Article{SourceID: src, CanonicalURL: "https://a/recent", ContentHash: "crecent", TitleHash: "trecent", Title: "recently published, freshly fetched", PublishedAt: 940, FetchedAt: 1000}) + aFresh, _ := s.InsertArticle(Article{SourceID: src, CanonicalURL: "https://a/fresh", ContentHash: "cfresh", TitleHash: "tfresh", Title: "fresh", PublishedAt: 1000, FetchedAt: 1000}) + + if err := s.ReplaceClusters([]ClusterRow{ + {Key: "old", Members: []int64{aOld}, FirstSeen: 100, LastSeen: 100}, + {Key: "recent", Members: []int64{aRecent}, FirstSeen: 940, LastSeen: 940}, + {Key: "fresh", Members: []int64{aFresh}, FirstSeen: 1000, LastSeen: 1000}, + }); err != nil { + t.Fatal(err) + } + + got, err := s.NewlyFetchedClusters(1000) + if err != nil { + t.Fatal(err) + } + seen := map[string]bool{} + for _, c := range got { + for _, a := range c.Articles { + seen[a.CanonicalURL] = true + } + } + if len(got) != 2 { + t.Fatalf("newly fetched clusters = %d, want 2 (recent + fresh)", len(got)) + } + if !seen["https://a/recent"] { + t.Error("a story fetched this cycle whose publish time is just before the watermark must be surfaced by the fetch-time query") + } + if !seen["https://a/fresh"] { + t.Error("the fresh story must be surfaced") + } + if seen["https://a/old"] { + t.Error("a story fetched before the watermark must be excluded") + } + + dig, err := s.DigestClusters(1000) + if err != nil { + t.Fatal(err) + } + for _, c := range dig { + for _, a := range c.Articles { + if a.CanonicalURL == "https://a/recent" { + t.Error("this is the whole point of the separate query: the publish-time DigestClusters filter drops the recent-but-just-fetched story, so watch must use the fetch-time query instead") + } + } + } +} + +func TestNewlyFetchedClustersEmpty(t *testing.T) { + s := openTemp(t) + got, err := s.NewlyFetchedClusters(0) + if err != nil { + t.Fatal(err) + } + if len(got) != 0 { + t.Fatalf("expected no clusters on an empty store, got %d", len(got)) + } +} + +func TestNewlyFetchedClustersAttachesCVEs(t *testing.T) { + s := openTemp(t) + src, _ := s.UpsertSource(SourceInput{Name: "a", URL: "https://a/f", Type: "rss", Weight: 1, Enabled: true}) + a1, _ := s.InsertArticle(Article{SourceID: src, CanonicalURL: "https://a/1", ContentHash: "c1", TitleHash: "t1", Title: "kev story", PublishedAt: 5000, FetchedAt: 5000}) + if err := s.UpsertCVEStub("CVE-2026-9"); err != nil { + t.Fatal(err) + } + score := 9.8 + if err := s.UpdateCVEEnrichment(CVE{ID: "CVE-2026-9", CVSSScore: &score, IsKEV: true, EnrichedAt: 1, EnrichStatus: EnrichStatusOK}); err != nil { + t.Fatal(err) + } + if err := s.LinkArticleCVE(a1, "CVE-2026-9"); err != nil { + t.Fatal(err) + } + if err := s.ReplaceClusters([]ClusterRow{{Key: "1", Members: []int64{a1}, FirstSeen: 5000, LastSeen: 5000}}); err != nil { + t.Fatal(err) + } + + got, err := s.NewlyFetchedClusters(1000) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 || len(got[0].CVEs) != 1 { + t.Fatalf("want 1 cluster with 1 CVE, got %d clusters", len(got)) + } + if !got[0].CVEs[0].IsKEV { + t.Error("KEV flag should be attached to the newly fetched cluster") + } +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/watch/notify.go b/PROJECTS/intermediate/security-news-scraper/internal/watch/notify.go new file mode 100644 index 00000000..134c1138 --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/internal/watch/notify.go @@ -0,0 +1,114 @@ +// ©AngelaMos | 2026 +// notify.go + +package watch + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" +) + +const ( + notifyMaxStatus = 300 + defaultNotifyTimeout = 15 * time.Second + headerContentType = "Content-Type" + mimeJSON = "application/json" +) + +type Notifier interface { + Notify(ctx context.Context, r Report) error +} + +type WebhookNotifier struct { + URL string + Client *http.Client +} + +type webhookPayload struct { + Text string `json:"text"` + Content string `json:"content"` + Items []webhookItem `json:"items"` +} + +type webhookItem struct { + Title string `json:"title"` + URL string `json:"url"` + Score float64 `json:"score"` + MaxCVSS float64 `json:"max_cvss"` + IsKEV bool `json:"is_kev"` + CVEs []string `json:"cves,omitempty"` + Sources int `json:"sources"` +} + +func (w WebhookNotifier) Notify(ctx context.Context, r Report) error { + if len(r.Notable) == 0 { + return nil + } + body, err := json.Marshal(buildPayload(r)) + if err != nil { + return fmt.Errorf("notify: marshal: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, w.URL, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("notify: new request: %w", err) + } + req.Header.Set(headerContentType, mimeJSON) + resp, err := w.client().Do(req) + if err != nil { + return fmt.Errorf("notify: post: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode >= notifyMaxStatus { + return fmt.Errorf("notify: webhook returned %s", resp.Status) + } + return nil +} + +func (w WebhookNotifier) client() *http.Client { + if w.Client != nil { + return w.Client + } + return &http.Client{Timeout: defaultNotifyTimeout} +} + +func buildPayload(r Report) webhookPayload { + summary := summarize(r) + items := make([]webhookItem, len(r.Notable)) + for i, n := range r.Notable { + items[i] = webhookItem{ + Title: n.Title, + URL: n.URL, + Score: n.Score, + MaxCVSS: n.MaxCVSS, + IsKEV: n.IsKEV, + CVEs: n.CVEs, + Sources: n.Sources, + } + } + return webhookPayload{Text: summary, Content: summary, Items: items} +} + +func summarize(r Report) string { + var sb strings.Builder + fmt.Fprintf(&sb, "nadezhda: %d notable %s\n", len(r.Notable), storyWord(len(r.Notable))) + for _, n := range r.Notable { + tag := "" + if n.IsKEV { + tag = " [KEV]" + } + fmt.Fprintf(&sb, "- %s%s (%s)\n", n.Title, tag, n.URL) + } + return strings.TrimRight(sb.String(), "\n") +} + +func storyWord(n int) string { + if n == 1 { + return "story" + } + return "stories" +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/watch/notify_test.go b/PROJECTS/intermediate/security-news-scraper/internal/watch/notify_test.go new file mode 100644 index 00000000..619225c3 --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/internal/watch/notify_test.go @@ -0,0 +1,89 @@ +// ©AngelaMos | 2026 +// notify_test.go + +package watch + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func sampleReport() Report { + return Report{ + Notable: []NotableItem{ + {Title: "Critical bug in Foo", URL: "https://x/1", Score: 0.91, MaxCVSS: 9.8, IsKEV: true, CVEs: []string{"CVE-2026-1"}, Sources: 3}, + {Title: "Breach at Bar", URL: "https://x/2", Score: 0.62, Sources: 2}, + }, + } +} + +func TestWebhookNotifierPostsPayload(t *testing.T) { + got := make(chan webhookPayload, 1) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Errorf("method = %s, want POST", r.Method) + } + if ct := r.Header.Get("Content-Type"); ct != "application/json" { + t.Errorf("content-type = %q, want application/json", ct) + } + var p webhookPayload + if err := json.NewDecoder(r.Body).Decode(&p); err != nil { + t.Errorf("decode: %v", err) + } + got <- p + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + n := WebhookNotifier{URL: srv.URL, Client: srv.Client()} + if err := n.Notify(context.Background(), sampleReport()); err != nil { + t.Fatal(err) + } + + p := <-got + if len(p.Items) != 2 { + t.Fatalf("items = %d, want 2", len(p.Items)) + } + if p.Items[0].Title != "Critical bug in Foo" || !p.Items[0].IsKEV { + t.Errorf("first item wrong: %+v", p.Items[0]) + } + if p.Text == "" || p.Content == "" { + t.Error("text and content must both be set for Slack/Discord compatibility") + } + if !strings.Contains(p.Text, "[KEV]") { + t.Errorf("summary should flag KEV items, got %q", p.Text) + } +} + +func TestWebhookNotifierNoopWhenNothingNotable(t *testing.T) { + hits := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits++ + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + n := WebhookNotifier{URL: srv.URL, Client: srv.Client()} + if err := n.Notify(context.Background(), Report{NewArticles: 5}); err != nil { + t.Fatal(err) + } + if hits != 0 { + t.Errorf("webhook was called %d times, want 0 when nothing is notable", hits) + } +} + +func TestWebhookNotifierErrorsOnBadStatus(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + n := WebhookNotifier{URL: srv.URL, Client: srv.Client()} + if err := n.Notify(context.Background(), sampleReport()); err == nil { + t.Error("expected an error when the webhook returns 500") + } +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/watch/report.go b/PROJECTS/intermediate/security-news-scraper/internal/watch/report.go new file mode 100644 index 00000000..f293c25f --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/internal/watch/report.go @@ -0,0 +1,28 @@ +// ©AngelaMos | 2026 +// report.go + +package watch + +import "time" + +type NotableItem struct { + Title string + URL string + Score float64 + MaxCVSS float64 + IsKEV bool + CVEs []string + Sources int +} + +type Report struct { + Start time.Time + Duration time.Duration + NewArticles int + Duplicates int + Clusters int + Enriched int + KEVHits int + Failed int + Notable []NotableItem +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/watch/watch.go b/PROJECTS/intermediate/security-news-scraper/internal/watch/watch.go new file mode 100644 index 00000000..d35c9a7c --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/internal/watch/watch.go @@ -0,0 +1,127 @@ +// ©AngelaMos | 2026 +// watch.go + +package watch + +import ( + "context" + "fmt" + "io" + "time" +) + +type Ticker interface { + C() <-chan time.Time + Stop() +} + +type realTicker struct{ t *time.Ticker } + +func (r realTicker) C() <-chan time.Time { return r.t.C } +func (r realTicker) Stop() { r.t.Stop() } + +func NewRealTicker(d time.Duration) Ticker { return realTicker{t: time.NewTicker(d)} } + +type Options struct { + Interval time.Duration + RunAtStart bool + Cycle func(context.Context) (Report, error) + Notifier Notifier + NewTicker func(time.Duration) Ticker + Out io.Writer +} + +func (o Options) validate() error { + if o.Cycle == nil { + return fmt.Errorf("watch: Cycle is required") + } + if o.Interval <= 0 { + return fmt.Errorf("watch: Interval must be > 0, got %v", o.Interval) + } + return nil +} + +func Run(ctx context.Context, opts Options) error { + if err := opts.validate(); err != nil { + return err + } + out := resolveOut(opts.Out) + newTicker := opts.NewTicker + if newTicker == nil { + newTicker = NewRealTicker + } + + fmt.Fprintf(out, "watch: starting, interval %s\n", opts.Interval) + + if opts.RunAtStart { + if stop, err := cycleAndNotify(ctx, opts, out); stop { + return shutdown(out) + } else if err != nil { + fmt.Fprintf(out, "watch: cycle error: %v\n", err) + } + } + + ticker := newTicker(opts.Interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return shutdown(out) + case <-ticker.C(): + if stop, err := cycleAndNotify(ctx, opts, out); stop { + return shutdown(out) + } else if err != nil { + fmt.Fprintf(out, "watch: cycle error: %v\n", err) + } + } + } +} + +func Once(ctx context.Context, opts Options) error { + if err := opts.validate(); err != nil { + return err + } + out := resolveOut(opts.Out) + _, err := cycleAndNotify(ctx, opts, out) + return err +} + +func cycleAndNotify(ctx context.Context, opts Options, out io.Writer) (stop bool, err error) { + if ctx.Err() != nil { + return true, nil + } + report, err := opts.Cycle(ctx) + if err != nil { + if ctx.Err() != nil { + return true, nil + } + return false, err + } + logReport(out, report) + if opts.Notifier != nil && len(report.Notable) > 0 { + if nerr := opts.Notifier.Notify(ctx, report); nerr != nil { + fmt.Fprintf(out, "watch: notify error: %v\n", nerr) + } else { + fmt.Fprintf(out, "watch: notified %d notable\n", len(report.Notable)) + } + } + return false, nil +} + +func shutdown(out io.Writer) error { + fmt.Fprintln(out, "watch: shutdown") + return nil +} + +func resolveOut(out io.Writer) io.Writer { + if out != nil { + return out + } + return io.Discard +} + +func logReport(out io.Writer, r Report) { + fmt.Fprintf(out, "watch: cycle done in %s: %d new, %d dup, %d clusters, %d enriched (%d KEV), %d failed, %d notable\n", + r.Duration.Round(time.Millisecond), r.NewArticles, r.Duplicates, r.Clusters, r.Enriched, r.KEVHits, r.Failed, len(r.Notable)) +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/watch/watch_test.go b/PROJECTS/intermediate/security-news-scraper/internal/watch/watch_test.go new file mode 100644 index 00000000..cd9732a3 --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/internal/watch/watch_test.go @@ -0,0 +1,198 @@ +// ©AngelaMos | 2026 +// watch_test.go + +package watch + +import ( + "context" + "errors" + "sync" + "testing" + "time" +) + +type fakeTicker struct{ ch chan time.Time } + +func (f *fakeTicker) C() <-chan time.Time { return f.ch } +func (f *fakeTicker) Stop() {} + +type fakeNotifier struct { + mu sync.Mutex + calls int + last Report +} + +func (f *fakeNotifier) Notify(_ context.Context, r Report) error { + f.mu.Lock() + defer f.mu.Unlock() + f.calls++ + f.last = r + return nil +} + +func (f *fakeNotifier) count() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.calls +} + +func recvWithin(t *testing.T, ch <-chan int, d time.Duration) int { + t.Helper() + select { + case v := <-ch: + return v + case <-time.After(d): + t.Fatal("timed out waiting for a watch cycle") + return 0 + } +} + +func TestRunAtStartThenTicksThenGracefulShutdown(t *testing.T) { + ticker := &fakeTicker{ch: make(chan time.Time)} + ran := make(chan int, 4) + count := 0 + ctx, cancel := context.WithCancel(context.Background()) + + opts := Options{ + Interval: time.Hour, + RunAtStart: true, + NewTicker: func(time.Duration) Ticker { return ticker }, + Cycle: func(context.Context) (Report, error) { + count++ + ran <- count + return Report{}, nil + }, + } + + done := make(chan error, 1) + go func() { done <- Run(ctx, opts) }() + + if n := recvWithin(t, ran, 2*time.Second); n != 1 { + t.Fatalf("RunAtStart cycle = %d, want 1", n) + } + ticker.ch <- time.Unix(0, 0) + if n := recvWithin(t, ran, 2*time.Second); n != 2 { + t.Fatalf("post-tick cycle = %d, want 2", n) + } + + cancel() + select { + case err := <-done: + if err != nil { + t.Fatalf("Run returned %v, want nil on graceful shutdown", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Run did not return after ctx cancel") + } +} + +func TestRunReturnsNilAndSkipsCycleOnImmediateCancel(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + called := false + err := Run(ctx, Options{ + Interval: time.Hour, + RunAtStart: true, + NewTicker: func(time.Duration) Ticker { return &fakeTicker{ch: make(chan time.Time)} }, + Cycle: func(context.Context) (Report, error) { + called = true + return Report{}, nil + }, + }) + if err != nil { + t.Fatalf("Run = %v, want nil", err) + } + if called { + t.Error("cycle should not run when the context is already cancelled") + } +} + +func TestRunContinuesAfterCycleError(t *testing.T) { + ticker := &fakeTicker{ch: make(chan time.Time)} + ran := make(chan int, 4) + count := 0 + ctx, cancel := context.WithCancel(context.Background()) + + err := make(chan error, 1) + go func() { + err <- Run(ctx, Options{ + Interval: time.Hour, + RunAtStart: true, + NewTicker: func(time.Duration) Ticker { return ticker }, + Cycle: func(context.Context) (Report, error) { + count++ + if count == 1 { + return Report{}, errors.New("boom") + } + ran <- count + return Report{}, nil + }, + }) + }() + + ticker.ch <- time.Unix(0, 0) + if n := recvWithin(t, ran, 2*time.Second); n != 2 { + t.Fatalf("second cycle = %d, want 2 (daemon survived the first cycle error)", n) + } + cancel() + if e := <-err; e != nil { + t.Fatalf("Run = %v, want nil", e) + } +} + +func TestOnceNotifiesWhenNotable(t *testing.T) { + fn := &fakeNotifier{} + err := Once(context.Background(), Options{ + Interval: time.Hour, + Cycle: func(context.Context) (Report, error) { + return Report{Notable: []NotableItem{{Title: "x"}}}, nil + }, + Notifier: fn, + }) + if err != nil { + t.Fatal(err) + } + if fn.count() != 1 { + t.Fatalf("notify calls = %d, want 1", fn.count()) + } +} + +func TestOnceSkipsNotifyWhenNoNotable(t *testing.T) { + fn := &fakeNotifier{} + err := Once(context.Background(), Options{ + Interval: time.Hour, + Cycle: func(context.Context) (Report, error) { + return Report{NewArticles: 3}, nil + }, + Notifier: fn, + }) + if err != nil { + t.Fatal(err) + } + if fn.count() != 0 { + t.Fatalf("notify calls = %d, want 0 (nothing notable)", fn.count()) + } +} + +func TestOnceReturnsCycleError(t *testing.T) { + want := errors.New("cycle failed") + err := Once(context.Background(), Options{ + Interval: time.Hour, + Cycle: func(context.Context) (Report, error) { + return Report{}, want + }, + }) + if !errors.Is(err, want) { + t.Fatalf("Once error = %v, want %v", err, want) + } +} + +func TestValidateRejectsBadOptions(t *testing.T) { + if err := (Options{Interval: time.Hour}).validate(); err == nil { + t.Error("validate should reject a nil Cycle") + } + if err := (Options{Cycle: func(context.Context) (Report, error) { return Report{}, nil }}).validate(); err == nil { + t.Error("validate should reject a non-positive Interval") + } +}