Problem: using pre-0.6.0 fork

This is not convenient, especially because it requires a forked version
of `cargo-pgx`

Solution: backport pgcentralfoundation/pgrx#887 into pg_crdt

pgcentralfoundation/pgrx#887 has stalled and it is unknown when
and if it'll make it into the mainline. The approach may change.

Cargo.toml

@@ -13,7 +13,7 @@ pg15 = ["pgx/pg15", "pgx-tests/pg15" ]
 pg_test = ["pgx-tests", "postgres", "lib0", "bytes"]
 
 [dependencies]
-pgx = "0.6.0-alpha.2"
+pgx = "0.6.0"
 serde = "1.0"
 itertools = "0.10.3"
 yrs = "0.13.0"
@@ -27,12 +27,7 @@ bytes = { version = "1.3.0", optional = true }
 automerge = "0.1.0"
 
 [dev-dependencies]
-pgx-tests = "0.6.0-alpha.2"
-
-[patch.crates-io]
-pgx = { version = "0.6.0-alpha.2", git = "https://github.com/yrashk/pgx", rev = "91d0d682" }
-pgx-tests = { version = "0.6.0-alpha.2", git = "https://github.com/yrashk/pgx", rev = "91d0d682" }
-
+pgx-tests = "0.6.0"
 
 [profile.dev]
 panic = "unwind"

README.md

@@ -9,10 +9,6 @@ user friendly, complete, or fully correct interface.
 
 Current version requires a patched version of cargo-pgx:
 
-```shell
-cargo install --git https://github.com/tcdi/pgx --rev 91d0d682 --force cargo-pgx
-```
-
 ## Design
 
 The database's internal representation of a CRDT Doc is the Doc's state vector encoded as an update. This format can be

src/automerge.rs

@@ -1,4 +1,4 @@
-use crate::serialization_primitives;
+use crate::{read_internal, serialization_primitives};
 use automerge::{Automerge, Change};
 use pgx::*;
 use serde::de::{Error, Visitor};
@@ -26,7 +26,6 @@ extension_sql!(
 
 #[derive(PostgresType)]
 #[inoutfuncs]
-#[sendrecvfuncs]
 pub struct AutoDoc(Automerge);
 
 impl AutoDoc {
@@ -64,7 +63,6 @@ serialization_primitives!(AutoDoc);
 
 #[derive(PostgresType)]
 #[inoutfuncs]
-#[sendrecvfuncs]
 pub struct AutoChange(Change);
 
 impl TryFrom<&[u8]> for AutoChange {
@@ -158,6 +156,30 @@ fn autochange_from_bytea(array: Vec<u8>) -> AutoChange {
         .expect("invalid AutoChange (binary)")
 }
 
+#[pg_extern(immutable)]
+#[search_path(@extschema@)]
+fn autodoc_send(doc: AutoDoc) -> Vec<u8> {
+    (&doc).into()
+}
+
+#[pg_extern(immutable)]
+#[search_path(@extschema@)]
+fn autodoc_receive(internal: Internal) -> AutoDoc {
+    AutoDoc::try_from(read_internal(internal).as_slice()).expect("invalid AutoDoc")
+}
+
+#[pg_extern(immutable)]
+#[search_path(@extschema@)]
+fn autochange_send(change: AutoChange) -> Vec<u8> {
+    (&change).into()
+}
+
+#[pg_extern(immutable)]
+#[search_path(@extschema@)]
+fn autochange_receive(internal: Internal) -> AutoChange {
+    AutoChange::try_from(read_internal(internal).as_slice()).expect("invalid AutoChange")
+}
+
 #[cfg(any(test, feature = "pg_test"))]
 #[pg_schema]
 mod tests {

src/lib.rs

@@ -2,6 +2,18 @@ use pgx::*;
 
 pg_module_magic!();
 
+extension_sql!(
+    r#"
+    ALTER TYPE crdt.ydoc SET (SEND = ydoc_send, RECEIVE = ydoc_receive);
+    ALTER TYPE crdt.yupdate SET (SEND = yupdate_send, RECEIVE = yupdate_receive);
+    ALTER TYPE crdt.autodoc SET (SEND = autodoc_send, RECEIVE = autodoc_receive);
+    ALTER TYPE crdt.autochange SET (SEND = autochange_send, RECEIVE = autochange_receive);
+
+    "#,
+    name = "binary_format",
+    finalize
+);
+
 pub mod automerge;
 pub mod y;
 
@@ -65,19 +77,19 @@ macro_rules! serialization_primitives {
                     .expect("can't encode");
             }
         }
-
-        impl SendRecvFuncs for $ty {
-            fn send(&self) -> Vec<u8> {
-                self.into()
-            }
-
-            fn recv(buffer: &[u8]) -> Self {
-                buffer.try_into().expect("invalid $ty")
-            }
-        }
     };
 }
 
+pub(crate) fn read_internal(internal: Internal) -> Vec<u8> {
+    let mut buffer0 = unsafe {
+        internal
+            .get_mut::<pg_sys::StringInfoData>()
+            .expect("Can't retrieve StringInfo pointer")
+    };
+    let buffer = StringInfo::from_pg(buffer0 as *mut _).expect("failed to construct StringInfo");
+    (*buffer0).cursor = (*buffer0).len;
+    buffer.as_bytes().to_vec()
+}
 #[cfg(test)]
 pub mod pg_test {
     pub fn setup(_options: Vec<&str>) {

src/y.rs

@@ -1,4 +1,4 @@
-use crate::serialization_primitives;
+use crate::{read_internal, serialization_primitives};
 use pgx::*;
 use serde::de::{Error, Visitor};
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
@@ -28,7 +28,6 @@ extension_sql!(
 
 #[derive(PostgresType)]
 #[inoutfuncs]
-#[sendrecvfuncs]
 pub struct YDoc(Doc);
 
 impl YDoc {
@@ -79,7 +78,6 @@ serialization_primitives!(YDoc);
 
 #[derive(PostgresType)]
 #[inoutfuncs]
-#[sendrecvfuncs]
 pub struct YUpdate(Update);
 
 impl TryFrom<&[u8]> for YUpdate {
@@ -170,6 +168,30 @@ fn yupdate_from_bytea(array: Vec<u8>) -> YUpdate {
         .expect("invalid YUpdate (binary)")
 }
 
+#[pg_extern(immutable)]
+#[search_path(@extschema@)]
+fn ydoc_send(doc: YDoc) -> Vec<u8> {
+    (&doc).into()
+}
+
+#[pg_extern(immutable)]
+#[search_path(@extschema@)]
+fn ydoc_receive(internal: Internal) -> YDoc {
+    YDoc::try_from(read_internal(internal).as_slice()).expect("invalid YDoc")
+}
+
+#[pg_extern(immutable)]
+#[search_path(@extschema@)]
+fn yupdate_send(update: YUpdate) -> Vec<u8> {
+    (&update).into()
+}
+
+#[pg_extern(immutable)]
+#[search_path(@extschema@)]
+fn yupdate_receive(internal: Internal) -> YUpdate {
+    YUpdate::try_from(read_internal(internal).as_slice()).expect("invalid YDoc")
+}
+
 #[cfg(any(test, feature = "pg_test"))]
 #[pg_schema]
 mod tests {