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

prototype: sidebar reports unread presence, not counts

Browse Source 
Booleans for ordinary unread and attention, existence probes instead of
counts to a cap, mentions from event_mentions, threads pruned by root
last_reply_received_at. Benchmark scaffold included; not for merge.
This commit is contained in:
hART
2026-10-06 21:13:28 -07:00
parent c0bf08ff18
commit 4bd9c1e0d2
15 changed files with 416 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
crates/buzz-db/src/store/event.rs
View File

@@ -1696,20 +1696,21 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
.execute(&mut **tx)
.await?;
if let Some(root_id) = meta.root_event_id {
sqlx::query(
r#"
UPDATE thread_metadata
SET descendant_count = descendant_count + 1
WHERE community_id = $1 AND event_id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(root_id)
.execute(&mut **tx)
.await?;
}
let root = meta.root_event_id.unwrap_or(pid);
sqlx::query(
r#"
UPDATE thread_metadata
SET descendant_count = descendant_count + $3::int,
last_reply_received_at = GREATEST(last_reply_received_at, $4)
WHERE community_id = $1 AND event_id = $2
"#,
)
.bind(community_id.as_uuid())
.bind(root)
.bind(i32::from(meta.root_event_id.is_some()))
.bind(received_at)
.execute(&mut **tx)
.await?;
crate::personal_read::record_reply(
tx,
community_id,

26
crates/buzz-db/src/store/personal_read/arrival_postgres_tests.rs
View File

@@ -154,7 +154,7 @@ async fn late_arrival_with_old_author_time_is_unread() {
states(&db, community, &actor, channel, &[&read, &late]).await,
["read", "unread"]
);
assert_eq!(sidebar(&db, community, &actor).await.unread, 1);
assert!(sidebar(&db, community, &actor).await.unread);
}
#[tokio::test]
@@ -179,7 +179,7 @@ async fn future_dated_anchor_does_not_swallow_later_arrivals() {
states(&db, community, &actor, channel, &[&earlier, &ahead, &later]).await,
["read", "read", "unread"]
);
assert_eq!(sidebar(&db, community, &actor).await.unread, 1);
assert!(sidebar(&db, community, &actor).await.unread);
}
/// Marking the sidebar's own latest message must clear the badge even when the
@@ -192,7 +192,7 @@ async fn mark_as_read_with_the_sidebar_anchor_clears_a_late_arrival() {
arrive(&pool, community, &first, now - 60).await;
post(&db, &pool, community, channel, now + 1, now - 40).await;
post(&db, &pool, community, channel, now - 600, now - 30).await;
assert_eq!(sidebar(&db, community, &actor).await.unread, 3);
assert!(sidebar(&db, community, &actor).await.unread);
let anchor = sidebar(&db, community, &actor)
.await
@@ -200,7 +200,7 @@ async fn mark_as_read_with_the_sidebar_anchor_clears_a_late_arrival() {
.expect("a channel with messages has a latest message");
apply(&db, community, &actor, mark_channel_read(channel, anchor)).await;
assert_eq!(sidebar(&db, community, &actor).await.unread, 0);
assert!(!sidebar(&db, community, &actor).await.unread);
}
#[tokio::test]
@@ -248,7 +248,7 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() {
.await
.unwrap();
let row = sidebar(&db, community, &actor).await;
assert_eq!((row.unread, row.unread_thread_count), (2, 1));
assert_eq!((row.unread, row.attention, row.threads.len()), (false, true, 1));
assert_eq!(row.threads.len(), 1);
let root_id = root.id.to_hex();
@@ -262,7 +262,7 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() {
.await;
let row = sidebar(&db, community, &actor).await;
assert_eq!((row.unread, row.unread_thread_count), (0, 0));
assert_eq!((row.unread, row.attention, row.threads.len()), (false, false, 0));
assert!(row.threads.is_empty());
}
@@ -283,11 +283,11 @@ async fn mark_as_read_clears_a_late_arrival_behind_the_latest_probe() {
.latest_message_id
.unwrap();
apply(&db, community, &actor, mark_channel_read(channel, anchor)).await;
assert_eq!(sidebar(&db, community, &actor).await.unread, 0);
assert!(!sidebar(&db, community, &actor).await.unread);
let late = post(&db, &pool, community, channel, now - 600, now - 10).await;
let row = sidebar(&db, community, &actor).await;
assert_eq!(row.unread, 1);
assert!(row.unread);
assert_eq!(row.latest_message_id, Some(late.id.to_hex()));
apply(
&db,
@@ -296,7 +296,7 @@ async fn mark_as_read_clears_a_late_arrival_behind_the_latest_probe() {
mark_channel_read(channel, late.id.to_hex()),
)
.await;
assert_eq!(sidebar(&db, community, &actor).await.unread, 0);
assert!(!sidebar(&db, community, &actor).await.unread);
}
/// Reactions never fill the latest probe.
@@ -321,7 +321,7 @@ async fn latest_message_behind_newer_reactions_is_found() {
let row = sidebar(&db, community, &actor).await;
assert_eq!(row.latest_message_id, Some(only.id.to_hex()));
assert_eq!(row.latest_message_at, Some((now - 600) as i64));
assert_eq!(row.unread, 1);
assert!(row.unread);
}
/// Store `event` as having arrived at exactly `seconds` plus `micros`. Built
@@ -355,7 +355,7 @@ async fn arrive_exact(
/// Two messages with the same author time arriving within one second; marks
/// the one at `pick` and returns both channel states.
async fn mark_within_one_second(micros: [u32; 2], pick: usize) -> (Vec<String>, u32) {
async fn mark_within_one_second(micros: [u32; 2], pick: usize) -> (Vec<String>, bool) {
let (db, pool, community, channel, actor, first) = fixture().await;
let now = first.created_at.as_secs();
let second = post(&db, &pool, community, channel, now, now).await;
@@ -383,7 +383,7 @@ async fn mark_within_one_second(micros: [u32; 2], pick: usize) -> (Vec<String>,
async fn arrivals_one_microsecond_apart_in_the_same_second_are_ordered() {
assert_eq!(
mark_within_one_second([500_000, 500_001], 0).await,
(vec!["read".to_owned(), "unread".to_owned()], 1)
(vec!["read".to_owned(), "unread".to_owned()], true)
);
}
@@ -395,7 +395,7 @@ async fn marking_either_of_two_identical_arrivals_reads_both() {
for pick in [0, 1] {
assert_eq!(
mark_within_one_second([500_000, 500_000], pick).await,
(vec!["read".to_owned(), "read".to_owned()], 0),
(vec!["read".to_owned(), "read".to_owned()], false),
"marked index {pick}"
);
}

199
crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs Normal file
View File

@@ -0,0 +1,199 @@
//! Side-by-side benchmark (not for merge). Identical file in both branches.
//! BENCH_THREADS (default 5000) followed threads, 3 replies each.
use super::postgres_tests::{apply, channel_read, mention};
use super::*;
use buzz_core::CommunityId;
use crate::{
channel::{ChannelType, ChannelVisibility},
Db,
};
use nostr::{EventBuilder, Keys, Kind, Tag};
use sqlx::postgres::PgPoolOptions;
use std::time::{Duration, Instant};
use uuid::Uuid;
fn ts() -> u64 {
nostr::Timestamp::now().as_secs()
}
fn ev(author: &Keys, at: u64, tags: Vec<Tag>) -> nostr::Event {
EventBuilder::new(Kind::Custom(9), format!("bench {at} {}", Uuid::new_v4()))
.tags(tags)
.custom_created_at(nostr::Timestamp::from(at))
.sign_with_keys(author)
.unwrap()
}
async fn top(db: &Db, c: CommunityId, ch: Uuid, e: &nostr::Event) -> Duration {
let t = Instant::now();
db.insert_event(c, e, Some(ch)).await.unwrap();
t.elapsed()
}
async fn rep(db: &Db, c: CommunityId, ch: Uuid, root: &nostr::Event, e: &nostr::Event) -> Duration {
let stamp = |e: &nostr::Event| chrono::DateTime::from_timestamp(e.created_at.as_secs() as i64, 0).unwrap();
let t = Instant::now();
db.insert_event_with_thread_metadata(
c,
e,
Some(ch),
Some(crate::event::ThreadMetadataParams {
event_id: e.id.as_bytes(),
event_created_at: stamp(e),
channel_id: ch,
parent_event_id: Some(root.id.as_bytes()),
parent_event_created_at: Some(stamp(root)),
root_event_id: Some(root.id.as_bytes()),
root_event_created_at: Some(stamp(root)),
depth: 1,
broadcast: false,
}),
)
.await
.unwrap();
t.elapsed()
}
fn stats(label: &str, mut v: Vec<Duration>) {
v.sort();
let p = |q: f64| v[((v.len() as f64 - 1.0) * q) as usize].as_secs_f64() * 1000.0;
println!(
"BENCH {label}: n={} p50={:.3}ms p95={:.3}ms p99={:.3}ms max={:.3}ms",
v.len(),
p(0.5),
p(0.95),
p(0.99),
p(1.0)
);
}
async fn sidebar_bench(db: &Db, c: CommunityId, actor: &Keys, label: &str) {
let mut v = Vec::new();
let mut last = None;
for _ in 0..30 {
let t = Instant::now();
let page = db.personal_read_sidebar(c, &actor.public_key(), 20, None).await.unwrap();
v.push(t.elapsed());
last = Some(page);
}
let page = last.unwrap();
let busy = &page.channels.iter().max_by_key(|r| r.unread).unwrap();
println!(
"BENCH {label}: busiest unread={} attention={} threads={} channels={}",
busy.unread,
busy.attention,
busy.threads.len(),
page.channels.len()
);
stats(&format!("sidebar {label}"), v);
}
#[tokio::test]
#[ignore = "benchmark"]
async fn bench_unread() {
let threads: usize = std::env::var("BENCH_THREADS").ok().and_then(|s| s.parse().ok()).unwrap_or(5000);
let pool = PgPoolOptions::new()
.max_connections(40)
.connect(&crate::test_support::database_url())
.await
.unwrap();
let db = Db::from_pool(pool.clone());
let c = db
.ensure_configured_community(&format!("bench-{}.local", Uuid::new_v4()))
.await
.unwrap()
.id;
let actor = Keys::generate();
let mut channels = Vec::new();
for i in 0..20 {
channels.push(
db.create_channel(c, &format!("bench {i}"), ChannelType::Stream, ChannelVisibility::Open, None, &actor.public_key().to_bytes(), None)
.await
.unwrap()
.id,
);
}
let busy = channels[0];
let others: Vec<Keys> = (0..8).map(|_| Keys::generate()).collect();
let base = ts() - 3600;
// A first message in every channel, and the first intent starts the actor.
let mut firsts = Vec::new();
for ch in &channels {
let e = ev(&others[0], base, vec![]);
top(&db, c, *ch, &e).await;
firsts.push(e);
}
assert_eq!(apply(&db, c, &actor, channel_read(busy, &firsts[0])).await, IntentOutcome::Applied);
// Phase 1: followed threads, serial ingest.
let (mut roots_t, mut reps_t) = (Vec::new(), Vec::new());
let mut roots = Vec::new();
let mut latest = firsts[0].clone();
for i in 0..threads {
let at = base + 1 + (i as u64 % 3000);
let root = ev(&others[i % 8], at, vec![]);
roots_t.push(top(&db, c, busy, &root).await);
let mine = ev(&actor, at, vec![]);
reps_t.push(rep(&db, c, busy, &root, &mine).await);
for j in 1..3 {
let r = ev(&others[(i + j) % 8], at, vec![]);
reps_t.push(rep(&db, c, busy, &root, &r).await);
latest = r;
}
roots.push(root);
}
stats("ingest top-level serial", roots_t);
stats("ingest reply serial", reps_t);
assert_eq!(apply(&db, c, &actor, channel_read(busy, &latest)).await, IntentOutcome::Applied);
for (ch, e) in channels.iter().zip(&firsts).skip(1) {
apply(&db, c, &actor, channel_read(*ch, e)).await;
}
sqlx::query("ANALYZE").execute(&pool).await.unwrap();
sidebar_bench(&db, c, &actor, &format!("caught-up {threads} threads")).await;
// Phase 2: a 20k timeline backlog with 12 mentions.
let now = ts();
let mut t2 = Vec::new();
for i in 0..20_000u64 {
let tags = if i % 1667 == 0 { mention(&actor) } else { vec![] };
let e = ev(&others[(i % 8) as usize], now - 600 + i / 40, tags);
t2.push(top(&db, c, busy, &e).await);
}
stats("ingest top-level serial (busy)", t2);
sqlx::query("ANALYZE").execute(&pool).await.unwrap();
sidebar_bench(&db, c, &actor, "20k timeline backlog").await;
// Phase 3: 500 followed threads get a new reply.
for root in roots.iter().take(500) {
let r = ev(&others[3], ts(), vec![]);
rep(&db, c, busy, root, &r).await;
}
sidebar_bench(&db, c, &actor, "20k backlog + 500 unread threads").await;
// Concurrency: 16 writers into one channel, then into 16 channels.
for (label, spread) in [("same channel", false), ("16 channels", true)] {
let started = Instant::now();
let mut handles = Vec::new();
for w in 0..16usize {
let db = db.clone();
let author = others[w % 8].clone();
let ch = if spread { channels[w + 1] } else { busy };
handles.push(tokio::spawn(async move {
let mut v = Vec::new();
for i in 0..300u64 {
let e = ev(&author, ts() - (i % 30), vec![]);
v.push(top(&db, c, ch, &e).await);
}
v
}));
}
let mut all = Vec::new();
for h in handles {
all.extend(h.await.unwrap());
}
let n = all.len();
let secs = started.elapsed().as_secs_f64();
println!("BENCH concurrent {label}: {n} inserts in {secs:.2}s = {:.0}/s", n as f64 / secs);
stats(&format!("ingest concurrent {label}"), all);
}
}

10
crates/buzz-db/src/store/personal_read/classification.rs
View File

@@ -3,7 +3,13 @@
use super::model::Reason;
/// Why a message is directed, before conversation membership is known.
pub(super) fn reason(channel_type: &str, actor_hex: &str, tags: &[Vec<String>]) -> Option<Reason> {
/// `broadcast_reply` is a depth-1 reply broadcast to the channel timeline.
pub(super) fn reason(
channel_type: &str,
actor_hex: &str,
tags: &[Vec<String>],
broadcast_reply: bool,
) -> Option<Reason> {
let tagged = |name: &str, matches: &dyn Fn(&str) -> bool| {
tags.iter()
.any(|tag| tag.len() >= 2 && tag[0] == name && matches(&tag[1]))
@@ -12,7 +18,7 @@ pub(super) fn reason(channel_type: &str, actor_hex: &str, tags: &[Vec<String>])
Some(Reason::Direct)
} else if tagged("p", &|value| value.eq_ignore_ascii_case(actor_hex)) {
Some(Reason::Mention)
} else if tagged("broadcast", &|value| value == "1") {
} else if broadcast_reply {
Some(Reason::Broadcast)
} else {
None

7
crates/buzz-db/src/store/personal_read/context.rs
View File

@@ -108,9 +108,9 @@ impl Db {
let thread = row
.try_get::<Option<Vec<u8>>, _>("root_event_id")?
.filter(|r| r != &message_id);
let on_timeline = thread.is_none()
|| (row.try_get::<Option<i32>, _>("depth")? == Some(1)
&& row.try_get::<Option<bool>, _>("broadcast")? == Some(true));
let broadcast_reply = row.try_get::<Option<i32>, _>("depth")? == Some(1)
&& row.try_get::<Option<bool>, _>("broadcast")? == Some(true);
let on_timeline = thread.is_none() || broadcast_reply;
// A thread context holds its replies, not its root; a
// broadcast reply belongs to both of its contexts.
let in_context = if root.is_empty() {
@@ -140,6 +140,7 @@ impl Db {
&row.try_get::<String, _>("channel_type")?,
&actor_hex,
&tags,
broadcast_reply,
);
match position {
None => MessageReadState::NotCounted,

3
crates/buzz-db/src/store/personal_read/mod.rs
View File

@@ -25,3 +25,6 @@ mod threads_postgres_tests;
#[cfg(test)]
mod arrival_postgres_tests;
#[cfg(test)]
mod bench_postgres_tests;

27
crates/buzz-db/src/store/personal_read/model.rs
View File

@@ -51,8 +51,6 @@ pub enum IntentOutcome {
pub const MAX_CHANNELS: usize = 20;
/// Maximum unread-thread summaries per channel row.
pub const MAX_THREAD_SUMMARIES: usize = 5;
/// Counts stop here: a count equal to the cap means at least the cap ("99+").
pub const UNREAD_CAP: u32 = 99;
/// Ingest rejects author times further than this from relay time, so anything
/// that arrived after a position has an author time no earlier than this
/// before it. Forward scans range over author time and filter on arrival.
@@ -73,18 +71,15 @@ pub struct ChannelReadSummary {
pub archived: bool,
/// Existing DM visibility preference (not an authorization decision).
pub hidden: bool,
/// Unread timeline messages plus unread replies in the actor's threads,
/// clamped at [`UNREAD_CAP`].
pub unread: u32,
/// The unread subset with a [`Reason`]: everything but ordinary timeline
/// messages. Clamped at [`UNREAD_CAP`].
pub attention: u32,
/// The actor's threads with at least one unread reply, clamped at
/// [`UNREAD_CAP`].
pub unread_thread_count: u32,
/// An unread timeline message without a [`Reason`] exists: ordinary
/// backlog. Never true in a DM, where every message is direct.
pub unread: bool,
/// An unread message with a [`Reason`] exists: a direct, mention or
/// broadcast timeline message, or a reply in one of the actor's threads.
pub attention: bool,
/// Last eligible nondeleted event to arrive among the channel's newest
/// and everything counted, whatever its author: `mark_channel_read`
/// through it reads the row.
/// and the unread messages found, whatever its author:
/// `mark_channel_read` through it reads the row.
pub latest_message_id: Option<String>,
/// Display activity: the greatest author time (Unix seconds) among the
/// channel's newest, not necessarily latest_message_id's own. None
@@ -94,13 +89,11 @@ pub struct ChannelReadSummary {
pub threads: Vec<ThreadReadSummary>,
}
/// Unread replies in one of the actor's threads. No conversation bytes.
#[derive(Debug, Serialize)]
/// One of the actor's threads with an unread reply. No conversation bytes.
#[derive(Debug, Deserialize, Serialize)]
pub struct ThreadReadSummary {
/// Canonical thread-root event ID.
pub root_id: String,
/// Unread replies in this thread, clamped at [`UNREAD_CAP`].
pub unread: u32,
/// Last unread reply to arrive: marking through it reads the thread.
pub latest_reply_id: String,
/// Greatest author time (Unix seconds) among unread replies, not

52
crates/buzz-db/src/store/personal_read/postgres_tests.rs
View File

@@ -236,24 +236,19 @@ async fn accounts(pool: &PgPool, community: CommunityId) -> i64 {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn counts_stop_at_the_advertised_cap_and_serialize_as_integers() {
async fn unread_is_presence_and_serializes_as_booleans() {
let (db, _, community, channel, actor, _) = fixture().await;
let author = Keys::generate();
let base = now();
let mut last = None;
for i in 0..u64::from(UNREAD_CAP) + 20 {
for i in 0..120 {
last = Some(post(&db, community, channel, &author, base + i, vec![]).await);
}
let row = sidebar(&db, community, &actor).await;
assert_eq!(
row.unread, UNREAD_CAP,
"a count at the cap means at least the cap"
);
let wire = serde_json::to_value(&row).unwrap();
assert_eq!(wire["unread"], 99);
assert_eq!(wire["attention"], 0);
assert_eq!(wire["unread_thread_count"], 0);
for gone in ["latest_message_complete", "account", "status"] {
assert_eq!(wire["unread"], true);
assert_eq!(wire["attention"], false);
for gone in ["latest_message_complete", "account", "status", "unread_thread_count"] {
assert!(wire.get(gone).is_none(), "{gone}");
}
let last = last.unwrap();
@@ -262,14 +257,14 @@ async fn counts_stop_at_the_advertised_cap_and_serialize_as_integers() {
apply(&db, community, &actor, mark(channel, None, &last)).await,
IntentOutcome::Applied
);
assert_eq!(sidebar(&db, community, &actor).await.unread, 0);
assert!(!sidebar(&db, community, &actor).await.unread);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn joining_starts_caught_up_and_counts_only_what_arrives_after() {
let (db, pool, community, channel, owner, event) = fixture().await;
assert_eq!(sidebar(&db, community, &owner).await.unread, 1);
assert!(sidebar(&db, community, &owner).await.unread);
let joiner = Keys::generate();
start(&pool, community, &joiner).await;
let before = accounts(&pool, community).await;
@@ -285,7 +280,7 @@ async fn joining_starts_caught_up_and_counts_only_what_arrives_after() {
let row = sidebar(&db, community, &joiner).await;
assert_eq!(
(row.unread, row.attention),
(0, 0),
(false, false),
"history before joining is read"
);
assert_eq!(row.latest_message_id, Some(event.id.to_hex()));
@@ -308,7 +303,7 @@ async fn joining_starts_caught_up_and_counts_only_what_arrives_after() {
)
.await;
let row = sidebar(&db, community, &joiner).await;
assert_eq!((row.unread, row.attention), (1, 1));
assert_eq!((row.unread, row.attention), (false, true), "a mention is attention, not backlog");
assert_eq!(
states(&db, community, &joiner, channel, None, &[&later]).await,
[serde_json::json!({"status":"unread","reason":"mention"})]
@@ -350,18 +345,18 @@ async fn rejoining_starts_caught_up_again() {
)
.await
.unwrap();
assert_eq!(sidebar(&db, community, &owner).await.unread, 2);
assert!(sidebar(&db, community, &owner).await.unread);
db.add_member(community, channel, &pk, MemberRole::Member, None)
.await
.unwrap();
let row = sidebar(&db, community, &member).await;
assert_eq!(row.unread, 0, "messages from while away are read");
assert!(!row.unread, "messages from while away are read");
assert_eq!(
status(&db, community, &member, channel, None, &[&away]).await,
["read"]
);
post(&db, community, channel, &Keys::generate(), now(), vec![]).await;
assert_eq!(sidebar(&db, community, &member).await.unread, 1);
assert!(sidebar(&db, community, &member).await.unread);
}
#[tokio::test]
@@ -408,8 +403,8 @@ async fn read_state_starts_caught_up_at_the_first_intent() {
.await;
let row = sidebar(&db, community, &member).await;
assert_eq!(
(row.unread, row.attention, row.unread_thread_count),
(0, 0, 0)
(row.unread, row.attention, row.threads.len()),
(false, false, 0)
);
assert_eq!(row.latest_message_id, Some(before.id.to_hex()));
assert_eq!(
@@ -423,7 +418,7 @@ async fn read_state_starts_caught_up_at_the_first_intent() {
IntentOutcome::Applied
);
let row = sidebar(&db, community, &member).await;
assert_eq!((row.unread, row.unread_thread_count), (0, 0));
assert_eq!((row.unread, row.threads.len()), (false, 0));
assert_eq!(
status(&db, community, &member, channel, None, &[&before]).await,
["read"]
@@ -441,7 +436,7 @@ async fn read_state_starts_caught_up_at_the_first_intent() {
.await;
post(&db, community, channel, &Keys::generate(), now(), vec![]).await;
let row = sidebar(&db, community, &member).await;
assert_eq!((row.unread, row.unread_thread_count), (2, 1));
assert_eq!((row.unread, row.attention, row.threads.len()), (true, true, 1));
}
#[tokio::test]
@@ -461,7 +456,7 @@ async fn sidebar_is_read_only_and_does_not_wait_for_account() {
.fetch_all(&mut *held)
.await
.unwrap();
assert_eq!(sidebar(&db, community, &actor).await.unread, 0);
assert!(!sidebar(&db, community, &actor).await.unread);
held.rollback().await.unwrap();
}
@@ -539,7 +534,7 @@ async fn diff_alone_leaves_the_sidebar_row_unchanged() {
)
.await;
let after = serde_json::to_value(sidebar(&db, community, &actor).await).unwrap();
assert_eq!(before["unread"], 1);
assert_eq!(before["unread"], true);
assert_eq!(before, after, "not unread, not attention, not latest");
}
@@ -560,7 +555,7 @@ async fn latest_includes_own_and_excludes_deleted_and_auxiliary() {
let row = sidebar(&db, community, &actor).await;
assert_eq!(row.latest_message_id, Some(own.id.to_hex()));
assert_eq!(row.latest_message_at, Some((base + 1) as i64));
assert_eq!(row.unread, 1, "own, deleted and reaction do not count");
assert!(row.unread, "own, deleted and reaction do not count");
// An empty channel has no latest message.
db.create_channel(
community,
@@ -621,16 +616,15 @@ async fn position_is_monotonic_and_rejects_malformed_anchors() {
vec![],
)
.await;
assert_eq!(sidebar(&db, community, &actor).await.unread, 1);
assert!(sidebar(&db, community, &actor).await.unread);
for anchor in [&later, &event] {
assert_eq!(
apply(&db, community, &actor, mark(channel, None, anchor)).await,
IntentOutcome::Applied
);
}
assert_eq!(
sidebar(&db, community, &actor).await.unread,
0,
assert!(
!sidebar(&db, community, &actor).await.unread,
"never moves back"
);
let at_later: bool = sqlx::query_scalar(
@@ -684,7 +678,7 @@ async fn contexts_bound_selectors_and_share_one_state_per_message() {
status(&db, community, &actor, channel, Some(&root), &both).await,
["unavailable", "unread", "not_counted"]
);
assert_eq!(sidebar(&db, community, &actor).await.unread, 2);
assert!(sidebar(&db, community, &actor).await.unread);
assert_eq!(
accounts(&pool, community).await,
1,

163
crates/buzz-db/src/store/personal_read/projection.rs
View File

@@ -1,10 +1,9 @@
//! Forward counting from read positions. A caught-up reader examines almost
//! nothing: each count reads only what arrived after its position, and stops
//! at [`UNREAD_CAP`].
//! Unread presence from read positions. Every fact is an existence probe
//! that stops at the first qualifying row, so a caught-up reader examines
//! almost nothing and nothing is counted.
use super::{model::*, writes};
use buzz_core::CommunityId;
use serde::Deserialize;
use sqlx::{Acquire, Row};
use uuid::Uuid;
@@ -16,9 +15,13 @@ use crate::{observability, Db, DbError, Result};
const ON_TIMELINE: &str = "(tm.root_event_id IS NULL OR tm.root_event_id=e.id
OR (tm.depth=1 AND tm.broadcast))";
/// Whether event `e` is directed at the actor (hex `$hex`) by its tags.
const DIRECTED_TAGS: &str = "EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) t(tag)
WHERE (tag->>0='p' AND lower(tag->>1)=$hex) OR (tag->>0='broadcast' AND tag->>1='1'))";
/// Event `e` is eligible and unread against `r.position`, and `tm` is its
/// thread row. Author time ranges from the position less the arrival skew.
const UNREAD_EVENT: &str = "e.created_at >= r.position-$8 AND e.received_at > r.position
AND e.kind=ANY($6) AND e.pubkey<>$2 AND e.deleted_at IS NULL";
const EVENT_THREAD_ROW: &str = "LEFT JOIN thread_metadata tm ON tm.community_id=$1
AND tm.event_created_at=e.created_at AND tm.event_id=e.id";
impl Db {
/// Read a bounded joined roster from the writer in one read-only snapshot.
@@ -75,13 +78,18 @@ impl Db {
// the actor's first intent, so nothing counts). A channel's is its
// frontier, never earlier than joining: a new member starts caught
// up. A thread row exists only for the actor's threads (see
// `membership`), and a whole-channel cut covers them all. Each count
// ranges over author time from the position less the arrival skew
// (served by idx_events_community_channel_created or
// idx_thread_metadata_window), filters on arrival, and stops at the
// cap: rows the filters reject are read but bound nothing. The latest message is the last arrival among
// the newest messages and everything counted, so marking through it
// reads every counted message whatever its author time.
// `membership`), and a whole-channel cut covers them all.
//
// Each fact is the first qualifying row of one index range:
// - ordinary: the channel's timeline after the position, skipping
// directed messages (idx_events_community_channel_created);
// - mention: the actor's own mentions (idx_event_mentions_pubkey_created);
// - broadcast: the channel's depth-1 replies (idx_thread_metadata_channel_depth);
// - direct: any unread timeline message in a DM;
// - threads: roots in the channel whose last reply arrived after the
// actor's position, then their unread replies (anchor and display time).
// The latest message is the last arrival among the newest messages and
// every unread message found, so marking through it reads what was found.
let sql = format!(
r#"WITH roster AS MATERIALIZED (
SELECT c.id, c.name, c.channel_type::text AS channel_type,
@@ -100,11 +108,12 @@ impl Db {
ORDER BY c.id LIMIT $5
)
SELECT r.id, r.name, r.channel_type, r.archived, r.hidden,
(SELECT v.id FROM (VALUES (latest.arrival, latest.id),
(timeline.arrival, timeline.id), (threads.arrival, threads.id)) v(arrival, id)
(SELECT v.id FROM (VALUES (latest.arrival, latest.id), (ordinary.arrival, ordinary.id),
(directed.arrival, directed.id), (threads.arrival, threads.id)) v(arrival, id)
WHERE v.id IS NOT NULL ORDER BY v.arrival DESC, v.id LIMIT 1) AS latest_message_id,
latest.at AS latest_message_at,
COALESCE(timeline.unread, 0) AS unread, COALESCE(timeline.attention, 0) AS attention,
ordinary.id IS NOT NULL AS unread,
directed.id IS NOT NULL OR threads.id IS NOT NULL AS attention,
COALESCE(threads.items, '[]') AS threads
FROM roster r
LEFT JOIN LATERAL (
@@ -115,49 +124,68 @@ impl Db {
ORDER BY created_at DESC LIMIT $7) newest
) latest ON true
LEFT JOIN LATERAL (
SELECT count(*)::int AS unread, count(*) FILTER (WHERE directed)::int AS attention,
(array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id,
max(received_at) AS arrival
FROM (
SELECT e.id, e.received_at, r.channel_type='dm' OR {DIRECTED_TAGS} AS directed
FROM events e
LEFT JOIN thread_metadata tm ON tm.community_id=$1
AND tm.event_created_at=e.created_at AND tm.event_id=e.id
WHERE e.community_id=$1 AND e.channel_id=r.id
AND e.created_at >= r.position-$8 AND e.received_at > r.position
AND e.kind=ANY($6) AND e.pubkey<>$2 AND e.deleted_at IS NULL AND {ON_TIMELINE}
ORDER BY e.created_at LIMIT $9
) counted
) timeline ON true
SELECT encode(e.id,'hex') AS id, e.received_at AS arrival
FROM events e {EVENT_THREAD_ROW}
WHERE r.channel_type<>'dm' AND e.community_id=$1 AND e.channel_id=r.id
AND {UNREAD_EVENT} AND {ON_TIMELINE}
AND NOT (tm.depth=1 AND tm.broadcast) IS TRUE
AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) t(tag)
WHERE tag->>0='p' AND lower(tag->>1)=$hex)
ORDER BY e.created_at LIMIT 1
) ordinary ON true
LEFT JOIN LATERAL (
SELECT id, arrival FROM (
(SELECT encode(e.id,'hex') AS id, e.received_at AS arrival
FROM events e {EVENT_THREAD_ROW}
WHERE r.channel_type='dm' AND e.community_id=$1 AND e.channel_id=r.id
AND {UNREAD_EVENT} AND {ON_TIMELINE}
ORDER BY e.created_at LIMIT 1)
UNION ALL
(SELECT encode(e.id,'hex'), e.received_at
FROM event_mentions m JOIN events e ON e.community_id=$1
AND e.created_at=m.event_created_at AND e.id=m.event_id
{EVENT_THREAD_ROW}
WHERE r.channel_type<>'dm' AND m.community_id=$1 AND m.pubkey_hex=$hex
AND m.event_created_at >= r.position-$8 AND m.channel_id=r.id
AND e.channel_id=r.id AND {UNREAD_EVENT} AND {ON_TIMELINE}
LIMIT 1)
UNION ALL
(SELECT encode(e.id,'hex'), e.received_at
FROM thread_metadata tm JOIN events e ON e.community_id=$1
AND e.created_at=tm.event_created_at AND e.id=tm.event_id
WHERE r.channel_type<>'dm' AND tm.community_id=$1 AND tm.channel_id=r.id
AND tm.depth=1 AND tm.broadcast AND tm.event_created_at >= r.position-$8
AND {UNREAD_EVENT}
LIMIT 1)
) found ORDER BY arrival DESC LIMIT 1
) directed ON true
LEFT JOIN LATERAL (
SELECT jsonb_agg(jsonb_build_object('root_id',encode(tf.root_id,'hex'),
'unread',n.unread,'latest_reply_id',n.id,'latest_reply_at',n.at)) AS items,
'latest_reply_id',n.id,'latest_reply_at',n.at)) AS items,
(array_agg(n.id ORDER BY n.arrival DESC,n.id))[1] AS id, max(n.arrival) AS arrival
FROM personal_read_frontiers tf
JOIN thread_metadata root ON root.community_id=$1 AND root.channel_id=r.id
AND root.depth=0 AND root.event_id=tf.root_id
CROSS JOIN LATERAL (SELECT GREATEST(tf.through_timestamp, r.threads_floor) AS position) p
CROSS JOIN LATERAL (
SELECT count(*)::int AS unread,
(array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id,
max(extract(epoch FROM created_at)::bigint) AS at,
max(received_at) AS arrival
FROM (
SELECT e.id, e.created_at, e.received_at
FROM thread_metadata tm JOIN events e ON e.community_id=$1
AND e.created_at=tm.event_created_at AND e.id=tm.event_id
WHERE tm.community_id=$1 AND tm.root_event_id=tf.root_id
AND tm.channel_id=r.id AND tm.event_id<>tf.root_id
AND tm.event_created_at >= p.position-$8
AND e.received_at > p.position AND e.kind=ANY($6) AND e.pubkey<>$2
AND e.deleted_at IS NULL AND NOT (tm.depth=1 AND tm.broadcast)
ORDER BY tm.event_created_at LIMIT $9
) counted
SELECT (array_agg(encode(e.id,'hex') ORDER BY e.received_at DESC,e.id))[1] AS id,
max(extract(epoch FROM e.created_at)::bigint) AS at,
max(e.received_at) AS arrival
FROM thread_metadata tm JOIN events e ON e.community_id=$1
AND e.created_at=tm.event_created_at AND e.id=tm.event_id
WHERE tm.community_id=$1 AND tm.root_event_id=tf.root_id
AND tm.channel_id=r.id AND tm.event_id<>tf.root_id
AND tm.event_created_at >= p.position-$8
AND e.received_at > p.position AND e.kind=ANY($6) AND e.pubkey<>$2
AND e.deleted_at IS NULL AND NOT (tm.depth=1 AND tm.broadcast)
) n
WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=r.id
AND tf.root_id<>''::bytea AND n.unread > 0
AND tf.root_id<>''::bytea AND root.last_reply_received_at > p.position
AND n.id IS NOT NULL
) threads ON true
ORDER BY r.id"#
)
.replace("$hex", "$10");
.replace("$hex", "$9");
// Only compile-time constants are interpolated.
let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(community.as_uuid())
@@ -168,7 +196,6 @@ impl Db {
.bind(ELIGIBLE_KINDS.as_slice())
.bind(LATEST_PROBE as i64)
.bind(skew())
.bind(i64::from(UNREAD_CAP))
.bind(actor.to_hex())
.fetch_all(&mut *tx)
.await?;
@@ -176,23 +203,19 @@ impl Db {
let has_more = rows.len() > limit;
let mut channels = Vec::with_capacity(limit);
for row in rows.into_iter().take(limit) {
let threads: Vec<ThreadRow> = serde_json::from_value(row.try_get("threads")?)
let threads: Vec<ThreadReadSummary> = serde_json::from_value(row.try_get("threads")?)
.map_err(|_| DbError::InvalidData("invalid thread summaries".into()))?;
let thread_unread: u32 = threads.iter().map(|t| t.unread).sum();
let unread = row.try_get::<i32, _>("unread")? as u32;
let attention = row.try_get::<i32, _>("attention")? as u32;
channels.push(ChannelReadSummary {
channel_id: row.try_get("id")?,
name: row.try_get("name")?,
channel_type: row.try_get("channel_type")?,
archived: row.try_get("archived")?,
hidden: row.try_get("hidden")?,
unread: clamp(unread + thread_unread),
attention: clamp(attention + thread_unread),
unread_thread_count: clamp(threads.len() as u32),
unread: row.try_get("unread")?,
attention: row.try_get("attention")?,
latest_message_id: row.try_get("latest_message_id")?,
latest_message_at: row.try_get("latest_message_at")?,
threads: summarize(threads.into_iter().map(ThreadRow::into).collect()),
threads: summarize(threads),
});
}
let next_cursor = if has_more {
@@ -214,29 +237,6 @@ pub(super) fn skew() -> chrono::Duration {
chrono::Duration::seconds(i64::from(MAX_ARRIVAL_SKEW_SECONDS))
}
fn clamp(n: u32) -> u32 {
n.min(UNREAD_CAP)
}
#[derive(Deserialize)]
struct ThreadRow {
root_id: String,
unread: u32,
latest_reply_id: String,
latest_reply_at: i64,
}
impl From<ThreadRow> for ThreadReadSummary {
fn from(t: ThreadRow) -> Self {
Self {
root_id: t.root_id,
unread: clamp(t.unread),
latest_reply_id: t.latest_reply_id,
latest_reply_at: t.latest_reply_at,
}
}
}
/// Order newest unread reply first (root ID breaks ties) and cap the list.
pub(super) fn summarize(mut items: Vec<ThreadReadSummary>) -> Vec<ThreadReadSummary> {
items.sort_unstable_by(|a, b| {
@@ -255,7 +255,6 @@ mod tests {
fn item(root: u8, at: i64) -> ThreadReadSummary {
ThreadReadSummary {
root_id: hex::encode([root; 32]),
unread: 1,
latest_reply_id: hex::encode([root; 32]),
latest_reply_at: at,
}

32
crates/buzz-db/src/store/personal_read/projection_postgres_tests.rs
View File

@@ -26,7 +26,7 @@ async fn sidebar_sql_eligibility_matches_selector() {
.unwrap();
let expected = u32::from(ELIGIBLE_KINDS.contains(&kind) && !own && !deleted);
assert_eq!(
page.channels[0].unread, expected,
u32::from(page.channels[0].unread), expected,
"kind={kind} own={own} deleted={deleted}"
);
let contexts = db
@@ -47,7 +47,7 @@ async fn sidebar_sql_eligibility_matches_selector() {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn sidebar_directed_fact_matches_selector_classifier() {
let (db, pool, community, channel, actor, _) = fixture().await;
let (db, pool, community, channel, actor, event) = fixture().await;
let actor_hex = actor.public_key().to_hex();
let fullwidth: String = actor_hex
.chars()
@@ -79,14 +79,36 @@ async fn sidebar_directed_fact_matches_selector_classifier() {
.await
.unwrap();
for tags in &cases {
let expected =
u32::from(classification::reason(channel_type, &actor_hex, tags).is_some());
// A top-level message: `broadcast` tags alone direct nothing.
let reason = classification::reason(channel_type, &actor_hex, tags, false);
let expected = (channel_type != "dm" && reason.is_none(), reason.is_some());
sqlx::query("UPDATE events SET tags=$2 WHERE community_id=$1")
.bind(community.as_uuid())
.bind(json!(tags))
.execute(&pool)
.await
.unwrap();
// Mirror ingest's mention index (`runtime::insert_mentions`).
sqlx::query("DELETE FROM event_mentions WHERE community_id=$1")
.bind(community.as_uuid())
.execute(&pool)
.await
.unwrap();
let mentioned = tags.iter().any(|t| {
t.len() >= 2 && t[0] == "p" && t[1].eq_ignore_ascii_case(&actor_hex)
});
if mentioned {
sqlx::query(
"INSERT INTO event_mentions (community_id,pubkey_hex,event_id,event_created_at,channel_id,event_kind)
SELECT community_id,$2,id,created_at,channel_id,kind FROM events WHERE community_id=$1 AND id=$3",
)
.bind(community.as_uuid())
.bind(&actor_hex)
.bind(event.id.as_bytes().as_slice())
.execute(&pool)
.await
.unwrap();
}
let row = db
.personal_read_sidebar(community, &actor.public_key(), 20, None)
.await
@@ -95,7 +117,7 @@ async fn sidebar_directed_fact_matches_selector_classifier() {
.remove(0);
assert_eq!(
(row.unread, row.attention),
(1, expected),
expected,
"channel_type={channel_type} tags={tags:?}"
);
}

59
crates/buzz-db/src/store/personal_read/threads_postgres_tests.rs
View File

@@ -65,9 +65,10 @@ async fn replying_mentioning_and_starting_a_thread_make_members() {
assert_eq!(thread_rows(&pool, community, &starter).await, root_row);
assert!(thread_rows(&pool, community, &bystander).await.is_empty());
let row = sidebar(&db, community, &starter).await;
// Only the reply is unread: attention, but no ordinary timeline backlog.
assert_eq!(
(row.unread, row.attention, row.unread_thread_count),
(1, 1, 1)
(row.unread, row.attention, row.threads.len()),
(false, true, 1)
);
assert_eq!(row.threads[0].latest_reply_id, own.id.to_hex());
assert_eq!(
@@ -75,7 +76,7 @@ async fn replying_mentioning_and_starting_a_thread_make_members() {
[serde_json::json!({"status":"unread","reason":"conversation"})]
);
let row = sidebar(&db, community, &actor).await;
assert_eq!((row.unread_thread_count, row.threads.len()), (0, 0));
assert!(row.threads.is_empty());
// Someone else replies: the actor, now a member, has it unread, beside
// the fixture's message and the root on the timeline.
@@ -92,10 +93,9 @@ async fn replying_mentioning_and_starting_a_thread_make_members() {
.await;
let row = sidebar(&db, community, &actor).await;
assert_eq!(
(row.unread, row.attention, row.unread_thread_count),
(3, 1, 1)
(row.unread, row.attention, row.threads.len()),
(true, true, 1)
);
assert_eq!(row.threads[0].unread, 1);
assert_eq!(row.threads[0].latest_reply_id, other.id.to_hex());
// A mention makes a channel member a thread member; a non-member gets nothing.
@@ -118,8 +118,8 @@ async fn replying_mentioning_and_starting_a_thread_make_members() {
let row = sidebar(&db, community, &bystander).await;
// The mention, and the root on the timeline.
assert_eq!(
(row.unread, row.attention, row.unread_thread_count),
(2, 1, 1)
(row.unread, row.attention, row.threads.len()),
(true, true, 1)
);
assert_eq!(
states(
@@ -200,13 +200,13 @@ async fn reading_a_thread_moves_a_members_row_and_never_joins() {
false,
)
.await;
assert_eq!(sidebar(&db, community, &actor).await.threads[0].unread, 1);
assert_eq!(sidebar(&db, community, &actor).await.threads.len(), 1);
assert_eq!(
apply(&db, community, &actor, mark(channel, Some(&root), &root)).await,
IntentOutcome::Applied
);
assert_eq!(
sidebar(&db, community, &actor).await.unread_thread_count,
sidebar(&db, community, &actor).await.threads.len(),
1,
"never backwards"
);
@@ -216,8 +216,8 @@ async fn reading_a_thread_moves_a_members_row_and_never_joins() {
);
let row = sidebar(&db, community, &actor).await;
assert_eq!(
(row.unread, row.unread_thread_count),
(1, 0),
(row.unread, row.threads.len()),
(true, 0),
"the root itself is on the timeline"
);
}
@@ -263,8 +263,8 @@ async fn dm_replies_make_every_member_a_thread_member() {
.channels
.remove(0);
assert_eq!(
(row.unread, row.attention, row.unread_thread_count),
(2, 2, 1)
(row.unread, row.attention, row.threads.len()),
(false, true, 1)
);
}
@@ -316,7 +316,11 @@ async fn mark_channel_read_covers_every_thread_through_a_reply_anchor() {
IntentOutcome::Applied
);
let row = sidebar(&db, community, &actor).await;
assert_eq!(row.unread, 2, "top at +20 and reply at +30 remain");
assert_eq!(
(row.unread, row.attention),
(true, true),
"top at +20 and reply at +30 remain"
);
assert_eq!(row.threads.len(), 1);
assert_eq!(row.threads[0].latest_reply_id, second.id.to_hex());
@@ -326,8 +330,8 @@ async fn mark_channel_read_covers_every_thread_through_a_reply_anchor() {
);
let row = sidebar(&db, community, &actor).await;
assert_eq!(
(row.unread, row.unread_thread_count),
(0, 0),
(row.unread, row.attention, row.threads.len()),
(false, false, 0),
"every thread is covered"
);
@@ -355,7 +359,7 @@ async fn mark_channel_read_covers_every_thread_through_a_reply_anchor() {
)
.await;
let row = sidebar(&db, community, &actor).await;
assert_eq!(row.unread, 2);
assert_eq!((row.unread, row.attention), (false, true));
assert_eq!(row.threads[0].latest_reply_id, newer.id.to_hex());
assert_eq!(
status(
@@ -453,11 +457,7 @@ async fn thread_summaries_order_cap_and_count_every_unread_thread() {
}
let row = sidebar(&db, community, &actor).await;
// The fixture's message is one ordinary timeline unread.
assert_eq!((row.unread, row.attention), (9, 8));
assert_eq!(
row.unread_thread_count, 6,
"counts threads beyond the listed five"
);
assert_eq!((row.unread, row.attention, row.threads.len()), (true, true, 5));
let mut tied = [roots[0].id.to_hex(), roots[1].id.to_hex()];
tied.sort();
let order: Vec<_> = row.threads.iter().map(|t| t.root_id.clone()).collect();
@@ -471,7 +471,6 @@ async fn thread_summaries_order_cap_and_count_every_unread_thread() {
roots[4].id.to_hex()
]
);
assert_eq!(row.threads[2].unread, 3);
assert_eq!(row.threads[2].latest_reply_id, last.unwrap().id.to_hex());
assert_eq!(row.threads[2].latest_reply_at, (base + 140) as i64);
// The row's activity is the greatest author time among the newest messages.
@@ -491,12 +490,10 @@ async fn thread_summaries_order_cap_and_count_every_unread_thread() {
IntentOutcome::Applied
);
let row = sidebar(&db, community, &actor).await;
assert_eq!((row.unread_thread_count, row.threads.len()), (5, 5));
assert_eq!(row.threads.len(), 5);
assert!(row.threads.iter().all(|t| t.root_id != first.root_id));
// Plus the fixture's timeline message.
assert_eq!(
row.unread,
1 + row.threads.iter().map(|t| t.unread).sum::<u32>()
);
assert!(row.unread);
}
#[tokio::test]
@@ -539,7 +536,7 @@ async fn thread_on_a_never_unread_root_is_selectable_and_readable_by_itself() {
.await;
let row = sidebar(&db, community, &actor).await;
assert_eq!(row.unread, 3, "the root and two replies");
assert_eq!((row.unread, row.attention), (true, true), "the root and two replies");
assert_eq!(row.threads.len(), 2);
let listed = &row.threads[1];
assert_eq!(listed.root_id, diff.id.to_hex());
@@ -562,7 +559,7 @@ async fn thread_on_a_never_unread_root_is_selectable_and_readable_by_itself() {
IntentOutcome::Applied
);
let row = sidebar(&db, community, &actor).await;
assert_eq!(row.unread, 2);
assert_eq!((row.unread, row.attention), (true, true));
assert_eq!(row.threads.len(), 1);
assert_eq!(row.threads[0].latest_reply_id, elsewhere.id.to_hex());
}

2
crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs
View File

@@ -464,7 +464,7 @@ async fn accessory_discovery_is_host_bound_and_opt_in() {
let d = &doc["buzz_v1"];
assert_eq!(d["version"], 1);
assert_eq!(d["base_path"], "/buzz/v1");
assert_eq!(d["unread_cap"], 99);
assert!(d.get("unread_cap").is_none());
assert!(d.get("retention_seconds").is_none());
assert_eq!(d["max_channels"], buzz_db::personal_read::MAX_CHANNELS);
assert_eq!(d["max_intents"], buzz_db::personal_read::MAX_INTENTS);

5
crates/buzz-relay/src/nip11.rs
View File

@@ -88,8 +88,6 @@ pub struct BuzzV1Descriptor {
pub version: u32,
/// Relay-relative API prefix; callers retain the requesting origin.
pub base_path: String,
/// Unread counts stop here: a count equal to the cap means at least it.
pub unread_cap: u32,
/// Maximum joined channels per sidebar page.
pub max_channels: usize,
/// Maximum independent write intents per request.
@@ -391,12 +389,11 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st
if state.config.buzz_v1_enabled {
use buzz_db::personal_read::{
ELIGIBLE_KINDS, MAX_CHANNELS, MAX_CONTEXTS, MAX_CONTEXT_MESSAGES, MAX_INTENTS,
MAX_THREAD_SUMMARIES, UNREAD_CAP,
MAX_THREAD_SUMMARIES,
};
info.buzz_v1 = Some(BuzzV1Descriptor {
version: 1,
base_path: crate::api::buzz_v1::BASE_PATH.to_owned(),
unread_cap: UNREAD_CAP,
max_channels: MAX_CHANNELS,
max_intents: MAX_INTENTS,
max_contexts: MAX_CONTEXTS,

3
migrations/0056_personal_read_state.sql
View File

@@ -32,6 +32,9 @@ CREATE TABLE personal_read_frontiers (
);
-- Arrival time of a thread root's latest reply: the sidebar skips threads
-- with nothing new since the reader's position without reading replies.
ALTER TABLE thread_metadata ADD COLUMN last_reply_received_at TIMESTAMPTZ;
SELECT attach_community_write_fence('personal_read_accounts');
SELECT attach_community_write_fence('personal_read_frontiers');

1
schema/schema.sql
View File

@@ -559,6 +559,7 @@ CREATE TABLE thread_metadata (
descendant_count INT NOT NULL DEFAULT 0,
last_reply_at TIMESTAMPTZ,
broadcast BOOLEAN NOT NULL DEFAULT FALSE,
last_reply_received_at TIMESTAMPTZ,
PRIMARY KEY (community_id, event_created_at, event_id),
FOREIGN KEY (community_id, channel_id) REFERENCES channels (community_id, id)
);