udb/generation/
dsn.rs

1use serde::{Deserialize, Serialize};
2use std::env;
3
4use crate::ast::ProtoSchema;
5use crate::backend::BackendKind;
6
7use super::manifest::{CatalogManifest, ManifestStore};
8
9#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
10pub struct DsnGenerationConfig {
11    pub sql_env_key: String,
12    pub nosql_env_key: String,
13    pub vector_env_key: String,
14    pub graph_env_key: String,
15    pub timeseries_env_key: String,
16    pub column_env_key: String,
17    pub cache_env_key: String,
18    pub object_env_key: String,
19    pub fallback_env_key: String,
20}
21
22impl Default for DsnGenerationConfig {
23    fn default() -> Self {
24        Self {
25            sql_env_key: "UDB_SQL_DSN".to_string(),
26            nosql_env_key: "UDB_NOSQL_DSN".to_string(),
27            vector_env_key: "UDB_VECTOR_DSN".to_string(),
28            graph_env_key: "UDB_GRAPH_DSN".to_string(),
29            timeseries_env_key: "UDB_TIMESERIES_DSN".to_string(),
30            column_env_key: "UDB_COLUMN_DSN".to_string(),
31            cache_env_key: "UDB_CACHE_DSN".to_string(),
32            object_env_key: "UDB_OBJECT_DSN".to_string(),
33            fallback_env_key: "UDB_GENERIC_DSN".to_string(),
34        }
35    }
36}
37
38#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
39pub struct UnifiedDsnCatalog {
40    pub checksum_sha256: String,
41    pub entries: Vec<UnifiedDsn>,
42}
43
44#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
45pub struct UnifiedDsn {
46    pub id: String,
47    pub message_name: String,
48    pub store_kind: String,
49    pub backend: String,
50    pub env_key: String,
51    pub dsn: String,
52    pub resource_uri: String,
53    pub owner_schema: String,
54    pub owner_table: String,
55}
56
57#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
58pub struct ParsedUnifiedDsn {
59    pub scheme: String,
60    pub tier: String,
61    pub backend: String,
62    pub env_key: String,
63    pub resource_path: String,
64    pub resource_parts: Vec<String>,
65}
66
67#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
68pub struct ResolvedUnifiedDsn {
69    pub id: String,
70    pub dsn: String,
71    pub env_key: String,
72    pub base_dsn: String,
73    pub redacted_base_dsn: String,
74    pub resource_path: String,
75    pub valid: bool,
76    pub error: String,
77}
78
79#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
80pub struct DsnValidationReport {
81    pub passed: bool,
82    pub errors: Vec<String>,
83}
84
85impl UnifiedDsn {
86    pub fn parsed(&self) -> Result<ParsedUnifiedDsn, String> {
87        parse_unified_dsn(&self.dsn)
88    }
89
90    pub fn validate(&self) -> DsnValidationReport {
91        validate_unified_dsn(self)
92    }
93
94    pub fn resolved_from_env(&self) -> ResolvedUnifiedDsn {
95        resolve_unified_dsn(self)
96    }
97}
98
99pub fn generate_unified_dsn_catalog(
100    schemas: &[ProtoSchema],
101    config: &DsnGenerationConfig,
102) -> Result<UnifiedDsnCatalog, serde_json::Error> {
103    let manifest = CatalogManifest::from_schemas(schemas)?;
104    let mut entries = Vec::new();
105
106    for table in &manifest.tables {
107        let env_key = config.sql_env_key.clone();
108        let backend = "postgres".to_string();
109        entries.push(UnifiedDsn {
110            id: format!("sql:{}:{}", table.schema, table.table),
111            message_name: table.message_name.clone(),
112            store_kind: "sql".to_string(),
113            backend: backend.clone(),
114            env_key: env_key.clone(),
115            dsn: format!(
116                "{}://env:{}/{}/{}",
117                BackendKind::Postgres.dsn_scheme(),
118                env_key,
119                table.schema,
120                table.table
121            ),
122            resource_uri: format!("sql://{}/{}", table.schema, table.table),
123            owner_schema: table.schema.clone(),
124            owner_table: table.table.clone(),
125        });
126    }
127
128    for store in &manifest.stores {
129        entries.push(dsn_from_store(store, config));
130    }
131
132    entries.sort_by(|a, b| a.id.cmp(&b.id));
133    Ok(UnifiedDsnCatalog {
134        checksum_sha256: manifest.checksum_sha256,
135        entries,
136    })
137}
138
139pub fn parse_unified_dsn(dsn: &str) -> Result<ParsedUnifiedDsn, String> {
140    let (scheme, rest) = dsn
141        .split_once("://")
142        .ok_or_else(|| "DSN must contain ://".to_string())?;
143    let scheme_parts = scheme.split('+').collect::<Vec<_>>();
144    if scheme_parts.len() != 3 || scheme_parts[0] != "udb" {
145        return Err("DSN scheme must be udb+<tier>+<backend>".to_string());
146    }
147    let tier = scheme_parts[1].trim();
148    let backend = scheme_parts[2].trim();
149    if tier.is_empty() || backend.is_empty() {
150        return Err("DSN tier and backend must be non-empty".to_string());
151    }
152    if BackendKind::from_store_kind(tier, backend).is_none() && backend != "generic" {
153        return Err(format!(
154            "unsupported backend '{}' for tier '{}'",
155            backend, tier
156        ));
157    }
158
159    let (authority, resource_path) = rest
160        .split_once('/')
161        .ok_or_else(|| "DSN must include env authority and resource path".to_string())?;
162    let env_key = authority
163        .strip_prefix("env:")
164        .ok_or_else(|| "DSN authority must be env:<ENV_KEY>".to_string())?
165        .trim();
166    if env_key.is_empty() {
167        return Err("DSN env key must be non-empty".to_string());
168    }
169    let resource_path = resource_path.trim_matches('/');
170    if resource_path.is_empty() {
171        return Err("DSN resource path must be non-empty".to_string());
172    }
173
174    Ok(ParsedUnifiedDsn {
175        scheme: scheme.to_string(),
176        tier: tier.to_string(),
177        backend: backend.to_string(),
178        env_key: env_key.to_string(),
179        resource_path: resource_path.to_string(),
180        resource_parts: resource_path
181            .split('/')
182            .filter(|part| !part.trim().is_empty())
183            .map(str::to_string)
184            .collect(),
185    })
186}
187
188pub fn validate_unified_dsn(entry: &UnifiedDsn) -> DsnValidationReport {
189    let mut errors = Vec::new();
190    match parse_unified_dsn(&entry.dsn) {
191        Ok(parsed) => {
192            if parsed.backend != entry.backend {
193                errors.push(format!(
194                    "backend mismatch: entry={} dsn={}",
195                    entry.backend, parsed.backend
196                ));
197            }
198            if parsed.env_key != entry.env_key {
199                errors.push(format!(
200                    "env key mismatch: entry={} dsn={}",
201                    entry.env_key, parsed.env_key
202                ));
203            }
204            if !entry.owner_schema.trim().is_empty()
205                && !parsed
206                    .resource_parts
207                    .iter()
208                    .any(|part| part == &entry.owner_schema)
209            {
210                errors.push(format!(
211                    "resource path '{}' does not include owner schema '{}'",
212                    parsed.resource_path, entry.owner_schema
213                ));
214            }
215        }
216        Err(err) => errors.push(err),
217    }
218    DsnValidationReport {
219        passed: errors.is_empty(),
220        errors,
221    }
222}
223
224pub fn resolve_unified_dsn(entry: &UnifiedDsn) -> ResolvedUnifiedDsn {
225    match parse_unified_dsn(&entry.dsn) {
226        Ok(parsed) => match env::var(&parsed.env_key) {
227            Ok(base_dsn) => {
228                if let Err(err) = validate_base_dsn(&base_dsn) {
229                    return ResolvedUnifiedDsn {
230                        id: entry.id.clone(),
231                        dsn: entry.dsn.clone(),
232                        env_key: parsed.env_key,
233                        redacted_base_dsn: redact_dsn(&base_dsn),
234                        base_dsn,
235                        resource_path: parsed.resource_path,
236                        valid: false,
237                        error: err,
238                    };
239                }
240                ResolvedUnifiedDsn {
241                    id: entry.id.clone(),
242                    dsn: entry.dsn.clone(),
243                    env_key: parsed.env_key,
244                    redacted_base_dsn: redact_dsn(&base_dsn),
245                    base_dsn,
246                    resource_path: parsed.resource_path,
247                    valid: true,
248                    error: String::new(),
249                }
250            }
251            Err(err) => ResolvedUnifiedDsn {
252                id: entry.id.clone(),
253                dsn: entry.dsn.clone(),
254                env_key: parsed.env_key,
255                resource_path: parsed.resource_path,
256                valid: false,
257                error: format!("{}: {}", entry.env_key, err),
258                ..ResolvedUnifiedDsn::default()
259            },
260        },
261        Err(err) => ResolvedUnifiedDsn {
262            id: entry.id.clone(),
263            dsn: entry.dsn.clone(),
264            valid: false,
265            error: err,
266            ..ResolvedUnifiedDsn::default()
267        },
268    }
269}
270
271fn validate_base_dsn(dsn: &str) -> Result<(), String> {
272    let (scheme, rest) = dsn
273        .split_once("://")
274        .ok_or_else(|| "base DSN is missing scheme://".to_string())?;
275    if scheme.trim().is_empty() {
276        return Err("base DSN scheme is empty".to_string());
277    }
278    let authority = rest.split('/').next().unwrap_or_default();
279    let host = authority.rsplit('@').next().unwrap_or_default();
280    if host.trim().is_empty() {
281        return Err("base DSN host is empty".to_string());
282    }
283    Ok(())
284}
285
286pub fn resolve_unified_dsn_catalog(catalog: &UnifiedDsnCatalog) -> Vec<ResolvedUnifiedDsn> {
287    catalog.entries.iter().map(resolve_unified_dsn).collect()
288}
289
290pub fn redact_dsn(dsn: &str) -> String {
291    let Some((scheme, rest)) = dsn.split_once("://") else {
292        return dsn.to_string();
293    };
294    // Use split_once('@') instead of find('@') + byte-slice indexing.
295    // Byte-slicing with find() is unsafe when the '@' symbol is followed by a
296    // multi-byte UTF-8 character: `rest[at_idx + 1..]` would panic if at_idx+1
297    // falls in the middle of a multi-byte sequence. split_once is always safe.
298    let Some((auth, host_and_path)) = rest.split_once('@') else {
299        return dsn.to_string();
300    };
301    if let Some((user, _password)) = auth.split_once(':') {
302        format!("{scheme}://{user}:***@{host_and_path}")
303    } else {
304        format!("{scheme}://***@{host_and_path}")
305    }
306}
307
308fn dsn_from_store(store: &ManifestStore, config: &DsnGenerationConfig) -> UnifiedDsn {
309    let env_key = if !store.dsn_env_key.trim().is_empty() {
310        store.dsn_env_key.clone()
311    } else {
312        env_key_for_kind(&store.store_kind, config).to_string()
313    };
314    let backend_kind = BackendKind::from_store_kind(&store.store_kind, &store.backend);
315    let backend = backend_kind
316        .as_ref()
317        .map(|kind| kind.as_str().to_string())
318        .unwrap_or_else(|| {
319            if store.backend.trim().is_empty() {
320                "generic".to_string()
321            } else {
322                store.backend.clone()
323            }
324        });
325    let scheme = backend_kind
326        .as_ref()
327        .map(|kind| kind.dsn_scheme())
328        .unwrap_or_else(|| format!("udb+{}+{}", store.store_kind, backend));
329    let dsn = if !store.dsn.trim().is_empty() {
330        store.dsn.clone()
331    } else {
332        format!("{}://env:{}/{}", scheme, env_key, store_path(store))
333    };
334
335    UnifiedDsn {
336        id: format!(
337            "{}:{}:{}:{}",
338            store.store_kind, backend, store.owner_schema, store.resource_name
339        ),
340        message_name: store.logical_name.clone(),
341        store_kind: store.store_kind.clone(),
342        backend: backend.clone(),
343        env_key,
344        dsn,
345        resource_uri: format!("{}://{}/{}", store.store_kind, backend, store_path(store)),
346        owner_schema: store.owner_schema.clone(),
347        owner_table: store.owner_table.clone(),
348    }
349}
350
351fn env_key_for_kind<'a>(kind: &str, config: &'a DsnGenerationConfig) -> &'a str {
352    match kind {
353        "sql" => &config.sql_env_key,
354        "nosql" | "document" => &config.nosql_env_key,
355        "vector" => &config.vector_env_key,
356        "graph" => &config.graph_env_key,
357        "timeseries" | "time-series" => &config.timeseries_env_key,
358        "column" | "columnar" | "wide-column" => &config.column_env_key,
359        "cache" => &config.cache_env_key,
360        "object" | "blob" | "storage" => &config.object_env_key,
361        _ => &config.fallback_env_key,
362    }
363}
364
365fn store_path(store: &ManifestStore) -> String {
366    let mut parts = Vec::new();
367    if !store.database_name.trim().is_empty() {
368        parts.push(store.database_name.as_str());
369    }
370    if !store.namespace.trim().is_empty() {
371        parts.push(store.namespace.as_str());
372    }
373    if !store.resource_name.trim().is_empty() {
374        parts.push(store.resource_name.as_str());
375    }
376    if parts.is_empty() {
377        format!("{}/{}", store.owner_schema, store.owner_table)
378    } else {
379        parts.join("/")
380    }
381}
udb/generation/dsn.rs

udb/generation/
dsn.rs
