func (s *Service) StartBlockListener() error {
	ch := make(chan *types.Header, 100)
	sub, err := s.ew.BC().SubscribeNewHead(s.ctx, ch)
	if err != nil {
		s.logger.Error("failed to create new head subscription", zap.Error(err))
		close(ch)
		return err
	}
	for {
		select {
		case <-s.ctx.Done():
			sub.Unsubscribe()
			close(ch)
			return nil
		case err := <-sub.Err():
			s.logger.Error("new head subscription error received", zap.Error(err))
			sub.Unsubscribe()
			close(ch)
			return err
		case header := <-ch:
			s.logger.Debug("new block header received", zap.Any("header", header))

			// do a single update for wait group incrementation instead of multiple times
			s.wg.Add(len(s.pools))

			for _, pool := range s.pools {

				// claim the named lock
				s.namedLock.Lock(pool.Name)

				// start the total supply update information in the background
				go func(pool config.Pool) {

					// defer named lock unlock
					defer s.namedLock.Unlock(pool.Name)
					defer s.wg.Done()

					// grab the previous pool information so we can determine if we should break the circuit
					dPool, err := s.db.GetPool(pool.Name)
					if err != nil {
						s.logger.Error("failed to get pool database entry", zap.Error(err), zap.String("pool", pool.Name))
						return
					}
					// get the information for the time this record was last updated at
					infos, err := s.db.GetInfo(dPool.Name, dPool.LastUpdateAt)
					if err != nil {
						s.logger.Error("failed to get pool infos database entry", zap.Error(err), zap.String("pool", pool.Name), zap.Uint64("updated.at", dPool.LastUpdateAt))
						return
					}

					// check the block number from the header
					blockNum := header.Number.Uint64()

					s.logger.Info("updating pool information (supply, balances, weights)", zap.String("pool", pool.Name), zap.Uint64("block.num", blockNum))

					// create bindings for the given contract
					contract, err := sigmacore.NewSigmacore(
						common.HexToAddress(pool.ContractAddress),
						s.ew.BC(),
					)
					if err != nil {
						s.logger.Error("failed to get contract binding", zap.Error(err), zap.String("pool", pool.Name))
						return
					}
					// update the information stored in the database
					_, _, totalSupplies, err := s.UpdateBalancesWeightsAndSupplies(blockNum, pool, contract)
					if err != nil {
						s.logger.Error("UpdateBalancesWeightsAndSupplies failed", zap.Error(err), zap.String("pool", pool.Name))
					}
					controllerAddress, err := contract.GetController(nil)
					if err != nil {
						s.logger.Error("failed to get controller address", zap.Error(err), zap.String("pool", pool.Name))
						return
					}
					// now start checking total supply shifts
					for tok, supply := range infos.TokenTotalSupplies {
						conContract, err := controller.NewController(controllerAddress, s.ew.BC())
						if err != nil {
							s.logger.Error("failed to get controller contract", zap.Error(err), zap.String("pool", pool.Name), zap.String("token", tok))
						} else {
							if err := s.circuitBreakCheck(tok, supply, totalSupplies, pool, conContract); err != nil {
								s.logger.Error("circuitBreakCheck failed", zap.Error(err), zap.String("pool", pool.Name), zap.String("token", tok))
							}
						}
					}
					// purge old records if need be
					s.purgeInfoCheck(pool.Name)

				}(pool)
			}
		}
	}
}
// StartBlockListener waits for new blocks to be mined to trigger blockchain db updates
StartBlockListener waits for new blocks to be mined to trigger blockchain db updates
// StartBlockListener waits for new blocks to be mined to trigger blockchain db updates
func (s *Service) StartBlockListener() error {
	ch := make(chan *types.Header, 100)
	sub, err := s.ew.BC().SubscribeNewHead(s.ctx, ch)
	if err != nil {
		s.logger.Error("failed to create new head subscription", zap.Error(err))
		close(ch)
		return err
	}
	for {
		select {
		case <-s.ctx.Done():
			sub.Unsubscribe()
			close(ch)
			return nil
		case err := <-sub.Err():
			s.logger.Error("new head subscription error received", zap.Error(err))
			sub.Unsubscribe()
			close(ch)
			return err
		case header := <-ch:
			s.logger.Debug("new block header received", zap.Any("header", header))
			s.wg.Add(len(s.pools))
			for _, pool := range s.pools {
				s.namedLock.Lock(pool.Name)
				go func(pool config.Pool) {
					defer s.namedLock.Unlock(pool.Name)
					defer s.wg.Done()
					dPool, err := s.db.GetPool(pool.Name)
					if err != nil {
						s.logger.Error("failed to get pool database entry", zap.Error(err), zap.String("pool", pool.Name))
						return
					}
					infos, err := s.db.GetInfo(dPool.Name, dPool.LastUpdateAt)
					if err != nil {
						s.logger.Error("failed to get pool infos database entry", zap.Error(err), zap.String("pool", pool.Name), zap.Uint64("updated.at", dPool.LastUpdateAt))
						return
					}
					blockNum := header.Number.Uint64()
					s.logger.Info("updating pool information (supply, balances, weights)", zap.String("pool", pool.Name), zap.Uint64("block.num", blockNum))
					contract, err := sigmacore.NewSigmacore(
						common.HexToAddress(pool.ContractAddress),
						s.ew.BC(),
					)
					if err != nil {
						s.logger.Error("failed to get contract binding", zap.Error(err), zap.String("pool", pool.Name))
						return
					}
					_, _, totalSupplies, err := s.UpdateBalancesWeightsAndSupplies(blockNum, pool, contract)
					if err != nil {
						s.logger.Error("UpdateBalancesWeightsAndSupplies failed", zap.Error(err), zap.String("pool", pool.Name))
					}
					controllerAddress, err := contract.GetController(nil)
					if err != nil {
						s.logger.Error("failed to get controller address", zap.Error(err), zap.String("pool", pool.Name))
						return
					}
					for tok, supply := range infos.TokenTotalSupplies {
						conContract, err := controller.NewController(controllerAddress, s.ew.BC())
						if err != nil {
							s.logger.Error("failed to get controller contract", zap.Error(err), zap.String("pool", pool.Name), zap.String("token", tok))
						} else {
							if err := s.circuitBreakCheck(tok, supply, totalSupplies, pool, conContract); err != nil {
								s.logger.Error("circuitBreakCheck failed", zap.Error(err), zap.String("pool", pool.Name), zap.String("token", tok))
							}
						}
					}
					s.purgeInfoCheck(pool.Name)
				}(pool)
			}
		}
	}
}
func ( s * Service ) StartBlockListener ( ) error { ch := make ( chan * types . Header , 100 ) sub , err := s . ew . BC ( ) . SubscribeNewHead ( s . ctx , ch ) if err != nil { s . logger . Error ( " " , zap . Error ( err ) ) close ( ch ) return err } for { select { case <- s . ctx . Done ( ) : sub . Unsubscribe ( ) close ( ch ) return nil case err := <- sub . Err ( ) : s . logger . Error ( " " , zap . Error ( err ) ) sub . Unsubscribe ( ) close ( ch ) return err case header := <- ch : s . logger . Debug ( " " , zap . Any ( " " , header ) ) s . wg . Add ( len ( s . pools ) ) for _ , pool := range s . pools { s . namedLock . Lock ( pool . Name ) go func ( pool config . Pool ) { defer s . namedLock . Unlock ( pool . Name ) defer s . wg . Done ( ) dPool , err := s . db . GetPool ( pool . Name ) if err != nil { s . logger . Error ( " " , zap . Error ( err ) , zap . String ( " " , pool . Name ) ) return } infos , err := s . db . GetInfo ( dPool . Name , dPool . LastUpdateAt ) if err != nil { s . logger . Error ( " " , zap . Error ( err ) , zap . String ( " " , pool . Name ) , zap . Uint64 ( " " , dPool . LastUpdateAt ) ) return } blockNum := header . Number . Uint64 ( ) s . logger . Info ( " " , zap . String ( " " , pool . Name ) , zap . Uint64 ( " " , blockNum ) ) contract , err := sigmacore . NewSigmacore ( common . HexToAddress ( pool . ContractAddress ) , s . ew . BC ( ) , ) if err != nil { s . logger . Error ( " " , zap . Error ( err ) , zap . String ( " " , pool . Name ) ) return } _ , _ , totalSupplies , err := s . UpdateBalancesWeightsAndSupplies ( blockNum , pool , contract ) if err != nil { s . logger . Error ( " " , zap . Error ( err ) , zap . String ( " " , pool . Name ) ) } controllerAddress , err := contract . GetController ( nil ) if err != nil { s . logger . Error ( " " , zap . Error ( err ) , zap . String ( " " , pool . Name ) ) return } for tok , supply := range infos . TokenTotalSupplies { conContract , err := controller . NewController ( controllerAddress , s . ew . BC ( ) ) if err != nil { s . logger . Error ( " " , zap . Error ( err ) , zap . String ( " " , pool . Name ) , zap . String ( " " , tok ) ) } else { if err := s . circuitBreakCheck ( tok , supply , totalSupplies , pool , conContract ) ; err != nil { s . logger . Error ( " " , zap . Error ( err ) , zap . String ( " " , pool . Name ) , zap . String ( " " , tok ) ) } } } s . purgeInfoCheck ( pool . Name ) } ( pool ) } } } }
StartBlockListener waits for new blocks to be mined to trigger blockchain db updates
StartBlockListener waits for new blocks to be mined to trigger blockchain db updates
// do a single update for wait group incrementation instead of multiple times // claim the named lock // start the total supply update information in the background // defer named lock unlock // grab the previous pool information so we can determine if we should break the circuit // get the information for the time this record was last updated at // check the block number from the header // create bindings for the given contract // update the information stored in the database // now start checking total supply shifts // purge old records if need be