Implement final move, link, unlink db APIs
This commit is contained in:
parent
7fa68a9aa4
commit
9afb67f2cf
8 changed files with 160 additions and 174 deletions
|
@ -48,7 +48,6 @@ impl Database {
|
|||
.insert(&*tx)
|
||||
.await?;
|
||||
|
||||
let channel_paths_stmt;
|
||||
if let Some(parent) = parent {
|
||||
let sql = r#"
|
||||
INSERT INTO channel_paths
|
||||
|
@ -60,7 +59,7 @@ impl Database {
|
|||
WHERE
|
||||
channel_id = $3
|
||||
"#;
|
||||
channel_paths_stmt = Statement::from_sql_and_values(
|
||||
let channel_paths_stmt = Statement::from_sql_and_values(
|
||||
self.pool.get_database_backend(),
|
||||
sql,
|
||||
[
|
||||
|
@ -796,6 +795,8 @@ impl Database {
|
|||
return Err(anyhow!("Cannot create a channel cycle").into());
|
||||
}
|
||||
}
|
||||
|
||||
// Now insert all of the new paths
|
||||
let sql = r#"
|
||||
INSERT INTO channel_paths
|
||||
(id_path, channel_id)
|
||||
|
@ -832,6 +833,21 @@ impl Database {
|
|||
}
|
||||
}
|
||||
|
||||
// If we're linking a channel, remove any root edges for the channel
|
||||
{
|
||||
let sql = r#"
|
||||
DELETE FROM channel_paths
|
||||
WHERE
|
||||
id_path = '/' || $1 || '/'
|
||||
"#;
|
||||
let channel_paths_stmt = Statement::from_sql_and_values(
|
||||
self.pool.get_database_backend(),
|
||||
sql,
|
||||
[channel.to_proto().into()],
|
||||
);
|
||||
tx.execute(channel_paths_stmt).await?;
|
||||
}
|
||||
|
||||
if let Some(channel) = from_descendants.get_mut(&channel) {
|
||||
// Remove the other parents
|
||||
channel.clear();
|
||||
|
@ -849,7 +865,7 @@ impl Database {
|
|||
&self,
|
||||
user: UserId,
|
||||
channel: ChannelId,
|
||||
from: Option<ChannelId>,
|
||||
from: ChannelId,
|
||||
) -> Result<()> {
|
||||
self.transaction(|tx| async move {
|
||||
// Note that even with these maxed permissions, this linking operation
|
||||
|
@ -927,10 +943,6 @@ impl Database {
|
|||
self.unlink_channel_internal(user, channel, from, &*tx)
|
||||
.await?;
|
||||
|
||||
dbg!(channel_path::Entity::find().all(&*tx).await);
|
||||
|
||||
dbg!(&moved_channels);
|
||||
|
||||
Ok(moved_channels)
|
||||
})
|
||||
.await
|
||||
|
|
|
@ -664,7 +664,7 @@ async fn test_channels_moving(db: &Arc<Database>) {
|
|||
.unlink_channel(
|
||||
a_id,
|
||||
livestreaming_dag_sub_id,
|
||||
Some(livestreaming_id),
|
||||
livestreaming_id,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
@ -688,7 +688,7 @@ async fn test_channels_moving(db: &Arc<Database>) {
|
|||
|
||||
// ========================================================================
|
||||
// Test unlinking in a complex DAG by removing the inner link
|
||||
db.unlink_channel(a_id, livestreaming_id, Some(gpui2_id))
|
||||
db.unlink_channel(a_id, livestreaming_id, gpui2_id)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
|
@ -709,7 +709,7 @@ async fn test_channels_moving(db: &Arc<Database>) {
|
|||
|
||||
// ========================================================================
|
||||
// Test moving DAG nodes by moving livestreaming to be below gpui2
|
||||
db.move_channel(a_id, livestreaming_id, Some(crdb_id), gpui2_id)
|
||||
db.move_channel(a_id, livestreaming_id, crdb_id, gpui2_id)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
|
@ -746,7 +746,7 @@ async fn test_channels_moving(db: &Arc<Database>) {
|
|||
|
||||
// ========================================================================
|
||||
// Unlinking a channel from it's parent should automatically promote it to a root channel
|
||||
db.unlink_channel(a_id, crdb_id, Some(zed_id))
|
||||
db.unlink_channel(a_id, crdb_id, zed_id)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
|
@ -764,29 +764,9 @@ async fn test_channels_moving(db: &Arc<Database>) {
|
|||
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
|
||||
]);
|
||||
|
||||
// ========================================================================
|
||||
// Unlinking a root channel should not have any effect
|
||||
db.unlink_channel(a_id, crdb_id, None)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// DAG is now:
|
||||
// crdb
|
||||
// zed
|
||||
// \- livestreaming - livestreaming_dag - livestreaming_dag_sub
|
||||
//
|
||||
let result = db.get_channels_for_user(a_id).await.unwrap();
|
||||
assert_dag(result.channels, &[
|
||||
(zed_id, None),
|
||||
(crdb_id, None),
|
||||
(livestreaming_id, Some(zed_id)),
|
||||
(livestreaming_dag_id, Some(livestreaming_id)),
|
||||
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
|
||||
]);
|
||||
|
||||
// ========================================================================
|
||||
// You should be able to move a root channel into a non-root channel
|
||||
db.move_channel(a_id, crdb_id, None, zed_id)
|
||||
db.link_channel(a_id, crdb_id, zed_id)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
|
@ -805,8 +785,8 @@ async fn test_channels_moving(db: &Arc<Database>) {
|
|||
|
||||
|
||||
// ========================================================================
|
||||
// Moving a non-root channel without a parent id should be the equivalent of a link operation
|
||||
db.move_channel(a_id, livestreaming_id, None, crdb_id)
|
||||
// Prep for DAG deletion test
|
||||
db.link_channel(a_id, livestreaming_id, crdb_id)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
|
@ -824,10 +804,10 @@ async fn test_channels_moving(db: &Arc<Database>) {
|
|||
(livestreaming_dag_sub_id, Some(livestreaming_dag_id)),
|
||||
]);
|
||||
|
||||
// ========================================================================
|
||||
// Deleting a parent of a DAG should delete the whole DAG:
|
||||
// Deleting the parent of a DAG should delete the whole DAG:
|
||||
db.delete_channel(zed_id, a_id).await.unwrap();
|
||||
let result = db.get_channels_for_user(a_id).await.unwrap();
|
||||
|
||||
assert!(
|
||||
result.channels.is_empty()
|
||||
)
|
||||
|
|
|
@ -2434,17 +2434,16 @@ async fn unlink_channel(
|
|||
) -> Result<()> {
|
||||
let db = session.db().await;
|
||||
let channel_id = ChannelId::from_proto(request.channel_id);
|
||||
let from = request.from.map(ChannelId::from_proto);
|
||||
|
||||
// Get the members before we remove it, so we know who to notify
|
||||
let members = db.get_channel_members(channel_id).await?;
|
||||
let from = ChannelId::from_proto(request.from);
|
||||
|
||||
db.unlink_channel(session.user_id, channel_id, from).await?;
|
||||
|
||||
let members = db.get_channel_members(from).await?;
|
||||
|
||||
let update = proto::UpdateChannels {
|
||||
delete_channel_edge: vec![proto::ChannelEdge {
|
||||
channel_id: channel_id.to_proto(),
|
||||
parent_id: from.map(ChannelId::to_proto),
|
||||
parent_id: from.to_proto(),
|
||||
}],
|
||||
..Default::default()
|
||||
};
|
||||
|
@ -2467,38 +2466,31 @@ async fn move_channel(
|
|||
) -> Result<()> {
|
||||
let db = session.db().await;
|
||||
let channel_id = ChannelId::from_proto(request.channel_id);
|
||||
let from_parent = request.from.map(ChannelId::from_proto);
|
||||
let from_parent = ChannelId::from_proto(request.from);
|
||||
let to = ChannelId::from_proto(request.to);
|
||||
|
||||
let mut members = db.get_channel_members(channel_id).await?;
|
||||
let members_from = db.get_channel_members(channel_id).await?;
|
||||
|
||||
let channels_to_send: Vec<Channel> = db
|
||||
.move_channel(session.user_id, channel_id, from_parent, to)
|
||||
.await?;
|
||||
|
||||
let members_after = db.get_channel_members(channel_id).await?;
|
||||
let members_to = db.get_channel_members(channel_id).await?;
|
||||
|
||||
members.extend(members_after);
|
||||
members.sort();
|
||||
members.dedup();
|
||||
|
||||
if let Some(from_parent) = from_parent {
|
||||
let update = proto::UpdateChannels {
|
||||
delete_channel_edge: vec![proto::ChannelEdge {
|
||||
channel_id: channel_id.to_proto(),
|
||||
parent_id: Some(from_parent.to_proto()),
|
||||
}],
|
||||
..Default::default()
|
||||
};
|
||||
let connection_pool = session.connection_pool().await;
|
||||
for member_id in members {
|
||||
for connection_id in connection_pool.user_connection_ids(member_id) {
|
||||
session.peer.send(connection_id, update.clone())?;
|
||||
}
|
||||
let update = proto::UpdateChannels {
|
||||
delete_channel_edge: vec![proto::ChannelEdge {
|
||||
channel_id: channel_id.to_proto(),
|
||||
parent_id: from_parent.to_proto(),
|
||||
}],
|
||||
..Default::default()
|
||||
};
|
||||
let connection_pool = session.connection_pool().await;
|
||||
for member_id in members_from {
|
||||
for connection_id in connection_pool.user_connection_ids(member_id) {
|
||||
session.peer.send(connection_id, update.clone())?;
|
||||
}
|
||||
}
|
||||
|
||||
let connection_pool = session.connection_pool().await;
|
||||
let update = proto::UpdateChannels {
|
||||
channels: channels_to_send
|
||||
.into_iter()
|
||||
|
@ -2510,7 +2502,7 @@ async fn move_channel(
|
|||
.collect(),
|
||||
..Default::default()
|
||||
};
|
||||
for member_id in members {
|
||||
for member_id in members_to {
|
||||
for connection_id in connection_pool.user_connection_ids(member_id) {
|
||||
session.peer.send(connection_id, update.clone())?;
|
||||
}
|
||||
|
|
|
@ -905,10 +905,15 @@ async fn test_channel_moving(
|
|||
(&client_a, cx_a),
|
||||
)
|
||||
.await;
|
||||
let channel_a_a_id = channels[0];
|
||||
let channel_a_b_id = channels[1];
|
||||
let channel_a_c_id = channels[2];
|
||||
let channel_a_d_id = channels[3];
|
||||
let channel_a_id = channels[0];
|
||||
let channel_b_id = channels[1];
|
||||
let channel_c_id = channels[2];
|
||||
let channel_d_id = channels[3];
|
||||
|
||||
dbg!(channel_a_id);
|
||||
dbg!(channel_b_id);
|
||||
dbg!(channel_c_id);
|
||||
dbg!(channel_d_id);
|
||||
|
||||
// Current shape:
|
||||
// a - b - c - d
|
||||
|
@ -916,17 +921,17 @@ async fn test_channel_moving(
|
|||
client_a.channel_store(),
|
||||
cx_a,
|
||||
&[
|
||||
(channel_a_a_id, 0),
|
||||
(channel_a_b_id, 1),
|
||||
(channel_a_c_id, 2),
|
||||
(channel_a_d_id, 3),
|
||||
(channel_a_id, 0),
|
||||
(channel_b_id, 1),
|
||||
(channel_c_id, 2),
|
||||
(channel_d_id, 3),
|
||||
],
|
||||
);
|
||||
|
||||
client_a
|
||||
.channel_store()
|
||||
.update(cx_a, |channel_store, cx| {
|
||||
channel_store.move_channel(channel_a_d_id, Some(channel_a_c_id), channel_a_b_id, cx)
|
||||
channel_store.move_channel(channel_d_id, channel_c_id, channel_b_id, cx)
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
@ -938,17 +943,17 @@ async fn test_channel_moving(
|
|||
client_a.channel_store(),
|
||||
cx_a,
|
||||
&[
|
||||
(channel_a_a_id, 0),
|
||||
(channel_a_b_id, 1),
|
||||
(channel_a_c_id, 2),
|
||||
(channel_a_d_id, 2),
|
||||
(channel_a_id, 0),
|
||||
(channel_b_id, 1),
|
||||
(channel_c_id, 2),
|
||||
(channel_d_id, 2),
|
||||
],
|
||||
);
|
||||
|
||||
client_a
|
||||
.channel_store()
|
||||
.update(cx_a, |channel_store, cx| {
|
||||
channel_store.link_channel(channel_a_d_id, channel_a_c_id, cx)
|
||||
channel_store.link_channel(channel_d_id, channel_c_id, cx)
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
@ -960,11 +965,11 @@ async fn test_channel_moving(
|
|||
client_a.channel_store(),
|
||||
cx_a,
|
||||
&[
|
||||
(channel_a_a_id, 0),
|
||||
(channel_a_b_id, 1),
|
||||
(channel_a_c_id, 2),
|
||||
(channel_a_d_id, 3),
|
||||
(channel_a_d_id, 2),
|
||||
(channel_a_id, 0),
|
||||
(channel_b_id, 1),
|
||||
(channel_c_id, 2),
|
||||
(channel_d_id, 3),
|
||||
(channel_d_id, 2),
|
||||
],
|
||||
);
|
||||
|
||||
|
@ -978,9 +983,9 @@ async fn test_channel_moving(
|
|||
(&client_b, cx_b),
|
||||
)
|
||||
.await;
|
||||
let channel_b_mu_id = b_channels[0];
|
||||
let channel_b_gamma_id = b_channels[1];
|
||||
let channel_b_epsilon_id = b_channels[2];
|
||||
let channel_mu_id = b_channels[0];
|
||||
let channel_ga_id = b_channels[1];
|
||||
let channel_ep_id = b_channels[2];
|
||||
|
||||
// Current shape for B:
|
||||
// /- ep
|
||||
|
@ -989,13 +994,13 @@ async fn test_channel_moving(
|
|||
client_b.channel_store(),
|
||||
cx_b,
|
||||
&[
|
||||
(channel_b_mu_id, 0),
|
||||
(channel_b_gamma_id, 1),
|
||||
(channel_b_epsilon_id, 1)
|
||||
(channel_mu_id, 0),
|
||||
(channel_ga_id, 1),
|
||||
(channel_ep_id, 1)
|
||||
],
|
||||
);
|
||||
|
||||
client_a.add_admin_to_channel((&client_b, cx_b), channel_a_b_id, cx_a).await;
|
||||
client_a.add_admin_to_channel((&client_b, cx_b), channel_b_id, cx_a).await;
|
||||
// Current shape for B:
|
||||
// /- ep
|
||||
// mu -- ga
|
||||
|
@ -1006,51 +1011,51 @@ async fn test_channel_moving(
|
|||
cx_b,
|
||||
&[
|
||||
// B's old channels
|
||||
(channel_b_mu_id, 0),
|
||||
(channel_b_gamma_id, 1),
|
||||
(channel_b_epsilon_id, 1),
|
||||
(channel_mu_id, 0),
|
||||
(channel_ga_id, 1),
|
||||
(channel_ep_id, 1),
|
||||
|
||||
// New channels from a
|
||||
(channel_a_b_id, 0),
|
||||
(channel_a_c_id, 1),
|
||||
(channel_a_d_id, 1),
|
||||
(channel_a_d_id, 2),
|
||||
(channel_b_id, 0),
|
||||
(channel_c_id, 1),
|
||||
(channel_d_id, 1),
|
||||
(channel_d_id, 2),
|
||||
],
|
||||
);
|
||||
|
||||
// client_b
|
||||
// .channel_store()
|
||||
// .update(cx_a, |channel_store, cx| {
|
||||
// channel_store.move_channel(channel_a_b_id, None, channel_b_epsilon_id, cx)
|
||||
// })
|
||||
// .await
|
||||
// .unwrap();
|
||||
|
||||
// // Current shape for B:
|
||||
// // /---------\
|
||||
// // /- ep -- b -- c -- d
|
||||
// // mu -- ga
|
||||
// assert_channels_list_shape(
|
||||
// client_b.channel_store(),
|
||||
// cx_b,
|
||||
// &[
|
||||
// // B's old channels
|
||||
// (channel_b_mu_id, 0),
|
||||
// (channel_b_gamma_id, 1),
|
||||
// (channel_b_epsilon_id, 1),
|
||||
|
||||
// // New channels from a, now under epsilon
|
||||
// (channel_a_b_id, 2),
|
||||
// (channel_a_c_id, 3),
|
||||
// (channel_a_d_id, 3),
|
||||
// (channel_a_d_id, 4),
|
||||
// ],
|
||||
// );
|
||||
|
||||
client_b
|
||||
.channel_store()
|
||||
.update(cx_a, |channel_store, cx| {
|
||||
channel_store.move_channel(channel_a_b_id, None, channel_b_epsilon_id, cx)
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Current shape for B:
|
||||
// /---------\
|
||||
// /- ep -- b -- c -- d
|
||||
// mu -- ga
|
||||
assert_channels_list_shape(
|
||||
client_b.channel_store(),
|
||||
cx_b,
|
||||
&[
|
||||
// B's old channels
|
||||
(channel_b_mu_id, 0),
|
||||
(channel_b_gamma_id, 1),
|
||||
(channel_b_epsilon_id, 1),
|
||||
|
||||
// New channels from a, now under epsilon
|
||||
(channel_a_b_id, 2),
|
||||
(channel_a_c_id, 3),
|
||||
(channel_a_d_id, 3),
|
||||
(channel_a_d_id, 4),
|
||||
],
|
||||
);
|
||||
|
||||
client_b
|
||||
.channel_store()
|
||||
.update(cx_a, |channel_store, cx| {
|
||||
channel_store.link_channel(channel_b_gamma_id, channel_a_b_id, cx)
|
||||
channel_store.link_channel(channel_ga_id, channel_b_id, cx)
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
@ -1065,16 +1070,16 @@ async fn test_channel_moving(
|
|||
cx_b,
|
||||
&[
|
||||
// B's old channels
|
||||
(channel_b_mu_id, 0),
|
||||
(channel_b_gamma_id, 1),
|
||||
(channel_b_epsilon_id, 1),
|
||||
(channel_mu_id, 0),
|
||||
(channel_ga_id, 1),
|
||||
(channel_ep_id, 1),
|
||||
|
||||
// New channels from a, now under epsilon, with gamma
|
||||
(channel_a_b_id, 2),
|
||||
(channel_b_gamma_id, 3),
|
||||
(channel_a_c_id, 3),
|
||||
(channel_a_d_id, 3),
|
||||
(channel_a_d_id, 4),
|
||||
(channel_b_id, 2),
|
||||
(channel_ga_id, 3),
|
||||
(channel_c_id, 3),
|
||||
(channel_d_id, 3),
|
||||
(channel_d_id, 4),
|
||||
],
|
||||
);
|
||||
|
||||
|
@ -1083,12 +1088,12 @@ async fn test_channel_moving(
|
|||
client_a.channel_store(),
|
||||
cx_a,
|
||||
&[
|
||||
(channel_a_a_id, 0),
|
||||
(channel_a_b_id, 1),
|
||||
(channel_b_gamma_id, 1),
|
||||
(channel_a_c_id, 2),
|
||||
(channel_a_d_id, 3),
|
||||
(channel_a_d_id, 2),
|
||||
(channel_a_id, 0),
|
||||
(channel_b_id, 1),
|
||||
(channel_ga_id, 1),
|
||||
(channel_c_id, 2),
|
||||
(channel_d_id, 3),
|
||||
(channel_d_id, 2),
|
||||
],
|
||||
);
|
||||
// TODO: Make sure to test that non-local root removing problem I was thinking about
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue