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

opendal_service_cloudflare_kv/
backend.rs
