PaymentEventConsumer.java

package com.checkout.paymentgw.query.projection;

import com.checkout.paymentgw.events.PaymentEvent;
import com.checkout.paymentgw.events.Topics;
import com.checkout.paymentgw.query.domain.PaymentView;
import com.checkout.paymentgw.query.domain.PaymentViewRepository;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

/**
 * Idempotent consumer of {@code payments.events}. Each event is upserted into
 * {@code payments_view} keyed by {@code paymentId}.
 *
 * <p>Idempotency is achieved by primary-key upsert plus a sequence/timestamp guard.
 * For replay, we can drop the read DB and let the consumer rebuild the view from the
 * earliest offset.
 */
@Component
public class PaymentEventConsumer {

    private static final Logger LOG = LoggerFactory.getLogger(PaymentEventConsumer.class);

    private final PaymentViewRepository repository;
    private final ObjectMapper mapper;

    public PaymentEventConsumer(PaymentViewRepository repository, ObjectMapper mapper) {
        this.repository = repository;
        this.mapper = mapper;
    }

    @KafkaListener(topics = Topics.PAYMENTS_EVENTS, groupId = "payment-query-service")
    @Transactional
    public void onMessage(String payload) {
        try {
            PaymentEvent ev = mapper.readValue(payload, PaymentEvent.class);
            apply(ev);
        } catch (Exception e) {
            LOG.error("failed to project event payload={}", payload, e);
            // Don't rethrow — would block the partition. In production, send to a DLT.
        }
    }

    private void apply(PaymentEvent ev) {
        if (ev.paymentId() == null) {
            LOG.warn("event without paymentId — skipping; eventId={}", ev.eventId());
            return;
        }

        PaymentView existing = repository.findById(ev.paymentId()).orElse(null);
        // ordering guard — only overwrite with a newer event
        if (existing != null && existing.getOccurredAt() != null
                && ev.occurredAt() != null
                && existing.getOccurredAt().isAfter(ev.occurredAt())) {
            return;
        }

        PaymentView view = PaymentView.builder()
                .id(ev.paymentId())
                .merchantId(nullSafe(ev.merchantId(), existing == null ? null : existing.getMerchantId()))
                .status(nullSafe(ev.status(), existing == null ? null : existing.getStatus()))
                .cardNumberLastFour(nullSafe(ev.cardNumberLastFour(), existing == null ? null : existing.getCardNumberLastFour()))
                .currency(nullSafe(ev.currency(), existing == null ? null : existing.getCurrency()))
                .amount(ev.amount() != null ? ev.amount() : (existing == null ? null : existing.getAmount()))
                .occurredAt(ev.occurredAt())
                .expiryMonth(existing == null ? null : existing.getExpiryMonth())
                .expiryYear(existing == null ? null : existing.getExpiryYear())
                .build();
        repository.save(view);
    }

    private static String nullSafe(String a, String b) {
        return a != null ? a : b;
    }
}