jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-04 16:31:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

relay: reshape buzz/v1 sidebar into channel and thread read scopes

Browse Source 
Channel rows are { unread, mentions, read_through_id }; thread rows are
{ root_id, unread, mentions, read_through_id, latest_id }. Clients only
handle message ids.

- Drop attention, latest_message_*, latest_reply_* and the
  max_thread_summaries cap (NIP-11 descriptor too).
- mentions: exact count of unread p-tag mentions per scope; a thread
  mention counts on the thread row, not the channel.
- read_through_id: last displayed message at or before the read position.
- Posting marks the author read through their own message at ingest, so
  projections no longer filter own messages.
- A mention follows the thread (thread row at rank 1); unfollow stays
  sticky and a later mention does not re-follow.
- Docs updated to the new shape.

Signed-off-by: hMB2 <0400eb0da32afbbf16cdd1dee6ffd243587e8f3d6a2823fe7dfa95d16b002b7d@buzz.block.builderlab.xyz>
This commit is contained in:
hMB2
2026-10-07 00:49:31 -07:00
parent 7e09d56c7b
commit a650b9efb0
18 changed files with 906 additions and 553 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
crates/buzz-db/src/runtime/mod.rs
View File

@@ -107,9 +107,17 @@ pub(crate) async fn insert_mentions_in_transaction(
// transaction so all chunks share its commit boundary.
const MENTION_INSERT_CHUNK_ROWS: usize = 5_000;
for chunk in valid_pubkeys.chunks(MENTION_INSERT_CHUNK_ROWS) {
// Arrival time and read scope come from the stored event and its
// thread row, which every caller writes before (or with) its mentions.
let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
"INSERT INTO event_mentions \
(community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) ",
(community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind, \
received_at, root_id) \
SELECT v.community_id, v.pubkey_hex, v.event_id, v.event_created_at, v.channel_id, \
v.event_kind, e.received_at, \
CASE WHEN tm.root_event_id <> v.event_id AND NOT (tm.depth = 1 AND tm.broadcast) \
THEN tm.root_event_id END \
FROM (",
);
qb.push_values(chunk, |mut b, pubkey| {
@@ -121,7 +129,14 @@ pub(crate) async fn insert_mentions_in_transaction(
.push_bind(kind as i32);
});
qb.push(" ON CONFLICT DO NOTHING");
qb.push(
") v (community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) \
LEFT JOIN events e ON e.community_id = v.community_id \
AND e.created_at = v.event_created_at AND e.id = v.event_id \
LEFT JOIN thread_metadata tm ON tm.community_id = v.community_id \
AND tm.event_created_at = v.event_created_at AND tm.event_id = v.event_id \
ON CONFLICT DO NOTHING",
);
qb.build().execute(&mut **tx).await?;
}

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

@@ -444,6 +444,17 @@ pub async fn insert_event_in_transaction(
) -> Result<(StoredEvent, bool)> {
let result = insert_event_on(tx.as_mut(), community_id, event, channel_id).await?;
if result.1 {
if let Some(channel) = channel_id {
crate::personal_read::record_message(
tx,
community_id,
event,
channel,
crate::personal_read::Place::TopLevel,
result.0.received_at,
)
.await?;
}
crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?;
}
Ok(result)
@@ -1853,12 +1864,15 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
.bind(received_at)
.execute(&mut **tx)
.await?;
crate::personal_read::record_reply(
crate::personal_read::record_message(
tx,
community_id,
event,
meta.channel_id,
root,
crate::personal_read::Place::Reply {
root,
broadcast: meta.depth == 1 && meta.broadcast,
},
received_at,
)
.await?;
@@ -1866,6 +1880,17 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
}
}
if let (None, Some(channel)) = (&thread_meta, channel_id) {
crate::personal_read::record_message(
tx,
community_id,
event,
channel,
crate::personal_read::Place::TopLevel,
received_at,
)
.await?;
}
crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?;
}

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

@@ -1,6 +1,6 @@
//! Read progress follows relay arrival (`received_at`), never author time.
//! Each case sets every arrival explicitly: back-to-back inserts share a clock.
use super::{postgres_tests, projection::LATEST_PROBE, *};
use super::{postgres_tests, *};
use crate::Db;
use buzz_core::CommunityId;
use nostr::{EventBuilder, Keys, Kind, Tag};
@@ -182,30 +182,82 @@ async fn future_dated_anchor_does_not_swallow_later_arrivals() {
assert!(sidebar(&db, community, &actor).await.unread);
}
/// Marking the sidebar's own latest message must clear the badge even when the
/// last arrival is not the newest by author time.
/// Clients name only display-order IDs. Marking the newest message shown
/// must clear the badge even when the last arrival is displayed above it.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn mark_as_read_with_the_sidebar_anchor_clears_a_late_arrival() {
async fn marking_the_newest_displayed_message_clears_a_late_arrival_above_it() {
let (db, pool, community, channel, actor, first) = fixture().await;
let now = first.created_at.as_secs();
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;
let newest = post(&db, &pool, community, channel, now + 1, now - 40).await;
let late = post(&db, &pool, community, channel, now - 600, now - 30).await;
assert!(sidebar(&db, community, &actor).await.unread);
let anchor = sidebar(&db, community, &actor)
.await
.latest_message_id
.expect("a channel with messages has a latest message");
apply(&db, community, &actor, mark_channel_read(channel, anchor)).await;
apply(
&db,
community,
&actor,
mark_through(channel, None, &newest.id.to_hex()),
)
.await;
let row = sidebar(&db, community, &actor).await;
assert!(!row.unread);
assert_eq!(
row.read_through_id,
Some(newest.id.to_hex()),
"the divider stays under the newest displayed message"
);
assert_eq!(
states(&db, community, &actor, channel, &[&late, &newest]).await,
["read", "read"]
);
}
/// A mark reads only what is displayed at or before its anchor: a message
/// shown below it stays unread, even one that arrived first.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn marking_leaves_messages_displayed_below_the_anchor_unread() {
let (db, pool, community, channel, actor, first) = fixture().await;
let now = first.created_at.as_secs();
arrive(&pool, community, &first, now - 60).await;
let below = post(&db, &pool, community, channel, now + 5, now - 50).await;
apply(
&db,
community,
&actor,
mark_through(channel, None, &first.id.to_hex()),
)
.await;
assert_eq!(
states(&db, community, &actor, channel, &[&first, &below]).await,
["read", "unread"]
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn mark_channel_read_with_the_newest_displayed_message_clears_a_late_arrival() {
let (db, pool, community, channel, actor, first) = fixture().await;
let now = first.created_at.as_secs();
arrive(&pool, community, &first, now - 60).await;
let newest = post(&db, &pool, community, channel, now + 1, now - 40).await;
post(&db, &pool, community, channel, now - 600, now - 30).await;
apply(
&db,
community,
&actor,
mark_channel_read(channel, newest.id.to_hex()),
)
.await;
assert!(!sidebar(&db, community, &actor).await.unread);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() {
async fn mark_thread_read_with_the_newest_displayed_reply_clears_a_late_reply() {
let (db, pool, community, channel, actor, root) = fixture().await;
let now = root.created_at.as_secs();
arrive(&pool, community, &root, now - 60).await;
@@ -216,7 +268,7 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() {
mark_through(channel, None, &root.id.to_hex()),
)
.await;
reply(
let newest = reply(
&db,
&pool,
community,
@@ -227,7 +279,7 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() {
now - 40,
)
.await;
reply(
let late = reply(
&db,
&pool,
community,
@@ -248,80 +300,25 @@ 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.attention, row.threads.len()), (false, true, 1));
assert_eq!(row.threads.len(), 1);
assert_eq!((row.unread, row.threads.len()), (false, 1));
assert_eq!(
row.threads[0].latest_id,
late.id.to_hex(),
"the last arrival"
);
assert_eq!(row.threads[0].mentions, 2);
let root_id = root.id.to_hex();
let anchor = row.threads[0].latest_reply_id.clone();
apply(
&db,
community,
&actor,
mark_through(channel, Some(&root_id), &anchor),
mark_through(channel, Some(&root_id), &newest.id.to_hex()),
)
.await;
let row = sidebar(&db, community, &actor).await;
assert_eq!((row.unread, row.attention, row.threads.len()), (false, false, 0));
assert!(row.threads.is_empty());
}
/// The latest probe holds only the newest messages by author time. A late
/// arrival authored before all of them is still counted, so it must still be
/// the anchor, or Mark as read with the sidebar anchor leaves it unread.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn mark_as_read_clears_a_late_arrival_behind_the_latest_probe() {
let (db, pool, community, channel, actor, first) = fixture().await;
let now = first.created_at.as_secs();
arrive(&pool, community, &first, now - 60).await;
for i in 0..LATEST_PROBE as u64 {
post(&db, &pool, community, channel, now - 500 + i, now - 50).await;
}
let anchor = sidebar(&db, community, &actor)
.await
.latest_message_id
.unwrap();
apply(&db, community, &actor, mark_channel_read(channel, anchor)).await;
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!(row.unread);
assert_eq!(row.latest_message_id, Some(late.id.to_hex()));
apply(
&db,
community,
&actor,
mark_channel_read(channel, late.id.to_hex()),
)
.await;
assert!(!sidebar(&db, community, &actor).await.unread);
}
/// Reactions never fill the latest probe.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn latest_message_behind_newer_reactions_is_found() {
let (db, pool, community, channel, actor, first) = fixture().await;
let now = first.created_at.as_secs();
let demoted = sqlx::query("UPDATE events SET kind=7 WHERE community_id=$1 AND id=$2")
.bind(community.as_uuid())
.bind(first.id.as_bytes().as_slice())
.execute(&pool)
.await
.unwrap()
.rows_affected();
assert_eq!(demoted, 1);
let only = post(&db, &pool, community, channel, now - 600, now - 10).await;
for i in 0..LATEST_PROBE as u64 {
post_kind(&db, &pool, community, channel, 7, now - 500 + i, now - 50).await;
}
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!(row.unread);
assert_eq!((row.unread, row.threads.len()), (false, 0));
}
/// Store `event` as having arrived at exactly `seconds` plus `micros`. Built
@@ -359,10 +356,13 @@ async fn mark_within_one_second(micros: [u32; 2], pick: usize) -> (Vec<String>,
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;
// Equal author times display in ID order; index 0 displays first, so a
// mark reads only what arrived by then, not the other by display order.
let mut both = [&first, &second];
both.sort_by_key(|e| e.id);
let arrived = now as i64 - 30;
arrive_exact(&pool, community, &first, arrived, micros[0]).await;
arrive_exact(&pool, community, &second, arrived, micros[1]).await;
let both = [&first, &second];
arrive_exact(&pool, community, both[0], arrived, micros[0]).await;
arrive_exact(&pool, community, both[1], arrived, micros[1]).await;
apply(
&db,
community,

57
crates/buzz-db/src/store/personal_read/bench_postgres_tests.rs
View File

@@ -2,11 +2,11 @@
//! 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 buzz_core::CommunityId;
use nostr::{EventBuilder, Keys, Kind, Tag};
use sqlx::postgres::PgPoolOptions;
use std::time::{Duration, Instant};
@@ -31,7 +31,9 @@ async fn top(db: &Db, c: CommunityId, ch: Uuid, e: &nostr::Event) -> Duration {
}
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 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,
@@ -72,16 +74,19 @@ async fn sidebar_bench(db: &Db, c: CommunityId, actor: &Keys, label: &str) {
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();
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={}",
"BENCH {label}: busiest unread={} mentions={} threads={} channels={}",
busy.unread,
busy.attention,
busy.mentions,
busy.threads.len(),
page.channels.len()
);
@@ -91,7 +96,10 @@ async fn sidebar_bench(db: &Db, c: CommunityId, actor: &Keys, label: &str) {
#[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 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())
@@ -107,10 +115,18 @@ async fn bench_unread() {
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,
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];
@@ -123,7 +139,10 @@ async fn bench_unread() {
top(&db, c, *ch, &e).await;
firsts.push(e);
}
assert_eq!(apply(&db, c, &actor, channel_read(busy, &firsts[0])).await, IntentOutcome::Applied);
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());
@@ -144,7 +163,10 @@ async fn bench_unread() {
}
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);
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;
}
@@ -155,7 +177,11 @@ async fn bench_unread() {
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 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);
}
@@ -193,7 +219,10 @@ async fn bench_unread() {
}
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);
println!(
"BENCH concurrent {label}: {n} inserts in {secs:.2}s = {:.0}/s",
n as f64 / secs
);
stats(&format!("ingest concurrent {label}"), all);
}
}

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

@@ -50,7 +50,7 @@ impl Db {
// The positions the sidebar counts against, both floored at the
// account's start and absent before it: the channel's, never
// before joining and absent for a non-member; and the thread's,
// which exists only for the actor's threads and includes any
// which exists only for threads the actor follows and includes any
// whole-channel cut.
let positions = sqlx::query(
"SELECT CASE WHEN m.joined_at IS NOT NULL AND s.started IS NOT NULL
@@ -65,7 +65,7 @@ impl Db {
LEFT JOIN personal_read_frontiers cf ON cf.community_id=$1 AND cf.actor=$2
AND cf.channel_id=$3 AND cf.root_id=''::bytea
LEFT JOIN personal_read_frontiers tf ON tf.community_id=$1 AND tf.actor=$2
AND tf.channel_id=$3 AND tf.root_id=$4 AND $4<>''::bytea",
AND tf.channel_id=$3 AND tf.root_id=$4 AND $4<>''::bytea AND tf.following",
)
.bind(community.as_uuid())
.bind(actor_bytes.as_slice())

86
crates/buzz-db/src/store/personal_read/membership.rs
View File

@@ -1,5 +1,7 @@
//! Thread membership. A thread frontier row exists exactly for the actor's
//! threads, and only ingest creates one: read intents move existing rows.
//! Read state written by ingest: posting marks the author read, and replies
//! and mentions follow threads. A thread frontier row exists for each thread
//! the actor follows or has unfollowed (`following=false`, so unfollow
//! sticks); only ingest and follow intents create one.
use super::model::ELIGIBLE_KINDS;
use crate::Result;
use buzz_core::CommunityId;
@@ -8,17 +10,34 @@ use nostr::Event;
use sqlx::{Postgres, Transaction};
use uuid::Uuid;
/// Record who a newly stored reply makes a thread member: its author, the
/// root's author, members it mentions, and in a DM (at most nine members)
/// every member. The author's row starts at the reply, which it has read;
/// everyone else's starts just before it, so the reply is unread. Existing
/// rows are left alone.
pub(crate) async fn record_reply(
/// Where a newly stored channel message sits.
#[derive(Clone, Copy)]
pub(crate) enum Place<'a> {
/// A top-level timeline message.
TopLevel,
/// A reply in `root`'s thread; `broadcast` when it also shows on the
/// timeline (a depth-1 reply broadcast to the channel).
Reply { root: &'a [u8], broadcast: bool },
}
/// Record what a newly stored message changes:
/// - its author has read through it, in each context it shows in, and a
/// reply (re-)follows the author;
/// - members it mentions follow the thread it is in, or the thread rooted at
/// it;
/// - a reply also makes the root's author, and in a DM (at most nine
/// members) every member, follow the thread.
///
/// Nobody but the author is re-followed: an unfollowed row stays unfollowed,
/// even for a new mention (whether a mention should re-follow is open).
///
/// Everyone but the author starts just before the message, so it is unread.
pub(crate) async fn record_message(
tx: &mut Transaction<'_, Postgres>,
community: CommunityId,
event: &Event,
channel: Uuid,
root: &[u8],
place: Place<'_>,
received_at: DateTime<Utc>,
) -> Result<()> {
if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) {
@@ -29,29 +48,54 @@ pub(crate) async fn record_reply(
.public_keys()
.map(|key| key.to_bytes().to_vec())
.collect();
// Both inserts' fence and foreign-key checks run at the end of the
// statement, after the account rows exist.
let (root, reply, on_timeline) = match place {
Place::TopLevel => (event.id.as_bytes().as_slice(), false, true),
Place::Reply { root, broadcast } => (root, true, broadcast),
};
// Rank 0 is the replying author, who follows (again) and has read; 1 a
// mentioned member and 2 anyone else the reply makes follow, whose
// existing rows are left alone. Every insert's fence and foreign-key
// checks run at the end of the statement, after the account rows exist.
sqlx::query(
"WITH joined AS (
SELECT $4::bytea AS actor, $6::timestamptz AS through
SELECT $4::bytea AS actor, $6::timestamptz AS through, 0 AS rank WHERE $8
UNION ALL
SELECT pubkey, $6-interval '1 microsecond' FROM events
WHERE community_id=$1 AND channel_id=$2 AND id=$3
SELECT cm.pubkey, $6-interval '1 microsecond', 1 FROM channel_members cm
WHERE cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL
AND cm.pubkey=ANY($5) AND cm.pubkey<>$4
UNION ALL
SELECT cm.pubkey, $6-interval '1 microsecond'
SELECT pubkey, $6-interval '1 microsecond', 2 FROM events
WHERE $8 AND community_id=$1 AND channel_id=$2 AND id=$3
UNION ALL
SELECT cm.pubkey, $6-interval '1 microsecond', 2
FROM channel_members cm JOIN channels c
ON c.community_id=cm.community_id AND c.id=cm.channel_id
WHERE cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL
AND (cm.pubkey=ANY($5) OR c.channel_type='dm')
WHERE $8 AND cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL
AND c.channel_type='dm'
), members AS (
SELECT DISTINCT ON (actor) actor, through FROM joined ORDER BY actor, through DESC
SELECT DISTINCT ON (actor) actor, through, rank FROM joined ORDER BY actor, rank
), accounts AS (
INSERT INTO personal_read_accounts (community_id, actor)
SELECT $1, actor FROM members ON CONFLICT DO NOTHING
SELECT $1, actor FROM members UNION SELECT $1, $4
ON CONFLICT DO NOTHING
), timeline AS (
INSERT INTO personal_read_frontiers
(community_id, actor, channel_id, root_id, through_timestamp)
SELECT $1, $4, $2, ''::bytea, $6 WHERE $7
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE
SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp,
excluded.through_timestamp)
), replied AS (
INSERT INTO personal_read_frontiers
(community_id, actor, channel_id, root_id, through_timestamp)
SELECT $1, actor, $2, $3, through FROM members WHERE rank=0
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE
SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp,
excluded.through_timestamp), following=true
)
INSERT INTO personal_read_frontiers
(community_id, actor, channel_id, root_id, through_timestamp)
SELECT $1, actor, $2, $3, through FROM members
SELECT $1, actor, $2, $3, through FROM members WHERE rank>0
ON CONFLICT DO NOTHING",
)
.bind(community.as_uuid())
@@ -60,6 +104,8 @@ pub(crate) async fn record_reply(
.bind(event.pubkey.to_bytes().as_slice())
.bind(&mentioned)
.bind(received_at)
.bind(on_timeline)
.bind(reply)
.execute(&mut **tx)
.await?;
Ok(())

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

@@ -1,8 +1,9 @@
//! Private, signer-owned accessory read progress, separate from NIP-RS events.
//!
//! A frontier is the relay arrival time of the message a context was read
//! through, and unread counts forward from it. Only fixed context intents
//! advance frontiers; ingest only creates thread membership rows.
//! through, and unread counts forward from it. Fixed intents advance and
//! follow; ingest advances the author's own frontiers (posting marks read)
//! and creates follow rows for replies and mentions.
mod classification;
mod context;
@@ -11,7 +12,7 @@ mod model;
mod projection;
mod writes;
pub(crate) use membership::record_reply;
pub(crate) use membership::{record_message, Place};
pub use model::*;
#[cfg(test)]

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

@@ -18,7 +18,8 @@ pub struct ReadTarget {
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
pub enum ReadIntent {
/// Advance a context through one fixed message, including equal arrivals.
/// Advance a context through one fixed message: everything displayed at
/// or before it in that context, and anything with an equal arrival.
MarkThrough {
/// Channel or canonical thread being marked.
target: ReadTarget,
@@ -26,13 +27,25 @@ pub enum ReadIntent {
message_id: String,
},
/// Advance the channel timeline and every thread in it through one fixed
/// message's arrival time. The anchor may be a reply; ancestry is irrelevant.
/// message: everything in the channel displayed at or before it. The
/// anchor may be a reply; ancestry is irrelevant.
MarkChannelRead {
/// Channel being marked, including all of its threads.
channel_id: Uuid,
/// Fixed anchor; retry must not substitute the latest message.
message_id: String,
},
/// Follow a thread. A thread you did not follow starts caught up.
Follow {
/// The thread; a target without a root is invalid.
target: ReadTarget,
},
/// Stop following a thread. Sticky: later replies and mentions do not
/// re-follow; your own reply or `follow` does.
Unfollow {
/// The thread; a target without a root is invalid.
target: ReadTarget,
},
}
/// Outcome for one independent transaction, never acknowledged before commit.
@@ -49,8 +62,6 @@ pub enum IntentOutcome {
/// Maximum channel summaries in one sidebar page.
pub const MAX_CHANNELS: usize = 20;
/// Maximum unread-thread summaries per channel row.
pub const MAX_THREAD_SUMMARIES: usize = 5;
/// 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.
@@ -59,6 +70,7 @@ pub const MAX_ARRIVAL_SKEW_SECONDS: u32 = 900;
pub const ELIGIBLE_KINDS: [i32; 4] = [9, 40002, 45001, 45003];
/// One joined-channel summary, not a second conversation/history API.
/// Clients receive message IDs only; positions stay relay-internal.
#[derive(Debug, Serialize)]
pub struct ChannelReadSummary {
/// Joined channel UUID.
@@ -71,34 +83,33 @@ pub struct ChannelReadSummary {
pub archived: bool,
/// Existing DM visibility preference (not an authorization decision).
pub hidden: bool,
/// An unread timeline message without a [`Reason`] exists: ordinary
/// backlog. Never true in a DM, where every message is direct.
/// The channel timeline has a message past the actor's position. Thread
/// replies never set it; they show on their thread row.
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 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
/// exactly when it is.
pub latest_message_at: Option<i64>,
/// The actor's threads with unread replies, newest unread reply first.
/// Unread timeline messages that mention the actor. Exact, uncapped.
pub mentions: i64,
/// The last timeline message, in display order, at or before the actor's
/// position: the "new" divider goes below it. None when there is none.
pub read_through_id: Option<String>,
/// Followed threads with an unread reply, newest unread reply first.
pub threads: Vec<ThreadReadSummary>,
}
/// One of the actor's threads with an unread reply. No conversation bytes.
/// One followed thread with an unread reply. No conversation bytes.
#[derive(Debug, Deserialize, Serialize)]
pub struct ThreadReadSummary {
/// Canonical thread-root event ID.
pub root_id: String,
/// 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
/// necessarily latest_reply_id's own; summaries order by it.
pub latest_reply_at: i64,
/// A reply past the actor's thread position. Listed rows are unread.
pub unread: bool,
/// Unread replies that mention the actor. Exact, uncapped.
pub mentions: i64,
/// The last reply, in display order, at or before the actor's thread
/// position. None when the actor has read no reply.
pub read_through_id: Option<String>,
/// Last unread reply to arrive: marking through it reads the thread, and
/// clients fetch it by ID for a preview.
pub latest_id: String,
}
/// A bounded roster page, with no cross-page snapshot or removal inference.

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

@@ -236,7 +236,7 @@ async fn accounts(pool: &PgPool, community: CommunityId) -> i64 {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn unread_is_presence_and_serializes_as_booleans() {
async fn sidebar_row_has_the_wire_shape() {
let (db, _, community, channel, actor, _) = fixture().await;
let author = Keys::generate();
let base = now();
@@ -246,18 +246,37 @@ async fn unread_is_presence_and_serializes_as_booleans() {
}
let row = sidebar(&db, community, &actor).await;
let wire = serde_json::to_value(&row).unwrap();
let mut keys: Vec<_> = wire.as_object().unwrap().keys().cloned().collect();
keys.sort();
assert_eq!(
keys,
[
"archived",
"channel_id",
"channel_type",
"hidden",
"mentions",
"name",
"read_through_id",
"threads",
"unread"
]
);
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}");
}
assert_eq!(wire["mentions"], 0);
assert_eq!(
wire["read_through_id"],
serde_json::Value::Null,
"nothing read yet"
);
let last = last.unwrap();
assert_eq!(row.latest_message_id, Some(last.id.to_hex()));
assert_eq!(
apply(&db, community, &actor, mark(channel, None, &last)).await,
IntentOutcome::Applied
);
assert!(!sidebar(&db, community, &actor).await.unread);
let row = sidebar(&db, community, &actor).await;
assert!(!row.unread);
assert_eq!(row.read_through_id, Some(last.id.to_hex()));
}
#[tokio::test]
@@ -279,11 +298,11 @@ async fn joining_starts_caught_up_and_counts_only_what_arrives_after() {
.unwrap();
let row = sidebar(&db, community, &joiner).await;
assert_eq!(
(row.unread, row.attention),
(false, false),
(row.unread, row.mentions),
(false, 0),
"history before joining is read"
);
assert_eq!(row.latest_message_id, Some(event.id.to_hex()));
assert_eq!(row.read_through_id, Some(event.id.to_hex()));
assert_eq!(
status(&db, community, &joiner, channel, None, &[&event]).await,
["read"]
@@ -303,7 +322,11 @@ 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), (false, true), "a mention is attention, not backlog");
assert_eq!(
(row.unread, row.mentions),
(true, 1),
"a mention is unread and counted"
);
assert_eq!(
states(&db, community, &joiner, channel, None, &[&later]).await,
[serde_json::json!({"status":"unread","reason":"mention"})]
@@ -402,11 +425,7 @@ 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.threads.len()),
(false, false, 0)
);
assert_eq!(row.latest_message_id, Some(before.id.to_hex()));
assert_eq!((row.unread, row.mentions, row.threads.len()), (false, 0, 0));
assert_eq!(
status(&db, community, &member, channel, None, &[&before]).await,
["not_counted"]
@@ -436,7 +455,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.attention, row.threads.len()), (true, true, 1));
assert_eq!((row.unread, row.mentions, row.threads.len()), (true, 0, 1));
}
#[tokio::test]
@@ -535,12 +554,12 @@ 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"], true);
assert_eq!(before, after, "not unread, not attention, not latest");
assert_eq!(before, after, "not unread, not a mention");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn latest_includes_own_and_excludes_deleted_and_auxiliary() {
async fn posting_marks_read_and_read_through_skips_deleted_and_auxiliary() {
let (db, pool, community, channel, actor, _) = fixture().await;
let base = now();
let own = post(&db, community, channel, &actor, base + 1, vec![]).await;
@@ -553,10 +572,13 @@ async fn latest_includes_own_and_excludes_deleted_and_auxiliary() {
.await
.unwrap();
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!(row.unread, "own, deleted and reaction do not count");
// An empty channel has no latest message.
assert!(!row.unread, "posting read through everything before it");
assert_eq!(
row.read_through_id,
Some(own.id.to_hex()),
"not deleted, not a reaction"
);
// An empty channel has nothing read.
db.create_channel(
community,
"truly empty",
@@ -577,7 +599,7 @@ async fn latest_includes_own_and_excludes_deleted_and_auxiliary() {
.iter()
.find(|c| c.channel_id != channel)
.unwrap();
assert!(empty.latest_message_id.is_none() && empty.latest_message_at.is_none());
assert!(!empty.unread && empty.mentions == 0 && empty.read_through_id.is_none());
}
#[tokio::test]

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

@@ -1,6 +1,8 @@
//! 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.
//! Sidebar read state from read positions. `unread` is an existence probe
//! that stops at the first qualifying row, and `mentions` walks only the
//! actor's mentions past the position, so a caught-up reader examines almost
//! nothing. Posting marks the author read at ingest, so no probe filters out
//! the actor's own messages.
use super::{model::*, writes};
use buzz_core::CommunityId;
@@ -17,8 +19,8 @@ const ON_TIMELINE: &str = "(tm.root_event_id IS NULL OR tm.root_event_id=e.id
/// 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 UNREAD_EVENT: &str = "e.created_at >= r.position-$7 AND e.received_at > r.position
AND e.kind=ANY($6) 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";
@@ -77,19 +79,18 @@ impl Db {
// Every position is floored at the account's start (infinity before
// 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.
// up. A thread's is its row, which exists only for threads the actor
// follows (see `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.
// - unread: the first timeline message past the position
// (idx_events_community_channel_created);
// - mentions: the actor's mentions in that scope that arrived past
// the position (idx_event_mentions_scope_received);
// - read_through_id: the newest timeline message, by author time, that
// arrived at or before the position. Anything that arrived by then
// was authored no later than the position plus the skew;
// - threads: followed roots whose last reply arrived after the
// thread position, then their unread replies and mentions.
let sql = format!(
r#"WITH roster AS MATERIALIZED (
SELECT c.id, c.name, c.channel_type::text AS channel_type,
@@ -108,84 +109,64 @@ 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), (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,
ordinary.id IS NOT NULL AS unread,
directed.id IS NOT NULL OR threads.id IS NOT NULL AS attention,
EXISTS (SELECT 1 FROM events e {EVENT_THREAD_ROW}
WHERE e.community_id=$1 AND e.channel_id=r.id
AND {UNREAD_EVENT} AND {ON_TIMELINE}) AS unread,
(SELECT count(*) 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
WHERE m.community_id=$1 AND m.pubkey_hex=$8 AND m.channel_id=r.id
AND m.root_id IS NULL AND m.received_at > r.position
AND e.channel_id=r.id AND {UNREAD_EVENT}) AS mentions,
(SELECT encode(e.id,'hex') FROM events e {EVENT_THREAD_ROW}
WHERE e.community_id=$1 AND e.channel_id=r.id
AND e.created_at <= r.position+$7 AND e.received_at <= r.position
AND e.kind=ANY($6) AND e.deleted_at IS NULL AND {ON_TIMELINE}
ORDER BY e.created_at DESC, e.id LIMIT 1) AS read_through_id,
COALESCE(threads.items, '[]') AS threads
FROM roster r
LEFT JOIN LATERAL (
SELECT (array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id,
max(received_at) AS arrival, max(extract(epoch FROM created_at)::bigint) AS at
FROM (SELECT id,created_at,received_at FROM events
WHERE community_id=$1 AND channel_id=r.id AND kind=ANY($6) AND deleted_at IS NULL
ORDER BY created_at DESC LIMIT $7) newest
) latest ON true
LEFT JOIN LATERAL (
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'),
'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
'unread',true,'mentions',nm.mentions,'read_through_id',rt.id,
'latest_id',n.id) ORDER BY n.at DESC, tf.root_id) AS items
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 (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
max(e.created_at) AS 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 tm.event_created_at >= p.position-$7
AND e.received_at > p.position AND e.kind=ANY($6)
AND e.deleted_at IS NULL AND NOT (tm.depth=1 AND tm.broadcast)
) n
CROSS JOIN LATERAL (
SELECT count(*) AS mentions
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
WHERE m.community_id=$1 AND m.pubkey_hex=$8 AND m.channel_id=r.id
AND m.root_id=tf.root_id AND m.received_at > p.position
AND e.kind=ANY($6) AND e.deleted_at IS NULL
) nm
LEFT JOIN LATERAL (
SELECT encode(tm.event_id,'hex') AS id
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+$7
AND e.received_at <= p.position AND e.kind=ANY($6) AND e.deleted_at IS NULL
ORDER BY tm.event_created_at DESC, tm.event_id LIMIT 1
) rt ON true
WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=r.id
AND tf.root_id<>''::bytea AND root.last_reply_received_at > p.position
AND n.id IS NOT NULL
AND tf.root_id<>''::bytea AND tf.following
AND root.last_reply_received_at > p.position AND n.id IS NOT NULL
) threads ON true
ORDER BY r.id"#
)
.replace("$hex", "$9");
);
// Only compile-time constants are interpolated.
let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(community.as_uuid())
@@ -194,7 +175,6 @@ impl Db {
.bind(only)
.bind((limit + 1) as i64)
.bind(ELIGIBLE_KINDS.as_slice())
.bind(LATEST_PROBE as i64)
.bind(skew())
.bind(actor.to_hex())
.fetch_all(&mut *tx)
@@ -203,8 +183,9 @@ 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<ThreadReadSummary> = serde_json::from_value(row.try_get("threads")?)
.map_err(|_| DbError::InvalidData("invalid thread summaries".into()))?;
let threads: Vec<ThreadReadSummary> =
serde_json::from_value(row.try_get("threads")?)
.map_err(|_| DbError::InvalidData("invalid thread summaries".into()))?;
channels.push(ChannelReadSummary {
channel_id: row.try_get("id")?,
name: row.try_get("name")?,
@@ -212,10 +193,9 @@ impl Db {
archived: row.try_get("archived")?,
hidden: row.try_get("hidden")?,
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),
mentions: row.try_get("mentions")?,
read_through_id: row.try_get("read_through_id")?,
threads,
});
}
let next_cursor = if has_more {
@@ -230,55 +210,6 @@ impl Db {
}
}
/// Newest messages, by author time, that join the counted ones as latest candidates.
pub(super) const LATEST_PROBE: usize = 32;
pub(super) fn skew() -> chrono::Duration {
chrono::Duration::seconds(i64::from(MAX_ARRIVAL_SKEW_SECONDS))
}
/// 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| {
b.latest_reply_at
.cmp(&a.latest_reply_at)
.then_with(|| a.root_id.cmp(&b.root_id))
});
items.truncate(MAX_THREAD_SUMMARIES);
items
}
#[cfg(test)]
mod tests {
use super::*;
fn item(root: u8, at: i64) -> ThreadReadSummary {
ThreadReadSummary {
root_id: hex::encode([root; 32]),
latest_reply_id: hex::encode([root; 32]),
latest_reply_at: at,
}
}
#[test]
fn thread_summaries_order_newest_first_break_ties_by_root_and_cap_at_five() {
// Literal contract boundaries deliberately do not derive from the constant.
for count in [0_u8, 1, 4, 5, 6, 7] {
// Descending roots with pairwise-equal times exercise the tie-break.
let items: Vec<_> = (0..count)
.rev()
.map(|i| item(i, i64::from(i / 2)))
.collect();
let summaries = summarize(items);
let roots: Vec<_> = summaries.iter().map(|t| t.root_id.clone()).collect();
let mut expected: Vec<_> = (0..count).map(|i| (i64::from(i / 2), i)).collect();
expected.sort_by(|a, b| b.0.cmp(&a.0).then(a.1.cmp(&b.1)));
let expected: Vec<_> = expected
.into_iter()
.take(5)
.map(|(_, i)| hex::encode([i; 32]))
.collect();
assert_eq!(roots, expected, "count {count}");
}
}
}

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

@@ -14,39 +14,38 @@ async fn sidebar_sql_eligibility_matches_selector() {
message_ids: vec![event.id.to_hex()],
}];
for kind in [9, 40002, 45001, 45003, 1, 7, 39002, 40008] {
for own in [false, true] {
for deleted in [false, true] {
sqlx::query("UPDATE events SET kind=$2,pubkey=$3,deleted_at=CASE WHEN $4 THEN now() ELSE NULL END WHERE community_id=$1")
.bind(community.as_uuid()).bind(kind)
.bind(if own { actor.public_key().to_bytes() } else { event.pubkey.to_bytes() }.as_slice())
.bind(deleted).execute(&pool).await.unwrap();
let page = db
.personal_read_sidebar(community, &actor.public_key(), 20, None)
.await
.unwrap();
let expected = u32::from(ELIGIBLE_KINDS.contains(&kind) && !own && !deleted);
assert_eq!(
u32::from(page.channels[0].unread), expected,
"kind={kind} own={own} deleted={deleted}"
);
let contexts = db
.personal_read_contexts(community, &actor.public_key(), &query)
.await
.unwrap();
assert_eq!(
serde_json::to_value(&contexts).unwrap()["contexts"][0]["messages"][0]
["status"],
["not_counted", "unread"][expected as usize],
"kind={kind} own={own} deleted={deleted}"
);
}
// Own messages need no rule here: posting marks the author read at
// ingest (`posting_marks_read_and_read_through_skips_deleted_and_auxiliary`).
for deleted in [false, true] {
sqlx::query("UPDATE events SET kind=$2,deleted_at=CASE WHEN $3 THEN now() ELSE NULL END WHERE community_id=$1")
.bind(community.as_uuid()).bind(kind)
.bind(deleted).execute(&pool).await.unwrap();
let page = db
.personal_read_sidebar(community, &actor.public_key(), 20, None)
.await
.unwrap();
let expected = u32::from(ELIGIBLE_KINDS.contains(&kind) && !deleted);
assert_eq!(
u32::from(page.channels[0].unread),
expected,
"kind={kind} deleted={deleted}"
);
let contexts = db
.personal_read_contexts(community, &actor.public_key(), &query)
.await
.unwrap();
assert_eq!(
serde_json::to_value(&contexts).unwrap()["contexts"][0]["messages"][0]["status"],
["not_counted", "unread"][expected as usize],
"kind={kind} deleted={deleted}"
);
}
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn sidebar_directed_fact_matches_selector_classifier() {
async fn sidebar_mentions_match_selector_classifier() {
let (db, pool, community, channel, actor, event) = fixture().await;
let actor_hex = actor.public_key().to_hex();
let fullwidth: String = actor_hex
@@ -79,9 +78,9 @@ async fn sidebar_directed_fact_matches_selector_classifier() {
.await
.unwrap();
for tags in &cases {
// A top-level message: `broadcast` tags alone direct nothing.
// A top-level message: `broadcast` tags alone direct nothing, and
// in a DM every message is direct but only mentions count.
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))
@@ -94,13 +93,13 @@ async fn sidebar_directed_fact_matches_selector_classifier() {
.execute(&pool)
.await
.unwrap();
let mentioned = tags.iter().any(|t| {
t.len() >= 2 && t[0] == "p" && t[1].eq_ignore_ascii_case(&actor_hex)
});
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",
"INSERT INTO event_mentions (community_id,pubkey_hex,event_id,event_created_at,channel_id,event_kind,received_at)
SELECT community_id,$2,id,created_at,channel_id,kind,received_at FROM events WHERE community_id=$1 AND id=$3",
)
.bind(community.as_uuid())
.bind(&actor_hex)
@@ -116,10 +115,13 @@ async fn sidebar_directed_fact_matches_selector_classifier() {
.channels
.remove(0);
assert_eq!(
(row.unread, row.attention),
expected,
(row.unread, row.mentions),
(true, i64::from(mentioned)),
"channel_type={channel_type} tags={tags:?}"
);
if channel_type != "dm" {
assert_eq!(reason.is_some(), mentioned, "tags={tags:?}");
}
}
}
}

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

@@ -65,12 +65,11 @@ 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.threads.len()),
(false, true, 1)
);
assert_eq!(row.threads[0].latest_reply_id, own.id.to_hex());
// Only the reply is unread: posting read the root, and a reply never
// makes the channel unread.
assert_eq!((row.unread, row.mentions, row.threads.len()), (false, 0, 1));
assert_eq!(row.threads[0].latest_id, own.id.to_hex());
assert_eq!(row.threads[0].read_through_id, None, "no reply read yet");
assert_eq!(
states(&db, community, &starter, channel, Some(&root), &[&own]).await,
[serde_json::json!({"status":"unread","reason":"conversation"})]
@@ -92,11 +91,13 @@ async fn replying_mentioning_and_starting_a_thread_make_members() {
)
.await;
let row = sidebar(&db, community, &actor).await;
assert_eq!((row.unread, row.mentions, row.threads.len()), (true, 0, 1));
assert_eq!(row.threads[0].latest_id, other.id.to_hex());
assert_eq!(
(row.unread, row.attention, row.threads.len()),
(true, true, 1)
row.threads[0].read_through_id,
Some(own.id.to_hex()),
"posting read the actor's own reply"
);
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.
let outsider = Keys::generate();
@@ -116,11 +117,9 @@ async fn replying_mentioning_and_starting_a_thread_make_members() {
assert_eq!(thread_rows(&pool, community, &bystander).await, root_row);
assert!(thread_rows(&pool, community, &outsider).await.is_empty());
let row = sidebar(&db, community, &bystander).await;
// The mention, and the root on the timeline.
assert_eq!(
(row.unread, row.attention, row.threads.len()),
(true, true, 1)
);
// The root on the timeline; the mention counts on the thread row only.
assert_eq!((row.unread, row.mentions, row.threads.len()), (true, 0, 1));
assert_eq!(row.threads[0].mentions, 1);
assert_eq!(
states(
&db,
@@ -262,10 +261,8 @@ async fn dm_replies_make_every_member_a_thread_member() {
.unwrap()
.channels
.remove(0);
assert_eq!(
(row.unread, row.attention, row.threads.len()),
(false, true, 1)
);
// The root is unread on the timeline, the reply on the thread row.
assert_eq!((row.unread, row.mentions, row.threads.len()), (true, 0, 1));
}
#[tokio::test]
@@ -316,13 +313,9 @@ 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, 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());
assert!(row.unread, "top at +20 remains");
assert_eq!(row.threads.len(), 1, "reply at +30 remains");
assert_eq!(row.threads[0].latest_id, second.id.to_hex());
assert_eq!(
apply(&db, community, &actor, channel_read(channel, &second)).await,
@@ -330,8 +323,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.attention, row.threads.len()),
(false, false, 0),
(row.unread, row.threads.len()),
(false, 0),
"every thread is covered"
);
@@ -359,8 +352,8 @@ async fn mark_channel_read_covers_every_thread_through_a_reply_anchor() {
)
.await;
let row = sidebar(&db, community, &actor).await;
assert_eq!((row.unread, row.attention), (false, true));
assert_eq!(row.threads[0].latest_reply_id, newer.id.to_hex());
assert!(!row.unread);
assert_eq!(row.threads[0].latest_id, newer.id.to_hex());
assert_eq!(
status(
&db,
@@ -412,7 +405,7 @@ async fn mark_channel_read_validates_its_anchor() {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn thread_summaries_order_cap_and_count_every_unread_thread() {
async fn thread_summaries_order_newest_first_and_list_every_unread_thread() {
let (db, _, community, channel, actor, _) = fixture().await;
let base = now();
let mut roots = Vec::new();
@@ -456,8 +449,9 @@ 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, row.threads.len()), (true, true, 5));
// Posting the roots read the timeline through them; no thread cap.
assert_eq!((row.unread, row.threads.len()), (false, 6));
assert_eq!(row.read_through_id, Some(roots[5].id.to_hex()));
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();
@@ -468,22 +462,21 @@ async fn thread_summaries_order_cap_and_count_every_unread_thread() {
tied[1].clone(),
roots[2].id.to_hex(),
roots[3].id.to_hex(),
roots[4].id.to_hex()
roots[4].id.to_hex(),
roots[5].id.to_hex()
]
);
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.
assert_eq!(row.latest_message_at, Some((base + 150) as i64));
// Ordered by its newest reply's author time, anchored at its last arrival.
assert_eq!(row.threads[2].latest_id, last.unwrap().id.to_hex());
// Reading a listed thread through its anchor promotes the sixth.
// Reading a listed thread through its anchor removes only it.
let first = &row.threads[0];
let intent = ReadIntent::MarkThrough {
target: ReadTarget {
channel_id: channel,
root_id: Some(first.root_id.clone()),
},
message_id: first.latest_reply_id.clone(),
message_id: first.latest_id.clone(),
};
assert_eq!(
apply(&db, community, &actor, intent).await,
@@ -492,8 +485,7 @@ async fn thread_summaries_order_cap_and_count_every_unread_thread() {
let row = sidebar(&db, community, &actor).await;
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!(row.unread);
assert!(!row.unread);
}
#[tokio::test]
@@ -536,11 +528,13 @@ 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, row.attention), (true, true), "the root and two replies");
// The fixture's root; each reply's mention counts on its own thread.
assert_eq!((row.unread, row.mentions), (true, 0));
assert_eq!(row.threads.len(), 2);
assert!(row.threads.iter().all(|t| t.mentions == 1));
let listed = &row.threads[1];
assert_eq!(listed.root_id, diff.id.to_hex());
assert_eq!(listed.latest_reply_id, on_diff.id.to_hex());
assert_eq!(listed.latest_id, on_diff.id.to_hex());
assert_eq!(
states(&db, community, &actor, channel, Some(&diff), &[&on_diff]).await,
[serde_json::json!({"status":"unread","reason":"mention"})]
@@ -559,9 +553,9 @@ 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, row.attention), (true, true));
assert!(row.unread);
assert_eq!(row.threads.len(), 1);
assert_eq!(row.threads[0].latest_reply_id, elsewhere.id.to_hex());
assert_eq!(row.threads[0].latest_id, elsewhere.id.to_hex());
}
#[tokio::test]
@@ -608,3 +602,177 @@ async fn targeted_sidebar_returns_only_requested_joined_channels() {
.is_err());
}
}
fn follow(channel: Uuid, root: &nostr::Event, following: bool) -> ReadIntent {
let target = ReadTarget {
channel_id: channel,
root_id: Some(root.id.to_hex()),
};
if following {
ReadIntent::Follow { target }
} else {
ReadIntent::Unfollow { target }
}
}
async fn following(pool: &PgPool, community: CommunityId, actor: &Keys) -> Vec<bool> {
sqlx::query_scalar(
"SELECT following FROM personal_read_frontiers
WHERE community_id=$1 AND actor=$2 AND root_id<>''::bytea",
)
.bind(community.as_uuid())
.bind(actor.public_key().to_bytes().as_slice())
.fetch_all(pool)
.await
.unwrap()
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn follow_starts_caught_up_and_unfollow_is_sticky() {
let (db, pool, community, channel, actor, root) = fixture().await;
let base = now();
let other = Keys::generate();
let say = |at: u64, tags: Vec<nostr::Tag>| {
let (db, root, other) = (db.clone(), root.clone(), other.clone());
async move { reply(&db, community, channel, &root, &other, at, tags, false).await }
};
say(base, vec![]).await;
assert!(following(&pool, community, &actor).await.is_empty());
// Follow starts at the thread's last reply: nothing unread yet.
assert_eq!(
apply(&db, community, &actor, follow(channel, &root, true)).await,
IntentOutcome::Applied
);
assert!(sidebar(&db, community, &actor).await.threads.is_empty());
let next = say(base + 1, vec![]).await;
let row = sidebar(&db, community, &actor).await;
assert_eq!(row.threads.len(), 1);
assert_eq!(row.threads[0].latest_id, next.id.to_hex());
// Unfollow keeps the row; later replies, mentions included, don't
// re-follow (whether a new mention should is an open product question).
assert_eq!(
apply(&db, community, &actor, follow(channel, &root, false)).await,
IntentOutcome::Applied
);
assert!(sidebar(&db, community, &actor).await.threads.is_empty());
say(base + 2, vec![]).await;
say(base + 3, mention(&actor)).await;
assert_eq!(following(&pool, community, &actor).await, [false]);
assert!(sidebar(&db, community, &actor).await.threads.is_empty());
// The actor's own reply re-follows, read through it.
let own = reply(
&db,
community,
channel,
&root,
&actor,
base + 4,
vec![],
false,
)
.await;
assert_eq!(following(&pool, community, &actor).await, [true]);
assert!(sidebar(&db, community, &actor).await.threads.is_empty());
say(base + 5, vec![]).await;
let row = sidebar(&db, community, &actor).await;
assert_eq!(row.threads.len(), 1);
assert_eq!(
(
row.threads[0].mentions,
row.threads[0].read_through_id.clone()
),
(0, Some(own.id.to_hex())),
"the earlier mention was read when the actor replied"
);
// Following again catches up; a target without a root is invalid.
assert_eq!(
apply(&db, community, &actor, follow(channel, &root, false)).await,
IntentOutcome::Applied
);
assert_eq!(
apply(&db, community, &actor, follow(channel, &root, true)).await,
IntentOutcome::Applied
);
assert!(sidebar(&db, community, &actor).await.threads.is_empty());
let channel_only = ReadIntent::Follow {
target: ReadTarget {
channel_id: channel,
root_id: None,
},
};
assert_eq!(
apply(&db, community, &actor, channel_only).await,
IntentOutcome::Invalid
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn mentions_count_exactly_in_their_own_scope() {
let (db, _, community, channel, actor, root) = fixture().await;
let base = now();
let author = Keys::generate();
let first = post(&db, community, channel, &author, base + 1, mention(&actor)).await;
post(&db, community, channel, &author, base + 2, mention(&actor)).await;
post(&db, community, channel, &author, base + 3, vec![]).await;
for at in [base + 4, base + 5, base + 6] {
reply(
&db,
community,
channel,
&root,
&author,
at,
mention(&actor),
false,
)
.await;
}
// A broadcast reply shows on the timeline, so its mention counts there.
reply(
&db,
community,
channel,
&root,
&author,
base + 7,
mention(&actor),
true,
)
.await;
let row = sidebar(&db, community, &actor).await;
assert_eq!((row.unread, row.mentions), (true, 3));
let thread = row
.threads
.iter()
.find(|t| t.root_id == root.id.to_hex())
.unwrap();
assert_eq!(
thread.mentions, 3,
"thread mentions never count on the channel"
);
// Reading through the first mention leaves the rest, still exact.
assert_eq!(
apply(&db, community, &actor, mark(channel, None, &first)).await,
IntentOutcome::Applied
);
let row = sidebar(&db, community, &actor).await;
assert_eq!(row.mentions, 2);
assert_eq!(row.read_through_id, Some(first.id.to_hex()));
let thread = row
.threads
.iter()
.find(|t| t.root_id == root.id.to_hex())
.unwrap();
assert_eq!(
thread.mentions, 3,
"a channel mark leaves thread positions alone"
);
}

128
crates/buzz-db/src/store/personal_read/writes.rs
View File

@@ -86,6 +86,7 @@ async fn access(
struct Message {
id: Vec<u8>,
created_at: DateTime<Utc>,
received_at: DateTime<Utc>,
/// Canonical thread root, when the message is a reply.
thread: Option<Vec<u8>>,
@@ -103,7 +104,7 @@ async fn message(
kinds: Option<&[i32]>,
) -> Result<Option<Message>> {
let row = sqlx::query(
"SELECT e.id, e.received_at, tm.root_event_id, tm.depth, tm.broadcast
"SELECT e.id, e.created_at, e.received_at, tm.root_event_id, tm.depth, tm.broadcast
FROM events e LEFT JOIN thread_metadata tm ON tm.community_id=e.community_id
AND tm.event_created_at=e.created_at AND tm.event_id=e.id AND tm.channel_id=e.channel_id
WHERE e.community_id=$1 AND e.channel_id=$2 AND e.id=$3
@@ -118,6 +119,7 @@ async fn message(
&& row.try_get::<Option<bool>, _>("broadcast")? == Some(true);
Ok(Message {
on_timeline: thread.is_none() || broadcast_reply,
created_at: row.try_get("created_at")?,
received_at: row.try_get("received_at")?,
thread,
id,
@@ -126,15 +128,16 @@ async fn message(
.transpose()
}
/// An eligible-kind message's arrival time, deleted or not, without tags.
async fn anchor_received_at(
/// An eligible-kind message's author and arrival times, deleted or not,
/// without tags.
async fn anchor_times(
conn: &mut PgConnection,
community: CommunityId,
channel: Uuid,
id: &[u8],
) -> Result<Option<DateTime<Utc>>> {
Ok(sqlx::query_scalar(
"SELECT received_at FROM events
) -> Result<Option<(DateTime<Utc>, DateTime<Utc>)>> {
Ok(sqlx::query_as(
"SELECT created_at, received_at FROM events
WHERE community_id=$1 AND channel_id=$2 AND id=$3 AND kind=ANY($4) LIMIT 1",
)
.bind(community.as_uuid())
@@ -145,6 +148,47 @@ async fn anchor_received_at(
.await?)
}
/// Where a mark through an anchor leaves the frontier: the last arrival among
/// the messages displayed (by author time, then ID) at or before the anchor
/// in `scope`. Clients name only display-order IDs, so a late arrival shown
/// above the anchor is read with it instead of leaving a badge no mark can
/// clear. Anything that arrived after the anchor was authored no earlier
/// than the skew before the anchor's arrival, which bounds the range.
/// `scope` is a thread root, empty for the channel timeline, or None for
/// the whole channel.
async fn display_through(
conn: &mut PgConnection,
community: CommunityId,
channel: Uuid,
anchor: (&[u8], DateTime<Utc>, DateTime<Utc>),
scope: Option<&[u8]>,
) -> Result<DateTime<Utc>> {
let (id, created_at, received_at) = anchor;
let through: Option<DateTime<Utc>> = sqlx::query_scalar(
"SELECT max(e.received_at) 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=$2 AND e.kind=ANY($3)
AND e.created_at >= $4::timestamptz-$7 AND e.created_at <= $5
AND (e.created_at, e.id) <= ($5, $6) AND e.received_at > $4
AND CASE WHEN $8::bytea IS NULL THEN true
WHEN $8=''::bytea THEN tm.root_event_id IS NULL OR tm.root_event_id=e.id
OR (tm.depth=1 AND tm.broadcast)
ELSE e.id=$8 OR tm.root_event_id=$8 END",
)
.bind(community.as_uuid())
.bind(channel)
.bind(ELIGIBLE_KINDS.as_slice())
.bind(received_at)
.bind(created_at)
.bind(id)
.bind(super::projection::skew())
.bind(scope)
.fetch_one(conn)
.await?;
Ok(through.map_or(received_at, |later| later.max(received_at)))
}
pub(super) async fn valid_target(
conn: &mut PgConnection,
community: CommunityId,
@@ -203,6 +247,14 @@ pub(super) async fn apply(
if !msg.on_timeline {
return Ok(IntentOutcome::Blocked);
}
let through = display_through(
conn,
community,
target.channel_id,
(msg.id.as_slice(), msg.created_at, msg.received_at),
Some(root.as_slice()),
)
.await?;
sqlx::query(
"INSERT INTO personal_read_frontiers
(community_id, actor, channel_id, root_id, through_timestamp)
@@ -213,13 +265,21 @@ pub(super) async fn apply(
.bind(community.as_uuid())
.bind(actor)
.bind(target.channel_id)
.bind(msg.received_at)
.bind(through)
.execute(&mut *conn)
.await?;
} else {
if msg.id != root && msg.thread.as_ref() != Some(&root) {
return Ok(IntentOutcome::Blocked);
}
let through = display_through(
conn,
community,
target.channel_id,
(msg.id.as_slice(), msg.created_at, msg.received_at),
Some(root.as_slice()),
)
.await?;
// Reading a thread never joins it: only the actor's threads
// have a row to advance.
sqlx::query(
@@ -231,7 +291,7 @@ pub(super) async fn apply(
.bind(actor)
.bind(target.channel_id)
.bind(&root)
.bind(msg.received_at)
.bind(through)
.execute(&mut *conn)
.await?;
}
@@ -246,11 +306,21 @@ pub(super) async fn apply(
if !access(conn, community, actor, *channel_id).await? {
return Ok(IntentOutcome::Blocked);
}
// Only the anchor's arrival matters: ancestry cannot change which
// messages a whole-channel cut covers.
let Some(through) = anchor_received_at(conn, community, *channel_id, &id).await? else {
// Ancestry cannot change which messages a whole-channel cut
// covers: everything displayed at or before the anchor.
let Some((created_at, received_at)) =
anchor_times(conn, community, *channel_id, &id).await?
else {
return Ok(IntentOutcome::Blocked);
};
let through = display_through(
conn,
community,
*channel_id,
(&id, created_at, received_at),
None,
)
.await?;
sqlx::query(
"INSERT INTO personal_read_frontiers (community_id, actor, channel_id, root_id,
through_timestamp, threads_through_timestamp) VALUES ($1,$2,$3,''::bytea,$4,$4)
@@ -260,6 +330,42 @@ pub(super) async fn apply(
).bind(community.as_uuid()).bind(actor).bind(channel_id).bind(through)
.execute(&mut *conn).await?;
}
ReadIntent::Follow { target } | ReadIntent::Unfollow { target } => {
if target.root_id.is_none() {
return Ok(IntentOutcome::Invalid);
}
let Some(root) = valid_target(conn, community, actor, target).await? else {
return Ok(IntentOutcome::Blocked);
};
let following = matches!(intent, ReadIntent::Follow { .. });
// A new row starts caught up: at the thread's last reply, or the
// root itself before any reply. Following again also catches up,
// so replies that arrived while unfollowed stay muted. Unfollow
// keeps the row and its position, so later replies don't
// re-follow.
sqlx::query(
"INSERT INTO personal_read_frontiers
(community_id, actor, channel_id, root_id, through_timestamp, following)
SELECT $1, $2, $3, $4, COALESCE(
(SELECT max(last_reply_received_at) FROM thread_metadata
WHERE community_id=$1 AND channel_id=$3 AND event_id=$4 AND depth=0),
(SELECT max(received_at) FROM events
WHERE community_id=$1 AND channel_id=$3 AND id=$4)), $5
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE
SET following=excluded.following,
through_timestamp=GREATEST(personal_read_frontiers.through_timestamp,
excluded.through_timestamp)
WHERE excluded.following AND NOT personal_read_frontiers.following
OR NOT excluded.following",
)
.bind(community.as_uuid())
.bind(actor)
.bind(target.channel_id)
.bind(&root)
.bind(following)
.execute(&mut *conn)
.await?;
}
}
Ok(IntentOutcome::Applied)
}

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

@@ -412,12 +412,15 @@ async fn accessory_write_revocation_is_terminal_before_persistence() {
// Exercise the same per-item function the batch handler uses after admission.
let result = super::handlers::write_intent(&state, &headers, &principal, &intent).await;
assert_eq!(result, json!({"status":"blocked"}));
let persisted: i64 =
sqlx::query_scalar("SELECT count(*) FROM personal_read_accounts WHERE community_id=$1")
.bind(community.as_uuid())
.fetch_one(state.db.pool())
.await
.unwrap();
// Ingest gives the message's author an account; the actor gets none.
let persisted: i64 = sqlx::query_scalar(
"SELECT count(*) FROM personal_read_accounts WHERE community_id=$1 AND actor=$2",
)
.bind(community.as_uuid())
.bind(actor.public_key().to_bytes().as_slice())
.fetch_one(state.db.pool())
.await
.unwrap();
assert_eq!(persisted, 0, "denied intent must not persist");
}
@@ -473,7 +476,6 @@ async fn accessory_discovery_is_host_bound_and_opt_in() {
d["max_context_messages"],
buzz_db::personal_read::MAX_CONTEXT_MESSAGES
);
assert_eq!(d["max_thread_summaries"], 5);
assert_eq!(d["eligible_kinds"], json!([9, 40002, 45001, 45003]));
}
}
@@ -636,7 +638,7 @@ async fn signed_sidebar_deletion(deletion_kind: u16) {
.sign_with_keys(signer)
.unwrap();
let path = format!("/buzz/v1/me/sidebar?channel_ids={channel}");
for (event, key, count) in [(&message, &author, 1), (&deletion, signer, 0)] {
for (event, key, unread) in [(&message, &author, true), (&deletion, signer, false)] {
let body = serde_json::to_vec(event).unwrap();
let auth = proof(key, &host, "/events", "POST", Some(&body));
// /events has a different response contract from the accessory helper.
@@ -669,7 +671,7 @@ async fn signed_sidebar_deletion(deletion_kind: u16) {
assert_eq!(channels[0]["channel_id"], channel.to_string());
assert_eq!(
channels[0]["unread"],
json!(count),
json!(unread),
"kind {deletion_kind}, after kind {}: {sidebar}",
event.kind.as_u16()
);
@@ -678,12 +680,12 @@ async fn signed_sidebar_deletion(deletion_kind: u16) {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn accessory_signed_kind5_deletion_clears_another_readers_sidebar_count() {
async fn accessory_signed_kind5_deletion_clears_another_readers_sidebar_unread() {
signed_sidebar_deletion(5).await;
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn accessory_signed_kind9005_deletion_clears_another_readers_sidebar_count() {
async fn accessory_signed_kind9005_deletion_clears_another_readers_sidebar_unread() {
signed_sidebar_deletion(9005).await;
}

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

@@ -96,10 +96,8 @@ pub struct BuzzV1Descriptor {
pub max_contexts: usize,
/// Maximum message selectors across one context request.
pub max_context_messages: usize,
/// Maximum unread-thread summaries per sidebar channel row.
pub max_thread_summaries: usize,
/// Message kinds that count as unread and as latest activity. Clients
/// classify live arrivals with this set instead of keeping a copy.
/// Message kinds that count as unread. Clients classify live arrivals
/// with this set instead of keeping a copy.
pub eligible_kinds: [i32; 4],
}
@@ -389,7 +387,6 @@ 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,
};
info.buzz_v1 = Some(BuzzV1Descriptor {
version: 1,
@@ -398,7 +395,6 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st
max_intents: MAX_INTENTS,
max_contexts: MAX_CONTEXTS,
max_context_messages: MAX_CONTEXT_MESSAGES,
max_thread_summaries: MAX_THREAD_SUMMARIES,
eligible_kinds: ELIGIBLE_KINDS,
});
}

252
docs/buzz-v1-read-state.md
View File

@@ -11,9 +11,8 @@ On a known community host, NIP-11 (`GET /` with `Accept:
application/nostr+json`, or `GET /info`) includes `buzz_v1` only when enabled:
```json
{"buzz_v1":{"version":1,"base_path":"/buzz/v1","retention_seconds":2592000,
"max_channels":20,"max_intents":100,"max_contexts":20,"max_context_messages":100,
"max_thread_summaries":5,"eligible_kinds":[9,40002,45001,45003]}}
{"buzz_v1":{"version":1,"base_path":"/buzz/v1","max_channels":20,"max_intents":100,
"max_contexts":20,"max_context_messages":100,"eligible_kinds":[9,40002,45001,45003]}}
```
Use the requesting origin plus this relative prefix. Discovery is a configured
@@ -50,16 +49,10 @@ Unknown request fields are rejected. Never turn a transport failure into read.
## Sidebar
`GET /buzz/v1/me/sidebar?limit=20&cursor=<exclusive-channel-uuid>` returns
`account`, `channels`, and `next_cursor`. Omit the cursor on the first request.
Each channel includes identity/name/type, archived and hidden flags, `unread`,
`attention`, `latest_message_id` (the last eligible message to arrive: the
row's read anchor, see [Order](#order)), `latest_message_at` (the greatest
author time among eligible messages, in seconds: display activity, not
necessarily that message's own time; null exactly when the ID is null),
`latest_message_complete`, and `threads`. Only joined, nondeleted channels are
listed. Hidden/archived presentation remains client-owned. Each page has a
writer-consistent snapshot; separate pages do not share a snapshot, and an
unfinished traversal cannot prove channel removal.
`channels` and `next_cursor`. Omit the cursor on the first request. Only joined,
nondeleted channels are listed. Hidden/archived presentation remains
client-owned. Each page has a writer-consistent snapshot; separate pages do not
share a snapshot, and an unfinished traversal cannot prove channel removal.
`GET /buzz/v1/me/sidebar?channel_ids=<uuid>,<uuid>` refreshes 1–20 unique
channels in one snapshot, ordered by ID with `next_cursor: null`. It cannot be
@@ -67,78 +60,61 @@ combined with `limit` or `cursor`. A requested ID absent from the result was not
a joined, nondeleted, accessible sidebar row at that snapshot: remove its row.
Absence says nothing else about access to an open channel.
`threads` lists unread threads in the row, newest unread reply first:
Each row is one read scope for the channel timeline plus one per followed
thread with an unread reply:
```json
{"items":[{"root_id":"<64-hex>","unread":{"status":"exact","value":2},
"latest_reply_id":"<64-hex>","latest_reply_at":1700000000}],"complete":true}
{"channel_id":"<uuid>","name":"general","channel_type":"stream","archived":false,
"hidden":false,"unread":true,"mentions":2,"read_through_id":"<64-hex>",
"threads":[{"root_id":"<64-hex>","unread":true,"mentions":1,
"read_through_id":"<64-hex>","latest_id":"<64-hex>"}]}
```
Items are canonical roots with unread replies that count (see below), ordered
by `latest_reply_at` descending, then `root_id`; at most 5. `latest_reply_id` is
the last such reply to arrive (equal arrivals prefer the smaller ID), so a
thread `mark_through` at it reads every reply listed; `latest_reply_at` is its
author time. Replies that do not count are filtered out before the count, the
latest reply and the cap are chosen. `unread` uses the row's
definition; every counted reply is also attention, so items carry no separate
attention count. `complete=true` means the unread window was exhausted, no
evidence had unresolved ancestry, unusable tags or undecided membership, and no
thread was omitted; then item unread counts sum to the row's unread replies.
Otherwise the list is a cut of observed evidence and counts may be lower bounds
or unknown. No message bytes are included.
- `unread`: a message in that scope arrived after your read position. The
channel scope is the timeline only (top-level messages and broadcast
replies). A reply in a thread never makes the channel unread; it shows on
its thread row.
- `mentions`: exact, uncapped count of unread messages in that scope that tag
you with `p`. A thread mention counts on the thread row, not the channel.
- `read_through_id`: the last message in that scope, in display order, at or
before your read position, or null when there is none. Draw the "new"
divider below it and open the view there.
- `latest_id` (threads only): the last unread reply to arrive. Fetch it by ID
for a preview; marking the thread through it reads every listed reply.
Counts have exactly three representations:
`threads` lists every followed thread with an unread reply, newest reply
(author time) first, then by `root_id`. There is no cap. A badge or a threads
page is computed on the client from these rows. Clients receive message IDs
only; positions and timestamps stay relay-internal. No message bytes are
included.
```json
[{"status":"exact","value":0},{"status":"at_least","value":7},{"status":"unknown"}]
```
A message counts when it is nondeleted and of the advertised `eligible_kinds`,
so an edit, reaction or diff (40008) never makes a scope unread. Classify live
arrivals with the advertised set, not a client copy. Your own messages need
no rule: posting marks you read through what you post (see
[Ingest](#ingest-posting-and-following)).
Only exact zero proves absence. Unknown has no numeric value. A message is
eligible when it is non-own, nondeleted, of the advertised `eligible_kinds` and
inside the horizon. The same kinds alone define latest activity, so an edit,
reaction or diff (40008) neither makes a channel unread nor moves it. Classify
live arrivals with the advertised set, not a client copy.
## Ingest: posting and following
An eligible message beyond the matching context frontier counts as unread for
the first reason that holds:
The relay updates read state when it stores an eligible message:
| `reason` | Holds when |
|---|---|
| `direct` | its channel is a DM |
| `mention` | it tags the actor with `p` |
| `conversation` | it is a reply, and the actor wrote its direct parent or has a reply to that same parent, in that channel |
| `broadcast` | it carries `broadcast=1` |
| null | it is top-level |
- **Posting marks read.** The author's position moves through their own
message in each scope it shows in: the timeline for a top-level message or a
broadcast reply, the thread for a reply.
- **Following is a relay-side thread row.** You follow a thread when you reply
to it, when someone replies to your top-level message, when a message
mentions you (the thread it is in, or the thread rooted at it), in a DM when
any member replies, or with a `follow` intent. A new follow starts just
before the message that caused it, so that message is unread.
- **Unfollow is sticky.** `unfollow` keeps the row with `following=false`, so
later replies and mentions do not follow you again. Your own reply, or
`follow`, does. Whether a new mention should re-follow is an open product
question; today it does not.
A reply with no reason does not count: it is not unread and appears in no
thread list. `unread` counts the messages that count; `attention` is the subset
with a reason. Conversation membership uses only the actor's live eligible
messages (a deleted parent proves nothing; a surviving reply to it still does),
looks at the direct parent only (owning the root or replying elsewhere in the
thread proves nothing), and is independent of read progress and retention. A
reply whose membership is undecided is left out, and the row's counts become
lower bounds or unknown. The sidebar counts a broadcast reply without asking
which of the two reasons applies. This is not Desktop notification policy:
follows and mutes do not affect these counts.
The unread horizon defaults to 30 days (`BUZZ_V1_RETENTION_SECONDS`) and is
measured in author time (`created_at`): a message counts while its author time
is at or after `account.cutoff_ms`. It filters unread/attention, not latest
activity, event storage or frontier state. Frontiers use a different clock,
relay arrival (see [Order](#order)). Three consequences:
- A message accepted late with an author time beyond the horizon (an import, a
backfill, a long-offline sender) is excluded under the current horizon,
however recently the relay accepted it. It can still be latest when the
horizon holds no message.
- Unread expires at author time plus the horizon, so a future-dated author
time extends how long a message counts.
- A context's unread set is two tests, not one range: arrived after the
frontier, and author time at or after the cutoff.
A later configuration expansion can change counts without having lost progress.
Latest activity is independent of actor and frontiers. A null latest ID proves
an empty eligible history only when `latest_message_complete=true`.
An account starts caught up at its first read intent (`started_at`): every
position is floored there, so history from before then never counts, and
nothing counts before the first intent. Joining a channel starts that channel
caught up.
## Explicit contexts
@@ -157,8 +133,8 @@ and one `contexts` entry per request entry, in order. Context status is
frontier includes any whole-channel cut. Message status is `read`, `not_counted`,
`unread` (with `reason`), `unknown`, or `unavailable`. Wrong-context, missing
and forbidden selectors share unavailable. Status is decided in this order:
ancestry and context; eligibility (`not_counted` for own, deleted, other kinds
and outside the horizon); the frontier (`read`, with no membership lookup); then
ancestry and context; eligibility (`not_counted` for own, deleted, other kinds,
or before the account started); the frontier (`read`, with no membership lookup); then
the reason. A reply past the frontier with no reason is `not_counted` when it is
proven outside the actor's conversations and `unknown` when membership is
undecided. A broadcast reply whose membership is undecided is `unread` with
@@ -173,7 +149,9 @@ Conversation bytes must still come from the existing Nostr path.
{"intents":[
{"type":"mark_through","target":{"channel_id":"<uuid>"},"message_id":"<64-hex-event>"},
{"type":"mark_through","target":{"channel_id":"<uuid>","root_id":"<64-hex-root>"},"message_id":"<64-hex-event>"},
{"type":"mark_channel_read","channel_id":"<uuid>","message_id":"<64-hex-event>"}
{"type":"mark_channel_read","channel_id":"<uuid>","message_id":"<64-hex-event>"},
{"type":"follow","target":{"channel_id":"<uuid>","root_id":"<64-hex-root>"}},
{"type":"unfollow","target":{"channel_id":"<uuid>","root_id":"<64-hex-root>"}}
]}
```
@@ -184,89 +162,85 @@ later failures. `projection_status` is `not_requested`; a successful write does
not assert a client has refreshed. Retry the same operands, never substitute
latest. Keep pending intent durably on the client until its outcome is resolved.
A mark-through validates a fixed message and advances its context's monotone
frontier to that message's relay arrival (see [Order](#order)). Channel and
thread frontiers never inherit in either direction. Opening a view is not
itself a reading action; client dwell/focus policy determines when to send an
actual observed anchor. Old or deleted valid anchors may advance a frontier.
A mark-through validates a fixed message and reads everything displayed at or
before it in its scope (see [Order](#order)). Channel and thread positions never
inherit in either direction. Opening a view is not itself a reading action;
client dwell/focus policy determines when to send an actual observed anchor.
Old or deleted valid anchors may advance a position. Marking a thread you do
not follow changes nothing.
`mark_channel_read` is the one whole-channel cut: it advances the channel
timeline and every thread in that channel, including unlisted ones, through the
anchor's arrival. The anchor must be an accessible eligible-kind message in
the channel, top-level or reply, deleted or not; ancestry is not checked. A
reply is read at or below the greater of its thread frontier and this cut. An
anchor that no longer exists is `blocked`. `latest_message_id` is the anchor
that reads the whole row. A null ID with `latest_message_complete=false` does
not prove empty history; it only leaves the client without an anchor. Thread
marks and channel `mark_through` never set the cut.
`mark_channel_read` is the one whole-channel cut: it reads the channel
timeline and every thread in that channel, including unlisted ones, through
everything in the channel displayed at or before the anchor. The anchor must be
an accessible eligible-kind message in the channel, top-level or reply, deleted
or not; ancestry is not checked. A reply is read at or below the greater of its
thread position and this cut. An anchor that no longer exists is `blocked`.
Thread marks and channel `mark_through` never set the cut.
`follow` and `unfollow` take a thread target; one without `root_id` is
`invalid`. Following starts caught up at the thread's last reply, including
when you follow again after unfollowing.
### Order
A frontier is the relay arrival time (`events.received_at`) of the message a
context was read through, never its author time, which the sender chooses and
the relay accepts up to 15 minutes either way. Everything that arrived at or
before the anchor is read, whatever its author time. A message that arrives
later is unread even when backdated, and a future-dated anchor reads nothing
that arrives after it.
A position is a relay arrival time (`events.received_at`), never author time,
which the sender chooses and the relay accepts up to 15 minutes either way. A
message that arrives after your position is unread even when backdated, and a
future-dated message cannot read anything that arrives after it.
The order is the relay's and is not exposed: no response carries a frontier,
and no field lets a client compute what a mark will cover. Send the anchors the
user actually saw, let the relay take the greatest, and ask a context which
messages are read. Do not compare author times, IDs or local receipt order to
drop one pending anchor in favor of another.
Clients see display order (author time, then ID) and name only message IDs. A
mark therefore reads everything displayed at or before its anchor in that
scope: the position becomes the last arrival among them. A late arrival shown
above the newest message is read when you mark that message, so marking the
bottom of a view always clears its badge. The cost, only when a late arrival
sits above the anchor, is that a message displayed below the anchor but arrived
before the late one is read too. With no late arrival, a mark reads exactly
what arrived by the anchor, and a message shown below it stays unread.
Positions are never exposed. `read_through_id` is the last message displayed at
or before the position; in the rare late-arrival case an unread message can sit
above that divider, while the row's `unread` stays correct.
Arrival is the accepting relay process's clock, read just before the insert, at
microsecond resolution. Three limits follow, none of which strands a badge:
microsecond resolution. It is not commit order, relay processes with different
clocks can stamp out of true order by their skew, and messages with the
identical stamp are read together.
- It is not commit order. An insert that commits after a later-stamped message
was already marked read lands read.
- Relay processes with different clocks can stamp out of true order by their
skew.
- Messages with the identical stamp are read together.
`latest_message_id` is the last message to arrive among those the unread count
examined: the 4,096 most recent events by author time inside the horizon. So
marking through it reads everything counted. When the horizon holds no message,
it is the last to arrive among the channel's 256 most recent events.
There is no import of earlier client read state: an account starts with no
frontiers, and the horizon bounds what that can show as unread. Manual unread
remains device-local.
There is no import of earlier client read state. Manual unread remains
device-local.
## Bounds and deployment
- 20 sidebar rows, 100 intents, 20 contexts / 100 selectors per request.
- 64 KiB write body; 16 KiB context URL; 1 MiB serialized API response.
- 4096 raw events per channel inside the horizon plus one exhaustion sentinel,
before eligibility.
- Latest activity probes 256 events plus a sentinel; long ineligible tails may
leave latest incomplete even when unread is exact.
- Tag documents over 8192 bytes or malformed relevant tags yield uncertainty.
Compact boolean facts cross the database boundary, never raw tag payloads.
- Conversation membership: at most 1024 unique parents per request, with a
500 ms savepoint budget. The lookup is exact, so its work grows with the
replies under each parent. Past either bound a reply is undecided, never
absent: counts become lower bounds or unknown and a context reports `unknown`.
- `unread` is an existence probe that stops at the first message past the
position, so a caught-up reader examines almost nothing. Counting is forward
from the position: work grows with what is unread, not with history.
- Mention counts walk only your mentions in that scope past the position
(`idx_event_mentions_scope_received`).
- A followed thread with nothing new is skipped by comparing the root's
`thread_metadata.last_reply_received_at` with the thread position, without
reading replies.
- DB statement/lock deadlines and HTTP read deadlines bound work; writes use a
shared eight-second intent-processing deadline after admission. Limits are
containment, not a production capacity claim.
Apply migration 0056 (or the equivalent desired schema). It creates two empty
private tables and no index on `events`: both sidebar scans are served by the
existing `idx_events_community_channel_created`. No new per-message ingest write
path or stored unread counters are introduced.
Apply migration 0056 (or the equivalent desired schema). It creates two
private tables, adds `thread_metadata.last_reply_received_at` and
`event_mentions.received_at`/`root_id` (NULL for rows from before it, which no
position can reach), and one index on `event_mentions`. Ingest writes the
author's own position and follow rows in the same transaction as the message.
Use existing HTTP route/status/latency metrics for `/buzz/v1/me/sidebar` and
`/buzz/v1/me/read-state`, plus database pool/statement metrics. Inspect exact /
lower-bound / unknown proportions in controlled acceptance captures; no payload,
actor, channel or frontier values should become metric labels. The measured
local seed is not a DAU/concurrency or p95/p99 production acceptance result.
`/buzz/v1/me/read-state`, plus database pool/statement metrics. No payload,
actor, channel or position values should become metric labels.
## Compatibility and extension rules
Within `/buzz/v1`, clients must ignore unknown response object fields. Existing
required fields, status variants and their meanings remain stable; additive
fields do not authorize silently changing `attention` or frontier semantics.
fields do not authorize silently changing `unread`, `mentions` or position
semantics.
Breaking changes require an explicitly negotiated contract or a new API version.
Requests remain strict: send new parameters or intent types only after the relay
advertises the corresponding capability. Missing optional data means unsupported
@@ -281,8 +255,8 @@ additional capabilities.
These typed relational frontiers are signer-private application state, **not
self-encrypted**. Database operators can see reading progress; ordinary Nostr
queries, search and moderator interfaces do not expose it. No public receipts
are emitted. Storage grows by touched contexts, not observed messages, and has
no fixed context-count ceiling.
are emitted. Storage grows by touched channels and followed threads, not observed
messages, and has no fixed ceiling.
Leaving/rejoining does not erase progress; revoked access hides it. Soft-deleted
channels are inaccessible, while hard channel deletion cascades their frontiers.

18
migrations/0056_personal_read_state.sql
View File

@@ -2,7 +2,9 @@
-- A frontier is the relay arrival time (events.received_at) of the message a
-- context was read through, never signed event time, which the sender chooses.
-- An empty root_id covers only the channel timeline; a root-specific frontier
-- covers that thread, without inheritance.
-- covers that thread, without inheritance. A thread row exists for each
-- thread the actor follows or has unfollowed. Ingest moves the author's own
-- frontiers through each message they post.
-- threads_through_timestamp is the only cross-context cut: an explicit
-- whole-channel read that also covers every thread in that channel.
-- started_at is the actor's first read intent and the floor of every
@@ -24,6 +26,9 @@ CREATE TABLE personal_read_frontiers (
-- Whole-channel cut covering every thread; channel rows only.
threads_through_timestamp TIMESTAMPTZ
CHECK (threads_through_timestamp IS NULL OR root_id = ''::bytea),
-- Thread rows only: false after unfollow, kept so later replies don't
-- re-follow. Replying or a follow intent sets it again.
following BOOLEAN NOT NULL DEFAULT true CHECK (following OR root_id <> ''::bytea),
PRIMARY KEY (community_id, actor, channel_id, root_id),
FOREIGN KEY (community_id, actor)
REFERENCES personal_read_accounts (community_id, actor) ON DELETE CASCADE,
@@ -36,5 +41,16 @@ CREATE TABLE personal_read_frontiers (
-- with nothing new since the reader's position without reading replies.
ALTER TABLE thread_metadata ADD COLUMN last_reply_received_at TIMESTAMPTZ;
-- Arrival time and read scope of each mention, so the sidebar counts a
-- reader's unread mentions in one channel timeline or thread from its
-- position. root_id is the thread a reply belongs to, NULL when the mention
-- shows on the channel timeline (a top-level message or broadcast reply).
-- Existing rows stay NULL: every position is floored at its account's
-- started_at, set by a read intent after this migration, so no earlier
-- mention can be unread.
ALTER TABLE event_mentions ADD COLUMN received_at TIMESTAMPTZ, ADD COLUMN root_id BYTEA;
CREATE INDEX idx_event_mentions_scope_received
ON event_mentions (community_id, pubkey_hex, channel_id, root_id, received_at);
SELECT attach_community_write_fence('personal_read_accounts');
SELECT attach_community_write_fence('personal_read_frontiers');

9
schema/schema.sql
View File

@@ -216,6 +216,9 @@ CREATE TABLE personal_read_frontiers (
-- Whole-channel cut covering every thread; channel rows only.
threads_through_timestamp TIMESTAMPTZ
CHECK (threads_through_timestamp IS NULL OR root_id = ''::bytea),
-- Thread rows only: false after unfollow, kept so later replies don't
-- re-follow. Replying or a follow intent sets it again.
following BOOLEAN NOT NULL DEFAULT true CHECK (following OR root_id <> ''::bytea),
PRIMARY KEY (community_id, actor, channel_id, root_id),
FOREIGN KEY (community_id, actor)
REFERENCES personal_read_accounts (community_id, actor) ON DELETE CASCADE,
@@ -325,6 +328,10 @@ CREATE TABLE event_mentions (
event_created_at TIMESTAMPTZ NOT NULL,
channel_id UUID,
event_kind INT,
-- Arrival and read scope for personal read state: root_id is a reply's
-- thread, NULL on the channel timeline. NULL received_at predates it.
received_at TIMESTAMPTZ,
root_id BYTEA,
PRIMARY KEY (community_id, pubkey_hex, event_id)
);
@@ -332,6 +339,8 @@ CREATE INDEX idx_event_mentions_pubkey_created
ON event_mentions (community_id, pubkey_hex, event_created_at DESC);
CREATE INDEX idx_event_mentions_pubkey_kind_created
ON event_mentions (community_id, pubkey_hex, event_kind, event_created_at DESC);
CREATE INDEX idx_event_mentions_scope_received
ON event_mentions (community_id, pubkey_hex, channel_id, root_id, received_at);
-- ── Subscriptions ─────────────────────────────────────────────────────────────
-- Conformance: "Mesh, agents, ACP/MCP, and CLI" (persisted subscriptions).