jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-02 07:22:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(db): move feed store ownership

Browse Source 
Signed-off-by: OpenAI Codex <codex@openai.com>
This commit is contained in:
OpenAI Codex
2026-08-26 14:53:44 +00:00
committed by Luke Tornquist
parent ae9ce79afc
commit b2dc379dfb
2 changed files with 231 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files

232
crates/buzz-db/src/feed.rs
View File

@@ -28,6 +28,7 @@
/// before the query is issued so the SQL `LIMIT` clause always reflects this cap.
pub const FEED_MAX_LIMIT: i64 = 100;
use buzz_datastore_tracing::datastore_span;
use chrono::{DateTime, Utc};
use sqlx::postgres::PgRow;
use sqlx::{PgPool, QueryBuilder};
@@ -41,8 +42,8 @@ use buzz_core::kind::{
};
use buzz_core::{CommunityId, StoredEvent};
use crate::error::Result;
use crate::event::row_to_stored_event;
use crate::{error::Result, Db, RouteDecision, RoutePredicate};
/// Column list shared by every feed subquery that aliases the `events` table as `e`.
const EVENT_COLS: &str =
@@ -303,6 +304,235 @@ pub(crate) async fn query_activity_on(
collect_stored_events(rows)
}
// -- Db API -------------------------------------------------------------------
impl Db {
/// Find events that @mention the given pubkey.
#[datastore_span(name = "query_feed_mentions", system = "postgresql")]
pub async fn query_feed_mentions(
&self,
community: CommunityId,
pubkey_bytes: &[u8],
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
crate::feed::query_mentions(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
/// [`Db::query_feed_mentions`] with replica routing — same contract and
/// classification-table requirement as [`Db::query_events_routed`].
///
/// Feed queries route on the BOUNDED arm only: the `accessible_channel_ids`
/// parameter admits community-global rows alongside channel rows, so no
/// single channel's fence floor can prove completeness — the covered arm
/// is structurally unavailable, not merely unchosen.
#[datastore_span(name = "query_feed_mentions_routed", system = "postgresql")]
pub async fn query_feed_mentions_routed(
&self,
path: &'static str,
community: CommunityId,
pubkey_bytes: &[u8],
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
match self.route_read(path, RoutePredicate::Bounded).await {
RouteDecision::Replica(mut tx, _entry, reason) => match crate::feed::query_mentions_on(
&mut tx,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
{
Ok(events) => {
Self::record_route(path, "replica", reason);
Ok(events)
}
Err(e) => {
tracing::warn!(path, "replica read failed; re-running on writer: {e}");
Self::record_route(path, "writer", "replica_error");
crate::feed::query_mentions(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
},
RouteDecision::Writer => {
crate::feed::query_mentions(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
/// Find events that require action from the given pubkey.
#[datastore_span(name = "query_feed_needs_action", system = "postgresql")]
pub async fn query_feed_needs_action(
&self,
community: CommunityId,
pubkey_bytes: &[u8],
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
crate::feed::query_needs_action(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
/// [`Db::query_feed_needs_action`] with replica routing — BOUNDED arm
/// only; see [`Db::query_feed_mentions_routed`] for why the covered arm
/// is structurally unavailable to feed queries.
#[datastore_span(name = "query_feed_needs_action_routed", system = "postgresql")]
pub async fn query_feed_needs_action_routed(
&self,
path: &'static str,
community: CommunityId,
pubkey_bytes: &[u8],
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
match self.route_read(path, RoutePredicate::Bounded).await {
RouteDecision::Replica(mut tx, _entry, reason) => {
match crate::feed::query_needs_action_on(
&mut tx,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
{
Ok(events) => {
Self::record_route(path, "replica", reason);
Ok(events)
}
Err(e) => {
tracing::warn!(path, "replica read failed; re-running on writer: {e}");
Self::record_route(path, "writer", "replica_error");
crate::feed::query_needs_action(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
RouteDecision::Writer => {
crate::feed::query_needs_action(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
/// Find recent activity across accessible channels.
#[datastore_span(name = "query_feed_activity", system = "postgresql")]
pub async fn query_feed_activity(
&self,
community: CommunityId,
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
crate::feed::query_activity(&self.pool, community, accessible_channel_ids, since, limit)
.await
}
/// [`Db::query_feed_activity`] with replica routing — BOUNDED arm only;
/// see [`Db::query_feed_mentions_routed`] for why the covered arm is
/// structurally unavailable to feed queries.
#[datastore_span(name = "query_feed_activity_routed", system = "postgresql")]
pub async fn query_feed_activity_routed(
&self,
path: &'static str,
community: CommunityId,
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
match self.route_read(path, RoutePredicate::Bounded).await {
RouteDecision::Replica(mut tx, _entry, reason) => match crate::feed::query_activity_on(
&mut tx,
community,
accessible_channel_ids,
since,
limit,
)
.await
{
Ok(events) => {
Self::record_route(path, "replica", reason);
Ok(events)
}
Err(e) => {
tracing::warn!(path, "replica read failed; re-running on writer: {e}");
Self::record_route(path, "writer", "replica_error");
crate::feed::query_activity(
&self.pool,
community,
accessible_channel_ids,
since,
limit,
)
.await
}
},
RouteDecision::Writer => {
crate::feed::query_activity(
&self.pool,
community,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
}
// -- Tests --------------------------------------------------------------------
#[cfg(test)]

222
crates/buzz-db/src/lib.rs
View File

@@ -1665,228 +1665,6 @@ impl Db {
}
}
/// Find events that @mention the given pubkey.
#[datastore_span(name = "query_feed_mentions", system = "postgresql")]
pub async fn query_feed_mentions(
&self,
community: CommunityId,
pubkey_bytes: &[u8],
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
feed::query_mentions(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
/// [`Db::query_feed_mentions`] with replica routing — same contract and
/// classification-table requirement as [`Db::query_events_routed`].
///
/// Feed queries route on the BOUNDED arm only: the `accessible_channel_ids`
/// parameter admits community-global rows alongside channel rows, so no
/// single channel's fence floor can prove completeness — the covered arm
/// is structurally unavailable, not merely unchosen.
#[datastore_span(name = "query_feed_mentions_routed", system = "postgresql")]
pub async fn query_feed_mentions_routed(
&self,
path: &'static str,
community: CommunityId,
pubkey_bytes: &[u8],
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
match self.route_read(path, RoutePredicate::Bounded).await {
RouteDecision::Replica(mut tx, _entry, reason) => {
match feed::query_mentions_on(
&mut tx,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
{
Ok(events) => {
Self::record_route(path, "replica", reason);
Ok(events)
}
Err(e) => {
tracing::warn!(path, "replica read failed; re-running on writer: {e}");
Self::record_route(path, "writer", "replica_error");
feed::query_mentions(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
RouteDecision::Writer => {
feed::query_mentions(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
/// Find events that require action from the given pubkey.
#[datastore_span(name = "query_feed_needs_action", system = "postgresql")]
pub async fn query_feed_needs_action(
&self,
community: CommunityId,
pubkey_bytes: &[u8],
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
feed::query_needs_action(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
/// [`Db::query_feed_needs_action`] with replica routing — BOUNDED arm
/// only; see [`Db::query_feed_mentions_routed`] for why the covered arm
/// is structurally unavailable to feed queries.
#[datastore_span(name = "query_feed_needs_action_routed", system = "postgresql")]
pub async fn query_feed_needs_action_routed(
&self,
path: &'static str,
community: CommunityId,
pubkey_bytes: &[u8],
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
match self.route_read(path, RoutePredicate::Bounded).await {
RouteDecision::Replica(mut tx, _entry, reason) => {
match feed::query_needs_action_on(
&mut tx,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
{
Ok(events) => {
Self::record_route(path, "replica", reason);
Ok(events)
}
Err(e) => {
tracing::warn!(path, "replica read failed; re-running on writer: {e}");
Self::record_route(path, "writer", "replica_error");
feed::query_needs_action(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
RouteDecision::Writer => {
feed::query_needs_action(
&self.pool,
community,
pubkey_bytes,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
/// Find recent activity across accessible channels.
#[datastore_span(name = "query_feed_activity", system = "postgresql")]
pub async fn query_feed_activity(
&self,
community: CommunityId,
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
feed::query_activity(&self.pool, community, accessible_channel_ids, since, limit).await
}
/// [`Db::query_feed_activity`] with replica routing — BOUNDED arm only;
/// see [`Db::query_feed_mentions_routed`] for why the covered arm is
/// structurally unavailable to feed queries.
#[datastore_span(name = "query_feed_activity_routed", system = "postgresql")]
pub async fn query_feed_activity_routed(
&self,
path: &'static str,
community: CommunityId,
accessible_channel_ids: &[Uuid],
since: Option<DateTime<Utc>>,
limit: i64,
) -> Result<Vec<StoredEvent>> {
match self.route_read(path, RoutePredicate::Bounded).await {
RouteDecision::Replica(mut tx, _entry, reason) => {
match feed::query_activity_on(
&mut tx,
community,
accessible_channel_ids,
since,
limit,
)
.await
{
Ok(events) => {
Self::record_route(path, "replica", reason);
Ok(events)
}
Err(e) => {
tracing::warn!(path, "replica read failed; re-running on writer: {e}");
Self::record_route(path, "writer", "replica_error");
feed::query_activity(
&self.pool,
community,
accessible_channel_ids,
since,
limit,
)
.await
}
}
}
RouteDecision::Writer => {
feed::query_activity(&self.pool, community, accessible_channel_ids, since, limit)
.await
}
}
}
/// Create a new workflow.
#[datastore_span(name = "create_workflow", system = "postgresql")]
pub async fn create_workflow(