samod 0.9.0

#![cfg_attr(docsrs, feature(doc_cfg))]
//! # Samod
//!
//! `samod` is a library for building collaborative applications which work offlne
//! and don't require servers (though servers certainly can be useful). This is
//! achieved by representing data as [`automerge`](https://docs.rs/automerge/latest/automerge/)
//! documents. `samod` is wire compatible with the `automerge-repo` JavaScript library.
//!
//! ## What does all that mean?
//!
//! `samod` helps you manage automerge "documents", which are hierarchical data
//! structures composed of maps, lists, text, and primitive values - a little
//! like JSON. Every change you make to a document is recorded and you can move
//! back and forth through the history of a document - it's a bit like Git for
//! JSON. `samod` takes care of storing changes for you and synchronizing them
//! with connected peers. The interesting part is that given this very detailed
//! history which we never discard, we can merge documents with changes which
//! were made concurrently. This means that we can build applications which
//! allow multiple users to edit the same document without having to have all
//! changes go through a server.
//!
//! ## How it all works
//!
//! The library is structured around a [`Repo`], which talks to a [`Storage`]
//! instance and to which you can connect to other peers using
//! [`Repo::connect`](crate::Repo::connect). Once you have a [`Repo`] you can
//! create documents using [`Repo::create`], or look up existing docuements using
//! [`Repo::find`]. In either case you will get back a [`DocHandle`] which you can
//! use to interact with the document.
//!
//! Typically then, your workflow will look like this:
//!
//! * Initialize a `Repo` at application startup, passing it a [`RuntimeHandle`]
//!   implementation and [`Storage`] implementation
//! * Whenever you have connections available (maybe you are connecting to a
//!   sync server, maybe you are receiving peer-to-peer connections) you call
//!   [`Repo::connect`] to add connections to the repo.
//! * Create `DocHandle`s using `Repo::create` and look up existing documents
//!   using `Repo::find`
//! * Modify documents using `DocHandle::with_document`
//!
//! Let's walk through each of those steps.
//!
//! ### Initializing a [`Repo`]
//!
//! To initialize a [`Repo`] you call [`Repo::builder()`] to obtain a
//! [`RepoBuilder`] which you use to configure the repo before calling [`RepoBuilder::load()`]
//! to actually load the repository. For example:
//!
//! ```rust
//! # #[cfg(feature="tokio")]
//! # tokio_test::block_on(async {
//! let repo = samod::Repo::builder(tokio::runtime::Handle::current())
//!     .with_storage(samod::storage::InMemoryStorage::new())
//!     .load()
//!     .await;
//! })
//! ```
//!
//! The first argument to `builder` is an implementation of [`RuntimeHandle`].
//! Default implementations are provided for `tokio` and `gio` which can be
//! conveniently used via [`Repo::build_tokio`] and [`Repo::build_gio`]
//! respectively. The [`RuntimeHandle`] trait is straightforward to implement if
//! you want to use some other async runtime.
//!
//! By default `samod` uses an in-memory storage implementation. This is great
//! for prototyping but in most cases you do actually want to persist data somewhere.
//! In this case you'll need an implementation of [`Storage`] to pass to
//! [`RepoBuilder::with_storage`]
//!
//! It is possible to use [`Storage`] and [`AnnouncePolicy`] implementations which
//! do not produce `Send` futures. In this case you will also need a runtime which
//! can spawn non-`Send` futures. See the [runtimes](#runtimes) section for more
//! details.
//!
//!
//! ### Connecting to peers
//!
//! Connections are managed via **dialers** and **acceptors** (listeners).
//!
//! - A **dialer** actively connects to a remote endpoint and automatically
//!   reconnects with exponential backoff. Create one with [`Repo::dial()`],
//!   which takes a [`BackoffConfig`] and an `Arc<dyn Dialer>`
//! - `Repo::find` will wait for any dialers which are in the process of
//!   connection to either establish a connection, or start retrying,
//!   before marking a document as unavailable. This means that as long
//!   as you add the [`Dialer`] before calling `Repo::find` you won't have to
//!   coordinate the timing of your dialers and document lookups.
//! - An **acceptor** passively accepts inbound connections. Create one with
//!   [`Repo::make_acceptor()`], then feed accepted transports via
//!   [`AcceptorHandle::accept()`].
//!
//! ```rust,no_run
//! // In this case we use the built in ChannelDialer which is useful for simple tests
//! // more useful implementations wrap actual network transports
//! # use samod::{BackoffConfig, Repo};
//! # use samod::transport::channel::ChannelDialer;
//! # use std::sync::Arc;
//! # async fn doit() {
//! let bob: Repo = todo!();
//! let alice: Repo = todo!();
//! let url = url::Url::parse("samod-channel://my-channel").unwrap();
//! let acceptor = bob.make_acceptor(url).unwrap();
//! let channel_dialer = ChannelDialer::new(acceptor);
//! let dialer_handle = alice.dial(BackoffConfig::default(), Arc::new(channel_dialer)).unwrap();
//! // Wait for the first successful connection + handshake
//! let peer_info = dialer_handle.established().await.unwrap();
//! # }
//! ```
//!
//! #### Dialer handles
//!
//! The [`DialerHandle`] returned by [`Repo::dial()`] can be used to:
//! - Wait for the first successful connection with [`DialerHandle::established()`]
//! - Observe lifecycle events (connect, disconnect, retry, failure) with
//!   [`DialerHandle::events()`]
//! - Check if connected with [`DialerHandle::is_connected()`]
//! - Close the dialer with [`DialerHandle::close()`]
//!
//! #### Acceptor handles
//!
//! The [`AcceptorHandle`] returned by [`Repo::make_acceptor()`] can be used to:
//! - Accept inbound connections with [`AcceptorHandle::accept()`]
//! - Observe lifecycle events with [`AcceptorHandle::events()`]
//! - Check connection count with [`AcceptorHandle::connection_count()`]
//! - Close the acceptor with [`AcceptorHandle::close()`]
//!
//! ### Managing Documents
//!
//! Once you have a [`Repo`] you can use it to manage [`DocHandle`]s. A
//! [`DocHandle`] represents an [`automerge`] document which the [`Repo`]
//! is managing. "managing" here means a few things:
//!
//! * Any changes made to the document using [`DocHandle::with_document`]
//!   will be persisted to storage and synchronized with connected peers
//!   (subject to the [`AnnouncePolicy`]).
//! * Any changes received from connected peers will be applied to the
//!   document and made visible to the application. You can listen for
//!   these changes using [`DocHandle::changes`].
//!
//! To create a new document you use [`Repo::create`] which will return
//! once the document has been persisted to storage. To look up an existing
//! document you use [`Repo::find`]. This will first look in storage, then
//! if the document is not found in storage it will request the document
//! from all connected peers (again subject to the [`AnnouncePolicy`]). If
//! any peer has the document the future returned by [`Repo::find`] will
//! resolve once we have synchronized with at least one remote peer which
//! has the document.
//!
//! You can make changes to a document using [`DocHandle::with_document`].
//!
//! ### Announce Policies
//!
//! By default, `samod` will announce all the [`DocHandle`]s it is synchronizing
//! to all connected peers and will also send requests to any connected peers
//! when you call [`Repo::find`]. This is often not what you want. To customize
//! this logic you pass an implementation of [`AnnouncePolicy`] to
//! [`RepoBuilder::with_announce_policy`]. Note that `AnnouncePolicy` is implemented
//! for `Fn(&DocumentId) -> bool` so you can just pass a closure if you want.
//!
//! ```rust
//! # #[cfg(feature="tokio")]
//! # tokio_test::block_on(async{
//! let authorized_peer = samod::PeerId::from("alice");
//! let repo = samod::Repo::build_tokio().with_announce_policy(move |_doc_id, peer_id| {
//!    // Only announce documents to alice
//!    &peer_id == &authorized_peer
//! }).load().await;
//! # });
//! ```
//!
//! ## Runtimes
//!
//! [`RuntimeHandle`] is a trait which is intended to abstract over the various
//! runtimes available in the rust ecosystem. The most common runtime is `tokio`.
//! `tokio` is a work-stealing runtime which means that the futures spawned on it
//! must be [`Send`], so that they can be moved between threads. This means that
//! [`RuntimeHandle::spawn`] requires [`Send`] futures. This in turn means that
//! the futures returned by the [`Storage`] and [`AnnouncePolicy`] traits are
//! also [`Send`] so that they can be spawned onto the [`RuntimeHandle`].
//!
//! In many cases though, you may have a runtime which doesn't require [`Send`]
//! futures and you may have storage and announce policy implementations which
//! cannot produce [`Send`] futures. This would often be the case in single
//! threaded runtimes for example. In these cases you can instead implement
//! [`LocalRuntimeHandle`], which doesn't require [`Send`] futures and then
//! you implement [`LocalStorage`] and [`LocalAnnouncePolicy`] traits for
//! your storage and announce policy implementations. You configure all these
//! things via the [`RepoBuilder`] struct. Once you've configured the storage
//! and announce policy implementations to use local variants you can then
//! create a local [`Repo`] using [`RepoBuilder::load_local`].
//!
//! ## Concurrency
//!
//! Typically `samod` will be managing many documents. One for each [`DocHandle`]
//! you retrieve via [`Repo::create`] or [`Repo::find`] but also one for any
//! sync messages received about a particular document from remote peers (e.g.
//! a sync server would have no [`DocHandle`]s open but would still be running
//! many document processes). By default document tasks will be handled on the
//! async runtime provided to the [`RepoBuilder`] but this can be undesirable.
//! Document operations can be compute intensive and so responsiveness may
//! benefit from running them on a separate thread pool. This is the purpose
//! of the [`RepoBuilder::with_concurrency`] method, which allows you to
//! configure how document operations are processed. If you want to use the
//! threadpool approach you will need to enable the `threadpool` feature.
//!
//! ## Why not just Automerge?
//!
//! `automerge` is a low level library. It provides routines for manipulating
//! documents in memory and an abstract data sync protocol. It does not actually
//! hook this up to any kind of network or storage. Most of the work involved
//! in doing this plumbing is straightforward, but if every application does
//! it themselves, we don't end up with interoperable applications. In particular
//! we don't end up with fungible sync servers. One of the core goals of this
//! library is to allow application authors to be agnostic as to where the
//! user synchronises data by implementing a generic network and storage layer
//! which all applications can use.
//!
//! ## Example
//!
//! Here's a somewhat fully featured example of using `samod` to manage a todo list
//! across two repos (representing two different devices):
//!
//! ```rust
//! # #[cfg(feature="tokio")]
//! # tokio_test::block_on(async {
//! use automerge::{ReadDoc, transaction::{Transactable as _}};
//! use futures::StreamExt as _;
//! use samod::{BackoffConfig, transport::channel::ChannelDialer};
//! use std::sync::Arc;
//!
//! # let _ = tracing_subscriber::fmt().try_init();
//!
//! // Create two repos, representing two different devices
//! let alice = samod::Repo::build_tokio().load().await;
//! let bob = samod::Repo::build_tokio().load().await;
//!
//! // Create an initial skeleton for our todo list on alice
//! let mut initial_doc = automerge::Automerge::new();
//! initial_doc.transact::<_, _, automerge::AutomergeError>(|tx| {
//!     let _todos = tx.put_object(automerge::ROOT, "todos", automerge::ObjType::List)?;
//!     Ok(())
//! }).unwrap();
//!
//! // Now create a `samod::DocHandle` on alice using `Repo::create`
//! let alice_handle = alice.create(initial_doc).await.unwrap();
//!
//! // Bob registers an acceptor; alice dials bob via an in-process ChannelDialer
//! let url = url::Url::parse("channel://alice-to-bob").unwrap();
//! let acceptor = bob.make_acceptor(url).unwrap();
//! let channel_dialer = ChannelDialer::new(acceptor);
//! let dialer_handle = alice.dial(BackoffConfig::default(), Arc::new(channel_dialer)).unwrap();
//!
//! // Wait for alice to be connected to bob
//! dialer_handle.established().await.unwrap();
//!
//! // Bob can now fetch alice's document
//! let bob_handle = bob.find(alice_handle.document_id().clone()).await.unwrap().unwrap();
//!
//! // Make a change on bob's side
//! bob_handle.with_document(|doc| {
//!     doc.transact(|tx| {
//!        let todos = tx.get(automerge::ROOT, "todos").unwrap()
//!           .expect("todos should exist").1;
//!        tx.insert(todos, 0, "Buy milk")?;
//!        Ok::<_, automerge::AutomergeError>(())
//!     }).unwrap();
//! });
//!
//! // Wait for the change to be received on alice's side
//! alice_handle.changes().next().await.unwrap();
//!
//! // Alice's document now reflects bob's change
//! alice_handle.with_document(|doc| {
//!     let todos = doc.get(automerge::ROOT, "todos").unwrap()
//!         .expect("todos should exist").1;
//!     let item = doc.get(todos, 0).unwrap().expect("item should exist").0;
//!     let automerge::Value::Scalar(val) = item else {
//!         panic!("item should be a scalar");
//!     };
//!     let automerge::ScalarValue::Str(s) = val.as_ref() else {
//!         panic!("item should be a string");
//!     };
//!     assert_eq!(s, "Buy milk");
//!     Ok::<_, automerge::AutomergeError>(())
//! }).unwrap();
//! # });
//! ```
//!
use std::{
    collections::HashMap,
    sync::{Arc, Mutex},
};

use automerge::Automerge;
use futures::{FutureExt, Stream, StreamExt, channel::oneshot, stream::FuturesUnordered};
use rand::SeedableRng;
pub use samod_core::{
    AutomergeUrl, BackoffConfig, ConnectionId, DialerId, DocumentId, ListenerId, PeerId, StorageId,
    network::ConnDirection,
};
use samod_core::{
    CommandId, CommandResult, DocumentActorId, LoaderState, UnixTimestamp,
    actors::{
        DocToHubMsg,
        document::{DocumentActor, SpawnArgs},
        hub::{DispatchedCommand, Hub, HubEvent, HubResults, io::HubIoAction},
    },
    io::{IoResult, IoTask},
    network::{ConnectionEvent, ConnectionOwner, DialerConfig, ListenerConfig},
};
use tracing::Instrument;
pub use url::Url;

mod actor_task;
use actor_task::ActorTask;
mod actor_handle;
use actor_handle::ActorHandle;
mod announce_policy;
mod builder;
pub use builder::RepoBuilder;
mod acceptor_handle;
pub use acceptor_handle::{AcceptorEvent, AcceptorHandle};
mod conn_finished_reason;
mod connection;
mod dialer_handle;
pub use conn_finished_reason::ConnFinishedReason;
pub use connection::Connection;
pub use dialer_handle::{DialerEvent, DialerFailed, DialerHandle};
mod dialer;
pub use dialer::Dialer;
mod doc_actor_inner;
mod doc_handle;
mod doc_runner;
mod io_loop;
mod observer;
pub use doc_handle::DocHandle;
pub use observer::{RepoEvent, RepoObserver, StorageOperation};
mod peer_connection_info;
mod peer_info;
pub use peer_connection_info::{ConnectionInfo, ConnectionState, PeerDocState};
pub use peer_info::PeerInfo;
mod stopped;
pub use stopped::Stopped;
pub mod storage;
pub mod transport;
pub use crate::announce_policy::{
    AlwaysAnnounce, AnnouncePolicy, LocalAnnouncePolicy, NeverAnnounce,
};
pub use crate::builder::ConcurrencyConfig;
use crate::{
    connection::ConnectionHandle,
    doc_actor_inner::DocActorInner,
    doc_runner::{DocRunner, SpawnedActor},
    io_loop::{DriveConnectionTask, IoLoopTask},
    storage::Storage,
    unbounded::{UnboundedReceiver, UnboundedSender},
};
use crate::{
    runtime::{LocalRuntimeHandle, RuntimeHandle},
    storage::{InMemoryStorage, LocalStorage},
};
pub use transport::Transport;
pub mod runtime;
#[cfg(feature = "tokio")]
pub mod tokio_io;
mod unbounded;
pub mod websocket;

/// The entry point to this library
///
/// A [`Repo`] represents a set of running [`DocHandle`]s, active connections to
/// other peers over which we are synchronizing the active [`DocHandle`]s, and
/// an instance of [`Storage`] where document data is persisted.
///
/// Individual documents require exclusive access to mutate (including receiving
/// sync messages). In order to make this non-blocking the `Repo` spawns a
/// task for each document on an underlying threadpool. All method calls on
/// the `Repo` can consequently be called in asynchronous contexts as they do
/// not block.
///
/// ## Lifecycle
///
/// To obtain a [`Repo`] call [`Repo::builder`] (or the various `Repo::build_*`
/// variants specific to different runtimes) to obtain a [`RepoBuilder`] and
/// then call [`RepoBuilder::load`] to actually load the [`Repo`].
///
/// Once you have a repo you can connect new network connections to it using
/// [`Repo::connect`]. You can create new documents using [`Repo::create`] and
/// lookup existing documents using [`Repo::find`].
///
/// When you are finished with a [`Repo`] you can call [`Repo::stop`] to flush
/// everything to storage before shutting down the application.
#[derive(Clone)]
pub struct Repo {
    inner: Arc<Mutex<Inner>>,
}

impl std::fmt::Debug for Repo {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Repo").finish()
    }
}

impl Repo {
    // Create a new [`RepoBuilder`] which will build a [`Repo`] that spawns its
    // tasks onto the provided runtime
    pub fn builder<R: runtime::RuntimeHandle>(
        runtime: R,
    ) -> RepoBuilder<InMemoryStorage, R, AlwaysAnnounce> {
        builder::RepoBuilder::new(runtime)
    }

    /// Create a new [`RepoBuilder`] which will build a [`Repo`] that spawns it's
    /// tasks onto the current tokio runtime
    ///
    /// ## Panics
    /// If called outside of the dynamic scope of a tokio runtime
    #[cfg(feature = "tokio")]
    pub fn build_tokio() -> RepoBuilder<InMemoryStorage, ::tokio::runtime::Handle, AlwaysAnnounce> {
        builder::RepoBuilder::new(::tokio::runtime::Handle::current())
    }

    // Create a new [`RepoBuilder`] which will build a [`Repo`] that spawns it's
    // tasks onto a [`futures::executor::LocalPool`]
    pub fn build_localpool(
        spawner: futures::executor::LocalSpawner,
    ) -> RepoBuilder<InMemoryStorage, futures::executor::LocalSpawner, AlwaysAnnounce> {
        builder::RepoBuilder::new(spawner)
    }

    /// Create a new [`Repo`] instance which will build a [`Repo`] that spawns
    /// its tasks onto the current gio mainloop
    ///
    /// # Panics
    ///
    /// This function will panic if called outside of a gio mainloop context.
    #[cfg(feature = "gio")]
    pub fn build_gio()
    -> RepoBuilder<InMemoryStorage, crate::runtime::gio::GioRuntime, AlwaysAnnounce> {
        builder::RepoBuilder::new(crate::runtime::gio::GioRuntime::new())
    }

    pub(crate) async fn load<
        R: runtime::RuntimeHandle + Clone + Send,
        S: Storage,
        A: AnnouncePolicy,
    >(
        builder: RepoBuilder<S, R, A>,
    ) -> Self {
        let RepoBuilder {
            storage,
            runtime,
            peer_id,
            announce_policy,
            concurrency,
            observer,
        } = builder;
        let task_setup = TaskSetup::new(storage.clone(), peer_id, concurrency, observer).await;
        let inner = task_setup.inner.clone();
        task_setup.spawn_tasks(runtime, storage, announce_policy);
        Self { inner }
    }

    pub(crate) async fn load_local<
        'a,
        R: runtime::LocalRuntimeHandle + Clone + 'static,
        S: LocalStorage + 'a,
        A: LocalAnnouncePolicy + 'a,
    >(
        builder: RepoBuilder<S, R, A>,
    ) -> Self {
        let RepoBuilder {
            storage,
            runtime,
            peer_id,
            announce_policy,
            concurrency,
            observer,
        } = builder;
        let task_setup = TaskSetup::new(storage.clone(), peer_id, concurrency, observer).await;
        let inner = task_setup.inner.clone();
        task_setup.spawn_tasks_local(runtime, storage, announce_policy);
        Self { inner }
    }

    /// Create a new document and return a handle to it
    ///
    /// # Arguments
    /// * `initial_content` - The initial content of the document. If this is an
    ///   empty document a single empty commit will be created and added to the
    ///   document. This ensures that when a document is created, _something_ is
    ///   in storage
    ///
    /// The returned future will resolve once the document has been persisted to storage
    pub async fn create(&self, initial_content: Automerge) -> Result<DocHandle, Stopped> {
        let (tx, rx) = oneshot::channel();
        {
            let DispatchedCommand { command_id, event } =
                HubEvent::create_document(initial_content);
            let mut inner = self.inner.lock().unwrap();
            inner.handle_event(event);
            inner.pending_commands.insert(command_id, tx);
            drop(inner);
        }
        let inner = self.inner.clone();
        match rx.await {
            Ok(r) => match r {
                CommandResult::CreateDocument {
                    actor_id,
                    document_id: _,
                } => {
                    {
                        let inner = inner.lock().unwrap();
                        // By this point the document should have been spawned

                        Ok(inner
                            .actors
                            .get(&actor_id)
                            .map(|ActorHandle { doc: handle, .. }| handle.clone())
                            .expect("actor should exist"))
                    }
                }
                other => {
                    panic!("unexpected command result for create: {other:?}");
                }
            },
            Err(_) => Err(Stopped),
        }
    }

    /// Lookup a document by ID
    ///
    /// The [`Repo`] will first attempt to load the document from [`Storage`] and
    /// if it is not found the [`Repo`] will then request the document from all
    /// connected peers (subject to the configured [`AnnouncePolicy`]). If any peer
    /// responds with the document the future will resolve once we have
    /// synchronized with at least one remote peer which has the document. Otherwise,
    /// the future will resolve to `Ok(None)` once all peers have responded that
    /// they do not have the document.
    ///
    /// If there are existing [`Dialer`]s which have not yet established a connection,
    /// but are not in the process of retrying or marked as failed then the future
    /// will wait for them to either establish a connection or fail before resolving.
    pub fn find(
        &self,
        doc_id: DocumentId,
    ) -> impl Future<Output = Result<Option<DocHandle>, Stopped>> + 'static {
        let mut inner = self.inner.lock().unwrap();
        let DispatchedCommand { command_id, event } = HubEvent::find_document(doc_id);
        let (tx, rx) = oneshot::channel();
        inner.pending_commands.insert(command_id, tx);
        inner.handle_event(event);
        drop(inner);
        let inner = self.inner.clone();
        async move {
            match rx.await {
                Ok(r) => match r {
                    CommandResult::FindDocument { actor_id, found } => {
                        if found {
                            // By this point the document should have been spawned
                            let handle = inner
                                .lock()
                                .unwrap()
                                .actors
                                .get(&actor_id)
                                .map(|ActorHandle { doc: handle, .. }| handle.clone())
                                .expect("actor should exist");
                            Ok(Some(handle))
                        } else {
                            Ok(None)
                        }
                    }
                    other => {
                        panic!("unexpected command result for create: {other:?}");
                    }
                },
                Err(_) => Err(Stopped),
            }
        }
    }

    /// Wait for some connection to be established with the given remote peer ID
    ///
    /// This will resolve immediately if the peer is already connected, otherwise
    /// it will resolve when a connection with the given peer ID is established.
    pub fn when_connected(
        &self,
        peer_id: PeerId,
    ) -> impl Future<Output = Result<Connection, Stopped>> + 'static {
        let inner = self.inner.clone();
        async move {
            let rx = {
                let mut inner = inner.lock().unwrap();

                for conn_handle in inner.connections.values() {
                    if conn_handle.info().map(|i| i.peer_id).as_ref() == Some(&peer_id) {
                        return Ok(Connection::new(conn_handle.clone()));
                    }
                }
                let (tx, rx) = oneshot::channel();
                inner
                    .waiting_for_connection
                    .entry(peer_id)
                    .or_default()
                    .push(tx);
                rx
            };
            rx.await.map_err(|_| Stopped)
        }
    }

    /// The peer ID of this instance
    pub fn peer_id(&self) -> PeerId {
        self.inner.lock().unwrap().hub.peer_id().clone()
    }

    /// Get the currently connected peers and a stream of changes to the connected peers
    ///
    /// The returned stream will be the entire state of the connected peers,
    /// emitted whenever any part of that state changes (such as when new peers
    /// are added, or a new message is sent)
    pub fn connected_peers(
        &self,
    ) -> (
        Vec<ConnectionInfo>,
        impl Stream<Item = Vec<ConnectionInfo>> + Unpin + 'static,
    ) {
        let (tx_events, rx_events) = unbounded::channel();
        let mut inner = self.inner.lock().unwrap();
        inner.conn_listeners.push(tx_events);
        let now = inner
            .hub
            .connections()
            .clone()
            .into_iter()
            .map(|c| c.into())
            .collect();
        (now, rx_events)
    }

    /// Register a dialer with automatic reconnection.
    ///
    /// The [`Dialer`] knows both *where* to connect ([`Dialer::url`]) and
    /// *how* to connect ([`Dialer::connect`]). It is called each time a new
    /// transport is needed — on the initial dial and on each reconnection
    /// attempt after backoff.
    ///
    /// Reconnection timing is controlled by the `backoff` parameter. The
    /// dialer will automatically reconnect with exponential backoff and
    /// jitter when a connection is lost or a transport establishment fails.
    ///
    /// Returns a [`DialerHandle`] that can be used to observe lifecycle events
    /// (connected, disconnected, reconnecting, failed) and to wait for the
    /// first successful connection.
    ///
    /// # Arguments
    ///
    /// * `backoff` - Backoff configuration for reconnection attempts.
    /// * `dialer` - A [`Dialer`] that establishes transports.
    ///
    /// # Returns
    ///
    /// A [`DialerHandle`] for observing and controlling the dialer.
    pub fn dial(
        &self,
        backoff: BackoffConfig,
        dialer: Arc<dyn Dialer>,
    ) -> Result<DialerHandle, Stopped> {
        let mut inner = self.inner.lock().unwrap();

        let url = dialer.url();
        let config = DialerConfig { url, backoff };

        let DispatchedCommand { command_id, event } = HubEvent::add_dialer(config);
        let (tx_result, mut rx_result) = oneshot::channel();
        inner.pending_commands.insert(command_id, tx_result);
        inner.handle_event(event);

        let dialer_id = match rx_result.try_recv() {
            Ok(Some(CommandResult::AddDialer { dialer_id })) => dialer_id,
            Ok(None) => return Err(Stopped), // Hub is stopped, event was ignored
            Ok(Some(other)) => panic!("unexpected command result {:?} for add_dialer", other),
            Err(_) => return Err(Stopped),
        };

        // Register the dialer so the IO loop can use it for (re)connections
        inner.dialers.lock().unwrap().insert(dialer_id, dialer);

        // Create and register the dialer handle
        let handle = DialerHandle::new(dialer_id, self.clone());
        inner.dialer_handles.insert(dialer_id, handle.clone());

        Ok(handle)
    }

    /// Register an acceptor for inbound connections on the given URL.
    ///
    /// This is typically called once at startup for each endpoint the server
    /// listens on (e.g. a WebSocket handler on port 8080). The returned
    /// [`AcceptorHandle`] is then used to feed individual accepted connections
    /// via [`AcceptorHandle::accept()`].
    ///
    /// The URL is used for logging and identifying the endpoint in
    /// debugging output.
    pub fn make_acceptor(&self, url: url::Url) -> Result<AcceptorHandle, Stopped> {
        let mut inner = self.inner.lock().unwrap();

        // Check if a listener already exists for this URL
        if let Some(listener_id) = inner.find_listener_for_url(&url)
            && let Some(handle) = inner.acceptor_handles.get(&listener_id)
        {
            return Ok(handle.clone());
        }

        let config = ListenerConfig { url };

        let DispatchedCommand { command_id, event } = HubEvent::add_listener(config);
        let (tx_result, mut rx_result) = oneshot::channel();
        inner.pending_commands.insert(command_id, tx_result);
        inner.handle_event(event);

        let listener_id = match rx_result.try_recv() {
            Ok(Some(CommandResult::AddListener { listener_id })) => listener_id,
            Ok(None) => return Err(Stopped),
            Ok(Some(other)) => panic!("unexpected command result {:?} for add_listener", other),
            Err(_) => return Err(Stopped),
        };

        // Create and register the acceptor handle
        let handle = AcceptorHandle::new(listener_id, self.clone());
        inner.acceptor_handles.insert(listener_id, handle.clone());

        Ok(handle)
    }

    /// Remove a dialer by ID. Used by `DialerHandle::close()`.
    pub(crate) fn remove_dialer_by_id(&self, dialer_id: DialerId) -> Result<(), Stopped> {
        let mut inner = self.inner.lock().unwrap();

        // Remove the dialer
        inner.dialers.lock().unwrap().remove(&dialer_id);

        // Remove handle
        inner.dialer_handles.remove(&dialer_id);

        let event = HubEvent::remove_dialer(dialer_id);
        inner.handle_event(event);
        Ok(())
    }

    /// Remove a listener by ID. Used by `AcceptorHandle::close()`.
    pub(crate) fn remove_listener_by_id(&self, listener_id: ListenerId) -> Result<(), Stopped> {
        let mut inner = self.inner.lock().unwrap();

        // Remove handle
        inner.acceptor_handles.remove(&listener_id);

        let event = HubEvent::remove_listener(listener_id);
        inner.handle_event(event);
        Ok(())
    }

    /// Internal: accept a connection on an existing listener. Used by
    /// `AcceptorHandle::accept()`.
    pub(crate) fn accept_on_listener(
        &self,
        listener_id: ListenerId,
        transport: crate::Transport,
    ) -> Result<ConnectionHandle, Stopped> {
        let mut inner = self.inner.lock().unwrap();

        // Create the connection atomically associated with the listener
        let DispatchedCommand { command_id, event } =
            HubEvent::create_listener_connection(listener_id);
        let (tx_result, mut rx_result) = oneshot::channel();
        inner.pending_commands.insert(command_id, tx_result);
        inner.handle_event(event);

        let connection_id = match rx_result.try_recv() {
            Ok(Some(CommandResult::CreateConnection { connection_id })) => connection_id,
            Ok(other) => panic!(
                "unexpected command result {:?} for create_listener_connection",
                other
            ),
            Err(_) => return Err(Stopped),
        };

        if let Some(obs) = inner.observer.as_ref() {
            obs.observe(&RepoEvent::ConnectionEstablished { connection_id })
        }

        let conn_handle = inner
            .connections
            .get(&connection_id)
            .expect("connection not found")
            .clone();

        let loop_task = IoLoopTask::DriveConnection(DriveConnectionTask {
            conn_handle: conn_handle.clone(),
            stream: transport.stream,
            sink: transport.sink,
        });
        inner.tx_io.unbounded_send(loop_task).map_err(|_| Stopped)?;

        Ok(conn_handle)
    }

    /// Stop the `Samod` instance.
    ///
    /// This will wait until all storage tasks have completed before stopping all
    /// the documents and returning
    pub fn stop(&self) -> impl Future<Output = ()> + 'static {
        let (tx, rx) = oneshot::channel();
        {
            let mut inner = self.inner.lock().unwrap();
            inner.stop_waiters.push(tx);
            inner.handle_event(HubEvent::stop());
        }
        async move {
            if rx.await.is_err() {
                tracing::warn!("stop signal was dropped");
            }
        }
    }
}

struct Inner {
    doc_runner: DocRunner,
    actors: HashMap<DocumentActorId, ActorHandle>,
    hub: Hub,
    pending_commands: HashMap<CommandId, oneshot::Sender<CommandResult>>,
    connections: HashMap<ConnectionId, ConnectionHandle>,
    conn_listeners: Vec<unbounded::UnboundedSender<Vec<ConnectionInfo>>>,
    tx_io: UnboundedSender<io_loop::IoLoopTask>,
    tx_to_core: UnboundedSender<(DocumentActorId, DocToHubMsg)>,
    waiting_for_connection: HashMap<PeerId, Vec<oneshot::Sender<Connection>>>,
    stop_waiters: Vec<oneshot::Sender<()>>,
    rng: rand::rngs::StdRng,
    dialers: Arc<Mutex<HashMap<DialerId, io_loop::DynDialer>>>,
    dialer_handles: HashMap<DialerId, DialerHandle>,
    acceptor_handles: HashMap<ListenerId, AcceptorHandle>,
    observer: Option<Arc<dyn observer::RepoObserver>>,
}

impl Inner {
    /// Find a listener for the given URL.
    fn find_listener_for_url(&self, url: &url::Url) -> Option<ListenerId> {
        self.hub.find_listener_for_url(url)
    }

    /// Dispatch a task to a document actor.
    ///
    /// In async mode, this sends the task via the actor's channel stored in DocRunner.
    /// In threadpool mode, this spawns the task directly on the threadpool.
    fn dispatch_task(&self, actor_id: DocumentActorId, task: ActorTask) {
        match &self.doc_runner {
            #[cfg(feature = "threadpool")]
            DocRunner::Threadpool(threadpool) => {
                let Some(actor_handle) = self.actors.get(&actor_id) else {
                    tracing::warn!(?actor_id, "received task for unknown actor");
                    return;
                };
                let inner = actor_handle.inner.clone();
                threadpool.spawn(move || {
                    let mut guard = inner.lock().unwrap();
                    guard.handle_task(task);
                });
            }
            DocRunner::Async { task_senders, .. } => {
                let Some(tx) = task_senders.get(&actor_id) else {
                    tracing::warn!(?actor_id, "received task for unknown actor");
                    return;
                };
                let _ = tx.unbounded_send(task);
            }
        }
    }

    #[tracing::instrument(skip(self, event), fields(local_peer_id=%self.hub.peer_id()))]
    fn handle_event(&mut self, event: HubEvent) {
        // The hub itself will gracefully ignore events after it has stopped,
        // but we short-circuit here to avoid unnecessary work.
        if self.hub.is_stopped() {
            tracing::trace!("ignoring event on stopped hub");
            return;
        }
        let hub_start = std::time::Instant::now();
        let now = UnixTimestamp::now();
        let HubResults {
            new_tasks,
            completed_commands,
            spawn_actors,
            actor_messages,
            stopped,
            connection_events,
            dial_requests,
            dialer_events,
            event_type,
            connections_count,
            documents_count,
        } = self.hub.handle_event(&mut self.rng, now, event);

        for spawn_args in spawn_actors {
            self.spawn_actor(spawn_args);
        }

        for (command_id, command) in completed_commands {
            if let CommandResult::Receive { .. } = &command {
                // We don't track receive commands
                continue;
            }
            if let Some(tx) = self.pending_commands.remove(&command_id) {
                if let CommandResult::CreateConnection { connection_id } = &command {
                    self.connections
                        .insert(*connection_id, ConnectionHandle::new(*connection_id));
                }
                let _ = tx.send(command);
            } else {
                tracing::warn!("Received result for unknown command: {:?}", command_id);
            }
        }

        for task in new_tasks {
            match task.action {
                HubIoAction::Send { connection_id, msg } => {
                    if let Some(connhandle) = self.connections.get(&connection_id) {
                        connhandle.send(msg);
                    } else {
                        tracing::warn!(
                            "Tried to send message on unknown connection: {:?}",
                            connection_id
                        );
                    }
                }
                HubIoAction::Disconnect { connection_id } => {
                    if self.connections.remove(&connection_id).is_none() {
                        tracing::warn!(
                            "Tried to disconnect unknown connection: {:?}",
                            connection_id
                        );
                    }
                }
            }
        }

        for (actor_id, actor_msg) in actor_messages {
            self.dispatch_task(actor_id, ActorTask::HandleMessage(actor_msg));
        }

        if !connection_events.is_empty() && !self.conn_listeners.is_empty() {
            let new_infos = self.hub.connections();
            self.conn_listeners.retain(|tx| {
                tx.unbounded_send(new_infos.clone().into_iter().map(|c| c.into()).collect())
                    .is_ok()
            });
        }

        for evt in connection_events {
            match evt {
                ConnectionEvent::HandshakeCompleted {
                    connection_id,
                    owner,
                    peer_info,
                } => {
                    if let Some(conn_handle) = self.connections.get(&connection_id) {
                        let samod_peer_info: PeerInfo = peer_info.clone().into();
                        conn_handle.notify_handshake_complete(samod_peer_info.clone());
                        if let Some(tx) = self.waiting_for_connection.get_mut(&peer_info.peer_id) {
                            for tx in tx.drain(..) {
                                let _ = tx.send(Connection::new(conn_handle.clone()));
                            }
                        }

                        // Route to DialerHandle or AcceptorHandle based on owner
                        match owner {
                            ConnectionOwner::Dialer(dialer_id) => {
                                if let Some(dh) = self.dialer_handles.get(&dialer_id) {
                                    dh.notify_connected(samod_peer_info, connection_id);
                                }
                            }
                            ConnectionOwner::Listener(listener_id) => {
                                if let Some(ah) = self.acceptor_handles.get(&listener_id) {
                                    ah.notify_client_connected(
                                        samod_peer_info.clone(),
                                        connection_id,
                                    );
                                }
                                conn_handle.notify_client_connected(samod_peer_info);
                            }
                        }
                    }
                }
                ConnectionEvent::ConnectionFailed {
                    connection_id,
                    owner,
                    error,
                } => {
                    tracing::error!(
                        ?connection_id,
                        ?error,
                        "connection failed, notifying waiting tasks",
                    );

                    if let Some(ref obs) = self.observer {
                        obs.observe(&observer::RepoEvent::ConnectionLost { connection_id });
                    }

                    // Notify dialer/acceptor handle of disconnection
                    match owner {
                        ConnectionOwner::Dialer(dialer_id) => {
                            if let Some(dh) = self.dialer_handles.get(&dialer_id) {
                                dh.notify_disconnected(error.clone());
                            }
                        }
                        ConnectionOwner::Listener(listener_id) => {
                            if let Some(ah) = self.acceptor_handles.get(&listener_id) {
                                ah.notify_client_disconnected(
                                    connection_id,
                                    ConnFinishedReason::ErrorReceiving(error.clone()),
                                );
                                if let Some(conn_handle) = self.connections.get(&connection_id) {
                                    conn_handle.notify_client_disconnected(
                                        ConnFinishedReason::ErrorReceiving(error.clone()),
                                    );
                                };
                            }
                        }
                    }

                    // This will drop the sender which will in turn cause the stream handling
                    // code in io_loop::drive_connection to emit disconnection events
                    self.connections.remove(&connection_id);
                }
                ConnectionEvent::StateChanged { .. } => {
                    // StateChanged events are already handled above for conn_listeners.
                    // Individual connection state changes don't directly map to
                    // DialerEvent/AcceptorEvent — those are driven by
                    // HandshakeCompleted and ConnectionFailed events.
                }
            }
        }

        // Process dial requests from dialers — dispatch to IO loop
        for request in dial_requests {
            tracing::debug!(
                dialer_id = ?request.dialer_id,
                url = %request.url,
                "dispatching dial request to IO loop"
            );

            // Notify dialer handle of reconnection attempt (if attempt > 0)
            if let Some(dh) = self.dialer_handles.get(&request.dialer_id) {
                // We check if this is a retry by looking at the hub's dialer state.
                // The first dial request (attempt 0) is the initial dial, not a reconnection.
                if let Some(attempt) = self.hub.dialer_attempt(request.dialer_id)
                    && attempt > 0
                {
                    dh.notify_reconnecting(attempt);
                }
            }

            let task = IoLoopTask::EstablishTransport {
                dialer_id: request.dialer_id,
                url: request.url,
            };
            if self.tx_io.unbounded_send(task).is_err() {
                tracing::error!("IO loop channel closed, cannot dispatch dial request");
            }
        }

        // Process dialer events (dialer lifecycle)
        for event in dialer_events {
            match &event {
                samod_core::DialerEvent::MaxRetriesReached { dialer_id, url } => {
                    tracing::warn!(
                        ?dialer_id,
                        %url,
                        "dialer exhausted retry budget"
                    );
                    if let Some(dh) = self.dialer_handles.get(dialer_id) {
                        dh.notify_max_retries_reached();
                    }
                }
            }
        }

        if let Some(ref obs) = self.observer {
            obs.observe(&observer::RepoEvent::HubEventProcessed {
                duration: hub_start.elapsed(),
                event_type,
                connections: connections_count,
                documents: documents_count,
            });
        }

        if stopped {
            for waiter in self.stop_waiters.drain(..) {
                let _ = waiter.send(());
            }
        }
    }

    #[tracing::instrument(skip(self, args))]
    fn spawn_actor(&mut self, args: SpawnArgs) {
        let actor_id = args.actor_id();
        let doc_id = args.document_id().clone();
        let (actor, init_results) = DocumentActor::new(UnixTimestamp::now(), args);

        if let Some(ref obs) = self.observer {
            obs.observe(&observer::RepoEvent::DocumentOpened {
                document_id: doc_id.clone(),
            });
        }

        let doc_inner = Arc::new(Mutex::new(DocActorInner::new(
            doc_id.clone(),
            actor_id,
            actor,
            self.tx_to_core.clone(),
            self.tx_io.clone(),
            self.observer.clone(),
        )));
        let handle = DocHandle::new(doc_id.clone(), doc_inner.clone());
        self.actors.insert(
            actor_id,
            ActorHandle {
                inner: doc_inner.clone(),
                doc: handle,
            },
        );

        match &mut self.doc_runner {
            #[cfg(feature = "threadpool")]
            DocRunner::Threadpool(_threadpool) => {
                // In threadpool mode, we process init_results synchronously and then
                // dispatch subsequent tasks via dispatch_task() which spawns them
                // directly on the threadpool. This avoids dedicating one thread per
                // document, which would limit the number of documents to the thread count.
                doc_inner.lock().unwrap().handle_results(init_results);
            }
            DocRunner::Async {
                tx_spawn,
                task_senders,
            } => {
                // Create channel for this actor and store the sender
                let (tx, rx) = unbounded::channel();
                task_senders.insert(actor_id, tx);

                if tx_spawn
                    .unbounded_send(SpawnedActor {
                        doc_id,
                        actor_id,
                        inner: doc_inner,
                        rx_tasks: rx,
                        init_results,
                    })
                    .is_err()
                {
                    tracing::error!(?actor_id, "actor spawner is gone");
                }
            }
        }
    }
}

/// Spawns a task which listens for new actors to spawn and runs them
///
/// `samod` has two ways of running document actors, on a rayon threadpool, or
/// on the async runtime which was provided to the `SamodBuilder`. In the latter
/// case we don't actually hold on to a reference to the `RuntimeHandle` because
/// that requires it to be `Send` which is not always the case (e.g. when using
/// futures::executor::LocalPool). Instead, we spawn a task on the runtime which
/// listens on a channel for new actors to spawn and then runs them on a
/// `FuturesUnordered`. This function is that task.
async fn async_actor_runner(rx: UnboundedReceiver<SpawnedActor>) {
    let mut running_actors = FuturesUnordered::new();

    loop {
        futures::select! {
            spawn_actor = rx.recv().fuse() => {
                match spawn_actor {
                    Err(_e) => {
                        tracing::trace!("actor spawner task finished");
                        break;
                    }
                    Ok(SpawnedActor { inner, rx_tasks, init_results, doc_id, actor_id }) => {
                        running_actors.push(async move {
                            inner.lock().unwrap().handle_results(init_results);

                            while let Ok(actor_task) = rx_tasks.recv().await {
                                let mut inner = inner.lock().unwrap();
                                inner.handle_task(actor_task);
                                if inner.is_stopped() {
                                    tracing::debug!(?doc_id, ?actor_id, "actor stopped");
                                    break;
                                }
                            }
                        });
                    }
                }
            },
            _ = running_actors.next() => {
                // nothing to do
            }
        }
    }

    // Wait for all actors to stop
    while running_actors.next().await.is_some() {
        // nothing to do
    }
}

/// All the information needed to spawn the background tasks
///
/// When we construct a `Repo` we need to spawn a number of tasks onto the
/// runtime to do things like handle storage tasks. We have to split the
/// spawn process into two stages:
///
/// * Create the channels which are used to communicate with the background tasks
/// * Spawn the background tasks onto the runtime
///
/// The reason we have to split into these two stages is so that we can work with
/// runtimes that don't support non-`Send` tasks. This split is represented by the
/// `TaskSetup::spawn_tasks` and `TaskSetup::spawn_tasks_local` methods.
struct TaskSetup {
    peer_id: PeerId,
    inner: Arc<Mutex<Inner>>,
    tx_io: UnboundedSender<IoLoopTask>,
    rx_storage: UnboundedReceiver<IoLoopTask>,
    rx_from_core: UnboundedReceiver<(DocumentActorId, DocToHubMsg)>,
    rx_actor: Option<UnboundedReceiver<SpawnedActor>>,
    dialers: Arc<Mutex<HashMap<DialerId, io_loop::DynDialer>>>,
    observer: Option<Arc<dyn observer::RepoObserver>>,
}

impl TaskSetup {
    async fn new<S: LocalStorage>(
        storage: S,
        peer_id: Option<PeerId>,
        concurrency: ConcurrencyConfig,
        observer: Option<Arc<dyn observer::RepoObserver>>,
    ) -> TaskSetup {
        let mut rng = rand::rngs::StdRng::from_rng(&mut rand::rng());
        let peer_id = peer_id.unwrap_or_else(|| PeerId::new_with_rng(&mut rng));
        let hub = load_hub(storage.clone(), Hub::load(peer_id.clone())).await;

        let (tx_storage, rx_storage) = unbounded::channel();
        let tx_io_for_tick = tx_storage.clone();
        let (tx_to_core, rx_from_core) = unbounded::channel();
        let rx_actor: Option<UnboundedReceiver<SpawnedActor>>;
        let doc_runner = match concurrency {
            #[cfg(feature = "threadpool")]
            ConcurrencyConfig::Threadpool(threadpool) => {
                rx_actor = None;
                DocRunner::Threadpool(threadpool)
            }
            ConcurrencyConfig::AsyncRuntime => {
                let (tx, rx) = unbounded::channel();
                rx_actor = Some(rx);
                DocRunner::Async {
                    tx_spawn: tx,
                    task_senders: HashMap::new(),
                }
            }
        };

        let dialers = Arc::new(Mutex::new(HashMap::new()));
        let inner = Arc::new(Mutex::new(Inner {
            doc_runner,
            actors: HashMap::new(),
            hub: *hub,
            pending_commands: HashMap::new(),
            connections: HashMap::new(),
            conn_listeners: Vec::new(),
            tx_io: tx_storage,
            tx_to_core,
            waiting_for_connection: HashMap::new(),
            stop_waiters: Vec::new(),
            rng: rand::rngs::StdRng::from_os_rng(),
            dialers: dialers.clone(),
            dialer_handles: HashMap::new(),
            acceptor_handles: HashMap::new(),
            observer: observer.clone(),
        }));

        TaskSetup {
            peer_id,
            inner,
            tx_io: tx_io_for_tick,
            rx_actor,
            rx_from_core,
            rx_storage,
            dialers,
            observer,
        }
    }
    fn spawn_tasks_local<
        R: LocalRuntimeHandle + Clone + 'static,
        S: LocalStorage,
        A: LocalAnnouncePolicy,
    >(
        self,
        runtime: R,
        storage: S,
        announce_policy: A,
    ) {
        runtime.spawn(
            io_loop::io_loop(
                self.peer_id.clone(),
                self.inner.clone(),
                storage,
                announce_policy,
                self.rx_storage,
                self.dialers.clone(),
                self.observer.clone(),
            )
            .boxed_local(),
        );
        runtime.spawn({
            let peer_id = self.peer_id.clone();
            let inner = self.inner.clone();
            async move {
                let rx = self.rx_from_core;
                while let Ok((actor_id, msg)) = rx.recv().await {
                    let event = HubEvent::actor_message(actor_id, msg);
                    inner.lock().unwrap().handle_event(event);
                }
            }
            .instrument(tracing::info_span!("actor_loop", local_peer_id=%peer_id))
            .boxed_local()
        });
        {
            let tx_io = self.tx_io.clone();
            let runtime_for_tick = runtime.clone();
            let sleep = move |d| runtime_for_tick.sleep(d);
            runtime.spawn(connector_tick_loop(tx_io, sleep).boxed_local());
        }
        if let Some(rx_actor) = self.rx_actor {
            runtime.spawn(async_actor_runner(rx_actor).boxed_local());
        }
    }

    fn spawn_tasks<R: RuntimeHandle + Clone + Send, S: Storage, A: AnnouncePolicy>(
        self,
        runtime: R,
        storage: S,
        announce_policy: A,
    ) {
        runtime.spawn(
            io_loop::io_loop(
                self.peer_id.clone(),
                self.inner.clone(),
                storage,
                announce_policy,
                self.rx_storage,
                self.dialers.clone(),
                self.observer.clone(),
            )
            .boxed(),
        );
        runtime.spawn({
            let peer_id = self.peer_id.clone();
            let inner = self.inner.clone();
            async move {
                let rx = self.rx_from_core;
                while let Ok((actor_id, msg)) = rx.recv().await {
                    let event = HubEvent::actor_message(actor_id, msg);
                    inner.lock().unwrap().handle_event(event);
                }
            }
            .instrument(tracing::info_span!("actor_loop", local_peer_id=%peer_id))
            .boxed()
        });
        {
            let tx_io = self.tx_io.clone();
            let runtime_for_tick = runtime.clone();
            let sleep = move |d| runtime_for_tick.sleep(d);
            runtime.spawn(connector_tick_loop(tx_io, sleep).boxed());
        }
        if let Some(rx_actor) = self.rx_actor {
            runtime.spawn(async_actor_runner(rx_actor).boxed());
        }
    }
}

/// Periodically sends tick events to the IO loop to drive time-based connector
/// logic (e.g. retry backoff for dialers).
///
/// `sleep` is provided by the runtime so this loop is not coupled to any
/// specific async runtime.
async fn connector_tick_loop<F, Fut>(tx_io: UnboundedSender<IoLoopTask>, sleep: F)
where
    F: Fn(std::time::Duration) -> Fut,
    Fut: std::future::Future<Output = ()>,
{
    loop {
        sleep(std::time::Duration::from_millis(100)).await;
        if tx_io.unbounded_send(IoLoopTask::Tick).is_err() {
            // IO loop channel closed — repo is shutting down
            break;
        }
    }
}

async fn load_hub<S: LocalStorage>(storage: S, mut loading: samod_core::SamodLoader) -> Box<Hub> {
    let mut rng = rand::rngs::StdRng::from_os_rng();
    let mut running_tasks = FuturesUnordered::new();
    loop {
        match loading.step(&mut rng, UnixTimestamp::now()) {
            LoaderState::NeedIo(items) => {
                for IoTask {
                    task_id,
                    action: task,
                } in items
                {
                    let storage = storage.clone();
                    running_tasks.push(async move {
                        let result = io_loop::dispatch_storage_task(task, storage).await;
                        (task_id, result)
                    })
                }
            }
            LoaderState::Loaded(hub) => break hub,
        }
        let (task_id, next_result) = running_tasks.select_next_some().await;
        loading.provide_io_result(IoResult {
            task_id,
            payload: next_result,
        });
    }
}

#[cfg(test)]
mod tests {
    use std::marker::PhantomData;

    fn assert_send<S: Send>(_s: PhantomData<S>) {}

    #[cfg(feature = "tokio")]
    fn assert_send_value<S: Send>(_s: impl Fn() -> S) {}

    #[test]
    fn make_sure_it_is_send() {
        assert_send::<super::storage::InMemoryStorage>(PhantomData);
        assert_send::<super::Repo>(PhantomData);

        #[cfg(feature = "tokio")]
        assert_send_value(|| crate::Repo::build_tokio().load());
    }
}