use std::collections::BTreeMap;
use std::path::Path;
use crate::ids::WorkspaceId;
use anyhow::{Result, bail};
use sqlx::SqliteConnection;
use tracing::{info, warn};
use crate::change_log::{ChangeEntity, ChangePayload, append_change, op_type};
use crate::choices::{TaskPriority, TaskStatus};
use crate::db::{Database, begin_immediate, set_field_version};
use crate::ids::{TaskId, new_id, now};
use crate::labels::resolve_labels_in_workspace;
use crate::mutation::{set_task_field, set_task_project};
use crate::projects::resolve_or_create_project_in_workspace;
use crate::refs::get_task_in_workspace;
use crate::task_fields::TaskField;
use crate::types::Task;
use crate::workspaces::Workspace;
pub struct TaskDraft {
pub title: String,
pub description: String,
pub project: Option<String>,
pub status: String,
pub priority: String,
pub labels: Vec<String>,
pub available_at: Option<String>,
pub due_on: Option<String>,
pub is_epic: bool,
}
#[derive(Debug)]
pub struct TaskOutcome {
pub task: Task,
pub create_change_id: Option<String>,
pub attachment_change_ids: Vec<String>,
}
struct InsertedTask {
id: TaskId,
change_id: String,
project_key: String,
label_count: usize,
}
#[derive(Default)]
pub struct TaskUpdate {
pub title: Option<String>,
pub description: Option<String>,
pub project: Option<String>,
pub status: Option<String>,
pub priority: Option<String>,
pub available_at: Option<Option<String>>,
pub due_on: Option<Option<String>>,
pub is_epic: Option<bool>,
pub add_labels: Vec<String>,
pub remove_labels: Vec<String>,
}
pub struct TaskUpdateOutcome {
pub task: Task,
pub changed: bool,
}
pub struct NoteDeleteOutcome {
#[allow(dead_code)]
pub task_id: TaskId,
#[allow(dead_code)]
pub note_id: String,
pub changed: bool,
}
pub struct NoteOutcome {
#[allow(dead_code)]
pub task_id: TaskId,
pub note_id: String,
pub change_id: String,
}
impl Database {
pub async fn create_task(
&self,
workspace: &Workspace,
draft: TaskDraft,
) -> Result<TaskOutcome> {
let mut conn = self.acquire().await?;
create_task(&mut conn, workspace, draft).await
}
pub async fn create_task_with_attachments(
&self,
workspace: &Workspace,
blob_dir: &Path,
lifecycle_policy: crate::attachments::lifecycle::LifecyclePolicy,
draft: TaskDraft,
attachments: Vec<super::attachments::TaskAttachmentAddInput>,
) -> Result<TaskOutcome> {
let mut conn = self.acquire().await?;
create_task_with_attachments(
&mut conn,
workspace,
blob_dir,
lifecycle_policy,
draft,
attachments,
)
.await
}
pub async fn update_task(
&self,
workspace: &Workspace,
task_id: &TaskId,
update: TaskUpdate,
) -> Result<TaskUpdateOutcome> {
let mut conn = self.acquire().await?;
update_task(&mut conn, workspace, task_id, update).await
}
pub async fn update_tasks(
&self,
workspace: &Workspace,
updates: Vec<(TaskId, TaskUpdate)>,
) -> Result<Vec<TaskUpdateOutcome>> {
let mut conn = self.acquire().await?;
let mut outcomes = Vec::with_capacity(updates.len());
for (task_id, update) in updates {
outcomes.push(update_task(&mut conn, workspace, &task_id, update).await?);
}
Ok(outcomes)
}
pub async fn set_task_deleted(
&self,
workspace: &Workspace,
task_id: &TaskId,
deleted: bool,
) -> Result<TaskOutcome> {
let mut conn = self.acquire().await?;
set_task_deleted(&mut conn, workspace, task_id, deleted).await
}
pub async fn add_note(
&self,
workspace: &Workspace,
task_id: &TaskId,
body: String,
) -> Result<NoteOutcome> {
let mut conn = self.acquire().await?;
add_note(&mut conn, workspace, task_id, body).await
}
pub async fn delete_note(
&self,
workspace: &Workspace,
task_id: &TaskId,
note_id: &str,
) -> Result<NoteDeleteOutcome> {
let mut conn = self.acquire().await?;
delete_note(&mut conn, workspace, task_id, note_id).await
}
}
pub async fn create_task(
conn: &mut SqliteConnection,
workspace: &Workspace,
draft: TaskDraft,
) -> Result<TaskOutcome> {
validate_task_draft(&draft)?;
let mut tx = begin_immediate(conn).await?;
let inserted = insert_task(&mut tx, workspace, draft).await?;
tx.commit().await?;
info!(
task_id = %inserted.id,
project_key = %inserted.project_key,
label_count = inserted.label_count,
"task created"
);
Ok(TaskOutcome {
task: get_task_in_workspace(conn, workspace, &inserted.id).await?,
create_change_id: Some(inserted.change_id),
attachment_change_ids: Vec::new(),
})
}
pub async fn create_task_with_attachments(
conn: &mut SqliteConnection,
workspace: &Workspace,
blob_dir: &Path,
lifecycle_policy: crate::attachments::lifecycle::LifecyclePolicy,
draft: TaskDraft,
attachments: Vec<super::attachments::TaskAttachmentAddInput>,
) -> Result<TaskOutcome> {
validate_task_draft(&draft)?;
let mut prepared = Vec::with_capacity(attachments.len());
for attachment in attachments {
prepared.push(super::attachments::prepare_task_attachment(attachment).await?);
}
let mut unique = BTreeMap::new();
for attachment in &prepared {
unique
.entry(attachment.sha256.clone())
.or_insert_with(|| attachment.clone());
}
let mut capacity_reservations = Vec::new();
for attachment in unique.values() {
let available: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM blob_inventory WHERE sha256 = ? AND available = 1)",
)
.bind(&attachment.sha256)
.fetch_one(&mut *conn)
.await?;
if available {
continue;
}
match crate::attachments::lifecycle::ensure_local_capacity(
conn,
blob_dir,
&attachment.sha256,
attachment.byte_size,
lifecycle_policy,
&crate::attachments::lifecycle::SystemClock,
)
.await
{
Ok(Some(reservation_id)) => capacity_reservations.push(reservation_id),
Ok(None) => {}
Err(error) => {
for reservation_id in capacity_reservations {
let _ =
crate::attachments::lifecycle::release_reservation(conn, &reservation_id)
.await;
}
return Err(error);
}
}
}
let mut staging_leases = Vec::with_capacity(unique.len());
for attachment in unique.values() {
match crate::attachments::lifecycle::acquire_lease(
conn,
&attachment.sha256,
"staging",
&crate::attachments::lifecycle::SystemClock,
)
.await
{
Ok(lease_id) => staging_leases.push(lease_id),
Err(error) => {
for lease_id in staging_leases {
let _ = crate::attachments::lifecycle::release_lease(conn, &lease_id).await;
}
for reservation_id in capacity_reservations {
let _ =
crate::attachments::lifecycle::release_reservation(conn, &reservation_id)
.await;
}
return Err(error);
}
}
}
let mut created_hashes = Vec::new();
for attachment in unique.values() {
match crate::attachments::storage::stage_blob(
blob_dir,
&attachment.sha256,
&attachment.bytes,
)
.await
{
Ok(staged) if staged.byte_size == attachment.byte_size => {
if staged.created {
created_hashes.push(staged.sha256);
}
}
Ok(_) => {
cleanup_attachment_guards(conn, &staging_leases, &capacity_reservations).await;
cleanup_created_objects(conn, blob_dir, &created_hashes).await;
bail!("error attachment-staged-size-mismatch");
}
Err(error) => {
cleanup_attachment_guards(conn, &staging_leases, &capacity_reservations).await;
cleanup_created_objects(conn, blob_dir, &created_hashes).await;
return Err(error);
}
}
}
let database_result = async {
let mut tx = begin_immediate(conn).await?;
for attachment in unique.values() {
crate::attachments::storage::upsert_inventory_available(
&mut tx,
&attachment.sha256,
attachment.byte_size,
&attachment.facts.media_type,
)
.await?;
}
let inserted = insert_task(&mut tx, workspace, draft).await?;
let mut attachment_change_ids = Vec::with_capacity(prepared.len());
let attachment_base = chrono::DateTime::parse_from_rfc3339(&now())?.to_utc();
for (index, attachment) in prepared.iter().enumerate() {
let created_at = (attachment_base
+ chrono::TimeDelta::microseconds(i64::try_from(index)?))
.to_rfc3339_opts(chrono::SecondsFormat::Micros, true);
attachment_change_ids.push(
super::attachments::insert_prepared_attachment(
&mut tx,
workspace,
&inserted.id,
attachment,
&created_at,
)
.await?,
);
}
let task = get_task_in_workspace(&mut tx, workspace, &inserted.id).await?;
tx.commit().await?;
Ok::<_, anyhow::Error>((inserted, attachment_change_ids, task))
}
.await;
let (inserted, attachment_change_ids, task) = match database_result {
Ok(value) => value,
Err(error) => {
cleanup_attachment_guards(conn, &staging_leases, &capacity_reservations).await;
cleanup_created_objects(conn, blob_dir, &created_hashes).await;
return Err(error);
}
};
cleanup_attachment_guards(conn, &staging_leases, &capacity_reservations).await;
if let Err(error) = crate::attachments::lifecycle::reconcile_liveness(
conn,
&crate::attachments::lifecycle::SystemClock,
)
.await
{
warn!(%error, "failed to reconcile attachment liveness");
}
info!(
task_id = %inserted.id,
project_key = %inserted.project_key,
label_count = inserted.label_count,
attachment_count = prepared.len(),
"task created"
);
Ok(TaskOutcome {
task,
create_change_id: Some(inserted.change_id),
attachment_change_ids,
})
}
fn validate_task_draft(draft: &TaskDraft) -> Result<()> {
TaskStatus::parse(&draft.status)?;
TaskPriority::parse(&draft.priority)?;
if let Some(available_at) = draft.available_at.as_deref() {
crate::time_validation::validate_available_at_value(available_at)?;
}
if let Some(due_on) = draft.due_on.as_deref() {
crate::time_validation::validate_due_on_value(due_on)?;
}
Ok(())
}
async fn insert_task(
conn: &mut SqliteConnection,
workspace: &Workspace,
draft: TaskDraft,
) -> Result<InsertedTask> {
let status = TaskStatus::parse(&draft.status)?;
let priority = TaskPriority::parse(&draft.priority)?;
let available_at = draft.available_at.as_deref().unwrap_or("");
let due_on = draft.due_on.as_deref().unwrap_or("");
let id = TaskId::new();
let ts = now();
let project = draft
.project
.as_deref()
.ok_or_else(|| anyhow::anyhow!("error project-required"))?;
let project = resolve_or_create_project_in_workspace(conn, &workspace.id, project).await?;
let labels = resolve_labels_in_workspace(conn, &workspace.id, &draft.labels).await?;
sqlx::query(
"INSERT INTO tasks(workspace_id, id, title, description, project_id, status, priority, created_at, updated_at, queue_activity_at, available_at, due_on, is_epic)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
)
.bind(&workspace.id)
.bind(&id)
.bind(&draft.title)
.bind(&draft.description)
.bind(&project.id)
.bind(status.as_str())
.bind(priority.as_str())
.bind(&ts)
.bind(&ts)
.bind(&ts)
.bind(available_at)
.bind(due_on)
.bind(i64::from(draft.is_epic))
.execute(&mut *conn)
.await?;
for label in &labels {
sqlx::query(
"INSERT OR IGNORE INTO task_labels(workspace_id, task_id, label) VALUES (?, ?, ?)",
)
.bind(&workspace.id)
.bind(&id)
.bind(label)
.execute(&mut *conn)
.await?;
}
let change_id = append_change(
conn,
ChangeEntity::Task,
&id,
None,
op_type::CREATE_TASK,
ChangePayload::workspace(workspace)
.set("title", draft.title)
.set("description", draft.description)
.set("project_id", project.id.clone())
.set("project_key", project.key.clone())
.set("project_name", project.name.clone())
.set("project_prefix", project.prefix.clone())
.set("status", status.as_str())
.set("priority", priority.as_str())
.set("available_at", available_at)
.set("due_on", due_on)
.set("is_epic", if draft.is_epic { "1" } else { "0" })
.set("labels", &labels)
.set("created_at", ts),
)
.await?;
for field in TaskField::VERSIONED {
set_field_version(conn, &id, field.as_str(), &change_id).await?;
}
Ok(InsertedTask {
id,
change_id,
project_key: project.key,
label_count: labels.len(),
})
}
async fn cleanup_attachment_guards(
conn: &mut SqliteConnection,
leases: &[String],
reservations: &[String],
) {
for lease_id in leases {
if let Err(error) = crate::attachments::lifecycle::release_lease(conn, lease_id).await {
warn!(%error, "failed to release attachment staging lease");
}
}
for reservation_id in reservations {
if let Err(error) =
crate::attachments::lifecycle::release_reservation(conn, reservation_id).await
{
warn!(%error, "failed to release attachment capacity reservation");
}
}
}
async fn cleanup_created_objects(conn: &mut SqliteConnection, blob_dir: &Path, hashes: &[String]) {
for sha256 in hashes {
crate::attachments::storage::remove_staged_blob_if_unreferenced(conn, blob_dir, sha256)
.await;
}
}
pub async fn update_task(
conn: &mut SqliteConnection,
workspace: &Workspace,
task_id: &crate::ids::TaskId,
update: TaskUpdate,
) -> Result<TaskUpdateOutcome> {
if let Some(status) = update.status.as_deref() {
TaskStatus::parse(status)?;
}
if let Some(priority) = update.priority.as_deref() {
TaskPriority::parse(priority)?;
}
if let Some(Some(available_at)) = update.available_at.as_ref() {
crate::time_validation::validate_available_at_value(available_at)?;
}
if let Some(Some(due_on)) = update.due_on.as_ref() {
crate::time_validation::validate_due_on_value(due_on)?;
}
let mut changed = false;
let mut tx = begin_immediate(conn).await?;
if let Some(title) = update.title {
changed |= update_task_field(&mut tx, workspace, task_id, "title", &title).await?;
}
if let Some(description) = update.description {
changed |=
update_task_field(&mut tx, workspace, task_id, "description", &description).await?;
}
if let Some(project) = update.project {
let project =
resolve_or_create_project_in_workspace(&mut tx, &workspace.id, &project).await?;
changed |= set_task_project(&mut tx, workspace, task_id, &project).await?;
}
if let Some(status) = update.status {
changed |= update_task_field(&mut tx, workspace, task_id, "status", &status).await?;
}
if let Some(priority) = update.priority {
changed |= update_task_field(&mut tx, workspace, task_id, "priority", &priority).await?;
}
if let Some(available_at) = update.available_at {
changed |= update_task_field(
&mut tx,
workspace,
task_id,
"available_at",
available_at.as_deref().unwrap_or(""),
)
.await?;
}
if let Some(due_on) = update.due_on {
changed |= update_task_field(
&mut tx,
workspace,
task_id,
"due_on",
due_on.as_deref().unwrap_or(""),
)
.await?;
}
if let Some(is_epic) = update.is_epic {
if !is_epic {
let task = get_task_in_workspace(&mut tx, workspace, task_id).await?;
if super::epics::task_has_epic_children(&mut tx, &task.workspace_id, task_id).await? {
bail!("error epic-has-children task_id={task_id}");
}
}
changed |= update_task_field(
&mut tx,
workspace,
task_id,
"is_epic",
if is_epic { "1" } else { "0" },
)
.await?;
}
if update_task_labels_in_workspace(
&mut tx,
&workspace.id,
task_id,
&update.add_labels,
&update.remove_labels,
)
.await?
{
changed = true;
}
tx.commit().await?;
info!(task_id = %task_id, changed, "task updated");
Ok(TaskUpdateOutcome {
task: get_task_in_workspace(conn, workspace, task_id).await?,
changed,
})
}
pub async fn update_task_field(
conn: &mut SqliteConnection,
workspace: &Workspace,
task_id: &crate::ids::TaskId,
field: &str,
value: &str,
) -> Result<bool> {
set_task_field(conn, workspace, task_id, field, value).await
}
pub async fn update_task_labels_in_workspace(
conn: &mut SqliteConnection,
workspace_id: &WorkspaceId,
task_id: &crate::ids::TaskId,
add_labels: &[String],
remove_labels: &[String],
) -> Result<bool> {
let workspace = crate::workspaces::workspace_for_id(conn, workspace_id).await?;
let mut changed = false;
for label in resolve_labels_in_workspace(conn, &workspace.id, add_labels).await? {
let rows_affected = sqlx::query(
"INSERT OR IGNORE INTO task_labels(workspace_id, task_id, label) VALUES (?, ?, ?)",
)
.bind(&workspace.id)
.bind(task_id)
.bind(&label)
.execute(&mut *conn)
.await?
.rows_affected();
if rows_affected > 0 {
append_change(
conn,
ChangeEntity::Task,
task_id,
Some("labels"),
op_type::LABEL_ADD,
ChangePayload::workspace(&workspace).set("label", label),
)
.await?;
changed = true;
}
}
for label in resolve_labels_in_workspace(conn, &workspace.id, remove_labels).await? {
let rows_affected = sqlx::query(
"DELETE FROM task_labels WHERE workspace_id = ? AND task_id = ? AND label = ?",
)
.bind(&workspace.id)
.bind(task_id)
.bind(&label)
.execute(&mut *conn)
.await?
.rows_affected();
if rows_affected > 0 {
append_change(
conn,
ChangeEntity::Task,
task_id,
Some("labels"),
op_type::LABEL_REMOVE,
ChangePayload::workspace(&workspace).set("label", label),
)
.await?;
changed = true;
}
}
if changed {
info!(
task_id = %task_id,
added = add_labels.len(),
removed = remove_labels.len(),
"task labels changed"
);
}
Ok(changed)
}
pub async fn set_task_deleted(
conn: &mut SqliteConnection,
workspace: &Workspace,
task_id: &crate::ids::TaskId,
deleted: bool,
) -> Result<TaskOutcome> {
set_task_field(
conn,
workspace,
task_id,
"deleted",
if deleted { "1" } else { "0" },
)
.await?;
crate::attachments::lifecycle::reconcile_liveness(
conn,
&crate::attachments::lifecycle::SystemClock,
)
.await?;
info!(task_id = %task_id, deleted, "task deleted flag changed");
Ok(TaskOutcome {
task: get_task_in_workspace(conn, workspace, task_id).await?,
create_change_id: None,
attachment_change_ids: Vec::new(),
})
}
pub async fn add_note(
conn: &mut SqliteConnection,
workspace: &Workspace,
task_id: &crate::ids::TaskId,
body: String,
) -> Result<NoteOutcome> {
let note_id = new_id();
let ts = now();
let mut tx = begin_immediate(conn).await?;
let change_id = append_change(
&mut tx,
ChangeEntity::Task,
task_id,
Some("notes"),
op_type::NOTE_ADD,
ChangePayload::workspace(workspace)
.set("note_id", ¬e_id)
.set("body", &body)
.set("created_at", &ts),
)
.await?;
sqlx::query(
"INSERT INTO notes(workspace_id, id, task_id, body, created_at, change_id) VALUES (?, ?, ?, ?, ?, ?)",
)
.bind(&workspace.id)
.bind(¬e_id)
.bind(task_id)
.bind(&body)
.bind(&ts)
.bind(&change_id)
.execute(&mut *tx)
.await?;
sqlx::query("UPDATE tasks SET queue_activity_at = ? WHERE workspace_id = ? AND id = ?")
.bind(&ts)
.bind(&workspace.id)
.bind(task_id)
.execute(&mut *tx)
.await?;
tx.commit().await?;
info!(task_id = %task_id, note_id = %note_id, "note added");
Ok(NoteOutcome {
task_id: task_id.clone(),
note_id,
change_id,
})
}
pub async fn delete_note(
conn: &mut SqliteConnection,
workspace: &Workspace,
task_id: &crate::ids::TaskId,
note_id: &str,
) -> Result<NoteDeleteOutcome> {
let mut tx = begin_immediate(conn).await?;
let deleted_at = now();
let deleted =
sqlx::query("DELETE FROM notes WHERE workspace_id = ? AND task_id = ? AND id = ?")
.bind(&workspace.id)
.bind(task_id)
.bind(note_id)
.execute(&mut *tx)
.await?
.rows_affected();
if deleted > 0 {
sqlx::query("UPDATE tasks SET queue_activity_at = ? WHERE workspace_id = ? AND id = ?")
.bind(&deleted_at)
.bind(&workspace.id)
.bind(task_id)
.execute(&mut *tx)
.await?;
append_change(
&mut tx,
ChangeEntity::Task,
task_id,
Some("notes"),
op_type::NOTE_DELETE,
ChangePayload::workspace(workspace)
.set("note_id", note_id)
.set("deleted_at", deleted_at),
)
.await?;
}
tx.commit().await?;
if deleted > 0 {
info!(task_id = %task_id, note_id = %note_id, "note deleted");
}
Ok(NoteDeleteOutcome {
task_id: task_id.clone(),
note_id: note_id.to_string(),
changed: deleted > 0,
})
}