package com.assu.study.order.command.application;

import com.assu.study.order.NoOrderException;
import com.assu.study.order.command.domain.*;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@RequiredArgsConstructor
@Service
public class CancelOrderService {
  private final OrderRepository orderRepository;
  private final CancelPolicy cancelPolicy;

  @Transactional
  public void cancel(OrderNo orderNo, Canceller canceller) {
    Order order = orderRepository.findById(orderNo)
            .orElseThrow(() -> new NoOrderException());
    if (!cancelPolicy.hasCancellationPermission(order, canceller)) {
      throw new NoCancellablePermission();
    }

    // order.cancel() 에서 OrderCanceledEvent 발생
    // 만일 외부 연동 과정에서 예외 발생 시 트랜잭션 처리를 어떻게 할 것인지?
    order.cancel(); 
  }
}

package com.assu.study.order.infra;

import com.assu.study.order.command.application.RefundService;
import com.assu.study.order.command.domain.OrderCanceledEvent;
import lombok.RequiredArgsConstructor;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;

// OrderCanceledEvent 처리하는 핸들러
@RequiredArgsConstructor
@Service
public class OrderCanceledEventHandler {
    private final RefundService refundService;

    @EventListener(OrderCanceledEvent.class)
    public void handle(OrderCanceledEvent event) {
        
        // refundService.refund() 가 느려지거나 예외가 발생하면 어떻게 처리할지?
        refundService.refund(event.getOrderNumber());
    }
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;

@EnableAsync    // 비동기 기능 활성화
@Configuration
public class SampleAsyncExecutionConfig implements AsyncConfigurer {
  // 프레임워크가 스레드 풀을 설정할 때 사용하는 콜백 메서드
  @Override
  public Executor getAsyncExecutor() {
    return getExecutor();
  }

  private Executor getExecutor() {
    ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    // 기본 개수는 10개
    threadPoolTaskExecutor.setCorePoolSize(10);
    // 최대 10개까지 늘어남
    threadPoolTaskExecutor.setMaxPoolSize(10);
    threadPoolTaskExecutor.setThreadNamePrefix("asyncExecutor-");
    // 컨테이너가 모든 속성값을 적용한 후 initialize() 호출
    threadPoolTaskExecutor.afterPropertiesSet();
    return threadPoolTaskExecutor;
  }

  @Bean("asyncExecutor")
  public Executor asyncExecutor() {
    return getExecutor();
  }
}

package com.assu.study.order.infra;

import com.assu.study.order.command.application.RefundService;
import com.assu.study.order.command.domain.OrderCanceledEvent;
import lombok.RequiredArgsConstructor;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

// OrderCanceledEvent 처리하는 핸들러
@RequiredArgsConstructor
@Service
public class OrderCanceledEventHandler {
    private final RefundService refundService;

    @Async
    @EventListener(OrderCanceledEvent.class)
    public void handle(OrderCanceledEvent event) {
        refundService.refund(event.getOrderNumber());
    }
}

package com.assu.study.eventstore.api;

import lombok.Getter;


// 이벤트 저장소에 보관할 데이터
@Getter
public class EventEntry {
  private Long id;    // 이벤트 식별자
  private String type;    // 이벤트 타입
  private String contentType; // 직렬화한 데이터 형식
  private String payload; // 이벤트 데이터 자체
  private long timestamp; // 이벤트 시간

  public EventEntry(String type, String contentType, String payload) {
    this.type = type;
    this.contentType = contentType;
    this.payload = payload;
    this.timestamp = System.currentTimeMillis();
  }

  public EventEntry(Long id, String type, String contentType, String payload) {
    this.id = id;
    this.type = type;
    this.contentType = contentType;
    this.payload = payload;
    this.timestamp = System.currentTimeMillis();
  }
}

package com.assu.study.eventstore.api;

import java.util.List;

// 이벤트 객체를 직렬화하여 EventEntry 의 payload 에 저장하는 인터페이스
// 이벤트를 저장하고 조회하는 인터페이스 제공
public interface EventStore {
    void save(Object event);

    List<EventEntry> get(long offset, long limit);
}

package com.assu.study.eventstore.infra;

import com.assu.study.eventstore.api.EventEntry;
import com.assu.study.eventstore.api.EventStore;
import com.assu.study.eventstore.api.PayloadConvertException;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import org.springframework.http.MediaType;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import java.sql.Timestamp;
import java.util.List;


// JDBC 를 이용한 EventStore 구현 클래스
@RequiredArgsConstructor
@Component
public class JdbcEventStore implements EventStore {

  private final ObjectMapper objectMapper;
  private final JdbcTemplate jdbcTemplate;

  @Override
  public void save(Object event) {
    // EventEntry 객체 생성
    // event 객체를 JSON 문자열로 변환하여 payload 에 전달
    EventEntry entry = new EventEntry(event.getClass().getName(), MediaType.APPLICATION_JSON_VALUE, toJson(event));

    jdbcTemplate.update(
            "INSERT INTO evententry (type, content_type, payload, timestamp) " +
                    "VALUES (?, ?, ?, ?)",
            ps -> {
              ps.setString(1, entry.getType());
              ps.setString(2, entry.getContentType());
              ps.setString(3, entry.getPayload());
              ps.setTimestamp(4, new Timestamp(entry.getTimestamp()));
            }
    );
  }

  private String toJson(Object event) {
    try {
      return objectMapper.writeValueAsString(event);
    } catch (JsonProcessingException e) {
      throw new PayloadConvertException(e);
    }
  }

  @Override
  public List<EventEntry> get(long offset, long limit) {
    return jdbcTemplate.query(
            "SELECT * FROM evententry ORDER BY id ASC LIMIT ?, ?",
            ps -> {
              ps.setLong(1, offset);
              ps.setLong(2, limit);
            },
            (rs, rowNum) -> new EventEntry(
                    rs.getLong("id"),
                    rs.getString("type"),
                    rs.getString("content_type"),
                    rs.getString("payload"),
                    rs.getTimestamp("timestamp").getTime()
            )
    );
  }
}

package com.assu.study.eventstore.api;

public class PayloadConvertException extends RuntimeException {
    public PayloadConvertException(Exception e) {
        super(e);
    }
}

package com.assu.study.common.event;

import com.assu.study.eventstore.api.EventStore;
import lombok.RequiredArgsConstructor;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

@RequiredArgsConstructor
@Component
public class EventStoreHandler {
    private final EventStore eventStore;

    // 모든 이벤트가 공통으로 갖는 프로퍼티를 갖는 공통 추상 클래스인 Event 타입을 상속받은
    // 이벤트 타입만 이벤트 저장소에 보관하기 위해 @EventListener 애너테이션에 값으로 Event.class 설정
    @EventListener(Event.class)
    public void handle(Event event) {
        eventStore.save(event);
    }
}

package com.assu.study.integration;

import com.assu.study.eventstore.api.EventEntry;
import com.assu.study.eventstore.api.EventStore;
import lombok.RequiredArgsConstructor;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;

@RequiredArgsConstructor
@Component
public class EventForwarder {
    private static final int DEFAULT_LIMIT_SIZE = 100;
    private final EventStore eventStore;    // 이벤트를 저장하고 조회하는 인터페이스
    private final OffsetStore offsetStore;
    private final EventSender eventSender;
    private int limitSize = DEFAULT_LIMIT_SIZE;

    @Scheduled(initialDelay = 1000L, fixedDelay = 1000L)
    public void getAndSend() {
        // 읽어올 이벤트의 다음 offset 조회
        long nextOffset = getNextOffset();

        // 이벤트 저장소에서 offset 부터 limitSize 만큼 이벤트 구함
        List<EventEntry> events = eventStore.get(nextOffset, limitSize);
        
        // 구한 이벤트가 존재하는지 검사
        if (!events.isEmpty()) {
            // 이벤트 전송
            int processedCount = sendEvent(events);

            // 처리한 이벤트 개수가 0보다 크면 ㅁ다음에 읽어올 offset 저장
            if (processedCount > 0) {
                saveNextOffset(nextOffset + processedCount);
            }
        }
    }

    private long getNextOffset() {
        return offsetStore.get();
    }

    private int sendEvent(List<EventEntry> events) {
        int processedCount = 0;

        // 이벤트를 차례대로 발송함
        // 예외가 발생하면 이벤트 전송을 멈추고 전송에 성공한 이벤트의 개수 리턴
        // 따라서 다음번 getAndSend() 메서드를 실행하면 
        // 마지막으로 전송에 성공한 이벤트의 다음 이벤트부터 읽어와서 전송 시도
        try {
            for (EventEntry entry : events) {
                eventSender.send(entry);
                processedCount++;
            }
        } catch (Exception e) {
            // 로깅 처리
        }
        return processedCount;
    }

    private void saveNextOffset(long nextOffset) {
        offsetStore.update(nextOffset);
    }
}

package com.assu.study.integration;

// OffsetStore 를 구현하는 클래스는 offset 값을 DB 에 저장하거나 로컬 파일에 보관하여 
// 마지막 offset 값을 물리적 저장소에 보관해야 함
public interface OffsetStore {
    long get();

    void update(long nextOffset);
}

package com.assu.study.integration.infra;

import com.assu.study.integration.OffsetStore;
import org.springframework.stereotype.Component;

@Component
public class MemoryOffsetStore implements OffsetStore {
    private long nextOffset = 0;

    @Override
    public long get() {
        return nextOffset;
    }

    @Override
    public void update(long nextOffset) {
        this.nextOffset = nextOffset;
    }
}

package com.assu.study.integration;

import com.assu.study.eventstore.api.EventEntry;

// 이 인페이스를 구현하는 클래스는 send() 메서드에서 외부 메시징 시스템에 이벤트를 전송하거나
// 원하는 핸들러에 이벤트를 전달하면 됨

// 이벤트 처리 중 예외가 발생하면 그대로 전파하여 다으 주기에 getAndSend() 메서드를 실행할 때 재처리할 수 있도록 해야 함
public interface EventSender {
    void send(EventEntry event);
}

package com.assu.study.integration.infra;

import com.assu.study.eventstore.api.EventEntry;
import com.assu.study.integration.EventSender;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class SysoutEventSender implements EventSender {
    @Override
    public void send(EventEntry event) {
        log.info("EventSender send event: {}", event);
    }
}

package com.assu.study.order.infra;

import com.assu.study.order.command.application.RefundService;
import com.assu.study.order.command.domain.OrderCanceledEvent;
import lombok.RequiredArgsConstructor;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;

// OrderCanceledEvent 처리하는 핸들러
@RequiredArgsConstructor
@Service
public class OrderCanceledEventHandler {
    private final RefundService refundService;

    @Async
    //@EventListener(OrderCanceledEvent.class)
    @TransactionalEventListener(
            classes = OrderCanceledEvent.class,
            phase = TransactionPhase.AFTER_COMMIT   // 트랜잭션 커밋이 성공한 뒤 핸들러 메서드 실행
    )
    public void handle(OrderCanceledEvent event) {
        refundService.refund(event.getOrderNumber());
    }
}