이미지 그룹핑

CLIP 기반 이미지 임베딩과 유사도 그룹핑을 통해 룹의 사진을 자동 분류하는 파이프라인입니다. 비즈니스 규칙은 이미지 그룹핑 참조.

아키텍처 개요

SQS Media Ready Consumer 재시도 로직

ProcessMediaReadyUseCase의 반환 타입이 sealed interface Result로 변경되어 3가지 결과를 명시적으로 처리합니다:

Result	의미	Consumer 동작
`Success`	정상 처리	메시지 삭제
`MediaNotFound`	미디어 미발견 (race condition)	receiveCount ≤ 3이면 재배달, 초과 시 Sentry 리포트 + 삭제
`Skipped`	처리 불필요 (중복 등)	메시지 삭제

Race Condition 대응: media-service의 media-ready SQS 메시지가 Core에 도달했을 때 미디어 레코드가 아직 DB에 반영되지 않은 경우가 있음. 기존에는 즉시 실패+삭제였으나, 이제 SQS의 ApproximateReceiveCount를 활용하여 최대 maxMediaNotFoundRetries(기본 3)회까지 재배달합니다.

환경변수	타입	기본값	설명
`MAX_MEDIA_NOT_FOUND_RETRIES`	int	`3`	MediaNotFound 시 최대 재시도 횟수

Hexagonal Architecture 구조

Port (Application Layer)

모든 포트는 application/loopmedia/port/ 패키지에 위치합니다.

Port	역할	패턴
`ImageGroupingPort`	외부 그룹핑 마이크로서비스(truloop-ai-server) 통신	임베딩 CRUD + 그룹 조회
`GroupingJobPublisherPort`	SQS 그룹핑 작업 큐 발행	단일 메서드 (`publishGroupingRequest`)
`LoopGroupingRepository` / `LoopGroupingQueries`	그룹핑 결과 캐시 DB	CQS 패턴 (Command/Query 분리)
`GroupRepresentativeRepository` / `GroupRepresentativeQueries`	대표 사진 오버라이드 DB	CQS 패턴

kotlin

interface ImageGroupingPort {
    suspend fun createEmbedding(loopEid: String, mediaEid: String, imageUrl: String)
    suspend fun getGroups(loopEid: String): ImageGroupsResult
    suspend fun deleteEmbedding(loopEid: String, mediaEid: String)
    suspend fun deleteAllEmbeddings(loopEid: String)
}

kotlin

interface GroupingJobPublisherPort {
    suspend fun publishGroupingRequest(loopEid: String)
}

Adapter (External)

Adapter	패키지	구현 대상
`HttpImageGroupingAdapter`	`adapter.external.imagegrouping`	`ImageGroupingPort` — HTTP로 truloop-ai-server 호출, retry with linear backoff
`NoOpImageGroupingAdapter`	`adapter.external.imagegrouping`	`ImageGroupingPort` — 기능 비활성화 시 (graceful degradation)
`SqsGroupingJobPublisher`	`adapter.external.sqs`	`GroupingJobPublisherPort` — SQS 큐에 그룹핑 요청 발행
`NoOpGroupingJobPublisher`	`adapter.external.sqs`	`GroupingJobPublisherPort` — 큐 미설정 시

정보

HttpImageGroupingAdapter는 truloopAssistantHttpClient를 공유합니다. 해당 클라이언트에 expectSuccess가 설정되지 않았으므로, 어댑터 내부에서 requireSuccess()로 응답 상태를 직접 검증합니다.

Adapter (Persistence)

두 어댑터 모두 기존 adapter.persistence.loopmedia 패키지에 위치합니다.

Adapter	테이블
`ExposedLoopGroupingAdapter`	`loop_grouping_results` (upsert)
`ExposedGroupRepresentativeAdapter`	`loop_group_representative_overrides` (upsert)

UseCase

UseCase	트리거	역할
`CreateMediaEmbeddingUseCase`	`MediaReadyEvent`	PHOTO만 임베딩 생성 → SQS 큐 발행
`ProcessGroupingQueueUseCase`	`POST /internal/grouping/process-queue`	배치 그룹핑 처리 (3-phase)
`GetGroupedLoopMediaUseCase`	`GET /api/v3/loops/{eid}/media`	미디어 목록 + 그룹 메타데이터 + 댓글 수 + 태그 사용자 반환
`SetGroupRepresentativeUseCase`	`PUT .../groups/{group_eid}/representative`	사용자 지정 대표 사진 설정
`RemoveGroupRepresentativeUseCase`	`DELETE .../groups/{group_eid}/representative`	대표 사진 오버라이드 제거

Event Handler

이벤트 핸들러는 bootstrap/event/handler/loopmedia/ 패키지에 위치합니다.

Handler	이벤트	동작
`MediaReadyEmbeddingHandler`	`MediaReadyEvent`	`CreateMediaEmbeddingUseCase` 호출
`MediaDeletedEmbeddingHandler`	`MediaDeletedEvent`	`ImageGroupingPort.deleteEmbedding()` 직접 호출

GetGroupedLoopMediaUseCase Enrichment

GetGroupedLoopMediaUseCase는 페이지네이션된 미디어에 대해 다음 부가 정보를 배치 조회하여 응답에 포함합니다:

필드	소스	설명
`commentsCount`	`CommentQueries.findActiveByMediaEids()`	미디어별 활성 댓글 수
`taggedUsers`	`MediaTaggedUserQueries.findByMediaEids()`	미디어에 태그된 사용자 목록

ProcessGroupingQueueUseCase 3-Phase 패턴

DB 트랜잭션 중 외부 HTTP 호출을 피하기 위해 3단계로 분리합니다.

Phase 1: Read (read-only TX)

루프 존재 확인
전체 미디어에서 PHOTO만 필터링
최소 2장 미만 시 캐시 무효화 후 skip
시간 분산(time spread) 검증: capturedAt 범위가 threshold(기본 24h) 초과 시 캐시 무효화 후 skip

Phase 2: HTTP Call (no TX)

ImageGroupingPort.getGroups(loopEid) — truloop-ai-server의 threshold-free v2 API 호출
트랜잭션 밖에서 실행하여 DB 커넥션을 점유하지 않음

Phase 3: Write (TX)

동시성 보호: Phase 1에서 읽은 PHOTO 수와 현재 PHOTO 수 비교
수가 변경되었으면 캐시 무효화 후 skip
변경 없으면 LoopGroupingRepository.upsert() — 결과 캐시 저장

kotlin

// Phase 1: Read data (read-only transaction)
val readResult = uow.transactional(readOnly = true) {
    val photos = loopMediaQueries.findByLoopId(loop.dbId)
        .filter { it.contentType == MediaContentType.PHOTO }
    LoopReadResult(loopDbId = loop.dbId, photos = photos)
}

// Phase 2: Call grouping microservice (NO transaction)
val groupsResult = imageGroupingPort.getGroups(loopEid)

// Phase 3: Store results (transaction) — re-validate media count
uow.transactional {
    val currentPhotoCount = loopMediaQueries.findByLoopId(readResult.loopDbId)
        .count { it.contentType == MediaContentType.PHOTO }
    if (currentPhotoCount != readResult.photos.size) {
        loopGroupingRepository.deleteByLoopId(readResult.loopDbId)
        return@transactional false
    }
    loopGroupingRepository.upsert(loopId = readResult.loopDbId, /* ... */)
    true
}

대표 사진 선택 로직

GetGroupedLoopMediaUseCase.loadGroupInfoMap()에서 처리합니다.

우선순위	조건	소스
1	사용자 지정 오버라이드	`GroupRepresentativeQueries` — 해당 미디어가 그룹에 소속된 경우에만 유효
2	좋아요 수 최다 (자동)	`MediaLikeQueries.getLikeCounts()` 기반 `maxByOrNull`

주의

그룹 최소 크기: 2장 미만 그룹은 응답에서 제외됩니다 (if (group.mediaEids.size < 2) continue).

외부 마이크로서비스 API

HttpImageGroupingAdapter → truloop-ai-server (CLIP 기반):

Method	Path	설명
`POST`	`/embeddings`	임베딩 생성 (`loop_eid`, `media_eid`, `image_url`)
`GET`	`/loops/{loopEid}/groups/v2`	Threshold-free v2 그룹핑 결과 조회
`DELETE`	`/loops/{loopEid}/embeddings/{mediaEid}`	단일 임베딩 삭제
`DELETE`	`/loops/{loopEid}/embeddings`	루프 전체 임베딩 삭제

Retry 전략: Linear backoff (200ms base, 2s max), 기본 3회 재시도. ClientRequestException(4xx)은 즉시 실패, IOException/ServerResponseException(5xx)만 재시도 대상.

조건부 DI

ExternalServicesModule에서 설정 존재 여부에 따라 실제 어댑터 또는 NoOp 어댑터를 조건부 등록합니다.

kotlin

// Image Grouping (HTTP adapter when configured, NoOp fallback)
val imageGroupingConfig = config.imageGrouping
if (imageGroupingConfig != null && imageGroupingConfig.baseUrl.isNotBlank()) {
    single<ImageGroupingPort> {
        HttpImageGroupingAdapter(httpClient = truloopAssistantHttpClient, config = /* ... */)
    }
} else {
    single<ImageGroupingPort> { NoOpImageGroupingAdapter() }
}

// Grouping Job Publisher (SQS when configured, NoOp fallback)
val groupingQueueUrl = config.aws.sqsGroupingQueueUrl
if (!groupingQueueUrl.isNullOrBlank()) {
    single<GroupingJobPublisherPort> { SqsGroupingJobPublisher(groupingSqsClient, groupingQueueUrl) }
} else {
    single<GroupingJobPublisherPort> { NoOpGroupingJobPublisher() }
}

정보

설정이 없으면 기능이 완전히 비활성화됩니다 (graceful degradation). 이벤트 핸들러는 여전히 호출되지만 NoOp 어댑터가 조용히 무시합니다.

환경변수

환경변수	타입	기본값	설명
`IMAGE_GROUPING_BASE_URL`	string	(없음, 미설정 시 비활성화)	CLIP 그룹핑 마이크로서비스 URL
`IMAGE_GROUPING_RETRY_COUNT`	int	`3`	외부 호출 재시도 횟수
`IMAGE_GROUPING_TIME_SPREAD_THRESHOLD_HOURS`	int	`24`	그룹핑 유효 시간 범위(시간)
`IMAGE_GROUPING_TIMEOUT_MILLIS`	long	`30000`	HTTP 요청 타임아웃(ms)
`AWS_SQS_GROUPING_QUEUE_URL`	string	(없음, 미설정 시 NoOp)	SQS 큐 URL

변경 이력

날짜	내용
2026-03-16	SQS Media Ready Consumer 재시도 로직 추가, GetGroupedLoopMediaUseCase enrichment 추가 (commentsCount, taggedUsers), DTO 필드명 items→media 변경
2026-03-12	최초 작성 — PR #1132 기반 이미지 그룹핑 파이프라인 문서화

Loop

Recap

Mission

User

Admin

Chat

AI 비서

알림

미디어

AI 콘텐츠

2026

Core (Ktor)

Media Service (Go)

AI Server (FastAPI)

Assistant (pydantic-ai)

Functions (Lambda)

Story Worker (Serverless)

Invite (Next.js)

Admin Dashboard (React/Vite)

이미지 그룹핑 ​

아키텍처 개요 ​

SQS Media Ready Consumer 재시도 로직 ​

Hexagonal Architecture 구조 ​

Port (Application Layer) ​

Adapter (External) ​

Adapter (Persistence) ​

UseCase ​

Event Handler ​

GetGroupedLoopMediaUseCase Enrichment ​

ProcessGroupingQueueUseCase 3-Phase 패턴 ​

Phase 1: Read (read-only TX) ​

Phase 2: HTTP Call (no TX) ​

Phase 3: Write (TX) ​

대표 사진 선택 로직 ​

외부 마이크로서비스 API ​

조건부 DI ​

환경변수 ​

변경 이력 ​

이미지 그룹핑

아키텍처 개요

SQS Media Ready Consumer 재시도 로직

Hexagonal Architecture 구조

Port (Application Layer)

Adapter (External)

Adapter (Persistence)

UseCase

Event Handler

GetGroupedLoopMediaUseCase Enrichment

ProcessGroupingQueueUseCase 3-Phase 패턴

Phase 1: Read (read-only TX)

Phase 2: HTTP Call (no TX)

Phase 3: Write (TX)

대표 사진 선택 로직

외부 마이크로서비스 API

조건부 DI

환경변수

변경 이력