package i2es import ( "encoding/json" "errors" "strings" log "github.com/Sirupsen/logrus" "github.com/google/uuid" ) // 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 "" } }