Feat/staged upload review

.env.example

@@ -41,6 +41,6 @@ totp_issuer=MultiAI
 
 GOOGLE_CLIENT_ID=
 GOOGLE_CLIENT_SECRET=
-GOOGLE_REDIRECT_URI=http://localhost:8000/staff/drive/callback
+GOOGLE_REDIRECT_URI=http://127.0.0.1:8000/stuff/drive/callback
 GOOGLE_OAUTH_SCOPES=https://www.googleapis.com/auth/drive.readonly openid email profile
 FACE_ENCRYPTION_KEY=hkbribvfirirbvivbibvib

.github/workflows/docker-publish.yml

@@ -38,22 +38,12 @@ jobs:
           username: ${{ github.actor }}
           password: ${{ secrets.GITHUB_TOKEN }}
 
-      - name: Extract metadata
-        id: meta
-        uses: docker/metadata-action@v5
-        with:
-          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
-          tags: |
-            type=raw,value=latest
-            type=sha,prefix=
-
       - name: Build and push
         uses: docker/build-push-action@v5
         with:
           context: .
           push: true
           platforms: linux/amd64,linux/arm64
-          tags: ${{ steps.meta.outputs.tags }}
-          labels: ${{ steps.meta.outputs.labels }}
-          cache-from: type=gha
-          cache-to: type=gha,mode=max
+          tags: |
+            ${{ env.IMAGE_NAME }}:latest
+            ${{ env.IMAGE_NAME }}:${{ github.sha }}

.gitignore

@@ -7,3 +7,5 @@ db/schema.sql
 multiai-c9380-firebase-adminsdk-fbsvc-cb6e5ce41b.json
 db.txt
 
+.venv
+multiai-c9380-firebase-adminsdk-fbsvc-cb6e5ce41b.json

app/container.py

@@ -17,11 +17,13 @@
 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
 from db.generated import staff_notifications as staff_notification_queries
 from db.generated import stuff_user as staff_user_queries
+from db.generated import upload_request_groups as upload_request_group_queries
 from db.generated import upload_request_photos as upload_request_photo_queries
 from db.generated import upload_requests as upload_request_queries
 from db.generated import user as user_queries
@@ -51,9 +53,11 @@ def __init__(
         self.device_querier = device_queries.AsyncQuerier(conn)
         self.staff_user_querier = staff_user_queries.AsyncQuerier(conn)
         self.staff_drive_querier = staff_drive_queries.AsyncQuerier(conn)
+        self.upload_request_group_querier = upload_request_group_queries.AsyncQuerier(conn)
         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)
@@ -94,6 +98,7 @@ def __init__(
         self.staged_upload_storage_service = StagedUploadStorageService()
 
         self.upload_requests_service = UploadRequestsService(
+            upload_request_group_querier=self.upload_request_group_querier,
             upload_request_querier=self.upload_request_querier,
             upload_request_photo_querier=self.upload_request_photo_querier,
             photo_querier=self.photo_querier,
@@ -115,7 +120,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
@@ -35,13 +40,22 @@ class Settings(BaseSettings):
     MOBILE_SESSION_LIMIT: int = 3
     MOBILE_SESSION_TTL_SECONDS: int = 180
     MOBILE_SESSION_DAYS: int = 7
-
+    # Admin list defaults
+    ADMIN_USERS_DEFAULT_LIMIT: int = 20
+    ADMIN_USERS_MAX_LIMIT: int = 100
     # Security
     jwt_secret: str
     jwt_algorithm: str = "HS256"
     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 = ""
@@ -53,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

@@ -9,6 +9,9 @@ class RedisKey(str, Enum):
 
 NOTIFICATION_EVENT_SUBJECT = "notification_event"
 AUDIT_EVENT_SUBJECT = "audit.event"
+FINAL_BUCKET_CLEANUP_SUBJECT = "ai.final_bucket.completed"
+FINAL_BUCKET_CLEANUP_STREAM = "ai-final-bucket-cleanup"
+FINAL_BUCKET_CLEANUP_DURABLE_NAME = "ai-final-bucket-cleaner"
 
 
 class AuditEventType(str, Enum):
@@ -20,14 +23,26 @@ class AuditEventType(str, Enum):
     UPLOAD_REQUEST_REJECTED = "upload_request.rejected"
 
 
-
 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/deps/token_auth.py

@@ -43,6 +43,8 @@ async def get_current_mobile_user(
     user = await container.auth_service.user_querier.get_user_by_id(id=session.user_id)
     if not user:
         raise HTTPException(status_code=401, detail="User not found")
+    if user.blocked:
+        raise HTTPException(status_code=403, detail="User is blocked")
 
     return MobileUserSchema(
         user_id=user.id,

app/infra/google_drive.py

@@ -9,12 +9,13 @@
 
 from app.core.exceptions import AppException
 from app.core.config import settings
-
-
-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}"
+from app.core.constant import (
+    GOOGLE_AUTH_URL,
+    GOOGLE_DRIVE_FILES_URL,
+    GOOGLE_TOKEN_URL,
+    GOOGLE_USERINFO_URL,
+)
+GOOGLE_DRIVE_LIST_FILES_URL = "https://www.googleapis.com/drive/v3/files"
 
 
 @dataclass
@@ -48,6 +49,8 @@ class GoogleDriveFileDownload:
 
 
 class GoogleDriveClient:
+    _drive_folder_mime_type = "application/vnd.google-apps.folder"
+
     @staticmethod
     def _require_str(data: dict[str, object], key: str) -> str:
         value = data.get(key)
@@ -181,6 +184,71 @@ async def download_file(
         )
         return GoogleDriveFileDownload(metadata=metadata, content=content)
 
+    @staticmethod
+    async def list_folder_files(
+        *,
+        access_token: str,
+        folder_id: str,
+    ) -> list[GoogleDriveFileMetadata]:
+        files: list[GoogleDriveFileMetadata] = []
+        next_page_token: str | None = None
+
+        while True:
+            query_params = {
+                "q": f"'{folder_id}' in parents and trashed = false",
+                "fields": "nextPageToken,files(id,name,mimeType,size)",
+                "supportsAllDrives": "true",
+                "includeItemsFromAllDrives": "true",
+                "pageSize": "100",
+            }
+            if next_page_token is not None:
+                query_params["pageToken"] = next_page_token
+
+            data = await GoogleDriveClient._get_json(
+                GOOGLE_DRIVE_LIST_FILES_URL,
+                headers={"Authorization": f"Bearer {access_token}"},
+                query_params=query_params,
+                error_context="Google Drive folder listing request",
+            )
+
+            raw_files = data.get("files", [])
+            if not isinstance(raw_files, list):
+                raise AppException.bad_request("Google Drive folder listing response is invalid")
+
+            for raw_file in raw_files:
+                if not isinstance(raw_file, dict):
+                    raise AppException.bad_request("Google Drive folder entry is invalid")
+                metadata = GoogleDriveClient._file_metadata_from_dict(raw_file)
+                if metadata.mime_type == GoogleDriveClient._drive_folder_mime_type:
+                    continue
+                files.append(metadata)
+
+            next_page_token_raw = data.get("nextPageToken")
+            if next_page_token_raw is None:
+                break
+            if not isinstance(next_page_token_raw, str) or not next_page_token_raw:
+                raise AppException.bad_request("Google Drive next page token is invalid")
+            next_page_token = next_page_token_raw
+
+        return files
+
+    @staticmethod
+    def _file_metadata_from_dict(data: dict[str, object]) -> GoogleDriveFileMetadata:
+        size_raw = data.get("size", "0")
+        if not isinstance(size_raw, (str, int)):
+            raise AppException.bad_request("Google Drive file size is invalid")
+        try:
+            size_bytes = int(size_raw)
+        except (TypeError, ValueError) as exc:
+            raise AppException.bad_request("Google Drive file size is invalid") from exc
+
+        return GoogleDriveFileMetadata(
+            id=GoogleDriveClient._require_str(data, "id"),
+            name=GoogleDriveClient._require_str(data, "name"),
+            mime_type=GoogleDriveClient._require_str(data, "mimeType"),
+            size_bytes=size_bytes,
+        )
+
     @staticmethod
     async def _post_form(url: str, payload: dict[str, str]) -> dict[str, object]:
         encoded = urllib.parse.urlencode(payload).encode("utf-8")

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}")