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 }