feat: add logs websocket subscription for live smart-contract events

data/interface.go

@@ -240,6 +240,8 @@ type LogHandler interface {
 	GetAddress() []byte
 	// GetLogEvents returns the events from a transaction log entry
 	GetLogEvents() []transaction.EventHandler
+	// GetContractID returns the numeric identifier of the contract that produced this log
+	GetContractID() int32
 
 	IsInterfaceNil() bool
 }

data/transaction/interface.go

@@ -32,6 +32,10 @@ type EventHandler interface {
 	// GetData returns the rest of the event data, which will not be indexed, so storing
 	//  information here should be cheaper
 	GetData() [][]byte
+	// GetIsSystemLog reports whether this event was generated by the node itself (e.g.
+	// an internal VM error) rather than by contract code — callers that only want
+	// genuine contract-emitted events should filter on this.
+	GetIsSystemLog() bool
 
 	IsInterfaceNil() bool
 }

indexer/data/prepared.go

@@ -6,4 +6,10 @@ type PreparedBlockData struct {
 	Txs     []*Transaction
 	TxsMap  map[string]*Transaction
 	Altered *AlteredData
+	// LogsResults and LogsDB, when set, are the websocket dispatcher's own
+	// ExtractDataFromLogs/PrepareLogsForDB results for this block, computed
+	// synchronously on the commit goroutine and reused by the elastic worker instead of
+	// recomputing them on its own goroutine — see eventsProcessor.SaveBlock.
+	LogsResults *PreparedLogsResults
+	LogsDB      []*Logs
 }

indexer/data/transaction.go

@@ -41,7 +41,11 @@ type Logs struct {
 	Caller     string        `json:"caller,omitempty"`
 	ContractID int32         `json:"contractId"`
 	Timestamp  time.Duration `json:"timestamp"`
-	Events     []*Event      `json:"events"`
+	// Status/ResultCode mirror the producing transaction's own fields, so a reverted/failed
+	// transaction's logs are distinguishable from a committed one's.
+	Status     string   `json:"status,omitempty"`
+	ResultCode string   `json:"resultCode,omitempty"`
+	Events     []*Event `json:"events"`
 }
 
 // Events represents the events generated by a transaction with changed fields
@@ -51,6 +55,11 @@ type Event struct {
 	Topics     []string `json:"topics"`
 	Data       []string `json:"data"`
 	Order      int      `json:"order"`
+	// IsSystemLog marks an event generated by the node itself (e.g. an internal VM
+	// error) rather than by contract code, so consumers of a live/public feed can filter
+	// out node-internal noise instead of it being indistinguishable from a real
+	// contract-emitted event.
+	IsSystemLog bool `json:"isSystemLog,omitempty"`
 }
 
 type KDAFee struct {

indexer/elasticProcessor.go

@@ -534,15 +534,24 @@ func (ei *elasticProcessor) SaveTransactions(
 ) error {
 	headerTimestamp := header.GetTimestamp()
 
-	txs, txsMap, ad, err := ei.resolvePreparedBlockData(header, pool, prepared)
+	txs, txsMap, ad, ld, logsDB, err := ei.resolvePreparedBlockData(header, pool, prepared)
 	if err != nil {
 		return err
 	}
 
-	ld := ei.logsAndEventsProc.ExtractDataFromLogs(pool, txs, headerTimestamp)
+	// Non-nil only when the websocket dispatcher already computed them synchronously on
+	// the commit goroutine (see eventsProcessor.SaveBlock) — reusing them here avoids both
+	// a redundant conversion pass and the data race that computing them here would race.
+	// That reuse only ever happens when it was computed full=true (indexerEnabled was true
+	// at commit time, the same condition that hands prepared to this worker at all), so the
+	// fallback below — computing it ourselves — always needs full=true too: this path only
+	// runs when ES will actually index ScDeploys/AlteredSCs.
+	if ld == nil {
+		ld = ei.logsAndEventsProc.ExtractDataFromLogs(pool, txs, headerTimestamp, true)
+	}
 	buffers := data.NewBufferSlice(data.DefaultMaxBulkSize)
 
-	if err := ei.indexBlockArtifacts(buffers, headerTimestamp, txs, txsMap, ad, ld, pool); err != nil {
+	if err := ei.indexBlockArtifacts(buffers, headerTimestamp, txs, txsMap, ad, ld, logsDB, pool); err != nil {
 		return err
 	}
 
@@ -551,20 +560,35 @@ func (ei *elasticProcessor) SaveTransactions(
 		return err
 	}
 
+	// Drop the cached conversion only once this work item has fully succeeded and won't be
+	// retried: PreparedBlockData sits in the dispatcher's work-item queue (up to
+	// indexerCacheSize deep) until this method returns without error, pinning the
+	// hex-expanded logs for that whole wait; clearing it here caps the exposure to one
+	// in-flight block instead of the full queue depth. Clearing it before a possible retry
+	// (dataDispatcher.doWork re-runs this on the same PreparedBlockData on error) would
+	// force ExtractDataFromLogs to recompute on this goroutine — reintroducing the exact
+	// race with the websocket hub's marshal that computing it here was meant to avoid.
+	if p, ok := prepared.(*data.PreparedBlockData); ok && p != nil {
+		p.LogsDB = nil
+		p.LogsResults = nil
+	}
+
 	return nil
 }
 
-// resolvePreparedBlockData returns the prepared block data if provided,
-// otherwise it runs prepareTransactionsForDatabase as a fallback.
+// resolvePreparedBlockData returns the prepared block data if provided (including any
+// LogsResults/LogsDB the websocket dispatcher already computed, nil otherwise), or runs
+// prepareTransactionsForDatabase as a fallback.
 func (ei *elasticProcessor) resolvePreparedBlockData(
 	header nodeData.HeaderHandler,
 	pool *indexer.Pool,
 	prepared any,
-) ([]*data.Transaction, map[string]*data.Transaction, *data.AlteredData, error) {
+) ([]*data.Transaction, map[string]*data.Transaction, *data.AlteredData, *data.PreparedLogsResults, []*data.Logs, error) {
 	if p, ok := prepared.(*data.PreparedBlockData); ok && p != nil {
-		return p.Txs, p.TxsMap, p.Altered, nil
+		return p.Txs, p.TxsMap, p.Altered, p.LogsResults, p.LogsDB, nil
 	}
-	return ei.prepareTransactionsForDatabase(header, pool)
+	txs, txsMap, ad, err := ei.prepareTransactionsForDatabase(header, pool)
+	return txs, txsMap, ad, nil, nil, err
 }
 
 // indexBlockArtifacts runs every sub-index step for a block. The list is
@@ -576,6 +600,7 @@ func (ei *elasticProcessor) indexBlockArtifacts(
 	txsMap map[string]*data.Transaction,
 	ad *data.AlteredData,
 	ld *data.PreparedLogsResults,
+	logsDB []*data.Logs,
 	pool *indexer.Pool,
 ) error {
 	steps := []func() error{
@@ -587,7 +612,7 @@ func (ei *elasticProcessor) indexBlockArtifacts(
 		func() error { return ei.indexMarketplaces(ad.Marketplaces.GetAll(), buffers) },
 		func() error { return ei.indexOrders(ad.Orders.GetAll(), buffers) },
 		func() error { return ei.indexAlteredAccounts(headerTimestamp, ad.Accounts.GetAll(), buffers) },
-		func() error { return ei.prepareAndIndexLogs(pool.Logs, txsMap, headerTimestamp, buffers) },
+		func() error { return ei.prepareAndIndexLogs(pool.Logs, txsMap, headerTimestamp, logsDB, buffers) },
 		func() error { return ei.indexScDeploys(ld.ScDeploys, buffers) },
 		func() error { return ei.indexAlteredSmartContracts(ld.AlteredSCs, buffers) },
 	}

indexer/elasticProcessor_test.go

@@ -202,6 +202,76 @@ func TestElasticProcessor_SaveTransactions_FallbackPrepWhenPreparedNil(t *testin
 		"fallback must run the full indexing pipeline when prepared is nil")
 }
 
+// TestElasticProcessor_SaveTransactions_ClearsCachedLogsAfterConsuming guards a memory
+// retention fix: PreparedBlockData.LogsDB/LogsResults (the websocket dispatcher's
+// precomputed conversion, reused here to avoid a duplicate pass — see resolvePreparedBlockData)
+// must not outlive this call. Left set, they'd stay pinned for as long as the containing
+// work item sits in the elastic dispatcher's queue after this point, for no further
+// benefit since nothing downstream reads them again.
+func TestElasticProcessor_SaveTransactions_ClearsCachedLogsAfterConsuming(t *testing.T) {
+	dbWriter := &imock.DatabaseWriterStub{
+		DoBulkRequestCalled: func(_ *bytes.Buffer, _ string) error { return nil },
+	}
+
+	contract := transaction.TransferContract{ToAddress: []byte("klv1d05ju9jaj6u99zph0ant9jh7gksg"), Amount: 45}
+	tx, err := createTransactionHandlerMock(&contract, transaction.TXContract_TransferContractType,
+		[]byte("klv1d05ju9jaj6u99zph0ant9jh7gksf"))
+	require.NoError(t, err)
+
+	header := &dataBlock.Block{Header: &dataBlock.BlockHeader{Nonce: 7, Timestamp: 100}, TxHashes: [][]byte{[]byte("h1")}}
+	pool := &indexer.Pool{Txs: map[string]nodeData.TransactionHandler{"h1": tx}}
+
+	prepared := &data.PreparedBlockData{
+		Txs:         []*data.Transaction{{Hash: "h1"}},
+		TxsMap:      map[string]*data.Transaction{"h1": {Hash: "h1"}},
+		Altered:     data.NewAlteredData(),
+		LogsDB:      []*data.Logs{{ID: "h1", Address: "klv1contract"}},
+		LogsResults: &data.PreparedLogsResults{ScDeploys: map[string]*data.ScDeployInfo{}, AlteredSCs: data.NewAlteredSmartContracts()},
+	}
+
+	ep := newTestElasticSearchDatabase(dbWriter, createMockElasticProcessorArgs())
+	require.NoError(t, ep.SaveTransactions(header, pool, prepared))
+
+	require.Nil(t, prepared.LogsDB, "LogsDB must be cleared once the elastic worker has consumed it")
+	require.Nil(t, prepared.LogsResults, "LogsResults must be cleared once the elastic worker has consumed it")
+}
+
+// TestElasticProcessor_SaveTransactions_KeepsCachedLogsOnBulkRequestError guards a
+// regression in the fix above: dataDispatcher.doWork retries a failed work item on the
+// same *data.PreparedBlockData. Clearing LogsDB/LogsResults before doBulkRequests could
+// fail would force the retry's resolvePreparedBlockData to see them as nil and recompute
+// ExtractDataFromLogs on this (elastic worker) goroutine — reintroducing the exact race
+// with the websocket hub's marshal that computing it on the commit goroutine was meant to
+// avoid. So on error, both fields must survive untouched for the retry to reuse.
+func TestElasticProcessor_SaveTransactions_KeepsCachedLogsOnBulkRequestError(t *testing.T) {
+	localErr := errors.New("bulk request failed")
+	dbWriter := &imock.DatabaseWriterStub{
+		DoBulkRequestCalled: func(_ *bytes.Buffer, _ string) error { return localErr },
+	}
+
+	contract := transaction.TransferContract{ToAddress: []byte("klv1d05ju9jaj6u99zph0ant9jh7gksg"), Amount: 45}
+	tx, err := createTransactionHandlerMock(&contract, transaction.TXContract_TransferContractType,
+		[]byte("klv1d05ju9jaj6u99zph0ant9jh7gksf"))
+	require.NoError(t, err)
+
+	header := &dataBlock.Block{Header: &dataBlock.BlockHeader{Nonce: 7, Timestamp: 100}, TxHashes: [][]byte{[]byte("h1")}}
+	pool := &indexer.Pool{Txs: map[string]nodeData.TransactionHandler{"h1": tx}}
+
+	prepared := &data.PreparedBlockData{
+		Txs:         []*data.Transaction{{Hash: "h1"}},
+		TxsMap:      map[string]*data.Transaction{"h1": {Hash: "h1"}},
+		Altered:     data.NewAlteredData(),
+		LogsDB:      []*data.Logs{{ID: "h1", Address: "klv1contract"}},
+		LogsResults: &data.PreparedLogsResults{ScDeploys: map[string]*data.ScDeployInfo{}, AlteredSCs: data.NewAlteredSmartContracts()},
+	}
+
+	ep := newTestElasticSearchDatabase(dbWriter, createMockElasticProcessorArgs())
+	require.ErrorIs(t, ep.SaveTransactions(header, pool, prepared), localErr)
+
+	require.NotNil(t, prepared.LogsDB, "LogsDB must survive a doBulkRequests error for a retry to reuse")
+	require.NotNil(t, prepared.LogsResults, "LogsResults must survive a doBulkRequests error for a retry to reuse")
+}
+
 func TestElasticseachSaveTransactions_ShouldReturnErr(t *testing.T) {
 	localErr := errors.New("localErr")
 	arguments := createMockElasticProcessorArgs()

indexer/events.go

@@ -11,6 +11,16 @@ const eventQueueBufferSize = 1000
 var EventQueue = make(chan Event, eventQueueBufferSize)
 var UseEventQueue bool
 
+// LogsSubscriberChecker, when set (by the websocket hub during its construction),
+// reports whether dispatching a LOGS event would actually be delivered anywhere — an
+// address-scoped LOGS subscriber, or a configured mirror endpoint. dispatchLogEvents
+// consults it before paying the full bech32/hex-encoding conversion cost on the
+// block-commit goroutine, so a block with many SC events costs nothing extra when nobody
+// would receive them. nil (no hub wired yet, or this indexer package used outside the
+// websocket feature) is treated as "yes, convert" so nothing is silently dropped absent a
+// hub that could report otherwise.
+var LogsSubscriberChecker func() bool
+
 type Event struct {
 	EvType  EventType
 	Message interface{}
@@ -24,6 +34,7 @@ const (
 	ACCOUNTS          EventType = "accounts"
 	BLOCKS            EventType = "blocks"
 	TRANSACTIONS      EventType = "transactions"
+	LOGS              EventType = "logs"
 )
 
 const dropLogIntervalSeconds = 10
@@ -60,22 +71,18 @@ func NewEventTypeStrict(evType string) (EventType, error) {
 		return BLOCKS, nil
 	case "user_transactions":
 		return USER_TRANSACTIONS, nil
+	case "logs":
+		return LOGS, nil
 	default:
 		return UNKNOWN, ErrUnknownEventType
 	}
 }
 
+// NewEventType is the non-strict counterpart of NewEventTypeStrict, returning UNKNOWN
+// instead of an error for an unrecognized type. Delegates to it rather than duplicating
+// the switch, so the two can't silently drift out of sync (e.g. a new type added to one
+// and forgotten in the other).
 func NewEventType(evType string) EventType {
-	switch evType {
-	case "transactions":
-		return TRANSACTIONS
-	case "accounts":
-		return ACCOUNTS
-	case "blocks":
-		return BLOCKS
-	case "user_transactions":
-		return USER_TRANSACTIONS
-	default:
-		return UNKNOWN
-	}
+	t, _ := NewEventTypeStrict(evType)
+	return t
 }

indexer/eventsProcessor.go

@@ -10,16 +10,18 @@ import (
 	indexerData "github.com/klever-io/klever-go/data/indexer"
 	dataState "github.com/klever-io/klever-go/data/state"
 	"github.com/klever-io/klever-go/indexer/data"
+	"github.com/klever-io/klever-go/indexer/logsevents"
 	"github.com/klever-io/klever-go/indexer/workItems"
 	"github.com/klever-io/klever-go/tools/check"
 )
 
 type eventsProcessor struct {
 	*txDatabaseProcessor
-	indexer         Indexer
-	parser          *dataParser
-	kappsController kapp.KAppController
-	accountsDB      dataState.AccountsAdapter
+	indexer           Indexer
+	parser            *dataParser
+	kappsController   kapp.KAppController
+	accountsDB        dataState.AccountsAdapter
+	logsAndEventsProc LogsAndEventsHandler
 }
 
 func NewEventsProcessor(arguments ArgEventsProcessor) (*eventsProcessor, error) {
@@ -28,6 +30,18 @@ func NewEventsProcessor(arguments ArgEventsProcessor) (*eventsProcessor, error)
 		return nil, err
 	}
 
+	// Reuses the same converter (bech32 addresses, hex topics/data) already used to
+	// prepare logs for the Elasticsearch index, so a LOGS websocket event and its
+	// Elasticsearch counterpart for the same block never drift in shape.
+	logsAndEventsProc, err := logsevents.NewLogsAndEventsProcessor(logsevents.ArgsLogsAndEventsProcessor{
+		PubKeyConverter: arguments.AddressPubkeyConverter,
+		Marshalizer:     arguments.Marshalizer,
+		Hasher:          arguments.Hasher,
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	ep := &eventsProcessor{
 		txDatabaseProcessor: newTxDatabaseProcessor(
 			arguments.Hasher,
@@ -41,8 +55,9 @@ func NewEventsProcessor(arguments ArgEventsProcessor) (*eventsProcessor, error)
 			hasher:      arguments.Hasher,
 			marshalizer: arguments.Marshalizer,
 		},
-		kappsController: arguments.KAppController,
-		accountsDB:      arguments.AccountsDB,
+		kappsController:   arguments.KAppController,
+		accountsDB:        arguments.AccountsDB,
+		logsAndEventsProc: logsAndEventsProc,
 	}
 
 	return ep, nil
@@ -85,10 +100,25 @@ func (ep *eventsProcessor) SaveBlock(args *indexerData.ArgsSaveBlockData) {
 	if wsEnabled {
 		prepared := ep.prepare(args)
 		ep.dispatchBlockEvent(args)
+		var txsMap map[string]*data.Transaction
 		if prepared != nil {
+			txsMap = prepared.TxsMap
+			// Run unconditionally (not just when indexerEnabled): this sets
+			// tx.HasLogs/HasOperations/Status, which the websocket payload itself reports —
+			// gating it on indexerEnabled would make a ws-only node and a ws+ES node emit
+			// different payloads for the same block. Doing it here also keeps it
+			// synchronous with dispatchTransactionEvents, so the elastic worker's async
+			// goroutine can't race the hub's marshal of the same prepared.Txs pointers.
+			// full=indexerEnabled: only pay for the ScDeploys/AlteredSCs extraction (which
+			// decodes contract-controlled event topics as addresses) when Elasticsearch
+			// will actually consume it — the websocket payload never reads those fields.
+			prepared.LogsResults = ep.logsAndEventsProc.ExtractDataFromLogs(args.TransactionsPool, prepared.Txs, args.Header.GetTimestamp(), indexerEnabled)
 			ep.dispatchTransactionEvents(prepared.Txs)
 			ep.dispatchAccountEventsFromAlteredAccounts(args.Header.GetTimestamp(), prepared.Altered.Accounts)
 		}
+		// Dispatched outside the prepared-only branch: a tx-prep failure must not silently
+		// drop a block's logs, same as BLOCKS already ships regardless of prepare().
+		ep.dispatchLogEvents(prepared, args.TransactionsPool, txsMap, args.Header.GetTimestamp())
 		if indexerEnabled {
 			args.Prepared = prepared
 		}
@@ -149,6 +179,34 @@ func (ep *eventsProcessor) dispatchTransactionEvents(txs []*data.Transaction) {
 	})
 }
 
+// dispatchLogEvents converts the block's raw smart-contract logs into the same shape
+// already used for the Elasticsearch index (bech32 addresses, hex topics/data) and
+// dispatches them as one LOGS event; the websocket hub fans each entry out by its
+// contract address, same as it does per-account for ACCOUNTS. When prepared is non-nil,
+// the conversion is also stashed on it (PreparedBlockData.LogsDB) so the elastic worker's
+// own logs indexing reuses it instead of paying the same bech32/hex-encoding pass twice.
+func (ep *eventsProcessor) dispatchLogEvents(prepared *data.PreparedBlockData, pool *indexerData.Pool, txsMap map[string]*data.Transaction, blockTimestamp int64) {
+	if pool == nil || len(pool.Logs) == 0 {
+		return
+	}
+	if LogsSubscriberChecker != nil && !LogsSubscriberChecker() {
+		return
+	}
+
+	logsDB := ep.logsAndEventsProc.PrepareLogsForDB(pool.Logs, txsMap, blockTimestamp)
+	if prepared != nil {
+		prepared.LogsDB = logsDB
+	}
+	if len(logsDB) == 0 {
+		return
+	}
+
+	trySendEvent(Event{
+		EvType:  LOGS,
+		Message: logsDB,
+	})
+}
+
 // dispatchAccountEventsFromAlteredAccounts uses GetExistingAccount so that
 // addresses with no persisted state (e.g. ZeroAddress) are silently dropped
 // rather than broadcast as empty ghost accounts.