char/phoebe

git clone https://git.t4t.associates/char/phoebe

videogame hackerAdd message editing support & impl for Discord4fc8c15

main
brancheslegacymainall refs →
4.5 KiB137 linesraw
1use color_eyre::Result;
2use futures::{future, StreamExt};
3use tracing::info;
4use tracing_subscriber::EnvFilter;
5
6use phoebe::{
7    get_linked_channels, get_linked_messages, link_messages,
8    prelude::{ChatEvent, SqlitePool},
9    service::Service,
10    unlink_message, DynServiceLookup,
11};
12
13async fn handle_events(
14    dyn_service: DynServiceLookup,
15    db: SqlitePool,
16    mut service: Box<dyn Service + Send + Sync>,
17    mut rx: tokio::sync::broadcast::Receiver<ChatEvent>,
18) {
19    info!("Handling events for {}…", service.tag());
20
21    let mut conn = db
22        .acquire()
23        .await
24        .expect("Failed to acquire core DB connection");
25
26    while let Ok(event) = rx.recv().await {
27        match event {
28            ChatEvent::NewMessage(message) => {
29                let linked_channels =
30                    get_linked_channels(&mut conn, dyn_service, &message.origin.channel).await;
31
32                let mut resulting_messages = vec![];
33                for destination_channel in linked_channels {
34                    resulting_messages.extend(
35                        service
36                            .send_chat_message(&message, destination_channel)
37                            .await,
38                    )
39                }
40
41                if !resulting_messages.is_empty() {
42                    if let Err(e) =
43                        link_messages(&mut conn, &message.origin, &resulting_messages).await
44                    {
45                        tracing::error!("Failed to link messages: {e}");
46                    }
47                }
48            }
49
50            ChatEvent::DeleteMessage(origin) => {
51                let messages = if let Ok(message_stream) =
52                    get_linked_messages(&mut conn, dyn_service, &origin).await
53                {
54                    message_stream
55                        .filter(|r| future::ready(r.channel.service == service.tag()))
56                        .collect::<Vec<_>>()
57                        .await
58                } else {
59                    vec![]
60                };
61
62                if !messages.is_empty() {
63                    if let Err(e) = unlink_message(&mut conn, &origin).await {
64                        tracing::error!("Failed to unlink origin message: {e}");
65                    }
66
67                    for message in messages {
68                        if service.delete_message(&message).await {
69                            if let Err(e) = unlink_message(&mut conn, &message).await {
70                                tracing::error!("Failed to unlink related message: {e}");
71                            }
72                        }
73                    }
74                }
75            }
76            ChatEvent::EditMessage(prev_origin, message_edit) => {
77                let messages = if let Ok(message_stream) =
78                    get_linked_messages(&mut conn, dyn_service, &prev_origin).await
79                {
80                    message_stream
81                        .filter(|r| future::ready(r.channel.service == service.tag()))
82                        .collect::<Vec<_>>()
83                        .await
84                } else {
85                    vec![]
86                };
87
88                let mut resulting_messages = vec![];
89                for message in messages {
90                    resulting_messages.extend(service.edit_message(&message, &message_edit).await)
91                }
92
93                if !resulting_messages.is_empty() {
94                    if let Err(e) =
95                        link_messages(&mut conn, &prev_origin, &resulting_messages).await
96                    {
97                        tracing::error!("Failed to link messages: {e}");
98                    }
99                }
100            }
101        }
102    }
103}
104
105#[tokio::main]
106async fn main() -> Result<()> {
107    color_eyre::install()?;
108
109    tracing_subscriber::fmt()
110        .with_target(true)
111        .with_env_filter(EnvFilter::from_default_env())
112        .init();
113
114    let (tx, _) = tokio::sync::broadcast::channel(512);
115
116    let db = phoebe::open_core_db().await?;
117
118    fn dyn_service(service: &str) -> &'static str {
119        match service {
120            "discord" => "discord",
121            "matrix" => "matrix",
122            _ => panic!("Unsupported service: {}", service),
123        }
124    }
125
126    let services: Vec<Box<dyn Service + Send + Sync>> = vec![Box::new(
127        phoebe_discord::setup(db.clone(), tx.clone(), dyn_service).await?,
128    )];
129
130    let handles = services
131        .into_iter()
132        .map(|srv| tokio::spawn(handle_events(dyn_service, db.clone(), srv, tx.subscribe())));
133
134    let _ = future::join_all(handles).await;
135
136    Ok(())
137}