diff --git a/alembic/versions/016_add_users_and_invite_codes.py b/alembic/versions/016_add_users_and_invite_codes.py
index 6c86023..1cec973 100644
--- a/alembic/versions/016_add_users_and_invite_codes.py
+++ b/alembic/versions/016_add_users_and_invite_codes.py
@@ -4,8 +4,6 @@ Revision ID: 016_add_users_and_invite_codes
 Revises: 015_add_creator_profile
 """
 from alembic import op
-import sqlalchemy as sa
-from sqlalchemy.dialects.postgresql import UUID
 
 revision = "016_add_users_and_invite_codes"
 down_revision = "015_add_creator_profile"
@@ -14,37 +12,41 @@ depends_on = None
 
 
 def upgrade() -> None:
-    # Create user_role enum type
-    user_role_enum = sa.Enum("creator", "admin", name="user_role", create_constraint=True)
-    user_role_enum.create(op.get_bind(), checkfirst=True)
+    # Use raw SQL to avoid SQLAlchemy's Enum double-creation bug with asyncpg
+    op.execute("""
+        DO $$ BEGIN
+            CREATE TYPE user_role AS ENUM ('creator', 'admin');
+        EXCEPTION WHEN duplicate_object THEN NULL;
+        END $$
+    """)
 
-    # Create users table
-    op.create_table(
-        "users",
-        sa.Column("id", UUID(as_uuid=True), primary_key=True, server_default=sa.func.gen_random_uuid()),
-        sa.Column("email", sa.String(255), nullable=False, unique=True),
-        sa.Column("hashed_password", sa.String(255), nullable=False),
-        sa.Column("display_name", sa.String(255), nullable=False),
-        sa.Column("role", user_role_enum, nullable=False, server_default="creator"),
-        sa.Column("creator_id", UUID(as_uuid=True), sa.ForeignKey("creators.id", ondelete="SET NULL"), nullable=True),
-        sa.Column("is_active", sa.Boolean(), nullable=False, server_default="true"),
-        sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
-        sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
-    )
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS users (
+            id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+            email VARCHAR(255) NOT NULL UNIQUE,
+            hashed_password VARCHAR(255) NOT NULL,
+            display_name VARCHAR(255) NOT NULL,
+            role user_role NOT NULL DEFAULT 'creator',
+            creator_id UUID REFERENCES creators(id) ON DELETE SET NULL,
+            is_active BOOLEAN NOT NULL DEFAULT TRUE,
+            created_at TIMESTAMP NOT NULL DEFAULT now(),
+            updated_at TIMESTAMP NOT NULL DEFAULT now()
+        )
+    """)
 
-    # Create invite_codes table
-    op.create_table(
-        "invite_codes",
-        sa.Column("id", UUID(as_uuid=True), primary_key=True, server_default=sa.func.gen_random_uuid()),
-        sa.Column("code", sa.String(100), nullable=False, unique=True),
-        sa.Column("uses_remaining", sa.Integer(), nullable=False, server_default="1"),
-        sa.Column("created_by", UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
-        sa.Column("expires_at", sa.DateTime(), nullable=True),
-        sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
-    )
+    op.execute("""
+        CREATE TABLE IF NOT EXISTS invite_codes (
+            id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+            code VARCHAR(100) NOT NULL UNIQUE,
+            uses_remaining INTEGER NOT NULL DEFAULT 1,
+            created_by UUID REFERENCES users(id) ON DELETE SET NULL,
+            expires_at TIMESTAMP,
+            created_at TIMESTAMP NOT NULL DEFAULT now()
+        )
+    """)
 
 
 def downgrade() -> None:
-    op.drop_table("invite_codes")
-    op.drop_table("users")
-    sa.Enum(name="user_role").drop(op.get_bind(), checkfirst=True)
+    op.execute("DROP TABLE IF EXISTS invite_codes")
+    op.execute("DROP TABLE IF EXISTS users")
+    op.execute("DROP TYPE IF EXISTS user_role")
diff --git a/backend/main.py b/backend/main.py
index 9aba2ef..c16f54c 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -12,7 +12,7 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
 from config import get_settings
-from routers import auth, consent, creators, health, ingest, pipeline, reports, search, stats, techniques, topics, videos
+from routers import auth, consent, creator_dashboard, creators, health, ingest, pipeline, reports, search, stats, techniques, topics, videos
 
 
 def _setup_logging() -> None:
@@ -80,6 +80,7 @@ app.include_router(health.router)
 # Versioned API
 app.include_router(auth.router, prefix="/api/v1")
 app.include_router(consent.router, prefix="/api/v1")
+app.include_router(creator_dashboard.router, prefix="/api/v1")
 app.include_router(creators.router, prefix="/api/v1")
 app.include_router(ingest.router, prefix="/api/v1")
 app.include_router(pipeline.router, prefix="/api/v1")
diff --git a/backend/routers/creator_dashboard.py b/backend/routers/creator_dashboard.py
new file mode 100644
index 0000000..2b89b27
--- /dev/null
+++ b/backend/routers/creator_dashboard.py
@@ -0,0 +1,162 @@
+"""Creator dashboard endpoint — authenticated analytics for a linked creator.
+
+Returns aggregate counts (videos, technique pages, key moments, search
+impressions) and content lists for the logged-in creator's dashboard.
+"""
+
+import logging
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from auth import get_current_user
+from database import get_session
+from models import (
+    Creator,
+    KeyMoment,
+    SearchLog,
+    SourceVideo,
+    TechniquePage,
+    User,
+)
+from schemas import (
+    CreatorDashboardResponse,
+    CreatorDashboardTechnique,
+    CreatorDashboardVideo,
+)
+
+logger = logging.getLogger("chrysopedia.creator_dashboard")
+
+router = APIRouter(prefix="/creator", tags=["creator-dashboard"])
+
+
+@router.get("/dashboard", response_model=CreatorDashboardResponse)
+async def get_creator_dashboard(
+    current_user: Annotated[User, Depends(get_current_user)],
+    db: AsyncSession = Depends(get_session),
+) -> CreatorDashboardResponse:
+    """Return dashboard analytics for the authenticated creator.
+
+    Requires the user to have a linked creator_id. Returns 404 if the
+    user has no linked creator profile.
+    """
+    if current_user.creator_id is None:
+        raise HTTPException(
+            status_code=404,
+            detail="No creator profile linked to this account",
+        )
+
+    creator_id = current_user.creator_id
+
+    # Verify creator exists (defensive — FK should guarantee this)
+    creator = (await db.execute(
+        select(Creator).where(Creator.id == creator_id)
+    )).scalar_one_or_none()
+    if creator is None:
+        logger.error("User %s has creator_id %s but creator row missing", current_user.id, creator_id)
+        raise HTTPException(
+            status_code=404,
+            detail="Linked creator profile not found",
+        )
+
+    # ── Aggregate counts ─────────────────────────────────────────────────
+
+    video_count = (await db.execute(
+        select(func.count()).select_from(SourceVideo)
+        .where(SourceVideo.creator_id == creator_id)
+    )).scalar() or 0
+
+    technique_count = (await db.execute(
+        select(func.count()).select_from(TechniquePage)
+        .where(TechniquePage.creator_id == creator_id)
+    )).scalar() or 0
+
+    key_moment_count = (await db.execute(
+        select(func.count()).select_from(KeyMoment)
+        .join(SourceVideo, KeyMoment.source_video_id == SourceVideo.id)
+        .where(SourceVideo.creator_id == creator_id)
+    )).scalar() or 0
+
+    # Search impressions: count distinct search_log rows where the query
+    # exactly matches (case-insensitive) any of this creator's technique titles.
+    search_impressions = (await db.execute(
+        select(func.count(func.distinct(SearchLog.id)))
+        .where(
+            select(TechniquePage.id)
+            .where(
+                TechniquePage.creator_id == creator_id,
+                func.lower(SearchLog.query) == func.lower(TechniquePage.title),
+            )
+            .correlate(SearchLog)
+            .exists()
+        )
+    )).scalar() or 0
+
+    # ── Content lists ────────────────────────────────────────────────────
+
+    # Techniques with per-page key moment count
+    km_count_sq = (
+        select(func.count(KeyMoment.id))
+        .where(KeyMoment.technique_page_id == TechniquePage.id)
+        .correlate(TechniquePage)
+        .scalar_subquery()
+        .label("key_moment_count")
+    )
+    technique_rows = (await db.execute(
+        select(
+            TechniquePage.title,
+            TechniquePage.slug,
+            TechniquePage.topic_category,
+            TechniquePage.created_at,
+            km_count_sq,
+        )
+        .where(TechniquePage.creator_id == creator_id)
+        .order_by(TechniquePage.created_at.desc())
+    )).all()
+
+    techniques = [
+        CreatorDashboardTechnique(
+            title=r.title,
+            slug=r.slug,
+            topic_category=r.topic_category,
+            created_at=r.created_at,
+            key_moment_count=r.key_moment_count or 0,
+        )
+        for r in technique_rows
+    ]
+
+    # Videos
+    video_rows = (await db.execute(
+        select(
+            SourceVideo.filename,
+            SourceVideo.processing_status,
+            SourceVideo.created_at,
+        )
+        .where(SourceVideo.creator_id == creator_id)
+        .order_by(SourceVideo.created_at.desc())
+    )).all()
+
+    videos = [
+        CreatorDashboardVideo(
+            filename=r.filename,
+            processing_status=r.processing_status.value if hasattr(r.processing_status, 'value') else str(r.processing_status),
+            created_at=r.created_at,
+        )
+        for r in video_rows
+    ]
+
+    logger.info(
+        "Dashboard loaded for creator %s: %d videos, %d techniques, %d moments, %d impressions",
+        creator_id, video_count, technique_count, key_moment_count, search_impressions,
+    )
+
+    return CreatorDashboardResponse(
+        video_count=video_count,
+        technique_count=technique_count,
+        key_moment_count=key_moment_count,
+        search_impressions=search_impressions,
+        techniques=techniques,
+        videos=videos,
+    )
diff --git a/backend/schemas.py b/backend/schemas.py
index 615938d..5add550 100644
--- a/backend/schemas.py
+++ b/backend/schemas.py
@@ -621,3 +621,39 @@ class ConsentSummary(BaseModel):
     kb_inclusion_granted: int = 0
     training_usage_granted: int = 0
     public_display_granted: int = 0
+
+
+# ── Creator Dashboard ────────────────────────────────────────────────────────
+
+class CreatorDashboardTechnique(BaseModel):
+    """Technique page summary for creator dashboard."""
+    title: str
+    slug: str
+    topic_category: str
+    created_at: datetime
+    key_moment_count: int = 0
+
+
+class CreatorDashboardVideo(BaseModel):
+    """Source video summary for creator dashboard."""
+    filename: str
+    processing_status: str
+    created_at: datetime
+
+
+class CreatorDashboardStats(BaseModel):
+    """Aggregate counts for dashboard header."""
+    video_count: int = 0
+    technique_count: int = 0
+    key_moment_count: int = 0
+    search_impressions: int = 0
+
+
+class CreatorDashboardResponse(BaseModel):
+    """Full creator dashboard payload."""
+    video_count: int = 0
+    technique_count: int = 0
+    key_moment_count: int = 0
+    search_impressions: int = 0
+    techniques: list[CreatorDashboardTechnique] = Field(default_factory=list)
+    videos: list[CreatorDashboardVideo] = Field(default_factory=list)