added elevenlabs grant

.env.example

@@ -0,0 +1,49 @@
+# ---- App
+ENV=development
+FRONTEND_URL=http://localhost:5173
+API_URL=http://localhost:8000
+CORS_ORIGINS=http://localhost:5173
+JWT_SECRET=replace-with-64-random-chars
+INTERNAL_TOKEN=replace-with-random-token
+BILLING_ENABLED=false
+
+# ---- Google sign in (OAuth client, type Web; add http://localhost:5173 to authorized JS origins)
+GOOGLE_CLIENT_ID=
+
+# ---- Email (Resend). EMAIL_PROVIDER=console prints emails to the API log instead.
+EMAIL_PROVIDER=console
+RESEND_API_KEY=
+FROM_EMAIL=Notestack <hello@notestack.ai>
+ALERTS_EMAIL=arslan@firebird-technologies.com
+UNSUBSCRIBE_SECRET=replace-me
+
+# ---- Storage: Cloudflare R2
+# Prod: set account id + keys, leave R2_ENDPOINT_URL empty.
+# Local: docker-compose points R2_ENDPOINT_URL at MinIO (keys minioadmin/minioadmin).
+R2_ACCOUNT_ID=
+R2_ACCESS_KEY_ID=minioadmin
+R2_SECRET_ACCESS_KEY=minioadmin
+R2_BUCKET=notestack-assets
+R2_ENDPOINT_URL=
+R2_PUBLIC_BASE_URL=
+
+# ---- LLM (any LiteLLM model string). Defaults target Z.ai GLM-5.3 via its OpenAI compatible API.
+# GLM-5.x always reasons; LLM_REASONING_EFFORT (low | high | max) trades cost for depth.
+# Cheaper: LLM_MODEL=openai/glm-5.3-flash. Free for local dev: openai/glm-4.7-flash
+LLM_MODEL=openai/glm-5.3
+LLM_FAST_MODEL=openai/glm-5.3-flash
+LLM_API_BASE=https://api.z.ai/api/paas/v4
+LLM_API_KEY=
+LLM_TEMPERATURE=1.0
+LLM_MAX_TOKENS=16000
+LLM_REASONING_EFFORT=low
+
+# ---- Embeddings (separate provider: Z.ai's international API has no embedding model)
+# Must support a `dimensions` parameter so the pgvector column stays EMBEDDING_DIM wide.
+EMBEDDING_MODEL=openai/text-embedding-3-small
+EMBEDDING_DIM=1024
+EMBEDDING_API_BASE=
+EMBEDDING_API_KEY=
+
+# ---- ElevenLabs (Phase 1 audio)
+ELEVENLABS_API_KEY=

.github/workflows/ci.yml

@@ -0,0 +1,35 @@
+name: CI
+on: [push, pull_request]
+
+jobs:
+  backend:
+    runs-on: ubuntu-latest
+    defaults: { run: { working-directory: backend } }
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with: { python-version: "3.12", cache: pip }
+      - run: pip install -r requirements-dev.txt
+      - run: ruff check app tests
+      - run: pytest -q
+
+  frontend:
+    runs-on: ubuntu-latest
+    defaults: { run: { working-directory: frontend } }
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with: { node-version: 22, cache: npm, cache-dependency-path: frontend/package-lock.json }
+      - run: npm ci
+      - run: npm run check:copy
+      - run: npm run build
+
+  renderer:
+    runs-on: ubuntu-latest
+    defaults: { run: { working-directory: renderer } }
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with: { node-version: 22, cache: npm, cache-dependency-path: renderer/package-lock.json }
+      - run: npm ci
+      - run: npm run typecheck

.gitignore

@@ -0,0 +1,10 @@
+.env
+node_modules/
+dist/
+.venv/
+__pycache__/
+*.pyc
+.pytest_cache/
+.ruff_cache/
+out/
+renderer/build/

README.md

@@ -0,0 +1,70 @@
+# Notestack
+
+Your archive, in orbit. Turn a Substack (or any feed) into a grounded research notebook, audio
+overviews, videos and launch kits, all traced back to what the writer actually wrote.
+
+- Design doc: [docs/DESIGN.md](docs/DESIGN.md)
+- Phase 0 task list: [docs/PHASE0.md](docs/PHASE0.md)
+
+## Layout
+
+```
+backend/    FastAPI + SQLAlchemy + Alembic, arq worker, DSPy (LLM agnostic, Z.ai first)
+frontend/   React + Vite + TS: night sky landing, pricing, blog, auth, Mission Control
+renderer/   Remotion compositions + render service that uploads to R2 via presigned PUT
+docs/       Design doc and task lists
+```
+
+## Run locally
+
+```bash
+cp .env.example .env            # fill LLM_API_KEY (Z.ai), EMBEDDING_API_KEY, GOOGLE_CLIENT_ID, RESEND_API_KEY
+docker compose up --build
+```
+
+- App: http://localhost:5173, API: http://localhost:8000/docs, MinIO console: http://localhost:9001
+- Local storage is MinIO standing in for R2. Presigned URLs point at `http://minio:9000`, so add
+  `127.0.0.1 minio` to your hosts file for browser uploads to work locally.
+- With `EMAIL_PROVIDER=console`, verification codes print in the `api` logs.
+
+Without Docker:
+
+```bash
+cd backend && python -m venv .venv && .venv/Scripts/pip install -r requirements-dev.txt
+alembic upgrade head && uvicorn app.main:app --reload     # needs Postgres (pgvector) + Redis
+arq app.worker.WorkerSettings                              # second terminal
+cd frontend && npm install && npm run dev
+cd renderer && npm install && npm start                    # or: npm run studio
+```
+
+## Key choices
+
+| Concern | Where |
+| --- | --- |
+| Cloudflare R2 storage (keys, presign, uploads) | `backend/app/services/storage.py`, `backend/app/routers/storage.py` |
+| Resend email + unsubscribe + campaigns | `backend/app/services/email.py`, `backend/app/worker.py` |
+| Google + email code auth, JWT revocation | `backend/app/auth.py`, `backend/app/routers/auth.py` |
+| LLM provider (swap by env) | `backend/app/llm/provider.py`, `.env` `LLM_*` |
+| Three plans, billing off | `backend/app/services/plans.py`, `BILLING_ENABLED` |
+| Blog posts | `frontend/src/content/blogPosts.ts` (`/blogs`, `/blogs/:slug`) |
+| Logo | replace `frontend/public/logo.svg` (white tile, black silhouettes) |
+| Sky background | `frontend/src/components/SkyCanvas.tsx` |
+
+## Switching LLM providers
+
+```bash
+# Z.ai (default). Cheaper: glm-5.3-flash. Free for dev: glm-4.7-flash
+LLM_MODEL=openai/glm-5.3  LLM_API_BASE=https://api.z.ai/api/paas/v4  LLM_REASONING_EFFORT=low
+# Anthropic
+LLM_MODEL=anthropic/claude-sonnet-5  LLM_API_BASE=
+# OpenAI
+LLM_MODEL=openai/gpt-5  LLM_API_BASE=
+```
+
+Changing `EMBEDDING_DIM` needs a new migration and a re-embed.
+
+## Rules
+
+- No em dashes in any UI, email or generated copy. `npm run check:copy` and
+  `app/llm/postprocess.py` enforce it.
+- Palette is black, `#217cff` and white only.

backend/Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3.12-slim
+WORKDIR /app
+ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY . .
+CMD ["sh", "-c", "alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port 8000"]

backend/alembic.ini

@@ -0,0 +1,28 @@
+[alembic]
+script_location = alembic
+prepend_sys_path = .
+
+[loggers]
+keys = root,sqlalchemy,alembic
+[handlers]
+keys = console
+[formatters]
+keys = generic
+[logger_root]
+level = WARN
+handlers = console
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s

backend/alembic/env.py

@@ -0,0 +1,35 @@
+from logging.config import fileConfig
+
+from alembic import context
+from sqlalchemy import engine_from_config, pool
+
+import app.models  # noqa: F401  (register tables)
+from app.config import settings
+from app.db import Base
+
+config = context.config
+config.set_main_option("sqlalchemy.url", settings.database_url)
+if config.config_file_name:
+    fileConfig(config.config_file_name)
+target_metadata = Base.metadata
+
+
+def run_migrations_offline() -> None:
+    context.configure(url=settings.database_url, target_metadata=target_metadata, literal_binds=True)
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online() -> None:
+    connectable = engine_from_config(config.get_section(config.config_ini_section), prefix="sqlalchemy.",
+                                     poolclass=pool.NullPool)
+    with connectable.connect() as connection:
+        context.configure(connection=connection, target_metadata=target_metadata)
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

backend/alembic/script.py.mako

@@ -0,0 +1,22 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+"""
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+revision = ${repr(up_revision)}
+down_revision = ${repr(down_revision)}
+branch_labels = ${repr(branch_labels)}
+depends_on = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}

backend/alembic/versions/0001_initial.py

@@ -0,0 +1,35 @@
+"""initial schema: pgvector + all Phase 0 tables
+
+Initial revision builds from the models so schema and code start identical. Every later change
+must be a normal autogenerated revision (alembic revision --autogenerate -m "...").
+
+Revision ID: 0001_initial
+Revises:
+Create Date: 2026-09-22
+"""
+from alembic import op
+
+import app.models  # noqa: F401
+from app.config import settings
+from app.db import Base
+
+revision = "0001_initial"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute("CREATE EXTENSION IF NOT EXISTS vector")
+    bind = op.get_bind()
+    Base.metadata.create_all(bind=bind)
+    # HNSW index for cosine search over chunk embeddings (pgvector supports HNSW up to 2000 dims).
+    if settings.embedding_dim <= 2000:
+        op.execute(
+            "CREATE INDEX IF NOT EXISTS ix_chunks_embedding_hnsw ON chunks "
+            "USING hnsw (embedding vector_cosine_ops)"
+        )
+
+
+def downgrade() -> None:
+    Base.metadata.drop_all(bind=op.get_bind())

backend/app/auth.py

@@ -0,0 +1,87 @@
+"""JWT helpers. Revocation = bump User.token_version; stale tokens get the same 401 as expired ones."""
+
+import uuid
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+
+import jwt
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.config import settings
+from app.db import get_db
+from app.models import User, Workspace, WorkspaceMember
+
+_bearer = HTTPBearer(auto_error=False)
+_UNAUTHORIZED = HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid or expired token")
+
+
+def _encode(user_id: uuid.UUID, token_version: int, kind: str, ttl: timedelta) -> str:
+    now = datetime.now(UTC)
+    payload = {"sub": str(user_id), "tv": token_version, "typ": kind, "iat": now, "exp": now + ttl}
+    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
+
+
+def create_access_token(user_id: uuid.UUID, token_version: int) -> str:
+    return _encode(user_id, token_version, "access", timedelta(hours=settings.jwt_expiration_hours))
+
+
+def create_refresh_token(user_id: uuid.UUID, token_version: int) -> str:
+    return _encode(user_id, token_version, "refresh", timedelta(days=settings.jwt_refresh_expiration_days))
+
+
+def decode_token_full(token: str, expected_type: str = "access") -> dict:
+    try:
+        payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
+    except jwt.PyJWTError as exc:
+        raise _UNAUTHORIZED from exc
+    if payload.get("typ") != expected_type:
+        raise _UNAUTHORIZED
+    return payload
+
+
+def token_version_is_current(user: User, payload: dict) -> bool:
+    return payload.get("tv") == user.token_version
+
+
+def load_user_from_token(db: Session, token: str, expected_type: str = "access") -> User:
+    payload = decode_token_full(token, expected_type)
+    try:
+        user_id = uuid.UUID(payload["sub"])
+    except (KeyError, ValueError) as exc:
+        raise _UNAUTHORIZED from exc
+    user = db.get(User, user_id)
+    if not user or not user.is_active or user.deleted_at or not token_version_is_current(user, payload):
+        raise _UNAUTHORIZED
+    return user
+
+
+def get_current_user(
+    creds: HTTPAuthorizationCredentials | None = Depends(_bearer), db: Session = Depends(get_db)
+) -> User:
+    if not creds:
+        raise _UNAUTHORIZED
+    return load_user_from_token(db, creds.credentials)
+
+
+@dataclass
+class Ctx:
+    user: User
+    workspace: Workspace
+    db: Session
+
+
+def get_ctx(user: User = Depends(get_current_user), db: Session = Depends(get_db)) -> Ctx:
+    """Current user plus their active workspace (first membership for now; team switching in Phase 4)."""
+    workspace = db.scalar(
+        select(Workspace)
+        .join(WorkspaceMember, WorkspaceMember.workspace_id == Workspace.id)
+        .where(WorkspaceMember.user_id == user.id)
+        .order_by(WorkspaceMember.created_at)
+        .limit(1)
+    )
+    if not workspace:
+        raise HTTPException(status.HTTP_409_CONFLICT, "No workspace for user")
+    return Ctx(user=user, workspace=workspace, db=db)

backend/app/config.py

@@ -0,0 +1,87 @@
+from functools import lru_cache
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+
+    env: str = "development"
+    app_name: str = "Notestack"
+    frontend_url: str = "http://localhost:5173"
+    api_url: str = "http://localhost:8000"
+    cors_origins: str = "http://localhost:5173"
+
+    database_url: str = "postgresql+psycopg://notestack:notestack@localhost:5432/notestack"
+    redis_url: str = "redis://localhost:6379/0"
+
+    # Auth
+    jwt_secret: str = "change-me"
+    jwt_algorithm: str = "HS256"
+    jwt_expiration_hours: int = 72
+    jwt_refresh_expiration_days: int = 30
+    google_client_id: str = ""
+
+    # Email (Resend)
+    email_provider: str = "console"  # resend | console
+    resend_api_key: str = ""
+    from_email: str = "Notestack <hello@notestack.ai>"
+    alerts_email: str = "arslan@firebird-technologies.com"
+    unsubscribe_secret: str = "change-me-too"
+    update_email_send_hour: int = 9
+
+    # Storage (Cloudflare R2, S3 compatible)
+    r2_account_id: str = ""
+    r2_access_key_id: str = ""
+    r2_secret_access_key: str = ""
+    r2_bucket: str = "notestack-assets"
+    r2_endpoint_url: str = ""  # override for MinIO locally; defaults to the R2 account endpoint
+    r2_public_base_url: str = ""  # optional custom domain for public objects
+    r2_presign_ttl_seconds: int = 3600
+    max_upload_bytes: int = 200 * 1024 * 1024
+
+    # LLM (LiteLLM model strings; Z.ai GLM by default)
+    llm_model: str = "openai/glm-5.3"
+    llm_fast_model: str = "openai/glm-5.3-flash"
+    llm_api_base: str = "https://api.z.ai/api/paas/v4"
+    llm_api_key: str = ""
+    llm_temperature: float = 1.0  # Z.ai recommends 1.0 for GLM-5.x
+    # Reasoning output counts against max_tokens, so leave headroom for the structured answer.
+    llm_max_tokens: int = 16000
+    # GLM-5.x always reasons (it cannot be disabled); this sets how hard. low | high | max
+    llm_reasoning_effort: str = "low"
+
+    # Embeddings are a separate provider: Z.ai's international API has no embedding model.
+    # LiteLLM picks up OPENAI_API_KEY etc. from env when embedding_api_key is empty.
+    embedding_model: str = "openai/text-embedding-3-small"
+    embedding_api_base: str = ""
+    embedding_api_key: str = ""
+    embedding_dim: int = 1024
+
+    # ElevenLabs
+    elevenlabs_api_key: str = ""
+
+    # Renderer
+    renderer_url: str = "http://localhost:3100"
+    internal_token: str = "internal-change-me"
+
+    # Billing
+    billing_enabled: bool = False
+
+    @property
+    def r2_endpoint(self) -> str:
+        if self.r2_endpoint_url:
+            return self.r2_endpoint_url
+        return f"https://{self.r2_account_id}.r2.cloudflarestorage.com"
+
+    @property
+    def cors_origin_list(self) -> list[str]:
+        return [o.strip() for o in self.cors_origins.split(",") if o.strip()]
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()
+
+
+settings = get_settings()

backend/app/db.py

@@ -0,0 +1,23 @@
+from collections.abc import Iterator
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
+
+from app.config import settings
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+_connect_args = {"check_same_thread": False} if settings.database_url.startswith("sqlite") else {}
+engine = create_engine(settings.database_url, pool_pre_ping=True, connect_args=_connect_args)
+SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
+
+
+def get_db() -> Iterator[Session]:
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

backend/app/llm/embeddings.py

@@ -0,0 +1,46 @@
+"""Embeddings through LiteLLM. Always requests EMBEDDING_DIM so the pgvector column stays fixed."""
+
+from dataclasses import dataclass
+
+import litellm
+
+from app.config import settings
+
+BATCH = 64
+
+
+@dataclass
+class EmbeddingResult:
+    vectors: list[list[float]]
+    tokens: int
+
+
+def embed_texts(texts: list[str]) -> EmbeddingResult:
+    vectors: list[list[float]] = []
+    tokens = 0
+    for i in range(0, len(texts), BATCH):
+        batch = [t[:8000] for t in texts[i : i + BATCH]]
+        resp = litellm.embedding(
+            model=settings.embedding_model,
+            input=batch,
+            api_base=settings.embedding_api_base or None,
+            api_key=settings.embedding_api_key or None,
+            dimensions=settings.embedding_dim,
+        )
+        vectors.extend(item["embedding"] for item in resp.data)
+        tokens += int(getattr(resp, "usage", None) and resp.usage.prompt_tokens or 0)
+    for v in vectors:
+        if len(v) != settings.embedding_dim:
+            raise ValueError(
+                f"Embedding model returned {len(v)} dims, expected {settings.embedding_dim}. "
+                "Check EMBEDDING_MODEL supports the dimensions parameter."
+            )
+    return EmbeddingResult(vectors=vectors, tokens=tokens)
+
+
+def embedding_provider() -> str:
+    return settings.embedding_model.split("/", 1)[0]
+
+
+def embed_query(text: str) -> list[float]:
+    return embed_texts([text]).vectors[0]

backend/app/llm/postprocess.py

@@ -0,0 +1,23 @@
+"""Last pass on every generated string. Non-negotiable: no em dashes in generated copy."""
+
+import re
+
+_SPACED_DASH = re.compile(r"\s*[—―]\s*")
+_EN_DASH_SPACED = re.compile(r"\s+–\s+")
+
+
+def strip_em_dashes(text: str) -> str:
+    text = _SPACED_DASH.sub(", ", text)
+    text = _EN_DASH_SPACED.sub(", ", text)
+    return text.replace(",,", ",").replace(" ,", ",")
+
+
+def clean_output(value):
+    """Recursively clean strings inside dicts/lists/pydantic dumps."""
+    if isinstance(value, str):
+        return strip_em_dashes(value)
+    if isinstance(value, list):
+        return [clean_output(v) for v in value]
+    if isinstance(value, dict):
+        return {k: clean_output(v) for k, v in value.items()}
+    return value

backend/app/llm/provider.py

@@ -0,0 +1,87 @@
+"""Provider agnostic LLM factory. Any LiteLLM model string works; Z.ai GLM is the default.
+
+Examples:
+  Z.ai:       LLM_MODEL=openai/glm-5.3  LLM_API_BASE=https://api.z.ai/api/paas/v4
+  Anthropic:  LLM_MODEL=anthropic/claude-sonnet-5  LLM_API_BASE=
+  OpenAI:     LLM_MODEL=openai/gpt-5  LLM_API_BASE=
+  Local:      LLM_MODEL=ollama_chat/qwen3  LLM_API_BASE=http://localhost:11434
+"""
+
+from contextlib import contextmanager
+from dataclasses import dataclass
+from functools import lru_cache
+
+import dspy
+
+from app.config import settings
+
+
+def is_zai() -> bool:
+    return "z.ai" in settings.llm_api_base or "bigmodel.cn" in settings.llm_api_base
+
+
+def _extra_kwargs(effort: str | None) -> dict:
+    if not is_zai():
+        return {}
+    # GLM-5.x always reasons; thinking.type only accepts "enabled". Cost and latency are tuned
+    # with reasoning_effort (low | high | max) instead. extra_body fields land top level in the request.
+    effort = effort or settings.llm_reasoning_effort
+    return {"extra_body": {"thinking": {"type": "enabled"}, "reasoning_effort": effort}}
+
+
+def _build(model: str, *, effort: str | None, temperature: float | None, max_tokens: int | None) -> dspy.LM:
+    return dspy.LM(
+        model,
+        api_base=settings.llm_api_base or None,
+        api_key=settings.llm_api_key or None,
+        temperature=settings.llm_temperature if temperature is None else temperature,
+        max_tokens=max_tokens or settings.llm_max_tokens,
+        cache=True,
+        **_extra_kwargs(effort),
+    )
+
+
+@lru_cache
+def main_lm(effort: str | None = None) -> dspy.LM:
+    """effort overrides LLM_REASONING_EFFORT per module, e.g. main_lm("high") for the groundedness judge."""
+    return _build(settings.llm_model, effort=effort, temperature=None, max_tokens=None)
+
+
+@lru_cache
+def fast_lm() -> dspy.LM:
+    return _build(settings.llm_fast_model, effort="low", temperature=None, max_tokens=8000)
+
+
+def configure_default() -> None:
+    dspy.configure(lm=main_lm(), adapter=dspy.JSONAdapter())
+
+
+@dataclass
+class Usage:
+    prompt_tokens: int = 0
+    completion_tokens: int = 0
+    cost_usd: float = 0.0
+
+    @property
+    def total(self) -> int:
+        return self.prompt_tokens + self.completion_tokens
+
+
+@contextmanager
+def track_usage(lm: dspy.LM):
+    """Collect token usage for calls made inside the block (for usage_events metering)."""
+    usage = Usage()
+    start = len(lm.history)
+    try:
+        yield usage
+    finally:
+        for entry in lm.history[start:]:
+            u = entry.get("usage") or {}
+            usage.prompt_tokens += int(u.get("prompt_tokens") or 0)
+            usage.completion_tokens += int(u.get("completion_tokens") or 0)
+            usage.cost_usd += float(entry.get("cost") or 0.0)
+
+
+def provider_name(model: str | None = None) -> str:
+    model = model or settings.llm_model
+    return "zai" if is_zai() and model.startswith("openai/") else model.split("/", 1)[0]

backend/app/llm/signatures.py

@@ -0,0 +1,141 @@
+"""Typed DSPy signatures with Pydantic outputs. Phase 1 modules are implemented; later ones are
+declared now so the output contract is fixed early."""
+
+from typing import Literal
+
+import dspy
+from pydantic import BaseModel, Field
+
+# Shared output models
+
+
+class Section(BaseModel):
+    heading: str | None = None
+    text: str
+
+
+class CitedSentence(BaseModel):
+    sentence: str
+    chunk_ids: list[int] = Field(description="Indices of supporting passages from the context list")
+
+
+class GroundedAnswerOut(BaseModel):
+    answer: str = Field(description="Answer text with [n] markers that refer to passage indices")
+    sentences: list[CitedSentence]
+    unsupported: bool = Field(description="True if the passages do not contain enough to answer")
+
+
+class VoiceProfileOut(BaseModel):
+    tone: list[str]
+    sentence_length: Literal["short", "medium", "long", "varied"]
+    vocabulary: list[str] = Field(description="Signature words and phrases the writer uses")
+    structure_habits: list[str]
+    openings: list[str] = Field(description="How the writer typically opens a piece")
+    avoid: list[str] = Field(description="Things this writer never does")
+    summary: str
+
+
+class Claim(BaseModel):
+    claim: str
+    chunk_ids: list[int]
+
+
+class ScriptLine(BaseModel):
+    speaker: Literal["host_a", "host_b"]
+    text: str
+    chunk_ids: list[int] = Field(default_factory=list)
+
+
+class Scene(BaseModel):
+    type: Literal["title", "section", "pull_quote", "number", "outro"]
+    on_screen_text: str
+    narration: str
+    duration_hint_s: float
+    visual: str = ""
+
+
+class Hook(BaseModel):
+    text: str
+    strength: float = Field(ge=0, le=1)
+    rationale: str
+
+
+# Signatures
+
+
+class CleanAndSegment(dspy.Signature):
+    """Turn a raw blog post into clean sections with headings. Keep the writer's words; drop
+    navigation, subscribe prompts, footers and share buttons."""
+
+    title: str = dspy.InputField()
+    raw_text: str = dspy.InputField()
+    sections: list[Section] = dspy.OutputField()
+
+
+class GroundedAnswer(dspy.Signature):
+    """Answer the question using ONLY the numbered passages. Every factual sentence must cite at least
+    one passage with [n]. If the passages do not support an answer, say so and set unsupported=true.
+    Never use em dashes."""
+
+    question: str = dspy.InputField()
+    passages: list[str] = dspy.InputField(desc="Numbered passages from the writer's own posts")
+    result: GroundedAnswerOut = dspy.OutputField()
+
+
+class GroundednessJudge(dspy.Signature):
+    """Decide whether the sentence is fully supported by the cited passages."""
+
+    sentence: str = dspy.InputField()
+    passages: list[str] = dspy.InputField()
+    supported: bool = dspy.OutputField()
+
+
+class BuildVoiceProfile(dspy.Signature):
+    """Describe the writer's style from their sample posts so other modules can write like them."""
+
+    samples: list[str] = dspy.InputField()
+    profile: VoiceProfileOut = dspy.OutputField()
+
+
+class ExtractClaims(dspy.Signature):
+    """List the key claims of the post, each with the passages that support it."""
+
+    passages: list[str] = dspy.InputField()
+    claims: list[Claim] = dspy.OutputField()
+
+
+class PodcastScript(dspy.Signature):
+    """Write a two host conversation about the material. Hosts stay grounded in the passages and cite
+    them. Natural, warm, curious. No em dashes."""
+
+    passages: list[str] = dspy.InputField()
+    format: Literal["deep_dive", "brief", "debate"] = dspy.InputField()
+    target_minutes: int = dspy.InputField()
+    lines: list[ScriptLine] = dspy.OutputField()
+
+
+class VideoStoryboard(dspy.Signature):
+    """Turn the post into a scene list for a Remotion composition."""
+
+    passages: list[str] = dspy.InputField()
+    aspect: Literal["16:9", "9:16", "1:1"] = dspy.InputField()
+    scenes: list[Scene] = dspy.OutputField()
+
+
+class HookGenerator(dspy.Signature):
+    """Write scroll stopping hooks for the post in the writer's voice. No em dashes."""
+
+    passages: list[str] = dspy.InputField()
+    voice_profile: str = dspy.InputField()
+    n: int = dspy.InputField()
+    hooks: list[Hook] = dspy.OutputField()
+
+
+class PlatformAdapter(dspy.Signature):
+    """Adapt claims into a native post for the platform, in the writer's voice, following the rules."""
+
+    claims: list[Claim] = dspy.InputField()
+    voice_profile: str = dspy.InputField()
+    platform: Literal["x_thread", "linkedin", "substack_notes", "bluesky"] = dspy.InputField()
+    platform_rules: str = dspy.InputField()
+    posts: list[str] = dspy.OutputField()

backend/app/main.py

@@ -0,0 +1,62 @@
+import logging
+import time
+import uuid
+from contextlib import asynccontextmanager
+
+from arq import create_pool
+from arq.connections import RedisSettings
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.config import settings
+from app.llm.provider import configure_default
+from app.routers import artifacts, auth, billing, jobs, notebooks, sources, storage, unsubscribe
+
+log = logging.getLogger("notestack")
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    logging.basicConfig(level=logging.INFO)
+    configure_default()
+    app.state.arq = await create_pool(RedisSettings.from_dsn(settings.redis_url))
+    if settings.env == "development" and settings.r2_endpoint_url:
+        try:
+            from app.services.storage import storage as store
+
+            store.ensure_bucket()  # MinIO locally
+        except Exception:
+            log.warning("could not ensure local bucket", exc_info=True)
+    yield
+    await app.state.arq.aclose()
+
+
+app = FastAPI(title="Notestack API", version="0.1.0", lifespan=lifespan)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=settings.cors_origin_list,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.middleware("http")
+async def request_id(request: Request, call_next):
+    rid = request.headers.get("x-request-id") or uuid.uuid4().hex[:12]
+    start = time.perf_counter()
+    response = await call_next(request)
+    response.headers["x-request-id"] = rid
+    log.info("%s %s %s %.0fms rid=%s", request.method, request.url.path, response.status_code,
+             (time.perf_counter() - start) * 1000, rid)
+    return response
+
+
+@app.get("/api/health")
+def health():
+    return {"ok": True, "llm": settings.llm_model, "billing_enabled": settings.billing_enabled}
+
+
+for r in (auth, billing, storage, sources, notebooks, artifacts, jobs, unsubscribe):
+    app.include_router(r.router)
+app.include_router(sources.documents_router)

backend/app/models/__init__.py

@@ -0,0 +1,52 @@
+from app.models.content import (
+    Artifact,
+    CalendarItem,
+    Chat,
+    Chunk,
+    Citation,
+    Document,
+    DspyTrace,
+    EngagementEvent,
+    Job,
+    Message,
+    Notebook,
+    NotebookDocument,
+    Source,
+    TrackedLink,
+    Upload,
+    UsageEvent,
+    VoiceConsent,
+    VoiceProfile,
+)
+from app.models.email import EmailVerificationCode, UpdateEmail, UpdateEmailSend, VerificationPurpose
+from app.models.user import AuthProvider, Subscription, User, Workspace, WorkspaceMember
+
+__all__ = [
+    "Artifact",
+    "AuthProvider",
+    "CalendarItem",
+    "Chat",
+    "Chunk",
+    "Citation",
+    "Document",
+    "DspyTrace",
+    "EmailVerificationCode",
+    "EngagementEvent",
+    "Job",
+    "Message",
+    "Notebook",
+    "NotebookDocument",
+    "Source",
+    "Subscription",
+    "TrackedLink",
+    "UpdateEmail",
+    "UpdateEmailSend",
+    "Upload",
+    "UsageEvent",
+    "User",
+    "VerificationPurpose",
+    "VoiceConsent",
+    "VoiceProfile",
+    "Workspace",
+    "WorkspaceMember",
+]

backend/app/models/base.py

@@ -0,0 +1,18 @@
+import uuid
+from datetime import UTC, datetime
+
+from sqlalchemy import DateTime, Uuid
+from sqlalchemy.orm import Mapped, mapped_column
+
+
+def utcnow() -> datetime:
+    return datetime.now(UTC)
+
+
+class IdMixin:
+    id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
+
+
+class TimestampMixin:
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow)

backend/app/models/content.py

@@ -0,0 +1,249 @@
+import uuid
+from datetime import datetime
+
+from pgvector.sqlalchemy import Vector
+from sqlalchemy import (
+    JSON,
+    BigInteger,
+    Boolean,
+    DateTime,
+    Float,
+    ForeignKey,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+    Uuid,
+)
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.config import settings
+from app.db import Base
+from app.models.base import IdMixin, TimestampMixin
+
+
+def _ws_fk() -> Mapped[uuid.UUID]:
+    return mapped_column(Uuid, ForeignKey("workspaces.id", ondelete="CASCADE"), index=True)
+
+
+class Source(IdMixin, TimestampMixin, Base):
+    __tablename__ = "sources"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    feed_url: Mapped[str] = mapped_column(String(1000))
+    site_url: Mapped[str | None] = mapped_column(String(1000))
+    platform: Mapped[str] = mapped_column(String(20), default="rss")  # substack | ghost | medium | rss | url
+    title: Mapped[str | None] = mapped_column(String(500))
+    sync_status: Mapped[str] = mapped_column(String(20), default="pending")  # pending | syncing | ok | error
+    sync_error: Mapped[str | None] = mapped_column(Text)
+    last_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
+class Document(IdMixin, TimestampMixin, Base):
+    __tablename__ = "documents"
+    __table_args__ = (UniqueConstraint("source_id", "url"),)
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    source_id: Mapped[uuid.UUID | None] = mapped_column(Uuid, ForeignKey("sources.id", ondelete="CASCADE"))
+    title: Mapped[str] = mapped_column(String(1000))
+    url: Mapped[str] = mapped_column(String(1000))
+    published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    raw_html_key: Mapped[str | None] = mapped_column(String(500))  # R2 key
+    clean_text: Mapped[str] = mapped_column(Text, default="")
+    content_hash: Mapped[str | None] = mapped_column(String(64))
+    metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
+
+    chunks: Mapped[list["Chunk"]] = relationship(back_populates="document", cascade="all, delete-orphan")
+
+
+class Chunk(IdMixin, TimestampMixin, Base):
+    __tablename__ = "chunks"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    document_id: Mapped[uuid.UUID] = mapped_column(
+        Uuid, ForeignKey("documents.id", ondelete="CASCADE"), index=True
+    )
+    position: Mapped[int] = mapped_column(Integer)
+    heading: Mapped[str | None] = mapped_column(String(500))
+    text: Mapped[str] = mapped_column(Text)
+    token_count: Mapped[int] = mapped_column(Integer, default=0)
+    embedding: Mapped[list[float] | None] = mapped_column(Vector(settings.embedding_dim))
+    entities: Mapped[list] = mapped_column(JSON, default=list)
+
+    document: Mapped[Document] = relationship(back_populates="chunks")
+
+
+class Notebook(IdMixin, TimestampMixin, Base):
+    __tablename__ = "notebooks"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    title: Mapped[str] = mapped_column(String(300))
+    description: Mapped[str | None] = mapped_column(Text)
+    summary: Mapped[str | None] = mapped_column(Text)
+
+
+class NotebookDocument(Base):
+    __tablename__ = "notebook_documents"
+
+    notebook_id: Mapped[uuid.UUID] = mapped_column(
+        Uuid, ForeignKey("notebooks.id", ondelete="CASCADE"), primary_key=True
+    )
+    document_id: Mapped[uuid.UUID] = mapped_column(
+        Uuid, ForeignKey("documents.id", ondelete="CASCADE"), primary_key=True
+    )
+
+
+class VoiceProfile(IdMixin, TimestampMixin, Base):
+    __tablename__ = "voice_profiles"
+
+    workspace_id: Mapped[uuid.UUID] = mapped_column(
+        Uuid, ForeignKey("workspaces.id", ondelete="CASCADE"), unique=True
+    )
+    profile_json: Mapped[dict] = mapped_column(JSON, default=dict)
+    sample_doc_ids: Mapped[list] = mapped_column(JSON, default=list)
+
+
+class VoiceConsent(IdMixin, TimestampMixin, Base):
+    """Required before any voice cloning. The recorded sample lives in R2."""
+
+    __tablename__ = "voice_consents"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    user_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("users.id"))
+    sample_key: Mapped[str] = mapped_column(String(500))
+    consent_text: Mapped[str] = mapped_column(Text)
+    ip_address: Mapped[str | None] = mapped_column(String(64))
+    elevenlabs_voice_id: Mapped[str | None] = mapped_column(String(100))
+    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
+class Chat(IdMixin, TimestampMixin, Base):
+    __tablename__ = "chats"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    notebook_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("notebooks.id", ondelete="CASCADE"))
+    title: Mapped[str | None] = mapped_column(String(300))
+
+
+class Message(IdMixin, TimestampMixin, Base):
+    __tablename__ = "messages"
+
+    chat_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("chats.id", ondelete="CASCADE"), index=True)
+    role: Mapped[str] = mapped_column(String(20))  # user | assistant
+    content: Mapped[str] = mapped_column(Text)
+
+    citations: Mapped[list["Citation"]] = relationship(cascade="all, delete-orphan")
+
+
+class Citation(IdMixin, Base):
+    __tablename__ = "citations"
+
+    message_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("messages.id", ondelete="CASCADE"))
+    chunk_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("chunks.id", ondelete="CASCADE"))
+    marker: Mapped[int] = mapped_column(Integer)  # [1], [2] in the answer text
+    span: Mapped[str | None] = mapped_column(Text)  # quoted supporting text
+
+
+class Artifact(IdMixin, TimestampMixin, Base):
+    __tablename__ = "artifacts"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    notebook_id: Mapped[uuid.UUID | None] = mapped_column(Uuid, ForeignKey("notebooks.id", ondelete="SET NULL"))
+    document_id: Mapped[uuid.UUID | None] = mapped_column(Uuid, ForeignKey("documents.id", ondelete="SET NULL"))
+    # summary | audio_overview | video | thread | linkedin | notes | quote_card | carousel | seo
+    type: Mapped[str] = mapped_column(String(30))
+    status: Mapped[str] = mapped_column(String(20), default="pending")
+    content_json: Mapped[dict] = mapped_column(JSON, default=dict)
+    storage_key: Mapped[str | None] = mapped_column(String(500))
+
+
+class Job(IdMixin, TimestampMixin, Base):
+    """Covers ingest, tts and render jobs. kind decides which fields matter."""
+
+    __tablename__ = "jobs"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    kind: Mapped[str] = mapped_column(String(30))  # ingest | tts | render | generate
+    status: Mapped[str] = mapped_column(String(20), default="queued")  # queued | running | done | failed
+    progress: Mapped[float] = mapped_column(Float, default=0.0)
+    message: Mapped[str | None] = mapped_column(String(500))
+    error: Mapped[str | None] = mapped_column(Text)
+    params: Mapped[dict] = mapped_column(JSON, default=dict)
+    result: Mapped[dict] = mapped_column(JSON, default=dict)
+    artifact_id: Mapped[uuid.UUID | None] = mapped_column(Uuid, ForeignKey("artifacts.id", ondelete="SET NULL"))
+    cost_usd: Mapped[float] = mapped_column(Float, default=0.0)
+    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
+class Upload(IdMixin, TimestampMixin, Base):
+    __tablename__ = "uploads"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    user_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("users.id"))
+    key: Mapped[str] = mapped_column(String(500), unique=True)
+    filename: Mapped[str] = mapped_column(String(300))
+    content_type: Mapped[str] = mapped_column(String(100))
+    size_bytes: Mapped[int | None] = mapped_column(BigInteger)
+    status: Mapped[str] = mapped_column(String(20), default="pending")  # pending | complete
+
+
+class CalendarItem(IdMixin, TimestampMixin, Base):
+    __tablename__ = "calendar_items"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    artifact_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("artifacts.id", ondelete="CASCADE"))
+    platform: Mapped[str] = mapped_column(String(30))
+    scheduled_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
+    status: Mapped[str] = mapped_column(String(20), default="scheduled")
+
+
+class TrackedLink(IdMixin, TimestampMixin, Base):
+    __tablename__ = "tracked_links"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    slug: Mapped[str] = mapped_column(String(32), unique=True)
+    target_url: Mapped[str] = mapped_column(String(2000))
+    calendar_item_id: Mapped[uuid.UUID | None] = mapped_column(
+        Uuid, ForeignKey("calendar_items.id", ondelete="SET NULL")
+    )
+
+
+class EngagementEvent(IdMixin, TimestampMixin, Base):
+    __tablename__ = "engagement_events"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    tracked_link_id: Mapped[uuid.UUID | None] = mapped_column(
+        Uuid, ForeignKey("tracked_links.id", ondelete="CASCADE")
+    )
+    kind: Mapped[str] = mapped_column(String(30))  # click | like | repost | reply | subscribe
+    value: Mapped[float] = mapped_column(Float, default=1.0)
+    metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
+
+
+class DspyTrace(IdMixin, TimestampMixin, Base):
+    __tablename__ = "dspy_traces"
+
+    workspace_id: Mapped[uuid.UUID | None] = mapped_column(Uuid, ForeignKey("workspaces.id", ondelete="SET NULL"))
+    module: Mapped[str] = mapped_column(String(60), index=True)
+    model: Mapped[str] = mapped_column(String(100))
+    inputs: Mapped[dict] = mapped_column(JSON, default=dict)
+    outputs: Mapped[dict] = mapped_column(JSON, default=dict)
+    score: Mapped[float | None] = mapped_column(Float)
+    feedback: Mapped[str | None] = mapped_column(Text)
+    # Only usable for optimization when the workspace opted in.
+    training_allowed: Mapped[bool] = mapped_column(Boolean, default=False)
+
+
+class UsageEvent(IdMixin, TimestampMixin, Base):
+    """Metering: LLM tokens, TTS characters, render seconds. One row per billable call."""
+
+    __tablename__ = "usage_events"
+
+    workspace_id: Mapped[uuid.UUID] = _ws_fk()
+    job_id: Mapped[uuid.UUID | None] = mapped_column(Uuid, ForeignKey("jobs.id", ondelete="SET NULL"))
+    kind: Mapped[str] = mapped_column(String(20))  # llm | embedding | tts | render
+    provider: Mapped[str] = mapped_column(String(40))
+    model: Mapped[str | None] = mapped_column(String(100))
+    quantity: Mapped[float] = mapped_column(Float)  # tokens, characters or seconds
+    unit: Mapped[str] = mapped_column(String(20))
+    cost_usd: Mapped[float] = mapped_column(Float, default=0.0)

backend/app/models/email.py

@@ -0,0 +1,58 @@
+import enum
+import uuid
+from datetime import datetime
+
+from sqlalchemy import DateTime, Enum, ForeignKey, Integer, String, Text, UniqueConstraint, Uuid
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db import Base
+from app.models.base import IdMixin, TimestampMixin
+
+
+class VerificationPurpose(str, enum.Enum):
+    SIGNUP = "signup"
+    PASSWORD_RESET = "password_reset"
+
+
+class EmailVerificationCode(IdMixin, TimestampMixin, Base):
+    __tablename__ = "email_verification_codes"
+
+    email: Mapped[str] = mapped_column(String(320), index=True)
+    purpose: Mapped[VerificationPurpose] = mapped_column(
+        Enum(VerificationPurpose, native_enum=False, length=20)
+    )
+    code_hash: Mapped[str] = mapped_column(String(128))
+    # Signup stores the pending name + password hash until the code is verified.
+    pending_name: Mapped[str | None] = mapped_column(String(200))
+    pending_password_hash: Mapped[str | None] = mapped_column(String(200))
+    attempts: Mapped[int] = mapped_column(Integer, default=0)
+    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
+    consumed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
+class UpdateEmail(IdMixin, TimestampMixin, Base):
+    """Broadcast campaign. Body is plain text (it gets HTML escaped)."""
+
+    __tablename__ = "update_emails"
+
+    subject: Mapped[str] = mapped_column(String(300))
+    body: Mapped[str] = mapped_column(Text)
+    user_filter: Mapped[str] = mapped_column(String(20), default="all")  # all | free | paid | writer | studio
+    batch_size: Mapped[int] = mapped_column(Integer, default=50)
+    send_hour: Mapped[int] = mapped_column(Integer, default=-1)  # -1 = config default, UTC
+    total_users: Mapped[int] = mapped_column(Integer, default=0)
+    sent_count: Mapped[int] = mapped_column(Integer, default=0)
+    failed_count: Mapped[int] = mapped_column(Integer, default=0)
+    status: Mapped[str] = mapped_column(String(20), default="scheduled")  # scheduled | running | completed
+    last_batch_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
+class UpdateEmailSend(IdMixin, TimestampMixin, Base):
+    __tablename__ = "update_email_sends"
+    __table_args__ = (UniqueConstraint("update_email_id", "user_id"),)
+
+    update_email_id: Mapped[uuid.UUID] = mapped_column(
+        Uuid, ForeignKey("update_emails.id", ondelete="CASCADE")
+    )
+    user_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("users.id", ondelete="CASCADE"))
+    status: Mapped[str] = mapped_column(String(10))  # sent | failed

backend/app/models/user.py

@@ -0,0 +1,72 @@
+import enum
+import uuid
+from datetime import datetime
+
+from sqlalchemy import JSON, Boolean, DateTime, Enum, ForeignKey, Integer, String, UniqueConstraint, Uuid
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.db import Base
+from app.models.base import IdMixin, TimestampMixin
+
+
+class AuthProvider(str, enum.Enum):
+    GOOGLE = "google"
+    EMAIL = "email"
+
+
+class User(IdMixin, TimestampMixin, Base):
+    """One auth provider per account for life, no linking (same rule as blog2video)."""
+
+    __tablename__ = "users"
+
+    email: Mapped[str] = mapped_column(String(320), unique=True, index=True)
+    name: Mapped[str | None] = mapped_column(String(200))
+    avatar_url: Mapped[str | None] = mapped_column(String(1000))
+    auth_provider: Mapped[AuthProvider] = mapped_column(Enum(AuthProvider, native_enum=False, length=16))
+    google_id: Mapped[str | None] = mapped_column(String(64), unique=True)
+    password_hash: Mapped[str | None] = mapped_column(String(200))
+    token_version: Mapped[int] = mapped_column(Integer, default=0)
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    email_unsubscribed: Mapped[bool] = mapped_column(Boolean, default=False)
+    welcome_email_sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+    memberships: Mapped[list["WorkspaceMember"]] = relationship(back_populates="user")
+
+
+class Workspace(IdMixin, TimestampMixin, Base):
+    __tablename__ = "workspaces"
+
+    name: Mapped[str] = mapped_column(String(200))
+    owner_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("users.id"))
+    training_opt_in: Mapped[bool] = mapped_column(Boolean, default=False)
+    brand_json: Mapped[dict] = mapped_column(JSON, default=dict)
+
+    members: Mapped[list["WorkspaceMember"]] = relationship(back_populates="workspace")
+
+
+class WorkspaceMember(IdMixin, TimestampMixin, Base):
+    __tablename__ = "workspace_members"
+    __table_args__ = (UniqueConstraint("workspace_id", "user_id"),)
+
+    workspace_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("workspaces.id", ondelete="CASCADE"))
+    user_id: Mapped[uuid.UUID] = mapped_column(Uuid, ForeignKey("users.id", ondelete="CASCADE"))
+    role: Mapped[str] = mapped_column(String(20), default="owner")
+
+    workspace: Mapped[Workspace] = relationship(back_populates="members")
+    user: Mapped[User] = relationship(back_populates="memberships")
+
+
+class Subscription(IdMixin, TimestampMixin, Base):
+    __tablename__ = "subscriptions"
+
+    workspace_id: Mapped[uuid.UUID] = mapped_column(
+        Uuid, ForeignKey("workspaces.id", ondelete="CASCADE"), unique=True
+    )
+    plan: Mapped[str] = mapped_column(String(20), default="free")  # free | writer | studio
+    status: Mapped[str] = mapped_column(String(20), default="active")
+    provider: Mapped[str | None] = mapped_column(String(20))
+    provider_customer_id: Mapped[str | None] = mapped_column(String(100))
+    provider_subscription_id: Mapped[str | None] = mapped_column(String(100))
+    current_period_end: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

backend/app/pipeline/ingest.py

@@ -0,0 +1,168 @@
+"""Ingestion: feed -> documents (raw HTML to R2) -> clean text -> chunks -> embeddings."""
+
+import hashlib
+import logging
+import re
+import uuid
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from time import mktime
+from urllib.parse import urlparse
+
+import feedparser
+import httpx
+import tiktoken
+from bs4 import BeautifulSoup
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.config import settings
+from app.llm.embeddings import embed_texts, embedding_provider
+from app.models import Chunk, Document, Job, Source
+from app.services.jobs import record_usage, update_job
+from app.services.storage import keys, storage
+
+log = logging.getLogger(__name__)
+
+CHUNK_TOKENS = 350
+CHUNK_OVERLAP = 60
+_enc = tiktoken.get_encoding("cl100k_base")
+_DROP = re.compile(r"(subscribe|share this post|leave a comment|thanks for reading)", re.I)
+
+
+def normalize_feed_url(url: str) -> tuple[str, str]:
+    """Accept 'name.substack.com', a Substack post URL, or any feed URL. Returns (feed_url, platform)."""
+    url = url.strip()
+    if not url.startswith(("http://", "https://")):
+        url = f"https://{url}"
+    parsed = urlparse(url)
+    host = parsed.netloc.lower()
+    if host.endswith("substack.com") or "/p/" in parsed.path:
+        return f"{parsed.scheme}://{host}/feed", "substack"
+    if parsed.path.rstrip("/").endswith(("/feed", "/rss", ".xml", "/atom")):
+        return url, "rss"
+    if host.endswith("medium.com"):
+        return f"https://medium.com/feed{parsed.path.rstrip('/')}", "medium"
+    return f"{parsed.scheme}://{host}/feed", "rss"
+
+
+def html_to_sections(html: str) -> list[tuple[str | None, str]]:
+    soup = BeautifulSoup(html, "html.parser")
+    for tag in soup(["script", "style", "nav", "footer", "form", "button", "iframe"]):
+        tag.decompose()
+    sections: list[tuple[str | None, list[str]]] = [(None, [])]
+    for el in soup.find_all(["h1", "h2", "h3", "p", "li", "blockquote"]):
+        text = el.get_text(" ", strip=True)
+        if not text or (len(text) < 120 and _DROP.search(text)):
+            continue
+        if el.name in {"h1", "h2", "h3"}:
+            sections.append((text, []))
+        else:
+            sections[-1][1].append(text)
+    return [(h, "\n\n".join(p)) for h, p in sections if p]
+
+
+def chunk_sections(sections: list[tuple[str | None, str]]) -> list[tuple[str | None, str, int]]:
+    out: list[tuple[str | None, str, int]] = []
+    for heading, text in sections:
+        tokens = _enc.encode(text)
+        start = 0
+        while start < len(tokens):
+            window = tokens[start : start + CHUNK_TOKENS]
+            out.append((heading, _enc.decode(window), len(window)))
+            if start + CHUNK_TOKENS >= len(tokens):
+                break
+            start += CHUNK_TOKENS - CHUNK_OVERLAP
+    return out
+
+
+@dataclass
+class FeedEntry:
+    title: str
+    url: str
+    published_at: datetime | None
+    html: str
+
+
+def fetch_feed(feed_url: str, limit: int) -> tuple[str | None, list[FeedEntry]]:
+    resp = httpx.get(feed_url, follow_redirects=True, timeout=30, headers={"User-Agent": "NotestackBot/0.1"})
+    resp.raise_for_status()
+    parsed = feedparser.parse(resp.content)
+    entries = []
+    for e in parsed.entries[:limit]:
+        content = (e.get("content") or [{}])[0].get("value") or e.get("summary") or ""
+        published = e.get("published_parsed") or e.get("updated_parsed")
+        entries.append(
+            FeedEntry(
+                title=e.get("title") or "Untitled",
+                url=e.get("link") or "",
+                published_at=datetime.fromtimestamp(mktime(published), UTC) if published else None,
+                html=content,
+            )
+        )
+    return parsed.feed.get("title"), entries
+
+
+def ingest_source(db: Session, source: Source, job: Job, max_posts: int) -> dict:
+    source.sync_status = "syncing"
+    update_job(db, job, status="running", progress=0.02, message="Contacting your feed")
+
+    title, entries = fetch_feed(source.feed_url, max_posts)
+    source.title = source.title or title
+    total = max(len(entries), 1)
+    indexed = skipped = 0
+    embed_tokens = 0
+
+    for i, entry in enumerate(entries):
+        content_hash = hashlib.sha256(entry.html.encode()).hexdigest()
+        doc = db.scalar(select(Document).where(Document.source_id == source.id, Document.url == entry.url))
+        if doc and doc.content_hash == content_hash:
+            skipped += 1
+            continue
+        if not doc:
+            doc = Document(id=uuid.uuid4(), workspace_id=source.workspace_id, source_id=source.id, url=entry.url)
+            db.add(doc)
+        doc.title = entry.title
+        doc.published_at = entry.published_at
+        doc.content_hash = content_hash
+        doc.raw_html_key = storage.put_text(
+            keys.raw_html(source.workspace_id, source.id, doc.id), entry.html, "text/html; charset=utf-8"
+        )
+        sections = html_to_sections(entry.html)
+        doc.clean_text = "\n\n".join(f"## {h}\n{t}" if h else t for h, t in sections)
+        doc.chunks.clear()
+        pieces = chunk_sections(sections)
+        if pieces:
+            result = embed_texts([f"{entry.title}\n{h or ''}\n{t}" for h, t, _ in pieces])
+            embed_tokens += result.tokens
+            for pos, ((heading, text, n_tokens), vec) in enumerate(zip(pieces, result.vectors, strict=True)):
+                doc.chunks.append(
+                    Chunk(
+                        workspace_id=source.workspace_id,
+                        position=pos,
+                        heading=heading,
+                        text=text,
+                        token_count=n_tokens,
+                        embedding=vec,
+                    )
+                )
+        indexed += 1
+        db.commit()
+        update_job(db, job, progress=0.05 + 0.9 * (i + 1) / total, message=f"{entry.title[:80]} is in orbit")
+
+    if embed_tokens:
+        record_usage(
+            db,
+            workspace_id=source.workspace_id,
+            kind="embedding",
+            provider=embedding_provider(),
+            model=settings.embedding_model,
+            quantity=embed_tokens,
+            unit="tokens",
+            job=job,
+        )
+    source.sync_status = "ok"
+    source.sync_error = None
+    source.last_synced_at = datetime.now(UTC)
+    db.commit()
+    return {"indexed": indexed, "skipped": skipped, "found": len(entries)}

backend/app/pipeline/retrieval.py

@@ -0,0 +1,62 @@
+"""Vector retrieval + grounded answering with citations."""
+
+import uuid
+from dataclasses import dataclass
+
+import dspy
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.llm.embeddings import embed_query
+from app.llm.postprocess import strip_em_dashes
+from app.llm.provider import main_lm, track_usage
+from app.llm.signatures import GroundedAnswer
+from app.models import Chunk, Document, NotebookDocument
+
+
+@dataclass
+class Retrieved:
+    chunk: Chunk
+    document: Document
+
+
+def retrieve(db: Session, workspace_id: uuid.UUID, query: str, notebook_id: uuid.UUID | None, k: int = 8):
+    vec = embed_query(query)
+    stmt = (
+        select(Chunk, Document)
+        .join(Document, Document.id == Chunk.document_id)
+        .where(Chunk.workspace_id == workspace_id, Chunk.embedding.is_not(None))
+    )
+    if notebook_id:
+        stmt = stmt.join(NotebookDocument, NotebookDocument.document_id == Document.id).where(
+            NotebookDocument.notebook_id == notebook_id
+        )
+    stmt = stmt.order_by(Chunk.embedding.cosine_distance(vec)).limit(k)
+    return [Retrieved(c, d) for c, d in db.execute(stmt).all()]
+
+
+@dataclass
+class Answer:
+    text: str
+    unsupported: bool
+    citations: list[tuple[int, Retrieved]]  # (marker, retrieved)
+    prompt_tokens: int
+    completion_tokens: int
+
+
+def grounded_answer(question: str, hits: list[Retrieved]) -> Answer:
+    if not hits:
+        return Answer("I could not find anything about that in your posts yet.", True, [], 0, 0)
+    passages = [f"[{i + 1}] {h.document.title}: {h.chunk.text}" for i, h in enumerate(hits)]
+    lm = main_lm()
+    with track_usage(lm) as usage, dspy.context(lm=lm, adapter=dspy.JSONAdapter()):
+        pred = dspy.Predict(GroundedAnswer)(question=question, passages=passages)
+    result = pred.result
+    used = sorted({i for s in result.sentences for i in s.chunk_ids if 1 <= i <= len(hits)})
+    return Answer(
+        text=strip_em_dashes(result.answer),
+        unsupported=result.unsupported,
+        citations=[(i, hits[i - 1]) for i in used],
+        prompt_tokens=usage.prompt_tokens,
+        completion_tokens=usage.completion_tokens,
+    )

backend/app/routers/artifacts.py

@@ -0,0 +1,77 @@
+import uuid
+from typing import Literal
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from pydantic import BaseModel
+from sqlalchemy import select
+
+from app.auth import Ctx, get_ctx
+from app.models import Artifact
+from app.services.jobs import create_job, serialize_job
+from app.services.storage import storage
+
+router = APIRouter(prefix="/api/artifacts", tags=["artifacts"])
+
+COMPOSITIONS = {"ShortVertical": "mp4", "QuoteCard": "png"}
+
+
+class ArtifactIn(BaseModel):
+    type: Literal["summary", "audio_overview", "video", "thread", "linkedin", "notes", "quote_card", "carousel", "seo"]
+    notebook_id: uuid.UUID | None = None
+    document_id: uuid.UUID | None = None
+    content: dict = {}
+
+
+class RenderIn(BaseModel):
+    composition: Literal["ShortVertical", "QuoteCard"]
+    props: dict
+
+
+def _serialize(a: Artifact) -> dict:
+    return {
+        "id": str(a.id),
+        "type": a.type,
+        "status": a.status,
+        "content": a.content_json,
+        "storage_key": a.storage_key,
+        "url": storage.presign_get(a.storage_key) if a.storage_key else None,
+    }
+
+
+def _get(ctx: Ctx, artifact_id: uuid.UUID) -> Artifact:
+    a = ctx.db.scalar(select(Artifact).where(Artifact.id == artifact_id, Artifact.workspace_id == ctx.workspace.id))
+    if not a:
+        raise HTTPException(404, "Artifact not found")
+    return a
+
+
+@router.post("")
+def create_artifact(body: ArtifactIn, ctx: Ctx = Depends(get_ctx)):
+    a = Artifact(
+        workspace_id=ctx.workspace.id,
+        type=body.type,
+        notebook_id=body.notebook_id,
+        document_id=body.document_id,
+        content_json=body.content,
+        status="draft",
+    )
+    ctx.db.add(a)
+    ctx.db.commit()
+    return _serialize(a)
+
+
+@router.get("/{artifact_id}")
+def get_artifact(artifact_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
+    return _serialize(_get(ctx, artifact_id))
+
+
+@router.post("/{artifact_id}/render")
+async def render_artifact(artifact_id: uuid.UUID, body: RenderIn, request: Request, ctx: Ctx = Depends(get_ctx)):
+    a = _get(ctx, artifact_id)
+    fmt = COMPOSITIONS[body.composition]
+    job = create_job(ctx.db, ctx.workspace.id, "render", {"composition": body.composition})
+    job.artifact_id = a.id
+    a.status = "rendering"
+    ctx.db.commit()
+    await request.app.state.arq.enqueue_job("render_task", str(a.id), str(job.id), body.composition, body.props, fmt)
+    return serialize_job(job)

backend/app/routers/auth.py

@@ -0,0 +1,244 @@
+from datetime import UTC, datetime
+
+from fastapi import APIRouter, Depends, HTTPException
+from google.auth.transport import requests as google_requests
+from google.oauth2 import id_token as google_id_token
+from pydantic import BaseModel, EmailStr
+from sqlalchemy.orm import Session
+
+from app.auth import create_access_token, create_refresh_token, get_current_user, load_user_from_token
+from app.config import settings
+from app.db import get_db
+from app.models import AuthProvider, User, VerificationPurpose
+from app.services import auth_identity as ident
+from app.services import email_verification as ev
+from app.services.email import email_service
+
+router = APIRouter(prefix="/api/auth", tags=["auth"])
+
+
+def _err(exc: ident.IdentityError | ev.VerificationError, status: int = 400) -> HTTPException:
+    return HTTPException(getattr(exc, "status", status), {"code": exc.code, "message": exc.message})
+
+
+def _serialize_user(user: User) -> dict:
+    return {
+        "id": str(user.id),
+        "email": user.email,
+        "name": user.name,
+        "avatar_url": user.avatar_url,
+        "auth_provider": user.auth_provider.value,
+        "email_unsubscribed": user.email_unsubscribed,
+    }
+
+
+def _finalize_login(db: Session, user: User, created: bool) -> dict:
+    user.last_login_at = datetime.now(UTC)
+    if created and not user.welcome_email_sent_at:
+        if email_service.send_welcome(user.email, user.name, str(user.id)):
+            user.welcome_email_sent_at = datetime.now(UTC)
+    db.commit()
+    return {
+        "access_token": create_access_token(user.id, user.token_version),
+        "refresh_token": create_refresh_token(user.id, user.token_version),
+        "user": _serialize_user(user),
+        "created": created,
+    }
+
+
+# Google
+
+
+class GoogleIn(BaseModel):
+    credential: str  # Google Identity Services ID token
+
+
+@router.post("/google")
+def google_login(body: GoogleIn, db: Session = Depends(get_db)):
+    if not settings.google_client_id:
+        raise HTTPException(503, "Google sign in is not configured")
+    try:
+        info = google_id_token.verify_oauth2_token(
+            body.credential, google_requests.Request(), settings.google_client_id
+        )
+    except ValueError as exc:
+        raise HTTPException(401, "Invalid Google credential") from exc
+    if not info.get("email_verified"):
+        raise HTTPException(401, "Google email is not verified")
+    try:
+        resolved = ident.resolve_or_create_google_user(
+            db, email=info["email"], google_id=info["sub"], name=info.get("name"), avatar_url=info.get("picture")
+        )
+    except ident.IdentityError as exc:
+        raise _err(exc) from exc
+    return _finalize_login(db, resolved.user, resolved.created)
+
+
+# Email signup (6 digit code)
+
+
+class RegisterStartIn(BaseModel):
+    email: EmailStr
+    password: str
+    name: str | None = None
+
+
+class CodeIn(BaseModel):
+    email: EmailStr
+    code: str
+
+
+class EmailIn(BaseModel):
+    email: EmailStr
+
+
+@router.post("/email/register/start")
+def register_start(body: RegisterStartIn, db: Session = Depends(get_db)):
+    email = ident.normalize_email(body.email)
+    try:
+        ident.validate_password(body.password)
+        existing = ident.find_user(db, email)
+        if existing:
+            if existing.auth_provider != AuthProvider.EMAIL:
+                raise ident.wrong_provider_error(existing)
+            raise ident.IdentityError("exists", "An account with this email already exists. Sign in instead.", 409)
+        issued = ev.issue_code(
+            db,
+            email,
+            VerificationPurpose.SIGNUP,
+            pending_name=(body.name or "").strip() or None,
+            pending_password_hash=ident.hash_password(body.password),
+        )
+    except (ident.IdentityError, ev.VerificationError) as exc:
+        raise _err(exc, 429 if getattr(exc, "code", "") == "cooldown" else 400) from exc
+    db.commit()
+    email_service.send_verification_code(email, issued.code)
+    return {"ok": True}
+
+
+@router.post("/email/register/verify")
+def register_verify(body: CodeIn, db: Session = Depends(get_db)):
+    email = ident.normalize_email(body.email)
+    try:
+        record = ev.consume_code(db, email, VerificationPurpose.SIGNUP, body.code)
+        user = ident.create_password_user(
+            db, email=email, name=record.pending_name, password_hash=record.pending_password_hash or ""
+        )
+    except (ident.IdentityError, ev.VerificationError) as exc:
+        raise _err(exc) from exc
+    ev.clear_codes(db, email, VerificationPurpose.SIGNUP)
+    return _finalize_login(db, user, True)
+
+
+@router.post("/email/register/resend")
+def register_resend(body: EmailIn, db: Session = Depends(get_db)):
+    email = ident.normalize_email(body.email)
+    pending = ev._latest(db, email, VerificationPurpose.SIGNUP)
+    if not pending:
+        raise HTTPException(400, {"code": "not_found", "message": "Start sign up again."})
+    try:
+        issued = ev.issue_code(
+            db,
+            email,
+            VerificationPurpose.SIGNUP,
+            pending_name=pending.pending_name,
+            pending_password_hash=pending.pending_password_hash,
+        )
+    except ev.VerificationError as exc:
+        raise _err(exc, 429) from exc
+    db.commit()
+    email_service.send_verification_code(email, issued.code)
+    return {"ok": True}
+
+
+# Password login + reset
+
+
+class LoginIn(BaseModel):
+    email: EmailStr
+    password: str
+
+
+@router.post("/email/login")
+def email_login(body: LoginIn, db: Session = Depends(get_db)):
+    try:
+        user = ident.resolve_password_user(db, email=body.email, password=body.password)
+    except ident.IdentityError as exc:
+        raise _err(exc) from exc
+    return _finalize_login(db, user, False)
+
+
+@router.post("/password/forgot/start")
+def forgot_start(body: EmailIn, db: Session = Depends(get_db)):
+    user = ident.find_user(db, body.email)
+    # Always 200 so the endpoint does not reveal which emails exist.
+    if user and user.auth_provider == AuthProvider.EMAIL and user.is_active:
+        try:
+            issued = ev.issue_code(db, user.email, VerificationPurpose.PASSWORD_RESET)
+            db.commit()
+            email_service.send_password_reset_code(user.email, issued.code)
+        except ev.VerificationError:
+            pass
+    return {"ok": True}
+
+
+@router.post("/password/forgot/check")
+def forgot_check(body: CodeIn, db: Session = Depends(get_db)):
+    try:
+        ev.check_code(db, ident.normalize_email(body.email), VerificationPurpose.PASSWORD_RESET, body.code)
+    except ev.VerificationError as exc:
+        raise _err(exc) from exc
+    return {"ok": True}
+
+
+class ResetIn(CodeIn):
+    password: str
+
+
+@router.post("/password/forgot/complete")
+def forgot_complete(body: ResetIn, db: Session = Depends(get_db)):
+    email = ident.normalize_email(body.email)
+    try:
+        ident.validate_password(body.password)
+        ev.consume_code(db, email, VerificationPurpose.PASSWORD_RESET, body.code)
+    except (ident.IdentityError, ev.VerificationError) as exc:
+        raise _err(exc) from exc
+    user = ident.find_user(db, email)
+    if not user:
+        raise HTTPException(400, {"code": "not_found", "message": "Account not found."})
+    user.password_hash = ident.hash_password(body.password)
+    user.token_version += 1  # sign out everywhere
+    ev.clear_codes(db, email, VerificationPurpose.PASSWORD_RESET)
+    return _finalize_login(db, user, False)
+
+
+# Session
+
+
+class RefreshIn(BaseModel):
+    refresh_token: str
+
+
+@router.post("/refresh")
+def refresh(body: RefreshIn, db: Session = Depends(get_db)):
+    user = load_user_from_token(db, body.refresh_token, expected_type="refresh")
+    return {"access_token": create_access_token(user.id, user.token_version)}
+
+
+@router.get("/me")
+def me(user: User = Depends(get_current_user)):
+    return _serialize_user(user)
+
+
+@router.post("/logout")
+def logout(user: User = Depends(get_current_user), db: Session = Depends(get_db)):
+    user.token_version += 1
+    db.commit()
+    return {"ok": True}
+
+
+@router.post("/delete-account")
+def delete_account(user: User = Depends(get_current_user), db: Session = Depends(get_db)):
+    ident.soft_delete(db, user)
+    db.commit()
+    return {"ok": True}

backend/app/routers/billing.py

@@ -0,0 +1,32 @@
+from fastapi import APIRouter, Depends, HTTPException
+
+from app.auth import Ctx, get_ctx
+from app.config import settings
+from app.services.plans import PLANS, effective_plan, plan_dict
+
+router = APIRouter(prefix="/api/billing", tags=["billing"])
+
+
+@router.get("/plans")
+def list_plans():
+    return {"billing_enabled": settings.billing_enabled, "plans": [plan_dict(p) for p in PLANS.values()]}
+
+
+@router.get("/me")
+def my_plan(ctx: Ctx = Depends(get_ctx)):
+    plan = effective_plan(ctx.db, ctx.workspace)
+    return {"billing_enabled": settings.billing_enabled, "plan": plan_dict(plan)}
+
+
+@router.post("/checkout")
+def checkout(ctx: Ctx = Depends(get_ctx)):
+    if not settings.billing_enabled:
+        raise HTTPException(409, {"code": "billing_disabled", "message": "Notestack is free during early access."})
+    raise HTTPException(501, "Checkout provider not wired yet")
+
+
+@router.post("/webhook")
+def webhook():
+    if not settings.billing_enabled:
+        return {"ignored": True}
+    raise HTTPException(501, "Webhook not wired yet")

backend/app/routers/jobs.py

@@ -0,0 +1,83 @@
+import json
+import uuid
+
+from fastapi import APIRouter, Depends, Header, HTTPException
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.auth import Ctx, get_ctx
+from app.config import settings
+from app.db import get_db
+from app.models import Artifact, Job
+from app.services.jobs import record_usage, serialize_job, subscribe, update_job
+
+router = APIRouter(prefix="/api", tags=["jobs"])
+
+
+def _get(ctx: Ctx, job_id: uuid.UUID) -> Job:
+    job = ctx.db.scalar(select(Job).where(Job.id == job_id, Job.workspace_id == ctx.workspace.id))
+    if not job:
+        raise HTTPException(404, "Job not found")
+    return job
+
+
+@router.get("/jobs/{job_id}")
+def get_job(job_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
+    return serialize_job(_get(ctx, job_id))
+
+
+@router.get("/jobs/{job_id}/events")
+async def job_events(job_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
+    job = _get(ctx, job_id)
+    snapshot = serialize_job(job)
+
+    async def stream():
+        yield f"data: {json.dumps(snapshot)}\n\n"
+        if snapshot["status"] in {"done", "failed"}:
+            return
+        async for data in subscribe(job_id):
+            yield f"data: {data}\n\n"
+            if json.loads(data).get("status") in {"done", "failed"}:
+                return
+
+    return StreamingResponse(stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache"})
+
+
+class ProgressIn(BaseModel):
+    status: str | None = None
+    progress: float | None = None
+    message: str | None = None
+    error: str | None = None
+    render_seconds: float | None = None
+    storage_key: str | None = None
+
+
+@router.post("/internal/jobs/{job_id}/progress", include_in_schema=False)
+def renderer_progress(
+    job_id: uuid.UUID,
+    body: ProgressIn,
+    x_internal_token: str = Header(...),
+    db: Session = Depends(get_db),
+):
+    if x_internal_token != settings.internal_token:
+        raise HTTPException(403, "Forbidden")
+    job = db.get(Job, job_id)
+    if not job:
+        raise HTTPException(404, "Job not found")
+    if body.status == "done" and job.artifact_id:
+        artifact = db.get(Artifact, job.artifact_id)
+        if artifact and body.storage_key:
+            artifact.storage_key = body.storage_key
+            artifact.status = "ready"
+        if body.render_seconds:
+            record_usage(
+                db, workspace_id=job.workspace_id, kind="render", provider="remotion",
+                quantity=body.render_seconds, unit="seconds", job=job,
+            )
+    update_job(
+        db, job, status=body.status, progress=body.progress, message=body.message, error=body.error,
+        result={"storage_key": body.storage_key} if body.storage_key else None,
+    )
+    return {"ok": True}

backend/app/routers/notebooks.py

@@ -0,0 +1,153 @@
+import json
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.concurrency import run_in_threadpool
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+
+from app.auth import Ctx, get_ctx
+from app.config import settings
+from app.llm.provider import provider_name
+from app.models import Chat, Citation, Document, Message, Notebook, NotebookDocument
+from app.pipeline.retrieval import grounded_answer, retrieve
+from app.services.jobs import record_usage
+
+router = APIRouter(prefix="/api/notebooks", tags=["notebooks"])
+
+
+class NotebookIn(BaseModel):
+    title: str = Field(min_length=1, max_length=300)
+    description: str | None = None
+
+
+class DocumentsIn(BaseModel):
+    document_ids: list[uuid.UUID]
+
+
+class ChatIn(BaseModel):
+    question: str = Field(min_length=1, max_length=4000)
+    chat_id: uuid.UUID | None = None
+
+
+def _get(ctx: Ctx, notebook_id: uuid.UUID) -> Notebook:
+    nb = ctx.db.scalar(
+        select(Notebook).where(Notebook.id == notebook_id, Notebook.workspace_id == ctx.workspace.id)
+    )
+    if not nb:
+        raise HTTPException(404, "Notebook not found")
+    return nb
+
+
+@router.post("")
+def create_notebook(body: NotebookIn, ctx: Ctx = Depends(get_ctx)):
+    nb = Notebook(workspace_id=ctx.workspace.id, title=body.title, description=body.description)
+    ctx.db.add(nb)
+    ctx.db.commit()
+    return {"id": str(nb.id), "title": nb.title}
+
+
+@router.get("")
+def list_notebooks(ctx: Ctx = Depends(get_ctx)):
+    rows = ctx.db.scalars(
+        select(Notebook).where(Notebook.workspace_id == ctx.workspace.id).order_by(Notebook.updated_at.desc())
+    ).all()
+    return [{"id": str(n.id), "title": n.title, "description": n.description} for n in rows]
+
+
+@router.get("/{notebook_id}")
+def get_notebook(notebook_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
+    nb = _get(ctx, notebook_id)
+    docs = ctx.db.scalars(
+        select(Document)
+        .join(NotebookDocument, NotebookDocument.document_id == Document.id)
+        .where(NotebookDocument.notebook_id == nb.id)
+        .order_by(Document.published_at.desc())
+    ).all()
+    return {
+        "id": str(nb.id),
+        "title": nb.title,
+        "description": nb.description,
+        "summary": nb.summary,
+        "documents": [
+            {"id": str(d.id), "title": d.title, "url": d.url,
+             "published_at": d.published_at.isoformat() if d.published_at else None}
+            for d in docs
+        ],
+    }
+
+
+@router.post("/{notebook_id}/documents")
+def add_documents(notebook_id: uuid.UUID, body: DocumentsIn, ctx: Ctx = Depends(get_ctx)):
+    nb = _get(ctx, notebook_id)
+    valid = set(
+        ctx.db.scalars(
+            select(Document.id).where(Document.id.in_(body.document_ids), Document.workspace_id == ctx.workspace.id)
+        ).all()
+    )
+    existing = set(
+        ctx.db.scalars(select(NotebookDocument.document_id).where(NotebookDocument.notebook_id == nb.id)).all()
+    )
+    for doc_id in valid - existing:
+        ctx.db.add(NotebookDocument(notebook_id=nb.id, document_id=doc_id))
+    ctx.db.commit()
+    return {"added": len(valid - existing)}
+
+
+def _sse(event: str, data: dict) -> str:
+    return f"event: {event}\ndata: {json.dumps(data)}\n\n"
+
+
+@router.post("/{notebook_id}/chat")
+async def chat(notebook_id: uuid.UUID, body: ChatIn, ctx: Ctx = Depends(get_ctx)):
+    nb = _get(ctx, notebook_id)
+    db = ctx.db
+    chat_row = db.get(Chat, body.chat_id) if body.chat_id else None
+    if not chat_row or chat_row.notebook_id != nb.id:
+        chat_row = Chat(workspace_id=ctx.workspace.id, notebook_id=nb.id, title=body.question[:120])
+        db.add(chat_row)
+        db.flush()
+    db.add(Message(chat_id=chat_row.id, role="user", content=body.question))
+    db.commit()
+
+    async def stream():
+        yield _sse("status", {"chat_id": str(chat_row.id), "message": "Scanning your archive"})
+        hits = await run_in_threadpool(retrieve, db, ctx.workspace.id, body.question, nb.id)
+        yield _sse("status", {"message": f"Found {len(hits)} relevant passages"})
+        answer = await run_in_threadpool(grounded_answer, body.question, hits)
+        msg = Message(chat_id=chat_row.id, role="assistant", content=answer.text)
+        db.add(msg)
+        db.flush()
+        citations = []
+        for marker, hit in answer.citations:
+            db.add(Citation(message_id=msg.id, chunk_id=hit.chunk.id, marker=marker, span=hit.chunk.text[:500]))
+            citations.append({
+                "marker": marker,
+                "chunk_id": str(hit.chunk.id),
+                "document_id": str(hit.document.id),
+                "title": hit.document.title,
+                "url": hit.document.url,
+                "heading": hit.chunk.heading,
+                "span": hit.chunk.text[:500],
+            })
+        db.commit()
+        if answer.prompt_tokens or answer.completion_tokens:
+            record_usage(
+                db,
+                workspace_id=ctx.workspace.id,
+                kind="llm",
+                provider=provider_name(),
+                model=settings.llm_model,
+                quantity=answer.prompt_tokens + answer.completion_tokens,
+                unit="tokens",
+            )
+        yield _sse("answer", {
+            "message_id": str(msg.id),
+            "text": answer.text,
+            "unsupported": answer.unsupported,
+            "citations": citations,
+        })
+        yield _sse("done", {})
+
+    return StreamingResponse(stream(), media_type="text/event-stream")

backend/app/routers/sources.py

@@ -0,0 +1,97 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from pydantic import BaseModel
+from sqlalchemy import func, select
+
+from app.auth import Ctx, get_ctx
+from app.models import Document, Job, Source
+from app.pipeline.ingest import normalize_feed_url
+from app.services.jobs import create_job, serialize_job
+from app.services.plans import effective_plan
+
+router = APIRouter(prefix="/api/sources", tags=["sources"])
+
+
+class SourceIn(BaseModel):
+    url: str
+
+
+def _serialize(source: Source, doc_count: int = 0) -> dict:
+    return {
+        "id": str(source.id),
+        "feed_url": source.feed_url,
+        "platform": source.platform,
+        "title": source.title,
+        "sync_status": source.sync_status,
+        "sync_error": source.sync_error,
+        "last_synced_at": source.last_synced_at.isoformat() if source.last_synced_at else None,
+        "document_count": doc_count,
+    }
+
+
+async def _enqueue_sync(request: Request, ctx: Ctx, source: Source) -> Job:
+    job = create_job(ctx.db, ctx.workspace.id, "ingest", {"source_id": str(source.id)})
+    await request.app.state.arq.enqueue_job("ingest_source_task", str(source.id), str(job.id))
+    return job
+
+
+@router.post("")
+async def create_source(body: SourceIn, request: Request, ctx: Ctx = Depends(get_ctx)):
+    feed_url, platform = normalize_feed_url(body.url)
+    existing = ctx.db.scalar(
+        select(Source).where(Source.workspace_id == ctx.workspace.id, Source.feed_url == feed_url)
+    )
+    if existing:
+        job = await _enqueue_sync(request, ctx, existing)
+        return {"source": _serialize(existing), "job": serialize_job(job)}
+    plan = effective_plan(ctx.db, ctx.workspace)
+    count = ctx.db.scalar(select(func.count()).select_from(Source).where(Source.workspace_id == ctx.workspace.id))
+    if count >= plan.sources:
+        raise HTTPException(402, {"code": "plan_limit", "message": f"Your plan includes {plan.sources} sources."})
+    source = Source(workspace_id=ctx.workspace.id, feed_url=feed_url, platform=platform, site_url=body.url)
+    ctx.db.add(source)
+    ctx.db.commit()
+    job = await _enqueue_sync(request, ctx, source)
+    return {"source": _serialize(source), "job": serialize_job(job)}
+
+
+@router.get("")
+def list_sources(ctx: Ctx = Depends(get_ctx)):
+    rows = ctx.db.execute(
+        select(Source, func.count(Document.id))
+        .outerjoin(Document, Document.source_id == Source.id)
+        .where(Source.workspace_id == ctx.workspace.id)
+        .group_by(Source.id)
+        .order_by(Source.created_at)
+    ).all()
+    return [_serialize(s, n) for s, n in rows]
+
+
+@router.get("/{source_id}/sync")
+async def sync_source(source_id: uuid.UUID, request: Request, ctx: Ctx = Depends(get_ctx)):
+    source = ctx.db.scalar(
+        select(Source).where(Source.id == source_id, Source.workspace_id == ctx.workspace.id)
+    )
+    if not source:
+        raise HTTPException(404, "Source not found")
+    job = await _enqueue_sync(request, ctx, source)
+    return {"source": _serialize(source), "job": serialize_job(job)}
+
+
+documents_router = APIRouter(prefix="/api/documents", tags=["documents"])
+
+
+@documents_router.get("")
+def list_documents(ctx: Ctx = Depends(get_ctx), limit: int = 500):
+    docs = ctx.db.scalars(
+        select(Document)
+        .where(Document.workspace_id == ctx.workspace.id)
+        .order_by(Document.published_at.desc())
+        .limit(min(limit, 5000))
+    ).all()
+    return [
+        {"id": str(d.id), "title": d.title, "url": d.url,
+         "published_at": d.published_at.isoformat() if d.published_at else None}
+        for d in docs
+    ]

backend/app/routers/storage.py

@@ -0,0 +1,91 @@
+"""Browser uploads go straight to R2 with presigned PUTs; reads redirect to presigned GETs."""
+
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import RedirectResponse
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+
+from app.auth import Ctx, get_ctx
+from app.config import settings
+from app.models import Upload
+from app.services.storage import keys, storage, workspace_owns_key
+
+router = APIRouter(prefix="/api/storage", tags=["storage"])
+
+ALLOWED_TYPES = {
+    "image/png",
+    "image/jpeg",
+    "image/webp",
+    "image/svg+xml",
+    "audio/mpeg",
+    "audio/wav",
+    "audio/webm",
+    "audio/mp4",
+    "video/mp4",
+    "application/pdf",
+    "text/markdown",
+    "text/plain",
+    "text/html",
+}
+
+
+class UploadStartIn(BaseModel):
+    filename: str = Field(max_length=300)
+    content_type: str
+    size_bytes: int = Field(gt=0)
+
+
+@router.post("/uploads")
+def start_upload(body: UploadStartIn, ctx: Ctx = Depends(get_ctx)):
+    if body.content_type not in ALLOWED_TYPES:
+        raise HTTPException(415, "File type not supported")
+    if body.size_bytes > settings.max_upload_bytes:
+        raise HTTPException(413, "File too large")
+    upload_id = uuid.uuid4()
+    key = keys.upload(ctx.workspace.id, upload_id, body.filename)
+    upload = Upload(
+        id=upload_id,
+        workspace_id=ctx.workspace.id,
+        user_id=ctx.user.id,
+        key=key,
+        filename=body.filename,
+        content_type=body.content_type,
+    )
+    ctx.db.add(upload)
+    ctx.db.commit()
+    return {
+        "upload_id": str(upload_id),
+        "key": key,
+        "upload_url": storage.presign_put(key, body.content_type, ttl=900),
+        "headers": {"Content-Type": body.content_type},
+    }
+
+
+@router.post("/uploads/{upload_id}/complete")
+def complete_upload(upload_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)):
+    upload = ctx.db.scalar(
+        select(Upload).where(Upload.id == upload_id, Upload.workspace_id == ctx.workspace.id)
+    )
+    if not upload:
+        raise HTTPException(404, "Upload not found")
+    head = storage.head(upload.key)
+    if not head:
+        raise HTTPException(409, "Object not found in storage yet")
+    size = int(head.get("ContentLength") or 0)
+    if size > settings.max_upload_bytes:
+        storage.delete(upload.key)
+        raise HTTPException(413, "File too large")
+    upload.size_bytes = size
+    upload.status = "complete"
+    ctx.db.commit()
+    return {"upload_id": str(upload.id), "key": upload.key, "size_bytes": size}
+
+
+@router.get("/objects/{key:path}")
+def read_object(key: str, download: bool = False, ctx: Ctx = Depends(get_ctx)):
+    if not workspace_owns_key(ctx.workspace.id, key):
+        raise HTTPException(404, "Not found")
+    name = key.rsplit("/", 1)[-1] if download else None
+    return RedirectResponse(storage.presign_get(key, download_name=name), status_code=302)

backend/app/routers/unsubscribe.py

@@ -0,0 +1,44 @@
+import uuid
+
+from fastapi import APIRouter, Depends
+from fastapi.responses import HTMLResponse
+from sqlalchemy.orm import Session
+
+from app.db import get_db
+from app.models import User
+from app.services.email import verify_unsubscribe_token
+
+router = APIRouter(prefix="/api", tags=["email"])
+
+_PAGE = """<!doctype html><html><body style="background:#000;color:#fff;font-family:Inter,Arial,sans-serif;
+display:grid;place-items:center;min-height:100vh;margin:0"><div style="text-align:center;padding:16px">
+<p style="color:#217cff;letter-spacing:.2em;text-transform:uppercase;font-size:13px">Notestack</p>
+<h1 style="font-weight:600">{title}</h1><p style="color:rgba(255,255,255,.7)">{body}</p></div></body></html>"""
+
+
+def _apply(token: str, db: Session) -> bool:
+    user_id = verify_unsubscribe_token(token)
+    if not user_id:
+        return False
+    try:
+        user = db.get(User, uuid.UUID(user_id))
+    except ValueError:
+        return False
+    if not user:
+        return False
+    user.email_unsubscribed = True
+    db.commit()
+    return True
+
+
+@router.get("/unsubscribe", response_class=HTMLResponse)
+def unsubscribe(token: str, db: Session = Depends(get_db)):
+    if _apply(token, db):
+        return _PAGE.format(title="You are unsubscribed", body="You will still get account and security emails.")
+    return HTMLResponse(_PAGE.format(title="Link not valid", body="This unsubscribe link is invalid."), 400)
+
+
+@router.post("/unsubscribe")
+def unsubscribe_one_click(token: str, db: Session = Depends(get_db)):
+    """RFC 8058 one click (List-Unsubscribe-Post)."""
+    return {"ok": _apply(token, db)}

backend/app/services/auth_identity.py

@@ -0,0 +1,122 @@
+"""Account resolution. One provider per account for life: Google accounts never get a password and
+email accounts never get linked to Google."""
+
+from dataclasses import dataclass
+from datetime import UTC, datetime
+
+import bcrypt
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.models import AuthProvider, Subscription, User, Workspace, WorkspaceMember
+
+
+class IdentityError(Exception):
+    def __init__(self, code: str, message: str, status: int = 400):
+        super().__init__(message)
+        self.code = code
+        self.message = message
+        self.status = status
+
+
+@dataclass
+class Resolved:
+    user: User
+    created: bool
+
+
+def normalize_email(email: str) -> str:
+    return email.strip().lower()
+
+
+def hash_password(password: str) -> str:
+    return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
+
+
+def verify_password(password: str, password_hash: str | None) -> bool:
+    if not password_hash:
+        return False
+    return bcrypt.checkpw(password.encode(), password_hash.encode())
+
+
+def validate_password(password: str) -> None:
+    if len(password) < 8:
+        raise IdentityError("weak_password", "Use at least 8 characters.")
+
+
+def wrong_provider_error(user: User) -> IdentityError:
+    if user.auth_provider == AuthProvider.GOOGLE:
+        return IdentityError("wrong_provider", "This email uses Google sign in. Continue with Google.", 409)
+    return IdentityError("wrong_provider", "This email uses a password. Sign in with email instead.", 409)
+
+
+def find_user(db: Session, email: str) -> User | None:
+    return db.scalar(select(User).where(User.email == normalize_email(email)))
+
+
+def _ensure_active(user: User) -> None:
+    if user.deleted_at or not user.is_active:
+        raise IdentityError("account_deleted", "This account was deleted. Contact support to restore it.", 403)
+
+
+def _bootstrap_workspace(db: Session, user: User) -> None:
+    first = (user.name or user.email.split("@")[0]).split(" ")[0]
+    workspace = Workspace(name=f"{first}'s workspace", owner_id=user.id)
+    db.add(workspace)
+    db.flush()
+    db.add(WorkspaceMember(workspace_id=workspace.id, user_id=user.id, role="owner"))
+    db.add(Subscription(workspace_id=workspace.id, plan="free"))
+
+
+def resolve_or_create_google_user(
+    db: Session, *, email: str, google_id: str, name: str | None, avatar_url: str | None
+) -> Resolved:
+    email = normalize_email(email)
+    user = db.scalar(select(User).where(User.google_id == google_id)) or find_user(db, email)
+    if user:
+        _ensure_active(user)
+        if user.auth_provider != AuthProvider.GOOGLE:
+            raise wrong_provider_error(user)
+        user.name = user.name or name
+        user.avatar_url = avatar_url or user.avatar_url
+        return Resolved(user, False)
+    user = User(
+        email=email, name=name, avatar_url=avatar_url, auth_provider=AuthProvider.GOOGLE, google_id=google_id
+    )
+    db.add(user)
+    db.flush()
+    _bootstrap_workspace(db, user)
+    return Resolved(user, True)
+
+
+def create_password_user(db: Session, *, email: str, name: str | None, password_hash: str) -> User:
+    email = normalize_email(email)
+    existing = find_user(db, email)
+    if existing:
+        _ensure_active(existing)
+        if existing.auth_provider != AuthProvider.EMAIL:
+            raise wrong_provider_error(existing)
+        raise IdentityError("exists", "An account with this email already exists. Sign in instead.", 409)
+    user = User(email=email, name=name, auth_provider=AuthProvider.EMAIL, password_hash=password_hash)
+    db.add(user)
+    db.flush()
+    _bootstrap_workspace(db, user)
+    return user
+
+
+def resolve_password_user(db: Session, *, email: str, password: str) -> User:
+    user = find_user(db, email)
+    if not user:
+        raise IdentityError("invalid_credentials", "Email or password is incorrect.", 401)
+    _ensure_active(user)
+    if user.auth_provider != AuthProvider.EMAIL:
+        raise wrong_provider_error(user)
+    if not verify_password(password, user.password_hash):
+        raise IdentityError("invalid_credentials", "Email or password is incorrect.", 401)
+    return user
+
+
+def soft_delete(db: Session, user: User) -> None:
+    user.deleted_at = datetime.now(UTC)
+    user.is_active = False
+    user.token_version += 1

backend/app/services/email.py

@@ -0,0 +1,214 @@
+"""EmailService over a provider interface (Resend in prod, console locally).
+
+Rule: no em dashes in any copy. Broadcast bodies are plain text and get HTML escaped.
+"""
+
+import hashlib
+import hmac
+import html
+import logging
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+from app.config import settings
+
+log = logging.getLogger(__name__)
+
+
+@dataclass
+class OutgoingEmail:
+    to: str
+    subject: str
+    html: str
+    text: str
+    headers: dict[str, str] | None = None
+
+
+class BaseEmailProvider(ABC):
+    @abstractmethod
+    def send(self, email: OutgoingEmail) -> str | None: ...
+
+
+class ResendEmailProvider(BaseEmailProvider):
+    def __init__(self, api_key: str):
+        import resend
+
+        resend.api_key = api_key
+        self._resend = resend
+
+    def send(self, email: OutgoingEmail) -> str | None:
+        params = {
+            "from": settings.from_email,
+            "to": [email.to],
+            "subject": email.subject,
+            "html": email.html,
+            "text": email.text,
+        }
+        if email.headers:
+            params["headers"] = email.headers
+        result = self._resend.Emails.send(params)
+        return result.get("id") if isinstance(result, dict) else None
+
+
+class ConsoleEmailProvider(BaseEmailProvider):
+    sent: list[OutgoingEmail] = []
+
+    def send(self, email: OutgoingEmail) -> str | None:
+        ConsoleEmailProvider.sent.append(email)
+        log.info("[email] to=%s subject=%s\n%s", email.to, email.subject, email.text)
+        return None
+
+
+def make_unsubscribe_token(user_id: str) -> str:
+    sig = hmac.new(settings.unsubscribe_secret.encode(), user_id.encode(), hashlib.sha256).hexdigest()[:32]
+    return f"{user_id}.{sig}"
+
+
+def verify_unsubscribe_token(token: str) -> str | None:
+    user_id, _, _sig = token.partition(".")
+    if user_id and hmac.compare_digest(make_unsubscribe_token(user_id), token):
+        return user_id
+    return None
+
+
+def _layout(title: str, body_html: str, footer_html: str = "") -> str:
+    # Inline styles only; email clients ignore <style>. Black, #217cff, white.
+    return f"""<!doctype html><html><body style="margin:0;background:#000000;padding:32px 16px;
+font-family:Inter,Arial,sans-serif;color:#ffffff">
+<table role="presentation" width="100%" cellpadding="0" cellspacing="0"><tr><td align="center">
+<table role="presentation" width="560" cellpadding="0" cellspacing="0" style="max-width:560px;width:100%;
+border:1px solid rgba(33,124,255,0.35);border-radius:16px;padding:32px;background:#000000">
+<tr><td style="font-size:13px;letter-spacing:0.2em;text-transform:uppercase;color:#217cff">Notestack</td></tr>
+<tr><td style="padding-top:16px;font-size:22px;font-weight:600;color:#ffffff">{html.escape(title)}</td></tr>
+<tr><td style="padding-top:16px;font-size:15px;line-height:1.6;color:rgba(255,255,255,0.85)">{body_html}</td></tr>
+<tr><td style="padding-top:28px;font-size:12px;color:rgba(255,255,255,0.5)">{footer_html}</td></tr>
+</table></td></tr></table></body></html>"""
+
+
+def _button(label: str, url: str) -> str:
+    return (
+        f'<p style="padding-top:12px"><a href="{html.escape(url)}" style="display:inline-block;'
+        f"background:#217cff;color:#ffffff;text-decoration:none;padding:12px 20px;border-radius:999px;"
+        f'font-weight:600">{html.escape(label)}</a></p>'
+    )
+
+
+def _code_block(code: str) -> str:
+    return (
+        f'<p style="font-family:JetBrains Mono,monospace;font-size:32px;letter-spacing:0.3em;'
+        f'color:#ffffff;padding:8px 0">{html.escape(code)}</p>'
+    )
+
+
+class EmailService:
+    def __init__(self, provider: BaseEmailProvider):
+        self.provider = provider
+
+    def _send(self, email: OutgoingEmail) -> bool:
+        try:
+            self.provider.send(email)
+            return True
+        except Exception:  # never let email failures break the request
+            log.exception("email send failed to=%s subject=%s", email.to, email.subject)
+            return False
+
+    # Transactional
+
+    def send_verification_code(self, to: str, code: str) -> bool:
+        body = f"Enter this code to finish creating your account:{_code_block(code)}It expires in 10 minutes."
+        return self._send(
+            OutgoingEmail(
+                to=to,
+                subject=f"{code} is your Notestack code",
+                html=_layout("Confirm your email", body, "If you did not sign up, ignore this email."),
+                text=f"Your Notestack code is {code}. It expires in 10 minutes.",
+            )
+        )
+
+    def send_password_reset_code(self, to: str, code: str) -> bool:
+        body = f"Use this code to reset your password:{_code_block(code)}It expires in 10 minutes."
+        return self._send(
+            OutgoingEmail(
+                to=to,
+                subject="Reset your Notestack password",
+                html=_layout("Password reset", body, "If you did not ask for this, you can ignore it."),
+                text=f"Your Notestack password reset code is {code}. It expires in 10 minutes.",
+            )
+        )
+
+    def send_welcome(self, to: str, name: str | None, user_id: str) -> bool:
+        first = (name or "there").split(" ")[0]
+        url = f"{settings.frontend_url}/app"
+        body = (
+            f"Hi {html.escape(first)},<br><br>Your archive is about to come into orbit. Paste your Substack "
+            "URL and Notestack will index your posts, learn your voice and get your first audio overview ready."
+            + _button("Open Mission Control", url)
+        )
+        return self._send(
+            OutgoingEmail(
+                to=to,
+                subject="Welcome to Notestack",
+                html=_layout("Welcome aboard", body, self._unsubscribe_footer(user_id)),
+                text=f"Hi {first}, welcome to Notestack. Open Mission Control: {url}",
+                headers=self._list_unsubscribe(user_id),
+            )
+        )
+
+    def send_job_ready(self, to: str, what: str, url: str) -> bool:
+        body = f"Your {html.escape(what)} is ready." + _button("View it", url)
+        return self._send(
+            OutgoingEmail(
+                to=to,
+                subject=f"Your {what} is ready",
+                html=_layout("Launch complete", body),
+                text=f"Your {what} is ready: {url}",
+            )
+        )
+
+    # Internal alerts
+
+    def send_alert(self, subject: str, text: str) -> bool:
+        return self._send(
+            OutgoingEmail(
+                to=settings.alerts_email,
+                subject=f"[Notestack alert] {subject}",
+                html=_layout(subject, f"<pre style='white-space:pre-wrap'>{html.escape(text)}</pre>"),
+                text=text,
+            )
+        )
+
+    # Broadcast
+
+    def send_blast_email(self, user_id: str, user_email: str, user_name: str | None, subject: str, body: str) -> bool:
+        first = (user_name or "there").split(" ")[0]
+        body_html = f"Hi {html.escape(first)},<br><br>" + html.escape(body).replace("\n", "<br>")
+        return self._send(
+            OutgoingEmail(
+                to=user_email,
+                subject=subject,
+                html=_layout(subject, body_html, self._unsubscribe_footer(user_id)),
+                text=f"Hi {first},\n\n{body}\n\nUnsubscribe: {self._unsubscribe_url(user_id)}",
+                headers=self._list_unsubscribe(user_id),
+            )
+        )
+
+    def _unsubscribe_url(self, user_id: str) -> str:
+        return f"{settings.api_url}/api/unsubscribe?token={make_unsubscribe_token(user_id)}"
+
+    def _unsubscribe_footer(self, user_id: str) -> str:
+        return f'<a href="{self._unsubscribe_url(user_id)}" style="color:rgba(255,255,255,0.5)">Unsubscribe</a>'
+
+    def _list_unsubscribe(self, user_id: str) -> dict[str, str]:
+        return {
+            "List-Unsubscribe": f"<{self._unsubscribe_url(user_id)}>",
+            "List-Unsubscribe-Post": "List-Unsubscribe=One-Click",
+        }
+
+
+def _build() -> EmailService:
+    if settings.email_provider == "resend" and settings.resend_api_key:
+        return EmailService(ResendEmailProvider(settings.resend_api_key))
+    return EmailService(ConsoleEmailProvider())
+
+
+email_service = _build()

backend/app/services/email_verification.py

@@ -0,0 +1,117 @@
+"""6 digit codes: hashed, 10 minute TTL, 5 attempts, 60s resend cooldown, purged after 1 day."""
+
+import hashlib
+import hmac
+import secrets
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import delete, select
+from sqlalchemy.orm import Session
+
+from app.config import settings
+from app.models import EmailVerificationCode, VerificationPurpose
+
+CODE_TTL = timedelta(minutes=10)
+MAX_ATTEMPTS = 5
+RESEND_COOLDOWN = timedelta(seconds=60)
+PURGE_AFTER = timedelta(days=1)
+
+
+class VerificationError(Exception):
+    def __init__(self, code: str, message: str):
+        super().__init__(message)
+        self.code = code
+        self.message = message
+
+
+@dataclass
+class IssuedCode:
+    code: str
+    record: EmailVerificationCode
+
+
+def _hash(email: str, code: str) -> str:
+    return hmac.new(settings.jwt_secret.encode(), f"{email}:{code}".encode(), hashlib.sha256).hexdigest()
+
+
+def _aware(dt: datetime) -> datetime:
+    return dt if dt.tzinfo else dt.replace(tzinfo=UTC)
+
+
+def _latest(db: Session, email: str, purpose: VerificationPurpose) -> EmailVerificationCode | None:
+    return db.scalar(
+        select(EmailVerificationCode)
+        .where(
+            EmailVerificationCode.email == email,
+            EmailVerificationCode.purpose == purpose,
+            EmailVerificationCode.consumed_at.is_(None),
+        )
+        .order_by(EmailVerificationCode.created_at.desc())
+        .limit(1)
+    )
+
+
+def issue_code(
+    db: Session,
+    email: str,
+    purpose: VerificationPurpose,
+    pending_name: str | None = None,
+    pending_password_hash: str | None = None,
+) -> IssuedCode:
+    now = datetime.now(UTC)
+    latest = _latest(db, email, purpose)
+    if latest and now - _aware(latest.created_at) < RESEND_COOLDOWN:
+        raise VerificationError("cooldown", "Please wait a minute before requesting another code.")
+    clear_codes(db, email, purpose)
+    code = f"{secrets.randbelow(1_000_000):06d}"
+    record = EmailVerificationCode(
+        email=email,
+        purpose=purpose,
+        code_hash=_hash(email, code),
+        pending_name=pending_name,
+        pending_password_hash=pending_password_hash,
+        expires_at=now + CODE_TTL,
+    )
+    db.add(record)
+    db.flush()
+    return IssuedCode(code=code, record=record)
+
+
+def check_code(db: Session, email: str, purpose: VerificationPurpose, code: str) -> EmailVerificationCode:
+    """Validate without consuming (used by the password reset 'check' step)."""
+    record = _latest(db, email, purpose)
+    if not record:
+        raise VerificationError("not_found", "No active code. Request a new one.")
+    if datetime.now(UTC) > _aware(record.expires_at):
+        raise VerificationError("expired", "That code expired. Request a new one.")
+    if record.attempts >= MAX_ATTEMPTS:
+        raise VerificationError("too_many_attempts", "Too many attempts. Request a new code.")
+    if not hmac.compare_digest(record.code_hash, _hash(email, code.strip())):
+        record.attempts += 1
+        db.commit()
+        raise VerificationError("invalid", "That code is not right.")
+    return record
+
+
+def consume_code(db: Session, email: str, purpose: VerificationPurpose, code: str) -> EmailVerificationCode:
+    record = check_code(db, email, purpose, code)
+    record.consumed_at = datetime.now(UTC)
+    return record
+
+
+def clear_codes(db: Session, email: str, purpose: VerificationPurpose) -> None:
+    db.execute(
+        delete(EmailVerificationCode).where(
+            EmailVerificationCode.email == email, EmailVerificationCode.purpose == purpose
+        )
+    )
+
+
+def purge_old_codes(db: Session) -> None:
+    db.execute(
+        delete(EmailVerificationCode).where(
+            EmailVerificationCode.created_at < datetime.now(UTC) - PURGE_AFTER
+        )
+    )
+    db.commit()

backend/app/services/jobs.py

@@ -0,0 +1,119 @@
+"""Job rows + progress fan-out over Redis pub/sub (consumed by the SSE endpoint)."""
+
+import json
+import uuid
+from datetime import UTC, datetime
+
+import redis
+import redis.asyncio as aredis
+from sqlalchemy.orm import Session
+
+from app.config import settings
+from app.models import Job, UsageEvent
+
+_sync_redis: redis.Redis | None = None
+
+
+def _redis() -> redis.Redis:
+    global _sync_redis
+    if _sync_redis is None:
+        _sync_redis = redis.Redis.from_url(settings.redis_url)
+    return _sync_redis
+
+
+def channel(job_id: uuid.UUID | str) -> str:
+    return f"job:{job_id}"
+
+
+def serialize_job(job: Job) -> dict:
+    return {
+        "id": str(job.id),
+        "kind": job.kind,
+        "status": job.status,
+        "progress": job.progress,
+        "message": job.message,
+        "error": job.error,
+        "result": job.result,
+        "artifact_id": str(job.artifact_id) if job.artifact_id else None,
+        "cost_usd": job.cost_usd,
+        "created_at": job.created_at.isoformat() if job.created_at else None,
+        "finished_at": job.finished_at.isoformat() if job.finished_at else None,
+    }
+
+
+def create_job(db: Session, workspace_id: uuid.UUID, kind: str, params: dict | None = None) -> Job:
+    job = Job(workspace_id=workspace_id, kind=kind, params=params or {})
+    db.add(job)
+    db.commit()
+    return job
+
+
+def update_job(
+    db: Session,
+    job: Job,
+    *,
+    status: str | None = None,
+    progress: float | None = None,
+    message: str | None = None,
+    error: str | None = None,
+    result: dict | None = None,
+) -> None:
+    if status:
+        job.status = status
+        if status in {"done", "failed"}:
+            job.finished_at = datetime.now(UTC)
+    if progress is not None:
+        job.progress = max(0.0, min(1.0, progress))
+    if message is not None:
+        job.message = message
+    if error is not None:
+        job.error = error
+    if result is not None:
+        job.result = result
+    db.commit()
+    try:
+        _redis().publish(channel(job.id), json.dumps(serialize_job(job)))
+    except redis.RedisError:
+        pass  # progress is also readable from GET /jobs/{id}
+
+
+async def subscribe(job_id: uuid.UUID):
+    client = aredis.Redis.from_url(settings.redis_url)
+    pubsub = client.pubsub()
+    await pubsub.subscribe(channel(job_id))
+    try:
+        async for msg in pubsub.listen():
+            if msg["type"] == "message":
+                yield msg["data"].decode()
+    finally:
+        await pubsub.unsubscribe(channel(job_id))
+        await client.aclose()
+
+
+def record_usage(
+    db: Session,
+    *,
+    workspace_id: uuid.UUID,
+    kind: str,
+    provider: str,
+    quantity: float,
+    unit: str,
+    model: str | None = None,
+    cost_usd: float = 0.0,
+    job: Job | None = None,
+) -> None:
+    db.add(
+        UsageEvent(
+            workspace_id=workspace_id,
+            job_id=job.id if job else None,
+            kind=kind,
+            provider=provider,
+            model=model,
+            quantity=quantity,
+            unit=unit,
+            cost_usd=cost_usd,
+        )
+    )
+    if job:
+        job.cost_usd = (job.cost_usd or 0.0) + cost_usd
+    db.commit()

backend/app/services/plans.py

@@ -0,0 +1,104 @@
+"""Three tiers. While BILLING_ENABLED is false every workspace gets Studio limits for free."""
+
+from dataclasses import asdict, dataclass
+
+from sqlalchemy.orm import Session
+
+from app.config import settings
+from app.models import Subscription, Workspace
+
+
+@dataclass(frozen=True)
+class Plan:
+    id: str
+    name: str
+    tagline: str
+    price_monthly_usd: int
+    sources: int
+    indexed_posts: int
+    audio_minutes: int
+    video_minutes: int
+    launch_kits: int  # -1 = unlimited
+    voice_cloning: bool
+    brand_kit: bool
+    features: tuple[str, ...]
+
+
+PLANS: dict[str, Plan] = {
+    "free": Plan(
+        id="free",
+        name="Free",
+        tagline="For trying Notestack on your archive",
+        price_monthly_usd=0,
+        sources=1,
+        indexed_posts=50,
+        audio_minutes=10,
+        video_minutes=3,
+        launch_kits=5,
+        voice_cloning=False,
+        brand_kit=False,
+        features=(
+            "1 source, 50 indexed posts",
+            "Grounded research chat with citations",
+            "10 min of audio overviews a month",
+            "5 Launch Kits a month",
+        ),
+    ),
+    "writer": Plan(
+        id="writer",
+        name="Writer",
+        tagline="For writers publishing every week",
+        price_monthly_usd=19,
+        sources=3,
+        indexed_posts=500,
+        audio_minutes=60,
+        video_minutes=30,
+        launch_kits=50,
+        voice_cloning=True,
+        brand_kit=True,
+        features=(
+            "3 sources, 500 indexed posts",
+            "60 min of audio overviews a month",
+            "30 min of video renders a month",
+            "50 Launch Kits a month",
+            "Voice cloning with consent",
+            "Your brand colors and logo",
+        ),
+    ),
+    "studio": Plan(
+        id="studio",
+        name="Studio",
+        tagline="For publications and power users",
+        price_monthly_usd=49,
+        sources=10,
+        indexed_posts=5000,
+        audio_minutes=240,
+        video_minutes=120,
+        launch_kits=-1,
+        voice_cloning=True,
+        brand_kit=True,
+        features=(
+            "10 sources, 5,000 indexed posts",
+            "240 min of audio overviews a month",
+            "120 min of video renders a month",
+            "Unlimited Launch Kits",
+            "Launchpad calendar and resurfacing",
+            "Priority rendering",
+        ),
+    ),
+}
+
+
+def plan_dict(plan: Plan) -> dict:
+    data = asdict(plan)
+    data["features"] = list(plan.features)
+    return data
+
+
+def effective_plan(db: Session, workspace: Workspace) -> Plan:
+    if not settings.billing_enabled:
+        return PLANS["studio"]
+    sub = db.query(Subscription).filter_by(workspace_id=workspace.id).one_or_none()
+    if not sub or sub.status not in {"active", "trialing"}:
+        return PLANS["free"]
+    return PLANS.get(sub.plan, PLANS["free"])

backend/app/services/storage.py

@@ -0,0 +1,130 @@
+"""Cloudflare R2 storage (S3 API). Postgres stores keys; bytes live here.
+
+Locally, point R2_ENDPOINT_URL at MinIO and nothing else changes.
+"""
+
+import hashlib
+import mimetypes
+import re
+import uuid
+from functools import cached_property
+
+import boto3
+from botocore.client import Config
+from botocore.exceptions import ClientError
+
+from app.config import settings
+
+
+class keys:  # noqa: N801  (namespace for key builders)
+    @staticmethod
+    def raw_html(workspace_id: uuid.UUID, source_id: uuid.UUID, document_id: uuid.UUID) -> str:
+        return f"ws/{workspace_id}/sources/{source_id}/documents/{document_id}/raw.html"
+
+    @staticmethod
+    def upload(workspace_id: uuid.UUID, upload_id: uuid.UUID, filename: str) -> str:
+        return f"ws/{workspace_id}/uploads/{upload_id}/{safe_filename(filename)}"
+
+    @staticmethod
+    def artifact(workspace_id: uuid.UUID, artifact_id: uuid.UUID, variant: str, ext: str) -> str:
+        return f"ws/{workspace_id}/artifacts/{artifact_id}/{variant}.{ext.lstrip('.')}"
+
+    @staticmethod
+    def tts_cache(workspace_id: uuid.UUID, script: str, voice_id: str, settings_json: str) -> str:
+        digest = hashlib.sha256(f"{script}|{voice_id}|{settings_json}".encode()).hexdigest()
+        return f"ws/{workspace_id}/tts-cache/{digest}.mp3"
+
+    @staticmethod
+    def dspy_program(module: str, version: str) -> str:
+        return f"system/dspy/{module}/{version}.json"
+
+
+def safe_filename(name: str) -> str:
+    name = name.rsplit("/", 1)[-1].rsplit("\\", 1)[-1]
+    name = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip("-.")
+    return name[:120] or "file"
+
+
+def workspace_owns_key(workspace_id: uuid.UUID, key: str) -> bool:
+    return key.startswith(f"ws/{workspace_id}/") and ".." not in key
+
+
+class Storage:
+    def __init__(self, bucket: str | None = None):
+        self.bucket = bucket or settings.r2_bucket
+
+    @cached_property
+    def client(self):
+        return boto3.client(
+            "s3",
+            endpoint_url=settings.r2_endpoint,
+            aws_access_key_id=settings.r2_access_key_id,
+            aws_secret_access_key=settings.r2_secret_access_key,
+            region_name="auto",
+            config=Config(signature_version="s3v4", s3={"addressing_style": "path"}),
+        )
+
+    def ensure_bucket(self) -> None:
+        """Local dev convenience (MinIO). On R2 create the bucket in the dashboard."""
+        try:
+            self.client.head_bucket(Bucket=self.bucket)
+        except ClientError:
+            self.client.create_bucket(Bucket=self.bucket)
+
+    def put_bytes(self, key: str, data: bytes, content_type: str | None = None) -> str:
+        content_type = content_type or mimetypes.guess_type(key)[0] or "application/octet-stream"
+        self.client.put_object(Bucket=self.bucket, Key=key, Body=data, ContentType=content_type)
+        return key
+
+    def put_text(self, key: str, text: str, content_type: str = "text/plain; charset=utf-8") -> str:
+        return self.put_bytes(key, text.encode("utf-8"), content_type)
+
+    def get_bytes(self, key: str) -> bytes:
+        return self.client.get_object(Bucket=self.bucket, Key=key)["Body"].read()
+
+    def head(self, key: str) -> dict | None:
+        try:
+            return self.client.head_object(Bucket=self.bucket, Key=key)
+        except ClientError as exc:
+            if exc.response.get("Error", {}).get("Code") in {"404", "NoSuchKey", "NotFound"}:
+                return None
+            raise
+
+    def exists(self, key: str) -> bool:
+        return self.head(key) is not None
+
+    def delete(self, key: str) -> None:
+        self.client.delete_object(Bucket=self.bucket, Key=key)
+
+    def delete_prefix(self, prefix: str) -> int:
+        deleted = 0
+        paginator = self.client.get_paginator("list_objects_v2")
+        for page in paginator.paginate(Bucket=self.bucket, Prefix=prefix):
+            objects = [{"Key": o["Key"]} for o in page.get("Contents", [])]
+            if objects:
+                self.client.delete_objects(Bucket=self.bucket, Delete={"Objects": objects})
+                deleted += len(objects)
+        return deleted
+
+    def presign_get(self, key: str, ttl: int | None = None, download_name: str | None = None) -> str:
+        params = {"Bucket": self.bucket, "Key": key}
+        if download_name:
+            params["ResponseContentDisposition"] = f'attachment; filename="{safe_filename(download_name)}"'
+        return self.client.generate_presigned_url(
+            "get_object", Params=params, ExpiresIn=ttl or settings.r2_presign_ttl_seconds
+        )
+
+    def presign_put(self, key: str, content_type: str, ttl: int | None = None) -> str:
+        return self.client.generate_presigned_url(
+            "put_object",
+            Params={"Bucket": self.bucket, "Key": key, "ContentType": content_type},
+            ExpiresIn=ttl or settings.r2_presign_ttl_seconds,
+        )
+
+    def public_url(self, key: str) -> str | None:
+        if not settings.r2_public_base_url:
+            return None
+        return f"{settings.r2_public_base_url.rstrip('/')}/{key}"
+
+
+storage = Storage()

backend/app/worker.py

@@ -0,0 +1,139 @@
+"""arq worker: long running jobs + scheduled email batches.
+
+Run: arq app.worker.WorkerSettings
+"""
+
+import logging
+import uuid
+from datetime import UTC, datetime
+
+import httpx
+from arq import cron
+from arq.connections import RedisSettings
+from sqlalchemy import select
+
+from app.config import settings
+from app.db import SessionLocal
+from app.models import (
+    Artifact,
+    Job,
+    Source,
+    Subscription,
+    UpdateEmail,
+    UpdateEmailSend,
+    User,
+    Workspace,
+    WorkspaceMember,
+)
+from app.pipeline.ingest import ingest_source
+from app.services.email import email_service
+from app.services.email_verification import purge_old_codes
+from app.services.jobs import update_job
+from app.services.plans import effective_plan
+from app.services.storage import keys, storage
+
+log = logging.getLogger(__name__)
+
+
+async def ingest_source_task(ctx, source_id: str, job_id: str) -> dict:
+    with SessionLocal() as db:
+        job = db.get(Job, uuid.UUID(job_id))
+        source = db.get(Source, uuid.UUID(source_id))
+        if not job or not source:
+            return {"error": "missing"}
+        try:
+            plan = effective_plan(db, db.get(Workspace, source.workspace_id))
+            result = ingest_source(db, source, job, max_posts=min(plan.indexed_posts, 200))
+            update_job(db, job, status="done", progress=1.0, message="All posts in orbit", result=result)
+            return result
+        except Exception as exc:
+            log.exception("ingest failed source=%s", source_id)
+            db.rollback()
+            source.sync_status = "error"
+            source.sync_error = str(exc)[:500]
+            update_job(db, job, status="failed", error=str(exc)[:1000], message="Ingestion failed")
+            return {"error": str(exc)}
+
+
+async def render_task(ctx, artifact_id: str, job_id: str, composition: str, props: dict, fmt: str) -> dict:
+    """Hand off to the Remotion renderer with a presigned PUT so it never holds R2 credentials."""
+    with SessionLocal() as db:
+        job = db.get(Job, uuid.UUID(job_id))
+        artifact = db.get(Artifact, uuid.UUID(artifact_id))
+        ext = "png" if fmt == "png" else "mp4"
+        key = keys.artifact(artifact.workspace_id, artifact.id, composition.lower(), ext)
+        content_type = "image/png" if ext == "png" else "video/mp4"
+        update_job(db, job, status="running", progress=0.01, message="T-minus: preparing render")
+        async with httpx.AsyncClient(timeout=30) as client:
+            resp = await client.post(
+                f"{settings.renderer_url}/render",
+                headers={"x-internal-token": settings.internal_token},
+                json={
+                    "jobId": job_id,
+                    "compositionId": composition,
+                    "props": props,
+                    "format": fmt,
+                    "uploadUrl": storage.presign_put(key, content_type, ttl=6 * 3600),
+                    "uploadContentType": content_type,
+                    "callbackUrl": f"{settings.api_url}/api/internal/jobs/{job_id}/progress",
+                    "storageKey": key,
+                },
+            )
+            resp.raise_for_status()
+        return {"accepted": True, "key": key}
+
+
+async def update_email_batch(ctx) -> None:
+    """Hourly check; sends one daily batch per scheduled campaign at its send hour (UTC)."""
+    now = datetime.now(UTC)
+    with SessionLocal() as db:
+        campaigns = db.scalars(select(UpdateEmail).where(UpdateEmail.status.in_(["scheduled", "running"]))).all()
+        for campaign in campaigns:
+            hour = campaign.send_hour if campaign.send_hour >= 0 else settings.update_email_send_hour
+            if now.hour != hour:
+                continue
+            if campaign.last_batch_at and campaign.last_batch_at.date() == now.date():
+                continue
+            already = select(UpdateEmailSend.user_id).where(UpdateEmailSend.update_email_id == campaign.id)
+            q = select(User).where(
+                User.is_active.is_(True), User.email_unsubscribed.is_(False), User.id.not_in(already)
+            )
+            if campaign.user_filter != "all":
+                plans = {"paid": ["writer", "studio"]}.get(campaign.user_filter, [campaign.user_filter])
+                q = (
+                    q.join(WorkspaceMember, WorkspaceMember.user_id == User.id)
+                    .join(Subscription, Subscription.workspace_id == WorkspaceMember.workspace_id)
+                    .where(Subscription.plan.in_(plans))
+                )
+            users = db.scalars(q.limit(campaign.batch_size)).unique().all()
+            campaign.status = "running"
+            for user in users:
+                ok = email_service.send_blast_email(
+                    str(user.id), user.email, user.name, campaign.subject, campaign.body
+                )
+                db.add(UpdateEmailSend(update_email_id=campaign.id, user_id=user.id, status="sent" if ok else "failed"))
+                campaign.sent_count += int(ok)
+                campaign.failed_count += int(not ok)
+            campaign.total_users = campaign.sent_count + campaign.failed_count
+            campaign.last_batch_at = now
+            if len(users) < campaign.batch_size:
+                campaign.status = "completed"
+            db.commit()
+
+
+async def daily_cleanup(ctx) -> None:
+    with SessionLocal() as db:
+        purge_old_codes(db)
+
+
+async def startup(ctx) -> None:
+    logging.basicConfig(level=logging.INFO)
+
+
+class WorkerSettings:
+    functions = [ingest_source_task, render_task]
+    cron_jobs = [cron(update_email_batch, minute=5), cron(daily_cleanup, hour=3, minute=17)]
+    redis_settings = RedisSettings.from_dsn(settings.redis_url)
+    on_startup = startup
+    max_jobs = 10
+    job_timeout = 60 * 30

backend/pyproject.toml

@@ -0,0 +1,10 @@
+[tool.ruff]
+line-length = 120
+target-version = "py312"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "B", "UP"]
+ignore = ["B008"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]

backend/requirements-dev.txt

@@ -0,0 +1,3 @@
+-r requirements.txt
+pytest==8.*
+ruff==0.8.*

backend/requirements.txt

@@ -0,0 +1,22 @@
+fastapi==0.115.*
+uvicorn[standard]==0.32.*
+sqlalchemy==2.0.*
+psycopg[binary]==3.2.*
+alembic==1.14.*
+pgvector==0.3.*
+pydantic==2.*
+pydantic-settings==2.*
+email-validator==2.*
+pyjwt==2.*
+bcrypt==4.*
+google-auth==2.*
+requests==2.*
+boto3==1.35.*
+resend==2.*
+httpx==0.28.*
+arq==0.26.*
+redis==5.*
+dspy==2.6.*
+feedparser==6.*
+beautifulsoup4==4.*
+tiktoken==0.8.*

backend/tests/conftest.py

@@ -0,0 +1,41 @@
+import os
+
+os.environ.setdefault("DATABASE_URL", "sqlite://")
+os.environ.setdefault("EMAIL_PROVIDER", "console")
+os.environ.setdefault("JWT_SECRET", "test-secret-that-is-at-least-32-bytes-long")
+os.environ.setdefault("BILLING_ENABLED", "false")
+
+import pytest  # noqa: E402
+from fastapi.testclient import TestClient  # noqa: E402
+from sqlalchemy import create_engine  # noqa: E402
+from sqlalchemy.orm import sessionmaker  # noqa: E402
+from sqlalchemy.pool import StaticPool  # noqa: E402
+
+import app.models  # noqa: E402,F401
+from app.db import Base, get_db  # noqa: E402
+from app.main import app  # noqa: E402
+from app.services.email import ConsoleEmailProvider  # noqa: E402
+
+
+@pytest.fixture()
+def db_session():
+    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    Base.metadata.create_all(engine)
+    Session = sessionmaker(bind=engine, expire_on_commit=False)
+    session = Session()
+    yield session
+    session.close()
+
+
+@pytest.fixture()
+def client(db_session):
+    app.dependency_overrides[get_db] = lambda: db_session
+    ConsoleEmailProvider.sent.clear()
+    yield TestClient(app)  # no context manager: skips lifespan (no Redis needed)
+    app.dependency_overrides.clear()
+
+
+def last_code() -> str:
+    import re
+
+    return re.search(r"\b(\d{6})\b", ConsoleEmailProvider.sent[-1].text).group(1)

backend/tests/test_auth.py

@@ -0,0 +1,80 @@
+from app.services.email import ConsoleEmailProvider
+from tests.conftest import last_code
+
+
+def _signup(client, email="writer@example.com", password="stardust-42"):
+    r = client.post("/api/auth/email/register/start", json={"email": email, "password": password, "name": "Ada L"})
+    assert r.status_code == 200, r.text
+    r = client.post("/api/auth/email/register/verify", json={"email": email, "code": last_code()})
+    assert r.status_code == 200, r.text
+    return r.json()
+
+
+def test_signup_login_me_logout(client):
+    data = _signup(client)
+    assert data["created"] is True
+    assert any(e.subject == "Welcome to Notestack" for e in ConsoleEmailProvider.sent)
+    headers = {"Authorization": f"Bearer {data['access_token']}"}
+    assert client.get("/api/auth/me", headers=headers).json()["email"] == "writer@example.com"
+
+    r = client.post("/api/auth/email/login", json={"email": "WRITER@example.com", "password": "stardust-42"})
+    assert r.status_code == 200
+
+    assert client.post("/api/auth/logout", headers=headers).status_code == 200
+    assert client.get("/api/auth/me", headers=headers).status_code == 401  # token_version bumped
+
+
+def test_wrong_code_and_attempt_limit(client):
+    client.post("/api/auth/email/register/start", json={"email": "a@b.co", "password": "longenough"})
+    for _ in range(5):
+        r = client.post("/api/auth/email/register/verify", json={"email": "a@b.co", "code": "000000"})
+        if last_code() == "000000":
+            return  # astronomically unlikely collision
+        assert r.status_code == 400
+    r = client.post("/api/auth/email/register/verify", json={"email": "a@b.co", "code": last_code()})
+    assert r.json()["detail"]["code"] == "too_many_attempts"
+
+
+def test_resend_cooldown(client):
+    client.post("/api/auth/email/register/start", json={"email": "c@d.co", "password": "longenough"})
+    r = client.post("/api/auth/email/register/resend", json={"email": "c@d.co"})
+    assert r.status_code == 429
+
+
+def test_bad_password_and_weak_password(client):
+    _signup(client)
+    r = client.post("/api/auth/email/login", json={"email": "writer@example.com", "password": "nope-nope"})
+    assert r.status_code == 401
+    r = client.post("/api/auth/email/register/start", json={"email": "x@y.co", "password": "short"})
+    assert r.status_code == 400
+
+
+def test_password_reset_revokes_old_tokens(client):
+    data = _signup(client)
+    old = {"Authorization": f"Bearer {data['access_token']}"}
+    client.post("/api/auth/password/forgot/start", json={"email": "writer@example.com"})
+    code = last_code()
+    r = client.post("/api/auth/password/forgot/check", json={"email": "writer@example.com", "code": code})
+    assert r.status_code == 200
+    r = client.post("/api/auth/password/forgot/complete",
+                    json={"email": "writer@example.com", "code": code, "password": "new-orbit-99"})
+    assert r.status_code == 200
+    assert client.get("/api/auth/me", headers=old).status_code == 401
+
+
+def test_google_user_cannot_use_password(client, db_session):
+    from app.services.auth_identity import resolve_or_create_google_user
+
+    resolve_or_create_google_user(db_session, email="g@example.com", google_id="123", name="G", avatar_url=None)
+    db_session.commit()
+    r = client.post("/api/auth/email/register/start", json={"email": "g@example.com", "password": "longenough"})
+    assert r.status_code == 409
+    assert r.json()["detail"]["code"] == "wrong_provider"
+
+
+def test_delete_account(client):
+    data = _signup(client)
+    headers = {"Authorization": f"Bearer {data['access_token']}"}
+    assert client.post("/api/auth/delete-account", headers=headers).status_code == 200
+    r = client.post("/api/auth/email/login", json={"email": "writer@example.com", "password": "stardust-42"})
+    assert r.status_code == 403