Add Wildberries product fetching and rate limiting functionality

internal/wb/common.go

@@ -0,0 +1,59 @@
+package wb
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/golang-jwt/jwt/v5"
+	"net/http"
+	"sipro-mps/internal/marketplace"
+	wbclient "sipro-mps/pkg/api/wb/client"
+	"time"
+)
+
+type WbAuthData struct {
+	Token string `json:"token"`
+}
+
+func NewWbAuthData(token string) WbAuthData {
+	return WbAuthData{
+		Token: token,
+	}
+}
+
+func DecodeWildberriesJwt(token []byte) (WbAuthData, jwt.MapClaims, error) {
+	var authData WbAuthData
+	err := json.Unmarshal(token, &authData)
+	if err != nil {
+		return authData, nil, fmt.Errorf("failed to unmarshal JWT: %w", err)
+	}
+	claims := jwt.MapClaims{}
+	_, _, err = jwt.NewParser().ParseUnverified(authData.Token, claims)
+	if err != nil {
+		return authData, nil, fmt.Errorf("invalid JWT: %w", err)
+	}
+
+	return authData, claims, nil
+}
+
+func GetClientFromMarketplace(mp *marketplace.Marketplace) (*wbclient.Client, error) {
+
+	authData, claims, err := DecodeWildberriesJwt(mp.AuthDataJson)
+	if err != nil {
+		return nil, fmt.Errorf("failed to decode Wildberries JWT")
+	}
+	exp := claims["exp"].(float64)
+	// chec if token is expired, for now unix date
+	now := float64(time.Now().Unix())
+	if exp < now {
+		return nil, fmt.Errorf("token is expired")
+	}
+	securityHandler := NewWildberriesSecurityHandler(authData.Token)
+	httpClient := &http.Client{
+		Transport: NewRateLimitTransport(),
+	}
+	client, err := wbclient.NewClient("https://content-api.wildberries.ru", securityHandler, wbclient.WithClient(httpClient))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create Wildberries client: %w", err)
+	}
+	return client, nil
+}

internal/wb/products/adapter_grpc.go

@@ -0,0 +1,63 @@
+package products
+
+import (
+	"fmt"
+	"github.com/samber/lo"
+	"google.golang.org/grpc"
+	pb "sipro-mps/api/generated/v1/wb/products"
+	"sipro-mps/internal/marketplace"
+)
+
+type AdapterGRPC struct {
+	pb.UnimplementedProductsServiceServer
+	repo Repository
+}
+
+func NewAdapterGRPC(repo Repository) *AdapterGRPC {
+	return &AdapterGRPC{
+		repo: repo,
+	}
+}
+
+func RegisterAdapterGRPC(server *grpc.Server, marketplacesRepository marketplace.Repository) (*Repository, error) {
+	repo := NewAPIRepository(marketplacesRepository)
+	adapter := NewAdapterGRPC(repo)
+	pb.RegisterProductsServiceServer(server, adapter)
+	return &repo, nil
+}
+func (a *AdapterGRPC) GetProducts(req *pb.GetProductsRequest, stream pb.ProductsService_GetProductsServer) error {
+	ctx := stream.Context()
+	resultChan := make(chan []pb.Product)
+	errChan := make(chan error)
+	go a.repo.StreamAllProductsCache(ctx, int(req.MarketplaceId), resultChan, errChan)
+	for {
+		select {
+		case <-ctx.Done():
+			fmt.Println("context done")
+			return ctx.Err()
+		case products, ok := <-resultChan:
+			if !ok {
+				fmt.Println("result channel closed")
+				return nil
+			}
+			resp := &pb.GetProductsResponse{
+				Products: lo.Map(products, func(p pb.Product, _ int) *pb.Product {
+					return &p
+				}),
+			}
+			if err := stream.Send(resp); err != nil {
+				fmt.Println("error sending response", err)
+				return err
+			}
+		case err, ok := <-errChan:
+			if !ok {
+				return nil
+			}
+			if ok && err != nil {
+				fmt.Println("error in channel", err)
+				return err
+			}
+		}
+
+	}
+}

internal/wb/products/entities.go

@@ -0,0 +1,9 @@
+package products
+
+import (
+	pb "sipro-mps/api/generated/v1/wb/products"
+	"sipro-mps/pkg/api/wb/client"
+)
+
+type WbProduct = api.ContentV2GetCardsListPostOKCardsItem
+type PbProduct = pb.Product

internal/wb/products/mapping/converter.go

@@ -0,0 +1,28 @@
+package mapping
+
+import wbclient "sipro-mps/pkg/api/wb/client"
+
+// import (
+//
+//	proto "sipro-mps/api/generated/v1/wb/products"
+//	internal "sipro-mps/internal/wb/products"
+//	wbclient "sipro-mps/pkg/api/wb/client"
+//
+// )
+//
+// //go:generate go run github.com/jmattheis/goverter/cmd/goverter gen  -global "ignoreUnexported yes" .
+//
+// // goverter:converter
+// // goverter:extend  OptIntToInt64 OptStringToString
+//
+//	type Converter interface {
+//		// goverter:ignore state sizeCache unknownFields
+//
+//		ToProto(details *internal.WbProduct) *proto.Product
+//	}
+func OptIntToInt64(i wbclient.OptInt) int64 {
+	return int64(i.Value)
+}
+func OptStringToString(s wbclient.OptString) string {
+	return s.Value
+}

internal/wb/products/mapping/generated/generated.go

@@ -0,0 +1,40 @@
+// Code generated by github.com/jmattheis/goverter, DO NOT EDIT.
+//go:build !goverter
+
+package generated
+
+import (
+	products "sipro-mps/api/generated/v1/wb/products"
+	mapping "sipro-mps/internal/wb/products/mapping"
+	client "sipro-mps/pkg/api/wb/client"
+)
+
+type ConverterImpl struct{}
+
+func (c *ConverterImpl) ToProto(source *client.ContentV2GetCardsListPostOKCardsItem) *products.Product {
+	var pProductsProduct *products.Product
+	if source != nil {
+		var productsProduct products.Product
+		productsProduct.NmID = mapping.OptIntToInt64((*source).NmID)
+		productsProduct.SubjectID = mapping.OptIntToInt64((*source).SubjectID)
+		productsProduct.VendorCode = mapping.OptStringToString((*source).VendorCode)
+		if (*source).Sizes != nil {
+			productsProduct.Sizes = make([]*products.Product_Size, len((*source).Sizes))
+			for i := 0; i < len((*source).Sizes); i++ {
+				productsProduct.Sizes[i] = c.apiContentV2GetCardsListPostOKCardsItemSizesItemToPProductsProduct_Size((*source).Sizes[i])
+			}
+		}
+		pProductsProduct = &productsProduct
+	}
+	return pProductsProduct
+}
+func (c *ConverterImpl) apiContentV2GetCardsListPostOKCardsItemSizesItemToPProductsProduct_Size(source client.ContentV2GetCardsListPostOKCardsItemSizesItem) *products.Product_Size {
+	var productsProduct_Size products.Product_Size
+	if source.Skus != nil {
+		productsProduct_Size.Skus = make([]string, len(source.Skus))
+		for i := 0; i < len(source.Skus); i++ {
+			productsProduct_Size.Skus[i] = source.Skus[i]
+		}
+	}
+	return &productsProduct_Size
+}

internal/wb/products/repository.go

@@ -0,0 +1,13 @@
+package products
+
+import (
+	"context"
+	"sipro-mps/internal/marketplace"
+)
+
+type Repository interface {
+	GetAllProducts(ctx context.Context, marketplaceId int) ([]WbProduct, error)
+	StreamAllProducts(ctx context.Context, marketplaceId int, resultChan chan<- []WbProduct, errChan chan<- error)
+	StreamAllProductsCache(ctx context.Context, marketplaceId int, resultChan chan<- []PbProduct, errChan chan<- error)
+	ParseMarketplace(ctx context.Context, marketplaceId int) (*marketplace.Marketplace, string, error)
+}

internal/wb/products/repository_api.go

@@ -0,0 +1,224 @@
+package products
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/go-faster/errors"
+	"github.com/redis/rueidis"
+	"github.com/samber/lo"
+	pb "sipro-mps/api/generated/v1/wb/products"
+	"sipro-mps/internal/marketplace"
+	"sipro-mps/internal/redis"
+	"sipro-mps/internal/tasks/client"
+	"sipro-mps/internal/tasks/types"
+	"sipro-mps/internal/wb"
+	"sipro-mps/internal/wb/products/mapping/generated"
+	wbapi "sipro-mps/pkg/api/wb/client"
+)
+
+const (
+	maxRetries            = 5
+	maxProductsPerRequest = 100
+)
+
+type apiRepository struct {
+	marketplaceRepository marketplace.Repository
+}
+
+func (a apiRepository) ParseMarketplace(ctx context.Context, marketplaceId int) (*marketplace.Marketplace, string, error) {
+	marketplaceByID, err := a.marketplaceRepository.GetMarketplaceByID(ctx, marketplaceId)
+	if err != nil {
+		return nil, "", err
+	}
+	_, claims, err := wb.DecodeWildberriesJwt(marketplaceByID.AuthDataJson)
+	if err != nil {
+		return nil, "", err
+	}
+	sellerId := claims["sid"].(string)
+	return marketplaceByID, sellerId, nil
+}
+
+func fetchProducts(
+	ctx context.Context,
+	client *wbapi.Client,
+	sellerId string,
+	resultChan chan<- []WbProduct,
+	errChan chan<- error,
+) {
+	defer close(resultChan)
+	defer close(errChan)
+	request := wbapi.ContentV2GetCardsListPostReq{}
+	request.Settings.SetTo(wbapi.ContentV2GetCardsListPostReqSettings{})
+
+	request.Settings.Value.Cursor.SetTo(wbapi.ContentV2GetCardsListPostReqSettingsCursor{})
+	request.Settings.Value.Cursor.Value.Limit.SetTo(maxProductsPerRequest)
+
+	request.Settings.Value.Filter.SetTo(wbapi.ContentV2GetCardsListPostReqSettingsFilter{})
+	request.Settings.Value.Filter.Value.WithPhoto.SetTo(-1)
+	currentRetry := 0
+	for {
+		response, err := client.ContentV2GetCardsListPost(ctx, &request, wbapi.ContentV2GetCardsListPostParams{Locale: wbapi.NewOptString("ru")})
+		if err != nil {
+			currentRetry++
+			if currentRetry >= maxRetries {
+				errChan <- fmt.Errorf("fetching product IDs: %w", err)
+				return
+			}
+			continue
+		}
+		currentRetry = 0
+
+		switch r := response.(type) {
+		case *wbapi.ContentV2GetCardsListPostOKHeaders:
+			err = wb.SyncRateLimitRemaining(ctx, sellerId, r.XRatelimitRemaining.Value)
+			if err != nil {
+				errChan <- fmt.Errorf("syncing rate limit: %w", err)
+				return
+			}
+			resultChan <- r.Response.Cards
+			if r.Response.Cursor.Value.Total.Value < maxProductsPerRequest {
+				return
+			}
+			request.Settings.Value.Cursor.Value.UpdatedAt.SetTo(r.Response.Cursor.Value.UpdatedAt.Value)
+			request.Settings.Value.Cursor.Value.NmID.SetTo(r.Response.Cursor.Value.NmID.Value)
+		case *wbapi.R429Headers:
+			err = wb.SetRateLimitRetry(ctx, sellerId, r.XRatelimitRetry.Value, r.XRatelimitLimit.Value, r.XRatelimitReset.Value)
+			if err != nil {
+				errChan <- fmt.Errorf("setting rate limit retry: %w", err)
+				return
+			}
+		default:
+			errChan <- fmt.Errorf("unexpected response type: %T", r)
+			return
+		}
+	}
+}
+
+func (a apiRepository) StreamAllProductsCache(ctx context.Context, marketplaceId int, resultChan chan<- []pb.Product, errChan chan<- error) {
+	defer close(resultChan)
+	defer close(errChan)
+	_, sellerId, err := a.ParseMarketplace(ctx, marketplaceId)
+	if err != nil {
+		errChan <- err
+		return
+	}
+	c := *redis.Client
+	key := fmt.Sprintf("wb:products:%s", sellerId)
+	jsonString, err := c.Do(ctx, c.B().Get().Key(key).Build()).ToString()
+	if err == nil && jsonString != "null" {
+		var result []pb.Product
+		err = json.Unmarshal([]byte(jsonString), &result)
+		if err != nil {
+			errChan <- fmt.Errorf("unmarshalling products from cache: %w", err)
+			return
+		}
+		task, err := types.NewFetchProductsTask(marketplaceId)
+		if err != nil {
+			errChan <- fmt.Errorf("creating fetch products task: %w", err)
+			return
+		}
+		_, err = client.Client.Enqueue(task)
+		if err != nil {
+			errChan <- fmt.Errorf("enqueueing fetch products task: %w", err)
+			return
+		}
+
+		resultChan <- result
+		return
+	}
+	if !errors.As(err, &rueidis.Nil) && err != nil {
+		errChan <- fmt.Errorf("fetching products from cache: %w", err)
+		return
+	}
+	converter := generated.ConverterImpl{}
+
+	innerResultChan := make(chan []WbProduct)
+	innerErrChan := make(chan error)
+	go a.StreamAllProducts(ctx, marketplaceId, innerResultChan, innerErrChan)
+	var allProducts []pb.Product
+	defer func() {
+		jsonData, err := json.Marshal(allProducts)
+		if err != nil {
+			errChan <- fmt.Errorf("marshalling products to cache: %w", err)
+			return
+		}
+		err = c.Do(ctx, c.B().Set().Key(key).Value(string(jsonData)).Build()).Error()
+		if err != nil {
+			errChan <- fmt.Errorf("setting products to cache: %w", err)
+			return
+		}
+	}()
+	for {
+		select {
+		case err, ok := <-innerErrChan:
+			if !ok {
+				return
+			}
+			errChan <- fmt.Errorf("streaming products: %w", err)
+			return
+		case products, ok := <-innerResultChan:
+			if !ok {
+				return
+			}
+			pbProducts := lo.Map(products, func(p WbProduct, _ int) pb.Product {
+				return *converter.ToProto(&p)
+			})
+			allProducts = append(allProducts, pbProducts...)
+			resultChan <- pbProducts
+		}
+	}
+
+}
+func (a apiRepository) GetAllProducts(ctx context.Context, marketplaceId int) ([]WbProduct, error) {
+	marketplaceByID, sellerId, err := a.ParseMarketplace(ctx, marketplaceId)
+	fromMarketplace, err := wb.GetClientFromMarketplace(marketplaceByID)
+	if err != nil {
+		return nil, err
+	}
+	resultChan := make(chan []WbProduct)
+	errChan := make(chan error)
+	go fetchProducts(ctx, fromMarketplace, sellerId, resultChan, errChan)
+
+	var products []WbProduct
+	isWaiting := true
+	for {
+		if !isWaiting {
+			break
+		}
+		select {
+		case err, ok := <-errChan:
+			if !ok {
+				isWaiting = false
+				continue
+			}
+			return nil, err
+		case newProducts, ok := <-resultChan:
+			if !ok {
+				isWaiting = false
+			}
+			products = append(products, newProducts...)
+		}
+	}
+	return products, nil
+}
+
+func (a apiRepository) StreamAllProducts(ctx context.Context, marketplaceId int, resultChan chan<- []WbProduct, errChan chan<- error) {
+	marketplaceByID, sellerId, err := a.ParseMarketplace(ctx, marketplaceId)
+	if err != nil {
+		errChan <- err
+		return
+	}
+	fromMarketplace, err := wb.GetClientFromMarketplace(marketplaceByID)
+	if err != nil {
+		errChan <- err
+		return
+	}
+	go fetchProducts(ctx, fromMarketplace, sellerId, resultChan, errChan)
+}
+
+func NewAPIRepository(marketplaceRepository marketplace.Repository) Repository {
+	return &apiRepository{
+		marketplaceRepository: marketplaceRepository,
+	}
+}

internal/wb/rate_limiter.go

@@ -0,0 +1,170 @@
+package wb
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/redis/rueidis"
+	"net/http"
+	"sipro-mps/internal/redis"
+	"time"
+)
+
+const (
+	defaultBucketCapacity = 10            // max burst size
+	refillRate            = 100.0 / 60000 // 300 requests per minute → 1 token per 200ms
+	tokenTTLMillis        = 60000         // Redis key TTL: 60s
+)
+
+var tokenBucketScript = rueidis.NewLuaScript(`
+local key = KEYS[1]
+local now = tonumber(ARGV[1])
+local default_capacity = tonumber(ARGV[2])
+local refill_rate = tonumber(ARGV[3])
+local ttl = tonumber(ARGV[4])
+
+-- Retry lock
+local retry_key = key .. ":retry_until"
+local retry_until = tonumber(redis.call("GET", retry_key))
+if retry_until and now < retry_until then
+	return retry_until - now
+end
+
+-- Token Bucket
+local capacity_key = key .. ":capacity"
+local token_key = key .. ":tokens"
+local time_key = key .. ":last_refill"
+
+local capacity = tonumber(redis.call("GET", capacity_key)) or default_capacity
+local tokens = tonumber(redis.call("GET", token_key))
+local last_refill = tonumber(redis.call("GET", time_key))
+
+if tokens == nil then tokens = capacity end
+if last_refill == nil then last_refill = now end
+
+local elapsed = now - last_refill
+local refill = elapsed * refill_rate
+tokens = math.min(capacity, tokens + refill)
+last_refill = now
+
+if tokens >= 1 then
+	tokens = tokens - 1
+	redis.call("SET", token_key, tokens)
+	redis.call("SET", time_key, last_refill)
+	redis.call("PEXPIRE", token_key, ttl)
+	redis.call("PEXPIRE", time_key, ttl)
+	return 0
+else
+	local wait_time = math.ceil((1 - tokens) / refill_rate)
+	return wait_time
+end
+`)
+
+type RateLimitTransport struct {
+	http.RoundTripper
+}
+
+func (t *RateLimitTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	ctx := req.Context()
+
+	tokenString := req.Header.Get("Authorization")
+	authData := NewWbAuthData(tokenString)
+	authDataBytes, err := json.Marshal(authData)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal Wildberries auth data: %w", err)
+	}
+	_, claims, err := DecodeWildberriesJwt(authDataBytes)
+	if err != nil {
+		return nil, fmt.Errorf("failed to decode Wildberries JWT: %w", err)
+	}
+	sellerId := claims["sid"].(string)
+	if sellerId == "" {
+		return nil, fmt.Errorf("sellerId is required in JWT claims")
+	}
+	now := time.Now().UnixMilli()
+	client := *redis.Client
+
+	waitTime, err := tokenBucketScript.Exec(ctx, client, []string{sellerId}, []string{
+		fmt.Sprintf("%d", now),
+		fmt.Sprintf("%d", defaultBucketCapacity),
+		fmt.Sprintf("%f", refillRate),
+		fmt.Sprintf("%d", tokenTTLMillis),
+	}).ToInt64()
+
+	if err != nil {
+		return nil, fmt.Errorf("rate limit script error: %w", err)
+	}
+
+	if waitTime > 0 {
+		select {
+		case <-time.After(time.Duration(waitTime) * time.Millisecond):
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+
+	return t.RoundTripper.RoundTrip(req)
+}
+
+func SyncRateLimitRemaining(ctx context.Context, sellerId string, remaining int) error {
+	if sellerId == "" || remaining < 0 {
+		return fmt.Errorf("invalid sellerId or remaining")
+	}
+	now := time.Now().UnixMilli()
+	client := *redis.Client
+
+	cmds := []rueidis.Completed{
+		client.B().Set().Key(sellerId + ":capacity").Value(fmt.Sprintf("%d", defaultBucketCapacity)).Ex(time.Minute).Build(),
+		client.B().Set().Key(sellerId + ":tokens").Value(fmt.Sprintf("%d", remaining)).Ex(time.Minute).Build(),
+		client.B().Set().Key(sellerId + ":last_refill").Value(fmt.Sprintf("%d", now)).Ex(time.Minute).Build(),
+	}
+
+	results := client.DoMulti(ctx, cmds...)
+	for _, res := range results {
+		if res.Error() != nil {
+			return fmt.Errorf("failed to sync rate limit: %w", res.Error())
+		}
+	}
+	return nil
+}
+
+func SetRateLimitRetry(ctx context.Context, sellerId string, retrySeconds int, limit int, resetSeconds int) error {
+	if sellerId == "" {
+		return fmt.Errorf("sellerId is required")
+	}
+	now := time.Now()
+	retryUntil := now.Add(time.Duration(retrySeconds) * time.Second).UnixMilli()
+	client := *redis.Client
+
+	cmds := []rueidis.Completed{
+		client.B().Set().
+			Key(sellerId + ":retry_until").
+			Value(fmt.Sprintf("%d", retryUntil)).
+			Px(time.Duration(retrySeconds+5) * time.Second).Build(),
+	}
+
+	if limit > 0 {
+		cmds = append(cmds, client.B().Set().
+			Key(sellerId+":capacity").
+			Value(fmt.Sprintf("%d", limit)).
+			Ex(time.Hour).Build())
+	}
+
+	if resetSeconds > 0 {
+		resetAt := now.Add(time.Duration(resetSeconds) * time.Second)
+		fmt.Printf("Seller %s rate limit resets at %v (limit: %d)\n", sellerId, resetAt, limit)
+	}
+
+	results := client.DoMulti(ctx, cmds...)
+	for _, res := range results {
+		if res.Error() != nil {
+			return fmt.Errorf("failed to set retry info: %w", res.Error())
+		}
+	}
+
+	return nil
+}
+
+func NewRateLimitTransport() *RateLimitTransport {
+	return &RateLimitTransport{RoundTripper: http.DefaultTransport}
+}

internal/wb/security_handler.go

@@ -0,0 +1,22 @@
+package wb
+
+import (
+	"context"
+	wbclient "sipro-mps/pkg/api/wb/client"
+)
+
+type WildberriesSecurityHandler struct {
+	ApiKey string
+}
+
+func (sh WildberriesSecurityHandler) HeaderApiKey(ctx context.Context, operationName wbclient.OperationName, client *wbclient.Client) (wbclient.HeaderApiKey, error) {
+	return wbclient.HeaderApiKey{
+		APIKey: sh.ApiKey,
+		Roles:  nil,
+	}, nil
+}
+func NewWildberriesSecurityHandler(apiKey string) WildberriesSecurityHandler {
+	return WildberriesSecurityHandler{
+		ApiKey: apiKey,
+	}
+}

internal/wb/types.go

@@ -0,0 +1 @@
+package wb