sensor

var SensorCmd = &cobra.Command{
	Use:   "sensor [nodes file]",
	Short: "Start a devp2p sensor that discovers other peers and will receive blocks and transactions.",
	Long:  sensorUsage,
	Args:  cobra.MinimumNArgs(1),
	PreRunE: func(cmd *cobra.Command, args []string) (err error) {
		inputSensorParams.NodesFile = args[0]
		_, err = p2p.ReadNodeSet(inputSensorParams.NodesFile)
		if err != nil {
			log.Warn().Err(err).Msgf("Creating nodes file %v because it does not exist", inputSensorParams.NodesFile)
		}

		if len(inputSensorParams.StaticNodesFile) > 0 {
			inputSensorParams.staticNodes, err = p2p.ReadNodeSet(inputSensorParams.StaticNodesFile)
			if err != nil {
				log.Warn().Err(err).Msgf("Static nodes file %v not found", inputSensorParams.StaticNodesFile)
			}
		}

		if len(inputSensorParams.TrustedNodesFile) > 0 {
			inputSensorParams.trustedNodes, err = p2p.ReadNodeSet(inputSensorParams.TrustedNodesFile)
			if err != nil {
				log.Warn().Err(err).Msgf("Trusted nodes file %v not found", inputSensorParams.TrustedNodesFile)
			}
		}

		if len(inputSensorParams.Bootnodes) > 0 {
			inputSensorParams.bootnodes, err = p2p.ParseBootnodes(inputSensorParams.Bootnodes)
			if err != nil {
				return fmt.Errorf("unable to parse bootnodes: %w", err)
			}
		}

		if inputSensorParams.NetworkID == 0 {
			return errors.New("network ID must be greater than zero")
		}

		inputSensorParams.privateKey, err = crypto.GenerateKey()
		if err != nil {
			return err
		}

		if len(inputSensorParams.KeyFile) > 0 {
			var privateKey *ecdsa.PrivateKey
			privateKey, err = crypto.LoadECDSA(inputSensorParams.KeyFile)

			if err != nil {
				log.Warn().Err(err).Msg("Key file was not found, generating a new key file")

				err = crypto.SaveECDSA(inputSensorParams.KeyFile, inputSensorParams.privateKey)
				if err != nil {
					return err
				}
			} else {
				inputSensorParams.privateKey = privateKey
			}
		}

		if len(inputSensorParams.PrivateKey) > 0 {
			inputSensorParams.privateKey, err = crypto.HexToECDSA(inputSensorParams.PrivateKey)
			if err != nil {
				log.Error().Err(err).Msg("Failed to parse PrivateKey")
				return err
			}
		}

		inputSensorParams.nat, err = nat.Parse(inputSensorParams.NAT)
		if err != nil {
			log.Error().Err(err).Msg("Failed to parse NAT")
			return err
		}

		if inputSensorParams.ValidateBlockSigner &&
			(inputSensorParams.ShouldBroadcastBlocks || inputSensorParams.ShouldBroadcastBlockHashes) {
			if inputSensorParams.HeimdallURL == "" {
				return errors.New("--heimdall-url is required when --validate-block-signer is enabled with block broadcasting")
			}
			if inputSensorParams.ValidatorSetRefresh <= 0 {
				return errors.New("--validator-set-refresh must be greater than zero when --validate-block-signer is enabled with block broadcasting")
			}
		}

		if inputSensorParams.PeerSnapshotInterval <= 0 {
			return errors.New("--peer-snapshot-interval must be greater than zero")
		}
		if inputSensorParams.ClickHouseFlushInterval <= 0 {
			return errors.New("--clickhouse-flush-interval must be greater than zero")
		}

		return nil
	},
	RunE: func(cmd *cobra.Command, args []string) error {
		ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGINT, syscall.SIGTERM)
		defer stop()

		db, err := newDatabase(ctx)
		if err != nil {
			return err
		}

		defer func() {
			if cerr := db.Close(); cerr != nil {
				log.Error().Err(cerr).Msg("Failed to close database")
			}
		}()

		rpcBlock, err := getLatestBlock(inputSensorParams.RPC)
		if err != nil {
			return err
		}

		head := p2p.NewBlockPacket{
			Block: rpcBlock.ToBlock(),
			TD:    rpcBlock.TotalDifficulty.ToBigInt(),
		}

		peersGauge := p2p.NewPeersGauge()
		metrics := p2p.NewBlockMetrics(head.Block)

		// When block rebroadcasting is enabled and signer validation is on, load
		// the validator set from Heimdall so we only rebroadcast blocks signed
		// by a known validator. The initial fetch is blocking and fatal on
		// failure so we never rebroadcast against an empty validator set.
		var validators *p2p.ValidatorSet
		if inputSensorParams.ValidateBlockSigner &&
			(inputSensorParams.ShouldBroadcastBlocks || inputSensorParams.ShouldBroadcastBlockHashes) {
			validators = p2p.NewValidatorSet(inputSensorParams.HeimdallURL, inputSensorParams.ValidatorSetRefresh)
			if err = validators.Start(ctx); err != nil {
				return fmt.Errorf("failed to load validator set from %s: %w", inputSensorParams.HeimdallURL, err)
			}
		}

		// When transaction rebroadcasting is enabled, validate transactions
		// before forwarding them so peers sending junk cannot use the sensor to
		// amplify it. Everything received is still cached and persisted.
		var txValidator *p2p.TxValidator
		broadcastingTxs := inputSensorParams.ShouldBroadcastTx || inputSensorParams.ShouldBroadcastTxHashes
		if inputSensorParams.ValidateBroadcastTxs && broadcastingTxs {

			log.Info().
				Uint64("chain_id", inputSensorParams.NetworkID).
				Uint64("min_gas_price", inputSensorParams.BroadcastMinGasPrice).
				Uint64("min_tip", inputSensorParams.BroadcastMinTip).
				Msg("Validating transactions before rebroadcast, using the network ID as the chain ID")

			txValidator, err = p2p.NewTxValidator(p2p.TxValidatorOptions{
				ChainID:     inputSensorParams.NetworkID,
				MinGasPrice: new(big.Int).SetUint64(inputSensorParams.BroadcastMinGasPrice),
				MinTip:      new(big.Int).SetUint64(inputSensorParams.BroadcastMinTip),
			})
			if err != nil {
				return fmt.Errorf("failed to create transaction validator: %w", err)
			}
		} else if !broadcastingTxs {
			log.Info().Msg("Transaction rebroadcasting is off, transaction validation flags have no effect")
		} else {
			log.Warn().Msg("Rebroadcasting transactions without validation, peers can use this sensor to amplify invalid transactions")
		}

		conns := p2p.NewConns(p2p.ConnsOptions{
			BlocksCache:                inputSensorParams.BlocksCache,
			TxsCache:                   inputSensorParams.TxsCache,
			KnownTxsBloom:              inputSensorParams.KnownTxsBloom,
			KnownBlocksMax:             inputSensorParams.KnownBlocksMax,
			Head:                       head,
			ShouldBroadcastTx:          inputSensorParams.ShouldBroadcastTx,
			ShouldBroadcastTxHashes:    inputSensorParams.ShouldBroadcastTxHashes,
			ShouldBroadcastBlocks:      inputSensorParams.ShouldBroadcastBlocks,
			ShouldBroadcastBlockHashes: inputSensorParams.ShouldBroadcastBlockHashes,
			BroadcastWorkers:           inputSensorParams.BroadcastWorkers,
			TxBatchTimeout:             inputSensorParams.TxBatchTimeout,
			TxBroadcastQueueSize:       inputSensorParams.TxBroadcastQueueSize,
			MaxTxPacketSize:            inputSensorParams.MaxTxPacketSize,
			MaxQueuedTxs:               inputSensorParams.MaxQueuedTxs,
			ValidatorSet:               validators,
			CacheOnlyValidatedBlocks:   inputSensorParams.CacheOnlyValidatedBlocks,
			TxValidator:                txValidator,
		})

		opts := p2p.EthProtocolOptions{
			Context:                    ctx,
			Database:                   db,
			GenesisHash:                common.HexToHash(inputSensorParams.GenesisHash),
			RPC:                        inputSensorParams.RPC,
			SensorID:                   inputSensorParams.SensorID,
			NetworkID:                  inputSensorParams.NetworkID,
			Conns:                      conns,
			ForkID:                     forkid.ID{Hash: [4]byte(inputSensorParams.ForkID)},
			RequestsCache:              inputSensorParams.RequestsCache,
			ParentsCache:               inputSensorParams.ParentsCache,
			ShouldBroadcastTx:          inputSensorParams.ShouldBroadcastTx,
			ShouldBroadcastTxHashes:    inputSensorParams.ShouldBroadcastTxHashes,
			ShouldBroadcastBlocks:      inputSensorParams.ShouldBroadcastBlocks,
			ShouldBroadcastBlockHashes: inputSensorParams.ShouldBroadcastBlockHashes,
		}

		config := ethp2p.Config{
			PrivateKey:     inputSensorParams.privateKey,
			BootstrapNodes: inputSensorParams.bootnodes,
			StaticNodes:    inputSensorParams.staticNodes,
			TrustedNodes:   inputSensorParams.trustedNodes,
			MaxPeers:       inputSensorParams.MaxPeers,
			ListenAddr:     fmt.Sprintf(":%d", inputSensorParams.Port),
			DiscAddr:       fmt.Sprintf(":%d", inputSensorParams.DiscoveryPort),
			DialRatio:      inputSensorParams.DialRatio,
			NAT:            inputSensorParams.nat,
			DiscoveryV4:    !inputSensorParams.NoDiscovery,
			DiscoveryV5:    !inputSensorParams.NoDiscovery,
			Protocols: []ethp2p.Protocol{
				p2p.NewEthProtocol(66, opts),
				p2p.NewEthProtocol(67, opts),
				p2p.NewEthProtocol(68, opts),
				p2p.NewEthProtocol(69, opts),
			},
		}

		server := ethp2p.Server{Config: config}

		log.Info().Str("enode", server.Self().URLv4()).Msg("Starting sensor")

		if err = server.Start(); err != nil {
			return err
		}
		defer stopServer(&server)
		defer conns.Close()

		events := make(chan *ethp2p.PeerEvent)
		sub := server.SubscribeEvents(events)
		defer sub.Unsubscribe()

		ticker := time.NewTicker(2 * time.Second)
		defer ticker.Stop()

		peerSnapshotTicker := time.NewTicker(inputSensorParams.PeerSnapshotInterval)
		defer peerSnapshotTicker.Stop()

		if inputSensorParams.ShouldRunPprof {
			go handlePprof()
		}

		if inputSensorParams.ShouldRunPrometheus {
			go handlePrometheus()
		}

		go handleAPI(&server, conns)
		go handleRPC(conns, inputSensorParams.NetworkID)
		go handleDNSDiscovery(&server)

		for {
			select {
			case <-ticker.C:
				peersGauge.Set(float64(server.PeerCount()))
				metrics.Update(conns.HeadBlock().Block, conns.OldestBlock())
				writePeers(server.Peers())
			case <-peerSnapshotTicker.C:
				db.WritePeers(ctx, server.Peers(), time.Now())
			case <-ctx.Done():
				log.Info().Msg("Stopping sensor")
				return nil
			case event := <-events:
				log.Debug().Any("event", event).Send()
			case err := <-sub.Err():
				log.Error().Err(err).Send()
			}
		}
	},
}