Merge pull request #17 from constata-eu/re_score

Re score

api/Cargo.toml

@@ -25,6 +25,7 @@ rust_decimal = "*"
 rust_decimal_macros = "*"
 tokio = { version = "1.2.0", features = ["full"] }
 chrono = { version = "0.4", features = ["serde"] }
+time = "0.3.37"
 jwt-simple = "0.11.6"
 base64-serde = "0.7.0"
 base64 = "0.21.2"

api/migrations/20241227110000_store_campaign_tweet_metrics.sql

@@ -0,0 +1,21 @@
+ALTER TABLE campaigns ADD COLUMN impression_count int4 NOT NULL default 0;
+ALTER TABLE campaigns ADD COLUMN reply_count int4 NOT NULL default 0;
+ALTER TABLE campaigns ADD COLUMN repost_count int4 NOT NULL default 0;
+ALTER TABLE campaigns ADD COLUMN like_count int4 NOT NULL default 0;
+
+ALTER TABLE handles ALTER COLUMN score SET DEFAULT '0x0000000000000000000000000000000000000000000000000000000000000000';
+ALTER TABLE handles ALTER COLUMN score DROP NOT NULL;
+ALTER TABLE handles ALTER COLUMN last_scoring DROP NOT NULL;
+
+ALTER TABLE handles ADD COLUMN avg_impression_count int4 NOT NULL default 0;
+ALTER TABLE handles ADD COLUMN avg_reply_count int4 NOT NULL default 0;
+ALTER TABLE handles ADD COLUMN avg_repost_count int4 NOT NULL default 0;
+ALTER TABLE handles ADD COLUMN avg_like_count int4 NOT NULL default 0;
+ALTER TABLE handles ADD COLUMN scored_tweet_count int4 NOT NULL default 0;
+ALTER TABLE handles ADD COLUMN legacy_score varchar NOT NULL default '0x0000000000000000000000000000000000000000000000000000000000000000';
+
+
+UPDATE handles SET legacy_score = score;
+UPDATE handles SET last_scoring = NULL;
+
+INSERT INTO topics (name) VALUES ('speaks_english'), ('speaks_spanish');

api/src/api.rs

@@ -9,7 +9,7 @@ use juniper::{
 use sqlx_models_orm::*;
 
 use juniper_rocket::{graphiql_source, GraphQLResponse};
-use rocket::{http::Status, State};
+use rocket::{http::Status, serde::json::Json, State};
 
 mod current_session;
 use current_session::*;
@@ -33,6 +33,8 @@ mod audit_log_entry;
 use audit_log_entry::*;
 mod stats;
 use stats::*;
+mod topic;
+use topic::*;
 
 type JsonResult<T> = AsamiResult<Json<T>>;
 
@@ -303,6 +305,7 @@ make_graphql_query! {
     [Collab, allCollabs, allCollabsMeta, "_allCollabsMeta", CollabFilter, i32],
     [CampaignPreference, allCampaignPreferences, allCampaignPreferencesMeta, "_allCampaignPreferencesMeta", CampaignPreferenceFilter, i32],
     [OnChainJob, allOnChainJobs, allOnChainJobsMeta, "_allOnChainJobsMeta", OnChainJobFilter, i32],
+    [Topic, allTopics, allTopicsMeta, "_allTopicsMeta", TopicFilter, i32],
     [AuditLogEntry, allAuditLogEntries, allAuditLogEntriesMeta, "_allAuditLogEntriesMeta", AuditLogEntryFilter, i32],
   }
 

api/src/api/current_session.rs

@@ -175,6 +175,8 @@ impl CurrentSession {
             }
         }
 
+        app.account().hydrate_on_chain_columns_for([session.account_id().clone()].iter()).await?;
+
         Ok(session)
     }
 

api/src/api/handle.rs

@@ -21,6 +21,8 @@ pub struct Handle {
     user_id: Option<String>,
     #[graphql(description = "The score given to this handle by Asami's admin.")]
     score: Option<String>,
+    #[graphql(description = "Topics assigned to this handle")]
+    topic_ids: Vec<i32>,
     #[graphql(description = "Status of this handle.")]
     status: HandleStatus,
     #[graphql(description = "Collabs made")]
@@ -78,13 +80,15 @@ impl Showable<models::Handle, HandleFilter> for Handle {
     }
 
     async fn db_to_graphql(_context: &Context, d: models::Handle) -> AsamiResult<Self> {
+        let topic_ids = d.topic_ids().await?;
         Ok(Handle {
             id: d.attrs.id,
             account_id: hex_to_i32(&d.attrs.account_id)?,
             site: d.attrs.site,
             username: d.attrs.username,
             user_id: d.attrs.user_id,
             score: d.attrs.score,
+            topic_ids,
             status: d.attrs.status,
             total_collabs: d.attrs.total_collabs,
             total_collab_rewards: d.attrs.total_collab_rewards,

api/src/api/topic.rs

@@ -0,0 +1,57 @@
+
+use super::{
+    models::{self, *},
+    *,
+};
+
+#[derive(Debug, GraphQLObject, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Topic {
+    #[graphql(description = "Topic ID as integer")]
+    id: i32,
+    #[graphql(description = "Human readable topic name, in english")]
+    name: String,
+}
+
+#[derive(Debug, Clone, Default, GraphQLInputObject, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TopicFilter {
+    pub ids: Option<Vec<i32>>,
+    pub name_like: Option<String>,
+}
+
+#[rocket::async_trait]
+impl Showable<models::Topic, TopicFilter> for Topic {
+    fn sort_field_to_order_by(field: &str) -> Option<models::TopicOrderBy> {
+        match field {
+            "id" => Some(TopicOrderBy::Id),
+            _ => None,
+        }
+    }
+
+    fn filter_to_select(_context: &Context, filter: Option<TopicFilter>) -> FieldResult<models::SelectTopic> {
+        if let Some(f) = filter {
+            Ok(models::SelectTopic {
+                id_in: f.ids,
+                name_like: into_like_search(f.name_like),
+                ..Default::default()
+            })
+        } else {
+            Ok(Default::default())
+        }
+    }
+
+    fn select_by_id(_context: &Context, id: i32) -> FieldResult<models::SelectTopic> {
+        Ok(models::SelectTopic {
+            id_eq: Some(id),
+            ..Default::default()
+        })
+    }
+
+    async fn db_to_graphql(_context: &Context, d: models::Topic) -> AsamiResult<Self> {
+        Ok(Topic {
+            id: d.attrs.id,
+            name: d.attrs.name,
+        })
+    }
+}

api/src/error.rs

@@ -48,11 +48,15 @@ impl From<validators::errors::EmailError> for Error {
     }
 }
 
+impl From<time::error::ComponentRange> for Error {
+    fn from(err: time::error::ComponentRange) -> Error {
+        Error::precondition(&err.to_string())
+    }
+}
+
 impl<A: ethers::middleware::Middleware> From<ethers::contract::ContractError<A>> for Error {
     fn from(err: ethers::contract::ContractError<A>) -> Error {
-        let desc = err
-            .decode_revert::<String>()
-            .unwrap_or_else(|| format!("{:?}.{:?}", err, err.source()));
+        let desc = err.decode_revert::<String>().unwrap_or_else(|| format!("{:?}.{:?}", err, err.source()));
         Error::Service("rsk_contract".into(), desc)
     }
 }
@@ -131,11 +135,7 @@ impl<S: ScalarValue> IntoFieldError<S> for Error {
                 graphql_value!({ "error": { "field": "third_party_service", "message": service.as_str() } }),
             ),
             _ => {
-                warn!(
-                    "A wild error appeared: {:?}\n\n{:?}\n",
-                    &self,
-                    &self.source()
-                );
+                warn!("A wild error appeared: {:?}\n\n{:?}\n", &self, &self.source());
                 FieldError::new("unexpected error", graphql_value!(None))
             }
         }
@@ -149,15 +149,9 @@ impl<'r> Responder<'r, 'static> for Error {
                 Status::UnprocessableEntity,
                 Json(json![{"error": { "field": field, "message": message}}]),
             ),
-            Error::DatabaseError(sqlx::Error::RowNotFound) => {
-                (Status::NotFound, Json(json![{ "error": "Not found" }]))
-            }
+            Error::DatabaseError(sqlx::Error::RowNotFound) => (Status::NotFound, Json(json![{ "error": "Not found" }])),
             _ => {
-                warn!(
-                    "A wild error appeared: {:?}\n\n{:?}\n",
-                    &self,
-                    &self.source()
-                );
+                warn!("A wild error appeared: {:?}\n\n{:?}\n", &self, &self.source());
                 (
                     Status::InternalServerError,
                     Json(json![{ "error": "Unexpected Error" }]),

api/src/lib.rs

@@ -1,5 +1,5 @@
 pub use anyhow::Context;
-use rocket::{self, fairing::AdHoc, figment, http::Method, routes, serde::json::Json, State};
+use rocket::{self, fairing::AdHoc, figment, http::Method, routes, State};
 use rocket_cors::{AllowedHeaders, AllowedOrigins, Origins};
 use rocket_recaptcha_v3::ReCaptcha;
 pub use rust_decimal::prelude::{Decimal, FromPrimitive, ToPrimitive};

api/src/models/campaign.rs

@@ -45,6 +45,16 @@ model! {
     total_spent: String,
     #[sqlx_model_hints(varchar, default)]
     total_budget: String,
+
+    // These fields have X metrics for the campaign
+    #[sqlx_model_hints(int4, default)]
+    impression_count: i32,
+    #[sqlx_model_hints(int4, default)]
+    reply_count: i32,
+    #[sqlx_model_hints(int4, default)]
+    repost_count: i32,
+    #[sqlx_model_hints(int4, default)]
+    like_count: i32,
   },
   queries {
       needing_report("valid_until IS NOT NULL AND valid_until < now() AND report_hash IS NULL"),
@@ -222,6 +232,20 @@ impl CampaignHub {
                 .map_err(|_| Error::Validation("content_id".into(), "was stored in the db not as u64".into()))?;
 
             self.state.info("sync_x_collabs", "fetching_retweets", &post_id).await;
+            if let Some(metrics) = api.get_tweet(post_id).send().await?.data.as_ref().and_then(|o| o.organic_metrics.as_ref()) {
+                campaign
+                    .clone()
+                    .update()
+                    .impression_count(metrics.impression_count as i32)
+                    .reply_count(metrics.reply_count as i32)
+                    .repost_count(metrics.retweet_count as i32)
+                    .like_count(metrics.like_count as i32)
+                    .save()
+                    .await?;
+           } else {
+                self.state.info("sync_x_collabs", "no_organic_metrics_for", &post_id).await;
+           }
+
             let reposts = api.get_tweet_retweeted_by(post_id).send().await?;
             self.state.info("sync_x_collabs", "got_reposts", ()).await;