From 7b1de3f99583f43c6e2be3c71196b9c156000c05 Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Sun, 19 Apr 2026 17:28:38 -0500
Subject: [PATCH] feat(prop-grid-rs): surface grid_tasks.kind in claim_next
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Introduces TaskKind { Forecast, Analysis } and threads it through the
claim flow. Forecast-only filter on the claim query keeps the claim
lane clean while Elixir still owns f00 — a half-deployed cluster won't
deadlock on an unfulfillable analysis row.

Prep for Stream A of Phase 3: once the Rust-side native-duct + NEXRAD +
commercial pipeline lands, a sibling claim_next_analysis function pulls
kind='analysis' rows and dispatches to the new pipeline.
---
 rust/prop_grid_rs/src/db.rs | 38 +++++++++++++++++++++++++++++++++++--
 1 file changed, 36 insertions(+), 2 deletions(-)

diff --git a/rust/prop_grid_rs/src/db.rs b/rust/prop_grid_rs/src/db.rs
index c55bc8fc..04614389 100644
--- a/rust/prop_grid_rs/src/db.rs
+++ b/rust/prop_grid_rs/src/db.rs
@@ -15,6 +15,30 @@ use uuid::Uuid;
 
 pub const NOTIFY_CHANNEL: &str = "propagation_ready";
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum TaskKind {
+    /// f01..f18 — fetch + decode + score the forecast-hour grid. This is
+    /// what Rust has owned since Phase 2 cutover.
+    Forecast,
+    /// f00 enrichment — adds native-level duct merge, NEXRAD composite,
+    /// and commercial-link degradation on top of the forecast pipeline,
+    /// and writes the ProfilesFile consumed by /weather. Introduced in
+    /// Phase 3 Stream A.
+    Analysis,
+}
+
+impl TaskKind {
+    fn from_str(s: &str) -> Self {
+        match s {
+            "analysis" => TaskKind::Analysis,
+            // Default to Forecast so older producers (which wrote rows
+            // before the kind column existed, defaulted to 'forecast'
+            // by migration) keep working.
+            _ => TaskKind::Forecast,
+        }
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct ClaimedTask {
     pub id: Uuid,
@@ -22,6 +46,7 @@ pub struct ClaimedTask {
     pub forecast_hour: i16,
     pub valid_time: DateTime<Utc>,
     pub attempt: i32,
+    pub kind: TaskKind,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -44,11 +69,16 @@ pub async fn connect(url: &str, max_connections: u32) -> Result<PgPool, DbError>
 /// stamps claimed_at. Returns `None` if the queue is empty.
 pub async fn claim_next(pool: &PgPool) -> Result<Option<ClaimedTask>, DbError> {
     let mut tx = pool.begin().await?;
+    // Forecast-only claim. Analysis rows (kind='analysis') go through
+    // `claim_next_analysis` once the Rust-side f00 pipeline is wired;
+    // keeping them out of the forecast lane means a half-deployed cluster
+    // (Elixir still owns f00, Rust pulls forecasts) doesn't deadlock on
+    // an unfulfillable analysis row.
     let row = sqlx::query(
         r#"
-        SELECT id, run_time, forecast_hour, valid_time, attempt
+        SELECT id, run_time, forecast_hour, valid_time, attempt, kind
         FROM grid_tasks
-        WHERE status = 'queued' AND forecast_hour > 0
+        WHERE status = 'queued' AND forecast_hour > 0 AND kind = 'forecast'
         ORDER BY run_time ASC, forecast_hour ASC
         FOR UPDATE SKIP LOCKED
         LIMIT 1
@@ -74,6 +104,8 @@ pub async fn claim_next(pool: &PgPool) -> Result<Option<ClaimedTask>, DbError> {
     let valid_time_naive: NaiveDateTime = row.try_get("valid_time")?;
     let valid_time = DateTime::<Utc>::from_naive_utc_and_offset(valid_time_naive, Utc);
     let attempt: i32 = row.try_get("attempt")?;
+    let kind_str: String = row.try_get("kind")?;
+    let kind = TaskKind::from_str(&kind_str);
 
     sqlx::query(
         r#"
@@ -97,6 +129,7 @@ pub async fn claim_next(pool: &PgPool) -> Result<Option<ClaimedTask>, DbError> {
         forecast_hour,
         valid_time,
         attempt: attempt + 1,
+        kind,
     }))
 }
 
@@ -204,6 +237,7 @@ mod tests {
         let claimed = claim_next(&pool).await.unwrap().expect("had a row");
         assert!(claimed.forecast_hour > 0);
         assert_eq!(claimed.attempt, 1);
+        assert_eq!(claimed.kind, TaskKind::Forecast);
 
         complete(&pool, &claimed).await.unwrap();