diff --git a/app/core/deps.py b/app/core/deps.py index c73a368..d8aed6e 100644 --- a/app/core/deps.py +++ b/app/core/deps.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from typing import Annotated import httpx @@ -27,6 +29,7 @@ from app.services.IssueService import IssueService from app.services.internal.IssuePinDailyRateLimitService import IssuePinDailyRateLimitService from app.services.internal.IssuePinBackgroundRunner import IssuePinBackgroundRunner +from app.services.internal.map.PinGeoRedisPublisher import PinGeoRedisPublisher from app.services.UserService import UserService from app.services.ComplaintEmailService import ComplaintEmailService from app.services.ContestPinService import ContestPinService @@ -325,6 +328,8 @@ def get_issue_service( s3_util: S3UtilDep, background_runner: IssuePinBackgroundRunnerDep, issue_pin_daily_rate_limit_service: IssuePinDailyRateLimitServiceDep, + location_repo: LocationRepoDep, + pin_geo_redis_publisher: PinGeoRedisPublisherDep, ) -> IssueService: return IssueService( vector_store_service=vector_store_service, @@ -341,6 +346,8 @@ def get_issue_service( s3_util=s3_util, background_runner=background_runner, issue_pin_daily_rate_limit_service=issue_pin_daily_rate_limit_service, + location_repo=location_repo, + pin_geo_redis_publisher=pin_geo_redis_publisher, ) @@ -356,6 +363,18 @@ def get_async_redis_client(request: Request) -> AsyncRedis: AsyncRedisDep = Annotated[AsyncRedis, Depends(get_async_redis_client)] + +def get_pin_geo_redis_publisher( + redis_client: AsyncRedisDep, +) -> PinGeoRedisPublisher: + return PinGeoRedisPublisher(redis_client) + + +PinGeoRedisPublisherDep = Annotated[ + PinGeoRedisPublisher, + Depends(get_pin_geo_redis_publisher), +] + CurrentUserIdDep = Annotated[str, Depends(get_current_user_id)] OptionalUserIdDep = Annotated[str | None, Depends(get_optional_user_id)] @@ -420,6 +439,8 @@ def get_festival_event_ingest_service( pin_location_repo: PinLocationRepoDep, pin_image_repo: PinImageRepoDep, location_resolve_client: LocationResolveClientDep, + location_repo: LocationRepoDep, + pin_geo_redis_publisher: PinGeoRedisPublisherDep, ) -> FestivalEventIngestService: return FestivalEventIngestService( pin_repo=pin_repo, @@ -427,6 +448,8 @@ def get_festival_event_ingest_service( pin_location_repo=pin_location_repo, pin_image_repo=pin_image_repo, location_resolve_client=location_resolve_client, + location_repo=location_repo, + pin_geo_redis_publisher=pin_geo_redis_publisher, ) diff --git a/app/repositories/LocationRepo.py b/app/repositories/LocationRepo.py index 757fb32..11db225 100644 --- a/app/repositories/LocationRepo.py +++ b/app/repositories/LocationRepo.py @@ -18,6 +18,12 @@ async def get_all_ordered(self) -> list[Location]: ) return list(result.scalars().all()) + async def get_region_by_id(self, location_id: int) -> str | None: + location = await self.get_by_id(location_id) + if location is None: + return None + return location.region + async def get_target_petition_by_location_id(self, *, location_id: int) -> int | None: result = await self.session.execute( select(PopulationDensity.target_petition) diff --git a/app/services/FestivalEventIngestService.py b/app/services/FestivalEventIngestService.py index 56cd4f5..f7fd543 100644 --- a/app/services/FestivalEventIngestService.py +++ b/app/services/FestivalEventIngestService.py @@ -17,6 +17,7 @@ from app.models.enum.PinType import PinType from app.models.enum.ToneType import ToneType from app.repositories.EventPinRepo import EventPinRepo +from app.repositories.LocationRepo import LocationRepo from app.repositories.PinImageRepo import PinImageRepo from app.repositories.PinLocationRepo import PinLocationRepo from app.repositories.PinRepo import PinRepo @@ -47,6 +48,8 @@ ) from app.services.internal.geo.LocationResolveClient import LocationResolveClient from app.services.internal.geo.location_resolve_fields import resolve_pin_location_fields +from app.services.internal.map.PinGeoRedisPublisher import PinGeoCacheItem, PinGeoRedisPublisher +from app.utils.geo import wgs84_from_pin_point from app.utils.festival_date_filter import current_year_festival_range, festival_overlaps_range from app.utils.visitkorea_area import area_display_name, resolve_row_area_code, row_matches_area_filter from rag.scripts.chunk_module import write_jsonl @@ -85,12 +88,16 @@ def __init__( pin_location_repo: PinLocationRepo, pin_image_repo: PinImageRepo, location_resolve_client: LocationResolveClient, + location_repo: LocationRepo, + pin_geo_redis_publisher: PinGeoRedisPublisher, ) -> None: self._pin_repo = pin_repo self._event_pin_repo = event_pin_repo self._pin_location_repo = pin_location_repo self._pin_image_repo = pin_image_repo self._location_resolve_client = location_resolve_client + self._location_repo = location_repo + self._pin_geo_redis_publisher = pin_geo_redis_publisher async def commit(self) -> None: await self._pin_repo.commit() @@ -369,6 +376,7 @@ async def _import_handoff_rows( errors: list[dict[str, Any]] = [] items: list[FestivalBatchItemResult] = [] pin_ids: list[int] = [] + geo_cache_items: list[PinGeoCacheItem] = [] pending_rows: list[dict[str, Any]] = [] for row in handoff_rows: @@ -403,7 +411,7 @@ async def _import_handoff_rows( try: async with self._pin_repo.session.begin_nested(): if existing is None: - pin_id = await self._insert_festival_pin( + pin_id, cache_item = await self._insert_festival_pin( admin_uid=admin_uid, row=row, festival_api_id=festival_api_id, @@ -411,6 +419,7 @@ async def _import_handoff_rows( inserted_count += 1 db_ids.add(festival_api_id) pin_ids.append(pin_id) + geo_cache_items.append(cache_item) items.append( FestivalBatchItemResult( festival_api_id=festival_api_id, @@ -419,9 +428,10 @@ async def _import_handoff_rows( ), ) elif allow_update: - pin_id = await self._update_festival_pin(existing=existing, row=row) + pin_id, cache_item = await self._update_festival_pin(existing=existing, row=row) updated_count += 1 pin_ids.append(pin_id) + geo_cache_items.append(cache_item) items.append( FestivalBatchItemResult( festival_api_id=festival_api_id, @@ -449,6 +459,7 @@ async def _import_handoff_rows( ) await self._pin_repo.commit() + await self._pin_geo_redis_publisher.add_pins_batch(geo_cache_items) pending_import = 0 for row in handoff_rows: @@ -561,7 +572,7 @@ async def _insert_festival_pin( admin_uid: str, row: dict[str, Any], festival_api_id: int, - ) -> int: + ) -> tuple[int, PinGeoCacheItem]: location_fields = await self._resolve_location(row) if location_fields is None: raise ValueError("위치 resolve 실패") @@ -596,9 +607,17 @@ async def _insert_festival_pin( await self._pin_location_repo.save(pin_location, flush_immediately=True) await self._replace_pin_images(pin_id=pin.pin_id, row=row) - return int(pin.pin_id) + cache_item = await self._build_festival_geo_cache_item( + pin_id=int(pin.pin_id), + location_id=location_id, + detail_address=detail_address, + pin_point=pin_point, + ) + return int(pin.pin_id), cache_item - async def _update_festival_pin(self, *, existing: EventPin, row: dict[str, Any]) -> int: + async def _update_festival_pin( + self, *, existing: EventPin, row: dict[str, Any] + ) -> tuple[int, PinGeoCacheItem]: pin = existing.pin if pin is None: raise ValueError("연결된 pin 없음") @@ -633,7 +652,33 @@ async def _update_festival_pin(self, *, existing: EventPin, row: dict[str, Any]) await self._pin_image_repo.delete_by_pin_id(pin.pin_id) await self._replace_pin_images(pin_id=pin.pin_id, row=row) - return int(pin.pin_id) + cache_item = await self._build_festival_geo_cache_item( + pin_id=int(pin.pin_id), + location_id=location_id, + detail_address=detail_address, + pin_point=pin_point, + ) + return int(pin.pin_id), cache_item + + async def _build_festival_geo_cache_item( + self, + *, + pin_id: int, + location_id: int, + detail_address: str, + pin_point, + ) -> PinGeoCacheItem: + lat, lng = wgs84_from_pin_point(pin_point) + region = await self._location_repo.get_region_by_id(location_id) or "" + return PinGeoCacheItem( + pin_id=pin_id, + pin_type=PinType.FESTIVAL.name, + lat=lat, + lng=lng, + detail_address=detail_address, + region=region, + discount=None, + ) async def _replace_pin_images(self, *, pin_id: int, row: dict[str, Any]) -> None: for spec in pin_images_for_db_row(row): diff --git a/app/services/IssueService.py b/app/services/IssueService.py index 27ed58f..b6590dd 100644 --- a/app/services/IssueService.py +++ b/app/services/IssueService.py @@ -27,6 +27,7 @@ from app.models.enum.ToneType import ToneType from app.repositories.CommunityRepo import CommunityRepo from app.repositories.IssuePinRepo import IssuePinRepo +from app.repositories.LocationRepo import LocationRepo from app.repositories.PinLikeRepo import PinLikeRepo from app.repositories.PinImageRepo import PinImageRepo from app.repositories.PinLocationRepo import PinLocationRepo @@ -53,6 +54,7 @@ RateLimitSnapshot, ) from app.services.internal.IssuePinBackgroundRunner import IssuePinBackgroundRunner +from app.services.internal.map.PinGeoRedisPublisher import PinGeoRedisPublisher from app.services.internal.ai.IssuePinLLMService import IssuePinLLMService from app.services.internal.ai.IssueRagPlannerService import IssueRagPlannerService from app.services.internal.geo.LocationResolveClient import LocationResolveClient @@ -104,6 +106,8 @@ def __init__( s3_util: S3Util, background_runner: IssuePinBackgroundRunner, issue_pin_daily_rate_limit_service: IssuePinDailyRateLimitService, + location_repo: LocationRepo, + pin_geo_redis_publisher: PinGeoRedisPublisher, ) -> None: self._vector_store_service = vector_store_service self._issue_rag_planner_service = issue_rag_planner_service @@ -119,6 +123,38 @@ def __init__( self._user_repo = user_repo self._background_runner = background_runner self._rate_limit = issue_pin_daily_rate_limit_service + self._location_repo = location_repo + self._pin_geo_redis_publisher = pin_geo_redis_publisher + + async def _sync_issue_pin_geo_cache( + self, + *, + pin_id: int, + pin_location: PinLocation, + ) -> None: + """commit 성공 후 Redis GEO 캐시를 best-effort로 동기화한다. + + DB 커밋 이후에만 호출해야 한다. 실패해도 API 응답은 성공으로 유지하고 + warning 로그만 남긴다(BE 야간 배치로 캐시 정합성 복구 가능). + """ + try: + lat, lng = wgs84_from_pin_point(pin_location.pin_point) + region = await self._location_repo.get_region_by_id(pin_location.location_id) or "" + await self._pin_geo_redis_publisher.add_pin( + pin_id=pin_id, + pin_type=PinType.ISSUE.name, + lat=lat, + lng=lng, + detail_address=pin_location.detail_address or "", + region=region, + discount=None, + ) + except Exception: + logger.warning( + "issue pin geo cache sync failed pin_id=%s", + pin_id, + exc_info=True, + ) async def _build_rate_limit_quota_dict( self, @@ -753,6 +789,11 @@ async def create_issue_pin( logger.exception("issue pin create commit failed uid=%s", uid) raise_business_exception(ErrorCode.ISSUE_PIN_IMPORT_FAILED) + await self._sync_issue_pin_geo_cache( + pin_id=pin.pin_id, + pin_location=pin_location, + ) + snapshot = await self._rate_limit.record_daily_quota_success( RateLimitKind.CREATE, uid=uid, @@ -872,6 +913,11 @@ async def update_issue_pin( logger.exception("issue pin update failed pin_id=%s", pin.pin_id) raise_business_exception(ErrorCode.ISSUE_PIN_EDIT_FAILED) + await self._sync_issue_pin_geo_cache( + pin_id=pin.pin_id, + pin_location=pin_location, + ) + snapshot = await self._rate_limit.record_daily_quota_success( RateLimitKind.EDIT, uid=uid, diff --git a/app/services/internal/map/PinGeoRedisPublisher.py b/app/services/internal/map/PinGeoRedisPublisher.py new file mode 100644 index 0000000..ec91c3d --- /dev/null +++ b/app/services/internal/map/PinGeoRedisPublisher.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from datetime import timedelta +from typing import Any + +from redis.asyncio import Redis as AsyncRedis + +logger = logging.getLogger(__name__) + +# BE PinGeoRedisService 와 동일 +GEO_KEY_ALL = "geo:pins" +GEO_KEY_PREFIX = "geo:pins:" +PIN_INFO_KEY_PREFIX = "pin:info:" +PIN_INFO_TTL = timedelta(hours=48) +PIN_INFO_TTL_SECONDS = int(PIN_INFO_TTL.total_seconds()) + + +@dataclass(frozen=True, slots=True) +class PinGeoCacheItem: + pin_id: int + pin_type: str + lat: float + lng: float + detail_address: str + region: str + discount: str | None = None + + +class PinGeoRedisPublisher: + """BE PinGeoRedisService write 규약을 미러링하는 Redis GEO 캐시 발행기.""" + + def __init__(self, redis_client: AsyncRedis) -> None: + self._redis = redis_client + + async def add_pin( + self, + *, + pin_id: int, + pin_type: str, + lat: float, + lng: float, + detail_address: str, + region: str, + discount: str | None = None, + ) -> None: + try: + member = str(pin_id) + await self._redis.geoadd(GEO_KEY_ALL, (lng, lat, member)) + await self._redis.geoadd(f"{GEO_KEY_PREFIX}{pin_type}", (lng, lat, member)) + payload = _build_pin_info_json( + pin_id=pin_id, + pin_type=pin_type, + lat=lat, + lng=lng, + detail_address=detail_address, + region=region, + discount=discount, + ) + await self._redis.set( + f"{PIN_INFO_KEY_PREFIX}{pin_id}", + payload, + ex=PIN_INFO_TTL_SECONDS, + ) + except Exception as exc: + logger.warning("Redis GEO 핀 추가 실패 pin_id=%s: %s", pin_id, exc) + + async def add_pins_batch(self, items: list[PinGeoCacheItem]) -> None: + if not items: + return + try: + async with self._redis.pipeline(transaction=False) as pipe: + for item in items: + member = str(item.pin_id) + pipe.geoadd(GEO_KEY_ALL, (item.lng, item.lat, member)) + pipe.geoadd(f"{GEO_KEY_PREFIX}{item.pin_type}", (item.lng, item.lat, member)) + pipe.set( + f"{PIN_INFO_KEY_PREFIX}{item.pin_id}", + _build_pin_info_json( + pin_id=item.pin_id, + pin_type=item.pin_type, + lat=item.lat, + lng=item.lng, + detail_address=item.detail_address, + region=item.region, + discount=item.discount, + ), + ex=PIN_INFO_TTL_SECONDS, + ) + await pipe.execute() + except Exception as exc: + logger.warning("Redis GEO 핀 batch 추가 실패 count=%s: %s", len(items), exc) + + async def remove_pin(self, *, pin_id: int, pin_type: str) -> None: + try: + member = str(pin_id) + await self._redis.zrem(GEO_KEY_ALL, member) + await self._redis.zrem(f"{GEO_KEY_PREFIX}{pin_type}", member) + await self._redis.delete(f"{PIN_INFO_KEY_PREFIX}{pin_id}") + except Exception as exc: + logger.warning("Redis GEO 핀 삭제 실패 pin_id=%s: %s", pin_id, exc) + + +def _build_pin_info_json( + *, + pin_id: int, + pin_type: str, + lat: float, + lng: float, + detail_address: str, + region: str, + discount: str | None, +) -> str: + payload: dict[str, Any] = { + "pinId": pin_id, + "pinType": pin_type, + "latitude": lat, + "longitude": lng, + "pinDetailAddress": detail_address, + "pinLocation": region, + "discount": discount, + } + return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) diff --git a/app/services/internal/map/__init__.py b/app/services/internal/map/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_festival_admin_batch.py b/tests/test_festival_admin_batch.py index 814a59d..0fcc043 100644 --- a/tests/test_festival_admin_batch.py +++ b/tests/test_festival_admin_batch.py @@ -221,6 +221,8 @@ async def test_import_batch_skips_existing_when_allow_update_false(self) -> None pin_location_repo=MagicMock(), pin_image_repo=MagicMock(), location_resolve_client=MagicMock(), + location_repo=MagicMock(), + pin_geo_redis_publisher=MagicMock(add_pins_batch=AsyncMock()), ) service._event_pin_repo.list_festival_api_ids = AsyncMock(return_value={100}) service._event_pin_repo.get_by_festival_api_id = AsyncMock() @@ -260,6 +262,8 @@ async def test_import_batch_uses_savepoint_per_row(self) -> None: pin_location_repo=MagicMock(), pin_image_repo=MagicMock(), location_resolve_client=MagicMock(), + location_repo=MagicMock(), + pin_geo_redis_publisher=MagicMock(add_pins_batch=AsyncMock()), ) service._event_pin_repo.list_festival_api_ids = AsyncMock(return_value=set()) service._event_pin_repo.get_by_festival_api_id = AsyncMock(return_value=None) @@ -302,6 +306,8 @@ async def test_import_batch_expunges_session_new_after_nested_failure(self) -> N pin_location_repo=MagicMock(), pin_image_repo=MagicMock(), location_resolve_client=MagicMock(), + location_repo=MagicMock(), + pin_geo_redis_publisher=MagicMock(add_pins_batch=AsyncMock()), ) service._event_pin_repo.list_festival_api_ids = AsyncMock(return_value=set()) service._event_pin_repo.get_by_festival_api_id = AsyncMock(return_value=None) @@ -329,6 +335,8 @@ async def test_import_all_delegates_to_import_handoff_rows(self) -> None: pin_location_repo=MagicMock(), pin_image_repo=MagicMock(), location_resolve_client=MagicMock(), + location_repo=MagicMock(), + pin_geo_redis_publisher=MagicMock(add_pins_batch=AsyncMock()), ) expected = FestivalImportBatchResult( import_all=True, diff --git a/tests/test_issue_pin_image_required.py b/tests/test_issue_pin_image_required.py index fce829d..818f252 100644 --- a/tests/test_issue_pin_image_required.py +++ b/tests/test_issue_pin_image_required.py @@ -30,6 +30,8 @@ def _build_issue_service() -> IssueService: s3_util=MagicMock(), background_runner=MagicMock(), issue_pin_daily_rate_limit_service=MagicMock(), + location_repo=MagicMock(), + pin_geo_redis_publisher=MagicMock(), ) diff --git a/tests/test_pin_geo_redis_publisher.py b/tests/test_pin_geo_redis_publisher.py new file mode 100644 index 0000000..a07b2a6 --- /dev/null +++ b/tests/test_pin_geo_redis_publisher.py @@ -0,0 +1,149 @@ +from __future__ import annotations + +import json +import unittest +from unittest.mock import AsyncMock, MagicMock + +from app.services.internal.map.PinGeoRedisPublisher import ( + GEO_KEY_ALL, + GEO_KEY_PREFIX, + PIN_INFO_KEY_PREFIX, + PIN_INFO_TTL_SECONDS, + PinGeoCacheItem, + PinGeoRedisPublisher, + _build_pin_info_json, +) + + +class PinGeoRedisPublisherConstantsTest(unittest.TestCase): + def test_pin_info_ttl_matches_backend_duration_of_hours_48(self) -> None: + self.assertEqual(PIN_INFO_TTL_SECONDS, 172_800) + + +class PinGeoRedisPublisherJsonTest(unittest.TestCase): + def test_build_pin_info_json_uses_java_camel_case_fields(self) -> None: + payload = json.loads( + _build_pin_info_json( + pin_id=123, + pin_type="ISSUE", + lat=37.5, + lng=127.0, + detail_address="상세주소", + region="서울특별시", + discount=None, + ), + ) + self.assertEqual( + payload, + { + "pinId": 123, + "pinType": "ISSUE", + "latitude": 37.5, + "longitude": 127.0, + "pinDetailAddress": "상세주소", + "pinLocation": "서울특별시", + "discount": None, + }, + ) + + +class PinGeoRedisPublisherAddPinTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self.redis = MagicMock() + self.redis.geoadd = AsyncMock() + self.redis.set = AsyncMock() + self.publisher = PinGeoRedisPublisher(self.redis) + + async def test_add_pin_writes_geo_and_pin_info_with_backend_ttl(self) -> None: + await self.publisher.add_pin( + pin_id=42, + pin_type="ISSUE", + lat=37.5, + lng=127.0, + detail_address="주소", + region="서울", + discount=None, + ) + + self.redis.geoadd.assert_any_await(GEO_KEY_ALL, (127.0, 37.5, "42")) + self.redis.geoadd.assert_any_await(f"{GEO_KEY_PREFIX}ISSUE", (127.0, 37.5, "42")) + self.redis.set.assert_awaited_once() + set_args, set_kwargs = self.redis.set.await_args + self.assertEqual(set_args[0], f"{PIN_INFO_KEY_PREFIX}42") + self.assertEqual(set_kwargs["ex"], 172_800) + payload = json.loads(set_args[1]) + self.assertEqual(payload["pinId"], 42) + self.assertEqual(payload["pinType"], "ISSUE") + + async def test_add_pin_swallows_redis_errors(self) -> None: + self.redis.geoadd.side_effect = RuntimeError("redis down") + + await self.publisher.add_pin( + pin_id=1, + pin_type="FESTIVAL", + lat=37.0, + lng=127.0, + detail_address="주소", + region="서울", + ) + + +class PinGeoRedisPublisherBatchTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self.redis = MagicMock() + self.pipeline = MagicMock() + self.pipeline.__aenter__ = AsyncMock(return_value=self.pipeline) + self.pipeline.__aexit__ = AsyncMock(return_value=None) + self.pipeline.geoadd = MagicMock() + self.pipeline.set = MagicMock() + self.pipeline.execute = AsyncMock(return_value=[]) + self.redis.pipeline = MagicMock(return_value=self.pipeline) + self.publisher = PinGeoRedisPublisher(self.redis) + + async def test_add_pins_batch_uses_pipeline(self) -> None: + items = [ + PinGeoCacheItem( + pin_id=1, + pin_type="FESTIVAL", + lat=37.1, + lng=127.1, + detail_address="A", + region="서울", + ), + PinGeoCacheItem( + pin_id=2, + pin_type="FESTIVAL", + lat=37.2, + lng=127.2, + detail_address="B", + region="부산", + ), + ] + + await self.publisher.add_pins_batch(items) + + self.redis.pipeline.assert_called_once_with(transaction=False) + self.assertEqual(self.pipeline.geoadd.call_count, 4) + self.assertEqual(self.pipeline.set.call_count, 2) + self.pipeline.execute.assert_awaited_once() + _, set_kwargs = self.pipeline.set.call_args_list[0] + self.assertEqual(set_kwargs["ex"], 172_800) + + async def test_add_pins_batch_noop_for_empty_list(self) -> None: + await self.publisher.add_pins_batch([]) + self.redis.pipeline.assert_not_called() + + +class PinGeoRedisPublisherRemovePinTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self.redis = MagicMock() + self.redis.zrem = AsyncMock() + self.redis.delete = AsyncMock() + self.publisher = PinGeoRedisPublisher(self.redis) + + async def test_remove_pin_deletes_geo_and_info_keys(self) -> None: + await self.publisher.remove_pin(pin_id=9, pin_type="ISSUE") + + self.redis.zrem.assert_any_await(GEO_KEY_ALL, "9") + self.redis.zrem.assert_any_await(f"{GEO_KEY_PREFIX}ISSUE", "9") + self.redis.delete.assert_awaited_once_with(f"{PIN_INFO_KEY_PREFIX}9")