hugo/RuDS
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement the bDS2-compatible core Lua API

Browse Source
This commit is contained in:
Chili Palmer
2026-07-19 09:24:39 +02:00
parent 33929fd04b
commit 9a72287fc6
37 changed files with 11990 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
crates/bds-core/src/engine/blogmark.rs
View File

@@ -1,5 +1,6 @@
use std::fs;
use std::path::Path;
use std::sync::Arc;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
@@ -9,7 +10,7 @@ use crate::db::DbConnection as Connection;
use crate::db::queries::script as script_queries;
use crate::engine::{EngineError, EngineResult};
use crate::model::{Post, Script, ScriptKind};
use crate::scripting::{self, ExecutionControl, ExecutionKind};
use crate::scripting::{self, CoreHost, ExecutionControl, ExecutionKind, HostApi, UnavailableHost};
const MAX_TITLE_LENGTH: usize = 200;
const MAX_URL_LENGTH: usize = 2_048;
@@ -80,6 +81,27 @@ pub fn receive_deep_link(
data_dir: &Path,
project_id: &str,
raw: &str,
) -> EngineResult<BlogmarkImportResult> {
let host = CoreHost::from_connection(conn, project_id, data_dir)
.map(|host| Arc::new(host) as Arc<dyn HostApi>)
.unwrap_or_else(|_| Arc::new(UnavailableHost));
receive_deep_link_with_host(
conn,
data_dir,
project_id,
raw,
&ExecutionControl::default(),
host,
)
}
pub fn receive_deep_link_with_host(
conn: &Connection,
data_dir: &Path,
project_id: &str,
raw: &str,
control: &ExecutionControl,
host: Arc<dyn HostApi>,
) -> EngineResult<BlogmarkImportResult> {
let mut candidate = parse_deep_link(raw)?;
if let Some(target) = &candidate.project_id
@@ -98,7 +120,10 @@ pub fn receive_deep_link(
));
}
let (mut candidate, toasts, transform_errors) =
run_transforms(conn, data_dir, project_id, candidate)?;
run_transforms(conn, data_dir, project_id, candidate, control, host)?;
if control.is_cancelled() {
return Err(EngineError::Validation("script cancelled".into()));
}
if candidate.categories.is_empty() {
let metadata = crate::engine::meta::read_project_json(data_dir)?;
if let Some(category) = metadata
@@ -133,6 +158,8 @@ fn run_transforms(
data_dir: &Path,
project_id: &str,
candidate: BlogmarkCandidate,
control: &ExecutionControl,
host: Arc<dyn HostApi>,
) -> EngineResult<(BlogmarkCandidate, Vec<String>, Vec<String>)> {
let mut transforms = script_queries::list_scripts_by_project(conn, project_id)?
.into_iter()
@@ -149,6 +176,9 @@ fn run_transforms(
let mut toasts = Vec::new();
let mut errors = Vec::new();
for script in transforms {
if control.is_cancelled() {
return Err(EngineError::Validation("script cancelled".into()));
}
if script.entrypoint.trim().is_empty() {
continue;
}
@@ -157,12 +187,13 @@ fn run_transforms(
"source": "blogmark",
"url": current.get("url").cloned().unwrap_or(Value::Null),
});
match scripting::execute_many(
match scripting::execute_many_with_host(
&source,
&script.entrypoint,
&[current.clone(), context],
ExecutionKind::Transform,
&ExecutionControl::default(),
control,
Arc::clone(&host),
) {
Ok(execution) => {
let (next, returned_toasts) = split_transform_result(execution.value, &current);
@@ -172,6 +203,9 @@ fn run_transforms(
execution.toasts.into_iter().chain(returned_toasts),
);
}
Err(_error) if control.is_cancelled() => {
return Err(EngineError::Validation("script cancelled".into()));
}
Err(error) => errors.push(format!("{}: {error}", script.slug)),
}
}
@@ -287,6 +321,22 @@ mod tests {
use crate::model::ScriptKind;
use tempfile::TempDir;
struct PostsHost;
impl HostApi for PostsHost {
fn call(
&self,
namespace: &str,
method: &str,
_arguments: Vec<Value>,
) -> Result<Value, String> {
match (namespace, method) {
("posts", "get_all") => Ok(json!([{"id":"one"}, {"id":"two"}])),
_ => Err("unsupported test capability".into()),
}
}
}
#[test]
fn parses_and_hardens_blogmark_links() {
let candidate = parse_deep_link(
@@ -358,4 +408,80 @@ mod tests {
assert_eq!(result.toasts, vec!["done"]);
assert!(result.transform_errors.is_empty());
}
#[test]
fn transforms_use_the_supplied_project_host() {
let db = Database::open_in_memory().unwrap();
db.migrate().unwrap();
crate::db::fts::ensure_fts_tables(db.conn()).unwrap();
let directory = TempDir::new().unwrap();
let project = crate::engine::project::create_project(
db.conn(),
"Blog",
Some(directory.path().to_str().unwrap()),
)
.unwrap();
crate::engine::script::create_script(
db.conn(),
&project.id,
"Count posts",
ScriptKind::Transform,
"function main(data) data.title = tostring(#bds.posts.get_all()) .. ' posts'; return data end",
None,
)
.unwrap();
let result = receive_deep_link_with_host(
db.conn(),
directory.path(),
&project.id,
"ruds://new-post?title=Example",
&ExecutionControl::default(),
Arc::new(PostsHost),
)
.unwrap();
assert_eq!(result.post.title, "2 posts");
}
#[test]
fn cancellation_stops_transform_import_before_post_creation() {
let db = Database::open_in_memory().unwrap();
db.migrate().unwrap();
crate::db::fts::ensure_fts_tables(db.conn()).unwrap();
let directory = TempDir::new().unwrap();
let project = crate::engine::project::create_project(
db.conn(),
"Blog",
Some(directory.path().to_str().unwrap()),
)
.unwrap();
crate::engine::script::create_script(
db.conn(),
&project.id,
"Cancelled",
ScriptKind::Transform,
"function main(data) return data end",
None,
)
.unwrap();
let control = ExecutionControl::default();
control.cancel();
assert!(
receive_deep_link_with_host(
db.conn(),
directory.path(),
&project.id,
"ruds://new-post?title=Example",
&control,
Arc::new(PostsHost),
)
.is_err()
);
assert_eq!(
crate::db::queries::post::count_posts_by_project(db.conn(), &project.id).unwrap(),
0
);
}
}

203
crates/bds-core/src/engine/search.rs
View File

@@ -93,6 +93,26 @@ pub fn rebuild_search_index(
}
}
/// Reindex one project without disturbing rows belonging to other projects.
pub fn reindex_project(
conn: &Connection,
project_id: &str,
on_item: Option<ItemProgressFn>,
) -> EngineResult<ReindexReport> {
let project = project_q::get_project_by_id(conn, project_id)?;
let total = post_q::count_posts_by_project(conn, project_id)? as usize
+ media_q::count_media_by_project(conn, project_id)? as usize;
let mut current = 0;
index_project(
conn,
project_id,
project.data_path.as_deref(),
total,
&mut current,
on_item.as_ref(),
)
}
fn index_all_projects(
conn: &Connection,
on_item: Option<&ItemProgressFn>,
@@ -112,71 +132,94 @@ fn index_all_projects(
};
for project in projects {
let main_language = project
.data_path
.as_deref()
.and_then(|path| crate::engine::meta::read_project_json(Path::new(path)).ok())
.and_then(|metadata| metadata.main_language)
.unwrap_or_else(|| "en".to_string());
let indexed = index_project(
conn,
&project.id,
project.data_path.as_deref(),
total,
&mut current,
on_item,
)?;
report.posts_indexed += indexed.posts_indexed;
report.media_indexed += indexed.media_indexed;
}
for post in post_q::list_posts_by_project(conn, &project.id)? {
current += 1;
if let Some(callback) = on_item {
callback(current, total, &post.title);
}
let translations = post_translation::list_post_translations_by_post(conn, &post.id)?;
let translation_data = translations
.iter()
.map(|translation| fts::PostTranslationFts {
title: translation.title.clone(),
excerpt: translation.excerpt.clone(),
content: translation.content.clone(),
language: translation.language.clone(),
})
.collect::<Vec<_>>();
fts::index_post(
conn,
&post.id,
&post.title,
post.excerpt.as_deref(),
post.content.as_deref(),
&post.tags,
&post.categories,
&translation_data,
post.language.as_deref().unwrap_or(&main_language),
)?;
report.posts_indexed += 1;
}
Ok(report)
}
for media in media_q::list_media_by_project(conn, &project.id)? {
current += 1;
if let Some(callback) = on_item {
callback(current, total, &media.original_name);
}
let translations =
media_translation::list_media_translations_by_media(conn, &media.id)?;
let translation_data = translations
.iter()
.map(|translation| fts::MediaTranslationFts {
title: translation.title.clone(),
alt: translation.alt.clone(),
caption: translation.caption.clone(),
language: translation.language.clone(),
})
.collect::<Vec<_>>();
fts::index_media(
conn,
&media.id,
media.title.as_deref(),
media.alt.as_deref(),
media.caption.as_deref(),
&media.original_name,
&media.tags,
&translation_data,
media.language.as_deref().unwrap_or(&main_language),
)?;
report.media_indexed += 1;
fn index_project(
conn: &Connection,
project_id: &str,
data_path: Option<&str>,
total: usize,
current: &mut usize,
on_item: Option<&ItemProgressFn>,
) -> EngineResult<ReindexReport> {
let main_language = data_path
.and_then(|path| crate::engine::meta::read_project_json(Path::new(path)).ok())
.and_then(|metadata| metadata.main_language)
.unwrap_or_else(|| "en".to_string());
let mut report = ReindexReport {
posts_indexed: 0,
media_indexed: 0,
};
for post in post_q::list_posts_by_project(conn, project_id)? {
*current += 1;
if let Some(callback) = on_item {
callback(*current, total, &post.title);
}
let translations = post_translation::list_post_translations_by_post(conn, &post.id)?;
let translation_data = translations
.iter()
.map(|translation| fts::PostTranslationFts {
title: translation.title.clone(),
excerpt: translation.excerpt.clone(),
content: translation.content.clone(),
language: translation.language.clone(),
})
.collect::<Vec<_>>();
fts::index_post(
conn,
&post.id,
&post.title,
post.excerpt.as_deref(),
post.content.as_deref(),
&post.tags,
&post.categories,
&translation_data,
post.language.as_deref().unwrap_or(&main_language),
)?;
report.posts_indexed += 1;
}
for media in media_q::list_media_by_project(conn, project_id)? {
*current += 1;
if let Some(callback) = on_item {
callback(*current, total, &media.original_name);
}
let translations = media_translation::list_media_translations_by_media(conn, &media.id)?;
let translation_data = translations
.iter()
.map(|translation| fts::MediaTranslationFts {
title: translation.title.clone(),
alt: translation.alt.clone(),
caption: translation.caption.clone(),
language: translation.language.clone(),
})
.collect::<Vec<_>>();
fts::index_media(
conn,
&media.id,
media.title.as_deref(),
media.alt.as_deref(),
media.caption.as_deref(),
&media.original_name,
&media.tags,
&translation_data,
media.language.as_deref().unwrap_or(&main_language),
)?;
report.media_indexed += 1;
}
Ok(report)
@@ -286,4 +329,40 @@ mod tests {
assert_eq!(report.posts_indexed, 2);
assert_eq!(results.len(), 2);
}
#[test]
fn project_reindex_leaves_other_project_rows_intact() {
let (db, first_project_id) = setup();
let first_dir = tempfile::tempdir().unwrap();
let second_dir = tempfile::tempdir().unwrap();
let second = engine::project::create_project(
db.conn(),
"Second Project",
Some(second_dir.path().to_str().unwrap()),
)
.unwrap();
for (project_id, data_dir, title) in [
(&first_project_id, first_dir.path(), "First Searchable"),
(&second.id, second_dir.path(), "Second Searchable"),
] {
engine::post::create_post(
db.conn(),
data_dir,
project_id,
title,
Some("body"),
vec![],
vec![],
None,
Some("en"),
None,
)
.unwrap();
}
reindex_project(db.conn(), &first_project_id, None).unwrap();
let results = crate::db::fts::search_posts(db.conn(), "searchable", "en").unwrap();
assert_eq!(results.len(), 2);
}
}

29
crates/bds-core/src/engine/task.rs
View File

@@ -179,6 +179,16 @@ impl TaskManager {
.unwrap_or(false)
}
/// Shared cancellation flag for a worker owned by this task.
pub fn cancellation_flag(&self, task_id: TaskId) -> Option<Arc<AtomicBool>> {
self.tasks
.lock()
.unwrap()
.iter()
.find(|task| task.id == task_id)
.map(|task| Arc::clone(&task.cancel_flag))
}
/// Return the current status of a task.
pub fn status(&self, task_id: TaskId) -> Option<TaskStatus> {
let tasks = self.tasks.lock().unwrap();
@@ -216,6 +226,14 @@ impl TaskManager {
});
}
/// Remove every finished task while preserving running and queued work.
pub fn clear_completed(&self) {
self.tasks
.lock()
.unwrap()
.retain(|task| matches!(task.status, TaskStatus::Pending | TaskStatus::Running));
}
/// Update progress for a running task. Throttled to at most once per 250ms.
pub fn report_progress(&self, task_id: TaskId, progress: Option<f32>, message: Option<String>) {
let mut tasks = self.tasks.lock().unwrap();
@@ -403,6 +421,17 @@ mod tests {
assert_eq!(mgr.snapshots().len(), 10);
}
#[test]
fn clear_completed_preserves_active_tasks() {
let mgr = TaskManager::new(2);
let done = mgr.submit("done");
let running = mgr.submit("running");
mgr.complete(done);
mgr.clear_completed();
assert_eq!(mgr.status(done), None);
assert_eq!(mgr.status(running), Some(TaskStatus::Running));
}
#[test]
fn completing_task_starts_next_queued() {
let mgr = TaskManager::new(1);