feat: implement RecordingDataManager::between and between_in_vc, start using it in /render

src/command/render.rs

@@ -1,3 +1,4 @@
+use futures::TryStreamExt as _;
 use snafu::{OptionExt as _, Report, ResultExt as _, Snafu};
 use std::{collections::BTreeMap, sync::LazyLock};
 use time::{UtcDateTime, format_description::well_known::Rfc3339};
@@ -85,7 +86,7 @@ pub static COMMAND: LazyLock<Command> = LazyLock::new(|| {
 });
 
 struct Options {
-    voice_channel: Id<ChannelMarker>,
+    voice_channel_id: Id<ChannelMarker>,
     start: UtcDateTime,
     end: UtcDateTime,
 }
@@ -152,15 +153,15 @@ fn parse_options(interaction: &Interaction) -> Result<Options, ParseOptionsError
         })
         .collect();
 
-    let voice_channel = options
+    let voice_channel_id = options
         .remove(OPTION_VOICE_CHANNEL)
         .context(NoVoiceChannelSnafu)?;
     let start = options.remove(OPTION_START).context(NoStartSnafu)?;
     let end = options.remove(OPTION_END).context(NoEndSnafu)?;
 
-    let &CommandOptionValue::Channel(voice_channel) = voice_channel else {
+    let &CommandOptionValue::Channel(voice_channel_id) = voice_channel_id else {
         return Err(ParseOptionsError::VoiceChannelInvalidType {
-            actual: voice_channel.clone(),
+            actual: voice_channel_id.clone(),
         });
     };
     let &CommandOptionValue::String(ref start) = start else {
@@ -178,7 +179,7 @@ fn parse_options(interaction: &Interaction) -> Result<Options, ParseOptionsError
     let end = UtcDateTime::parse(end, &DATE_FORMAT).context(EndInvalidDateSnafu)?;
 
     Ok(Options {
-        voice_channel,
+        voice_channel_id,
         start,
         end,
     })
@@ -186,6 +187,36 @@ fn parse_options(interaction: &Interaction) -> Result<Options, ParseOptionsError
 
 #[tracing::instrument]
 pub async fn handle(state: State, interaction: Interaction) {
+    let Some(guild_id) = interaction.guild_id else {
+        state
+            .discord_client
+            .interaction(state.discord_application_id)
+            .create_response(
+                interaction.id,
+                &interaction.token,
+                &InteractionResponse {
+                    kind: InteractionResponseType::ChannelMessageWithSource,
+                    data: Some(
+                        InteractionResponseDataBuilder::new()
+                            .embeds([EmbedBuilder::new()
+                                .title("Not in a server")
+                                .description(
+                                    "This command can only work when used in a Discord server.",
+                                )
+                                .validate()
+                                .unwrap()
+                                .build()])
+                            .flags(MessageFlags::EPHEMERAL)
+                            .build(),
+                    ),
+                },
+            )
+            .await
+            .expect("TODO");
+
+        return;
+    };
+
     let bot_owner_user_id = state.discord_bot_owner_user_id;
 
     let is_bot_owner = interaction
@@ -224,7 +255,7 @@ pub async fn handle(state: State, interaction: Interaction) {
     }
 
     let Options {
-        voice_channel,
+        voice_channel_id,
         start,
         end,
     } = match parse_options(&interaction) {
@@ -253,9 +284,22 @@ pub async fn handle(state: State, interaction: Interaction) {
         }
     };
 
-    tracing::info!(?voice_channel, ?start, ?end);
-
     let duration = end - start;
+    tracing::info!(?voice_channel_id, ?start, ?end, ?duration);
 
+    let mut recordings =
+        state
+            .recording_data_manager
+            .between_in_vc(start, end, guild_id, voice_channel_id);
+
+    while let Some(recording) = recordings.try_next().await.expect("TODO") {
+        tracing::debug!(?recording);
+        let recording_data = state
+            .recording_data_manager
+            .read(&recording)
+            .await
+            .expect("TODO");
+        tracing::debug!(?recording, ?recording_data);
+    }
     todo!();
 }

src/recording_data/between.rs

@@ -1,44 +1,159 @@
-use futures::{TryStream, TryStreamExt as _};
-use snafu::{ResultExt as _, Snafu};
-use std::{fmt::Display, str::FromStr};
-use time::UtcDateTime;
+use futures::{SinkExt, StreamExt as _, TryStream, TryStreamExt as _};
+use snafu::Snafu;
+use time::{Month, UtcDateTime};
 use twilight_model::id::{
     Id,
     marker::{ChannelMarker, GuildMarker},
 };
 
-use super::{ListError, Recording, RecordingDataManager, recording};
+use super::{ClipEntryError, ListError, Recording, RecordingDataManager};
+
+const BUFFER_SIZE: usize = 2048;
 
 #[derive(Debug, Snafu)]
 pub enum RecordingEntryError {
-    /// failed to get an entry from the storage operator's lister
-    ReceiveEntryError { source: opendal::Error },
+    /// could not list (at least some) clips
+    ListClipsError { source: ListError },
 
-    /// failed to parse the entry as a recording
-    ParseError { source: recording::TakeError },
+    /// could not receive this clip entry
+    ClipEntryError { source: ClipEntryError },
 }
 
-// impl RecordingDataManager {
-//     pub async fn between(
-//         &self,
-//         start: UtcDateTime,
-//         end: UtcDateTime,
-//     ) -> Result<impl TryStream<Ok = Recording, Error = RecordingEntryError> + Unpin, ListError>
-//     {
-//         todo!();
-//     }
-// }
+impl RecordingDataManager {
+    pub fn between(
+        &self,
+        start: UtcDateTime,
+        end: UtcDateTime,
+    ) -> impl TryStream<Ok = Recording, Error = RecordingEntryError> + Unpin {
+        let this = self.clone();
 
-// impl RecordingDataManager {
-//     pub async fn between_in_vc(
-//         &self,
-//         start: UtcDateTime,
-//         end: UtcDateTime,
-//         guild_id: Id<GuildMarker>,
-//         voice_channel_id: Id<ChannelMarker>,
-//     ) -> Result<impl TryStream<Ok = Recording, Error = RecordingEntryError> + Unpin, ListError>
-//     {
-//         todo!();
-//         Ok(self.between(start, end)?)
-//     }
-// }
+        let (mut sink, stream) = futures::channel::mpsc::channel(BUFFER_SIZE);
+
+        tokio::spawn(async move {
+            let year_start = start.year();
+            let year_end = end.year();
+
+            let years = year_start..=year_end;
+
+            for year in years {
+                let mut month_start = start.month();
+                let mut month_end = end.month();
+
+                if year > year_start {
+                    month_start = Month::January;
+                }
+
+                if year < year_end {
+                    month_end = Month::December;
+                }
+
+                let months = month_start as u8..=month_end as u8;
+                let months = months.map(|month| Month::try_from(month).unwrap());
+
+                for month in months {
+                    let mut day_start = start.day();
+                    let mut day_end = end.day();
+
+                    if month > month_start {
+                        day_start = 1;
+                    }
+
+                    if month < month_end {
+                        day_end = 31;
+                    }
+
+                    let days = day_start..=day_end;
+
+                    for day in days {
+                        let mut hour_start = start.hour();
+                        let mut hour_end = end.hour();
+
+                        if day > day_start {
+                            hour_start = 0;
+                        }
+
+                        if day < day_end {
+                            hour_end = 23;
+                        }
+
+                        let hours = hour_start..=hour_end;
+
+                        for hour in hours {
+                            let mut minute_start = start.minute();
+                            let mut minute_end = end.minute();
+
+                            if hour > hour_start {
+                                minute_start = 0;
+                            }
+
+                            if hour < hour_end {
+                                minute_end = 59;
+                            }
+
+                            let minutes = minute_start..=minute_end;
+
+                            for minute in minutes {
+                                match this.clips(year, month, day, hour, minute).await {
+                                    Err(list_error) => {
+                                        let _ = sink
+                                            .send(Err(RecordingEntryError::ListClipsError {
+                                                source: list_error,
+                                            }))
+                                            .await;
+                                    }
+                                    Ok(clips) => {
+                                        let mut clips = clips.into_stream();
+                                        while let Some(clip_result) = clips.next().await {
+                                            match clip_result {
+                                                Err(entry_error) => {
+                                                    let _ = sink
+                                                        .send(Err(
+                                                            RecordingEntryError::ClipEntryError {
+                                                                source: entry_error,
+                                                            },
+                                                        ))
+                                                        .await;
+                                                }
+                                                Ok(clip) => {
+                                                    let recording = Recording {
+                                                        year,
+                                                        month,
+                                                        day,
+                                                        hour,
+                                                        minute,
+                                                        clip,
+                                                    };
+
+                                                    let _ = sink.send(Ok(recording)).await;
+                                                }
+                                            }
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        });
+
+        stream
+    }
+}
+
+impl RecordingDataManager {
+    pub fn between_in_vc(
+        &self,
+        start: UtcDateTime,
+        end: UtcDateTime,
+        guild_id: Id<GuildMarker>,
+        voice_channel_id: Id<ChannelMarker>,
+    ) -> impl TryStream<Ok = Recording, Error = RecordingEntryError> + Unpin {
+        self.between(start, end).try_filter(move |recording| {
+            std::future::ready(
+                recording.clip.guild == guild_id
+                    && recording.clip.voice_channel == voice_channel_id,
+            )
+        })
+    }
+}

src/recording_data/mod.rs

@@ -18,7 +18,8 @@ mod user;
 mod voice_channel;
 mod year;
 
-pub use clip::Clip;
+pub use between::RecordingEntryError;
+pub use clip::{Clip, ClipEntryError};
 use day::Day;
 use guild::Guild;
 use hour::Hour;