feat: implement locking mechanism for marketplace operations to prevent race conditions

internal/ozon/products/repository_api.go

@@ -149,6 +149,12 @@ func (a *apiRepository) StreamAllProductsCache(ctx context.Context, marketplaceI
 
 		return
 	}
+	locker := *redis.Locker
+	_, cancel, err := locker.TryWithContext(ctx, fmt.Sprintf("ozon:products:marketplace:%s:lock", key))
+	if err != nil {
+		return
+	}
+	defer cancel()
 
 	innerResultChan := make(chan []OzonProduct)
 	innerErrChan := make(chan error)
@@ -198,11 +204,24 @@ func (a *apiRepository) StreamProductAttributesCache(ctx context.Context, market
 		errChan <- err
 		return
 	}
+	identifier, err := mp.GetIdentifier()
+	if err != nil {
+		errChan <- fmt.Errorf("getting marketplace identifier: %w", err)
+		return
+	}
 	ozonClient, err := ozon.GetClientFromMarketplace(mp)
 	if err != nil {
 		errChan <- err
 		return
 	}
+	key := fmt.Sprintf("ozon:product_attributes:%s:lock", identifier)
+	locker := *redis.Locker
+	_, cancel, err := locker.WithContext(ctx, key)
+	if err != nil {
+		return
+	}
+	defer cancel()
+
 	converter := generated.ConverterImpl{}
 
 	for _, chunk := range lo.Chunk(productIds, 1000) {
@@ -235,10 +254,26 @@ func (a *apiRepository) DeleteProducts(ctx context.Context, marketplaceId int, i
 
 		return nil, err
 	}
+
+	identifier, err := mp.GetIdentifier()
+	if err != nil {
+		return nil, fmt.Errorf("getting marketplace identifier: %w", err)
+	}
+
 	ozonClient, err := ozon.GetClientFromMarketplace(mp)
 	if err != nil {
 		return nil, err
 	}
+
+	key := fmt.Sprintf("ozon:products_delete:%s:lock", identifier)
+	locker := *redis.Locker
+	_, cancel, err := locker.WithContext(ctx, key)
+	if err != nil {
+		return nil, err
+
+	}
+	defer cancel()
+
 	// Step 1: map the items into a slice
 	mapped := lo.Map(items, func(item *PbDeleteProductRequestItem, _ int) *PbDeleteProductResponseItem {
 		return &PbDeleteProductResponseItem{
@@ -293,6 +328,21 @@ func (a *apiRepository) CreateOrUpdateProducts(ctx context.Context, marketplaceI
 	if err != nil {
 		return nil, err
 	}
+	identifier, err := mp.GetIdentifier()
+	if err != nil {
+		return nil, fmt.Errorf("getting marketplace identifier: %w", err)
+
+	}
+
+	key := fmt.Sprintf("ozon:products_create_update:%s:lock", identifier)
+	locker := *redis.Locker
+	_, cancel, err := locker.WithContext(ctx, key)
+	if err != nil {
+		return nil, err
+
+	}
+	defer cancel()
+
 	converter := generated.ConverterImpl{}
 	pageSize := 100
 	result := make([]int64, (len(items)+pageSize-1)/pageSize)