fakz9/Sipro-Marketplaces
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor marketplace product fetching and caching logic; update environment configuration for Redis and PostgreSQL

Browse Source
This commit is contained in:
admin
2025-07-07 19:20:05 +03:00
parent c7be7e2cea
commit 3976c7d0cf
22 changed files with 319 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
internal/marketplace/entities.go
View File

@@ -1,5 +1,17 @@
package marketplace
import (
"encoding/json"
"github.com/go-faster/errors"
"sipro-mps/pkg/utils"
)
const (
WildberriesBaseMarketplace = 0
OzonBaseMarketplace = 1
YandexBaseMarketplace = 2
)
type Marketplace struct {
ID int `json:"id"`
BaseMarketplace int `json:"base_marketplace"`
@@ -7,3 +19,39 @@ type Marketplace struct {
WarehouseID string `json:"warehouse_id"`
AuthDataJson []byte `json:"auth_data_json,omitempty"`
}
func (m *Marketplace) getIdentifierWildberries() (string, error) {
_, claims, err := utils.DecodeWildberriesJwt(m.AuthDataJson)
if err != nil {
return "", err
}
sellerId := claims["sid"].(string)
return sellerId, err
}
func (m *Marketplace) getIdentifierOzon() (string, error) {
if m.AuthDataJson == nil {
return "", nil
}
var authData map[string]interface{}
if err := json.Unmarshal(m.AuthDataJson, &authData); err != nil {
return "", errors.Wrap(err, "unmarshal AuthDataJson")
}
if identifier, ok := authData["clientId"].(string); ok {
return identifier, nil
}
return "", errors.New("identifier not found in AuthDataJson")
}
func (m *Marketplace) GetIdentifier() (string, error) {
switch m.BaseMarketplace {
case WildberriesBaseMarketplace:
return m.getIdentifierWildberries()
case OzonBaseMarketplace:
return m.getIdentifierOzon()
case YandexBaseMarketplace:
return "", errors.New("Yandex marketplace does not support identifier retrieval")
default:
return "", errors.Errorf("unknown base marketplace: %d", m.BaseMarketplace)
}
}

14
internal/ozon/products/adapter_grpc.go
View File

@@ -6,7 +6,6 @@ import (
"google.golang.org/grpc"
pb "sipro-mps/api/generated/v1/ozon/products"
"sipro-mps/internal/marketplace"
"sipro-mps/internal/ozon/products/mapping/generated"
)
type AdapterGRPC struct {
@@ -30,12 +29,10 @@ func RegisterAdapterGRPC(server *grpc.Server, marketplaceRepo marketplace.Reposi
func (g *AdapterGRPC) GetListOfProducts(req *pb.GetListOfProductsRequest, stream pb.ProductsService_GetListOfProductsServer) error {
ctx := stream.Context()
fmt.Printf("GetListOfProducts called with req: %+v\n", req.MarketplaceId)
converter := generated.ConverterImpl{}
resultChan := make(chan []OzonProduct)
resultChan := make(chan []pb.Product)
errChan := make(chan error)
go g.repo.StreamAllProducts(ctx, int(req.MarketplaceId), resultChan, errChan) // Запускаем в горутине
go g.repo.StreamAllProductsCache(ctx, int(req.MarketplaceId), resultChan, errChan) // Запускаем в горутине
for {
select {
case <-ctx.Done():
@@ -45,11 +42,10 @@ func (g *AdapterGRPC) GetListOfProducts(req *pb.GetListOfProductsRequest, stream
if !ok {
return nil // Завершаем, только если результат закрыт
}
protoProducts := lo.Map(products, func(product OzonProduct, _ int) *pb.Product {
return converter.ToProto(&product)
})
resp := &pb.GetListOfProductsResponse{
Products: protoProducts,
Products: lo.Map(products, func(p pb.Product, _ int) *pb.Product {
return &p
}),
}
if err := stream.Send(resp); err != nil {
fmt.Println("GetListOfProducts: error sending response:", err)

6
internal/ozon/products/entities.go
View File

@@ -1,5 +1,9 @@
package products
import "git.denco.store/fakz9/ozon-api-client/ozon"
import (
"git.denco.store/fakz9/ozon-api-client/ozon"
pb "sipro-mps/api/generated/v1/ozon/products"
)
type OzonProduct = ozon.ProductDetails
type PbProduct = pb.Product

1
internal/ozon/products/repository.go
View File

@@ -5,4 +5,5 @@ import "context"
type Repository interface {
GetAllProducts(ctx context.Context, marketplaceId int) ([]OzonProduct, error)
StreamAllProducts(ctx context.Context, marketplaceId int, resultChan chan<- []OzonProduct, errChan chan<- error)
StreamAllProductsCache(ctx context.Context, marketplaceId int, resultChan chan<- []PbProduct, errChan chan<- error)
}

102
internal/ozon/products/repository_api.go
View File

@@ -2,11 +2,16 @@ package products
import (
"context"
"encoding/json"
"fmt"
api "git.denco.store/fakz9/ozon-api-client/ozon"
"github.com/samber/lo"
"sipro-mps/internal/marketplace"
"sipro-mps/internal/ozon"
"sipro-mps/internal/ozon/products/mapping/generated"
"sipro-mps/internal/redis"
"sipro-mps/internal/tasks/client"
"sipro-mps/internal/tasks/types"
"sync"
)
@@ -81,7 +86,7 @@ func (a *apiRepository) GetAllProducts(ctx context.Context, marketplaceId int) (
if err != nil {
return nil, err
}
client, err := ozon.GetClientFromMarketplace(mp)
clientFromMarketplace, err := ozon.GetClientFromMarketplace(mp)
if err != nil {
return nil, err
}
@@ -89,15 +94,13 @@ func (a *apiRepository) GetAllProducts(ctx context.Context, marketplaceId int) (
productIdsChan := make(chan []int64)
producsChan := make(chan []OzonProduct)
errChan := make(chan error)
go fetchProductIds(ctx, client, productIdsChan, errChan)
go fetchProducts(ctx, client, productIdsChan, producsChan, errChan)
go fetchProductIds(ctx, clientFromMarketplace, productIdsChan, errChan)
go fetchProducts(ctx, clientFromMarketplace, productIdsChan, producsChan, errChan)
for products := range producsChan {
for _, product := range products {
fmt.Println(product.Name)
items = append(items, product)
}
}
fmt.Println(len(items))
return items, nil
}
@@ -107,12 +110,95 @@ func (a *apiRepository) StreamAllProducts(ctx context.Context, marketplaceId int
errChan <- err
return
}
client, err := ozon.GetClientFromMarketplace(mp)
clientFromMarketplace, err := ozon.GetClientFromMarketplace(mp)
if err != nil {
errChan <- err
return
}
productIdsChan := make(chan []int64)
go fetchProductIds(ctx, client, productIdsChan, errChan)
go fetchProducts(ctx, client, productIdsChan, resultChan, errChan)
go fetchProductIds(ctx, clientFromMarketplace, productIdsChan, errChan)
go fetchProducts(ctx, clientFromMarketplace, productIdsChan, resultChan, errChan)
}
func (a *apiRepository) StreamAllProductsCache(ctx context.Context, marketplaceId int, resultChan chan<- []PbProduct, errChan chan<- error) {
defer close(resultChan)
defer close(errChan)
mp, err := a.marketplaceRepository.GetMarketplaceByID(ctx, marketplaceId)
if err != nil {
errChan <- err
return
}
identifier, err := mp.GetIdentifier()
if err != nil {
errChan <- fmt.Errorf("getting marketplace identifier: %w", err)
return
}
r := *redis.Client
key := fmt.Sprintf("ozon:products:%s", identifier)
jsonString, err := r.Do(ctx, r.B().Get().Key(key).Build()).ToString()
if err == nil && jsonString != "null" {
var result []PbProduct
err = json.Unmarshal([]byte(jsonString), &result)
if err != nil {
errChan <- fmt.Errorf("unmarshalling products from cache: %w", err)
return
}
task, err := types.NewFetchProductsTask(types.TypeOzonFetchProducts, 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
}
innerResultChan := make(chan []OzonProduct)
innerErrChan := make(chan error)
go a.StreamAllProducts(ctx, marketplaceId, innerResultChan, innerErrChan)
converter := generated.ConverterImpl{}
var allProducts []PbProduct
defer func() {
value, err := json.Marshal(allProducts)
if err != nil {
errChan <- fmt.Errorf("marshalling products to JSON: %w", err)
return
}
err = r.Do(ctx, r.B().Set().Key(key).Value(string(value)).Build()).Error()
if err != nil {
errChan <- fmt.Errorf("setting products in cache: %w", err)
return
}
}()
for {
select {
case err, ok := <-innerErrChan:
if !ok {
return
}
if err != nil {
errChan <- fmt.Errorf("streaming products: %w", err)
return
}
case products, ok := <-innerResultChan:
if !ok && len(products) == 0 {
return
}
pbProducts := lo.Map(products, func(p OzonProduct, _ int) PbProduct {
return *converter.ToProto(&p)
})
allProducts = append(allProducts, pbProducts...)
resultChan <- pbProducts
if !ok {
return
}
}
}
}

66
internal/tasks/ozon/ozon.go Normal file
View File

@@ -0,0 +1,66 @@
package ozon
import (
"context"
"encoding/json"
"fmt"
"github.com/hibiken/asynq"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/samber/lo"
"sipro-mps/internal/marketplace"
"sipro-mps/internal/ozon/products/mapping/generated"
"sipro-mps/internal/redis"
"sipro-mps/internal/tasks/types"
"sipro-mps/internal/ozon/products"
)
type FetchProductsProcessor struct {
Dbpool *pgxpool.Pool
}
func (p *FetchProductsProcessor) ProcessTask(ctx context.Context, task *asynq.Task) error {
var payload types.FetchProductsTask
err := payload.Unmarshal(task)
if err != nil {
return asynq.SkipRetry
}
marketplaceRepo := marketplace.NewDBRepository(p.Dbpool)
marketplaceById, err := marketplaceRepo.GetMarketplaceByID(ctx, payload.MarketplaceId)
if err != nil {
return asynq.SkipRetry
}
identifier, err := marketplaceById.GetIdentifier()
if err != nil {
return asynq.SkipRetry
}
lockKey := fmt.Sprintf("ozon:products:marketplace:%s:lock", identifier)
locker := *redis.Locker
_, cancel, err := locker.TryWithContext(ctx, lockKey)
if err != nil {
fmt.Printf("Failed to acquire lock for marketplace %s: %v\n", identifier, err)
return asynq.SkipRetry
}
defer cancel()
ozonRepo := products.NewAPIRepository(marketplaceRepo)
productsRaw, err := ozonRepo.GetAllProducts(ctx, payload.MarketplaceId)
if err != nil {
return fmt.Errorf("failed to fetch products for marketplace %d: %w", payload.MarketplaceId, err)
}
converter := generated.ConverterImpl{}
productsProto := lo.Map(productsRaw, func(item products.OzonProduct, _ int) *products.PbProduct {
return converter.ToProto(&item)
})
productsJson, err := json.Marshal(productsProto)
if err != nil {
return asynq.SkipRetry
}
redisClient := *redis.Client
productsKey := fmt.Sprintf("ozon:products:%s", identifier)
err = redisClient.Do(ctx, redisClient.B().Set().Key(productsKey).Value(string(productsJson)).Build()).Error()
if err != nil {
return err
}
return nil
}

4
internal/tasks/server/server.go
View File

@@ -4,6 +4,7 @@ import (
"github.com/hibiken/asynq"
"github.com/jackc/pgx/v5/pgxpool"
"sipro-mps/internal/config"
"sipro-mps/internal/tasks/ozon"
"sipro-mps/internal/tasks/types"
"sipro-mps/internal/tasks/wb"
)
@@ -24,13 +25,14 @@ func (s *AsynqServer) createMux() *asynq.ServeMux {
// Register task handlers here
mux.Handle(types.TypeWbFetchProducts, &wb.FetchProductsProcessor{Dbpool: s.dbpool})
mux.Handle(types.TypeOzonFetchProducts, &ozon.FetchProductsProcessor{Dbpool: s.dbpool})
return mux
}
func (s *AsynqServer) Run() {
srv := asynq.NewServer(
asynq.RedisClientOpt{Addr: s.redisConfig.Addr, Password: s.redisConfig.Password},
asynq.Config{Concurrency: 10},
asynq.Config{},
)
mux := s.createMux()
if err := srv.Run(mux); err != nil {

3
internal/tasks/types/common.go
View File

@@ -1,5 +1,6 @@
package types
const (
TypeWbFetchProducts = "wb:fetch_products"
TypeWbFetchProducts = "wb:fetch_products"
TypeOzonFetchProducts = "ozon:fetch_products"
)

24
internal/tasks/types/types.go Normal file
View File

@@ -0,0 +1,24 @@
package types
import (
"encoding/json"
"github.com/hibiken/asynq"
"time"
)
type FetchProductsTask struct {
MarketplaceId int
}
func (t *FetchProductsTask) Unmarshal(task *asynq.Task) error {
return json.Unmarshal(task.Payload(), t)
}
func NewFetchProductsTask(taskType string, marketplaceId int) (*asynq.Task, error) {
payload, err := json.Marshal(&FetchProductsTask{MarketplaceId: marketplaceId})
if err != nil {
return nil, err
}
return asynq.NewTask(taskType, payload, asynq.MaxRetry(2), asynq.Timeout(20*time.Minute)), nil
}

19
internal/tasks/types/wb.go
View File

@@ -1,19 +0,0 @@
package types
import (
"encoding/json"
"github.com/hibiken/asynq"
"time"
)
type FetchProductsTask struct {
MarketplaceId int
}
func NewFetchProductsTask(marketplaceId int) (*asynq.Task, error) {
payload, err := json.Marshal(&FetchProductsTask{MarketplaceId: marketplaceId})
if err != nil {
return nil, err
}
return asynq.NewTask(TypeWbFetchProducts, payload, asynq.MaxRetry(2), asynq.Timeout(20*time.Minute)), nil
}

25
internal/tasks/wb/wb.go
View File

@@ -8,16 +8,15 @@ import (
"github.com/jackc/pgx/v5/pgxpool"
"github.com/samber/lo"
pb "sipro-mps/api/generated/v1/wb/products"
mp_repo "sipro-mps/internal/marketplace"
"sipro-mps/internal/marketplace"
"sipro-mps/internal/redis"
"sipro-mps/internal/tasks/types"
wb_products_repo "sipro-mps/internal/wb/products"
conv "sipro-mps/internal/wb/products/mapping/generated"
"sipro-mps/internal/wb/products"
"sipro-mps/internal/wb/products/mapping/generated"
)
type FetchProductsProcessor struct {
Dbpool *pgxpool.Pool
wbRepo wb_products_repo.Repository
}
func (p *FetchProductsProcessor) ProcessTask(ctx context.Context, task *asynq.Task) error {
@@ -25,11 +24,15 @@ func (p *FetchProductsProcessor) ProcessTask(ctx context.Context, task *asynq.Ta
if err := json.Unmarshal(task.Payload(), &payload); err != nil {
return asynq.SkipRetry
}
marketplaceRepo := mp_repo.NewDBRepository(p.Dbpool)
repo := wb_products_repo.NewAPIRepository(marketplaceRepo)
_, sellerId, err := repo.ParseMarketplace(ctx, payload.MarketplaceId)
marketplaceRepo := marketplace.NewDBRepository(p.Dbpool)
repo := products.NewAPIRepository(marketplaceRepo)
marketplaceById, err := marketplaceRepo.GetMarketplaceByID(ctx, payload.MarketplaceId)
if err != nil {
return fmt.Errorf("failed to parse marketplace %d: %w", payload.MarketplaceId, err)
return fmt.Errorf("failed to get marketplace by ID %d: %w", payload.MarketplaceId, err)
}
sellerId, err := marketplaceById.GetIdentifier()
if err != nil {
return fmt.Errorf("failed to get identifier for marketplace %d: %w", payload.MarketplaceId, err)
}
locker := *redis.Locker
_, cancel, err := locker.TryWithContext(ctx, fmt.Sprintf("wb:products:marketplace:%s:lock", sellerId))
@@ -47,12 +50,12 @@ func (p *FetchProductsProcessor) ProcessTask(ctx context.Context, task *asynq.Ta
if err != nil {
return fmt.Errorf("failed to fetch products for marketplace %d: %w", payload.MarketplaceId, err)
}
converter := conv.ConverterImpl{}
products := lo.Map(productsRaw, func(item wb_products_repo.WbProduct, _ int) *pb.Product {
converter := generated.ConverterImpl{}
productsProto := lo.Map(productsRaw, func(item products.WbProduct, _ int) *pb.Product {
return converter.ToProto(&item)
})
redisClient := *redis.Client
productsJson, err := json.Marshal(products)
productsJson, err := json.Marshal(productsProto)
if err != nil {
return fmt.Errorf("failed to marshal products: %w", err)
}

30
internal/wb/common.go
View File

@@ -1,43 +1,17 @@
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"
"sipro-mps/pkg/utils"
"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)
authData, claims, err := utils.DecodeWildberriesJwt(mp.AuthDataJson)
if err != nil {
return nil, fmt.Errorf("failed to decode Wildberries JWT")
}

5
internal/wb/products/repository_api.go
View File

@@ -15,6 +15,7 @@ import (
"sipro-mps/internal/wb"
"sipro-mps/internal/wb/products/mapping/generated"
wbapi "sipro-mps/pkg/api/wb/client"
"sipro-mps/pkg/utils"
)
const (
@@ -31,7 +32,7 @@ func (a apiRepository) ParseMarketplace(ctx context.Context, marketplaceId int)
if err != nil {
return nil, "", err
}
_, claims, err := wb.DecodeWildberriesJwt(marketplaceByID.AuthDataJson)
_, claims, err := utils.DecodeWildberriesJwt(marketplaceByID.AuthDataJson)
if err != nil {
return nil, "", err
}
@@ -113,7 +114,7 @@ func (a apiRepository) StreamAllProductsCache(ctx context.Context, marketplaceId
errChan <- fmt.Errorf("unmarshalling products from cache: %w", err)
return
}
task, err := types.NewFetchProductsTask(marketplaceId)
task, err := types.NewFetchProductsTask(types.TypeWbFetchProducts, marketplaceId)
if err != nil {
errChan <- fmt.Errorf("creating fetch products task: %w", err)
return

5
internal/wb/rate_limiter.go
View File

@@ -7,6 +7,7 @@ import (
"github.com/redis/rueidis"
"net/http"
"sipro-mps/internal/redis"
"sipro-mps/pkg/utils"
"time"
)
@@ -68,12 +69,12 @@ func (t *RateLimitTransport) RoundTrip(req *http.Request) (*http.Response, error
ctx := req.Context()
tokenString := req.Header.Get("Authorization")
authData := NewWbAuthData(tokenString)
authData := utils.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)
_, claims, err := utils.DecodeWildberriesJwt(authDataBytes)
if err != nil {
return nil, fmt.Errorf("failed to decode Wildberries JWT: %w", err)
}