diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e8da259ea..8d0637846 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -133,6 +133,14 @@ jobs: wait_healthy "Postgres" "sprout-postgres" wait_healthy "Redis" "sprout-redis" wait_healthy "Typesense" "sprout-typesense" + - name: Apply database schema + run: ./bin/pgschema apply --file schema/schema.sql --auto-approve + env: + PGHOST: localhost + PGPORT: "5432" + PGUSER: sprout + PGPASSWORD: sprout_dev + PGDATABASE: sprout - name: Build relay run: cargo build -p sprout-relay - name: Start relay diff --git a/crates/sprout-db/src/channel.rs b/crates/sprout-db/src/channel.rs index af68fe8f9..23f2dfe90 100644 --- a/crates/sprout-db/src/channel.rs +++ b/crates/sprout-db/src/channel.rs @@ -9,7 +9,6 @@ use sqlx::{PgPool, Postgres, Row, Transaction}; use uuid::Uuid; use crate::error::{DbError, Result}; -use crate::event::uuid_from_bytes; /// Whether a channel is publicly visible or invite-only. #[derive(Debug, Clone, PartialEq, Eq)] @@ -214,7 +213,6 @@ pub async fn create_channel( } let id = Uuid::new_v4(); - let id_bytes = id.as_bytes().as_slice().to_vec(); let mut tx = pool.begin().await?; @@ -224,7 +222,7 @@ pub async fn create_channel( VALUES ($1, $2, $3, $4, $5, $6) "#, ) - .bind(&id_bytes) + .bind(id) .bind(name) .bind(channel_type.as_str()) .bind(visibility.as_str()) @@ -243,7 +241,7 @@ pub async fn create_channel( role = EXCLUDED.role "#, ) - .bind(&id_bytes) + .bind(id) .bind(created_by) .bind(created_by) .execute(&mut *tx) @@ -259,7 +257,7 @@ pub async fn create_channel( FROM channels WHERE id = $1 "#, ) - .bind(&id_bytes) + .bind(id) .fetch_one(&mut *tx) .await?; @@ -270,8 +268,6 @@ pub async fn create_channel( /// Fetches a channel record by ID. Returns `ChannelNotFound` if missing or deleted. pub async fn get_channel(pool: &PgPool, channel_id: Uuid) -> Result { - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); - let row = sqlx::query( r#" SELECT id, name, channel_type, visibility, description, canvas, @@ -282,7 +278,7 @@ pub async fn get_channel(pool: &PgPool, channel_id: Uuid) -> Result Result Result> { - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); let row = sqlx::query("SELECT canvas FROM channels WHERE id = $1 AND deleted_at IS NULL") - .bind(&id_bytes) + .bind(channel_id) .fetch_optional(pool) .await? .ok_or(DbError::ChannelNotFound(channel_id))?; @@ -303,10 +298,9 @@ pub async fn get_canvas(pool: &PgPool, channel_id: Uuid) -> Result) -> Result<()> { - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); let rows = sqlx::query("UPDATE channels SET canvas = $1 WHERE id = $2 AND deleted_at IS NULL") .bind(canvas) - .bind(&id_bytes) + .bind(channel_id) .execute(pool) .await?; if rows.rows_affected() == 0 { @@ -340,8 +334,6 @@ pub async fn add_member( ))); } - let channel_id_bytes = channel_id.as_bytes().as_slice().to_vec(); - let mut tx = pool.begin().await?; let channel = get_channel_tx(&mut tx, channel_id).await?; @@ -411,7 +403,7 @@ pub async fn add_member( role = EXCLUDED.role "#, ) - .bind(&channel_id_bytes) + .bind(channel_id) .bind(pubkey) .bind(effective_role.as_str()) .bind(invited_by) @@ -424,7 +416,7 @@ pub async fn add_member( FROM channel_members WHERE channel_id = $1 AND pubkey = $2 "#, ) - .bind(&channel_id_bytes) + .bind(channel_id) .bind(pubkey) .fetch_one(&mut *tx) .await?; @@ -447,8 +439,6 @@ pub async fn remove_member( pubkey: &[u8], actor_pubkey: &[u8], ) -> Result<()> { - let channel_id_bytes = channel_id.as_bytes().as_slice().to_vec(); - let mut tx = pool.begin().await?; let is_self_remove = pubkey == actor_pubkey; @@ -475,7 +465,7 @@ pub async fn remove_member( "SELECT COUNT(*) as cnt FROM channel_members \ WHERE channel_id = $1 AND role = 'owner' AND removed_at IS NULL", ) - .bind(&channel_id_bytes) + .bind(channel_id) .fetch_one(&mut *tx) .await?; let owner_count: i64 = row.try_get("cnt")?; @@ -494,7 +484,7 @@ pub async fn remove_member( "#, ) .bind(actor_pubkey) - .bind(&channel_id_bytes) + .bind(channel_id) .bind(pubkey) .execute(&mut *tx) .await?; @@ -509,13 +499,12 @@ pub async fn remove_member( /// Returns `true` if the given pubkey is an active member of the channel. pub async fn is_member(pool: &PgPool, channel_id: Uuid, pubkey: &[u8]) -> Result { - let channel_id_bytes = channel_id.as_bytes().as_slice().to_vec(); let row = sqlx::query( "SELECT COUNT(*) as cnt FROM channel_members cm \ JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL \ WHERE cm.channel_id = $1 AND cm.pubkey = $2 AND cm.removed_at IS NULL", ) - .bind(&channel_id_bytes) + .bind(channel_id) .bind(pubkey) .fetch_one(pool) .await?; @@ -527,7 +516,6 @@ pub async fn is_member(pool: &PgPool, channel_id: Uuid, pubkey: &[u8]) -> Result /// /// Returns an empty list if the channel has been soft-deleted. pub async fn get_members(pool: &PgPool, channel_id: Uuid) -> Result> { - let channel_id_bytes = channel_id.as_bytes().as_slice().to_vec(); let rows = sqlx::query( r#" SELECT cm.channel_id, cm.pubkey, cm.role, cm.joined_at, cm.invited_by, cm.removed_at @@ -538,7 +526,7 @@ pub async fn get_members(pool: &PgPool, channel_id: Uuid) -> Result Result< .await?; rows.into_iter() - .map(|r| { - let bytes: Vec = r.try_get("channel_id")?; - uuid_from_bytes(&bytes) - }) + .map(|r| Ok(r.try_get("channel_id")?)) .collect() } @@ -620,12 +605,11 @@ async fn get_active_role_tx( channel_id: Uuid, pubkey: &[u8], ) -> Result> { - let channel_id_bytes = channel_id.as_bytes().as_slice().to_vec(); let row = sqlx::query( "SELECT role FROM channel_members \ WHERE channel_id = $1 AND pubkey = $2 AND removed_at IS NULL", ) - .bind(&channel_id_bytes) + .bind(channel_id) .bind(pubkey) .fetch_optional(&mut **tx) .await?; @@ -637,7 +621,6 @@ async fn get_channel_tx( tx: &mut Transaction<'_, Postgres>, channel_id: Uuid, ) -> Result { - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); let row = sqlx::query( r#" SELECT id, name, channel_type, visibility, description, canvas, @@ -648,7 +631,7 @@ async fn get_channel_tx( FROM channels WHERE id = $1 AND deleted_at IS NULL "#, ) - .bind(&id_bytes) + .bind(channel_id) .fetch_optional(&mut **tx) .await? .ok_or(DbError::ChannelNotFound(channel_id))?; @@ -829,8 +812,7 @@ pub async fn get_users_bulk(pool: &PgPool, pubkeys: &[Vec]) -> Result Result { - let id_bytes: Vec = row.try_get("id")?; - let id = uuid_from_bytes(&id_bytes)?; + let id: Uuid = row.try_get("id")?; let topic_required: bool = row.try_get("topic_required")?; // topic/purpose fields are new — use try_get and fall back to None if the @@ -867,8 +849,7 @@ fn row_to_channel_record(row: sqlx::postgres::PgRow) -> Result { } fn row_to_member_record(row: sqlx::postgres::PgRow) -> Result { - let channel_id_bytes: Vec = row.try_get("channel_id")?; - let channel_id = uuid_from_bytes(&channel_id_bytes)?; + let channel_id: Uuid = row.try_get("channel_id")?; Ok(MemberRecord { channel_id, @@ -905,8 +886,6 @@ pub async fn update_channel( )); } - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); - // Build SET clause dynamically — only include fields that are Some. // Track parameter index for positional placeholders. let mut set_parts: Vec = Vec::new(); @@ -931,7 +910,7 @@ pub async fn update_channel( if let Some(ref desc) = updates.description { q = q.bind(desc); } - q = q.bind(&id_bytes); + q = q.bind(channel_id); let result = q.execute(pool).await?; if result.rows_affected() == 0 { @@ -943,14 +922,13 @@ pub async fn update_channel( /// Sets the topic for a channel, recording who set it and when. pub async fn set_topic(pool: &PgPool, channel_id: Uuid, topic: &str, set_by: &[u8]) -> Result<()> { - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); let result = sqlx::query( "UPDATE channels SET topic = $1, topic_set_by = $2, topic_set_at = NOW() \ WHERE id = $3 AND deleted_at IS NULL", ) .bind(topic) .bind(set_by) - .bind(&id_bytes) + .bind(channel_id) .execute(pool) .await?; if result.rows_affected() == 0 { @@ -966,14 +944,13 @@ pub async fn set_purpose( purpose: &str, set_by: &[u8], ) -> Result<()> { - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); let result = sqlx::query( "UPDATE channels SET purpose = $1, purpose_set_by = $2, purpose_set_at = NOW() \ WHERE id = $3 AND deleted_at IS NULL", ) .bind(purpose) .bind(set_by) - .bind(&id_bytes) + .bind(channel_id) .execute(pool) .await?; if result.rows_affected() == 0 { @@ -987,11 +964,9 @@ pub async fn set_purpose( /// Returns `AccessDenied` if the channel is already archived. /// Returns `ChannelNotFound` if the channel does not exist or is deleted. pub async fn archive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> { - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); - // First check: does the channel exist and what is its state? let row = sqlx::query("SELECT archived_at FROM channels WHERE id = $1 AND deleted_at IS NULL") - .bind(&id_bytes) + .bind(channel_id) .fetch_optional(pool) .await?; @@ -1011,7 +986,7 @@ pub async fn archive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> { "UPDATE channels SET archived_at = NOW() \ WHERE id = $1 AND deleted_at IS NULL AND archived_at IS NULL", ) - .bind(&id_bytes) + .bind(channel_id) .execute(pool) .await?; @@ -1023,11 +998,9 @@ pub async fn archive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> { /// Returns `AccessDenied` if the channel is not currently archived. /// Returns `ChannelNotFound` if the channel does not exist or is deleted. pub async fn unarchive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> { - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); - // First check: does the channel exist and what is its state? let row = sqlx::query("SELECT archived_at FROM channels WHERE id = $1 AND deleted_at IS NULL") - .bind(&id_bytes) + .bind(channel_id) .fetch_optional(pool) .await?; @@ -1045,7 +1018,7 @@ pub async fn unarchive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> { "UPDATE channels SET archived_at = NULL \ WHERE id = $1 AND deleted_at IS NULL AND archived_at IS NOT NULL", ) - .bind(&id_bytes) + .bind(channel_id) .execute(pool) .await?; @@ -1057,10 +1030,9 @@ pub async fn unarchive_channel(pool: &PgPool, channel_id: Uuid) -> Result<()> { /// Returns `Ok(true)` if the channel was deleted, `Ok(false)` if already /// deleted or not found. pub async fn soft_delete_channel(pool: &PgPool, channel_id: Uuid) -> Result { - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); let result = sqlx::query("UPDATE channels SET deleted_at = NOW() WHERE id = $1 AND deleted_at IS NULL") - .bind(&id_bytes) + .bind(channel_id) .execute(pool) .await?; @@ -1069,11 +1041,10 @@ pub async fn soft_delete_channel(pool: &PgPool, channel_id: Uuid) -> Result Result { - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); let row = sqlx::query( "SELECT COUNT(*) as cnt FROM channel_members WHERE channel_id = $1 AND removed_at IS NULL", ) - .bind(&id_bytes) + .bind(channel_id) .fetch_one(pool) .await?; Ok(row.try_get("cnt")?) @@ -1087,8 +1058,6 @@ pub async fn get_member_counts_bulk( pool: &PgPool, channel_ids: &[Uuid], ) -> Result> { - use crate::event::uuid_from_bytes; - if channel_ids.is_empty() { return Ok(std::collections::HashMap::new()); } @@ -1099,7 +1068,7 @@ pub async fn get_member_counts_bulk( ); let mut sep = qb.separated(", "); for id in channel_ids { - sep.push_bind(id.as_bytes().to_vec()); + sep.push_bind(*id); } qb.push(") GROUP BY channel_id"); @@ -1107,8 +1076,7 @@ pub async fn get_member_counts_bulk( let mut map = std::collections::HashMap::with_capacity(rows.len()); for row in rows { - let id_bytes: Vec = row.try_get("channel_id")?; - let id = uuid_from_bytes(&id_bytes)?; + let id: Uuid = row.try_get("channel_id")?; let cnt: i64 = row.try_get("cnt")?; map.insert(id, cnt); } @@ -1123,13 +1091,12 @@ pub async fn get_member_role( channel_id: Uuid, pubkey: &[u8], ) -> Result> { - let channel_id_bytes = channel_id.as_bytes().as_slice().to_vec(); let row = sqlx::query( "SELECT cm.role FROM channel_members cm \ JOIN channels c ON cm.channel_id = c.id AND c.deleted_at IS NULL \ WHERE cm.channel_id = $1 AND cm.pubkey = $2 AND cm.removed_at IS NULL", ) - .bind(&channel_id_bytes) + .bind(channel_id) .bind(pubkey) .fetch_optional(pool) .await?; diff --git a/crates/sprout-db/src/dm.rs b/crates/sprout-db/src/dm.rs index 79c72e7a6..1ecbc807e 100644 --- a/crates/sprout-db/src/dm.rs +++ b/crates/sprout-db/src/dm.rs @@ -10,7 +10,6 @@ use uuid::Uuid; use crate::channel::ChannelRecord; use crate::error::{DbError, Result}; -use crate::event::uuid_from_bytes; // -- Public structs ----------------------------------------------------------- @@ -154,7 +153,6 @@ pub async fn create_dm( }; let id = Uuid::new_v4(); - let id_bytes = id.as_bytes().as_slice().to_vec(); sqlx::query( r#" @@ -163,7 +161,7 @@ pub async fn create_dm( VALUES ($1, $2, 'dm', 'private', $3, $4) "#, ) - .bind(&id_bytes) + .bind(id) .bind(&name) .bind(created_by) .bind(hash.as_slice()) @@ -182,7 +180,7 @@ pub async fn create_dm( role = EXCLUDED.role "#, ) - .bind(&id_bytes) + .bind(id) .bind(*pk) .bind(created_by) .execute(&mut *tx) @@ -199,7 +197,7 @@ pub async fn create_dm( FROM channels WHERE id = $1 "#, ) - .bind(&id_bytes) + .bind(id) .fetch_one(&mut *tx) .await?; @@ -222,9 +220,8 @@ pub async fn list_dms_for_user( // Resolve cursor to a timestamp for keyset pagination. let cursor_ts: Option> = if let Some(cid) = cursor { - let cid_bytes = cid.as_bytes().as_slice().to_vec(); let row = sqlx::query("SELECT updated_at FROM channels WHERE id = $1") - .bind(&cid_bytes) + .bind(cid) .fetch_optional(pool) .await?; row.map(|r| r.try_get::, _>("updated_at")) @@ -279,8 +276,7 @@ pub async fn list_dms_for_user( let mut results = Vec::with_capacity(channel_rows.len()); for row in channel_rows { - let id_bytes: Vec = row.try_get("id")?; - let channel_id = uuid_from_bytes(&id_bytes)?; + let channel_id: Uuid = row.try_get("id")?; let created_at: DateTime = row.try_get("created_at")?; let updated_at: DateTime = row.try_get("updated_at")?; @@ -295,7 +291,7 @@ pub async fn list_dms_for_user( ORDER BY cm.joined_at ASC "#, ) - .bind(&id_bytes) + .bind(channel_id) .fetch_all(pool) .await?; @@ -363,8 +359,7 @@ pub async fn open_dm( // -- Row mapping -------------------------------------------------------------- fn row_to_channel_record(row: sqlx::postgres::PgRow) -> Result { - let id_bytes: Vec = row.try_get("id")?; - let id = uuid_from_bytes(&id_bytes)?; + let id: Uuid = row.try_get("id")?; let topic_required: bool = row.try_get("topic_required")?; Ok(ChannelRecord { diff --git a/crates/sprout-db/src/event.rs b/crates/sprout-db/src/event.rs index c1bb52393..cc4591f62 100644 --- a/crates/sprout-db/src/event.rs +++ b/crates/sprout-db/src/event.rs @@ -64,8 +64,6 @@ pub async fn insert_event( let created_at = DateTime::from_timestamp(created_at_secs, 0) .ok_or(DbError::InvalidTimestamp(created_at_secs))?; let received_at = Utc::now(); - let channel_id_bytes: Option<[u8; 16]> = channel_id.map(|u| *u.as_bytes()); - let result = sqlx::query( r#" INSERT INTO events (id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id) @@ -81,7 +79,7 @@ pub async fn insert_event( .bind(&event.content) .bind(sig_bytes.as_slice()) .bind(received_at) - .bind(channel_id_bytes.as_ref().map(|b| b.as_slice())) + .bind(channel_id) .execute(pool) .await?; @@ -129,7 +127,7 @@ pub async fn query_events(pool: &PgPool, q: &EventQuery) -> Result = row.try_get("received_at")?; - let channel_id_bytes: Option> = row.try_get("channel_id")?; - let channel_id: Option = channel_id_bytes.map(|b| uuid_from_bytes(&b)).transpose()?; + let channel_id: Option = row.try_get("channel_id")?; // kind is stored as i32 (Postgres INT) but Nostr uses u16. Values > 65535 are corrupt. let kind_u16 = u16::try_from(kind_i32) @@ -282,13 +279,12 @@ pub async fn get_last_message_at( pool: &PgPool, channel_id: uuid::Uuid, ) -> Result>> { - let id_bytes = channel_id.as_bytes().as_slice().to_vec(); let row = sqlx::query( "SELECT created_at FROM events \ WHERE channel_id = $1 AND deleted_at IS NULL \ ORDER BY created_at DESC LIMIT 1", ) - .bind(&id_bytes) + .bind(channel_id) .fetch_optional(pool) .await?; @@ -316,7 +312,7 @@ pub async fn get_last_message_at_bulk( ); let mut sep = qb.separated(", "); for id in channel_ids { - sep.push_bind(id.as_bytes().to_vec()); + sep.push_bind(*id); } qb.push(") GROUP BY channel_id"); @@ -324,8 +320,7 @@ pub async fn get_last_message_at_bulk( let mut map = std::collections::HashMap::with_capacity(rows.len()); for row in rows { - let id_bytes: Vec = row.try_get("channel_id")?; - let id = uuid_from_bytes(&id_bytes)?; + let id: Uuid = row.try_get("channel_id")?; let last_at: DateTime = row.try_get("last_at")?; map.insert(id, last_at); } @@ -462,8 +457,6 @@ pub async fn insert_event_with_thread_metadata( let created_at = DateTime::from_timestamp(created_at_secs, 0) .ok_or(DbError::InvalidTimestamp(created_at_secs))?; let received_at = Utc::now(); - let channel_id_bytes: Option<[u8; 16]> = channel_id.map(|u| *u.as_bytes()); - let mut tx = pool.begin().await?; // ── Insert event ────────────────────────────────────────────────────────── @@ -482,7 +475,7 @@ pub async fn insert_event_with_thread_metadata( .bind(&event.content) .bind(sig_bytes.as_slice()) .bind(received_at) - .bind(channel_id_bytes.as_ref().map(|b| b.as_slice())) + .bind(channel_id) .execute(&mut *tx) .await?; @@ -491,7 +484,6 @@ pub async fn insert_event_with_thread_metadata( // ── Insert thread metadata (if provided and event was actually inserted) ── if was_inserted { if let Some(ref meta) = thread_meta { - let ch_bytes = meta.channel_id.as_bytes().as_slice().to_vec(); let broadcast_val: bool = meta.broadcast; let tm_result = sqlx::query( @@ -507,7 +499,7 @@ pub async fn insert_event_with_thread_metadata( ) .bind(meta.event_created_at) .bind(meta.event_id) - .bind(ch_bytes.as_slice()) + .bind(meta.channel_id) .bind(meta.parent_event_id) .bind(meta.parent_event_created_at) .bind(meta.root_event_id) @@ -539,7 +531,7 @@ pub async fn insert_event_with_thread_metadata( ) .bind(parent_ts) .bind(pid) - .bind(ch_bytes.as_slice()) + .bind(meta.channel_id) .execute(&mut *tx) .await?; @@ -561,7 +553,7 @@ pub async fn insert_event_with_thread_metadata( ) .bind(root_ts) .bind(root_id) - .bind(ch_bytes.as_slice()) + .bind(meta.channel_id) .execute(&mut *tx) .await?; } diff --git a/crates/sprout-db/src/feed.rs b/crates/sprout-db/src/feed.rs index d2853465f..1ed6a702b 100644 --- a/crates/sprout-db/src/feed.rs +++ b/crates/sprout-db/src/feed.rs @@ -76,7 +76,7 @@ pub async fn query_mentions( qb.push(" AND e.channel_id IN ("); let mut sep = qb.separated(", "); for id in accessible_channel_ids { - sep.push_bind(id.as_bytes().to_vec()); + sep.push_bind(*id); } qb.push(")"); } @@ -135,7 +135,7 @@ pub async fn query_needs_action( qb.push(" AND e.channel_id IN ("); let mut sep = qb.separated(", "); for id in accessible_channel_ids { - sep.push_bind(id.as_bytes().to_vec()); + sep.push_bind(*id); } qb.push(")"); } @@ -185,7 +185,7 @@ pub async fn query_activity( qb.push(" AND channel_id IN ("); let mut sep = qb.separated(", "); for id in accessible_channel_ids { - sep.push_bind(id.as_bytes().to_vec()); + sep.push_bind(*id); } qb.push(")"); } diff --git a/crates/sprout-db/src/lib.rs b/crates/sprout-db/src/lib.rs index 33732154b..bd917dadb 100644 --- a/crates/sprout-db/src/lib.rs +++ b/crates/sprout-db/src/lib.rs @@ -75,7 +75,6 @@ pub async fn insert_mentions( let created_at_secs = event.created_at.as_u64() as i64; let created_at = DateTime::from_timestamp(created_at_secs, 0) .ok_or(crate::error::DbError::InvalidTimestamp(created_at_secs))?; - let channel_id_bytes = channel_id.map(|id| id.as_bytes().to_vec()); let kind = event.kind.as_u16() as u32; // Validate and normalize pubkeys, logging any malformed ones. @@ -110,7 +109,7 @@ pub async fn insert_mentions( b.push_bind(pubkey.as_str()) .push_bind(event_id_bytes.as_slice()) .push_bind(created_at) - .push_bind(channel_id_bytes.as_deref()) + .push_bind(channel_id) .push_bind(kind as i32); }); @@ -1185,12 +1184,11 @@ impl Db { channel_id: Uuid, relay_pubkey: &[u8], ) -> Result { - let channel_id_bytes = channel_id.as_bytes().to_vec(); let result = sqlx::query( "UPDATE events SET deleted_at = NOW() \ WHERE channel_id = $1 AND pubkey = $2 AND deleted_at IS NULL AND kind IN (39000, 39001, 39002)", ) - .bind(channel_id_bytes.as_slice()) + .bind(channel_id) .bind(relay_pubkey) .execute(&self.pool) .await?; @@ -1208,8 +1206,6 @@ impl Db { ) -> Result<(StoredEvent, bool)> { let kind_i32 = sprout_core::kind::event_kind_i32(event); let pubkey_bytes = event.pubkey.to_bytes(); - let channel_id_bytes: Option> = channel_id.map(|u| u.as_bytes().to_vec()); - let mut tx = self.pool.begin().await?; // Soft-delete existing events with the same (kind, pubkey, channel_id). @@ -1220,7 +1216,7 @@ impl Db { ) .bind(kind_i32) .bind(pubkey_bytes.as_slice()) - .bind(channel_id_bytes.as_deref()) + .bind(channel_id) .execute(&mut *tx) .await?; diff --git a/crates/sprout-db/src/thread.rs b/crates/sprout-db/src/thread.rs index 6da39ea93..4809be706 100644 --- a/crates/sprout-db/src/thread.rs +++ b/crates/sprout-db/src/thread.rs @@ -9,7 +9,6 @@ use sqlx::{PgPool, Row}; use uuid::Uuid; use crate::error::Result; -use crate::event::uuid_from_bytes; // -- Structs ------------------------------------------------------------------ @@ -120,8 +119,6 @@ pub async fn insert_thread_metadata( depth: i32, broadcast: bool, ) -> Result<()> { - let channel_id_bytes = channel_id.as_bytes().as_slice().to_vec(); - let mut tx = pool.begin().await?; let result = sqlx::query( @@ -137,7 +134,7 @@ pub async fn insert_thread_metadata( ) .bind(event_created_at) .bind(event_id) - .bind(channel_id_bytes.as_slice()) + .bind(channel_id) .bind(parent_event_id) .bind(parent_event_created_at) .bind(root_event_id) @@ -168,7 +165,7 @@ pub async fn insert_thread_metadata( ) .bind(parent_ts) .bind(pid) - .bind(channel_id_bytes.as_slice()) + .bind(channel_id) .execute(&mut *tx) .await?; @@ -189,7 +186,7 @@ pub async fn insert_thread_metadata( ) .bind(root_ts) .bind(root_id) - .bind(channel_id_bytes.as_slice()) + .bind(channel_id) .execute(&mut *tx) .await?; } @@ -392,7 +389,7 @@ pub async fn get_thread_replies( let event_id: Vec = row.try_get("event_id")?; let parent_event_id: Option> = row.try_get("parent_event_id")?; let root_event_id_col: Option> = row.try_get("root_event_id")?; - let channel_id_bytes: Vec = row.try_get("channel_id")?; + let channel_id: Uuid = row.try_get("channel_id")?; let pubkey: Vec = row.try_get("pubkey")?; let tags: serde_json::Value = row.try_get("tags")?; let content: String = row.try_get("content")?; @@ -401,8 +398,6 @@ pub async fn get_thread_replies( let created_at: DateTime = row.try_get("event_created_at")?; let broadcast_val: bool = row.try_get("broadcast")?; - let channel_id = uuid_from_bytes(&channel_id_bytes)?; - replies.push(ThreadReply { event_id, parent_event_id, @@ -495,8 +490,6 @@ pub async fn get_channel_messages_top_level( before_cursor: Option>, kind_filter: Option<&[u32]>, ) -> Result> { - let channel_id_bytes = channel_id.as_bytes().as_slice().to_vec(); - let mut param_idx = 2u32; // $1 is channel_id let mut sql = String::from( r#" @@ -540,7 +533,7 @@ pub async fn get_channel_messages_top_level( sql.push_str(&format!(" ORDER BY e.created_at DESC LIMIT ${param_idx}")); - let mut q = sqlx::query(&sql).bind(channel_id_bytes.as_slice()); + let mut q = sqlx::query(&sql).bind(channel_id); if let Some(cursor) = before_cursor { q = q.bind(cursor); @@ -557,8 +550,7 @@ pub async fn get_channel_messages_top_level( let content: String = row.try_get("content")?; let kind: i32 = row.try_get("kind")?; let created_at: DateTime = row.try_get("created_at")?; - let channel_id_col: Vec = row.try_get("channel_id_bytes")?; - let ch_id = uuid_from_bytes(&channel_id_col)?; + let ch_id: Uuid = row.try_get("channel_id_bytes")?; messages.push(TopLevelMessage { event_id, @@ -611,7 +603,7 @@ pub async fn get_thread_metadata_by_event( let event_id_col: Vec = row.try_get("event_id")?; let event_created_at: DateTime = row.try_get("event_created_at")?; - let channel_id_bytes: Vec = row.try_get("channel_id")?; + let channel_id: Uuid = row.try_get("channel_id")?; let parent_event_id: Option> = row.try_get("parent_event_id")?; let root_event_id: Option> = row.try_get("root_event_id")?; let depth: i32 = row.try_get("depth")?; @@ -619,8 +611,6 @@ pub async fn get_thread_metadata_by_event( let descendant_count: i32 = row.try_get("descendant_count")?; let broadcast_val: bool = row.try_get("broadcast")?; - let channel_id = uuid_from_bytes(&channel_id_bytes)?; - Ok(Some(ThreadMetadataRecord { event_id: event_id_col, event_created_at, diff --git a/crates/sprout-db/src/workflow.rs b/crates/sprout-db/src/workflow.rs index be304fc3c..3fda9694a 100644 --- a/crates/sprout-db/src/workflow.rs +++ b/crates/sprout-db/src/workflow.rs @@ -1,6 +1,6 @@ //! Workflow CRUD -- workflows, workflow_runs, and workflow_approvals tables. //! -//! All IDs are stored as BINARY(16) (UUID bytes). Never uses string interpolation +//! All IDs are stored as native UUID columns. Never uses string interpolation //! for query values -- all user data goes through bind parameters. //! //! Security notes: @@ -16,7 +16,6 @@ use sqlx::{PgPool, Row}; use uuid::Uuid; use crate::error::{DbError, Result}; -use crate::event::uuid_from_bytes; // -- Token hashing ------------------------------------------------------------ @@ -252,7 +251,6 @@ pub async fn create_workflow( definition_hash: &[u8], ) -> Result { let id = Uuid::new_v4(); - let channel_id_bytes: Option> = channel_id.map(|u| u.as_bytes().to_vec()); sqlx::query( r#" @@ -261,10 +259,10 @@ pub async fn create_workflow( VALUES ($1, $2, $3, $4, $5, $6, 'active', TRUE) "#, ) - .bind(id.as_bytes().to_vec()) + .bind(id) .bind(name) .bind(owner_pubkey) - .bind(channel_id_bytes) + .bind(channel_id) .bind(definition_json) .bind(definition_hash) .execute(pool) @@ -283,7 +281,7 @@ pub async fn get_workflow(pool: &PgPool, id: Uuid) -> Result { WHERE id = $1 "#, ) - .bind(id.as_bytes().to_vec()) + .bind(id) .fetch_optional(pool) .await? .ok_or_else(|| DbError::NotFound(format!("workflow {id}")))?; @@ -314,7 +312,7 @@ pub async fn list_channel_workflows( LIMIT $2 OFFSET $3 "#, ) - .bind(channel_id.as_bytes().to_vec()) + .bind(channel_id) .bind(limit) .bind(offset) .fetch_all(pool) @@ -345,7 +343,7 @@ pub async fn list_enabled_channel_workflows( LIMIT $2 "#, ) - .bind(channel_id.as_bytes().to_vec()) + .bind(channel_id) .bind(LIST_MAX_LIMIT) .fetch_all(pool) .await?; @@ -396,7 +394,7 @@ pub async fn update_workflow( .bind(name) .bind(definition_json) .bind(definition_hash) - .bind(id.as_bytes().to_vec()) + .bind(id) .execute(pool) .await? .rows_affected(); @@ -417,7 +415,7 @@ pub async fn update_workflow_status(pool: &PgPool, id: Uuid, status: WorkflowSta "#, ) .bind(status.to_string()) - .bind(id.as_bytes().to_vec()) + .bind(id) .execute(pool) .await? .rows_affected(); @@ -438,7 +436,7 @@ pub async fn set_workflow_enabled(pool: &PgPool, id: Uuid, enabled: bool) -> Res "#, ) .bind(enabled) - .bind(id.as_bytes().to_vec()) + .bind(id) .execute(pool) .await? .rows_affected(); @@ -452,7 +450,7 @@ pub async fn set_workflow_enabled(pool: &PgPool, id: Uuid, enabled: bool) -> Res /// Delete a workflow and all its runs/approvals (CASCADE). pub async fn delete_workflow(pool: &PgPool, id: Uuid) -> Result<()> { let affected = sqlx::query("DELETE FROM workflows WHERE id = $1") - .bind(id.as_bytes().to_vec()) + .bind(id) .execute(pool) .await? .rows_affected(); @@ -485,8 +483,8 @@ pub async fn create_workflow_run( VALUES ($1, $2, 'pending', $3, 0, '[]', $4) "#, ) - .bind(id.as_bytes().to_vec()) - .bind(workflow_id.as_bytes().to_vec()) + .bind(id) + .bind(workflow_id) .bind(trigger_event_id) .bind(trigger_context) .execute(pool) @@ -505,7 +503,7 @@ pub async fn get_workflow_run(pool: &PgPool, id: Uuid) -> Result) -> "#, ) .bind(token_hash) - .bind(workflow_id.as_bytes().to_vec()) - .bind(run_id.as_bytes().to_vec()) + .bind(workflow_id) + .bind(run_id) .bind(step_id) .bind(step_index) .bind(approver_spec) @@ -710,13 +708,8 @@ pub async fn update_approval( // -- Row mappers -------------------------------------------------------------- fn row_to_workflow_record(row: sqlx::postgres::PgRow) -> Result { - let id_bytes: Vec = row.try_get("id")?; - let id = uuid_from_bytes(&id_bytes)?; - - let channel_id: Option = { - let raw: Option> = row.try_get("channel_id")?; - raw.map(|b| uuid_from_bytes(&b)).transpose()? - }; + let id: Uuid = row.try_get("id")?; + let channel_id: Option = row.try_get("channel_id")?; let status_str: String = row.try_get("status")?; let status = status_str.parse::()?; @@ -738,11 +731,8 @@ fn row_to_workflow_record(row: sqlx::postgres::PgRow) -> Result } fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result { - let id_bytes: Vec = row.try_get("id")?; - let id = uuid_from_bytes(&id_bytes)?; - - let wf_bytes: Vec = row.try_get("workflow_id")?; - let workflow_id = uuid_from_bytes(&wf_bytes)?; + let id: Uuid = row.try_get("id")?; + let workflow_id: Uuid = row.try_get("workflow_id")?; let status_str: String = row.try_get("status")?; let status = status_str.parse::()?; @@ -763,11 +753,8 @@ fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result { } fn row_to_approval_record(row: sqlx::postgres::PgRow) -> Result { - let wf_bytes: Vec = row.try_get("workflow_id")?; - let workflow_id = uuid_from_bytes(&wf_bytes)?; - - let run_bytes: Vec = row.try_get("run_id")?; - let run_id = uuid_from_bytes(&run_bytes)?; + let workflow_id: Uuid = row.try_get("workflow_id")?; + let run_id: Uuid = row.try_get("run_id")?; let status_str: String = row.try_get("status")?; let status = status_str.parse::()?; @@ -1240,37 +1227,4 @@ mod tests { assert_eq!(record.status, ApprovalStatus::Pending); assert_eq!(cloned.status, ApprovalStatus::Granted); } - - // -- uuid_from_bytes (helper) --------------------------------------------- - - #[test] - fn uuid_from_bytes_round_trips() { - let original = Uuid::new_v4(); - let bytes = original.as_bytes().to_vec(); - let recovered = uuid_from_bytes(&bytes).expect("uuid_from_bytes failed"); - assert_eq!(original, recovered); - } - - #[test] - fn uuid_from_bytes_rejects_wrong_length() { - let bad_bytes = vec![0u8; 10]; // UUID requires exactly 16 bytes - let err = uuid_from_bytes(&bad_bytes).unwrap_err(); - assert!( - matches!(err, DbError::InvalidData(_)), - "expected InvalidData, got: {err}" - ); - } - - #[test] - fn uuid_from_bytes_rejects_empty() { - let err = uuid_from_bytes(&[]).unwrap_err(); - assert!(matches!(err, DbError::InvalidData(_))); - } - - #[test] - fn uuid_from_bytes_accepts_nil_uuid() { - let nil_bytes = [0u8; 16]; - let result = uuid_from_bytes(&nil_bytes).expect("nil UUID should parse"); - assert_eq!(result, Uuid::nil()); - } } diff --git a/schema/schema.sql b/schema/schema.sql index c41afee2a..9fb0e5fa6 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -2,27 +2,17 @@ -- -- This file represents the desired state of the database schema. -- Use `pgschema apply --file schema/schema.sql` to bring the database up to date. - --- ── Custom types ────────────────────────────────────────────────────────────── - -CREATE TYPE channel_type AS ENUM ('stream', 'forum', 'dm', 'workflow'); -CREATE TYPE channel_visibility AS ENUM ('open', 'private'); -CREATE TYPE member_role AS ENUM ('owner', 'admin', 'member', 'guest', 'bot'); -CREATE TYPE workflow_status AS ENUM ('active', 'disabled', 'archived'); -CREATE TYPE run_status AS ENUM ('pending', 'running', 'waiting_approval', 'completed', 'failed', 'cancelled'); -CREATE TYPE approval_status AS ENUM ('pending', 'granted', 'denied', 'expired'); -CREATE TYPE delivery_method AS ENUM ('webhook', 'websocket'); -CREATE TYPE subscription_status AS ENUM ('active', 'paused', 'deleted'); -CREATE TYPE pause_reason AS ENUM ('user', 'system', 'rate_limit'); -CREATE TYPE channel_add_policy AS ENUM ('anyone', 'owner_only', 'nobody'); +-- +-- All "enum" columns use TEXT + CHECK constraints instead of CREATE TYPE so that +-- sqlx runtime queries can decode them as plain String without custom type registration. -- ── Channels ────────────────────────────────────────────────────────────────── CREATE TABLE channels ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name VARCHAR(255) NOT NULL, - channel_type channel_type NOT NULL DEFAULT 'stream', - visibility channel_visibility NOT NULL DEFAULT 'open', + channel_type TEXT NOT NULL DEFAULT 'stream', + visibility TEXT NOT NULL DEFAULT 'open', description TEXT, canvas TEXT, created_by BYTEA NOT NULL, @@ -39,7 +29,9 @@ CREATE TABLE channels ( purpose TEXT, purpose_set_by BYTEA, purpose_set_at TIMESTAMPTZ, - participant_hash BYTEA + participant_hash BYTEA, + CONSTRAINT chk_channel_type CHECK (channel_type IN ('stream', 'forum', 'dm', 'workflow')), + CONSTRAINT chk_channel_visibility CHECK (visibility IN ('open', 'private')) ); CREATE INDEX idx_channels_type ON channels (channel_type); @@ -52,12 +44,13 @@ CREATE UNIQUE INDEX idx_channels_dm_hash ON channels (participant_hash); CREATE TABLE channel_members ( channel_id UUID NOT NULL REFERENCES channels(id) ON DELETE CASCADE, pubkey BYTEA NOT NULL, - role member_role NOT NULL DEFAULT 'member', + role TEXT NOT NULL DEFAULT 'member', joined_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), invited_by BYTEA, removed_at TIMESTAMPTZ, removed_by BYTEA, - PRIMARY KEY (channel_id, pubkey) + PRIMARY KEY (channel_id, pubkey), + CONSTRAINT chk_member_role CHECK (role IN ('owner', 'admin', 'member', 'guest', 'bot')) ); -- ── Users ───────────────────────────────────────────────────────────────────── @@ -76,8 +69,9 @@ CREATE TABLE users ( deactivated_at TIMESTAMPTZ, metadata_event_id BYTEA, agent_owner_pubkey BYTEA REFERENCES users(pubkey) ON DELETE SET NULL, - channel_add_policy channel_add_policy NOT NULL DEFAULT 'anyone', - CONSTRAINT chk_users_pubkey_len CHECK (LENGTH(pubkey) = 32) + channel_add_policy TEXT NOT NULL DEFAULT 'anyone', + CONSTRAINT chk_users_pubkey_len CHECK (LENGTH(pubkey) = 32), + CONSTRAINT chk_channel_add_policy CHECK (channel_add_policy IN ('anyone', 'owner_only', 'nobody')) ); -- ── Events (partitioned by month on created_at) ────────────────────────────── @@ -142,14 +136,17 @@ CREATE TABLE subscriptions ( filter_channel_ids JSONB, filter_since TIMESTAMPTZ, filter_until TIMESTAMPTZ, - delivery_method delivery_method NOT NULL DEFAULT 'webhook', + delivery_method TEXT NOT NULL DEFAULT 'webhook', delivery_url TEXT, - status subscription_status NOT NULL DEFAULT 'active', - pause_reason pause_reason, + status TEXT NOT NULL DEFAULT 'active', + pause_reason TEXT, delivered_count BIGINT NOT NULL DEFAULT 0, error_count BIGINT NOT NULL DEFAULT 0, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT chk_delivery_method CHECK (delivery_method IN ('webhook', 'websocket')), + CONSTRAINT chk_subscription_status CHECK (status IN ('active', 'paused', 'deleted')), + CONSTRAINT chk_pause_reason CHECK (pause_reason IS NULL OR pause_reason IN ('user', 'system', 'rate_limit')) ); -- ── Delivery log (partitioned by month on delivered_at) ────────────────────── @@ -158,7 +155,7 @@ CREATE TABLE delivery_log ( id BIGINT GENERATED ALWAYS AS IDENTITY, subscription_id VARCHAR(255), event_id BYTEA, - method delivery_method, + method TEXT, delivered_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), success BOOLEAN, http_status INT, @@ -187,10 +184,11 @@ CREATE TABLE workflows ( channel_id UUID REFERENCES channels(id), definition JSONB NOT NULL, definition_hash BYTEA NOT NULL, - status workflow_status NOT NULL DEFAULT 'active', + status TEXT NOT NULL DEFAULT 'active', enabled BOOLEAN NOT NULL DEFAULT TRUE, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT chk_workflow_status CHECK (status IN ('active', 'disabled', 'archived')) ); CREATE INDEX idx_workflows_channel_active ON workflows (channel_id, status, enabled); @@ -200,7 +198,7 @@ CREATE INDEX idx_workflows_channel_active ON workflows (channel_id, status, enab CREATE TABLE workflow_runs ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), workflow_id UUID NOT NULL REFERENCES workflows(id) ON DELETE CASCADE, - status run_status NOT NULL DEFAULT 'pending', + status TEXT NOT NULL DEFAULT 'pending', trigger_event_id BYTEA, current_step INT NOT NULL DEFAULT 0, execution_trace JSONB NOT NULL DEFAULT '[]', @@ -208,7 +206,8 @@ CREATE TABLE workflow_runs ( started_at TIMESTAMPTZ, completed_at TIMESTAMPTZ, error_message TEXT, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT chk_run_status CHECK (status IN ('pending', 'running', 'waiting_approval', 'completed', 'failed', 'cancelled')) ); CREATE INDEX idx_workflow_runs_workflow ON workflow_runs (workflow_id); @@ -223,13 +222,14 @@ CREATE TABLE workflow_approvals ( step_id VARCHAR(64) NOT NULL, step_index INT NOT NULL, approver_spec TEXT NOT NULL, - status approval_status NOT NULL DEFAULT 'pending', + status TEXT NOT NULL DEFAULT 'pending', approver_pubkey BYTEA, note TEXT, granted_at TIMESTAMPTZ, denied_at TIMESTAMPTZ, expires_at TIMESTAMPTZ NOT NULL, - created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT chk_approval_status CHECK (status IN ('pending', 'granted', 'denied', 'expired')) ); CREATE INDEX idx_workflow_approvals_workflow ON workflow_approvals (workflow_id); @@ -239,7 +239,7 @@ CREATE INDEX idx_workflow_approvals_status ON workflow_approvals (status); -- ── API tokens ──────────────────────────────────────────────────────────────── CREATE TABLE api_tokens ( - id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + id BYTEA PRIMARY KEY, token_hash BYTEA NOT NULL UNIQUE, owner_pubkey BYTEA NOT NULL REFERENCES users(pubkey), name VARCHAR(255) NOT NULL,