task schema + queries 2.0 wip

This commit is contained in:
Brandon Thomas committed 2025-10-08 15:06:09 -04:00
1 parent 0fec772d9a
commit 68caef8e1f
11 files changed
+211 -3

No files matched your search

@@ -0,0 +1,20 @@
-- noinspection SqlDialectInspectionForFile
-- noinspection SqlNoDataSourceInspectionForFile
-- noinspection SqlResolveForFile
select
u.display_name,
count(*) as sessions,
a.session_duration_seconds / 60 / 60 as hours,
a.measurement_count,
u.email_address,
a.os_platform,
a.os_version,
u.created_at,
u.maybe_source,
a.last_active_at
from analytics_app_sessions as a
left outer join users as u
on a.user_token = u.token
group by a.user_token
order by hours desc;
@@ -9,6 +9,7 @@
-- tasks_v1.sqlite - initial version
-- tasks_v2.sqlite - added model_type (nullable)
-- tasks_v3.sqlite - added frontend_caller, comments
-- tasks_v4.sqlite - added is_dismissed_by_user
CREATE TABLE tasks (
-- Task auto-incrementing primary key.
@@ -50,6 +51,9 @@ CREATE TABLE tasks (
-- will be re-emitted back to the frontend.
frontend_subscriber_payload TEXT,
-- Whether the user has dismissed the task from view.
is_dismissed_by_user INTEGER NOT NULL DEFAULT 0,
created_at INTEGER NOT NULL DEFAULT (unixepoch('now')),
updated_at INTEGER NOT NULL DEFAULT (unixepoch('now')),
completed_at INTEGER DEFAULT NULL
@@ -7,4 +7,5 @@ pub mod load_without_cors_command;
pub mod media_files;
pub mod platform_info_command;
pub mod providers;
pub mod response;
pub mod response;
mod task_queue;
@@ -0,0 +1,85 @@
use crate::core::commands::response::shorthand::ResponseOrErrorMessage;
use crate::core::commands::response::success_response_wrapper::SerializeMarker;
use crate::core::events::generation_events::generation_enqueue_success_event::GenerationEnqueueSuccessEvent;
use crate::core::state::app_env_configs::app_env_configs::AppEnvConfigs;
use crate::core::state::data_dir::app_data_root::AppDataRoot;
use crate::core::state::provider_priority::{Provider, ProviderPriorityStore};
use crate::core::state::task_database::TaskDatabase;
use crate::services::fal::state::fal_credential_manager::FalCredentialManager;
use crate::services::fal::state::fal_task_queue::FalTaskQueue;
use crate::services::sora::state::sora_task_queue::SoraTaskQueue;
use crate::services::storyteller::state::storyteller_credential_manager::StorytellerCredentialManager;
use chrono::{DateTime, Utc};
use enums::common::generation_provider::GenerationProvider;
use enums::tauri::tasks::task_model_type::TaskModelType;
use enums::tauri::tasks::task_status::TaskStatus;
use enums::tauri::tasks::task_type::TaskType;
use errors::AnyhowResult;
use log::{error, info, warn};
use serde_derive::{Deserialize, Serialize};
use storyteller_client::endpoints::media_files::delete_media_file::delete_media_file;
use tauri::{AppHandle, State};
use tokens::tokens::media_files::MediaFileToken;
use tokens::tokens::sqlite::tasks::TaskId;
#[derive(Serialize)]
pub struct GetTaskQueueCommandResponse {
tasks: Vec<TaskQueueTask>,
}
#[derive(Serialize)]
pub struct TaskQueueTask {
pub id: TaskId,
pub task_status: TaskStatus,
pub task_type: TaskType,
pub model_type: Option<TaskModelType>,
pub provider: Option<GenerationProvider>,
pub provider_job_id: Option<String>,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
pub completed_at: Option<DateTime<Utc>>,
}
impl SerializeMarker for GetTaskQueueCommandResponse {}
#[tauri::command]
pub async fn get_task_queue_command(
app: AppHandle,
app_env_configs: State<'_, AppEnvConfigs>,
task_database: State<'_, TaskDatabase>,
) -> ResponseOrErrorMessage<GetTaskQueueCommandResponse> {
info!("get_task_queue_command called");
let result = handle_request(
&app,
&app_env_configs,
&task_database,
).await;
if let Err(err) = result {
error!("get_task_queue_command failed: {:?}", err);
return Err("get_task_queue_command failed".into())
}
Ok(GetTaskQueueCommandResponse{}.into())
}
pub async fn handle_request(
app: &AppHandle,
app_env_configs: &AppEnvConfigs,
task_database: State<'_, TaskDatabase>,
) -> AnyhowResult<()> {
let creds = storyteller_creds_manager.get_credentials()?;
let _result = delete_media_file(
&app_env_configs.storyteller_host,
creds.as_ref(),
&request.media_file_token
).await?;
// TODO: Send event to frontend that the file was deleted.
Ok(())
}
@@ -0,0 +1 @@
mod get_task_queue_command;
@@ -1,4 +1,5 @@
use crate::core::state::data_dir::trait_data_subdir::DataSubdir;
use crate::core::state::task_database_version::TASK_DATABASE_VERSION;
use std::path::{Path, PathBuf};
#[derive(Clone)]
@@ -24,7 +25,7 @@ impl AppStateDir {
pub fn get_tasks_sqlite_database_path(&self) -> PathBuf {
// NB: This must be incremented on any change to the database schema (including comments).
// The database houses ephemeral content and will be migrated automatically.
self.path.join("tasks_v3.sqlite")
self.path.join(format!("tasks_v{TASK_DATABASE_VERSION}.sqlite"))
}
pub fn get_window_size_config_file(&self) -> PathBuf {
@@ -7,4 +7,5 @@ pub mod expanduser;
pub mod os_platform;
pub mod provider_priority;
pub mod task_database;
pub mod task_database_version;
pub mod window;
@@ -0,0 +1,10 @@
/// Used to bump the task database schema version.
/// Migrations on customers' machines tend to fail on existing databases,
/// even if we instruct that the database file be deleted first.
/// It's easiest to just write to a new file and migrate from scratch.
/// We increment this whenever we make a change to the database schema.
/// This prevents deadlock on startup if the schema has changed from the
/// previous version.
pub const TASK_DATABASE_VERSION: u32 = 4;
@@ -0,0 +1,81 @@
use crate::connection::TaskDbConnection;
use crate::error::SqliteTasksError;
use enums::common::generation_provider::GenerationProvider;
use enums::tauri::tasks::task_model_type::TaskModelType;
use enums::tauri::tasks::task_status::TaskStatus;
use enums::tauri::tasks::task_type::TaskType;
use enums::tauri::ux::tauri_command_caller::TauriCommandCaller;
use tokens::tokens::sqlite::tasks::TaskId;
pub struct CreateTaskArgs<'a> {
pub status: TaskStatus,
pub task_type: TaskType,
pub model_type: Option<TaskModelType>,
pub provider: GenerationProvider,
pub provider_job_id: Option<&'a str>,
pub frontend_caller: Option<TauriCommandCaller>,
pub frontend_subscriber_id: Option<&'a str>,
pub frontend_subscriber_payload: Option<&'a str>,
}
pub struct TaskList {
pub tasks: Vec<TaskItem>,
}
pub struct TaskItem {
pub status: TaskStatus,
pub task_type: TaskType,
pub model_type: Option<TaskModelType>,
pub provider: GenerationProvider,
pub provider_job_id: Option<String>,
pub frontend_caller: Option<TauriCommandCaller>,
pub frontend_subscriber_id: Option<String>,
pub frontend_subscriber_payload: Option<String>,
}
pub async fn list_tasks_for_frontend(
db: & TaskDbConnection
) -> Result<TaskList, SqliteTasksError> {
let task_id = TaskId::generate();
//model_type,
//frontend_caller,
let query = sqlx::query_as!(
TaskItemRaw,
r#"
SELECT
id,
task_status,
task_type,
provider,
provider_job_id,
frontend_subscriber_id,
frontend_subscriber_payload
FROM tasks
"#);
let _r = query.fetch(db.get_pool()).await?;
Ok(())
}
//fn query(&self) -> Query<MySql, MySqlArguments> {
//}
struct TaskItemRaw {
id: String,
task_status: String,
task_type: String,
//model_type: Option<String>,
//id: TaskId,
//task_status: TaskStatus,
//task_type: TaskType,
//model_type: Option<TaskModelType>,
provider: Option<String>,
provider_job_id: Option<String>,
//frontend_caller: Option<TauriCommandCaller>,
frontend_subscriber_id: Option<String>,
frontend_subscriber_payload: Option<String>,
}
@@ -1,4 +1,5 @@
pub mod create_task;
pub mod list_tasks_by_provider_and_status;
pub mod list_tasks_by_provider_and_tokens;
pub mod update_task_status;
pub mod update_task_status;
mod list_tasks_for_frontend;
@@ -22,6 +22,9 @@ impl_enum_display_and_debug_using_to_str!(GenerationProvider);
impl_mysql_enum_coders!(GenerationProvider);
impl_mysql_from_row!(GenerationProvider);
// For Tauri
impl_sqlite_enum_coders!(GenerationProvider);
// NB: We can derive `sqlx::Type` instead of using `impl_mysql_enum_coders`
impl GenerationProvider {