worker/internal/sender/run.go

package sender

import (
	"errors"
	"log"
	"time"

	"rocketgit.ru/rsmon/worker/app/models"
)

// Run provides functionality.
func Run() {
	messages := make([]models.Message, 0)

	tx := models.DB().Begin().Set("gorm:association_autoupdate", false)

	q := tx
	// q = q.Set("gorm:query_option", "FOR UPDATE")
	err := models.MessageScope(q).Where("state IN ('queued')").Find(&messages).Error
	if err != nil {
		tx.Rollback()
		log.Println(err)
		return
		// panic(err)
	}

	for _, msg := range messages { //nolint:gocritic // range copy is acceptable here
		msg.State = "sending"
		tx.Model(&msg).Update("state", "sending")
		// spew.Dump(msg.ID)
		// err = tx.Save(&msg).Error
		// if err != nil {
		// panic(err)
		// }
	}
	_ = tx.Commit()

	for _, msg := range messages { //nolint:gocritic // range copy is acceptable here
		_, _ = RunMessage(&msg)
	}
}

// RunMessage provides functionality.
func RunMessage(message *models.Message) (*string, error) {
	var err error
	var response *string

	log.Println("send message", message.ID)
	switch message.Contact.Kind {
	case "telegram_group", "telegram_private":
		err = RunTelegram(message)
	case "email":
		err = RunEmail(message)
	case "webhook":
		response, err = RunWebhook(message)
	case "mattermost":
		response, err = RunMattermost(message)
	case "sms":
		err = RunSMS(message)
	case "voice":
		err = RunVoice(message)
	default:
		err = errors.New("notification kind not implemented: " + message.Contact.Kind)
	}
	if response != nil {
		log.Println("notification done. response:", *response, "error:", err)
	} else {
		log.Println("notification done. error:", err)
	}

	if message.ID != 0 {
		if response != nil {
			message.Response = response
		}
		if err == nil {
			message.State = "sent"
			message.SentAt = time.Now()
		} else {
			message.State = "error"
			et := err.Error()
			message.Error = &et
		}
		err = models.DB().Save(&message).Error
		if err != nil {
			log.Println("ERROR:", err)
			return response, err
		}
	}
	return response, err
}