systemprompt_sync/local/
content_sync.rs

1//! Two-way sync between content stored on disk (markdown + frontmatter) and
2//! the database via the `systemprompt-content` ingestion + repository layers.
3//!
4//! Copyright (c) systemprompt.io — Business Source License 1.1.
5//! See <https://systemprompt.io> for licensing details.
6
7use crate::diff::ContentDiffCalculator;
8use crate::error::{SyncError, SyncResult};
9use crate::export::export_content_to_file;
10use crate::models::{ContentDiffResult, LocalSyncDirection, LocalSyncResult};
11use std::path::{Path, PathBuf};
12use systemprompt_content::models::{IngestionOptions, IngestionSource};
13use systemprompt_content::repository::ContentRepository;
14use systemprompt_content::services::IngestionService;
15use systemprompt_database::DbPool;
16use systemprompt_identifiers::{CategoryId, ContentId, LocaleCode, SourceId};
17use tracing::info;
18
19#[derive(Debug)]
20pub struct ContentDiffEntry {
21    pub name: String,
22    pub source_id: SourceId,
23    pub category_id: CategoryId,
24    pub path: PathBuf,
25    pub allowed_content_types: Vec<String>,
26    pub diff: ContentDiffResult,
27}
28
29#[derive(Debug)]
30pub struct ContentLocalSync {
31    db: DbPool,
32}
33
34impl ContentLocalSync {
35    pub const fn new(db: DbPool) -> Self {
36        Self { db }
37    }
38
39    pub async fn calculate_diff(
40        &self,
41        source_id: &SourceId,
42        disk_path: &Path,
43        allowed_types: &[String],
44    ) -> SyncResult<ContentDiffResult> {
45        let calculator = ContentDiffCalculator::new(&self.db).map_err(SyncError::internal)?;
46        calculator
47            .calculate_diff(source_id, disk_path, allowed_types)
48            .await
49            .map_err(SyncError::internal)
50    }
51
52    pub async fn sync_to_disk(
53        &self,
54        diffs: &[ContentDiffEntry],
55        delete_orphans: bool,
56    ) -> SyncResult<LocalSyncResult> {
57        let content_repo = ContentRepository::new(&self.db).map_err(SyncError::internal)?;
58        let mut result = LocalSyncResult {
59            direction: LocalSyncDirection::ToDisk,
60            ..Default::default()
61        };
62
63        for entry in diffs {
64            let source_path = &entry.path;
65
66            for item in &entry.diff.modified {
67                match content_repo
68                    .get_by_source_and_slug(&entry.source_id, &item.slug, &LocaleCode::new("en"))
69                    .await
70                    .map_err(SyncError::internal)?
71                {
72                    Some(content) => {
73                        export_content_to_file(&content, source_path, &entry.name)?;
74                        result.items_synced += 1;
75                        info!("Exported modified content: {}", item.slug);
76                    },
77                    None => {
78                        result
79                            .errors
80                            .push(format!("Content not found in DB: {}", item.slug));
81                    },
82                }
83            }
84
85            for item in &entry.diff.removed {
86                match content_repo
87                    .get_by_source_and_slug(&entry.source_id, &item.slug, &LocaleCode::new("en"))
88                    .await
89                    .map_err(SyncError::internal)?
90                {
91                    Some(content) => {
92                        export_content_to_file(&content, source_path, &entry.name)?;
93                        result.items_synced += 1;
94                        info!("Created content on disk: {}", item.slug);
95                    },
96                    None => {
97                        result
98                            .errors
99                            .push(format!("Content not found in DB: {}", item.slug));
100                    },
101                }
102            }
103
104            if delete_orphans {
105                for item in &entry.diff.added {
106                    let file_path = if entry.name == "blog" {
107                        source_path.join(&item.slug).join("index.md")
108                    } else {
109                        source_path.join(format!("{}.md", item.slug))
110                    };
111
112                    if file_path.exists() {
113                        if entry.name == "blog" {
114                            std::fs::remove_dir_all(source_path.join(&item.slug))?;
115                        } else {
116                            std::fs::remove_file(&file_path)?;
117                        }
118                        result.items_deleted += 1;
119                        info!("Deleted orphan content: {}", item.slug);
120                    }
121                }
122            } else {
123                result.items_skipped += entry.diff.added.len();
124            }
125        }
126
127        Ok(result)
128    }
129
130    pub async fn sync_to_db(
131        &self,
132        diffs: &[ContentDiffEntry],
133        delete_orphans: bool,
134        override_existing: bool,
135    ) -> SyncResult<LocalSyncResult> {
136        let ingestion_service = IngestionService::new(&self.db).map_err(SyncError::internal)?;
137        let content_repo = ContentRepository::new(&self.db).map_err(SyncError::internal)?;
138        let mut result = LocalSyncResult {
139            direction: LocalSyncDirection::ToDatabase,
140            ..Default::default()
141        };
142
143        for entry in diffs {
144            let source_path = &entry.path;
145            let source = IngestionSource::new(&entry.source_id, &entry.name, &entry.category_id);
146            let report = ingestion_service
147                .ingest_directory(
148                    source_path,
149                    &source,
150                    IngestionOptions::default()
151                        .with_recursive(true)
152                        .with_override(override_existing),
153                )
154                .await
155                .map_err(SyncError::internal)?;
156
157            result.items_synced += report.files_processed;
158            result.items_skipped_modified += report.skipped_count;
159
160            for error in report.errors {
161                result.errors.push(error);
162            }
163
164            info!(
165                "Ingested {} files from {}",
166                report.files_processed, entry.name
167            );
168
169            if delete_orphans {
170                for item in &entry.diff.removed {
171                    let content_id = ContentId::new(&item.slug);
172                    content_repo.delete(&content_id).await.map_err(|e| {
173                        SyncError::internal(format!("Failed to delete {}: {e}", item.slug))
174                    })?;
175                    result.items_deleted += 1;
176                    info!("Deleted from DB: {}", item.slug);
177                }
178            } else {
179                result.items_skipped += entry.diff.removed.len();
180            }
181        }
182
183        Ok(result)
184    }
185}
systemprompt_sync/local/content_sync.rs

systemprompt_sync/local/
content_sync.rs
