diff --git a/.env.exemple b/.env.exemple index bd955ac..11bfcf2 100644 --- a/.env.exemple +++ b/.env.exemple @@ -38,6 +38,14 @@ STORAGE_ACCESS_KEY=minioadmin STORAGE_SECRET_KEY=minioadmin STORAGE_REGION=us-east-1 +# ── Realtime (Redis pub/sub) ── +# Fire-and-forget SSE hints consumed by crm-backend's realtime hub. Optional +# everywhere, including production: an empty URL makes core.realtime.emit() a +# silent no-op, so a broker outage degrades the UI to polling and never affects +# a business write. db 1 on purpose — db 0 is crm-backend's cache. +REALTIME_ENABLED=false +REALTIME_REDIS_URL=redis://:changeme_redis_password@localhost:6379/1 + # ── CORS ── # Comma-separated list of allowed origins. "*" is allowed only in local. ALLOW_ORIGIN=* diff --git a/.gitignore b/.gitignore index a96b88b..93ebac6 100644 --- a/.gitignore +++ b/.gitignore @@ -210,7 +210,11 @@ pyrightconfig.json [Ll]ib [Ll]ib64 [Ll]ocal -[Ss]cripts +# [Ss]cripts — REMOVED. This unanchored toptal venv entry matched this +# project's own top-level scripts/ directory, which silently kept every +# file under scripts/sql/migrations/ out of the repository. The virtualenv +# case it was meant to cover is already handled by the `.venv` and `venv/` +# rules above, so the entry was pure downside. pyvenv.cfg pip-selfcheck.json diff --git a/api/generation/mappers.py b/api/generation/mappers.py index ed32a8d..a95a0a9 100644 --- a/api/generation/mappers.py +++ b/api/generation/mappers.py @@ -1,10 +1,15 @@ from api.generation.schemas import ( AllocationKeyGenerated, ConsumerGenerated, + CrmDataPreview, Generation, + IncompleteMeter, IterationGenerated, PartialAllocationKeyGenerated, + PreviewBlocker, ) +from core.i18n import translate +from shared.crm_preflight import Preflight from shared.models.crm_models import AllocationKeyModel, ConsumerModel, IterationModel from shared.models.local_models import ( AllocationKeyGeneratedModel, @@ -91,3 +96,41 @@ def to_allocation_key_crm( id_community=allocation_key.id_community, iterations=[to_iteration_crm(i) for i in allocation_key.iterations], ) + + +def to_crm_data_preview(preflight: Preflight, locale: str) -> CrmDataPreview: + """Render a pre-flight verdict for the manager's screen. + + Blocker messages are localised here rather than in the service so that + translation stays at the API edge — the worker reaches the same verdict via + ``Preflight`` and needs no locale at all. + """ + summary = preflight.summary + return CrmDataPreview( + can_generate=preflight.ok, + # The participant count, not the raw meter count: injection-only sites + # contribute production but never receive a share. + meter_count=len(preflight.consumer_eans), + reading_count=summary.total_rows, + first_timestamp=summary.first_timestamp, + last_timestamp=summary.last_timestamp, + total_consumption_kwh=summary.total_consumption_kwh, + total_injection_kwh=summary.total_injection_kwh, + incomplete_meters=[ + IncompleteMeter( + ean=e.ean, + readings=e.distinct_ts, + expected=summary.grid_size, + missing=summary.grid_size - e.distinct_ts, + ) + for e in summary.incomplete + ], + blockers=[ + PreviewBlocker( + error_code=b.error.code, + message=translate(b.error.key, locale=locale), + detail=b.detail, + ) + for b in preflight.blockers + ], + ) diff --git a/api/generation/routes.py b/api/generation/routes.py index 89c1c6f..b56ebad 100644 --- a/api/generation/routes.py +++ b/api/generation/routes.py @@ -1,12 +1,16 @@ import json +from datetime import date from typing import Annotated, Any from fastapi import APIRouter, Body, Depends, File, Form, Query, UploadFile from sqlalchemy.ext.asyncio import AsyncSession from algorithms.registry import AlgorithmMetadata, registry +from api.generation.mappers import to_crm_data_preview from api.generation.schemas import ( AllocationKeyGenerated, + CrmDataPreview, + GenerateFromCrmRequest, GenerateRequest, GenerateResponse, Generation, @@ -95,6 +99,34 @@ async def get_algorithm_inputs(algorithm_name: str): return ApiResponse[LocalizedAlgorithmMetadata](data=data) +# GET (/crm-data-preview) : What the CRM holds for a sharing operation + period +# +# Declared BEFORE `GET /{id}`: FastAPI matches in declaration order, so putting +# this after the integer catch-all would make "/crm-data-preview" try to parse +# as an id and 422. +@generation_routes.get("/crm-data-preview", response_model=ApiResponse[CrmDataPreview]) +@with_default_error(default_error=errors.generation.GET_CRM_PREVIEW) +async def get_crm_data_preview( + local_session: Annotated[AsyncSession, Depends(get_local_session)], + crm_session: Annotated[AsyncSession, Depends(get_crm_session)], + id_sharing_operation: Annotated[int, Query(description="CRM sharing operation id.")], + period_start: Annotated[date, Query(description="First day of the period (inclusive).")], + period_end: Annotated[date, Query(description="Last day of the period (inclusive).")], +): + internal_community_id = current_internal_community_id.get() + if internal_community_id is None: + raise ErrorException(error=errors.auth.UNAUTHORIZED, status_code=401) + service = GenerationService(local_session, crm_session) + preflight = await service.preview_crm_data( + id_sharing_operation=id_sharing_operation, + period_start=period_start, + period_end=period_end, + community_id=internal_community_id, + ) + locale = current_locale.get().split("_")[0] + return ApiResponse[CrmDataPreview](data=to_crm_data_preview(preflight, locale)) + + # GET (/key/{id}) : Key generated @generation_routes.get("/key/{id_key}", response_model=ApiResponse[AllocationKeyGenerated]) @with_default_error(default_error=errors.generation.GET_ALLOCATION_KEY) @@ -172,6 +204,25 @@ async def start_generation( return ApiResponse[GenerateResponse](data=data) +# POST (/from-crm) Generate from CRM meter data +# +# Plain JSON, unlike POST / — with no file part nothing forces multipart. It +# also correctly keeps the default 2 MB body cap rather than the upload one. +@generation_routes.post("/from-crm", response_model=ApiResponse[GenerateResponse]) +@with_default_error(default_error=errors.generation.START_GENERATION) +async def start_generation_from_crm( + body: Annotated[GenerateFromCrmRequest, Body()], + local_session: Annotated[AsyncSession, Depends(get_local_session)], + crm_session: Annotated[AsyncSession, Depends(get_crm_session)], +): + internal_community_id = current_internal_community_id.get() + if internal_community_id is None: + raise ErrorException(error=errors.auth.UNAUTHORIZED, status_code=401) + service = GenerationService(local_session, crm_session) + data = await service.start_generation_from_crm(body, internal_community_id) + return ApiResponse[GenerateResponse](data=data) + + # POST (/save): Save a key @generation_routes.post("/save", response_model=ApiResponse[str]) @with_default_error(default_error=errors.generation.SAVE_KEY) diff --git a/api/generation/schemas.py b/api/generation/schemas.py index b766157..9fed001 100644 --- a/api/generation/schemas.py +++ b/api/generation/schemas.py @@ -1,3 +1,4 @@ +import datetime from typing import Any from pydantic import BaseModel, Field @@ -102,3 +103,66 @@ class LocalizedAlgorithmMetadata(BaseModel): input_schema: dict[str, Any] tags: list[str] timeout_seconds: int | None + + +# --------------------------------------------------------------------------- +# CRM-sourced generation (source = DataSource.CRM) +# --------------------------------------------------------------------------- + + +class GenerateFromCrmRequest(BaseModel): + """Body of ``POST /from-crm``. + + Unlike ``GenerateRequest`` this *is* the FastAPI body model: with no file + part there is nothing forcing multipart, so the request is plain JSON. + + ``injection_name`` is deliberately absent — the production profile is summed + from the meters themselves, which is the whole reason this path is simpler + for the user than uploading a file. + """ + + name: str = Field(..., min_length=1, description="User-facing label for the generation.") + algorithm_name: str = Field(..., description="Algorithm registry key, e.g. 'olagsa'.") + inputs: dict[str, Any] = Field( + ..., + description=( + "Algorithm-specific input parameters; validated against the algorithm's input schema." + ), + ) + id_sharing_operation: int = Field(..., description="CRM sharing operation to read meters from.") + period_start: datetime.date = Field(..., description="First day of the period (inclusive).") + period_end: datetime.date = Field(..., description="Last day of the period (inclusive).") + + +class IncompleteMeter(BaseModel): + """A meter missing part of the period. Zero-filled, not fatal.""" + + ean: str + readings: int = Field(..., description="Distinct timestamps this meter actually has.") + expected: int = Field(..., description="Distinct timestamps across the whole operation.") + missing: int = Field(..., description="expected - readings.") + + +class PreviewBlocker(BaseModel): + """A reason the period cannot be used, already localised.""" + + error_code: int = Field(..., description="Matches the error_code of the eventual 4xx.") + message: str = Field(..., description="Localised, manager-facing explanation.") + detail: str = Field(..., description="Which meters/values triggered it.") + + +class CrmDataPreview(BaseModel): + """What ``GET /crm-data-preview`` shows before the manager commits to a run. + + ``can_generate`` is the single flag the UI binds its submit button to. + """ + + can_generate: bool + meter_count: int = Field(..., description="Meters that drew energy and will be participants.") + reading_count: int + first_timestamp: datetime.datetime | None + last_timestamp: datetime.datetime | None + total_consumption_kwh: float + total_injection_kwh: float + incomplete_meters: list[IncompleteMeter] + blockers: list[PreviewBlocker] diff --git a/api/generation/service.py b/api/generation/service.py index 397eb89..668bca4 100644 --- a/api/generation/service.py +++ b/api/generation/service.py @@ -1,3 +1,4 @@ +import datetime import logging from uuid import uuid4 @@ -14,6 +15,7 @@ from api.generation.repository import GenerationRepository from api.generation.schemas import ( AllocationKeyGenerated, + GenerateFromCrmRequest, GenerateRequest, GenerateResponse, Generation, @@ -29,7 +31,10 @@ from core.middleware.request_limits import UPLOAD_MAX_BODY_BYTES from core.queue.helper import Event, send_event from core.queue.init import get_jetstream -from shared.const import GenerationStatus +from shared import crm_preflight +from shared.const import DataSource, GenerationStatus +from shared.crm_meter_repository import CrmMeterRepository +from shared.crm_preflight import Preflight from shared.crm_repository import CRMRepository from shared.custom_errors import errors from shared.models.local_models import GenerationModel @@ -255,6 +260,128 @@ async def _mark_failed_to_queue(generation_id: int, reason: str) -> None: ) await crm_session.commit() + # ------------------------------------------------------------------ + # CRM-sourced generation + # ------------------------------------------------------------------ + + async def preview_crm_data( + self, + *, + id_sharing_operation: int, + period_start: datetime.date, + period_end: datetime.date, + community_id: int, + ) -> Preflight: + """Aggregate the requested period and classify it, without running anything. + + Also the pre-flight for ``start_generation_from_crm`` — one code path, so + the answer the manager saw and the answer that gates the run cannot drift. + """ + if period_start > period_end: + raise ErrorException(error=errors.generation.INVALID_PERIOD, status_code=422) + + crm_meters = CrmMeterRepository(self.crm_session) + # Explicit tenant check: without it a foreign operation id is + # indistinguishable from an empty period, which is a confusing 422 for a + # legitimate user and a soft information leak for everyone else. + if not await crm_meters.sharing_operation_exists( + id_community=community_id, id_sharing_operation=id_sharing_operation + ): + raise ErrorException( + error=errors.generation.SHARING_OPERATION_NOT_FOUND, status_code=404 + ) + + summary = await crm_meters.summarize( + id_community=community_id, + id_sharing_operation=id_sharing_operation, + period_start=period_start, + period_end=period_end, + ) + return crm_preflight.evaluate(summary) + + async def start_generation_from_crm( + self, req: GenerateFromCrmRequest, community_id: int + ) -> GenerateResponse: + """Queue a generation that reads its input from the CRM. + + Same ordering as the file path minus the upload: validate, commit, then + publish. There is no object to roll back, so the ``_best_effort_delete`` + branches have no counterpart here. + """ + if req.algorithm_name not in registry: + raise ErrorException(error=errors.generation.ALGORITHM_NOT_FOUND, status_code=404) + meta = registry.metadata(req.algorithm_name) + + try: + validated_inputs = meta.input_schema.model_validate(req.inputs) + except ValidationError as e: + logger.info("Invalid inputs for algorithm '%s': %s", req.algorithm_name, e) + raise ErrorException( + error=errors.generation.INVALID_ALGORITHM_INPUTS, status_code=422 + ) from e + + # Re-run the pre-flight rather than trusting whatever the client saw: + # the preview may be minutes old, and an import can have landed since. + preflight = await self.preview_crm_data( + id_sharing_operation=req.id_sharing_operation, + period_start=req.period_start, + period_end=req.period_end, + community_id=community_id, + ) + if preflight.blockers: + first = preflight.blockers[0] + logger.info( + "CRM generation refused for community %d op %d: %s", + community_id, + req.id_sharing_operation, + first.detail, + ) + raise ErrorException(error=first.error, status_code=422) + + model = GenerationModel( + name=req.name, + id_community=community_id, + source=DataSource.CRM, + id_sharing_operation=req.id_sharing_operation, + period_start=req.period_start, + period_end=req.period_end, + algorithm_name=meta.name, + algorithm_version=meta.version, + inputs=validated_inputs.model_dump(mode="json"), + status=GenerationStatus.PENDING, + data_warnings=preflight.warnings, + ) + await self.repository.create_generation(model) + await self.local_session.commit() + generation_id = model.id + app_metrics.generations_created.add(1, {"algorithm": meta.name}) + await self.audit_log_service.log( + AuditLogInput( + action=AuditActions.GENERATION_CREATED, + entity_type="generation", + entity_id=str(generation_id), + payload={ + "name": req.name, + "algorithm_name": meta.name, + "algorithm_version": meta.version, + "source": DataSource.CRM.name, + "id_sharing_operation": req.id_sharing_operation, + "period_start": req.period_start.isoformat(), + "period_end": req.period_end.isoformat(), + }, + ) + ) + + event = Event(type="generation.requested", data={"generation_id": generation_id}) + try: + await send_event(get_jetstream(), meta.queue, event) + except Exception as exc: + logger.exception("Failed to publish generation %d to %s", generation_id, meta.queue) + await self._mark_failed_to_queue(generation_id, str(exc)) + raise ErrorException(error=errors.generation.START_GENERATION, status_code=500) from exc + + return GenerateResponse(id=generation_id, status=GenerationStatus.PENDING) + async def save_key(self, saved_key: SaveKey): # Retrieve it in this database key = await self.repository.get_allocation_key(saved_key.id_key) diff --git a/core/config.py b/core/config.py index ae650af..a8a556a 100644 --- a/core/config.py +++ b/core/config.py @@ -61,6 +61,14 @@ class Settings(BaseSettings): # MinIO ignores region but botocore still requires it to sign requests. STORAGE_REGION: str = "us-east-1" + # ---- Realtime (Redis pub/sub) ---- + # Fire-and-forget SSE hints. Deliberately NOT in validate_env_config's + # required set (contrast NATS_URL below): realtime is optional by design, and + # making it mandatory would turn a broker outage into a boot failure. An + # empty URL makes core.realtime.emit() a silent no-op. + REALTIME_ENABLED: bool = False + REALTIME_REDIS_URL: str = "" + # ---- CORS ---- ALLOW_ORIGIN: str = "*" diff --git a/core/middleware/request_limits.py b/core/middleware/request_limits.py index 7fc7310..747f988 100644 --- a/core/middleware/request_limits.py +++ b/core/middleware/request_limits.py @@ -38,15 +38,18 @@ # 2 MB — generous for a JSON API, covers PDF upload metadata but blocks abuse MAX_BODY_BYTES = 2 * 1024 * 1024 -# Larger cap for explicit upload endpoints (POST /generation/ multipart). +# Larger cap for explicit upload endpoints (the generation multipart upload). # 50 MB covers realistic energy datasets (hourly * 365 days * hundreds of # consumers) loaded fully into memory in the route handler. Bump this only # after moving the upload to a streaming put_object. UPLOAD_MAX_BODY_BYTES = 50 * 1024 * 1024 -# Path + method pairs that get the larger upload cap. Match on POST only — -# GET/DELETE on the same path keep the default. -_UPLOAD_ROUTES: tuple[tuple[str, str], ...] = (("POST", "/generation/"),) +# Path + method pairs that get the larger upload cap. These are matched against +# the path THIS APP sees, not the public gateway path: the generation upload is +# POST / here (generation_routes is mounted without a prefix in main.py, and the +# /generation prefix is added by KrakenD). Match on POST only — GET/DELETE on +# the same path keep the default. +_UPLOAD_ROUTES: tuple[tuple[str, str], ...] = (("POST", "/"),) # 30 seconds — covers complex DB queries / PDF generation TIMEOUT_SECONDS = 30 diff --git a/core/realtime/__init__.py b/core/realtime/__init__.py new file mode 100644 index 0000000..5606326 --- /dev/null +++ b/core/realtime/__init__.py @@ -0,0 +1,54 @@ +"""Shared realtime publisher — BYTE-IDENTICAL across every producing service. + +Copied verbatim into billing, administrative-document, news-board, +allocation-key-generation and simulation-key, mirroring the ``core/notifications`` +convention. ``scripts/check-realtime-parity.sh`` at the monorepo root is the gate: +make the change in the reference service (news-board) and copy it out, never edit +one copy. + +Consumed by crm-backend's realtime hub (``src/shared/realtime/``) and delivered +to browsers over SSE. Fire-and-forget by contract: if the recipient has no stream +open the event is dropped, which is correct — every event is a hint, and the +client refetches authoritative state through the API gateway. + +Usage, always AFTER the owning transaction commits:: + + from core.realtime import CommunityAudience, Tier, emit + + await session.commit() + await emit( + topic="generation.finished", + audience=CommunityAudience(community_id=cid, tier=Tier.MANAGER), + resource=("generation", generation_id), + scope_community_id=cid, + hint={"status": "success"}, + ) +""" + +from .bus import close, emit, log_realtime_state +from .channels import ( + Audience, + CommunityAudience, + Tier, + UserAudience, + UsersAudience, + community_channel, + user_channel, +) +from .envelope import MAX_ENVELOPE_BYTES, TOPICS, build_envelope + +__all__ = [ + "MAX_ENVELOPE_BYTES", + "TOPICS", + "Audience", + "CommunityAudience", + "Tier", + "UserAudience", + "UsersAudience", + "build_envelope", + "close", + "community_channel", + "emit", + "log_realtime_state", + "user_channel", +] diff --git a/core/realtime/bus.py b/core/realtime/bus.py new file mode 100644 index 0000000..8abecc5 --- /dev/null +++ b/core/realtime/bus.py @@ -0,0 +1,154 @@ +"""Fire-and-forget realtime publisher. + +One lazily-created ``redis.asyncio`` client per process. Every failure mode — +no configuration, unreachable broker, hung broker, malformed envelope — is a +silent no-op, because the *only* thing lost is UI freshness: crm-backend's +clients refetch authoritative state on every reconnect, and every poller in the +SPA keeps running (slower) as a durability backstop. +""" + +import asyncio +import json +import logging +from collections.abc import Mapping + +import redis.asyncio as redis + +from core.config import settings + +from .channels import Audience +from .envelope import build_envelope + +logger = logging.getLogger(__name__) + +_client: redis.Redis | None = None + +#: Publishing must never add latency to a request or a worker tick. A wedged +#: broker is bounded here rather than by the socket, because a *connected but +#: hung* Redis would otherwise await forever. +_PUBLISH_TIMEOUT_SECONDS = 1.0 + + +def _redacted_url() -> str: + """The DSN with its password removed. NEVER log the raw value. + + ``REALTIME_REDIS_URL`` is composed from ``REDIS_PASSWORD`` in + docker-compose, so it carries a live secret in userinfo position. + """ + url = settings.REALTIME_REDIS_URL + scheme, sep, rest = url.partition("://") + if not sep or "@" not in rest: + return url + return f"{scheme}://***@{rest.rpartition('@')[2]}" + + +def log_realtime_state(component: str) -> None: + """Announce this process's realtime publishing state, once, at startup. + + *** CALL THIS AFTER configure_logging(). NEVER at module import time. *** + + This module is imported long before logging is configured: ``worker/main.py`` + imports the dispatcher, which reaches ``worker/persistence.py``, which imports + this file — all in the import block at the top — while ``configure_logging()`` + runs inside ``async def main()``. And ``core/logging.py`` opens with + ``root.handlers.clear()``. So at import time the root logger has no handlers, + Python falls back to ``logging.lastResort`` at WARNING, and an INFO line here + is dropped with no trace whatsoever — reproducing the exact silence this + function exists to break. Do not "simplify" it into the module body. + + Why it exists: four worker containers once ran images built before this + package existed. ``core/realtime/`` was absent and so were the emit call + sites, so three topics were published by nothing at all — for three days, + while their environment variables looked perfectly correct, because + ``--force-recreate`` rebuilds the container from the EXISTING image. The + absence of this line is the cheapest signal that an image predates the + feature. ``scripts/check-realtime-images.sh`` is the automatable form of the + same check, and ``scripts/check-realtime-parity.sh`` cannot see it at all — + it compares source trees, which were green throughout. + """ + if not settings.REALTIME_ENABLED: + logger.info("Realtime disabled — %s publishes nothing", component) + return + if not settings.REALTIME_REDIS_URL: + # Enabled but unconfigured is a misconfiguration, not a deployment choice. + logger.warning( + "Realtime ENABLED but REALTIME_REDIS_URL is empty — %s publishes nothing", component + ) + return + logger.info("Realtime publisher ready — %s publishing to %s", component, _redacted_url()) + + +def _get_client() -> redis.Redis: + global _client + if _client is None: + _client = redis.from_url( + settings.REALTIME_REDIS_URL, + socket_connect_timeout=1, + socket_timeout=1, + health_check_interval=30, + decode_responses=False, + ) + return _client + + +async def emit( + *, + topic: str, + audience: Audience, + resource: tuple[str, str | int], + hint: Mapping[str, str | int | float | bool | None] | None = None, + scope_community_id: int | None = None, +) -> None: + """Publish a realtime hint. NEVER raises. + + *** CALL THIS AFTER THE COMMIT. NEVER inside ``begin_nested()``, and never + inside the ``try`` that owns the business write. *** + + Publishing pre-commit does not merely lose an event — it tells the browser to + refetch and read PRE-COMMIT state, and because the transport is + fire-and-forget there is NO second event, ever. The result is a permanently + stale UI behind a 200, with no error anywhere. That is the same silhouette as + the sweep-commit-ordering and notification-savepoint traps. + + Note the deliberate asymmetry with ``core.notifications.service.publish()``, + which MUST run inside the caller's transaction because it writes rows. These + two have opposite requirements. Do not "unify" them. + + Do not fire this as a bare ``asyncio.create_task`` either: in a worker that + finishes immediately the task is orphaned (the event is lost anyway) and may + log after teardown. + """ + if not settings.REALTIME_ENABLED or not settings.REALTIME_REDIS_URL: + return # no-op: the default everywhere realtime is not deployed + + try: + envelope = build_envelope( + topic=topic, + resource=resource, + hint=hint, + scope_community_id=scope_community_id, + ) + if envelope is None: + logger.warning("realtime: envelope rejected locally topic=%s", topic) + return + + body = json.dumps(envelope, separators=(",", ":")) + client = _get_client() + async with asyncio.timeout(_PUBLISH_TIMEOUT_SECONDS): + for channel in audience.channels(): + await client.publish(channel, body) + # Blanket by contract: nothing this function can hit is worth propagating + # into a caller that has already committed. + except Exception: + logger.warning("realtime: emit failed topic=%s", topic, exc_info=True) + + +async def close() -> None: + """Release the client. For worker shutdown and test teardown.""" + global _client + if _client is not None: + try: + await _client.aclose() + except Exception: + logger.warning("realtime: client close failed", exc_info=True) + _client = None diff --git a/core/realtime/channels.py b/core/realtime/channels.py new file mode 100644 index 0000000..2144039 --- /dev/null +++ b/core/realtime/channels.py @@ -0,0 +1,89 @@ +"""THE ONLY PLACE A REALTIME CHANNEL STRING IS BUILT (Python side). + +This is a security control, not a style rule. A producer that accidentally +publishes a per-user thing onto a community tier is a cross-tenant leak, and +``tier`` below is a required argument with no default precisely so that mistake +cannot be made by omission. A test greps each service for the literal +``notify:v1:`` outside this module and fails on a hit. + +Grammar (fixed arity, so Redis 6 ACLs can later be granted per prefix without a +redesign) — byte-identical to +``crm-backend/src/shared/realtime/realtime.channels.ts``:: + + notify:v1:u:{internal_app_user_id} + notify:v1:c:{internal_community_id}:{MEMBER|MANAGER} + +Ids are the INTERNAL integer keys (``app_user.id``, ``community.id``) — what +every producer here already holds, and what ``notification.id_user`` is. They are +NOT Keycloak subs or org uuids. + +The community family is what lets a worker with NO user attribution at all (the +generation and simulation jobs carry only ``id_community``) address exactly the +right audience with zero database lookups. Its safety comes from the subscribe +side: crm-backend only ever subscribes a connection to tiers the ticket mint +proved the user holds, from gateway-verified claims. +""" + +from collections.abc import Iterable, Sequence +from dataclasses import dataclass +from enum import Enum + +_PREFIX = "notify:v1" + + +class Tier(str, Enum): + """Channel tiers. Deliberately coarser than a role: there is no ADMIN bucket.""" + + #: Everyone in the community, managers included. + MEMBER = "MEMBER" + #: ADMIN and MANAGER only. + MANAGER = "MANAGER" + + +def user_channel(internal_user_id: int) -> str: + """Channel for one user, in every community and outside all of them.""" + return f"{_PREFIX}:u:{internal_user_id}" + + +def community_channel(internal_community_id: int, tier: Tier) -> str: + """Channel for one tier of one community.""" + return f"{_PREFIX}:c:{internal_community_id}:{tier.value}" + + +@dataclass(frozen=True) +class UserAudience: + """One recipient, addressed by internal ``app_user.id``.""" + + user_id: int + + def channels(self) -> Sequence[str]: + return (user_channel(self.user_id),) + + +@dataclass(frozen=True) +class UsersAudience: + """An explicit set of recipients. Duplicates are collapsed.""" + + user_ids: Iterable[int] + + def channels(self) -> Sequence[str]: + return tuple(user_channel(uid) for uid in dict.fromkeys(self.user_ids)) + + +@dataclass(frozen=True) +class CommunityAudience: + """One tier of one community. + + ``Tier.MANAGER`` reaches ADMIN and MANAGER only; ``Tier.MEMBER`` reaches + everyone in the community, managers included — a manager's connection + subscribes to both tiers, so "everyone" is one publish, not two. + """ + + community_id: int + tier: Tier + + def channels(self) -> Sequence[str]: + return (community_channel(self.community_id, self.tier),) + + +Audience = UserAudience | UsersAudience | CommunityAudience diff --git a/core/realtime/envelope.py b/core/realtime/envelope.py new file mode 100644 index 0000000..a2db705 --- /dev/null +++ b/core/realtime/envelope.py @@ -0,0 +1,92 @@ +"""Realtime envelope construction and validation. + +An envelope is a HINT — "something about resource X changed" — never data. The +rules below mirror ``crm-backend/src/shared/realtime/realtime.envelope.ts``, +which re-validates everything on the way out to a browser: + +* **No business data.** No names, emails, EANs, amounts, invoice numbers, + storage keys, error messages. +* **No display strings.** Toast text is chosen client-side from ``topic`` + + ``hint["status"]`` against the i18n bundle. This is a security control: a + compromised publisher gets a nuisance channel, never a text-injection channel + into every open browser. +* **No recipient field.** The channel already says who. A recipient in the body + invites a subscriber-side "is this for me?" check — authorization on the wrong + leg. + +``ref.id`` is permitted: any authorized reader can already see it, and the +client needs it to decide *which* row to refetch. +""" + +import json +import secrets +from collections.abc import Mapping +from datetime import UTC, datetime +from typing import Any, Final + +#: Hard ceiling on a serialized envelope, in BYTES (not characters). +MAX_ENVELOPE_BYTES: Final[int] = 1024 + +#: The topic registry. Mirrors crm-backend's realtime.topics.ts; an unknown topic +#: is dropped by the hub rather than forwarded, so publishing one is a silent +#: no-op that is much easier to find here. +TOPICS: Final[frozenset[str]] = frozenset( + { + "notification.created", + "generation.finished", + "simulation.finished", + "billing_run.finished", + "session.revoked", + } +) + +_SCALARS = (str, int, float, bool) + + +def _is_scalar(value: Any) -> bool: + # bool is a subclass of int, so it is already covered; None is allowed. + return value is None or isinstance(value, _SCALARS) + + +def build_envelope( + *, + topic: str, + resource: tuple[str, str | int], + hint: Mapping[str, str | int | float | bool | None] | None = None, + scope_community_id: int | None = None, +) -> dict[str, Any] | None: + """Build a valid envelope, or ``None`` if the input violates the contract. + + Returns ``None`` rather than raising: every caller is a fire-and-forget side + effect that must never affect a business write, so a malformed hint has to + degrade to "no event", not to an exception travelling up through a commit + path. + """ + if topic not in TOPICS: + return None + + kind, ref_id = resource + if not kind or ref_id is None: + return None + + flat: dict[str, Any] = {} + for key, value in (hint or {}).items(): + if not _is_scalar(value): + return None + flat[str(key)] = value + + envelope: dict[str, Any] = { + "v": 1, + # A client-side dedupe key. Not sortable on purpose: there is no replay, + # so ordering buys nothing. + "id": secrets.token_hex(8), + "topic": topic, + "at": datetime.now(UTC).isoformat().replace("+00:00", "Z"), + "scope": {"community_id": scope_community_id}, + "ref": {"kind": str(kind), "id": str(ref_id)}, + "hint": flat, + } + + if len(json.dumps(envelope, separators=(",", ":")).encode("utf-8")) > MAX_ENVELOPE_BYTES: + return None + return envelope diff --git a/locales/de.json b/locales/de.json index b9755cd..23f8f63 100644 --- a/locales/de.json +++ b/locales/de.json @@ -27,7 +27,15 @@ "INVALID_ALGORITHM_INPUTS": "Ungültige Algorithmus-Eingaben (Code: 2012)", "STORAGE_UPLOAD_FAILED": "Hochladen der Quelldatei fehlgeschlagen (Code: 2013)", "INVALID_FILE": "Ungültige oder leere Datei (Code: 2014)", - "FILE_TOO_LARGE": "Die Datei überschreitet die maximal zulässige Größe (Code: 2015)" + "FILE_TOO_LARGE": "Die Datei überschreitet die maximal zulässige Größe (Code: 2015)", + "INVALID_PERIOD": "Das Startdatum muss vor oder auf dem Enddatum liegen (Code: 2016)", + "SHARING_OPERATION_NOT_FOUND": "Sharing-Vorgang nicht gefunden (Code: 2017)", + "CRM_NO_DATA": "Keine Messdaten für diesen Sharing-Vorgang in diesem Zeitraum (Code: 2018)", + "CRM_DUPLICATE_READINGS": "Doppelte Messwerte erkannt: die Daten wurden vermutlich zweimal importiert (Code: 2019)", + "CRM_NO_INJECTION": "In diesem Zeitraum wurde nichts eingespeist, es gibt also nichts zu teilen (Code: 2020)", + "CRM_NO_CONSUMERS": "Kein Zähler hat in diesem Zeitraum Energie bezogen (Code: 2021)", + "CRM_RANGE_TOO_LARGE": "Der gewählte Zeitraum enthält zu viele Messwerte, bitte kürzen Sie ihn (Code: 2022)", + "GET_CRM_PREVIEW": "Die Messdaten konnten nicht geprüft werden (Code: 2023)" } }, "ALGORITHMS": { diff --git a/locales/en.json b/locales/en.json index c096c5d..831114f 100644 --- a/locales/en.json +++ b/locales/en.json @@ -27,7 +27,15 @@ "INVALID_ALGORITHM_INPUTS": "Invalid algorithm inputs (code: 2012)", "STORAGE_UPLOAD_FAILED": "Unable to upload the source file (code: 2013)", "INVALID_FILE": "Invalid or empty file (code: 2014)", - "FILE_TOO_LARGE": "The file exceeds the maximum allowed size (code: 2015)" + "FILE_TOO_LARGE": "The file exceeds the maximum allowed size (code: 2015)", + "INVALID_PERIOD": "The start date must be on or before the end date (code: 2016)", + "SHARING_OPERATION_NOT_FOUND": "Sharing operation not found (code: 2017)", + "CRM_NO_DATA": "No metering data for this sharing operation over this period (code: 2018)", + "CRM_DUPLICATE_READINGS": "Duplicate readings detected: the data was most likely imported twice (code: 2019)", + "CRM_NO_INJECTION": "No energy was injected over this period, so there is nothing to share (code: 2020)", + "CRM_NO_CONSUMERS": "No meter drew any energy over this period (code: 2021)", + "CRM_RANGE_TOO_LARGE": "The selected period holds too many readings, please shorten it (code: 2022)", + "GET_CRM_PREVIEW": "The metering data could not be checked (code: 2023)" } }, "ALGORITHMS": { diff --git a/locales/fr.json b/locales/fr.json index 8589d25..04a215f 100644 --- a/locales/fr.json +++ b/locales/fr.json @@ -27,7 +27,15 @@ "INVALID_ALGORITHM_INPUTS": "Entrées d'algorithme invalides (code: 2012)", "STORAGE_UPLOAD_FAILED": "Impossible d'envoyer le fichier source (code: 2013)", "INVALID_FILE": "Fichier invalide ou vide (code: 2014)", - "FILE_TOO_LARGE": "Le fichier dépasse la taille maximale autorisée (code: 2015)" + "FILE_TOO_LARGE": "Le fichier dépasse la taille maximale autorisée (code: 2015)", + "INVALID_PERIOD": "La date de début doit précéder la date de fin (code: 2016)", + "SHARING_OPERATION_NOT_FOUND": "Opération de partage introuvable (code: 2017)", + "CRM_NO_DATA": "Aucune donnée de comptage pour cette opération de partage sur cette période (code: 2018)", + "CRM_DUPLICATE_READINGS": "Des relevés en double ont été détectés : les données ont probablement été importées deux fois (code: 2019)", + "CRM_NO_INJECTION": "Aucune injection sur cette période : il n'y a rien à partager (code: 2020)", + "CRM_NO_CONSUMERS": "Aucun compteur n'a consommé d'énergie sur cette période (code: 2021)", + "CRM_RANGE_TOO_LARGE": "La période sélectionnée contient trop de relevés, veuillez la raccourcir (code: 2022)", + "GET_CRM_PREVIEW": "Impossible de vérifier les données de comptage (code: 2023)" } }, "ALGORITHMS": { diff --git a/locales/nl.json b/locales/nl.json index a16e501..3518934 100644 --- a/locales/nl.json +++ b/locales/nl.json @@ -27,7 +27,15 @@ "INVALID_ALGORITHM_INPUTS": "Ongeldige algoritme-invoer (code: 2012)", "STORAGE_UPLOAD_FAILED": "Uploaden van het bronbestand is mislukt (code: 2013)", "INVALID_FILE": "Ongeldig of leeg bestand (code: 2014)", - "FILE_TOO_LARGE": "Het bestand overschrijdt de maximaal toegestane grootte (code: 2015)" + "FILE_TOO_LARGE": "Het bestand overschrijdt de maximaal toegestane grootte (code: 2015)", + "INVALID_PERIOD": "De begindatum moet op of vóór de einddatum liggen (code: 2016)", + "SHARING_OPERATION_NOT_FOUND": "Deeloperatie niet gevonden (code: 2017)", + "CRM_NO_DATA": "Geen meetgegevens voor deze deeloperatie in deze periode (code: 2018)", + "CRM_DUPLICATE_READINGS": "Dubbele meetwaarden gevonden: de gegevens zijn waarschijnlijk twee keer geïmporteerd (code: 2019)", + "CRM_NO_INJECTION": "Er is in deze periode niets geïnjecteerd, dus er valt niets te delen (code: 2020)", + "CRM_NO_CONSUMERS": "Geen enkele meter heeft in deze periode energie afgenomen (code: 2021)", + "CRM_RANGE_TOO_LARGE": "De geselecteerde periode bevat te veel meetwaarden, kies een kortere periode (code: 2022)", + "GET_CRM_PREVIEW": "De meetgegevens konden niet worden gecontroleerd (code: 2023)" } }, "ALGORITHMS": { diff --git a/main.py b/main.py index 3f35488..ebffd8e 100644 --- a/main.py +++ b/main.py @@ -17,6 +17,7 @@ from core.middleware.request_limits import RequestLimitsMiddleware from core.middleware.set_auth_context import GatewayScopeMiddleware from core.queue.init import close_nats, init_nats +from core.realtime import log_realtime_state from core.tracing import enrich_span, setup_tracer_provider configure_logging() @@ -25,6 +26,9 @@ @asynccontextmanager async def lifespan(app: FastAPI): + # Absence of this line means the image predates the realtime feature — + # see core/realtime/bus.py. Must come after configure_logging(). + log_realtime_state("allocation-key-generation api") setup_tracer_provider() await init_nats() autodiscover() diff --git a/requirements/base.txt b/requirements/base.txt index 56b4bad..7714dfa 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -33,4 +33,9 @@ opentelemetry-semantic-conventions==0.60b1 opentelemetry-util-http==0.60b1 reportlab==4.4.10 -aiosmtplib>=3.0.0 \ No newline at end of file +aiosmtplib>=3.0.0 + +# Realtime SSE hints (fire-and-forget pub/sub consumed by crm-backend's hub). +# redis.asyncio, not the unmaintained aioredis — it is aioredis's successor and +# ships in the same package. +redis==5.2.1 diff --git a/scripts/export_openapi.py b/scripts/export_openapi.py index 38d1f4a..64392ec 100644 --- a/scripts/export_openapi.py +++ b/scripts/export_openapi.py @@ -2,13 +2,14 @@ Usage: python scripts/export_openapi.py """ + import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -from main import app # noqa: E402 +from main import app def export(output_path: Path) -> None: diff --git a/scripts/sql/migrations/002_file_storage_key.sql b/scripts/sql/migrations/002_file_storage_key.sql new file mode 100644 index 0000000..fd6afa9 --- /dev/null +++ b/scripts/sql/migrations/002_file_storage_key.sql @@ -0,0 +1,17 @@ +-- Migration 002: rename generation.file_url to file_storage_key. +-- +-- Before this migration, file_url stored an externally hosted URL passed by +-- the client. After it, the column stores an opaque S3 object key inside +-- STORAGE_BUCKET (MinIO). The service uploads the file at creation time and +-- the worker deletes it on terminal outcomes. Widening to VARCHAR(512) gives +-- headroom for the longer keys (allocations///). + +BEGIN; + +ALTER TABLE generation RENAME COLUMN file_url TO file_storage_key; +ALTER TABLE generation ALTER COLUMN file_storage_key TYPE VARCHAR(512); + +INSERT INTO schema_version (version, description) +VALUES (2, 'Rename generation.file_url to file_storage_key, widen to 512'); + +COMMIT; diff --git a/scripts/sql/migrations/003_crm_source.sql b/scripts/sql/migrations/003_crm_source.sql new file mode 100644 index 0000000..7b3523b --- /dev/null +++ b/scripts/sql/migrations/003_crm_source.sql @@ -0,0 +1,50 @@ +-- Migration 003: allow a generation to source its input from the CRM database +-- instead of an uploaded file. +-- +-- Until now every generation carried an uploaded CSV/XLSX, so file_storage_key, +-- file_name and injection_name were all NOT NULL. A CRM-sourced run has none of +-- them: it names a sharing operation and a date range, and the worker reads +-- meter_consumption directly. +-- +-- The three file columns therefore become nullable, and a CHECK constraint +-- takes over the job they were doing — each source shape must be fully +-- populated, so a half-specified row is still impossible. +-- +-- data_warnings holds non-blocking findings from the pre-flight (currently: +-- meters with gaps, which are zero-filled). It is persisted rather than only +-- shown before launch, because a warning the manager sees once and never again +-- is not really a warning. + +BEGIN; + +ALTER TABLE generation + ALTER COLUMN file_storage_key DROP NOT NULL, + ALTER COLUMN file_name DROP NOT NULL, + ALTER COLUMN injection_name DROP NOT NULL; + +ALTER TABLE generation + ADD COLUMN IF NOT EXISTS source SMALLINT NOT NULL DEFAULT 1, + ADD COLUMN IF NOT EXISTS id_sharing_operation INTEGER NULL, + ADD COLUMN IF NOT EXISTS period_start DATE NULL, + ADD COLUMN IF NOT EXISTS period_end DATE NULL, + ADD COLUMN IF NOT EXISTS data_warnings JSONB NULL; + +-- 1=FILE, 2=CRM. Existing rows keep the DEFAULT 1 and satisfy the FILE branch. +ALTER TABLE generation DROP CONSTRAINT IF EXISTS ck_generation_source; +ALTER TABLE generation ADD CONSTRAINT ck_generation_source CHECK ( + (source = 1 + AND file_storage_key IS NOT NULL + AND file_name IS NOT NULL + AND injection_name IS NOT NULL) + OR (source = 2 + AND id_sharing_operation IS NOT NULL + AND period_start IS NOT NULL + AND period_end IS NOT NULL + AND period_start <= period_end) +); + +INSERT INTO schema_version (version, description) +VALUES (3, 'Allow CRM-sourced generations (source, sharing operation, period)') +ON CONFLICT DO NOTHING; + +COMMIT; diff --git a/scripts/sql/schema.sql b/scripts/sql/schema.sql index d504631..8858b8f 100644 --- a/scripts/sql/schema.sql +++ b/scripts/sql/schema.sql @@ -32,6 +32,10 @@ INSERT into schema_version (version, description) VALUES( 2, 'Rename generation.file_url to file_storage_key, widen to 512' ) ON CONFLICT DO NOTHING; +INSERT into schema_version (version, description) VALUES( + 3, 'Allow CRM-sourced generations (source, sharing operation, period)' +) ON CONFLICT DO NOTHING; + -- ---- generation ------------------------------------------------------------ -- One row per allocation-key generation request. Holds the source file -- reference, the chosen algorithm + its input payload snapshot, and the @@ -42,12 +46,24 @@ CREATE TABLE IF NOT EXISTS generation ( name VARCHAR(255) NOT NULL, id_community INTEGER NOT NULL, - -- Source data - -- file_storage_key is the object key inside STORAGE_BUCKET (MinIO). The - -- service uploads on creation; the worker deletes on terminal outcomes. - file_storage_key VARCHAR(512) NOT NULL, - file_name VARCHAR(255) NOT NULL, - injection_name VARCHAR(255) NOT NULL, + -- Source data: 1=FILE (uploaded CSV/XLSX), 2=CRM (meter_consumption). + -- Exactly one of the two column groups below is populated; the + -- ck_generation_source CHECK is what enforces that, now that the file + -- columns can no longer be NOT NULL. + source SMALLINT NOT NULL DEFAULT 1, + + -- FILE only. file_storage_key is the object key inside STORAGE_BUCKET + -- (MinIO). The service uploads on creation; the worker deletes on terminal + -- outcomes. injection_name names the production column inside the file. + file_storage_key VARCHAR(512) NULL, + file_name VARCHAR(255) NULL, + injection_name VARCHAR(255) NULL, + + -- CRM only. The sharing operation and the inclusive local date range read + -- from meter_consumption. No FK: the CRM lives in a separate database. + id_sharing_operation INTEGER NULL, + period_start DATE NULL, + period_end DATE NULL, -- Algorithm snapshot (keyed to algorithms.registry) algorithm_name VARCHAR(64) NOT NULL, @@ -58,8 +74,24 @@ CREATE TABLE IF NOT EXISTS generation ( status INTEGER NOT NULL DEFAULT 0, error_message TEXT NULL, + -- Non-blocking findings from the CRM pre-flight (meters with gaps, which + -- are zero-filled). Persisted so the warning outlives the preview screen. + data_warnings JSONB NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT ck_generation_source CHECK ( + (source = 1 + AND file_storage_key IS NOT NULL + AND file_name IS NOT NULL + AND injection_name IS NOT NULL) + OR (source = 2 + AND id_sharing_operation IS NOT NULL + AND period_start IS NOT NULL + AND period_end IS NOT NULL + AND period_start <= period_end) + ) ); CREATE INDEX IF NOT EXISTS idx_generation_id_community ON generation (id_community); diff --git a/shared/const.py b/shared/const.py index 4aa2c9f..861eee6 100644 --- a/shared/const.py +++ b/shared/const.py @@ -10,3 +10,15 @@ class GenerationStatus(IntEnum): class FeatureName(StrEnum): ALGORITHM = "algorithm" # SIMULATION = "simulation" # future + + +class DataSource(IntEnum): + """Where a generation's input timeseries comes from. + + FILE is the historical path (a CSV/XLSX uploaded by the manager). CRM + reads the same numbers straight out of the core database's + ``meter_consumption`` for one sharing operation over a date range. + """ + + FILE = 1 + CRM = 2 diff --git a/shared/crm_meter_repository.py b/shared/crm_meter_repository.py new file mode 100644 index 0000000..0dcbaa9 --- /dev/null +++ b/shared/crm_meter_repository.py @@ -0,0 +1,279 @@ +"""Read-only access to the CRM core's metering tables. + +This is the single place coupled to the CRM ``meter_consumption`` layout, in +the same spirit as ``billing/ports/crm_core_sqlalchemy.py``. Every statement is +SELECT-only and runs on a CRM ``AsyncSession``. + +Two deliberate constraints on this module: + +* **No pandas.** The API imports it for the pre-flight/preview and + ``requirements/api.txt`` carries no numpy/pandas. The pivot into a DataFrame + lives in ``shared/crm_timeseries.py``, which only the worker imports. +* **No fastapi.** The worker imports it too, and ``Dockerfile.worker`` + installs no HTTP stack. + +Community scope is passed **explicitly** on every call rather than read from a +ContextVar: the worker has no request context, and +``core.database.with_community.with_community_scope`` would silently degrade to +``WHERE false`` there. + +Period boundaries are interpreted in Belgian local time, so a month aligns to +local midnights and is DST-safe. Windows are half-open: +``timestamp >= start AND timestamp < end_exclusive``. +""" + +import datetime +from dataclasses import dataclass +from datetime import date +from zoneinfo import ZoneInfo + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +# Metering timestamps are absolute instants (timestamptz); a period expressed +# as local dates is bounded at Belgian local midnights. Mirrors +# crm-backend's CONSUMPTION_TIMEZONE. +_METERING_TZ = ZoneInfo("Europe/Brussels") + + +def period_bounds( + period_start: date, period_end: date +) -> tuple[datetime.datetime, datetime.datetime]: + """Half-open instant bounds ``[start, end_exclusive)`` for an inclusive date range.""" + start = datetime.datetime.combine(period_start, datetime.time.min, tzinfo=_METERING_TZ) + end_exclusive = datetime.datetime.combine( + period_end + datetime.timedelta(days=1), datetime.time.min, tzinfo=_METERING_TZ + ) + return start, end_exclusive + + +@dataclass(frozen=True) +class EanCoverage: + """Per-EAN aggregate over the requested period.""" + + ean: str + row_count: int + distinct_ts: int + consumption_kwh: float + injection_kwh: float + + @property + def has_duplicate_rows(self) -> bool: + """True when the same (ean, timestamp) appears more than once. + + ``meter_consumption`` has no unique constraint on (ean, timestamp), so a + workbook imported twice lands twice and silently doubles this meter's + energy. Callers treat this as fatal, not as a warning. + """ + return self.row_count != self.distinct_ts + + @property + def is_consumer(self) -> bool: + """A meter that actually drew energy over the period. + + A pure injection point would otherwise become an all-zero participant + holding a 0% share of the generated key. + """ + return self.consumption_kwh > 0 + + +@dataclass(frozen=True) +class CrmDataSummary: + """Everything the pre-flight and the preview screen need, in two queries.""" + + eans: list[EanCoverage] + # Distinct timestamps across the whole operation — the length of the grid + # every meter is reindexed onto. + grid_size: int + first_timestamp: datetime.datetime | None + last_timestamp: datetime.datetime | None + + @property + def total_rows(self) -> int: + return sum(e.row_count for e in self.eans) + + @property + def total_consumption_kwh(self) -> float: + return sum(e.consumption_kwh for e in self.eans) + + @property + def total_injection_kwh(self) -> float: + return sum(e.injection_kwh for e in self.eans) + + @property + def all_eans(self) -> list[str]: + """Every meter with a reading in the period, consumers and injectors alike.""" + return [e.ean for e in self.eans] + + @property + def consumer_eans(self) -> list[str]: + return [e.ean for e in self.eans if e.is_consumer] + + @property + def duplicate_eans(self) -> list[str]: + return [e.ean for e in self.eans if e.has_duplicate_rows] + + @property + def incomplete(self) -> list[EanCoverage]: + """Meters missing at least one timestamp of the common grid. + + These are allowed through (the gaps are zero-filled) but are reported to + the manager both before the run and on the finished run. + """ + return [e for e in self.eans if e.distinct_ts < self.grid_size] + + +@dataclass(frozen=True) +class ConsumptionRow: + """One metering reading, as consumed by the pivot.""" + + timestamp: datetime.datetime + ean: str + gross: float + inj_gross: float + + +class CrmMeterRepository: + """SELECT-only reader over the CRM metering tables.""" + + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def sharing_operation_exists( + self, *, id_community: int, id_sharing_operation: int + ) -> bool: + """Tenant gate: does this operation belong to the caller's community? + + Checked explicitly so a cross-tenant id yields a clean "not found" + rather than an indistinguishable "no data in this period". + """ + result = await self._session.execute( + text( + """ + SELECT EXISTS ( + SELECT 1 FROM sharing_operation so + WHERE so.id = :op AND so.id_community = :cid + ) AS present + """ + ), + {"cid": id_community, "op": id_sharing_operation}, + ) + return bool(result.scalar()) + + async def summarize( + self, + *, + id_community: int, + id_sharing_operation: int, + period_start: date, + period_end: date, + ) -> CrmDataSummary: + """Aggregate the period without transferring the readings themselves. + + Deliberately cheap: ``RequestLimitsMiddleware.TIMEOUT_SECONDS`` caps + every API request at 30 s, and this runs on the request path. + """ + start, end_exclusive = period_bounds(period_start, period_end) + params = { + "cid": id_community, + "op": id_sharing_operation, + "start": start, + "end_excl": end_exclusive, + } + + per_ean = await self._session.execute( + text( + """ + SELECT mc.ean AS ean, + COUNT(*) AS row_count, + COUNT(DISTINCT mc.timestamp) AS distinct_ts, + COALESCE(SUM(mc.gross), 0) AS consumption_kwh, + COALESCE(SUM(mc.inj_gross), 0) AS injection_kwh + FROM meter_consumption mc + WHERE mc.id_community = :cid + AND mc.id_sharing_operation = :op + AND mc.timestamp >= :start AND mc.timestamp < :end_excl + GROUP BY mc.ean + ORDER BY mc.ean + """ + ), + params, + ) + eans = [ + EanCoverage( + ean=row["ean"], + row_count=int(row["row_count"]), + distinct_ts=int(row["distinct_ts"]), + consumption_kwh=float(row["consumption_kwh"]), + injection_kwh=float(row["injection_kwh"]), + ) + for row in per_ean.mappings() + ] + + grid = await self._session.execute( + text( + """ + SELECT COUNT(DISTINCT mc.timestamp) AS grid_size, + MIN(mc.timestamp) AS first_ts, + MAX(mc.timestamp) AS last_ts + FROM meter_consumption mc + WHERE mc.id_community = :cid + AND mc.id_sharing_operation = :op + AND mc.timestamp >= :start AND mc.timestamp < :end_excl + """ + ), + params, + ) + grid_row = grid.mappings().one() + + return CrmDataSummary( + eans=eans, + grid_size=int(grid_row["grid_size"] or 0), + first_timestamp=grid_row["first_ts"], + last_timestamp=grid_row["last_ts"], + ) + + async def fetch_rows( + self, + *, + id_community: int, + id_sharing_operation: int, + period_start: date, + period_end: date, + ) -> list[ConsumptionRow]: + """The readings themselves, ordered for a stable pivot. + + Only ever called from the worker: a year of quarter-hours across a few + dozen meters is well past what belongs on a 30-second request path. + """ + start, end_exclusive = period_bounds(period_start, period_end) + result = await self._session.execute( + text( + """ + SELECT mc.timestamp AS ts, + mc.ean AS ean, + COALESCE(mc.gross, 0) AS gross, + COALESCE(mc.inj_gross, 0) AS inj_gross + FROM meter_consumption mc + WHERE mc.id_community = :cid + AND mc.id_sharing_operation = :op + AND mc.timestamp >= :start AND mc.timestamp < :end_excl + ORDER BY mc.timestamp, mc.ean + """ + ), + { + "cid": id_community, + "op": id_sharing_operation, + "start": start, + "end_excl": end_exclusive, + }, + ) + return [ + ConsumptionRow( + timestamp=row["ts"], + ean=row["ean"], + gross=float(row["gross"]), + inj_gross=float(row["inj_gross"]), + ) + for row in result.mappings() + ] diff --git a/shared/crm_preflight.py b/shared/crm_preflight.py new file mode 100644 index 0000000..37a579a --- /dev/null +++ b/shared/crm_preflight.py @@ -0,0 +1,130 @@ +"""Decide whether a CRM-sourced period can produce a meaningful allocation key. + +One definition of "blocking", used in three places: the preview endpoint (so the +manager sees the problem before launching), ``POST /from-crm`` (so a stale +preview cannot slip a bad run through), and the worker (which re-reads the data +at execution time and must reach the same verdict). + +Framework-free and pandas-free on purpose — the API, which has neither pandas +nor a request-scoped community, imports this too. +""" + +from dataclasses import dataclass +from typing import Any + +from core.errors.errors import Error +from shared.crm_meter_repository import CrmDataSummary +from shared.custom_errors import errors + +# A year of quarter-hours across ~60 meters is ~2.1 M rows, which the worker +# pivots comfortably. Well past that we would rather refuse than risk the +# worker being OOM-killed mid-run, which reads to the manager as a silent hang. +MAX_READING_ROWS = 5_000_000 + + +@dataclass(frozen=True) +class Blocker: + """A reason this period cannot be used, with the detail the manager needs.""" + + error: Error + detail: str + + +@dataclass(frozen=True) +class Preflight: + summary: CrmDataSummary + consumer_eans: list[str] + blockers: list[Blocker] + warnings: dict[str, Any] | None + + @property + def ok(self) -> bool: + return not self.blockers + + +def evaluate(summary: CrmDataSummary) -> Preflight: + """Classify a period's metering data into blockers and warnings. + + Gaps are deliberately **not** blocking: missing quarters are zero-filled and + reported. Duplicates are, because there is no unique constraint on + ``meter_consumption(ean, timestamp)`` and a repeated import would inflate a + participant's share with no visible symptom. + """ + blockers: list[Blocker] = [] + + if not summary.eans: + # Nothing else can be said about an empty period; return early so the + # manager gets one clear message rather than four derived ones. + return Preflight( + summary=summary, + consumer_eans=[], + blockers=[ + Blocker( + error=errors.generation.CRM_NO_DATA, + detail="no readings for this sharing operation over this period", + ) + ], + warnings=None, + ) + + if summary.total_rows > MAX_READING_ROWS: + blockers.append( + Blocker( + error=errors.generation.CRM_RANGE_TOO_LARGE, + detail=( + f"{summary.total_rows} readings exceed the {MAX_READING_ROWS} limit; " + "choose a shorter period" + ), + ) + ) + + duplicates = summary.duplicate_eans + if duplicates: + blockers.append( + Blocker( + error=errors.generation.CRM_DUPLICATE_READINGS, + detail=( + "the same timestamp appears more than once for meter(s) " + f"{', '.join(duplicates)} — the data was most likely imported twice" + ), + ) + ) + + consumer_eans = summary.consumer_eans + if not consumer_eans: + blockers.append( + Blocker( + error=errors.generation.CRM_NO_CONSUMERS, + detail="no meter drew any energy over this period", + ) + ) + + if summary.total_injection_kwh <= 0: + blockers.append( + Blocker( + error=errors.generation.CRM_NO_INJECTION, + detail="no energy was injected over this period — there is nothing to share", + ) + ) + + incomplete = summary.incomplete + warnings: dict[str, Any] | None = None + if incomplete: + warnings = { + "incomplete_meters": [ + { + "ean": e.ean, + "readings": e.distinct_ts, + "expected": summary.grid_size, + "missing": summary.grid_size - e.distinct_ts, + } + for e in incomplete + ] + } + + return Preflight( + summary=summary, + consumer_eans=consumer_eans, + blockers=blockers, + warnings=warnings, + ) diff --git a/shared/crm_timeseries.py b/shared/crm_timeseries.py new file mode 100644 index 0000000..b96e212 --- /dev/null +++ b/shared/crm_timeseries.py @@ -0,0 +1,89 @@ +"""Pivot CRM metering rows into the wide frame the algorithms already expect. + +The whole point of this module is that it produces a DataFrame **shaped exactly +like a parsed upload** — one column per participant plus a single injection +column — so the existing converter in ``shared/data_loading`` consumes it +unchanged. Nothing downstream needs to know the data came from the database +rather than from a file. + +Imports pandas, and is therefore **worker-only**: ``requirements/api.txt`` +carries no pandas, exactly as for ``shared/data_loading.py``. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import pandas as pd + +from shared.crm_meter_repository import ConsumptionRow + +# Column name standing in for the file-based ``injection_name``. EANs are digit +# strings, so a dunder label cannot collide with a participant column. +INJECTION_COLUMN = "__injection__" + + +class CrmPivotError(ValueError): + """Raised when the fetched rows cannot form a rectangular timeseries.""" + + +def build_dataframe( + rows: Sequence[ConsumptionRow], + consumer_eans: Sequence[str], +) -> pd.DataFrame: + """Build ``[timestamp x consumer EAN] + __injection__`` from raw readings. + + ``consumer_eans`` selects (and orders) the participant columns — meters that + actually drew energy over the period. Injection is summed over **every** EAN + in ``rows``, including pure production sites that are not participants, and + is taken from the same rows as the consumption so the two series can never + disagree in length. + + Timestamps present for some meters but not others are zero-filled; the + caller has already reported those gaps to the manager. + """ + if not rows: + raise CrmPivotError("no readings to pivot") + if not consumer_eans: + raise CrmPivotError("no consumer EANs requested") + if INJECTION_COLUMN in consumer_eans: + raise CrmPivotError(f"{INJECTION_COLUMN!r} is reserved and cannot be a participant") + + frame = pd.DataFrame( + { + "timestamp": [r.timestamp for r in rows], + "ean": [r.ean for r in rows], + "gross": [r.gross for r in rows], + "inj_gross": [r.inj_gross for r in rows], + } + ) + + # Duplicate (ean, timestamp) pairs are refused upstream (they would double a + # participant's energy). Pivoting defensively rather than aggregating means a + # regression surfaces as a loud failure instead of silently inflated volumes. + try: + consumption = frame.pivot(index="timestamp", columns="ean", values="gross") + except ValueError as exc: + raise CrmPivotError(f"duplicate (ean, timestamp) readings: {exc}") from exc + + # A requested participant with no reading at all in the period would be + # silently reindexed into an all-zero column — for simulation that means a + # key participant whose EAN does not exist quietly gets a 0 kWh profile + # instead of an error. Refuse instead; the pre-flight normally catches this + # first and reports it far more helpfully, so reaching here is a backstop. + missing = [ean for ean in consumer_eans if ean not in consumption.columns] + if missing: + raise CrmPivotError(f"no readings for requested participant(s): {missing}") + + # Select the requested participants in order, dropping injection-only meters. + consumption = consumption.reindex(columns=list(consumer_eans)) + # Remaining NaNs are per-timestamp gaps in an otherwise present meter. + consumption = consumption.fillna(0.0).astype(float) + + # One shared production profile, summed across every meter at each instant. + injection = frame.groupby("timestamp")["inj_gross"].sum() + consumption[INJECTION_COLUMN] = injection.reindex(consumption.index).fillna(0.0).astype(float) + + # Chronological order is the only ordering guarantee the algorithms have: + # row t of every column must be contemporaneous. + return consumption.sort_index().reset_index(drop=True) diff --git a/shared/custom_errors.py b/shared/custom_errors.py index 6a4c200..69e66c0 100644 --- a/shared/custom_errors.py +++ b/shared/custom_errors.py @@ -33,6 +33,23 @@ class _GenerationErrors: INVALID_FILE = Error(code=2014, key="ERRORS.GENERATION.INVALID_FILE") FILE_TOO_LARGE = Error(code=2015, key="ERRORS.GENERATION.FILE_TOO_LARGE") + # --- CRM-sourced input (source = DataSource.CRM) ----------------------- + # Raised before a run is queued (pre-flight on POST /from-crm) and again + # in the worker, which re-reads the data at execution time. + INVALID_PERIOD = Error(code=2016, key="ERRORS.GENERATION.INVALID_PERIOD") + SHARING_OPERATION_NOT_FOUND = Error( + code=2017, key="ERRORS.GENERATION.SHARING_OPERATION_NOT_FOUND" + ) + CRM_NO_DATA = Error(code=2018, key="ERRORS.GENERATION.CRM_NO_DATA") + # No unique constraint on meter_consumption(ean, timestamp): a repeated + # import silently doubles a participant's energy, so this is fatal rather + # than a warning. + CRM_DUPLICATE_READINGS = Error(code=2019, key="ERRORS.GENERATION.CRM_DUPLICATE_READINGS") + CRM_NO_INJECTION = Error(code=2020, key="ERRORS.GENERATION.CRM_NO_INJECTION") + CRM_NO_CONSUMERS = Error(code=2021, key="ERRORS.GENERATION.CRM_NO_CONSUMERS") + CRM_RANGE_TOO_LARGE = Error(code=2022, key="ERRORS.GENERATION.CRM_RANGE_TOO_LARGE") + GET_CRM_PREVIEW = Error(code=2023, key="ERRORS.GENERATION.GET_CRM_PREVIEW") + class _Errors: auth = _AuthErrors() diff --git a/shared/models/local_models.py b/shared/models/local_models.py index 2eb295f..4b8d6c4 100644 --- a/shared/models/local_models.py +++ b/shared/models/local_models.py @@ -1,12 +1,12 @@ import datetime from typing import Any -from sqlalchemy import TIMESTAMP, Float, ForeignKey, Integer, String, Text +from sqlalchemy import TIMESTAMP, Date, Float, ForeignKey, Integer, SmallInteger, String, Text from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship from core.database.database import LocalBase -from shared.const import GenerationStatus +from shared.const import DataSource, GenerationStatus class GenerationModel(LocalBase): @@ -19,14 +19,28 @@ class GenerationModel(LocalBase): id_community: Mapped[int] = mapped_column(Integer, nullable=False) # --- Source data --- - # file_storage_key is the object key inside STORAGE_BUCKET (MinIO). The - # service uploads the user-supplied file at creation time; the worker - # deletes it once the row reaches SUCCESS or FAILED. - file_storage_key: Mapped[str] = mapped_column(String(512), nullable=False) - file_name: Mapped[str] = mapped_column(String(255), nullable=False) + # Which of the two column groups below is populated. The DB-level + # ck_generation_source CHECK enforces the pairing; these columns cannot be + # NOT NULL any more because a CRM-sourced run has no file. + source: Mapped[DataSource] = mapped_column( + SmallInteger, nullable=False, default=DataSource.FILE + ) + + # FILE only. file_storage_key is the object key inside STORAGE_BUCKET + # (MinIO). The service uploads the user-supplied file at creation time; the + # worker deletes it once the row reaches SUCCESS or FAILED. + file_storage_key: Mapped[str | None] = mapped_column(String(512), nullable=True) + file_name: Mapped[str | None] = mapped_column(String(255), nullable=True) # Name of the column inside the uploaded file that holds the shared # production profile (the "injection"). Shared across all algorithms. - injection_name: Mapped[str] = mapped_column(String(255), nullable=False) + injection_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + + # CRM only. The sharing operation and the inclusive Brussels-local date + # range read out of meter_consumption. Plain columns, never FKs — the CRM + # is a separate database. + id_sharing_operation: Mapped[int | None] = mapped_column(Integer, nullable=True) + period_start: Mapped[datetime.date | None] = mapped_column(Date, nullable=True) + period_end: Mapped[datetime.date | None] = mapped_column(Date, nullable=True) # --- Algorithm snapshot --- # Keyed to algorithms.registry at creation time; version is snapshotted @@ -44,6 +58,10 @@ class GenerationModel(LocalBase): # Populated by the worker on failure so the UI can surface the cause # without needing to hit logs. Nullable because success rows have none. error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + # Non-blocking findings from the CRM pre-flight — currently the meters that + # had gaps in the period and were zero-filled. Persisted rather than only + # shown before launch, so the manager can still see it on the finished run. + data_warnings: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) created_at: Mapped[datetime.datetime] = mapped_column( TIMESTAMP(timezone=True), diff --git a/tests/api/generation/test_crm_source_routes.py b/tests/api/generation/test_crm_source_routes.py new file mode 100644 index 0000000..28f9d5e --- /dev/null +++ b/tests/api/generation/test_crm_source_routes.py @@ -0,0 +1,385 @@ +"""Integration tests for the CRM-sourced generation routes. + +Full ASGI stack against a real Postgres, same conventions as +test_generation_routes.py: gateway headers rather than dependency overrides, +NATS patched at the import site in the service module, and an active +`algorithm` subscription on every request. + +Nothing here touches MinIO — that is the point of the CRM source. +""" + +import datetime +from unittest.mock import AsyncMock, MagicMock, patch + +from sqlalchemy import select + +from core.database.models import Community +from shared.const import DataSource, GenerationStatus +from shared.custom_errors import errors +from shared.models.local_models import GenerationModel +from tests.factories.meter_factory import ( + create_meter, + create_readings, + create_sharing_operation, +) +from tests.factories.subscription_factory import create_community, create_subscription + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_PERIOD = {"period_start": "2025-02-01", "period_end": "2025-02-28"} + + +def _admin_headers(community: Community) -> dict[str, str]: + return { + "x-user-id": "test|admin", + "x-community-id": community.auth_community_id, + "x-user-role": "ADMIN", + } + + +async def _community_with_subscription(db_session) -> Community: + community = await create_community(db_session) + await create_subscription(db_session, id_community=community.id, is_active=True) + return community + + +async def _operation_with_data( + db_session, + community: Community, + *, + consumer_skip: set[int] | None = None, + duplicate: bool = False, + injection: float = 40.0, +) -> int: + """One consuming meter + one PV meter, four quarter-hours in Feb 2025.""" + op = await create_sharing_operation(db_session, id_community=community.id) + await create_meter(db_session, ean="541448000000000001", id_community=community.id) + await create_meter(db_session, ean="541448000000000002", id_community=community.id) + + await create_readings( + db_session, + ean="541448000000000001", + id_community=community.id, + id_sharing_operation=op, + gross=10.0, + inj_gross=0.0, + skip=consumer_skip, + ) + if duplicate: + # A second import of the same quarters — the exact shape that has no + # unique constraint to stop it in production. + await create_readings( + db_session, + ean="541448000000000001", + id_community=community.id, + id_sharing_operation=op, + gross=10.0, + inj_gross=0.0, + skip=consumer_skip, + ) + await create_readings( + db_session, + ean="541448000000000002", + id_community=community.id, + id_sharing_operation=op, + gross=0.0, + inj_gross=injection, + ) + return op + + +def _from_crm_body(op: int, **overrides) -> dict: + body = { + "name": "february run", + "algorithm_name": "brute_force", + "inputs": {"iterations": 1}, + "id_sharing_operation": op, + **_PERIOD, + } + body.update(overrides) + return body + + +# --------------------------------------------------------------------------- +# 1. GET /crm-data-preview +# --------------------------------------------------------------------------- + + +async def test_preview_reports_participants_and_totals(client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + + response = await client.get( + "/crm-data-preview", + params={"id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(community), + ) + + assert response.status_code == 200 + data = response.json()["data"] + assert data["can_generate"] is True + # Only the drawing meter is a participant; the PV site just contributes. + assert data["meter_count"] == 1 + assert data["reading_count"] == 8 + assert data["total_consumption_kwh"] == 40.0 + assert data["total_injection_kwh"] == 160.0 + assert data["blockers"] == [] + assert data["incomplete_meters"] == [] + + +async def test_preview_is_reachable_and_not_swallowed_by_the_id_route(client, db_session): + # `GET /{id}` is declared after this route; if the order regressed, the + # path would be parsed as an integer id and 422. + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + + response = await client.get( + "/crm-data-preview", + params={"id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(community), + ) + + assert response.status_code == 200 + + +async def test_preview_reports_gaps_as_warnings_without_blocking(client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community, consumer_skip={1, 2}) + + response = await client.get( + "/crm-data-preview", + params={"id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(community), + ) + + data = response.json()["data"] + assert data["can_generate"] is True, "a gap must warn, not block" + assert data["incomplete_meters"] == [ + {"ean": "541448000000000001", "readings": 2, "expected": 4, "missing": 2} + ] + + +async def test_preview_blocks_on_duplicate_readings(client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community, duplicate=True) + + response = await client.get( + "/crm-data-preview", + params={"id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(community), + ) + + data = response.json()["data"] + assert data["can_generate"] is False + codes = {b["error_code"] for b in data["blockers"]} + assert errors.generation.CRM_DUPLICATE_READINGS.code in codes + # The message is localised, not a bare translation key. + blocker = next(b for b in data["blockers"] if b["error_code"] in codes) + assert not blocker["message"].startswith("ERRORS.") + + +async def test_preview_blocks_on_empty_period(client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + + response = await client.get( + "/crm-data-preview", + params={ + "id_sharing_operation": op, + "period_start": "2024-01-01", + "period_end": "2024-01-31", + }, + headers=_admin_headers(community), + ) + + data = response.json()["data"] + assert data["can_generate"] is False + assert [b["error_code"] for b in data["blockers"]] == [errors.generation.CRM_NO_DATA.code] + + +async def test_preview_rejects_inverted_period(client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + + response = await client.get( + "/crm-data-preview", + params={ + "id_sharing_operation": op, + "period_start": "2025-02-28", + "period_end": "2025-02-01", + }, + headers=_admin_headers(community), + ) + + assert response.status_code == 422 + assert response.json()["error_code"] == errors.generation.INVALID_PERIOD.code + + +async def test_preview_cannot_reach_another_communitys_operation(client, db_session): + owner = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, owner) + intruder = await _community_with_subscription(db_session) + + response = await client.get( + "/crm-data-preview", + params={"id_sharing_operation": op, **_PERIOD}, + headers=_admin_headers(intruder), + ) + + assert response.status_code == 404 + assert response.json()["error_code"] == errors.generation.SHARING_OPERATION_NOT_FOUND.code + + +# --------------------------------------------------------------------------- +# 2. POST /from-crm +# --------------------------------------------------------------------------- + + +@patch("api.generation.service.get_jetstream", MagicMock()) +@patch("api.generation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_persists_a_crm_sourced_row(send_event, client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + + response = await client.post( + "/from-crm", json=_from_crm_body(op), headers=_admin_headers(community) + ) + + assert response.status_code == 200, response.text + generation_id = response.json()["data"]["id"] + assert response.json()["data"]["status"] == GenerationStatus.PENDING + + row = ( + await db_session.execute(select(GenerationModel).where(GenerationModel.id == generation_id)) + ).scalar_one() + assert row.source == DataSource.CRM + assert row.id_sharing_operation == op + assert row.period_start == datetime.date(2025, 2, 1) + assert row.period_end == datetime.date(2025, 2, 28) + # No file was uploaded, so the file columns stay empty — which the + # ck_generation_source CHECK only permits for source = CRM. + assert row.file_storage_key is None + assert row.file_name is None + assert row.injection_name is None + assert row.data_warnings is None + send_event.assert_awaited_once() + + +@patch("api.generation.service.get_jetstream", MagicMock()) +@patch("api.generation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_persists_gap_warnings(send_event, client, db_session): + # "Warn but allow" is only honest if the warning outlives the preview. + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community, consumer_skip={1}) + + response = await client.post( + "/from-crm", json=_from_crm_body(op), headers=_admin_headers(community) + ) + + assert response.status_code == 200, response.text + row = ( + await db_session.execute( + select(GenerationModel).where(GenerationModel.id == response.json()["data"]["id"]) + ) + ).scalar_one() + assert row.data_warnings == { + "incomplete_meters": [ + {"ean": "541448000000000001", "readings": 3, "expected": 4, "missing": 1} + ] + } + + +@patch("api.generation.service.get_jetstream", MagicMock()) +@patch("api.generation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_refuses_duplicate_readings(send_event, client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community, duplicate=True) + + response = await client.post( + "/from-crm", json=_from_crm_body(op), headers=_admin_headers(community) + ) + + assert response.status_code == 422 + assert response.json()["error_code"] == errors.generation.CRM_DUPLICATE_READINGS.code + send_event.assert_not_awaited() + + +@patch("api.generation.service.get_jetstream", MagicMock()) +@patch("api.generation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_refuses_a_period_with_no_injection(send_event, client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community, injection=0.0) + + response = await client.post( + "/from-crm", json=_from_crm_body(op), headers=_admin_headers(community) + ) + + assert response.status_code == 422 + assert response.json()["error_code"] == errors.generation.CRM_NO_INJECTION.code + send_event.assert_not_awaited() + + +@patch("api.generation.service.get_jetstream", MagicMock()) +@patch("api.generation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_cannot_use_another_communitys_operation( + send_event, client, db_session +): + owner = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, owner) + intruder = await _community_with_subscription(db_session) + + response = await client.post( + "/from-crm", json=_from_crm_body(op), headers=_admin_headers(intruder) + ) + + assert response.status_code == 404 + assert response.json()["error_code"] == errors.generation.SHARING_OPERATION_NOT_FOUND.code + send_event.assert_not_awaited() + + +@patch("api.generation.service.get_jetstream", MagicMock()) +@patch("api.generation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_rejects_an_unknown_algorithm(send_event, client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + + response = await client.post( + "/from-crm", + json=_from_crm_body(op, algorithm_name="does_not_exist"), + headers=_admin_headers(community), + ) + + assert response.status_code == 404 + assert response.json()["error_code"] == errors.generation.ALGORITHM_NOT_FOUND.code + + +@patch("api.generation.service.get_jetstream", MagicMock()) +@patch("api.generation.service.send_event", new_callable=AsyncMock) +async def test_start_from_crm_rejects_invalid_algorithm_inputs(send_event, client, db_session): + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + + response = await client.post( + "/from-crm", + json=_from_crm_body(op, inputs={"iterations": 99}), + headers=_admin_headers(community), + ) + + assert response.status_code == 422 + assert response.json()["error_code"] == errors.generation.INVALID_ALGORITHM_INPUTS.code + + +async def test_start_from_crm_body_is_json_not_query_params(client, db_session): + # Guards the with_default_error / `from __future__ import annotations` trap: + # if the route module ever gains that import, the Pydantic body is demoted + # to query params and every well-formed request 422s with loc=[query, body]. + community = await _community_with_subscription(db_session) + op = await _operation_with_data(db_session, community) + + response = await client.post( + "/from-crm", json=_from_crm_body(op), headers=_admin_headers(community) + ) + + assert response.status_code != 422, response.text diff --git a/tests/api/generation/test_generation_routes.py b/tests/api/generation/test_generation_routes.py index 0d26efc..4847693 100644 --- a/tests/api/generation/test_generation_routes.py +++ b/tests/api/generation/test_generation_routes.py @@ -19,6 +19,7 @@ from sqlalchemy import select from core.database.models import Community +from core.middleware.request_limits import MAX_BODY_BYTES, UPLOAD_MAX_BODY_BYTES from shared.const import GenerationStatus from shared.models.crm_models import ( AllocationKeyModel, @@ -603,6 +604,51 @@ async def _chunked_stream(): mock_upload.assert_not_awaited() # rejected before the storage write +@patch("api.generation.service.storage.delete", new_callable=AsyncMock) +@patch("api.generation.service.storage.upload", new_callable=AsyncMock) +@patch("api.generation.service.get_jetstream", return_value=MagicMock()) +@patch("api.generation.service.send_event", new_callable=AsyncMock) +async def test_start_generation_large_body_with_content_length_is_accepted( + mock_send, mock_jetstream, mock_upload, mock_delete, client, db_session +): + """A declared-length upload between the default and upload caps must pass. + + Regression test. RequestLimitsMiddleware grants the larger + UPLOAD_MAX_BODY_BYTES only to the method/path pairs in `_UPLOAD_ROUTES`, + matched against the path THIS APP sees. That entry once read + ``("POST", "/generation/")`` — the public KrakenD path — but + generation_routes is mounted with no prefix, so the upload arrives as + ``POST /``, never matched, and every real upload over 2 MB was rejected by + the middleware with 413 / error_code 0. + + Unlike the chunked test above, this body carries a Content-Length, which is + the *only* thing the middleware gate looks at — so this is the path a real + browser upload takes, and the one that was broken. + """ + community = await _community_with_subscription(db_session) + + # Over MAX_BODY_BYTES (2 MB), under UPLOAD_MAX_BODY_BYTES (50 MB). Sent via + # files=/data= so httpx computes a real Content-Length. The route never + # parses the file — it uploads the bytes and publishes — so filler is fine. + file_bytes = b"col1,col2,production\n" + b"1,2,3\n" * 400_000 + assert MAX_BODY_BYTES < len(file_bytes) < UPLOAD_MAX_BODY_BYTES + + response = await client.post( + "/", + headers=_admin_headers(community), + **_multipart_payload( + name="large upload", + file=("large.csv", file_bytes, "text/csv"), + ), + ) + + assert response.status_code == 200 + assert response.json()["error_code"] == 0 + # Reaching the storage write proves the middleware gate let the body past. + mock_upload.assert_awaited_once() + assert mock_upload.await_args.args[1] == file_bytes + + # --------------------------------------------------------------------------- # GET /generation/{id} (allocation keys list) # --------------------------------------------------------------------------- diff --git a/tests/core/test_request_limits.py b/tests/core/test_request_limits.py new file mode 100644 index 0000000..c45c18e --- /dev/null +++ b/tests/core/test_request_limits.py @@ -0,0 +1,48 @@ +"""Per-route body-size cap tests for the request-limits middleware. + +The generation upload is ``POST /`` (generation_routes is mounted without a +prefix; the ``/generation`` prefix in the public API is added by KrakenD), so it +must receive the larger ``UPLOAD_MAX_BODY_BYTES`` cap; every other route keeps +the conservative ``MAX_BODY_BYTES`` default. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +from core.middleware import request_limits + + +def _request(method: str, path: str) -> SimpleNamespace: + return SimpleNamespace(method=method, url=SimpleNamespace(path=path)) + + +def test_upload_route_gets_large_cap(): + req = _request("POST", "/") + assert request_limits._max_body_for(req) == request_limits.UPLOAD_MAX_BODY_BYTES + + +def test_get_root_keeps_default_cap(): + # The upload cap is POST-only; a GET on the same path keeps the default. + req = _request("GET", "/") + assert request_limits._max_body_for(req) == request_limits.MAX_BODY_BYTES + + +def test_save_route_keeps_default_cap(): + # POST /save is a JSON route on the same router — no upload cap. + req = _request("POST", "/save") + assert request_limits._max_body_for(req) == request_limits.MAX_BODY_BYTES + + +def test_other_post_route_keeps_default_cap(): + req = _request("POST", "/health/readiness") + assert request_limits._max_body_for(req) == request_limits.MAX_BODY_BYTES + + +def test_gateway_prefixed_path_is_not_what_gets_matched(): + # Regression guard: the cap must key off the path the APP sees, not the + # public KrakenD path. `_UPLOAD_ROUTES` once held ("POST", "/generation/"), + # which no request to this service ever matches — so every declared-length + # upload over 2 MB was rejected with 413 by the middleware. + req = _request("POST", "/generation/") + assert request_limits._max_body_for(req) == request_limits.MAX_BODY_BYTES diff --git a/tests/factories/meter_factory.py b/tests/factories/meter_factory.py new file mode 100644 index 0000000..43be406 --- /dev/null +++ b/tests/factories/meter_factory.py @@ -0,0 +1,99 @@ +"""Factories for the CRM metering tables. + +These services map no ORM models for ``meter`` / ``meter_consumption`` — the +production code reads them with raw ``text()`` SQL through +``shared/crm_meter_repository.py`` — so the factories insert with raw SQL too. +That is deliberate: a test that went through an ORM model would stop exercising +the column names the real query actually depends on. + +Like the other factories here they flush and never commit; ``conftest`` wraps +each test in a transaction that is rolled back. +""" + +import datetime + +from sqlalchemy import text + +# Brussels local midnight, expressed as the UTC instant Postgres stores. February +# is CET (UTC+1), so 00:00 local is 23:00 UTC the previous day. Hard-coding the +# offset rather than importing ZoneInfo keeps the fixture obvious about which +# instant it means. +_CET = datetime.timezone(datetime.timedelta(hours=1)) + + +async def create_sharing_operation(session, *, id_community: int, name: str = "Test op") -> int: + result = await session.execute( + text( + """ + INSERT INTO sharing_operation (name, type, is_public, id_community) + VALUES (:name, 1, FALSE, :cid) + RETURNING id + """ + ), + {"name": name, "cid": id_community}, + ) + await session.flush() + return int(result.scalar_one()) + + +async def create_meter(session, *, ean: str, id_community: int) -> str: + await session.execute( + text( + """ + INSERT INTO meter (ean, meter_number, id_community) + VALUES (:ean, :num, :cid) + ON CONFLICT (ean) DO NOTHING + """ + ), + {"ean": ean, "num": f"M-{ean}", "cid": id_community}, + ) + await session.flush() + return ean + + +async def create_readings( + session, + *, + ean: str, + id_community: int, + id_sharing_operation: int, + start: datetime.datetime | None = None, + count: int = 4, + gross: float | None = 1.0, + inj_gross: float | None = 0.0, + step_minutes: int = 15, + skip: set[int] | None = None, +) -> list[datetime.datetime]: + """Insert ``count`` quarter-hourly readings, optionally skipping some. + + ``skip`` holds indices to omit, which is how a gap is produced: the meter is + then missing from those timestamps of the operation-wide grid and the + pre-flight reports it as incomplete. + """ + start = start or datetime.datetime(2025, 2, 1, 0, 0, tzinfo=_CET) + skip = skip or set() + written: list[datetime.datetime] = [] + for i in range(count): + if i in skip: + continue + ts = start + datetime.timedelta(minutes=step_minutes * i) + await session.execute( + text( + """ + INSERT INTO meter_consumption + (ean, id_sharing_operation, timestamp, gross, inj_gross, id_community) + VALUES (:ean, :op, :ts, :gross, :inj, :cid) + """ + ), + { + "ean": ean, + "op": id_sharing_operation, + "ts": ts, + "gross": gross, + "inj": inj_gross, + "cid": id_community, + }, + ) + written.append(ts) + await session.flush() + return written diff --git a/tests/shared/__init__.py b/tests/shared/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/shared/test_crm_preflight.py b/tests/shared/test_crm_preflight.py new file mode 100644 index 0000000..fba3453 --- /dev/null +++ b/tests/shared/test_crm_preflight.py @@ -0,0 +1,133 @@ +"""Unit tests for the CRM pre-flight verdict. + +No database — ``evaluate`` is a pure function of a ``CrmDataSummary``. These +pin the classification the whole feature hangs on: which findings block a run +and which are only reported. +""" + +import datetime + +from shared.crm_meter_repository import CrmDataSummary, EanCoverage +from shared.crm_preflight import MAX_READING_ROWS, evaluate +from shared.custom_errors import errors + + +def _coverage( + ean: str, + *, + row_count: int = 4, + distinct_ts: int = 4, + consumption_kwh: float = 10.0, + injection_kwh: float = 0.0, +) -> EanCoverage: + return EanCoverage( + ean=ean, + row_count=row_count, + distinct_ts=distinct_ts, + consumption_kwh=consumption_kwh, + injection_kwh=injection_kwh, + ) + + +def _summary(eans: list[EanCoverage], *, grid_size: int = 4) -> CrmDataSummary: + return CrmDataSummary( + eans=eans, + grid_size=grid_size, + first_timestamp=datetime.datetime(2025, 2, 1, tzinfo=datetime.UTC), + last_timestamp=datetime.datetime(2025, 2, 28, tzinfo=datetime.UTC), + ) + + +def _codes(preflight) -> set[int]: + return {b.error.code for b in preflight.blockers} + + +# --------------------------------------------------------------------------- +# 1. The happy path +# --------------------------------------------------------------------------- + + +def test_complete_period_with_consumption_and_injection_is_accepted(): + result = evaluate( + _summary([_coverage("A"), _coverage("PV", consumption_kwh=0.0, injection_kwh=50.0)]) + ) + assert result.ok + assert result.warnings is None + # The injection-only meter is not a participant. + assert result.consumer_eans == ["A"] + + +# --------------------------------------------------------------------------- +# 2. Blocking findings +# --------------------------------------------------------------------------- + + +def test_empty_period_blocks_with_a_single_message(): + result = evaluate(_summary([], grid_size=0)) + # One clear cause, not four derived ones. + assert _codes(result) == {errors.generation.CRM_NO_DATA.code} + + +def test_duplicate_readings_block(): + # row_count > distinct_ts means the same (ean, timestamp) landed twice, + # which would double this meter's energy with no visible symptom. + result = evaluate( + _summary( + [ + _coverage("A", row_count=8, distinct_ts=4, injection_kwh=1.0), + ] + ) + ) + assert errors.generation.CRM_DUPLICATE_READINGS.code in _codes(result) + assert "A" in result.blockers[0].detail + + +def test_no_injection_blocks(): + result = evaluate(_summary([_coverage("A", injection_kwh=0.0)])) + assert errors.generation.CRM_NO_INJECTION.code in _codes(result) + + +def test_no_consuming_meter_blocks(): + result = evaluate(_summary([_coverage("PV", consumption_kwh=0.0, injection_kwh=50.0)])) + assert errors.generation.CRM_NO_CONSUMERS.code in _codes(result) + + +def test_oversized_period_blocks(): + result = evaluate( + _summary( + [ + _coverage( + "A", + row_count=MAX_READING_ROWS + 1, + distinct_ts=MAX_READING_ROWS + 1, + injection_kwh=1.0, + ) + ], + grid_size=MAX_READING_ROWS + 1, + ) + ) + assert errors.generation.CRM_RANGE_TOO_LARGE.code in _codes(result) + + +# --------------------------------------------------------------------------- +# 3. Gaps warn but do not block — the behaviour the user chose +# --------------------------------------------------------------------------- + + +def test_gaps_warn_but_do_not_block(): + result = evaluate( + _summary( + [ + _coverage("A", injection_kwh=10.0), + _coverage("B", row_count=2, distinct_ts=2), + ], + grid_size=4, + ) + ) + assert result.ok, "a gap must not stop the run" + assert result.warnings is not None + incomplete = result.warnings["incomplete_meters"] + assert [m["ean"] for m in incomplete] == ["B"] + assert incomplete[0] == {"ean": "B", "readings": 2, "expected": 4, "missing": 2} + # The incomplete meter is still a participant; its gaps get zero-filled. + assert "B" in result.consumer_eans diff --git a/tests/shared/test_crm_timeseries.py b/tests/shared/test_crm_timeseries.py new file mode 100644 index 0000000..bc46be9 --- /dev/null +++ b/tests/shared/test_crm_timeseries.py @@ -0,0 +1,167 @@ +"""Unit tests for the CRM rows -> wide DataFrame pivot. + +No database: the pivot is a pure function over ``ConsumptionRow`` values, and +these tests pin the three behaviours the rest of the feature relies on — a +common timestamp grid, zero-filled gaps, and a single summed injection series. +""" + +import datetime + +import numpy as np +import pandas as pd +import pytest + +from shared import crm_timeseries, data_loading +from shared.crm_meter_repository import ConsumptionRow + +_TZ = datetime.UTC + + +def _ts(index: int) -> datetime.datetime: + return datetime.datetime(2025, 2, 1, 0, 0, tzinfo=_TZ) + datetime.timedelta(minutes=15 * index) + + +def _rows(spec: dict[str, list[tuple[int, float, float]]]) -> list[ConsumptionRow]: + """Build rows from {ean: [(timestamp_index, gross, inj_gross), ...]}.""" + return [ + ConsumptionRow(timestamp=_ts(i), ean=ean, gross=gross, inj_gross=inj) + for ean, entries in spec.items() + for (i, gross, inj) in entries + ] + + +# --------------------------------------------------------------------------- +# 1. Shape and ordering +# --------------------------------------------------------------------------- + + +def test_columns_are_participants_plus_injection_in_requested_order(): + rows = _rows({"B": [(0, 2.0, 0.0)], "A": [(0, 1.0, 0.0)]}) + frame = crm_timeseries.build_dataframe(rows, ["A", "B"]) + assert list(frame.columns) == ["A", "B", crm_timeseries.INJECTION_COLUMN] + + +def test_rows_are_ordered_chronologically_regardless_of_input_order(): + # Row t of every column must be contemporaneous; ordering is the only + # alignment guarantee the algorithms have. + rows = [ + ConsumptionRow(timestamp=_ts(2), ean="A", gross=30.0, inj_gross=3.0), + ConsumptionRow(timestamp=_ts(0), ean="A", gross=10.0, inj_gross=1.0), + ConsumptionRow(timestamp=_ts(1), ean="A", gross=20.0, inj_gross=2.0), + ] + frame = crm_timeseries.build_dataframe(rows, ["A"]) + assert frame["A"].tolist() == [10.0, 20.0, 30.0] + + +# --------------------------------------------------------------------------- +# 2. Gaps — the "warn but allow" behaviour +# --------------------------------------------------------------------------- + + +def test_missing_timestamps_are_zero_filled_onto_the_common_grid(): + # A is present at 0,1,2; B only at 0 and 2. The grid is the union, so B + # gets a 0.0 at index 1 rather than the frame collapsing to A's shape. + rows = _rows( + { + "A": [(0, 1.0, 0.0), (1, 1.0, 0.0), (2, 1.0, 0.0)], + "B": [(0, 5.0, 0.0), (2, 5.0, 0.0)], + } + ) + frame = crm_timeseries.build_dataframe(rows, ["A", "B"]) + assert len(frame) == 3 + assert frame["B"].tolist() == [5.0, 0.0, 5.0] + + +# --------------------------------------------------------------------------- +# 3. Injection +# --------------------------------------------------------------------------- + + +def test_injection_sums_every_meter_including_non_participants(): + # C injects but never consumes, so it is not a participant — its production + # must still reach the shared profile. + rows = _rows( + { + "A": [(0, 1.0, 2.0)], + "C": [(0, 0.0, 10.0)], + } + ) + frame = crm_timeseries.build_dataframe(rows, ["A"]) + assert list(frame.columns) == ["A", crm_timeseries.INJECTION_COLUMN] + assert frame[crm_timeseries.INJECTION_COLUMN].tolist() == [12.0] + + +def test_non_participant_meters_are_not_consumer_columns(): + rows = _rows({"A": [(0, 1.0, 0.0)], "C": [(0, 0.0, 9.0)]}) + frame = crm_timeseries.build_dataframe(rows, ["A"]) + assert "C" not in frame.columns + + +# --------------------------------------------------------------------------- +# 4. Refusals +# --------------------------------------------------------------------------- + + +def test_requested_participant_with_no_readings_is_refused(): + # Reindexing would otherwise invent an all-zero column, which for a + # simulation means a key participant silently gets a 0 kWh profile. + rows = _rows({"A": [(0, 1.0, 1.0)]}) + with pytest.raises(crm_timeseries.CrmPivotError, match="GHOST"): + crm_timeseries.build_dataframe(rows, ["A", "GHOST"]) + + +def test_duplicate_ean_timestamp_pairs_are_refused(): + rows = [ + ConsumptionRow(timestamp=_ts(0), ean="A", gross=1.0, inj_gross=1.0), + ConsumptionRow(timestamp=_ts(0), ean="A", gross=1.0, inj_gross=1.0), + ] + with pytest.raises(crm_timeseries.CrmPivotError): + crm_timeseries.build_dataframe(rows, ["A"]) + + +def test_empty_inputs_are_refused(): + with pytest.raises(crm_timeseries.CrmPivotError): + crm_timeseries.build_dataframe([], ["A"]) + with pytest.raises(crm_timeseries.CrmPivotError): + crm_timeseries.build_dataframe(_rows({"A": [(0, 1.0, 1.0)]}), []) + + +def test_injection_column_name_cannot_be_a_participant(): + rows = _rows({"A": [(0, 1.0, 1.0)]}) + with pytest.raises(crm_timeseries.CrmPivotError, match="reserved"): + crm_timeseries.build_dataframe(rows, [crm_timeseries.INJECTION_COLUMN]) + + +# --------------------------------------------------------------------------- +# 5. The whole point: the frame drops into the existing file-path converter +# --------------------------------------------------------------------------- + + +def test_frame_feeds_the_existing_algorithm_converter_unchanged(): + rows = _rows( + { + "541448000000000001": [(0, 10.0, 0.0), (1, 11.0, 0.0)], + "541448000000000002": [(0, 5.0, 0.0), (1, 6.0, 0.0)], + "541448000000000003": [(0, 0.0, 100.0), (1, 0.0, 200.0)], + } + ) + participants = ["541448000000000001", "541448000000000002"] + + frame = crm_timeseries.build_dataframe(rows, participants) + raw = data_loading.to_algorithm_raw_data(frame, crm_timeseries.INJECTION_COLUMN) + + assert raw.consumer_names == participants + assert raw.C.shape == (2, 2) + np.testing.assert_array_equal(raw.C, np.array([[10.0, 11.0], [5.0, 6.0]])) + # VA is the single production series broadcast across every consumer row. + assert raw.VA.shape == (2, 2) + np.testing.assert_array_equal(raw.VA, np.array([[100.0, 200.0], [100.0, 200.0]])) + + +def test_frame_is_shaped_like_a_parsed_upload(): + # The contract that lets the converter stay untouched: a plain RangeIndex + # and float columns, exactly what pd.read_csv would produce. + rows = _rows({"A": [(0, 1.0, 1.0), (1, 2.0, 2.0)]}) + frame = crm_timeseries.build_dataframe(rows, ["A"]) + assert isinstance(frame.index, pd.RangeIndex) + assert all(pd.api.types.is_float_dtype(dtype) for dtype in frame.dtypes) diff --git a/tests/sql/crm_test_schema.sql b/tests/sql/crm_test_schema.sql index dc938c2..8ba0de6 100644 --- a/tests/sql/crm_test_schema.sql +++ b/tests/sql/crm_test_schema.sql @@ -92,3 +92,65 @@ CREATE TABLE IF NOT EXISTS audit_log ( user_email VARCHAR(256), payload JSONB NOT NULL DEFAULT '{}'::jsonb ); + + +-- ---- Metering tables ------------------------------------------------------- +-- Mirrors crm-backend/database_script/init.sql (meter / meter_data / +-- meter_consumption / sharing_operation), trimmed to the columns this service +-- actually SELECTs. Adapted from billing/tests/sql/crm_test_schema.sql, which +-- carries the same block for the same reason. +-- +-- Two production properties are reproduced deliberately, because the code under +-- test depends on both: +-- * meter_consumption has NO unique constraint on (ean, timestamp) -- that is +-- what makes a double import possible and CRM_DUPLICATE_READINGS necessary. +-- * every measure column is nullable, so COALESCE in the queries is load- +-- bearing rather than defensive. + +CREATE TABLE IF NOT EXISTS sharing_operation ( + id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + name VARCHAR(255) NOT NULL, + type INTEGER NOT NULL DEFAULT 1, + is_public BOOLEAN NOT NULL DEFAULT FALSE, + id_community INTEGER NOT NULL REFERENCES community (id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS meter ( + ean VARCHAR(64) PRIMARY KEY, + meter_number VARCHAR(255), + tarif_group INTEGER, + phases_number INTEGER, + reading_frequency INTEGER, + id_community INTEGER NOT NULL REFERENCES community (id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS meter_data ( + id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + ean VARCHAR(64) NOT NULL REFERENCES meter (ean) ON DELETE CASCADE, + id_member INTEGER, + id_sharing_operation INTEGER REFERENCES sharing_operation (id), + status INTEGER, -- 1=ACTIVE + client_type INTEGER, -- 1=Residentiel, 2=Professionnel, 3=Industriel + injection_status INTEGER, + production_chain INTEGER, + start_date DATE, + end_date DATE, + id_community INTEGER NOT NULL REFERENCES community (id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS meter_consumption ( + id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + ean VARCHAR(64) NOT NULL REFERENCES meter (ean) ON DELETE CASCADE, + id_sharing_operation INTEGER REFERENCES sharing_operation (id), + timestamp TIMESTAMPTZ NOT NULL, + gross DOUBLE PRECISION, + net DOUBLE PRECISION, + shared DOUBLE PRECISION, + inj_gross DOUBLE PRECISION, + inj_shared DOUBLE PRECISION, + inj_net DOUBLE PRECISION, + id_community INTEGER NOT NULL REFERENCES community (id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_meter_consumption_lookup + ON meter_consumption (id_sharing_operation, timestamp); diff --git a/tests/worker/test_dispatcher_crm_source.py b/tests/worker/test_dispatcher_crm_source.py new file mode 100644 index 0000000..c318003 --- /dev/null +++ b/tests/worker/test_dispatcher_crm_source.py @@ -0,0 +1,229 @@ +"""Worker tests for the CRM-sourced branch of ``_process``. + +Same style as test_dispatcher_failure_paths.py: drive ``_load_from_crm`` +directly with patched collaborators, so no live NATS, Postgres or MinIO is +needed. + +What matters here is the failure classification, which differs from the file +path in one respect that is easy to get wrong: a CRM **read** error must be +transient (NAK, redeliver) while **rejected data** must be deterministic +(FAILED, ack). Getting that backwards either burns a run on a blip or hot-loops +the queue on a permanent data problem. +""" + +from __future__ import annotations + +import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from shared.const import DataSource +from shared.crm_meter_repository import ConsumptionRow, CrmDataSummary, EanCoverage +from worker import dispatcher + +_TZ = datetime.UTC + + +def _snapshot( + *, + generation_id: int = 1, + source: DataSource = DataSource.CRM, + id_sharing_operation: int | None = 7, + period_start: datetime.date | None = datetime.date(2025, 2, 1), + period_end: datetime.date | None = datetime.date(2025, 2, 28), +) -> dispatcher._GenerationSnapshot: + return dispatcher._GenerationSnapshot( + id=generation_id, + source=source, + file_storage_key=None, + file_name=None, + injection_name=None, + id_sharing_operation=id_sharing_operation, + period_start=period_start, + period_end=period_end, + inputs={"value": 1}, + id_community=42, + status=0, + ) + + +def _coverage(ean: str, *, consumption: float, injection: float, dup: bool = False) -> EanCoverage: + return EanCoverage( + ean=ean, + row_count=4 if not dup else 8, + distinct_ts=4, + consumption_kwh=consumption, + injection_kwh=injection, + ) + + +def _summary(eans: list[EanCoverage]) -> CrmDataSummary: + return CrmDataSummary( + eans=eans, + grid_size=4, + first_timestamp=datetime.datetime(2025, 2, 1, tzinfo=_TZ), + last_timestamp=datetime.datetime(2025, 2, 1, 0, 45, tzinfo=_TZ), + ) + + +def _rows() -> list[ConsumptionRow]: + base = datetime.datetime(2025, 2, 1, tzinfo=_TZ) + out: list[ConsumptionRow] = [] + for i in range(4): + ts = base + datetime.timedelta(minutes=15 * i) + out.append(ConsumptionRow(timestamp=ts, ean="A", gross=10.0, inj_gross=0.0)) + out.append(ConsumptionRow(timestamp=ts, ean="PV", gross=0.0, inj_gross=25.0)) + return out + + +@pytest.fixture +def patched_save(monkeypatch): + save_failure = AsyncMock() + monkeypatch.setattr(dispatcher.persistence, "save_failure", save_failure) + return save_failure + + +def _patch_crm(monkeypatch, *, summary, rows=None, raises: Exception | None = None): + """Patch AsyncSessionCRMFactory + CrmMeterRepository on the dispatcher.""" + repository = MagicMock() + if raises is not None: + repository.summarize = AsyncMock(side_effect=raises) + else: + repository.summarize = AsyncMock(return_value=summary) + repository.fetch_rows = AsyncMock(return_value=rows or []) + + session_cm = MagicMock() + session_cm.__aenter__ = AsyncMock(return_value=MagicMock()) + session_cm.__aexit__ = AsyncMock(return_value=False) + + monkeypatch.setattr(dispatcher, "AsyncSessionCRMFactory", MagicMock(return_value=session_cm)) + monkeypatch.setattr(dispatcher, "CrmMeterRepository", MagicMock(return_value=repository)) + return repository + + +# --------------------------------------------------------------------------- +# 1. Happy path +# --------------------------------------------------------------------------- + + +async def test_returns_raw_data_built_from_crm_rows(monkeypatch, patched_save): + repository = _patch_crm( + monkeypatch, + summary=_summary( + [ + _coverage("A", consumption=40.0, injection=0.0), + _coverage("PV", consumption=0.0, injection=100.0), + ] + ), + rows=_rows(), + ) + + result = await dispatcher._load_from_crm(_snapshot()) + + assert not isinstance(result, dispatcher._Terminal) + assert result.consumer_names == ["A"] + assert result.C.shape == (1, 4) + # VA carries the PV site's production even though it is not a participant. + assert result.VA.tolist() == [[25.0, 25.0, 25.0, 25.0]] + patched_save.assert_not_awaited() + # The community is passed explicitly — the worker has no ContextVar. + assert repository.summarize.await_args.kwargs["id_community"] == 42 + assert repository.summarize.await_args.kwargs["id_sharing_operation"] == 7 + + +# --------------------------------------------------------------------------- +# 2. Transient: the CRM is unreachable +# --------------------------------------------------------------------------- + + +async def test_crm_read_failure_is_transient(monkeypatch, patched_save): + _patch_crm(monkeypatch, summary=None, raises=OSError("connection reset")) + + with pytest.raises(dispatcher._TransientError, match="crm read"): + await dispatcher._load_from_crm(_snapshot()) + + # A blip must not burn the run. + patched_save.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# 3. Deterministic: the data itself is unusable +# --------------------------------------------------------------------------- + + +async def test_empty_period_fails_deterministically(monkeypatch, patched_save): + repository = _patch_crm(monkeypatch, summary=_summary([])) + + result = await dispatcher._load_from_crm(_snapshot()) + + assert isinstance(result, dispatcher._Terminal) + # Nothing to clean up — there is no uploaded object on this path. + assert result.storage_key is None + patched_save.assert_awaited_once() + assert "crm_data_rejected" in patched_save.await_args.args[1] + # The expensive read is skipped once the period is already rejected. + repository.fetch_rows.assert_not_awaited() + + +async def test_duplicate_readings_fail_deterministically(monkeypatch, patched_save): + _patch_crm( + monkeypatch, + summary=_summary([_coverage("A", consumption=40.0, injection=10.0, dup=True)]), + ) + + result = await dispatcher._load_from_crm(_snapshot()) + + assert isinstance(result, dispatcher._Terminal) + detail = patched_save.await_args.args[1] + assert "crm_data_rejected" in detail + assert "imported twice" in detail + + +async def test_no_injection_fails_deterministically(monkeypatch, patched_save): + _patch_crm(monkeypatch, summary=_summary([_coverage("A", consumption=40.0, injection=0.0)])) + + result = await dispatcher._load_from_crm(_snapshot()) + + assert isinstance(result, dispatcher._Terminal) + assert "nothing to share" in patched_save.await_args.args[1] + + +async def test_incomplete_crm_columns_fail_deterministically(monkeypatch, patched_save): + # ck_generation_source makes this unreachable via the API, but a row written + # straight to the DB could still land here. + _patch_crm(monkeypatch, summary=_summary([])) + + result = await dispatcher._load_from_crm(_snapshot(id_sharing_operation=None)) + + assert isinstance(result, dispatcher._Terminal) + patched_save.assert_awaited_once_with(1, "crm_source_incomplete") + + +# --------------------------------------------------------------------------- +# 4. Gaps still run — the user-chosen behaviour, verified end to end +# --------------------------------------------------------------------------- + + +async def test_gaps_do_not_stop_the_run_and_are_zero_filled(monkeypatch, patched_save): + # B is missing the last two quarters; the grid stays 4 long. + rows = _rows() + rows = [r for r in rows if not (r.ean == "A" and r.timestamp.minute >= 30)] + _patch_crm( + monkeypatch, + summary=_summary( + [ + EanCoverage( + "A", row_count=2, distinct_ts=2, consumption_kwh=20.0, injection_kwh=0.0 + ), + _coverage("PV", consumption=0.0, injection=100.0), + ] + ), + rows=rows, + ) + + result = await dispatcher._load_from_crm(_snapshot()) + + assert not isinstance(result, dispatcher._Terminal) + assert result.C.tolist() == [[10.0, 10.0, 0.0, 0.0]] + patched_save.assert_not_awaited() diff --git a/tests/worker/test_dispatcher_failure_paths.py b/tests/worker/test_dispatcher_failure_paths.py index bf10b12..98023bb 100644 --- a/tests/worker/test_dispatcher_failure_paths.py +++ b/tests/worker/test_dispatcher_failure_paths.py @@ -40,7 +40,7 @@ AlgorithmResult, ) from core.queue.helper import Event -from shared.const import GenerationStatus +from shared.const import DataSource, GenerationStatus from worker import dispatcher # --------------------------------------------------------------------------- @@ -102,9 +102,13 @@ def _make_snapshot( ) -> dispatcher._GenerationSnapshot: return dispatcher._GenerationSnapshot( id=generation_id, + source=DataSource.FILE, file_storage_key=file_storage_key, file_name="data.csv", injection_name="production", + id_sharing_operation=None, + period_start=None, + period_end=None, inputs=inputs if inputs is not None else {"value": 1}, id_community=1, status=int(status), @@ -227,7 +231,7 @@ async def test_handler_calls_save_failure_when_algorithm_raises( monkeypatch.setattr( dispatcher, "_snapshot_generation", - AsyncMock(return_value=_make_snapshot()), + AsyncMock(return_value=_make_snapshot(generation_id=42)), ) monkeypatch.setattr( "algorithms.registry.registry.implementation", @@ -262,7 +266,7 @@ async def test_handler_calls_save_failure_when_implementation_missing( monkeypatch.setattr( dispatcher, "_snapshot_generation", - AsyncMock(return_value=_make_snapshot()), + AsyncMock(return_value=_make_snapshot(generation_id=7)), ) def _missing(_name): @@ -332,7 +336,7 @@ async def test_handler_naks_when_save_success_raises_db_error( monkeypatch.setattr( dispatcher, "_snapshot_generation", - AsyncMock(return_value=_make_snapshot()), + AsyncMock(return_value=_make_snapshot(generation_id=99)), ) monkeypatch.setattr( "algorithms.registry.registry.implementation", @@ -396,7 +400,7 @@ async def test_handler_deletes_object_on_success( monkeypatch.setattr( dispatcher, "_snapshot_generation", - AsyncMock(return_value=_make_snapshot()), + AsyncMock(return_value=_make_snapshot(generation_id=33)), ) monkeypatch.setattr( "algorithms.registry.registry.implementation", @@ -430,7 +434,7 @@ async def test_handler_marks_failed_when_storage_object_missing( monkeypatch.setattr( dispatcher, "_snapshot_generation", - AsyncMock(return_value=_make_snapshot()), + AsyncMock(return_value=_make_snapshot(generation_id=44)), ) handler = dispatcher._make_handler(_make_meta()) @@ -461,7 +465,7 @@ async def test_handler_naks_on_storage_transient_error(monkeypatch, patched_save monkeypatch.setattr( dispatcher, "_snapshot_generation", - AsyncMock(return_value=_make_snapshot()), + AsyncMock(return_value=_make_snapshot(generation_id=77)), ) handler = dispatcher._make_handler(_make_meta()) diff --git a/worker/dispatcher.py b/worker/dispatcher.py index 94771d3..4b25649 100644 --- a/worker/dispatcher.py +++ b/worker/dispatcher.py @@ -34,6 +34,7 @@ import asyncio import dataclasses +import datetime import logging import time from collections.abc import Awaitable, Callable @@ -54,10 +55,11 @@ from algorithms.registry import registry from core import metrics as app_metrics from core import storage -from core.database.database import AsyncSessionLocalFactory +from core.database.database import AsyncSessionCRMFactory, AsyncSessionLocalFactory from core.queue.helper import Event -from shared import data_loading -from shared.const import GenerationStatus +from shared import crm_preflight, crm_timeseries, data_loading +from shared.const import DataSource, GenerationStatus +from shared.crm_meter_repository import CrmMeterRepository from shared.models.local_models import GenerationModel from worker import persistence @@ -79,9 +81,15 @@ class _GenerationSnapshot: """Per-message snapshot of the row, captured before the session closes.""" id: int - file_storage_key: str - file_name: str - injection_name: str + source: DataSource + # FILE only — None on a CRM-sourced row. + file_storage_key: str | None + file_name: str | None + injection_name: str | None + # CRM only — None on a file-sourced row. + id_sharing_operation: int | None + period_start: datetime.date | None + period_end: datetime.date | None inputs: dict id_community: int status: int @@ -282,30 +290,17 @@ async def _process( ) return _Terminal(storage_key=snapshot.file_storage_key) - # ---- Step 2: download the source file ------------------------------ - try: - content = await storage.download(snapshot.file_storage_key) - except storage.ObjectNotFound: - # The object is gone — possibly a previous run cleaned it up but - # the row was somehow still PENDING (admin intervention, partial - # crash). Terminal: mark FAILED and ack. Nothing left to delete. - await persistence.save_failure(generation_id, "storage_object_missing") - return _Terminal(storage_key=None) - except storage.TransientStorageError as exc: - raise _TransientError(f"storage download: {exc}") from exc - - # ---- Step 3: parse the file into the algorithm raw-data triple ----- - try: - raw_data = data_loading.load(content, snapshot.file_name, snapshot.injection_name) - except ( - data_loading.InvalidInjectionColumnError, - data_loading.UnsupportedFileFormatError, - ) as exc: - await persistence.save_failure(generation_id, f"parse_failed: {exc}") - return _Terminal(storage_key=snapshot.file_storage_key) - except Exception as exc: - await persistence.save_failure(generation_id, f"parse_failed_unexpected: {exc}") - return _Terminal(storage_key=snapshot.file_storage_key) + # ---- Steps 2+3: obtain the algorithm raw-data triple --------------- + # Two sources, one output. FILE downloads and parses the upload; CRM reads + # meter_consumption and pivots it into the same wide frame. Everything + # downstream of here is identical and knows nothing about the source. + if snapshot.source is DataSource.CRM: + loaded = await _load_from_crm(snapshot) + else: + loaded = await _load_from_file(snapshot) + if isinstance(loaded, _Terminal): + return loaded + raw_data = loaded # ---- Step 4: re-validate inputs against the algorithm schema ------- # Defensive: the API validates these on creation, but a row written @@ -426,6 +421,106 @@ async def _delete_safely(key: str) -> None: await storage.delete(key) +async def _load_from_file( + snapshot: _GenerationSnapshot, +) -> AlgorithmRawData | _Terminal: + """Download the uploaded object and parse it. The historical path.""" + if snapshot.file_storage_key is None or snapshot.file_name is None: + # Unreachable through the API (ck_generation_source enforces it), but a + # row written directly to the DB could get here. Same defensiveness as + # the inputs re-validation below. + await persistence.save_failure(snapshot.id, "file_source_incomplete") + return _Terminal(storage_key=None) + + try: + content = await storage.download(snapshot.file_storage_key) + except storage.ObjectNotFound: + # The object is gone — possibly a previous run cleaned it up but + # the row was somehow still PENDING (admin intervention, partial + # crash). Terminal: mark FAILED and ack. Nothing left to delete. + await persistence.save_failure(snapshot.id, "storage_object_missing") + return _Terminal(storage_key=None) + except storage.TransientStorageError as exc: + raise _TransientError(f"storage download: {exc}") from exc + + try: + return data_loading.load(content, snapshot.file_name, snapshot.injection_name or "") + except ( + data_loading.InvalidInjectionColumnError, + data_loading.UnsupportedFileFormatError, + ) as exc: + await persistence.save_failure(snapshot.id, f"parse_failed: {exc}") + return _Terminal(storage_key=snapshot.file_storage_key) + except Exception as exc: + await persistence.save_failure(snapshot.id, f"parse_failed_unexpected: {exc}") + return _Terminal(storage_key=snapshot.file_storage_key) + + +async def _load_from_crm( + snapshot: _GenerationSnapshot, +) -> AlgorithmRawData | _Terminal: + """Read meter_consumption for this row's sharing operation and period. + + The pre-flight is re-run here rather than trusted from creation time: the + data can have changed since the run was queued, and this is the read that + actually feeds the algorithm. + + Failure classification follows the module's existing matrix — a CRM read + error is transient (NAK, redeliver), while rejected or unpivotable data is + deterministic (FAILED, ack). There is never an object to delete. + """ + if ( + snapshot.id_sharing_operation is None + or snapshot.period_start is None + or snapshot.period_end is None + ): + await persistence.save_failure(snapshot.id, "crm_source_incomplete") + return _Terminal(storage_key=None) + + try: + async with AsyncSessionCRMFactory() as crm_session: + repository = CrmMeterRepository(crm_session) + # The worker has no request context, so the community is passed + # explicitly; with_community_scope would degrade to WHERE false. + summary = await repository.summarize( + id_community=snapshot.id_community, + id_sharing_operation=snapshot.id_sharing_operation, + period_start=snapshot.period_start, + period_end=snapshot.period_end, + ) + preflight = crm_preflight.evaluate(summary) + # Skip the expensive read when the period is already rejected. + rows = ( + await repository.fetch_rows( + id_community=snapshot.id_community, + id_sharing_operation=snapshot.id_sharing_operation, + period_start=snapshot.period_start, + period_end=snapshot.period_end, + ) + if preflight.ok + else [] + ) + except Exception as exc: + raise _TransientError(f"crm read: {exc}") from exc + + if preflight.blockers: + detail = "; ".join(b.detail for b in preflight.blockers) + await persistence.save_failure(snapshot.id, f"crm_data_rejected: {detail}") + return _Terminal(storage_key=None) + + try: + frame = crm_timeseries.build_dataframe(rows, preflight.consumer_eans) + # The same converter the file path uses — the frame is deliberately + # shaped like a parsed upload so nothing below this line differs. + return data_loading.to_algorithm_raw_data(frame, crm_timeseries.INJECTION_COLUMN) + except (crm_timeseries.CrmPivotError, data_loading.InvalidInjectionColumnError) as exc: + await persistence.save_failure(snapshot.id, f"crm_pivot_failed: {exc}") + return _Terminal(storage_key=None) + except Exception as exc: + await persistence.save_failure(snapshot.id, f"crm_pivot_failed_unexpected: {exc}") + return _Terminal(storage_key=None) + + async def _snapshot_generation(generation_id: int) -> _GenerationSnapshot | None: """Read the row in a short-lived session and return a frozen snapshot. @@ -438,9 +533,13 @@ async def _snapshot_generation(generation_id: int) -> _GenerationSnapshot | None return None return _GenerationSnapshot( id=row.id, + source=DataSource(row.source), file_storage_key=row.file_storage_key, file_name=row.file_name, injection_name=row.injection_name, + id_sharing_operation=row.id_sharing_operation, + period_start=row.period_start, + period_end=row.period_end, inputs=dict(row.inputs) if row.inputs else {}, id_community=row.id_community, status=int(row.status), diff --git a/worker/main.py b/worker/main.py index 1450163..6138bc7 100644 --- a/worker/main.py +++ b/worker/main.py @@ -30,6 +30,7 @@ from core.database.database import crm_engine, local_engine from core.logging import configure_logging from core.queue.init import close_nats, get_jetstream, init_nats +from core.realtime import log_realtime_state from core.tracing import setup_tracer_provider from worker import dispatcher @@ -222,6 +223,9 @@ async def _poll_queue_depth(js, shutdown_event: asyncio.Event) -> None: async def main() -> None: configure_logging() + # Absence of this line means the image predates the realtime feature — + # see core/realtime/bus.py. Must come after configure_logging(). + log_realtime_state("allocation-key-generation-worker") setup_tracer_provider() # Heavy import path: pulls numpy/cvxpy/pandas via each algorithm's diff --git a/worker/persistence.py b/worker/persistence.py index e2b3ae4..41a4450 100644 --- a/worker/persistence.py +++ b/worker/persistence.py @@ -32,6 +32,7 @@ from core import metrics as app_metrics from core.audit_log import AuditActions, AuditLogInput, AuditLogService from core.database.database import AsyncSessionCRMFactory, AsyncSessionLocalFactory +from core.realtime import CommunityAudience, Tier, emit from shared.const import GenerationStatus from shared.models.local_models import ( AllocationKeyGeneratedModel, @@ -139,6 +140,26 @@ async def save_success(generation_id: int, result: AlgorithmResult) -> None: ) await crm_session.commit() + # Realtime hint, AFTER both commits. Fire-and-forget: if no manager has the + # hub open it is dropped, which is correct — the row is already durable and + # the hub's own poller converges regardless. + # + # Audience is the community's MANAGER tier, not a user: `generation` carries + # only `id_community`, and the hub route is manager-gated anyway + # (annexes-services.routes.ts, minRole GESTIONNAIRE). That is the whole point + # of the community channel family — a worker with no request context and no + # user attribution addresses exactly the right people with zero lookups. + # + # No error message and no key count in the envelope: it is a hint, and the + # client refetches through the gateway (which re-authorizes the read). + await emit( + topic="generation.finished", + audience=CommunityAudience(community_id=community_id, tier=Tier.MANAGER), + resource=("generation", generation_id), + scope_community_id=community_id, + hint={"status": "success"}, + ) + async def save_failure(generation_id: int, error_message: str) -> None: """Mark a generation FAILED with the given message. @@ -194,3 +215,14 @@ async def save_failure(generation_id: int, error_message: str) -> None: id_community=id_community, ) await crm_session.commit() + + # Same contract as save_success. `id_community` was captured before the + # session closed; both factories set expire_on_commit=False, so reading it + # here does not re-issue a query against a closed session. + await emit( + topic="generation.finished", + audience=CommunityAudience(community_id=id_community, tier=Tier.MANAGER), + resource=("generation", generation_id), + scope_community_id=id_community, + hint={"status": "failed"}, + )