feat(monitor/collectors/usgs): earthquakes repository (upsert, known-ids, recent-by-mag)

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

@@ -0,0 +1,84 @@
+// ©AngelaMos | 2026
+// repo.go
+
+package usgs
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+	"github.com/lib/pq"
+)
+
+type Row struct {
+	ID         string          `db:"id"`
+	OccurredAt time.Time       `db:"occurred_at"`
+	Mag        float64         `db:"mag"`
+	Place      string          `db:"place"`
+	GeomLon    float64         `db:"geom_lon"`
+	GeomLat    float64         `db:"geom_lat"`
+	DepthKm    float64         `db:"depth_km"`
+	Payload    json.RawMessage `db:"payload"`
+}
+
+type Repo struct {
+	db *sqlx.DB
+}
+
+func NewRepo(db *sqlx.DB) *Repo { return &Repo{db: db} }
+
+func (r *Repo) Upsert(ctx context.Context, row Row) error {
+	_, err := r.db.ExecContext(ctx, `
+		INSERT INTO earthquakes (id, occurred_at, mag, place, geom_lon, geom_lat, depth_km, payload)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+		ON CONFLICT (id) DO UPDATE SET
+			occurred_at = EXCLUDED.occurred_at,
+			mag         = EXCLUDED.mag,
+			place       = EXCLUDED.place,
+			geom_lon    = EXCLUDED.geom_lon,
+			geom_lat    = EXCLUDED.geom_lat,
+			depth_km    = EXCLUDED.depth_km,
+			payload     = EXCLUDED.payload`,
+		row.ID, row.OccurredAt, row.Mag, row.Place,
+		row.GeomLon, row.GeomLat, row.DepthKm, []byte(row.Payload),
+	)
+	if err != nil {
+		return fmt.Errorf("upsert earthquake %s: %w", row.ID, err)
+	}
+	return nil
+}
+
+func (r *Repo) KnownIDs(ctx context.Context, ids []string) (map[string]bool, error) {
+	out := make(map[string]bool, len(ids))
+	if len(ids) == 0 {
+		return out, nil
+	}
+	var found []string
+	err := r.db.SelectContext(ctx, &found,
+		`SELECT id FROM earthquakes WHERE id = ANY($1)`,
+		pq.Array(ids),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("known earthquake ids: %w", err)
+	}
+	for _, id := range found {
+		out[id] = true
+	}
+	return out, nil
+}
+
+func (r *Repo) RecentByMag(ctx context.Context, limit int) ([]Row, error) {
+	var rows []Row
+	err := r.db.SelectContext(ctx, &rows, `
+		SELECT id, occurred_at, mag, place, geom_lon, geom_lat, depth_km, payload
+		  FROM earthquakes
+		 ORDER BY mag DESC, occurred_at DESC
+		 LIMIT $1`, limit)
+	if err != nil {
+		return nil, fmt.Errorf("recent earthquakes: %w", err)
+	}
+	return rows, nil
+}

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

@@ -0,0 +1,124 @@
+// ©AngelaMos | 2026
+// repo_test.go
+
+package usgs_test
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	_ "github.com/jackc/pgx/v5/stdlib"
+	"github.com/jmoiron/sqlx"
+	"github.com/stretchr/testify/require"
+	"github.com/testcontainers/testcontainers-go/modules/postgres"
+
+	"github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/usgs"
+)
+
+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 earthquakes (
+			id              text PRIMARY KEY,
+			occurred_at     timestamptz NOT NULL,
+			mag             numeric(3,1) NOT NULL,
+			place           text,
+			geom_lon        double precision NOT NULL,
+			geom_lat        double precision NOT NULL,
+			depth_km        double precision,
+			payload         jsonb NOT NULL
+		);`)
+	require.NoError(t, err)
+	return db
+}
+
+func TestRepo_UpsertOverwritesRevisions(t *testing.T) {
+	db := setupDB(t)
+	repo := usgs.NewRepo(db)
+	ctx := context.Background()
+
+	row := usgs.Row{
+		ID:         "us6000abcd",
+		OccurredAt: time.Now().UTC().Add(-time.Hour),
+		Mag:        5.5,
+		Place:      "100 km SW of test",
+		GeomLon:    -130.5,
+		GeomLat:    49.2,
+		DepthKm:    10.0,
+		Payload:    json.RawMessage(`{"v":1}`),
+	}
+	require.NoError(t, repo.Upsert(ctx, row))
+
+	row.Mag = 5.8
+	row.Payload = json.RawMessage(`{"v":2}`)
+	require.NoError(t, repo.Upsert(ctx, row))
+
+	got, err := repo.RecentByMag(ctx, 1)
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	require.InDelta(t, 5.8, got[0].Mag, 0.0001)
+}
+
+func TestRepo_KnownIDsBulkLookup(t *testing.T) {
+	db := setupDB(t)
+	repo := usgs.NewRepo(db)
+	ctx := context.Background()
+
+	for _, id := range []string{"a", "b", "c"} {
+		require.NoError(t, repo.Upsert(ctx, usgs.Row{
+			ID: id, OccurredAt: time.Now().UTC(), Mag: 4.5,
+			GeomLon: 0, GeomLat: 0, Payload: json.RawMessage(`{}`),
+		}))
+	}
+
+	known, err := repo.KnownIDs(ctx, []string{"a", "b", "x", "y"})
+	require.NoError(t, err)
+	require.True(t, known["a"])
+	require.True(t, known["b"])
+	require.False(t, known["x"])
+	require.False(t, known["y"])
+}
+
+func TestRepo_RecentByMagOrdered(t *testing.T) {
+	db := setupDB(t)
+	repo := usgs.NewRepo(db)
+	ctx := context.Background()
+
+	now := time.Now().UTC()
+	rows := []usgs.Row{
+		{ID: "small", OccurredAt: now, Mag: 4.5, GeomLon: 0, GeomLat: 0, Payload: json.RawMessage(`{}`)},
+		{ID: "huge", OccurredAt: now, Mag: 7.8, GeomLon: 0, GeomLat: 0, Payload: json.RawMessage(`{}`)},
+		{ID: "mid", OccurredAt: now, Mag: 6.0, GeomLon: 0, GeomLat: 0, Payload: json.RawMessage(`{}`)},
+	}
+	for _, r := range rows {
+		require.NoError(t, repo.Upsert(ctx, r))
+	}
+
+	got, err := repo.RecentByMag(ctx, 3)
+	require.NoError(t, err)
+	require.Len(t, got, 3)
+	require.Equal(t, "huge", got[0].ID)
+	require.Equal(t, "mid", got[1].ID)
+	require.Equal(t, "small", got[2].ID)
+}