radicle-feed 0.5.3

A feed service for Radicle
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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

use diesel::dsl::count;
use diesel::{
    insert_into, Connection, ExpressionMethods, OptionalExtension, PgConnection, QueryDsl,
    RunQueryDsl, SelectableHelper,
};
use snafu::ResultExt;

use radicle::cob::{ObjectId, TypeName};
use radicle::git::Oid;
use radicle::prelude::RepoId;

use crate::models::activity_feed::ActivityFeedOperation;
use crate::models::entry::{OperationEntry, TimelineEntry};
use crate::schema::activity_feed_timeline;
use crate::schema::{activity_feed_operations, seeded_radicle_repository};
use crate::storage::{FeedStorage, StorageStats};

pub struct PostgresStorage {
    connection: PgConnection,
}

impl PostgresStorage {
    pub fn new(db_url: String) -> Result<Self, snafu::Whatever> {
        let conn =
            PgConnection::establish(&db_url).whatever_context("Unable to establish connection")?;

        Ok(Self { connection: conn })
    }
}

impl FeedStorage for PostgresStorage {
    type Error = snafu::Whatever;

    fn get_last_processed_operation(
        &mut self,
        rid: &RepoId,
        cob_id: &ObjectId,
        typename: &TypeName,
    ) -> Result<Option<String>, Self::Error> {
        let rows = activity_feed_timeline::table
            .filter(activity_feed_timeline::repo.eq(rid.to_string()))
            .filter(activity_feed_timeline::cob_id.eq(cob_id.to_string()))
            .filter(activity_feed_timeline::typename.eq(typename.to_string()))
            .select(activity_feed_timeline::last_operation_id)
            .load::<Option<String>>(&mut self.connection)
            .whatever_context("Unable to get_last_processed_operation")?;

        if rows.is_empty() {
            Ok(None)
        } else {
            Ok(rows[0].clone())
        }
    }

    fn get_operation_by_id(
        &mut self,
        id: &radicle::git::Oid,
    ) -> Result<ActivityFeedOperation, Self::Error> {
        activity_feed_operations::table
            .filter(activity_feed_operations::operation_id.eq(id.to_string()))
            .select(ActivityFeedOperation::as_select())
            .first::<ActivityFeedOperation>(&mut self.connection)
            .whatever_context("Unable to get_operation_by_id")
    }

    fn operation_exists(&mut self, operation_id: &Oid) -> Result<bool, Self::Error> {
        let rows = activity_feed_operations::table
            .filter(activity_feed_operations::operation_id.eq(operation_id.to_string()))
            .select(activity_feed_operations::operation_id)
            .first::<String>(&mut self.connection)
            .optional()
            .whatever_context("Unable to load last feed operation")?;

        Ok(rows.is_some())
    }

    /// Inserts a new timeline entry into the database, in case of conflict it tries to update the last operation ID.
    fn insert_timeline_entry(&mut self, entry: &TimelineEntry) -> Result<(), Self::Error> {
        insert_into(activity_feed_timeline::table)
            .values(entry)
            .on_conflict((
                activity_feed_timeline::repo,
                activity_feed_timeline::node,
                activity_feed_timeline::cob_id,
            ))
            .do_update()
            .set(activity_feed_timeline::last_operation_id.eq(entry.last_operation_id.clone()))
            .execute(&mut self.connection)
            .whatever_context("Unable to insert")?;

        Ok(())
    }

    fn resolve_rid(&mut self, rid: &RepoId) -> Result<Option<String>, Self::Error> {
        seeded_radicle_repository::table
            .filter(seeded_radicle_repository::repository_id.eq(rid.to_string()))
            .select(seeded_radicle_repository::alias)
            .first::<Option<String>>(&mut self.connection)
            .whatever_context("Unable to resolve rid")
    }

    fn insert_batch(&mut self, entries: &[OperationEntry]) -> Result<(), Self::Error> {
        if entries.is_empty() {
            return Ok(());
        }

        insert_into(activity_feed_operations::table)
            .values(entries)
            .execute(&mut self.connection)
            .whatever_context("context")?;

        Ok(())
    }

    fn get_stats(&mut self) -> Result<StorageStats, Self::Error> {
        let mut stats = StorageStats::default();

        let statement_total_operations = activity_feed_operations::table
            .count()
            .first::<i64>(&mut self.connection)
            .whatever_context("context")?;
        stats.total_operations = statement_total_operations as u64;

        let statement_operations_by_type = activity_feed_operations::table
            .count()
            .group_by(activity_feed_operations::typename)
            .select((
                activity_feed_operations::typename,
                count(activity_feed_operations::id),
            ))
            .load::<(String, i64)>(&mut self.connection)
            .whatever_context("context")?;

        for (typename, count) in statement_operations_by_type {
            stats.operations_by_type.insert(typename, count as u64);
        }

        let statement_tracked_objects = activity_feed_operations::table
            .count()
            .get_results::<i64>(&mut self.connection)
            .whatever_context("context")?;

        for count in statement_tracked_objects {
            stats.tracked_objects = count as u64;
        }

        Ok(stats)
    }
}