use super::*;use prost::Message;use text::{EditOperation, UndoOperation}; pub struct LeftChannelBuffer {    pub channel_id: ChannelId,    pub collaborators: Vec<proto::Collaborator>,    pub connections: Vec<ConnectionId>,} impl Database {    /// Open a channel buffer. Returns the current contents, and adds you to the list of people    /// to notify on changes.    pub async fn join_channel_buffer(        &self,        channel_id: ChannelId,        user_id: UserId,        connection: ConnectionId,    ) -> Result<proto::JoinChannelBufferResponse> {        self.transaction(|tx| async move {            let channel = self.get_channel_internal(channel_id, &tx).await?;            self.check_user_is_channel_participant(&channel, user_id, &tx)                .await?;             let buffer = channel::Model {                id: channel_id,                ..Default::default()            }            .find_related(buffer::Entity)            .one(&*tx)            .await?;             let buffer = if let Some(buffer) = buffer {                buffer            } else {                let buffer = buffer::ActiveModel {                    channel_id: ActiveValue::Set(channel_id),                    ..Default::default()                }                .insert(&*tx)                .await?;                buffer_snapshot::ActiveModel {                    buffer_id: ActiveValue::Set(buffer.id),                    epoch: ActiveValue::Set(0),                    text: ActiveValue::Set(String::new()),                    operation_serialization_version: ActiveValue::Set(                        storage::SERIALIZATION_VERSION,                    ),                }                .insert(&*tx)                .await?;                buffer            };             // Join the collaborators            let mut collaborators = channel_buffer_collaborator::Entity::find()                .filter(channel_buffer_collaborator::Column::ChannelId.eq(channel_id))                .all(&*tx)                .await?;            let replica_ids = collaborators                .iter()                .map(|c| c.replica_id)                .collect::<HashSet<_>>();            let mut replica_id = ReplicaId(0);            while replica_ids.contains(&replica_id) {                replica_id.0 += 1;            }            let collaborator = channel_buffer_collaborator::ActiveModel {                channel_id: ActiveValue::Set(channel_id),                connection_id: ActiveValue::Set(connection.id as i32),                connection_server_id: ActiveValue::Set(ServerId(connection.owner_id as i32)),                user_id: ActiveValue::Set(user_id),                replica_id: ActiveValue::Set(replica_id),                ..Default::default()            }            .insert(&*tx)            .await?;            collaborators.push(collaborator);             let (base_text, operations, max_operation) =                self.get_buffer_state(&buffer, &tx).await?;             // Save the last observed operation            if let Some(op) = max_operation {                observed_buffer_edits::Entity::insert(observed_buffer_edits::ActiveModel {                    user_id: ActiveValue::Set(user_id),                    buffer_id: ActiveValue::Set(buffer.id),                    epoch: ActiveValue::Set(op.epoch),                    lamport_timestamp: ActiveValue::Set(op.lamport_timestamp),                    replica_id: ActiveValue::Set(op.replica_id),                })                .on_conflict(                    OnConflict::columns([                        observed_buffer_edits::Column::UserId,                        observed_buffer_edits::Column::BufferId,                    ])                    .update_columns([                        observed_buffer_edits::Column::Epoch,                        observed_buffer_edits::Column::LamportTimestamp,                    ])                    .to_owned(),                )                .exec(&*tx)                .await?;            }             Ok(proto::JoinChannelBufferResponse {                buffer_id: buffer.id.to_proto(),                replica_id: replica_id.to_proto() as u32,                base_text,                operations,                epoch: buffer.epoch as u64,                collaborators: collaborators                    .into_iter()                    .map(|collaborator| proto::Collaborator {                        peer_id: Some(collaborator.connection().into()),                        user_id: collaborator.user_id.to_proto(),                        replica_id: collaborator.replica_id.0 as u32,                        is_host: false,                    })                    .collect(),            })        })        .await    }     /// Rejoin a channel buffer (after a connection interruption)    pub async fn rejoin_channel_buffers(        &self,        buffers: &[proto::ChannelBufferVersion],        user_id: UserId,        connection_id: ConnectionId,    ) -> Result<Vec<RejoinedChannelBuffer>> {        self.transaction(|tx| async move {            let mut results = Vec::new();            for client_buffer in buffers {                let channel = self                    .get_channel_internal(ChannelId::from_proto(client_buffer.channel_id), &tx)                    .await?;                if self                    .check_user_is_channel_participant(&channel, user_id, &tx)                    .await                    .is_err()                {                    log::info!("user is not a member of channel");                    continue;                }                 let buffer = self.get_channel_buffer(channel.id, &tx).await?;                let mut collaborators = channel_buffer_collaborator::Entity::find()                    .filter(channel_buffer_collaborator::Column::ChannelId.eq(channel.id))                    .all(&*tx)                    .await?;                 // If the buffer epoch hasn't changed since the client lost                // connection, then the client's buffer can be synchronized with                // the server's buffer.                if buffer.epoch as u64 != client_buffer.epoch {                    log::info!("can't rejoin buffer, epoch has changed");                    continue;                }                 // Find the collaborator record for this user's previous lost                // connection. Update it with the new connection id.                let Some(self_collaborator) =                    collaborators.iter_mut().find(|c| c.user_id == user_id)                else {                    log::info!("can't rejoin buffer, no previous collaborator found");                    continue;                };                let old_connection_id = self_collaborator.connection();                *self_collaborator = channel_buffer_collaborator::ActiveModel {                    id: ActiveValue::Unchanged(self_collaborator.id),                    connection_id: ActiveValue::Set(connection_id.id as i32),                    connection_server_id: ActiveValue::Set(ServerId(connection_id.owner_id as i32)),                    connection_lost: ActiveValue::Set(false),                    ..Default::default()                }                .update(&*tx)                .await?;                 let client_version = version_from_wire(&client_buffer.version);                let serialization_version = self                    .get_buffer_operation_serialization_version(buffer.id, buffer.epoch, &tx)                    .await?;                 let mut rows = buffer_operation::Entity::find()                    .filter(                        buffer_operation::Column::BufferId                            .eq(buffer.id)                            .and(buffer_operation::Column::Epoch.eq(buffer.epoch)),                    )                    .stream(&*tx)                    .await?;                 // Find the server's version vector and any operations                // that the client has not seen.                let mut server_version = clock::Global::new();                let mut operations = Vec::new();                while let Some(row) = rows.next().await {                    let row = row?;                    let timestamp = clock::Lamport {                        replica_id: row.replica_id as u16,                        value: row.lamport_timestamp as u32,                    };                    server_version.observe(timestamp);                    if !client_version.observed(timestamp) {                        operations.push(proto::Operation {                            variant: Some(operation_from_storage(row, serialization_version)?),                        })                    }                }                 results.push(RejoinedChannelBuffer {                    old_connection_id,                    buffer: proto::RejoinedChannelBuffer {                        channel_id: client_buffer.channel_id,                        version: version_to_wire(&server_version),                        operations,                        collaborators: collaborators                            .into_iter()                            .map(|collaborator| proto::Collaborator {                                peer_id: Some(collaborator.connection().into()),                                user_id: collaborator.user_id.to_proto(),                                replica_id: collaborator.replica_id.0 as u32,                                is_host: false,                            })                            .collect(),                    },                });            }             Ok(results)        })        .await    }     /// Clear out any buffer collaborators who are no longer collaborating.    pub async fn clear_stale_channel_buffer_collaborators(        &self,        channel_id: ChannelId,        server_id: ServerId,    ) -> Result<RefreshedChannelBuffer> {        self.transaction(|tx| async move {            let db_collaborators = channel_buffer_collaborator::Entity::find()                .filter(channel_buffer_collaborator::Column::ChannelId.eq(channel_id))                .all(&*tx)                .await?;             let mut connection_ids = Vec::new();            let mut collaborators = Vec::new();            let mut collaborator_ids_to_remove = Vec::new();            for db_collaborator in &db_collaborators {                if !db_collaborator.connection_lost                    && db_collaborator.connection_server_id == server_id                {                    connection_ids.push(db_collaborator.connection());                    collaborators.push(proto::Collaborator {                        peer_id: Some(db_collaborator.connection().into()),                        replica_id: db_collaborator.replica_id.0 as u32,                        user_id: db_collaborator.user_id.to_proto(),                        is_host: false,                    })                } else {                    collaborator_ids_to_remove.push(db_collaborator.id);                }            }             channel_buffer_collaborator::Entity::delete_many()                .filter(channel_buffer_collaborator::Column::Id.is_in(collaborator_ids_to_remove))                .exec(&*tx)                .await?;             Ok(RefreshedChannelBuffer {                connection_ids,                collaborators,            })        })        .await    }     /// Close the channel buffer, and stop receiving updates for it.    pub async fn leave_channel_buffer(        &self,        channel_id: ChannelId,        connection: ConnectionId,    ) -> Result<LeftChannelBuffer> {        self.transaction(|tx| async move {            self.leave_channel_buffer_internal(channel_id, connection, &tx)                .await        })        .await    }     /// Close the channel buffer, and stop receiving updates for it.    pub async fn channel_buffer_connection_lost(        &self,        connection: ConnectionId,        tx: &DatabaseTransaction,    ) -> Result<()> {        channel_buffer_collaborator::Entity::update_many()            .filter(                Condition::all()                    .add(channel_buffer_collaborator::Column::ConnectionId.eq(connection.id as i32))                    .add(                        channel_buffer_collaborator::Column::ConnectionServerId                            .eq(connection.owner_id as i32),                    ),            )            .set(channel_buffer_collaborator::ActiveModel {                connection_lost: ActiveValue::set(true),                ..Default::default()            })            .exec(tx)            .await?;        Ok(())    }     /// Close all open channel buffers    pub async fn leave_channel_buffers(        &self,        connection: ConnectionId,    ) -> Result<Vec<LeftChannelBuffer>> {        self.transaction(|tx| async move {            #[derive(Debug, Clone, Copy, EnumIter, DeriveColumn)]            enum QueryChannelIds {                ChannelId,            }             let channel_ids: Vec<ChannelId> = channel_buffer_collaborator::Entity::find()                .select_only()                .column(channel_buffer_collaborator::Column::ChannelId)                .filter(Condition::all().add(                    channel_buffer_collaborator::Column::ConnectionId.eq(connection.id as i32),                ))                .into_values::<_, QueryChannelIds>()                .all(&*tx)                .await?;             let mut result = Vec::new();            for channel_id in channel_ids {                let left_channel_buffer = self                    .leave_channel_buffer_internal(channel_id, connection, &tx)                    .await?;                result.push(left_channel_buffer);            }             Ok(result)        })        .await    }     async fn leave_channel_buffer_internal(        &self,        channel_id: ChannelId,        connection: ConnectionId,        tx: &DatabaseTransaction,    ) -> Result<LeftChannelBuffer> {        let result = channel_buffer_collaborator::Entity::delete_many()            .filter(                Condition::all()                    .add(channel_buffer_collaborator::Column::ChannelId.eq(channel_id))                    .add(channel_buffer_collaborator::Column::ConnectionId.eq(connection.id as i32))                    .add(                        channel_buffer_collaborator::Column::ConnectionServerId                            .eq(connection.owner_id as i32),                    ),            )            .exec(tx)            .await?;        if result.rows_affected == 0 {            Err(anyhow!("not a collaborator on this project"))?;        }         let mut collaborators = Vec::new();        let mut connections = Vec::new();        let mut rows = channel_buffer_collaborator::Entity::find()            .filter(                Condition::all().add(channel_buffer_collaborator::Column::ChannelId.eq(channel_id)),            )            .stream(tx)            .await?;        while let Some(row) = rows.next().await {            let row = row?;            let connection = row.connection();            connections.push(connection);            collaborators.push(proto::Collaborator {                peer_id: Some(connection.into()),                replica_id: row.replica_id.0 as u32,                user_id: row.user_id.to_proto(),                is_host: false,            });        }         drop(rows);         if collaborators.is_empty() {            self.snapshot_channel_buffer(channel_id, tx).await?;        }         Ok(LeftChannelBuffer {            channel_id,            collaborators,            connections,        })    }     pub async fn get_channel_buffer_collaborators(        &self,        channel_id: ChannelId,    ) -> Result<Vec<UserId>> {        self.transaction(|tx| async move {            self.get_channel_buffer_collaborators_internal(channel_id, &tx)                .await        })        .await    }     async fn get_channel_buffer_collaborators_internal(        &self,        channel_id: ChannelId,        tx: &DatabaseTransaction,    ) -> Result<Vec<UserId>> {        #[derive(Debug, Clone, Copy, EnumIter, DeriveColumn)]        enum QueryUserIds {            UserId,        }         let users: Vec<UserId> = channel_buffer_collaborator::Entity::find()            .select_only()            .column(channel_buffer_collaborator::Column::UserId)            .filter(                Condition::all().add(channel_buffer_collaborator::Column::ChannelId.eq(channel_id)),            )            .into_values::<_, QueryUserIds>()            .all(tx)            .await?;         Ok(users)    }     pub async fn update_channel_buffer(        &self,        channel_id: ChannelId,        user: UserId,        operations: &[proto::Operation],    ) -> Result<(HashSet<ConnectionId>, i32, Vec<proto::VectorClockEntry>)> {        self.transaction(move |tx| async move {            let channel = self.get_channel_internal(channel_id, &tx).await?;             let mut requires_write_permission = false;            for op in operations.iter() {                match op.variant {                    None | Some(proto::operation::Variant::UpdateSelections(_)) => {}                    Some(_) => requires_write_permission = true,                }            }            if requires_write_permission {                self.check_user_is_channel_member(&channel, user, &tx)                    .await?;            } else {                self.check_user_is_channel_participant(&channel, user, &tx)                    .await?;            }             let buffer = buffer::Entity::find()                .filter(buffer::Column::ChannelId.eq(channel_id))                .one(&*tx)                .await?                .ok_or_else(|| anyhow!("no such buffer"))?;             let serialization_version = self                .get_buffer_operation_serialization_version(buffer.id, buffer.epoch, &tx)                .await?;             let operations = operations                .iter()                .filter_map(|op| operation_to_storage(op, &buffer, serialization_version))                .collect::<Vec<_>>();             let max_version;             if !operations.is_empty() {                let max_operation = operations                    .iter()                    .max_by_key(|op| (op.lamport_timestamp.as_ref(), op.replica_id.as_ref()))                    .unwrap();                 max_version = vec![proto::VectorClockEntry {                    replica_id: *max_operation.replica_id.as_ref() as u32,                    timestamp: *max_operation.lamport_timestamp.as_ref() as u32,                }];                 // get current channel participants and save the max operation above                self.save_max_operation(                    user,                    buffer.id,                    buffer.epoch,                    *max_operation.replica_id.as_ref(),                    *max_operation.lamport_timestamp.as_ref(),

Lines 1–500 of 1010. Load more