use crate::ids::{ProjectId, WorkspaceId};
use anyhow::{Result, bail};
use serde_json::Value;
use sqlx::SqliteConnection;
use crate::operations::{ProjectMetadata, set_project_metadata as apply_project_metadata};
use crate::projects::{normalize_key, prefix_base};
use crate::sync::wire::ChangeWire;
use super::payload::{CreateProjectPayload, SetProjectMetadataPayload};
use super::shared::str_payload;
pub(super) async fn delete_project(conn: &mut SqliteConnection, change: &ChangeWire) -> Result<()> {
let workspace_id = super::shared::workspace_id_payload(conn, change).await?;
let project_id = resolve_project_for_delete(conn, &workspace_id, change).await?;
let Some(project_id) = project_id else {
return Ok(());
};
let deleted_at = str_payload(&change.payload, "deleted_at")?;
sqlx::query(
"UPDATE projects SET deleted = 1, updated_at = ? WHERE workspace_id = ? AND id = ?",
)
.bind(&deleted_at)
.bind(&workspace_id)
.bind(&project_id)
.execute(&mut *conn)
.await?;
Ok(())
}
async fn resolve_project_for_delete(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
change: &ChangeWire,
) -> Result<Option<ProjectId>> {
let remote_project_id: ProjectId = change.entity_id.parse()?;
if let Some(local_id) = project_id_alias(conn, workspace_id, &remote_project_id).await? {
return Ok(Some(local_id));
}
let exists = sqlx::query_scalar::<_, i64>(
"SELECT count(*) FROM projects WHERE workspace_id = ? AND id = ?",
)
.bind(workspace_id)
.bind(&remote_project_id)
.fetch_one(&mut *conn)
.await?
> 0;
if exists {
return Ok(Some(remote_project_id));
}
Ok(None)
}
pub(super) async fn create_project(conn: &mut SqliteConnection, change: &ChangeWire) -> Result<()> {
let p = CreateProjectPayload::from_change(change)?;
let workspace_id = super::shared::workspace_id_payload(conn, change).await?;
let created_at = p.created_at.unwrap_or_else(crate::ids::now);
let project_id: ProjectId = change.entity_id.parse()?;
ensure_remote_project(
conn,
&workspace_id,
&project_id,
&p.key,
&p.name,
&p.prefix,
&created_at,
)
.await?;
Ok(())
}
pub(super) async fn set_project_metadata(
conn: &mut SqliteConnection,
change: &ChangeWire,
) -> Result<()> {
let p = SetProjectMetadataPayload::from_change(change)?;
let workspace_id = super::shared::workspace_id_payload(conn, change).await?;
validate_project_metadata_payload(&p.key, &p.name, &p.prefix)?;
let project_id = ensure_project_for_metadata(conn, &workspace_id, change).await?;
let resolved_key = unique_remote_key(conn, &workspace_id, &p.key, Some(&project_id)).await?;
let resolved_prefix = unique_remote_prefix(
conn,
&workspace_id,
&p.prefix,
&resolved_key,
Some(&project_id),
)
.await?;
let workspace = crate::workspaces::workspace_for_id(conn, &workspace_id).await?;
apply_project_metadata(
conn,
&workspace,
&project_id,
ProjectMetadata {
key: &resolved_key,
name: &p.name,
prefix: &resolved_prefix,
},
false,
)
.await?;
Ok(())
}
fn validate_project_metadata_payload(key: &str, name: &str, prefix: &str) -> Result<()> {
anyhow::ensure!(
!key.is_empty() && normalize_key(name) == key,
"error invalid-sync-change project-metadata key"
);
anyhow::ensure!(
(2..=8).contains(&prefix.len()) && prefix.chars().all(|ch| ch.is_ascii_alphanumeric()),
"error invalid-sync-change project-metadata prefix"
);
Ok(())
}
async fn ensure_project_for_metadata(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
change: &ChangeWire,
) -> Result<ProjectId> {
let remote_project_id: ProjectId = change.entity_id.parse()?;
if let Some(local_id) = project_id_alias(conn, workspace_id, &remote_project_id).await? {
return Ok(local_id);
}
if let Some(existing_id) = live_project_by_id(conn, workspace_id, &remote_project_id).await? {
return Ok(existing_id);
}
let key = str_payload(&change.payload, "key")?;
let name = str_payload(&change.payload, "name")?;
let prefix = str_payload(&change.payload, "prefix")?;
ensure_remote_project(
conn,
workspace_id,
&remote_project_id,
&key,
&name,
&prefix,
&change.created_at,
)
.await
}
pub(super) async fn ensure_project_for_payload(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
project_id: &ProjectId,
change: &ChangeWire,
) -> Result<ProjectId> {
if change
.payload
.get("series_id")
.and_then(Value::as_str)
.is_some()
{
if let Some(local_id) = project_id_alias_any(conn, workspace_id, project_id).await? {
return Ok(local_id);
}
if let Some(existing_id) = project_by_id_any(conn, workspace_id, project_id).await? {
return Ok(existing_id);
}
bail!("error recurrence-project-not-found project_id={project_id}");
}
let key = str_payload(&change.payload, "project_key")?;
let name = str_payload(&change.payload, "project_name").unwrap_or_else(|_| key.clone());
let prefix =
str_payload(&change.payload, "project_prefix").unwrap_or_else(|_| prefix_base(&key));
ensure_remote_project(
conn,
workspace_id,
project_id,
&key,
&name,
&prefix,
&change.created_at,
)
.await
}
async fn ensure_remote_project(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
remote_project_id: &ProjectId,
key: &str,
name: &str,
prefix: &str,
created_at: &str,
) -> Result<ProjectId> {
if let Some(local_id) = project_id_alias(conn, workspace_id, remote_project_id).await? {
return Ok(local_id);
}
if let Some(existing_id) = live_project_by_id(conn, workspace_id, remote_project_id).await? {
return Ok(existing_id);
}
if let Some(local_id) = live_project_by_key(conn, workspace_id, key).await? {
insert_project_alias(conn, workspace_id, remote_project_id, &local_id).await?;
return Ok(local_id);
}
if deleted_project_by_id(conn, workspace_id, remote_project_id).await? {
restore_remote_project(
conn,
workspace_id,
remote_project_id,
key,
name,
prefix,
created_at,
)
.await?;
return Ok(remote_project_id.clone());
}
if let Some(local_id) = deleted_project_by_key(conn, workspace_id, key).await? {
restore_remote_project(conn, workspace_id, &local_id, key, name, prefix, created_at)
.await?;
insert_project_alias(conn, workspace_id, remote_project_id, &local_id).await?;
return Ok(local_id);
}
let prefix = unique_remote_prefix(conn, workspace_id, prefix, key, None).await?;
sqlx::query(
"INSERT INTO projects(id, workspace_id, key, name, prefix, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?)",
)
.bind(remote_project_id)
.bind(workspace_id)
.bind(key)
.bind(name)
.bind(&prefix)
.bind(created_at)
.bind(created_at)
.execute(&mut *conn)
.await?;
Ok(remote_project_id.clone())
}
async fn live_project_by_id(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
project_id: &ProjectId,
) -> Result<Option<ProjectId>> {
sqlx::query_scalar::<_, ProjectId>(
"SELECT id FROM projects WHERE workspace_id = ? AND id = ? AND deleted = 0",
)
.bind(workspace_id)
.bind(project_id)
.fetch_optional(&mut *conn)
.await
.map_err(Into::into)
}
async fn project_by_id_any(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
project_id: &ProjectId,
) -> Result<Option<ProjectId>> {
sqlx::query_scalar::<_, ProjectId>("SELECT id FROM projects WHERE workspace_id = ? AND id = ?")
.bind(workspace_id)
.bind(project_id)
.fetch_optional(&mut *conn)
.await
.map_err(Into::into)
}
async fn live_project_by_key(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
key: &str,
) -> Result<Option<ProjectId>> {
sqlx::query_scalar::<_, ProjectId>(
"SELECT id FROM projects WHERE workspace_id = ? AND key = ? AND deleted = 0",
)
.bind(workspace_id)
.bind(key)
.fetch_optional(&mut *conn)
.await
.map_err(Into::into)
}
async fn deleted_project_by_id(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
project_id: &ProjectId,
) -> Result<bool> {
Ok(sqlx::query_scalar::<_, i64>(
"SELECT count(*) FROM projects WHERE workspace_id = ? AND id = ? AND deleted = 1",
)
.bind(workspace_id)
.bind(project_id)
.fetch_one(&mut *conn)
.await?
> 0)
}
async fn deleted_project_by_key(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
key: &str,
) -> Result<Option<ProjectId>> {
sqlx::query_scalar::<_, ProjectId>(
"SELECT id FROM projects WHERE workspace_id = ? AND key = ? AND deleted = 1",
)
.bind(workspace_id)
.bind(key)
.fetch_optional(&mut *conn)
.await
.map_err(Into::into)
}
async fn restore_remote_project(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
project_id: &ProjectId,
key: &str,
name: &str,
prefix: &str,
updated_at: &str,
) -> Result<()> {
let prefix = unique_remote_prefix(conn, workspace_id, prefix, key, Some(project_id)).await?;
sqlx::query(
"UPDATE projects SET name = ?, prefix = ?, updated_at = ?, deleted = 0
WHERE workspace_id = ? AND id = ?",
)
.bind(name)
.bind(&prefix)
.bind(updated_at)
.bind(workspace_id)
.bind(project_id)
.execute(&mut *conn)
.await?;
Ok(())
}
async fn insert_project_alias(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
remote_project_id: &ProjectId,
local_project_id: &ProjectId,
) -> Result<()> {
sqlx::query(
"INSERT OR IGNORE INTO project_id_aliases(workspace_id, remote_project_id, local_project_id)
VALUES (?, ?, ?)",
)
.bind(workspace_id)
.bind(remote_project_id)
.bind(local_project_id)
.execute(&mut *conn)
.await?;
Ok(())
}
async fn unique_remote_key(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
preferred: &str,
ignore_project_id: Option<&ProjectId>,
) -> Result<String> {
let mut candidate = preferred.to_string();
let mut n = 2;
while sqlx::query_scalar::<_, i64>(
"SELECT count(*) FROM projects
WHERE workspace_id = ? AND key = ? AND (? IS NULL OR id != ?)",
)
.bind(workspace_id)
.bind(&candidate)
.bind(ignore_project_id)
.bind(ignore_project_id)
.fetch_one(&mut *conn)
.await?
> 0
{
candidate = format!("{preferred}-{n}");
n += 1;
}
Ok(candidate)
}
async fn unique_remote_prefix(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
preferred: &str,
key: &str,
ignore_project_id: Option<&ProjectId>,
) -> Result<String> {
let base = if preferred.trim().is_empty() {
prefix_base(key)
} else {
preferred.to_string()
};
let mut candidate = base.clone();
let mut n = 2;
while sqlx::query_scalar::<_, i64>(
"SELECT count(*) FROM projects
WHERE workspace_id = ? AND prefix = ? AND (? IS NULL OR id != ?)",
)
.bind(workspace_id)
.bind(&candidate)
.bind(ignore_project_id)
.bind(ignore_project_id)
.fetch_one(&mut *conn)
.await?
> 0
{
candidate = format!("{}{}", base.chars().take(2).collect::<String>(), n);
n += 1;
}
Ok(candidate)
}
async fn project_id_alias(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
remote_project_id: &ProjectId,
) -> Result<Option<ProjectId>> {
Ok(sqlx::query_scalar::<_, ProjectId>(
"SELECT a.local_project_id
FROM project_id_aliases a
JOIN projects p ON p.workspace_id = a.workspace_id
AND p.id = a.local_project_id
AND p.deleted = 0
WHERE a.workspace_id = ? AND a.remote_project_id = ?",
)
.bind(workspace_id)
.bind(remote_project_id)
.fetch_optional(&mut *conn)
.await?)
}
async fn project_id_alias_any(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
remote_project_id: &ProjectId,
) -> Result<Option<ProjectId>> {
Ok(sqlx::query_scalar::<_, ProjectId>(
"SELECT a.local_project_id
FROM project_id_aliases a
JOIN projects p ON p.workspace_id = a.workspace_id
AND p.id = a.local_project_id
WHERE a.workspace_id = ? AND a.remote_project_id = ?",
)
.bind(workspace_id)
.bind(remote_project_id)
.fetch_optional(&mut *conn)
.await?)
}
#[cfg(test)]
mod tests {
use serde_json::json;
use super::*;
use crate::projects::create_project;
use crate::test_support::test_conn;
use crate::workspaces::Workspace;
fn recurrence_task_change(workspace: &Workspace, project_id: &ProjectId) -> ChangeWire {
ChangeWire {
change_id: "AAAAAAAAAAAAAAAA".to_string(),
client_id: "remote".to_string(),
local_seq: 1,
entity_type: "task".to_string(),
entity_id: "BBBBBBBBBBBBBBBB".to_string(),
field: None,
op_type: "create_task".to_string(),
payload: json!({
"workspace_id": workspace.id,
"workspace_key": workspace.key,
"series_id": "CCCCCCCCCCCCCCCC",
"project_id": project_id,
"title": "projected task",
}),
base_version: None,
created_at: "2026-08-24T00:00:00Z".to_string(),
server_seq: Some(1),
}
}
#[tokio::test]
async fn recurrence_task_resolves_soft_deleted_project_without_restoring_it() {
let (_temp, mut conn) = test_conn().await;
let workspace = Workspace::default();
let project = create_project(&mut conn, &workspace, "app").await.unwrap();
sqlx::query("UPDATE projects SET deleted = 1 WHERE workspace_id = ? AND id = ?")
.bind(&workspace.id)
.bind(&project.id)
.execute(&mut *conn)
.await
.unwrap();
let change = recurrence_task_change(&workspace, &project.id);
super::super::task::create_task(&mut conn, &change)
.await
.unwrap();
let task_project_id: ProjectId =
sqlx::query_scalar("SELECT project_id FROM tasks WHERE workspace_id = ? AND id = ?")
.bind(&workspace.id)
.bind(&change.entity_id)
.fetch_one(&mut *conn)
.await
.unwrap();
assert_eq!(task_project_id, project.id);
let deleted: bool =
sqlx::query_scalar("SELECT deleted FROM projects WHERE workspace_id = ? AND id = ?")
.bind(&workspace.id)
.bind(&project.id)
.fetch_one(&mut *conn)
.await
.unwrap();
assert!(deleted);
}
#[tokio::test]
async fn recurrence_task_resolves_alias_to_soft_deleted_project() {
let (_temp, mut conn) = test_conn().await;
let workspace = Workspace::default();
let project = create_project(&mut conn, &workspace, "app").await.unwrap();
let remote_project_id: ProjectId = "DDDDDDDDDDDDDDDD".parse().unwrap();
sqlx::query("UPDATE projects SET deleted = 1 WHERE workspace_id = ? AND id = ?")
.bind(&workspace.id)
.bind(&project.id)
.execute(&mut *conn)
.await
.unwrap();
sqlx::query(
"INSERT INTO project_id_aliases(workspace_id, remote_project_id, local_project_id)
VALUES (?, ?, ?)",
)
.bind(&workspace.id)
.bind(&remote_project_id)
.bind(&project.id)
.execute(&mut *conn)
.await
.unwrap();
let change = recurrence_task_change(&workspace, &remote_project_id);
let resolved =
ensure_project_for_payload(&mut conn, &workspace.id, &remote_project_id, &change)
.await
.unwrap();
assert_eq!(resolved, project.id);
}
}