jammi-server 0.35.0

//! `CatalogService` gRPC implementation: the engine's single control-plane
//! surface.
//!
//! Every catalog / metadata / lifecycle / observability verb a remote client
//! reaches lands here, folding what were five straddling services
//! (session / sources+models-on-embedding / channel / mutable-table /
//! topic-admin-on-trigger) into one. The verbs split into two backings:
//!
//! * **Engine-free** — the tenant trio (`SetTenant` / `GetTenant` /
//!   `ClearTenant`) and the `GetServerInfo` handshake update / read the
//!   in-process [`SessionStore`] and report compile-time + runtime tier facts.
//!   They need no engine, so they answer even on an engine-light deployment.
//! * **Engine-backed** — sources / models / channels / mutable tables / topics
//!   delegate 1:1 to the transport-agnostic [`Session`]
//!   abstraction (`session_tenant_traced(&request)` → [`scoped`] → [`map_engine_error`]
//!   / [`map_trigger_error`]), exactly mirroring the handlers they were lifted
//!   from. These require an engine; an engine-light deployment returns a
//!   truthful `Unavailable` rather than a faked empty result.
//!
//! Tenant scope is read from the request's [`SessionTenant`] extension (set
//! upstream by [`crate::grpc::session::TenantInterceptor`]).
//!
//! [`SessionTenant`]: crate::grpc::session::SessionTenant
//! [`SessionStore`]: crate::grpc::session::SessionStore

use std::str::FromStr;
use std::sync::Arc;

use jammi_ai::session::InferenceSession;
use jammi_ai::Session;
use jammi_db::source::SourceConnection;
use jammi_db::trigger::ids::TopicId;
use jammi_db::trigger::TriggerError;
use jammi_db::TenantId;
use jammi_wire::{
    channel_to_proto, definition_to_proto, derives_from_edge_to_proto, match_verdict_to_proto,
    model_to_proto, parse_table_id, source_type_from_proto, staleness_to_proto, topic_to_proto,
};
use tonic::{Request, Response, Status};

use crate::grpc::proto::catalog as pb;
use crate::grpc::proto::catalog::catalog_service_server::CatalogService;
use crate::grpc::session::{SessionId, SessionStore, SESSION_HEADER};
use crate::grpc::wire::{
    map_engine_error, map_trigger_error, require_nonempty, scoped, session_tenant_traced,
};
use crate::tiers::TierSet;

/// Server-side handler for the control-plane gRPC surface.
///
/// Holds the in-process [`SessionStore`] + mounted [`TierSet`] that back the
/// engine-free verbs (the tenant trio and `GetServerInfo`), and an optional
/// shared engine session that backs the catalog / lifecycle verbs. The engine
/// is `None` on an engine-light deployment (one that mounts the control plane
/// but no data-plane engine); those verbs then report a truthful `Unavailable`.
pub struct CatalogServer {
    store: SessionStore,
    tiers: TierSet,
    session: Option<Arc<InferenceSession>>,
}

impl CatalogServer {
    pub fn new(
        store: SessionStore,
        tiers: TierSet,
        session: Option<Arc<InferenceSession>>,
    ) -> Self {
        Self {
            store,
            tiers,
            session,
        }
    }

    /// The shared engine session backing the catalog / lifecycle verbs, or a
    /// truthful `Unavailable` when this deployment mounts no engine.
    fn engine(&self) -> Result<&Arc<InferenceSession>, Status> {
        self.session.as_ref().ok_or_else(|| {
            Status::unavailable("catalog engine verbs are not enabled on this deployment")
        })
    }

    /// A [`Session`] over the shared engine; see [`crate::grpc::inference`] for
    /// the tenant-scope wiring rationale.
    fn local(&self) -> Result<Session, Status> {
        Ok(Session::new(Arc::clone(self.engine()?)))
    }
}

#[tonic::async_trait]
impl CatalogService for CatalogServer {
    // --- session -----------------------------------------------------------

    async fn set_tenant(
        &self,
        request: Request<pb::SetTenantRequest>,
    ) -> Result<Response<()>, Status> {
        let session = session_id_from_request(&request)?;
        let body = request.into_inner();
        let tenant = body
            .tenant
            .ok_or_else(|| Status::invalid_argument("tenant field is required"))?;
        let parsed = parse_tenant(&tenant)?;
        self.store.set(session, parsed);
        Ok(Response::new(()))
    }

    async fn get_tenant(
        &self,
        request: Request<()>,
    ) -> Result<Response<pb::GetTenantResponse>, Status> {
        let session = session_id_from_request(&request)?;
        let tenant = self.store.get(&session);
        Ok(Response::new(pb::GetTenantResponse {
            tenant: Some(pb::Tenant {
                id: tenant.map(|t| t.to_string()).unwrap_or_default(),
            }),
        }))
    }

    async fn clear_tenant(&self, request: Request<()>) -> Result<Response<()>, Status> {
        let session = session_id_from_request(&request)?;
        self.store.clear(&session);
        Ok(Response::new(()))
    }

    async fn get_server_info(
        &self,
        _request: Request<()>,
    ) -> Result<Response<pb::ServerInfo>, Status> {
        // The build's compile-time self-description (`version` / `features` /
        // `storage_backends`) plus this deployment's *runtime* tier handshake:
        // `services` is the set of gRPC tiers this server actually mounted, not
        // a compile-time fact — so it comes from the injected `TierSet`, not
        // from `ServerInfo::current` (which leaves it empty).
        let info = jammi_db::ServerInfo::current();
        Ok(Response::new(pb::ServerInfo {
            version: info.version,
            features: info.features,
            storage_backends: info.storage_backends,
            services: self.tiers.as_wire(),
        }))
    }

    // --- sources / models --------------------------------------------------

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn add_source(
        &self,
        request: Request<pb::AddSourceRequest>,
    ) -> Result<Response<()>, Status> {
        let tenant = session_tenant_traced(&request);
        let req = request.into_inner();
        require_nonempty(&req.source_id, "source_id")?;
        let source_type = source_type_from_proto(req.source_kind)?;
        let connection: SourceConnection = req
            .connection
            .ok_or_else(|| Status::invalid_argument("connection is required"))?
            .try_into()?;
        let session = self.local()?;

        scoped(self.engine()?, tenant, || {
            session.add_source(&req.source_id, source_type, connection)
        })
        .await
        .map_err(map_engine_error)?;
        Ok(Response::new(()))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn remove_source(
        &self,
        request: Request<pb::RemoveSourceRequest>,
    ) -> Result<Response<()>, Status> {
        let tenant = session_tenant_traced(&request);
        let req = request.into_inner();
        require_nonempty(&req.source_id, "source_id")?;
        let session = self.local()?;

        scoped(self.engine()?, tenant, || {
            session.remove_source(&req.source_id)
        })
        .await
        .map_err(map_engine_error)?;
        Ok(Response::new(()))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn list_sources(
        &self,
        request: Request<pb::ListSourcesRequest>,
    ) -> Result<Response<pb::ListSourcesResponse>, Status> {
        let tenant = session_tenant_traced(&request);
        let session = self.local()?;

        let descriptors = scoped(self.engine()?, tenant, || session.list_sources())
            .await
            .map_err(map_engine_error)?;

        let sources = descriptors
            .into_iter()
            .map(pb::SourceDescriptor::from)
            .collect();
        Ok(Response::new(pb::ListSourcesResponse { sources }))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn describe_source(
        &self,
        request: Request<pb::DescribeSourceRequest>,
    ) -> Result<Response<pb::SourceDescriptor>, Status> {
        let tenant = session_tenant_traced(&request);
        let req = request.into_inner();
        require_nonempty(&req.source_id, "source_id")?;
        let session = self.local()?;

        // An absent source returns NotFound — the truthful "no such source for
        // this tenant" the remote arm maps back to `None`, never a faked empty
        // descriptor.
        let descriptor = scoped(self.engine()?, tenant, || {
            session.describe_source(&req.source_id)
        })
        .await
        .map_err(map_engine_error)?
        .ok_or_else(|| Status::not_found(format!("source '{}' not found", req.source_id)))?;

        Ok(Response::new(pb::SourceDescriptor::from(descriptor)))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn list_models(
        &self,
        request: Request<pb::ListModelsRequest>,
    ) -> Result<Response<pb::ListModelsResponse>, Status> {
        let tenant = session_tenant_traced(&request);
        let session = self.local()?;

        let records = scoped(self.engine()?, tenant, || session.list_models())
            .await
            .map_err(map_engine_error)?;

        let models = records.iter().map(model_to_proto).collect();
        Ok(Response::new(pb::ListModelsResponse { models }))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn describe_model(
        &self,
        request: Request<pb::DescribeModelRequest>,
    ) -> Result<Response<pb::Model>, Status> {
        let tenant = session_tenant_traced(&request);
        let req = request.into_inner();
        require_nonempty(&req.model_id, "model_id")?;
        let session = self.local()?;

        // An absent model returns NotFound — the truthful "no such model for this
        // tenant" the remote arm maps back to `None`, never a faked record.
        let record = scoped(self.engine()?, tenant, || {
            session.describe_model(&req.model_id)
        })
        .await
        .map_err(map_engine_error)?
        .ok_or_else(|| Status::not_found(format!("model '{}' not found", req.model_id)))?;

        Ok(Response::new(model_to_proto(&record)))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn delete_model(
        &self,
        request: Request<pb::DeleteModelRequest>,
    ) -> Result<Response<()>, Status> {
        let tenant = session_tenant_traced(&request);
        let req = request.into_inner();
        require_nonempty(&req.model_id, "model_id")?;
        let session = self.local()?;

        // A model outside the caller's scope (absent without `if_exists`, or a
        // GLOBAL model a tenant session cannot delete) surfaces as
        // `JammiError::ModelNotFound` — a NotFound at the wire. A still-referenced
        // model surfaces as `JammiError::ModelReferenced`, which `map_engine_error`
        // renders as `FailedPrecondition`. Every fault keeps its faithful
        // `map_engine_error` mapping.
        scoped(self.engine()?, tenant, || {
            session.delete_model(&req.model_id, req.version, req.if_exists)
        })
        .await
        .map_err(map_engine_error)?;
        Ok(Response::new(()))
    }

    // --- channels ----------------------------------------------------------

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn register_channel(
        &self,
        request: Request<pb::RegisterChannelRequest>,
    ) -> Result<Response<()>, Status> {
        let tenant = session_tenant_traced(&request);
        // Decode through the shared `jammi_ai::wire` seam — the same decode the
        // embedded binding's `_register_channel_proto` drives — so both transports
        // validate and submit an identical request.
        let spec = jammi_ai::wire::register_channel_from_proto(request.into_inner())?;
        let session = self.local()?;

        scoped(self.engine()?, tenant, || session.register_channel(&spec))
            .await
            .map_err(map_engine_error)?;

        Ok(Response::new(()))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn add_channel_columns(
        &self,
        request: Request<pb::AddChannelColumnsRequest>,
    ) -> Result<Response<()>, Status> {
        let tenant = session_tenant_traced(&request);
        // Decode through the shared `jammi_ai::wire` seam — the same decode the
        // embedded binding's `_add_channel_columns_proto` drives — so both
        // transports validate and submit an identical request.
        let args = jammi_ai::wire::add_channel_columns_from_proto(request.into_inner())?;
        let session = self.local()?;

        scoped(self.engine()?, tenant, || {
            session.add_channel_columns(&args.id, &args.columns)
        })
        .await
        .map_err(map_engine_error)?;

        Ok(Response::new(()))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn list_channels(
        &self,
        request: Request<pb::ListChannelsRequest>,
    ) -> Result<Response<pb::ListChannelsResponse>, Status> {
        let tenant = session_tenant_traced(&request);
        let session = self.local()?;

        let specs = scoped(self.engine()?, tenant, || session.list_channels())
            .await
            .map_err(map_engine_error)?;

        let channels = specs.iter().map(channel_to_proto).collect();
        Ok(Response::new(pb::ListChannelsResponse { channels }))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn verify_materialization(
        &self,
        request: Request<pb::VerifyMaterializationRequest>,
    ) -> Result<Response<pb::VerifyMaterializationResponse>, Status> {
        let tenant = session_tenant_traced(&request);
        let session = self.local()?;
        let req = request.into_inner();
        let table = req.table;
        let expected = req
            .expected_definition
            .map(jammi_db::store::manifest::DefinitionHash);

        let verdict = scoped(self.engine()?, tenant, || {
            session.verify_materialization(&table, expected.clone())
        })
        .await
        .map_err(map_engine_error)?;

        Ok(Response::new(pb::VerifyMaterializationResponse {
            verdict: Some(match_verdict_to_proto(verdict)),
        }))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn staleness(
        &self,
        request: Request<pb::StalenessRequest>,
    ) -> Result<Response<pb::StalenessResponse>, Status> {
        let tenant = session_tenant_traced(&request);
        let session = self.local()?;
        let req = request.into_inner();
        let table = req.table;
        let current = jammi_db::store::manifest::DefinitionHash(req.current_definition);

        let verdict = scoped(self.engine()?, tenant, || {
            session.staleness(&table, current.clone())
        })
        .await
        .map_err(map_engine_error)?;

        Ok(Response::new(pb::StalenessResponse {
            staleness: Some(staleness_to_proto(verdict)),
        }))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn derives_from(
        &self,
        request: Request<pb::DerivesFromRequest>,
    ) -> Result<Response<pb::DerivesFromResponse>, Status> {
        let tenant = session_tenant_traced(&request);
        let session = self.local()?;
        let req = request.into_inner();
        let table = req.table;

        let edges = scoped(self.engine()?, tenant, || session.derives_from(&table))
            .await
            .map_err(map_engine_error)?;

        Ok(Response::new(pb::DerivesFromResponse {
            edges: edges.into_iter().map(derives_from_edge_to_proto).collect(),
        }))
    }

    // --- mutable tables ----------------------------------------------------

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn create_mutable_table(
        &self,
        request: Request<pb::CreateMutableTableRequest>,
    ) -> Result<Response<pb::CreateMutableTableResponse>, Status> {
        let tenant = session_tenant_traced(&request);
        // Decode through the shared `jammi_ai::wire` seam — the same decode the
        // embedded binding's `_create_mutable_table_proto` drives — so both
        // transports validate and submit an identical request. The wire body is
        // tenant-free; the decode stamps the session's resolved tenant.
        let def = jammi_ai::wire::create_mutable_table_from_proto(request.into_inner(), tenant)?;
        let session = self.local()?;

        let id = scoped(self.engine()?, tenant, || session.create_mutable_table(def))
            .await
            .map_err(map_engine_error)?;

        Ok(Response::new(pb::CreateMutableTableResponse {
            mutable_table_id: id.to_string(),
        }))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn drop_mutable_table(
        &self,
        request: Request<pb::DropMutableTableRequest>,
    ) -> Result<Response<()>, Status> {
        let tenant = session_tenant_traced(&request);
        let req = request.into_inner();
        let id = parse_table_id(&req.mutable_table_id)?;
        let session = self.local()?;

        scoped(self.engine()?, tenant, || session.drop_mutable_table(&id))
            .await
            .map_err(map_engine_error)?;

        Ok(Response::new(()))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn list_mutable_tables(
        &self,
        request: Request<pb::ListMutableTablesRequest>,
    ) -> Result<Response<pb::ListMutableTablesResponse>, Status> {
        let tenant = session_tenant_traced(&request);
        let session = self.local()?;

        let defs = scoped(self.engine()?, tenant, || session.list_mutable_tables())
            .await
            .map_err(map_engine_error)?;

        // The wire body stays tenant-free (the catalog row's tenant is the
        // session scope, not a message field), so each definition encodes the
        // same projection the create path carries.
        let definitions = defs
            .iter()
            .map(definition_to_proto)
            .collect::<Result<Vec<_>, Status>>()?;
        Ok(Response::new(pb::ListMutableTablesResponse { definitions }))
    }

    // --- topics ------------------------------------------------------------

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn register_topic(
        &self,
        request: Request<pb::RegisterTopicRequest>,
    ) -> Result<Response<pb::RegisterTopicResponse>, Status> {
        let tenant = session_tenant_traced(&request);
        // Decode through the shared `jammi_ai::wire` seam — the same decode the
        // embedded binding's `_register_topic_proto` drives — so both transports
        // validate and submit an identical request. The wire body is tenant-free;
        // the decode stamps the session's resolved tenant and mints the topic id
        // server-side (the seam ignores any caller-supplied `topic_id`, so a
        // caller cannot replay a known UUID to PK-collide another tenant's
        // registration). The minted id is returned below.
        let topic = jammi_ai::wire::register_topic_from_proto(request.into_inner(), tenant)?;
        let session = self.local()?;

        // `register_topic` dual-registers the broker driver and the catalog (so a
        // later `publish` resolves the topic) — the engine path owns that
        // invariant, not the handler.
        scoped(self.engine()?, tenant, || session.register_topic(&topic))
            .await
            .map_err(map_trigger_error)?;

        Ok(Response::new(pb::RegisterTopicResponse {
            topic_id: topic.id.to_string(),
        }))
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn drop_topic(
        &self,
        request: Request<pb::DropTopicRequest>,
    ) -> Result<Response<()>, Status> {
        let tenant = session_tenant_traced(&request);
        let req = request.into_inner();
        if req.topic_id.is_empty() {
            return Err(Status::invalid_argument("topic_id is required"));
        }
        let topic_id = TopicId::from_str(&req.topic_id)
            .map_err(|e| Status::invalid_argument(format!("invalid topic_id: {e}")))?;
        let session = self.local()?;

        // The engine resolves the topic by id (tenant-scoped) and returns
        // `TopicNotFound` when it is absent; `if_exists` turns that into a no-op.
        match scoped(self.engine()?, tenant, || session.drop_topic(topic_id)).await {
            Ok(()) => Ok(Response::new(())),
            Err(TriggerError::TopicNotFound(_)) if req.if_exists => Ok(Response::new(())),
            Err(e) => Err(map_trigger_error(e)),
        }
    }

    #[tracing::instrument(skip(self, request), fields(tenant_id = tracing::field::Empty))]
    async fn list_topics(
        &self,
        request: Request<pb::ListTopicsRequest>,
    ) -> Result<Response<pb::ListTopicsResponse>, Status> {
        let tenant = session_tenant_traced(&request);
        let session = self.local()?;

        let topics = scoped(self.engine()?, tenant, || session.list_topics())
            .await
            .map_err(map_trigger_error)?;
        let topics = topics
            .iter()
            .map(topic_to_proto)
            .collect::<Result<Vec<_>, Status>>()?;
        Ok(Response::new(pb::ListTopicsResponse {
            topics,
            // Pagination is not yet implemented — `next_page_token` empty
            // means "this is the complete result set."
            next_page_token: String::new(),
        }))
    }
}

/// Extract the session id and return a typed `InvalidArgument` error if it's
/// missing. Used by the tenant verbs that read or mutate session state directly.
fn session_id_from_request<T>(request: &Request<T>) -> Result<SessionId, Status> {
    request
        .metadata()
        .get(SESSION_HEADER)
        .and_then(|v| v.to_str().ok())
        .map(SessionId::new)
        .ok_or_else(|| {
            Status::invalid_argument(format!(
                "request missing required `{SESSION_HEADER}` metadata header"
            ))
        })
}

/// Parse a wire-format [`pb::Tenant`] into the engine's [`TenantId`] newtype.
/// Empty string is interpreted as "no tenant" → `Ok(None)`. Any other value
/// must parse as a non-nil UUID per ADR-00.
fn parse_tenant(t: &pb::Tenant) -> Result<Option<TenantId>, Status> {
    if t.id.is_empty() {
        return Ok(None);
    }
    TenantId::from_str(&t.id)
        .map(Some)
        .map_err(|e| Status::invalid_argument(format!("invalid tenant id: {e}")))
}