External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix: add pgschema apply step to CI before relay start

Browse Source 
The relay no longer runs migrations itself — pgschema must be
invoked explicitly after Postgres is healthy and before the relay
starts.

Co-authored-by: Claude Code <noreply@anthropic.com>
Ai-assisted: true
This commit is contained in:
Alec Thomas
2026-03-20 06:56:16 +11:00
co-authored by Claude Code
parent 688846e9bb
commit 2d24656a45
9 changed files with 122 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+8
View File
@@ -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
crates/sprout-db/src/channel.rs
+29 -62
View File
@@ -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<ChannelRecord> {
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<ChannelRecor
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))?;
@@ -292,9 +288,8 @@ pub async fn get_channel(pool: &PgPool, channel_id: Uuid) -> Result<ChannelRecor
/// Returns the canvas content for a channel, if any.
pub async fn get_canvas(pool: &PgPool, channel_id: Uuid) -> Result<Option<String>> {
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<Option<String
/// Sets or clears the canvas content for a channel.
pub async fn set_canvas(pool: &PgPool, channel_id: Uuid, canvas: Option<&str>) -> 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<bool> {
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<Vec<MemberRecord>> {
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<Vec<MemberRe
LIMIT 1000
"#,
)
.bind(&channel_id_bytes)
.bind(channel_id)
.fetch_all(pool)
.await?;
rows.into_iter().map(row_to_member_record).collect()
@@ -567,10 +555,7 @@ pub async fn get_accessible_channel_ids(pool: &PgPool, pubkey: &[u8]) -> Result<
.await?;
rows.into_iter()
.map(|r| {
let bytes: Vec<u8> = 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<Option<String>> {
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<ChannelRecord> {
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<u8>]) -> Result<Vec<Us
}
fn row_to_channel_record(row: sqlx::postgres::PgRow) -> Result<ChannelRecord> {
let id_bytes: Vec<u8> = 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<ChannelRecord> {
}
fn row_to_member_record(row: sqlx::postgres::PgRow) -> Result<MemberRecord> {
let channel_id_bytes: Vec<u8> = 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<String> = 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<bool> {
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<bool
/// Returns the count of active (non-removed) members in a channel.
pub async fn get_member_count(pool: &PgPool, channel_id: Uuid) -> Result<i64> {
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<std::collections::HashMap<Uuid, i64>> {
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<u8> = 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<Option<String>> {
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?;
crates/sprout-db/src/dm.rs
+7 -12
View File
@@ -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<DateTime<Utc>> = 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::<DateTime<Utc>, _>("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<u8> = row.try_get("id")?;
let channel_id = uuid_from_bytes(&id_bytes)?;
let channel_id: Uuid = row.try_get("id")?;
let created_at: DateTime<Utc> = row.try_get("created_at")?;
let updated_at: DateTime<Utc> = 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<ChannelRecord> {
let id_bytes: Vec<u8> = 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 {
crates/sprout-db/src/event.rs
+10 -18
View File
@@ -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<Vec<StoredEve
if let Some(ch) = q.channel_id {
qb.push(format!(" AND {col_prefix}channel_id = "))
.push_bind(ch.as_bytes().to_vec());
.push_bind(ch);
}
if let Some(ks) = q.kinds.as_deref().filter(|k| !k.is_empty()) {
@@ -179,8 +177,7 @@ pub(crate) fn row_to_stored_event(row: sqlx::postgres::PgRow) -> Result<Option<S
let sig_bytes: Vec<u8> = row.try_get("sig")?;
let received_at: DateTime<Utc> = row.try_get("received_at")?;
let channel_id_bytes: Option<Vec<u8>> = row.try_get("channel_id")?;
let channel_id: Option<Uuid> = channel_id_bytes.map(|b| uuid_from_bytes(&b)).transpose()?;
let channel_id: Option<Uuid> = 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<Option<DateTime<Utc>>> {
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<u8> = row.try_get("channel_id")?;
let id = uuid_from_bytes(&id_bytes)?;
let id: Uuid = row.try_get("channel_id")?;
let last_at: DateTime<Utc> = 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?;
}
crates/sprout-db/src/feed.rs
+3 -3
View File
@@ -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(")");
}
crates/sprout-db/src/lib.rs
+3 -7
View File
@@ -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<u64> {
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<Vec<u8>> = 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?;
crates/sprout-db/src/thread.rs
+7 -17
View File
@@ -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<u8> = row.try_get("event_id")?;
let parent_event_id: Option<Vec<u8>> = row.try_get("parent_event_id")?;
let root_event_id_col: Option<Vec<u8>> = row.try_get("root_event_id")?;
let channel_id_bytes: Vec<u8> = row.try_get("channel_id")?;
let channel_id: Uuid = row.try_get("channel_id")?;
let pubkey: Vec<u8> = 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<Utc> = 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<DateTime<Utc>>,
kind_filter: Option<&[u32]>,
) -> Result<Vec<TopLevelMessage>> {
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<Utc> = row.try_get("created_at")?;
let channel_id_col: Vec<u8> = 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<u8> = row.try_get("event_id")?;
let event_created_at: DateTime<Utc> = row.try_get("event_created_at")?;
let channel_id_bytes: Vec<u8> = row.try_get("channel_id")?;
let channel_id: Uuid = row.try_get("channel_id")?;
let parent_event_id: Option<Vec<u8>> = row.try_get("parent_event_id")?;
let root_event_id: Option<Vec<u8>> = 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,
crates/sprout-db/src/workflow.rs
+23 -69
View File
@@ -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<Uuid> {
let id = Uuid::new_v4();
let channel_id_bytes: Option<Vec<u8>> = 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<WorkflowRecord> {
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<WorkflowRunReco
WHERE id = $1
"#,
)
.bind(id.as_bytes().to_vec())
.bind(id)
.fetch_optional(pool)
.await?
.ok_or_else(|| DbError::NotFound(format!("workflow_run {id}")))?;
@@ -530,7 +528,7 @@ pub async fn list_workflow_runs(
LIMIT $2
"#,
)
.bind(workflow_id.as_bytes().to_vec())
.bind(workflow_id)
.bind(limit)
.fetch_all(pool)
.await?;
@@ -573,7 +571,7 @@ pub async fn update_workflow_run(
.bind(error)
.bind(&status_str) // for started_at CASE
.bind(&status_str) // for completed_at CASE
.bind(id.as_bytes().to_vec())
.bind(id)
.execute(pool)
.await?
.rows_affected();
@@ -628,8 +626,8 @@ pub async fn create_approval(pool: &PgPool, params: CreateApprovalParams<'_>) ->
"#,
)
.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<WorkflowRecord> {
let id_bytes: Vec<u8> = row.try_get("id")?;
let id = uuid_from_bytes(&id_bytes)?;
let channel_id: Option<Uuid> = {
let raw: Option<Vec<u8>> = row.try_get("channel_id")?;
raw.map(|b| uuid_from_bytes(&b)).transpose()?
};
let id: Uuid = row.try_get("id")?;
let channel_id: Option<Uuid> = row.try_get("channel_id")?;
let status_str: String = row.try_get("status")?;
let status = status_str.parse::<WorkflowStatus>()?;
@@ -738,11 +731,8 @@ fn row_to_workflow_record(row: sqlx::postgres::PgRow) -> Result<WorkflowRecord>
}
fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result<WorkflowRunRecord> {
let id_bytes: Vec<u8> = row.try_get("id")?;
let id = uuid_from_bytes(&id_bytes)?;
let wf_bytes: Vec<u8> = 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::<RunStatus>()?;
@@ -763,11 +753,8 @@ fn row_to_run_record(row: sqlx::postgres::PgRow) -> Result<WorkflowRunRecord> {
}
fn row_to_approval_record(row: sqlx::postgres::PgRow) -> Result<ApprovalRecord> {
let wf_bytes: Vec<u8> = row.try_get("workflow_id")?;
let workflow_id = uuid_from_bytes(&wf_bytes)?;
let run_bytes: Vec<u8> = 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::<ApprovalStatus>()?;
@@ -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());
}
}
schema/schema.sql
+32 -32
View File
@@ -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,