From 11a68de2b79d1c418bd28659c950798f83e0dec8 Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Sat, 2 May 2026 04:20:49 -0400 Subject: [PATCH] feat(monitor/collectors/usgs): earthquakes repository (upsert, known-ids, recent-by-mag) --- .../backend/internal/collectors/usgs/repo.go | 84 ++++++++++++ .../internal/collectors/usgs/repo_test.go | 124 ++++++++++++++++++ 2 files changed, 208 insertions(+) create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/usgs/repo.go create mode 100644 PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/usgs/repo_test.go diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/usgs/repo.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/usgs/repo.go new file mode 100644 index 00000000..a1d585e7 --- /dev/null +++ b/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 +} diff --git a/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/usgs/repo_test.go b/PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/usgs/repo_test.go new file mode 100644 index 00000000..84c115e4 --- /dev/null +++ b/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) +}