Merge pull request #92 from asiniscalchi/refactor-froid

Refactor froid

Author: asiniscalchi

.gitignore

@@ -3,3 +3,4 @@
 *.sqlite3
 data/
 froid.wiki
+.antigravitycli/

src/app.rs

@@ -44,7 +44,7 @@ use crate::{
     prompts::{PromptKey, PromptRepository, PromptSource},
     version,
     workers::{
-        ReconciliationWorker,
+        ReconciliationWorker, ReconciliationWorkerConfig,
         daily_review::{DailyReviewDeliveryWorker, TelegramDailyReviewSender},
         embedding::EmbeddingCycle,
         extraction::ExtractionCycle,
@@ -440,7 +440,7 @@ fn spawn_daily_review_delivery_worker(
         return Ok(false);
     };
 
-    let worker = DailyReviewDeliveryWorker::new(
+    let cycle = DailyReviewDeliveryWorker::new(
         JournalRepository::new(pool.clone()),
         crate::journal::review::repository::DailyReviewRepository::new(pool.clone()),
         daily_review_service,
@@ -450,6 +450,12 @@ fn spawn_daily_review_delivery_worker(
         ),
         config.daily_review_delivery.clone(),
     );
+    let worker_config = ReconciliationWorkerConfig {
+        enabled: config.daily_review_delivery.enabled,
+        batch_size: 1,
+        interval: config.daily_review_delivery.interval,
+    };
+    let worker = ReconciliationWorker::new(cycle, worker_config);
     let token = shutdown.clone();
     workers.spawn(async move {
         worker.run_forever(token).await;
@@ -477,7 +483,7 @@ fn spawn_weekly_review_delivery_worker(
         return Ok(());
     };
 
-    let worker = WeeklyReviewDeliveryWorker::new(
+    let cycle = WeeklyReviewDeliveryWorker::new(
         JournalRepository::new(pool.clone()),
         crate::journal::week_review::repository::WeeklyReviewRepository::new(pool.clone()),
         weekly_review_service,
@@ -487,6 +493,12 @@ fn spawn_weekly_review_delivery_worker(
         ),
         config.weekly_review_delivery.clone(),
     );
+    let worker_config = ReconciliationWorkerConfig {
+        enabled: config.weekly_review_delivery.enabled,
+        batch_size: 1,
+        interval: config.weekly_review_delivery.interval,
+    };
+    let worker = ReconciliationWorker::new(cycle, worker_config);
     let token = shutdown.clone();
     workers.spawn(async move {
         worker.run_forever(token).await;

src/dashboard/api.rs

@@ -51,14 +51,91 @@ struct ExportEnvelope {
     messages: Vec<ExportedMessage>,
 }
 
-async fn export_messages(State(repo): State<JournalRepository>) -> Response {
-    let records = match repo.fetch_all_for_export().await {
-        Ok(records) => records,
-        Err(err) => {
-            error!(error = %err, "failed to fetch journal entries for export");
-            return (StatusCode::INTERNAL_SERVER_ERROR, "failed to load messages").into_response();
+#[derive(Debug)]
+enum DashboardError {
+    ExportRepository(sqlx::Error),
+    Serialization(serde_json::Error),
+    UnsupportedVersion {
+        version: u32,
+    },
+    ImportConflict {
+        source: String,
+        source_conversation_id: String,
+        source_message_id: String,
+    },
+    ImportRepository(sqlx::Error),
+}
+
+impl IntoResponse for DashboardError {
+    fn into_response(self) -> Response {
+        match self {
+            Self::ExportRepository(err) => {
+                error!(error = %err, "failed to fetch journal entries for export");
+                (StatusCode::INTERNAL_SERVER_ERROR, "failed to load messages").into_response()
+            }
+            Self::Serialization(err) => {
+                error!(error = %err, "failed to serialize journal entries for export");
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    "failed to serialize messages",
+                )
+                    .into_response()
+            }
+            Self::UnsupportedVersion { version } => {
+                (
+                    StatusCode::BAD_REQUEST,
+                    Json(ImportError {
+                        error: format!(
+                            "unsupported export version {} (supported {}..={})",
+                            version, MIN_SUPPORTED_IMPORT_VERSION, EXPORT_FORMAT_VERSION
+                        ),
+                        conflict: None,
+                    }),
+                )
+                    .into_response()
+            }
+            Self::ImportConflict {
+                source,
+                source_conversation_id,
+                source_message_id,
+            } => {
+                (
+                    StatusCode::CONFLICT,
+                    Json(ImportError {
+                        error: format!(
+                            "import aborted: entry ({source}, {source_conversation_id}, {source_message_id}) collides with an existing message"
+                        ),
+                        conflict: Some(ConflictDetails {
+                            source: source.clone(),
+                            source_conversation_id: source_conversation_id.clone(),
+                            source_message_id: source_message_id.clone(),
+                        }),
+                    }),
+                )
+                    .into_response()
+            }
+            Self::ImportRepository(err) => {
+                error!(error = %err, "failed to import journal entries");
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(ImportError {
+                        error: "failed to import messages".to_string(),
+                        conflict: None,
+                    }),
+                )
+                    .into_response()
+            }
         }
-    };
+    }
+}
+
+async fn export_messages(
+    State(repo): State<JournalRepository>,
+) -> Result<impl IntoResponse, DashboardError> {
+    let records = repo
+        .fetch_all_for_export()
+        .await
+        .map_err(DashboardError::ExportRepository)?;
 
     let envelope = ExportEnvelope {
         version: EXPORT_FORMAT_VERSION,
@@ -76,24 +153,14 @@ async fn export_messages(State(repo): State<JournalRepository>) -> Response {
             .collect(),
     };
 
-    let body = match serde_json::to_vec(&envelope) {
-        Ok(body) => body,
-        Err(err) => {
-            error!(error = %err, "failed to serialize journal entries for export");
-            return (
-                StatusCode::INTERNAL_SERVER_ERROR,
-                "failed to serialize messages",
-            )
-                .into_response();
-        }
-    };
+    let body = serde_json::to_vec(&envelope).map_err(DashboardError::Serialization)?;
 
     let filename = format!(
         "froid-messages-{}.json",
         envelope.exported_at.format("%Y-%m-%d")
     );
 
-    (
+    Ok((
         StatusCode::OK,
         [
             (header::CONTENT_TYPE, "application/json".to_string()),
@@ -103,8 +170,7 @@ async fn export_messages(State(repo): State<JournalRepository>) -> Response {
             ),
         ],
         body,
-    )
-        .into_response()
+    ))
 }
 
 #[derive(Deserialize)]
@@ -149,23 +215,15 @@ struct ImportError {
 async fn import_messages(
     State(repo): State<JournalRepository>,
     Json(envelope): Json<ImportEnvelope>,
-) -> Response {
+) -> Result<impl IntoResponse, DashboardError> {
     if envelope.version < MIN_SUPPORTED_IMPORT_VERSION || envelope.version > EXPORT_FORMAT_VERSION {
-        return (
-            StatusCode::BAD_REQUEST,
-            Json(ImportError {
-                error: format!(
-                    "unsupported export version {} (supported {}..={})",
-                    envelope.version, MIN_SUPPORTED_IMPORT_VERSION, EXPORT_FORMAT_VERSION
-                ),
-                conflict: None,
-            }),
-        )
-            .into_response();
+        return Err(DashboardError::UnsupportedVersion {
+            version: envelope.version,
+        });
     }
 
     if envelope.messages.is_empty() {
-        return (StatusCode::OK, Json(ImportResult { imported: 0 })).into_response();
+        return Ok((StatusCode::OK, Json(ImportResult { imported: 0 })).into_response());
     }
 
     let records: Vec<JournalEntryRecord> = envelope
@@ -181,38 +239,20 @@ async fn import_messages(
         })
         .collect();
 
-    match repo.bulk_import(&records).await {
-        Ok(imported) => (StatusCode::OK, Json(ImportResult { imported })).into_response(),
-        Err(BulkImportError::Conflict {
+    let imported = repo.bulk_import(&records).await.map_err(|err| match err {
+        BulkImportError::Conflict {
             source,
             source_conversation_id,
             source_message_id,
-        }) => (
-            StatusCode::CONFLICT,
-            Json(ImportError {
-                error: format!(
-                    "import aborted: entry ({source}, {source_conversation_id}, {source_message_id}) collides with an existing message"
-                ),
-                conflict: Some(ConflictDetails {
-                    source,
-                    source_conversation_id,
-                    source_message_id,
-                }),
-            }),
-        )
-            .into_response(),
-        Err(BulkImportError::Database(err)) => {
-            error!(error = %err, "failed to import journal entries");
-            (
-                StatusCode::INTERNAL_SERVER_ERROR,
-                Json(ImportError {
-                    error: "failed to import messages".to_string(),
-                    conflict: None,
-                }),
-            )
-                .into_response()
-        }
-    }
+        } => DashboardError::ImportConflict {
+            source,
+            source_conversation_id,
+            source_message_id,
+        },
+        BulkImportError::Database(e) => DashboardError::ImportRepository(e),
+    })?;
+
+    Ok((StatusCode::OK, Json(ImportResult { imported })).into_response())
 }
 
 #[derive(Serialize)]
@@ -564,7 +604,7 @@ mod tests {
         let parsed: Value = serde_json::from_slice(&body).unwrap();
         assert_eq!(parsed["imported"], 2);
 
-        let entries = repo.fetch_recent("7", 10).await.unwrap();
+        let entries = repo.fetch_recent(10).await.unwrap();
         assert_eq!(entries.len(), 2);
     }
 
@@ -616,7 +656,7 @@ mod tests {
         assert_eq!(parsed["conflict"]["source_conversation_id"], "42");
         assert_eq!(parsed["conflict"]["source_message_id"], "dup");
 
-        let entries = repo.fetch_recent("7", 10).await.unwrap();
+        let entries = repo.fetch_recent(10).await.unwrap();
         assert_eq!(entries.len(), 1);
         assert_eq!(entries[0].entry.text, "existing");
     }

src/journal/analyzer/journal.rs

@@ -63,7 +63,7 @@ fn map_storage_error(err: sqlx::Error) -> AnalyzerError {
 impl JournalReadService for DefaultJournalReadService {
     async fn get_recent(
         &self,
-        ctx: &UserContext,
+        _ctx: &UserContext,
         request: GetRecentRequest,
     ) -> Result<Vec<JournalEntryView>, AnalyzerError> {
         let limit = validate_limit(request.limit, MAX_RECENT_LIMIT)?;
@@ -72,14 +72,14 @@ impl JournalReadService for DefaultJournalReadService {
         let entries = match (request.from_date, request.to_date_exclusive) {
             (None, None) => self
                 .repository
-                .fetch_recent(&ctx.user_id, limit)
+                .fetch_recent(limit)
                 .await
                 .map_err(map_storage_error)?,
             (from, to) => {
                 let from = from.unwrap_or(chrono::NaiveDate::from_ymd_opt(1970, 1, 1).unwrap());
                 let to = to.unwrap_or(chrono::NaiveDate::from_ymd_opt(9999, 1, 1).unwrap());
                 self.repository
-                    .fetch_in_range(&ctx.user_id, from, to, limit)
+                    .fetch_in_range(from, to, limit)
                     .await
                     .map_err(map_storage_error)?
             }
@@ -90,7 +90,7 @@ impl JournalReadService for DefaultJournalReadService {
 
     async fn search_text(
         &self,
-        ctx: &UserContext,
+        _ctx: &UserContext,
         request: SearchTextRequest,
     ) -> Result<Vec<JournalEntryView>, AnalyzerError> {
         let limit = validate_limit(request.limit, MAX_TEXT_SEARCH_LIMIT)?;
@@ -104,13 +104,7 @@ impl JournalReadService for DefaultJournalReadService {
 
         let entries = self
             .repository
-            .search_text(
-                &ctx.user_id,
-                trimmed,
-                request.from_date,
-                request.to_date_exclusive,
-                limit,
-            )
+            .search_text(trimmed, request.from_date, request.to_date_exclusive, limit)
             .await
             .map_err(map_storage_error)?;
 
@@ -147,12 +141,12 @@ impl JournalReadService for DefaultJournalReadService {
 
     async fn get_by_id(
         &self,
-        ctx: &UserContext,
+        _ctx: &UserContext,
         id: &str,
     ) -> Result<Option<JournalEntryView>, AnalyzerError> {
         let mut rows = self
             .repository
-            .fetch_by_ids(&ctx.user_id, &[id.to_string()])
+            .fetch_by_ids(&[id.to_string()])
             .await
             .map_err(map_storage_error)?;