From dca6f717c94a29c8f6f287d033fcab2549bdbc59 Mon Sep 17 00:00:00 2001 From: andrew Date: Mon, 26 Jan 2026 17:15:50 +0900 Subject: [PATCH] =?UTF-8?q?UPLUS-150=20fix=20:=20Processor=20message=20?= =?UTF-8?q?=EB=B3=80=ED=99=98=20=EB=A1=9C=EC=A7=81=20=EC=88=98=EC=A0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../processor/NotificationSendProcessor.java | 2 +- .../invoicesend/job/InvoiceSendJobConfig.java | 4 +- .../model/InvoiceAggregateRecord.java | 18 ---- .../model/InvoiceNotificationEvent.java | 26 ++++++ .../processor/InvoiceSendProcessor.java | 84 +++++++++++++------ .../invoicesend/writer/InvoiceSendWriter.java | 14 ++-- 6 files changed, 95 insertions(+), 53 deletions(-) delete mode 100644 src/main/java/com/template/worker/jobs/invoicesend/model/InvoiceAggregateRecord.java create mode 100644 src/main/java/com/template/worker/jobs/invoicesend/model/InvoiceNotificationEvent.java diff --git a/src/main/java/com/template/worker/batch/notificationsend/processor/NotificationSendProcessor.java b/src/main/java/com/template/worker/batch/notificationsend/processor/NotificationSendProcessor.java index 450a8b5..63b939c 100644 --- a/src/main/java/com/template/worker/batch/notificationsend/processor/NotificationSendProcessor.java +++ b/src/main/java/com/template/worker/batch/notificationsend/processor/NotificationSendProcessor.java @@ -22,7 +22,7 @@ public NotificationMessage process(UsageNotificationOutboxRow item) { return new NotificationMessage( UUID.randomUUID(), item.id(), - 101L, + 2L, Map.of( "subId", item.subId(), "phoneNumber", item.phoneNumber(), diff --git a/src/main/java/com/template/worker/jobs/invoicesend/job/InvoiceSendJobConfig.java b/src/main/java/com/template/worker/jobs/invoicesend/job/InvoiceSendJobConfig.java index 8122f8b..86648ba 100644 --- a/src/main/java/com/template/worker/jobs/invoicesend/job/InvoiceSendJobConfig.java +++ b/src/main/java/com/template/worker/jobs/invoicesend/job/InvoiceSendJobConfig.java @@ -13,7 +13,7 @@ import org.springframework.transaction.PlatformTransactionManager; import com.template.worker.global.listener.JobResultListener; -import com.template.worker.jobs.invoicesend.model.InvoiceAggregateRecord; +import com.template.worker.jobs.invoicesend.model.InvoiceNotificationEvent; import com.template.worker.jobs.invoicesend.model.InvoiceSendRecord; import com.template.worker.jobs.invoicesend.processor.InvoiceSendProcessor; import com.template.worker.jobs.invoicesend.writer.InvoiceSendWriter; @@ -45,7 +45,7 @@ public Job invoiceSendJob(Step invoiceSendStep) { @Bean public Step invoiceSendStep(JdbcCursorItemReader invoiceJoinReader) { return new StepBuilder("invoiceSendStep", jobRepository) - .chunk(chunkSize, transactionManager) + .chunk(chunkSize, transactionManager) .reader(invoiceJoinReader) .processor(invoiceSendProcessor) .writer(invoiceSendWriter) diff --git a/src/main/java/com/template/worker/jobs/invoicesend/model/InvoiceAggregateRecord.java b/src/main/java/com/template/worker/jobs/invoicesend/model/InvoiceAggregateRecord.java deleted file mode 100644 index 4cf6723..0000000 --- a/src/main/java/com/template/worker/jobs/invoicesend/model/InvoiceAggregateRecord.java +++ /dev/null @@ -1,18 +0,0 @@ -package com.template.worker.jobs.invoicesend.model; - -import java.time.LocalDateTime; -import java.util.List; - -public record InvoiceAggregateRecord( - Long templateGroupId, - Long eventId, - Long invId, - Long subId, - String invMonth, - String phoneEnc, - String emailEnc, - String name, - Integer totalPrice, - LocalDateTime createdAt, - LocalDateTime dueDate, - List items) {} diff --git a/src/main/java/com/template/worker/jobs/invoicesend/model/InvoiceNotificationEvent.java b/src/main/java/com/template/worker/jobs/invoicesend/model/InvoiceNotificationEvent.java new file mode 100644 index 0000000..eac9c7a --- /dev/null +++ b/src/main/java/com/template/worker/jobs/invoicesend/model/InvoiceNotificationEvent.java @@ -0,0 +1,26 @@ +package com.template.worker.jobs.invoicesend.model; + +import java.util.List; +import java.util.UUID; + +public record InvoiceNotificationEvent( + UUID eventId, + Long templateGroupId, + SubscriptionInfo subscriptionInfo, + Variables variables) { + + public record SubscriptionInfo(Long subId, String phoneNumber, String email) {} + + public record Variables( + String invoiceNumber, + String customerName, + String phoneNumber, + String planName, + String billingDate, + String dueDate, + Integer totalAmount, + String paymentStatus, + List items) {} + + public record Item(String name, String detail, Integer amount) {} +} diff --git a/src/main/java/com/template/worker/jobs/invoicesend/processor/InvoiceSendProcessor.java b/src/main/java/com/template/worker/jobs/invoicesend/processor/InvoiceSendProcessor.java index 9805d9e..1496cd7 100644 --- a/src/main/java/com/template/worker/jobs/invoicesend/processor/InvoiceSendProcessor.java +++ b/src/main/java/com/template/worker/jobs/invoicesend/processor/InvoiceSendProcessor.java @@ -1,30 +1,34 @@ package com.template.worker.jobs.invoicesend.processor; +import java.time.format.DateTimeFormatter; import java.util.ArrayList; import java.util.List; +import java.util.UUID; import org.springframework.batch.core.configuration.annotation.StepScope; import org.springframework.batch.item.ItemProcessor; import org.springframework.stereotype.Component; -import com.template.worker.jobs.invoicesend.model.InvoiceAggregateRecord; import com.template.worker.jobs.invoicesend.model.InvoiceItemRecord; +import com.template.worker.jobs.invoicesend.model.InvoiceNotificationEvent; import com.template.worker.jobs.invoicesend.model.InvoiceSendRecord; @Component @StepScope public class InvoiceSendProcessor - implements ItemProcessor { + implements ItemProcessor { + + private static final DateTimeFormatter DATE_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd"); private Long currentInvId; private InvoiceSendRecord currentInvoice; private List currentItems; @Override - public InvoiceAggregateRecord process(InvoiceSendRecord row) { + public InvoiceNotificationEvent process(InvoiceSendRecord row) { if (currentInvId == null) { - startNewInvoice(row); + start(row); return null; } @@ -33,14 +37,12 @@ public InvoiceAggregateRecord process(InvoiceSendRecord row) { return null; } - InvoiceAggregateRecord finished = buildAggregate(); - - startNewInvoice(row); - + InvoiceNotificationEvent finished = buildEvent(); + start(row); return finished; } - private void startNewInvoice(InvoiceSendRecord row) { + private void start(InvoiceSendRecord row) { this.currentInvId = row.invId(); this.currentInvoice = row; this.currentItems = new ArrayList<>(4); @@ -56,26 +58,58 @@ private void addItem(InvoiceSendRecord row) { row.itemValue().intValue())); } - public InvoiceAggregateRecord flushLast() { + public InvoiceNotificationEvent flushLast() { if (currentInvId == null) { return null; } - return buildAggregate(); + return buildEvent(); + } + + private InvoiceNotificationEvent buildEvent() { + + String planName = + currentItems.stream() + .filter(i -> "PLAN".equals(i.invoiceType())) + .map(InvoiceItemRecord::invoiceName) + .findFirst() + .orElse(null); + + List items = + currentItems.stream() + .map( + i -> + new InvoiceNotificationEvent.Item( + mapItemName(i.invoiceType()), + i.invoiceName(), + i.value())) + .toList(); + + return new InvoiceNotificationEvent( + UUID.randomUUID(), + 1L, // templateGroupId 고정 + new InvoiceNotificationEvent.SubscriptionInfo( + currentInvoice.subId(), + currentInvoice.phone_enc(), + currentInvoice.email_enc()), + new InvoiceNotificationEvent.Variables( + currentInvoice.invNo().toString(), + currentInvoice.name(), + currentInvoice.phone_enc(), + planName, + currentInvoice.createdAt().format(DATE_FORMAT), + currentInvoice.dueDate().format(DATE_FORMAT), + currentInvoice.totalPrice(), + "PAID", + items)); } - private InvoiceAggregateRecord buildAggregate() { - return new InvoiceAggregateRecord( - 2L, - currentInvoice.invNo(), - currentInvoice.invId(), - currentInvoice.subId(), - currentInvoice.invMonth(), - currentInvoice.phone_enc(), - currentInvoice.email_enc(), - currentInvoice.name(), - currentInvoice.totalPrice(), - currentInvoice.createdAt(), - currentInvoice.dueDate(), - List.copyOf(currentItems)); + private String mapItemName(String type) { + return switch (type) { + case "PLAN" -> "기본 요금"; + case "VAS" -> "부가서비스"; + case "DISCOUNT" -> "할인"; + case "MICRO" -> "소액 결제"; + default -> "기타"; + }; } } diff --git a/src/main/java/com/template/worker/jobs/invoicesend/writer/InvoiceSendWriter.java b/src/main/java/com/template/worker/jobs/invoicesend/writer/InvoiceSendWriter.java index b0ba2a8..c4f76f4 100644 --- a/src/main/java/com/template/worker/jobs/invoicesend/writer/InvoiceSendWriter.java +++ b/src/main/java/com/template/worker/jobs/invoicesend/writer/InvoiceSendWriter.java @@ -7,7 +7,7 @@ import org.springframework.stereotype.Component; import com.fasterxml.jackson.databind.ObjectMapper; -import com.template.worker.jobs.invoicesend.model.InvoiceAggregateRecord; +import com.template.worker.jobs.invoicesend.model.InvoiceNotificationEvent; import com.template.worker.jobs.invoicesend.processor.InvoiceSendProcessor; import lombok.RequiredArgsConstructor; @@ -16,7 +16,7 @@ @Slf4j @Component @RequiredArgsConstructor -public class InvoiceSendWriter implements ItemWriter, ItemStream { +public class InvoiceSendWriter implements ItemWriter, ItemStream { private static final String TOPIC = "invoice-noti"; @@ -24,12 +24,12 @@ public class InvoiceSendWriter implements ItemWriter, It private final ObjectMapper objectMapper; private final InvoiceSendProcessor processor; - private InvoiceAggregateRecord lastBuffered; + private InvoiceNotificationEvent lastBuffered; @Override - public void write(Chunk items) throws Exception { + public void write(Chunk items) throws Exception { - for (InvoiceAggregateRecord invoice : items) { + for (InvoiceNotificationEvent invoice : items) { send(invoice); } @@ -53,8 +53,8 @@ public void close() { } } - private void send(InvoiceAggregateRecord invoice) throws Exception { + private void send(InvoiceNotificationEvent invoice) throws Exception { String payload = objectMapper.writeValueAsString(invoice); - kafkaTemplate.send(TOPIC, String.valueOf(invoice.invId()), payload); + kafkaTemplate.send(TOPIC, String.valueOf(invoice.eventId()), payload); } }