feat(nadezhda): M7 watch daemon with optional webhook notify
Add a watch command (--interval, --once, --no-enrich): a long-lived daemon that re-ingests every enabled source on a ticker and, when a webhook is configured, POSTs genuinely new high-signal stories to it (Slack, Discord, or any JSON endpoint via a text/content/items payload). internal/watch is a pure, stdlib-only scheduler: the pipeline is injected as a Cycle closure, so the daemon is fully unit-testable with a fake ticker and notifier. Graceful shutdown on SIGINT/SIGTERM returns cleanly; a cycle error is logged and the loop continues (fail-soft). Notify uses a dedicated fetch-time watermark (store.NewlyFetchedClusters) so a freshly ingested but older-dated advisory is surfaced, which a publish-time filter would drop. Notable stories are capped by watch.notify_max_items and filtered by score threshold or KEV status. Extract the shared ingest and cluster sequence (pipeline.go) so scrape and watch cannot drift, and route scrape and enrich through cmd.Context() so they honor SIGTERM too.
This commit is contained in:
CarterPerez-dev
parent
d57bcc15b6
commit
f65d197c96
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
@ -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)
|
||||
}
|
||||
|
|
@ -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))
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
@ -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")
|
||||
}
|
||||
}
|
||||
|
|
@ -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"
|
||||
}
|
||||
|
|
@ -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")
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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))
|
||||
}
|
||||
|
|
@ -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")
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue