[HCR-299] LLM에게 전달할 테이블 배치작업

src/main/java/site/holliverse/worker/batch/jobs/memberllmcontext/MemberLlmContextJobConfig.java

@@ -0,0 +1,94 @@
+package site.holliverse.worker.batch.jobs.memberllmcontext;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.job.builder.JobBuilder;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.step.builder.StepBuilder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.transaction.PlatformTransactionManager;
+import site.holliverse.worker.batch.common.listener.BatchJobExecutionListener;
+import site.holliverse.worker.batch.common.listener.BatchStepExecutionListener;
+import site.holliverse.worker.batch.jobs.memberllmcontext.tasklet.BuildMemberLlmContextTasklet;
+import site.holliverse.worker.batch.jobs.memberllmcontext.tasklet.MemberLlmContextGateTasklet;
+import site.holliverse.worker.batch.jobs.memberllmcontext.tasklet.VerifyMemberLlmContextTasklet;
+
+/**
+ * member_llm_context 적재 전용 배치 잡 구성.
+ *
+ * 흐름은 단순하게 세 단계로 나눈다.
+ * 1. Gate: 기준일과 전월 yyyymm 계산, 필수 테이블 확인
+ * 2. Upsert: 회원 컨텍스트를 한 번에 계산해서 upsert
+ * 3. Verify: 적재 대상 수와 결과 수가 맞는지 최소 검증
+ */
+@Configuration
+@RequiredArgsConstructor
+public class MemberLlmContextJobConfig {
+
+    public static final String JOB_NAME = "memberLlmContextJob";
+
+    private final BatchJobExecutionListener batchJobExecutionListener;
+    private final BatchStepExecutionListener batchStepExecutionListener;
+
+    @Bean
+    public Job memberLlmContextJob(
+            JobRepository jobRepository,
+            Step memberLlmContextGateStep,
+            Step memberLlmContextUpsertStep,
+            Step memberLlmContextVerifyStep
+    ) {
+        return new JobBuilder(JOB_NAME, jobRepository)
+                .listener(batchJobExecutionListener)
+                .start(memberLlmContextGateStep)
+                .next(memberLlmContextUpsertStep)
+                .next(memberLlmContextVerifyStep)
+                .build();
+    }
+
+    /**
+     * 배치 파라미터와 실행 환경을 정리하는 선행 step.
+     */
+    @Bean
+    public Step memberLlmContextGateStep(
+            JobRepository jobRepository,
+            PlatformTransactionManager tx,
+            MemberLlmContextGateTasklet tasklet
+    ) {
+        return new StepBuilder("Step00_Gate", jobRepository)
+                .listener(batchStepExecutionListener)
+                .tasklet(tasklet, tx)
+                .build();
+    }
+
+    /**
+     * 실제 member_llm_context upsert SQL을 수행하는 핵심 step.
+     */
+    @Bean
+    public Step memberLlmContextUpsertStep(
+            JobRepository jobRepository,
+            PlatformTransactionManager tx,
+            BuildMemberLlmContextTasklet tasklet
+    ) {
+        return new StepBuilder("Step01_UpsertMemberLlmContext", jobRepository)
+                .listener(batchStepExecutionListener)
+                .tasklet(tasklet, tx)
+                .build();
+    }
+
+    /**
+     * 적재 결과 건수와 필수 컬럼을 검증하는 마무리 step.
+     */
+    @Bean
+    public Step memberLlmContextVerifyStep(
+            JobRepository jobRepository,
+            PlatformTransactionManager tx,
+            VerifyMemberLlmContextTasklet tasklet
+    ) {
+        return new StepBuilder("Step02_Verify", jobRepository)
+                .listener(batchStepExecutionListener)
+                .tasklet(tasklet, tx)
+                .build();
+    }
+}

src/main/java/site/holliverse/worker/batch/jobs/memberllmcontext/tasklet/BuildMemberLlmContextTasklet.java

@@ -0,0 +1,61 @@
+package site.holliverse.worker.batch.jobs.memberllmcontext.tasklet;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.core.step.tasklet.Tasklet;
+import org.springframework.batch.repeat.RepeatStatus;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.stereotype.Component;
+import site.holliverse.worker.batch.common.support.SqlFileLoader;
+
+/**
+ * member_llm_context upsert SQL을 실제로 실행하는 tasklet.
+ *
+ * 설계상 이 step은 reader/processor/writer가 아니라,
+ * SQL 한 번으로 전체 적재를 끝내는 set-based 배치다.
+ */
+@Component
+@RequiredArgsConstructor
+public class BuildMemberLlmContextTasklet implements Tasklet {
+
+    private static final String UPSERT_SQL_PATH = "sql/member-llm-context/upsert_member_llm_context.sql";
+
+    private final NamedParameterJdbcTemplate namedParameterJdbcTemplate;
+    private final SqlFileLoader sqlFileLoader;
+
+    @Override
+    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) {
+        String snapshotDate = chunkContext.getStepContext()
+                .getStepExecution()
+                .getJobExecution()
+                .getExecutionContext()
+                .getString("snapshotDate");
+
+        String yyyymm = chunkContext.getStepContext()
+                .getStepExecution()
+                .getJobExecution()
+                .getExecutionContext()
+                .getString("yyyymm");
+
+        /**
+         * SQL 내부에서 기준일과 전월 집계월을 함께 사용한다.
+         * snapshotDate는 나이/활성 구독/약정/로그 범위 계산에 쓰이고,
+         * yyyymm은 usage_monthly 전월 데이터 조회에 쓰인다.
+         */
+        MapSqlParameterSource params = new MapSqlParameterSource()
+                .addValue("snapshotDate", snapshotDate)
+                .addValue("yyyymm", yyyymm);
+
+        /**
+         * 예외 처리 이유:
+         * - SQL 파일을 읽지 못하면 실제 적재 로직 자체가 없다는 뜻이므로 바로 실패해야 한다.
+         * - SQL 실행 중 예외가 나면 조인 대상 테이블, 타입, 데이터 형식, 제약조건 중 하나가 어긋난 상황일 가능성이 크다.
+         * - 이 step은 핵심 적재 단계이므로 예외를 삼키지 않고 상위로 그대로 올려 배치를 실패시키는 편이 맞다.
+         */
+        String upsertSql = sqlFileLoader.load(UPSERT_SQL_PATH);
+        namedParameterJdbcTemplate.update(upsertSql, params);
+        return RepeatStatus.FINISHED;
+    }
+}

src/main/java/site/holliverse/worker/batch/jobs/memberllmcontext/tasklet/MemberLlmContextGateTasklet.java

@@ -0,0 +1,124 @@
+package site.holliverse.worker.batch.jobs.memberllmcontext.tasklet;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.core.step.tasklet.Tasklet;
+import org.springframework.batch.repeat.RepeatStatus;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * member_llm_context 배치 실행 전 공통 값을 준비하는 tasklet.
+ *
+ * 역할:
+ * - snapshotDate 파라미터를 읽고 기본값을 보정한다.
+ * - usage_monthly 조회에 쓸 전월 yyyymm을 계산한다.
+ * - 실제 적재에 필요한 필수 테이블이 모두 있는지 확인한다.
+ */
+@Component
+@RequiredArgsConstructor
+public class MemberLlmContextGateTasklet implements Tasklet {
+
+    private static final ZoneId KST = ZoneId.of("Asia/Seoul");
+    private static final DateTimeFormatter YYYYMM = DateTimeFormatter.ofPattern("yyyyMM");
+
+    /**
+     * 현재 잡이 정상적으로 동작하려면 반드시 필요한 테이블 목록.
+     * churn 스냅샷이 생긴 이후에는 해당 테이블도 필수로 본다.
+     */
+    private static final List<String> REQUIRED_TABLES = List.of(
+            "member",
+            "subscription",
+            "product",
+            "mobile_plan",
+            "support_case",
+            "usage_monthly",
+            "user_event_features_7d",
+            "index_persona_snapshot",
+            "index_tscore_snapshot",
+            "churn_score_snapshot",
+            "member_llm_context"
+    );
+
+    private final JdbcTemplate jdbcTemplate;
+
+    @Override
+    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) {
+        String snapshotDateParam = (String) chunkContext.getStepContext()
+                .getJobParameters()
+                .get("snapshotDate");
+
+        LocalDate snapshotDate = resolveSnapshotDate(snapshotDateParam);
+        String yyyymm = snapshotDate.minusMonths(1).format(YYYYMM);
+
+        chunkContext.getStepContext()
+                .getStepExecution()
+                .getJobExecution()
+                .getExecutionContext()
+                .putString("snapshotDate", snapshotDate.toString());
+
+        chunkContext.getStepContext()
+                .getStepExecution()
+                .getJobExecution()
+                .getExecutionContext()
+                .putString("yyyymm", yyyymm);
+
+        verifyRequiredTables();
+        return RepeatStatus.FINISHED;
+    }
+
+    /**
+     * snapshotDate를 yyyy-MM-dd 형식으로 파싱한다.
+     * 값이 없으면 한국 시간 기준 오늘 날짜를 사용한다.
+     *
+     * 예외 처리 이유:
+     * - 배치 기준일이 잘못되면 이후 usage, 로그 14일 범위, 약정 계산이 전부 틀어진다.
+     * - 그래서 형식이 맞지 않으면 조용히 보정하지 않고 즉시 실패시킨다.
+     */
+    private LocalDate resolveSnapshotDate(String snapshotDateParam) {
+        if (snapshotDateParam == null || snapshotDateParam.isBlank()) {
+            return LocalDate.now(KST);
+        }
+
+        try {
+            return LocalDate.parse(snapshotDateParam);
+        } catch (DateTimeParseException e) {
+            throw new IllegalArgumentException(
+                    "snapshotDate 파라미터 형식이 올바르지 않습니다. yyyy-MM-dd 형식을 사용하세요.",
+                    e
+            );
+        }
+    }
+
+    /**
+     * 필수 테이블이 하나라도 빠져 있으면 즉시 실패시키되,
+     * 운영에서 한 번에 원인을 볼 수 있도록 누락 목록 전체를 함께 보여준다.
+     *
+     * 예외 처리 이유:
+     * - 이 잡은 여러 테이블을 동시에 조인하므로 필수 테이블 하나만 없어도 중간 step에서 실패한다.
+     * - Upsert step까지 갔다가 SQL 오류로 터지는 것보다 Gate 단계에서 빠르게 실패시키는 편이 원인 파악이 쉽다.
+     */
+    private void verifyRequiredTables() {
+        List<String> missingTables = new ArrayList<>();
+        for (String table : REQUIRED_TABLES) {
+            String regClass = jdbcTemplate.queryForObject("select to_regclass(?)", String.class, table);
+            if (regClass == null) {
+                missingTables.add(table);
+            }
+        }
+
+        if (!missingTables.isEmpty()) {
+            throw new IllegalStateException(
+                    "member_llm_context 배치 실행에 필요한 테이블이 없습니다: " + String.join(", ", missingTables)
+            );
+        }
+    }
+}

src/main/java/site/holliverse/worker/batch/jobs/memberllmcontext/tasklet/VerifyMemberLlmContextTasklet.java

@@ -0,0 +1,104 @@
+package site.holliverse.worker.batch.jobs.memberllmcontext.tasklet;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.core.step.tasklet.Tasklet;
+import org.springframework.batch.repeat.RepeatStatus;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.stereotype.Component;
+import site.holliverse.worker.batch.common.support.SqlFileLoader;
+
+import java.util.Map;
+
+/**
+ * member_llm_context 적재 결과를 최소 기준으로 검증하는 tasklet.
+ *
+ * 이 step의 목적은 "SQL이 실행됐다" 수준에서 멈추지 않고,
+ * 실제 결과가 우리가 기대한 형태로 들어갔는지 빠르게 확인하는 데 있다.
+ *
+ * 현재 검증 항목은 다음과 같다.
+ * 1. 적재 대상자 수와 실제 적재 row 수가 같은가
+ * 2. PK(member_id)가 비어 있는 row가 없는가
+ * 3. segment가 허용값(CHURN_RISK, UPSELL, NORMAL) 안에 있는가
+ * 4. current_product_types가 null 없이 채워졌는가
+ */
+@Component
+@RequiredArgsConstructor
+public class VerifyMemberLlmContextTasklet implements Tasklet {
+
+    /**
+     * 검증 SQL 파일 경로.
+     *
+     * SQL에서 적재 대상 건수, 실제 적재 건수, null 건수, 잘못된 segment 건수를
+     * 한 번에 가져와서 자바 쪽에서 판정한다.
+     */
+    private static final String VERIFY_SQL_PATH = "sql/member-llm-context/verify_member_llm_context.sql";
+
+    private final NamedParameterJdbcTemplate namedParameterJdbcTemplate;
+    private final SqlFileLoader sqlFileLoader;
+
+    @Override
+    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) {
+        // 검증 SQL을 읽어 현재 적재 상태를 한 번에 조회한다.
+        String verifySql = sqlFileLoader.load(VERIFY_SQL_PATH);
+        Map<String, Object> row = namedParameterJdbcTemplate.queryForMap(verifySql, new MapSqlParameterSource());
+
+        long eligibleCount = getLong(row, "eligible_count");
+        long contextCount = getLong(row, "context_count");
+        long nullPkCount = getLong(row, "null_pk_count");
+        long invalidSegmentCount = getLong(row, "invalid_segment_count");
+        long nullProductTypesCount = getLong(row, "null_product_types_count");
+
+        // 현재 배치 기준 적재 대상자 수와 실제 member_llm_context row 수가 달라지면
+        // 중간 누락 또는 과적재가 발생한 것이므로 바로 실패시킨다.
+        if (eligibleCount != contextCount) {
+            throw new IllegalStateException(String.format(
+                    "member_llm_context 건수 검증에 실패했습니다. 적재 대상 수=%d, 실제 적재 수=%d",
+                    eligibleCount,
+                    contextCount
+            ));
+        }
+
+        // PK는 절대 비면 안 된다.
+        if (nullPkCount > 0) {
+            throw new IllegalStateException(
+                    "member_llm_context PK(member_id) null 검증에 실패했습니다. nullPkCount=" + nullPkCount
+            );
+        }
+
+        // segment는 허용된 세 값만 들어가야 한다.
+        if (invalidSegmentCount > 0) {
+            throw new IllegalStateException(
+                    "member_llm_context segment 값 검증에 실패했습니다. invalidSegmentCount=" + invalidSegmentCount
+            );
+        }
+
+        // current_product_types는 후속 추천/프롬프트 구성에 바로 쓰이므로 null이면 안 된다.
+        if (nullProductTypesCount > 0) {
+            throw new IllegalStateException(
+                    "member_llm_context current_product_types null 검증에 실패했습니다. nullProductTypesCount="
+                            + nullProductTypesCount
+            );
+        }
+
+        return RepeatStatus.FINISHED;
+    }
+
+    /**
+     * queryForMap 결과에서 숫자 값을 long으로 안전하게 꺼낸다.
+     *
+     * 검증 SQL이 바뀌었거나 예상과 다른 타입이 들어오면
+     * 조용히 넘어가지 않고 즉시 실패시켜 원인을 빨리 드러낸다.
+     */
+    private long getLong(Map<String, Object> row, String key) {
+        Object value = row.get(key);
+        if (value instanceof Number number) {
+            return number.longValue();
+        }
+        throw new IllegalStateException(
+                "검증 SQL 결과를 숫자로 변환하지 못했습니다. key=" + key + ", value=" + value
+        );
+    }
+}