Nvidia/BossAI
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add native MoneyPrinter short video factory

Browse Source
This commit is contained in:
boos
2026-06-02 15:26:42 +08:00
parent 056e9fb651
commit b29c7ffa7a
6 changed files with 1226 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/mod.rs
+2
View File
@@ -23,6 +23,7 @@ mod digital_human;
mod events;
mod health;
mod monitor;
mod money_printer;
mod oauth;
mod organizations;
mod picture_agent;
@@ -46,6 +47,7 @@ pub fn build_router(state: AppState) -> Router {
.merge(admin::routes(state.clone()))
.merge(proxy::routes())
.merge(picture_agent::routes())
.merge(money_printer::routes())
.merge(comfyui_agent::routes())
.merge(digital_human::routes())
.merge(comfyui::routes())
src/routes/money_printer.rs
+723
View File
@@ -0,0 +1,723 @@
use crate::{
auth::jwt::validate_jwt,
db::models::{background_task::BackgroundTask, downstream_api_key::DownstreamApiKey},
error::{AppError, AppResult},
services::workspace::{self, WorkspaceAssetInput},
AppState,
};
use axum::{
extract::{Path, Query, State},
http::{header, HeaderMap},
routing::{get, post},
Json, Router,
};
use reqwest::Client;
use serde::Deserialize;
use serde_json::{json, Value};
use std::time::Duration;
const TASK_TYPE_MONEY_PRINTER: &str = "money-printer-video";
const VIDEO_POLL_INTERVAL_SECS: u64 = 6;
const VIDEO_MAX_POLLS: usize = 120;
#[derive(Debug, Clone)]
struct MoneyPrinterAuth {
user_id: Option<i64>,
role: String,
outbound_key: String,
key_name: String,
}
#[derive(Debug, Deserialize)]
struct JobListQuery {
limit: Option<i64>,
}
pub fn routes() -> Router<AppState> {
Router::new()
.route("/api/money-printer/status", get(status))
.route("/api/money-printer/jobs", get(list_jobs).post(create_job))
.route("/api/money-printer/jobs/clear", post(clear_jobs))
.route("/api/money-printer/jobs/{id}", get(get_job))
.route("/api/money-printer/jobs/{id}/cancel", post(cancel_job))
}
fn local_v1_base(state: &AppState) -> String {
let host = if state.config.server.host == "0.0.0.0" || state.config.server.host == "::" {
"127.0.0.1".to_string()
} else {
state.config.server.host.clone()
};
format!("http://{}:{}/v1", host, state.config.server.port)
}
fn extract_bearer_token(headers: &HeaderMap) -> Option<&str> {
headers
.get(header::AUTHORIZATION)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.strip_prefix("Bearer ").or_else(|| value.strip_prefix("bearer ")))
.map(str::trim)
.filter(|token| !token.is_empty())
}
async fn user_downstream_key(state: &AppState, user_id: i64) -> Result<DownstreamApiKey, String> {
sqlx::query_as::<_, DownstreamApiKey>(
"SELECT *
FROM downstream_api_keys
WHERE user_id = ?
AND COALESCE(enabled, 1) = 1
AND (
expires_at IS NULL
OR expires_at = ''
OR datetime(expires_at) > datetime('now')
)
ORDER BY updated_at DESC, created_at DESC, id DESC
LIMIT 1",
)
.bind(user_id)
.fetch_optional(&*state.db)
.await
.map_err(|err| err.to_string())?
.ok_or_else(|| "当前用户没有可用的 API Key，请先在用户中心创建或启用 API Key".to_string())
}
async fn resolve_auth(headers: &HeaderMap, state: &AppState) -> Result<MoneyPrinterAuth, AppError> {
let token = extract_bearer_token(headers)
.ok_or_else(|| AppError::Unauthorized("Missing Authorization header".into()))?;
if token == state.config.auth.auth_token || token == state.config.auth.proxy_token {
return Ok(MoneyPrinterAuth {
user_id: None,
role: "admin".to_string(),
outbound_key: state.config.auth.proxy_token.clone(),
key_name: "global-proxy".to_string(),
});
}
let claims = validate_jwt(&state.config.auth.jwt_secret, token)
.map_err(|_| AppError::Unauthorized("Invalid or expired token".into()))?;
if claims.role != "admin" && claims.role != "user" {
return Err(AppError::Forbidden("Forbidden".into()));
}
let user_id = claims
.sub
.parse::<i64>()
.map_err(|_| AppError::Unauthorized("Invalid user ID in token".into()))?;
match user_downstream_key(state, user_id).await {
Ok(key) => Ok(MoneyPrinterAuth {
user_id: Some(user_id),
role: claims.role,
outbound_key: key.key,
key_name: key.name,
}),
Err(_) if claims.role == "admin" => Ok(MoneyPrinterAuth {
user_id: Some(user_id),
role: claims.role,
outbound_key: state.config.auth.proxy_token.clone(),
key_name: "global-proxy".to_string(),
}),
Err(message) => Err(AppError::Forbidden(message)),
}
}
async fn status(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json<Value>> {
let auth = resolve_auth(&headers, &state).await?;
Ok(Json(json!({
"success": true,
"mode": "boosapi-native",
"keyName": auth.key_name,
"capabilities": [
"topic_to_script",
"boosapi_chat_completion",
"boosapi_text_to_video",
"workspace_archive",
"background_jobs"
]
})))
}
fn string_from(body: &Value, keys: &[&str]) -> Option<String> {
keys.iter()
.find_map(|key| body.get(*key).and_then(Value::as_str))
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn task_visible_to_auth(task: &BackgroundTask, auth: &MoneyPrinterAuth) -> bool {
if auth.role == "admin" && auth.user_id.is_none() {
return true;
}
let result = task
.result_json
.as_deref()
.and_then(|value| serde_json::from_str::<Value>(value).ok())
.unwrap_or(Value::Null);
result.get("userId").and_then(Value::as_i64) == auth.user_id
}
fn terminal_status(status: &str) -> bool {
matches!(status, "succeeded" | "failed" | "cancelled")
}
fn task_to_json(task: BackgroundTask) -> Value {
let result = task
.result_json
.as_deref()
.and_then(|value| serde_json::from_str::<Value>(value).ok())
.unwrap_or(Value::Null);
json!({
"id": task.id,
"type": task.task_type,
"title": task.title,
"status": task.status,
"message": task.message,
"error": task.error,
"progress": task.progress.unwrap_or(0),
"progressTotal": task.progress_total.unwrap_or(5),
"result": result,
"createdAt": task.created_at,
"updatedAt": task.updated_at,
"startedAt": task.started_at,
"finishedAt": task.finished_at,
})
}
async fn load_job(state: &AppState, id: &str, auth: &MoneyPrinterAuth) -> AppResult<Value> {
let task: BackgroundTask =
sqlx::query_as("SELECT * FROM background_tasks WHERE id = ? AND task_type = ? LIMIT 1")
.bind(id)
.bind(TASK_TYPE_MONEY_PRINTER)
.fetch_optional(&*state.db)
.await?
.ok_or_else(|| AppError::NotFound("MoneyPrinter job not found".into()))?;
if !task_visible_to_auth(&task, auth) {
return Err(AppError::NotFound("MoneyPrinter job not found".into()));
}
Ok(task_to_json(task))
}
async fn update_progress(state: &AppState, id: &str, progress: i64, message: &str) -> AppResult<()> {
sqlx::query(
"UPDATE background_tasks
SET progress = ?, message = ?, updated_at = datetime('now')
WHERE id = ? AND status IN ('pending', 'running')",
)
.bind(progress)
.bind(message)
.bind(id)
.execute(&*state.db)
.await?;
Ok(())
}
async fn is_cancelled(state: &AppState, id: &str) -> AppResult<bool> {
let status: Option<String> =
sqlx::query_scalar("SELECT status FROM background_tasks WHERE id = ? LIMIT 1")
.bind(id)
.fetch_optional(&*state.db)
.await?;
Ok(status.as_deref() == Some("cancelled"))
}
fn http_client(timeout_secs: u64) -> Client {
Client::builder()
.timeout(Duration::from_secs(timeout_secs))
.build()
.unwrap_or_else(|_| Client::new())
}
fn script_prompt(topic: &str, body: &Value) -> String {
let language = string_from(body, &["language"]).unwrap_or_else(|| "中文".to_string());
let style = string_from(body, &["style"]).unwrap_or_else(|| "爆款短视频，节奏紧凑，适合口播".to_string());
let duration = string_from(body, &["duration"]).unwrap_or_else(|| "30".to_string());
let audience = string_from(body, &["audience"]).unwrap_or_else(|| "泛大众用户".to_string());
let extra = string_from(body, &["requirements", "extra"]).unwrap_or_default();
format!(
"你是 MoneyPrinterTurbo 风格的短视频编导。请基于主题生成一个可直接用于 AI 文生视频的成片方案。\n\
主题：{topic}\n语言：{language}\n目标受众：{audience}\n时长：约 {duration} 秒\n风格：{style}\n额外要求：{extra}\n\n\
输出 JSON，字段必须包含：title、hook、script、video_prompt、hashtags、scenes。\n\
script 是完整口播文案；video_prompt 是英文或中英混合的视频生成提示词，包含镜头、场景、人物/产品、光线、运动、质感；scenes 是 3-6 个分镜对象数组。不要输出 Markdown。"
)
}
fn extract_json_object(text: &str) -> Option<Value> {
if let Ok(value) = serde_json::from_str::<Value>(text) {
return Some(value);
}
let start = text.find('{')?;
let end = text.rfind('}')?;
serde_json::from_str::<Value>(&text[start..=end]).ok()
}
async fn generate_script(
state: &AppState,
auth: &MoneyPrinterAuth,
body: &Value,
) -> AppResult<Value> {
let topic = string_from(body, &["topic", "prompt"]).ok_or_else(|| {
AppError::BadRequest("topic is required".into())
})?;
let script_model = string_from(body, &["scriptModel", "script_model"])
.unwrap_or_else(|| "gpt-5.4".to_string());
let payload = json!({
"model": script_model,
"messages": [
{ "role": "system", "content": "你是短视频自动化生产助手，只输出严格 JSON。" },
{ "role": "user", "content": script_prompt(&topic, body) }
],
"temperature": 0.7,
"stream": false
});
let response = http_client(120)
.post(format!("{}/chat/completions", local_v1_base(state)))
.bearer_auth(&auth.outbound_key)
.json(&payload)
.send()
.await
.map_err(|err| AppError::UpstreamError(format!("生成短视频脚本失败: {err}")))?;
let status = response.status();
let value: Value = response
.json()
.await
.map_err(|err| AppError::UpstreamError(format!("解析脚本响应失败: {err}")))?;
if !status.is_success() {
return Err(AppError::UpstreamError(value.to_string()));
}
let content = value
.get("choices")
.and_then(Value::as_array)
.and_then(|choices| choices.first())
.and_then(|choice| choice.get("message"))
.and_then(|message| message.get("content"))
.and_then(Value::as_str)
.unwrap_or_default();
let parsed = extract_json_object(content).unwrap_or_else(|| {
json!({
"title": topic,
"hook": "",
"script": content,
"video_prompt": content,
"hashtags": [],
"scenes": []
})
});
Ok(json!({
"model": script_model,
"raw": value,
"content": content,
"plan": parsed,
}))
}
fn video_url_from_response(value: &Value) -> Option<String> {
let keys = ["url", "video", "video_url", "output_url", "content_url"];
for key in keys {
if let Some(url) = value.get(key).and_then(Value::as_str) {
if url.starts_with("http") || url.starts_with('/') {
return Some(url.to_string());
}
}
}
if let Some(data) = value.get("data").and_then(Value::as_array) {
for item in data {
if let Some(url) = video_url_from_response(item) {
return Some(url);
}
}
}
if let Some(output) = value.get("output") {
if let Some(url) = video_url_from_response(output) {
return Some(url);
}
}
None
}
fn video_task_id_from_response(value: &Value) -> Option<String> {
let keys = ["id", "task_id", "taskId", "request_id", "job_id"];
keys.iter()
.find_map(|key| value.get(*key).and_then(Value::as_str))
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| {
value
.get("data")
.and_then(Value::as_array)
.and_then(|data| data.first())
.and_then(video_task_id_from_response)
})
}
async fn submit_video(
state: &AppState,
auth: &MoneyPrinterAuth,
body: &Value,
plan: &Value,
) -> AppResult<Value> {
let model = string_from(body, &["videoModel", "video_model", "model"])
.unwrap_or_else(|| "seedance-1-0-lite-t2v".to_string());
let prompt = string_from(body, &["videoPrompt", "video_prompt"]).unwrap_or_else(|| {
plan.get("video_prompt")
.and_then(Value::as_str)
.or_else(|| plan.get("script").and_then(Value::as_str))
.unwrap_or_default()
.to_string()
});
if prompt.trim().is_empty() {
return Err(AppError::BadRequest("video prompt is empty".into()));
}
let size = string_from(body, &["size", "resolution"]).unwrap_or_else(|| "1080p".to_string());
let ratio = string_from(body, &["ratio", "aspectRatio", "aspect_ratio"]).unwrap_or_else(|| "9:16".to_string());
let duration = body
.get("duration")
.and_then(Value::as_i64)
.or_else(|| string_from(body, &["duration"]).and_then(|value| value.parse::<i64>().ok()))
.unwrap_or(30);
let payload = json!({
"model": model,
"prompt": prompt,
"size": size,
"ratio": ratio,
"duration": duration,
"metadata": {
"source": "money-printer",
"script": plan.get("script").cloned().unwrap_or(Value::Null),
"title": plan.get("title").cloned().unwrap_or(Value::Null)
}
});
let response = http_client(180)
.post(format!("{}/videos", local_v1_base(state)))
.bearer_auth(&auth.outbound_key)
.json(&payload)
.send()
.await
.map_err(|err| AppError::UpstreamError(format!("提交视频生成失败: {err}")))?;
let status = response.status();
let value: Value = response
.json()
.await
.map_err(|err| AppError::UpstreamError(format!("解析视频响应失败: {err}")))?;
if !status.is_success() {
return Err(AppError::UpstreamError(value.to_string()));
}
Ok(json!({
"model": model,
"request": payload,
"response": value,
}))
}
async fn poll_video(
state: &AppState,
auth: &MoneyPrinterAuth,
model: &str,
task_id: &str,
) -> AppResult<Value> {
for _ in 0..VIDEO_MAX_POLLS {
let response = http_client(60)
.get(format!("{}/videos/{}", local_v1_base(state), task_id))
.query(&[("model", model)])
.bearer_auth(&auth.outbound_key)
.send()
.await
.map_err(|err| AppError::UpstreamError(format!("查询视频任务失败: {err}")))?;
let status = response.status();
let value: Value = response
.json()
.await
.map_err(|err| AppError::UpstreamError(format!("解析视频任务失败: {err}")))?;
if !status.is_success() {
return Err(AppError::UpstreamError(value.to_string()));
}
if video_url_from_response(&value).is_some() {
return Ok(value);
}
let state_text = value
.get("status")
.or_else(|| value.get("state"))
.and_then(Value::as_str)
.unwrap_or_default()
.to_ascii_lowercase();
if matches!(state_text.as_str(), "failed" | "error" | "cancelled" | "canceled") {
return Err(AppError::UpstreamError(value.to_string()));
}
tokio::time::sleep(Duration::from_secs(VIDEO_POLL_INTERVAL_SECS)).await;
}
Err(AppError::GatewayTimeout("视频生成任务等待超时".into()))
}
async fn archive_video(
state: &AppState,
auth: &MoneyPrinterAuth,
title: &str,
prompt: &str,
model: &str,
video_url: &str,
metadata: Value,
) -> Option<Value> {
let input = WorkspaceAssetInput {
user_id: auth.user_id,
source: "money-printer".to_string(),
kind: "video".to_string(),
title: Some(title.to_string()),
prompt: Some(prompt.to_string()),
model: Some(model.to_string()),
remote_url: Some(video_url.to_string()),
metadata,
};
match workspace::record_remote_asset(state, input).await {
Ok(asset) => Some(json!({
"publicId": asset.public_id,
"kind": "video",
"status": asset.status,
"remoteUrl": asset.remote_url,
"fileName": asset.file_name,
})),
Err(err) => Some(json!({
"kind": "video",
"status": "failed",
"remoteUrl": video_url,
"error": err.to_string(),
})),
}
}
async fn create_job(
State(state): State<AppState>,
headers: HeaderMap,
Json(body): Json<Value>,
) -> AppResult<Json<Value>> {
let auth = resolve_auth(&headers, &state).await?;
let topic = string_from(&body, &["topic", "prompt"])
.ok_or_else(|| AppError::BadRequest("topic is required".into()))?;
let task_id = format!("mp_{}", uuid::Uuid::new_v4().simple());
let title = topic.chars().take(40).collect::<String>();
let initial_result = json!({
"userId": auth.user_id,
"role": auth.role,
"keyName": auth.key_name,
"topic": topic,
"request": body,
});
sqlx::query(
"INSERT INTO background_tasks
(id, task_type, title, status, message, dedupe_key, result_json, progress, progress_total, started_at)
VALUES (?, ?, ?, 'running', '短视频任务已创建', ?, ?, 1, 5, datetime('now'))",
)
.bind(&task_id)
.bind(TASK_TYPE_MONEY_PRINTER)
.bind(if title.trim().is_empty() { "短视频工厂任务" } else { &title })
.bind(format!("money-printer:{}", auth.user_id.map(|id| id.to_string()).unwrap_or_else(|| "admin".into())))
.bind(initial_result.to_string())
.execute(&*state.db)
.await?;
let state_for_task = state.clone();
let task_id_for_spawn = task_id.clone();
let auth_for_task = auth.clone();
tokio::spawn(async move {
let result = async {
update_progress(&state_for_task, &task_id_for_spawn, 2, "正在生成脚本和分镜").await?;
let script = generate_script(&state_for_task, &auth_for_task, &body).await?;
if is_cancelled(&state_for_task, &task_id_for_spawn).await? {
return Ok(());
}
let plan = script.get("plan").cloned().unwrap_or(Value::Null);
let mid_result = json!({
"userId": auth_for_task.user_id,
"role": auth_for_task.role,
"keyName": auth_for_task.key_name,
"topic": topic,
"request": body,
"script": script,
});
sqlx::query("UPDATE background_tasks SET result_json = ?, updated_at = datetime('now') WHERE id = ?")
.bind(mid_result.to_string())
.bind(&task_id_for_spawn)
.execute(&*state_for_task.db)
.await?;
update_progress(&state_for_task, &task_id_for_spawn, 3, "正在提交 BoosAPI 视频模型").await?;
let submitted = submit_video(&state_for_task, &auth_for_task, &body, &plan).await?;
if is_cancelled(&state_for_task, &task_id_for_spawn).await? {
return Ok(());
}
let model = submitted
.get("model")
.and_then(Value::as_str)
.unwrap_or("default")
.to_string();
let first_response = submitted.get("response").cloned().unwrap_or(Value::Null);
let final_video_response = if video_url_from_response(&first_response).is_some() {
first_response
} else if let Some(video_task_id) = video_task_id_from_response(&first_response) {
update_progress(&state_for_task, &task_id_for_spawn, 4, "视频生成中，正在轮询上游任务").await?;
poll_video(&state_for_task, &auth_for_task, &model, &video_task_id).await?
} else {
first_response
};
let video_url = video_url_from_response(&final_video_response);
let prompt = submitted
.get("request")
.and_then(|request| request.get("prompt"))
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let title = plan
.get("title")
.and_then(Value::as_str)
.unwrap_or("短视频工厂成片")
.to_string();
let mut assets = Vec::new();
if let Some(url) = video_url.as_deref() {
update_progress(&state_for_task, &task_id_for_spawn, 5, "视频完成，正在归档到工作空间").await?;
if let Some(asset) = archive_video(
&state_for_task,
&auth_for_task,
&title,
&prompt,
&model,
url,
json!({ "moneyPrinter": { "topic": topic, "plan": plan, "response": final_video_response } }),
)
.await
{
assets.push(asset);
}
}
let final_result = json!({
"userId": auth_for_task.user_id,
"role": auth_for_task.role,
"keyName": auth_for_task.key_name,
"topic": topic,
"request": body,
"script": script,
"video": submitted,
"finalVideoResponse": final_video_response,
"videoUrl": video_url,
"assets": assets,
});
sqlx::query(
"UPDATE background_tasks
SET status = 'succeeded', message = '短视频生成完成', result_json = ?, progress = 5,
updated_at = datetime('now'), finished_at = datetime('now')
WHERE id = ? AND status IN ('pending', 'running')",
)
.bind(final_result.to_string())
.bind(&task_id_for_spawn)
.execute(&*state_for_task.db)
.await?;
Ok::<(), AppError>(())
}
.await;
if let Err(err) = result {
let _ = sqlx::query(
"UPDATE background_tasks
SET status = 'failed', message = '短视频生成失败', error = ?, updated_at = datetime('now'), finished_at = datetime('now')
WHERE id = ? AND status IN ('pending', 'running')",
)
.bind(err.to_string())
.bind(&task_id_for_spawn)
.execute(&*state_for_task.db)
.await;
}
});
let job = load_job(&state, &task_id, &auth).await?;
Ok(Json(json!({ "success": true, "job": job })))
}
async fn get_job(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<String>,
) -> AppResult<Json<Value>> {
let auth = resolve_auth(&headers, &state).await?;
let job = load_job(&state, &id, &auth).await?;
Ok(Json(json!({ "success": true, "job": job })))
}
async fn list_jobs(
State(state): State<AppState>,
headers: HeaderMap,
Query(query): Query<JobListQuery>,
) -> AppResult<Json<Value>> {
let auth = resolve_auth(&headers, &state).await?;
let limit = query.limit.unwrap_or(20).clamp(1, 100);
let tasks: Vec<BackgroundTask> = sqlx::query_as(
"SELECT * FROM background_tasks
WHERE task_type = ?
ORDER BY created_at DESC
LIMIT ?",
)
.bind(TASK_TYPE_MONEY_PRINTER)
.bind(limit * 3)
.fetch_all(&*state.db)
.await?;
let jobs = tasks
.into_iter()
.filter(|task| task_visible_to_auth(task, &auth))
.take(limit as usize)
.map(task_to_json)
.collect::<Vec<_>>();
Ok(Json(json!({ "success": true, "jobs": jobs })))
}
async fn cancel_job(
State(state): State<AppState>,
headers: HeaderMap,
Path(id): Path<String>,
) -> AppResult<Json<Value>> {
let auth = resolve_auth(&headers, &state).await?;
let task: BackgroundTask =
sqlx::query_as("SELECT * FROM background_tasks WHERE id = ? AND task_type = ? LIMIT 1")
.bind(&id)
.bind(TASK_TYPE_MONEY_PRINTER)
.fetch_optional(&*state.db)
.await?
.ok_or_else(|| AppError::NotFound("MoneyPrinter job not found".into()))?;
if !task_visible_to_auth(&task, &auth) {
return Err(AppError::NotFound("MoneyPrinter job not found".into()));
}
if terminal_status(&task.status) {
return Ok(Json(json!({ "success": true, "job": task_to_json(task) })));
}
sqlx::query(
"UPDATE background_tasks
SET status = 'cancelled', message = '已取消', updated_at = datetime('now'), finished_at = datetime('now')
WHERE id = ? AND task_type = ? AND status IN ('pending', 'running')",
)
.bind(&id)
.bind(TASK_TYPE_MONEY_PRINTER)
.execute(&*state.db)
.await?;
let job = load_job(&state, &id, &auth).await?;
Ok(Json(json!({ "success": true, "job": job })))
}
async fn clear_jobs(State(state): State<AppState>, headers: HeaderMap) -> AppResult<Json<Value>> {
let auth = resolve_auth(&headers, &state).await?;
let tasks: Vec<BackgroundTask> = sqlx::query_as(
"SELECT * FROM background_tasks WHERE task_type = ? ORDER BY created_at DESC",
)
.bind(TASK_TYPE_MONEY_PRINTER)
.fetch_all(&*state.db)
.await?;
let mut cleared = 0_i64;
for task in tasks {
if !task_visible_to_auth(&task, &auth) || !terminal_status(&task.status) {
continue;
}
let affected = sqlx::query("DELETE FROM background_tasks WHERE id = ? AND task_type = ?")
.bind(&task.id)
.bind(TASK_TYPE_MONEY_PRINTER)
.execute(&*state.db)
.await?
.rows_affected();
cleared += i64::try_from(affected).unwrap_or(0);
}
Ok(Json(json!({ "success": true, "cleared": cleared })))
}