From 2f216b077968243ae26d748d0b7560c564867f9e Mon Sep 17 00:00:00 2001 From: Michael Feth Date: Mon, 7 Sep 2026 11:16:09 -0400 Subject: [PATCH 1/2] fix(tasks): preserve audit changes and paginate visible work Signed-off-by: Michael Feth --- crates/buzz-core/src/task.rs | 18 +- crates/buzz-db/src/task.rs | 542 ++++----------- crates/buzz-db/src/task/postgres_tests.rs | 766 ++++++++++++++++++++++ crates/buzz-relay/src/api/tasks.rs | 617 ++--------------- crates/buzz-relay/src/api/tasks/tests.rs | 676 +++++++++++++++++++ migrations/0047_task_event_changes.sql | 8 + schema/schema.sql | 3 +- 7 files changed, 1648 insertions(+), 982 deletions(-) create mode 100644 crates/buzz-db/src/task/postgres_tests.rs create mode 100644 crates/buzz-relay/src/api/tasks/tests.rs create mode 100644 migrations/0047_task_event_changes.sql diff --git a/crates/buzz-core/src/task.rs b/crates/buzz-core/src/task.rs index 744bc0abbd8..5d7d49a6127 100644 --- a/crates/buzz-core/src/task.rs +++ b/crates/buzz-core/src/task.rs @@ -82,9 +82,9 @@ impl FromStr for TaskStatus { /// A row in the append-only `task_events` log. /// -/// Stored as free `TEXT` rather than a database enum so a new action can ship -/// across a rolling upgrade without a migration; this enum is the set the -/// relay itself writes. +/// Stored as free `TEXT` rather than a database enum. New spellings do not +/// require a constraint migration, but readers must understand a spelling +/// before writers emit it; parsing an unknown action intentionally fails. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum TaskAction { /// The task was created. @@ -97,6 +97,10 @@ pub enum TaskAction { Commented, /// `title` changed. TitleChanged, + /// `priority` changed. + PriorityChanged, + /// `due_at` changed or was cleared. + DueAtChanged, /// An agent persisted its summary of the task. At most one per task. SummaryPersisted, } @@ -110,6 +114,8 @@ impl TaskAction { Self::Assigned => "assigned", Self::Commented => "commented", Self::TitleChanged => "title_changed", + Self::PriorityChanged => "priority_changed", + Self::DueAtChanged => "due_at_changed", Self::SummaryPersisted => "summary_persisted", } } @@ -138,6 +144,8 @@ impl FromStr for TaskAction { "assigned" => Ok(Self::Assigned), "commented" => Ok(Self::Commented), "title_changed" => Ok(Self::TitleChanged), + "priority_changed" => Ok(Self::PriorityChanged), + "due_at_changed" => Ok(Self::DueAtChanged), "summary_persisted" => Ok(Self::SummaryPersisted), other => Err(format!("unknown task action: {other:?}")), } @@ -185,6 +193,8 @@ mod tests { TaskAction::Assigned, TaskAction::Commented, TaskAction::TitleChanged, + TaskAction::PriorityChanged, + TaskAction::DueAtChanged, TaskAction::SummaryPersisted, ] { assert_eq!(action.as_str().parse::(), Ok(action)); @@ -263,6 +273,8 @@ mod tests { TaskAction::Assigned, TaskAction::Commented, TaskAction::TitleChanged, + TaskAction::PriorityChanged, + TaskAction::DueAtChanged, ] { assert!(!action.is_singleton_per_task()); } diff --git a/crates/buzz-db/src/task.rs b/crates/buzz-db/src/task.rs index c8b6018b461..eafa49c830d 100644 --- a/crates/buzz-db/src/task.rs +++ b/crates/buzz-db/src/task.rs @@ -18,7 +18,9 @@ //! transition neither of them made. use buzz_core::task::{status_change_action, TaskAction, TaskStatus}; -use chrono::{DateTime, Utc}; +use chrono::{DateTime, SubsecRound, Utc}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; use sqlx::{PgPool, Postgres, QueryBuilder, Row as _, Transaction}; use uuid::Uuid; @@ -43,7 +45,7 @@ macro_rules! task_columns { /// Columns selected for every [`TaskEventRecord`]. macro_rules! task_event_columns { () => { - "id, task_id, actor_pubkey, action, from_status, to_status, body, created_at" + "id, task_id, actor_pubkey, action, from_status, to_status, body, changes, created_at" }; } @@ -101,6 +103,8 @@ pub struct TaskEventRecord { pub to_status: Option, /// Comment or summary text. pub body: Option, + /// Structured before/after values, absent for legacy events and comments. + pub changes: Option, /// When it happened. pub created_at: DateTime, } @@ -130,6 +134,19 @@ pub struct NewTask { pub due_at: Option>, } +/// Exclusive boundary for newest-modified-first task pagination. +/// +/// The full timestamp precision and id tie-breaker must both survive the wire. +/// This is a live keyset, not a snapshot: tasks modified between pages move +/// ahead of the cursor and can be found by refreshing the first page. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TaskCursor { + /// Last task's modification timestamp, with database precision. + pub updated_at: DateTime, + /// Last task's id, breaking equal-timestamp ties. + pub id: Uuid, +} + /// Filters for [`list_tasks`]. `None` means "do not filter on this field". #[derive(Debug, Clone, Default)] pub struct TaskFilter { @@ -147,6 +164,11 @@ pub struct TaskFilter { pub source_ref: Option, /// Include archived tasks. Archived tasks are hidden by default. pub include_archived: bool, + /// Caller-visible channels, applied before the limit. `Some([])` permits + /// only channel-less tasks; `None` leaves visibility to a trusted caller. + pub visible_channel_ids: Option>, + /// Return rows strictly after this boundary in newest-modified order. + pub before: Option, /// Maximum rows to return. pub limit: i64, } @@ -219,32 +241,35 @@ fn parse_task_event_row(row: &sqlx::postgres::PgRow) -> Result from_status: from_status.as_deref().map(parse_status).transpose()?, to_status: to_status.as_deref().map(parse_status).transpose()?, body: row.try_get("body")?, + changes: row.try_get("changes")?, created_at: row.try_get("created_at")?, }) } -/// Append one row to a task's history inside an open transaction. -/// -/// `transition` carries the `(from, to)` pair for -/// [`TaskAction::StatusChanged`] and is `None` for every other action — the -/// two ends are only ever meaningful together, so they travel together. +#[derive(Default)] +struct TaskEventContent<'a> { + transition: Option<(TaskStatus, TaskStatus)>, + body: Option<&'a str>, + changes: Option, +} + +/// Append a history row in the same transaction as its task mutation. async fn insert_task_event( tx: &mut Transaction<'_, Postgres>, community: CommunityId, task_id: Uuid, actor_pubkey: Option<&[u8]>, action: TaskAction, - transition: Option<(TaskStatus, TaskStatus)>, - body: Option<&str>, + content: TaskEventContent<'_>, ) -> Result { - let (from_status, to_status) = match transition { + let (from_status, to_status) = match content.transition { Some((from, to)) => (Some(from), Some(to)), None => (None, None), }; let row = sqlx::query(concat!( "INSERT INTO task_events \ - (community_id, task_id, actor_pubkey, action, from_status, to_status, body) \ - VALUES ($1, $2, $3, $4, $5, $6, $7) \ + (community_id, task_id, actor_pubkey, action, from_status, to_status, body, changes) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) \ RETURNING ", task_event_columns!() )) @@ -254,7 +279,8 @@ async fn insert_task_event( .bind(action.as_str()) .bind(from_status.map(|status| status.as_str())) .bind(to_status.map(|status| status.as_str())) - .bind(body) + .bind(content.body) + .bind(content.changes) .fetch_one(&mut **tx) .await?; parse_task_event_row(&row) @@ -300,8 +326,15 @@ pub async fn create_task( task.id, new_task.created_by_pubkey.as_deref(), TaskAction::Created, - None, - None, + TaskEventContent { + changes: Some(json!({ + "title": {"from": null, "to": task.title}, + "assignee": {"from": null, "to": task.assignee_pubkey.as_ref().map(hex::encode)}, + "priority": {"from": null, "to": task.priority}, + "due_at": {"from": null, "to": task.due_at}, + })), + ..TaskEventContent::default() + }, ) .await?; @@ -355,6 +388,18 @@ pub async fn list_tasks( if !filter.include_archived { builder.push(" AND archived_at IS NULL"); } + if let Some(channels) = &filter.visible_channel_ids { + builder.push(" AND (channel_id IS NULL OR channel_id = ANY("); + builder.push_bind(channels); + builder.push("))"); + } + if let Some(cursor) = &filter.before { + builder.push(" AND (updated_at, id) < ("); + builder.push_bind(cursor.updated_at); + builder.push(", "); + builder.push_bind(cursor.id); + builder.push(")"); + } builder.push(" ORDER BY updated_at DESC, id DESC LIMIT "); builder.push_bind(filter.limit); @@ -423,7 +468,12 @@ pub async fn update_task( let new_status = patch.status.unwrap_or(current.status); let new_title = patch.title.clone().unwrap_or_else(|| current.title.clone()); let new_priority = patch.priority.unwrap_or(current.priority); - let new_due_at = patch.due_at.unwrap_or(current.due_at); + // PostgreSQL stores microseconds; normalize before comparison so a client + // retry with nanoseconds neither inflates history nor records phantom digits. + let new_due_at = patch + .due_at + .unwrap_or(current.due_at) + .map(|value| value.trunc_subsecs(6)); let new_assignee = patch .assignee_pubkey .clone() @@ -436,9 +486,19 @@ pub async fn update_task( None }; + if new_status == current.status + && new_title == current.title + && new_priority == current.priority + && new_due_at == current.due_at + && new_assignee == current.assignee_pubkey + { + tx.commit().await?; + return Ok(current); + } + let row = sqlx::query(concat!( "UPDATE tasks SET status = $3, title = $4, priority = $5, due_at = $6, \ - assignee_pubkey = $7, done_at = $8, updated_at = NOW() \ + assignee_pubkey = $7, done_at = $8, updated_at = clock_timestamp() \ WHERE community_id = $1 AND id = $2 \ RETURNING ", task_columns!() @@ -462,8 +522,10 @@ pub async fn update_task( id, actor_pubkey, action, - Some((current.status, new_status)), - None, + TaskEventContent { + transition: Some((current.status, new_status)), + ..TaskEventContent::default() + }, ) .await?; } @@ -474,8 +536,11 @@ pub async fn update_task( id, actor_pubkey, TaskAction::TitleChanged, - None, - Some(&new_title), + TaskEventContent { + body: Some(&new_title), + changes: Some(json!({"title": {"from": current.title, "to": new_title}})), + ..TaskEventContent::default() + }, ) .await?; } @@ -486,12 +551,45 @@ pub async fn update_task( id, actor_pubkey, TaskAction::Assigned, - None, - None, + TaskEventContent { + changes: Some(json!({"assignee": { + "from": current.assignee_pubkey.as_ref().map(hex::encode), + "to": new_assignee.as_ref().map(hex::encode), + }})), + ..TaskEventContent::default() + }, ) .await?; } + for (action, changes) in [ + ( + TaskAction::PriorityChanged, + (new_priority != current.priority) + .then(|| json!({"priority": {"from": current.priority, "to": new_priority}})), + ), + ( + TaskAction::DueAtChanged, + (new_due_at != current.due_at) + .then(|| json!({"due_at": {"from": current.due_at, "to": new_due_at}})), + ), + ] { + if let Some(changes) = changes { + insert_task_event( + &mut tx, + community, + id, + actor_pubkey, + action, + TaskEventContent { + changes: Some(changes), + ..TaskEventContent::default() + }, + ) + .await?; + } + } + tx.commit().await?; Ok(updated) } @@ -529,8 +627,10 @@ pub async fn append_task_event( task_id, actor_pubkey, action, - None, - body, + TaskEventContent { + body, + ..TaskEventContent::default() + }, ) .await .map_err(|error| match &error { @@ -589,25 +689,6 @@ mod tests { "duplicate column in projection" ); } - - // ── Live-Postgres integration coverage ────────────────────────────────── - // - // `#[ignore]`d, exactly like every other Postgres-backed test in this - // crate: `just test-unit` runs `-p buzz-db --lib`, which skips them, and - // `just test` (Docker Postgres + Redis) is what turns them on. Run one - // directly with: - // - // cargo test -p buzz-db --lib crate::task::tests -- --ignored - // - // against a database that has migration 0033 applied. - - const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 - - pub(super) fn test_database_url() -> String { - std::env::var("BUZZ_TEST_DATABASE_URL") - .or_else(|_| std::env::var("DATABASE_URL")) - .unwrap_or_else(|_| TEST_DB_URL.to_owned()) - } } use crate::Db; @@ -687,374 +768,5 @@ impl Db { } #[cfg(test)] -mod postgres_tests { - use super::tests::test_database_url; - use super::*; - async fn setup_pool() -> PgPool { - PgPool::connect(&test_database_url()) - .await - .expect("connect to test DB") - } - - async fn make_test_community(pool: &PgPool) -> CommunityId { - let id = Uuid::new_v4(); - sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") - .bind(id) - .bind(format!("task-test-{}.example", id.simple())) - .execute(pool) - .await - .expect("insert test community"); - CommunityId::from_uuid(id) - } - - async fn make_test_user(pool: &PgPool, community: CommunityId, seed: u8) -> Vec { - let pubkey = vec![seed; 32]; - crate::user::ensure_user(pool, community, &pubkey) - .await - .expect("ensure test user"); - pubkey - } - - async fn delete_test_community(pool: &PgPool, community: CommunityId) { - for table in ["task_events", "tasks", "users"] { - sqlx::query(sqlx::AssertSqlSafe(format!( - "DELETE FROM {table} WHERE community_id = $1" - ))) - .bind(community.as_uuid()) - .execute(pool) - .await - .expect("delete test rows"); - } - sqlx::query("DELETE FROM communities WHERE id = $1") - .bind(community.as_uuid()) - .execute(pool) - .await - .expect("delete test community"); - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn create_then_list_then_get_round_trips_a_task_and_its_history() { - let pool = setup_pool().await; - let community = make_test_community(&pool).await; - let creator = make_test_user(&pool, community, 0x11).await; - - let created = create_task( - &pool, - community, - NewTask { - created_by_pubkey: Some(creator.clone()), - title: "ship the task system".to_owned(), - body: Some("phase 1".to_owned()), - priority: 5, - source: Some("claude".to_owned()), - ..NewTask::default() - }, - ) - .await - .expect("create task"); - - assert_eq!(created.title, "ship the task system"); - assert_eq!(created.status, TaskStatus::Todo); - assert_eq!(created.priority, 5); - assert_eq!(created.done_at, None); - assert_eq!( - created.created_by_pubkey.as_deref(), - Some(creator.as_slice()) - ); - - let listed = list_tasks( - &pool, - community, - &TaskFilter { - limit: 10, - ..TaskFilter::default() - }, - ) - .await - .expect("list tasks"); - assert_eq!(listed, vec![created.clone()]); - - let fetched = get_task(&pool, community, created.id) - .await - .expect("get task"); - assert_eq!(fetched, created); - - // create_task commits the task and its opening history entry together. - let events = list_task_events(&pool, community, created.id) - .await - .expect("list events"); - assert_eq!(events.len(), 1); - assert_eq!(events[0].action, TaskAction::Created); - - delete_test_community(&pool, community).await; - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn a_task_is_findable_by_its_source_ref() { - let pool = setup_pool().await; - let community = make_test_community(&pool).await; - let creator = make_test_user(&pool, community, 0x21).await; - - let wanted = create_task( - &pool, - community, - NewTask { - created_by_pubkey: Some(creator.clone()), - title: "from the thread we care about".to_owned(), - source: Some("app".to_owned()), - source_ref: Some("thread-head-aaa".to_owned()), - ..NewTask::default() - }, - ) - .await - .expect("create linked task"); - - let other = create_task( - &pool, - community, - NewTask { - created_by_pubkey: Some(creator.clone()), - title: "from a different thread".to_owned(), - source: Some("app".to_owned()), - source_ref: Some("thread-head-bbb".to_owned()), - ..NewTask::default() - }, - ) - .await - .expect("create unrelated task"); - - // Exact equality: the reader queries the same key the writer wrote. - let found = list_tasks( - &pool, - community, - &TaskFilter { - source_ref: Some("thread-head-aaa".to_owned()), - limit: 10, - ..TaskFilter::default() - }, - ) - .await - .expect("list by source_ref"); - assert_eq!(found, vec![wanted.clone()]); - - // An unknown reference is an empty page, never an error and never a - // fallback to "everything". - let missing = list_tasks( - &pool, - community, - &TaskFilter { - source_ref: Some("thread-head-does-not-exist".to_owned()), - limit: 10, - ..TaskFilter::default() - }, - ) - .await - .expect("list unknown source_ref"); - assert!(missing.is_empty()); - - // Omitting the filter must keep today's behaviour: both tasks. - let unfiltered = list_tasks( - &pool, - community, - &TaskFilter { - limit: 10, - ..TaskFilter::default() - }, - ) - .await - .expect("list unfiltered"); - assert_eq!(unfiltered.len(), 2); - assert!(unfiltered.contains(&wanted)); - assert!(unfiltered.contains(&other)); - - delete_test_community(&pool, community).await; - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn a_status_change_sets_done_at_and_appends_exactly_one_event() { - let pool = setup_pool().await; - let community = make_test_community(&pool).await; - let creator = make_test_user(&pool, community, 0x22).await; - - let task = create_task( - &pool, - community, - NewTask { - created_by_pubkey: Some(creator.clone()), - title: "finish it".to_owned(), - ..NewTask::default() - }, - ) - .await - .expect("create task"); - - let done = update_task( - &pool, - community, - task.id, - &TaskPatch { - status: Some(TaskStatus::Done), - ..TaskPatch::default() - }, - Some(&creator), - ) - .await - .expect("mark done"); - assert_eq!(done.status, TaskStatus::Done); - assert!( - done.done_at.is_some(), - "done_at is derived from the status, not supplied by the caller" - ); - - let events = list_task_events(&pool, community, task.id) - .await - .expect("list events"); - assert_eq!(events.len(), 2, "created + status_changed"); - assert_eq!(events[1].action, TaskAction::StatusChanged); - assert_eq!(events[1].from_status, Some(TaskStatus::Todo)); - assert_eq!(events[1].to_status, Some(TaskStatus::Done)); - - // Restating the same status is idempotent: no second event. - update_task( - &pool, - community, - task.id, - &TaskPatch { - status: Some(TaskStatus::Done), - ..TaskPatch::default() - }, - Some(&creator), - ) - .await - .expect("restate done"); - let events = list_task_events(&pool, community, task.id) - .await - .expect("list events again"); - assert_eq!(events.len(), 2, "restating a status must append nothing"); - - // Reopening clears done_at, keeping chk_tasks_done_at_matches_status - // satisfiable. - let reopened = update_task( - &pool, - community, - task.id, - &TaskPatch { - status: Some(TaskStatus::Todo), - ..TaskPatch::default() - }, - Some(&creator), - ) - .await - .expect("reopen"); - assert_eq!(reopened.done_at, None); - - delete_test_community(&pool, community).await; - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn a_task_id_is_invisible_to_another_community() { - let pool = setup_pool().await; - let owner = make_test_community(&pool).await; - let stranger = make_test_community(&pool).await; - let creator = make_test_user(&pool, owner, 0x33).await; - - let task = create_task( - &pool, - owner, - NewTask { - created_by_pubkey: Some(creator), - title: "tenant-private".to_owned(), - ..NewTask::default() - }, - ) - .await - .expect("create task"); - - // The bare id is not a capability: presented against another tenant it - // reads as absent, never as the owner's row. - assert!(matches!( - get_task(&pool, stranger, task.id).await, - Err(DbError::NotFound(_)) - )); - assert!(matches!( - append_task_event( - &pool, - stranger, - task.id, - None, - TaskAction::Commented, - Some("leak?") - ) - .await, - Err(DbError::NotFound(_)) - )); - - delete_test_community(&pool, owner).await; - delete_test_community(&pool, stranger).await; - } - - #[tokio::test] - #[ignore = "requires Postgres"] - async fn a_task_keeps_at_most_one_persisted_summary() { - let pool = setup_pool().await; - let community = make_test_community(&pool).await; - let actor = make_test_user(&pool, community, 0x44).await; - - let task = create_task( - &pool, - community, - NewTask { - created_by_pubkey: Some(actor.clone()), - title: "summarize me".to_owned(), - ..NewTask::default() - }, - ) - .await - .expect("create task"); - - append_task_event( - &pool, - community, - task.id, - Some(&actor), - TaskAction::SummaryPersisted, - Some("first summary"), - ) - .await - .expect("first summary"); - - let second = append_task_event( - &pool, - community, - task.id, - Some(&actor), - TaskAction::SummaryPersisted, - Some("second summary"), - ) - .await; - assert!( - matches!(second, Err(DbError::InvalidData(_))), - "the partial unique index must reject a second summary, got {second:?}" - ); - - // Ordinary comments stay unbounded. - for _ in 0..2 { - append_task_event( - &pool, - community, - task.id, - Some(&actor), - TaskAction::Commented, - Some("a comment"), - ) - .await - .expect("comment"); - } - - delete_test_community(&pool, community).await; - } -} +#[path = "task/postgres_tests.rs"] +mod postgres_tests; diff --git a/crates/buzz-db/src/task/postgres_tests.rs b/crates/buzz-db/src/task/postgres_tests.rs new file mode 100644 index 00000000000..55eaceae7a4 --- /dev/null +++ b/crates/buzz-db/src/task/postgres_tests.rs @@ -0,0 +1,766 @@ +use super::*; +async fn setup_pool() -> PgPool { + PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to test DB") +} + +async fn make_test_community(pool: &PgPool) -> CommunityId { + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(format!("task-test-{}.example", id.simple())) + .execute(pool) + .await + .expect("insert test community"); + CommunityId::from_uuid(id) +} + +async fn make_test_user(pool: &PgPool, community: CommunityId, seed: u8) -> Vec { + let pubkey = vec![seed; 32]; + crate::user::ensure_user(pool, community, &pubkey) + .await + .expect("ensure test user"); + pubkey +} + +async fn delete_test_community(pool: &PgPool, community: CommunityId) { + for table in ["task_events", "tasks", "users"] { + sqlx::query(sqlx::AssertSqlSafe(format!( + "DELETE FROM {table} WHERE community_id = $1" + ))) + .bind(community.as_uuid()) + .execute(pool) + .await + .expect("delete test rows"); + } + sqlx::query("DELETE FROM communities WHERE id = $1") + .bind(community.as_uuid()) + .execute(pool) + .await + .expect("delete test community"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn create_then_list_then_get_round_trips_a_task_and_its_history() { + let pool = setup_pool().await; + let community = make_test_community(&pool).await; + let creator = make_test_user(&pool, community, 0x11).await; + + let created = create_task( + &pool, + community, + NewTask { + created_by_pubkey: Some(creator.clone()), + title: "ship the task system".to_owned(), + body: Some("phase 1".to_owned()), + priority: 5, + source: Some("claude".to_owned()), + ..NewTask::default() + }, + ) + .await + .expect("create task"); + + assert_eq!(created.title, "ship the task system"); + assert_eq!(created.status, TaskStatus::Todo); + assert_eq!(created.priority, 5); + assert_eq!(created.done_at, None); + assert_eq!( + created.created_by_pubkey.as_deref(), + Some(creator.as_slice()) + ); + + let listed = list_tasks( + &pool, + community, + &TaskFilter { + limit: 10, + ..TaskFilter::default() + }, + ) + .await + .expect("list tasks"); + assert_eq!(listed, vec![created.clone()]); + + let fetched = get_task(&pool, community, created.id) + .await + .expect("get task"); + assert_eq!(fetched, created); + + // create_task commits the task and its opening history entry together. + let events = list_task_events(&pool, community, created.id) + .await + .expect("list events"); + assert_eq!(events.len(), 1); + assert_eq!(events[0].action, TaskAction::Created); + + delete_test_community(&pool, community).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_task_is_findable_by_its_source_ref() { + let pool = setup_pool().await; + let community = make_test_community(&pool).await; + let creator = make_test_user(&pool, community, 0x21).await; + + let wanted = create_task( + &pool, + community, + NewTask { + created_by_pubkey: Some(creator.clone()), + title: "from the thread we care about".to_owned(), + source: Some("app".to_owned()), + source_ref: Some("thread-head-aaa".to_owned()), + ..NewTask::default() + }, + ) + .await + .expect("create linked task"); + + let other = create_task( + &pool, + community, + NewTask { + created_by_pubkey: Some(creator.clone()), + title: "from a different thread".to_owned(), + source: Some("app".to_owned()), + source_ref: Some("thread-head-bbb".to_owned()), + ..NewTask::default() + }, + ) + .await + .expect("create unrelated task"); + + // Exact equality: the reader queries the same key the writer wrote. + let found = list_tasks( + &pool, + community, + &TaskFilter { + source_ref: Some("thread-head-aaa".to_owned()), + limit: 10, + ..TaskFilter::default() + }, + ) + .await + .expect("list by source_ref"); + assert_eq!(found, vec![wanted.clone()]); + + // An unknown reference is an empty page, never an error and never a + // fallback to "everything". + let missing = list_tasks( + &pool, + community, + &TaskFilter { + source_ref: Some("thread-head-does-not-exist".to_owned()), + limit: 10, + ..TaskFilter::default() + }, + ) + .await + .expect("list unknown source_ref"); + assert!(missing.is_empty()); + + // Omitting the filter must keep today's behaviour: both tasks. + let unfiltered = list_tasks( + &pool, + community, + &TaskFilter { + limit: 10, + ..TaskFilter::default() + }, + ) + .await + .expect("list unfiltered"); + assert_eq!(unfiltered.len(), 2); + assert!(unfiltered.contains(&wanted)); + assert!(unfiltered.contains(&other)); + + delete_test_community(&pool, community).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_status_change_sets_done_at_and_appends_exactly_one_event() { + let pool = setup_pool().await; + let community = make_test_community(&pool).await; + let creator = make_test_user(&pool, community, 0x22).await; + + let task = create_task( + &pool, + community, + NewTask { + created_by_pubkey: Some(creator.clone()), + title: "finish it".to_owned(), + ..NewTask::default() + }, + ) + .await + .expect("create task"); + + let done = update_task( + &pool, + community, + task.id, + &TaskPatch { + status: Some(TaskStatus::Done), + ..TaskPatch::default() + }, + Some(&creator), + ) + .await + .expect("mark done"); + assert_eq!(done.status, TaskStatus::Done); + assert!( + done.done_at.is_some(), + "done_at is derived from the status, not supplied by the caller" + ); + + let events = list_task_events(&pool, community, task.id) + .await + .expect("list events"); + assert_eq!(events.len(), 2, "created + status_changed"); + assert_eq!(events[1].action, TaskAction::StatusChanged); + assert_eq!(events[1].from_status, Some(TaskStatus::Todo)); + assert_eq!(events[1].to_status, Some(TaskStatus::Done)); + + // Restating the same status is idempotent: no second event. + update_task( + &pool, + community, + task.id, + &TaskPatch { + status: Some(TaskStatus::Done), + ..TaskPatch::default() + }, + Some(&creator), + ) + .await + .expect("restate done"); + let events = list_task_events(&pool, community, task.id) + .await + .expect("list events again"); + assert_eq!(events.len(), 2, "restating a status must append nothing"); + + // Reopening clears done_at, keeping chk_tasks_done_at_matches_status + // satisfiable. + let reopened = update_task( + &pool, + community, + task.id, + &TaskPatch { + status: Some(TaskStatus::Todo), + ..TaskPatch::default() + }, + Some(&creator), + ) + .await + .expect("reopen"); + assert_eq!(reopened.done_at, None); + + delete_test_community(&pool, community).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_task_id_is_invisible_to_another_community() { + let pool = setup_pool().await; + let owner = make_test_community(&pool).await; + let stranger = make_test_community(&pool).await; + let creator = make_test_user(&pool, owner, 0x33).await; + + let task = create_task( + &pool, + owner, + NewTask { + created_by_pubkey: Some(creator), + title: "tenant-private".to_owned(), + ..NewTask::default() + }, + ) + .await + .expect("create task"); + + // The bare id is not a capability: presented against another tenant it + // reads as absent, never as the owner's row. + assert!(matches!( + get_task(&pool, stranger, task.id).await, + Err(DbError::NotFound(_)) + )); + assert!(matches!( + append_task_event( + &pool, + stranger, + task.id, + None, + TaskAction::Commented, + Some("leak?") + ) + .await, + Err(DbError::NotFound(_)) + )); + + delete_test_community(&pool, owner).await; + delete_test_community(&pool, stranger).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_task_keeps_at_most_one_persisted_summary() { + let pool = setup_pool().await; + let community = make_test_community(&pool).await; + let actor = make_test_user(&pool, community, 0x44).await; + + let task = create_task( + &pool, + community, + NewTask { + created_by_pubkey: Some(actor.clone()), + title: "summarize me".to_owned(), + ..NewTask::default() + }, + ) + .await + .expect("create task"); + + append_task_event( + &pool, + community, + task.id, + Some(&actor), + TaskAction::SummaryPersisted, + Some("first summary"), + ) + .await + .expect("first summary"); + + let second = append_task_event( + &pool, + community, + task.id, + Some(&actor), + TaskAction::SummaryPersisted, + Some("second summary"), + ) + .await; + assert!( + matches!(second, Err(DbError::InvalidData(_))), + "the partial unique index must reject a second summary, got {second:?}" + ); + + // Ordinary comments stay unbounded. + for _ in 0..2 { + append_task_event( + &pool, + community, + task.id, + Some(&actor), + TaskAction::Commented, + Some("a comment"), + ) + .await + .expect("comment"); + } + + delete_test_community(&pool, community).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn assignment_and_schedule_history_preserves_before_after_and_noop_retry() { + let pool = setup_pool().await; + let community = make_test_community(&pool).await; + let actor = make_test_user(&pool, community, 0x51).await; + let assignee = make_test_user(&pool, community, 0x52).await; + let due: DateTime = "2026-09-09T14:15:16.123456Z".parse().expect("date"); + let task = create_task( + &pool, + community, + NewTask { + created_by_pubkey: Some(actor.clone()), + title: "original".into(), + ..NewTask::default() + }, + ) + .await + .expect("create"); + let patch = TaskPatch { + title: Some("renamed".into()), + assignee_pubkey: Some(Some(assignee.clone())), + priority: Some(8), + due_at: Some(Some(due + chrono::TimeDelta::nanoseconds(789))), + ..TaskPatch::default() + }; + let updated = update_task(&pool, community, task.id, &patch, Some(&actor)) + .await + .expect("update"); + assert_eq!(updated.assignee_pubkey, Some(assignee.clone())); + assert_eq!(updated.priority, 8); + assert_eq!(updated.due_at, Some(due)); + let events = list_task_events(&pool, community, task.id) + .await + .expect("history"); + assert_eq!(events.len(), 5); + assert_eq!( + events[0].changes.as_ref().expect("initial snapshot")["priority"], + json!({"from": null, "to": 0}) + ); + for (action, changes) in [ + ( + TaskAction::TitleChanged, + json!({"title": {"from": "original", "to": "renamed"}}), + ), + ( + TaskAction::Assigned, + json!({"assignee": {"from": null, "to": hex::encode(&assignee)}}), + ), + ( + TaskAction::PriorityChanged, + json!({"priority": {"from": 0, "to": 8}}), + ), + ( + TaskAction::DueAtChanged, + json!({"due_at": {"from": null, "to": due}}), + ), + ] { + let event = events + .iter() + .find(|e| e.action == action) + .expect("field history"); + assert_eq!(event.changes, Some(changes)); + assert_eq!(event.actor_pubkey, Some(actor.clone())); + } + // A transport retry must change neither history nor pagination order. + let retried = update_task(&pool, community, task.id, &patch, Some(&actor)) + .await + .expect("retry"); + assert_eq!(retried, updated); + assert_eq!( + list_task_events(&pool, community, task.id) + .await + .expect("retry history"), + events + ); + + update_task( + &pool, + community, + task.id, + &TaskPatch { + assignee_pubkey: Some(Some(actor.clone())), + ..TaskPatch::default() + }, + Some(&actor), + ) + .await + .expect("reassign"); + update_task( + &pool, + community, + task.id, + &TaskPatch { + assignee_pubkey: Some(None), + due_at: Some(None), + ..TaskPatch::default() + }, + Some(&actor), + ) + .await + .expect("clear assignment and deadline"); + let events = list_task_events(&pool, community, task.id) + .await + .expect("cleared history"); + assert_eq!( + events[5].changes, + Some(json!({"assignee": {"from": hex::encode(&assignee), "to": hex::encode(&actor)}})) + ); + assert_eq!( + events[6].changes, + Some(json!({"assignee": {"from": hex::encode(&actor), "to": null}})) + ); + assert_eq!( + events[7].changes, + Some(json!({"due_at": {"from": due, "to": null}})) + ); + delete_test_community(&pool, community).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn task_mutation_rolls_back_when_its_history_cannot_be_written() { + let pool = setup_pool().await; + let community = make_test_community(&pool).await; + let actor = make_test_user(&pool, community, 0x53).await; + let task = create_task( + &pool, + community, + NewTask { + created_by_pubkey: Some(actor.clone()), + title: "atomic mutation".into(), + ..NewTask::default() + }, + ) + .await + .expect("create"); + // Deliberately fail the history insert after the task UPDATE. A user FK + // violation is an existing production constraint; no test-only write seam. + let result = update_task( + &pool, + community, + task.id, + &TaskPatch { + priority: Some(99), + ..TaskPatch::default() + }, + Some(&[0xfe; 32]), + ) + .await; + assert!( + result.is_err(), + "invalid audit actor must reject the mutation" + ); + assert_eq!( + get_task(&pool, community, task.id) + .await + .expect("persisted task"), + task + ); + assert_eq!( + list_task_events(&pool, community, task.id) + .await + .expect("history") + .len(), + 1 + ); + delete_test_community(&pool, community).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn concurrent_schedule_updates_record_the_locked_before_image() { + let pool = setup_pool().await; + let community = make_test_community(&pool).await; + let actor = make_test_user(&pool, community, 0x54).await; + let task = create_task( + &pool, + community, + NewTask { + created_by_pubkey: Some(actor.clone()), + title: "concurrent priority".into(), + ..NewTask::default() + }, + ) + .await + .expect("create"); + let first = TaskPatch { + priority: Some(5), + ..TaskPatch::default() + }; + let second = TaskPatch { + priority: Some(7), + ..TaskPatch::default() + }; + let (a, b) = tokio::join!( + update_task(&pool, community, task.id, &first, Some(&actor)), + update_task(&pool, community, task.id, &second, Some(&actor)), + ); + a.expect("first mutation"); + b.expect("second mutation"); + let events = list_task_events(&pool, community, task.id) + .await + .expect("history"); + // The normal history read must preserve the actual mutation order. + let changes: Vec<_> = events + .iter() + .filter(|e| e.action == TaskAction::PriorityChanged) + .map(|e| &e.changes.as_ref().expect("structured history")["priority"]) + .collect(); + assert_eq!(changes.len(), 2); + assert_eq!(changes[0]["from"], 0); + assert_eq!(changes[1]["from"], changes[0]["to"]); + assert_eq!( + changes[1]["to"], + get_task(&pool, community, task.id) + .await + .expect("task") + .priority + ); + delete_test_community(&pool, community).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn visibility_precedes_limit_and_cursor_keeps_equal_timestamp_rows() { + use buzz_core::channel::{ChannelType, ChannelVisibility}; + let pool = setup_pool().await; + let community = make_test_community(&pool).await; + let actor = make_test_user(&pool, community, 0x55).await; + let channel = crate::channel::create_channel( + &pool, + community, + "hidden", + ChannelType::Stream, + ChannelVisibility::Private, + None, + &actor, + None, + ) + .await + .expect("channel"); + let mut visible = Vec::new(); + for title in ["visible one", "visible two", "visible three"] { + visible.push( + create_task( + &pool, + community, + NewTask { + title: title.into(), + ..NewTask::default() + }, + ) + .await + .expect("visible task"), + ); + } + let same_time: DateTime = "2026-09-07T10:11:12.123456Z".parse().expect("timestamp"); + sqlx::query("UPDATE tasks SET updated_at = $2 WHERE community_id = $1") + .bind(community.as_uuid()) + .bind(same_time) + .execute(&pool) + .await + .expect("equal timestamps"); + for _ in 0..3 { + create_task( + &pool, + community, + NewTask { + title: "hidden newer".into(), + channel_id: Some(channel.id), + ..NewTask::default() + }, + ) + .await + .expect("hidden task"); + } + visible.sort_by_key(|task| std::cmp::Reverse(task.id)); + let mut filter = TaskFilter { + visible_channel_ids: Some(vec![]), + limit: 2, + ..TaskFilter::default() + }; + let first = list_tasks(&pool, community, &filter) + .await + .expect("first visible page"); + assert_eq!( + first.iter().map(|t| t.id).collect::>(), + visible[..2].iter().map(|t| t.id).collect::>() + ); + let last = first.last().expect("first page tail"); + filter.before = Some(TaskCursor { + updated_at: last.updated_at, + id: last.id, + }); + let second = list_tasks(&pool, community, &filter) + .await + .expect("second page"); + assert_eq!( + second.iter().map(|t| t.id).collect::>(), + vec![visible[2].id] + ); + filter.before = Some(TaskCursor { + updated_at: second[0].updated_at, + id: second[0].id, + }); + assert!(list_tasks(&pool, community, &filter) + .await + .expect("end of pages") + .is_empty()); + filter.before = None; + filter.visible_channel_ids = Some(vec![channel.id]); + assert!(list_tasks(&pool, community, &filter) + .await + .expect("member page") + .iter() + .all(|t| t.channel_id == Some(channel.id))); + // The per-test database is discarded by the runner, but normal cleanup + // also leaves this scenario reusable in the focused local invocation. + sqlx::query("DELETE FROM task_events WHERE community_id = $1") + .bind(community.as_uuid()) + .execute(&pool) + .await + .expect("events cleanup"); + sqlx::query("DELETE FROM tasks WHERE community_id = $1") + .bind(community.as_uuid()) + .execute(&pool) + .await + .expect("tasks cleanup"); + sqlx::query("DELETE FROM channel_members WHERE community_id = $1") + .bind(community.as_uuid()) + .execute(&pool) + .await + .expect("members cleanup"); + sqlx::query("DELETE FROM channels WHERE community_id = $1") + .bind(community.as_uuid()) + .execute(&pool) + .await + .expect("channels cleanup"); + delete_test_community(&pool, community).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_task_history_upgrade_preserves_legacy_rows() { + let pool = setup_pool().await; + crate::migration::run_migrations_through(&pool, 46) + .await + .expect("legacy migrations"); + let community = make_test_community(&pool).await; + let id: Uuid = sqlx::query_scalar( + "INSERT INTO tasks (community_id, title) VALUES ($1, 'legacy task') RETURNING id", + ) + .bind(community.as_uuid()) + .fetch_one(&pool) + .await + .expect("legacy task"); + sqlx::query( + "INSERT INTO task_events (community_id, task_id, action) VALUES ($1, $2, 'assigned')", + ) + .bind(community.as_uuid()) + .bind(id) + .execute(&pool) + .await + .expect("legacy history"); + crate::migration::run_migrations(&pool) + .await + .expect("upgrade migration"); + let legacy = list_task_events(&pool, community, id) + .await + .expect("read old history"); + assert_eq!(legacy.len(), 1); + assert_eq!( + legacy[0].changes, None, + "must not fabricate past before-images" + ); + update_task( + &pool, + community, + id, + &TaskPatch { + priority: Some(9), + ..TaskPatch::default() + }, + None, + ) + .await + .expect("new write after upgrade"); + let history = list_task_events(&pool, community, id) + .await + .expect("read upgraded history"); + assert_eq!( + history[1].changes, + Some(json!({"priority": {"from": 0, "to": 9}})) + ); + delete_test_community(&pool, community).await; +} diff --git a/crates/buzz-relay/src/api/tasks.rs b/crates/buzz-relay/src/api/tasks.rs index 030a3b91ca8..6308a16f22a 100644 --- a/crates/buzz-relay/src/api/tasks.rs +++ b/crates/buzz-relay/src/api/tasks.rs @@ -21,6 +21,7 @@ use axum::{ http::{HeaderMap, StatusCode}, response::Json, }; +use base64::Engine as _; use chrono::{DateTime, Utc}; use serde::Deserialize; use serde_json::Value; @@ -28,7 +29,7 @@ use uuid::Uuid; use buzz_core::task::{TaskAction, TaskStatus}; use buzz_core::TenantContext; -use buzz_db::task::{NewTask, TaskEventRecord, TaskFilter, TaskPatch, TaskRecord}; +use buzz_db::task::{NewTask, TaskCursor, TaskEventRecord, TaskFilter, TaskPatch, TaskRecord}; use crate::{ api::{api_error, bridge, internal_error}, @@ -48,6 +49,7 @@ pub struct TasksQuery { source_ref: Option, include_archived: Option, limit: Option, + before: Option, } /// Body of `POST /api/tasks`. @@ -315,6 +317,28 @@ pub async fn create_task( Ok(Json(task_json(&task))) } +// Cursor timestamps retain subsecond precision; the task wire format intentionally +// keeps its existing seconds representation for old clients. +fn decode_task_cursor(raw: &str) -> Result)> { + let invalid = || api_error(StatusCode::BAD_REQUEST, "invalid task cursor"); + if raw.len() > 256 { + return Err(invalid()); + } + let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(raw) + .map_err(|_| invalid())?; + serde_json::from_slice(&bytes).map_err(|_| invalid()) +} + +fn encode_task_cursor(task: &TaskRecord) -> Result)> { + let bytes = serde_json::to_vec(&TaskCursor { + updated_at: task.updated_at, + id: task.id, + }) + .map_err(|error| internal_error(&format!("encode task cursor: {error}")))?; + Ok(base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes)) +} + /// `GET /api/tasks` — list this community's tasks, newest-modified first. pub async fn list_tasks( State(state): State>, @@ -329,6 +353,11 @@ pub async fn list_tasks( "limit must be between 1 and 200", )); } + let before = query + .before + .as_deref() + .map(decode_task_cursor) + .transpose()?; let status = query.status.as_deref().map(parse_status).transpose()?; let assignee = query .assignee @@ -350,7 +379,13 @@ pub async fn list_tasks( enforce_channel_access(&state, &tenant, &pubkey, Some(channel_id)).await?; } - let tasks = state + // The visibility predicate must be part of the database query: filtering + // an already-limited page can hide all accessible work behind private tasks. + let accessible = state + .get_accessible_channel_ids_cached(tenant.community(), &pubkey.to_bytes()) + .await + .map_err(|error| internal_error(&format!("task channel access lookup: {error}")))?; + let mut tasks = state .db .list_tasks( tenant.community(), @@ -360,28 +395,25 @@ pub async fn list_tasks( channel_id: query.channel, source_ref: query.source_ref.clone(), include_archived: query.include_archived.unwrap_or(false), - limit, + visible_channel_ids: Some(accessible.into_iter().collect()), + before, + limit: limit + 1, }, ) .await .map_err(|error| map_task_error("list tasks", error))?; - // Channel-bound tasks the caller cannot see are filtered out rather than - // failing the whole page: a list is a view of what you may see. - let accessible = state - .get_accessible_channel_ids_cached(tenant.community(), &pubkey.to_bytes()) - .await - .map_err(|error| internal_error(&format!("task channel access lookup: {error}")))?; - let visible: Vec = tasks - .iter() - .filter(|task| { - task.channel_id - .is_none_or(|channel_id| accessible.contains(&channel_id)) - }) - .map(task_json) - .collect(); - - Ok(Json(serde_json::json!({ "tasks": visible }))) + let has_more = tasks.len() > limit as usize; + tasks.truncate(limit as usize); + let next_cursor = if has_more { + tasks.last().map(encode_task_cursor).transpose()? + } else { + None + }; + let visible: Vec = tasks.iter().map(task_json).collect(); + Ok(Json( + serde_json::json!({ "tasks": visible, "next_cursor": next_cursor }), + )) } /// `GET /api/tasks/{id}` — one task plus its full event history. @@ -562,552 +594,11 @@ fn task_event_json(event: &TaskEventRecord) -> Value { "from_status": event.from_status.map(|status| status.as_str()), "to_status": event.to_status.map(|status| status.as_str()), "body": event.body, + "changes": event.changes, "created_at": event.created_at.timestamp(), }) } #[cfg(test)] -mod tests { - use super::*; - use axum::http::Uri; - - #[test] - fn request_path_preserves_signed_query_verbatim() { - assert_eq!( - request_path("/api/tasks", Some("status=todo&limit=10")), - "/api/tasks?status=todo&limit=10" - ); - assert_eq!(request_path("/api/tasks", None), "/api/tasks"); - assert_eq!(request_path("/api/tasks", Some("")), "/api/tasks"); - } - - #[test] - fn source_ref_survives_verbatim_into_the_signed_path() { - // The client signs the raw query, so the relay must reconstruct it - // byte-for-byte. A normalised or re-ordered `source_ref` would break - // the NIP-98 signature rather than merely filter differently. - let raw = "channel=6f1b0e2c-0000-4000-8000-000000000001&source_ref=abc123"; - assert_eq!( - request_path("/api/tasks", Some(raw)), - format!("/api/tasks?{raw}") - ); - } - - #[test] - fn source_ref_is_parsed_as_an_opaque_optional_string() { - // Opaque TEXT by design (migrations/0046_task_system.sql): the relay - // must not validate it as an event id, and its absence must stay - // distinct from a present value. - fn parse(query: &str) -> TasksQuery { - let uri: Uri = format!("http://relay.invalid/api/tasks?{query}") - .parse() - .expect("valid uri"); - Query::::try_from_uri(&uri).expect("parses").0 - } - - assert_eq!(parse("status=todo").source_ref, None); - assert_eq!( - parse("source_ref=not-an-event-id").source_ref.as_deref(), - Some("not-an-event-id") - ); - } - - #[test] - fn title_length_is_counted_in_characters_not_bytes() { - // 200 multi-byte characters is 600 bytes but a legal title; counting - // bytes here would 400 a request the database would have accepted. - let multibyte = "é".repeat(200); - assert_eq!( - validate_title(&multibyte).expect("200 chars is legal"), - multibyte - ); - assert!(validate_title(&"é".repeat(201)).is_err()); - } - - #[test] - fn title_is_trimmed_and_must_not_be_blank() { - assert_eq!(validate_title(" ship it ").expect("trims"), "ship it"); - assert!(validate_title(" ").is_err()); - assert!(validate_title("").is_err()); - } - - #[test] - fn assignee_must_be_a_32_byte_hex_pubkey() { - let valid = "ab".repeat(32); - assert_eq!( - parse_pubkey("assignee", &valid).expect("valid"), - vec![0xab; 32] - ); - assert!(parse_pubkey("assignee", "not-hex").is_err()); - assert!(parse_pubkey("assignee", &"ab".repeat(31)).is_err()); - assert!(parse_pubkey("assignee", &"ab".repeat(33)).is_err()); - } - - #[test] - fn absent_and_null_assignee_are_different_patches() { - // The whole point of the double option: `{}` leaves the assignee - // alone, `{"assignee": null}` unassigns. - let absent: UpdateTaskRequest = serde_json::from_str("{}").expect("absent"); - assert_eq!(absent.assignee, None); - - let cleared: UpdateTaskRequest = - serde_json::from_str(r#"{"assignee": null}"#).expect("null"); - assert_eq!(cleared.assignee, Some(None)); - - let set: UpdateTaskRequest = serde_json::from_str(r#"{"assignee": "abc"}"#).expect("set"); - assert_eq!(set.assignee, Some(Some("abc".to_owned()))); - } - - #[test] - fn absent_and_null_due_at_are_different_patches() { - let absent: UpdateTaskRequest = serde_json::from_str("{}").expect("absent"); - assert_eq!(absent.due_at, None); - - let cleared: UpdateTaskRequest = serde_json::from_str(r#"{"due_at": null}"#).expect("null"); - assert_eq!(cleared.due_at, Some(None)); - } - - #[test] - fn task_wire_renders_status_and_hex_pubkeys() { - let task = TaskRecord { - id: Uuid::nil(), - channel_id: None, - created_by_pubkey: Some(vec![0xab; 32]), - assignee_pubkey: None, - parent_task_id: None, - title: "ship it".to_owned(), - body: None, - status: TaskStatus::InProgress, - priority: 3, - source: Some("claude".to_owned()), - source_ref: None, - due_at: None, - done_at: None, - archived_at: None, - created_at: Utc::now(), - updated_at: Utc::now(), - }; - let wire = task_json(&task); - assert_eq!(wire["status"], "in_progress"); - assert_eq!(wire["created_by"], hex::encode([0xab; 32])); - assert!(wire["assignee"].is_null()); - assert_eq!(wire["priority"], 3); - // Raw bytes must never reach the wire. - assert!(wire.get("created_by_pubkey").is_none()); - } - - #[test] - fn task_event_wire_renders_both_status_ends() { - let event = TaskEventRecord { - id: 7, - task_id: Uuid::nil(), - actor_pubkey: None, - action: TaskAction::StatusChanged, - from_status: Some(TaskStatus::Todo), - to_status: Some(TaskStatus::Done), - body: None, - created_at: Utc::now(), - }; - let wire = task_event_json(&event); - assert_eq!(wire["action"], "status_changed"); - assert_eq!(wire["from_status"], "todo"); - assert_eq!(wire["to_status"], "done"); - } - - /// Route-level private-channel authorization (COMPAT LANE 3, §7 closure). - /// - /// Drives the REAL router (`build_router` + `oneshot`) with REAL NIP-98 - /// auth headers against a REAL Postgres community containing a private - /// channel and a channel-bound task. Proves at the route seam — not the - /// db seam — that a relay member who is NOT a channel member: - /// * gets 404 (never 403, never the task) on GET/PATCH/POST-events, - /// * gets the task silently filtered out of a channel list, and - /// * cannot even create a task bound to the private channel. - /// - /// The relay-membership gate is exercised with `require_relay_membership - /// = true` so the 404s below are authz verdicts, not gate bypasses. - /// - /// Postgres + Redis are required: run with - /// `cargo test -p buzz-relay --lib api::tasks -- --ignored`. - mod route_authz { - use super::super::*; - use crate::state::AppState; - use buzz_core::channel::{ChannelType, ChannelVisibility}; - use buzz_db::task::NewTask; - use nostr::Keys; - use sha2::{Digest, Sha256}; - - use axum::body::{to_bytes, Body}; - use axum::http::{header, Request, StatusCode}; - use tower::ServiceExt; - - const TEST_DB_URL: &str = "postgres://buzz:***@localhost:5432/buzz"; // sadscan:disable np.postgres.1 - - /// Same trick as the invites tests: the shared AlwaysFreshReplayGuard - /// is gated behind buzz-auth/test-utils, which this crate doesn't - /// enable, so define the pass-through locally. - struct AlwaysFreshReplayGuard; - - impl buzz_auth::Nip98ReplayGuard for AlwaysFreshReplayGuard { - fn try_mark_in_scope<'a>( - &'a self, - _scope: &'a str, - _event_id: &'a nostr::EventId, - _ttl_secs: u64, - ) -> std::pin::Pin< - Box< - dyn std::future::Future> - + Send - + 'a, - >, - > { - Box::pin(async { Ok(true) }) - } - } - - /// Clone of the invites.rs NIP-98 helper: signs kind:27235 over the - /// exact URL the relay will reconstruct (scheme from config.relay_url, - /// host from the tenant, path + raw query). - fn nip98_auth_header(keys: &Keys, method: &str, url: &str, body: &[u8]) -> String { - let hash: [u8; 32] = Sha256::digest(body).into(); - let tags = vec![ - nostr::Tag::parse(["u", url]).expect("u tag"), - nostr::Tag::parse(["method", method]).expect("method tag"), - nostr::Tag::parse(["payload", hex::encode(hash).as_str()]).expect("payload tag"), - ]; - let event = nostr::EventBuilder::new(nostr::Kind::HttpAuth, "") - .tags(tags) - .sign_with_keys(keys) - .expect("sign NIP-98 event"); - let event_json = serde_json::to_string(&event).expect("serialize NIP-98 event"); - let encoded = - base64::Engine::encode(&base64::engine::general_purpose::STANDARD, event_json); - format!("Nostr {encoded}") - } - - #[allow(dead_code)] // AGENT-HOMES-001: shared fixture; fields used by sibling test mods - pub(super) struct Fixture { - state: Arc, - #[allow(dead_code)] - pool: sqlx::PgPool, - host: String, - community: buzz_core::CommunityId, - private_channel_id: Uuid, - task_id: Uuid, - owner: Keys, - outsider: Keys, - } - - /// Boot an AppState bound to a fresh community whose Postgres + Redis - /// are live. Redis must be real: the HTTP admission gate fails closed - /// (503) when the shared limiter is unavailable, which would mask the - /// authorization verdict under test. - pub(super) async fn fixture() -> Option { - let host = format!("task-authz-{}.example", Uuid::new_v4().simple()); - let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") - .or_else(|_| std::env::var("DATABASE_URL")) - .unwrap_or_else(|_| TEST_DB_URL.to_string()); - let redis_url = std::env::var("BUZZ_TEST_REDIS_URL") - .unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()); - - let mut config = crate::config::Config::from_env().ok()?; - config.database_url = database_url.clone(); - config.redis_url = redis_url.clone(); - config.relay_url = format!("wss://{host}"); - config.require_relay_membership = true; - config.require_auth_token = false; - - let pool = sqlx::PgPool::connect(&database_url).await.ok()?; - let db = buzz_db::Db::from_pool(pool.clone()); - let ensured = db.ensure_configured_community(&host).await.ok()?; - - // Live Redis pool for admission + pubsub, mirroring invite tests. - let redis_pool = deadpool_redis::Config::from_url(&redis_url) - .create_pool(Some(deadpool_redis::Runtime::Tokio1)) - .ok()?; - let pubsub = Arc::new( - buzz_pubsub::PubSubManager::new(&redis_url, redis_pool.clone()) - .await - .ok()?, - ); - - let owner = Keys::generate(); - let outsider = Keys::generate(); - let owner_pk = owner.public_key().to_bytes().to_vec(); - let outsider_pk = outsider.public_key().to_bytes().to_vec(); - - // Both are relay members (the outer gate) so every response below - // isolates the CHANNEL gate, not relay membership. - buzz_db::user::ensure_user(&pool, ensured.id, &owner_pk) - .await - .ok()?; - buzz_db::user::ensure_user(&pool, ensured.id, &outsider_pk) - .await - .ok()?; - db.add_relay_member(ensured.id, &owner.public_key().to_hex(), "member", None) - .await - .ok()?; - db.add_relay_member(ensured.id, &outsider.public_key().to_hex(), "member", None) - .await - .ok()?; - - // Private channel owned by `owner` — outsider is not a member. - let channel = buzz_db::channel::create_channel( - &pool, - ensured.id, - "task-authz-private", - ChannelType::Stream, - ChannelVisibility::Private, - None, - &owner_pk, - None, - ) - .await - .ok()?; - - // A task bound to that private channel. - let task = db - .create_task( - ensured.id, - NewTask { - channel_id: Some(channel.id), - created_by_pubkey: Some(owner_pk.clone()), - title: "route authz probe".to_owned(), - ..NewTask::default() - }, - ) - .await - .ok()?; - - let audit = buzz_audit::AuditService::new(pool.clone()); - let auth = buzz_auth::AuthService::new(config.auth.clone()); - let search = buzz_search::SearchService::new(pool.clone()); - let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new( - db.clone(), - buzz_workflow::WorkflowConfig::default(), - )); - let media_storage = buzz_media::MediaStorage::new(&config.media).ok()?; - let (mut state, _audit_shutdown) = AppState::new( - config, - db, - redis_pool, - audit, - pubsub, - auth, - search, - workflow_engine, - Keys::generate(), - media_storage, - ); - state.nip98_replay = Arc::new(AlwaysFreshReplayGuard); - let state = Arc::new(state); - - Some(Fixture { - state, - pool, - host, - community: ensured.id, - private_channel_id: channel.id, - task_id: task.id, - owner, - outsider, - }) - } - - #[allow(dead_code)] // AGENT-HOMES-001: retained for future integration tests - async fn cleanup(f: &Fixture) { - for table in ["task_events", "tasks", "channel_members", "channels"] { - let sql = format!("DELETE FROM {table} WHERE community_id = $1"); - sqlx::query(sqlx::AssertSqlSafe(sql)) - .bind(f.community.as_uuid()) - .execute(&f.pool) - .await - .expect("cleanup channel/task rows"); - } - let _ = f - .state - .db - .remove_relay_member(f.community, &f.outsider.public_key().to_hex()) - .await; - let _ = f - .state - .db - .remove_relay_member(f.community, &f.owner.public_key().to_hex()) - .await; - sqlx::query("DELETE FROM users WHERE community_id = $1") - .bind(f.community.as_uuid()) - .execute(&f.pool) - .await - .expect("cleanup users"); - sqlx::query("DELETE FROM communities WHERE id = $1") - .bind(f.community.as_uuid()) - .execute(&f.pool) - .await - .expect("cleanup community"); - } - - impl Fixture { - async fn request( - &self, - method: &str, - path_and_query: &str, - keys: &Keys, - body: Option<&str>, - ) -> (StatusCode, serde_json::Value) { - let url = format!("https://{}{}", self.host, path_and_query); - let body_bytes = body.map(str::as_bytes).unwrap_or_default(); - let auth = nip98_auth_header(keys, method, &url, body_bytes); - let mut builder = Request::builder() - .method(method) - .uri(path_and_query) - .header(header::HOST, &self.host) - .header(header::AUTHORIZATION, auth); - if body.is_some() { - builder = builder.header(header::CONTENT_TYPE, "application/json"); - } - let response = crate::router::build_router(self.state.clone()) - .oneshot( - builder - .body(Body::from(body_bytes.to_vec())) - .expect("request"), - ) - .await - .expect("response"); - let status = response.status(); - let bytes = to_bytes(response.into_body(), 1024 * 1024) - .await - .expect("read body"); - let json: serde_json::Value = if bytes.is_empty() { - serde_json::Value::Null - } else { - serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null) - }; - (status, json) - } - } - - pub(super) async fn private_channel_assertions(f: &Fixture) { - let task_path = format!("/api/tasks/{}", f.task_id); - - // --- Positive control: the owner sees the task. Without this, 404s - // for the outsider could be any breakage at all. - let (status, body) = f.request("GET", &task_path, &f.owner, None).await; - assert_eq!( - status, - StatusCode::OK, - "owner must see the task; got {status} {body}" - ); - assert_eq!(body["task"]["title"], "route authz probe"); - - // --- GET detail as outsider: 404, never 403, never the task. - let (status, body) = f.request("GET", &task_path, &f.outsider, None).await; - assert_eq!(status, StatusCode::NOT_FOUND, "got {status} {body}"); - assert_eq!(body["error"], "task not found"); - - // --- PATCH as outsider: 404 too. - let (status, body) = f - .request( - "PATCH", - &task_path, - &f.outsider, - Some(r#"{"status":"done"}"#), - ) - .await; - assert_eq!(status, StatusCode::NOT_FOUND, "got {status} {body}"); - - // --- POST comment as outsider: 404. - let (status, body) = f - .request( - "POST", - &format!("{task_path}/events"), - &f.outsider, - Some(r#"{"action":"commented","body":"leak?"}"#), - ) - .await; - assert_eq!(status, StatusCode::NOT_FOUND, "got {status} {body}"); - - // --- Channel-filtered list as outsider: 404, not 200-with-empty. - // The explicit channel filter is itself gated by - // enforce_channel_access (list_tasks), so an outsider cannot even - // probe whether a channel exists — same anti-oracle rule as the - // detail routes. The invisible-channel task is simply unreadable. - let list_path = format!("/api/tasks?channel={}", f.private_channel_id); - let (status, body) = f.request("GET", &list_path, &f.outsider, None).await; - assert_eq!( - status, - StatusCode::NOT_FOUND, - "channel filter must 404 for an invisible channel; got {status} {body}" - ); - - // --- Unfiltered list as outsider: the task must also vanish. - let (status, body) = f.request("GET", "/api/tasks", &f.outsider, None).await; - assert_eq!(status, StatusCode::OK); - let titles: Vec<&str> = body["tasks"] - .as_array() - .map(|tasks| tasks.iter().filter_map(|t| t["title"].as_str()).collect()) - .unwrap_or_default(); - assert!( - !titles.contains(&"route authz probe"), - "task leaked into unfiltered list" - ); - - // --- The owner's list DOES contain it (control for both lists). - let (_, body) = f.request("GET", "/api/tasks", &f.owner, None).await; - let titles: Vec<&str> = body["tasks"] - .as_array() - .map(|tasks| tasks.iter().filter_map(|t| t["title"].as_str()).collect()) - .unwrap_or_default(); - assert!( - titles.contains(&"route authz probe"), - "owner must see the task in the unfiltered list" - ); - - // --- Create bound to the private channel as outsider: 404. - let (status, body) = f - .request( - "POST", - "/api/tasks", - &f.outsider, - Some( - &serde_json::json!({ - "title": "should not exist", - "channel_id": f.private_channel_id, - }) - .to_string(), - ), - ) - .await; - assert_eq!( - status, - StatusCode::NOT_FOUND, - "create must not bind to an invisible channel; got {status} {body}" - ); - } - } - - mod postgres_tests { - use super::route_authz::{fixture, private_channel_assertions}; - - /// The single route-level scenario: a relay member outside a private - /// channel must receive 404 (not 403, not data) on every task route, - /// and the channel-bound task must vanish from listings. The owner's - /// positive control proves the 404s are authz, not breakage. - #[tokio::test] - #[ignore = "requires Postgres"] - async fn private_channel_task_is_invisible_to_non_members_at_the_route() { - let Some(f) = fixture().await else { - eprintln!("SKIP: Postgres/Redis unavailable"); - return; - }; - // Catch assertion panics so cleanup ALWAYS runs, then resume them. - let result = futures::FutureExt::catch_unwind(std::panic::AssertUnwindSafe( - private_channel_assertions(&f), - )) - .await; - drop(f); - if let Err(payload) = result { - std::panic::resume_unwind(payload); - } - } - } -} +#[path = "tasks/tests.rs"] +mod tests; diff --git a/crates/buzz-relay/src/api/tasks/tests.rs b/crates/buzz-relay/src/api/tasks/tests.rs new file mode 100644 index 00000000000..4d6fd116232 --- /dev/null +++ b/crates/buzz-relay/src/api/tasks/tests.rs @@ -0,0 +1,676 @@ +use super::*; +use axum::http::Uri; + +#[test] +fn request_path_preserves_signed_query_verbatim() { + assert_eq!( + request_path("/api/tasks", Some("status=todo&limit=10")), + "/api/tasks?status=todo&limit=10" + ); + assert_eq!(request_path("/api/tasks", None), "/api/tasks"); + assert_eq!(request_path("/api/tasks", Some("")), "/api/tasks"); +} + +#[test] +fn source_ref_survives_verbatim_into_the_signed_path() { + // The client signs the raw query, so the relay must reconstruct it + // byte-for-byte. A normalised or re-ordered `source_ref` would break + // the NIP-98 signature rather than merely filter differently. + let raw = "channel=6f1b0e2c-0000-4000-8000-000000000001&source_ref=abc123"; + assert_eq!( + request_path("/api/tasks", Some(raw)), + format!("/api/tasks?{raw}") + ); +} + +#[test] +fn source_ref_is_parsed_as_an_opaque_optional_string() { + // Opaque TEXT by design (migrations/0046_task_system.sql): the relay + // must not validate it as an event id, and its absence must stay + // distinct from a present value. + fn parse(query: &str) -> TasksQuery { + let uri: Uri = format!("http://relay.invalid/api/tasks?{query}") + .parse() + .expect("valid uri"); + Query::::try_from_uri(&uri).expect("parses").0 + } + + assert_eq!(parse("status=todo").source_ref, None); + assert_eq!( + parse("source_ref=not-an-event-id").source_ref.as_deref(), + Some("not-an-event-id") + ); +} + +#[test] +fn title_length_is_counted_in_characters_not_bytes() { + // 200 multi-byte characters is 600 bytes but a legal title; counting + // bytes here would 400 a request the database would have accepted. + let multibyte = "é".repeat(200); + assert_eq!( + validate_title(&multibyte).expect("200 chars is legal"), + multibyte + ); + assert!(validate_title(&"é".repeat(201)).is_err()); +} + +#[test] +fn title_is_trimmed_and_must_not_be_blank() { + assert_eq!(validate_title(" ship it ").expect("trims"), "ship it"); + assert!(validate_title(" ").is_err()); + assert!(validate_title("").is_err()); +} + +#[test] +fn assignee_must_be_a_32_byte_hex_pubkey() { + let valid = "ab".repeat(32); + assert_eq!( + parse_pubkey("assignee", &valid).expect("valid"), + vec![0xab; 32] + ); + assert!(parse_pubkey("assignee", "not-hex").is_err()); + assert!(parse_pubkey("assignee", &"ab".repeat(31)).is_err()); + assert!(parse_pubkey("assignee", &"ab".repeat(33)).is_err()); +} + +#[test] +fn absent_and_null_assignee_are_different_patches() { + // The whole point of the double option: `{}` leaves the assignee + // alone, `{"assignee": null}` unassigns. + let absent: UpdateTaskRequest = serde_json::from_str("{}").expect("absent"); + assert_eq!(absent.assignee, None); + + let cleared: UpdateTaskRequest = serde_json::from_str(r#"{"assignee": null}"#).expect("null"); + assert_eq!(cleared.assignee, Some(None)); + + let set: UpdateTaskRequest = serde_json::from_str(r#"{"assignee": "abc"}"#).expect("set"); + assert_eq!(set.assignee, Some(Some("abc".to_owned()))); +} + +#[test] +fn absent_and_null_due_at_are_different_patches() { + let absent: UpdateTaskRequest = serde_json::from_str("{}").expect("absent"); + assert_eq!(absent.due_at, None); + + let cleared: UpdateTaskRequest = serde_json::from_str(r#"{"due_at": null}"#).expect("null"); + assert_eq!(cleared.due_at, Some(None)); +} + +#[test] +fn task_wire_renders_status_and_hex_pubkeys() { + let task = TaskRecord { + id: Uuid::nil(), + channel_id: None, + created_by_pubkey: Some(vec![0xab; 32]), + assignee_pubkey: None, + parent_task_id: None, + title: "ship it".to_owned(), + body: None, + status: TaskStatus::InProgress, + priority: 3, + source: Some("claude".to_owned()), + source_ref: None, + due_at: None, + done_at: None, + archived_at: None, + created_at: Utc::now(), + updated_at: Utc::now(), + }; + let wire = task_json(&task); + assert_eq!(wire["status"], "in_progress"); + assert_eq!(wire["created_by"], hex::encode([0xab; 32])); + assert!(wire["assignee"].is_null()); + assert_eq!(wire["priority"], 3); + // Raw bytes must never reach the wire. + assert!(wire.get("created_by_pubkey").is_none()); +} + +#[test] +fn task_event_wire_renders_both_status_ends() { + let event = TaskEventRecord { + id: 7, + task_id: Uuid::nil(), + actor_pubkey: None, + action: TaskAction::StatusChanged, + changes: None, + from_status: Some(TaskStatus::Todo), + to_status: Some(TaskStatus::Done), + body: None, + created_at: Utc::now(), + }; + let wire = task_event_json(&event); + assert_eq!(wire["action"], "status_changed"); + assert_eq!(wire["from_status"], "todo"); + assert_eq!(wire["to_status"], "done"); +} + +/// Route-level private-channel authorization (COMPAT LANE 3, §7 closure). +/// +/// Drives the REAL router (`build_router` + `oneshot`) with REAL NIP-98 +/// auth headers against a REAL Postgres community containing a private +/// channel and a channel-bound task. Proves at the route seam — not the +/// db seam — that a relay member who is NOT a channel member: +/// * gets 404 (never 403, never the task) on GET/PATCH/POST-events, +/// * gets the task silently filtered out of a channel list, and +/// * cannot even create a task bound to the private channel. +/// +/// The relay-membership gate is exercised with `require_relay_membership +/// = true` so the 404s below are authz verdicts, not gate bypasses. +/// +/// Postgres + Redis are required: run with +/// `cargo test -p buzz-relay --lib api::tasks -- --ignored`. +mod route_authz { + use super::super::*; + use crate::state::AppState; + use buzz_core::channel::{ChannelType, ChannelVisibility}; + use buzz_db::task::NewTask; + use nostr::Keys; + use sha2::{Digest, Sha256}; + + use axum::body::{to_bytes, Body}; + use axum::http::{header, Request, StatusCode}; + use tower::ServiceExt; + + const TEST_DB_URL: &str = "postgres://buzz:***@localhost:5432/buzz"; // sadscan:disable np.postgres.1 + + /// Same trick as the invites tests: the shared AlwaysFreshReplayGuard + /// is gated behind buzz-auth/test-utils, which this crate doesn't + /// enable, so define the pass-through locally. + struct AlwaysFreshReplayGuard; + + impl buzz_auth::Nip98ReplayGuard for AlwaysFreshReplayGuard { + fn try_mark_in_scope<'a>( + &'a self, + _scope: &'a str, + _event_id: &'a nostr::EventId, + _ttl_secs: u64, + ) -> std::pin::Pin< + Box> + Send + 'a>, + > { + Box::pin(async { Ok(true) }) + } + } + + /// Clone of the invites.rs NIP-98 helper: signs kind:27235 over the + /// exact URL the relay will reconstruct (scheme from config.relay_url, + /// host from the tenant, path + raw query). + fn nip98_auth_header(keys: &Keys, method: &str, url: &str, body: &[u8]) -> String { + let hash: [u8; 32] = Sha256::digest(body).into(); + let tags = vec![ + nostr::Tag::parse(["u", url]).expect("u tag"), + nostr::Tag::parse(["method", method]).expect("method tag"), + nostr::Tag::parse(["payload", hex::encode(hash).as_str()]).expect("payload tag"), + ]; + let event = nostr::EventBuilder::new(nostr::Kind::HttpAuth, "") + .tags(tags) + .sign_with_keys(keys) + .expect("sign NIP-98 event"); + let event_json = serde_json::to_string(&event).expect("serialize NIP-98 event"); + let encoded = + base64::Engine::encode(&base64::engine::general_purpose::STANDARD, event_json); + format!("Nostr {encoded}") + } + + #[allow(dead_code)] // AGENT-HOMES-001: shared fixture; fields used by sibling test mods + pub(super) struct Fixture { + state: Arc, + #[allow(dead_code)] + pool: sqlx::PgPool, + host: String, + community: buzz_core::CommunityId, + private_channel_id: Uuid, + task_id: Uuid, + owner: Keys, + outsider: Keys, + pub(super) http_base: Option, + } + + /// Boot an AppState bound to a fresh community whose Postgres + Redis + /// are live. Redis must be real: the HTTP admission gate fails closed + /// (503) when the shared limiter is unavailable, which would mask the + /// authorization verdict under test. + pub(super) async fn fixture() -> Option { + let host = format!("task-authz-{}.example", Uuid::new_v4().simple()); + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_string()); + let redis_url = std::env::var("BUZZ_TEST_REDIS_URL") + .unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()); + + let mut config = crate::config::Config::from_env().ok()?; + config.database_url = database_url.clone(); + config.redis_url = redis_url.clone(); + config.relay_url = format!("wss://{host}"); + config.require_relay_membership = true; + config.require_auth_token = false; + + let pool = sqlx::PgPool::connect(&database_url).await.ok()?; + let db = buzz_db::Db::from_pool(pool.clone()); + let ensured = db.ensure_configured_community(&host).await.ok()?; + + // Live Redis pool for admission + pubsub, mirroring invite tests. + let redis_pool = deadpool_redis::Config::from_url(&redis_url) + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .ok()?; + let pubsub = Arc::new( + buzz_pubsub::PubSubManager::new(&redis_url, redis_pool.clone()) + .await + .ok()?, + ); + + let owner = Keys::generate(); + let outsider = Keys::generate(); + let owner_pk = owner.public_key().to_bytes().to_vec(); + let outsider_pk = outsider.public_key().to_bytes().to_vec(); + + // Both are relay members (the outer gate) so every response below + // isolates the CHANNEL gate, not relay membership. + buzz_db::user::ensure_user(&pool, ensured.id, &owner_pk) + .await + .ok()?; + buzz_db::user::ensure_user(&pool, ensured.id, &outsider_pk) + .await + .ok()?; + db.add_relay_member(ensured.id, &owner.public_key().to_hex(), "member", None) + .await + .ok()?; + db.add_relay_member(ensured.id, &outsider.public_key().to_hex(), "member", None) + .await + .ok()?; + + // Private channel owned by `owner` — outsider is not a member. + let channel = buzz_db::channel::create_channel( + &pool, + ensured.id, + "task-authz-private", + ChannelType::Stream, + ChannelVisibility::Private, + None, + &owner_pk, + None, + ) + .await + .ok()?; + + // A task bound to that private channel. + let task = db + .create_task( + ensured.id, + NewTask { + channel_id: Some(channel.id), + created_by_pubkey: Some(owner_pk.clone()), + title: "route authz probe".to_owned(), + ..NewTask::default() + }, + ) + .await + .ok()?; + + let audit = buzz_audit::AuditService::new(pool.clone()); + let auth = buzz_auth::AuthService::new(config.auth.clone()); + let search = buzz_search::SearchService::new(pool.clone()); + let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new( + db.clone(), + buzz_workflow::WorkflowConfig::default(), + )); + let media_storage = buzz_media::MediaStorage::new(&config.media).ok()?; + let (mut state, _audit_shutdown) = AppState::new( + config, + db, + redis_pool, + audit, + pubsub, + auth, + search, + workflow_engine, + Keys::generate(), + media_storage, + ); + state.nip98_replay = Arc::new(AlwaysFreshReplayGuard); + let state = Arc::new(state); + + Some(Fixture { + state, + pool, + host, + community: ensured.id, + private_channel_id: channel.id, + task_id: task.id, + owner, + outsider, + http_base: None, + }) + } + + #[allow(dead_code)] // AGENT-HOMES-001: retained for future integration tests + async fn cleanup(f: &Fixture) { + for table in ["task_events", "tasks", "channel_members", "channels"] { + let sql = format!("DELETE FROM {table} WHERE community_id = $1"); + sqlx::query(sqlx::AssertSqlSafe(sql)) + .bind(f.community.as_uuid()) + .execute(&f.pool) + .await + .expect("cleanup channel/task rows"); + } + let _ = f + .state + .db + .remove_relay_member(f.community, &f.outsider.public_key().to_hex()) + .await; + let _ = f + .state + .db + .remove_relay_member(f.community, &f.owner.public_key().to_hex()) + .await; + sqlx::query("DELETE FROM users WHERE community_id = $1") + .bind(f.community.as_uuid()) + .execute(&f.pool) + .await + .expect("cleanup users"); + sqlx::query("DELETE FROM communities WHERE id = $1") + .bind(f.community.as_uuid()) + .execute(&f.pool) + .await + .expect("cleanup community"); + } + + impl Fixture { + pub(super) fn router(&self) -> axum::Router { + crate::router::build_router(self.state.clone()) + } + + async fn request( + &self, + method: &str, + path_and_query: &str, + keys: &Keys, + body: Option<&str>, + ) -> (StatusCode, serde_json::Value) { + let url = format!("https://{}{}", self.host, path_and_query); + let body_bytes = body.map(str::as_bytes).unwrap_or_default(); + let auth = nip98_auth_header(keys, method, &url, body_bytes); + if let Some(base) = &self.http_base { + let client = reqwest::Client::new(); + let response = client + .request( + method.parse().expect("method"), + format!("{base}{path_and_query}"), + ) + .header(header::HOST, &self.host) + .header(header::AUTHORIZATION, auth) + .header(header::CONTENT_TYPE, "application/json") + .body(body_bytes.to_vec()) + .send() + .await + .expect("HTTP response"); + let status = response.status(); + let json = response.json().await.expect("HTTP JSON response"); + return (status, json); + } + let mut builder = Request::builder() + .method(method) + .uri(path_and_query) + .header(header::HOST, &self.host) + .header(header::AUTHORIZATION, auth); + if body.is_some() { + builder = builder.header(header::CONTENT_TYPE, "application/json"); + } + let response = crate::router::build_router(self.state.clone()) + .oneshot( + builder + .body(Body::from(body_bytes.to_vec())) + .expect("request"), + ) + .await + .expect("response"); + let status = response.status(); + let bytes = to_bytes(response.into_body(), 1024 * 1024) + .await + .expect("read body"); + let json: serde_json::Value = if bytes.is_empty() { + serde_json::Value::Null + } else { + serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null) + }; + (status, json) + } + } + + pub(super) async fn pagination_and_history_assertions(f: &Fixture) { + let mut ids = Vec::new(); + for title in ["visible one", "visible two", "visible three"] { + let payload = serde_json::json!({"title": title}).to_string(); + let (status, body) = f + .request("POST", "/api/tasks", &f.owner, Some(&payload)) + .await; + assert_eq!(status, StatusCode::OK, "create: {body}"); + ids.push(body["id"].as_str().expect("task id").to_owned()); + } + // Tie timestamps with subsecond precision to exercise the id part of + // the cursor and its wire encoding, behind a newer invisible task. + sqlx::query("UPDATE tasks SET updated_at = '2026-01-01T00:00:00.123456Z' WHERE community_id = $1 AND channel_id IS NULL") + .bind(f.community.as_uuid()).execute(&f.pool).await.expect("fixture timestamps"); + ids.sort_by(|a, b| b.cmp(a)); + let (status, first) = f + .request("GET", "/api/tasks?limit=2", &f.outsider, None) + .await; + assert_eq!(status, StatusCode::OK, "first page: {first}"); + let rows = first["tasks"].as_array().expect("tasks array"); + assert_eq!( + rows.len(), + 2, + "invisible newer tasks must not consume the limit" + ); + assert_eq!(rows[0]["id"], ids[0]); + assert_eq!(rows[1]["id"], ids[1]); + let cursor = first["next_cursor"].as_str().expect("next cursor"); + let path = format!("/api/tasks?limit=2&before={cursor}"); + let (status, second) = f.request("GET", &path, &f.outsider, None).await; + assert_eq!(status, StatusCode::OK, "second page: {second}"); + assert_eq!(second["tasks"].as_array().expect("second tasks").len(), 1); + assert_eq!(second["tasks"][0]["id"], ids[2]); + assert!(second["next_cursor"].is_null()); + let (status, _) = f + .request("GET", "/api/tasks?before=invalid", &f.owner, None) + .await; + assert_eq!(status, StatusCode::BAD_REQUEST); + + let task_path = format!("/api/tasks/{}", ids[0]); + let payload = serde_json::json!({ + "assignee": f.owner.public_key().to_hex(), "priority": 4, + "due_at": "2026-09-10T11:12:13.123456Z", + }) + .to_string(); + let (status, updated) = f + .request("PATCH", &task_path, &f.owner, Some(&payload)) + .await; + assert_eq!(status, StatusCode::OK, "patch: {updated}"); + assert_eq!(updated["priority"], 4); + let (status, detail) = f.request("GET", &task_path, &f.owner, None).await; + assert_eq!(status, StatusCode::OK, "detail: {detail}"); + let events = detail["events"].as_array().expect("event history"); + assert_eq!(events.len(), 4); + for (action, expected) in [ + ( + "assigned", + serde_json::json!({"assignee": {"from": null, "to": f.owner.public_key().to_hex()}}), + ), + ( + "priority_changed", + serde_json::json!({"priority": {"from": 0, "to": 4}}), + ), + ( + "due_at_changed", + serde_json::json!({"due_at": {"from": null, "to": "2026-09-10T11:12:13.123456Z"}}), + ), + ] { + let event = events + .iter() + .find(|event| event["action"] == action) + .expect("change event"); + assert_eq!(event["changes"], expected); + assert_eq!(event["actor"], f.owner.public_key().to_hex()); + } + let (status, _) = f + .request("PATCH", &task_path, &f.owner, Some(&payload)) + .await; + assert_eq!(status, StatusCode::OK); + let (_, retry) = f.request("GET", &task_path, &f.owner, None).await; + assert_eq!( + retry["events"], detail["events"], + "signed retry must not append duplicate history" + ); + eprintln!("PASS signed task create/update/history; private tasks excluded before limit; cursor pages 2+1; retry unchanged"); + } + + pub(super) async fn private_channel_assertions(f: &Fixture) { + let task_path = format!("/api/tasks/{}", f.task_id); + + // --- Positive control: the owner sees the task. Without this, 404s + // for the outsider could be any breakage at all. + let (status, body) = f.request("GET", &task_path, &f.owner, None).await; + assert_eq!( + status, + StatusCode::OK, + "owner must see the task; got {status} {body}" + ); + assert_eq!(body["task"]["title"], "route authz probe"); + + // --- GET detail as outsider: 404, never 403, never the task. + let (status, body) = f.request("GET", &task_path, &f.outsider, None).await; + assert_eq!(status, StatusCode::NOT_FOUND, "got {status} {body}"); + assert_eq!(body["error"], "task not found"); + + // --- PATCH as outsider: 404 too. + let (status, body) = f + .request( + "PATCH", + &task_path, + &f.outsider, + Some(r#"{"status":"done"}"#), + ) + .await; + assert_eq!(status, StatusCode::NOT_FOUND, "got {status} {body}"); + + // --- POST comment as outsider: 404. + let (status, body) = f + .request( + "POST", + &format!("{task_path}/events"), + &f.outsider, + Some(r#"{"action":"commented","body":"leak?"}"#), + ) + .await; + assert_eq!(status, StatusCode::NOT_FOUND, "got {status} {body}"); + + // --- Channel-filtered list as outsider: 404, not 200-with-empty. + // The explicit channel filter is itself gated by + // enforce_channel_access (list_tasks), so an outsider cannot even + // probe whether a channel exists — same anti-oracle rule as the + // detail routes. The invisible-channel task is simply unreadable. + let list_path = format!("/api/tasks?channel={}", f.private_channel_id); + let (status, body) = f.request("GET", &list_path, &f.outsider, None).await; + assert_eq!( + status, + StatusCode::NOT_FOUND, + "channel filter must 404 for an invisible channel; got {status} {body}" + ); + + // --- Unfiltered list as outsider: the task must also vanish. + let (status, body) = f.request("GET", "/api/tasks", &f.outsider, None).await; + assert_eq!(status, StatusCode::OK); + let titles: Vec<&str> = body["tasks"] + .as_array() + .map(|tasks| tasks.iter().filter_map(|t| t["title"].as_str()).collect()) + .unwrap_or_default(); + assert!( + !titles.contains(&"route authz probe"), + "task leaked into unfiltered list" + ); + + // --- The owner's list DOES contain it (control for both lists). + let (_, body) = f.request("GET", "/api/tasks", &f.owner, None).await; + let titles: Vec<&str> = body["tasks"] + .as_array() + .map(|tasks| tasks.iter().filter_map(|t| t["title"].as_str()).collect()) + .unwrap_or_default(); + assert!( + titles.contains(&"route authz probe"), + "owner must see the task in the unfiltered list" + ); + + // --- Create bound to the private channel as outsider: 404. + let (status, body) = f + .request( + "POST", + "/api/tasks", + &f.outsider, + Some( + &serde_json::json!({ + "title": "should not exist", + "channel_id": f.private_channel_id, + }) + .to_string(), + ), + ) + .await; + assert_eq!( + status, + StatusCode::NOT_FOUND, + "create must not bind to an invisible channel; got {status} {body}" + ); + } +} + +mod postgres_tests { + use super::route_authz::{ + fixture, pagination_and_history_assertions, private_channel_assertions, + }; + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn signed_task_routes_page_visible_work_and_expose_durable_changes() { + let mut f = fixture() + .await + .expect("Postgres and Redis fixture must be available"); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("test HTTP listener"); + f.http_base = Some(format!( + "http://{}", + listener.local_addr().expect("bound address") + )); + let router = f.router(); + struct Server(tokio::task::JoinHandle<()>); + impl Drop for Server { + fn drop(&mut self) { + self.0.abort(); + } + } + let _server = Server(tokio::spawn(async move { + axum::serve(listener, router).await.expect("HTTP server"); + })); + pagination_and_history_assertions(&f).await; + } + + /// The single route-level scenario: a relay member outside a private + /// channel must receive 404 (not 403, not data) on every task route, + /// and the channel-bound task must vanish from listings. The owner's + /// positive control proves the 404s are authz, not breakage. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn private_channel_task_is_invisible_to_non_members_at_the_route() { + let f = fixture() + .await + .expect("Postgres and Redis fixture must be available"); + // Catch assertion panics so cleanup ALWAYS runs, then resume them. + let result = futures::FutureExt::catch_unwind(std::panic::AssertUnwindSafe( + private_channel_assertions(&f), + )) + .await; + drop(f); + if let Err(payload) = result { + std::panic::resume_unwind(payload); + } + } +} diff --git a/migrations/0047_task_event_changes.sql b/migrations/0047_task_event_changes.sql new file mode 100644 index 00000000000..bf4b25a06ad --- /dev/null +++ b/migrations/0047_task_event_changes.sql @@ -0,0 +1,8 @@ +-- Preserve structured before/after task changes without changing legacy rows. +-- Deploy readers that understand the new action strings before new writers. +ALTER TABLE task_events ADD COLUMN changes JSONB + CHECK (changes IS NULL OR jsonb_typeof(changes) = 'object'); + +-- Transaction start can precede another writer's commit while waiting on a +-- task row lock. Timestamp the actual append so history remains chronological. +ALTER TABLE task_events ALTER COLUMN created_at SET DEFAULT clock_timestamp(); diff --git a/schema/schema.sql b/schema/schema.sql index a741e1cfb6d..a30346aae51 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -1263,7 +1263,8 @@ CREATE TABLE task_events ( from_status TEXT, to_status TEXT, body TEXT, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + changes JSONB CHECK (changes IS NULL OR jsonb_typeof(changes) = 'object'), + created_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(), PRIMARY KEY (community_id, id), CONSTRAINT chk_task_events_actor_len CHECK (actor_pubkey IS NULL OR length(actor_pubkey) = 32), From 8a35c1f14ae5f618ffbf416b1a6d1786e7b46f86 Mon Sep 17 00:00:00 2001 From: Michael Feth Date: Mon, 7 Sep 2026 11:33:05 -0400 Subject: [PATCH 2/2] test(db): pin additive task history migration inventory Signed-off-by: Michael Feth --- crates/buzz-db/src/runtime/migration.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index fbc8c79e98b..7923e61736c 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -703,8 +703,15 @@ mod postgres_tests { migrations.sort_by_key(|migration| migration.version); // upstream carries 44 (0032-0034 and 0040 adopted from our PRs); - // fork adds 0046_task_system (PR #6425 pending upstream). - assert_eq!(migrations.len(), 45); + // fork adds 0046_task_system (PR #6425 pending upstream) and 0047 + // structured task history. Both stay additive for existing deployments. + assert_eq!(migrations.len(), 46); + assert_eq!(migrations[44].version, 46); + assert_eq!(migrations[45].version, 47); + let task_changes = migrations[45].sql.as_str(); + assert!(task_changes.contains("ALTER TABLE task_events ADD COLUMN changes JSONB")); + assert!(task_changes.contains("ALTER COLUMN created_at SET DEFAULT clock_timestamp()")); + assert!(!migrations[44].sql.as_str().contains("ADD COLUMN changes")); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0]