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

package lk

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

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

// agentTask — один запрос кабинета к ядру, ожидающий ответа.
type agentTask struct {
	ID      string `json:"id"`
	Method  string `json:"method"`
	Path    string `json:"path"`
	Body    []byte `json:"body,omitempty"`
	Session string `json:"session,omitempty"`
	// Адрес и браузер посетителя: ядро привязывает к ним сессию и считает по адресу
	// попытки входа. В прямом режиме они идут заголовками, здесь — полями запроса.
	ClientIP    string `json:"client_ip,omitempty"`
	ClientAgent string `json:"client_agent,omitempty"`

	done chan agentResult
}

// agentResult — ответ ядра.
type agentResult struct {
	Status int    `json:"status"`
	Body   []byte `json:"body,omitempty"`
}

// agentQueue — очередь запросов к ядру и ожидающие ответа.
type agentQueue struct {
	mu      sync.Mutex
	queue   []*agentTask
	waiting map[string]*agentTask
	signal  chan struct{}
	seq     uint64
}

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

// push — поставить запрос в очередь и ждать ответа ядра.
func (q *agentQueue) push(ctx context.Context, method, path string, body []byte, session string, wait time.Duration) (agentResult, error) {
	q.mu.Lock()
	q.seq++
	t := &agentTask{ID: idFromSeq(q.seq), Method: method, Path: path, Body: body, Session: session, done: make(chan agentResult, 1)}
	if v, ok := ctx.Value(visitorKey{}).(visitor); ok {
		t.ClientIP, t.ClientAgent = v.ip, v.agent
	}
	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 agentResult{}, ctx.Err()
	case <-timer.C:
		q.forget(t.ID)
		return agentResult{}, 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 agentResult) 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) {
	keyOK := func(w http.ResponseWriter, r *http.Request) bool {
		if !secureEqual(r.Header.Get("X-BJ-Key"), s.cfg.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
		t := s.agent.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(map[string]any{
			"id": t.ID, "method": t.Method, "path": t.Path, "session": t.Session,
			"client_ip": t.ClientIP, "client_agent": t.ClientAgent,
			"body": base64.StdEncoding.EncodeToString(t.Body),
		})
	})
	// Ядро возвращает ответ на запрос.
	mux.HandleFunc("/agent/result", func(w http.ResponseWriter, r *http.Request) {
		if !keyOK(w, r) {
			return
		}
		var in struct {
			ID     string `json:"id"`
			Status int    `json:"status"`
			Body   string `json:"body"`
		}
		if json.NewDecoder(http.MaxBytesReader(w, r.Body, 8<<20)).Decode(&in) != nil || in.ID == "" {
			http.Error(w, "не разобрали ответ", http.StatusBadRequest)
			return
		}
		body, _ := base64.StdEncoding.DecodeString(in.Body)
		if !s.agent.reply(in.ID, agentResult{Status: in.Status, Body: body}) {
			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"})
	})
}