Improve

Author: iambriccardo

etl/src/conversions/event.rs

@@ -20,12 +20,12 @@ use crate::{bail, etl_error};
 /// event trigger. Messages with this prefix contain JSON-encoded schema information.
 pub const DDL_MESSAGE_PREFIX: &str = "supabase_etl_ddl";
 
-/// Represents a DDL schema change message emitted by Postgres event trigger.
+/// Represents a schema change message emitted by Postgres event trigger.
 ///
 /// This message is emitted when ALTER TABLE commands are executed on tables
 /// that are part of a publication.
 #[derive(Debug, Clone, Deserialize)]
-pub struct DdlSchemaChangeMessage {
+pub struct SchemaChangeMessage {
     /// The DDL command that triggered this message (e.g., "ALTER TABLE").
     pub event: String,
     /// The schema name of the affected table.
@@ -35,13 +35,49 @@ pub struct DdlSchemaChangeMessage {
     /// The OID of the affected table.
     pub table_id: i64,
     /// The columns of the table after the schema change.
-    pub columns: Vec<DdlColumnSchema>,
+    pub columns: Vec<ColumnSchemaMessage>,
 }
 
-/// Represents a column schema in a DDL schema change message.
+impl SchemaChangeMessage {
+
+    /// Converts a [`SchemaChangeMessage`] to a [`TableSchema`] with a specific snapshot ID.
+    ///
+    /// This is used to update the stored table schema when a DDL change is detected.
+    /// The snapshot_id should be the start_lsn of the DDL message.
+    pub fn into_table_schema(
+        self,
+        snapshot_id: SnapshotId,
+    ) -> TableSchema {
+        let table_name = TableName::new(self.schema_name, self.table_name);
+        let column_schemas = self
+            .columns
+            .into_iter()
+            .map(|column| {
+                let typ = convert_type_oid_to_type(column.type_oid);
+                ColumnSchema::new(
+                    column.name,
+                    typ,
+                    column.type_modifier,
+                    column.ordinal_position,
+                    column.primary_key_ordinal_position,
+                    column.nullable,
+                )
+            })
+            .collect();
+
+        TableSchema::with_snapshot_id(
+            TableId::new(self.table_id as u32),
+            table_name,
+            column_schemas,
+            snapshot_id,
+        )
+    }
+}
+
+/// Represents a column schema in a schema change message.
 #[allow(dead_code)]
 #[derive(Debug, Clone, Deserialize)]
-pub struct DdlColumnSchema {
+pub struct ColumnSchemaMessage {
     /// The name of the column.
     pub name: String,
     /// The OID of the column's data type.
@@ -316,45 +352,12 @@ pub fn convert_tuple_to_row<'a>(
 /// Parses a DDL schema change message from its JSON content.
 ///
 /// Returns the parsed message if successful, or an error if the JSON is malformed.
-pub fn parse_ddl_schema_change_message(content: &str) -> EtlResult<DdlSchemaChangeMessage> {
+pub fn parse_schema_change_message(content: &str) -> EtlResult<SchemaChangeMessage> {
     serde_json::from_str(content).map_err(|e| {
         etl_error!(
             ErrorKind::ConversionError,
-            "Failed to parse DDL schema change message",
-            format!("Invalid JSON in DDL message: {}", e)
+            "Failed to parse schema change message",
+            format!("Invalid JSON in schema change message: {}", e)
         )
     })
 }
-
-/// Converts a [`DdlSchemaChangeMessage`] to a [`TableSchema`] with a specific snapshot ID.
-///
-/// This is used to update the stored table schema when a DDL change is detected.
-/// The snapshot_id should be the start_lsn of the DDL message.
-pub fn ddl_message_to_table_schema(
-    message: &DdlSchemaChangeMessage,
-    snapshot_id: SnapshotId,
-) -> TableSchema {
-    let table_name = TableName::new(message.schema_name.clone(), message.table_name.clone());
-    let column_schemas = message
-        .columns
-        .iter()
-        .map(|col| {
-            let typ = convert_type_oid_to_type(col.type_oid);
-            ColumnSchema::new(
-                col.name.clone(),
-                typ,
-                col.type_modifier,
-                col.ordinal_position,
-                col.primary_key_ordinal_position,
-                col.nullable,
-            )
-        })
-        .collect();
-
-    TableSchema::with_snapshot_id(
-        TableId::new(message.table_id as u32),
-        table_name,
-        column_schemas,
-        snapshot_id,
-    )
-}

etl/src/replication/apply.rs

@@ -3,7 +3,7 @@ use crate::concurrency::shutdown::ShutdownRx;
 use crate::concurrency::signal::SignalRx;
 use crate::concurrency::stream::{TimeoutStream, TimeoutStreamResult};
 use crate::conversions::event::{
-    DDL_MESSAGE_PREFIX, ddl_message_to_table_schema, parse_ddl_schema_change_message,
+    DDL_MESSAGE_PREFIX, parse_schema_change_message,
     parse_event_from_begin_message, parse_event_from_commit_message,
     parse_event_from_delete_message, parse_event_from_insert_message,
     parse_event_from_truncate_message, parse_event_from_update_message,
@@ -1524,14 +1524,14 @@ where
     }
 
     let content = message.content()?;
-    let Ok(ddl_message) = parse_ddl_schema_change_message(content) else {
+    let Ok(schema_change_message) = parse_schema_change_message(content) else {
         bail!(
             ErrorKind::SourceConnectionFailed,
             "PostgreSQL connection has been closed during the apply loop"
         );
     };
 
-    let table_id = TableId::new(ddl_message.table_id as u32);
+    let table_id = TableId::new(schema_change_message.table_id as u32);
     // TODO: check if this check is required or we can leverage the idempotency of schema writing and
     //  we always unconditionally update the schema.
     if let Some(remote_final_lsn) = state.remote_final_lsn {
@@ -1544,26 +1544,27 @@ where
     }
 
     info!(
-        table_id = ddl_message.table_id,
-        table_name = %ddl_message.table_name,
-        schema_name = %ddl_message.schema_name,
-        event = %ddl_message.event,
-        columns = ddl_message.columns.len(),
+        table_id = schema_change_message.table_id,
+        table_name = %schema_change_message.table_name,
+        schema_name = %schema_change_message.schema_name,
+        event = %schema_change_message.event,
+        columns = schema_change_message.columns.len(),
         "received ddl schema change message"
     );
 
     // Build table schema from DDL message with start_lsn as the snapshot_id.
     let snapshot_id: SnapshotId = u64::from(start_lsn) as i64;
-    let table_schema = ddl_message_to_table_schema(&ddl_message, snapshot_id);
+    let table_schema = schema_change_message.into_table_schema(snapshot_id);
 
-    // Store the new schema version.
+    // Store the new schema version in the store.
     schema_store.store_table_schema(table_schema).await?;
 
     // Update the current schema snapshot in the state.
     state.update_schema_snapshot(snapshot_id);
 
+    let table_id: u32 = table_id.into();
     info!(
-        table_id = ddl_message.table_id,
+        table_id = table_id,
         snapshot_id = snapshot_id,
         "stored new schema version from ddl message"
     );