src/bot_manager.rs

@@ -17,18 +17,15 @@ use crate::{
     BotController, BotResult, BotRuntime,
 };
 
-pub type BotThread = JoinHandle<BotResult<()>>;
-
 pub struct BotRunner {
     controller: BotController,
     info: BotInfo,
     notificator: NotificatorThread,
-    thread: Option<BotThread>,
+    thread: Option<JoinHandle<BotResult<()>>>,
 }
 
-#[derive(Debug)]
 pub enum NotificatorThread {
-    Running(Option<BotThread>),
+    Running(Option<JoinHandle<BotResult<()>>>),
     Done,
 }
 
@@ -75,7 +72,7 @@ where
 
     pub async fn dispatch(mut self, db: &mut DB) -> BotResult<()> {
         loop {
-            for bi in (self.bi_getter)().await {
+            'biter: for bi in (self.bi_getter)().await {
                 // removing handler to force restart
                 // TODO: wait till all updates are processed in bot
                 // Temporarly disabling code, because it's free of js runtime
@@ -91,12 +88,26 @@ where
                 let mut bot_runner = match self.bot_pool.remove(&bi.name) {
                     Some(br) => br,
                     None => {
+                        let handlers = (self.h_mapper)(bi.clone()).await;
                         info!("NEW INSTANCE: Starting new instance! bot name: {}", bi.name);
-                        self.create_bot_runner(&bi, db).await?
+                        self.start_bot(bi, db, handlers.collect()).await?;
+                        continue 'biter;
                     }
                 };
 
-                bot_runner.thread = clear_finished_thread(bot_runner.thread, &bi);
+                // checking if thread is not finished, otherwise clearing handler
+                bot_runner.thread = match bot_runner.thread {
+                    Some(thread) => {
+                        if thread.is_finished() {
+                            let err = thread.join();
+                            error!("Thread bot `{}` finished with error: {:?}", bi.name, err);
+                            None
+                        } else {
+                            Some(thread)
+                        }
+                    }
+                    None => None,
+                };
 
                 // checking if thread is running, otherwise start thread
                 bot_runner.thread = match bot_runner.thread {
@@ -118,84 +129,41 @@ where
                         )
                     }
                 };
-
-                bot_runner.notificator = check_notificator_done(bot_runner.notificator);
-
-                bot_runner.notificator = match bot_runner.notificator {
-                    NotificatorThread::Done => NotificatorThread::Done,
-                    NotificatorThread::Running(thread) => {
-                        NotificatorThread::Running(match thread {
-                            Some(thread) => Some(thread),
-                            None => {
-                                let thread =
-                                    spawn_notificator_thread(bot_runner.controller.clone()).await?;
-                                Some(thread)
-                            }
-                        })
-                    }
-                };
-
                 self.bot_pool.insert(bi.name.clone(), bot_runner);
             }
             tokio::time::sleep(Duration::from_secs(1)).await;
         }
     }
 
-    pub async fn create_bot_runner(
+    pub async fn start_bot(
         &mut self,
-        bi: &BotInstance,
+        bi: BotInstance,
         db: &mut DB,
-    ) -> BotResult<BotRunner> {
+        plug_handlers: Vec<BotHandler>,
+    ) -> BotResult<BotInfo> {
         let db = db.clone().with_name(bi.name.clone());
         let controller = BotController::with_db(db.clone(), &bi.token, &bi.script).await?;
 
+        let handler = script_handler_gen(controller.runtime.clone(), plug_handlers).await;
+
+        let thread =
+            spawn_bot_thread(controller.bot.clone(), controller.db.clone(), handler).await?;
+        let notificator = spawn_notificator_thread(controller.clone()).await?;
+        let notificator = NotificatorThread::Running(Some(notificator));
+
         let info = BotInfo {
             name: bi.name.clone(),
         };
         let runner = BotRunner {
             controller,
-            info,
+            info: info.clone(),
-            notificator: NotificatorThread::Running(None),
+            notificator,
-            thread: None,
+            thread: Some(thread),
         };
 
-        Ok(runner)
+        self.bot_pool.insert(bi.name.clone(), runner);
-    }
-}
 
-/// checking if thread is not finished, otherwise clearing handler
+        Ok(info)
-fn clear_finished_thread(thread: Option<BotThread>, bi: &BotInstance) -> Option<BotThread> {
-    thread.and_then(|thread| match thread.is_finished() {
-        false => Some(thread),
-        // if finished, join it (should return immidiatly), and print cause of stop
-        true => {
-            let err = thread.join();
-            error!("Thread bot `{}` finished with error: {:?}", bi.name, err);
-            None
-        }
-    })
-}
-
-// sets NotificatorThread to Done if running thread returned Ok(...)
-fn check_notificator_done(n: NotificatorThread) -> NotificatorThread {
-    match n {
-        NotificatorThread::Running(Some(thread)) if thread.is_finished() => {
-            match thread.join() {
-                // if thread returns Ok(_), then do not run it again
-                Ok(result) if result.is_ok() => NotificatorThread::Done,
-
-                // but try to restart, if returned an error
-                Ok(result) => {
-                    error!("Notificator thread returned error: {result:?}");
-                    NotificatorThread::Running(None)
-                }
-                Err(panicerr) => {
-                    error!("Notificator thread paniced: {panicerr:?}");
-                    NotificatorThread::Running(None)
-                }
-            }
-        }
-        other => other,
     }
 }
 
@@ -213,13 +181,20 @@ async fn script_handler_gen(
     handler
 }
 
-pub async fn spawn_bot_thread(bot: Bot, mut db: DB, handler: BotHandler) -> BotResult<BotThread> {
+pub async fn spawn_bot_thread(
+    bot: Bot,
+    mut db: DB,
+    handler: BotHandler,
+) -> BotResult<JoinHandle<BotResult<()>>> {
     let state_mgr = MongodbStorage::from_db(&mut db, Json)
         .await
         .map_err(DbError::from)?;
     let thread = std::thread::spawn(move || -> BotResult<()> {
         let state_mgr = state_mgr;
 
+        // let rt = tokio::runtime::Builder::new_current_thread()
+        //     .enable_all()
+        //     .build()?;
         let rt = tokio::runtime::Runtime::new()?;
 
         rt.block_on(
@@ -235,7 +210,9 @@ pub async fn spawn_bot_thread(bot: Bot, mut db: DB, handler: BotHandler) -> BotR
     Ok(thread)
 }
 
-pub async fn spawn_notificator_thread(mut c: BotController) -> BotResult<BotThread> {
+pub async fn spawn_notificator_thread(
+    mut c: BotController,
+) -> BotResult<JoinHandle<BotResult<()>>> {
     let thread = std::thread::spawn(move || -> BotResult<()> {
         let rt = tokio::runtime::Runtime::new()?;
 

src/botscript.rs

@@ -2,14 +2,14 @@ pub mod application;
 pub mod db;
 pub mod message_info;
 use std::collections::HashMap;
-use std::sync::{Mutex, PoisonError};
+use std::sync::Mutex;
 use std::time::Duration;
 
 use crate::db::raw_calls::RawCallError;
 use crate::db::{CallDB, DbError, User, DB};
 use crate::message_answerer::MessageAnswererError;
+use crate::notify_admin;
 use crate::utils::parcelable::{ParcelType, Parcelable, ParcelableError, ParcelableResult};
-use crate::{notify_admin, BotError};
 use chrono::{DateTime, Days, NaiveTime, ParseError, TimeDelta, Timelike, Utc};
 use db::attach_db_obj;
 use futures::future::join_all;
@@ -48,23 +48,6 @@ pub enum ScriptError {
     MutexError(String),
     #[error("can't send message to user to user: {0:?}")]
     MAError(#[from] MessageAnswererError),
-    #[error("other script error: {0:?}")]
-    Other(String),
-}
-
-impl From<BotError> for ScriptError {
-    fn from(value: BotError) -> Self {
-        match value {
-            crate::BotError::DBError(db_error) => ScriptError::DBError(db_error),
-            error => ScriptError::Other(format!("BotError: {error}")),
-        }
-    }
-}
-
-impl<T> From<PoisonError<T>> for ScriptError {
-    fn from(value: PoisonError<T>) -> Self {
-        Self::MutexError(format!("Can't lock Mutex in script, err: {}", value))
-    }
 }
 
 #[derive(thiserror::Error, Debug)]
@@ -688,7 +671,8 @@ pub enum NotificationTime {
 }
 
 impl NotificationTime {
-    pub fn when_next(&self, start_time: DateTime<Utc>, now: DateTime<Utc>) -> DateTime<Utc> {
+    pub fn when_next(&self, start_time: &DateTime<Utc>, now: &DateTime<Utc>) -> DateTime<Utc> {
+        let now = *now;
         match self {
             NotificationTime::Delta {
                 delta_hours,
@@ -696,15 +680,15 @@ impl NotificationTime {
             } => {
                 let delta = TimeDelta::minutes((delta_minutes + delta_hours * 60).into());
 
-                let secs_period = delta.num_seconds();
+                let mut estimation = *start_time;
-                if secs_period == 0 {
+                // super non-optimal, but fun :)
-                    return now;
+                loop {
-                };
+                    if estimation < now + Duration::from_secs(1) {
-
+                        estimation += delta;
-                let diff = now - start_time;
+                    } else {
-                let passed = diff.num_seconds().abs() % secs_period;
+                        break estimation;
-
+                    }
-                now - Duration::from_secs(passed as u64) + delta
+                }
             }
             NotificationTime::Specific(time) => {
                 let estimation = now;
@@ -712,11 +696,13 @@ impl NotificationTime {
                 let mut estimation = estimation
                     .with_minute(time.minutes.into())
                     .unwrap_or(estimation);
-
+                // super non-optimal, but fun :)
+                loop {
                     if estimation < now {
-                    estimation + Days::new(1)
+                        estimation = estimation + Days::new(1);
                     } else {
-                    estimation
+                        break estimation;
+                    }
                 }
             }
         }
@@ -875,11 +861,11 @@ impl Parcelable<BotFunction> for BotNotification {
 }
 
 impl BotNotification {
-    pub fn left_time(&self, start_time: DateTime<Utc>, now: DateTime<Utc>) -> Duration {
+    pub fn left_time(&self, start_time: &DateTime<Utc>, now: &DateTime<Utc>) -> Duration {
         let next = self.time.when_next(start_time, now);
 
         // immidate notification if time to do it passed
-        let duration = (next - now).to_std().unwrap_or(Duration::from_secs(0));
+        let duration = (next - now).to_std().unwrap_or(Duration::from_secs(1));
 
         // Rounding partitions of seconds
         Duration::from_secs(duration.as_secs())
@@ -986,19 +972,19 @@ impl RunnerConfig {
         let ordered = self
             .notifications
             .iter()
-            .filter(|f| f.left_time(start_time, now) > Duration::from_secs(1))
+            .filter(|f| f.left_time(&start_time, &now) > Duration::from_secs(1))
-            .sorted_by_key(|f| f.left_time(start_time, now))
+            .sorted_by_key(|f| f.left_time(&start_time, &now))
             .collect::<Vec<_>>();
 
         let left = match ordered.first() {
-            Some(notification) => notification.left_time(start_time, now),
+            Some(notification) => notification.left_time(&start_time, &now),
             // No notifications provided
             None => return None,
         };
         // get all that should be sent at the same time
         let notifications = ordered
             .into_iter()
-            .filter(|n| n.left_time(start_time, now) == left)
+            .filter(|n| n.left_time(&start_time, &now) == left)
             .cloned()
             .collect::<Vec<_>>();
 
@@ -1186,7 +1172,7 @@ mod tests {
         let start_time = chrono::offset::Utc::now();
         // let start_time = chrono::offset::Utc::now() + TimeDelta::try_hours(5).unwrap();
         let start_time = start_time.with_hour(13).unwrap().with_minute(23).unwrap();
-        let left = n.left_time(start_time, start_time);
+        let left = n.left_time(&start_time, &start_time);
         let secs = left.as_secs();
         let minutes = secs / 60;
         let hours = minutes / 60;
@@ -1204,37 +1190,4 @@ mod tests {
 
         assert_eq!(left, should_left)
     }
-
-    #[test]
-    fn test_notification_time_nextday() {
-        let botn = json!({
-            "time": "11:00",
-            "filter": {"random": 2},
-            "message": {"text": "some"},
-        });
-        let n: BotNotification = serde_json::from_value(botn).unwrap();
-        println!("BotNotification: {n:#?}");
-        let start_time = chrono::offset::Utc::now();
-        // let start_time = chrono::offset::Utc::now() + TimeDelta::try_hours(5).unwrap();
-        let start_time = start_time.with_hour(13).unwrap().with_minute(23).unwrap();
-        let left = n.left_time(start_time, start_time);
-        let secs = left.as_secs();
-        let minutes = secs / 60;
-        let hours = minutes / 60;
-        let minutes = minutes % 60;
-        println!("Left: {hours}:{minutes}");
-
-        let when_should = chrono::offset::Utc::now()
-            .with_hour(11)
-            .unwrap()
-            .with_minute(00)
-            .unwrap();
-
-        let should_left = (when_should + TimeDelta::days(1) - start_time)
-            .to_std()
-            .unwrap();
-        let should_left = Duration::from_secs(should_left.as_secs());
-
-        assert_eq!(left, should_left)
-    }
 }

src/botscript/application.rs

@@ -1,3 +1,8 @@
+// just keeping track locks in asychronous calls
+#![allow(clippy::await_holding_lock)]
+use std::sync::RwLock;
+
+use log::info;
 use quickjs_rusty::{context::Context, serde::from_js, OwnedJsObject};
 use teloxide::Bot;
 use tokio::runtime::Handle;
@@ -13,41 +18,58 @@ use super::ScriptError;
 pub fn attach_user_application(
     c: &Context,
     o: &mut OwnedJsObject,
-    db: DB,
+    db: &DB,
-    bot: Bot,
+    bot: &Bot,
 ) -> Result<(), ScriptError> {
-    // To guarantee that closure is valid if thread panics
+    let db: std::sync::Arc<RwLock<DB>> = std::sync::Arc::new(RwLock::new(db.clone()));
-    let db: std::sync::Mutex<DB> = std::sync::Mutex::new(db);
+
-    let bot: std::sync::Mutex<Bot> = std::sync::Mutex::new(bot);
+    let bot: std::sync::Arc<RwLock<Bot>> = std::sync::Arc::new(RwLock::new(bot.clone()));
 
     let user_application =
         c.create_callback(move |q: OwnedJsObject| -> Result<_, ScriptError> {
-            let mut db = { db.lock().map_err(ScriptError::from)?.clone() };
+            let bot1 = bot.clone();
-            let bot = { bot.lock().map_err(ScriptError::from)?.clone() };
+            let bot1 = bot1.read().expect("Can't read lock bot");
+            let bot2 = bot.read().expect("Can't read lock bot");
             let user: teloxide::types::User = match from_js(q.context(), &q) {
                 Ok(q) => q,
                 Err(_) => todo!(),
             };
 
-            let application =
+            let application = futures::executor::block_on(
-                futures::executor::block_on(Application::new(user.clone()).store_db(&mut db))?;
+                Application::new(user.clone())
+                    .store_db(&mut db.write().expect("Can't write lock db")),
+            )?;
 
-            let msg = tokio::task::block_in_place(|| {
+            let db2 = db.clone();
-                Handle::current()
+            let msg = tokio::task::block_in_place(move || {
-                    .block_on(async { send_application_to_chat(&bot, &mut db, &application).await })
+                Handle::current().block_on(async move {
-            });
+                    send_application_to_chat(
-            let msg = msg.map_err(ScriptError::from)?;
+                        &bot1,
-
+                        &mut db2.write().expect("Can't write lock db"),
-            let (chat_id, msg_id) = tokio::task::block_in_place(|| {
+                        &application,
-                Handle::current().block_on(async {
+                    )
-                    MessageAnswerer::new(&bot, &mut db, user.id.0 as i64)
-                        .answer("left_application_msg", None, None)
                     .await
                 })
-            })?;
+            });
+            let msg = match msg {
+                Ok(msg) => msg,
+                Err(err) => {
+                    info!("Got err: {err}");
+                    return Err(ScriptError::MutexError("🤦‍♂️".to_string()));
+                }
+            };
+
+            let (chat_id, msg_id) = futures::executor::block_on(
+                MessageAnswerer::new(
+                    &bot2,
+                    &mut db.write().expect("Can't write lock db"),
+                    user.id.0 as i64,
+                )
+                .answer("left_application_msg", None, None),
+            )?;
             futures::executor::block_on(
                 MessageForward::new(msg.chat.id.0, msg.id.0, chat_id, msg_id, false)
-                    .store_db(&mut db),
+                    .store_db(&mut db.write().expect("Can't write lock db")),
             )?;
 
             let ret = true;

src/db/tests/mod.rs

@@ -10,7 +10,7 @@ async fn setup_db() -> DB {
     dotenvy::dotenv().unwrap();
     let db_url = std::env::var("DATABASE_URL").unwrap();
 
-    DB::new(db_url, "tests".to_string()).await.unwrap()
+    DB::new(db_url, "gongbot".to_string()).await.unwrap()
 }
 
 #[tokio::test]
@@ -72,8 +72,6 @@ async fn test_add_media() {
 async fn test_drop_media() {
     let mut db = setup_db().await;
 
-    let _result = db.drop_media("test_drop_media_literal").await.unwrap();
-
     let _result = db
         .add_media("test_drop_media_literal", "photo", "file_id_1", None)
         .await
@@ -179,9 +177,7 @@ async fn test_drop_media_except() {
 
 #[tokio::test]
 async fn test_get_random_users() {
-    let mut db = setup_db().await;
+    let db = setup_db().await;
-
-    let _ = db.get_or_init_user(1, "Nick").await;
 
     let users = db.get_random_users(1).await.unwrap();
     assert_eq!(users.len(), 1);

src/main.rs

@@ -126,7 +126,7 @@ impl BotController {
         let bot = Bot::new(token);
 
         let mut runner = Runner::init_with_db(&mut db)?;
-        runner.call_attacher(|c, o| attach_user_application(c, o, db.clone(), bot.clone()))??;
+        runner.call_attacher(|c, o| attach_user_application(c, o, &db, &bot))??;
         let rc = runner.init_config(script)?;
         let runtime = Arc::new(Mutex::new(BotRuntime { rc, runner }));