node: Integrate new p2panda high-level API

This patch integrates the new `p2panda` crate into the "backend" of
Reflection.

`p2panda` is the new all-in-one high-level Node API which should make
usage of p2panda features easier, removing a bunch of code in
`reflection-node` which is now handled in p2panda directly, such as:

* Encoding / Decoding of operations and their message payloads
* Make sure changes to the append-only log are strictly serialized, to
  avoid breaking consistency / integrity of data due to race conditions
* Re-playing all operations on demand, this is required by Reflection to
  re-materialize all state for a document
* Manage header extensions to include log ids and prune flags
* Process local and remotely incoming operations ("ingest", "pruning")
* SQLite storage handling for logs and operations
* Mapping between logs and topics
* De-duplicate messages in ephemeral streams by adding timestamp
* Add public key / signatures to give provenance to ephemeral messages

Some features implemented in `reflection-node` stay as they're adding
functionality _on top_ of `p2panda`. Some of them might be moved there
as well in the future, such as:

* Author Presence: Indicate which authors have contributed to which
  topic and if they're currently online.
  Related issue: p2panda/p2panda#1002
* Persisted Topics: Store to persist all previously used topics

Unfortunately we need to remove one feature for now which is to change
"connection modes", causing the network to be disabled / enabled.
`reflection-node` handled that well in the past by only disabling the
networking layer while keeping the event processing layer intact. We
will have to do something similar in `p2panda` to offer this feature as
well since the new API surface will not allow such low-level access
anymore.

See related issue: p2panda/p2panda#1093

BREAKING CHANGES

These changes introduce breaking changes to the core operation format
(`previous` field was removed) and the header extensions changed which
means that all previously created, persisted operations will _not_ be
compatible anymore.

THIS RESULTS IN LOOSING ALL PREVIOUSLY CREATED DOCUMENTS.

Since `p2panda` is not stable yet this might happen again in the near
future, though this was the most important and largest update and we're
much closer to a stable API and protocol now.

Final remarks:

* Some minor changes in `reflection-doc` where also necessary, I
  specifically decided to make `p2panda-core` a concrete dependency
  there since too much functionality is used to justify re-exports from
  `reflection-node`.

Author: adzialocha

Cargo.lock

@@ -14,6 +14,7 @@ glib = "0.21"
 hex = "0.4.3"
 indexmap = "2.13.0"
 loro = { tag = "loro-crdt@1.10.6", git = "https://github.com/loro-dev/loro.git" }
+p2panda-core = { git = "https://github.com/p2panda/p2panda", rev = "534363cfc33d2e3275748bf8b0a9307e6eb261c0" }
 rand = "0.10.0"
 reflection-node = { path = "../reflection-node" }
 serde = { version = "1.0.228", features = ["derive"] }

reflection-doc/Cargo.toml

@@ -9,11 +9,9 @@ use glib::{Properties, clone};
 pub use hex::FromHexError;
 use loro::{ExportMode, LoroDoc, LoroText, event::Diff};
 use p2panda_core::cbor::{decode_cbor, encode_cbor};
-use reflection_node::p2panda_core;
-use reflection_node::topic::{
-    SubscribableTopic, Subscription as TopicSubscription,
-    SubscriptionError as TopicSubscriptionError,
-};
+use p2panda_core::{self, Topic};
+use reflection_node::subscription::Subscription as TopicSubscription;
+use reflection_node::traits::{SubscribableTopic, SubscriptionError as TopicSubscriptionError};
 use tracing::error;
 
 use crate::author::Author;
@@ -37,6 +35,18 @@ impl From<[u8; 32]> for DocumentId {
     }
 }
 
+impl From<Topic> for DocumentId {
+    fn from(value: Topic) -> Self {
+        Self(value.to_bytes())
+    }
+}
+
+impl From<DocumentId> for Topic {
+    fn from(value: DocumentId) -> Self {
+        Topic::from(value.0)
+    }
+}
+
 impl fmt::Display for DocumentId {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str(&self.to_hex())
@@ -298,7 +308,7 @@ mod imp {
                     #[weak]
                     subscription,
                     async move {
-                        if let Err(error) = subscription.send_ephemeral(cursor_bytes).await {
+                        if let Err(error) = subscription.publish_ephemeral(cursor_bytes).await {
                             error!("Failed to send cursor position: {}", error);
                         }
                     }
@@ -495,7 +505,7 @@ mod imp {
                             subscription,
                             async move {
                                 // Broadcast a "text delta" to all peers
-                                if let Err(error) = subscription.send_delta(delta_bytes).await {
+                                if let Err(error) = subscription.publish_delta(delta_bytes).await {
                                     error!(
                                         "Failed to send delta of document to the network: {}",
                                         error
@@ -828,7 +838,7 @@ impl Document {
                 .export(ExportMode::Snapshot)
                 .expect("encoded crdt snapshot");
 
-            if let Err(error) = subscription.send_snapshot(snapshot_bytes).await {
+            if let Err(error) = subscription.publish_snapshot(snapshot_bytes).await {
                 error!(
                     "Failed to send snapshot of document to the network: {}",
                     error
@@ -872,7 +882,7 @@ impl Document {
                 .expect("crdt_doc to be set")
                 .export(ExportMode::Snapshot)
                 .expect("encoded crdt snapshot");
-            if let Err(error) = subscription.send_snapshot(snapshot_bytes).await {
+            if let Err(error) = subscription.publish_snapshot(snapshot_bytes).await {
                 error!(
                     "Failed to send snapshot of document to the network: {}",
                     error

reflection-doc/src/document.rs

@@ -70,7 +70,7 @@ impl Documents {
     pub(crate) async fn load(&self, service: &Service) -> Result<(), StartupError> {
         let public_key = service.private_key().public_key();
 
-        let documents = service.node().topics::<DocumentId>().await?;
+        let documents = service.node().topics().await?;
 
         let mut list = self.imp().list.write().unwrap();
         assert!(list.is_empty());
@@ -102,14 +102,14 @@ impl Documents {
 
             let obj = Document::with_state(
                 service,
-                Some(&document.id),
+                Some(&document.id.into()),
                 document.name.as_deref(),
                 last_accessed.as_ref(),
             );
 
             obj.authors().load(authors);
 
-            list.insert(document.id, obj);
+            list.insert(document.id.into(), obj);
         }
 
         drop(list);

reflection-doc/src/documents.rs

@@ -5,11 +5,11 @@ pub mod documents;
 pub mod service;
 
 pub mod identity {
+    use std::fmt;
     use std::hash::Hash;
 
-    use reflection_node::p2panda_core;
-    pub use reflection_node::p2panda_core::identity::IdentityError;
-    use std::fmt;
+    use p2panda_core;
+    pub use p2panda_core::identity::IdentityError;
 
     #[derive(Clone, Debug, glib::Boxed)]
     #[boxed_type(name = "ReflectionPrivateKey", nullable)]

reflection-doc/src/lib.rs

@@ -1,29 +1,26 @@
+use std::sync::{Mutex, OnceLock};
+
 use gio::prelude::{FileExt, ListModelExtManual, NetworkMonitorExt};
 use glib::object::ObjectExt;
 use glib::subclass::prelude::*;
 use glib::{Properties, clone};
-use reflection_node::p2panda_core::Hash;
-use std::sync::{Mutex, OnceLock};
+use p2panda_core::Hash;
+use reflection_node::node;
+use reflection_node::node::{Node, NodeError};
+use reflection_node::subscription::SubscriptionError;
 use thiserror::Error;
 use tracing::error;
 
+use crate::document::{Document, DocumentId};
+use crate::documents::Documents;
 use crate::identity::PrivateKey;
-use crate::{
-    document::{Document, DocumentId},
-    documents::Documents,
-};
-use reflection_node::{
-    node,
-    node::{Node, NodeError},
-    topic::TopicError,
-};
 
 #[derive(Error, Debug)]
 pub enum StartupError {
     #[error(transparent)]
     Node(#[from] NodeError),
     #[error(transparent)]
-    Topic(#[from] TopicError),
+    Topic(#[from] SubscriptionError),
 }
 
 #[derive(Debug, Copy, Clone, PartialEq, Eq, glib::Enum, Default)]

reflection-doc/src/service.rs

@@ -12,21 +12,16 @@ authors = [
 test_utils = []
 
 [dependencies]
-thiserror = "2.0.18"
 chrono = "0.4.43"
-ciborium = "0.2.2"
-p2panda-core = { git = "https://github.com/p2panda/p2panda", rev = "a1308c116c1bd345f2cd6af7df41d2a6e0a4a682" }
-p2panda-discovery = { git = "https://github.com/p2panda/p2panda", rev = "a1308c116c1bd345f2cd6af7df41d2a6e0a4a682" }
-p2panda-net = { git = "https://github.com/p2panda/p2panda", rev = "a1308c116c1bd345f2cd6af7df41d2a6e0a4a682" }
-p2panda-store = { git = "https://github.com/p2panda/p2panda", rev = "a1308c116c1bd345f2cd6af7df41d2a6e0a4a682", features = ["sqlite"], default-features = false }
-p2panda-stream = { git = "https://github.com/p2panda/p2panda", rev = "a1308c116c1bd345f2cd6af7df41d2a6e0a4a682" }
-p2panda-sync = { git = "https://github.com/p2panda/p2panda", rev = "a1308c116c1bd345f2cd6af7df41d2a6e0a4a682" }
+p2panda = { git = "https://github.com/p2panda/p2panda", rev = "534363cfc33d2e3275748bf8b0a9307e6eb261c0" }
+p2panda-core = { git = "https://github.com/p2panda/p2panda", rev = "534363cfc33d2e3275748bf8b0a9307e6eb261c0" }
+p2panda-store = { git = "https://github.com/p2panda/p2panda", rev = "534363cfc33d2e3275748bf8b0a9307e6eb261c0" }
 serde = { version = "1.0.228", features = ["derive"] }
-serde_bytes = "0.11.19"
 sqlx = { version = "0.8.6", features = ["runtime-tokio", "sqlite", "chrono"], default-features = false }
+thiserror = "2.0.18"
 tokio = { version = "1.49.0", features = ["rt", "sync"] }
 tokio-stream = "0.1.18"
 tracing = "0.1"
+
+[dev-dependencies]
 test-log = { version = "0.2.19", default-features = false, features = ["trace", "color"] }
-hex = "0.4.3"
-rand_chacha = "0.10.0"

reflection-node/Cargo.toml

@@ -10,4 +10,4 @@ CREATE TABLE IF NOT EXISTS topics (
     id	 	        TEXT NOT NULL PRIMARY KEY,
     name		TEXT,
     last_accessed	INTEGER
-);
+);