fix: Remove ApiEndpoint::name to be consistent with other kinds of sinks (#2316)

* fix: Remove `ApiEndpoint::name` to be consistent with other kinds of sinks

* fix e2e tests

* fix compilation

* fix test

Author: chubei

dozer-api/src/cache_builder/builder_impl.rs

@@ -131,9 +131,9 @@ impl CacheBuilderImpl {
             LogOperation::Op { op } => match op {
                 Operation::Delete { old } => {
                     if let Some(meta) = self.building.delete(&old)? {
-                        if let Some((endpoint_name, operations_sender)) = operations_sender {
+                        if let Some((table_name, operations_sender)) = operations_sender {
                             let operation = types_helper::map_delete_operation(
-                                endpoint_name.clone(),
+                                table_name.clone(),
                                 CacheRecord::new(meta.id, meta.version, old),
                             );
                             send_and_log_error(operations_sender, operation);
@@ -157,9 +157,9 @@ impl CacheBuilderImpl {
                     );
                     increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);
 
-                    if let Some((endpoint_name, operations_sender)) = operations_sender {
+                    if let Some((table_name, operations_sender)) = operations_sender {
                         send_upsert_result(
-                            endpoint_name,
+                            table_name,
                             operations_sender,
                             result,
                             &self.building.get_schema().0,
@@ -178,9 +178,9 @@ impl CacheBuilderImpl {
                     );
                     increment_counter!(CACHE_OPERATION_COUNTER_NAME, labels);
 
-                    if let Some((endpoint_name, operations_sender)) = operations_sender {
+                    if let Some((table_name, operations_sender)) = operations_sender {
                         send_upsert_result(
-                            endpoint_name,
+                            table_name,
                             operations_sender,
                             upsert_result,
                             &self.building.get_schema().0,
@@ -200,9 +200,9 @@ impl CacheBuilderImpl {
 
                     for record in new {
                         let upsert_result = self.building.insert(&record)?;
-                        if let Some((endpoint_name, operations_sender)) = operations_sender {
+                        if let Some((table_name, operations_sender)) = operations_sender {
                             send_upsert_result(
-                                endpoint_name,
+                                table_name,
                                 operations_sender,
                                 upsert_result,
                                 &self.building.get_schema().0,
@@ -276,7 +276,7 @@ fn snapshotting_str(snapshotting: bool) -> &'static str {
 }
 
 fn send_upsert_result(
-    endpoint_name: &str,
+    table_name: &str,
     operations_sender: &Sender<GrpcOperation>,
     upsert_result: UpsertResult,
     schema: &Schema,
@@ -286,7 +286,7 @@ fn send_upsert_result(
     match upsert_result {
         UpsertResult::Inserted { meta } => {
             let op = types_helper::map_insert_operation(
-                endpoint_name.to_string(),
+                table_name.to_string(),
                 CacheRecord::new(meta.id, meta.version, new),
             );
             send_and_log_error(operations_sender, op);
@@ -303,7 +303,7 @@ fn send_upsert_result(
                 record
             });
             let op = types_helper::map_update_operation(
-                endpoint_name.to_string(),
+                table_name.to_string(),
                 CacheRecord::new(old_meta.id, old_meta.version, old),
                 CacheRecord::new(new_meta.id, new_meta.version, new),
             );
@@ -365,7 +365,7 @@ mod tests {
 
     fn test_endpoint_meta(log_id: String) -> EndpointMeta {
         EndpointMeta {
-            name: Default::default(),
+            table_name: Default::default(),
             log_id,
             schema: EndpointSchema {
                 path: Default::default(),

dozer-api/src/cache_builder/endpoint_meta.rs

@@ -9,23 +9,23 @@ use crate::{cache_alias_and_labels, errors::ApiInitError};
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct EndpointMeta {
-    pub name: String,
+    pub table_name: String,
     pub log_id: String,
     pub schema: EndpointSchema,
 }
 
 impl EndpointMeta {
     pub async fn load_from_client(
         client: &mut InternalPipelineServiceClient<Channel>,
-        endpoint: String,
+        table_name: String,
     ) -> Result<(Self, LogClient), ApiInitError> {
         // We establish the log stream first to avoid tonic auto-reconnecting without us knowing.
-        let (log_client, schema) = LogClient::new(client, endpoint.clone()).await?;
+        let (log_client, schema) = LogClient::new(client, table_name.clone()).await?;
         let log_id = client.get_id(()).await?.into_inner().id;
 
         Ok((
             Self {
-                name: endpoint,
+                table_name,
                 log_id,
                 schema,
             },
@@ -34,12 +34,12 @@ impl EndpointMeta {
     }
 
     pub fn cache_alias_and_labels(&self, extra_labels: Labels) -> (String, Labels) {
-        let (alias, mut labels) = cache_alias_and_labels(self.name.clone());
+        let (alias, mut labels) = cache_alias_and_labels(self.table_name.clone());
         labels.extend(extra_labels);
         (alias, labels)
     }
 
     pub fn cache_name(&self) -> String {
-        format!("{}_{}", self.log_id, self.name)
+        format!("{}_{}", self.log_id, self.table_name)
     }
 }

dozer-api/src/cache_builder/mod.rs

@@ -33,7 +33,7 @@ const READ_LOG_RETRY_INTERVAL: Duration = Duration::from_secs(1);
 #[derive(Debug)]
 pub struct CacheBuilder {
     client: InternalPipelineServiceClient<Channel>,
-    endpoint: String,
+    table_name: String,
     cache_manager: Arc<dyn RwCacheManager>,
     serving: Arc<ArcSwap<CacheReader>>,
     labels: Labels,
@@ -52,6 +52,7 @@ impl CacheBuilder {
     pub async fn new(
         cache_manager: Arc<dyn RwCacheManager>,
         app_server_url: String,
+        table_name: String,
         endpoint: &ApiEndpoint,
         labels: LabelsAndProgress,
     ) -> Result<(Self, EndpointSchema), ApiInitError> {
@@ -63,7 +64,7 @@ impl CacheBuilder {
                 error,
             })?;
         let (endpoint_meta, _) =
-            EndpointMeta::load_from_client(&mut client, endpoint.name.clone()).await?;
+            EndpointMeta::load_from_client(&mut client, table_name.clone()).await?;
 
         // Open or create cache.
         let cache_write_options = cache_write_options(endpoint.conflict_resolution);
@@ -74,14 +75,15 @@ impl CacheBuilder {
             cache_write_options,
         )
         .map_err(ApiInitError::OpenOrCreateCache)?;
-        let progress_bar = labels.create_progress_bar(format!("cache: {}", endpoint_meta.name));
+        let progress_bar =
+            labels.create_progress_bar(format!("cache: {}", endpoint_meta.table_name));
 
         let log_reader_options = get_log_reader_options(endpoint);
 
         Ok((
             Self {
                 client,
-                endpoint: endpoint.name.clone(),
+                table_name,
                 cache_manager,
                 serving: Arc::new(ArcSwap::from_pointee(serving)),
                 labels: labels.labels().clone(),
@@ -106,7 +108,7 @@ impl CacheBuilder {
         loop {
             // Connect to the endpoint's log.
             let Some(connect_result) = runtime.block_on(with_cancel(
-                connect_until_success(&mut self.client, &self.endpoint),
+                connect_until_success(&mut self.client, &self.table_name),
                 cancel,
             )) else {
                 return Ok(());

dozer-api/src/generator/oapi/generator.rs

@@ -12,15 +12,16 @@ use serde_json::{json, Value};
 pub struct OpenApiGenerator<'a> {
     schema: &'a dozer_types::types::Schema,
     secondary_indexes: &'a [IndexDefinition],
+    table_name: String,
     endpoint: ApiEndpoint,
     server_host: Vec<String>,
 }
 impl<'a> OpenApiGenerator<'a> {
     fn get_singular_name(&self) -> String {
-        self.endpoint.name.to_string()
+        self.table_name.to_string()
     }
     fn get_plural_name(&self) -> String {
-        format!("{}_array", self.endpoint.name)
+        format!("{}_array", self.table_name)
     }
 
     // Generate first secondary_index as an example
@@ -80,18 +81,18 @@ impl<'a> OpenApiGenerator<'a> {
         let responses = Responses {
             responses: indexmap::indexmap! {
                 StatusCode::Code(200) =>
-                ReferenceOr::Item(create_reference_response(format!("Get by id {}", self.endpoint.name), format!("#/components/schemas/{}", self.get_singular_name())))
+                ReferenceOr::Item(create_reference_response(format!("Get by id {}", self.table_name), format!("#/components/schemas/{}", self.get_singular_name())))
             },
             ..Default::default()
         };
         let get_operation = Some(Operation {
-            tags: vec![format!("{}", self.endpoint.name)],
+            tags: vec![format!("{}", self.table_name)],
             summary: Some("Fetch a single document record by primary key".to_owned()),
             description: Some(
                 "Generated API to fetch a single record. Primary key specified will be used for lookup"
                     .to_owned(),
             ),
-            operation_id: Some(format!("{}-by-id", self.endpoint.name)),
+            operation_id: Some(format!("{}-by-id", self.table_name)),
             parameters: vec![ReferenceOr::Item(Parameter::Path {
                 parameter_data: ParameterData {
                     name: "id".to_owned(),
@@ -123,17 +124,17 @@ impl<'a> OpenApiGenerator<'a> {
     fn generate_list_route(&self) -> ReferenceOr<PathItem> {
         let responses = Responses {
             responses: indexmap::indexmap! {
-                StatusCode::Code(200) => ReferenceOr::Item(create_reference_response(format!("A page array of {}", self.endpoint.name.to_owned()), format!("#/components/schemas/{}",self.get_plural_name())))
+                StatusCode::Code(200) => ReferenceOr::Item(create_reference_response(format!("A page array of {}", self.table_name.to_owned()), format!("#/components/schemas/{}",self.get_plural_name())))
             },
             ..Default::default()
         };
         let operation = Some(Operation {
-            tags: vec![format!("{}", self.endpoint.name)],
+            tags: vec![format!("{}", self.table_name)],
             summary: Some("Fetch multiple documents in the default sort order".to_owned()),
             description: Some(
                 "This is used when no filter expression or sort is needed.".to_owned(),
             ),
-            operation_id: Some(format!("list-{}", self.endpoint.name.to_owned())),
+            operation_id: Some(format!("list-{}", self.table_name.to_owned())),
             responses,
             ..Default::default()
         });
@@ -170,10 +171,10 @@ impl<'a> OpenApiGenerator<'a> {
             ..Default::default()
         };
         let operation = Some(Operation {
-            tags: vec![format!("{}", self.endpoint.name)],
+            tags: vec![format!("{}", self.table_name)],
             summary: Some("Count documents based on an expression".to_string()),
             description: Some("Count documents based on an expression".to_string()),
-            operation_id: Some(format!("count-{}", self.endpoint.name)),
+            operation_id: Some(format!("count-{}", self.table_name)),
             request_body: Some(ReferenceOr::Item(request_body)),
             responses,
             ..Default::default()
@@ -194,17 +195,17 @@ impl<'a> OpenApiGenerator<'a> {
         };
         let responses = Responses {
             responses: indexmap::indexmap! {
-                StatusCode::Code(200) => ReferenceOr::Item(create_reference_response(format!("A page array of {}", self.endpoint.name.to_owned()), format!("#/components/schemas/{}", self.get_plural_name()) ))
+                StatusCode::Code(200) => ReferenceOr::Item(create_reference_response(format!("A page array of {}", self.table_name.to_owned()), format!("#/components/schemas/{}", self.get_plural_name()) ))
             },
             ..Default::default()
         };
         let operation = Some(Operation {
-            tags: vec![format!("{}", self.endpoint.name)],
+            tags: vec![format!("{}", self.table_name)],
             summary: Some("Query documents based on an expression".to_owned()),
             description: Some(
                 "Documents can be queried based on a simple or a composite expression".to_owned(),
             ),
-            operation_id: Some(format!("query-{}", self.endpoint.name)),
+            operation_id: Some(format!("query-{}", self.table_name)),
             request_body: Some(ReferenceOr::Item(request_body)),
             responses,
             ..Default::default()
@@ -234,13 +235,13 @@ impl<'a> OpenApiGenerator<'a> {
 
     fn generate_component_schema(&self) -> Components {
         let generated_schema =
-            convert_cache_to_oapi_schema(self.schema.to_owned(), &self.endpoint.name);
+            convert_cache_to_oapi_schema(self.schema.to_owned(), &self.table_name);
 
         let schemas = indexmap::indexmap! {
             self.get_singular_name() => ReferenceOr::Item(generated_schema),
             self.get_plural_name() => ReferenceOr::Item(Schema {
                         schema_data: SchemaData {
-                            description: Some(format!("Array of {}", &self.endpoint.name)),
+                            description: Some(format!("Array of {}", &self.table_name)),
                             ..Default::default()
                         },
                         schema_kind: SchemaKind::Type(Type::Array(ArrayType {
@@ -267,17 +268,17 @@ impl<'a> OpenApiGenerator<'a> {
         OpenAPI {
             openapi: "3.0.0".to_owned(),
             info: Info {
-                title: self.endpoint.name.to_uppercase(),
+                title: self.table_name.to_uppercase(),
                 description: Some(format!(
                     "API documentation for {}. Powered by Dozer Data.",
-                    self.endpoint.name.to_lowercase()
+                    self.table_name.to_lowercase()
                 )),
                 version: "1.0.0".to_owned(),
                 contact: create_contact_info(),
                 ..Default::default()
             },
             tags: vec![Tag {
-                name: self.endpoint.name.to_string(),
+                name: self.table_name.to_string(),
                 ..Default::default()
             }],
             servers: self
@@ -297,12 +298,14 @@ impl<'a> OpenApiGenerator<'a> {
     pub fn new(
         schema: &'a dozer_types::types::Schema,
         secondary_indexes: &'a [IndexDefinition],
+        table_name: String,
         endpoint: ApiEndpoint,
         server_host: Vec<String>,
     ) -> Self {
         Self {
             schema,
             secondary_indexes,
+            table_name,
             endpoint,
             server_host,
         }