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

relay: prototype unread counts by relay-assigned sequence numbers

Browse Source 
Not for merge. Side-by-side comparison with #8138: ingest numbers each
eligible message under a per-channel counter row, unread is the counter
less the reader's position, posting marks the author read, deleted
messages count until read past. Includes a shared benchmark test.

Signed-off-by: hMurderbot <219c58afff8ac2c76e4e527a329aa61c2ebbb896786b1916d4f9cca5140ecd04@buzz.block.builderlab.xyz>
This commit is contained in:
hMurderbot
2026-10-06 19:54:55 -07:00
parent c0bf08ff18
commit 0e0b02520f
10 changed files with 737 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -93,6 +93,7 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[
"moderation_reports",
"parameterized_event_watermarks",
"personal_read_accounts",
"personal_read_counters",
"personal_read_frontiers",
"pubkey_allowlist",
"push_leases",
@@ -113,6 +114,7 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[
/// Foreign-key-safe child-before-parent order for the PostgreSQL purge.
pub const PURGE_SCOPED_TABLES: &[&str] = &[
"personal_read_frontiers",
"personal_read_counters",
"personal_read_accounts",
"workflow_approvals",
"scheduled_workflow_fires",

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

@@ -477,10 +477,14 @@ async fn insert_event_on(
let received_at = Utc::now();
let d_tag = extract_d_tag(event);
let not_before = extract_not_before(event);
// No thread metadata: on the timeline.
let numbers =
crate::personal_read::reserve(&mut *connection, community_id, event, channel_id, true)
.await?;
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, channel_seq, timeline_seq)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
ON CONFLICT DO NOTHING
"#,
)
@@ -496,10 +500,16 @@ async fn insert_event_on(
.bind(channel_id)
.bind(d_tag.as_deref())
.bind(not_before)
.execute(connection)
.bind(numbers.map(|n| n.channel))
.bind(numbers.map(|n| n.timeline))
.execute(&mut *connection)
.await?;
let was_inserted = result.rows_affected() > 0;
if let (true, Some(n), Some(channel)) = (was_inserted, numbers, channel_id) {
crate::personal_read::commit(connection, community_id, event, channel, n, received_at)
.await?;
}
Ok((
StoredEvent::with_received_at(event.clone(), received_at, channel_id, true),
@@ -1580,10 +1590,17 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
let d_tag = extract_d_tag(event);
let not_before = extract_not_before(event);
let on_timeline = thread_meta
.as_ref()
.is_none_or(|m| m.parent_event_id.is_none() || (m.depth == 1 && m.broadcast));
let numbers =
crate::personal_read::reserve(&mut **tx, community_id, event, channel_id, on_timeline)
.await?;
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, channel_seq, timeline_seq)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
ON CONFLICT DO NOTHING
"#,
)
@@ -1599,12 +1616,18 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
.bind(channel_id)
.bind(d_tag.as_deref())
.bind(not_before)
.bind(numbers.map(|n| n.channel))
.bind(numbers.map(|n| n.timeline))
.execute(&mut **tx)
.await?;
let was_inserted = result.rows_affected() > 0;
if was_inserted {
if let (Some(n), Some(channel)) = (numbers, channel_id) {
crate::personal_read::commit(&mut **tx, community_id, event, channel, n, received_at)
.await?;
}
if let Some(ref meta) = thread_meta {
let broadcast_val: bool = meta.broadcast;
@@ -1696,29 +1719,45 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
.execute(&mut **tx)
.await?;
// An eligible reply takes the thread's next number unless it
// counts on the timeline instead (broadcast depth-1).
let counted = numbers.filter(|_| !(meta.depth == 1 && meta.broadcast));
let mut thread_seq = None;
if let Some(root_id) = meta.root_event_id {
sqlx::query(
thread_seq = sqlx::query_scalar::<_, i64>(
r#"
UPDATE thread_metadata
SET descendant_count = descendant_count + 1
SET descendant_count = descendant_count + 1,
reply_seq = reply_seq + $3::int,
reply_channel_seq = CASE WHEN $3 = 1 THEN $4 ELSE reply_channel_seq END,
reply_last_id = CASE WHEN $3 = 1 THEN $5 ELSE reply_last_id END
WHERE community_id = $1 AND event_id = $2
RETURNING reply_seq
"#,
)
.bind(community_id.as_uuid())
.bind(root_id)
.execute(&mut **tx)
.bind(i32::from(counted.is_some()))
.bind(counted.map(|n| n.channel))
.bind(meta.event_id)
.fetch_optional(&mut **tx)
.await?;
}
let root = meta.root_event_id.unwrap_or(pid);
crate::personal_read::record_reply(
tx,
community_id,
event,
meta.channel_id,
root,
received_at,
)
.await?;
if let (Some(n), Some(seq)) = (numbers, thread_seq) {
crate::personal_read::record_reply(
tx,
community_id,
event,
meta.channel_id,
root,
n,
seq,
counted.is_some(),
received_at,
)
.await?;
}
}
}
}

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

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

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

@@ -2,7 +2,6 @@
use super::{classification, model::*, writes};
use crate::{observability, Db, DbError, Result};
use buzz_core::CommunityId;
use chrono::{DateTime, Utc};
use sqlx::{Acquire, Row};
use std::collections::HashMap;
use uuid::Uuid;
@@ -47,17 +46,15 @@ impl Db {
contexts.push(ContextState::Unavailable);
continue;
};
// The positions the sidebar counts against, both floored at the
// account's start and absent before it: the channel's, never
// before joining and absent for a non-member; and the thread's,
// which exists only for the actor's threads and includes any
// whole-channel cut.
// The channel-wide cuts the sidebar subtracts against, absent
// before the account starts: the channel's for a member, and the
// thread's, which exists only for the actor's threads. A cut in
// channel numbering orders every message in either context.
let positions = sqlx::query(
"SELECT CASE WHEN m.joined_at IS NOT NULL AND s.started IS NOT NULL
THEN GREATEST(cf.through_timestamp, m.joined_at, s.started) END AS channel,
THEN COALESCE(cf.through_channel_seq, 0) END AS channel,
CASE WHEN tf.root_id IS NOT NULL AND s.started IS NOT NULL
THEN GREATEST(tf.through_timestamp, cf.threads_through_timestamp, s.started)
END AS thread
THEN tf.through_channel_seq END AS thread
FROM (SELECT (SELECT started_at FROM personal_read_accounts
WHERE community_id=$1 AND actor=$2) AS started) s
LEFT JOIN channel_members m ON m.community_id=$1 AND m.channel_id=$3
@@ -73,15 +70,15 @@ impl Db {
.bind(&root)
.fetch_one(&mut *tx)
.await?;
let channel_position: Option<DateTime<Utc>> = positions.try_get("channel")?;
let thread_position: Option<DateTime<Utc>> = positions.try_get("thread")?;
let channel_position: Option<i64> = positions.try_get("channel")?;
let thread_position: Option<i64> = positions.try_get("thread")?;
let ids: Vec<Vec<u8>> = query
.message_ids
.iter()
.filter_map(|id| writes::event_id(id))
.collect();
let rows = sqlx::query(
"SELECT encode(e.id,'hex') AS id,e.kind,e.received_at,
"SELECT encode(e.id,'hex') AS id,e.kind,e.channel_seq,
e.deleted_at IS NOT NULL AS deleted,e.pubkey=$3 AS own,e.tags,
tm.root_event_id,tm.depth,tm.broadcast,c.channel_type::text AS channel_type
FROM events e JOIN channels c ON c.community_id=e.community_id AND c.id=e.channel_id
@@ -133,7 +130,8 @@ impl Db {
} else {
thread_position
};
let received: DateTime<Utc> = row.try_get("received_at")?;
// Unnumbered messages predate numbering: read.
let seq: Option<i64> = row.try_get("channel_seq")?;
let tags: Vec<Vec<String>> =
serde_json::from_value(row.try_get("tags")?).unwrap_or_default();
let reason = classification::reason(
@@ -143,7 +141,7 @@ impl Db {
);
match position {
None => MessageReadState::NotCounted,
Some(p) if received <= p => MessageReadState::Read,
Some(p) if seq.is_none_or(|n| n <= p) => MessageReadState::Read,
Some(_) if on_timeline => MessageReadState::Unread { reason },
Some(_) => MessageReadState::Unread {
reason: match reason {

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

@@ -1,58 +1,157 @@
//! Thread membership. A thread frontier row exists exactly for the actor's
//! threads, and only ingest creates one: read intents move existing rows.
//! Ingest's share of read state: numbering each eligible message under its
//! channel's counter, posting marking the author's position, and thread
//! membership. A thread frontier row exists exactly for the actor's threads,
//! and only ingest creates one: read intents move existing rows.
use super::model::ELIGIBLE_KINDS;
use crate::Result;
use buzz_core::CommunityId;
use chrono::{DateTime, Utc};
use nostr::Event;
use sqlx::{Postgres, Transaction};
use sqlx::{PgConnection, Postgres, Row, Transaction};
use uuid::Uuid;
/// The numbers a message takes if it is stored.
#[derive(Clone, Copy, Debug)]
pub(crate) struct Numbers {
/// The channel's timeline count once stored: its own number on the timeline.
pub timeline: i64,
/// Its number among every eligible message in the channel.
pub channel: i64,
/// Shown on the timeline: top-level, or a broadcast depth-1 reply.
pub on_timeline: bool,
}
/// Lock the channel's counter until commit and return the numbers an eligible
/// message would take. The lock is what makes numbers follow commit order.
pub(crate) async fn reserve(
conn: &mut PgConnection,
community: CommunityId,
event: &Event,
channel: Option<Uuid>,
on_timeline: bool,
) -> Result<Option<Numbers>> {
let Some(channel) = channel else {
return Ok(None);
};
if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) {
return Ok(None);
}
let row = sqlx::query(
"INSERT INTO personal_read_counters (community_id, channel_id) VALUES ($1, $2)
ON CONFLICT (community_id, channel_id)
DO UPDATE SET channel_seq = personal_read_counters.channel_seq
RETURNING timeline_seq, channel_seq",
)
.bind(community.as_uuid())
.bind(channel)
.fetch_one(&mut *conn)
.await?;
Ok(Some(Numbers {
timeline: row.try_get::<i64, _>("timeline_seq")? + i64::from(on_timeline),
channel: row.try_get::<i64, _>("channel_seq")? + 1,
on_timeline,
}))
}
/// Advance the counter for a stored message. Posting reads the author's
/// timeline through it once the author has started.
pub(crate) async fn commit(
conn: &mut PgConnection,
community: CommunityId,
event: &Event,
channel: Uuid,
numbers: Numbers,
received_at: DateTime<Utc>,
) -> Result<()> {
sqlx::query(
"UPDATE personal_read_counters SET timeline_seq=$3, channel_seq=$4, latest_id=$5
WHERE community_id=$1 AND channel_id=$2",
)
.bind(community.as_uuid())
.bind(channel)
.bind(numbers.timeline)
.bind(numbers.channel)
.bind(event.id.as_bytes().as_slice())
.execute(&mut *conn)
.await?;
if numbers.on_timeline {
sqlx::query(
"INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
through_seq, through_channel_seq, through_timestamp)
SELECT $1, $2, $3, $4, $5, $6 FROM personal_read_accounts
WHERE community_id=$1 AND actor=$2 AND started_at IS NOT NULL
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
through_seq=GREATEST(personal_read_frontiers.through_seq, EXCLUDED.through_seq),
through_channel_seq=GREATEST(personal_read_frontiers.through_channel_seq,
EXCLUDED.through_channel_seq),
through_timestamp=GREATEST(personal_read_frontiers.through_timestamp,
EXCLUDED.through_timestamp)",
)
.bind(community.as_uuid())
.bind(event.pubkey.to_bytes().as_slice())
.bind(channel)
.bind(numbers.timeline)
.bind(numbers.channel)
.bind(received_at)
.execute(&mut *conn)
.await?;
}
Ok(())
}
/// Record who a newly stored reply makes a thread member: its author, the
/// root's author, members it mentions, and in a DM (at most nine members)
/// every member. The author's row starts at the reply, which it has read;
/// everyone else's starts just before it, so the reply is unread. Existing
/// rows are left alone.
/// every member. `thread_seq` is the thread's count once the reply is stored.
/// The author reads through the reply (posting marks read, also on an
/// existing row); everyone else's new row starts just before it, so the reply
/// is unread. Other existing rows are left alone.
pub(crate) async fn record_reply(
tx: &mut Transaction<'_, Postgres>,
community: CommunityId,
event: &Event,
channel: Uuid,
root: &[u8],
numbers: Numbers,
thread_seq: i64,
counts_in_thread: bool,
received_at: DateTime<Utc>,
) -> Result<()> {
if !ELIGIBLE_KINDS.contains(&i32::from(event.kind.as_u16())) {
return Ok(());
}
let mentioned: Vec<Vec<u8>> = event
.tags
.public_keys()
.map(|key| key.to_bytes().to_vec())
.collect();
let before = thread_seq - i64::from(counts_in_thread);
// Both inserts' fence and foreign-key checks run at the end of the
// statement, after the account rows exist.
sqlx::query(
"WITH joined AS (
SELECT $4::bytea AS actor, $6::timestamptz AS through
SELECT $4::bytea AS actor, $7::bigint AS seq, $8::bigint AS cseq, $6::timestamptz AS at
UNION ALL
SELECT pubkey, $6-interval '1 microsecond' FROM events
SELECT pubkey, $9, $8-1, $6-interval '1 microsecond' FROM events
WHERE community_id=$1 AND channel_id=$2 AND id=$3
UNION ALL
SELECT cm.pubkey, $6-interval '1 microsecond'
SELECT cm.pubkey, $9, $8-1, $6-interval '1 microsecond'
FROM channel_members cm JOIN channels c
ON c.community_id=cm.community_id AND c.id=cm.channel_id
WHERE cm.community_id=$1 AND cm.channel_id=$2 AND cm.removed_at IS NULL
AND (cm.pubkey=ANY($5) OR c.channel_type='dm')
), members AS (
SELECT DISTINCT ON (actor) actor, through FROM joined ORDER BY actor, through DESC
SELECT DISTINCT ON (actor) actor, seq, cseq, at FROM joined ORDER BY actor, cseq DESC
), accounts AS (
INSERT INTO personal_read_accounts (community_id, actor)
SELECT $1, actor FROM members ON CONFLICT DO NOTHING
)
INSERT INTO personal_read_frontiers
(community_id, actor, channel_id, root_id, through_timestamp)
SELECT $1, actor, $2, $3, through FROM members
ON CONFLICT DO NOTHING",
INSERT INTO personal_read_frontiers (community_id, actor, channel_id, root_id,
through_seq, through_channel_seq, through_timestamp)
SELECT $1, actor, $2, $3, seq, cseq, at FROM members
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
through_seq=GREATEST(personal_read_frontiers.through_seq, EXCLUDED.through_seq),
through_channel_seq=GREATEST(personal_read_frontiers.through_channel_seq,
EXCLUDED.through_channel_seq),
through_timestamp=GREATEST(personal_read_frontiers.through_timestamp,
EXCLUDED.through_timestamp)
WHERE personal_read_frontiers.actor=$4",
)
.bind(community.as_uuid())
.bind(channel)
@@ -60,6 +159,9 @@ pub(crate) async fn record_reply(
.bind(event.pubkey.to_bytes().as_slice())
.bind(&mentioned)
.bind(received_at)
.bind(thread_seq)
.bind(numbers.channel)
.bind(before)
.execute(&mut **tx)
.await?;
Ok(())

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

@@ -11,7 +11,7 @@ mod model;
mod projection;
mod writes;
pub(crate) use membership::record_reply;
pub(crate) use membership::{commit, record_reply, reserve};
pub use model::*;
#[cfg(test)]
@@ -25,3 +25,6 @@ mod threads_postgres_tests;
#[cfg(test)]
mod arrival_postgres_tests;
#[cfg(test)]
mod bench_postgres_tests;

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

@@ -71,89 +71,78 @@ impl Db {
.await?;
writes::deadlines(&mut tx).await?;
sqlx::query("SET LOCAL jit = off").execute(&mut *tx).await?;
// Every position is floored at the account's start (infinity before
// the actor's first intent, so nothing counts). A channel's is its
// frontier, never earlier than joining: a new member starts caught
// up. A thread row exists only for the actor's threads (see
// `membership`), and a whole-channel cut covers them all. Each count
// ranges over author time from the position less the arrival skew
// (served by idx_events_community_channel_created or
// idx_thread_metadata_window), filters on arrival, and stops at the
// cap: rows the filters reject are read but bound nothing. The latest message is the last arrival among
// the newest messages and everything counted, so marking through it
// reads every counted message whatever its author time.
// Unread is subtraction: the channel's counter less the actor's
// position (starting and joining write one, so a started member
// without a row has read nothing), and each of the actor's threads' reply count less its
// position. Nothing counts before the account starts. Attention needs
// the messages themselves, so it walks forward from the position
// through the numbered messages after it, only when something is
// unread, and stops at the cap. A whole-channel mark through the
// counter's latest message reads every counted message.
let sql = format!(
r#"WITH roster AS MATERIALIZED (
SELECT c.id, c.name, c.channel_type::text AS channel_type,
c.archived_at IS NOT NULL AS archived, cm.hidden_at IS NOT NULL AS hidden,
GREATEST(cf.through_timestamp, cm.joined_at, s.started) AS position,
GREATEST(cf.threads_through_timestamp, s.started) AS threads_floor
s.started IS NOT NULL AS started,
COALESCE(k.timeline_seq, 0) AS timeline_seq,
COALESCE(cf.through_seq, 0) AS through_seq,
COALESCE(cf.through_channel_seq, 0) AS through_channel_seq,
COALESCE(cf.through_timestamp, '-infinity') AS position,
encode(k.latest_id, 'hex') AS latest_id
FROM channel_members cm JOIN channels c
ON c.community_id=cm.community_id AND c.id=cm.channel_id
CROSS JOIN (SELECT COALESCE((SELECT started_at FROM personal_read_accounts
WHERE community_id=$1 AND actor=$2), 'infinity') AS started) s
CROSS JOIN (SELECT (SELECT started_at FROM personal_read_accounts
WHERE community_id=$1 AND actor=$2) AS started) s
LEFT JOIN personal_read_counters k ON k.community_id=cm.community_id
AND k.channel_id=c.id
LEFT JOIN personal_read_frontiers cf ON cf.community_id=cm.community_id
AND cf.actor=cm.pubkey 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 $5
), counted AS (
SELECT r.*, CASE WHEN r.started
THEN GREATEST(r.timeline_seq - r.through_seq, 0) ELSE 0 END AS unread
FROM roster r
)
SELECT r.id, r.name, r.channel_type, r.archived, r.hidden,
(SELECT v.id FROM (VALUES (latest.arrival, latest.id),
(timeline.arrival, timeline.id), (threads.arrival, threads.id)) v(arrival, id)
WHERE v.id IS NOT NULL ORDER BY v.arrival DESC, v.id LIMIT 1) AS latest_message_id,
latest.at AS latest_message_at,
COALESCE(timeline.unread, 0) AS unread, COALESCE(timeline.attention, 0) AS attention,
r.latest_id AS latest_message_id, latest.at AS latest_message_at,
LEAST(r.unread, $9)::int AS unread,
CASE WHEN r.channel_type='dm' THEN LEAST(r.unread, $9)::int
ELSE COALESCE(attention.n, 0) END AS attention,
COALESCE(threads.items, '[]') AS threads
FROM roster r
FROM counted r
LEFT JOIN LATERAL (
SELECT (array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id,
max(received_at) AS arrival, max(extract(epoch FROM created_at)::bigint) AS at
FROM (SELECT id,created_at,received_at FROM events
SELECT max(extract(epoch FROM created_at)::bigint) AS at
FROM (SELECT created_at FROM events
WHERE community_id=$1 AND channel_id=r.id AND kind=ANY($6) AND deleted_at IS NULL
ORDER BY created_at DESC LIMIT $7) newest
) latest ON true
LEFT JOIN LATERAL (
SELECT count(*)::int AS unread, count(*) FILTER (WHERE directed)::int AS attention,
(array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id,
max(received_at) AS arrival
FROM (
SELECT e.id, e.received_at, r.channel_type='dm' OR {DIRECTED_TAGS} AS directed
FROM events e
SELECT count(*)::int AS n FROM (
SELECT 1 FROM events e
LEFT JOIN thread_metadata tm ON tm.community_id=$1
AND tm.event_created_at=e.created_at AND tm.event_id=e.id
WHERE e.community_id=$1 AND e.channel_id=r.id
AND e.created_at >= r.position-$8 AND e.received_at > r.position
AND e.kind=ANY($6) AND e.pubkey<>$2 AND e.deleted_at IS NULL AND {ON_TIMELINE}
WHERE r.unread > 0 AND r.channel_type<>'dm'
AND e.community_id=$1 AND e.channel_id=r.id
AND e.created_at >= r.position-$8 AND e.channel_seq > r.through_channel_seq
AND e.kind=ANY($6) AND e.pubkey<>$2 AND {ON_TIMELINE} AND {DIRECTED_TAGS}
ORDER BY e.created_at LIMIT $9
) counted
) timeline ON true
) directed
) attention ON true
LEFT JOIN LATERAL (
SELECT jsonb_agg(jsonb_build_object('root_id',encode(tf.root_id,'hex'),
'unread',n.unread,'latest_reply_id',n.id,'latest_reply_at',n.at)) AS items,
(array_agg(n.id ORDER BY n.arrival DESC,n.id))[1] AS id, max(n.arrival) AS arrival
'unread',LEAST(root.reply_seq - tf.through_seq, $9),
'latest_reply_id',encode(root.reply_last_id,'hex'),
'latest_reply_at',extract(epoch FROM last.event_created_at)::bigint)) AS items
FROM personal_read_frontiers tf
CROSS JOIN LATERAL (SELECT GREATEST(tf.through_timestamp, r.threads_floor) AS position) p
CROSS JOIN LATERAL (
SELECT count(*)::int AS unread,
(array_agg(encode(id,'hex') ORDER BY received_at DESC,id))[1] AS id,
max(extract(epoch FROM created_at)::bigint) AS at,
max(received_at) AS arrival
FROM (
SELECT e.id, e.created_at, e.received_at
FROM thread_metadata tm JOIN events e ON e.community_id=$1
AND e.created_at=tm.event_created_at AND e.id=tm.event_id
WHERE tm.community_id=$1 AND tm.root_event_id=tf.root_id
AND tm.channel_id=r.id AND tm.event_id<>tf.root_id
AND tm.event_created_at >= p.position-$8
AND e.received_at > p.position AND e.kind=ANY($6) AND e.pubkey<>$2
AND e.deleted_at IS NULL AND NOT (tm.depth=1 AND tm.broadcast)
ORDER BY tm.event_created_at LIMIT $9
) counted
) n
WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=r.id
AND tf.root_id<>''::bytea AND n.unread > 0
JOIN thread_metadata root ON root.community_id=$1 AND root.event_id=tf.root_id
AND root.depth=0
CROSS JOIN LATERAL (SELECT event_created_at FROM thread_metadata
WHERE community_id=$1 AND event_id=root.reply_last_id LIMIT 1) last
WHERE r.started AND tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=r.id
AND tf.root_id<>''::bytea AND root.reply_seq > tf.through_seq
) threads ON true
ORDER BY r.id"#
)

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

@@ -26,28 +26,60 @@ pub(super) async fn deadlines(conn: &mut PgConnection) -> Result<()> {
/// Start the account if this is the actor's first read intent, then serialize
/// private frontier writes, never shared conversation rows. NO KEY UPDATE
/// leaves ingest's foreign-key checks (which create thread rows) unblocked.
/// Starting reads every joined channel and existing thread through its
/// current count.
pub(super) async fn lock_account(
conn: &mut PgConnection,
community: CommunityId,
actor: &[u8],
) -> Result<()> {
sqlx::query(
let started = sqlx::query(
"INSERT INTO personal_read_accounts (community_id,actor,started_at) VALUES ($1,$2,now())
ON CONFLICT (community_id,actor) DO UPDATE SET started_at=now()
WHERE personal_read_accounts.started_at IS NULL",
WHERE personal_read_accounts.started_at IS NULL RETURNING actor",
)
.bind(community.as_uuid())
.bind(actor)
.execute(&mut *conn)
.await?;
.fetch_optional(&mut *conn)
.await?
.is_some();
sqlx::query(
"SELECT actor FROM personal_read_accounts WHERE community_id=$1 AND actor=$2
FOR NO KEY UPDATE",
)
.bind(community.as_uuid())
.bind(actor)
.fetch_one(conn)
.fetch_one(&mut *conn)
.await?;
if started {
sqlx::query(
"INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
through_seq, through_channel_seq, through_timestamp)
SELECT $1, $2, cm.channel_id, COALESCE(k.timeline_seq,0), COALESCE(k.channel_seq,0), now()
FROM channel_members cm LEFT JOIN personal_read_counters k
ON k.community_id=cm.community_id AND k.channel_id=cm.channel_id
WHERE cm.community_id=$1 AND cm.pubkey=$2 AND cm.removed_at IS NULL
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
through_seq=EXCLUDED.through_seq, through_channel_seq=EXCLUDED.through_channel_seq,
through_timestamp=EXCLUDED.through_timestamp",
)
.bind(community.as_uuid())
.bind(actor)
.execute(&mut *conn)
.await?;
sqlx::query(
"UPDATE personal_read_frontiers tf SET through_seq=root.reply_seq,
through_channel_seq=COALESCE(root.reply_channel_seq, tf.through_channel_seq),
through_timestamp=now()
FROM thread_metadata root
WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.root_id<>''::bytea
AND root.community_id=$1 AND root.event_id=tf.root_id AND root.depth=0",
)
.bind(community.as_uuid())
.bind(actor)
.execute(&mut *conn)
.await?;
}
Ok(())
}
@@ -87,6 +119,9 @@ async fn access(
struct Message {
id: Vec<u8>,
received_at: DateTime<Utc>,
/// Numbers, absent for messages stored before numbering.
timeline_seq: Option<i64>,
channel_seq: Option<i64>,
/// Canonical thread root, when the message is a reply.
thread: Option<Vec<u8>>,
/// Shown on the channel timeline: not a reply, or a broadcast depth-1 reply.
@@ -103,7 +138,8 @@ async fn message(
kinds: Option<&[i32]>,
) -> Result<Option<Message>> {
let row = sqlx::query(
"SELECT e.id, e.received_at, tm.root_event_id, tm.depth, tm.broadcast
"SELECT e.id, e.received_at, e.timeline_seq, e.channel_seq,
tm.root_event_id, tm.depth, tm.broadcast
FROM events e LEFT JOIN thread_metadata tm ON tm.community_id=e.community_id
AND tm.event_created_at=e.created_at AND tm.event_id=e.id AND tm.channel_id=e.channel_id
WHERE e.community_id=$1 AND e.channel_id=$2 AND e.id=$3
@@ -119,6 +155,8 @@ async fn message(
Ok(Message {
on_timeline: thread.is_none() || broadcast_reply,
received_at: row.try_get("received_at")?,
timeline_seq: row.try_get("timeline_seq")?,
channel_seq: row.try_get("channel_seq")?,
thread,
id,
})
@@ -126,25 +164,6 @@ async fn message(
.transpose()
}
/// An eligible-kind message's arrival time, deleted or not, without tags.
async fn anchor_received_at(
conn: &mut PgConnection,
community: CommunityId,
channel: Uuid,
id: &[u8],
) -> Result<Option<DateTime<Utc>>> {
Ok(sqlx::query_scalar(
"SELECT received_at FROM events
WHERE community_id=$1 AND channel_id=$2 AND id=$3 AND kind=ANY($4) LIMIT 1",
)
.bind(community.as_uuid())
.bind(channel)
.bind(id)
.bind(ELIGIBLE_KINDS.as_slice())
.fetch_optional(conn)
.await?)
}
pub(super) async fn valid_target(
conn: &mut PgConnection,
community: CommunityId,
@@ -174,6 +193,17 @@ pub(super) async fn valid_target(
Ok(Some(root))
}
/// A thread's count through channel cut `cut`: its counted replies numbered
/// at or before it. Equal to the count ingest assigned, since both follow the
/// channel's numbering. The root's latest reply answers the usual case.
const THREAD_SEQ_AT: &str = "CASE WHEN root.reply_channel_seq IS NULL
OR root.reply_channel_seq <= $cut THEN root.reply_seq
ELSE (SELECT count(*) FROM thread_metadata tm JOIN events e ON e.community_id=tm.community_id
AND e.created_at=tm.event_created_at AND e.id=tm.event_id
WHERE tm.community_id=root.community_id AND tm.root_event_id=root.event_id
AND tm.event_id<>root.event_id AND NOT (tm.depth=1 AND tm.broadcast)
AND e.kind=ANY($kinds) AND e.channel_seq <= $cut) END";
pub(super) async fn apply(
conn: &mut PgConnection,
community: CommunityId,
@@ -199,20 +229,29 @@ pub(super) async fn apply(
else {
return Ok(IntentOutcome::Blocked);
};
let (Some(timeline_seq), Some(channel_seq)) = (msg.timeline_seq, msg.channel_seq)
else {
// Stored before numbering: already read.
return Ok(IntentOutcome::Applied);
};
if root.is_empty() {
if !msg.on_timeline {
return Ok(IntentOutcome::Blocked);
}
sqlx::query(
"INSERT INTO personal_read_frontiers
(community_id, actor, channel_id, root_id, through_timestamp)
VALUES ($1,$2,$3,''::bytea,$4)
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE
SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, $4)",
"INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
through_seq, through_channel_seq, through_timestamp)
VALUES ($1,$2,$3,$4,$5,$6)
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
through_seq=GREATEST(personal_read_frontiers.through_seq, $4),
through_channel_seq=GREATEST(personal_read_frontiers.through_channel_seq, $5),
through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, $6)",
)
.bind(community.as_uuid())
.bind(actor)
.bind(target.channel_id)
.bind(timeline_seq)
.bind(channel_seq)
.bind(msg.received_at)
.execute(&mut *conn)
.await?;
@@ -222,18 +261,27 @@ pub(super) async fn apply(
}
// Reading a thread never joins it: only the actor's threads
// have a row to advance.
sqlx::query(
"UPDATE personal_read_frontiers
SET through_timestamp=GREATEST(through_timestamp, $5)
WHERE community_id=$1 AND actor=$2 AND channel_id=$3 AND root_id=$4",
let sql = format!(
"UPDATE personal_read_frontiers tf SET
through_seq=GREATEST(tf.through_seq, {THREAD_SEQ_AT}),
through_channel_seq=GREATEST(tf.through_channel_seq, $5),
through_timestamp=GREATEST(tf.through_timestamp, $6)
FROM thread_metadata root
WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=$3 AND tf.root_id=$4
AND root.community_id=$1 AND root.event_id=$4 AND root.depth=0"
)
.bind(community.as_uuid())
.bind(actor)
.bind(target.channel_id)
.bind(&root)
.bind(msg.received_at)
.execute(&mut *conn)
.await?;
.replace("$cut", "$5")
.replace("$kinds", "$7");
sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(community.as_uuid())
.bind(actor)
.bind(target.channel_id)
.bind(&root)
.bind(channel_seq)
.bind(msg.received_at)
.bind(ELIGIBLE_KINDS.as_slice())
.execute(&mut *conn)
.await?;
}
}
ReadIntent::MarkChannelRead {
@@ -246,19 +294,55 @@ pub(super) async fn apply(
if !access(conn, community, actor, *channel_id).await? {
return Ok(IntentOutcome::Blocked);
}
// Only the anchor's arrival matters: ancestry cannot change which
// Only the anchor's numbers matter: ancestry cannot change which
// messages a whole-channel cut covers.
let Some(through) = anchor_received_at(conn, community, *channel_id, &id).await? else {
let Some(msg) =
message(conn, community, *channel_id, &id, Some(&ELIGIBLE_KINDS)).await?
else {
return Ok(IntentOutcome::Blocked);
};
let (Some(timeline_seq), Some(cut)) = (msg.timeline_seq, msg.channel_seq) else {
return Ok(IntentOutcome::Applied);
};
sqlx::query(
"INSERT INTO personal_read_frontiers (community_id, actor, channel_id, root_id,
through_timestamp, threads_through_timestamp) VALUES ($1,$2,$3,''::bytea,$4,$4)
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE
SET through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, $4),
threads_through_timestamp=GREATEST(personal_read_frontiers.threads_through_timestamp, $4)",
).bind(community.as_uuid()).bind(actor).bind(channel_id).bind(through)
.execute(&mut *conn).await?;
"INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
through_seq, through_channel_seq, through_timestamp)
VALUES ($1,$2,$3,$4,$5,$6)
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
through_seq=GREATEST(personal_read_frontiers.through_seq, $4),
through_channel_seq=GREATEST(personal_read_frontiers.through_channel_seq, $5),
through_timestamp=GREATEST(personal_read_frontiers.through_timestamp, $6)",
)
.bind(community.as_uuid())
.bind(actor)
.bind(channel_id)
.bind(timeline_seq)
.bind(cut)
.bind(msg.received_at)
.execute(&mut *conn)
.await?;
// The cut covers every one of the actor's threads in the channel.
let sql = format!(
"UPDATE personal_read_frontiers tf SET
through_seq=GREATEST(tf.through_seq, {THREAD_SEQ_AT}),
through_channel_seq=GREATEST(tf.through_channel_seq, $4),
through_timestamp=GREATEST(tf.through_timestamp, $5)
FROM thread_metadata root
WHERE tf.community_id=$1 AND tf.actor=$2 AND tf.channel_id=$3
AND tf.root_id<>''::bytea AND root.community_id=$1
AND root.event_id=tf.root_id AND root.depth=0"
)
.replace("$cut", "$4")
.replace("$kinds", "$6");
sqlx::query(sqlx::AssertSqlSafe(sql))
.bind(community.as_uuid())
.bind(actor)
.bind(channel_id)
.bind(cut)
.bind(msg.received_at)
.bind(ELIGIBLE_KINDS.as_slice())
.execute(&mut *conn)
.await?;
}
}
Ok(IntentOutcome::Applied)

79
migrations/0056_personal_read_state.sql
View File

@@ -1,13 +1,23 @@
-- Private accessory read progress. Never included in Nostr event queries.
-- A frontier is the relay arrival time (events.received_at) of the message a
-- context was read through, never signed event time, which the sender chooses.
-- An empty root_id covers only the channel timeline; a root-specific frontier
-- covers that thread, without inheritance.
-- threads_through_timestamp is the only cross-context cut: an explicit
-- whole-channel read that also covers every thread in that channel.
-- started_at is the actor's first read intent and the floor of every
-- position: until then nothing counts, and from then the actor starts caught
-- up. Ingest creates accounts for thread membership without starting them.
--
-- Unread is counted by subtraction over relay-assigned message numbers.
-- Ingest numbers every eligible message in its channel under one locked
-- counter row:
-- events.channel_seq every eligible message in the channel, in commit order;
-- events.timeline_seq the channel's timeline count once the message is
-- stored (its own number when it is on the timeline);
-- thread_metadata.reply_seq on a root: the thread's eligible replies so far,
-- broadcast depth-1 replies excluded (they count on the
-- timeline), with reply_channel_seq and reply_last_id
-- the latest one's.
-- A position is the numbers a context was read through. Unread is the latest
-- number minus the position, so every numbered message counts for every
-- reader: posting marks the author's position, and deleted messages count
-- until read past.
--
-- started_at is the actor's first read intent: until then nothing counts.
-- Starting writes a caught-up position for every joined channel and thread,
-- and joining a channel afterwards writes one at its current count.
CREATE TABLE personal_read_accounts (
community_id UUID NOT NULL REFERENCES communities(id),
actor BYTEA NOT NULL CHECK (octet_length(actor) = 32),
@@ -15,15 +25,32 @@ CREATE TABLE personal_read_accounts (
PRIMARY KEY (community_id, actor)
);
CREATE TABLE personal_read_counters (
community_id UUID NOT NULL,
channel_id UUID NOT NULL,
timeline_seq BIGINT NOT NULL DEFAULT 0,
channel_seq BIGINT NOT NULL DEFAULT 0,
-- The message numbered channel_seq: marking the channel read through it
-- reads everything, so a badge has an anchor without a scan.
latest_id BYTEA,
PRIMARY KEY (community_id, channel_id),
FOREIGN KEY (community_id, channel_id)
REFERENCES channels (community_id, id) ON DELETE CASCADE
);
-- An empty root_id is the channel timeline. A thread row exists exactly for
-- the actor's threads. through_seq is in the context's own numbering
-- (timeline or thread); through_channel_seq is the channel-wide cut it was
-- read through. through_timestamp is the anchor's arrival, used only to
-- bound the mention scan.
CREATE TABLE personal_read_frontiers (
community_id UUID NOT NULL,
actor BYTEA NOT NULL,
channel_id UUID NOT NULL,
root_id BYTEA NOT NULL DEFAULT ''::bytea CHECK (octet_length(root_id) IN (0, 32)),
through_seq BIGINT NOT NULL,
through_channel_seq BIGINT NOT NULL,
through_timestamp TIMESTAMPTZ NOT NULL,
-- Whole-channel cut covering every thread; channel rows only.
threads_through_timestamp TIMESTAMPTZ
CHECK (threads_through_timestamp IS NULL 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,
@@ -31,7 +58,35 @@ CREATE TABLE personal_read_frontiers (
REFERENCES channels (community_id, id) ON DELETE CASCADE
);
ALTER TABLE events ADD COLUMN channel_seq BIGINT, ADD COLUMN timeline_seq BIGINT;
ALTER TABLE thread_metadata ADD COLUMN reply_seq BIGINT NOT NULL DEFAULT 0,
ADD COLUMN reply_channel_seq BIGINT, ADD COLUMN reply_last_id BYTEA;
-- Joining (or rejoining) a channel after starting begins caught up.
CREATE FUNCTION personal_read_on_join() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
IF NEW.removed_at IS NULL AND (TG_OP = 'INSERT' OR OLD.removed_at IS NOT NULL) THEN
INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
through_seq, through_channel_seq, through_timestamp)
SELECT NEW.community_id, NEW.pubkey, NEW.channel_id,
COALESCE(k.timeline_seq, 0), COALESCE(k.channel_seq, 0), now()
FROM personal_read_accounts a
LEFT JOIN personal_read_counters k
ON k.community_id = NEW.community_id AND k.channel_id = NEW.channel_id
WHERE a.community_id = NEW.community_id AND a.actor = NEW.pubkey
AND a.started_at IS NOT NULL
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
through_seq = EXCLUDED.through_seq,
through_channel_seq = EXCLUDED.through_channel_seq,
through_timestamp = EXCLUDED.through_timestamp;
END IF;
RETURN NULL;
END $$;
CREATE TRIGGER trg_channel_members_personal_read_join
AFTER INSERT OR UPDATE OF removed_at ON channel_members
FOR EACH ROW EXECUTE FUNCTION personal_read_on_join();
SELECT attach_community_write_fence('personal_read_accounts');
SELECT attach_community_write_fence('personal_read_counters');
SELECT attach_community_write_fence('personal_read_frontiers');

78
schema/schema.sql
View File

@@ -194,12 +194,25 @@ CREATE UNIQUE INDEX idx_users_okta ON users (community_id, okta_user_id)
WHERE okta_user_id IS NOT NULL;
-- Private accessory read progress. Never included in Nostr event queries.
-- A frontier is the relay arrival time (events.received_at) of the message a
-- context was read through, never signed event time. An empty root_id denotes
-- a channel frontier.
-- started_at is the actor's first read intent and the floor of every
-- position: until then nothing counts, and from then the actor starts caught
-- up. Ingest creates accounts for thread membership without starting them.
--
-- Unread is counted by subtraction over relay-assigned message numbers.
-- Ingest numbers every eligible message in its channel under one locked
-- counter row:
-- events.channel_seq every eligible message in the channel, in commit order;
-- events.timeline_seq the channel's timeline count once the message is
-- stored (its own number when it is on the timeline);
-- thread_metadata.reply_seq on a root: the thread's eligible replies so far,
-- broadcast depth-1 replies excluded (they count on the
-- timeline), with reply_channel_seq and reply_last_id
-- the latest one's.
-- A position is the numbers a context was read through. Unread is the latest
-- number minus the position, so every numbered message counts for every
-- reader: posting marks the author's position, and deleted messages count
-- until read past.
--
-- started_at is the actor's first read intent: until then nothing counts.
-- Starting writes a caught-up position for every joined channel and thread,
-- and joining a channel afterwards writes one at its current count.
CREATE TABLE personal_read_accounts (
community_id UUID NOT NULL REFERENCES communities(id),
actor BYTEA NOT NULL CHECK (octet_length(actor) = 32),
@@ -207,15 +220,32 @@ CREATE TABLE personal_read_accounts (
PRIMARY KEY (community_id, actor)
);
CREATE TABLE personal_read_counters (
community_id UUID NOT NULL,
channel_id UUID NOT NULL,
timeline_seq BIGINT NOT NULL DEFAULT 0,
channel_seq BIGINT NOT NULL DEFAULT 0,
-- The message numbered channel_seq: marking the channel read through it
-- reads everything, so a badge has an anchor without a scan.
latest_id BYTEA,
PRIMARY KEY (community_id, channel_id),
FOREIGN KEY (community_id, channel_id)
REFERENCES channels (community_id, id) ON DELETE CASCADE
);
-- An empty root_id is the channel timeline. A thread row exists exactly for
-- the actor's threads. through_seq is in the context's own numbering
-- (timeline or thread); through_channel_seq is the channel-wide cut it was
-- read through. through_timestamp is the anchor's arrival, used only to
-- bound the mention scan.
CREATE TABLE personal_read_frontiers (
community_id UUID NOT NULL,
actor BYTEA NOT NULL,
channel_id UUID NOT NULL,
root_id BYTEA NOT NULL DEFAULT ''::bytea CHECK (octet_length(root_id) IN (0, 32)),
through_seq BIGINT NOT NULL,
through_channel_seq BIGINT NOT NULL,
through_timestamp TIMESTAMPTZ NOT NULL,
-- Whole-channel cut covering every thread; channel rows only.
threads_through_timestamp TIMESTAMPTZ
CHECK (threads_through_timestamp IS NULL 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,
@@ -223,6 +253,31 @@ CREATE TABLE personal_read_frontiers (
REFERENCES channels (community_id, id) ON DELETE CASCADE
);
-- Joining (or rejoining) a channel after starting begins caught up.
CREATE FUNCTION personal_read_on_join() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
IF NEW.removed_at IS NULL AND (TG_OP = 'INSERT' OR OLD.removed_at IS NOT NULL) THEN
INSERT INTO personal_read_frontiers (community_id, actor, channel_id,
through_seq, through_channel_seq, through_timestamp)
SELECT NEW.community_id, NEW.pubkey, NEW.channel_id,
COALESCE(k.timeline_seq, 0), COALESCE(k.channel_seq, 0), now()
FROM personal_read_accounts a
LEFT JOIN personal_read_counters k
ON k.community_id = NEW.community_id AND k.channel_id = NEW.channel_id
WHERE a.community_id = NEW.community_id AND a.actor = NEW.pubkey
AND a.started_at IS NOT NULL
ON CONFLICT (community_id, actor, channel_id, root_id) DO UPDATE SET
through_seq = EXCLUDED.through_seq,
through_channel_seq = EXCLUDED.through_channel_seq,
through_timestamp = EXCLUDED.through_timestamp;
END IF;
RETURN NULL;
END $$;
CREATE TRIGGER trg_channel_members_personal_read_join
AFTER INSERT OR UPDATE OF removed_at ON channel_members
FOR EACH ROW EXECUTE FUNCTION personal_read_on_join();
-- ── Events (partitioned by month on created_at) ──────────────────────────────
-- Conformance: "Channel-less global events and DMs". `community_id` leads the
-- PK and every hot-path index. Partition stays BY RANGE (created_at) — the
@@ -563,6 +618,10 @@ CREATE TABLE thread_metadata (
FOREIGN KEY (community_id, channel_id) REFERENCES channels (community_id, id)
);
ALTER TABLE events ADD COLUMN channel_seq BIGINT, ADD COLUMN timeline_seq BIGINT;
ALTER TABLE thread_metadata ADD COLUMN reply_seq BIGINT NOT NULL DEFAULT 0,
ADD COLUMN reply_channel_seq BIGINT, ADD COLUMN reply_last_id BYTEA;
CREATE INDEX idx_thread_metadata_parent ON thread_metadata (community_id, parent_event_id);
CREATE INDEX idx_thread_metadata_root ON thread_metadata (community_id, root_event_id);
CREATE INDEX idx_thread_metadata_window
@@ -1820,6 +1879,7 @@ SELECT attach_community_write_fence('moderation_actions');
SELECT attach_community_write_fence('moderation_reports');
SELECT attach_community_write_fence('parameterized_event_watermarks');
SELECT attach_community_write_fence('personal_read_accounts');
SELECT attach_community_write_fence('personal_read_counters');
SELECT attach_community_write_fence('personal_read_frontiers');
SELECT attach_community_write_fence('pubkey_allowlist');
SELECT attach_community_write_fence('push_leases');