refactor(SyncEngine): introduce a TransportTrait: Sink<ClientMsg> + Stream<ServerMsg> and use that

This makes it easier to do testing (hopefully).

Author: davidatsurge

shared/src/sync_engine/changes.rs

@@ -1,4 +1,3 @@
-use crate::sync_engine::websocket_updates::typed_transport::TypedTransportTrait;
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use typesafe_idb::{ReadOnly, Store};
@@ -10,6 +9,7 @@ use crate::types::issue_comment::{IssueComment, IssueCommentId};
 use crate::types::label::{Label, LabelId};
 
 use super::optimistic::db::{DbWithOptimisticChanges, TxnBuilderWithOptimisticChanges, TxnWithOptimisticChanges};
+use super::websocket_updates::transport::TransportTrait;
 use super::{
     super::types::{
         github_app::{GithubApp, GithubAppId},
@@ -425,7 +425,7 @@ where
     }
 }
 
-impl<W: TypedTransportTrait, GithubApi> SyncEngine<W, GithubApi> {
+impl<W: TransportTrait, GithubApi> SyncEngine<W, GithubApi> {
     pub async fn persist_changes<
         Marker: StoreMarkersForChanges,
         Mode: TxnMode<SupportsReadOnly = Present, SupportsReadWrite = Present>,
@@ -457,7 +457,7 @@ impl<W: TypedTransportTrait, GithubApi> SyncEngine<W, GithubApi> {
     }
 }
 
-async fn persist_changes_to_issues<W: TypedTransportTrait, Marker, Mode>(
+async fn persist_changes_to_issues<W: TransportTrait, Marker, Mode>(
     txn: &TxnWithOptimisticChanges<Marker, Mode>,
     issues: HashMap<IssueId, ExistingOrDeleted<Issue>>,
 ) -> SyncResult<(), W>
@@ -485,7 +485,7 @@ where
     Ok(())
 }
 
-async fn persist_changes_to_issue_comments<W: TypedTransportTrait, Marker, Mode>(
+async fn persist_changes_to_issue_comments<W: TransportTrait, Marker, Mode>(
     txn: &TxnWithOptimisticChanges<Marker, Mode>,
     issue_comments: HashMap<IssueCommentId, ExistingOrDeleted<IssueComment>>,
 ) -> SyncResult<(), W>
@@ -514,7 +514,7 @@ where
     Ok(())
 }
 
-async fn persist_changes_to_github_apps<W: TypedTransportTrait, Marker, Mode>(
+async fn persist_changes_to_github_apps<W: TransportTrait, Marker, Mode>(
     txn: &TxnWithOptimisticChanges<Marker, Mode>,
     github_apps: HashMap<GithubAppId, ExistingOrDeleted<GithubApp>>,
 ) -> SyncResult<(), W>
@@ -541,7 +541,7 @@ where
     Ok(())
 }
 
-async fn persist_changes_to_users<W: TypedTransportTrait, Marker, Mode>(
+async fn persist_changes_to_users<W: TransportTrait, Marker, Mode>(
     txn: &TxnWithOptimisticChanges<Marker, Mode>,
     users: HashMap<UserId, ExistingOrDeleted<User>>,
 ) -> SyncResult<(), W>
@@ -569,7 +569,7 @@ where
     Ok(())
 }
 
-async fn persist_changes_to_licenses<W: TypedTransportTrait, Marker, Mode>(
+async fn persist_changes_to_licenses<W: TransportTrait, Marker, Mode>(
     txn: &TxnWithOptimisticChanges<Marker, Mode>,
     licenses: HashMap<LicenseId, ExistingOrDeleted<License>>,
 ) -> SyncResult<(), W>
@@ -597,7 +597,7 @@ where
     Ok(())
 }
 
-async fn persist_changes_to_milestones<W: TypedTransportTrait, Marker, Mode>(
+async fn persist_changes_to_milestones<W: TransportTrait, Marker, Mode>(
     txn: &TxnWithOptimisticChanges<Marker, Mode>,
     milestones: HashMap<MilestoneId, ExistingOrDeleted<Milestone>>,
 ) -> SyncResult<(), W>
@@ -625,7 +625,7 @@ where
     Ok(())
 }
 
-async fn persist_changes_to_repositorys<W: TypedTransportTrait, Marker, Mode>(
+async fn persist_changes_to_repositorys<W: TransportTrait, Marker, Mode>(
     txn: &TxnWithOptimisticChanges<Marker, Mode>,
     repositorys: HashMap<RepositoryId, ExistingOrDeleted<Repository>>,
 ) -> SyncResult<(), W>
@@ -653,7 +653,7 @@ where
     Ok(())
 }
 
-async fn upsert_labels<W: TypedTransportTrait, Marker, Mode>(
+async fn upsert_labels<W: TransportTrait, Marker, Mode>(
     txn: &TxnWithOptimisticChanges<Marker, Mode>,
     labels: HashMap<LabelId, ExistingOrDeleted<Label>>,
 ) -> SyncResult<(), W>

shared/src/sync_engine/error.rs

@@ -4,10 +4,10 @@ use std::{fmt::Debug, panic::Location};
 use super::{
     conversions::conversion_error::ConversionError,
     optimistic::db::Error,
-    websocket_updates::typed_transport::{TypedTransportError, TypedTransportTrait},
+    websocket_updates::transport::TransportTrait,
 };
 
-pub enum SyncErrorSrc<T: TypedTransportTrait> {
+pub enum SyncErrorSrc<T: TransportTrait> {
     OwnApi(OwnApiError),
     Github(github_api::simple_error::SimpleError),
     Db(Error),
@@ -18,11 +18,11 @@ pub enum SyncErrorSrc<T: TypedTransportTrait> {
     Ewebsock(ewebsock::Error),
     /// These are things like: the user that owns a repository in our db not existing in our db.
     DataModel(String),
-    WebSocket(TypedTransportError<T::ConnError>),
+    Transport(T::TransportError),
     NotAvailable(NotAvailableError)
 }
 
-impl<T: TypedTransportTrait> Debug for SyncErrorSrc<T> {
+impl<T: TransportTrait> Debug for SyncErrorSrc<T> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             SyncErrorSrc::OwnApi(err) => write!(f, "SyncErrorSrc::OwnApi({:?})", err),
@@ -34,13 +34,13 @@ impl<T: TypedTransportTrait> Debug for SyncErrorSrc<T> {
             SyncErrorSrc::Merge(err) => write!(f, "SyncErrorSrc::Merge({:?})", err),
             SyncErrorSrc::Ewebsock(err) => write!(f, "SyncErrorSrc::Ewebsock({:?})", err),
             SyncErrorSrc::DataModel(msg) => write!(f, "SyncErrorSrc::DataModel({})", msg),
-            SyncErrorSrc::WebSocket(err) => write!(f, "SyncErrorSrc::WebSocket({:?})", err),
+            SyncErrorSrc::Transport(err) => write!(f, "SyncErrorSrc::WebSocket({:?})", err),
             SyncErrorSrc::NotAvailable(err) => write!(f, "SyncErrorSrc::NotAvailable({:?})", err),
         }
     }
 }
 
-impl<T: TypedTransportTrait> From<SyncErrorSrc<T>> for SyncError<T> {
+impl<T: TransportTrait> From<SyncErrorSrc<T>> for SyncError<T> {
     #[track_caller]
     fn from(value: SyncErrorSrc<T>) -> Self {
         Self {
@@ -51,12 +51,12 @@ impl<T: TypedTransportTrait> From<SyncErrorSrc<T>> for SyncError<T> {
 }
 
 #[allow(dead_code)]
-pub struct SyncError<T: TypedTransportTrait> {
+pub struct SyncError<T: TransportTrait> {
     source: SyncErrorSrc<T>,
     location: &'static Location<'static>
 }
 
-impl<T: TypedTransportTrait> Debug for SyncError<T> {
+impl<T: TransportTrait> Debug for SyncError<T> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("SyncError")
             .field("source", &self.source)
@@ -65,7 +65,7 @@ impl<T: TypedTransportTrait> Debug for SyncError<T> {
     }
 }
 
-impl<T: TypedTransportTrait> From<ConversionError> for SyncError<T> {
+impl<T: TransportTrait> From<ConversionError> for SyncError<T> {
     fn from(value: ConversionError) -> Self {
         match value {
             ConversionError::Merge(merge_error) => SyncErrorSrc::Merge(merge_error),
@@ -76,7 +76,7 @@ impl<T: TypedTransportTrait> From<ConversionError> for SyncError<T> {
     }
 }
 
-impl<T: TypedTransportTrait> SyncError<T> {
+impl<T: TransportTrait> SyncError<T> {
     /// We don't derive `From<>` because a `String` might accidentally get converted (which is what
     /// `ewebsock::Error` really is).
     pub fn from_ewebsock(error: ewebsock::Error) -> Self {
@@ -86,48 +86,42 @@ impl<T: TypedTransportTrait> SyncError<T> {
 
 pub type SyncResult<T, W> = Result<T, SyncError<W>>;
 
-impl<W: TypedTransportTrait, T> From<github_api::apis::Error<T>> for SyncError<W> {
+impl<W: TransportTrait, T> From<github_api::apis::Error<T>> for SyncError<W> {
     #[track_caller]
     fn from(value: github_api::apis::Error<T>) -> Self {
         SyncErrorSrc::Github(value.into()).into()
     }
 }
 
-impl<W: TypedTransportTrait> From<TypedTransportError<W::ConnError>> for SyncError<W> {
-    #[track_caller]
-    fn from(value: TypedTransportError<W::ConnError>) -> Self {
-        SyncErrorSrc::WebSocket(value).into()
-    }
-}
 
-impl<W: TypedTransportTrait> From<OwnApiError> for SyncError<W> {
+impl<W: TransportTrait> From<OwnApiError> for SyncError<W> {
     #[track_caller]
     fn from(value: OwnApiError) -> Self {
         SyncErrorSrc::OwnApi(value).into()
     }
 }
 
-impl<W: TypedTransportTrait> From<Error> for SyncError<W> {
+impl<W: TransportTrait> From<Error> for SyncError<W> {
     #[track_caller]
     fn from(value: Error) -> Self {
         SyncErrorSrc::Db(value).into()
     }
 }
 
-impl<W: TypedTransportTrait> From<serde_json::Error> for SyncError<W> {
+impl<W: TransportTrait> From<serde_json::Error> for SyncError<W> {
     #[track_caller]
     fn from(value: serde_json::Error) -> Self {
         SyncErrorSrc::SerdeToString(value).into()
     }
 }
-impl<W: TypedTransportTrait> From<MergeError> for SyncError<W> {
+impl<W: TransportTrait> From<MergeError> for SyncError<W> {
     #[track_caller]
     fn from(value: MergeError) -> Self {
         SyncErrorSrc::Merge(value).into()
     }
 }
 
-impl<W: TypedTransportTrait> From<NotAvailableError> for SyncError<W> {
+impl<W: TransportTrait> From<NotAvailableError> for SyncError<W> {
     #[track_caller]
     fn from(value: NotAvailableError) -> Self {
         SyncErrorSrc::NotAvailable(value).into()

shared/src/sync_engine/initial_sync/ensure_initial_sync_issue_comments.rs

@@ -10,7 +10,7 @@ use super::super::{
     SyncEngine, MAX_PER_PAGE,
 };
 use crate::{
-    sync_engine::websocket_updates::typed_transport::TypedTransportTrait,
+    sync_engine::websocket_updates::transport::TransportTrait,
     types::{
         installation::InstallationId,
         issue::{Issue, IssueId, NumberIndex},
@@ -22,7 +22,7 @@ use crate::{
     },
 };
 
-impl<W: TypedTransportTrait, GithubApi: GithubApiTrait> SyncEngine<W, GithubApi> {
+impl<W: TransportTrait, GithubApi: GithubApiTrait> SyncEngine<W, GithubApi> {
     /// This function will try to find issue ids in the db by using the issue number in `issue_url`
     /// of issue_comment`.
     pub async fn ensure_initial_sync_issue_comments(

shared/src/sync_engine/initial_sync/ensure_initial_sync_issues.rs

@@ -4,7 +4,7 @@ use super::super::{
     changes::{AddChanges, Changes},
     conversions::ToDb,
     error::{SyncErrorSrc, SyncResult},
-    SyncEngine, TypedTransportTrait, MAX_PER_PAGE,
+    SyncEngine, MAX_PER_PAGE,
 };
 use crate::{
     avail::MergeError,
@@ -17,8 +17,9 @@ use crate::{
     },
 };
 use github_api::github_api_trait::GithubApiTrait;
+use crate::sync_engine::websocket_updates::transport::TransportTrait;
 
-impl<W: TypedTransportTrait, GithubApi: GithubApiTrait> SyncEngine<W, GithubApi> {
+impl<W: TransportTrait, GithubApi: GithubApiTrait> SyncEngine<W, GithubApi> {
     pub async fn ensure_initial_sync_issues(
         &self,
         id: &RepositoryId,

shared/src/sync_engine/initial_sync/ensure_initial_sync_repository.rs

@@ -1,13 +1,13 @@
 use github_api::github_api_trait::GithubApiTrait;
 
-use crate::{sync_engine::websocket_updates::typed_transport, types::{
+use crate::{sync_engine::websocket_updates::transport::TransportTrait, types::{
     repository::{Repository, RepositoryId},
     repository_initial_sync_status::{RepoSyncStatus, RepositoryInitialSyncStatus},
 }};
 
 use super::super::{error::SyncResult, SyncEngine};
 
-impl<T: typed_transport::TypedTransportTrait, GithubApi: GithubApiTrait> SyncEngine<T, GithubApi> {
+impl<T: TransportTrait, GithubApi: GithubApiTrait> SyncEngine<T, GithubApi> {
     /// `force_initial_sync` means we ignore the RepositoryInitialSyncStatus. This will come into
     /// play when we implement the "if the last time we were in touch is less than 7 days, do a
     /// full resync."

shared/src/sync_engine/initial_sync/fetch_repositorys_for_installation_id.rs

@@ -2,7 +2,7 @@ use futures::future::try_join_all;
 use github_api::github_api_trait::GithubApiTrait;
 
 use crate::{
-    avail::MergeError, sync_engine::websocket_updates::typed_transport::TypedTransportTrait,
+    avail::MergeError, sync_engine::websocket_updates::transport::TransportTrait,
     types::installation::InstallationId,
 };
 
@@ -12,7 +12,7 @@ use super::super::{
     SyncEngine, SyncResult, MAX_PER_PAGE,
 };
 
-impl<W: TypedTransportTrait, GithubApi: GithubApiTrait> SyncEngine<W, GithubApi> {
+impl<W: TransportTrait, GithubApi: GithubApiTrait> SyncEngine<W, GithubApi> {
     pub async fn fetch_repositorys_for_installation_id(
         &self,
         id: &InstallationId,

shared/src/sync_engine/mod.rs

@@ -2,7 +2,7 @@ use optimistic::db::{DbWithOptimisticChanges, ReactivityTrackers};
 use registry::Registry;
 use send_wrapper::SendWrapper;
 use std::{marker::PhantomData, sync::Arc};
-pub use websocket_updates::typed_transport::*;
+pub use websocket_updates::transport::*;
 mod conversions;
 mod initial_sync;
 
@@ -12,7 +12,7 @@ pub mod mutations;
 pub mod optimistic;
 mod registry;
 pub mod storage_traits;
-mod websocket_updates;
+pub mod websocket_updates;
 
 use std::{cmp::Ordering, rc::Rc};
 
@@ -49,7 +49,7 @@ pub struct SyncEngine<Transport, GithubApi> {
 
 const MAX_PER_PAGE: i32 = 100;
 
-impl<W: TypedTransportTrait, GithubApi> SyncEngine<W, GithubApi> {
+impl<W: TransportTrait, GithubApi> SyncEngine<W, GithubApi> {
     async fn get_api_conf(
         &self,
         id: &InstallationId,

shared/src/sync_engine/mutations/issues.rs

@@ -4,16 +4,16 @@ use jiff::Timestamp;
 use crate::{
     random::random, sync_engine::{
         error::{SyncError, SyncErrorSrc},
-        SyncEngine, TypedTransportTrait,
+        SyncEngine, 
     }, types::{
         installation::InstallationId,
         issue::{Issue, IssueId},
         repository::{Repository, RepositoryId},
         user::{User, UserId},
-    }
+    }, sync_engine::websocket_updates::transport::TransportTrait,
 };
 
-impl<T: TypedTransportTrait, GithubApi: GithubApiTrait> SyncEngine<T, GithubApi> {
+impl<T: TransportTrait, GithubApi: GithubApiTrait> SyncEngine<T, GithubApi> {
     /// Returns the optimistic id of the issue.
     /// 
     /// Invariant upheld: The issue number and id will be a negative number for the optimistic issue.

shared/src/sync_engine/new_defn.rs

@@ -22,7 +22,7 @@ use typesafe_idb::{StoreMarker, TypesafeDb};
 
 use super::optimistic::db::DbWithOptimisticChanges;
 use super::registry::Registry;
-use super::websocket_updates::typed_transport::TypedTransportTrait;
+use super::websocket_updates::transport::TransportTrait;
 use super::DbSubscription;
 use super::{error::SyncResult, SyncEngine};
 
@@ -41,7 +41,7 @@ pub type DbStoreMarkers = impl StoreMarker<IssueCommentsInitialSyncStatus>
     + StoreMarker<LastWebhookUpdateAt>
     + Default;
 
-impl<W: TypedTransportTrait, GithubApi> SyncEngine<W, GithubApi> {
+impl<W: TransportTrait, GithubApi> SyncEngine<W, GithubApi> {
     pub async fn new(endpoint_client: EndpointClient) -> SyncResult<Self, W> {
         let db = TypesafeDb::builder("heimisch".into())
             .with_store::<Issue>()

shared/src/sync_engine/optimistic/db/tests.rs

@@ -9,7 +9,7 @@ use wasm_bindgen_test::wasm_bindgen_test;
 use crate::{
     endpoints::endpoint_client::EndpointClient,
     sync_engine::{
-        websocket_updates::tests::MockTypedTransport, DbStoreMarkers, DbSubscription, SyncEngine,
+        websocket_updates::tests::MockTransport, DbStoreMarkers, DbSubscription, SyncEngine,
     },
     types::{
         issue::{Issue, RepositoryIdIndex},
@@ -19,8 +19,8 @@ use crate::{
 
 use super::{TxnBuilderWithOptimisticChanges, TxnWithOptimisticChanges};
 
-async fn get_sync_engine() -> SyncEngine<MockTypedTransport, ()> {
-    SyncEngine::<MockTypedTransport, ()>::new(EndpointClient::new(
+async fn get_sync_engine() -> SyncEngine<MockTransport, ()> {
+    SyncEngine::<MockTransport, ()>::new(EndpointClient::new(
         |_| (),
         Url::parse("https://www.example.com/").unwrap(),
     ))