class DisplayCacheModule(
    private val circuitBreaker: CircuitBreaker,
    private val redisTemplate: RedisTemplate<String, Any?>,
    private val objectMapper: ObjectMapper
) {
    private val logger = LoggerFactory.getLogger(DisplayCacheModule::class.java)

    companion object {
      private val SAFE_NULL_OBJECT = Object()

      /**
       * 파라미터 기반의 캐시 키를 생성
       */
      fun makeDefaultCacheKey(
         baseCacheKey: String,
         paramValuesForKey: List<String> = emptyList(),
         dateTimeSuffixType: DateTimeKeySuffixType = DateTimeKeySuffixType.NOT_USE,
      ): String {
         // paramValuesForKey 처리
         val parameterCacheKey = paramValuesForKey.joinToString(separator = "") { it }
            .let { if (it.isNotBlank()) ":$it" else "" } // ":xyz" or ""

         // dateTimeKeySuffix 처리
         val dateTimeKeySuffix = if (dateTimeSuffixType != DateTimeKeySuffixType.NOT_USE) {
            dateTimeSuffixType.getSuffixString(LocalDateTime.now())
               .let { if (it.isNotBlank()) "-$it" else "" } // "-yyyyMMdd" or ""
         } else ""

         // 최종적으로 base, parameterCacheKey, dateTimeKeySuffix를 조립하여 반환
         return baseCacheKey + parameterCacheKey + dateTimeKeySuffix
      }
   }

   /**
    * 캐시를 읽어옵니다. 캐시를 획득하지 못한 경우 함수 본체를 수행한 결과를 캐싱 후 반환합니다.
    * @param redisDataType 레디스 데이터 타입
    * @param cacheKey 캐시 키
    * @param hashKey 해시 키 ([RedisDataType.HASH]의 경우 사용)
    * @param ttlBySeconds 데이터의 TTL(초)
    * @param putDataAfterProceed fallback로직 수행 후 데이터 적재 수행 여부
    * @param throwWhenReadFail 읽기작업 실패 시 예외 던지기 여부
    * @param fallbackFunction 읽기작업 실패 시 수행할 fallback 함수
    * @param fallbackReturnType fallback함수의 반환 자료형
    */
   fun fetchAndPut(
      redisDataType: RedisDataType,
      cacheKey: String,
      hashKey: String,
      ttlBySeconds: Long,
      putDataAfterProceed: Boolean = true,
      throwWhenReadFail: Boolean = false,
      fallbackFunction: Any?.() -> Any?,
      fallbackReturnType: Class<*>
    ): Any? {
      // 2.레디스로부터 데이터 읽기 시도
      val fetchResult = tryReadFromRedis(
         redisDataType = redisDataType,
         cacheKey = cacheKey,
         hashKey = hashKey,
         // 3-3.throwWhenReadFail가 true면 데이터 읽기 실패 예외를 던지고 종료
         throwWhenReadFail = throwWhenReadFail,
         returnType = fallbackReturnType
      )

      // 3-1.읽기 성공 시 데이터 반환
      if (fetchResult !== SAFE_NULL_OBJECT) {
         return fetchResult
      }

      // 3-4.데이터 획득을 위한 fallback함수 수행
      val fallbackResult = fallbackFunction()

      // 4-2.fallback함수 수행 결과를 반환
      if (!putDataAfterProceed || isNullOrEmptyData(fallbackResult)) {
         return fallbackResult
      }

      // 4-1.레디스에 데이터 적재 시도
      tryWriteIntoRedis(
         redisDataType = redisDataType,
         cacheKey = cacheKey,
         hashKey = hashKey,
         ttlBySeconds = ttlBySeconds,
         writeData = fallbackResult!!, // isNullOrEmptyData(..)에서 null 검증을 수행
      )

      // 7.적재 후 fallback함수 수행 결과를 반환
      return fallbackResult
   }

   /**
    * 빈 데이터인지 검증합니다.
    */
   private fun isNullOrEmptyData(data: Any?): Boolean {
      return when (data) {
         null -> true
         is Collection<*> -> data.size == 0
         is Map<*, *> -> data.size == 0
         else -> data.toString().isBlank()
      }
    }

   /**
    * Resilience4j적용, 레디스에서 데이터를 읽어옵니다.
    * 데이터 읽기에 실패하거나 데이터가 없는 경우(null) [SAFE_NULL_OBJECT]를 반환합니다.
    */
   private fun tryReadFromRedis(
      redisDataType: RedisDataType,
      cacheKey: String,
      hashKey: String,
      throwWhenReadFail: Boolean,
      returnType: Class<*>
   ): Any { /* 생략 ... */ }

   /**
    * Resilience4j적용, 레디스에 데이터를 쓰고 예외 발생 시 핸들링을 수행합니다.
    */
   private fun tryWriteIntoRedis(
      redisDataType: RedisDataType,
      cacheKey: String,
      hashKey: String,
      ttlBySeconds: Long,
      writeData: Any
   ): Boolean { /* 생략 ... */ }
}

@Aspect
@Component
@Order(Ordered.LOWEST_PRECEDENCE)
class DisplayCachingAspect(
   private val circuitBreaker: CircuitBreaker,
   private val redisTemplate: RedisTemplate<String, Any?>,
   private val objectMapper: ObjectMapper,
) {
   /**
    * joinPoint와 method로부터 @DisplayCachingKey가 붙은 모든 매개변수들을 추출하여 키 값을 생성합니다.
    */
   private fun makeCacheKey(joinPoint: ProceedingJoinPoint, displayCaching: DisplayCaching, method: Method): String {
      // @DisplayCachingParamKey가 있는 파라미터 획득
      val annotatedParamAndValues = method.parameters
         .zip(joinPoint.args) // Pair(first: Parameter, second: Actual value)
         .filter { (param, _) ->
            param.isAnnotationPresent(DisplayCachingKey::class.java) &&
            !param.getAnnotation(DisplayCachingKey::class.java).forHashKey
         }

      // 파라미터 접두사 및 값 변환 일괄 처리
      val prefixProcessedParamValue = annotatedParamAndValues.map {
         val param = it.first
         val annotation = param.getAnnotation(DisplayCachingKey::class.java)
         val prefix = annotation.prefix
         val convertedValue = DisplayCachingKey.convertParamValueToKeyString(it.second)
         prefix + convertedValue
      }

      // 캐시 키 생성
      return DisplayCacheModule.makeDefaultCacheKey(
         baseCacheKey = displayCaching.displayCacheInfo.key,
         paramValuesForKey = prefixProcessedParamValue,
         dateTimeSuffixType = displayCaching.dateTimeKeySuffix
      )
   }

   /**
    * joinPoint와 method로부터 @DisplayCachingKey(forHash = true)인 매개변수를 추출하여
    * 레디스 해시에 사용할 키 값을 생성합니다.
    */
   private fun readHashKeyOrThrow(
      joinPoint: ProceedingJoinPoint,
      method: Method,
      cacheKey: String
   ): String {
      try {
         // @DisplayCachingParamKey가 있는 파라미터중 forHashKey = true인 파라미터 획득
         val hashKeyAnnotatedParam = method.parameters
            .zip(joinPoint.args)
            .first { (param, _) ->
               param.isAnnotationPresent(DisplayCachingKey::class.java) &&
               param.getAnnotation(DisplayCachingKey::class.java).forHashKey
            }
         val annotatedValue = hashKeyAnnotatedParam.second
               ?: throw IllegalArgumentException(
               "'@${DisplayCachingKey::class.simpleName}.forHashKey = true' parameter value should not be null.")

         // 해시 키 생성 및 반환
         val hashKey = DisplayCachingKey.convertParamValueToKeyString(annotatedValue)
         if (hashKey.isBlank()) {
            throw IllegalArgumentException(
               "'@${DisplayCachingKey::class.simpleName}.forHashKey = true' parameter value should not be blank.")
         }
         return hashKey
      } catch (e: Exception) {
         val exceptionMessage = "Fail to proper '@${DisplayCachingKey::class.simpleName}.forHashKey=true' " +
            "annotated parameter or value from the '${cacheKey}'. Please check annotation or actual value."
         throw IllegalStateException(exceptionMessage, e)
      }
   }

   /**
    * [@DisplayCaching]어노테이션의 around로 동작합니다.
    */
   @Around("@annotation(display.module.cache.annotation.DisplayCaching)")
   @Throws(Throwable::class)
   fun aroundDisplayCaching(joinPoint: ProceedingJoinPoint): Any? {
      // 캐시 모듈을 호출하기 위한 전처리: 리플랙션을 통한 fallback함수 속성, 캐시 키 생성, 어노테이션 속성 추출 등 수행
      val method = (joinPoint.signature as MethodSignature).method
      val annotation = method.getAnnotation(DisplayCaching::class.java)
      val cacheKey = makeCacheKey(joinPoint, annotation, method)
      val hashKey = when (annotation.displayCacheInfo.redisDataType) {
         RedisDataType.HASH -> readHashKeyOrThrow(joinPoint, method, cacheKey)
         else -> "" // hashKey 는 RedisDataType.HASH일때만 사용
      }
      // 캐시 모듈 호출
      return DisplayCacheModule(
         circuitBreaker = circuitBreaker,
         redisTemplate = redisTemplate,
         objectMapper = objectMapper
      ).fetchAndPut(
         redisDataType = annotation.displayCacheInfo.redisDataType,
         cacheKey = cacheKey,
         hashKey = hashKey,
         ttlBySeconds = annotation.displayCacheInfo.ttl,
         putDataAfterProceed = annotation.putDataAfterProceed,
         throwWhenReadFail = annotation.throwWhenReadFail,
         fallbackFunction = { joinPoint.proceed(joinPoint.args) },
         fallbackReturnType = method.returnType
      )
   }
}

/**
 * fallback 기능을 분산락을 사용하여 한 번만 수행할 때,
 * 락의 수명과 다시 읽는 주기 등을 설정하기 위한 열거형 클래스
 */
enum class FallbackLockOption(
   val maxLockDurationMs: Long, // 락 수명(밀리초)
   val sleepMsBetweenObtainLock: Long // 읽기/락 획득 주기(밀리초)
) {
   // 락 미사용
   DISABLED(0L, 0L),

   // 기본 락 (최대 10초간 락을 걸고, 1초마다 락 해제/데이터 적재여부를 확인)
   LOCK_10S_CHECK_EVERY_1S(10000L, 1000L);
}

class DisplayCacheModule(
   private val circuitBreaker: CircuitBreaker,
   private val redisTemplate: RedisTemplate<String, Any?>,
   private val objectMapper: ObjectMapper
) {
   companion object {

      // 이전 코드 생략 ...

      /**
       * Resilience4j적용, 레디스에서 데이터를 읽어오고 예외 발생 시 핸들링 수행합니다.
       * 추가로, 데이터 획득 실패 시 락 획득을 시도하고 획득에 성공한 경우 '[isCurrentThreadHasFallbackLock] = true'로 플래그를 설정합니다.
       */
      private fun tryReadFromRedisWithObtainLock(
         redisDataType: RedisDataType,
         cacheKey: String,
         hashKey: String,
         throwWhenReadFail: Boolean = false,
         fallbackReturnType: Class<*>,
         fallbackLockOption: FallbackLockOption = FallbackLockOption.DISABLED,
         isCurrentThreadHasFallbackLock: AtomicBoolean
      ): Any {
         val isFallbackLockEnabled = (fallbackLockOption != FallbackLockOption.DISABLED)
         val readRetryCount =
            if (isFallbackLockEnabled) (fallbackLockOption.maxLockDurationMs / fallbackLockOption.sleepMsBetweenObtainLock).toInt()
            else 1

         repeat(readRetryCount) {
            try {
               // 1.레디스로부터 데이터 읽기 시도
               val readData = circuitBreaker.executeSupplier {
                  tryReadFromRedis(
                     redisDataType = redisDataType,
                     cacheKey = cacheKey,
                     hashKey = hashKey,
                     throwWhenReadFail = throwWhenReadFail,
                     returnType = fallbackReturnType
                  )
               }
               // 2-1.읽기 성공 시 반환
               if (readData !== SAFE_NULL_OBJECT) {
                  return readData
               }
            } catch (e: Exception) {
               logger.warn("No data '${cacheKey}' in redis. (readRetryCount: ${readRetryCount})")
               if (throwWhenReadFail) throw e
            }

            // 2-2.읽기 실패 시
            if (throwWhenReadFail) {
               throw NoSuchElementException("No such data '${cacheKey}' in redis.")
            }

            // 3.락 획득 시도
            if (isFallbackLockEnabled) {
               if (tryObtainRedisLock(
                  cacheKey = cacheKey,
                  fallbackLockOption = fallbackLockOption,
                  isCurrentThreadHasFallbackLock = isCurrentThreadHasFallbackLock)
               ) {
                  return@repeat
               }
               // 락 획득 실패 시 (1)부터 다시 시작
            }
         }

         // 레디스에 데이터가 없고, 락 획득도 실패한 경우 도달
         return SAFE_NULL_OBJECT
      }

      /**
       * Redis에 .lock 데이터를 생성하여 락 획득을 시도합니다.
       */
      private fun tryObtainRedisLock(
         cacheKey: String,
         fallbackLockOption: FallbackLockOption,
         isCurrentThreadHasFallbackLock: AtomicBoolean
      ): Boolean {
         val lockKey = cacheKey + ".lock"
         val lockValue = "true"

         try {
            val lockObtained = circuitBreaker.executeSupplier {
               redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, fallbackLockOption.maxLockDurationMs, TimeUnit.MILLISECONDS)
            }
            if (lockObtained) {
               // 4-1.락 획득 성공
               isCurrentThreadHasFallbackLock.set(true)
               return true
            } else {
               // 4-2.락 획득 실패
               // 과부하 방지를 위해 일정 시간만큼(최소 1ms) 대기 후 레디스 읽기 재시도
               // 대기 후 다음 싸이클에 데이터 읽기가 시도됐을 때,
               // 다른 컨테이너/스레드에서 fallback수행 결과를 적재했다면 여기에 다시 진입하지 않음
               Thread.sleep(fallbackLockOption.sleepMsBetweenObtainLock.coerceAtLeast(1L))
               return false
            }
         } catch (e: Exception) {
            logger.warn("Fail to obtain lock '${lockKey}' from redis. (${e.message})")
            return false
         }
      }

      /**
       * Resilience4j적용, 레디스에 데이터를 쓰고 예외 발생 시 핸들링을 수행합니다.
       * 추가로, 레디스에 생성되어있던 .lock 데이터의 삭제를 시도합니다.
       */
      private fun tryWriteIntoRedisWithReleaseLock(
         redisDataType: RedisDataType,
         cacheKey: String,
         hashKey: String,
         ttlBySeconds: Long,
         writeData: Any,
         isCurrentThreadHasFallbackLock: AtomicBoolean
      ) {
         var isWriteSuccess = false
         try {
            // 5.레디스에 데이터 적재 시
            isWriteSuccess = tryWriteIntoRedis(
               redisDataType = redisDataType,
               cacheKey = cacheKey,
               hashKey = hashKey,
               ttlBySeconds = ttlBySeconds,
               writeData = writeData
            )
         } catch (e: Exception) {
            logger.warn("Fail to write data '${cacheKey}' into redis.", e)
         } finally {
            // 6.락 해제
            if (isWriteSuccess && isCurrentThreadHasFallbackLock.get()) {
               tryReleaseRedisLock(cacheKey = cacheKey)
            }
         }
      }
   }
}

/**
 * 캐시를 읽어옵니다. 캐시를 획득하지 못한 경우 함수 본체를 수행한 결과를 캐싱 후 반환합니다.
 * @param redisDataType 레디스 데이터 타입
 * @param cacheKey 캐시 키
 * @param hashKey 해시 키 ([RedisDataType.HASH]의 경우 사용)
 * @param ttlBySeconds 데이터의 TTL(초)
 * @param putDataAfterProceed fallback로직 수행 후 데이터 적재 수행 여부
 * @param throwWhenReadFail 읽기작업 실패 시 예외 던지기 여부
 * @param fallbackFunction 읽기작업 실패 시 수행할 fallback 함수
 * @param fallbackReturnType fallback함수의 반환 자료형
 * @param fallbackLockOption fallback함수 수행 시 Lock 사용 옵션
 */
fun fetchAndPut(
   redisDataType: RedisDataType,
   cacheKey: String,
   hashKey: String,
   ttlBySeconds: Long,
   putDataAfterProceed: Boolean = true,
   throwWhenReadFail: Boolean = false,
   fallbackFunction: Any?.() -> Any?,
   fallbackReturnType: Class<*>,
   fallbackLockOption: FallbackLockOption = FallbackLockOption.DISABLED
): Any? {
   val threadHasFallbackLock = AtomicBoolean(false) // [Note] 지역 변수는 동시성을 고려하지 않아도 되지만, 락 획득 여부를 참조로 전달하기 위해 사용

   // 2.레디스로부터 데이터 읽기 시도 (읽기 실패 시 Lock획득 시도)
   val fetchResult = tryReadFromRedisWithObtainLock(
      redisDataType = redisDataType,
      cacheKey = cacheKey,
      hashKey = hashKey,
      // 3-3.throwWhenReadFail가 true면 데이터 읽기 실패 예외를 던지고 종료
      throwWhenReadFail = throwWhenReadFail,
      fallbackReturnType = fallbackReturnType,
      fallbackLockOption = fallbackLockOption,
      isCurrentThreadHasFallbackLock = threadHasFallbackLock
   )

   // 3-1.읽기 성공 시 데이터 반환
   if (fetchResult !== SAFE_NULL_OBJECT) {
      return fetchResult
   }

   // 3-4.데이터 획득을 위한 fallback함수 수행
   val fallbackResult = try {
      fallbackFunction()
   } catch (e: Exception) {
      // fallback수행 중 오류 발생 시 빠른 재시도를 위해 락 해제를 시도하고 예외를 던짐
      if (threadHasFallbackLock.get()) { tryReleaseRedisLock(cacheKey) }
      throw e
   }

   // 4-2.fallback함수 수행 결과를 반환
   if (!putDataAfterProceed || isNullOrEmptyData(fallbackResult)) {
      return fallbackResult
   }

   // 4-1.레디스에 데이터 적재 시도 (적재 후 Lock해제 시도)
   tryWriteIntoRedisWithReleaseLock(
      redisDataType = redisDataType,
      cacheKey = cacheKey,
      hashKey = hashKey,
      ttlBySeconds = ttlBySeconds,
      writeData = fallbackResult!!, // isNullOrEmptyData(..)에서 null 검증을 수행
      isCurrentThreadHasFallbackLock = threadHasFallbackLock
   )

   // 7.적재 후 fallback함수 수행 결과를 반환
   return fallbackResult
}