[Yeah] Architecture & Tech Stack Reference

Heading

֍ Mental Model

• Runtime Path: Ingestion → Consensus Engine → Personalization → Verdict API → Ghost Runtime. Every component in the architecture maps exactly to a slot in this path or the operational surface surrounding it.

• Forcing Function ("10,000 users in 60 days"): stack is deliberately boring, proven, and mostly in-process to minimize moving parts and operational surface area. The goal is to ship features rather than wrangle infra. Punt: queue, cache, or AI orchestration layer when needed.

---

Language, API & Validation

Python 3.12+ version floor

• Cleaner generics, type aliases
• Core Mechanics: async + typing with Pydantic's; has mature asyncio performance & modern typing ergonomics

# The frame to hold: Modern async + typing contract boundary
import asyncio
from typing import TypeAlias, Generic, TypeVar

T = TypeVar('T')
PayloadID: TypeAlias = str

class RuntimeSlot(Generic[T]):
    def __init__(self, slot_id: PayloadID, data: T):
        self.slot_id = slot_id
        self.data = data

async def execute_runtime_path() -> None:
    # 3.12+ optimized asyncio event loop execution path
    pass

FastAPI HTTP Layer

• ASGI/async-native, pairing natively with async SQLAlchemy so the whole request path stays non-blocking.

• Core Mechanics: Owns route definitions, request parsing, and Dependency Injection (DI). Auto-generates the OpenAPI schema directly from Pydantic models.

from fastapi import FastAPI, Depends, Security
from pydantic import BaseModel

app = FastAPI(title="Bakery", version="1.0.0")

class VerdictResponse(BaseModel):
    product_id: str
    verdict: str
    confidence: float

# Exposing /v1/verdicts/{product_id} and /healthz
@app.get("/v1/verdicts/{product_id}", response_model=VerdictResponse)
async def get_verdict(product_id: str, principal: dict = Depends(validate_supabase_jwt)):
    # Non-blocking async path with embedded DI validation boundary
    return {"product_id": product_id, "verdict": "consensus_approved", "confidence": 0.98}

@app.get("/healthz")
async def health_check():
    return {"status": "healthy"}

Pydantic v2 (Strict Mode) for Contract Boundary

• Rust core makes thousands of per-mention validations cheap.

• Core Mechanics: At the API edge, it defines request/response models. At the runtime edge, every extraction/planner output crossing a boundary is a BaseModel with ConfigDict(strict=True). No silent coercion ("3" stays a string, never becomes 3), no raw dicts, no untyped downstream data.

from pydantic import BaseModel, ConfigDict, Field

class ExtractedMention(BaseModel):
    # Enforce strict parsing at runtime edge to stop malformed model outputs
    model_config = ConfigDict(strict=True)
    
    mention: str
    canonical_product_id: str
    sentiment: str = Field(description="Extracted sentiment analysis")
    is_primary_subject: bool
    is_sponsored: bool

---

Data Persistence & Migrations

PostgreSQL — The Durable System of Record

• Official Reference: Postgres Documentation

• Core Mechanics: Holds relational data: products, aliases, sources, mentions, consensus snapshots, and the append-only event log. Relational because the truth model is relational. Enforces the replayability invariant: recompute verdicts from durable rows instead of mutating state in place.

-- Schema representation of relational truth and append-only invariants
CREATE TABLE products (
    id UUID PRIMARY KEY,
    canonical_name VARCHAR(255) NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE mentions (
    id UUID PRIMARY KEY,
    canonical_product_id UUID REFERENCES products(id),
    source_attribution TEXT NOT NULL,
    payload JSONB NOT NULL, -- Strict shape validated by Pydantic on write
    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

async SQLAlchemy 2.0 & Alembic — Data Access & Schema Evolution

• Official References: SQLAlchemy 2.0 Docs | Alembic Docs

• Core Mechanics: Two distinct, decoupled jobs. SQLAlchemy maps objects to SQL using 2.0 style exclusively (AsyncSession, no legacy .query()). The async part is load-bearing; sync drivers block the event loop. Alembic manages versioned, reversible DDL upgrades/downgrades without raw ALTER TABLE operations.

# SQLAlchemy 2.0 modern async data-access layer pattern
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.future import select

class Base(DeclarativeBase):
    pass

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/bakery")

async def fetch_consensus_snapshot(session: AsyncSession, product_id: str):
    # Load-bearing async execution pattern protecting concurrency
    stmt = select(Base.metadata.tables['products']).where(Base.metadata.tables['products'].c.id == product_id)
    result = await session.execute(stmt)
    return result.fetchone()

---

AI Inference & Authentication

Anthropic via Direct HTTP — The Generative Funnel

• Official Reference: Anthropic Messages API

• Core Mechanics: The generative half of the deterministic-to-generative funnel. After deterministic alias matching narrows a transcript to relevant slices, Claude extracts structured data. Made via direct httpx raw HTTP calls parsed into strict Pydantic models. No SDK wrapper, explicitly no LangChain—keeping the prompt/parse boundary completely visible.

import httpx

async def extract_structured_mentions(transcript_slice: str) -> ExtractedMention:
    # Direct HTTP implementation using raw Messages API and httpx
    url = "https://api.anthropic.com/v1/messages"
    headers = {"x-api-key": "your_key", "anthropic-version": "2023-06-01"}
    payload = {
        "model": "claude-3-5-sonnet-20240620",
        "messages": [{"role": "user", "content": f"Extract structured metrics from: {transcript_slice}"}]
    }
    
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload, headers=headers)
        raw_data = response.json()
        # Parse straight into Pydantic strict contract boundary
        return ExtractedMention.model_validate(raw_data["content"][0]["text"])

Supabase JWT Verification — The Auth Boundary

• Official Reference: Supabase Auth | JWT.io

• Core Mechanics: An auth boundary, not an auth system. Bakery does not host logins, sessions, or user profiles. It receives a JWT minted by Supabase (where Ghost/Atomic owns real auth), verifies signature/claims as a FastAPI dependency, and extracts identity. Read and verify only—preserving repo isolation.

from jose import jwt

def validate_supabase_jwt(token: str) -> dict:
    # Only read and verify signature + claims; no backend DB calls to Supabase
    SUPABASE_JWT_SECRET = "your-isolated-repo-secret"
    ALGORITHMS = ["HS256"]
    
    try:
        payload = jwt.decode(token, SUPABASE_JWT_SECRET, algorithms=ALGORITHMS)
        return payload  # Returns principal identity context
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="Invalid token boundary constraint")

---

Infrastructure & Asynchronous Work

Railway — Hosting Platform (PaaS)

• Official Reference: Railway Documentation

• Core Mechanics: Runs the FastAPI container and managed Postgres instance with git-push deploys and managed env vars. Chosen over AWS/k8s to maintain zero infra ops at 1,000 users. No clusters, no Terraform, no brokers to babysit.

# Conceptual Railway Deployment Environment Profile
services:
  bakery-api:
    build:
      dockerfile: Dockerfile
    environment:
      PYTHONPATH: .
      DATABASE_URL: ${PostgreSQL.DATABASE_URL}
  database:
    image: postgres:16

FastAPI BackgroundTasks & Cron — Minimal Work Layer

• Official Reference: FastAPI Background Tasks Docs

• Core Mechanics: Kept deliberately minimal as an explicit substitute for a heavy task queue. BackgroundTasks executes fire-and-forget in-process work directly after a response is dispatched (e.g. transcript ingestion initialization). Cron handles out-of-band scheduled execution (nightly backfills, consensus recalculations).

from fastapi import BackgroundTasks

async def heavy_transcript_ingestion(raw_data: str):
    # In-process asynchronous task execution execution path
    pass

@app.post("/v1/ingest")
async def trigger_ingestion(data: str, background_tasks: BackgroundTasks):
    # Fire-and-forget directly after responding to caller
    background_tasks.add_task(heavy_transcript_ingestion, data)
    return {"status": "ingestion_queued"}

---

The "Deferred" Stack (Intentional Exclusions)

• No Celery & Redis Broker: Multi-worker fan-out and out-of-process queues introduce overhead only justified by high job volume. BackgroundTasks + cron easily covers 10,000 users.

• No Redis-as-Cache: Premature optimization; Postgres safely handles the read load at this scale.

• No LangChain: Abstractions over LLM calls hide the underlying HTTP request and create dependency churn, fighting the discipline of structured output from controlled prompts. Direct HTTP provides absolute control.