next-tjwater-server/app/mcp/peripheral.py

"""Read existing project-owned peripheral facts without forwarding MCP tokens."""

from datetime import datetime, timezone
from uuid import uuid4
from fastapi import HTTPException
from app.infra.db.postgresql.read_queries import MAX_QUERY_BYTES, read_rows, scada_devices, series_rows
from app.domain.schemas.operational_data import TaskDataSnapshot, TaskDataQuery
from app.services.content_digest import digest
from samples.peripheral.store import PostgresStore
from app.infra.integrations.operational_data import _select_demo_observations


async def read_peripheral_query(scope, args):
    conn = scope.connection
    project = str(scope.project.project_id)
    truncated = False
    origin = "system_data"
    if args.source in {"peripheral_work_orders", "peripheral_test_data"}:
        table, column = (
            ("work_orders", "work_order_id")
            if args.source == "peripheral_work_orders"
            else ("test_data_records", "record_id")
        )
        rows = await read_rows(
            conn,
            "SELECT document FROM peripheral_sample."
            + table
            + " WHERE project_id=%s AND deleted_at IS NULL ORDER BY "
            + column
            + " LIMIT %s OFFSET %s",
            (project, args.limit + 1, args.offset),
        )
        items = [
            PostgresStore._active_managed_document(row["document"])
            for row in rows[: args.limit]
        ]
        truncated = len(rows) > args.limit
        if any(item.get("project_id") != project for item in items):
            raise HTTPException(502, "Peripheral project mismatch")
        data = {
            "items": items,
            "truncated": truncated,
            "limit": args.limit,
            "offset": args.offset,
        }
        origin = (
            "synthetic"
            if items and any(item.get("origin") == "synthetic" for item in items)
            else "system_data"
        )
    elif args.source == "peripheral_points":
        rows = await scada_devices(conn, limit=args.limit + 1, offset=args.offset)
        data = {
            "origin": "historical_replay",
            "items": [
                {
                    "point_id": row["device_id"],
                    "label": row["device_id"] + " · " + row["device_type"],
                    "unit": row["measurement_unit"],
                    "status": "unknown",
                    "origin": "historical_replay",
                }
                for row in rows[: args.limit]
            ],
            "truncated": len(rows) > args.limit,
        }
        truncated = data["truncated"]
    elif args.point_id.startswith("DEMO-"):
        rows = await read_rows(
            conn,
            "SELECT CASE WHEN octet_length(document::text)<=%s THEN document END AS document,"
            "digest,octet_length(document::text)>%s AS too_large "
            "FROM peripheral_sample.demo_datasets WHERE project_id=%s AND dataset_id='water-demo-v1'",
            (MAX_QUERY_BYTES, MAX_QUERY_BYTES, project),
        )
        if rows and rows[0]["too_large"]:
            raise HTTPException(413, "Synthetic dataset exceeds the integrity-check byte budget")
        if not rows or digest(rows[0]["document"]) != rows[0]["digest"]:
            raise HTTPException(503, "Synthetic dataset is unavailable or invalid")
        query = TaskDataQuery(
            source=args.source, point_id=args.point_id, start=args.start, end=args.end
        )
        data = _select_demo_observations(
            {**rows[0]["document"], "project_id": project}, query, project
        )
        items = data["items"]
        truncated = len(items) > args.limit + args.offset
        data["items"] = items[args.offset : args.offset + args.limit]
        data["truncated"] = truncated
        origin = "synthetic"
    else:
        devices = await scada_devices(conn, device_id=args.point_id, limit=1)
        if not devices:
            raise HTTPException(404, "Peripheral point was not found")
        device = devices[0]
        if not device.get("api_query_id"):
            raise HTTPException(503, "Peripheral history mapping is unavailable")
        async with scope.timescale() as ts:
            rows = await series_rows(
                ts,
                source="scada",
                element_type="junction",
                ids=[device["api_query_id"]],
                start=args.start,
                end=args.end,
                field="monitored_value",
                limit=args.limit + 1,
                offset=args.offset,
                end_exclusive=True,
            )
        truncated = len(rows) > args.limit
        data = {
            "origin": "historical_replay",
            "items": [
                {
                    "point_id": args.point_id,
                    "observed_at": row["time"].isoformat(),
                    "value": row["value"],
                    "unit": device["measurement_unit"],
                }
                for row in rows[: args.limit]
            ],
            "truncated": truncated,
        }
    return TaskDataSnapshot(
        source=args.source,
        source_ref="mcp:" + str(uuid4()),
        queried_at=datetime.now(timezone.utc),
        data=data,
        origin=origin,
        truncated=truncated,
    )