JezzWTF/phokus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improve indexing updates and thumbnail streaming

Browse Source
This commit is contained in:
LyAhn
2026-04-05 16:49:10 +01:00
parent c5e9c83ac9
commit c299c7d452
7 changed files with 382 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/db.rs
+97
View File
@@ -65,6 +65,13 @@ pub struct EmbeddingJob {
pub updated_at: String,
}
#[derive(Debug, Clone)]
pub struct ThumbnailJob {
pub image_id: i64,
pub path: String,
pub media_kind: String,
}
pub fn create_pool(db_path: &Path) -> Result<DbPool> {
vector::register_sqlite_vec();
let manager = SqliteConnectionManager::file(db_path);
@@ -111,9 +118,19 @@ pub fn migrate(conn: &Connection) -> Result<()> {
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS thumbnail_jobs (
image_id INTEGER PRIMARY KEY REFERENCES images(id) ON DELETE CASCADE,
status TEXT NOT NULL DEFAULT 'pending',
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_images_folder_id ON images(folder_id);
CREATE INDEX IF NOT EXISTS idx_images_modified_at ON images(modified_at);
CREATE INDEX IF NOT EXISTS idx_embedding_jobs_status ON embedding_jobs(status);
CREATE INDEX IF NOT EXISTS idx_thumbnail_jobs_status ON thumbnail_jobs(status);
",
)?;
@@ -209,6 +226,19 @@ pub fn enqueue_embedding_job(conn: &Connection, image_id: i64) -> Result<()> {
Ok(())
}
pub fn enqueue_thumbnail_job(conn: &Connection, image_id: i64) -> Result<()> {
conn.execute(
"INSERT INTO thumbnail_jobs (image_id, status, attempts, last_error, created_at, updated_at)
VALUES (?1, 'pending', 0, NULL, datetime('now'), datetime('now'))
ON CONFLICT(image_id) DO UPDATE SET
status = 'pending',
last_error = NULL,
updated_at = datetime('now')",
[image_id],
)?;
Ok(())
}
#[allow(dead_code)]
pub fn get_next_embedding_job(conn: &Connection) -> Result<Option<EmbeddingJob>> {
let mut stmt = conn.prepare(
@@ -282,6 +312,61 @@ pub fn update_folder_count(conn: &Connection, folder_id: i64) -> Result<()> {
Ok(())
}
pub fn get_next_thumbnail_job(conn: &Connection) -> Result<Option<ThumbnailJob>> {
let mut stmt = conn.prepare(
"SELECT j.image_id, i.path, i.media_kind
FROM thumbnail_jobs j
JOIN images i ON i.id = j.image_id
WHERE j.status = 'pending'
ORDER BY j.updated_at, j.image_id
LIMIT 1",
)?;
let mut rows = stmt.query([])?;
let Some(row) = rows.next()? else {
return Ok(None);
};
Ok(Some(ThumbnailJob {
image_id: row.get(0)?,
path: row.get(1)?,
media_kind: row.get(2)?,
}))
}
pub fn mark_thumbnail_job_processing(conn: &Connection, image_id: i64) -> Result<()> {
conn.execute(
"UPDATE thumbnail_jobs
SET status = 'processing', attempts = attempts + 1, updated_at = datetime('now')
WHERE image_id = ?1",
[image_id],
)?;
Ok(())
}
pub fn mark_thumbnail_ready(
conn: &Connection,
image_id: i64,
thumbnail_path: Option<&str>,
) -> Result<ImageRecord> {
conn.execute(
"UPDATE images SET thumbnail_path = ?2 WHERE id = ?1",
params![image_id, thumbnail_path],
)?;
conn.execute("DELETE FROM thumbnail_jobs WHERE image_id = ?1", [image_id])?;
get_image_by_id(conn, image_id)
}
pub fn mark_thumbnail_failed(conn: &Connection, image_id: i64, error: &str) -> Result<()> {
conn.execute(
"UPDATE thumbnail_jobs
SET status = 'failed', last_error = ?2, updated_at = datetime('now')
WHERE image_id = ?1",
params![image_id, error],
)?;
Ok(())
}
pub fn update_image_details(
conn: &Connection,
image_id: i64,
@@ -307,6 +392,18 @@ pub fn update_image_details(
.map_err(Into::into)
}
pub fn get_image_by_id(conn: &Connection, image_id: i64) -> Result<ImageRecord> {
conn.query_row(
"SELECT id, folder_id, path, filename, thumbnail_path, width, height, file_size, created_at, modified_at, mime_type,
media_kind, favorite, rating, embedding_status, embedding_model, embedding_updated_at, embedding_error
FROM images
WHERE id = ?1",
[image_id],
map_image_row,
)
.map_err(Into::into)
}
pub fn get_folders(conn: &Connection) -> Result<Vec<Folder>> {
let mut stmt =
conn.prepare("SELECT id, path, name, image_count, indexed_at FROM folders ORDER BY name")?;
src-tauri/src/indexer.rs
+103 -11
View File
@@ -63,7 +63,13 @@ pub struct IndexedImagesBatch {
pub images: Vec<ImageRecord>,
}
#[derive(Clone, Serialize)]
pub struct ThumbnailBatch {
pub images: Vec<ImageRecord>,
}
const INDEX_BATCH_SIZE: usize = 25;
const THUMBNAIL_BATCH_SIZE: usize = 12;
pub fn index_folder(
app: AppHandle,
@@ -101,25 +107,40 @@ fn do_index(app: AppHandle, pool: DbPool, folder_id: i64, folder_path: PathBuf)
},
);
// Parallel: read file metadata and dimensions across all CPU cores
let records: Vec<ImageRecord> = image_paths
.par_iter()
.filter_map(|path| build_record(path, folder_id))
.collect();
// Sequential: commit in batches and stream results to the frontend
let mut indexed = 0usize;
for chunk in records.chunks(INDEX_BATCH_SIZE) {
let committed = commit_batch(&pool, chunk)?;
for path_chunk in image_paths.chunks(INDEX_BATCH_SIZE) {
let records: Vec<ImageRecord> = path_chunk
.par_iter()
.filter_map(|path| build_record(path, folder_id))
.collect();
if records.is_empty() {
continue;
}
let committed = commit_batch(&pool, &records)?;
indexed += committed.len();
emit_images(&app, &IndexedImagesBatch { folder_id, images: committed });
emit_images(
&app,
&IndexedImagesBatch {
folder_id,
images: committed,
},
);
let current_file = path_chunk
.last()
.and_then(|path| path.file_name())
.map(|name| name.to_string_lossy().to_string())
.unwrap_or_default();
emit_progress(
&app,
&IndexProgress {
folder_id,
total,
indexed,
current_file: String::new(),
current_file,
done: false,
},
);
@@ -191,6 +212,10 @@ fn emit_images(app: &AppHandle, batch: &IndexedImagesBatch) {
let _ = app.emit("indexed-images", batch);
}
fn emit_thumbnails(app: &AppHandle, batch: &ThumbnailBatch) {
let _ = app.emit("thumbnail-updated", batch);
}
fn commit_batch(pool: &DbPool, records: &[ImageRecord]) -> Result<Vec<ImageRecord>> {
let mut conn = pool.get()?;
let tx = conn.transaction()?;
@@ -200,9 +225,76 @@ fn commit_batch(pool: &DbPool, records: &[ImageRecord]) -> Result<Vec<ImageRecor
let mut committed_record = record.clone();
committed_record.id = db::upsert_image(&tx, record)?;
db::enqueue_embedding_job(&tx, committed_record.id)?;
db::enqueue_thumbnail_job(&tx, committed_record.id)?;
committed.push(committed_record);
}
tx.commit()?;
Ok(committed)
}
pub fn start_thumbnail_worker(app: AppHandle, pool: DbPool, cache_dir: PathBuf) {
std::thread::spawn(move || loop {
if let Err(error) = process_thumbnail_batch(&app, &pool, &cache_dir) {
eprintln!("Thumbnail worker error: {}", error);
}
std::thread::sleep(std::time::Duration::from_millis(250));
});
}
fn process_thumbnail_batch(app: &AppHandle, pool: &DbPool, cache_dir: &Path) -> Result<()> {
let mut updated_images = Vec::new();
for _ in 0..THUMBNAIL_BATCH_SIZE {
let job = {
let conn = pool.get()?;
db::get_next_thumbnail_job(&conn)?
};
let Some(job) = job else {
break;
};
{
let conn = pool.get()?;
db::mark_thumbnail_job_processing(&conn, job.image_id)?;
}
let thumbnail_result = match job.media_kind.as_str() {
"image" => thumbnail::generate_thumbnail(Path::new(&job.path), cache_dir),
"video" => thumbnail::generate_video_poster(Path::new(&job.path), cache_dir),
_ => {
let conn = pool.get()?;
updated_images.push(db::mark_thumbnail_ready(&conn, job.image_id, None)?);
continue;
}
};
let thumbnail_path = match thumbnail_result {
Ok(path) => Some(path.to_string_lossy().to_string()),
Err(error) => {
let conn = pool.get()?;
db::mark_thumbnail_failed(&conn, job.image_id, &error.to_string())?;
continue;
}
};
let updated_image = {
let conn = pool.get()?;
db::mark_thumbnail_ready(&conn, job.image_id, thumbnail_path.as_deref())?
};
updated_images.push(updated_image);
}
if !updated_images.is_empty() {
emit_thumbnails(
app,
&ThumbnailBatch {
images: updated_images,
},
);
}
Ok(())
}
src-tauri/src/lib.rs
+5
View File
@@ -28,6 +28,11 @@ pub fn run() {
db::migrate(&conn).expect("Failed to run migrations");
}
let thumb_dir = app_dir.join("thumbnails");
std::fs::create_dir_all(&thumb_dir).expect("Failed to create thumbnail dir");
indexer::start_thumbnail_worker(app.handle().clone(), pool.clone(), thumb_dir);
app.manage(pool);
Ok(())
src-tauri/src/thumbnail.rs
+83 -1
View File
@@ -1,4 +1,86 @@
use std::path::Path;
use anyhow::{anyhow, Result};
use image::ImageFormat;
use std::path::{Path, PathBuf};
use std::process::Command;
pub const THUMB_SIZE: u32 = 320;
pub fn thumb_path(cache_dir: &Path, image_path: &str) -> PathBuf {
thumb_path_with_ext(cache_dir, image_path, "webp")
}
pub fn video_poster_path(cache_dir: &Path, video_path: &str) -> PathBuf {
thumb_path_with_ext(cache_dir, video_path, "jpg")
}
fn thumb_path_with_ext(cache_dir: &Path, input_path: &str, extension: &str) -> PathBuf {
let hash = hash_path(input_path);
cache_dir.join(format!("{}.{}", hash, extension))
}
fn hash_path(s: &str) -> String {
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
let mut hasher = DefaultHasher::new();
s.hash(&mut hasher);
format!("{:016x}", hasher.finish())
}
pub fn generate_thumbnail(image_path: &Path, cache_dir: &Path) -> Result<PathBuf> {
let path_str = image_path.to_string_lossy();
let out_path = thumb_path(cache_dir, &path_str);
if out_path.exists() {
return Ok(out_path);
}
let img = image::open(image_path)?;
let thumb = img.thumbnail(THUMB_SIZE, THUMB_SIZE);
if let Some(parent) = out_path.parent() {
std::fs::create_dir_all(parent)?;
}
thumb.save_with_format(&out_path, ImageFormat::WebP)?;
Ok(out_path)
}
pub fn generate_video_poster(video_path: &Path, cache_dir: &Path) -> Result<PathBuf> {
let path_str = video_path.to_string_lossy();
let out_path = video_poster_path(cache_dir, &path_str);
if out_path.exists() {
return Ok(out_path);
}
if let Some(parent) = out_path.parent() {
std::fs::create_dir_all(parent)?;
}
let status = Command::new("ffmpeg")
.args([
"-y",
"-ss",
"00:00:01.000",
"-i",
&path_str,
"-frames:v",
"1",
"-vf",
"scale=320:-1:force_original_aspect_ratio=decrease",
"-q:v",
"4",
out_path.to_string_lossy().as_ref(),
])
.status()?;
if !status.success() {
return Err(anyhow!("ffmpeg failed generating poster for {}", path_str));
}
Ok(out_path)
}
/// Gets image dimensions without fully decoding.
pub fn get_dimensions(image_path: &Path) -> Option<(u32, u32)> {