fix(transfer): preserve PostgreSQL bytea migration

crates/dbx-core/src/transfer.rs

@@ -714,6 +714,10 @@ pub fn escape_value_typed(val: &serde_json::Value, db_type: &DatabaseType, colum
             _ => n.to_string(),
         },
         serde_json::Value::String(s) => {
+            if let Some(binary_literal) = format_postgres_binary_sql_literal(s, db_type, column_type) {
+                return binary_literal;
+            }
+
             let literal = format_literal_string(s, db_type, column_type);
             let escaped = if is_postgres_family_target(db_type) {
                 literal.replace('\'', "''")
@@ -747,6 +751,29 @@ fn is_mysql_bit_type(column_type: &str) -> bool {
     lower == "bit" || lower.starts_with("bit(") || lower.starts_with("bit ")
 }
 
+fn is_binary_transfer_column_type(column_type: &str) -> bool {
+    let lower = column_type.trim().to_ascii_lowercase();
+    let base = lower.split(['(', ' ', '\t', '\n']).next().unwrap_or("");
+    matches!(base, "binary" | "varbinary" | "blob" | "tinyblob" | "mediumblob" | "longblob" | "bytea" | "image")
+}
+
+fn format_postgres_binary_sql_literal(
+    value: &str,
+    db_type: &DatabaseType,
+    column_type: Option<&str>,
+) -> Option<String> {
+    if !matches!(db_type, DatabaseType::Postgres) || !column_type.is_some_and(is_binary_transfer_column_type) {
+        return None;
+    }
+
+    let hex = value.strip_prefix("0x").or_else(|| value.strip_prefix("0X"))?;
+    if hex.len() % 2 != 0 || !hex.as_bytes().iter().all(|byte| byte.is_ascii_hexdigit()) {
+        return None;
+    }
+
+    Some(format!("decode('{hex}', 'hex')"))
+}
+
 pub fn format_pg_array_sql_literal(arr: &[serde_json::Value]) -> String {
     if arr.is_empty() {
         return "'{}'".to_string();
@@ -4664,6 +4691,24 @@ mod tests {
         );
     }
 
+    #[test]
+    fn postgres_insert_formats_bytea_prefixed_hex_as_binary_literal() {
+        let sql = generate_insert_typed(
+            &[String::from("id"), String::from("payload"), String::from("note")],
+            &[Some(String::from("integer")), Some(String::from("BYTEA")), Some(String::from("text"))],
+            &[vec![json!(1), json!("0x48656c6c6f"), json!("0x48656c6c6f")]],
+            "files",
+            "public",
+            &DatabaseType::Postgres,
+        );
+
+        assert_eq!(
+            sql,
+            r#"INSERT INTO "public"."files" ("id", "payload", "note") VALUES
+(1, decode('48656c6c6f', 'hex'), '0x48656c6c6f')"#
+        );
+    }
+
     #[test]
     fn mysql_insert_keeps_backslash_escape_style() {
         let sql = generate_insert_typed(

crates/dbx-core/tests/live_postgres_transfer.rs

@@ -110,6 +110,14 @@ async fn live_postgres_transfer_preserves_data_and_schema_objects() {
             )",
             source_schema, source_schema
         ),
+        format!(
+            "CREATE TABLE \"{}\".\"files\" (\
+                \"id\" integer PRIMARY KEY,\
+                \"payload\" bytea NOT NULL,\
+                \"note\" text NOT NULL\
+            )",
+            source_schema
+        ),
         format!(
             "CREATE INDEX \"users_display_name_idx\" ON \"{}\".\"users\" USING btree (lower(display_name))",
             source_schema
@@ -139,6 +147,11 @@ async fn live_postgres_transfer_preserves_data_and_schema_objects() {
              ('beta@example.com', 'disabled', false, 'Beta')",
             source_schema
         ),
+        format!(
+            "INSERT INTO \"{}\".\"files\" (\"id\", \"payload\", \"note\") VALUES \
+             (1, decode('48656c6c6f', 'hex'), '0x48656c6c6f')",
+            source_schema
+        ),
         format!(
             "CREATE VIEW \"{}\".\"active_users\" AS \
              SELECT \"id\", \"email\", \"display_name\" FROM \"{}\".\"users\" WHERE \"active\"",
@@ -192,7 +205,7 @@ async fn live_postgres_transfer_preserves_data_and_schema_objects() {
         target_connection_id: target_connection_id.to_string(),
         target_database: target_database.clone(),
         target_schema: target_schema.clone(),
-        tables: vec!["users".to_string(), "audit_logs".to_string()],
+        tables: vec!["users".to_string(), "audit_logs".to_string(), "files".to_string()],
         create_table: true,
         mode: TransferMode::Append,
         target_table_name_case: TransferTableNameCase::Preserve,
@@ -229,6 +242,27 @@ async fn live_postgres_transfer_preserves_data_and_schema_objects() {
         query_scalar(&target_pool, &format!("SELECT count(*) FROM \"{}\".\"audit_logs\"", target_schema)).await,
         json!(2)
     );
+    assert_eq!(
+        query_scalar(
+            &target_pool,
+            &format!("SELECT octet_length(\"payload\") FROM \"{}\".\"files\" WHERE \"id\" = 1", target_schema)
+        )
+        .await,
+        json!(5)
+    );
+    assert_eq!(
+        query_scalar(
+            &target_pool,
+            &format!("SELECT encode(\"payload\", 'hex') FROM \"{}\".\"files\" WHERE \"id\" = 1", target_schema)
+        )
+        .await,
+        json!("48656c6c6f")
+    );
+    assert_eq!(
+        query_scalar(&target_pool, &format!("SELECT \"note\" FROM \"{}\".\"files\" WHERE \"id\" = 1", target_schema))
+            .await,
+        json!("0x48656c6c6f")
+    );
     assert_eq!(
         query_scalar(
             &target_pool,