// SPDX-License-Identifier: AGPL-3.0-or-later

//! 教育行业初始化频道的原生文档与表格数据层。
//!
//! 公共主世界只投影模块状态；真实内容必须进入已登录账号的独立存储根。
//! WebView 不直接持有文件路径或 SQLite 连接，所有写入都要求当前修订号，防止静默覆盖。

use ring::digest::{digest, SHA256};
use rusqlite::{params, types::Type, Connection, OptionalExtension};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::fs;
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tauri::AppHandle;
use uuid::Uuid;

const WORKSPACE_SCHEMA: &str = "hololake.education-workspace/v1";
const MAX_DOCUMENT_BYTES: usize = 2_000_000;
const MAX_TITLE_BYTES: usize = 300;
const MAX_TABLE_COLUMNS: usize = 30;
const MAX_TABLE_ROWS: usize = 1_000;
const MAX_CELL_BYTES: usize = 10_000;
const MODULE_NUMBER: &str = "HLP-MOD-OFFICIAL-EDUCATION-WORKBENCH-0001";
const ADAPTER: &str = "education-workbench-v1";

fn require_active(app: &AppHandle) -> Result<(), String> {
    crate::module_package_runtime::require_active_module_adapter(app, MODULE_NUMBER, ADAPTER)
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EducationDocumentSummary {
    pub document_id: String,
    pub title: String,
    pub revision: i64,
    pub updated_at_unix_ms: i64,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EducationDocument {
    pub document_id: String,
    pub title: String,
    pub body: String,
    pub revision: i64,
    pub created_at_unix_ms: i64,
    pub updated_at_unix_ms: i64,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EducationTableSummary {
    pub table_id: String,
    pub title: String,
    pub revision: i64,
    pub column_count: usize,
    pub row_count: usize,
    pub updated_at_unix_ms: i64,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct UnassignedTableSummary {
    pub table_id: String,
    pub title: String,
    pub column_count: usize,
    pub row_count: usize,
    pub imported_at_unix_ms: i64,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct EducationTableColumn {
    pub column_id: String,
    pub title: String,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct EducationTableRow {
    pub row_id: String,
    pub cells: Vec<String>,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EducationTable {
    pub table_id: String,
    pub title: String,
    pub columns: Vec<EducationTableColumn>,
    pub rows: Vec<EducationTableRow>,
    pub revision: i64,
    pub created_at_unix_ms: i64,
    pub updated_at_unix_ms: i64,
}

#[derive(Clone, Debug)]
pub(crate) struct ImportedEducationTable {
    pub title: String,
    pub columns: Vec<EducationTableColumn>,
    pub rows: Vec<EducationTableRow>,
}

#[derive(Clone, Debug)]
pub(crate) struct EducationImportRegistration {
    pub source_filename: String,
    pub source_format: String,
    pub source_sha256: String,
    pub profile_json: String,
    pub imported_at_unix_ms: i64,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EducationImportRegistrySummary {
    pub import_id: String,
    pub source_filename: String,
    pub source_format: String,
    pub table_count: usize,
    pub imported_at_unix_ms: i64,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EducationWorkspaceSnapshot {
    pub schema: &'static str,
    pub state: &'static str,
    pub documents: Vec<EducationDocumentSummary>,
    pub tables: Vec<EducationTableSummary>,
    pub automation_rules: Vec<EducationAutomationRule>,
    pub recent_automation_runs: Vec<EducationAutomationRunReceipt>,
    pub recent_imports: Vec<EducationImportRegistrySummary>,
    pub unassigned_tables: Vec<UnassignedTableSummary>,
    pub document_count: usize,
    pub table_count: usize,
    pub automation_rule_count: usize,
    pub unassigned_count: usize,
    pub storage: &'static str,
    pub authority: &'static str,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EducationAutomationRule {
    pub rule_id: String,
    pub title: String,
    pub table_id: String,
    pub condition_column_id: String,
    pub operator: String,
    pub condition_value: String,
    pub action_column_id: String,
    pub action_value: String,
    pub enabled: bool,
    pub revision: i64,
    pub created_at_unix_ms: i64,
    pub updated_at_unix_ms: i64,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EducationAutomationPreview {
    pub state: &'static str,
    pub rule_id: String,
    pub table_id: String,
    pub rule_revision: i64,
    pub table_revision: i64,
    pub matched_rows: usize,
    pub changed_cells: usize,
    pub preview_token: String,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EducationAutomationRunReceipt {
    pub state: String,
    pub run_id: String,
    pub rule_id: String,
    pub table_id: String,
    pub matched_rows: usize,
    pub changed_cells: usize,
    pub from_table_revision: i64,
    pub to_table_revision: i64,
    pub ran_at_unix_ms: i64,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct CreateEducationItemInput {
    pub title: String,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct ReadEducationDocumentInput {
    pub document_id: String,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SaveEducationDocumentInput {
    pub document_id: String,
    pub title: String,
    pub body: String,
    pub expected_revision: i64,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct ReadEducationTableInput {
    pub table_id: String,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SaveEducationTableInput {
    pub table_id: String,
    pub title: String,
    pub columns: Vec<EducationTableColumn>,
    pub rows: Vec<EducationTableRow>,
    pub expected_revision: i64,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct CreateEducationAutomationRuleInput {
    pub title: String,
    pub table_id: String,
    pub condition_column_id: String,
    pub operator: String,
    pub condition_value: String,
    pub action_column_id: String,
    pub action_value: String,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SaveEducationAutomationRuleInput {
    pub rule_id: String,
    pub title: String,
    pub table_id: String,
    pub condition_column_id: String,
    pub operator: String,
    pub condition_value: String,
    pub action_column_id: String,
    pub action_value: String,
    pub enabled: bool,
    pub expected_revision: i64,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct EducationAutomationRuleInput {
    pub rule_id: String,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct PreviewEducationAutomationRuleInput {
    pub rule_id: String,
    pub expected_rule_revision: i64,
    pub expected_table_revision: i64,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct ExecuteEducationAutomationRuleInput {
    pub rule_id: String,
    pub expected_rule_revision: i64,
    pub expected_table_revision: i64,
    pub preview_token: String,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct EducationArchiveReceipt {
    pub state: &'static str,
    pub entity_id: String,
}

pub async fn get_education_workspace_snapshot(
    app: AppHandle,
) -> Result<EducationWorkspaceSnapshot, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || snapshot_at(&database))
        .await
        .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn create_education_document(
    app: AppHandle,
    input: CreateEducationItemInput,
) -> Result<EducationDocument, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || create_document_at(&database, input))
        .await
        .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn read_education_document(
    app: AppHandle,
    input: ReadEducationDocumentInput,
) -> Result<EducationDocument, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || read_document_at(&database, &input.document_id))
        .await
        .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn save_education_document(
    app: AppHandle,
    input: SaveEducationDocumentInput,
) -> Result<EducationDocument, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || save_document_at(&database, input))
        .await
        .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn archive_education_document(
    app: AppHandle,
    input: ReadEducationDocumentInput,
) -> Result<EducationArchiveReceipt, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || {
        archive_at(
            &database,
            "education_documents",
            "document_id",
            input.document_id,
        )
    })
    .await
    .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn create_education_table(
    app: AppHandle,
    input: CreateEducationItemInput,
) -> Result<EducationTable, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || create_table_at(&database, input))
        .await
        .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn read_education_table(
    app: AppHandle,
    input: ReadEducationTableInput,
) -> Result<EducationTable, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || read_table_at(&database, &input.table_id))
        .await
        .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn save_education_table(
    app: AppHandle,
    input: SaveEducationTableInput,
) -> Result<EducationTable, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || save_table_at(&database, input))
        .await
        .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn archive_education_table(
    app: AppHandle,
    input: ReadEducationTableInput,
) -> Result<EducationArchiveReceipt, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || {
        archive_at(&database, "education_tables", "table_id", input.table_id)
    })
    .await
    .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn assign_imported_table_to_education(
    app: AppHandle,
    input: ReadEducationTableInput,
) -> Result<EducationTable, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || {
        assign_imported_table_to_education_at(&database, &input.table_id)
    })
    .await
    .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn create_education_automation_rule(
    app: AppHandle,
    input: CreateEducationAutomationRuleInput,
) -> Result<EducationAutomationRule, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || create_automation_rule_at(&database, input))
        .await
        .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn save_education_automation_rule(
    app: AppHandle,
    input: SaveEducationAutomationRuleInput,
) -> Result<EducationAutomationRule, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || save_automation_rule_at(&database, input))
        .await
        .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn archive_education_automation_rule(
    app: AppHandle,
    input: EducationAutomationRuleInput,
) -> Result<EducationArchiveReceipt, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || {
        archive_automation_rule_at(&database, input.rule_id)
    })
    .await
    .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn preview_education_automation_rule(
    app: AppHandle,
    input: PreviewEducationAutomationRuleInput,
) -> Result<EducationAutomationPreview, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || preview_automation_rule_at(&database, input))
        .await
        .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub async fn execute_education_automation_rule(
    app: AppHandle,
    input: ExecuteEducationAutomationRuleInput,
) -> Result<EducationAutomationRunReceipt, String> {
    require_active(&app)?;
    let database = education_workspace_database(&app)?;
    tauri::async_runtime::spawn_blocking(move || execute_automation_rule_at(&database, input))
        .await
        .map_err(|error| format!("HOLOLAKE_EDUCATION_JOIN_FAILED: {error}"))?
}

pub(crate) fn education_workspace_database(app: &AppHandle) -> Result<PathBuf, String> {
    let root = crate::authenticated_storage::account_storage_root(app, "education-workspace-v1")?;
    create_private_directory(&root)?;
    Ok(root.join("education-workspace.sqlite3"))
}

fn create_private_directory(path: &Path) -> Result<(), String> {
    fs::create_dir_all(path)
        .map_err(|error| format!("HOLOLAKE_EDUCATION_STORAGE_UNAVAILABLE: {error}"))?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(path, fs::Permissions::from_mode(0o700))
            .map_err(|error| format!("HOLOLAKE_EDUCATION_STORAGE_PERMISSION_FAILED: {error}"))?;
    }
    Ok(())
}

fn open_database(path: &Path) -> Result<Connection, String> {
    if let Some(parent) = path.parent() {
        create_private_directory(parent)?;
    }
    let connection = Connection::open(path)
        .map_err(|error| format!("HOLOLAKE_EDUCATION_DATABASE_UNAVAILABLE: {error}"))?;
    connection
        .busy_timeout(Duration::from_secs(5))
        .map_err(|error| format!("HOLOLAKE_EDUCATION_DATABASE_UNAVAILABLE: {error}"))?;
    connection
        .execute_batch(
            "PRAGMA foreign_keys = ON;
             PRAGMA journal_mode = DELETE;
             PRAGMA synchronous = FULL;
             PRAGMA trusted_schema = OFF;
             CREATE TABLE IF NOT EXISTS workspace_meta (
               key TEXT PRIMARY KEY NOT NULL,
               value TEXT NOT NULL
             );
             INSERT OR IGNORE INTO workspace_meta(key, value) VALUES ('schema_version', '1');
             CREATE TABLE IF NOT EXISTS education_documents (
               document_id TEXT PRIMARY KEY NOT NULL,
               title TEXT NOT NULL,
               body TEXT NOT NULL,
               revision INTEGER NOT NULL,
               created_at_unix_ms INTEGER NOT NULL,
               updated_at_unix_ms INTEGER NOT NULL,
               archived INTEGER NOT NULL DEFAULT 0 CHECK(archived IN (0, 1))
             );
             CREATE TABLE IF NOT EXISTS education_tables (
               table_id TEXT PRIMARY KEY NOT NULL,
               title TEXT NOT NULL,
               columns_json TEXT NOT NULL,
               rows_json TEXT NOT NULL,
               revision INTEGER NOT NULL,
               created_at_unix_ms INTEGER NOT NULL,
               updated_at_unix_ms INTEGER NOT NULL,
               assignment_scope TEXT NOT NULL DEFAULT 'EDUCATION' CHECK(assignment_scope IN ('EDUCATION', 'UNASSIGNED')),
               archived INTEGER NOT NULL DEFAULT 0 CHECK(archived IN (0, 1))
             );
             CREATE TABLE IF NOT EXISTS education_automation_rules (
               rule_id TEXT PRIMARY KEY NOT NULL,
               title TEXT NOT NULL,
               table_id TEXT NOT NULL,
               condition_column_id TEXT NOT NULL,
               operator TEXT NOT NULL,
               condition_value TEXT NOT NULL,
               action_column_id TEXT NOT NULL,
               action_value TEXT NOT NULL,
               enabled INTEGER NOT NULL DEFAULT 1 CHECK(enabled IN (0, 1)),
               revision INTEGER NOT NULL,
               created_at_unix_ms INTEGER NOT NULL,
               updated_at_unix_ms INTEGER NOT NULL,
               archived INTEGER NOT NULL DEFAULT 0 CHECK(archived IN (0, 1))
             );
             CREATE TABLE IF NOT EXISTS education_automation_runs (
               run_id TEXT PRIMARY KEY NOT NULL,
               rule_id TEXT NOT NULL,
               table_id TEXT NOT NULL,
               matched_rows INTEGER NOT NULL,
               changed_cells INTEGER NOT NULL,
               from_table_revision INTEGER NOT NULL,
               to_table_revision INTEGER NOT NULL,
               ran_at_unix_ms INTEGER NOT NULL
             );
             CREATE TABLE IF NOT EXISTS education_import_registry (
               import_id TEXT PRIMARY KEY NOT NULL,
               source_filename TEXT NOT NULL,
               source_format TEXT NOT NULL,
               source_sha256 TEXT NOT NULL,
               profile_json TEXT NOT NULL,
               table_ids_json TEXT NOT NULL,
               imported_at_unix_ms INTEGER NOT NULL
             );",
        )
        .map_err(|error| format!("HOLOLAKE_EDUCATION_DATABASE_MIGRATION_FAILED: {error}"))?;
    if !table_has_column(&connection, "education_tables", "assignment_scope")? {
        connection
            .execute(
                "ALTER TABLE education_tables ADD COLUMN assignment_scope TEXT NOT NULL DEFAULT 'EDUCATION' CHECK(assignment_scope IN ('EDUCATION', 'UNASSIGNED'))",
                [],
            )
            .map_err(database_write_error)?;
        let imported_table_ids = {
            let mut statement = connection
                .prepare("SELECT table_ids_json FROM education_import_registry")
                .map_err(database_read_error)?;
            let rows = statement
                .query_map([], |row| row.get::<_, String>(0))
                .map_err(database_read_error)?;
            let encoded = rows
                .collect::<Result<Vec<_>, _>>()
                .map_err(database_read_error)?;
            encoded
                .into_iter()
                .map(|value| decode_sqlite_json::<Vec<String>>(&value, 0))
                .collect::<Result<Vec<_>, _>>()
                .map_err(database_read_error)?
                .into_iter()
                .flatten()
                .collect::<Vec<_>>()
        };
        for table_id in imported_table_ids {
            connection
                .execute(
                    "UPDATE education_tables SET assignment_scope = 'UNASSIGNED' WHERE table_id = ?1 AND archived = 0",
                    [table_id],
                )
                .map_err(database_write_error)?;
        }
        connection
            .execute(
                "INSERT INTO workspace_meta(key, value) VALUES ('schema_version', '2') ON CONFLICT(key) DO UPDATE SET value = excluded.value",
                [],
            )
            .map_err(database_write_error)?;
    }
    #[cfg(unix)]
    if path.exists() {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(path, fs::Permissions::from_mode(0o600))
            .map_err(|error| format!("HOLOLAKE_EDUCATION_DATABASE_PERMISSION_FAILED: {error}"))?;
    }
    Ok(connection)
}

fn snapshot_at(path: &Path) -> Result<EducationWorkspaceSnapshot, String> {
    let connection = open_database(path)?;
    let documents = {
        let mut statement = connection
            .prepare(
                "SELECT document_id, title, revision, updated_at_unix_ms
                 FROM education_documents WHERE archived = 0
                 ORDER BY updated_at_unix_ms DESC, title ASC",
            )
            .map_err(database_read_error)?;
        let rows = statement
            .query_map([], |row| {
                Ok(EducationDocumentSummary {
                    document_id: row.get(0)?,
                    title: row.get(1)?,
                    revision: row.get(2)?,
                    updated_at_unix_ms: row.get(3)?,
                })
            })
            .map_err(database_read_error)?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(database_read_error)?
    };
    let tables = {
        let mut statement = connection
            .prepare(
                "SELECT table_id, title, columns_json, rows_json, revision, updated_at_unix_ms
                 FROM education_tables WHERE archived = 0 AND assignment_scope = 'EDUCATION'
                 ORDER BY updated_at_unix_ms DESC, title ASC",
            )
            .map_err(database_read_error)?;
        let rows = statement
            .query_map([], |row| {
                let columns_json: String = row.get(2)?;
                let rows_json: String = row.get(3)?;
                let columns: Vec<EducationTableColumn> = decode_sqlite_json(&columns_json, 2)?;
                let rows: Vec<EducationTableRow> = decode_sqlite_json(&rows_json, 3)?;
                Ok(EducationTableSummary {
                    table_id: row.get(0)?,
                    title: row.get(1)?,
                    column_count: columns.len(),
                    row_count: rows.len(),
                    revision: row.get(4)?,
                    updated_at_unix_ms: row.get(5)?,
                })
            })
            .map_err(database_read_error)?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(database_read_error)?
    };
    let automation_rules = read_automation_rules_with_connection(&connection)?;
    let recent_automation_runs = read_recent_automation_runs_with_connection(&connection)?;
    let recent_imports = read_recent_imports_with_connection(&connection)?;
    let unassigned_tables = read_unassigned_tables_with_connection(&connection)?;
    Ok(EducationWorkspaceSnapshot {
        schema: WORKSPACE_SCHEMA,
        state: "READY",
        document_count: documents.len(),
        table_count: tables.len(),
        automation_rule_count: automation_rules.len(),
        documents,
        tables,
        automation_rules,
        recent_automation_runs,
        recent_imports,
        unassigned_count: unassigned_tables.len(),
        unassigned_tables,
        storage: "AUTHENTICATED_ACCOUNT_SCOPED_SQLITE",
        authority: "CURRENT_AUTHENTICATED_ACCOUNT_ONLY",
    })
}

fn read_unassigned_tables_with_connection(
    connection: &Connection,
) -> Result<Vec<UnassignedTableSummary>, String> {
    let mut statement = connection
        .prepare(
            "SELECT table_id, title, columns_json, rows_json, updated_at_unix_ms
             FROM education_tables
             WHERE archived = 0 AND assignment_scope = 'UNASSIGNED'
             ORDER BY updated_at_unix_ms DESC, title ASC",
        )
        .map_err(database_read_error)?;
    let rows = statement
        .query_map([], |row| {
            let columns_json: String = row.get(2)?;
            let rows_json: String = row.get(3)?;
            let columns: Vec<EducationTableColumn> = decode_sqlite_json(&columns_json, 2)?;
            let rows: Vec<EducationTableRow> = decode_sqlite_json(&rows_json, 3)?;
            Ok(UnassignedTableSummary {
                table_id: row.get(0)?,
                title: row.get(1)?,
                column_count: columns.len(),
                row_count: rows.len(),
                imported_at_unix_ms: row.get(4)?,
            })
        })
        .map_err(database_read_error)?;
    rows.collect::<Result<Vec<_>, _>>()
        .map_err(database_read_error)
}

fn read_recent_imports_with_connection(
    connection: &Connection,
) -> Result<Vec<EducationImportRegistrySummary>, String> {
    let mut statement = connection
        .prepare(
            "SELECT import_id, source_filename, source_format, table_ids_json, imported_at_unix_ms
             FROM education_import_registry ORDER BY imported_at_unix_ms DESC LIMIT 20",
        )
        .map_err(database_read_error)?;
    let rows = statement
        .query_map([], |row| {
            let table_ids_json: String = row.get(3)?;
            let table_ids: Vec<String> = decode_sqlite_json(&table_ids_json, 3)?;
            Ok(EducationImportRegistrySummary {
                import_id: row.get(0)?,
                source_filename: row.get(1)?,
                source_format: row.get(2)?,
                table_count: table_ids.len(),
                imported_at_unix_ms: row.get(4)?,
            })
        })
        .map_err(database_read_error)?;
    rows.collect::<Result<Vec<_>, _>>()
        .map_err(database_read_error)
}

fn read_automation_rules_with_connection(
    connection: &Connection,
) -> Result<Vec<EducationAutomationRule>, String> {
    let mut statement = connection
        .prepare(
            "SELECT rule_id, title, table_id, condition_column_id, operator,
                    condition_value, action_column_id, action_value, enabled, revision,
                    created_at_unix_ms, updated_at_unix_ms
             FROM education_automation_rules WHERE archived = 0
             ORDER BY updated_at_unix_ms DESC, title ASC",
        )
        .map_err(database_read_error)?;
    let rows = statement
        .query_map([], |row| {
            Ok(EducationAutomationRule {
                rule_id: row.get(0)?,
                title: row.get(1)?,
                table_id: row.get(2)?,
                condition_column_id: row.get(3)?,
                operator: row.get(4)?,
                condition_value: row.get(5)?,
                action_column_id: row.get(6)?,
                action_value: row.get(7)?,
                enabled: row.get::<_, i64>(8)? == 1,
                revision: row.get(9)?,
                created_at_unix_ms: row.get(10)?,
                updated_at_unix_ms: row.get(11)?,
            })
        })
        .map_err(database_read_error)?;
    rows.collect::<Result<Vec<_>, _>>()
        .map_err(database_read_error)
}

fn read_recent_automation_runs_with_connection(
    connection: &Connection,
) -> Result<Vec<EducationAutomationRunReceipt>, String> {
    let mut statement = connection
        .prepare(
            "SELECT run_id, rule_id, table_id, matched_rows, changed_cells,
                    from_table_revision, to_table_revision, ran_at_unix_ms
             FROM education_automation_runs ORDER BY ran_at_unix_ms DESC LIMIT 20",
        )
        .map_err(database_read_error)?;
    let rows = statement
        .query_map([], |row| {
            let changed_cells = row.get::<_, i64>(4)? as usize;
            Ok(EducationAutomationRunReceipt {
                state: if changed_cells > 0 {
                    "APPLIED"
                } else {
                    "UNCHANGED"
                }
                .into(),
                run_id: row.get(0)?,
                rule_id: row.get(1)?,
                table_id: row.get(2)?,
                matched_rows: row.get::<_, i64>(3)? as usize,
                changed_cells,
                from_table_revision: row.get(5)?,
                to_table_revision: row.get(6)?,
                ran_at_unix_ms: row.get(7)?,
            })
        })
        .map_err(database_read_error)?;
    rows.collect::<Result<Vec<_>, _>>()
        .map_err(database_read_error)
}

fn create_document_at(
    path: &Path,
    input: CreateEducationItemInput,
) -> Result<EducationDocument, String> {
    let title = normalized_title(&input.title, "未命名教育文档")?;
    let document_id = format!("EDU-DOC-{}", Uuid::new_v4());
    let now = now_ms();
    let connection = open_database(path)?;
    connection
        .execute(
            "INSERT INTO education_documents(
               document_id, title, body, revision, created_at_unix_ms, updated_at_unix_ms, archived
             ) VALUES (?1, ?2, '', 1, ?3, ?3, 0)",
            params![document_id, title, now],
        )
        .map_err(database_write_error)?;
    read_document_with_connection(&connection, &document_id)
}

fn read_document_at(path: &Path, document_id: &str) -> Result<EducationDocument, String> {
    let connection = open_database(path)?;
    read_document_with_connection(&connection, document_id)
}

fn read_document_with_connection(
    connection: &Connection,
    document_id: &str,
) -> Result<EducationDocument, String> {
    validate_entity_id(document_id, "EDU-DOC-")?;
    connection
        .query_row(
            "SELECT document_id, title, body, revision, created_at_unix_ms, updated_at_unix_ms
             FROM education_documents WHERE document_id = ?1 AND archived = 0",
            [document_id],
            |row| {
                Ok(EducationDocument {
                    document_id: row.get(0)?,
                    title: row.get(1)?,
                    body: row.get(2)?,
                    revision: row.get(3)?,
                    created_at_unix_ms: row.get(4)?,
                    updated_at_unix_ms: row.get(5)?,
                })
            },
        )
        .optional()
        .map_err(database_read_error)?
        .ok_or_else(|| "HOLOLAKE_EDUCATION_DOCUMENT_NOT_FOUND".to_string())
}

fn save_document_at(
    path: &Path,
    input: SaveEducationDocumentInput,
) -> Result<EducationDocument, String> {
    validate_entity_id(&input.document_id, "EDU-DOC-")?;
    let title = normalized_title(&input.title, "未命名教育文档")?;
    if input.body.len() > MAX_DOCUMENT_BYTES || input.expected_revision < 1 {
        return Err("HOLOLAKE_EDUCATION_DOCUMENT_INPUT_INVALID".into());
    }
    let connection = open_database(path)?;
    let changed = connection
        .execute(
            "UPDATE education_documents
             SET title = ?1, body = ?2, revision = revision + 1, updated_at_unix_ms = ?3
             WHERE document_id = ?4 AND revision = ?5 AND archived = 0",
            params![
                title,
                input.body,
                now_ms(),
                input.document_id,
                input.expected_revision
            ],
        )
        .map_err(database_write_error)?;
    if changed != 1 {
        return Err("HOLOLAKE_EDUCATION_DOCUMENT_REVISION_CONFLICT".into());
    }
    read_document_with_connection(&connection, &input.document_id)
}

fn create_table_at(path: &Path, input: CreateEducationItemInput) -> Result<EducationTable, String> {
    let title = normalized_title(&input.title, "未命名教育表格")?;
    let table_id = format!("EDU-TABLE-{}", Uuid::new_v4());
    let columns = vec![
        EducationTableColumn {
            column_id: format!("COL-{}", Uuid::new_v4()),
            title: "姓名".into(),
        },
        EducationTableColumn {
            column_id: format!("COL-{}", Uuid::new_v4()),
            title: "类别".into(),
        },
        EducationTableColumn {
            column_id: format!("COL-{}", Uuid::new_v4()),
            title: "状态".into(),
        },
    ];
    let rows: Vec<EducationTableRow> = Vec::new();
    let now = now_ms();
    let connection = open_database(path)?;
    connection
        .execute(
            "INSERT INTO education_tables(
               table_id, title, columns_json, rows_json, revision,
               created_at_unix_ms, updated_at_unix_ms, assignment_scope, archived
             ) VALUES (?1, ?2, ?3, ?4, 1, ?5, ?5, 'EDUCATION', 0)",
            params![
                table_id,
                title,
                serde_json::to_string(&columns).map_err(json_write_error)?,
                serde_json::to_string(&rows).map_err(json_write_error)?,
                now
            ],
        )
        .map_err(database_write_error)?;
    read_table_with_connection(&connection, &table_id)
}

pub(crate) fn read_table_at(path: &Path, table_id: &str) -> Result<EducationTable, String> {
    let connection = open_database(path)?;
    read_table_with_connection(&connection, table_id)
}

pub(crate) fn import_table_batch_at(
    path: &Path,
    imported: Vec<ImportedEducationTable>,
    registration: EducationImportRegistration,
) -> Result<(String, Vec<EducationTable>), String> {
    if imported.is_empty() {
        return Err("HOLOLAKE_EDUCATION_IMPORT_EMPTY".into());
    }
    for table in &imported {
        normalized_title(&table.title, "导入的教育表格")?;
        validate_table_data(&table.columns, &table.rows)?;
    }
    let mut connection = open_database(path)?;
    let transaction = connection.transaction().map_err(database_write_error)?;
    let now = now_ms();
    let mut table_ids = Vec::with_capacity(imported.len());
    for table in imported {
        let table_id = format!("EDU-TABLE-{}", Uuid::new_v4());
        let title = normalized_title(&table.title, "导入的教育表格")?;
        transaction
            .execute(
                "INSERT INTO education_tables(
                   table_id, title, columns_json, rows_json, revision,
                   created_at_unix_ms, updated_at_unix_ms, assignment_scope, archived
                 ) VALUES (?1, ?2, ?3, ?4, 1, ?5, ?5, 'UNASSIGNED', 0)",
                params![
                    table_id,
                    title,
                    serde_json::to_string(&table.columns).map_err(json_write_error)?,
                    serde_json::to_string(&table.rows).map_err(json_write_error)?,
                    now
                ],
            )
            .map_err(database_write_error)?;
        table_ids.push(table_id);
    }
    let import_id = format!("EDU-IMPORT-{}", Uuid::new_v4());
    transaction
        .execute(
            "INSERT INTO education_import_registry(
               import_id, source_filename, source_format, source_sha256,
               profile_json, table_ids_json, imported_at_unix_ms
             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
            params![
                import_id,
                registration.source_filename,
                registration.source_format,
                registration.source_sha256,
                registration.profile_json,
                serde_json::to_string(&table_ids).map_err(json_write_error)?,
                registration.imported_at_unix_ms
            ],
        )
        .map_err(database_write_error)?;
    transaction.commit().map_err(database_write_error)?;
    let tables = table_ids
        .iter()
        .map(|table_id| read_table_with_connection(&connection, table_id))
        .collect::<Result<Vec<_>, _>>()?;
    Ok((import_id, tables))
}

fn assign_imported_table_to_education_at(
    path: &Path,
    table_id: &str,
) -> Result<EducationTable, String> {
    validate_entity_id(table_id, "EDU-TABLE-")?;
    let connection = open_database(path)?;
    let changed = connection
        .execute(
            "UPDATE education_tables
             SET assignment_scope = 'EDUCATION', updated_at_unix_ms = ?1
             WHERE table_id = ?2 AND assignment_scope = 'UNASSIGNED' AND archived = 0",
            params![now_ms(), table_id],
        )
        .map_err(database_write_error)?;
    if changed != 1 {
        return Err("HOLOLAKE_EDUCATION_IMPORT_ASSIGNMENT_NOT_FOUND".into());
    }
    read_table_with_connection(&connection, table_id)
}

fn table_has_column(connection: &Connection, table: &str, column: &str) -> Result<bool, String> {
    let mut statement = connection
        .prepare(&format!("PRAGMA table_info({table})"))
        .map_err(database_read_error)?;
    let rows = statement
        .query_map([], |row| row.get::<_, String>(1))
        .map_err(database_read_error)?;
    let found = rows.filter_map(Result::ok).any(|name| name == column);
    Ok(found)
}

fn read_table_with_connection(
    connection: &Connection,
    table_id: &str,
) -> Result<EducationTable, String> {
    validate_entity_id(table_id, "EDU-TABLE-")?;
    let stored = connection
        .query_row(
            "SELECT table_id, title, columns_json, rows_json, revision,
                    created_at_unix_ms, updated_at_unix_ms
             FROM education_tables WHERE table_id = ?1 AND archived = 0",
            [table_id],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, i64>(4)?,
                    row.get::<_, i64>(5)?,
                    row.get::<_, i64>(6)?,
                ))
            },
        )
        .optional()
        .map_err(database_read_error)?
        .ok_or_else(|| "HOLOLAKE_EDUCATION_TABLE_NOT_FOUND".to_string())?;
    let columns = serde_json::from_str(&stored.2)
        .map_err(|error| format!("HOLOLAKE_EDUCATION_TABLE_DATA_INVALID: {error}"))?;
    let rows = serde_json::from_str(&stored.3)
        .map_err(|error| format!("HOLOLAKE_EDUCATION_TABLE_DATA_INVALID: {error}"))?;
    Ok(EducationTable {
        table_id: stored.0,
        title: stored.1,
        columns,
        rows,
        revision: stored.4,
        created_at_unix_ms: stored.5,
        updated_at_unix_ms: stored.6,
    })
}

fn save_table_at(path: &Path, input: SaveEducationTableInput) -> Result<EducationTable, String> {
    validate_entity_id(&input.table_id, "EDU-TABLE-")?;
    let title = normalized_title(&input.title, "未命名教育表格")?;
    validate_table_data(&input.columns, &input.rows)?;
    if input.expected_revision < 1 {
        return Err("HOLOLAKE_EDUCATION_TABLE_INPUT_INVALID".into());
    }
    let columns_json = serde_json::to_string(&input.columns).map_err(json_write_error)?;
    let rows_json = serde_json::to_string(&input.rows).map_err(json_write_error)?;
    let connection = open_database(path)?;
    let changed = connection
        .execute(
            "UPDATE education_tables
             SET title = ?1, columns_json = ?2, rows_json = ?3,
                 revision = revision + 1, updated_at_unix_ms = ?4
             WHERE table_id = ?5 AND revision = ?6 AND archived = 0",
            params![
                title,
                columns_json,
                rows_json,
                now_ms(),
                input.table_id,
                input.expected_revision
            ],
        )
        .map_err(database_write_error)?;
    if changed != 1 {
        return Err("HOLOLAKE_EDUCATION_TABLE_REVISION_CONFLICT".into());
    }
    read_table_with_connection(&connection, &input.table_id)
}

fn normalized_operator(value: &str) -> Result<String, String> {
    let operator = value.trim().to_ascii_uppercase();
    if matches!(
        operator.as_str(),
        "EQUALS" | "CONTAINS" | "IS_EMPTY" | "IS_NOT_EMPTY"
    ) {
        Ok(operator)
    } else {
        Err("HOLOLAKE_EDUCATION_AUTOMATION_OPERATOR_INVALID".into())
    }
}

fn validate_automation_rule_against_table(
    table: &EducationTable,
    condition_column_id: &str,
    action_column_id: &str,
    action_value: &str,
) -> Result<(), String> {
    let column_ids = table
        .columns
        .iter()
        .map(|column| column.column_id.as_str())
        .collect::<HashSet<_>>();
    if !column_ids.contains(condition_column_id)
        || !column_ids.contains(action_column_id)
        || action_value.len() > MAX_CELL_BYTES
    {
        return Err("HOLOLAKE_EDUCATION_AUTOMATION_COLUMN_INVALID".into());
    }
    Ok(())
}

fn create_automation_rule_at(
    path: &Path,
    input: CreateEducationAutomationRuleInput,
) -> Result<EducationAutomationRule, String> {
    validate_entity_id(&input.table_id, "EDU-TABLE-")?;
    let title = normalized_title(&input.title, "未命名自动化规则")?;
    let operator = normalized_operator(&input.operator)?;
    if input.condition_value.len() > MAX_CELL_BYTES {
        return Err("HOLOLAKE_EDUCATION_AUTOMATION_VALUE_INVALID".into());
    }
    let connection = open_database(path)?;
    let table = read_table_with_connection(&connection, &input.table_id)?;
    validate_automation_rule_against_table(
        &table,
        &input.condition_column_id,
        &input.action_column_id,
        &input.action_value,
    )?;
    let rule_id = format!("EDU-AUTO-{}", Uuid::new_v4());
    let now = now_ms();
    connection
        .execute(
            "INSERT INTO education_automation_rules(
               rule_id, title, table_id, condition_column_id, operator, condition_value,
               action_column_id, action_value, enabled, revision,
               created_at_unix_ms, updated_at_unix_ms, archived
             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 1, 1, ?9, ?9, 0)",
            params![
                rule_id,
                title,
                input.table_id,
                input.condition_column_id,
                operator,
                input.condition_value,
                input.action_column_id,
                input.action_value,
                now
            ],
        )
        .map_err(database_write_error)?;
    read_automation_rule_with_connection(&connection, &rule_id)
}

fn read_automation_rule_with_connection(
    connection: &Connection,
    rule_id: &str,
) -> Result<EducationAutomationRule, String> {
    validate_entity_id(rule_id, "EDU-AUTO-")?;
    connection
        .query_row(
            "SELECT rule_id, title, table_id, condition_column_id, operator,
                    condition_value, action_column_id, action_value, enabled, revision,
                    created_at_unix_ms, updated_at_unix_ms
             FROM education_automation_rules WHERE rule_id = ?1 AND archived = 0",
            [rule_id],
            |row| {
                Ok(EducationAutomationRule {
                    rule_id: row.get(0)?,
                    title: row.get(1)?,
                    table_id: row.get(2)?,
                    condition_column_id: row.get(3)?,
                    operator: row.get(4)?,
                    condition_value: row.get(5)?,
                    action_column_id: row.get(6)?,
                    action_value: row.get(7)?,
                    enabled: row.get::<_, i64>(8)? == 1,
                    revision: row.get(9)?,
                    created_at_unix_ms: row.get(10)?,
                    updated_at_unix_ms: row.get(11)?,
                })
            },
        )
        .optional()
        .map_err(database_read_error)?
        .ok_or_else(|| "HOLOLAKE_EDUCATION_AUTOMATION_RULE_NOT_FOUND".to_string())
}

fn save_automation_rule_at(
    path: &Path,
    input: SaveEducationAutomationRuleInput,
) -> Result<EducationAutomationRule, String> {
    validate_entity_id(&input.rule_id, "EDU-AUTO-")?;
    validate_entity_id(&input.table_id, "EDU-TABLE-")?;
    if input.expected_revision < 1 || input.condition_value.len() > MAX_CELL_BYTES {
        return Err("HOLOLAKE_EDUCATION_AUTOMATION_INPUT_INVALID".into());
    }
    let title = normalized_title(&input.title, "未命名自动化规则")?;
    let operator = normalized_operator(&input.operator)?;
    let connection = open_database(path)?;
    let table = read_table_with_connection(&connection, &input.table_id)?;
    validate_automation_rule_against_table(
        &table,
        &input.condition_column_id,
        &input.action_column_id,
        &input.action_value,
    )?;
    let changed = connection
        .execute(
            "UPDATE education_automation_rules
             SET title = ?1, table_id = ?2, condition_column_id = ?3, operator = ?4,
                 condition_value = ?5, action_column_id = ?6, action_value = ?7,
                 enabled = ?8, revision = revision + 1, updated_at_unix_ms = ?9
             WHERE rule_id = ?10 AND revision = ?11 AND archived = 0",
            params![
                title,
                input.table_id,
                input.condition_column_id,
                operator,
                input.condition_value,
                input.action_column_id,
                input.action_value,
                if input.enabled { 1 } else { 0 },
                now_ms(),
                input.rule_id,
                input.expected_revision
            ],
        )
        .map_err(database_write_error)?;
    if changed != 1 {
        return Err("HOLOLAKE_EDUCATION_AUTOMATION_RULE_REVISION_CONFLICT".into());
    }
    read_automation_rule_with_connection(&connection, &input.rule_id)
}

fn archive_automation_rule_at(
    path: &Path,
    rule_id: String,
) -> Result<EducationArchiveReceipt, String> {
    validate_entity_id(&rule_id, "EDU-AUTO-")?;
    let connection = open_database(path)?;
    let changed = connection
        .execute(
            "UPDATE education_automation_rules
             SET archived = 1, updated_at_unix_ms = ?1
             WHERE rule_id = ?2 AND archived = 0",
            params![now_ms(), rule_id],
        )
        .map_err(database_write_error)?;
    if changed != 1 {
        return Err("HOLOLAKE_EDUCATION_AUTOMATION_RULE_NOT_FOUND".into());
    }
    Ok(EducationArchiveReceipt {
        state: "ARCHIVED_RECOVERABLE",
        entity_id: rule_id,
    })
}

fn rule_matches(operator: &str, condition_value: &str, cell: &str) -> bool {
    match operator {
        "EQUALS" => cell == condition_value,
        "CONTAINS" => cell.contains(condition_value),
        "IS_EMPTY" => cell.trim().is_empty(),
        "IS_NOT_EMPTY" => !cell.trim().is_empty(),
        _ => false,
    }
}

fn automation_changes(
    rule: &EducationAutomationRule,
    table: &EducationTable,
) -> Result<(usize, usize, Vec<EducationTableRow>), String> {
    validate_automation_rule_against_table(
        table,
        &rule.condition_column_id,
        &rule.action_column_id,
        &rule.action_value,
    )?;
    let condition_index = table
        .columns
        .iter()
        .position(|column| column.column_id == rule.condition_column_id)
        .ok_or_else(|| "HOLOLAKE_EDUCATION_AUTOMATION_COLUMN_INVALID".to_string())?;
    let action_index = table
        .columns
        .iter()
        .position(|column| column.column_id == rule.action_column_id)
        .ok_or_else(|| "HOLOLAKE_EDUCATION_AUTOMATION_COLUMN_INVALID".to_string())?;
    let mut matched_rows = 0;
    let mut changed_cells = 0;
    let mut rows = table.rows.clone();
    for row in &mut rows {
        if rule_matches(
            &rule.operator,
            &rule.condition_value,
            &row.cells[condition_index],
        ) {
            matched_rows += 1;
            if row.cells[action_index] != rule.action_value {
                row.cells[action_index] = rule.action_value.clone();
                changed_cells += 1;
            }
        }
    }
    Ok((matched_rows, changed_cells, rows))
}

fn automation_preview_token(
    rule: &EducationAutomationRule,
    table: &EducationTable,
    matched_rows: usize,
    changed_cells: usize,
) -> String {
    let material = format!(
        "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}",
        rule.rule_id,
        rule.revision,
        table.table_id,
        table.revision,
        rule.condition_column_id,
        rule.operator,
        rule.condition_value,
        rule.action_column_id,
        rule.action_value,
        matched_rows,
        changed_cells
    );
    digest(&SHA256, material.as_bytes())
        .as_ref()
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect()
}

fn preview_automation_rule_with_connection(
    connection: &Connection,
    input: &PreviewEducationAutomationRuleInput,
) -> Result<
    (
        EducationAutomationPreview,
        EducationAutomationRule,
        EducationTable,
        Vec<EducationTableRow>,
    ),
    String,
> {
    let rule = read_automation_rule_with_connection(connection, &input.rule_id)?;
    if !rule.enabled {
        return Err("HOLOLAKE_EDUCATION_AUTOMATION_RULE_DISABLED".into());
    }
    let table = read_table_with_connection(connection, &rule.table_id)?;
    if rule.revision != input.expected_rule_revision
        || table.revision != input.expected_table_revision
    {
        return Err("HOLOLAKE_EDUCATION_AUTOMATION_PREVIEW_STALE".into());
    }
    let (matched_rows, changed_cells, rows) = automation_changes(&rule, &table)?;
    let preview_token = automation_preview_token(&rule, &table, matched_rows, changed_cells);
    let preview = EducationAutomationPreview {
        state: "PREVIEW_ONLY_NO_WRITE",
        rule_id: rule.rule_id.clone(),
        table_id: table.table_id.clone(),
        rule_revision: rule.revision,
        table_revision: table.revision,
        matched_rows,
        changed_cells,
        preview_token,
    };
    Ok((preview, rule, table, rows))
}

fn preview_automation_rule_at(
    path: &Path,
    input: PreviewEducationAutomationRuleInput,
) -> Result<EducationAutomationPreview, String> {
    let connection = open_database(path)?;
    preview_automation_rule_with_connection(&connection, &input).map(|result| result.0)
}

fn execute_automation_rule_at(
    path: &Path,
    input: ExecuteEducationAutomationRuleInput,
) -> Result<EducationAutomationRunReceipt, String> {
    if input.preview_token.len() != 64
        || !input
            .preview_token
            .chars()
            .all(|character| character.is_ascii_hexdigit())
    {
        return Err("HOLOLAKE_EDUCATION_AUTOMATION_PREVIEW_TOKEN_INVALID".into());
    }
    let mut connection = open_database(path)?;
    let transaction = connection
        .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
        .map_err(database_write_error)?;
    let preview_input = PreviewEducationAutomationRuleInput {
        rule_id: input.rule_id.clone(),
        expected_rule_revision: input.expected_rule_revision,
        expected_table_revision: input.expected_table_revision,
    };
    let (preview, _rule, table, rows) =
        preview_automation_rule_with_connection(&transaction, &preview_input)?;
    if preview.preview_token != input.preview_token {
        return Err("HOLOLAKE_EDUCATION_AUTOMATION_PREVIEW_TOKEN_MISMATCH".into());
    }
    let now = now_ms();
    let to_revision = if preview.changed_cells > 0 {
        table.revision + 1
    } else {
        table.revision
    };
    if preview.changed_cells > 0 {
        let rows_json = serde_json::to_string(&rows).map_err(json_write_error)?;
        let changed = transaction
            .execute(
                "UPDATE education_tables SET rows_json = ?1, revision = revision + 1,
                        updated_at_unix_ms = ?2
                 WHERE table_id = ?3 AND revision = ?4 AND archived = 0",
                params![rows_json, now, table.table_id, table.revision],
            )
            .map_err(database_write_error)?;
        if changed != 1 {
            return Err("HOLOLAKE_EDUCATION_AUTOMATION_TABLE_REVISION_CONFLICT".into());
        }
    }
    let run_id = format!("EDU-RUN-{}", Uuid::new_v4());
    transaction
        .execute(
            "INSERT INTO education_automation_runs(
               run_id, rule_id, table_id, matched_rows, changed_cells,
               from_table_revision, to_table_revision, ran_at_unix_ms
             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
            params![
                run_id,
                preview.rule_id,
                preview.table_id,
                preview.matched_rows as i64,
                preview.changed_cells as i64,
                table.revision,
                to_revision,
                now
            ],
        )
        .map_err(database_write_error)?;
    transaction.commit().map_err(database_write_error)?;
    Ok(EducationAutomationRunReceipt {
        state: if preview.changed_cells > 0 {
            "APPLIED"
        } else {
            "UNCHANGED"
        }
        .into(),
        run_id,
        rule_id: input.rule_id,
        table_id: table.table_id,
        matched_rows: preview.matched_rows,
        changed_cells: preview.changed_cells,
        from_table_revision: table.revision,
        to_table_revision: to_revision,
        ran_at_unix_ms: now,
    })
}

fn archive_at(
    path: &Path,
    table: &str,
    id_column: &str,
    entity_id: String,
) -> Result<EducationArchiveReceipt, String> {
    if !matches!(
        (table, id_column),
        ("education_documents", "document_id") | ("education_tables", "table_id")
    ) {
        return Err("HOLOLAKE_EDUCATION_ARCHIVE_TARGET_INVALID".into());
    }
    let expected_prefix = if table == "education_documents" {
        "EDU-DOC-"
    } else {
        "EDU-TABLE-"
    };
    validate_entity_id(&entity_id, expected_prefix)?;
    let connection = open_database(path)?;
    let sql = format!(
        "UPDATE {table} SET archived = 1, updated_at_unix_ms = ?1
         WHERE {id_column} = ?2 AND archived = 0"
    );
    let changed = connection
        .execute(&sql, params![now_ms(), entity_id])
        .map_err(database_write_error)?;
    if changed != 1 {
        return Err("HOLOLAKE_EDUCATION_ARCHIVE_NOT_FOUND".into());
    }
    Ok(EducationArchiveReceipt {
        state: "ARCHIVED_RECOVERABLE",
        entity_id,
    })
}

pub(crate) fn validate_table_data(
    columns: &[EducationTableColumn],
    rows: &[EducationTableRow],
) -> Result<(), String> {
    if columns.is_empty() || columns.len() > MAX_TABLE_COLUMNS || rows.len() > MAX_TABLE_ROWS {
        return Err("HOLOLAKE_EDUCATION_TABLE_BOUNDS_INVALID".into());
    }
    let mut column_ids = HashSet::new();
    for column in columns {
        if !column_ids.insert(column.column_id.as_str())
            || column.column_id.len() > 80
            || column.title.trim().is_empty()
            || column.title.len() > MAX_TITLE_BYTES
        {
            return Err("HOLOLAKE_EDUCATION_TABLE_COLUMN_INVALID".into());
        }
    }
    let mut row_ids = HashSet::new();
    for row in rows {
        if !row_ids.insert(row.row_id.as_str())
            || row.row_id.len() > 80
            || row.cells.len() != columns.len()
            || row.cells.iter().any(|cell| cell.len() > MAX_CELL_BYTES)
        {
            return Err("HOLOLAKE_EDUCATION_TABLE_ROW_INVALID".into());
        }
    }
    Ok(())
}

fn normalized_title(value: &str, fallback: &str) -> Result<String, String> {
    let title = if value.trim().is_empty() {
        fallback
    } else {
        value.trim()
    };
    if title.len() > MAX_TITLE_BYTES
        || title
            .chars()
            .any(|character| matches!(character, '\0' | '\r' | '\n'))
    {
        return Err("HOLOLAKE_EDUCATION_TITLE_INVALID".into());
    }
    Ok(title.to_string())
}

fn validate_entity_id(value: &str, prefix: &str) -> Result<(), String> {
    if value.starts_with(prefix)
        && value.len() <= 80
        && value
            .chars()
            .all(|character| character.is_ascii_alphanumeric() || character == '-')
    {
        Ok(())
    } else {
        Err("HOLOLAKE_EDUCATION_ENTITY_ID_INVALID".into())
    }
}

fn now_ms() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|duration| duration.as_millis().min(i64::MAX as u128) as i64)
        .unwrap_or(0)
}

fn decode_sqlite_json<T: DeserializeOwned>(value: &str, column: usize) -> rusqlite::Result<T> {
    serde_json::from_str(value).map_err(|error| {
        rusqlite::Error::FromSqlConversionFailure(column, Type::Text, Box::new(error))
    })
}

fn database_read_error(error: rusqlite::Error) -> String {
    format!("HOLOLAKE_EDUCATION_DATABASE_READ_FAILED: {error}")
}

fn database_write_error(error: rusqlite::Error) -> String {
    format!("HOLOLAKE_EDUCATION_DATABASE_WRITE_FAILED: {error}")
}

fn json_write_error(error: serde_json::Error) -> String {
    format!("HOLOLAKE_EDUCATION_TABLE_JSON_FAILED: {error}")
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    #[test]
    fn documents_persist_across_reopen_and_reject_stale_revisions() {
        let directory = tempdir().unwrap();
        let database = directory.path().join("education.sqlite3");
        let created = create_document_at(
            &database,
            CreateEducationItemInput {
                title: "师训计划".into(),
            },
        )
        .unwrap();
        let saved = save_document_at(
            &database,
            SaveEducationDocumentInput {
                document_id: created.document_id.clone(),
                title: created.title.clone(),
                body: "# 第一周\n\n完成课程准备。".into(),
                expected_revision: created.revision,
            },
        )
        .unwrap();
        assert_eq!(saved.revision, 2);
        assert_eq!(
            read_document_at(&database, &created.document_id)
                .unwrap()
                .body,
            "# 第一周\n\n完成课程准备。"
        );
        assert!(save_document_at(
            &database,
            SaveEducationDocumentInput {
                document_id: created.document_id,
                title: "错误覆盖".into(),
                body: "旧修订".into(),
                expected_revision: 1,
            },
        )
        .unwrap_err()
        .contains("REVISION_CONFLICT"));
    }

    #[test]
    fn tables_persist_cells_and_archive_without_hard_delete() {
        let directory = tempdir().unwrap();
        let database = directory.path().join("education.sqlite3");
        let created = create_table_at(
            &database,
            CreateEducationItemInput {
                title: "学员进度".into(),
            },
        )
        .unwrap();
        let row = EducationTableRow {
            row_id: format!("ROW-{}", Uuid::new_v4()),
            cells: vec!["张同学".into(), "学员".into(), "进行中".into()],
        };
        let saved = save_table_at(
            &database,
            SaveEducationTableInput {
                table_id: created.table_id.clone(),
                title: created.title,
                columns: created.columns,
                rows: vec![row],
                expected_revision: created.revision,
            },
        )
        .unwrap();
        assert_eq!(saved.rows[0].cells[0], "张同学");
        archive_at(
            &database,
            "education_tables",
            "table_id",
            created.table_id.clone(),
        )
        .unwrap();
        assert!(read_table_at(&database, &created.table_id).is_err());
        let connection = open_database(&database).unwrap();
        let archived: i64 = connection
            .query_row(
                "SELECT archived FROM education_tables WHERE table_id = ?1",
                [created.table_id],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(archived, 1);
    }

    #[test]
    fn imported_tables_stay_unassigned_until_a_human_routes_them_to_education() {
        let directory = tempdir().unwrap();
        let database = directory.path().join("education.sqlite3");
        let imported = ImportedEducationTable {
            title: "待识别业务数据".into(),
            columns: vec![EducationTableColumn {
                column_id: format!("COL-{}", Uuid::new_v4()),
                title: "业务字段".into(),
            }],
            rows: vec![EducationTableRow {
                row_id: format!("ROW-{}", Uuid::new_v4()),
                cells: vec!["尚未归属教育".into()],
            }],
        };
        let (_, tables) = import_table_batch_at(
            &database,
            vec![imported],
            EducationImportRegistration {
                source_filename: "业务数据.xlsx".into(),
                source_format: "XLSX".into(),
                source_sha256: "0".repeat(64),
                profile_json: "{}".into(),
                imported_at_unix_ms: now_ms(),
            },
        )
        .unwrap();
        let table_id = tables[0].table_id.clone();
        let quarantined = snapshot_at(&database).unwrap();
        assert_eq!(quarantined.table_count, 0);
        assert_eq!(quarantined.unassigned_count, 1);
        assert_eq!(quarantined.unassigned_tables[0].table_id, table_id);

        assign_imported_table_to_education_at(&database, &table_id).unwrap();
        let assigned = snapshot_at(&database).unwrap();
        assert_eq!(assigned.table_count, 1);
        assert_eq!(assigned.unassigned_count, 0);
    }

    #[test]
    fn table_bounds_are_enforced() {
        let columns = (0..=MAX_TABLE_COLUMNS)
            .map(|index| EducationTableColumn {
                column_id: format!("COL-{index}"),
                title: format!("列 {index}"),
            })
            .collect::<Vec<_>>();
        assert!(validate_table_data(&columns, &[]).is_err());
    }

    fn seeded_automation_table(database: &Path) -> EducationTable {
        let created = create_table_at(
            database,
            CreateEducationItemInput {
                title: "学员跟进".into(),
            },
        )
        .unwrap();
        save_table_at(
            database,
            SaveEducationTableInput {
                table_id: created.table_id.clone(),
                title: created.title,
                columns: created.columns,
                rows: vec![
                    EducationTableRow {
                        row_id: format!("ROW-{}", Uuid::new_v4()),
                        cells: vec!["林同学".into(), "学员".into(), "待跟进".into()],
                    },
                    EducationTableRow {
                        row_id: format!("ROW-{}", Uuid::new_v4()),
                        cells: vec!["周老师".into(), "教师".into(), "已完成".into()],
                    },
                ],
                expected_revision: created.revision,
            },
        )
        .unwrap()
    }

    #[test]
    fn automation_rules_persist_preview_execute_and_receipt() {
        let directory = tempdir().unwrap();
        let database = directory.path().join("education.sqlite3");
        let table = seeded_automation_table(&database);
        let rule = create_automation_rule_at(
            &database,
            CreateEducationAutomationRuleInput {
                title: "待跟进转为已联系".into(),
                table_id: table.table_id.clone(),
                condition_column_id: table.columns[2].column_id.clone(),
                operator: "EQUALS".into(),
                condition_value: "待跟进".into(),
                action_column_id: table.columns[2].column_id.clone(),
                action_value: "已联系".into(),
            },
        )
        .unwrap();
        let preview = preview_automation_rule_at(
            &database,
            PreviewEducationAutomationRuleInput {
                rule_id: rule.rule_id.clone(),
                expected_rule_revision: rule.revision,
                expected_table_revision: table.revision,
            },
        )
        .unwrap();
        assert_eq!(preview.matched_rows, 1);
        assert_eq!(preview.changed_cells, 1);
        assert_eq!(snapshot_at(&database).unwrap().automation_rule_count, 1);
        let receipt = execute_automation_rule_at(
            &database,
            ExecuteEducationAutomationRuleInput {
                rule_id: rule.rule_id,
                expected_rule_revision: rule.revision,
                expected_table_revision: table.revision,
                preview_token: preview.preview_token,
            },
        )
        .unwrap();
        assert_eq!(receipt.state, "APPLIED");
        assert_eq!(receipt.to_table_revision, table.revision + 1);
        let updated = read_table_at(&database, &table.table_id).unwrap();
        assert_eq!(updated.rows[0].cells[2], "已联系");
        assert_eq!(
            snapshot_at(&database).unwrap().recent_automation_runs.len(),
            1
        );
    }

    #[test]
    fn automation_execution_rejects_stale_or_changed_preview() {
        let directory = tempdir().unwrap();
        let database = directory.path().join("education.sqlite3");
        let table = seeded_automation_table(&database);
        let rule = create_automation_rule_at(
            &database,
            CreateEducationAutomationRuleInput {
                title: "教师标记".into(),
                table_id: table.table_id.clone(),
                condition_column_id: table.columns[1].column_id.clone(),
                operator: "EQUALS".into(),
                condition_value: "教师".into(),
                action_column_id: table.columns[2].column_id.clone(),
                action_value: "已确认".into(),
            },
        )
        .unwrap();
        let preview = preview_automation_rule_at(
            &database,
            PreviewEducationAutomationRuleInput {
                rule_id: rule.rule_id.clone(),
                expected_rule_revision: rule.revision,
                expected_table_revision: table.revision,
            },
        )
        .unwrap();
        let mut wrong_token = preview.preview_token;
        wrong_token.replace_range(0..1, if &wrong_token[0..1] == "0" { "1" } else { "0" });
        assert!(execute_automation_rule_at(
            &database,
            ExecuteEducationAutomationRuleInput {
                rule_id: rule.rule_id,
                expected_rule_revision: rule.revision,
                expected_table_revision: table.revision,
                preview_token: wrong_token,
            },
        )
        .unwrap_err()
        .contains("TOKEN_MISMATCH"));
    }
}