feat: 6 critical Kafka→DB projection nodes [OMN-7610]

.github/workflows/ci.yml

@@ -11,6 +11,8 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
+      - name: Clone omnibase_core
+        run: git clone --depth=1 https://github.com/OmniNode-ai/omnibase_core.git ../omnibase_core
       - name: Install uv
         uses: astral-sh/setup-uv@v4
         with:
@@ -28,6 +30,8 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
+      - name: Clone omnibase_core
+        run: git clone --depth=1 https://github.com/OmniNode-ai/omnibase_core.git ../omnibase_core
       - name: Install uv
         uses: astral-sh/setup-uv@v4
         with:
@@ -43,6 +47,8 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
+      - name: Clone omnibase_core
+        run: git clone --depth=1 https://github.com/OmniNode-ai/omnibase_core.git ../omnibase_core
       - name: Install uv
         uses: astral-sh/setup-uv@v4
         with:

Dockerfile.projection

@@ -0,0 +1,18 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+# Install uv for fast dependency resolution
+RUN pip install --no-cache-dir uv
+
+# Copy dependency file first for layer caching
+COPY pyproject.toml ./
+COPY src/ ./src/
+
+# Install the package with projection dependencies
+RUN uv pip install --system --no-cache ".[projection]" 2>/dev/null || \
+    uv pip install --system --no-cache aiokafka asyncpg python-dateutil && \
+    uv pip install --system --no-cache -e .
+
+# Default: overridden by docker-compose command
+CMD ["python", "-m", "omnimarket"]

docker-compose.projection.yml

@@ -0,0 +1,62 @@
+###############################################################################
+# docker-compose.projection.yml — OMN-7610 projection nodes
+#
+# Deploys all 6 critical Kafka→DB projection consumers as independent
+# Docker services on .201 server. Each service is a standalone Python
+# process consuming from its Kafka topic(s) and projecting into the
+# omnidash_analytics PostgreSQL database.
+#
+# Usage:
+#   docker compose -f docker-compose.projection.yml up -d
+#   docker compose -f docker-compose.projection.yml logs -f
+#   docker compose -f docker-compose.projection.yml down
+###############################################################################
+
+x-projection-defaults: &projection-defaults
+  build:
+    context: .
+    dockerfile: Dockerfile.projection
+  restart: unless-stopped
+  network_mode: host
+  env_file: .env
+  environment:
+    KAFKA_BROKERS: ${KAFKA_BROKERS:-192.168.86.201:19092}
+    OMNIDASH_ANALYTICS_DB_URL: ${OMNIDASH_ANALYTICS_DB_URL}
+    KAFKA_CONSUMER_GROUP: ${KAFKA_CONSUMER_GROUP:-omnimarket-projections-v1}
+    PYTHONUNBUFFERED: "1"
+  logging:
+    driver: json-file
+    options:
+      max-size: "10m"
+      max-file: "3"
+
+services:
+  projection-session-outcome:
+    <<: *projection-defaults
+    container_name: omnimarket-projection-session-outcome
+    command: ["python", "-m", "omnimarket.nodes.node_projection_session_outcome.handlers.handler_session_outcome"]
+
+  projection-llm-cost:
+    <<: *projection-defaults
+    container_name: omnimarket-projection-llm-cost
+    command: ["python", "-m", "omnimarket.nodes.node_projection_llm_cost.handlers.handler_llm_cost"]
+
+  projection-savings:
+    <<: *projection-defaults
+    container_name: omnimarket-projection-savings
+    command: ["python", "-m", "omnimarket.nodes.node_projection_savings.handlers.handler_savings"]
+
+  projection-delegation:
+    <<: *projection-defaults
+    container_name: omnimarket-projection-delegation
+    command: ["python", "-m", "omnimarket.nodes.node_projection_delegation.handlers.handler_delegation"]
+
+  projection-baselines:
+    <<: *projection-defaults
+    container_name: omnimarket-projection-baselines
+    command: ["python", "-m", "omnimarket.nodes.node_projection_baselines.handlers.handler_baselines"]
+
+  projection-registration:
+    <<: *projection-defaults
+    container_name: omnimarket-projection-registration
+    command: ["python", "-m", "omnimarket.nodes.node_projection_registration.handlers.handler_registration"]

pyproject.toml

@@ -32,6 +32,9 @@ dependencies = [
     "omnibase_core>=0.39.0",
     "pydantic>=2.12.5,<3.0.0",
     "pyyaml>=6.0.2,<7.0.0",
+    "aiokafka>=0.12.0",
+    "asyncpg>=0.30.0",
+    "python-dateutil>=2.9.0",
 ]
 
 [project.urls]
@@ -105,7 +108,7 @@ ignore = [
 ]
 
 [tool.ruff.lint.per-file-ignores]
-"tests/**/*.py" = ["T201"]
+"tests/**/*.py" = ["T201", "SLF001"]
 
 [tool.mypy]
 strict = true

src/omnimarket/adapters/asyncpg_adapter.py

@@ -0,0 +1,71 @@
+"""Asyncpg implementation of DatabaseAdapter."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any
+
+import asyncpg  # type: ignore[import-untyped]
+
+logger = logging.getLogger(__name__)
+
+DB_URL_ENV = "OMNIDASH_ANALYTICS_DB_URL"
+
+
+class AsyncpgAdapter:
+    """DatabaseAdapter backed by asyncpg connection pool."""
+
+    def __init__(
+        self, dsn: str | None = None, min_size: int = 2, max_size: int = 10
+    ) -> None:
+        self._dsn = dsn or os.environ.get(DB_URL_ENV, "")
+        self._min_size = min_size
+        self._max_size = max_size
+        self._pool: asyncpg.Pool | None = None
+
+    async def connect(self) -> None:
+        if not self._dsn:
+            raise RuntimeError(f"{DB_URL_ENV} not set and no DSN provided")
+        self._pool = await asyncpg.create_pool(
+            self._dsn,
+            min_size=self._min_size,
+            max_size=self._max_size,
+            command_timeout=30,
+        )
+        logger.info(
+            "asyncpg pool connected (min=%d, max=%d)", self._min_size, self._max_size
+        )
+
+    async def execute(self, query: str, *params: Any) -> list[dict[str, Any]]:
+        assert self._pool is not None, "call connect() first"
+        async with self._pool.acquire() as conn:
+            rows = await conn.fetch(query, *params)
+            return [dict(r) for r in rows]
+
+    async def execute_many(
+        self, query: str, params_list: list[tuple[Any, ...]]
+    ) -> None:
+        assert self._pool is not None, "call connect() first"
+        async with self._pool.acquire() as conn:
+            await conn.executemany(query, params_list)
+
+    async def fetchval(self, query: str, *params: Any) -> Any:
+        assert self._pool is not None, "call connect() first"
+        async with self._pool.acquire() as conn:
+            return await conn.fetchval(query, *params)
+
+    async def execute_in_transaction(
+        self, queries: list[tuple[str, tuple[Any, ...]]]
+    ) -> None:
+        """Execute multiple queries in a single transaction."""
+        assert self._pool is not None, "call connect() first"
+        async with self._pool.acquire() as conn, conn.transaction():
+            for query, params in queries:
+                await conn.execute(query, *params)
+
+    async def close(self) -> None:
+        if self._pool:
+            await self._pool.close()
+            self._pool = None
+            logger.info("asyncpg pool closed")

src/omnimarket/adapters/database.py

@@ -0,0 +1,27 @@
+"""DatabaseAdapter protocol for projection nodes."""
+
+from __future__ import annotations
+
+from typing import Any, Protocol
+
+
+class DatabaseAdapter(Protocol):
+    """Async database adapter for projection handlers."""
+
+    async def execute(self, query: str, *params: Any) -> list[dict[str, Any]]:
+        """Execute a parameterized query and return rows as dicts."""
+        ...
+
+    async def execute_many(
+        self, query: str, params_list: list[tuple[Any, ...]]
+    ) -> None:
+        """Execute a parameterized query for each set of params."""
+        ...
+
+    async def fetchval(self, query: str, *params: Any) -> Any:
+        """Execute a query and return a single value."""
+        ...
+
+    async def close(self) -> None:
+        """Close the connection pool."""
+        ...