Repository: p2panda/aquadoggo (language: rust)

Issue:
Replication strategy: naive protocol



Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 68a17749a..bc4c46f82 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -19,9 +19,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - SQL query for collections [#311](https://github.com/p2panda/aquadoggo/pull/311)
 - Add custom validation to GraphQL scalar types [#318](https://github.com/p2panda/aquadoggo/pull/318)
 - Introduce property tests for GraphQL query API with `proptest` [#338](https://github.com/p2panda/aquadoggo/pull/338)
-- Introduce initial libp2p network behaviour for replication protocol [#365](https://github.com/p2panda/aquadoggo/pull/365) 
+- Introduce initial libp2p network behaviour for replication protocol [#365](https://github.com/p2panda/aquadoggo/pull/365)
 - Replication protocol session manager [#363](https://github.com/p2panda/aquadoggo/pull/363)
 - Replication message de- / serialization [#375](https://github.com/p2panda/aquadoggo/pull/375)
+- Naive protocol replication [#380](https://github.com/p2panda/aquadoggo/pull/380)
 
 ### Changed
 
diff --git a/Cargo.lock b/Cargo.lock
index 9012f7ccb..e54f203b3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3314,7 +3314,7 @@ dependencies = [
 [[package]]
 name = "p2panda-rs"
 version = "0.7.0"
-source = "git+https://github.com/p2panda/p2panda?rev=679107b636eea39e7713d9b21ac69df28aaad97d#679107b636eea39e7713d9b21ac69df28aaad97d"
+source = "git+https://github.com/p2panda/p2panda?rev=ae60bf754a60a1daf9c6862e9ae51ee7501b007f#ae60bf754a60a1daf9c6862e9ae51ee7501b007f"
 dependencies = [
  "arrayvec 0.5.2",
  "async-trait",
diff --git a/aquadoggo/Cargo.toml b/aquadoggo/Cargo.toml
index d9edf17e1..1f51d48eb 100644
--- a/aquadoggo/Cargo.toml
+++ b/aquadoggo/Cargo.toml
@@ -51,7 +51,7 @@ lipmaa-link = "0.2.2"
 log = "0.4.17"
 once_cell = "1.17.0"
 openssl-probe = "0.1.5"
-p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "679107b636eea39e7713d9b21ac69df28aaad97d", features = [
+p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "ae60bf754a60a1daf9c6862e9ae51ee7501b007f", features = [
     "storage-provider",
 ] }
 serde = { version = "1.0.152", features = ["derive"] }
@@ -83,7 +83,7 @@ http = "0.2.9"
 hyper = "0.14.19"
 libp2p-swarm-test = "0.1.0"
 once_cell = "1.17.0"
-p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "679107b636eea39e7713d9b21ac69df28aaad97d", features = [
+p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "ae60bf754a60a1daf9c6862e9ae51ee7501b007f", features = [
     "test-utils",
     "storage-provider",
 ] }
diff --git a/aquadoggo/src/db/models/log.rs b/aquadoggo/src/db/models/log.rs
index 3b7abca5c..f75d13e5f 100644
--- a/aquadoggo/src/db/models/log.rs
+++ b/aquadoggo/src/db/models/log.rs
@@ -21,3 +21,15 @@ pub struct LogRow {
     /// SchemaId which identifies the schema for operations in this log.
     pub schema: String,
 }
+
+#[derive(FromRow, Debug, Clone)]
+pub struct LogHeightRow {
+    /// PublicKey of this entry.
+    pub public_key: String,
+
+    /// Used log for this entry.
+    pub log_id: String,
+
+    /// Sequence number of this entry.
+    pub seq_num: String,
+}
diff --git a/aquadoggo/src/db/models/mod.rs b/aquadoggo/src/db/models/mod.rs
index c4249778b..44394d262 100644
--- a/aquadoggo/src/db/models/mod.rs
+++ b/aquadoggo/src/db/models/mod.rs
@@ -10,7 +10,7 @@ mod query;
 mod task;
 pub mod utils;
 
-pub use self::log::LogRow;
+pub use self::log::{LogHeightRow, LogRow};
 pub use document::{DocumentRow, DocumentViewFieldRow};
 pub use entry::EntryRow;
 pub use operation::{OperationFieldsJoinedRow, OperationRow};
diff --git a/aquadoggo/src/db/stores/entry.rs b/aquadoggo/src/db/stores/entry.rs
index 7d1dcde36..3a7a7d220 100644
--- a/aquadoggo/src/db/stores/entry.rs
+++ b/aquadoggo/src/db/stores/entry.rs
@@ -1,5 +1,8 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+use std::collections::HashMap;
+use std::vec;
+
 use async_trait::async_trait;
 use lipmaa_link::get_lipmaa_links_back_to;
 use p2panda_rs::entry::traits::{AsEncodedEntry, AsEntry};
@@ -12,7 +15,7 @@ use p2panda_rs::storage_provider::error::EntryStorageError;
 use p2panda_rs::storage_provider::traits::EntryStore;
 use sqlx::{query, query_as};
 
-use crate::db::models::EntryRow;
+use crate::db::models::{EntryRow, LogHeightRow};
 use crate::db::types::StorageEntry;
 use crate::db::SqlStore;
 
@@ -314,6 +317,101 @@ impl EntryStore for SqlStore {
     }
 }
 
+impl SqlStore {
+    pub async fn get_log_heights(
+        &self,
+        schema_id: &SchemaId,
+    ) -> Result<Vec<(PublicKey, Vec<(LogId, SeqNum)>)>, EntryStorageError> {
+        let log_height_rows = query_as::<_, LogHeightRow>(
+            "
+            SELECT
+                entries.public_key,
+                entries.log_id,
+                CAST(MAX(CAST(entries.seq_num AS NUMERIC)) AS TEXT) as seq_num
+            FROM
+                entries
+            INNER JOIN logs
+                ON entries.log_id = logs.log_id
+                    AND entries.public_key = logs.public_key
+            WHERE
+                logs.schema = $1
+            GROUP BY
+                entries.public_key, entries.log_id
+            ",
+        )
+        .bind(schema_id.to_string())
+        .fetch_all(&self.pool)
+        .await
+        .map_err(|e| EntryStorageError::Custom(e.to_string()))?;
+
+        let mut log_heights = HashMap::<PublicKey, Vec<(LogId, SeqNum)>>::new();
+
+        // Aggregate log height rows into a map.
+        for LogHeightRow {
+            public_key,
+            log_id,
+            seq_num,
+        } in log_height_rows
+        {
+            let public_key: PublicKey = public_key
+                .parse()
+                .expect("Values stored in the database are valid");
+            let log_id: LogId = log_id
+                .parse()
+                .expect("Values stored in the database are valid");
+            let seq_num: SeqNum = seq_num
+                .parse()
+                .expect("Values stored in the database are valid");
+
+            if let Some(author_logs) = log_heights.get_mut(&public_key) {
+                author_logs.push((log_id, seq_num));
+            } else {
+                let author_logs = vec![(log_id, seq_num)];
+                log_heights.insert(public_key, author_logs);
+            }
+        }
+
+        // Convert log heights map back into vec.
+        Ok(log_heights.into_iter().collect())
+    }
+
+    pub async fn get_entries_from(
+        &self,
+        public_key: &PublicKey,
+        log_id: &LogId,
+        seq_num: &SeqNum,
+    ) -> Result<Vec<StorageEntry>, EntryStorageError> {
+        let entries = query_as::<_, EntryRow>(
+            "
+            SELECT
+                public_key,
+                entry_bytes,
+                entry_hash,
+                log_id,
+                payload_bytes,
+                payload_hash,
+                seq_num
+            FROM
+                entries
+            WHERE
+                public_key = $1
+                AND log_id = $2
+                AND CAST(seq_num AS NUMERIC) >= CAST($3 AS NUMERIC)
+            ORDER BY
+                CAST(seq_num AS NUMERIC)
+            ",
+        )
+        .bind(public_key.to_string())
+        .bind(log_id.as_u64().to_string())
+        .bind(seq_num.as_u64().to_string())
+        .fetch_all(&self.pool)
+        .await
+        .map_err(|e| EntryStorageError::Custom(e.to_string()))?;
+
+        Ok(entries.into_iter().map(|row| row.into()).collect())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use p2panda_rs::entry::traits::{AsEncodedEntry, AsEntry};
@@ -725,4 +823,58 @@ mod tests {
             assert_eq!(cert_pool_seq_nums, vec![19, 18, 17, 13, 4, 1]);
         });
     }
+
+    #[rstest]
+    fn get_log_heights(
+        #[from(populate_store_config)]
+        #[with(5, 5, 5)]
+        config: PopulateStoreConfig,
+    ) {
+        test_runner(|node: TestNode| async move {
+            // Populate the store with some entries and operations but DON'T materialise any resulting documents.
+            let (key_pairs, _) = populate_store(&node.context.store, &config).await;
+
+            let log_heights = node
+                .context
+                .store
+                .get_log_heights(config.schema.id())
+                .await
+                .unwrap();
+
+            assert_eq!(log_heights.len(), 5);
+
+            for (public_key, author_logs) in log_heights {
+                assert!(key_pairs
+                    .iter()
+                    .find(|key_pair| key_pair.public_key() == public_key)
+                    .is_some());
+                assert_eq!(author_logs.len(), 5);
+
+                for (_, seq_num) in author_logs {
+                    assert_eq!(seq_num.as_u64(), 5)
+                }
+            }
+        });
+    }
+
+    #[rstest]
+    fn get_entries_from(
+        #[from(populate_store_config)]
+        #[with(20, 2, 1)]
+        config: PopulateStoreConfig,
+    ) {
+        test_runner(|node: TestNode| async move {
+            // Populate the store with some entries and operations but DON'T materialise any resulting documents.
+            let (key_pairs, _) = populate_store(&node.context.store, &config).await;
+            let public_key = key_pairs[0].public_key();
+            let entries = node
+                .context
+                .store
+                .get_entries_from(&public_key, &LogId::default(), &SeqNum::new(10).unwrap())
+                .await
+                .unwrap();
+
+            assert_eq!(entries.len(), 11);
+        });
+    }
 }
diff --git a/aquadoggo/src/db/stores/query.rs b/aquadoggo/src/db/stores/query.rs
index 857460aa8..8c01869c1 100644
--- a/aquadoggo/src/db/stores/query.rs
+++ b/aquadoggo/src/db/stores/query.rs
@@ -24,6 +24,7 @@ use crate::db::types::StorageDocument;
 use crate::db::SqlStore;
 
 /// Configure query to select documents based on a relation list field.
+#[derive(Debug)]
 pub struct RelationList {
     /// View id of document which holds relation list field.
     pub root: DocumentViewId,
@@ -35,6 +36,7 @@ pub struct RelationList {
     pub list_type: RelationListType,
 }
 
+#[derive(Debug)]
 pub enum RelationListType {
     Pinned,
     Unpinned,
diff --git a/aquadoggo/src/db/types/operation.rs b/aquadoggo/src/db/types/operation.rs
index 271911ead..231a89239 100644
--- a/aquadoggo/src/db/types/operation.rs
+++ b/aquadoggo/src/db/types/operation.rs
@@ -7,6 +7,7 @@ use p2panda_rs::operation::{OperationAction, OperationFields, OperationId, Opera
 use p2panda_rs::schema::SchemaId;
 use p2panda_rs::WithId;
 
+#[derive(Debug)]
 pub struct StorageOperation {
     /// The id of the document this operation is part of.
     pub(crate) document_id: DocumentId,
diff --git a/aquadoggo/src/graphql/queries/collection.rs b/aquadoggo/src/graphql/queries/collection.rs
index 36d387c9f..c149cabe1 100644
--- a/aquadoggo/src/graphql/queries/collection.rs
+++ b/aquadoggo/src/graphql/queries/collection.rs
@@ -1162,7 +1162,6 @@ mod tests {
 
             assert_eq!(data["query"]["documents"].as_array().unwrap().len(), 1);
             let end_cursor = data["query"]["endCursor"].clone();
-            println!("{end_cursor}");
 
             let data = query_songs_meta_fields_only(
                 &client,
diff --git a/aquadoggo/src/graphql/queries/next_args.rs b/aquadoggo/src/graphql/queries/next_args.rs
index 249ec996d..51bc98e07 100644
--- a/aquadoggo/src/graphql/queries/next_args.rs
+++ b/aquadoggo/src/graphql/queries/next_args.rs
@@ -183,7 +183,6 @@ mod tests {
                 .json::<Response>()
                 .await;
 
-            print!("{:?}", received_entry_args.errors);
             assert!(received_entry_args.is_ok());
             assert_eq!(
                 received_entry_args.data,
diff --git a/aquadoggo/src/replication/errors.rs b/aquadoggo/src/replication/errors.rs
index b11028e00..0451886f4 100644
--- a/aquadoggo/src/replication/errors.rs
+++ b/aquadoggo/src/replication/errors.rs
@@ -12,6 +12,33 @@ pub enum ReplicationError {
 
     #[error("Tried to initialise duplicate outbound replication session with id {0}")]
     DuplicateOutboundRequest(u64),
+
+    #[error("No session found with id {0}")]
+    NoSessionFound(u64),
+
+    #[error("Received entry which is not in target set")]
+    UnmatchedTargetSet,
+
+    #[error("Replication strategy failed with error: {0}")]
+    StrategyFailed(String),
+
+    #[error("Incoming data could not be ingested")]
+    Validation(#[from] IngestError),
+}
+
+#[derive(Error, Debug)]
+pub enum IngestError {
+    #[error("Schema is not supported")]
+    UnsupportedSchema,
+
+    #[error("Received entry and operation is invalid")]
+    Domain(#[from] p2panda_rs::api::DomainError),
+
+    #[error("Decoding entry failed")]
+    DecodeEntry(#[from] p2panda_rs::entry::error::DecodeEntryError),
+
+    #[error("Decoding operation failed")]
+    DecodeOperation(#[from] p2panda_rs::operation::error::DecodeOperationError),
 }
 
 #[derive(Error, Debug)]
diff --git a/aquadoggo/src/replication/ingest.rs b/aquadoggo/src/replication/ingest.rs
new file mode 100644
index 000000000..e92672c63
--- /dev/null
+++ b/aquadoggo/src/replication/ingest.rs
@@ -0,0 +1,189 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use p2panda_rs::api::validation::{
+    ensure_document_not_deleted, get_checked_document_id_for_view_id, get_expected_skiplink,
+    is_next_seq_num, validate_claimed_schema_id,
+};
+use p2panda_rs::api::DomainError;
+use p2panda_rs::document::DocumentId;
+use p2panda_rs::entry::decode::decode_entry;
+use p2panda_rs::entry::traits::{AsEncodedEntry, AsEntry};
+use p2panda_rs::entry::EncodedEntry;
+use p2panda_rs::operation::decode::decode_operation;
+use p2panda_rs::operation::plain::PlainOperation;
+use p2panda_rs::operation::traits::{AsOperation, Schematic};
+use p2panda_rs::operation::validate::validate_operation_with_entry;
+use p2panda_rs::operation::{EncodedOperation, Operation, OperationAction, OperationId};
+use p2panda_rs::schema::Schema;
+use p2panda_rs::storage_provider::traits::{EntryStore, LogStore, OperationStore};
+
+use crate::bus::{ServiceMessage, ServiceSender};
+use crate::db::SqlStore;
+use crate::replication::errors::IngestError;
+use crate::schema::SchemaProvider;
+
+// @TODO: This method exists in `p2panda-rs`, we need to make it public there
+async fn validate_entry_and_operation<S: EntryStore + OperationStore + LogStore>(
+    store: &S,
+    schema: &Schema,
+    entry: &impl AsEntry,
+    encoded_entry: &impl AsEncodedEntry,
+    plain_operation: &PlainOperation,
+    encoded_operation: &EncodedOperation,
+) -> Result<(Operation, OperationId), DomainError> {
+    // Verify that the claimed seq num matches the expected seq num for this public_key and log.
+    let latest_entry = store
+        .get_latest_entry(entry.public_key(), entry.log_id())
+        .await?;
+    let latest_seq_num = latest_entry.as_ref().map(|entry| entry.seq_num());
+    is_next_seq_num(latest_seq_num, entry.seq_num())?;
+
+    // If a skiplink is claimed, get the expected skiplink from the database, errors if it can't be found.
+    let skiplink = match entry.skiplink() {
+        Some(_) => Some(
+            get_expected_skiplink(store, entry.public_key(), entry.log_id(), entry.seq_num())
+                .await?,
+        ),
+        None => None,
+    };
+
+    // Construct params as `validate_operation_with_entry` expects.
+    let skiplink_params = skiplink.as_ref().map(|entry| {
+        let hash = entry.hash();
+        (entry.clone(), hash)
+    });
+
+    // The backlink for this entry is the latest entry from this public key's log.
+    let backlink_params = latest_entry.as_ref().map(|entry| {
+        let hash = entry.hash();
+        (entry.clone(), hash)
+    });
+
+    // Perform validation of the entry and it's operation.
+    let (operation, operation_id) = validate_operation_with_entry(
+        entry,
+        encoded_entry,
+        skiplink_params.as_ref().map(|(entry, hash)| (entry, hash)),
+        backlink_params.as_ref().map(|(entry, hash)| (entry, hash)),
+        plain_operation,
+        encoded_operation,
+        schema,
+    )?;
+
+    Ok((operation, operation_id))
+}
+
+// @TODO: This method exists in `p2panda-rs`, we need to make it public there
+async fn determine_document_id<S: OperationStore>(
+    store: &S,
+    operation: &impl AsOperation,
+    operation_id: &OperationId,
+) -> Result<DocumentId, DomainError> {
+    let document_id = match operation.action() {
+        OperationAction::Create => {
+            // Derive the document id for this new document.
+            Ok::<DocumentId, DomainError>(DocumentId::new(operation_id))
+        }
+        _ => {
+            // We can unwrap previous operations here as we know all UPDATE and DELETE operations contain them.
+            let previous = operation.previous().unwrap();
+
+            // Validate claimed schema for this operation matches the expected found in the
+            // previous operations.
+            validate_claimed_schema_id(store, &operation.schema_id(), &previous).await?;
+
+            // Get the document_id for the document_view_id contained in previous operations.
+            // This performs several validation steps (check method doc string).
+            let document_id = get_checked_document_id_for_view_id(store, &previous).await?;
+
+            Ok(document_id)
+        }
+    }?;
+
+    // Ensure the document isn't deleted.
+    ensure_document_not_deleted(store, &document_id)
+        .await
+        .map_err(|_| DomainError::DeletedDocument)?;
+
+    Ok(document_id)
+}
+
+#[derive(Debug, Clone)]
+pub struct SyncIngest {
+    tx: ServiceSender,
+    schema_provider: SchemaProvider,
+}
+
+impl SyncIngest {
+    pub fn new(schema_provider: SchemaProvider, tx: ServiceSender) -> Self {
+        Self {
+            tx,
+            schema_provider,
+        }
+    }
+
+    pub async fn handle_entry(
+        &self,
+        store: &SqlStore,
+        encoded_entry: &EncodedEntry,
+        encoded_operation: &EncodedOperation,
+    ) -> Result<(), IngestError> {
+        let entry = decode_entry(encoded_entry)?;
+        let plain_operation = decode_operation(encoded_operation)?;
+
+        let schema = self
+            .schema_provider
+            .get(plain_operation.schema_id())
+            .await
+            .ok_or_else(|| IngestError::UnsupportedSchema)?;
+
+        let (operation, operation_id) = validate_entry_and_operation(
+            store,
+            &schema,
+            &entry,
+            encoded_entry,
+            &plain_operation,
+            encoded_operation,
+        )
+        .await?;
+
+        let document_id = determine_document_id(store, &operation, &operation_id).await?;
+
+        // If the entries' seq num is 1 we insert a new log here
+        if entry.seq_num().is_first() {
+            store
+                .insert_log(
+                    entry.log_id(),
+                    entry.public_key(),
+                    Schematic::schema_id(&operation),
+                    &document_id,
+                )
+                .await
+                .expect("Fatal database error");
+        }
+
+        store
+            .insert_entry(&entry, encoded_entry, Some(encoded_operation))
+            .await
+            .expect("Fatal database error");
+
+        store
+            .insert_operation(&operation_id, entry.public_key(), &operation, &document_id)
+            .await
+            .expect("Fatal database error");
+
+        // Inform other services about received data
+        let operation_id: OperationId = encoded_entry.hash().into();
+
+        if self
+            .tx
+            .send(ServiceMessage::NewOperation(operation_id))
+            .is_err()
+        {
+            // Silently fail here as we don't mind if there are no subscribers. We have
+            // tests in other places to check if messages arrive.
+        }
+
+        Ok(())
+    }
+}
diff --git a/aquadoggo/src/replication/manager.rs b/aquadoggo/src/replication/manager.rs
index 7c1e360de..d2e037937 100644
--- a/aquadoggo/src/replication/manager.rs
+++ b/aquadoggo/src/replication/manager.rs
@@ -3,16 +3,47 @@
 use std::collections::HashMap;
 
 use anyhow::Result;
+use p2panda_rs::entry::EncodedEntry;
+use p2panda_rs::operation::EncodedOperation;
 
+use crate::db::SqlStore;
 use crate::replication::errors::ReplicationError;
-use crate::replication::{Message, Mode, Session, SessionId, SessionState, SyncMessage, TargetSet};
+use crate::replication::{
+    Message, Mode, Session, SessionId, SessionState, SyncIngest, SyncMessage, TargetSet,
+};
 
 pub const INITIAL_SESSION_ID: SessionId = 0;
 
 pub const SUPPORTED_MODES: [Mode; 1] = [Mode::Naive];
 
+pub const SUPPORT_LIVE_MODE: bool = false;
+
+fn to_sync_messages(session_id: SessionId, messages: Vec<Message>) -> Vec<SyncMessage> {
+    messages
+        .into_iter()
+        .map(|message| SyncMessage::new(session_id, message))
+        .collect()
+}
+
+#[derive(Clone, Debug)]
+pub struct SyncResult {
+    pub messages: Vec<SyncMessage>,
+    pub is_done: bool,
+}
+
+impl SyncResult {
+    pub fn from_messages(session_id: SessionId, messages: Vec<Message>, is_done: bool) -> Self {
+        Self {
+            is_done,
+            messages: to_sync_messages(session_id, messages),
+        }
+    }
+}
+
 #[derive(Debug)]
 pub struct SyncManager<P> {
+    store: SqlStore,
+    ingest: SyncIngest,
     local_peer: P,
     sessions: HashMap<P, Vec<Session>>,
 }
@@ -21,9 +52,11 @@ impl<P> SyncManager<P>
 where
     P: Clone + std::hash::Hash + Eq + PartialOrd,
 {
-    pub fn new(local_peer: P) -> Self {
+    pub fn new(store: SqlStore, ingest: SyncIngest, local_peer: P) -> Self {
         Self {
+            store,
             local_peer,
+            ingest,
             sessions: HashMap::new(),
         }
     }
@@ -37,8 +70,29 @@ where
             .to_owned()
     }
 
+    /// Register a new session in manager and retrieve initial messages from it.
+    async fn insert_and_initialize_session(
+        &mut self,
+        remote_peer: &P,
+        session_id: &SessionId,
+        target_set: &TargetSet,
+        mode: &Mode,
+        local: bool,
+    ) -> Vec<Message> {
+        let mut session = Session::new(session_id, target_set, mode, local, SUPPORT_LIVE_MODE);
+        let initial_messages = session.initial_messages(&self.store).await;
+
+        if let Some(sessions) = self.sessions.get_mut(remote_peer) {
+            sessions.push(session);
+        } else {
+            self.sessions.insert(remote_peer.clone(), vec![session]);
+        }
+
+        initial_messages
+    }
+
     /// Register a new session in manager.
-    fn insert_session(
+    async fn insert_session(
         &mut self,
         remote_peer: &P,
         session_id: &SessionId,
@@ -46,7 +100,7 @@ where
         mode: &Mode,
         local: bool,
     ) {
-        let session = Session::new(session_id, target_set, mode, local);
+        let session = Session::new(session_id, target_set, mode, local, SUPPORT_LIVE_MODE);
 
         if let Some(sessions) = self.sessions.get_mut(remote_peer) {
             sessions.push(session);
@@ -55,12 +109,36 @@ where
         }
     }
 
-    pub fn initiate_session(
+    fn remove_session(&mut self, remote_peer: &P, session_id: &SessionId) {
+        let sessions = self.sessions.get_mut(remote_peer);
+
+        if let Some(sessions) = sessions {
+            if let Some((index, _)) = sessions
+                .iter()
+                .enumerate()
+                .find(|(_, session)| session.id == *session_id)
+            {
+                sessions.remove(index);
+            }
+        }
+    }
+
+    // @TODO: Make error type smaller in size
+    #[allow(clippy::result_large_err)]
+    fn is_mode_supported(mode: &Mode) -> Result<(), ReplicationError> {
+        if !SUPPORTED_MODES.contains(mode) {
+            return Err(ReplicationError::UnsupportedMode);
+        }
+
+        Ok(())
+    }
+
+    pub async fn initiate_session(
         &mut self,
         remote_peer: &P,
         target_set: &TargetSet,
         mode: &Mode,
-    ) -> Result<(), ReplicationError> {
+    ) -> Result<Vec<SyncMessage>, ReplicationError> {
         SyncManager::<P>::is_mode_supported(mode)?;
 
         let sessions = self.get_sessions(remote_peer);
@@ -83,36 +161,29 @@ where
             }
         };
 
-        self.insert_session(remote_peer, &session_id, target_set, mode, true);
+        // Ignore initial messages when we initiated the session, they will come from the other
+        // peer
+        self.insert_session(remote_peer, &session_id, target_set, mode, true)
+            .await;
 
-        Ok(())
+        Ok(vec![SyncMessage::new(
+            session_id,
+            Message::SyncRequest(mode.clone(), target_set.clone()),
+        )])
     }
 
-    fn is_mode_supported(mode: &Mode) -> Result<(), ReplicationError> {
-        if !SUPPORTED_MODES.contains(mode) {
-            return Err(ReplicationError::UnsupportedMode);
-        }
-
-        Ok(())
-    }
-
-    fn handle_duplicate_session(
+    async fn handle_duplicate_session(
         &mut self,
         remote_peer: &P,
         target_set: &TargetSet,
-        index: usize,
         session: &Session,
-    ) -> Result<(), ReplicationError> {
+    ) -> Result<SyncResult, ReplicationError> {
         let accept_inbound_request = match session.state {
             // Handle only duplicate sessions when they haven't started yet
             SessionState::Pending => {
                 if &self.local_peer < remote_peer {
                     // Drop our pending session
-                    let sessions = self
-                        .sessions
-                        .get_mut(remote_peer)
-                        .expect("Expected at least one pending session");
-                    sessions.remove(index);
+                    self.remove_session(remote_peer, &session.id);
 
                     // Accept the inbound request
                     true
@@ -124,44 +195,56 @@ where
             _ => return Err(ReplicationError::DuplicateInboundRequest(session.id)),
         };
 
-        if accept_inbound_request {
-            self.insert_session(remote_peer, &session.id, target_set, &session.mode(), false);
+        let mut all_messages: Vec<SyncMessage> = vec![];
 
-            // @TODO: Session needs to generate some messages on creation and
-            // it will pass them back up to us to then forward onto
-            // the swarm
+        if accept_inbound_request {
+            let messages = self
+                .insert_and_initialize_session(
+                    remote_peer,
+                    &session.id,
+                    target_set,
+                    &session.mode(),
+                    false,
+                )
+                .await;
+            all_messages.extend(to_sync_messages(session.id, messages));
 
             // If we dropped our own outbound session request regarding a different target set, we
             // need to re-establish it with another session id, otherwise it would get lost
             if session.target_set() != *target_set {
-                self.initiate_session(remote_peer, target_set, &session.mode())?;
-                // @TODO: Again, the new session will generate a message
-                // which we send onto the swarm
+                let messages = self
+                    .initiate_session(remote_peer, target_set, &session.mode())
+                    .await?;
+                all_messages.extend(messages)
             }
         }
 
-        Ok(())
+        Ok(SyncResult {
+            messages: all_messages,
+            is_done: false,
+        })
     }
 
-    fn handle_sync_request(
+    async fn handle_sync_request(
         &mut self,
         remote_peer: &P,
         mode: &Mode,
         session_id: &SessionId,
         target_set: &TargetSet,
-    ) -> Result<(), ReplicationError> {
+    ) -> Result<SyncResult, ReplicationError> {
         SyncManager::<P>::is_mode_supported(mode)?;
 
         let sessions = self.get_sessions(remote_peer);
 
         // Check if a session with this id already exists for this peer, this can happen if both
         // peers started to initiate a session at the same time, we can try to resolve this
-        if let Some((index, session)) = sessions
+        if let Some(session) = sessions
             .iter()
-            .enumerate()
-            .find(|(_, session)| session.id == *session_id && session.local)
+            .find(|session| session.id == *session_id && session.local)
         {
-            return self.handle_duplicate_session(remote_peer, target_set, index, session);
+            return self
+                .handle_duplicate_session(remote_peer, target_set, session)
+                .await;
         }
 
         // Check if a session with this target set already exists for this peer, this always gets
@@ -173,33 +256,128 @@ where
             return Err(ReplicationError::DuplicateInboundRequest(session.id));
         }
 
-        self.insert_session(remote_peer, session_id, target_set, mode, false);
+        let messages = self
+            .insert_and_initialize_session(remote_peer, session_id, target_set, mode, false)
+            .await;
 
-        Ok(())
+        Ok(SyncResult::from_messages(*session_id, messages, false))
+    }
+
+    async fn handle_session_message(
+        &mut self,
+        remote_peer: &P,
+        session_id: &SessionId,
+        message: &Message,
+    ) -> Result<SyncResult, ReplicationError> {
+        let sessions = self.sessions.get_mut(remote_peer);
+
+        let (is_both_done, messages) = match sessions {
+            Some(sessions) => {
+                // Check if a session exists with the given id for this peer.
+                if let Some(session) = sessions
+                    .iter_mut()
+                    .find(|session| session.id == *session_id)
+                {
+                    // Pass the message onto the session when found.
+                    let messages = session.handle_message(&self.store, message).await?;
+                    let is_both_done = session.state == SessionState::Done;
+                    Ok((is_both_done, messages))
+                } else {
+                    Err(ReplicationError::NoSessionFound(*session_id))
+                }
+            }
+            None => Err(ReplicationError::NoSessionFound(*session_id)),
+        }?;
+
+        // We're done, clean up after ourselves
+        if is_both_done {
+            self.remove_session(remote_peer, session_id);
+        }
+
+        Ok(SyncResult::from_messages(
+            *session_id,
+            messages,
+            is_both_done,
+        ))
     }
 
-    pub fn handle_message(
+    async fn handle_entry(
+        &mut self,
+        remote_peer: &P,
+        session_id: &SessionId,
+        entry_bytes: &EncodedEntry,
+        operation_bytes: &Option<EncodedOperation>,
+    ) -> Result<SyncResult, ReplicationError> {
+        let mut sessions = self.get_sessions(remote_peer);
+
+        if let Some(session) = sessions
+            .iter_mut()
+            .find(|session| session.id == *session_id)
+        {
+            session.validate_entry(entry_bytes, operation_bytes.as_ref())?;
+
+            self.ingest
+                .handle_entry(
+                    &self.store,
+                    entry_bytes,
+                    // @TODO: This should handle entries with removed payloads
+                    operation_bytes
+                        .as_ref()
+                        .expect("For now we always expect an operation here"),
+                )
+                .await?;
+
+            Ok(SyncResult {
+                messages: vec![],
+                is_done: session.state == SessionState::Done,
+            })
+        } else {
+            Err(ReplicationError::NoSessionFound(*session_id))
+        }
+    }
+
+    pub async fn handle_message(
         &mut self,
         remote_peer: &P,
         sync_message: &SyncMessage,
-    ) -> Result<(), ReplicationError> {
+    ) -> Result<SyncResult, ReplicationError> {
         match sync_message.message() {
             Message::SyncRequest(mode, target_set) => {
                 self.handle_sync_request(remote_peer, mode, &sync_message.session_id(), target_set)
+                    .await
+            }
+            Message::Entry(entry_bytes, operation_bytes) => {
+                self.handle_entry(
+                    remote_peer,
+                    &sync_message.session_id(),
+                    entry_bytes,
+                    operation_bytes,
+                )
+                .await
+            }
+            message => {
+                self.handle_session_message(remote_peer, &sync_message.session_id(), message)
+                    .await
             }
-            _ => todo!(),
         }
     }
 }
 
 #[cfg(test)]
 mod tests {
+    use p2panda_rs::test_utils::memory_store::helpers::PopulateStoreConfig;
     use rstest::rstest;
+    use tokio::sync::broadcast;
 
     use crate::replication::errors::ReplicationError;
-    use crate::replication::message::Message;
-    use crate::replication::{Mode, SyncMessage, TargetSet};
+    use crate::replication::message::{Message, HAVE_TYPE, SYNC_DONE_TYPE};
+    use crate::replication::{Mode, SyncIngest, SyncMessage, TargetSet};
+    use crate::schema::SchemaProvider;
     use crate::test_utils::helpers::random_target_set;
+    use crate::test_utils::{
+        populate_and_materialize, populate_store_config, test_runner, test_runner_with_manager,
+        TestNode, TestNodeManager,
+    };
 
     use super::{SyncManager, INITIAL_SESSION_ID};
 
@@ -211,21 +389,31 @@ mod tests {
         #[from(random_target_set)] target_set_1: TargetSet,
         #[from(random_target_set)] target_set_2: TargetSet,
     ) {
-        let mode = Mode::Naive;
-
-        let mut manager = SyncManager::new(PEER_ID_LOCAL);
-        let result = manager.initiate_session(&PEER_ID_REMOTE, &target_set_1, &mode);
-        assert!(result.is_ok());
-
-        let result = manager.initiate_session(&PEER_ID_REMOTE, &target_set_2, &mode);
-        assert!(result.is_ok());
-
-        // Expect error when initiating a session for the same target set
-        let result = manager.initiate_session(&PEER_ID_REMOTE, &target_set_1, &mode);
-        assert!(matches!(
-            result,
-            Err(ReplicationError::DuplicateOutboundRequest(0))
-        ));
+        test_runner(move |node: TestNode| async move {
+            let mode = Mode::Naive;
+            let (tx, _rx) = broadcast::channel(8);
+            let ingest = SyncIngest::new(SchemaProvider::default(), tx);
+
+            let mut manager = SyncManager::new(node.context.store.clone(), ingest, PEER_ID_LOCAL);
+            let result = manager
+                .initiate_session(&PEER_ID_REMOTE, &target_set_1, &mode)
+                .await;
+            assert!(result.is_ok());
+
+            let result = manager
+                .initiate_session(&PEER_ID_REMOTE, &target_set_2, &mode)
+                .await;
+            assert!(result.is_ok());
+
+            // Expect error when initiating a session for the same target set
+            let result = manager
+                .initiate_session(&PEER_ID_REMOTE, &target_set_1, &mode)
+                .await;
+            assert!(matches!(
+                result,
+                Err(ReplicationError::DuplicateOutboundRequest(0))
+            ));
+        })
     }
 
     #[rstest]
@@ -233,51 +421,163 @@ mod tests {
         #[from(random_target_set)] target_set_1: TargetSet,
         #[from(random_target_set)] target_set_2: TargetSet,
     ) {
-        let mut manager = SyncManager::new(PEER_ID_LOCAL);
-
-        let message = SyncMessage::new(0, Message::SyncRequest(Mode::Naive, target_set_1.clone()));
-        let result = manager.handle_message(&PEER_ID_REMOTE, &message);
-        assert!(result.is_ok());
-
-        let message = SyncMessage::new(1, Message::SyncRequest(Mode::Naive, target_set_2.clone()));
-        let result = manager.handle_message(&PEER_ID_REMOTE, &message);
-        assert!(result.is_ok());
-
-        // Reject attempt to create session again
-        let message = SyncMessage::new(0, Message::SyncRequest(Mode::Naive, target_set_1.clone()));
-        let result = manager.handle_message(&PEER_ID_REMOTE, &message);
-        assert!(matches!(
-            result,
-            Err(ReplicationError::DuplicateInboundRequest(0))
-        ));
-
-        // Reject different session concerning same target set
-        let message = SyncMessage::new(2, Message::SyncRequest(Mode::Naive, target_set_2.clone()));
-        let result = manager.handle_message(&PEER_ID_REMOTE, &message);
-        assert!(matches!(
-            result,
-            Err(ReplicationError::DuplicateInboundRequest(1))
-        ));
+        test_runner(move |node: TestNode| async move {
+            let (tx, _rx) = broadcast::channel(8);
+            let ingest = SyncIngest::new(SchemaProvider::default(), tx);
+
+            let mut manager = SyncManager::new(node.context.store.clone(), ingest, PEER_ID_LOCAL);
+
+            let message =
+                SyncMessage::new(0, Message::SyncRequest(Mode::Naive, target_set_1.clone()));
+            let result = manager.handle_message(&PEER_ID_REMOTE, &message).await;
+            assert!(result.is_ok());
+
+            let message =
+                SyncMessage::new(1, Message::SyncRequest(Mode::Naive, target_set_2.clone()));
+            let result = manager.handle_message(&PEER_ID_REMOTE, &message).await;
+            assert!(result.is_ok());
+
+            // Reject attempt to create session again
+            let message =
+                SyncMessage::new(0, Message::SyncRequest(Mode::Naive, target_set_1.clone()));
+            let result = manager.handle_message(&PEER_ID_REMOTE, &message).await;
+            assert!(matches!(
+                result,
+                Err(ReplicationError::DuplicateInboundRequest(0))
+            ));
+
+            // Reject different session concerning same target set
+            let message =
+                SyncMessage::new(2, Message::SyncRequest(Mode::Naive, target_set_2.clone()));
+            let result = manager.handle_message(&PEER_ID_REMOTE, &message).await;
+            assert!(matches!(
+                result,
+                Err(ReplicationError::DuplicateInboundRequest(1))
+            ));
+        })
     }
 
     #[rstest]
     fn inbound_checks_supported_mode(#[from(random_target_set)] target_set: TargetSet) {
-        // Should not fail when requesting supported replication mode
-        let mut manager = SyncManager::new(PEER_ID_LOCAL);
-        let message = SyncMessage::new(
-            INITIAL_SESSION_ID,
-            Message::SyncRequest(Mode::Naive, target_set.clone()),
-        );
-        let result = manager.handle_message(&PEER_ID_REMOTE, &message);
-        assert!(result.is_ok());
-
-        // Should fail when requesting unsupported replication mode
-        let mut manager = SyncManager::new(PEER_ID_LOCAL);
-        let message = SyncMessage::new(
-            INITIAL_SESSION_ID,
-            Message::SyncRequest(Mode::SetReconciliation, target_set.clone()),
-        );
-        let result = manager.handle_message(&PEER_ID_REMOTE, &message);
-        assert!(result.is_err());
+        test_runner(move |node: TestNode| async move {
+            let (tx, _rx) = broadcast::channel(8);
+            let ingest = SyncIngest::new(SchemaProvider::default(), tx);
+
+            // Should not fail when requesting supported replication mode
+            let mut manager =
+                SyncManager::new(node.context.store.clone(), ingest.clone(), PEER_ID_LOCAL);
+            let message = SyncMessage::new(
+                INITIAL_SESSION_ID,
+                Message::SyncRequest(Mode::Naive, target_set.clone()),
+            );
+            let result = manager.handle_message(&PEER_ID_REMOTE, &message).await;
+            assert!(result.is_ok());
+
+            // Should fail when requesting unsupported replication mode
+            let mut manager = SyncManager::new(node.context.store.clone(), ingest, PEER_ID_LOCAL);
+            let message = SyncMessage::new(
+                INITIAL_SESSION_ID,
+                Message::SyncRequest(Mode::SetReconciliation, target_set.clone()),
+            );
+            let result = manager.handle_message(&PEER_ID_REMOTE, &message).await;
+            assert!(result.is_err());
+        })
+    }
+
+    #[rstest]
+    fn sync_lifetime(
+        #[from(populate_store_config)]
+        #[with(2, 1, 3)]
+        config_a: PopulateStoreConfig,
+        #[from(populate_store_config)] config_b: PopulateStoreConfig,
+    ) {
+        test_runner_with_manager(|manager: TestNodeManager| async move {
+            let mut node_a = manager.create().await;
+            let mut node_b = manager.create().await;
+
+            populate_and_materialize(&mut node_a, &config_a).await;
+            populate_and_materialize(&mut node_b, &config_b).await;
+
+            let (tx, _rx) = broadcast::channel(8);
+            let target_set = TargetSet::new(&vec![config_a.schema.id().to_owned()]);
+
+            let mut manager_a = SyncManager::new(
+                node_a.context.store.clone(),
+                SyncIngest::new(node_a.context.schema_provider.clone(), tx.clone()),
+                PEER_ID_LOCAL,
+            );
+
+            let mut manager_b = SyncManager::new(
+                node_b.context.store.clone(),
+                SyncIngest::new(node_b.context.schema_provider.clone(), tx),
+                PEER_ID_REMOTE,
+            );
+
+            // Send `SyncRequest` to remote
+            let messages = manager_a
+                .initiate_session(&PEER_ID_REMOTE, &target_set, &Mode::Naive)
+                .await
+                .unwrap();
+
+            assert_eq!(
+                messages,
+                vec![SyncMessage::new(
+                    0,
+                    Message::SyncRequest(Mode::Naive, target_set.clone())
+                )]
+            );
+
+            // Receive `Have` and `SyncDone` from remote
+            let result = manager_b
+                .handle_message(&PEER_ID_LOCAL, &messages[0])
+                .await
+                .unwrap();
+
+            assert_eq!(result.is_done, false);
+            assert_eq!(
+                result.messages,
+                vec![
+                    SyncMessage::new(0, Message::Have(vec![])),
+                    SyncMessage::new(0, Message::SyncDone(false)),
+                ]
+            );
+
+            // Send `Have`, `Entry` and `SyncDone` messages to remote
+            let result_have = manager_a
+                .handle_message(&PEER_ID_REMOTE, &result.messages[0])
+                .await
+                .unwrap();
+            assert_eq!(result_have.is_done, false);
+
+            let result_done = manager_a
+                .handle_message(&PEER_ID_REMOTE, &result.messages[1])
+                .await
+                .unwrap();
+            assert_eq!(result_done.is_done, true);
+
+            assert_eq!(result_have.messages.len(), 8);
+            assert_eq!(
+                result_have.messages.first().unwrap().message_type(),
+                HAVE_TYPE
+            );
+            assert_eq!(
+                result_have.messages.last().unwrap().message_type(),
+                SYNC_DONE_TYPE
+            );
+
+            // Receive `SyncDone` from remote
+            for (index, message) in result_have.messages.iter().enumerate() {
+                let result = manager_b
+                    .handle_message(&PEER_ID_LOCAL, &message)
+                    .await
+                    .unwrap();
+
+                // We're done when we received the last message (SyncDone)
+                assert_eq!(result.is_done, index == result_have.messages.len() - 1);
+
+                // We don't expect any messages anymore, Manager B didn't have any data
+                assert!(result.messages.is_empty());
+            }
+        })
     }
 }
diff --git a/aquadoggo/src/replication/message.rs b/aquadoggo/src/replication/message.rs
index 3d25b4404..d310d677f 100644
--- a/aquadoggo/src/replication/message.rs
+++ b/aquadoggo/src/replication/message.rs
@@ -2,6 +2,10 @@
 
 use std::fmt;
 
+use p2panda_rs::entry::EncodedEntry;
+use p2panda_rs::entry::{LogId, SeqNum};
+use p2panda_rs::identity::PublicKey;
+use p2panda_rs::operation::EncodedOperation;
 use serde::de::Visitor;
 use serde::ser::SerializeSeq;
 use serde::{Deserialize, Serialize};
@@ -11,29 +15,29 @@ use crate::replication::{Mode, SessionId, TargetSet};
 pub const SYNC_REQUEST_TYPE: MessageType = 0;
 pub const ENTRY_TYPE: MessageType = 8;
 pub const SYNC_DONE_TYPE: MessageType = 9;
+pub const HAVE_TYPE: MessageType = 10;
 
 pub type MessageType = u64;
 
-// @TODO
-#[derive(Clone, Debug)]
-pub enum StrategyMessage {
-    Have,
-    Entry,
-}
+pub type LiveMode = bool;
+
+pub type LogHeight = (PublicKey, Vec<(LogId, SeqNum)>);
 
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub enum Message {
     SyncRequest(Mode, TargetSet),
-    Entry(Vec<u8>, Vec<u8>),
-    SyncDone(bool),
+    Entry(EncodedEntry, Option<EncodedOperation>),
+    SyncDone(LiveMode),
+    Have(Vec<LogHeight>),
 }
 
 impl Message {
     pub fn message_type(&self) -> MessageType {
         match self {
-            Message::SyncRequest { .. } => SYNC_REQUEST_TYPE,
+            Message::SyncRequest(_, _) => SYNC_REQUEST_TYPE,
             Message::Entry(_, _) => ENTRY_TYPE,
             Message::SyncDone(_) => SYNC_DONE_TYPE,
+            Message::Have(_) => HAVE_TYPE,
         }
     }
 }
@@ -89,6 +93,11 @@ impl Serialize for SyncMessage {
                 seq.serialize_element(live_mode)?;
                 seq.end()
             }
+            Message::Have(log_heights) => {
+                let mut seq = serialize_header(serializer.serialize_seq(Some(3))?)?;
+                seq.serialize_element(log_heights)?;
+                seq.end()
+            }
         }
     }
 }
@@ -130,13 +139,11 @@ impl<'de> Deserialize<'de> for SyncMessage {
 
                     Ok(Message::SyncRequest(mode, target_set))
                 } else if message_type == ENTRY_TYPE {
-                    let entry_bytes: Vec<u8> = seq.next_element()?.ok_or_else(|| {
+                    let entry_bytes: EncodedEntry = seq.next_element()?.ok_or_else(|| {
                         serde::de::Error::custom("missing entry bytes in entry message")
                     })?;
 
-                    let operation_bytes: Vec<u8> = seq.next_element()?.ok_or_else(|| {
-                        serde::de::Error::custom("missing operation bytes in entry message")
-                    })?;
+                    let operation_bytes: Option<EncodedOperation> = seq.next_element()?;
 
                     Ok(Message::Entry(entry_bytes, operation_bytes))
                 } else if message_type == SYNC_DONE_TYPE {
@@ -145,6 +152,13 @@ impl<'de> Deserialize<'de> for SyncMessage {
                     })?;
 
                     Ok(Message::SyncDone(live_mode))
+                } else if message_type == HAVE_TYPE {
+                    let log_heights: Vec<(PublicKey, Vec<(LogId, SeqNum)>)> =
+                        seq.next_element()?.ok_or_else(|| {
+                            serde::de::Error::custom("missing log heights in have message")
+                        })?;
+
+                    Ok(Message::Have(log_heights))
                 } else {
                     Err(serde::de::Error::custom(format!(
                         "unknown message type {} in replication message",
@@ -172,7 +186,10 @@ impl<'de> Deserialize<'de> for SyncMessage {
 mod tests {
     use ciborium::cbor;
     use ciborium::value::{Error, Value};
+    use p2panda_rs::entry::{LogId, SeqNum};
+    use p2panda_rs::identity::PublicKey;
     use p2panda_rs::serde::{deserialize_into, serialize_from, serialize_value};
+    use p2panda_rs::test_utils::fixtures::public_key;
     use rstest::rstest;
 
     use crate::replication::{Mode, TargetSet};
@@ -181,7 +198,7 @@ mod tests {
     use super::{Message, SyncMessage};
 
     #[rstest]
-    fn serialize(#[from(random_target_set)] target_set: TargetSet) {
+    fn serialize(#[from(random_target_set)] target_set: TargetSet, public_key: PublicKey) {
         assert_eq!(
             serialize_from(SyncMessage::new(
                 51,
@@ -189,15 +206,35 @@ mod tests {
             )),
             serialize_value(cbor!([0, 51, 1, target_set]))
         );
+
+        let log_heights = vec![(public_key, vec![(LogId::default(), SeqNum::default())])];
+        assert_eq!(
+            serialize_from(SyncMessage::new(51, Message::Have(log_heights.clone()))),
+            serialize_value(cbor!([10, 51, log_heights]))
+        );
     }
 
     #[rstest]
-    fn deserialize(#[from(random_target_set)] target_set: TargetSet) {
+    fn deserialize(#[from(random_target_set)] target_set: TargetSet, public_key: PublicKey) {
         assert_eq!(
             deserialize_into::<SyncMessage>(&serialize_value(cbor!([0, 12, 0, target_set])))
                 .unwrap(),
             SyncMessage::new(12, Message::SyncRequest(Mode::Naive, target_set.clone()))
         );
+
+        let log_heights: Vec<(PublicKey, Vec<(LogId, SeqNum)>)> = vec![];
+        assert_eq!(
+            deserialize_into::<SyncMessage>(&serialize_value(cbor!([10, 12, log_heights])))
+                .unwrap(),
+            SyncMessage::new(12, Message::Have(vec![]))
+        );
+
+        let log_heights = vec![(public_key, vec![(LogId::default(), SeqNum::default())])];
+        assert_eq!(
+            deserialize_into::<SyncMessage>(&serialize_value(cbor!([10, 12, log_heights.clone()])))
+                .unwrap(),
+            SyncMessage::new(12, Message::Have(log_heights))
+        );
     }
 
     #[rstest]
diff --git a/aquadoggo/src/replication/mod.rs b/aquadoggo/src/replication/mod.rs
index 57d0d7ca1..bb943f877 100644
--- a/aquadoggo/src/replication/mod.rs
+++ b/aquadoggo/src/replication/mod.rs
@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 pub mod errors;
+mod ingest;
 mod manager;
 mod message;
 mod mode;
@@ -9,8 +10,9 @@ mod strategies;
 mod target_set;
 pub mod traits;
 
+pub use ingest::SyncIngest;
 pub use manager::SyncManager;
-pub use message::{Message, StrategyMessage, SyncMessage};
+pub use message::{LiveMode, LogHeight, Message, SyncMessage};
 pub use mode::Mode;
 pub use session::{Session, SessionId, SessionState};
 pub use strategies::{NaiveStrategy, SetReconciliationStrategy, StrategyResult};
diff --git a/aquadoggo/src/replication/session.rs b/aquadoggo/src/replication/session.rs
index 9bb54a5b0..445d72537 100644
--- a/aquadoggo/src/replication/session.rs
+++ b/aquadoggo/src/replication/session.rs
@@ -1,11 +1,20 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+use p2panda_rs::entry::EncodedEntry;
+use p2panda_rs::operation::decode::decode_operation;
+use p2panda_rs::operation::traits::Schematic;
+use p2panda_rs::operation::EncodedOperation;
+
+use crate::db::SqlStore;
+use crate::replication::errors::ReplicationError;
 use crate::replication::traits::Strategy;
-use crate::replication::{Mode, NaiveStrategy, SetReconciliationStrategy, TargetSet};
+use crate::replication::{
+    Message, Mode, NaiveStrategy, SetReconciliationStrategy, StrategyResult, TargetSet,
+};
 
 pub type SessionId = u64;
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Eq, PartialEq)]
 pub enum SessionState {
     Pending,
     Established,
@@ -14,8 +23,6 @@ pub enum SessionState {
 
 #[derive(Clone, Debug)]
 pub struct Session {
-    // @TODO: Access to the store
-    // store: Store
     /// Unique identifier of this session for that peer.
     pub id: SessionId,
 
@@ -27,24 +34,50 @@ pub struct Session {
 
     /// Replication strategy handler.
     pub strategy: Box<dyn Strategy>,
+
+    /// True if we're done locally with this replication session.
+    pub is_local_done: bool,
+
+    /// True if we received a `SyncDone` message from the remote peer.
+    pub is_remote_done: bool,
+
+    /// True if the we support live-mode with remote peer.
+    pub is_local_live_mode: bool,
+
+    /// True if the remote peer suggested entering live-mode.
+    pub is_remote_live_mode: bool,
 }
 
 impl Session {
-    pub fn new(id: &SessionId, target_set: &TargetSet, mode: &Mode, local: bool) -> Self {
+    pub fn new(
+        id: &SessionId,
+        target_set: &TargetSet,
+        mode: &Mode,
+        local: bool,
+        live_mode: bool,
+    ) -> Self {
         let strategy: Box<dyn Strategy> = match mode {
-            Mode::Naive => Box::new(NaiveStrategy::new(target_set, mode)),
+            Mode::Naive => Box::new(NaiveStrategy::new(target_set)),
             Mode::SetReconciliation => Box::new(SetReconciliationStrategy::new()),
-            Mode::Unknown => panic!("Unknown replication mode found"),
+            Mode::Unknown => panic!("Unknown replication mode"),
         };
 
         Self {
             id: *id,
             state: SessionState::Pending,
-            strategy,
             local,
+            strategy,
+            is_local_done: false,
+            is_remote_done: false,
+            is_remote_live_mode: false,
+            is_local_live_mode: live_mode,
         }
     }
 
+    pub fn live_mode(&self) -> bool {
+        self.is_local_live_mode && self.is_remote_live_mode
+    }
+
     pub fn mode(&self) -> Mode {
         self.strategy.mode()
     }
@@ -52,4 +85,136 @@ impl Session {
     pub fn target_set(&self) -> TargetSet {
         self.strategy.target_set()
     }
+
+    /// Send `SyncDone` message last as soon as the done flag flipped.
+    fn flippy_flaggy(&mut self, result: &mut StrategyResult) {
+        if result.is_local_done && !self.is_local_done {
+            result
+                .messages
+                .push(Message::SyncDone(self.is_local_live_mode));
+
+            self.is_local_done = result.is_local_done;
+        }
+    }
+
+    pub async fn initial_messages(&mut self, store: &SqlStore) -> Vec<Message> {
+        let mut result = self.strategy.initial_messages(store).await;
+        self.flippy_flaggy(&mut result);
+        result.messages
+    }
+
+    /// Validate entry and operation.
+    ///
+    /// This checks if the received data is actually what we've asked for.
+    // @TODO: Make error type smaller in size
+    #[allow(clippy::result_large_err)]
+    pub fn validate_entry(
+        &self,
+        _entry_bytes: &EncodedEntry,
+        operation_bytes: Option<&EncodedOperation>,
+    ) -> Result<(), ReplicationError> {
+        if let Some(operation_bytes) = operation_bytes {
+            let operation = decode_operation(operation_bytes).map_err(|_| {
+                ReplicationError::StrategyFailed("Could not decode operation".into())
+            })?;
+
+            if !self.target_set().contains(operation.schema_id()) {
+                return Err(ReplicationError::UnmatchedTargetSet);
+            }
+        }
+
+        Ok(())
+    }
+
+    pub async fn handle_message(
+        &mut self,
+        store: &SqlStore,
+        message: &Message,
+    ) -> Result<Vec<Message>, ReplicationError> {
+        let result = match message {
+            Message::SyncDone(live_mode) => {
+                self.is_remote_done = true;
+                self.is_remote_live_mode = *live_mode;
+                vec![]
+            }
+            message => {
+                let mut result = self.strategy.handle_message(store, message).await?;
+                self.flippy_flaggy(&mut result);
+                result.messages
+            }
+        };
+
+        // As soon as we've received any message from the remote peer we can consider the session
+        // to be "established"
+        if self.state == SessionState::Pending {
+            self.state = SessionState::Established;
+        }
+
+        // If local and remote peer decided they're done, we can consider this whole session to be
+        // "done"
+        if self.is_local_done && self.is_remote_done {
+            self.state = SessionState::Done;
+        }
+
+        Ok(result)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use p2panda_rs::test_utils::memory_store::helpers::{populate_store, PopulateStoreConfig};
+    use rstest::rstest;
+
+    use crate::replication::manager::INITIAL_SESSION_ID;
+    use crate::replication::{Message, Mode, SessionState, TargetSet};
+    use crate::test_utils::helpers::random_target_set;
+    use crate::test_utils::{populate_store_config, test_runner, TestNode};
+
+    use super::Session;
+
+    #[rstest]
+    fn state_machine(#[from(random_target_set)] target_set: TargetSet) {
+        test_runner(move |node: TestNode| async move {
+            let mut session =
+                Session::new(&INITIAL_SESSION_ID, &target_set, &Mode::Naive, true, false);
+            assert!(!session.is_local_done);
+            assert!(!session.is_local_live_mode);
+            assert!(!session.is_remote_live_mode);
+            assert!(!session.is_remote_done);
+            assert!(session.state == SessionState::Pending);
+
+            session
+                .handle_message(&node.context.store, &Message::SyncDone(true))
+                .await
+                .expect("No errors expected");
+            assert!(!session.is_local_done);
+            assert!(!session.is_local_live_mode);
+            assert!(session.is_remote_live_mode);
+            assert!(session.is_remote_done);
+            assert!(session.state == SessionState::Established);
+        })
+    }
+
+    #[rstest]
+    fn correct_strategy_messages(
+        #[from(populate_store_config)]
+        #[with(5, 2, 1)]
+        config: PopulateStoreConfig,
+    ) {
+        test_runner(move |node: TestNode| async move {
+            populate_store(&node.context.store, &config).await;
+
+            let target_set = TargetSet::new(&vec![config.schema.id().to_owned()]);
+            let mut session =
+                Session::new(&INITIAL_SESSION_ID, &target_set, &Mode::Naive, true, false);
+
+            let response_messages = session
+                .handle_message(&node.context.store, &Message::Have(vec![]))
+                .await
+                .unwrap();
+
+            // 1x Have + 10x Entry + 1x SyncDone = 12 messages
+            assert_eq!(response_messages.len(), 12);
+        });
+    }
 }
diff --git a/aquadoggo/src/replication/strategies.rs b/aquadoggo/src/replication/strategies.rs
deleted file mode 100644
index bafbf12fa..000000000
--- a/aquadoggo/src/replication/strategies.rs
+++ /dev/null
@@ -1,95 +0,0 @@
-// SPDX-License-Identifier: AGPL-3.0-or-later
-
-use anyhow::Result;
-use async_trait::async_trait;
-
-use crate::replication::traits::Strategy;
-use crate::replication::{Mode, StrategyMessage, TargetSet};
-
-#[derive(Clone, Debug)]
-pub struct StrategyResult {
-    is_done: bool,
-    messages: Vec<StrategyMessage>,
-}
-
-#[derive(Clone, Debug)]
-pub struct NaiveStrategy {
-    target_set: TargetSet,
-    mode: Mode,
-}
-
-impl NaiveStrategy {
-    pub fn new(target_set: &TargetSet, mode: &Mode) -> Self {
-        Self {
-            target_set: target_set.clone(),
-            mode: mode.clone(),
-        }
-    }
-}
-
-#[async_trait]
-impl Strategy for NaiveStrategy {
-    fn mode(&self) -> Mode {
-        self.mode.clone()
-    }
-
-    fn target_set(&self) -> TargetSet {
-        self.target_set.clone()
-    }
-
-    async fn initial_messages(&self) -> Vec<StrategyMessage> {
-        // TODO: Access the store and compose a have message which contains our local log heights over
-        // the TargetSet.
-        let _target_set = self.target_set();
-
-        vec![StrategyMessage::Have]
-    }
-
-    async fn handle_message(&self, message: StrategyMessage) -> Result<StrategyResult> {
-        // TODO: Verify that the TargetSet contained in the message is a sub-set of the passed
-        // local TargetSet.
-        let _target_set = self.target_set();
-        let messages = Vec::new();
-        let mut is_done = false;
-
-        match message {
-            StrategyMessage::Have => {
-                // Compose Have message and push to messages
-                is_done = true;
-            }
-            StrategyMessage::Entry => {
-                // self.handle_entry(..)
-            }
-        }
-
-        Ok(StrategyResult { is_done, messages })
-    }
-}
-
-#[derive(Clone, Debug)]
-pub struct SetReconciliationStrategy();
-
-impl SetReconciliationStrategy {
-    pub fn new() -> Self {
-        Self()
-    }
-}
-
-#[async_trait]
-impl Strategy for SetReconciliationStrategy {
-    fn mode(&self) -> Mode {
-        todo!()
-    }
-
-    fn target_set(&self) -> TargetSet {
-        todo!()
-    }
-
-    async fn initial_messages(&self) -> Vec<StrategyMessage> {
-        todo!()
-    }
-
-    async fn handle_message(&self, _message: StrategyMessage) -> Result<StrategyResult> {
-        todo!()
-    }
-}
diff --git a/aquadoggo/src/replication/strategies/diff.rs b/aquadoggo/src/replication/strategies/diff.rs
new file mode 100644
index 000000000..0956785bf
--- /dev/null
+++ b/aquadoggo/src/replication/strategies/diff.rs
@@ -0,0 +1,137 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use p2panda_rs::entry::{LogId, SeqNum};
+
+use crate::replication::LogHeight;
+
+pub fn diff_log_heights(
+    local_log_heights: &[LogHeight],
+    remote_log_heights: &[LogHeight],
+) -> Vec<LogHeight> {
+    let mut remote_needs = Vec::new();
+
+    for (local_author, local_author_logs) in local_log_heights {
+        // Helper for diffing local log heights against remote log heights.
+        let diff_logs = |(remote_log_id, remote_seq_num): (LogId, SeqNum)| {
+            // Get the remote log by it's id.
+            let local_log = local_author_logs
+                .iter()
+                .find(|(local_log_id, _)| remote_log_id == *local_log_id);
+
+            match local_log {
+                // If a log exists then compare heights of local and remote logs.
+                Some((log_id, local_seq_num)) => {
+                    // If the local log is higher we increment their log id (we want all entries
+                    // greater than or equal to this). Otherwise we return none.
+                    if local_seq_num > &remote_seq_num {
+                        // We can unwrap as we are incrementing the remote peers seq num here and
+                        // this means it's will not reach max seq number.
+                        Some((log_id.to_owned(), remote_seq_num.clone().next().unwrap()))
+                    } else {
+                        None
+                    }
+                }
+                // If no log exists then the remote has never had this log and they need all
+                // entries from seq num 1.
+                None => Some((remote_log_id.to_owned(), SeqNum::default())),
+            }
+        };
+
+        // Find local log for a public key sent by the remote peer.
+        //
+        // If none is found we don't do anything as this means we are missing entries they should
+        // send us.
+        if let Some((_, remote_author_logs)) = remote_log_heights
+            .iter()
+            .find(|(remote_author, _)| remote_author == local_author)
+        {
+            // Diff our local log heights against the remote.
+            let remote_needs_logs: Vec<(LogId, SeqNum)> = remote_author_logs
+                .iter()
+                .copied()
+                .filter_map(diff_logs)
+                .collect();
+
+            // If the remote needs at least one log we push it to the remote needs.
+            if !remote_needs_logs.is_empty() {
+                remote_needs.push((local_author.to_owned(), remote_needs_logs));
+            };
+        } else {
+            remote_needs.push((
+                local_author.to_owned(),
+                local_author_logs
+                    .iter()
+                    .map(|(log_id, _)| (*log_id, SeqNum::default()))
+                    .collect(),
+            ));
+        }
+    }
+
+    remote_needs
+}
+
+#[cfg(test)]
+mod tests {
+    use p2panda_rs::entry::{LogId, SeqNum};
+    use p2panda_rs::identity::KeyPair;
+    use p2panda_rs::test_utils::fixtures::random_key_pair;
+    use rstest::rstest;
+
+    use super::diff_log_heights;
+
+    #[rstest]
+    fn correctly_diffs_log_heights(#[from(random_key_pair)] author_a: KeyPair) {
+        let author_a = author_a.public_key();
+        let peer_a_log_heights = vec![(
+            author_a,
+            vec![
+                (LogId::new(0), SeqNum::new(5).unwrap()),
+                (LogId::new(1), SeqNum::new(2).unwrap()),
+            ],
+        )];
+        let peer_b_log_heights = vec![(
+            author_a,
+            vec![
+                (LogId::new(0), SeqNum::new(8).unwrap()),
+                (LogId::new(1), SeqNum::new(2).unwrap()),
+            ],
+        )];
+
+        let peer_b_needs = diff_log_heights(&peer_a_log_heights, &peer_b_log_heights);
+        let peer_a_needs = diff_log_heights(&peer_b_log_heights, &peer_a_log_heights);
+
+        assert_eq!(
+            peer_a_needs,
+            vec![(author_a, vec![(LogId::new(0), SeqNum::new(6).unwrap())])]
+        );
+        assert_eq!(peer_b_needs, vec![]);
+    }
+
+    #[rstest]
+    fn diff_when_remote_is_empty(#[from(random_key_pair)] author_a: KeyPair) {
+        let author_a = author_a.public_key();
+        let peer_a_log_heights = vec![(
+            author_a,
+            vec![
+                (LogId::new(0), SeqNum::new(5).unwrap()),
+                (LogId::new(1), SeqNum::new(5).unwrap()),
+            ],
+        )];
+        let peer_b_log_heights = vec![];
+
+        let peer_b_needs = diff_log_heights(&peer_a_log_heights, &peer_b_log_heights);
+        let peer_a_needs = diff_log_heights(&peer_b_log_heights, &peer_a_log_heights);
+
+        assert_eq!(peer_a_needs, vec![]);
+        assert_eq!(
+            peer_b_needs,
+            vec![(
+                author_a,
+                vec![
+                    (LogId::new(0), SeqNum::new(1).unwrap()),
+                    (LogId::new(1), SeqNum::new(1).unwrap())
+                ]
+            ),]
+        );
+    }
+}
diff --git a/aquadoggo/src/replication/strategies/mod.rs b/aquadoggo/src/replication/strategies/mod.rs
new file mode 100644
index 000000000..c509580fc
--- /dev/null
+++ b/aquadoggo/src/replication/strategies/mod.rs
@@ -0,0 +1,24 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+mod diff;
+mod naive;
+mod set_reconciliation;
+
+pub use diff::diff_log_heights;
+pub use naive::NaiveStrategy;
+pub use set_reconciliation::SetReconciliationStrategy;
+
+use crate::replication::Message;
+
+#[derive(Clone, Debug)]
+pub struct StrategyResult {
+    pub messages: Vec<Message>,
+    pub is_local_done: bool,
+}
+
+impl StrategyResult {
+    pub fn merge(&mut self, result: StrategyResult) {
+        self.messages.extend(result.messages);
+        self.is_local_done = self.is_local_done || result.is_local_done;
+    }
+}
diff --git a/aquadoggo/src/replication/strategies/naive.rs b/aquadoggo/src/replication/strategies/naive.rs
new file mode 100644
index 000000000..fd0e04d4f
--- /dev/null
+++ b/aquadoggo/src/replication/strategies/naive.rs
@@ -0,0 +1,130 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use anyhow::Result;
+use async_trait::async_trait;
+
+use crate::db::SqlStore;
+use crate::replication::errors::ReplicationError;
+use crate::replication::strategies::diff_log_heights;
+use crate::replication::traits::Strategy;
+use crate::replication::{LogHeight, Message, Mode, StrategyResult, TargetSet};
+
+#[derive(Clone, Debug)]
+pub struct NaiveStrategy {
+    target_set: TargetSet,
+    received_remote_have: bool,
+    sent_have: bool,
+}
+
+impl NaiveStrategy {
+    pub fn new(target_set: &TargetSet) -> Self {
+        Self {
+            target_set: target_set.clone(),
+            received_remote_have: false,
+            sent_have: false,
+        }
+    }
+
+    async fn local_log_heights(&self, store: &SqlStore) -> Vec<LogHeight> {
+        let mut result = vec![];
+
+        // For every schema id in the target set retrieve log heights for all contributing authors
+        for schema_id in self.target_set().0.iter() {
+            let log_heights = store
+                .get_log_heights(schema_id)
+                .await
+                .expect("Fatal database error");
+            result.extend(log_heights);
+        }
+
+        result
+    }
+
+    async fn entry_responses(
+        &self,
+        store: &SqlStore,
+        remote_log_heights: &[LogHeight],
+    ) -> Vec<Message> {
+        let mut messages = Vec::new();
+
+        let local_log_heights = self.local_log_heights(store).await;
+        let remote_needs = diff_log_heights(&local_log_heights, remote_log_heights);
+
+        for (public_key, log_heights) in remote_needs {
+            for (log_id, seq_num) in log_heights {
+                let entry_messages: Vec<Message> = store
+                    .get_entries_from(&public_key, &log_id, &seq_num)
+                    .await
+                    .expect("Fatal database error")
+                    .iter()
+                    .map(|entry| {
+                        Message::Entry(entry.clone().encoded_entry, entry.payload().cloned())
+                    })
+                    .collect();
+                messages.extend(entry_messages);
+            }
+        }
+
+        messages
+    }
+}
+
+#[async_trait]
+impl Strategy for NaiveStrategy {
+    fn mode(&self) -> Mode {
+        Mode::Naive
+    }
+
+    fn target_set(&self) -> TargetSet {
+        self.target_set.clone()
+    }
+
+    async fn initial_messages(&mut self, store: &SqlStore) -> StrategyResult {
+        let log_heights = self.local_log_heights(store).await;
+        self.sent_have = true;
+
+        StrategyResult {
+            is_local_done: log_heights.is_empty(),
+            messages: vec![Message::Have(log_heights)],
+        }
+    }
+
+    async fn handle_message(
+        &mut self,
+        store: &SqlStore,
+        message: &Message,
+    ) -> Result<StrategyResult, ReplicationError> {
+        let mut result = StrategyResult {
+            is_local_done: false,
+            messages: vec![],
+        };
+
+        // Send our Have message to remote if we haven't done it yet
+        if !self.sent_have {
+            result.merge(self.initial_messages(store).await);
+        }
+
+        match message {
+            Message::Have(remote_log_heights) => {
+                if self.received_remote_have {
+                    return Err(ReplicationError::StrategyFailed(
+                        "Received Have from remote message twice".into(),
+                    ));
+                }
+
+                let response = self.entry_responses(store, remote_log_heights).await;
+                result.messages.extend(response);
+                result.is_local_done = true;
+
+                self.received_remote_have = true;
+            }
+            _ => {
+                return Err(ReplicationError::StrategyFailed(
+                    "Received unknown message type".into(),
+                ));
+            }
+        }
+
+        Ok(result)
+    }
+}
diff --git a/aquadoggo/src/replication/strategies/set_reconciliation.rs b/aquadoggo/src/replication/strategies/set_reconciliation.rs
new file mode 100644
index 000000000..12377edf6
--- /dev/null
+++ b/aquadoggo/src/replication/strategies/set_reconciliation.rs
@@ -0,0 +1,41 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+use anyhow::Result;
+use async_trait::async_trait;
+
+use crate::db::SqlStore;
+use crate::replication::errors::ReplicationError;
+use crate::replication::traits::Strategy;
+use crate::replication::{Message, Mode, StrategyResult, TargetSet};
+
+#[derive(Clone, Debug)]
+pub struct SetReconciliationStrategy;
+
+impl SetReconciliationStrategy {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+#[async_trait]
+impl Strategy for SetReconciliationStrategy {
+    fn mode(&self) -> Mode {
+        Mode::SetReconciliation
+    }
+
+    fn target_set(&self) -> TargetSet {
+        todo!()
+    }
+
+    async fn initial_messages(&mut self, _store: &SqlStore) -> StrategyResult {
+        todo!()
+    }
+
+    async fn handle_message(
+        &mut self,
+        _store: &SqlStore,
+        _message: &Message,
+    ) -> Result<StrategyResult, ReplicationError> {
+        todo!()
+    }
+}
diff --git a/aquadoggo/src/replication/target_set.rs b/aquadoggo/src/replication/target_set.rs
index 03ac3e64e..86cb76fe8 100644
--- a/aquadoggo/src/replication/target_set.rs
+++ b/aquadoggo/src/replication/target_set.rs
@@ -9,7 +9,7 @@ use crate::replication::errors::TargetSetError;
 /// De-duplicated and sorted set of schema ids which define the target data for the replication
 /// session.
 #[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize)]
-pub struct TargetSet(Vec<SchemaId>);
+pub struct TargetSet(pub Vec<SchemaId>);
 
 impl TargetSet {
     pub fn new(schema_ids: &[SchemaId]) -> Self {
@@ -27,6 +27,10 @@ impl TargetSet {
         Self(deduplicated_set)
     }
 
+    pub fn contains(&self, schema_id: &SchemaId) -> bool {
+        self.0.contains(schema_id)
+    }
+
     fn from_untrusted(schema_ids: Vec<SchemaId>) -> Result<Self, TargetSetError> {
         // Create target set with potentially invalid data
         let target_set = Self(schema_ids);
diff --git a/aquadoggo/src/replication/traits.rs b/aquadoggo/src/replication/traits.rs
index 323bb2931..9bbea4c90 100644
--- a/aquadoggo/src/replication/traits.rs
+++ b/aquadoggo/src/replication/traits.rs
@@ -1,52 +1,31 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-use anyhow::Result;
 use async_trait::async_trait;
-use p2panda_rs::schema::SchemaId;
 
-use crate::replication::{Mode, StrategyMessage, StrategyResult, TargetSet};
+use crate::db::SqlStore;
+use crate::replication::errors::ReplicationError;
+use crate::replication::{Message, Mode, StrategyResult, TargetSet};
 
 #[async_trait]
-pub trait Strategy: std::fmt::Debug + StrategyClone {
+pub trait Strategy: std::fmt::Debug + StrategyClone + Sync + Send {
     /// Replication mode of this strategy.
     fn mode(&self) -> Mode;
 
     /// Target set replication is occurring over.
     fn target_set(&self) -> TargetSet;
 
-    // Generate initial messages.
-    //
-    // @TODO: we want to pass the store in here too eventually.
-    async fn initial_messages(&self) -> Vec<StrategyMessage>;
+    /// Generate initial messages.
+    async fn initial_messages(&mut self, store: &SqlStore) -> StrategyResult;
 
-    // Handle incoming message and return response.
-    //
-    // @TODO: we want to pass the store in here too eventually.
-    async fn handle_message(&self, message: StrategyMessage) -> Result<StrategyResult>;
-
-    // Validate and store entry and operation.
-    //
-    // @TODO: we want to pass the store in here too eventually.
-    async fn handle_entry(
-        &self,
-        _schema_id: &SchemaId,
-        _entry_bytes: Vec<u8>,
-        _operation_bytes: Vec<u8>,
-    ) -> Result<()> {
-        // Validation:
-        // Check against schema_id and target_set if entry is what we've asked for
-        let _target_set = self.target_set();
-
-        // Further validation through our publish api stuff (?!)
-
-        // Have an entry waiting lobby service here, to batch stuff?!
-        // Nice to check certificate pool in one go.
-        // Nice to not put too much pressure on the database.
-        Ok(())
-    }
+    /// Handle incoming message and return response.
+    async fn handle_message(
+        &mut self,
+        store: &SqlStore,
+        message: &Message,
+    ) -> Result<StrategyResult, ReplicationError>;
 }
 
-/// This is a little trick so we can clone trait objects.
+// This is a little trick so we can clone trait objects.
 pub trait StrategyClone {
     fn clone_box(&self) -> Box<dyn Strategy>;
 }


Test patch:
diff --git a/aquadoggo/src/test_utils/db.rs b/aquadoggo/src/test_utils/db.rs
index 20a54022d..8f474a39a 100644
--- a/aquadoggo/src/test_utils/db.rs
+++ b/aquadoggo/src/test_utils/db.rs
@@ -6,10 +6,7 @@ use sqlx::Any;
 use crate::db::{connection_pool, create_database, run_pending_migrations, Pool};
 use crate::test_utils::TestConfiguration;
 
-/// Create test database.
-pub async fn initialize_db() -> (TestConfiguration, Pool) {
-    let config = TestConfiguration::new();
-
+async fn db_from_config(config: TestConfiguration) -> (TestConfiguration, Pool) {
     drop_database(&config).await;
     create_database(&config.database_url).await.unwrap();
 
@@ -23,6 +20,17 @@ pub async fn initialize_db() -> (TestConfiguration, Pool) {
     (config, pool)
 }
 
+/// Create test database.
+pub async fn initialize_db() -> (TestConfiguration, Pool) {
+    let config = TestConfiguration::new();
+    db_from_config(config).await
+}
+
+pub async fn initialize_sqlite_db() -> (TestConfiguration, Pool) {
+    let config = TestConfiguration::default();
+    db_from_config(config).await
+}
+
 /// Delete test database.
 pub async fn drop_database(config: &TestConfiguration) {
     if Any::database_exists(&config.database_url).await.unwrap() {
diff --git a/aquadoggo/src/test_utils/mod.rs b/aquadoggo/src/test_utils/mod.rs
index dcda6881e..19d444074 100644
--- a/aquadoggo/src/test_utils/mod.rs
+++ b/aquadoggo/src/test_utils/mod.rs
@@ -9,10 +9,10 @@ mod runner;
 
 pub use client::{graphql_test_client, TestClient};
 pub use config::TestConfiguration;
-pub use db::{drop_database, initialize_db};
+pub use db::{drop_database, initialize_db, initialize_sqlite_db};
 pub use helpers::{build_document, doggo_fields, doggo_schema, schema_from_fields};
 pub use node::{
     add_document, add_schema, add_schema_and_documents, populate_and_materialize,
     populate_store_config, TestNode,
 };
-pub use runner::test_runner;
+pub use runner::{test_runner, test_runner_with_manager, TestNodeManager};
diff --git a/aquadoggo/src/test_utils/runner.rs b/aquadoggo/src/test_utils/runner.rs
index 8d31d0f36..fccf72bdb 100644
--- a/aquadoggo/src/test_utils/runner.rs
+++ b/aquadoggo/src/test_utils/runner.rs
@@ -1,15 +1,18 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 use std::panic;
+use std::sync::Arc;
 
 use futures::Future;
 use tokio::runtime::Builder;
+use tokio::sync::Mutex;
 
 use crate::context::Context;
+use crate::db::Pool;
 use crate::db::SqlStore;
 use crate::schema::SchemaProvider;
-use crate::test_utils::initialize_db;
 use crate::test_utils::TestNode;
+use crate::test_utils::{initialize_db, initialize_sqlite_db};
 use crate::Configuration;
 
 #[async_trait::async_trait]
@@ -28,6 +31,53 @@ where
     }
 }
 
+#[async_trait::async_trait]
+pub trait AsyncTestFnWithManager {
+    async fn call(self, manager: TestNodeManager);
+}
+
+#[async_trait::async_trait]
+impl<FN, F> AsyncTestFnWithManager for FN
+where
+    FN: FnOnce(TestNodeManager) -> F + Sync + Send,
+    F: Future<Output = ()> + Send,
+{
+    async fn call(self, manager: TestNodeManager) {
+        self(manager).await
+    }
+}
+
+/// A manager which can create many databases and retain a handle on their connection pools.
+#[derive(Default)]
+pub struct TestNodeManager {
+    pools: Arc<Mutex<Vec<Pool>>>,
+}
+
+impl TestNodeManager {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub async fn create(&self) -> TestNode {
+        let (_config, pool) = initialize_sqlite_db().await;
+
+        // Initialise test store using pool.
+        let store = SqlStore::new(pool.clone());
+
+        // Construct the actual test node
+        let test_node = TestNode {
+            context: Context::new(
+                store.clone(),
+                Configuration::default(),
+                SchemaProvider::default(),
+            ),
+        };
+
+        self.pools.lock().await.push(pool);
+        test_node
+    }
+}
+
 /// Provides a safe way to write tests using a database which closes the pool connection
 /// automatically when the test succeeds or fails.
 ///
@@ -80,3 +130,48 @@ pub fn test_runner<F: AsyncTestFn + Send + Sync + 'static>(test: F) {
         };
     });
 }
+
+/// Method which provides a safe way to write tests with the ability to build many databases and
+/// have their pool connections closed automatically when the test succeeds or fails.
+///
+/// Takes an (async) test function as an argument and passes over the `TestNodeManager`
+/// instance which can be used to build nodes with databases from inside the tests.
+pub fn test_runner_with_manager<F: AsyncTestFnWithManager + Send + Sync + 'static>(test: F) {
+    let runtime = Builder::new_current_thread()
+        .worker_threads(1)
+        .enable_all()
+        .thread_name("with_db_teardown")
+        .build()
+        .expect("Could not build tokio Runtime for test");
+
+    // Instantiate the database manager
+    let manager = TestNodeManager::new();
+
+    // Get a handle onto it's collection of pools
+    let pools = manager.pools.clone();
+
+    runtime.block_on(async {
+        // Spawn the test in a separate task to make sure we have control over the possible
+        // panics which might happen inside of it
+        let handle = tokio::task::spawn(async move {
+            // Execute the actual test
+            test.call(manager).await;
+        });
+
+        // Get a handle of the task so we can use it later
+        let result = handle.await;
+
+        // Unwind the test by closing down the connections to all the database pools. This
+        // will be reached even when the test panicked
+        for pool in pools.lock().await.iter() {
+            pool.close().await;
+        }
+
+        // Panic here when test failed. The test fails within its own async task and stays
+        // there, we need to propagate it further to inform the test runtime about the result
+        match result {
+            Ok(_) => (),
+            Err(err) => panic::resume_unwind(err.into_panic()),
+        };
+    });
+}