diff --git a/.env.example b/.env.example index a53aa8a..a3d09f9 100644 --- a/.env.example +++ b/.env.example @@ -288,6 +288,13 @@ JOB_PRICE_FEED_UPDATE_INTERVAL_SECONDS=900 # Cache cleanup job (default: 3600 seconds = 1 hour) JOB_CACHE_CLEANUP_ENABLED=true JOB_CACHE_CLEANUP_INTERVAL_SECONDS=3600 + +# Daily active accounts job (default: 900 seconds = 15 minutes). +# Computes DAA/tx-count for the current UTC calendar day and upserts into +# network_daily_metrics; backs the /api/v1/stats/summary and +# /api/v1/network/daily-active-accounts endpoints. +JOB_DAA_ENABLED=true +JOB_DAA_INTERVAL_SECONDS=900 # --------------------------------------------------------------------------- # Telegram Bot Configuration # --------------------------------------------------------------------------- diff --git a/migrations/031_create_network_metrics_timeseries.sql b/migrations/031_create_network_metrics_timeseries.sql new file mode 100644 index 0000000..3d041af --- /dev/null +++ b/migrations/031_create_network_metrics_timeseries.sql @@ -0,0 +1,27 @@ +-- Time-series schema for the Network Dashboard (issue #45). +-- +-- One wide row per UTC calendar day, rather than narrow per-metric rows, +-- since #15-#19's endpoints each want "give me the last N days of X" and a +-- wide table lets a single row backfill/upsert cover every metric for that +-- day without a multi-table join. `date` is the UTC calendar day the metrics +-- were computed for (see #2/#3's day-boundary convention). +CREATE TABLE IF NOT EXISTS network_daily_metrics ( + date TEXT PRIMARY KEY, -- YYYY-MM-DD, UTC calendar day + daily_active_accounts INTEGER NOT NULL DEFAULT 0, + new_accounts INTEGER NOT NULL DEFAULT 0, + transaction_count INTEGER NOT NULL DEFAULT 0, + payment_volume_usd REAL NOT NULL DEFAULT 0, + avg_fee_stroops REAL NOT NULL DEFAULT 0, + median_fee_stroops REAL NOT NULL DEFAULT 0, + active_soroban_contracts INTEGER NOT NULL DEFAULT 0, + -- distinguishes "job ran and computed a real value" from "no row yet" for + -- #15's gap-vs-zero requirement; a present row with is_complete=0 means a + -- job run was interrupted partway through (see #2's partial-run note). + is_complete INTEGER NOT NULL DEFAULT 1, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +-- date is already the primary key (and therefore indexed), but the explicit +-- index documents intent and survives if the PK strategy ever changes. +CREATE INDEX IF NOT EXISTS idx_network_daily_metrics_date ON network_daily_metrics(date DESC); diff --git a/src/jobs/daily_active_accounts.rs b/src/jobs/daily_active_accounts.rs new file mode 100644 index 0000000..147a934 --- /dev/null +++ b/src/jobs/daily_active_accounts.rs @@ -0,0 +1,131 @@ +use anyhow::Result; +use sqlx::SqlitePool; +use std::sync::Arc; +use tokio::time::{interval, Duration as TokioDuration}; +use tracing::{info, warn}; + +use crate::observability::job_metrics::JobMetricsCollector; + +/// Configuration for the daily-active-accounts job. +#[derive(Debug, Clone)] +pub struct DaaJobConfig { + pub enabled: bool, + pub interval_seconds: u64, +} + +impl Default for DaaJobConfig { + fn default() -> Self { + Self { + enabled: std::env::var("JOB_DAA_ENABLED") + .unwrap_or_else(|_| "true".to_string()) + .parse() + .unwrap_or(true), + interval_seconds: std::env::var("JOB_DAA_INTERVAL_SECONDS") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(900), + } + } +} + +/// Computes Daily Active Accounts (and companion counters) for the current +/// UTC calendar day and upserts the result into `network_daily_metrics`. +/// +/// **Window convention:** this job uses the *UTC calendar day* (`date('now')` +/// in SQLite), not a rolling 24h window from "now". This matches +/// `network_daily_metrics.date`'s definition (see migration 031) and is the +/// convention #1/#3 build their "24h" fields and delta calculations against +/// — pick one, and this is it. A rolling-24h number would drift against the +/// snapshot-diff math in #3. +/// +/// **Definition:** "active account" = a distinct `source_account` on a +/// `payments` row created today, per this issue's guidance to reuse the +/// already-extracted payment data (`ExtractedPayment.source_account` in +/// `src/ingestion/ledger.rs`) rather than re-parsing ledgers from scratch. +pub struct DailyActiveAccountsJob { + pool: SqlitePool, + config: DaaJobConfig, +} + +impl DailyActiveAccountsJob { + #[must_use] + pub const fn new(pool: SqlitePool, config: DaaJobConfig) -> Self { + Self { pool, config } + } + + pub async fn start(self: Arc) { + if !self.config.enabled { + info!("Daily active accounts job is disabled"); + return; + } + + info!( + "Starting daily-active-accounts job (interval: {}s)", + self.config.interval_seconds + ); + + let mut ticker = interval(TokioDuration::from_secs(self.config.interval_seconds)); + + loop { + ticker.tick().await; + + let _metrics = JobMetricsCollector::new("daily-active-accounts"); + match self.run_once().await { + Ok(count) => { + info!("Daily active accounts computed: {}", count); + _metrics.complete_success(); + } + Err(e) => { + warn!("Daily active accounts job failed: {}", e); + _metrics.complete_failure(&e.to_string()); + } + } + } + } + + /// Computes today's DAA figure and upserts it. The full aggregate is + /// computed in memory *before* any write happens, so a crash mid-query + /// simply means "no write occurred this run" rather than a partial or + /// corrupt figure landing in `network_daily_metrics` — the same + /// all-or-nothing guarantee `contract_event_listener.rs` gets from only + /// advancing its cursor after a batch fully succeeds. + pub async fn run_once(&self) -> Result { + let daa: i64 = sqlx::query_scalar( + r" + SELECT COUNT(DISTINCT source_account) + FROM payments + WHERE date(created_at) = date('now') + ", + ) + .fetch_one(&self.pool) + .await?; + + let tx_count: i64 = sqlx::query_scalar( + r" + SELECT COUNT(*) + FROM payments + WHERE date(created_at) = date('now') + ", + ) + .fetch_one(&self.pool) + .await?; + + sqlx::query( + r" + INSERT INTO network_daily_metrics (date, daily_active_accounts, transaction_count, is_complete, updated_at) + VALUES (date('now'), ?1, ?2, 1, CURRENT_TIMESTAMP) + ON CONFLICT(date) DO UPDATE SET + daily_active_accounts = excluded.daily_active_accounts, + transaction_count = excluded.transaction_count, + is_complete = 1, + updated_at = CURRENT_TIMESTAMP + ", + ) + .bind(daa) + .bind(tx_count) + .execute(&self.pool) + .await?; + + Ok(daa) + } +} diff --git a/src/jobs/mod.rs b/src/jobs/mod.rs index 02bd782..debf66a 100644 --- a/src/jobs/mod.rs +++ b/src/jobs/mod.rs @@ -1,10 +1,12 @@ pub mod asset_revalidation; pub mod backfill; pub mod contract_event_listener; +pub mod daily_active_accounts; pub mod scheduler; pub mod trace_aware_executor; pub use asset_revalidation::{AssetRevalidationJob, RevalidationConfig, RevalidationStats}; +pub use daily_active_accounts::{DaaJobConfig, DailyActiveAccountsJob}; pub use backfill::{ BackfillJob, BackfillRequest, BackfillState, BackfillStateRef, BackfillStatus, LedgerGap, }; diff --git a/src/jobs/scheduler.rs b/src/jobs/scheduler.rs index 96cdd6c..6c3013c 100644 --- a/src/jobs/scheduler.rs +++ b/src/jobs/scheduler.rs @@ -115,7 +115,7 @@ impl JobScheduler { } pub fn start( - _db: Arc, + db: Arc, cache: Arc, _rpc: Arc, ingestion: Arc, @@ -170,6 +170,24 @@ impl JobScheduler { }) }); + // Daily active accounts job (issue #2) + let config = JobConfig::from_env("daa", 900); + let daa_pool = db.pool().clone(); + scheduler.add_job(config, move || { + let pool = daa_pool.clone(); + Box::pin(async move { + let job = crate::jobs::daily_active_accounts::DailyActiveAccountsJob::new( + pool, + crate::jobs::daily_active_accounts::DaaJobConfig { + enabled: true, + interval_seconds: 900, + }, + ); + job.run_once().await?; + Ok(()) + }) + }); + scheduler }