1use chrono::{DateTime, Utc};
2use sqlx::PgPool;
3use uuid::Uuid;
4
5use dk_core::{RepoId, SymbolId};
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub enum ChangesetState {
10 Draft,
11 Submitted,
12 Verifying,
13 Approved,
14 Rejected,
15 Merged,
16 Closed,
17}
18
19impl ChangesetState {
20 pub fn parse(s: &str) -> Option<Self> {
22 match s {
23 "draft" => Some(Self::Draft),
24 "submitted" => Some(Self::Submitted),
25 "verifying" => Some(Self::Verifying),
26 "approved" => Some(Self::Approved),
27 "rejected" => Some(Self::Rejected),
28 "merged" => Some(Self::Merged),
29 "closed" => Some(Self::Closed),
30 _ => None,
31 }
32 }
33
34 pub fn as_str(&self) -> &'static str {
36 match self {
37 Self::Draft => "draft",
38 Self::Submitted => "submitted",
39 Self::Verifying => "verifying",
40 Self::Approved => "approved",
41 Self::Rejected => "rejected",
42 Self::Merged => "merged",
43 Self::Closed => "closed",
44 }
45 }
46
47 pub fn can_transition_to(&self, target: Self) -> bool {
56 if target == Self::Closed {
57 return true;
58 }
59 matches!(
60 (self, target),
61 (Self::Draft, Self::Submitted)
62 | (Self::Submitted, Self::Verifying)
63 | (Self::Verifying, Self::Approved)
64 | (Self::Verifying, Self::Rejected)
65 | (Self::Approved, Self::Merged)
66 )
67 }
68
69 pub fn is_terminal(&self) -> bool {
79 matches!(self, Self::Draft | Self::Merged | Self::Rejected | Self::Closed)
80 }
81}
82
83impl std::fmt::Display for ChangesetState {
84 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
85 f.write_str(self.as_str())
86 }
87}
88
89#[derive(Debug, Clone, sqlx::FromRow)]
90pub struct Changeset {
91 pub id: Uuid,
92 pub repo_id: RepoId,
93 pub number: i32,
94 pub title: String,
95 pub intent_summary: Option<String>,
96 pub source_branch: String,
97 pub target_branch: String,
98 pub state: String,
99 pub reason: String,
100 pub session_id: Option<Uuid>,
101 pub agent_id: Option<String>,
102 pub agent_name: Option<String>,
103 pub author_id: Option<Uuid>,
104 pub base_version: Option<String>,
105 pub merged_version: Option<String>,
106 pub created_at: DateTime<Utc>,
107 pub updated_at: DateTime<Utc>,
108 pub merged_at: Option<DateTime<Utc>>,
109 #[sqlx(default)]
120 pub parent_changeset_id: Option<Uuid>,
121}
122
123impl Changeset {
124 pub fn parsed_state(&self) -> Option<ChangesetState> {
126 ChangesetState::parse(&self.state)
127 }
128
129 pub fn transition(
132 &mut self,
133 target: ChangesetState,
134 reason: impl Into<String>,
135 ) -> dk_core::Result<()> {
136 let current = self.parsed_state().ok_or_else(|| {
137 dk_core::Error::Internal(format!("unknown current state: '{}'", self.state))
138 })?;
139
140 if !current.can_transition_to(target) {
141 return Err(dk_core::Error::InvalidInput(format!(
142 "invalid state transition: '{}' -> '{}'",
143 current, target,
144 )));
145 }
146
147 self.state = target.as_str().to_string();
148 self.reason = reason.into();
149 Ok(())
150 }
151}
152
153#[derive(Debug, Clone)]
154pub struct ChangesetFile {
155 pub changeset_id: Uuid,
156 pub file_path: String,
157 pub operation: String,
158 pub content: Option<String>,
159}
160
161#[derive(Debug, Clone)]
162pub struct ChangesetFileMeta {
163 pub file_path: String,
164 pub operation: String,
165 pub size_bytes: i64,
166}
167
168pub struct ChangesetStore {
169 db: PgPool,
170}
171
172impl ChangesetStore {
173 pub fn new(db: PgPool) -> Self {
174 Self { db }
175 }
176
177 pub async fn create(
183 &self,
184 repo_id: RepoId,
185 session_id: Option<Uuid>,
186 agent_id: &str,
187 intent: &str,
188 base_version: Option<&str>,
189 agent_name: &str,
190 ) -> dk_core::Result<Changeset> {
191 let intent_slug: String = intent
192 .to_lowercase()
193 .chars()
194 .map(|c| if c.is_alphanumeric() || c == '-' { c } else { '-' })
195 .collect::<String>()
196 .trim_matches('-')
197 .to_string();
198 let slug = if intent_slug.len() > 50 {
199 let cut = intent_slug
200 .char_indices()
201 .take_while(|(i, _)| *i < 50)
202 .last()
203 .map(|(i, c)| i + c.len_utf8())
204 .unwrap_or(0);
205 intent_slug[..cut].trim_end_matches('-').to_string()
206 } else {
207 intent_slug
208 };
209 let source_branch = format!("{}/{}", slug, agent_name);
210 let target_branch = "main";
211
212 let mut tx = self.db.begin().await?;
213
214 sqlx::query("SELECT pg_advisory_xact_lock(hashtext('changeset:' || $1::text))")
215 .bind(repo_id)
216 .execute(&mut *tx)
217 .await?;
218
219 let row: (Uuid, i32, String, String, DateTime<Utc>, DateTime<Utc>) = sqlx::query_as(
220 r#"INSERT INTO changesets
221 (repo_id, number, title, intent_summary, source_branch, target_branch,
222 state, reason, session_id, agent_id, agent_name, base_version)
223 SELECT $1, COALESCE(MAX(number), 0) + 1, $2, $2, $3, $4,
224 'draft', 'created via agent connect', $5, $6, $7, $8
225 FROM changesets WHERE repo_id = $1
226 RETURNING id, number, state, reason, created_at, updated_at"#,
227 )
228 .bind(repo_id)
229 .bind(intent)
230 .bind(&source_branch)
231 .bind(target_branch)
232 .bind(session_id)
233 .bind(agent_id)
234 .bind(agent_name)
235 .bind(base_version)
236 .fetch_one(&mut *tx)
237 .await?;
238
239 tx.commit().await?;
240
241 Ok(Changeset {
242 id: row.0,
243 repo_id,
244 number: row.1,
245 title: intent.to_string(),
246 intent_summary: Some(intent.to_string()),
247 source_branch,
248 target_branch: target_branch.to_string(),
249 state: row.2,
250 reason: row.3,
251 session_id,
252 agent_id: Some(agent_id.to_string()),
253 agent_name: Some(agent_name.to_string()),
254 author_id: None,
255 base_version: base_version.map(String::from),
256 merged_version: None,
257 created_at: row.4,
258 updated_at: row.5,
259 merged_at: None,
260 parent_changeset_id: None,
261 })
262 }
263
264 pub async fn get(&self, id: Uuid) -> dk_core::Result<Changeset> {
265 sqlx::query_as::<_, Changeset>(
268 r#"SELECT id, repo_id, number, title, intent_summary,
269 source_branch, target_branch, state, reason,
270 session_id, agent_id, agent_name, author_id,
271 base_version, merged_version,
272 created_at, updated_at, merged_at
273 FROM changesets WHERE id = $1"#,
274 )
275 .bind(id)
276 .fetch_optional(&self.db)
277 .await?
278 .ok_or_else(|| dk_core::Error::Internal(format!("changeset {} not found", id)))
279 }
280
281 pub async fn update_status(&self, id: Uuid, status: &str) -> dk_core::Result<()> {
282 self.update_status_with_reason(id, status, "").await
283 }
284
285 pub async fn update_status_with_reason(
287 &self,
288 id: Uuid,
289 status: &str,
290 reason: &str,
291 ) -> dk_core::Result<()> {
292 sqlx::query(
293 "UPDATE changesets SET state = $1, reason = $2, updated_at = now() WHERE id = $3",
294 )
295 .bind(status)
296 .bind(reason)
297 .bind(id)
298 .execute(&self.db)
299 .await?;
300 Ok(())
301 }
302
303 pub async fn update_status_if(
307 &self,
308 id: Uuid,
309 new_status: &str,
310 expected_states: &[&str],
311 ) -> dk_core::Result<()> {
312 self.update_status_if_with_reason(id, new_status, expected_states, "").await
313 }
314
315 pub async fn update_status_if_with_reason(
317 &self,
318 id: Uuid,
319 new_status: &str,
320 expected_states: &[&str],
321 reason: &str,
322 ) -> dk_core::Result<()> {
323 let states: Vec<String> = expected_states.iter().map(|s| s.to_string()).collect();
324 let result = sqlx::query(
325 "UPDATE changesets SET state = $1, reason = $2, updated_at = now() WHERE id = $3 AND state = ANY($4)",
326 )
327 .bind(new_status)
328 .bind(reason)
329 .bind(id)
330 .bind(&states)
331 .execute(&self.db)
332 .await?;
333
334 if result.rows_affected() == 0 {
335 return Err(dk_core::Error::Internal(format!(
336 "changeset {} not found or not in expected state (expected one of: {:?})",
337 id, expected_states,
338 )));
339 }
340 Ok(())
341 }
342
343 pub async fn set_merged(&self, id: Uuid, commit_hash: &str) -> dk_core::Result<()> {
344 sqlx::query(
345 "UPDATE changesets SET state = 'merged', reason = 'merge completed', merged_version = $1, merged_at = now(), updated_at = now() WHERE id = $2",
346 )
347 .bind(commit_hash)
348 .bind(id)
349 .execute(&self.db)
350 .await?;
351 Ok(())
352 }
353
354 pub async fn upsert_file(
355 &self,
356 changeset_id: Uuid,
357 file_path: &str,
358 operation: &str,
359 content: Option<&str>,
360 ) -> dk_core::Result<()> {
361 sqlx::query(
362 r#"INSERT INTO changeset_files (changeset_id, file_path, operation, content)
363 VALUES ($1, $2, $3, $4)
364 ON CONFLICT (changeset_id, file_path) DO UPDATE SET
365 operation = EXCLUDED.operation,
366 content = EXCLUDED.content"#,
367 )
368 .bind(changeset_id)
369 .bind(file_path)
370 .bind(operation)
371 .bind(content)
372 .execute(&self.db)
373 .await?;
374 Ok(())
375 }
376
377 pub async fn get_files(&self, changeset_id: Uuid) -> dk_core::Result<Vec<ChangesetFile>> {
378 let rows: Vec<(Uuid, String, String, Option<String>)> = sqlx::query_as(
379 "SELECT changeset_id, file_path, operation, content FROM changeset_files WHERE changeset_id = $1",
380 )
381 .bind(changeset_id)
382 .fetch_all(&self.db)
383 .await?;
384
385 Ok(rows
386 .into_iter()
387 .map(|r| ChangesetFile {
388 changeset_id: r.0,
389 file_path: r.1,
390 operation: r.2,
391 content: r.3,
392 })
393 .collect())
394 }
395
396 pub async fn get_files_metadata(&self, changeset_id: Uuid) -> dk_core::Result<Vec<ChangesetFileMeta>> {
399 let rows: Vec<(String, String, i64)> = sqlx::query_as(
400 "SELECT file_path, operation, COALESCE(LENGTH(content), 0)::bigint AS size_bytes FROM changeset_files WHERE changeset_id = $1",
401 )
402 .bind(changeset_id)
403 .fetch_all(&self.db)
404 .await?;
405
406 Ok(rows
407 .into_iter()
408 .map(|r| ChangesetFileMeta {
409 file_path: r.0,
410 operation: r.1,
411 size_bytes: r.2,
412 })
413 .collect())
414 }
415
416 pub async fn record_affected_symbol(
417 &self,
418 changeset_id: Uuid,
419 symbol_id: SymbolId,
420 qualified_name: &str,
421 ) -> dk_core::Result<()> {
422 sqlx::query(
423 r#"INSERT INTO changeset_symbols (changeset_id, symbol_id, symbol_qualified_name)
424 VALUES ($1, $2, $3)
425 ON CONFLICT DO NOTHING"#,
426 )
427 .bind(changeset_id)
428 .bind(symbol_id)
429 .bind(qualified_name)
430 .execute(&self.db)
431 .await?;
432 Ok(())
433 }
434
435 pub async fn get_affected_symbols(&self, changeset_id: Uuid) -> dk_core::Result<Vec<(SymbolId, String)>> {
436 let rows: Vec<(Uuid, String)> = sqlx::query_as(
437 "SELECT symbol_id, symbol_qualified_name FROM changeset_symbols WHERE changeset_id = $1",
438 )
439 .bind(changeset_id)
440 .fetch_all(&self.db)
441 .await?;
442 Ok(rows)
443 }
444
445 pub async fn list_path_competitors(
457 &self,
458 repo_id: RepoId,
459 path: &str,
460 ) -> dk_core::Result<Vec<(Uuid, Option<Uuid>, String, DateTime<Utc>)>> {
461 let rows: Vec<(Uuid, Option<Uuid>, String, DateTime<Utc>)> = sqlx::query_as(
462 r#"SELECT c.id, c.session_id, c.state, c.updated_at
463 FROM changesets c
464 JOIN changeset_files cf ON cf.changeset_id = c.id
465 WHERE c.repo_id = $1
466 AND cf.file_path = $2
467 AND c.state IN ('submitted', 'verifying', 'approved')"#,
468 )
469 .bind(repo_id)
470 .bind(path)
471 .fetch_all(&self.db)
472 .await?;
473 Ok(rows)
474 }
475
476 pub async fn find_conflicting_changesets(
480 &self,
481 repo_id: RepoId,
482 base_version: &str,
483 my_changeset_id: Uuid,
484 ) -> dk_core::Result<Vec<(Uuid, Vec<String>)>> {
485 let rows: Vec<(Uuid, String)> = sqlx::query_as(
486 r#"SELECT DISTINCT cs.changeset_id, cs.symbol_qualified_name
487 FROM changeset_symbols cs
488 JOIN changesets c ON c.id = cs.changeset_id
489 WHERE c.repo_id = $1
490 AND c.state = 'merged'
491 AND c.id != $2
492 AND c.merged_version IS NOT NULL
493 AND c.merged_version != $3
494 AND cs.symbol_qualified_name IN (
495 SELECT symbol_qualified_name FROM changeset_symbols WHERE changeset_id = $2
496 )"#,
497 )
498 .bind(repo_id)
499 .bind(my_changeset_id)
500 .bind(base_version)
501 .fetch_all(&self.db)
502 .await?;
503
504 let mut map: std::collections::HashMap<Uuid, Vec<String>> = std::collections::HashMap::new();
505 for (cs_id, sym_name) in rows {
506 map.entry(cs_id).or_default().push(sym_name);
507 }
508 Ok(map.into_iter().collect())
509 }
510}
511
512#[cfg(test)]
513mod tests {
514 use super::*;
515
516 fn slugify_intent(intent: &str) -> String {
519 let slug: String = intent
520 .to_lowercase()
521 .chars()
522 .map(|c| if c.is_alphanumeric() || c == '-' { c } else { '-' })
523 .collect::<String>()
524 .trim_matches('-')
525 .to_string();
526 if slug.len() > 50 {
527 slug[..50].trim_end_matches('-').to_string()
528 } else {
529 slug
530 }
531 }
532
533 #[test]
534 fn source_branch_format_uses_intent_slug() {
535 let intent = "Fix UI bugs";
536 let agent_name = "agent-1";
537 let source_branch = format!("{}/{}", slugify_intent(intent), agent_name);
538 assert_eq!(source_branch, "fix-ui-bugs/agent-1");
539 }
540
541 #[test]
542 fn source_branch_format_with_custom_name() {
543 let intent = "Add comments endpoint";
544 let agent_name = "feature-bot";
545 let source_branch = format!("{}/{}", slugify_intent(intent), agent_name);
546 assert_eq!(source_branch, "add-comments-endpoint/feature-bot");
547 }
548
549 #[test]
550 fn target_branch_is_main() {
551 let target_branch = "main";
553 assert_eq!(target_branch, "main");
554 }
555
556 fn test_changeset_fixture() -> Changeset {
563 let now = Utc::now();
564 Changeset {
565 id: Uuid::new_v4(),
566 repo_id: Uuid::new_v4(),
567 number: 1,
568 title: String::new(),
569 intent_summary: None,
570 source_branch: "agent/test".to_string(),
571 target_branch: "main".to_string(),
572 state: "draft".to_string(),
573 reason: String::new(),
574 session_id: None,
575 agent_id: None,
576 agent_name: None,
577 author_id: None,
578 base_version: None,
579 merged_version: None,
580 created_at: now,
581 updated_at: now,
582 merged_at: None,
583 parent_changeset_id: None,
584 }
585 }
586
587 #[test]
590 fn changeset_create_shape_has_correct_branches() {
591 let repo_id = Uuid::new_v4();
592 let session_id = Uuid::new_v4();
593 let agent_id = "test-agent";
594 let intent = "fix all the bugs";
595
596 let source_branch = format!("agent/{}", agent_id);
597
598 let cs = Changeset {
599 repo_id,
600 title: intent.to_string(),
601 intent_summary: Some(intent.to_string()),
602 source_branch: source_branch.clone(),
603 session_id: Some(session_id),
604 agent_id: Some(agent_id.to_string()),
605 agent_name: Some(agent_id.to_string()),
606 base_version: Some("abc123".to_string()),
607 ..test_changeset_fixture()
608 };
609
610 assert_eq!(cs.source_branch, "agent/test-agent");
611 assert_eq!(cs.target_branch, "main");
612 assert_eq!(cs.agent_name.as_deref(), Some("test-agent"));
613 assert_eq!(cs.agent_id, cs.agent_name, "agent_name should equal agent_id per create()");
614 assert!(cs.merged_at.is_none());
615 assert!(cs.merged_version.is_none());
616 }
617
618 #[test]
621 fn changeset_all_fields_accessible() {
622 let id = Uuid::new_v4();
623 let repo_id = Uuid::new_v4();
624
625 let cs = Changeset {
626 id,
627 repo_id,
628 number: 42,
629 title: "test".to_string(),
630 source_branch: "agent/a".to_string(),
631 ..test_changeset_fixture()
632 };
633
634 assert_eq!(cs.id, id);
635 assert_eq!(cs.repo_id, repo_id);
636 assert_eq!(cs.number, 42);
637 assert_eq!(cs.title, "test");
638 assert!(cs.intent_summary.is_none());
639 assert!(cs.session_id.is_none());
640 assert!(cs.agent_id.is_none());
641 assert!(cs.agent_name.is_none());
642 assert!(cs.author_id.is_none());
643 assert!(cs.base_version.is_none());
644 assert!(cs.merged_version.is_none());
645 assert!(cs.merged_at.is_none());
646 }
647
648 #[test]
649 fn changeset_file_meta_struct() {
650 let meta = ChangesetFileMeta {
651 file_path: "src/main.rs".to_string(),
652 operation: "modify".to_string(),
653 size_bytes: 1024,
654 };
655 assert_eq!(meta.file_path, "src/main.rs");
656 assert_eq!(meta.operation, "modify");
657 assert_eq!(meta.size_bytes, 1024);
658 }
659
660 #[test]
661 fn changeset_file_struct() {
662 let cf = ChangesetFile {
663 changeset_id: Uuid::new_v4(),
664 file_path: "lib.rs".to_string(),
665 operation: "add".to_string(),
666 content: Some("fn main() {}".to_string()),
667 };
668 assert_eq!(cf.file_path, "lib.rs");
669 assert_eq!(cf.operation, "add");
670 assert!(cf.content.is_some());
671 }
672
673 #[test]
674 fn changeset_clone_produces_equal_values() {
675 let cs = Changeset {
676 title: "clone test".to_string(),
677 intent_summary: Some("intent".to_string()),
678 source_branch: "agent/x".to_string(),
679 agent_id: Some("x".to_string()),
680 agent_name: Some("x".to_string()),
681 ..test_changeset_fixture()
682 };
683
684 let cloned = cs.clone();
685 assert_eq!(cs.id, cloned.id);
686 assert_eq!(cs.source_branch, cloned.source_branch);
687 assert_eq!(cs.target_branch, cloned.target_branch);
688 assert_eq!(cs.state, cloned.state);
689 }
690
691 #[test]
692 fn is_terminal_partitions_changeset_states() {
693 assert!(!ChangesetState::Submitted.is_terminal());
694 assert!(!ChangesetState::Verifying.is_terminal());
695 assert!(!ChangesetState::Approved.is_terminal());
696 assert!(ChangesetState::Merged.is_terminal());
697 assert!(ChangesetState::Rejected.is_terminal());
698 assert!(ChangesetState::Closed.is_terminal());
699 assert!(ChangesetState::Draft.is_terminal()); }
701}