fix(desktop): scope observer feed by channel and add session-boundary dividers

desktop/src-tauri/src/archive/mod.rs

@@ -475,6 +475,120 @@ pub fn delete_save_subscription(
 
 // ── read_archived_events ─────────────────────────────────────────────────────
 
+// ── read_archived_observer_events_for_channel ────────────────────────────────
+
+/// Read a paginated page of archived kind 24200 events scoped to one channel,
+/// using the `observer_channel_index` as the primary lookup.
+///
+/// The index only contains rows with a known, non-null `channelId` (frames
+/// pre-dating the channelId stamp, or rows where decryption failed, are
+/// absent from the index and thus absent from every scoped channel view —
+/// Will's (a) ruling, 2026-07-08).
+///
+/// Returns at most `limit` events (default `DEFAULT_READ_LIMIT`) in
+/// newest-first order. Compound cursor `(before_created_at, before_id)` works
+/// identically to `read_archived_events`.
+#[tauri::command]
+pub fn read_archived_observer_events_for_channel(
+    state: State<'_, AppState>,
+    channel_id: String,
+    before_created_at: Option<i64>,
+    before_id: Option<String>,
+    limit: Option<i64>,
+) -> Result<Vec<String>, String> {
+    let identity_pk = identity_pubkey(&state)?;
+    let relay_url = relay_ws_url_with_override(&state);
+    let conn = open_db()?;
+    store::read_archived_observer_events_for_channel(
+        &conn,
+        &identity_pk,
+        &relay_url,
+        &channel_id,
+        before_created_at,
+        before_id.as_deref(),
+        limit.unwrap_or(DEFAULT_READ_LIMIT),
+    )
+}
+
+// ── index_observer_channel_id ─────────────────────────────────────────────────
+
+/// Index one or more archived observer frame ids with their decoded channelId.
+///
+/// Called from the TS-side backfill after attempting `decryptObserverEvent`.
+/// `channel_id` is `Some` for frames with a non-null channelId; `None` for
+/// frames where decryption yielded no channelId or failed entirely.  Both cases
+/// write a status row so a re-run skips the frame (INSERT OR IGNORE on PK).
+///
+/// Idempotent: rows that are already indexed are left unchanged.
+#[tauri::command]
+pub fn index_observer_channel_id(
+    state: State<'_, AppState>,
+    entries: Vec<ObserverChannelIndexEntry>,
+) -> Result<(), String> {
+    let identity_pk = identity_pubkey(&state)?;
+    let relay_url = relay_ws_url_with_override(&state);
+    let conn = open_db()?;
+    for entry in &entries {
+        store::upsert_observer_channel_index(
+            &conn,
+            &identity_pk,
+            &relay_url,
+            &entry.event_id,
+            entry.channel_id.as_deref(),
+            entry.created_at,
+        )?;
+    }
+    Ok(())
+}
+
+/// A single (event_id, channel_id?, created_at) record used by
+/// `index_observer_channel_id`.
+///
+/// `channel_id` is `None` for frames where decryption found no channelId or
+/// failed; those rows are written to `observer_channel_index` with a NULL
+/// channel_id so the frame is treated as processed (no re-decrypt on re-run).
+#[derive(Debug, Deserialize)]
+pub struct ObserverChannelIndexEntry {
+    pub event_id: String,
+    pub channel_id: Option<String>,
+    pub created_at: i64,
+}
+
+// ── read_unindexed_observer_rows ─────────────────────────────────────────────
+
+/// Return raw event JSON + id + created_at for all `owner_p` kind 24200 rows
+/// that are NOT yet in `observer_channel_index`.
+///
+/// The TS-side backfill driver calls this once, decrypts each row, and sends
+/// the (id, channelId, created_at) triples back via `index_observer_channel_id`.
+/// Together these constitute the one-shot idempotent backfill required by the
+/// Slice 1 acceptance criteria (Thufir Pass 4).
+#[tauri::command]
+pub fn read_unindexed_observer_rows(
+    state: State<'_, AppState>,
+) -> Result<Vec<RawObserverRow>, String> {
+    let identity_pk = identity_pubkey(&state)?;
+    let relay_url = relay_ws_url_with_override(&state);
+    let conn = open_db()?;
+    let rows = store::read_unindexed_observer_rows(&conn, &identity_pk, &relay_url)?;
+    Ok(rows
+        .into_iter()
+        .map(|(id, raw_json, created_at)| RawObserverRow {
+            id,
+            raw_json,
+            created_at,
+        })
+        .collect())
+}
+
+/// Wire type returned by `read_unindexed_observer_rows`.
+#[derive(Debug, Serialize)]
+pub struct RawObserverRow {
+    pub id: String,
+    pub raw_json: String,
+    pub created_at: i64,
+}
+
 /// Default page size for `read_archived_events`.
 const DEFAULT_READ_LIMIT: i64 = 50;
 

desktop/src-tauri/src/archive/pipeline.rs

@@ -423,6 +423,26 @@ pub(super) fn commit_archive(
                 &p.matched_scope.scope_value,
                 now,
             )?;
+
+            // Index at ingest: attempt to decrypt and extract channelId.
+            // Write a status row regardless of outcome so backfill never
+            // re-processes this frame (INSERT OR IGNORE on PK is a no-op if
+            // the row is already present from a prior run).
+            let channel_id_for_index: Option<String> =
+                buzz_core_pkg::observer::decrypt_observer_payload::<serde_json::Value>(
+                    owner_keys, &p.event,
+                )
+                .ok()
+                .and_then(|v| v.get("channelId")?.as_str().map(|s| s.to_owned()));
+            store::upsert_observer_channel_index(
+                &tx,
+                identity_pk,
+                relay_url,
+                eid,
+                channel_id_for_index.as_deref(),
+                p.event.created_at.as_secs() as i64,
+            )?;
+
             persisted += 1;
         }
 

desktop/src-tauri/src/archive/store.rs

@@ -46,8 +46,62 @@ CREATE TABLE IF NOT EXISTS save_subscriptions (
     created_at      INTEGER NOT NULL,
     PRIMARY KEY (identity_pubkey, relay_url, scope_type, scope_value)
 );
+
+-- Processing-status index for kind 24200 (observer) frames.
+--
+-- Every examined observer row — whether its channelId was successfully
+-- decrypted or not — gets exactly one row here keyed by (identity, relay, id).
+-- `channel_id` is nullable:
+--   NOT NULL  → frame was attributed to a channel; visible in that channel's feed.
+--   NULL      → frame was examined but had no channelId (pre-stamp or decrypt
+--               failure); excluded from all scoped views per Will's (a) ruling
+--               (2026-07-08), but the row exists so a re-run is a no-op.
+--
+-- Scoped reads filter `channel_id = ?`; NULL rows never match, staying hidden.
+CREATE TABLE IF NOT EXISTS observer_channel_index (
+    identity_pubkey TEXT NOT NULL,
+    relay_url       TEXT NOT NULL,
+    id              TEXT NOT NULL,
+    channel_id      TEXT,
+    created_at      INTEGER NOT NULL,
+    PRIMARY KEY (identity_pubkey, relay_url, id)
+);
+CREATE INDEX IF NOT EXISTS idx_observer_channel
+    ON observer_channel_index (identity_pubkey, relay_url, channel_id, created_at DESC, id DESC);
+
+-- One-row migration state table: tracks which idempotent migrations have run.
+CREATE TABLE IF NOT EXISTS archive_migrations (
+    name       TEXT PRIMARY KEY,
+    applied_at INTEGER NOT NULL
+);
 ";
 
+// ── Migration helpers ────────────────────────────────────────────────────────
+
+/// Returns true if a named migration has already been applied.
+pub fn migration_applied(conn: &Connection, name: &str) -> Result<bool, String> {
+    // The table is created in SCHEMA above, so it always exists after open.
+    let count: i64 = conn
+        .query_row(
+            "SELECT COUNT(*) FROM archive_migrations WHERE name = ?1",
+            params![name],
+            |row| row.get(0),
+        )
+        .map_err(|e| format!("migration_applied query: {e}"))?;
+    Ok(count > 0)
+}
+
+/// Mark a named migration as applied (idempotent: no-op if already recorded).
+pub fn mark_migration_applied(conn: &Connection, name: &str, now: i64) -> Result<(), String> {
+    conn.execute(
+        "INSERT INTO archive_migrations (name, applied_at) VALUES (?1, ?2)
+         ON CONFLICT (name) DO NOTHING",
+        params![name, now],
+    )
+    .map_err(|e| format!("mark_migration_applied: {e}"))?;
+    Ok(())
+}
+
 // ── Open / init ─────────────────────────────────────────────────────────────
 
 /// Open (or create) the archive database at the given path.
@@ -568,6 +622,157 @@ pub fn read_archived_events(
         .map_err(|e| format!("read read_archived_events row: {e}"))
 }
 
+/// Upsert a row in `observer_channel_index` for an examined kind 24200 frame.
+///
+/// `channel_id` is `Some` for frames with a successfully-decrypted, non-null
+/// channelId; `None` for frames that had no channelId or where decryption
+/// failed. Null-channel_id rows are excluded from all scoped channel reads
+/// (scoped queries filter `channel_id = ?`; NULLs never match), satisfying
+/// Will's (a) ruling (2026-07-08), while ensuring the row exists so a re-run
+/// is a no-op.
+///
+/// Idempotent: if the (identity, relay, id) PK already exists the row is left
+/// unchanged (INSERT OR IGNORE). Called both at ingest time (new frames) and
+/// from the one-shot backfill for existing archived rows.
+pub fn upsert_observer_channel_index(
+    conn: &Connection,
+    identity_pubkey: &str,
+    relay_url: &str,
+    event_id: &str,
+    channel_id: Option<&str>,
+    created_at: i64,
+) -> Result<(), String> {
+    conn.execute(
+        "INSERT OR IGNORE INTO observer_channel_index
+             (identity_pubkey, relay_url, id, channel_id, created_at)
+         VALUES (?1, ?2, ?3, ?4, ?5)",
+        params![identity_pubkey, relay_url, event_id, channel_id, created_at],
+    )
+    .map_err(|e| format!("failed to upsert observer_channel_index: {e}"))?;
+    Ok(())
+}
+
+/// Read all `owner_p` kind 24200 archived event rows (id + raw_json +
+/// created_at) that have NOT yet been processed (i.e., have no row in
+/// `observer_channel_index`, regardless of what channel_id would be).
+///
+/// Used by the one-shot backfill migration to discover which existing rows
+/// still need channel attribution attempted. A row is "processed" as soon as
+/// we attempt decryption — whether it succeeds (non-null channel_id written)
+/// or fails (null channel_id written) — so re-runs skip those rows.
+pub fn read_unindexed_observer_rows(
+    conn: &Connection,
+    identity_pubkey: &str,
+    relay_url: &str,
+) -> Result<Vec<(String, String, i64)>, String> {
+    let mut stmt = conn
+        .prepare(
+            "SELECT ae.id, ae.raw_json, ae.created_at
+             FROM archived_events ae
+             INNER JOIN archived_event_scopes aes
+                 ON aes.identity_pubkey = ae.identity_pubkey
+                AND aes.relay_url       = ae.relay_url
+                AND aes.id              = ae.id
+             WHERE ae.identity_pubkey = ?1
+               AND ae.relay_url       = ?2
+               AND ae.kind            = 24200
+               AND aes.scope_type     = 'owner_p'
+               AND ae.id NOT IN (
+                   SELECT id FROM observer_channel_index
+                   WHERE identity_pubkey = ?1
+                     AND relay_url       = ?2
+               )
+             ORDER BY ae.created_at DESC, ae.id DESC",
+        )
+        .map_err(|e| format!("prepare read_unindexed_observer_rows: {e}"))?;
+
+    let rows = stmt
+        .query_map(params![identity_pubkey, relay_url], |row| {
+            Ok((
+                row.get::<_, String>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, i64>(2)?,
+            ))
+        })
+        .map_err(|e| format!("query read_unindexed_observer_rows: {e}"))?;
+
+    rows.collect::<Result<Vec<_>, _>>()
+        .map_err(|e| format!("read read_unindexed_observer_rows row: {e}"))
+}
+
+/// Read a paginated page of archived kind 24200 events scoped to one channel,
+/// using the `observer_channel_index` as the primary lookup.
+///
+/// Returns `raw_json` strings in newest-first order. The compound cursor
+/// `(before_created_at, before_id)` works identically to `read_archived_events`.
+/// A page shorter than `limit` signals the archive is exhausted for this channel.
+#[allow(clippy::too_many_arguments)]
+pub fn read_archived_observer_events_for_channel(
+    conn: &Connection,
+    identity_pubkey: &str,
+    relay_url: &str,
+    channel_id: &str,
+    before_created_at: Option<i64>,
+    before_id: Option<&str>,
+    limit: i64,
+) -> Result<Vec<String>, String> {
+    let mut next_slot: usize = 4;
+    let mut extra_clauses = String::new();
+    let mut before_at_val: Option<i64> = None;
+    let mut before_id_val: Option<String> = None;
+
+    if let (Some(bat), Some(bid)) = (before_created_at, before_id) {
+        before_at_val = Some(bat);
+        before_id_val = Some(bid.to_owned());
+        extra_clauses.push_str(&format!(
+            " AND (oci.created_at < ?{next_slot} \
+              OR (oci.created_at = ?{next_slot} AND oci.id < ?{}))",
+            next_slot + 1,
+        ));
+        next_slot += 2;
+    }
+    let limit_slot = next_slot;
+
+    let sql = format!(
+        "SELECT ae.raw_json \
+         FROM observer_channel_index oci \
+         INNER JOIN archived_events ae \
+             ON ae.identity_pubkey = oci.identity_pubkey \
+            AND ae.relay_url       = oci.relay_url \
+            AND ae.id              = oci.id \
+         WHERE oci.identity_pubkey = ?1 \
+           AND oci.relay_url       = ?2 \
+           AND oci.channel_id      = ?3\
+         {extra_clauses}\
+         ORDER BY oci.created_at DESC, oci.id DESC \
+         LIMIT ?{limit_slot}",
+    );
+
+    let mut params_vec: Vec<Box<dyn rusqlite::ToSql>> = vec![
+        Box::new(identity_pubkey.to_owned()),
+        Box::new(relay_url.to_owned()),
+        Box::new(channel_id.to_owned()),
+    ];
+    if let (Some(bat), Some(bid)) = (before_at_val, before_id_val) {
+        params_vec.push(Box::new(bat));
+        params_vec.push(Box::new(bid));
+    }
+    params_vec.push(Box::new(limit));
+
+    let param_refs: Vec<&dyn rusqlite::ToSql> = params_vec.iter().map(|p| p.as_ref()).collect();
+
+    let mut stmt = conn
+        .prepare(&sql)
+        .map_err(|e| format!("prepare read_archived_observer_events_for_channel: {e}"))?;
+
+    let rows = stmt
+        .query_map(param_refs.as_slice(), |row| row.get::<_, String>(0))
+        .map_err(|e| format!("query read_archived_observer_events_for_channel: {e}"))?;
+
+    rows.collect::<Result<Vec<_>, _>>()
+        .map_err(|e| format!("read read_archived_observer_events_for_channel row: {e}"))
+}
+
 /// GC: delete orphaned event rows whose last scope row was just removed.
 ///
 /// Called after any batch deletion of scope rows. Uses a LEFT JOIN so only