meguri/src/store.rs

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,
        email: row.email,
        instance_admin: row.instance_admin,
        has_password: row.password_hash.is_some(),
    })
}
pub fn record(row: WorkspaceRow) -> Result<WorkspaceRecord> {
    if row.demo_deleted_at.is_some() || row.disabled_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 is_instance_admin(&self, user: Uuid) -> Result<bool> {
        let id = user.to_string();
        Ok(query!(UserRow,$id==id && $instance_admin==true)
            .exists(&self.db)
            .await?)
    }
    pub async fn workspace_key(&self, workspace: Uuid) -> Result<Option<String>> {
        let id = workspace.to_string();
        Ok(query!(WorkspaceRow,$id==id)
            .get(&self.db)
            .await?
            .and_then(|row| row.key))
    }
    pub async fn permissions(&self, workspace: Uuid, actor: Uuid) -> Result<Vec<Permission>> {
        self.workspace(workspace).await?;
        if self.is_instance_admin(actor).await? {
            return Ok(Permission::all());
        }
        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?;
        if self.is_instance_admin(actor).await? {
            return Ok(());
        }
        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)?
                },
                is_owner: m.is_owner,
            });
        }
        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,
    /// employee-role grants (if changed), history, audit, and durable job as one
    /// PostgreSQL statement. Instance administrators pass the permission check.
    pub async fn commit(
        &self,
        record: &mut WorkspaceRecord,
        actor: Uuid,
        permission: Permission,
        action: &str,
        details: Value,
        role_permissions: Option<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 role_permissions = role_permissions
            .map(|grants| serde_json::to_string(&grants))
            .transpose()?;
        let result = self
            .db
            .raw_with(
                COMMIT_WORKSPACE,
                &[
                    &workspace,
                    &record.revision,
                    &document,
                    &actor,
                    &action,
                    &details,
                    &required,
                    &role_permissions,
                ],
            )
            .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(())
    }
    /// Revision-checked change made by the system (configuration sync or identity linking)
    /// rather than by a person with a workspace grant. Never expose this to request input.
    pub async fn system_commit(
        &self,
        record: &mut WorkspaceRecord,
        actor: Option<Uuid>,
        action: &str,
        details: Value,
    ) -> Result<()> {
        let workspace = record.id.to_string();
        let actor = actor.map(|a| a.to_string());
        let document = serde_json::to_string(&record.document)?;
        let details = details.to_string();
        let result = self
            .db
            .raw_with(
                SYSTEM_COMMIT_WORKSPACE,
                &[
                    &workspace,
                    &record.revision,
                    &document,
                    &actor,
                    &action,
                    &details,
                ],
            )
            .await?;
        if result.rows_affected().0 == 0 {
            return Err(Error::Conflict(
                crate::i18n::text("err_this_team_changed_refresh_before_saving_7f93ad").into(),
            ));
        }
        record.revision += 1;
        Ok(())
    }
    pub async fn instance_event(
        &self,
        actor: Option<Uuid>,
        action: &str,
        details: Value,
    ) -> Result<()> {
        let actor = actor.map(|a| a.to_string());
        let details = details.to_string();
        self.db
            .raw_with(
                "INSERT INTO meguri__instance_audit(actor_id,action,details,created_at) VALUES($1,$2,$3,now())",
                &[&actor, &action, &details],
            )
            .await?;
        if action == "login.failed" {
            // Anonymous callers can create these rows, so only the newest few thousand are kept.
            self.db
                .raw(
                    "DELETE FROM meguri__instance_audit WHERE action='login.failed' AND id<=COALESCE((SELECT id FROM meguri__instance_audit WHERE action='login.failed' ORDER BY id DESC OFFSET 2000 LIMIT 1),0)",
                )
                .await?;
        }
        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 disabled_at IS NULL
 AND (EXISTS(SELECT 1 FROM meguri__users WHERE id=$4 AND instance_admin)
  OR EXISTS(SELECT 1 FROM meguri__workspace_members WHERE workspace_id=$1 AND user_id=$4 AND (is_owner OR permissions::jsonb @> $7::jsonb))) RETURNING *
), roles AS (
 UPDATE meguri__workspace_members SET permissions=$8 WHERE $8::text IS NOT NULL AND NOT is_owner AND workspace_id IN(SELECT id FROM updated)
), 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
"#;

const SYSTEM_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 disabled_at IS NULL RETURNING *
), 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::text 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::text,$5,$6,(SELECT effective_at FROM history) FROM updated
"#;