I will give you my documentation of the Python service I am currently working on. It is a worker service that will work in parallel with FastAPI app and will handle all the calculation of points that were claimed through this app. What I want you to do is to improve this service to make it more robust and prevent false claims or double claims by implementing checking and storing of idempotency key in DB for each type of claim. I also want you to improve it based on this criterias: robustness, architecture, highload, effectivmess, performance, pricision. Here is the full doc:

 ## Table of Contents  1. Project Structure 2. Configuration Management (config.py) 3. Logging Setup (logger.py) 4. Database Models (models.py) 5. Repository Class (repository.py) 6. Database Access Layer (db.py) 7. Redis Client Wrapper (redis_client.py) 8. Pulsar Client Wrapper (pulsar_client.py) 9. Lua Scripts for Redis (lua_scripts/cooldown.lua) 10. Processing Logic (processor.py) 11. Worker Execution Logic (worker.py) 12. Dockerization (Dockerfile) 13. Dependency Management (pyproject.toml) 14. Testing (test_processor.py) 15. Environment Variables (.env)  ---  ## 1. Project Structure  Maintain a clear and organized project structure to facilitate maintainability and scalability.

worker/
├── app/
│   ├── init.py
│   ├── config.py
│   ├── db.py
│   ├── logger.py
│   ├── pulsar_client.py
│   ├── redis_client.py
│   ├── processor.py
│   ├── repository.py
│   ├── worker.py
│   └── models.py
├── lua_scripts/
│   └── cooldown.lua
├── tests/
│   ├── init.py
│   ├── test_processor.py
│   └── ... # Additional tests
├── Dockerfile
├── pyproject.toml
├── poetry.lock
└── README.md


---

## 2. Configuration Management (config.py)

Use pydantic for robust and type-safe configuration management, incorporating configurations for PostgreSQL, Redis, and other components.

# app/config.py
from pydantic import BaseSettings, Field
from typing import Optional

class Settings(BaseSettings):
    # Pulsar Configuration
    pulsar_service_url: str = Field(..., env='PULSAR_SERVICE_URL')
    pulsar_consumer_topic: str = Field(..., env='PULSAR_CONSUMER_TOPIC')
    pulsar_consumer_subscription: str = Field(..., env='PULSAR_CONSUMER_SUBSCRIPTION')
    pulsar_producer_topic: str = Field(..., env='PULSAR_PRODUCER_TOPIC')
    
    # Database Configuration
    db_url: str = Field(..., env='DATABASE_URL')
    
    # Redis Configuration
    redis_host: str = Field('localhost', env='REDIS_HOST')
    redis_port: int = Field(6379, env='REDIS_PORT')
    redis_db: int = Field(0, env='REDIS_DB')
    redis_password: Optional[str] = Field(None, env='REDIS_PASSWORD')
    
    # Worker Configuration
    max_retries: int = Field(5, env='MAX_RETRIES')
    retry_delay_seconds: int = Field(5, env='RETRY_DELAY_SECONDS')
    max_concurrent_tasks: int = Field(100, env='MAX_CONCURRENT_TASKS')
    
    # Idempotency Configuration
    idempotency_enabled: bool = Field(True, env='IDEMPOTENCY_ENABLED')
    
    # Logging Configuration
    log_level: str = Field('INFO', env='LOG_LEVEL')
    
    class Config:
        env_file = '.env'
        env_file_encoding = 'utf-8'

settings = Settings()


Explanation:

- Comprehensive Settings: Includes configurations for Pulsar, PostgreSQL, Redis, worker behavior, and logging.
- Environment Variables: Utilizes .env files for configuration, enhancing security and flexibility.

---

## 3. Logging Setup (logger.py)

Implement structured and asynchronous logging to ensure non-blocking log operations, crucial for high-performance workers.

# app/logger.py
import logging
import sys
from .config import settings
from pythonjsonlogger import jsonlogger

def setup_logger():
    logger = logging.getLogger("worker")
    logger.setLevel(getattr(logging, settings.log_level.upper(), logging.INFO))
    
    handler = logging.StreamHandler(sys.stdout)
    formatter = jsonlogger.JsonFormatter(
        "%(asctime)s %(levelname)s %(name)s %(message)s"
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    
    # Avoid duplicate logs
    logger.propagate = False
    return logger

logger = setup_logger()


Explanation:

- Structured Logging: Uses pythonjsonlogger for JSON-formatted logs, enhancing log parsing and analysis.
- Performance: Ensures logging does not block the event loop, maintaining high throughput.

---

## 4. Database Models (models.py)

Define the SQLAlchemy models with appropriate indexing and include a model for processed messages to enforce idempotency.

# app/models.py
from sqlalchemy import Column, Integer, String, DateTime, Index, UniqueConstraint
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func
import uuid

Base = declarative_base()

class Profile(Base):
    tablename = 'profiles'

    profile_uuid = Column(String, primary_key=True, index=True)
    auth_user_uuid = Column(String, unique=True, index=True)
    direct_points = Column(Integer, nullable=False, default=0)
    ref_points = Column(Integer, nullable=False, default=0)
    idempotency_key = Column(String, nullable=False, default=lambda: str(uuid.uuid4()), unique=True)
    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

    table_args = (
        Index('idx_auth_user_uuid', 'auth_user_uuid'),
    )


Explanation:

- Profile Model: Represents the user's profile with both profile_uuid and auth_user_uuid.
- Idempotency Key: The idempotency_key acts as an ETag, ensuring that each claim is unique and preventing duplicate or malicious claims.
- Indexing: Ensures that queries on auth_user_uuid, profile_uuid, and idempotency_key are performant.

---

## 5. Repository Class (repository.py)

Implement a repository class to handle database operations, enhancing code readability and maintainability.

# app/repository.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy import update
from models import Profile
from .logger import logger
import uuid
from sqlalchemy.sql import func

class ProfileRepository:
    def init(self, session: AsyncSession):
        self.session = session

    async def get_profile(self, profile_uuid: str) -> Profile:
        stmt = select(Profile).where(Profile.profile_uuid == profile_uuid)
        result = await self.session.execute(stmt)
        profile = result.scalars().first()
        if not profile:
            logger.error(f"Profile not found: {profile_uuid}")
            raise ValueError("Profile not found.")
        return profile

    async def update_profile_idempotency_and_points(
        self,
        profile_uuid: str,
        direct_points: int,
        ref_points: int,
        new_idempotency_key: str
    ) -> Profile:
        stmt = (
            update(Profile)
            .where(Profile.profile_uuid == profile_uuid)
            .values(
                direct_points=Profile.direct_points + direct_points,
                ref_points=Profile.ref_points + ref_points,
                idempotency_key=new_idempotency_key,
                updated_at=func.now()
            )
            .returning(Profile)
        )
        result = await self.session.execute(stmt)
        profile = result.scalars().first()
        if not profile:
            logger.error(f"Profile not found during update: {profile_uuid}")
            raise ValueError("Profile not found during update.")
        return profile


Explanation:

- Encapsulation: The repository encapsulates all database interactions related to the Profile model.
- Readability: Enhances code readability by abstracting away raw SQL queries and updates.

---

## 6. Database Access Layer (db.py)

Use SQLAlchemy with asynchronous support for non-blocking database operations.

# app/db.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from .config import settings
from .logger import logger

DATABASE_URL = settings.db_url

# Create asynchronous engine
engine = create_async_engine(DATABASE_URL, echo=False, future=True)

# Create asynchronous sessionmaker
async_session = sessionmaker(
    engine, expire_on_commit=False, class_=AsyncSession
)

async def get_db_session() -> AsyncSession:
    async with async_session() as session:
        try:
            yield session
        except Exception as e:
            logger.error(f"Database session error: {e}")
            raise


Explanation:

- Asynchronous Operations: Utilizes SQLAlchemy’s async capabilities to ensure non-blocking database interactions.
- Session Management: Uses context managers to ensure sessions are properly handled, preventing resource leaks.

---

## 7. Redis Client Wrapper (redis_client.py)

Implement a Redis client capable of executing comprehensive Lua scripts for atomic operations.

# app/redis_client.py
import aioredis
from .config import settings
from .logger import logger
import os

class RedisClient:
    def init(self):
        self.redis = None

    async def connect(self):
        if not self.redis:
            try:
                self.redis = await aioredis.create_redis_pool(
                    (settings.redis_host, settings.redis_port),
                    db=settings.redis_db,
                    password=settings.redis_password,
                    maxsize=20
                )
                logger.info("Connected to Redis.")
            except Exception as e:
                logger.error(f"Failed to connect to Redis: {e}")
                raise

    async def close(self):
        if self.redis:
            self.redis.close()
            await self.redis.wait_closed()
            logger.info("Redis connection closed.")

    async def execute_lua_script(self, script_path: str, keys: list, args: list):
        try:
            with open(script_path, 'r') as file:
                lua_script = file.read()
            sha = await self.redis.script_load(lua_script)
            return await self.redis.evalsha(sha, keys, args)
        except Exception as e:
            logger.error(f"Failed to execute Lua script {script_path}: {e}")
            raise

redis_client = RedisClient()


Explanation:

- Comprehensive Lua Execution: Prepares and executes complex Lua scripts that handle multiple operations atomically.
- Connection Pooling: Maintains efficient Redis connections for high concurrency.
- Lifecycle Management: Provides methods to connect and close Redis connections gracefully.

---

## 8. Pulsar Client Wrapper (pulsar_client.py)

Enhance the Pulsar client for high throughput, ensuring efficient message consumption and production.

# app/pulsar_client.py
import pulsar
from pulsar import Client, Producer, Consumer
from .config import settings
from .logger import logger
import asyncio

class PulsarClientWrapper:
    _client: Client = None

    @classmethod
    def get_client(cls) -> Client:
        if not cls._client:
            try:
                cls._client = pulsar.Client(settings.pulsar_service_url)
                logger.info("Connected to Pulsar.")
            except Exception as e:
                logger.error(f"Failed to connect to Pulsar: {e}")
                raise
        return cls._client

    @classmethod
    def get_consumer(cls) -> Consumer:
        client = cls.get_client()
        try:
            consumer = client.subscribe(
                settings.pulsar_consumer_topic,
                settings.pulsar_consumer_subscription,
                consumer_type=pulsar.ConsumerType.Shared,
                ack_timeout_milliseconds=60000
            )
            logger.info(f"Subscribed to Pulsar topic {settings.pulsar_consumer_topic}.")
            return consumer
        except Exception as e:
            logger.error(f"Failed to subscribe to Pulsar topic: {e}")
            raise

    @classmethod
    def get_producer(cls) -> Producer:
        client = cls.get_client()
        try:
            producer = client.create_producer(settings.pulsar_producer_topic)
            logger.info(f"Producer created for Pulsar topic {settings.pulsar_producer_topic}.")
            return producer
        except Exception as e:
            logger.error(f"Failed to create Pulsar producer: {e}")
            raise

    @classmethod
    def close_client(cls):
        if cls._client:
            cls._client.close()
            logger.info("Pulsar client closed.")


Explanation:

- Shared Consumer Type: Utilizes Shared consumer type for load-balanced message consumption, enhancing throughput.
- Timeout Management: Ensures messages are acknowledged within a reasonable timeframe to prevent message re-delivery.
- Lifecycle Management: Provides methods to connect and close Pulsar clients gracefully.

---

## 9. Lua Scripts for Redis (lua_scripts/cooldown.lua)

Use Lua scripts to perform atomic and idempotent operations in Redis, ensuring consistency when updating cooldowns and caches.

-- lua_scripts/cooldown.lua
-- KEYS[1] - last_claim_key
-- ARGV[1] - current_time (as integer)
-- ARGV[2] - cooldown_seconds (as integer)

local last_claim = redis.call("GET", KEYS[1])
if last_claim then
    local elapsed = ARGV[1] - tonumber(last_claim)
    if elapsed < tonumber(ARGV[2]) then
        return 0 -- Cooldown active
    end
end

-- Set the last_claim_time with expiration
redis.call("SET", KEYS[1], ARGV[1], "EX", 86400) -- Set cooldown with TTL of 24 hours
return 1 -- Claim allowed


Explanation:

- Cooldown Enforcement: Checks if a cooldown is active (EXISTS). If yes, the claim is rejected; otherwise, sets the new cooldown.
- Atomicity: All operations are performed within the Lua script, ensuring atomic execution and eliminating race conditions.
- Return Values:
- 1: Successful claim processing.
- 0: Cooldown active; claim rejected.

---

## 10. Processing Logic (processor.py)

Implement the core processing logic, handling message validation, cooldown checks, database updates, cache synchronization, and message production.

# app/processor.py
import json
from typing import Any, Dict
from .logger import logger
from .db import get_db_session
from .redis_client import redis_client
from .pulsar_client import PulsarClientWrapper
from models import Profile
from repository import ProfileRepository
import asyncio
import os
import uuid
from .config import settings

class MessageProcessor:
    def init(self, message: pulsar.Message):
        self.message = message

    async def process(self):
        try:
            data = self._parse_message(self.message)
            if settings.idempotency_enabled:
                await self._check_idempotency(data)
            await self._handle_claim(data)
            self.message.ack()
            logger.info(f"Successfully processed message {self.message.message_id()}")
        except Exception as e:
            self.message.nack()
            logger.error(f"Failed to process message {self.message.message_id()}: {e}")
            raise

    def _parse_message(self, message: pulsar.Message) -> Dict[str, Any]:
        try:
            data = json.loads(message.data())
            logger.debug(f"Parsed message data: {data}")
            return data
        except json.JSONDecodeError as e:
            logger.error(f"JSON decoding failed: {e}")
            raise

    async def _check_idempotency(self, data: Dict[str, Any]):
        idempotency_key = data.get("idempotency_key")
        if not idempotency_key:
            logger.error("Missing idempotency_key in message.")
            raise ValueError("Missing idempotency_key in message.")

        profile_uuid = data.get("profile_uuid")
        if not profile_uuid:
            logger.error("Missing profile_uuid in message.")
            raise ValueError("Missing profile_uuid in message.")

        async for session in get_db_session():
            repository = ProfileRepository(session)
            profile = await repository.get_profile(profile_uuid)
            if profile.idempotency_key != idempotency_key:
                logger.warning(
                    f"Invalid idempotency_key for profile {profile_uuid}. "
                    f"Expected: {profile.idempotency_key}, Received: {idempotency_key}"
                )
                raise Exception("Invalid idempotency_key. Possible duplicate or invalid claim.")
            # Proceed; will set a new key

    async def _handle_claim(self, data: Dict[str, Any]):
        auth_user_uuid = data["auth_user_uuid"]
        profile_uuid = data["profile_uuid"]
        direct_points = data["direct_points"]
        ref_points = data["ref_points"]
        idempotency_key = data["idempotency_key"]

        # Determine claim type
        if direct_points > 0 and ref_points == 0:
            claim_type = "direct"
        elif ref_points > 0 and direct_points == 0:
            claim_type = "ref"
        elif direct_points > 0 and ref_points < 0:
            claim_type = "convert"
        else:
            logger.error("Invalid claim type based on points.")
            raise ValueError("Invalid claim type based on points.")

        # Calculate cooldown
        cooldown_seconds = await self._calculate_cooldown(auth_user_uuid)

        # Execute Lua script for atomic cooldown check and set
        script_path = os.path.join(os.path.dirname(file), '..', 'lua_scripts', 'cooldown.lua')
        last_claim_key = f"last_claim:{auth_user_uuid}:{claim_type}"
        current_time = int(asyncio.get_event_loop().time())
        result = await redis_client.execute_lua_script(script_path, [last_claim_key], [current_time, cooldown_seconds])
        if result == 0:
            logger.warning(f"Cooldown active for user {auth_user_uuid} on {claim_type} claim.")
            raise Exception("Cooldown active. Claim cannot be processed at this time.")

        # Generate new idempotency_key
        new_idempotency_key = str(uuid.uuid4())

        async for session in get_db_session():
            repository = ProfileRepository(session)
            # Update profile's points and set new idempotency_key
            profile = await repository.update_profile_idempotency_and_points(
                profile_uuid, direct_points, ref_points, new_idempotency_key
            )
            logger.debug(f"Profile {profile_uuid} updated with points and new idempotency_key.")

            # Update cache with new profile data (Write-Through Strategy)
            cache_key = f"profile:{auth_user_uuid}"
            updated_cache_value = {
                "profile_uuid": profile.profile_uuid,
                "auth_user_uuid": profile.auth_user_uuid,
                "direct_points": profile.direct_points,
                "ref_points": profile.ref_points,
                "idempotency_key": profile.idempotency_key,
                "updated_at": int(profile.updated_at.timestamp())
            }
            await redis_client.redis.hmset_dict(cache_key, updated_cache_value)
            logger.debug(f"Redis cache updated for user {auth_user_uuid}.")

        # Produce update message
        await self._produce_update(profile.auth_user_uuid, profile.profile_uuid, direct_points, ref_points)

    async def _calculate_cooldown(self, auth_user_uuid: str) -> int:
        streak_key = f"streak:{auth_user_uuid}"
        streak = await redis_client.redis.get(streak_key)
        streak_value = int(streak) if streak else 0
        cooldown = max(24 - streak_value, 8) * 3600  # Convert hours to seconds
        logger.debug(f"Calculated cooldown for user {auth_user_uuid}: {cooldown} seconds.")
        return cooldown

    async def _produce_update(self, auth_user_uuid: str, profile_uuid: str, direct_points: int, ref_points: int):
        producer = PulsarClientWrapper.get_producer()
        update_data = {
            "auth_user_uuid": auth_user_uuid,
            "profile_uuid": profile_uuid,
            "direct_points": direct_points,
            "ref_points": ref_points
        }
        try:
            message = json.dumps(update_data).encode('utf-8')
            producer.send(message)
            logger.debug(f"Produced update message for user {auth_user_uuid}.")
        except Exception as e:
            logger.error(f"Failed to produce update message for user {auth_user_uuid}: {e}")
            raise


Key Enhancements:

1. Idempotency via ETags:
- Validation: Compares the idempotency_key from the message with the one stored in the Profile.
- Generation: Upon successful claim processing, generates a new idempotency_key using uuid.uuid4().

2. Atomic Cooldown Checks with Lua Scripts:
- Cooldown Enforcement: Utilizes a Lua script (cooldown.lua) to atomically check if the cooldown period has passed and set the new last_claim_time.

3. Write-Through Cache Strategy:
- Immediate Cache Update: Updates the Redis cache immediately after the successful DB transaction, ensuring data safety and consistency.

4. Repository Pattern:
- Encapsulation: Encapsulates all DB operations within the ProfileRepository class, enhancing code readability and maintainability.

5. Error Handling and Logging:
- Comprehensive Exception Management: Ensures that any failure during processing results in appropriate actions (nack the message, log errors, and raise exceptions).
- Detailed Logging: Provides structured logs for easier monitoring and troubleshooting.

---

## 11. Worker Execution Logic (worker.py)

Refine the worker to support high concurrency, handle graceful shutdowns, execute atomic operations, and ensure robustness under high load.

# app/worker.py
import asyncio
import signal
from concurrent.futures import ThreadPoolExecutor
from pulsar import Consumer, Message, Timeout
from .pulsar_client import PulsarClientWrapper
from .processor import MessageProcessor
from .logger import logger
from .config import settings
from .redis_client import redis_client
from typing import Optional, Set

class Worker:
    def init(self):
        self.consumer: Optional[Consumer] = None
        self.shutdown_signal = asyncio.Event()
        self.executor = ThreadPoolExecutor(max_workers=10)
        self.active_tasks: Set[asyncio.Task] = set()
        self.semaphore = asyncio.Semaphore(settings.max_concurrent_tasks)

    async def start(self):
        await redis_client.connect()
        self.consumer = PulsarClientWrapper.get_consumer()
        loop = asyncio.get_event_loop()
        for sig in (signal.SIGTERM, signal.SIGINT):
            loop.add_signal_handler(sig, self.initiate_shutdown)
        logger.info("Worker started and listening for messages.")

        try:
            while not self.shutdown_signal.is_set():
                try:
                    message: Message = await loop.run_in_executor(self.executor, self.consumer.receive, 1)
                    if message:
                        await self.semaphore.acquire()
                        task = asyncio.create_task(self.handle_message(message))
                        self.active_tasks.add(task)
                        task.add_done_callback(self.active_tasks.discard)
                except Timeout:
                    continue  # No message received, continue polling
                except Exception as e:
                    logger.error(f"Error receiving message: {e}")
                    await asyncio.sleep(settings.retry_delay_seconds)
        finally:
            await self.shutdown()

    def initiate_shutdown(self):
        logger.info("Shutdown signal received.")
        self.shutdown_signal.set()

    async def handle_message(self, message: Message):
        try:
            processor = MessageProcessor(message)
            retry_count = 0
            while retry_count < settings.max_retries:
                try:
                    await processor.process()
                    break
                except Exception as e:
                    retry_count += 1
                    logger.error(f"Error processing message {message.message_id()}, attempt {retry_count}/{settings.max_retries}: {e}")
                    await asyncio.sleep(settings.retry_delay_seconds)
            else:
                logger.error(f"Exceeded max retries for message {message.message_id()}. Giving up.")
        finally:
            self.semaphore.release()

    async def shutdown(self):
        logger.info("Shutting down worker gracefully...")
        self.shutdown_signal.set()
        # Wait for all active tasks to complete
        if self.active_tasks:
            await asyncio.wait(self.active_tasks)
        # Close Redis connection
        await redis_client.close()
        # Close Pulsar client
        if self.consumer:
            PulsarClientWrapper.close_client()
        self.executor.shutdown(wait=True)
        logger.info("Worker shutdown complete.")

def main():
    worker = Worker()
    try:
        asyncio.run(worker.start())
    except Exception as e:
        logger.error(f"Worker encountered an error: {e}")
    finally:
        logger.info("Worker has stopped.")

if name == "main":
    main()


Key Enhancements:

1. Concurrency Control:
- asyncio.Semaphore: Manages the number of concurrent tasks based on max_concurrent_tasks to prevent resource exhaustion.
- Task Management: Maintains a set of active tasks to ensure all are completed before shutdown.

2. Graceful Shutdown:
- Signal Handling: Listens for SIGTERM and SIGINT to initiate a graceful shutdown.
- Shutdown Sequence: Waits for all active tasks to finish, closes Redis and Pulsar connections, and shuts down the executor.

3. Error Resilience:
- Retry Mechanism: Retries message processing up to max_retries times with delays in between.
- Logging: Logs errors and warnings to aid in monitoring and troubleshooting.

4. Optimized Concurrency:
- Semaphore Usage: Ensures that the number of concurrent tasks does not exceed max_concurrent_tasks, managing system resources effectively.

---

## 10. Lua Scripts for Redis (lua_scripts/cooldown.lua)

Use Lua scripts to perform atomic and idempotent operations in Redis, ensuring consistency when updating cooldowns and caches.

-- lua_scripts/cooldown.lua
-- KEYS[1] - last_claim_key
-- ARGV[1] - current_time (as integer)
-- ARGV[2] - cooldown_seconds (as integer)

local last_claim = redis.call("GET", KEYS[1])
if last_claim then
    local elapsed = ARGV[1] - tonumber(last_claim)
    if elapsed < tonumber(ARGV[2]) then
        return 0 -- Cooldown active
    end
end

-- Set the last_claim_time with expiration
redis.call("SET", KEYS[1], ARGV[1], "EX", 86400) -- Set cooldown with TTL of 24 hours
return 1 -- Claim allowed


Explanation:

- Cooldown Enforcement: Checks if a cooldown is active (EXISTS). If yes, the claim is rejected; otherwise, sets the new cooldown.
- Atomicity: All operations are performed within the Lua script, ensuring atomic execution and eliminating race conditions.
- Return Values:
- 1: Successful claim processing.
- 0: Cooldown active; claim rejected.

---

## 12. Dockerization (Dockerfile)

Containerize the worker for consistent and reproducible deployments, optimizing for production environments.

# Dockerfile
FROM python:3.10-slim

# Set environment variables
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

# Set work directory
WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y build-essential libpq-dev gcc && rm -rf /var/lib/apt/lists/*

# Install Poetry
COPY pyproject.toml poetry.lock ./
RUN pip install --upgrade pip
RUN pip install poetry
RUN poetry config virtualenvs.create false
RUN poetry install --no-dev --no-interaction --no-ansi

# Copy application code
COPY app /app/app
COPY lua_scripts /app/lua_scripts

# Set environment variables for Poetry
ENV PYTHONPATH=/app

# Run the worker
CMD ["python", "app/worker.py"]


Explanation:

- Lean Base Image: Uses python:3.10-slim to reduce image size and surface area for vulnerabilities.
- System Dependencies: Installs necessary build tools and libraries for dependencies like aioredis and SQLAlchemy.
- Poetry Installation: Ensures reproducible builds by installing dependencies via Poetry.
- Inclusion of Lua Scripts: Copies the lua_scripts directory into the container, ensuring scripts are available during execution.
- Environment Variables: Sets PYTHONPATH to include the application directory, facilitating module imports.

---

## 13. Dependency Management (pyproject.toml)

Manage dependencies with Poetry, ensuring reproducibility and ease of maintenance.

# pyproject.toml
[tool.poetry]
name = "worker"
version = "0.1.0"
description = "Async Worker for Points Recalculation"
authors = ["Your Name <you@example.com>"]
license = "MIT"
repository = "https://github.com/yourusername/worker"

[tool.poetry.dependencies]
python = "^3.10"
pulsar-client = "^2.11.0"
SQLAlchemy = { version = "^1.4", extras = ["asyncio"] }
asyncpg = "^0.25.0"  # Async PostgreSQL driver
pydantic = "^1.10.2"
aioredis = "^2.0.1"
python-json-logger = "^2.0.4"

[tool.poetry.dev-dependencies]
pytest = "^7.3.1"
pytest-asyncio = "^0.21.1"
black = "^23.3.0"
isort = "^5.12.0"
flake8 = "^5.0.4"
mypy = "^0.991"  # Type checking
types-pulsar-client = "^2.11.0"  # Type definitions for pulsar-client
types-json = "^2.0.9"  # Type definitions for JSON

[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"


Explanation:

- Essential Dependencies: Includes pulsar-client, SQLAlchemy, aioredis, asyncpg, and other necessary libraries for asynchronous operations.
- Development Tools: Incorporates pytest, black, isort, flake8, and mypy for testing and code quality assurance.
- Type Checking: Added mypy and type definitions to enforce type safety and improve code quality.

---

## 14. Testing (test_processor.py)

Implement unit tests to ensure reliability and correctness of the processing logic, covering various scenarios including idempotency, cooldown enforcement, and claim types.

# tests/test_processor.py
import pytest
from unittest.mock import AsyncMock, patch
from app.processor import MessageProcessor
from pulsar import Message
import json

@pytest.fixture
def sample_message():
    class MockMessage(Message):
        def init(self, data):
            self._data = data.encode('utf-8')
            self._message_id = "test-message-id"

        def data(self):
            return self._data

        def message_id(self):
            return self._message_id

        def ack(self):
            pass

        def nack(self):
            pass

    return MockMessage

@pytest.mark.asyncio
@patch('app.processor.PulsarClientWrapper.get_producer')
@patch('app.processor.get_db_session')
@patch('app.processor.redis_client')
async def test_process_direct_claim(mock_redis, mock_db_session, mock_get_producer, sample_message):
    # Setup mocks
    mock_db = AsyncMock()
    mock_db.execute.return_value.scalars.return_value.first.return_value = AsyncMock(
        profile_uuid="profile-uuid-456",
        auth_user_uuid="auth-uuid-123",
        direct_points=10,
        ref_points=0,
        idempotency_key="unique-idempotency-key-1",
        updated_at=AsyncMock(timestamp=lambda: 1609459200)  # 2021-01-01
    )
    mock_db_session.return_value = mock_db
    mock_producer = AsyncMock()
    mock_get_producer.return_value = mock_producer
    mock_redis.execute_lua_script.return_value = 1  # Allow claim

    # Create message data
    data = {
        "auth_user_uuid": "auth-uuid-123",
        "profile_uuid": "profile-uuid-456",
        "direct_points": 10,
        "ref_points": 0,
        "idempotency_key": "unique-idempotency-key-1"
    }
    message = sample_message(json.dumps(data))

    # Initialize processor
    processor = MessageProcessor(message)

    # Execute process
    await processor.process()

    # Assertions
    message.ack.assert_called_once()
    mock_db.execute.assert_awaited()
    mock_db.commit.assert_awaited_once()
    mock_producer.send.assert_called_once()

@pytest.mark.asyncio
@patch('app.processor.PulsarClientWrapper.get_producer')
@patch('app.processor.get_db_session')
@patch('app.processor.redis_client')
async def test_process_ref_claim(mock_redis, mock_db_session, mock_get_producer, sample_message):
    # Setup mocks
    mock_db = AsyncMock()
    mock_db.execute.return_value.scalars.return_value.first.return_value = AsyncMock(
        profile_uuid="profile-uuid-012",
        auth_user_uuid="auth-uuid-789",
        direct_points=0,
        ref_points=15,
        idempotency_key="unique-idempotency-key-2",
        updated_at=AsyncMock(timestamp=lambda: 1609459200)  # 2021-01-01
    )
    mock_db_session.return_value = mock_db
    mock_producer = AsyncMock()
    mock_get_producer.return_value = mock_producer
    mock_redis.execute_lua_script.return_value = 1  # Allow claim

    # Create message data
    data = {
        "auth_user_uuid": "auth-uuid-789",
        "profile_uuid": "profile-uuid-012",
        "direct_points": 0,
        "ref_points": 15,
        "idempotency_key": "unique-idempotency-key-2"
    }
    message = sample_message(json.dumps(data))

    # Initialize processor
    processor = MessageProcessor(message)

    # Execute process
    await processor.process()

    # Assertions
    message.ack.assert_called_once()
    mock_db.execute.assert_awaited()
    mock_db.commit.assert_awaited_once()
    mock_producer.send.assert_called_once()

@pytest.mark.asyncio
@patch('app.processor.PulsarClientWrapper.get_producer')
@patch('app.processor.get_db_session')
@patch('app.processor.redis_client')
async def test_process_convert_claim(mock_redis, mock_db_session, mock_get_producer, sample_message):
    # Setup mocks
    mock_db = AsyncMock()
    mock_db.execute.return_value.scalars.return_value.first.return_value = AsyncMock(
        profile_uuid="profile-uuid-678",
        auth_user_uuid="auth-uuid-345",
        direct_points=30,  # 20 added
        ref_points=-10,    # -10 added
        idempotency_key="unique-idempotency-key-3",
        updated_at=AsyncMock(timestamp=lambda: 1609459200)  # 2021-01-01
    )
    mock_db_session.return_value = mock_db
    mock_producer = AsyncMock()
    mock_get_producer.return_value = mock_producer
    mock_redis.execute_lua_script.return_value = 1  # Allow claim

    # Create message data
    data = {
        "auth_user_uuid": "auth-uuid-345",
        "profile_uuid": "profile-uuid-678",
        "direct_points": 20,
        "ref_points": -10,
        "idempotency_key": "unique-idempotency-key-3"
    }
    message = sample_message(json.dumps(data))

    # Initialize processor
    processor = MessageProcessor(message)

    # Execute process
    await processor.process()

    # Assertions
    message.ack.assert_called_once()
    mock_db.execute.assert_awaited()
    mock_db.commit.assert_awaited_once()
    mock_producer.send.assert_called_once()

@pytest.mark.asyncio
@patch('app.processor.PulsarClientWrapper.get_producer')
@patch('app.processor.get_db_session')
@patch('app.processor.redis_client')
async def test_process_idempotency(mock_redis, mock_db_session, mock_get_producer, sample_message):
    # Setup mocks
    mock_db = AsyncMock()
    # Simulate the profile with different idempotency_key
    profile = AsyncMock()
    profile.profile_uuid = "profile-uuid-888"
    profile.auth_user_uuid = "auth-uuid-999"
    profile.idempotency_key = "unique-idempotency-key-4"
    mock_db.execute.return_value.scalars.return_value.first.return_value = profile
    mock_db_session.return_value = mock_db
    mock_producer = AsyncMock()
    mock_get_producer.return_value = mock_producer

    # Create message data with a different idempotency_key
    data = {
        "auth_user_uuid": "auth-uuid-999",
        "profile_uuid": "profile-uuid-888",
        "direct_points": 10,
        "ref_points": 0,
        "idempotency_key": "invalid-idempotency-key"
    }
    message = sample_message(json.dumps(data))

    # Initialize processor
    processor = MessageProcessor(message)

    # Execute process and expect exception
    with pytest.raises(Exception) as excinfo:
        await processor.process()

    # Assertions
    assert "Invalid idempotency_key" in str(excinfo.value)
    message.nack.assert_called_once()
    mock_db.execute.assert_awaited_once()
    mock_get_producer.return_value.send.assert_not_called()

@pytest.mark.asyncio
@patch('app.processor.PulsarClientWrapper.get_producer')
@patch('app.processor.get_db_session')
@patch('app.processor.redis_client')
async def test_cooldown_active(mock_redis, mock_db_session, mock_get_producer, sample_message):
    # Setup mocks
    mock_db = AsyncMock()
    profile = AsyncMock()
    profile.profile_uuid = "profile-uuid-888"
    profile.auth_user_uuid = "auth-uuid-999"
    profile.idempotency_key = "unique-idempotency-key-5"
    mock_db.execute.return_value.scalars.return_value.first.return_value = profile
    mock_db_session.return_value = mock_db
    mock_producer = AsyncMock()
    mock_get_producer.return_value = mock_producer

    # Simulate cooldown active (Lua script returns 0)
    mock_redis.execute_lua_script.return_value = 0  # Cooldown active

    # Create message data
    data = {
        "auth_user_uuid": "auth-uuid-999",
        "profile_uuid": "profile-uuid-888",
        "direct_points": 10,
        "ref_points": 0,
        "idempotency_key": "unique-idempotency-key-5"
    }
    message = sample_message(json.dumps(data))

    # Initialize processor
    processor = MessageProcessor(message)

    # Execute process and expect exception
    with pytest.raises(Exception) as excinfo:
        await processor.process()

    # Assertions
    assert "Cooldown active" in str(excinfo.value)
    message.nack.assert_called_once()
    mock_db.execute.assert_awaited_once()
    mock_producer.send.assert_not_called()

@pytest.mark.asyncio
@patch('app.processor.PulsarClientWrapper.get_producer')
@patch('app.processor.get_db_session')
@patch('app.processor.redis_client')
async def test_cache_initialization(mock_redis, mock_db_session, mock_get_producer, sample_message):
    # Setup mocks
    mock_db = AsyncMock()
    profile = AsyncMock()
    profile.profile_uuid = "profile-uuid-666"
    profile.auth_user_uuid = "auth-uuid-555"
    profile.direct_points = 5
    profile.ref_points = 0
    profile.idempotency_key = "unique-idempotency-key-6"
    profile.updated_at = AsyncMock(timestamp=lambda: 1609459200)  # 2021-01-01
    mock_db.execute.return_value.scalars.return_value.first.return_value = profile
    mock_db_session.return_value = mock_db
    mock_producer = AsyncMock()
    mock_get_producer.return_value = mock_producer
    mock_redis.execute_lua_script.return_value = 1  # Allow claim

    # Simulate cache absence by ensuring redis.hmset_dict and redis.set are called
    # Since the cache was absent before, hmset_dict should be called to set it

    # Create message data
    data = {
        "auth_user_uuid": "auth-uuid-555",
        "profile_uuid": "profile-uuid-666",
        "direct_points": 5,
        "ref_points": 0,
        "idempotency_key": "unique-idempotency-key-6"
    }
    message = sample_message(json.dumps(data))

    # Initialize processor
    processor = MessageProcessor(message)

    # Execute process
    await processor.process()

    # Assertions
    message.ack.assert_called_once()
    mock_db.execute.assert_awaited_once()
    mock_db.commit.assert_awaited_once()
    mock_producer.send.assert_called_once()
    mock_redis.redis.hmset_dict.assert_called_once_with(
        "profile:auth-uuid-555",
        {
            "profile_uuid": "profile-uuid-666",
            "auth_user_uuid": "auth-uuid-555",
            "direct_points": 5,
            "ref_points": 0,
            "idempotency_key": "unique-idempotency-key-6",
            "updated_at": 1609459200
        }
    )

@pytest.mark.asyncio
@patch('app.processor.PulsarClientWrapper.get_producer')
@patch('app.processor.get_db_session')
@patch('app.processor.redis_client')
async def test_process_with_external_cache_deletion(mock_redis, mock_db_session, mock_get_producer, sample_message):
    # Setup mocks
    mock_db = AsyncMock()
    profile = AsyncMock()
    profile.profile_uuid = "profile-uuid-888"
    profile.auth_user_uuid = "auth-uuid-777"
    profile.direct_points = 8
    profile.ref_points = 0
    profile.idempotency_key = "unique-idempotency-key-7"
    profile.updated_at = AsyncMock(timestamp=lambda: 1609459200)  # 2021-01-01
    mock_db.execute.return_value.scalars.return_value.first.return_value = profile
    mock_db_session.return_value = mock_db
    mock_producer = AsyncMock()
    mock_get_producer.return_value = mock_producer

    # Simulate cache absence during processing by mocking no cache data
    # The cache will be set post DB transaction

    # Create message data
    data = {
        "auth_user_uuid": "auth-uuid-777",
        "profile_uuid": "profile-uuid-888",
        "direct_points": 8,
        "ref_points": 0,
        "idempotency_key": "unique-idempotency-key-7"
    }
    message = sample_message(json.dumps(data))

    # Initialize processor
    processor = MessageProcessor(message)

    # Execute process
    await processor.process()

    # Assertions
    message.ack.assert_called_once()
    mock_db.execute.assert_awaited_once()
    mock_db.commit.assert_awaited_once()
    mock_producer.send.assert_called_once()
    mock_redis.redis.hmset_dict.assert_called_once_with(
        "profile:auth-uuid-777",
        {
            "profile_uuid": "profile-uuid-888",
            "auth_user_uuid": "auth-uuid-777",
            "direct_points": 8,
            "ref_points": 0,
            "idempotency_key": "unique-idempotency-key-7",
            "updated_at": 1609459200
        }
    )


Explanation:

- Mocking: Uses unittest.mock to simulate external dependencies like Pulsar, Redis, and the database, isolating the processor's logic for testing.

- Coverage:
- Direct Claims: Tests processing of direct points claims.
- Referral Claims: Tests processing of referral points claims.
- Conversion Claims: Tests processing of referral to direct points conversion claims.
- Idempotency: Ensures that duplicate claims with the same idempotency_key are rejected.
- Cooldown Enforcement: Validates that claims made within the cooldown period are rejected.
- Cache Initialization: Ensures that the cache is correctly initialized when absent.

- Assertions: Verifies that appropriate methods (ack, nack, DB executions, cache updates, and message production) are called based on different scenarios.

---

## 15. Environment Variables (.env)

Securely manage configurations using environment variables, ensuring sensitive data is not hard-coded.

# .env

# Pulsar Configuration
PULSAR_SERVICE_URL=pulsar://your-pulsar-service:6650
PULSAR_CONSUMER_TOPIC=your-consumer-topic
PULSAR_CONSUMER_SUBSCRIPTION=your-consumer-subscription
PULSAR_PRODUCER_TOPIC=your-producer-topic

# Database Configuration
DATABASE_URL=postgresql+asyncpg://user:password@db-host:5432/yourdb

# Redis Configuration
REDIS_HOST=redis-host
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD=yourredispassword

# Worker Configuration
MAX_RETRIES=5
RETRY_DELAY_SECONDS=5
MAX_CONCURRENT_TASKS=100

# Idempotency Configuration
IDEMPOTENCY_ENABLED=True

# Logging Configuration
LOG_LEVEL=INFO


Explanation:

- Separation of Concerns: Keeps configuration separate from code, enhancing security and flexibility.
- Sensitive Data Protection: Ensures that sensitive information like database credentials and Redis passwords are not exposed in the codebase.

---

## Summary

This refined implementation addresses all your requirements by:

1. Implementing Idempotency via ETags: Each Profile includes an idempotency_key which acts as an ETag, ensuring that each claim is unique. The worker validates this key before processing and updates it upon successful claim processing.

2. Atomic Cooldown Checks with Lua Scripts: Utilizes a Lua script to atomically check and set cooldowns in Redis, ensuring that claims adhere to business rules without race conditions.

3. Comprehensive Cache Management: Ensures that the Redis cache is correctly initialized and updated alongside the PostgreSQL database, adhering to a write-through cache strategy for data safety.

4. Optimized Database Transactions: Consolidates database operations into a single atomic transaction without using row-level locks (FOR UPDATE), leveraging message partitioning to ensure sequential processing per user, thus enhancing performance.

5. Repository Pattern: Implements a ProfileRepository class to encapsulate all database interactions, enhancing code readability and maintainability.

6. Comprehensive Testing: Incorporates unit tests covering various scenarios, including idempotency, cooldown enforcement, and different claim types, ensuring reliability and correctness.

7. Structured Logging and Error Handling: Employs structured JSON logging for better observability and comprehensive error handling to maintain system integrity.

8. Dockerization for Deployment: Provides a Dockerfile to containerize the worker, ensuring consistent and reproducible deployments across environments.

9. Secure Configuration Management: Utilizes environment variables to manage configurations securely, preventing sensitive data exposure.