import uuid
from collections.abc import AsyncIterator, Sequence
from contextlib import asynccontextmanager
from typing import Any, Literal
from urllib.parse import urlencode

import stripe as stripe_lib
import structlog
from sqlalchemy import func, select
from sqlalchemy.orm import contains_eager, joinedload

from polar.account.repository import AccountRepository
from polar.auth.models import AuthSubject
from polar.billing_entry.service import billing_entry as billing_entry_service
from polar.checkout.eventstream import CheckoutEvent, publish_checkout_event
from polar.checkout.guard import has_product_checkout
from polar.config import settings
from polar.customer.repository import CustomerRepository
from polar.customer_portal.schemas.order import (
    CustomerOrderPaymentConfirmation,
    CustomerOrderUpdate,
)
from polar.customer_session.service import customer_session as customer_session_service
from polar.email.react import render_email_template
from polar.email.schemas import EmailAdapter
from polar.email.sender import Attachment, enqueue_email
from polar.enums import PaymentProcessor
from polar.event.service import event as event_service
from polar.event.system import OrderPaidMetadata, SystemEvent, build_system_event
from polar.eventstream.service import publish as eventstream_publish
from polar.exceptions import PolarError
from polar.file.s3 import S3_SERVICES
from polar.held_balance.service import held_balance as held_balance_service
from polar.integrations.stripe.service import stripe as stripe_service
from polar.invoice.service import invoice as invoice_service
from polar.kit.db.postgres import AsyncReadSession, AsyncSession
from polar.kit.metadata import MetadataQuery, apply_metadata_clause
from polar.kit.pagination import PaginationParams
from polar.kit.sorting import Sorting
from polar.kit.tax import (
    IncompleteTaxLocation,
    InvalidTaxLocation,
    TaxabilityReason,
    TaxCalculation,
    TaxRate,
    calculate_tax,
    from_stripe_tax_rate_details,
)
from polar.kit.utils import utc_now
from polar.logging import Logger
from polar.models import (
    Checkout,
    Customer,
    Order,
    OrderItem,
    Organization,
    Payment,
    PaymentMethod,
    Product,
    Subscription,
    Transaction,
    User,
    WalletTransaction,
)
from polar.models.held_balance import HeldBalance
from polar.models.order import OrderBillingReasonInternal, OrderStatus
from polar.models.product import ProductBillingType
from polar.models.subscription import SubscriptionStatus
from polar.models.transaction import TransactionType
from polar.models.webhook_endpoint import WebhookEventType
from polar.notifications.notification import (
    MaintainerNewProductSaleNotificationPayload,
    NotificationType,
)
from polar.notifications.service import PartialNotification
from polar.notifications.service import notifications as notifications_service
from polar.organization.repository import OrganizationRepository
from polar.organization.service import organization as organization_service
from polar.payment.repository import PaymentRepository
from polar.payment_method.repository import PaymentMethodRepository
from polar.payment_method.service import payment_method as payment_method_service
from polar.product.guard import is_custom_price, is_seat_price, is_static_price
from polar.subscription.service import subscription as subscription_service
from polar.transaction.service.balance import PaymentTransactionForChargeDoesNotExist
from polar.transaction.service.balance import (
    balance_transaction as balance_transaction_service,
)
from polar.transaction.service.platform_fee import (
    platform_fee_transaction as platform_fee_transaction_service,
)
from polar.wallet.repository import WalletTransactionRepository
from polar.wallet.service import wallet as wallet_service
from polar.webhook.service import webhook as webhook_service
from polar.worker import enqueue_job, make_bulk_job_delay_calculator

from .repository import OrderRepository
from .schemas import OrderInvoice, OrderUpdate
from .sorting import OrderSortProperty

log: Logger = structlog.get_logger()


class OrderError(PolarError): ...


class RecurringProduct(OrderError):
    def __init__(self, checkout: Checkout, product: Product) -> None:
        self.checkout = checkout
        self.product = product
        message = (
            f"Checkout {checkout.id} is for product {product.id}, "
            "which is a recurring product."
        )
        super().__init__(message)


class NotRecurringProduct(OrderError):
    def __init__(self, checkout: Checkout, product: Product) -> None:
        self.checkout = checkout
        self.product = product
        message = (
            f"Checkout {checkout.id} is for product {product.id}, "
            "which is not a recurring product."
        )
        super().__init__(message)


class MissingCheckoutCustomer(OrderError):
    def __init__(self, checkout: Checkout) -> None:
        self.checkout = checkout
        message = f"Checkout {checkout.id} is missing a customer."
        super().__init__(message)


class AlreadyBalancedOrder(OrderError):
    def __init__(self, order: Order, payment_transaction: Transaction) -> None:
        self.order = order
        self.payment_transaction = payment_transaction
        message = (
            f"The order {order.id} with payment {payment_transaction.id} "
            "has already been balanced."
        )
        super().__init__(message)


class NotPaidOrder(OrderError):
    def __init__(self, order: Order) -> None:
        self.order = order
        message = f"Order {order.id} is not paid, so an invoice cannot be generated."
        super().__init__(message, 422)


class MissingInvoiceBillingDetails(OrderError):
    def __init__(self, order: Order) -> None:
        self.order = order
        message = (
            "Billing name and address are required "
            "to generate an invoice for this order."
        )
        super().__init__(message, 422)


class InvoiceDoesNotExist(OrderError):
    def __init__(self, order: Order) -> None:
        self.order = order
        message = f"No invoice exists for order {order.id}."
        super().__init__(message, 404)


class OrderNotEligibleForRetry(OrderError):
    def __init__(self, order: Order) -> None:
        self.order = order
        message = f"Order {order.id} is not eligible for payment retry."
        super().__init__(message, 422)


class NoPendingBillingEntries(OrderError):
    def __init__(self, subscription: Subscription) -> None:
        self.subscription = subscription
        message = (
            f"No pending billing entries found for subscription {subscription.id}."
        )
        super().__init__(message)


class OrderNotPending(OrderError):
    def __init__(self, order: Order) -> None:
        self.order = order
        message = f"Order {order.id} is not pending"
        super().__init__(message)


class PaymentAlreadyInProgress(OrderError):
    def __init__(self, order: Order) -> None:
        self.order = order
        message = f"Payment for order {order.id} is already in progress"
        super().__init__(message, 409)


class CardPaymentFailed(OrderError):
    """Exception for card-related payment failures that should not be retried."""

    def __init__(
        self,
        order: Order,
        stripe_error: stripe_lib.CardError | stripe_lib.InvalidRequestError,
    ) -> None:
        self.order = order
        self.stripe_error = stripe_error
        message = f"Card payment failed for order {order.id}: {stripe_error.user_message or stripe_error.code}"
        super().__init__(message, 402)


class PaymentRetryValidationError(OrderError):
    def __init__(self, message: str) -> None:
        super().__init__(message, 422)


class SubscriptionNotTrialing(OrderError):
    def __init__(self, subscription: Subscription) -> None:
        self.subscription = subscription
        message = f"Subscription {subscription.id} is not in trialing status."
        super().__init__(message)


def _is_empty_customer_address(customer_address: dict[str, Any] | None) -> bool:
    return customer_address is None or customer_address["country"] is None


class OrderService:
    @asynccontextmanager
    async def acquire_payment_lock(
        self, session: AsyncSession, order: Order, *, release_on_success: bool = True
    ) -> AsyncIterator[None]:
        """
        Context manager to acquire and release a payment lock for an order.
        """

        repository = OrderRepository.from_session(session)

        # Try to acquire the lock
        lock_acquired = await repository.acquire_payment_lock_by_id(order.id)
        if not lock_acquired:
            raise PaymentAlreadyInProgress(order)

        try:
            yield
        except Exception:
            await repository.release_payment_lock(order, flush=True)
            raise
        else:
            if release_on_success:
                await repository.release_payment_lock(order, flush=True)

    async def list(
        self,
        session: AsyncReadSession,
        auth_subject: AuthSubject[User | Organization],
        *,
        organization_id: Sequence[uuid.UUID] | None = None,
        product_id: Sequence[uuid.UUID] | None = None,
        product_billing_type: Sequence[ProductBillingType] | None = None,
        discount_id: Sequence[uuid.UUID] | None = None,
        customer_id: Sequence[uuid.UUID] | None = None,
        checkout_id: Sequence[uuid.UUID] | None = None,
        metadata: MetadataQuery | None = None,
        pagination: PaginationParams,
        sorting: list[Sorting[OrderSortProperty]] = [
            (OrderSortProperty.created_at, True)
        ],
    ) -> tuple[Sequence[Order], int]:
        repository = OrderRepository.from_session(session)
        statement = repository.get_readable_statement(auth_subject)

        statement = (
            statement.join(Order.discount, isouter=True)
            .join(Order.product, isouter=True)
            .options(
                *repository.get_eager_options(
                    customer_load=contains_eager(Order.customer),
                    product_load=contains_eager(Order.product),
                    discount_load=contains_eager(Order.discount),
                )
            )
        )

        if organization_id is not None:
            statement = statement.where(Customer.organization_id.in_(organization_id))

        if product_id is not None:
            statement = statement.where(Order.product_id.in_(product_id))

        if product_billing_type is not None:
            statement = statement.where(Product.billing_type.in_(product_billing_type))

        if discount_id is not None:
            statement = statement.where(Order.discount_id.in_(discount_id))

        # TODO:
        # Once we add `external_customer_id` be sure to filter for non-deleted.
        # Since it could be shared across soft deleted records whereas the unique ID cannot.
        if customer_id is not None:
            statement = statement.where(Order.customer_id.in_(customer_id))

        if checkout_id is not None:
            statement = statement.where(Order.checkout_id.in_(checkout_id))

        if metadata is not None:
            statement = apply_metadata_clause(Order, statement, metadata)

        statement = repository.apply_sorting(statement, sorting)

        return await repository.paginate(
            statement, limit=pagination.limit, page=pagination.page
        )

    async def get(
        self,
        session: AsyncReadSession,
        auth_subject: AuthSubject[User | Organization],
        id: uuid.UUID,
    ) -> Order | None:
        repository = OrderRepository.from_session(session)
        statement = (
            repository.get_readable_statement(auth_subject)
            .options(
                *repository.get_eager_options(
                    customer_load=contains_eager(Order.customer),
                    product_load=joinedload(Order.product),
                )
            )
            .where(Order.id == id)
        )
        return await repository.get_one_or_none(statement)

    async def update(
        self,
        session: AsyncSession,
        order: Order,
        order_update: OrderUpdate | CustomerOrderUpdate,
    ) -> Order:
        repository = OrderRepository.from_session(session)
        order = await repository.update(
            order, update_dict=order_update.model_dump(exclude_unset=True)
        )

        await self.send_webhook(session, order, WebhookEventType.order_updated)

        return order

    async def set_refunds_blocked(
        self,
        session: AsyncSession,
        order: Order,
        blocked: bool,
    ) -> Order:
        repository = OrderRepository.from_session(session)
        refunds_blocked_at = utc_now() if blocked else None
        order = await repository.update(
            order, update_dict={"refunds_blocked_at": refunds_blocked_at}
        )

        log.info(
            "order.refunds_blocked_changed",
            order_id=order.id,
            refunds_blocked=blocked,
            refunds_blocked_at=refunds_blocked_at,
        )

        return order

    async def trigger_invoice_generation(
        self, session: AsyncSession, order: Order
    ) -> None:
        if not order.paid:
            raise NotPaidOrder(order)

        if order.billing_name is None or order.billing_address is None:
            raise MissingInvoiceBillingDetails(order)

        enqueue_job("order.invoice", order_id=order.id)

    async def generate_invoice(self, session: AsyncSession, order: Order) -> Order:
        invoice_path = await invoice_service.create_order_invoice(order)
        repository = OrderRepository.from_session(session)
        order = await repository.update(
            order, update_dict={"invoice_path": invoice_path}
        )

        await eventstream_publish(
            "order.invoice_generated",
            {"order_id": order.id},
            customer_id=order.customer_id,
            organization_id=order.organization.id,
        )

        await self.send_webhook(session, order, WebhookEventType.order_updated)

        return order

    async def get_order_invoice(self, order: Order) -> OrderInvoice:
        if order.invoice_path is None:
            raise InvoiceDoesNotExist(order)

        url, _ = await invoice_service.get_order_invoice_url(order)
        return OrderInvoice(url=url)

    async def create_from_checkout_one_time(
        self, session: AsyncSession, checkout: Checkout, payment: Payment | None = None
    ) -> Order:
        assert has_product_checkout(checkout)

        product = checkout.product
        if product.is_recurring:
            raise RecurringProduct(checkout, product)

        order = await self._create_order_from_checkout(
            session, checkout, OrderBillingReasonInternal.purchase, payment
        )

        # For seat-based orders, benefits are granted when seats are claimed
        # For non-seat orders, grant benefits immediately
        prices = checkout.prices[product.id]
        has_seat_price = any(is_seat_price(price) for price in prices)
        if not has_seat_price:
            enqueue_job(
                "benefit.enqueue_benefits_grants",
                task="grant",
                customer_id=order.customer.id,
                product_id=product.id,
                order_id=order.id,
            )

        # Trigger notifications
        organization = checkout.organization
        await self.send_admin_notification(session, organization, order)

        return order

    async def create_from_checkout_subscription(
        self,
        session: AsyncSession,
        checkout: Checkout,
        subscription: Subscription,
        billing_reason: Literal[
            OrderBillingReasonInternal.subscription_create,
            OrderBillingReasonInternal.subscription_update,
        ],
        payment: Payment | None = None,
    ) -> Order:
        assert has_product_checkout(checkout)

        product = checkout.product
        if not product.is_recurring:
            raise NotRecurringProduct(checkout, product)

        if subscription.trialing:
            return await self.create_trial_order(
                session, subscription, billing_reason, checkout
            )

        return await self._create_order_from_checkout(
            session, checkout, billing_reason, payment, subscription
        )

    async def _create_order_from_checkout(
        self,
        session: AsyncSession,
        checkout: Checkout,
        billing_reason: OrderBillingReasonInternal,
        payment: Payment | None = None,
        subscription: Subscription | None = None,
    ) -> Order:
        customer = checkout.customer
        if customer is None:
            raise MissingCheckoutCustomer(checkout)

        items: list[OrderItem] = []
        if has_product_checkout(checkout):
            prices = checkout.prices[checkout.product_id]
            for price in prices:
                # Don't create an item for metered prices
                if not is_static_price(price):
                    continue
                if is_custom_price(price):
                    item = OrderItem.from_price(price, 0, checkout.amount)
                else:
                    item = OrderItem.from_price(price, 0, seats=checkout.seats)
                items.append(item)

        discount_amount = checkout.discount_amount

        # Retrieve tax data
        tax_amount = checkout.tax_amount or 0
        taxability_reason = None
        tax_rate: TaxRate | None = None
        tax_id = customer.tax_id
        if checkout.tax_processor_id is not None:
            calculation = await stripe_service.get_tax_calculation(
                checkout.tax_processor_id
            )
            assert tax_amount == calculation.tax_amount_exclusive
            assert len(calculation.tax_breakdown) > 0
            if len(calculation.tax_breakdown) > 1:
                log.warning(
                    "Multiple tax breakdowns found for checkout",
                    checkout_id=checkout.id,
                    calculation_id=calculation.id,
                )
            breakdown = calculation.tax_breakdown[0]
            taxability_reason = TaxabilityReason.from_stripe(
                breakdown.taxability_reason, tax_amount
            )
            tax_rate = from_stripe_tax_rate_details(breakdown.tax_rate_details)

        organization = checkout.organization
        invoice_number = await organization_service.get_next_invoice_number(
            session, organization, customer
        )

        repository = OrderRepository.from_session(session)
        order = await repository.create(
            Order(
                status=OrderStatus.paid,
                subtotal_amount=checkout.amount,
                discount_amount=discount_amount,
                tax_amount=tax_amount,
                currency=checkout.currency,
                billing_reason=billing_reason,
                billing_name=customer.billing_name,
                billing_address=customer.billing_address,
                taxability_reason=taxability_reason,
                tax_id=tax_id,
                tax_rate=tax_rate,
                invoice_number=invoice_number,
                customer=customer,
                product=checkout.product,
                discount=checkout.discount,
                subscription=subscription,
                checkout=checkout,
                user_metadata=checkout.user_metadata,
                custom_field_data=checkout.custom_field_data,
                items=items,
                seats=checkout.seats,
            ),
            flush=True,
        )

        # Link payment and balance transaction to the order
        if payment is not None:
            payment_repository = PaymentRepository.from_session(session)
            assert payment.amount == order.total_amount
            await payment_repository.update(payment, update_dict={"order": order})
            enqueue_job(
                "order.balance", order_id=order.id, charge_id=payment.processor_id
            )

        # Record tax transaction
        if checkout.tax_processor_id is not None:
            transaction = await stripe_service.create_tax_transaction(
                checkout.tax_processor_id, str(order.id)
            )
            await repository.update(
                order, update_dict={"tax_transaction_processor_id": transaction.id}
            )

        await self._on_order_created(session, order)

        return order

    async def create_subscription_order(
        self,
        session: AsyncSession,
        subscription: Subscription,
        billing_reason: OrderBillingReasonInternal,
    ) -> Order:
        async with billing_entry_service.create_order_items_from_pending(
            session, subscription
        ) as items:
            if len(items) == 0:
                raise NoPendingBillingEntries(subscription)

            order_id = uuid.uuid4()
            customer = subscription.customer
            billing_address = customer.billing_address
            product = subscription.product

            subtotal_amount = sum(item.amount for item in items)

            discount = subscription.discount
            discount_amount = 0
            if discount is not None:
                # Discount only applies to cycle and meter items, as prorations
                # use "last month's" discount and so this month's discount
                # shouldn't apply to those.
                discountable_amount = sum(
                    item.amount for item in items if item.discountable
                )
                discount_amount = discount.get_discount_amount(discountable_amount)

            # Calculate tax
            tax_calculation: TaxCalculation | None = None
            taxable_amount = subtotal_amount - discount_amount
            tax_amount = 0
            taxability_reason: TaxabilityReason | None = None
            tax_rate: TaxRate | None = None
            tax_id = customer.tax_id
            tax_calculation_processor_id: str | None = None

            if (
                taxable_amount != 0
                and product.is_tax_applicable
                and billing_address is not None
            ):
                try:
                    tax_calculation = await calculate_tax(
                        order_id,
                        subscription.currency,
                        # Stripe doesn't support calculating negative tax amounts
                        taxable_amount if taxable_amount >= 0 else -taxable_amount,
                        product.tax_code,
                        billing_address,
                        [tax_id] if tax_id is not None else [],
                        subscription.tax_exempted,
                    )
                except (IncompleteTaxLocation, InvalidTaxLocation):
                    log.warning(
                        "Failed to calculate tax for subscription order due to invalid or incomplete address",
                        subscription_id=subscription.id,
                        order_id=order_id,
                        customer_id=customer.id,
                    )
                    tax_amount = 0
                    tax_calculation_processor_id = None
                else:
                    if taxable_amount >= 0:
                        tax_calculation_processor_id = tax_calculation["processor_id"]
                        tax_amount = tax_calculation["amount"]
                    else:
                        # When the taxable amount is negative it's usually due to a credit proration
                        # this means we "owe" the customer money -- but we don't pay it back at this
                        # point. This also means that there's no money transaction going on, and we
                        # don't have to record the tax transaction either.
                        tax_calculation_processor_id = None
                        tax_amount = -tax_calculation["amount"]

                if tax_calculation is not None:
                    taxability_reason = tax_calculation["taxability_reason"]
                    tax_rate = tax_calculation["tax_rate"]

            invoice_number = await organization_service.get_next_invoice_number(
                session, subscription.organization, customer
            )

            total_amount = subtotal_amount - discount_amount + tax_amount
            customer_balance = await wallet_service.get_billing_wallet_balance(
                session, customer, subscription.currency
            )

            # Calculate balance change and applied amount
            if total_amount >= 0:
                # Order is a charge: use customer balance if available
                balance_change = -min(total_amount, customer_balance)
                applied_balance_amount = balance_change
            else:
                # Order is a credit: always add to balance
                balance_change = -total_amount
                # Track how much existing debt was cleared
                if customer_balance < 0:
                    applied_balance_amount = min(-total_amount, -customer_balance)
                else:
                    applied_balance_amount = 0

            repository = OrderRepository.from_session(session)
            order = await repository.create(
                Order(
                    id=order_id,
                    status=OrderStatus.pending,
                    subtotal_amount=subtotal_amount,
                    discount_amount=discount_amount,
                    tax_amount=tax_amount,
                    applied_balance_amount=applied_balance_amount,
                    currency=subscription.currency,
                    billing_reason=billing_reason,
                    billing_name=customer.billing_name,
                    billing_address=billing_address,
                    taxability_reason=taxability_reason,
                    tax_id=tax_id,
                    tax_rate=tax_rate,
                    tax_calculation_processor_id=tax_calculation_processor_id,
                    invoice_number=invoice_number,
                    customer=customer,
                    product=subscription.product,
                    discount=discount,
                    subscription=subscription,
                    checkout=None,
                    items=items,
                    user_metadata=subscription.user_metadata,
                    custom_field_data=subscription.custom_field_data,
                ),
                flush=True,
            )

            # Impact customer's balance
            if balance_change != 0:
                await wallet_service.create_balance_transaction(
                    session,
                    customer,
                    balance_change,
                    subscription.currency,
                    order=order,
                )

            # Reset the associated meters, if any
            if billing_reason in {
                OrderBillingReasonInternal.subscription_cycle,
                OrderBillingReasonInternal.subscription_cycle_after_trial,
                OrderBillingReasonInternal.subscription_update,
            }:
                await subscription_service.reset_meters(session, subscription)

            # If the due amount is less or equal than zero, mark it as paid immediately
            if order.due_amount <= 0:
                order = await repository.update(
                    order, update_dict={"status": OrderStatus.paid}
                )
            elif subscription.payment_method_id is None:
                order = await self.handle_payment_failure(session, order)
            else:
                enqueue_job(
                    "order.trigger_payment",
                    order_id=order.id,
                    payment_method_id=subscription.payment_method_id,
                )

            await self._on_order_created(session, order)

            return order

    async def create_trial_order(
        self,
        session: AsyncSession,
        subscription: Subscription,
        billing_reason: Literal[
            OrderBillingReasonInternal.subscription_create,
            OrderBillingReasonInternal.subscription_update,
        ],
        checkout: Checkout | None = None,
    ) -> Order:
        if not subscription.trialing:
            raise SubscriptionNotTrialing(subscription)
        assert subscription.trial_start is not None
        assert subscription.trial_end is not None

        product = subscription.product
        customer = subscription.customer

        items: list[OrderItem] = [
            OrderItem.from_trial(
                product, subscription.trial_start, subscription.trial_end
            )
        ]

        organization = subscription.organization
        invoice_number = await organization_service.get_next_invoice_number(
            session, organization, customer
        )

        repository = OrderRepository.from_session(session)
        order = await repository.create(
            Order(
                status=OrderStatus.paid,
                subtotal_amount=sum(item.amount for item in items),
                discount_amount=0,
                tax_amount=0,
                currency=subscription.currency,
                billing_reason=billing_reason,
                billing_name=customer.billing_name,
                billing_address=customer.billing_address,
                taxability_reason=None,
                tax_id=customer.tax_id,
                tax_rate=None,
                invoice_number=invoice_number,
                customer=customer,
                product=product,
                discount=None,
                subscription=subscription,
                checkout=checkout,
                user_metadata=subscription.user_metadata,
                custom_field_data=subscription.custom_field_data,
                items=items,
            ),
            flush=True,
        )

        await self._on_order_created(session, order)

        return order

    async def create_wallet_order(
        self,
        session: AsyncSession,
        wallet_transaction: WalletTransaction,
        payment: Payment | None,
    ) -> Order:
        wallet = wallet_transaction.wallet
        items: list[OrderItem] = [
            OrderItem.from_wallet(wallet, wallet_transaction.amount)
        ]

        customer = wallet.customer
        billing_address = customer.billing_address

        subtotal_amount = sum(item.amount for item in items)

        # Retrieve tax data
        tax_amount = wallet_transaction.tax_amount or 0
        taxability_reason = None
        tax_rate: TaxRate | None = None
        tax_id = customer.tax_id
        if wallet_transaction.tax_calculation_processor_id is not None:
            calculation = await stripe_service.get_tax_calculation(
                wallet_transaction.tax_calculation_processor_id
            )
            assert tax_amount == calculation.tax_amount_exclusive
            assert len(calculation.tax_breakdown) > 0
            breakdown = calculation.tax_breakdown[0]
            taxability_reason = TaxabilityReason.from_stripe(
                breakdown.taxability_reason, tax_amount
            )
            tax_rate = from_stripe_tax_rate_details(breakdown.tax_rate_details)

        invoice_number = await organization_service.get_next_invoice_number(
            session, wallet.organization, wallet.customer
        )

        repository = OrderRepository.from_session(session)
        order = await repository.create(
            Order(
                status=OrderStatus.paid,
                subtotal_amount=subtotal_amount,
                discount_amount=0,
                tax_amount=tax_amount,
                applied_balance_amount=0,
                currency=wallet.currency,
                billing_reason=OrderBillingReasonInternal.purchase,
                billing_name=customer.billing_name,
                billing_address=billing_address,
                taxability_reason=taxability_reason,
                tax_id=tax_id,
                tax_rate=tax_rate,
                invoice_number=invoice_number,
                customer=customer,
                items=items,
                product=None,
                discount=None,
                subscription=None,
                checkout=None,
            ),
            flush=True,
        )

        # Link payment and balance transaction to the order
        if payment is not None:
            payment_repository = PaymentRepository.from_session(session)
