Fix notes unread status (#7643)

1. The client-side comparison was wrong
2. The server never told the client about the version it remembered
3. The server generated broken timestamps in some cases

Release Notes:

- Fixed the notes/chat appearing as unread too often

**or**

- N/A

crates/collab/src/db/queries/buffers.rs

@@ -561,7 +561,6 @@ impl Database {
         tx: &DatabaseTransaction,
     ) -> Result<()> {
         use observed_buffer_edits::Column;
-
         observed_buffer_edits::Entity::insert(observed_buffer_edits::ActiveModel {
             user_id: ActiveValue::Set(user_id),
             buffer_id: ActiveValue::Set(buffer_id),
@@ -671,7 +670,7 @@ impl Database {
                 buffer_id: row.buffer_id,
                 epoch: row.epoch,
                 lamport_timestamp: row.lamport_timestamp,
-                replica_id: row.lamport_timestamp,
+                replica_id: row.replica_id,
                 value: Default::default(),
             });
             operations.push(proto::Operation {
@@ -750,20 +749,9 @@ impl Database {
 
     pub async fn latest_channel_buffer_changes(
         &self,
-        channel_ids: &[ChannelId],
+        channel_ids_by_buffer_id: &HashMap<BufferId, ChannelId>,
         tx: &DatabaseTransaction,
     ) -> Result<Vec<proto::ChannelBufferVersion>> {
-        let mut channel_ids_by_buffer_id = HashMap::default();
-        let mut rows = buffer::Entity::find()
-            .filter(buffer::Column::ChannelId.is_in(channel_ids.iter().copied()))
-            .stream(&*tx)
-            .await?;
-        while let Some(row) = rows.next().await {
-            let row = row?;
-            channel_ids_by_buffer_id.insert(row.id, row.channel_id);
-        }
-        drop(rows);
-
         let latest_operations = self
             .get_latest_operations_for_buffers(channel_ids_by_buffer_id.keys().copied(), &*tx)
             .await?;
@@ -783,6 +771,36 @@ impl Database {
             .collect())
     }
 
+    pub async fn observed_channel_buffer_changes(
+        &self,
+        channel_ids_by_buffer_id: &HashMap<BufferId, ChannelId>,
+        user_id: UserId,
+        tx: &DatabaseTransaction,
+    ) -> Result<Vec<proto::ChannelBufferVersion>> {
+        let observed_operations = observed_buffer_edits::Entity::find()
+            .filter(observed_buffer_edits::Column::UserId.eq(user_id))
+            .filter(
+                observed_buffer_edits::Column::BufferId
+                    .is_in(channel_ids_by_buffer_id.keys().copied()),
+            )
+            .all(&*tx)
+            .await?;
+
+        Ok(observed_operations
+            .iter()
+            .flat_map(|op| {
+                Some(proto::ChannelBufferVersion {
+                    channel_id: channel_ids_by_buffer_id.get(&op.buffer_id)?.to_proto(),
+                    epoch: op.epoch as u64,
+                    version: vec![proto::VectorClockEntry {
+                        replica_id: op.replica_id as u32,
+                        timestamp: op.lamport_timestamp as u32,
+                    }],
+                })
+            })
+            .collect())
+    }
+
     /// Returns the latest operations for the buffers with the specified IDs.
     pub async fn get_latest_operations_for_buffers(
         &self,

crates/collab/src/db/queries/channels.rs

@@ -673,18 +673,40 @@ impl Database {
         }
 
         let channel_ids = channels.iter().map(|c| c.id).collect::<Vec<_>>();
+
+        let mut channel_ids_by_buffer_id = HashMap::default();
+        let mut rows = buffer::Entity::find()
+            .filter(buffer::Column::ChannelId.is_in(channel_ids.iter().copied()))
+            .stream(&*tx)
+            .await?;
+        while let Some(row) = rows.next().await {
+            let row = row?;
+            channel_ids_by_buffer_id.insert(row.id, row.channel_id);
+        }
+        drop(rows);
+
         let latest_buffer_versions = self
-            .latest_channel_buffer_changes(&channel_ids, &*tx)
+            .latest_channel_buffer_changes(&channel_ids_by_buffer_id, &*tx)
             .await?;
 
-        let latest_messages = self.latest_channel_messages(&channel_ids, &*tx).await?;
+        let latest_channel_messages = self.latest_channel_messages(&channel_ids, &*tx).await?;
+
+        let observed_buffer_versions = self
+            .observed_channel_buffer_changes(&channel_ids_by_buffer_id, user_id, &*tx)
+            .await?;
+
+        let observed_channel_messages = self
+            .observed_channel_messages(&channel_ids, user_id, &*tx)
+            .await?;
 
         Ok(ChannelsForUser {
             channel_memberships,
             channels,
             channel_participants,
             latest_buffer_versions,
-            latest_channel_messages: latest_messages,
+            latest_channel_messages,
+            observed_buffer_versions,
+            observed_channel_messages,
         })
     }
 

crates/collab/src/db/queries/messages.rs

@@ -388,6 +388,30 @@ impl Database {
         Ok(())
     }
 
+    pub async fn observed_channel_messages(
+        &self,
+        channel_ids: &[ChannelId],
+        user_id: UserId,
+        tx: &DatabaseTransaction,
+    ) -> Result<Vec<proto::ChannelMessageId>> {
+        let rows = observed_channel_messages::Entity::find()
+            .filter(observed_channel_messages::Column::UserId.eq(user_id))
+            .filter(
+                observed_channel_messages::Column::ChannelId
+                    .is_in(channel_ids.iter().map(|id| id.0)),
+            )
+            .all(&*tx)
+            .await?;
+
+        Ok(rows
+            .into_iter()
+            .map(|message| proto::ChannelMessageId {
+                channel_id: message.channel_id.to_proto(),
+                message_id: message.channel_message_id.to_proto(),
+            })
+            .collect())
+    }
+
     pub async fn latest_channel_messages(
         &self,
         channel_ids: &[ChannelId],