diff --git a/docs/README.md b/docs/README.md index 6dad911..161ed3a 100644 --- a/docs/README.md +++ b/docs/README.md @@ -30,3 +30,4 @@ Building an AppView from scratch means wiring up firehose connections, record st - [Quickstart](getting-started/deployment/railway.md): Deploy HappyView on Railway or run it locally - [Lexicons](guides/lexicons.md): Upload lexicon schemas and start indexing records - [Lua Scripting](guides/scripting.md): Write custom query and procedure logic +- [Event Logs](guides/event-logs.md): Monitor system activity, debug script errors, and audit admin actions diff --git a/docs/getting-started/configuration.md b/docs/getting-started/configuration.md index 8364cbc..78d4189 100644 --- a/docs/getting-started/configuration.md +++ b/docs/getting-started/configuration.md @@ -14,6 +14,7 @@ HappyView is configured via environment variables. A `.env` file in the project | `TAP_ADMIN_PASSWORD` | no | --- | Shared secret for authenticating with Tap's admin endpoints | | `RELAY_URL` | no | `https://bsky.network` | Relay URL for [backfill](../guides/backfill.md) repo discovery | | `PLC_URL` | no | `https://plc.directory` | [PLC directory](https://github.com/did-method-plc/did-method-plc) URL for DID resolution | +| `EVENT_LOG_RETENTION_DAYS` | no | `30` | Number of days to keep event logs before automatic cleanup. Set to `0` to disable cleanup | | `RUST_LOG` | no | `happyview=debug,tower_http=debug` | Log filter (uses `tracing_subscriber::EnvFilter`) | ## Example `.env` @@ -29,5 +30,6 @@ AIP_URL=http://localhost:8080 # TAP_ADMIN_PASSWORD=your-secret-here # RELAY_URL=https://bsky.network # PLC_URL=https://plc.directory +# EVENT_LOG_RETENTION_DAYS=30 # RUST_LOG=happyview=debug,tower_http=debug ``` diff --git a/docs/guides/event-logs.md b/docs/guides/event-logs.md new file mode 100644 index 0000000..8a88910 --- /dev/null +++ b/docs/guides/event-logs.md @@ -0,0 +1,96 @@ +# Event Logs + +HappyView maintains an internal event log that records system activity — lexicon changes, record operations, Lua script executions and errors, admin actions, backfill jobs, and Tap connectivity. Events are stored in a Postgres table and queryable via the [admin API](../reference/admin-api.md#event-logs). + +## Event types + +Events follow a `category.action` naming convention. Each event has a severity level (`info`, `warn`, or `error`), an optional `actor_did` (the user who triggered it), an optional `subject` (what was affected), and a `detail` JSON object with event-specific data. + +### Lexicon events + +| Event Type | Severity | Subject | Detail | +|---|---|---|---| +| `lexicon.created` | info | Lexicon NSID | `revision`, `has_script`, `source` | +| `lexicon.updated` | info | Lexicon NSID | `revision`, `has_script`, `source` | +| `lexicon.deleted` | info | Lexicon NSID | — | + +Logged when lexicons are uploaded, updated, or deleted via the [admin API](../reference/admin-api.md#lexicons). The `actor_did` is the admin who performed the action. + +### Record events + +| Event Type | Severity | Subject | Detail | +|---|---|---|---| +| `record.created` | info | Record AT URI | `collection`, `did`, `rkey` | +| `record.deleted` | info | Record AT URI | `collection`, `did`, `rkey` | + +Logged when records are received from Tap and stored or removed from the local database. These are system-triggered events (`actor_did` is null). If a database error occurs during the operation, the same event type is logged with `error` severity and the error message is included in the detail. + +### Script events + +| Event Type | Severity | Subject | Detail | +|---|---|---|---| +| `script.executed` | info | Method NSID | `method`, `caller_did`, `duration_ms` | +| `script.error` | error | Method NSID | `error`, `script_source`, `input`, `caller_did`, `method` | + +Logged when Lua scripts run for XRPC query or procedure endpoints. Script errors capture the full context needed to reproduce and debug the issue: the error message, the complete Lua script source, the input that triggered it, and the caller's DID. + +:::note +For query scripts (unauthenticated), `caller_did` and `input` are omitted from the detail since queries don't have an authenticated user or request body. +::: + +### Admin events + +| Event Type | Severity | Subject | Detail | +|---|---|---|---| +| `admin.created` | info | New admin DID | — | +| `admin.deleted` | info | Removed admin ID | — | +| `admin.bootstrapped` | info | Bootstrapped admin DID | — | + +The `admin.bootstrapped` event is logged when the first user is auto-promoted to admin (see [Auth - Auto-bootstrap](../reference/admin-api.md#auth)). + +### Backfill events + +| Event Type | Severity | Subject | Detail | +|---|---|---|---| +| `backfill.started` | info | Collection NSID | `job_id` | +| `backfill.completed` | info | Collection NSID | `job_id`, `total_repos` | +| `backfill.failed` | error | Collection NSID | `job_id`, `error` | + +See [Backfill](backfill.md) for background on backfill jobs. + +### Tap events + +| Event Type | Severity | Subject | Detail | +|---|---|---|---| +| `tap.connected` | info | — | `url` | +| `tap.disconnected` | warn | — | `reason` | + +Logged when the WebSocket connection to [Tap](https://github.com/bluesky-social/indigo/tree/main/cmd/tap) is established or lost. + +## Querying events + +Use the admin API to query event logs with filters: + +```sh +# Get all errors +curl "http://localhost:3000/admin/events?severity=error" -H "$AUTH" + +# Get script errors for a specific lexicon +curl "http://localhost:3000/admin/events?event_type=script.error&subject=com.example.feed.like" -H "$AUTH" + +# Get all lexicon-related events +curl "http://localhost:3000/admin/events?category=lexicon" -H "$AUTH" + +# Paginate through results +curl "http://localhost:3000/admin/events?limit=20&cursor=2026-03-01T11:59:00Z" -H "$AUTH" +``` + +See the [Admin API reference](../reference/admin-api.md#list-event-logs) for full parameter documentation. + +## Retention + +Event logs are automatically cleaned up based on the `EVENT_LOG_RETENTION_DAYS` environment variable (default: 30 days). A background task runs hourly to delete events older than the configured retention period. + +Set `EVENT_LOG_RETENTION_DAYS=0` to disable automatic cleanup and keep logs indefinitely. + +See [Configuration](../getting-started/configuration.md) for all environment variables. diff --git a/docs/reference/admin-api.md b/docs/reference/admin-api.md index d4b9443..81ea360 100644 --- a/docs/reference/admin-api.md +++ b/docs/reference/admin-api.md @@ -301,6 +301,56 @@ curl http://localhost:3000/admin/backfill/status -H "$AUTH" ] ``` +## Event Logs + +HappyView records an audit trail of system events: lexicon changes, record operations, Lua script executions and errors, admin actions, backfill jobs, and Tap connectivity. See the [Event Logs guide](../guides/event-logs.md) for details on event types and retention. + +### List event logs + +``` +GET /admin/events +``` + +```sh +curl "http://localhost:3000/admin/events?severity=error&limit=10" -H "$AUTH" +``` + +| Param | Type | Required | Description | +| ------------ | ------ | -------- | ----------------------------------------------------------------- | +| `event_type` | string | no | Filter by exact event type (e.g. `script.error`) | +| `category` | string | no | Filter by category prefix (e.g. `lexicon` matches all lexicon events) | +| `severity` | string | no | Filter by severity: `info`, `warn`, or `error` | +| `subject` | string | no | Filter by subject (lexicon ID, record URI, admin DID, etc.) | +| `cursor` | string | no | Pagination cursor (ISO 8601 timestamp from previous response) | +| `limit` | number | no | Results per page (default `50`, max `100`) | + +**Response**: `200 OK` + +```json +{ + "events": [ + { + "id": "550e8400-e29b-41d4-a716-446655440000", + "event_type": "script.error", + "severity": "error", + "actor_did": "did:plc:abc123", + "subject": "com.example.feed.like", + "detail": { + "error": "attempt to index nil value", + "script_source": "function handle() ... end", + "input": { "status": "hello" }, + "caller_did": "did:plc:abc123", + "method": "com.example.feed.like" + }, + "created_at": "2026-03-01T12:00:00Z" + } + ], + "cursor": "2026-03-01T11:59:00Z" +} +``` + +Events are returned in reverse chronological order (newest first). Pass the `cursor` value from the response to fetch the next page. + ## Admin management ### Add an admin diff --git a/docs/reference/changelog.md b/docs/reference/changelog.md new file mode 100644 index 0000000..c0b130e --- /dev/null +++ b/docs/reference/changelog.md @@ -0,0 +1,52 @@ +# Changelog + +## v1.9.0 — Event Logs + +- **Event logging** — system-wide audit trail for lexicon changes, record operations, Lua script executions/errors, admin actions, backfill jobs, and Tap connectivity +- **`GET /admin/events`** — query event logs with filtering by event type, category, severity, and subject, with cursor pagination +- **Lua error context** — script errors capture full debugging context: error message, script source, input payload, and caller DID +- **Automatic retention cleanup** — configurable via `EVENT_LOG_RETENTION_DAYS` (default 30 days) + +## v1.8.0 — Advanced Queries + +- **`db.backlinks()`** — find records that reference a given AT URI +- **`db.raw()`** — run raw read-only SQL with parameterized queries and automatic column type mapping + +## v1.7.1 — Patch + +- Fixed Docker Compose database URLs for local dev + +## v1.7.0 — Lua DB API Improvements + +- **`toarray()`** utility — force Lua tables to serialize as JSON arrays (fixes empty `{}` vs `[]`) +- **`db.search()`** — text search on record fields with relevance ranking +- **Array serialization fix** — `db.query()` and `db.search()` now always return proper arrays for `records` + +## v1.6.2 — Patch + +- Fixed auth: use original auth scheme instead of hardcoded DPoP + +## v1.6.1 — Patch + +- Fixed broken dynamic page routes + +## v1.6.0 — Record Management + +- **Delete records** from the dashboard and API (individual and bulk collection deletion) +- **"View Records" buttons** on lexicon pages +- Bug fixes: backfill now loads previously deleted records, empty collections shown in dropdown + +## v1.5.1 — Patch + +- Removed backfill toggle from query/procedure lexicons (only applies to record lexicons) + +## v1.5.0 — Lua Scripting & Dashboard Overhaul + +- **Lua scripting** — attach custom Lua scripts to query and procedure lexicons +- **Docusaurus docs site** with GitHub Pages deploy +- **Dark mode** for the dashboard +- **Records table** reworked with dynamic columns, column visibility, and better scrolling +- **Backfill stats tracking** +- **Network and local lexicons merged** into a unified view +- **Shiki code highlighting** in the dashboard +- Bug fixes: rogue record storage, collection dropdown, dynamic page builds diff --git a/sidebars.ts b/sidebars.ts index e9158b0..bded7cb 100644 --- a/sidebars.ts +++ b/sidebars.ts @@ -84,6 +84,11 @@ const sidebars: SidebarsConfig = { id: "guides/backfill", label: "Backfill", }, + { + type: "doc", + id: "guides/event-logs", + label: "Event Logs", + }, ], }, { diff --git a/src/admin/admins.rs b/src/admin/admins.rs index d4384b7..8b9bbf3 100644 --- a/src/admin/admins.rs +++ b/src/admin/admins.rs @@ -5,6 +5,7 @@ use serde_json::Value; use crate::AppState; use crate::error::AppError; +use crate::event_log::{EventLog, Severity, log_event}; use super::auth::AdminAuth; use super::types::{AdminSummary, CreateAdminBody}; @@ -12,7 +13,7 @@ use super::types::{AdminSummary, CreateAdminBody}; /// POST /admin/admins — add a new admin by DID. pub(super) async fn create_admin( State(state): State, - _admin: AdminAuth, + auth: AdminAuth, Json(body): Json, ) -> Result<(StatusCode, Json), AppError> { let row: (String,) = sqlx::query_as("INSERT INTO admins (did) VALUES ($1) RETURNING id::text") @@ -21,6 +22,18 @@ pub(super) async fn create_admin( .await .map_err(|e| AppError::Internal(format!("failed to create admin: {e}")))?; + log_event( + &state.db, + EventLog { + event_type: "admin.created".to_string(), + severity: Severity::Info, + actor_did: Some(auth.did.clone()), + subject: Some(body.did.clone()), + detail: serde_json::json!({}), + }, + ) + .await; + Ok(( StatusCode::CREATED, Json(serde_json::json!({ @@ -64,7 +77,7 @@ pub(super) async fn list_admins( /// DELETE /admin/admins/:id — remove an admin. pub(super) async fn delete_admin( State(state): State, - _admin: AdminAuth, + auth: AdminAuth, Path(id): Path, ) -> Result { let result = sqlx::query("DELETE FROM admins WHERE id::text = $1") @@ -77,5 +90,17 @@ pub(super) async fn delete_admin( return Err(AppError::NotFound(format!("admin '{id}' not found"))); } + log_event( + &state.db, + EventLog { + event_type: "admin.deleted".to_string(), + severity: Severity::Info, + actor_did: Some(auth.did.clone()), + subject: Some(id.to_string()), + detail: serde_json::json!({}), + }, + ) + .await; + Ok(StatusCode::NO_CONTENT) } diff --git a/src/admin/auth.rs b/src/admin/auth.rs index dd159bb..bcbd89f 100644 --- a/src/admin/auth.rs +++ b/src/admin/auth.rs @@ -4,12 +4,15 @@ use axum::http::request::Parts; use crate::AppState; use crate::auth::middleware::Claims; use crate::error::AppError; +use crate::event_log::{EventLog, Severity, log_event}; /// Axum extractor for admin auth. Validates the Bearer token via AIP OAuth /// (same as `Claims`), then checks if the returned DID exists in the `admins` /// table. If no admins exist yet, the first authenticated user is /// auto-bootstrapped as the initial admin. -pub struct AdminAuth; +pub struct AdminAuth { + pub did: String, +} impl FromRequestParts for AdminAuth { type Rejection = AppError; @@ -37,6 +40,18 @@ impl FromRequestParts for AdminAuth { .map_err(|e| AppError::Internal(format!("auto-bootstrap admin failed: {e}")))?; tracing::info!(did = %did, "auto-bootstrapped first admin"); + + log_event( + &state.db, + EventLog { + event_type: "admin.bootstrapped".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(did.clone()), + detail: serde_json::json!({}), + }, + ) + .await; } // Look up the DID in the admins table. @@ -59,6 +74,6 @@ impl FromRequestParts for AdminAuth { .await; }); - Ok(AdminAuth) + Ok(AdminAuth { did }) } } diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index 8781528..529300e 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -6,6 +6,7 @@ use serde_json::Value; use crate::AppState; use crate::error::AppError; +use crate::event_log::{EventLog, Severity, log_event}; use crate::tap; use super::auth::AdminAuth; @@ -81,7 +82,7 @@ async fn list_repos_by_collection( /// POST /admin/backfill — create a backfill job, discover repos, and add them to Tap. pub(super) async fn create_backfill( State(state): State, - _admin: AdminAuth, + admin: AdminAuth, Json(body): Json, ) -> Result<(StatusCode, Json), AppError> { // Create a backfill_jobs record for tracking/audit. @@ -104,6 +105,20 @@ pub(super) async fn create_backfill( .execute(&state.db) .await; + log_event( + &state.db, + EventLog { + event_type: "backfill.started".to_string(), + severity: Severity::Info, + actor_did: Some(admin.did.clone()), + subject: body.collection.clone(), + detail: serde_json::json!({ + "job_id": job_id.clone(), + }), + }, + ) + .await; + // Determine target collections. let collections: Vec = if let Some(ref col) = body.collection { vec![col.clone()] @@ -202,6 +217,21 @@ pub(super) async fn create_backfill( .execute(&state.db) .await; + log_event( + &state.db, + EventLog { + event_type: "backfill.failed".to_string(), + severity: Severity::Error, + actor_did: None, + subject: body.collection.clone(), + detail: serde_json::json!({ + "job_id": job_id.clone(), + "error": e, + }), + }, + ) + .await; + return Ok(( StatusCode::CREATED, Json(serde_json::json!({ @@ -223,6 +253,21 @@ pub(super) async fn create_backfill( .execute(&state.db) .await; + log_event( + &state.db, + EventLog { + event_type: "backfill.completed".to_string(), + severity: Severity::Info, + actor_did: None, + subject: body.collection.clone(), + detail: serde_json::json!({ + "job_id": job_id.clone(), + "total_repos": total_repos, + }), + }, + ) + .await; + Ok(( StatusCode::CREATED, Json(serde_json::json!({ diff --git a/src/admin/events.rs b/src/admin/events.rs new file mode 100644 index 0000000..33f3008 --- /dev/null +++ b/src/admin/events.rs @@ -0,0 +1,133 @@ +use axum::{ + Json, + extract::{Query, State}, +}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use super::auth::AdminAuth; +use crate::AppState; +use crate::error::AppError; + +#[derive(Deserialize)] +pub struct EventsQuery { + pub event_type: Option, + pub category: Option, + pub severity: Option, + pub subject: Option, + pub cursor: Option, + pub limit: Option, +} + +#[derive(Serialize)] +pub struct EventResponse { + pub id: String, + pub event_type: String, + pub severity: String, + pub actor_did: Option, + pub subject: Option, + pub detail: Value, + pub created_at: chrono::DateTime, +} + +#[derive(Serialize)] +pub struct EventsListResponse { + pub events: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + pub cursor: Option, +} + +/// GET /admin/events — list event logs with optional filters and pagination. +pub(super) async fn list_events( + _auth: AdminAuth, + State(state): State, + Query(query): Query, +) -> Result, AppError> { + let limit = query.limit.unwrap_or(50).clamp(1, 100); + + let mut sql = String::from( + "SELECT id::text, event_type, severity, actor_did, subject, detail, created_at + FROM event_logs WHERE 1=1", + ); + let mut param_count = 0u32; + + if query.event_type.is_some() { + param_count += 1; + sql.push_str(&format!(" AND event_type = ${param_count}")); + } + if query.category.is_some() { + param_count += 1; + sql.push_str(&format!(" AND event_type LIKE ${param_count}")); + } + if query.severity.is_some() { + param_count += 1; + sql.push_str(&format!(" AND severity = ${param_count}")); + } + if query.subject.is_some() { + param_count += 1; + sql.push_str(&format!(" AND subject = ${param_count}")); + } + if query.cursor.is_some() { + param_count += 1; + sql.push_str(&format!(" AND created_at < ${param_count}")); + } + + param_count += 1; + sql.push_str(&format!(" ORDER BY created_at DESC LIMIT ${param_count}")); + + #[allow(clippy::type_complexity)] + let mut q = sqlx::query_as::< + _, + ( + String, + String, + String, + Option, + Option, + Value, + chrono::DateTime, + ), + >(&sql); + + if let Some(ref event_type) = query.event_type { + q = q.bind(event_type); + } + if let Some(ref category) = query.category { + q = q.bind(format!("{category}.%")); + } + if let Some(ref severity) = query.severity { + q = q.bind(severity); + } + if let Some(ref subject) = query.subject { + q = q.bind(subject); + } + if let Some(ref cursor) = query.cursor { + let ts = cursor + .parse::>() + .map_err(|_| AppError::BadRequest("invalid cursor format".to_string()))?; + q = q.bind(ts); + } + q = q.bind(limit); + + let rows = q + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to query events: {e}")))?; + + let events: Vec = rows + .into_iter() + .map(|row| EventResponse { + id: row.0, + event_type: row.1, + severity: row.2, + actor_did: row.3, + subject: row.4, + detail: row.5, + created_at: row.6, + }) + .collect(); + + let cursor = events.last().map(|e| e.created_at.to_rfc3339()); + + Ok(Json(EventsListResponse { events, cursor })) +} diff --git a/src/admin/lexicons.rs b/src/admin/lexicons.rs index ba964cb..b303f28 100644 --- a/src/admin/lexicons.rs +++ b/src/admin/lexicons.rs @@ -5,6 +5,7 @@ use serde_json::Value; use crate::AppState; use crate::error::AppError; +use crate::event_log::{EventLog, Severity, log_event}; use crate::lexicon::{LexiconType, ParsedLexicon, ProcedureAction}; use super::auth::AdminAuth; @@ -20,7 +21,7 @@ async fn notify_collections(state: &AppState) { /// POST /admin/lexicons — upload (upsert) a lexicon. pub(super) async fn upload_lexicon( State(state): State, - _admin: AdminAuth, + auth: AdminAuth, Json(body): Json, ) -> Result<(StatusCode, Json), AppError> { // Validate basic structure @@ -65,6 +66,7 @@ pub(super) async fn upload_lexicon( } let action_str = action.to_optional_str(); + let has_script = body.script.is_some(); // Upsert into database let row: (i32,) = sqlx::query_as( @@ -117,6 +119,27 @@ pub(super) async fn upload_lexicon( StatusCode::OK }; + let event_type = if status == StatusCode::CREATED { + "lexicon.created" + } else { + "lexicon.updated" + }; + log_event( + &state.db, + EventLog { + event_type: event_type.to_string(), + severity: Severity::Info, + actor_did: Some(auth.did.clone()), + subject: Some(id.clone()), + detail: serde_json::json!({ + "revision": revision, + "has_script": has_script, + "source": "manual", + }), + }, + ) + .await; + Ok(( status, Json(serde_json::json!({ @@ -254,7 +277,7 @@ pub(super) async fn get_lexicon( /// DELETE /admin/lexicons/:id — remove a lexicon. pub(super) async fn delete_lexicon( State(state): State, - _admin: AdminAuth, + auth: AdminAuth, Path(id): Path, ) -> Result { let result = sqlx::query("DELETE FROM lexicons WHERE id = $1") @@ -270,5 +293,17 @@ pub(super) async fn delete_lexicon( state.lexicons.remove(&id).await; notify_collections(&state).await; + log_event( + &state.db, + EventLog { + event_type: "lexicon.deleted".to_string(), + severity: Severity::Info, + actor_did: Some(auth.did.clone()), + subject: Some(id.clone()), + detail: serde_json::json!({}), + }, + ) + .await; + Ok(StatusCode::NO_CONTENT) } diff --git a/src/admin/mod.rs b/src/admin/mod.rs index 7e5d6de..32dfeeb 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -1,6 +1,7 @@ mod admins; pub(crate) mod auth; mod backfill; +mod events; mod lexicons; mod network_lexicons; mod records; @@ -26,6 +27,7 @@ pub fn admin_routes(_state: AppState) -> Router { .route("/stats", get(stats::stats)) .route("/backfill", post(backfill::create_backfill)) .route("/backfill/status", get(backfill::backfill_status)) + .route("/events", get(events::list_events)) .route( "/admins", post(admins::create_admin).get(admins::list_admins), diff --git a/src/aip.rs b/src/aip.rs index 3ac7e40..7941edd 100644 --- a/src/aip.rs +++ b/src/aip.rs @@ -108,6 +108,7 @@ mod tests { relay_url: String::new(), plc_url: String::new(), static_dir: String::new(), + event_log_retention_days: 30, }; let (tx, _) = watch::channel(vec![]); AppState { diff --git a/src/config.rs b/src/config.rs index 34e21c6..ed08be7 100644 --- a/src/config.rs +++ b/src/config.rs @@ -13,6 +13,7 @@ pub struct Config { pub relay_url: String, pub plc_url: String, pub static_dir: String, + pub event_log_retention_days: u32, } impl Config { @@ -32,6 +33,10 @@ impl Config { relay_url: env::var("RELAY_URL").unwrap_or_else(|_| "https://bsky.network".into()), plc_url: env::var("PLC_URL").unwrap_or_else(|_| "https://plc.directory".into()), static_dir: env::var("STATIC_DIR").unwrap_or_else(|_| "./web/out".into()), + event_log_retention_days: std::env::var("EVENT_LOG_RETENTION_DAYS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(30), } } @@ -57,6 +62,7 @@ mod tests { "TAP_ADMIN_PASSWORD", "RELAY_URL", "PLC_URL", + "EVENT_LOG_RETENTION_DAYS", ] { unsafe { env::remove_var(key); @@ -84,6 +90,7 @@ mod tests { relay_url: String::new(), plc_url: String::new(), static_dir: String::new(), + event_log_retention_days: 30, }; assert_eq!( config.listen_addr(), @@ -158,4 +165,39 @@ mod tests { } Config::from_env(); } + + #[test] + #[serial] + fn default_event_log_retention_days() { + unsafe { + clear_env(); + set_required_env(); + } + let config = Config::from_env(); + assert_eq!(config.event_log_retention_days, 30); + } + + #[test] + #[serial] + fn custom_event_log_retention_days() { + unsafe { + clear_env(); + set_required_env(); + env::set_var("EVENT_LOG_RETENTION_DAYS", "7"); + } + let config = Config::from_env(); + assert_eq!(config.event_log_retention_days, 7); + } + + #[test] + #[serial] + fn zero_event_log_retention_days_disables_cleanup() { + unsafe { + clear_env(); + set_required_env(); + env::set_var("EVENT_LOG_RETENTION_DAYS", "0"); + } + let config = Config::from_env(); + assert_eq!(config.event_log_retention_days, 0); + } } diff --git a/src/event_log.rs b/src/event_log.rs new file mode 100644 index 0000000..84ec1e2 --- /dev/null +++ b/src/event_log.rs @@ -0,0 +1,116 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sqlx::PgPool; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "lowercase")] +pub enum Severity { + Info, + Warn, + Error, +} + +impl std::fmt::Display for Severity { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Severity::Info => write!(f, "info"), + Severity::Warn => write!(f, "warn"), + Severity::Error => write!(f, "error"), + } + } +} + +pub struct EventLog { + pub event_type: String, + pub severity: Severity, + pub actor_did: Option, + pub subject: Option, + pub detail: Value, +} + +pub async fn spawn_retention_cleanup(db: PgPool, retention_days: u32) { + if retention_days == 0 { + tracing::info!("event log retention cleanup disabled"); + return; + } + + tracing::info!(retention_days, "starting event log retention cleanup task"); + + let interval = tokio::time::Duration::from_secs(3600); // 1 hour + loop { + tokio::time::sleep(interval).await; + + let result = sqlx::query( + "DELETE FROM event_logs WHERE created_at < NOW() - make_interval(days => $1)", + ) + .bind(retention_days as i32) + .execute(&db) + .await; + + match result { + Ok(result) => { + let count = result.rows_affected(); + if count > 0 { + tracing::info!(count, "cleaned up old event logs"); + } + } + Err(e) => { + tracing::warn!("failed to clean up event logs: {e}"); + } + } + } +} + +pub async fn log_event(db: &PgPool, event: EventLog) { + let severity = event.severity.to_string(); + let result = sqlx::query( + "INSERT INTO event_logs (event_type, severity, actor_did, subject, detail) + VALUES ($1, $2, $3, $4, $5)", + ) + .bind(&event.event_type) + .bind(&severity) + .bind(&event.actor_did) + .bind(&event.subject) + .bind(&event.detail) + .execute(db) + .await; + + if let Err(e) = result { + tracing::warn!(event_type = %event.event_type, "failed to log event: {e}"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn severity_display() { + assert_eq!(Severity::Info.to_string(), "info"); + assert_eq!(Severity::Warn.to_string(), "warn"); + assert_eq!(Severity::Error.to_string(), "error"); + } + + #[test] + fn severity_serializes_lowercase() { + assert_eq!(serde_json::to_string(&Severity::Info).unwrap(), "\"info\""); + assert_eq!( + serde_json::to_string(&Severity::Error).unwrap(), + "\"error\"" + ); + } + + #[test] + fn event_log_construction() { + let event = EventLog { + event_type: "lexicon.created".to_string(), + severity: Severity::Info, + actor_did: Some("did:plc:test".to_string()), + subject: Some("com.example.test".to_string()), + detail: serde_json::json!({"revision": 1}), + }; + assert_eq!(event.event_type, "lexicon.created"); + assert_eq!(event.severity, Severity::Info); + assert_eq!(event.actor_did.unwrap(), "did:plc:test"); + } +} diff --git a/src/lib.rs b/src/lib.rs index 911ae69..0439395 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,6 +3,7 @@ pub mod aip; pub mod auth; pub mod config; pub mod error; +pub mod event_log; pub mod lexicon; pub mod lua; pub mod profile; diff --git a/src/lua/execute.rs b/src/lua/execute.rs index 7bebf1e..1417284 100644 --- a/src/lua/execute.rs +++ b/src/lua/execute.rs @@ -4,10 +4,12 @@ use mlua::LuaSerdeExt; use serde_json::Value; use std::collections::HashMap; use std::sync::Arc; +use std::time::Instant; use crate::AppState; use crate::auth::Claims; use crate::error::AppError; +use crate::event_log::{EventLog, Severity, log_event}; use crate::lexicon::ParsedLexicon; use crate::repo; @@ -25,50 +27,254 @@ pub async fn execute_procedure_script( lexicon: &ParsedLexicon, script: &str, ) -> Result { + let start = Instant::now(); let collection = lexicon.target_collection.as_deref().unwrap_or_default(); - let session = repo::get_atp_session(state, claims.token()).await?; + // Capture script source and input for error logging before anything is consumed. + let script_source = script.to_string(); + let input_json = input.clone(); - let lua = sandbox::create_sandbox() - .map_err(|e| AppError::Internal(format!("failed to create Lua VM: {e}")))?; + let session = match repo::get_atp_session(state, claims.token()).await { + Ok(s) => s, + Err(e) => { + let error_message = format!("{e}"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: Some(claims.did().to_string()), + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "input": input_json, + "caller_did": claims.did(), + "method": method, + }), + }, + ) + .await; + return Err(e); + } + }; + + let lua = match sandbox::create_sandbox() { + Ok(l) => l, + Err(e) => { + let error_message = format!("failed to create Lua VM: {e}"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: Some(claims.did().to_string()), + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "input": input_json, + "caller_did": claims.did(), + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal(error_message)); + } + }; let state_arc = Arc::new(state.clone()); let claims_arc = Arc::new(claims.clone()); let session_arc = Arc::new(session); - db_api::register_db_api(&lua, state_arc.clone()) - .map_err(|e| AppError::Internal(format!("failed to register db API: {e}")))?; + if let Err(e) = db_api::register_db_api(&lua, state_arc.clone()) { + let error_message = format!("failed to register db API: {e}"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: Some(claims.did().to_string()), + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "input": input_json, + "caller_did": claims.did(), + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal(error_message)); + } - record::register_record_api(&lua, state_arc, claims_arc, session_arc) - .map_err(|e| AppError::Internal(format!("failed to register Record API: {e}")))?; + if let Err(e) = record::register_record_api(&lua, state_arc, claims_arc, session_arc) { + let error_message = format!("failed to register Record API: {e}"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: Some(claims.did().to_string()), + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "input": input_json, + "caller_did": claims.did(), + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal(error_message)); + } - context::set_procedure_context(&lua, method, input, claims.did(), collection) - .map_err(|e| AppError::Internal(format!("failed to set context: {e}")))?; + if let Err(e) = context::set_procedure_context(&lua, method, input, claims.did(), collection) { + let error_message = format!("failed to set context: {e}"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: Some(claims.did().to_string()), + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "input": input_json, + "caller_did": claims.did(), + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal(error_message)); + } - lua.load(script).exec().map_err(|e| { + if let Err(e) = lua.load(script).exec() { + let error_message = format!("{e}"); tracing::error!(method, error = %e, "lua script load failed"); - AppError::Internal("script execution failed".into()) - })?; - - let handle: mlua::Function = lua.globals().get("handle").map_err(|e| { - tracing::error!(method, error = %e, "lua script missing handle function"); - AppError::Internal("script execution failed".into()) - })?; - - let result: mlua::Value = handle.call_async(()).await.map_err(|e| { - let msg = e.to_string(); - tracing::error!(method, error = %msg, "lua script execution failed"); - if msg.contains("execution limit") { - AppError::Internal("script exceeded execution time limit".into()) - } else { - AppError::Internal("script execution failed".into()) + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: Some(claims.did().to_string()), + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "input": input_json, + "caller_did": claims.did(), + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal("script execution failed".into())); + } + + let handle: mlua::Function = match lua.globals().get("handle") { + Ok(f) => f, + Err(e) => { + let error_message = format!("{e}"); + tracing::error!(method, error = %e, "lua script missing handle function"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: Some(claims.did().to_string()), + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "input": input_json, + "caller_did": claims.did(), + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal("script execution failed".into())); + } + }; + + let result: mlua::Value = match handle.call_async(()).await { + Ok(r) => r, + Err(e) => { + let msg = e.to_string(); + tracing::error!(method, error = %msg, "lua script execution failed"); + let app_error = if msg.contains("execution limit") { + AppError::Internal("script exceeded execution time limit".into()) + } else { + AppError::Internal("script execution failed".into()) + }; + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: Some(claims.did().to_string()), + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": msg, + "script_source": script_source, + "input": input_json, + "caller_did": claims.did(), + "method": method, + }), + }, + ) + .await; + return Err(app_error); + } + }; + + let json_value: Value = match lua.from_value(result) { + Ok(v) => v, + Err(e) => { + let error_message = format!("{e}"); + tracing::error!(method, error = %e, "failed to convert lua result to JSON"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: Some(claims.did().to_string()), + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "input": input_json, + "caller_did": claims.did(), + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal("script execution failed".into())); } - })?; + }; - let json_value: Value = lua.from_value(result).map_err(|e| { - tracing::error!(method, error = %e, "failed to convert lua result to JSON"); - AppError::Internal("script execution failed".into()) - })?; + log_event( + &state.db, + EventLog { + event_type: "script.executed".to_string(), + severity: Severity::Info, + actor_did: Some(claims.did().to_string()), + subject: Some(method.to_string()), + detail: serde_json::json!({ + "method": method, + "caller_did": claims.did(), + "duration_ms": start.elapsed().as_millis() as u64, + }), + }, + ) + .await; Ok(Json(json_value).into_response()) } @@ -81,43 +287,189 @@ pub async fn execute_query_script( lexicon: &ParsedLexicon, script: &str, ) -> Result { + let start = Instant::now(); let collection = lexicon.target_collection.as_deref().unwrap_or_default(); - let lua = sandbox::create_sandbox() - .map_err(|e| AppError::Internal(format!("failed to create Lua VM: {e}")))?; + // Capture script source for error logging. + let script_source = script.to_string(); + + let lua = match sandbox::create_sandbox() { + Ok(l) => l, + Err(e) => { + let error_message = format!("failed to create Lua VM: {e}"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal(error_message)); + } + }; let state_arc = Arc::new(state.clone()); - db_api::register_db_api(&lua, state_arc) - .map_err(|e| AppError::Internal(format!("failed to register db API: {e}")))?; + if let Err(e) = db_api::register_db_api(&lua, state_arc) { + let error_message = format!("failed to register db API: {e}"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal(error_message)); + } - context::set_query_context(&lua, method, params, collection) - .map_err(|e| AppError::Internal(format!("failed to set context: {e}")))?; + if let Err(e) = context::set_query_context(&lua, method, params, collection) { + let error_message = format!("failed to set context: {e}"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal(error_message)); + } - lua.load(script).exec().map_err(|e| { + if let Err(e) = lua.load(script).exec() { + let error_message = format!("{e}"); tracing::error!(method, error = %e, "lua script load failed"); - AppError::Internal("script execution failed".into()) - })?; - - let handle: mlua::Function = lua.globals().get("handle").map_err(|e| { - tracing::error!(method, error = %e, "lua script missing handle function"); - AppError::Internal("script execution failed".into()) - })?; - - let result: mlua::Value = handle.call_async(()).await.map_err(|e| { - let msg = e.to_string(); - tracing::error!(method, error = %msg, "lua script execution failed"); - if msg.contains("execution limit") { - AppError::Internal("script exceeded execution time limit".into()) - } else { - AppError::Internal("script execution failed".into()) + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal("script execution failed".into())); + } + + let handle: mlua::Function = match lua.globals().get("handle") { + Ok(f) => f, + Err(e) => { + let error_message = format!("{e}"); + tracing::error!(method, error = %e, "lua script missing handle function"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal("script execution failed".into())); + } + }; + + let result: mlua::Value = match handle.call_async(()).await { + Ok(r) => r, + Err(e) => { + let msg = e.to_string(); + tracing::error!(method, error = %msg, "lua script execution failed"); + let app_error = if msg.contains("execution limit") { + AppError::Internal("script exceeded execution time limit".into()) + } else { + AppError::Internal("script execution failed".into()) + }; + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": msg, + "script_source": script_source, + "method": method, + }), + }, + ) + .await; + return Err(app_error); + } + }; + + let json_value: Value = match lua.from_value(result) { + Ok(v) => v, + Err(e) => { + let error_message = format!("{e}"); + tracing::error!(method, error = %e, "failed to convert lua result to JSON"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "method": method, + }), + }, + ) + .await; + return Err(AppError::Internal("script execution failed".into())); } - })?; + }; - let json_value: Value = lua.from_value(result).map_err(|e| { - tracing::error!(method, error = %e, "failed to convert lua result to JSON"); - AppError::Internal("script execution failed".into()) - })?; + log_event( + &state.db, + EventLog { + event_type: "script.executed".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(method.to_string()), + detail: serde_json::json!({ + "method": method, + "duration_ms": start.elapsed().as_millis() as u64, + }), + }, + ) + .await; Ok(Json(json_value).into_response()) } diff --git a/src/main.rs b/src/main.rs index 88eb3be..69237c6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -136,6 +136,11 @@ async fn main() { state.collections_tx.clone(), ); + tokio::spawn(happyview::event_log::spawn_retention_cleanup( + state.db.clone(), + state.config.event_log_retention_days, + )); + let app = server::router(state); let addr = config.listen_addr(); diff --git a/src/tap.rs b/src/tap.rs index 0641dcb..3602094 100644 --- a/src/tap.rs +++ b/src/tap.rs @@ -6,6 +6,7 @@ use tokio::sync::watch; use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use crate::event_log::{EventLog, Severity, log_event}; use crate::lexicon::{LexiconRegistry, LexiconType, ParsedLexicon, ProcedureAction}; // --------------------------------------------------------------------------- @@ -314,6 +315,18 @@ async fn run( ) = tokio_tungstenite::connect_async(request).await?; tracing::info!("connected to tap"); + log_event( + db, + EventLog { + event_type: "tap.connected".to_string(), + severity: Severity::Info, + actor_did: None, + subject: None, + detail: serde_json::json!({ "url": ws_url }), + }, + ) + .await; + let (mut write, mut read) = ws.split(); loop { @@ -386,6 +399,18 @@ async fn run( } } + log_event( + db, + EventLog { + event_type: "tap.disconnected".to_string(), + severity: Severity::Warn, + actor_did: None, + subject: None, + detail: serde_json::json!({ "reason": "connection closed" }), + }, + ) + .await; + Ok(()) } @@ -429,7 +454,7 @@ async fn handle_record_event( }; let cid = record.cid.as_deref().unwrap_or_default(); - if let Err(e) = sqlx::query( + match sqlx::query( r#" INSERT INTO records (uri, did, collection, rkey, record, cid, indexed_at) VALUES ($1, $2, $3, $4, $5, $6, NOW()) @@ -448,16 +473,86 @@ async fn handle_record_event( .execute(db) .await { - tracing::warn!(uri = %uri, "failed to upsert record: {e}"); + Ok(_) => { + log_event( + db, + EventLog { + event_type: "record.created".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + }), + }, + ) + .await; + } + Err(e) => { + tracing::warn!(uri = %uri, "failed to upsert record: {e}"); + log_event( + db, + EventLog { + event_type: "record.created".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + "error": e.to_string(), + }), + }, + ) + .await; + } } } "delete" => { - if let Err(e) = sqlx::query("DELETE FROM records WHERE uri = $1") + match sqlx::query("DELETE FROM records WHERE uri = $1") .bind(&uri) .execute(db) .await { - tracing::warn!(uri = %uri, "failed to delete record: {e}"); + Ok(_) => { + log_event( + db, + EventLog { + event_type: "record.deleted".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + }), + }, + ) + .await; + } + Err(e) => { + tracing::warn!(uri = %uri, "failed to delete record: {e}"); + log_event( + db, + EventLog { + event_type: "record.deleted".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + "error": e.to_string(), + }), + }, + ) + .await; + } } } _ => {} diff --git a/tests/common/app.rs b/tests/common/app.rs index 65574f2..955fbbe 100644 --- a/tests/common/app.rs +++ b/tests/common/app.rs @@ -41,6 +41,7 @@ impl TestApp { relay_url: mock_url.clone(), plc_url: mock_url.clone(), static_dir: "./web/out".into(), + event_log_retention_days: 30, }; // Seed the admin DID directly so tests don't rely on auto-bootstrap.