diff --git a/.env.example b/.env.example index 0135929..cef0154 100644 --- a/.env.example +++ b/.env.example @@ -1,11 +1,27 @@ -APP__URL=https://app.example.com -DB__URL=postgresql+asyncpg://user:password@localhost:5432/tgex -LOGGER__APP_NAME="tgex-backend" +# backend (src) +DB__URL=postgres://user:password@localhost:5432/tgex + +LOGGER__APP_NAME=tgex-backend LOGGER__PRETTY_CONSOLE=true -TELEGRAM__TOKEN=your_bot_token_here +LOGGER__LEVEL=info + JWT__SECRET_KEY=your_secret_key_here_please_generate_a_strong_random_string + PARSER__URL=http://localhost:8080 + S3__ENDPOINT_URL=http://localhost:9000 S3__ACCESS_KEY_ID=user S3__SECRET_ACCESS_KEY=password -S3__BUCKET_NAME=tgex-files \ No newline at end of file +S3__BUCKET_NAME=tgex-files + +# tg_bot (telegram_bot) +TELEGRAM__TOKEN=your_bot_token_here +BACKEND__BASE_URL=http://localhost:8000 +LOGIN_URL=https://app.example.com/api/v1/auth/complete?token= + +# tg_parser (tg_parser) +TELEGRAM__API_ID=123456 +TELEGRAM__API_HASH=your_api_hash_here +TELEGRAM__SESSION_FILE=/data/tg_parser.json +TELEGRAM__PHONE=+79999999999 +TELEGRAM__PASSWORD= diff --git a/.gitignore b/.gitignore index e645b08..ee448f0 100644 --- a/.gitignore +++ b/.gitignore @@ -129,4 +129,5 @@ docker-compose.yml .uv_cache/ -session.json \ No newline at end of file +session.json +tg_parser.json diff --git a/docker-compose.yml.example b/docker-compose.yml.example index f44c220..dcd1dfb 100644 --- a/docker-compose.yml.example +++ b/docker-compose.yml.example @@ -31,6 +31,7 @@ services: - minio environment: DB__URL: "postgres://user:password@postgres:5432/tgex" + PARSER__URL: "http://tg_parser:8080" S3__ENDPOINT_URL: "http://minio:9000" env_file: - .env @@ -45,6 +46,23 @@ services: - .env restart: unless-stopped + tg_parser: + build: + context: tg_parser + ports: + - "8080:8080" + depends_on: + - postgres + environment: + DB__URL: "postgres://user:password@postgres:5432/tgex" + LOGGER__LEVEL: "info" + LOGGER__PRETTY_CONSOLE: "true" + env_file: + - .env + volumes: + - ./tg_parser.session:/data/tg_parser.session + restart: unless-stopped + volumes: postgres_data: minio_data: diff --git a/src/config.py b/src/config.py index 48138a3..567f807 100644 --- a/src/config.py +++ b/src/config.py @@ -9,10 +9,6 @@ from shared.telegram_base import TelegramConfig from src.adapter.s3 import S3Config -class AppConfig(BaseModel): - URL: str - - class ParserConfig(BaseModel): URL: str @@ -20,7 +16,6 @@ class ParserConfig(BaseModel): class Settings(BaseSettings): model_config = SettingsConfigDict(env_file='.env', case_sensitive=False, env_nested_delimiter='__') - app: AppConfig db: DatabaseConfig logger: LoggerConfig telegram: TelegramConfig diff --git a/src/usecase/purchase/create_purchase.py b/src/usecase/purchase/create_purchase.py index 45501f7..11b991e 100644 --- a/src/usecase/purchase/create_purchase.py +++ b/src/usecase/purchase/create_purchase.py @@ -1,5 +1,4 @@ import logging -import random import uuid from typing import TYPE_CHECKING @@ -131,13 +130,17 @@ async def create_purchase( channel = await self.database.get_channel(username=channel_input.username) if not channel: + parser_response = await self.parser.fetch_telegram_channel(channel_input.username) + if not parser_response: + raise domain.TelegramChannelNotFound(channel_input.username) + channel = domain.Channel( - username=channel_input.username, - telegram_id=random.randint(1, 10000), - title='parser_response.title', + username=parser_response.username, + telegram_id=parser_response.telegram_id, + title=parser_response.title, ) await self.database.create_channel(channel) - log.info('Created channel @%s with telegram_id=%s', channel_input.username, channel.telegram_id) + log.info('Created channel @%s with telegram_id=%s', channel.username, channel.telegram_id) invite_link = await self.telegram_bot.create_chat_invite_link(project.channel.telegram_id, requires_approval) diff --git a/telegram_bot/main.go b/telegram_bot/main.go index 5634317..7778957 100644 --- a/telegram_bot/main.go +++ b/telegram_bot/main.go @@ -1,6 +1,7 @@ package main import ( + "os" "time" "example.com/m/adapter/backend" @@ -12,10 +13,6 @@ import ( "github.com/rs/zerolog/log" ) -const ( - botToken = "7718883894:AAHUqtjqHtSs2Rp6Tb4PokqU6syfKmcFvOo" -) - func main() { logger.Init(logger.Config{ AppName: "tgex-bot", @@ -23,10 +20,13 @@ func main() { PrettyConsole: true, }) - // Инициализация backend client + botToken := mustEnv("TELEGRAM__TOKEN") + loginURL := mustEnv("LOGIN_URL") + baseURL := os.Getenv("BACKEND__BASE_URL") + backendClient := backend.New(backend.Config{ - BaseURL: "http://backend:8000", - LoginURL: "https://unstabilising-lora-unframable.ngrok-free.dev/api/v1/auth/complete?token=", + BaseURL: baseURL, + LoginURL: loginURL, }) var mainMenu = func() bot.State { return &screens.MainMenu{} } @@ -47,3 +47,12 @@ func main() { break } } + +func mustEnv(key string) string { + value := os.Getenv(key) + if value == "" { + log.Fatal().Str("env", key).Msg("missing required environment variable") + } + + return value +} diff --git a/telegram_bot/screens/add_creative.go b/telegram_bot/screens/add_creative.go index 0a822d4..c21fafb 100644 --- a/telegram_bot/screens/add_creative.go +++ b/telegram_bot/screens/add_creative.go @@ -74,6 +74,7 @@ https://t.me/+AbCdEfGhIjKlMn`, [][]echotron.InlineKeyboardButton{ case data == "add_button": // Начинаем процесс добавления кнопки s.WaitingForButtonText = true + s.PendingButtonType = "invite" s.ShowControlPanel(b, `＋ Добавить кнопку @@ -81,7 +82,13 @@ https://t.me/+AbCdEfGhIjKlMn`, [][]echotron.InlineKeyboardButton{ Например: • Перейти на сайт -• Написать нам`, [][]echotron.InlineKeyboardButton{ +• Написать нам + +Ссылка на канал подставится автоматически при создании закупа.`, [][]echotron.InlineKeyboardButton{ + bot.Row( + bot.Button("✓ Ссылка на канал", "button_type_invite"), + bot.Button("Своя ссылка", "button_type_custom"), + ), bot.Row(bot.Button("« Отмена", "cancel_add_button")), }) @@ -115,13 +122,14 @@ https://t.me/+AbCdEfGhIjKlMn`, [][]echotron.InlineKeyboardButton{ // Если мы уже создали кнопку с placeholder - удаляем её if s.WaitingForButtonURL && len(s.Buttons) > 0 { lastButton := s.Buttons[len(s.Buttons)-1] - if lastButton.URL == "https://example.com" { + if lastButton.URL == "https://example.com" || lastButton.URL == inviteLinkPlaceholder { s.Buttons = s.Buttons[:len(s.Buttons)-1] s.UpdateCreativePreview(b) } } s.WaitingForButtonText = false s.WaitingForButtonURL = false + s.PendingButtonType = "" s.CurrentButtonText = nil if s.CreativeMessageSent { @@ -130,6 +138,42 @@ https://t.me/+AbCdEfGhIjKlMn`, [][]echotron.InlineKeyboardButton{ s.ShowManagementPanel(b, "✓ Сохранить", "confirm_create", "cancel") } + case data == "button_type_invite": + s.PendingButtonType = "invite" + s.ShowControlPanel(b, `＋ Добавить кнопку + +Введите текст кнопки: + +Например: +• Перейти на сайт +• Написать нам + +Ссылка на канал подставится автоматически при создании закупа.`, [][]echotron.InlineKeyboardButton{ + bot.Row( + bot.Button("✓ Ссылка на канал", "button_type_invite"), + bot.Button("Своя ссылка", "button_type_custom"), + ), + bot.Row(bot.Button("« Отмена", "cancel_add_button")), + }) + + case data == "button_type_custom": + s.PendingButtonType = "custom" + s.ShowControlPanel(b, `＋ Добавить кнопку + +Введите текст кнопки: + +Например: +• Перейти на сайт +• Написать нам + +Ссылка на канал подставится автоматически при создании закупа.`, [][]echotron.InlineKeyboardButton{ + bot.Row( + bot.Button("Ссылка на канал", "button_type_invite"), + bot.Button("✓ Своя ссылка", "button_type_custom"), + ), + bot.Row(bot.Button("« Отмена", "cancel_add_button")), + }) + case data == "cancel_edit": // Отменяем редактирование текста s.WaitingForTextEdit = false @@ -255,7 +299,6 @@ func (s *AddCreative) HandleMessage(b *bot.Bot, u *echotron.Update) { buttonText := u.Message.Text s.CurrentButtonText = &buttonText s.WaitingForButtonText = false - s.WaitingForButtonURL = true // Удаляем сообщение пользователя _, err := b.DeleteMessage(b.ChatID, u.Message.ID) @@ -263,22 +306,35 @@ func (s *AddCreative) HandleMessage(b *bot.Bot, u *echotron.Update) { log.Error().Err(err).Msg("Failed to delete user message") } - s.Buttons = append(s.Buttons, InlineButton{ - Text: buttonText, - URL: "https://example.com", - }) - - s.UpdateCreativePreview(b) - - s.ShowControlPanel(b, `✧ Введите URL для кнопки: + if s.PendingButtonType == "custom" { + s.WaitingForButtonURL = true + s.Buttons = append(s.Buttons, InlineButton{ + Text: buttonText, + URL: "https://example.com", + }) + s.UpdateCreativePreview(b) + s.ShowControlPanel(b, `✧ Введите URL для кнопки: URL должен начинаться с https:// Например: https://example.com https://t.me/yourchannel`, [][]echotron.InlineKeyboardButton{ - bot.Row(bot.Button("« Отмена", "cancel_add_button")), - }) + bot.Row(bot.Button("« Отмена", "cancel_add_button")), + }) + } else { + s.Buttons = append(s.Buttons, InlineButton{ + Text: buttonText, + URL: inviteLinkPlaceholder, + }) + s.UpdateCreativePreview(b) + s.PendingButtonType = "" + if s.CreativeMessageSent { + s.showCreativeConfirmation(b) + } else { + s.ShowManagementPanel(b, "✓ Сохранить", "confirm_create", "cancel") + } + } return } @@ -312,6 +368,7 @@ URL должен начинаться с https:// } s.WaitingForButtonURL = false s.CurrentButtonText = nil + s.PendingButtonType = "" s.UpdateCreativePreview(b) diff --git a/telegram_bot/screens/creative_details.go b/telegram_bot/screens/creative_details.go index d536fb4..8682ee5 100644 --- a/telegram_bot/screens/creative_details.go +++ b/telegram_bot/screens/creative_details.go @@ -101,13 +101,20 @@ https://t.me/+AbCdEfGhIjKlMn`, [][]echotron.InlineKeyboardButton{ case data == "add_button": s.WaitingForButtonText = true + s.PendingButtonType = "invite" s.ShowControlPanel(b, `＋ Добавить кнопку Введите текст кнопки: Например: • Перейти на сайт -• Написать нам`, [][]echotron.InlineKeyboardButton{ +• Написать нам + +Ссылка на канал подставится автоматически при создании закупа.`, [][]echotron.InlineKeyboardButton{ + bot.Row( + bot.Button("✓ Ссылка на канал", "button_type_invite"), + bot.Button("Своя ссылка", "button_type_custom"), + ), bot.Row(bot.Button("« Отмена", "cancel_add_button")), }) @@ -132,16 +139,53 @@ https://t.me/+AbCdEfGhIjKlMn`, [][]echotron.InlineKeyboardButton{ // Отменяем добавление кнопки if s.WaitingForButtonURL && len(s.Buttons) > 0 { lastButton := s.Buttons[len(s.Buttons)-1] - if lastButton.URL == "https://example.com" { + if lastButton.URL == "https://example.com" || lastButton.URL == inviteLinkPlaceholder { s.Buttons = s.Buttons[:len(s.Buttons)-1] s.UpdateCreativePreview(b) } } s.WaitingForButtonText = false s.WaitingForButtonURL = false + s.PendingButtonType = "" s.CurrentButtonText = nil s.showManagementPanelWithDelete(b) + case data == "button_type_invite": + s.PendingButtonType = "invite" + s.ShowControlPanel(b, `＋ Добавить кнопку + +Введите текст кнопки: + +Например: +• Перейти на сайт +• Написать нам + +Ссылка на канал подставится автоматически при создании закупа.`, [][]echotron.InlineKeyboardButton{ + bot.Row( + bot.Button("✓ Ссылка на канал", "button_type_invite"), + bot.Button("Своя ссылка", "button_type_custom"), + ), + bot.Row(bot.Button("« Отмена", "cancel_add_button")), + }) + + case data == "button_type_custom": + s.PendingButtonType = "custom" + s.ShowControlPanel(b, `＋ Добавить кнопку + +Введите текст кнопки: + +Например: +• Перейти на сайт +• Написать нам + +Ссылка на канал подставится автоматически при создании закупа.`, [][]echotron.InlineKeyboardButton{ + bot.Row( + bot.Button("Ссылка на канал", "button_type_invite"), + bot.Button("✓ Своя ссылка", "button_type_custom"), + ), + bot.Row(bot.Button("« Отмена", "cancel_add_button")), + }) + case data == "cancel_edit", data == "cancel_media": s.WaitingForMedia = false s.showManagementPanelWithDelete(b) @@ -235,7 +279,6 @@ func (s *CreativeDetails) HandleMessage(b *bot.Bot, u *echotron.Update) { buttonText := u.Message.Text s.CurrentButtonText = &buttonText s.WaitingForButtonText = false - s.WaitingForButtonURL = true // Удаляем сообщение пользователя _, err := b.DeleteMessage(b.ChatID, u.Message.ID) @@ -244,19 +287,28 @@ func (s *CreativeDetails) HandleMessage(b *bot.Bot, u *echotron.Update) { } // Создаём кнопку с placeholder URL - s.Buttons = append(s.Buttons, InlineButton{ - Text: buttonText, - URL: "https://example.com", - }) - - s.UpdateCreativePreview(b) - - s.ShowControlPanel(b, `✧ Введите URL для кнопки: + if s.PendingButtonType == "custom" { + s.WaitingForButtonURL = true + s.Buttons = append(s.Buttons, InlineButton{ + Text: buttonText, + URL: "https://example.com", + }) + s.UpdateCreativePreview(b) + s.ShowControlPanel(b, `✧ Введите URL для кнопки: Например: https://example.com`, [][]echotron.InlineKeyboardButton{ - bot.Row(bot.Button("« Отмена", "cancel_add_button")), - }) + bot.Row(bot.Button("« Отмена", "cancel_add_button")), + }) + } else { + s.Buttons = append(s.Buttons, InlineButton{ + Text: buttonText, + URL: inviteLinkPlaceholder, + }) + s.UpdateCreativePreview(b) + s.PendingButtonType = "" + s.showManagementPanelWithDelete(b) + } return } @@ -289,6 +341,7 @@ URL должен начинаться с http:// или https s.WaitingForButtonURL = false s.CurrentButtonText = nil + s.PendingButtonType = "" s.UpdateCreativePreview(b) s.showManagementPanelWithDelete(b) diff --git a/telegram_bot/screens/creative_editor.go b/telegram_bot/screens/creative_editor.go index da3a0e9..a5bcd73 100644 --- a/telegram_bot/screens/creative_editor.go +++ b/telegram_bot/screens/creative_editor.go @@ -33,6 +33,7 @@ type CreativeEditorFields struct { WaitingForButtonURL bool CurrentButtonText *string WaitingForMedia bool + PendingButtonType string // ID сообщения с креативом (которое постоянно обновляется) CreativeMessageID *int @@ -43,6 +44,9 @@ type CreativeEditorFields struct { previousHasMedia bool } +const inviteLinkPlaceholder = "{{invite_link}}" +const inviteLinkPreviewURL = "https://t.me/placeholder" + // GetCreativeText возвращает текст для превью креатива func (e *CreativeEditorFields) GetCreativeText() string { var result string @@ -78,10 +82,14 @@ func (e *CreativeEditorFields) GetCreativeButtons() [][]echotron.InlineKeyboardB var buttons [][]echotron.InlineKeyboardButton for _, btn := range e.Buttons { + url := btn.URL + if url == inviteLinkPlaceholder { + url = inviteLinkPreviewURL + } buttons = append(buttons, []echotron.InlineKeyboardButton{ { Text: btn.Text, - URL: btn.URL, + URL: url, }, }) } diff --git a/telegram_bot/screens/purchase_optional_details.go b/telegram_bot/screens/purchase_optional_details.go index 3d06f97..c67ab4c 100644 --- a/telegram_bot/screens/purchase_optional_details.go +++ b/telegram_bot/screens/purchase_optional_details.go @@ -1443,9 +1443,13 @@ func (s *PurchaseOptionalDetails) sendCreativeWithInviteLink( if len(creative.Buttons) > 0 { editor.Buttons = make([]InlineButton, 0, len(creative.Buttons)) for _, btn := range creative.Buttons { + url := btn.URL + if url == inviteLinkPlaceholder { + url = inviteLink + } editor.Buttons = append(editor.Buttons, InlineButton{ Text: btn.Text, - URL: btn.URL, + URL: url, }) } } diff --git a/tg_parser/Dockerfile b/tg_parser/Dockerfile new file mode 100644 index 0000000..8964ce1 --- /dev/null +++ b/tg_parser/Dockerfile @@ -0,0 +1,17 @@ +FROM golang:1.25-alpine AS build + +WORKDIR /app + +# Modules layer +COPY go.mod go.sum ./ +RUN go mod download + +# Build layer +COPY . . +RUN CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -o /parser . + +FROM alpine:latest AS run + +COPY --from=build /parser /parser + +CMD ["/parser"] diff --git a/tg_parser/cmd/auth/main.go b/tg_parser/cmd/auth/main.go new file mode 100644 index 0000000..755b287 --- /dev/null +++ b/tg_parser/cmd/auth/main.go @@ -0,0 +1,80 @@ +package main + +import ( + "bufio" + "context" + "fmt" + "os" + "strconv" + "strings" + + "github.com/gotd/td/session" + "github.com/gotd/td/telegram" + "github.com/gotd/td/telegram/auth" + "github.com/gotd/td/tg" +) + +func main() { + apiID := mustEnvInt("TELEGRAM__API_ID") + apiHash := mustEnv("TELEGRAM__API_HASH") + sessionFile := mustEnv("TELEGRAM__SESSION_FILE") + + phone := strings.TrimSpace(os.Getenv("TELEGRAM__PHONE")) + if phone == "" { + phone = prompt("Phone (e.g. +79991234567): ") + } + + password := strings.TrimSpace(os.Getenv("TELEGRAM__PASSWORD")) + if password == "" { + password = prompt("2FA password (empty if not set): ") + } + + ctx := context.Background() + + client := telegram.NewClient(apiID, apiHash, telegram.Options{ + SessionStorage: &session.FileStorage{Path: sessionFile}, + }) + + err := client.Run(ctx, func(ctx context.Context) error { + return client.Auth().IfNecessary(ctx, auth.NewFlow( + auth.Constant(phone, password, auth.CodeAuthenticatorFunc( + func(ctx context.Context, sentCode *tg.AuthSentCode) (string, error) { + return prompt("Enter code: "), nil + }, + )), + auth.SendCodeOptions{}, + )) + }) + if err != nil { + fmt.Fprintf(os.Stderr, "auth failed: %v\n", err) + os.Exit(1) + } + + fmt.Println("✓ Authorized") +} + +func mustEnv(key string) string { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + fmt.Fprintf(os.Stderr, "missing required env: %s\n", key) + os.Exit(1) + } + return value +} + +func mustEnvInt(key string) int { + raw := mustEnv(key) + v, err := strconv.Atoi(raw) + if err != nil { + fmt.Fprintf(os.Stderr, "invalid int in %s: %v\n", key, err) + os.Exit(1) + } + return v +} + +func prompt(label string) string { + fmt.Print(label) + reader := bufio.NewReader(os.Stdin) + value, _ := reader.ReadString('\n') + return strings.TrimSpace(value) +} diff --git a/tg_parser/config/config.go b/tg_parser/config/config.go new file mode 100644 index 0000000..90bdcd8 --- /dev/null +++ b/tg_parser/config/config.go @@ -0,0 +1,44 @@ +package config + +import ( + "errors" + "fmt" + "os" + + "github.com/TelegramExchange/tgex-parser/pkg/logger" + "github.com/TelegramExchange/tgex-parser/pkg/postgres" + "github.com/joho/godotenv" + "github.com/kelseyhightower/envconfig" + + "github.com/TelegramExchange/tgex-parser/internal/controller/worker" + "github.com/TelegramExchange/tgex-parser/pkg/telegram" +) + +type HTTP struct { + Addr string `envconfig:"HTTP__ADDR" default:":8080"` +} + +type Config struct { + Logger logger.Config + Postgres postgres.Config + Telegram telegram.Config + ChannelWorker worker.ChannelConfig + ViewsWorker worker.ViewsConfig + HTTP HTTP +} + +func New() (Config, error) { + var config Config + + err := godotenv.Load(".env") + if err != nil && !errors.Is(err, os.ErrNotExist) { + return config, fmt.Errorf("godotenv.Load: %w", err) + } + + err = envconfig.Process("", &config) + if err != nil { + return config, fmt.Errorf("envconfig.Process: %w", err) + } + + return config, nil +} diff --git a/tg_parser/go.mod b/tg_parser/go.mod new file mode 100644 index 0000000..784b40c --- /dev/null +++ b/tg_parser/go.mod @@ -0,0 +1,51 @@ +module github.com/TelegramExchange/tgex-parser + +go 1.25.0 + +require ( + github.com/gotd/td v0.136.0 + github.com/jackc/pgx/v5 v5.7.6 + github.com/joho/godotenv v1.5.1 + github.com/kelseyhightower/envconfig v1.4.0 + github.com/rs/zerolog v1.34.0 +) + +require ( + github.com/cenkalti/backoff/v4 v4.3.0 // indirect + github.com/coder/websocket v1.8.14 // indirect + github.com/dlclark/regexp2 v1.11.5 // indirect + github.com/fatih/color v1.18.0 // indirect + github.com/ghodss/yaml v1.0.0 // indirect + github.com/go-faster/errors v0.7.1 // indirect + github.com/go-faster/jx v1.2.0 // indirect + github.com/go-faster/xor v1.0.0 // indirect + github.com/go-faster/yaml v0.4.6 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/gotd/ige v0.2.2 // indirect + github.com/gotd/neo v0.1.5 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/klauspost/compress v1.18.2 // indirect + github.com/mattn/go-colorable v0.1.14 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/ogen-go/ogen v1.16.0 // indirect + github.com/segmentio/asm v1.2.1 // indirect + github.com/shopspring/decimal v1.4.0 // indirect + go.opentelemetry.io/otel v1.38.0 // indirect + go.opentelemetry.io/otel/metric v1.38.0 // indirect + go.opentelemetry.io/otel/trace v1.38.0 // indirect + go.uber.org/atomic v1.11.0 // indirect + go.uber.org/multierr v1.11.0 // indirect + go.uber.org/zap v1.27.1 // indirect + golang.org/x/crypto v0.45.0 // indirect + golang.org/x/exp v0.0.0-20230725093048-515e97ebf090 // indirect + golang.org/x/mod v0.30.0 // indirect + golang.org/x/net v0.47.0 // indirect + golang.org/x/sync v0.18.0 // indirect + golang.org/x/sys v0.38.0 // indirect + golang.org/x/text v0.31.0 // indirect + golang.org/x/tools v0.39.0 // indirect + gopkg.in/yaml.v2 v2.4.0 // indirect + rsc.io/qr v0.2.0 // indirect +) diff --git a/tg_parser/go.sum b/tg_parser/go.sum new file mode 100644 index 0000000..e2a6a74 --- /dev/null +++ b/tg_parser/go.sum @@ -0,0 +1,130 @@ +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g= +github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= +github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dlclark/regexp2 v1.11.5 h1:Q/sSnsKerHeCkc/jSTNq1oCm7KiVgUMZRDUoRu0JQZQ= +github.com/dlclark/regexp2 v1.11.5/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= +github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM= +github.com/fatih/color v1.18.0/go.mod h1:4FelSpRwEGDpQ12mAdzqdOukCy4u8WUtOY6lkT/6HfU= +github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk= +github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04= +github.com/go-faster/errors v0.7.1 h1:MkJTnDoEdi9pDabt1dpWf7AA8/BaSYZqibYyhZ20AYg= +github.com/go-faster/errors v0.7.1/go.mod h1:5ySTjWFiphBs07IKuiL69nxdfd5+fzh1u7FPGZP2quo= +github.com/go-faster/jx v1.2.0 h1:T2YHJPrFaYu21fJtUxC9GzmluKu8rVIFDwwGBKTDseI= +github.com/go-faster/jx v1.2.0/go.mod h1:UWLOVDmMG597a5tBFPLIWJdUxz5/2emOpfsj9Neg0PE= +github.com/go-faster/xor v0.3.0/go.mod h1:x5CaDY9UKErKzqfRfFZdfu+OSTfoZny3w5Ak7UxcipQ= +github.com/go-faster/xor v1.0.0 h1:2o8vTOgErSGHP3/7XwA5ib1FTtUsNtwCoLLBjl31X38= +github.com/go-faster/xor v1.0.0/go.mod h1:x5CaDY9UKErKzqfRfFZdfu+OSTfoZny3w5Ak7UxcipQ= +github.com/go-faster/yaml v0.4.6 h1:lOK/EhI04gCpPgPhgt0bChS6bvw7G3WwI8xxVe0sw9I= +github.com/go-faster/yaml v0.4.6/go.mod h1:390dRIvV4zbnO7qC9FGo6YYutc+wyyUSHBgbXL52eXk= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gotd/ige v0.2.2 h1:XQ9dJZwBfDnOGSTxKXBGP4gMud3Qku2ekScRjDWWfEk= +github.com/gotd/ige v0.2.2/go.mod h1:tuCRb+Y5Y3eNTo3ypIfNpQ4MFjrnONiL2jN2AKZXmb0= +github.com/gotd/neo v0.1.5 h1:oj0iQfMbGClP8xI59x7fE/uHoTJD7NZH9oV1WNuPukQ= +github.com/gotd/neo v0.1.5/go.mod h1:9A2a4bn9zL6FADufBdt7tZt+WMhvZoc5gWXihOPoiBQ= +github.com/gotd/td v0.136.0 h1:f7vx/1rlvP59L5EKR820XpMRO2k267wW8/F0rAWbepc= +github.com/gotd/td v0.136.0/go.mod h1:mStcqs/9FXhNhWnPTguptSwqkQbRIwXLw3SCSpzPJxM= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.7.6 h1:rWQc5FwZSPX58r1OQmkuaNicxdmExaEz5A2DO2hUuTk= +github.com/jackc/pgx/v5 v5.7.6/go.mod h1:aruU7o91Tc2q2cFp5h4uP3f6ztExVpyVv88Xl/8Vl8M= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= +github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= +github.com/kelseyhightower/envconfig v1.4.0 h1:Im6hONhd3pLkfDFsbRgu68RDNkGF1r3dvMUtDTo2cv8= +github.com/kelseyhightower/envconfig v1.4.0/go.mod h1:cccZRl6mQpaq41TPp5QxidR+Sa3axMbJDNb//FQX6Gg= +github.com/klauspost/compress v1.18.2 h1:iiPHWW0YrcFgpBYhsA6D1+fqHssJscY/Tm/y2Uqnapk= +github.com/klauspost/compress v1.18.2/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= +github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= +github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= +github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= +github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= +github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= +github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/ogen-go/ogen v1.16.0 h1:fKHEYokW/QrMzVNXId74/6RObRIUs9T2oroGKtR25Iw= +github.com/ogen-go/ogen v1.16.0/go.mod h1:s3nWiMzybSf8fhxckyO+wtto92+QHpEL8FmkPnhL3jI= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0= +github.com/rs/zerolog v1.34.0 h1:k43nTLIwcTVQAncfCw4KZ2VY6ukYoZaBPNOE8txlOeY= +github.com/rs/zerolog v1.34.0/go.mod h1:bJsvje4Z08ROH4Nhs5iH600c3IkWhwp44iRc54W6wYQ= +github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0= +github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs= +github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= +github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= +go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= +go.opentelemetry.io/otel v1.38.0 h1:RkfdswUDRimDg0m2Az18RKOsnI8UDzppJAtj01/Ymk8= +go.opentelemetry.io/otel v1.38.0/go.mod h1:zcmtmQ1+YmQM9wrNsTGV/q/uyusom3P8RxwExxkZhjM= +go.opentelemetry.io/otel/metric v1.38.0 h1:Kl6lzIYGAh5M159u9NgiRkmoMKjvbsKtYRwgfrA6WpA= +go.opentelemetry.io/otel/metric v1.38.0/go.mod h1:kB5n/QoRM8YwmUahxvI3bO34eVtQf2i4utNVLr9gEmI= +go.opentelemetry.io/otel/trace v1.38.0 h1:Fxk5bKrDZJUH+AMyyIXGcFAPah0oRcT+LuNtJrmcNLE= +go.opentelemetry.io/otel/trace v1.38.0/go.mod h1:j1P9ivuFsTceSWe1oY+EeW3sc+Pp42sO++GHkg4wwhs= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= +go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc= +go.uber.org/zap v1.27.1/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= +golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q= +golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4= +golang.org/x/exp v0.0.0-20230725093048-515e97ebf090 h1:Di6/M8l0O2lCLc6VVRWhgCiApHV8MnQurBnFSHsQtNY= +golang.org/x/exp v0.0.0-20230725093048-515e97ebf090/go.mod h1:FXUEEKJgO7OQYeo8N01OfiKP8RXMtf6e8aTskBGqWdc= +golang.org/x/mod v0.30.0 h1:fDEXFVZ/fmCKProc/yAXXUijritrDzahmwwefnjoPFk= +golang.org/x/mod v0.30.0/go.mod h1:lAsf5O2EvJeSFMiBxXDki7sCgAxEUcZHXoXMKT4GJKc= +golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= +golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU= +golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= +golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= +golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= +golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= +golang.org/x/tools v0.39.0 h1:ik4ho21kwuQln40uelmciQPp9SipgNDdrafrYA4TmQQ= +golang.org/x/tools v0.39.0/go.mod h1:JnefbkDPyD8UU2kI5fuf8ZX4/yUeh9W877ZeBONxUqQ= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= +gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +nhooyr.io/websocket v1.8.17 h1:KEVeLJkUywCKVsnLIDlD/5gtayKp8VoCkksHCGGfT9Y= +nhooyr.io/websocket v1.8.17/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c= +rsc.io/qr v0.2.0 h1:6vBLea5/NRMVTz8V66gipeLycZMl/+UlFmk8DvqQ6WY= +rsc.io/qr v0.2.0/go.mod h1:IF+uZjkb9fqyeF/4tlBoynqmQxUoPfWEKh921coOuXs= diff --git a/tg_parser/internal/adapter/database/create_post.go b/tg_parser/internal/adapter/database/create_post.go new file mode 100644 index 0000000..a5ac93b --- /dev/null +++ b/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 +} diff --git a/tg_parser/internal/adapter/database/create_views_snapshot.go b/tg_parser/internal/adapter/database/create_views_snapshot.go new file mode 100644 index 0000000..da7d357 --- /dev/null +++ b/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 +} diff --git a/tg_parser/internal/adapter/database/database.go b/tg_parser/internal/adapter/database/database.go new file mode 100644 index 0000000..65305af --- /dev/null +++ b/tg_parser/internal/adapter/database/database.go @@ -0,0 +1,7 @@ +package database + +type Database struct{} + +func New() *Database { + return &Database{} +} diff --git a/tg_parser/internal/adapter/database/delete_post.go b/tg_parser/internal/adapter/database/delete_post.go new file mode 100644 index 0000000..4450e04 --- /dev/null +++ b/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 +} diff --git a/tg_parser/internal/adapter/database/get_channels.go b/tg_parser/internal/adapter/database/get_channels.go new file mode 100644 index 0000000..17fc0d4 --- /dev/null +++ b/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), + } +} 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 new file mode 100644 index 0000000..fcbb055 --- /dev/null +++ b/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), + } +} diff --git a/tg_parser/internal/adapter/database/get_tracked_posts.go b/tg_parser/internal/adapter/database/get_tracked_posts.go new file mode 100644 index 0000000..815f528 --- /dev/null +++ b/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), + } +} diff --git a/tg_parser/internal/adapter/database/update_channel.go b/tg_parser/internal/adapter/database/update_channel.go new file mode 100644 index 0000000..1cdb700 --- /dev/null +++ b/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 +} diff --git a/tg_parser/internal/adapter/telegram/get_active_views.go b/tg_parser/internal/adapter/telegram/get_active_views.go new file mode 100644 index 0000000..278b048 --- /dev/null +++ b/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 +} diff --git a/tg_parser/internal/adapter/telegram/get_channel_diff.go b/tg_parser/internal/adapter/telegram/get_channel_diff.go new file mode 100644 index 0000000..394e74f --- /dev/null +++ b/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 +} diff --git a/tg_parser/internal/adapter/telegram/history.go b/tg_parser/internal/adapter/telegram/history.go new file mode 100644 index 0000000..7ec1e62 --- /dev/null +++ b/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 +} diff --git a/tg_parser/internal/adapter/telegram/resolve.go b/tg_parser/internal/adapter/telegram/resolve.go new file mode 100644 index 0000000..ae3dbee --- /dev/null +++ b/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 +} diff --git a/tg_parser/internal/adapter/telegram/telegram.go b/tg_parser/internal/adapter/telegram/telegram.go new file mode 100644 index 0000000..13c5540 --- /dev/null +++ b/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() +} diff --git a/tg_parser/internal/app/app.go b/tg_parser/internal/app/app.go new file mode 100644 index 0000000..60a1a01 --- /dev/null +++ b/tg_parser/internal/app/app.go @@ -0,0 +1,91 @@ +package app + +import ( + "context" + "errors" + "fmt" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/rs/zerolog/log" + + "github.com/TelegramExchange/tgex-parser/config" + "github.com/TelegramExchange/tgex-parser/internal/adapter/database" + tgadapter "github.com/TelegramExchange/tgex-parser/internal/adapter/telegram" + "github.com/TelegramExchange/tgex-parser/internal/controller/httpserver" + "github.com/TelegramExchange/tgex-parser/internal/controller/worker" + "github.com/TelegramExchange/tgex-parser/internal/usecase" + "github.com/TelegramExchange/tgex-parser/pkg/postgres" + "github.com/TelegramExchange/tgex-parser/pkg/telegram" + "github.com/TelegramExchange/tgex-parser/pkg/transaction" +) + +func Run(ctx context.Context, c config.Config) error { + // Telegram client + tgClient, err := telegram.New(c.Telegram) + if err != nil { + return fmt.Errorf("telegram.New: %w", err) + } + + // PostgreSQL + pgPool, err := postgres.New(ctx, c.Postgres) + if err != nil { + return fmt.Errorf("broker.New: %w", err) + } + defer pgPool.Close() + + transaction.Init(pgPool) + + // Adapters + tg := tgadapter.New(tgClient) + db := database.New() + + // UseCase + uc := usecase.New(tg, db) + + // Controllers + channelWorker := worker.NewChannelWorker(uc, c.ChannelWorker) + viewsWorker := worker.NewViewsWorker(uc, c.ViewsWorker) + + httpServer := httpserver.New(uc, c.HTTP.Addr) + serverErr := make(chan error, 1) + go func() { + err := httpServer.ListenAndServe() + if err != nil && !errors.Is(err, http.ErrServerClosed) { + serverErr <- err + } + }() + + log.Info().Str("addr", c.HTTP.Addr).Msg("HTTP server started") + log.Info().Msg("App started") + + sig := make(chan os.Signal, 1) + signal.Notify(sig, os.Interrupt, syscall.SIGTERM) + select { + case <-sig: + case err := <-serverErr: + return fmt.Errorf("http server: %w", err) + } + + log.Info().Msg("App got signal to stop") + + // Controllers + viewsWorker.Stop() + channelWorker.Stop() + + ctxShutdown, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + if err := httpServer.Shutdown(ctxShutdown); err != nil { + log.Error().Err(err).Msg("HTTP server shutdown failed") + } + + // Adapters + tgClient.Close() + + log.Info().Msg("App stopped") + + return nil +} diff --git a/tg_parser/internal/controller/httpserver/server.go b/tg_parser/internal/controller/httpserver/server.go new file mode 100644 index 0000000..62f9c0b --- /dev/null +++ b/tg_parser/internal/controller/httpserver/server.go @@ -0,0 +1,90 @@ +package httpserver + +import ( + "context" + "encoding/json" + "net/http" + "strings" + + "github.com/TelegramExchange/tgex-parser/internal/usecase" + "github.com/gotd/td/tgerr" + "github.com/rs/zerolog/log" +) + +type Server struct { + httpServer *http.Server +} + +func New(uc *usecase.UseCase, addr string) *Server { + mux := http.NewServeMux() + mux.HandleFunc("/fetch-telegram-channel", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + username := strings.TrimSpace(r.URL.Query().Get("username")) + username = strings.TrimPrefix(username, "@") + if username == "" { + http.Error(w, "missing username", http.StatusBadRequest) + return + } + + channel, err := uc.FetchChannelMeta(r.Context(), username) + if err != nil { + if isNotFoundError(err) { + http.Error(w, "channel not found", http.StatusNotFound) + return + } + log.Error().Err(err).Str("username", username).Msg("fetch channel meta failed") + http.Error(w, "internal error", http.StatusInternalServerError) + return + } + + resp := struct { + ID string `json:"id"` + TelegramID int64 `json:"telegram_id"` + Username string `json:"username"` + Title string `json:"title"` + AccessHash int64 `json:"access_hash"` + Pts int `json:"pts"` + }{ + ID: channel.ID.String(), + TelegramID: channel.TelegramID, + Username: channel.Username, + Title: channel.Title, + AccessHash: channel.AccessHash, + Pts: channel.Pts, + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(resp); err != nil { + log.Error().Err(err).Msg("encode channel response") + } + }) + + return &Server{ + httpServer: &http.Server{ + Addr: addr, + Handler: mux, + }, + } +} + +func (s *Server) ListenAndServe() error { + return s.httpServer.ListenAndServe() +} + +func (s *Server) Shutdown(ctx context.Context) error { + return s.httpServer.Shutdown(ctx) +} + +func isNotFoundError(err error) bool { + if tgerr.Is(err, "USERNAME_NOT_OCCUPIED", "USERNAME_INVALID", "CHANNEL_INVALID", "CHANNEL_PRIVATE") { + return true + } + + msg := err.Error() + return strings.Contains(msg, "no chats in resolve result") || + strings.Contains(msg, "not a channel") +} diff --git a/tg_parser/internal/controller/worker/channel_worker.go b/tg_parser/internal/controller/worker/channel_worker.go new file mode 100644 index 0000000..0103566 --- /dev/null +++ b/tg_parser/internal/controller/worker/channel_worker.go @@ -0,0 +1,85 @@ +package worker + +import ( + "context" + "sync" + "time" + + "github.com/rs/zerolog/log" + + "github.com/TelegramExchange/tgex-parser/internal/usecase" +) + +type ChannelConfig struct { + MaxWorkers int `envconfig:"WORKER__MAX_WORKERS" default:"1"` + ChannelDelay time.Duration `envconfig:"WORKER__CHANNEL_DELAY" default:"500ms"` + RequestTimeout time.Duration `envconfig:"WORKER__REQUEST_TIMEOUT" default:"10s"` + MessagesLimit int `envconfig:"WORKER__MESSAGES_LIMIT" default:"20"` + PollInterval time.Duration `envconfig:"WORKER__POLL_INTERVAL" default:"20s"` +} + +type ChannelWorker struct { + usecase *usecase.UseCase + config ChannelConfig + stop chan struct{} + done chan struct{} +} + +func NewChannelWorker(uc *usecase.UseCase, cfg ChannelConfig) *ChannelWorker { + w := &ChannelWorker{ + usecase: uc, + config: cfg, + stop: make(chan struct{}), + done: make(chan struct{}), + } + + go w.run() + + return w +} + +func (w *ChannelWorker) run() { + log.Info().Msg("channel worker: started") + + var wg sync.WaitGroup + + wg.Add(w.config.MaxWorkers) + + for range w.config.MaxWorkers { + go w.spawnWorker(&wg) + } + + wg.Wait() + + log.Info().Msg("channel worker: stopped") + + close(w.done) +} + +func (w *ChannelWorker) spawnWorker(wg *sync.WaitGroup) { + defer wg.Done() + + limiter := time.NewTicker(w.config.ChannelDelay) + defer limiter.Stop() + + poll := time.NewTicker(w.config.PollInterval) + defer poll.Stop() + + for { + select { + case <-w.stop: + return + + case <-poll.C: + err := w.usecase.FetchChannels(context.Background()) + if err != nil { + log.Error().Err(err).Msg("usecase.FetchChannels error") + } + } + } +} + +func (w *ChannelWorker) Stop() { + close(w.stop) + <-w.done +} diff --git a/tg_parser/internal/controller/worker/views_worker.go b/tg_parser/internal/controller/worker/views_worker.go new file mode 100644 index 0000000..ae76224 --- /dev/null +++ b/tg_parser/internal/controller/worker/views_worker.go @@ -0,0 +1,65 @@ +package worker + +import ( + "context" + "time" + + "github.com/rs/zerolog/log" + + "github.com/TelegramExchange/tgex-parser/internal/usecase" +) + +type ViewsConfig struct { + Interval time.Duration `envconfig:"WORKER__VIEWS_INTERVAL" default:"30s"` + RequestTimeout time.Duration `envconfig:"WORKER__REQUEST_TIMEOUT" default:"10s"` +} + +type ViewsWorker struct { + usecase *usecase.UseCase + config ViewsConfig + stop chan struct{} + done chan struct{} +} + +func NewViewsWorker(uc *usecase.UseCase, cfg ViewsConfig) *ViewsWorker { + w := &ViewsWorker{ + usecase: uc, + config: cfg, + stop: make(chan struct{}), + done: make(chan struct{}), + } + + go w.run() + + return w +} + +func (w *ViewsWorker) run() { + log.Info().Msg("views worker: started") + + for { + select { + case <-w.stop: + log.Info().Msg("views worker: stopped") + close(w.done) + return + case <-time.After(w.config.Interval): + log.Debug().Dur("interval", w.config.Interval).Msg("views worker: refresh tick") + ctx, cancel := context.WithTimeout(context.Background(), w.config.RequestTimeout) + + err := w.usecase.FetchViews(ctx) + if err != nil { + log.Error().Err(err).Msg("views worker: FetchViews error") + } else { + log.Debug().Msg("views worker: refresh finished") + } + + cancel() + } + } +} + +func (w *ViewsWorker) Stop() { + close(w.stop) + <-w.done +} diff --git a/tg_parser/internal/domain/channel.go b/tg_parser/internal/domain/channel.go new file mode 100644 index 0000000..3d4abcb --- /dev/null +++ b/tg_parser/internal/domain/channel.go @@ -0,0 +1,65 @@ +package domain + +import ( + "fmt" + + "github.com/google/uuid" +) + +type Channel struct { + ID uuid.UUID + TelegramID int64 + Username string + Title string + AccessHash int64 + Pts int +} + +const channelIDOffset int64 = 1000000000000 + +// ChannelID returns the positive channel identifier expected by Telegram API. +func (c Channel) ChannelID() int64 { + return ChannelIDFromChatID(c.TelegramID) +} + +// ChannelIDFromChatID converts stored chat IDs (Bot API style) to channel IDs used by TDLib. +func ChannelIDFromChatID(chatID int64) int64 { + if chatID >= 0 { + return chatID + } + + return -chatID - channelIDOffset +} + +// ChatIDFromChannelID converts Telegram channel IDs to Bot API style chat IDs (-100...). +func ChatIDFromChannelID(channelID int64) int64 { + return -channelIDOffset - channelID +} + +// NormalizeChatID ensures that TelegramID is stored in chat-id form (negative). +func NormalizeChatID(id int64) int64 { + if id < 0 { + return id + } + + return ChatIDFromChannelID(id) +} + +func (c Channel) String() string { + return fmt.Sprintf( + "Channel{id=%s telegram_id=%d username=%q title=%q access_hash=%d pts=%d}", + c.ID, + c.TelegramID, + c.Username, + c.Title, + c.AccessHash, + c.Pts, + ) +} + +type ChannelDiff struct { + NewPts int + NewPosts []Post + DeletedPosts []Post + UpdatedChannel *Channel +} diff --git a/tg_parser/internal/domain/post.go b/tg_parser/internal/domain/post.go new file mode 100644 index 0000000..64b8632 --- /dev/null +++ b/tg_parser/internal/domain/post.go @@ -0,0 +1,27 @@ +package domain + +import ( + "fmt" + + "github.com/google/uuid" +) + +type Post struct { + ID uuid.UUID + ChannelID uuid.UUID + MessageID int + Text string + Link string + Views int +} + +func NewPost(channel Channel, messageID int, text string, views int) Post { + return Post{ + ID: uuid.New(), + ChannelID: channel.ID, + MessageID: messageID, + Text: text, + Link: fmt.Sprintf("https://t.me/%s/%d", channel.Username, messageID), + Views: views, + } +} diff --git a/tg_parser/internal/domain/views_snapshot.go b/tg_parser/internal/domain/views_snapshot.go new file mode 100644 index 0000000..a265e9a --- /dev/null +++ b/tg_parser/internal/domain/views_snapshot.go @@ -0,0 +1,14 @@ +package domain + +import ( + "time" + + "github.com/google/uuid" +) + +type ViewsSnapshot struct { + ViewsCount int + FetchedAt time.Time + + PostID uuid.UUID +} diff --git a/tg_parser/internal/usecase/fetch_channel_meta.go b/tg_parser/internal/usecase/fetch_channel_meta.go new file mode 100644 index 0000000..3dde6fd --- /dev/null +++ b/tg_parser/internal/usecase/fetch_channel_meta.go @@ -0,0 +1,11 @@ +package usecase + +import ( + "context" + + "github.com/TelegramExchange/tgex-parser/internal/domain" +) + +func (uc *UseCase) FetchChannelMeta(ctx context.Context, username string) (domain.Channel, error) { + return uc.telegram.ParseChannelMeta(ctx, username) +} diff --git a/tg_parser/internal/usecase/fetch_channels.go b/tg_parser/internal/usecase/fetch_channels.go new file mode 100644 index 0000000..9d9c267 --- /dev/null +++ b/tg_parser/internal/usecase/fetch_channels.go @@ -0,0 +1,112 @@ +package usecase + +import ( + "context" + "fmt" + + "github.com/TelegramExchange/tgex-parser/pkg/transaction" + "github.com/rs/zerolog/log" + + "github.com/TelegramExchange/tgex-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 + } + + 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 { + 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 (uc *UseCase) initializeChannel(ctx context.Context, channel domain.Channel) (domain.Channel, error) { + c, err := uc.telegram.ParseChannelMeta(ctx, channel.Username) + if err != nil { + return domain.Channel{}, fmt.Errorf("telegram.ParseChannelMeta: %w", err) + } + + // 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 +} diff --git a/tg_parser/internal/usecase/fetch_views.go b/tg_parser/internal/usecase/fetch_views.go new file mode 100644 index 0000000..3015613 --- /dev/null +++ b/tg_parser/internal/usecase/fetch_views.go @@ -0,0 +1,46 @@ +package usecase + +import ( + "context" + "time" + + "github.com/rs/zerolog/log" + + "github.com/TelegramExchange/tgex-parser/internal/domain" +) + +func (uc *UseCase) FetchViews(ctx context.Context) error { + channels, err := uc.database.GetChannelsWithTrackedPosts(ctx) + if err != nil { + return err + } + + for _, channel := range channels { + posts, err := uc.database.GetTrackedPosts(ctx, channel) + if err != nil { + return err + } + + err = uc.telegram.UpdatePostsViews(ctx, channel, posts) + if err != nil { + return err + } + + for _, p := range posts { + v := domain.ViewsSnapshot{ + ViewsCount: p.Views, + FetchedAt: time.Now().UTC(), + PostID: p.ID, + } + + err = uc.database.CreateViewsSnapshot(ctx, v) + if err != nil { + return err + } + + log.Info().Msgf("New ViewsSnapshot: %s - Post ID: %d, Views: %d", p.Link, p.MessageID, p.Views) + } + } + + return nil +} diff --git a/tg_parser/internal/usecase/usecase.go b/tg_parser/internal/usecase/usecase.go new file mode 100644 index 0000000..2acaee1 --- /dev/null +++ b/tg_parser/internal/usecase/usecase.go @@ -0,0 +1,38 @@ +package usecase + +import ( + "context" + + "github.com/TelegramExchange/tgex-parser/internal/domain" +) + +type Telegram interface { + ParseChannelMeta(ctx context.Context, username string) (domain.Channel, error) + GetChannelHistory(ctx context.Context, channel domain.Channel, limit int) ([]domain.Post, error) + GetChannelDiff(ctx context.Context, channel domain.Channel, limit int) (domain.ChannelDiff, error) + UpdatePostsViews(ctx context.Context, channel domain.Channel, posts []domain.Post) error +} + +type Database interface { + CreatePost(ctx context.Context, post domain.Post) error + + CreateViewsSnapshot(ctx context.Context, snapshot domain.ViewsSnapshot) error + + GetChannels(ctx context.Context) []domain.Channel + UpdateChannel(ctx context.Context, channel domain.Channel) error + GetChannelsWithTrackedPosts(ctx context.Context) ([]domain.Channel, error) + GetTrackedPosts(ctx context.Context, channel domain.Channel) ([]domain.Post, error) + DeletePost(ctx context.Context, p domain.Post) error +} + +type UseCase struct { + telegram Telegram + database Database +} + +func New(telegram Telegram, database Database) *UseCase { + return &UseCase{ + telegram: telegram, + database: database, + } +} diff --git a/tg_parser/main.go b/tg_parser/main.go new file mode 100644 index 0000000..232b1f4 --- /dev/null +++ b/tg_parser/main.go @@ -0,0 +1,29 @@ +package main + +import ( + "context" + + "github.com/rs/zerolog/log" + + "github.com/TelegramExchange/tgex-parser/config" + "github.com/TelegramExchange/tgex-parser/internal/app" + "github.com/TelegramExchange/tgex-parser/pkg/logger" +) + +func main() { + log.Info().Msg("parser starting") + + c, err := config.New() + if err != nil { + log.Fatal().Err(err).Msg("config.New") + } + + logger.Init(c.Logger) + log.Info().Msg("parser initialized") + + ctx := context.Background() + + if err := app.Run(ctx, c); err != nil { + log.Error().Err(err).Msg("app.Run") + } +} diff --git a/tg_parser/pkg/logger/logger.go b/tg_parser/pkg/logger/logger.go new file mode 100644 index 0000000..8793b49 --- /dev/null +++ b/tg_parser/pkg/logger/logger.go @@ -0,0 +1,36 @@ +package logger + +import ( + "os" + "time" + + "github.com/rs/zerolog" + "github.com/rs/zerolog/log" +) + +type Config struct { + Level string `default:"info" envconfig:"LOGGER__LEVEL"` + PrettyConsole bool `default:"false" envconfig:"LOGGER__PRETTY_CONSOLE"` +} + +func Init(c Config) { + zerolog.TimeFieldFormat = time.RFC3339 + + level := zerolog.InfoLevel + if parsedLevel, err := zerolog.ParseLevel(c.Level); err == nil { + level = parsedLevel + } + zerolog.SetGlobalLevel(level) + + log.Logger = log.With(). + // Caller(). + // Str("app_name", c.AppName). + // Str("app_version", c.AppVersion). + Logger() + + if c.PrettyConsole { + log.Logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr, TimeFormat: "15:04:05"}) + } + + log.Info().Msg("Logger initialized") +} diff --git a/tg_parser/pkg/postgres/postgres.go b/tg_parser/pkg/postgres/postgres.go new file mode 100644 index 0000000..848eb9e --- /dev/null +++ b/tg_parser/pkg/postgres/postgres.go @@ -0,0 +1,37 @@ +package postgres + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/rs/zerolog/log" +) + +type Config struct { + URL string `envconfig:"DB__URL" required:"true"` +} + +type Pool struct { + *pgxpool.Pool +} + +func New(ctx context.Context, c Config) (*Pool, error) { + cfg, err := pgxpool.ParseConfig(c.URL) + if err != nil { + return nil, fmt.Errorf("pgxpool.ParseConfig: %w", err) + } + + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + return nil, fmt.Errorf("pgxpool.NewWithConfig: %w", err) + } + + return &Pool{Pool: pool}, nil +} + +func (p *Pool) Close() { + p.Pool.Close() + + log.Info().Msg("Postgres closed") +} diff --git a/tg_parser/pkg/telegram/telegram.go b/tg_parser/pkg/telegram/telegram.go new file mode 100644 index 0000000..bd56e42 --- /dev/null +++ b/tg_parser/pkg/telegram/telegram.go @@ -0,0 +1,76 @@ +package telegram + +import ( + "context" + "fmt" + + "github.com/rs/zerolog/log" + + "github.com/gotd/td/session" + "github.com/gotd/td/telegram" +) + +type Config struct { + ApiID int `envconfig:"TELEGRAM__API_ID" required:"true"` + ApiHash string `envconfig:"TELEGRAM__API_HASH" required:"true"` + SessionFile string `envconfig:"TELEGRAM__SESSION_FILE" required:"true"` +} + +type Client struct { + *telegram.Client + cancel context.CancelFunc + done chan struct{} +} + +func New(cfg Config) (*Client, error) { + ctx := context.Background() + + tg := telegram.NewClient(cfg.ApiID, cfg.ApiHash, telegram.Options{ + SessionStorage: &session.FileStorage{Path: cfg.SessionFile}, + }) + + runCtx, cancel := context.WithCancel(ctx) + + ready := make(chan struct{}) + done := make(chan struct{}) + + client := &Client{ + Client: tg, + cancel: cancel, + done: done, + } + + go func() { + err := tg.Run(runCtx, func(ctx context.Context) error { + status, err := tg.Auth().Status(ctx) + if err != nil { + close(ready) + return fmt.Errorf("telegram.Auth error: %w", err) + } + if !status.Authorized { + close(ready) + return fmt.Errorf("session not authorized") + } + + close(ready) + + <-ctx.Done() + return ctx.Err() + }) + + if err != nil { + log.Error().Err(err).Msg("telegram client stopped") + } + + close(done) + }() + + <-ready + + return client, nil +} + +func (c *Client) Close() { + c.cancel() + <-c.done +} diff --git a/tg_parser/pkg/transaction/transaction.go b/tg_parser/pkg/transaction/transaction.go new file mode 100644 index 0000000..50b2a4a --- /dev/null +++ b/tg_parser/pkg/transaction/transaction.go @@ -0,0 +1,101 @@ +package transaction + +import ( + "context" + "errors" + "fmt" + + "github.com/TelegramExchange/tgex-parser/pkg/postgres" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/rs/zerolog/log" +) + +var ( + errMissingInit = errors.New("missing `transaction.Init' call before `transaction.Begin'") + errMissingBegin = errors.New("missing `transaction.Begin' call before 'transaction.Get'") +) + +var ( + pool *pgxpool.Pool + IsUnitTest bool +) + +type ctxKey struct{} + +func Init(p *postgres.Pool) { + pool = p.Pool +} + +type Transaction struct { + pgx.Tx +} + +func Begin(ctx context.Context) (context.Context, error) { + if IsUnitTest { + return ctx, nil + } + + if pool == nil { + return nil, errMissingInit + } + + tx, err := pool.Begin(ctx) + if err != nil { + return nil, fmt.Errorf("pool.Begin: %w", err) + } + + ctx = context.WithValue(ctx, ctxKey{}, &Transaction{tx}) + + return ctx, nil +} + +func Rollback(ctx context.Context) { + if IsUnitTest { + return + } + + tx, ok := ctx.Value(ctxKey{}).(*Transaction) + if !ok { + return + } + + err := tx.Rollback(ctx) + if err != nil && !errors.Is(err, pgx.ErrTxClosed) { + log.Error().Err(err).Msg("transaction: Rollback") + } +} + +func Commit(ctx context.Context) error { + if IsUnitTest { + return nil + } + + tx, ok := ctx.Value(ctxKey{}).(*Transaction) + if !ok { + return errMissingBegin + } + + err := tx.Commit(ctx) + if err != nil { + return fmt.Errorf("tx.Commit: %w", err) + } + + return nil +} + +type Executor interface { + Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) + Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +} + +func TryExtractTX(ctx context.Context) Executor { + tx, ok := ctx.Value(ctxKey{}).(*Transaction) + if !ok { + return pool + } + + return tx +} diff --git a/tg_parser/pkg/transaction/wrap.go b/tg_parser/pkg/transaction/wrap.go new file mode 100644 index 0000000..97245e3 --- /dev/null +++ b/tg_parser/pkg/transaction/wrap.go @@ -0,0 +1,38 @@ +package transaction + +import ( + "context" + "errors" + "fmt" + + "github.com/jackc/pgx/v5" + "github.com/rs/zerolog/log" +) + +func Wrap(ctx context.Context, fn func(context.Context) error) error { + tx, err := pool.Begin(ctx) + if err != nil { + return fmt.Errorf("pool.Begin: %w", err) + } + + defer func() { + err = tx.Rollback(ctx) + if err != nil && !errors.Is(err, pgx.ErrTxClosed) { + log.Error().Err(err).Msg("transaction: Rollback") + } + }() + + ctx = context.WithValue(ctx, ctxKey{}, &Transaction{tx}) + + err = fn(ctx) + if err != nil { + return fmt.Errorf("fn: %w", err) + } + + err = tx.Commit(ctx) + if err != nil { + return fmt.Errorf("tx.Commit: %w", err) + } + + return nil +}