1use super::*;
2use crate::test_both_dbs;
3use language::proto::{self, serialize_version};
4use text::Buffer;
5
6test_both_dbs!(
7 test_channel_buffers,
8 test_channel_buffers_postgres,
9 test_channel_buffers_sqlite
10);
11
12async fn test_channel_buffers(db: &Arc<Database>) {
13 let a_id = db
14 .create_user(
15 "user_a@example.com",
16 None,
17 false,
18 NewUserParams {
19 github_login: "user_a".into(),
20 github_user_id: 101,
21 },
22 )
23 .await
24 .unwrap()
25 .user_id;
26 let b_id = db
27 .create_user(
28 "user_b@example.com",
29 None,
30 false,
31 NewUserParams {
32 github_login: "user_b".into(),
33 github_user_id: 102,
34 },
35 )
36 .await
37 .unwrap()
38 .user_id;
39
40 // This user will not be a part of the channel
41 let c_id = db
42 .create_user(
43 "user_c@example.com",
44 None,
45 false,
46 NewUserParams {
47 github_login: "user_c".into(),
48 github_user_id: 103,
49 },
50 )
51 .await
52 .unwrap()
53 .user_id;
54
55 let owner_id = db.create_server("production").await.unwrap().0 as u32;
56
57 let zed_id = db.create_root_channel("zed", a_id).await.unwrap();
58
59 db.invite_channel_member(zed_id, b_id, a_id, ChannelRole::Member)
60 .await
61 .unwrap();
62
63 db.respond_to_channel_invite(zed_id, b_id, true)
64 .await
65 .unwrap();
66
67 let connection_id_a = ConnectionId { owner_id, id: 1 };
68 let _ = db
69 .join_channel_buffer(zed_id, a_id, connection_id_a)
70 .await
71 .unwrap();
72
73 let mut buffer_a = Buffer::new(0, text::BufferId::new(1).unwrap(), "".to_string());
74 let operations = vec![
75 buffer_a.edit([(0..0, "hello world")]),
76 buffer_a.edit([(5..5, ", cruel")]),
77 buffer_a.edit([(0..5, "goodbye")]),
78 buffer_a.undo().unwrap().1,
79 ];
80 assert_eq!(buffer_a.text(), "hello, cruel world");
81
82 let operations = operations
83 .into_iter()
84 .map(|op| proto::serialize_operation(&language::Operation::Buffer(op)))
85 .collect::<Vec<_>>();
86
87 db.update_channel_buffer(zed_id, a_id, &operations)
88 .await
89 .unwrap();
90
91 let connection_id_b = ConnectionId { owner_id, id: 2 };
92 let buffer_response_b = db
93 .join_channel_buffer(zed_id, b_id, connection_id_b)
94 .await
95 .unwrap();
96
97 let mut buffer_b = Buffer::new(
98 0,
99 text::BufferId::new(1).unwrap(),
100 buffer_response_b.base_text,
101 );
102 buffer_b.apply_ops(buffer_response_b.operations.into_iter().map(|operation| {
103 let operation = proto::deserialize_operation(operation).unwrap();
104 if let language::Operation::Buffer(operation) = operation {
105 operation
106 } else {
107 unreachable!()
108 }
109 }));
110
111 assert_eq!(buffer_b.text(), "hello, cruel world");
112
113 // Ensure that C fails to open the buffer
114 assert!(db
115 .join_channel_buffer(zed_id, c_id, ConnectionId { owner_id, id: 3 })
116 .await
117 .is_err());
118
119 // Ensure that both collaborators have shown up
120 assert_eq!(
121 buffer_response_b.collaborators,
122 &[
123 rpc::proto::Collaborator {
124 user_id: a_id.to_proto(),
125 peer_id: Some(rpc::proto::PeerId { id: 1, owner_id }),
126 replica_id: 0,
127 is_host: false,
128 },
129 rpc::proto::Collaborator {
130 user_id: b_id.to_proto(),
131 peer_id: Some(rpc::proto::PeerId { id: 2, owner_id }),
132 replica_id: 1,
133 is_host: false,
134 }
135 ]
136 );
137
138 // Ensure that get_channel_buffer_collaborators works
139 let zed_collaborats = db.get_channel_buffer_collaborators(zed_id).await.unwrap();
140 assert_eq!(zed_collaborats, &[a_id, b_id]);
141
142 let left_buffer = db
143 .leave_channel_buffer(zed_id, connection_id_b)
144 .await
145 .unwrap();
146
147 assert_eq!(left_buffer.connections, &[connection_id_a],);
148
149 let cargo_id = db.create_root_channel("cargo", a_id).await.unwrap();
150 let _ = db
151 .join_channel_buffer(cargo_id, a_id, connection_id_a)
152 .await
153 .unwrap();
154
155 db.leave_channel_buffers(connection_id_a).await.unwrap();
156
157 let zed_collaborators = db.get_channel_buffer_collaborators(zed_id).await.unwrap();
158 let cargo_collaborators = db.get_channel_buffer_collaborators(cargo_id).await.unwrap();
159 assert_eq!(zed_collaborators, &[]);
160 assert_eq!(cargo_collaborators, &[]);
161
162 // When everyone has left the channel, the operations are collapsed into
163 // a new base text.
164 let buffer_response_b = db
165 .join_channel_buffer(zed_id, b_id, connection_id_b)
166 .await
167 .unwrap();
168 assert_eq!(buffer_response_b.base_text, "hello, cruel world");
169 assert_eq!(buffer_response_b.operations, &[]);
170}
171
172test_both_dbs!(
173 test_channel_buffers_last_operations,
174 test_channel_buffers_last_operations_postgres,
175 test_channel_buffers_last_operations_sqlite
176);
177
178async fn test_channel_buffers_last_operations(db: &Database) {
179 let user_id = db
180 .create_user(
181 "user_a@example.com",
182 None,
183 false,
184 NewUserParams {
185 github_login: "user_a".into(),
186 github_user_id: 101,
187 },
188 )
189 .await
190 .unwrap()
191 .user_id;
192 let observer_id = db
193 .create_user(
194 "user_b@example.com",
195 None,
196 false,
197 NewUserParams {
198 github_login: "user_b".into(),
199 github_user_id: 102,
200 },
201 )
202 .await
203 .unwrap()
204 .user_id;
205 let owner_id = db.create_server("production").await.unwrap().0 as u32;
206 let connection_id = ConnectionId {
207 owner_id,
208 id: user_id.0 as u32,
209 };
210
211 let mut buffers = Vec::new();
212 let mut text_buffers = Vec::new();
213 for i in 0..3 {
214 let channel = db
215 .create_root_channel(&format!("channel-{i}"), user_id)
216 .await
217 .unwrap();
218
219 db.invite_channel_member(channel, observer_id, user_id, ChannelRole::Member)
220 .await
221 .unwrap();
222 db.respond_to_channel_invite(channel, observer_id, true)
223 .await
224 .unwrap();
225
226 db.join_channel_buffer(channel, user_id, connection_id)
227 .await
228 .unwrap();
229
230 buffers.push(
231 db.transaction(|tx| async move { db.get_channel_buffer(channel, &tx).await })
232 .await
233 .unwrap(),
234 );
235
236 text_buffers.push(Buffer::new(
237 0,
238 text::BufferId::new(1).unwrap(),
239 "".to_string(),
240 ));
241 }
242
243 update_buffer(
244 buffers[0].channel_id,
245 user_id,
246 db,
247 vec![
248 text_buffers[0].edit([(0..0, "a")]),
249 text_buffers[0].edit([(0..0, "b")]),
250 text_buffers[0].edit([(0..0, "c")]),
251 ],
252 )
253 .await;
254
255 update_buffer(
256 buffers[1].channel_id,
257 user_id,
258 db,
259 vec![
260 text_buffers[1].edit([(0..0, "d")]),
261 text_buffers[1].edit([(1..1, "e")]),
262 text_buffers[1].edit([(2..2, "f")]),
263 ],
264 )
265 .await;
266
267 // cause buffer 1's epoch to increment.
268 db.leave_channel_buffer(buffers[1].channel_id, connection_id)
269 .await
270 .unwrap();
271 db.join_channel_buffer(buffers[1].channel_id, user_id, connection_id)
272 .await
273 .unwrap();
274 text_buffers[1] = Buffer::new(1, text::BufferId::new(1).unwrap(), "def".to_string());
275 update_buffer(
276 buffers[1].channel_id,
277 user_id,
278 db,
279 vec![
280 text_buffers[1].edit([(0..0, "g")]),
281 text_buffers[1].edit([(0..0, "h")]),
282 ],
283 )
284 .await;
285
286 update_buffer(
287 buffers[2].channel_id,
288 user_id,
289 db,
290 vec![text_buffers[2].edit([(0..0, "i")])],
291 )
292 .await;
293
294 let channels_for_user = db.get_channels_for_user(user_id).await.unwrap();
295
296 pretty_assertions::assert_eq!(
297 channels_for_user.latest_buffer_versions,
298 [
299 rpc::proto::ChannelBufferVersion {
300 channel_id: buffers[0].channel_id.to_proto(),
301 epoch: 0,
302 version: serialize_version(&text_buffers[0].version()),
303 },
304 rpc::proto::ChannelBufferVersion {
305 channel_id: buffers[1].channel_id.to_proto(),
306 epoch: 1,
307 version: serialize_version(&text_buffers[1].version())
308 .into_iter()
309 .filter(|vector| vector.replica_id == text_buffers[1].replica_id() as u32)
310 .collect::<Vec<_>>(),
311 },
312 rpc::proto::ChannelBufferVersion {
313 channel_id: buffers[2].channel_id.to_proto(),
314 epoch: 0,
315 version: serialize_version(&text_buffers[2].version()),
316 },
317 ]
318 );
319}
320
321async fn update_buffer(
322 channel_id: ChannelId,
323 user_id: UserId,
324 db: &Database,
325 operations: Vec<text::Operation>,
326) {
327 let operations = operations
328 .into_iter()
329 .map(|op| proto::serialize_operation(&language::Operation::Buffer(op)))
330 .collect::<Vec<_>>();
331 db.update_channel_buffer(channel_id, user_id, &operations)
332 .await
333 .unwrap();
334}