парсер

tg_parser/internal/adapter/database/create_post.go

@@ -0,0 +1,42 @@
+package database
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5/pgtype"
+
+	"github.com/TelegramExchange/tgex-parser/internal/domain"
+	"github.com/TelegramExchange/tgex-parser/pkg/transaction"
+)
+
+func (d *Database) CreatePost(ctx context.Context, post domain.Post) error {
+	query := `INSERT INTO post (id, channel_id, message_id, text)
+VALUES ($1, $2, $3, $4)
+ON CONFLICT (channel_id, message_id) DO UPDATE
+SET text = EXCLUDED.text,
+    updated_at = CURRENT_TIMESTAMP;`
+
+	txOrPool := transaction.TryExtractTX(ctx)
+
+	dto := createPostDTO{
+		ID:        pgtype.UUID{Bytes: post.ID, Valid: true},
+		ChannelID: pgtype.UUID{Bytes: post.ChannelID, Valid: true},
+		MessageID: pgtype.Int4{Int32: int32(post.MessageID), Valid: true},
+		Text:      pgtype.Text{String: post.Text, Valid: true},
+	}
+
+	_, err := txOrPool.Exec(ctx, query, dto.ID, dto.ChannelID, dto.MessageID, dto.Text)
+	if err != nil {
+		return fmt.Errorf("txOrPool.Exec: %w", err)
+	}
+
+	return nil
+}
+
+type createPostDTO struct {
+	ID        pgtype.UUID
+	ChannelID pgtype.UUID
+	MessageID pgtype.Int4
+	Text      pgtype.Text
+}

tg_parser/internal/adapter/database/create_views_snapshot.go

@@ -0,0 +1,40 @@
+package database
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+
+	"github.com/TelegramExchange/tgex-parser/internal/domain"
+	"github.com/TelegramExchange/tgex-parser/pkg/transaction"
+)
+
+func (d *Database) CreateViewsSnapshot(ctx context.Context, snapshot domain.ViewsSnapshot) error {
+	query := `INSERT INTO post_views_history (id, views_count, fetched_at, post_id)
+VALUES ($1, $2, $3, $4);`
+
+	txOrPool := transaction.TryExtractTX(ctx)
+
+	dto := createViewsSnapshotDTO{
+		ID:         pgtype.UUID{Bytes: uuid.New(), Valid: true},
+		ViewsCount: pgtype.Int4{Int32: int32(snapshot.ViewsCount), Valid: true},
+		FetchedAt:  pgtype.Timestamptz{Time: snapshot.FetchedAt, Valid: true},
+		PostID:     pgtype.UUID{Bytes: snapshot.PostID, Valid: true},
+	}
+
+	_, err := txOrPool.Exec(ctx, query, dto.ID, dto.ViewsCount, dto.FetchedAt, dto.PostID)
+	if err != nil {
+		return fmt.Errorf("txOrPool.Exec: %w", err)
+	}
+
+	return nil
+}
+
+type createViewsSnapshotDTO struct {
+	ID         pgtype.UUID
+	ViewsCount pgtype.Int4
+	FetchedAt  pgtype.Timestamptz
+	PostID     pgtype.UUID
+}

tg_parser/internal/adapter/database/database.go

@@ -0,0 +1,7 @@
+package database
+
+type Database struct{}
+
+func New() *Database {
+	return &Database{}
+}

tg_parser/internal/adapter/database/delete_post.go

@@ -0,0 +1,37 @@
+package database
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5/pgtype"
+
+	"github.com/TelegramExchange/tgex-parser/internal/domain"
+	"github.com/TelegramExchange/tgex-parser/pkg/transaction"
+)
+
+func (d *Database) DeletePost(ctx context.Context, p domain.Post) error {
+	query := `UPDATE post
+SET deleted_from_channel_at = CURRENT_TIMESTAMP,
+    updated_at = CURRENT_TIMESTAMP
+WHERE channel_id = $1 AND message_id = $2;`
+
+	txOrPool := transaction.TryExtractTX(ctx)
+
+	dto := deletePostDTO{
+		ChannelID: pgtype.UUID{Bytes: p.ChannelID, Valid: true},
+		MessageID: pgtype.Int4{Int32: int32(p.MessageID), Valid: true},
+	}
+
+	_, err := txOrPool.Exec(ctx, query, dto.ChannelID, dto.MessageID)
+	if err != nil {
+		return fmt.Errorf("txOrPool.Exec: %w", err)
+	}
+
+	return nil
+}
+
+type deletePostDTO struct {
+	ChannelID pgtype.UUID
+	MessageID pgtype.Int4
+}

tg_parser/internal/adapter/database/get_channels.go

@@ -0,0 +1,79 @@
+package database
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/rs/zerolog/log"
+
+	"github.com/TelegramExchange/tgex-parser/internal/domain"
+	"github.com/TelegramExchange/tgex-parser/pkg/transaction"
+)
+
+func (d *Database) GetChannels(ctx context.Context) []domain.Channel {
+	query := `SELECT
+    id,
+    telegram_id,
+    username,
+    title,
+    access_hash,
+    pts
+FROM channel
+WHERE deleted_at IS NULL;`
+
+	txOrPool := transaction.TryExtractTX(ctx)
+
+	rows, err := txOrPool.Query(ctx, query)
+	if err != nil {
+		log.Error().Err(err).Msg("txOrPool.Query failed")
+		return []domain.Channel{}
+	}
+	defer rows.Close()
+
+	channels := make([]domain.Channel, 0)
+
+	for rows.Next() {
+		var dto getChannelsDTO
+
+		err = rows.Scan(dto.destination()...)
+		if err != nil {
+			log.Error().Err(err).Msg("rows.Scan failed")
+			return []domain.Channel{}
+		}
+
+		channels = append(channels, dto.toDomain())
+	}
+
+	return channels
+}
+
+type getChannelsDTO struct {
+	ID         pgtype.UUID
+	TelegramID pgtype.Int8
+	Username   pgtype.Text
+	Title      pgtype.Text
+	AccessHash pgtype.Int8
+	Pts        pgtype.Int4
+}
+
+func (dto *getChannelsDTO) destination() []any {
+	return []any{
+		&dto.ID,
+		&dto.TelegramID,
+		&dto.Username,
+		&dto.Title,
+		&dto.AccessHash,
+		&dto.Pts,
+	}
+}
+
+func (dto *getChannelsDTO) 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),
+	}
+}

tg_parser/internal/adapter/database/get_channels_with_tracked_posts.go

@@ -0,0 +1,79 @@
+package database
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5/pgtype"
+
+	"github.com/TelegramExchange/tgex-parser/internal/domain"
+	"github.com/TelegramExchange/tgex-parser/pkg/transaction"
+)
+
+func (d *Database) GetChannelsWithTrackedPosts(ctx context.Context) ([]domain.Channel, error) {
+	query := `SELECT DISTINCT 
+    c.id,
+    c.telegram_id,
+    c.username,
+    c.title,
+    c.access_hash,
+    c.pts
+FROM channel c
+INNER JOIN placement p ON p.placement_channel_id = c.id
+INNER JOIN post po ON po.id = p.post_id
+WHERE po.deleted_from_channel_at IS NULL;`
+
+	txOrPool := transaction.TryExtractTX(ctx)
+
+	rows, err := txOrPool.Query(ctx, query)
+	if err != nil {
+		return nil, fmt.Errorf("txOrPool.Query: %w", err)
+	}
+	defer rows.Close()
+
+	channels := make([]domain.Channel, 0)
+
+	for rows.Next() {
+		var dto getChannelsWithTrackedPostsDTO
+
+		err = rows.Scan(dto.destination()...)
+		if err != nil {
+			return nil, fmt.Errorf("rows.Scan: %w", err)
+		}
+
+		channels = append(channels, dto.toDomain())
+	}
+
+	return channels, nil
+}
+
+type getChannelsWithTrackedPostsDTO struct {
+	ID         pgtype.UUID
+	TelegramID pgtype.Int8
+	Username   pgtype.Text
+	Title      pgtype.Text
+	AccessHash pgtype.Int8
+	Pts        pgtype.Int4
+}
+
+func (dto *getChannelsWithTrackedPostsDTO) destination() []any {
+	return []any{
+		&dto.ID,
+		&dto.TelegramID,
+		&dto.Username,
+		&dto.Title,
+		&dto.AccessHash,
+		&dto.Pts,
+	}
+}
+
+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),
+	}
+}

tg_parser/internal/adapter/database/get_tracked_posts.go

@@ -0,0 +1,90 @@
+package database
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5/pgtype"
+
+	"github.com/TelegramExchange/tgex-parser/internal/domain"
+	"github.com/TelegramExchange/tgex-parser/pkg/transaction"
+)
+
+func (d *Database) GetTrackedPosts(ctx context.Context, channel domain.Channel) ([]domain.Post, error) {
+	query := `SELECT DISTINCT
+    p.id,
+    p.channel_id,
+    p.message_id,
+    p.text,
+    CASE
+        WHEN c.username IS NOT NULL THEN CONCAT('https://t.me/', c.username, '/', p.message_id)
+        ELSE ''
+    END as link,
+    COALESCE(
+        (SELECT pvh.views_count
+         FROM post_views_history pvh
+         WHERE pvh.post_id = p.id
+         ORDER BY pvh.fetched_at DESC
+         LIMIT 1),
+        0
+    ) as views
+FROM post p
+INNER JOIN channel c ON c.id = p.channel_id
+INNER JOIN placement pl ON pl.post_id = p.id
+WHERE p.channel_id = $1
+  AND p.deleted_from_channel_at IS NULL;`
+
+	txOrPool := transaction.TryExtractTX(ctx)
+
+	rows, err := txOrPool.Query(ctx, query, pgtype.UUID{Bytes: channel.ID, Valid: true})
+	if err != nil {
+		return nil, fmt.Errorf("txOrPool.Query: %w", err)
+	}
+	defer rows.Close()
+
+	posts := make([]domain.Post, 0)
+
+	for rows.Next() {
+		var dto getTrackedPostsDTO
+
+		err = rows.Scan(dto.destination()...)
+		if err != nil {
+			return nil, fmt.Errorf("rows.Scan: %w", err)
+		}
+
+		posts = append(posts, dto.toDomain())
+	}
+
+	return posts, nil
+}
+
+type getTrackedPostsDTO struct {
+	ID        pgtype.UUID
+	ChannelID pgtype.UUID
+	MessageID pgtype.Int4
+	Text      pgtype.Text
+	Link      pgtype.Text
+	Views     pgtype.Int4
+}
+
+func (dto *getTrackedPostsDTO) destination() []any {
+	return []any{
+		&dto.ID,
+		&dto.ChannelID,
+		&dto.MessageID,
+		&dto.Text,
+		&dto.Link,
+		&dto.Views,
+	}
+}
+
+func (dto *getTrackedPostsDTO) toDomain() domain.Post {
+	return domain.Post{
+		ID:        dto.ID.Bytes,
+		ChannelID: dto.ChannelID.Bytes,
+		MessageID: int(dto.MessageID.Int32),
+		Text:      dto.Text.String,
+		Link:      dto.Link.String,
+		Views:     int(dto.Views.Int32),
+	}
+}

tg_parser/internal/adapter/database/update_channel.go

@@ -0,0 +1,49 @@
+package database
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5/pgtype"
+
+	"github.com/TelegramExchange/tgex-parser/internal/domain"
+	"github.com/TelegramExchange/tgex-parser/pkg/transaction"
+)
+
+func (d *Database) UpdateChannel(ctx context.Context, channel domain.Channel) error {
+	query := `UPDATE channel
+SET telegram_id = $2,
+    username = $3,
+    title = $4,
+    access_hash = $5,
+    pts = $6,
+    updated_at = CURRENT_TIMESTAMP
+WHERE id = $1;`
+
+	txOrPool := transaction.TryExtractTX(ctx)
+
+	dto := updateChannelDTO{
+		ID:         pgtype.UUID{Bytes: channel.ID, Valid: true},
+		TelegramID: pgtype.Int8{Int64: channel.TelegramID, Valid: true},
+		Username:   pgtype.Text{String: channel.Username, Valid: true},
+		Title:      pgtype.Text{String: channel.Title, Valid: true},
+		AccessHash: pgtype.Int8{Int64: channel.AccessHash, Valid: true},
+		Pts:        pgtype.Int4{Int32: int32(channel.Pts), Valid: true},
+	}
+
+	_, err := txOrPool.Exec(ctx, query, dto.ID, dto.TelegramID, dto.Username, dto.Title, dto.AccessHash, dto.Pts)
+	if err != nil {
+		return fmt.Errorf("txOrPool.Exec: %w", err)
+	}
+
+	return nil
+}
+
+type updateChannelDTO struct {
+	ID         pgtype.UUID
+	TelegramID pgtype.Int8
+	Username   pgtype.Text
+	Title      pgtype.Text
+	AccessHash pgtype.Int8
+	Pts        pgtype.Int4
+}

tg_parser/internal/adapter/telegram/get_active_views.go

@@ -0,0 +1,37 @@
+package telegram
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/TelegramExchange/tgex-parser/internal/domain"
+	"github.com/gotd/td/tg"
+)
+
+func (t *Telegram) UpdatePostsViews(ctx context.Context, channel domain.Channel, posts []domain.Post) error {
+	ids := make([]int, len(posts))
+	for i, p := range posts {
+		ids[i] = p.MessageID
+	}
+
+	req := &tg.MessagesGetMessagesViewsRequest{
+		Peer: &tg.InputPeerChannel{
+			ChannelID:  channel.ChannelID(),
+			AccessHash: channel.AccessHash,
+		},
+		ID:        ids,
+		Increment: false,
+	}
+
+	resp, err := t.API().MessagesGetMessagesViews(ctx, req)
+	if err != nil {
+		return fmt.Errorf("get messages views: %w", err)
+	}
+
+	// Telegram гарантирует, что resp.Views соответствует порядку ids
+	for i, v := range resp.Views {
+		posts[i].Views = v.Views
+	}
+
+	return nil
+}

tg_parser/internal/adapter/telegram/get_channel_diff.go

@@ -0,0 +1,119 @@
+package telegram
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/TelegramExchange/tgex-parser/internal/domain"
+	"github.com/gotd/td/tg"
+)
+
+func (t *Telegram) GetChannelDiff(ctx context.Context, channel domain.Channel, limit int) (domain.ChannelDiff, error) {
+	req := &tg.UpdatesGetChannelDifferenceRequest{
+		Channel: &tg.InputChannel{
+			ChannelID:  channel.ChannelID(),
+			AccessHash: channel.AccessHash,
+		},
+		Filter: &tg.ChannelMessagesFilterEmpty{},
+		Pts:    channel.Pts,
+		Limit:  limit,
+	}
+
+	rawDiff, err := t.API().UpdatesGetChannelDifference(ctx, req)
+	if err != nil {
+		return domain.ChannelDiff{}, fmt.Errorf("get difference: %w", err)
+	}
+
+	result := domain.ChannelDiff{}
+
+	switch d := rawDiff.(type) {
+	case *tg.UpdatesChannelDifferenceEmpty:
+		result.NewPts = d.Pts
+
+	case *tg.UpdatesChannelDifferenceTooLong:
+		if dialog, ok := d.Dialog.(*tg.Dialog); ok {
+			result.NewPts = dialog.Pts
+		}
+		result.NewPosts = extractPosts(channel, d.Messages)
+		result.UpdatedChannel = extractChannelMeta(channel, d.Chats)
+
+	case *tg.UpdatesChannelDifference:
+		result.NewPts = d.Pts
+		result.NewPosts = extractPosts(channel, d.NewMessages)
+		result.DeletedPosts = extractDeletedPosts(channel, d.OtherUpdates)
+		result.UpdatedChannel = extractChannelMeta(channel, d.Chats)
+
+	default:
+		return domain.ChannelDiff{}, fmt.Errorf("unexpected rawDiff type: %T", rawDiff)
+	}
+
+	return result, nil
+}
+
+func extractPosts(channel domain.Channel, msgs []tg.MessageClass) []domain.Post {
+	posts := make([]domain.Post, 0, len(msgs))
+
+	for _, raw := range msgs {
+		m, ok := raw.(*tg.Message)
+		if !ok {
+			continue
+		}
+
+		p := domain.NewPost(channel, m.ID, m.Message, m.Views)
+		posts = append(posts, p)
+	}
+
+	return posts
+}
+
+func extractDeletedPosts(channel domain.Channel, updates []tg.UpdateClass) []domain.Post {
+	var deleted []domain.Post
+
+	for _, upd := range updates {
+		if u, ok := upd.(*tg.UpdateDeleteChannelMessages); ok {
+			for _, msgID := range u.Messages {
+				p := domain.NewPost(channel, msgID, "", 0)
+				deleted = append(deleted, p)
+			}
+		}
+	}
+
+	return deleted
+}
+
+func extractChannelMeta(currentChannel domain.Channel, chats []tg.ChatClass) *domain.Channel {
+	for _, chat := range chats {
+		ch, ok := chat.(*tg.Channel)
+		if !ok {
+			continue
+		}
+
+		if ch.ID != currentChannel.ChannelID() {
+			continue
+		}
+
+		username := ""
+		if usernameVal, ok := ch.GetUsername(); ok {
+			username = usernameVal
+		}
+
+		// Keep current access hash if new one is not provided or is zero
+		accessHash := currentChannel.AccessHash
+		if accessHashVal, ok := ch.GetAccessHash(); ok && accessHashVal != 0 {
+			accessHash = accessHashVal
+		}
+
+		updated := domain.Channel{
+			ID:         currentChannel.ID,
+			TelegramID: domain.ChatIDFromChannelID(ch.ID),
+			Username:   username,
+			Title:      ch.Title,
+			AccessHash: accessHash,
+			Pts:        currentChannel.Pts,
+		}
+
+		return &updated
+	}
+
+	return nil
+}

tg_parser/internal/adapter/telegram/history.go

@@ -0,0 +1,46 @@
+package telegram
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/TelegramExchange/tgex-parser/internal/domain"
+	"github.com/gotd/td/tg"
+	"github.com/rs/zerolog/log"
+)
+
+func (t *Telegram) GetChannelHistory(ctx context.Context, channel domain.Channel, limit int) ([]domain.Post, error) {
+	log.Info().Msg(channel.String())
+
+	req := &tg.MessagesGetHistoryRequest{
+		Peer: &tg.InputPeerChannel{
+			ChannelID:  channel.ChannelID(),
+			AccessHash: channel.AccessHash,
+		},
+		Limit: limit,
+	}
+
+	resp, err := t.API().MessagesGetHistory(ctx, req)
+	if err != nil {
+		return nil, fmt.Errorf("resp: %w", err)
+	}
+
+	messages, ok := resp.(*tg.MessagesChannelMessages)
+	if !ok {
+		return nil, nil
+	}
+
+	placements := make([]domain.Post, 0, len(messages.Messages))
+
+	for _, raw := range messages.Messages {
+		m, ok := raw.(*tg.Message)
+		if !ok {
+			continue
+		}
+
+		p := domain.NewPost(channel, m.ID, m.Message, m.Views)
+		placements = append(placements, p)
+	}
+
+	return placements, nil
+}

tg_parser/internal/adapter/telegram/resolve.go

@@ -0,0 +1,59 @@
+package telegram
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/TelegramExchange/tgex-parser/internal/domain"
+	"github.com/google/uuid"
+	"github.com/gotd/td/tg"
+)
+
+func (t *Telegram) ParseChannelMeta(ctx context.Context, username string) (domain.Channel, error) {
+	resolved, err := t.API().ContactsResolveUsername(ctx, &tg.ContactsResolveUsernameRequest{
+		Username: username,
+	})
+	if err != nil {
+		return domain.Channel{}, fmt.Errorf("resolve username: %w", err)
+	}
+
+	if len(resolved.Chats) == 0 {
+		return domain.Channel{}, fmt.Errorf("no chats in resolve result")
+	}
+
+	ch, ok := resolved.Chats[0].(*tg.Channel)
+	if !ok {
+		return domain.Channel{}, fmt.Errorf("not a channel")
+	}
+
+	req := []tg.InputDialogPeerClass{
+		&tg.InputDialogPeer{
+			Peer: &tg.InputPeerChannel{
+				ChannelID:  ch.ID,
+				AccessHash: ch.AccessHash,
+			},
+		},
+	}
+
+	dialogs, err := t.API().MessagesGetPeerDialogs(ctx, req)
+	if err != nil {
+		return domain.Channel{}, fmt.Errorf("peer dialogs: %w", err)
+	}
+
+	pts := 0
+	if len(dialogs.Dialogs) > 0 {
+		d, ok := dialogs.Dialogs[0].(*tg.Dialog)
+		if ok {
+			pts = d.Pts
+		}
+	}
+
+	return domain.Channel{
+		ID:         uuid.New(),
+		TelegramID: domain.ChatIDFromChannelID(ch.ID),
+		Username:   username,
+		Title:      ch.Title,
+		AccessHash: ch.AccessHash,
+		Pts:        pts,
+	}, nil
+}

tg_parser/internal/adapter/telegram/telegram.go

@@ -0,0 +1,20 @@
+package telegram
+
+import (
+	"github.com/TelegramExchange/tgex-parser/pkg/telegram"
+	"github.com/gotd/td/tg"
+)
+
+type Telegram struct {
+	client *telegram.Client
+}
+
+func New(client *telegram.Client) *Telegram {
+	return &Telegram{
+		client: client,
+	}
+}
+
+func (t *Telegram) API() *tg.Client {
+	return t.client.API()
+}