Bridge-and-Join-s/internal/lkgateway/npr.go

package lkgateway

// Модуль «НПР / УДС» — риск-монитор клиентов маржинальной торговли по базе экспорта QUIK.
// Сам модуль работает отдельной службой bj-npr (modules/npr, обновляется независимо). Здесь —
// его настройки и проверка подключения, а раздел /admin/npr bj-server проксирует службе: проверяет
// включение и лицензию (права — authMiddleware), передаёт служебный ключ и встраивает полные
// страницы модуля в свой макет. Если служба не отвечает, показывает понятную страницу.

import (
	"context"
	"fmt"
	"html/template"
	"io"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
)

// NPRSettings — модуль НПР/УДС.
type NPRSettings struct {
	Enabled bool   `json:"enabled"`
	DSN     string `json:"dsn"` // postgres://quik_ro:...@host:5432/quik_export
}

func (r *RuntimeConfig) UpdateNPR(s NPRSettings) error {
	r.mu.Lock()
	r.data.NPR = s
	r.data.UpdatedAt = time.Now().UTC()
	r.mu.Unlock()
	return r.save()
}

// nprAvailable — модуль включён в настройках и разрешён лицензией.
func nprAvailable(rc *RuntimeConfig) bool {
	if rc == nil {
		return false
	}
	s := rc.Snapshot().NPR
	return s.Enabled && strings.TrimSpace(s.DSN) != "" && moduleAllowed(rc, "npr")
}

// ---- служба модуля ----

var nprHTTP = &http.Client{Timeout: 60 * time.Second, CheckRedirect: noRedirect}

func nprURL() string {
	if u := strings.TrimSpace(os.Getenv("BJ_NPR_URL")); u != "" {
		return strings.TrimRight(u, "/")
	}
	return "http://127.0.0.1:8083"
}

// registerNPR — прокси раздела /admin/npr к службе bj-npr.
func registerNPR(mux *http.ServeMux, a *admin, rc *RuntimeConfig) {
	unavailable := func(w http.ResponseWriter, r *http.Request, status int, title, text, hint string) {
		if isAjax(r) || !strings.HasPrefix(r.Header.Get("Accept"), "text/html") && r.URL.Path != "/admin/npr" && !strings.HasPrefix(r.URL.Path, "/admin/npr/c/") {
			http.Error(w, text, status)
			return
		}
		w.WriteHeader(status)
		render(w, a.modulePage, modulePageData{page: nowPage("НПР / УДС", "npr"), Content: moduleMessage(title, text, hint)})
	}
	proxy := func(w http.ResponseWriter, r *http.Request) {
		if !strings.HasPrefix(r.URL.Path, "/admin/npr/static/") && !nprAvailable(rc) {
			unavailable(w, r, http.StatusForbidden, "Модуль «НПР / УДС» недоступен",
				"Модуль не включён в настройках или не входит в лицензию.", "Включение и подключение к базе: Настройка → НПР / УДС. Состав лицензии: Настройка → Лицензия.")
			return
		}
		key, err := moduleKey()
		if err != nil {
			http.Error(w, "служебный ключ модулей: "+err.Error(), http.StatusInternalServerError)
			return
		}
		req, err := http.NewRequestWithContext(r.Context(), r.Method, nprURL()+r.URL.RequestURI(), r.Body)
		if err != nil {
			http.Error(w, err.Error(), http.StatusBadRequest)
			return
		}
		for _, h := range []string{"Accept", "Content-Type", "HX-Request", "HX-Target", "HX-Trigger", "HX-Current-URL", "If-Modified-Since", "If-None-Match", "Range"} {
			if v := r.Header.Get(h); v != "" {
				req.Header.Set(h, v)
			}
		}
		req.Header.Set("X-BJ-Module-Key", key)
		if u := currentUser(w); u != nil {
			req.Header.Set("X-BJ-User", url.QueryEscape(u.Login))
		}
		resp, err := nprHTTP.Do(req)
		if err != nil {
			unavailable(w, r, http.StatusServiceUnavailable, "Служба модуля не отвечает",
				"Служба bj-npr не запущена или ещё не установлена на этом сервере.",
				"Установка и обновление модуля: Настройка → Обновления. Состояние службы: systemctl status bj-npr.")
			return
		}
		defer resp.Body.Close()
		if resp.Header.Get("X-BJ-Layout") == "page" {
			body, err := io.ReadAll(io.LimitReader(resp.Body, 16<<20))
			if err != nil {
				http.Error(w, err.Error(), http.StatusBadGateway)
				return
			}
			title, _ := url.PathUnescape(resp.Header.Get("X-BJ-Title"))
			render(w, a.modulePage, modulePageData{page: nowPage(pickStr(title, "НПР / УДС"), "npr"), Content: template.HTML(body)})
			return
		}
		for _, h := range []string{"Content-Type", "Content-Disposition", "Content-Length", "Cache-Control", "Last-Modified", "ETag", "Accept-Ranges", "Content-Range", "Location"} {
			if v := resp.Header.Get(h); v != "" {
				w.Header().Set(h, v)
			}
		}
		w.WriteHeader(resp.StatusCode)
		_, _ = io.Copy(w, resp.Body)
	}
	mux.HandleFunc("/admin/npr", proxy)
	mux.HandleFunc("/admin/npr/", proxy)
}

// ---- настройки: проверка подключения ----

var (
	nprPoolMu  sync.Mutex
	nprPool    *pgxpool.Pool
	nprPoolDSN string
)

func nprDB(rc *RuntimeConfig) (*pgxpool.Pool, error) {
	dsn := strings.TrimSpace(rc.Snapshot().NPR.DSN)
	if dsn == "" {
		return nil, fmt.Errorf("модуль НПР: не задано подключение к базе экспорта QUIK")
	}
	nprPoolMu.Lock()
	defer nprPoolMu.Unlock()
	if nprPool != nil && nprPoolDSN == dsn {
		return nprPool, nil
	}
	if nprPool != nil {
		nprPool.Close()
		nprPool = nil
	}
	cfg, err := pgxpool.ParseConfig(dsn)
	if err != nil {
		return nil, err
	}
	cfg.MaxConns = 2
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	p, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		return nil, err
	}
	nprPool, nprPoolDSN = p, dsn
	return p, nil
}

// nprTest — проверка подключения для настроек.
func nprTest(rc *RuntimeConfig) (string, error) {
	p, err := nprDB(rc)
	if err != nil {
		return "", err
	}
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	var clients, hist int64
	var tick *time.Time
	if err := p.QueryRow(ctx, "SELECT count(DISTINCT client_code) FROM npr.v_current").Scan(&clients); err != nil {
		return "", fmt.Errorf("npr.v_current: %v", err)
	}
	_ = p.QueryRow(ctx, "SELECT count(*) FROM npr.history").Scan(&hist)
	_ = p.QueryRow(ctx, "SELECT last_tick FROM npr.heartbeat WHERE id=1").Scan(&tick)
	age := "нет пульса"
	if tick != nil {
		age = fmt.Sprintf("пульс экспортёра %s назад", time.Since(*tick).Round(time.Second))
	}
	service := "служба bj-npr не отвечает"
	if resp, err := (&http.Client{Timeout: 3 * time.Second}).Get(nprURL() + "/healthz"); err == nil {
		resp.Body.Close()
		service = "служба bj-npr работает"
	}
	return fmt.Sprintf("клиентов в текущем состоянии %d, записей истории %d, %s; %s", clients, hist, age, service), nil
}

func (h *setupHandlers) saveNPR(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method", http.StatusMethodNotAllowed)
		return
	}
	s := NPRSettings{Enabled: r.FormValue("enabled") != "", DSN: strings.TrimSpace(r.FormValue("dsn"))}
	if err := h.rc.UpdateNPR(s); err != nil {
		setupFlash(w, r, "Модуль НПР: ошибка сохранения: "+err.Error())
		return
	}
	if s.Enabled {
		setupFlash(w, r, "Модуль НПР / УДС включён")
	} else {
		setupFlash(w, r, "Модуль НПР / УДС выключен")
	}
}

func (h *setupHandlers) testNPR(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method", http.StatusMethodNotAllowed)
		return
	}
	if dsn := strings.TrimSpace(r.FormValue("dsn")); dsn != "" {
		s := h.rc.Snapshot().NPR
		if s.DSN != dsn {
			s.DSN = dsn
			_ = h.rc.UpdateNPR(s)
		}
	}
	msg, err := nprTest(h.rc)
	if err != nil {
		setupFlash(w, r, "Модуль НПР: подключение НЕ удалось — "+err.Error())
		return
	}
	setupFlash(w, r, "Модуль НПР: подключение OK — "+msg)
}