### Start a Broadcast with Task ID

Source: https://context7.com/snoups/remnashop/llms.txt

Initiates a broadcast process, creates a broadcast record, and enqueues a task for sending. Requires BroadcastDao, UnitOfWork, and send_broadcast_task.

```python
from dataclasses import dataclass
from typing import Optional
from uuid import UUID, uuid4

from src.application.common import Interactor
from src.application.common.dao import BroadcastDao
from src.application.common.uow import UnitOfWork
from src.application.dto import BroadcastDto, MessagePayloadDto, UserDto
from src.core.enums import BroadcastAudience, BroadcastStatus


@dataclass(frozen=True)
class StartBroadcastDto:
    audience: BroadcastAudience  # ALL, PLAN, SUBSCRIBED, UNSUBSCRIBED, EXPIRED, TRIAL
    payload: MessagePayloadDto
    plan_id: Optional[int] = None


class StartBroadcast(Interactor[StartBroadcastDto, UUID]):
    async def _execute(self, actor: UserDto, data: StartBroadcastDto) -> UUID:
        from src.infrastructure.taskiq.tasks.broadcast import send_broadcast_task

        total_count = await self.get_broadcast_audience_count.system(
            GetBroadcastAudienceCountDto(data.audience, data.plan_id)
        )

        async with self.uow:
            task_id = uuid4()
            broadcast = BroadcastDto(
                task_id=task_id,
                status=BroadcastStatus.PROCESSING,
                total_count=total_count,
                audience=data.audience,
                payload=data.payload,
            )
            await self.broadcast_dao.create(broadcast)
            await self.uow.commit()

            await send_broadcast_task.kicker().with_task_id(str(task_id)).kiq(
                broadcast, data.plan_id
            )

        return task_id
```

--------------------------------

### TaskIQ Scheduler Service Configuration

Source: https://context7.com/snoups/remnashop/llms.txt

Configures the TaskIQ scheduler service, similar to the worker. It uses the Remnashop image, specifies the container name, environment file, and a command to start the scheduler with task pattern and file discovery options. It also depends on Redis and the database.

```yaml
  remnashop-taskiq-scheduler:
    image: ghcr.io/snoups/remnashop:latest
    container_name: "remnashop-taskiq-scheduler"
    env_file: .env
    command: >
      taskiq scheduler src.infrastructure.taskiq.scheduler:scheduler
      --tasks-pattern src/infrastructure/taskiq/tasks/*.py -fsd
    depends_on:
      - remnashop-redis
      - remnashop-db
```

--------------------------------

### TaskIQ Worker Service Configuration

Source: https://context7.com/snoups/remnashop/llms.txt

Defines the TaskIQ worker service, using the same Remnashop image. It specifies the container name, environment file, and a command to start the worker with specific configurations for acknowledgment, task patterns, and file discovery. It depends on Redis and the database.

```yaml
  remnashop-taskiq-worker:
    image: ghcr.io/snoups/remnashop:latest
    container_name: "remnashop-taskiq-worker"
    env_file: .env
    command: >
      taskiq worker src.infrastructure.taskiq.worker:worker
      --ack-type when_received --tasks-pattern src/infrastructure/taskiq/tasks/*.py -fsd
    depends_on:
      - remnashop-redis
      - remnashop-db
```

--------------------------------

### Initialize Application Entry Point

Source: https://context7.com/snoups/remnashop/llms.txt

The main entry point configures the FastAPI app and aiogram dispatcher using the Dishka dependency injection container.

```python
# src/__main__.py - Application entry point
import uvicorn
from dishka.integrations.aiogram import setup_dishka as setup_aiogram_dishka
from dishka.integrations.fastapi import setup_dishka as setup_fastapi_dishka
from fastapi import FastAPI

from src.core.config import AppConfig
from src.core.logger import setup_logger
from src.infrastructure.di import create_aiogram_container
from src.telegram.dispatcher import get_dispatcher, get_bg_manager_factory, setup_dispatcher
from src.web.app import get_app


def application() -> FastAPI:
    config = AppConfig.get()
    setup_logger(config)

    dispatcher = get_dispatcher(config)
    bg_manager_factory = get_bg_manager_factory(dispatcher)
    setup_dispatcher(dispatcher)

    app = get_app(config, dispatcher)
    container = create_aiogram_container(config, bg_manager_factory)

    setup_aiogram_dishka(container, dispatcher, auto_inject=True)
    setup_fastapi_dishka(container, app)
    return app


if __name__ == "__main__":
    uvicorn.run(
        app=application,
        host="0.0.0.0",
        port=8000,
        factory=True,
    )
```

--------------------------------

### Configure Environment Variables

Source: https://context7.com/snoups/remnashop/llms.txt

Required settings for bot operation, including API tokens, database credentials, and Remnawave integration parameters.

```bash
# Required environment variables
APP_DOMAIN=shop.example.com          # Domain for webhooks (without https://)
APP_CRYPT_KEY=your-44-char-base64-key  # Encryption key for sensitive data

BOT_TOKEN=123456789:ABCdef...        # Telegram Bot API token from @BotFather
BOT_SECRET_TOKEN=webhook-secret      # Secret token for webhook verification
BOT_OWNER_ID=123456789               # Telegram user ID for bot owner
BOT_SUPPORT_USERNAME=support_user    # Support contact (without @)

REMNAWAVE_HOST=remnawave             # Remnawave API hostname
REMNAWAVE_TOKEN=your-api-token       # Remnawave API authentication token
REMNAWAVE_WEBHOOK_SECRET=webhook-key # Webhook secret from Remnawave

DATABASE_PASSWORD=secure-password    # PostgreSQL password

# Optional settings
APP_LOCALES="en, ru"                 # Supported locales
APP_DEFAULT_LOCALE="en"              # Default language
BOT_USE_BANNERS=true                 # Enable banner images
```

--------------------------------

### Execute Purchase Subscription Logic

Source: https://context7.com/snoups/remnashop/llms.txt

Handles new purchases, renewals, and plan changes for subscriptions. It interacts with the Unit of Work, Remnawave service, and DAOs for subscription and transaction management. Redirects to success payment upon completion.

```python
async def _execute(self, actor: UserDto, data: PurchaseSubscriptionDto) -> None:
        user = data.user
        transaction = data.transaction
        subscription = data.subscription
        plan = transaction.plan_snapshot
        purchase_type = transaction.purchase_type

        async with self.uow:
            # NEW PURCHASE
            if purchase_type == PurchaseType.NEW:
                created_user = await self.remnawave.create_user(user, plan=plan)
                new_sub = SubscriptionDto(
                    user_remna_id=created_user.uuid,
                    status=SubscriptionStatus(created_user.status),
                    traffic_limit=plan.traffic_limit,
                    device_limit=plan.device_limit,
                    traffic_limit_strategy=plan.traffic_limit_strategy,
                    expire_at=created_user.expire_at,
                    url=created_user.subscription_url,
                    plan_snapshot=plan,
                )
                await self.subscription_dao.create(new_sub, telegram_id=user.telegram_id)

            # RENEW EXISTING
            elif purchase_type == PurchaseType.RENEW:
                base_date = max(subscription.expire_at, datetime_now())
                subscription.expire_at = base_date + timedelta(days=plan.duration)
                await self.remnawave.update_user(user, uuid=subscription.user_remna_id,
                                                  subscription=subscription, reset_traffic=True)
                await self.subscription_dao.update(subscription)

            # CHANGE PLAN
            elif purchase_type == PurchaseType.CHANGE:
                await self.subscription_dao.update_status(subscription.id, SubscriptionStatus.DELETED)
                updated_user = await self.remnawave.update_user(
                    user, uuid=subscription.user_remna_id, plan=plan, reset_traffic=True
                )
                new_sub = self._build_subscription_dto(updated_user, plan)
                await self.subscription_dao.create(new_sub, telegram_id=user.telegram_id)

            await self.uow.commit()
            await self.redirect.to_success_payment(user.telegram_id, purchase_type)
```

--------------------------------

### Create Payment Use Case

Source: https://context7.com/snoups/remnashop/llms.txt

Handles the creation of a payment transaction, including free plans and integration with payment gateways. Requires UserDto, CreatePaymentDto, and various DAOs and services.

```python
import uuid
from dataclasses import dataclass
from decimal import Decimal

from src.application.common import Interactor, TranslatorHub
from src.application.common.dao import PaymentGatewayDao, TransactionDao
from src.application.common.uow import UnitOfWork
from src.application.dto import PaymentResultDto, PlanSnapshotDto, PriceDetailsDto, TransactionDto
from src.core.enums import PaymentGatewayType, PurchaseType, TransactionStatus


@dataclass(frozen=True)
class CreatePaymentDto:
    plan_snapshot: PlanSnapshotDto
    pricing: PriceDetailsDto
    purchase_type: PurchaseType
    gateway_type: PaymentGatewayType


class CreatePayment(Interactor[CreatePaymentDto, PaymentResultDto]):
    async def _execute(self, actor: UserDto, data: CreatePaymentDto) -> PaymentResultDto:
        gateway_instance = await self.get_payment_gateway_instance.system(data.gateway_type)
        i18n = self.translator_hub.get_translator_by_locale(actor.language)

        details = i18n.get(
            "payment-invoice-description",
            purchase_type=data.purchase_type,
            name=data.plan_snapshot.name,
            duration=i18n.get(*i18n_format_days(data.plan_snapshot.duration)),
        )

        transaction = TransactionDto(
            payment_id=uuid.uuid4(),
            user_telegram_id=actor.telegram_id,
            status=TransactionStatus.PENDING,
            purchase_type=data.purchase_type,
            gateway_type=gateway_instance.data.type,
            pricing=data.pricing,
            currency=gateway_instance.data.currency,
            plan_snapshot=data.plan_snapshot,
        )

        async with self.uow:
            if data.pricing.is_free:
                await self.transaction_dao.create(transaction)
                await self.uow.commit()
                return PaymentResultDto(id=transaction.payment_id, url=None)

            payment = await gateway_instance.handle_create_payment(
                amount=data.pricing.final_amount,
                details=details,
            )

            transaction.payment_id = payment.id
            await self.transaction_dao.create(transaction)
            await self.uow.commit()

        return payment
```

--------------------------------

### Pricing Service Calculation

Source: https://context7.com/snoups/remnashop/llms.txt

Calculates final prices with discount support and applies currency-specific rounding rules. Handles zero or negative prices by returning zero. Discounts are capped at 100%.

```python
from decimal import ROUND_DOWN, Decimal

from src.application.dto import PriceDetailsDto, UserDto
from src.core.enums import Currency


class PricingService:
    def calculate(self, user: UserDto, price: Decimal, currency: Currency) -> PriceDetailsDto:
        if price <= 0:
            return PriceDetailsDto(
                original_amount=Decimal(0),
                discount_percent=0,
                final_amount=Decimal(0),
            )

        discount_percent = min(user.purchase_discount or user.personal_discount or 0, 100)

        if discount_percent >= 100:
            return PriceDetailsDto(original_amount=price, discount_percent=100, final_amount=Decimal(0))

        discounted = price * (Decimal(100) - Decimal(discount_percent)) / Decimal(100)
        final_amount = self.apply_currency_rules(discounted, currency)

        return PriceDetailsDto(
            original_amount=price,
            discount_percent=discount_percent if final_amount != price else 0,
            final_amount=final_amount,
        )

    def apply_currency_rules(self, amount: Decimal, currency: Currency) -> Decimal:
        match currency:
            case Currency.XTR | Currency.RUB:  # Telegram Stars, Rubles - whole numbers
                amount = amount.to_integral_value(rounding=ROUND_DOWN)
                min_amount = Decimal(1)
            case _:  # USD and others - 2 decimal places
                amount = amount.quantize(Decimal("0.01")).normalize()
                min_amount = Decimal("0.01")

        return max(amount, min_amount)
```

--------------------------------

### Docker Deployment Configuration

Source: https://context7.com/snoups/remnashop/llms.txt

Docker Compose configuration for deploying the Remnashop stack, including PostgreSQL, Redis, the main application, and TaskIQ workers.

```yaml
version: "3.8"

services:
  postgres:
    image: postgres:15
    volumes:
      - postgres_data:/var/lib/postgresql/data/
    environment:
      POSTGRES_DB: "${POSTGRES_DB:-remnashop}"
      POSTGRES_USER: "${POSTGRES_USER:-remnashop}"
      POSTGRES_PASSWORD: "${POSTGRES_PASSWORD:-remnashop}"
    ports:
      - "5432:5432"

  redis:
    image: redis:7
    ports:
      - "6379:6379"

  app:
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    depends_on:
      - postgres
      - redis
    environment:
      DATABASE_URL: "postgresql://${POSTGRES_USER:-remnashop}:${POSTGRES_PASSWORD:-remnashop}@postgres:5432/${POSTGRES_DB:-remnashop}"
      REDIS_URL: "redis://redis:6379/0"
      SECRET_KEY: "${SECRET_KEY:-changeme}"
      DEBUG: "${DEBUG:-true}"

  worker:
    build:
      context: .
      dockerfile: Dockerfile.worker
    depends_on:
      - postgres
      - redis
    environment:
      DATABASE_URL: "postgresql://${POSTGRES_USER:-remnashop}:${POSTGRES_PASSWORD:-remnashop}@postgres:5432/${POSTGRES_DB:-remnashop}"
      REDIS_URL: "redis://redis:6379/0"
      SECRET_KEY: "${SECRET_KEY:-changeme}"
      DEBUG: "${DEBUG:-true}"

volumes:
  postgres_data:

```

--------------------------------

### Remnawave User Creation

Source: https://context7.com/snoups/remnashop/llms.txt

Creates a new user in the Remnawave panel. Maps user, plan, and subscription details to Remnawave's CreateUserRequestDto. Requires RemnawaveSDK.

```python
# src/infrastructure/services/remnawave.py
from typing import Optional
from uuid import UUID

from remnapy import RemnawaveSDK
from remnapy.models import CreateUserRequestDto, UpdateUserRequestDto, UserResponseDto

from src.application.common import Remnawave
from src.application.dto import PlanSnapshotDto, SubscriptionDto, UserDto
from src.core.utils.converters import days_to_datetime, gb_to_bytes


class RemnawaveImpl(Remnawave):
    def __init__(self, sdk: RemnawaveSDK) -> None:
        self.sdk = sdk

    async def try_connection(self) -> None:
        response = await self.sdk.system.get_stats()
        if not isinstance(response, GetStatsResponseDto):
            raise ValueError("Invalid response from Remnawave panel")

    async def create_user(
        self, 
        user: UserDto, 
        plan: Optional[PlanSnapshotDto] = None, 
        subscription: Optional[SubscriptionDto] = None,
    ) -> UserResponseDto:
        request = CreateUserRequestDto(
            username=user.remna_name,
            telegram_id=user.telegram_id,
            expire_at=days_to_datetime(plan.duration), 
            traffic_limit_strategy=plan.traffic_limit_strategy,
            traffic_limit_bytes=gb_to_bytes(plan.traffic_limit),
            hwid_device_limit=plan.device_limit,
            description=user.remna_description,
            tag=plan.tag,
            active_internal_squads=plan.internal_squads,
            external_squad_uuid=plan.external_squad,
        )
        return await self.sdk.users.create_user(request)
```

--------------------------------

### Handle Payment Gateway Webhooks

Source: https://context7.com/snoups/remnashop/llms.txt

Processes payment callbacks by dynamically resolving the gateway instance and triggering background transaction tasks.

```python
# src/web/endpoints/payments.py - POST /api/v1/payments/{gateway_type}
from dishka import FromDishka
from dishka.integrations.fastapi import inject
from fastapi import APIRouter, Request, Response, status
from loguru import logger

from src.application.use_cases.gateways.queries.providers import GetPaymentGatewayInstance
from src.core.enums import PaymentGatewayType
from src.infrastructure.taskiq.tasks.payments import handle_payment_transaction_task

router = APIRouter(prefix="/api/v1/payments")


@router.post("/{gateway_type}")
@inject
async def payments_webhook(
    gateway_type: str,
    request: Request,
    get_payment_gateway_instance: FromDishka[GetPaymentGatewayInstance],
) -> Response:
    try:
        gateway_enum = PaymentGatewayType(gateway_type.upper())
    except ValueError:
        return Response(status_code=status.HTTP_404_NOT_FOUND)

    gateway = await get_payment_gateway_instance.system(gateway_enum)

    if not gateway.data.is_active or not gateway.data.settings.is_configured:
        return Response(status_code=status.HTTP_404_NOT_FOUND)

    payment_id, payment_status = await gateway.handle_webhook(request)
    await handle_payment_transaction_task.kiq(payment_id, payment_status)

    return await gateway.build_webhook_response(request)

# Supported payment gateways:
# TELEGRAM_STARS, YOOKASSA, YOOMONEY, CRYPTOMUS, HELEKET,
# CRYPTOPAY, FREEKASSA, MULENPAY, PAYMASTER, PLATEGA, ROBOKASSA, URLPAY, WATA
```

--------------------------------

### Assign Referral Rewards Logic

Source: https://context7.com/snoups/remnashop/llms.txt

Calculates and assigns referral rewards based on user transactions and referral chain. Requires settings and referral data to be available.

```python
from dataclasses import dataclass

from src.application.common import EventPublisher, Interactor
from src.application.common.dao import ReferralDao, SettingsDao
from src.application.dto import ReferralRewardDto, TransactionDto, UserDto
from src.application.events import ReferralRewardReceivedEvent
from src.core.enums import PurchaseType, ReferralAccrualStrategy, ReferralLevel, ReferralRewardType


@dataclass(frozen=True)
class AssignReferralRewardsDto:
    user: UserDto
    transaction: TransactionDto


class AssignReferralRewards(Interactor[AssignReferralRewardsDto, None]):
    async def _execute(self, actor: UserDto, data: AssignReferralRewardsDto) -> None:
        settings = await self.settings_dao.get()

        if not settings.referral.enable:
            return

        if (settings.referral.accrual_strategy == ReferralAccrualStrategy.ON_FIRST_PAYMENT
            and data.transaction.purchase_type != PurchaseType.NEW):
            return

        referral, parent = await self.referral_dao.get_referral_chain(data.user.telegram_id)
        if not referral:
            return

        reward_chain = {ReferralLevel.FIRST: referral.referrer}
        if parent:
            reward_chain[ReferralLevel.SECOND] = parent.referrer

        for level, referrer in reward_chain.items():
            if level > settings.referral.level:
                continue

            config_value = settings.referral.reward.config.get(level)
            reward_amount = await self.calculate_referral_reward.system(
                CalculateReferralRewardDto(settings.referral, data.transaction, config_value)
            )

            if reward_amount and reward_amount > 0:
                async with self.uow:
                    reward = await self.referral_dao.create_reward(
                        ReferralRewardDto(
                            user_telegram_id=referrer.telegram_id,
                            type=settings.referral.reward.type,  # POINTS or EXTRA_DAYS
                            amount=reward_amount,
                        ),
                        referral_id=referral.id,
                    )
                    await self.uow.commit()

                    await self.give_referrer_reward.system(
                        GiveReferrerRewardDto(referrer.telegram_id, reward, data.user.name)
                    )
```

--------------------------------

### Handle Telegram Stars Pre-Checkout Query

Source: https://context7.com/snoups/remnashop/llms.txt

Answers pre-checkout queries for Telegram Stars payments. Responds with 'ok=True' if an invoice payload exists, otherwise 'ok=False'.

```python
from uuid import UUID

from aiogram import Bot, F, Router
from aiogram.types import Message, PreCheckoutQuery
from dishka import FromDishka

from src.application.dto import UserDto
from src.application.use_cases.gateways.commands.payment import ProcessPayment, ProcessPaymentDto
from src.core.enums import TransactionStatus

router = Router(name=__name__)


@router.pre_checkout_query()
async def on_pre_checkout(pre_checkout_query: PreCheckoutQuery, user: UserDto) -> None:
    if pre_checkout_query.invoice_payload:
        await pre_checkout_query.answer(ok=True)
    else:
        await pre_checkout_query.answer(ok=False)
```

--------------------------------

### Remnashop Application Service Configuration

Source: https://context7.com/snoups/remnashop/llms.txt

Sets up the main Remnashop application service. It specifies the Docker image, container name, environment file for configuration, port mapping, volumes for logs and assets, and dependencies on healthy database and Redis services.

```yaml
  remnashop:
    image: ghcr.io/snoups/remnashop:latest
    container_name: "remnashop"
    env_file: .env
    ports:
      - 127.0.0.1:5000:${APP_PORT:-5000}
    volumes:
      - ./logs:/opt/remnashop/logs
      - ./assets:/opt/remnashop/assets
    depends_on:
      remnashop-db:
        condition: service_healthy
      remnashop-redis:
        condition: service_healthy
```

--------------------------------

### PostgreSQL Database Service Configuration

Source: https://context7.com/snoups/remnashop/llms.txt

Defines the PostgreSQL database service, including image, container name, environment variables for credentials and database name, data volume, and a health check command. Uses environment variables for sensitive information and defaults if not provided.

```yaml
services:
  remnashop-db:
    image: postgres:17
    container_name: "remnashop-db"
    environment:
      - POSTGRES_USER=${DATABASE_USER:-remnashop}
      - POSTGRES_PASSWORD=${DATABASE_PASSWORD}
      - POSTGRES_DB=${DATABASE_NAME:-remnashop}
    volumes:
      - remnashop-db-data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $${DATABASE_USER:-remnashop}"]
      interval: 3s
      timeout: 10s
      retries: 3
```

--------------------------------

### Implement User Registration Interactor

Source: https://context7.com/snoups/remnashop/llms.txt

Manages user creation or retrieval during Telegram interactions, including referral code attachment and locale configuration.

```python
# src/application/use_cases/user/commands/registration.py
from dataclasses import dataclass
from typing import Optional, Self

from aiogram.types import TelegramObject
from aiogram.types import User as AiogramUser

from src.application.common import Cryptographer, EventPublisher, Interactor
from src.application.common.dao import UserDao
from src.application.common.uow import UnitOfWork
from src.application.dto import UserDto
from src.application.events import UserRegisteredEvent
from src.core.enums import Locale, Role


@dataclass
class GetOrCreateUserDto:
    telegram_id: int
    username: Optional[str]
    full_name: str
    language_code: Optional[str]
    event: TelegramObject
    role: Role = Role.USER

    @classmethod
    def from_aiogram(cls, user: AiogramUser, event: TelegramObject) -> Self:
        return cls(
            telegram_id=user.id,
            username=user.username,
            full_name=user.full_name,
            language_code=user.language_code,
            event=event,
        )


class GetOrCreateUser(Interactor[GetOrCreateUserDto, Optional[UserDto]]):
    async def _execute(self, actor: UserDto, data: GetOrCreateUserDto) -> Optional[UserDto]:
        async with self.uow:
            user = await self.user_dao.get_by_telegram_id(data.telegram_id)
            if user:
                return user

            user_dto = UserDto(
                telegram_id=data.telegram_id,
                username=data.username,
                referral_code=self.cryptographer.generate_short_code(data.telegram_id),
                name=data.full_name,
                role=data.role,
                language=Locale(data.language_code) if data.language_code in self.config.locales else self.config.default_locale,
            )

            user = await self.user_dao.create(user_dto)
            await self.uow.commit()

        # Attach referral if present
        referral_code = await self.get_referral_code_from_event.system(data.event)
        if referral_code:
            await self.attach_referral.system(AttachReferralDto(user.telegram_id, referral_code))

        await self.event_publisher.publish(UserRegisteredEvent(
            telegram_id=user.telegram_id,
            username=user.username,
            name=user.name,
        ))

        return user
```

--------------------------------

### POST /api/v1/payments/{gateway_type}

Source: https://context7.com/snoups/remnashop/llms.txt

Endpoint to handle incoming webhooks from various supported payment providers.

```APIDOC
## POST /api/v1/payments/{gateway_type}

### Description
Processes payment callbacks from configured gateways. Validates the gateway type and triggers a background task to handle the transaction.

### Method
POST

### Endpoint
/api/v1/payments/{gateway_type}

### Parameters
#### Path Parameters
- **gateway_type** (string) - Required - The identifier for the payment provider (e.g., TELEGRAM_STARS, YOOKASSA, CRYPTOMUS).

### Response
#### Success Response (200)
- **response** (object) - Returns a gateway-specific webhook response.

#### Error Response (404)
- **status** (integer) - Returned if the gateway type is invalid or the gateway is not active/configured.
```

--------------------------------

### Docker Networks Configuration

Source: https://context7.com/snoups/remnashop/llms.txt

Defines the Docker network 'remnawave-network' using the bridge driver, which will be used for communication between the services.

```yaml
networks:
  remnawave-network:
    name: remnawave-network
    driver: bridge
```

--------------------------------

### Docker Volumes Configuration

Source: https://context7.com/snoups/remnashop/llms.txt

Declares the named volumes used for persistent storage for the PostgreSQL database and Redis data.

```yaml
volumes:
  remnashop-db-data:
  remnashop-redis-data:
```

--------------------------------

### Redis Cache Service Configuration

Source: https://context7.com/snoups/remnashop/llms.txt

Configures the Redis service for caching and task queueing. Specifies the image, container name, command to disable persistence and eviction, and a volume for data storage.

```yaml
  remnashop-redis:
    image: valkey/valkey:9-alpine
    container_name: "remnashop-redis"
    command: valkey-server --save "" --appendonly no --maxmemory-policy noeviction
    volumes:
      - remnashop-redis-data:/data
```

--------------------------------

### Implement Telegram Webhook Endpoint

Source: https://context7.com/snoups/remnashop/llms.txt

Handles incoming Telegram updates with secret token validation and asynchronous task processing.

```python
import asyncio
import secrets
from typing import Annotated

from aiogram import Bot, Dispatcher
from aiogram.methods import TelegramMethod
from aiogram.types import Update
from dishka.integrations.fastapi import FromDishka, inject
from fastapi import Body, FastAPI, Header, HTTPException, Response, status
from loguru import logger


class TelegramWebhookEndpoint:
    def __init__(self, dispatcher: Dispatcher, secret_token: str) -> None:
        self.dispatcher = dispatcher
        self.secret_token = secret_token
        self._feed_update_tasks: set[asyncio.Task] = set()

    def register(self, app: FastAPI, path: str) -> None:
        app.add_api_route(path, endpoint=self._handle_request, methods=["POST"])

    @inject
    async def _handle_request(
        self,
        update: Annotated[Update, Body()],
        bot: FromDishka[Bot],
        x_telegram_bot_api_secret_token: Annotated[str, Header()] = "",
    ) -> Response:
        if not x_telegram_bot_api_secret_token:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

        if not secrets.compare_digest(x_telegram_bot_api_secret_token, self.secret_token):
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)

        task = asyncio.create_task(self._feed_update(bot, update))
        self._feed_update_tasks.add(task)
        task.add_done_callback(self._feed_update_tasks.discard)

        return Response(status_code=status.HTTP_200_OK)

    async def _feed_update(self, bot: Bot, update: Update) -> None:
        result = await self.dispatcher.feed_update(bot, update)
        if isinstance(result, TelegramMethod):
            await result.as_(bot)
```

--------------------------------

### Handle Remnawave Webhooks in FastAPI

Source: https://context7.com/snoups/remnashop/llms.txt

Processes incoming webhooks from Remnawave, validates the payload, and routes events to the appropriate service handler.

```python
from typing import cast

from dishka import FromDishka
from dishka.integrations.fastapi import inject
from fastapi import APIRouter, HTTPException, Request, Response, status
from remnapy.controllers import WebhookUtility
from remnapy.models.webhook import NodeDto, UserDto, UserHwidDeviceEventDto

from src.application.services import RemnaWebhookService
from src.core.config import AppConfig

router = APIRouter(prefix="/api/v1")


@router.post("/remnawave")
@inject
async def remnawave_webhook(
    request: Request,
    config: FromDishka[AppConfig],
    remna_webhook_service: FromDishka[RemnaWebhookService],
) -> Response:
    raw_body = await request.body()
    data = await request.json()

    payload = WebhookUtility.parse_webhook(
        body=raw_body.decode("utf-8"),
        headers=dict(request.headers),
        webhook_secret=config.remnawave.webhook_secret.get_secret_value(),
        validate=True,
    )

    if WebhookUtility.is_user_event(payload.event):
        user = cast(UserDto, WebhookUtility.get_typed_data(payload))
        await remna_webhook_service.handle_user_event(payload.event, user)

    elif WebhookUtility.is_user_hwid_devices_event(payload.event):
        event = cast(UserHwidDeviceEventDto, WebhookUtility.get_typed_data(payload))
        await remna_webhook_service.handle_device_event(
            payload.event, event.user, event.hwid_user_device
        )

    elif WebhookUtility.is_node_event(payload.event):
        node = cast(NodeDto, WebhookUtility.get_typed_data(payload))
        await remna_webhook_service.handle_node_event(payload.event, node)

    return Response(status_code=status.HTTP_200_OK)
```

--------------------------------

### POST /api/v1/remnawave

Source: https://context7.com/snoups/remnashop/llms.txt

Handles incoming webhooks from Remnawave. It parses the payload, validates it against a secret, and routes events to the appropriate service handler based on the event type.

```APIDOC
## POST /api/v1/remnawave

### Description
Processes incoming webhooks from Remnawave. The endpoint validates the request using a configured secret and dispatches events (User, Device, or Node) to the internal RemnaWebhookService.

### Method
POST

### Endpoint
/api/v1/remnawave

### Request Body
- **payload** (JSON) - Required - The webhook payload containing event data, validated via webhook_secret.

### Response
#### Success Response (200)
- **status** (integer) - Returns 200 OK upon successful processing of the webhook event.
```

--------------------------------

### Remnawave User Update

Source: https://context7.com/snoups/remnashop/llms.txt

Updates an existing user in the Remnawave panel. Allows for resetting traffic and uses subscription details if available. Requires RemnawaveSDK.

```python
    async def update_user(
        self, 
        user: UserDto, 
        uuid: UUID, 
        plan: Optional[PlanSnapshotDto] = None, 
        subscription: Optional[SubscriptionDto] = None,
        reset_traffic: bool = False,
    ) -> UserResponseDto:
        request = UpdateUserRequestDto(
            uuid=uuid,
            telegram_id=user.telegram_id,
            expire_at=subscription.expire_at if subscription else days_to_datetime(plan.duration),
            traffic_limit_bytes=gb_to_bytes(subscription.traffic_limit if subscription else plan.traffic_limit),
            hwid_device_limit=subscription.device_limit if subscription else plan.device_limit,
        )
        remna_user = await self.sdk.users.update_user(request)

        if reset_traffic:
            await self.sdk.users.reset_user_traffic(uuid)

        return remna_user
```

--------------------------------

### POST /telegram/webhook

Source: https://context7.com/snoups/remnashop/llms.txt

Endpoint to receive and process Telegram bot updates via webhooks. Requires a secret token for authentication.

```APIDOC
## POST /telegram/webhook

### Description
Receives Telegram updates and dispatches them to the bot dispatcher. Validates the request using a secret token provided in the headers.

### Method
POST

### Endpoint
/telegram/webhook

### Parameters
#### Header Parameters
- **x-telegram-bot-api-secret-token** (string) - Required - Secret token for authentication.

#### Request Body
- **update** (object) - Required - The Telegram Update object.

### Response
#### Success Response (200)
- **status** (integer) - Returns 200 OK upon successful receipt and task creation.
```

--------------------------------

### Handle Successful Telegram Stars Payment

Source: https://context7.com/snoups/remnashop/llms.txt

Processes successful Telegram Stars payments, including refunding test payments for the owner and updating transaction status.

```python
@router.message(F.successful_payment)
async def on_successful_payment(
    message: Message,
    user: UserDto,
    bot: Bot,
    process_payment: FromDishka[ProcessPayment],
) -> None:
    payment = message.successful_payment
    if not payment:
        return

    # Refund test payments for owner
    if user.is_owner:
        await bot.refund_star_payment(
            user_id=user.telegram_id,
            telegram_payment_charge_id=payment.telegram_payment_charge_id,
        )

    await process_payment.system(
        ProcessPaymentDto(
            payment_id=UUID(payment.invoice_payload),
            new_transaction_status=TransactionStatus.COMPLETED,
        )
    )
```

--------------------------------

### Remnawave Subscription Revocation

Source: https://context7.com/snoups/remnashop/llms.txt

Revokes a user's subscription in the Remnawave panel. Requires RemnawaveSDK.

```python
    async def revoke_subscription(self, uuid: UUID) -> None:
        await self.sdk.users.revoke_user_subscription(uuid)
```

--------------------------------

### Remnawave User Deletion

Source: https://context7.com/snoups/remnashop/llms.txt

Deletes a user from the Remnawave panel. Returns a boolean indicating if the deletion was successful. Requires RemnawaveSDK.

```python
    async def delete_user(self, uuid: UUID) -> bool:
        response = await self.sdk.users.delete_user(uuid)
        return response.is_deleted
```

--------------------------------

### Cancel a Broadcast by Task ID

Source: https://context7.com/snoups/remnashop/llms.txt

Cancels a broadcast if its status is PROCESSING. Updates the broadcast status to CANCELED. Requires BroadcastDao and UnitOfWork.

```python
class CancelBroadcast(Interactor[UUID, None]):
    async def _execute(self, actor: UserDto, task_id: UUID) -> None:
        async with self.uow:
            broadcast = await self.broadcast_dao.get_by_task_id(task_id)
            if broadcast.status != BroadcastStatus.PROCESSING:
                raise ValueError("Broadcast is not cancelable")

            await self.broadcast_dao.update_status(task_id, BroadcastStatus.CANCELED)
            await self.uow.commit()
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.