package usecase import ( "context" "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) for _, c := range channels { if c.Pts == 0 { updatedChannel, err := uc.initializeChannel(ctx, c) if err != nil { return fmt.Errorf("uc.initializeChannel: %w", err) } c = updatedChannel } if c.Pts == 0 { log.Warn().Int64("telegram_id", c.TelegramID).Msg("Channel pts still empty, skipping diff") continue } err := uc.processChannel(ctx, c) if err != nil { return fmt.Errorf("uc.processChannel: %w", err) } } return nil } func (uc *UseCase) processChannel(ctx context.Context, channel domain.Channel) error { diff, err := uc.telegram.GetChannelDiff(ctx, channel, 20) if err != nil { if isPrivateChannelError(err) { log.Warn().Err(err).Int64("telegram_id", channel.TelegramID).Msg("Private channel, attempting rejoin") if channel.InviteLink == "" { log.Warn().Int64("telegram_id", channel.TelegramID).Msg("No invite link stored, skipping channel") return nil } updatedChannel, joinErr := uc.telegram.ParseChannelMetaByInvite(ctx, channel.InviteLink) if joinErr != nil { log.Warn().Err(joinErr).Int64("telegram_id", channel.TelegramID).Msg("Invite rejoin failed") return nil } updatedChannel.ID = channel.ID updatedChannel.InviteLink = channel.InviteLink updatedChannel.Pts = 0 if err := uc.database.UpdateChannel(ctx, updatedChannel); err != nil { return fmt.Errorf("database.UpdateChannel: %w", err) } return 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) } } else { channel.Pts = diff.NewPts err = uc.database.UpdateChannel(ctx, channel) if err != nil { return fmt.Errorf("database.UpdateChannel: %w", err) } } return nil } func isPrivateChannelError(err error) bool { return tgerr.Is(err, "CHANNEL_PRIVATE", "CHANNEL_INVALID", "CHANNEL_FORBIDDEN") } func (uc *UseCase) initializeChannel(ctx context.Context, channel domain.Channel) (domain.Channel, error) { c := channel if channel.Username != "" { updated, err := uc.telegram.ParseChannelMeta(ctx, channel.Username) if err != nil { return domain.Channel{}, fmt.Errorf("telegram.ParseChannelMeta: %w", err) } c = updated } else if channel.AccessHash == 0 { return domain.Channel{}, fmt.Errorf("channel %s missing access hash", channel.ID) } else { pts, err := uc.telegram.GetChannelPTS(ctx, channel) if err != nil { return domain.Channel{}, fmt.Errorf("telegram.GetChannelPTS: %w", err) } c.Pts = pts } // Preserve persistent channel ID from the database so foreign keys remain valid. c.ID = channel.ID posts, err := uc.telegram.GetChannelHistory(ctx, c, 20) if err != nil { return domain.Channel{}, fmt.Errorf("telegram.GetChannelHistory: %w", err) } err = transaction.Wrap(ctx, func(ctx context.Context) error { err = uc.database.UpdateChannel(ctx, c) 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 { return domain.Channel{}, fmt.Errorf("transaction.Wrap: %w", err) } return c, nil }