diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index d62086bf85..63466f762f 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -705,12 +705,17 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 56); + assert_eq!(migrations.len(), 57); assert_eq!(migrations[55].version, 56); assert!(migrations[55] .sql .as_str() .contains("CREATE TABLE personal_read_accounts")); + assert_eq!(migrations[56].version, 57); + assert!(migrations[56] + .sql + .as_str() + .contains("ALTER TABLE personal_read_accounts ADD COLUMN started_at")); assert_eq!(migrations[48].version, 49); assert_eq!(migrations[49].version, 50); assert_eq!(migrations[50].version, 51); @@ -2210,9 +2215,30 @@ mod postgres_tests { .sql .as_ref(), ); + // 0057 adds personal_read_accounts.started_at after 0056's CREATE TABLE. + let started_at_column = "started_at timestamptz, "; + assert!(MIGRATOR + .iter() + .find(|m| m.version == 57) + .expect("personal read start migration") + .sql + .as_str() + .contains("ADD COLUMN started_at TIMESTAMPTZ")); + assert!(schema + .tables + .get("personal_read_accounts") + .expect("schema.sql personal read accounts") + .contains(started_at_column)); for (table, definition) in personal.tables { + let in_schema = schema.tables.get(&table).map(|schema_definition| { + if table == "personal_read_accounts" { + schema_definition.replacen(started_at_column, "", 1) + } else { + schema_definition.clone() + } + }); assert_eq!( - schema.tables.get(&table), + in_schema.as_ref(), Some(&definition), "personal read table {table} differs" ); diff --git a/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs index 6e4248a740..beda4b199f 100644 --- a/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs +++ b/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs @@ -311,7 +311,16 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer() production_result.is_ok(), "production migrator must preserve ingestion progress: {production_result:?}" ); - assert_eq!(version, 56); + let latest = sqlx::migrate::Migrator::new( + std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../migrations"), + ) + .await + .unwrap() + .iter() + .map(|m| m.version) + .max() + .unwrap(); + assert_eq!(version, latest); assert_eq!(final_oid, oid, "prebuild must not be replaced"); assert_eq!(count, 4, "all writer witnesses must persist"); } diff --git a/crates/buzz-db/src/store/personal_read/context.rs b/crates/buzz-db/src/store/personal_read/context.rs index 39a537575a..4ecc5a2cb4 100644 --- a/crates/buzz-db/src/store/personal_read/context.rs +++ b/crates/buzz-db/src/store/personal_read/context.rs @@ -39,7 +39,9 @@ impl Db { .await?; writes::deadlines(&mut tx).await?; let actor_bytes = actor.to_bytes(); - let account = read_account(&mut tx, retention_seconds).await?; + let account = read_account(&mut tx, community, &actor_bytes, retention_seconds).await?; + // Every frontier is floored at the account's start. + let floor = account.floor(); let mut contexts = Vec::with_capacity(queries.len()); // Replies whose state turns on conversation membership, by position. let mut pending = Vec::new(); @@ -128,7 +130,7 @@ impl Db { account.cutoff_ms, ) { MessageReadState::NotCounted - } else if prefix.is_some_and(|p| received <= p) { + } else if received <= prefix.map_or(floor, |p| p.max(floor)) { MessageReadState::Read } else { let reason = classification::reason( diff --git a/crates/buzz-db/src/store/personal_read/model.rs b/crates/buzz-db/src/store/personal_read/model.rs index 5786881ba8..1b329b46ac 100644 --- a/crates/buzz-db/src/store/personal_read/model.rs +++ b/crates/buzz-db/src/store/personal_read/model.rs @@ -1,3 +1,4 @@ +use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -56,6 +57,18 @@ pub struct ReadAccount { pub retention_seconds: u32, /// Read-time author-time cutoff (Unix milliseconds), not a discard boundary. pub cutoff_ms: i64, + /// Arrival floor of every frontier: the account's first applied read + /// intent. `None` until then, and nothing counts as unread. Not on the wire. + #[serde(skip)] + pub started_at: Option>, +} + +impl ReadAccount { + /// The arrival every frontier is floored at: the start, or the end of time + /// for an account that has not started. + pub(super) fn floor(&self) -> DateTime { + self.started_at.unwrap_or(DateTime::::MAX_UTC) + } } /// Maximum channel summaries in one sidebar page. diff --git a/crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs b/crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs index 682cb045af..ec65afb51d 100644 --- a/crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/participation_postgres_tests.rs @@ -34,11 +34,13 @@ impl World { .await .unwrap() .id; + let actor = Keys::generate(); + super::postgres_tests::start_before_everything(&pool, community, &actor.public_key()).await; Self { db, pool, community, - actor: Keys::generate(), + actor, peer: Keys::generate(), now: Timestamp::now().as_secs(), } diff --git a/crates/buzz-db/src/store/personal_read/postgres_tests.rs b/crates/buzz-db/src/store/personal_read/postgres_tests.rs index 8339cb9a4a..f6f87d127e 100644 --- a/crates/buzz-db/src/store/personal_read/postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/postgres_tests.rs @@ -38,9 +38,28 @@ pub(super) async fn fixture() -> (Db, PgPool, CommunityId, Uuid, Keys, nostr::Ev db.insert_event(community, &event, Some(channel)) .await .unwrap(); + start_before_everything(&pool, community, &actor.public_key()).await; (db, pool, community, channel, actor, event) } +/// Start an account before any fixture arrival, so unread semantics can be +/// tested without the start floor covering every message. +pub(super) async fn start_before_everything( + pool: &PgPool, + community: CommunityId, + actor: &nostr::PublicKey, +) { + sqlx::query( + "INSERT INTO personal_read_accounts (community_id,actor,started_at) + VALUES ($1,$2,'1900-01-01T00:00:00Z')", + ) + .bind(community.as_uuid()) + .bind(actor.to_bytes().as_slice()) + .execute(pool) + .await + .unwrap(); +} + /// Move every community message past the default horizon by author time, the /// only clock the unread window reads. async fn expire(pool: &PgPool, community: CommunityId) { @@ -253,6 +272,7 @@ async fn personal_read_sidebar_window_budget_counts_boundary_and_ineligible_tail #[ignore = "requires Postgres"] async fn personal_read_sidebar_is_read_only_and_does_not_wait_for_account() { let (db, pool, community, channel, actor, event) = fixture().await; + unstart(&pool, community).await; db.personal_read_sidebar( community, &actor.public_key(), @@ -540,6 +560,7 @@ async fn personal_read_latest_old_message_is_not_an_empty_channel() { #[ignore = "requires Postgres"] async fn personal_read_frontier_is_monotonic_and_rejects_malformed_anchors() { let (db, pool, community, channel, actor, event) = fixture().await; + unstart(&pool, community).await; let target = ReadTarget { channel_id: channel, root_id: None, @@ -626,6 +647,189 @@ async fn personal_read_frontier_is_monotonic_and_rejects_malformed_anchors() { assert!(sparse.is_none()); } +/// Undo the fixture's start, for tests of what creates an account. +async fn unstart(pool: &PgPool, community: CommunityId) { + sqlx::query("DELETE FROM personal_read_accounts WHERE community_id=$1") + .bind(community.as_uuid()) + .execute(pool) + .await + .unwrap(); +} + +async fn started_at( + pool: &PgPool, + community: CommunityId, + actor: &nostr::PublicKey, +) -> Option> { + sqlx::query_scalar( + "SELECT started_at FROM personal_read_accounts WHERE community_id=$1 AND actor=$2", + ) + .bind(community.as_uuid()) + .bind(actor.to_bytes().as_slice()) + .fetch_one(pool) + .await + .unwrap() +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn personal_read_first_applied_intent_starts_the_account_once() { + let (db, pool, community, channel, _, event) = fixture().await; + let intent = ReadIntent::MarkThrough { + target: ReadTarget { + channel_id: channel, + root_id: None, + }, + message_id: event.id.to_hex(), + }; + // The fixture actor is already started; the channel is open to anyone. + let actor = Keys::generate().public_key(); + assert!(sqlx::query_scalar::<_, Vec>( + "SELECT actor FROM personal_read_accounts WHERE community_id=$1 AND actor=$2" + ) + .bind(community.as_uuid()) + .bind(actor.to_bytes().as_slice()) + .fetch_optional(&pool) + .await + .unwrap() + .is_none()); + for _ in 0..2 { + assert_eq!( + db.apply_personal_read_intent(community, &actor, &intent) + .await + .unwrap(), + IntentOutcome::Applied + ); + } + let first = started_at(&pool, community, &actor).await; + assert!( + first.is_some(), + "the first applied intent starts the account" + ); + + // An account can exist before its actor starts; the first intent starts + // it, and later intents never move the start. + let pending = Keys::generate().public_key(); + sqlx::query("INSERT INTO personal_read_accounts (community_id,actor) VALUES ($1,$2)") + .bind(community.as_uuid()) + .bind(pending.to_bytes().as_slice()) + .execute(&pool) + .await + .unwrap(); + assert_eq!(started_at(&pool, community, &pending).await, None); + db.apply_personal_read_intent(community, &pending, &intent) + .await + .unwrap(); + let pending_start = started_at(&pool, community, &pending).await; + assert!(pending_start >= first); + db.apply_personal_read_intent(community, &pending, &intent) + .await + .unwrap(); + assert_eq!(started_at(&pool, community, &pending).await, pending_start); + assert_eq!(started_at(&pool, community, &actor).await, first); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn personal_read_counts_only_arrivals_after_the_account_starts() { + let (db, pool, community, channel, actor, _) = fixture().await; + unstart(&pool, community).await; + let unread = |page: &SidebarPage, id: Uuid| { + let row = page.channels.iter().find(|c| c.channel_id == id).unwrap(); + match row.unread { + ReadCount::Exact { value } => value, + _ => panic!("expected an exact count"), + } + }; + let sidebar = || async { + db.personal_read_sidebar( + community, + &actor.public_key(), + DEFAULT_RETENTION_SECONDS, + 20, + None, + ) + .await + .unwrap() + }; + // Before the first read intent nothing counts, not the 30-day window. + assert_eq!(unread(&sidebar().await, channel), 0); + + // The first intent, in another channel, starts the account everywhere. + let other = db + .create_channel( + community, + "elsewhere", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &actor.public_key().to_bytes(), + None, + ) + .await + .unwrap() + .id; + let elsewhere = EventBuilder::new(Kind::Custom(9), "elsewhere") + .sign_with_keys(&Keys::generate()) + .unwrap(); + db.insert_event(community, &elsewhere, Some(other)) + .await + .unwrap(); + db.apply_personal_read_intent( + community, + &actor.public_key(), + &ReadIntent::MarkThrough { + target: ReadTarget { + channel_id: other, + root_id: None, + }, + message_id: elsewhere.id.to_hex(), + }, + ) + .await + .unwrap(); + // A channel never marked starts caught up: earlier arrivals stay read. + assert_eq!(unread(&sidebar().await, channel), 0); + + // Arrivals after the start count. + let later = EventBuilder::new(Kind::Custom(9), "after the start") + .sign_with_keys(&Keys::generate()) + .unwrap(); + db.insert_event(community, &later, Some(channel)) + .await + .unwrap(); + assert_eq!(unread(&sidebar().await, channel), 1); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn personal_read_unstarted_account_is_exactly_caught_up_past_the_scan_cap() { + let (db, pool, community, channel, actor, _) = fixture().await; + unstart(&pool, community).await; + let last = add_history(&db, community, channel, MAX_UNREAD_SCAN + 44).await; + let page = db + .personal_read_sidebar( + community, + &actor.public_key(), + DEFAULT_RETENTION_SECONDS, + 20, + None, + ) + .await + .unwrap(); + let row = &page.channels[0]; + // The start floor covers rows the cap left unexamined: zero, not unknown. + assert!(matches!(row.unread, ReadCount::Exact { value: 0 })); + assert!(matches!(row.attention, ReadCount::Exact { value: 0 })); + assert!(row.threads.items.is_empty()); + assert!(row.threads.complete); + assert_eq!( + row.latest_message_id.as_deref(), + Some(last.id.to_hex().as_str()) + ); + assert!(row.latest_message_complete); +} + #[tokio::test] #[ignore = "requires Postgres"] async fn personal_read_channel_and_thread_never_inherit_each_other() { @@ -777,7 +981,7 @@ async fn personal_read_contexts_bound_selectors_and_use_only_matching_frontiers( .fetch_one(&pool) .await .unwrap(); - assert_eq!(accounts, 0, "context GET must never create authority"); + assert_eq!(accounts, 1, "context GET must never create authority"); db.apply_personal_read_intent( community, &actor.public_key(), @@ -824,17 +1028,26 @@ async fn personal_read_contexts_bound_selectors_and_use_only_matching_frontiers( .unwrap(); assert_eq!(page["contexts"][1]["messages"][1]["status"], "read"); let other = Keys::generate(); - let page = serde_json::to_value( - db.personal_read_contexts( - community, - &other.public_key(), - DEFAULT_RETENTION_SECONDS, - &queries, + let other_page = || async { + serde_json::to_value( + db.personal_read_contexts( + community, + &other.public_key(), + DEFAULT_RETENTION_SECONDS, + &queries, + ) + .await + .unwrap(), ) - .await - .unwrap(), - ) - .unwrap(); + .unwrap() + }; + // Nothing counts before an account starts. + let page = other_page().await; + assert_eq!(page["contexts"][0]["messages"][0]["status"], "read"); + assert_eq!(page["contexts"][1]["messages"][1]["status"], "read"); + // Frontiers are per actor: the actor's marks never read for another. + start_before_everything(&pool, community, &other.public_key()).await; + let page = other_page().await; assert_eq!(page["contexts"][0]["messages"][0]["status"], "unread"); assert_eq!(page["contexts"][1]["messages"][1]["status"], "unread"); assert!(db diff --git a/crates/buzz-db/src/store/personal_read/projection.rs b/crates/buzz-db/src/store/personal_read/projection.rs index b839ca7459..45dd619956 100644 --- a/crates/buzz-db/src/store/personal_read/projection.rs +++ b/crates/buzz-db/src/store/personal_read/projection.rs @@ -77,7 +77,7 @@ impl Db { 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, retention_seconds).await?; + let account = read_account(&mut tx, community, &actor_bytes, retention_seconds).await?; // The inner event LIMIT is deliberately before eligibility filtering. // This bounds rows/joins even with long deleted or self-authored runs. // Aggregate equivalent eligible evidence before transfer. Multiplicity @@ -132,10 +132,11 @@ impl Db { ), classified AS ( SELECT e.*, tm.root_event_id AS root, tm.parent_event_id AS parent, COALESCE(tm.root_event_id<>e.id,false) AS is_reply, - COALESCE(e.received_at <= + -- Every frontier is floored at the account's start. + COALESCE(e.received_at <= GREATEST(COALESCE($10::timestamptz,'infinity'), CASE WHEN tm.root_event_id IS NOT NULL AND tm.root_event_id<>e.id THEN GREATEST(tf.through_timestamp, cf.threads_through_timestamp) - ELSE cf.through_timestamp END,false) AS covered + ELSE cf.through_timestamp END),false) AS covered FROM (SELECT * FROM candidates ORDER BY created_at DESC,id LIMIT $8-1) e LEFT JOIN thread_metadata tm ON tm.community_id=$1 AND tm.channel_id=r.id AND tm.event_created_at=e.created_at AND tm.event_id=e.id @@ -189,8 +190,12 @@ impl Db { .ok_or_else(|| DbError::InvalidData("invalid unread cutoff".into()))?) .bind((MAX_UNREAD_SCAN+1) as i64) .bind(only) + .bind(account.started_at) .fetch_all(&mut *tx).await?; let has_more = rows.len() > limit; + // An unstarted account has read every arrival, scanned or not, so its + // counts are exactly zero whatever the scan cap left unexamined. + let started = account.started_at.is_some(); let mut channels = Vec::new(); let mut pending = Vec::new(); for row in rows.into_iter().take(limit) { @@ -198,7 +203,8 @@ impl Db { let evidence = evidence .as_array() .ok_or_else(|| DbError::InvalidData("invalid sidebar evidence".into()))?; - let complete = row.try_get::("scanned")? <= MAX_UNREAD_SCAN as i64; + let evidence = if started { evidence.as_slice() } else { &[] }; + let complete = !started || row.try_get::("scanned")? <= MAX_UNREAD_SCAN as i64; let channel_type: String = row.try_get("channel_type")?; let mut unread = 0; let mut attention = 0; @@ -379,20 +385,28 @@ pub(super) fn summarize( ThreadSummaries { items, complete } } -/// Read-time horizon only: frontier state is not discarded on expiry. +/// Read-time horizon and the account's start, in the caller's snapshot. +/// Frontier state is not discarded on expiry. pub(super) async fn read_account( conn: &mut PgConnection, + community: CommunityId, + actor: &[u8], retention_seconds: u32, ) -> Result { - let cutoff: DateTime = sqlx::query_scalar( - "SELECT date_trunc('milliseconds',transaction_timestamp()-make_interval(secs=>$1::double precision))", + let row = sqlx::query( + "SELECT date_trunc('milliseconds',transaction_timestamp()-make_interval(secs=>$1::double precision)) AS cutoff, + (SELECT started_at FROM personal_read_accounts WHERE community_id=$2 AND actor=$3) AS started_at", ) .bind(f64::from(retention_seconds)) + .bind(community.as_uuid()) + .bind(actor) .fetch_one(conn) .await?; + let cutoff: DateTime = row.try_get("cutoff")?; Ok(ReadAccount { retention_seconds, cutoff_ms: cutoff.timestamp_millis(), + started_at: row.try_get("started_at")?, }) } diff --git a/crates/buzz-db/src/store/personal_read/writes.rs b/crates/buzz-db/src/store/personal_read/writes.rs index 9947d2b7ab..448e3f5367 100644 --- a/crates/buzz-db/src/store/personal_read/writes.rs +++ b/crates/buzz-db/src/store/personal_read/writes.rs @@ -23,15 +23,17 @@ pub(super) async fn deadlines(conn: &mut PgConnection) -> Result<()> { Ok(()) } -/// Serialize private frontier writes, never shared conversation rows. +/// Start the account if this is the actor's first read intent, then serialize +/// private frontier writes, never shared conversation rows. pub(super) async fn lock_account( conn: &mut PgConnection, community: CommunityId, actor: &[u8], ) -> Result<()> { sqlx::query( - "INSERT INTO personal_read_accounts (community_id,actor) VALUES ($1,$2) - ON CONFLICT (community_id,actor) DO NOTHING", + "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", ) .bind(community.as_uuid()) .bind(actor) diff --git a/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs b/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs index 58ed929bec..49f3de7f55 100644 --- a/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs +++ b/crates/buzz-relay/src/api/buzz_v1/postgres_tests.rs @@ -56,6 +56,24 @@ async fn request( (status, serde_json::from_slice(&bytes).unwrap()) } +/// Start a reader's account before any fixture arrival. Nothing counts as +/// unread until an account starts. +async fn start_before_everything( + state: &crate::state::AppState, + community: buzz_core::CommunityId, + key: &Keys, +) { + sqlx::query( + "INSERT INTO personal_read_accounts (community_id,actor,started_at) + VALUES ($1,$2,'1900-01-01T00:00:00Z')", + ) + .bind(community.as_uuid()) + .bind(key.public_key().to_bytes().as_slice()) + .execute(state.db.pool()) + .await + .unwrap(); +} + #[tokio::test] #[ignore = "requires Postgres"] async fn accessory_router_signed_url_body_replay_and_actor_boundary() { @@ -217,6 +235,7 @@ async fn accessory_router_signed_url_body_replay_and_actor_boundary() { .map(|b| format!("%{b:02X}")) .collect(); let path = format!("/buzz/v1/me/read-state?targets={targets}"); + start_before_everything(&state, community, &other).await; for (key, status) in [(&actor, "read"), (&other, "unread")] { let auth = proof(key, &host, &path, "GET", None); let page = request(state.clone(), &host, &path, "GET", Some(&auth), b"").await; @@ -275,6 +294,7 @@ async fn accessory_context_get_signed_query_and_independent_batch_outcomes() { "/buzz/v1/me/read-state?targets={}", encode(&targets.to_string()) ); + start_before_everything(&state, community, &actor).await; 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); @@ -609,6 +629,7 @@ async fn signed_sidebar_deletion(deletion_kind: u16) { .sign_with_keys(signer) .unwrap(); let path = format!("/buzz/v1/me/sidebar?channel_ids={channel}"); + start_before_everything(&state, community, &reader).await; for (event, key, count) in [(&message, &author, 1), (&deletion, signer, 0)] { let body = serde_json::to_vec(event).unwrap(); let auth = proof(key, &host, "/events", "POST", Some(&body)); diff --git a/docs/buzz-v1-read-state.md b/docs/buzz-v1-read-state.md index b6210a75b4..befd690f29 100644 --- a/docs/buzz-v1-read-state.md +++ b/docs/buzz-v1-read-state.md @@ -229,9 +229,12 @@ examined: the 4,096 most recent events by author time inside the horizon. So marking through it reads everything counted. When the horizon holds no message, it is the last to arrive among the channel's 256 most recent events. -There is no import of earlier client read state: an account starts with no -frontiers, and the horizon bounds what that can show as unread. Manual unread -remains device-local. +There is no import of earlier client read state. The relay records the +account's `started_at` at its first applied read intent and never moves it. +Every frontier is floored at `started_at`: until the account starts nothing +counts as unread, and from then it starts caught up, so a context it has never +marked counts only arrivals after the start. Manual unread remains +device-local. ## Bounds and deployment @@ -251,10 +254,11 @@ remains device-local. shared eight-second intent-processing deadline after admission. Limits are containment, not a production capacity claim. -Apply migration 0056 (or the equivalent desired schema). It creates two empty -private tables and no index on `events`: both sidebar scans are served by the -existing `idx_events_community_channel_created`. No new per-message ingest write -path or stored unread counters are introduced. +Apply migrations 0056 and 0057 (or the equivalent desired schema). 0056 +creates two empty private tables and no index on `events`, and 0057 adds the +nullable `personal_read_accounts.started_at` column. Both sidebar scans are +served by the existing `idx_events_community_channel_created`. No new +per-message ingest write path or stored unread counters are introduced. Use existing HTTP route/status/latency metrics for `/buzz/v1/me/sidebar` and `/buzz/v1/me/read-state`, plus database pool/statement metrics. Inspect exact / @@ -292,10 +296,8 @@ is no new public account export/reset endpoint. Operator-assisted erasure/export must use the established authenticated operational process and explicitly scope both community and actor; never equate the read-time horizon with data erasure. -Migration 0056 must be applied before this relay serves, enabled or not: started -without it and with auto-migration off, the relay stops before readiness. There -is no down migration, and disabling the API is not a rollback. A relay built -before 0056 that restarts with `BUZZ_AUTO_MIGRATE=true` (the Helm default) +There is no down migration, and disabling the API is not a rollback. A relay +built before 0056 that restarts with `BUZZ_AUTO_MIGRATE=true` (the Helm default) refuses to start on the migrated schema. Whole-community deletion run from a build before 0056 rejects the two new tables. A deletion approved on the earlier schema and not yet fenced fails structural revalidation after 0056: diff --git a/migrations/0057_personal_read_started_at.sql b/migrations/0057_personal_read_started_at.sql new file mode 100644 index 0000000000..4d73f5f2b4 --- /dev/null +++ b/migrations/0057_personal_read_started_at.sql @@ -0,0 +1,4 @@ +-- started_at is the actor's first read intent. NULL until then: an account +-- can exist before its actor has started. Read positions are floored at it, +-- so a started account begins caught up and an unstarted one counts nothing. +ALTER TABLE personal_read_accounts ADD COLUMN started_at TIMESTAMPTZ; diff --git a/schema/tables/public/personal_read_accounts.sql b/schema/tables/public/personal_read_accounts.sql index 2761fa55b8..0573c45807 100644 --- a/schema/tables/public/personal_read_accounts.sql +++ b/schema/tables/public/personal_read_accounts.sql @@ -2,8 +2,11 @@ -- A frontier is the relay arrival time (events.received_at) of the message a -- context was read through; the unread cutoff alone is signed event time. An -- empty root_id denotes a channel frontier. +-- started_at is the actor's first read intent. NULL until then: an account +-- can exist before its actor has started. CREATE TABLE personal_read_accounts ( community_id UUID NOT NULL REFERENCES communities(id), actor BYTEA NOT NULL CHECK (octet_length(actor) = 32), + started_at TIMESTAMPTZ, PRIMARY KEY (community_id, actor) );