Patch assignment

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pgmq"
-version = "0.14.1"
+version = "0.14.2"
 edition = "2021"
 authors = ["Tembo.io"]
 description = "Postgres extension for PGMQ"

core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pgmq"
-version = "0.16.0"
+version = "0.16.1"
 edition = "2021"
 authors = ["Tembo.io"]
 description = "A distributed message queue for Rust applications, on Postgres."

core/src/query.rs

@@ -336,7 +336,7 @@ pub fn assign_queue(name: CheckedName<'_>) -> Result<String, PgmqError> {
 }
 
 pub fn assign_archive(name: CheckedName<'_>) -> Result<String, PgmqError> {
-    Ok(assign(&format!("{name}_archive; ")))
+    Ok(assign(&format!("{name}_archive")))
 }
 
 pub fn unassign_queue(name: CheckedName<'_>) -> Result<String, PgmqError> {
@@ -362,14 +362,15 @@ pub fn assign(table_name: &str) -> String {
             SELECT 1
             FROM pg_depend
             WHERE refobjid = (SELECT oid FROM pg_extension WHERE extname = 'pgmq')
-            AND objid = (SELECT oid FROM pg_class WHERE relname = '{TABLE_PREFIX}_{table_name}')
+            AND objid = (
+                SELECT oid
+                FROM pg_class
+                WHERE relname = '{TABLE_PREFIX}_{table_name}'
+            )
         ) THEN
-
             EXECUTE 'ALTER EXTENSION pgmq ADD TABLE {PGMQ_SCHEMA}.{TABLE_PREFIX}_{table_name}';
-
         END IF;
-
-        END $$;
+    END $$;
     "
     )
 }
@@ -405,6 +406,12 @@ pub fn check_input(input: &str) -> Result<(), PgmqError> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn test_assign() {
+        let query = assign("my_queue_archive");
+        assert!(query.contains("WHERE relname = 'pgmq_my_queue_archive'"));
+    }
+
     #[test]
     fn test_create() {
         let queue_name = CheckedName::new("yolo").unwrap();

tests/integration_tests.rs

@@ -44,6 +44,13 @@ async fn test_lifecycle() {
         .await
         .expect("failed to create queue");
 
+    // creating a queue must be idempotent
+    // create with same name again, must be no  error
+    let _ = sqlx::query(&format!("SELECT pgmq_create('{test_default_queue}');"))
+        .execute(&conn)
+        .await
+        .expect("failed to create queue");
+
     let msg_id = sqlx::query(&format!(
         "SELECT * from pgmq_send('{test_default_queue}', '{{\"hello\": \"world\"}}');"
     ))