mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
feat: ephemeral channels with TTL-based auto-archiving (#232)
Co-authored-by: Wes <wesbillman@users.noreply.github.com>
This commit is contained in:
Bradley Axen
@@ -81,6 +81,17 @@ OKTA_AUDIENCE=sprout-desktop
|
||||
# OKTA_AUDIENCE=sprout-api
|
||||
# OKTA_PUBKEY_CLAIM=nostr_pubkey
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Ephemeral Channels (TTL testing)
|
||||
# -----------------------------------------------------------------------------
|
||||
# Override the TTL for all ephemeral channels (in seconds). When set, any
|
||||
# channel created with a TTL tag will use this value instead of the
|
||||
# client-provided one. Unset to use the client-provided TTL.
|
||||
# SPROUT_EPHEMERAL_TTL_OVERRIDE=60
|
||||
|
||||
# How often the reaper checks for expired ephemeral channels (default: 60s).
|
||||
# SPROUT_REAPER_INTERVAL_SECS=5
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Logging / Tracing
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
@@ -58,6 +58,10 @@ pub struct ChannelRecord {
|
||||
pub purpose_set_by: Option<Vec<u8>>,
|
||||
/// When the purpose was last set.
|
||||
pub purpose_set_at: Option<DateTime<Utc>>,
|
||||
/// TTL in seconds for ephemeral channels. `None` means permanent.
|
||||
pub ttl_seconds: Option<i32>,
|
||||
/// Deadline by which a new message must arrive or the channel is auto-archived.
|
||||
pub ttl_deadline: Option<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
/// A channel membership row as returned from the database.
|
||||
@@ -85,6 +89,7 @@ pub async fn create_channel(
|
||||
visibility: ChannelVisibility,
|
||||
description: Option<&str>,
|
||||
created_by: &[u8],
|
||||
ttl_seconds: Option<i32>,
|
||||
) -> Result<ChannelRecord> {
|
||||
if created_by.len() != 32 {
|
||||
return Err(DbError::InvalidData(format!(
|
||||
@@ -99,8 +104,9 @@ pub async fn create_channel(
|
||||
|
||||
sqlx::query(
|
||||
r#"
|
||||
INSERT INTO channels (id, name, channel_type, visibility, description, created_by)
|
||||
VALUES ($1, $2, $3::channel_type, $4::channel_visibility, $5, $6)
|
||||
INSERT INTO channels (id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline)
|
||||
VALUES ($1, $2, $3::channel_type, $4::channel_visibility, $5, $6, $7,
|
||||
CASE WHEN $7 IS NOT NULL THEN NOW() + ($7 || ' seconds')::interval ELSE NULL END)
|
||||
"#,
|
||||
)
|
||||
.bind(id)
|
||||
@@ -109,6 +115,7 @@ pub async fn create_channel(
|
||||
.bind(visibility.as_str())
|
||||
.bind(description)
|
||||
.bind(created_by)
|
||||
.bind(ttl_seconds)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
@@ -135,7 +142,8 @@ pub async fn create_channel(
|
||||
created_by, created_at, updated_at, archived_at, deleted_at,
|
||||
nip29_group_id, topic_required, max_members,
|
||||
topic, topic_set_by, topic_set_at,
|
||||
purpose, purpose_set_by, purpose_set_at
|
||||
purpose, purpose_set_by, purpose_set_at,
|
||||
ttl_seconds, ttl_deadline
|
||||
FROM channels WHERE id = $1
|
||||
"#,
|
||||
)
|
||||
@@ -152,6 +160,7 @@ pub async fn create_channel(
|
||||
///
|
||||
/// Returns `(record, true)` if the channel was newly created, or `(record, false)` if a
|
||||
/// channel with `channel_id` already exists (duplicate — caller should reject the event).
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn create_channel_with_id(
|
||||
pool: &PgPool,
|
||||
channel_id: Uuid,
|
||||
@@ -160,6 +169,7 @@ pub async fn create_channel_with_id(
|
||||
visibility: ChannelVisibility,
|
||||
description: Option<&str>,
|
||||
created_by: &[u8],
|
||||
ttl_seconds: Option<i32>,
|
||||
) -> Result<(ChannelRecord, bool)> {
|
||||
if created_by.len() != 32 {
|
||||
return Err(DbError::InvalidData(format!(
|
||||
@@ -172,8 +182,9 @@ pub async fn create_channel_with_id(
|
||||
|
||||
let rows_affected = sqlx::query(
|
||||
r#"
|
||||
INSERT INTO channels (id, name, channel_type, visibility, description, created_by)
|
||||
VALUES ($1, $2, $3::channel_type, $4::channel_visibility, $5, $6)
|
||||
INSERT INTO channels (id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline)
|
||||
VALUES ($1, $2, $3::channel_type, $4::channel_visibility, $5, $6, $7,
|
||||
CASE WHEN $7 IS NOT NULL THEN NOW() + ($7 || ' seconds')::interval ELSE NULL END)
|
||||
ON CONFLICT (id) DO NOTHING
|
||||
"#,
|
||||
)
|
||||
@@ -183,6 +194,7 @@ pub async fn create_channel_with_id(
|
||||
.bind(visibility.as_str())
|
||||
.bind(description)
|
||||
.bind(created_by)
|
||||
.bind(ttl_seconds)
|
||||
.execute(&mut *tx)
|
||||
.await?
|
||||
.rows_affected();
|
||||
@@ -215,7 +227,8 @@ pub async fn create_channel_with_id(
|
||||
created_by, created_at, updated_at, archived_at, deleted_at,
|
||||
nip29_group_id, topic_required, max_members,
|
||||
topic, topic_set_by, topic_set_at,
|
||||
purpose, purpose_set_by, purpose_set_at
|
||||
purpose, purpose_set_by, purpose_set_at,
|
||||
ttl_seconds, ttl_deadline
|
||||
FROM channels WHERE id = $1
|
||||
"#,
|
||||
)
|
||||
@@ -237,7 +250,8 @@ pub async fn get_channel(pool: &PgPool, channel_id: Uuid) -> Result<ChannelRecor
|
||||
created_by, created_at, updated_at, archived_at, deleted_at,
|
||||
nip29_group_id, topic_required, max_members,
|
||||
topic, topic_set_by, topic_set_at,
|
||||
purpose, purpose_set_by, purpose_set_at
|
||||
purpose, purpose_set_by, purpose_set_at,
|
||||
ttl_seconds, ttl_deadline
|
||||
FROM channels WHERE id = $1 AND deleted_at IS NULL
|
||||
"#,
|
||||
)
|
||||
@@ -535,7 +549,8 @@ pub async fn list_channels(pool: &PgPool, visibility: Option<&str>) -> Result<Ve
|
||||
created_by, created_at, updated_at, archived_at, deleted_at,
|
||||
nip29_group_id, topic_required, max_members,
|
||||
topic, topic_set_by, topic_set_at,
|
||||
purpose, purpose_set_by, purpose_set_at
|
||||
purpose, purpose_set_by, purpose_set_at,
|
||||
ttl_seconds, ttl_deadline
|
||||
FROM channels
|
||||
WHERE deleted_at IS NULL AND visibility::text = $1
|
||||
ORDER BY created_at DESC
|
||||
@@ -553,7 +568,8 @@ pub async fn list_channels(pool: &PgPool, visibility: Option<&str>) -> Result<Ve
|
||||
created_by, created_at, updated_at, archived_at, deleted_at,
|
||||
nip29_group_id, topic_required, max_members,
|
||||
topic, topic_set_by, topic_set_at,
|
||||
purpose, purpose_set_by, purpose_set_at
|
||||
purpose, purpose_set_by, purpose_set_at,
|
||||
ttl_seconds, ttl_deadline
|
||||
FROM channels
|
||||
WHERE deleted_at IS NULL
|
||||
ORDER BY created_at DESC
|
||||
@@ -596,7 +612,8 @@ async fn get_channel_tx(
|
||||
created_by, created_at, updated_at, archived_at, deleted_at,
|
||||
nip29_group_id, topic_required, max_members,
|
||||
topic, topic_set_by, topic_set_at,
|
||||
purpose, purpose_set_by, purpose_set_at
|
||||
purpose, purpose_set_by, purpose_set_at,
|
||||
ttl_seconds, ttl_deadline
|
||||
FROM channels WHERE id = $1 AND deleted_at IS NULL
|
||||
"#,
|
||||
)
|
||||
@@ -685,6 +702,7 @@ pub async fn get_accessible_channels(
|
||||
c.nip29_group_id, c.topic_required, c.max_members,
|
||||
c.topic, c.topic_set_by, c.topic_set_at,
|
||||
c.purpose, c.purpose_set_by, c.purpose_set_at,
|
||||
c.ttl_seconds, c.ttl_deadline,
|
||||
(cm.channel_id IS NOT NULL) AS is_member
|
||||
FROM channels c
|
||||
LEFT JOIN channel_members cm
|
||||
@@ -807,6 +825,8 @@ fn row_to_channel_record(row: sqlx::postgres::PgRow) -> Result<ChannelRecord> {
|
||||
let purpose: Option<String> = row.try_get("purpose").unwrap_or(None);
|
||||
let purpose_set_by: Option<Vec<u8>> = row.try_get("purpose_set_by").unwrap_or(None);
|
||||
let purpose_set_at: Option<DateTime<Utc>> = row.try_get("purpose_set_at").unwrap_or(None);
|
||||
let ttl_seconds: Option<i32> = row.try_get("ttl_seconds").unwrap_or(None);
|
||||
let ttl_deadline: Option<DateTime<Utc>> = row.try_get("ttl_deadline").unwrap_or(None);
|
||||
|
||||
Ok(ChannelRecord {
|
||||
id,
|
||||
@@ -829,6 +849,8 @@ fn row_to_channel_record(row: sqlx::postgres::PgRow) -> Result<ChannelRecord> {
|
||||
purpose,
|
||||
purpose_set_by,
|
||||
purpose_set_at,
|
||||
ttl_seconds,
|
||||
ttl_deadline,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1086,3 +1108,42 @@ pub async fn get_member_role(
|
||||
.await?;
|
||||
Ok(row.map(|r| r.try_get("role")).transpose()?)
|
||||
}
|
||||
|
||||
/// Bump the TTL deadline for an ephemeral channel after a new message.
|
||||
///
|
||||
/// No-op for permanent channels or channels that are already archived/deleted.
|
||||
pub async fn bump_ttl_deadline(pool: &PgPool, channel_id: Uuid) -> Result<()> {
|
||||
sqlx::query(
|
||||
"UPDATE channels SET ttl_deadline = NOW() + (ttl_seconds || ' seconds')::interval \
|
||||
WHERE id = $1 AND ttl_seconds IS NOT NULL AND archived_at IS NULL AND deleted_at IS NULL",
|
||||
)
|
||||
.bind(channel_id)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Archive ephemeral channels whose TTL deadline has passed.
|
||||
///
|
||||
/// Returns the list of channel IDs that were archived. Idempotent — the
|
||||
/// `archived_at IS NULL` guard prevents double-archiving even if called
|
||||
/// concurrently from multiple relay pods.
|
||||
pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result<Vec<Uuid>> {
|
||||
let rows = sqlx::query(
|
||||
"UPDATE channels SET archived_at = NOW() \
|
||||
WHERE ttl_seconds IS NOT NULL \
|
||||
AND ttl_deadline < NOW() \
|
||||
AND archived_at IS NULL \
|
||||
AND deleted_at IS NULL \
|
||||
RETURNING id",
|
||||
)
|
||||
.fetch_all(pool)
|
||||
.await?;
|
||||
|
||||
rows.into_iter()
|
||||
.map(|row| {
|
||||
let id: Uuid = row.try_get("id")?;
|
||||
Ok(id)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
@@ -439,6 +439,8 @@ fn row_to_channel_record(row: sqlx::postgres::PgRow) -> Result<ChannelRecord> {
|
||||
purpose: row.try_get("purpose").unwrap_or(None),
|
||||
purpose_set_by: row.try_get("purpose_set_by").unwrap_or(None),
|
||||
purpose_set_at: row.try_get("purpose_set_at").unwrap_or(None),
|
||||
ttl_seconds: row.try_get("ttl_seconds").unwrap_or(None),
|
||||
ttl_deadline: row.try_get("ttl_deadline").unwrap_or(None),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -298,6 +298,7 @@ impl Db {
|
||||
visibility: channel::ChannelVisibility,
|
||||
description: Option<&str>,
|
||||
created_by: &[u8],
|
||||
ttl_seconds: Option<i32>,
|
||||
) -> Result<channel::ChannelRecord> {
|
||||
channel::create_channel(
|
||||
&self.pool,
|
||||
@@ -306,6 +307,7 @@ impl Db {
|
||||
visibility,
|
||||
description,
|
||||
created_by,
|
||||
ttl_seconds,
|
||||
)
|
||||
.await
|
||||
}
|
||||
@@ -313,6 +315,7 @@ impl Db {
|
||||
/// Creates a channel with a client-supplied UUID.
|
||||
///
|
||||
/// Returns `(record, true)` if newly created, `(record, false)` if already exists.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn create_channel_with_id(
|
||||
&self,
|
||||
channel_id: Uuid,
|
||||
@@ -321,6 +324,7 @@ impl Db {
|
||||
visibility: channel::ChannelVisibility,
|
||||
description: Option<&str>,
|
||||
created_by: &[u8],
|
||||
ttl_seconds: Option<i32>,
|
||||
) -> Result<(channel::ChannelRecord, bool)> {
|
||||
channel::create_channel_with_id(
|
||||
&self.pool,
|
||||
@@ -330,6 +334,7 @@ impl Db {
|
||||
visibility,
|
||||
description,
|
||||
created_by,
|
||||
ttl_seconds,
|
||||
)
|
||||
.await
|
||||
}
|
||||
@@ -465,6 +470,16 @@ impl Db {
|
||||
channel::get_member_role(&self.pool, channel_id, pubkey).await
|
||||
}
|
||||
|
||||
/// Bump the TTL deadline for an ephemeral channel after a new message.
|
||||
pub async fn bump_ttl_deadline(&self, channel_id: Uuid) -> Result<()> {
|
||||
channel::bump_ttl_deadline(&self.pool, channel_id).await
|
||||
}
|
||||
|
||||
/// Archive ephemeral channels whose TTL deadline has passed.
|
||||
pub async fn reap_expired_ephemeral_channels(&self) -> Result<Vec<Uuid>> {
|
||||
channel::reap_expired_ephemeral_channels(&self.pool).await
|
||||
}
|
||||
|
||||
// ── Users ────────────────────────────────────────────────────────────────
|
||||
|
||||
/// Ensure a user record exists (upsert).
|
||||
|
||||
@@ -112,6 +112,8 @@ fn channel_record_to_json(
|
||||
"participants": participants,
|
||||
"participant_pubkeys": participant_pubkeys,
|
||||
"is_member": is_member,
|
||||
"ttl_seconds": channel.ttl_seconds,
|
||||
"ttl_deadline": channel.ttl_deadline.map(|t| t.to_rfc3339()),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -50,6 +50,8 @@ fn channel_detail_to_json(record: &ChannelRecord, member_count: i64) -> serde_js
|
||||
"topic_required": record.topic_required,
|
||||
"max_members": record.max_members,
|
||||
"nip29_group_id": record.nip29_group_id,
|
||||
"ttl_seconds": record.ttl_seconds,
|
||||
"ttl_deadline": record.ttl_deadline.map(|t| t.to_rfc3339()),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -61,6 +61,13 @@ pub struct Config {
|
||||
pub pubkey_allowlist_enabled: bool,
|
||||
/// Media storage configuration (S3/MinIO).
|
||||
pub media: sprout_media::MediaConfig,
|
||||
|
||||
/// Optional override for ephemeral channel TTL (in seconds).
|
||||
/// When set, any channel created with a TTL tag will use this value instead
|
||||
/// of the client-provided one. Useful for testing ephemeral expiry quickly.
|
||||
/// Example: `SPROUT_EPHEMERAL_TTL_OVERRIDE=60` → all ephemeral channels expire
|
||||
/// 60 seconds after the last message.
|
||||
pub ephemeral_ttl_override: Option<i32>,
|
||||
}
|
||||
|
||||
impl Config {
|
||||
@@ -194,6 +201,18 @@ impl Config {
|
||||
}),
|
||||
};
|
||||
|
||||
let ephemeral_ttl_override = std::env::var("SPROUT_EPHEMERAL_TTL_OVERRIDE")
|
||||
.ok()
|
||||
.and_then(|v| v.parse::<i32>().ok())
|
||||
.filter(|&v| v > 0);
|
||||
|
||||
if let Some(ttl) = ephemeral_ttl_override {
|
||||
warn!(
|
||||
"SPROUT_EPHEMERAL_TTL_OVERRIDE={ttl}s — all ephemeral channels will use \
|
||||
this TTL instead of the client-provided value."
|
||||
);
|
||||
}
|
||||
|
||||
Ok(Self {
|
||||
bind_addr,
|
||||
database_url,
|
||||
@@ -213,6 +232,7 @@ impl Config {
|
||||
metrics_port,
|
||||
pubkey_allowlist_enabled,
|
||||
media,
|
||||
ephemeral_ttl_override,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1029,6 +1029,8 @@ pub async fn ingest_event(
|
||||
}
|
||||
});
|
||||
|
||||
let ttl_seconds = super::resolve_ttl(&event, state.config.ephemeral_ttl_override);
|
||||
|
||||
let actor_bytes = event.pubkey.serialize().to_vec();
|
||||
let (_, was_created) = state
|
||||
.db
|
||||
@@ -1039,6 +1041,7 @@ pub async fn ingest_event(
|
||||
visibility,
|
||||
description.as_deref(),
|
||||
&actor_bytes,
|
||||
ttl_seconds,
|
||||
)
|
||||
.await
|
||||
.map_err(|e| IngestError::Internal(format!("error: {e}")))?;
|
||||
@@ -1304,6 +1307,17 @@ pub async fn ingest_event(
|
||||
});
|
||||
}
|
||||
|
||||
// ── 20b. Bump ephemeral channel TTL deadline ──────────────────────
|
||||
// Any successfully stored channel-scoped event keeps the channel alive.
|
||||
// Skip kind:9007 (create) — the deadline was just set during creation.
|
||||
if let Some(ch_id) = channel_id {
|
||||
if kind_u32 != KIND_NIP29_CREATE_GROUP {
|
||||
if let Err(e) = state.db.bump_ttl_deadline(ch_id).await {
|
||||
warn!(channel = %ch_id, "TTL deadline bump failed: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── 21. Side effects ─────────────────────────────────────────────────
|
||||
if crate::handlers::side_effects::is_side_effect_kind(kind_u32) {
|
||||
if let Err(e) =
|
||||
|
||||
@@ -8,3 +8,29 @@ pub mod ingest;
|
||||
pub mod req;
|
||||
/// NIP-29 and NIP-25 side-effect handlers.
|
||||
pub mod side_effects;
|
||||
|
||||
/// Extract an optional TTL (in seconds) from a Nostr event's `ttl` tag,
|
||||
/// applying the server-side override when configured.
|
||||
///
|
||||
/// Returns `None` when the event carries no `ttl` tag — the channel is permanent.
|
||||
pub fn resolve_ttl(event: &nostr::Event, ephemeral_ttl_override: Option<i32>) -> Option<i32> {
|
||||
let from_tag: Option<i32> = event.tags.iter().find_map(|t| {
|
||||
if t.kind().to_string() == "ttl" {
|
||||
t.content().and_then(|s| s.parse::<i32>().ok())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
});
|
||||
|
||||
match (from_tag, ephemeral_ttl_override) {
|
||||
(Some(original), Some(ovr)) => {
|
||||
tracing::debug!(
|
||||
original,
|
||||
override_val = ovr,
|
||||
"Applying SPROUT_EPHEMERAL_TTL_OVERRIDE"
|
||||
);
|
||||
Some(ovr)
|
||||
}
|
||||
(ttl, _) => ttl,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -948,6 +948,7 @@ async fn handle_create_group(event: &Event, state: &Arc<AppState>) -> anyhow::Re
|
||||
|
||||
let actor_bytes = event.pubkey.serialize().to_vec();
|
||||
let description = extract_tag_value(event, "about");
|
||||
let ttl_seconds = super::resolve_ttl(event, state.config.ephemeral_ttl_override);
|
||||
|
||||
// If the event has an h-tag UUID, ingest_event() already created the channel
|
||||
// via create_channel_with_id(). Fetch it rather than creating a duplicate.
|
||||
@@ -966,6 +967,7 @@ async fn handle_create_group(event: &Event, state: &Arc<AppState>) -> anyhow::Re
|
||||
visibility,
|
||||
description.as_deref(),
|
||||
&actor_bytes,
|
||||
ttl_seconds,
|
||||
)
|
||||
.await?
|
||||
}
|
||||
@@ -979,6 +981,7 @@ async fn handle_create_group(event: &Event, state: &Arc<AppState>) -> anyhow::Re
|
||||
visibility,
|
||||
description.as_deref(),
|
||||
&actor_bytes,
|
||||
ttl_seconds,
|
||||
)
|
||||
.await?
|
||||
};
|
||||
|
||||
@@ -149,6 +149,67 @@ async fn main() -> anyhow::Result<()> {
|
||||
let wf_cron = Arc::clone(&workflow_engine);
|
||||
tokio::spawn(async move { wf_cron.run().await });
|
||||
|
||||
// Ephemeral channel reaper — archives channels whose TTL deadline has passed.
|
||||
// Runs every 60s, matching the workflow cron loop pattern. The SQL UPDATE
|
||||
// uses `archived_at IS NULL` as a guard, so concurrent runs from multiple
|
||||
// pods are harmless (at worst, duplicate system messages — same trade-off
|
||||
// as the workflow cron loop). Will be upgraded to use pg_advisory_lock
|
||||
// together with the workflow engine in a future multi-pod coordination pass.
|
||||
{
|
||||
let reaper_state = Arc::clone(&state);
|
||||
let reaper_interval_secs: u64 = std::env::var("SPROUT_REAPER_INTERVAL_SECS")
|
||||
.ok()
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(60);
|
||||
tokio::spawn(async move {
|
||||
info!(
|
||||
interval_secs = reaper_interval_secs,
|
||||
"Ephemeral channel reaper started"
|
||||
);
|
||||
loop {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(reaper_interval_secs)).await;
|
||||
|
||||
let expired = match reaper_state.db.reap_expired_ephemeral_channels().await {
|
||||
Ok(ids) => ids,
|
||||
Err(e) => {
|
||||
error!("Ephemeral reaper tick failed: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
if expired.is_empty() {
|
||||
continue;
|
||||
}
|
||||
|
||||
info!(count = expired.len(), "Ephemeral reaper archived channels");
|
||||
|
||||
for channel_id in &expired {
|
||||
// Emit a system message so members see why the channel was archived.
|
||||
if let Err(e) = sprout_relay::handlers::side_effects::emit_system_message(
|
||||
&reaper_state,
|
||||
*channel_id,
|
||||
serde_json::json!({ "type": "channel_auto_archived" }),
|
||||
)
|
||||
.await
|
||||
{
|
||||
error!(channel = %channel_id, "reaper system message failed: {e}");
|
||||
}
|
||||
|
||||
// Update NIP-29 discovery events so clients see the archived state.
|
||||
if let Err(e) =
|
||||
sprout_relay::handlers::side_effects::emit_group_discovery_events(
|
||||
&reaper_state,
|
||||
*channel_id,
|
||||
)
|
||||
.await
|
||||
{
|
||||
error!(channel = %channel_id, "reaper discovery update failed: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Multi-node fan-out consumer: receive events from Redis pub/sub
|
||||
// (published by other relay instances) and fan out to local WS subscribers.
|
||||
{
|
||||
|
||||
@@ -48,7 +48,7 @@ const overrides = new Map([
|
||||
["src/features/agents/ui/AgentsView.tsx", 790], // remote agent stop/delete + channel UUID resolution + presence-aware delete guard + persona/team import + provider/model fields
|
||||
["src/features/agents/ui/CreateAgentDialog.tsx", 685], // provider selector + config form + schema-typed config coercion + required field validation + locked scopes
|
||||
["src/features/channels/ui/AddChannelBotDialog.tsx", 640], // provider mode: Run on selector, trust warning, probe effect, single-agent enforcement, provider warnings display
|
||||
["src/shared/api/types.ts", 525], // persona provider/model fields + forum types + workflow type re-exports
|
||||
["src/shared/api/types.ts", 530], // persona provider/model fields + forum types + workflow type re-exports + ephemeral channel TTL fields
|
||||
]);
|
||||
|
||||
async function walkFiles(directory) {
|
||||
|
||||
@@ -48,6 +48,7 @@ pub async fn create_channel(
|
||||
channel_type: String,
|
||||
visibility: String,
|
||||
description: Option<String>,
|
||||
ttl_seconds: Option<i32>,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<ChannelInfo, String> {
|
||||
let channel_uuid = uuid::Uuid::new_v4();
|
||||
@@ -61,8 +62,14 @@ pub async fn create_channel(
|
||||
other => return Err(format!("invalid channel_type: {other}")),
|
||||
};
|
||||
|
||||
let builder =
|
||||
events::build_create_channel(channel_uuid, &name, vis, ct, description.as_deref())?;
|
||||
let builder = events::build_create_channel(
|
||||
channel_uuid,
|
||||
&name,
|
||||
vis,
|
||||
ct,
|
||||
description.as_deref(),
|
||||
ttl_seconds,
|
||||
)?;
|
||||
submit_event(builder, &state).await?;
|
||||
|
||||
// Follow-up GET to return the full ChannelInfo the frontend expects.
|
||||
|
||||
@@ -110,6 +110,7 @@ pub fn build_create_channel(
|
||||
visibility: &str,
|
||||
channel_type: &str,
|
||||
about: Option<&str>,
|
||||
ttl_seconds: Option<i32>,
|
||||
) -> Result<EventBuilder, String> {
|
||||
let mut tags = vec![
|
||||
tag(vec!["h", &channel_id.to_string()])?,
|
||||
@@ -120,6 +121,9 @@ pub fn build_create_channel(
|
||||
if let Some(a) = about {
|
||||
tags.push(tag(vec!["about", a])?);
|
||||
}
|
||||
if let Some(ttl) = ttl_seconds {
|
||||
tags.push(tag(vec!["ttl", &ttl.to_string()])?);
|
||||
}
|
||||
Ok(EventBuilder::new(Kind::Custom(9007), "").tags(tags))
|
||||
}
|
||||
|
||||
|
||||
@@ -70,6 +70,8 @@ pub struct ChannelInfo {
|
||||
pub participant_pubkeys: Vec<String>,
|
||||
#[serde(default = "default_true")]
|
||||
pub is_member: bool,
|
||||
pub ttl_seconds: Option<i32>,
|
||||
pub ttl_deadline: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
@@ -94,6 +96,8 @@ pub struct ChannelDetailInfo {
|
||||
pub topic_required: bool,
|
||||
pub max_members: Option<i32>,
|
||||
pub nip29_group_id: Option<String>,
|
||||
pub ttl_seconds: Option<i32>,
|
||||
pub ttl_deadline: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
|
||||
@@ -449,22 +449,34 @@ export function AppShell() {
|
||||
isOpeningDm={openDmMutation.isPending}
|
||||
isPresencePending={presenceSession.isPending}
|
||||
selfPresenceStatus={presenceSession.currentStatus}
|
||||
onCreateChannel={async ({ description, name, visibility }) => {
|
||||
onCreateChannel={async ({
|
||||
description,
|
||||
name,
|
||||
visibility,
|
||||
ttlSeconds,
|
||||
}) => {
|
||||
const createdChannel = await createChannelMutation.mutateAsync({
|
||||
name,
|
||||
description,
|
||||
channelType: "stream",
|
||||
visibility,
|
||||
ttlSeconds,
|
||||
});
|
||||
|
||||
openChannelView(createdChannel.id);
|
||||
}}
|
||||
onCreateForum={async ({ description, name, visibility }) => {
|
||||
onCreateForum={async ({
|
||||
description,
|
||||
name,
|
||||
visibility,
|
||||
ttlSeconds,
|
||||
}) => {
|
||||
const createdForum = await createForumMutation.mutateAsync({
|
||||
name,
|
||||
description,
|
||||
channelType: "forum",
|
||||
visibility,
|
||||
ttlSeconds,
|
||||
});
|
||||
|
||||
openChannelView(createdForum.id);
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
import { Bot, Home, Lock, PenSquare, Plus, Search, Zap } from "lucide-react";
|
||||
import * as React from "react";
|
||||
|
||||
/** Default TTL for ephemeral channels: 1 day of inactivity. */
|
||||
const EPHEMERAL_TTL_SECONDS = 86400;
|
||||
|
||||
import { useManagedAgentsQuery } from "@/features/agents/hooks";
|
||||
import { getPresenceLabel } from "@/features/presence/lib/presence";
|
||||
import { PresenceDot } from "@/features/presence/ui/PresenceBadge";
|
||||
@@ -68,11 +71,13 @@ type AppSidebarProps = {
|
||||
name: string;
|
||||
description?: string;
|
||||
visibility: ChannelVisibility;
|
||||
ttlSeconds?: number;
|
||||
}) => Promise<void>;
|
||||
onCreateForum: (input: {
|
||||
name: string;
|
||||
description?: string;
|
||||
visibility: ChannelVisibility;
|
||||
ttlSeconds?: number;
|
||||
}) => Promise<void>;
|
||||
onOpenBrowseChannels: () => void;
|
||||
onOpenBrowseForums: () => void;
|
||||
@@ -97,6 +102,7 @@ function useCreateForm(
|
||||
name: string;
|
||||
description?: string;
|
||||
visibility: ChannelVisibility;
|
||||
ttlSeconds?: number;
|
||||
}) => Promise<void>,
|
||||
entityLabel: string,
|
||||
) {
|
||||
@@ -105,6 +111,7 @@ function useCreateForm(
|
||||
const [draftDescription, setDraftDescription] = React.useState("");
|
||||
const [draftVisibility, setDraftVisibility] =
|
||||
React.useState<ChannelVisibility>("open");
|
||||
const [draftEphemeral, setDraftEphemeral] = React.useState(false);
|
||||
const [errorMessage, setErrorMessage] = React.useState<string | undefined>();
|
||||
const inputRef = React.useRef<HTMLInputElement>(null);
|
||||
|
||||
@@ -124,6 +131,7 @@ function useCreateForm(
|
||||
setDraftName("");
|
||||
setDraftDescription("");
|
||||
setDraftVisibility("open");
|
||||
setDraftEphemeral(false);
|
||||
setIsOpen(false);
|
||||
}
|
||||
|
||||
@@ -142,6 +150,11 @@ function useCreateForm(
|
||||
setDraftVisibility(value);
|
||||
}
|
||||
|
||||
function changeEphemeral(value: boolean) {
|
||||
setErrorMessage(undefined);
|
||||
setDraftEphemeral(value);
|
||||
}
|
||||
|
||||
async function handleSubmit(event: React.FormEvent<HTMLFormElement>) {
|
||||
event.preventDefault();
|
||||
|
||||
@@ -158,11 +171,13 @@ function useCreateForm(
|
||||
name,
|
||||
description: description || undefined,
|
||||
visibility: draftVisibility,
|
||||
ttlSeconds: draftEphemeral ? EPHEMERAL_TTL_SECONDS : undefined,
|
||||
});
|
||||
|
||||
setDraftName("");
|
||||
setDraftDescription("");
|
||||
setDraftVisibility("open");
|
||||
setDraftEphemeral(false);
|
||||
setIsOpen(false);
|
||||
} catch (error) {
|
||||
setErrorMessage(
|
||||
@@ -178,6 +193,7 @@ function useCreateForm(
|
||||
draftName,
|
||||
draftDescription,
|
||||
draftVisibility,
|
||||
draftEphemeral,
|
||||
errorMessage,
|
||||
inputRef,
|
||||
toggle,
|
||||
@@ -185,6 +201,7 @@ function useCreateForm(
|
||||
changeName,
|
||||
changeDescription,
|
||||
changeVisibility,
|
||||
changeEphemeral,
|
||||
handleSubmit,
|
||||
};
|
||||
}
|
||||
@@ -311,6 +328,40 @@ function PrivateCheckbox({
|
||||
);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// EphemeralCheckbox — checkbox toggle for auto-archiving channels
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function EphemeralCheckbox({
|
||||
disabled,
|
||||
isEphemeral,
|
||||
onChange,
|
||||
}: {
|
||||
disabled: boolean;
|
||||
isEphemeral: boolean;
|
||||
onChange: (isEphemeral: boolean) => void;
|
||||
}) {
|
||||
const id = React.useId();
|
||||
|
||||
return (
|
||||
<div className="flex items-center gap-2">
|
||||
<Checkbox
|
||||
checked={isEphemeral}
|
||||
disabled={disabled}
|
||||
id={id}
|
||||
onCheckedChange={(checked) => onChange(checked === true)}
|
||||
/>
|
||||
<label
|
||||
className="flex cursor-pointer items-center gap-1.5 text-xs font-medium text-sidebar-foreground/70 select-none peer-disabled:cursor-not-allowed peer-disabled:opacity-50"
|
||||
htmlFor={id}
|
||||
>
|
||||
<Zap className="h-3 w-3" />
|
||||
Ephemeral — auto-archives after 1 day of inactivity
|
||||
</label>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// ChannelGroupSection — unified Channels / Forums section
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -411,6 +462,11 @@ function ChannelGroupSection({
|
||||
}
|
||||
testId={createVisibilityTestId}
|
||||
/>
|
||||
<EphemeralCheckbox
|
||||
disabled={isCreating}
|
||||
isEphemeral={form.draftEphemeral}
|
||||
onChange={form.changeEphemeral}
|
||||
/>
|
||||
<div className="flex items-center gap-2">
|
||||
<Button
|
||||
disabled={isCreating || form.draftName.trim().length === 0}
|
||||
|
||||
@@ -103,6 +103,8 @@ type RawChannel = {
|
||||
participants: string[];
|
||||
participant_pubkeys: string[];
|
||||
is_member?: boolean;
|
||||
ttl_seconds: number | null;
|
||||
ttl_deadline: string | null;
|
||||
};
|
||||
|
||||
type RawChannelDetail = RawChannel & {
|
||||
@@ -353,6 +355,8 @@ function fromRawChannel(channel: RawChannel): Channel {
|
||||
participants: channel.participants,
|
||||
participantPubkeys: channel.participant_pubkeys,
|
||||
isMember: channel.is_member ?? true,
|
||||
ttlSeconds: channel.ttl_seconds,
|
||||
ttlDeadline: channel.ttl_deadline,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -16,6 +16,8 @@ export type Channel = {
|
||||
participants: string[];
|
||||
participantPubkeys: string[];
|
||||
isMember: boolean;
|
||||
ttlSeconds: number | null;
|
||||
ttlDeadline: string | null;
|
||||
};
|
||||
|
||||
export type ChannelDetail = Channel & {
|
||||
@@ -43,6 +45,7 @@ export type CreateChannelInput = {
|
||||
channelType: Exclude<ChannelType, "dm">;
|
||||
visibility: ChannelVisibility;
|
||||
description?: string;
|
||||
ttlSeconds?: number;
|
||||
};
|
||||
|
||||
export type OpenDmInput = {
|
||||
|
||||
+5
-1
@@ -39,13 +39,17 @@ CREATE TABLE channels (
|
||||
purpose TEXT,
|
||||
purpose_set_by BYTEA,
|
||||
purpose_set_at TIMESTAMPTZ,
|
||||
participant_hash BYTEA
|
||||
participant_hash BYTEA,
|
||||
ttl_seconds INT,
|
||||
ttl_deadline TIMESTAMPTZ
|
||||
);
|
||||
|
||||
CREATE INDEX idx_channels_type ON channels (channel_type);
|
||||
CREATE INDEX idx_channels_visibility ON channels (visibility);
|
||||
CREATE INDEX idx_channels_created_by ON channels (created_by);
|
||||
CREATE UNIQUE INDEX idx_channels_dm_hash ON channels (participant_hash);
|
||||
CREATE INDEX idx_channels_ttl_expiry ON channels (ttl_deadline)
|
||||
WHERE ttl_seconds IS NOT NULL AND archived_at IS NULL AND deleted_at IS NULL;
|
||||
|
||||
-- ── Channel members ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
Reference in New Issue
Block a user