Add API key requirement to Ingest methods

Add API key management methods
Update UI

Cargo.toml

@@ -28,12 +28,13 @@ reqwest = { version = "0.12.12", default-features = false, features = ["json", "
 axum = { version = "0.8.1", features = ["macros"] }
 axum-extra = { version = "0.10.0", features = ["typed-header"] }
 clap = { version = "4.5.23", features = ["derive", "color", "usage", "env"] }
+metrics = "0.24.1"
 tokio = { version = "1.43", features = ["macros", "rt", "sync", "rt-multi-thread", "signal", "io-util"] }
 url = { version = "2.5.4", features = ["serde"] }
-utoipa = { version = "5.3.1", features = ["axum_extras", "uuid", "url"] }
+utoipa = { version = "5.3.1", features = ["axum_extras", "chrono", "uuid", "url"] }
 utoipa-axum = "0.2"
 utoipa-swagger-ui = { version = "9", features = ["axum", "vendored"] }
-uuid = { version = "1.12.0", features = ["serde", "v7", "fast-rng"] }
+uuid = { version = "1.12.0", features = ["serde", "v4", "v7", "fast-rng"] }
 
 [workspace.dependencies.sea-orm-migration]
 version = "1.1.4"

notifico-app/Cargo.toml

@@ -17,13 +17,17 @@ async-trait = "0.1.84"
 axum = { workspace = true }
 axum-extra = { workspace = true }
 backoff = { version = "0.4.0", features = ["tokio"] }
+chrono = "0.4.39"
 clap = { workspace = true }
 dotenvy = "0.15.7"
 fe2o3-amqp = { version = "0.13.1" }
 flume = "0.11.1"
 futures = "0.3.31"
 jsonwebtoken = "9.3.0"
 log = "0.4.22"
+metrics = { workspace = true }
+moka = "0.12.10"
+multimap = "0.10.0"
 rust-embed = { version = "8.5.0", features = ["mime-guess"] }
 sea-orm = { workspace = true }
 serde = { version = "1.0.217", features = ["derive"] }
@@ -38,4 +42,3 @@ utoipa = { workspace = true }
 utoipa-axum = { workspace = true }
 utoipa-swagger-ui = { workspace = true }
 uuid = { workspace = true }
-multimap = "0.10.0"

notifico-app/assets/index.html

@@ -22,7 +22,7 @@
     <title>
       Refine - Build your React-based CRUD applications, without constraints.
     </title>
-    <script type="module" crossorigin src="/assets/index-DL7kKpS3.js"></script>
+    <script type="module" crossorigin src="/assets/index-DKhqOV6l.js"></script>
     <link rel="modulepreload" crossorigin href="/assets/monaco_editor-RvQpGkV2.js">
     <link rel="modulepreload" crossorigin href="/assets/mui-BkgdmF8m.js">
     <link rel="stylesheet" crossorigin href="/assets/monaco_editor-DyX1CsEw.css">

notifico-app/migration/src/lib.rs

@@ -5,6 +5,7 @@ mod m20250108_000001_create_project_table;
 mod m20250108_000002_create_pipeline_event;
 mod m20250108_000003_recipient;
 mod m20250121_000001_create_template_table;
+mod m20250203_000001_create_apikey_table;
 
 pub struct Migrator;
 
@@ -16,6 +17,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20250108_000002_create_pipeline_event::Migration),
             Box::new(m20250108_000003_recipient::Migration),
             Box::new(m20250121_000001_create_template_table::Migration),
+            Box::new(m20250203_000001_create_apikey_table::Migration),
         ]
     }
 

notifico-app/migration/src/m20250203_000001_create_apikey_table.rs

@@ -0,0 +1,50 @@
+use sea_orm_migration::{prelude::*, schema::*};
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .create_table(
+                Table::create()
+                    .table(ApiKey::Table)
+                    .if_not_exists()
+                    .col(pk_uuid(ApiKey::Id))
+                    .col(uuid_uniq(ApiKey::Key))
+                    .col(uuid(ApiKey::ProjectId))
+                    .col(string(ApiKey::Description).default(""))
+                    .col(date_time(ApiKey::CreatedAt).default(Expr::current_timestamp()))
+                    .foreign_key(
+                        ForeignKey::create()
+                            .from(ApiKey::Table, ApiKey::ProjectId)
+                            .to(Project::Table, Project::Id)
+                            .on_delete(ForeignKeyAction::Cascade)
+                            .on_update(ForeignKeyAction::Restrict),
+                    )
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> {
+        unimplemented!()
+    }
+}
+
+#[derive(DeriveIden)]
+enum ApiKey {
+    Table,
+    Id,
+    Key,
+    ProjectId,
+    Description,
+    CreatedAt,
+}
+
+#[derive(DeriveIden)]
+enum Project {
+    Table,
+    Id,
+}

notifico-app/src/controllers/api_key.rs

@@ -0,0 +1,201 @@
+use crate::crud_table::{
+    AdminCrudError, AdminCrudTable, ItemWithId, ListQueryParams, ListableTrait, PaginatedResult,
+};
+use crate::entity;
+use futures::FutureExt;
+use metrics::{counter, gauge, Counter, Gauge};
+use moka::future::Cache;
+use moka::notification::ListenerFuture;
+use sea_orm::ActiveValue::Unchanged;
+use sea_orm::{
+    ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, NotSet, PaginatorTrait,
+    QueryFilter, Set,
+};
+use serde::{Deserialize, Serialize};
+use std::time::Duration;
+use utoipa::ToSchema;
+use uuid::Uuid;
+
+#[derive(Clone, Debug, Serialize, Deserialize, ToSchema)]
+pub struct ApiKey {
+    pub key: Uuid,
+    pub description: String,
+    pub project_id: Uuid,
+    pub created_at: Option<chrono::NaiveDateTime>,
+}
+
+pub struct ApiKeyController {
+    db: DatabaseConnection,
+    authorization_cache: Cache<Uuid, Uuid>,
+    authorization_cache_gauge: Gauge,
+    authorization_cache_hit: Counter,
+    authorization_cache_miss: Counter,
+    authorization_invalid_key: Counter,
+}
+
+impl ApiKeyController {
+    pub fn new(db: DatabaseConnection) -> Self {
+        let authorization_cache_capacity = 100;
+        gauge!("ingest_api_key_cache_capacity").set(authorization_cache_capacity as f64);
+
+        let authorization_cache_gauge = gauge!("ingest_api_key_cache_total");
+        let authorization_cache_gauge_for_fut = authorization_cache_gauge.clone();
+
+        let authorization_cache = Cache::builder()
+            .max_capacity(authorization_cache_capacity)
+            .time_to_live(Duration::from_secs(1))
+            .async_eviction_listener(move |_, _, _| -> ListenerFuture {
+                authorization_cache_gauge_for_fut.decrement(1);
+                async {}.boxed()
+            })
+            .build();
+
+        Self {
+            db,
+            authorization_cache,
+            authorization_cache_gauge,
+            authorization_cache_hit: counter!("ingest_api_key_cache_hit"),
+            authorization_cache_miss: counter!("ingest_api_key_cache_miss"),
+            authorization_invalid_key: counter!("ingest_api_key_invalid"),
+        }
+    }
+}
+
+impl From<entity::api_key::Model> for ApiKey {
+    fn from(value: entity::api_key::Model) -> Self {
+        ApiKey {
+            key: value.key,
+            description: value.description,
+            project_id: value.project_id,
+            created_at: Some(value.created_at),
+        }
+    }
+}
+
+impl AdminCrudTable for ApiKeyController {
+    type Item = ApiKey;
+
+    async fn get_by_id(&self, id: Uuid) -> Result<Option<Self::Item>, AdminCrudError> {
+        let query = entity::api_key::Entity::find_by_id(id)
+            .one(&self.db)
+            .await?;
+        Ok(query.map(ApiKey::from))
+    }
+
+    async fn list(
+        &self,
+        params: ListQueryParams,
+    ) -> Result<PaginatedResult<ItemWithId<Self::Item>>, AdminCrudError> {
+        let params = params.try_into()?;
+        let query = entity::api_key::Entity::find()
+            .apply_params(&params)
+            .unwrap()
+            .all(&self.db)
+            .await?;
+
+        Ok(PaginatedResult {
+            items: query
+                .into_iter()
+                .map(|m| ItemWithId {
+                    id: m.id,
+                    item: ApiKey::from(m),
+                })
+                .collect(),
+            total: entity::api_key::Entity::find()
+                .apply_filter(&params)?
+                .count(&self.db)
+                .await?,
+        })
+    }
+
+    async fn create(&self, item: Self::Item) -> Result<ItemWithId<Self::Item>, AdminCrudError> {
+        let id = Uuid::now_v7();
+        let key = Uuid::new_v4();
+
+        entity::api_key::ActiveModel {
+            id: Set(id),
+            key: Set(key),
+            project_id: Set(item.project_id),
+            description: Set(item.description.to_string()),
+            created_at: NotSet,
+        }
+        .insert(&self.db)
+        .await?;
+
+        Ok(ItemWithId {
+            id,
+            item: ApiKey {
+                key,
+                description: item.description.to_string(),
+                project_id: item.project_id,
+                created_at: Some(chrono::Utc::now().naive_utc()),
+            },
+        })
+    }
+
+    async fn update(
+        &self,
+        id: Uuid,
+        item: Self::Item,
+    ) -> Result<ItemWithId<Self::Item>, AdminCrudError> {
+        entity::api_key::ActiveModel {
+            id: Unchanged(id),
+            key: NotSet,
+            project_id: NotSet,
+            description: Set(item.description.to_string()),
+            created_at: NotSet,
+        }
+        .update(&self.db)
+        .await?;
+        Ok(ItemWithId { id, item })
+    }
+
+    async fn delete(&self, id: Uuid) -> Result<(), AdminCrudError> {
+        entity::api_key::Entity::delete_by_id(id)
+            .exec(&self.db)
+            .await?;
+        Ok(())
+    }
+}
+
+pub enum ApiKeyError {
+    InvalidApiKey,
+    InternalError,
+}
+
+impl ApiKeyController {
+    pub async fn authorize_api_key(&self, key: &str) -> Result<Uuid, ApiKeyError> {
+        let Ok(key_uuid) = Uuid::try_parse(key) else {
+            self.authorization_invalid_key.increment(1);
+            return Err(ApiKeyError::InvalidApiKey);
+        };
+
+        let cached_project_id = self.authorization_cache.get(&key_uuid).await;
+
+        if let Some(project_id) = cached_project_id {
+            // Cache Hit
+            self.authorization_cache_hit.increment(1);
+            Ok(project_id)
+        } else {
+            // Cache Miss
+            self.authorization_cache_miss.increment(1);
+
+            let Some(api_key_entry) = entity::api_key::Entity::find()
+                .filter(entity::api_key::Column::Key.eq(key_uuid))
+                .one(&self.db)
+                .await
+                .map_err(|_| ApiKeyError::InternalError)?
+            else {
+                self.authorization_invalid_key.increment(1);
+                return Err(ApiKeyError::InvalidApiKey);
+            };
+
+            let project_id = api_key_entry.project_id;
+
+            self.authorization_cache.insert(key_uuid, project_id).await;
+            self.authorization_cache_gauge.increment(1);
+
+            Ok(project_id)
+        }
+    }
+}

notifico-app/src/controllers/mod.rs

@@ -1,3 +1,4 @@
+pub mod api_key;
 pub mod event;
 pub mod group;
 pub mod pipeline;