starred/minne
1
0
Fork 0
mirror of https://github.com/perstarkse/minne.git synced 2026-01-11 20:50:24 +01:00
Code Issues Packages Projects Releases 10 Wiki Activity

fix: remove remnant job table, normalize taskstatus enum

Browse Source
This commit is contained in:
Per Stark
2025-06-27 23:18:16 +02:00
parent 43263fa77e
commit ec16f2100c
9 changed files with 33 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
common/migrations/20250627_231035_remove_job_table.surql Normal file
View File

@@ -0,0 +1 @@
REMOVE TABLE job;

2
common/migrations/definitions/20250627_120000_add_image_processing_settings.json
View File

@@ -1 +1 @@
{"schemas":"--- original\n+++ modified\n@@ -157,10 +157,12 @@\n DEFINE FIELD IF NOT EXISTS require_email_verification ON system_settings TYPE bool;\n DEFINE FIELD IF NOT EXISTS query_model ON system_settings TYPE string;\n DEFINE FIELD IF NOT EXISTS processing_model ON system_settings TYPE string;\n+DEFINE FIELD IF NOT EXISTS image_processing_model ON system_settings TYPE string;\n DEFINE FIELD IF NOT EXISTS embedding_model ON system_settings TYPE string;\n DEFINE FIELD IF NOT EXISTS embedding_dimensions ON system_settings TYPE int;\n DEFINE FIELD IF NOT EXISTS query_system_prompt ON system_settings TYPE string;\n DEFINE FIELD IF NOT EXISTS ingestion_system_prompt ON system_settings TYPE string;\n+DEFINE FIELD IF NOT EXISTS image_processing_prompt ON system_settings TYPE string;\n\n # Defines the schema for the 'text_chunk' table.\n\n","events":null}
{"schemas":"--- original\n+++ modified\n@@ -57,10 +57,7 @@\n DEFINE FIELD IF NOT EXISTS created_at ON ingestion_task TYPE string;\n DEFINE FIELD IF NOT EXISTS updated_at ON ingestion_task TYPE string;\n\n-# Custom fields from the IngestionTask struct\n-# IngestionPayload is complex, store as object\n DEFINE FIELD IF NOT EXISTS content ON ingestion_task TYPE object;\n-# IngestionTaskStatus can hold data (InProgress), store as object\n DEFINE FIELD IF NOT EXISTS status ON ingestion_task TYPE object;\n DEFINE FIELD IF NOT EXISTS user_id ON ingestion_task TYPE string;\n\n@@ -157,10 +154,12 @@\n DEFINE FIELD IF NOT EXISTS require_email_verification ON system_settings TYPE bool;\n DEFINE FIELD IF NOT EXISTS query_model ON system_settings TYPE string;\n DEFINE FIELD IF NOT EXISTS processing_model ON system_settings TYPE string;\n+DEFINE FIELD IF NOT EXISTS image_processing_model ON system_settings TYPE string;\n DEFINE FIELD IF NOT EXISTS embedding_model ON system_settings TYPE string;\n DEFINE FIELD IF NOT EXISTS embedding_dimensions ON system_settings TYPE int;\n DEFINE FIELD IF NOT EXISTS query_system_prompt ON system_settings TYPE string;\n DEFINE FIELD IF NOT EXISTS ingestion_system_prompt ON system_settings TYPE string;\n+DEFINE FIELD IF NOT EXISTS image_processing_prompt ON system_settings TYPE string;\n\n # Defines the schema for the 'text_chunk' table.\n\n","events":null}

3
common/schemas/ingestion_task.surql
View File

@@ -6,10 +6,7 @@ DEFINE TABLE IF NOT EXISTS ingestion_task SCHEMALESS;
DEFINE FIELD IF NOT EXISTS created_at ON ingestion_task TYPE string;
DEFINE FIELD IF NOT EXISTS updated_at ON ingestion_task TYPE string;
# Custom fields from the IngestionTask struct
# IngestionPayload is complex, store as object
DEFINE FIELD IF NOT EXISTS content ON ingestion_task TYPE object;
# IngestionTaskStatus can hold data (InProgress), store as object
DEFINE FIELD IF NOT EXISTS status ON ingestion_task TYPE object;
DEFINE FIELD IF NOT EXISTS user_id ON ingestion_task TYPE string;

17
common/src/storage/types/ingestion_task.rs
View File

@@ -7,6 +7,7 @@ use crate::{error::AppError, storage::db::SurrealDbClient, stored_object};
use super::ingestion_payload::IngestionPayload;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(tag = "name")]
pub enum IngestionTaskStatus {
Created,
InProgress {
@@ -14,7 +15,9 @@ pub enum IngestionTaskStatus {
last_attempt: DateTime<Utc>,
},
Completed,
Error(String),
Error {
message: String,
},
Cancelled,
}
@@ -85,10 +88,10 @@ impl IngestionTask {
.query(
"SELECT * FROM type::table($table)
WHERE
status = 'Created'
status.name = 'Created'
OR (
status.InProgress != NONE
AND status.InProgress.attempts < $max_attempts
status.name = 'InProgress'
AND status.attempts < $max_attempts
)
ORDER BY created_at ASC",
)
@@ -241,7 +244,9 @@ mod tests {
completed_task.status = IngestionTaskStatus::Completed;
let mut error_task = IngestionTask::new(payload.clone(), user_id.to_string()).await;
error_task.status = IngestionTaskStatus::Error("Test error".to_string());
error_task.status = IngestionTaskStatus::Error {
message: "Test error".to_string(),
};
// Store all tasks
db.store_item(created_task)
@@ -280,7 +285,7 @@ mod tests {
}
}
IngestionTaskStatus::Completed => "Completed",
IngestionTaskStatus::Error(_) => "Error",
IngestionTaskStatus::Error { .. } => "Error",
IngestionTaskStatus::Cancelled => "Cancelled",
})
.collect();

6
html-router/src/routes/content/handlers.rs
View File

@@ -7,7 +7,7 @@ use axum_htmx::{HxBoosted, HxRequest};
use serde::{Deserialize, Serialize};
use common::storage::types::{
conversation::Conversation, file_info::FileInfo, text_content::TextContent, user::User,
conversation::Conversation, file_info::FileInfo, text_content::TextContent, user::User, knowledge_entity::KnowledgeEntity, text_chunk::TextChunk,
};
use crate::{
@@ -138,6 +138,10 @@ pub async fn delete_text_content(
FileInfo::delete_by_id(&file_info.id, &state.db).await?;
}
// Delete related knowledge entities and text chunks
KnowledgeEntity::delete_by_source_id(&id, &state.db).await?;
TextChunk::delete_by_source_id(&id, &state.db).await?;
// Delete the text content
state.db.delete_item::<TextContent>(&id).await?;

10
html-router/src/routes/ingestion/handlers.rs
View File

@@ -186,9 +186,9 @@ pub async fn get_task_updates_stream(
format!("In progress, attempt {}", attempts)
}
IngestionTaskStatus::Completed => "Completed".to_string(),
IngestionTaskStatus::Error(ref err_msg) => {
IngestionTaskStatus::Error { message } => {
// Providing a user-friendly error message from the status
format!("Error: {}", err_msg)
format!("Error: {}", message)
}
IngestionTaskStatus::Cancelled => "Cancelled".to_string(),
};
@@ -197,9 +197,9 @@ pub async fn get_task_updates_stream(
// Check for terminal states to close the stream
match updated_task.status {
IngestionTaskStatus::Completed |
IngestionTaskStatus::Error(_) |
IngestionTaskStatus::Cancelled => {
IngestionTaskStatus::Completed
| IngestionTaskStatus::Error { .. }
| IngestionTaskStatus::Cancelled => {
// Send a specific event that HTMX uses to close the connection
// Send a event to reload the recent content
// Send a event to remove the loading indicatior

8
html-router/templates/dashboard/active_jobs.html
View File

@@ -19,10 +19,12 @@
</div>
<div>
<div class="[&:before]:content-['Status:_'] [&:before]:opacity-60">
{% if item.status.InProgress %}
In Progress, attempt {{item.status.InProgress.attempts}}
{% if item.status.name == "InProgress" %}
In Progress, attempt {{item.status.attempts}}
{% elif item.status.name == "Error" %}
Error: {{item.status.message}}
{% else %}
{{item.status}}
{{item.status.name}}
{% endif %}
</div>
<div class="text-xs font-semibold opacity-60">

4
ingestion-pipeline/src/lib.rs
View File

@@ -44,7 +44,7 @@ pub async fn run_worker_loop(
Action::Update => {
match notification.data.status {
IngestionTaskStatus::Completed
| IngestionTaskStatus::Error(_)
| IngestionTaskStatus::Error { .. }
| IngestionTaskStatus::Cancelled => {
info!(
"Skipping already completed/error/cancelled task: {}",
@@ -58,7 +58,7 @@ pub async fn run_worker_loop(
db.get_item::<IngestionTask>(&notification.data.id).await
{
match current_task.status {
IngestionTaskStatus::Error(_)
IngestionTaskStatus::Error { .. }
if attempts
< common::storage::types::ingestion_task::MAX_ATTEMPTS =>
{

4
ingestion-pipeline/src/pipeline.rs
View File

@@ -72,7 +72,9 @@ impl IngestionPipeline {
if current_attempts >= MAX_ATTEMPTS {
IngestionTask::update_status(
&task.id,
IngestionTaskStatus::Error(format!("Max attempts reached: {}", e)),
IngestionTaskStatus::Error {
message: format!("Max attempts reached: {}", e),
},
&self.db,
)
.await?;