Chore/refactor and rate limit

README.md

@@ -146,6 +146,14 @@ This can be introduced incrementally after Phase 2 or alongside Phase 3 without
 - CI deploy writes that per-folder `.env` before running `docker compose --env-file /srv/throwback-<env>/.env ...` on the VPS; reuse the same pattern for manual runs.
 - `REDIS_PASSWORD` is required for staging/prod; the dev `.env` ships with a default for local use only. The dev compose binds Redis to `127.0.0.1` and requires that password; avoid publishing Redis to the internet.
 - Auth config: set `AUTH_SECRET_KEY` (required for staging/prod) and optionally `ACCESS_TOKEN_TTL_MINUTES` (default 60).
+- Rate limiting and quotas: tune `REQUEST_RATE_LIMIT_PER_MINUTE` (default 60) and `DAILY_UPLOAD_LIMIT_BYTES` (default 1 GB) as needed per environment.
+
+## Redis usage
+
+- Request throttling via `fastapi-limiter`: per-user (or IP if unauthenticated) caps using `REQUEST_RATE_LIMIT_PER_MINUTE`.
+- Per-user daily upload quota using Redis counters keyed by date and user (`DAILY_UPLOAD_LIMIT_BYTES`).
+- Short-lived caching for album/asset list responses.
+- Celery broker and result backend.
 
 ---
 
@@ -172,6 +180,7 @@ This can be introduced incrementally after Phase 2 or alongside Phase 3 without
   mode; schema bootstrap honors the configured `tbm` schema.
 - Upload pipeline: assets land in MinIO (local) or S3-compatible buckets with
   Celery tasks ready for metadata and variants.
+- Basic protections: per-user/IP rate limiting (default 60 req/min) and per-user daily upload caps (default 1 GB) backed by Redis.
 - Auth baseline: JWT-based `/api/auth/signup` + `/api/auth/login` with bearer protection on `/api/albums` and `/api/assets`.
 - Local dev loop: `make db-up`, `make worker-up`, `make run`, `make migrate`.
 - Quality gates: `make fmt lint type test-single`.
@@ -190,6 +199,7 @@ This can be introduced incrementally after Phase 2 or alongside Phase 3 without
   prefixes; optional schema-per-tenant for stricter separation.
 - **Delivery polish:** Document fly/railway deploy scripts, add image tags per
   branch, and provide a smoke-test checklist post-deploy.
+- **Traffic shaping:** Tune rate limits/quotas per endpoint and surface counters/alerts in metrics.
 - **Security hardening (before user accounts/asset ACLs):**
   - Keep Redis/Postgres off public interfaces; require strong passwords and close host firewalls/SGs to 0.0.0.0 on 6379/5432.
   - Enforce HTTPS via proxy/CDN; set tight CORS and CSRF protections.

app/api/deps.py

@@ -2,7 +2,7 @@
 
 from typing import Annotated
 
-from fastapi import Depends, Header, HTTPException, status
+from fastapi import Depends, Header, HTTPException, Request, status
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.db import get_session
@@ -11,6 +11,7 @@
 
 
 async def get_current_user(
+    request: Request,
     authorization: Annotated[str | None, Header()] = None,
     db: AsyncSession = Depends(get_session),
 ) -> User:
@@ -21,4 +22,7 @@ async def get_current_user(
             headers={"WWW-Authenticate": "Bearer"},
         )
     token = authorization.split(" ", 1)[1].strip()
-    return await AuthService(db).validate_token(token)
+    user = await AuthService(db).validate_token(token)
+    request.state.user_id = user.id
+    request.state.user_email = user.email
+    return user

app/api/routers/assets.py

@@ -5,8 +5,10 @@
 from fastapi import APIRouter, Depends, File, Form, Query, Response, UploadFile
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.api import deps
 from app.constants import API_PREFIX
 from app.db import get_session
+from app.models.users import User
 from app.schemas.assets import AssetOut, AssetVariantOut, MediaType
 from app.services.assets import AssetService
 from app.settings import settings
@@ -20,11 +22,12 @@ async def upload_asset(
     media_type: Annotated[MediaType, Form(...)],
     file: Annotated[UploadFile, File(...)],
     response: Response,
+    current_user: User = Depends(deps.get_current_user),
     db: AsyncSession = Depends(get_session),
 ) -> AssetOut:
     svc = AssetService(db)
     out = await svc.create_from_upload(
-        album_id=album_id, media_type=media_type, file=file
+        album_id=album_id, media_type=media_type, file=file, user_id=current_user.id
     )
     response.headers["Location"] = settings.api_url(
         f"{API_PREFIX}/assets/{out.id}/variants"

app/main.py

@@ -1,8 +1,10 @@
 from fastapi import APIRouter, Depends, FastAPI
+from fastapi_limiter.depends import RateLimiter
 
 from app.api import deps
 from app.api.routers import albums, assets, auth, health
 from app.constants import API_PREFIX
+from app.rate_limiter import close_rate_limiter, init_rate_limiter
 from app.settings import settings
 
 servers = [{"url": settings.app_base_url}] if settings.app_base_url else None
@@ -15,14 +17,44 @@
 )
 
 public_api = APIRouter(prefix=API_PREFIX)
-public_api.include_router(auth.router)
+public_api.include_router(
+    auth.router,
+    dependencies=[
+        Depends(
+            RateLimiter(
+                times=settings.request_rate_limit_per_minute,
+                seconds=60,
+            )
+        )
+    ],
+)
 public_api.include_router(health.router)
 
 protected_api = APIRouter(
-    prefix=API_PREFIX, dependencies=[Depends(deps.get_current_user)]
+    prefix=API_PREFIX,
+    dependencies=[
+        Depends(deps.get_current_user),
+        Depends(
+            RateLimiter(
+                times=settings.request_rate_limit_per_minute,
+                seconds=60,
+            )
+        ),
+    ],
 )
 protected_api.include_router(albums.router)
 protected_api.include_router(assets.router)
 
 app.include_router(public_api)
 app.include_router(protected_api)
+
+
+@app.on_event("startup")
+async def _startup() -> None:
+    redis_client = getattr(app.state, "rate_limit_redis", None)
+    await init_rate_limiter(redis_client)
+
+
+@app.on_event("shutdown")
+async def _shutdown() -> None:
+    await close_rate_limiter()

app/rate_limiter.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import redis.asyncio as aioredis
+from fastapi import Request
+from fastapi_limiter import FastAPILimiter
+
+from app.settings import settings
+
+
+async def rate_limit_identifier(request: Request) -> str:
+    """
+    Prefer user id when available (set in get_current_user), otherwise fall
+    back to client IP. Keeps limits per-authenticated user across hosts.
+    """
+    user_id = getattr(request.state, "user_id", None)
+    if user_id:
+        return f"user:{user_id}"
+    if request.client and request.client.host:
+        return f"ip:{request.client.host}"
+    return "anonymous"
+
+
+async def init_rate_limiter(redis_client: aioredis.Redis | None = None) -> None:
+    """
+    Initialize FastAPI-Limiter with a Redis client. Uses decode_responses to
+    keep counters as strings for portability.
+    """
+    redis = redis_client or aioredis.from_url(
+        settings.redis_url, encoding="utf-8", decode_responses=True
+    )
+    await FastAPILimiter.init(redis, identifier=rate_limit_identifier)
+
+
+async def close_rate_limiter() -> None:
+    try:
+        await FastAPILimiter.close()
+    except Exception:
+        # best effort; don't crash shutdown
+        return

app/services/assets.py

@@ -6,6 +6,7 @@
 import mimetypes
 import subprocess
 import tempfile
+from datetime import datetime, time, timedelta, timezone
 from pathlib import Path
 from typing import Iterable, cast
 
@@ -93,7 +94,12 @@ def __init__(self, db: AsyncSession) -> None:
         self.repo = AssetRepository(db)
 
     async def create_from_upload(
-        self, *, album_id: int, media_type: MediaType, file: UploadFile
+        self,
+        *,
+        album_id: int,
+        media_type: MediaType,
+        file: UploadFile,
+        user_id: int | None = None,
     ) -> AssetOut:
         # 1) validate album exists
         album = await self.db.get(Album, album_id)
@@ -107,56 +113,77 @@ async def create_from_upload(
         if not data:
             raise HTTPException(status_code=400, detail="Empty upload")
 
+        quota_bytes = len(data)
+        quota_charged = False
+
+        if user_id is not None:
+            await self._enforce_daily_upload_quota(
+                user_id=user_id, bytes_added=quota_bytes
+            )
+            quota_charged = True
+
         # 3) persist original to storage
         ext = _ext_for(file.filename, file.content_type)
         storage_key = storage_key_for(
             asset_id=0, kind="original_tmp", ext=ext
         )  # placeholder; we’ll rename after insert
 
-        await write_bytes_async(
-            storage_key,
-            data,
-            content_type=file.content_type or "application/octet-stream",
-        )
-
-        # 3b) derive metadata
-        width: int | None = None
-        height: int | None = None
-        duration: int | None = None
-        if media_type == "photo":
-            width, height = _photo_dimensions(data)
-        elif media_type == "video":
-            width, height, duration = _video_metadata(data)
-
-        # 4) checksum for dedupe/idempotency (optional use)
-        checksum = sha256_bytes([data])
-
-        # 5) insert DB row (we’ll rename the storage key with the new asset id)
-        asset = await self.repo.create(
-            album_id=album_id,
-            media_type=media_type,
-            storage_key=storage_key,
-            mime_type=file.content_type or "application/octet-stream",
-            size_bytes=len(data),
-            original_filename=file.filename,
-            checksum_sha256=checksum,
-        )
-        asset.width = width
-        asset.height = height
-        asset.duration_sec = duration
-        # rename key to include real id (optional but nice)
-        final_key = storage_key_for(asset_id=asset.id, kind="original", ext=ext)
-        if final_key != storage_key:
-            await copy_bytes_async(
-                src_key=storage_key,
-                dst_key=final_key,
+        try:
+            await write_bytes_async(
+                storage_key,
+                data,
                 content_type=file.content_type or "application/octet-stream",
             )
-            storage_key = final_key
 
-        asset.storage_key = storage_key
-        await self.db.commit()
-        await self.db.refresh(asset)
+            # 3b) derive metadata
+            width: int | None = None
+            height: int | None = None
+            duration: int | None = None
+            if media_type == "photo":
+                width, height = _photo_dimensions(data)
+            elif media_type == "video":
+                width, height, duration = _video_metadata(data)
+
+            # 4) checksum for dedupe/idempotency (optional use)
+            checksum = sha256_bytes([data])
+
+            # 5) insert DB row (we’ll rename the storage key with the new asset id)
+            asset = await self.repo.create(
+                album_id=album_id,
+                media_type=media_type,
+                storage_key=storage_key,
+                mime_type=file.content_type or "application/octet-stream",
+                size_bytes=len(data),
+                original_filename=file.filename,
+                checksum_sha256=checksum,
+            )
+            asset.width = width
+            asset.height = height
+            asset.duration_sec = duration
+            # rename key to include real id (optional but nice)
+            final_key = storage_key_for(asset_id=asset.id, kind="original", ext=ext)
+            if final_key != storage_key:
+                await copy_bytes_async(
+                    src_key=storage_key,
+                    dst_key=final_key,
+                    content_type=file.content_type or "application/octet-stream",
+                )
+                storage_key = final_key
+
+            asset.storage_key = storage_key
+            await self.db.commit()
+            await self.db.refresh(asset)
+        except Exception as exc:
+            if quota_charged and user_id is not None:
+                await self._rollback_daily_upload_quota(
+                    user_id=user_id, bytes_removed=quota_bytes
+                )
+            if isinstance(exc, HTTPException):
+                raise
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Upload failed, please retry",
+            ) from exc
 
         # 6) bump album cache namespace and enqueue processing
         await bump(f"album:{album_id}")
@@ -264,3 +291,54 @@ async def list_variants(self, asset_id: int) -> list[AssetVariantOut]:
             )
             for v in row.variants
         ]
+
+    async def _enforce_daily_upload_quota(
+        self, *, user_id: int, bytes_added: int
+    ) -> None:
+        """
+        Enforce per-user daily upload quota using Redis counters. We increment
+        optimistically then roll back if the cap is exceeded.
+        """
+        key = self._upload_quota_key(user_id=user_id)
+        ttl = self._seconds_until_tomorrow()
+
+        from app.cache.redis import get_redis_async
+
+        r = get_redis_async()
+        current = await r.incrby(key, bytes_added)
+        # set TTL only when first created
+        if current == bytes_added:
+            await r.expire(key, ttl)
+
+        if current > settings.daily_upload_limit_bytes:
+            await r.incrby(key, -bytes_added)
+            raise HTTPException(
+                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+                detail="Daily upload limit exceeded",
+            )
+
+    async def _rollback_daily_upload_quota(
+        self, *, user_id: int, bytes_removed: int
+    ) -> None:
+        """
+        Best-effort rollback to keep the counter accurate when downstream work fails.
+        """
+        key = self._upload_quota_key(user_id=user_id)
+
+        from app.cache.redis import get_redis_async
+
+        r = get_redis_async()
+        new_value = await r.incrby(key, -bytes_removed)
+        if new_value <= 0:
+            await r.delete(key)
+
+    def _upload_quota_key(self, *, user_id: int) -> str:
+        today = datetime.now(timezone.utc).date().isoformat()
+        return f"user:{user_id}:upload_bytes:{today}"
+
+    def _seconds_until_tomorrow(self) -> int:
+        now = datetime.now(timezone.utc)
+        tomorrow = datetime.combine(
+            now.date() + timedelta(days=1), time(0, 0, tzinfo=timezone.utc)
+        )
+        return int((tomorrow - now).total_seconds())

app/settings.py

@@ -65,6 +65,12 @@ class Settings(BaseSettings):
     media_base_url: str | None = None
     jwt_secret: str = Field(default="dev-secret-change-me", alias="AUTH_SECRET_KEY")
     access_token_ttl_minutes: int = Field(default=60, alias="ACCESS_TOKEN_TTL_MINUTES")
+    request_rate_limit_per_minute: int = Field(
+        default=60, alias="REQUEST_RATE_LIMIT_PER_MINUTE"
+    )
+    daily_upload_limit_bytes: int = Field(
+        default=1_000_000_000, alias="DAILY_UPLOAD_LIMIT_BYTES"
+    )
 
     def model_post_init(self, __context: object) -> None:
         # Compute URLs from components when not explicitly provided.