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"}) }) }