ironflow_artifacts/
lib.rs

1//! # ironflow-artifacts
2//!
3//! Blob storage for **ironflow** workflow artifacts: the bytes a step produces
4//! and a later step consumes.
5//!
6//! This crate owns the payloads only. Their metadata (name, MIME type, size,
7//! SHA-256, owning step) lives in `ironflow-store` and is the source of truth:
8//! a blob without a metadata row is never listed and never served.
9//!
10//! # Implementations
11//!
12//! | Store | Feature | Description |
13//! |-------|---------|-------------|
14//! | [`LocalBlobStore`](local::LocalBlobStore) | `artifact-local` (default) | Local filesystem, for development, CI and shared-volume deployments. |
15//!
16//! # Quick start
17//!
18//! ```no_run
19//! use ironflow_artifacts::prelude::*;
20//!
21//! # async fn example() -> Result<(), ArtifactError> {
22//! let store = LocalBlobStore::new("/var/lib/ironflow/artifacts");
23//!
24//! let key = storage_key(uuid::Uuid::now_v7(), uuid::Uuid::now_v7(), uuid::Uuid::now_v7());
25//! let digest = store.put(&key, stream_from_bytes(b"report".to_vec())).await?;
26//!
27//! println!("{} bytes, sha256 {}", digest.size_bytes, digest.sha256);
28//! # Ok(())
29//! # }
30//! ```
31
32pub mod blob_store;
33pub mod error;
34pub mod name;
35
36#[cfg(feature = "artifact-local")]
37pub mod local;
38
39use std::path::Path;
40
41use bytes::Bytes;
42use futures_util::stream::{once, unfold};
43use tokio::fs::File;
44use tokio::io::AsyncReadExt;
45
46use crate::blob_store::ByteStream;
47use crate::error::ArtifactError;
48
49/// Size of the chunks yielded when reading a file.
50const READ_CHUNK_BYTES: usize = 64 * 1024;
51
52/// Wrap an in-memory buffer as a [`ByteStream`].
53///
54/// Convenience for callers that already hold the whole payload -- custom
55/// operations, tests, small generated files. Large payloads should be streamed
56/// from their source instead.
57///
58/// # Examples
59///
60/// ```
61/// use ironflow_artifacts::stream_from_bytes;
62///
63/// let stream = stream_from_bytes(b"hello".to_vec());
64/// # drop(stream);
65/// ```
66pub fn stream_from_bytes(bytes: impl Into<Bytes>) -> ByteStream {
67    let bytes = bytes.into();
68    Box::pin(once(async move { Ok(bytes) }))
69}
70
71/// Read an open file as a [`ByteStream`], in fixed-size chunks.
72///
73/// # Examples
74///
75/// ```no_run
76/// use ironflow_artifacts::stream_from_file;
77/// use tokio::fs::File;
78///
79/// # async fn example() -> Result<(), ironflow_artifacts::error::ArtifactError> {
80/// let file = File::open("report.html").await?;
81/// let stream = stream_from_file(file);
82/// # drop(stream);
83/// # Ok(())
84/// # }
85/// ```
86pub fn stream_from_file(file: File) -> ByteStream {
87    Box::pin(unfold(Some(file), |state| async move {
88        let mut file = state?;
89        let mut buf = vec![0u8; READ_CHUNK_BYTES];
90        match file.read(&mut buf).await {
91            Ok(0) => None,
92            Ok(read) => {
93                buf.truncate(read);
94                Some((Ok(Bytes::from(buf)), Some(file)))
95            }
96            Err(err) => Some((Err(ArtifactError::from(err)), None)),
97        }
98    }))
99}
100
101/// Open a file and read it as a [`ByteStream`].
102///
103/// # Errors
104///
105/// Returns [`ArtifactError::NotFound`] when the path does not exist, and
106/// [`ArtifactError::Io`] on any other filesystem failure.
107///
108/// # Examples
109///
110/// ```no_run
111/// use ironflow_artifacts::stream_from_path;
112///
113/// # async fn example() -> Result<(), ironflow_artifacts::error::ArtifactError> {
114/// let stream = stream_from_path("target/report.html").await?;
115/// # drop(stream);
116/// # Ok(())
117/// # }
118/// ```
119pub async fn stream_from_path(path: impl AsRef<Path>) -> Result<ByteStream, ArtifactError> {
120    let path = path.as_ref();
121    let file = File::open(path).await.map_err(|err| match err.kind() {
122        std::io::ErrorKind::NotFound => ArtifactError::NotFound(path.display().to_string()),
123        _ => ArtifactError::from(err),
124    })?;
125
126    Ok(stream_from_file(file))
127}
128
129/// Convenience re-exports for common usage.
130pub mod prelude {
131    pub use crate::blob_store::{BlobDigest, BlobFuture, BlobStore, ByteStream};
132    pub use crate::error::ArtifactError;
133    pub use crate::name::{
134        MAX_ARTIFACT_NAME_LEN, guess_content_type, storage_key, validate_artifact_name,
135    };
136    pub use crate::{stream_from_bytes, stream_from_file, stream_from_path};
137
138    #[cfg(feature = "artifact-local")]
139    pub use crate::local::{DEFAULT_MAX_ARTIFACT_BYTES, LocalBlobStore};
140}
141
142#[cfg(test)]
143mod tests {
144    use futures_util::TryStreamExt;
145
146    use super::*;
147
148    #[tokio::test]
149    async fn stream_from_bytes_yields_the_whole_buffer() {
150        let chunks: Vec<Bytes> = stream_from_bytes(b"hello".to_vec())
151            .try_collect()
152            .await
153            .expect("collect");
154
155        assert_eq!(chunks.concat(), b"hello");
156    }
157
158    #[tokio::test]
159    async fn stream_from_bytes_supports_an_empty_buffer() {
160        let chunks: Vec<Bytes> = stream_from_bytes(Vec::new())
161            .try_collect()
162            .await
163            .expect("collect");
164
165        assert!(chunks.concat().is_empty());
166    }
167
168    #[tokio::test]
169    async fn stream_from_path_reads_a_file_larger_than_one_chunk() {
170        let dir = tempfile::TempDir::new().expect("temp dir");
171        let path = dir.path().join("big.bin");
172        let payload: Vec<u8> = (0..200_000u32).map(|i| (i % 256) as u8).collect();
173        std::fs::write(&path, &payload).expect("write");
174
175        let chunks: Vec<Bytes> = stream_from_path(&path)
176            .await
177            .expect("open")
178            .try_collect()
179            .await
180            .expect("collect");
181
182        assert_eq!(chunks.concat(), payload);
183        assert!(chunks.len() > 1, "large file should stream in chunks");
184    }
185
186    #[tokio::test]
187    async fn stream_from_path_reports_a_missing_file_as_not_found() {
188        let result = stream_from_path("/nonexistent/ironflow/artifact").await;
189
190        assert!(matches!(result, Err(ArtifactError::NotFound(_))));
191    }
192}
ironflow_artifacts/lib.rs

ironflow_artifacts/
lib.rs
