chore: update lockfile

.dockerignore

@@ -0,0 +1,2 @@
+/target
+.history/

Cargo.toml

@@ -4,8 +4,17 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
+async-compression = { version = "0.4.41", features = ["brotli", "futures-io"] }
+async-trait = "0.1.89"
+bytes = "1.11.1"
+capnp = "0.25.3"
 clap = { version = "4.5.40", features = ["derive", "env"] }
-opendal = { git = "https://github.com/apache/opendal", features = [
+dashmap = "6.1.0"
+extension-traits = "2.0.2"
+futures = "0.3.32"
+hound = "3.5.1"
+moka = { version = "0.12.15", features = ["future"] }
+opendal = { git = "https://github.com/apache/opendal", rev = "ecf840b04afd2be109830b9978ba89759adfee79", features = [
     "services-azfile",
     "services-aliyun-drive",
     "services-alluxio",
@@ -42,11 +51,13 @@ opendal = { git = "https://github.com/apache/opendal", features = [
     "services-sled",
     "services-webdav",
 ] }
+opus2 = "0.4.0"
+patricia_tree = "0.10.1"
 rhai = "1.23.6"
 rustls = "0.23"
 secrecy = { version = "0.10.3", features = ["serde"] }
 snafu = { version = "0.8.9", features = ["futures"] }
-songbird = { version = "0.5.0", default-features = false, features = [
+songbird = { version = "0.6.0", default-features = false, features = [
     "driver",
     "gateway",
     "receive",
@@ -54,11 +65,25 @@ songbird = { version = "0.5.0", default-features = false, features = [
     "twilight",
     "tws",
 ] }
-tokio = { version = "1.46.0", features = ["rt-multi-thread", "macros"] }
+strum = { version = "0.28.0", features = ["derive"] }
+time = "0.3.47"
+tokio = { version = "1.46.0", features = [
+    "rt-multi-thread",
+    "macros",
+    "signal",
+] }
+tokio-util = { version = "0.7.18", features = ["io"] }
+tokio-websockets-0-13 = { package = "tokio-websockets", version = "0.13", features = [
+    "rustls-webpki-roots",
+] }
+tokio-websockets-0-11 = { package = "tokio-websockets", version = "0.11", features = [
+    "rustls-webpki-roots",
+] }
 tracing = "0.1.41"
 tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
 twilight-gateway = { version = "0.17", default-features = false, features = [
     "rustls-webpki-roots",
+    "twilight-http",
 ] }
 twilight-http = { version = "0.17", default-features = false, features = [
     "rustls-webpki-roots",
@@ -68,6 +93,7 @@ twilight-http = { version = "0.17", default-features = false, features = [
 twilight-model = "0.17"
 twilight-util = { version = "0.17", features = ["builder"] }
 typed-builder = "0.23.2"
+yoke = "0.8.2"
 
-[patch.crates-io]
-audiopus_sys = { git = "https://github.com/amsam0/audiopus_sys", rev = "3955ddb2b7b87e772e1c7bb93b8726a864f8c8f9" }
+[build-dependencies]
+capnpc = "0.25.3"

Dockerfile

@@ -0,0 +1,39 @@
+ARG RUST_TAG=1.94.1-alpine3.23
+FROM rust:${RUST_TAG} AS builder
+
+ARG BUILD_BASE_VERSION=0.5-r3
+RUN --mount=type=cache,sharing=locked,target=/var/cache/apk \
+    apk add --update build-base=${BUILD_BASE_VERSION}
+
+ARG CAPNPROTO_DEV_VERSION=1.2.0-r0
+RUN --mount=type=cache,sharing=locked,target=/var/cache/apk \
+    apk add --update capnproto-dev=${CAPNPROTO_DEV_VERSION}
+
+ARG CMAKE_VERSION=4.1.3-r0
+RUN --mount=type=cache,sharing=locked,target=/var/cache/apk \
+    apk add --update cmake=${CMAKE_VERSION}
+
+ARG PROTOC_VERSION=31.1-r1
+RUN --mount=type=cache,sharing=locked,target=/var/cache/apk \
+    apk add --update protoc=${PROTOC_VERSION}
+
+RUN \
+    # This one would be nice if it worked:
+    # --mount=type=bind,source=.,target=/root/app \
+    # Use this one if there's a hard to track down "read-only filesystem" error:
+    --mount=type=bind,source=.,target=/root/app,rw \
+
+    # --mount=type=bind,source=./target,target=/root/app/target,rw \
+    --mount=type=cache,sharing=private,target=/root/app/target \
+    # --mount=type=cache,sharing=private,target=/root/app/target,uid=1000,gid=1000 \
+    --mount=type=cache,target=/usr/local/cargo/git/db \
+    --mount=type=cache,target=/usr/local/cargo/registry/ \
+    cd /root/app && \
+    cargo build --release && \
+    cp target/release/fomo-reducer /root/program
+
+FROM scratch AS runner
+
+COPY --from=builder /root/program /program
+
+ENTRYPOINT ["/program"]

bot.capnp

@@ -0,0 +1,5 @@
+@0x993a671d7aa374f3;
+
+struct Bot {
+    heatScript @0 :Text;
+}

build.rs

@@ -0,0 +1,7 @@
+fn main() {
+    capnpc::CompilerCommand::new()
+        .file("bot.capnp")
+        .file("user.capnp")
+        .run()
+        .expect("couldn't compile capnproto schemas");
+}

src/bot_data.rs

@@ -0,0 +1,175 @@
+use async_compression::futures::{bufread::BrotliDecoder, write::BrotliEncoder};
+use capnp::message::TypedBuilder;
+use futures::{AsyncReadExt, AsyncWriteExt};
+use opendal::Operator;
+use snafu::{ResultExt as _, Snafu};
+
+use crate::{OperatorExt, bot_capnp, option_ext::OptionExt as _};
+
+#[derive(Debug, Clone)]
+pub struct BotDataManager {
+    operator: Operator,
+}
+
+impl BotDataManager {
+    pub fn new(operator: Operator) -> Self {
+        Self { operator }
+    }
+}
+
+const PATH: &str = "data.bin.brotli";
+
+#[derive(Debug, Snafu)]
+#[snafu(module)]
+pub enum WithError {
+    /// couldn't read data for this bot from the storage operator
+    ReadError { source: opendal::Error },
+
+    /// couldn't decompress the bot data from storage
+    DecompressionError { source: std::io::Error },
+
+    /// couldn't deserialize the bot data
+    DeserializeError { source: capnp::Error },
+}
+
+impl BotDataManager {
+    pub async fn with<R>(
+        &self,
+        f: impl FnOnce(bot_capnp::bot::Reader<'_>) -> R,
+    ) -> Result<R, WithError> {
+        let compressed_buffer = self
+            .operator
+            .async_reader_if_exists(PATH)
+            .await
+            .context(with_error::ReadSnafu)?;
+
+        let decompressed_reader = compressed_buffer.map(BrotliDecoder::new);
+        let decompressed = decompressed_reader
+            .map_async(|mut reader| async move {
+                let mut vec = Vec::new();
+                reader.read_to_end(&mut vec).await?;
+                Ok(vec)
+            })
+            .await
+            .transpose()
+            .context(with_error::DecompressionSnafu)?;
+
+        let mut message = TypedBuilder::<bot_capnp::bot::Owned>::new_default();
+        let fallback = message.init_root();
+
+        let mut bot_data = fallback.into_reader();
+        let message_reader;
+
+        if let Some(mut bytes) = decompressed.as_deref() {
+            message_reader = capnp::serialize::read_message_from_flat_slice_no_alloc(
+                &mut bytes,
+                Default::default(),
+            )
+            .context(with_error::DeserializeSnafu)?;
+
+            bot_data = message_reader
+                .get_root()
+                .context(with_error::DeserializeSnafu)?;
+        }
+
+        Ok(f(bot_data))
+    }
+}
+
+#[derive(Debug, Snafu)]
+#[snafu(module)]
+pub enum UpdateError {
+    /// couldn't read data for this bot from the storage operator
+    ReadError { source: opendal::Error },
+
+    /// couldn't decompress the bot data from storage
+    DecompressionError { source: std::io::Error },
+
+    /// couldn't deserialize the bot data
+    DeserializeError { source: capnp::Error },
+
+    /// couldn't serialize the (modified) bot data
+    SerializeError { source: capnp::Error },
+
+    /// couldn't create a writer for this bot data in the storage operator
+    WriterError { source: opendal::Error },
+
+    /// couldn't write (modified) data for this bot to the storage operator
+    WriteError { source: std::io::Error },
+
+    /// couldn't finalize writing (modified) data for this bot to the storage operator
+    FinalizeError { source: std::io::Error },
+}
+
+impl BotDataManager {
+    pub async fn update<R>(
+        &self,
+        f: impl FnOnce(bot_capnp::bot::Builder<'_>) -> R,
+    ) -> Result<R, UpdateError> {
+        let compressed_buffer = self
+            .operator
+            .async_reader_if_exists(PATH)
+            .await
+            .context(update_error::ReadSnafu)?;
+
+        let decompressed_reader = compressed_buffer.map(BrotliDecoder::new);
+        let decompressed = decompressed_reader
+            .map_async(|mut reader| async move {
+                let mut vec = Vec::new();
+                reader.read_to_end(&mut vec).await?;
+                Ok(vec)
+            })
+            .await
+            .transpose()
+            .context(update_error::DecompressionSnafu)?;
+
+        let mut message = TypedBuilder::<bot_capnp::bot::Owned>::new_default();
+
+        let ret = if let Some(mut bytes) = decompressed.as_deref() {
+            let message_reader = capnp::serialize::read_message_from_flat_slice_no_alloc(
+                &mut bytes,
+                Default::default(),
+            )
+            .context(update_error::DeserializeSnafu)?;
+
+            let bot_data = message_reader
+                .get_root()
+                .context(update_error::DeserializeSnafu)?;
+
+            message
+                .set_root(bot_data)
+                .context(update_error::DeserializeSnafu)?;
+
+            f(
+                message.get_root().unwrap(), // this is logically impossible
+            )
+        } else {
+            f(message.init_root())
+        };
+
+        let mut buffer = Vec::new();
+        capnp::serialize::write_message(&mut buffer, message.borrow_inner())
+            .context(update_error::SerializeSnafu)?;
+
+        let compressed_writer = self
+            .operator
+            .writer(PATH)
+            .await
+            .context(update_error::WriterSnafu)?
+            .into_futures_async_write();
+
+        let mut decompressed_writer = BrotliEncoder::new(compressed_writer);
+
+        decompressed_writer
+            .write_all(&buffer)
+            .await
+            .context(update_error::WriteSnafu)?;
+
+        decompressed_writer
+            .close()
+            .await
+            .context(update_error::FinalizeSnafu)?;
+
+        Ok(ret)
+    }
+}

src/command/info.rs

@@ -0,0 +1,185 @@
+use futures::TryStreamExt;
+use snafu::{OptionExt, Snafu};
+use std::sync::LazyLock;
+use twilight_model::{
+    application::{
+        command::{Command, CommandType},
+        interaction::Interaction,
+    },
+    channel::message::{Embed, MessageFlags},
+    http::interaction::{InteractionResponse, InteractionResponseType},
+    id::{Id, marker::UserMarker},
+};
+use twilight_util::builder::{
+    InteractionResponseDataBuilder,
+    command::CommandBuilder,
+    embed::{EmbedAuthorBuilder, EmbedBuilder, EmbedFieldBuilder},
+};
+
+use crate::command::State;
+
+const NAME: &str = "info";
+const DESCRIPTION: &str = "Show various information";
+
+pub static COMMAND: LazyLock<Command> = LazyLock::new(|| {
+    CommandBuilder::new(NAME, DESCRIPTION, CommandType::ChatInput)
+        .validate()
+        .expect("command wasn't correct")
+        .build()
+});
+
+#[derive(Debug, Snafu)]
+enum NoPermission {
+    /// there isn't a user who invoked this command
+    NoUser,
+
+    /// the user isn't allowed to use this command because they're not the bot owner
+    NotInvokedByBotOwner,
+}
+
+fn no_permission_to_embed(error: NoPermission) -> Embed {
+    match error {
+        NoPermission::NoUser => {
+            EmbedBuilder::new().title("Not invoked by a user").description("This command works by joining the same VC as the user, but this bot didn't receive any user data. So did no user invoke it?! (This error should be impossible!)").validate().unwrap().build()
+        },
+        NoPermission::NotInvokedByBotOwner => {
+            EmbedBuilder::new().title("No permission to see debug info").description("Only the owner of this bot is allowed to see its information for debugging purposes.").validate().unwrap().build()
+        },
+    }
+}
+
+fn check_permission(
+    interaction: &Interaction,
+    bot_owner_user_id: Id<UserMarker>,
+) -> Result<(), NoPermission> {
+    let user_id = interaction
+        .member
+        .as_ref()
+        .and_then(|member| member.user.as_ref().map(|user| user.id))
+        .context(NoUserSnafu)?;
+
+    if user_id != bot_owner_user_id {
+        return Err(NoPermission::NotInvokedByBotOwner);
+    }
+
+    Ok(())
+}
+
+#[tracing::instrument]
+pub async fn handle(state: State, interaction: Interaction) {
+    if let Err(no_permission) = check_permission(&interaction, state.discord_bot_owner_user_id) {
+        state
+            .discord_client
+            .interaction(state.discord_application_id)
+            .create_response(
+                interaction.id,
+                &interaction.token,
+                &InteractionResponse {
+                    kind: InteractionResponseType::ChannelMessageWithSource,
+                    data: Some(
+                        InteractionResponseDataBuilder::new()
+                            .embeds([no_permission_to_embed(no_permission)])
+                            .flags(MessageFlags::EPHEMERAL)
+                            .build(),
+                    ),
+                },
+            )
+            .await
+            .expect("TODO");
+
+        return;
+    }
+
+    state
+        .discord_client
+        .interaction(state.discord_application_id)
+        .create_response(
+            interaction.id,
+            &interaction.token,
+            &InteractionResponse {
+                kind: InteractionResponseType::ChannelMessageWithSource,
+                data: Some(
+                    InteractionResponseDataBuilder::new()
+                        .flags(MessageFlags::EPHEMERAL)
+                        .content("some debug info is coming your way!")
+                        .build(),
+                ),
+            },
+        )
+        .await
+        .expect("TODO");
+
+    let heat_script_description = state
+        .bot_data_manager
+        .with(|bot_data| {
+            let heat_script_option = bot_data.has_heat_script().then(|| {
+                bot_data
+                    .get_heat_script()
+                    .expect("TODO")
+                    .to_string()
+                    .expect("TODO")
+            });
+            heat_script_option.map_or("none set yet".into(), |heat_script| {
+                format!("```\n{heat_script}\n```")
+            })
+        })
+        .await
+        .expect("TODO");
+
+    state
+        .discord_client
+        .interaction(state.discord_application_id)
+        .create_followup(&interaction.token)
+        .embeds(&[EmbedBuilder::new()
+            .field(EmbedFieldBuilder::new("Heat Script", heat_script_description).build())
+            .validate()
+            .unwrap()
+            .build()])
+        .flags(MessageFlags::EPHEMERAL)
+        .await
+        .expect("TODO");
+
+    let mut user_id_stream = state.user_data_manager.list().await.expect("TODO");
+
+    while let Some(user_id) = user_id_stream.try_next().await.expect("TODO") {
+        let (consent, notification_script) = state
+            .user_data_manager
+            .with(user_id, |user_data| {
+                let consent = user_data.get_voice_recording_consent().unwrap();
+                let notification_script = user_data.has_notification_script().then_some(
+                    user_data
+                        .get_notification_script()
+                        .expect("TODO")
+                        .to_string()
+                        .expect("TODO"),
+                );
+
+                (consent, notification_script)
+            })
+            .await
+            .expect("TODO");
+
+        let user_mention = format!("<@{user_id}>");
+
+        state
+            .discord_client
+            .interaction(state.discord_application_id)
+            .create_followup(&interaction.token)
+            .embeds(&[EmbedBuilder::new()
+                .author(EmbedAuthorBuilder::new(user_mention))
+                .field(EmbedFieldBuilder::new("Consent", format!("{consent:?}")).build())
+                .field(
+                    EmbedFieldBuilder::new(
+                        "Notification Script",
+                        format!("{notification_script:?}"),
+                    )
+                    .build(),
+                )
+                .validate()
+                .unwrap()
+                .build()])
+            .flags(MessageFlags::EPHEMERAL)
+            .await
+            .expect("TODO");
+    }
+}

src/command/join.rs

@@ -0,0 +1,362 @@
+use crate::{
+    OneToManyUniqueBTreeMap, UserDataManager, VCs, command::State, option_ext::OptionExt as _,
+    user_capnp::user::Consent, user_data::RECORD_IF_CONSENT_UNSPECIFIED,
+};
+use async_trait::async_trait;
+use futures::FutureExt;
+use hound::{SampleFormat, WavSpec};
+use opendal::Operator;
+use snafu::{OptionExt as _, Snafu};
+use songbird::{CoreEvent, Event, EventContext, EventHandler};
+use std::{
+    io::Cursor,
+    sync::{Arc, LazyLock, Mutex},
+    time::Instant,
+};
+use time::UtcDateTime;
+use twilight_model::{
+    application::{
+        command::{Command, CommandType},
+        interaction::Interaction,
+    },
+    channel::message::{Embed, MessageFlags},
+    http::interaction::{InteractionResponse, InteractionResponseType},
+    id::{
+        Id,
+        marker::{ChannelMarker, GuildMarker, UserMarker},
+    },
+};
+use twilight_util::builder::{
+    InteractionResponseDataBuilder,
+    command::CommandBuilder,
+    embed::{EmbedBuilder, EmbedFieldBuilder, EmbedFooterBuilder},
+};
+
+const NAME: &str = "join";
+const DESCRIPTION: &str = "The bot will join the same VC as you (with intention to record)";
+
+pub static COMMAND: LazyLock<Command> = LazyLock::new(|| {
+    CommandBuilder::new(NAME, DESCRIPTION, CommandType::ChatInput)
+        .validate()
+        .expect("command wasn't correct")
+        .build()
+});
+
+#[derive(Debug, Snafu)]
+enum GetGuildAndVoiceChannelIdError {
+    /// this command was not used inside a guild (Discord server)
+    NotInGuild,
+
+    /// there is no user who invoked this command
+    NoUser,
+
+    /// there are no voice chats in this guild
+    NoVCsInGuild,
+
+    /// the user is not in a voice chat in this guild
+    UserNotInVC,
+}
+
+#[tracing::instrument]
+fn get_guild_and_voice_channel_id(
+    interaction: &Interaction,
+    vcs: &VCs,
+) -> Result<(Id<GuildMarker>, Id<ChannelMarker>), GetGuildAndVoiceChannelIdError> {
+    let guild_id = interaction.guild_id.context(NotInGuildSnafu)?;
+
+    let user_id = interaction
+        .member
+        .as_ref()
+        .and_then(|member| member.user.as_ref().map(|user| user.id))
+        .context(NoUserSnafu)?;
+
+    let guild_vcs = vcs.get(&guild_id).context(NoVCsInGuildSnafu)?;
+
+    let &voice_channel_id = guild_vcs.get_left_for(&user_id).context(UserNotInVCSnafu)?;
+
+    Ok((guild_id, voice_channel_id))
+}
+
+fn get_guild_and_vc_error_to_embed(error: GetGuildAndVoiceChannelIdError) -> Embed {
+    match error {
+        GetGuildAndVoiceChannelIdError::NotInGuild => {
+            EmbedBuilder::new().title("Use this in a server").description("This bot can't find a VC to join if the command is used outside of a server (you might've used it in a DM?).").validate().unwrap().build()
+        },
+        GetGuildAndVoiceChannelIdError::NoUser => {
+            EmbedBuilder::new().title("Not invoked by a user").description("This command works by joining the same VC as the user, but this bot didn't receive any user data. So did no user invoke it?! (This error should be impossible!)").validate().unwrap().build()
+        },
+        GetGuildAndVoiceChannelIdError::NoVCsInGuild => {
+            EmbedBuilder::new().title("No VCs in this server").description("This bot can't find a VC to join because there aren't any in this server right now.").validate().unwrap().build()
+        },
+        GetGuildAndVoiceChannelIdError::UserNotInVC => {
+            EmbedBuilder::new().title("You're not in a VC").description("This bot can't follow you into VC if you aren't in one in this server.").validate().unwrap().build()
+        },
+    }
+}
+
+#[derive(Debug, Clone)]
+struct Handler {
+    start_instant: Instant,
+    start_utc: UtcDateTime,
+
+    recordings: Operator,
+
+    guild_id: Id<GuildMarker>,
+    channel_id: Id<ChannelMarker>,
+
+    known_ssrcs: Arc<Mutex<OneToManyUniqueBTreeMap<Id<UserMarker>, u32>>>,
+
+    audio_channels: u16,
+    audio_sample_rate: u32,
+
+    user_data_manager: UserDataManager,
+}
+
+#[async_trait]
+impl EventHandler for Handler {
+    async fn act(&self, ctx: &EventContext<'_>) -> Option<Event> {
+        match ctx {
+            EventContext::Track(_items) => {
+                // Not expected to fire
+            }
+            EventContext::SpeakingStateUpdate(speaking) => {
+                tracing::error!(?speaking);
+
+                if let Some(user_id) = speaking.user_id {
+                    let user_id = Id::new(user_id.0);
+
+                    self.known_ssrcs
+                        .lock()
+                        .unwrap()
+                        .insert(user_id, speaking.ssrc);
+                }
+            }
+            EventContext::VoiceTick(voice_tick) => {
+                tracing::error!(?voice_tick);
+
+                for (ssrc, voice_data) in &voice_tick.speaking {
+                    let user_id = self.known_ssrcs.lock().unwrap().get_left_for(ssrc).cloned();
+
+                    tracing::info!(?user_id);
+
+                    if let Some(pcm) = &voice_data.decoded_voice {
+                        let may_record = user_id
+                            .map_async(|user_id| {
+                                self.user_data_manager
+                                    .with(user_id, |user_data| {
+                                        user_data.get_voice_recording_consent().unwrap()
+                                    })
+                                    .map(|result| result.expect("TODO"))
+                            })
+                            .await
+                            .map_or(RECORD_IF_CONSENT_UNSPECIFIED, |consent| match consent {
+                                Consent::Unspecified => RECORD_IF_CONSENT_UNSPECIFIED,
+                                Consent::Granted => true,
+                                Consent::Withheld => false,
+                            });
+
+                        if !may_record {
+                            tracing::warn!(?user_id, "may not be recorded");
+                            continue;
+                        }
+
+                        let elapsed = self.start_instant.elapsed();
+                        let elapsed = elapsed.try_into().expect("TODO");
+
+                        let now_utc = self.start_utc.checked_add(elapsed).expect("TODO");
+                        tracing::error!(?now_utc, "TODO");
+
+                        let year = now_utc.year();
+                        let month = now_utc.month();
+                        let day = now_utc.day();
+
+                        let hour = now_utc.hour();
+                        let minute = now_utc.minute();
+                        let second = now_utc.second();
+
+                        let microseconds = now_utc.microsecond();
+
+                        let guild_id = self.guild_id;
+                        let channel_id = self.channel_id;
+
+                        let user = user_id
+                            .as_ref()
+                            .map_or_else(|| "UNKNOWN".into(), ToString::to_string);
+
+                        let path = format!(
+                            "{year}/{month}/{day}/{hour}/{minute}/audio-{second}.{microseconds}-{guild_id}-{channel_id}-{user}.wav"
+                        );
+
+                        let channels = self.audio_channels;
+                        let sample_rate = self.audio_sample_rate;
+
+                        let wav_spec = WavSpec {
+                            channels,
+                            sample_rate,
+                            bits_per_sample: 16,
+                            sample_format: SampleFormat::Int,
+                        };
+
+                        let mut buffer = Vec::new();
+                        let writer = Cursor::new(&mut buffer);
+
+                        let mut wav_writer = hound::WavWriter::new(writer, wav_spec).expect("TODO");
+
+                        let mut sample_writer = wav_writer.get_i16_writer(pcm.len() as u32);
+
+                        for sample in pcm {
+                            sample_writer.write_sample(*sample);
+                        }
+                        sample_writer.flush().expect("TODO");
+
+                        wav_writer.finalize().expect("TODO");
+
+                        tracing::info!("going to write the audio shortly");
+
+                        let recordings = self.recordings.clone();
+                        tokio::spawn(async move {
+                            recordings.write(&path, buffer).await.expect("TODO");
+                            tracing::info!("successfully wrote the audio!");
+                        });
+                    }
+                }
+            }
+            EventContext::RtpPacket(_rtp_data) => {}
+            EventContext::RtcpPacket(_rtcp_data) => {}
+            EventContext::ClientDisconnect(_client_disconnect) => {
+                // This is already taken care of elsewhere
+            }
+            EventContext::DriverConnect(_connect_data) => {}
+            EventContext::DriverReconnect(_connect_data) => {}
+            EventContext::DriverDisconnect(_disconnect_data) => {}
+            other => {
+                tracing::warn!(?other, "cannot be handled yet");
+            }
+        }
+
+        None
+    }
+}
+
+#[tracing::instrument(skip(state))]
+pub async fn handle(state: State, interaction: Interaction) {
+    let vcs = state.vcs;
+
+    let (guild_id, voice_channel_id) = match get_guild_and_voice_channel_id(&interaction, &vcs) {
+        Ok((guild_id, voice_channel_id)) => (guild_id, voice_channel_id),
+        Err(error) => {
+            state
+                .discord_client
+                .interaction(state.discord_application_id)
+                .create_response(
+                    interaction.id,
+                    &interaction.token,
+                    &InteractionResponse {
+                        kind: InteractionResponseType::ChannelMessageWithSource,
+                        data: Some(
+                            InteractionResponseDataBuilder::new()
+                                .embeds([get_guild_and_vc_error_to_embed(error)])
+                                .flags(MessageFlags::EPHEMERAL)
+                                .build(),
+                        ),
+                    },
+                )
+                .await
+                .expect("TODO");
+
+            return;
+        }
+    };
+
+    state
+        .discord_client
+        .interaction(state.discord_application_id)
+        .create_response(
+            interaction.id,
+            &interaction.token,
+            &InteractionResponse {
+                kind: InteractionResponseType::DeferredChannelMessageWithSource,
+                data: None,
+            },
+        )
+        .await
+        .expect("TODO");
+
+    let call = state
+        .songbird
+        .join(guild_id, voice_channel_id)
+        .await
+        .expect("TODO");
+
+    tracing::error!(?call, "successfully joined");
+
+    let start_instant = Instant::now();
+    let start_utc = UtcDateTime::now();
+
+    let audio_channels = opus2::Channels::from(state.audio_channels) as u16;
+
+    let audio_sample_rate = u32::from(state.audio_sample_rate);
+
+    let handler = Handler {
+        start_instant,
+        start_utc,
+        recordings: state.recording_data,
+        guild_id,
+        channel_id: voice_channel_id,
+        known_ssrcs: Default::default(),
+
+        audio_channels,
+        audio_sample_rate,
+
+        user_data_manager: state.user_data_manager,
+    };
+
+    {
+        let mut call = call.lock().await;
+
+        call.add_global_event(CoreEvent::SpeakingStateUpdate.into(), handler.clone());
+        call.add_global_event(CoreEvent::VoiceTick.into(), handler);
+
+        call.mute(true).await.expect("TODO");
+    }
+
+    let channel_mention = format!("<#{voice_channel_id}>");
+    let bot_owner_mention = format!("<@{}>", state.discord_bot_owner_user_id);
+
+    let opt_in_mention = format!(
+        "</{}:{}>",
+        state.discord_opt_in_command_name, state.discord_opt_in_command_id
+    );
+    let opt_out_mention = format!(
+        "</{}:{}>",
+        state.discord_opt_out_command_name, state.discord_opt_out_command_id
+    );
+
+    state
+        .discord_client
+        .interaction(state.discord_application_id)
+        .update_response(
+            &interaction.token,
+        ).embeds(Some(&[
+            EmbedBuilder::new()
+            .title("Joined VC to record")
+            .description(format!("This bot joined {channel_mention} and intends to record. You can opt out with {opt_out_mention} or explicitly opt in with {opt_in_mention} (I'd appreciate this one). Here are some pledges backed by faith (because there is no way to verify them yourself) in {bot_owner_mention}:"))
+            .field(
+                EmbedFieldBuilder::new("Recordings are never shared", "Audio recordings are only stored on my home server and desktop computer and will never be uploaded to services or hardware that is owned by another person: not even curated clips, and not even to people who were in the recording. When transcription to text is implemented, this will only be run on my personally owned devices and not on any internet or cloud offering.").build()
+            )
+            .field(
+                EmbedFieldBuilder::new("You won't be \"audited\"", "I will not reference things said in past recordings with the goal of \"making a point\", nor pull them up on the spot (even by the request of the person who said it). Ideally, these are just peace of mind for me that I'm not missing out by not being in a Discord call all the time and can take my life back, so using them in an unhealthy way isn't in my interest.").build()
+            )
+            .field(
+                EmbedFieldBuilder::new("Code is publicly available", "The latest source code is at https://gitea.katniss.top/jacob/fomo-reducer so that I don't have to write guarantees about the technology here (e.g. what data is acquired, how it's used or stored) and you can just check it yourself.").build()
+            )
+            .footer(
+                EmbedFooterBuilder::new("Thanks for your patience and understanding as I have bad and unusual mental health and it's crazy that I need this. This - especially if I learn if I can record streams or webcams so I don't miss out on those experiences either - should be the end of abrasion and force about how we spend our time. Again, thank you, I appreciate it.")
+            )
+            .validate()
+            .unwrap()
+            .build()
+        ]))
+        .await
+        .expect("TODO");
+}

src/command/leave.rs

@@ -0,0 +1,174 @@
+use crate::VCs;
+use crate::command::State;
+use snafu::{OptionExt, Snafu};
+use std::sync::LazyLock;
+use twilight_model::channel::message::{Embed, MessageFlags};
+use twilight_model::http::interaction::{InteractionResponse, InteractionResponseType};
+use twilight_model::id::marker::UserMarker;
+use twilight_model::{
+    application::{
+        command::{Command, CommandType},
+        interaction::Interaction,
+    },
+    id::{
+        Id,
+        marker::{ChannelMarker, GuildMarker},
+    },
+};
+use twilight_util::builder::InteractionResponseDataBuilder;
+use twilight_util::builder::command::CommandBuilder;
+use twilight_util::builder::embed::EmbedBuilder;
+
+const NAME: &str = "leave";
+const DESCRIPTION: &str = "The bot will leave the VC it's in (so it won't record anyone anymore)";
+
+pub static COMMAND: LazyLock<Command> = LazyLock::new(|| {
+    CommandBuilder::new(NAME, DESCRIPTION, CommandType::ChatInput)
+        .validate()
+        .expect("command wasn't correct")
+        .build()
+});
+
+#[derive(Debug, Snafu)]
+pub enum GetGuildAndVoiceChannelIdError {
+    /// this command was not used inside a guild (Discord server)
+    NotInGuild,
+
+    /// there is no user who invoked this command
+    NoUser,
+
+    /// there are no voice chats in this guild
+    NoVCsInGuild,
+
+    /// the bot is not in a voice chat in this guild
+    BotNotInVC,
+}
+
+#[tracing::instrument]
+pub fn get_user_and_guild_and_voice_channel_id(
+    bot_user_id: Id<UserMarker>,
+    interaction: &Interaction,
+    vcs: &VCs,
+) -> Result<(Id<UserMarker>, Id<GuildMarker>, Id<ChannelMarker>), GetGuildAndVoiceChannelIdError> {
+    let user_id = interaction
+        .member
+        .as_ref()
+        .and_then(|member| member.user.as_ref().map(|user| user.id))
+        .context(NoUserSnafu)?;
+
+    let guild_id = interaction.guild_id.context(NotInGuildSnafu)?;
+
+    let guild_vcs = vcs.get(&guild_id).context(NoVCsInGuildSnafu)?;
+
+    let &voice_channel_id = guild_vcs
+        .get_left_for(&bot_user_id)
+        .context(BotNotInVCSnafu)?;
+
+    Ok((user_id, guild_id, voice_channel_id))
+}
+
+fn get_guild_and_vc_error_to_embed(error: GetGuildAndVoiceChannelIdError) -> Embed {
+    match error {
+        GetGuildAndVoiceChannelIdError::NotInGuild => {
+            EmbedBuilder::new().title("Use this in a server").description("This bot can't tell which VC to leave if the command is used outside of a server (you might've used it in a DM?).").validate().unwrap().build()
+        },
+        GetGuildAndVoiceChannelIdError::NoUser => {
+            EmbedBuilder::new().title("Not invoked by a user").description("This command works by joining the same VC as the user, but this bot didn't receive any user data. So did no user invoke it?! (This error should be impossible!)").validate().unwrap().build()
+        },
+        GetGuildAndVoiceChannelIdError::NoVCsInGuild => {
+            EmbedBuilder::new().title("No VCs in this server").description("This bot can't leave VC because there aren't any in this server right now (therefore the bot must not be in any).").validate().unwrap().build()
+        },
+        GetGuildAndVoiceChannelIdError::BotNotInVC => {
+            EmbedBuilder::new().title("Not in a VC").description("This bot can't leave VC if it isn't in one in this server.").validate().unwrap().build()
+        },
+    }
+}
+
+#[tracing::instrument]
+pub async fn handle(state: State, interaction: Interaction) {
+    let (user_id, guild_id, voice_channel_id) = match get_user_and_guild_and_voice_channel_id(
+        state.discord_user_id,
+        &interaction,
+        &state.vcs,
+    ) {
+        Ok((user_id, guild_id, voice_channel_id)) => (user_id, guild_id, voice_channel_id),
+        Err(error) => {
+            state
+                .discord_client
+                .interaction(state.discord_application_id)
+                .create_response(
+                    interaction.id,
+                    &interaction.token,
+                    &InteractionResponse {
+                        kind: InteractionResponseType::ChannelMessageWithSource,
+                        data: Some(
+                            InteractionResponseDataBuilder::new()
+                                .embeds([get_guild_and_vc_error_to_embed(error)])
+                                .flags(MessageFlags::EPHEMERAL)
+                                .build(),
+                        ),
+                    },
+                )
+                .await
+                .expect("TODO");
+
+            return;
+        }
+    };
+
+    if user_id != state.discord_bot_owner_user_id {
+        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("No permission to make this bot leave").description("Only the owner of this bot is allowed to make the bot leave RIGHT NOW. You might be looking for the opt out command.").validate().unwrap().build()
+                            ])
+                            .flags(MessageFlags::EPHEMERAL)
+                            .build(),
+                    ),
+                },
+            )
+            .await
+            .expect("TODO");
+
+        return;
+    }
+
+    state.songbird.leave(guild_id).await.expect("TODO");
+
+    tracing::error!("TODO: successfully left the call");
+
+    let channel_mention = format!("<#{voice_channel_id}>");
+
+    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("Left VC")
+            .description(format!(
+                "This bot left {channel_mention} (and is thereby unable to record anymore)."
+            ))
+            .validate()
+            .unwrap()
+            .build()
+                            ])
+                            .flags(MessageFlags::EPHEMERAL)
+                            .build(),
+                    ),
+                },)
+        .await
+        .expect("TODO");
+}

src/command/mod.rs

@@ -0,0 +1,118 @@
+use std::{fmt::Debug, sync::Arc};
+
+use futures::future::BoxFuture;
+use opendal::Operator;
+use patricia_tree::StringPatriciaMap;
+use songbird::{
+    Songbird,
+    driver::{Channels, SampleRate},
+};
+use tokio_util::sync::CancellationToken;
+use twilight_model::{
+    application::{command::Command, interaction::Interaction},
+    id::{
+        Id,
+        marker::{ApplicationMarker, CommandMarker, UserMarker},
+    },
+};
+
+use crate::{BotDataManager, GuildVoiceChannelToTextChannel, UserDataManager, VCs};
+
+pub mod info;
+pub mod join;
+pub mod leave;
+pub mod opt_in;
+pub mod opt_out;
+
+#[derive(Debug, Clone)]
+pub struct State {
+    pub audio_channels: Channels,
+    pub audio_sample_rate: SampleRate,
+    pub bot_data_manager: BotDataManager,
+    pub cancellation_token: CancellationToken,
+    pub discord_application_id: Id<ApplicationMarker>,
+    pub discord_bot_owner_user_id: Id<UserMarker>,
+    pub discord_client: Arc<twilight_http::Client>,
+    pub discord_opt_in_command_id: Id<CommandMarker>,
+    pub discord_opt_in_command_name: Arc<str>,
+    pub discord_opt_out_command_id: Id<CommandMarker>,
+    pub discord_opt_out_command_name: Arc<str>,
+    pub discord_user_id: Id<UserMarker>,
+    pub discord_voice_channel_corresponding_text_channel: Arc<GuildVoiceChannelToTextChannel>,
+    pub recording_data: Operator,
+    pub songbird: Arc<Songbird>,
+    pub user_data_manager: UserDataManager,
+    pub vcs: Arc<VCs>,
+}
+
+type Return = ();
+type BoxedHandler = Box<dyn Send + Sync + Fn(State, Interaction) -> BoxFuture<'static, Return>>;
+
+fn box_handler<Handler, Fut>(handler: Handler) -> BoxedHandler
+where
+    Fut: Future<Output = Return> + Send + 'static,
+    Handler: Send + Sync + Fn(State, Interaction) -> Fut + 'static,
+{
+    Box::new(move |state, interaction| Box::pin(handler(state, interaction)))
+}
+
+pub fn all() -> Vec<(&'static Command, BoxedHandler)> {
+    vec![
+        (&info::COMMAND, box_handler(info::handle)),
+        (&join::COMMAND, box_handler(join::handle)),
+        (&leave::COMMAND, box_handler(leave::handle)),
+        (&opt_in::COMMAND, box_handler(opt_in::handle)),
+        (&opt_out::COMMAND, box_handler(opt_out::handle)),
+    ]
+}
+
+#[derive(Default)]
+pub struct Router {
+    map: StringPatriciaMap<BoxedHandler>,
+}
+
+impl Router {
+    pub fn add_route<Fut, Handler>(&mut self, name: &str, handler: Handler)
+    where
+        Fut: Future<Output = Return> + Send + 'static,
+        Handler: Send + Sync + Fn(State, Interaction) -> Fut + 'static,
+    {
+        self.add_route_already_boxed(name, box_handler(handler));
+    }
+
+    fn add_route_already_boxed(&mut self, name: &str, boxed_handler: BoxedHandler) {
+        self.map.insert(name, boxed_handler);
+    }
+
+    pub async fn handle(
+        &self,
+        state: State,
+        command_name: &str,
+        interaction: Interaction,
+    ) -> Option<Return> {
+        let handler = self.map.get(command_name)?;
+
+        Some(handler(state, interaction).await)
+    }
+}
+
+impl<'a> FromIterator<(&'a Command, BoxedHandler)> for Router {
+    #[inline]
+    fn from_iter<T: IntoIterator<Item = (&'a Command, BoxedHandler)>>(iter: T) -> Self {
+        let mut this = Self::default();
+
+        this.extend(iter);
+
+        this
+    }
+}
+
+impl<'a> Extend<(&'a Command, BoxedHandler)> for Router {
+    #[inline]
+    fn extend<T: IntoIterator<Item = (&'a Command, BoxedHandler)>>(&mut self, iter: T) {
+        for (command, boxed_handler) in iter {
+            let name = &command.name;
+            self.add_route_already_boxed(name, boxed_handler);
+        }
+    }
+}

src/command/opt_in.rs

@@ -0,0 +1,90 @@
+use std::sync::LazyLock;
+
+use twilight_model::{
+    application::{
+        command::{Command, CommandType},
+        interaction::Interaction,
+    },
+    channel::message::MessageFlags,
+    http::interaction::{InteractionResponse, InteractionResponseType},
+};
+use twilight_util::builder::{InteractionResponseDataBuilder, command::CommandBuilder};
+
+use crate::{command::State, user_capnp::user::Consent};
+
+const NAME: &str = "opt-in";
+const DESCRIPTION: &str = "Opt in to being recorded";
+
+pub static COMMAND: LazyLock<Command> = LazyLock::new(|| {
+    CommandBuilder::new(NAME, DESCRIPTION, CommandType::ChatInput)
+        .validate()
+        .expect("command wasn't correct")
+        .build()
+});
+
+#[tracing::instrument]
+pub async fn handle(state: State, interaction: Interaction) {
+    let user_id = interaction
+        .member
+        .as_ref()
+        .and_then(|member| member.user.as_ref().map(|user| user.id));
+
+    let user_id = match user_id {
+        Some(user_id) => user_id,
+        None => {
+            state
+                .discord_client
+                .interaction(state.discord_application_id)
+                .create_response(
+                    interaction.id,
+                    &interaction.token,
+                    &InteractionResponse {
+                        kind: InteractionResponseType::ChannelMessageWithSource,
+                        data: Some(
+                            InteractionResponseDataBuilder::new()
+                                .content("TODO")
+                                .build(),
+                        ),
+                    },
+                )
+                .await
+                .expect("TODO");
+            return;
+        }
+    };
+
+    let previous_consent = state
+        .user_data_manager
+        .update(user_id, |mut user_data| {
+            let previous_consent = user_data
+                .reborrow()
+                .get_voice_recording_consent()
+                .expect("TODO");
+            user_data.set_voice_recording_consent(Consent::Granted);
+
+            previous_consent
+        })
+        .await
+        .expect("TODO");
+
+    state
+        .discord_client
+        .interaction(state.discord_application_id)
+        .create_response(
+            interaction.id,
+            &interaction.token,
+            &InteractionResponse {
+                kind: InteractionResponseType::ChannelMessageWithSource,
+                data: Some(
+                    InteractionResponseDataBuilder::new()
+                        .content(format!(
+                            "opted you in, your previous consent was {previous_consent:?}"
+                        ))
+                        .flags(MessageFlags::EPHEMERAL)
+                        .build(),
+                ),
+            },
+        )
+        .await
+        .expect("TODO");
+}

src/command/opt_out.rs

@@ -0,0 +1,90 @@
+use std::sync::LazyLock;
+
+use twilight_model::{
+    application::{
+        command::{Command, CommandType},
+        interaction::Interaction,
+    },
+    channel::message::MessageFlags,
+    http::interaction::{InteractionResponse, InteractionResponseType},
+};
+use twilight_util::builder::{InteractionResponseDataBuilder, command::CommandBuilder};
+
+use crate::{command::State, user_capnp::user::Consent};
+
+const NAME: &str = "opt-out";
+const DESCRIPTION: &str = "Opt out of being recorded";
+
+pub static COMMAND: LazyLock<Command> = LazyLock::new(|| {
+    CommandBuilder::new(NAME, DESCRIPTION, CommandType::ChatInput)
+        .validate()
+        .expect("command wasn't correct")
+        .build()
+});
+
+#[tracing::instrument]
+pub async fn handle(state: State, interaction: Interaction) {
+    let user_id = interaction
+        .member
+        .as_ref()
+        .and_then(|member| member.user.as_ref().map(|user| user.id));
+
+    let user_id = match user_id {
+        Some(user_id) => user_id,
+        None => {
+            state
+                .discord_client
+                .interaction(state.discord_application_id)
+                .create_response(
+                    interaction.id,
+                    &interaction.token,
+                    &InteractionResponse {
+                        kind: InteractionResponseType::ChannelMessageWithSource,
+                        data: Some(
+                            InteractionResponseDataBuilder::new()
+                                .content("TODO")
+                                .build(),
+                        ),
+                    },
+                )
+                .await
+                .expect("TODO");
+            return;
+        }
+    };
+
+    let previous_consent = state
+        .user_data_manager
+        .update(user_id, |mut user_data| {
+            let previous_consent = user_data
+                .reborrow()
+                .get_voice_recording_consent()
+                .expect("TODO");
+            user_data.set_voice_recording_consent(Consent::Withheld);
+
+            previous_consent
+        })
+        .await
+        .expect("TODO");
+
+    state
+        .discord_client
+        .interaction(state.discord_application_id)
+        .create_response(
+            interaction.id,
+            &interaction.token,
+            &InteractionResponse {
+                kind: InteractionResponseType::ChannelMessageWithSource,
+                data: Some(
+                    InteractionResponseDataBuilder::new()
+                        .content(format!(
+                            "opted you out, your previous consent was {previous_consent:?}"
+                        ))
+                        .flags(MessageFlags::EPHEMERAL)
+                        .build(),
+                ),
+            },
+        )
+        .await
+        .expect("TODO");
+}

src/lib.rs

@@ -1,12 +1,24 @@
-mod one_to_many;
-mod one_to_many_with_data;
-mod one_to_one;
-mod track_vcs;
-mod vc_user;
-
-pub use one_to_many::OneToManyUniqueBTreeMap;
-pub use one_to_many_with_data::OneToManyUniqueBTreeMapWithData;
-pub use one_to_one::OneToOneBTreeMap;
-
-pub use track_vcs::{VCs, initialize_vcs, update_vcs};
-pub use vc_user::{UserInVCData, VoiceStatus};
+mod bot_data;
+pub mod command;
+mod one_to_many;
+mod one_to_many_with_data;
+mod one_to_one;
+mod operator_ext;
+mod option_ext;
+mod storage;
+mod track_vcs;
+mod user_data;
+mod vc_user;
+capnp::generated_code!(mod bot_capnp);
+capnp::generated_code!(mod user_capnp);
+
+pub use bot_data::BotDataManager;
+pub use command::{Router as CommandRouter, State, all as all_commands};
+pub use one_to_many::OneToManyUniqueBTreeMap;
+pub use one_to_many_with_data::OneToManyUniqueBTreeMapWithData;
+pub use one_to_one::OneToOneBTreeMap;
+pub use operator_ext::OperatorExt;
+pub use storage::Storage;
+pub use track_vcs::{GuildVoiceChannelToTextChannel, VCs, initialize_vcs, update_vcs};
+pub use user_data::UserDataManager;
+pub use vc_user::{UserInVCData, VoiceStatus};

src/main.rs

@@ -1,46 +1,107 @@
 use clap::Parser;
-use fomo_reducer::{VCs, initialize_vcs, update_vcs};
-use opendal::{IntoOperatorUri, Operator, OperatorUri};
+use fomo_reducer::{
+    BotDataManager, CommandRouter, GuildVoiceChannelToTextChannel, State, Storage, UserDataManager,
+    all_commands, command, initialize_vcs, update_vcs,
+};
 use secrecy::{ExposeSecret, SecretString};
-use snafu::Snafu;
-use std::{fmt::Debug, str::FromStr};
+use snafu::{OptionExt, ResultExt, Snafu};
+use songbird::{
+    Config, Songbird,
+    driver::{Channels, DecodeConfig, SampleRate},
+    shards::TwilightMap,
+};
+use std::{collections::BTreeMap, fmt::Debug, str::FromStr, sync::Arc};
+use strum::EnumString;
+use tokio::{select, signal::ctrl_c, task::JoinSet};
+use tokio_util::{sync::CancellationToken, time::FutureExt as _};
 use tracing_subscriber::{EnvFilter, fmt::format::FmtSpan};
-use twilight_gateway::{Event, EventTypeFlags, Intents, Shard, ShardId, StreamExt};
-use twilight_model::id::{Id, marker::UserMarker};
+use twilight_gateway::{Event, EventTypeFlags, Intents, Shard, StreamExt};
+use twilight_model::{
+    application::interaction::InteractionData,
+    gateway::{
+        payload::{incoming::InteractionCreate, outgoing::UpdatePresence},
+        presence::{ActivityType, MinimalActivity, Status},
+    },
+    id::{
+        Id,
+        marker::{ChannelMarker, GuildMarker, UserMarker},
+    },
+};
 
-#[derive(Clone)]
-struct OpendalOperator {
-    uri: OperatorUri,
-    operator: Operator,
+#[derive(Clone, Copy, Debug, strum::Display, EnumString)]
+enum AudioChannels {
+    Mono,
+    Stereo,
 }
 
-impl FromStr for OpendalOperator {
-    type Err = opendal::Error;
-
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
-        let uri = s.into_operator_uri()?;
-        let operator = Operator::from_uri(&uri)?;
-
-        Ok(Self { uri, operator })
+impl From<AudioChannels> for Channels {
+    fn from(value: AudioChannels) -> Self {
+        match value {
+            AudioChannels::Mono => Channels::Mono,
+            AudioChannels::Stereo => Channels::Stereo,
+        }
     }
 }
 
-impl OpendalOperator {
-    fn into_inner(self) -> Operator {
-        self.operator
+#[derive(Clone, Copy, Debug, strum::Display, EnumString)]
+enum AudioSampleRate {
+    #[strum(serialize = "8000Hz")]
+    Hz8000,
+    #[strum(serialize = "12000Hz")]
+    Hz12000,
+    #[strum(serialize = "16000Hz")]
+    Hz16000,
+    #[strum(serialize = "24000Hz")]
+    Hz24000,
+    #[strum(serialize = "48000Hz")]
+    Hz48000,
+}
+
+impl From<AudioSampleRate> for SampleRate {
+    fn from(value: AudioSampleRate) -> Self {
+        match value {
+            AudioSampleRate::Hz8000 => SampleRate::Hz8000,
+            AudioSampleRate::Hz12000 => SampleRate::Hz12000,
+            AudioSampleRate::Hz16000 => SampleRate::Hz16000,
+            AudioSampleRate::Hz24000 => SampleRate::Hz24000,
+            AudioSampleRate::Hz48000 => SampleRate::Hz48000,
+        }
     }
 }
 
-impl From<OpendalOperator> for Operator {
-    fn from(wrapper: OpendalOperator) -> Self {
-        wrapper.into_inner()
-    }
+#[derive(Debug, Snafu)]
+enum ParseGuildVCToTextChannelError {
+    NoScope,
+
+    NoRelation,
+
+    ParseGuildError {
+        source: <Id<GuildMarker> as FromStr>::Err,
+    },
+
+    ParseVoiceChannelError {
+        source: <Id<ChannelMarker> as FromStr>::Err,
+    },
+
+    ParseTextChannelError {
+        source: <Id<ChannelMarker> as FromStr>::Err,
+    },
 }
 
-impl Debug for OpendalOperator {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        Debug::fmt(&self.uri, f)
-    }
+fn parse_guild_vc_to_text_channel(
+    source: &str,
+) -> Result<(Id<GuildMarker>, Id<ChannelMarker>, Id<ChannelMarker>), ParseGuildVCToTextChannelError>
+{
+    let (guild, voice_channel_and_text_channel) = source.split_once(':').context(NoScopeSnafu)?;
+    let (voice_channel, text_channel) = voice_channel_and_text_channel
+        .split_once("->")
+        .context(NoRelationSnafu)?;
+
+    let guild = guild.parse().context(ParseGuildSnafu)?;
+    let voice_channel = voice_channel.parse().context(ParseVoiceChannelSnafu)?;
+    let text_channel = text_channel.parse().context(ParseTextChannelSnafu)?;
+
+    Ok((guild, voice_channel, text_channel))
 }
 
 #[derive(Debug, Parser)]
@@ -49,15 +110,41 @@ struct AppArgs {
     discord_token: SecretString,
 
     #[arg(long, env)]
-    storage: OpendalOperator,
+    discord_bot_owner_user_id: Id<UserMarker>,
 
     #[arg(long, env)]
-    bot_owner: Id<UserMarker>,
+    discord_nickname: Option<Arc<str>>,
+
+    #[arg(long, env)]
+    discord_status: Option<Arc<str>>,
+
+    #[arg(long, env, value_parser = parse_guild_vc_to_text_channel)]
+    discord_voice_channel_corresponding_text_channel:
+        Vec<(Id<GuildMarker>, Id<ChannelMarker>, Id<ChannelMarker>)>,
+
+    #[arg(long, env, default_value_t = AudioChannels::Mono)]
+    audio_channels: AudioChannels,
+
+    #[arg(long, env, default_value_t = AudioSampleRate::Hz12000)]
+    audio_sample_rate: AudioSampleRate,
+
+    #[arg(long, env)]
+    bot_data: Storage,
+
+    #[arg(long, env)]
+    user_data: Storage,
+
+    #[arg(long, env)]
+    recording_data: Storage,
 }
 
 #[derive(Parser)]
 struct LoggingArgs {
-    #[arg(long = "logging-directives", env = "RUST_LOG", default_value = "info")]
+    #[arg(
+        long = "logging-directives",
+        env = "RUST_LOG",
+        default_value = "warn,fomo_reducer=debug"
+    )]
     env_filter: EnvFilter,
 }
 
@@ -71,7 +158,10 @@ struct Args {
 }
 
 #[derive(Debug, Snafu)]
-enum MainError {}
+enum MainError {
+    /// the program was cancelled, perhaps by Ctrl-C / SIGINT
+    Cancelled,
+}
 
 #[snafu::report]
 #[tokio::main]
@@ -93,49 +183,317 @@ async fn main() -> Result<(), MainError> {
 
     let AppArgs {
         discord_token,
-        storage,
-        bot_owner,
+        discord_bot_owner_user_id,
+        discord_nickname,
+        discord_status,
+        discord_voice_channel_corresponding_text_channel,
+        audio_channels,
+        audio_sample_rate,
+        bot_data,
+        user_data,
+        recording_data,
     } = app_args;
 
+    let cancellation_token = CancellationToken::new();
+
     rustls::crypto::aws_lc_rs::default_provider()
         .install_default()
         .unwrap();
 
-    let shard_id = ShardId::new(0, 1);
-    let intents = Intents::GUILD_VOICE_STATES;
-    let mut shard = Shard::new(shard_id, discord_token.expose_secret().to_owned(), intents);
-
-    let vc_event_types = EventTypeFlags::GUILD_VOICE_STATES;
-    let mut next_event = shard.next_event(vc_event_types);
-
     let discord_client = twilight_http::Client::new(discord_token.expose_secret().to_owned());
-    let mut voice_status = initialize_vcs(&discord_client).await;
-    while let Some(event_res) = next_event.await {
+
+    let guilds = discord_client
+        .current_user_guilds()
+        .limit(200)
+        .await
+        .expect("TODO")
+        .model()
+        .await
+        .expect("TODO");
+
+    JoinSet::from_iter(guilds.into_iter().map(|guild| {
+        discord_client
+            .update_current_member(guild.id)
+            .nick(discord_nickname.as_deref())
+            .into_future()
+    }))
+    .join_all()
+    .await;
+
+    let discord_user = discord_client
+        .current_user()
+        .await
+        .expect("couldn't fetch current user") // TODO
+        .model()
+        .await
+        .expect("couldn't deserialize current user"); // TODO
+
+    let discord_user_id = discord_user.id;
+
+    let current_application = discord_client
+        .current_user_application()
+        .await
+        .expect("couldn't get current Discord application"); // TODO
+
+    let current_application = current_application
+        .model()
+        .await
+        .expect("couldn't get current Discord application"); // TODO
+
+    let discord_application_id = current_application.id;
+
+    let intents = Intents::GUILD_VOICE_STATES;
+    let config = twilight_gateway::Config::new(discord_token.expose_secret().to_owned(), intents);
+
+    let shards = twilight_gateway::create_recommended(&discord_client, config, |_id, builder| {
+        builder.build()
+    })
+    .await
+    .expect("TODO");
+    let shards = Vec::from_iter(shards);
+
+    let senders = TwilightMap::new(
+        shards
+            .iter()
+            .map(|shard| (shard.id().number(), shard.sender()))
+            .collect(),
+    );
+
+    let audio_channels = audio_channels.into();
+    let audio_sample_rate = audio_sample_rate.into();
+
+    let senders = Arc::new(senders);
+    let songbird = Songbird::twilight(senders, discord_user_id);
+    songbird.set_config(
+        Config::default().decode_mode(songbird::driver::DecodeMode::Decode(DecodeConfig::new(
+            audio_channels,
+            audio_sample_rate,
+        ))),
+    );
+
+    let interaction_client = discord_client.interaction(discord_application_id);
+
+    let commands = all_commands();
+
+    let returned_commands = interaction_client
+        .set_global_commands(
+            Vec::from_iter(
+                commands
+                    .iter()
+                    .map(|(command, _handler)| (*command).clone()),
+            )
+            .as_slice(),
+        )
+        .await
+        .expect("failed to set interaction commands") // TODO
+        .models()
+        .await
+        .expect("failed to deserialize set commands"); // TODO
+
+    let mut discord_command_name_to_returned_command = BTreeMap::from_iter(
+        returned_commands
+            .into_iter()
+            .map(|command| (command.name.clone(), command)),
+    );
+
+    let discord_opt_in_command = discord_command_name_to_returned_command
+        .remove(&command::opt_in::COMMAND.name)
+        .expect("TODO");
+    let discord_opt_out_command = discord_command_name_to_returned_command
+        .remove(&command::opt_out::COMMAND.name)
+        .expect("TODO");
+
+    let discord_opt_in_command_id = discord_opt_in_command.id.expect("TODO");
+    let discord_opt_out_command_id = discord_opt_out_command.id.expect("TODO");
+
+    let discord_opt_in_command_name = discord_opt_in_command.name.into();
+    let discord_opt_out_command_name = discord_opt_out_command.name.into();
+
+    let vcs = initialize_vcs(&discord_client).await;
+
+    let command_router = CommandRouter::from_iter(commands);
+    let command_router = Arc::new(command_router);
+
+    let discord_client = Arc::new(discord_client);
+    let songbird = Arc::new(songbird);
+    let vcs = Arc::new(vcs);
+
+    let bot_data = bot_data.into_inner();
+    let recording_data = recording_data.into_inner();
+    let user_data = user_data.into_inner();
+
+    let bot_data_manager = BotDataManager::new(bot_data);
+    let user_data_manager = UserDataManager::new(user_data);
+
+    let discord_voice_channel_corresponding_text_channel = {
+        let mut map = GuildVoiceChannelToTextChannel::default();
+
+        for (guild_id, voice_channel_id, text_channel_id) in
+            discord_voice_channel_corresponding_text_channel
+        {
+            map.entry(guild_id)
+                .or_default()
+                .insert(voice_channel_id, text_channel_id);
+        }
+
+        map
+    };
+    let discord_voice_channel_corresponding_text_channel =
+        Arc::new(discord_voice_channel_corresponding_text_channel);
+
+    let state = State {
+        audio_channels,
+        audio_sample_rate,
+        bot_data_manager,
+        cancellation_token: cancellation_token.clone(),
+        discord_application_id,
+        discord_bot_owner_user_id,
+        discord_client,
+        discord_opt_in_command_id,
+        discord_opt_in_command_name,
+        discord_opt_out_command_id,
+        discord_opt_out_command_name,
+        discord_user_id,
+        discord_voice_channel_corresponding_text_channel,
+        recording_data,
+        songbird,
+        user_data_manager,
+        vcs,
+    };
+
+    if let Some(discord_status) = discord_status {
+        shards.iter().for_each(|shard| {
+            shard.command(
+                &UpdatePresence::new(
+                    vec![
+                        MinimalActivity {
+                            kind: ActivityType::Listening,
+                            name: (*discord_status).to_owned(),
+                            url: None,
+                        }
+                        .into(),
+                    ],
+                    false,
+                    None,
+                    Status::Idle,
+                )
+                .expect("TODO"),
+            )
+        });
+    }
+
+    let run_shards = JoinSet::from_iter(
+        shards
+            .into_iter()
+            .map(|shard| handle_events(command_router.clone(), state.clone(), shard)),
+    );
+    let run_shards = run_shards.join_all();
+    tokio::pin!(run_shards);
+
+    tokio::spawn({
+        let cancellation_token = cancellation_token.clone();
+        async move {
+            match ctrl_c().await {
+                Ok(()) => cancellation_token.cancel(),
+                Err(error) => tracing::error!(?error, "failed to listen for interrupt signal"),
+            }
+        }
+    });
+
+    select! {
+        _ = &mut run_shards => {
+            Ok(())
+        }
+        () = cancellation_token.cancelled() => {
+            tracing::warn!("waiting for tasks to gracefully shut down");
+            run_shards.await;
+
+            Err(MainError::Cancelled)
+        }
+    }
+}
+
+#[tracing::instrument(skip(command_router, shard, state))]
+async fn handle_events(command_router: Arc<CommandRouter>, state: State, mut shard: Shard) {
+    let event_types = EventTypeFlags::GUILD_VOICE_STATES
+        | EventTypeFlags::INTERACTION_CREATE
+        | EventTypeFlags::VOICE_SERVER_UPDATE
+        | EventTypeFlags::VOICE_STATE_UPDATE;
+
+    while let Some(Some(event_res)) = shard
+        .next_event(event_types)
+        .with_cancellation_token(&state.cancellation_token)
+        .await
+    {
         match event_res {
+            Ok(twilight_model::gateway::event::Event::GatewayClose(frame_option)) => {
+                tracing::warn!(?frame_option);
+                return;
+            }
             Ok(event) => {
-                tracing::debug!(?voice_status, "before handling");
-                handle_event(event, &mut voice_status).await;
-                tracing::debug!(?voice_status, "after handling");
+                handle_event(command_router.clone(), state.clone(), event).await;
+            }
+            Err(reconnect_error)
+                if matches!(
+                    reconnect_error.kind(),
+                    &twilight_gateway::error::ReceiveMessageErrorType::Reconnect
+                ) =>
+            {
+                tracing::error!(?reconnect_error);
+                return;
             }
             Err(error) => {
                 tracing::error!(?error);
             }
         }
-
-        next_event = shard.next_event(vc_event_types);
     }
-
-    Ok(())
 }
 
-#[tracing::instrument(skip(vcs))]
-async fn handle_event(event: Event, vcs: &mut VCs) {
+#[tracing::instrument(skip(command_router, state))]
+async fn handle_event(command_router: Arc<CommandRouter>, state: State, event: Event) {
+    state.songbird.process(&event).await;
+
     match event {
         Event::VoiceStateUpdate(voice_state_update) => {
-            update_vcs(&voice_state_update, vcs);
+            update_vcs(&voice_state_update, &state.vcs);
         }
-        other => {
-            tracing::warn!(?other, "wasn't expected");
+        Event::InteractionCreate(interaction_create) => {
+            let InteractionCreate(interaction) = *interaction_create;
+
+            match &interaction.data {
+                None => {
+                    tracing::warn!("missing expected interaction data");
+                }
+                Some(InteractionData::ApplicationCommand(command_data)) => {
+                    let command_name = command_data.name.clone();
+                    tokio::spawn(async move {
+                        command_router
+                            .handle(state, &command_name, interaction)
+                            .await;
+                    });
+                }
+
+                Some(InteractionData::MessageComponent(component_data)) => {
+                    tracing::warn!(
+                        ?component_data,
+                        "wasn't expected because this bot has no modal features"
+                    );
+                }
+
+                Some(InteractionData::ModalSubmit(modal_data)) => {
+                    tracing::warn!(
+                        ?modal_data,
+                        "wasn't expected because this bot has no modal features"
+                    );
+                }
+
+                Some(other_interaction_data) => {
+                    tracing::warn!(?other_interaction_data, "wasn't expected");
+                }
+            }
+        }
+        other_event => {
+            tracing::warn!(?other_event, "wasn't expected");
         }
     }
 }

src/one_to_many_with_data.rs

@@ -7,6 +7,7 @@ pub struct OneToManyUniqueBTreeMapWithData<Left, Right, RightData> {
 }
 
 impl<Left, Right, RightData> Default for OneToManyUniqueBTreeMapWithData<Left, Right, RightData> {
+    #[inline]
     fn default() -> Self {
         Self {
             left_to_rights: Default::default(),
@@ -47,10 +48,25 @@ where
         self.right_to_left.get(right)
     }
 
+    pub fn get_left_and_data_for(&self, right: &Right) -> Option<(&Left, &RightData)> {
+        let left = self.right_to_left.get(right)?;
+        let rights = self.get_rights_for(left)?;
+
+        let right_data = rights.get(right)?;
+
+        Some((left, right_data))
+    }
+
+    pub fn get_data_for(&self, right: &Right) -> Option<&RightData> {
+        let (_left, right_data) = self.get_left_and_data_for(right)?;
+
+        Some(right_data)
+    }
+
     pub fn remove_left(&mut self, left: &Left) -> Option<(Left, BTreeMap<Right, RightData>)> {
         let (left, rights) = self.left_to_rights.remove_entry(left)?;
 
-        for (right, _right_data) in &rights {
+        for right in rights.keys() {
             self.right_to_left.remove(right);
         }
 
@@ -69,3 +85,33 @@ where
         Some((left, right, right_data))
     }
 }
+
+impl<Left, Right, RightData> FromIterator<(Left, Right, RightData)>
+    for OneToManyUniqueBTreeMapWithData<Left, Right, RightData>
+where
+    Left: Ord + Clone,
+    Right: Ord + Clone,
+{
+    #[inline]
+    fn from_iter<T: IntoIterator<Item = (Left, Right, RightData)>>(iter: T) -> Self {
+        let mut this = Self::default();
+
+        this.extend(iter);
+
+        this
+    }
+}
+
+impl<Left, Right, RightData> Extend<(Left, Right, RightData)>
+    for OneToManyUniqueBTreeMapWithData<Left, Right, RightData>
+where
+    Left: Ord + Clone,
+    Right: Ord + Clone,
+{
+    #[inline]
+    fn extend<T: IntoIterator<Item = (Left, Right, RightData)>>(&mut self, iter: T) {
+        for (left, right, right_data) in iter {
+            self.insert(left, right, right_data);
+        }
+    }
+}

src/operator_ext.rs

@@ -0,0 +1,25 @@
+use extension_traits::extension;
+use opendal::{Buffer, Error, ErrorKind, FuturesAsyncReader, Operator};
+
+#[extension(pub trait OperatorExt)]
+impl Operator {
+    async fn read_if_exists(&self, path: &str) -> Result<Option<Buffer>, Error> {
+        match self.read(path).await {
+            Ok(buffer) => Ok(Some(buffer)),
+            Err(error) if matches!(error.kind(), ErrorKind::NotFound) => Ok(None),
+            Err(error) => Err(error),
+        }
+    }
+
+    async fn async_reader_if_exists(
+        &self,
+        path: &str,
+    ) -> Result<Option<FuturesAsyncReader>, Error> {
+        let reader = self.reader(path).await?;
+        match reader.into_futures_async_read(..).await {
+            Ok(reader) => Ok(Some(reader)),
+            Err(error) if matches!(error.kind(), ErrorKind::NotFound) => Ok(None),
+            Err(error) => Err(error),
+        }
+    }
+}

src/option_ext.rs

@@ -0,0 +1,11 @@
+use extension_traits::extension;
+
+#[extension(pub trait OptionExt)]
+impl<S> Option<S> {
+    async fn map_async<M, Fut: Future<Output = M>, F: FnOnce(S) -> Fut>(self, f: F) -> Option<M> {
+        match self {
+            Some(s) => Some(f(s).await),
+            None => None,
+        }
+    }
+}

src/storage.rs

@@ -0,0 +1,38 @@
+use std::{fmt::Debug, str::FromStr};
+
+use opendal::{IntoOperatorUri, Operator, OperatorUri};
+
+#[derive(Clone)]
+pub struct Storage {
+    uri: OperatorUri,
+    operator: Operator,
+}
+
+impl FromStr for Storage {
+    type Err = opendal::Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let uri = s.into_operator_uri()?;
+        let operator = Operator::from_uri(&uri)?;
+
+        Ok(Self { uri, operator })
+    }
+}
+
+impl Storage {
+    pub fn into_inner(self) -> Operator {
+        self.operator
+    }
+}
+
+impl From<Storage> for Operator {
+    fn from(wrapper: Storage) -> Self {
+        wrapper.into_inner()
+    }
+}
+
+impl Debug for Storage {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        Debug::fmt(&self.uri, f)
+    }
+}

src/track_vcs.rs

@@ -1,102 +1,135 @@
-type VCsInServer = OneToManyUniqueBTreeMapWithData<Id<ChannelMarker>, Id<UserMarker>, UserInVCData>;
-
-pub type VCs = BTreeMap<Id<GuildMarker>, VCsInServer>;
-
-use std::collections::BTreeMap;
-
-use twilight_model::{
-    gateway::payload::incoming::VoiceStateUpdate,
-    id::{
-        Id,
-        marker::{ChannelMarker, GuildMarker, UserMarker},
-    },
-};
-
-use crate::{OneToManyUniqueBTreeMapWithData, UserInVCData, VoiceStatus};
-
-#[tracing::instrument(skip(discord_client), ret)]
-pub async fn initialize_vcs(discord_client: &twilight_http::Client) -> VCs {
-    let mut vcs = VCs::default();
-
-    if let Ok(guilds_res) = discord_client.current_user_guilds().limit(200).await
-        && let Ok(guilds) = guilds_res.model().await
-    {
-        for guild in guilds {
-            if let Ok(guild_members_res) = discord_client.guild_members(guild.id).limit(999).await
-                && let Ok(guild_members) = guild_members_res.model().await
-            {
-                for member in guild_members {
-                    if let Ok(voice_state_res) = discord_client
-                        .user_voice_state(guild.id, member.user.id)
-                        .await
-                        && let Ok(voice_state) = voice_state_res.model().await
-                    {
-                        tracing::info!(?member.user.id, ?voice_state);
-
-                        let voice_status = VoiceStatus::builder()
-                            .self_deafened(voice_state.self_deaf)
-                            .self_muted(voice_state.self_mute)
-                            .server_deafened(voice_state.deaf)
-                            .server_muted(voice_state.mute)
-                            .camming(voice_state.self_video)
-                            .streaming(voice_state.self_stream)
-                            .build();
-                        let user_in_vc_data = voice_status.into();
-
-                        if let Some(channel_id) = voice_state.channel_id {
-                            vcs.entry(guild.id).or_default().insert(
-                                channel_id,
-                                member.user.id,
-                                user_in_vc_data,
-                            );
-                        }
-                    }
-                }
-            }
-        }
-    }
-
-    vcs
-}
-
-pub fn update_vcs(voice_state_update: &VoiceStateUpdate, vcs: &mut VCs) {
-    let user_id = voice_state_update.user_id;
-    match voice_state_update.guild_id {
-        Some(guild_id) => match voice_state_update.channel_id {
-            Some(channel_id) => {
-                let voice_status = VoiceStatus::builder()
-                    .self_deafened(voice_state_update.self_deaf)
-                    .self_muted(voice_state_update.self_mute)
-                    .server_deafened(voice_state_update.deaf)
-                    .server_muted(voice_state_update.mute)
-                    .camming(voice_state_update.self_video)
-                    .streaming(voice_state_update.self_stream)
-                    .build();
-                let user_in_vc_data = voice_status.into();
-
-                vcs.entry(guild_id)
-                    .or_default()
-                    .insert(channel_id, user_id, user_in_vc_data);
-
-                tracing::info!(
-                    ?guild_id,
-                    ?channel_id,
-                    ?user_id,
-                    "connected or otherwise changed state while connected"
-                );
-            }
-
-            None => {
-                if let Some(channel_vcers) = vcs.get_mut(&guild_id) {
-                    channel_vcers.remove_right(&user_id);
-                }
-
-                tracing::info!(?guild_id, ?user_id, "disconnected");
-            }
-        },
-
-        None => {
-            tracing::error!("why doesn't this have a guild id attached?!");
-        }
-    }
-}
+use std::collections::BTreeMap;
+
+use dashmap::DashMap;
+use futures::{StreamExt, stream::FuturesUnordered};
+use twilight_model::{
+    gateway::payload::incoming::VoiceStateUpdate,
+    id::{
+        Id,
+        marker::{ChannelMarker, GuildMarker, UserMarker},
+    },
+};
+
+use crate::{OneToManyUniqueBTreeMapWithData, OneToOneBTreeMap, UserInVCData, VoiceStatus};
+
+pub type GuildVoiceChannelToTextChannel =
+    BTreeMap<Id<GuildMarker>, OneToOneBTreeMap<Id<ChannelMarker>, Id<ChannelMarker>>>;
+
+type VCsInGuild = OneToManyUniqueBTreeMapWithData<Id<ChannelMarker>, Id<UserMarker>, UserInVCData>;
+pub type VCs = DashMap<Id<GuildMarker>, VCsInGuild>;
+
+#[tracing::instrument(skip(discord_client), ret)]
+async fn initialize_user_in_vc(
+    discord_client: &twilight_http::Client,
+    guild_id: Id<GuildMarker>,
+    user_id: Id<UserMarker>,
+) -> Option<(Id<ChannelMarker>, UserInVCData)> {
+    if let Ok(voice_state_res) = discord_client.user_voice_state(guild_id, user_id).await
+        && let Ok(voice_state) = voice_state_res.model().await
+    {
+        tracing::info!(?user_id, ?voice_state);
+
+        let voice_status = VoiceStatus::builder()
+            .self_deafened(voice_state.self_deaf)
+            .self_muted(voice_state.self_mute)
+            .server_deafened(voice_state.deaf)
+            .server_muted(voice_state.mute)
+            .camming(voice_state.self_video)
+            .streaming(voice_state.self_stream)
+            .build();
+        let user_in_vc_data = voice_status.into();
+
+        voice_state
+            .channel_id
+            .map(|channel_id| (channel_id, user_in_vc_data))
+    } else {
+        None // TODO
+    }
+}
+
+#[tracing::instrument(skip(discord_client), ret)]
+async fn initialize_server_vcs(
+    discord_client: &twilight_http::Client,
+    id: Id<GuildMarker>,
+) -> VCsInGuild {
+    if let Ok(guild_members_res) = discord_client.guild_members(id).limit(999).await
+        && let Ok(guild_members) = guild_members_res.model().await
+    {
+        FuturesUnordered::from_iter(guild_members.into_iter().map(|member| async move {
+            (
+                member.user.id,
+                initialize_user_in_vc(discord_client, id, member.user.id).await,
+            )
+        }))
+        .filter_map(
+            |(user_id, channel_id_and_user_in_vc_data_option)| async move {
+                channel_id_and_user_in_vc_data_option
+                    .map(|(channel_id, user_in_vc_data)| (channel_id, user_id, user_in_vc_data))
+            },
+        )
+        .collect()
+        .await
+    } else {
+        Default::default()
+    }
+}
+
+#[tracing::instrument(skip(discord_client), ret)]
+pub async fn initialize_vcs(discord_client: &twilight_http::Client) -> VCs {
+    if let Ok(guilds_res) = discord_client.current_user_guilds().limit(200).await
+        && let Ok(guilds) = guilds_res.model().await
+    {
+        FuturesUnordered::from_iter(guilds.into_iter().map(|guild| async move {
+            let guild_vcs = initialize_server_vcs(discord_client, guild.id).await;
+
+            (guild.id, guild_vcs)
+        }))
+        .collect()
+        .await
+    } else {
+        Default::default()
+    }
+}
+
+#[tracing::instrument(skip(vcs))]
+pub fn update_vcs(voice_state_update: &VoiceStateUpdate, vcs: &VCs) {
+    let user_id = voice_state_update.user_id;
+    match voice_state_update.guild_id {
+        Some(guild_id) => match voice_state_update.channel_id {
+            Some(channel_id) => {
+                let voice_status = VoiceStatus::builder()
+                    .self_deafened(voice_state_update.self_deaf)
+                    .self_muted(voice_state_update.self_mute)
+                    .server_deafened(voice_state_update.deaf)
+                    .server_muted(voice_state_update.mute)
+                    .camming(voice_state_update.self_video)
+                    .streaming(voice_state_update.self_stream)
+                    .build();
+                let user_in_vc_data = voice_status.into();
+
+                vcs.entry(guild_id)
+                    .or_default()
+                    .insert(channel_id, user_id, user_in_vc_data);
+
+                tracing::info!(
+                    ?guild_id,
+                    ?channel_id,
+                    ?user_id,
+                    "connected or otherwise changed state while connected"
+                );
+            }
+
+            None => {
+                if let Some(mut channel_vcers) = vcs.get_mut(&guild_id) {
+                    channel_vcers.remove_right(&user_id);
+                }
+
+                tracing::info!(?guild_id, ?user_id, "disconnected");
+            }
+        },
+
+        None => {
+            tracing::error!("why doesn't this have a guild id attached?!");
+        }
+    }
+}

src/user_data.rs

@@ -0,0 +1,254 @@
+use std::str::FromStr;
+
+use async_compression::futures::{bufread::BrotliDecoder, write::BrotliEncoder};
+use capnp::message::TypedBuilder;
+use futures::{AsyncReadExt, AsyncWriteExt, TryStream, TryStreamExt};
+use opendal::Operator;
+use snafu::{OptionExt as _, ResultExt as _, Snafu, ensure};
+use twilight_model::id::{Id, marker::UserMarker};
+
+use crate::{OperatorExt, option_ext::OptionExt as _, user_capnp};
+
+pub const RECORD_IF_CONSENT_UNSPECIFIED: bool = true;
+
+#[derive(Debug, Clone)]
+pub struct UserDataManager {
+    operator: Operator,
+}
+
+impl UserDataManager {
+    pub fn new(operator: Operator) -> Self {
+        Self { operator }
+    }
+}
+
+fn path(id: Id<UserMarker>) -> String {
+    format!("{id}/data.bin.brotli")
+}
+
+#[derive(Debug, Snafu)]
+#[snafu(module)]
+pub enum ParsePathError {
+    /// paths must have a / in them because that's how user data is stored, but this one doesn't have one
+    MissingSlashError,
+
+    /// if this isn't a directory, then the file must be "data.bin.brotli" but it was actually {actual:?}
+    WrongFileError { actual: String },
+
+    /// couldn't parse the directory as a user ID
+    ParseUserIdError {
+        source: <Id<UserMarker> as FromStr>::Err,
+    },
+}
+
+fn parse(path: &str) -> Result<Id<UserMarker>, ParsePathError> {
+    let (directory, file) = path
+        .rsplit_once("/")
+        .context(parse_path_error::MissingSlashSnafu)?;
+
+    ensure!(
+        file == "" || file == "data.bin.brotli",
+        parse_path_error::WrongFileSnafu {
+            actual: file.to_owned()
+        }
+    );
+
+    let user_id = directory
+        .parse()
+        .context(parse_path_error::ParseUserIdSnafu)?;
+
+    Ok(user_id)
+}
+
+#[derive(Debug, Snafu)]
+#[snafu(module)]
+pub enum ListError {
+    /// error creating a lister through the storage operator
+    CreateListerError { source: opendal::Error },
+}
+
+#[derive(Debug, Snafu)]
+#[snafu(module)]
+pub enum EntryError {
+    /// failed to get an entry from the storage operator's lister
+    ReceiveEntryError { source: opendal::Error },
+
+    /// failed to parse the entry as an acceptable path
+    ParsePathError { source: ParsePathError },
+}
+
+impl UserDataManager {
+    pub async fn list(
+        &self,
+    ) -> Result<impl TryStream<Ok = Id<UserMarker>, Error = EntryError> + Unpin, ListError> {
+        let lister = self
+            .operator
+            .lister("")
+            .await
+            .context(list_error::CreateListerSnafu)?;
+
+        Ok(lister
+            .map_err(|error| EntryError::ReceiveEntryError { source: error })
+            .and_then(|entry| {
+                std::future::ready(parse(entry.path()).context(entry_error::ParsePathSnafu))
+            }))
+    }
+}
+
+#[derive(Debug, Snafu)]
+#[snafu(module)]
+pub enum WithError {
+    /// couldn't read data for this user from the storage operator
+    ReadError { source: opendal::Error },
+
+    /// couldn't decompress the user data from storage
+    DecompressionError { source: std::io::Error },
+
+    /// couldn't deserialize the user data
+    DeserializeError { source: capnp::Error },
+}
+
+impl UserDataManager {
+    pub async fn with<R>(
+        &self,
+        id: Id<UserMarker>,
+        f: impl FnOnce(user_capnp::user::Reader<'_>) -> R,
+    ) -> Result<R, WithError> {
+        let compressed_buffer = self
+            .operator
+            .async_reader_if_exists(&path(id))
+            .await
+            .context(with_error::ReadSnafu)?;
+
+        let decompressed_reader = compressed_buffer.map(BrotliDecoder::new);
+        let decompressed = decompressed_reader
+            .map_async(|mut reader| async move {
+                let mut vec = Vec::new();
+                reader.read_to_end(&mut vec).await?;
+                Ok(vec)
+            })
+            .await
+            .transpose()
+            .context(with_error::DecompressionSnafu)?;
+
+        let mut message = TypedBuilder::<user_capnp::user::Owned>::new_default();
+        let fallback = message.init_root();
+
+        let mut user_data = fallback.into_reader();
+        let message_reader;
+
+        if let Some(mut bytes) = decompressed.as_deref() {
+            message_reader = capnp::serialize::read_message_from_flat_slice_no_alloc(
+                &mut bytes,
+                Default::default(),
+            )
+            .context(with_error::DeserializeSnafu)?;
+
+            user_data = message_reader
+                .get_root()
+                .context(with_error::DeserializeSnafu)?;
+        }
+
+        Ok(f(user_data))
+    }
+}
+
+#[derive(Debug, Snafu)]
+#[snafu(module)]
+pub enum UpdateError {
+    /// couldn't read data for this user from the storage operator
+    ReadError { source: opendal::Error },
+
+    /// couldn't decompress the user data from storage
+    DecompressionError { source: std::io::Error },
+
+    /// couldn't deserialize the user data
+    DeserializeError { source: capnp::Error },
+
+    /// couldn't serialize the (modified) user data
+    SerializeError { source: capnp::Error },
+
+    /// couldn't create a writer for this user data in the storage operator
+    WriterError { source: opendal::Error },
+
+    /// couldn't write (modified) data for this user to the storage operator
+    WriteError { source: std::io::Error },
+
+    /// couldn't finalize writing (modified) data for this user to the storage operator
+    FinalizeError { source: std::io::Error },
+}
+
+impl UserDataManager {
+    pub async fn update<R>(
+        &self,
+        id: Id<UserMarker>,
+        f: impl FnOnce(user_capnp::user::Builder<'_>) -> R,
+    ) -> Result<R, UpdateError> {
+        let path = path(id);
+        let compressed_buffer = self
+            .operator
+            .async_reader_if_exists(&path)
+            .await
+            .context(update_error::ReadSnafu)?;
+
+        let decompressed_reader = compressed_buffer.map(BrotliDecoder::new);
+        let decompressed = decompressed_reader
+            .map_async(|mut reader| async move {
+                let mut vec = Vec::new();
+                reader.read_to_end(&mut vec).await?;
+                Ok(vec)
+            })
+            .await
+            .transpose()
+            .context(update_error::DecompressionSnafu)?;
+
+        let mut message = TypedBuilder::<user_capnp::user::Owned>::new_default();
+
+        let ret = if let Some(mut bytes) = decompressed.as_deref() {
+            let message_reader = capnp::serialize::read_message_from_flat_slice_no_alloc(
+                &mut bytes,
+                Default::default(),
+            )
+            .context(update_error::DeserializeSnafu)?;
+
+            let user_data = message_reader
+                .get_root()
+                .context(update_error::DeserializeSnafu)?;
+
+            message
+                .set_root(user_data)
+                .context(update_error::DeserializeSnafu)?;
+
+            f(
+                message.get_root().unwrap(), // this is logically impossible
+            )
+        } else {
+            f(message.init_root())
+        };
+
+        let mut buffer = Vec::new();
+        capnp::serialize::write_message(&mut buffer, message.borrow_inner())
+            .context(update_error::SerializeSnafu)?;
+
+        let compressed_writer = self
+            .operator
+            .writer(&path)
+            .await
+            .context(update_error::WriterSnafu)?
+            .into_futures_async_write();
+
+        let mut decompressed_writer = BrotliEncoder::new(compressed_writer);
+
+        decompressed_writer
+            .write_all(&buffer)
+            .await
+            .context(update_error::WriteSnafu)?;
+
+        decompressed_writer
+            .close()
+            .await
+            .context(update_error::FinalizeSnafu)?;
+
+        Ok(ret)
+    }
+}

user.capnp

@@ -0,0 +1,13 @@
+@0xc3e8b8ea9947b0c5;
+
+struct User {
+    notificationScript @0 :Text;
+
+    voiceRecordingConsent @1 :Consent = unspecified;
+
+    enum Consent {
+        unspecified @0;
+        granted @1;
+        withheld @2;
+    }
+}