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

package lkgateway

// lkagent.go — подключение ядра к кабинету, когда у ядра нет адреса в интернете.
//
// Обычный порядок: кабинет на хостинге сам обращается к bj-server. Но ядро депозитария
// часто стоит во внутренней сети без публичного имени, и «достучаться» до него снаружи
// нельзя. Тогда связь разворачивается: ядро само подключается к кабинету по его
// доменному имени, забирает накопившиеся запросы клиентов, выполняет их у себя и
// возвращает ответы. Наружу из сети депозитария идёт только исходящее соединение по
// TLS — открывать порты и пробрасывать адрес не нужно.
//
// Запрос выполняется тем же обработчиком /api/lk/*, что и в прямом режиме: ядро просто
// прогоняет его через свой mux, как если бы кабинет пришёл по сети.

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"git.zetit.ru/zuevav/Bridge-and-Join-s/modules/lk"
)

// lkAgentState — состояние канала для страницы настроек.
type lkAgentStateT struct {
	Connected atomic.Bool
	LastOK    atomic.Int64  // unix-время последнего успешного обращения
	LastErr   atomic.Value  // string
	Served    atomic.Uint64 // сколько запросов кабинета выполнено
}

var lkAgentState lkAgentStateT

func (a *lkAgentStateT) fail(err error) {
	a.Connected.Store(false)
	a.LastErr.Store(err.Error())
}

func (a *lkAgentStateT) ok() {
	a.Connected.Store(true)
	a.LastOK.Store(time.Now().Unix())
	a.LastErr.Store("")
}

// LKAgentStatus — что показать администратору в настройках.
type LKAgentStatus struct {
	Enabled   bool   `json:"enabled"`
	Cabinet   string `json:"cabinet"`
	Connected bool   `json:"connected"`
	LastOK    string `json:"last_ok,omitempty"`
	LastErr   string `json:"last_error,omitempty"`
	Served    uint64 `json:"served"`
}

func lkAgentStatus(s Settings) LKAgentStatus {
	st := LKAgentStatus{
		Enabled: s.LK.AgentMode, Cabinet: strings.TrimSpace(s.LK.CabinetURL),
		Connected: lkAgentState.Connected.Load(), Served: lkAgentState.Served.Load(),
	}
	if v := lkAgentState.LastOK.Load(); v > 0 {
		st.LastOK = time.Unix(v, 0).Format("02.01.2006 15:04:05")
	}
	if v, _ := lkAgentState.LastErr.Load().(string); v != "" {
		st.LastErr = v
	}
	return st
}

// lkAgentLoop — вечный цикл: подключиться к кабинету, забрать запрос, выполнить, ответить.
// Запускается всегда, но работает только когда режим ожидания включён в настройках:
// так администратор может включить и выключить его без перезапуска службы.
func (s *Server) lkAgentLoop(ctx context.Context) {
	// Несколько подключений сразу: запрос к учётной системе может идти десятки секунд,
	// и один медленный клиент не должен задерживать остальных.
	var wg sync.WaitGroup
	for i := 0; i < lkAgentWorkers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			s.lkAgentWorker(ctx)
		}()
	}
	wg.Wait()
}

// lkAgentWorkers — сколько запросов кабинета ядро обслуживает одновременно.
const lkAgentWorkers = 4

func (s *Server) lkAgentWorker(ctx context.Context) {
	defer func() {
		if rec := recover(); rec != nil {
			log.Printf("кабинет: сбой при обслуживании запроса: %v", rec)
		}
	}()
	client := &http.Client{Timeout: 40 * time.Second}
	for {
		if ctx.Err() != nil {
			return
		}
		cfg := s.rc.Snapshot().LK
		cabinet := strings.TrimRight(strings.TrimSpace(cfg.CabinetURL), "/")
		if !cfg.AgentMode || cabinet == "" || strings.TrimSpace(cfg.APIKey) == "" {
			lkAgentState.Connected.Store(false)
			if !sleepCtx(ctx, 15*time.Second) {
				return
			}
			continue
		}
		task, err := lkAgentFetch(ctx, client, cabinet, cfg.APIKey)
		if err != nil {
			lkAgentState.fail(err)
			log.Printf("кабинет: нет связи с %s: %v", cabinet, err)
			if !sleepCtx(ctx, 20*time.Second) { // не долбим хостинг при аварии
				return
			}
			continue
		}
		lkAgentState.ok()
		if task == nil {
			continue // запросов не было, кабинет отпустил соединение — подключаемся заново
		}
		res := s.lkAgentServe(ctx, task)
		if err := lkAgentReply(ctx, client, cabinet, cfg.APIKey, res); err != nil {
			log.Printf("кабинет: ответ на запрос %s не доставлен: %v", task.ID, err)
			continue
		}
		lkAgentState.Served.Add(1)
	}
}

// lkAgentTask — запрос, ожидающий в кабинете: вызов API или страница в конверте.
type lkAgentTask = lk.PageRequest

// lkAgentFetch — забрать очередной запрос. nil без ошибки — запросов не было.
func lkAgentFetch(ctx context.Context, c *http.Client, cabinet, key string) (*lkAgentTask, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, cabinet+"/agent/tasks", nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("X-BJ-Key", key)
	resp, err := c.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	switch resp.StatusCode {
	case http.StatusNoContent:
		return nil, nil
	case http.StatusOK:
		var t lkAgentTask
		if err := json.NewDecoder(io.LimitReader(resp.Body, 8<<20)).Decode(&t); err != nil {
			return nil, err
		}
		if t.ID == "" || (t.Kind != "page" && !strings.HasPrefix(t.Path, "/api/lk/")) {
			return nil, fmt.Errorf("кабинет прислал неизвестный запрос")
		}
		return &t, nil
	case http.StatusForbidden:
		return nil, fmt.Errorf("кабинет не принял ключ модуля: сверьте BJ_KEY в /etc/bj-lk/lk.env")
	default:
		return nil, fmt.Errorf("кабинет ответил %d", resp.StatusCode)
	}
}

// lkAgentServe — выполнить запрос кабинета у себя: страницу — встроенным кабинетом,
// вызов API — тем же обработчиком /api/lk/*, что и по сети.
func (s *Server) lkAgentServe(ctx context.Context, t *lkAgentTask) lk.PageResponse {
	if t.Kind == "page" {
		if globalLKCabinet == nil {
			return lk.PageResponse{ID: t.ID, Status: http.StatusServiceUnavailable, Body: []byte("кабинет в ядре не запущен")}
		}
		req, _ := http.NewRequestWithContext(ctx, http.MethodGet, "/", nil)
		return globalLKCabinet.serve(req, *t)
	}
	cctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	req, err := http.NewRequestWithContext(cctx, t.Method, t.Path, bytes.NewReader(t.Body))
	if err != nil {
		return lk.PageResponse{ID: t.ID, Status: http.StatusBadRequest, Body: []byte(`{"error":"некорректный запрос кабинета"}`)}
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("X-BJ-Key", s.rc.Snapshot().LK.APIKey) // запрос пришёл по ключу, он же идёт дальше
	if t.Session != "" {
		req.Header.Set("X-BJ-Session", t.Session)
	}
	if t.ClientIP != "" {
		req.Header.Set("X-BJ-Client-IP", t.ClientIP)
	}
	if t.ClientAgent != "" {
		req.Header.Set("X-BJ-Client-Agent", t.ClientAgent)
	}
	req.RemoteAddr = "127.0.0.1:0" // обращение пришло по нашему исходящему каналу
	rec := &lkAgentRecorder{status: http.StatusOK}
	s.mux.ServeHTTP(rec, req)
	return lk.PageResponse{ID: t.ID, Status: rec.status, Body: rec.buf.Bytes()}
}

// lkAgentRecorder — приёмник ответа обработчика: пишем в память, а не в сеть.
type lkAgentRecorder struct {
	status int
	buf    bytes.Buffer
	hdr    http.Header
}

func (w *lkAgentRecorder) Header() http.Header {
	if w.hdr == nil {
		w.hdr = http.Header{}
	}
	return w.hdr
}
func (w *lkAgentRecorder) Write(b []byte) (int, error) { return w.buf.Write(b) }
func (w *lkAgentRecorder) WriteHeader(code int)        { w.status = code }

// lkAgentReply — вернуть кабинету ответ на его запрос.
func lkAgentReply(ctx context.Context, c *http.Client, cabinet, key string, res lk.PageResponse) error {
	payload, _ := json.Marshal(res)
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, cabinet+"/agent/result", bytes.NewReader(payload))
	if err != nil {
		return err
	}
	req.Header.Set("X-BJ-Key", key)
	req.Header.Set("Content-Type", "application/json")
	resp, err := c.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10))
	if resp.StatusCode >= 400 && resp.StatusCode != http.StatusGone {
		return fmt.Errorf("кабинет ответил %d", resp.StatusCode)
	}
	return nil
}

// sleepCtx — пауза, прерываемая остановкой службы.
func sleepCtx(ctx context.Context, d time.Duration) bool {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}