nodedb 0.2.0

// SPDX-License-Identifier: BUSL-1.1

//! HTTP poll endpoint for change stream consumption.
//!
//! `GET /v1/streams/{stream}/poll?group={group}&limit=100&partition=3`
//!
//! Returns a JSON batch of events from the stream buffer, starting after
//! the consumer group's committed offsets. Does NOT auto-commit.

use axum::Json;
use axum::extract::{Path, Query, State};
use axum::http::StatusCode;
use axum::response::IntoResponse;
use serde::{Deserialize, Serialize};

use super::super::auth::{AppState, ResolvedIdentity};
use crate::event::cdc::consume::{ConsumeError, ConsumeParams, ConsumeResult, consume_stream};

/// Query parameters.
#[derive(Deserialize, Default)]
pub struct PollParams {
    /// Consumer group name (required).
    pub group: Option<String>,
    /// Maximum events to return. Default: 100.
    pub limit: Option<usize>,
    /// Optional: consume from a specific partition only.
    pub partition: Option<u32>,
    /// Detected and rejected — callers must not supply `tenant_id` as a
    /// query parameter. Tenant is always sourced from the bearer token.
    pub tenant_id: Option<u64>,
}

/// Response body.
///
/// All fields that were present before v1.0 remain. `evicted_since_last_poll`
/// and `oldest_available_lsn` are additive — HTTP clients and ORMs that ignore
/// unknown JSON fields will not break.
#[derive(Serialize)]
pub struct PollResponse {
    /// Events in this batch.
    pub events: Vec<serde_json::Value>,
    /// Per-partition latest LSN in this batch.
    pub partition_offsets: std::collections::BTreeMap<String, u64>,
    /// Total events returned.
    pub count: usize,
    /// Events dropped from this stream's buffer since the previous poll for
    /// this consumer group. Zero on the first poll or when the buffer has not
    /// overflowed. A non-zero value means the consumer has a gap: events
    /// between the last committed LSN and `oldest_available_lsn` are gone.
    pub evicted_since_last_poll: u64,
    /// Oldest LSN still available in the stream buffer. Zero when the buffer
    /// is empty. If `evicted_since_last_poll > 0`, seek here to resume
    /// consumption (events before this LSN are permanently lost).
    pub oldest_available_lsn: u64,
}

/// `GET /v1/streams/{stream}/poll`
pub async fn poll_stream(
    identity: ResolvedIdentity,
    Path(stream_name): Path<String>,
    Query(params): Query<PollParams>,
    State(state): State<AppState>,
) -> impl IntoResponse {
    // Reject any attempt to override the caller's tenant via query string.
    if params.tenant_id.is_some() {
        return (
            StatusCode::FORBIDDEN,
            Json(serde_json::json!({
                "error": "tenant_id must not be supplied as a query parameter; \
                          tenant is determined from the bearer token"
            })),
        )
            .into_response();
    }

    let group = match params.group {
        Some(g) => g.to_lowercase(),
        None => {
            return (
                StatusCode::BAD_REQUEST,
                Json(serde_json::json!({"error": "missing 'group' query parameter"})),
            )
                .into_response();
        }
    };

    let tenant_id = identity.tenant_id().as_u64();
    let limit = params.limit.unwrap_or(100).min(10_000);
    let stream_name = stream_name.to_lowercase();

    let consume_params = ConsumeParams {
        tenant_id,
        stream_name: &stream_name,
        group_name: &group,
        partition: params.partition,
        limit,
    };

    let result = match consume_stream(&state.shared, &consume_params) {
        Ok(r) => r,
        Err(ConsumeError::RemotePartition { leader_node, .. }) => {
            // Forward to remote node.
            match crate::event::cdc::consume::consume_remote(
                &state.shared,
                &consume_params,
                leader_node,
            )
            .await
            {
                Ok(r) => r,
                Err(e) => {
                    return (
                        StatusCode::BAD_GATEWAY,
                        Json(serde_json::json!({"error": e.to_string()})),
                    )
                        .into_response();
                }
            }
        }
        Err(ConsumeError::BufferEmpty(_)) => ConsumeResult {
            events: Vec::new(),
            partition_offsets: Vec::new(),
            evicted_since_last_poll: 0,
            oldest_available_lsn: 0,
        },
        Err(e) => {
            return (
                StatusCode::NOT_FOUND,
                Json(serde_json::json!({"error": e.to_string()})),
            )
                .into_response();
        }
    };

    let events: Vec<serde_json::Value> = result
        .events
        .iter()
        .map(|e| serde_json::to_value(e).unwrap_or_default())
        .collect();
    let count = events.len();
    let partition_offsets: std::collections::BTreeMap<String, u64> = result
        .partition_offsets
        .into_iter()
        .map(|(pid, lsn)| (pid.to_string(), lsn))
        .collect();

    Json(PollResponse {
        events,
        partition_offsets,
        count,
        evicted_since_last_poll: result.evicted_since_last_poll,
        oldest_available_lsn: result.oldest_available_lsn,
    })
    .into_response()
}