feat: add DA inclusion status tracking from ev-node

.env.example

@@ -30,6 +30,18 @@ RPC_BATCH_SIZE=20
 # API_DB_MAX_CONNECTIONS=20
 # SSE_REPLAY_BUFFER_BLOCKS=4096  # replay tail used only for active connected clients
 
+# Optional: enable DA (Data Availability) inclusion tracking from ev-node.
+# Set this to true only when you also provide EVNODE_URL below.
+ENABLE_DA_TRACKING=false
+
+# Required when ENABLE_DA_TRACKING=true.
+# Must be reachable from the atlas-server process/container.
+# EVNODE_URL=http://<ev-node-host-reachable-from-atlas-server>:7331
+
+# Optional when ENABLE_DA_TRACKING=true.
+# DA_RPC_REQUESTS_PER_SECOND=50
+# DA_WORKER_CONCURRENCY=50
+
 # Branding / white-label (all optional)
 # CHAIN_LOGO_URL=                      # URL or path to logo (e.g., /branding/logo.svg). Default: bundled logo
 # ACCENT_COLOR=                        # Primary accent hex (e.g. #3b82f6). Default: #dc2626 (red)

CLAUDE.md

@@ -66,7 +66,7 @@ let cursor = (total_count - 1) - (pagination.page.saturating_sub(1) as i64) * li
 ### Row count estimation
 For large tables (transactions, addresses), use `pg_class.reltuples` instead of `COUNT(*)`:
 ```rust
-// handlers/mod.rs — get_table_count(pool)
+// handlers/mod.rs — get_table_count(pool, table_name)
 // Partition-aware: sums child reltuples, falls back to parent
 // For tables < 100k rows: falls back to exact COUNT(*)
 ```
@@ -77,19 +77,25 @@ For large tables (transactions, addresses), use `pg_class.reltuples` instead of
 ### AppState (API)
 ```rust
 pub struct AppState {
-    pub pool: PgPool,                           // API pool only
-    pub block_events_tx: broadcast::Sender<()>, // shared with indexer
+    pub pool: PgPool,                                // API pool only
+    pub block_events_tx: broadcast::Sender<()>,      // shared with indexer
+    pub da_events_tx: broadcast::Sender<Vec<DaSseUpdate>>, // shared with DA worker
+    pub head_tracker: Arc<HeadTracker>,
     pub rpc_url: String,
-    pub solc_path: String,
-    pub admin_api_key: Option<String>,
+    pub da_tracking_enabled: bool,
+    pub chain_id: u64,
+    pub chain_name: String,
 }
 ```
 
+### DA tracking (optional)
+When `ENABLE_DA_TRACKING=true`, a background DA worker queries ev-node for Celestia inclusion heights per block. `EVNODE_URL` is required only in that mode. Updates are pushed to SSE clients via an in-process `broadcast::Sender<Vec<DaSseUpdate>>`. The SSE handler streams `da_batch` events for incremental updates and emits `da_resync` when a client falls behind and should refetch visible DA state.
+
 ### Frontend API client
 - Base URL: `/api` (proxied by nginx to `atlas-server:3000`)
-- Fast polling endpoint: `GET /api/height` → `{ block_height, indexed_at }` — single key-value lookup from `indexer_state`, sub-ms. Used by the navbar as a polling fallback when SSE is disconnected.
+- Fast polling endpoint: `GET /api/height` → `{ block_height, indexed_at, features: { da_tracking } }` — serves from `head_tracker` first and falls back to `indexer_state` when the in-memory head is empty. Used by the navbar as a polling fallback when SSE is disconnected and by feature-flag consumers.
 - Chain status: `GET /api/status` → `{ chain_id, chain_name, block_height, total_transactions, total_addresses, indexed_at }` — full chain info, fetched once on page load.
-- `GET /api/events` → SSE stream of `new_block` events, one per block in order. Primary live-update path for navbar counter and blocks page. Falls back to `/api/height` polling on disconnect.
+- `GET /api/events` → SSE stream of `new_block`, `da_batch`, and `da_resync` events. Primary live-update path for navbar counter, blocks page, block detail DA status, and DA resync handling. Falls back to `/api/height` polling on disconnect.
 
 ## Important Conventions
 
@@ -109,13 +115,18 @@ Key vars (see `.env.example` for full list):
 |---|---|---|
 | `DATABASE_URL` | all | required |
 | `RPC_URL` | server | required |
+| `CHAIN_NAME` | server | `"Unknown"` |
 | `DB_MAX_CONNECTIONS` | indexer pool | `20` |
 | `API_DB_MAX_CONNECTIONS` | API pool | `20` |
 | `BATCH_SIZE` | indexer | `100` |
 | `FETCH_WORKERS` | indexer | `10` |
 | `ADMIN_API_KEY` | API | none |
 | `API_HOST` | API | `127.0.0.1` |
 | `API_PORT` | API | `3000` |
+| `ENABLE_DA_TRACKING` | server | `false` |
+| `EVNODE_URL` | server | none |
+| `DA_RPC_REQUESTS_PER_SECOND` | DA worker | `50` |
+| `DA_WORKER_CONCURRENCY` | DA worker | `50` |
 
 ## Running Locally
 

backend/crates/atlas-common/src/types.rs

@@ -16,6 +16,18 @@ pub struct Block {
     pub indexed_at: DateTime<Utc>,
 }
 
+/// DA (Data Availability) status for a block on L2 chains using Celestia.
+/// Only populated when DA tracking is enabled and the DA worker has checked the block.
+#[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
+pub struct BlockDaStatus {
+    pub block_number: i64,
+    /// Celestia height where the block header was submitted. 0 = pending.
+    pub header_da_height: i64,
+    /// Celestia height where the block data was submitted. 0 = pending.
+    pub data_da_height: i64,
+    pub updated_at: DateTime<Utc>,
+}
+
 /// Transaction data as stored in the database
 #[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
 pub struct Transaction {

backend/crates/atlas-server/src/api/handlers/blocks.rs

@@ -2,16 +2,27 @@ use axum::{
     extract::{Path, Query, State},
     Json,
 };
+use serde::Serialize;
 use std::sync::Arc;
 
 use crate::api::error::ApiResult;
 use crate::api::AppState;
-use atlas_common::{AtlasError, Block, PaginatedResponse, Pagination, Transaction};
+use atlas_common::{AtlasError, Block, BlockDaStatus, PaginatedResponse, Pagination, Transaction};
+
+/// Block response with optional DA status.
+/// DA fields are always present in the JSON (null when no data),
+/// so the frontend can rely on a stable schema.
+#[derive(Serialize)]
+pub struct BlockResponse {
+    #[serde(flatten)]
+    pub block: Block,
+    pub da_status: Option<BlockDaStatus>,
+}
 
 pub async fn list_blocks(
     State(state): State<Arc<AppState>>,
     Query(pagination): Query<Pagination>,
-) -> ApiResult<Json<PaginatedResponse<Block>>> {
+) -> ApiResult<Json<PaginatedResponse<BlockResponse>>> {
     // Use MAX(number) + 1 instead of COUNT(*) - blocks are sequential so this is accurate
     // This is ~6500x faster than COUNT(*) on large tables
     let total: (Option<i64>,) = sqlx::query_as("SELECT MAX(number) + 1 FROM blocks")
@@ -30,15 +41,37 @@ pub async fn list_blocks(
          FROM blocks
          WHERE number <= $2
          ORDER BY number DESC
-         LIMIT $1"
+         LIMIT $1",
     )
     .bind(limit)
     .bind(cursor)
     .fetch_all(&state.pool)
     .await?;
 
+    // Batch-fetch DA status for all blocks in this page
+    let block_numbers: Vec<i64> = blocks.iter().map(|b| b.number).collect();
+    let da_rows: Vec<BlockDaStatus> = sqlx::query_as(
+        "SELECT block_number, header_da_height, data_da_height, updated_at
+         FROM block_da_status
+         WHERE block_number = ANY($1)",
+    )
+    .bind(&block_numbers)
+    .fetch_all(&state.pool)
+    .await?;
+
+    let da_map: std::collections::HashMap<i64, BlockDaStatus> =
+        da_rows.into_iter().map(|d| (d.block_number, d)).collect();
+
+    let responses: Vec<BlockResponse> = blocks
+        .into_iter()
+        .map(|block| {
+            let da_status = da_map.get(&block.number).cloned();
+            BlockResponse { block, da_status }
+        })
+        .collect();
+
     Ok(Json(PaginatedResponse::new(
-        blocks,
+        responses,
         pagination.page,
         pagination.limit,
         total_count,
@@ -48,18 +81,27 @@ pub async fn list_blocks(
 pub async fn get_block(
     State(state): State<Arc<AppState>>,
     Path(number): Path<i64>,
-) -> ApiResult<Json<Block>> {
+) -> ApiResult<Json<BlockResponse>> {
     let block: Block = sqlx::query_as(
         "SELECT number, hash, parent_hash, timestamp, gas_used, gas_limit, transaction_count, indexed_at
          FROM blocks
-         WHERE number = $1"
+         WHERE number = $1",
     )
     .bind(number)
     .fetch_optional(&state.pool)
     .await?
     .ok_or_else(|| AtlasError::NotFound(format!("Block {} not found", number)))?;
 
-    Ok(Json(block))
+    let da_status: Option<BlockDaStatus> = sqlx::query_as(
+        "SELECT block_number, header_da_height, data_da_height, updated_at
+         FROM block_da_status
+         WHERE block_number = $1",
+    )
+    .bind(number)
+    .fetch_optional(&state.pool)
+    .await?;
+
+    Ok(Json(BlockResponse { block, da_status }))
 }
 
 pub async fn get_block_transactions(
@@ -77,7 +119,7 @@ pub async fn get_block_transactions(
          FROM transactions
          WHERE block_number = $1
          ORDER BY block_index ASC
-         LIMIT $2 OFFSET $3"
+         LIMIT $2 OFFSET $3",
     )
     .bind(number)
     .bind(pagination.limit())

backend/crates/atlas-server/src/api/handlers/faucet.rs

@@ -161,11 +161,14 @@ mod tests {
             .expect("lazy pool");
         let head_tracker = Arc::new(crate::head::HeadTracker::empty(10));
         let (tx, _) = broadcast::channel(1);
+        let (da_tx, _) = broadcast::channel(1);
         Arc::new(AppState {
             pool,
             block_events_tx: tx,
+            da_events_tx: da_tx,
             head_tracker,
             rpc_url: String::new(),
+            da_tracking_enabled: false,
             faucet,
             chain_id: 1,
             chain_name: "Test Chain".to_string(),