feat: initial commit

internal/infrastructure/n8n/payload.go

@@ -0,0 +1,8 @@
+package n8n
+
+// payload.go contains shared payload types for n8n webhook communication.
+// The primary types (n8nPayload, n8nResponse) are defined in webhook_dispatcher.go
+// to keep them co-located with their usage.
+//
+// This file is reserved for any additional payload structures needed
+// for extended n8n integration (e.g., batch requests, webhook registration).

internal/infrastructure/n8n/webhook_dispatcher.go

@@ -0,0 +1,124 @@
+package n8n
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"time"
+
+	"github.com/paramah/gw_telegram/internal/domain/entity"
+)
+
+type WorkflowConfig struct {
+	ID         string
+	WebhookURL string
+	AuthToken  string
+	Timeout    time.Duration
+}
+
+type WebhookDispatcher struct {
+	workflows map[string]WorkflowConfig
+	client    *http.Client
+	logger    *slog.Logger
+}
+
+func NewWebhookDispatcher(workflows map[string]WorkflowConfig, logger *slog.Logger) *WebhookDispatcher {
+	return &WebhookDispatcher{
+		workflows: workflows,
+		client:    &http.Client{Timeout: 30 * time.Second},
+		logger:    logger,
+	}
+}
+
+type n8nPayload struct {
+	RequestID   string         `json:"request_id"`
+	ChatID      int64          `json:"chat_id"`
+	UserID      int64          `json:"user_id"`
+	Username    string         `json:"username"`
+	MessageText string         `json:"message_text"`
+	IntentName  string         `json:"intent_name"`
+	Timestamp   time.Time      `json:"timestamp"`
+	Metadata    map[string]any `json:"metadata,omitempty"`
+}
+
+type n8nResponse struct {
+	Reply        string          `json:"reply"`
+	Actions      []entity.Action `json:"actions,omitempty"`
+	NextWorkflow string          `json:"next_workflow,omitempty"`
+}
+
+func (d *WebhookDispatcher) Dispatch(ctx context.Context, req entity.WorkflowRequest) (entity.WorkflowResponse, error) {
+	wf, ok := d.workflows[req.Intent.Name]
+	if !ok {
+		// Try default workflow
+		wf, ok = d.workflows["default"]
+		if !ok {
+			return entity.WorkflowResponse{}, fmt.Errorf("no workflow configured for intent %q", req.Intent.Name)
+		}
+	}
+
+	payload := n8nPayload{
+		RequestID:   req.RequestID,
+		ChatID:      req.ChatID,
+		UserID:      req.UserID,
+		Username:    req.Username,
+		MessageText: req.MessageText,
+		IntentName:  req.Intent.Name,
+		Timestamp:   req.Timestamp,
+		Metadata:    req.Metadata,
+	}
+
+	body, err := json.Marshal(payload)
+	if err != nil {
+		return entity.WorkflowResponse{}, fmt.Errorf("marshal payload: %w", err)
+	}
+
+	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, wf.WebhookURL, bytes.NewReader(body))
+	if err != nil {
+		return entity.WorkflowResponse{}, fmt.Errorf("create request: %w", err)
+	}
+	httpReq.Header.Set("Content-Type", "application/json")
+	if wf.AuthToken != "" {
+		httpReq.Header.Set("Authorization", "Bearer "+wf.AuthToken)
+	}
+
+	client := d.client
+	if wf.Timeout > 0 {
+		client = &http.Client{Timeout: wf.Timeout}
+	}
+
+	resp, err := client.Do(httpReq)
+	if err != nil {
+		return entity.WorkflowResponse{}, fmt.Errorf("n8n webhook call: %w", err)
+	}
+	defer resp.Body.Close()
+
+	respBody, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return entity.WorkflowResponse{}, fmt.Errorf("read response: %w", err)
+	}
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return entity.WorkflowResponse{}, fmt.Errorf("n8n returned status %d: %s", resp.StatusCode, string(respBody))
+	}
+
+	var n8nResp n8nResponse
+	if err := json.Unmarshal(respBody, &n8nResp); err != nil {
+		// If not valid JSON, treat the raw body as reply text
+		return entity.WorkflowResponse{
+			RequestID: req.RequestID,
+			ReplyText: string(respBody),
+		}, nil
+	}
+
+	return entity.WorkflowResponse{
+		RequestID:    req.RequestID,
+		ReplyText:    n8nResp.Reply,
+		Actions:      n8nResp.Actions,
+		NextWorkflow: n8nResp.NextWorkflow,
+	}, nil
+}

internal/infrastructure/router/rule_based_router.go

@@ -0,0 +1,44 @@
+package router
+
+import (
+	"context"
+	"regexp"
+	"sort"
+
+	"github.com/paramah/gw_telegram/internal/domain/apperror"
+	"github.com/paramah/gw_telegram/internal/domain/entity"
+)
+
+type Rule struct {
+	Pattern    *regexp.Regexp
+	IntentName string
+	Target     entity.RouteTarget
+	Priority   int
+}
+
+type RuleBasedRouter struct {
+	rules []Rule
+}
+
+func NewRuleBasedRouter(rules []Rule) *RuleBasedRouter {
+	sorted := make([]Rule, len(rules))
+	copy(sorted, rules)
+	sort.Slice(sorted, func(i, j int) bool {
+		return sorted[i].Priority > sorted[j].Priority
+	})
+	return &RuleBasedRouter{rules: sorted}
+}
+
+func (r *RuleBasedRouter) Route(_ context.Context, msg entity.Message) (entity.Route, error) {
+	for _, rule := range r.rules {
+		if rule.Pattern.MatchString(msg.Text) {
+			return entity.Route{
+				Pattern:    rule.Pattern.String(),
+				IntentName: rule.IntentName,
+				Target:     rule.Target,
+				Priority:   rule.Priority,
+			}, nil
+		}
+	}
+	return entity.Route{}, apperror.ErrRouteNotFound
+}

internal/infrastructure/speech/audio_converter.go

@@ -0,0 +1,63 @@
+package speech
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+
+	"github.com/google/uuid"
+)
+
+type FFmpegConverter struct {
+	ffmpegPath string
+	tempDir    string
+}
+
+func NewFFmpegConverter(ffmpegPath, tempDir string) *FFmpegConverter {
+	if ffmpegPath == "" {
+		ffmpegPath = "ffmpeg"
+	}
+	if tempDir == "" {
+		tempDir = os.TempDir()
+	}
+	return &FFmpegConverter{ffmpegPath: ffmpegPath, tempDir: tempDir}
+}
+
+func (c *FFmpegConverter) Convert(ctx context.Context, input []byte, fromMime, toMime string) ([]byte, error) {
+	id := uuid.New().String()
+	inFile := filepath.Join(c.tempDir, id+".input")
+	outFile := filepath.Join(c.tempDir, id+".wav")
+
+	defer os.Remove(inFile)
+	defer os.Remove(outFile)
+
+	if err := os.WriteFile(inFile, input, 0600); err != nil {
+		return nil, fmt.Errorf("write temp input: %w", err)
+	}
+
+	cmd := exec.CommandContext(ctx, c.ffmpegPath,
+		"-i", inFile,
+		"-ar", "16000",
+		"-ac", "1",
+		"-f", "wav",
+		"-y",
+		outFile,
+	)
+
+	var stderr bytes.Buffer
+	cmd.Stderr = &stderr
+
+	if err := cmd.Run(); err != nil {
+		return nil, fmt.Errorf("ffmpeg conversion: %w: %s", err, stderr.String())
+	}
+
+	out, err := os.ReadFile(outFile)
+	if err != nil {
+		return nil, fmt.Errorf("read converted file: %w", err)
+	}
+
+	return out, nil
+}

internal/infrastructure/speech/openai_whisper.go

@@ -0,0 +1,89 @@
+package speech
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"net/http"
+	"time"
+)
+
+const whisperAPIURL = "https://api.openai.com/v1/audio/transcriptions"
+
+type OpenAIWhisper struct {
+	apiKey   string
+	model    string
+	language string
+	client   *http.Client
+}
+
+func NewOpenAIWhisper(apiKey, model, language string) *OpenAIWhisper {
+	return &OpenAIWhisper{
+		apiKey:   apiKey,
+		model:    model,
+		language: language,
+		client:   &http.Client{Timeout: 60 * time.Second},
+	}
+}
+
+type whisperResponse struct {
+	Text string `json:"text"`
+}
+
+func (w *OpenAIWhisper) Transcribe(ctx context.Context, audioData []byte, mimeType string) (string, error) {
+	var buf bytes.Buffer
+	mw := multipart.NewWriter(&buf)
+
+	fw, err := mw.CreateFormFile("file", "audio.wav")
+	if err != nil {
+		return "", fmt.Errorf("create form file: %w", err)
+	}
+	if _, err := fw.Write(audioData); err != nil {
+		return "", fmt.Errorf("write audio data: %w", err)
+	}
+
+	if err := mw.WriteField("model", w.model); err != nil {
+		return "", fmt.Errorf("write model field: %w", err)
+	}
+	if err := mw.WriteField("response_format", "json"); err != nil {
+		return "", fmt.Errorf("write response_format: %w", err)
+	}
+	if w.language != "" {
+		if err := mw.WriteField("language", w.language); err != nil {
+			return "", fmt.Errorf("write language field: %w", err)
+		}
+	}
+	mw.Close()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, whisperAPIURL, &buf)
+	if err != nil {
+		return "", fmt.Errorf("create whisper request: %w", err)
+	}
+	req.Header.Set("Authorization", "Bearer "+w.apiKey)
+	req.Header.Set("Content-Type", mw.FormDataContentType())
+
+	resp, err := w.client.Do(req)
+	if err != nil {
+		return "", fmt.Errorf("whisper API call: %w", err)
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", fmt.Errorf("read whisper response: %w", err)
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("whisper API error %d: %s", resp.StatusCode, string(body))
+	}
+
+	var result whisperResponse
+	if err := json.Unmarshal(body, &result); err != nil {
+		return "", fmt.Errorf("parse whisper response: %w", err)
+	}
+
+	return result.Text, nil
+}

internal/infrastructure/storage/redis_session_store.go

@@ -0,0 +1,62 @@
+package storage
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+	"github.com/paramah/gw_telegram/internal/domain/apperror"
+	"github.com/paramah/gw_telegram/internal/domain/entity"
+)
+
+type RedisSessionStore struct {
+	client *redis.Client
+	ttl    time.Duration
+}
+
+func NewRedisSessionStore(client *redis.Client, ttlHours int) *RedisSessionStore {
+	return &RedisSessionStore{
+		client: client,
+		ttl:    time.Duration(ttlHours) * time.Hour,
+	}
+}
+
+func sessionKey(userID int64) string {
+	return fmt.Sprintf("session:%d", userID)
+}
+
+func (s *RedisSessionStore) Get(ctx context.Context, userID int64) (entity.Session, error) {
+	data, err := s.client.Get(ctx, sessionKey(userID)).Bytes()
+	if err == redis.Nil {
+		return entity.Session{}, apperror.ErrSessionNotFound
+	}
+	if err != nil {
+		return entity.Session{}, fmt.Errorf("redis get session: %w", err)
+	}
+
+	var session entity.Session
+	if err := json.Unmarshal(data, &session); err != nil {
+		return entity.Session{}, fmt.Errorf("unmarshal session: %w", err)
+	}
+	return session, nil
+}
+
+func (s *RedisSessionStore) Set(ctx context.Context, session entity.Session) error {
+	data, err := json.Marshal(session)
+	if err != nil {
+		return fmt.Errorf("marshal session: %w", err)
+	}
+	if err := s.client.Set(ctx, sessionKey(session.UserID), data, s.ttl).Err(); err != nil {
+		return fmt.Errorf("redis set session: %w", err)
+	}
+	return nil
+}
+
+func (s *RedisSessionStore) Delete(ctx context.Context, userID int64) error {
+	if err := s.client.Del(ctx, sessionKey(userID)).Err(); err != nil {
+		return fmt.Errorf("redis del session: %w", err)
+	}
+	return nil
+}

internal/infrastructure/telegram/bot_gateway.go

@@ -0,0 +1,37 @@
+package telegram
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api/v5"
+)
+
+type BotGateway struct {
+	bot    *tgbotapi.BotAPI
+	logger *slog.Logger
+}
+
+func NewBotGateway(bot *tgbotapi.BotAPI, logger *slog.Logger) *BotGateway {
+	return &BotGateway{bot: bot, logger: logger}
+}
+
+func (g *BotGateway) SendText(ctx context.Context, chatID int64, text string) error {
+	msg := tgbotapi.NewMessage(chatID, text)
+	msg.ParseMode = tgbotapi.ModeMarkdown
+	_, err := g.bot.Send(msg)
+	if err != nil {
+		return fmt.Errorf("telegram send text: %w", err)
+	}
+	return nil
+}
+
+func (g *BotGateway) SendTyping(ctx context.Context, chatID int64) error {
+	action := tgbotapi.NewChatAction(chatID, tgbotapi.ChatTyping)
+	_, err := g.bot.Request(action)
+	if err != nil {
+		g.logger.WarnContext(ctx, "failed to send typing action", "error", err, "chat_id", chatID)
+	}
+	return nil
+}

internal/infrastructure/telegram/file_downloader.go

@@ -0,0 +1,48 @@
+package telegram
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+
+	tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api/v5"
+)
+
+type TelegramFileDownloader struct {
+	bot    *tgbotapi.BotAPI
+	client *http.Client
+}
+
+func NewTelegramFileDownloader(bot *tgbotapi.BotAPI) *TelegramFileDownloader {
+	return &TelegramFileDownloader{
+		bot:    bot,
+		client: &http.Client{},
+	}
+}
+
+func (d *TelegramFileDownloader) Download(ctx context.Context, fileID string) ([]byte, string, error) {
+	file, err := d.bot.GetFile(tgbotapi.FileConfig{FileID: fileID})
+	if err != nil {
+		return nil, "", fmt.Errorf("get file info: %w", err)
+	}
+
+	url := file.Link(d.bot.Token)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return nil, "", fmt.Errorf("create download request: %w", err)
+	}
+
+	resp, err := d.client.Do(req)
+	if err != nil {
+		return nil, "", fmt.Errorf("download file: %w", err)
+	}
+	defer resp.Body.Close()
+
+	data, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, "", fmt.Errorf("read file body: %w", err)
+	}
+
+	return data, "audio/ogg", nil
+}

internal/infrastructure/telegram/update_poller.go

@@ -0,0 +1,50 @@
+package telegram
+
+import (
+	"context"
+	"log/slog"
+
+	tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api/v5"
+)
+
+type UpdateHandler interface {
+	Handle(ctx context.Context, update tgbotapi.Update)
+}
+
+type UpdatePoller struct {
+	bot     *tgbotapi.BotAPI
+	handler UpdateHandler
+	logger  *slog.Logger
+	timeout int
+}
+
+func NewUpdatePoller(bot *tgbotapi.BotAPI, handler UpdateHandler, logger *slog.Logger) *UpdatePoller {
+	return &UpdatePoller{
+		bot:     bot,
+		handler: handler,
+		logger:  logger,
+		timeout: 60,
+	}
+}
+
+func (p *UpdatePoller) Start(ctx context.Context) error {
+	u := tgbotapi.NewUpdate(0)
+	u.Timeout = p.timeout
+
+	updates := p.bot.GetUpdatesChan(u)
+	p.logger.InfoContext(ctx, "update poller started")
+
+	for {
+		select {
+		case <-ctx.Done():
+			p.bot.StopReceivingUpdates()
+			p.logger.InfoContext(ctx, "update poller stopped")
+			return ctx.Err()
+		case update, ok := <-updates:
+			if !ok {
+				return nil
+			}
+			go p.handler.Handle(ctx, update)
+		}
+	}
+}