Zum Hauptinhalt springen

Streams Filter Examples (Go & JavaScript)

Aktualisiert am
Aug 18, 2026

This page provides ready-to-use filter examples for common use cases. Copy and modify these examples to match your specific needs.

Before You Start​

Every filter function must follow these rules:


  • Name: Your function must be named Filter (Go) or main (JavaScript)
  • Return value: Return data to deliver it, or return nil/null to skip delivery for that block
  • Input: The function receives a payload with .Data (Go) / .data (JavaScript) and .Metadata / .metadata

Note that .Data / .data is always an Array, not a single object.

Batching. stream.data holds one entry per block delivered. dataset_batch_size controls how many (default 1), so stream.data.length is 1 unless you raise it. Examples on this page iterate stream.data, which stays correct at any batch size — indexing stream.data[0] would silently process only the first block of each batch and drop the rest, with no error.


type Payload struct {
  Data     []interface{}          `json:"data"`
  Metadata map[string]interface{} `json:"metadata"`
}

func Filter(qn *qn.QNContext, payload Payload) interface{} {
  // payload.Data contains the blockchain data
  // payload.Metadata contains stream information
  return payload.Data // Return data to deliver, or nil to skip
}

Basic Examples​

Extract Block Hash and Number​

Dataset: Block

Extracts the block hash and converts the block number from hex to decimal. This example reads only the first block of the batch (data[0]); see Chargenverarbeitung to handle a dataset_batch_size above 1.


type Payload struct {
	Data     []Block                `json:"data"`
	Metadata map[string]interface{} `json:"metadata"`
}
type Block struct {
	Hash   string `json:"hash"`
	Number string `json:"number"`
}

func Filter(qn *qn.QNContext, payload Payload) interface{} {
	numberDecimal, _ := strconv.ParseInt(strings.TrimPrefix(payload.Data[0].Number, "0x"), 16, 64)

	return map[string]interface{}{
		"hash":   payload.Data[0].Hash,
		"number": numberDecimal,
	}
}

Test block: 20999911 — returns number: 20999911

Filter ERC-20 Transfer Events​

Dataset: block_with_receipts

Finds all ERC-20 token transfers in a block. The Transfer event has a specific topic signature and exactly 3 topics (event signature + from + to addresses).


type Payload struct {
	Data     []Entry                `json:"data"`
	Metadata map[string]interface{} `json:"metadata"`
}
type Entry struct {
	// Keep receipts as raw JSON so the full original object is preserved —
	// only Receipt/Log below are decoded to evaluate the filter condition.
	Receipts []json.RawMessage `json:"receipts"`
}
type Receipt struct {
	Logs []Log `json:"logs"`
}
type Log struct {
	Topics []string `json:"topics"`
}

func Filter(qn *qn.QNContext, payload Payload) interface{} {
	const transferTopic = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"

	var filteredReceipts []json.RawMessage
	totalReceipts := 0
	for _, entry := range payload.Data {
		totalReceipts += len(entry.Receipts)
		for _, raw := range entry.Receipts {
			var receipt Receipt
			if err := json.Unmarshal(raw, &receipt); err != nil {
				continue
			}
			relevant := false
			for _, log := range receipt.Logs {
				if len(log.Topics) == 3 && log.Topics[0] == transferTopic {
					relevant = true
					break
				}
			}
			if relevant {
				filteredReceipts = append(filteredReceipts, raw)
			}
		}
	}

	if len(filteredReceipts) == 0 {
		return nil
	}

	return map[string]interface{}{
		"totalReceipts": totalReceipts,
		"filteredCount": len(filteredReceipts),
		"receipts":      filteredReceipts,
	}
}

Test block: 20999911 — returns totalReceipts: 206, filteredCount: 96

Monitor Specific Addresses​

Dataset: block_with_receipts

Tracks transactions and events involving a list of addresses. Useful for monitoring wallets or contracts.


type Payload struct {
	Data     []Entry                `json:"data"`
	Metadata map[string]interface{} `json:"metadata"`
}
type Entry struct {
	Block    Block     `json:"block"`
	Receipts []Receipt `json:"receipts"`
}
type Block struct {
	Transactions []Transaction `json:"transactions"`
}
type Transaction struct {
	From string `json:"from"`
	To   string `json:"to"`
}
type Receipt struct {
	Logs []Log `json:"logs"`
}
type Log struct {
	Topics []string `json:"topics"`
}

func Filter(qn *qn.QNContext, payload Payload) interface{} {
	addresses := []string{
		"0x21a31ee1afc51d94c2efccaa2092ad1028285549", // Binance 15
		"0xdfd5293d8e347dfe59e90efd55b2956a1343963d", // Binance 16
		"0x28c6c06298d514db089934071355e5743bf21d60", // Binance 14
	}

	addressSet := make(map[string]bool)
	paddedAddressSet := make(map[string]bool)
	for _, addr := range addresses {
		lower := strings.ToLower(addr)
		addressSet[lower] = true
		hex := lower[2:]
		paddedAddressSet["0x"+strings.Repeat("0", 64-len(hex))+hex] = true
	}

	var matchingTransactions []Transaction
	var matchingReceipts []Receipt

	for _, entry := range payload.Data {
		for _, tx := range entry.Block.Transactions {
			from := strings.ToLower(tx.From)
			to := strings.ToLower(tx.To)
			if (tx.From != "" && addressSet[from]) || (tx.To != "" && addressSet[to]) {
				matchingTransactions = append(matchingTransactions, tx)
			}
		}

		for _, receipt := range entry.Receipts {
			matches := false
			for _, log := range receipt.Logs {
				if len(log.Topics) > 1 && (paddedAddressSet[log.Topics[1]] ||
					(len(log.Topics) > 2 && paddedAddressSet[log.Topics[2]])) {
					matches = true
					break
				}
			}
			if matches {
				matchingReceipts = append(matchingReceipts, receipt)
			}
		}
	}

	if len(matchingTransactions) == 0 && len(matchingReceipts) == 0 {
		return nil
	}

	return map[string]interface{}{
		"transactions": matchingTransactions,
		"receipts":     matchingReceipts,
	}
}

Test block: 20999911 — returns 4 transactions and 3 receipts

Decoding EVM Events​

Die decodeEVMReceipts (JavaScript) / qn.DecodeEVMReceipts (Go) function transforms raw event logs into human-readable data using contract ABIs. This section covers the basics of decoding and then shows practical examples that combine decoding with filtering for common use cases like tracking token transfers, NFT sales, and DeFi activity.

How it works:

  1. Matches event signatures in logs with the provided ABIs
  2. Decodes parameters according to their types (addresses, integers, etc.)
  3. Returns structured data with named parameters in a decodedLogs Array

Parameters:

  • receipts — Array of transaction receipts to decode
  • abis — Array of contract ABIs (strings or objects)

Store ABIs Externally

Instead of hardcoding ABIs in your filter, you can store them in Key-Value Store for easier updates and reuse. See Storing Contract ABIs.

Decode ERC-20 Transfers​

Dataset: block_with_receipts

Decodes Transfer events into readable format with aus, zu, und Wert fields.


type Payload struct {
	Data     []Entry                `json:"data"`
	Metadata map[string]interface{} `json:"metadata"`
}
type Entry struct {
	Receipts []interface{} `json:"receipts"`
}

func Filter(qn *qn.QNContext, payload Payload) interface{} {
	const erc20ABI = `[{
    "anonymous": false,
    "inputs": [
      {"indexed": true, "type": "address", "name": "from"},
      {"indexed": true, "type": "address", "name": "to"},
      {"indexed": false, "type": "uint256", "name": "value"}
    ],
    "name": "Transfer",
    "type": "event"
  }]`

	// Collect receipts from every block in the batch
	var receipts []interface{}
	for _, entry := range payload.Data {
		receipts = append(receipts, entry.Receipts...)
	}
	decoded := qn.DecodeEVMReceipts(receipts, []string{erc20ABI})

	// Filter for receipts with decoded logs
	var result []interface{}
	for _, r := range decoded {
		receipt, ok := r.(map[string]interface{})
		if !ok {
			continue
		}
		if logs, ok := receipt["decodedLogs"].([]interface{}); ok && len(logs) > 0 {
			result = append(result, receipt)
		}
	}

	return map[string]interface{}{"result": result}
}

Test block: 20999911 — returns 96 decoded receipts

Decoded output example:

{
  "address": "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
  "name": "Transfer",
  "from": "0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640",
  "to": "0x6f1cdbbb4d53d226cf4b917bf768b94acbab6168",
  "value": "138908045566"
}

Track NFT Marketplace Sales (OpenSea)​

Dataset: block_with_receipts

Decodes OpenSea Seaport OrderFulfilled events to track NFT sales. Sales are sparse, so most blocks legitimately return no results — use the test block above to confirm the filter works.


type Payload struct {
	Data     []Entry                `json:"data"`
	Metadata map[string]interface{} `json:"metadata"`
}
type Entry struct {
	Receipts []interface{} `json:"receipts"`
}

func Filter(qn *qn.QNContext, payload Payload) interface{} {
	// OpenSea Seaport OrderFulfilled event
	// test on Ethereum, 21292520 block
	const seaportABI = `[{
    "anonymous": false,
    "inputs": [
      {"type": "bytes32", "name": "orderHash", "indexed": false},
      {"type": "address", "name": "offerer", "indexed": true},
      {"type": "address", "name": "zone", "indexed": true},
      {"type": "address", "name": "recipient", "indexed": false},
      {
        "components": [
          {"type": "uint8", "name": "itemType"},
          {"type": "address", "name": "token"},
          {"type": "uint256", "name": "identifier"},
          {"type": "uint256", "name": "amount"}
        ],
        "type": "tuple[]",
        "name": "offer",
        "indexed": false
      },
      {
        "components": [
          {"type": "uint8", "name": "itemType"},
          {"type": "address", "name": "token"},
          {"type": "uint256", "name": "identifier"},
          {"type": "uint256", "name": "amount"},
          {"type": "address", "name": "recipient"}
        ],
        "type": "tuple[]",
        "name": "consideration",
        "indexed": false
      }
    ],
    "name": "OrderFulfilled",
    "type": "event"
  }]`

	const seaportAddress = "0x00000000006c3852cbef3e08e8df289169ede581"

	// Collect receipts from every block in the batch
	var receipts []interface{}
	for _, entry := range payload.Data {
		receipts = append(receipts, entry.Receipts...)
	}
	decoded := qn.DecodeEVMReceipts(receipts, []string{seaportABI})

	var result []interface{}
	for _, r := range decoded {
		receipt, ok := r.(map[string]interface{})
		if !ok {
			continue
		}
		logs, ok := receipt["decodedLogs"].([]interface{})
		if !ok {
			continue
		}
		for _, l := range logs {
			logMap, ok := l.(map[string]interface{})
			if !ok {
				continue
			}
			address := strings.ToLower(fmt.Sprintf("%v", logMap["address"]))
			name := fmt.Sprintf("%v", logMap["name"])
			if address == seaportAddress && name == "OrderFulfilled" {
				result = append(result, receipt)
				break
			}
		}
	}

	return map[string]interface{}{"result": result}
}

Test block: 20999911 — returns 2 Seaport sales

Nächste Schritte​


Zuletzt aktualisiert auf