Add daily-active-accounts computation job

.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
 # ---------------------------------------------------------------------------

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);

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<Self>) {
+        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<i64> {
+        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)
+    }
+}

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,
 };

src/jobs/scheduler.rs

@@ -115,7 +115,7 @@ impl JobScheduler {
     }
 
     pub fn start(
-        _db: Arc<Database>,
+        db: Arc<Database>,
         cache: Arc<CacheManager>,
         _rpc: Arc<StellarRpcClient>,
         ingestion: Arc<DataIngestionService>,
@@ -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
     }