IdeA/crates/infrastructure/src/store/scheduled_task.rs

//! Machine-local JSON persistence for project scheduled task definitions.

use std::path::{Path, PathBuf};

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use tokio::sync::Mutex;

use domain::{
    ProjectId, ScheduledProjectTask, ScheduledProjectTaskStore, ScheduledTaskApproval,
    ScheduledTaskAuditEntry, ScheduledTaskId, ScheduledTaskStoreError,
};

const DOCUMENT_VERSION: u32 = 1;

#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct Document {
    version: u32,
    project_id: ProjectId,
    #[serde(default)]
    tasks: Vec<ScheduledProjectTask>,
    #[serde(default)]
    audit: Vec<ScheduledTaskAuditEntry>,
}

/// JSON store rooted at `<app_data_dir>/scheduled-project-tasks`.
pub struct FsScheduledProjectTaskStore {
    dir: PathBuf,
    lock: Mutex<()>,
}

impl FsScheduledProjectTaskStore {
    /// Creates a store rooted at `<app_data_dir>/scheduled-project-tasks`.
    #[must_use]
    pub fn new(app_data_dir: impl AsRef<Path>) -> Self {
        Self {
            dir: app_data_dir.as_ref().join("scheduled-project-tasks"),
            lock: Mutex::new(()),
        }
    }

    /// Returns the machine-local storage directory.
    #[must_use]
    pub fn dir(&self) -> &Path {
        &self.dir
    }

    fn path(&self, project_id: ProjectId) -> PathBuf {
        self.dir.join(format!("{project_id}.json"))
    }

    async fn read_unlocked(
        &self,
        project_id: ProjectId,
    ) -> Result<Document, ScheduledTaskStoreError> {
        let path = self.path(project_id);
        let bytes = match tokio::fs::read(&path).await {
            Ok(bytes) => bytes,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
                return Ok(Document {
                    version: DOCUMENT_VERSION,
                    project_id,
                    tasks: Vec::new(),
                    audit: Vec::new(),
                })
            }
            Err(err) => return Err(store_err(err)),
        };
        let document: Document = serde_json::from_slice(&bytes)
            .map_err(|err| ScheduledTaskStoreError::Store(err.to_string()))?;
        if document.version != DOCUMENT_VERSION || document.project_id != project_id {
            return Err(ScheduledTaskStoreError::Invalid(format!(
                "unexpected scheduled task document identity/version in {}",
                path.display()
            )));
        }
        for task in &document.tasks {
            if task.project_id != project_id {
                return Err(ScheduledTaskStoreError::Invalid(
                    "foreign task in project document".to_owned(),
                ));
            }
            task.validate()
                .map_err(|err| ScheduledTaskStoreError::Invalid(err.to_string()))?;
        }
        Ok(document)
    }

    async fn write_unlocked(&self, document: &Document) -> Result<(), ScheduledTaskStoreError> {
        tokio::fs::create_dir_all(&self.dir)
            .await
            .map_err(store_err)?;
        let path = self.path(document.project_id);
        let tmp = self.dir.join(format!(
            ".{}.{}.tmp",
            document.project_id,
            uuid::Uuid::new_v4()
        ));
        let bytes = serde_json::to_vec_pretty(document)
            .map_err(|err| ScheduledTaskStoreError::Store(err.to_string()))?;
        tokio::fs::write(&tmp, bytes).await.map_err(store_err)?;
        if let Err(err) = tokio::fs::rename(&tmp, &path).await {
            let _ = tokio::fs::remove_file(&tmp).await;
            return Err(store_err(err));
        }
        Ok(())
    }
}

fn store_err(err: std::io::Error) -> ScheduledTaskStoreError {
    ScheduledTaskStoreError::Store(err.to_string())
}

#[async_trait]
impl ScheduledProjectTaskStore for FsScheduledProjectTaskStore {
    async fn list(
        &self,
        project_id: ProjectId,
    ) -> Result<Vec<ScheduledProjectTask>, ScheduledTaskStoreError> {
        let _guard = self.lock.lock().await;
        let mut tasks = self.read_unlocked(project_id).await?.tasks;
        tasks.sort_by_key(|task| (task.created_at_ms, task.id));
        Ok(tasks)
    }

    async fn get(
        &self,
        project_id: ProjectId,
        scheduled_task_id: ScheduledTaskId,
    ) -> Result<Option<ScheduledProjectTask>, ScheduledTaskStoreError> {
        let _guard = self.lock.lock().await;
        Ok(self
            .read_unlocked(project_id)
            .await?
            .tasks
            .into_iter()
            .find(|task| task.id == scheduled_task_id))
    }

    async fn save(
        &self,
        task: &ScheduledProjectTask,
        audit: ScheduledTaskAuditEntry,
    ) -> Result<(), ScheduledTaskStoreError> {
        task.validate()
            .map_err(|err| ScheduledTaskStoreError::Invalid(err.to_string()))?;
        if audit.project_id != task.project_id
            || audit.scheduled_task_id != task.id
            || audit.revision != task.audit_revision
        {
            return Err(ScheduledTaskStoreError::Invalid(
                "audit identity/revision mismatch".to_owned(),
            ));
        }
        let _guard = self.lock.lock().await;
        let mut document = self.read_unlocked(task.project_id).await?;
        if let Some(existing) = document
            .tasks
            .iter_mut()
            .find(|existing| existing.id == task.id)
        {
            if task.audit_revision <= existing.audit_revision {
                return Err(ScheduledTaskStoreError::Invalid(
                    "stale scheduled task revision".to_owned(),
                ));
            }
            *existing = task.clone();
        } else {
            document.tasks.push(task.clone());
        }
        document.audit.push(audit);
        self.write_unlocked(&document).await
    }

    async fn delete(
        &self,
        project_id: ProjectId,
        scheduled_task_id: ScheduledTaskId,
        audit: ScheduledTaskAuditEntry,
    ) -> Result<(), ScheduledTaskStoreError> {
        let _guard = self.lock.lock().await;
        let mut document = self.read_unlocked(project_id).await?;
        let before = document.tasks.len();
        document.tasks.retain(|task| task.id != scheduled_task_id);
        if before == document.tasks.len() {
            return Err(ScheduledTaskStoreError::NotFound);
        }
        document.audit.push(audit);
        self.write_unlocked(&document).await
    }

    async fn list_due(
        &self,
        now_ms: u64,
    ) -> Result<Vec<ScheduledProjectTask>, ScheduledTaskStoreError> {
        let _guard = self.lock.lock().await;
        let mut due = Vec::new();
        let mut entries = match tokio::fs::read_dir(&self.dir).await {
            Ok(entries) => entries,
            Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(err) => return Err(store_err(err)),
        };
        while let Some(entry) = entries.next_entry().await.map_err(store_err)? {
            if entry.path().extension().and_then(|v| v.to_str()) != Some("json") {
                continue;
            }
            let bytes = tokio::fs::read(entry.path()).await.map_err(store_err)?;
            let document: Document = serde_json::from_slice(&bytes)
                .map_err(|err| ScheduledTaskStoreError::Store(err.to_string()))?;
            due.extend(document.tasks.into_iter().filter(|task| {
                task.enabled
                    && task.approval == ScheduledTaskApproval::Approved
                    && task.next_run_at_ms.is_some_and(|at| at <= now_ms)
            }));
        }
        Ok(due)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use domain::{ScheduledTaskActor, ScheduledTaskCommand, ScheduledTaskSchedule};

    fn task(project_id: ProjectId) -> ScheduledProjectTask {
        ScheduledProjectTask {
            id: ScheduledTaskId::new_random(),
            project_id,
            name: "cleanup".to_owned(),
            description: None,
            enabled: true,
            approval: ScheduledTaskApproval::Approved,
            sandbox: domain::ScheduledTaskSandbox::ProjectReadWrite,
            command: ScheduledTaskCommand {
                command: "find".to_owned(),
                args: vec!["build".to_owned()],
                cwd: Some(".".to_owned()),
            },
            schedule: ScheduledTaskSchedule::Interval { every_minutes: 15 },
            timezone: "UTC".to_owned(),
            next_run_at_ms: Some(1_000),
            last_run_at_ms: None,
            last_task_id: None,
            created_by: ScheduledTaskActor::Human,
            updated_by: ScheduledTaskActor::Human,
            created_at_ms: 1,
            updated_at_ms: 1,
            audit_revision: 1,
            timeout_ms: None,
        }
    }

    #[tokio::test]
    async fn persists_in_app_data_project_document_and_lists_due() {
        let root =
            std::env::temp_dir().join(format!("idea-scheduled-store-{}", uuid::Uuid::new_v4()));
        let store = FsScheduledProjectTaskStore::new(&root);
        let project_id = ProjectId::new_random();
        let task = task(project_id);
        store
            .save(
                &task,
                ScheduledTaskAuditEntry {
                    revision: 1,
                    project_id,
                    scheduled_task_id: task.id,
                    actor: ScheduledTaskActor::Human,
                    action: "create".to_owned(),
                    summary: "created".to_owned(),
                    at_ms: 1,
                },
            )
            .await
            .unwrap();
        assert_eq!(store.dir(), root.join("scheduled-project-tasks"));
        assert_eq!(store.list(project_id).await.unwrap(), vec![task.clone()]);
        assert_eq!(store.list_due(999).await.unwrap(), Vec::new());
        assert_eq!(store.list_due(1_000).await.unwrap(), vec![task]);
        let _ = tokio::fs::remove_dir_all(root).await;
    }
}