hiqlite 0.13.0

Hiqlite - highly-available, embeddable, raft-based SQLite + cache
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

use crate::Error;
pub use cryptr::stream::s3::*;
use cryptr::stream::writer::channel_writer::{ChannelReceiver, ChannelWriter};
pub use cryptr::EncKeys;
use cryptr::{EncValue, FileReader, FileWriter, S3Reader, S3Writer, StreamReader, StreamWriter};
use std::env;
use std::sync::Arc;
use tokio::task;

#[derive(Debug, Clone)]
pub struct S3Config {
    pub bucket: Bucket,
}

impl S3Config {
    pub fn new<S>(
        endpoint: &str,
        bucket_name: S,
        region: S,
        key: S,
        secret: S,
        path_style: bool,
    ) -> Result<Arc<Self>, Error>
    where
        S: Into<String>,
    {
        let endpoint = reqwest::Url::parse(endpoint).map_err(|err| Error::S3(err.to_string()))?;
        let region = Region(region.into());
        let credentials = Credentials {
            access_key_id: AccessKeyId(key.into()),
            access_key_secret: AccessKeySecret(secret.into()),
        };
        let options = Some(BucketOptions {
            path_style,
            list_objects_v2: true,
        });
        let bucket = Bucket::new(endpoint, bucket_name.into(), region, credentials, options)
            .map_err(|err| Error::S3(err.to_string()))?;

        // TODO try to list bucket and make sure access creds work fine

        Ok(Arc::new(Self { bucket }))
    }

    pub fn try_from_env() -> Option<Arc<Self>> {
        if let Ok(url) = env::var("HQL_S3_URL") {
            // we assume that all values exist when we can read the url successfully

            let url = reqwest::Url::parse(&url).expect("Cannot parse HQL_S3_URL as URL");
            let bucket_name = env::var("HQL_S3_BUCKET").expect("HQL_S3_BUCKET not found");
            let region = Region(env::var("HQL_S3_REGION").expect("HQL_S3_REGION not found"));
            let path_style = env::var("HQL_S3_PATH_STYLE")
                .expect("HQL_S3_PATH_STYLE not found")
                .parse::<bool>()
                .expect("Cannot parse HQL_S3_PATH_STYLE as bool");

            let access_key_id = AccessKeyId(env::var("HQL_S3_KEY").expect("HQL_S3_KEY not found"));
            let access_key_secret =
                AccessKeySecret(env::var("HQL_S3_SECRET").expect("HQL_S3_SECRET not found"));
            let credentials = Credentials {
                access_key_id,
                access_key_secret,
            };

            let options = Some(BucketOptions {
                path_style,
                list_objects_v2: true,
            });

            let bucket = Bucket::new(url, bucket_name, region, credentials, options).unwrap();

            Some(Arc::new(S3Config { bucket }))
        } else {
            None
        }
    }

    pub(crate) async fn push(&self, path: &str, object: &str) -> Result<(), Error> {
        let reader = StreamReader::File(FileReader {
            path,
            print_progress: false,
        });
        let writer = StreamWriter::S3(S3Writer {
            bucket: &self.bucket,
            object,
        });

        EncValue::encrypt_stream(reader, writer)
            .await
            .map_err(|err| Error::S3(err.to_string()))
    }

    pub(crate) async fn pull(&self, object: &str, path: &str) -> Result<(), Error> {
        let reader = StreamReader::S3(S3Reader {
            bucket: &self.bucket,
            object,
            print_progress: false,
        });
        let writer = StreamWriter::File(FileWriter {
            path,
            overwrite_target: true,
        });

        EncValue::decrypt_stream(reader, writer)
            .await
            .map_err(|err| Error::S3(err.to_string()))
    }

    pub(crate) fn pull_channel(&self, object: String) -> Result<ChannelReceiver, Error> {
        let (channel_writer, rx) = ChannelWriter::new();
        let writer = StreamWriter::Channel(channel_writer.clone());

        let bucket = self.bucket.clone();
        task::spawn(async move {
            let reader = StreamReader::S3(S3Reader {
                bucket: &bucket,
                object: &object,
                print_progress: false,
            });

            if let Err(err) = EncValue::decrypt_stream(reader, writer).await {
                channel_writer.err(Some(err)).await;
            }
        });

        Ok(rx)
    }
}