sofie_storage/
local.rs

1use std::io::ErrorKind;
2use std::path::{Path, PathBuf};
3use std::sync::Arc;
4use std::time::Duration;
5
6use bytes::{Bytes, BytesMut};
7use sofie_auth::SigningKeys;
8use sofie_core::encoding::percent;
9use sofie_core::{Error, Result, Timestamp};
10use tokio::fs::File;
11use tokio::io::{AsyncReadExt, AsyncWriteExt};
12
13use crate::{
14    ContentType, DownloadLink, Pending, Reader, SERVED_PATH, Served, StorageKey, Store, Writer,
15};
16
17const CHUNK: usize = 64 * 1024;
18const EXPIRES: &str = "?expires=";
19const SIGNATURE: &str = "&signature=";
20const NAME: &str = "&name=";
21const PARTIAL: &str = "partial";
22
23pub struct Local {
24    root: Arc<Path>,
25    keys: Arc<SigningKeys>,
26}
27
28impl Local {
29    /// `keys` must be the ones sofie-http checks requests with, so it accepts the links signed here.
30    pub fn new(root: impl Into<PathBuf>, keys: Arc<SigningKeys>) -> Result<Self> {
31        let root = root.into();
32        std::fs::create_dir_all(&root).map_err(|error| {
33            Error::internal(
34                format!("creating the storage root {}", root.display()),
35                error,
36            )
37        })?;
38        Ok(Self {
39            root: root.into(),
40            keys,
41        })
42    }
43
44    fn path(&self, key: &StorageKey) -> PathBuf {
45        self.root.join(key.as_str())
46    }
47}
48
49impl Store for Local {
50    fn writer(&self, key: &StorageKey) -> Pending<'_, Box<dyn Writer>> {
51        let path = self.path(key);
52        Box::pin(async move {
53            // Its own name: a variant's key is derived, and a killed writer's partial file stays.
54            let partial = path.with_extension(format!("{}.{PARTIAL}", StorageKey::generate()?));
55            let file = File::options()
56                .write(true)
57                .create_new(true)
58                .open(&partial)
59                .await
60                .map_err(|error| failed("creating", &partial, error))?;
61            Ok(Box::new(LocalWriter {
62                file: Some(file),
63                partial,
64                path,
65                settled: false,
66            }) as Box<dyn Writer>)
67        })
68    }
69
70    fn reader(&self, key: &StorageKey) -> Pending<'_, Option<Box<dyn Reader>>> {
71        let path = self.path(key);
72        Box::pin(async move {
73            match File::open(&path).await {
74                Ok(file) => {
75                    let size = file
76                        .metadata()
77                        .await
78                        .map_err(|error| failed("reading the size of", &path, error))?
79                        .len();
80                    Ok(Some(Box::new(LocalReader { file, size }) as Box<dyn Reader>))
81                }
82                Err(error) if error.kind() == ErrorKind::NotFound => Ok(None),
83                Err(error) => Err(failed("opening", &path, error)),
84            }
85        })
86    }
87
88    fn delete(&self, key: &StorageKey) -> Pending<'_, ()> {
89        let path = self.path(key);
90        Box::pin(async move {
91            match tokio::fs::remove_file(&path).await {
92                Err(error) if error.kind() != ErrorKind::NotFound => {
93                    Err(failed("deleting", &path, error))
94                }
95                _ => Ok(()),
96            }
97        })
98    }
99
100    fn exists(&self, key: &StorageKey) -> Pending<'_, bool> {
101        let path = self.path(key);
102        Box::pin(async move {
103            tokio::fs::try_exists(&path)
104                .await
105                .map_err(|error| failed("looking for", &path, error))
106        })
107    }
108
109    fn download_url(&self, link: &DownloadLink<'_>) -> Result<String> {
110        let expires = link
111            .now
112            .as_second()
113            .saturating_add(i64::try_from(link.valid_for.as_secs()).unwrap_or(i64::MAX));
114        let mut signed = format!(
115            "{SERVED_PATH}{}/{}{EXPIRES}{expires}",
116            link.key, link.content_type
117        );
118        if let Some(filename) = link.filename {
119            signed.push_str(NAME);
120            percent::push_encoded(&mut signed, filename, percent::is_unreserved);
121        }
122        let signature = self
123            .keys
124            .sign(sofie_auth::Signed::StoredFile, signed.as_bytes());
125        Ok(format!("{signed}{SIGNATURE}{signature}"))
126    }
127
128    fn served(&self, path_and_query: &str, now: Timestamp) -> Option<Served> {
129        let (signed, signature) = path_and_query.split_once(SIGNATURE)?;
130        let (path, query) = signed.split_once(EXPIRES)?;
131        let (expires, filename) = match query.split_once(NAME) {
132            Some((expires, name)) => (expires, Some(query_value(name)?)),
133            None => (query, None),
134        };
135        let expires: i64 = expires.parse().ok()?;
136        let verified =
137            self.keys
138                .verifies(sofie_auth::Signed::StoredFile, signed.as_bytes(), signature);
139        if now.as_second() >= expires || !verified {
140            return None;
141        }
142        let (key, content_type) = path.strip_prefix(SERVED_PATH)?.split_once('/')?;
143        Some(Served {
144            key: StorageKey::parse(key).ok()?,
145            content_type: ContentType::parse(content_type).ok()?,
146            filename,
147            fresh_for: Duration::from_secs(u64::try_from(expires - now.as_second()).ok()?),
148        })
149    }
150}
151
152/// The filename `download_url` wrote; anything else is refused.
153fn query_value(encoded: &str) -> Option<String> {
154    String::from_utf8(percent::decode(encoded.as_bytes())?).ok()
155}
156
157/// Written beside its final name and renamed once complete, so a reader never sees half a file.
158struct LocalWriter {
159    file: Option<File>,
160    partial: PathBuf,
161    path: PathBuf,
162    settled: bool,
163}
164
165impl LocalWriter {
166    fn file(&mut self) -> &mut File {
167        self.file
168            .as_mut()
169            .expect("only finish takes the file, and it consumes the writer")
170    }
171}
172
173impl Writer for LocalWriter {
174    fn write(&mut self, chunk: Bytes) -> Pending<'_, ()> {
175        Box::pin(async move {
176            self.file()
177                .write_all(&chunk)
178                .await
179                .map_err(|error| failed("writing", &self.partial, error))
180        })
181    }
182
183    fn finish(mut self: Box<Self>) -> Pending<'static, ()> {
184        Box::pin(async move {
185            self.file()
186                .sync_all()
187                .await
188                .map_err(|error| failed("syncing", &self.partial, error))?;
189            drop(self.file.take());
190            tokio::fs::rename(&self.partial, &self.path)
191                .await
192                .map_err(|error| failed("renaming", &self.partial, error))?;
193            self.settled = true;
194            Ok(())
195        })
196    }
197
198    fn abort(mut self: Box<Self>) -> Pending<'static, ()> {
199        Box::pin(async move {
200            self.settled = true;
201            drop(self.file.take());
202            tokio::fs::remove_file(&self.partial)
203                .await
204                .map_err(|error| failed("removing", &self.partial, error))
205        })
206    }
207}
208
209/// A writer dropped before it settled, by a failed request or finish, leaves no partial file.
210impl Drop for LocalWriter {
211    fn drop(&mut self) {
212        if self.settled {
213            return;
214        }
215        drop(self.file.take());
216        if let Err(error) = std::fs::remove_file(&self.partial) {
217            tracing::warn!(
218                error.kind = "storage",
219                error = %error,
220                "removing an unfinished file failed"
221            );
222        }
223    }
224}
225
226struct LocalReader {
227    file: File,
228    size: u64,
229}
230
231impl Reader for LocalReader {
232    fn chunk(&mut self) -> Pending<'_, Option<Bytes>> {
233        Box::pin(async move {
234            let mut buffer = BytesMut::with_capacity(CHUNK);
235            let read = self
236                .file
237                .read_buf(&mut buffer)
238                .await
239                .map_err(|error| Error::internal("reading a stored file", error))?;
240            Ok((read > 0).then(|| buffer.freeze()))
241        })
242    }
243
244    fn size(&self) -> Option<u64> {
245        Some(self.size)
246    }
247}
248
249fn failed(doing: &str, path: &Path, error: std::io::Error) -> Error {
250    Error::internal(format!("{doing} {}", path.display()), error)
251}
sofie_storage/local.rs

sofie_storage/
local.rs
