1use std::path::{Path, PathBuf};
2use std::sync::Mutex;
3
4use anyhow::{Context, Result};
5use rusqlite::{Connection, OptionalExtension, params};
6
7use crate::flow_meta::{FlowMeta, FlowMetaSource};
8
9pub struct FlowRegistry {
10 path: PathBuf,
11 conn: Mutex<Connection>,
12}
13
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct FlowRevision {
16 pub id: i64,
17 pub flow_name: String,
18 pub version: String,
19 pub content: String,
20 pub content_hash: String,
21 pub ts: chrono::DateTime<chrono::Utc>,
22 pub author: Option<String>,
23 pub source_tag: String,
24 pub origin_path: Option<String>,
25}
26
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub enum SnapshotOutcome {
29 Inserted(FlowRevision),
30 UnchangedFromLatest(FlowRevision),
31}
32
33impl FlowRegistry {
34 pub fn open(project_dir: &Path) -> Result<Self> {
35 let path = project_dir.join(".atman").join("flow-registry.db");
36 if let Some(parent) = path.parent() {
37 std::fs::create_dir_all(parent)
38 .with_context(|| format!("mkdir {}", parent.display()))?;
39 }
40 let conn = Connection::open(&path).with_context(|| format!("open {}", path.display()))?;
41 conn.pragma_update(None, "journal_mode", "WAL")?;
42 conn.pragma_update(None, "busy_timeout", 5000)?;
43 conn.pragma_update(None, "synchronous", "NORMAL")?;
44 conn.execute_batch(SCHEMA)
45 .with_context(|| format!("apply schema on {}", path.display()))?;
46 migrate(&conn).with_context(|| format!("migrate registry at {}", path.display()))?;
47 Ok(Self {
48 path,
49 conn: Mutex::new(conn),
50 })
51 }
52
53 pub fn path(&self) -> &Path {
54 &self.path
55 }
56
57 pub fn snapshot(
58 &self,
59 flow_name: &str,
60 content: &str,
61 meta: &FlowMeta,
62 origin_path: Option<&Path>,
63 ) -> Result<SnapshotOutcome> {
64 let content_hash = FlowMeta::short_hash(content);
65 let origin_str = origin_path.map(|p| p.to_string_lossy().into_owned());
66 if let Some(mut latest) = self.latest(flow_name)?
67 && latest.content_hash == content_hash
68 {
69 if latest.origin_path.is_none() && origin_str.is_some() {
70 let conn = self.conn.lock().unwrap();
71 conn.execute(
72 "UPDATE flow_revisions SET origin_path = ?1 WHERE id = ?2",
73 params![origin_str, latest.id],
74 )?;
75 latest.origin_path = origin_str.clone();
76 }
77 return Ok(SnapshotOutcome::UnchangedFromLatest(latest));
78 }
79 let ts = chrono::Utc::now();
80 let source_tag = match meta.source {
81 FlowMetaSource::Sidecar => "sidecar",
82 FlowMetaSource::HashFallback => "hash",
83 };
84 let conn = self.conn.lock().unwrap();
85 conn.execute(
86 "INSERT INTO flow_revisions \
87 (flow_name, version, content, content_hash, ts, author, source_tag, origin_path) \
88 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
89 params![
90 flow_name,
91 meta.version,
92 content,
93 content_hash,
94 ts.to_rfc3339(),
95 meta.author,
96 source_tag,
97 origin_str,
98 ],
99 )?;
100 let id = conn.last_insert_rowid();
101 Ok(SnapshotOutcome::Inserted(FlowRevision {
102 id,
103 flow_name: flow_name.to_string(),
104 version: meta.version.clone(),
105 content: content.to_string(),
106 content_hash,
107 ts,
108 author: meta.author.clone(),
109 source_tag: source_tag.to_string(),
110 origin_path: origin_str,
111 }))
112 }
113
114 pub fn list_versions(&self, flow_name: &str) -> Result<Vec<FlowRevision>> {
115 let conn = self.conn.lock().unwrap();
116 let mut stmt = conn.prepare(
117 "SELECT id, flow_name, version, content, content_hash, ts, author, source_tag, origin_path \
118 FROM flow_revisions WHERE flow_name = ?1 ORDER BY id DESC",
119 )?;
120 let rows = stmt.query_map([flow_name], row_to_revision)?;
121 let mut out = Vec::new();
122 for r in rows {
123 out.push(r?);
124 }
125 Ok(out)
126 }
127
128 pub fn latest(&self, flow_name: &str) -> Result<Option<FlowRevision>> {
129 let conn = self.conn.lock().unwrap();
130 conn.query_row(
131 "SELECT id, flow_name, version, content, content_hash, ts, author, source_tag, origin_path \
132 FROM flow_revisions WHERE flow_name = ?1 ORDER BY id DESC LIMIT 1",
133 [flow_name],
134 row_to_revision,
135 )
136 .optional()
137 .map_err(Into::into)
138 }
139
140 pub fn find_by_version(
141 &self,
142 flow_name: &str,
143 version_or_hash: &str,
144 ) -> Result<Option<FlowRevision>> {
145 let conn = self.conn.lock().unwrap();
146 let exact = conn
147 .query_row(
148 "SELECT id, flow_name, version, content, content_hash, ts, author, source_tag, origin_path \
149 FROM flow_revisions WHERE flow_name = ?1 AND version = ?2 \
150 ORDER BY id DESC LIMIT 1",
151 params![flow_name, version_or_hash],
152 row_to_revision,
153 )
154 .optional()?;
155 if let Some(r) = exact {
156 return Ok(Some(r));
157 }
158 let stripped = version_or_hash
159 .strip_prefix("hash:")
160 .unwrap_or(version_or_hash);
161 if stripped.is_empty() {
162 return Ok(None);
163 }
164 let prefix_pattern = format!("{stripped}%");
165 conn.query_row(
166 "SELECT id, flow_name, version, content, content_hash, ts, author, source_tag, origin_path \
167 FROM flow_revisions WHERE flow_name = ?1 AND content_hash LIKE ?2 \
168 ORDER BY id DESC LIMIT 1",
169 params![flow_name, prefix_pattern],
170 row_to_revision,
171 )
172 .optional()
173 .map_err(Into::into)
174 }
175
176 pub fn count(&self, flow_name: &str) -> Result<u64> {
177 let conn = self.conn.lock().unwrap();
178 let n: i64 = conn.query_row(
179 "SELECT COUNT(*) FROM flow_revisions WHERE flow_name = ?1",
180 [flow_name],
181 |r| r.get(0),
182 )?;
183 Ok(n as u64)
184 }
185
186 pub fn flow_names(&self) -> Result<Vec<String>> {
187 let conn = self.conn.lock().unwrap();
188 let mut stmt =
189 conn.prepare("SELECT DISTINCT flow_name FROM flow_revisions ORDER BY flow_name ASC")?;
190 let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
191 let mut out = Vec::new();
192 for r in rows {
193 out.push(r?);
194 }
195 Ok(out)
196 }
197}
198
199fn row_to_revision(row: &rusqlite::Row<'_>) -> rusqlite::Result<FlowRevision> {
200 let ts_str: String = row.get(5)?;
201 let ts = chrono::DateTime::parse_from_rfc3339(&ts_str)
202 .map(|dt| dt.with_timezone(&chrono::Utc))
203 .unwrap_or_else(|_| chrono::Utc::now());
204 Ok(FlowRevision {
205 id: row.get(0)?,
206 flow_name: row.get(1)?,
207 version: row.get(2)?,
208 content: row.get(3)?,
209 content_hash: row.get(4)?,
210 ts,
211 author: row.get(6)?,
212 source_tag: row.get(7)?,
213 origin_path: row.get(8)?,
214 })
215}
216
217fn migrate(conn: &Connection) -> Result<()> {
218 let mut has_origin = false;
219 let mut stmt = conn.prepare("PRAGMA table_info(flow_revisions)")?;
220 let rows = stmt.query_map([], |r| r.get::<_, String>(1))?;
221 for name in rows {
222 if name? == "origin_path" {
223 has_origin = true;
224 }
225 }
226 drop(stmt);
227 if !has_origin {
228 conn.execute("ALTER TABLE flow_revisions ADD COLUMN origin_path TEXT", [])?;
229 }
230 Ok(())
231}
232
233const SCHEMA: &str = r#"
234CREATE TABLE IF NOT EXISTS flow_revisions (
235 id INTEGER PRIMARY KEY AUTOINCREMENT,
236 flow_name TEXT NOT NULL,
237 version TEXT NOT NULL,
238 content TEXT NOT NULL,
239 content_hash TEXT NOT NULL,
240 ts TEXT NOT NULL,
241 author TEXT,
242 source_tag TEXT NOT NULL,
243 origin_path TEXT
244);
245CREATE INDEX IF NOT EXISTS flow_revisions_by_name_id
246 ON flow_revisions (flow_name, id DESC);
247CREATE INDEX IF NOT EXISTS flow_revisions_by_hash
248 ON flow_revisions (flow_name, content_hash);
249"#;
250
251#[cfg(test)]
252mod tests {
253 use super::*;
254 use crate::flow_meta::{FlowMeta, FlowMetaSource};
255
256 fn mk_meta(version: &str, source: FlowMetaSource) -> FlowMeta {
257 FlowMeta {
258 version: version.to_string(),
259 description: None,
260 last_modified: None,
261 author: Some("w-mai".into()),
262 tags: Vec::new(),
263 source,
264 }
265 }
266
267 #[test]
268 fn snapshot_inserts_and_returns_row() {
269 let dir = tempfile::tempdir().unwrap();
270 let reg = FlowRegistry::open(dir.path()).unwrap();
271 let out = reg
272 .snapshot(
273 "greet",
274 "flow greet() { return 1 }",
275 &mk_meta("0.1.0", FlowMetaSource::Sidecar),
276 None,
277 )
278 .unwrap();
279 let SnapshotOutcome::Inserted(rev) = out else {
280 panic!("expected inserted, got {out:?}");
281 };
282 assert_eq!(rev.flow_name, "greet");
283 assert_eq!(rev.version, "0.1.0");
284 assert_eq!(rev.source_tag, "sidecar");
285 assert!(rev.origin_path.is_none());
286 assert_eq!(reg.count("greet").unwrap(), 1);
287 }
288
289 #[test]
290 fn snapshot_skips_when_latest_is_identical() {
291 let dir = tempfile::tempdir().unwrap();
292 let reg = FlowRegistry::open(dir.path()).unwrap();
293 let src = "flow greet() { return 1 }";
294 reg.snapshot(
295 "greet",
296 src,
297 &mk_meta("0.1.0", FlowMetaSource::Sidecar),
298 None,
299 )
300 .unwrap();
301 let again = reg
302 .snapshot(
303 "greet",
304 src,
305 &mk_meta("0.1.0", FlowMetaSource::Sidecar),
306 None,
307 )
308 .unwrap();
309 assert!(matches!(again, SnapshotOutcome::UnchangedFromLatest(_)));
310 assert_eq!(reg.count("greet").unwrap(), 1);
311 }
312
313 #[test]
314 fn snapshot_appends_when_content_changes() {
315 let dir = tempfile::tempdir().unwrap();
316 let reg = FlowRegistry::open(dir.path()).unwrap();
317 reg.snapshot(
318 "greet",
319 "flow greet() { return 1 }",
320 &mk_meta("0.1.0", FlowMetaSource::Sidecar),
321 None,
322 )
323 .unwrap();
324 reg.snapshot(
325 "greet",
326 "flow greet() { return 2 }",
327 &mk_meta("0.2.0", FlowMetaSource::Sidecar),
328 None,
329 )
330 .unwrap();
331 assert_eq!(reg.count("greet").unwrap(), 2);
332 let versions = reg.list_versions("greet").unwrap();
333 assert_eq!(versions.len(), 2);
334 assert_eq!(versions[0].version, "0.2.0");
335 assert_eq!(versions[1].version, "0.1.0");
336 }
337
338 #[test]
339 fn find_by_version_exact_match() {
340 let dir = tempfile::tempdir().unwrap();
341 let reg = FlowRegistry::open(dir.path()).unwrap();
342 reg.snapshot(
343 "greet",
344 "flow greet() { return 1 }",
345 &mk_meta("0.1.0", FlowMetaSource::Sidecar),
346 None,
347 )
348 .unwrap();
349 let hit = reg.find_by_version("greet", "0.1.0").unwrap().unwrap();
350 assert_eq!(hit.version, "0.1.0");
351 assert!(reg.find_by_version("greet", "9.9.9").unwrap().is_none());
352 }
353
354 #[test]
355 fn find_by_version_short_hash_prefix() {
356 let dir = tempfile::tempdir().unwrap();
357 let reg = FlowRegistry::open(dir.path()).unwrap();
358 let src = "flow greet() { return 1 }";
359 let hash = FlowMeta::short_hash(src);
360 reg.snapshot(
361 "greet",
362 src,
363 &mk_meta(&format!("hash:{hash}"), FlowMetaSource::HashFallback),
364 None,
365 )
366 .unwrap();
367 let short_prefix: String = hash.chars().take(6).collect();
368 let hit = reg
369 .find_by_version("greet", &short_prefix)
370 .unwrap()
371 .expect("prefix should match");
372 assert_eq!(hit.content_hash, hash);
373 let hit2 = reg
374 .find_by_version("greet", &format!("hash:{short_prefix}"))
375 .unwrap()
376 .expect("hash: prefix should also match");
377 assert_eq!(hit2.content_hash, hash);
378 }
379
380 #[test]
381 fn flow_names_returns_distinct_sorted() {
382 let dir = tempfile::tempdir().unwrap();
383 let reg = FlowRegistry::open(dir.path()).unwrap();
384 reg.snapshot(
385 "b_flow",
386 "flow b_flow() { return 1 }",
387 &mk_meta("1", FlowMetaSource::Sidecar),
388 None,
389 )
390 .unwrap();
391 reg.snapshot(
392 "a_flow",
393 "flow a_flow() { return 1 }",
394 &mk_meta("1", FlowMetaSource::Sidecar),
395 None,
396 )
397 .unwrap();
398 reg.snapshot(
399 "a_flow",
400 "flow a_flow() { return 2 }",
401 &mk_meta("2", FlowMetaSource::Sidecar),
402 None,
403 )
404 .unwrap();
405 let names = reg.flow_names().unwrap();
406 assert_eq!(names, vec!["a_flow", "b_flow"]);
407 }
408
409 #[test]
410 fn latest_returns_most_recent_revision() {
411 let dir = tempfile::tempdir().unwrap();
412 let reg = FlowRegistry::open(dir.path()).unwrap();
413 reg.snapshot("greet", "a", &mk_meta("0.1", FlowMetaSource::Sidecar), None)
414 .unwrap();
415 reg.snapshot("greet", "b", &mk_meta("0.2", FlowMetaSource::Sidecar), None)
416 .unwrap();
417 let latest = reg.latest("greet").unwrap().unwrap();
418 assert_eq!(latest.version, "0.2");
419 assert!(reg.latest("unknown").unwrap().is_none());
420 }
421
422 #[test]
423 fn source_tag_reflects_meta_source() {
424 let dir = tempfile::tempdir().unwrap();
425 let reg = FlowRegistry::open(dir.path()).unwrap();
426 reg.snapshot(
427 "a",
428 "x",
429 &mk_meta("hash:aaaa", FlowMetaSource::HashFallback),
430 None,
431 )
432 .unwrap();
433 reg.snapshot("b", "y", &mk_meta("1.0", FlowMetaSource::Sidecar), None)
434 .unwrap();
435 assert_eq!(reg.latest("a").unwrap().unwrap().source_tag, "hash");
436 assert_eq!(reg.latest("b").unwrap().unwrap().source_tag, "sidecar");
437 }
438
439 #[test]
440 fn snapshot_records_origin_path() {
441 let dir = tempfile::tempdir().unwrap();
442 let reg = FlowRegistry::open(dir.path()).unwrap();
443 let out = reg
444 .snapshot(
445 "greet",
446 "flow greet() { return 1 }",
447 &mk_meta("0.1.0", FlowMetaSource::Sidecar),
448 Some(Path::new("/tmp/greet.at")),
449 )
450 .unwrap();
451 let SnapshotOutcome::Inserted(rev) = out else {
452 panic!("expected inserted");
453 };
454 assert_eq!(rev.origin_path.as_deref(), Some("/tmp/greet.at"));
455 let latest = reg.latest("greet").unwrap().unwrap();
456 assert_eq!(latest.origin_path.as_deref(), Some("/tmp/greet.at"));
457 }
458
459 #[test]
460 fn unchanged_snapshot_backfills_missing_origin_path() {
461 let dir = tempfile::tempdir().unwrap();
462 let reg = FlowRegistry::open(dir.path()).unwrap();
463 let src = "flow greet() { return 1 }";
464 reg.snapshot(
465 "greet",
466 src,
467 &mk_meta("0.1.0", FlowMetaSource::Sidecar),
468 None,
469 )
470 .unwrap();
471 let outcome = reg
472 .snapshot(
473 "greet",
474 src,
475 &mk_meta("0.1.0", FlowMetaSource::Sidecar),
476 Some(Path::new("/tmp/greet.at")),
477 )
478 .unwrap();
479 let SnapshotOutcome::UnchangedFromLatest(rev) = outcome else {
480 panic!("expected UnchangedFromLatest");
481 };
482 assert_eq!(rev.origin_path.as_deref(), Some("/tmp/greet.at"));
483 let latest = reg.latest("greet").unwrap().unwrap();
484 assert_eq!(latest.origin_path.as_deref(), Some("/tmp/greet.at"));
485 assert_eq!(reg.count("greet").unwrap(), 1);
486 }
487
488 #[test]
489 fn open_migrates_legacy_registry_without_origin_path_column() {
490 let dir = tempfile::tempdir().unwrap();
491 let db_path = dir.path().join(".atman").join("flow-registry.db");
492 std::fs::create_dir_all(db_path.parent().unwrap()).unwrap();
493 {
494 let conn = Connection::open(&db_path).unwrap();
495 conn.execute_batch(
496 r#"
497 CREATE TABLE flow_revisions (
498 id INTEGER PRIMARY KEY AUTOINCREMENT,
499 flow_name TEXT NOT NULL,
500 version TEXT NOT NULL,
501 content TEXT NOT NULL,
502 content_hash TEXT NOT NULL,
503 ts TEXT NOT NULL,
504 author TEXT,
505 source_tag TEXT NOT NULL
506 );
507 "#,
508 )
509 .unwrap();
510 conn.execute(
511 "INSERT INTO flow_revisions \
512 (flow_name, version, content, content_hash, ts, author, source_tag) \
513 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
514 params![
515 "legacy",
516 "0.1",
517 "flow legacy() { return 1 }",
518 "abcd",
519 "2026-07-05T00:00:00Z",
520 Option::<String>::None,
521 "sidecar",
522 ],
523 )
524 .unwrap();
525 }
526 let reg = FlowRegistry::open(dir.path()).unwrap();
527 let latest = reg.latest("legacy").unwrap().unwrap();
528 assert_eq!(latest.origin_path, None);
529 assert_eq!(latest.version, "0.1");
530 }
531}