laterite_media/
variant.rs

1//! Variants: named, cached derivatives of a media record (a `thumb`, a
2//! `card`), each a row pointing at its own blob path, made on the first
3//! request behind a lock per file and preset so a burst renders once.
4
5use chrono::{SecondsFormat, Utc};
6use futures::TryStreamExt;
7use laterite_core::strata::*;
8use laterite_core::Db;
9use std::collections::HashMap;
10use std::sync::Arc;
11use tokio::sync::Mutex;
12
13use crate::image::{self, ImageError};
14use crate::ingest::blob_path;
15use crate::preset::Preset;
16use crate::record::{MediaRecord, RecordError};
17use crate::{Kind, StorageDriver, StorageError};
18
19#[derive(Iden)]
20pub enum MediaVariants {
21    Table,
22    Id,
23    MediaId,
24    Preset,
25    Path,
26    Width,
27    Height,
28    Size,
29    ContentType,
30    CreatedAt,
31}
32
33/// A variant as stored.
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub struct Variant {
36    pub id: i64,
37    pub media_id: i64,
38    pub preset: String,
39    pub path: String,
40    pub width: Option<i64>,
41    pub height: Option<i64>,
42    pub size: i64,
43    pub content_type: String,
44    pub created_at: String,
45}
46
47#[derive(Debug, thiserror::Error)]
48pub enum VariantError {
49    #[error(transparent)]
50    Record(#[from] RecordError),
51    #[error(transparent)]
52    Storage(#[from] StorageError),
53    #[error(transparent)]
54    Image(#[from] ImageError),
55    #[error("the file is not an image")]
56    NotImage,
57    #[error("the blob could not be read: {0}")]
58    Read(String),
59}
60
61const COLUMNS: [MediaVariants; 9] = [
62    MediaVariants::Id,
63    MediaVariants::MediaId,
64    MediaVariants::Preset,
65    MediaVariants::Path,
66    MediaVariants::Width,
67    MediaVariants::Height,
68    MediaVariants::Size,
69    MediaVariants::ContentType,
70    MediaVariants::CreatedAt,
71];
72
73fn from_row(r: &sqlx::any::AnyRow) -> Result<Variant, sqlx::Error> {
74    Ok(Variant {
75        id: r.get_int("id")?,
76        media_id: r.get_int("media_id")?,
77        preset: r.get_text("preset")?,
78        path: r.get_text("path")?,
79        width: r.get_int_opt("width")?,
80        height: r.get_int_opt("height")?,
81        size: r.get_int("size")?,
82        content_type: r.get_text("content_type")?,
83        created_at: r.get_text("created_at")?,
84    })
85}
86
87/// Where a variant's blob lives: beside the originals, under the hash.
88pub fn variant_path(hash: &str, preset: &str, extension: &str) -> String {
89    format!(
90        "variants/{}/{}/{hash}/{preset}.{extension}",
91        &hash[0..2],
92        &hash[2..4]
93    )
94}
95
96/// The variant of `media_id` in `preset`, if it was made.
97pub async fn find(db: &Db, media_id: i64, preset: &str) -> Result<Option<Variant>, RecordError> {
98    let (sql, values) = build(
99        db.backend,
100        Query::select()
101            .columns(COLUMNS)
102            .from(MediaVariants::Table)
103            .and_where(Expr::col(MediaVariants::MediaId).eq(media_id))
104            .and_where(Expr::col(MediaVariants::Preset).eq(preset))
105            .to_owned(),
106    );
107    let row = bind_values(sqlx::query(&sql), values)
108        .fetch_optional(&db.pool)
109        .await?;
110    Ok(row.as_ref().map(from_row).transpose()?)
111}
112
113/// Every variant made of `media_id`.
114pub async fn list_for(db: &Db, media_id: i64) -> Result<Vec<Variant>, RecordError> {
115    let (sql, values) = build(
116        db.backend,
117        Query::select()
118            .columns(COLUMNS)
119            .from(MediaVariants::Table)
120            .and_where(Expr::col(MediaVariants::MediaId).eq(media_id))
121            .order_by(MediaVariants::Id, Order::Asc)
122            .to_owned(),
123    );
124    let rows = bind_values(sqlx::query(&sql), values)
125        .fetch_all(&db.pool)
126        .await?;
127    Ok(rows.iter().map(from_row).collect::<Result<_, _>>()?)
128}
129
130/// Forgets every variant of `media_id`; the blobs are the caller's to remove.
131pub async fn delete_for(db: &Db, media_id: i64) -> Result<u64, RecordError> {
132    let (sql, values) = build(
133        db.backend,
134        Query::delete()
135            .from_table(MediaVariants::Table)
136            .and_where(Expr::col(MediaVariants::MediaId).eq(media_id))
137            .to_owned(),
138    );
139    Ok(bind_values(sqlx::query(&sql), values)
140        .execute(&db.pool)
141        .await?
142        .rows_affected())
143}
144
145async fn insert(
146    db: &Db,
147    media_id: i64,
148    preset: &str,
149    path: &str,
150    rendered: &image::Rendered,
151) -> Result<Variant, RecordError> {
152    let now = Utc::now().to_rfc3339_opts(SecondsFormat::Micros, true);
153    let stmt = Query::insert()
154        .into_table(MediaVariants::Table)
155        .columns([
156            MediaVariants::MediaId,
157            MediaVariants::Preset,
158            MediaVariants::Path,
159            MediaVariants::Width,
160            MediaVariants::Height,
161            MediaVariants::Size,
162            MediaVariants::ContentType,
163            MediaVariants::CreatedAt,
164        ])
165        .values_panic([
166            media_id.into(),
167            preset.into(),
168            path.into(),
169            i64::from(rendered.width).into(),
170            i64::from(rendered.height).into(),
171            (rendered.bytes.len() as i64).into(),
172            rendered.content_type.into(),
173            now.clone().into(),
174        ])
175        .to_owned();
176    let id = insert_returning_id(db, stmt, MediaVariants::Id).await?;
177    Ok(Variant {
178        id,
179        media_id,
180        preset: preset.to_string(),
181        path: path.to_string(),
182        width: Some(i64::from(rendered.width)),
183        height: Some(i64::from(rendered.height)),
184        size: rendered.bytes.len() as i64,
185        content_type: rendered.content_type.to_string(),
186        created_at: now,
187    })
188}
189
190/// The maker of variants: holds one lock per file and preset, so a burst of
191/// requests for a variant that does not exist yet renders it once.
192#[derive(Default)]
193pub struct Variants {
194    locks: Mutex<HashMap<String, Arc<Mutex<()>>>>,
195}
196
197impl Variants {
198    pub fn new() -> Self {
199        Self::default()
200    }
201
202    /// The variant of `record` in `preset`, made and stored on `disk` when it
203    /// does not exist yet.
204    pub async fn ensure(
205        &self,
206        db: &Db,
207        disk: &dyn StorageDriver,
208        record: &MediaRecord,
209        preset: &Preset,
210    ) -> Result<Variant, VariantError> {
211        if record.kind() != Kind::Image || record.content_type == "image/svg+xml" {
212            return Err(VariantError::NotImage);
213        }
214        if let Some(v) = find(db, record.id, &preset.name).await? {
215            return Ok(v);
216        }
217        let key = format!("{}:{}:{}", record.disk, record.hash, preset.name);
218        let lock = {
219            let mut locks = self.locks.lock().await;
220            Arc::clone(locks.entry(key.clone()).or_default())
221        };
222        let result = {
223            let _held = lock.lock().await;
224            match find(db, record.id, &preset.name).await? {
225                Some(v) => Ok(v),
226                None => make(db, disk, record, preset).await,
227            }
228        };
229        let mut locks = self.locks.lock().await;
230        if Arc::strong_count(&lock) <= 2 {
231            locks.remove(&key);
232        }
233        result
234    }
235}
236
237async fn make(
238    db: &Db,
239    disk: &dyn StorageDriver,
240    record: &MediaRecord,
241    preset: &Preset,
242) -> Result<Variant, VariantError> {
243    let mut stream = disk.get(&blob_path(&record.hash)).await?;
244    let mut bytes: Vec<u8> = Vec::with_capacity(usize::try_from(record.size).unwrap_or(0));
245    while let Some(chunk) = stream
246        .try_next()
247        .await
248        .map_err(|e| VariantError::Read(e.to_string()))?
249    {
250        bytes.extend_from_slice(&chunk);
251    }
252    let orientation = record
253        .meta
254        .get("orientation")
255        .and_then(|o| o.as_u64())
256        .and_then(|o| u8::try_from(o).ok())
257        .unwrap_or(1);
258    let preset_owned = preset.clone();
259    let rendered =
260        tokio::task::spawn_blocking(move || image::render(&bytes, &preset_owned, orientation))
261            .await
262            .map_err(|e| VariantError::Read(e.to_string()))??;
263    let path = variant_path(&record.hash, &preset.name, rendered.extension);
264    disk.put(&path, crate::stream_of(rendered.bytes.clone()))
265        .await?;
266    Ok(insert(db, record.id, &preset.name, &path, &rendered).await?)
267}
laterite_media/variant.rs

laterite_media/
variant.rs
