package com.food.order.system.order.messaging.listener.kafka;

import com.food.order.sysyem.ports.input.message.listener.restaurantapproval.RestaurantApprovalResponseMessageListener;
import com.food.order.system.kafka.consumer.KafkaConsumer;
import com.food.order.system.order.messaging.mapper.OrderMessagingDataMapper;
import com.food.ordering.system.kafka.order.avro.model.OrderApprovalStatus;
import com.food.ordering.system.kafka.order.avro.model.RestaurantApprovalResponseAvroModel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@RequiredArgsConstructor
@Component
public class RestaurantApprovalResponseKafkaListener implements KafkaConsumer<RestaurantApprovalResponseAvroModel> {

    private final OrderMessagingDataMapper orderMessagingDataMapper;
    private final RestaurantApprovalResponseMessageListener responseMessageListener;

    @Override
    @KafkaListener(
            id = "${kafka-consumer-config.restaurant-approval-consumer-group-id}",
            topics = "${order-service.restaurant-approval-response-topic-name}"
    )
    public void receive(@Payload List<RestaurantApprovalResponseAvroModel> messages,
                        @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) List<String> keys,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                        @Header(KafkaHeaders.OFFSET) List<Long> offSets) {

        log.info("{} number of restaurant approval responses received with keys : {} , partitions : {} , offsets : {}",
                messages.size(), keys, partitions, offSets);

        messages.forEach(message -> {
            if (message.getOrderApprovalStatus().equals(OrderApprovalStatus.APPROVED)) {
                log.info("Processing successful restaurant approval response for order id: {}", message.getOrderId());
                responseMessageListener.orderApproved(orderMessagingDataMapper.
                        restaurantApprovalResponseAvroModelToRestaurantApprovalResponse(message));
            }
            else if (message.getOrderApprovalStatus().equals(OrderApprovalStatus.REJECTED)) {
                log.info("Processing failed restaurant approval response for order id: {}", message.getOrderId());
                responseMessageListener.orderRejected(orderMessagingDataMapper.
                        restaurantApprovalResponseAvroModelToRestaurantApprovalResponse(message));
            }
        });
    }
}
