txnotify

README

txnotify

txnotify is an Ethereum transaction observer that tracks incoming and outgoing transactions for subscribed addresses using Ethereum's JSON-RPC API. It pushes real-time updates to clients via WebSockets, making it suitable for integration with notification systems or wallet UIs.

It includes both a CLI debugging tool, a simple websockets client/server, as well as a Vue SPA that embeds it using WASM.

Check out the app here: aalbacetef.github.io/txnotify

Architecture

Components

• Watcher: Core engine that polls new blocks, fetches transactions, and notifies clients.
• RPC Client: Low-level JSON-RPC interface for Ethereum endpoints.
• Cache: In-memory store for blocks, transactions, and processing state. Can be easily extended to any data storage backend.
• Notifier: Interface for pushing updates to subscribers (WebSockets implementation included).
• CLI / Server: Commands to run the observer as a server or test client.

Data Flow

1. Watcher polls latest block using JSON-RPC.
2. If a new block exists, it is fetched and stored in cache.
3. Transactions from the block are filtered and matched against subscribed addresses.
4. For matching transactions, the notifier sends data to connected clients.

Usage

We'll use two addresses as examples:

• USDT: 0xdAC17F958D2ee523a2206206994597C13D831ec7
• USDC: 0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48

You can see them on etherscan.io

WebSockets

Run the WebSocket server:

go run ./cmd/server

## if you prefer pretty printed output 

go run ./cmd/server | jq 

Run a test client:

go run ./cmd/client --addresses 0xdAC17F958D2ee523a2206206994597C13D831ec7,0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48

Note these are the addresses for USDT and USDC.

CLI tool for watching txs

Run the block watcher:

go run ./cmd/watch --address 0xdAC17F958D2ee523a2206206994597C13D831ec7

## if you prefer pretty printed output 

go run ./cmd/watch --address 0xdAC17F958D2ee523a2206206994597C13D831ec7 | jq

Roadmap

• Add persistent storage backend (e.g: file or Redis)
• Improve reprocessing of failed or incomplete blocks
• Move a lot of the processing to a separate job system/message queue
• Enrich notifications with detailed transaction data
• Extend CLI for manual queries (e.g., list txs for address)

Known Limitations

• Could potentially retry blocks continuously
• In-memory cache only; restarts clear state
• Clients will timeout if no messages come in

Documentation

Index

• Variables
• type Cache
• type Config
• type InMemoryCache
  • func NewInMemoryCache() *InMemoryCache
  • func (cache *InMemoryCache) AddBlock(blockNum string, block ethereum.Block) error
  • func (cache *InMemoryCache) AddTx(tx ethereum.Transaction) error
  • func (cache *InMemoryCache) GetBlock(blockNum string) (ethereum.Block, error)
  • func (cache *InMemoryCache) GetBlockProcessed(blockNum string) (bool, error)
  • func (cache *InMemoryCache) GetTx(hash string) (ethereum.Transaction, error)
  • func (cache *InMemoryCache) SetBlockProcessed(blockNum string) error
  • func (cache *InMemoryCache) Subscribe(address string) error
  • func (cache *InMemoryCache) TxForAddress(address string) ([]ethereum.Transaction, error)
  • func (cache *InMemoryCache) Unsubscribe(address string) error
• type Notifier
• type RPCClient
• type State
• type TxNotFoundError
  • func (e TxNotFoundError) Error() string
• type Watcher
  • func NewWatcher(rpcEndpoint string, cfg Config, notifier Notifier) (*Watcher, error)
  • func (watcher *Watcher) Close() error
  • func (watcher *Watcher) Listen(backgroundCtx context.Context) error
  • func (watcher *Watcher) Subscribe(address string) error

Variables

var ErrBadFormat = errors.New("invalid string format, expected '0x' prefix")

Types

type Cache

type Cache interface {
	AddBlock(blockNum string, block ethereum.Block) error
	GetBlock(blockNum string) (ethereum.Block, error)
	GetBlockProcessed(blockNum string) (bool, error)
	SetBlockProcessed(blockNum string) error
	AddTx(tx ethereum.Transaction) error
	GetTx(hash string) (ethereum.Transaction, error)
	TxForAddress(address string) ([]ethereum.Transaction, error)
	Subscribe(address string) error
	Unsubscribe(address string) error
}

type Config

type Config struct {
	PollInterval time.Duration
	BatchSize    int
	BatchDelay   time.Duration
}

type InMemoryCache

type InMemoryCache struct {
	// contains filtered or unexported fields
}

func NewInMemoryCache

func NewInMemoryCache() *InMemoryCache

func (*InMemoryCache) AddBlock

func (cache *InMemoryCache) AddBlock(blockNum string, block ethereum.Block) error

func (*InMemoryCache) AddTx

func (cache *InMemoryCache) AddTx(tx ethereum.Transaction) error

func (*InMemoryCache) GetBlock

func (cache *InMemoryCache) GetBlock(blockNum string) (ethereum.Block, error)

func (*InMemoryCache) GetBlockProcessed

func (cache *InMemoryCache) GetBlockProcessed(blockNum string) (bool, error)

func (*InMemoryCache) GetTx

func (cache *InMemoryCache) GetTx(hash string) (ethereum.Transaction, error)

func (*InMemoryCache) SetBlockProcessed

func (cache *InMemoryCache) SetBlockProcessed(blockNum string) error

func (*InMemoryCache) Subscribe

func (cache *InMemoryCache) Subscribe(address string) error

func (*InMemoryCache) TxForAddress

func (cache *InMemoryCache) TxForAddress(address string) ([]ethereum.Transaction, error)

func (*InMemoryCache) Unsubscribe

func (cache *InMemoryCache) Unsubscribe(address string) error

type Notifier

type Notifier interface {
	Notify(address string, txList []ethereum.Transaction)
}

type RPCClient

type RPCClient interface {
	GetBlockByNumber(blockNum string) (*rpc.Response[ethereum.Block], error)
	GetCurrentBlockNumber() (*rpc.Response[string], error)
}

type State

type State struct {
	// contains filtered or unexported fields
}

type TxNotFoundError

type TxNotFoundError struct {
	// contains filtered or unexported fields
}

func (TxNotFoundError) Error

func (e TxNotFoundError) Error() string

type Watcher

type Watcher struct {
	// contains filtered or unexported fields
}

func NewWatcher

func NewWatcher(rpcEndpoint string, cfg Config, notifier Notifier) (*Watcher, error)

NewWatcher initializes a new Watcher instance with a JSON-RPC client, logger, in-memory cache, and notifier.

func (*Watcher) Close

func (watcher *Watcher) Close() error

func (*Watcher) Listen

func (watcher *Watcher) Listen(backgroundCtx context.Context) error

Listen starts the polling loop to watch for new Ethereum blocks and process transactions in real-time.

func (*Watcher) Subscribe

func (watcher *Watcher) Subscribe(address string) error

Subscribe registers a new address for monitoring by normalizing and adding it to the subscription list.