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

sofie_storage/
local.rs
