diff --git a/crates/abyss-backend/migrations/20260824000000_initial_schema/down.sql b/crates/abyss-backend/migrations/20260824000000_initial_schema/down.sql index bdeb4a5..1e929fa 100644 --- a/crates/abyss-backend/migrations/20260824000000_initial_schema/down.sql +++ b/crates/abyss-backend/migrations/20260824000000_initial_schema/down.sql @@ -1,3 +1,5 @@ +-- Drop in reverse dependency order so every foreign key remains valid while +-- the consolidated standalone schema is removed. DROP TABLE IF EXISTS search_outbox; DROP TABLE IF EXISTS agent_diagnostic_capture_events; DROP TABLE IF EXISTS agent_diagnostic_captures; diff --git a/crates/abyss-backend/migrations/20260824000000_initial_schema/up.sql b/crates/abyss-backend/migrations/20260824000000_initial_schema/up.sql index 15410a6..551ee1b 100644 --- a/crates/abyss-backend/migrations/20260824000000_initial_schema/up.sql +++ b/crates/abyss-backend/migrations/20260824000000_initial_schema/up.sql @@ -1,3 +1,6 @@ +-- The standalone backend currently authenticates one deployment-wide owner. +-- Keeping the owner as a real row preserves explicit foreign-key ownership and +-- leaves the event schema ready for a future identity model without nullable IDs. CREATE TABLE app_users ( id uuid PRIMARY KEY, email text NOT NULL, @@ -12,6 +15,8 @@ VALUES ( 'Abyss Owner' ); +-- A device is a user-visible context, not a hardware identity. Host name and +-- platform form the stable bucket; observed OS version and time bounds evolve. CREATE TABLE devices ( id uuid PRIMARY KEY, user_id uuid NOT NULL REFERENCES app_users(id) ON DELETE CASCADE, @@ -28,6 +33,8 @@ CREATE TABLE devices ( CREATE INDEX devices_user_seen_idx ON devices (user_id, last_seen_at DESC); +-- Sessions use the Agent-native session ID plus canonical Agent name. The +-- device reference follows the most recently observed device for that session. CREATE TABLE agent_sessions ( id uuid PRIMARY KEY, user_id uuid NOT NULL REFERENCES app_users(id) ON DELETE CASCADE, @@ -46,6 +53,8 @@ CREATE TABLE agent_sessions ( CREATE INDEX agent_sessions_user_time_idx ON agent_sessions (user_id, started_at DESC); +-- Turn indexes are backend-normalized during ingest when stable provider or +-- Agent metadata reveals that a collector restarted its local counter. CREATE TABLE agent_turns ( id uuid PRIMARY KEY, user_id uuid NOT NULL REFERENCES app_users(id) ON DELETE CASCADE, @@ -61,6 +70,8 @@ CREATE TABLE agent_turns ( CREATE INDEX agent_turns_user_session_idx ON agent_turns (user_id, session_pk, turn_index); +-- Usage events are immutable source records. event_id provides global ingest +-- idempotency, while denormalized labels keep filters and aggregation direct. CREATE TABLE llm_usage_events ( id uuid PRIMARY KEY, user_id uuid NOT NULL REFERENCES app_users(id) ON DELETE CASCADE, @@ -100,6 +111,8 @@ CREATE INDEX llm_usage_events_session_turn_idx CREATE INDEX llm_usage_events_agent_model_time_idx ON llm_usage_events (agent_name, llm_provider, llm_model, observed_at DESC); +-- Attachment content is optional so collectors can report image usage metadata +-- without uploading bytes. Hash, size, type, and position remain queryable. CREATE TABLE llm_usage_event_attachments ( id uuid PRIMARY KEY, user_id uuid NOT NULL REFERENCES app_users(id) ON DELETE CASCADE, @@ -119,6 +132,8 @@ CREATE TABLE llm_usage_event_attachments ( CREATE INDEX llm_usage_event_attachments_user_event_idx ON llm_usage_event_attachments (user_id, event_pk, position); +-- Diagnostic payloads are deliberately opaque; the join table records the +-- usage events that establish their owner, session, and device context. CREATE TABLE agent_diagnostic_captures ( id uuid PRIMARY KEY, user_id uuid NOT NULL REFERENCES app_users(id) ON DELETE CASCADE, @@ -148,6 +163,9 @@ CREATE TABLE agent_diagnostic_capture_events ( CREATE INDEX agent_diagnostic_capture_events_event_idx ON agent_diagnostic_capture_events (event_pk); +-- Search is a rebuildable projection. Outbox rows intentionally do not foreign +-- key event_pk so a delete operation can survive removal of its source row. +-- Expiring claims and terminal dead letters support concurrent crash recovery. CREATE TABLE search_outbox ( id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, event_pk uuid NOT NULL, diff --git a/crates/abyss-backend/src/api.rs b/crates/abyss-backend/src/api.rs index 9716767..0eb73d5 100644 --- a/crates/abyss-backend/src/api.rs +++ b/crates/abyss-backend/src/api.rs @@ -1,4 +1,9 @@ -//! HTTP routes for standalone Agent event ingestion and queries. +//! HTTP boundary for standalone Agent event ingestion and queries. +//! +//! The health endpoints are public, while every event, attachment, summary, +//! timeline, and search endpoint authenticates the deployment bearer token. +//! Diesel is synchronous, so all database access is moved onto Tokio's blocking +//! pool through [`run_db`] rather than running on asynchronous executor threads. use axum::{ Json, Router, @@ -31,16 +36,28 @@ use crate::{ const MAX_INGEST_REQUEST_BODY_BYTES: usize = 16 * 1024 * 1024; #[derive(Clone)] +/// Cloneable dependencies and request limits shared by all Axum handlers. pub struct AppState { + /// Deployment label exposed by the informational root endpoint. pub environment: String, + /// Maximum events and diagnostic captures accepted per ingest request. pub max_ingest_batch_size: usize, + /// Default upper bound for summary aggregation rows. pub summary_scan_limit: i64, + /// Default raw-event page size. pub default_page_size: i64, + /// Deployment-wide bearer-token validator. pub identity: IdentityAuthenticator, + /// Optional full-text search service. pub search: Option, + /// PostgreSQL connection pool used by request handlers. pub pool: DbPool, } +/// Builds the complete HTTP router for one backend process. +/// +/// The ingest body limit is intentionally route-local so future endpoints do +/// not inherit a large request allowance by accident. pub fn router(state: AppState) -> Router { Router::new() .route("/", get(root)) @@ -82,6 +99,8 @@ async fn health() -> Json { } async fn ready(State(state): State) -> Result, AppError> { + // Readiness checks the source of truth only. Search is an optional derived + // service and its temporary failure must not remove ingestion capacity. run_db(state, db::check_ready).await?; Ok(Json(ServiceStatus { service: "abyss-backend", @@ -133,6 +152,8 @@ async fn session_search( .ok_or_else(|| AppError::unavailable("session search is not configured".to_owned()))?; let execution = search.search(user_id, query).await?; let session_ids = execution.session_ids(); + // Elasticsearch contains only a bounded search projection. Authoritative + // session/device details are reloaded from PostgreSQL under the owner scope. let details = run_db(state, move |connection| { SearchOutboxRepository::session_details(connection, user_id, &session_ids) }) @@ -189,6 +210,8 @@ async fn image_attachment( header::CACHE_CONTROL, HeaderValue::from_static("private, no-store"), ); + // Attachments may contain sensitive conversation context. Disallow MIME + // sniffing, shared caching, and cross-origin embedding even for valid images. response_headers.insert( header::X_CONTENT_TYPE_OPTIONS, HeaderValue::from_static("nosniff"), @@ -219,6 +242,8 @@ where T: Send + 'static, F: FnOnce(&mut PgConnection) -> Result + Send + 'static, { + // Diesel and r2d2 are blocking APIs. Acquiring the pool connection inside + // spawn_blocking also keeps pool contention off Tokio worker threads. task::spawn_blocking(move || { let mut connection = state.pool.get()?; task_fn(&mut connection) diff --git a/crates/abyss-backend/src/config.rs b/crates/abyss-backend/src/config.rs index f2a5513..ea920dc 100644 --- a/crates/abyss-backend/src/config.rs +++ b/crates/abyss-backend/src/config.rs @@ -1,4 +1,9 @@ //! Environment-backed standalone backend configuration. +//! +//! Every process setting is read once during startup. Empty or whitespace-only +//! values are treated as absent, positive numeric settings are rejected at the +//! boundary, and Elasticsearch is enabled only when its URL is present. This +//! leaves the rest of the service with a fully validated, immutable snapshot. use std::{env, net::SocketAddr, num::ParseIntError}; @@ -15,22 +20,36 @@ const DEFAULT_SEARCH_REQUEST_TIMEOUT_SECONDS: u64 = 10; const DEFAULT_SEARCH_POLL_INTERVAL_MILLISECONDS: u64 = 500; const DEFAULT_SEARCH_BATCH_SIZE: i64 = 100; +/// Complete runtime configuration shared by process startup and HTTP state. pub struct Config { + /// Socket address on which the HTTP server listens. pub addr: SocketAddr, + /// Deployment label returned by the root endpoint and used by safety checks. pub environment: String, + /// Explicit escape hatch for running black-box instances in containers. pub blackbox_allow_non_loopback: bool, + /// Default tracing filter used when `RUST_LOG` is not set. pub log_level: String, + /// PostgreSQL connection string. pub database_url: String, + /// Maximum number of PostgreSQL connections held by the r2d2 pool. pub database_pool_size: u32, + /// Whether embedded Diesel migrations run before the listener starts. pub run_migrations: bool, + /// Maximum events and diagnostic captures accepted in one ingest request. pub max_ingest_batch_size: usize, + /// Default and maximum row limit used by summary aggregation. pub summary_scan_limit: i64, + /// Default number of events returned by paginated raw-event queries. pub default_page_size: i64, + /// Deployment-wide bearer-token authentication configuration. pub identity: IdentityConfig, + /// Optional Elasticsearch projection configuration. pub search: Option, } impl Config { + /// Reads and validates all `ABYSS_BACKEND_*` environment variables. pub fn from_env() -> Result { let addr = read_env("ABYSS_BACKEND_ADDR", DEFAULT_ADDR) .parse::() @@ -73,12 +92,19 @@ impl Config { } #[derive(Clone)] +/// Settings required by the Elasticsearch client and projection worker. pub struct SearchConfig { + /// Base Elasticsearch URL, without an index or API path suffix. pub endpoint: String, + /// Optional HTTP Basic Authentication username. pub username: Option, + /// Optional HTTP Basic Authentication password. pub password: Option, + /// Per-request Elasticsearch timeout in seconds. pub request_timeout_seconds: u64, + /// Idle polling delay for the search outbox worker. pub poll_interval_milliseconds: u64, + /// Maximum outbox rows claimed or backfilled per worker iteration. pub batch_size: i64, } @@ -88,6 +114,8 @@ impl SearchConfig { let username = env_value("ABYSS_BACKEND_ELASTICSEARCH_USERNAME"); let password = env_value("ABYSS_BACKEND_ELASTICSEARCH_PASSWORD"); + // Credentials without an endpoint almost always indicate a misspelled + // or missing secret, so fail startup instead of silently disabling search. let Some(endpoint) = endpoint else { if username.is_some() || password.is_some() { return Err(AppError::config( @@ -98,6 +126,7 @@ impl SearchConfig { return Ok(None); }; + // Basic Authentication is only meaningful as a complete pair. if username.is_some() != password.is_some() { return Err(AppError::config( "ABYSS_BACKEND_ELASTICSEARCH_USERNAME and ABYSS_BACKEND_ELASTICSEARCH_PASSWORD must be configured together" @@ -134,6 +163,8 @@ fn read_required_env(key: &str) -> Result { } fn env_value(key: &str) -> Option { + // Normalizing empty secrets to None makes mounted-but-empty Kubernetes + // secrets behave the same as missing environment variables. env::var(key) .ok() .map(|value| value.trim().to_owned()) diff --git a/crates/abyss-backend/src/db/mod.rs b/crates/abyss-backend/src/db/mod.rs index ab73841..0196460 100644 --- a/crates/abyss-backend/src/db/mod.rs +++ b/crates/abyss-backend/src/db/mod.rs @@ -1,6 +1,12 @@ -//! Database connection pool, migrations, and persistence modules. +//! PostgreSQL connection pool, embedded migrations, and persistence modules. +//! +//! A synchronous Diesel pool is shared by HTTP handlers and the search worker. +//! Callers running on Tokio are responsible for entering this module through a +//! blocking task so neither connection acquisition nor SQL blocks the executor. +/// Diesel models used for reads and inserts. pub mod models; +/// Diesel's compile-time representation of the PostgreSQL schema. pub mod schema; use diesel::{PgConnection, RunQueryDsl, r2d2::ConnectionManager, sql_query}; @@ -8,10 +14,12 @@ use diesel_migrations::{EmbeddedMigrations, MigrationHarness, embed_migrations}; use crate::{config::Config, error::AppError}; +/// Cloneable pool of synchronous PostgreSQL connections. pub type DbPool = r2d2::Pool>; const MIGRATIONS: EmbeddedMigrations = embed_migrations!("migrations"); +/// Creates the PostgreSQL pool and verifies that its initial connection opens. pub fn create_pool(config: &Config) -> Result { let manager = ConnectionManager::::new(config.database_url.clone()); r2d2::Pool::builder() @@ -20,12 +28,14 @@ pub fn create_pool(config: &Config) -> Result { .map_err(AppError::from) } +/// Applies all embedded migrations that are not recorded by Diesel yet. pub fn run_migrations(pool: &DbPool) -> Result<(), Box> { let mut connection = pool.get()?; connection.run_pending_migrations(MIGRATIONS)?; Ok(()) } +/// Executes the minimal PostgreSQL query used by the readiness endpoint. pub fn check_ready(connection: &mut PgConnection) -> Result<(), AppError> { sql_query("SELECT 1").execute(connection)?; Ok(()) diff --git a/crates/abyss-backend/src/db/models.rs b/crates/abyss-backend/src/db/models.rs index f2ba98d..9e3b48a 100644 --- a/crates/abyss-backend/src/db/models.rs +++ b/crates/abyss-backend/src/db/models.rs @@ -1,4 +1,8 @@ //! Diesel row and insert models owned by the Agent event service. +//! +//! Query models mirror complete table rows because Diesel uses them for typed +//! selection. Insert models make server-assigned fields explicit and prevent +//! HTTP payloads from being written directly without normalization. use chrono::{DateTime, Utc}; use diesel::{Insertable, Queryable, Selectable}; @@ -13,206 +17,356 @@ use super::schema::{ #[derive(Clone, Debug, Queryable, Selectable)] #[diesel(table_name = devices)] #[diesel(check_for_backend(diesel::pg::Pg))] +/// Persisted device context keyed by owner, host name, and platform. pub struct Device { + /// Server-generated device primary key. pub id: Uuid, + /// Owning user identifier. pub user_id: Uuid, + /// User-visible host name. pub host_name: String, + /// Normalized platform slug. pub platform: String, + /// Most recently observed OS version. pub os_version: Option, + /// Earliest event observation associated with the device. pub first_seen_at: DateTime, + /// Latest event observation associated with the device. pub last_seen_at: DateTime, + /// Row creation time. pub created_at: DateTime, + /// Last metadata update time. pub updated_at: DateTime, } #[derive(Debug, Insertable)] #[diesel(table_name = devices)] +/// Values used to create a device context. pub struct NewDevice { + /// Server-generated device primary key. pub id: Uuid, + /// Owning user identifier. pub user_id: Uuid, + /// User-visible host name. pub host_name: String, + /// Normalized platform slug. pub platform: String, + /// Optional observed OS version. pub os_version: Option, + /// Initial earliest observation time. pub first_seen_at: DateTime, + /// Initial latest observation time. pub last_seen_at: DateTime, + /// Row creation time. pub created_at: DateTime, + /// Initial metadata update time. pub updated_at: DateTime, } #[derive(Clone, Debug, Queryable, Selectable)] #[diesel(table_name = agent_sessions)] #[diesel(check_for_backend(diesel::pg::Pg))] +/// Persisted Agent session containing many normalized turns. pub struct AgentSession { + /// Server-generated session primary key. pub id: Uuid, + /// Owning user identifier. pub user_id: Uuid, + /// Most recently associated device-context key. pub device_context_id: Uuid, + /// Canonical Agent name. pub agent_name: String, + /// Most recently observed Agent version. pub agent_version: Option, + /// Agent-native session identifier. pub session_id: String, + /// Earliest event observation in the session. pub started_at: DateTime, + /// Latest event observation in the session. pub ended_at: Option>, + /// Stable allowlisted session metadata. pub metadata: Value, + /// Row creation time. pub created_at: DateTime, + /// Last session metadata update time. pub updated_at: DateTime, } #[derive(Debug, Insertable)] #[diesel(table_name = agent_sessions)] +/// Values used to create an Agent session. pub struct NewAgentSession { + /// Server-generated session primary key. pub id: Uuid, + /// Owning user identifier. pub user_id: Uuid, + /// Associated device-context key. pub device_context_id: Uuid, + /// Canonical Agent name. pub agent_name: String, + /// Optional Agent version. pub agent_version: Option, + /// Agent-native session identifier. pub session_id: String, + /// Initial earliest observation time. pub started_at: DateTime, + /// Initial latest observation time. pub ended_at: Option>, + /// Stable allowlisted session metadata. pub metadata: Value, + /// Row creation time. pub created_at: DateTime, + /// Initial metadata update time. pub updated_at: DateTime, } #[derive(Clone, Debug, Queryable, Selectable)] #[diesel(table_name = agent_turns)] #[diesel(check_for_backend(diesel::pg::Pg))] +/// Persisted logical turn within one Agent session. pub struct AgentTurn { + /// Server-generated turn primary key. pub id: Uuid, + /// Owning user identifier. pub user_id: Uuid, + /// Parent session primary key. pub session_pk: Uuid, + /// Positive turn number unique within the session. pub turn_index: i32, + /// Earliest event observation in the turn. pub started_at: DateTime, + /// Latest event observation in the turn. pub ended_at: Option>, + /// Row creation time. pub created_at: DateTime, + /// Last turn-boundary update time. pub updated_at: DateTime, } #[derive(Debug, Insertable)] #[diesel(table_name = agent_turns)] +/// Values used to create a logical Agent turn. pub struct NewAgentTurn { + /// Server-generated turn primary key. pub id: Uuid, + /// Owning user identifier. pub user_id: Uuid, + /// Parent session primary key. pub session_pk: Uuid, + /// Positive turn number unique within the session. pub turn_index: i32, + /// Initial earliest observation time. pub started_at: DateTime, + /// Initial latest observation time. pub ended_at: Option>, + /// Row creation time. pub created_at: DateTime, + /// Initial boundary update time. pub updated_at: DateTime, } #[derive(Clone, Debug, Queryable, Selectable)] #[diesel(table_name = llm_usage_events)] #[diesel(check_for_backend(diesel::pg::Pg))] +/// Source-of-truth row for one immutable LLM usage observation. pub struct UsageEvent { + /// Server-generated event primary key. pub id: Uuid, + /// Owning user identifier. pub user_id: Uuid, + /// Device context observed for the event. pub device_context_id: Uuid, + /// Parent session primary key. pub session_pk: Uuid, + /// Parent normalized-turn primary key. pub turn_pk: Uuid, + /// Collector-generated idempotency key. pub event_id: String, + /// Canonical Agent name. pub agent_name: String, + /// Agent version observed for the event. pub agent_version: Option, + /// Agent-native session identifier copied for direct filtering. pub session_id: String, + /// Normalized turn number copied for direct filtering. pub turn_index: i32, + /// Canonical LLM provider slug. pub llm_provider: String, + /// Provider-specific model identifier. pub llm_model: String, + /// Canonical request or response value. pub event_type: String, + /// Optional prompt or response text. pub text: Option, + /// Binary SHA-256 digest of text when present. pub text_sha256: Option>, + /// Normalized input-token count. pub input_tokens: i64, + /// Normalized output-token count. pub output_tokens: i64, + /// Normalized cache-read-token count. pub cache_read_tokens: i64, + /// Normalized cache-write-token count. pub cache_write_tokens: i64, + /// Normalized reasoning-token count. pub reasoning_tokens: i64, + /// Provider-reported or derived total-token count. pub total_tokens: i64, + /// Collector observation time. pub observed_at: DateTime, + /// Collector metadata retained with the event. pub metadata: Value, + /// Server insertion time. pub created_at: DateTime, } #[derive(Queryable, Selectable)] #[diesel(table_name = llm_usage_event_attachments)] #[diesel(check_for_backend(diesel::pg::Pg))] +/// Persisted metadata and optional bytes for one event image. pub struct UsageEventAttachment { + /// Server-generated attachment primary key. pub id: Uuid, + /// Owning user identifier copied for direct authorization. pub user_id: Uuid, + /// Parent usage-event primary key. pub event_pk: Uuid, + /// Unique display position within the event. pub position: i32, + /// Validated browser-safe MIME type. pub media_type: String, + /// Declared decoded content length. pub byte_size: i64, + /// Binary SHA-256 digest. pub sha256: Vec, + /// Optional image bytes for metadata-only attachments. pub content: Option>, + /// Server insertion time. pub created_at: DateTime, } #[derive(Insertable)] #[diesel(table_name = agent_diagnostic_captures)] +/// Values used to create an opaque diagnostic capture. pub struct NewAgentDiagnosticCapture { + /// Server-generated capture primary key. pub id: Uuid, + /// Owning user identifier. pub user_id: Uuid, + /// Correlated device-context key. pub device_context_id: Uuid, + /// Correlated session key. pub session_pk: Uuid, + /// Collector-generated idempotency key. pub capture_id: String, + /// Collector flow identifier. pub flow_id: String, + /// Collector observation time. pub captured_at: DateTime, + /// Collector version that produced the payload. pub collector_version: String, + /// Opaque diagnostic JSON. pub payload: Value, + /// Server insertion time. pub created_at: DateTime, } #[derive(Insertable)] #[diesel(table_name = agent_diagnostic_capture_events)] +/// Join row linking one diagnostic capture to one usage event. pub struct NewAgentDiagnosticCaptureEvent { + /// Diagnostic capture primary key. pub capture_pk: Uuid, + /// Correlated usage-event primary key. pub event_pk: Uuid, } #[derive(Debug, Insertable)] #[diesel(table_name = llm_usage_events)] +/// Normalized values used to create one immutable usage event. pub struct NewUsageEvent { + /// Server-generated event primary key. pub id: Uuid, + /// Owning user identifier. pub user_id: Uuid, + /// Associated device-context key. pub device_context_id: Uuid, + /// Parent session primary key. pub session_pk: Uuid, + /// Parent normalized-turn primary key. pub turn_pk: Uuid, + /// Collector-generated idempotency key. pub event_id: String, + /// Canonical Agent name. pub agent_name: String, + /// Optional Agent version. pub agent_version: Option, + /// Agent-native session identifier. pub session_id: String, + /// Backend-normalized turn number. pub turn_index: i32, + /// Canonical LLM provider slug. pub llm_provider: String, + /// Provider-specific model identifier. pub llm_model: String, + /// Canonical request or response value. pub event_type: String, + /// Optional prompt or response text. pub text: Option, + /// Optional binary SHA-256 of text. pub text_sha256: Option>, + /// Normalized input-token count. pub input_tokens: i64, + /// Normalized output-token count. pub output_tokens: i64, + /// Normalized cache-read-token count. pub cache_read_tokens: i64, + /// Normalized cache-write-token count. pub cache_write_tokens: i64, + /// Normalized reasoning-token count. pub reasoning_tokens: i64, + /// Provider-reported or derived total-token count. pub total_tokens: i64, + /// Collector observation time. pub observed_at: DateTime, + /// Collector metadata retained with the event. pub metadata: Value, + /// Server insertion time. pub created_at: DateTime, } #[derive(Insertable)] #[diesel(table_name = llm_usage_event_attachments)] +/// Validated values used to create one event attachment. pub struct NewUsageEventAttachment { + /// Server-generated attachment primary key. pub id: Uuid, + /// Owning user identifier. pub user_id: Uuid, + /// Parent usage-event primary key. pub event_pk: Uuid, + /// Unique display position within the event. pub position: i32, + /// Validated browser-safe MIME type. pub media_type: String, + /// Decoded content length. pub byte_size: i64, + /// Binary SHA-256 digest. pub sha256: Vec, + /// Optional decoded image bytes. pub content: Option>, + /// Server insertion time. pub created_at: DateTime, } #[derive(Insertable)] #[diesel(table_name = search_outbox)] +/// Durable search-projection task created with a new usage event. pub struct NewSearchOutboxTask { + /// Source usage-event primary key and Elasticsearch document key. pub event_pk: Uuid, + /// Owning user identifier retained for projection bookkeeping. pub user_id: Uuid, + /// Source event insertion time. pub created_at: DateTime, } diff --git a/crates/abyss-backend/src/error.rs b/crates/abyss-backend/src/error.rs index befa6a7..9784337 100644 --- a/crates/abyss-backend/src/error.rs +++ b/crates/abyss-backend/src/error.rs @@ -1,4 +1,8 @@ -//! Shared error types for HTTP and database operations. +//! Shared error types and their safe HTTP representation. +//! +//! Client-actionable failures retain their messages. Configuration, database, +//! pool, and internal failures are logged server-side and deliberately reduced +//! to a generic response so implementation details and secrets are not exposed. use axum::{ Json, @@ -8,46 +12,61 @@ use axum::{ use serde::Serialize; #[derive(Debug, thiserror::Error)] +/// Error vocabulary shared across configuration, HTTP, and persistence layers. pub enum AppError { + /// Invalid process configuration detected during startup. #[error("configuration error: {0}")] Config(String), + /// Diesel query or transaction failure. #[error("database error: {0}")] Database(#[from] diesel::result::Error), + /// PostgreSQL connection-pool failure. #[error("connection pool error: {0}")] Pool(#[from] r2d2::Error), + /// Authenticated resource does not exist. #[error("not found: {0}")] NotFound(String), + /// Missing or invalid deployment bearer token. #[error("unauthorized: {0}")] Unauthorized(String), + /// Optional dependency is disabled or temporarily unavailable. #[error("unavailable: {0}")] Unavailable(String), + /// Request data violates an API contract. #[error("validation error: {0}")] Validation(String), + /// Unexpected application invariant or task failure. #[error("internal error: {0}")] Internal(String), } impl AppError { + /// Constructs a configuration error. pub const fn config(message: String) -> Self { Self::Config(message) } + /// Constructs a request validation error. pub const fn validation(message: String) -> Self { Self::Validation(message) } + /// Constructs a resource-not-found error. pub const fn not_found(message: String) -> Self { Self::NotFound(message) } + /// Constructs an authentication error. pub const fn unauthorized(message: String) -> Self { Self::Unauthorized(message) } + /// Constructs a dependency-unavailable error. pub const fn unavailable(message: String) -> Self { Self::Unavailable(message) } + /// Constructs an unexpected internal error. pub const fn internal(message: String) -> Self { Self::Internal(message) } diff --git a/crates/abyss-backend/src/identity.rs b/crates/abyss-backend/src/identity.rs index 90a5d4f..1b312d5 100644 --- a/crates/abyss-backend/src/identity.rs +++ b/crates/abyss-backend/src/identity.rs @@ -1,4 +1,9 @@ //! Standalone bearer authentication mapped to one deployment owner. +//! +//! This module is intentionally not a login or SSO implementation. Operators +//! provision one opaque API token, store only its SHA-256 hash in configuration, +//! and every valid request maps to [`OWNER_ID`]. This keeps authorization data +//! ownership explicit until a multi-user identity system is introduced. use axum::http::{HeaderMap, header}; use sha2::{Digest, Sha256}; @@ -7,14 +12,20 @@ use uuid::Uuid; use crate::error::AppError; +/// Stable database owner seeded by the initial migration. pub const OWNER_ID: Uuid = Uuid::from_u128(1); #[derive(Clone)] +/// Canonical SHA-256 digest of the deployment bearer token. pub struct IdentityConfig { token_hash: [u8; 32], } impl IdentityConfig { + /// Parses exactly 32 bytes encoded as lowercase hexadecimal. + /// + /// Requiring canonical encoding avoids accepting multiple textual forms of + /// the same secret and catches malformed deployment secrets at startup. pub fn parse(encoded_hash: &str) -> Result { let bytes = hex::decode(encoded_hash).map_err(|_error| invalid_hash())?; let token_hash = <[u8; 32]>::try_from(bytes.as_slice()).map_err(|_error| invalid_hash())?; @@ -26,19 +37,23 @@ impl IdentityConfig { } #[derive(Clone)] +/// Validates HTTP bearer credentials for the standalone owner. pub struct IdentityAuthenticator { config: IdentityConfig, } impl IdentityAuthenticator { + /// Creates an authenticator from startup-validated configuration. #[must_use] pub const fn new(config: IdentityConfig) -> Self { Self { config } } + /// Authenticates one request and returns the database owner identifier. pub fn authenticate(&self, headers: &HeaderMap) -> Result { let token = bearer_token(headers)?; let presented_hash = Sha256::digest(token.as_bytes()); + // Constant-time comparison avoids leaking how many digest bytes match. if !bool::from( presented_hash .as_slice() diff --git a/crates/abyss-backend/src/main.rs b/crates/abyss-backend/src/main.rs index 24fb66b..b643572 100644 --- a/crates/abyss-backend/src/main.rs +++ b/crates/abyss-backend/src/main.rs @@ -1,5 +1,12 @@ -//! HTTP entrypoint for the Abyss backend service. - +//! Process entrypoint and lifecycle management for the Abyss backend. +//! +//! Startup is deliberately ordered: configuration is validated before any +//! external connection is opened, database migrations finish before the HTTP +//! listener accepts traffic, and the optional search worker shares the same +//! shutdown signal as the server. PostgreSQL remains the source of truth; +//! Elasticsearch is only a derived, eventually consistent projection. + +#![warn(missing_docs)] #![expect( clippy::multiple_crate_versions, reason = "Axum and Diesel currently pull a few distinct transitive crate versions." @@ -31,6 +38,8 @@ async fn main() -> Result<(), Box> { run_migrations(&pool)?; } + // Search is optional. Keeping the service itself in an Option makes the + // disabled state explicit all the way through routing and worker startup. let search = config .search .as_ref() @@ -92,6 +101,9 @@ async fn stop_search_worker(worker: Option>) { } fn validate_runtime_config(config: &Config) -> Result<(), error::AppError> { + // Black-box tests use disposable credentials and data. Requiring loopback + // prevents an accidentally started test instance from becoming reachable + // on the surrounding network; containers must opt out explicitly. if config.environment == "blackbox" && !config.addr.ip().is_loopback() && !config.blackbox_allow_non_loopback @@ -105,6 +117,8 @@ fn validate_runtime_config(config: &Config) -> Result<(), error::AppError> { } fn init_tracing(config: &Config) { + // RUST_LOG takes precedence for standard operational overrides. The + // configured level is only the fallback when no valid filter is supplied. let filter = EnvFilter::try_from_default_env() .unwrap_or_else(|_| EnvFilter::new(config.log_level.clone())); tracing_subscriber::fmt().with_env_filter(filter).init(); diff --git a/crates/abyss-backend/src/search/document.rs b/crates/abyss-backend/src/search/document.rs index d1d1465..8937d7d 100644 --- a/crates/abyss-backend/src/search/document.rs +++ b/crates/abyss-backend/src/search/document.rs @@ -1,4 +1,9 @@ //! Search document projection and bounded extraction from usage-event metadata. +//! +//! Only explicitly allowlisted metadata enters the index. Extraction limits +//! depth, value count, and character length to bound CPU, allocation, request +//! size, and accidental indexing of unrelated collector metadata. PostgreSQL +//! retains the complete source event and remains the recovery source. use serde::Serialize; use serde_json::Value; @@ -16,30 +21,47 @@ const MAX_JSON_DEPTH: usize = 8; /// Elasticsearch representation of one immutable usage event. #[derive(Serialize)] pub struct SearchDocument { + /// Source usage-event primary key and Elasticsearch document identifier. pub event_pk: Uuid, + /// Owner identifier used as a mandatory search filter. pub user_id: Uuid, + /// Parent session key used to collapse event hits into sessions. pub session_pk: Uuid, + /// Agent-native session identifier searchable as text and keyword. pub session_id: String, + /// Parent turn primary key. pub turn_pk: Uuid, + /// Backend-normalized turn number. pub turn_index: i32, + /// Canonical Agent name used for exact filtering. pub agent_name: String, + /// Canonical LLM provider slug used for exact filtering. pub llm_provider: String, + /// Provider-specific model identifier used for exact filtering. pub llm_model: String, + /// Request or response value used for exact filtering. pub event_type: String, + /// Collector observation time used for range filtering. pub observed_at: chrono::DateTime, + /// Bounded prompt or response text. #[serde(skip_serializing_if = "Option::is_none")] pub content: Option, + /// Bounded names extracted from tool-call metadata. #[serde(skip_serializing_if = "Vec::is_empty")] pub tool_names: Vec, + /// Bounded raw tool inputs and outputs. #[serde(skip_serializing_if = "Vec::is_empty")] pub tool_content: Vec, + /// Bounded command strings extracted from structured tool input. #[serde(skip_serializing_if = "Vec::is_empty")] pub commands: Vec, + /// Bounded paths and working directories extracted from allowlisted fields. #[serde(skip_serializing_if = "Vec::is_empty")] pub file_paths: Vec, } impl SearchDocument { + /// Projects one source event into the strict Elasticsearch mapping. #[must_use] pub fn from_event(event: UsageEvent) -> Self { let metadata = SearchableMetadata::extract(&event.metadata); @@ -87,6 +109,9 @@ impl SearchableMetadata { extracted.push_file_path(working_directory); } + // Deliberately ignore every top-level key except working_directory and + // content_segments so headers, credentials, and image data stay out of + // the derived index even when present in raw metadata. let Some(segments) = metadata.get("content_segments").and_then(Value::as_array) else { return extracted; }; @@ -131,6 +156,8 @@ impl SearchableMetadata { _ => {} } } + // Recurse after examining the current key so nested command + // or path fields are discoverable without indexing all JSON. self.extract_structured_input(child, remaining_depth.saturating_sub(1)); } } @@ -161,6 +188,8 @@ fn push_bounded(values: &mut Vec, value: &str) { } fn sanitize_search_text(value: &str) -> String { + // Marker tokens are controlled by this service during highlighting. Strip + // collector-supplied copies so clients cannot forge highlighted segments. value .replace(HIGHLIGHT_START, "") .replace(HIGHLIGHT_END, "") diff --git a/crates/abyss-backend/src/search/elasticsearch.rs b/crates/abyss-backend/src/search/elasticsearch.rs index c132fb2..28187f7 100644 --- a/crates/abyss-backend/src/search/elasticsearch.rs +++ b/crates/abyss-backend/src/search/elasticsearch.rs @@ -1,4 +1,9 @@ //! Minimal Elasticsearch HTTP boundary for fixed-index search and bulk projection. +//! +//! This module owns every Elasticsearch-specific request and response shape so +//! the rest of the service depends on typed domain results. The index name and +//! strict mapping are fixed by the backend; operators configure only the base +//! endpoint, optional Basic Authentication, and request timeout. use std::{collections::HashMap, time::Duration}; @@ -12,13 +17,18 @@ use crate::{config::SearchConfig, error::AppError}; use super::{ValidatedSearchQuery, document::SearchDocument}; +/// Fixed name of the derived usage-event index. pub const SEARCH_INDEX: &str = "abyss_usage_events"; +/// Sentinel inserted before text matched by Elasticsearch highlighting. pub const HIGHLIGHT_START: &str = "[[[ABYSS_SEARCH_HIGHLIGHT_START]]]"; +/// Sentinel inserted after text matched by Elasticsearch highlighting. pub const HIGHLIGHT_END: &str = "[[[ABYSS_SEARCH_HIGHLIGHT_END]]]"; /// One idempotent operation submitted through Elasticsearch's Bulk API. pub enum BulkOperation { + /// Create or replace a document from current source-event state. Index(Box), + /// Remove a document whose source event no longer exists. Delete(Uuid), } @@ -36,6 +46,7 @@ impl BulkOperation { } #[derive(Clone)] +/// Small HTTP client for the backend-owned Elasticsearch index. pub struct ElasticsearchClient { client: reqwest::Client, endpoint: Url, @@ -44,6 +55,7 @@ pub struct ElasticsearchClient { } impl ElasticsearchClient { + /// Builds a client after validating endpoint restrictions and credentials. pub fn new(config: &SearchConfig) -> Result { let endpoint = Url::parse(&config.endpoint).map_err(|error| { AppError::config(format!( @@ -66,6 +78,8 @@ impl ElasticsearchClient { "ABYSS_BACKEND_ELASTICSEARCH_URL must not contain a query or fragment".to_owned(), )); } + // Ignore ambient HTTP proxy variables. A private Elasticsearch endpoint + // and its credentials should not be routed through an unrelated proxy. let client = reqwest::Client::builder() .timeout(Duration::from_secs(config.request_timeout_seconds)) .no_proxy() @@ -150,6 +164,7 @@ impl ElasticsearchClient { return Err(response_error("write Elasticsearch bulk request", response).await); } let response = response.json::().await?; + // Positional correspondence is required to update the durable outbox. if response.items.len() != operations.len() { return Err(SearchClientError::Protocol(format!( "Elasticsearch bulk response returned {} items for {} operations", @@ -168,6 +183,7 @@ impl ElasticsearchClient { .collect()) } + /// Executes an owner-scoped, session-collapsed full-text query. pub async fn search( &self, user_id: Uuid, @@ -203,45 +219,71 @@ impl ElasticsearchClient { } } +/// Parsed Elasticsearch page before PostgreSQL hydration. pub struct SearchMatchPage { + /// Approximate number of distinct matching sessions. pub total_sessions: u64, + /// Collapsed session hits in Elasticsearch relevance order. pub sessions: Vec, } +/// Matching events collapsed under one session. pub struct SessionMatches { + /// Backend session primary key. pub session_pk: Uuid, + /// Total events matching within the session. pub match_count: u64, + /// Bounded strongest matching events. pub events: Vec, } +/// Parsed event metadata and raw highlight fragments from Elasticsearch. pub struct MatchedEvent { + /// Backend event primary key. pub event_pk: Uuid, + /// Backend turn primary key. pub turn_pk: Uuid, + /// Normalized turn number. pub turn_index: i32, + /// Request or response side. pub event_type: String, + /// Canonical LLM provider slug. pub llm_provider: String, + /// Provider-specific model identifier. pub llm_model: String, + /// Collector observation time. pub observed_at: DateTime, + /// Ordered raw fragments containing backend sentinel markers. pub fragments: Vec, } #[derive(Debug, thiserror::Error)] +/// Failures produced by the Elasticsearch protocol boundary. pub enum SearchClientError { + /// Transport, TLS, timeout, or response-decoding failure from reqwest. #[error("Elasticsearch request failed: {0}")] Request(#[from] reqwest::Error), + /// Failure while constructing NDJSON or JSON request data. #[error("serialize Elasticsearch request: {0}")] Serialization(#[from] serde_json::Error), + /// Non-success HTTP response with a bounded body for diagnostics. #[error("{operation} returned HTTP {status}: {body}")] Response { + /// Backend operation that received the response. operation: &'static str, + /// Elasticsearch HTTP status. status: StatusCode, + /// Response body truncated to a safe diagnostic bound. body: String, }, + /// Structurally valid JSON that violates an expected ES response contract. #[error("invalid Elasticsearch response: {0}")] Protocol(String), } fn index_definition() -> Value { + // Strict dynamic mapping rejects accidental expansion when a projection + // field is added without an intentional mapping and review. json!({ "mappings": { "dynamic": "strict", @@ -268,6 +310,8 @@ fn index_definition() -> Value { } fn search_request(user_id: Uuid, query: &ValidatedSearchQuery) -> Value { + // Authorization is encoded as a mandatory filter rather than a scoring + // clause, guaranteeing that foreign documents cannot enter the hit set. let mut filters = vec![json!({"term": {"user_id": user_id}})]; if query.from.is_some() || query.to.is_some() { let mut range = serde_json::Map::new(); @@ -309,6 +353,8 @@ fn search_request(user_id: Uuid, query: &ValidatedSearchQuery) -> Value { }] } }, + // Pagination applies to collapsed sessions, while inner_hits returns a + // bounded sample of the strongest matching events for each session. "collapse": { "field": "session_pk", "inner_hits": { @@ -375,6 +421,8 @@ async fn response_error(operation: &'static str, response: reqwest::Response) -> } async fn bounded_response_body(response: reqwest::Response) -> String { + // Elasticsearch errors can echo request data. Bound retained/logged text to + // avoid turning a dependency failure into uncontrolled memory or log use. let body = response .text() .await diff --git a/crates/abyss-backend/src/search/mod.rs b/crates/abyss-backend/src/search/mod.rs index f1f822b..26ede7a 100644 --- a/crates/abyss-backend/src/search/mod.rs +++ b/crates/abyss-backend/src/search/mod.rs @@ -1,8 +1,15 @@ //! Traditional full-text session search backed by a derived Elasticsearch index. +//! +//! Search results are grouped by session in Elasticsearch, then hydrated with +//! authoritative session and device rows from PostgreSQL. The index never acts +//! as an authorization source: every query includes the authenticated owner and +//! missing PostgreSQL details cause a stale search hit to be omitted. mod document; mod elasticsearch; +/// PostgreSQL outbox leasing and hydration queries. pub mod outbox; +/// Background outbox-to-Elasticsearch projection worker. pub mod worker; use std::collections::{HashMap, HashSet}; @@ -25,22 +32,26 @@ const MAX_FILTER_CHARACTERS: usize = 256; const MAX_RESULT_WINDOW: u32 = 10_000; #[derive(Clone)] +/// Validating facade over the Elasticsearch HTTP client. pub struct SearchService { client: ElasticsearchClient, } impl SearchService { + /// Creates a search service from startup-validated settings. pub fn new(config: &SearchConfig) -> Result { Ok(Self { client: ElasticsearchClient::new(config)?, }) } + /// Returns a cloned client for the background indexer. #[must_use] pub fn client(&self) -> ElasticsearchClient { self.client.clone() } + /// Validates and executes one owner-scoped session search. pub async fn search( &self, user_id: Uuid, @@ -56,15 +67,25 @@ impl SearchService { } #[derive(Debug, Deserialize)] +/// Query-string contract for session full-text search. pub struct SessionSearchQuery { + /// Required full-text query. pub q: String, + /// Inclusive lower observation-time bound. pub from: Option>, + /// Exclusive upper observation-time bound. pub to: Option>, + /// Optional canonical Agent filter. pub agent_name: Option, + /// Optional canonical LLM provider filter. pub llm_provider: Option, + /// Optional model filter. pub llm_model: Option, + /// Optional request or response filter. pub event_type: Option, + /// One-based result page. pub page: Option, + /// Sessions returned per page. pub page_size: Option, } @@ -88,6 +109,8 @@ impl SessionSearchQuery { "page_size must be between 1 and {MAX_PAGE_SIZE}" ))); } + // Elasticsearch's from/size pagination has a bounded result window. + // Checked arithmetic makes oversized user input a validation error. let offset = page .saturating_sub(1) .checked_mul(page_size) @@ -126,25 +149,38 @@ impl SessionSearchQuery { } } +/// Normalized query safe to translate directly into Elasticsearch JSON. pub struct ValidatedSearchQuery { + /// Trimmed full-text query. pub text: String, + /// Inclusive lower observation-time bound. pub from: Option>, + /// Exclusive upper observation-time bound. pub to: Option>, + /// Canonical Agent-name filter. pub agent_name: Option, + /// Canonical provider slug filter. pub llm_provider: Option, + /// Trimmed model filter. pub llm_model: Option, + /// Canonical request or response filter. pub event_type: Option, + /// One-based result page. pub page: u32, + /// Sessions returned per page. pub page_size: u32, + /// Zero-based Elasticsearch result offset. pub offset: u32, } +/// Elasticsearch matches paired with the validated query that produced them. pub struct SearchExecution { query: ValidatedSearchQuery, page: SearchMatchPage, } impl SearchExecution { + /// Returns session keys that must be hydrated from PostgreSQL. #[must_use] pub fn session_ids(&self) -> Vec { self.page @@ -154,6 +190,10 @@ impl SearchExecution { .collect() } + /// Combines search matches with authoritative session/device details. + /// + /// Stale index entries whose session no longer exists are omitted instead + /// of returning partially authorized or incomplete data. #[must_use] pub fn hydrate( self, @@ -227,45 +267,76 @@ impl SearchExecution { } #[derive(Serialize)] +/// Paginated session search response. pub struct SessionSearchResponse { + /// Normalized full-text query. pub query: String, + /// Approximate distinct session count returned by Elasticsearch cardinality. pub total_sessions: u64, + /// One-based current page. pub page: u32, + /// Requested sessions per page. pub page_size: u32, + /// Whether the reported session count extends past this page. pub has_more: bool, + /// Hydrated matching sessions. pub items: Vec, } #[derive(Serialize)] +/// One hydrated session with its strongest matching events. pub struct SessionSearchResult { + /// Backend session primary key. pub session_pk: Uuid, + /// Agent-native session identifier. pub session_id: String, + /// Canonical Agent name. pub agent_name: String, + /// Most recently observed Agent version. pub agent_version: Option, + /// Authoritative device host name. pub host_name: String, + /// Authoritative device platform. pub platform: String, + /// Earliest event observation in the session. pub started_at: DateTime, + /// Latest event observation in the session. pub ended_at: Option>, + /// Sorted providers represented by the returned matching events. pub providers: Vec, + /// Sorted models represented by the returned matching events. pub models: Vec, + /// Total matching events in the session. pub match_count: u64, + /// Bounded strongest matching events with fragments. pub matches: Vec, } #[derive(Serialize)] +/// Search metadata and fragments for one matching usage event. pub struct SessionSearchMatch { + /// Backend event primary key. pub event_pk: Uuid, + /// Backend turn primary key. pub turn_pk: Uuid, + /// Normalized turn number. pub turn_index: i32, + /// Request or response side. pub event_type: String, + /// Canonical LLM provider slug. pub llm_provider: String, + /// Provider-specific model identifier. pub llm_model: String, + /// Collector observation time. pub observed_at: DateTime, + /// Safely segmented highlighted snippets. pub fragments: Vec, } #[derive(Serialize)] +/// One Elasticsearch highlight fragment split into plain and matching text. pub struct SearchFragment { + /// Ordered text segments suitable for structured UI rendering. pub segments: Vec, } @@ -279,6 +350,8 @@ impl SearchFragment { let highlighted = tagged .strip_prefix(HIGHLIGHT_START) .expect("the marker position came from find"); + // Treat malformed or truncated marker pairs as plain text. The API + // never emits raw HTML and therefore does not trust ES fragments. let Some(end) = highlighted.find(HIGHLIGHT_END) else { push_fragment_segment(&mut segments, tagged, false); remainder = ""; @@ -296,8 +369,11 @@ impl SearchFragment { } #[derive(Serialize)] +/// Plain or highlighted portion of a search fragment. pub struct SearchFragmentSegment { + /// Fragment text with internal marker tokens removed by parsing. pub text: String, + /// Whether this segment matched the full-text query. pub highlighted: bool, } diff --git a/crates/abyss-backend/src/search/outbox.rs b/crates/abyss-backend/src/search/outbox.rs index df46eeb..d7e2876 100644 --- a/crates/abyss-backend/src/search/outbox.rs +++ b/crates/abyss-backend/src/search/outbox.rs @@ -1,4 +1,10 @@ -//! `PostgreSQL` outbox leasing, retry transitions, and session ownership hydration. +//! PostgreSQL outbox leasing, retry transitions, and session ownership hydration. +//! +//! Event ingest and outbox insertion share one transaction, which prevents a +//! committed source event from being permanently missed by search. Workers use +//! `FOR UPDATE SKIP LOCKED` and expiring claims so replicas can process rows +//! concurrently and recover work abandoned by a crashed process. Item failures +//! are retried with bounded backoff and eventually retained as dead letters. use std::collections::HashMap; @@ -27,23 +33,33 @@ const MAX_ERROR_CHARACTERS: usize = 2_000; /// One leased outbox row paired with the Elasticsearch operation it requires. pub struct PreparedOutboxTask { + /// Durable outbox primary key. pub id: i64, + /// Number of previously recorded failures. pub attempt_count: i32, + /// Idempotent Elasticsearch operation derived from current source state. pub operation: BulkOperation, } /// Result of one Elasticsearch bulk item, used to advance durable outbox state. pub struct OutboxTaskResult { + /// Durable outbox primary key. pub id: i64, + /// Attempt count observed when the task was leased. pub attempt_count: i32, + /// Per-item Elasticsearch outcome. pub result: Result<(), String>, } +/// Authoritative PostgreSQL data used to hydrate a search result. pub struct SearchSessionDetails { + /// Session row visible to the authenticated owner. pub session: AgentSession, + /// Device row associated with the session. pub device: Device, } +/// Synchronous persistence boundary for search projection state. pub struct SearchOutboxRepository; impl SearchOutboxRepository { @@ -52,6 +68,8 @@ impl SearchOutboxRepository { connection: &mut PgConnection, batch_size: i64, ) -> Result { + // The unique (event_pk, operation) constraint makes every backfill pass + // idempotent and allows startup by multiple replicas. let queued = sql_query( "INSERT INTO search_outbox (event_pk, user_id, operation, created_at) \ SELECT source.id, source.user_id, 'upsert', source.created_at \ @@ -94,6 +112,8 @@ impl SearchOutboxRepository { worker_id: &str, batch_size: i64, ) -> Result, AppError> { + // SKIP LOCKED distributes rows across replicas without serializing all + // workers. A stale claimed_at is eligible again after the lease window. let tasks = sql_query( "WITH claimable AS (\ SELECT id \ @@ -181,6 +201,8 @@ impl SearchOutboxRepository { Err(error) => { let attempt_count = task.attempt_count.saturating_add(1); let error = truncate_error(error); + // Dead letters preserve the final bounded error for operator + // inspection without allowing a poison row to loop forever. if attempt_count >= MAX_OUTBOX_ATTEMPTS { diesel::update(search_outbox::table.find(task.id)) .set(( @@ -229,6 +251,9 @@ impl SearchOutboxRepository { if session_ids.is_empty() { return Ok(HashMap::new()); } + // This owner predicate is authoritative even though Elasticsearch also + // applies an owner filter. Defense in depth protects against stale or + // incorrectly projected documents. let rows = agent_sessions::table .inner_join(devices::table) .filter(agent_sessions::user_id.eq(user_id)) diff --git a/crates/abyss-backend/src/search/worker.rs b/crates/abyss-backend/src/search/worker.rs index 4348ac3..4d1c9ee 100644 --- a/crates/abyss-backend/src/search/worker.rs +++ b/crates/abyss-backend/src/search/worker.rs @@ -1,4 +1,9 @@ -//! Background projection worker from the `PostgreSQL` outbox into Elasticsearch. +//! Background projection worker from the PostgreSQL outbox into Elasticsearch. +//! +//! The worker performs blocking database operations on Tokio's blocking pool, +//! batches Elasticsearch writes, and persists each bulk-item result separately. +//! It polls only while idle or unhealthy; full batches are drained immediately +//! to reduce projection lag without busy-looping an empty queue. use std::time::Duration; @@ -15,9 +20,13 @@ use crate::{ }, }; +/// Factory for the detached search projection task. pub struct SearchIndexer; impl SearchIndexer { + /// Spawns one indexer with a unique lease owner identifier. + /// + /// The returned handle must be joined or aborted during service shutdown. #[must_use] pub fn spawn( pool: DbPool, @@ -47,7 +56,9 @@ struct SearchIndexerWorker { } enum PollSchedule { + /// Start another iteration without sleeping because work may remain. Immediately, + /// Wait for the configured interval or shutdown notification. AfterInterval, } @@ -81,6 +92,8 @@ impl SearchIndexerWorker { return PollSchedule::AfterInterval; } + // Historical rows are queued incrementally so enabling search on an + // existing installation does not require a separate migration job. if !*backfill_complete && let Err(error) = self.advance_backfill(backfill_complete).await { tracing::error!(%error, "queue session search backfill batch"); return PollSchedule::AfterInterval; @@ -123,6 +136,8 @@ impl SearchIndexerWorker { index_ready: &mut bool, ) -> usize { let task_count = tasks.len(); + // Keep durable state aligned by position with bulk operations. The ES + // boundary guarantees one response result for every submitted item. let (task_states, operations): (Vec<_>, Vec<_>) = tasks .into_iter() .map(|task| ((task.id, task.attempt_count), task.operation)) @@ -135,6 +150,8 @@ impl SearchIndexerWorker { results } Err(error) => { + // A request-level failure has no trustworthy per-item result; + // retry every leased task and force the index check to rerun. *index_ready = false; let message = error.to_string(); task_states diff --git a/crates/abyss-backend/src/usage/attachments.rs b/crates/abyss-backend/src/usage/attachments.rs index 38cf2e7..02750aa 100644 --- a/crates/abyss-backend/src/usage/attachments.rs +++ b/crates/abyss-backend/src/usage/attachments.rs @@ -1,4 +1,9 @@ //! Image attachment ingest contracts, validation, and download response models. +//! +//! Attachments may carry either metadata alone or inline base64 content. All +//! declared sizes, hashes, media signatures, positions, and aggregate limits are +//! checked before a database transaction begins so an invalid image cannot +//! partially persist the surrounding event batch. use std::collections::HashSet; @@ -18,17 +23,22 @@ const MAX_BASE64_IMAGE_CHARACTERS: usize = 11_184_812; /// Browser-safe raster media types accepted by the audit service. #[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)] pub enum ImageMediaType { + /// Portable Network Graphics image. #[serde(rename = "image/png")] Png, + /// JPEG image. #[serde(rename = "image/jpeg")] Jpeg, + /// WebP image. #[serde(rename = "image/webp")] Webp, + /// Graphics Interchange Format image. #[serde(rename = "image/gif")] Gif, } impl ImageMediaType { + /// Returns the canonical MIME type persisted and emitted by the API. #[must_use] pub const fn as_str(self) -> &'static str { match self { @@ -39,6 +49,10 @@ impl ImageMediaType { } } + /// Parses a media type read from a database row. + /// + /// An unknown stored value is an internal data-integrity error rather than + /// a client validation error because the migration constrains this column. pub fn from_stored(value: &str) -> Result { match value { "image/png" => Ok(Self::Png), @@ -51,6 +65,7 @@ impl ImageMediaType { } } + /// Returns the safe file extension used in attachment responses. #[must_use] pub const fn file_extension(self) -> &'static str { match self { @@ -76,38 +91,57 @@ impl ImageMediaType { /// Image attachment supplied with an Agent usage ingest event. #[derive(Debug, Deserialize)] pub struct IngestImageAttachment { + /// Zero-based display position within the event. pub position: i32, + /// Declared browser-safe raster media type. pub media_type: ImageMediaType, + /// Declared decoded byte length. pub byte_size: u64, + /// Lowercase hexadecimal SHA-256 of the decoded content. pub sha256: String, + /// Optional standard-base64 encoded image bytes. pub content_base64: Option, } /// Validated attachment ready for database persistence. #[derive(Debug)] pub struct ValidatedImageAttachment { + /// Unique non-negative display position. pub position: i32, + /// Media type verified against the file signature when content is present. pub media_type: ImageMediaType, + /// Decoded size converted to PostgreSQL's signed integer representation. pub byte_size: i64, + /// Decoded 32-byte SHA-256 digest. pub sha256: Vec, + /// Optional decoded image bytes. pub content: Option>, } /// Attachment metadata returned with usage event APIs. #[derive(Debug, Serialize)] pub struct UsageEventAttachmentResponse { + /// Backend-generated attachment primary key. pub id: Uuid, + /// Display position within the event. pub position: i32, + /// Validated image media type. pub media_type: ImageMediaType, + /// Decoded image byte length. pub byte_size: i64, + /// Lowercase hexadecimal SHA-256 digest. pub sha256: String, + /// Whether bytes can be fetched from the attachment endpoint. pub content_available: bool, } /// Authorized image bytes returned by the attachment download repository. pub struct StoredImageAttachment { + /// Validated media type used for the HTTP `Content-Type` header. pub media_type: ImageMediaType, + /// Lowercase hexadecimal digest used for the HTTP entity tag. pub sha256: String, + /// Authorized image bytes. pub content: Vec, } @@ -148,6 +182,9 @@ pub fn validate_image_attachments( } let sha256 = decode_sha256(&attachment.sha256)?; + // Metadata-only attachments intentionally retain a digest and size but + // have no downloadable body. When bytes are present, every declaration + // is verified before they enter the persistence layer. let content = attachment .content_base64 .as_deref() @@ -186,6 +223,8 @@ fn decode_content( expected_size: usize, expected_sha256: &[u8], ) -> Result, AppError> { + // Reject overlong encoded data before allocating the decoded buffer. The + // limit includes base64 expansion and a small allowance for padding. if encoded.len() > MAX_BASE64_IMAGE_CHARACTERS { return Err(AppError::validation( "image attachment content exceeds the decoded size limit".to_owned(), diff --git a/crates/abyss-backend/src/usage/diagnostics.rs b/crates/abyss-backend/src/usage/diagnostics.rs index a3e1d59..2400928 100644 --- a/crates/abyss-backend/src/usage/diagnostics.rs +++ b/crates/abyss-backend/src/usage/diagnostics.rs @@ -14,17 +14,27 @@ use serde_json::Value; use crate::error::AppError; #[derive(Deserialize)] +/// One opaque diagnostic capture correlated with usage events in the batch. pub struct IngestDiagnosticCapture { + /// Collector-generated idempotency key scoped to the owner. pub capture_id: String, + /// Time at which the diagnostic evidence was observed. pub captured_at: DateTime, + /// Collector flow identifier used to correlate related captures. pub flow_id: String, + /// Usage-event identifiers that explain the capture context. pub event_ids: Vec, + /// Version of the collector that produced the payload. pub collector_version: String, /// Agent Hook evidence stored without content-level validation. pub payload: Value, } impl IngestDiagnosticCapture { + /// Ensures every referenced event appears exactly once in the same batch. + /// + /// Database-backed validation later verifies that the persisted events also + /// share an authenticated owner, session, and device context. pub fn validate_event_correlation( &self, request_event_ids: &HashSet<&str>, diff --git a/crates/abyss-backend/src/usage/event_order.rs b/crates/abyss-backend/src/usage/event_order.rs index 535e797..4fd3e45 100644 --- a/crates/abyss-backend/src/usage/event_order.rs +++ b/crates/abyss-backend/src/usage/event_order.rs @@ -18,6 +18,10 @@ use crate::db::models::UsageEvent; pub struct UsageEventTimelineOrder; impl UsageEventTimelineOrder { + /// Sorts events by turn, provider chain, observation time, side, and ID. + /// + /// The final event identifier tie-breaker makes output deterministic even + /// for corrupt cycles or collectors with identical timestamps. pub fn sort(events: &mut [UsageEvent]) { let provider_ranks = ProviderResponseRanks::from_events(events); events.sort_by(|left, right| { @@ -44,6 +48,9 @@ struct ProviderResponseRanks { impl ProviderResponseRanks { fn from_events(events: &[UsageEvent]) -> Self { let mut nodes_by_turn = BTreeMap::>::new(); + // Native provider ordering is safe only when every event in a turn has + // a response_id. A partially instrumented turn falls back as a whole so + // ranked and unranked events cannot interleave unpredictably. let mut native_turns = events .iter() .map(|event| event.turn_index) @@ -100,6 +107,8 @@ impl ProviderResponseRanks { } } + // Branches can occur after retries or malformed evidence. Sort roots and + // siblings by stable observed/event fallback data before traversal. let compare_ids = |left: &String, right: &String| { nodes .get(left) @@ -128,6 +137,8 @@ impl ProviderResponseRanks { ); } + // Cycles have no root. Traversing remaining nodes after rooted chains + // guarantees every response still receives a deterministic rank. let mut remaining = nodes .keys() .filter(|response_id| !visited.contains(response_id.as_str())) @@ -175,6 +186,8 @@ impl ProviderResponseRanks { let left_response_id = metadata_string(&left.metadata, "response_id"); let right_response_id = metadata_string(&right.metadata, "response_id"); + // Request and response observations that describe the same provider + // call remain adjacent and request-first regardless of timestamp ties. if left_response_id.is_some() && left_response_id == right_response_id { return event_side_rank(&left.event_type).cmp(&event_side_rank(&right.event_type)); } diff --git a/crates/abyss-backend/src/usage/mod.rs b/crates/abyss-backend/src/usage/mod.rs index adeb34e..47625c7 100644 --- a/crates/abyss-backend/src/usage/mod.rs +++ b/crates/abyss-backend/src/usage/mod.rs @@ -1,8 +1,16 @@ -//! API request and response types for Agent usage collection. - +//! API contracts and shared transformations for Agent usage collection. +//! +//! Ingest types preserve collector-provided observations, while response types +//! expose the normalized device/session/turn hierarchy stored by PostgreSQL. +//! Repository code owns persistence and validation that requires database state; +//! small deterministic metadata transformations remain in this module. + +/// Image attachment contracts and validation. pub mod attachments; +/// Opaque diagnostic-capture ingest contracts. pub mod diagnostics; mod event_order; +/// PostgreSQL-backed ingest and query operations. pub mod repository; use chrono::{DateTime, Utc}; @@ -13,62 +21,91 @@ use uuid::Uuid; const WORKING_DIRECTORY_METADATA_KEY: &str = "working_directory"; #[derive(Deserialize)] +/// Atomic batch accepted by the event-ingestion endpoint. pub struct IngestEventsRequest { + /// Usage observations to validate and persist. pub events: Vec, + /// Optional captures correlated with events from this same request. #[serde(default)] pub diagnostic_captures: Vec, } #[derive(Debug, Deserialize)] +/// One collector-observed LLM request or response. pub struct IngestUsageEvent { + /// Collector-generated global idempotency key. pub event_id: String, + /// Time at which the collector observed the event. pub observed_at: DateTime, + /// Host context associated with the event. pub device: DevicePayload, + /// Agent implementation that produced the event. pub agent: AgentPayload, + /// Agent-native session identifier. pub session_id: String, /// Collector-side best-effort turn number. Ingest normalizes this when a /// stable logical-turn/provider identity shows that a restarted collector /// reused an earlier local sequence number. pub turn_index: i32, + /// LLM provider and model labels. pub llm: LlmPayload, + /// Whether the event represents the request or response side of a call. pub event_type: UsageEventType, + /// Optional prompt or response text. pub text: Option, + /// Provider-reported token counters; missing counters default to zero. #[serde(default)] pub token_usage: TokenUsagePayload, + /// Collector metadata retained with the raw event. #[serde(default = "empty_metadata")] pub metadata: Value, + /// Optional image metadata and content ordered within the event. #[serde(default)] pub attachments: Vec, } #[derive(Debug, Deserialize)] +/// Collector-supplied host identity used to group events by device context. pub struct DevicePayload { + /// User-visible host name; `hostname` remains an accepted JSON alias. #[serde(alias = "hostname")] pub host_name: String, + /// Normalized operating-system or runtime platform label. pub platform: String, + /// Optional operating-system version observed by the collector. pub os_version: Option, } #[derive(Debug, Deserialize)] +/// Collector-supplied Agent identity. pub struct AgentPayload { + /// Agent name, normalized to a canonical slug during ingest. pub name: String, + /// Optional Agent build or release version. pub version: Option, } #[derive(Debug, Deserialize)] +/// Collector-supplied LLM target identity. pub struct LlmPayload { + /// Provider name, normalized to a lowercase slug during ingest. pub provider: String, + /// Provider-specific model identifier. pub model: String, } #[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)] #[serde(rename_all = "lowercase")] +/// Side of the LLM exchange represented by an event. pub enum UsageEventType { + /// Prompt or tool request sent to the provider. Request, + /// Provider response returned to the Agent. Response, } impl UsageEventType { + /// Returns the canonical value stored in PostgreSQL and returned by APIs. pub const fn as_str(self) -> &'static str { match self { Self::Request => "request", @@ -83,101 +120,164 @@ impl UsageEventType { reason = "Public JSON token usage fields intentionally include the token suffix." )] pub struct TokenUsagePayload { + /// Input tokens reported by the provider. pub input_tokens: Option, + /// Output tokens reported by the provider. pub output_tokens: Option, + /// Tokens served from a provider cache. pub cache_read_tokens: Option, + /// Tokens written to a provider cache. pub cache_write_tokens: Option, + /// Provider-reported reasoning tokens. pub reasoning_tokens: Option, + /// Provider total, or a derived sum when omitted. pub total_tokens: Option, } #[derive(Debug, Serialize)] +/// Counts returned after an atomic ingest batch finishes. pub struct IngestEventsResponse { + /// Newly persisted usage events. pub accepted: usize, + /// Events skipped because their `event_id` already exists. pub duplicates: usize, + /// Per-event rejection count; currently always zero because batches are atomic. pub rejected: usize, + /// Per-event errors; currently empty because validation fails the whole request. pub errors: Vec, + /// Newly persisted diagnostic captures. pub accepted_diagnostic_captures: usize, + /// Captures skipped because their `(user_id, capture_id)` already exists. pub duplicate_diagnostic_captures: usize, } #[derive(Clone, Debug, Deserialize)] +/// Filters, dimensions, and bounds for token-usage aggregation. pub struct SummaryQuery { + /// Inclusive lower observation-time bound. pub from: Option>, + /// Exclusive upper observation-time bound. pub to: Option>, + /// Comma-separated grouping dimensions. pub group_by: Option, + /// Selects the full or token-only response shape. pub fields: Option, + /// User UUID, name, or email filter retained for the summary contract. pub user_filter: Option, + /// Agent-name filter. pub agent_name: Option, + /// Agent-native session identifier filter. pub session_id: Option, + /// LLM provider filter. pub llm_provider: Option, + /// LLM model filter. pub llm_model: Option, + /// Event-side filter. pub event_type: Option, + /// Requested aggregate row limit. pub limit: Option, } #[derive(Clone, Copy, Debug, Deserialize, PartialEq)] #[serde(rename_all = "snake_case")] +/// Response projection requested from the summary endpoint. pub enum SummaryFields { + /// Return dimensions, event counts, and every token counter. Full, + /// Return only day, input-token, and total-token fields. TokenUsage, } #[derive(Debug, Serialize)] +/// Full summary response grouped by the requested dimensions. pub struct SummaryResponse { + /// Inclusive lower bound echoed from the request. pub from: Option>, + /// Exclusive upper bound echoed from the request. pub to: Option>, + /// Normalized grouping dimensions actually applied. pub group_by: Vec, + /// Aggregate rows ordered by token usage and stable tie-breakers. pub rows: Vec, + /// Reserved cursor field; summary queries are currently single-page. pub next_page_token: Option, } #[derive(Debug, Serialize)] +/// One aggregate bucket from the full usage summary. pub struct SummaryRow { + /// UTC calendar day when grouped by day. #[serde(skip_serializing_if = "Option::is_none")] pub day: Option, + /// Owner UUID when grouped by user. #[serde(skip_serializing_if = "Option::is_none")] pub user_id: Option, + /// Owner display name when grouped by user. #[serde(skip_serializing_if = "Option::is_none")] pub user_name: Option, + /// Owner email when grouped by user. #[serde(skip_serializing_if = "Option::is_none")] pub user_email: Option, + /// Device host name when grouped by device. #[serde(skip_serializing_if = "Option::is_none")] pub host_name: Option, + /// Device platform when grouped by device. #[serde(skip_serializing_if = "Option::is_none")] pub platform: Option, + /// Device OS version when grouped by device. #[serde(skip_serializing_if = "Option::is_none")] pub os_version: Option, + /// Canonical Agent name when grouped by Agent. #[serde(skip_serializing_if = "Option::is_none")] pub agent_name: Option, + /// LLM provider when grouped by provider. #[serde(skip_serializing_if = "Option::is_none")] pub llm_provider: Option, + /// LLM model when grouped by model. #[serde(skip_serializing_if = "Option::is_none")] pub llm_model: Option, + /// Request or response value when grouped by event type. #[serde(skip_serializing_if = "Option::is_none")] pub event_type: Option, + /// Distinct sessions in this bucket. pub sessions: usize, + /// Distinct turns in this bucket. pub turns: usize, + /// Request events in this bucket. pub requests: i64, + /// Response events in this bucket. pub responses: i64, + /// Summed input tokens. pub input_tokens: i64, + /// Summed output tokens. pub output_tokens: i64, + /// Summed cache-read tokens. pub cache_read_tokens: i64, + /// Summed cache-write tokens. pub cache_write_tokens: i64, + /// Summed reasoning tokens. pub reasoning_tokens: i64, + /// Summed provider or derived total tokens. pub total_tokens: i64, } #[derive(Debug, Serialize)] +/// Reduced summary response used by token-usage-only consumers. pub struct TokenUsageSummaryResponse { + /// Inclusive lower bound echoed from the request. pub from: Option>, + /// Exclusive upper bound echoed from the request. pub to: Option>, + /// Normalized grouping dimensions actually applied. pub group_by: Vec, + /// Token-only aggregate rows. pub rows: Vec, + /// Reserved cursor field; summary queries are currently single-page. pub next_page_token: Option, } impl TokenUsageSummaryResponse { + /// Projects a full summary into its stable token-only representation. pub fn from_summary(summary: SummaryResponse) -> Self { Self { from: summary.from, @@ -194,10 +294,14 @@ impl TokenUsageSummaryResponse { } #[derive(Debug, Serialize)] +/// Minimal token counters for one summary bucket. pub struct TokenUsageSummaryRow { + /// UTC calendar day when day grouping was requested. #[serde(skip_serializing_if = "Option::is_none")] pub day: Option, + /// Summed input tokens. pub input_tokens: i64, + /// Summed provider or derived total tokens. pub total_tokens: i64, } @@ -212,74 +316,129 @@ impl From for TokenUsageSummaryRow { } #[derive(Clone, Debug, Deserialize)] +/// Filters and offset pagination for the raw-event endpoint. pub struct RawEventsQuery { + /// Inclusive lower observation-time bound. pub from: Option>, + /// Exclusive upper observation-time bound. pub to: Option>, + /// Agent-name filter. pub agent_name: Option, + /// Agent-native session identifier filter. pub session_id: Option, + /// Backend session primary-key filter. pub session_pk: Option, + /// Normalized turn-number filter. pub turn_index: Option, + /// LLM provider filter. pub llm_provider: Option, + /// LLM model filter. pub llm_model: Option, + /// Request or response filter. pub event_type: Option, + /// Requested page size. pub limit: Option, + /// Number of newest matching events to skip. pub offset: Option, } #[derive(Debug, Serialize)] +/// One page of raw usage events. pub struct RawEventsResponse { + /// Events ordered from newest to oldest observation time. pub events: Vec, + /// Decimal offset for the next page, or `None` at the end. pub next_page_token: Option, } #[derive(Debug, Serialize)] +/// API representation of one persisted usage event. pub struct UsageEventResponse { + /// Backend-generated event primary key. pub id: Uuid, + /// Collector-generated idempotency key. pub event_id: String, + /// Authenticated owner identifier. pub user_id: Uuid, + /// Backend device-context primary key. pub device_context_id: Uuid, + /// Device host name, when the referenced device still exists. #[serde(skip_serializing_if = "Option::is_none")] pub host_name: Option, + /// Device platform, when the referenced device still exists. #[serde(skip_serializing_if = "Option::is_none")] pub platform: Option, + /// Backend session primary key. pub session_pk: Uuid, + /// Backend turn primary key. pub turn_pk: Uuid, + /// Canonical Agent name. pub agent_name: String, + /// Agent version observed on this event. pub agent_version: Option, + /// Agent-native session identifier. pub session_id: String, + /// Backend-normalized turn number. pub turn_index: i32, + /// Canonical LLM provider slug. pub llm_provider: String, + /// Provider-specific model identifier. pub llm_model: String, + /// Canonical request or response value. pub event_type: String, + /// Optional prompt or response text. pub text: Option, + /// Lowercase SHA-256 of text when text is present. pub text_sha256: Option, + /// Normalized input-token count. pub input_tokens: i64, + /// Normalized output-token count. pub output_tokens: i64, + /// Normalized cache-read-token count. pub cache_read_tokens: i64, + /// Normalized cache-write-token count. pub cache_write_tokens: i64, + /// Normalized reasoning-token count. pub reasoning_tokens: i64, + /// Provider-reported or derived total-token count. pub total_tokens: i64, + /// Collector observation timestamp. pub observed_at: DateTime, + /// Collector metadata retained without schema expansion. pub metadata: Value, + /// Ordered attachment metadata associated with the event. pub attachments: Vec, } #[derive(Debug, Serialize)] +/// Session metadata and its ordered turn timelines. pub struct SessionTimelineResponse { + /// Authoritative session information. pub session: SessionInfo, + /// Turns ordered by normalized turn index. pub turns: Vec, } #[derive(Debug, Serialize)] +/// API representation of an Agent session. pub struct SessionInfo { + /// Backend session primary key. pub session_pk: Uuid, + /// Authenticated owner identifier. pub user_id: Uuid, + /// Most recently associated device context. pub device_context_id: Uuid, + /// Canonical Agent name. pub agent_name: String, + /// Most recently known Agent version. pub agent_version: Option, + /// Agent-native session identifier. pub session_id: String, + /// Earliest event observation time in the session. pub started_at: DateTime, + /// Latest event observation time in the session. pub ended_at: Option>, + /// Stable, allowlisted session metadata. pub metadata: Value, } @@ -303,14 +462,21 @@ pub fn session_metadata_from_event(metadata: &Value) -> Value { } #[derive(Debug, Serialize)] +/// One normalized turn and its deterministically ordered events. pub struct TurnTimeline { + /// Backend turn primary key. pub turn_pk: Uuid, + /// Backend-normalized turn number. pub turn_index: i32, + /// Earliest event observation time in the turn. pub started_at: DateTime, + /// Latest event observation time in the turn. pub ended_at: Option>, + /// Request and response events in authoritative timeline order. pub events: Vec, } +/// Returns the JSON object used when optional metadata is absent. pub fn empty_metadata() -> Value { json!({}) } diff --git a/crates/abyss-backend/src/usage/repository.rs b/crates/abyss-backend/src/usage/repository.rs index 5b038c9..3285cf2 100644 --- a/crates/abyss-backend/src/usage/repository.rs +++ b/crates/abyss-backend/src/usage/repository.rs @@ -1,4 +1,10 @@ //! Diesel-backed repository functions for Agent usage APIs. +//! +//! This module is the transactional boundary for the event hierarchy. Ingest +//! validates the complete request before opening a transaction, upserts device, +//! session, and turn aggregates, inserts immutable events idempotently, and +//! enqueues search projection in the same commit. Query functions always apply +//! the authenticated owner before returning conversation or attachment data. use std::collections::{HashMap, HashSet}; @@ -96,6 +102,10 @@ const SUMMARY_AGGREGATE_SQL: &str = "\ ORDER BY total_tokens DESC, agent_name ASC NULLS LAST, user_email ASC NULLS LAST, day ASC NULLS LAST \ LIMIT $31"; +/// Validates and atomically ingests one batch for the authenticated owner. +/// +/// Collector `event_id` and `(user_id, capture_id)` values are idempotency keys; +/// replays are counted as duplicates without changing the original rows. pub fn ingest_events( connection: &mut PgConnection, request: &IngestEventsRequest, @@ -153,6 +163,8 @@ fn ingest_one_diagnostic_capture( user_id: Uuid, capture: &IngestDiagnosticCapture, ) -> Result { + // Reload event rows rather than trusting request correlation alone. This + // establishes authenticated ownership and a single session/device boundary. let events = llm_usage_events::table .filter(llm_usage_events::user_id.eq(user_id)) .filter(llm_usage_events::event_id.eq_any(&capture.event_ids)) @@ -212,6 +224,7 @@ fn ingest_one_diagnostic_capture( Ok(inserted == 1) } +/// Aggregates owner-scoped event and token counts by requested dimensions. pub fn usage_summary( connection: &mut PgConnection, query: &SummaryQuery, @@ -280,6 +293,9 @@ fn load_filtered_summary_rows( let event_type = sql_filters.event_type.unwrap_or_default(); let has_event_type_filter = !event_type.is_empty(); + // The static query uses boolean gates for optional dimensions and filters. + // This keeps user values in typed bind parameters and avoids dynamic SQL. + // Bind order intentionally mirrors the numbered placeholders in the query. sql_query(SUMMARY_AGGREGATE_SQL) .bind::(dimensions.enabled(SummaryDimension::Day)) .bind::(dimensions.enabled(SummaryDimension::User)) @@ -317,6 +333,7 @@ fn load_filtered_summary_rows( .map_err(AppError::from) } +/// Loads one owner-scoped session with normalized turns and ordered events. pub fn session_timeline( connection: &mut PgConnection, user_id: Uuid, @@ -392,6 +409,7 @@ pub fn session_timeline( }) } +/// Returns one newest-first page of raw owner-scoped events. pub fn raw_events( connection: &mut PgConnection, query: &RawEventsQuery, @@ -422,6 +440,8 @@ pub fn raw_events( offset, }, )?; + // Fetch one sentinel row beyond the requested page so no count query is + // needed merely to determine whether a next offset exists. let has_next_page = events.len() > page_size_usize; if has_next_page { events.truncate(page_size_usize); @@ -446,6 +466,7 @@ pub fn raw_events( }) } +/// Loads authorized attachment bytes and verifies persisted size metadata. pub fn image_attachment( connection: &mut PgConnection, user_id: Uuid, @@ -470,6 +491,9 @@ pub fn image_attachment( ))); } let media_type = ImageMediaType::from_stored(&attachment.media_type)?; + // These columns are selected by the reusable Diesel row model but are not + // part of the download response. Touch them to keep dead-code linting useful + // without creating a second, nearly identical query model. std::hint::black_box(( attachment.user_id, attachment.event_pk, @@ -554,6 +578,8 @@ fn ingest_one_event( .do_nothing() .execute(connection)?; + // Child rows are written only for a newly inserted event. A replay must not + // replace the attachments associated with the original idempotency key. if inserted == 1 && !attachments.is_empty() { let attachment_rows = attachments .into_iter() @@ -574,6 +600,8 @@ fn ingest_one_event( .execute(connection)?; } + // The enclosing transaction commits the event and its projection task + // together, providing at-least-once delivery to Elasticsearch. if inserted == 1 { diesel::insert_into(search_outbox::table) .values(NewSearchOutboxTask { @@ -677,6 +705,9 @@ fn resolve_turn_index( session_pk: Uuid, event: &IngestUsageEvent, ) -> Result { + // Older collectors without a stable provider identity retain their supplied + // index. Newer evidence lets restarted collectors avoid merging new work + // into an already-used local turn number. let Some(identity) = turn_identity(event) else { return Ok(event.turn_index); }; @@ -768,9 +799,13 @@ const fn choose_turn_index( existing_turn_index: Option, next_turn_index: i32, ) -> i32 { + // Events with the same stable provider identity must remain in one turn, + // even if collector-local numbering changes between observations. if let Some(existing_turn_index) = existing_turn_index { return existing_turn_index; } + // A previously used local number indicates a restarted counter. Append at + // the next index; otherwise preserve intentional gaps from the collector. if requested_turn_index < next_turn_index { next_turn_index } else { @@ -783,6 +818,8 @@ fn turn_identity(event: &IngestUsageEvent) -> Option { } fn turn_identity_from_metadata(metadata: &serde_json::Value) -> Option { + // Prefer Agent-level turn identities over provider message/response ids + // because a single user turn may contain several provider round trips. metadata_string(metadata, "codex_turn_id") .map(|value| TurnIdentity { kind: TurnIdentityKind::CodexTurnId, @@ -951,6 +988,8 @@ fn load_attachments_for_events( let mut by_event = HashMap::new(); for attachment in attachments { let event_pk = attachment.event_pk; + // The shared row model includes authorization/audit columns that are not + // repeated in each event response attachment object. std::hint::black_box((attachment.user_id, attachment.created_at)); by_event .entry(event_pk) @@ -972,6 +1011,8 @@ fn event_response( device: Option<&Device>, attachments: Vec, ) -> UsageEventResponse { + // created_at is an internal ingestion timestamp; APIs expose observed_at as + // the user-facing event time while still selecting a complete Diesel row. std::hint::black_box(event.created_at); UsageEventResponse { id: event.id, @@ -1039,6 +1080,8 @@ fn validate_batch( ))); } + // Decode and validate all potentially expensive attachment content before + // obtaining a connection transaction and any table locks. let mut attachments = Vec::with_capacity(request.events.len()); for event in &request.events { validate_event(event)?; @@ -1107,6 +1150,8 @@ fn normalized_tokens(event: &IngestUsageEvent) -> Result normalize_token(Some(value), "total_tokens")?, None => input @@ -1145,6 +1190,8 @@ fn validate_event_type(value: &str) -> Result { } fn parse_group_by(raw: Option<&str>) -> Vec { + // Unknown dimensions are ignored and historical aliases remain readable. + // Falling back after filtering prevents an empty GROUP BY contract. let parsed: Vec<_> = raw .unwrap_or("user,agent,provider,model") .split(',') @@ -1233,6 +1280,8 @@ const fn unix_epoch() -> DateTime { } fn escape_like_pattern(value: &str) -> String { + // The SQL uses LIKE ... ESCAPE '\\'; escape wildcard characters so a user + // filter remains a literal substring search rather than a pattern language. let mut escaped = String::with_capacity(value.len()); for character in value.chars() { if matches!(character, '\\' | '%' | '_') {