jellytau_lib/download/
cache.rs

1//! Smart caching engine for predictive downloads
2
3#[cfg(test)]
4use crate::utils::lock::MutexSafe;
5use log::{debug, info};
6use std::collections::HashMap;
7use std::sync::{Arc, Mutex};
8
9use serde::{Deserialize, Serialize};
10
11use crate::storage::db_service::{DatabaseService, Query, QueryParam};
12
13/// Smart caching configuration
14#[derive(specta::Type, Debug, Clone, Serialize, Deserialize)]
15#[serde(rename_all = "camelCase")]
16pub struct CacheConfig {
17    /// Enable queue pre-caching
18    pub queue_precache_enabled: bool,
19    /// Number of tracks to pre-cache from queue
20    pub queue_precache_count: usize,
21    /// Enable album affinity detection
22    pub album_affinity_enabled: bool,
23    /// Threshold for album affinity (tracks played before caching)
24    pub album_affinity_threshold: usize,
25    /// Storage limit in bytes (0 = unlimited)
26    pub storage_limit: u64,
27    /// Only cache on WiFi
28    pub wifi_only: bool,
29    /// How long a temporary (`download_source = 'auto'`) download lives before
30    /// it is reclaimed, in hours. 0 disables expiry, leaving space pressure as
31    /// the only reclaim trigger.
32    ///
33    /// TRACES: UR-071 | DR-127
34    pub temporary_ttl_hours: u64,
35}
36
37impl Default for CacheConfig {
38    fn default() -> Self {
39        Self {
40            queue_precache_enabled: true,
41            queue_precache_count: 3, // Preload next 3 tracks by default
42            album_affinity_enabled: true,
43            album_affinity_threshold: 3,
44            storage_limit: 10 * 1024 * 1024 * 1024, // 10GB
45            wifi_only: false,                       // Allow preloading on any connection by default
46            // A week: long enough that re-watching over a weekend still hits
47            // disk, short enough that a one-off play does not hold space
48            // indefinitely.
49            temporary_ttl_hours: 24 * 7,
50        }
51    }
52}
53
54/// Smart caching engine
55#[derive(Clone)]
56pub struct SmartCache {
57    config: Arc<Mutex<CacheConfig>>,
58    /// Track recently played items per album
59    album_play_history: Arc<Mutex<HashMap<String, Vec<String>>>>,
60}
61
62impl SmartCache {
63    pub fn new(config: CacheConfig) -> Self {
64        Self {
65            config: Arc::new(Mutex::new(config)),
66            album_play_history: Arc::new(Mutex::new(HashMap::new())),
67        }
68    }
69
70    /// Update configuration
71    pub fn update_config(&self, config: CacheConfig) {
72        if let Ok(mut cfg) = self.config.lock() {
73            *cfg = config;
74        }
75    }
76
77    /// Check if should pre-cache queue items.
78    ///
79    /// Note this deliberately does NOT consult `wifi_only`. It used to return
80    /// `queue_precache_enabled && !wifi_only`, which disabled precaching
81    /// outright whenever the user enabled WiFi-only — regardless of the network
82    /// actually in use. The network check now lives in the download queue pump
83    /// (`downloads_allowed_on_current_network`), which is the single gate for
84    /// all download traffic, so this only answers "is precaching enabled?".
85    ///
86    /// TRACES: UR-053 | DR-074
87    pub fn should_precache_queue(&self) -> bool {
88        self.config
89            .lock()
90            .map(|cfg| cfg.queue_precache_enabled)
91            .unwrap_or(false)
92    }
93
94    /// Get number of queue items to pre-cache
95    pub fn queue_precache_count(&self) -> usize {
96        self.config
97            .lock()
98            .map(|cfg| cfg.queue_precache_count)
99            .unwrap_or(5)
100    }
101
102    /// Track that an item was played
103    pub fn track_play(&self, item_id: &str, album_id: Option<&str>) {
104        if let Some(album) = album_id {
105            if let Ok(mut history) = self.album_play_history.lock() {
106                let plays = history.entry(album.to_string()).or_insert_with(Vec::new);
107                if !plays.contains(&item_id.to_string()) {
108                    plays.push(item_id.to_string());
109                }
110            }
111        }
112    }
113
114    /// Check if album affinity threshold reached for caching
115    pub fn should_cache_album(&self, album_id: &str) -> Option<bool> {
116        let config = self.config.lock().ok()?;
117        if !config.album_affinity_enabled {
118            return Some(false);
119        }
120
121        let history = self.album_play_history.lock().ok()?;
122        let play_count = history.get(album_id).map(|v| v.len()).unwrap_or(0);
123
124        Some(play_count >= config.album_affinity_threshold)
125    }
126
127    /// Get configuration
128    pub fn get_config(&self) -> Option<CacheConfig> {
129        self.config.lock().ok().map(|cfg| cfg.clone())
130    }
131
132    /// Get all tracked albums with their play counts
133    /// Returns Vec<(album_id, unique_tracks_played)>
134    pub fn get_album_play_history(&self) -> Vec<(String, usize)> {
135        self.album_play_history
136            .lock()
137            .ok()
138            .map(|history| {
139                history
140                    .iter()
141                    .map(|(album_id, tracks)| (album_id.clone(), tracks.len()))
142                    .collect()
143            })
144            .unwrap_or_default()
145    }
146
147    // ============= Async versions for DatabaseService =============
148
149    /// Get total download size for a user (async version)
150    pub async fn get_total_download_size_async<S: DatabaseService>(
151        &self,
152        db_service: &Arc<S>,
153        user_id: &str,
154    ) -> Result<u64, String> {
155        let query = Query::with_params(
156            "SELECT COALESCE(SUM(file_size), 0) FROM downloads
157             WHERE user_id = ? AND status = 'completed'",
158            vec![QueryParam::String(user_id.to_string())],
159        );
160
161        let size: i64 = db_service
162            .query_one(query, |row| row.get(0))
163            .await
164            .map_err(|e| e.to_string())?;
165
166        Ok(size as u64)
167    }
168
169    /// Check if storage limit allows download (async version)
170    pub async fn can_download_async<S: DatabaseService>(
171        &self,
172        db_service: &Arc<S>,
173        user_id: &str,
174        new_size: u64,
175    ) -> bool {
176        // Clone config to avoid holding lock across await
177        let storage_limit = {
178            match self.config.lock() {
179                Ok(cfg) => cfg.storage_limit,
180                Err(_) => return true,
181            }
182        };
183
184        if storage_limit == 0 {
185            return true; // Unlimited
186        }
187
188        let current_size = self
189            .get_total_download_size_async(db_service, user_id)
190            .await
191            .unwrap_or(0);
192
193        current_size + new_size <= storage_limit
194    }
195
196    /// Reclaim temporary downloads whose life limit has passed.
197    ///
198    /// The time-based half of the temporary tier (DR-127); [`evict_lru_async`]
199    /// is the space-pressure half. A row is reclaimed by whichever fires first.
200    ///
201    /// Scoped to `download_source = 'auto'` for the same reason eviction is: a
202    /// `'user'` row is someone's own download and has no expiry. `COALESCE`
203    /// guards rows predating migration 012, whose source is NULL and whose
204    /// provenance must therefore be treated as the user's.
205    ///
206    /// Expiry is normally *derived* — `completed_at` plus the configured TTL —
207    /// rather than stamped at completion. That means a TTL change applies to
208    /// entries already on disk instead of only to future ones, and entries
209    /// predating the column expire without a backfill. `expires_at` is honoured
210    /// as a per-row override when something sets it.
211    ///
212    /// `now` is passed in rather than read from the clock so the policy is
213    /// testable without sleeping. Both sides go through SQLite's `datetime()`
214    /// because `completed_at` is written as `CURRENT_TIMESTAMP`
215    /// (`YYYY-MM-DD HH:MM:SS`) while callers pass RFC-3339 (`…T…+00:00`) — a raw
216    /// string comparison between the two formats is wrong, since `' ' < 'T'`.
217    ///
218    /// Returns the number of entries reclaimed.
219    ///
220    /// TRACES: UR-071 | DR-127 | UT-120
221    pub async fn reclaim_expired_async<S: DatabaseService>(
222        &self,
223        db_service: &Arc<S>,
224        user_id: &str,
225        now: &str,
226    ) -> Result<usize, String> {
227        let ttl_hours = {
228            let config = self.config.lock().map_err(|e| e.to_string())?;
229            config.temporary_ttl_hours
230        };
231        // 0 disables time-based reclaim; space pressure remains the only trigger.
232        if ttl_hours == 0 {
233            return Ok(0);
234        }
235
236        let expired: Vec<(i64, String)> = db_service
237            .query_many(
238                Query::with_params(
239                    "SELECT id, file_path FROM downloads
240                      WHERE user_id = ?
241                        AND COALESCE(download_source, 'user') = 'auto'
242                        AND status = 'completed'
243                        AND datetime(
244                              COALESCE(expires_at, datetime(completed_at, '+' || ? || ' hours'))
245                            ) < datetime(?)",
246                    vec![
247                        QueryParam::String(user_id.to_string()),
248                        QueryParam::String(ttl_hours.to_string()),
249                        QueryParam::String(now.to_string()),
250                    ],
251                ),
252                |row| Ok((row.get(0)?, row.get(1)?)),
253            )
254            .await
255            .map_err(|e| e.to_string())?;
256
257        let mut reclaimed = 0usize;
258        for (id, file_path) in expired {
259            // Best-effort on the file: a missing one still needs its row gone,
260            // or the sweep retries it forever.
261            let _ = std::fs::remove_file(&file_path);
262            db_service
263                .execute(Query::with_params(
264                    "DELETE FROM downloads WHERE id = ?",
265                    vec![QueryParam::Int64(id)],
266                ))
267                .await
268                .map_err(|e| e.to_string())?;
269            reclaimed += 1;
270        }
271
272        if reclaimed > 0 {
273            info!("[SmartCache] Reclaimed {} expired entries", reclaimed);
274        }
275        Ok(reclaimed)
276    }
277
278    /// Evict least recently used items to make space (async version).
279    ///
280    /// The space-pressure half of the temporary tier; [`reclaim_expired_async`]
281    /// is the time-based half.
282    pub async fn evict_lru_async<S: DatabaseService>(
283        &self,
284        db_service: &Arc<S>,
285        user_id: &str,
286        space_needed: u64,
287    ) -> Result<u64, String> {
288        let current_size = self
289            .get_total_download_size_async(db_service, user_id)
290            .await?;
291
292        // Get limit without holding lock across await
293        let limit = {
294            let config = self.config.lock().map_err(|e| e.to_string())?;
295            config.storage_limit
296        };
297
298        if limit == 0 || current_size + space_needed <= limit {
299            return Ok(0); // No eviction needed
300        }
301
302        let to_free = (current_size + space_needed) - limit;
303        let mut freed: u64 = 0;
304
305        // Only the *temporary* tier is evictable. `download_source = 'auto'` is
306        // precache — the cache put it there, the cache may reclaim it. A 'user'
307        // row is a download someone explicitly asked for; deleting it to make
308        // room for a predictive fetch is data loss, and because the old query
309        // ordered purely by `completed_at ASC` it took the oldest — typically
310        // exactly the film saved for a flight.
311        //
312        // COALESCE, not `= 'auto'` alone: migration 012 added the column with a
313        // 'user' default, but rows predating it can be NULL, and an unknown
314        // provenance must be treated as the user's, never as disposable.
315        //
316        // Freeing less than requested is the correct outcome when only user
317        // downloads remain — the caller surfaces "unable to free enough space"
318        // rather than silently deleting them.
319        //
320        // TRACES: UR-071 | DR-126 | UT-108
321        let query = Query::with_params(
322            "SELECT id, file_size, file_path FROM downloads
323             WHERE user_id = ? AND status = 'completed'
324               AND COALESCE(download_source, 'user') = 'auto'
325             ORDER BY completed_at ASC",
326            vec![QueryParam::String(user_id.to_string())],
327        );
328
329        let downloads: Vec<(i64, i64, String)> = db_service
330            .query_many(query, |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))
331            .await
332            .map_err(|e| e.to_string())?;
333
334        for (id, size, file_path) in downloads {
335            if freed >= to_free {
336                break;
337            }
338
339            // Delete file
340            let _ = std::fs::remove_file(&file_path);
341            debug!("[SmartCache] Evicted: {} ({} bytes)", file_path, size);
342
343            // Delete from database
344            let delete_query = Query::with_params(
345                "DELETE FROM downloads WHERE id = ?",
346                vec![QueryParam::Int64(id)],
347            );
348            db_service
349                .execute(delete_query)
350                .await
351                .map_err(|e| e.to_string())?;
352
353            freed += size as u64;
354        }
355
356        info!("[SmartCache] Freed {} bytes ({} needed)", freed, to_free);
357        Ok(freed)
358    }
359}
360
361#[cfg(test)]
362mod tests {
363    use super::*;
364
365    #[test]
366    fn test_default_config() {
367        let config = CacheConfig::default();
368        assert_eq!(config.queue_precache_count, 3);
369        assert_eq!(config.album_affinity_threshold, 3);
370        assert!(!config.wifi_only); // wifi_only is false by default for easier preloading
371    }
372
373    #[test]
374    fn test_album_affinity_tracking() {
375        let cache = SmartCache::new(CacheConfig::default());
376
377        // Track plays from same album
378        cache.track_play("track1", Some("album1"));
379        cache.track_play("track2", Some("album1"));
380
381        // Below threshold
382        assert!(!cache.should_cache_album("album1").unwrap_or(false));
383
384        cache.track_play("track3", Some("album1"));
385
386        // At threshold - should cache
387        assert!(cache.should_cache_album("album1").unwrap_or(false));
388    }
389
390    #[test]
391    fn test_queue_precache_config() {
392        let config = CacheConfig {
393            queue_precache_enabled: false,
394            ..CacheConfig::default()
395        };
396
397        let cache = SmartCache::new(config);
398        assert!(!cache.should_precache_queue());
399
400        let new_config = CacheConfig {
401            wifi_only: false,
402            ..CacheConfig::default()
403        };
404        cache.update_config(new_config);
405
406        assert!(cache.should_precache_queue());
407    }
408
409    #[test]
410    fn test_wifi_only_does_not_disable_precaching() {
411        // wifi_only must not short-circuit precaching: the network gate lives in
412        // the download pump, which checks the *actual* transport. Enabling
413        // WiFi-only while on WiFi should still precache.
414        let config = CacheConfig {
415            queue_precache_enabled: true,
416            wifi_only: true,
417            ..CacheConfig::default()
418        };
419
420        let cache = SmartCache::new(config);
421        assert!(cache.should_precache_queue());
422    }
423
424    /// Expiry reclaims only temporary entries that are actually past their life
425    /// limit — never a user's download (which has no expiry), and never a
426    /// temporary entry still within its life.
427    ///
428    /// TRACES: UR-071 | DR-127 | UT-120
429    #[tokio::test]
430    async fn test_reclaim_expired_only_takes_expired_temporary_entries() {
431        use crate::storage::db_service::RusqliteService;
432        use rusqlite::Connection;
433        use std::sync::{Arc, Mutex};
434
435        let conn = Connection::open_in_memory().unwrap();
436        conn.execute(
437            "CREATE TABLE downloads (
438                id INTEGER PRIMARY KEY,
439                user_id TEXT,
440                status TEXT,
441                file_size INTEGER,
442                file_path TEXT,
443                completed_at TEXT,
444                download_source TEXT DEFAULT 'user',
445                expires_at TEXT
446            )",
447            [],
448        )
449        .unwrap();
450
451        // `completed_at` is in SQLite's CURRENT_TIMESTAMP format (space, not
452        // 'T'), deliberately: the query has to compare it against an RFC-3339
453        // "now" and must not do so as raw strings.
454        for (path, source, completed, expires) in [
455            // Completed long ago, no override => derived expiry has passed.
456            ("/tmp/jt-expired.mp4", "auto", "2026-01-01 00:00:00", None),
457            // Completed yesterday => still inside the 7-day default TTL.
458            ("/tmp/jt-fresh.mp4", "auto", "2026-05-31 00:00:00", None),
459            // Old, but an explicit override keeps it alive.
460            (
461                "/tmp/jt-override.mp4",
462                "auto",
463                "2026-01-01 00:00:00",
464                Some("2026-12-01T00:00:00+00:00"),
465            ),
466            // A user download must never carry an expiry, but assert the sweep
467            // ignores it even if one were somehow set.
468            (
469                "/tmp/jt-user.mp4",
470                "user",
471                "2026-01-01 00:00:00",
472                Some("2026-01-01T00:00:00+00:00"),
473            ),
474            (
475                "/tmp/jt-user-noexp.mp4",
476                "user",
477                "2026-01-01 00:00:00",
478                None,
479            ),
480        ] {
481            conn.execute(
482                "INSERT INTO downloads (user_id, status, file_size, file_path, download_source, completed_at, expires_at)
483                 VALUES ('user1', 'completed', 10, ?1, ?2, ?3, ?4)",
484                rusqlite::params![path, source, completed, expires],
485            )
486            .unwrap();
487        }
488
489        let conn_arc = Arc::new(Mutex::new(conn));
490        let db_service = Arc::new(RusqliteService::new(conn_arc.clone()));
491        let cache = SmartCache::new(CacheConfig::default());
492
493        let reclaimed = cache
494            .reclaim_expired_async(&db_service, "user1", "2026-06-01T00:00:00+00:00")
495            .await
496            .unwrap();
497        assert_eq!(
498            reclaimed, 1,
499            "only the expired temporary entry is reclaimed"
500        );
501
502        let surviving: Vec<String> = {
503            let guard = conn_arc.lock_safe();
504            let mut stmt = guard
505                .prepare("SELECT file_path FROM downloads ORDER BY id")
506                .unwrap();
507            let rows = stmt
508                .query_map([], |row| row.get::<_, String>(0))
509                .unwrap()
510                .map(|r| r.unwrap())
511                .collect();
512            rows
513        };
514        assert_eq!(
515            surviving,
516            vec![
517                "/tmp/jt-fresh.mp4".to_string(),
518                "/tmp/jt-override.mp4".to_string(),
519                "/tmp/jt-user.mp4".to_string(),
520                "/tmp/jt-user-noexp.mp4".to_string(),
521            ],
522            "entries within their life, those with a later override, and every user download must survive"
523        );
524
525        // TTL of 0 disables time-based reclaim entirely.
526        let cache_no_ttl = SmartCache::new(CacheConfig {
527            temporary_ttl_hours: 0,
528            ..Default::default()
529        });
530        assert_eq!(
531            cache_no_ttl
532                .reclaim_expired_async(&db_service, "user1", "2027-01-01T00:00:00+00:00")
533                .await
534                .unwrap(),
535            0,
536            "a zero TTL leaves space pressure as the only reclaim trigger"
537        );
538    }
539
540    /// Eviction must only reclaim *temporary* (`download_source = 'auto'`)
541    /// downloads — the precache tier. A download the user explicitly asked for
542    /// is their file: it may be deleted by them, never by the cache making room
543    /// for a predictive fetch.
544    ///
545    /// Before the fix, `evict_lru_async` selected every completed row ordered by
546    /// `completed_at ASC` with no source filter, so hitting the storage limit
547    /// deleted the *oldest* download — typically the film someone downloaded for
548    /// a flight — in favour of a newer auto-precached track.
549    ///
550    /// TRACES: UR-071 | DR-126 | UT-108
551    #[tokio::test]
552    async fn test_evict_lru_never_deletes_user_downloads() {
553        use crate::storage::db_service::RusqliteService;
554        use rusqlite::Connection;
555        use std::sync::{Arc, Mutex};
556
557        let conn = Connection::open_in_memory().unwrap();
558        conn.execute(
559            "CREATE TABLE downloads (
560                id INTEGER PRIMARY KEY,
561                user_id TEXT,
562                status TEXT,
563                file_size INTEGER,
564                file_path TEXT,
565                completed_at TEXT,
566                download_source TEXT DEFAULT 'user'
567            )",
568            [],
569        )
570        .unwrap();
571
572        // The user's own download is the OLDEST, so a purely time-ordered
573        // eviction would take it first.
574        conn.execute(
575            "INSERT INTO downloads (user_id, status, file_size, file_path, completed_at, download_source)
576             VALUES ('user1', 'completed', 600, '/tmp/jellytau-test-user.mp4', '2026-01-01', 'user')",
577            [],
578        )
579        .unwrap();
580        // A newer, auto-precached item.
581        conn.execute(
582            "INSERT INTO downloads (user_id, status, file_size, file_path, completed_at, download_source)
583             VALUES ('user1', 'completed', 600, '/tmp/jellytau-test-auto.mp4', '2026-06-01', 'auto')",
584            [],
585        )
586        .unwrap();
587
588        let conn_arc = Arc::new(Mutex::new(conn));
589        let db_service = Arc::new(RusqliteService::new(conn_arc.clone()));
590
591        let cache = SmartCache::new(CacheConfig {
592            storage_limit: 1000,
593            ..Default::default()
594        });
595
596        // 1200 bytes held against a 1000 limit: eviction must free something.
597        let freed = cache
598            .evict_lru_async(&db_service, "user1", 0)
599            .await
600            .unwrap();
601        assert!(freed > 0, "eviction should have reclaimed the auto entry");
602
603        let surviving: Vec<String> = {
604            let guard = conn_arc.lock_safe();
605            let mut stmt = guard
606                .prepare("SELECT download_source FROM downloads ORDER BY id")
607                .unwrap();
608            let rows = stmt
609                .query_map([], |row| row.get::<_, String>(0))
610                .unwrap()
611                .map(|r| r.unwrap())
612                .collect();
613            rows
614        };
615
616        assert_eq!(
617            surviving,
618            vec!["user".to_string()],
619            "the user's own download must survive; only the 'auto' entry is evictable"
620        );
621    }
622
623    #[tokio::test]
624    async fn test_storage_limit_check() {
625        use crate::storage::db_service::RusqliteService;
626        use rusqlite::Connection;
627        use std::sync::{Arc, Mutex};
628
629        let conn = Connection::open_in_memory().unwrap();
630        conn.execute(
631            "CREATE TABLE downloads (
632                id INTEGER PRIMARY KEY,
633                user_id TEXT,
634                status TEXT,
635                file_size INTEGER
636            )",
637            [],
638        )
639        .unwrap();
640
641        let conn_arc = Arc::new(Mutex::new(conn));
642        let db_service = Arc::new(RusqliteService::new(conn_arc.clone()));
643
644        let config = CacheConfig {
645            storage_limit: 1000,
646            ..Default::default()
647        };
648        let cache = SmartCache::new(config);
649
650        // Empty - can download
651        assert!(cache.can_download_async(&db_service, "user1", 500).await);
652
653        // Add some downloads
654        {
655            let conn_guard = conn_arc.lock_safe();
656            conn_guard.execute(
657                "INSERT INTO downloads (user_id, status, file_size) VALUES ('user1', 'completed', 600)",
658                [],
659            )
660            .unwrap();
661        }
662
663        // Total would be 1100 > 1000
664        assert!(!cache.can_download_async(&db_service, "user1", 500).await);
665
666        // Smaller size fits
667        assert!(cache.can_download_async(&db_service, "user1", 300).await);
668    }
669}
jellytau_lib/download/cache.rs

jellytau_lib/download/
cache.rs
