implemented primary mount shared with me

crates/pdfs-core/src/db/nodes.rs

@@ -1,11 +1,13 @@
 //! Node rows: the persisted mirror of the remote tree, plus the trigram
 //! full-text index over names and paths that backs `Request::Search`.
 
-use rusqlite::{OptionalExtension, params};
+use std::collections::HashSet;
+
+use rusqlite::{OptionalExtension, Transaction, params};
 
 use super::Db;
-use crate::Result;
-use proton_drive_rs::proton_sdk::ids::NodeUid;
+use crate::{Access, Result};
+use proton_drive_rs::proton_sdk::ids::{LinkId, NodeUid, VolumeId};
 use proton_drive_rs::{Node, NodeKind};
 
 use super::utils::{TRIGRAM_MIN, collect_pairs, like_escape, pair, path_of};
@@ -22,6 +24,11 @@ pub struct SearchHit {
     pub path: String,
 }
 
+pub struct PublishedSharedRoot {
+    pub node: Node,
+    pub access: Access,
+}
+
 impl Db {
     pub fn upsert_node(&self, node: &Node) -> Result<()> {
         self.upsert_nodes(std::slice::from_ref(node))
@@ -42,87 +49,176 @@ impl Db {
         let mut conn = self.conn.lock();
         let tx = conn.transaction()?;
         for node in nodes {
-            let json = serde_json::to_string(node)?;
-            let uid = node.uid.to_string();
-            tx.execute(
-                "INSERT INTO nodes
-                   (uid, parent_uid, name, is_dir, size, mtime, trashed, node_json)
-                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
-                 ON CONFLICT(uid) DO UPDATE SET
-                   parent_uid = excluded.parent_uid,
-                   name       = excluded.name,
-                   is_dir     = excluded.is_dir,
-                   size       = excluded.size,
-                   mtime      = excluded.mtime,
-                   trashed    = excluded.trashed,
-                   node_json  = excluded.node_json",
-                params![
-                    uid,
-                    node.parent_uid.as_ref().map(|u| u.to_string()),
-                    node.name,
-                    node.is_folder() as i64,
-                    node_size(node),
-                    node.modification_time,
-                    node.trashed as i64,
-                    json,
-                ],
-            )?;
-            // FTS5 has no UPSERT, so refresh the row by delete-then-insert.
-            // Trashed nodes are kept out of the index entirely so they never
-            // surface in search results.
-            //
-            // Keyed by the node's rowid, never by its uid: the uid is not an
-            // indexed FTS column, so deleting by it scans the entire index once
-            // per node written (B12). The rowid is stable across the upsert
-            // above — `ON CONFLICT DO UPDATE` keeps the existing row — so it is
-            // the same key before and after.
-            let rowid: i64 = tx.query_row(
-                "SELECT rowid FROM nodes WHERE uid = ?1",
-                params![uid],
-                |row| row.get(0),
-            )?;
-            // A folder rename/move changes every descendant's searchable path.
-            // Refresh the affected subtree, while retaining efficient rowid
-            // deletes for the overwhelmingly common one-file case.
-            let mut affected = vec![rowid];
-            if node.is_folder() {
-                let mut stmt = tx.prepare(
-                    "WITH RECURSIVE descendants(rowid, uid) AS (
-                       SELECT rowid, uid FROM nodes WHERE parent_uid = ?1
-                       UNION ALL
-                       SELECT n.rowid, n.uid FROM nodes n
-                         JOIN descendants d ON n.parent_uid = d.uid
-                     ) SELECT rowid FROM descendants",
-                )?;
-                for descendant in stmt.query_map(params![uid], |row| row.get(0))? {
-                    affected.push(descendant?);
-                }
-            }
-            for affected_rowid in affected {
-                tx.execute(
-                    "DELETE FROM nodes_fts WHERE rowid = ?1",
-                    params![affected_rowid],
-                )?;
-                let indexed: Option<(String, String)> = tx
-                    .query_row(
-                        "SELECT name, uid FROM nodes WHERE rowid = ?1 AND trashed = 0",
-                        params![affected_rowid],
-                        |row| Ok((row.get(0)?, row.get(1)?)),
-                    )
-                    .optional()?;
-                if let Some((name, indexed_uid)) = indexed {
-                    let path = path_of(&tx, &indexed_uid)?;
-                    tx.execute(
-                        "INSERT INTO nodes_fts (rowid, name, path) VALUES (?1, ?2, ?3)",
-                        params![affected_rowid, name, path],
-                    )?;
-                }
-            }
+            upsert_node_tx(&tx, node)?;
         }
         tx.commit()?;
         Ok(())
     }
 
+    /// Return visible direct children without consulting the parent's `listed`
+    /// flag. Synthetic shared listings use this to serve the last completed
+    /// snapshot while offline, including after a live event expired its TTL.
+    pub fn visible_children(&self, parent: &NodeUid) -> Result<Vec<Node>> {
+        let conn = self.conn.lock();
+        let mut stmt = conn.prepare(
+            "SELECT node_json FROM nodes
+              WHERE parent_uid = ?1 AND trashed = 0 AND node_json IS NOT NULL
+              ORDER BY name, uid",
+        )?;
+        let rows = stmt.query_map([parent.to_string()], |row| row.get::<_, String>(0))?;
+        let mut nodes = Vec::new();
+        for row in rows {
+            nodes.push(serde_json::from_str(&row?)?);
+        }
+        Ok(nodes)
+    }
+
+    /// Atomically publish one completed `Shared with me` root listing.
+    ///
+    /// Roots absent from `accepted` and every persisted descendant are
+    /// tombstoned, not deleted. Accepted roots omitted from materialization keep
+    /// their visible snapshot but are downgraded to Viewer until membership is
+    /// verified again. Both cases preserve queued operations and staged writes.
+    pub fn publish_shared_roots(
+        &self,
+        parent: &NodeUid,
+        accepted: &[NodeUid],
+        roots: &[PublishedSharedRoot],
+    ) -> Result<Vec<NodeUid>> {
+        let parent = parent.to_string();
+        let present: HashSet<String> = accepted.iter().map(NodeUid::to_string).collect();
+        let materialized: HashSet<String> = roots
+            .iter()
+            .map(|published| published.node.uid.to_string())
+            .collect();
+        let mut conn = self.conn.lock();
+        let tx = conn.transaction()?;
+
+        let existing = direct_child_uids_tx(&tx, &parent)?;
+        let removed: Vec<String> = existing
+            .into_iter()
+            .filter(|uid| !present.contains(uid))
+            .collect();
+
+        for published in roots {
+            if !present.contains(&published.node.uid.to_string()) {
+                continue;
+            }
+            upsert_node_tx(&tx, &published.node)?;
+            upsert_share_access_tx(&tx, &published.node.uid.to_string(), published.access)?;
+        }
+        for uid in present.difference(&materialized) {
+            upsert_share_access_tx(&tx, uid, Access::Viewer)?;
+        }
+        for root in &removed {
+            tombstone_subtree_tx(&tx, root)?;
+            upsert_share_access_tx(&tx, root, Access::Viewer)?;
+        }
+        tx.execute(
+            "UPDATE nodes SET listed = 1 WHERE uid = ?1",
+            params![parent],
+        )?;
+        tx.commit()?;
+
+        Ok(removed
+            .into_iter()
+            .filter_map(|uid| parse_node_uid(&uid))
+            .collect())
+    }
+
+    /// Atomically withdraw a deleted foreign subtree and deny stale handles.
+    ///
+    /// Node rows and queued operations remain so retries and staged writes keep
+    /// their references. FTS and completed-listing state are withdrawn for the
+    /// whole subtree, and the deleted UID becomes a fail-closed authority.
+    pub fn tombstone_foreign_subtree(&self, uid: &NodeUid) -> Result<()> {
+        let uid = uid.to_string();
+        let mut conn = self.conn.lock();
+        let tx = conn.transaction()?;
+        tombstone_subtree_tx(&tx, &uid)?;
+        upsert_share_access_tx(&tx, &uid, Access::Viewer)?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    /// Publish a completed foreign-folder listing from its authoritative UID
+    /// list and the subset that materialized successfully. Accepted-but-omitted
+    /// children retain their previous snapshot; only absent UIDs are tombstoned.
+    pub fn publish_foreign_children(
+        &self,
+        parent: &NodeUid,
+        accepted: &[NodeUid],
+        materialized: &[Node],
+    ) -> Result<Vec<NodeUid>> {
+        let parent = parent.to_string();
+        let present: HashSet<String> = accepted.iter().map(NodeUid::to_string).collect();
+        let mut conn = self.conn.lock();
+        let tx = conn.transaction()?;
+        let removed: Vec<String> = direct_child_uids_tx(&tx, &parent)?
+            .into_iter()
+            .filter(|uid| !present.contains(uid))
+            .collect();
+        for node in materialized {
+            if present.contains(&node.uid.to_string()) {
+                upsert_node_tx(&tx, node)?;
+            }
+        }
+        for uid in &removed {
+            tombstone_subtree_tx(&tx, uid)?;
+        }
+        tx.execute(
+            "UPDATE nodes SET listed = 1 WHERE uid = ?1",
+            params![parent],
+        )?;
+        tx.commit()?;
+        Ok(removed
+            .into_iter()
+            .filter_map(|uid| parse_node_uid(&uid))
+            .collect())
+    }
+
+    /// Persist the synthetic root and its Viewer authority only when its
+    /// presentation state changed. Returning `false` guarantees no node/FTS
+    /// write occurred, which keeps every cached root lookup O(1).
+    pub fn ensure_virtual_root(&self, node: &Node) -> Result<bool> {
+        let uid = node.uid.to_string();
+        let desired_parent = node.parent_uid.as_ref().map(NodeUid::to_string);
+        let mut conn = self.conn.lock();
+        let tx = conn.transaction()?;
+        let stored: Option<(Option<String>, String, i64, i64)> = tx
+            .query_row(
+                "SELECT parent_uid, name, is_dir, trashed FROM nodes WHERE uid = ?1",
+                params![uid],
+                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
+            )
+            .optional()?;
+        let access: Option<String> = tx
+            .query_row(
+                "SELECT access FROM share_access WHERE root_uid = ?1",
+                params![uid],
+                |row| row.get(0),
+            )
+            .optional()?;
+        let unchanged = stored.is_some_and(|(parent, name, is_dir, trashed)| {
+            parent == desired_parent
+                && name == node.name
+                && is_dir == 1
+                && trashed == node.trashed as i64
+        }) && access.as_deref() == Some(Access::Viewer.as_db_str());
+        if unchanged {
+            tx.commit()?;
+            return Ok(false);
+        }
+        tx.execute(
+            "INSERT INTO share_access (root_uid, access) VALUES (?1, ?2)
+             ON CONFLICT(root_uid) DO UPDATE SET access = excluded.access",
+            params![uid, Access::Viewer.as_db_str()],
+        )?;
+        upsert_node_tx(&tx, node)?;
+        tx.commit()?;
+        Ok(true)
+    }
+
     /// Drop a node row (delete or trash from the hot cache). Children rows are
     /// not cascaded here; the daemon forgets a whole subtree node-by-node.
     pub fn delete_node(&self, uid: &NodeUid) -> Result<()> {
@@ -413,6 +509,168 @@ impl Db {
     // in sync on every store/read/evict, so it is authoritative for eviction.
 }
 
+fn upsert_node_tx(tx: &Transaction<'_>, node: &Node) -> Result<()> {
+    let json = serde_json::to_string(node)?;
+    let uid = node.uid.to_string();
+    tx.execute(
+        "INSERT INTO nodes
+           (uid, parent_uid, name, is_dir, size, mtime, trashed, node_json)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
+         ON CONFLICT(uid) DO UPDATE SET
+           parent_uid = excluded.parent_uid,
+           name       = excluded.name,
+           is_dir     = excluded.is_dir,
+           size       = excluded.size,
+           mtime      = excluded.mtime,
+           trashed    = excluded.trashed,
+           node_json  = excluded.node_json",
+        params![
+            uid,
+            node.parent_uid.as_ref().map(|u| u.to_string()),
+            node.name,
+            node.is_folder() as i64,
+            node_size(node),
+            node.modification_time,
+            node.trashed as i64,
+            json,
+        ],
+    )?;
+    // FTS5 has no UPSERT. Folder moves also change every descendant path, so
+    // refresh the whole affected subtree while keeping rowid deletes indexed.
+    let rowid: i64 = tx.query_row(
+        "SELECT rowid FROM nodes WHERE uid = ?1",
+        params![uid],
+        |row| row.get(0),
+    )?;
+    let mut affected = vec![rowid];
+    if node.is_folder() {
+        let mut stmt = tx.prepare(
+            "WITH RECURSIVE descendants(rowid, uid) AS (
+               SELECT rowid, uid FROM nodes WHERE parent_uid = ?1
+               UNION ALL
+               SELECT n.rowid, n.uid FROM nodes n
+                 JOIN descendants d ON n.parent_uid = d.uid
+             ) SELECT rowid FROM descendants",
+        )?;
+        for descendant in stmt.query_map(params![uid], |row| row.get(0))? {
+            affected.push(descendant?);
+        }
+    }
+    for affected_rowid in affected {
+        tx.execute(
+            "DELETE FROM nodes_fts WHERE rowid = ?1",
+            params![affected_rowid],
+        )?;
+        let indexed: Option<(String, String)> = tx
+            .query_row(
+                "SELECT name, uid FROM nodes WHERE rowid = ?1",
+                params![affected_rowid],
+                |row| Ok((row.get(0)?, row.get(1)?)),
+            )
+            .optional()?;
+        if let Some((name, indexed_uid)) = indexed
+            && node_is_indexable_tx(tx, &indexed_uid)?
+        {
+            let path = path_of(tx, &indexed_uid)?;
+            tx.execute(
+                "INSERT INTO nodes_fts (rowid, name, path) VALUES (?1, ?2, ?3)",
+                params![affected_rowid, name, path],
+            )?;
+        }
+    }
+    Ok(())
+}
+
+fn direct_child_uids_tx(tx: &Transaction<'_>, parent: &str) -> Result<Vec<String>> {
+    let mut stmt = tx.prepare("SELECT uid FROM nodes WHERE parent_uid = ?1")?;
+    let rows = stmt.query_map([parent], |row| row.get::<_, String>(0))?;
+    let mut uids = Vec::new();
+    for row in rows {
+        uids.push(row?);
+    }
+    Ok(uids)
+}
+
+fn node_is_indexable_tx(tx: &Transaction<'_>, uid: &str) -> Result<bool> {
+    let indexable: i64 = tx.query_row(
+        "WITH RECURSIVE ancestors(uid, parent_uid, trashed, path) AS (
+           SELECT uid, parent_uid, trashed, char(31) || uid || char(31)
+             FROM nodes WHERE uid = ?1
+           UNION ALL
+           SELECT n.uid, n.parent_uid, n.trashed,
+                  a.path || n.uid || char(31)
+             FROM ancestors a JOIN nodes n ON n.uid = a.parent_uid
+            WHERE instr(a.path, char(31) || n.uid || char(31)) = 0
+         )
+         SELECT CASE
+           WHEN COUNT(*) = 0 THEN 0
+           WHEN MAX(trashed) != 0 THEN 0
+           WHEN MAX(CASE WHEN parent_uid IS NULL THEN 1 ELSE 0 END) = 0 THEN 0
+           ELSE 1
+         END
+         FROM ancestors",
+        [uid],
+        |row| row.get(0),
+    )?;
+    Ok(indexable == 1)
+}
+
+fn tombstone_subtree_tx(tx: &Transaction<'_>, root_uid: &str) -> Result<()> {
+    let rows = {
+        let mut stmt = tx.prepare(
+            "WITH RECURSIVE subtree(rowid, uid, node_json) AS (
+               SELECT rowid, uid, node_json FROM nodes WHERE uid = ?1
+               UNION
+               SELECT n.rowid, n.uid, n.node_json FROM nodes n
+                 JOIN subtree s ON n.parent_uid = s.uid
+             )
+             SELECT rowid, uid, node_json FROM subtree",
+        )?;
+        let mapped = stmt.query_map([root_uid], |row| {
+            Ok((
+                row.get::<_, i64>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, Option<String>>(2)?,
+            ))
+        })?;
+        let mut rows = Vec::new();
+        for row in mapped {
+            rows.push(row?);
+        }
+        rows
+    };
+    for (rowid, uid, json) in rows {
+        let json = match json {
+            Some(json) => {
+                let mut node: Node = serde_json::from_str(&json)?;
+                node.trashed = true;
+                Some(serde_json::to_string(&node)?)
+            }
+            None => None,
+        };
+        tx.execute(
+            "UPDATE nodes SET trashed = 1, listed = 0, node_json = ?2 WHERE uid = ?1",
+            params![uid, json],
+        )?;
+        tx.execute("DELETE FROM nodes_fts WHERE rowid = ?1", [rowid])?;
+    }
+    Ok(())
+}
+
+fn upsert_share_access_tx(tx: &Transaction<'_>, uid: &str, access: Access) -> Result<()> {
+    tx.execute(
+        "INSERT INTO share_access (root_uid, access) VALUES (?1, ?2)
+         ON CONFLICT(root_uid) DO UPDATE SET access = excluded.access",
+        params![uid, access.as_db_str()],
+    )?;
+    Ok(())
+}
+
+fn parse_node_uid(value: &str) -> Option<NodeUid> {
+    let (volume, link) = value.split_once('~')?;
+    Some(NodeUid::new(VolumeId::from(volume), LinkId::from(link)))
+}
+
 /// Unique quoted character trigrams suitable for an FTS5 OR expression.
 pub(super) fn candidate_trigrams(query: &str) -> Vec<String> {
     let chars: Vec<char> = query.to_lowercase().chars().collect();