zebflow 0.8.3

Reactive web automation platform — pipelines, templates, and real-time collaboration in a single binary.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

//! Portable SQLite schema export/import for project bundles.
//!
//! Project bundles do not ship local SQLite data through Hub. When requested,
//! they carry `schemas/sqlite/schema.sql`, which is enough to initialize a new
//! project's `data/local.db` with the same table/index/view/trigger structure.

use std::path::{Path, PathBuf};

use rusqlite::Connection;

use crate::platform::error::PlatformError;
use crate::platform::model::slug_segment;

pub const REPO_SCHEMA_PATH: &str = "schemas/sqlite/schema.sql";

fn project_repo_dir(data_root: &Path, owner: &str, project: &str) -> PathBuf {
    data_root
        .join("users")
        .join(slug_segment(owner))
        .join(slug_segment(project))
        .join("repo")
}

fn project_data_dir(data_root: &Path, owner: &str, project: &str) -> PathBuf {
    data_root
        .join("users")
        .join(slug_segment(owner))
        .join(slug_segment(project))
        .join("data")
}

pub fn local_db_path(data_root: &Path, owner: &str, project: &str) -> PathBuf {
    project_data_dir(data_root, owner, project).join("local.db")
}

pub fn repo_schema_path(data_root: &Path, owner: &str, project: &str) -> PathBuf {
    project_repo_dir(data_root, owner, project).join(REPO_SCHEMA_PATH)
}

pub fn has_schema(data_root: &Path, owner: &str, project: &str) -> Result<bool, PlatformError> {
    let path = local_db_path(data_root, owner, project);
    if !path.is_file() {
        return Ok(false);
    }
    let conn = Connection::open(&path)
        .map_err(|err| PlatformError::new("SQLITE_SCHEMA_OPEN", err.to_string()))?;
    let count: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM sqlite_master \
             WHERE type IN ('table','index','view','trigger') \
             AND name NOT LIKE 'sqlite_%' \
             AND sql IS NOT NULL",
            [],
            |row| row.get(0),
        )
        .map_err(|err| PlatformError::new("SQLITE_SCHEMA_READ", err.to_string()))?;
    Ok(count > 0)
}

pub fn export_schema_sql(
    data_root: &Path,
    owner: &str,
    project: &str,
) -> Result<Option<String>, PlatformError> {
    if !has_schema(data_root, owner, project)? {
        return Ok(None);
    }
    let path = local_db_path(data_root, owner, project);
    let conn = Connection::open(&path)
        .map_err(|err| PlatformError::new("SQLITE_SCHEMA_OPEN", err.to_string()))?;
    let mut stmt = conn
        .prepare(
            "SELECT sql FROM sqlite_master \
             WHERE type IN ('table','index','view','trigger') \
             AND name NOT LIKE 'sqlite_%' \
             AND sql IS NOT NULL \
             ORDER BY CASE type \
                WHEN 'table' THEN 0 WHEN 'view' THEN 1 WHEN 'index' THEN 2 ELSE 3 END, name",
        )
        .map_err(|err| PlatformError::new("SQLITE_SCHEMA_READ", err.to_string()))?;
    let rows = stmt
        .query_map([], |row| row.get::<_, String>(0))
        .map_err(|err| PlatformError::new("SQLITE_SCHEMA_READ", err.to_string()))?;
    let mut statements = Vec::new();
    for row in rows {
        let sql = row.map_err(|err| PlatformError::new("SQLITE_SCHEMA_READ", err.to_string()))?;
        let sql = sql.trim().trim_end_matches(';');
        if !sql.is_empty() {
            statements.push(format!("{sql};"));
        }
    }
    if statements.is_empty() {
        Ok(None)
    } else {
        Ok(Some(format!(
            "-- Zebflow portable SQLite schema\n{}\n",
            statements.join("\n")
        )))
    }
}

pub fn apply_schema_from_repo(
    data_root: &Path,
    owner: &str,
    project: &str,
) -> Result<bool, PlatformError> {
    let schema_path = repo_schema_path(data_root, owner, project);
    if !schema_path.is_file() {
        return Ok(false);
    }
    if has_schema(data_root, owner, project)? {
        return Ok(false);
    }
    let sql = std::fs::read_to_string(&schema_path)?;
    let db_path = local_db_path(data_root, owner, project);
    if let Some(parent) = db_path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let conn = Connection::open(&db_path)
        .map_err(|err| PlatformError::new("SQLITE_SCHEMA_OPEN", err.to_string()))?;
    conn.execute_batch(&sql)
        .map_err(|err| PlatformError::new("SQLITE_SCHEMA_APPLY", err.to_string()))?;
    Ok(true)
}

pub fn execute_sql(
    data_root: &Path,
    owner: &str,
    project: &str,
    sql: &str,
) -> Result<(), PlatformError> {
    let trimmed = sql.trim();
    if trimmed.is_empty() {
        return Ok(());
    }
    let db_path = local_db_path(data_root, owner, project);
    if let Some(parent) = db_path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let conn = Connection::open(&db_path)
        .map_err(|err| PlatformError::new("SQLITE_INITIAL_DATA_OPEN", err.to_string()))?;
    conn.execute_batch(trimmed)
        .map_err(|err| PlatformError::new("SQLITE_INITIAL_DATA_APPLY", err.to_string()))?;
    Ok(())
}