fix: short_url duplicating to meta (#5365)

src/infra/src/table/migration/m20241215_190333_delete_metas.rs

@@ -0,0 +1,68 @@
+// Copyright 2024 OpenObserve Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Deletes dashboard records from the meta table.
+
+use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, TransactionTrait};
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        let txn = manager.get_connection().begin().await?;
+        meta::Entity::delete_many()
+            .filter(meta::Column::Module.eq("short_urls"))
+            .exec(&txn)
+            .await?;
+        txn.commit().await?;
+        Ok(())
+    }
+
+    async fn down(&self, _: &SchemaManager) -> Result<(), DbErr> {
+        // The deletion of records from the meta table is not reversable.
+        Ok(())
+    }
+}
+
+// The schemas of tables might change after subsequent migrations. Therefore
+// this migration only references ORM models in private submodules that should
+// remain unchanged rather than ORM models in the `entity` module that will be
+// updated to reflect the latest changes to table schemas.
+
+/// Representation of the meta table at the time this migration executes.
+mod meta {
+    use sea_orm::entity::prelude::*;
+
+    #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)]
+    #[sea_orm(table_name = "meta")]
+    pub struct Model {
+        #[sea_orm(primary_key)]
+        pub id: i64,
+        pub module: String,
+        pub key1: String,
+        pub key2: String,
+        pub start_dt: i64,
+        #[sea_orm(column_type = "Text")]
+        pub value: String,
+    }
+
+    #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+    pub enum Relation {}
+
+    impl ActiveModelBehavior for ActiveModel {}
+}

src/infra/src/table/migration/mod.rs

@@ -24,6 +24,7 @@ mod m20241119_000002_populate_dashboards_table;
 mod m20241119_000003_delete_metas;
 mod m20241204_143100_create_table_search_queue;
 mod m20241209_120000_create_alerts_table;
+mod m20241215_190333_delete_metas;
 
 pub struct Migrator;
 
@@ -39,7 +40,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20241119_000002_populate_dashboards_table::Migration),
             Box::new(m20241119_000003_delete_metas::Migration),
             Box::new(m20241204_143100_create_table_search_queue::Migration),
-            Box::new(m20241209_120000_create_alerts_table::Migration),
+            Box::new(m20241215_190333_delete_metas::Migration),
         ]
     }
 }

src/service/db/short_url.rs

@@ -50,14 +50,16 @@ pub async fn set(short_id: &str, entry: short_urls::ShortUrlRecord) -> Result<()
         return Err(e).context("Failed to add short URL to DB");
     }
 
-    // trigger watch event
-    db::put(
-        &format!("{SHORT_URL_KEY}{short_id}"),
-        Bytes::new(),
-        NEED_WATCH,
-        None,
-    )
-    .await?;
+    // trigger watch event by putting empty value to cluster coordinator
+    let cluster_coordinator = super::get_coordinator().await;
+    cluster_coordinator
+        .put(
+            &format!("{SHORT_URL_KEY}{short_id}"),
+            Bytes::new(),
+            NEED_WATCH,
+            None,
+        )
+        .await?;
 
     Ok(())
 }
@@ -153,14 +155,16 @@ pub async fn gc_cache(retention_period_minutes: i64) -> Result<(), anyhow::Error
             short_urls::batch_remove(expired_short_ids.clone()).await?;
 
             // delete from cache
+            let cluster_coordinator = super::get_coordinator().await;
             for short_id in expired_short_ids {
-                db::delete(
-                    &format!("{SHORT_URL_KEY}{short_id}"),
-                    false,
-                    db::NEED_WATCH,
-                    None,
-                )
-                .await?;
+                cluster_coordinator
+                    .delete(
+                        &format!("{SHORT_URL_KEY}{short_id}"),
+                        false,
+                        NEED_WATCH,
+                        None,
+                    )
+                    .await?;
             }
         }
     }