team archive more

archiver/src/archive.rs

@@ -6,7 +6,8 @@ use std::str::FromStr;
 use thiserror::Error;
 use tokio_stream::Stream;
 use ugc_scraper_types::{
-    GameMode, MatchInfo, Membership, MembershipRole, NameChange, Record, Region, Team,
+    GameMode, MatchInfo, Membership, MembershipRole, NameChange, Record, Region, RosterHistory,
+    Team,
 };
 
 #[derive(Debug, Error)]
@@ -167,6 +168,38 @@ impl Archive {
         Ok((row.min.unwrap_or_default() as u32)..(row.max.unwrap_or_default() as u32))
     }
 
+    pub fn get_team_ids(
+        &self,
+        min: u32,
+    ) -> impl Stream<Item = Result<u32, ArchiveError>> + use<'_> {
+        query!(
+            "select id from teams where id > $1 order by id asc",
+            min as i32
+        )
+        .fetch(&self.pool)
+        .map_err(|error| ArchiveError::Query {
+            description: "getting team ids",
+            error,
+        })
+        .map_ok(|map| map.id as u32)
+    }
+
+    pub async fn get_max_roster_history(&self) -> Result<u32, ArchiveError> {
+        if let Some(row) =
+            query!("select team_id as max from membership_history order by team_id desc limit 1;")
+                .fetch_optional(&self.pool)
+                .await
+                .map_err(|error| ArchiveError::Query {
+                    description: "getting latest team membership history",
+                    error,
+                })?
+        {
+            Ok(row.max as u32)
+        } else {
+            Ok(0)
+        }
+    }
+
     pub fn get_no_region_teams(&self) -> impl Stream<Item = Result<u32, ArchiveError>> + use<'_> {
         query!("select id from teams where region IS NULL and format != 'eights' order by id desc")
             .fetch(&self.pool)
@@ -268,4 +301,48 @@ impl Archive {
         })?;
         Ok(())
     }
+
+    pub async fn store_membership_history(
+        &self,
+        team_id: u32,
+        memberships: &[RosterHistory],
+    ) -> Result<(), ArchiveError> {
+        let mut transaction = self
+            .pool
+            .begin()
+            .await
+            .map_err(|error| ArchiveError::Query {
+                description: "beginning membership history transaction",
+                error,
+            })?;
+
+        for membership in memberships {
+            query!(
+                r#"INSERT INTO membership_history (
+                team_id, steam_id, role, joined, "left"
+              ) VALUES ($1, $2, $3, $4, $5)"#,
+                team_id as i32,
+                u64::from(membership.steam_id) as i64,
+                membership.role as MembershipRole,
+                membership.joined,
+                membership.left,
+            )
+            .execute(&mut *transaction)
+            .await
+            .map_err(|error| ArchiveError::Query {
+                description: "inserting membership history",
+                error,
+            })?;
+        }
+
+        transaction
+            .commit()
+            .await
+            .map_err(|error| ArchiveError::Query {
+                description: "commiting membership history transaction",
+                error,
+            })?;
+
+        Ok(())
+    }
 }

archiver/src/client.rs

@@ -3,7 +3,7 @@ use serde::de::DeserializeOwned;
 use thiserror::Error;
 use ugc_scraper_types::{
     GameMode, MapHistory, MatchInfo, MembershipHistory, Player, RosterHistory, Team,
-    TeamSeasonMatch, Transaction,
+    TeamRosterData, TeamSeasonMatch, Transaction,
 };
 
 #[derive(Debug, Error)]
@@ -56,7 +56,9 @@ impl UgcClient {
     }
 
     pub async fn get_team_roster(&self, id: u32) -> Result<Vec<RosterHistory>, UgcClientError> {
-        self.send_request(Endpoint::TeamRoster { id }).await
+        self.send_request::<TeamRosterData>(Endpoint::TeamRoster { id })
+            .await
+            .map(|data| data.history)
     }
 
     pub async fn get_team_matches(&self, id: u32) -> Result<Vec<TeamSeasonMatch>, UgcClientError> {

archiver/src/main.rs

@@ -27,6 +27,7 @@ enum Command {
     Matches,
     Teams,
     FixupTeams,
+    MembershipHistory,
 }
 
 const LAST_MATCH: u32 = 117047;
@@ -50,6 +51,9 @@ async fn main() -> MainResult {
         Command::FixupTeams => {
             fixup_teams(&client, &archive).await?;
         }
+        Command::MembershipHistory => {
+            archive_team_roster_history(&client, &archive).await?;
+        }
     }
     Ok(())
 }
@@ -107,6 +111,30 @@ async fn archive_teams(client: &UgcClient, archive: &Archive) -> MainResult {
     Ok(())
 }
 
+async fn archive_team_roster_history(client: &UgcClient, archive: &Archive) -> MainResult {
+    let last = archive.get_max_roster_history().await?;
+    let mut ids = pin!(archive.get_team_ids(last));
+
+    while let Some(Ok(id)) = ids.next().await {
+        let _span = span!(Level::INFO, "archive_team_roster_history", id = id).entered();
+        match client.get_team_roster(id).await.check_not_found() {
+            Ok(Some(team_data)) => {
+                info!(count = team_data.len(), "storing team roster history");
+                archive.store_membership_history(id, &team_data).await?;
+            }
+            Ok(None) => {
+                warn!("team roster history not found");
+            }
+            Err(e) => {
+                error!("error fetching team roster history: {:?}", e);
+                panic!();
+            }
+        }
+        sleep(Duration::from_millis(500)).await;
+    }
+    Ok(())
+}
+
 async fn fixup_teams(client: &UgcClient, archive: &Archive) -> MainResult {
     let mut ids = pin!(archive.get_no_region_teams());