diff --git a/.env.example b/.env.example new file mode 100644 index 0000000000000000000000000000000000000000..bded2b41f7d356c781bbcf2133a64af5f0af1c00 --- /dev/null +++ b/.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 +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= diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000000000000000000000000000000000000..f853f171a004d9f05bc6236838701a29181d5b80 --- /dev/null +++ b/.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 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000000000000000000000000000000000000..d5580e4f6ee2a344364a94f95a1a739f4e2d25f5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +.env +node_modules/ +dist/ +.venv/ +__pycache__/ +*.pyc +.pytest_cache/ +.ruff_cache/ +out/ +renderer/build/ diff --git a/README.md b/README.md new file mode 100644 index 0000000000000000000000000000000000000000..4c910047de9c1405ffdda8aa7c855c53dca69926 --- /dev/null +++ b/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. diff --git a/backend/Dockerfile b/backend/Dockerfile new file mode 100644 index 0000000000000000000000000000000000000000..b4abb98fb3268cd9df96932f4b5dc3bee200306f --- /dev/null +++ b/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"] diff --git a/backend/alembic.ini b/backend/alembic.ini new file mode 100644 index 0000000000000000000000000000000000000000..d4e4fbb265cefcf24e462f156098169bb789d4ab --- /dev/null +++ b/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 diff --git a/backend/alembic/env.py b/backend/alembic/env.py new file mode 100644 index 0000000000000000000000000000000000000000..91c9973b74237303de9f3ed7b265fc56953234cf --- /dev/null +++ b/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() diff --git a/backend/alembic/script.py.mako b/backend/alembic/script.py.mako new file mode 100644 index 0000000000000000000000000000000000000000..36cd8efeda0d3f671a90fc6c7b4131af5758bbb4 --- /dev/null +++ b/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"} diff --git a/backend/alembic/versions/0001_initial.py b/backend/alembic/versions/0001_initial.py new file mode 100644 index 0000000000000000000000000000000000000000..5959255ca9a36561f1b5bfbebe7559499b46cbaa --- /dev/null +++ b/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()) diff --git a/backend/app/__init__.py b/backend/app/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/backend/app/auth.py b/backend/app/auth.py new file mode 100644 index 0000000000000000000000000000000000000000..426a9d41a582b67ee24d6508b4469187ca930c11 --- /dev/null +++ b/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) diff --git a/backend/app/config.py b/backend/app/config.py new file mode 100644 index 0000000000000000000000000000000000000000..1b6b810d1fbbd7ea77dca1c8d189237f0ead6a1e --- /dev/null +++ b/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 " + 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() diff --git a/backend/app/db.py b/backend/app/db.py new file mode 100644 index 0000000000000000000000000000000000000000..9806460568ad635b72853c9ddc20622326554d6b --- /dev/null +++ b/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() diff --git a/backend/app/llm/__init__.py b/backend/app/llm/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/backend/app/llm/embeddings.py b/backend/app/llm/embeddings.py new file mode 100644 index 0000000000000000000000000000000000000000..dd2f0321c4bf1ffc5e4ba6c5c1befffc93fd19e6 --- /dev/null +++ b/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] diff --git a/backend/app/llm/postprocess.py b/backend/app/llm/postprocess.py new file mode 100644 index 0000000000000000000000000000000000000000..266f73cab456518208fe2691659dfca43471e307 --- /dev/null +++ b/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 diff --git a/backend/app/llm/provider.py b/backend/app/llm/provider.py new file mode 100644 index 0000000000000000000000000000000000000000..ca80349533834091d88405380282a96de5f5a00f --- /dev/null +++ b/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] diff --git a/backend/app/llm/signatures.py b/backend/app/llm/signatures.py new file mode 100644 index 0000000000000000000000000000000000000000..7c76421bbb7d73b1bafb2f814daacc079de5cb87 --- /dev/null +++ b/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() diff --git a/backend/app/main.py b/backend/app/main.py new file mode 100644 index 0000000000000000000000000000000000000000..3f92af736c58fc6198d95258e1fee113c4837490 --- /dev/null +++ b/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) diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e3a7f7f8fbe154f893825e8fac8f45faf8fa96fc --- /dev/null +++ b/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", +] diff --git a/backend/app/models/base.py b/backend/app/models/base.py new file mode 100644 index 0000000000000000000000000000000000000000..3da5af57bb071ef79cd612f54d7830f494a6489d --- /dev/null +++ b/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) diff --git a/backend/app/models/content.py b/backend/app/models/content.py new file mode 100644 index 0000000000000000000000000000000000000000..5b093bcae5c5a5287a981d40d14643a6e024b03a --- /dev/null +++ b/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) diff --git a/backend/app/models/email.py b/backend/app/models/email.py new file mode 100644 index 0000000000000000000000000000000000000000..beafa08516875e822cacb7d72cd6d6d3d58a3067 --- /dev/null +++ b/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 diff --git a/backend/app/models/user.py b/backend/app/models/user.py new file mode 100644 index 0000000000000000000000000000000000000000..20c2042e0358d3972e4f14d6df488131e8f0d8ed --- /dev/null +++ b/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)) diff --git a/backend/app/pipeline/__init__.py b/backend/app/pipeline/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/backend/app/pipeline/ingest.py b/backend/app/pipeline/ingest.py new file mode 100644 index 0000000000000000000000000000000000000000..9179dbd477824c3138c15c6c23b1d41192f548b1 --- /dev/null +++ b/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)} diff --git a/backend/app/pipeline/retrieval.py b/backend/app/pipeline/retrieval.py new file mode 100644 index 0000000000000000000000000000000000000000..83dd6df566d8cab0b0795ff3b1fd6507d2a099f2 --- /dev/null +++ b/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, + ) diff --git a/backend/app/routers/__init__.py b/backend/app/routers/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/backend/app/routers/artifacts.py b/backend/app/routers/artifacts.py new file mode 100644 index 0000000000000000000000000000000000000000..e2072e54cb9a0f0f8d99bf84f27c29b90f4b01aa --- /dev/null +++ b/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) diff --git a/backend/app/routers/auth.py b/backend/app/routers/auth.py new file mode 100644 index 0000000000000000000000000000000000000000..8c52526608804c87c5bdecfa46c2edb0ec5243c5 --- /dev/null +++ b/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} diff --git a/backend/app/routers/billing.py b/backend/app/routers/billing.py new file mode 100644 index 0000000000000000000000000000000000000000..a50518df597d0fea1527c6248d7901f3fb301b19 --- /dev/null +++ b/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") diff --git a/backend/app/routers/jobs.py b/backend/app/routers/jobs.py new file mode 100644 index 0000000000000000000000000000000000000000..d29ec59acdc903ac5b5373000bb4b68904023f76 --- /dev/null +++ b/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} diff --git a/backend/app/routers/notebooks.py b/backend/app/routers/notebooks.py new file mode 100644 index 0000000000000000000000000000000000000000..88591145f90b9afa29bb3379b3dbf110bce00b1c --- /dev/null +++ b/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") diff --git a/backend/app/routers/sources.py b/backend/app/routers/sources.py new file mode 100644 index 0000000000000000000000000000000000000000..0856c57d4408d31340edde87f04452a074d833b2 --- /dev/null +++ b/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 + ] diff --git a/backend/app/routers/storage.py b/backend/app/routers/storage.py new file mode 100644 index 0000000000000000000000000000000000000000..ae399f209e0c00f8c3e5ed358fc14cdb2f974c23 --- /dev/null +++ b/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) diff --git a/backend/app/routers/unsubscribe.py b/backend/app/routers/unsubscribe.py new file mode 100644 index 0000000000000000000000000000000000000000..19b8b79ae1222ba342f641f1587b7d2f86cba075 --- /dev/null +++ b/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 = """
+

Notestack

+

{title}

{body}

""" + + +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)} diff --git a/backend/app/services/__init__.py b/backend/app/services/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/backend/app/services/auth_identity.py b/backend/app/services/auth_identity.py new file mode 100644 index 0000000000000000000000000000000000000000..d3434601f30c30bf6c72555e0176623379ec8105 --- /dev/null +++ b/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 diff --git a/backend/app/services/email.py b/backend/app/services/email.py new file mode 100644 index 0000000000000000000000000000000000000000..14198dd6d228805b7f2c2070def0bfe99a914d88 --- /dev/null +++ b/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