Invalidation Strategies.

# TTL-Based Invalidation

class TTLCache:
    """
    Simple TTL-based caching.
    
    Pros: Simple, self-healing
    Cons: Stale for up to TTL duration
    """
    
    def __init__(self, redis_client, default_ttl: int = 300):
        self.redis = redis_client
        self.default_ttl = default_ttl
    
    async def get_product(self, product_id: str) -> dict:
        cache_key = f"product:{product_id}"
        
        # Check cache
        cached = await self.redis.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # Miss: load from database
        product = await self.db.fetch_one(
            "SELECT * FROM products WHERE id = $1",
            product_id
        )
        
        # Cache with TTL
        await self.redis.setex(
            cache_key,
            self.default_ttl,  # Expires in 300 seconds
            json.dumps(dict(product))
        )
        
        return dict(product)

# Event-Driven Invalidation

from dataclasses import dataclass
from typing import List, Optional
import json
from datetime import datetime, timezone


@dataclass
class ProductUpdatedEvent:
    """Event published when a product changes."""
    product_id: str
    changed_fields: List[str]
    timestamp: str
    source: str  # Which service made the change


class ProductService:
    """
    Product service that publishes events on changes.
    """
    
    def __init__(self, db, kafka_producer, cache):
        self.db = db
        self.kafka = kafka_producer
        self.cache = cache
    
    async def update_product(self, product_id: str, updates: dict) -> dict:
        """Update product and publish event for cache invalidation."""
        
        # Update database
        product = await self.db.fetch_one(
            """
            UPDATE products 
            SET name = COALESCE($2, name),
                price = COALESCE($3, price),
                updated_at = NOW()
            WHERE id = $1
            RETURNING *
            """,
            product_id, updates.get('name'), updates.get('price')
        )
        
        # Publish event for cache invalidation
        # (This is a dual write; use the Week 3 outbox for guaranteed delivery.)
        event = ProductUpdatedEvent(
            product_id=product_id,
            changed_fields=list(updates.keys()),
            timestamp=datetime.now(timezone.utc).isoformat(),
            source="product-service"
        )
        
        await self.kafka.send(
            "product-events",
            key=product_id.encode(),
            value=json.dumps(event.__dict__).encode()
        )
        
        return dict(product)


class CacheInvalidationConsumer:
    """
    Consumes product events and invalidates cache.
    
    Runs as a separate service/process.
    """
    
    def __init__(self, kafka_consumer, redis_client):
        self.kafka = kafka_consumer
        self.redis = redis_client
    
    async def run(self):
        """Main consumer loop."""
        async for message in self.kafka:
            try:
                event = json.loads(message.value.decode())
                await self.handle_event(event)
                await self.kafka.commit()
            except Exception as e:
                logger.error(f"Failed to process event: {e}")
                # Send to DLQ (Week 3, Day 4)
    
    async def handle_event(self, event: dict):
        """Handle a product event."""
        event_type = event.get('type', 'product.updated')
        product_id = event['product_id']
        
        if event_type in ('product.updated', 'product.deleted'):
            # Invalidate product cache
            await self.redis.delete(f"product:{product_id}")
            
            # Also invalidate related caches
            await self.redis.delete(f"product_page:{product_id}")
            await self.redis.delete(f"category_products:{event.get('category_id')}")
            
            logger.info(f"Invalidated cache for product {product_id}")
        
        elif event_type == 'product.price_changed':
            # Price change might affect more things
            await self.redis.delete(f"product:{product_id}")
            await self.redis.delete(f"deals_page")  # Might appear in deals
            # DEL takes literal key names, not patterns: use SCAN + DEL
            await delete_by_pattern("search_results:*")

# Products appear in multiple caches - invalidate all of them

CACHE_DEPENDENCIES = {
    "product": [
        "product:{id}",           # Direct product cache
        "product_page:{id}",      # Rendered product page
        "category:{category_id}", # Category listing
        "search:*",               # Search results (pattern)
    ],
    "inventory": [
        "product:{id}",           # Product includes stock info
        "product_page:{id}",
        "availability:{id}",
    ],
    "price": [
        "product:{id}",
        "product_page:{id}",
        "deals_page",             # If product is on sale
        "cart:{user_id}",         # User's cart might show price
    ]
}


async def invalidate_product(product_id: str, change_type: str, metadata: dict):
    """Invalidate all caches affected by a product change."""
    
    patterns = CACHE_DEPENDENCIES.get(change_type, ["product:{id}"])
    
    for pattern in patterns:
        # Replace placeholders
        key = pattern.format(
            id=product_id,
            category_id=metadata.get('category_id', '*'),
            user_id='*'  # Wildcard for user-specific caches
        )
        
        if '*' in key:
            # Pattern delete (use with caution - expensive!)
            await delete_by_pattern(key)
        else:
            await redis.delete(key)

# Versioned Key Invalidation

class VersionedCache:
    """
    Cache with versioned keys.
    
    Version is stored separately and incremented on updates.
    Old versions expire naturally via TTL.
    """
    
    def __init__(self, redis_client, default_ttl: int = 3600):
        self.redis = redis_client
        self.default_ttl = default_ttl
    
    async def get_version(self, entity: str, entity_id: str) -> int:
        """Get current version for an entity."""
        version_key = f"version:{entity}:{entity_id}"
        version = await self.redis.get(version_key)
        # A missing key reads as 0: INCR creates it at 0 and returns 1, so the first
        # increment must move to a new version (a default of 1 would never change the key)
        return int(version) if version else 0
    
    async def increment_version(self, entity: str, entity_id: str) -> int:
        """Increment version (effectively invalidates cache)."""
        version_key = f"version:{entity}:{entity_id}"
        new_version = await self.redis.incr(version_key)
        return new_version
    
    def _build_key(self, entity: str, entity_id: str, version: int) -> str:
        """Build versioned cache key."""
        return f"{entity}:{entity_id}:v{version}"
    
    async def get(self, entity: str, entity_id: str) -> Optional[dict]:
        """Get cached value using current version."""
        version = await self.get_version(entity, entity_id)
        cache_key = self._build_key(entity, entity_id, version)
        
        cached = await self.redis.get(cache_key)
        if cached:
            return json.loads(cached)
        return None
    
    async def set(self, entity: str, entity_id: str, value: dict) -> None:
        """Set cached value with current version."""
        version = await self.get_version(entity, entity_id)
        cache_key = self._build_key(entity, entity_id, version)
        
        await self.redis.setex(
            cache_key,
            self.default_ttl,
            json.dumps(value)
        )
    
    async def invalidate(self, entity: str, entity_id: str) -> int:
        """Invalidate by incrementing version."""
        return await self.increment_version(entity, entity_id)


# Usage
class ProductRepository:
    def __init__(self, db, cache: VersionedCache):
        self.db = db
        self.cache = cache
    
    async def get_product(self, product_id: str) -> dict:
        # Try cache with current version
        cached = await self.cache.get("product", product_id)
        if cached:
            return cached
        
        # Load from database
        product = await self.db.fetch_one(
            "SELECT * FROM products WHERE id = $1",
            product_id
        )
        
        # Cache with current version
        await self.cache.set("product", product_id, dict(product))
        return dict(product)
    
    async def update_product(self, product_id: str, data: dict) -> dict:
        # Update database
        product = await self.db.fetch_one(
            "UPDATE products SET ... WHERE id = $1 RETURNING *",
            product_id
        )
        
        # Invalidate by incrementing version
        # Next read will miss (new version key doesn't exist)
        await self.cache.invalidate("product", product_id)
        
        return dict(product)

# Invalidate ALL products at once (e.g., after bulk import)

class GlobalVersionCache:
    """Cache with global version for bulk invalidation."""
    
    async def get_global_version(self, entity: str) -> int:
        """Global version affects all entities of this type."""
        version = await self.redis.get(f"global_version:{entity}")
        # A missing key reads as 0: INCR creates it at 0 and returns 1, so the first
        # increment must move to a new version (a default of 1 would never change the key)
        return int(version) if version else 0
    
    async def increment_global_version(self, entity: str) -> int:
        """Invalidate ALL cached entities of this type."""
        return await self.redis.incr(f"global_version:{entity}")
    
    def _build_key(self, entity: str, entity_id: str, 
                   entity_version: int, global_version: int) -> str:
        """Key includes both entity and global version."""
        return f"{entity}:{entity_id}:v{entity_version}:g{global_version}"


# Usage: After bulk product import
await cache.increment_global_version("product")
# All product cache keys now miss (global version changed)

# Write-Invalidate (Recommended)

async def update_product_invalidate(product_id: str, data: dict):
    """Update database, then DELETE cache."""
    
    # 1. Update database
    product = await db.fetch_one(
        "UPDATE products SET price = $2 WHERE id = $1 RETURNING *",
        product_id, data['price']
    )
    
    # 2. Delete cache (invalidate)
    await cache.delete(f"product:{product_id}")
    
    # Next read will miss and load fresh data
    return dict(product)


# Write-Update (Use with caution)

async def update_product_update(product_id: str, data: dict):
    """Update database, then UPDATE cache."""
    
    # 1. Update database
    product = await db.fetch_one(
        "UPDATE products SET price = $2 WHERE id = $1 RETURNING *",
        product_id, data['price']
    )
    
    # 2. Update cache with new value
    # RISK: Race condition if concurrent updates!
    await cache.set(f"product:{product_id}", dict(product), ttl=300)
    
    return dict(product)

# Write-Update is safe when:
# 1. Single writer (no concurrent updates)
# 2. Using atomic operations
# 3. Data doesn't have complex dependencies

# Example: Incrementing a counter (atomic operation)
async def increment_view_count(product_id: str):
    """Atomic increment - safe for concurrent updates."""
    
    # Each increment is atomic on its own (the pair is not one transaction)
    await db.execute(
        "UPDATE products SET view_count = view_count + 1 WHERE id = $1",
        product_id
    )
    
    # INCR is atomic in Redis
    await cache.incr(f"product:{product_id}:views")

# Safety Net Pattern: Event-driven + TTL
from datetime import datetime, timezone


class SafetyNetCache:
    """
    Event-driven invalidation with TTL safety net.
    
    - Events provide fast invalidation
    - TTL ensures stale data eventually expires
    - Best of both worlds
    """
    
    def __init__(self, redis_client, kafka_producer, safety_ttl: int = 300):
        self.redis = redis_client
        self.kafka = kafka_producer
        self.safety_ttl = safety_ttl  # Maximum staleness
    
    async def set(self, key: str, value: any) -> None:
        """Cache with safety TTL."""
        await self.redis.setex(
            key,
            self.safety_ttl,  # Even without invalidation, expires in 5 min
            json.dumps(value)
        )
    
    async def invalidate(self, key: str) -> None:
        """Explicit invalidation via event."""
        await self.redis.delete(key)
    
    async def update_product(self, product_id: str, data: dict):
        """Update with event-driven invalidation."""
        
        # Update database
        product = await self.db.fetch_one(
            "UPDATE products SET ... WHERE id = $1 RETURNING *",
            product_id
        )
        
        # Publish invalidation event
        await self.kafka.send("cache-invalidation", {
            "type": "invalidate",
            "keys": [f"product:{product_id}"],
            "timestamp": datetime.now(timezone.utc).isoformat()
        })
        
        # Even if event is lost, TTL ensures max 5 min staleness
        return dict(product)


# Invalidation consumer
async def handle_invalidation_event(event: dict):
    """Process invalidation events."""
    for key in event.get("keys", []):
        await redis.delete(key)
        logger.info(f"Invalidated {key}")

# Multi-Strategy Cache Configuration

from dataclasses import dataclass
from enum import Enum
from typing import Optional, List


class InvalidationStrategy(Enum):
    TTL_ONLY = "ttl_only"
    EVENT_DRIVEN = "event_driven"
    VERSIONED = "versioned"
    WRITE_THROUGH = "write_through"


@dataclass
class CachePolicy:
    """Defines caching behavior for a data type."""
    strategy: InvalidationStrategy
    ttl: int
    event_topics: List[str] = None
    version_key: str = None


# Define policies for different data types
CACHE_POLICIES = {
    "product_details": CachePolicy(
        strategy=InvalidationStrategy.TTL_ONLY,
        ttl=3600  # 1 hour - rarely changes
    ),
    
    "product_price": CachePolicy(
        strategy=InvalidationStrategy.EVENT_DRIVEN,
        ttl=300,  # 5 min safety net
        event_topics=["price-changes", "flash-sales"]
    ),
    
    "product_inventory": CachePolicy(
        strategy=InvalidationStrategy.EVENT_DRIVEN,
        ttl=30,  # 30 sec safety net - changes frequently
        event_topics=["inventory-updates"]
    ),
    
    "user_session": CachePolicy(
        strategy=InvalidationStrategy.TTL_ONLY,
        ttl=1800  # 30 min sliding expiration
    ),
    
    "homepage_content": CachePolicy(
        strategy=InvalidationStrategy.VERSIONED,
        ttl=86400,  # 24 hours
        version_key="homepage_version"
    ),
}


class SmartCache:
    """
    Cache that applies different strategies based on data type.
    """
    
    def __init__(self, redis_client, policies: dict = CACHE_POLICIES):
        self.redis = redis_client
        self.policies = policies
    
    def get_policy(self, data_type: str) -> CachePolicy:
        """Get caching policy for data type."""
        return self.policies.get(data_type, CachePolicy(
            strategy=InvalidationStrategy.TTL_ONLY,
            ttl=300  # Default 5 min
        ))
    
    async def get(self, data_type: str, key: str) -> Optional[dict]:
        """Get from cache using appropriate strategy."""
        policy = self.get_policy(data_type)
        
        if policy.strategy == InvalidationStrategy.VERSIONED:
            return await self._get_versioned(data_type, key, policy)
        else:
            return await self._get_simple(data_type, key)
    
    async def set(self, data_type: str, key: str, value: dict) -> None:
        """Set in cache using appropriate strategy."""
        policy = self.get_policy(data_type)
        
        if policy.strategy == InvalidationStrategy.VERSIONED:
            await self._set_versioned(data_type, key, value, policy)
        else:
            await self._set_simple(data_type, key, value, policy.ttl)
    
    async def _get_simple(self, data_type: str, key: str) -> Optional[dict]:
        cache_key = f"{data_type}:{key}"
        cached = await self.redis.get(cache_key)
        return json.loads(cached) if cached else None
    
    async def _set_simple(self, data_type: str, key: str, value: dict, ttl: int):
        cache_key = f"{data_type}:{key}"
        await self.redis.setex(cache_key, ttl, json.dumps(value))
    
    async def _get_versioned(self, data_type: str, key: str, policy: CachePolicy):
        version = int(await self.redis.get(policy.version_key) or 0)
        cache_key = f"{data_type}:{key}:v{version}"
        cached = await self.redis.get(cache_key)
        return json.loads(cached) if cached else None
    
    async def _set_versioned(self, data_type: str, key: str, value: dict, policy: CachePolicy):
        version = int(await self.redis.get(policy.version_key) or 0)
        cache_key = f"{data_type}:{key}:v{version}"
        await self.redis.setex(cache_key, policy.ttl, json.dumps(value))

# Production Cache Invalidation System

import asyncio
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, List, Optional, Set
from enum import Enum

logger = logging.getLogger(__name__)


# =============================================================================
# Configuration
# =============================================================================

@dataclass
class InvalidationConfig:
    """Configuration for invalidation system."""
    
    # Event processing
    event_topic: str = "cache-invalidation"
    consumer_group: str = "cache-invalidator"
    batch_size: int = 100
    
    # Safety
    max_keys_per_invalidation: int = 1000
    enable_pattern_delete: bool = False  # Dangerous at scale
    
    # Monitoring
    enable_metrics: bool = True
    slow_invalidation_threshold_ms: float = 100


# =============================================================================
# Invalidation Events
# =============================================================================

class InvalidationType(Enum):
    DELETE = "delete"           # Delete specific keys
    PATTERN = "pattern"         # Delete keys matching pattern
    VERSION = "version"         # Increment version
    REFRESH = "refresh"         # Delete and pre-warm


@dataclass
class InvalidationEvent:
    """Event requesting cache invalidation."""
    type: InvalidationType
    keys: List[str] = field(default_factory=list)
    pattern: Optional[str] = None
    version_key: Optional[str] = None
    source: str = "unknown"
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    correlation_id: Optional[str] = None
    
    def to_dict(self) -> dict:
        return {
            "type": self.type.value,
            "keys": self.keys,
            "pattern": self.pattern,
            "version_key": self.version_key,
            "source": self.source,
            "timestamp": self.timestamp.isoformat(),
            "correlation_id": self.correlation_id
        }
    
    @classmethod
    def from_dict(cls, data: dict) -> 'InvalidationEvent':
        return cls(
            type=InvalidationType(data["type"]),
            keys=data.get("keys", []),
            pattern=data.get("pattern"),
            version_key=data.get("version_key"),
            source=data.get("source", "unknown"),
            timestamp=datetime.fromisoformat(data["timestamp"]),
            correlation_id=data.get("correlation_id")
        )


# =============================================================================
# Invalidation Publisher (Use in application services)
# =============================================================================

class InvalidationPublisher:
    """
    Publishes cache invalidation events.
    
    Use this in your application services when data changes.
    """
    
    def __init__(self, kafka_producer, config: InvalidationConfig):
        self.kafka = kafka_producer
        self.config = config
    
    async def invalidate_keys(
        self,
        keys: List[str],
        source: str = "application",
        correlation_id: str = None
    ) -> None:
        """Invalidate specific cache keys."""
        if len(keys) > self.config.max_keys_per_invalidation:
            # Split into batches
            for i in range(0, len(keys), self.config.max_keys_per_invalidation):
                batch = keys[i:i + self.config.max_keys_per_invalidation]
                await self._publish(InvalidationEvent(
                    type=InvalidationType.DELETE,
                    keys=batch,
                    source=source,
                    correlation_id=correlation_id
                ))
        else:
            await self._publish(InvalidationEvent(
                type=InvalidationType.DELETE,
                keys=keys,
                source=source,
                correlation_id=correlation_id
            ))
    
    async def invalidate_pattern(
        self,
        pattern: str,
        source: str = "application"
    ) -> None:
        """Invalidate keys matching pattern (use sparingly!)."""
        if not self.config.enable_pattern_delete:
            raise ValueError("Pattern delete is disabled in config")
        
        await self._publish(InvalidationEvent(
            type=InvalidationType.PATTERN,
            pattern=pattern,
            source=source
        ))
    
    async def increment_version(
        self,
        version_key: str,
        source: str = "application"
    ) -> None:
        """Increment a version key for versioned invalidation."""
        await self._publish(InvalidationEvent(
            type=InvalidationType.VERSION,
            version_key=version_key,
            source=source
        ))
    
    async def _publish(self, event: InvalidationEvent) -> None:
        """Publish event to Kafka."""
        await self.kafka.send(
            self.config.event_topic,
            value=json.dumps(event.to_dict()).encode()
        )
        logger.debug(f"Published invalidation event: {event.type.value}")


# =============================================================================
# Invalidation Consumer (Background service)
# =============================================================================

class InvalidationConsumer:
    """
    Consumes and processes cache invalidation events.
    
    Run this as a separate service or background worker.
    """
    
    def __init__(
        self,
        kafka_consumer,
        redis_client,
        config: InvalidationConfig,
        metrics_client = None
    ):
        self.kafka = kafka_consumer
        self.redis = redis_client
        self.config = config
        self.metrics = metrics_client
        
        self._running = False
    
    async def start(self):
        """Start consuming invalidation events."""
        self._running = True
        logger.info("Invalidation consumer started")
        
        while self._running:
            # Wait at most 100ms for up to batch_size events, so a lone
            # invalidation is never held back waiting for the batch to fill
            records = await self.kafka.getmany(
                timeout_ms=100, max_records=self.config.batch_size
            )
            
            batch = []
            for messages in records.values():
                for message in messages:
                    try:
                        batch.append(InvalidationEvent.from_dict(
                            json.loads(message.value.decode())
                        ))
                    except Exception as e:
                        logger.error(f"Failed to parse event: {e}")
                        # Malformed event - log (or DLQ) and continue
            
            if batch:
                await self._process_batch(batch)
                await self.kafka.commit()
    
    async def stop(self):
        """Stop the consumer."""
        self._running = False
        logger.info("Invalidation consumer stopped")
    
    async def _process_batch(self, events: List[InvalidationEvent]) -> None:
        """Process a batch of invalidation events."""
        start_time = datetime.now(timezone.utc)
        
        # Group by type for efficient processing
        delete_keys: Set[str] = set()
        patterns: List[str] = []
        version_keys: List[str] = []
        
        for event in events:
            if event.type == InvalidationType.DELETE:
                delete_keys.update(event.keys)
            elif event.type == InvalidationType.PATTERN:
                patterns.append(event.pattern)
            elif event.type == InvalidationType.VERSION:
                version_keys.append(event.version_key)
        
        # Process deletions in bulk
        if delete_keys:
            await self._delete_keys(list(delete_keys))
        
        # Process patterns (one at a time - expensive!)
        for pattern in patterns:
            await self._delete_pattern(pattern)
        
        # Process version increments
        for version_key in version_keys:
            await self._increment_version(version_key)
        
        # Record metrics
        duration_ms = (datetime.now(timezone.utc) - start_time).total_seconds() * 1000
        
        if self.metrics:
            self.metrics.timing("invalidation.batch_duration_ms", duration_ms)
            self.metrics.increment("invalidation.keys_deleted", len(delete_keys))
            self.metrics.increment("invalidation.events_processed", len(events))
        
        if duration_ms > self.config.slow_invalidation_threshold_ms:
            logger.warning(f"Slow invalidation batch: {duration_ms:.2f}ms")
    
    async def _delete_keys(self, keys: List[str]) -> int:
        """Delete multiple keys efficiently."""
        if not keys:
            return 0
        
        # Use pipeline for bulk delete
        pipe = self.redis.pipeline()
        for key in keys:
            pipe.delete(key)
        results = await pipe.execute()
        
        deleted = sum(1 for r in results if r)
        logger.debug(f"Deleted {deleted}/{len(keys)} keys")
        return deleted
    
    async def _delete_pattern(self, pattern: str) -> int:
        """Delete keys matching pattern (use sparingly!)."""
        # SCAN to find matching keys (non-blocking)
        deleted = 0
        cursor = 0
        
        while True:
            cursor, keys = await self.redis.scan(
                cursor=cursor,
                match=pattern,
                count=100
            )
            
            if keys:
                await self.redis.delete(*keys)
                deleted += len(keys)
            
            if cursor == 0:
                break
        
        logger.info(f"Pattern delete '{pattern}' removed {deleted} keys")
        return deleted
    
    async def _increment_version(self, version_key: str) -> int:
        """Increment a version key."""
        new_version = await self.redis.incr(version_key)
        logger.debug(f"Incremented {version_key} to {new_version}")
        return new_version


# =============================================================================
# Helper: Invalidation Decorator
# =============================================================================

def invalidate_cache(*keys: str, publisher: InvalidationPublisher = None):
    """
    Decorator that invalidates cache after function execution.
    
    Usage:
        @invalidate_cache("product:{product_id}", "category:{category_id}")
        async def update_product(product_id: str, category_id: str, data: dict):
            ...
    """
    def decorator(func):
        async def wrapper(*args, **kwargs):
            result = await func(*args, **kwargs)
            
            # Format keys with function arguments
            formatted_keys = []
            for key_template in keys:
                try:
                    formatted_key = key_template.format(**kwargs)
                    formatted_keys.append(formatted_key)
                except KeyError:
                    logger.warning(f"Could not format key template: {key_template}")
            
            if formatted_keys and publisher:
                await publisher.invalidate_keys(
                    formatted_keys,
                    source=func.__name__
                )
            
            return result
        return wrapper
    return decorator


# =============================================================================
# Usage Example
# =============================================================================

# In your product service:
class ProductService:
    def __init__(self, db, cache, invalidation: InvalidationPublisher):
        self.db = db
        self.cache = cache
        self.invalidation = invalidation
    
    async def update_product(self, product_id: str, data: dict) -> dict:
        # Get old product for category info
        old_product = await self.db.fetch_one(
            "SELECT category_id FROM products WHERE id = $1",
            product_id
        )
        
        # Update database
        product = await self.db.fetch_one(
            "UPDATE products SET ... WHERE id = $1 RETURNING *",
            product_id
        )
        
        # Invalidate all related caches
        await self.invalidation.invalidate_keys([
            f"product:{product_id}",
            f"product_page:{product_id}",
            f"category:{old_product['category_id']}",
        ], source="ProductService.update_product")
        
        return dict(product)
    
    async def bulk_import(self, products: List[dict]) -> int:
        """Import many products - use version increment."""
        # Import to database
        count = await self._bulk_insert(products)
        
        # Invalidate all product caches via version
        await self.invalidation.increment_version("product_cache_version")
        
        return count