feat(workbench): add runtime GIS and sample integrations

app/infra/db/postgresql/analysis.py

@@ -101,7 +101,11 @@ class AnalysisRepository:
             return cur.fetchone()
 
     @staticmethod
-    async def list_runs(conn: AsyncConnection) -> list[dict]:
+    async def list_runs(
+        conn: AsyncConnection, *, limit: int = 100, offset: int = 0
+    ) -> list[dict]:
+        if not 1 <= limit <= 100 or offset < 0:
+            raise ValueError("analysis run page is out of range")
         async with conn.cursor() as cur:
             await cur.execute(
                 """
@@ -109,7 +113,52 @@ class AnalysisRepository:
                        started_at, status, parameters
                 FROM analysis.runs
                 ORDER BY created_at DESC, run_id
+                LIMIT %s OFFSET %s
+                """,
+                (limit, offset),
+            )
+            return await cur.fetchall()
+
+    @staticmethod
+    async def list_recent_runs(
+        conn: AsyncConnection, *, limit: int = 100
+    ) -> list[dict]:
+        if not 1 <= limit <= 100:
+            raise ValueError("recent analysis run limit must be between 1 and 100")
+        async with conn.cursor() as cur:
+            await cur.execute(
                 """
+                SELECT run_id, name, run_type, created_by, created_at,
+                       started_at, status, parameters
+                FROM analysis.runs
+                ORDER BY created_at DESC, run_id
+                LIMIT %s
+                """,
+                (limit,),
+            )
+            return await cur.fetchall()
+
+    @staticmethod
+    async def list_run_summaries(
+        conn: AsyncConnection, *, limit: int = 21, offset: int = 0
+    ) -> list[dict]:
+        if not 1 <= limit <= 101 or offset < 0:
+            raise ValueError("analysis run summary page is out of range")
+        async with conn.cursor() as cur:
+            await cur.execute(
+                """
+                SELECT r.run_id, r.name, r.run_type, r.created_by,
+                       r.created_at, r.started_at, r.status,
+                       EXISTS (
+                           SELECT 1
+                           FROM analysis.results AS result
+                           WHERE result.run_id = r.run_id
+                       ) AS has_results
+                FROM analysis.runs AS r
+                ORDER BY r.created_at DESC, r.run_id
+                LIMIT %s OFFSET %s
+                """,
+                (limit, offset),
             )
             return await cur.fetchall()
 
@@ -129,20 +178,75 @@ class AnalysisRepository:
 
     @staticmethod
     async def list_results(
-        conn: AsyncConnection, run_id: UUID, result_type: str | None = None
+        conn: AsyncConnection,
+        run_id: UUID,
+        result_type: str | None = None,
+        *,
+        limit: int = 100,
+        offset: int = 0,
     ) -> list[dict]:
+        if not 1 <= limit <= 500 or offset < 0:
+            raise ValueError("analysis result page is out of range")
         query = """
             SELECT result_id, run_id, result_type, node_id, link_id,
                    payload, created_at
             FROM analysis.results
             WHERE run_id = %s
         """
-        params: tuple[UUID] | tuple[UUID, str] = (run_id,)
+        params: tuple = (run_id,)
         if result_type is not None:
             query += " AND result_type = %s"
             params = (run_id, result_type)
-        query += " ORDER BY created_at, result_id"
+        query += " ORDER BY created_at, result_id LIMIT %s OFFSET %s"
+        params += (limit, offset)
 
         async with conn.cursor() as cur:
             await cur.execute(query, params)
             return await cur.fetchall()
+
+    @staticmethod
+    async def list_recent_results(
+        conn: AsyncConnection, run_id: UUID, *, limit: int = 100
+    ) -> list[dict]:
+        if not 1 <= limit <= 100:
+            raise ValueError("recent analysis result limit must be between 1 and 100")
+        async with conn.cursor() as cur:
+            await cur.execute(
+                """
+                SELECT result_id, run_id, result_type, node_id, link_id,
+                       payload, created_at
+                FROM (
+                    SELECT result_id, run_id, result_type, node_id, link_id,
+                           payload, created_at
+                    FROM analysis.results
+                    WHERE run_id = %s
+                    ORDER BY created_at DESC, result_id DESC
+                    LIMIT %s
+                ) AS recent_results
+                ORDER BY created_at, result_id
+                """,
+                (run_id, limit),
+            )
+            return await cur.fetchall()
+
+    @staticmethod
+    async def list_junction_map_points(
+        conn: AsyncConnection, node_ids: list[str]
+    ) -> list[dict]:
+        if not node_ids:
+            return []
+        if len(node_ids) > 500:
+            raise ValueError("analysis map projection supports at most 500 junctions")
+        async with conn.cursor() as cur:
+            await cur.execute(
+                """
+                SELECT g.id AS node_id,
+                       ST_X(ST_Transform(g.geom, 4326)) AS longitude,
+                       ST_Y(ST_Transform(g.geom, 4326)) AS latitude
+                FROM gis.junctions AS g
+                WHERE g.id = ANY(%s)
+                ORDER BY g.id
+                """,
+                (node_ids,),
+            )
+            return await cur.fetchall()

app/infra/db/postgresql/gis_attributes.py

@@ -0,0 +1,80 @@
+"""Read-only access to imported source GIS attributes."""
+
+from typing import Any
+
+from psycopg import AsyncConnection
+
+
+class GisAttributeRepository:
+    @staticmethod
+    async def list_element_groups(
+        conn: AsyncConnection,
+        element_type: str,
+        element_id: str,
+    ) -> list[dict[str, Any]]:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                "SELECT to_regclass('source_gis.element_links') AS relation"
+            )
+            availability = await cur.fetchone()
+            if not availability or availability["relation"] is None:
+                return []
+            await cur.execute(
+                """
+                SELECT
+                    dataset.dataset_key,
+                    dataset.title AS dataset_title,
+                    dataset.source_date,
+                    layer.layer_key,
+                    layer.display_name AS layer_name,
+                    feature.source_fid,
+                    link.match_method,
+                    link.distance_m AS match_distance_m,
+                    feature.attributes
+                FROM source_gis.element_links AS link
+                JOIN source_gis.features AS feature
+                  ON feature.layer_key = link.layer_key
+                 AND feature.source_fid = link.source_fid
+                JOIN source_gis.layers AS layer
+                  ON layer.layer_key = feature.layer_key
+                JOIN source_gis.datasets AS dataset
+                  ON dataset.dataset_key = layer.dataset_key
+                WHERE link.element_type = %s
+                  AND link.element_id = %s
+                ORDER BY dataset.source_date DESC, layer.display_name, feature.source_fid
+                """,
+                (element_type, element_id),
+            )
+            return [dict(row) for row in await cur.fetchall()]
+
+    @staticmethod
+    async def get_source_feature_group(
+        conn: AsyncConnection,
+        layer_key: str,
+        source_fid: int,
+    ) -> dict[str, Any] | None:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                """
+                SELECT
+                    dataset.dataset_key,
+                    dataset.title AS dataset_title,
+                    dataset.source_date,
+                    layer.layer_key,
+                    layer.display_name AS layer_name,
+                    feature.source_fid,
+                    'source' AS match_method,
+                    NULL::double precision AS match_distance_m,
+                    feature.attributes
+                FROM source_gis.features AS feature
+                JOIN source_gis.layers AS layer
+                  ON layer.layer_key = feature.layer_key
+                JOIN source_gis.datasets AS dataset
+                  ON dataset.dataset_key = layer.dataset_key
+                WHERE feature.layer_key = %s
+                  AND feature.source_fid = %s
+                """,
+                (layer_key, source_fid),
+            )
+            row = await cur.fetchone()
+            return dict(row) if row else None

app/infra/db/postgresql/sensor_placement.py

@@ -24,6 +24,8 @@ def _placement_row(row: dict[str, Any]) -> dict[str, Any]:
         "created_by": row["created_by"],
         "created_at": row["created_at"],
         "status": row["status"],
+        "sensor_type": payload.get("sensor_type", "pressure"),
+        "method": payload.get("method"),
         "sensor_locations": locations,
     }
 
@@ -57,11 +59,15 @@ def create_sensor_placement(
     min_diameter: int,
     created_by: str,
     sensor_locations: list[str],
+    sensor_type: str = "pressure",
+    method: str | None = None,
 ) -> dict[str, Any]:
     run_id = uuid4()
     payload = {
         "sensor_number": len(sensor_locations),
         "minimum_diameter": min_diameter,
+        "sensor_type": sensor_type,
+        "method": method,
         "sensor_locations": sensor_locations,
     }
     with project_connection(name) as conn, conn.transaction():

app/infra/db/postgresql/task_alerts.py

@@ -0,0 +1,88 @@
+"""PostgreSQL persistence for task-produced alert events and current state."""
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+
+from psycopg import AsyncConnection
+from psycopg.errors import UndefinedTable
+from psycopg.types.json import Jsonb
+
+from app.domain.domain_errors import DomainError
+
+
+class TaskAlertRepository:
+    def __init__(self, connection: AsyncConnection, project_id):
+        self.connection = connection
+        self.project_id = str(project_id)
+
+    @asynccontextmanager
+    async def transaction(self):
+        try:
+            async with self.connection.transaction():
+                yield self
+        except UndefinedTable as exc:
+            raise DomainError(
+                "DEPENDENCY_UNAVAILABLE",
+                "Project alert storage is not provisioned",
+                status=503,
+            ) from exc
+
+    async def lock_alert(self, alert_id: str) -> None:
+        await self.connection.execute(
+            "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
+            (f"tjwater-alert:{self.project_id}:{alert_id}",),
+        )
+
+    async def get_event(self, producer_id: str, event_id: str) -> dict | None:
+        cursor = await self.connection.execute(
+            """SELECT document FROM workbench.alert_events
+               WHERE project_id=%s AND producer_id=%s AND event_id=%s""",
+            (self.project_id, producer_id, event_id),
+        )
+        row = await cursor.fetchone()
+        return row["document"] if row else None
+
+    async def get_alert(self, alert_id: str) -> tuple[int, dict] | None:
+        cursor = await self.connection.execute(
+            """SELECT revision,document FROM workbench.alerts
+               WHERE project_id=%s AND alert_id=%s""",
+            (self.project_id, alert_id),
+        )
+        row = await cursor.fetchone()
+        return (row["revision"], row["document"]) if row else None
+
+    async def save_event(self, event: dict, resulting_state: dict) -> None:
+        document = event | {"resulting_state": resulting_state}
+        await self.connection.execute(
+            """INSERT INTO workbench.alert_events
+               (project_id,producer_id,event_id,alert_id,event_type,occurred_at,document)
+               VALUES (%s,%s,%s,%s,%s,%s,%s)""",
+            (
+                self.project_id,
+                event["producer_id"],
+                event["event_id"],
+                event["alert_id"],
+                event["event_type"],
+                event["occurred_at"],
+                Jsonb(document),
+            ),
+        )
+
+    async def save_alert(self, state: dict) -> None:
+        await self.connection.execute(
+            """INSERT INTO workbench.alerts
+               (project_id,alert_id,revision,status,occurred_at,updated_at,document)
+               VALUES (%s,%s,%s,%s,%s,%s,%s)
+               ON CONFLICT(project_id,alert_id) DO UPDATE SET
+                 revision=excluded.revision,status=excluded.status,
+                 updated_at=excluded.updated_at,document=excluded.document""",
+            (
+                self.project_id,
+                state["alert_id"],
+                state["revision"],
+                state["status"],
+                state["occurred_at"],
+                state["updated_at"],
+                Jsonb({key: value for key, value in state.items() if key != "revision"}),
+            ),
+        )

app/infra/db/postgresql/workbench_operations.py

@@ -0,0 +1,70 @@
+"""Optional project-local schedule, work-order and alert projections."""
+
+from dataclasses import dataclass
+from typing import Any
+
+from psycopg import AsyncConnection
+
+
+@dataclass(frozen=True)
+class WorkbenchOperationsAvailability:
+    work_orders: bool
+    alerts: bool
+    schedules: bool = False
+
+
+class WorkbenchOperationsRepository:
+    """Read provisioned workbench records without creating or mutating storage."""
+
+    @staticmethod
+    async def availability(
+        conn: AsyncConnection,
+    ) -> WorkbenchOperationsAvailability:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                """
+                SELECT
+                    to_regclass('workbench.work_orders') IS NOT NULL AS work_orders,
+                    to_regclass('workbench.alerts') IS NOT NULL AS alerts
+                """
+            )
+            row = await cur.fetchone()
+        return WorkbenchOperationsAvailability(
+            work_orders=bool(row and row["work_orders"]),
+            alerts=bool(row and row["alerts"]),
+        )
+
+    @staticmethod
+    async def list_work_orders(
+        conn: AsyncConnection, project_id: str, *, limit: int = 100
+    ) -> list[dict[str, Any]]:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                """
+                SELECT document
+                FROM workbench.work_orders
+                WHERE project_id = %s
+                ORDER BY updated_at DESC, work_order_id
+                LIMIT %s
+                """,
+                (project_id, limit),
+            )
+            return [row["document"] for row in await cur.fetchall()]
+
+    @staticmethod
+    async def list_alerts(
+        conn: AsyncConnection, project_id: str, *, limit: int = 100
+    ) -> list[dict[str, Any]]:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                """
+                SELECT document
+                FROM workbench.alerts
+                WHERE project_id = %s
+                  AND status = 'active'
+                ORDER BY occurred_at DESC, alert_id
+                LIMIT %s
+                """,
+                (project_id, limit),
+            )
+            return [row["document"] for row in await cur.fetchall()]