couchbase_core/mgmtx/
mgmt_collection.rs

1/*
2 *
3 *  * Copyright (c) 2025 Couchbase, Inc.
4 *  *
5 *  * Licensed under the Apache License, Version 2.0 (the "License");
6 *  * you may not use this file except in compliance with the License.
7 *  * You may obtain a copy of the License at
8 *  *
9 *  *    http://www.apache.org/licenses/LICENSE-2.0
10 *  *
11 *  * Unless required by applicable law or agreed to in writing, software
12 *  * distributed under the License is distributed on an "AS IS" BASIS,
13 *  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 *  * See the License for the specific language governing permissions and
15 *  * limitations under the License.
16 *
17 */
18
19use crate::cbconfig::CollectionManifest;
20use crate::httpx::client::Client;
21use crate::mgmtx::error;
22use crate::mgmtx::mgmt::{parse_response_json, Management};
23use crate::mgmtx::options::{
24    CreateCollectionOptions, CreateScopeOptions, DeleteCollectionOptions, DeleteScopeOptions,
25    GetCollectionManifestOptions, UpdateCollectionOptions,
26};
27use crate::mgmtx::responses::{
28    CreateCollectionResponse, CreateScopeResponse, DeleteCollectionResponse, DeleteScopeResponse,
29    UpdateCollectionResponse,
30};
31use crate::tracingcomponent::{BeginDispatchFields, EndDispatchFields};
32use crate::util::get_host_port_tuple_from_uri;
33use bytes::Bytes;
34use http::Method;
35use serde::Deserialize;
36
37impl<C: Client> Management<C> {
38    pub async fn get_collection_manifest(
39        &self,
40        opts: &GetCollectionManifestOptions<'_>,
41    ) -> error::Result<CollectionManifest> {
42        let method = Method::GET;
43        let path = format!("pools/default/buckets/{}/scopes", opts.bucket_name).to_string();
44
45        let resp = self
46            .execute(
47                method.clone(),
48                &path,
49                "",
50                opts.on_behalf_of_info.cloned(),
51                None,
52                None,
53            )
54            .await
55            .map_err(|e| {
56                error::Error::new_message_error(format!("could not get collections manifest: {e}"))
57            })?;
58
59        if resp.status() != 200 {
60            return Err(
61                Self::decode_common_error(method, path, "get_collection_manifest", resp).await,
62            );
63        }
64
65        parse_response_json(resp).await
66    }
67
68    pub async fn create_scope(
69        &self,
70        opts: &CreateScopeOptions<'_>,
71    ) -> error::Result<CreateScopeResponse> {
72        let method = Method::POST;
73        let path = format!("pools/default/buckets/{}/scopes", opts.bucket_name).to_string();
74
75        let body = url::form_urlencoded::Serializer::new(String::new())
76            .append_pair("name", opts.scope_name)
77            .finish();
78
79        let peer_addr = get_host_port_tuple_from_uri(&self.endpoint).unwrap_or_default();
80        let canonical_addr =
81            get_host_port_tuple_from_uri(&self.canonical_endpoint).unwrap_or_default();
82        let resp = self
83            .tracing
84            .orchestrate_dispatch_span(
85                BeginDispatchFields::new(
86                    (&peer_addr.0, &peer_addr.1),
87                    (&canonical_addr.0, &canonical_addr.1),
88                    None,
89                ),
90                self.execute(
91                    method.clone(),
92                    &path,
93                    "application/x-www-form-urlencoded",
94                    opts.on_behalf_of_info.cloned(),
95                    None,
96                    Some(Bytes::from(body)),
97                ),
98                |_| EndDispatchFields::new(None, None),
99            )
100            .await?;
101
102        if resp.status() != 200 {
103            return Err(Self::decode_common_error(method, path, "create_scope", resp).await);
104        }
105
106        let manifest_uid: ManifestUidJson = parse_response_json(resp).await?;
107
108        Ok(CreateScopeResponse {
109            manifest_uid: manifest_uid.manifest_uid,
110        })
111    }
112
113    pub async fn delete_scope(
114        &self,
115        opts: &DeleteScopeOptions<'_>,
116    ) -> error::Result<DeleteScopeResponse> {
117        let method = Method::DELETE;
118        let path = format!(
119            "pools/default/buckets/{}/scopes/{}",
120            opts.bucket_name, opts.scope_name
121        )
122        .to_string();
123
124        let peer_addr = get_host_port_tuple_from_uri(&self.endpoint).unwrap_or_default();
125        let canonical_addr =
126            get_host_port_tuple_from_uri(&self.canonical_endpoint).unwrap_or_default();
127        let resp = self
128            .tracing
129            .orchestrate_dispatch_span(
130                BeginDispatchFields::new(
131                    (&peer_addr.0, &peer_addr.1),
132                    (&canonical_addr.0, &canonical_addr.1),
133                    None,
134                ),
135                self.execute(
136                    method.clone(),
137                    &path,
138                    "",
139                    opts.on_behalf_of_info.cloned(),
140                    None,
141                    None,
142                ),
143                |_| EndDispatchFields::new(None, None),
144            )
145            .await?;
146
147        if resp.status() != 200 {
148            return Err(Self::decode_common_error(method, path, "delete_scope", resp).await);
149        }
150
151        let manifest_uid: ManifestUidJson = parse_response_json(resp).await?;
152
153        Ok(DeleteScopeResponse {
154            manifest_uid: manifest_uid.manifest_uid,
155        })
156    }
157
158    pub async fn create_collection(
159        &self,
160        opts: &CreateCollectionOptions<'_>,
161    ) -> error::Result<CreateCollectionResponse> {
162        let method = Method::POST;
163        let path = format!(
164            "pools/default/buckets/{}/scopes/{}/collections",
165            opts.bucket_name, opts.scope_name
166        )
167        .to_string();
168
169        let body = {
170            // Serializer is not Send so we need to drop it before making the request.
171            let mut form = url::form_urlencoded::Serializer::new(String::new());
172            form.append_pair("name", opts.collection_name);
173
174            let max_ttl = opts.max_ttl.map(|m| m.to_string());
175            let max_ttl = max_ttl.as_deref();
176            let history = opts.history_enabled.map(|h| h.to_string());
177            let history = history.as_deref();
178            if let Some(max_ttl) = max_ttl {
179                form.append_pair("maxTTL", max_ttl);
180            }
181            if let Some(history) = history {
182                form.append_pair("history", history);
183            }
184
185            Bytes::from(form.finish())
186        };
187
188        let peer_addr = get_host_port_tuple_from_uri(&self.endpoint).unwrap_or_default();
189        let canonical_addr =
190            get_host_port_tuple_from_uri(&self.canonical_endpoint).unwrap_or_default();
191        let resp = self
192            .tracing
193            .orchestrate_dispatch_span(
194                BeginDispatchFields::new(
195                    (&peer_addr.0, &peer_addr.1),
196                    (&canonical_addr.0, &canonical_addr.1),
197                    None,
198                ),
199                self.execute(
200                    method.clone(),
201                    &path,
202                    "application/x-www-form-urlencoded",
203                    opts.on_behalf_of_info.cloned(),
204                    None,
205                    Some(body),
206                ),
207                |_| EndDispatchFields::new(None, None),
208            )
209            .await?;
210
211        if resp.status() != 200 {
212            return Err(Self::decode_common_error(method, path, "create_collection", resp).await);
213        }
214
215        let manifest_uid: ManifestUidJson = parse_response_json(resp).await?;
216
217        Ok(CreateCollectionResponse {
218            manifest_uid: manifest_uid.manifest_uid,
219        })
220    }
221
222    pub async fn update_collection(
223        &self,
224        opts: &UpdateCollectionOptions<'_>,
225    ) -> error::Result<UpdateCollectionResponse> {
226        let method = Method::PATCH;
227        let path = format!(
228            "pools/default/buckets/{}/scopes/{}/collections/{}",
229            opts.bucket_name, opts.scope_name, opts.collection_name
230        )
231        .to_string();
232
233        let body = {
234            // Serializer is not Send so we need to drop it before making the request.
235            let mut form = url::form_urlencoded::Serializer::new(String::new());
236
237            let max_ttl = opts.max_ttl.map(|m| m.to_string());
238            let max_ttl = max_ttl.as_deref();
239            let history = opts.history_enabled.map(|h| h.to_string());
240            let history = history.as_deref();
241            if let Some(max_ttl) = max_ttl {
242                form.append_pair("maxTTL", max_ttl);
243            }
244            if let Some(history) = history {
245                form.append_pair("history", history);
246            }
247
248            Bytes::from(form.finish())
249        };
250
251        let peer_addr = get_host_port_tuple_from_uri(&self.endpoint).unwrap_or_default();
252        let canonical_addr =
253            get_host_port_tuple_from_uri(&self.canonical_endpoint).unwrap_or_default();
254        let resp = self
255            .tracing
256            .orchestrate_dispatch_span(
257                BeginDispatchFields::new(
258                    (&peer_addr.0, &peer_addr.1),
259                    (&canonical_addr.0, &canonical_addr.1),
260                    None,
261                ),
262                self.execute(
263                    method.clone(),
264                    &path,
265                    "application/x-www-form-urlencoded",
266                    opts.on_behalf_of_info.cloned(),
267                    None,
268                    Some(body),
269                ),
270                |_| EndDispatchFields::new(None, None),
271            )
272            .await?;
273
274        if resp.status() != 200 {
275            return Err(Self::decode_common_error(method, path, "update_collection", resp).await);
276        }
277
278        let manifest_uid: ManifestUidJson = parse_response_json(resp).await?;
279
280        Ok(UpdateCollectionResponse {
281            manifest_uid: manifest_uid.manifest_uid,
282        })
283    }
284
285    pub async fn delete_collection(
286        &self,
287        opts: &DeleteCollectionOptions<'_>,
288    ) -> error::Result<DeleteCollectionResponse> {
289        let method = Method::DELETE;
290        let path = format!(
291            "pools/default/buckets/{}/scopes/{}/collections/{}",
292            opts.bucket_name, opts.scope_name, opts.collection_name
293        )
294        .to_string();
295
296        let peer_addr = get_host_port_tuple_from_uri(&self.endpoint).unwrap_or_default();
297        let canonical_addr =
298            get_host_port_tuple_from_uri(&self.canonical_endpoint).unwrap_or_default();
299        let resp = self
300            .tracing
301            .orchestrate_dispatch_span(
302                BeginDispatchFields::new(
303                    (&peer_addr.0, &peer_addr.1),
304                    (&canonical_addr.0, &canonical_addr.1),
305                    None,
306                ),
307                self.execute(
308                    method.clone(),
309                    &path,
310                    "",
311                    opts.on_behalf_of_info.cloned(),
312                    None,
313                    None,
314                ),
315                |_| EndDispatchFields::new(None, None),
316            )
317            .await?;
318
319        if resp.status() != 200 {
320            return Err(Self::decode_common_error(method, path, "delete_collection", resp).await);
321        }
322
323        let manifest_uid: ManifestUidJson = parse_response_json(resp).await?;
324
325        Ok(DeleteCollectionResponse {
326            manifest_uid: manifest_uid.manifest_uid,
327        })
328    }
329}
330
331#[derive(Deserialize)]
332struct ManifestUidJson {
333    #[serde(rename = "uid")]
334    pub manifest_uid: String,
335}
couchbase_core/mgmtx/mgmt_collection.rs

couchbase_core/mgmtx/
mgmt_collection.rs
