use std::sync::Arc;
use sqlx::PgPool;
use tonic::{Request, Response, Status};
use crate::generation::CatalogManifest;
use crate::metrics::{MetricsRecorder, NoopMetrics};
use crate::proto::udb::core::tenant::services::v1 as tenant_pb;
use crate::proto::udb::core::tenant::services::v1::tenant_service_server::TenantService;
use crate::runtime::DataBrokerRuntime;
use crate::runtime::channels::ChannelManager;
pub use crate::proto::udb::core::tenant::services::v1::tenant_service_server::TenantServiceServer;
use super::DataBrokerService;
mod config;
mod errors;
mod events;
mod handlers;
mod model;
mod store;
#[cfg(test)]
mod tests;
pub struct TenantServiceImpl {
pub(crate) pg_pool: Option<PgPool>,
pub(crate) runtime: Option<Arc<DataBrokerRuntime>>,
pub(crate) channels: Option<ChannelManager>,
pub(crate) metrics: Arc<dyn MetricsRecorder>,
pub(crate) outbox_relation: Option<String>,
#[cfg(feature = "redis")]
pub(crate) jti_denylist: Option<crate::runtime::authn::revocation::JtiDenylist>,
pub(crate) manifest: Option<CatalogManifest>,
}
impl TenantServiceImpl {
pub fn new() -> Self {
Self {
pg_pool: None,
runtime: None,
channels: None,
metrics: Arc::new(NoopMetrics),
outbox_relation: None,
#[cfg(feature = "redis")]
jti_denylist: None,
manifest: None,
}
}
pub(crate) fn with_manifest(mut self, manifest: Option<CatalogManifest>) -> Self {
self.manifest = manifest;
self
}
pub(crate) fn require_manifest(&self) -> Result<&CatalogManifest, Status> {
self.manifest.as_ref().ok_or_else(|| {
errors::tenant_capability_status(
"purge_tenant",
"catalog_manifest",
"tenant service requires the catalog manifest for purge",
)
})
}
pub fn with_postgres(mut self, pool: Option<PgPool>) -> Self {
self.pg_pool = pool;
self
}
pub(crate) fn with_runtime(mut self, runtime: Option<Arc<DataBrokerRuntime>>) -> Self {
self.runtime = runtime;
self
}
pub(crate) fn require_runtime(&self) -> Result<&DataBrokerRuntime, Status> {
self.runtime.as_deref().ok_or_else(|| {
errors::tenant_capability_status(
"native_entity_dispatch",
"runtime_native_entity_dispatch",
"tenant service requires runtime native entity dispatch",
)
})
}
pub(crate) fn with_metrics(mut self, metrics: Arc<dyn MetricsRecorder>) -> Self {
self.metrics = metrics;
self
}
pub(crate) fn with_outbox(mut self, relation: Option<String>) -> Self {
self.outbox_relation = relation;
self
}
#[cfg(feature = "redis")]
pub(crate) fn with_jti_denylist(
mut self,
denylist: Option<crate::runtime::authn::revocation::JtiDenylist>,
) -> Self {
self.jti_denylist = denylist;
self
}
pub(crate) fn with_channels(mut self, channels: Option<ChannelManager>) -> Self {
self.channels = channels;
self
}
pub(crate) fn require_pool(&self) -> Result<&PgPool, Status> {
self.pg_pool.as_ref().ok_or_else(|| {
errors::tenant_capability_status(
"postgres_store",
"postgres_store",
"tenant service requires a Postgres-backed store (no PG pool configured)",
)
})
}
}
impl Default for TenantServiceImpl {
fn default() -> Self {
Self::new()
}
}
#[tonic::async_trait]
impl TenantService for TenantServiceImpl {
async fn create_tenant(
&self,
request: Request<tenant_pb::CreateTenantRequest>,
) -> Result<Response<tenant_pb::CreateTenantResponse>, Status> {
handlers::create_tenant(self, request).await
}
async fn purge_tenant(
&self,
request: Request<tenant_pb::PurgeTenantRequest>,
) -> Result<Response<tenant_pb::PurgeTenantResponse>, Status> {
handlers::purge_tenant(self, request).await
}
async fn get_tenant(
&self,
request: Request<tenant_pb::GetTenantRequest>,
) -> Result<Response<tenant_pb::GetTenantResponse>, Status> {
handlers::get_tenant(self, request).await
}
async fn list_tenants(
&self,
request: Request<tenant_pb::ListTenantsRequest>,
) -> Result<Response<tenant_pb::ListTenantsResponse>, Status> {
handlers::list_tenants(self, request).await
}
async fn update_tenant(
&self,
request: Request<tenant_pb::UpdateTenantRequest>,
) -> Result<Response<tenant_pb::UpdateTenantResponse>, Status> {
handlers::update_tenant(self, request).await
}
async fn get_tenant_config(
&self,
request: Request<tenant_pb::GetTenantConfigRequest>,
) -> Result<Response<tenant_pb::GetTenantConfigResponse>, Status> {
handlers::get_tenant_config(self, request).await
}
async fn update_tenant_config(
&self,
request: Request<tenant_pb::UpdateTenantConfigRequest>,
) -> Result<Response<tenant_pb::UpdateTenantConfigResponse>, Status> {
handlers::update_tenant_config(self, request).await
}
}
impl DataBrokerService {
pub(crate) fn build_tenant_service(&self) -> TenantServiceImpl {
let runtime = self.runtime.load_full();
let pg_pool = runtime
.native_store_pool_for_service("tenant", true, "")
.ok();
let outbox = runtime.config().cdc.outbox_relation();
let channels = Some(runtime.channels().clone());
#[cfg(feature = "redis")]
let jti_denylist = runtime.redis_clone().map(|redis| {
crate::runtime::authn::revocation::JtiDenylist::new(
redis,
crate::runtime::security::SecurityConfig::current().jwt_access_ttl_secs,
)
});
let service = TenantServiceImpl::new()
.with_postgres(pg_pool)
.with_runtime(Some(runtime))
.with_channels(channels)
.with_metrics(self.metrics.clone())
.with_outbox(Some(outbox))
.with_manifest(Some(self.manifest.clone()));
#[cfg(feature = "redis")]
let service = service.with_jti_denylist(jti_denylist);
service
}
}