Phase 2: analytics rollups, geo breakdown, event feed (backend)

- Idempotent hourly rollup (run_rollup, INSERT ... ON CONFLICT DO UPDATE into
  analytics_rollups_hourly) + Tokio background task (startup then hourly)
- GET /v1/cards/{id}/analytics/geo — country-grouped visit counts
- GeoResolver trait: real MaxMind under the `geoip` feature, no-op default;
  raw IP resolved then discarded (only hash + coarse country/city persisted)
- analytics_events now store country/city; rollups read query for the dashboard

2 new tests (geo grouping, rollup aggregation + idempotency); 75 backend tests.
fmt + clippy clean; geoip feature compiles.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

cardclaws-backend/crates/cardclaws-db/src/models/analytics.rs

@@ -14,6 +14,26 @@ pub struct AnalyticsSummary {
     pub link_clicks: i64,
 }
 
+/// One hourly rollup bucket (PRD §9.3 `analytics_rollups_hourly`).
+#[derive(Debug, Clone, FromRow, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RollupRow {
+    pub hour: DateTime<Utc>,
+    pub visits: i32,
+    pub qr_scans: i32,
+    pub nfc_taps: i32,
+    pub saves: i32,
+    pub link_clicks: i32,
+}
+
+/// A country's visit count for the geo distribution (PRD §6.7.1).
+#[derive(Debug, Clone, FromRow, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GeoCount {
+    pub country: String,
+    pub visits: i64,
+}
+
 /// One row of the chronological event feed (PRD §6.7.2).
 #[derive(Debug, Clone, FromRow, Serialize)]
 #[serde(rename_all = "camelCase")]

cardclaws-backend/crates/cardclaws-db/src/queries/analytics.rs

@@ -5,7 +5,7 @@
 
 use uuid::Uuid;
 
-use crate::models::analytics::{AnalyticsSummary, FeedEvent};
+use crate::models::analytics::{AnalyticsSummary, FeedEvent, GeoCount, RollupRow};
 use crate::Db;
 
 pub struct NewEvent<'a> {
@@ -14,18 +14,23 @@ pub struct NewEvent<'a> {
     /// Correlates the event to the share link that produced it (PRD §6.5.2).
     pub share_token: Option<&'a str>,
     pub ip_hash: Option<&'a str>,
+    pub country: Option<&'a str>,
+    pub city: Option<&'a str>,
     pub user_agent: Option<&'a str>,
 }
 
 pub async fn insert_event(db: &Db, ev: NewEvent<'_>) -> Result<(), sqlx::Error> {
     sqlx::query(
-        r#"INSERT INTO analytics_events (card_id, event_type, share_token, ip_hash, user_agent)
-           VALUES ($1, $2, $3, $4, $5)"#,
+        r#"INSERT INTO analytics_events
+             (card_id, event_type, share_token, ip_hash, country, city, user_agent)
+           VALUES ($1, $2, $3, $4, $5, $6, $7)"#,
     )
     .bind(ev.card_id)
     .bind(ev.event_type)
     .bind(ev.share_token)
     .bind(ev.ip_hash)
+    .bind(ev.country)
+    .bind(ev.city)
     .bind(ev.user_agent)
     .execute(db)
     .await?;
@@ -66,3 +71,62 @@ pub async fn feed(db: &Db, card_id: Uuid, limit: i64) -> Result<Vec<FeedEvent>,
     .fetch_all(db)
     .await
 }
+
+/// Visit counts grouped by country, descending (PRD §6.7.1 geo distribution).
+/// Rows with no resolved country are excluded.
+pub async fn geo_breakdown(db: &Db, card_id: Uuid) -> Result<Vec<GeoCount>, sqlx::Error> {
+    sqlx::query_as(
+        r#"SELECT country, COUNT(*) AS visits
+           FROM analytics_events
+           WHERE card_id = $1 AND country IS NOT NULL
+           GROUP BY country
+           ORDER BY visits DESC"#,
+    )
+    .bind(card_id)
+    .fetch_all(db)
+    .await
+}
+
+/// Read a card's hourly rollup buckets, newest first.
+pub async fn rollups(db: &Db, card_id: Uuid) -> Result<Vec<RollupRow>, sqlx::Error> {
+    sqlx::query_as(
+        r#"SELECT hour, visits, qr_scans, nfc_taps, saves, link_clicks
+           FROM analytics_rollups_hourly
+           WHERE card_id = $1
+           ORDER BY hour DESC"#,
+    )
+    .bind(card_id)
+    .fetch_all(db)
+    .await
+}
+
+/// Recompute the hourly rollups from the raw event log (PRD §18.2). Idempotent:
+/// `ON CONFLICT DO UPDATE` overwrites each (card, hour) bucket, so re-running is
+/// safe and self-correcting. Returns the number of buckets written.
+pub async fn run_rollup(db: &Db) -> Result<u64, sqlx::Error> {
+    let result = sqlx::query(
+        r#"
+        INSERT INTO analytics_rollups_hourly
+            (card_id, hour, visits, qr_scans, nfc_taps, saves, link_clicks)
+        SELECT
+            card_id,
+            date_trunc('hour', occurred_at) AS hour,
+            COUNT(*) FILTER (WHERE event_type = 'profile_visit'),
+            COUNT(*) FILTER (WHERE event_type = 'qr_scan'),
+            COUNT(*) FILTER (WHERE event_type = 'nfc_tap'),
+            COUNT(*) FILTER (WHERE event_type = 'contact_save'),
+            COUNT(*) FILTER (WHERE event_type = 'link_click')
+        FROM analytics_events
+        GROUP BY card_id, date_trunc('hour', occurred_at)
+        ON CONFLICT (card_id, hour) DO UPDATE SET
+            visits = EXCLUDED.visits,
+            qr_scans = EXCLUDED.qr_scans,
+            nfc_taps = EXCLUDED.nfc_taps,
+            saves = EXCLUDED.saves,
+            link_clicks = EXCLUDED.link_clicks
+        "#,
+    )
+    .execute(db)
+    .await?;
+    Ok(result.rows_affected())
+}