Bridge-and-Join-s/modules/lk/agent.go

package lk

// agent.go — режим ожидания для случая, когда у ядра нет адреса, доступного кабинету.
//
// Обычно кабинет сам ходит в ядро: BJ_URL указывает на bj-server. Но ядро часто стоит во
// внутренней сети депозитария без публичного имени, а кабинет — на хостинге. Тогда связь
// разворачивается наоборот: кабинет держит очередь запросов, а ядро само подключается к
// кабинету, забирает запрос, выполняет его у себя и возвращает ответ. Наружу из сети
// депозитария идёт только исходящее соединение.
//
//	клиент → страница кабинета → очередь → (ядро забирает) → ответ → страница
//
// Оба обращения ядра защищены тем же ключом модуля, что и прямой режим.

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"sync"
	"sync/atomic"
	"time"
)

// agentTask — один запрос к ядру, ожидающий ответа: вызов API (старый кабинет) или
// страница целиком в конверте (ретранслятор).
type agentTask struct {
	PageRequest
	done chan PageResponse
}

// agentQueue — очередь запросов к ядру и ожидающие ответа.
type agentQueue struct {
	mu      sync.Mutex
	queue   []*agentTask
	waiting map[string]*agentTask
	signal  chan struct{}
	seq     uint64
	// lastFetch — когда ядро в последний раз забирало запросы: по этому установщик
	// понимает, что ядро дозвонилось до кабинета.
	lastFetch atomic.Int64
}

// coreSeen — ядро подключалось за последнюю минуту.
func (q *agentQueue) coreSeen() bool {
	t := q.lastFetch.Load()
	return t > 0 && time.Since(time.Unix(t, 0)) < time.Minute
}

func newAgentQueue() *agentQueue {
	return &agentQueue{waiting: map[string]*agentTask{}, signal: make(chan struct{}, 1)}
}

// push — вызов API через очередь (кабинет с отрисовкой на своём сервере).
func (q *agentQueue) push(ctx context.Context, method, path string, body []byte, session string, wait time.Duration) (PageResponse, error) {
	env := PageRequest{Method: method, Path: path, Body: body, Session: session}
	if v, ok := ctx.Value(visitorKey{}).(visitor); ok {
		env.ClientIP, env.ClientAgent = v.ip, v.agent
	}
	return q.pushEnvelope(ctx, env, wait)
}

// pushEnvelope — поставить конверт в очередь и ждать ответа ядра.
func (q *agentQueue) pushEnvelope(ctx context.Context, env PageRequest, wait time.Duration) (PageResponse, error) {
	q.mu.Lock()
	q.seq++
	env.ID = idFromSeq(q.seq)
	t := &agentTask{PageRequest: env, done: make(chan PageResponse, 1)}
	q.queue = append(q.queue, t)
	q.waiting[t.ID] = t
	q.mu.Unlock()
	select {
	case q.signal <- struct{}{}:
	default:
	}
	timer := time.NewTimer(wait)
	defer timer.Stop()
	select {
	case res := <-t.done:
		return res, nil
	case <-ctx.Done():
		q.forget(t.ID)
		return PageResponse{}, ctx.Err()
	case <-timer.C:
		q.forget(t.ID)
		return PageResponse{}, errors.New("ядро не ответило: проверьте, что bj-server подключён к кабинету")
	}
}

func (q *agentQueue) forget(id string) {
	q.mu.Lock()
	delete(q.waiting, id)
	q.mu.Unlock()
}

// next — выдать ядру очередной запрос; ждёт до wait, если очередь пуста.
func (q *agentQueue) next(ctx context.Context, wait time.Duration) *agentTask {
	for {
		q.mu.Lock()
		if len(q.queue) > 0 {
			t := q.queue[0]
			q.queue = q.queue[1:]
			q.mu.Unlock()
			return t
		}
		q.mu.Unlock()
		timer := time.NewTimer(wait)
		select {
		case <-q.signal:
			timer.Stop()
		case <-timer.C:
			return nil
		case <-ctx.Done():
			timer.Stop()
			return nil
		}
	}
}

// reply — вернуть ответ ядра ожидающему запросу.
func (q *agentQueue) reply(id string, res PageResponse) bool {
	q.mu.Lock()
	t, ok := q.waiting[id]
	delete(q.waiting, id)
	q.mu.Unlock()
	if !ok {
		return false
	}
	select {
	case t.done <- res:
	default:
	}
	return true
}

func idFromSeq(n uint64) string {
	const alphabet = "0123456789abcdef"
	buf := make([]byte, 0, 16)
	for i := 0; i < 16; i++ {
		buf = append(buf, alphabet[(n>>(uint(i)*4))&0xf])
	}
	return string(buf)
}

// registerAgent — точки, куда подключается ядро (только в режиме ожидания).
func (s *Server) registerAgent(mux *http.ServeMux) {
	// В прямом режиме этих точек как бы нет: очередь не создана — отвечаем 404.
	keyOK := func(w http.ResponseWriter, r *http.Request) bool {
		if s.agentQueue() == nil {
			http.NotFound(w, r)
			return false
		}
		if !secureEqual(r.Header.Get("X-BJ-Key"), s.conf().BJKey) {
			http.Error(w, "неверный ключ модуля", http.StatusForbidden)
			return false
		}
		return true
	}
	// Ядро забирает очередной запрос; если запросов нет — ответ 204 через несколько секунд.
	mux.HandleFunc("/agent/tasks", func(w http.ResponseWriter, r *http.Request) {
		if !keyOK(w, r) {
			return
		}
		wait := 25 * time.Second
		q := s.agentQueue()
		q.lastFetch.Store(time.Now().Unix())
		t := q.next(r.Context(), wait)
		if t == nil {
			w.WriteHeader(http.StatusNoContent)
			return
		}
		w.Header().Set("Content-Type", "application/json; charset=utf-8")
		_ = json.NewEncoder(w).Encode(t.PageRequest)
	})
	// Ядро возвращает ответ на запрос.
	mux.HandleFunc("/agent/result", func(w http.ResponseWriter, r *http.Request) {
		if !keyOK(w, r) {
			return
		}
		var in PageResponse
		if json.NewDecoder(http.MaxBytesReader(w, r.Body, 32<<20)).Decode(&in) != nil || in.ID == "" {
			http.Error(w, "не разобрали ответ", http.StatusBadRequest)
			return
		}
		if !s.agentQueue().reply(in.ID, in) {
			w.WriteHeader(http.StatusGone) // запрос уже не ждут: клиент ушёл или истёк срок
			return
		}
		w.WriteHeader(http.StatusNoContent)
	})
	// Состояние связи для страницы «не настроено» и проверки администратором.
	mux.HandleFunc("/agent/state", func(w http.ResponseWriter, r *http.Request) {
		if !keyOK(w, r) {
			return
		}
		w.Header().Set("Content-Type", "application/json; charset=utf-8")
		_ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "mode": "agent"})
	})
}