{
    id: node:⟨uuid⟩,                     // SurrealDB record ID (primary key, auto-indexed)
    type: "task",                        // Current active type
    data: task:⟨uuid⟩,                   // Record link to type-specific table (null if no properties)
    variants: {                          // Type history for lossless switching
        task: task:⟨uuid⟩,
        text: null  // No properties to preserve
    },
    content: "Implement feature",        // Primary content
    before_sibling_id: "uuid",           // Linked list for sibling ordering
    _schema_version: 1,                  // Universal versioning (moved from type tables)
    created_at: "2025-11-16T...",
    modified_at: "2025-11-16T...",
    embedding_vector: [0.1, 0.2, ...],   // f32 array for AI semantic search
    embedding_stale: false,              // Track if embedding needs regeneration
    mentions: ["node-uuid-1", ...],      // Outgoing mentions
    mentioned_by: ["node-uuid-2", ...]   // Incoming mentions

    // REMOVED: uuid, properties, parent_id, container_node_id
}

{
    id: task:⟨uuid⟩,
    priority: "MEDIUM",      // Flattened (not nested under 'task' object)
    status: "OPEN",
    due_date: "2025-11-20",
    assignee: "user-id",
    started_at: "2025-11-16T...",
    completed_at: null

    // REMOVED: uuid, _schema_version, nested task object
}

// Get node by ID
db.query("SELECT * FROM nodes WHERE uuid = $uuid LIMIT 1;")
    .bind(("uuid", "550e8400-..."))
    .await?;

// Get children
db.query("SELECT * FROM nodes WHERE parent_id = $parent_id")
    .bind(("parent_id", parent_id))
    .await?;

// Get node by ID (use record ID directly)
db.query("SELECT * FROM node:⟨$id⟩ FETCH data;")
    .bind(("id", "550e8400-..."))
    .await?;

// Get children (graph edge traversal)
db.query("SELECT ->has_child->node.* FROM node:⟨$parent_id⟩ FETCH data;")
    .bind(("parent_id", parent_id))
    .await?;

// Get entire tree recursively (ONE QUERY!)
db.query(r#"
    SELECT *,
        ->has_child->node.{..}.{
            id, type, content, before_sibling_id,
            children: ->has_child->node.@
        } AS children
    FROM node:⟨$root_id⟩
    FETCH data;
"#)
    .bind(("root_id", root_id))
    .await?;

// In surreal_store.rs::initialize_schema()

async fn initialize_schema(db: &Arc<Surreal<Db>>) -> Result<()> {
    // 1. Rename nodes to node
    db.query("DEFINE TABLE IF NOT EXISTS node SCHEMALESS;").await?;

    // 2. Add new fields with defaults (backward compatible)
    db.query(r#"
        UPDATE node SET
            data = data ?? null,
            variants = variants ?? {},
            _schema_version = _schema_version ?? 1
        WHERE data IS NONE OR variants IS NONE OR _schema_version IS NONE;
    "#).await?;

    Ok(())
}

// Migration script: packages/core/src/db/migrations/flatten_task_properties.rs

pub async fn migrate_flatten_task_properties(db: &Arc<Surreal<Db>>) -> Result<()> {
    // Get all task records with old structure
    let tasks: Vec<Value> = db.query("SELECT * FROM task;").await?.take(0)?;

    for task in tasks {
        let id = task["id"].as_str().unwrap();

        // Extract properties from nested 'task' object or root level
        let priority = task.get("task")
            .and_then(|t| t.get("priority"))
            .or_else(|| task.get("priority"))
            .and_then(|v| v.as_str())
            .unwrap_or("MEDIUM");

        let status = task.get("task")
            .and_then(|t| t.get("status"))
            .or_else(|| task.get("status"))
            .and_then(|v| v.as_str())
            .unwrap_or("OPEN");

        // Update with flattened structure
        db.query(r#"
            UPDATE $id SET
                priority = $priority,
                status = $status,
                task = NONE,              -- Remove nested object
                uuid = NONE,              -- Remove redundant uuid
                _schema_version = NONE    -- Moved to node table
        "#)
            .bind(("id", id))
            .bind(("priority", priority))
            .bind(("status", status))
            .await?;
    }

    Ok(())
}

// In surreal_store.rs::create_node() - around line 723-742

// Insert into type-specific table (if properties exist)
if !node.properties.as_object().unwrap_or(&serde_json::Map::new()).is_empty() {
    // ❌ OLD WAY (nested, with uuid)
    // let mut props = node.properties.clone();
    // if let Some(obj) = props.as_object_mut() {
    //     obj.insert("uuid".to_string(), serde_json::json!(node.id));
    // }

    // ✅ NEW WAY (flattened, no uuid)
    let props = node.properties.clone();

    self.db
        .query("CREATE type::thing($table, $id) CONTENT $properties;")
        .bind(("table", node.node_type.clone()))
        .bind(("id", node.id.clone()))
        .bind(("properties", props))  // Direct properties, not nested
        .await
        .context("Failed to create node in type-specific table")?;

    // Set data link on node
    let data_link = format!("{}:{}", node.node_type, node.id);
    self.db
        .query("UPDATE node:⟨$id⟩ SET data = type::thing($table, $uuid);")
        .bind(("id", node.id.clone()))
        .bind(("table", node.node_type.clone()))
        .bind(("uuid", node.id.clone()))
        .await?;
}

// Migration script: packages/core/src/db/migrations/create_has_child_edges.rs

pub async fn migrate_create_has_child_edges(db: &Arc<Surreal<Db>>) -> Result<()> {
    // Get all nodes with parents
    let nodes: Vec<SurrealNode> = db.query(
        "SELECT uuid, parent_id FROM node WHERE parent_id IS NOT NONE;"
    ).await?.take(0)?;

    tracing::info!("Creating has_child edges for {} nodes", nodes.len());

    for node in nodes {
        let parent_id = node.parent_id.unwrap();
        let child_id = node.uuid;

        // Create bidirectional edge
        db.query("RELATE node:⟨$parent⟩->has_child->node:⟨$child⟩;")
            .bind(("parent", parent_id))
            .bind(("child", child_id))
            .await
            .with_context(|| format!("Failed to create edge: {} -> {}", parent_id, child_id))?;
    }

    tracing::info!("✅ Created has_child edges successfully");
    Ok(())
}

// In node_service.rs or operations layer

pub async fn validate_no_cycle(
    db: &SurrealStore,
    parent_id: &str,
    child_id: &str
) -> Result<(), NodeServiceError> {
    // Check if parent is a descendant of child (would create cycle)
    let is_descendant: Vec<Value> = db.query(r#"
        SELECT * FROM node:⟨$parent⟩
        WHERE id IN (
            SELECT ->has_child->node.{..} FROM node:⟨$child⟩
        )
        LIMIT 1;
    "#)
        .bind(("parent", parent_id))
        .bind(("child", child_id))
        .await?
        .take(0)?;

    if !is_descendant.is_empty() {
        return Err(NodeServiceError::InvalidOperation(
            format!("Cannot add child {} to parent {} - would create cycle", child_id, parent_id)
        ));
    }

    Ok(())
}

// BEFORE (field-based lookup)
pub async fn get_node(&self, id: &str) -> Result<Option<Node>> {
    let query = "SELECT * FROM nodes WHERE uuid = $uuid LIMIT 1;";
    let mut response = self.db
        .query(query)
        .bind(("uuid", id.to_string()))
        .await?;
    // ...
}

// AFTER (record ID lookup)
pub async fn get_node(&self, id: &str) -> Result<Option<Node>> {
    // Direct record lookup (O(1) primary key access)
    let query = "SELECT * FROM node:⟨$id⟩ FETCH data;";
    let mut response = self.db
        .query(query)
        .bind(("id", id.to_string()))
        .await?;
    // ...
}

// Find all instances of this pattern:
WHERE uuid = $uuid

// Replace with:
FROM node:⟨$id⟩  // Or type::thing('node', $id)

// BEFORE (field-based)
pub async fn get_children(&self, parent_id: &str) -> Result<Vec<Node>> {
    let nodes = self.query_nodes(
        NodeQuery::new().with_parent_id(parent_id.to_string())
    ).await?;
    // Then sort by before_sibling_id in memory
    Ok(nodes)
}

// AFTER (graph-based)
pub async fn get_children(&self, parent_id: &str) -> Result<Vec<Node>> {
    let query = r#"
        SELECT ->has_child->node.*
        FROM node:⟨$parent_id⟩
        FETCH data
        ORDER BY before_sibling_id;
    "#;

    let mut response = self.db
        .query(query)
        .bind(("parent_id", parent_id))
        .await?;

    let surreal_nodes: Vec<SurrealNode> = response.take(0)?;
    Ok(surreal_nodes.into_iter().map(Node::from).collect())
}

// NEW METHOD: Get entire tree in ONE query
pub async fn get_node_tree(&self, root_id: &str) -> Result<NodeTree> {
    let query = r#"
        SELECT *,
            ->has_child->node.{..}.{
                id,
                type,
                content,
                before_sibling_id,
                data,
                created_at,
                modified_at,
                children: ->has_child->node.@
            } AS children
        FROM node:⟨$root_id⟩
        FETCH data;
    "#;

    let mut response = self.db
        .query(query)
        .bind(("root_id", root_id))
        .await?;

    response.take(0)
}

// Return type
#[derive(Debug, Serialize, Deserialize)]
pub struct NodeTree {
    #[serde(flatten)]
    pub node: Node,
    pub children: Vec<NodeTree>,  // Recursive structure
}

// BEFORE (uses container_node_id)
let query = "SELECT * FROM nodes WHERE
    (node_type = 'task' OR container_node_id IS NONE)
    AND content CONTAINS $query
    LIMIT 20;";

// AFTER (uses graph query)
let query = r#"
    SELECT * FROM node
    WHERE (
        type = 'task'
        OR !EXISTS(<-has_child<-node)  -- No parent = root/container node
    )
    AND content CONTAINS $query
    FETCH data
    LIMIT 20;
"#;

// Add to packages/core/src/db/surreal_store_test.rs

#[tokio::test]
async fn benchmark_recursive_tree_query() {
    let store = setup_test_store().await;

    // Create deep tree: 1000 nodes, 10 levels deep
    let root = create_deep_tree(&store, 1000, 10).await;

    // Benchmark recursive query
    let start = Instant::now();
    let tree = store.get_node_tree(&root.id).await.unwrap();
    let duration = start.elapsed();

    // Should complete in <200ms for 1000 nodes
    assert!(duration < Duration::from_millis(200),
        "Recursive query took {:?}, expected <200ms", duration);

    // Verify correctness
    assert_eq!(count_nodes(&tree), 1000);
}

#[tokio::test]
async fn benchmark_get_children() {
    let store = setup_test_store().await;

    // Create parent with 100 children
    let parent = create_node_with_children(&store, 100).await;

    // Benchmark children query
    let start = Instant::now();
    let children = store.get_children(&parent.id).await.unwrap();
    let duration = start.elapsed();

    // Should complete in <50ms for 100 children
    assert!(duration < Duration::from_millis(50),
        "Get children took {:?}, expected <50ms", duration);

    assert_eq!(children.len(), 100);
}

pub async fn migrate_drop_old_fields(db: &Arc<Surreal<Db>>) -> Result<()> {
    tracing::warn!("⚠️  Dropping old fields - this cannot be easily undone!");

    // Confirm all queries have been updated
    tracing::info!("Verifying no code still uses old fields...");

    // Drop fields by setting to NONE (SurrealDB SCHEMALESS allows this)
    db.query(r#"
        UPDATE node SET
            uuid = NONE,
            parent_id = NONE,
            container_node_id = NONE,
            properties = NONE
        WHERE
            uuid IS NOT NONE
            OR parent_id IS NOT NONE
            OR container_node_id IS NOT NONE
            OR properties IS NOT NONE;
    "#).await?;

    tracing::info!("✅ Old fields removed from node table");
    Ok(())
}

pub async fn migrate_delete_empty_type_tables(db: &Arc<Surreal<Db>>) -> Result<()> {
    let empty_tables = [
        "text",
        "date",
        "header",
        "code_block",
        "quote_block",
        "ordered_list"
    ];

    for table in empty_tables {
        tracing::info!("Deleting empty table: {}", table);

        // Verify table is truly empty or only has metadata
        let count: Option<i64> = db.query(format!("SELECT count() FROM {} GROUP ALL;", table))
            .await?
            .take(0)?;

        if let Some(c) = count {
            tracing::warn!("Table {} has {} records - verifying they're metadata only", table, c);
        }

        // Delete the table
        db.query(format!("REMOVE TABLE {};", table))
            .await
            .with_context(|| format!("Failed to delete table {}", table))?;
    }

    tracing::info!("✅ Deleted empty type tables");
    Ok(())
}

// In surreal_store.rs::initialize_schema() - around line 319-336

// BEFORE
let core_types = [
    "task", "text", "date", "header",
    "code_block", "quote_block", "ordered_list"
];

// AFTER
let core_types = [
    "task",  // Only type with properties
];

for node_type in core_types {
    db.query(format!("DEFINE TABLE IF NOT EXISTS {} SCHEMALESS;", node_type))
        .await?;
}

// In create_node() - around line 723-742

// Insert into type-specific table ONLY for types with properties
if node.node_type == "task" && !node.properties.is_empty() {
    // Create task record and set data link
    // ... (existing logic)
} else {
    // Types without properties: set data = null
    self.db
        .query("UPDATE node:⟨$id⟩ SET data = null;")
        .bind(("id", node.id.clone()))
        .await?;
}

// Background job: Run weekly or on-demand
pub async fn cleanup_orphaned_type_records(db: &SurrealStore) -> Result<()> {
    // Find task records not referenced by any node
    let orphaned: Vec<Value> = db.query(r#"
        SELECT id FROM task
        WHERE id NOT IN (
            SELECT data FROM node WHERE type = 'task' AND data IS NOT NONE
            UNION
            SELECT variants.task FROM node WHERE variants.task IS NOT NONE
        );
    "#).await?.take(0)?;

    tracing::info!("Found {} orphaned task records", orphaned.len());

    for record in orphaned {
        let id = record["id"].as_str().unwrap();
        db.query("DELETE $id;").bind(("id", id)).await?;
        tracing::debug!("Deleted orphaned record: {}", id);
    }

    Ok(())
}

pub async fn cleanup_old_variants(db: &SurrealStore) -> Result<()> {
    const MAX_VARIANTS: usize = 3;  // Keep last 3 type changes

    // For nodes with >3 variants, remove oldest
    // (This requires custom logic based on your variant tracking)
    // For now, just document the concern

    Ok(())
}

// packages/core/src/db/surreal_store_test.rs

#[tokio::test]
async fn test_create_node_without_properties() {
    // Text node - no type table
    let text_node = Node::new("text".into(), "Just text".into(), None, json!({}));
    let created = store.create_node(text_node).await.unwrap();
    assert_eq!(created.data, None);  // No data link
}

#[tokio::test]
async fn test_create_task_with_properties() {
    // Task node - has type table
    let task_node = Node::new(
        "task".into(),
        "Do thing".into(),
        None,
        json!({"priority": "HIGH", "status": "OPEN"})
    );
    let created = store.create_node(task_node).await.unwrap();
    assert!(created.data.is_some());  // Has data link to task table
}

#[tokio::test]
async fn test_type_switching_preserves_data() {
    // Create task
    let task = create_task_node(&store, "Test task").await;

    // Switch to text
    store.update_node(&task.id, NodeUpdate {
        node_type: Some("text".into()),
        ..Default::default()
    }).await.unwrap();

    // Verify old task data in variants
    let node = store.get_node(&task.id).await.unwrap().unwrap();
    assert!(node.variants.contains_key("task"));

    // Switch back to task
    store.update_node(&node.id, NodeUpdate {
        node_type: Some("task".into()),
        ..Default::default()
    }).await.unwrap();

    // Verify task properties restored
    let restored = store.get_node(&node.id).await.unwrap().unwrap();
    // Check if data link points to old task record (from variants)
}

#[tokio::test]
async fn test_cyclic_reference_prevented() {
    let node_a = create_node(&store, "A").await;
    let node_b = create_node(&store, "B").await;
    let node_c = create_node(&store, "C").await;

    // A -> B -> C
    store.add_child(&node_a.id, &node_b.id).await.unwrap();
    store.add_child(&node_b.id, &node_c.id).await.unwrap();

    // Try C -> A (would create cycle)
    let result = store.add_child(&node_c.id, &node_a.id).await;
    assert!(result.is_err());
    assert!(result.unwrap_err().to_string().contains("cycle"));
}

#[tokio::test]
async fn test_recursive_tree_fetch() {
    // Create tree: root -> [child1, child2 -> [grandchild1, grandchild2]]
    let tree = create_test_tree(&store).await;

    // Fetch entire tree in one query
    let result = store.get_node_tree(&tree.root.id).await.unwrap();

    // Verify structure
    assert_eq!(result.children.len(), 2);
    assert_eq!(result.children[1].children.len(), 2);

    // Verify data hydrated (FETCH data worked)
    if result.node.type == "task" {
        assert!(result.node.data.is_some());
    }
}

// Node compositionally IS a task
{
    id: node:⟨abc⟩,
    type: "task",
    data: task:⟨abc⟩,  // ✅ Record link (composition)
}

// Query: Get task with properties
SELECT * FROM node:⟨abc⟩ FETCH data;