chore: fluvio tests

Randoooom · Randoooom · commit 35105ec12cb8 · 2025-07-13T17:21:26.000+02:00
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -120,6 +120,11 @@ jobs:
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
 
+      - name: Install FluvioCLI
+        run: |
+          curl -fsS https://hub.infinyon.cloud/install/install.sh | bash
+          export PATH=${PATH}:~/.fluvio/bin:~/.fvm/bin
+
       - name: Tests
         run: just test-all 
 
diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml
@@ -73,6 +73,11 @@ jobs:
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
 
+      - name: Install FluvioCLI
+        run: |
+          curl -fsS https://hub.infinyon.cloud/install/install.sh | bash
+          export PATH=${PATH}:~/.fluvio/bin:~/.fvm/bin
+
       - name: Tests
         run: just test-all 
 
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
@@ -73,6 +73,11 @@ jobs:
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
 
+      - name: Install FluvioCLI
+        run: |
+          curl -fsS https://hub.infinyon.cloud/install/install.sh | bash
+          export PATH=${PATH}:~/.fluvio/bin:~/.fvm/bin
+
       - name: Tests
         run: just test-all 
 
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -14,6 +14,7 @@ aliri_clock = "0.1.4"
 aliri_oauth2 = { version = "0.10.1", features = ["rustls-tls"] }
 aliri_tower = "0.6.0"
 aliri_traits = "0.1.1"
+axum = "0.8.4"
 async-recursion = "1.1.1"
 async-trait = "0.1.80"
 bincode = { version = "2.0.1", features = ["serde"] }
diff --git a/common/src/database/mssql.sql b/common/src/database/mssql.sql
@@ -1,4 +1,3 @@
-
 IF NOT EXISTS (SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'target')
 BEGIN
     CREATE TABLE target (
@@ -92,7 +91,7 @@ BEGIN
         data                VARBINARY(MAX) NOT NULL,
         made_by             VARCHAR(32)    NOT NULL  REFERENCES oidc_user(id),
         action              VARCHAR(32)    NOT NULL,
-        version             VARCHAR(32)    NOT NULL,
+        version             INT            NOT NULL,
         created_at          DATETIME2(3),
     );
 END;
diff --git a/common/src/database/mysql.sql b/common/src/database/mysql.sql
@@ -70,7 +70,7 @@ CREATE TABLE IF NOT EXISTS audit_version (
   data                BLOB NOT NULL,
   made_by             VARCHAR(32) NOT NULL,
   action              VARCHAR(32) NOT NULL,
-  version             VARCHAR(32) NOT NULL,
+  version             INT UNSIGNED NOT NULL,
   created_at          DATETIME(3),
   FOREIGN KEY (made_by) REFERENCES oidc_user(id)
 );
diff --git a/common/src/database/postgres.sql b/common/src/database/postgres.sql
@@ -65,6 +65,6 @@ CREATE TABLE IF NOT EXISTS audit_version (
   data                BYTEA NOT NULL, 
   made_by             VARCHAR(32) REFERENCES oidc_user(id) NOT NULL,
   action              VARCHAR(32) NOT NULL,
-  version             VARCHAR(32) NOT NULL,
+  version             INTEGER NOT NULL,
   created_at  TIMESTAMP(3)
 );
diff --git a/common/src/database/schema/audit.rs b/common/src/database/schema/audit.rs
@@ -77,7 +77,10 @@ pub struct AuditVersion {
 }
 
 crud!(AuditVersion {});
-impl_select_page!(AuditVersion { select_page_by_resource_type_and_resource_id(resource_type: &impl Serialize, resource_id: &str) => "`WHERE resource_type = #{resource_type} resource_id = #{resource_id}`" });
+impl_select_page!(AuditVersion { select_page_by_resource_type_and_resource_id(resource_type: &impl Serialize, resource_id: &str) => "
+        `WHERE resource_type = #{resource_type} AND resource_id = #{resource_id}`
+    if do_count == flase:
+        `ORDER BY version DESC`" });
 
 impl From<&AuditResource> for ProtoResourceKind {
     fn from(value: &AuditResource) -> Self {
diff --git a/indexer/Cargo.toml b/indexer/Cargo.toml
@@ -8,6 +8,7 @@ feedback_fusion_common = { path = "../common" }
 
 anyhow = "1.0.98"
 async-trait = "0.1.88"
+axum = "0.8.4"
 chrono = { version = "0.4.38", features = ["serde"] }
 fluvio = "0.50.1"
 futures = "0.3.31"
diff --git a/indexer/src/main.rs b/indexer/src/main.rs
@@ -28,16 +28,22 @@ use crate::{
     prelude::*,
     processor::FeedbackFusionIndexerProcessor,
 };
+use axum::{Router, routing::get};
 use feedback_fusion_common::database::DatabaseConnection;
+use tokio::net::TcpListener;
 
 mod broker;
 mod config;
 mod error;
 mod processor;
 
+const HEALTHCHECK_ADDRESS: &str = "0.0.0.0:3000";
+
 #[tokio::main]
 async fn main() {
-    rustls::crypto::aws_lc_rs::default_provider().install_default().unwrap();
+    rustls::crypto::aws_lc_rs::default_provider()
+        .install_default()
+        .unwrap();
     lazy_static::initialize(&CONFIG);
 
     feedback_fusion_common::observability::otlp::init_tracing(CONFIG.otlp());
@@ -62,9 +68,23 @@ async fn main() {
     let processor =
         FeedbackFusionIndexerProcessor::initialize(connection, broker_event_batch_receiver);
     processor
-        .start_worker(shutdown_sender.clone(), shutdown_receiver)
+        .start_worker(shutdown_sender.clone(), shutdown_receiver.clone())
         .await;
 
+    // start a axum server to serve a healthcheck
+    tokio::spawn(async move {
+        let router = Router::new().route("/", get(|| async { "OK" }));
+        let listener = TcpListener::bind(HEALTHCHECK_ADDRESS).await.unwrap();
+        axum::serve(listener, router)
+            .with_graceful_shutdown(async move {
+                shutdown_receiver.recv().await.ok();
+
+                info!("Got shotdown signal, gracefully stopping gRPC server");
+            })
+            .await
+            .unwrap();
+    });
+
     debug!("Trying to listen for graceful shutdown");
     match tokio::signal::ctrl_c().await {
         Ok(()) => {}
diff --git a/indexer/src/processor/audit.rs b/indexer/src/processor/audit.rs
@@ -37,7 +37,7 @@ use rbatis::rbdc::DateTime;
 struct MaxVersionQuery {
     resource_id: String,
     resource_type: AuditResource,
-    max_version: String,
+    max_version: u32,
 }
 
 /// Create the audit versions for the given events.
@@ -137,7 +137,7 @@ pub async fn create_audit_versions(
         result = vec![MaxVersionQuery {
             resource_type: AuditResource::Target,
             resource_id: "".to_owned(),
-            max_version: "0".to_owned(),
+            max_version: 0,
         }]
     }
 
@@ -170,7 +170,7 @@ pub async fn create_audit_versions(
                         {
                             is_match = true;
 
-                            latest_version.max_version.parse().unwrap()
+                            latest_version.max_version
                         } else {
                             0
                         }
diff --git a/justfile b/justfile
@@ -12,6 +12,7 @@ test-all: cleanup
   just test mysql
   just test mssql
   just test skytable
+  just test fluvio
 
   cargo llvm-cov report
   cargo llvm-cov report --lcov --output-path coverage.info
@@ -47,10 +48,10 @@ build-all DOCKERFILE="./Dockerfile":
 backend TYPE=DEFAULT_TEST:
   FEEDBACK_FUSION_CONFIG="./tests/_common/configs/indexer/{{TYPE}}.hcl" RUST_LOG=DEBUG cargo llvm-cov run --bin indexer --no-report > ./target/feedback-fusion-indexer.log 2>&1 &
 
-  while ! nc -z localhost 7000; do \
+  while ! nc -z localhost 3000; do \
     sleep 1; \
   done
-  @echo "Application ready"
+  @echo "Indexer ready"
 
   FEEDBACK_FUSION_CONFIG="./tests/_common/configs/{{TYPE}}.hcl" RUST_LOG=DEBUG cargo llvm-cov run --bin feedback-fusion --no-report > ./target/feedback-fusion.log 2>&1 &
 
@@ -64,7 +65,7 @@ stop-backend:
     kill -2 $PID; \
   fi
 
-  -@PID=$(lsof -t -i:7000) && if [ -n "$PID" ]; then \
+  -@PID=$(lsof -t -i:3000) && if [ -n "$PID" ]; then \
     kill -2 $PID; \
   fi
 
@@ -104,6 +105,19 @@ protoc-docs:
 @skytable: postgres
   docker run -p 2003:2003 --entrypoint skyd --rm --name skytable -d skytable/skytable --auth-root-password=passwordpassword --endpoint=tcp@0.0.0.0:2003
 
+@fluvio: postgres
+  fluvio cluster start
+
+  fluvio topic create feedback-fusion 
+
+@stop-fluvio:
+  #!/usr/bin/expect -f
+
+  spawn fluvio cluster delete
+  expect "Please type the cluster name to confirm:"
+  send "local\r"
+  expect eof
+
 # 
 # Testing
 #
@@ -114,6 +128,7 @@ protoc-docs:
   -just stop-backend
   -docker rm -f skytable > /dev/null 2>&1
   -docker network rm {{DOCKER_NETWORK}} > /dev/null 2>&1
+  -just stop-fluvio
 
 unittest:
   FEEDBACK_FUSION_CONFIG="./tests/_common/configs/postgres.hcl" cargo llvm-cov --bin feedback-fusion --no-report -- --nocapture
diff --git a/src/broker/producer.rs b/src/broker/producer.rs
@@ -29,14 +29,11 @@ use tokio_retry::strategy::{ExponentialBackoff, jitter};
 
 use crate::broker::FeedbackFusionBroker;
 
-const BATCH_SIZE: usize = 10;
-const BATCH_TIMEOUT: Duration = Duration::from_secs(10);
-
 pub async fn start_loop(
     mut broker: FeedbackFusionBroker,
     receiver: AsyncReceiver<ProtoEvent>,
 ) -> Result<()> {
-    let mut batch = Vec::with_capacity(BATCH_SIZE);
+    let mut batch = Vec::with_capacity(*CONFIG.broker().max_batch_size() as usize);
 
     loop {
         let first_event = match receiver.recv().await {
@@ -48,9 +45,9 @@ pub async fn start_loop(
         };
         batch.push(first_event);
 
-        let timeout = tokio::time::sleep(BATCH_TIMEOUT);
+        let timeout = tokio::time::sleep(Duration::from_millis(*CONFIG.broker().batch_interval() as u64));
         tokio::pin!(timeout);
-        while batch.len() < BATCH_SIZE {
+        while batch.len() < *CONFIG.broker().max_batch_size() as usize {
             tokio::select! {
                 biased;
 
diff --git a/src/database/schema/authorization.rs b/src/database/schema/authorization.rs
@@ -43,6 +43,7 @@ use rbatis::rbdc::DateTime;
     PartialOrd,
     Ord,
 )]
+#[serde(rename_all = "lowercase")]
 pub enum ResourceKind {
     Target,
     Prompt,
diff --git a/src/services/v1/audit.rs b/src/services/v1/audit.rs
@@ -119,7 +119,7 @@ pub async fn rollback_resource(
     {
         // try to fetch the rollback version
         let version: Vec<AuditVersion> = database_request!(
-            AuditVersion::select_by_map(connection, value!{ "resource_id": data.resource_id, "resource_type": data.resource_type, "version": data.version }).await,
+            AuditVersion::select_by_map(connection, value!{ "resource_id": data.resource_id, "resource_type": resource_kind, "version": data.version }).await,
             "Fetch the rollback data"
         )?;
 
diff --git a/tests/_common/configs/fluvio.hcl b/tests/_common/configs/fluvio.hcl
diff --git a/tests/_common/configs/indexer/fluvio.hcl b/tests/_common/configs/indexer/fluvio.hcl
diff --git a/tests/integration/audit.rs b/tests/integration/audit.rs
diff --git a/tests/integration/mod.rs b/tests/integration/mod.rs
