cdb: implement data_source_requests table

CHANGELOG_UNRELEASED.md

@@ -16,14 +16,15 @@
 
 ### Emitter & Flusher
 
+- (impv) [\#2549](https://github.com/bandprotocol/bandchain/pull/2549) Implemented `data_source_requests` table
 - (feat) [\#2551](https://github.com/bandprotocol/bandchain/pull/2551) fast-sync: add flag enable fast sync and emit all account and validator
 
 ### Scan
 
 - (impv) [\#2557](https://github.com/bandprotocol/bandchain/pull/2557) Fix and clean up copy button
 - (impv) [\#2552](https://github.com/bandprotocol/bandchain/pull/2552) Wire up related data source, handle nullable timestamp
 - (impv) [\#2550](https://github.com/bandprotocol/bandchain/pull/2550) Polish request index page
-- (bugs) [\#2527](https://github.com/bandprotocol/bandchain/pull/2527) Fix serach bugs on ds & os home page
+- (bugs) [\#2527](https://github.com/bandprotocol/bandchain/pull/2527) Fix search bugs on ds & os home page
 - (impv) [\#2516](https://github.com/bandprotocol/bandchain/pull/2516) Fix loading width and bg color
 - (impv) [\#2499](https://github.com/bandprotocol/bandchain/pull/2499) Implemented Total request chart
 - (impv) [\#2455](https://github.com/bandprotocol/bandchain/pull/2455) Implement full copy non evm proof

chain/emitter/app.go

@@ -163,7 +163,7 @@ func (app *App) InitChain(req abci.RequestInitChain) abci.ResponseInitChain {
 		app.emitSetDataSource(types.DataSourceID(idx+1), ds, nil)
 	}
 	for idx, os := range oracleState.OracleScripts {
-		app.emitNewOracleScript(types.OracleScriptID(idx+1), os, nil)
+		app.emitSetOracleScript(types.OracleScriptID(idx+1), os, nil)
 	}
 	app.FlushMessages()
 	return res

chain/emitter/oracle.go

@@ -25,19 +25,6 @@ func (app *App) emitSetDataSource(id types.DataSourceID, ds types.DataSource, tx
 	})
 }
 
-func (app *App) emitNewOracleScript(id types.OracleScriptID, os types.OracleScript, txHash []byte) {
-	app.Write("NEW_ORACLE_SCRIPT", JsDict{
-		"id":              id,
-		"name":            os.Name,
-		"description":     os.Description,
-		"owner":           os.Owner.String(),
-		"schema":          os.Schema,
-		"codehash":        os.Filename,
-		"source_code_url": os.SourceCodeURL,
-		"tx_hash":         txHash,
-	})
-}
-
 func (app *App) emitSetOracleScript(id types.OracleScriptID, os types.OracleScript, txHash []byte) {
 	app.Write("SET_ORACLE_SCRIPT", JsDict{
 		"id":              id,
@@ -135,7 +122,7 @@ func (app *App) handleMsgCreateOracleScript(
 ) {
 	id := types.OracleScriptID(atoi(evMap[types.EventTypeCreateOracleScript+"."+types.AttributeKeyID][0]))
 	os := app.BandApp.OracleKeeper.MustGetOracleScript(app.DeliverContext, id)
-	app.emitNewOracleScript(id, os, txHash)
+	app.emitSetOracleScript(id, os, txHash)
 	extra["id"] = id
 }
 

flusher/flusher/db.py

@@ -360,6 +360,13 @@ def Column(*args, **kwargs):
     Column("timestamp", CustomDateTime, primary_key=True),
 )
 
+data_source_requests = sa.Table(
+    "data_source_requests",
+    metadata,
+    Column("data_source_id", sa.Integer, sa.ForeignKey("data_sources.id"), primary_key=True),
+    Column("count", sa.Integer),
+)
+
 oracle_script_requests = sa.Table(
     "oracle_script_requests",
     metadata,

flusher/flusher/handler.py

@@ -27,6 +27,7 @@
     reporters,
     related_data_source_oracle_scripts,
     historical_oracle_statuses,
+    data_source_requests,
     oracle_script_requests,
     request_count_per_days,
 )
@@ -90,10 +91,11 @@ def handle_set_data_source(self, msg):
             .values(**msg)
             .on_conflict_do_update(constraint="data_sources_pkey", set_=msg)
         )
-
-    def handle_new_oracle_script(self, msg):
-        self.handle_set_oracle_script(msg)
-        self.handle_new_oracle_script_request({"oracle_script_id": msg["id"], "count": 0})
+        self.conn.execute(
+            insert(data_source_requests)
+            .values({"data_source_id": msg["id"], "count": 0})
+            .on_conflict_do_nothing(constraint="data_source_requests_pkey")
+        )
 
     def handle_set_oracle_script(self, msg):
         if msg["tx_hash"] is not None:
@@ -106,6 +108,11 @@ def handle_set_oracle_script(self, msg):
             .values(**msg)
             .on_conflict_do_update(constraint="oracle_scripts_pkey", set_=msg)
         )
+        self.conn.execute(
+            insert(oracle_script_requests)
+            .values({"oracle_script_id": msg["id"], "count": 0})
+            .on_conflict_do_nothing(constraint="oracle_script_requests_pkey")
+        )
 
     def handle_new_request(self, msg):
         msg["transaction_id"] = self.get_transaction_id(msg["tx_hash"])
@@ -138,6 +145,7 @@ def handle_new_raw_request(self, msg):
             }
         )
         self.conn.execute(raw_requests.insert(), msg)
+        self.handle_set_data_source_request({"data_source_id": msg["data_source_id"]})
 
     def handle_new_val_request(self, msg):
         msg["validator_id"] = self.get_validator_id(msg["validator"])
@@ -309,8 +317,13 @@ def handle_set_historical_validator_status(self, msg):
             .on_conflict_do_update(constraint="historical_oracle_statuses_pkey", set_=msg)
         )
 
-    def handle_new_oracle_script_request(self, msg):
-        self.conn.execute(oracle_script_requests.insert(), msg)
+    def handle_set_data_source_request(self, msg):
+        condition = True
+        for col in data_source_requests.primary_key.columns.values():
+            condition = (col == msg[col.name]) & condition
+        self.conn.execute(
+            data_source_requests.update(condition).values(count=data_source_requests.c.count + 1)
+        )
 
     def handle_set_oracle_script_request(self, msg):
         condition = True