diff --git a/tg_parser/internal/adapter/database/get_channels_with_tracked_posts.go b/tg_parser/internal/adapter/database/get_channels_with_tracked_posts.go index 6e53509..c09ceb0 100644 --- a/tg_parser/internal/adapter/database/get_channels_with_tracked_posts.go +++ b/tg_parser/internal/adapter/database/get_channels_with_tracked_posts.go @@ -11,19 +11,21 @@ import ( ) func (d *Database) GetChannelsWithTrackedPosts(ctx context.Context) ([]domain.Channel, error) { - query := `SELECT DISTINCT + query := `SELECT DISTINCT c.id, c.telegram_id, c.username, c.title, c.access_hash, c.pts, - c.invite_link + c.invite_link, + c.is_accessible FROM channel c INNER JOIN placement p ON p.channel_id = c.id INNER JOIN placement_post pp ON pp.placement_id = p.id INNER JOIN post po ON po.id = pp.post_id -WHERE po.deleted_from_channel_at IS NULL;` +WHERE po.deleted_from_channel_at IS NULL + AND c.is_accessible = true;` txOrPool := transaction.TryExtractTX(ctx) @@ -50,13 +52,14 @@ WHERE po.deleted_from_channel_at IS NULL;` } type getChannelsWithTrackedPostsDTO struct { - ID pgtype.UUID - TelegramID pgtype.Int8 - Username pgtype.Text - Title pgtype.Text - AccessHash pgtype.Int8 - Pts pgtype.Int4 - InviteLink pgtype.Text + ID pgtype.UUID + TelegramID pgtype.Int8 + Username pgtype.Text + Title pgtype.Text + AccessHash pgtype.Int8 + Pts pgtype.Int4 + InviteLink pgtype.Text + IsAccessible pgtype.Bool } func (dto *getChannelsWithTrackedPostsDTO) destination() []any { @@ -68,17 +71,19 @@ func (dto *getChannelsWithTrackedPostsDTO) destination() []any { &dto.AccessHash, &dto.Pts, &dto.InviteLink, + &dto.IsAccessible, } } func (dto *getChannelsWithTrackedPostsDTO) toDomain() domain.Channel { return domain.Channel{ - ID: dto.ID.Bytes, - TelegramID: domain.NormalizeChatID(dto.TelegramID.Int64), - Username: dto.Username.String, - Title: dto.Title.String, - AccessHash: dto.AccessHash.Int64, - Pts: int(dto.Pts.Int32), - InviteLink: dto.InviteLink.String, + ID: dto.ID.Bytes, + TelegramID: domain.NormalizeChatID(dto.TelegramID.Int64), + Username: dto.Username.String, + Title: dto.Title.String, + AccessHash: dto.AccessHash.Int64, + Pts: int(dto.Pts.Int32), + InviteLink: dto.InviteLink.String, + IsAccessible: dto.IsAccessible.Bool, } } diff --git a/tg_parser/internal/adapter/database/update_channel.go b/tg_parser/internal/adapter/database/update_channel.go index 624dc1b..34d6ff7 100644 --- a/tg_parser/internal/adapter/database/update_channel.go +++ b/tg_parser/internal/adapter/database/update_channel.go @@ -5,11 +5,9 @@ import ( "fmt" "strings" - "github.com/jackc/pgx/v5/pgtype" - "github.com/rs/zerolog/log" - "github.com/TelegramExchange/pkg/transaction" "github.com/TelegramExchange/tgex-backend/tg_parser/internal/domain" + "github.com/jackc/pgx/v5/pgtype" ) func (d *Database) UpdateChannel(ctx context.Context, channel domain.Channel) error { @@ -42,15 +40,6 @@ WHERE id = $1;` inviteLinkDTO = pgtype.Text{Valid: false} } - // Log warning if both identity fields are empty (will violate CHECK constraint) - if !usernameDTO.Valid && !inviteLinkDTO.Valid { - log.Warn(). - Str("channel_id", channel.ID.String()). - Int64("telegram_id", channel.TelegramID). - Str("title", channel.Title). - Msg("UpdateChannel: both username and invite_link are empty, this may violate CHECK constraint") - } - dto := updateChannelDTO{ ID: pgtype.UUID{Bytes: channel.ID, Valid: true}, TelegramID: pgtype.Int8{Int64: channel.TelegramID, Valid: true}, diff --git a/tg_parser/internal/adapter/telegram/history.go b/tg_parser/internal/adapter/telegram/history.go index 79a4e99..28e5d75 100644 --- a/tg_parser/internal/adapter/telegram/history.go +++ b/tg_parser/internal/adapter/telegram/history.go @@ -31,7 +31,7 @@ func (t *Telegram) GetChannelHistory(ctx context.Context, channel domain.Channel return nil, nil } - placements := make([]domain.Post, 0, len(messages.Messages)) + posts := make([]domain.Post, 0, len(messages.Messages)) for _, raw := range messages.Messages { m, ok := raw.(*tg.Message) @@ -42,8 +42,8 @@ func (t *Telegram) GetChannelHistory(ctx context.Context, channel domain.Channel text := messageToHTML(m.Message, m.Entities) publishedAt := time.Unix(int64(m.Date), 0).UTC() p := domain.NewPost(channel, m.ID, text, m.Views, publishedAt) - placements = append(placements, p) + posts = append(posts, p) } - return placements, nil + return posts, nil } diff --git a/tg_parser/internal/adapter/telegram/resolve.go b/tg_parser/internal/adapter/telegram/resolve.go index e09a33a..a637eec 100644 --- a/tg_parser/internal/adapter/telegram/resolve.go +++ b/tg_parser/internal/adapter/telegram/resolve.go @@ -5,7 +5,6 @@ import ( "fmt" "github.com/TelegramExchange/tgex-backend/tg_parser/internal/domain" - "github.com/google/uuid" "github.com/gotd/td/tg" ) @@ -49,7 +48,6 @@ func (t *Telegram) ParseChannelMeta(ctx context.Context, username string) (domai } return domain.Channel{ - ID: uuid.New(), TelegramID: domain.ChatIDFromChannelID(ch.ID), Username: username, Title: ch.Title, diff --git a/tg_parser/internal/adapter/telegram/resolve_invite.go b/tg_parser/internal/adapter/telegram/resolve_invite.go index 39fa37f..406627d 100644 --- a/tg_parser/internal/adapter/telegram/resolve_invite.go +++ b/tg_parser/internal/adapter/telegram/resolve_invite.go @@ -5,7 +5,6 @@ import ( "fmt" "github.com/TelegramExchange/tgex-backend/tg_parser/internal/domain" - "github.com/google/uuid" "github.com/gotd/td/telegram/deeplink" "github.com/gotd/td/tg" ) @@ -64,12 +63,12 @@ func (t *Telegram) ParseChannelMetaByInvite(ctx context.Context, inviteLink stri } return domain.Channel{ - ID: uuid.New(), TelegramID: domain.ChatIDFromChannelID(channel.ID), Username: username, Title: channel.Title, AccessHash: channel.AccessHash, Pts: pts, + InviteLink: inviteLink, IsAccessible: true, }, nil } diff --git a/tg_parser/internal/usecase/fetch_channel_meta.go b/tg_parser/internal/usecase/fetch_channel_meta.go index 0fbf8a3..0e790ce 100644 --- a/tg_parser/internal/usecase/fetch_channel_meta.go +++ b/tg_parser/internal/usecase/fetch_channel_meta.go @@ -2,10 +2,21 @@ package usecase import ( "context" + "fmt" "github.com/TelegramExchange/tgex-backend/tg_parser/internal/domain" ) func (uc *UseCase) FetchChannelMeta(ctx context.Context, username string) (domain.Channel, error) { - return uc.telegram.ParseChannelMeta(ctx, username) + channel, err := uc.telegram.ParseChannelMeta(ctx, username) + if err != nil { + return domain.Channel{}, fmt.Errorf("uc.telegram.ParseChannelMeta: %s", err) + } + + err = uc.database.UpdateChannelIfNotAccessible(ctx, channel.TelegramID) + if err != nil { + return domain.Channel{}, fmt.Errorf("uc.database.UpdateChannelIfNotAccessible: %s", err) + } + + return channel, nil } diff --git a/tg_parser/internal/usecase/fetch_channel_meta_by_invite.go b/tg_parser/internal/usecase/fetch_channel_meta_by_invite.go index a358003..6f65e11 100644 --- a/tg_parser/internal/usecase/fetch_channel_meta_by_invite.go +++ b/tg_parser/internal/usecase/fetch_channel_meta_by_invite.go @@ -2,10 +2,21 @@ package usecase import ( "context" + "fmt" "github.com/TelegramExchange/tgex-backend/tg_parser/internal/domain" ) func (uc *UseCase) FetchChannelMetaByInvite(ctx context.Context, inviteLink string) (domain.Channel, error) { - return uc.telegram.ParseChannelMetaByInvite(ctx, inviteLink) + channel, err := uc.telegram.ParseChannelMetaByInvite(ctx, inviteLink) + if err != nil { + return domain.Channel{}, fmt.Errorf("uc.telegram.ParseChannelMetaByInvite: %s", err) + } + + err = uc.database.UpdateChannelIfNotAccessible(ctx, channel.TelegramID) + if err != nil { + return domain.Channel{}, fmt.Errorf("uc.database.UpdateChannelIfNotAccessible: %s", err) + } + + return channel, nil } diff --git a/tg_parser/internal/usecase/fetch_channels.go b/tg_parser/internal/usecase/fetch_channels.go index a2fdb2b..8ff62f8 100644 --- a/tg_parser/internal/usecase/fetch_channels.go +++ b/tg_parser/internal/usecase/fetch_channels.go @@ -2,6 +2,7 @@ package usecase import ( "context" + "errors" "fmt" "github.com/TelegramExchange/pkg/transaction" @@ -16,21 +17,18 @@ func (uc *UseCase) FetchChannels(ctx context.Context) error { 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") + 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) } } @@ -38,37 +36,13 @@ func (uc *UseCase) FetchChannels(ctx context.Context) error { 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 { - 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, marking inaccessible") - channel.IsAccessible = false - if err := uc.database.UpdateChannel(ctx, channel); err != nil { - return fmt.Errorf("database.UpdateChannel (mark inaccessible): %w", err) - } - 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, marking inaccessible") - channel.IsAccessible = false - if err := uc.database.UpdateChannel(ctx, channel); err != nil { - return fmt.Errorf("database.UpdateChannel (mark inaccessible): %w", err) - } - return nil - } - updatedChannel.ID = channel.ID - updatedChannel.InviteLink = channel.InviteLink - updatedChannel.IsAccessible = true - 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) } @@ -97,49 +71,90 @@ func (uc *UseCase) processChannel(ctx context.Context, channel domain.Channel) e 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 + } + + 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) 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)") + } + }() -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 + if channel.InviteLink == "" { + log.Warn().Int64("telegram_id", channel.TelegramID).Msg("No invite link stored, marking inaccessible") + channel.IsAccessible = false + + return } - // Preserve persistent channel ID from the database so foreign keys remain valid. - c.ID = channel.ID - - posts, err := uc.telegram.GetChannelHistory(ctx, c, 20) + updated, err := uc.telegram.ParseChannelMetaByInvite(ctx, channel.InviteLink) if err != nil { - return domain.Channel{}, fmt.Errorf("telegram.GetChannelHistory: %w", err) + 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) { + 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, c) + err = uc.database.UpdateChannel(ctx, channel) if err != nil { return fmt.Errorf("database.UpdateChannel: %w", err) } @@ -155,8 +170,6 @@ func (uc *UseCase) initializeChannel(ctx context.Context, channel domain.Channel return nil }) if err != nil { - return domain.Channel{}, fmt.Errorf("transaction.Wrap: %w", err) + log.Error().Err(err).Msg("transaction.Wrap") } - - return c, nil } diff --git a/tg_parser/internal/usecase/fetch_views.go b/tg_parser/internal/usecase/fetch_views.go index 3a75d9e..338cadd 100644 --- a/tg_parser/internal/usecase/fetch_views.go +++ b/tg_parser/internal/usecase/fetch_views.go @@ -23,6 +23,14 @@ func (uc *UseCase) FetchViews(ctx context.Context) error { err = uc.telegram.UpdatePostsViews(ctx, channel, posts) 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") + } + return nil + } return err } diff --git a/tg_parser/internal/usecase/usecase.go b/tg_parser/internal/usecase/usecase.go index 7516578..2e50e75 100644 --- a/tg_parser/internal/usecase/usecase.go +++ b/tg_parser/internal/usecase/usecase.go @@ -20,6 +20,7 @@ type Database interface { CreateViewsSnapshot(ctx context.Context, snapshot domain.ViewsSnapshot) error + UpdateChannelIfNotAccessible(ctx context.Context, channel domain.Channel) error GetChannels(ctx context.Context) []domain.Channel UpdateChannel(ctx context.Context, channel domain.Channel) error GetChannelsWithTrackedPosts(ctx context.Context) ([]domain.Channel, error)