arama_cache/core/
video.rs

1//! `VideoCacheWriter` / `VideoCacheReader` — video-specific cache handles.
2//!
3//! Backed by `localcache` (RFC 002). Same architecture as the image
4//! handles; video adds ffmpeg poster-thumbnail extraction and a second
5//! feature vector (wav2vec2). `None` fields in an upsert request preserve
6//! the stored value (the v1 SQL `COALESCE` semantics).
7
8use std::path::{Path, PathBuf};
9use std::sync::Arc;
10
11use localcache::{CacheEntry, CacheStatus, ConnectionPool, ReadPool};
12use rayon::prelude::*;
13
14use crate::core::engine::{
15    CacheConfig, DbLocation, NAMESPACE_VIDEO, Result, VIDEO_PAYLOAD_VERSION, cache_options,
16    ensure_db_dir, ensure_schema, is_fresh, read_pool_size,
17};
18use crate::core::payload::VideoPayload;
19use crate::core::thumbnail::{generate_video_thumbnail, thumbnail_dest};
20use crate::types::{
21    CacheRead, DirCacheSummary, LookupResult, UpsertVideoRequest, VideoCacheEntry, VideoFeatures,
22};
23
24// ---------------------------------------------------------------------------
25// Config
26// ---------------------------------------------------------------------------
27
28#[derive(Debug, Clone, Default)]
29pub struct VideoCacheConfig {
30    pub cache_config: CacheConfig,
31    /// Path of the ffmpeg executable. `None` skips thumbnail generation.
32    pub ffmpeg_path: Option<PathBuf>,
33}
34
35// ---------------------------------------------------------------------------
36// VideoCacheWriter
37// ---------------------------------------------------------------------------
38
39/// Update handle for video files.
40///
41/// - Generates poster thumbnails with ffmpeg (frame at 5 s, falling back
42///   to 0 s).
43/// - `Clone` only bumps `Arc` counters.
44#[derive(Clone)]
45pub struct VideoCacheWriter {
46    write: ConnectionPool<VideoPayload>,
47    read: ReadPool<VideoPayload>,
48    config: Arc<VideoCacheConfig>,
49}
50
51impl VideoCacheWriter {
52    pub fn as_session(config: VideoCacheConfig) -> Result<Self> {
53        let options = cache_options(&config.cache_config, NAMESPACE_VIDEO, VIDEO_PAYLOAD_VERSION);
54        // Create the parent directory before localcache touches SQLite.
55        ensure_db_dir(&options)?;
56        let write = ConnectionPool::open(options.clone())?;
57        write.with(|e| e.purge_stale_versions())?;
58        let read = ReadPool::open(options, read_pool_size(&config.cache_config))?;
59        Ok(Self {
60            write,
61            read,
62            config: Arc::new(config),
63        })
64    }
65
66    pub fn onetime(
67        location: DbLocation,
68        thumbnail_dir: Option<PathBuf>,
69        ffmpeg_path: Option<PathBuf>,
70    ) -> Result<Self> {
71        Self::as_session(VideoCacheConfig {
72            cache_config: CacheConfig {
73                db_location: location,
74                thumbnail_dir,
75                ..CacheConfig::default()
76            },
77            ffmpeg_path,
78        })
79    }
80
81    // -----------------------------------------------------------------------
82    // Update API
83    // -----------------------------------------------------------------------
84
85    pub fn upsert(&self, req: UpsertVideoRequest) -> Result<()> {
86        let status = self.write.check_status(&req.path)?;
87        let thumbnail = self.ensure_thumbnail(&req.path, &status)?;
88        self.commit(&req, Prepared { status, thumbnail })
89    }
90
91    /// Batch variant of `upsert`. Returns `(PathBuf, Result<()>)` per
92    /// request. Freshness checks and thumbnail extraction run in
93    /// parallel; writes are serialized; fresh nothing-new entries are
94    /// skipped without a write. Individual errors are stored per
95    /// element; other requests continue.
96    pub fn upsert_all(&self, reqs: Vec<UpsertVideoRequest>) -> Vec<(PathBuf, Result<()>)> {
97        let prepared: Vec<(UpsertVideoRequest, Result<Prepared>)> = reqs
98            .into_par_iter()
99            .map(|req| {
100                let prep = (|| {
101                    let status = self.read.check_status(&req.path)?;
102                    let thumbnail = self.ensure_thumbnail(&req.path, &status)?;
103                    Ok(Prepared { status, thumbnail })
104                })();
105                (req, prep)
106            })
107            .collect();
108
109        prepared
110            .into_iter()
111            .map(|(req, prep)| {
112                let path = req.path.clone();
113                let result = prep.and_then(|p| self.commit(&req, p));
114                (path, result)
115            })
116            .collect()
117    }
118
119    pub fn delete(&self, path: &Path) -> Result<bool> {
120        Ok(self.write.remove(path)?)
121    }
122
123    /// Remove every cached entry whose file lives directly in `dir`,
124    /// deleting the associated thumbnail file (when recorded) as well.
125    /// Non-recursive. Returns the number of entries removed (RFC 004).
126    pub fn delete_in_dir(&self, dir: &Path) -> Result<usize> {
127        let entries = self.read.query_run(|q| q.path_in_dir(dir, false))?;
128        let mut removed = 0;
129        for entry in entries {
130            if let Some(thumb) = &entry.payload.thumbnail_path {
131                let _ = std::fs::remove_file(thumb);
132            }
133            if self.write.remove(&entry.path)? {
134                removed += 1;
135            }
136        }
137        Ok(removed)
138    }
139
140    pub fn list_paths(&self) -> Result<Vec<String>> {
141        let keys = self.write.with(|e| e.keys(None))?;
142        Ok(keys
143            .into_iter()
144            .map(|p| p.to_string_lossy().into_owned())
145            .collect())
146    }
147
148    pub fn as_reader(&self) -> VideoCacheReader {
149        VideoCacheReader {
150            read: self.read.clone(),
151        }
152    }
153
154    pub fn lookup(&self, path: &Path) -> Result<LookupResult<VideoCacheEntry>> {
155        self.as_reader().lookup(path)
156    }
157
158    // -----------------------------------------------------------------------
159    // Internal
160    // -----------------------------------------------------------------------
161
162    /// Merge with the existing payload (when fresh) and write. `None`
163    /// vectors in the request preserve stored values, matching the v1
164    /// `COALESCE` update semantics.
165    fn commit(&self, req: &UpsertVideoRequest, prep: Prepared) -> Result<()> {
166        let existing = if is_fresh(&prep.status) {
167            self.write.get(&req.path)?.map(|e| e.payload)
168        } else {
169            None
170        };
171
172        // Steady-state skip: fresh entry, no new vectors, thumbnail
173        // already recorded.
174        if is_fresh(&prep.status)
175            && req.clip_vector.is_none()
176            && req.wav2vec2_vector.is_none()
177            && let Some(p) = &existing
178            && (prep.thumbnail.is_none() || p.thumbnail_path == prep.thumbnail)
179        {
180            return Ok(());
181        }
182
183        let mut payload = existing.unwrap_or_default();
184        if let Some(t) = prep.thumbnail {
185            payload.thumbnail_path = Some(t);
186        }
187        if let Some(v) = &req.clip_vector {
188            payload.clip_vector = Some(v.clone());
189        }
190        if let Some(v) = &req.wav2vec2_vector {
191            payload.wav2vec2_vector = Some(v.clone());
192        }
193        self.write.set(&req.path, &payload)?;
194        Ok(())
195    }
196
197    /// Extract (or reuse) the poster thumbnail for `path`. Skipped when
198    /// either `ffmpeg_path` or `thumbnail_dir` is unset. Regenerated when
199    /// the file changed.
200    fn ensure_thumbnail(&self, path: &Path, status: &CacheStatus) -> Result<Option<String>> {
201        let (Some(ffmpeg), Some(thumb_dir)) = (
202            &self.config.ffmpeg_path,
203            &self.config.cache_config.thumbnail_dir,
204        ) else {
205            return Ok(None);
206        };
207        let dest = thumbnail_dest(thumb_dir, path)?;
208        if !dest.exists() || !is_fresh(status) {
209            generate_video_thumbnail(path, &dest, ffmpeg)?;
210        }
211        Ok(Some(dest.to_string_lossy().into_owned()))
212    }
213}
214
215/// Intermediate result of the parallel preparation phase.
216struct Prepared {
217    status: CacheStatus,
218    thumbnail: Option<String>,
219}
220
221// ---------------------------------------------------------------------------
222// VideoCacheReader
223// ---------------------------------------------------------------------------
224
225/// Read-only handle for video files. `Clone` only bumps `Arc` counters;
226/// clones share the same read pool and may be used from many threads.
227#[derive(Clone)]
228pub struct VideoCacheReader {
229    read: ReadPool<VideoPayload>,
230}
231
232impl VideoCacheReader {
233    pub fn as_session(config: VideoCacheConfig) -> Result<Self> {
234        let options = cache_options(&config.cache_config, NAMESPACE_VIDEO, VIDEO_PAYLOAD_VERSION);
235        // Create the parent directory before localcache touches SQLite.
236        ensure_db_dir(&options)?;
237        ensure_schema::<VideoPayload>(&options)?;
238        let read = ReadPool::open(options, read_pool_size(&config.cache_config))?;
239        Ok(Self { read })
240    }
241
242    pub fn onetime(location: DbLocation) -> Result<Self> {
243        Self::as_session(VideoCacheConfig {
244            cache_config: CacheConfig {
245                db_location: location,
246                ..CacheConfig::default()
247            },
248            ffmpeg_path: None,
249        })
250    }
251
252    pub fn lookup(&self, path: &Path) -> Result<LookupResult<VideoCacheEntry>> {
253        let canonical = match path.canonicalize() {
254            Ok(p) => p,
255            Err(_) => return Ok(LookupResult::Miss),
256        };
257
258        match self.read.check_status(&canonical)? {
259            CacheStatus::Missing => Ok(LookupResult::Miss),
260            CacheStatus::Stale => Ok(LookupResult::Invalidated),
261            CacheStatus::Fresh => match self.read.get(&canonical)? {
262                None => Ok(LookupResult::Miss),
263                Some(entry) => Ok(LookupResult::Hit(to_video_entry(entry))),
264            },
265        }
266    }
267
268    /// Batch variant of `lookup`, parallelized with rayon over the
269    /// read pool's connections.
270    pub fn lookup_all(
271        &self,
272        paths: &[&Path],
273    ) -> Vec<(PathBuf, Result<LookupResult<VideoCacheEntry>>)> {
274        paths
275            .par_iter()
276            .map(|p| (p.to_path_buf(), self.lookup(p)))
277            .collect()
278    }
279
280    pub fn check(&self, path: &Path) -> Result<bool> {
281        Ok(is_fresh(&self.read.check_status(path)?))
282    }
283
284    pub fn list_paths(&self) -> Result<Vec<String>> {
285        Ok(self
286            .read
287            .keys(None)?
288            .into_iter()
289            .map(|p| p.to_string_lossy().into_owned())
290            .collect())
291    }
292
293    pub fn all(&self) -> Result<Vec<Result<VideoCacheEntry>>> {
294        let entries = self.read.query_run(|q| q)?;
295        Ok(entries.into_iter().map(|e| Ok(to_video_entry(e))).collect())
296    }
297
298    /// Group all cached entries by their parent directory and aggregate
299    /// count, total size, and the newest cached-at timestamp (RFC 004).
300    pub fn summarize_by_dir(&self) -> Result<Vec<DirCacheSummary>> {
301        crate::core::image::summarize_entries(self.read.list_entries()?)
302    }
303
304    pub fn all_in_dir(&self, path: &Path) -> Result<Vec<Result<VideoCacheEntry>>> {
305        let dir = dir_of(path);
306        let entries = self.read.query_run(|q| q.path_in_dir(dir, false))?;
307        Ok(entries.into_iter().map(|e| Ok(to_video_entry(e))).collect())
308    }
309
310    pub fn all_in_dir_and_sub_dirs(&self, path: &Path) -> Result<Vec<Result<VideoCacheEntry>>> {
311        let dir = dir_of(path);
312        let entries = self.read.query_run(|q| q.path_in_dir(dir, true))?;
313        Ok(entries.into_iter().map(|e| Ok(to_video_entry(e))).collect())
314    }
315}
316
317impl CacheRead for VideoCacheReader {
318    fn check(&self, path: &Path) -> Result<bool> {
319        VideoCacheReader::check(self, path)
320    }
321
322    fn check_all(&self, paths: &[&Path]) -> Vec<(PathBuf, Result<bool>)> {
323        paths
324            .par_iter()
325            .map(|p| (p.to_path_buf(), self.check(p)))
326            .collect()
327    }
328
329    fn list_paths(&self) -> Result<Vec<String>> {
330        VideoCacheReader::list_paths(self)
331    }
332}
333
334// ---------------------------------------------------------------------------
335// Helpers
336// ---------------------------------------------------------------------------
337
338fn dir_of(path: &Path) -> &Path {
339    if path.is_dir() {
340        path
341    } else {
342        path.parent().unwrap_or(path)
343    }
344}
345
346// ---------------------------------------------------------------------------
347// Mapping
348// ---------------------------------------------------------------------------
349
350fn to_video_entry(entry: CacheEntry<VideoPayload>) -> VideoCacheEntry {
351    let has_features =
352        entry.payload.clip_vector.is_some() || entry.payload.wav2vec2_vector.is_some();
353    VideoCacheEntry {
354        path: entry.path.to_string_lossy().into_owned(),
355        thumbnail_path: entry.payload.thumbnail_path,
356        features: has_features.then_some(VideoFeatures {
357            clip_vector: entry.payload.clip_vector,
358            wav2vec2_vector: entry.payload.wav2vec2_vector,
359        }),
360    }
361}
arama_cache/core/video.rs

arama_cache/core/
video.rs
