opendal_service_cloudflare_kv/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  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,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::fmt::Debug;
19use std::sync::Arc;
20
21use bytes::Buf;
22use http::StatusCode;
23use opendal_core::raw::*;
24use opendal_core::*;
25
26use super::CLOUDFLARE_KV_SCHEME;
27use super::config::CloudflareKvConfig;
28use super::core::CloudflareKvCore;
29use super::core::parse_error;
30use super::deleter::CloudflareKvDeleter;
31use super::lister::CloudflareKvLister;
32use super::model::*;
33use super::reader::*;
34use super::writer::CloudflareWriter;
35
36#[doc = include_str!("docs.md")]
37#[derive(Default)]
38pub struct CloudflareKvBuilder {
39    pub(super) config: CloudflareKvConfig,
40}
41
42impl Debug for CloudflareKvBuilder {
43    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
44        f.debug_struct("CloudflareKvBuilder")
45            .field("config", &self.config)
46            .finish_non_exhaustive()
47    }
48}
49
50impl CloudflareKvBuilder {
51    /// Set the token used to authenticate with CloudFlare.
52    pub fn api_token(mut self, api_token: &str) -> Self {
53        if !api_token.is_empty() {
54            self.config.api_token = Some(api_token.to_string())
55        }
56        self
57    }
58
59    /// Set the account ID used to authenticate with CloudFlare.
60    pub fn account_id(mut self, account_id: &str) -> Self {
61        if !account_id.is_empty() {
62            self.config.account_id = Some(account_id.to_string())
63        }
64        self
65    }
66
67    /// Set the namespace ID.
68    pub fn namespace_id(mut self, namespace_id: &str) -> Self {
69        if !namespace_id.is_empty() {
70            self.config.namespace_id = Some(namespace_id.to_string())
71        }
72        self
73    }
74
75    /// Set the default ttl for cloudflare_kv services.
76    ///
77    /// If set, we will specify `EX` for write operations.
78    pub fn default_ttl(mut self, ttl: Duration) -> Self {
79        self.config.default_ttl = Some(ttl);
80        self
81    }
82
83    /// Set the root within this backend.
84    pub fn root(mut self, root: &str) -> Self {
85        self.config.root = if root.is_empty() {
86            None
87        } else {
88            Some(root.to_string())
89        };
90
91        self
92    }
93}
94
95impl Builder for CloudflareKvBuilder {
96    type Config = CloudflareKvConfig;
97
98    fn build(self) -> Result<impl Service> {
99        let api_token = match &self.config.api_token {
100            Some(api_token) => format_authorization_by_bearer(api_token)?,
101            None => {
102                return Err(Error::new(
103                    ErrorKind::ConfigInvalid,
104                    "api_token is required",
105                ));
106            }
107        };
108
109        let Some(account_id) = self.config.account_id.clone() else {
110            return Err(Error::new(
111                ErrorKind::ConfigInvalid,
112                "account_id is required",
113            ));
114        };
115
116        let Some(namespace_id) = self.config.namespace_id.clone() else {
117            return Err(Error::new(
118                ErrorKind::ConfigInvalid,
119                "namespace_id is required",
120            ));
121        };
122
123        // Validate default TTL is at least 60 seconds if specified
124        if let Some(ttl) = self.config.default_ttl
125            && ttl < Duration::from_secs(60)
126        {
127            return Err(Error::new(
128                ErrorKind::ConfigInvalid,
129                "Default TTL must be at least 60 seconds",
130            ));
131        }
132
133        let root = normalize_root(
134            self.config
135                .root
136                .clone()
137                .unwrap_or_else(|| "/".to_string())
138                .as_str(),
139        );
140
141        Ok(CloudflareKvBackend {
142            core: Arc::new(CloudflareKvCore {
143                api_token,
144                account_id,
145                namespace_id,
146                expiration_ttl: self.config.default_ttl,
147                info: ServiceInfo::new(CLOUDFLARE_KV_SCHEME, &root, ""),
148                capability: Capability {
149                    create_dir: true,
150
151                    stat: true,
152                    stat_with_if_match: true,
153                    stat_with_if_none_match: true,
154                    stat_with_if_modified_since: true,
155                    stat_with_if_unmodified_since: true,
156
157                    read: true,
158                    read_with_if_match: true,
159                    read_with_if_none_match: true,
160                    read_with_if_modified_since: true,
161                    read_with_if_unmodified_since: true,
162
163                    write: true,
164                    write_can_empty: true,
165                    write_total_max_size: Some(25 * 1024 * 1024),
166
167                    list: true,
168                    list_with_limit: true,
169                    list_with_recursive: true,
170
171                    delete: true,
172                    delete_max_size: Some(10000),
173
174                    ..Default::default()
175                },
176            }),
177        })
178    }
179}
180
181#[derive(Debug, Clone)]
182pub struct CloudflareKvBackend {
183    pub(crate) core: Arc<CloudflareKvCore>,
184}
185
186impl Service for CloudflareKvBackend {
187    type Reader = oio::StreamReader<CloudflareKvReader>;
188    type Writer = oio::OneShotWriter<CloudflareWriter>;
189    type Lister = oio::PageLister<CloudflareKvLister>;
190    type Deleter = oio::BatchDeleter<CloudflareKvDeleter>;
191    type Copier = ();
192
193    fn info(&self) -> ServiceInfo {
194        self.core.info.clone()
195    }
196
197    fn capability(&self) -> Capability {
198        self.core.capability
199    }
200
201    async fn create_dir(
202        &self,
203        ctx: &OperationContext,
204        path: &str,
205        _args: OpCreateDir,
206    ) -> Result<RpCreateDir> {
207        let path = build_abs_path(&self.core.info.root(), path);
208
209        if path == build_abs_path(&self.core.info.root(), "") {
210            return Ok(RpCreateDir::default());
211        }
212
213        // Split path into segments and create directories for each level
214        let segments: Vec<&str> = path
215            .trim_start_matches('/')
216            .trim_end_matches('/')
217            .split('/')
218            .collect();
219
220        // Create each directory level
221        let mut current_path = String::from("/");
222        for segment in segments {
223            // Build the current directory path
224            if !current_path.ends_with('/') {
225                current_path.push('/');
226            }
227            current_path.push_str(segment);
228            current_path.push('/');
229
230            // Create metadata for current directory
231            let cf_kv_metadata = CfKvMetadata {
232                etag: build_tmp_path_of(&current_path),
233                last_modified: Timestamp::now().to_string(),
234                content_length: 0,
235                is_dir: true,
236            };
237
238            // Set the directory entry
239            self.core
240                .set(ctx, &current_path, Buffer::new(), cf_kv_metadata)
241                .await?;
242        }
243
244        Ok(RpCreateDir::default())
245    }
246
247    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
248        let path = build_abs_path(&self.core.info.root(), path);
249        let new_path = path.trim_end_matches('/');
250
251        let resp = self.core.metadata(ctx, new_path).await?;
252
253        // Handle non-OK response
254        if resp.status() != StatusCode::OK {
255            // Special handling for potential directory paths
256            if path.ends_with('/') && resp.status() == StatusCode::NOT_FOUND {
257                // Try listing the path to check if it's a directory
258                let list_resp = self.core.list(ctx, &path, None, None).await?;
259
260                if list_resp.status() == StatusCode::OK {
261                    let list_body = list_resp.into_body();
262                    let list_result: CfKvListResponse = serde_json::from_reader(list_body.reader())
263                        .map_err(new_json_deserialize_error)?;
264
265                    // If listing returns results, treat as directory
266                    if let Some(entries) = list_result.result
267                        && !entries.is_empty()
268                    {
269                        return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
270                    }
271
272                    // Empty or no results means not found
273                    return Err(Error::new(
274                        ErrorKind::NotFound,
275                        "key not found in CloudFlare KV",
276                    ));
277                }
278            }
279
280            // For all other error cases, parse the error response
281            return Err(parse_error(resp));
282        }
283
284        let resp_body = resp.into_body();
285        let cf_response: CfKvStatResponse =
286            serde_json::from_reader(resp_body.reader()).map_err(new_json_deserialize_error)?;
287
288        if !cf_response.success {
289            return Err(Error::new(
290                ErrorKind::Unexpected,
291                "cloudflare_kv stat this key failed for reason we don't know",
292            ));
293        }
294
295        let metadata = match cf_response.result {
296            Some(metadata) => {
297                if path.ends_with('/') && !metadata.is_dir {
298                    return Err(Error::new(
299                        ErrorKind::NotFound,
300                        "key not found in CloudFlare KV",
301                    ));
302                } else {
303                    metadata
304                }
305            }
306            None => {
307                return Err(Error::new(
308                    ErrorKind::NotFound,
309                    "key not found in CloudFlare KV",
310                ));
311            }
312        };
313
314        // Check if_match condition
315        if let Some(if_match) = &args.if_match()
316            && if_match != &metadata.etag
317        {
318            return Err(Error::new(ErrorKind::ConditionNotMatch, "etag mismatch"));
319        }
320
321        // Check if_none_match condition
322        if let Some(if_none_match) = &args.if_none_match()
323            && if_none_match == &metadata.etag
324        {
325            return Err(Error::new(
326                ErrorKind::ConditionNotMatch,
327                "etag match when expected none match",
328            ));
329        }
330
331        // Parse since time once for both time-based conditions
332        let last_modified = metadata
333            .last_modified
334            .parse::<Timestamp>()
335            .map_err(|_| Error::new(ErrorKind::Unsupported, "invalid since format"))?;
336
337        // Check modified_since condition
338        if let Some(modified_since) = &args.if_modified_since()
339            && !last_modified.gt(modified_since)
340        {
341            return Err(Error::new(
342                ErrorKind::ConditionNotMatch,
343                "not modified since specified time",
344            ));
345        }
346
347        // Check unmodified_since condition
348        if let Some(unmodified_since) = &args.if_unmodified_since()
349            && !last_modified.le(unmodified_since)
350        {
351            return Err(Error::new(
352                ErrorKind::ConditionNotMatch,
353                "modified since specified time",
354            ));
355        }
356
357        let meta = Metadata::new(if metadata.is_dir {
358            EntryMode::DIR
359        } else {
360            EntryMode::FILE
361        })
362        .with_etag(metadata.etag)
363        .with_content_length(metadata.content_length as u64)
364        .with_last_modified(metadata.last_modified.parse::<Timestamp>()?);
365
366        Ok(RpStat::new(meta))
367    }
368    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
369        let output: oio::StreamReader<CloudflareKvReader> = {
370            Ok(oio::StreamReader::new(CloudflareKvReader::new(
371                self.clone(),
372                ctx.clone(),
373                path,
374                args,
375            )))
376        }?;
377
378        Ok(output)
379    }
380
381    fn write(&self, ctx: &OperationContext, path: &str, _: OpWrite) -> Result<Self::Writer> {
382        let output: oio::OneShotWriter<CloudflareWriter> = {
383            let path = build_abs_path(&self.core.info.root(), path);
384            let writer = CloudflareWriter::new(self.core.clone(), ctx.clone(), path);
385
386            let w = oio::OneShotWriter::new(writer);
387
388            Ok(w)
389        }?;
390
391        Ok(output)
392    }
393
394    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
395        let output: oio::BatchDeleter<CloudflareKvDeleter> = {
396            Ok(oio::BatchDeleter::new(
397                CloudflareKvDeleter::new(self.core.clone(), ctx.clone()),
398                self.core.capability.delete_max_size,
399            ))
400        }?;
401
402        Ok(output)
403    }
404
405    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
406        let output: oio::PageLister<CloudflareKvLister> = {
407            let path = build_abs_path(&self.core.info.root(), path);
408
409            let limit = match args.limit() {
410                Some(limit) => {
411                    // The list limit of cloudflare_kv is limited to 10..1000.
412                    if !(10..=1000).contains(&limit) {
413                        1000
414                    } else {
415                        limit
416                    }
417                }
418                None => 1000,
419            };
420
421            let l = CloudflareKvLister::new(
422                self.core.clone(),
423                ctx.clone(),
424                &path,
425                args.recursive(),
426                Some(limit),
427            );
428
429            Ok(oio::PageLister::new(l))
430        }?;
431
432        Ok(output)
433    }
434
435    fn copy(
436        &self,
437        _ctx: &OperationContext,
438        _from: &str,
439        _to: &str,
440        _args: OpCopy,
441        _opts: OpCopier,
442    ) -> Result<Self::Copier> {
443        Err(Error::new(
444            ErrorKind::Unsupported,
445            "operation is not supported",
446        ))
447    }
448
449    async fn rename(
450        &self,
451        _ctx: &OperationContext,
452        _from: &str,
453        _to: &str,
454        _args: OpRename,
455    ) -> Result<RpRename> {
456        Err(Error::new(
457            ErrorKind::Unsupported,
458            "operation is not supported",
459        ))
460    }
461
462    async fn presign(
463        &self,
464        _ctx: &OperationContext,
465        _path: &str,
466        _args: OpPresign,
467    ) -> Result<RpPresign> {
468        Err(Error::new(
469            ErrorKind::Unsupported,
470            "operation is not supported",
471        ))
472    }
473}
opendal_service_cloudflare_kv/backend.rs

opendal_service_cloudflare_kv/
backend.rs
