fetcher/i2es/topics.go

package i2es

import (
	"encoding/json"
	"errors"
	"strings"

	"github.com/google/uuid"
	log "github.com/sirupsen/logrus"
)

// getorcreate topicID. Generate new unique topicID if message is start message
func (t topicid) getOrCreate(msg *ESDoc) {
	if msg.TopicID == "" && msg.Repto == "" {
		log.Warn("New topic assign for message " + msg.MsgID)
		msg.TopicID = uuid.New().String()
	}
}

// AssignTopics ...
func (es ESConf) AssignTopics(messages []ESDoc) error {
	log.Info("====\nTrying assign topics\n====")

	for _, message := range messages {
		log.Info("Working for " + message.MsgID)
		// generate new topicid
		if message.Repto == "" && message.TopicID == "" {
			topic := uuid.New().String()
			message.TopicID = topic
			_, err := es.PutToIndex(message)
			if err != nil {
				log.Error(err.Error())
				continue
			}
			log.Info("New topic " + topic + " for message " + message.MsgID + " assigned")
		}

		// skip message with assigned topicid
		if message.TopicID != "" {
			log.Info("message " + message.MsgID + " has already topicid " + message.TopicID)
			continue
		}

		// find top message of the thread and get they topicid
		if message.Repto != "" {
			log.Info("Repto " + message.Repto + " found in message " + message.MsgID)
			t := topicid{es}
			topicid, err := t.findTopicID(message.Repto, 0)
			if err != nil {
				log.Error(err.Error())
				message.Misplaced = "yes"
				log.Error("Message " + message.MsgID + " misplaced")
				_, err := es.PutToIndex(message)
				if err != nil {
					log.Error(err.Error())
					continue
				}
				continue
			}

			message.TopicID = topicid
			_, err = es.PutToIndex(message)
			if err != nil {
				log.Error(err.Error())
				continue
			}
			log.Info("Topic " + topicid + "found for message " + message.MsgID)
		}
	}
	return nil
}

// findTopicID for msgid
func (t topicid) findTopicID(msgid string, enter int) (string, error) {
	searchURI := strings.Join([]string{t.es.Host, t.es.Index, t.es.Type, "_search"}, "/")
	searchQ := strings.Join([]string{`{"sort": [{"date": {"order": "desc"}}, {"_score": { "order": "desc" }}], "query": { "query_string" : {"fields": ["msgid"], "query": "msgid.keyword:` + msgid + ` "}}}`}, "")
	log.Info(searchQ)

	body, err := doRequest(searchURI, searchQ, "POST")
	if err != nil {
		return "", err
	}

	var res ESRes
	err = json.Unmarshal(body, &res)
	if err != nil {
		return "", err
	}

	if res.Hits.Total == 0 {
		e := errors.New("MSGID " + msgid + " not found ")
		log.Error(string(body))
		return "", e
	}

	if res.Hits.Hits[0].Source.TopicID != "" {
		topicid := res.Hits.Hits[0].Source.TopicID
		return topicid, nil
	}

	if res.Hits.Hits[0].Source.Repto != "" && res.Hits.Hits[0].Source.TopicID == "" {
		if enter > 25 {
			topicid := uuid.New().String()
			e := errors.New("Maximum recurse depth. Top message not in index? Forse assign new topicID " + topicid)
			log.Error(e.Error())
			return topicid, err
		} else {
			t.findTopicID(res.Hits.Hits[0].Source.Repto, enter+1)
		}
	} else if res.Hits.Hits[0].Source.Repto == "" && res.Hits.Hits[0].Source.TopicID != "" {
		return res.Hits.Hits[0].Source.TopicID, nil
	}

	if res.Hits.Hits[0].Source.Repto == "" && res.Hits.Hits[0].Source.TopicID == "" {
		// This is a old message without topicid => generate new topicid and update a top document
		topicid := uuid.New().String()
		log.Warn(topicid)
		log.Warn("Top message " + res.Hits.Hits[0].Source.MsgID + " found but it is doesn not have a topicID")
		log.Warn("Generated new topicID " + topicid)
		log.Warn("Update top message")

		updateDoc := res.Hits.Hits[0].Source
		updateDoc.TopicID = topicid

		updateDocPlain, err := json.Marshal(updateDoc)
		if err != nil {
			e := errors.New("Cant serialize document: " + err.Error())
			return "", e
		}

		postURI := strings.Join([]string{t.es.Host, t.es.Index, t.es.Type, updateDoc.MsgID}, "/")
		_, err = doRequest(postURI, string(updateDocPlain), "PUT")
		if err != nil {
			e := errors.New("Cant update document: " + err.Error())
			return "", e
		}

		log.Info(string(updateDocPlain))
		return topicid, nil
	}
	return "", errors.New("Something went wrong")
}

// check if
func (t topicid) checkTopicID(msg *ESDoc) func() string {
	if msg.TopicID != "" {
		return func() string { return msg.TopicID }
	}

	if msg.Repto == "" {
		var genTopicID = func() string {
			topicid := uuid.New().String()
			log.Warn("New start message for topic " + topicid + " found")
			return topicid
		}
		return genTopicID
	}

	if msg.Repto != "" {
		var findAndSetTopicID = func() string {
			topicid, err := t.findTopicID(msg.Repto, 0)
			if err != nil {
				log.Error(err.Error())
				return ""
			}
			log.Info("Topic ID " + topicid + " found for message " + msg.MsgID)
			return topicid
		}
		return findAndSetTopicID
	}

	return func() string { return "" }
}