Init

src/store.rs

@@ -0,0 +1,398 @@
+use crate::{
+    domain::*,
+    error::{Error, Result},
+    models::*,
+};
+use chrono::{DateTime, Utc};
+use cot::db::{Database, Model, query};
+use serde::{Deserialize, Serialize};
+use serde_json::{Value, json};
+use std::collections::BTreeSet;
+use uuid::Uuid;
+
+#[derive(Clone, Debug)]
+pub struct Store {
+    pub db: Database,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct WorkspaceRecord {
+    pub id: Uuid,
+    pub revision: i64,
+    pub document: WorkspaceDocument,
+}
+
+pub fn uuid(value: &str) -> Result<Uuid> {
+    Uuid::parse_str(value).map_err(|_| {
+        Error::Internal(crate::i18n::text("err_invalid_stored_identifier_7c6a5b").into())
+    })
+}
+pub fn user(row: UserRow) -> Result<User> {
+    Ok(User {
+        language: row.language,
+        id: uuid(&row.id)?,
+        username: row.username,
+        display_name: row.display_name,
+        country: row.country,
+        timezone: row.timezone,
+    })
+}
+pub fn record(row: WorkspaceRow) -> Result<WorkspaceRecord> {
+    if row.demo_deleted_at.is_some() {
+        return Err(Error::NotFound);
+    }
+    Ok(WorkspaceRecord {
+        id: uuid(&row.id)?,
+
+        revision: row.revision,
+        document: serde_json::from_str(&row.document)?,
+    })
+}
+
+impl Store {
+    pub async fn connect(url: &str) -> Result<Self> {
+        let db = Database::new(url).await?;
+        cot::db::migrations::MigrationEngine::new(crate::migrations::MIGRATIONS.iter().copied())?
+            .run(&db)
+            .await?;
+        Ok(Self { db })
+    }
+    pub async fn workspace(&self, workspace: Uuid) -> Result<WorkspaceRecord> {
+        let id = workspace.to_string();
+        record(
+            query!(WorkspaceRow,$id==id)
+                .get(&self.db)
+                .await?
+                .ok_or(Error::NotFound)?,
+        )
+    }
+    pub async fn permissions(&self, workspace: Uuid, actor: Uuid) -> Result<Vec<Permission>> {
+        self.workspace(workspace).await?;
+        let workspace = workspace.to_string();
+        let actor = actor.to_string();
+        let row = query!(WorkspaceMemberRow,$workspace_id==workspace && $user_id==actor)
+            .get(&self.db)
+            .await?
+            .ok_or(Error::NotFound)?;
+        if row.is_owner {
+            Ok(Permission::all())
+        } else {
+            Ok(serde_json::from_str(&row.permissions)?)
+        }
+    }
+    pub async fn require(
+        &self,
+        workspace: Uuid,
+        actor: Uuid,
+        permission: Permission,
+    ) -> Result<Vec<Permission>> {
+        let grants = self.permissions(workspace, actor).await?;
+        if grants.contains(&permission) {
+            Ok(grants)
+        } else {
+            Err(Error::Forbidden)
+        }
+    }
+    pub async fn owner(&self, workspace: Uuid, actor: Uuid) -> Result<()> {
+        self.workspace(workspace).await?;
+        let workspace = workspace.to_string();
+        let actor = actor.to_string();
+        if query!(WorkspaceMemberRow,$workspace_id==workspace && $user_id==actor && $is_owner==true)
+            .exists(&self.db)
+            .await?
+        {
+            Ok(())
+        } else {
+            Err(Error::Forbidden)
+        }
+    }
+    pub async fn members(&self, workspace: Uuid) -> Result<Vec<Member>> {
+        let workspace = workspace.to_string();
+        let mut members = vec![];
+        for m in query!(WorkspaceMemberRow,$workspace_id==workspace)
+            .all(&self.db)
+            .await?
+        {
+            let id = m.user_id;
+            let row = query!(UserRow,$id==id)
+                .get(&self.db)
+                .await?
+                .ok_or(Error::NotFound)?;
+            members.push(Member {
+                user: user(row)?,
+                permissions: if m.is_owner {
+                    Permission::all()
+                } else {
+                    serde_json::from_str(&m.permissions)?
+                },
+            });
+        }
+        members.sort_by(|a, b| a.user.display_name.cmp(&b.user.display_name));
+        Ok(members)
+    }
+    pub async fn member_ids(&self, workspace: Uuid) -> Result<BTreeSet<Uuid>> {
+        Ok(self
+            .members(workspace)
+            .await?
+            .into_iter()
+            .map(|m| m.user.id)
+            .collect())
+    }
+    pub async fn directory(&self, workspace: Uuid) -> Result<Vec<User>> {
+        Ok(self
+            .members(workspace)
+            .await?
+            .into_iter()
+            .map(|m| m.user)
+            .collect())
+    }
+    pub async fn working_interval(&self, workspace: Uuid) -> Result<WorkingWindow> {
+        Ok(self.workspace(workspace).await?.document.working_interval)
+    }
+    pub async fn prepare(
+        &self,
+        workspace: Uuid,
+        revision: i64,
+        actor: Uuid,
+        permission: Permission,
+    ) -> Result<WorkspaceRecord> {
+        self.require(workspace, actor, permission).await?;
+        let record = self.workspace(workspace).await?;
+        if record.revision != revision {
+            return Err(Error::Conflict(
+                crate::i18n::text("err_this_team_changed_refresh_before_saving_7f93ad").into(),
+            ));
+        }
+        Ok(record)
+    }
+    /// Cot 0.7 has no transaction API. A single parameterized CTE commits the CAS,
+    /// membership (if any), history, audit, and durable job as one PostgreSQL statement.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn commit(
+        &self,
+        record: &mut WorkspaceRecord,
+        actor: Uuid,
+        permission: Permission,
+        action: &str,
+        details: Value,
+        member: Option<(Uuid, Vec<Permission>)>,
+    ) -> Result<()> {
+        let workspace = record.id.to_string();
+        let actor = actor.to_string();
+        let document = serde_json::to_string(&record.document)?;
+        let details = details.to_string();
+        let required = serde_json::to_string(&vec![permission])?;
+        let member_id = member.as_ref().map(|(id, _)| id.to_string());
+        let grants = serde_json::to_string(&member.map(|(_, g)| g).unwrap_or_default())?;
+        let result = self
+            .db
+            .raw_with(
+                COMMIT_WORKSPACE,
+                &[
+                    &workspace,
+                    &record.revision,
+                    &document,
+                    &actor,
+                    &action,
+                    &details,
+                    &required,
+                    &member_id,
+                    &grants,
+                ],
+            )
+            .await?;
+        if result.rows_affected().0 == 0 {
+            return Err(Error::Conflict(
+                crate::i18n::text("err_this_team_or_your_permissions_changed_refresh_be_c1e4c2")
+                    .into(),
+            ));
+        }
+        record.revision += 1;
+        Ok(())
+    }
+    pub async fn audit(&self, workspace: Uuid) -> Result<Vec<Value>> {
+        let mut names = crate::audit::Names::new();
+        for person in self.directory(workspace).await? {
+            names.insert(person.id.to_string(), person.display_name);
+        }
+        let current = self.workspace(workspace).await?;
+        crate::audit::remember(&mut names, &current.document);
+        let history_team = workspace.to_string();
+        let mut history = query!(HistoryRow, $workspace_id == history_team)
+            .all(&self.db)
+            .await?;
+        history.sort_by_key(|h| (h.effective_at, h.revision));
+        let mut history = history.into_iter().peekable();
+        let workspace = workspace.to_string();
+        let mut events = query!(AuditRow,$workspace_id==workspace)
+            .all(&self.db)
+            .await?;
+        events.sort_by_key(|a| std::cmp::Reverse(a.id.unwrap()));
+        events.truncate(100);
+        events.sort_by_key(|a| (a.created_at, a.id.unwrap()));
+        let mut output = vec![];
+        for e in events {
+            while history
+                .peek()
+                .is_some_and(|h| h.effective_at <= e.created_at)
+            {
+                let snapshot = history.next().unwrap();
+                let document: WorkspaceDocument = serde_json::from_str(&snapshot.document)?;
+                crate::audit::remember(&mut names, &document);
+            }
+            let actor_name = e.actor_id.as_ref().and_then(|id| names.get(id)).cloned();
+            let actor = if let Some(id) = e.actor_id {
+                query!(UserRow,$id==id)
+                    .get(&self.db)
+                    .await?
+                    .map(|u| u.username)
+                    .unwrap_or("Unknown".into())
+            } else {
+                "System".into()
+            };
+            let actor_name = actor_name
+                .map(|name| format!("{name} (@{actor})"))
+                .unwrap_or_else(|| actor.clone());
+            let details: Value = serde_json::from_str(&e.details)?;
+            let resolved_details = crate::audit::resolve(&details, &names);
+            output.push(json!({"id":e.id.unwrap(),"action":e.action,"details":details,"resolved_details":resolved_details,"created_at":e.created_at,"actor":actor,"actor_name":actor_name}));
+        }
+        output.sort_by_key(|e| std::cmp::Reverse(e["id"].as_i64().unwrap_or_default()));
+        Ok(output)
+    }
+    pub async fn sync_status(&self, workspace: Uuid) -> Result<Value> {
+        let workspace = workspace.to_string();
+        Ok(
+            match query!(SyncJobRow,$workspace_id==workspace)
+                .get(&self.db)
+                .await?
+            {
+                None => json!({"state":"pending"}),
+                Some(r) => {
+                    json!({"requested_revision":r.requested_revision,"published_revision":r.published_revision,"state":r.state,"attempts":r.attempts,"last_error":r.last_error,"last_success":r.last_success,"window_end":r.window_end})
+                }
+            },
+        )
+    }
+    pub async fn historical_timeline(
+        &self,
+        workspace: &WorkspaceRecord,
+        from: DateTime<Utc>,
+        to: DateTime<Utc>,
+    ) -> Result<Timeline> {
+        crate::error::ensure(
+            from < to && (to - from) <= chrono::Duration::days(367),
+            crate::i18n::text("err_invalid_calendar_range_c177b5"),
+        )?;
+        let id = workspace.id.to_string();
+        let revision = workspace.revision;
+        let mut history =
+            query!(HistoryRow,$workspace_id==id && $effective_at<to && $revision<=revision)
+                .all(&self.db)
+                .await?;
+        history.sort_by_key(|r| (r.effective_at, r.revision));
+        let mut result = Timeline {
+            from,
+            to,
+            assignments: vec![],
+        };
+        for (index, row) in history.iter().enumerate() {
+            let start = row.effective_at.max(from);
+            let end = history
+                .get(index + 1)
+                .map(|r| r.effective_at)
+                .unwrap_or(to)
+                .min(to);
+            if start < end {
+                result.assignments.extend(
+                    crate::engine::timeline(&serde_json::from_str(&row.document)?, start, end)?
+                        .assignments,
+                );
+            }
+        }
+        // History records actions as well as rule changes. Adjacent identical
+        // assignments remain one interval in the calendar.
+        result.assignments.sort_by_key(|a| (a.schedule_id, a.start));
+        let mut merged: Vec<Assignment> = vec![];
+        for assignment in result.assignments {
+            if let Some(last) = merged.last_mut()
+                && last.schedule_id == assignment.schedule_id
+                && last.end == assignment.start
+                && last.user_id == assignment.user_id
+                && last.reason == assignment.reason
+                && last.unavailable == assignment.unavailable
+            {
+                last.end = assignment.end;
+            } else {
+                merged.push(assignment);
+            }
+        }
+        result.assignments = merged;
+        Ok(result)
+    }
+    pub async fn objects(&self, workspace: Uuid) -> Result<Vec<ProviderObjectRow>> {
+        let workspace = workspace.to_string();
+        Ok(query!(ProviderObjectRow,$workspace_id==workspace)
+            .all(&self.db)
+            .await?)
+    }
+    pub async fn checkpoint(
+        &self,
+        workspace: Uuid,
+        key: &str,
+        remote_id: &str,
+        content: &Value,
+    ) -> Result<()> {
+        ProviderObjectRow {
+            id: format!("{workspace}:{key}"),
+            workspace_id: workspace.to_string(),
+            object_key: key.into(),
+            remote_id: remote_id.into(),
+            content: content.to_string(),
+            updated_at: Utc::now(),
+        }
+        .save(&self.db)
+        .await?;
+        Ok(())
+    }
+    pub async fn forget(&self, workspace: Uuid, key: &str) -> Result<()> {
+        let id = format!("{workspace}:{key}");
+        query!(ProviderObjectRow,$id==id).delete(&self.db).await?;
+        Ok(())
+    }
+    pub async fn retry(&self, workspace: Uuid, actor: Uuid) -> Result<()> {
+        self.require(workspace, actor, Permission::ManageIntegration)
+            .await?;
+        let mut record = self.workspace(workspace).await?;
+        self.commit(
+            &mut record,
+            actor,
+            Permission::ManageIntegration,
+            "publication.requested",
+            json!({}),
+            None,
+        )
+        .await
+    }
+}
+
+const COMMIT_WORKSPACE: &str = r#"
+WITH updated AS (
+ UPDATE meguri__workspaces SET document=$3,revision=revision+1
+ WHERE id=$1 AND revision=$2 AND demo_deleted_at IS NULL
+ AND EXISTS(SELECT 1 FROM meguri__workspace_members WHERE workspace_id=$1 AND user_id=$4 AND (is_owner OR permissions::jsonb @> $7::jsonb)) RETURNING *
+), member AS (
+ INSERT INTO meguri__workspace_members(id,workspace_id,user_id,permissions,is_owner)
+ SELECT $1||':'||$8,$1,$8,$9,false FROM updated WHERE $8::text IS NOT NULL
+ ON CONFLICT(id) DO UPDATE SET permissions=EXCLUDED.permissions
+), history AS (
+ INSERT INTO meguri__workspace_history(id,workspace_id,revision,document,effective_at,actor_id)
+ SELECT id||':'||revision::text,id,revision,document,clock_timestamp(),$4 FROM updated RETURNING effective_at
+), jobs AS (
+ INSERT INTO meguri__sync_jobs(workspace_id,requested_revision,state,attempts,next_attempt)
+ SELECT id,revision,'pending',0,now() FROM updated
+ ON CONFLICT(workspace_id) DO UPDATE SET requested_revision=EXCLUDED.requested_revision,state='pending',attempts=0,next_attempt=now(),last_error=NULL
+)
+INSERT INTO meguri__audit_events(workspace_id,actor_id,action,details,created_at)
+SELECT id,$4,$5,$6,(SELECT effective_at FROM history) FROM updated
+"#;