AI Data & Database Architecture

# FIX: Offload blob to S3, persist clean pointer in PostgreSQL
import boto3
import hashlib

s3_client = boto3.client('s3')

async def store_user_document(file_bytes: bytes, filename: str, user_id: str, tenant_id: str, db):
    file_hash = hashlib.sha256(file_bytes).hexdigest()
    s3_key = f"{tenant_id}/documents/{file_hash}/{filename}"
    
    # 1. Store raw binary in Object Storage
    s3_client.put_object(
        Bucket="ai-production-documents",
        Key=s3_key,
        Body=file_bytes,
        ContentType="application/pdf"
    )
    
    # 2. Store metadata pointer in PostgreSQL
    db.execute(
        """
        INSERT INTO documents (tenant_id, owner_id, filename, s3_bucket, s3_key, size_bytes, sha256_hash, status)
        VALUES (%s, %s, %s, %s, %s, %s, %s, 'uploaded')
        """,
        (tenant_id, user_id, filename, "ai-production-documents", s3_key, len(file_bytes), file_hash)
    )

# FIX: Strict Tenant-Partitioned Cache Key & Search
def get_cache_key(tenant_id: str, query_text: str) -> str:
    # Hash query text combined with immutable tenant identity
    query_hash = hashlib.sha256(query_text.strip().lower().encode()).hexdigest()
    return f"ai_cache:{tenant_id}:{query_hash}"

async def search_semantic_cache(tenant_id: str, query_vector: list[float], redis_client):
    # Search ONLY within this tenant's vector namespace
    query = (
        Query(f"@tenant_id:{`tenant_id`}=>[KNN 1 @vector $vec AS score]")
        .return_fields("completion_text", "score")
        .dialect(2)
    )
    return redis_client.ft("cache_idx").search(query, query_params={"vec": query_vector})

# FIX: Durable Job State Machine in PostgreSQL
from enum import Enum
import uuid

class JobStatus(str, Enum):
    QUEUED = "queued"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"

def create_ingestion_job(document_id: str, tenant_id: str, db):
    job_id = str(uuid.uuid4())
    db.execute(
        """
        INSERT INTO ai_processing_jobs (id, tenant_id, document_id, status, attempts, created_at)
        VALUES (%s, %s, %s, %s, 0, NOW())
        """,
        (job_id, tenant_id, document_id, JobStatus.QUEUED)
    )
    return job_id

def claim_next_job(worker_id: str, db):
    # Atomic row-lock prevents race conditions among concurrent workers
    return db.query_one(
        """
        UPDATE ai_processing_jobs
        SET status = 'processing', worker_id = %s, updated_at = NOW()
        WHERE id = (
            SELECT id FROM ai_processing_jobs
            WHERE status = 'queued'
            ORDER BY created_at ASC
            LIMIT 1
            FOR UPDATE SKIP LOCKED
        )
        RETURNING *
        """,
        (worker_id,)
    )

# FIX: Monotonic Sequence Numbering in SQL Schema
"""
CREATE TABLE conversation_messages (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    conversation_id UUID NOT NULL REFERENCES conversations(id) ON DELETE CASCADE,
    sequence_number INT NOT NULL,
    role VARCHAR(20) NOT NULL CHECK (role IN ('system', 'user', 'assistant', 'tool')),
    content TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    CONSTRAINT uq_conv_seq UNIQUE (conversation_id, sequence_number)
);

CREATE INDEX idx_messages_conv_seq ON conversation_messages(conversation_id, sequence_number ASC);
"""

def append_message(conversation_id: str, role: str, content: str, db):
    # Atomically calculate next sequence number
    db.execute(
        """
        INSERT INTO conversation_messages (conversation_id, sequence_number, role, content)
        VALUES (
            %s,
            COALESCE((SELECT MAX(sequence_number) + 1 FROM conversation_messages WHERE conversation_id = %s), 1),
            %s,
            %s
        )
        """,
        (conversation_id, conversation_id, role, content)
    )

# FIX: Coordinated Cascading Deletion across Relational & Vector Stores
async def delete_document_cascade(document_id: str, tenant_id: str, db, s3, vector_index):
    # 1. Fetch document metadata to ensure ownership
    doc = db.query_one("SELECT * FROM documents WHERE id = %s AND tenant_id = %s", (document_id, tenant_id))
    if not doc:
        raise ResourceNotFoundException("Document not found")
        
    # 2. Delete all vector embeddings matching document_id metadata
    vector_index.delete(
        filter={
            "document_id": {"$eq": document_id},
            "tenant_id": {"$eq": tenant_id}
        }
    )
    
    # 3. Delete raw file from Object Storage
    s3.delete_object(Bucket=doc["s3_bucket"], Key=doc["s3_key"])
    
    # 4. Invalidate related caches
    invalidate_document_cache(tenant_id, document_id)
    
    # 5. Delete metadata row in PostgreSQL
    db.execute("DELETE FROM documents WHERE id = %s", (document_id,))

# FIX: Distributed Mutex on Cache Miss (Prevent Stampede)
import redis
import time

redis_client = redis.Redis()

async def get_or_compute_faq(query_text: str, tenant_id: str):
    cache_key = f"faq:{tenant_id}:{hashlib.sha256(query_text.encode()).hexdigest()}"
    cached_val = redis_client.get(cache_key)
    if cached_val:
        return cached_val.decode('utf-8')
        
    lock_key = f"lock:{cache_key}"
    # Acquire distributed lock: only 1 request recomputes
    acquired = redis_client.set(lock_key, "locked", nx=True, ex=10)
    if acquired:
        try:
            # Generate answer from LLM
            answer = await call_llm_faq(query_text)
            # Store with jittered TTL (e.g. 3600s + random 300s)
            redis_client.set(cache_key, answer, ex=3600 + int(random.random() * 300))
            return answer
        finally:
            redis_client.delete(lock_key)
    else:
        # Another worker is already generating; sleep briefly and re-read cache
        time.sleep(0.5)
        return redis_client.get(cache_key) or await call_llm_faq(query_text)

# FIX: Dedicated AI Usage & Cost Telemetry Table
"""
CREATE TABLE ai_usage_logs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    request_id UUID NOT NULL,
    tenant_id VARCHAR(64) NOT NULL,
    user_id VARCHAR(64) NOT NULL,
    model_name VARCHAR(64) NOT NULL,
    prompt_tokens INT NOT NULL,
    completion_tokens INT NOT NULL,
    total_tokens INT NOT NULL,
    simulated_cost_usd NUMERIC(8, 6) NOT NULL,
    latency_ms INT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_usage_tenant_month ON ai_usage_logs(tenant_id, created_at);
"""

def log_ai_usage(req_id: str, tenant_id: str, user_id: str, model: str, usage: dict, latency_ms: int, db):
    p_tokens = usage.get("prompt_tokens", 0)
    c_tokens = usage.get("completion_tokens", 0)
    # Estimated blended price per 1k tokens for synthetic metering
    cost = (p_tokens * 0.000003) + (c_tokens * 0.000015)
    db.execute(
        """
        INSERT INTO ai_usage_logs (request_id, tenant_id, user_id, model_name, prompt_tokens, completion_tokens, total_tokens, simulated_cost_usd, latency_ms)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
        """,
        (req_id, tenant_id, user_id, model, p_tokens, c_tokens, p_tokens + c_tokens, cost, latency_ms)
    )

# FIX: Asynchronous Ingestion with HTTP 202 Accepted
@app.post("/api/v1/documents/upload", status_code=status.HTTP_202_ACCEPTED)
async def upload_document(
    file: UploadFile,
    auth_ctx: AuthContext = Depends(get_current_user),
    db = Depends(get_db)
):
    # 1. Quick validation and S3 upload
    content = await file.read()
    doc_id = save_document_metadata_and_s3(file.filename, content, auth_ctx, db)
    
    # 2. Queue background ingestion task (FastAPI BackgroundTasks or Celery)
    job_id = enqueue_vector_ingestion_job(doc_id, auth_ctx.tenant_id)
    
    # 3. Return immediately to client within 40ms!
    return {
        "status": "accepted",
        "document_id": doc_id,
        "job_id": job_id,
        "poll_url": f"/api/v1/jobs/{job_id}"
    }