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

feat(db): complete SQLite reaction and feed paths

Browse Source 
Signed-off-by: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>
This commit is contained in:
npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je
2026-08-10 12:11:35 -04:00
committed by Brother Darryl
parent 2d433a4d99
commit b0e7be0bd1
2 changed files with 282 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/buzz-db/src/lib.rs
+163 -65
View File
@@ -3348,16 +3348,32 @@ impl Db {
emoji: &str,
reaction_event_id: Option<&[u8]>,
) -> Result<bool> {
reaction::add_reaction(
self.pg_pool()?,
community,
event_id,
event_created_at,
pubkey,
emoji,
reaction_event_id,
)
.await
match &self.backend {
DbBackend::SQLite(pool) => {
sqlite::add_reaction(
pool,
community,
event_id,
event_created_at,
pubkey,
emoji,
reaction_event_id,
)
.await
}
DbBackend::Postgres => {
reaction::add_reaction(
self.pg_pool()?,
community,
event_id,
event_created_at,
pubkey,
emoji,
reaction_event_id,
)
.await
}
}
}
/// Soft-delete a reaction.
@@ -3414,15 +3430,30 @@ impl Db {
pubkey: &[u8],
emoji: &str,
) -> Result<Option<reaction::ActiveReactionRecord>> {
reaction::get_active_reaction_record(
self.pg_pool()?,
community,
event_id,
event_created_at,
pubkey,
emoji,
)
.await
match &self.backend {
DbBackend::SQLite(pool) => {
sqlite::get_active_reaction_record(
pool,
community,
event_id,
event_created_at,
pubkey,
emoji,
)
.await
}
DbBackend::Postgres => {
reaction::get_active_reaction_record(
self.pg_pool()?,
community,
event_id,
event_created_at,
pubkey,
emoji,
)
.await
}
}
}
/// Backfill the source event ID on an active reaction row.
@@ -3435,16 +3466,32 @@ impl Db {
emoji: &str,
reaction_event_id: &[u8],
) -> Result<bool> {
reaction::set_reaction_event_id(
self.pg_pool()?,
community,
event_id,
event_created_at,
pubkey,
emoji,
reaction_event_id,
)
.await
match &self.backend {
DbBackend::SQLite(pool) => {
sqlite::set_reaction_event_id(
pool,
community,
event_id,
event_created_at,
pubkey,
emoji,
reaction_event_id,
)
.await
}
DbBackend::Postgres => {
reaction::set_reaction_event_id(
self.pg_pool()?,
community,
event_id,
event_created_at,
pubkey,
emoji,
reaction_event_id,
)
.await
}
}
}
/// Get all active reactions for an event, grouped by emoji.
@@ -3456,15 +3503,23 @@ impl Db {
limit: u32,
cursor: Option<&str>,
) -> Result<Vec<reaction::ReactionGroup>> {
reaction::get_reactions(
self.pg_pool()?,
community,
event_id,
event_created_at,
limit,
cursor,
)
.await
match &self.backend {
DbBackend::SQLite(pool) => {
sqlite::get_reactions(pool, community, event_id, event_created_at, limit, cursor)
.await
}
DbBackend::Postgres => {
reaction::get_reactions(
self.pg_pool()?,
community,
event_id,
event_created_at,
limit,
cursor,
)
.await
}
}
}
/// Batch-fetch emoji counts for a set of (event_id, event_created_at) pairs.
@@ -3473,7 +3528,12 @@ impl Db {
community: CommunityId,
event_ids: &[(&[u8], DateTime<Utc>)],
) -> Result<Vec<reaction::BulkReactionEntry>> {
reaction::get_reactions_bulk(self.pg_pool()?, community, event_ids).await
match &self.backend {
DbBackend::SQLite(pool) => sqlite::get_reactions_bulk(pool, community, event_ids).await,
DbBackend::Postgres => {
reaction::get_reactions_bulk(self.pg_pool()?, community, event_ids).await
}
}
}
/// Find events that @mention the given pubkey.
@@ -3485,15 +3545,30 @@ impl Db {
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
feed::query_mentions(
self.pg_pool()?,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
match &self.backend {
DbBackend::SQLite(pool) => {
sqlite::query_feed_mentions(
pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
DbBackend::Postgres => {
feed::query_mentions(
self.pg_pool()?,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
/// [`Db::query_feed_mentions`] with replica routing — same contract and
@@ -3577,15 +3652,30 @@ impl Db {
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
feed::query_needs_action(
self.pg_pool()?,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
match &self.backend {
DbBackend::SQLite(pool) => {
sqlite::query_feed_needs_action(
pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
DbBackend::Postgres => {
feed::query_needs_action(
self.pg_pool()?,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
/// [`Db::query_feed_needs_action`] with replica routing — BOUNDED arm
@@ -3664,14 +3754,22 @@ impl Db {
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
feed::query_activity(
self.pg_pool()?,
community,
accessible_channel_ids,
since,
limit,
)
.await
match &self.backend {
DbBackend::SQLite(pool) => {
sqlite::query_feed_activity(pool, community, accessible_channel_ids, since, limit)
.await
}
DbBackend::Postgres => {
feed::query_activity(
self.pg_pool()?,
community,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
/// [`Db::query_feed_activity`] with replica routing — BOUNDED arm only;
crates/buzz-db/src/sqlite.rs
+119 -4
View File
@@ -205,10 +205,10 @@ pub(crate) async fn migrate(pool: &SqlitePool) -> Result<()> {
let mut tx = pool.begin().await?;
for statement in [
"CREATE VIRTUAL TABLE IF NOT EXISTS events_fts USING fts5(content, content='events', content_rowid='rowid', tokenize='unicode61')",
"CREATE TRIGGER IF NOT EXISTS events_fts_insert AFTER INSERT ON events WHEN new.kind NOT IN (1059, 30300, 30350, 30622, 44100, 44101, 44200) BEGIN INSERT INTO events_fts(rowid, content) VALUES (new.rowid, new.content); END",
"CREATE TRIGGER IF NOT EXISTS events_fts_delete AFTER DELETE ON events WHEN old.kind NOT IN (1059, 30300, 30350, 30622, 44100, 44101, 44200) BEGIN INSERT INTO events_fts(events_fts, rowid, content) VALUES ('delete', old.rowid, old.content); END",
"CREATE TRIGGER IF NOT EXISTS events_fts_update AFTER UPDATE OF content, kind ON events BEGIN INSERT INTO events_fts(events_fts, rowid, content) SELECT 'delete', old.rowid, old.content WHERE old.kind NOT IN (1059, 30300, 30350, 30622, 44100, 44101, 44200); INSERT INTO events_fts(rowid, content) SELECT new.rowid, new.content WHERE new.kind NOT IN (1059, 30300, 30350, 30622, 44100, 44101, 44200); END",
"INSERT INTO events_fts(rowid, content) SELECT rowid, content FROM events WHERE kind NOT IN (1059, 30300, 30350, 30622, 44100, 44101, 44200)",
"CREATE TRIGGER IF NOT EXISTS events_fts_insert AFTER INSERT ON events WHEN new.kind IN (0, 9, 40002, 45001, 45003) BEGIN INSERT INTO events_fts(rowid, content) VALUES (new.rowid, new.content); END",
"CREATE TRIGGER IF NOT EXISTS events_fts_delete AFTER DELETE ON events WHEN old.kind IN (0, 9, 40002, 45001, 45003) BEGIN INSERT INTO events_fts(events_fts, rowid, content) VALUES ('delete', old.rowid, old.content); END",
"CREATE TRIGGER IF NOT EXISTS events_fts_update AFTER UPDATE OF content, kind ON events BEGIN INSERT INTO events_fts(events_fts, rowid, content) SELECT 'delete', old.rowid, old.content WHERE old.kind IN (0, 9, 40002, 45001, 45003); INSERT INTO events_fts(rowid, content) SELECT new.rowid, new.content WHERE new.kind IN (0, 9, 40002, 45001, 45003); END",
"INSERT INTO events_fts(rowid, content) SELECT rowid, content FROM events WHERE kind IN (0, 9, 40002, 45001, 45003)",
] {
sqlx::query(statement).execute(&mut *tx).await?;
}
@@ -661,6 +661,121 @@ pub(crate) async fn insert_reaction_event(
})
}
pub(crate) async fn add_reaction(
pool: &SqlitePool,
community: CommunityId,
event_id: &[u8],
event_created_at: chrono::DateTime<chrono::Utc>,
pubkey: &[u8],
emoji: &str,
reaction_event_id: Option<&[u8]>,
) -> Result<bool> {
let result = sqlx::query("INSERT INTO reactions (community_id,event_created_at,event_id,pubkey,emoji,reaction_event_id) VALUES (?1,?2,?3,?4,?5,?6) ON CONFLICT (community_id,event_created_at,event_id,pubkey,emoji) DO UPDATE SET removed_at=NULL, reaction_event_id=COALESCE(excluded.reaction_event_id,reactions.reaction_event_id) WHERE reactions.removed_at IS NOT NULL")
.bind(community.as_uuid().to_string()).bind(event_created_at.timestamp()).bind(event_id).bind(pubkey).bind(emoji).bind(reaction_event_id).execute(pool).await?;
Ok(result.rows_affected() != 0)
}
pub(crate) async fn get_active_reaction_record(
pool: &SqlitePool,
community: CommunityId,
event_id: &[u8],
event_created_at: chrono::DateTime<chrono::Utc>,
pubkey: &[u8],
emoji: &str,
) -> Result<Option<crate::reaction::ActiveReactionRecord>> {
let row = sqlx::query("SELECT reaction_event_id FROM reactions WHERE community_id=?1 AND event_id=?2 AND event_created_at=?3 AND pubkey=?4 AND emoji=?5 AND removed_at IS NULL LIMIT 1")
.bind(community.as_uuid().to_string()).bind(event_id).bind(event_created_at.timestamp()).bind(pubkey).bind(emoji).fetch_optional(pool).await?;
row.map(|r| {
Ok(crate::reaction::ActiveReactionRecord {
reaction_event_id: r.try_get("reaction_event_id")?,
})
})
.transpose()
}
pub(crate) async fn set_reaction_event_id(
pool: &SqlitePool,
community: CommunityId,
event_id: &[u8],
event_created_at: chrono::DateTime<chrono::Utc>,
pubkey: &[u8],
emoji: &str,
reaction_event_id: &[u8],
) -> Result<bool> {
Ok(sqlx::query("UPDATE reactions SET reaction_event_id=?1 WHERE community_id=?2 AND event_created_at=?3 AND event_id=?4 AND pubkey=?5 AND emoji=?6 AND removed_at IS NULL")
.bind(reaction_event_id).bind(community.as_uuid().to_string()).bind(event_created_at.timestamp()).bind(event_id).bind(pubkey).bind(emoji).execute(pool).await?.rows_affected() > 0)
}
pub(crate) async fn get_reactions(
pool: &SqlitePool,
community: CommunityId,
event_id: &[u8],
event_created_at: chrono::DateTime<chrono::Utc>,
limit: u32,
_cursor: Option<&str>,
) -> Result<Vec<crate::reaction::ReactionGroup>> {
let rows = sqlx::query("SELECT r.emoji,r.pubkey,r.reaction_event_id FROM reactions r WHERE r.community_id=?1 AND r.event_id=?2 AND r.event_created_at=?3 AND r.removed_at IS NULL AND r.emoji IN (SELECT emoji FROM reactions WHERE community_id=?1 AND event_id=?2 AND event_created_at=?3 AND removed_at IS NULL GROUP BY emoji ORDER BY emoji LIMIT ?4) ORDER BY r.emoji,r.rowid")
.bind(community.as_uuid().to_string()).bind(event_id).bind(event_created_at.timestamp()).bind(limit as i64).fetch_all(pool).await?;
let mut groups = Vec::new();
let mut current: Option<String> = None;
let mut users = Vec::new();
for row in rows {
let emoji: String = row.try_get("emoji")?;
if current.as_ref() != Some(&emoji) {
if let Some(e) = current.take() {
groups.push(crate::reaction::ReactionGroup {
emoji: e,
count: users.len() as i64,
users: std::mem::take(&mut users),
});
}
current = Some(emoji);
}
users.push(crate::reaction::ReactionUser {
pubkey: row.try_get("pubkey")?,
display_name: None,
reaction_event_id: row.try_get("reaction_event_id")?,
});
}
if let Some(e) = current {
groups.push(crate::reaction::ReactionGroup {
emoji: e,
count: users.len() as i64,
users,
});
}
Ok(groups)
}
pub(crate) async fn get_reactions_bulk(
pool: &SqlitePool,
community: CommunityId,
event_ids: &[(&[u8], chrono::DateTime<chrono::Utc>)],
) -> Result<Vec<crate::reaction::BulkReactionEntry>> {
let mut out = Vec::new();
for (event_id, ts) in event_ids {
let rows=sqlx::query("SELECT emoji,COUNT(*) AS count FROM reactions WHERE community_id=?1 AND event_id=?2 AND event_created_at=?3 AND removed_at IS NULL GROUP BY emoji ORDER BY emoji").bind(community.as_uuid().to_string()).bind(*event_id).bind(ts.timestamp()).fetch_all(pool).await?;
if rows.is_empty() {
continue;
}
let reactions = rows
.into_iter()
.map(|r| {
Ok(crate::reaction::ReactionSummary {
emoji: r.try_get("emoji")?,
count: r.try_get("count")?,
})
})
.collect::<Result<Vec<_>>>()?;
out.push(crate::reaction::BulkReactionEntry {
event_id: event_id.to_vec(),
event_created_at: *ts,
reactions,
});
}
Ok(out)
}
pub(crate) async fn remove_reaction(
pool: &SqlitePool,
community: CommunityId,