fix: Enhanced Data Collection and Response Time for SendWork and Twitter

.env.example

@@ -34,6 +34,11 @@ DISCORD_SCRAPER=true
 DISCORD_BOT_TOKEN= your discord bot token
 TWITTER_SCRAPER=true
 WEB_SCRAPER=true
+TELEGRAM_SCRAPER=false
+
+# Go to my.telegram.org/auth and after logging in click the developer option to get these
+TELEGRAM_APP_ID=
+TELEGRAM_APP_HASH=
 
 ## Optional added features for ORACLE or VALIDATOR nodes only
 

Makefile

@@ -23,7 +23,7 @@ client: build
 	@./bin/masa-node-cli
 
 test:
-	@go test ./...
+	@go test -v -count=1 ./...
 
 clean:
 	@rm -rf bin

pkg/api/handlers_data.go

@@ -100,7 +100,7 @@ func handleWorkResponse(c *gin.Context, responseCh chan []byte) {
 
 			c.JSON(http.StatusOK, result)
 			return
-		case <-time.After(90 * time.Second):
+		case <-time.After(60 * time.Second):
 			c.JSON(http.StatusGatewayTimeout, gin.H{"error": "Request timed out"})
 			return
 		case <-c.Done():

pkg/config/app.go

@@ -112,8 +112,8 @@ func GetInstance() *AppConfig {
 	once.Do(func() {
 		instance = &AppConfig{}
 
-		instance.setDefaultConfig()
 		instance.setEnvVariableConfig()
+		instance.setDefaultConfig()
 		instance.setFileConfig(viper.GetString("FILE_PATH"))
 		err := instance.setCommandLineConfig()
 		if err != nil {

pkg/llmbridge/sentiment.go

@@ -19,11 +19,21 @@ import (
 // AnalyzeSentimentTweets analyzes the sentiment of the provided tweets by sending them to the Claude API.
 // It concatenates the tweets, creates a payload, sends a request to Claude, parses the response,
 // and returns the concatenated tweets content, a sentiment summary, and any error.
-func AnalyzeSentimentTweets(tweets []*twitterscraper.Tweet, model string, prompt string) (string, string, error) {
+func AnalyzeSentimentTweets(tweets []*twitterscraper.TweetResult, model string, prompt string) (string, string, error) {
 	// check if we are using claude or gpt, can add others easily
 	if strings.Contains(model, "claude-") {
 		client := NewClaudeClient() // Adjusted to call without arguments
-		tweetsContent := ConcatenateTweets(tweets)
+
+		var validTweets []*twitterscraper.TweetResult
+		for _, tweet := range tweets {
+			if tweet.Error != nil {
+				logrus.WithError(tweet.Error).Warn("[-] Error in tweet")
+				continue
+			}
+			validTweets = append(validTweets, tweet)
+		}
+
+		tweetsContent := ConcatenateTweets(validTweets)
 		payloadBytes, err := CreatePayload(tweetsContent, model, prompt)
 		if err != nil {
 			logrus.Errorf("[-] Error creating payload: %v", err)
@@ -101,10 +111,10 @@ func AnalyzeSentimentTweets(tweets []*twitterscraper.Tweet, model string, prompt
 
 // ConcatenateTweets concatenates the text of the provided tweets into a single string,
 // with each tweet separated by a newline character.
-func ConcatenateTweets(tweets []*twitterscraper.Tweet) string {
+func ConcatenateTweets(tweets []*twitterscraper.TweetResult) string {
 	var tweetsTexts []string
-	for _, tweet := range tweets {
-		tweetsTexts = append(tweetsTexts, tweet.Text)
+	for _, t := range tweets {
+		tweetsTexts = append(tweetsTexts, t.Tweet.Text)
 	}
 	return strings.Join(tweetsTexts, "\n")
 }

pkg/pubsub/node_data.go

@@ -44,36 +44,36 @@ func (m *JSONMultiaddr) UnmarshalJSON(b []byte) error {
 }
 
 type NodeData struct {
-	Multiaddrs           []JSONMultiaddr `json:"multiaddrs,omitempty"`
-	PeerId               peer.ID         `json:"peerId"`
-	FirstJoinedUnix      int64           `json:"firstJoined,omitempty"`
-	LastJoinedUnix       int64           `json:"lastJoined,omitempty"`
-	LastLeftUnix         int64           `json:"-"`
-	LastUpdatedUnix      int64           `json:"lastUpdated,omitempty"`
-	CurrentUptime        time.Duration   `json:"uptime,omitempty"`
-	CurrentUptimeStr     string          `json:"uptimeStr,omitempty"`
-	AccumulatedUptime    time.Duration   `json:"accumulatedUptime,omitempty"`
-	AccumulatedUptimeStr string          `json:"accumulatedUptimeStr,omitempty"`
-	EthAddress           string          `json:"ethAddress,omitempty"`
-	Activity             int             `json:"activity,omitempty"`
-	IsActive             bool            `json:"isActive"`
-	IsStaked             bool            `json:"isStaked"`
-	SelfIdentified       bool            `json:"-"`
-	IsValidator          bool            `json:"isValidator"`
-	IsTwitterScraper     bool            `json:"isTwitterScraper"`
-	IsDiscordScraper     bool            `json:"isDiscordScraper"`
-	IsTelegramScraper    bool            `json:"isTelegramScraper"`
-	IsWebScraper         bool            `json:"isWebScraper"`
-	Records              any             `json:"records,omitempty"`
-	Version              string          `json:"version"`
+	Multiaddrs           []JSONMultiaddr      `json:"multiaddrs,omitempty"`
+	PeerId               peer.ID              `json:"peerId"`
+	FirstJoinedUnix      int64                `json:"firstJoined,omitempty"`
+	LastJoinedUnix       int64                `json:"lastJoined,omitempty"`
+	LastLeftUnix         int64                `json:"-"`
+	LastUpdatedUnix      int64                `json:"lastUpdated,omitempty"`
+	CurrentUptime        time.Duration        `json:"uptime,omitempty"`
+	CurrentUptimeStr     string               `json:"uptimeStr,omitempty"`
+	AccumulatedUptime    time.Duration        `json:"accumulatedUptime,omitempty"`
+	AccumulatedUptimeStr string               `json:"accumulatedUptimeStr,omitempty"`
+	EthAddress           string               `json:"ethAddress,omitempty"`
+	Activity             int                  `json:"activity,omitempty"`
+	IsActive             bool                 `json:"isActive"`
+	IsStaked             bool                 `json:"isStaked"`
+	SelfIdentified       bool                 `json:"-"`
+	IsValidator          bool                 `json:"isValidator"`
+	IsTwitterScraper     bool                 `json:"isTwitterScraper"`
+	IsDiscordScraper     bool                 `json:"isDiscordScraper"`
+	IsTelegramScraper    bool                 `json:"isTelegramScraper"`
+	IsWebScraper         bool                 `json:"isWebScraper"`
+	Records              any                  `json:"records,omitempty"`
+	Version              string               `json:"version"`
+	TimeoutChecks        map[string]time.Time `json:"timeoutChecks,omitempty"`
 }
 
 // NewNodeData creates a new NodeData struct initialized with the given
 // parameters. It is used to represent data about a node in the network.
 func NewNodeData(addr multiaddr.Multiaddr, peerId peer.ID, publicKey string, activity int) *NodeData {
 	multiaddrs := make([]JSONMultiaddr, 0)
 	multiaddrs = append(multiaddrs, JSONMultiaddr{addr})
-	// cfg := config.GetInstance()
 
 	return &NodeData{
 		PeerId:            peerId,
@@ -94,6 +94,38 @@ func (n *NodeData) Address() string {
 	return fmt.Sprintf("%s/p2p/%s", n.Multiaddrs[0].String(), n.PeerId.String())
 }
 
+// WorkerCategory represents the main categories of workers
+type WorkerCategory int
+
+const (
+	CategoryDiscord WorkerCategory = iota
+	CategoryTelegram
+	CategoryTwitter
+	CategoryWeb
+)
+
+// String returns the string representation of the WorkerCategory
+func (wc WorkerCategory) String() string {
+	return [...]string{"Discord", "Telegram", "Twitter", "Web"}[wc]
+}
+
+// CanDoWork checks if the node can perform work of the specified WorkerType.
+// It returns true if the node is configured for the given worker type, false otherwise.
+func (n *NodeData) CanDoWork(workerType WorkerCategory) bool {
+	switch workerType {
+	case CategoryTwitter:
+		return n.IsActive && n.IsTwitterScraper
+	case CategoryDiscord:
+		return n.IsActive && n.IsDiscordScraper
+	case CategoryTelegram:
+		return n.IsActive && n.IsTelegramScraper
+	case CategoryWeb:
+		return n.IsActive && n.IsWebScraper
+	default:
+		return false
+	}
+}
+
 // TwitterScraper checks if the current node is configured as a Twitter scraper.
 // It retrieves the configuration instance and returns the value of the TwitterScraper field.
 func (n *NodeData) TwitterScraper() bool {

pkg/scrapers/twitter/tweets.go

@@ -13,6 +13,11 @@
 	"github.com/sirupsen/logrus"
 )
 
+type TweetResult struct {
+	Tweet *twitterscraper.Tweet
+	Error error
+}
+
 // auth initializes and returns a new Twitter scraper instance. It attempts to load cookies from a file to reuse an existing session.
 // If no valid session is found, it performs a login with credentials specified in the application's configuration.
 // On successful login, it saves the session cookies for future use. If the login fails, it returns nil.
@@ -69,7 +74,7 @@
 //   - An error if the scraping or sentiment analysis process encounters any issues.
 func ScrapeTweetsForSentiment(query string, count int, model string) (string, string, error) {
 	scraper := auth()
-	var tweets []*twitterscraper.Tweet
+	var tweets []*TweetResult
 
 	if scraper == nil {
 		return "", "", fmt.Errorf("there was an error authenticating with your Twitter credentials")
@@ -80,14 +85,30 @@
 
 	// Perform the search with the specified query and count
 	for tweetResult := range scraper.SearchTweets(context.Background(), query, count) {
+		var tweet TweetResult
 		if tweetResult.Error != nil {
-			logrus.Printf("Error fetching tweet: %v", tweetResult.Error)
-			continue
+			tweet = TweetResult{
+				Tweet: nil,
+				Error: tweetResult.Error,
+			}
+		} else {
+			tweet = TweetResult{
+				Tweet: tweetResult.Tweet,
+				Error: nil,
+			}
 		}
-		tweets = append(tweets, &tweetResult.Tweet)
+		tweets = append(tweets, &tweet)
 	}
 	sentimentPrompt := "Please perform a sentiment analysis on the following tweets, using an unbiased approach. Sentiment analysis involves identifying and categorizing opinions expressed in text, particularly to determine whether the writer's attitude towards a particular topic, product, etc., is positive, negative, or neutral. After analyzing, please provide a summary of the overall sentiment expressed in these tweets, including the proportion of positive, negative, and neutral sentiments if applicable."
-	prompt, sentiment, err := llmbridge.AnalyzeSentimentTweets(tweets, model, sentimentPrompt)
+
+	twitterScraperTweets := make([]*twitterscraper.TweetResult, len(tweets))
+	for i, tweet := range tweets {
+		twitterScraperTweets[i] = &twitterscraper.TweetResult{
+			Tweet: tweet.Tweet,
+			Error: tweet.Error,
+		}
+	}
+	prompt, sentiment, err := llmbridge.AnalyzeSentimentTweets(twitterScraperTweets, model, sentimentPrompt)
 	if err != nil {
 		return "", "", err
 	}
@@ -103,9 +124,9 @@
 // Returns:
 //   - A slice of pointers to twitterscraper.Tweet objects that match the search query.
 //   - An error if the scraping process encounters any issues.
-func ScrapeTweetsByQuery(query string, count int) ([]*twitterscraper.Tweet, error) {
+func ScrapeTweetsByQuery(query string, count int) ([]*TweetResult, error) {
 	scraper := auth()
-	var tweets []*twitterscraper.Tweet
+	var tweets []*TweetResult
 
 	if scraper == nil {
 		return nil, fmt.Errorf("there was an error authenticating with your Twitter credentials")
@@ -116,21 +137,29 @@
 
 	// Perform the search with the specified query and count
 	for tweetResult := range scraper.SearchTweets(context.Background(), query, count) {
+		var tweet TweetResult
 		if tweetResult.Error != nil {
-			logrus.Printf("Error fetching tweet: %v", tweetResult.Error)
-			continue
+			tweet = TweetResult{
+				Tweet: nil,
+				Error: tweetResult.Error,
+			}
+		} else {
+			tweet = TweetResult{
+				Tweet: tweetResult.Tweet,
+				Error: nil,
+			}
 		}
-		tweets = append(tweets, &tweetResult.Tweet)
+		tweets = append(tweets, &tweet)
 	}
 	return tweets, nil
 }
 
 // ScrapeTweetsByTrends scrapes the current trending topics on Twitter.
 // It returns a slice of strings representing the trending topics.
 // If an error occurs during the scraping process, it returns an error.
-func ScrapeTweetsByTrends() ([]string, error) {
+func ScrapeTweetsByTrends() ([]*TweetResult, error) {
 	scraper := auth()
-	var tweets []string
+	var trendResults []*TweetResult
 
 	if scraper == nil {
 		return nil, fmt.Errorf("there was an error authenticating with your Twitter credentials")
@@ -141,13 +170,18 @@
 
 	trends, err := scraper.GetTrends()
 	if err != nil {
-		logrus.Printf("Error fetching tweet: %v", err)
 		return nil, err
 	}
 
-	tweets = append(tweets, trends...)
+	for _, trend := range trends {
+		trendResult := &TweetResult{
+			Tweet: &twitterscraper.Tweet{Text: trend},
+			Error: nil,
+		}
+		trendResults = append(trendResults, trendResult)
+	}
 
-	return tweets, nil
+	return trendResults, nil
 }
 
 // ScrapeTweetsProfile scrapes the profile and tweets of a specific Twitter user.