call db migration on init

will contain indexes

src/db/mod.rs

@@ -3,12 +3,13 @@ use chrono::{DateTime, Utc};
 use enum_stringify::EnumStringify;
 use futures::stream::{StreamExt, TryStreamExt};
 
-use mongodb::Database;
+use mongodb::options::IndexOptions;
 use mongodb::{
     bson::doc,
     options::{ClientOptions, ResolverConfig},
     Client,
 };
+use mongodb::{Database, IndexModel};
 use serde::{Deserialize, Serialize};
 
 #[derive(EnumStringify)]
@@ -80,6 +81,19 @@ impl DB {
 
         DB { client }
     }
+
+    pub async fn migrate(&mut self) -> Result<(), mongodb::error::Error> {
+        let events = self.get_database().await.collection::<Event>("events");
+        events
+            .create_index(
+                IndexModel::builder()
+                    .keys(doc! {"time": 1})
+                    .options(IndexOptions::builder().unique(true).build())
+                    .build(),
+            )
+            .await?;
+        Ok(())
+    }
 }
 
 #[async_trait]
@@ -129,7 +143,10 @@ pub trait CallDB {
         users
             .update_one(
                 doc! { "id": userid },
-                doc! { "$set": { "first_name": firstname } },
+                doc! {
+                    "$set": doc! { "first_name": firstname},
+                    "$setOnInsert": doc! { "is_admin": false },
+                },
             )
             .upsert(true)
             .await
@@ -196,7 +213,7 @@ pub trait CallDB {
         literal: &str,
     ) -> Result<Option<Literal>, Box<dyn std::error::Error>> {
         let db = self.get_database().await;
-        let messages = db.collection::<Literal>("messages");
+        let messages = db.collection::<Literal>("literals");
 
         let literal = messages.find_one(doc! { "token": literal }).await?;
 

src/main.rs

@@ -1,18 +1,20 @@
 pub mod admin;
 pub mod db;
+pub mod mongodb_storage;
 
 use std::time::Duration;
 
 use crate::admin::{admin_command_handler, AdminCommands};
 use crate::admin::{secret_command_handler, SecretCommands};
 use crate::db::{CallDB, DB};
+use crate::mongodb_storage::MongodbStorage;
 
 use chrono::{DateTime, Utc};
 use chrono_tz::Asia;
 use envconfig::Envconfig;
 use serde::{Deserialize, Serialize};
 use teloxide::dispatching::dialogue::serializer::Json;
-use teloxide::dispatching::dialogue::{GetChatId, PostgresStorage};
+use teloxide::dispatching::dialogue::GetChatId;
 use teloxide::types::{
     InlineKeyboardButton, InlineKeyboardMarkup, InputFile, InputMedia, MediaKind, MessageKind,
     ParseMode, ReplyMarkup,
@@ -23,7 +25,7 @@ use teloxide::{
     utils::{command::BotCommands, render::RenderMessageTextHelper},
 };
 
-type BotDialogue = Dialogue<State, PostgresStorage<Json>>;
+type BotDialogue = Dialogue<State, MongodbStorage<Json>>;
 
 #[derive(Envconfig)]
 struct Config {
@@ -72,9 +74,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let config = Config::init_from_env()?;
 
     let bot = Bot::new(&config.bot_token);
-    let db = DB::new(&config.db_url).await;
+    let mut db = DB::new(&config.db_url).await;
+    db.migrate().await.unwrap();
     let db_url2 = config.db_url.clone();
-    let state_mgr = PostgresStorage::open(&db_url2, 8, Json).await?;
+    let state_mgr = MongodbStorage::open(&db_url2, "gongbot", Json).await?;
 
     // TODO: delete this in production
     let events: Vec<DateTime<Utc>> = vec!["2025-04-09T18:00:00+04:00", "2025-04-11T16:00:00+04:00"]
@@ -105,7 +108,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                         .await;
                     user.is_admin
                 })
-                .enter_dialogue::<Message, PostgresStorage<Json>, State>()
+                .enter_dialogue::<Message, MongodbStorage<Json>, State>()
                 .branch(
                     Update::filter_message()
                         .filter(|msg: Message| {

src/mongodb_storage.rs

@@ -0,0 +1,105 @@
+use std::{
+    fmt::{Debug, Display},
+    sync::Arc,
+};
+
+use futures::future::BoxFuture;
+use mongodb::bson::doc;
+use mongodb::Database;
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
+use teloxide::dispatching::dialogue::{Serializer, Storage};
+
+pub struct MongodbStorage<S> {
+    database: Database,
+    serializer: S,
+}
+
+impl<S> MongodbStorage<S> {
+    pub async fn open(
+        database_url: &str,
+        database_name: &str,
+        serializer: S,
+    ) -> Result<Arc<Self>, mongodb::error::Error> {
+        let client = mongodb::Client::with_uri_str(database_url).await?;
+        let database = client.database(database_name);
+
+        Ok(Arc::new(Self {
+            database,
+            serializer,
+        }))
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct Dialogue {
+    chat_id: i64,
+    dialogue: Vec<u32>,
+}
+
+impl<S, D> Storage<D> for MongodbStorage<S>
+where
+    S: Send + Sync + Serializer<D> + 'static,
+    D: Send + Serialize + DeserializeOwned + 'static,
+
+    <S as Serializer<D>>::Error: Debug + Display,
+{
+    type Error = mongodb::error::Error;
+
+    fn remove_dialogue(
+        self: std::sync::Arc<Self>,
+        chat_id: teloxide::prelude::ChatId,
+    ) -> BoxFuture<'static, Result<(), Self::Error>>
+    where
+        D: Send + 'static,
+    {
+        Box::pin(async move {
+            let d = self.database.collection::<Dialogue>("dialogues");
+            d.delete_one(doc! { "chat_id": chat_id.0 })
+                .await
+                .map(|_| ())
+        })
+    }
+
+    fn update_dialogue(
+        self: std::sync::Arc<Self>,
+        chat_id: teloxide::prelude::ChatId,
+        dialogue: D,
+    ) -> BoxFuture<'static, Result<(), Self::Error>>
+    where
+        D: Send + 'static,
+    {
+        Box::pin(async move {
+            let d = self.database.collection::<Dialogue>("dialogues");
+            d.update_one(
+                        doc! {
+                            "chat_id": chat_id.0
+                        },
+                        doc! {
+                            "$set": doc! {
+                                "dialogue": self.serializer.serialize(&dialogue).unwrap().into_iter().map(|v| v as u32).collect::<Vec<u32>>()
+                            }
+                    }).upsert(true).await?;
+            Ok(())
+        })
+    }
+
+    fn get_dialogue(
+        self: std::sync::Arc<Self>,
+        chat_id: teloxide::prelude::ChatId,
+    ) -> BoxFuture<'static, Result<Option<D>, Self::Error>> {
+        Box::pin(async move {
+            let d = self.database.collection::<Dialogue>("dialogues");
+            Ok(d.find_one(doc! { "chat_id": chat_id.0 }).await?.map(|d| {
+                self.serializer
+                    .deserialize(
+                        d.dialogue
+                            .into_iter()
+                            .map(|i| i as u8)
+                            .collect::<Vec<_>>()
+                            .as_slice(),
+                    )
+                    .unwrap()
+            }))
+        })
+    }
+}