hotfix: Emit Kafka msg for staking genesis state.

CHANGELOG_UNRELEASED.md

@@ -20,6 +20,7 @@
 
 ### Emitter & Flusher
 
+- (impv) [\#2240](https://github.com/bandprotocol/bandchain/pull/2240) Emit Kafka msg for staking genesis state.
 - (feat) [\#2238](https://github.com/bandprotocol/bandchain/pull/2238) Implement handle MsgSubmitProposal for emitter and flusher.
 
 ### Scan

chain/emitter/app.go

@@ -107,7 +107,7 @@ func (app *App) InitChain(req abci.RequestInitChain) abci.ResponseInitChain {
 			"balance": app.BankKeeper.GetCoins(app.DeliverContext, account.GetAddress()).String(),
 		})
 	}
-	// Staking module
+	// GenUtil module for create validator genesis transactions.
 	var genutilState genutil.GenesisState
 	app.Codec().MustUnmarshalJSON(genesisState[genutil.ModuleName], &genutilState)
 	for _, genTx := range genutilState.GenTxs {
@@ -120,6 +120,40 @@ func (app *App) InitChain(req abci.RequestInitChain) abci.ResponseInitChain {
 		}
 	}
 
+	// Staking module
+	var stakingState staking.GenesisState
+	app.Codec().MustUnmarshalJSON(genesisState[staking.ModuleName], &stakingState)
+	for _, val := range stakingState.Validators {
+		app.emitSetValidator(val.OperatorAddress)
+	}
+
+	for _, del := range stakingState.Delegations {
+		app.emitDelegation(del.ValidatorAddress, del.DelegatorAddress)
+	}
+
+	for _, unbonding := range stakingState.UnbondingDelegations {
+		for _, entry := range unbonding.Entries {
+			app.Write("NEW_UNBONDING_DELEGATION", JsDict{
+				"delegator_address": unbonding.DelegatorAddress,
+				"operator_address":  unbonding.ValidatorAddress,
+				"completion_time":   entry.CompletionTime.UnixNano(),
+				"amount":            entry.Balance,
+			})
+		}
+	}
+
+	for _, redelegate := range stakingState.Redelegations {
+		for _, entry := range redelegate.Entries {
+			app.Write("NEW_REDELEGATION", JsDict{
+				"delegator_address":    redelegate.DelegatorAddress,
+				"operator_src_address": redelegate.ValidatorSrcAddress,
+				"operator_dst_address": redelegate.ValidatorDstAddress,
+				"completion_time":      entry.CompletionTime.UnixNano(),
+				"amount":               entry.InitialBalance,
+			})
+		}
+	}
+
 	// Oracle module
 	var oracleState oracle.GenesisState
 	app.Codec().MustUnmarshalJSON(genesisState[oracle.ModuleName], &oracleState)

flusher/flusher/db.py

@@ -244,7 +244,7 @@ def Column(*args, **kwargs):
     metadata,
     Column("delegator_id", sa.Integer, sa.ForeignKey("accounts.id")),
     Column("validator_id", sa.Integer, sa.ForeignKey("validators.id")),
-    Column("creation_height", sa.Integer, sa.ForeignKey("blocks.height")),
+    Column("creation_height", sa.Integer, sa.ForeignKey("blocks.height"), nullable=True),
     Column("completion_time", CustomDateTime),
     Column("amount", sa.DECIMAL),
 )