From 1fc6d09910f1c8c764240f49f83ec0c7c8306452 Mon Sep 17 00:00:00 2001
From: Natsu Kagami <natsukagami@gmail.com>
Date: Mon, 7 Sep 2020 22:38:09 -0400
Subject: [PATCH] Codeforces: asyncify announcer

---
 youmubot-cf/src/announcer.rs | 137 +++++++++++++++++++----------------
 1 file changed, 74 insertions(+), 63 deletions(-)

diff --git a/youmubot-cf/src/announcer.rs b/youmubot-cf/src/announcer.rs
index 2cab8eb..cea3290 100644
--- a/youmubot-cf/src/announcer.rs
+++ b/youmubot-cf/src/announcer.rs
@@ -1,85 +1,62 @@
-use crate::db::{CfSavedUsers, CfUser};
+use crate::{
+    db::{CfSavedUsers, CfUser},
+    CFClient,
+};
 use announcer::MemberToChannels;
 use chrono::Utc;
 use codeforces::{RatingChange, User};
-use serenity::{
-    framework::standard::{CommandError, CommandResult},
-    http::CacheHttp,
-    model::id::{ChannelId, UserId},
-    CacheAndHttp,
-};
+use serenity::{http::CacheHttp, model::id::UserId, CacheAndHttp};
 use std::sync::Arc;
 use youmubot_prelude::*;
 
 type Reqwest = <HTTPClient as TypeMapKey>::Value;
 
 /// Updates the rating and rating changes of the users.
-pub fn updates(
-    http: Arc<CacheAndHttp>,
-    data: AppData,
-    channels: MemberToChannels,
-) -> CommandResult {
-    let mut users = CfSavedUsers::open(&*data.read()).borrow()?.clone();
-    let reqwest = data.get_cloned::<HTTPClient>();
+pub struct Announcer;
 
-    for (user_id, cfu) in users.iter_mut() {
-        if let Err(e) = update_user(http.clone(), &channels, &reqwest, *user_id, cfu) {
-            dbg!((*user_id, e));
-        }
+#[async_trait]
+impl youmubot_prelude::Announcer for Announcer {
+    async fn updates(
+        &mut self,
+        http: Arc<CacheAndHttp>,
+        data: AppData,
+        channels: MemberToChannels,
+    ) -> Result<()> {
+        let data = data.read().await;
+        let client = data.get::<CFClient>().unwrap();
+        let mut users = CfSavedUsers::open(&*data).borrow()?.clone();
+
+        users
+            .iter_mut()
+            .map(|(user_id, cfu)| update_user(http.clone(), &channels, &client, *user_id, cfu))
+            .collect::<stream::FuturesUnordered<_>>()
+            .try_collect::<()>()
+            .await?;
+        *CfSavedUsers::open(&*data).borrow_mut()? = users;
+        Ok(())
     }
-
-    *CfSavedUsers::open(&*data.read()).borrow_mut()? = users;
-    Ok(())
 }
 
-fn update_user(
+async fn update_user(
     http: Arc<CacheAndHttp>,
     channels: &MemberToChannels,
-    reqwest: &Reqwest,
+    client: &codeforces::Client,
     user_id: UserId,
     cfu: &mut CfUser,
-) -> CommandResult {
-    // Ensure this takes 200ms
-    let after = crossbeam_channel::after(std::time::Duration::from_secs_f32(0.2));
-    let info = User::info(reqwest, &[cfu.handle.as_str()])?
+) -> Result<()> {
+    let info = User::info(client, &[cfu.handle.as_str()])
+        .await?
         .into_iter()
         .next()
-        .ok_or(CommandError::from("Not found"))?;
+        .ok_or(Error::msg("Not found"))?;
 
-    let rating_changes = info.rating_changes(reqwest)?;
+    let rating_changes = info.rating_changes(client).await?;
 
-    let mut channels_list: Option<Vec<ChannelId>> = None;
+    let channels_list = channels.channels_of(&http, user_id).await;
     cfu.last_update = Utc::now();
     // Update the rating
     cfu.rating = info.rating;
 
-    let mut send_message = |rc: RatingChange| -> CommandResult {
-        let channels =
-            channels_list.get_or_insert_with(|| channels.channels_of(http.clone(), user_id));
-        if channels.is_empty() {
-            return Ok(());
-        }
-        let (contest, _, _) =
-            codeforces::Contest::standings(reqwest, rc.contest_id, |f| f.limit(1, 1))?;
-        for channel in channels {
-            if let Err(e) = channel.send_message(http.http(), |e| {
-                e.content(format!("Rating change for {}!", user_id.mention()))
-                    .embed(|c| {
-                        crate::embed::rating_change_embed(
-                            &rc,
-                            &info,
-                            &contest,
-                            &user_id.mention(),
-                            c,
-                        )
-                    })
-            }) {
-                dbg!(e);
-            }
-        }
-        Ok(())
-    };
-
     let rating_changes = match cfu.last_contest_id {
         None => rating_changes,
         Some(v) => {
@@ -101,12 +78,46 @@ fn update_user(
         .or(cfu.last_contest_id);
 
     // Check for any good announcements to make
-    for rc in rating_changes {
-        if let Err(v) = send_message(rc) {
-            dbg!(v);
-        }
-    }
-    after.recv().ok();
+    rating_changes
+        .into_iter()
+        .map(|rc: RatingChange| {
+            let channels = channels_list.clone();
+            let http = http.clone();
+            let info = info.clone();
+            async move {
+                if channels.is_empty() {
+                    return Ok(());
+                }
+                let (contest, _, _) =
+                    codeforces::Contest::standings(client, rc.contest_id, |f| f.limit(1, 1))
+                        .await?;
+                channels
+                    .iter()
+                    .map(|channel| {
+                        channel.send_message(http.http(), |e| {
+                            e.content(format!("Rating change for {}!", user_id.mention()))
+                                .embed(|c| {
+                                    crate::embed::rating_change_embed(
+                                        &rc,
+                                        &info,
+                                        &contest,
+                                        &user_id.mention(),
+                                        c,
+                                    )
+                                })
+                        })
+                    })
+                    .collect::<stream::FuturesUnordered<_>>()
+                    .map(|v| v.map(|_| ()))
+                    .try_collect::<()>()
+                    .await?;
+                let r: Result<_> = Ok(());
+                r
+            }
+        })
+        .collect::<stream::FuturesUnordered<_>>()
+        .try_collect::<()>()
+        .await?;
 
     Ok(())
 }