diff --git a/apps/chat2db-web/src/legacy.rs b/apps/chat2db-web/src/legacy.rs index 1f07ee2..557d291 100644 --- a/apps/chat2db-web/src/legacy.rs +++ b/apps/chat2db-web/src/legacy.rs @@ -245,6 +245,43 @@ pub struct LegacyDriverResponse { pub driver_config_list: Vec, } +/// One database type offered by the retained Community connection form. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacySupportedDatabase { + pub db_type: String, + pub name: String, + pub support_database: bool, + pub support_schema: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub sql_dialect: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub jdbc_driver_class: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub url_sample: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub icon: Option, +} + +/// Identity-colour update posted by the retained Community connection tree. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyIdentityColorRequest { + pub id: LegacyIdentifier, + #[serde(default)] + pub identity_color: Option, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyIdentityColorResponse { + pub id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub identity_color: Option, + pub environment_id: u64, + pub environment: LegacyEnvironment, +} + #[derive(Clone, Default, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LegacyConnectionProperty { @@ -314,6 +351,8 @@ pub struct LegacyDatasourceResponse { pub support_schema: bool, pub has_secret: bool, pub revision: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub identity_color: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -438,6 +477,250 @@ pub struct LegacyTaskListQuery { pub task_status: String, } +/// Task-center query posted by the newer Community import/export panel. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportTaskQuery { + #[serde(default = "default_page_no")] + pub page_no: u32, + #[serde(default = "default_page_size")] + pub page_size: u32, + #[serde(default)] + pub status: String, +} + +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportTaskIdQuery { + pub task_id: LegacyIdentifier, +} + +/// Event-window query used by the task-center log view. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportEventQuery { + pub task_id: LegacyIdentifier, + #[serde(default)] + pub after_sequence: Option, + #[serde(default)] + pub before_sequence: Option, + #[serde(default)] + pub limit: Option, +} + +/// Accepted task id returned to the task-center panel. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportTaskAccepted { + pub task_id: i64, +} + +/// Active-transaction query used by the `MySQL` operations monitor. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyActiveTransactionQuery { + pub data_source_id: LegacyIdentifier, + #[serde(default)] + pub database_name: String, + #[serde(default)] + pub schema_name: String, +} + +/// One active transaction in the shape the monitor expects. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyActiveTransaction { + pub trx_id: String, + pub state: String, + pub started_at: i64, + pub age_seconds: u64, + pub isolation_level: String, + pub rows_locked: u64, + pub rows_modified: u64, + pub lock_structs: u64, + pub thread_id: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub user: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub host: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub db: Option, + pub query: String, + pub connection_inspection_sql: String, + /// Deep lock metadata is not collected yet, so the panel must not wait for it. + pub lock_metadata_state: String, +} + +/// Staged import-preview upload sent by the desktop bridge. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewLocalUploadRequest { + pub source_file: String, + #[serde(default)] + pub original_file_name: String, +} + +/// Import-preview request carrying a staged file id. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewFileRequest { + pub data_source_id: LegacyIdentifier, + #[serde(default)] + pub database_name: String, + #[serde(default)] + pub schema_name: String, + #[serde(default)] + pub table_name: String, + pub file_id: String, +} + +/// Column mapping submitted with an import-preview execution. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewMappingRequest { + #[serde(default)] + pub source_column: Option, + pub target_column: String, +} + +/// Import-preview execution posted after the mapping step. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewExecuteRequest { + pub data_source_id: LegacyIdentifier, + #[serde(default)] + pub database_name: String, + #[serde(default)] + pub schema_name: String, + #[serde(default)] + pub table_name: String, + pub file_id: String, + #[serde(default)] + pub mappings: Vec, + #[serde(default)] + pub unmapped_target: String, +} + +/// Mapping suggestion returned by the preview step. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewMapping { + pub source_column: String, + pub target_column: String, +} + +/// Target column metadata shown by the mapping step. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreviewColumn { + pub name: String, + pub data_type: String, + pub nullable: bool, + pub auto_increment: bool, + pub default_value: Option, + pub comment: Option, +} + +/// Preview payload rendered by the mapping step. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportPreview { + pub source_columns: Vec, + pub preview_data: Vec>, + pub target_table_name: String, + pub target_columns: Vec, + pub suggested_mapping: Vec, + pub preview_limit: u32, +} + +/// Submission posted by the task-center import/export panel. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportSubmitRequest { + pub data_source_id: LegacyIdentifier, + #[serde(default)] + pub database_name: String, + #[serde(default)] + pub schema_name: String, + #[serde(default)] + pub table_name: String, + #[serde(default)] + pub table_names: Vec, + #[serde(default)] + pub task_type: String, + #[serde(default)] + pub task_name: String, + #[serde(default)] + pub format: String, + #[serde(default)] + pub scope: String, + #[serde(default = "default_true")] + pub contain_data: bool, + #[serde(default = "default_true")] + pub contains_header: bool, + #[serde(default)] + pub source_file: String, + #[serde(default)] + pub file_id: String, + #[serde(default)] + pub export_path: String, +} + +/// One task-center log event in the shape the newer Community panel expects. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportEvent { + pub event_id: i64, + pub task_id: i64, + pub sequence: u64, + pub level: String, + pub code: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub stage: Option, + pub message: String, + pub created_at: i64, +} + +/// One task-center row in the shape the newer Community panel expects. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportTask { + pub id: i64, + pub name: String, + #[serde(rename = "type")] + pub task_type: String, + pub status: String, + pub progress: u32, + #[serde(skip_serializing_if = "Option::is_none")] + pub stage: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub progress_message: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub target: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error_code: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error_message: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub artifact_id: Option, + pub created_at: i64, + #[serde(skip_serializing_if = "Option::is_none")] + pub started_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub finished_at: Option, + pub updated_at: i64, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyImportExportTaskTarget { + pub data_source_id: String, + pub database_name: String, + pub schema_name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub table_name: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LegacyTransferTask { @@ -858,6 +1141,20 @@ pub struct LegacyTableCopyRequest { pub copy_data: bool, } +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyTableCopyPrepareQuery { + pub data_source_id: LegacyIdentifier, + #[serde(default, deserialize_with = "deserialize_string_or_default")] + pub database_name: String, + #[serde(default, deserialize_with = "deserialize_string_or_default")] + pub schema_name: String, + #[serde(default, deserialize_with = "deserialize_string_or_default")] + pub database_type: String, + #[serde(default, deserialize_with = "deserialize_string_or_default")] + pub table_name: String, +} + #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase")] pub struct LegacyDatabaseDefinitionRequest { @@ -1949,6 +2246,38 @@ pub fn drivers(application: &Application, requested_type: &str) -> LegacyDriverR } } +/// Rejects uploaded JDBC drivers: this runtime only loads verified driver packs. +pub(crate) fn custom_driver_upload_unsupported() -> LegacyFailure { + LegacyFailure { + code: "unsupported_custom_driver_upload".to_owned(), + message: + "This runtime loads verified driver packs and cannot accept an uploaded JDBC driver yet" + .to_owned(), + } +} + +pub(crate) fn list_supported_databases(application: &Application) -> Vec { + application + .list_drivers() + .items + .into_iter() + .map(|driver| { + let db_type = database_type_for_driver(&driver); + let support_schema = !matches!(db_type.as_str(), "MYSQL" | "SQLITE"); + LegacySupportedDatabase { + db_type, + name: driver.name, + support_database: true, + support_schema, + sql_dialect: None, + jdbc_driver_class: Some(driver.driver_class), + url_sample: None, + icon: None, + } + }) + .collect() +} + pub(crate) async fn list_community_dashboards( application: &Application, query: CommunityDashboardListQuery, @@ -2905,6 +3234,511 @@ pub(crate) async fn legacy_transfer_task_download( .await?) } +fn import_export_task_status(status: TransferTaskStatus) -> &'static str { + match status { + TransferTaskStatus::Queued => "PENDING", + TransferTaskStatus::Running => "RUNNING", + TransferTaskStatus::Succeeded => "SUCCESS", + TransferTaskStatus::Failed | TransferTaskStatus::Interrupted => "FAILED", + TransferTaskStatus::Cancelled => "CANCELLED", + } +} + +fn import_export_status_filter(value: &str) -> LegacyResult> { + match value.trim().to_ascii_uppercase().as_str() { + "" => Ok(Vec::new()), + "PENDING" => Ok(vec![TransferTaskStatus::Queued]), + "RUNNING" => Ok(vec![TransferTaskStatus::Running]), + "SUCCESS" => Ok(vec![TransferTaskStatus::Succeeded]), + "FAILED" => Ok(vec![ + TransferTaskStatus::Failed, + TransferTaskStatus::Interrupted, + ]), + "CANCELLED" => Ok(vec![TransferTaskStatus::Cancelled]), + _ => Err(LegacyFailure::invalid( + "invalid_task_status", + "status is not supported", + )), + } +} + +fn import_export_task_type(kind: TransferTaskKind) -> &'static str { + match kind { + TransferTaskKind::ImportFile => "DATA_FILE_IMPORT", + TransferTaskKind::ExportSql => "SQL_EXPORT", + TransferTaskKind::ExportFile => "TABLE_DATA_EXPORT", + } +} + +fn legacy_epoch_millis(value: &str) -> i64 { + value.parse().unwrap_or_default() +} + +fn import_export_task(task: TransferTask) -> LegacyImportExportTask { + let error_message = task + .error_log + .lines() + .map(str::trim) + .rfind(|line| !line.is_empty()) + .map(str::to_owned); + let progress = transfer_progress_percent(&task).parse().unwrap_or(0); + LegacyImportExportTask { + id: task.id, + name: task.task_name, + task_type: import_export_task_type(task.kind).to_owned(), + status: import_export_task_status(task.status).to_owned(), + progress, + stage: None, + progress_message: Some(task.progress_description).filter(|value| !value.is_empty()), + target: Some(LegacyImportExportTaskTarget { + data_source_id: task.datasource_id, + database_name: task.database_name, + schema_name: task.schema_name, + table_name: task.table_name, + }), + error_code: None, + error_message, + artifact_id: task.artifact_id, + created_at: legacy_epoch_millis(&task.created_at_ms), + started_at: None, + finished_at: task.finished_at_ms.as_deref().map(legacy_epoch_millis), + updated_at: legacy_epoch_millis(&task.updated_at_ms), + } +} + +pub(crate) async fn list_import_export_tasks( + application: &Application, + query: &LegacyImportExportTaskQuery, +) -> LegacyResult> { + let statuses = import_export_status_filter(&query.status)?; + let page = application + .list_transfer_tasks_by_statuses(query.page_no, query.page_size, &statuses) + .await?; + Ok(LegacyPage { + total: usize::try_from(page.total).unwrap_or(usize::MAX), + has_next_page: u64::from(page.page_no).saturating_mul(u64::from(page.page_size)) + < page.total, + data: page.items.into_iter().map(import_export_task).collect(), + page_no: page.page_no, + page_size: page.page_size, + }) +} + +pub(crate) async fn get_import_export_task( + application: &Application, + query: &LegacyImportExportTaskIdQuery, +) -> LegacyResult { + let task = application + .transfer_task(legacy_transfer_task_id(&query.task_id)?) + .await?; + Ok(import_export_task(task)) +} + +pub(crate) async fn active_import_export_task_count( + application: &Application, +) -> LegacyResult { + let page = application + .list_transfer_tasks_by_statuses( + 1, + 100, + &[TransferTaskStatus::Queued, TransferTaskStatus::Running], + ) + .await?; + Ok(page.total) +} + +pub(crate) async fn import_export_task_artifact( + application: &Application, + query: &LegacyImportExportTaskIdQuery, +) -> LegacyResult { + legacy_transfer_task_download(application, &query.task_id).await +} + +pub(crate) async fn delete_import_export_task( + application: &Application, + query: &LegacyImportExportTaskIdQuery, +) -> LegacyResult<()> { + application + .delete_transfer_task(legacy_transfer_task_id(&query.task_id)?) + .await?; + Ok(()) +} + +pub(crate) async fn list_import_export_task_events( + application: &Application, + query: &LegacyImportExportEventQuery, +) -> LegacyResult> { + let events = application + .list_transfer_task_events( + legacy_transfer_task_id(&query.task_id)?, + query.after_sequence, + query.before_sequence, + query.limit, + ) + .await?; + Ok(events + .into_iter() + .map(|event| LegacyImportExportEvent { + event_id: event.id, + task_id: event.task_id, + sequence: event.sequence, + level: event.level, + code: event.code, + stage: event.stage, + message: event.message, + created_at: legacy_epoch_millis(&event.created_at_ms), + }) + .collect()) +} + +const IMPORT_PREVIEW_ROW_LIMIT: u32 = 100; + +fn import_preview_artifact_format( + file_name: &str, +) -> LegacyResult<(&'static str, &'static str, &'static str)> { + let extension = file_name + .rsplit('.') + .next() + .unwrap_or_default() + .to_ascii_lowercase(); + match extension.as_str() { + "csv" => Ok(("text/csv", "CSV", "csv")), + "xls" => Ok(("application/vnd.ms-excel", "XLS", "xls")), + "xlsx" => Ok(( + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "XLSX", + "xlsx", + )), + "sql" => Ok(("application/sql; charset=utf-8", "SQL", "sql")), + _ => Err(LegacyFailure::invalid( + "invalid_import_file", + "The staged import file must be CSV, XLS, XLSX, or SQL", + )), + } +} + +fn import_preview_transfer_format(value: &str) -> LegacyResult { + match value.trim().to_ascii_uppercase().as_str() { + "CSV" => Ok(TransferFileFormat::Csv), + "XLS" => Ok(TransferFileFormat::Xls), + "XLSX" => Ok(TransferFileFormat::Xlsx), + "SQL" => Ok(TransferFileFormat::Sql), + other => Err(LegacyFailure { + code: "unsupported_import_file".to_owned(), + message: format!("Unsupported import preview format: {other}"), + }), + } +} + +async fn stage_import_preview_bytes( + application: &Application, + file_name: String, + content: Vec, +) -> LegacyResult { + let (media_type, format_name, extension) = import_preview_artifact_format(&file_name)?; + let expires_at_ms = i64::try_from(unix_epoch_millis()) + .unwrap_or(i64::MAX) + .saturating_add(LEGACY_IMPORT_UPLOAD_TTL_MS); + let storage = legacy_storage(application)?; + legacy_storage_call(move || { + let mut writer = storage.begin_transfer_artifact( + None, + &file_name, + media_type, + format_name, + extension, + Some(expires_at_ms), + )?; + writer.write_all(&content).map_err(|_| { + StorageError::InvalidTransfer("the staged import file could not be stored") + })?; + Ok(writer.finish()?.id) + }) + .await +} + +pub(crate) async fn stage_import_preview_upload( + application: &Application, + multipart: Multipart, +) -> LegacyResult { + let upload = read_legacy_multipart_file(multipart).await?; + stage_import_preview_bytes(application, upload.file_name, upload.content).await +} + +pub(crate) async fn stage_import_preview_local_file( + application: &Application, + request: &LegacyImportPreviewLocalUploadRequest, +) -> LegacyResult { + let source = request.source_file.trim().to_owned(); + if source.is_empty() { + return Err(LegacyFailure::invalid( + "invalid_import_file", + "sourceFile is required", + )); + } + let metadata = tokio::fs::metadata(&source) + .await + .map_err(|_| LegacyFailure { + code: "import_file_not_found".to_owned(), + message: "The selected import file could not be read".to_owned(), + })?; + if !metadata.is_file() || metadata.len() > MAX_LEGACY_DATASOURCE_IMPORT_BYTES as u64 { + return Err(LegacyFailure::invalid( + "invalid_import_file", + "sourceFile must be a readable file within the upload size limit", + )); + } + let content = tokio::fs::read(&source).await.map_err(|_| LegacyFailure { + code: "import_file_not_found".to_owned(), + message: "The selected import file could not be read".to_owned(), + })?; + let file_name = non_blank(&request.original_file_name).unwrap_or_else(|| { + source + .rsplit(['/', '\\']) + .next() + .unwrap_or("import.dat") + .to_owned() + }); + stage_import_preview_bytes(application, file_name, content).await +} + +pub(crate) async fn import_preview_sheets( + application: &Application, + request: &LegacyImportPreviewFileRequest, +) -> LegacyResult> { + Ok(application.import_preview_sheets(&request.file_id).await?) +} + +async fn import_preview_target_columns( + application: &Application, + request: &LegacyImportPreviewFileRequest, +) -> LegacyResult> { + let datasource_id = request.data_source_id.as_string(); + let database_type = resolve_database_type(application, &datasource_id, "").await?; + Ok(application + .list_community_columns(ListCommunityColumnsRequest { + datasource_id, + database_type, + database_name: request.database_name.clone(), + schema_name: request.schema_name.clone(), + table_name: request.table_name.clone(), + }) + .await? + .items) +} + +pub(crate) async fn import_preview( + application: &Application, + request: &LegacyImportPreviewFileRequest, +) -> LegacyResult { + let columns = import_preview_target_columns(application, request).await?; + let (source_columns, rows) = application + .import_preview_table(&request.file_id, true, IMPORT_PREVIEW_ROW_LIMIT) + .await?; + let suggested_mapping = source_columns + .iter() + .filter_map(|source| { + columns + .iter() + .find(|target| target.name.eq_ignore_ascii_case(source)) + .map(|target| LegacyImportPreviewMapping { + source_column: source.clone(), + target_column: target.name.clone(), + }) + }) + .collect(); + let target_columns = columns + .into_iter() + .map(|column| LegacyImportPreviewColumn { + name: column.name, + data_type: column.column_type, + nullable: column.nullable != Some(0), + auto_increment: column.auto_increment.unwrap_or(false), + default_value: column.default_value, + comment: non_blank(&column.comment), + }) + .collect(); + Ok(LegacyImportPreview { + source_columns, + preview_data: rows, + target_table_name: request.table_name.clone(), + target_columns, + suggested_mapping, + preview_limit: IMPORT_PREVIEW_ROW_LIMIT, + }) +} + +pub(crate) async fn execute_import_preview( + application: &Application, + request: &LegacyImportPreviewExecuteRequest, +) -> LegacyResult { + if request.unmapped_target.trim().eq_ignore_ascii_case("NULL") { + return Err(LegacyFailure { + code: "unsupported_import_mapping".to_owned(), + message: "Writing NULL into unmapped target columns is not supported yet".to_owned(), + }); + } + let mappings: Vec<(String, String)> = request + .mappings + .iter() + .filter_map(|mapping| { + non_blank(mapping.source_column.as_deref().unwrap_or_default()) + .map(|source| (source, mapping.target_column.clone())) + }) + .collect(); + if mappings + .iter() + .any(|(source, target)| !source.eq_ignore_ascii_case(target)) + { + return Err(LegacyFailure { + code: "unsupported_import_mapping".to_owned(), + message: "Renaming columns during import is not supported yet".to_owned(), + }); + } + let (columns, _) = application + .import_preview_table(&request.file_id, true, 1) + .await?; + if columns.len() != mappings.len() + || !columns.iter().all(|column| { + mappings + .iter() + .any(|(source, _)| source.eq_ignore_ascii_case(column)) + }) + { + return Err(LegacyFailure { + code: "unsupported_import_mapping".to_owned(), + message: "Every file column must map to a same-named table column; skipped columns are not supported yet" + .to_owned(), + }); + } + let download = application + .transfer_artifact_download(&request.file_id) + .await?; + let format = import_preview_transfer_format(&download.artifact.format)?; + let Some(file_path) = download.path.to_str().map(str::to_owned) else { + return Err(LegacyFailure { + code: "invalid_import_path".to_owned(), + message: "The staged import file path cannot be represented as UTF-8".to_owned(), + }); + }; + let accepted = application + .import_mysql_file(ImportFileRequest { + datasource_id: request.data_source_id.as_string(), + database_name: request.database_name.clone(), + schema_name: request.schema_name.clone(), + table_name: non_blank(&request.table_name), + file_path, + format, + contains_header: true, + tabular_encoding: TabularImportEncoding::Plain, + }) + .await?; + Ok(LegacyImportExportTaskAccepted { + task_id: accepted.task_id, + }) +} + +async fn submitted_import_path( + application: &Application, + request: &LegacyImportExportSubmitRequest, +) -> LegacyResult { + if let Some(path) = non_blank(&request.source_file) { + return Ok(path); + } + if let Some(file_id) = non_blank(&request.file_id) { + let download = application.transfer_artifact_download(&file_id).await?; + return download + .path + .to_str() + .map(str::to_owned) + .ok_or_else(|| LegacyFailure { + code: "invalid_import_path".to_owned(), + message: "The staged import file path cannot be represented as UTF-8".to_owned(), + }); + } + Err(LegacyFailure::invalid( + "invalid_import_request", + "sourceFile or fileId is required", + )) +} + +pub(crate) async fn submit_import_export_task( + application: &Application, + request: &LegacyImportExportSubmitRequest, +) -> LegacyResult { + let task_type = request.task_type.trim().to_ascii_uppercase(); + match task_type.as_str() { + "SQL_EXPORT" => { + let task_id = export_legacy_mysql_sql_file( + application, + &LegacySqlFileExportRequest { + data_source_id: request.data_source_id.clone(), + database_name: request.database_name.clone(), + schema_name: request.schema_name.clone(), + table_name: request.table_name.clone(), + table_names: request.table_names.clone(), + scope: request.scope.clone(), + contain_data: request.contain_data, + export_path: String::new(), + }, + ) + .await?; + Ok(LegacyImportExportTaskAccepted { task_id }) + } + "TABLE_DATA_EXPORT" => { + let task_id = export_legacy_mysql_other_file( + application, + &LegacyOtherFileExportRequest { + data_source_id: request.data_source_id.clone(), + database_name: request.database_name.clone(), + schema_name: request.schema_name.clone(), + table_name: request.table_name.clone(), + table_names: request.table_names.clone(), + export_type: request.format.clone(), + contains_header: request.contains_header, + export_path: String::new(), + }, + ) + .await?; + Ok(LegacyImportExportTaskAccepted { task_id }) + } + "DATA_FILE_IMPORT" | "SQL_FILE_IMPORT" => { + let sql_route = task_type == "SQL_FILE_IMPORT"; + let file_name = submitted_import_path(application, request).await?; + let import_type = if sql_route { + "SQL".to_owned() + } else { + request.format.clone() + }; + let task_id = import_legacy_mysql_desktop_file( + application, + &LegacyImportFileRequest { + data_source_id: request.data_source_id.clone(), + database_name: request.database_name.clone(), + schema_name: request.schema_name.clone(), + table_name: request.table_name.clone(), + file_name, + import_type, + contains_header: request.contains_header, + tabular_encoding: TabularImportEncoding::Plain, + }, + sql_route, + ) + .await?; + Ok(LegacyImportExportTaskAccepted { task_id }) + } + "QUERY_RESULT_EXPORT" => Err(LegacyFailure { + code: "unsupported_task_type".to_owned(), + message: "Exporting a query result as a background task is not supported yet" + .to_owned(), + }), + _ => Err(LegacyFailure::invalid( + "invalid_task_type", + "taskType is not supported", + )), + } +} + pub(crate) async fn export_legacy_mysql_dml( application: &Application, request: &LegacyDmlExportRequest, @@ -4222,6 +5056,39 @@ pub(crate) async fn list_simple_tables( } /// Lists table or view columns in the historical `ColumnResponse` shape. +pub(crate) async fn list_active_transactions( + application: &Application, + query: &LegacyActiveTransactionQuery, +) -> LegacyResult> { + let datasource_id = query.data_source_id.as_string(); + resolve_mysql_database_type(application, &datasource_id, "").await?; + Ok(application + .list_mysql_active_transactions(&datasource_id) + .await? + .into_iter() + .map(|item| LegacyActiveTransaction { + connection_inspection_sql: format!( + "SELECT * FROM information_schema.PROCESSLIST WHERE ID = {}", + item.thread_id + ), + lock_metadata_state: "UNAVAILABLE".to_owned(), + started_at: legacy_epoch_millis(&item.started_at_ms), + trx_id: item.trx_id, + state: item.state, + age_seconds: item.age_seconds, + isolation_level: item.isolation_level, + rows_locked: item.rows_locked, + rows_modified: item.rows_modified, + lock_structs: item.lock_structs, + thread_id: item.thread_id, + user: item.user, + host: item.host, + db: item.database, + query: item.query, + }) + .collect()) +} + pub(crate) async fn list_columns( application: &Application, query: &LegacyTableDetailQuery, @@ -4732,6 +5599,26 @@ pub(crate) async fn truncate_table( .await } +/// Returns the default copy-table name the Community copy dialog pre-fills. +pub(crate) async fn prepare_copy_table_name( + application: &Application, + request: &LegacyTableCopyPrepareQuery, +) -> LegacyResult { + let datasource_id = request.data_source_id.as_string(); + resolve_mysql_database_type(application, &datasource_id, &request.database_type).await?; + if request.table_name.trim().is_empty() { + return Err(LegacyFailure::invalid( + "invalid_table_request", + "tableName is required", + )); + } + Ok(default_copy_table_name(&request.table_name)) +} + +fn default_copy_table_name(table_name: &str) -> String { + format!("{}_copy", table_name.trim()) +} + pub(crate) async fn copy_table( application: &Application, request: &LegacyTableCopyRequest, @@ -4739,7 +5626,7 @@ pub(crate) async fn copy_table( let datasource_id = request.data_source_id.as_string(); resolve_mysql_database_type(application, &datasource_id, &request.database_type).await?; let new_name = if request.new_name.trim().is_empty() { - format!("{}_copy", request.table_name.trim()) + default_copy_table_name(&request.table_name) } else { request.new_name.trim().to_owned() }; @@ -6366,6 +7253,22 @@ fn storage_failure(error: StorageError) -> LegacyFailure { } } +pub(crate) async fn update_datasource_identity_color( + application: &Application, + request: &LegacyIdentityColorRequest, +) -> LegacyResult { + let id = request.id.as_string(); + let datasource = application + .update_datasource_identity_color(&id, request.identity_color.clone()) + .await?; + Ok(LegacyIdentityColorResponse { + id: datasource.id, + identity_color: datasource.identity_color, + environment_id: 1, + environment: default_environment(), + }) +} + fn datasource_response( application: &Application, datasource: DatasourceEditProjection, @@ -6436,6 +7339,7 @@ fn datasource_response( support_schema, has_secret: datasource.has_secret, revision: datasource.revision, + identity_color: datasource.identity_color, } } @@ -8190,6 +9094,8 @@ async fn dispatch_inner( Err(error) => Err(error), }, ("get", "/api/common/environment/list_all") => serialized(Ok(environments())), + ("get", "/api/database/supported") => serialize_data(list_supported_databases(application)), + ("post", "/api/jdbc/driver/upload") => Err(custom_driver_upload_unsupported()), ("get", "/api/jdbc/driver/list") => decode(request.message) .map(|query: LegacyDriverQuery| drivers(application, &query.db_type)) .and_then(serialize_data), @@ -8364,6 +9270,12 @@ async fn dispatch_inner( Err(error) => Err(error), } } + ("post", "/api/connection/datasource/identity_color") => { + match decode::(request.message) { + Ok(body) => serialized(update_datasource_identity_color(application, &body).await), + Err(error) => Err(error), + } + } ("post", "/api/import/sql_file") => { match decode::(request.message) { Ok(body) if desktop_paths => { @@ -8421,6 +9333,85 @@ async fn dispatch_inner( Ok(_) => Err(desktop_file_operation_required()), Err(error) => Err(error), }, + ("get", "/api/tasks/list") => { + match decode::(request.message) { + Ok(query) => serialized(list_import_export_tasks(application, &query).await), + Err(error) => Err(error), + } + } + ("get", "/api/tasks/get") => { + match decode::(request.message) { + Ok(query) => serialized(get_import_export_task(application, &query).await), + Err(error) => Err(error), + } + } + ("get", "/api/tasks/active-count") => { + serialized(active_import_export_task_count(application).await) + } + ("delete", "/api/tasks/delete") => { + match decode::(request.message) { + Ok(query) => serialized(delete_import_export_task(application, &query).await), + Err(error) => Err(error), + } + } + ("get", "/api/tasks/events") => { + match decode::(request.message) { + Ok(query) => serialized(list_import_export_task_events(application, &query).await), + Err(error) => Err(error), + } + } + ("post", "/api/tasks/export" | "/api/tasks/import") => { + match decode::(request.message) { + Ok(body) => serialized(submit_import_export_task(application, &body).await), + Err(error) => Err(error), + } + } + ("post", "/api/tasks/prepare-user-exit") => serialized( + application + .prepare_user_exit() + .await + .map_err(LegacyFailure::from), + ), + ("post", "/api/tasks/abort-user-exit") => serialized( + application + .abort_user_exit() + .await + .map_err(LegacyFailure::from), + ), + ("get", "/api/rdb/active_transaction/list") => { + match decode::(request.message) { + Ok(query) => serialized(list_active_transactions(application, &query).await), + Err(error) => Err(error), + } + } + ("post", "/api/rdb/import_preview/upload_local") => { + match decode::(request.message) { + Ok(body) if desktop_paths => { + serialized(stage_import_preview_local_file(application, &body).await) + } + Ok(_) => Err(desktop_file_operation_required()), + Err(error) => Err(error), + } + } + ("post", "/api/rdb/import_preview/upload") => Err(desktop_file_operation_required()), + ("post", "/api/rdb/import_preview/sheets") => { + match decode::(request.message) { + Ok(body) => serialized(import_preview_sheets(application, &body).await), + Err(error) => Err(error), + } + } + ("post", "/api/rdb/import_preview/preview") => { + match decode::(request.message) { + Ok(body) => serialized(import_preview(application, &body).await), + Err(error) => Err(error), + } + } + ("post", "/api/rdb/import_preview/execute") => { + match decode::(request.message) { + Ok(body) => serialized(execute_import_preview(application, &body).await), + Err(error) => Err(error), + } + } ("get", "/api/sql/format") => match decode::(request.message) { Ok(query) => serialized(format_legacy_sql(application, &query).await), Err(error) => Err(error), @@ -8817,6 +9808,12 @@ async fn dispatch_inner( Err(error) => Err(error), } } + ("get", "/api/rdb/table/copy/prepare") => { + match decode::(request.message) { + Ok(query) => serialized(prepare_copy_table_name(application, &query).await), + Err(error) => Err(error), + } + } ("post", "/api/rdb/table/copy") => { match decode::(request.message) { Ok(body) => serialized(copy_table(application, &body).await), @@ -8923,6 +9920,7 @@ const LEGACY_PATHS: &[&str] = &[ "/api/connection/datasource/create", "/api/connection/datasource/pre_connect", "/api/connection/datasource/update", + "/api/connection/datasource/identity_color", "/api/connection/datasource/clone", "/api/connection/datasource/connect", "/api/connection/datasource/close", @@ -8938,12 +9936,28 @@ const LEGACY_PATHS: &[&str] = &[ "/api/connection/console/connect", "/api/import/sql_file", "/api/import/other_file", + "/api/rdb/active_transaction/list", + "/api/rdb/import_preview/upload", + "/api/rdb/import_preview/upload_local", + "/api/rdb/import_preview/sheets", + "/api/rdb/import_preview/preview", + "/api/rdb/import_preview/execute", "/api/export/sql_file", "/api/export/other_file", "/api/task/list", "/api/task/get", "/api/task/stop", "/api/task/download", + "/api/tasks/list", + "/api/tasks/get", + "/api/tasks/active-count", + "/api/tasks/artifact", + "/api/tasks/delete", + "/api/tasks/events", + "/api/tasks/export", + "/api/tasks/import", + "/api/tasks/prepare-user-exit", + "/api/tasks/abort-user-exit", "/api/sql/format", "/api/sql/valid_select", "/api/sql_parser/get_keywords", @@ -8974,6 +9988,8 @@ const LEGACY_PATHS: &[&str] = &[ "/api/rdb/account/grants", "/api/rdb/account/preview", "/api/rdb/account/execute", + "/api/database/supported", + "/api/jdbc/driver/upload", "/api/diff/sql", "/api/rdb/database/list", "/api/rdb/database/create_database_sql", @@ -8981,6 +9997,7 @@ const LEGACY_PATHS: &[&str] = &[ "/api/rdb/schema/create_schema_sql", "/api/rdb/ddl/schema_list", "/api/rdb/ddl/database_schema_list", + "/api/rdb/table/copy/prepare", "/api/rdb/table/list", "/api/rdb/table/table_meta", "/api/rdb/table/query", @@ -9116,6 +10133,11 @@ pub(crate) fn routes() -> Router { .route("/api/v1/chart/update", post(chart_update_handler)) .route("/api/chart", axum::routing::delete(chart_delete_handler)) .route("/api/common/environment/list_all", get(environment_handler)) + .route("/api/database/supported", get(supported_database_handler)) + .route( + "/api/jdbc/driver/upload", + post(driver_upload_handler).layer(DefaultBodyLimit::max(MAX_LEGACY_MULTIPART_BYTES)), + ) .route("/api/jdbc/driver/list", get(driver_handler)) .route("/api/jdbc/driver/download", get(driver_download_handler)) .route("/api/jdbc/driver/save", post(driver_save_handler)) @@ -9197,6 +10219,10 @@ pub(crate) fn routes() -> Router { "/api/connection/datasource/update", post(update_datasource_handler).put(update_datasource_handler), ) + .route( + "/api/connection/datasource/identity_color", + post(update_datasource_identity_color_handler), + ) .route( "/api/import/sql_file", post(import_sql_file_handler).layer(DefaultBodyLimit::max(MAX_LEGACY_MULTIPART_BYTES)), @@ -9208,9 +10234,55 @@ pub(crate) fn routes() -> Router { ) .route("/api/export/sql_file", post(export_sql_file_handler)) .route("/api/export/other_file", post(export_other_file_handler)) + .route( + "/api/rdb/import_preview/upload", + post(import_preview_upload_handler) + .layer(DefaultBodyLimit::max(MAX_LEGACY_MULTIPART_BYTES)), + ) + .route( + "/api/rdb/import_preview/upload_local", + post(import_preview_local_upload_handler), + ) + .route( + "/api/rdb/import_preview/sheets", + post(import_preview_sheets_handler), + ) + .route( + "/api/rdb/import_preview/preview", + post(import_preview_handler), + ) + .route( + "/api/rdb/import_preview/execute", + post(import_preview_execute_handler), + ) .route("/api/task/list", get(transfer_task_list_handler)) .route("/api/task/get", get(transfer_task_get_handler)) .route("/api/task/stop", get(transfer_task_stop_handler)) + .route("/api/tasks/list", get(import_export_task_list_handler)) + .route("/api/tasks/get", get(import_export_task_get_handler)) + .route( + "/api/tasks/active-count", + get(import_export_task_active_count_handler), + ) + .route( + "/api/tasks/artifact", + get(import_export_task_artifact_handler), + ) + .route( + "/api/tasks/delete", + axum::routing::delete(import_export_task_delete_handler), + ) + .route("/api/tasks/events", get(import_export_task_events_handler)) + .route("/api/tasks/export", post(import_export_task_export_handler)) + .route("/api/tasks/import", post(import_export_task_import_handler)) + .route( + "/api/tasks/prepare-user-exit", + post(import_export_task_prepare_exit_handler), + ) + .route( + "/api/tasks/abort-user-exit", + post(import_export_task_abort_exit_handler), + ) .route("/api/task/download", get(transfer_task_download_handler)) .route("/api/sql/format", get(sql_format_handler)) .route("/api/sql/valid_select", get(sql_valid_select_handler)) @@ -9313,6 +10385,14 @@ pub(crate) fn routes() -> Router { .route("/api/rdb/table/modify/sql", post(table_modify_sql_handler)) .route("/api/rdb/table/truncate", post(table_truncate_handler)) .route("/api/rdb/table/copy", post(table_copy_handler)) + .route( + "/api/rdb/active_transaction/list", + get(active_transaction_list_handler), + ) + .route( + "/api/rdb/table/copy/prepare", + get(table_copy_prepare_handler), + ) .route("/api/rdb/table/table_list", get(simple_table_list_handler)) .route("/api/rdb/table/column_list", get(table_column_list_handler)) .route("/api/rdb/table/index_list", get(table_index_list_handler)) @@ -9521,6 +10601,16 @@ async fn environment_handler() -> Json>> { envelope(Ok(environments())) } +async fn supported_database_handler( + State(application): State, +) -> Json>> { + envelope(Ok(list_supported_databases(&application))) +} + +async fn driver_upload_handler() -> Json>> { + envelope(Err(custom_driver_upload_unsupported())) +} + async fn driver_handler( State(application): State, Query(query): Query, @@ -9717,6 +10807,13 @@ async fn update_datasource_handler( envelope(update_datasource(&application, &request).await) } +async fn update_datasource_identity_color_handler( + State(application): State, + Json(request): Json, +) -> Json> { + envelope(update_datasource_identity_color(&application, &request).await) +} + async fn delete_datasource_handler( State(application): State, Query(query): Query, @@ -9786,6 +10883,122 @@ async fn transfer_task_download_handler( transfer_attachment_response(legacy_transfer_task_download(&application, &query.id).await) } +async fn import_export_task_list_handler( + State(application): State, + Query(query): Query, +) -> Json>> { + envelope(list_import_export_tasks(&application, &query).await) +} + +async fn import_export_task_get_handler( + State(application): State, + Query(query): Query, +) -> Json> { + envelope(get_import_export_task(&application, &query).await) +} + +async fn import_export_task_active_count_handler( + State(application): State, +) -> Json> { + envelope(active_import_export_task_count(&application).await) +} + +async fn import_export_task_artifact_handler( + State(application): State, + Query(query): Query, +) -> Response { + transfer_attachment_response(import_export_task_artifact(&application, &query).await) +} + +async fn import_export_task_delete_handler( + State(application): State, + Query(query): Query, +) -> Json> { + envelope(delete_import_export_task(&application, &query).await) +} + +async fn import_export_task_events_handler( + State(application): State, + Query(query): Query, +) -> Json>> { + envelope(list_import_export_task_events(&application, &query).await) +} + +async fn active_transaction_list_handler( + State(application): State, + Query(query): Query, +) -> Json>> { + envelope(list_active_transactions(&application, &query).await) +} + +async fn import_preview_upload_handler( + State(application): State, + multipart: Multipart, +) -> Json> { + envelope(stage_import_preview_upload(&application, multipart).await) +} + +async fn import_preview_local_upload_handler() -> Json> { + envelope(Err(desktop_file_operation_required())) +} + +async fn import_preview_sheets_handler( + State(application): State, + Json(request): Json, +) -> Json>> { + envelope(import_preview_sheets(&application, &request).await) +} + +async fn import_preview_handler( + State(application): State, + Json(request): Json, +) -> Json> { + envelope(import_preview(&application, &request).await) +} + +async fn import_preview_execute_handler( + State(application): State, + Json(request): Json, +) -> Json> { + envelope(execute_import_preview(&application, &request).await) +} + +async fn import_export_task_export_handler( + State(application): State, + Json(request): Json, +) -> Json> { + envelope(submit_import_export_task(&application, &request).await) +} + +async fn import_export_task_import_handler( + State(application): State, + Json(request): Json, +) -> Json> { + envelope(submit_import_export_task(&application, &request).await) +} + +async fn import_export_task_prepare_exit_handler( + State(application): State, +) -> Json> { + envelope( + application + .prepare_user_exit() + .await + .map_err(LegacyFailure::from), + ) +} + +async fn import_export_task_abort_exit_handler( + State(application): State, +) -> Json> { + envelope( + application + .abort_user_exit() + .await + .map_err(LegacyFailure::from), + ) +} + async fn dml_export_handler( State(application): State, Json(request): Json, @@ -10152,6 +11365,13 @@ async fn table_copy_handler( envelope(copy_table(&application, &request).await) } +async fn table_copy_prepare_handler( + State(application): State, + Query(query): Query, +) -> Json> { + envelope(prepare_copy_table_name(&application, &query).await) +} + async fn database_delete_prepare_handler( State(application): State, Json(request): Json, @@ -10587,6 +11807,7 @@ mod tests { }), has_secret: true, revision: "1".to_owned(), + identity_color: None, }, ); let json = serde_json::to_string(&response).expect("response serializes"); @@ -10645,6 +11866,7 @@ mod tests { ("get", "/api/rdb/ddl/database_schema_list"), ("post", "/api/rdb/ddl/execute"), ("get", "/api/rdb/table/table_meta"), + ("get", "/api/rdb/table/copy/prepare"), ("get", "/api/rdb/table/query"), ("get", "/api/rdb/table/export"), ("get", "/api/rdb/table/create/example"), @@ -10693,6 +11915,8 @@ mod tests { ]; const REQUIRED_WORKSPACE_PATHS: &[(&str, &str)] = &[ + ("get", "/api/database/supported"), + ("post", "/api/jdbc/driver/upload"), ("get", "/api/jdbc/driver/download"), ("post", "/api/jdbc/driver/save"), ("delete", "/api/jdbc/driver/delete"), @@ -10710,6 +11934,7 @@ mod tests { ("post", "/api/converter/dbp/upload"), ("post", "/api/converter/chat2db/upload"), ("post", "/api/converter/datagrip/upload"), + ("post", "/api/connection/datasource/identity_color"), ("post", "/api/namespaces/create"), ("post", "/api/namespaces/update"), ("post", "/api/namespaces/delete"), @@ -10720,12 +11945,27 @@ mod tests { const REQUIRED_TRANSFER_PATHS: &[(&str, &str)] = &[ ("post", "/api/import/sql_file"), ("post", "/api/import/other_file"), + ("post", "/api/rdb/import_preview/upload"), + ("post", "/api/rdb/import_preview/upload_local"), + ("post", "/api/rdb/import_preview/sheets"), + ("post", "/api/rdb/import_preview/preview"), + ("post", "/api/rdb/import_preview/execute"), ("post", "/api/export/sql_file"), ("post", "/api/export/other_file"), ("get", "/api/task/list"), ("get", "/api/task/get"), ("get", "/api/task/stop"), ("get", "/api/task/download"), + ("get", "/api/tasks/list"), + ("get", "/api/tasks/get"), + ("get", "/api/tasks/active-count"), + ("get", "/api/tasks/artifact"), + ("delete", "/api/tasks/delete"), + ("get", "/api/tasks/events"), + ("post", "/api/tasks/export"), + ("post", "/api/tasks/import"), + ("post", "/api/tasks/prepare-user-exit"), + ("post", "/api/tasks/abort-user-exit"), ("post", "/api/rdb/dml/export"), ("post", "/api/rdb/table/generate/class"), ]; @@ -11389,6 +12629,16 @@ mod tests { } } + async fn dashboard_response_json(response: axum::response::Response) -> serde_json::Value { + let body = response + .into_body() + .collect() + .await + .expect("response body must collect") + .to_bytes(); + serde_json::from_slice(&body).expect("response body must be JSON") + } + async fn dashboard_http_json( router: &Router, method: &str, @@ -11421,6 +12671,171 @@ mod tests { serde_json::from_slice(&body).expect("response body must be JSON") } + #[tokio::test] + async fn supported_databases_route_serves_the_driver_inventory() { + let directory = tempfile::TempDir::new().expect("temporary directory"); + let storage = Storage::open(directory.path(), Arc::new(EmptyVault)).expect("storage opens"); + let router = routes().with_state(Application::with_storage(storage)); + + let body = dashboard_http_json(&router, "GET", "/api/database/supported", None).await; + assert_eq!(body["success"], true); + assert!(body["errorCode"].is_null()); + let databases = body["data"] + .as_array() + .expect("supported databases must be an array"); + for database in databases { + assert!(database["dbType"].is_string(), "dbType must be a string"); + assert_eq!(database["supportDatabase"], true); + assert!(database["supportSchema"].is_boolean()); + } + } + + #[tokio::test] + async fn driver_upload_route_reports_the_unsupported_gap() { + let directory = tempfile::TempDir::new().expect("temporary directory"); + let storage = Storage::open(directory.path(), Arc::new(EmptyVault)).expect("storage opens"); + let router = routes().with_state(Application::with_storage(storage)); + + let body = dashboard_http_json(&router, "POST", "/api/jdbc/driver/upload", None).await; + assert_eq!(body["success"], false); + assert_eq!(body["errorCode"], "unsupported_custom_driver_upload"); + } + + #[tokio::test] + async fn import_preview_stages_files_and_rejects_unsupported_mappings() { + let directory = tempfile::TempDir::new().expect("temporary directory"); + let storage = Storage::open(directory.path(), Arc::new(EmptyVault)).expect("storage opens"); + let router = routes().with_state(Application::with_storage(storage)); + + let boundary = "chat2db-preview-boundary"; + let multipart = format!( + "--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"rows.csv\"\r\nContent-Type: text/csv\r\n\r\nname,value\r\nalpha,1\r\n--{boundary}--\r\n" + ); + let upload_request = Request::builder() + .method(axum::http::Method::POST) + .uri("/api/rdb/import_preview/upload") + .header( + "content-type", + format!("multipart/form-data; boundary={boundary}"), + ) + .body(Body::from(multipart)) + .expect("multipart request must build"); + let uploaded = router + .clone() + .oneshot(upload_request) + .await + .expect("upload route must respond"); + let uploaded = dashboard_response_json(uploaded).await; + assert_eq!(uploaded["success"], true); + let file_id = uploaded["data"] + .as_str() + .expect("staged file id must be a string") + .to_owned(); + + let sheets = dashboard_http_json( + &router, + "POST", + "/api/rdb/import_preview/sheets", + Some(serde_json::json!({ + "dataSourceId": "datasource-1", + "databaseName": "app", + "tableName": "items", + "fileId": file_id + })), + ) + .await; + assert_eq!(sheets["success"], true); + assert_eq!(sheets["data"], serde_json::json!([])); + + let renamed = dashboard_http_json( + &router, + "POST", + "/api/rdb/import_preview/execute", + Some(serde_json::json!({ + "dataSourceId": "datasource-1", + "databaseName": "app", + "tableName": "items", + "fileId": file_id, + "unmappedTarget": "DEFAULT", + "mappings": [ + {"sourceColumn": "name", "targetColumn": "label"}, + {"sourceColumn": "value", "targetColumn": "value"} + ] + })), + ) + .await; + assert_eq!(renamed["success"], false); + assert_eq!(renamed["errorCode"], "unsupported_import_mapping"); + } + + #[tokio::test] + async fn prepared_exit_blocks_new_task_submissions_until_aborted() { + let directory = tempfile::TempDir::new().expect("temporary directory"); + let storage = Storage::open(directory.path(), Arc::new(EmptyVault)).expect("storage opens"); + let router = routes().with_state(Application::with_storage(storage)); + + let prepared = + dashboard_http_json(&router, "POST", "/api/tasks/prepare-user-exit", None).await; + assert_eq!(prepared["success"], true); + + let blocked = dashboard_http_json( + &router, + "POST", + "/api/tasks/import", + Some(serde_json::json!({ + "dataSourceId": "datasource-1", + "databaseName": "app", + "taskType": "DATA_FILE_IMPORT", + "format": "CSV", + "sourceFile": "/tmp/rows.csv" + })), + ) + .await; + assert_eq!(blocked["success"], false); + assert_eq!(blocked["errorCode"], "runtime_not_accepting_work"); + + let aborted = + dashboard_http_json(&router, "POST", "/api/tasks/abort-user-exit", None).await; + assert_eq!(aborted["success"], true); + } + + #[tokio::test] + async fn task_submission_reports_unsupported_and_invalid_task_types() { + let directory = tempfile::TempDir::new().expect("temporary directory"); + let storage = Storage::open(directory.path(), Arc::new(EmptyVault)).expect("storage opens"); + let router = routes().with_state(Application::with_storage(storage)); + + let unsupported = dashboard_http_json( + &router, + "POST", + "/api/tasks/export", + Some(serde_json::json!({ + "dataSourceId": "datasource-1", + "databaseName": "app", + "taskType": "QUERY_RESULT_EXPORT", + "format": "CSV" + })), + ) + .await; + assert_eq!(unsupported["success"], false); + assert_eq!(unsupported["errorCode"], "unsupported_task_type"); + + let invalid = dashboard_http_json( + &router, + "POST", + "/api/tasks/import", + Some(serde_json::json!({ + "dataSourceId": "datasource-1", + "databaseName": "app", + "taskType": "NOPE", + "format": "CSV" + })), + ) + .await; + assert_eq!(invalid["success"], false); + assert_eq!(invalid["errorCode"], "invalid_task_type"); + } + #[tokio::test] async fn dashboard_http_routes_preserve_community_crud_and_envelopes() { let directory = tempfile::TempDir::new().expect("temporary directory"); diff --git a/apps/chat2db-web/src/legacy_ai.rs b/apps/chat2db-web/src/legacy_ai.rs index 7e2e060..ae12d42 100644 --- a/apps/chat2db-web/src/legacy_ai.rs +++ b/apps/chat2db-web/src/legacy_ai.rs @@ -181,6 +181,13 @@ pub struct LegacyAiSessionDeleteRequest { pub id: String, } +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct LegacyAiSessionRenameRequest { + pub id: String, + pub title: String, +} + #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase")] struct LegacyAiMessagesQuery { @@ -662,6 +669,38 @@ pub async fn delete_session(application: &Application, id: &str) -> Result<(), L Ok(()) } +/// Renames one durable Community AI session through revision CAS. +/// +/// # Errors +/// +/// Returns validation, lookup, revision, or storage failures. +pub async fn rename_session( + application: &Application, + id: &str, + title: &str, +) -> Result<(), LegacyAiFailure> { + let title = title.trim(); + if title.is_empty() { + return Err(LegacyAiFailure::invalid( + "invalid_ai_request", + "title must not be empty", + )); + } + let session = application.get_agent_session(id.trim()).await?; + application + .update_agent_session( + &session.id, + UpdateAgentSessionRequest { + expected_revision: session.revision.clone(), + title: title.to_owned(), + provider_id: session.provider_id.clone(), + datasource_id: session.datasource_id.clone(), + }, + ) + .await?; + Ok(()) +} + /// Lists provider profiles that have usable credentials as frontend model options. /// /// # Errors @@ -1088,6 +1127,23 @@ pub async fn dispatch( .unwrap_or_else(|_| internal_failure_value()), ) } + ("post", "/api/v3/ai/chat/history/session/rename") => { + let request = serde_json::from_value::(message); + let result = match request { + Ok(request) => rename_session(application, &request.id, &request.title).await, + Err(_) => Err(LegacyAiFailure::invalid( + "invalid_ai_request", + "id and title must not be empty", + )), + }; + Some( + serde_json::to_value(match result { + Ok(()) => LegacyAiEnvelope::success(()), + Err(error) => LegacyAiEnvelope::failure(error), + }) + .unwrap_or_else(|_| internal_failure_value()), + ) + } ("get", "/api/v3/ai/model/options") => Some( serde_json::to_value(match model_options(application).await { Ok(data) => LegacyAiEnvelope::success(data), @@ -1186,6 +1242,10 @@ pub(crate) fn routes() -> Router { "/api/v3/ai/chat/history/session/delete", post(delete_session_handler), ) + .route( + "/api/v3/ai/chat/history/session/rename", + post(rename_session_handler), + ) .route("/api/v3/ai/model/options", get(model_options_handler)) .route("/api/v3/ai/model/list", get(model_catalog_handler)) .route( @@ -1312,6 +1372,21 @@ async fn delete_session_handler( ) } +async fn rename_session_handler( + State(application): State, + Json(request): Json, +) -> Json { + Json( + serde_json::to_value( + match rename_session(&application, &request.id, &request.title).await { + Ok(()) => LegacyAiEnvelope::success(()), + Err(error) => LegacyAiEnvelope::failure(error), + }, + ) + .unwrap_or_else(|_| internal_failure_value()), + ) +} + async fn model_options_handler(State(application): State) -> Json { Json( serde_json::to_value(match model_options(&application).await { @@ -1978,7 +2053,7 @@ mod tests { }; use chat2db_contract::{ AgentEvent, AgentEventEnvelope, AgentPermissionRequest, ApiError, - CreateProviderProfileRequest, ProviderKind, + CreateAgentSessionRequest, CreateProviderProfileRequest, ProviderKind, }; use chat2db_core::Application; use chat2db_storage::{SecretRef, SecretValue, SecretVault, SecretVaultError, Storage}; @@ -2454,4 +2529,54 @@ mod tests { .contains("alpha") ); } + + #[tokio::test] + async fn session_rename_route_updates_the_visible_title() { + let fixture = test_application(); + let profile = fixture + .application + .create_provider_profile(CreateProviderProfileRequest { + name: "Rename provider".to_owned(), + kind: ProviderKind::OpenAiCompatible, + base_url: "https://provider.example/v1".to_owned(), + model: "mock-model".to_owned(), + context_window_tokens: "4096".to_owned(), + max_output_tokens: "1024".to_owned(), + credentials: None, + }) + .await + .expect("provider profile must be created"); + let session = fixture + .application + .create_agent_session(CreateAgentSessionRequest { + title: "First title".to_owned(), + provider_id: profile.id, + datasource_id: None, + system_prompt: None, + }) + .await + .expect("session must be created"); + let application = routes().with_state(fixture.application); + + let renamed = application + .clone() + .oneshot(json_request( + Method::POST, + "/api/v3/ai/chat/history/session/rename", + &serde_json::json!({ "id": session.id, "title": "Renamed title" }), + )) + .await + .expect("rename route must respond"); + assert_eq!(response_json(renamed).await["success"], true); + + let sessions = application + .oneshot(empty_request( + Method::GET, + "/api/v3/ai/chat/history/sessions", + )) + .await + .expect("session list must respond"); + let sessions = response_json(sessions).await; + assert_eq!(sessions["data"][0]["title"], "Renamed title"); + } } diff --git a/apps/frontend/src/generated/contract.ts b/apps/frontend/src/generated/contract.ts index f42da1d..0718722 100644 --- a/apps/frontend/src/generated/contract.ts +++ b/apps/frontend/src/generated/contract.ts @@ -1865,6 +1865,8 @@ export interface components { hasSecret: boolean; /** @description Opaque datasource id. */ id: string; + /** @description Optional `#RRGGBB` identity colour shown by the connection tree. */ + identityColor?: string | null; /** @description User-visible datasource name. */ name: string; /** @description Monotonic revision encoded as a decimal integer. */ diff --git a/contracts/openapi/chat2db-v1.json b/contracts/openapi/chat2db-v1.json index cce64e3..e60a784 100644 --- a/contracts/openapi/chat2db-v1.json +++ b/contracts/openapi/chat2db-v1.json @@ -7211,6 +7211,13 @@ "type": "string", "description": "Opaque datasource id." }, + "identityColor": { + "type": [ + "string", + "null" + ], + "description": "Optional `#RRGGBB` identity colour shown by the connection tree." + }, "name": { "type": "string", "description": "User-visible datasource name." diff --git a/crates/chat2db-contract/src/community.rs b/crates/chat2db-contract/src/community.rs index 36ba20e..cb4e0fe 100644 --- a/crates/chat2db-contract/src/community.rs +++ b/crates/chat2db-contract/src/community.rs @@ -2049,3 +2049,38 @@ mod tests { } } } + +/// One active `InnoDB` transaction reported by the `MySQL` operations monitor. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ActiveTransaction { + /// `InnoDB` transaction id. + pub trx_id: String, + /// `InnoDB` transaction state. + pub state: String, + /// Transaction start time as Unix epoch milliseconds. + pub started_at_ms: String, + /// Whole seconds since the transaction started. + pub age_seconds: u64, + /// Isolation level reported by `InnoDB`. + pub isolation_level: String, + /// Rows currently locked by the transaction. + pub rows_locked: u64, + /// Rows modified by the transaction. + pub rows_modified: u64, + /// `InnoDB` lock structures held by the transaction. + pub lock_structs: u64, + /// `MySQL` thread id owning the transaction. + pub thread_id: u64, + /// Connection user, when the process list entry is visible. + #[serde(skip_serializing_if = "Option::is_none")] + pub user: Option, + /// Connection host, when the process list entry is visible. + #[serde(skip_serializing_if = "Option::is_none")] + pub host: Option, + /// Default database of the connection, when visible. + #[serde(skip_serializing_if = "Option::is_none")] + pub database: Option, + /// Currently running statement, empty when the session is idle. + pub query: String, +} diff --git a/crates/chat2db-contract/src/datasource.rs b/crates/chat2db-contract/src/datasource.rs index 1323be7..158ca62 100644 --- a/crates/chat2db-contract/src/datasource.rs +++ b/crates/chat2db-contract/src/datasource.rs @@ -115,6 +115,9 @@ pub struct Datasource { pub created_at_ms: String, /// Unix epoch milliseconds encoded as a decimal integer. pub updated_at_ms: String, + /// Optional `#RRGGBB` identity colour shown by the connection tree. + #[serde(skip_serializing_if = "Option::is_none")] + pub identity_color: Option, } /// Stable datasource collection returned by list APIs. @@ -142,6 +145,7 @@ mod tests { revision: "9007199254740993".to_owned(), created_at_ms: "1784900000000".to_owned(), updated_at_ms: "1784900000001".to_owned(), + identity_color: None, }; let json = serde_json::to_string(&response).expect("response must serialize"); diff --git a/crates/chat2db-contract/src/datasource_edit.rs b/crates/chat2db-contract/src/datasource_edit.rs index 9aad373..a994533 100644 --- a/crates/chat2db-contract/src/datasource_edit.rs +++ b/crates/chat2db-contract/src/datasource_edit.rs @@ -31,6 +31,9 @@ pub struct DatasourceEditProjection { pub has_secret: bool, /// Monotonic revision encoded as a decimal integer. pub revision: String, + /// Optional `#RRGGBB` identity colour shown by the connection tree. + #[serde(skip_serializing_if = "Option::is_none")] + pub identity_color: Option, } impl Debug for DatasourceEditProjection { @@ -47,6 +50,7 @@ impl Debug for DatasourceEditProjection { .field("ssh", &self.ssh) .field("has_secret", &self.has_secret) .field("revision", &self.revision) + .field("identity_color", &self.identity_color) .finish() } } @@ -72,6 +76,7 @@ mod tests { ssh: None, has_secret: true, revision: "3".to_owned(), + identity_color: None, }; let json = serde_json::to_string(&projection).expect("projection serializes"); diff --git a/crates/chat2db-contract/src/lib.rs b/crates/chat2db-contract/src/lib.rs index 5b0e122..f5b4424 100644 --- a/crates/chat2db-contract/src/lib.rs +++ b/crates/chat2db-contract/src/lib.rs @@ -32,15 +32,16 @@ pub use agent::{ UpdateProviderProfileRequest, }; pub use community::{ - BuildCommunityCreateSchemaRequest, BuildCommunityDmlRequest, BuildCommunityNamespaceSqlRequest, - CommunityBuiltSql, CommunityDatabase, CommunityDatabaseList, CommunityDmlAssignment, - CommunityDmlColumn, CommunityDmlRow, CommunityDmlStatement, CommunityDmlTarget, - CommunityDmlTemporalKind, CommunityDmlValue, CommunityDriverConfig, CommunityForeignKey, - CommunityForeignKeyList, CommunityFormattedSql, CommunityFunction, CommunityFunctionList, - CommunityFunctionParameter, CommunityFunctionParameterList, CommunityNamespaceSqlOperation, - CommunityParsedStatement, CommunityPlugin, CommunityPluginBehavior, CommunityPluginCatalog, - CommunityPluginServices, CommunityPrimaryKey, CommunityPrimaryKeyList, CommunityProcedure, - CommunityProcedureList, CommunityProcedureParameter, CommunityProcedureParameterList, + ActiveTransaction, BuildCommunityCreateSchemaRequest, BuildCommunityDmlRequest, + BuildCommunityNamespaceSqlRequest, CommunityBuiltSql, CommunityDatabase, CommunityDatabaseList, + CommunityDmlAssignment, CommunityDmlColumn, CommunityDmlRow, CommunityDmlStatement, + CommunityDmlTarget, CommunityDmlTemporalKind, CommunityDmlValue, CommunityDriverConfig, + CommunityForeignKey, CommunityForeignKeyList, CommunityFormattedSql, CommunityFunction, + CommunityFunctionList, CommunityFunctionParameter, CommunityFunctionParameterList, + CommunityNamespaceSqlOperation, CommunityParsedStatement, CommunityPlugin, + CommunityPluginBehavior, CommunityPluginCatalog, CommunityPluginServices, CommunityPrimaryKey, + CommunityPrimaryKeyList, CommunityProcedure, CommunityProcedureList, + CommunityProcedureParameter, CommunityProcedureParameterList, CommunityRoutineInvocationPreview, CommunityRoutineMigrationExecution, CommunityRoutineMigrationRequest, CommunitySchema, CommunitySchemaList, CommunitySqlAnalysis, CommunitySqlCompletion, CommunitySqlCompletionActiveSnippetSlot, @@ -115,7 +116,8 @@ pub use transfer::{ DmlExportFormat, DmlExportRequest, DmlExportSize, GenerateMysqlClassRequest, GeneratedMysqlClassSet, ImportFileRequest, OtherFileExportRequest, SqlFileExportRequest, TabularImportEncoding, TransferArtifact, TransferFileFormat, TransferSqlScope, TransferTask, - TransferTaskAccepted, TransferTaskKind, TransferTaskPage, TransferTaskStatus, + TransferTaskAccepted, TransferTaskEvent, TransferTaskKind, TransferTaskPage, + TransferTaskStatus, }; pub use workspace::{ AssignDatasourceNamespaceRequest, CreateWorkspaceNamespaceRequest, MoveWorkspaceNodeRequest, diff --git a/crates/chat2db-contract/src/transfer.rs b/crates/chat2db-contract/src/transfer.rs index 83b75be..5ba7408 100644 --- a/crates/chat2db-contract/src/transfer.rs +++ b/crates/chat2db-contract/src/transfer.rs @@ -233,6 +233,29 @@ pub struct GeneratedMysqlClassSet { pub files: Vec, } +/// One ordered progress event of a durable transfer task. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct TransferTaskEvent { + /// Event row id. + pub id: i64, + /// Owning transfer task id. + pub task_id: i64, + /// Monotonic per-task sequence starting at one. + pub sequence: u64, + /// `INFO`, `WARN`, or `ERROR`. + pub level: String, + /// Stable machine-readable event code, empty when none applies. + pub code: String, + /// Optional pipeline stage. + #[serde(skip_serializing_if = "Option::is_none")] + pub stage: Option, + /// Human-readable event message. + pub message: String, + /// Unix epoch milliseconds encoded as a decimal integer. + pub created_at_ms: String, +} + const fn default_true() -> bool { true } diff --git a/crates/chat2db-core/src/convert.rs b/crates/chat2db-core/src/convert.rs index 091463b..cb66bdc 100644 --- a/crates/chat2db-core/src/convert.rs +++ b/crates/chat2db-core/src/convert.rs @@ -92,6 +92,7 @@ pub(crate) fn datasource(record: storage::DatasourceRecord) -> contract::Datasou revision: record.revision.to_string(), created_at_ms: record.created_at_ms.to_string(), updated_at_ms: record.updated_at_ms.to_string(), + identity_color: record.identity_color, } } diff --git a/crates/chat2db-core/src/datasource_edit.rs b/crates/chat2db-core/src/datasource_edit.rs index 6278871..b880723 100644 --- a/crates/chat2db-core/src/datasource_edit.rs +++ b/crates/chat2db-core/src/datasource_edit.rs @@ -64,6 +64,7 @@ impl Application { ssh: projected.ssh, has_secret, revision: record.revision.to_string(), + identity_color: record.identity_color, }) } diff --git a/crates/chat2db-core/src/lib.rs b/crates/chat2db-core/src/lib.rs index 9f458cc..6b98712 100644 --- a/crates/chat2db-core/src/lib.rs +++ b/crates/chat2db-core/src/lib.rs @@ -13,6 +13,7 @@ mod error; mod large_value; mod legacy_community_import; mod mysql_account; +mod mysql_active_transaction; mod mysql_dashboard; pub mod mysql_ddl; mod mysql_schema_diff; @@ -672,6 +673,28 @@ impl Application { .map(convert::datasource) } + /// Replaces only the identity colour of one datasource. + /// + /// The colour is presentation metadata and does not participate in the + /// revision CAS used for public datasource fields. + /// + /// # Errors + /// + /// Returns validation, not-found, availability, or storage failures. + pub async fn update_datasource_identity_color( + &self, + id: &str, + identity_color: Option, + ) -> Result { + let storage = self.require_storage()?; + let id = id.to_owned(); + storage_call(move || { + storage.update_datasource_identity_color(&id, identity_color.as_deref()) + }) + .await + .map(convert::datasource) + } + fn require_managed_driver(&self, driver_id: &str) -> Result<(), AppError> { if let Some(driver) = self.native_driver_for_datasource_driver_id(driver_id) && (driver.connection().is_some() @@ -839,6 +862,26 @@ impl Application { } /// Stops work admission and requests cancellation of active queries and agent runs. + /// Stops accepting new transfer work while the product prepares to exit. + /// + /// # Errors + /// + /// Returns availability failures only. + pub async fn prepare_user_exit(&self) -> Result<(), AppError> { + *self.inner.accepting_work.lock().await = false; + Ok(()) + } + + /// Re-enables transfer work after a cancelled exit. + /// + /// # Errors + /// + /// Returns availability failures only. + pub async fn abort_user_exit(&self) -> Result<(), AppError> { + *self.inner.accepting_work.lock().await = true; + Ok(()) + } + pub async fn begin_shutdown(&self) { let mut accepting_work = self.inner.accepting_work.lock().await; if !*accepting_work { diff --git a/crates/chat2db-core/src/mysql_active_transaction.rs b/crates/chat2db-core/src/mysql_active_transaction.rs new file mode 100644 index 0000000..e1d685f --- /dev/null +++ b/crates/chat2db-core/src/mysql_active_transaction.rs @@ -0,0 +1,155 @@ +//! `MySQL` active-transaction inspection for the Community operations monitor. + +use std::time::Duration; + +use chat2db_contract::ActiveTransaction; +use mysql_async::{Error as MysqlError, prelude::Queryable}; + +use crate::{ + AppError, Application, + native_mysql::{finish_connection, open_resolved_connection, resolve_native_connection}, +}; + +const ACTIVE_TRANSACTION_TIMEOUT: Duration = Duration::from_secs(30); +const ACTIVE_TRANSACTION_UNAVAILABLE: &str = "mysql.activeTransaction.unavailable"; +const ACTIVE_TRANSACTION_PRIVILEGE: &str = "mysql.activeTransaction.processPrivilegeRequired"; + +const SELECT_ACTIVE_TRANSACTIONS: &str = " + SELECT t.trx_id, + t.trx_state, + TIMESTAMPDIFF(MICROSECOND, t.trx_started, NOW(6)) DIV 1000, + t.trx_isolation_level, + t.trx_rows_locked, + t.trx_rows_modified, + t.trx_lock_structs, + t.trx_mysql_thread_id, + p.USER, + p.HOST, + p.DB, + COALESCE(p.INFO, '') + FROM information_schema.INNODB_TRX t + LEFT JOIN information_schema.PROCESSLIST p ON p.ID = t.trx_mysql_thread_id + ORDER BY t.trx_started, t.trx_id"; + +type ActiveTransactionRow = ( + String, + String, + Option, + String, + u64, + u64, + u64, + u64, + Option, + Option, + Option, + Option, +); + +impl Application { + /// Lists active `InnoDB` transactions of one native `MySQL` datasource. + /// + /// # Errors + /// + /// Returns connection, privilege, timeout, or query failures. A missing + /// `PROCESS` privilege answers `mysql.activeTransaction.processPrivilegeRequired` + /// so the monitor can explain the requirement. + pub async fn list_mysql_active_transactions( + &self, + datasource_id: &str, + ) -> Result, AppError> { + let resolved = resolve_native_connection(self, datasource_id).await?; + let mut connection = open_resolved_connection(&resolved).await?; + let rows = match tokio::time::timeout( + ACTIVE_TRANSACTION_TIMEOUT, + connection.query::(SELECT_ACTIVE_TRANSACTIONS), + ) + .await + { + Ok(Ok(rows)) => rows, + Ok(Err(error)) => { + return finish_connection(connection, Err(active_transaction_failure(&error))) + .await; + } + Err(_) => { + return finish_connection( + connection, + Err(AppError::invalid( + ACTIVE_TRANSACTION_UNAVAILABLE, + "The MySQL active transaction query timed out", + )), + ) + .await; + } + }; + finish_connection(connection, Ok(())).await?; + Ok(rows + .into_iter() + .map( + |( + trx_id, + state, + age_millis, + isolation_level, + rows_locked, + rows_modified, + lock_structs, + thread_id, + user, + host, + database, + query, + )| { + let age_millis = age_millis.unwrap_or_default(); + ActiveTransaction { + trx_id, + state, + started_at_ms: current_epoch_millis() + .saturating_sub(age_millis) + .to_string(), + age_seconds: u64::try_from(age_millis / 1_000).unwrap_or_default(), + isolation_level, + rows_locked, + rows_modified, + lock_structs, + thread_id, + user, + host, + database, + query: query.unwrap_or_default(), + } + }, + ) + .collect()) + } +} + +fn current_epoch_millis() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .ok() + .and_then(|duration| i64::try_from(duration.as_millis()).ok()) + .unwrap_or_default() +} + +fn active_transaction_failure(error: &MysqlError) -> AppError { + match error { + MysqlError::Server(server) if server.code == 1227 || server.code == 1142 => { + AppError::invalid( + ACTIVE_TRANSACTION_PRIVILEGE, + "Inspecting active transactions requires the PROCESS privilege", + ) + } + MysqlError::Server(server) => AppError::invalid( + ACTIVE_TRANSACTION_UNAVAILABLE, + format!( + "The MySQL active transaction query failed: {}", + server.message + ), + ), + _ => AppError::invalid( + ACTIVE_TRANSACTION_UNAVAILABLE, + "The MySQL connection ended before the active transaction query completed", + ), + } +} diff --git a/crates/chat2db-core/src/transfer/format.rs b/crates/chat2db-core/src/transfer/format.rs index e7abf94..74230f2 100644 --- a/crates/chat2db-core/src/transfer/format.rs +++ b/crates/chat2db-core/src/transfer/format.rs @@ -162,6 +162,56 @@ fn read_csv( Ok(ImportedTable { columns, rows }) } +/// Lists worksheet names of one spreadsheet file, empty for other formats. +pub(crate) fn worksheet_names( + path: &Path, + format: TransferFileFormat, +) -> Result, AppError> { + match format { + TransferFileFormat::Csv | TransferFileFormat::Sql => Ok(Vec::new()), + TransferFileFormat::Xls | TransferFileFormat::Xlsx => { + if format == TransferFileFormat::Xlsx { + validate_xlsx_archive(path)?; + } + let file = File::open(path).map_err(|_| { + AppError::not_found( + "import_file_not_found", + "The selected import file could not be opened", + ) + })?; + let workbook = match format { + TransferFileFormat::Xls => xls::core::xls::read(file), + TransferFileFormat::Xlsx => xls::core::xlsx::read(file), + TransferFileFormat::Csv | TransferFileFormat::Sql => unreachable!(), + } + .map_err(format_error)?; + Ok(workbook + .sheets + .iter() + .map(|sheet| sheet.name.clone()) + .collect()) + } + } +} + +/// Reads a row-bounded preview with string-encoded cells. +pub(crate) fn read_tabular_preview( + path: &Path, + format: TransferFileFormat, + contains_header: bool, + limit: usize, +) -> Result<(Vec, Vec>), AppError> { + let mut table = read_tabular_file(path, format, contains_header, TabularImportEncoding::Plain)?; + let columns = table.columns.take().unwrap_or_default(); + table.rows.truncate(limit); + let rows = table + .rows + .into_iter() + .map(|row| row.iter().map(encode_tabular_value).collect()) + .collect(); + Ok((columns, rows)) +} + fn read_spreadsheet( path: &Path, format: TransferFileFormat, diff --git a/crates/chat2db-core/src/transfer/mod.rs b/crates/chat2db-core/src/transfer/mod.rs index ae4d38b..c938466 100644 --- a/crates/chat2db-core/src/transfer/mod.rs +++ b/crates/chat2db-core/src/transfer/mod.rs @@ -11,8 +11,8 @@ use std::{ use chat2db_contract::{ DmlExportFormat, DmlExportRequest, DmlExportSize, GenerateMysqlClassRequest, GeneratedMysqlClassSet, ImportFileRequest, OtherFileExportRequest, SqlFileExportRequest, - TransferArtifact, TransferFileFormat, TransferTask, TransferTaskAccepted, TransferTaskKind, - TransferTaskPage, TransferTaskStatus, + TransferArtifact, TransferFileFormat, TransferTask, TransferTaskAccepted, TransferTaskEvent, + TransferTaskKind, TransferTaskPage, TransferTaskStatus, }; use chat2db_storage::{ CreateTransferTask, ResolvedTransferArtifact, Storage, StorageError, StoredTransferTaskKind, @@ -27,6 +27,7 @@ use tokio_util::sync::CancellationToken; use crate::{AppError, Application, storage_call}; const MAX_TASK_PAGE_SIZE: u32 = 100; +const MAX_TASK_EVENT_LIMIT: u32 = 200; const MAX_TRANSFER_FAILURE_MESSAGE_BYTES: usize = 64 * 1024; const TRANSFER_FAILURE_TRUNCATION_SUFFIX: &str = "\n[truncated]"; const TERMINAL_RETRY_INITIAL_DELAY: Duration = Duration::from_millis(25); @@ -444,6 +445,7 @@ impl Application { &self, request: ImportFileRequest, ) -> Result { + self.ensure_accepting_transfer_work().await?; self.import_file(request).await } @@ -478,6 +480,7 @@ impl Application { &self, request: SqlFileExportRequest, ) -> Result { + self.ensure_accepting_transfer_work().await?; self.export_sql_file(request).await } @@ -524,9 +527,26 @@ impl Application { &self, request: OtherFileExportRequest, ) -> Result { + self.ensure_accepting_transfer_work().await?; self.export_other_file(request).await } + /// Rejects new transfer work while the product is preparing to exit. + /// + /// # Errors + /// + /// Returns `runtime_not_accepting_work` when an exit was prepared and not + /// aborted yet. + pub(crate) async fn ensure_accepting_transfer_work(&self) -> Result<(), AppError> { + if *self.inner.accepting_work.lock().await { + return Ok(()); + } + Err(AppError::invalid( + "runtime_not_accepting_work", + "The application is preparing to exit and no longer accepts new transfer tasks", + )) + } + /// Lists retained transfer tasks newest first. /// /// # Errors @@ -616,6 +636,103 @@ impl Application { Ok(()) } + /// Deletes one finished transfer task and its artifact. + /// + /// # Errors + /// + /// Returns not-found, active-task, or durable-storage failures. + pub async fn delete_transfer_task(&self, task_id: i64) -> Result<(), AppError> { + let storage = self.require_storage()?; + storage_call(move || storage.delete_transfer_task(task_id)).await?; + Ok(()) + } + + /// Resolves one staged import preview artifact to its path and format. + async fn staged_import_file( + &self, + file_id: &str, + ) -> Result<(std::path::PathBuf, TransferFileFormat), AppError> { + let download = self.transfer_artifact_download(file_id.trim()).await?; + let format = match download + .artifact + .format + .trim() + .to_ascii_uppercase() + .as_str() + { + "CSV" => TransferFileFormat::Csv, + "XLS" => TransferFileFormat::Xls, + "XLSX" => TransferFileFormat::Xlsx, + "SQL" => TransferFileFormat::Sql, + other => { + return Err(AppError::invalid( + "unsupported_import_file", + format!("Unsupported import preview format: {other}"), + )); + } + }; + Ok((download.path, format)) + } + + /// Lists worksheet names of one staged import preview file. + /// + /// # Errors + /// + /// Returns not-found, expiry, corrupted-artifact, or format failures. + pub async fn import_preview_sheets(&self, file_id: &str) -> Result, AppError> { + let (path, format) = self.staged_import_file(file_id).await?; + format::worksheet_names(&path, format) + } + + /// Reads a bounded preview of one staged import file. + /// + /// # Errors + /// + /// Returns not-found, expiry, corrupted-artifact, or format failures. + pub async fn import_preview_table( + &self, + file_id: &str, + contains_header: bool, + limit: u32, + ) -> Result<(Vec, Vec>), AppError> { + let limit = usize::try_from(limit).unwrap_or(usize::MAX); + let (path, format) = self.staged_import_file(file_id).await?; + format::read_tabular_preview(&path, format, contains_header, limit) + } + + /// Lists ordered progress events of one transfer task. + /// + /// # Errors + /// + /// Returns validation, not-found, or durable-storage failures. + pub async fn list_transfer_task_events( + &self, + task_id: i64, + after_sequence: Option, + before_sequence: Option, + limit: Option, + ) -> Result, AppError> { + let limit = limit.unwrap_or(MAX_TASK_EVENT_LIMIT); + let storage = self.require_storage()?; + let records = storage_call(move || { + storage.list_transfer_task_events(task_id, after_sequence, before_sequence, limit) + }) + .await?; + Ok(records + .into_iter() + .map(|record| TransferTaskEvent { + id: record.id, + task_id: record.task_id, + sequence: record.sequence, + level: record.level, + code: record.code, + stage: record.stage, + message: record.message, + created_at_ms: record.created_at_ms.to_string(), + }) + .collect()) + } + /// Resolves a managed artifact and its owner-only local path for a delivery adapter. /// /// # Errors diff --git a/crates/chat2db-storage/migrations/009_community_client_routes.sql b/crates/chat2db-storage/migrations/009_community_client_routes.sql new file mode 100644 index 0000000..9c81017 --- /dev/null +++ b/crates/chat2db-storage/migrations/009_community_client_routes.sql @@ -0,0 +1,29 @@ +BEGIN IMMEDIATE; + +ALTER TABLE datasources ADD COLUMN identity_color TEXT + CHECK (identity_color IS NULL OR length(CAST(identity_color AS BLOB)) <= 16); + +ALTER TABLE transfer_tasks ADD COLUMN task_type TEXT + CHECK (task_type IS NULL OR length(CAST(task_type AS BLOB)) <= 64); + +ALTER TABLE transfer_tasks ADD COLUMN format TEXT + CHECK (format IS NULL OR length(CAST(format AS BLOB)) <= 32); + +CREATE TABLE transfer_task_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id INTEGER NOT NULL REFERENCES transfer_tasks(id) ON DELETE CASCADE, + sequence INTEGER NOT NULL CHECK (sequence >= 1), + level TEXT NOT NULL CHECK (level IN ('INFO', 'WARN', 'ERROR')), + code TEXT NOT NULL DEFAULT '', + stage TEXT, + message TEXT NOT NULL, + details_json TEXT, + created_at_ms INTEGER NOT NULL, + UNIQUE (task_id, sequence) +) STRICT; + +CREATE INDEX transfer_task_events_task_idx + ON transfer_task_events (task_id, sequence); + +PRAGMA user_version = 9; +COMMIT; diff --git a/crates/chat2db-storage/src/datasource.rs b/crates/chat2db-storage/src/datasource.rs index 7f1c159..379292f 100644 --- a/crates/chat2db-storage/src/datasource.rs +++ b/crates/chat2db-storage/src/datasource.rs @@ -63,6 +63,8 @@ pub struct DatasourceRecord { pub created_at_ms: i64, /// Last update time as Unix epoch milliseconds. pub updated_at_ms: i64, + /// Optional `#RRGGBB` identity colour shown by the connection tree. + pub identity_color: Option, } impl std::fmt::Debug for DatasourceRecord { @@ -76,6 +78,7 @@ impl std::fmt::Debug for DatasourceRecord { .field("revision", &self.revision) .field("created_at_ms", &self.created_at_ms) .field("updated_at_ms", &self.updated_at_ms) + .field("identity_color", &self.identity_color) .finish() } } @@ -128,6 +131,7 @@ impl Storage { revision: 1, created_at_ms: timestamp, updated_at_ms: timestamp, + identity_color: None, }; let mutation = (|| -> Result<(), StorageError> { @@ -189,7 +193,7 @@ impl Storage { let connection = self.connection()?; let mut statement = connection.prepare( "SELECT id, name, driver_id, secret_ref, - revision, created_at_ms, updated_at_ms + revision, created_at_ms, updated_at_ms, identity_color FROM datasources ORDER BY created_at_ms, id", )?; let rows = statement.query_map([], raw_datasource)?; @@ -249,6 +253,7 @@ impl Storage { revision: next_revision, created_at_ms: current.created_at_ms, updated_at_ms: timestamp, + identity_color: current.identity_color.clone(), }; let mutation = (|| -> Result<(), StorageError> { @@ -309,6 +314,35 @@ impl Storage { } } + /// Replaces only the identity colour of one datasource. + /// + /// The colour is presentation metadata, so it does not participate in the + /// revision CAS used for public fields. + /// + /// # Errors + /// + /// Returns validation, not-found, or `SQLite` failures. + pub fn update_datasource_identity_color( + &self, + id: &str, + identity_color: Option<&str>, + ) -> Result { + let identity_color = validate_identity_color(identity_color)?; + let timestamp = now_millis()?; + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + let changed = transaction.execute( + "UPDATE datasources SET identity_color = ?1, updated_at_ms = ?2 WHERE id = ?3", + params![identity_color, timestamp, id], + )?; + if changed == 0 { + return Err(StorageError::DatasourceNotFound(id.to_owned())); + } + transaction.commit()?; + self.get_datasource(id)? + .ok_or_else(|| StorageError::DatasourceNotFound(id.to_owned())) + } + /// Deletes a datasource using revision CAS and retires its vault reference. /// /// # Errors @@ -624,7 +658,30 @@ fn validate_datasource(name: &str, driver_id: &str) -> Result<(), StorageError> Ok(()) } -type RawDatasource = (String, String, String, Option, i64, i64, i64); +/// Normalizes the optional `#RRGGBB` identity colour. +fn validate_identity_color(value: Option<&str>) -> Result, StorageError> { + let Some(value) = value.map(str::trim).filter(|value| !value.is_empty()) else { + return Ok(None); + }; + let hex = value.strip_prefix('#').unwrap_or_default(); + if hex.len() != 6 || !hex.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Err(StorageError::InvalidDatasource( + "identity colour must use the #RRGGBB form", + )); + } + Ok(Some(format!("#{}", hex.to_ascii_uppercase()))) +} + +type RawDatasource = ( + String, + String, + String, + Option, + i64, + i64, + i64, + Option, +); fn raw_datasource(row: &rusqlite::Row<'_>) -> rusqlite::Result { Ok(( @@ -635,11 +692,12 @@ fn raw_datasource(row: &rusqlite::Row<'_>) -> rusqlite::Result { row.get(4)?, row.get(5)?, row.get(6)?, + row.get(7)?, )) } fn decode_datasource(raw: RawDatasource) -> Result { - let (id, name, driver_id, secret_ref, revision, created, updated) = raw; + let (id, name, driver_id, secret_ref, revision, created, updated, identity_color) = raw; Ok(DatasourceRecord { id, name, @@ -649,6 +707,7 @@ fn decode_datasource(raw: RawDatasource) -> Result Result<(), StorageError> { connection, include_str!("../migrations/008_community_dashboard.sql"), )?; + version = 8; + } + if version == 8 { + apply_migration( + connection, + include_str!("../migrations/009_community_client_routes.sql"), + )?; } Ok(()) } diff --git a/crates/chat2db-storage/src/transfer.rs b/crates/chat2db-storage/src/transfer.rs index bdaa271..dafa377 100644 --- a/crates/chat2db-storage/src/transfer.rs +++ b/crates/chat2db-storage/src/transfer.rs @@ -18,6 +18,7 @@ const MAX_LOG_BYTES: usize = 256 * 1024; const MAX_LOG_BYTES_I64: i64 = 256 * 1024; const MAX_FILE_NAME_BYTES: usize = 1_024; const MAX_MEDIA_TYPE_BYTES: usize = 255; +const MAX_TASK_EVENT_PAGE: u32 = 200; /// Durable transfer category. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -149,6 +150,19 @@ pub struct ResolvedTransferArtifact { pub file: File, } +/// One ordered progress event of a durable transfer task. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TransferTaskEventRecord { + pub id: i64, + pub task_id: i64, + pub sequence: u64, + pub level: String, + pub code: String, + pub stage: Option, + pub message: String, + pub created_at_ms: i64, +} + /// Startup cleanup and task recovery report. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct TransferRecoveryReport { @@ -427,6 +441,65 @@ impl Storage { .map_err(StorageError::from) } + /// Lists ordered events of one durable task inside a sequence window. + /// + /// # Errors + /// + /// Returns validation, not-found, range, or `SQLite` failures. + pub fn list_transfer_task_events( + &self, + task_id: i64, + after_sequence: Option, + before_sequence: Option, + limit: u32, + ) -> Result, StorageError> { + if limit == 0 || limit > MAX_TASK_EVENT_PAGE { + return Err(StorageError::InvalidTransfer( + "event limit must be between 1 and 200", + )); + } + let connection = self.connection()?; + ensure_task_exists(&connection, task_id)?; + let after = sequence_bound(after_sequence)?.unwrap_or(0); + let before = sequence_bound(before_sequence)?.unwrap_or(i64::MAX); + let mut statement = connection.prepare( + "SELECT id, task_id, sequence, level, code, stage, message, created_at_ms + FROM transfer_task_events + WHERE task_id = ?1 AND sequence > ?2 AND sequence < ?3 + ORDER BY sequence + LIMIT ?4", + )?; + let rows = + statement.query_map(params![task_id, after, before, i64::from(limit)], |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, i64>(1)?, + row.get::<_, i64>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, Option>(5)?, + row.get::<_, String>(6)?, + row.get::<_, i64>(7)?, + )) + })?; + let mut events = Vec::new(); + for row in rows { + let (id, task_id, sequence, level, code, stage, message, created_at_ms) = row?; + events.push(TransferTaskEventRecord { + id, + task_id, + sequence: u64::try_from(sequence) + .map_err(|_| StorageError::NumericRange("event sequence"))?, + level, + code, + stage, + message, + created_at_ms, + }); + } + Ok(events) + } + /// Transitions a queued task to running. /// /// # Errors @@ -476,8 +549,9 @@ impl Storage { .map_err(|_| StorageError::NumericRange("transfer progress total")) }) .transpose()?; - let connection = self.connection()?; - let updated = connection.execute( + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + let updated = transaction.execute( "UPDATE transfer_tasks SET progress_current = MAX(progress_current, ?2), progress_total = COALESCE(?3, progress_total), @@ -498,8 +572,19 @@ impl Storage { ], )?; if updated == 1 { + append_transfer_task_event( + &transaction, + id, + "INFO", + "", + None, + info.unwrap_or(description), + timestamp, + )?; + transaction.commit()?; return Ok(()); } + drop(transaction); ensure_task_exists(&connection, id)?; Err(StorageError::InvalidTransfer( "task progress can only update while running", @@ -584,8 +669,9 @@ impl Storage { pub fn complete_transfer_task(&self, id: i64, message: &str) -> Result<(), StorageError> { validate_text(message, MAX_LOG_BYTES, "task completion message")?; let timestamp = now_millis()?; - let connection = self.connection()?; - let updated = connection.execute( + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + let updated = transaction.execute( "UPDATE transfer_tasks SET status = 'succeeded', progress_description = 'Completed', info_log = CASE WHEN ?2 = '' THEN info_log @@ -601,11 +687,26 @@ impl Storage { ], )?; if updated != 1 { + drop(transaction); ensure_task_exists(&connection, id)?; return Err(StorageError::InvalidTransfer( "task cannot complete in its current state", )); } + append_transfer_task_event( + &transaction, + id, + "INFO", + "", + None, + if message.is_empty() { + "Completed" + } else { + message + }, + timestamp, + )?; + transaction.commit()?; // The terminal state is already durable. Retention is recoverable // maintenance and must not make callers retry a committed transition. let _ = self.prune_transfer_tasks(); @@ -624,8 +725,9 @@ impl Storage { } validate_text(message, MAX_LOG_BYTES, "task terminal message")?; let timestamp = now_millis()?; - let connection = self.connection()?; - let updated = connection.execute( + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + let updated = transaction.execute( "UPDATE transfer_tasks SET status = ?2, progress_description = ?3, error_log = ?4, updated_at_ms = ?5, finished_at_ms = ?5 @@ -633,11 +735,19 @@ impl Storage { params![id, status.as_str(), description, message, timestamp], )?; if updated != 1 { + drop(transaction); ensure_task_exists(&connection, id)?; return Err(StorageError::InvalidTransfer( "task is already in a terminal state", )); } + let level = if status == StoredTransferTaskStatus::Cancelled { + "WARN" + } else { + "ERROR" + }; + append_transfer_task_event(&transaction, id, level, "", None, message, timestamp)?; + transaction.commit()?; // The terminal state is already durable. Retention is recoverable // maintenance and must not make callers retry a committed transition. let _ = self.prune_transfer_tasks(); @@ -875,6 +985,49 @@ impl Storage { }) } + /// Deletes one terminal transfer task and its artifact file. + /// + /// # Errors + /// + /// Returns not-found, active-task, or `SQLite`/filesystem failures. + pub fn delete_transfer_task(&self, id: i64) -> Result<(), StorageError> { + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + let status = transaction + .query_row( + "SELECT status FROM transfer_tasks WHERE id = ?1", + [id], + |row| row.get::<_, String>(0), + ) + .optional()? + .ok_or(StorageError::TransferTaskNotFound(id))?; + if !StoredTransferTaskStatus::parse(&status)?.is_terminal() { + return Err(StorageError::InvalidTransfer( + "only finished transfer tasks can be deleted", + )); + } + let storage_name = transaction + .query_row( + "SELECT storage_name FROM transfer_artifacts WHERE task_id = ?1", + [id], + |row| row.get::<_, String>(0), + ) + .optional()?; + transaction.execute("DELETE FROM transfer_tasks WHERE id = ?1", [id])?; + transaction.commit()?; + if let Some(name) = storage_name { + validate_storage_name(&name)?; + let path = self.inner.artifacts_dir.join(name); + match fs::remove_file(&path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(StorageError::io(path, error)), + } + sync_directory(&self.inner.artifacts_dir)?; + } + Ok(()) + } + fn prune_transfer_tasks(&self) -> Result<(), StorageError> { let mut connection = self.connection()?; let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; @@ -1112,6 +1265,39 @@ fn validate_text(value: &str, maximum: usize, field: &'static str) -> Result<(), Ok(()) } +/// Appends one ordered event to a task inside an open transaction. +fn append_transfer_task_event( + transaction: &rusqlite::Transaction<'_>, + task_id: i64, + level: &str, + code: &str, + stage: Option<&str>, + message: &str, + timestamp: i64, +) -> Result<(), StorageError> { + if message.is_empty() { + return Ok(()); + } + validate_text(message, MAX_LOG_BYTES, "task event message")?; + transaction.execute( + "INSERT INTO transfer_task_events ( + task_id, sequence, level, code, stage, message, details_json, created_at_ms + ) VALUES ( + ?1, + (SELECT COALESCE(MAX(sequence), 0) + 1 FROM transfer_task_events WHERE task_id = ?1), + ?2, ?3, ?4, ?5, NULL, ?6 + )", + params![task_id, level, code, stage, message, timestamp], + )?; + Ok(()) +} + +fn sequence_bound(value: Option) -> Result, StorageError> { + value + .map(|value| i64::try_from(value).map_err(|_| StorageError::NumericRange("event sequence"))) + .transpose() +} + fn ensure_task_exists(connection: &rusqlite::Connection, id: i64) -> Result<(), StorageError> { let exists = connection.query_row( "SELECT EXISTS(SELECT 1 FROM transfer_tasks WHERE id = ?1)", @@ -1188,6 +1374,69 @@ mod tests { .id } + #[test] + fn delete_transfer_task_removes_finished_tasks_and_rejects_active_ones() { + let directory = TempDir::new().expect("temp directory"); + let storage = open(&directory); + let active = task(&storage, 1); + assert!(matches!( + storage.delete_transfer_task(active), + Err(StorageError::InvalidTransfer(_)) + )); + storage.start_transfer_task(active).expect("task starts"); + storage + .complete_transfer_task(active, "done") + .expect("task completes"); + storage.delete_transfer_task(active).expect("task deletes"); + assert!( + storage + .get_transfer_task(active) + .expect("lookup succeeds") + .is_none() + ); + assert!(matches!( + storage.delete_transfer_task(active), + Err(StorageError::TransferTaskNotFound(_)) + )); + } + + #[test] + fn task_events_record_progress_and_terminal_outcomes_in_sequence() { + let directory = TempDir::new().expect("temp directory"); + let storage = open(&directory); + let id = task(&storage, 1); + storage.start_transfer_task(id).expect("task starts"); + storage + .update_transfer_progress(id, 1, Some(2), "Importing", Some("Importing table rows")) + .expect("progress updates"); + storage + .update_transfer_progress(id, 2, Some(2), "Finishing", None) + .expect("progress updates"); + storage + .fail_transfer_task(id, "Import failed") + .expect("task fails"); + + let events = storage + .list_transfer_task_events(id, None, None, 10) + .expect("events list"); + assert_eq!(events.len(), 3); + assert_eq!(events[0].sequence, 1); + assert_eq!(events[0].level, "INFO"); + assert_eq!(events[0].message, "Importing table rows"); + assert_eq!(events[1].message, "Finishing"); + assert_eq!(events[2].level, "ERROR"); + assert_eq!(events[2].message, "Import failed"); + + let window = storage + .list_transfer_task_events(id, Some(1), None, 10) + .expect("window lists"); + assert_eq!(window.len(), 2); + assert!(matches!( + storage.list_transfer_task_events(id, None, None, 0), + Err(StorageError::InvalidTransfer(_)) + )); + } + #[test] fn artifact_publish_is_atomic_and_completes_the_task() { let directory = TempDir::new().expect("temp directory"); diff --git a/docs/architecture.md b/docs/architecture.md index 1329ed3..2f0d9c0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -47,7 +47,11 @@ The current product frontend is not the former repository-owned replacement workbench. The build exports a locked Community frontend commit that retains the original pages, components, interactions, and styles while applying a reviewable host-transport patch for CSP-safe callbacks and Web/Desktop file -flows. `scripts/community-frontend.lock.json` binds both commit and tree. Web maps its historical `/api` +flows. `scripts/community-frontend.lock.json` binds both commit and tree, and is +independent of the submodule checkout: `third_party/chat2db-community` stays on +the Java compatibility baseline recorded in +`third_party/community-h2-classpath.lock`, while the locked frontend commit is +fetched into the submodule object store and archived from there. Web maps its historical `/api` contract through Axum; desktop maps the existing `window.javaQuery` contract through one Tauri `legacy_request` command. Both paths call the same Rust legacy dispatcher. The implemented product slice covers native MySQL connection diff --git a/docs/releases.md b/docs/releases.md index 750dd3d..add3d69 100644 --- a/docs/releases.md +++ b/docs/releases.md @@ -25,7 +25,10 @@ commit in `main`. The tag annotation records: - `source_repository`: `OtterMind/Chat2DB-Rust`; - `source_commit`: the complete release commit SHA; -- `community_ref`: the pinned submodule commit; +- `community_ref`: the locked frontend commit from + `scripts/community-frontend.lock.json` (the Community revision shipped as the + visible UI; the Java compatibility baseline stays recorded in + `third_party/community-h2-classpath.lock`); - `package_repository`: `OtterMind/Chat2DB-Rust`; - `workflow_ref`: the complete release commit SHA; - `product_workflow`: `.github/workflows/package.yml`; and diff --git a/scripts/build-linux-package.sh b/scripts/build-linux-package.sh index c186eb6..a9a1e2a 100755 --- a/scripts/build-linux-package.sh +++ b/scripts/build-linux-package.sh @@ -119,7 +119,7 @@ cp -- "${appimage_artifacts[0]}" "${deb_artifacts[0]}" "${rpm_artifacts[0]}" "${ echo "Chat2DB Rust Linux package" echo "version=${version}" echo "git_commit=$(git -C "${repository_root}" rev-parse HEAD)" - echo "community_commit=$(git -C "${repository_root}/third_party/chat2db-community" rev-parse HEAD)" + echo "community_commit=$(node -e 'process.stdout.write(require(process.argv[1]).commit)' "${repository_root}/scripts/community-frontend.lock.json")" echo "architecture=$(uname -m)" echo "target=linux" echo "rust_toolchain=${rust_toolchain}" diff --git a/scripts/build-macos-package.sh b/scripts/build-macos-package.sh index 1149fec..54ce69e 100755 --- a/scripts/build-macos-package.sh +++ b/scripts/build-macos-package.sh @@ -308,7 +308,7 @@ verification_directory="" shasum -a 256 -- "$(basename "${zip_path}")" "$(basename "${dmg_path}")" > SHA256SUMS ) git_commit="$(git -C "${repository_root}" rev-parse HEAD)" -community_commit="$(git -C "${repository_root}/third_party/chat2db-community" rev-parse HEAD)" +community_commit="$(node -e 'process.stdout.write(require(process.argv[1]).commit)' "${repository_root}/scripts/community-frontend.lock.json")" app_kib="$(du -sk "${app_path}" | awk '{ print $1 }')" signature_details="$(codesign -dv --verbose=4 "${app_path}" 2>&1)" signing_team_id="$(awk -F= '/^TeamIdentifier=/ { print $2; exit }' <<<"${signature_details}")" diff --git a/scripts/build-windows-package.sh b/scripts/build-windows-package.sh index a0e88a5..6512ffd 100755 --- a/scripts/build-windows-package.sh +++ b/scripts/build-windows-package.sh @@ -121,7 +121,7 @@ cp -- "${msi_artifacts[0]}" "${package_directory}/" echo "Chat2DB Rust Windows package" echo "version=${version}" echo "git_commit=$(git -C "${repository_root}" rev-parse HEAD)" - echo "community_commit=$(git -C "${repository_root}/third_party/chat2db-community" rev-parse HEAD)" + echo "community_commit=$(node -e 'process.stdout.write(require(process.argv[1]).commit)' "${repository_root}/scripts/community-frontend.lock.json")" echo "architecture=x86_64" echo "target=windows" echo "rust_toolchain=${rust_toolchain}" diff --git a/scripts/community-frontend.lock.json b/scripts/community-frontend.lock.json index 0b27cb9..c0ed48c 100644 --- a/scripts/community-frontend.lock.json +++ b/scripts/community-frontend.lock.json @@ -2,7 +2,7 @@ "repository": "https://github.com/OtterMind/Chat2DB.git", "submodulePath": "third_party/chat2db-community", "sourcePath": "chat2db-community-client", - "commit": "a117f4895b3c63bbcedd2f79ab867ead800cf5c5", - "tree": "d8a24c8322700bc8373e165674092d4f0b446569", + "commit": "dedc12663d8d17eae239f79c2f31205406524aa5", + "tree": "9f4be62c090886b726e20fd9bf274d7bd279d5e1", "packageManager": "yarn@1.22.22" } diff --git a/scripts/community-frontend.mjs b/scripts/community-frontend.mjs index a3acfc0..62fe11b 100644 --- a/scripts/community-frontend.mjs +++ b/scripts/community-frontend.mjs @@ -90,21 +90,42 @@ function replaceDirectory(target) { mkdirSync(target, { recursive: true }); } -function verifySource() { - if (!existsSync(join(SUBMODULE_DIR, '.git'))) { - fail(`submodule is unavailable; run git submodule update --init --recursive ${lock.submodulePath}`); - } +function hasLockedCommit() { + const result = spawnSync('git', ['-C', SUBMODULE_DIR, 'cat-file', '-e', `${lock.commit}^{commit}`]); + return result.status === 0; +} - const indexedSubmodule = capture('git', ['ls-files', '--stage', '--', lock.submodulePath]); - const expectedIndex = `160000 ${lock.commit} 0\t${lock.submodulePath}`; - if (indexedSubmodule !== expectedIndex) { - fail(`repository index must pin ${lock.submodulePath} at ${lock.commit}`); +// The submodule checkout tracks the Java compatibility baseline recorded in +// `third_party/community-h2-classpath.lock`, so the locked frontend commit is +// fetched into the submodule object store instead of being checked out. +function ensureLockedCommit() { + if (hasLockedCommit()) { + return; + } + spawnSync('git', ['-C', SUBMODULE_DIR, 'fetch', '--no-tags', 'origin', lock.commit], { + stdio: 'inherit', + }); + if (!hasLockedCommit()) { + run('git', [ + '-C', + SUBMODULE_DIR, + 'fetch', + '--no-tags', + 'origin', + '+refs/heads/*:refs/remotes/origin/*', + '+refs/tags/*:refs/tags/*', + ]); + } + if (!hasLockedCommit()) { + fail(`locked Community commit ${lock.commit} is unavailable in ${lock.submodulePath}`); } +} - const commit = capture('git', ['rev-parse', 'HEAD'], SUBMODULE_DIR); - if (commit !== lock.commit) { - fail(`submodule HEAD is ${commit}; expected ${lock.commit}`); +function verifySource() { + if (!existsSync(join(SUBMODULE_DIR, '.git'))) { + fail(`submodule is unavailable; run git submodule update --init --recursive ${lock.submodulePath}`); } + ensureLockedCommit(); const tree = capture('git', ['rev-parse', `${lock.commit}:${lock.sourcePath}`], SUBMODULE_DIR); if (tree !== lock.tree) { @@ -120,10 +141,11 @@ function verifySource() { fail(`submodule worktree is not clean:\n${status}`); } - return { commit, tree }; + return { commit: lock.commit, tree }; } -function exportSource(destination) { +function exportSource(destination, sourcePath = lock.sourcePath) { + ensureLockedCommit(); replaceDirectory(destination); mkdirSync(VERSION_ROOT, { recursive: true }); const archiveName = `source-${process.pid}.tar`; @@ -139,7 +161,7 @@ function exportSource(destination) { '--format=tar', `--output=${archiveOutput}`, lock.commit, - lock.sourcePath, + sourcePath, ], { cwd: VERSION_ROOT }, ); @@ -150,6 +172,12 @@ function exportSource(destination) { rmSync(archivePath, { force: true }); } +// Upstream's community boundary check scans the sibling server tree, so every +// exported worktree needs the monorepo layout next to it. +function exportCommunityServer() { + exportSource(join(VERSION_ROOT, 'work', 'chat2db-community-server'), 'chat2db-community-server'); +} + function runYarn(args, cwd, env = {}) { if (!existsSync(YARN_JS)) { fail('pinned Yarn is missing; run npm ci in apps/frontend first'); @@ -179,6 +207,7 @@ function ensureDependencies() { function createWorktree(name) { ensureDependencies(); + exportCommunityServer(); const worktree = join(VERSION_ROOT, 'work', name); exportSource(worktree); symlinkSync( @@ -199,17 +228,12 @@ function setupUmi(worktree) { function test() { const worktree = createWorktree('test'); setupUmi(worktree); - runYarn(['test:chat-answer-update'], worktree); + // Upstream deleted the other per-behavior scripts this step used to pin. + // `prebuild:web:community` now runs the full upstream test battery before + // every community build, so only the surviving checks stay pinned here. runYarn(['test:tree-title-highlight'], worktree); - runYarn(['test:deep-clone'], worktree); runYarn(['test:ai-model-select'], worktree); runYarn(['test:export-connections'], worktree); - runYarn(['test:host-file-transfer'], worktree); - runYarn(['test:canvas-lifecycle'], worktree); - runYarn(['test:result-error-boundary'], worktree); - runYarn(['test:result-resource-activity'], worktree); - runYarn(['test:result-table-lifecycle'], worktree); - runYarn(['test:workspace-resource-activity'], worktree); } function build() {