golangLAKEHOUSE/internal/storeclient/client.go

// Package storeclient is the shared HTTP client to storaged's
// /storage/* surface. catalogd uses Get/Put/List for manifest
// persistence; vectord (G1+) uses the same shape for HNSW index
// persistence. Extracting it here keeps the dep direction clean —
// a service that talks to storaged depends on this package, not
// on another service's package.
package storeclient

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"
)

// ErrKeyNotFound mirrors storaged's not-found semantics on the
// caller side without exposing storaged's package types.
var ErrKeyNotFound = fmt.Errorf("storeclient: key not found")

// Client talks HTTP to a storaged service.
type Client struct {
	baseURL string
	hc      *http.Client
}

// listResponse mirrors storaged's GET /storage/list shape:
//
//	{"prefix":"_catalog/manifests/","objects":[{Key,Size,...}, ...]}
type listResponse struct {
	Prefix  string `json:"prefix"`
	Objects []struct {
		Key  string `json:"Key"`
		Size int64  `json:"Size"`
	} `json:"objects"`
}

// New builds a client against the given storaged base URL
// (e.g. "http://127.0.0.1:3211"). Timeout covers a single op only;
// callers that drive many ops sequentially get a fresh window per call.
func New(baseURL string) *Client {
	return &Client{
		baseURL: strings.TrimRight(baseURL, "/"),
		hc:      &http.Client{Timeout: 30 * time.Second},
	}
}

// Put writes raw bytes at key.
func (c *Client) Put(ctx context.Context, key string, body []byte) error {
	u := c.baseURL + "/storage/put/" + safeKey(key)
	req, err := http.NewRequestWithContext(ctx, http.MethodPut, u, bytes.NewReader(body))
	if err != nil {
		return fmt.Errorf("put req: %w", err)
	}
	req.ContentLength = int64(len(body))
	resp, err := c.hc.Do(req)
	if err != nil {
		return fmt.Errorf("put do: %w", err)
	}
	defer drainAndClose(resp.Body)
	if resp.StatusCode != http.StatusOK {
		preview, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
		return fmt.Errorf("put %s: status %d: %s", key, resp.StatusCode, string(preview))
	}
	return nil
}

// Get reads the bytes at key. Returns ErrKeyNotFound on 404.
func (c *Client) Get(ctx context.Context, key string) ([]byte, error) {
	u := c.baseURL + "/storage/get/" + safeKey(key)
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
	if err != nil {
		return nil, fmt.Errorf("get req: %w", err)
	}
	resp, err := c.hc.Do(req)
	if err != nil {
		return nil, fmt.Errorf("get do: %w", err)
	}
	defer drainAndClose(resp.Body)
	if resp.StatusCode == http.StatusNotFound {
		return nil, ErrKeyNotFound
	}
	if resp.StatusCode != http.StatusOK {
		preview, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
		return nil, fmt.Errorf("get %s: status %d: %s", key, resp.StatusCode, string(preview))
	}
	return io.ReadAll(resp.Body)
}

// Delete removes the key. Idempotent — a missing key is not an
// error (matches storaged's underlying S3 DeleteObject semantics).
func (c *Client) Delete(ctx context.Context, key string) error {
	u := c.baseURL + "/storage/delete/" + safeKey(key)
	req, err := http.NewRequestWithContext(ctx, http.MethodDelete, u, nil)
	if err != nil {
		return fmt.Errorf("delete req: %w", err)
	}
	resp, err := c.hc.Do(req)
	if err != nil {
		return fmt.Errorf("delete do: %w", err)
	}
	defer drainAndClose(resp.Body)
	if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK {
		preview, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
		return fmt.Errorf("delete %s: status %d: %s", key, resp.StatusCode, string(preview))
	}
	return nil
}

// List returns the keys under prefix. Caller-side filtering on
// suffix or other shape lives outside this package.
func (c *Client) List(ctx context.Context, prefix string) ([]string, error) {
	u := c.baseURL + "/storage/list?prefix=" + url.QueryEscape(prefix)
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
	if err != nil {
		return nil, fmt.Errorf("list req: %w", err)
	}
	resp, err := c.hc.Do(req)
	if err != nil {
		return nil, fmt.Errorf("list do: %w", err)
	}
	defer drainAndClose(resp.Body)
	if resp.StatusCode != http.StatusOK {
		preview, _ := io.ReadAll(io.LimitReader(resp.Body, 256))
		return nil, fmt.Errorf("list %s: status %d: %s", prefix, resp.StatusCode, string(preview))
	}
	var lr listResponse
	if err := json.NewDecoder(resp.Body).Decode(&lr); err != nil {
		return nil, fmt.Errorf("list decode: %w", err)
	}
	out := make([]string, 0, len(lr.Objects))
	for _, o := range lr.Objects {
		out = append(out, o.Key)
	}
	return out, nil
}

// safeKey URL-escapes path segments while preserving "/". storaged's
// chi `/storage/<verb>/*` routes accept literal slashes in the
// wildcard match, so we only escape the segments, not the separators.
func safeKey(key string) string {
	parts := strings.Split(key, "/")
	for i, p := range parts {
		parts[i] = url.PathEscape(p)
	}
	return strings.Join(parts, "/")
}

// drainAndClose reads any remaining body bytes (capped at 64 KiB) and
// closes the body — keeps HTTP/1.1 keep-alive pool reuse healthy on
// error paths.
func drainAndClose(body io.ReadCloser) {
	_, _ = io.Copy(io.Discard, io.LimitReader(body, 64<<10))
	_ = body.Close()
}