pipu-music/services/api/internal/library/agent.go

package library

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"time"

	"github.com/google/uuid"

	"pipu/api/internal/storageagent"
)

// Agent — клиент storage-agent (монтирование NFS, запись импортированных файлов).
type Agent interface {
	// Mount монтирует NFS; writable — источник принимает загрузки (ещё rw-точка, видимая только агенту).
	Mount(ctx context.Context, sourceID uuid.UUID, spec storageagent.NFSSpec, writable bool) (storageagent.MountResult, error)
	Unmount(ctx context.Context, sourceID uuid.UUID) error
	Test(ctx context.Context, spec storageagent.NFSSpec) (storageagent.MountResult, error)
	TestWrite(ctx context.Context, req storageagent.WriteTestRequest) (storageagent.WriteTestResult, error)
	WriteFile(ctx context.Context, req storageagent.WriteRequest) (storageagent.WriteResult, error)
}

// HTTPAgent ходит в storage-agent по внутренней сети с общим секретом.
type HTTPAgent struct {
	BaseURL string
	Token   string
	Client  *http.Client
}

// NewHTTPAgent создаёт клиента агента.
func NewHTTPAgent(baseURL, token string) *HTTPAgent {
	return &HTTPAgent{BaseURL: baseURL, Token: token, Client: &http.Client{Timeout: 40 * time.Second}}
}

// AgentError — агент отклонил запрос (например, недопустимые параметры NFS).
type AgentError struct {
	Status int
	Field  string
	Reason string
}

func (e *AgentError) Error() string {
	return fmt.Sprintf("storage-agent: %d %s %s", e.Status, e.Field, e.Reason)
}

func (a *HTTPAgent) do(ctx context.Context, method, path string, body any, out any) error {
	var rd *bytes.Reader
	if body != nil {
		raw, err := json.Marshal(body)
		if err != nil {
			return err
		}
		rd = bytes.NewReader(raw)
	} else {
		rd = bytes.NewReader(nil)
	}
	req, err := http.NewRequestWithContext(ctx, method, a.BaseURL+path, rd)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+a.Token)
	req.Header.Set("Content-Type", "application/json")
	res, err := a.Client.Do(req)
	if err != nil {
		return fmt.Errorf("storage-agent: %w", err)
	}
	defer func() { _ = res.Body.Close() }()
	if res.StatusCode >= 400 {
		var p struct {
			Detail string `json:"detail"`
			Errors []struct {
				Field string `json:"field"`
			} `json:"errors"`
		}
		_ = json.NewDecoder(res.Body).Decode(&p)
		ae := &AgentError{Status: res.StatusCode, Reason: p.Detail}
		if len(p.Errors) > 0 {
			ae.Field = p.Errors[0].Field
		}
		return ae
	}
	if out != nil {
		return json.NewDecoder(res.Body).Decode(out)
	}
	return nil
}

// Mount реализует Agent.
func (a *HTTPAgent) Mount(ctx context.Context, id uuid.UUID, spec storageagent.NFSSpec, writable bool) (storageagent.MountResult, error) {
	var res storageagent.MountResult
	err := a.do(ctx, http.MethodPost, "/internal/mounts",
		storageagent.MountRequest{SourceID: id.String(), NFSSpec: spec, Writable: writable}, &res)
	return res, err
}

// TestWrite реализует Agent.
func (a *HTTPAgent) TestWrite(ctx context.Context, req storageagent.WriteTestRequest) (storageagent.WriteTestResult, error) {
	var res storageagent.WriteTestResult
	err := a.do(ctx, http.MethodPost, "/internal/files/test", req, &res)
	return res, err
}

// WriteFile реализует Agent. Копирование большого файла по NFS может идти минуты — отдельный таймаут.
func (a *HTTPAgent) WriteFile(ctx context.Context, req storageagent.WriteRequest) (storageagent.WriteResult, error) {
	var res storageagent.WriteResult
	ctx, cancel := context.WithTimeout(ctx, 30*time.Minute)
	defer cancel()
	c := *a
	c.Client = &http.Client{}
	err := c.do(ctx, http.MethodPost, "/internal/files", req, &res)
	return res, err
}

// Unmount реализует Agent.
func (a *HTTPAgent) Unmount(ctx context.Context, id uuid.UUID) error {
	return a.do(ctx, http.MethodDelete, "/internal/mounts/"+id.String(), nil, nil)
}

// Test реализует Agent.
func (a *HTTPAgent) Test(ctx context.Context, spec storageagent.NFSSpec) (storageagent.MountResult, error) {
	var res storageagent.MountResult
	err := a.do(ctx, http.MethodPost, "/internal/mounts/test", spec, &res)
	return res, err
}