Compare commits

...

39 Commits

Author SHA1 Message Date
906ddec843 chore: update lockfile 2026-04-26 18:38:08 -04:00
48e92d1736 fix: make /opt-in and /opt-out's responses ephemeral 2026-04-25 20:58:35 -04:00
733e8f73ea feat: create a BotDataManager abstraction by copying and pasting then editing UserDataManager 2026-04-25 20:56:41 -04:00
2c0d5c8479 feat: advertise opt in and opt out commands 2026-04-22 12:11:36 -04:00
9b479d1236 feat: respect consent to be recorded 2026-04-22 11:57:57 -04:00
29f97f82c4 fix: move title to author field cause you can't @mention users in the title 2026-04-22 01:05:34 -04:00
dd17de79de fix: close the writer instead of merely dropping it 2026-04-22 00:45:59 -04:00
bb96724454 fix: check for NotFound error at the correct site in OperatorExt::async_reader_if_exists 2026-04-22 00:27:11 -04:00
4dae5bac7a fix: specify webpki roots for tokio-websockets dependencies 2026-04-21 23:31:11 -04:00
37753fe37c chore: revert opendal to before reqwest 0.13 upgrade to see if it'll run in a SCRATCH container now 2026-04-21 21:45:32 -04:00
7fe6980867 feat: inspect user data 2026-04-21 15:06:56 -04:00
0ce26fc0e5 feat: user consent setting and retrieving (NOTE: does not affect recording yet) 2026-04-21 03:11:27 -04:00
62399c2046 feat: allow setting a corresponding text channel for a voice channel 2026-04-17 01:19:09 -04:00
612a696829 feat: mute 2026-04-16 23:13:45 -04:00
2bf42e47c5 chore: eliminate rocksdb which makes everything better 2026-04-16 19:29:21 -04:00
50230c43b8 chore: update lockfile 2026-04-16 17:59:58 -04:00
e4e274a543 fix: have to use match to detect ReceiveMessageErrorType 2026-04-16 02:03:52 -04:00
6cd7f00028 fix: detect gateway close and failure to reconnect 2026-04-16 02:01:50 -04:00
6a1d8f060f fix: I REALLY did this without rust-analyzer so THIS TIME fix the gateway close error path 2026-04-16 01:57:30 -04:00
766582c9e8 fix: I did this without rust-analyzer so fix the gateway close error path 2026-04-16 01:53:50 -04:00
784ec5e867 fix: exit the program to trigger a restart when the gateway closes 2026-04-16 01:51:34 -04:00
dbcc155c4c chore: run the program in the same image/stage as the builder (eliminating the runner) until I fix static linking 2026-04-15 18:43:51 -04:00
31adbc2027 chore: go back to Alpine dynamic linking since I think it's the best I can do 2026-04-15 17:46:25 -04:00
74bd37a67f chore: remove static requirement from rocksdb 2026-04-15 17:42:23 -04:00
e609428f95 chore: use a patched version of opendal with rocksdb static 2026-04-15 17:28:47 -04:00
d129913235 fix: revert to Rust image 2026-04-15 13:11:59 -04:00
066bff4c07 fix: add the unspecified Linux target 2026-04-15 13:08:37 -04:00
509278c6eb fix: build for unspecified Linux (not specifically Alpine) 2026-04-15 13:04:29 -04:00
f31d06bdf9 fix: force static linking 2026-04-15 13:01:35 -04:00
d93b53267e fix: depend on capnproto-dev 2026-04-15 12:41:16 -04:00
6ad797eaea fix: depend on capnproto 2026-04-15 12:31:51 -04:00
4b71e5ef85 fix: try building from a base Alpine image and install cargo instead 2026-04-15 02:15:37 -04:00
a99840ffb7 fix: depend on versions of llvm dev and static that actually exist 2026-04-15 02:02:55 -04:00
38196e84ec fix: depend on clang and llvm static and dev 2026-04-15 02:00:37 -04:00
0fdb83a9d3 fix: add required clang static dependency 2026-04-15 01:53:19 -04:00
dd6c1723e5 fix: add required libclang dependency 2026-04-15 01:48:59 -04:00
01a55d42ec fix: add required clang libs dependency 2026-04-15 01:40:58 -04:00
33a7b15720 feat: support configuring audio channels and sample rates of recordings; audio recordings now work! 2026-04-15 01:22:38 -04:00
b457375e69 fix: address compiler errors related to recording vc 2026-04-14 21:54:12 -04:00
17 changed files with 1668 additions and 857 deletions

508
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -6,12 +6,15 @@ edition = "2024"
[dependencies] [dependencies]
async-compression = { version = "0.4.41", features = ["brotli", "futures-io"] } async-compression = { version = "0.4.41", features = ["brotli", "futures-io"] }
async-trait = "0.1.89" async-trait = "0.1.89"
bytes = "1.11.1"
capnp = "0.25.3" capnp = "0.25.3"
clap = { version = "4.5.40", features = ["derive", "env"] } clap = { version = "4.5.40", features = ["derive", "env"] }
dashmap = "6.1.0" dashmap = "6.1.0"
extension-traits = "2.0.2"
futures = "0.3.32" futures = "0.3.32"
hound = "3.5.1" hound = "3.5.1"
opendal = { git = "https://github.com/apache/opendal", features = [ moka = { version = "0.12.15", features = ["future"] }
opendal = { git = "https://github.com/apache/opendal", rev = "ecf840b04afd2be109830b9978ba89759adfee79", features = [
"services-azfile", "services-azfile",
"services-aliyun-drive", "services-aliyun-drive",
"services-alluxio", "services-alluxio",
@@ -44,11 +47,11 @@ opendal = { git = "https://github.com/apache/opendal", features = [
"services-postgresql", "services-postgresql",
"services-redb", "services-redb",
"services-redis", "services-redis",
"services-rocksdb",
"services-s3", "services-s3",
"services-sled", "services-sled",
"services-webdav", "services-webdav",
] } ] }
opus2 = "0.4.0"
patricia_tree = "0.10.1" patricia_tree = "0.10.1"
rhai = "1.23.6" rhai = "1.23.6"
rustls = "0.23" rustls = "0.23"
@@ -62,12 +65,26 @@ songbird = { version = "0.6.0", default-features = false, features = [
"twilight", "twilight",
"tws", "tws",
] } ] }
strum = { version = "0.28.0", features = ["derive"] }
time = "0.3.47" time = "0.3.47"
tokio = { version = "1.46.0", features = ["rt-multi-thread", "macros", "signal"] } tokio = { version = "1.46.0", features = [
tokio-util = "0.7.18" "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 = "0.1.41"
tracing-subscriber = { version = "0.3.19", features = ["env-filter"] } tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
twilight-gateway = { version = "0.17", default-features = false, features = ["rustls-webpki-roots", "twilight-http"] } twilight-gateway = { version = "0.17", default-features = false, features = [
"rustls-webpki-roots",
"twilight-http",
] }
twilight-http = { version = "0.17", default-features = false, features = [ twilight-http = { version = "0.17", default-features = false, features = [
"rustls-webpki-roots", "rustls-webpki-roots",
"hickory", "hickory",
@@ -76,6 +93,7 @@ twilight-http = { version = "0.17", default-features = false, features = [
twilight-model = "0.17" twilight-model = "0.17"
twilight-util = { version = "0.17", features = ["builder"] } twilight-util = { version = "0.17", features = ["builder"] }
typed-builder = "0.23.2" typed-builder = "0.23.2"
yoke = "0.8.2"
[build-dependencies] [build-dependencies]
capnpc = "0.25.3" capnpc = "0.25.3"

View File

@@ -5,6 +5,10 @@ ARG BUILD_BASE_VERSION=0.5-r3
RUN --mount=type=cache,sharing=locked,target=/var/cache/apk \ RUN --mount=type=cache,sharing=locked,target=/var/cache/apk \
apk add --update build-base=${BUILD_BASE_VERSION} 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 ARG CMAKE_VERSION=4.1.3-r0
RUN --mount=type=cache,sharing=locked,target=/var/cache/apk \ RUN --mount=type=cache,sharing=locked,target=/var/cache/apk \
apk add --update cmake=${CMAKE_VERSION} apk add --update cmake=${CMAKE_VERSION}
@@ -13,7 +17,6 @@ ARG PROTOC_VERSION=31.1-r1
RUN --mount=type=cache,sharing=locked,target=/var/cache/apk \ RUN --mount=type=cache,sharing=locked,target=/var/cache/apk \
apk add --update protoc=${PROTOC_VERSION} apk add --update protoc=${PROTOC_VERSION}
ENV CXXFLAGS="-include cstdint"
RUN \ RUN \
# This one would be nice if it worked: # This one would be nice if it worked:
# --mount=type=bind,source=.,target=/root/app \ # --mount=type=bind,source=.,target=/root/app \

175
src/bot_data.rs Normal file
View File

@@ -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)
}
}

View File

@@ -1,10 +1,6 @@
use std::sync::LazyLock; use futures::TryStreamExt;
use async_compression::futures::bufread::BrotliDecoder;
use capnp::message::ReaderOptions;
use futures::AsyncReadExt;
use opendal::ErrorKind;
use snafu::{OptionExt, Snafu}; use snafu::{OptionExt, Snafu};
use std::sync::LazyLock;
use twilight_model::{ use twilight_model::{
application::{ application::{
command::{Command, CommandType}, command::{Command, CommandType},
@@ -17,14 +13,13 @@ use twilight_model::{
use twilight_util::builder::{ use twilight_util::builder::{
InteractionResponseDataBuilder, InteractionResponseDataBuilder,
command::CommandBuilder, command::CommandBuilder,
embed::{EmbedBuilder, EmbedFieldBuilder}, embed::{EmbedAuthorBuilder, EmbedBuilder, EmbedFieldBuilder},
}; };
use crate::{bot_capnp, command::State}; use crate::command::State;
const NAME: &str = "debug"; const NAME: &str = "info";
const DESCRIPTION: &str = const DESCRIPTION: &str = "Show various information";
"(Only the bot owner can use this) Show various information for debugging purposes";
pub static COMMAND: LazyLock<Command> = LazyLock::new(|| { pub static COMMAND: LazyLock<Command> = LazyLock::new(|| {
CommandBuilder::new(NAME, DESCRIPTION, CommandType::ChatInput) CommandBuilder::new(NAME, DESCRIPTION, CommandType::ChatInput)
@@ -114,54 +109,22 @@ pub async fn handle(state: State, interaction: Interaction) {
.await .await
.expect("TODO"); .expect("TODO");
let heat_script_description = { let heat_script_description = state
let compressed_result = state .bot_data_manager
.bot_data .with(|bot_data| {
.reader("data.bin.brotli") let heat_script_option = bot_data.has_heat_script().then(|| {
.await bot_data
.expect("TODO") .get_heat_script()
.into_futures_async_read(..) .expect("TODO")
.await; .to_string()
.expect("TODO")
let mut buf = Vec::default(); });
let mut message = capnp::message::TypedBuilder::<bot_capnp::bot::Owned>::new_default(); heat_script_option.map_or("none set yet".into(), |heat_script| {
let fallback = message.init_root(); format!("```\n{heat_script}\n```")
})
let message_reader;
let mut bot_data = fallback.into_reader();
match compressed_result {
Ok(compressed) => {
let mut decompressed = BrotliDecoder::new(compressed);
decompressed.read_to_end(&mut buf).await.expect("TODO");
message_reader =
capnp::serialize_packed::read_message(&*buf, ReaderOptions::default())
.expect("TODO");
bot_data = message_reader
.get_root::<bot_capnp::bot::Reader>()
.expect("TODO");
}
Err(error) if error.kind() == ErrorKind::NotFound => {
tracing::error!("TODO: proceeding with fallback");
}
Err(error) => {
tracing::error!(?error, "TODO");
return;
}
}
let heat_script_option = bot_data
.has_heat_script()
.then(|| bot_data.get_heat_script().expect("TODO"));
let heat_script_option =
heat_script_option.map(|heat_script| heat_script.to_str().expect("TODO"));
heat_script_option.map_or("none set yet".into(), |heat_script| {
format!("```\n{heat_script}\n```")
}) })
}; .await
.expect("TODO");
state state
.discord_client .discord_client
@@ -175,4 +138,48 @@ pub async fn handle(state: State, interaction: Interaction) {
.flags(MessageFlags::EPHEMERAL) .flags(MessageFlags::EPHEMERAL)
.await .await
.expect("TODO"); .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");
}
} }

View File

@@ -1,8 +1,12 @@
use crate::{OneToManyUniqueBTreeMap, VCs, command::State}; 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 async_trait::async_trait;
use futures::FutureExt;
use hound::{SampleFormat, WavSpec}; use hound::{SampleFormat, WavSpec};
use opendal::Operator; use opendal::Operator;
use snafu::{OptionExt, Snafu}; use snafu::{OptionExt as _, Snafu};
use songbird::{CoreEvent, Event, EventContext, EventHandler}; use songbird::{CoreEvent, Event, EventContext, EventHandler};
use std::{ use std::{
io::Cursor, io::Cursor,
@@ -99,25 +103,25 @@ struct Handler {
guild_id: Id<GuildMarker>, guild_id: Id<GuildMarker>,
channel_id: Id<ChannelMarker>, channel_id: Id<ChannelMarker>,
vcs: Arc<VCs>,
known_ssrcs: Arc<Mutex<OneToManyUniqueBTreeMap<Id<UserMarker>, u32>>>, known_ssrcs: Arc<Mutex<OneToManyUniqueBTreeMap<Id<UserMarker>, u32>>>,
audio_channels: u16,
audio_sample_rate: u32,
user_data_manager: UserDataManager,
} }
#[async_trait] #[async_trait]
impl EventHandler for Handler { impl EventHandler for Handler {
async fn act(&self, ctx: &EventContext<'_>) -> Option<Event> { async fn act(&self, ctx: &EventContext<'_>) -> Option<Event> {
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");
match ctx { match ctx {
EventContext::Track(_items) => { EventContext::Track(_items) => {
// Not expected to fire // Not expected to fire
} }
EventContext::SpeakingStateUpdate(speaking) => { EventContext::SpeakingStateUpdate(speaking) => {
tracing::error!(?speaking);
if let Some(user_id) = speaking.user_id { if let Some(user_id) = speaking.user_id {
let user_id = Id::new(user_id.0); let user_id = Id::new(user_id.0);
@@ -128,10 +132,40 @@ impl EventHandler for Handler {
} }
} }
EventContext::VoiceTick(voice_tick) => { EventContext::VoiceTick(voice_tick) => {
tracing::error!(?voice_tick);
for (ssrc, voice_data) in &voice_tick.speaking { for (ssrc, voice_data) in &voice_tick.speaking {
let user_id = self.known_ssrcs.lock().unwrap().get_left_for(ssrc).cloned(); 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 { 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 year = now_utc.year();
let month = now_utc.month(); let month = now_utc.month();
let day = now_utc.day(); let day = now_utc.day();
@@ -145,15 +179,20 @@ impl EventHandler for Handler {
let guild_id = self.guild_id; let guild_id = self.guild_id;
let channel_id = self.channel_id; let channel_id = self.channel_id;
let user = user_id.map_or_else(|| "Unknown".into(), ToString::to_string); let user = user_id
.as_ref()
.map_or_else(|| "UNKNOWN".into(), ToString::to_string);
let path = format!( let path = format!(
"{year}/{month}/{day}/{hour}/{minute}/audio-{second}.{microseconds}-{guild_id}-{channel_id}-{user}.wav" "{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 { let wav_spec = WavSpec {
channels: 2, channels,
sample_rate: 48000, sample_rate,
bits_per_sample: 16, bits_per_sample: 16,
sample_format: SampleFormat::Int, sample_format: SampleFormat::Int,
}; };
@@ -254,27 +293,45 @@ pub async fn handle(state: State, interaction: Interaction) {
let start_instant = Instant::now(); let start_instant = Instant::now();
let start_utc = UtcDateTime::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 { let handler = Handler {
start_instant, start_instant,
start_utc, start_utc,
recordings: state.recording_data, recordings: state.recording_data,
guild_id, guild_id,
channel_id: voice_channel_id, channel_id: voice_channel_id,
vcs,
known_ssrcs: Default::default(), known_ssrcs: Default::default(),
audio_channels,
audio_sample_rate,
user_data_manager: state.user_data_manager,
}; };
{ {
let call = call.lock().await; let mut call = call.lock().await;
call.add_global_event(CoreEvent::SpeakingStateUpdate.into(), handler); call.add_global_event(CoreEvent::SpeakingStateUpdate.into(), handler.clone());
call.add_global_event(CoreEvent::VoiceTick.into(), handler); call.add_global_event(CoreEvent::VoiceTick.into(), handler);
call.mute(true).await.expect("TODO");
} }
let channel_mention = format!("<#{voice_channel_id}>"); let channel_mention = format!("<#{voice_channel_id}>");
let bot_owner_mention = format!("<@{}>", state.discord_bot_owner_user_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 state
.discord_client .discord_client
.interaction(state.discord_application_id) .interaction(state.discord_application_id)
@@ -283,7 +340,7 @@ pub async fn handle(state: State, interaction: Interaction) {
).embeds(Some(&[ ).embeds(Some(&[
EmbedBuilder::new() EmbedBuilder::new()
.title("Joined VC to record") .title("Joined VC to record")
.description(format!("This bot joined {channel_mention} and intends to record. Here are some pledges backed by faith (because there is no way to verify them yourself) in {bot_owner_mention}:")) .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( .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() 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()
) )
@@ -291,7 +348,7 @@ pub async fn handle(state: State, interaction: Interaction) {
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() 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( .field(
EmbedFieldBuilder::new("Code is publically 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() 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( .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.") 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.")

View File

@@ -3,35 +3,45 @@ use std::{fmt::Debug, sync::Arc};
use futures::future::BoxFuture; use futures::future::BoxFuture;
use opendal::Operator; use opendal::Operator;
use patricia_tree::StringPatriciaMap; use patricia_tree::StringPatriciaMap;
use songbird::Songbird; use songbird::{
Songbird,
driver::{Channels, SampleRate},
};
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
use twilight_model::{ use twilight_model::{
application::{command::Command, interaction::Interaction}, application::{command::Command, interaction::Interaction},
id::{ id::{
Id, Id,
marker::{ApplicationMarker, UserMarker}, marker::{ApplicationMarker, CommandMarker, UserMarker},
}, },
}; };
use crate::VCs; use crate::{BotDataManager, GuildVoiceChannelToTextChannel, UserDataManager, VCs};
mod debug; pub mod info;
mod join; pub mod join;
mod leave; pub mod leave;
mod opt_in; pub mod opt_in;
mod opt_out; pub mod opt_out;
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct State { pub struct State {
pub bot_data: Operator, pub audio_channels: Channels,
pub audio_sample_rate: SampleRate,
pub bot_data_manager: BotDataManager,
pub cancellation_token: CancellationToken, pub cancellation_token: CancellationToken,
pub discord_application_id: Id<ApplicationMarker>, pub discord_application_id: Id<ApplicationMarker>,
pub discord_bot_owner_user_id: Id<UserMarker>, pub discord_bot_owner_user_id: Id<UserMarker>,
pub discord_client: Arc<twilight_http::Client>, 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_user_id: Id<UserMarker>,
pub discord_voice_channel_corresponding_text_channel: Arc<GuildVoiceChannelToTextChannel>,
pub recording_data: Operator, pub recording_data: Operator,
pub songbird: Arc<Songbird>, pub songbird: Arc<Songbird>,
pub user_data: Operator, pub user_data_manager: UserDataManager,
pub vcs: Arc<VCs>, pub vcs: Arc<VCs>,
} }
@@ -48,7 +58,7 @@ where
pub fn all() -> Vec<(&'static Command, BoxedHandler)> { pub fn all() -> Vec<(&'static Command, BoxedHandler)> {
vec![ vec![
(&debug::COMMAND, box_handler(debug::handle)), (&info::COMMAND, box_handler(info::handle)),
(&join::COMMAND, box_handler(join::handle)), (&join::COMMAND, box_handler(join::handle)),
(&leave::COMMAND, box_handler(leave::handle)), (&leave::COMMAND, box_handler(leave::handle)),
(&opt_in::COMMAND, box_handler(opt_in::handle)), (&opt_in::COMMAND, box_handler(opt_in::handle)),

View File

@@ -1,12 +1,16 @@
use std::sync::LazyLock; use std::sync::LazyLock;
use twilight_model::application::{ use twilight_model::{
command::{Command, CommandType}, application::{
interaction::Interaction, command::{Command, CommandType},
interaction::Interaction,
},
channel::message::MessageFlags,
http::interaction::{InteractionResponse, InteractionResponseType},
}; };
use twilight_util::builder::command::CommandBuilder; use twilight_util::builder::{InteractionResponseDataBuilder, command::CommandBuilder};
use crate::command::State; use crate::{command::State, user_capnp::user::Consent};
const NAME: &str = "opt-in"; const NAME: &str = "opt-in";
const DESCRIPTION: &str = "Opt in to being recorded"; const DESCRIPTION: &str = "Opt in to being recorded";
@@ -20,5 +24,67 @@ pub static COMMAND: LazyLock<Command> = LazyLock::new(|| {
#[tracing::instrument] #[tracing::instrument]
pub async fn handle(state: State, interaction: Interaction) { pub async fn handle(state: State, interaction: Interaction) {
todo!(); 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");
} }

View File

@@ -1,12 +1,16 @@
use std::sync::LazyLock; use std::sync::LazyLock;
use twilight_model::application::{ use twilight_model::{
command::{Command, CommandType}, application::{
interaction::Interaction, command::{Command, CommandType},
interaction::Interaction,
},
channel::message::MessageFlags,
http::interaction::{InteractionResponse, InteractionResponseType},
}; };
use twilight_util::builder::command::CommandBuilder; use twilight_util::builder::{InteractionResponseDataBuilder, command::CommandBuilder};
use crate::command::State; use crate::{command::State, user_capnp::user::Consent};
const NAME: &str = "opt-out"; const NAME: &str = "opt-out";
const DESCRIPTION: &str = "Opt out of being recorded"; const DESCRIPTION: &str = "Opt out of being recorded";
@@ -20,5 +24,67 @@ pub static COMMAND: LazyLock<Command> = LazyLock::new(|| {
#[tracing::instrument] #[tracing::instrument]
pub async fn handle(state: State, interaction: Interaction) { pub async fn handle(state: State, interaction: Interaction) {
todo!(); 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");
} }

View File

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

View File

@@ -1,9 +1,17 @@
use clap::Parser; use clap::Parser;
use fomo_reducer::{CommandRouter, State, Storage, all_commands, initialize_vcs, update_vcs}; use fomo_reducer::{
BotDataManager, CommandRouter, GuildVoiceChannelToTextChannel, State, Storage, UserDataManager,
all_commands, command, initialize_vcs, update_vcs,
};
use secrecy::{ExposeSecret, SecretString}; use secrecy::{ExposeSecret, SecretString};
use snafu::Snafu; use snafu::{OptionExt, ResultExt, Snafu};
use songbird::{Songbird, shards::TwilightMap}; use songbird::{
use std::{fmt::Debug, sync::Arc}; 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::{select, signal::ctrl_c, task::JoinSet};
use tokio_util::{sync::CancellationToken, time::FutureExt as _}; use tokio_util::{sync::CancellationToken, time::FutureExt as _};
use tracing_subscriber::{EnvFilter, fmt::format::FmtSpan}; use tracing_subscriber::{EnvFilter, fmt::format::FmtSpan};
@@ -14,9 +22,88 @@ use twilight_model::{
payload::{incoming::InteractionCreate, outgoing::UpdatePresence}, payload::{incoming::InteractionCreate, outgoing::UpdatePresence},
presence::{ActivityType, MinimalActivity, Status}, presence::{ActivityType, MinimalActivity, Status},
}, },
id::{Id, marker::UserMarker}, id::{
Id,
marker::{ChannelMarker, GuildMarker, UserMarker},
},
}; };
#[derive(Clone, Copy, Debug, strum::Display, EnumString)]
enum AudioChannels {
Mono,
Stereo,
}
impl From<AudioChannels> for Channels {
fn from(value: AudioChannels) -> Self {
match value {
AudioChannels::Mono => Channels::Mono,
AudioChannels::Stereo => Channels::Stereo,
}
}
}
#[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,
}
}
}
#[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,
},
}
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)] #[derive(Debug, Parser)]
struct AppArgs { struct AppArgs {
#[arg(long, env)] #[arg(long, env)]
@@ -31,6 +118,16 @@ struct AppArgs {
#[arg(long, env)] #[arg(long, env)]
discord_status: Option<Arc<str>>, 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)] #[arg(long, env)]
bot_data: Storage, bot_data: Storage,
@@ -89,6 +186,9 @@ async fn main() -> Result<(), MainError> {
discord_bot_owner_user_id, discord_bot_owner_user_id,
discord_nickname, discord_nickname,
discord_status, discord_status,
discord_voice_channel_corresponding_text_channel,
audio_channels,
audio_sample_rate,
bot_data, bot_data,
user_data, user_data,
recording_data, recording_data,
@@ -159,15 +259,23 @@ async fn main() -> Result<(), MainError> {
.collect(), .collect(),
); );
let senders = Arc::new(senders); 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); 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 interaction_client = discord_client.interaction(discord_application_id);
let commands = all_commands(); let commands = all_commands();
let _returned_commands = interaction_client let returned_commands = interaction_client
.set_global_commands( .set_global_commands(
Vec::from_iter( Vec::from_iter(
commands commands
@@ -182,11 +290,30 @@ async fn main() -> Result<(), MainError> {
.await .await
.expect("failed to deserialize set commands"); // TODO .expect("failed to deserialize set commands"); // TODO
let command_router = CommandRouter::from_iter(commands); let mut discord_command_name_to_returned_command = BTreeMap::from_iter(
let command_router = Arc::new(command_router); 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 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 discord_client = Arc::new(discord_client);
let songbird = Arc::new(songbird); let songbird = Arc::new(songbird);
let vcs = Arc::new(vcs); let vcs = Arc::new(vcs);
@@ -195,16 +322,42 @@ async fn main() -> Result<(), MainError> {
let recording_data = recording_data.into_inner(); let recording_data = recording_data.into_inner();
let user_data = user_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 { let state = State {
bot_data, audio_channels,
audio_sample_rate,
bot_data_manager,
cancellation_token: cancellation_token.clone(), cancellation_token: cancellation_token.clone(),
discord_application_id, discord_application_id,
discord_bot_owner_user_id, discord_bot_owner_user_id,
discord_client, 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_user_id,
discord_voice_channel_corresponding_text_channel,
recording_data, recording_data,
songbird, songbird,
user_data, user_data_manager,
vcs, vcs,
}; };
@@ -273,9 +426,22 @@ async fn handle_events(command_router: Arc<CommandRouter>, state: State, mut sha
.await .await
{ {
match event_res { match event_res {
Ok(twilight_model::gateway::event::Event::GatewayClose(frame_option)) => {
tracing::warn!(?frame_option);
return;
}
Ok(event) => { Ok(event) => {
handle_event(command_router.clone(), state.clone(), event).await; 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) => { Err(error) => {
tracing::error!(?error); tracing::error!(?error);
} }

25
src/operator_ext.rs Normal file
View File

@@ -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),
}
}
}

11
src/option_ext.rs Normal file
View File

@@ -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,
}
}
}

View File

@@ -2,7 +2,6 @@ use std::{fmt::Debug, str::FromStr};
use opendal::{IntoOperatorUri, Operator, OperatorUri}; use opendal::{IntoOperatorUri, Operator, OperatorUri};
#[derive(Clone)] #[derive(Clone)]
pub struct Storage { pub struct Storage {
uri: OperatorUri, uri: OperatorUri,

View File

@@ -1,3 +1,5 @@
use std::collections::BTreeMap;
use dashmap::DashMap; use dashmap::DashMap;
use futures::{StreamExt, stream::FuturesUnordered}; use futures::{StreamExt, stream::FuturesUnordered};
use twilight_model::{ use twilight_model::{
@@ -8,10 +10,12 @@ use twilight_model::{
}, },
}; };
use crate::{OneToManyUniqueBTreeMapWithData, UserInVCData, VoiceStatus}; 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>; type VCsInGuild = OneToManyUniqueBTreeMapWithData<Id<ChannelMarker>, Id<UserMarker>, UserInVCData>;
pub type VCs = DashMap<Id<GuildMarker>, VCsInGuild>; pub type VCs = DashMap<Id<GuildMarker>, VCsInGuild>;
#[tracing::instrument(skip(discord_client), ret)] #[tracing::instrument(skip(discord_client), ret)]

254
src/user_data.rs Normal file
View File

@@ -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)
}
}

View File

@@ -3,5 +3,11 @@
struct User { struct User {
notificationScript @0 :Text; notificationScript @0 :Text;
voiceRecordingConsent @1 :Bool; voiceRecordingConsent @1 :Consent = unspecified;
enum Consent {
unspecified @0;
granted @1;
withheld @2;
}
} }