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" "git.vakhrushev.me/av/transcriber/internal/contract" 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{ UpdateTimeout: cfg.Telegram.UpdateTimeout, UserWhiteList: cfg.Server.UsersWhiteList, } // Клиента бота заводит единая точка: её отказ не несёт токена, тогда как // отказ `NewBotAPI` несёт — он ходит за `getMe`. tgController, err := newTelegramController(cfg.Telegram.BotToken, 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(ctx) 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") } // newTelegramController собирает бота и транспорт вокруг него. Токен доходит // до единой точки `internal/adapter/telegram` и дальше не идёт: транспорт его // не видит вовсе, а отказ, который увидит журнал, адреса с токеном не несёт. func newTelegramController( botToken string, cfg tgcontroller.TelegramConfig, transcribeService *service.TranscribeService, jobRepo contract.TranscriptJobRepository, logger *slog.Logger, ) (*tgcontroller.TelegramController, error) { bot, err := telegram.NewBot(botToken, logger) if err != nil { return nil, err } return tgcontroller.NewTelegramController(cfg, bot, transcribeService, jobRepo, logger) }