airweave / rules

---
globs: **/backend/**
alwaysApply: false
---
# Airweave Backend Architecture

## Core Layers

### API Layer
- **FastAPI Application** (`main.py`): Entry point for HTTP requests
- **API Routes** (`endpoints/`): RESTful endpoints organized by resource
- **Dependencies** (`api/deps.py`): Authentication and request validation

### Service Layer
- **Sync Service**: Orchestrates data synchronization between sources and destinations
- **DAG Service**: Manages directed acyclic graphs for data transformations
- **OAuth2 Service**: Handles authentication for integrations
- **Identity Provider** (protocol): Abstracts user authentication, roles, and organization memberships (Auth0, Null, Fake adapters)
- **Payment Gateway** (protocol): Abstracts subscription management and billing operations (Stripe adapter)
- **Resend Service**: Manages transactional email delivery and templates
- **PostHog Service**: Manages product analytics and user event tracking

### Data Access Layer
- **CRUD Modules** (`crud/`): Database operations for each entity
- **Unit of Work**: Transaction management with atomic operations
- **Database Session**: Async SQLAlchemy session management. This extra care must be applied with lazy objects and sessions.

### Domain Layer
- **Models** (`models/`): Database models (SQLAlchemy ORM)
- **Schemas** (`schemas/`): API data validation models (Pydantic)

Detail: `short_name` is a globally unique identifier used throughout the system, e.g. `slack`, `outlook` or `hubspot_crm`

## Key Components

### Platform Components
- **Sources** (`platform/sources/`): Connectors to external data sources (APIs, DBs)
- **Destinations** (`platform/destinations/`): Vector database adapters
- **Embedding Models** (`platform/embedding_models/`): Text vectorization services
- **Transformers** (`platform/transformers/`): Data transformation processors
- **Entities** (`platform/entities/`): Core search objects that flow through the system

### Sync Architecture
If necessary (like editing ), refer [sync-architecture.mdc](mdc:.cursor/rules/sync-architecture.mdc).

### Infrastructure
- **Config** (`core/config.py`): Environment-based configuration
- **Logging** (`core/logging.py`): Structured logging system
- **Exceptions** (`core/exceptions.py`): Centralized error handling
- **Migrations** (`alembic/`): Database schema versioning

## Data Flow

1. **Request Processing**: API request → Authentication → Schema Validation → Handler (often CRUD, otherwise service)
2. **Sync Flow**: Source → Transformation (DAG) → Embedding → Destination
3. **Database Operations**: CRUD with Unit of Work pattern for transactions


## Rules

### Style & Structure
- Python 3.11+, FastAPI, SQLAlchemy async ORM
- Black (100 char), isort, Ruff with Google docstrings including arg and return type annotation
- Models → `models/`, Pydantic schemas → `schemas/`
- Domain modules → `domains/` (protocols, service, repository, operations, fakes, subscribers)
- Platform connectors → `platform/` modules by function (sources, destinations, transformers, entities)
- Sync handlers → `platform/sync/`

### Code Principles
- Descriptive names over comments
- Async for all I/O operations
- Typed parameters and returns
- Keep functions under 50 lines

### Architecture
- SQLAlchemy models with UUID primary keys
- FastAPI dependency injection for services
- Error responses via custom exceptions
- Background processing with Redis workers (upcoming)

### API Convention
- RESTful endpoints in `endpoints/` -> the version is not part of the endpoint. It's just host.com/{endpoint}!
- Consistent response structures
- One router per resource type
- Logger injected via `ctx` dependency for contextual logging


### Logging convention
- Stick to log level standards
- Always try to use logger from ctx (in case of API) or sync_context (if during sync)

### PostHog Analytics Integration
- **Module Path**: `airweave.analytics.service` for core analytics and user behavior tracking
- **Environment Variables**: `POSTHOG_API_KEY`, `POSTHOG_HOST`, `ANALYTICS_ENABLED`, `ENVIRONMENT` in `core/config.py`
- **Event Tagging**: All events automatically include an `environment` property from `ENVIRONMENT` variable for filtering by deployment environment
- **Usage Patterns**:
  - **API Endpoints**: Import `analytics` directly and use `analytics.track_event()` for custom tracking
  - **Search Operations**: Automatically tracked via `SearchService.search()` - no manual tracking required
  - **Business Events**: Use `business_events.track_*()` methods for high-level metrics
- **Direct Import Pattern**: Use `from airweave.analytics.service import analytics` instead of `ApiContext` for direct access
- **Error Handling**: Analytics errors are logged but never crash the application
- **Property Guidelines**: Use `.value` for enums, avoid mutating caller's properties dict
- **Timing**: Use `time.monotonic()` for duration measurements, not `time.time()`
- **Session Replay Integration**:
  - **Header Flow**: Frontend sends PostHog session ID via `X-Airweave-Session-ID` header
  - **Header Extraction**: `deps._extract_headers_from_request()` extracts session ID into `RequestHeaders.session_id`
  - **PostHog Property**: `ContextualAnalyticsService` automatically adds `$session_id` property to all events
  - **Purpose**: Links backend events to frontend session replays for debugging and user journey analysis
  - **Implementation**: Session ID flows through `RequestHeaders` → `ContextualAnalyticsService` → `_build_base_properties()` → `$session_id` property

### Search Persistence & Analytics
- **Database-First Approach**: Every search is persisted to `search_queries` table via `CRUDSearchQuery`
- **Dual Analytics**: Both database persistence and PostHog events tracked automatically
- **Comprehensive Data**: Database + PostHog contain all search metadata, performance metrics, and feature usage data
- **Non-blocking**: Both persistence methods fail gracefully without affecting search functionality
- **Automatic Tracking**: All search operations (regular, streaming, legacy) tracked uniformly via `SearchService`

### Security

- Never use `random.*` for security-sensitive values — ruff
  rule `S311` bans it
- Use `secrets.choice()`, `secrets.randbelow()`,
  `secrets.token_bytes()`, or `secrets.token_urlsafe()` instead
- Seeded `random.Random(seed)` is acceptable **only** for
  deterministic test-data generation (stub sources)