flow_db/
migration.rs

1use flow_core::Result;
2use rusqlite::Connection;
3
4const CURRENT_VERSION: i32 = 1;
5
6/// Run database migrations to bring the schema up to date.
7pub fn run_migrations(conn: &Connection) -> Result<()> {
8    let version: i32 = conn
9        .pragma_query_value(None, "user_version", |row| row.get(0))
10        .map_err(|e| flow_core::FlowError::Database(format!("failed to get user_version: {e}")))?;
11
12    if version < 1 {
13        migrate_v1(conn)?;
14    }
15
16    // Future migrations would go here
17    // if version < 2 { migrate_v2(conn)?; }
18
19    Ok(())
20}
21
22fn migrate_v1(conn: &Connection) -> Result<()> {
23    tracing::info!("Running migration v1: creating initial schema");
24
25    // Create features table
26    conn.execute(
27        r"
28        CREATE TABLE IF NOT EXISTS features (
29            id INTEGER PRIMARY KEY AUTOINCREMENT,
30            priority INTEGER NOT NULL DEFAULT 0,
31            category TEXT NOT NULL DEFAULT '',
32            name TEXT NOT NULL,
33            description TEXT NOT NULL DEFAULT '',
34            steps TEXT NOT NULL DEFAULT '[]',
35            passes INTEGER NOT NULL DEFAULT 0,
36            in_progress INTEGER NOT NULL DEFAULT 0,
37            dependencies TEXT NOT NULL DEFAULT '[]',
38            created_at TEXT NOT NULL DEFAULT (datetime('now')),
39            updated_at TEXT NOT NULL DEFAULT (datetime('now'))
40        )
41        ",
42        [],
43    )
44    .map_err(|e| flow_core::FlowError::Database(format!("failed to create features table: {e}")))?;
45
46    // Create change_events table
47    conn.execute(
48        r"
49        CREATE TABLE IF NOT EXISTS change_events (
50            id INTEGER PRIMARY KEY AUTOINCREMENT,
51            feature_id INTEGER NOT NULL,
52            event_type TEXT NOT NULL,
53            field TEXT,
54            old_value TEXT,
55            new_value TEXT,
56            agent TEXT,
57            source TEXT NOT NULL,
58            created_at TEXT NOT NULL DEFAULT (datetime('now')),
59            FOREIGN KEY (feature_id) REFERENCES features(id) ON DELETE CASCADE
60        )
61        ",
62        [],
63    )
64    .map_err(|e| {
65        flow_core::FlowError::Database(format!("failed to create change_events table: {e}"))
66    })?;
67
68    // Create indexes for common queries
69    conn.execute(
70        "CREATE INDEX IF NOT EXISTS ix_feature_status ON features(passes, in_progress)",
71        [],
72    )
73    .map_err(|e| flow_core::FlowError::Database(format!("failed to create status index: {e}")))?;
74
75    conn.execute(
76        "CREATE INDEX IF NOT EXISTS ix_feature_priority ON features(priority)",
77        [],
78    )
79    .map_err(|e| flow_core::FlowError::Database(format!("failed to create priority index: {e}")))?;
80
81    conn.execute(
82        "CREATE INDEX IF NOT EXISTS ix_change_events_feature ON change_events(feature_id)",
83        [],
84    )
85    .map_err(|e| flow_core::FlowError::Database(format!("failed to create events index: {e}")))?;
86
87    // Update version
88    conn.pragma_update(None, "user_version", CURRENT_VERSION)
89        .map_err(|e| {
90            flow_core::FlowError::Database(format!("failed to update user_version: {e}"))
91        })?;
92
93    tracing::info!("Migration v1 complete");
94    Ok(())
95}
96
97#[cfg(test)]
98mod tests {
99    use super::*;
100
101    #[test]
102    fn test_migrations() {
103        let conn = Connection::open_in_memory().unwrap();
104
105        // Initially version should be 0
106        let version: i32 = conn
107            .pragma_query_value(None, "user_version", |row| row.get(0))
108            .unwrap();
109        assert_eq!(version, 0);
110
111        // Run migrations
112        run_migrations(&conn).unwrap();
113
114        // Version should now be CURRENT_VERSION
115        let version: i32 = conn
116            .pragma_query_value(None, "user_version", |row| row.get(0))
117            .unwrap();
118        assert_eq!(version, CURRENT_VERSION);
119
120        // Tables should exist
121        let tables: Vec<String> = conn
122            .prepare("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")
123            .unwrap()
124            .query_map([], |row| row.get(0))
125            .unwrap()
126            .collect::<std::result::Result<Vec<_>, _>>()
127            .unwrap();
128
129        assert!(tables.contains(&"features".to_string()));
130        assert!(tables.contains(&"change_events".to_string()));
131
132        // Running migrations again should be idempotent
133        run_migrations(&conn).unwrap();
134        let version: i32 = conn
135            .pragma_query_value(None, "user_version", |row| row.get(0))
136            .unwrap();
137        assert_eq!(version, CURRENT_VERSION);
138    }
139
140    #[test]
141    fn test_feature_table_schema() {
142        let conn = Connection::open_in_memory().unwrap();
143        run_migrations(&conn).unwrap();
144
145        // Verify we can insert and retrieve a feature
146        conn.execute(
147            r"
148            INSERT INTO features (name, description, priority, category, steps, dependencies)
149            VALUES (?, ?, ?, ?, ?, ?)
150            ",
151            rusqlite::params![
152                "Test Feature",
153                "Test Description",
154                1,
155                "Test",
156                "[]",
157                "[1, 2]"
158            ],
159        )
160        .unwrap();
161
162        let (name, deps): (String, String) = conn
163            .query_row(
164                "SELECT name, dependencies FROM features WHERE id = 1",
165                [],
166                |row| Ok((row.get(0)?, row.get(1)?)),
167            )
168            .unwrap();
169
170        assert_eq!(name, "Test Feature");
171        assert_eq!(deps, "[1, 2]");
172    }
173}
flow_db/migration.rs

flow_db/
migration.rs
