69 lines
2.2 KiB
Go
69 lines
2.2 KiB
Go
package wb
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"github.com/hibiken/asynq"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
"github.com/samber/lo"
|
|
pb "sipro-mps/api/generated/v1/wb/products"
|
|
"sipro-mps/internal/marketplace"
|
|
"sipro-mps/internal/redis"
|
|
"sipro-mps/internal/tasks/types"
|
|
"sipro-mps/internal/wb/products"
|
|
"sipro-mps/internal/wb/products/mapping/generated"
|
|
)
|
|
|
|
type FetchProductsProcessor struct {
|
|
Dbpool *pgxpool.Pool
|
|
}
|
|
|
|
func (p *FetchProductsProcessor) ProcessTask(ctx context.Context, task *asynq.Task) error {
|
|
var payload types.FetchProductsTask
|
|
if err := json.Unmarshal(task.Payload(), &payload); err != nil {
|
|
return asynq.SkipRetry
|
|
}
|
|
marketplaceRepo := marketplace.NewDBRepository(p.Dbpool)
|
|
repo := products.NewAPIRepository(marketplaceRepo)
|
|
marketplaceById, err := marketplaceRepo.GetMarketplaceByID(ctx, payload.MarketplaceId)
|
|
if err != nil {
|
|
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))
|
|
if err != nil {
|
|
fmt.Printf("Failed to acquire lock for marketplace %s: %v\n", sellerId, err)
|
|
return asynq.SkipRetry
|
|
|
|
}
|
|
fmt.Println("Working on marketplace", payload.MarketplaceId, "with seller ID", sellerId)
|
|
|
|
defer cancel()
|
|
|
|
redisKey := fmt.Sprintf("wb:products:%s", sellerId)
|
|
productsRaw, err := repo.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.WbProduct, _ int) *pb.Product {
|
|
return converter.ToProto(&item)
|
|
})
|
|
redisClient := *redis.Client
|
|
productsJson, err := json.Marshal(productsProto)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to marshal products: %w", err)
|
|
}
|
|
err = redisClient.Do(ctx, redisClient.B().Set().Key(redisKey).Value(string(productsJson)).Build()).Error()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|