nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

// SPDX-License-Identifier: BUSL-1.1

//! User CRUD operations for the system catalog.

use super::types::{METADATA, StoredUser, SystemCatalog, USERS, catalog_err};

impl SystemCatalog {
    /// Load all active users from the catalog.
    pub fn load_all_users(&self) -> crate::Result<Vec<StoredUser>> {
        let read_txn = self
            .db
            .begin_read()
            .map_err(|e| catalog_err("read txn", e))?;
        let table = read_txn
            .open_table(USERS)
            .map_err(|e| catalog_err("open users", e))?;

        let mut users = Vec::new();
        let range = table
            .range::<&str>(..)
            .map_err(|e| catalog_err("range users", e))?;
        for entry in range {
            let (_, value) = entry.map_err(|e| catalog_err("read entry", e))?;
            let user: StoredUser = zerompk::from_msgpack(value.value())
                .map_err(|e| catalog_err("deserialize user", e))?;
            users.push(user);
        }

        Ok(users)
    }

    /// Look up a single user by username. Matches the shape of
    /// `get_collection` / `get_trigger` / etc.
    pub fn get_user(&self, username: &str) -> crate::Result<Option<StoredUser>> {
        let read_txn = self
            .db
            .begin_read()
            .map_err(|e| catalog_err("read txn", e))?;
        let table = read_txn
            .open_table(USERS)
            .map_err(|e| catalog_err("open users", e))?;
        match table.get(username) {
            Ok(Some(value)) => {
                let user: StoredUser = zerompk::from_msgpack(value.value())
                    .map_err(|e| catalog_err("deserialize user", e))?;
                Ok(Some(user))
            }
            Ok(None) => Ok(None),
            Err(e) => Err(catalog_err("get user", e)),
        }
    }

    /// Write a user record to the catalog (insert or update).
    pub fn put_user(&self, user: &StoredUser) -> crate::Result<()> {
        let bytes = zerompk::to_msgpack_vec(user).map_err(|e| catalog_err("serialize user", e))?;

        let write_txn = self
            .db
            .begin_write()
            .map_err(|e| catalog_err("write txn", e))?;
        {
            let mut table = write_txn
                .open_table(USERS)
                .map_err(|e| catalog_err("open users", e))?;
            table
                .insert(user.username.as_str(), bytes.as_slice())
                .map_err(|e| catalog_err("insert user", e))?;
        }
        write_txn.commit().map_err(|e| catalog_err("commit", e))?;

        Ok(())
    }

    /// Atomically write a newly allocated user and advance the durable ID counter.
    pub fn put_user_with_next_user_id(
        &self,
        user: &StoredUser,
        next_user_id: u64,
    ) -> crate::Result<()> {
        let bytes = zerompk::to_msgpack_vec(user).map_err(|e| catalog_err("serialize user", e))?;
        let write_txn = self
            .db
            .begin_write()
            .map_err(|e| catalog_err("write txn", e))?;
        {
            let mut users = write_txn
                .open_table(USERS)
                .map_err(|e| catalog_err("open users", e))?;
            users
                .insert(user.username.as_str(), bytes.as_slice())
                .map_err(|e| catalog_err("insert user", e))?;
        }
        {
            let mut metadata = write_txn
                .open_table(METADATA)
                .map_err(|e| catalog_err("open metadata", e))?;
            metadata
                .insert("next_user_id", next_user_id.to_le_bytes().as_slice())
                .map_err(|e| catalog_err("insert next_user_id", e))?;
        }
        #[cfg(test)]
        if self
            .fail_next_user_counter_write
            .swap(false, std::sync::atomic::Ordering::SeqCst)
        {
            return Err(crate::Error::Storage {
                engine: "catalog".into(),
                detail: "injected user/counter transaction failure".into(),
            });
        }
        write_txn.commit().map_err(|e| catalog_err("commit", e))
    }

    /// Delete a user record from the catalog.
    pub fn delete_user(&self, username: &str) -> crate::Result<()> {
        let write_txn = self
            .db
            .begin_write()
            .map_err(|e| catalog_err("write txn", e))?;
        {
            let mut table = write_txn
                .open_table(USERS)
                .map_err(|e| catalog_err("open users", e))?;
            table
                .remove(username)
                .map_err(|e| catalog_err("remove user", e))?;
        }
        write_txn.commit().map_err(|e| catalog_err("commit", e))?;

        Ok(())
    }
}