Глава 13. Деплой и инфраструктура

package agentsvc

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/anthropics/anthropic-sdk-go"
)

// SessionStore — внешнее durable-хранилище состояния (Redis/Postgres).
// Обработчик stateless: историю читаем в начале, пишем в конце.
type SessionStore interface {
	Load(ctx context.Context, sessionID string) ([]anthropic.MessageParam, error)
	Save(ctx context.Context, sessionID string, msgs []anthropic.MessageParam) error
}

// Server — агент как сервис.
type Server struct {
	client anthropic.Client
	store  SessionStore
	// sem — семафор-пул: ограничивает число одновременных вызовов модели,
	// чтобы не исчерпать память и не упереться в rate limit провайдера.
	sem chan struct{}
}

func NewServer(client anthropic.Client, store SessionStore, maxConcurrent int) *Server {
	return &Server{
		client: client,
		store:  store,
		sem:    make(chan struct{}, maxConcurrent),
	}
}

type chatRequest struct {
	SessionID string // json:"session_id"
	Message   string // json:"message"
}

type chatResponse struct {
	Reply string // json:"reply"
}

func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) {
	// Таймаут на запрос: зависший вызов модели будет оборван по дедлайну,
	// а отмена клиента распространится вниз по ctx.
	ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
	defer cancel()

	var req chatRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "bad request", http.StatusBadRequest)
		return
	}

	// Занимаем слот в пуле; если все заняты — ждём или отдаём 503 по таймауту.
	select {
	case s.sem <- struct{}{}:
		defer func() { <-s.sem }()
	case <-ctx.Done():
		http.Error(w, "busy", http.StatusServiceUnavailable)
		return
	}

	// 1. Читаем состояние из внешнего хранилища (stateless-обработчик).
	history, err := s.store.Load(ctx, req.SessionID)
	if err != nil {
		http.Error(w, "store error", http.StatusInternalServerError)
		return
	}

	// 2. Добавляем реплику пользователя и вызываем модель.
	history = append(history,
		anthropic.NewUserMessage(anthropic.NewTextBlock(req.Message)))

	resp, err := s.client.Messages.New(ctx, anthropic.MessageNewParams{
		Model:     anthropic.ModelClaudeOpus4_8,
		MaxTokens: 4096,
		Messages:  history,
	})
	if err != nil {
		http.Error(w, "model error", http.StatusBadGateway)
		return
	}

	// 3. Сохраняем обновлённую историю обратно (ToParam конвертирует ответ).
	history = append(history, resp.ToParam())
	if err := s.store.Save(ctx, req.SessionID, history); err != nil {
		http.Error(w, "store error", http.StatusInternalServerError)
		return
	}

	var reply string
	for _, block := range resp.Content {
		if t, ok := block.AsAny().(anthropic.TextBlock); ok {
			reply += t.Text
		}
	}
	_ = json.NewEncoder(w).Encode(chatResponse{Reply: reply})
}

// Run поднимает сервер и обеспечивает graceful shutdown по SIGTERM/SIGINT.
func (s *Server) Run(addr string) error {
	mux := http.NewServeMux()
	mux.HandleFunc("/chat", s.handleChat)
	srv := &http.Server{Addr: addr, Handler: mux}

	// Слушаем сигналы завершения от оркестратора.
	stop := make(chan os.Signal, 1)
	signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		if err := srv.ListenAndServe(); err != nil &&
			!errors.Is(err, http.ErrServerClosed) {
			log.Fatalf("listen: %v", err)
		}
	}()

	<-stop
	// Перестаём принимать новые запросы, даём текущим доработать в дедлайн.
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	return srv.Shutdown(shutdownCtx)
}

package agentsvc

import (
	"context"
	"log"
	"sync"
)

// Task — асинхронная задача агента (например, генерация отчёта).
// IdempotencyKey защищает от двойного выполнения при ретрае.
type Task struct {
	ID             string
	IdempotencyKey string
	Payload        string
}

// ProcessedSet — хранилище уже обработанных ключей (в проде — Redis/Postgres).
type ProcessedSet interface {
	// SeenOrMark возвращает true, если ключ уже обработан; иначе помечает его.
	SeenOrMark(ctx context.Context, key string) (seen bool, err error)
}

// Pool — пул воркеров, разбирающий задачи из канала.
type Pool struct {
	tasks     chan Task
	processed ProcessedSet
	handle    func(ctx context.Context, t Task) error
	wg        sync.WaitGroup
}

func NewPool(workers int, processed ProcessedSet, handle func(context.Context, Task) error) *Pool {
	p := &Pool{
		tasks:     make(chan Task, 256),
		processed: processed,
		handle:    handle,
	}
	for i := 0; i < workers; i++ {
		p.wg.Add(1)
		go p.worker()
	}
	return p
}

func (p *Pool) worker() {
	defer p.wg.Done()
	ctx := context.Background()
	// Цикл завершится, когда канал tasks закроют (graceful shutdown).
	for t := range p.tasks {
		// Идемпотентность: повторно пришедшую задачу не выполняем дважды.
		seen, err := p.processed.SeenOrMark(ctx, t.IdempotencyKey)
		if err != nil {
			log.Printf("task %s: processed-check failed: %v", t.ID, err)
			continue
		}
		if seen {
			continue
		}
		if err := p.handle(ctx, t); err != nil {
			// В проде здесь ретрай с backoff или отправка в dead-letter-очередь.
			log.Printf("task %s failed: %v", t.ID, err)
		}
	}
}

// Submit кладёт задачу в очередь.
func (p *Pool) Submit(t Task) { p.tasks <- t }

// Shutdown закрывает вход и ждёт, пока воркеры дренируют очередь.
func (p *Pool) Shutdown() {
	close(p.tasks)
	p.wg.Wait()
}

package agentsvc

import (
	"context"
	"errors"
	"math"
	"math/rand"
	"strconv"
	"time"

	"github.com/anthropics/anthropic-sdk-go"
)

// callModel — функция одного вызова модели. Она ДОЛЖНА быть идемпотентной:
// повтор не создаёт побочных эффектов на нашей стороне. Иначе ретраить нельзя
// без ключа идемпотентности (см. воркер очереди выше).
type callModel func(context.Context) (*anthropic.Message, error)

// retryAfterSeconds достаёт паузу из заголовка Retry-After ответа об ошибке.
// Возвращает 0, если заголовка нет. Заголовки 429/529 и anthropic-ratelimit-*
// сверяйте с актуальными доками — имена и числа меняются.
func retryAfterSeconds(err error) (time.Duration, bool) {
	var apiErr *anthropic.Error
	if !errors.As(err, &apiErr) {
		return 0, false
	}
	// Ретраим только перегрузку/лимит: 429 (rate_limit) и 529 (overloaded).
	if apiErr.StatusCode != 429 && apiErr.StatusCode != 529 {
		return 0, false
	}
	if v := apiErr.Response.Header.Get("Retry-After"); v != "" {
		if secs, perr := strconv.Atoi(v); perr == nil {
			return time.Duration(secs) * time.Second, true
		}
	}
	// 429/529 без Retry-After: ретраим, но паузу посчитаем сами (backoff).
	return 0, true
}

// CallWithRetry повторяет идемпотентный вызов с уважением к Retry-After,
// иначе — экспоненциальный backoff с джиттером. SDK уже делает базовые ретраи
// (настраиваются через option.WithMaxRetries); здесь — слой нашей бизнес-логики.
func CallWithRetry(
	ctx context.Context,
	call callModel,
	maxAttempts int,
) (*anthropic.Message, error) {
	const base = 500 * time.Millisecond
	const cap = 30 * time.Second

	var lastErr error
	for attempt := 0; attempt < maxAttempts; attempt++ {
		msg, err := call(ctx)
		if err == nil {
			return msg, nil
		}
		lastErr = err

		wait, retryable := retryAfterSeconds(err)
		if !retryable {
			// Не лимит и не перегрузка — повтор не поможет, отдаём ошибку.
			return nil, err
		}
		if wait == 0 {
			// Retry-After не пришёл: экспоненциальный backoff + джиттер,
			// чтобы клиенты не пошли на повтор синхронно (thundering herd).
			backoff := time.Duration(float64(base) * math.Pow(2, float64(attempt)))
			if backoff > cap {
				backoff = cap
			}
			wait = backoff + time.Duration(rand.Int63n(int64(base)))
		}

		select {
		case <-time.After(wait):
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}
	return nil, lastErr
}

package agentsvc

import (
	"context"
	"errors"
	"sync"
)

// ErrTenantBusy возвращается, когда тенант исчерпал свою долю конкурентности.
// На уровне HTTP это превращается в 503 (load shedding), а не в бесконечную
// очередь до OOM.
var ErrTenantBusy = errors.New("tenant concurrency limit reached")

// TenantLimiter ограничивает число одновременных вызовов модели ОТДЕЛЬНО
// для каждого тенанта, чтобы «шумный сосед» не выел общий лимит провайдера.
type TenantLimiter struct {
	mu       sync.Mutex
	perLimit int
	sems     map[string]chan struct{}
}

func NewTenantLimiter(perTenant int) *TenantLimiter {
	return &TenantLimiter{
		perLimit: perTenant,
		sems:     make(map[string]chan struct{}),
	}
}

// semFor лениво создаёт семафор для тенанта.
func (l *TenantLimiter) semFor(tenant string) chan struct{} {
	l.mu.Lock()
	defer l.mu.Unlock()
	sem, ok := l.sems[tenant]
	if !ok {
		sem = make(chan struct{}, l.perLimit)
		l.sems[tenant] = sem
	}
	return sem
}

// Do выполняет fn под лимитом конкретного тенанта. Если слотов нет —
// немедленно отклоняет (ErrTenantBusy), а не копит очередь бесконечно.
// Так всплеск одного тенанта не обрушивает SLA остальных.
func (l *TenantLimiter) Do(
	ctx context.Context,
	tenant string,
	fn func(context.Context) error,
) error {
	sem := l.semFor(tenant)
	select {
	case sem <- struct{}{}:
		defer func() { <-sem }()
	default:
		// Слотов тенанта нет: отклоняем быстро (load shedding).
		return ErrTenantBusy
	}

	// Если хотим короткое ожидание вместо мгновенного отказа — здесь можно
	// добавить select с ctx и небольшим таймаутом перед тем, как сдаться.
	return fn(ctx)
}