feat(hubs): idempotent create with inline document (#6)

* feat(models): add relationships and hub response schemas

* feat(hubs): add tenant-safe list, get, and delete endpoints

* test(isolation): prove cross-tenant read, list, and delete are cloaked

* feat(ai): add gemini, embeddings, and chunking wrappers

* feat(hubs): implement idempotent create with inline document

* test(idempotency): cover replay, different body, and missing key

app/chunking.py

@@ -0,0 +1,9 @@
+def chunk_text(text: str, size_tokens: int = 500, overlap_tokens: int = 50) -> list[tuple[str, int]]:
+    words = text.split()
+    chunks = []
+    i = 0
+    while i < len(words):
+        chunk_words = words[i:i + size_tokens]
+        chunks.append((" ".join(chunk_words), len(chunk_words)))
+        i += size_tokens - overlap_tokens
+    return chunks

app/idempotency.py

@@ -0,0 +1,76 @@
+import hashlib
+import json
+from uuid import UUID
+from datetime import datetime, timezone, timedelta
+from sqlalchemy.orm import Session
+from sqlalchemy.dialects.postgresql import insert
+from fastapi import Header, Response
+from app.models import IdempotencyKey
+from app.errors import AppError
+
+def require_idempotency_key(x_idempotency_key: str | None = Header(default=None)) -> UUID:
+    if x_idempotency_key is None:
+        raise AppError(400, "IDEMPOTENCY_KEY_MISSING", "Header X-Idempotency-Key is required.")
+    try:
+        key = UUID(x_idempotency_key)
+        if key.version != 4:
+            raise ValueError
+        return key
+    except ValueError:
+        raise AppError(422, "IDEMPOTENCY_KEY_INVALID", "X-Idempotency-Key must be a UUIDv4.")
+
+def body_hash(data: dict) -> str:
+    canonical = json.dumps(data, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(canonical.encode()).hexdigest()
+
+def purge_stale(db: Session, user_id: UUID):
+    now = datetime.now(timezone.utc)
+    # Delete expired keys OR in_progress keys older than 60 seconds
+    db.query(IdempotencyKey).filter(
+        IdempotencyKey.user_id == user_id,
+        (IdempotencyKey.expires_at < now) | 
+        ((IdempotencyKey.status == "in_progress") & (IdempotencyKey.created_at < now - timedelta(seconds=60)))
+    ).delete()
+    db.commit()
+
+def claim_key(db: Session, user_id: UUID, key: UUID, req_hash: str, path: str) -> bool:
+    stmt = insert(IdempotencyKey).values(
+        user_id=user_id, key=key, request_method="POST", request_path=path,
+        request_hash=req_hash, status="in_progress",
+        expires_at=datetime.now(timezone.utc) + timedelta(hours=24)
+    ).on_conflict_do_nothing().returning(IdempotencyKey.key)
+    
+    result = db.execute(stmt)
+    db.commit()
+    # fetchone() returns None if the insert was skipped (conflict)
+    return result.fetchone() is not None
+
+def complete_key(db: Session, user_id: UUID, key: UUID, status: int, body: dict):
+    row = db.query(IdempotencyKey).filter_by(user_id=user_id, key=key).first()
+    if row:
+        row.status = "completed"
+        row.response_status = status
+        row.response_body = body
+        db.commit()
+
+def release_claim(db: Session, user_id: UUID, key: UUID):
+    db.query(IdempotencyKey).filter_by(user_id=user_id, key=key).delete()
+    db.commit()
+
+def replay_or_conflict(db: Session, user_id: UUID, key: UUID, req_hash: str):
+    row = db.query(IdempotencyKey).filter_by(user_id=user_id, key=key).first()
+    if not row:
+        raise AppError(500, "INTERNAL_ERROR", "Idempotency key vanished.")
+    
+    if row.request_hash != req_hash:
+        raise AppError(422, "IDEMPOTENCY_KEY_REUSED", "This idempotency key was used with a different payload.")
+    
+    if row.status == "in_progress":
+        raise AppError(409, "IDEMPOTENCY_REQUEST_IN_PROGRESS", "Original request still processing.", headers={"Retry-After": "2"})
+    
+    return Response(
+        content=json.dumps(row.response_body),
+        status_code=row.response_status,
+        media_type="application/json",
+        headers={"Idempotent-Replayed": "true"}
+    )

app/routers/hubs.py

@@ -3,6 +3,81 @@ from fastapi import APIRouter, Depends, status, Query, Response
 from sqlalchemy.orm import Session
 from app.db import get_db
 from app.deps import get_current_user
+from app.models import User, Hub, Document, Chunk
+from app.schemas import HubResponse, HubDetail, HubCreate
+from app.errors import AppError
+from app import repo
+from app.idempotency import require_idempotency_key, body_hash, purge_stale, claim_key, complete_key, release_claim, replay_or_conflict
+from app.chunking import chunk_text
+from app.services.embeddings import embed_texts
+from app.services.gemini import hub_insights
+router = APIRouter(prefix="/hubs", tags=["hubs"])
+
+
+
+@router.post("", response_model=HubResponse, status_code=status.HTTP_201_CREATED)
+def create_hub(
+    body: HubCreate,
+    key: UUID = Depends(require_idempotency_key),
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db)
+):
+    user_id = current_user.id
+    req_hash = body_hash(body.model_dump())
+    
+    purge_stale(db, user_id)
+    if not claim_key(db, user_id, key, req_hash, "/api/v1/hubs"):
+        return replay_or_conflict(db, user_id, key, req_hash)
+
+    # Name pre-check
+    if db.query(Hub).filter(Hub.owner_id == user_id, Hub.name == body.name).first():
+        release_claim(db, user_id, key)
+        raise AppError(409, "HUB_NAME_CONFLICT", "You already have a hub with this name.")
+
+    # AI Summary
+    summary, questions = hub_insights(body.content) if body.content else (None, [])
+
+    # Create Hub
+    new_hub = Hub(owner_id=user_id, name=body.name, description=body.description, ai_summary=summary, suggested_questions=questions)
+    db.add(new_hub)
+    db.flush()
+
+    doc_count = 0
+    chunk_count = 0
+
+    # Inline Document
+    if body.content.strip():
+        doc_count = 1
+        new_doc = Document(
+            hub_id=new_hub.id, owner_id=user_id, origin="inline", filename="inline.txt",
+            mime_type="text/plain", size_bytes=len(body.content), checksum_sha256="0"*64, status="pending"
+        )
+        db.add(new_doc)
+        db.flush()
+        try:
+            chunks_data = chunk_text(body.content)
+            texts = [c[0] for c in chunks_data]
+            embeddings = embed_texts(texts)
+            for i, ((text, tokens), emb) in enumerate(zip(chunks_data, embeddings)):
+                db.add(Chunk(document_id=new_doc.id, hub_id=new_hub.id, owner_id=user_id, chunk_index=i, text=text, token_count=tokens, embedding=emb))
+                chunk_count += 1
+            new_doc.status = "ready"
+        except Exception:
+            new_doc.status = "failed"
+            new_doc.error_message = "Embedding failed"
+
+    db.commit()
+    db.refresh(new_hub)
+
+    # Build response
+    hub_resp = HubResponse.model_validate(new_hub)
+    hub_resp.document_count = doc_count
+    hub_resp.chunk_count = chunk_count
+    payload = hub_resp.model_dump(mode="json")
+
+    complete_key(db, user_id, key, 201, payload)
+    return payload
+
 from app.models import User
 from app.schemas import HubResponse, HubDetail
 from app.errors import AppError

app/schemas.py

@@ -68,6 +68,13 @@ class HubResponse(BaseModel):
     updated_at: datetime
 
 class HubDetail(HubResponse):
+    documents: list[DocumentSummary] = []  
+
+class HubCreate(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    name: str = Field(min_length=1, max_length=120)
+    description: str | None = Field(default=None, max_length=500)
+    content: str = Field(default="", max_length=20000)      
     documents: list[DocumentSummary] = []    
 
     

app/services/embeddings.py

@@ -0,0 +1,17 @@
+from google import genai
+from google.genai import types
+from app.config import settings
+from app.services.gemini import AiUnavailable
+
+_client = genai.Client(api_key=settings.GEMINI_API_KEY, http_options=types.HttpOptions(timeout=15_000))
+
+def embed_texts(texts: list[str], batch_size: int = 20) -> list[list[float]]:
+    out: list[list[float]] = []
+    for i in range(0, len(texts), batch_size):
+        batch = texts[i:i + batch_size]
+        try:
+            r = _client.models.embed_content(model=settings.GEMINI_EMBEDDING_MODEL, contents=batch)
+            out.extend([e.values for e in r.embeddings])
+        except Exception as exc:
+            raise AiUnavailable("provider", str(exc)) from exc
+    return out

app/services/gemini.py

@@ -0,0 +1,26 @@
+import json
+from google import genai
+from google.genai import types
+from app.config import settings
+
+class AiUnavailable(Exception):
+    def __init__(self, kind: str, detail: str = ""):
+        super().__init__(detail)
+        self.kind = kind
+
+_client = genai.Client(api_key=settings.GEMINI_API_KEY, http_options=types.HttpOptions(timeout=15_000))
+
+def hub_insights(content: str) -> tuple[str | None, list[str]]:
+    try:
+        r = _client.models.generate_content(
+            model=settings.GEMINI_MODEL, 
+            contents=f"<context>{content}</context>",
+            config=types.GenerateContentConfig(
+                system_instruction='Reply with JSON only: {"summary": "at most 2 sentences", "questions": ["3 questions"]}',
+                temperature=0.2, max_output_tokens=1024, response_mime_type="application/json"
+            )
+        )
+        d = json.loads(r.text)
+        return str(d.get("summary", ""))[:500], [str(q)[:200] for q in d.get("questions", [])][:3]
+    except Exception:
+        return None, []

tests/conftest.py

@@ -35,9 +35,9 @@ def setup_database():
 
 @pytest.fixture(autouse=True)
 def clean_tables():
-    # Truncate tables between tests
-    with engine.begin() as conn:
-        conn.execute(text("TRUNCATE idempotency_keys, chunks, documents, hubs, users CASCADE"))
+    with engine.connect() as conn:
+        conn.execute(text("TRUNCATE idempotency_keys, chunks, documents, hubs, users RESTART IDENTITY CASCADE"))
+        conn.commit()
     yield
 
 @pytest.fixture

tests/test_idempotency.py

@@ -0,0 +1,64 @@
+import pytest
+from uuid import uuid4
+from unittest.mock import patch
+from app.models import Hub, Document, Chunk
+
+def test_create_hub_idempotency(client, db_session):
+    client.post("/api/v1/auth/signup", json={"email": "idem@example.com", "password": "Password123"})
+    login = client.post("/api/v1/auth/login", json={"email": "idem@example.com", "password": "Password123"})
+    token = login.json()["access_token"]
+    
+    key = str(uuid4())
+    headers = {"Authorization": f"Bearer {token}", "X-Idempotency-Key": key}
+    body = {"name": "HR Hub", "content": "This is a long text about HR policies and employee benefits."}
+
+    with patch("app.routers.hubs.hub_insights", return_value=("Summary here", ["Q1?", "Q2?"])), \
+         patch("app.routers.hubs.embed_texts", return_value=[[0.1]*768] * 3):
+        
+        # First Request
+        r1 = client.post("/api/v1/hubs", headers=headers, json=body)
+        if r1.status_code != 201:
+            print("\n\nFAILED FIRST REQUEST:", r1.json(), "\n\n")
+            
+        assert r1.status_code == 201
+        assert "Idempotent-Replayed" not in r1.headers
+
+        # Second Request (Replay)
+        r2 = client.post("/api/v1/hubs", headers=headers, json=body)
+        
+        # --- ADD THIS PRINT STATEMENT ---
+        if r2.status_code != 201:
+            print("\n\nFAILED SECOND REQUEST:", r2.json(), "\n\n")
+        # --------------------------------
+
+        assert r2.status_code == 201
+        assert r2.headers["Idempotent-Replayed"] == "true"
+        assert r1.json() == r2.json()
+
+        # Check DB counts
+        assert db_session.query(Hub).count() == 1
+        assert db_session.query(Document).count() == 1
+        assert db_session.query(Chunk).count() > 0
+
+        # Third Request (Different Body)
+        r3 = client.post("/api/v1/hubs", headers=headers, json={"name": "Finance Hub", "content": "Different text"})
+        assert r3.status_code == 422
+        assert r3.json()["error"]["code"] == "IDEMPOTENCY_KEY_REUSED"
+
+def test_create_hub_missing_key(client):
+    client.post("/api/v1/auth/signup", json={"email": "idem2@example.com", "password": "Password123"})
+    login = client.post("/api/v1/auth/login", json={"email": "idem2@example.com", "password": "Password123"})
+    token = login.json()["access_token"]
+    
+    r = client.post("/api/v1/hubs", headers={"Authorization": f"Bearer {token}"}, json={"name": "No Key"})
+    assert r.status_code == 400
+    assert r.json()["error"]["code"] == "IDEMPOTENCY_KEY_MISSING"
+
+def test_create_hub_missing_key(client):
+    client.post("/api/v1/auth/signup", json={"email": "idem2@example.com", "password": "Password123"})
+    login = client.post("/api/v1/auth/login", json={"email": "idem2@example.com", "password": "Password123"})
+    token = login.json()["access_token"]
+    
+    r = client.post("/api/v1/hubs", headers={"Authorization": f"Bearer {token}"}, json={"name": "No Key"})
+    assert r.status_code == 400
+    assert r.json()["error"]["code"] == "IDEMPOTENCY_KEY_MISSING"