Initial commit

tg_parser/internal/usecase/fetch_channels.go

@@ -0,0 +1,177 @@
+package usecase
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/TelegramExchange/pkg/transaction"
+	"github.com/gotd/td/tgerr"
+	"github.com/rs/zerolog/log"
+
+	"github.com/TelegramExchange/tgex-backend/tg_parser/internal/domain"
+)
+
+func (uc *UseCase) FetchChannels(ctx context.Context) error {
+	channels := uc.database.GetChannels(ctx)
+	log.Debug().Msg("start fetch channels")
+
+	for _, c := range channels {
+		if c.Pts == 0 {
+			go uc.initializeChannel(ctx, c)
+			continue
+		}
+
+		err := uc.processChannel(ctx, c)
+		if err != nil {
+			if isPrivateChannelError(err) {
+				log.Warn().Int64("telegram_id", c.TelegramID).Msg("Private channel, attempting rejoin")
+				go uc.rejoinChannel(ctx, c)
+
+				return nil
+			}
+			return fmt.Errorf("uc.processChannel: %w", err)
+		}
+	}
+
+	return nil
+}
+
+func isPrivateChannelError(err error) bool {
+	return tgerr.Is(err, "CHANNEL_PRIVATE", "CHANNEL_INVALID", "CHANNEL_FORBIDDEN")
+}
+
+func (uc *UseCase) processChannel(ctx context.Context, channel domain.Channel) error {
+	diff, err := uc.telegram.GetChannelDiff(ctx, channel, 20)
+	if err != nil {
+		return fmt.Errorf("telegram.GetChannelDiff: %w", err)
+	}
+
+	for _, p := range diff.NewPosts {
+		log.Info().Msgf("New post: %s - Views: %d, Text: %.10s...", p.Link, p.Views, p.Text)
+
+		err = uc.database.CreatePost(ctx, p)
+		if err != nil {
+			return fmt.Errorf("database.CreatePost: %w", err)
+		}
+	}
+
+	for _, p := range diff.DeletedPosts {
+		log.Info().Msgf("Deleted post: %s - Views: %d", p.Link, p.Views)
+
+		err = uc.database.DeletePost(ctx, p)
+		if err != nil {
+			return fmt.Errorf("database.DeletePost: %w", err)
+		}
+	}
+
+	// Update channel metadata if changed
+	if diff.UpdatedChannel != nil {
+		diff.UpdatedChannel.Pts = diff.NewPts
+		err = uc.database.UpdateChannel(ctx, *diff.UpdatedChannel)
+		if err != nil {
+			return fmt.Errorf("database.UpdateChannel: %w", err)
+		}
+
+		return nil
+	}
+
+	channel.Pts = diff.NewPts
+	err = uc.database.UpdateChannel(ctx, channel)
+	if err != nil {
+		return fmt.Errorf("database.UpdateChannel: %w", err)
+	}
+
+	return nil
+}
+
+func (uc *UseCase) rejoinChannel(ctx context.Context, channel domain.Channel) {
+	defer func() {
+		err := uc.database.UpdateChannel(ctx, channel)
+		if err != nil {
+			log.Err(err).Msg("database.UpdateChannel (rejoinChannel)")
+		}
+	}()
+
+	if channel.InviteLink == "" {
+		log.Warn().Int64("telegram_id", channel.TelegramID).Msg("No invite link stored, marking inaccessible")
+		channel.IsAccessible = false
+
+		return
+	}
+
+	updated, err := uc.telegram.ParseChannelMetaByInvite(ctx, channel.InviteLink)
+	if err != nil {
+		log.Warn().Err(err).Int64("telegram_id", channel.TelegramID).Msg("Invite rejoin failed, marking inaccessible")
+		channel.IsAccessible = false
+
+		return
+	}
+
+	channel.TelegramID = updated.TelegramID
+	channel.Title = updated.Title
+	channel.AccessHash = 0
+	channel.Pts = 0
+	channel.IsAccessible = true
+}
+
+func (uc *UseCase) initializeChannel(ctx context.Context, channel domain.Channel) {
+	log.Debug().Stringer("ch", channel).Msg("init channel")
+	var (
+		updated domain.Channel
+		err     error
+	)
+
+	switch {
+	case channel.Username != "":
+		updated, err = uc.telegram.ParseChannelMeta(ctx, channel.Username)
+	case channel.InviteLink != "":
+		updated, err = uc.telegram.ParseChannelMetaByInvite(ctx, channel.InviteLink)
+	default:
+		err = errors.New("channel state invalid")
+	}
+	if err != nil {
+		if isPrivateChannelError(err) {
+			channel.IsAccessible = false
+
+			err = uc.database.UpdateChannel(ctx, channel)
+			if err != nil {
+				log.Err(err).Msg("initializeChannel.uc.database.UpdateChannel")
+			}
+		}
+
+		log.Error().Err(err).Msg("initializeChannel.ParseChannel")
+		return
+	}
+
+	channel.Title = updated.Title
+	channel.AccessHash = updated.AccessHash
+	channel.Pts = updated.Pts
+
+	const initPosts = 20
+	posts, err := uc.telegram.GetChannelHistory(ctx, channel, initPosts)
+	if err != nil {
+		log.Error().Err(err).Msg("telegram.GetChannelHistory")
+		return
+	}
+
+	err = transaction.Wrap(ctx, func(ctx context.Context) error {
+		err = uc.database.UpdateChannel(ctx, channel)
+		if err != nil {
+			return fmt.Errorf("database.UpdateChannel: %w", err)
+		}
+
+		for _, p := range posts {
+			log.Info().Msgf("New post (init): %s - Views: %d", p.Link, p.Views)
+
+			err = uc.database.CreatePost(ctx, p)
+			if err != nil {
+				return fmt.Errorf("database.CreatePost: %w", err)
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		log.Error().Err(err).Msg("transaction.Wrap")
+	}
+}