diff --git a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/scrape.go b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/scrape.go index 6d3c5587..0b49e69d 100644 --- a/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/scrape.go +++ b/PROJECTS/intermediate/security-news-scraper/cmd/nadezhda/scrape.go @@ -12,12 +12,15 @@ import ( "github.com/spf13/cobra" + "github.com/CarterPerez-dev/nadezhda/internal/cluster" "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" ) +const secondsPerHour = 3600 + const ( statusNotModified = "304" statusError = "error" @@ -70,12 +73,21 @@ func runScrape(cmd *cobra.Command, args []string) error { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) defer stop() - summary, err := ingest.Run(ctx, fc, st, cfg, targets, time.Now()) + now := time.Now() + summary, err := ingest.Run(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) return nil } diff --git a/PROJECTS/intermediate/security-news-scraper/internal/cluster/cluster.go b/PROJECTS/intermediate/security-news-scraper/internal/cluster/cluster.go new file mode 100644 index 00000000..2176e2df --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/internal/cluster/cluster.go @@ -0,0 +1,208 @@ +// ©AngelaMos | 2026 +// cluster.go + +package cluster + +import ( + "sort" + "strconv" + "strings" + + "github.com/CarterPerez-dev/nadezhda/internal/normalize" +) + +type Item struct { + ID int64 + SourceID int64 + Title string + CVEs []string + Time int64 +} + +type Cluster struct { + Key string + Members []int64 + SourceCount int + FirstSeen int64 + LastSeen int64 +} + +type tokenized struct { + item Item + tokens map[string]struct{} + cves map[string]struct{} +} + +func Compute(items []Item, jaccardThreshold float64, windowSeconds int64) []Cluster { + n := len(items) + prepared := make([]tokenized, n) + for i, it := range items { + prepared[i] = tokenized{ + item: it, + tokens: tokenSet(it.Title), + cves: stringSet(it.CVEs), + } + } + + uf := newUnionFind(n) + for i := 0; i < n; i++ { + for j := i + 1; j < n; j++ { + if abs64(prepared[i].item.Time-prepared[j].item.Time) > windowSeconds { + continue + } + sharedCVE := shareAny(prepared[i].cves, prepared[j].cves) + crossSource := prepared[i].item.SourceID != prepared[j].item.SourceID + titleMatch := crossSource && + jaccard(prepared[i].tokens, prepared[j].tokens) >= jaccardThreshold + if sharedCVE || titleMatch { + uf.union(i, j) + } + } + } + + groups := make(map[int][]int) + for i := 0; i < n; i++ { + root := uf.find(i) + groups[root] = append(groups[root], i) + } + + clusters := make([]Cluster, 0, len(groups)) + for _, members := range groups { + clusters = append(clusters, buildCluster(prepared, members)) + } + sort.Slice(clusters, func(a, b int) bool { + if clusters[a].FirstSeen != clusters[b].FirstSeen { + return clusters[a].FirstSeen < clusters[b].FirstSeen + } + return clusters[a].Members[0] < clusters[b].Members[0] + }) + return clusters +} + +func buildCluster(prepared []tokenized, members []int) Cluster { + earliest := prepared[members[0]].item + ids := make([]int64, 0, len(members)) + sources := make(map[int64]struct{}, len(members)) + first := earliest.Time + last := earliest.Time + for _, m := range members { + it := prepared[m].item + ids = append(ids, it.ID) + sources[it.SourceID] = struct{}{} + if it.Time < first { + first = it.Time + } + if it.Time > last { + last = it.Time + } + if it.Time < earliest.Time || (it.Time == earliest.Time && it.ID < earliest.ID) { + earliest = it + } + } + sort.Slice(ids, func(a, b int) bool { return ids[a] < ids[b] }) + return Cluster{ + Key: strconv.FormatInt(earliest.ID, 10), + Members: ids, + SourceCount: len(sources), + FirstSeen: first, + LastSeen: last, + } +} + +func tokenSet(title string) map[string]struct{} { + fields := strings.Fields(normalize.NormalizeTitle(title)) + set := make(map[string]struct{}, len(fields)) + for _, f := range fields { + set[f] = struct{}{} + } + return set +} + +func stringSet(values []string) map[string]struct{} { + set := make(map[string]struct{}, len(values)) + for _, v := range values { + if v != "" { + set[v] = struct{}{} + } + } + return set +} + +func jaccard(a, b map[string]struct{}) float64 { + if len(a) == 0 || len(b) == 0 { + return 0 + } + small, large := a, b + if len(small) > len(large) { + small, large = large, small + } + intersection := 0 + for k := range small { + if _, ok := large[k]; ok { + intersection++ + } + } + union := len(a) + len(b) - intersection + if union == 0 { + return 0 + } + return float64(intersection) / float64(union) +} + +func shareAny(a, b map[string]struct{}) bool { + if len(a) == 0 || len(b) == 0 { + return false + } + small, large := a, b + if len(small) > len(large) { + small, large = large, small + } + for k := range small { + if _, ok := large[k]; ok { + return true + } + } + return false +} + +func abs64(v int64) int64 { + if v < 0 { + return -v + } + return v +} + +type unionFind struct { + parent []int + rank []int +} + +func newUnionFind(n int) *unionFind { + uf := &unionFind{parent: make([]int, n), rank: make([]int, n)} + for i := range uf.parent { + uf.parent[i] = i + } + return uf +} + +func (uf *unionFind) find(x int) int { + for uf.parent[x] != x { + uf.parent[x] = uf.parent[uf.parent[x]] + x = uf.parent[x] + } + return x +} + +func (uf *unionFind) union(a, b int) { + ra, rb := uf.find(a), uf.find(b) + if ra == rb { + return + } + if uf.rank[ra] < uf.rank[rb] { + ra, rb = rb, ra + } + uf.parent[rb] = ra + if uf.rank[ra] == uf.rank[rb] { + uf.rank[ra]++ + } +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/cluster/cluster_test.go b/PROJECTS/intermediate/security-news-scraper/internal/cluster/cluster_test.go new file mode 100644 index 00000000..4c93afb9 --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/internal/cluster/cluster_test.go @@ -0,0 +1,148 @@ +// ©AngelaMos | 2026 +// cluster_test.go + +package cluster + +import ( + "sort" + "testing" +) + +const bigWindow = 1 << 40 + +func sizes(clusters []Cluster) []int { + out := make([]int, len(clusters)) + for i, c := range clusters { + out[i] = len(c.Members) + } + sort.Ints(out) + return out +} + +func findCluster(clusters []Cluster, id int64) Cluster { + for _, c := range clusters { + for _, m := range c.Members { + if m == id { + return c + } + } + } + return Cluster{} +} + +func TestComputeTitleSimilarityGroups(t *testing.T) { + items := []Item{ + {ID: 1, SourceID: 1, Title: "Acme Corp data breach exposes millions", Time: 100}, + {ID: 2, SourceID: 2, Title: "Acme Corp data breach exposed millions", Time: 200}, + {ID: 3, SourceID: 3, Title: "Acme Corp data breach hits millions", Time: 300}, + {ID: 4, SourceID: 4, Title: "Linux kernel 7 released with new scheduler", Time: 400}, + } + clusters := Compute(items, 0.6, bigWindow) + + if got := sizes(clusters); len(got) != 2 || got[0] != 1 || got[1] != 3 { + t.Fatalf("cluster sizes = %v, want [1 3]", got) + } + c := findCluster(clusters, 1) + if len(c.Members) != 3 { + t.Errorf("article 1 cluster size = %d, want 3", len(c.Members)) + } + if c.SourceCount != 3 { + t.Errorf("SourceCount = %d, want 3 (three distinct outlets)", c.SourceCount) + } + if c.Key != "1" { + t.Errorf("cluster key = %q, want 1 (earliest article)", c.Key) + } + if c.FirstSeen != 100 || c.LastSeen != 300 { + t.Errorf("span = [%d,%d], want [100,300]", c.FirstSeen, c.LastSeen) + } +} + +func TestComputeSameSourceTitlesDoNotMerge(t *testing.T) { + items := []Item{ + {ID: 1, SourceID: 1, Title: "CISA Adds One Known Exploited Vulnerability to Catalog", Time: 100}, + {ID: 2, SourceID: 1, Title: "CISA Adds One Known Exploited Vulnerability to Catalog", Time: 200}, + } + if got := sizes(Compute(items, 0.6, bigWindow)); len(got) != 2 { + t.Errorf("same-source identical titles sizes = %v, want [1 1] (clustering is cross-outlet)", got) + } +} + +func TestComputeSharedCVEJoinsRegardlessOfSource(t *testing.T) { + items := []Item{ + {ID: 1, SourceID: 1, Title: "Router vendor patches critical flaw", Time: 100, CVEs: []string{"CVE-2026-1111"}}, + {ID: 2, SourceID: 1, Title: "Enterprise gear gets emergency update", Time: 200, CVEs: []string{"CVE-2026-1111"}}, + {ID: 3, SourceID: 2, Title: "Totally unrelated privacy story", Time: 300, CVEs: []string{"CVE-2026-9999"}}, + } + clusters := Compute(items, 0.6, bigWindow) + if got := sizes(clusters); len(got) != 2 || got[0] != 1 || got[1] != 2 { + t.Fatalf("cluster sizes = %v, want [1 2]", got) + } + joined := findCluster(clusters, 1) + if len(joined.Members) != 2 { + t.Error("shared-CVE articles should join even from the same source") + } + if joined.SourceCount != 1 { + t.Errorf("SourceCount = %d, want 1 (a same-source cluster is NOT multi-source)", joined.SourceCount) + } +} + +func TestComputeThresholdBoundary(t *testing.T) { + items := []Item{ + {ID: 1, SourceID: 1, Title: "alpha bravo charlie delta", Time: 100}, + {ID: 2, SourceID: 2, Title: "alpha bravo charlie echo", Time: 200}, + } + if got := sizes(Compute(items, 0.6, bigWindow)); len(got) != 1 || got[0] != 2 { + t.Errorf("at threshold 0.6 (jaccard==0.6) sizes = %v, want [2] (>= joins)", got) + } + if got := sizes(Compute(items, 0.61, bigWindow)); len(got) != 2 { + t.Errorf("at threshold 0.61 sizes = %v, want [1 1] (below threshold)", got) + } +} + +func TestComputeWindowSeparates(t *testing.T) { + items := []Item{ + {ID: 1, SourceID: 1, Title: "identical breaking headline text", Time: 0}, + {ID: 2, SourceID: 2, Title: "identical breaking headline text", Time: 200}, + } + if got := sizes(Compute(items, 0.6, 100)); len(got) != 2 { + t.Errorf("outside window sizes = %v, want [1 1] (time gap > window)", got) + } + if got := sizes(Compute(items, 0.6, 300)); len(got) != 1 || got[0] != 2 { + t.Errorf("inside window sizes = %v, want [2]", got) + } +} + +func TestComputeKeyIsEarliestByTime(t *testing.T) { + items := []Item{ + {ID: 10, SourceID: 1, Title: "shared story about the incident", Time: 500}, + {ID: 20, SourceID: 2, Title: "shared story about the incident", Time: 100}, + } + clusters := Compute(items, 0.6, bigWindow) + if len(clusters) != 1 { + t.Fatalf("clusters = %d, want 1", len(clusters)) + } + if clusters[0].Key != "20" { + t.Errorf("key = %q, want 20 (earliest by time, not lowest id)", clusters[0].Key) + } + if clusters[0].FirstSeen != 100 || clusters[0].LastSeen != 500 { + t.Errorf("span = [%d,%d], want [100,500]", clusters[0].FirstSeen, clusters[0].LastSeen) + } +} + +func TestComputeTransitiveChain(t *testing.T) { + items := []Item{ + {ID: 1, SourceID: 1, Title: "one two three four five", Time: 100}, + {ID: 2, SourceID: 2, Title: "one two three four six", Time: 100}, + {ID: 3, SourceID: 3, Title: "one two three six five", Time: 100}, + } + clusters := Compute(items, 0.6, bigWindow) + if len(clusters) != 1 || len(clusters[0].Members) != 3 { + t.Errorf("transitive similarity should merge all three, got %d clusters", len(clusters)) + } +} + +func TestComputeEmpty(t *testing.T) { + if c := Compute(nil, 0.6, bigWindow); len(c) != 0 { + t.Errorf("empty input -> %d clusters, want 0", len(c)) + } +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/cluster/rebuild.go b/PROJECTS/intermediate/security-news-scraper/internal/cluster/rebuild.go new file mode 100644 index 00000000..b1103f43 --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/internal/cluster/rebuild.go @@ -0,0 +1,51 @@ +// ©AngelaMos | 2026 +// rebuild.go + +package cluster + +import ( + "github.com/CarterPerez-dev/nadezhda/internal/store" +) + +type Stats struct { + Total int + MultiSource int + LargestSize int +} + +func Rebuild(st *store.Store, jaccardThreshold float64, windowSeconds, sinceUnix int64) (Stats, error) { + candidates, err := st.ClusterCandidates(sinceUnix) + if err != nil { + return Stats{}, err + } + cveMap, err := st.ArticleCVEMap() + if err != nil { + return Stats{}, err + } + + items := make([]Item, len(candidates)) + for i, c := range candidates { + items[i] = Item{ID: c.ID, SourceID: c.SourceID, Title: c.Title, Time: c.Time, CVEs: cveMap[c.ID]} + } + + clusters := Compute(items, jaccardThreshold, windowSeconds) + + rows := make([]store.ClusterRow, len(clusters)) + stats := Stats{Total: len(clusters)} + for i, c := range clusters { + rows[i] = store.ClusterRow{ + Key: c.Key, Members: c.Members, FirstSeen: c.FirstSeen, LastSeen: c.LastSeen, + } + if c.SourceCount > 1 { + stats.MultiSource++ + } + if size := len(c.Members); size > stats.LargestSize { + stats.LargestSize = size + } + } + + if err := st.ReplaceClusters(rows); err != nil { + return Stats{}, err + } + return stats, nil +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/cluster/rebuild_test.go b/PROJECTS/intermediate/security-news-scraper/internal/cluster/rebuild_test.go new file mode 100644 index 00000000..6204d029 --- /dev/null +++ b/PROJECTS/intermediate/security-news-scraper/internal/cluster/rebuild_test.go @@ -0,0 +1,133 @@ +// ©AngelaMos | 2026 +// rebuild_test.go + +package cluster + +import ( + "fmt" + "path/filepath" + "testing" + + "github.com/CarterPerez-dev/nadezhda/internal/store" +) + +func openStore(t *testing.T) *store.Store { + t.Helper() + st, err := store.Open(filepath.Join(t.TempDir(), "cluster.db")) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + return st +} + +func seedArticles(t *testing.T, st *store.Store) { + t.Helper() + titles := []string{ + "Acme Corp data breach exposes millions", + "Acme Corp data breach exposed millions", + "Acme Corp data breach hits millions", + "Linux kernel 7 released with new scheduler", + } + for i, title := range titles { + sourceID, err := st.UpsertSource(store.SourceInput{ + Name: fmt.Sprintf("src%d", i), URL: fmt.Sprintf("https://outlet%d.example/feed", i), + Type: "rss", Weight: 1, Enabled: true, + }) + if err != nil { + t.Fatalf("upsert source %d: %v", i, err) + } + if _, err := st.InsertArticle(store.Article{ + SourceID: sourceID, + CanonicalURL: fmt.Sprintf("https://example.com/a/%d", i), + ContentHash: fmt.Sprintf("chash-%d", i), + TitleHash: fmt.Sprintf("thash-%d", i), + Title: title, + PublishedAt: int64(100 * (i + 1)), + FetchedAt: int64(100 * (i + 1)), + }); err != nil { + t.Fatalf("insert article %d: %v", i, err) + } + } +} + +func clusterSizes(t *testing.T, st *store.Store) []int { + t.Helper() + rows, err := st.DB().Query(`SELECT size FROM clusters ORDER BY size`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []int + for rows.Next() { + var n int + if err := rows.Scan(&n); err != nil { + t.Fatal(err) + } + out = append(out, n) + } + return out +} + +func countMembers(t *testing.T, st *store.Store) int { + t.Helper() + var n int + if err := st.DB().QueryRow(`SELECT COUNT(*) FROM cluster_members`).Scan(&n); err != nil { + t.Fatal(err) + } + return n +} + +func TestRebuildClustersEndToEnd(t *testing.T) { + st := openStore(t) + seedArticles(t, st) + + stats, err := Rebuild(st, 0.6, bigWindow, 0) + if err != nil { + t.Fatalf("Rebuild: %v", err) + } + if stats.Total != 2 || stats.MultiSource != 1 || stats.LargestSize != 3 { + t.Fatalf("stats = %+v, want {Total:2 MultiSource:1 LargestSize:3}", stats) + } + if got := clusterSizes(t, st); len(got) != 2 || got[0] != 1 || got[1] != 3 { + t.Errorf("cluster sizes = %v, want [1 3]", got) + } + if m := countMembers(t, st); m != 4 { + t.Errorf("cluster_members = %d, want 4 (every article assigned once)", m) + } +} + +func TestRebuildIsIdempotent(t *testing.T) { + st := openStore(t) + seedArticles(t, st) + + if _, err := Rebuild(st, 0.6, bigWindow, 0); err != nil { + t.Fatalf("first Rebuild: %v", err) + } + stats, err := Rebuild(st, 0.6, bigWindow, 0) + if err != nil { + t.Fatalf("second Rebuild: %v", err) + } + if stats.Total != 2 { + t.Errorf("second run clusters = %d, want 2 (rebuild replaces, no accumulation)", stats.Total) + } + if m := countMembers(t, st); m != 4 { + t.Errorf("cluster_members = %d, want 4 after re-run", m) + } +} + +func TestRebuildLookbackExcludesOld(t *testing.T) { + st := openStore(t) + seedArticles(t, st) + + stats, err := Rebuild(st, 0.6, bigWindow, 250) + if err != nil { + t.Fatalf("Rebuild: %v", err) + } + if stats.Total != 2 { + t.Errorf("with since=250 only articles at t>=250 cluster, total = %d, want 2", stats.Total) + } + if m := countMembers(t, st); m != 2 { + t.Errorf("members = %d, want 2 (two articles below lookback excluded)", m) + } +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/config/config.go b/PROJECTS/intermediate/security-news-scraper/internal/config/config.go index a8692f5f..1f2677f2 100644 --- a/PROJECTS/intermediate/security-news-scraper/internal/config/config.go +++ b/PROJECTS/intermediate/security-news-scraper/internal/config/config.go @@ -24,8 +24,9 @@ const ( defaultCacheTTLHours = 24 defaultNegativeTTLHours = 3 - defaultTitleJaccard = 0.6 - defaultWindowHours = 72 + defaultTitleJaccard = 0.6 + defaultWindowHours = 72 + defaultLookbackHours = 168 trackingUTMPrefix = "utm_*" @@ -73,6 +74,7 @@ type Enrich struct { type Cluster struct { TitleJaccard float64 `yaml:"title_jaccard_threshold"` WindowHours int `yaml:"window_hours"` + LookbackHours int `yaml:"lookback_hours"` TrackingParams []string `yaml:"tracking_params"` } @@ -136,6 +138,7 @@ func Default() Config { Cluster: Cluster{ TitleJaccard: defaultTitleJaccard, WindowHours: defaultWindowHours, + LookbackHours: defaultLookbackHours, TrackingParams: defaultTrackingParams, }, Rank: Rank{ @@ -208,6 +211,15 @@ func (c Config) validate() error { if c.Cluster.TitleJaccard < 0 || c.Cluster.TitleJaccard > 1 { return fmt.Errorf("config: cluster.title_jaccard_threshold must be in [0,1], got %v", c.Cluster.TitleJaccard) } + if c.Cluster.WindowHours < 1 { + return fmt.Errorf("config: cluster.window_hours must be >= 1, got %d", c.Cluster.WindowHours) + } + if c.Cluster.LookbackHours < 1 { + return fmt.Errorf("config: cluster.lookback_hours must be >= 1, got %d", c.Cluster.LookbackHours) + } + if c.Cluster.LookbackHours < c.Cluster.WindowHours { + return fmt.Errorf("config: cluster.lookback_hours (%d) must be >= cluster.window_hours (%d) or window edges near the corpus boundary are silently dropped", c.Cluster.LookbackHours, c.Cluster.WindowHours) + } if c.Rank.HalfLifeHours < 1 { return fmt.Errorf("config: rank.half_life_hours must be >= 1, got %d", c.Rank.HalfLifeHours) } diff --git a/PROJECTS/intermediate/security-news-scraper/internal/config/config_test.go b/PROJECTS/intermediate/security-news-scraper/internal/config/config_test.go index 9a490a58..2eaaf370 100644 --- a/PROJECTS/intermediate/security-news-scraper/internal/config/config_test.go +++ b/PROJECTS/intermediate/security-news-scraper/internal/config/config_test.go @@ -92,3 +92,15 @@ func TestValidateRejectsBadProvider(t *testing.T) { t.Error("expected error for invalid ai.provider") } } + +func TestValidateRejectsLookbackBelowWindow(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "lb.yaml") + body := "cluster:\n window_hours: 72\n lookback_hours: 24\n" + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatal(err) + } + if _, err := Load(path); err == nil { + t.Error("expected error when lookback_hours < window_hours") + } +} diff --git a/PROJECTS/intermediate/security-news-scraper/internal/store/store.go b/PROJECTS/intermediate/security-news-scraper/internal/store/store.go index a19dddc7..b56b7588 100644 --- a/PROJECTS/intermediate/security-news-scraper/internal/store/store.go +++ b/PROJECTS/intermediate/security-news-scraper/internal/store/store.go @@ -179,6 +179,100 @@ func (s *Store) UpsertFetchState(sourceID int64, fs FetchState) error { return nil } +type CandidateArticle struct { + ID int64 + SourceID int64 + Title string + Time int64 +} + +type ClusterRow struct { + Key string + Members []int64 + FirstSeen int64 + LastSeen int64 +} + +func (s *Store) ClusterCandidates(since int64) ([]CandidateArticle, error) { + rows, err := s.db.Query(` + SELECT id, source_id, title, COALESCE(NULLIF(published_at, 0), fetched_at) AS t + FROM articles + WHERE COALESCE(NULLIF(published_at, 0), fetched_at) >= ? + ORDER BY id`, since) + if err != nil { + return nil, fmt.Errorf("cluster candidates: %w", err) + } + defer rows.Close() + + var out []CandidateArticle + for rows.Next() { + var c CandidateArticle + if err := rows.Scan(&c.ID, &c.SourceID, &c.Title, &c.Time); err != nil { + return nil, fmt.Errorf("cluster candidates: scan: %w", err) + } + out = append(out, c) + } + return out, rows.Err() +} + +func (s *Store) ArticleCVEMap() (map[int64][]string, error) { + rows, err := s.db.Query(`SELECT article_id, cve_id FROM article_cves`) + if err != nil { + return nil, fmt.Errorf("article cve map: %w", err) + } + defer rows.Close() + + out := make(map[int64][]string) + for rows.Next() { + var articleID int64 + var cveID string + if err := rows.Scan(&articleID, &cveID); err != nil { + return nil, fmt.Errorf("article cve map: scan: %w", err) + } + out[articleID] = append(out[articleID], cveID) + } + return out, rows.Err() +} + +func (s *Store) ReplaceClusters(rows []ClusterRow) error { + tx, err := s.db.Begin() + if err != nil { + return fmt.Errorf("replace clusters: begin: %w", err) + } + defer func() { _ = tx.Rollback() }() + + if _, err := tx.Exec(`DELETE FROM cluster_members`); err != nil { + return fmt.Errorf("replace clusters: clear members: %w", err) + } + if _, err := tx.Exec(`DELETE FROM clusters`); err != nil { + return fmt.Errorf("replace clusters: clear clusters: %w", err) + } + + for _, r := range rows { + var clusterID int64 + if err := tx.QueryRow(` + INSERT INTO clusters (cluster_key, first_seen, last_seen, size) + VALUES (?, ?, ?, ?) RETURNING id`, + r.Key, r.FirstSeen, r.LastSeen, len(r.Members), + ).Scan(&clusterID); err != nil { + return fmt.Errorf("replace clusters: insert cluster %q: %w", r.Key, err) + } + for _, articleID := range r.Members { + if _, err := tx.Exec(` + INSERT INTO cluster_members (cluster_id, article_id) VALUES (?, ?)`, + clusterID, articleID, + ); err != nil { + return fmt.Errorf("replace clusters: insert member %d: %w", articleID, err) + } + } + } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("replace clusters: commit: %w", err) + } + return nil +} + func boolToInt(b bool) int { if b { return 1