radicle-feed 0.3.0

use deadpool_postgres::{Object, Pool, Runtime};
use snafu::ResultExt;
use tokio_postgres::types::ToSql;
use tokio_postgres::{NoTls, Row, Statement};

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

use crate::entry::{OperationEntry, TimelineEntry};
use crate::radicle_extra::sql;
use crate::storage::{FeedStorage, StorageStats};

pub struct PostgresStorage {
    connection: Pool,
}

impl PostgresStorage {
    pub fn new(db_url: String) -> Result<Self, snafu::Whatever> {
        let pool = deadpool_postgres::Config {
            url: Some(db_url),
            ..Default::default()
        }
        .create_pool(Some(Runtime::Tokio1), NoTls)
        .whatever_context("Unable to create pool")?;

        Ok(Self { connection: pool })
    }

    pub async fn client(&self) -> Result<Object, snafu::Whatever> {
        self.connection
            .get()
            .await
            .whatever_context("Unable to get a client")
    }

    pub async fn statement(
        &self,
        client: &Object,
        query: &str,
    ) -> Result<Statement, snafu::Whatever> {
        client
            .prepare_cached(query)
            .await
            .whatever_context("Unable to prepare statement")
    }

    pub async fn query(
        &self,
        client: &Object,
        stmt: &Statement,
        params: &[&(dyn ToSql + Sync)],
    ) -> Result<Vec<Row>, snafu::Whatever> {
        client
            .query(stmt, params)
            .await
            .whatever_context("Unable to query db")
    }
}

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

    async fn get_last_processed_operation(
        &self,
        rid: &RepoId,
        cob_id: &ObjectId,
        typename: &TypeName,
    ) -> Result<Option<sql::Oid>, Self::Error> {
        let client = self.client().await?;
        let stmt = self
            .statement(
                &client,
                "SELECT last_operation_id FROM activity_feed_timeline WHERE repo = $1 AND cob_id = $2 AND typename = $3",
            )
            .await?;
        let rows = self
            .query(
                &client,
                &stmt,
                &[&rid.to_string(), &cob_id.to_string(), &typename.to_string()],
            )
            .await?;

        if rows.is_empty() {
            return Ok(None);
        }

        Ok(rows[0].get::<usize, Option<sql::Oid>>(0))
    }

    async fn operation_exists(&self, operation_id: &Oid) -> Result<bool, Self::Error> {
        let client = self.client().await?;
        let statement = self
            .statement(
                &client,
                "SELECT 1 FROM activity_feed_operations WHERE operation_id = $1 LIMIT 1",
            )
            .await?;
        let row = self
            .query(&client, &statement, &[&operation_id.to_string()])
            .await?;

        Ok(!row.is_empty())
    }

    /// Inserts a new timeline entry into the database, in case of conflict it tries to update the last operation ID.
    async fn insert_timeline_entry(&mut self, entry: &TimelineEntry) -> Result<(), Self::Error> {
        let client = self.client().await?;
        let stmt = self.statement(&client, "INSERT INTO activity_feed_timeline (repo, node, cob_id, cob_title, cob_status, typename, last_operation_id, operations)
            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?;

        self.query(
            &client,
            &stmt,
            &[
                &entry.repo.to_string(),
                &entry.node.to_string(),
                &entry.cob_id.to_string(),
                &entry.cob_title,
                &entry.cob_status,
                &entry.typename.to_string(),
                &entry.last_operation_id.map(|o| o.to_string()),
                &entry.operations,
            ],
        )
        .await?;

        Ok(())
    }

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

        let client = self.client().await?;
        let stmt = self.statement(&client, "INSERT INTO activity_feed_operations (rid, operation_id, author, author_alias, actions, created_at, typename)
            VALUES ($1, $2, $3, $4, $5, $6, $7)").await?;

        for entry in entries {
            self.query(
                &client,
                &stmt,
                &[
                    &entry.rid.to_string(),
                    &entry.operation_id.to_string(),
                    &entry.author.to_string(),
                    &entry.author_alias.as_ref().map(|alias| alias.to_string()),
                    &entry.actions,
                    &(entry.timestamp.as_secs() as i32),
                    &entry.typename.to_string(),
                ],
            )
            .await?;
        }

        Ok(())
    }

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

        let client = self.client().await?;
        let statement_total_operations = self
            .statement(&client, "SELECT COUNT(*) FROM activity_feed_operations")
            .await?;
        let rows = self
            .query(&client, &statement_total_operations, &[])
            .await?;
        let count: i64 = rows[0].get(0);
        stats.total_operations = count as u64;

        let statement_operations_by_type = self
            .statement(
                &client,
                "SELECT typename, COUNT(*) FROM activity_feed_operations GROUP BY typename",
            )
            .await?;
        let rows = self
            .query(&client, &statement_operations_by_type, &[])
            .await?;

        for row in rows {
            let typename: String = row.get(0);
            let count: i64 = row.get(1);
            stats.operations_by_type.insert(typename, count as u64);
        }

        let statement_tracked_objects = self
            .statement(&client, "SELECT COUNT(*) FROM activity_feed_operations")
            .await?;
        let rows = self.query(&client, &statement_tracked_objects, &[]).await?;

        for row in rows {
            let count: i64 = row.get(0);
            stats.tracked_objects = count as u64;
        }

        Ok(stats)
    }
}