idecnode/pkg/idec/message.go

package idec

import (
	"encoding/base64"
	"fmt"
	"log"
	"strings"

	"gitrepo.ru/neonxp/idecnode/pkg/model"
	"go.etcd.io/bbolt"
)

func (i *IDEC) GetMessageIDsByEcho(echoID string, offset, limit int) ([]string, int, error) {
	messages := make([]string, 0, limit)
	count := 0
	return messages, count, i.db.View(func(tx *bbolt.Tx) error {
		if _, ok := i.config.Echos[echoID]; !ok {
			return nil
		}

		bEcho := tx.Bucket([]byte(echoID))
		if bEcho == nil {
			return nil
		}

		count = bEcho.Stats().KeyN

		cur := bEcho.Cursor()
		cur.First()
		if limit == 0 {
			limit = count
		}
		if offset < 0 {
			offset = max(0, count+offset-1)
		}
		for i := 0; i < offset; i++ {
			// skip offset entries
			cur.Next()
		}
		for i := 0; i < limit; i++ {
			_, v := cur.Next()
			if v == nil {
				break
			}
			messages = append(messages, string(v))
		}

		return nil
	})
}

func (i *IDEC) GetMessagesByEcho(echoID string, parent string, offset, limit int) ([]*model.Message, error) {
	messages := make([]*model.Message, 0, limit)
	return messages, i.db.View(func(tx *bbolt.Tx) error {
		if _, ok := i.config.Echos[echoID]; !ok {
			return nil
		}

		bEcho := tx.Bucket([]byte(echoID))
		if bEcho == nil {
			return nil
		}
		bMessages := tx.Bucket([]byte(msgBucket))
		if bMessages == nil {
			return nil
		}

		count := bEcho.Stats().KeyN

		cur := bEcho.Cursor()
		cur.First()
		if limit == 0 {
			limit = count
		}
		if offset < 0 {
			offset = max(0, count+offset-1)
		}
		for i := 0; i < offset; i++ {
			// skip offset entries
			cur.Next()
		}
		for i := 0; len(messages) < limit; i++ {
			_, v := cur.Next()
			if v == nil {
				break
			}

			msgText := bMessages.Get(v)

			msg, err := model.MessageFromBundle(string(v), string(msgText))
			if err != nil {
				return err
			}
			if msg.RepTo == parent {
				messages = append(messages, msg)
			}
		}

		return nil
	})
}

func (i *IDEC) GetMessage(messageID string) (*model.Message, error) {
	var msg *model.Message
	return msg, i.db.View(func(tx *bbolt.Tx) error {
		bucket := tx.Bucket([]byte(msgBucket))
		if bucket == nil {
			return ErrMessageNotFound
		}
		b := bucket.Get([]byte(messageID))
		var err error
		msg, err = model.MessageFromBundle(messageID, string(b))

		return err
	})
}

func (i *IDEC) GetMessages(messageIDs []string) ([]*model.Message, error) {
	msgs := make([]*model.Message, 0, len(messageIDs))
	return msgs, i.db.View(func(tx *bbolt.Tx) error {
		bucket := tx.Bucket([]byte(msgBucket))
		if bucket == nil {
			return ErrMessageNotFound
		}
		for _, messageID := range messageIDs {
			b := bucket.Get([]byte(messageID))
			msg, err := model.MessageFromBundle(messageID, string(b))
			if err != nil {
				log.Println(err)
				continue
			}

			msgs = append(msgs, msg)
		}

		return nil
	})
}

func (i *IDEC) SavePointMessage(point string, rawMessage string) error {
	rawMessage = strings.NewReplacer("-", "+", "_", "/").Replace(rawMessage)

	messageBody, err := base64.StdEncoding.DecodeString(rawMessage)
	if err != nil {
		return ErrWrongMessageFormat
	}

	msg, err := model.MessageFromPointMessage(i.config.Node, point, string(messageBody))
	if err != nil {
		return err
	}

	return i.SaveMessage(msg)
}

func (i *IDEC) SaveBundleMessage(msgID, text string) error {
	msg, err := model.MessageFromBundle(msgID, text)
	if err != nil {
		return err
	}

	return i.SaveMessage(msg)
}

func (i *IDEC) SaveMessage(msg *model.Message) error {
	return i.db.Update(func(tx *bbolt.Tx) error {
		bMessages, err := tx.CreateBucketIfNotExists([]byte(msgBucket))
		if err != nil {
			return err
		}
		if err := bMessages.Put([]byte(msg.ID), []byte(msg.Bundle())); err != nil {
			return err
		}
		bEcho, err := tx.CreateBucketIfNotExists([]byte(msg.EchoArea))
		if err != nil {
			return err
		}

		bucketKey := fmt.Sprintf("%s.%s", msg.Date.Format("2006.01.02.15.04.05"), msg.ID)

		return bEcho.Put([]byte(bucketKey), []byte(msg.ID))
	})
}