// VulcanizeDB // Copyright © 2019 Vulcanize // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Affero General Public License for more details. // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . package logs import ( "errors" "github.com/ethereum/go-ethereum/common" "github.com/sirupsen/logrus" "github.com/vulcanize/vulcanizedb/libraries/shared/constants" "github.com/vulcanize/vulcanizedb/libraries/shared/fetcher" "github.com/vulcanize/vulcanizedb/libraries/shared/transactions" "github.com/vulcanize/vulcanizedb/libraries/shared/transformer" "github.com/vulcanize/vulcanizedb/pkg/core" "github.com/vulcanize/vulcanizedb/pkg/datastore" ) var ( ErrNoUncheckedHeaders = errors.New("no unchecked headers available for log fetching") ErrNoWatchedAddresses = errors.New("no watched addresses configured in the log extractor") ) type ILogExtractor interface { AddTransformerConfig(config transformer.EventTransformerConfig) error ExtractLogs(recheckHeaders constants.TransformerExecution) error } type LogExtractor struct { Addresses []common.Address CheckedHeadersRepository datastore.CheckedHeadersRepository CheckedLogsRepository datastore.CheckedLogsRepository Fetcher fetcher.ILogFetcher LogRepository datastore.HeaderSyncLogRepository StartingBlock *int64 Syncer transactions.ITransactionsSyncer Topics []common.Hash } // Add additional logs to extract func (extractor *LogExtractor) AddTransformerConfig(config transformer.EventTransformerConfig) error { checkedHeadersErr := extractor.updateCheckedHeaders(config) if checkedHeadersErr != nil { return checkedHeadersErr } if extractor.StartingBlock == nil { extractor.StartingBlock = &config.StartingBlockNumber } else if earlierStartingBlockNumber(config.StartingBlockNumber, *extractor.StartingBlock) { extractor.StartingBlock = &config.StartingBlockNumber } addresses := transformer.HexStringsToAddresses(config.ContractAddresses) extractor.Addresses = append(extractor.Addresses, addresses...) extractor.Topics = append(extractor.Topics, common.HexToHash(config.Topic)) return nil } // Fetch and persist watched logs func (extractor LogExtractor) ExtractLogs(recheckHeaders constants.TransformerExecution) error { if len(extractor.Addresses) < 1 { logrus.Errorf("error extracting logs: %s", ErrNoWatchedAddresses.Error()) return ErrNoWatchedAddresses } uncheckedHeaders, uncheckedHeadersErr := extractor.CheckedHeadersRepository.UncheckedHeaders(*extractor.StartingBlock, -1, getCheckCount(recheckHeaders)) if uncheckedHeadersErr != nil { logrus.Errorf("error fetching missing headers: %s", uncheckedHeadersErr) return uncheckedHeadersErr } if len(uncheckedHeaders) < 1 { return ErrNoUncheckedHeaders } for _, header := range uncheckedHeaders { logs, fetchLogsErr := extractor.Fetcher.FetchLogs(extractor.Addresses, extractor.Topics, header) if fetchLogsErr != nil { logError("error fetching logs for header: %s", fetchLogsErr, header) return fetchLogsErr } if len(logs) > 0 { transactionsSyncErr := extractor.Syncer.SyncTransactions(header.ID, logs) if transactionsSyncErr != nil { logError("error syncing transactions: %s", transactionsSyncErr, header) return transactionsSyncErr } createLogsErr := extractor.LogRepository.CreateHeaderSyncLogs(header.ID, logs) if createLogsErr != nil { logError("error persisting logs: %s", createLogsErr, header) return createLogsErr } } markHeaderCheckedErr := extractor.CheckedHeadersRepository.MarkHeaderChecked(header.ID) if markHeaderCheckedErr != nil { logError("error marking header checked: %s", markHeaderCheckedErr, header) return markHeaderCheckedErr } } return nil } func earlierStartingBlockNumber(transformerBlock, watcherBlock int64) bool { return transformerBlock < watcherBlock } func logError(description string, err error, header core.Header) { logrus.WithFields(logrus.Fields{ "headerId": header.ID, "headerHash": header.Hash, "blockNumber": header.BlockNumber, }).Errorf(description, err.Error()) } func getCheckCount(recheckHeaders constants.TransformerExecution) int64 { if recheckHeaders == constants.HeaderUnchecked { return 1 } return constants.RecheckHeaderCap } func (extractor *LogExtractor) updateCheckedHeaders(config transformer.EventTransformerConfig) error { alreadyWatchingLog, watchingLogErr := extractor.CheckedLogsRepository.AlreadyWatchingLog(config.ContractAddresses, config.Topic) if watchingLogErr != nil { return watchingLogErr } if !alreadyWatchingLog { uncheckHeadersErr := extractor.CheckedHeadersRepository.MarkHeadersUnchecked(config.StartingBlockNumber) if uncheckHeadersErr != nil { return uncheckHeadersErr } markLogWatchedErr := extractor.CheckedLogsRepository.MarkLogWatched(config.ContractAddresses, config.Topic) if markLogWatchedErr != nil { return markLogWatchedErr } } return nil }