jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-28 04:51:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(read-state): follow intents, threads view, arrival-ordered access paths

Browse Source 
Extend the merged /buzz/v1 read-state wire with the minimum the follow
model needs:

- `follow` / `unfollow` intents on a thread target. An unfollow sticks
  until the actor posts in the thread or is mentioned there again.
- `GET /me/threads`: followed threads in joined channels, newest reply
  arrival first, cursor-paged, at most 50 per page.
- Thread rows count `mentions` as direct mentions only.
- Broadcast replies count on the channel timeline and are valid channel
  anchors.
- NIP-11 advertises `capabilities: ["follow","threads"]` and `max_threads`.

Migration 0061 adds `events.read_timeline`, `thread_metadata.received_at`
and `personal_read_frontiers.unfollowed`, plus arrival-ordered partial
indexes so the sidebar and threads view probe an index per channel and
per follow instead of scanning a bounded window. No backfill.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Bradley Axen <baxen@squareup.com>
This commit is contained in:
Bradley Axen
2026-10-11 16:14:33 -07:00
parent 419a53029a
commit 4da2b9f4e4
23 changed files with 1193 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
crates/buzz-db/src/runtime/migration.rs
View File

@@ -705,7 +705,7 @@ mod postgres_tests {
let mut migrations: Vec<_> = MIGRATOR.iter().collect();
migrations.sort_by_key(|migration| migration.version);
assert_eq!(migrations.len(), 60);
assert_eq!(migrations.len(), 61);
assert_eq!(migrations[58].version, 59);
assert!(migrations[58].sql.as_str().contains("buzz.push_enabled"));
assert_eq!(migrations[59].version, 60);
@@ -713,6 +713,11 @@ mod postgres_tests {
.sql
.as_str()
.contains("ALTER TABLE push_leases DROP COLUMN app_profile"));
assert_eq!(migrations[60].version, 61);
assert!(migrations[60]
.sql
.as_str()
.contains("idx_events_read_timeline"));
assert_eq!(migrations[55].version, 56);
assert!(migrations[55]
.sql
@@ -2253,6 +2258,16 @@ mod postgres_tests {
.sql
.as_str();
assert!(through_message.contains("DROP COLUMN threads_through_timestamp"));
// 0061 adds the sticky thread unfollow.
let unfollowed_column = "unfollowed boolean not null default false \
check (not unfollowed or root_id <> ''::bytea), ";
assert!(MIGRATOR
.iter()
.find(|m| m.version == 61)
.expect("read-state access path migration")
.sql
.as_str()
.contains("$d$BOOLEAN NOT NULL DEFAULT false CHECK (NOT unfollowed OR root_id <> ''::bytea)$d$"));
for (table, definition) in personal.tables {
let in_schema = schema.tables.get(&table).map(|schema_definition| {
if table == "personal_read_accounts" {
@@ -2262,7 +2277,11 @@ mod postgres_tests {
}
});
let definition = if table == "personal_read_frontiers" {
definition.replacen(threads_through_column, through_message_column, 1)
definition.replacen(
threads_through_column,
&format!("{through_message_column}{unfollowed_column}"),
1,
)
} else {
definition
};

17
crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs
View File

@@ -270,9 +270,11 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer()
.await
.unwrap();
// Also bind the public migrator: the advisory schema/destruction lock,
// ledger write and post-migration catalog checks must work with ingestion.
let production_result = migration::run_migrations(&pool).await;
// Also bind the migrator: the advisory schema/destruction lock, ledger
// write and post-migration catalog checks must work with ingestion. 0061
// adds a column to thread_metadata, which takes a brief exclusive lock
// bounded by its lock_timeout, so it runs after the writer below.
let production_result = migration::run_migrations_through(&pool, 60).await;
sqlx::query(insert)
.bind(community)
.bind(channel)
@@ -280,10 +282,6 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer()
.execute(&mut *witness)
.await
.unwrap();
let version: i64 = sqlx::query_scalar("SELECT max(version) FROM _sqlx_migrations")
.fetch_one(&mut *witness)
.await
.unwrap();
let final_oid: i64 =
sqlx::query_scalar("SELECT 'idx_thread_metadata_window'::regclass::oid::bigint")
.fetch_one(&mut *witness)
@@ -297,6 +295,11 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer()
.unwrap();
drop(witness);
writer.rollback().await.unwrap();
migration::run_migrations(&pool).await.unwrap();
let version: i64 = sqlx::query_scalar("SELECT max(version) FROM _sqlx_migrations")
.fetch_one(&pool)
.await
.unwrap();
drop_scratch_db(&admin, pool, &name).await;
assert!(

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

@@ -447,15 +447,9 @@ pub async fn insert_event_in_transaction(
) -> Result<(StoredEvent, bool)> {
let result = insert_event_on(tx, event, channel_id).await?;
if result.1 {
if let Some(channel) = channel_id {
crate::personal_read::record_message(
tx,
event,
channel,
crate::personal_read::Place::TopLevel,
result.0.received_at,
)
.await?;
if let Some((channel, place)) = crate::personal_read::place(event, channel_id, None) {
crate::personal_read::record_message(tx, event, channel, place, result.0.received_at)
.await?;
}
crate::operator_listener::enqueue_mentions_in_transaction(tx, event).await?;
}
@@ -492,8 +486,8 @@ async fn insert_event_on(
let not_before = extract_not_before(event);
let result = sqlx::query(
r#"
INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before, read_timeline)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
ON CONFLICT DO NOTHING
"#,
)
@@ -509,6 +503,7 @@ async fn insert_event_on(
.bind(channel_id)
.bind(d_tag.as_deref())
.bind(not_before)
.bind(crate::personal_read::place(event, channel_id, None).is_some_and(|(_, p)| p.on_timeline()))
.execute(tx.conn())
.await?;
@@ -1740,11 +1735,22 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
let received_at = Utc::now();
let d_tag = extract_d_tag(event);
let not_before = extract_not_before(event);
let place = match &thread_meta {
Some(meta) => crate::personal_read::place(
event,
Some(meta.channel_id),
meta.parent_event_id.map(|parent| match meta.root_event_id {
Some(root) => (root, meta.root_event_created_at, meta.broadcast),
None => (parent, meta.parent_event_created_at, meta.broadcast),
}),
),
None => crate::personal_read::place(event, channel_id, None),
};
let result = sqlx::query(
r#"
INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id, d_tag, not_before, read_timeline)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
ON CONFLICT DO NOTHING
"#,
)
@@ -1760,6 +1766,7 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
.bind(channel_id)
.bind(d_tag.as_deref())
.bind(not_before)
.bind(place.is_some_and(|(_, p)| p.on_timeline()))
.execute(tx.conn())
.await?;
@@ -1774,6 +1781,12 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
)
.await?;
// Before the thread counters below: their root row is shared by every
// reply, so its lock must not be held across these writes.
if let Some((channel, place)) = place {
crate::personal_read::record_message(tx, event, channel, place, received_at).await?;
}
if let Some(ref meta) = thread_meta {
let broadcast_val: bool = meta.broadcast;
@@ -1783,8 +1796,8 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
(community_id, event_created_at, event_id, channel_id,
parent_event_id, parent_event_created_at,
root_event_id, root_event_created_at,
depth, broadcast)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
depth, broadcast, received_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
ON CONFLICT DO NOTHING
"#,
)
@@ -1798,6 +1811,8 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
.bind(meta.root_event_created_at)
.bind(meta.depth)
.bind(broadcast_val)
// An eligible reply's arrival, for the read-state thread index.
.bind(place.and(meta.parent_event_id).map(|_| received_at))
.execute(tx.conn())
.await?;
@@ -1882,22 +1897,6 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
}
}
let place = match &thread_meta {
Some(meta) => Some((
meta.channel_id,
match meta.parent_event_id {
Some(parent) => crate::personal_read::Place::Reply {
root: meta.root_event_id.unwrap_or(parent),
},
// Workflow messages store a depth-0 metadata row.
None => crate::personal_read::Place::TopLevel,
},
)),
None => channel_id.map(|channel| (channel, crate::personal_read::Place::TopLevel)),
};
if let Some((channel, place)) = place {
crate::personal_read::record_message(tx, event, channel, place, received_at).await?;
}
crate::operator_listener::enqueue_mentions_in_transaction(tx, event).await?;
}

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

@@ -1,7 +1,7 @@
//! 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::{fixture, insert_reply},
postgres_tests::{fixture, insert_reply, sync_reply_arrivals},
*,
};
use crate::Db;
@@ -23,6 +23,7 @@ async fn arrive(pool: &PgPool, community: CommunityId, event: &nostr::Event, arr
.unwrap()
.rows_affected();
assert_eq!(updated, 1, "arrival must land on exactly one stored event");
sync_reply_arrivals(pool).await;
}
/// A message authored at `authored` that arrives at `arrived`. It mentions the
@@ -254,10 +255,9 @@ async fn mark_thread_read_with_the_sidebar_anchor_clears_a_late_reply() {
assert!(row.threads.is_empty());
}
/// The latest probe reads the newest `MAX_TIMELINE_SCAN` events by author
/// time, of any kind. A late arrival with an older author time than the
/// uncounted events after it must still be the sidebar anchor, or Mark as read
/// with that anchor leaves it unread.
/// A late arrival with an older author time than the uncounted events after
/// it must still be the sidebar anchor, or Mark as read with that anchor
/// leaves it unread.
async fn reactions(
pool: &PgPool,
db: &Db,
@@ -310,37 +310,6 @@ async fn mark_as_read_clears_a_late_arrival_behind_reactions() {
assert_eq!(counts(&db, community, &actor).await, (false, 0));
}
/// Past the budget the probe finds no anchor, and nothing counts: a badge
/// with no latest to mark could never be cleared.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn a_message_behind_a_full_budget_of_reactions_is_not_a_badge_without_an_anchor() {
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);
post(&db, &pool, community, channel, &actor, now - 600, now - 10).await;
reactions(
&pool,
&db,
community,
channel,
MAX_TIMELINE_SCAN as u64,
now,
)
.await;
let row = sidebar(&db, community, &actor).await;
assert_eq!(row.latest_id, None);
assert_eq!((row.unread, row.mentions), (false, 0));
}
/// Store `event` as having arrived at exactly `seconds` plus `micros`. Built
/// from integers, never a float, so microsecond order cannot hinge on rounding.
async fn arrive_exact(
@@ -360,6 +329,7 @@ async fn arrive_exact(
.unwrap()
.rows_affected();
assert_eq!(updated, 1, "arrival must land on exactly one stored event");
sync_reply_arrivals(pool).await;
let stored: chrono::DateTime<chrono::Utc> =
sqlx::query_scalar("SELECT received_at FROM events WHERE community_id=$1 AND id=$2")
.bind(community.as_uuid())

116
crates/buzz-db/src/store/personal_read/follows.rs
View File

@@ -1,5 +1,5 @@
//! Read state that ingest writes. A thread frontier row exists only for a
//! thread its actor follows; only this module creates one.
//! Read state that ingest writes. A thread frontier row exists for a thread
//! its actor follows or has unfollowed; ingest and follow intents create one.
use super::model::ELIGIBLE_KINDS;
use crate::{AdmittedTx, Result};
use chrono::{DateTime, Utc};
@@ -11,19 +11,75 @@ use uuid::Uuid;
pub(crate) enum Place<'a> {
/// A top-level timeline message.
TopLevel,
/// A reply in the thread rooted at `root`.
Reply { root: &'a [u8] },
/// A reply in the thread rooted at `root`, authored at `root_created_at`;
/// a broadcast reply is also on the channel timeline.
Reply {
root: &'a [u8],
root_created_at: DateTime<Utc>,
broadcast: bool,
},
}
impl Place<'_> {
/// Whether the message counts on its channel timeline (`read_timeline`).
pub(crate) fn on_timeline(&self) -> bool {
matches!(
self,
Place::TopLevel
| Place::Reply {
broadcast: true,
..
}
)
}
}
/// A reply's recorded ancestry: its root, the root's author time, and whether
/// it is a broadcast.
pub(crate) type Ancestry<'a> = (&'a [u8], Option<DateTime<Utc>>, bool);
/// Where an eligible channel message sits, from its recorded ancestry. None
/// for other events, and for a reply whose ancestry was never (fully)
/// recorded: it is on neither read path.
pub(crate) fn place<'a>(
event: &Event,
channel: Option<Uuid>,
reply: Option<Ancestry<'a>>,
) -> Option<(Uuid, Place<'a>)> {
let channel = channel?;
if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) {
return None;
}
match reply {
Some((root, root_created_at, broadcast)) => Some((
channel,
Place::Reply {
root,
root_created_at: root_created_at?,
broadcast,
},
)),
None if buzz_core::nip10::parse_thread_markers(&event.tags)
.resolve()
.is_some() =>
{
None
}
None => Some((channel, Place::TopLevel)),
}
}
/// Record what a newly stored eligible message changes, in its transaction:
/// - its author has read through it: the timeline for a top-level message,
/// otherwise the thread, which the author then follows;
/// otherwise the thread, which the author then follows (and the timeline
/// too for a broadcast reply);
/// - each channel member it mentions follows its thread (for a top-level
/// message, the thread rooted at it);
/// - a reply makes the root's author, and in a DM every member, follow too.
///
/// Everyone but the author starts just before the message, so it is unread
/// to them. A follower keeps an existing position.
/// to them. A follower keeps an existing position. An unfollow sticks until
/// its actor posts in the thread or is mentioned there again.
pub(crate) async fn record_message(
tx: &mut AdmittedTx,
event: &Event,
@@ -41,40 +97,49 @@ pub(crate) async fn record_message(
.public_keys()
.map(|key| key.to_bytes().to_vec())
.collect();
let (read, thread, reply) = match place {
Place::TopLevel => (&[][..], id, false),
Place::Reply { root } => (root, root, true),
let (read, thread, reply, root_created_at, broadcast) = match place {
Place::TopLevel => (&[][..], id, false, None, false),
Place::Reply {
root,
root_created_at,
broadcast,
} => (root, root, true, Some(root_created_at), broadcast),
};
// Accounts, then frontiers, each in key order, so concurrent messages that
// involve the same people cannot deadlock: the frontier insert joins the
// finished account insert. Foreign-key checks run at the end of the
// statement, after the account rows exist.
sqlx::query(
"WITH involved (actor, root_id, through, message) AS (
SELECT $4::bytea, $5::bytea, $6::timestamptz, $3::bytea
"WITH involved (actor, root_id, through, message, refollow) AS (
SELECT $4::bytea, $5::bytea, $6::timestamptz, $3::bytea, true
UNION ALL
SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL FROM channel_members cm
SELECT $4, ''::bytea, $6, $3, true WHERE $11
UNION ALL
SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL, true FROM channel_members cm
WHERE cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL
AND cm.pubkey=ANY($8) AND cm.pubkey<>$4
UNION ALL
SELECT e.pubkey, $7, $6-interval '1 microsecond', NULL FROM events e
WHERE $9 AND e.community_id=$1 AND e.channel_id=$2 AND e.id=$7 AND e.pubkey<>$4
SELECT e.pubkey, $7, $6-interval '1 microsecond', NULL, false FROM events e
WHERE $9 AND e.community_id=$1 AND e.created_at=$10 AND e.id=$7
AND e.channel_id=$2 AND e.pubkey<>$4
UNION ALL
SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL
SELECT cm.pubkey, $7, $6-interval '1 microsecond', NULL, false
FROM channel_members cm JOIN channels c
ON c.community_id=cm.community_id AND c.id=cm.channel_id
WHERE $9 AND cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL
AND c.channel_type='dm' AND cm.pubkey<>$4
), rows AS (
-- An existing follower keeps their position, so is not touched.
-- Existence probes are LIMIT 1 laterals: always one index lookup.
-- An existing follower keeps their position, so is not touched;
-- nor is an unfollow, unless this message re-follows (the author
-- posting, or a mention). Existence probes are LIMIT 1 laterals:
-- always one index lookup.
SELECT DISTINCT ON (i.actor, i.root_id) i.* FROM involved i
LEFT JOIN LATERAL (
SELECT true AS found FROM personal_read_frontiers f WHERE f.community_id=$1
SELECT f.unfollowed FROM personal_read_frontiers f WHERE f.community_id=$1
AND f.actor=i.actor AND f.channel_id=$2 AND f.root_id=i.root_id LIMIT 1
) f ON true
WHERE i.message IS NOT NULL OR f.found IS NULL
ORDER BY i.actor, i.root_id
WHERE i.message IS NOT NULL OR f.unfollowed IS NULL OR (i.refollow AND f.unfollowed)
ORDER BY i.actor, i.root_id, i.message IS NULL, NOT i.refollow
), accounts AS (
INSERT INTO personal_read_accounts (community_id, actor)
SELECT $1, r.actor FROM (SELECT DISTINCT actor FROM rows) r
@@ -85,6 +150,9 @@ pub(crate) async fn record_message(
WHERE a.found IS NULL ORDER BY r.actor
ON CONFLICT DO NOTHING RETURNING 1
)
-- A re-follow starts just before this message, like a new follow,
-- unless the unfollow is later. A row that a concurrent message
-- created is only advanced when it carries an anchor.
INSERT INTO personal_read_frontiers
(community_id, actor, channel_id, root_id, through_timestamp, through_message_id)
SELECT $1, actor, $2, root_id, through, message
@@ -95,8 +163,10 @@ pub(crate) async fn record_message(
excluded.through_timestamp),
through_message_id=CASE
WHEN excluded.through_timestamp > personal_read_frontiers.through_timestamp
THEN excluded.through_message_id ELSE personal_read_frontiers.through_message_id END
WHERE excluded.through_message_id IS NOT NULL",
THEN excluded.through_message_id ELSE personal_read_frontiers.through_message_id END,
unfollowed=false
WHERE excluded.through_message_id IS NOT NULL OR (personal_read_frontiers.unfollowed
AND excluded.through_timestamp > personal_read_frontiers.through_timestamp)",
)
.bind(community.as_uuid())
.bind(channel)
@@ -107,6 +177,8 @@ pub(crate) async fn record_message(
.bind(thread)
.bind(&mentioned)
.bind(reply)
.bind(root_created_at)
.bind(broadcast)
.execute(tx.conn())
.await?;
Ok(())

294
crates/buzz-db/src/store/personal_read/follows_postgres_tests.rs
View File

@@ -95,7 +95,7 @@ async fn follow(
.unwrap()
}
/// (root, unread replies) for each listed thread.
/// (root, mentions) for each listed thread.
fn threads(row: &ChannelReadSummary) -> Vec<(String, u32)> {
row.threads
.iter()
@@ -129,7 +129,8 @@ async fn replying_follows_the_thread_and_reads_through_the_reply() {
let other = reply(&db, community, channel, &root, &Keys::generate(), vec![]).await;
let summary = row(&db, community, channel, &actor).await;
assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]);
// Unread, but not directed at the actor.
assert_eq!(threads(&summary), [(root.id.to_hex(), 0)]);
assert_eq!(summary.threads[0].latest_id, other.id.to_hex());
assert_eq!(summary.threads[0].read_through_id, Some(own.id.to_hex()));
}
@@ -148,9 +149,9 @@ async fn a_mention_follows_a_member_from_just_before_it() {
assert_eq!(follow(&pool, community, &bee, &root).await, Some(None));
assert_eq!(follow(&pool, community, &outsider, &root).await, None);
// The mention and what came after it; not the reply before it.
// The mention and what came after it are unread; not the reply before it.
let summary = row(&db, community, channel, &bee).await;
assert_eq!(threads(&summary), [(root.id.to_hex(), 2)]);
assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]);
assert_ne!(summary.threads[0].latest_id, earlier.id.to_hex());
assert_ne!(summary.threads[0].latest_id, mentioning.id.to_hex());
}
@@ -175,7 +176,7 @@ async fn a_later_mention_keeps_the_followers_position() {
Some(Some(own.id.to_hex()))
);
let summary = row(&db, community, channel, &actor).await;
assert_eq!(threads(&summary), [(root.id.to_hex(), 2)]);
assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]);
}
#[tokio::test]
@@ -188,7 +189,7 @@ async fn the_root_author_follows_on_the_first_reply() {
reply(&db, community, channel, &root, &Keys::generate(), vec![]).await;
assert_eq!(follow(&pool, community, &actor, &root).await, Some(None));
let summary = row(&db, community, channel, &actor).await;
assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]);
assert_eq!(threads(&summary), [(root.id.to_hex(), 0)]);
}
#[tokio::test]
@@ -202,7 +203,8 @@ async fn a_top_level_mention_follows_the_thread_rooted_at_it() {
let summary = row(&db, community, channel, &actor).await;
// The fixture message and the mention are unread; one is a mention.
assert_eq!((summary.unread, summary.mentions), (true, 1));
assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]);
// The reply is unread but does not mention the actor.
assert_eq!(threads(&summary), [(root.id.to_hex(), 0)]);
}
#[tokio::test]
@@ -236,8 +238,9 @@ async fn an_unfollowed_thread_never_counts_and_marking_it_follows_nothing() {
let last = reply(&db, community, channel, &root, &Keys::generate(), broadcast).await;
let summary = row(&db, community, channel, &actor).await;
assert!(summary.threads.is_empty());
// Replies are not on the timeline; only the fixture message is unread.
// Only the broadcast reply is also on the timeline.
assert_eq!((summary.unread, summary.mentions), (true, 0));
assert_eq!(summary.latest_id, Some(last.id.to_hex()));
let outcome = db
.apply_personal_read_intent(
@@ -270,7 +273,7 @@ async fn thread_counts_stop_at_the_cap() {
reply(&db, community, channel, &root, &actor, vec![]).await;
let other = Keys::generate();
for _ in 0..=MAX_UNREAD_COUNT {
reply(&db, community, channel, &root, &other, vec![]).await;
reply(&db, community, channel, &root, &other, mention(&actor)).await;
}
let summary = row(&db, community, channel, &actor).await;
assert_eq!(
@@ -355,7 +358,7 @@ async fn a_nested_reply_follows_the_root_thread() {
);
assert_eq!(follow(&pool, community, &bee, &parent).await, None);
let summary = row(&db, community, channel, &actor).await;
assert_eq!(threads(&summary), [(root.id.to_hex(), 2)]);
assert_eq!(threads(&summary), [(root.id.to_hex(), 0)]);
assert_eq!(summary.threads[0].latest_id, nested.id.to_hex());
}
@@ -371,3 +374,274 @@ async fn a_message_stored_without_thread_metadata_reads_the_timeline() {
assert!(!summary.unread);
assert_eq!(summary.read_through_id, Some(own.id.to_hex()));
}
fn follow_intent(channel: Uuid, root: Option<&nostr::Event>, follow: bool) -> ReadIntent {
let target = ReadTarget {
channel_id: channel,
root_id: root.map(|r| r.id.to_hex()),
};
if follow {
ReadIntent::Follow { target }
} else {
ReadIntent::Unfollow { target }
}
}
async fn intent(
db: &Db,
community: CommunityId,
actor: &Keys,
intent: ReadIntent,
) -> IntentOutcome {
db.apply_personal_read_intent(community, &actor.public_key(), &intent)
.await
.unwrap()
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn following_starts_now_and_needs_a_thread() {
let (db, pool, community, channel, actor, root) = fixture().await;
reply(&db, community, channel, &root, &Keys::generate(), vec![]).await;
assert_eq!(
intent(&db, community, &actor, follow_intent(channel, None, true)).await,
IntentOutcome::Invalid
);
let missing = message(&Keys::generate(), vec![]);
assert_eq!(
intent(
&db,
community,
&actor,
follow_intent(channel, Some(&missing), true)
)
.await,
IntentOutcome::Blocked
);
assert_eq!(
intent(
&db,
community,
&actor,
follow_intent(channel, Some(&root), true)
)
.await,
IntentOutcome::Applied
);
assert_eq!(follow(&pool, community, &actor, &root).await, Some(None));
// The reply before the follow is read; the next one is not.
assert!(row(&db, community, channel, &actor)
.await
.threads
.is_empty());
let next = reply(&db, community, channel, &root, &Keys::generate(), vec![]).await;
let summary = row(&db, community, channel, &actor).await;
assert_eq!(threads(&summary), [(root.id.to_hex(), 0)]);
assert_eq!(summary.threads[0].latest_id, next.id.to_hex());
// Following again keeps the position.
intent(
&db,
community,
&actor,
follow_intent(channel, Some(&root), true),
)
.await;
assert_eq!(
threads(&row(&db, community, channel, &actor).await).len(),
1
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn an_unfollow_sticks_until_a_mention_or_own_post() {
let (db, pool, community, channel, actor, _) = fixture().await;
// The actor's own root: its author follows on every reply.
let root = post(&db, community, channel, &actor, vec![]).await;
reply(&db, community, channel, &root, &Keys::generate(), vec![]).await;
assert_eq!(
threads(&row(&db, community, channel, &actor).await).len(),
1
);
assert_eq!(
intent(
&db,
community,
&actor,
follow_intent(channel, Some(&root), false)
)
.await,
IntentOutcome::Applied
);
reply(&db, community, channel, &root, &Keys::generate(), vec![]).await;
assert!(row(&db, community, channel, &actor)
.await
.threads
.is_empty());
// A mention follows again, from just before it.
let mentioning = reply(
&db,
community,
channel,
&root,
&Keys::generate(),
mention(&actor),
)
.await;
let summary = row(&db, community, channel, &actor).await;
assert_eq!(threads(&summary), [(root.id.to_hex(), 1)]);
assert_eq!(summary.threads[0].latest_id, mentioning.id.to_hex());
assert_eq!(summary.threads[0].read_through_id, None);
// So does posting, which reads through the post.
intent(
&db,
community,
&actor,
follow_intent(channel, Some(&root), false),
)
.await;
let own = reply(&db, community, channel, &root, &actor, vec![]).await;
assert_eq!(
follow(&pool, community, &actor, &root).await,
Some(Some(own.id.to_hex()))
);
reply(&db, community, channel, &root, &Keys::generate(), vec![]).await;
assert_eq!(
threads(&row(&db, community, channel, &actor).await).len(),
1
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn a_broadcast_reply_is_a_timeline_message_and_anchor() {
let (db, _pool, community, channel, actor, root) = fixture().await;
apply_mark(&db, community, channel, &actor, &root).await;
let broadcast = vec![
Tag::parse(["broadcast", "1"]).unwrap(),
Tag::public_key(actor.public_key()),
];
let last = reply(&db, community, channel, &root, &Keys::generate(), broadcast).await;
let summary = row(&db, community, channel, &actor).await;
assert_eq!((summary.unread, summary.mentions), (true, 1));
assert_eq!(summary.latest_id, Some(last.id.to_hex()));
assert_eq!(
apply_mark(&db, community, channel, &actor, &last).await,
IntentOutcome::Applied
);
assert!(!row(&db, community, channel, &actor).await.unread);
// A broadcast by the reader reads the timeline through it, as a post does.
post(&db, community, channel, &Keys::generate(), vec![]).await;
let broadcast = vec![Tag::parse(["broadcast", "1"]).unwrap()];
let own = reply(&db, community, channel, &root, &actor, broadcast).await;
let summary = row(&db, community, channel, &actor).await;
assert!(!summary.unread);
assert_eq!(summary.read_through_id, Some(own.id.to_hex()));
}
async fn apply_mark(
db: &Db,
community: CommunityId,
channel: Uuid,
actor: &Keys,
message: &nostr::Event,
) -> IntentOutcome {
let mark = ReadIntent::MarkThrough {
target: ReadTarget {
channel_id: channel,
root_id: None,
},
message_id: message.id.to_hex(),
};
intent(db, community, actor, mark).await
}
async fn threads_page(
db: &Db,
community: CommunityId,
actor: &Keys,
limit: usize,
after: Option<&str>,
) -> ThreadsPage {
let cursor = after.map(|c| ThreadsCursor::parse(c).unwrap());
db.personal_read_threads(
community,
&actor.public_key(),
DEFAULT_RETENTION_SECONDS,
limit,
cursor.as_ref(),
)
.await
.unwrap()
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn the_threads_view_lists_follows_newest_arrival_first() {
let (db, pool, community, channel, actor, _) = fixture().await;
let mut roots = Vec::new();
for _ in 0..3 {
let root = post(&db, community, channel, &Keys::generate(), vec![]).await;
reply(&db, community, channel, &root, &actor, vec![]).await;
roots.push(root);
}
// The first thread has news, so it comes first; the rest are read.
reply(
&db,
community,
channel,
&roots[0],
&Keys::generate(),
mention(&actor),
)
.await;
// Neither an unfollowed thread nor one with no reply is listed.
let quiet = post(&db, community, channel, &Keys::generate(), mention(&actor)).await;
let left = post(&db, community, channel, &Keys::generate(), vec![]).await;
reply(&db, community, channel, &left, &actor, vec![]).await;
intent(
&db,
community,
&actor,
follow_intent(channel, Some(&left), false),
)
.await;
assert_eq!(follow(&pool, community, &actor, &quiet).await, Some(None));
let first = threads_page(&db, community, &actor, 2, None).await;
let rows: Vec<_> = first
.threads
.iter()
.map(|t| (t.thread.root_id.clone(), t.thread.unread, t.thread.mentions))
.collect();
assert_eq!(
rows,
[
(roots[0].id.to_hex(), true, 1),
(roots[2].id.to_hex(), false, 0)
]
);
assert!(first.threads.iter().all(|t| t.channel_id == channel));
let rest = threads_page(&db, community, &actor, 2, first.next_cursor.as_deref()).await;
let rest: Vec<_> = rest
.threads
.iter()
.map(|t| t.thread.root_id.clone())
.collect();
assert_eq!(rest, [roots[1].id.to_hex()]);
// Leaving the channel hides its threads.
sqlx::query("UPDATE channel_members SET removed_at=now() WHERE community_id=$1 AND pubkey=$2")
.bind(community.as_uuid())
.bind(actor.public_key().to_bytes().as_slice())
.execute(&pool)
.await
.unwrap();
assert!(threads_page(&db, community, &actor, 2, None)
.await
.threads
.is_empty());
}

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

@@ -10,7 +10,7 @@ mod model;
mod projection;
mod writes;
pub(crate) use follows::{record_message, Place};
pub(crate) use follows::{place, record_message};
pub use model::*;
#[cfg(test)]

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

@@ -28,6 +28,29 @@ pub enum ReadIntent {
/// Fixed anchor; retry must not substitute the latest message.
message_id: String,
},
/// Follow a thread. A new follow, or one after an unfollow, starts now;
/// an existing follow keeps its position.
Follow {
/// The thread; a channel-only target is invalid.
target: ReadTarget,
},
/// Stop following a thread until the actor posts in it or is mentioned
/// there again.
Unfollow {
/// The thread; a channel-only target is invalid.
target: ReadTarget,
},
}
impl ReadIntent {
/// The intent's target.
pub fn target(&self) -> &ReadTarget {
match self {
ReadIntent::MarkThrough { target, .. }
| ReadIntent::Follow { target }
| ReadIntent::Unfollow { target } => target,
}
}
}
/// Outcome for one independent transaction, never acknowledged before commit.
@@ -57,16 +80,11 @@ pub const MAX_CHANNELS: usize = 20;
/// Maximum thread rows per channel row. A POST can return rows for
/// `MAX_INTENTS` channels, and they must fit the 1 MiB response limit.
pub const MAX_THREAD_SUMMARIES: usize = 25;
/// Events each walk examines before eligibility: newest-first for the
/// newest timeline message, then for the last arrival among those authored up
/// to twice the skew before it (in a thread too), and forward from the read
/// position for DM counts. What lies past the budget goes unseen.
pub const MAX_TIMELINE_SCAN: usize = 4096;
/// Counts stop here: a count of 100 means at least 100.
pub const MAX_UNREAD_COUNT: usize = 100;
/// The relay accepts author times up to this far from arrival either way.
/// Read positions are arrival times and the indexes are by author time, so
/// every forward walk starts this much before the position.
/// Read positions are arrival times and mentions are indexed by author time,
/// so a mention walk starts this much before the position.
pub const MAX_ARRIVAL_SKEW_SECONDS: u32 = 15 * 60;
/// Conversation kinds eligible for ordinary unread state (not edits/reactions).
pub const ELIGIBLE_KINDS: [i32; 4] = [9, 40002, 45001, 45003];
@@ -89,7 +107,7 @@ pub struct ChannelReadSummary {
/// Last eligible top-level message to arrive, whatever its author or read
/// progress: marking the timeline through it reads the timeline.
pub latest_id: Option<String>,
/// Followed threads with unread replies, newest unread reply first.
/// Followed threads with unread replies, newest arrival first.
pub threads: Vec<ThreadReadSummary>,
}
@@ -98,9 +116,10 @@ pub struct ChannelReadSummary {
pub struct ThreadReadSummary {
/// Canonical thread-root event ID.
pub root_id: String,
/// Whether an unread reply that counts was found.
/// Whether the last reply to arrive is unread.
pub unread: bool,
/// Unread replies by others in this followed thread.
/// Unread replies directed at the actor: every one in a DM, otherwise
/// those that tag the actor with `p`.
pub mentions: u32,
/// The anchor of the thread's frontier: the marked or own reply that
/// arrived last. None for a follower who has read nothing yet.
@@ -117,3 +136,43 @@ pub struct SidebarPage {
/// Exclusive UUID roster cursor. None means this roster scan exhausted.
pub next_cursor: Option<Uuid>,
}
/// Maximum rows in one threads page.
pub const MAX_THREADS: usize = 50;
/// One followed thread in the threads view.
#[derive(Debug, Serialize)]
pub struct FollowedThread {
/// Joined channel UUID.
pub channel_id: Uuid,
/// The thread's read state; `unread` may be false here.
#[serde(flatten)]
pub thread: ThreadReadSummary,
}
/// A page of followed threads, newest arrival first.
#[derive(Debug, Serialize)]
pub struct ThreadsPage {
/// Followed threads in joined channels.
pub threads: Vec<FollowedThread>,
/// Opaque exclusive cursor. None means this scan exhausted.
pub next_cursor: Option<String>,
}
/// Where a threads page resumes: after this arrival and root.
#[derive(Clone, Debug)]
pub struct ThreadsCursor {
pub(super) latest_at: DateTime<Utc>,
pub(super) root_id: Vec<u8>,
}
impl ThreadsCursor {
/// Parse a `next_cursor` value; None if it is not one.
pub fn parse(value: &str) -> Option<Self> {
let (micros, root) = value.split_once('.')?;
Some(Self {
latest_at: DateTime::from_timestamp_micros(micros.parse().ok()?)?,
root_id: super::writes::event_id(root)?,
})
}
}

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

@@ -62,6 +62,20 @@ pub(super) async fn start_before_everything(
}
/// Store `event` as a canonical reply in `root`'s thread, through ingest.
/// Copy rewritten event arrivals onto their replies' thread rows, as ingest
/// stores them.
pub(super) async fn sync_reply_arrivals(pool: &PgPool) {
sqlx::query(
"UPDATE thread_metadata tm SET received_at=e.received_at FROM events e
WHERE e.community_id=tm.community_id AND e.created_at=tm.event_created_at
AND e.id=tm.event_id AND tm.received_at IS NOT NULL
AND tm.received_at<>e.received_at",
)
.execute(pool)
.await
.unwrap();
}
pub(super) async fn insert_reply(
db: &Db,
community: CommunityId,
@@ -179,9 +193,9 @@ async fn add_history(db: &Db, community: CommunityId, channel: Uuid, count: usiz
#[tokio::test]
#[ignore = "requires Postgres"]
async fn personal_read_sidebar_marked_history_past_the_scan_cap_is_read_and_keeps_latest() {
async fn personal_read_sidebar_marked_history_is_read_and_keeps_latest() {
let (db, _pool, community, channel, actor, _) = fixture().await;
let last = add_history(&db, community, channel, MAX_TIMELINE_SCAN + 44).await;
let last = add_history(&db, community, channel, 300).await;
let outcome = db
.apply_personal_read_intent(
community,
@@ -191,7 +205,6 @@ async fn personal_read_sidebar_marked_history_past_the_scan_cap_is_read_and_keep
.await
.unwrap();
assert_eq!(outcome, IntentOutcome::Applied);
// The scan never reaches the oldest 45 messages; nothing is unread.
let row = &sidebar(&db, community, &actor).await.channels[0];
assert!(!row.unread);
assert_eq!(row.mentions, 0);
@@ -203,7 +216,7 @@ async fn personal_read_sidebar_marked_history_past_the_scan_cap_is_read_and_keep
#[ignore = "requires Postgres"]
async fn personal_read_sidebar_author_window_excludes_expired_and_late_old_messages() {
let (db, pool, community, channel, actor, _) = fixture().await;
let latest = add_history(&db, community, channel, MAX_TIMELINE_SCAN + 44).await;
let latest = add_history(&db, community, channel, 300).await;
expire(&pool, community).await;
let row = &sidebar(&db, community, &actor).await.channels[0];
assert!(
@@ -734,13 +747,13 @@ async fn personal_read_counts_only_arrivals_after_the_account_starts() {
#[tokio::test]
#[ignore = "requires Postgres"]
async fn personal_read_unstarted_account_is_caught_up_past_the_scan_cap() {
async fn personal_read_unstarted_account_is_caught_up() {
let (db, pool, community, channel, actor, _) = fixture().await;
unstart(&pool, community).await;
let last = add_history(&db, community, channel, MAX_TIMELINE_SCAN + 44).await;
let last = add_history(&db, community, channel, 300).await;
let page = sidebar(&db, community, &actor).await;
let row = &page.channels[0];
// The start floor covers every arrival, examined by the scan or not.
// The start floor covers every arrival.
assert!(!row.unread);
assert_eq!(row.mentions, 0);
assert!(row.threads.is_empty());

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

@@ -1,7 +1,7 @@
//! Sidebar read state, counted forward from each read position. A caught-up
//! reader's work is a few index probes per channel and per followed thread;
//! a reader who is behind walks only what arrived after their position, and
//! counts stop at `MAX_UNREAD_COUNT`.
//! Sidebar and threads read state, counted forward from each read position.
//! Each channel timeline and each followed thread is one index probe for its
//! last arrival; counts walk only what arrived after the position and stop at
//! `MAX_UNREAD_COUNT`.
use super::{model::*, writes};
use buzz_core::CommunityId;
@@ -11,31 +11,37 @@ use uuid::Uuid;
use crate::{observability, Db, DbError, Result};
/// Event `e` is eligible: a conversation kind, not deleted.
const ELIGIBLE: &str = "e.kind=ANY($5) AND e.deleted_at IS NULL";
/// The last eligible reply `tl` to arrive in thread `tf` (idx_thread_metadata_read_arrival).
/// Deletion is a per-row filter, not a join, so the walk stops at the first
/// living reply instead of sorting them all.
const LATEST_REPLY: &str = "SELECT tm.event_id AS id, tm.received_at, e.pubkey, e.created_at
FROM (SELECT tm.event_id, tm.event_created_at, tm.received_at FROM thread_metadata tm
WHERE tm.community_id=$1 AND tm.root_event_id=tf.root_id AND tm.received_at IS NOT NULL
AND tm.channel_id=tf.channel_id
AND (SELECT e.deleted_at IS NULL FROM events e WHERE e.community_id=$1
AND e.created_at=tm.event_created_at AND e.id=tm.event_id)
ORDER BY tm.received_at DESC, tm.event_id LIMIT 1) tm
JOIN events e ON e.community_id=$1 AND e.created_at=tm.event_created_at AND e.id=tm.event_id";
/// Event `e`, with its `thread_metadata` row `tm`, is on the channel timeline:
/// not a reply, and not a NIP-10 reply whose ancestry was never recorded
/// (neither counted nor a timeline anchor, since a mark cannot validate it).
/// Parity-tested against the shared NIP-10 parser.
const TIMELINE: &str = r#"(tm.root_event_id IS NULL OR tm.root_event_id=e.id)
AND NOT (tm.root_event_id IS NULL AND jsonb_typeof(e.tags)='array'
AND EXISTS (SELECT 1 FROM jsonb_array_elements(e.tags) t(tag)
WHERE jsonb_typeof(tag)='array' AND tag->>0='e' AND tag->>3='reply'
AND (tag->>1) COLLATE "C" ~ '^[0123456789abcdefABCDEF]{64}$'))"#;
/// What the budgeted timeline walks carry out of their `LIMIT`.
const COLUMNS: &str =
"e.id, e.pubkey, e.created_at, e.received_at, e.kind, e.deleted_at, e.tags, e.channel_id";
const EVENT_THREAD_ROW: &str = "LEFT JOIN thread_metadata tm ON tm.community_id=$1
AND tm.channel_id=e.channel_id AND tm.event_created_at=e.created_at AND tm.event_id=e.id";
/// Replies `tm` in followed thread `tf`, with their events `e`.
const THREAD_REPLIES: &str = "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";
/// Unread replies in thread `t` at position `t.position` that are directed at
/// the actor: every one in a DM (`t.dm`), otherwise those that tag the actor.
/// Nothing is walked unless `t.unread`.
const THREAD_MENTIONS: &str = "SELECT count(*) AS n FROM (
SELECT 1 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 t.unread AND t.dm AND tm.community_id=$1 AND tm.root_event_id=t.root_id
AND tm.received_at > t.position AND tm.channel_id=t.channel_id
AND e.deleted_at IS NULL AND e.pubkey<>$2 AND e.created_at >= $4
UNION ALL
SELECT 1 FROM event_mentions em JOIN thread_metadata tm ON tm.community_id=$1
AND tm.event_created_at=em.event_created_at AND tm.event_id=em.event_id
JOIN events e ON e.community_id=$1 AND e.created_at=em.event_created_at AND e.id=em.event_id
WHERE t.unread AND NOT t.dm AND em.community_id=$1 AND em.pubkey_hex=$5 AND em.channel_id=t.channel_id
AND em.event_created_at >= GREATEST(t.position-$6, $4)
AND tm.root_event_id=t.root_id AND tm.received_at > t.position
AND e.deleted_at IS NULL AND e.pubkey<>$2
LIMIT $7
) x";
impl Db {
/// Read a bounded joined roster from the writer in one read-only snapshot.
@@ -103,131 +109,77 @@ impl Db {
let actor_bytes = actor.to_bytes();
let account = read_account(&mut tx, community, &actor_bytes, retention_seconds).await?;
// A position is its frontier floored at the account's start (infinity
// before the actor's first intent, so nothing counts). Positions are
// arrival times and the indexes are by author time, which the relay
// accepts within the skew of arrival, so a walk forward from a
// position starts the skew before it (and never before the horizon).
// before the actor's first intent, so nothing counts).
//
// Timeline, per channel:
// - latest_id: the newest timeline message by author time within the
// scan budget, then the last to arrive among the newest budget of
// events authored up to twice the skew before it, which holds the
// last arrival unless that many arrived out of order;
// - latest_id: the last timeline message to arrive (idx_events_read_timeline);
// - unread: latest is someone else's, inside the horizon, and arrived
// after the position (an own post moves the position past it);
// - mentions: those of them that tag the actor (event_mentions), or
// every one in a DM.
// Threads: each followed thread whose newest reply could have arrived
// after its position (one index-only probe of
// idx_thread_metadata_window), then its last arrival the same way as
// latest_id, and its unread replies by others.
// - mentions: unread timeline messages that tag the actor, found by
// author time (event_mentions) from the skew before the position,
// or every one in a DM.
// Threads: each followed thread whose last reply is unread the same
// way, then its mentions for the listed ones only.
let sql = format!(
r#"WITH roster AS MATERIALIZED (
SELECT c.id, c.channel_type='dm' AS dm,
GREATEST(cf.through_timestamp, COALESCE($7::timestamptz, 'infinity')) AS position,
GREATEST(cf.through_timestamp, COALESCE($3::timestamptz, 'infinity')) AS position,
encode(cf.through_message_id,'hex') AS read_through_id
FROM channel_members cm JOIN channels c
ON c.community_id=cm.community_id AND c.id=cm.channel_id
LEFT JOIN personal_read_frontiers cf ON cf.community_id=$1 AND cf.actor=$2
AND cf.channel_id=c.id AND cf.root_id=''::bytea
WHERE cm.community_id=$1 AND cm.pubkey=$2 AND cm.removed_at IS NULL
AND c.deleted_at IS NULL AND ($3::uuid IS NULL OR c.id>$3)
AND ($4::uuid[] IS NULL OR c.id=ANY($4))
ORDER BY c.id LIMIT $6
AND c.deleted_at IS NULL AND ($8::uuid IS NULL OR c.id>$8)
AND ($9::uuid[] IS NULL OR c.id=ANY($9))
ORDER BY c.id LIMIT $10
)
SELECT r.id, r.read_through_id, encode(l.id,'hex') AS latest_id, u.unread,
COALESCE(m.mentions, 0) AS mentions, COALESCE(th.items, '[]') AS threads
FROM roster r
CROSS JOIN LATERAL (SELECT GREATEST(r.position-$8, $9) AS walk_from) w
LEFT JOIN LATERAL (
SELECT e.created_at FROM (
SELECT {COLUMNS} FROM events e WHERE e.community_id=$1 AND e.channel_id=r.id
ORDER BY e.created_at DESC, e.id DESC LIMIT $10
) e {EVENT_THREAD_ROW}
WHERE {ELIGIBLE} AND {TIMELINE}
ORDER BY e.created_at DESC, e.id DESC LIMIT 1
) n ON true
LEFT JOIN LATERAL (
SELECT e.id, e.pubkey, e.created_at, e.received_at FROM (
SELECT {COLUMNS} FROM events e WHERE e.community_id=$1 AND e.channel_id=r.id
AND e.created_at BETWEEN n.created_at-2*$8 AND n.created_at
ORDER BY e.created_at DESC, e.id DESC LIMIT $10
) e {EVENT_THREAD_ROW}
WHERE {ELIGIBLE} AND {TIMELINE}
SELECT e.id, e.pubkey, e.created_at, e.received_at FROM events e
WHERE e.community_id=$1 AND e.channel_id=r.id AND e.read_timeline
AND e.deleted_at IS NULL
ORDER BY e.received_at DESC, e.id LIMIT 1
) l ON true
-- The last arrival is unread exactly when the timeline is: the
-- actor's own posts read through themselves.
CROSS JOIN LATERAL (
SELECT COALESCE(l.received_at > r.position AND l.pubkey<>$2
AND l.created_at >= $9, false) AS unread
AND l.created_at >= $4, false) AS unread
) u
LEFT JOIN LATERAL (
SELECT count(*) AS mentions FROM (
SELECT 1 FROM (
SELECT {COLUMNS} FROM events e WHERE r.dm AND e.community_id=$1
AND e.channel_id=r.id AND e.created_at >= w.walk_from
ORDER BY e.created_at, e.id LIMIT $10
) e {EVENT_THREAD_ROW}
WHERE e.received_at > r.position AND e.pubkey<>$2 AND {ELIGIBLE} AND {TIMELINE}
SELECT 1 FROM events e WHERE r.dm AND e.community_id=$1
AND e.channel_id=r.id AND e.read_timeline AND e.deleted_at IS NULL
AND e.received_at > r.position AND e.pubkey<>$2 AND e.created_at >= $4
UNION ALL
SELECT 1 FROM event_mentions em JOIN events e ON e.community_id=$1
AND e.created_at=em.event_created_at AND e.id=em.event_id {EVENT_THREAD_ROW}
WHERE NOT r.dm AND em.community_id=$1 AND em.pubkey_hex=$11
AND em.channel_id=r.id AND em.event_created_at >= w.walk_from
AND e.channel_id=r.id AND e.received_at > r.position
AND e.pubkey<>$2 AND {ELIGIBLE} AND {TIMELINE}
LIMIT $12
AND e.created_at=em.event_created_at AND e.id=em.event_id
WHERE NOT r.dm AND em.community_id=$1 AND em.pubkey_hex=$5
AND em.channel_id=r.id AND em.event_created_at >= GREATEST(r.position-$6, $4)
AND e.read_timeline AND e.deleted_at IS NULL
AND e.received_at > r.position AND e.pubkey<>$2
LIMIT $7
) x WHERE u.unread
) m ON true
LEFT JOIN LATERAL (
SELECT jsonb_agg(jsonb_build_object('root_id',encode(t.root_id,'hex'),
'unread',true,'mentions',t.n,'read_through_id',t.read_through_id,
'latest_id',t.latest_id) ORDER BY t.latest_at DESC, t.root_id) AS items
'unread',true,'mentions',tc.n,'read_through_id',t.read_through_id,
'latest_id',encode(t.latest_id,'hex'))
ORDER BY t.latest_at DESC, t.root_id) AS items
FROM (
SELECT tf.root_id, encode(tf.through_message_id,'hex') AS read_through_id,
encode(tl.id,'hex') AS latest_id, tl.created_at AS latest_at, tc.n
SELECT tf.root_id, tf.channel_id, r.dm, p.position, true AS unread,
encode(tf.through_message_id,'hex') AS read_through_id,
tl.id AS latest_id, tl.received_at AS latest_at
FROM personal_read_frontiers tf
CROSS JOIN LATERAL (SELECT GREATEST(tf.through_timestamp, COALESCE($7, 'infinity')) AS position) p
CROSS JOIN LATERAL (
SELECT tm.event_created_at AS created_at FROM thread_metadata tm
WHERE tm.community_id=$1 AND tm.root_event_id=tf.root_id
ORDER BY tm.event_created_at DESC, tm.event_id LIMIT 1
) tq
CROSS JOIN LATERAL (
-- Eligibility is a per-row filter, not a join, so the
-- planner walks the window index newest-first and stops
-- at the first eligible reply instead of sorting them all.
SELECT tm.event_created_at AS created_at FROM thread_metadata tm
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 (SELECT {ELIGIBLE} FROM events e WHERE e.community_id=$1
AND e.created_at=tm.event_created_at AND e.id=tm.event_id)
ORDER BY tm.event_created_at DESC, tm.event_id LIMIT 1
) tn
CROSS JOIN LATERAL (
SELECT e.id, e.created_at, e.received_at FROM (
SELECT e.id, e.created_at, e.received_at FROM {THREAD_REPLIES}
AND tm.event_created_at BETWEEN tn.created_at-2*$8 AND tn.created_at
AND {ELIGIBLE}
ORDER BY tm.event_created_at DESC, tm.event_id LIMIT $10
) e
ORDER BY e.received_at DESC, e.id LIMIT 1
) tl
CROSS JOIN LATERAL (
SELECT count(*) AS n FROM (
SELECT 1 FROM {THREAD_REPLIES}
AND tm.event_created_at >= GREATEST(p.position-$8, $9)
AND e.received_at > p.position AND e.pubkey<>$2 AND {ELIGIBLE}
LIMIT $12
) x
) tc
CROSS JOIN LATERAL (SELECT GREATEST(tf.through_timestamp,
COALESCE($3, 'infinity')) AS position) p
CROSS JOIN LATERAL ({LATEST_REPLY}) tl
WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=r.id
AND tf.root_id<>''::bytea
AND tq.created_at+$8 > p.position AND tl.received_at > p.position
AND tc.n > 0
ORDER BY tl.created_at DESC, tf.root_id LIMIT $13
) t
AND tf.root_id<>''::bytea AND NOT tf.unfollowed
AND tl.received_at > p.position AND tl.pubkey<>$2 AND tl.created_at >= $4
ORDER BY tl.received_at DESC, tf.root_id LIMIT $11
) t CROSS JOIN LATERAL ({THREAD_MENTIONS}) tc
) th ON true
ORDER BY r.id"#
);
@@ -237,16 +189,14 @@ impl Db {
let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(community.as_uuid())
.bind(actor_bytes.as_slice())
.bind(account.started_at)
.bind(cutoff)
.bind(actor.to_hex())
.bind(skew())
.bind(MAX_UNREAD_COUNT as i64)
.bind(after)
.bind(only)
.bind(ELIGIBLE_KINDS.as_slice())
.bind((limit + 1) as i64)
.bind(account.started_at)
.bind(skew())
.bind(cutoff)
.bind(MAX_TIMELINE_SCAN as i64)
.bind(actor.to_hex())
.bind(MAX_UNREAD_COUNT as i64)
.bind(MAX_THREAD_SUMMARIES as i64)
.fetch_all(&mut *tx)
.await?;
@@ -277,6 +227,106 @@ impl Db {
next_cursor,
})
}
/// Followed threads in joined channels, newest arrival first, read or
/// not, from the writer in one read-only snapshot. A thread with no reply
/// yet is omitted. Callers must recheck admission/resource access before
/// releasing this data.
pub async fn personal_read_threads(
&self,
community: CommunityId,
actor: &nostr::PublicKey,
retention_seconds: u32,
limit: usize,
after: Option<&ThreadsCursor>,
) -> Result<ThreadsPage> {
if !(1..=MAX_THREADS).contains(&limit) {
return Err(DbError::InvalidData("invalid threads limit".into()));
}
let mut conn = observability::acquire_writer(
&self.pool,
observability::WriterOperation::SubscriptionHistory,
)
.await?;
let mut tx = conn.begin().await?;
sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY")
.execute(&mut *tx)
.await?;
writes::deadlines(&mut tx).await?;
sqlx::query("SET LOCAL jit = off").execute(&mut *tx).await?;
let actor_bytes = actor.to_bytes();
let account = read_account(&mut tx, community, &actor_bytes, retention_seconds).await?;
// Every follow is probed for its last arrival and sorted: the cost
// grows with the actor's lifetime follows, not with community size.
let sql = format!(
r#"SELECT t.channel_id, encode(t.root_id,'hex') AS root_id, t.unread,
tc.n AS mentions,
encode(t.through_message_id,'hex') AS read_through_id,
encode(t.latest_id,'hex') AS latest_id, t.latest_at
FROM (
SELECT tf.root_id, tf.channel_id, c.channel_type='dm' AS dm, p.position,
tf.through_message_id, tl.id AS latest_id, tl.received_at AS latest_at,
(tl.received_at > p.position AND tl.pubkey<>$2 AND tl.created_at >= $4) AS unread
FROM personal_read_frontiers tf
JOIN channel_members cm ON cm.community_id=$1 AND cm.channel_id=tf.channel_id
AND cm.pubkey=$2 AND cm.removed_at IS NULL
JOIN channels c ON c.community_id=$1 AND c.id=tf.channel_id AND c.deleted_at IS NULL
CROSS JOIN LATERAL (SELECT GREATEST(tf.through_timestamp,
COALESCE($3::timestamptz, 'infinity')) AS position) p
CROSS JOIN LATERAL ({LATEST_REPLY}) tl
WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.root_id<>''::bytea
AND NOT tf.unfollowed AND ($8::timestamptz IS NULL OR tl.received_at < $8
OR (tl.received_at=$8 AND tf.root_id>$9))
ORDER BY tl.received_at DESC, tf.root_id LIMIT $10
) t CROSS JOIN LATERAL ({THREAD_MENTIONS}) tc
ORDER BY t.latest_at DESC, t.root_id"#
);
let cutoff = DateTime::from_timestamp_millis(account.cutoff_ms)
.ok_or_else(|| DbError::InvalidData("invalid unread cutoff".into()))?;
// Only compile-time constants are interpolated.
let rows = sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(community.as_uuid())
.bind(actor_bytes.as_slice())
.bind(account.started_at)
.bind(cutoff)
.bind(actor.to_hex())
.bind(skew())
.bind(MAX_UNREAD_COUNT as i64)
.bind(after.map(|c| c.latest_at))
.bind(after.map(|c| c.root_id.as_slice()))
.bind((limit + 1) as i64)
.fetch_all(&mut *tx)
.await?;
tx.commit().await?;
let has_more = rows.len() > limit;
let mut threads = Vec::with_capacity(limit);
let mut last = None;
for row in rows.into_iter().take(limit) {
let mentions: i64 = row.try_get("mentions")?;
let root_id: String = row.try_get("root_id")?;
last = Some((
row.try_get::<DateTime<Utc>, _>("latest_at")?,
root_id.clone(),
));
threads.push(FollowedThread {
channel_id: row.try_get("channel_id")?,
thread: ThreadReadSummary {
root_id,
unread: row.try_get("unread")?,
mentions: mentions as u32,
read_through_id: row.try_get("read_through_id")?,
latest_id: row.try_get("latest_id")?,
},
});
}
let next_cursor = last
.filter(|_| has_more)
.map(|(at, root)| format!("{}.{root}", at.timestamp_micros()));
Ok(ThreadsPage {
threads,
next_cursor,
})
}
}
fn skew() -> chrono::Duration {

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

@@ -1,26 +1,16 @@
use super::{postgres_tests::fixture, *};
use crate::Db;
use nostr::{EventBuilder, Keys, Kind, Tag};
use serde_json::json;
/// Eligibility is classified at ingest (kind) and at read (author, deletion,
/// horizon). Each case is stored through ingest, then rewritten in place for
/// the read-time facts.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn sidebar_sql_eligibility_follows_kind_author_deletion_and_horizon() {
let (db, pool, community, _, actor, event) = fixture().await;
let (db, pool, community, channel, actor, _) = fixture().await;
let now = chrono::Utc::now().timestamp_millis();
let horizon = i64::from(DEFAULT_RETENTION_SECONDS) * 1000;
// Addressed to the actor, so a counted message is also exactly one mention.
let tags = json!([["p", actor.public_key().to_hex()]]);
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",
)
.bind(community.as_uuid())
.bind(actor.public_key().to_hex())
.execute(&pool)
.await
.unwrap();
for (kind, eligible_kind) in [
(9, true),
(40002, true),
@@ -28,9 +18,22 @@ async fn sidebar_sql_eligibility_follows_kind_author_deletion_and_horizon() {
(45003, true),
(1, false),
(7, false),
(39002, false),
(40008, false),
] {
sqlx::query("UPDATE events SET deleted_at=now() WHERE community_id=$1")
.bind(community.as_uuid())
.execute(&pool)
.await
.unwrap();
// Addressed to the actor, so a counted message is also exactly one mention.
let author = Keys::generate();
let event = EventBuilder::new(Kind::Custom(kind), "case")
.tags([Tag::public_key(actor.public_key())])
.sign_with_keys(&author)
.unwrap();
db.insert_event(community, &event, Some(channel))
.await
.unwrap();
for own in [false, true] {
for deleted in [false, true] {
// Now, then one minute inside and one minute outside the horizon.
@@ -40,12 +43,13 @@ async fn sidebar_sql_eligibility_follows_kind_author_deletion_and_horizon() {
(horizon + 60_000, false),
] {
let created = now - age;
sqlx::query("UPDATE events SET kind=$2,pubkey=$3,deleted_at=CASE WHEN $4 THEN now() ELSE NULL END,created_at=to_timestamp($5::double precision/1000),tags=$6 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).bind(created as f64).bind(&tags).execute(&pool).await.unwrap();
sqlx::query("UPDATE event_mentions SET event_created_at=to_timestamp($2::double precision/1000) WHERE community_id=$1")
.bind(community.as_uuid()).bind(created as f64).execute(&pool).await.unwrap();
sqlx::query("UPDATE events SET pubkey=$3,deleted_at=CASE WHEN $4 THEN now() ELSE NULL END,created_at=to_timestamp($5::double precision/1000) WHERE community_id=$1 AND id=$2")
.bind(community.as_uuid()).bind(event.id.as_bytes().as_slice())
.bind(if own { actor.public_key() } else { author.public_key() }.to_bytes().as_slice())
.bind(deleted).bind(created as f64).execute(&pool).await.unwrap();
sqlx::query("UPDATE event_mentions SET event_created_at=to_timestamp($3::double precision/1000) WHERE community_id=$1 AND event_id=$2")
.bind(community.as_uuid()).bind(event.id.as_bytes().as_slice())
.bind(created as f64).execute(&pool).await.unwrap();
let page = db
.personal_read_sidebar(
community,
@@ -69,10 +73,12 @@ async fn sidebar_sql_eligibility_follows_kind_author_deletion_and_horizon() {
}
}
/// A message with a NIP-10 reply marker but no recorded ancestry is on
/// neither read path, by the shared parser's definition of a reply.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn sidebar_ancestry_fact_matches_shared_nip10_parser() {
let (db, pool, community, _, actor, _) = fixture().await;
let (db, pool, community, channel, actor, _) = fixture().await;
let ids = [
"a".repeat(64),
"A".repeat(64),
@@ -104,12 +110,22 @@ async fn sidebar_ancestry_fact_matches_shared_nip10_parser() {
buzz_core::nip10::parse_thread_markers_from_parts(tags.iter().map(Vec::as_slice))
.resolve()
.is_some();
sqlx::query("UPDATE events SET tags=$2 WHERE community_id=$1")
sqlx::query("UPDATE events SET deleted_at=now() WHERE community_id=$1")
.bind(community.as_uuid())
.bind(json!(tags))
.execute(&pool)
.await
.unwrap();
let parsed: Vec<Tag> = tags
.iter()
.map(|tag| Tag::parse(tag.iter().map(String::as_str)).unwrap())
.collect();
let event = EventBuilder::new(Kind::Custom(9), "case")
.tags(parsed)
.sign_with_keys(&Keys::generate())
.unwrap();
db.insert_event(community, &event, Some(channel))
.await
.unwrap();
let page = db
.personal_read_sidebar(
community,

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

@@ -1,6 +1,6 @@
//! Thread summaries, anchor validation and targeted refresh.
use super::{
postgres_tests::{fixture, insert_reply},
postgres_tests::{fixture, insert_reply, sync_reply_arrivals},
*,
};
use crate::{
@@ -179,7 +179,7 @@ async fn thread_summaries_order_cap_and_anchor() {
channel_mark(channel, newest_root.id.to_hex()),
)
.await;
// Threads 0 and 1 tie on newest reply time and the rest follow, newest
// Threads 0 and 1 tie on last arrival and the rest follow, newest
// first. Thread 2 has one directed reply and two that arrive last,
// together.
let offset = |i: u64| if i < 2 { n * 10 } else { (n - i) * 10 };
@@ -206,14 +206,18 @@ async fn thread_summaries_order_cap_and_anchor() {
vec![],
)
.await;
// Replies arrive in author order, so the twin and thread 2's anchor
// arrive together.
sqlx::query(
"UPDATE events SET received_at=(SELECT received_at FROM events WHERE id=$1) WHERE id=$2",
"UPDATE events e SET received_at=e.created_at FROM thread_metadata tm
WHERE tm.community_id=e.community_id AND tm.event_id=e.id
AND tm.received_at IS NOT NULL AND e.community_id=$1",
)
.bind(twin.id.as_bytes().as_slice())
.bind(anchors[2].id.as_bytes().as_slice())
.bind(community.as_uuid())
.execute(&pool)
.await
.unwrap();
sync_reply_arrivals(&pool).await;
let row = sidebar(&db, community, &actor).await;
// Every root is read; the eight unread replies count only on threads.
@@ -230,7 +234,7 @@ async fn thread_summaries_order_cap_and_anchor() {
// The oldest unread thread is past the cap.
assert_eq!(order(&row), [tied.to_vec(), ids(&roots[2..last])].concat());
let third = &row.threads[2];
assert_eq!(third.mentions, 3);
assert_eq!(third.mentions, 1, "only the directed reply is a mention");
assert_eq!(
third.latest_id,
std::cmp::min(anchors[2].id.to_hex(), twin.id.to_hex()),
@@ -249,8 +253,8 @@ async fn thread_summaries_order_cap_and_anchor() {
[vec![tied[1].clone()], ids(&roots[2..])].concat()
);
let mentions: Vec<_> = row.threads.iter().map(|t| t.mentions).collect();
let mut expected = vec![1; MAX_THREAD_SUMMARIES];
expected[1] = 3;
let mut expected = vec![0; MAX_THREAD_SUMMARIES];
expected[1] = 1;
assert_eq!(mentions, expected);
}

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

@@ -88,6 +88,8 @@ struct Message {
id: Vec<u8>,
received_at: DateTime<Utc>,
root: Option<Vec<u8>>,
/// A reply that is also on the channel timeline.
broadcast: bool,
}
/// A channel event and its canonical ancestry. `kinds` bounds the lookup: an
@@ -100,7 +102,7 @@ async fn message(
kinds: Option<&[i32]>,
) -> Result<Option<Message>> {
let row = sqlx::query(
"SELECT e.id, e.received_at, e.tags, tm.root_event_id
"SELECT e.id, e.received_at, e.tags, tm.root_event_id, COALESCE(tm.broadcast, false) AS 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
@@ -125,6 +127,7 @@ async fn message(
id,
received_at,
root,
broadcast: row.try_get("broadcast")?,
})
})
.transpose()
@@ -199,6 +202,38 @@ async fn frontier(
Ok(())
}
/// Follow or unfollow a thread. A follow that changes state starts now, so
/// earlier replies are read; an unfollowed row keeps ingest from following
/// again, except as `record_message` allows.
async fn follow(
conn: &mut PgConnection,
community: CommunityId,
actor: &[u8],
channel: Uuid,
root: &[u8],
unfollow: bool,
) -> Result<()> {
sqlx::query(
"INSERT INTO personal_read_frontiers
(community_id, actor, channel_id, root_id, through_timestamp, unfollowed)
VALUES ($1,$2,$3,$4,now(),$5)
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE
SET unfollowed=$5,
through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, now()),
through_message_id=CASE WHEN now() > personal_read_frontiers.through_timestamp
THEN NULL ELSE personal_read_frontiers.through_message_id END
WHERE personal_read_frontiers.unfollowed <> $5",
)
.bind(community.as_uuid())
.bind(actor)
.bind(channel)
.bind(root)
.bind(unfollow)
.execute(conn)
.await?;
Ok(())
}
pub(super) async fn apply(
conn: &mut PgConnection,
community: CommunityId,
@@ -225,13 +260,30 @@ pub(super) async fn apply(
return Ok(IntentOutcome::Blocked);
};
let is_reply = msg.root.as_ref().is_some_and(|r| r != &msg.id);
if (root.is_empty() && is_reply)
if (root.is_empty() && is_reply && !msg.broadcast)
|| (!root.is_empty() && msg.id != root && msg.root.as_ref() != Some(&root))
{
return Ok(IntentOutcome::Blocked);
}
frontier(conn, community, actor, target, &root, &msg).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);
};
follow(
conn,
community,
actor,
target.channel_id,
&root,
matches!(intent, ReadIntent::Unfollow { .. }),
)
.await?;
}
}
Ok(IntentOutcome::Applied)
}

69
crates/buzz-relay/src/api/buzz_v1/handlers.rs
View File

@@ -7,7 +7,8 @@ use axum::{
response::Response,
};
use buzz_db::personal_read::{
ChannelReadSummary, ReadIntent, SidebarPage, MAX_CHANNELS, MAX_INTENTS,
ChannelReadSummary, ReadIntent, ThreadsCursor, ThreadsPage, MAX_CHANNELS, MAX_INTENTS,
MAX_THREADS,
};
use serde::Deserialize;
use serde_json::{json, Value};
@@ -74,27 +75,72 @@ pub(super) async fn sidebar(
}
}
.map_err(|_| Error::unavailable())?;
auth::response(release(&state, &headers, &principal, page).await?)
let channels: Vec<_> = page.channels.iter().map(|c| c.channel_id).collect();
release(&state, &headers, &principal, &channels).await?;
auth::response(page)
})
.await
.map_err(|_| Error::unavailable())?
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
pub(super) struct ThreadsQuery {
limit: Option<usize>,
cursor: Option<String>,
}
pub(super) async fn threads(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
OriginalUri(uri): OriginalUri,
query: Result<Query<ThreadsQuery>, axum::extract::rejection::QueryRejection>,
) -> Result<Response, Error> {
tokio::time::timeout(Duration::from_secs(8), async {
let principal = auth::authorize(&state, &headers, &uri, "GET", None).await?;
let Query(query) = query.map_err(|_| Error::invalid())?;
let limit = query.limit.unwrap_or(MAX_THREADS);
let cursor = query
.cursor
.map(|c| ThreadsCursor::parse(&c).ok_or_else(Error::invalid))
.transpose()?;
if !(1..=MAX_THREADS).contains(&limit) {
return Err(Error::invalid());
}
let page: ThreadsPage = state
.db
.personal_read_threads(
principal.tenant.community(),
&principal.actor,
state.config.buzz_v1_retention_seconds,
limit,
cursor.as_ref(),
)
.await
.map_err(|_| Error::unavailable())?;
let channels: BTreeSet<_> = page.threads.iter().map(|t| t.channel_id).collect();
let channels: Vec<_> = channels.into_iter().collect();
release(&state, &headers, &principal, &channels).await?;
auth::response(page)
})
.await
.map_err(|_| Error::unavailable())?
}
/// Release projected rows only after admission and the actor's membership of
/// every row are rechecked outside the projection snapshot.
/// every row's channel are rechecked outside the projection snapshot.
async fn release(
state: &AppState,
headers: &HeaderMap,
principal: &auth::Principal,
page: SidebarPage,
) -> Result<SidebarPage, Error> {
channels: &[Uuid],
) -> Result<(), Error> {
auth::recheck(state, headers, principal).await?;
let channels: Vec<_> = page.channels.iter().map(|c| c.channel_id).collect();
let memberships = state
.db
.membership_pairs(
principal.tenant.community(),
&channels,
channels,
&[principal.actor.to_bytes().to_vec()],
)
.await
@@ -102,7 +148,7 @@ async fn release(
if memberships.len() != channels.len() {
return Err(Error::unavailable());
}
Ok(page)
Ok(())
}
/// Updated rows for the channels a batch applied to. A channel the actor has
@@ -126,7 +172,9 @@ async fn refreshed(
)
.await
.map_err(|_| Error::unavailable())?;
rows.extend(release(state, headers, principal, page).await?.channels);
let ids: Vec<_> = page.channels.iter().map(|c| c.channel_id).collect();
release(state, headers, principal, &ids).await?;
rows.extend(page.channels);
}
Ok(rows)
}
@@ -168,8 +216,7 @@ pub(super) async fn write(
.await
.unwrap_or_else(|_| json!({"status":"unknown","retryable":true}));
if outcome["status"] == "applied" {
let ReadIntent::MarkThrough { target, .. } = &intent;
applied.insert(target.channel_id);
applied.insert(intent.target().channel_id);
}
outcomes.push(outcome);
}

1
crates/buzz-relay/src/api/buzz_v1/mod.rs
View File

@@ -17,6 +17,7 @@ pub const BASE_PATH: &str = "/buzz/v1";
pub fn router(state: Arc<AppState>) -> Router {
Router::new()
.route("/me/sidebar", get(handlers::sidebar))
.route("/me/threads", get(handlers::threads))
.route("/me/read-state", post(handlers::write))
.fallback(|| async { auth::Error::new(axum::http::StatusCode::NOT_FOUND, "not_found") })
.with_state(state)

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

@@ -361,6 +361,115 @@ async fn accessory_batch_outcomes_are_independent_and_return_applied_rows() {
assert_ne!(response.status(), StatusCode::OK);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn accessory_follow_intents_and_threads_view() {
let fixture = crate::api::bridge::postgres_tests::bridge_handler_test_state()
.await
.unwrap();
let mut state = (*fixture).clone();
Arc::make_mut(&mut state.config).buzz_v1_enabled = true;
let state = Arc::new(state);
let host = format!("bff-threads-{}.local", uuid::Uuid::new_v4());
let community = state
.db
.ensure_configured_community(&host)
.await
.unwrap()
.id;
let actor = Keys::generate();
let channel = state
.db
.create_channel(
community,
"threads",
buzz_db::channel::ChannelType::Stream,
buzz_db::channel::ChannelVisibility::Open,
None,
&actor.public_key().to_bytes(),
None,
)
.await
.unwrap()
.id;
start_before_everything(&state, community, &actor).await;
let mut roots = Vec::new();
for _ in 0..2 {
let root = EventBuilder::new(Kind::Custom(9), "root")
.sign_with_keys(&Keys::generate())
.unwrap();
state
.db
.insert_event(community, &root, Some(channel))
.await
.unwrap();
roots.push(root);
}
let write_path = "/buzz/v1/me/read-state";
let body = serde_json::to_vec(&json!({"intents":[
{"type":"follow","target":{"channel_id":channel}},
{"type":"follow","target":{"channel_id":channel,"root_id":roots[0].id.to_hex()}},
{"type":"follow","target":{"channel_id":channel,"root_id":roots[1].id.to_hex()}},
{"type":"unfollow","target":{"channel_id":channel,"root_id":roots[1].id.to_hex()}}
]}))
.unwrap();
let auth = proof(&actor, &host, write_path, "POST", Some(&body));
let result = request(state.clone(), &host, write_path, "POST", Some(&auth), &body).await;
assert_eq!(result.0, StatusCode::OK, "{}", result.1);
assert_eq!(
result.1["outcomes"],
json!([{"status":"invalid"},{"status":"applied"},{"status":"applied"},{"status":"applied"}])
);
assert_eq!(result.1["channels"][0]["channel_id"], json!(channel));
let mut replies = Vec::new();
for root in &roots {
let reply = EventBuilder::new(Kind::Custom(9), "reply")
.tags([Tag::parse(["e", &root.id.to_hex(), "", "root"]).unwrap()])
.sign_with_keys(&Keys::generate())
.unwrap();
let at = |e: &nostr::Event| {
chrono::DateTime::from_timestamp(e.created_at.as_secs() as i64, 0).unwrap()
};
state
.db
.insert_event_with_thread_metadata(
community,
&reply,
Some(channel),
Some(buzz_db::event::ThreadMetadataParams {
event_id: reply.id.as_bytes(),
event_created_at: at(&reply),
channel_id: channel,
parent_event_id: Some(root.id.as_bytes()),
parent_event_created_at: Some(at(root)),
root_event_id: Some(root.id.as_bytes()),
root_event_created_at: Some(at(root)),
depth: 1,
broadcast: false,
}),
)
.await
.unwrap();
replies.push(reply);
}
let path = "/buzz/v1/me/threads?limit=1";
let auth = proof(&actor, &host, path, "GET", None);
let result = request(state.clone(), &host, path, "GET", Some(&auth), b"").await;
assert_eq!(result.0, StatusCode::OK, "{}", result.1);
// Only the followed thread, unread since the follow.
assert_eq!(
result.1,
json!({"threads":[{"channel_id":channel,"root_id":roots[0].id.to_hex(),
"unread":true,"mentions":0,"read_through_id":null,
"latest_id":replies[0].id.to_hex()}],"next_cursor":null})
);
let path = "/buzz/v1/me/threads?cursor=not-a-cursor";
let auth = proof(&actor, &host, path, "GET", None);
let result = request(state.clone(), &host, path, "GET", Some(&auth), b"").await;
assert_eq!(result.0, StatusCode::BAD_REQUEST, "{}", result.1);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn accessory_write_revocation_is_terminal_before_persistence() {
@@ -495,7 +604,8 @@ async fn accessory_discovery_is_host_bound_and_opt_in() {
doc["buzz_v1"],
json!({"version":1,"base_path":"/buzz/v1","retention_seconds":1234,
"max_channels":20,"max_intents":100,
"eligible_kinds":[9, 40002, 45001, 45003]})
"eligible_kinds":[9, 40002, 45001, 45003],
"capabilities":["follow","threads"],"max_threads":50})
);
}
}

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

@@ -97,6 +97,11 @@ pub struct BuzzV1Descriptor {
/// Message kinds that count as unread and as latest activity. Clients
/// classify live arrivals with this set instead of keeping a copy.
pub eligible_kinds: [i32; 4],
/// Additions beyond the first v1 contract: `follow` (follow and unfollow
/// intents) and `threads` (`GET /me/threads`). Send one only when listed.
pub capabilities: Vec<String>,
/// Maximum followed threads per threads page.
pub max_threads: usize,
}
/// Public capability descriptor for relay-proxied GIF search.
@@ -382,7 +387,7 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st
);
if let Ok(tenant) = crate::tenant::bind_community(&state.db, raw_host).await {
if state.config.buzz_v1_enabled {
use buzz_db::personal_read::{ELIGIBLE_KINDS, MAX_CHANNELS, MAX_INTENTS};
use buzz_db::personal_read::{ELIGIBLE_KINDS, MAX_CHANNELS, MAX_INTENTS, MAX_THREADS};
info.buzz_v1 = Some(BuzzV1Descriptor {
version: 1,
base_path: crate::api::buzz_v1::BASE_PATH.to_owned(),
@@ -390,6 +395,8 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st
max_channels: MAX_CHANNELS,
max_intents: MAX_INTENTS,
eligible_kinds: ELIGIBLE_KINDS,
capabilities: vec!["follow".to_owned(), "threads".to_owned()],
max_threads: MAX_THREADS,
});
}
info.read_state_snapshot = Some(serde_json::json!({

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

@@ -12,9 +12,14 @@ 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,"eligible_kinds":[9,40002,45001,45003]}}
"max_channels":20,"max_intents":100,"eligible_kinds":[9,40002,45001,45003],
"capabilities":["follow","threads"],"max_threads":50}}
```
`capabilities` lists additions to the first v1 contract: `follow` for the
`follow` and `unfollow` intents, `threads` for `GET /me/threads`. Use one only
when it is listed.
Use the requesting origin plus this relative prefix. Discovery is a configured
capability, not a promise that the next request cannot fail. Unknown hosts and
disabled deployments omit it, and a disabled deployment does not mount
@@ -74,22 +79,23 @@ marking one never moves another.
| Field | Channel row | Thread row |
|---|---|---|
| `unread` | an unread top-level message counts | an unread reply counts |
| `mentions` | unread top-level messages that tag you with `p`, or every unread one in a DM (at most 100) | unread replies (at most 100) |
| `unread` | an unread timeline message counts | an unread reply counts |
| `mentions` | unread timeline messages that tag you with `p`, or every unread one in a DM (at most 100) | unread replies that tag you with `p`, or every unread one in a DM (at most 100) |
| `read_through_id` | the anchor of the timeline's frontier, or null | the anchor of the thread's frontier, or null |
| `latest_id` | the last eligible top-level message to arrive, or null | the last reply to arrive |
| `latest_id` | the last eligible timeline message to arrive, or null | the last reply to arrive |
A thread reply never makes the channel row unread and never becomes its
`latest_id`, including a reply carrying `broadcast=1`: it shows on its thread
row. `latest_id` is the anchor that reads its scope: "mark channel read" sends a
channel `mark_through` at the channel's `latest_id`, plus one thread
`mark_through` at each listed thread's `latest_id` to clear those too. The
channel `latest_id` counts any author and ignores read progress. Null means the
bounded scans found no top-level message, not that the channel is empty.
`read_through_id` may name a message that has since been deleted.
The timeline holds top-level messages and replies carrying `broadcast=1`; a
broadcast reply also counts on its thread row. No other reply makes the channel
row unread or becomes its `latest_id`. `latest_id` is the anchor that reads its
scope: "mark channel read" sends a channel `mark_through` at the channel's
`latest_id`, plus one thread `mark_through` at each listed thread's `latest_id`
to clear those too. The channel `latest_id` counts any author and ignores read
progress. Null means no timeline message was recorded since 0061, not that the
channel is empty. `read_through_id` may name a message that has since been
deleted.
`threads` lists the threads you follow that have unread replies, newest
unread reply first by author time (then `root_id`), at most 25. Threads past
arrival first (then `root_id`), at most 25. Threads past
the 25th are omitted with their mentions, and nothing signals the omission:
after marking the listed threads read, a refresh can list the next ones. A
thread row's `unread` is always true in this version. No message bytes are
@@ -107,7 +113,8 @@ thread you follow. The relay records follows when it stores an eligible message,
in the same transaction:
- Its author has read through it: the timeline for a top-level message,
otherwise the thread, which the author now follows.
otherwise the thread, which the author now follows. A broadcast reply reads
only its thread.
- Each channel member it tags with `p` follows its thread (for a top-level
message, the thread rooted at it).
- A reply also makes the root's author follow, and in a DM every member.
@@ -115,11 +122,16 @@ in the same transaction:
A new follower starts just before the message that made them follow, so it is
unread; one who already follows keeps their position. Reading never follows:
a thread `mark_through` moves an existing follow and otherwise changes nothing.
There is no unfollow yet. `broadcast=1` does not make anyone follow, so a
broadcast reply counts only for followers. Follows start when the relay that
`broadcast=1` does not make anyone follow. Follows start when the relay that
records them is deployed; there is no backfill. Desktop notification follows
and mutes do not affect these counts.
The `follow` intent follows a thread explicitly, from now: replies that arrived
before it are read. Following a thread you already follow changes nothing.
`unfollow` stops counting it, and the stored message rules above no longer
follow you again, with two exceptions: posting in the thread follows you through
your post, and a mention follows you from just before it.
Counts are what the bounded walks could establish, and may undercount. A
message whose ancestry is unresolved is left out, as are messages beyond the
bounds (see [Bounds](#bounds-and-deployment)).
@@ -140,6 +152,27 @@ clock, relay arrival (see [Order](#order)). Three consequences:
A later configuration expansion can change counts without having lost progress.
## Threads view
`GET /buzz/v1/me/threads?limit=50&cursor=<opaque>` lists the threads you follow
in joined, nondeleted channels, newest last reply first by arrival (then
`root_id`), read or not, at most 50 per page. Omit the cursor on the first
request and pass `next_cursor` back unchanged; null means the scan is done. A
thread with no reply yet is not listed.
```json
{"threads":[{"channel_id":"<uuid>","root_id":"<64-hex>","unread":false,
"mentions":0,"read_through_id":"<64-hex>","latest_id":"<64-hex>"}],
"next_cursor":"<opaque>"}
```
Each row has the sidebar thread row's fields and meanings plus its
`channel_id`, except that `unread` may be false and `mentions` is then 0. Pages
are separate snapshots: a thread that gets a new reply while you page moves
ahead of the cursor, so refresh from the first page after a change. Every
page probes each of your follows, so its cost grows with how many threads you
have ever followed, not with community size.
## Fixed-operand writes
`POST /buzz/v1/me/read-state` accepts 1–100 independent intents:
@@ -147,7 +180,9 @@ A later configuration expansion can change counts without having lost progress.
```json
{"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_through","target":{"channel_id":"<uuid>","root_id":"<64-hex-root>"},"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>"}}
]}
```
@@ -165,8 +200,9 @@ 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)). A channel
anchor must be a top-level message; a thread anchor must be the thread's root
or one of its replies. The frontier records the anchor as `read_through_id`
anchor must be a top-level message or a broadcast reply; a thread anchor must
be the thread's root or one of its replies. `follow` and `unfollow` need a
thread target; a channel-only one is `invalid`. The frontier records the anchor as `read_through_id`
when it advances; an anchor that arrived at or before the current frontier
changes nothing. Opening a view is not itself a reading action; client
dwell/focus policy determines when to send an actual observed anchor. Old or
@@ -195,16 +231,16 @@ microsecond resolution. Three limits follow, none of which strands a badge:
- Relay processes with different clocks can stamp out of true order by their
skew.
- Messages with the identical stamp are read together.
- A `follow` and the account's start use the database clock, so they can be
off from relay arrivals by the skew between the two.
Counting walks forward from a position. Positions are arrival times but the
event indexes are by author time, which the relay accepts within 15 minutes of
arrival, so a walk starts 15 minutes of author time before the position. A
message backdated further than that arrives unseen; Order above still holds for
what the walk finds. A channel's `latest_id` is found from the newest top-level
message by author time: it is the last top-level message to arrive among those
authored up to 30 minutes before that one, which includes the last arrival. So
marking through it reads every top-level message counted. A thread's
`latest_id` is found the same way among its replies.
Counting walks forward from a position over indexes in arrival order: the
timeline's messages per channel and each thread's replies. A channel's
`latest_id` is exactly its last timeline arrival and a thread's its last reply
arrival, so marking through it reads everything counted. Mentions are indexed
by author time, which the relay accepts within 15 minutes of arrival, so a
mention walk starts 15 minutes of author time before the position: a mention
backdated further than that is unread but not counted as a mention.
A channel row is unread exactly when its `latest_id` is someone else's, inside
the horizon, and arrived after the timeline's position. Nothing earlier needs
@@ -225,28 +261,62 @@ device-local.
a POST's rows for 100 channels fit the response limit.
- 64 KiB write body; 1 MiB serialized API response.
- Counts stop at 100 per channel and per thread.
- Each walk reads at most 4096 raw events, before eligibility. The `latest_id`
probe reads the newest 4096 per channel; a longer ineligible or reply-only
tail (reactions, edits, thread replies) leaves it null and the row read. The
last-arrival search reads the newest 4096 of the 30-minute window, so more
than that many backdated arrivals can hide one.
- A caught-up reader costs a few index probes per channel and per followed
- A caught-up reader costs one index probe per channel and per followed
thread; a reader who is behind walks only what arrived after their position.
- 50 rows per threads page; each page probes every follow of the reader.
- DB statement/lock deadlines and HTTP read deadlines bound work; a write's intents
and its row refresh share one eight-second deadline after admission. Limits are
containment, not a production capacity claim.
Apply migrations 0056 through 0058 (or the equivalent desired schema). 0056
creates two empty private tables and no index on `events`, 0057 adds the
nullable `personal_read_accounts.started_at` column, and 0058 adds the nullable
`personal_read_frontiers.through_message_id` column and drops
`threads_through_timestamp`. Timeline walks use the existing
`idx_events_community_channel_created`, thread walks
`idx_thread_metadata_window`, and mention counts
`idx_event_mentions_pubkey_created`. Ingest writes a few frontier rows per
eligible channel message (the author, and anyone it makes follow), whether or
not the API is enabled, so follows exist once it is; there are no stored unread
counters.
Apply migrations 0056 through 0058 and 0061 (or the equivalent desired
schema). 0056 creates two empty private tables and no index on `events`, 0057
adds the nullable `personal_read_accounts.started_at` column, and 0058 adds the
nullable `personal_read_frontiers.through_message_id` column and drops
`threads_through_timestamp`. 0061 adds `events.read_timeline` (set at insert
for timeline messages), `thread_metadata.received_at` (set at insert for
eligible replies) and `personal_read_frontiers.unfollowed`, plus three indexes:
`idx_events_read_timeline` for timeline walks, `idx_thread_metadata_read_arrival`
for thread walks and `idx_event_mentions_pubkey_channel_created` for mention
counts. There is no backfill: messages stored before the new relay serves,
including by older relay processes still running after 0061, are on neither
arrival index, so they never count as unread or become `latest_id`. Ingest
writes a few frontier rows per eligible channel message (the author, and anyone
it makes follow), whether or not the API is enabled, so follows exist once it
is; there are no stored unread counters.
The column additions are catalog-only but briefly take an exclusive lock on
`events`, `thread_metadata` and `personal_read_frontiers`, bounded by the
migration's one-second `lock_timeout`; a busy writer fails the deployment,
which retries, rather than stalling ingestion. Building the indexes inside the
migration is bounded the same way, so on a database with data, add the columns
and prebuild the indexes before deploying, as for
[0049](thread-window-deployment.md). 0061 then skips both and only validates
the indexes. Run each statement on its own, retrying any that times out;
`CREATE INDEX CONCURRENTLY` cannot run in a transaction block:
```sql
SET lock_timeout = '1s';
ALTER TABLE events ADD COLUMN IF NOT EXISTS read_timeline BOOLEAN NOT NULL DEFAULT false;
ALTER TABLE thread_metadata ADD COLUMN IF NOT EXISTS received_at TIMESTAMPTZ;
ALTER TABLE personal_read_frontiers ADD COLUMN IF NOT EXISTS unfollowed BOOLEAN NOT NULL
DEFAULT false CHECK (NOT unfollowed OR root_id <> ''::bytea);
CREATE INDEX CONCURRENTLY idx_thread_metadata_read_arrival ON public.thread_metadata
(community_id, root_event_id, received_at DESC, event_id) WHERE received_at IS NOT NULL;
CREATE INDEX CONCURRENTLY idx_event_mentions_pubkey_channel_created ON public.event_mentions
(community_id, pubkey_hex, channel_id, event_created_at DESC);
-- events is partitioned: an invalid parent, then each partition concurrently.
CREATE INDEX idx_events_read_timeline ON ONLY public.events
(community_id, channel_id, received_at DESC, id) WHERE read_timeline AND deleted_at IS NULL;
-- For each partition p in
-- SELECT inhrelid::regclass FROM pg_inherits WHERE inhparent = 'public.events'::regclass:
CREATE INDEX CONCURRENTLY p_read_timeline ON p
(community_id, channel_id, received_at DESC, id) WHERE read_timeline AND deleted_at IS NULL;
ALTER INDEX idx_events_read_timeline ATTACH PARTITION p_read_timeline;
```
The parent becomes valid once every partition's index is attached. Verify the
three like 0049's, against the definitions the migration expects; recover a
failed build the same way.
Use existing HTTP route/status/latency metrics for `/buzz/v1/me/sidebar` and
`/buzz/v1/me/read-state`, plus database pool/statement metrics. No payload,

68
migrations/0061_read_state_access_paths.sql Normal file
View File

@@ -0,0 +1,68 @@
-- Read-state access paths: one index seek answers "newest countable message
-- here" and "what arrived after this position" for a channel timeline or a
-- thread, and "my mentions in this channel since then". Arrival order
-- (received_at), not author time. No backfill: rows written before this
-- migration are not on the new paths.
--
-- events.read_timeline: an eligible message that counts on its channel
-- timeline (top-level, or a broadcast reply), set at insert.
-- thread_metadata.received_at: a reply's arrival, set at insert; NULL on root
-- stubs and earlier rows.
-- personal_read_frontiers.unfollowed: a sticky unfollow of a thread.
--
-- Brownfield operators MUST add the three columns and prebuild the three
-- indexes concurrently as documented in docs/buzz-v1-read-state.md. Startup is
-- bounded like 0049: a large table fails deployment rather than blocking
-- ingestion.
SET LOCAL lock_timeout = '1s';
SET LOCAL statement_timeout = '5s';
DO $$
DECLARE
spec RECORD;
BEGIN
-- Even ADD COLUMN IF NOT EXISTS requests a writer-conflicting lock: skip
-- a column an operator added beforehand.
FOR spec IN SELECT * FROM (VALUES
('events', 'read_timeline', 'BOOLEAN NOT NULL DEFAULT false'),
('thread_metadata', 'received_at', 'TIMESTAMPTZ'),
('personal_read_frontiers', 'unfollowed',
$d$BOOLEAN NOT NULL DEFAULT false CHECK (NOT unfollowed OR root_id <> ''::bytea)$d$)
) AS s(tbl, col, definition) LOOP
IF NOT EXISTS (
SELECT 1 FROM pg_attribute
WHERE attrelid = ('public.' || spec.tbl)::regclass
AND attname = spec.col AND NOT attisdropped
) THEN
EXECUTE format('ALTER TABLE public.%I ADD COLUMN %I %s',
spec.tbl, spec.col, spec.definition);
END IF;
END LOOP;
-- A partitioned parent's definition reads ON ONLY; CREATE recurses to
-- every partition regardless.
FOR spec IN SELECT * FROM (VALUES
('idx_events_read_timeline',
'CREATE INDEX idx_events_read_timeline ON ONLY public.events USING btree (community_id, channel_id, received_at DESC, id) WHERE (read_timeline AND (deleted_at IS NULL))'),
('idx_thread_metadata_read_arrival',
'CREATE INDEX idx_thread_metadata_read_arrival ON public.thread_metadata USING btree (community_id, root_event_id, received_at DESC, event_id) WHERE (received_at IS NOT NULL)'),
('idx_event_mentions_pubkey_channel_created',
'CREATE INDEX idx_event_mentions_pubkey_channel_created ON public.event_mentions USING btree (community_id, pubkey_hex, channel_id, event_created_at DESC)')
) AS s(name, definition) LOOP
-- IF NOT EXISTS still requests a writer-conflicting lock: skip CREATE
-- for a prebuilt index, then validate it.
IF to_regclass('public.' || spec.name) IS NULL THEN
EXECUTE replace(spec.definition, ' ON ONLY ', ' ON ');
END IF;
IF NOT EXISTS (
SELECT 1 FROM pg_index i
JOIN pg_class c ON c.oid = i.indexrelid
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname = 'public' AND c.relname = spec.name
AND i.indisvalid AND i.indisready AND i.indislive
AND pg_get_indexdef(i.indexrelid) = spec.definition
) THEN
RAISE EXCEPTION '% invalid or wrong definition; see docs/buzz-v1-read-state.md', spec.name;
END IF;
END LOOP;
END $$;

2
schema/tables/public/event_mentions.sql
View File

@@ -18,5 +18,7 @@ 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_pubkey_channel_created
ON event_mentions (community_id, pubkey_hex, channel_id, event_created_at DESC);
CREATE INDEX idx_event_mentions_community_event
ON event_mentions (community_id, event_id);

6
schema/tables/public/events.sql
View File

@@ -37,6 +37,8 @@ CREATE TABLE events (
d_tag TEXT,
not_before BIGINT,
delivered_at BIGINT,
-- An eligible message that counts on its channel's read-state timeline.
read_timeline BOOLEAN NOT NULL DEFAULT false,
PRIMARY KEY (community_id, created_at, id)
) PARTITION BY RANGE (created_at);
@@ -52,6 +54,10 @@ CREATE INDEX idx_events_community_pubkey_kind_created
CREATE INDEX idx_events_community_kind_created
ON events (community_id, kind, created_at DESC, id);
CREATE INDEX idx_events_community_deleted ON events (community_id, deleted_at);
-- Read state: newest timeline message, and what arrived after a position.
CREATE INDEX idx_events_read_timeline
ON events (community_id, channel_id, received_at DESC, id)
WHERE read_timeline AND deleted_at IS NULL;
-- Addressable (replaceable) and NIP-33 parameterized lookups.
CREATE INDEX idx_events_addressable
ON events (community_id, kind, pubkey, channel_id, deleted_at);

2
schema/tables/public/personal_read_frontiers.sql
View File

@@ -6,6 +6,8 @@ CREATE TABLE personal_read_frontiers (
through_timestamp TIMESTAMPTZ NOT NULL,
-- The message whose arrival is through_timestamp; NULL before 0058.
through_message_id BYTEA CHECK (octet_length(through_message_id) = 32),
-- A sticky thread unfollow; thread rows only.
unfollowed BOOLEAN NOT NULL DEFAULT false CHECK (NOT unfollowed 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,

5
schema/tables/public/thread_metadata.sql
View File

@@ -15,6 +15,8 @@ CREATE TABLE thread_metadata (
descendant_count INT NOT NULL DEFAULT 0,
last_reply_at TIMESTAMPTZ,
broadcast BOOLEAN NOT NULL DEFAULT FALSE,
-- A reply's relay arrival; NULL on root stubs and rows before 0061.
received_at TIMESTAMPTZ,
PRIMARY KEY (community_id, event_created_at, event_id),
FOREIGN KEY (community_id, channel_id) REFERENCES channels (community_id, id)
);
@@ -26,3 +28,6 @@ CREATE INDEX idx_thread_metadata_window
CREATE INDEX idx_thread_metadata_channel_depth
ON thread_metadata (community_id, channel_id, depth, event_created_at);
CREATE INDEX idx_thread_metadata_event_id ON thread_metadata (community_id, event_id);
CREATE INDEX idx_thread_metadata_read_arrival
ON thread_metadata (community_id, root_event_id, received_at DESC, event_id)
WHERE received_at IS NOT NULL;