Add topic admin ops & secure file download

Introduce multiple topic and file-download improvements:

- Add POST /v1/api/topics/latest-offsets endpoint to resolve partition head offsets (service/dba/system only).
- Make topic consume support optional consumer group (stateless inspection); honor requested position for stateless reads and route fetches accordingly.
- Add RESET CONSUMER GROUP SQL command/parser/classifier, integrate into extensions and dialect, and implement handler + result rows so admins (dba/system) can move group cursors. Wire into stream handler registry.
- Harden file download handler: rename path param to stored_name, validate/guess content type from stored_name, sanitize disposition, and restrict cross-user raw file downloads to dba/system via explicit authorization helper. Update DownloadQuery docs and tests.
- Improve health/job maintenance: add lower-frequency idle trim guard, epoch helper, and avoid unnecessary leadership polling in single-node mode; increase idle poll max and reduce unnecessary ticks.
- Minor cleanups and tests: small formatting fixes in backup/restore tests, add topic models (selectors/response), and update routes, models, and related tests and SDK/UI files.

These changes add admin tooling for topic maintenance, tighten file download security/authorization, and reduce background wakeups in non-cluster deployments.

Author: jamals86

backend/crates/kalamdb-api/src/http/files/download.rs

@@ -4,20 +4,26 @@ use std::sync::Arc;
 
 use actix_web::{get, web, HttpResponse, Responder};
 use kalamdb_auth::AuthSessionExtractor;
-use kalamdb_commons::{models::TableId, schemas::TableType, TableAccess};
+use kalamdb_commons::{
+    models::{TableId, UserId},
+    schemas::TableType,
+    Role, TableAccess,
+};
 use kalamdb_core::app_context::AppContext;
-use kalamdb_session::{can_access_shared_table, can_impersonate_target_user, AuthSession};
+use kalamdb_session::{
+    can_access_shared_table, can_access_user_table, can_impersonate_target_user, AuthSession,
+};
 use kalamdb_system::FileRef;
 
 use super::models::DownloadQuery;
 use crate::http::sql::models::{ErrorCode, SqlResponse};
 
-/// GET /v1/files/{namespace}/{table_name}/{subfolder}/{file_id} - Download a file
+/// GET /v1/files/{namespace}/{table_name}/{subfolder}/{stored_name} - Download a file
 ///
 /// Requires Bearer token (JWT) authorization and table access permissions.
 /// For user tables, downloads default to the authenticated user's table scope.
-/// Higher roles may supply `user_id` when the impersonation role matrix allows it.
-#[get("/files/{namespace}/{table_name}/{subfolder}/{file_id}")]
+/// DBA/system roles may supply `user_id` when the impersonation role matrix allows it.
+#[get("/files/{namespace}/{table_name}/{subfolder}/{stored_name}")]
 pub async fn download_file(
     extractor: AuthSessionExtractor,
     path: web::Path<(String, String, String, String)>,
@@ -27,7 +33,7 @@ pub async fn download_file(
     // Convert extractor to AuthSession
     let session: AuthSession = extractor.into();
 
-    let (namespace, table_name, subfolder, file_id) = path.into_inner();
+    let (namespace, table_name, subfolder, stored_name) = path.into_inner();
     let table_id = TableId::from_strings(&namespace, &table_name);
 
     // Look up table definition from schema registry
@@ -46,31 +52,29 @@ pub async fn download_file(
 
     let user_id = match table_type {
         TableType::User => {
+            if !can_access_user_table(session.role()) {
+                return HttpResponse::Forbidden().json(SqlResponse::error(
+                    ErrorCode::PermissionDenied,
+                    "User table file downloads require user-table access",
+                    0.0,
+                ));
+            }
+
             let effective_user_id = if let Some(requested_user_id) = query.user_id.as_ref() {
-                if requested_user_id == session.user_id() {
-                    requested_user_id.clone()
-                } else {
-                    let requested_user_id = requested_user_id.clone();
-                    let target_role = app_context
-                        .system_tables()
-                        .users()
-                        .role_for_impersonation_target(&requested_user_id);
-
-                    if !can_impersonate_target_user(
-                        session.user_id(),
-                        session.role(),
-                        &requested_user_id,
-                        target_role,
-                    ) {
-                        return HttpResponse::Forbidden().json(SqlResponse::error(
-                            ErrorCode::PermissionDenied,
-                            "Requested user is not allowed for the current role",
-                            0.0,
-                        ));
-                    }
-
-                    requested_user_id
+                let target_role = app_context
+                    .system_tables()
+                    .users()
+                    .role_for_impersonation_target(requested_user_id);
+
+                if !can_download_user_file_for_target(&session, requested_user_id, target_role) {
+                    return HttpResponse::Forbidden().json(SqlResponse::error(
+                        ErrorCode::PermissionDenied,
+                        "Requested user is not allowed for file download",
+                        0.0,
+                    ));
                 }
+
+                requested_user_id.clone()
             } else {
                 session.user_id().clone()
             };
@@ -113,18 +117,18 @@ pub async fn download_file(
         || subfolder.contains('/')
         || subfolder.contains('\\')
         || subfolder.contains('\0')
-        || file_id.contains("..")
-        || file_id.contains('/')
-        || file_id.contains('\\')
-        || file_id.contains('\0')
+        || stored_name.contains("..")
+        || stored_name.contains('/')
+        || stored_name.contains('\\')
+        || stored_name.contains('\0')
     {
         return HttpResponse::BadRequest().json(SqlResponse::error(
             ErrorCode::InvalidInput,
             "Invalid file path",
             0.0,
         ));
     }
-    let relative_path = format!("{}/{}", subfolder, file_id);
+    let relative_path = format!("{}/{}", subfolder, stored_name);
 
     // Fetch file from storage
     let file_service = app_context.file_storage_service();
@@ -134,25 +138,25 @@ pub async fn download_file(
     {
         Ok(data) => {
             // TODO: Get content type from the stored file metadata
-            // Guess content type from file extension in file_id
-            let content_type = guess_content_type(&file_id);
+            // Guess content type from file extension in stored_name
+            let content_type = guess_content_type(&stored_name);
 
-            // SECURITY: Sanitize file_id for Content-Disposition header to prevent
+            // SECURITY: Sanitize stored_name for Content-Disposition header to prevent
             // HTTP response header injection (CRLF injection) via crafted filenames.
-            let safe_file_id: String = file_id
+            let safe_stored_name: String = stored_name
                 .chars()
                 .filter(|c| *c != '"' && *c != '\r' && *c != '\n' && *c != '\0')
                 .collect();
             HttpResponse::Ok()
                 .content_type(content_type)
                 .append_header((
                     "Content-Disposition",
-                    format!("inline; filename=\"{}\"", safe_file_id),
+                    format!("inline; filename=\"{}\"", safe_stored_name),
                 ))
                 .body(data)
         },
         Err(e) => {
-            log::warn!("File download failed: table={}, file={}: {}", table_id, file_id, e);
+            log::warn!("File download failed: table={}, file={}: {}", table_id, stored_name, e);
             HttpResponse::NotFound().json(serde_json::json!({
                 "error": "File not found",
                 "code": "FILE_NOT_FOUND",
@@ -161,26 +165,45 @@ pub async fn download_file(
     }
 }
 
-fn guess_content_type(file_id: &str) -> String {
-    mime_guess::from_path(file_id).first_or_octet_stream().to_string()
+fn guess_content_type(stored_name: &str) -> String {
+    mime_guess::from_path(stored_name).first_or_octet_stream().to_string()
+}
+
+fn can_download_user_file_for_target(
+    session: &AuthSession,
+    requested_user_id: &UserId,
+    target_role: Role,
+) -> bool {
+    if requested_user_id == session.user_id() {
+        return true;
+    }
+
+    matches!(session.role(), Role::System | Role::Dba)
+        && can_impersonate_target_user(
+            session.user_id(),
+            session.role(),
+            requested_user_id,
+            target_role,
+        )
 }
 
 #[cfg(test)]
 mod tests {
-    use kalamdb_commons::models::UserId;
-    use kalamdb_commons::Role;
-
     use super::*;
 
     #[test]
-    fn download_user_id_query_uses_shared_impersonation_authorization() {
-        let actor = UserId::new("svc");
-        let same_user = UserId::new("svc");
+    fn download_user_id_query_is_self_or_admin_only() {
+        let service = AuthSession::new(UserId::new("svc"), Role::Service);
+        let dba = AuthSession::new(UserId::new("dba"), Role::Dba);
+        let system = AuthSession::new(UserId::new("system"), Role::System);
+        let user = AuthSession::new(UserId::new("alice"), Role::User);
         let regular_target = UserId::new("alice");
         let dba_target = UserId::new("dba-target");
 
-        assert!(can_impersonate_target_user(&actor, Role::Service, &same_user, Role::Service));
-        assert!(can_impersonate_target_user(&actor, Role::Service, &regular_target, Role::User));
-        assert!(!can_impersonate_target_user(&actor, Role::Service, &dba_target, Role::Dba));
+        assert!(can_download_user_file_for_target(&user, &regular_target, Role::User));
+        assert!(!can_download_user_file_for_target(&service, &regular_target, Role::User));
+        assert!(can_download_user_file_for_target(&dba, &regular_target, Role::User));
+        assert!(can_download_user_file_for_target(&dba, &dba_target, Role::Dba));
+        assert!(can_download_user_file_for_target(&system, &dba_target, Role::Dba));
     }
 }

backend/crates/kalamdb-api/src/http/files/models/download_query.rs

@@ -7,7 +7,7 @@ use serde::Deserialize;
 #[derive(Debug, Deserialize)]
 pub struct DownloadQuery {
     /// Optional user_id for user-table downloads.
-    /// Cross-user requests are authorized through the impersonation role matrix.
+    /// Cross-user raw byte downloads are limited to dba/system roles.
     #[serde(default, deserialize_with = "deserialize_optional_user_id")]
     pub user_id: Option<UserId>,
 }

backend/crates/kalamdb-api/src/http/topics/consume.rs

@@ -59,7 +59,7 @@ pub async fn consume_handler(
     }
 
     let topic_id = &body.topic_id;
-    let group_id = &body.group_id;
+    let group_id = body.group_id.as_ref();
 
     // Verify topic exists
     let topics_provider = app_context.system_tables().topics();
@@ -82,19 +82,22 @@ pub async fn consume_handler(
 
     // Determine start offset based on position.
     //
-    // All positions first check the consumer group's committed offset.
-    // If a committed offset exists, we resume from there (last_acked + 1).
-    // The position only matters when no offset has been committed yet:
+    // Grouped reads first check the consumer group's committed offset. If a
+    // committed offset exists, we resume from there (last_acked + 1). The
+    // requested position only matters when no offset has been committed yet.
+    // Stateless reads do not check or create group state, so the requested
+    // position is honored on every call:
     //   - Earliest: start from offset 0 (replay all history)
     //   - Latest: start from high-water mark (last offset + 1)
     //   - Offset: start from the explicit offset
-    let committed_offset =
+    let committed_offset = group_id.and_then(|group_id| {
         topic_publisher.get_group_offsets(topic_id, group_id).ok().and_then(|offsets| {
             offsets
                 .iter()
                 .find(|o| o.partition_id == body.partition_id)
                 .map(|o| o.last_acked_offset + 1)
-        });
+        })
+    });
 
     let start_offset = match committed_offset {
         Some(committed) => committed,
@@ -118,14 +121,25 @@ pub async fn consume_handler(
         },
     };
 
-    // Fetch messages
-    let messages = match topic_publisher.fetch_messages_for_group(
-        topic_id,
-        group_id,
-        body.partition_id,
-        start_offset,
-        body.limit as usize,
-    ) {
+    // Fetch messages.
+    let messages_result = if let Some(group_id) = group_id {
+        topic_publisher.fetch_messages_for_group(
+            topic_id,
+            group_id,
+            body.partition_id,
+            start_offset,
+            body.limit as usize,
+        )
+    } else {
+        topic_publisher.fetch_messages(
+            topic_id,
+            body.partition_id,
+            start_offset,
+            body.limit as usize,
+        )
+    };
+
+    let messages = match messages_result {
         Ok(msgs) => msgs,
         Err(e) => {
             return HttpResponse::InternalServerError().json(TopicErrorResponse::internal_error(

backend/crates/kalamdb-api/src/http/topics/latest_offsets.rs

@@ -0,0 +1,83 @@
+//! Topic latest offsets handler
+//!
+//! POST /v1/api/topics/latest-offsets - Resolve topic partition head offsets
+
+use std::collections::BTreeSet;
+use std::sync::Arc;
+
+use actix_web::{post, web, HttpResponse, Responder};
+use kalamdb_auth::AuthSessionExtractor;
+use kalamdb_commons::Role;
+use kalamdb_core::app_context::AppContext;
+use kalamdb_session::AuthSession;
+
+use super::models::{
+    LatestOffsetsRequest, LatestOffsetsResponse, TopicErrorResponse, TopicPartitionLatestOffset,
+};
+
+/// Check if role is allowed to resolve topic offsets.
+/// Must be service, dba, or system role (NOT user)
+fn is_topic_authorized(session: &AuthSession) -> bool {
+    matches!(session.role(), Role::Service | Role::Dba | Role::System)
+}
+
+/// POST /v1/api/topics/latest-offsets - Resolve topic partition head offsets
+///
+/// # Authentication
+/// Requires Bearer token authentication.
+///
+/// # Authorization
+/// Role must be `service`, `dba`, or `system` (NOT `user`).
+#[post("/latest-offsets")]
+pub async fn latest_offsets_handler(
+    extractor: AuthSessionExtractor,
+    body: web::Json<LatestOffsetsRequest>,
+    app_context: web::Data<Arc<AppContext>>,
+) -> impl Responder {
+    let session: AuthSession = extractor.into();
+
+    if !is_topic_authorized(&session) {
+        return HttpResponse::Forbidden().json(TopicErrorResponse::forbidden(
+            "Topic offset inspection requires service, dba, or system role",
+        ));
+    }
+
+    let topic_publisher = app_context.topic_publisher();
+    let mut seen = BTreeSet::new();
+    let mut offsets = Vec::with_capacity(body.partitions.len());
+
+    for selector in &body.partitions {
+        let dedupe_key = (selector.topic_id.to_string(), selector.partition_id);
+        if !seen.insert(dedupe_key) {
+            continue;
+        }
+
+        let last_offset = match topic_publisher.latest_offset(&selector.topic_id, selector.partition_id)
+        {
+            Ok(offset) => offset,
+            Err(error) => {
+                return HttpResponse::InternalServerError().json(TopicErrorResponse::internal_error(
+                    &format!("Failed to resolve latest offset: {}", error),
+                ));
+            },
+        };
+
+        offsets.push(TopicPartitionLatestOffset {
+            topic_id: selector.topic_id.clone(),
+            partition_id: selector.partition_id,
+            next_offset: last_offset.map(|offset| offset + 1).unwrap_or(0),
+            last_offset,
+        });
+    }
+
+    offsets.sort_by(|left, right| {
+        let topic_compare = left.topic_id.to_string().cmp(&right.topic_id.to_string());
+        if topic_compare == std::cmp::Ordering::Equal {
+            left.partition_id.cmp(&right.partition_id)
+        } else {
+            topic_compare
+        }
+    });
+
+    HttpResponse::Ok().json(LatestOffsetsResponse { offsets })
+}

backend/crates/kalamdb-api/src/http/topics/mod.rs

@@ -6,13 +6,16 @@
 //! ## Endpoints
 //! - POST /v1/api/topics/consume - Consume messages from a topic
 //! - POST /v1/api/topics/ack - Acknowledge offset for consumer group
+//! - POST /v1/api/topics/latest-offsets - Resolve topic partition head offsets
 //!
 //! **Authorization**: Endpoints require `service`, `dba`, or `system` role (NOT `user`).
 
 pub mod models;
 
 mod ack;
 mod consume;
+mod latest_offsets;
 
 pub(crate) use ack::ack_handler;
 pub(crate) use consume::consume_handler;
+pub(crate) use latest_offsets::latest_offsets_handler;