transcriber/main.go

package main

import (
	"context"
	"errors"
	"flag"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	ffmpegconv "git.vakhrushev.me/av/transcriber/internal/adapter/converter/ffmpeg"
	ffmpegmv "git.vakhrushev.me/av/transcriber/internal/adapter/metaviewer/ffmpeg"
	"git.vakhrushev.me/av/transcriber/internal/adapter/recognizer/yandex"
	pbrepo "git.vakhrushev.me/av/transcriber/internal/adapter/repo/pocketbase"
	"git.vakhrushev.me/av/transcriber/internal/adapter/telegram"
	"git.vakhrushev.me/av/transcriber/internal/config"
	httpcontroller "git.vakhrushev.me/av/transcriber/internal/controller/http"
	tgcontroller "git.vakhrushev.me/av/transcriber/internal/controller/tg"
	"git.vakhrushev.me/av/transcriber/internal/controller/worker"
	"git.vakhrushev.me/av/transcriber/internal/service"
	"github.com/joho/godotenv"
	"github.com/pocketbase/pocketbase/apis"
	"github.com/pocketbase/pocketbase/core"
	"github.com/prometheus/client_golang/prometheus/promhttp"

	"git.vakhrushev.me/av/transcriber/internal/clock"
)

func main() {
	// Создаем структурированный логгер
	logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
		Level: slog.LevelInfo,
	}))
	slog.SetDefault(logger)

	// Parse command line flags
	configPath := flag.String("c", "config.toml", "Path to config file")
	flag.StringVar(configPath, "config", "config.toml", "Path to config file (alias for -c)")
	flag.Parse()

	// Load configuration
	cfg, err := config.LoadConfig(*configPath)
	if err != nil {
		logger.Error("Unable to load configuration", "config_path", *configPath, "error", err)
		os.Exit(1)
	} else {
		logger.Info("Configuration loaded successfully", "config_path", *configPath)
	}

	// Незаполненный вход роняет старт: подняться с молча выключенным входом
	// значит остаться открытым наружу, и узнать об этом было бы неоткуда.
	if err := cfg.Auth.Validate(); err != nil {
		logger.Error("Unable to start with incomplete login settings", "error", err)
		os.Exit(1)
	}

	// Загружаем переменные окружения из .env файла
	if err := godotenv.Load(); err != nil {
		logger.Warn("Warning: .env file not found, using system environment variables")
	}

	// Хранилище поднимается библиотекой, а не её набором команд: разбор флагов
	// и мягкая остановка остаются нашими. Схему накатывает Serve — он гоняет
	// непринятые шаги прежде, чем поднять сервер.
	storage, err := pbrepo.New(cfg.Storage.DataDir)
	if err != nil {
		logger.Error("Failed to open storage", "error", err)
		os.Exit(1)
	}
	defer func() {
		if err := storage.ResetBootstrapState(); err != nil {
			logger.Error("Failed to close storage", "error", err)
		}
	}()

	pbrepo.BindPanelRules(storage)

	// Создаем репозитории
	fileRepo := pbrepo.NewFileRepository(storage)
	jobRepo := pbrepo.NewTranscriptJobRepository(storage)

	// Создаем адаптеры
	metaviewer := ffmpegmv.NewFfmpegMetaViewer()
	converter := ffmpegconv.NewFfmpegConverter()

	tgSender, err := telegram.NewTelegramMessageSender(cfg.Telegram.BotToken, logger)
	if err != nil {
		logger.Error("failed to create audio telegram sender", "error", err)
		os.Exit(1)
	}

	recognizer, err := yandex.NewYandexAudioRecognizerService(yandex.YandexAudioRecognizerConfig{
		Region:     cfg.Yandex.ObjStorageRegion,
		AccessKey:  cfg.Yandex.ObjStorageAccessKey,
		SecretKey:  cfg.Yandex.ObjStorageSecretKey,
		BucketName: cfg.Yandex.ObjStorageBucketName,
		Endpoint:   cfg.Yandex.ObjStorageEndpoint,
		ApiKey:     cfg.Yandex.SpeechKitAPIKey,
		FolderID:   cfg.Yandex.FolderID,
	})
	if err != nil {
		logger.Error("failed to create audio recognizer", "error", err)
		os.Exit(1)
	}
	// Отдавать отказ закрытия некому — процесс заканчивается, — поэтому он идёт
	// в журнал владельца. Что он означает: gRPC-клиент отдаёт здесь отказ лишь
	// при повторном закрытии, то есть запись говорит о нашей ошибке, а не о
	// недоступности Yandex.
	defer func() {
		if err := recognizer.Close(); err != nil {
			logger.Error("failed to close audio recognizer", "error", err)
		}
	}()

	// Создаем сервисы
	transcribeService := service.NewTranscribeService(
		jobRepo,
		fileRepo,
		metaviewer,
		converter,
		recognizer,
		tgSender,
		logger,
	)

	// Создаем контекст для graceful shutdown
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Создаем WaitGroup для ожидания завершения всех воркеров
	var wg sync.WaitGroup

	tgConfig := tgcontroller.TelegramConfig{
		BotToken:      cfg.Telegram.BotToken,
		UpdateTimeout: cfg.Telegram.UpdateTimeout,
		UserWhiteList: cfg.Server.UsersWhiteList,
	}

	// Создаем Telegram бот
	tgController, err := tgcontroller.NewTelegramController(tgConfig, transcribeService, jobRepo, logger)
	if err != nil {
		logger.Error("Failed to create Telegram controller", "error", err)
		// Не останавливаем приложение, если Telegram бот не создан
	} else {
		// Запускаем Telegram бот в отдельной горутине
		wg.Add(1)
		go func() {
			defer wg.Done()
			logger.Info("Starting Telegram bot")
			tgController.Start()
			logger.Info("Telegram bot stopped gracefully")
		}()
	}

	// Создаем воркеры
	conversionWorker := worker.NewCallbackWorker("conversion_worker", transcribeService.FindAndRunConversionJob, logger)
	transcribeWorker := worker.NewCallbackWorker("transcribe_worker", transcribeService.FindAndRunTranscribeJob, logger)
	checkWorker := worker.NewCallbackWorker("check_worker", transcribeService.FindAndRunTranscribeCheckJob, logger)

	workers := []worker.Worker{
		conversionWorker,
		transcribeWorker,
		checkWorker,
	}

	// Запускаем воркеры в отдельных горутинах
	for _, w := range workers {
		wg.Add(1)
		go func(worker worker.Worker) {
			defer wg.Done()
			worker.Start(ctx)
			logger.Info("Worker stopped gracefully", "worker", worker.Name())
		}(w)
	}

	// Наши маршруты живут на роутере хранилища: панель отдаётся тем же портом,
	// и второму серверу на нём взяться неоткуда.
	transcribeHandler := httpcontroller.NewTranscribeHandler(jobRepo, transcribeService, logger)
	authHandler := httpcontroller.NewAuthHandler(storage, httpcontroller.AuthHandlerConfig{
		AuthURL:      cfg.Auth.AuthURL,
		RedirectURL:  cfg.Auth.RedirectURL,
		ClientID:     cfg.Auth.ClientID,
		SecureCookie: cfg.Auth.SecureCookie,
	}, logger)

	// Сервер приезжает каналом, а не общей переменной: хук исполняется в
	// горутине сервера, а читает его горутина остановки, и связи «произошло
	// раньше» между ними иначе нет.
	srvCh := make(chan *http.Server, 1)
	storage.OnServe().BindFunc(func(se *core.ServeEvent) error {
		// Шесть часов записи по медленному каналу переживают любой фиксированный
		// таймаут чтения, а умолчание хранилища — пять минут. Стойкость к
		// целенаправленной нагрузке объявлена вне модели угроз проекта.
		se.Server.ReadTimeout = 0
		srvCh <- se.Server

		// Журнал входящих запросов вернулся своим слоем: вместе с gin ушёл
		// `sloggin`, а хранилище пишет запросы в свою таблицу, которой в
		// журнале контейнера не видно. Поля — те, что просит конвенция.
		se.Router.BindFunc(func(e *core.RequestEvent) error {
			start := clock.Start()
			err := e.Next()

			level := slog.LevelInfo
			if e.Request.URL.Path == "/health" || e.Request.URL.Path == "/metrics" {
				// Опрос здоровья и метрик идёт постоянно и полезного не несёт.
				level = slog.LevelDebug
			}

			logger.Log(e.Request.Context(), level, "Incoming request",
				"http.method", e.Request.Method,
				"http.route", e.Request.URL.Path,
				"http.status_code", e.Status(),
				"duration_ms", time.Since(start).Milliseconds(),
				"transport", "http")

			return err
		})

		// Настройки провайдера приводятся к конфигу при каждом подъёме:
		// применённый шаг схемы не переписывается, и секрет, положенный
		// однажды шагом, не пережил бы ротации.
		if err := pbrepo.ApplyProviderSettings(storage, pbrepo.ProviderSettings{
			AuthURL:      cfg.Auth.AuthURL,
			TokenURL:     cfg.Auth.TokenURL,
			UserInfoURL:  cfg.Auth.UserInfoURL,
			ClientID:     cfg.Auth.ClientID,
			ClientSecret: cfg.Auth.ClientSecret,
		}); err != nil {
			return fmt.Errorf("failed to apply provider settings: %w", err)
		}

		authHandler.Register(se.Router)
		transcribeHandler.Register(se.Router)

		se.Router.GET("/health", func(e *core.RequestEvent) error {
			return e.JSON(http.StatusOK, map[string]string{
				"status":  "ok",
				"message": "Transcriber service is running",
			})
		})

		se.Router.GET("/metrics", func(e *core.RequestEvent) error {
			promhttp.Handler().ServeHTTP(e.Response, e.Request)
			return nil
		})

		return se.Next()
	})

	// Запускаем HTTP сервер в отдельной горутине
	serveErr := make(chan error, 1)
	wg.Add(1)
	go func() {
		defer wg.Done()
		logger.Info("Starting HTTP server", "port", cfg.Server.Port)
		err := apis.Serve(storage, apis.ServeConfig{
			HttpAddr:        fmt.Sprintf(":%d", cfg.Server.Port),
			ShowStartBanner: false,
		})
		if err != nil && !errors.Is(err, http.ErrServerClosed) {
			logger.Error("HTTP server error", "error", err)
			serveErr <- err
		}
	}()

	// Настраиваем обработку сигналов для graceful shutdown
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	logger.Info("Transcriber service started with background workers")
	logger.Info("Workers: ConversionWorker, TranscribeWorker, CheckWorker")
	logger.Info("Press Ctrl+C to stop...")

	// Ждем сигнал завершения либо отказ сервера
	select {
	case <-sigChan:
		logger.Info("Received shutdown signal, initiating graceful shutdown...")
	case <-serveErr:
		logger.Error("HTTP server stopped unexpectedly, shutting down")
	}

	if tgController != nil {
		logger.Info("Shutting down Telegram bot...")
		tgController.Stop()
	}

	// Создаем контекст с таймаутом для graceful shutdown HTTP сервера
	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), time.Duration(cfg.Server.ShutdownTimeout)*time.Second)
	defer shutdownCancel()

	// Останавливаем HTTP сервер
	select {
	case srv := <-srvCh:
		logger.Info("Shutting down HTTP server...")
		if err := srv.Shutdown(shutdownCtx); err != nil {
			logger.Error("HTTP server forced to shutdown", "error", err)
		} else {
			logger.Info("HTTP server stopped gracefully")
		}
	default:
		logger.Info("HTTP server was not started, nothing to shut down")
	}

	// Отменяем контекст для остановки воркеров
	cancel()

	// Создаем канал для уведомления о завершении всех воркеров
	done := make(chan struct{})
	go func() {
		wg.Wait()
		close(done)
	}()

	// Ждем завершения всех воркеров или таймаута
	select {
	case <-done:
		logger.Info("All workers stopped gracefully")
	case <-time.After(time.Duration(cfg.Server.ForceShutdownTimeout) * time.Second):
		logger.Warn("Timeout reached, forcing shutdown")
	}

	logger.Info("Transcriber service stopped")
}