package com.assu.study.chap10.adapter.lock;

import java.util.Objects;

/**
 * 분산락을 사용하기 위한 레디스 키 디자인 클래스
 */
public class LockKey {
  private static final String PREFIX = "LOCK::";
  private final Long eventHotelId;

  public LockKey(Long eventHotelId) {
    if (Objects.isNull(eventHotelId)) {
      throw new IllegalArgumentException("eventHotelId can't be null.");
    }
    this.eventHotelId = eventHotelId;
  }

  public static LockKey from(Long eventHotelId) {
    return new LockKey(eventHotelId);
  }

  // 레디스에 저장된 키를 LockKey 객체로 역직렬화할 때 사용
  public static LockKey fromString(String key) {
    String idToken = key.substring(0, PREFIX.length());
    Long eventHotelId = Long.valueOf(idToken);

    return LockKey.from(eventHotelId);
  }

  // LockKey 객체를 레디스의 키로 저장할 때 직렬화 과정에서 사용할 메서드
  // LOCK::eventHotelId 문자열 포맷으로 직렬화되어 저장됨
  @Override
  public String toString() {
    return PREFIX + eventHotelId;
  }
}

package com.assu.study.chap10.adapter.lock;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.time.Duration;

@Slf4j
@Component
public class LockAdapter {
  private final RedisTemplate<LockKey, Long> lockRedisTemplate;
  private final ValueOperations<LockKey, Long> lockOperation;

  public LockAdapter(RedisTemplate<LockKey, Long> lockRedisTemplate) {
    this.lockRedisTemplate = lockRedisTemplate;
    this.lockOperation = lockRedisTemplate.opsForValue();
  }

  // 레디스에 락 생성
  // hotelId 를 사용하여 LockKey 객체를 생성하고, 레디스 key 에 저장
  // 이 때 value 는 userId
  public Boolean holdLock(Long hotelId, Long userId) {
    LockKey lockKey = LockKey.from(hotelId);
    // setIfAbsent() 는 레디스 key 와 매핑되는 값이 없을때만 레디스 데이터 생성
    // 데이터가 없으면 Boolean.TRUE 리턴
    // 즉, Boolean.FALSE 를 리턴하면 레디스에 이미 데이터가 있음을 의미하므로 분산락이 있다는 의미이기 때문에
    // 공유 자원에 작업하지 않음
    // 레디스의 유효 기간은 10초로 설정
    return lockOperation.setIfAbsent(lockKey, userId, Duration.ofSeconds(10));
  }

  // 레디스에 락이 있는지 확인
  public Long checkLock(Long hotelId) {
    LockKey lockKey = LockKey.from(hotelId);
    return lockOperation.get(lockKey);
  }

  // 레디스에서 락을 삭제
  public void clearLock(Long hotelId) {
    lockRedisTemplate.delete(LockKey.from(hotelId));
  }
}

package com.assu.study.chap10.adapter.lock;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@RequiredArgsConstructor
@Slf4j
@Service
public class LockEventService {
  private LockAdapter lockAdapter;

  @Transactional(timeout = 10)  // 트랜잭션의 타임아웃을 10초로 설정
  public Boolean attendEvent(Long hotelId, Long userId) {

    // 레디스에 분산락이 없는지 확인
    // holdLock() 이 false 를 리턴하면 다른 인스턴스나 스레드로 락이 생성되었음을 의미
    // 따라서 attendEvent() 도 false 를 리턴
    if (!lockAdapter.holdLock(hotelId, userId)) {
      return false;
    }

    EventHotelEntity eventHotelEntity = eventHotelRepository.findByHotelId(hotelId);

    // DB 에서 hotelId 와 일치하는 엔티티 객체 조회
    // nonEmptyUser() 는 엔티티 객체에 winnerUserId 의 null 여부 확인
    // null 이 아니면 다른 사용자가 이미 선착순 이벤트에 성공한 것이므로 false 리턴
    // 따라서 attendEvent() 도 false 를 리턴
    if (eventHotelEntity.nonEmptyUser()) {
      return false;
    }

    // 엔티티에 winnerUserId 를 설정하는 winner() 실행 후 DB 에 저장
    eventHotelEntity.winner(userId);
    eventHotelRepository.save(eventHotelEntity);

    return true;
  }
}

package com.assu.study.chap10.adapter.lock;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.List;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;

@Slf4j
@SpringBootTest
class LockAdapterTest {

  private final Long firstUserId = 1L;
  private final Long secondUserId = 2L;
  private final Long thirdUserId = 3L;

  @Autowired
  private LockAdapter lockAdapter;

  @Test
  @DisplayName("firstUserId 가 락 선점")
  public void testLock() {
    final Long hotelId = 1111L;

    boolean isSuccess = lockAdapter.holdLock(hotelId, firstUserId);
    Assertions.assertTrue(isSuccess);

    isSuccess = lockAdapter.holdLock(hotelId, secondUserId);
    Assertions.assertFalse(isSuccess);

    Long holderId = lockAdapter.checkLock(hotelId);
    Assertions.assertEquals(firstUserId, holderId);
  }

  @Test
  @DisplayName("동시에 3이 락을 선점하지만 1명만 락을 잡음")
  public void testConcurrentAccess() throws InterruptedException {
    final Long hotelId = 9999L;

    // CyclicBarrier 의 인자를 3으로 설정
    // 각 스레드는 CyclicBarrier 공유 객체의 await() 메서드를 호출하고, 3번 호출되면 CyclicBarrier 는 스레드 실행함 
    CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
    
    // Runnable 을 구현하는 Accessor 를 사용하여 3개의 스레드 실행
    // 공유 자원인 hotelId 와 사용자 아이디를 각각 인자로 입력
    // 그리고 CyclicBarrier 기능을 공유하려고 인자로 넘김
    new Thread(new Accessor(hotelId, firstUserId, cyclicBarrier)).start();
    new Thread(new Accessor(hotelId, secondUserId, cyclicBarrier)).start();
    new Thread(new Accessor(hotelId, thirdUserId, cyclicBarrier)).start();

    TimeUnit.SECONDS.sleep(1);
    Long holderId = lockAdapter.checkLock(hotelId);

    log.info("holderId: {}", holderId);

    // 스레드를 실행하고 1초 후 락의 유무 확인
    // 단, firstUserId, secondUserId, thirdUserId 중 하나가 레디스 value 에 저장되어 있음을 검증함
    Assertions.assertTrue(List.of(firstUserId, secondUserId, thirdUserId).contains(holderId));
    lockAdapter.clearLock(hotelId);
  }

  // 최대한 동시에 LockAdapter 의 holdLock() 을 실행하고자 CyclicBarrier 사용
  class Accessor implements Runnable {
    private final Long hotelId;
    private final Long userId;
    private final CyclicBarrier cyclicBarrier;

    public Accessor(Long hotelId, Long userId, CyclicBarrier cyclicBarrier) {
      this.hotelId = hotelId;
      this.userId = userId;
      this.cyclicBarrier = cyclicBarrier;
    }

    // 인자로 받은 cyclicBarrier 객체를 사용하여 await() 메서드를 실행하고,
    // lockAdapter.holdLock() 메서드 실행
    // await() 가 3번 호출될 때까지는 모든 스레드는 대기하고,
    // 3번 호출된 시점에 모든 스레드가 한 번에 lockAdapter.holdLock() 를 호출함
    @Override
    public void run() {
      try {
        cyclicBarrier.await();
        lockAdapter.holdLock(hotelId, userId);
      } catch (Exception e) {
        throw new RuntimeException(e);
      }

    }
  }
}

package com.assu.study.chap10.config;

import com.assu.study.chap10.adapter.lock.LockKey;
import com.assu.study.chap10.adapter.lock.LockKeySerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericToStringSerializer;

@Configuration
public class LockConfig {

  @Bean
  public RedisConnectionFactory lockRedisConnectionFactory() {
    RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration("127.0.0.1", 6379);

    // 레디스의 데이터베이스 번호 설정
    // 레디스 서버는 내부에서 16개의 데이터베이스를 구분해서 운영 가능하며, 0~15번까지의 데이터베이스를 가짐
    // 개발 환경에서 장비를 효츌적으로 사용하기 위해 데이터베이스를 구분하여 각 컴포넌트에 할당해서 운영 가능
//    configuration.setDatabase(0);
//    configuration.setUsername("username");
//    configuration.setPassword("password");

    return new LettuceConnectionFactory(configuration);
  }

  // redisTemplate 가 2개 (hotelCacheRedisTemplate, lockRedisTemplate) 인 경우 하나는
  // 디폴트 이름인 redisTemplate 로 해주어야 하나 봄 (lockRedisTemplate 가 아닌)
  @Bean(name = "redisTemplate")
  public RedisTemplate<LockKey, Long> lockRedisTemplate() {
    RedisTemplate<LockKey, Long> lockRedisTemplate = new RedisTemplate<>();

    // 위에서 생성한 RedisConnectionFactory 스프링 빈을 사용하여 RedisTemplate 객체 생성
    lockRedisTemplate.setConnectionFactory(lockRedisConnectionFactory());
    // key 와 value 값을 직렬화/역직렬화하는 RedisSerializer 구현체 설정
    lockRedisTemplate.setKeySerializer(new LockKeySerializer());
    lockRedisTemplate.setValueSerializer(new GenericToStringSerializer<>(Long.class));

    return lockRedisTemplate;
  }
}

package com.assu.study.chap10.adapter.lock;

import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Objects;

public class LockKeySerializer implements RedisSerializer<LockKey> {
  private final Charset UTF_8 = StandardCharsets.UTF_8;

  @Override
  public byte[] serialize(LockKey lockKey) throws SerializationException {
    // 레디스 데이터 중 key 는 null 이 될 수 없음
    if (Objects.isNull(lockKey)) {
      throw new SerializationException("lockKey is null.");
    }
    // HotelCacheKey 가 직렬화되면 byte[] 를 리턴해야 함
    // 이 때 Charset 를 설정하여 byte[] 로 변환하는 것이 좋음
    return lockKey.toString().getBytes(UTF_8);
  }

  @Override
  public LockKey deserialize(byte[] bytes) throws SerializationException {
    if (Objects.isNull(bytes)) {
      throw new SerializationException("bytes is null.");
    }
    // 레디스의 key 데이터는 byte[] 이므로 적절히 변환하여 HotelCacheKey 객체를 생성하여 리턴
    return LockKey.fromString(new String(bytes, UTF_8));
  }
}