Merge branch 'master' into clean-up

flusher/flusher/db.py

@@ -184,6 +184,7 @@ def Column(*args, **kwargs):
     Column("resolve_time", sa.Integer, nullable=True),
     Column("result", CustomBase64, nullable=True),
     sa.Index("ix_requests_oracle_script_id", "oracle_script_id", "id"),
+    sa.Index("ix_oracle_script_id_resolve_status_request_time", "oracle_script_id", "resolve_status", "request_time"),
 )
 
 raw_requests = sa.Table(

flusher/flusher/handler.py

@@ -173,6 +173,8 @@ def handle_new_raw_report(self, msg):
         self.conn.execute(raw_reports.insert(), msg)
 
     def handle_set_validator(self, msg):
+        last_update = msg["last_update"]
+        del msg["last_update"]
         msg["account_id"] = self.get_account_id(msg["delegator_address"])
         del msg["delegator_address"]
         if self.get_validator_id(msg["operator_address"]) is None:
@@ -182,8 +184,24 @@ def handle_set_validator(self, msg):
             for col in validators.primary_key.columns.values():
                 condition = (col == msg[col.name]) & condition
             self.conn.execute(validators.update().where(condition).values(**msg))
+        self.handle_new_historical_bonded_token_on_validator(
+            {
+                "validator_id": self.get_validator_id(msg["operator_address"]),
+                "bonded_tokens": msg["tokens"],
+                "timestamp": last_update,
+            }
+        )
 
     def handle_update_validator(self, msg):
+        if "tokens" in msg and "last_update" in msg:
+            self.handle_new_historical_bonded_token_on_validator(
+                {
+                    "validator_id": self.get_validator_id(msg["operator_address"]),
+                    "bonded_tokens": msg["tokens"],
+                    "timestamp": msg["last_update"],
+                }
+            )
+            del msg["last_update"]
         self.conn.execute(
             validators.update().where(validators.c.operator_address == msg["operator_address"]).values(**msg)
         )
@@ -286,7 +304,7 @@ def handle_set_reporter(self, msg):
         self.conn.execute(reporters.insert(), msg)
 
     def handle_remove_reporter(self, msg):
-        msg["validator_id"] = self.get_validator_id(msg["validator"])
+        msg["operator_address"] = msg["validator"]
         del msg["validator"]
         msg["reporter_id"] = self.get_account_id(msg["reporter"])
         del msg["reporter"]
@@ -350,3 +368,10 @@ def increase_oracle_script_count(self, id):
                 count=oracle_script_requests.c.count + 1
             )
         )
+
+    def handle_new_historical_bonded_token_on_validator(self, msg):
+        self.conn.execute(
+            insert(historical_bonded_token_on_validators)
+            .values(**msg)
+            .on_conflict_do_update(constraint="historical_bonded_token_on_validators_pkey", set_=msg)
+        )

flusher/flusher/init.py

@@ -128,7 +128,7 @@ def init(chain_id, topic, replay_topic, db):
   FROM   oracle_scripts
          join requests
            ON oracle_scripts.id = requests.oracle_script_id
-  WHERE  TO_TIMESTAMP(requests.request_time) >= Now() - '1 day' :: interval
+  WHERE  requests.request_time >= CAST(EXTRACT(epoch FROM NOW()) AS INT) - 86400
   GROUP  BY oracle_scripts.id,
             requests.resolve_status;
         """
@@ -144,7 +144,7 @@ def init(chain_id, topic, replay_topic, db):
   FROM   oracle_scripts
          join requests
            ON oracle_scripts.id = requests.oracle_script_id
-  WHERE  TO_TIMESTAMP(requests.request_time) >= Now() - '1 week' :: interval
+  WHERE  requests.request_time >= CAST(EXTRACT(epoch FROM NOW()) AS INT) - 604800
   GROUP  BY oracle_scripts.id,
             requests.resolve_status;
 """
@@ -160,7 +160,7 @@ def init(chain_id, topic, replay_topic, db):
   FROM   oracle_scripts
          join requests
            ON oracle_scripts.id = requests.oracle_script_id
-  WHERE  TO_TIMESTAMP(requests.request_time) >= Now() - '1 month' :: interval
+  WHERE  requests.request_time >= CAST(EXTRACT(epoch FROM NOW()) AS INT) - 2592000
   GROUP  BY oracle_scripts.id,
             requests.resolve_status;
 """
@@ -221,4 +221,3 @@ def init(chain_id, topic, replay_topic, db):
 COMMIT;
 """
     )
-

go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/poy/onpar v1.1.2 // indirect
 	github.com/prometheus/client_golang v1.8.0
 	github.com/rakyll/statik v0.1.7
-	github.com/segmentio/kafka-go v0.4.9
+	github.com/segmentio/kafka-go v0.3.7
 	github.com/spf13/cast v1.3.1
 	github.com/spf13/cobra v1.1.1
 	github.com/spf13/viper v1.7.1

go.sum

@@ -745,8 +745,8 @@ github.com/sasha-s/go-deadlock v0.2.1-0.20190427202633-1595213edefa h1:0U2s5loxr
 github.com/sasha-s/go-deadlock v0.2.1-0.20190427202633-1595213edefa/go.mod h1:F73l+cr82YSh10GxyRI6qZiCgK64VaZjwesgfQ1/iLM=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
 github.com/seccomp/libseccomp-golang v0.9.1/go.mod h1:GbW5+tmTXfcxTToHLXlScSlAvWlF4P2Ca7zGrPiEpWo=
-github.com/segmentio/kafka-go v0.4.9 h1:cMjsu4BDGrqKJDRcFYdNWfwf/ziITVFPWOs1As3AOu8=
-github.com/segmentio/kafka-go v0.4.9/go.mod h1:BVDwBTF24avtlj4l8/xsWNb4papVeg16+jO6/0qjvhA=
+github.com/segmentio/kafka-go v0.3.7 h1:UCFPJw6KoVkmrilA2LbWVuybJojHzj6gDDFdV7H7IBs=
+github.com/segmentio/kafka-go v0.3.7/go.mod h1:8rEphJEczp+yDE/R5vwmaqZgF1wllrl4ioQcNKB8wVA=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=

yoda/handler.go

@@ -233,7 +233,7 @@ func handleRawRequest(c *Context, l *Logger, req rawRequest, key keyring.Info, i
 		return
 	}
 
-	vmsg := NewVerificationMessage(cfg.ChainID, c.validator, id, req.externalID)
+	vmsg := types.NewRequestVerification(cfg.ChainID, c.validator, id, req.externalID)
 	sig, pubkey, err := kb.Sign(key.GetName(), vmsg.GetSignBytes())
 	if err != nil {
 		l.Error(":skull: Failed to sign verify message: %s", c, err.Error())
@@ -246,7 +246,7 @@ func handleRawRequest(c *Context, l *Logger, req rawRequest, key keyring.Info, i
 
 	result, err := c.executor.Exec(exec, req.calldata, map[string]interface{}{
 		"BAND_CHAIN_ID":    vmsg.ChainID,
-		"BAND_VALIDATOR":   vmsg.Validator.String(),
+		"BAND_VALIDATOR":   vmsg.Validator,
 		"BAND_REQUEST_ID":  strconv.Itoa(int(vmsg.RequestID)),
 		"BAND_EXTERNAL_ID": strconv.Itoa(int(vmsg.ExternalID)),
 		"BAND_REPORTER":    sdk.MustBech32ifyPubKey(sdk.Bech32PubKeyTypeAccPub, pubkey),