.

davidatsurge · davidatsurge · commit c9384c790811 · 2025-02-13T13:55:47.000+03:00
diff --git a/shared/src/sync_engine/optimistic/db/index.rs b/shared/src/sync_engine/optimistic/db/index.rs
@@ -8,7 +8,7 @@ use typesafe_idb::{Index, IndexSpec};
 use crate::sync_engine::optimistic::optimistic_changes::OptimisticChanges;
 
 use super::reactivity_trackers::ReactivityTrackers;
-use super::Error;
+use super::{Error, MaybeOptimistic};
 
 #[derive(derive_more::Constructor)]
 pub struct IndexWithOptimisticChanges<'txn, IS> {
@@ -18,7 +18,7 @@ pub struct IndexWithOptimisticChanges<'txn, IS> {
     txn_location: &'static Location<'static>,
 }
 impl<IS: IndexSpec> IndexWithOptimisticChanges<'_, IS> {
-    pub async fn get(&self, id: &IS::Type) -> Result<Option<IS::Store>, super::Error> {
+    pub async fn get(&self, id: &IS::Type) -> Result<Option<MaybeOptimistic<IS::Store>>, super::Error> {
         self.reactivity_trackers
             .borrow_mut()
             .add_bulk_read(IS::Store::NAME);
@@ -44,7 +44,8 @@ impl<IS: IndexSpec> IndexWithOptimisticChanges<'_, IS> {
             .optimistic_changes
             .updates
             .latest_downcasted(id)
-            .or(Some(row)))
+            .map(|o| MaybeOptimistic::new(o, true))
+            .or(Some(MaybeOptimistic::new(row, false))))
     }
 
     pub(crate) async fn no_optimism_get(
@@ -58,7 +59,7 @@ impl<IS: IndexSpec> IndexWithOptimisticChanges<'_, IS> {
         self.inner.get(id).await
     }
 
-    pub async fn get_all(&self, value: Option<&IS::Type>) -> Result<Vec<IS::Store>, Error> {
+    pub async fn get_all(&self, value: Option<&IS::Type>) -> Result<Vec<MaybeOptimistic<IS::Store>>, Error> {
         self.reactivity_trackers
             .borrow_mut()
             .add_bulk_read(IS::Store::NAME);
@@ -79,22 +80,24 @@ impl<IS: IndexSpec> IndexWithOptimisticChanges<'_, IS> {
                 self.optimistic_changes
                     .updates
                     .latest_downcasted(r.id())
-                    .unwrap_or(r)
+                    .map(|o| MaybeOptimistic::new(o, true))
+                    .unwrap_or(MaybeOptimistic::new(r, false))
             });
         let mut all = Vec::from_iter(from_db_filtered);
 
         let optimistic_creations = self
             .optimistic_changes
             .creations
-            .all_the_latest_downcasted();
+            .all_the_latest_downcasted() ;
         if let Some(value) = value {
             all.extend(
                 optimistic_creations
                     .into_iter()
-                    .filter(|row| IS::get_index_value(row) == value),
+                    .filter(|row| IS::get_index_value(row) == value)
+                    .map(|o| MaybeOptimistic::new(o, true))
             );
         } else {
-            all.extend(optimistic_creations)
+            all.extend(optimistic_creations.into_iter().map(|o| MaybeOptimistic::new(o, true)));
         }
         Ok(all)
     }
diff --git a/shared/src/sync_engine/optimistic/db/object_store.rs b/shared/src/sync_engine/optimistic/db/object_store.rs
@@ -25,6 +25,26 @@ pub struct MaybeOptimistic<S> {
     is_optimistic: bool,
 }
 
+impl<S> MaybeOptimistic<S> { 
+    pub fn into_inner(self) -> S {
+        self.inner
+    }
+
+    pub fn map<T>(self, f: impl FnOnce(S) -> T) -> MaybeOptimistic<T> {
+        MaybeOptimistic {
+            inner: f(self.inner),
+            is_optimistic: self.is_optimistic,
+        }
+    }
+
+    pub fn map_ref<T>(&self, f: impl FnOnce(&S) -> T) -> MaybeOptimistic<T> {
+        MaybeOptimistic {
+            inner: f(&self.inner),
+            is_optimistic: self.is_optimistic,
+        }
+    }
+}
+
 impl<S, Mode> ObjectStoreWithOptimisticChanges<S, Mode>
 where
     S: Store + 'static,
@@ -150,6 +170,11 @@ where
             .map_err(|e| Error::new(e, self.location))?;
         self.optimistic_changes
             .remove_obsoletes_for_id::<S>(item.id());
+
+        if let Some(commit_listener) = self.commit_listener.as_ref() {
+            commit_listener(&self.reactivity_trackers.borrow());
+        }
+
         Ok(())
     }
     pub fn update(&self, row: S, update_fut: impl Future<Output = Result<(), ()>> + 'static) {
@@ -180,4 +205,4 @@ where
             tracing::info!("invoked commit_listener() from within ObjectStore::create()");
         }
     }
-}
+}
diff --git a/shared/src/sync_engine/optimistic/optimistic_changes.rs b/shared/src/sync_engine/optimistic/optimistic_changes.rs
@@ -101,6 +101,7 @@ impl OptimisticChanges {
     }
 
     pub fn remove_obsoletes_for_id<S: Store>(&self, id: &S::Id) {
+        tracing::info!("Called remove obseletes_for_id: {id:?}");
         self.deletes.remove_all_successful::<S>(id, &());
         self.updates.remove_all_successful::<S>(id, &());
         self.creations
diff --git a/shared/src/sync_engine/websocket_updates/transport.rs b/shared/src/sync_engine/websocket_updates/transport.rs
@@ -1,17 +1,17 @@
-use std::fmt::Debug;
 use codee::{Decoder, Encoder};
 use futures::{Sink, Stream};
 use pin_project::pin_project;
+use std::fmt::Debug;
 use std::task::{ready, Poll};
 use url::Url;
 
 use crate::{
     endpoints::defns::api::websocket_updates::{ClientMsg, ServerMsg},
-    sync_engine::websocket_updates::binary_transport::{BinaryTransportError, BinaryTransportTrait, ConnOrClosedError, JsonSerdeToBinaryCodec},
+    sync_engine::websocket_updates::binary_transport::{
+        BinaryTransportError, BinaryTransportTrait, ConnOrClosedError, JsonSerdeToBinaryCodec,
+    },
 };
 
-
-
 #[pin_project]
 pub struct Transport<I> {
     #[pin]
@@ -34,7 +34,8 @@ where
 
     fn start_send(self: std::pin::Pin<&mut Self>, item: ClientMsg) -> Result<(), Self::Error> {
         let this = self.project();
-        let encoded = JsonSerdeToBinaryCodec::encode(&item).map_err(BinaryTransportError::Encode)?;
+        let encoded =
+            JsonSerdeToBinaryCodec::encode(&item).map_err(BinaryTransportError::Encode)?;
         this.inner
             .start_send(encoded)
             .map_err(BinaryTransportError::Conn)
@@ -95,14 +96,15 @@ where
     }
 }
 
-pub trait TransportTrait: Sized + Sink<ClientMsg> + Stream<Item = Result<ServerMsg, Self::TransportError>> {
+pub trait TransportTrait:
+    Sized + Sink<ClientMsg> + Stream<Item = Result<ServerMsg, Self::TransportError>>
+{
     type TransportError: Debug;
 
     #[allow(async_fn_in_trait)]
     async fn establish(url: &Url) -> Result<Self, Self::TransportError>;
 }
 
-
 impl<T> TransportTrait for Transport<T>
 where
     T: BinaryTransportTrait,
@@ -116,4 +118,72 @@ where
 
         Ok(Transport { inner })
     }
+}
+
+#[cfg(test)]
+pub mod tests {
+    use std::{pin::Pin, task::{Context, Poll}};
+
+    use futures::{channel::mpsc, Sink, Stream};
+
+    use crate::endpoints::defns::api::websocket_updates::{ClientMsg, ServerMsg};
+
+    pub struct MockTransportHandler {
+        send: mpsc::Sender<ServerMsg>,
+        recv: mpsc::Receiver<ClientMsg>,
+    }
+
+    pub struct TestTransport {
+        recv: mpsc::Receiver<ServerMsg>,
+        send: mpsc::Sender<ClientMsg>,
+    }
+
+    impl TestTransport {
+        pub fn new() -> (Self, MockTransportHandler) {
+            let (server_msg_sender, server_msg_receiver) = mpsc::channel(100);
+            let (client_msg_sender, client_msg_receiver) = mpsc::channel(100);
+
+            (
+                Self { recv: server_msg_receiver, send: client_msg_sender },
+                MockTransportHandler { send: server_msg_sender, recv: client_msg_receiver },
+            )
+        }
+    }
+
+    impl Stream for TestTransport {
+        type Item = Result<ServerMsg, mpsc::SendError>;
+
+        fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+            Pin::new(&mut self.recv).poll_next(cx).map(|opt| opt.map(Ok))
+        }
+    }
+
+    impl Sink<ClientMsg> for TestTransport {
+        type Error = mpsc::SendError;
+
+        fn poll_ready(
+            mut self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<Result<(), Self::Error>> {
+            Pin::new(&mut self.send).poll_ready(cx)
+        }
+
+        fn start_send(mut self: Pin<&mut Self>, item: ClientMsg) -> Result<(), Self::Error> {
+            Pin::new(&mut self.send).start_send(item)
+        }
+
+        fn poll_flush(
+            mut self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<Result<(), Self::Error>> {
+            Pin::new(&mut self.send).poll_flush(cx)
+        }
+
+        fn poll_close(
+            mut self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<Result<(), Self::Error>> {
+            Pin::new(&mut self.send).poll_close(cx)
+        }
+    }
 }
diff --git a/web/src/app/repository/issues_tab/list.rs b/web/src/app/repository/issues_tab/list.rs
@@ -1,11 +1,11 @@
 use jiff::{fmt::strtime, Timestamp};
 use leptos::prelude::*;
 use macros::zwang_url;
-use shared::types::{
+use shared::{sync_engine::optimistic::db::MaybeOptimistic, types::{
     issue::{Issue, RepositoryIdIndex},
     issue_comment::{IssueComment, IssueIdIndex},
     user::User,
-};
+}};
 
 use itertools::Itertools;
 use zwang_router::{ArgFromParent, RouteParams, A};
@@ -107,7 +107,7 @@ pub fn IssuesList(
 
 #[component]
 fn IssueRow(
-    issue: Issue,
+    issue: MaybeOptimistic<Issue>,
     #[prop(into)] owner_name: Signal<String>,
     #[prop(into)] repo_name: Signal<String>,
     #[prop(optional)] is_last: bool,
diff --git a/web/src/app/repository/issues_tab/one_issue.rs b/web/src/app/repository/issues_tab/one_issue.rs
@@ -25,7 +25,6 @@ use crate::{
         thirds::Thirds,
     },
     frontend_error::FrontendError,
-    idb_signal::IdbSignal,
     idb_signal_from_sync_engine::IdbSignalFromSyncEngine,
 };
 
@@ -47,7 +46,7 @@ pub fn OneIssue(
             Ok(i) => i,
             Err(_) => return view! { <NotFound /> }.into_any(),
         };
-        let issue_and_user: IdbSignal<Result<Option<_>, _>> = sync_engine.idb_signal(
+        let issue_and_user = sync_engine.idb_signal(
             move |txn| txn.with_store::<User>().with_store::<Issue>().build(),
             move |txn| async move {
                 let issue = txn
@@ -71,6 +70,10 @@ pub fn OneIssue(
             },
         );
 
+        Effect::new(move || {
+            tracing::info!("issue_and_user: {:?}", issue_and_user.get());
+        });
+
         (move || {
             issue_and_user.get().map(|issue_and_user| {
                 let sync_engine = use_sync_engine();
@@ -105,8 +108,6 @@ pub fn OneIssue(
                     Ok(issue_comments.into_iter().zip(users).collect::<Vec<_>>())
                 });
 
-
-
                 Ok(
                 view! {
                     <div>
@@ -165,6 +166,7 @@ pub fn OneIssue(
                                                             each=move || issue_comment_and_users.clone()
                                                             key=|(ic, _)| ic.id
                                                             children=|(issue_comment, user)| {
+                                                                let issue_comment = issue_comment.into_inner();
                                                                 view! {
                                                                     <IssueCommentBox
                                                                         body=issue_comment.body.to_option()
diff --git a/web/src/app/repository/mod.rs b/web/src/app/repository/mod.rs
@@ -1,12 +1,11 @@
 use crate::signal_ext::*;
 use leptos::{prelude::*, task::spawn_local};
 use shared::{
-    types::{
+    sync_engine::optimistic::db::MaybeOptimistic, types::{
         self,
         repository::Repository,
         user::{self, User},
-    },
-    utils::LogErr,
+    }, utils::LogErr
 };
 use top_bar::TopBar;
 use zwang_router::{set_pathname, Outlet, ParsedPath, RouteParams};
@@ -55,8 +54,8 @@ impl From<Tab> for RootOwnerNameRepoName {
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct RepositoryPageContextInner {
-    repository: Repository,
-    user: User
+    repository: MaybeOptimistic<Repository>,
+    user: MaybeOptimistic<User>
 }
 
 pub type RepositoryPageContext = Memo<RepositoryPageContextInner>;

Original file line number	Diff line number	Diff line change
`@@ -101,6 +101,7 @@ impl OptimisticChanges {`
`101`	`101`	`}`
`102`	`102`
`103`	`103`	`pub fn remove_obsoletes_for_id<S: Store>(&self, id: &S::Id) {`
	`104`	`+ tracing::info!("Called remove obseletes_for_id: {id:?}");`
`104`	`105`	`self.deletes.remove_all_successful::<S>(id, &());`
`105`	`106`	`self.updates.remove_all_successful::<S>(id, &());`
`106`	`107`	`self.creations`