@Service
public class CouponService implements CouponCommand {

    private final CouponRepository couponRepository;

    public CouponService(
        CouponRepository couponRepository
    ) {
        this.couponRepository = couponRepository;
    }

    @Override
    public void issueCoupon(Long userId) {
        long count = couponRepository.count();

        if (count > MAX_COUPON_COUNT) {
            return;
        }

        couponRepository.save(new Coupon(userId));
    }
}

@Service
public class CouponServiceWithRedis implements CouponCommand {
    private final CouponRepository couponRepository;

    private final CouponCountRepository couponCountRepository;

    public CouponServiceWithRedis(
        CouponRepository couponRepository,
        CouponCountRepository couponCountRepository
    ) {
        this.couponRepository = couponRepository;
        this.couponCountRepository = couponCountRepository;
    }

    @Override
    public void issueCoupon(Long userId) {
        long count = couponCountRepository.increment();

        if (count > MAX_COUPON_COUNT) {
            return;
        }

        couponRepository.save(new Coupon(userId));
    }
}

@Service
public class CouponServiceWithRedisAndKafka implements CouponCommand {
    private final CouponCountRepository couponCountRepository;

    private final CouponCreateProducer couponCreateProducer;

    public CouponServiceWithRedisAndKafka(
        CouponCountRepository couponCountRepository,
        CouponCreateProducer couponCreateProducer) {
        this.couponCountRepository = couponCountRepository;
        this.couponCreateProducer = couponCreateProducer;
    }

    @Override
    public void issueCoupon(Long userId) {
        long count = couponCountRepository.increment();

        if (count > MAX_COUPON_COUNT) {
            return;
        }

        couponCreateProducer.createCoupon(userId);
    }
}

@Component
public class CouponCreateProducer {

    public static final String COUPON_CREATE_TOPIC = "coupon-create";

    private final KafkaTemplate<String, Long> kafkaTemplate;

    public CouponCreateProducer(KafkaTemplate<String, Long> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void createCoupon(Long userId) {
        kafkaTemplate.send(COUPON_CREATE_TOPIC, userId);
    }
}

@Component
@Slf4j
public class CouponCreateConsumer {

    private static final String COUPON_CREATE_TOPIC = "coupon-create";

    private final CouponRepository couponRepository;

    private final FailedEventRepository failedEventRepository;

    public CouponCreateConsumer(CouponRepository couponRepository,
        FailedEventRepository failedEventRepository) {
        this.couponRepository = couponRepository;
        this.failedEventRepository = failedEventRepository;
    }

    @KafkaListener(topics = CouponCreateConsumer.COUPON_CREATE_TOPIC, groupId = "coupon-consumer-group-1")
    public void consume(Long userId) {
        try {
            log.info("Consumed message: {}", userId);
            couponRepository.save(new Coupon(userId));
        } catch (Exception e) {
            log.error("Failed to consume message: {}", userId);
            failedEventRepository.save(new FailedEvent(userId));
        }
    }
}

@Service
public class CouponLimitedServiceWithRedisAndKafka implements CouponCommand {
    private final CouponCountRepository couponCountRepository;

    private final CouponCreateProducer couponCreateProducer;

    private final CouponAppliedUserRepository couponAppliedUserRepository;

    public CouponLimitedServiceWithRedisAndKafka(
        CouponCountRepository couponCountRepository,
        CouponCreateProducer couponCreateProducer,
        CouponAppliedUserRepository couponAppliedUserRepository) {
        this.couponCountRepository = couponCountRepository;
        this.couponCreateProducer = couponCreateProducer;
        this.couponAppliedUserRepository = couponAppliedUserRepository;
    }

    @Override
    public void issueCoupon(Long userId) {
        couponAppliedUserRepository.add(userId);

        long count = couponCountRepository.increment();

        if (count > MAX_COUPON_COUNT) {
            return;
        }

        couponCreateProducer.createCoupon(userId);
    }
}

@Repository
public class CouponAppliedUserRepository {
    private final RedisTemplate<String, String> redisTemplate;

    public CouponAppliedUserRepository(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void add(Long userId) throws IllegalStateException {
        Long result = redisTemplate
            .opsForSet()
            .add("coupon-applied-user", userId.toString());

        if (result == null || result != 1) {
            throw new IllegalStateException("이미 쿠폰을 받은 사용자입니다.");
        }
    }
}