Refactored db directory

service/src/storage/guilds/mod.rs

@@ -0,0 +1,3 @@
+mod model;
+
+pub use model::*;

service/src/storage/guilds/model.rs

@@ -0,0 +1,43 @@
+use diesel::prelude::*;
+use serde::{Serialize, Deserialize};
+use siren::ServiceError;
+
+use crate::storage::{schema::guilds, connection};
+
+#[derive(Queryable, QueryableByName, Serialize, Deserialize)]
+#[diesel(table_name = guilds)]
+pub struct QueryGuild {
+  pub id: i64,
+  pub bot_id: i64,
+  pub volume: i32
+}
+
+impl QueryGuild {
+  pub fn get(id: i64) -> Result<Self, ServiceError> {
+    let mut conn = connection()?;
+    let guild = guilds::table.filter(guilds::id.eq(id)).first(&mut conn)?;
+    Ok(guild)
+  }
+}
+
+#[derive(Insertable, AsChangeset, Serialize, Deserialize)]
+#[diesel(table_name = guilds)]
+pub struct InsertGuild {
+  pub id: i64,
+  pub bot_id: i64,
+  pub volume: i32
+}
+
+impl InsertGuild {
+  pub fn insert(guild: Self) -> Result<QueryGuild, ServiceError> {
+    let mut conn = connection()?;
+    let guild = diesel::insert_into(guilds::table).values(guild).get_result(&mut conn)?;
+    Ok(guild)
+  }
+
+  pub fn update_audio(id: i64, volume: i32) -> Result<QueryGuild, ServiceError> {
+    let mut conn = connection()?;
+    let guild = diesel::update(guilds::table.filter(guilds::id.eq(id))).set(guilds::volume.eq(volume)).get_result(&mut conn)?;
+    Ok(guild)
+  }
+}

service/src/storage/messages/mod.rs

@@ -0,0 +1,5 @@
+mod model;
+mod routes;
+
+pub use model::*;
+pub use routes::init_routes;

service/src/storage/messages/model.rs

@@ -0,0 +1,150 @@
+use diesel::prelude::*;
+use serde::{Deserialize, Serialize};
+use siren::ServiceError;
+
+use crate::storage::{schema::messages::{self}, connection};
+
+#[derive(Queryable, Selectable, Serialize, Deserialize)]
+#[diesel(table_name = messages)]
+pub struct QueryMessage {
+  pub id: String,
+  pub guild_id: i64,
+  pub channel_id: i64,
+  pub user_id: i64,
+  pub created: i64,
+  pub model: String,
+  pub request: String,
+  pub response: String,
+  pub request_tags: Vec<String>,
+  pub response_tags: Vec<String>,
+}
+
+pub struct QueryFilters {
+  pub by_id: Option<String>,
+  pub by_guild_id: Option<i64>,
+  pub by_channel_id: Option<i64>,
+  pub by_user_id: Option<i64>,
+  pub by_model: Option<String>,
+  pub by_request: Option<String>,
+  pub by_response: Option<String>,
+  pub by_request_tags: Option<Vec<String>>,
+  pub by_response_tags: Option<Vec<String>>
+}
+
+impl Default for QueryFilters {
+  fn default() -> Self {
+    QueryFilters {
+      by_id: None,
+      by_guild_id: None,
+      by_channel_id: None,
+      by_user_id: None,
+      by_model: None,
+      by_request: None,
+      by_response: None,
+      by_request_tags: None,
+      by_response_tags: None
+    }
+  }
+}
+
+impl QueryMessage {
+  pub fn get_all(filters: &QueryFilters, limit: i32, page: i32) -> Result<Vec<Self>, ServiceError> {
+    let mut conn = connection()?;
+    let mut query = messages::table.limit(limit as i64).order(messages::created.asc()).into_boxed();
+    // Limit query to page and limit
+    let offset = (page - 1) * limit;
+    query = query.offset(offset as i64);
+    // Apply filters
+    if let Some(id) = &filters.by_id {
+      query = query.filter(messages::id.eq(id));
+    }
+    if let Some(guild_id) = &filters.by_guild_id {
+      query = query.filter(messages::guild_id.eq(guild_id));
+    }
+    if let Some(channel_id) = &filters.by_channel_id {
+      query = query.filter(messages::channel_id.eq(channel_id));
+    }
+    if let Some(user_id) = &filters.by_user_id {
+      query = query.filter(messages::user_id.eq(user_id));
+    }
+    if let Some(model) = &filters.by_model {
+      query = query.filter(messages::model.eq(model));
+    }
+    if let Some(request) = &filters.by_request {
+      query = query.filter(messages::request.eq(request));
+    }
+    if let Some(response) = &filters.by_response {
+      query = query.filter(messages::response.eq(response));
+    }
+    if let Some(request_tags) = &filters.by_request_tags {
+      query = query.filter(messages::request_tags.eq(request_tags));
+    }
+    if let Some(response_tags) = &filters.by_response_tags {
+      query = query.filter(messages::response_tags.eq(response_tags));
+    }
+    // Execute query
+    let messages = query.load::<Self>(&mut conn)?;
+    Ok(messages)
+  }
+
+  pub fn get_count(fitlers: &QueryFilters) -> Result<i64, ServiceError> {
+    let mut conn = connection()?;
+    let mut query = messages::table.into_boxed();
+    // Apply filters
+    if let Some(id) = &fitlers.by_id {
+      query = query.filter(messages::id.eq(id));
+    }
+    if let Some(guild_id) = &fitlers.by_guild_id {
+      query = query.filter(messages::guild_id.eq(guild_id));
+    }
+    if let Some(channel_id) = &fitlers.by_channel_id {
+      query = query.filter(messages::channel_id.eq(channel_id));
+    }
+    if let Some(user_id) = &fitlers.by_user_id {
+      query = query.filter(messages::user_id.eq(user_id));
+    }
+    if let Some(model) = &fitlers.by_model {
+      query = query.filter(messages::model.eq(model));
+    }
+    if let Some(request) = &fitlers.by_request {
+      query = query.filter(messages::request.eq(request));
+    }
+    if let Some(response) = &fitlers.by_response {
+      query = query.filter(messages::response.eq(response));
+    }
+    if let Some(request_tags) = &fitlers.by_request_tags {
+      query = query.filter(messages::request_tags.eq(request_tags));
+    }
+    if let Some(response_tags) = &fitlers.by_response_tags {
+      query = query.filter(messages::response_tags.eq(response_tags));
+    }
+    // Execute query
+    let count = query.count().get_result::<i64>(&mut conn)?;
+    Ok(count)
+  }
+}
+
+#[derive(Insertable, AsChangeset, Serialize, Deserialize)]
+#[diesel(table_name = messages)]
+pub struct InsertMessage {
+  pub id: String,
+  pub guild_id: i64,
+  pub channel_id: i64,
+  pub user_id: i64,
+  pub created: i64,
+  pub model: String,
+  pub request: String,
+  pub response: String,
+  pub request_tags: Vec<String>,
+  pub response_tags: Vec<String>,
+}
+
+impl InsertMessage {
+  pub fn insert(message: Self) -> Result<QueryMessage, ServiceError> {
+    let mut conn = connection()?;
+    let message = diesel::insert_into(messages::table)
+      .values(message)
+      .get_result(&mut conn)?;
+    Ok(message)
+  }
+}

service/src/storage/messages/routes.rs

@@ -0,0 +1,88 @@
+use actix_web::{get, post, web, HttpResponse, HttpRequest, ResponseError};
+use log::error;
+use serde::{Serialize, Deserialize};
+use siren::{GetResponse, Metadata, ServiceError};
+
+use crate::{storage::messages::{QueryMessage, QueryFilters, InsertMessage}, auth::{JwtAuth, verify_role}};
+
+#[derive(Serialize, Deserialize)]
+struct GetAllParams {
+  id: Option<String>,
+  guild_id: Option<i64>,
+  channel_id: Option<i64>,
+  user_id: Option<i64>,
+  model: Option<String>,
+  request: Option<String>,
+  response: Option<String>,
+  request_tags: Option<Vec<String>>,
+  response_tags: Option<Vec<String>>,
+  limit: Option<i32>,
+  page: Option<i32>,
+}
+
+#[get("/messages")]
+async fn get_all(req: HttpRequest, auth: JwtAuth) -> HttpResponse {
+  let _ = match verify_role(&auth, "admin") {
+    Ok(_) => {},
+    Err(err) => return ResponseError::error_response(&err)
+  };
+  let params = match web::Query::<GetAllParams>::from_query(req.query_string()) {
+    Ok(params) => params,
+    Err(err) => return ResponseError::error_response(&ServiceError {
+      status: 422,
+      message: err.to_string()
+    })
+  };
+  let mut filters = QueryFilters::default();
+  filters.by_id = params.id.clone();
+  filters.by_guild_id = params.guild_id;
+  filters.by_channel_id = params.channel_id;
+  filters.by_user_id = params.user_id;
+  filters.by_model = params.model.clone();
+  filters.by_request = params.request.clone();
+  filters.by_response = params.response.clone();
+  filters.by_request_tags = params.request_tags.clone();
+  filters.by_response_tags = params.response_tags.clone();
+  let limit = params.limit.unwrap_or(100);
+  let total_count = QueryMessage::get_count(&filters).unwrap();
+  let max_page = std::cmp::max((total_count as f64 / limit as f64).ceil() as i32, 1);
+  let page = std::cmp::min(std::cmp::max(params.page.unwrap_or(1), 1), max_page);
+  
+  match QueryMessage::get_all(&filters, limit, page) {
+    Ok(messages) => {
+      HttpResponse::Ok().json(GetResponse {
+        data: messages,
+        metadata: Some(Metadata {
+          total: total_count as i32,
+          limit,
+          page,
+          pages: max_page
+        })
+      })
+    },
+    Err(err) => {
+      error!("{:?}", err.message);
+      ResponseError::error_response(&err)
+    }
+  }
+}
+
+#[post("/messages")]
+async fn create(message: web::Json<InsertMessage>, auth: JwtAuth) -> HttpResponse {
+  let _ = match verify_role(&auth, "admin") {
+    Ok(_) => {},
+    Err(err) => return ResponseError::error_response(&err)
+  };
+  match InsertMessage::insert(message.into_inner()) {
+    Ok(message) => HttpResponse::Created().json(message),
+    Err(err) => {
+      error!("{:?}", err.message);
+      ResponseError::error_response(&err)
+    }
+  }
+}
+
+pub fn init_routes(config: &mut web::ServiceConfig) {
+  config.service(get_all);
+  config.service(create);
+}

service/src/storage/mod.rs

@@ -0,0 +1,93 @@
+use diesel::{r2d2::ConnectionManager as DieselConnectionManager, PgConnection};
+use minio::s3::{client::Client as MinioClient, http::BaseUrl, creds::StaticProvider, args::{MakeBucketArgs, BucketExistsArgs}};
+use redis::{Client as RedisClient, aio::Connection as RedisConnection};
+use siren::ServiceError;
+use crate::diesel_migrations::MigrationHarness;
+use lazy_static::lazy_static;
+use log::{error, info};
+use r2d2;
+use std::env;
+
+pub mod guilds;
+pub mod messages;
+pub mod schema;
+
+type DbPool = r2d2::Pool<DieselConnectionManager<PgConnection>>;
+pub type DbConnection = r2d2::PooledConnection<DieselConnectionManager<PgConnection>>;
+
+pub const MIGRATIONS: diesel_migrations::EmbeddedMigrations = embed_migrations!();
+
+lazy_static! {
+  static ref POOL: DbPool = {
+    let username = env::var("DATABASE_USER").expect("DATABASE_USERNAME is not set");
+    let password = env::var("DATABASE_PASSWORD").expect("DATABASE_PASSWORD is not set");
+    let host = env::var("DATABASE_HOST").unwrap_or("localhost".to_string());
+    let name = env::var("DATABASE_NAME").expect("DATABASE_NAME is not set");
+    let port = env::var("DATABASE_PORT").unwrap_or("5432".to_string());
+    let url = format!("postgres://{}:{}@{}:{}/{}", username, password, host, port, name);
+    let manager = DieselConnectionManager::<PgConnection>::new(url);
+    DbPool::builder().test_on_check_out(true).build(manager).expect("Failed to create db pool")
+  };
+  static ref REDIS: RedisClient = {
+    let host = env::var("REDIS_HOST").unwrap_or("localhost".to_string());
+    let port = env::var("REDIS_PORT").unwrap_or("6379".to_string());
+    let url = format!("redis://{}:{}", host, port);
+    RedisClient::open(url).expect("Failed to create redis client")
+  };
+  static ref MINIO: MinioClient = {
+    let url = env::var("MINIO_URL").unwrap_or("localhost".to_string());
+    let port = env::var("MINIO_PORT").unwrap_or("9000".to_string());
+    let base_url = format!("http://{}:{}", url, port).parse::<BaseUrl>().unwrap();
+
+    let user = env::var("MINIO_ROOT_USER").expect("MINIO_ROOT_USER is not set");
+    let password = env::var("MINIO_ROOT_PASSWORD").expect("MINIO_ROOT_PASSWORD is not set");
+
+    let static_provider = StaticProvider::new(
+      &user,
+      &password,
+      None
+    );
+
+    MinioClient::new(
+      base_url,
+      Some(Box::new(static_provider)),
+      None,
+      None
+    ).expect("Failed to create minio client")
+  };
+}
+
+pub fn init() {
+  lazy_static::initialize(&POOL);
+  lazy_static::initialize(&REDIS);
+  lazy_static::initialize(&MINIO);
+  let mut pool: DbConnection = connection().expect("Failed to get db connection");
+  match pool.run_pending_migrations(MIGRATIONS) {
+    Ok(_) => info!("Database initialized"),
+    Err(err) => error!("Failed to initialize database; {}", err)
+  };
+}
+
+pub fn connection() -> Result<DbConnection, ServiceError> {
+  POOL.get()
+    .map_err(|e| ServiceError::new(500, format!("Failed getting db connection: {}", e)))
+}
+
+pub fn redis_connection() -> Result<redis::Connection, ServiceError> {
+  let conn = REDIS.get_connection()?;
+  Ok(conn)
+}
+
+pub async fn redis_async_connection() -> Result<RedisConnection, ServiceError> {
+  let conn = REDIS.get_async_connection().await?;
+  Ok(conn)
+}
+
+pub async fn create_bucket(bucket_name: &str) -> Result<(), ServiceError> {
+  let exists  = MINIO.bucket_exists(&BucketExistsArgs::new(&bucket_name).unwrap()).await?;
+  if !exists {
+    MINIO.make_bucket(&MakeBucketArgs::new(&bucket_name).unwrap()).await?;
+  }
+
+  Ok(())
+}

service/src/storage/schema.rs

@@ -0,0 +1,53 @@
+diesel::table! {
+  messages (id) {
+    id -> Text,
+    guild_id -> BigInt,
+    channel_id -> BigInt,
+    user_id -> BigInt,
+    created -> BigInt,
+    model -> Text,
+    request -> Text,
+    response -> Text,
+    request_tags -> Array<Text>,
+    response_tags -> Array<Text>,
+  }
+}
+
+diesel::table! {
+  spells (id) {
+    id -> Integer,
+    name -> Text,
+    school -> Text,
+    level -> Integer,
+    ritual -> Bool,
+    concentration -> Bool,
+    classes -> Array<Text>,
+    damage_inflict -> Array<Text>,
+    damage_resist -> Array<Text>,
+    conditions -> Array<Text>,
+    saving_throw -> Array<Text>,
+    attack_type -> Nullable<Text>,
+    data -> Jsonb
+  }
+}
+
+diesel::table! {
+  guilds (id) {
+    id -> BigInt,
+    bot_id -> BigInt,
+    volume -> Integer,
+  }
+}
+
+diesel::table! {
+  users (email) {
+    email -> Text,
+    hash -> Text,
+    role -> Text,
+    first_name -> Text,
+    last_name -> Text,
+    updated_at -> Timestamp,
+    created_at -> Timestamp,
+    verified -> Bool,
+  }
+}