[deploy] issue pin, festival pin redis cache write 파트 구현 배포

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,13 +439,17 @@ 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,
         event_pin_repo=event_pin_repo,
         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,
     )
 
 

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)

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,14 +411,15 @@ 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,
                         )
                         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):

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,