Message queue module

Cargo.toml

@@ -12,6 +12,7 @@ members = [
     "neptune",
     "aries",
     "saturn",
+    "taurus",
     "lunar/src-tauri",
     "atlas",
 ]
@@ -31,6 +32,7 @@ gemini = { path = "gemini" }
 vault = { path = "vault" }
 neptune = { path = "neptune" }
 saturn = { path = "saturn" }
+taurus = { path = "taurus" }
 mega = { path = "mega" }
 anyhow = "1.0.86"
 serde = {version = "1.0.203", features = ["derive"]}

firedbg/target.json

@@ -0,0 +1,6 @@
+{
+  "binaries": [],
+  "examples": [],
+  "integration_tests": [],
+  "unit_tests": []
+}

firedbg/version.toml

@@ -0,0 +1 @@
+firedbg_cli = "1.74.0"

gateway/Cargo.toml

@@ -18,6 +18,7 @@ gemini = { workspace = true }
 vault = { workspace = true }
 venus = { workspace = true }
 mercury = { workspace = true }
+taurus = { workspace = true }
 anyhow = { workspace = true }
 axum = { workspace = true }
 axum-server = { version = "0.7", features = ["tls-rustls"] }
@@ -37,7 +38,7 @@ tower-http = { workspace = true, features = [
     "trace",
     "decompression-full",
 ] }
-axum-extra = { workspace = true, features = ["typed-header"]}
+axum-extra = { workspace = true, features = ["typed-header"] }
 tokio = { workspace = true, features = ["net"] }
 tokio-stream = { workspace = true }
 async-stream = { workspace = true }

gateway/src/api/api_router.rs

@@ -6,6 +6,7 @@ use axum::{
     Json, Router,
 };
 
+use taurus::event::api_request::{ApiRequestEvent, ApiType};
 use ceres::model::{
     create_file::CreateFileInfo,
     publish_path::PublishPathInfo,
@@ -39,6 +40,7 @@ async fn get_blob_object(
     Query(query): Query<BlobContentQuery>,
     state: State<ApiServiceState>,
 ) -> Result<Json<CommonResult<String>>, (StatusCode, String)> {
+    ApiRequestEvent::notify(ApiType::Blob, &state.0.context.config);
     let res = state
         .api_handler(query.path.clone().into())
         .await
@@ -80,6 +82,7 @@ async fn create_file(
     state: State<ApiServiceState>,
     Json(json): Json<CreateFileInfo>,
 ) -> Result<Json<CommonResult<String>>, (StatusCode, String)> {
+    ApiRequestEvent::notify(ApiType::CreateFile, &state.0.context.config);
     let res = state
         .api_handler(json.path.clone().into())
         .await
@@ -96,6 +99,7 @@ async fn get_latest_commit(
     Query(query): Query<CodePreviewQuery>,
     state: State<ApiServiceState>,
 ) -> Result<Json<LatestCommitInfo>, (StatusCode, String)> {
+    ApiRequestEvent::notify(ApiType::LastestCommit, &state.0.context.config);
     let res = state
         .api_handler(query.path.clone().into())
         .await
@@ -109,6 +113,7 @@ async fn get_tree_info(
     Query(query): Query<CodePreviewQuery>,
     state: State<ApiServiceState>,
 ) -> Result<Json<CommonResult<Vec<TreeBriefItem>>>, (StatusCode, String)> {
+    ApiRequestEvent::notify(ApiType::TreeInfo, &state.0.context.config);
     let res = state
         .api_handler(query.path.clone().into())
         .await
@@ -125,6 +130,7 @@ async fn get_tree_commit_info(
     Query(query): Query<CodePreviewQuery>,
     state: State<ApiServiceState>,
 ) -> Result<Json<CommonResult<Vec<TreeCommitItem>>>, (StatusCode, String)> {
+    ApiRequestEvent::notify(ApiType::CommitInfo, &state.0.context.config);
     let res = state
         .api_handler(query.path.clone().into())
         .await
@@ -141,6 +147,7 @@ async fn publish_path_to_repo(
     state: State<ApiServiceState>,
     Json(json): Json<PublishPathInfo>,
 ) -> Result<Json<CommonResult<String>>, (StatusCode, String)> {
+    ApiRequestEvent::notify(ApiType::Publish, &state.0.context.config);
     let res = state
         .api_handler(json.path.clone().into())
         .await

gateway/src/api/mr_router.rs

@@ -10,6 +10,7 @@ use axum::{
 use ceres::model::mr::{MRDetail, MrInfoItem};
 use common::model::CommonResult;
 
+use taurus::event::api_request::{ApiRequestEvent, ApiType};
 use crate::api::ApiServiceState;
 
 pub fn routers() -> Router<ApiServiceState> {
@@ -24,18 +25,22 @@ async fn merge(
     Path(mr_id): Path<i64>,
     state: State<ApiServiceState>,
 ) -> Result<Json<CommonResult<String>>, (StatusCode, String)> {
+    ApiRequestEvent::notify(ApiType::MergeRequest, &state.0.context.config);
+
     let res = state.monorepo().merge_mr(mr_id).await;
     let res = match res {
         Ok(_) => CommonResult::success(None),
         Err(err) => CommonResult::failed(&err.to_string()),
     };
+    ApiRequestEvent::notify(ApiType::MergeDone, &state.0.context.config);
     Ok(Json(res))
 }
 
 async fn get_mr_list(
     Query(query): Query<HashMap<String, String>>,
     state: State<ApiServiceState>,
 ) -> Result<Json<CommonResult<Vec<MrInfoItem>>>, (StatusCode, String)> {
+    ApiRequestEvent::notify(ApiType::MergeList, &state.0.context.config);
     let status = query.get("status").unwrap();
     let res = state.monorepo().mr_list(status).await;
     let res = match res {
@@ -49,6 +54,7 @@ async fn mr_detail(
     Path(mr_id): Path<i64>,
     state: State<ApiServiceState>,
 ) -> Result<Json<CommonResult<Option<MRDetail>>>, (StatusCode, String)> {
+    ApiRequestEvent::notify(ApiType::MergeDetail, &state.0.context.config);
     let res = state.monorepo().mr_detail(mr_id).await;
     let res = match res {
         Ok(data) => CommonResult::success(Some(data)),
@@ -61,6 +67,7 @@ async fn get_mr_files(
     Path(mr_id): Path<i64>,
     state: State<ApiServiceState>,
 ) -> Result<Json<CommonResult<Vec<PathBuf>>>, (StatusCode, String)> {
+    ApiRequestEvent::notify(ApiType::MergeFiles, &state.0.context.config);
     let res = state.monorepo().mr_tree_files(mr_id).await;
     let res = match res {
         Ok(data) => CommonResult::success(Some(data)),

jupiter/callisto/src/lib.rs

@@ -26,3 +26,4 @@ pub mod mega_tree;
 pub mod raw_blob;
 pub mod ztm_node;
 pub mod ztm_repo_info;
+pub mod mq_storage;

jupiter/callisto/src/mq_storage.rs

@@ -0,0 +1,19 @@
+//! `SeaORM` Entity. Generated by sea-orm-codegen 0.11.3
+
+use sea_orm::entity::prelude::*;
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
+#[sea_orm(table_name = "mq_storage")]
+pub struct Model {
+    #[sea_orm(primary_key, auto_increment = false)]
+    pub id: i64,
+    pub category: String,
+    pub create_time: DateTime,
+    #[sea_orm(column_type = "Text", nullable)]
+    pub content: Option<String>,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {}
+
+impl ActiveModelBehavior for ActiveModel {}

jupiter/src/context.rs

@@ -3,8 +3,7 @@ use std::sync::Arc;
 use common::config::Config;
 
 use crate::storage::{
-    git_db_storage::GitDbStorage, init::database_connection, lfs_storage::LfsStorage,
-    mega_storage::MegaStorage, ztm_storage::ZTMStorage,
+    git_db_storage::GitDbStorage, init::database_connection, lfs_storage::LfsStorage, mega_storage::MegaStorage, mq_storage::MQStorage, ztm_storage::ZTMStorage
 };
 
 #[derive(Clone)]
@@ -34,6 +33,7 @@ pub struct Service {
     pub git_db_storage: Arc<GitDbStorage>,
     pub lfs_storage: Arc<LfsStorage>,
     pub ztm_storage: Arc<ZTMStorage>,
+    pub mq_storage: Arc<MQStorage>,
 }
 
 impl Service {
@@ -48,6 +48,7 @@ impl Service {
             ),
             lfs_storage: Arc::new(LfsStorage::new(connection.clone()).await),
             ztm_storage: Arc::new(ZTMStorage::new(connection.clone()).await),
+            mq_storage: Arc::new(MQStorage::new(connection.clone()).await),
         }
     }
 
@@ -61,6 +62,7 @@ impl Service {
             git_db_storage: Arc::new(GitDbStorage::mock()),
             lfs_storage: Arc::new(LfsStorage::mock()),
             ztm_storage: Arc::new(ZTMStorage::mock()),
+            mq_storage: Arc::new(MQStorage::mock()),
         })
     }
 }

jupiter/src/storage/mod.rs

@@ -3,6 +3,7 @@ pub mod git_fs_storage;
 pub mod init;
 pub mod lfs_storage;
 pub mod mega_storage;
+pub mod mq_storage;
 pub mod ztm_storage;
 
 use async_trait::async_trait;

jupiter/src/storage/mq_storage.rs

@@ -0,0 +1,46 @@
+use std::sync::Arc;
+
+use callisto::mq_storage::*;
+use sea_orm::{DatabaseConnection, EntityTrait, QueryOrder, QuerySelect};
+
+use super::batch_save_model;
+
+
+#[derive(Clone)]
+pub struct MQStorage {
+    pub connection: Arc<DatabaseConnection>,
+}
+
+impl MQStorage {
+    pub fn get_connection(&self) -> &DatabaseConnection {
+        &self.connection
+    }
+
+    pub async fn new(connection: Arc<DatabaseConnection>) -> Self {
+        MQStorage { connection }
+    }
+
+    pub fn mock() -> Self {
+        MQStorage {
+            connection: Arc::new(DatabaseConnection::default()),
+        }
+    }
+
+    pub async fn save_messages(&self, msgs: Vec<Model>) {
+        if msgs.is_empty() {
+            return;
+        }
+
+        let msgs: Vec<ActiveModel> = msgs.into_iter().map(|m| m.into()).collect();
+        batch_save_model(self.get_connection(), msgs).await.unwrap();
+    }
+
+    pub async fn get_latest_message(&self) -> Option<Model> {
+        Entity::find()
+            .order_by_desc(Column::Id)
+            .limit(1)
+            .one(self.get_connection())
+            .await
+            .unwrap()
+    }
+}

mega/Cargo.toml

@@ -16,6 +16,7 @@ path = "src/main.rs"
 gateway = { workspace = true }
 common = { workspace = true }
 ceres = { workspace = true }
+taurus = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 tokio = { workspace = true, features = ["macros"] }
 clap = { workspace = true, features = ["derive"] }

mega/src/commands/service/mod.rs

@@ -25,6 +25,9 @@ pub fn cli() -> Command {
 // It determines which subcommand was used and calls the appropriate function.
 #[tokio::main]
 pub(crate) async fn exec(config: Config, args: &ArgMatches) -> MegaResult {
+    use taurus::init::init_mq;
+    init_mq(&config).await;
+
     let (cmd, subcommand_args) = match args.subcommand() {
         Some((cmd, args)) => (cmd, args),
         _ => {

saturn/src/context.rs

@@ -16,6 +16,7 @@ pub struct AppContext {
     schema: Schema,
 }
 
+#[allow(dead_code)]
 #[derive(Debug, Error)]
 pub enum ContextError {
     #[error("{0}")]
@@ -34,6 +35,7 @@ pub enum ContextError {
     Json(#[from] serde_json::Error),
 }
 
+#[allow(dead_code)]
 #[derive(Debug, Error)]
 pub enum Error {
     #[error("Authorization Denied")]
@@ -43,6 +45,7 @@ pub enum Error {
 }
 
 impl AppContext {
+    #[allow(dead_code)]
     pub fn new(
         entities: EntityStore,
         schema_path: impl Into<PathBuf>,
@@ -78,6 +81,7 @@ impl AppContext {
         }
     }
 
+    #[allow(dead_code)]
     pub fn is_authorized(
         &self,
         principal: impl AsRef<EntityUid>,

saturn/src/entitystore.rs

@@ -18,6 +18,7 @@ pub struct EntityStore {
 }
 
 impl EntityStore {
+    #[allow(dead_code)]
     pub fn as_entities(&self, schema: &Schema) -> Entities {
         let users = self.users.values().map(|user| user.clone().into());
         let repos = self.repos.values().map(|repo| repo.clone().into());
@@ -32,6 +33,7 @@ impl EntityStore {
         Entities::from_entities(all, Some(schema)).unwrap()
     }
 
+    #[allow(dead_code)]
     pub fn merge(&mut self, other: EntityStore) {
         self.users.extend(other.users);
         self.repos.extend(other.repos);

sql/postgres/pg_20240205__init.sql

@@ -243,4 +243,11 @@ CREATE TABLE IF NOT EXISTS "ztm_repo_info" (
   "origin" VARCHAR(64),
   "update_time" BIGINT NOT NULL,
   "commit" VARCHAR(64)
-);
+);
+
+CREATE TABLE IF NOT EXISTS "mq_storage" (
+  "id" BIGINT PRIMARY KEY,
+  "category" VARCHAR(64),
+  "create_time" TIMESTAMP NOT NULL,
+  "content" TEXT
+);

sql/sqlite/sqlite_20240711_init.sql

@@ -243,4 +243,11 @@ CREATE TABLE IF NOT EXISTS "ztm_repo_info" (
   "origin" TEXT,
   "update_time" INTEGER NOT NULL,
   "commit" TEXT
-);
+);
+
+CREATE TABLE IF NOT EXISTS "mq_storage" (
+  "id" INTEGER PRIMARY KEY,
+  "category" TEXT,
+  "create_time" TIMESTAMP NOT NULL,
+  "content" TEXT
+);

taurus/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "taurus"
+version = "0.1.0"
+edition = "2021"
+
+[lib]
+name = "taurus"
+path = "src/lib.rs"
+
+[dependencies]
+common = { workspace = true }
+jupiter = { workspace = true }
+callisto = { workspace = true }
+
+axum = { workspace = true }
+async-trait = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread"]}
+tracing = { workspace = true }
+thiserror = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+chrono = { workspace = true }
+crossbeam-channel = "0.5.10"

taurus/README.md

@@ -0,0 +1,2 @@
+# Message Queue Module
+This module offers mega the ability to send and handle specific events.