package main import ( "context" "errors" "flag" "fmt" "log/slog" "net/http" "os" "os/signal" "strings" "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/config" httpcontroller "git.vakhrushev.me/av/transcriber/internal/controller/http" "git.vakhrushev.me/av/transcriber/internal/controller/worker" "git.vakhrushev.me/av/transcriber/internal/metrics" "git.vakhrushev.me/av/transcriber/internal/service" "git.vakhrushev.me/av/transcriber/web" "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) } // Числа конвейера проверяются здесь же: ноль воркеров — объявленный режим, а // отрицательное число и нулевой предел простоя — опечатка, и подниматься с // ней значит остановить всякую запись первым же захватом. if err := cfg.Pipeline.Validate(); err != nil { logger.Error("Unable to start with incorrect pipeline 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) // Создаем репозитории recordRepo := pbrepo.NewAudioRecordRepository(storage) repos := service.Repositories{ Records: recordRepo, Files: pbrepo.NewFileRepository(storage), Texts: pbrepo.NewTextRepository(storage), Structures: pbrepo.NewStructureRepository(storage), Recognitions: pbrepo.NewRecognitionRepository(storage), Events: pbrepo.NewRecordEventRepository(storage), } // Создаем адаптеры metaviewer := ffmpegmv.NewFfmpegMetaViewer() converter := ffmpegconv.NewFfmpegConverter() 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( repos, metaviewer, converter, recognizer, cfg.Pipeline.StuckLimits(), logger, ) // Создаем контекст для graceful shutdown ctx, cancel := context.WithCancel(context.Background()) defer cancel() // Создаем WaitGroup для ожидания завершения всех воркеров var wg sync.WaitGroup // Пул одинаковых воркеров: специализации у них нет, шаг выбирается по рубежу // самой записи. Число приходит настройкой, ноль — законное значение. pool := worker.NewPool(cfg.Pipeline.Workers, transcribeService.RunStep, logger) wg.Add(1) go func() { defer wg.Done() pool.Start(ctx) }() // Вход у сервиса один — приём по HTTP, — и метка ставится только ему. Метки // убранного входа Telegram здесь нет намеренно: ноль читался бы как поломка, // а признак существует ради того дня, когда входов снова станет больше. metrics.IntakeUpGauge.WithLabelValues("http").Set(1) // Наши маршруты живут на роутере хранилища: панель отдаётся тем же портом, // и второму серверу на нём взяться неоткуда. appHandler := httpcontroller.NewAppHandler(recordRepo, repos.Texts, repos.Structures, 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) // Адресное пространство сервиса объявлено одним перечнем, и он порождает // регистрацию, а не описывает её: корень, заведённый мимо перечня, не // получит обработчика вовсе. Отсюда же уровень журнала для адресов // наблюдения и правило неизвестного пути у раздачи приложения. mounts := httpcontroller.ServiceMounts(appHandler, authHandler, promhttp.Handler()) dist, appBuilt := web.Dist() webappHandler := httpcontroller.NewWebappHandler(dist, appBuilt, mounts, 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 httpcontroller.IsObservationAddress(mounts, e.Request.URL.Path) { // Опрос здоровья и метрик идёт постоянно и полезного не несёт. level = slog.LevelDebug } attrs := []any{ "http.method", e.Request.Method, "http.route", journalRoute(e.Request.URL.Path, mounts), "http.status_code", e.Status(), "duration_ms", time.Since(start).Milliseconds(), "transport", "http", } // Путь, отданный приложению, в журнал не идёт — вместо него исход // и длина: по ним видно, что происходит, а множеством значений // самого пути распоряжается спрашивающий. if outcome := httpcontroller.WebappOutcome(e); outcome != "" { attrs = append(attrs, "webapp.outcome", outcome, "http.path_length", len(e.Request.URL.Path)) } logger.Log(e.Request.Context(), level, "Incoming request", attrs...) 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) } // Своё правило ограничителя частоты под корень приложения. Правило // хранилища настроено на его собственный корень и наших адресов больше не // покрывает: вместе с переездом приложения ограничитель перестал бы // существовать для него вовсе, и заметить это было бы нечем. if err := httpcontroller.ApplyAppRateLimit(storage); err != nil { return fmt.Errorf("failed to apply app rate limit: %w", err) } httpcontroller.RegisterServiceRoutes(se.Router, mounts) // Раздача приложения вешается последней: она занимает корень, и всё, // что не совпало ни с одним адресом сервиса, доходит до неё. webappHandler.Register(se.Router) 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", "pipeline_workers", pool.Size()) 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") } // Создаем контекст с таймаутом для 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") } // filesPathPrefix — начало пути, которым хранилище отдаёт файл записи. Последний // сегмент такого пути и есть имя файла в хранилище. const filesPathPrefix = "/api/files/" // webappRoute — чем в журнале обозначается всякий путь, отданный приложению. const webappRoute = "<приложение>" // journalRoute готовит путь запроса к записи в журнал. // // Инвариант проекта запрещает имени файла в хранилище попадать в журнал: имя — // последняя часть ссылки `/api/files/...`, и строка журнала вместе с // идентификатором записи собирала бы ссылку целиком. Слой журнала пишет путь // всякого запроса, поэтому имя срезается здесь — иначе оно уезжало бы в // собранные логи при каждом скачивании записи. // // Срезается только имя: маршрут остаётся различимым, и наблюдаемость от этого не // теряется. // // Путь, не принадлежащий сервису, в журнал не идёт вовсе. До появления раздачи // приложения такой путь ловил отказ маршрутизатора, а теперь получает разметку // с кодом `200`: множеством его значений распоряжается спрашивающий, и // дословная строка сделала бы журнал местом, куда аноним пишет свой текст. func journalRoute(path string, mounts []httpcontroller.Mount) string { if !httpcontroller.IsServiceAddress(mounts, path) { return webappRoute } if !strings.HasPrefix(path, filesPathPrefix) { return path } cut := strings.LastIndex(path, "/") if cut < len(filesPathPrefix) { return path } return path[:cut+1] + "<имя>" }