Skip to content

Commit bddfdb9

Browse files
codexTheSentinel454
codex
authored and
TheSentinel454
committed
refactor(db): move feed store ownership
Signed-off-by: OpenAI Codex <codex@openai.com>
1 parent c0c5ba0 commit bddfdb9

2 files changed

Lines changed: 231 additions & 223 deletions

File tree

‎crates/buzz-db/src/feed.rs‎

Lines changed: 231 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
/// before the query is issued so the SQL `LIMIT` clause always reflects this cap.
2929
pub const FEED_MAX_LIMIT: i64 = 100;
3030

31+
use buzz_datastore_tracing::datastore_span;
3132
use chrono::{DateTime, Utc};
3233
use sqlx::postgres::PgRow;
3334
use sqlx::{PgPool, QueryBuilder};
@@ -41,8 +42,8 @@ use buzz_core::kind::{
4142
};
4243
use buzz_core::{CommunityId, StoredEvent};
4344

44-
use crate::error::Result;
4545
use crate::event::row_to_stored_event;
46+
use crate::{error::Result, Db, RouteDecision, RoutePredicate};
4647

4748
/// Column list shared by every feed subquery that aliases the `events` table as `e`.
4849
const EVENT_COLS: &str =
@@ -303,6 +304,235 @@ pub(crate) async fn query_activity_on(
303304
collect_stored_events(rows)
304305
}
305306

307+
// -- Db API -------------------------------------------------------------------
308+
309+
impl Db {
310+
/// Find events that @mention the given pubkey.
311+
#[datastore_span(name = "query_feed_mentions", system = "postgresql")]
312+
pub async fn query_feed_mentions(
313+
&self,
314+
community: CommunityId,
315+
pubkey_bytes: &[u8],
316+
accessible_channel_ids: &[Uuid],
317+
since: Option<DateTime<Utc>>,
318+
limit: i64,
319+
) -> Result<Vec<StoredEvent>> {
320+
crate::feed::query_mentions(
321+
&self.pool,
322+
community,
323+
pubkey_bytes,
324+
accessible_channel_ids,
325+
since,
326+
limit,
327+
)
328+
.await
329+
}
330+
331+
/// [`Db::query_feed_mentions`] with replica routing — same contract and
332+
/// classification-table requirement as [`Db::query_events_routed`].
333+
///
334+
/// Feed queries route on the BOUNDED arm only: the `accessible_channel_ids`
335+
/// parameter admits community-global rows alongside channel rows, so no
336+
/// single channel's fence floor can prove completeness — the covered arm
337+
/// is structurally unavailable, not merely unchosen.
338+
#[datastore_span(name = "query_feed_mentions_routed", system = "postgresql")]
339+
pub async fn query_feed_mentions_routed(
340+
&self,
341+
path: &'static str,
342+
community: CommunityId,
343+
pubkey_bytes: &[u8],
344+
accessible_channel_ids: &[Uuid],
345+
since: Option<DateTime<Utc>>,
346+
limit: i64,
347+
) -> Result<Vec<StoredEvent>> {
348+
match self.route_read(path, RoutePredicate::Bounded).await {
349+
RouteDecision::Replica(mut tx, _entry, reason) => match crate::feed::query_mentions_on(
350+
&mut tx,
351+
community,
352+
pubkey_bytes,
353+
accessible_channel_ids,
354+
since,
355+
limit,
356+
)
357+
.await
358+
{
359+
Ok(events) => {
360+
Self::record_route(path, "replica", reason);
361+
Ok(events)
362+
}
363+
Err(e) => {
364+
tracing::warn!(path, "replica read failed; re-running on writer: {e}");
365+
Self::record_route(path, "writer", "replica_error");
366+
crate::feed::query_mentions(
367+
&self.pool,
368+
community,
369+
pubkey_bytes,
370+
accessible_channel_ids,
371+
since,
372+
limit,
373+
)
374+
.await
375+
}
376+
},
377+
RouteDecision::Writer => {
378+
crate::feed::query_mentions(
379+
&self.pool,
380+
community,
381+
pubkey_bytes,
382+
accessible_channel_ids,
383+
since,
384+
limit,
385+
)
386+
.await
387+
}
388+
}
389+
}
390+
391+
/// Find events that require action from the given pubkey.
392+
#[datastore_span(name = "query_feed_needs_action", system = "postgresql")]
393+
pub async fn query_feed_needs_action(
394+
&self,
395+
community: CommunityId,
396+
pubkey_bytes: &[u8],
397+
accessible_channel_ids: &[Uuid],
398+
since: Option<DateTime<Utc>>,
399+
limit: i64,
400+
) -> Result<Vec<StoredEvent>> {
401+
crate::feed::query_needs_action(
402+
&self.pool,
403+
community,
404+
pubkey_bytes,
405+
accessible_channel_ids,
406+
since,
407+
limit,
408+
)
409+
.await
410+
}
411+
412+
/// [`Db::query_feed_needs_action`] with replica routing — BOUNDED arm
413+
/// only; see [`Db::query_feed_mentions_routed`] for why the covered arm
414+
/// is structurally unavailable to feed queries.
415+
#[datastore_span(name = "query_feed_needs_action_routed", system = "postgresql")]
416+
pub async fn query_feed_needs_action_routed(
417+
&self,
418+
path: &'static str,
419+
community: CommunityId,
420+
pubkey_bytes: &[u8],
421+
accessible_channel_ids: &[Uuid],
422+
since: Option<DateTime<Utc>>,
423+
limit: i64,
424+
) -> Result<Vec<StoredEvent>> {
425+
match self.route_read(path, RoutePredicate::Bounded).await {
426+
RouteDecision::Replica(mut tx, _entry, reason) => {
427+
match crate::feed::query_needs_action_on(
428+
&mut tx,
429+
community,
430+
pubkey_bytes,
431+
accessible_channel_ids,
432+
since,
433+
limit,
434+
)
435+
.await
436+
{
437+
Ok(events) => {
438+
Self::record_route(path, "replica", reason);
439+
Ok(events)
440+
}
441+
Err(e) => {
442+
tracing::warn!(path, "replica read failed; re-running on writer: {e}");
443+
Self::record_route(path, "writer", "replica_error");
444+
crate::feed::query_needs_action(
445+
&self.pool,
446+
community,
447+
pubkey_bytes,
448+
accessible_channel_ids,
449+
since,
450+
limit,
451+
)
452+
.await
453+
}
454+
}
455+
}
456+
RouteDecision::Writer => {
457+
crate::feed::query_needs_action(
458+
&self.pool,
459+
community,
460+
pubkey_bytes,
461+
accessible_channel_ids,
462+
since,
463+
limit,
464+
)
465+
.await
466+
}
467+
}
468+
}
469+
470+
/// Find recent activity across accessible channels.
471+
#[datastore_span(name = "query_feed_activity", system = "postgresql")]
472+
pub async fn query_feed_activity(
473+
&self,
474+
community: CommunityId,
475+
accessible_channel_ids: &[Uuid],
476+
since: Option<DateTime<Utc>>,
477+
limit: i64,
478+
) -> Result<Vec<StoredEvent>> {
479+
crate::feed::query_activity(&self.pool, community, accessible_channel_ids, since, limit)
480+
.await
481+
}
482+
483+
/// [`Db::query_feed_activity`] with replica routing — BOUNDED arm only;
484+
/// see [`Db::query_feed_mentions_routed`] for why the covered arm is
485+
/// structurally unavailable to feed queries.
486+
#[datastore_span(name = "query_feed_activity_routed", system = "postgresql")]
487+
pub async fn query_feed_activity_routed(
488+
&self,
489+
path: &'static str,
490+
community: CommunityId,
491+
accessible_channel_ids: &[Uuid],
492+
since: Option<DateTime<Utc>>,
493+
limit: i64,
494+
) -> Result<Vec<StoredEvent>> {
495+
match self.route_read(path, RoutePredicate::Bounded).await {
496+
RouteDecision::Replica(mut tx, _entry, reason) => match crate::feed::query_activity_on(
497+
&mut tx,
498+
community,
499+
accessible_channel_ids,
500+
since,
501+
limit,
502+
)
503+
.await
504+
{
505+
Ok(events) => {
506+
Self::record_route(path, "replica", reason);
507+
Ok(events)
508+
}
509+
Err(e) => {
510+
tracing::warn!(path, "replica read failed; re-running on writer: {e}");
511+
Self::record_route(path, "writer", "replica_error");
512+
crate::feed::query_activity(
513+
&self.pool,
514+
community,
515+
accessible_channel_ids,
516+
since,
517+
limit,
518+
)
519+
.await
520+
}
521+
},
522+
RouteDecision::Writer => {
523+
crate::feed::query_activity(
524+
&self.pool,
525+
community,
526+
accessible_channel_ids,
527+
since,
528+
limit,
529+
)
530+
.await
531+
}
532+
}
533+
}
534+
}
535+
306536
// -- Tests --------------------------------------------------------------------
307537

308538
#[cfg(test)]

0 commit comments

Comments
 (0)