[BE] realtime.rs ( 실시간 데이터 )

let cache_key = format!("{}:{}", TOPIC_TRAIN_LOCATION, query.line);

if let Ok(Some(cached)) = state.cache.get_realtime(&cache_key).await {
    if let Ok(data) = serde_json::from_str::<Vec<TrainPositionResult>>(&cached) {
        return HttpResponse::Ok().json(RealtimeResponse {
            source: "cache".to_string(),
            data,
        });
    }
}

So what: Valkey에서 먼저 조회하고 캐시 히트일 시 바로 반환한다

So why:  TOPIC_TRAIN_LOCATION을 키 prefix로 쓰는 이유는 kafka.rs에서 Valkey에 저장할 때 realtime:{topic}:{key} 형식을 쓰기 때문에 같은 상수를 공유해서 Producer가 저장한 키와 여기서 조회하는 키가 자동으로 일치하게 한다. 파싱까지 성공해야 캐시 히트를 처리하는 이유는 Valkey에 손상된 데이터가 있다면 API를 다시 호출하기 위해서이다

match state.seoul.get_train_position(&query.line).await {
    Ok(data) => {
        if let Ok(json) = serde_json::to_string(&data) {
            let _ = state.cache.set_realtime(&cache_key, &json).await;
        }
        HttpResponse::Ok().json(RealtimeResponse {
            source: "api".to_string(),
            data,
        })
    }

So what: 캐스 미스일 때 서울 열린데이터 API를 직접 호출하고 결과를 Valkey에 저장한다

So why: Kafka Consumer가 30초마다 Valkey를 갱신하는게 정상 흐름이지만 서버 재시작 직후나 Kafka가 아직 데이터를 발행하기 전이면 Valkey에 데이터가 없다. 이 경우 API를 직접 호출하는 fallback이 없으면 Flutter가 빈 응답을 받게 되는데 source 필드로 캐시에서 왔는지 API에서 왔는지 구분할 수 있도록 하였다.