radicle_feed/storage/
postgres.rs

1use deadpool_postgres::{Object, Pool, Runtime};
2use snafu::ResultExt;
3use tokio_postgres::types::ToSql;
4use tokio_postgres::{NoTls, Row, Statement};
5
6use radicle::cob::{ObjectId, TypeName};
7use radicle::git::Oid;
8use radicle::prelude::RepoId;
9
10use crate::entry::{OperationEntry, TimelineEntry};
11use crate::radicle_extra::sql;
12use crate::storage::{FeedStorage, StorageStats};
13
14pub struct PostgresStorage {
15    connection: Pool,
16}
17
18impl PostgresStorage {
19    pub fn new(db_url: String) -> Result<Self, snafu::Whatever> {
20        let pool = deadpool_postgres::Config {
21            url: Some(db_url),
22            ..Default::default()
23        }
24        .create_pool(Some(Runtime::Tokio1), NoTls)
25        .whatever_context("Unable to create pool")?;
26
27        Ok(Self { connection: pool })
28    }
29
30    pub async fn client(&self) -> Result<Object, snafu::Whatever> {
31        self.connection
32            .get()
33            .await
34            .whatever_context("Unable to get a client")
35    }
36
37    pub async fn statement(
38        &self,
39        client: &Object,
40        query: &str,
41    ) -> Result<Statement, snafu::Whatever> {
42        client
43            .prepare_cached(query)
44            .await
45            .whatever_context("Unable to prepare statement")
46    }
47
48    pub async fn query(
49        &self,
50        client: &Object,
51        stmt: &Statement,
52        params: &[&(dyn ToSql + Sync)],
53    ) -> Result<Vec<Row>, snafu::Whatever> {
54        client
55            .query(stmt, params)
56            .await
57            .whatever_context("Unable to query db")
58    }
59}
60
61impl FeedStorage for PostgresStorage {
62    type Error = snafu::Whatever;
63
64    async fn get_last_processed_operation(
65        &self,
66        rid: &RepoId,
67        cob_id: &ObjectId,
68        typename: &TypeName,
69    ) -> Result<Option<sql::Oid>, Self::Error> {
70        let client = self.client().await?;
71        let stmt = self
72            .statement(
73                &client,
74                "SELECT last_operation_id FROM activity_feed_timeline WHERE repo = $1 AND cob_id = $2 AND typename = $3",
75            )
76            .await?;
77        let rows = self
78            .query(
79                &client,
80                &stmt,
81                &[&rid.to_string(), &cob_id.to_string(), &typename.to_string()],
82            )
83            .await?;
84
85        if rows.is_empty() {
86            return Ok(None);
87        }
88
89        Ok(rows[0].get::<usize, Option<sql::Oid>>(0))
90    }
91
92    async fn operation_exists(&self, operation_id: &Oid) -> Result<bool, Self::Error> {
93        let client = self.client().await?;
94        let statement = self
95            .statement(
96                &client,
97                "SELECT 1 FROM activity_feed_operations WHERE operation_id = $1 LIMIT 1",
98            )
99            .await?;
100        let row = self
101            .query(&client, &statement, &[&operation_id.to_string()])
102            .await?;
103
104        Ok(!row.is_empty())
105    }
106
107    /// Inserts a new timeline entry into the database, in case of conflict it tries to update the last operation ID.
108    async fn insert_timeline_entry(&mut self, entry: &TimelineEntry) -> Result<(), Self::Error> {
109        let client = self.client().await?;
110        let stmt = self.statement(&client, "INSERT INTO activity_feed_timeline (repo, node, cob_id, cob_title, cob_status, typename, last_operation_id, operations)
111            VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (repo, node, cob_id) DO UPDATE SET last_operation_id = EXCLUDED.last_operation_id").await?;
112
113        self.query(
114            &client,
115            &stmt,
116            &[
117                &entry.repo.to_string(),
118                &entry.node.to_string(),
119                &entry.cob_id.to_string(),
120                &entry.cob_title,
121                &entry.cob_status,
122                &entry.typename.to_string(),
123                &entry.last_operation_id.map(|o| o.to_string()),
124                &entry.operations,
125            ],
126        )
127        .await?;
128
129        Ok(())
130    }
131
132    async fn insert_batch(&mut self, entries: &[OperationEntry]) -> Result<(), Self::Error> {
133        if entries.is_empty() {
134            return Ok(());
135        }
136
137        let client = self.client().await?;
138        let stmt = self.statement(&client, "INSERT INTO activity_feed_operations (rid, operation_id, author, author_alias, actions, created_at, typename)
139            VALUES ($1, $2, $3, $4, $5, $6, $7)").await?;
140
141        for entry in entries {
142            self.query(
143                &client,
144                &stmt,
145                &[
146                    &entry.rid.to_string(),
147                    &entry.operation_id.to_string(),
148                    &entry.author.to_string(),
149                    &entry.author_alias.as_ref().map(|alias| alias.to_string()),
150                    &entry.actions,
151                    &(entry.timestamp.as_secs() as i32),
152                    &entry.typename.to_string(),
153                ],
154            )
155            .await?;
156        }
157
158        Ok(())
159    }
160
161    async fn get_stats(&self) -> Result<StorageStats, Self::Error> {
162        let mut stats = StorageStats::default();
163
164        let client = self.client().await?;
165        let statement_total_operations = self
166            .statement(&client, "SELECT COUNT(*) FROM activity_feed_operations")
167            .await?;
168        let rows = self
169            .query(&client, &statement_total_operations, &[])
170            .await?;
171        let count: i64 = rows[0].get(0);
172        stats.total_operations = count as u64;
173
174        let statement_operations_by_type = self
175            .statement(
176                &client,
177                "SELECT typename, COUNT(*) FROM activity_feed_operations GROUP BY typename",
178            )
179            .await?;
180        let rows = self
181            .query(&client, &statement_operations_by_type, &[])
182            .await?;
183
184        for row in rows {
185            let typename: String = row.get(0);
186            let count: i64 = row.get(1);
187            stats.operations_by_type.insert(typename, count as u64);
188        }
189
190        let statement_tracked_objects = self
191            .statement(&client, "SELECT COUNT(*) FROM activity_feed_operations")
192            .await?;
193        let rows = self.query(&client, &statement_tracked_objects, &[]).await?;
194
195        for row in rows {
196            let count: i64 = row.get(0);
197            stats.tracked_objects = count as u64;
198        }
199
200        Ok(stats)
201    }
202}
radicle_feed/storage/postgres.rs

radicle_feed/storage/
postgres.rs
