feat(monitor/collectors/cfradar): outages + hijacks repository with diff lookups

PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.mod

@@ -70,6 +70,7 @@ require (
 	github.com/lestrrat-go/httprc/v3 v3.0.1 // indirect
 	github.com/lestrrat-go/option v1.0.1 // indirect
 	github.com/lestrrat-go/option/v2 v2.0.0 // indirect
+	github.com/lib/pq v1.12.3 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/magiconair/properties v1.8.10 // indirect
 	github.com/mdelapenya/tlscert v0.2.0 // indirect

PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.sum

@@ -131,6 +131,8 @@ github.com/lestrrat-go/option/v2 v2.0.0 h1:XxrcaJESE1fokHy3FpaQ/cXW8ZsIdWcdFzzLO
 github.com/lestrrat-go/option/v2 v2.0.0/go.mod h1:oSySsmzMoR0iRzCDCaUfsCzxQHUEuhOViQObyy7S6Vg=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/lib/pq v1.12.3 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ=
+github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA=
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4=
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
 github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE=

PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cfradar/repo.go

@@ -0,0 +1,125 @@
+// ©AngelaMos | 2026
+// repo.go
+
+package cfradar
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+	"github.com/lib/pq"
+)
+
+type OutageRow struct {
+	ID         string          `db:"id"`
+	StartedAt  time.Time       `db:"started_at"`
+	EndedAt    *time.Time      `db:"ended_at"`
+	Locations  pq.StringArray  `db:"locations"`
+	ASNs       pq.Int32Array   `db:"asns"`
+	Cause      string          `db:"cause"`
+	OutageType string          `db:"outage_type"`
+	Payload    json.RawMessage `db:"payload"`
+}
+
+type HijackRow struct {
+	ID          int64           `db:"id"`
+	DetectedAt  time.Time       `db:"detected_at"`
+	StartedAt   time.Time       `db:"started_at"`
+	DurationSec int32           `db:"duration_sec"`
+	Confidence  int16           `db:"confidence"`
+	HijackerASN int32           `db:"hijacker_asn"`
+	VictimASNs  pq.Int32Array   `db:"victim_asns"`
+	Prefixes    []string        `db:"-"`
+	Payload     json.RawMessage `db:"payload"`
+}
+
+type Repo struct {
+	db *sqlx.DB
+}
+
+func NewRepo(db *sqlx.DB) *Repo { return &Repo{db: db} }
+
+func (r *Repo) UpsertOutage(ctx context.Context, o OutageRow) error {
+	if o.Locations == nil {
+		o.Locations = pq.StringArray{}
+	}
+	if o.ASNs == nil {
+		o.ASNs = pq.Int32Array{}
+	}
+	_, err := r.db.ExecContext(ctx, `
+		INSERT INTO outage_events (id, started_at, ended_at, locations, asns, cause, outage_type, payload)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+		ON CONFLICT (id) DO UPDATE SET
+			ended_at    = EXCLUDED.ended_at,
+			locations   = EXCLUDED.locations,
+			asns        = EXCLUDED.asns,
+			cause       = EXCLUDED.cause,
+			outage_type = EXCLUDED.outage_type,
+			payload     = EXCLUDED.payload`,
+		o.ID, o.StartedAt, o.EndedAt, o.Locations, o.ASNs, o.Cause, o.OutageType, []byte(o.Payload),
+	)
+	if err != nil {
+		return fmt.Errorf("upsert outage %s: %w", o.ID, err)
+	}
+	return nil
+}
+
+func (r *Repo) UpsertHijack(ctx context.Context, h HijackRow) error {
+	if h.VictimASNs == nil {
+		h.VictimASNs = pq.Int32Array{}
+	}
+	if h.Prefixes == nil {
+		h.Prefixes = []string{}
+	}
+	_, err := r.db.ExecContext(ctx, `
+		INSERT INTO bgp_hijack_events
+			(id, detected_at, started_at, duration_sec, confidence, hijacker_asn, victim_asns, prefixes, payload)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8::cidr[], $9)
+		ON CONFLICT (id) DO UPDATE SET
+			detected_at  = EXCLUDED.detected_at,
+			duration_sec = EXCLUDED.duration_sec,
+			confidence   = EXCLUDED.confidence,
+			payload      = EXCLUDED.payload`,
+		h.ID, h.DetectedAt, h.StartedAt, h.DurationSec, h.Confidence, h.HijackerASN,
+		h.VictimASNs, pq.Array(h.Prefixes), []byte(h.Payload),
+	)
+	if err != nil {
+		return fmt.Errorf("upsert hijack %d: %w", h.ID, err)
+	}
+	return nil
+}
+
+func (r *Repo) KnownOutageIDs(ctx context.Context, ids []string) (map[string]bool, error) {
+	if len(ids) == 0 {
+		return map[string]bool{}, nil
+	}
+	var found []string
+	if err := r.db.SelectContext(ctx, &found,
+		`SELECT id FROM outage_events WHERE id = ANY($1::text[])`, pq.Array(ids)); err != nil {
+		return nil, fmt.Errorf("known outage ids: %w", err)
+	}
+	out := make(map[string]bool, len(found))
+	for _, id := range found {
+		out[id] = true
+	}
+	return out, nil
+}
+
+func (r *Repo) KnownHijackIDs(ctx context.Context, ids []int64) (map[int64]bool, error) {
+	if len(ids) == 0 {
+		return map[int64]bool{}, nil
+	}
+	var found []int64
+	if err := r.db.SelectContext(ctx, &found,
+		`SELECT id FROM bgp_hijack_events WHERE id = ANY($1::bigint[])`, pq.Array(ids)); err != nil {
+		return nil, fmt.Errorf("known hijack ids: %w", err)
+	}
+	out := make(map[int64]bool, len(found))
+	for _, id := range found {
+		out[id] = true
+	}
+	return out, nil
+}

PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/cfradar/repo_test.go

@@ -0,0 +1,136 @@
+// ©AngelaMos | 2026
+// repo_test.go
+
+package cfradar_test
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	_ "github.com/jackc/pgx/v5/stdlib"
+	"github.com/jmoiron/sqlx"
+	"github.com/lib/pq"
+	"github.com/stretchr/testify/require"
+	"github.com/testcontainers/testcontainers-go/modules/postgres"
+
+	"github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/cfradar"
+)
+
+func setupDB(t *testing.T) *sqlx.DB {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+	defer cancel()
+
+	pg, err := postgres.Run(ctx, "postgres:17-alpine",
+		postgres.WithDatabase("monitor"),
+		postgres.WithUsername("monitor"),
+		postgres.WithPassword("monitor"),
+		postgres.BasicWaitStrategies(),
+	)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = pg.Terminate(context.Background()) })
+
+	dsn, err := pg.ConnectionString(ctx, "sslmode=disable")
+	require.NoError(t, err)
+
+	db, err := sqlx.ConnectContext(ctx, "pgx", dsn)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = db.Close() })
+
+	_, err = db.ExecContext(ctx, `
+		CREATE TABLE outage_events (
+			id          text PRIMARY KEY,
+			started_at  timestamptz NOT NULL,
+			ended_at    timestamptz,
+			locations   text[] NOT NULL DEFAULT '{}',
+			asns        integer[] NOT NULL DEFAULT '{}',
+			cause       text NOT NULL DEFAULT '',
+			outage_type text NOT NULL DEFAULT '',
+			payload     jsonb NOT NULL
+		);
+		CREATE TABLE bgp_hijack_events (
+			id            bigint PRIMARY KEY,
+			detected_at   timestamptz NOT NULL,
+			started_at    timestamptz NOT NULL,
+			duration_sec  integer NOT NULL DEFAULT 0,
+			confidence    smallint NOT NULL DEFAULT 0,
+			hijacker_asn  integer NOT NULL DEFAULT 0,
+			victim_asns   integer[] NOT NULL DEFAULT '{}',
+			prefixes      cidr[] NOT NULL DEFAULT '{}',
+			payload       jsonb NOT NULL
+		)`)
+	require.NoError(t, err)
+	return db
+}
+
+func TestRepo_UpsertOutageIsIdempotent(t *testing.T) {
+	db := setupDB(t)
+	repo := cfradar.NewRepo(db)
+	ctx := context.Background()
+
+	row := cfradar.OutageRow{
+		ID:         "out-1",
+		StartedAt:  time.Now().UTC().Add(-time.Hour),
+		Locations:  pq.StringArray{"DE", "AT"},
+		ASNs:       pq.Int32Array{15169},
+		Cause:      "fiber-cut",
+		OutageType: "regional",
+		Payload:    json.RawMessage(`{"id":"out-1"}`),
+	}
+	require.NoError(t, repo.UpsertOutage(ctx, row))
+	require.NoError(t, repo.UpsertOutage(ctx, row))
+}
+
+func TestRepo_UpsertHijackAndDiff(t *testing.T) {
+	db := setupDB(t)
+	repo := cfradar.NewRepo(db)
+	ctx := context.Background()
+
+	row := cfradar.HijackRow{
+		ID:          7700001,
+		DetectedAt:  time.Now().UTC(),
+		StartedAt:   time.Now().UTC().Add(-2 * time.Minute),
+		Confidence:  9,
+		HijackerASN: 64500,
+		VictimASNs:  pq.Int32Array{15169, 13335},
+		Prefixes:    []string{"203.0.113.0/24"},
+		Payload:     json.RawMessage(`{"id":7700001}`),
+	}
+	require.NoError(t, repo.UpsertHijack(ctx, row))
+
+	known, err := repo.KnownHijackIDs(ctx, []int64{7700001, 7700002})
+	require.NoError(t, err)
+	require.Equal(t, map[int64]bool{7700001: true}, known)
+}
+
+func TestRepo_KnownOutageIDs(t *testing.T) {
+	db := setupDB(t)
+	repo := cfradar.NewRepo(db)
+	ctx := context.Background()
+
+	require.NoError(t, repo.UpsertOutage(ctx, cfradar.OutageRow{
+		ID: "out-known", StartedAt: time.Now().UTC(),
+		Payload: json.RawMessage(`{}`),
+	}))
+
+	known, err := repo.KnownOutageIDs(ctx, []string{"out-known", "out-new"})
+	require.NoError(t, err)
+	require.Equal(t, map[string]bool{"out-known": true}, known)
+}
+
+func TestRepo_KnownIDsHandlesEmptySlice(t *testing.T) {
+	db := setupDB(t)
+	repo := cfradar.NewRepo(db)
+	ctx := context.Background()
+
+	out, err := repo.KnownOutageIDs(ctx, nil)
+	require.NoError(t, err)
+	require.Empty(t, out)
+
+	hij, err := repo.KnownHijackIDs(ctx, nil)
+	require.NoError(t, err)
+	require.Empty(t, hij)
+}