Plan Phase 5 Brand Kit

.specify/feature.json

@@ -1 +1 @@
-{"feature_directory":"specs/005-signup-access"}
+{"feature_directory":"specs/006-brand-kit"}

backend/app/config.py

@@ -80,6 +80,10 @@ def get_engine() -> Engine:
         AND has_table_privilege(current_user, 'public.brand_asset_operations', 'INSERT')
         AND has_table_privilege(current_user, 'public.brand_asset_operations', 'UPDATE')
         AND has_table_privilege(current_user, 'public.brand_asset_operations', 'DELETE')
+        AND has_table_privilege(current_user, 'public.brand_kits', 'SELECT')
+        AND has_table_privilege(current_user, 'public.brand_kits', 'INSERT')
+        AND has_table_privilege(current_user, 'public.brand_kits', 'UPDATE')
+        AND has_table_privilege(current_user, 'public.brand_kits', 'DELETE')
         AS application_dml,
       has_schema_privilege(current_user, 'vault', 'USAGE') AS vault_schema_usage,
       has_function_privilege(

backend/app/main.py

@@ -14,6 +14,7 @@
 
 from .config import assert_database_role_privileges, load_settings
 from .routes.auth import router as auth_router
+from .routes.brand_kits import router as brand_kits_router
 from .routes.brands import router as brands_router
 from .routes.health import router as health_router
 from .routes.me import router as me_router
@@ -150,6 +151,7 @@ async def root() -> dict[str, str]:
 
 
 app.include_router(auth_router)
+app.include_router(brand_kits_router)
 app.include_router(brands_router)
 app.include_router(health_router)
 app.include_router(me_router)

backend/app/models/brand.py

@@ -28,6 +28,7 @@ class Brand(BaseModel):
     name: str
     logo_url: str | None
     cleanup_state: Literal["normal", "cleanup_required"] = "normal"
+    kit_status: Literal["not_started", "in_progress", "complete"] = "not_started"
     created_at: datetime
 
 

backend/app/models/brand_kit.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+import re
+from datetime import datetime
+from enum import Enum
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+
+_COLOR_PATTERN = re.compile(r"#[0-9A-Fa-f]{6}\Z")
+
+
+class KitStatus(str, Enum):
+    NOT_STARTED = "not_started"
+    IN_PROGRESS = "in_progress"
+    COMPLETE = "complete"
+
+
+class Tone(str, Enum):
+    FORMAL = "formal"
+    CASUAL = "casual"
+    PLAYFUL = "playful"
+    PROFESSIONAL = "professional"
+    FRIENDLY = "friendly"
+
+
+class BrandKitAnswers(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    tagline: str | None = None
+    tone: Tone | None = None
+    audience: str | None = None
+    colors: list[str] = Field(default_factory=list)
+    avoid_words: str | None = None
+
+    @field_validator("tagline", "audience", "avoid_words", mode="before")
+    @classmethod
+    def trim_nullable_text(cls, value: object) -> object:
+        if not isinstance(value, str):
+            return value
+        normalized = value.strip()
+        return normalized or None
+
+    @field_validator("tone", mode="before")
+    @classmethod
+    def clear_blank_tone(cls, value: object) -> object:
+        if isinstance(value, str):
+            normalized = value.strip()
+            return normalized or None
+        return value
+
+    @field_validator("tagline")
+    @classmethod
+    def validate_tagline(cls, value: str | None) -> str | None:
+        if value is not None and len(value) > 160:
+            raise ValueError("tagline must be 160 characters or fewer.")
+        return value
+
+    @field_validator("audience")
+    @classmethod
+    def validate_audience(cls, value: str | None) -> str | None:
+        if value is not None and not 2 <= len(value) <= 500:
+            raise ValueError("audience must be between 2 and 500 characters.")
+        return value
+
+    @field_validator("colors", mode="before")
+    @classmethod
+    def clear_null_colors(cls, value: object) -> object:
+        return [] if value is None else value
+
+    @field_validator("colors")
+    @classmethod
+    def normalize_colors(cls, value: list[str]) -> list[str]:
+        if len(value) > 3:
+            raise ValueError("colors must contain no more than 3 values.")
+
+        normalized = [color.strip().upper() for color in value]
+        if any(_COLOR_PATTERN.fullmatch(color) is None for color in normalized):
+            raise ValueError("colors must use the #RRGGBB format.")
+        return normalized
+
+
+class BrandKitUpsert(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    name: str
+    answers: BrandKitAnswers = Field(default_factory=BrandKitAnswers)
+
+    @field_validator("name")
+    @classmethod
+    def validate_name(cls, value: str) -> str:
+        normalized = value.strip()
+        if not 2 <= len(normalized) <= 120:
+            raise ValueError("name must be between 2 and 120 characters.")
+        return normalized
+
+
+class BrandKit(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    brand_id: UUID
+    brand_name: str
+    answers: BrandKitAnswers
+    summary: str | None
+    status: KitStatus
+    completed_at: datetime | None
+    updated_at: datetime | None

backend/app/routes/brand_kits.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import logging
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+
+from ..auth import CurrentUserDep
+from ..models.brand_kit import BrandKit, BrandKitUpsert
+from ..services.brand_kit_store import (
+    BrandKitStore,
+    get_brand_kit_store,
+)
+from ..services.brand_store import BrandCleanupRequiredError, BrandNameTakenError
+
+
+router = APIRouter(prefix="/api/v1/brands", tags=["brand-kits"])
+logger = logging.getLogger(__name__)
+
+BrandKitStoreDep = Annotated[BrandKitStore, Depends(get_brand_kit_store)]
+
+
+def _not_found() -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_404_NOT_FOUND,
+        detail={"code": "BRAND_NOT_FOUND", "message": "Brand not found."},
+    )
+
+
+def _cleanup_required() -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_409_CONFLICT,
+        detail={
+            "code": "BRAND_CLEANUP_REQUIRED",
+            "message": "Brand cleanup is required. Retry deletion.",
+        },
+    )
+
+
+def _name_taken() -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_409_CONFLICT,
+        detail={
+            "code": "BRAND_NAME_TAKEN",
+            "message": "You already have a brand with this name.",
+        },
+    )
+
+
+def _map_access_error(exc: Exception) -> HTTPException:
+    if isinstance(exc, BrandCleanupRequiredError):
+        return _cleanup_required()
+    return _not_found()
+
+
+@router.get("/{brand_id}/kit", response_model=BrandKit)
+def get_brand_kit(
+    request: Request,
+    brand_id: UUID,
+    current_user: CurrentUserDep,
+    brand_kit_store: BrandKitStoreDep,
+) -> BrandKit:
+    try:
+        kit = brand_kit_store.get_kit(current_user.user_id, brand_id)
+    except (LookupError, BrandCleanupRequiredError) as exc:
+        raise _map_access_error(exc) from exc
+
+    logger.info(
+        "brand_kits.get_success",
+        extra={
+            "event": "brand_kits.get_success",
+            "request_id": getattr(request.state, "request_id", "unknown"),
+        },
+    )
+    return kit
+
+
+@router.put("/{brand_id}/kit", response_model=BrandKit)
+def put_brand_kit(
+    request: Request,
+    brand_id: UUID,
+    payload: BrandKitUpsert,
+    current_user: CurrentUserDep,
+    brand_kit_store: BrandKitStoreDep,
+) -> BrandKit:
+    try:
+        kit = brand_kit_store.upsert_kit(current_user.user_id, brand_id, payload)
+    except BrandNameTakenError as exc:
+        raise _name_taken() from exc
+    except (LookupError, BrandCleanupRequiredError) as exc:
+        raise _map_access_error(exc) from exc
+
+    logger.info(
+        "brand_kits.put_success",
+        extra={
+            "event": "brand_kits.put_success",
+            "request_id": getattr(request.state, "request_id", "unknown"),
+        },
+    )
+    return kit