Feat/ai jetstream listener

app/container.py

@@ -17,6 +17,7 @@
 from app.service.users import AuthService
 from app.service.user_notification import UserNotificationService
 from db.generated import devices as device_queries
+from db.generated import photo_faces as photo_face_queries
 from db.generated import photos as photo_queries
 from db.generated import session as session_queries
 from db.generated import staff_drive_connections as staff_drive_queries
@@ -54,6 +55,7 @@ def __init__(
         self.upload_request_querier = upload_request_queries.AsyncQuerier(conn)
         self.upload_request_photo_querier = upload_request_photo_queries.AsyncQuerier(conn)
         self.photo_querier = photo_queries.AsyncQuerier(conn)
+        self.photo_face_querier = photo_face_queries.AsyncQuerier(conn)
         self.staff_notification_querier = staff_notification_queries.AsyncQuerier(conn)
         self.notification_querier = notification_queries.AsyncQuerier(conn)
         self.audit_querier = audit_queries.AsyncQuerier(conn)
@@ -115,7 +117,6 @@ def __init__(
         )
 
         self.staff_user_service = StaffUserService()
-
         self.staff_user_service.init(
             staff_user_querier=self.staff_user_querier,)
 

app/core/config.py

@@ -1,4 +1,5 @@
-from pydantic_settings import BaseSettings
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from pydantic import field_validator
 
 
 class Settings(BaseSettings):
@@ -16,13 +17,17 @@ class Settings(BaseSettings):
     NATS_HOST: str
     NATS_PASSWORD: str
     NATS_USER: str
+    NATS_SINGLE_FACE_MATCH_STREAM: str = "single_face_matches"
+    NATS_SINGLE_FACE_MATCH_DURABLE: str = "single_face_match_worker"
 
 
     # MinIO
     MINIO_API_PORT: int
     MINIO_ROOT_USER: str
     MINIO_ROOT_PASSWORD: str
     MINIO_HOST: str
+    MINIO_RETRY_ATTEMPTS: int = 3
+    MINIO_RETRY_BASE_SECONDS: float = 0.5
 
     # PostgreSQL
     POSTGRES_USER: str
@@ -44,6 +49,13 @@ class Settings(BaseSettings):
     encryption_key: str
     totp_issuer: str = "multAI"
 
+    # Face embedding model
+    FACE_EMBEDDING_MODEL_NAME: str = "buffalo_l"
+    FACE_EMBEDDING_PROVIDERS: str = "CPUExecutionProvider"
+    FACE_EMBEDDING_CTX_ID: int = -1
+    FACE_EMBEDDING_DET_WIDTH: int = 640
+    FACE_EMBEDDING_DET_HEIGHT: int = 640
+
     # Google Drive OAuth
     GOOGLE_CLIENT_ID: str = ""
     GOOGLE_CLIENT_SECRET: str = ""
@@ -55,9 +67,21 @@ class Settings(BaseSettings):
     FACE_ENCRYPTION_KEY: str
     FIREBASE_CREDENTIALS_PATH: str = "multiai-c9380-firebase-adminsdk-fbsvc-cb6e5ce41b.json"
 
-    class Config:
-        env_file = ".env"
-        extra = "ignore"
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        extra="ignore",
+    )
+
+    @field_validator("debug", mode="before")
+    @classmethod
+    def _parse_debug(cls, value):  # type: ignore[no-untyped-def]
+        if isinstance(value, str):
+            lowered = value.strip().lower()
+            if lowered in {"release", "prod", "production", "false", "0", "no"}:
+                return False
+            if lowered in {"true", "1", "yes"}:
+                return True
+        return value
 
 
 settings = Settings()  # type: ignore

app/core/constant.py

@@ -23,15 +23,26 @@ class AuditEventType(str, Enum):
     UPLOAD_REQUEST_REJECTED = "upload_request.rejected"
 
 
-    BlacklistedSession = "blacklist:session:{session_id}"
-
 IMAGE_ALLOWED_TYPES = {
     "image/jpeg",
     "image/png",
     "image/heic",
     "image/heif"
 }
 
+DEFAULT_CONTENT_TYPE = "application/octet-stream"
+DRIVE_ALLOWED_HOSTS = {"drive.google.com", "docs.google.com"}
+MINIO_URL_PREFIX = "minio://"
+
+IMAGES_BUCKET_NAME = "images"
+DOCUMENTS_BUCKET_NAME = "documents"
+WA_SIM_BUCKET_NAME = "wa-sim"
+
+GOOGLE_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
+GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
+GOOGLE_USERINFO_URL = "https://www.googleapis.com/oauth2/v2/userinfo"
+GOOGLE_DRIVE_FILES_URL = "https://www.googleapis.com/drive/v3/files/{file_id}"
+
 MAX_IMAGE_SIZE = 5 * 1024 * 1024
 MIN_ENROLL_IMAGES = 3
 MAX_ENROLL_IMAGES = 5

app/core/exceptions.py

@@ -75,6 +75,9 @@ def handle_check_violation(exc: Exception) -> HTTPException:
     def handle(exc: Exception) -> HTTPException:
         logger.error("Database error: %s", exc)
 
+        if isinstance(exc, HTTPException):
+            return exc
+
         if isinstance(exc, IntegrityError):
             orig = getattr(exc, "orig", None)
             sqlstate = getattr(orig, "sqlstate", None)

app/infra/google_drive.py

@@ -9,12 +9,14 @@
 
 from app.core.exceptions import AppException
 from app.core.config import settings
+from app.core.constant import (
+    GOOGLE_AUTH_URL,
+    GOOGLE_DRIVE_FILES_URL,
+    GOOGLE_TOKEN_URL,
+    GOOGLE_USERINFO_URL,
+)
 
 
-GOOGLE_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
-GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
-GOOGLE_USERINFO_URL = "https://www.googleapis.com/oauth2/v2/userinfo"
-GOOGLE_DRIVE_FILES_URL = "https://www.googleapis.com/drive/v3/files/{file_id}"
 
 
 @dataclass

app/infra/minio.py

@@ -9,11 +9,18 @@
 
 from app.core.utils import check_extension
 from app.core.exceptions import AppException
+from app.core.constant import (
+    DEFAULT_CONTENT_TYPE,
+    DOCUMENTS_BUCKET_NAME as CORE_DOCUMENTS_BUCKET_NAME,
+    IMAGES_BUCKET_NAME as CORE_IMAGES_BUCKET_NAME,
+    WA_SIM_BUCKET_NAME as CORE_WA_SIM_BUCKET_NAME,
+)
 
 
-IMAGES_BUCKET_NAME = "images"
-DOCUMENTS_BUCKET_NAME = "documents"
-WA_SIM_BUCKET_NAME = "wa-sim"
+# Re-export bucket names for compatibility with existing imports.
+IMAGES_BUCKET_NAME = CORE_IMAGES_BUCKET_NAME
+DOCUMENTS_BUCKET_NAME = CORE_DOCUMENTS_BUCKET_NAME
+WA_SIM_BUCKET_NAME = CORE_WA_SIM_BUCKET_NAME
 
 async def init_minio_client(
     minio_host: str, minio_port: int, minio_root_user: str, minio_root_password: str
@@ -48,7 +55,7 @@ async def put(self, file: UploadFile, object_name: str | None = None) -> str:
             object_name = str(uuid.uuid4())
 
         if file.content_type is None:
-            file.content_type = "application/octet-stream"
+            file.content_type = DEFAULT_CONTENT_TYPE
 
         if file.filename is None:
             file.filename = object_name
@@ -80,7 +87,7 @@ async def get(self, object_name: str) -> tuple[bytes, str, str]:
 
         data = await res.read()
         content_type = (
-            res.content_type if res.content_type else "application/octet-stream"
+            res.content_type if res.content_type else DEFAULT_CONTENT_TYPE
         )
         filename = res.headers.get("x-amz-meta-filename", f"{object_name}")
 

app/infra/nats.py

@@ -2,7 +2,8 @@
 from typing import Any, Callable, Optional
 from nats.aio.client import Client as NATS
 from nats.js.client import JetStreamContext
-from nats.js.api import DeliverPolicy, AckPolicy
+from nats.js.api import DeliverPolicy, AckPolicy, StreamConfig
+from nats.js.errors import NotFoundError
 from nats.aio.msg import Msg
 from pydantic import BaseModel
 
@@ -28,6 +29,7 @@ class NatsSubjects(Enum):
     STAFF_UPLOAD_REQUEST_CREATED = "staff.upload_request.created"
     STAFF_UPLOAD_REQUEST_APPROVED = "staff.upload_request.approved"
     STAFF_UPLOAD_REQUEST_REJECTED = "staff.upload_request.rejected"
+    SINGLE_FACE_MATCH_REQUESTED = "photo_faces.single.requested"
 
 class NatsClient:
     _nc: Optional[NATS] = None
@@ -102,6 +104,8 @@ async def js_subscribe(
         if NatsClient._js is None:
             await NatsClient.connect()
 
+        await NatsClient.ensure_stream(stream_name=stream_name, subjects=[subject.value])
+
         async def _wrapper(msg: Msg) -> None:
             await callback(msg.data)
             await msg.ack()
@@ -116,3 +120,20 @@ async def _wrapper(msg: Msg) -> None:
             deliver_policy=DeliverPolicy.NEW,
             # ack_policy=ack_policy
         )
+
+    @staticmethod
+    async def ensure_stream(*, stream_name: str, subjects: list[str]) -> None:
+        if NatsClient._js is None:
+            await NatsClient.connect()
+        js = NatsClient._js
+        assert js is not None
+        try:
+            await js.stream_info(stream_name)
+        except NotFoundError:
+            await js.add_stream(
+                name=stream_name,
+                config=StreamConfig(
+                    name=stream_name,
+                    subjects=subjects,
+                ),
+            )

app/router/mobile/enrollement.py

@@ -5,7 +5,13 @@
 from app.container import Container, get_container
 from app.deps.token_auth import MobileUserSchema, get_current_mobile_user
 from app.core.exceptions import AppException
-from app.core.constant import IMAGE_ALLOWED_TYPES, MAX_ENROLL_IMAGES, MAX_IMAGE_SIZE, MIN_ENROLL_IMAGES
+from app.core.constant import (
+    DEFAULT_CONTENT_TYPE,
+    IMAGE_ALLOWED_TYPES,
+    MAX_ENROLL_IMAGES,
+    MAX_IMAGE_SIZE,
+    MIN_ENROLL_IMAGES,
+)
 from app.service.face_embedding import FaceImagePayload
 from db.generated.models import User
 
@@ -57,7 +63,7 @@ async def enroll_face(
 
         payload: FaceImagePayload = FaceImagePayload(
             filename=file.filename or "unknown",
-            content_type=file.content_type or "application/octet-stream",
+            content_type=file.content_type or DEFAULT_CONTENT_TYPE,
             bytes=contents,
         )
 

app/router/web/users.py

@@ -10,7 +10,6 @@
 from app.schema.response.web.user import AdminUserSchema, to_admin_user_schema
 from db.generated.models import StaffUser
 
-
 router = APIRouter(prefix="/users")
 
 @router.post("/", response_model=AdminUserSchema, status_code=status.HTTP_201_CREATED)
@@ -28,7 +27,6 @@ async def create_user(
     logger.info("admin %s created user %s", current_staff_user.id, user.id)
     return to_admin_user_schema(user)
 
-
 @router.get("/", response_model=list[AdminUserSchema])
 async def list_users(
     limit: int = Query(

app/schema/dto/single_face_match.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from uuid import UUID, uuid4
+
+from pydantic import BaseModel, Field
+
+
+class BBoxPayload(BaseModel):
+    x1: float
+    y1: float
+    x2: float
+    y2: float
+
+
+class SingleFaceMatchJob(BaseModel):
+    job_id: UUID = Field(default_factory=uuid4)
+    photo_id: UUID
+    face_index: int = 0
+    image_ref: str
+    bbox: BBoxPayload | None = None
+    faces_detected: int | None = None
+    submitted_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+
+    model_config = {"extra": "allow"}

app/service/face_embedding.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
 import asyncio
+from dataclasses import dataclass
 from typing import List, Literal, Optional, Sequence, Tuple, TypedDict
 
 import cv2 # type: ignore
 import numpy as np
 from insightface.app import FaceAnalysis  # type: ignore[import-untyped]
+from app.core.config import settings
 from app.core.exceptions import AppException
 
 
@@ -27,18 +29,35 @@ class FaceStub:
     embedding: Optional[np.ndarray] = None
 
 
+@dataclass(frozen=True)
+class DetectedFace:
+    embedding: list[float]
+    bbox: Tuple[float, float, float, float]
+
+
 class FaceEmbedding:
     def __init__(
         self,
-        model_name: str = "buffalo_l",
-        providers: Sequence[str] = ("CPUExecutionProvider",),
-        ctx_id: int = -1,
-        det_size: Tuple[int, int] = (640, 640),
+        model_name: str | None = None,
+        providers: Sequence[str] | None = None,
+        ctx_id: int | None = None,
+        det_size: Tuple[int, int] | None = None,
     ) -> None:
         self.model: FaceAnalysis | None = None
-        self.model_name = model_name
+        self.model_name = model_name or settings.FACE_EMBEDDING_MODEL_NAME
+        if providers is None:
+            providers = tuple(
+                p.strip()
+                for p in settings.FACE_EMBEDDING_PROVIDERS.split(",")
+                if p.strip()
+            )
         self.providers = providers
-        self.ctx_id = ctx_id
+        self.ctx_id = settings.FACE_EMBEDDING_CTX_ID if ctx_id is None else ctx_id
+        if det_size is None:
+            det_size = (
+                settings.FACE_EMBEDDING_DET_WIDTH,
+                settings.FACE_EMBEDDING_DET_HEIGHT,
+            )
         self.det_size = det_size
         self._initialized = False
 
@@ -184,6 +203,26 @@ async def compute_event_embedding(
 
         return results
 
+    async def detect_faces(
+        self,
+        payload: FaceImagePayload,
+    ) -> list[DetectedFace]:
+        image = self._decode_image(payload)
+        image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
+
+        faces: list[FaceStub] = await asyncio.to_thread( # type: ignore
+            self.face_embedding.model.get, image_rgb  # type: ignore
+        )
+
+        detected: list[DetectedFace] = []
+        for face in faces:
+            if face.embedding is None:
+                continue
+            embedding = face.embedding.astype(float).flatten().tolist()
+            detected.append(DetectedFace(embedding=embedding, bbox=face.bbox))
+
+        return detected
+
     def _decode_image(self, payload: FaceImagePayload) -> np.ndarray:
 
         buffer = np.frombuffer(payload["bytes"], dtype=np.uint8)

app/service/session.py

@@ -22,6 +22,8 @@ class SessionService :
     def init(self, session: session_queries.AsyncQuerier, redis: RedisClient) -> None:
         self.session_querier = session
         self.redis = redis
+        SessionService.session_querier = session
+        SessionService.redis = redis
 
     @staticmethod
     async def create_session(user_id:uuid.UUID,device_id:uuid.UUID)->UpsertSessionRow: