Add private product feedback sidecar (#1857)

Signed-off-by: Wes <wesbillman@users.noreply.github.com>
Co-authored-by: Pinky <44b8e82baa6e0e254e0208d68f335c283c94e7b78dd1fa10d5a49d3f13dd0435@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Wes
2026-07-14 10:41:48 -07:00
committed by GitHub
co-authored by Pinky
parent 1aec7ea7a7
commit af190c93e1
9 changed files with 521 additions and 4 deletions
+25
View File
@@ -76,6 +76,11 @@ enum Command {
GenerateKey,
/// Run pending database migrations.
Migrate,
/// Inspect deployment-wide Buzz product feedback.
ProductFeedback {
#[command(subcommand)]
command: ProductFeedbackCommand,
},
/// Emit kind:39000/39002 events for channels missing them.
///
/// Channels created via direct SQL (seed scripts, pre-migration data) won't
@@ -90,6 +95,16 @@ enum Command {
},
}
#[derive(Subcommand)]
enum ProductFeedbackCommand {
/// List feedback across every community as JSON.
List {
/// Maximum records to return.
#[arg(long, default_value_t = 100, value_parser = clap::value_parser!(u16).range(1..=1000))]
limit: u16,
},
}
#[tokio::main]
async fn main() {
// Install the ring CryptoProvider for rustls. The workspace redis TLS
@@ -130,6 +145,9 @@ async fn run(cli: Cli) -> Result<i32> {
Command::AddMember { pubkey, role } => cmd_add_member(pubkey, role).await,
Command::RemoveMember { pubkey, role } => cmd_remove_member(pubkey, role).await,
Command::ListMembers => cmd_list_members().await,
Command::ProductFeedback {
command: ProductFeedbackCommand::List { limit },
} => cmd_list_product_feedback(limit).await,
Command::ReconcileChannels { relay_key } => {
reconcile_channels(relay_key).await?;
Ok(0)
@@ -232,6 +250,13 @@ async fn cmd_remove_member(pubkey_arg: String, role_filter: Option<String>) -> R
Ok(0)
}
async fn cmd_list_product_feedback(limit: u16) -> Result<i32> {
let db = connect_db().await?;
let feedback = db.list_product_feedback(i64::from(limit)).await?;
println!("{}", serde_json::to_string_pretty(&feedback)?);
Ok(0)
}
async fn cmd_list_members() -> Result<i32> {
let db = connect_db().await?;
let tenant = resolve_admin_tenant(&db).await?;
+5
View File
@@ -190,6 +190,10 @@ pub const KIND_MANAGED_AGENT: u32 = 30177;
/// the relay never auto-actions on them (NIP-56).
pub const KIND_REPORT: u32 = 1984;
/// Buzz product feedback submission. Accepted at ingest, sidecarred to the
/// deployment feedback table, and never stored or fanned out as an event.
pub const KIND_PRODUCT_FEEDBACK: u32 = 42000;
// NIP-29 group admin events
/// NIP-29: Add a user to a group.
pub const KIND_NIP29_PUT_USER: u32 = 9000;
@@ -530,6 +534,7 @@ pub const ALL_KINDS: &[u32] = &[
KIND_TEAM,
KIND_MANAGED_AGENT,
KIND_REPORT,
KIND_PRODUCT_FEEDBACK,
KIND_NIP29_PUT_USER,
KIND_NIP29_REMOVE_USER,
KIND_NIP29_EDIT_METADATA,
+19
View File
@@ -31,6 +31,8 @@ pub mod migration;
pub mod moderation;
/// Monthly table partition management.
pub mod partition;
/// Buzz product-feedback sidecar persistence.
pub mod product_feedback;
/// Community-scoped push lease and durable wake-outbox persistence.
pub mod push;
/// Reaction persistence.
@@ -2559,6 +2561,23 @@ impl Db {
relay_members::backfill_from_allowlist(&self.pool, community).await
}
/// Sidecar an accepted product-feedback event, idempotent by event id.
pub async fn insert_product_feedback(
&self,
community: CommunityId,
feedback: product_feedback::NewProductFeedback<'_>,
) -> Result<Uuid> {
product_feedback::insert(&self.pool, community, feedback).await
}
/// List product feedback across the deployment, newest first.
pub async fn list_product_feedback(
&self,
limit: i64,
) -> Result<Vec<product_feedback::ProductFeedbackRecord>> {
product_feedback::list(&self.pool, limit).await
}
/// Insert a tenant-scoped NIP-56 report row, idempotent by report event id.
pub async fn insert_moderation_report(
&self,
+18
View File
@@ -338,6 +338,7 @@ mod tests {
"push_gateway_endpoint_quotas",
"push_gateway_delivery_auth_replays",
"push_gateway_delivery_request_replays",
"product_feedback",
] {
if normalized[insert_pos..].contains(&format!("'{value}'")) {
globals.insert(value.to_owned());
@@ -763,6 +764,23 @@ mod tests {
.sql
.as_str()
.contains("_operator_global_tables"));
// Product feedback is a deployment-private sidecar; community_id is
// provenance, not an operator-review authorization boundary.
assert_eq!(migrations[15].version, 16);
assert!(migrations[15]
.sql
.as_str()
.contains("CREATE TABLE product_feedback"));
assert!(migrations[15]
.sql
.as_str()
.contains("community_id UUID NOT NULL"));
assert!(migrations[15]
.sql
.as_str()
.contains("('product_feedback', 'deployment product inbox"));
assert!(!migrations[0].sql.as_str().contains("product_feedback"));
}
#[test]
+188
View File
@@ -0,0 +1,188 @@
//! Persistence for deployment-level Buzz product feedback.
//!
//! Feedback retains its source [`CommunityId`] as provenance, but is not a
//! community moderation concern and is never inserted into the events table.
use chrono::{DateTime, Utc};
use serde::Serialize;
use sqlx::{PgPool, Row as _};
use uuid::Uuid;
use crate::{error::Result, CommunityId};
/// Validated fields from an accepted product-feedback event.
#[derive(Debug, Clone)]
pub struct NewProductFeedback<'a> {
/// Signed feedback event id (32 bytes), used for idempotency.
pub event_id: &'a [u8],
/// Authenticated submitter's Nostr pubkey (32 bytes).
pub submitter_pubkey: &'a [u8],
/// Optional category from the relay-validated vocabulary.
pub category: Option<&'a str>,
/// Required free-text feedback body.
pub body: &'a str,
/// Full validated event tags (attachments and diagnostics metadata included).
pub tags: &'a serde_json::Value,
/// Timestamp signed into the source event.
pub event_created_at: DateTime<Utc>,
}
/// Product-feedback row returned to deployment-operator tooling.
#[derive(Debug, Clone, Serialize)]
pub struct ProductFeedbackRecord {
/// Sidecar row id.
pub id: Uuid,
/// Source community, retained as provenance only.
pub community_id: Uuid,
/// Signed source event id.
pub event_id: String,
/// Signed submitter pubkey.
pub submitter_pubkey: String,
/// Optional feedback category.
pub category: Option<String>,
/// Feedback body.
pub body: String,
/// Full source tags, including attachment and diagnostics metadata.
pub tags: serde_json::Value,
/// Timestamp signed into the source event.
pub event_created_at: DateTime<Utc>,
/// Time accepted by this deployment.
pub received_at: DateTime<Utc>,
}
/// Insert product feedback, idempotent deployment-wide by signed event id.
///
/// The first accepted submission owns the provenance row. Replaying the exact
/// same signed event through another community returns the same row without
/// changing its source community; callers intentionally receive the same
/// successful acknowledgment in both cases.
pub async fn insert(
pool: &PgPool,
community: CommunityId,
feedback: NewProductFeedback<'_>,
) -> Result<Uuid> {
let row = sqlx::query(
r#"
INSERT INTO product_feedback (
community_id, event_id, submitter_pubkey, category, body, tags,
event_created_at
) VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (event_id) DO UPDATE SET
event_id = EXCLUDED.event_id
RETURNING id
"#,
)
.bind(community.as_uuid())
.bind(feedback.event_id)
.bind(feedback.submitter_pubkey)
.bind(feedback.category)
.bind(feedback.body)
.bind(feedback.tags)
.bind(feedback.event_created_at)
.fetch_one(pool)
.await?;
Ok(row.try_get("id")?)
}
/// List feedback across all communities, newest received first.
pub async fn list(pool: &PgPool, limit: i64) -> Result<Vec<ProductFeedbackRecord>> {
let rows = sqlx::query(
r#"
SELECT id, community_id, event_id, submitter_pubkey, category, body,
tags, event_created_at, received_at
FROM product_feedback
ORDER BY received_at DESC, id
LIMIT $1
"#,
)
.bind(limit)
.fetch_all(pool)
.await?;
rows.into_iter()
.map(|row| {
Ok(ProductFeedbackRecord {
id: row.try_get("id")?,
community_id: row.try_get("community_id")?,
event_id: hex::encode(row.try_get::<Vec<u8>, _>("event_id")?),
submitter_pubkey: hex::encode(row.try_get::<Vec<u8>, _>("submitter_pubkey")?),
category: row.try_get("category")?,
body: row.try_get("body")?,
tags: row.try_get("tags")?,
event_created_at: row.try_get("event_created_at")?,
received_at: row.try_get("received_at")?,
})
})
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
#[ignore = "requires migrated Postgres"]
async fn duplicate_event_keeps_first_community_provenance() {
let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
.or_else(|_| std::env::var("DATABASE_URL"))
.expect("BUZZ_TEST_DATABASE_URL or DATABASE_URL");
let pool = PgPool::connect(&database_url)
.await
.expect("connect test DB");
let first = Uuid::new_v4();
let second = Uuid::new_v4();
for (id, host) in [
(first, format!("feedback-first-{first}.test")),
(second, format!("feedback-second-{second}.test")),
] {
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(id)
.bind(host)
.execute(&pool)
.await
.expect("insert community");
}
let mut event_id = [0_u8; 32];
event_id[..16].copy_from_slice(Uuid::new_v4().as_bytes());
event_id[16..].copy_from_slice(Uuid::new_v4().as_bytes());
let pubkey = [7_u8; 32];
let tags = serde_json::json!([["category", "bug"]]);
let feedback = || NewProductFeedback {
event_id: &event_id,
submitter_pubkey: &pubkey,
category: Some("bug"),
body: "same signed feedback",
tags: &tags,
event_created_at: Utc::now(),
};
let first_row = insert(&pool, CommunityId::from_uuid(first), feedback())
.await
.expect("first insert");
let duplicate_row = insert(&pool, CommunityId::from_uuid(second), feedback())
.await
.expect("duplicate insert");
assert_eq!(duplicate_row, first_row);
let rows: Vec<Uuid> =
sqlx::query_scalar("SELECT community_id FROM product_feedback WHERE event_id = $1")
.bind(event_id.as_slice())
.fetch_all(&pool)
.await
.expect("read feedback provenance");
assert_eq!(rows, vec![first]);
sqlx::query("DELETE FROM product_feedback WHERE event_id = $1")
.bind(event_id.as_slice())
.execute(&pool)
.await
.expect("delete feedback");
sqlx::query("DELETE FROM communities WHERE id = ANY($1)")
.bind(vec![first, second])
.execute(&pool)
.await
.expect("delete communities");
}
}
+91 -4
View File
@@ -28,8 +28,8 @@ use buzz_core::kind::{
KIND_NIP29_DELETE_GROUP, KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST,
KIND_NIP29_LEAVE_REQUEST, KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER,
KIND_NIP43_LEAVE_REQUEST, KIND_NIP65_RELAY_LIST_METADATA, KIND_PERSONA, KIND_PIN_LIST,
KIND_PRESENCE_UPDATE, KIND_PROFILE, KIND_REACTION, KIND_READ_STATE, KIND_REPORT,
KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_DIFF,
KIND_PRESENCE_UPDATE, KIND_PRODUCT_FEEDBACK, KIND_PROFILE, KIND_REACTION, KIND_READ_STATE,
KIND_REPORT, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_DIFF,
KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_PINNED, KIND_STREAM_MESSAGE_SCHEDULED,
KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, KIND_TEAM, KIND_TEXT_NOTE, KIND_USER_STATUS,
KIND_WORKFLOW_DEF, KIND_WORKFLOW_TRIGGER, RELAY_ADMIN_ADD_MEMBER, RELAY_ADMIN_CHANGE_ROLE,
@@ -125,6 +125,22 @@ impl IngestAuth {
}
}
fn emit_product_feedback_success(
tracer: &Arc<dyn buzz_conformance::Tracer>,
tenant: &TenantContext,
event: &Event,
auth: &IngestAuth,
) {
emit(
tracer,
TraceAction::WriteInsertGlobal {
msg_id: msg_id_label(event.id.as_bytes()),
claimed_community: claimed_community_from_event(event),
},
state_for_request(tenant, auth.pubkey()),
);
}
/// Successful ingestion result.
pub struct IngestResult {
/// Hex-encoded event ID.
@@ -172,7 +188,7 @@ fn required_scope_for_kind(kind: u32, event: &Event) -> Result<Scope, &'static s
// NIP-56 reports are ordinary member writes into the mod-only queue.
// Ingest persists them to `moderation_reports` and suppresses public
// storage/fanout; reports are signals, never enforcement triggers.
KIND_REPORT => Ok(Scope::MessagesWrite),
KIND_REPORT | KIND_PRODUCT_FEEDBACK => Ok(Scope::MessagesWrite),
// Community moderation commands are direct, mod-authz-gated writes.
// Scope only proves the transport can submit message writes; the
// command handler owns role/capability authorization.
@@ -1445,6 +1461,23 @@ async fn ingest_event_inner(
return super::command_executor::handle_command(tenant, state, event, auth).await;
}
// Product feedback is sidecarred directly into its private deployment table.
// It never enters ordinary event storage or subscription fan-out.
if kind_u32 == KIND_PRODUCT_FEEDBACK {
super::product_feedback::handle(tenant, &event, state)
.await
.map_err(IngestError::Rejected)?;
// Feedback is a host-resolved, channel-less write. Although its row is
// private to operator tooling rather than ordinary event reads, this is
// the matching modeled success action at the ingest isolation seam.
emit_product_feedback_success(tracer, tenant, &event, &auth);
return Ok(IngestResult {
event_id: event_id_hex,
accepted: true,
message: String::new(),
});
}
// NIP-56 reports are persisted only to the mod queue. They are not stored in
// the public events table and never fan out to subscribers. Reports remain
// available while timed out so users can signal abuse during a write-block.
@@ -2402,12 +2435,64 @@ async fn ingest_event_inner(
#[cfg(test)]
mod tests {
use std::sync::Mutex;
use super::*;
use buzz_conformance::{TraceStep, Tracer};
use buzz_core::kind::{
KIND_CANVAS, KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_LONG_FORM,
KIND_MANAGED_AGENT, KIND_PERSONA, KIND_PRESENCE_UPDATE, KIND_STREAM_MESSAGE,
KIND_STREAM_MESSAGE_DIFF, KIND_TEAM, KIND_USER_STATUS,
};
use nostr::{EventBuilder, Kind};
#[derive(Debug, Default)]
struct VecTracer {
steps: Mutex<Vec<TraceStep>>,
}
impl Tracer for VecTracer {
fn record(&self, step: TraceStep) {
self.steps.lock().expect("trace lock").push(step);
}
}
#[test]
fn feedback_success_action_satisfies_ingest_emit_guard() {
let community = buzz_core::CommunityId::from_uuid(Uuid::new_v4());
let tenant = TenantContext::resolved(community, "feedback.test");
let keys = nostr::Keys::generate();
let event = EventBuilder::new(
Kind::Custom(KIND_PRODUCT_FEEDBACK as u16),
"Useful feedback",
)
.sign_with_keys(&keys)
.expect("sign feedback");
let auth = IngestAuth::Http {
pubkey: keys.public_key(),
scopes: vec![Scope::MessagesWrite],
auth_method: HttpAuthMethod::Nip98,
};
let tracer = Arc::new(VecTracer::default());
let abstract_state = state_for_request(&tenant, auth.pubkey());
{
let (guard, counting) = EmitGuard::arm(
tracer.clone(),
abstract_state.clone(),
"ingest_event_exited_without_trace",
);
emit_product_feedback_success(&counting, &tenant, &event, &auth);
drop(guard);
}
let steps = tracer.steps.lock().expect("trace lock");
assert_eq!(steps.len(), 1);
assert!(matches!(
steps[0].action,
TraceAction::WriteInsertGlobal { .. }
));
}
#[test]
fn nip_ia_requests_are_global_only() {
@@ -2525,10 +2610,11 @@ mod tests {
}
#[test]
fn reports_and_moderation_commands_require_messages_write_scope() {
fn private_sidecars_and_moderation_commands_require_messages_write_scope() {
let dummy = make_dummy_event();
for kind in [
KIND_REPORT,
KIND_PRODUCT_FEEDBACK,
KIND_MODERATION_BAN,
KIND_MODERATION_UNBAN,
KIND_MODERATION_TIMEOUT,
@@ -2617,6 +2703,7 @@ mod tests {
KIND_DELETION,
KIND_REACTION,
KIND_REPORT,
KIND_PRODUCT_FEEDBACK,
KIND_MODERATION_BAN,
KIND_MODERATION_UNBAN,
KIND_MODERATION_TIMEOUT,
+2
View File
@@ -24,6 +24,8 @@ pub mod moderation_authz;
pub mod moderation_commands;
/// Relay-signed moderation notice DMs.
pub mod moderation_notices;
/// Product-feedback validation + deployment sidecar persistence.
pub mod product_feedback;
#[allow(dead_code, missing_docs)]
pub mod push_lease;
/// NIP-43 relay membership admin command handler (kinds 9030–9032).
@@ -0,0 +1,149 @@
//! Product-feedback event validation and sidecar persistence.
use std::sync::Arc;
use buzz_core::tenant::TenantContext;
use buzz_db::product_feedback::NewProductFeedback;
use nostr::Event;
use crate::state::AppState;
const CATEGORIES: &[&str] = &["bug", "praise", "needs-work"];
const MAX_BODY_BYTES: usize = 32 * 1024;
const MAX_TAGS_BYTES: usize = 64 * 1024;
/// Validate and persist a product-feedback event outside ordinary event storage.
pub async fn handle(
tenant: &TenantContext,
event: &Event,
state: &Arc<AppState>,
) -> Result<(), String> {
let category = parse_category(event)?;
validate_body(&event.content)?;
let tags = serialize_tags(event)?;
let event_created_at =
chrono::DateTime::from_timestamp(event.created_at.as_secs() as i64, 0)
.ok_or_else(|| "invalid: feedback timestamp is out of range".to_string())?;
state
.db
.insert_product_feedback(
tenant.community(),
NewProductFeedback {
event_id: event.id.as_bytes(),
submitter_pubkey: &event.pubkey.to_bytes(),
category,
body: &event.content,
tags: &tags,
event_created_at,
},
)
.await
.map_err(|e| format!("error: database error inserting product feedback: {e}"))?;
Ok(())
}
fn serialize_tags(event: &Event) -> Result<serde_json::Value, String> {
let tags_bytes = serde_json::to_vec(&event.tags)
.map_err(|e| format!("error: failed to serialize feedback tags: {e}"))?;
if tags_bytes.len() > MAX_TAGS_BYTES {
return Err(format!(
"invalid: feedback tags exceed maximum size of {MAX_TAGS_BYTES} bytes"
));
}
serde_json::from_slice(&tags_bytes)
.map_err(|e| format!("error: failed to deserialize feedback tags: {e}"))
}
fn parse_category(event: &Event) -> Result<Option<&str>, String> {
let values: Vec<&str> = event
.tags
.iter()
.filter(|tag| tag.kind().to_string() == "category")
.filter_map(|tag| tag.content())
.collect();
match values.as_slice() {
[] => Ok(None),
[category] if CATEGORIES.contains(category) => Ok(Some(category)),
[_] => Err("invalid: unsupported feedback category".to_string()),
_ => Err("invalid: feedback must include at most one category tag".to_string()),
}
}
fn validate_body(body: &str) -> Result<(), String> {
if body.trim().is_empty() {
return Err("invalid: feedback body must not be empty".to_string());
}
if body.len() > MAX_BODY_BYTES {
return Err(format!(
"invalid: feedback body exceeds maximum size of {MAX_BODY_BYTES} bytes"
));
}
Ok(())
}
#[cfg(test)]
mod tests {
use nostr::{Event, EventBuilder, Keys, Kind, Tag};
use super::{parse_category, serialize_tags, validate_body};
use buzz_core::kind::KIND_PRODUCT_FEEDBACK;
fn feedback(tags: Vec<Tag>) -> Event {
EventBuilder::new(
Kind::Custom(KIND_PRODUCT_FEEDBACK as u16),
"Useful feedback",
)
.tags(tags)
.sign_with_keys(&Keys::generate())
.expect("sign feedback")
}
#[test]
fn accepts_supported_or_absent_category() {
assert_eq!(parse_category(&feedback(vec![])).unwrap(), None);
let event = feedback(vec![Tag::parse(["category", "bug"]).unwrap()]);
assert_eq!(parse_category(&event).unwrap(), Some("bug"));
}
#[test]
fn rejects_unknown_or_duplicate_category() {
let unknown = feedback(vec![Tag::parse(["category", "idea"]).unwrap()]);
assert!(parse_category(&unknown)
.unwrap_err()
.contains("unsupported"));
let duplicate = feedback(vec![
Tag::parse(["category", "bug"]).unwrap(),
Tag::parse(["category", "praise"]).unwrap(),
]);
assert!(parse_category(&duplicate)
.unwrap_err()
.contains("at most one"));
}
#[test]
fn body_must_be_nonempty_and_bounded() {
assert!(validate_body(" \n ").is_err());
assert!(validate_body("works").is_ok());
assert!(validate_body(&"x".repeat(32 * 1024 + 1)).is_err());
}
#[test]
fn tags_are_serialized_and_bounded() {
let small = feedback(vec![Tag::parse([
"imeta",
"url https://example.test/a.png",
])
.unwrap()]);
assert!(serialize_tags(&small).is_ok());
let oversized = feedback(vec![
Tag::parse(["diagnostics", &"x".repeat(64 * 1024)]).unwrap()
]);
assert!(serialize_tags(&oversized)
.unwrap_err()
.contains("tags exceed maximum size"));
}
}
+24
View File
@@ -0,0 +1,24 @@
-- Buzz product feedback is accepted through a dedicated signed event kind and
-- sidecarred here instead of entering the ordinary events table. Rows remain
-- attributable to their source community, while deployment operators may
-- review the table across communities through internal tooling.
CREATE TABLE product_feedback (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
community_id UUID NOT NULL REFERENCES communities(id),
event_id BYTEA NOT NULL CHECK (length(event_id) = 32),
submitter_pubkey BYTEA NOT NULL CHECK (length(submitter_pubkey) = 32),
category TEXT CHECK (category IN ('bug', 'praise', 'needs-work')),
body TEXT NOT NULL CHECK (length(btrim(body)) > 0),
tags JSONB NOT NULL DEFAULT '[]'::jsonb CHECK (jsonb_typeof(tags) = 'array'),
event_created_at TIMESTAMPTZ NOT NULL,
received_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (event_id)
);
CREATE INDEX idx_product_feedback_received
ON product_feedback (received_at DESC, id);
CREATE INDEX idx_product_feedback_community_received
ON product_feedback (community_id, received_at DESC, id);
INSERT INTO _operator_global_tables (table_name, reason) VALUES
('product_feedback', 'deployment product inbox; community_id is provenance only');