package lkgateway import ( "context" "errors" "fmt" "log" "net/http" "time" "git.zetit.ru/zuevav/Bridge-and-Join-s/internal/m2m" "git.zetit.ru/zuevav/Bridge-and-Join-s/internal/m2mcore" "git.zetit.ru/zuevav/Bridge-and-Join-s/internal/nsdadapter" "git.zetit.ru/zuevav/Bridge-and-Join-s/internal/nsdadapter/igw" "git.zetit.ru/zuevav/Bridge-and-Join-s/internal/nsdadapter/mock" "git.zetit.ru/zuevav/Bridge-and-Join-s/internal/nsdxml" ) // igwClientAdapter адаптирует igw.Client под узкий nsdadapter.IGWClient: // разворачивает (channel, since, type) в igw.ListFilter. type igwClientAdapter struct{ c *igw.Client } func (a igwClientAdapter) SendPackage(ctx context.Context, channel, packageType string, body []byte) (string, error) { return a.c.SendPackage(ctx, channel, packageType, body) } func (a igwClientAdapter) ListIncoming(ctx context.Context, channel string, since time.Time, packageType string) ([]igw.Package, error) { return a.c.ListIncoming(ctx, igw.ListFilter{Channel: channel, Date: since, Type: packageType}) } // ServerConfig — конфигурация HTTP-сервера lk-gateway. type ServerConfig struct { Addr string DefaultSender m2m.DeponentCode DefaultReceiver m2m.DeponentCode CheckOptions func() CheckOptions MockDecisionDelay time.Duration // 0 = дефолт 3 секунды SetupPath string // путь к JSON-файлу runtime-конфига (пусто = ~/.bj/setup.json) } // Server — обвязка HTTP + сервис + workers. type Server struct { cfg ServerConfig svc *Service mock *mock.Sender store *SeedStore rc *RuntimeConfig mux *http.ServeMux server *http.Server // igwClient/igwChannel заполнены только в реальном режиме (ИШ настроен). // На них работает поллер входящих pollIncoming — забирает ответы НРД // (M2MTransferDecision/Response) и применяет через svc.ApplyDecision. igwClient *igw.Client igwChannel string // incoming — входящие переводы (M2MTransferRequest от робота/контрагента, // сценарий 3333), на которые мы отвечаем M2MTransferDecision. incoming *IncomingStore } // NewServer собирает Server с репозиторием, mock NSDSender, SeedStore // и REST + Admin маршрутами. Выбор Repository: // - если в runtime-конфиге (или ENV-fallback в cfg) задан PostgresDSN // и pgx-Pool успешно создаётся — используется PostgresRepository; // - иначе fallback на MemoryRepository (M2-демо). func NewServer(cfg ServerConfig) (*Server, error) { store := NewSeedStore() mockCfg := mock.DefaultConfig() mockCfg.NSDSenderCode = "MC0010300000" if cfg.MockDecisionDelay > 0 { mockCfg.DecisionDelay = cfg.MockDecisionDelay } rc, err := NewRuntimeConfig(cfg.SetupPath) if err != nil { return nil, err } // Код отправителя — наша организация из настроек (приоритет над BJ_M2M_SENDER). if code := orgCode(rc.Snapshot()); code != "" { cfg.DefaultSender = m2m.DeponentCode(code) } // Выбор NSD-сендера: если в runtime-конфиге задан профиль ИШ и URL — // используем реальный nsdadapter поверх REST ИШ; иначе mock-эмуляция. // mockSender остаётся не-nil только в mock-режиме — на нём висит // consumeDecisions (реальные Decision приходят поллером входящих ИШ). var sender m2mcore.NSDSender var mockSender *mock.Sender var igwClient *igw.Client var igwChannel string { s := rc.Snapshot() applyBusinessLine(s.NSD.IGWv904) // config.xml businessLine=MOST для ИШ 107.0.0.904+ if s.NSD.IGWBaseURL != "" && s.NSD.Profile != "" { prof, perr := nsdadapter.LookupProfile(s.NSD.Profile) if perr != nil { log.Printf("lk-gateway: профиль ИШ %q неизвестен (%v) — fallback mock", s.NSD.Profile, perr) } else { prof.IGWBaseURL = s.NSD.IGWBaseURL // override URL из setup.json cl := igw.NewClient(s.NSD.IGWBaseURL, igw.WithDeliveryHook(nsdDelivery)) sender = nsdadapter.NewSender(prof, igwClientAdapter{c: cl}) igwClient = cl // Канал ИШ резолвится по составному коду <канал>+<депонент>. igwChannel = prof.Channel + string(cfg.DefaultSender) log.Printf("lk-gateway: реальный ИШ-адаптер — профиль %s, канал %s, ИШ %s", prof.Name, igwChannel, s.NSD.IGWBaseURL) } } if sender == nil { mockSender = mock.NewSender(mockCfg) sender = mockSender log.Printf("lk-gateway: NSD mock-режим (Decision через эмуляцию)") } } // Repository: pgx если DSN указан, иначе in-memory. var repo m2mcore.Repository = m2mcore.NewMemoryRepository() if dsn := rc.Snapshot().Postgres.DSN; dsn != "" { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) pgRepo, pgErr := m2mcore.NewPostgresRepository(ctx, dsn) cancel() if pgErr != nil { log.Printf("lk-gateway: PostgresRepository отказал, fallback MemoryRepository: %v", pgErr) } else { repo = pgRepo log.Printf("lk-gateway: PostgresRepository подключён (m2m_core.deals)") } } svc := NewService(Config{ Repository: repo, Sender: sender, Store: store, Recorder: m2mcore.NewMemoryRecorder(), DefaultSender: cfg.DefaultSender, DefaultReceiver: cfg.DefaultReceiver, }) // Если runtime-конфиг уже содержит callback URL — применяем его. if s := rc.Snapshot(); s.LK.CallbackURL != "" { svc.callbackURL = s.LK.CallbackURL } mux := http.NewServeMux() RegisterAPI(mux, svc) // CheckOptions берётся из runtime-конфига при каждом запросе на дашборд. checkOpts := func() CheckOptions { s := rc.Snapshot() profile := "не настроен" if devMode(rc) { profile = "demo (mock NSD)" } if s.NSD.Profile != "" { profile = s.NSD.Profile } return CheckOptions{ PostgresDSN: s.Postgres.DSN, CryptoSocket: s.Crypto.SocketPath, NSDAdapterURL: s.NSD.IGWBaseURL, LKCallbackURL: s.LK.CallbackURL, Profile: profile, CryptoProvider: nonEmpty(s.Crypto.Provider, "stub"), Timeout: 2 * time.Second, } } if cfg.CheckOptions != nil { // Опциональный override (например, из cmd/lk-gateway для override ENV-перетягивания). checkOpts = cfg.CheckOptions } adminTpl, err := RegisterAdmin(mux, svc, rc, checkOpts) if err != nil { return nil, err } registerSetup(mux, adminTpl, rc, svc) registerM2MTest(mux, adminTpl, &setupHandlers{rc: rc, tpl: &adminTemplates{a: adminTpl}, svc: svc}) incomingStore := NewIncomingStore() globalIncomingStore = incomingStore // для сводной выгрузки отчётов тест-плана inDeps := incomingDeps{ store: incomingStore, igwClient: igwClient, igwChan: igwChannel, rc: rc, ourCode: string(cfg.DefaultSender), } globalIncomingDeps = &inDeps // автоматический режим M2M registerIncoming(mux, adminTpl, inDeps) globalOutgoingStore = NewOutgoingStore() registerDepoSim(mux, adminTpl) // корень сайта: http://сервер:8080/ → рабочая страница (до входа authMiddleware отправит на вход, // до первичной настройки — на /first-run); прочие неизвестные адреса — 404. mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/" { http.NotFound(w, r) return } if firstRunPending() { http.Redirect(w, r, "/first-run", http.StatusSeeOther) return } http.Redirect(w, r, "/admin/", http.StatusSeeOther) }) registerDepoTest(mux, adminTpl) // тестирование DEPO: сценарии с имитацией НРД, только база депозитария // песочница сверки с DEPO (имитация ответа Fancy) outDeps := outgoingDeps{ store: globalOutgoingStore, igwClient: igwClient, igwChan: igwChannel, svc: svc, // REFERENCE бумаг из M2MTR — для строчных событий DEPO } globalOutgoingDeps = &outDeps // автоматический режим M2M registerOutgoing(mux, adminTpl, outDeps) registerManual(mux, rc, svc) // ручной ввод исходящих переводов по бумажным поручениям registerJournal(mux, adminTpl, svc, incomingStore) registerNSDHealth(mux, rc) // связь с НРД: шапка (json) и «Проверить связь сейчас» registerBackup(mux, rc) // резервное копирование и восстановление (backup.go) registerIgate(mux, rc) // интеграционный шлюз НРД: состояние и установка (igate.go) registerReadiness(mux, rc) // итоговая проверка готовности (мастер, шаг 10) registerQuikReady(mux, rc) // проверка модуля неторговых поручений QUIK (только чтение) registerHTTPS(mux, rc) // HTTPS веб-интерфейса: перенаправление, свой сертификат registerTime(mux, rc) // сервер времени (NTP) registerTimeZone(mux) // часовой пояс сервера registerLetters(mux, rc) // готовые письма: НРД, ARQA, администратор DEPO registerLKAPI(mux, rc, svc) // API для модуля личного кабинета клиента (bj-lk) registerHealth(mux) registerQuikOps(mux, rc, svc) registerNPR(mux, adminTpl, rc) registerAuth(mux, rc) registerSetCallback(mux, svc, rc) registerSeedListing(mux, store) registerHandbookRequest(mux, igwClient, igwChannel) registerChainTest(mux, rc, igwClient != nil) // «Проверить всю цепочку» (мастер, глава «Запуск») return &Server{ cfg: cfg, svc: svc, mock: mockSender, store: store, rc: rc, mux: mux, igwClient: igwClient, igwChannel: igwChannel, incoming: incomingStore, server: &http.Server{ Addr: cfg.Addr, Handler: httpsRedirect(rc, authMiddleware(rc, mux)), ReadHeaderTimeout: 5 * time.Second, }, }, nil } // RuntimeConfig возвращает текущий runtime-конфиг (для тестов). func (s *Server) RuntimeConfig() *RuntimeConfig { return s.rc } func nonEmpty(s, def string) string { if s == "" { return def } return s } // SetCallbackURL обновляет адрес, куда отправлять PATCH callback'и в ЛК. func (s *Server) SetCallbackURL(url string) { s.svc.callbackURL = url } // Service возвращает Service для тестов. func (s *Server) Service() *Service { return s.svc } // Mock возвращает mock-сендер. func (s *Server) Mock() *mock.Sender { return s.mock } // Store возвращает SeedStore. func (s *Server) Store() *SeedStore { return s.store } // Mux возвращает обработчик (для httptest). func (s *Server) Mux() http.Handler { return s.mux } // Run поднимает HTTP-сервер и фоновый Decisions-consumer. // Блокируется до ctx.Done(). func (s *Server) Run(ctx context.Context) error { // Монитор здоровья + systemd watchdog: следит за фоновыми горутинами, // поднимает алерт при зависании и «пинает» systemd, пока всё живо. healthRegister("poll-lk", time.Minute) // Ключ продлеваем до запуска фоновых циклов: режим разработки (имитации и т.п.) должен быть // известен им с первого прохода. if licenseNeedsRenew(s.rc) { lctx, cancel := context.WithTimeout(ctx, 15*time.Second) if err := LicenseRenew(lctx, s.rc); err != nil { log.Printf("lk-gateway: лицензия при запуске: %v", err) } cancel() } go s.superviseHealth(ctx) // Данные модуля личного кабинета — схема lk в PostgreSQL bj. if err := ensureLKSchema(ctx, s.rc); err != nil { log.Printf("lk-gateway: схема lk в PostgreSQL: %v", err) } // Все фоновые горутины — под supervise: паника ловится recover(), пишется // в лог и в ленту, горутина перезапускается. Один сбой не роняет сервис. go superviseLoop(ctx, "consume-decisions", s.rc, s.consumeDecisions) go superviseLoop(ctx, "igw-watch", s.rc, s.igwWatchLoop) // сторож ИШ НРД: оповещения и автоперезапуск go superviseLoop(ctx, "units-watch", s.rc, s.unitsWatchLoop) // службы модулей и ключей Валидаты: письмо администратору go handbookLoop(ctx, s.rc) // справочник участников M2M с сервера лицензий (для установок без шлюза НРД) // Поллер входящих от НРД (только в реальном режиме ИШ): забирает // ответы робота/контрагента и применяет их через ApplyDecision. if s.igwClient != nil && s.igwChannel != "" { healthRegister("poll-incoming", 30*time.Second) go superviseLoop(ctx, "poll-incoming", s.rc, s.pollIncoming) } // Имитатор НРД для заявок ЛК в тестовом режиме (lkemulate.go): решения контрагента // применяются тем же обработчиком, что настоящие ответы из ИШ. lkEmuSender = newLKEmuSender() go superviseLoop(ctx, "lk-emulator", s.rc, s.consumeEmulatedDecisions) // Отказы шлюза по заявкам ЛК: ждём ответ Fancy (DEPO) и только тогда ставим «Отклонена» в ЛК. go superviseLoop(ctx, "lk-flow", s.rc, s.lkFlowLoop) // Ежедневный срез стоимости портфелей: из него кабинет рисует график. go superviseLoop(ctx, "lk-snapshot", s.rc, s.lkSnapshotLoop) // Кабинет клиента в режиме ожидания: ядро само подключается к кабинету на // хостинге и обслуживает запросы клиентов через исходящее соединение. go superviseLoop(ctx, "lk-agent", s.rc, s.lkAgentLoop) lkPreviewApply(s.rc) // кабинет на этом сервере для просмотра, если включён в настройках // Поллер заявок ЛК ЕСИА Финанс: раз в минуту забирает новые depo_m2m и // отправляет в НРД (или имитатору). Интервал 1 мин — время ответов регламентировано. go superviseLoop(ctx, "poll-lk", s.rc, s.pollLKClaims) // Авто-обновление справочника участников M2M раз в сутки (только реальный // ИШ). Тик суточный — в heartbeat-монитор не регистрируем, только recover. go superviseLoop(ctx, "nsd-health", s.rc, s.nsdHealthLoop) // связь с НРД: проверка раз в 10 мин (справочник), история, автоповтор // QUIK МНП: автоматический цикл M2M (регистрация до отправки — хук в // SendBuiltRequest; подпись/снятие/зачисление по MS010 — опросчик). quikSendHook = func(ctx context.Context, req *m2m.M2MTransferRequest) error { return quikOutgoingBeforeSend(ctx, s.rc, req) } healthRegister("poll-quik", 30*time.Second) go superviseLoop(ctx, "poll-quik", s.rc, s.quikCycleLoop) go superviseLoop(ctx, "quik-recipients", s.rc, s.quikRecipientsLoop) // справочник получателей M2M в МНП до 20:00 // Контроль сроков депо-поручений 16/16-1: MS010 за 5 мин, исполнение до 30 дней. healthRegister("order-watch", time.Minute) go superviseLoop(ctx, "order-watch", s.rc, s.orderWatchLoop) // DEPO (bj2depo): события BJ_* в трансферные таблицы, чтение DEPO_*, справочник участников. sentHook = func(req *m2m.M2MTransferRequest) { depoOnOutgoingSent(s.rc, req) } healthRegister("depo-events", time.Minute) go superviseLoop(ctx, "depo-events", s.rc, s.depoEventsLoop) healthRegister("depo-job", 2*time.Minute) go superviseLoop(ctx, "depo-job", s.rc, s.depoJobLoop) healthRegister("m2m-auto", 2*time.Minute) go superviseLoop(ctx, "m2m-auto", s.rc, s.m2mAutoLoop) // Перепроверка групп AD раз в 15 мин (если задана сервисная учётная запись). // Тик длинный, контроллер может отвечать медленно — только recover, без heartbeat. go superviseLoop(ctx, "ad-recheck", s.rc, s.adRecheckLoop) // Фоновая авто-проверка обновлений из артефактории (если включена). go NewUpdater(s.rc).updateLoop(ctx) go refreshVersions(s.rc) // версии систем для «Настройки» собираются заранее // Лицензия: ежесуточное продление ключа на сервере лицензий lic.zetit.ru. go superviseLoop(ctx, "license", s.rc, s.licenseLoop) go publicIPLoop(ctx) // внешний адрес сервера — для письма в ARQA (у сервера лицензий, раз в час) // ИШ старее дистрибутива из обновления bj (ядро перезапускалось при обновлении) — обновить go func() { select { case <-ctx.Done(): case <-time.After(2 * time.Minute): igateAutoUpgrade(ctx, s.rc) } }() // Авто-обновление сертификатов УЦ раз в сутки (если оператор включил). stopCACerts := StartCACertsAutoUpdater(s.rc) defer stopCACerts() // Доверенные сертификаты НРД: вшитый комплект → справочник активного // профиля ИШ (добавление/плановое удаление, с бэкапом и откатом). stopNSDCerts := StartNSDCertsSync(s.rc) stopCertExpiry := StartCertExpiryWatch(s.rc) defer stopNSDCerts() defer stopCertExpiry() // Doc-watcher: раз в сутки проверяет сайт НРД на новые PDF и // эмитирует новости в ленту. Дефолтные источники + дефолтные // новости (окно техработ TEST3, появление робота) сеются один раз. EnsureDocSources(s.rc) SeedDefaultNews(s.rc) stopDocWatcher := StartDocWatcher(s.rc) defer stopDocWatcher() errCh := make(chan error, 1) go func() { log.Printf("lk-gateway: listen %s", s.cfg.Addr) errCh <- s.server.ListenAndServe() }() if hs := startHTTPS(s.rc, s.server.Handler); hs != nil { defer hs.Close() } select { case <-ctx.Done(): shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() _ = s.server.Shutdown(shutdownCtx) return nil case err := <-errCh: if errors.Is(err, http.ErrServerClosed) { return nil } return err } } // consumeDecisions слушает Decisions от mock и обновляет соответствующие сделки. func (s *Server) consumeDecisions(ctx context.Context) { if s.mock == nil { // Реальный ИШ-режим: Decision приходят не из mock-канала, а через // поллер входящих пакетов ИШ (отдельный механизм). Здесь нечего слушать. <-ctx.Done() return } for { select { case <-ctx.Done(): return case d := <-s.mock.Decisions(): if d == nil { continue } if err := s.svc.ApplyDecision(ctx, d); err != nil { log.Printf("lk-gateway: ApplyDecision GUID=%s: %v", d.Header.GUID, err) } else { log.Printf("lk-gateway: Decision применён GUID=%s, callback в %s", d.Header.GUID, s.svc.callbackURL) } } } } // applyBusinessLine синхронизирует флаг «ИШ 107.0.0.904+» с генератором // config.xml: v904 → businessLine=MOST, иначе элемент опускается. func applyBusinessLine(v904 bool) { if v904 { igw.BusinessLine = "MOST" } else { igw.BusinessLine = "" } } func registerHealth(mux *http.ServeMux) { mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("ok\n")) }) } // sendHandbookRequest собирает и отправляет в ИШ запрос справочника (#M2MHR). func sendHandbookRequest(ctx context.Context, igwClient *igw.Client, channel string) (string, error) { if igwClient == nil || channel == "" { return "", fmt.Errorf("ИШ не настроен (реальный режим НРД выключен)") } xmlBytes, err := nsdxml.Marshal(m2m.M2MTransferHandbookRequest{}) if err != nil { return "", fmt.Errorf("сборка XML: %w", err) } zip, err := igw.PackXML(xmlBytes, "M2MTransferHandbookRequest.xml", "#M2MHR") if err != nil { return "", fmt.Errorf("упаковка: %w", err) } return igwClient.SendPackage(ctx, channel, "#M2MHR", zip) } // registerHandbookRequest — POST /admin/setup/handbook-request: ручной запрос // актуального справочника участников M2M. Ответ #M2MTH разберёт поллер. func registerHandbookRequest(mux *http.ServeMux, igwClient *igw.Client, channel string) { mux.HandleFunc("/admin/setup/handbook-request", func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method", http.StatusMethodNotAllowed) return } ctx, cancel := context.WithTimeout(r.Context(), 20*time.Second) defer cancel() id, err := sendHandbookRequest(ctx, igwClient, channel) if err != nil { setupFlash(w, r, "Запрос справочника: "+err.Error()) return } setupFlash(w, r, "Запрос справочника участников M2M отправлен (#M2MHR, пакет id="+id+"). Ответ придёт во входящие через ~30-60с — список появится в логе и в файле m2m-handbook.json.") }) } // registerSetCallback — служебный POST /admin/api/callback-url для // эмулятора ЛК, чтобы сообщить gateway свой URL. Если URL уже сохранён // в runtime-конфиге (пользователь явно настроил его через UI), запрос // эмулятора игнорируется — приоритет у явно настроенного. func registerSetCallback(mux *http.ServeMux, svc *Service, rc *RuntimeConfig) { mux.HandleFunc("/admin/api/callback-url", func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method", http.StatusMethodNotAllowed) return } url := r.URL.Query().Get("url") if url == "" { http.Error(w, "url required", http.StatusBadRequest) return } if rc != nil { if s := rc.Snapshot(); s.LK.CallbackURL != "" { w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("kept-user-configured")) return } } svc.callbackURL = url w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte("ok")) }) } func registerSeedListing(mux *http.ServeMux, store *SeedStore) { mux.HandleFunc("/admin/api/clients", func(w http.ResponseWriter, _ *http.Request) { type c struct { ID, LastName, FirstName, MiddleName string } out := make([]c, 0) for _, cl := range store.Clients() { out = append(out, c{ID: cl.ID, LastName: cl.LastName, FirstName: cl.FirstName, MiddleName: cl.MiddleName}) } writeJSON(w, http.StatusOK, out) }) }