go-receipt-tracker/workers/telegram/telegram.go

package telegram

import (
	"context"
	"log/slog"
	"sync/atomic"
	"time"

	"git.savin.nyc/alex/go-receipt-tracker/bus"
	"git.savin.nyc/alex/go-receipt-tracker/config"
	tele "gopkg.in/telebot.v4"
)

type Bot struct {
	id            string                    //
	opts          *config.Telegram          //
	client        *tele.Bot                 //
	subscriptions map[string]chan bus.Event //
	bus           *bus.Bus                  //
	log           *slog.Logger              //
	cancel        context.CancelFunc        //
	end           uint32                    // ensure the close methods are only called once
}

func NewTelegramBot(cfg *config.Telegram, bus *bus.Bus) (*Bot, error) {

	bot := &Bot{
		id:   "telegram",
		opts: cfg,
		bus:  bus,
	}

	return bot, nil
}

func (b *Bot) ID() string {
	return b.id
}

func (b *Bot) Type() string {
	return "telegram"
}

func (b *Bot) Init(log *slog.Logger) error {
	var err error
	b.log = log

	pref := tele.Settings{
		Token: b.opts.ApiKey,
		Poller: &tele.LongPoller{
			Timeout: 10 * time.Second,
		},
	}

	b.client, err = tele.NewBot(pref)
	if err != nil {
		b.log.Debug("cannot create telegram bot", "error", err)
		return err
	}

	b.client.Handle(tele.OnText, func(c tele.Context) error {
		return c.Send("")
	})

	b.client.Handle(tele.OnPhoto, func(c tele.Context) error {

		photo := c.Message().Photo
		b.client.Download(&photo.File, "./"+photo.UniqueID+".jpg")

		msg := &bus.Message{
			Image: &bus.Image{
				ID:       photo.UniqueID,
				Filename: b.opts.StoragePath + "/" + photo.UniqueID + ".jpg",
			},
			ReplyType: "send",
			TbContext: c,
		}
		err := b.bus.Publish("processor:image", msg)
		if err != nil {
			b.log.Error("couldn't publish to the channel", "channel", "processor:image", "error", err.Error())
		}

		return c.Send("")
	})

	b.client.Handle(tele.OnCallback, func(c tele.Context) error {
		b.log.Debug("CALLBACK (BEFORE) %s // %s", c.Callback().ID, c.Callback().Data)

		tbc := new(TelegramBotCommand)
		err := tbc.Unmarshal(c.Callback().Data)
		if err != nil {
			panic(err)
		}
		b.log.Debug("CALLBACK (AFTER) %d|%d|%d|%d|%d|%d|%s", tbc.Cmd, tbc.ID, tbc.RefCmd, tbc.RefID, tbc.TelegramID, tbc.Pagination, tbc.Extra)

		switch tbc.Cmd {
		case btnCmd["user_add"]:
			return c.Send("")
		case btnCmd["receipt_add"]:

			return c.Send("")
		case btnCmd["receipt_edit"]:
			return c.Send("")
		case btnCmd["receipts_list"]:
		case btnCmd["items_list"]:
		}
		cbr := &tele.CallbackResponse{
			CallbackID: c.Callback().ID,
			Text:       "No unique",
			ShowAlert:  false,
		}
		return c.Respond(cbr)
	})

	b.client.Handle("/start", func(c tele.Context) error {
		err = b.bus.Publish("processor:start", nil)
		if err != nil {
			b.log.Error("couldn't publish to a channel", "channel", "telegram:publish", "error", err.Error())
		}

		return c.Send("")
	})

	b.client.Handle("/receipts", func(c tele.Context) error {
		return c.Send("")
	})

	b.client.Handle("/reports", func(c tele.Context) error {
		return c.Send("")
	})

	return nil
}

func (b *Bot) Serve() {
	if atomic.LoadUint32(&b.end) == 1 {
		return
	}

	b.subscribe("messenger:" + b.ID())

	b.client.Start()

	ctx, cancel := context.WithCancel(context.Background())
	b.cancel = cancel

	go b.eventLoop(ctx)

	// if atomic.LoadUint32(&b.end) == 0 {
	// 	go func() {
	// 		if !w.mqtt.IsConnected() {
	// 			b.bus.Publish("app:connected", &bus.ConnectionStatus{WorkerID: b.ID(), WorkerType: b.Type(), IsConnected: false})
	// 			b.log.Warn(b.ID() + " client is disconnected")
	// 		}
	// 	}()
	// }
}

func (b *Bot) Stop() {
	b.client.Stop()
}

func (b *Bot) subscribe(chn string) error {
	s, err := b.bus.Subscribe(chn, b.Type()+":"+b.ID())
	if err != nil {
		b.log.Error("couldn't subscribe to a channel", "channel", chn, "error", err.Error())
		return err
	}
	b.subscriptions[chn] = s

	return nil
}

// eventLoop loops forever
func (b *Bot) eventLoop(ctx context.Context) {
	b.log.Debug(b.ID() + " communication event loop started")
	defer b.log.Debug(b.ID() + " communication event loop halted")

	for {
		for chn, ch := range b.subscriptions {
			select {
			case event := <-ch:
				b.log.Debug("got a new message to a channel", "channel", chn)
				for _, msg := range event.Payload.([]*bus.Message) {
					b.log.Debug("publishing telegram message", "topic", msg)
					// w.mqtt.Publish(message.Topic, message.QOS, message.Retained, message.Payload)
				}
			case <-ctx.Done():
				b.log.Info("stopping " + b.ID() + " communication event loop")
				return
			}
		}
	}

	// for {
	// 	select {
	// 	case event := <-chMQTTPublishStatus:
	// 		for _, message := range event.Data.([]*bus.Message) {
	// 			b.log.Debug("publishing mqtt message", "topic", message.Topic, "qos", message.QOS, "retained", message.Retained, "payload", message.Payload)
	// 			// w.mqtt.Publish(message.Topic, message.QOS, message.Retained, message.Payload)
	// 		}
	// 	case event := <-chMQTTSubscribeCommand:
	// 		for _, message := range event.Data.([]*bus.Message) {
	// 			b.log.Debug("subscribing to a topic", "topic", message.Topic, "qos", message.QOS)
	// 			// w.mqtt.SubscribeMultiple()
	// 			// b.mqtt.Subscribe(message.Topic, message.QOS, w.mqttHandlers.publish)
	// 		}
	// 	case <-ctx.Done():
	// 		b.log.Info("stopping mqtt communication event loop")
	// 		return
	// 	}
	// }
}