diff --git a/backend/app/api/main.py b/backend/app/api/main.py index bdb1178f5..a796c58b6 100644 --- a/backend/app/api/main.py +++ b/backend/app/api/main.py @@ -27,7 +27,6 @@ onboarding, openai_conversation, organization, - private, project, responses, threads, @@ -75,7 +74,6 @@ api_router.include_router(user_project.router) api_router.include_router(users.router) api_router.include_router(utils.router) -api_router.include_router(private.router) # if settings.ENVIRONMENT in ["development", "testing"]: # api_router.include_router(private.router) diff --git a/backend/app/api/routes/private.py b/backend/app/api/routes/private.py deleted file mode 100644 index 02c0c3855..000000000 --- a/backend/app/api/routes/private.py +++ /dev/null @@ -1,242 +0,0 @@ -import base64 -import logging -import time -from typing import Any - -from fastapi import APIRouter -from pydantic import BaseModel -from sqlmodel import col, func, select - -from app.api.deps import SessionDep -from app.core.cloud.storage import get_cloud_storage -from app.core.security import get_password_hash -from app.core.storage_utils import upload_audio_bytes_to_s3 -from app.core.util import now -from app.models import ( - LlmCall, - User, - UserPublic, -) - -logger = logging.getLogger(__name__) - -router = APIRouter(tags=["private"], prefix="/private") - - -class PrivateUserCreate(BaseModel): - email: str - password: str - full_name: str - is_verified: bool = False - - -MIGRATION_BATCH_SIZE = 50 -MIGRATION_LOG_INTERVAL = 100 - - -@router.post("/migrate/tts-base64-to-s3", include_in_schema=False) -def migrate_tts_base64_to_s3(session: SessionDep) -> dict: - """ - One-shot migration: find all llm_call rows with input_type=text / output_type=audio - whose content still holds raw base64, upload the audio to S3, and replace with a URI. - - Commits in batches so that partial progress is preserved on failure. - """ - fn = "migrate_tts_base64_to_s3" - start_time = time.monotonic() - - processed = skipped = failed = 0 - committed = 0 - pending_in_batch = 0 - errors: list[dict] = [] - - # Storage instances are cached per project_id to avoid redundant DB lookups. - storage_cache: dict[int, Any] = {} - - # --- count total candidates for progress logging --- - count_stmt = ( - select(func.count()) - .select_from(LlmCall) - .where( - LlmCall.input_type == "text", - LlmCall.output_type == "audio", - col(LlmCall.deleted_at).is_(None), - ) - ) - total_candidates = session.exec(count_stmt).one() - logger.info(f"[{fn}] Starting migration | total_candidates={total_candidates}") - - # Fetch all candidate IDs upfront to avoid server-side cursor invalidation - # on mid-loop commits. IDs are lightweight so this is safe for ~2k rows. - id_stmt = ( - select(LlmCall.id) - .where( - LlmCall.input_type == "text", - LlmCall.output_type == "audio", - col(LlmCall.deleted_at).is_(None), - ) - .order_by(col(LlmCall.inserted_at).desc()) - ) - candidate_ids = list(session.exec(id_stmt).all()) - logger.info(f"[{fn}] Fetched {len(candidate_ids)} candidate IDs") - - for idx, call_id in enumerate(candidate_ids, start=1): - call = session.get(LlmCall, call_id) - if call is None: - skipped += 1 - continue - content = call.content - if not content: - skipped += 1 - continue - - audio_content = content.get("content", {}) - if audio_content.get("format") != "base64": - skipped += 1 - continue - - b64_value = audio_content.get("value") - if not b64_value: - skipped += 1 - continue - - try: - if call.project_id not in storage_cache: - storage_cache[call.project_id] = get_cloud_storage( - session, call.project_id - ) - storage = storage_cache[call.project_id] - - audio_bytes = base64.b64decode(b64_value) - b64_size_kb = len(b64_value) / 1024 - audio_size_kb = len(audio_bytes) / 1024 - - prefix = f"orgs/{call.organization_id}/{call.project_id}/audio/tts" - s3_url = upload_audio_bytes_to_s3( - storage, - audio_bytes, - call.id, - audio_content.get("mime_type"), - prefix, - ) - - if not s3_url: - raise RuntimeError("upload_audio_bytes_to_s3 returned None") - - call.content = { - "type": "audio", - "content": { - "format": "uri", - "value": s3_url, - "mime_type": audio_content.get("mime_type"), - }, - } - call.updated_at = now() - session.add(call) - processed += 1 - pending_in_batch += 1 - - logger.debug( - f"[{fn}] Uploaded | call_id={call.id}, " - f"project_id={call.project_id}, " - f"b64_kb={b64_size_kb:.1f}, audio_kb={audio_size_kb:.1f}, " - f"s3_url={s3_url}" - ) - - except Exception as e: - failed += 1 - errors.append( - { - "call_id": str(call.id), - "project_id": str(call.project_id), - "error": str(e), - } - ) - logger.warning( - f"[{fn}] Row failed | call_id={call.id}, " - f"project_id={call.project_id}, error={e}", - exc_info=True, - ) - # Expunge the dirty object so the failed row doesn't poison the batch - session.expunge(call) - - # --- batch commit for partial progress --- - if pending_in_batch >= MIGRATION_BATCH_SIZE: - try: - session.commit() - committed += pending_in_batch - logger.info( - f"[{fn}] Batch committed | " - f"batch_size={pending_in_batch}, total_committed={committed}" - ) - except Exception as e: - logger.error( - f"[{fn}] Batch commit failed, rolling back | " - f"pending={pending_in_batch}, error={e}", - exc_info=True, - ) - session.rollback() - failed += pending_in_batch - processed -= pending_in_batch - pending_in_batch = 0 - - # --- periodic progress log --- - if idx % MIGRATION_LOG_INTERVAL == 0: - elapsed = time.monotonic() - start_time - logger.info( - f"[{fn}] Progress | " - f"scanned={idx}/{total_candidates}, " - f"processed={processed}, skipped={skipped}, failed={failed}, " - f"elapsed={elapsed:.1f}s" - ) - - # --- final batch --- - if pending_in_batch > 0: - try: - session.commit() - committed += pending_in_batch - logger.info( - f"[{fn}] Final batch committed | " - f"batch_size={pending_in_batch}, total_committed={committed}" - ) - except Exception as e: - logger.error( - f"[{fn}] Final batch commit failed, rolling back | " - f"pending={pending_in_batch}, error={e}", - exc_info=True, - ) - session.rollback() - failed += pending_in_batch - processed -= pending_in_batch - - elapsed = time.monotonic() - start_time - summary = { - "processed": processed, - "committed": committed, - "skipped": skipped, - "failed": failed, - "total_candidates": total_candidates, - "elapsed_seconds": round(elapsed, 2), - "errors": errors[:50], - } - logger.info(f"[{fn}] Migration complete | {summary}") - - return summary - - -@router.post("/users", response_model=UserPublic, include_in_schema=False) -def create_user(user_in: PrivateUserCreate, session: SessionDep) -> Any: - """ - Create a new user. - """ - - user = User( - email=user_in.email, - full_name=user_in.full_name, - hashed_password=get_password_hash(user_in.password), - ) - - session.add(user) - session.commit() - - return user diff --git a/backend/app/api/routes/utils.py b/backend/app/api/routes/utils.py index 0ee3dadfd..ffd9527a5 100644 --- a/backend/app/api/routes/utils.py +++ b/backend/app/api/routes/utils.py @@ -1,32 +1,8 @@ -from fastapi import APIRouter, Depends -from pydantic.networks import EmailStr - -from app.models import Message -from app.utils import generate_test_email, send_email -from app.api.permissions import Permission, require_permission +from fastapi import APIRouter router = APIRouter(prefix="/utils", tags=["utils"]) -@router.post( - "/test-email", - dependencies=[Depends(require_permission(Permission.SUPERUSER))], - status_code=201, - include_in_schema=False, -) -def test_email(email_to: EmailStr) -> Message: - """ - Test emails. - """ - email_data = generate_test_email(email_to=email_to) - send_email( - email_to=email_to, - subject=email_data.subject, - html_content=email_data.html_content, - ) - return Message(message="Test email sent") - - @router.get("/health", include_in_schema=False) async def health_check() -> bool: return True diff --git a/backend/app/tests/api/routes/test_private.py b/backend/app/tests/api/routes/test_private.py deleted file mode 100644 index 4c8380c0f..000000000 --- a/backend/app/tests/api/routes/test_private.py +++ /dev/null @@ -1,307 +0,0 @@ -import base64 -from types import SimpleNamespace -from unittest.mock import MagicMock, patch, call -from uuid import uuid4 - -from fastapi.testclient import TestClient -from sqlmodel import Session, select - -from app.core.config import settings -from app.models import User - - -# --------------------------------------------------------------------------- -# Existing user test (unchanged) -# --------------------------------------------------------------------------- - - -def test_create_user(client: TestClient, db: Session) -> None: - r = client.post( - f"{settings.API_V1_STR}/private/users", - json={ - "email": "pollo@listo.com", - "password": "password123", - "full_name": "Pollo Listo", - }, - ) - - assert r.status_code == 200 - - data = r.json() - - user = db.exec(select(User).where(User.id == data["id"])).first() - - assert user - assert user.email == "pollo@listo.com" - assert user.full_name == "Pollo Listo" - - -# --------------------------------------------------------------------------- -# Unit tests for migrate_tts_base64_to_s3 -# --------------------------------------------------------------------------- - -MODULE = "app.api.routes.private" -FAKE_AUDIO = b"\x00\x01\x02\x03audio-bytes" -FAKE_B64 = base64.b64encode(FAKE_AUDIO).decode() - - -def _fake_call( - content: dict | None = None, - project_id: int = 1, - organization_id: int = 10, -) -> SimpleNamespace: - """Lightweight stand-in for an LlmCall row.""" - return SimpleNamespace( - id=uuid4(), - project_id=project_id, - organization_id=organization_id, - content=content, - updated_at=None, - ) - - -def _b64_content(mime_type: str = "audio/mp3") -> dict: - return { - "type": "audio", - "content": {"format": "base64", "value": FAKE_B64, "mime_type": mime_type}, - } - - -def _uri_content() -> dict: - return { - "type": "audio", - "content": { - "format": "uri", - "value": "s3://bucket/existing.mp3", - "mime_type": "audio/mp3", - }, - } - - -def _mock_session(rows: list) -> MagicMock: - """Build a mock session whose .exec() returns count then IDs, - and .get() returns the corresponding row object.""" - session = MagicMock() - count_result = MagicMock() - count_result.one.return_value = len(rows) - - id_result = MagicMock() - id_result.all.return_value = [r.id for r in rows] - - # First exec → count, second → ID list - session.exec.side_effect = [count_result, id_result] - - # .get(LlmCall, call_id) → look up by id - row_map = {r.id: r for r in rows} - session.get.side_effect = lambda _model, call_id: row_map.get(call_id) - return session - - -@patch(f"{MODULE}.get_cloud_storage", return_value=MagicMock()) -@patch(f"{MODULE}.upload_audio_bytes_to_s3", return_value="s3://bucket/migrated.mp3") -def test_processes_base64_row(mock_upload: MagicMock, mock_storage: MagicMock) -> None: - """A row with base64 content is uploaded and rewritten to URI format.""" - from app.api.routes.private import migrate_tts_base64_to_s3 - - row = _fake_call(content=_b64_content()) - session = _mock_session([row]) - - result = migrate_tts_base64_to_s3(session) - - assert result["processed"] == 1 - assert result["failed"] == 0 - assert row.content["content"]["format"] == "uri" - assert row.content["content"]["value"] == "s3://bucket/migrated.mp3" - session.add.assert_called_once_with(row) - mock_upload.assert_called_once() - - -@patch(f"{MODULE}.get_cloud_storage", return_value=MagicMock()) -@patch(f"{MODULE}.upload_audio_bytes_to_s3", return_value="s3://bucket/migrated.mp3") -def test_skips_already_migrated_uri( - mock_upload: MagicMock, mock_storage: MagicMock -) -> None: - """Rows already in URI format are skipped.""" - from app.api.routes.private import migrate_tts_base64_to_s3 - - row = _fake_call(content=_uri_content()) - session = _mock_session([row]) - - result = migrate_tts_base64_to_s3(session) - - assert result["skipped"] == 1 - assert result["processed"] == 0 - mock_upload.assert_not_called() - - -@patch(f"{MODULE}.get_cloud_storage", return_value=MagicMock()) -@patch(f"{MODULE}.upload_audio_bytes_to_s3", return_value="s3://bucket/migrated.mp3") -def test_skips_null_content(mock_upload: MagicMock, mock_storage: MagicMock) -> None: - """Rows with None content are skipped.""" - from app.api.routes.private import migrate_tts_base64_to_s3 - - row = _fake_call(content=None) - session = _mock_session([row]) - - result = migrate_tts_base64_to_s3(session) - - assert result["skipped"] == 1 - assert result["processed"] == 0 - mock_upload.assert_not_called() - - -@patch(f"{MODULE}.get_cloud_storage", return_value=MagicMock()) -@patch(f"{MODULE}.upload_audio_bytes_to_s3", return_value=None) -def test_fails_when_upload_returns_none( - mock_upload: MagicMock, mock_storage: MagicMock -) -> None: - """upload returning None is recorded as a failure; original content is unchanged.""" - from app.api.routes.private import migrate_tts_base64_to_s3 - - original_content = _b64_content() - row = _fake_call(content=original_content) - session = _mock_session([row]) - - result = migrate_tts_base64_to_s3(session) - - assert result["failed"] == 1 - assert result["processed"] == 0 - assert any(e["call_id"] == str(row.id) for e in result["errors"]) - session.expunge.assert_called_once_with(row) - - -@patch(f"{MODULE}.get_cloud_storage", return_value=MagicMock()) -@patch(f"{MODULE}.upload_audio_bytes_to_s3", side_effect=RuntimeError("S3 timeout")) -def test_fails_on_upload_exception( - mock_upload: MagicMock, mock_storage: MagicMock -) -> None: - """An upload exception is caught and recorded in errors.""" - from app.api.routes.private import migrate_tts_base64_to_s3 - - row = _fake_call(content=_b64_content()) - session = _mock_session([row]) - - result = migrate_tts_base64_to_s3(session) - - assert result["failed"] == 1 - error = next(e for e in result["errors"] if e["call_id"] == str(row.id)) - assert "S3 timeout" in error["error"] - session.expunge.assert_called_once_with(row) - - -@patch(f"{MODULE}.get_cloud_storage", return_value=MagicMock()) -@patch(f"{MODULE}.upload_audio_bytes_to_s3", return_value="s3://bucket/out.mp3") -def test_uses_correct_s3_prefix( - mock_upload: MagicMock, mock_storage: MagicMock -) -> None: - """The prefix follows orgs/{org_id}/{project_id}/audio/tts.""" - from app.api.routes.private import migrate_tts_base64_to_s3 - - row = _fake_call(content=_b64_content(), project_id=42, organization_id=7) - session = _mock_session([row]) - - migrate_tts_base64_to_s3(session) - - args = mock_upload.call_args[0] - assert args[4] == "orgs/7/42/audio/tts" - - -@patch(f"{MODULE}.get_cloud_storage", return_value=MagicMock()) -@patch(f"{MODULE}.upload_audio_bytes_to_s3", return_value="s3://bucket/out.mp3") -def test_preserves_mime_type(mock_upload: MagicMock, mock_storage: MagicMock) -> None: - """The migrated content retains the original mime_type.""" - from app.api.routes.private import migrate_tts_base64_to_s3 - - row = _fake_call(content=_b64_content(mime_type="audio/wav")) - session = _mock_session([row]) - - migrate_tts_base64_to_s3(session) - - assert row.content["content"]["mime_type"] == "audio/wav" - - -@patch(f"{MODULE}.get_cloud_storage", return_value=MagicMock()) -@patch(f"{MODULE}.upload_audio_bytes_to_s3", return_value="s3://bucket/out.mp3") -def test_no_candidates(mock_upload: MagicMock, mock_storage: MagicMock) -> None: - """Zero rows means all counters are zero and no uploads happen.""" - from app.api.routes.private import migrate_tts_base64_to_s3 - - session = _mock_session([]) - - result = migrate_tts_base64_to_s3(session) - - assert result["processed"] == 0 - assert result["failed"] == 0 - assert result["committed"] == 0 - assert result["total_candidates"] == 0 - mock_upload.assert_not_called() - session.commit.assert_not_called() - - -@patch(f"{MODULE}.get_cloud_storage", return_value=MagicMock()) -@patch(f"{MODULE}.upload_audio_bytes_to_s3", return_value="s3://bucket/out.mp3") -def test_returns_all_summary_fields( - mock_upload: MagicMock, mock_storage: MagicMock -) -> None: - """The response dict contains every expected key.""" - from app.api.routes.private import migrate_tts_base64_to_s3 - - session = _mock_session([]) - - result = migrate_tts_base64_to_s3(session) - - for key in [ - "processed", - "committed", - "skipped", - "failed", - "total_candidates", - "elapsed_seconds", - "errors", - ]: - assert key in result, f"Missing key: {key}" - assert isinstance(result["elapsed_seconds"], (int, float)) - assert isinstance(result["errors"], list) - - -@patch(f"{MODULE}.get_cloud_storage", return_value=MagicMock()) -@patch(f"{MODULE}.upload_audio_bytes_to_s3", return_value="s3://bucket/out.mp3") -def test_mixed_rows(mock_upload: MagicMock, mock_storage: MagicMock) -> None: - """A mix of base64, URI, and null-content rows are handled correctly.""" - from app.api.routes.private import migrate_tts_base64_to_s3 - - rows = [ - _fake_call(content=_b64_content()), - _fake_call(content=_uri_content()), - _fake_call(content=None), - _fake_call(content=_b64_content(mime_type="audio/wav")), - ] - session = _mock_session(rows) - - result = migrate_tts_base64_to_s3(session) - - assert result["processed"] == 2 - assert result["skipped"] == 2 - assert result["failed"] == 0 - assert mock_upload.call_count == 2 - - -@patch(f"{MODULE}.get_cloud_storage", return_value=MagicMock()) -@patch(f"{MODULE}.upload_audio_bytes_to_s3", return_value="s3://bucket/out.mp3") -def test_caches_storage_per_project( - mock_upload: MagicMock, mock_storage: MagicMock -) -> None: - """get_cloud_storage is called once per unique project_id, not per row.""" - from app.api.routes.private import migrate_tts_base64_to_s3 - - rows = [ - _fake_call(content=_b64_content(), project_id=1), - _fake_call(content=_b64_content(), project_id=1), - _fake_call(content=_b64_content(), project_id=2), - ] - session = _mock_session(rows) - - migrate_tts_base64_to_s3(session) - - # Only 2 distinct project_ids → 2 calls - assert mock_storage.call_count == 2 diff --git a/docs/wiki/modules/tenancy.md b/docs/wiki/modules/tenancy.md index cc53cb496..917327480 100644 --- a/docs/wiki/modules/tenancy.md +++ b/docs/wiki/modules/tenancy.md @@ -10,7 +10,6 @@ All paths relative to `backend/app/`. - `api/routes/organization.py`, `api/routes/project.py` - `api/routes/api_keys.py` - `api/routes/onboarding.py` -- `api/routes/private.py` — internal endpoints ## Tables (SQLModel) | Table | Model |