diff --git a/README.md b/README.md index 6681383..d24cc19 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,43 @@ worker가 송금사 호출 결과를 `KafkaTemplate.send(...)`로 바로 발행 따라서 worker도 결과를 자체 `payout_outbox`에 트랜잭션 INSERT 하고 Debezium이 Kafka로 흘리는 동일한 패턴을 적용합니다. 송금사 호출 후 worker가 죽어도 outbox 행은 DB에 살아있고, 재기동 시 또는 다음 binlog tail에서 자연스럽게 발행됩니다. +## 정산 배치 (`reconciler`) — 왜 A·B 두 검증을 모두 돌리는가 + +`잔액 = Σ(거래 내역)` 이라는 한 줄짜리 정합성 명제는 실제 시스템에서는 **같은 BigDecimal 비교지만 잡는 결함의 종류가 서로 다른 두 검증** 으로 쪼개집니다. reconciler 는 매일 두 검증을 모두 수행해 `reconciliations.mismatch_count` 에 기록합니다. + +**A-검증 (총합 무결성)** + +``` +wallet.balance == Σ wallet_transactions.amount // 양수=입금, 음수=출금 +``` + +지갑의 모든 ledger row 의 amount 합과 현재 잔액이 일치하는지. **누락되거나 오기록된 ledger row** 를 잡습니다. 예: 송금 use case 에서 wallet.withdraw 만 하고 ledger insert 가 빠진 코드 경로 — 잔액은 차감됐지만 ledger row 가 없어 합이 안 맞습니다. + +**B-검증 (마지막 갱신 무결성)** + +``` +wallet.balance == 마지막 wallet_transactions.balance_after // id DESC LIMIT 1 +``` + +가장 최근 ledger row 의 `balance_after` 스냅샷과 현재 잔액이 일치하는지. **race 로 인한 last-write 손실, 트랜잭션 외부에서 일어난 balance 직접 수정** 을 잡습니다. 예: 운영자가 SQL 콘솔에서 `UPDATE wallets SET balance = ...` 를 직접 친 경우 — ledger 합은 그대로지만 마지막 balance_after 와 어긋납니다. + +**왜 둘 중 하나만으로는 부족한가** + +| 결함 시나리오 | A 만 돌리면 | B 만 돌리면 | A+B 둘 다 | +|---|---|---|---| +| ledger row 1건 누락 | 탐지 (합 불일치) | 탐지 못 함 (마지막 balance_after 는 우연히 일치 가능) | 탐지 | +| balance 직접 SQL 수정 | 탐지 못 함 (합은 그대로) | 탐지 (마지막 balance_after 와 어긋남) | 탐지 | +| 양쪽 모두 부정합 | 탐지 | 탐지 | 탐지 (두 플래그 모두 켜짐) | + +A 와 B 는 **서로 다른 클래스의 결함** 을 잡습니다. ledger 가 단일 출처(single source of truth)라면 둘이 동일해 보이지만, 실제 incident 는 ledger 외부 경로(콘솔, 데이터 패치, 잘못된 마이그레이션, race condition)에서도 발생합니다. 두 검증 모두 통과해야 비로소 "잔액 = Σ(거래 내역)" 의 의미적 보장이 성립합니다. + +mismatch 가 발견되어도 잡 자체는 SUCCESS 로 종료합니다. `reconciliations.mismatch_count > 0` 을 별도 alerting 트리거로 쓰는 편이, "잡 결함" 과 "데이터 결함" 을 메타데이터 측면에서 구분할 수 있어 운영상 명확합니다. + +실행: +```bash +./gradlew :reconciler:bootRun # 매일 04:00 KST 자동 (openremit.reconcile.cron 으로 변경) +``` + ## 테스트 ```bash diff --git a/reconciler/build.gradle.kts b/reconciler/build.gradle.kts index 4ddc986..5e0911b 100644 --- a/reconciler/build.gradle.kts +++ b/reconciler/build.gradle.kts @@ -1,6 +1,30 @@ -// 정산 배치 (Spring Batch) -// Day 10에 Spring Boot 플러그인 + 배치 구성 추가 예정. +plugins { + id("org.springframework.boot") + kotlin("plugin.jpa") +} + +allOpen { + annotation("jakarta.persistence.Entity") + annotation("jakarta.persistence.MappedSuperclass") + annotation("jakarta.persistence.Embeddable") +} dependencies { implementation(project(":common")) + + implementation("org.springframework.boot:spring-boot-starter-actuator") + implementation("org.springframework.boot:spring-boot-starter-data-jpa") + implementation("org.springframework.boot:spring-boot-starter-flyway") + implementation("org.springframework.boot:spring-boot-starter-batch") + implementation("org.springframework.boot:spring-boot-starter-json") + implementation("org.flywaydb:flyway-mysql") + runtimeOnly("com.mysql:mysql-connector-j") + + testImplementation("org.springframework.boot:spring-boot-starter-actuator-test") + testImplementation("org.springframework.boot:spring-boot-starter-data-jpa-test") + testImplementation("org.springframework.boot:spring-boot-starter-flyway-test") + testImplementation("org.springframework.boot:spring-boot-testcontainers") + testImplementation("org.springframework.boot:spring-boot-starter-batch-test") + testImplementation("org.testcontainers:junit-jupiter") + testImplementation("org.testcontainers:mysql") } diff --git a/reconciler/src/main/kotlin/com/openremit/reconcile/ReconcilerApplication.kt b/reconciler/src/main/kotlin/com/openremit/reconcile/ReconcilerApplication.kt new file mode 100644 index 0000000..9e07123 --- /dev/null +++ b/reconciler/src/main/kotlin/com/openremit/reconcile/ReconcilerApplication.kt @@ -0,0 +1,13 @@ +package com.openremit.reconcile + +import org.springframework.boot.autoconfigure.SpringBootApplication +import org.springframework.boot.runApplication +import org.springframework.scheduling.annotation.EnableScheduling + +@SpringBootApplication +@EnableScheduling +class ReconcilerApplication + +fun main(args: Array) { + runApplication(*args) +} diff --git a/reconciler/src/main/kotlin/com/openremit/reconcile/application/WalletReconcile.kt b/reconciler/src/main/kotlin/com/openremit/reconcile/application/WalletReconcile.kt new file mode 100644 index 0000000..4072b28 --- /dev/null +++ b/reconciler/src/main/kotlin/com/openremit/reconcile/application/WalletReconcile.kt @@ -0,0 +1,32 @@ +package com.openremit.reconcile.application + +import java.math.BigDecimal + +/** + * 단일 wallet의 정산 입력 스냅샷. + * - balance: 현재 wallet.balance + * - sumOfTransactions: 해당 wallet의 모든 wallet_transactions.amount 합 (A-검증) + * - lastBalanceAfter: 가장 최근 wallet_transaction의 balance_after, 거래 0건이면 null (B-검증) + */ +data class WalletSnapshot( + val walletId: Long, + val balance: BigDecimal, + val sumOfTransactions: BigDecimal, + val lastBalanceAfter: BigDecimal?, +) + +data class WalletMismatch( + val walletId: Long, + val balance: BigDecimal, + val sumOfTransactions: BigDecimal, + val lastBalanceAfter: BigDecimal?, + val violatesA: Boolean, + val violatesB: Boolean, +) + +data class ReconcileResult( + val totalCount: Int, + val mismatches: List, +) { + val mismatchCount: Int get() = mismatches.size +} diff --git a/reconciler/src/main/kotlin/com/openremit/reconcile/application/WalletReconcileService.kt b/reconciler/src/main/kotlin/com/openremit/reconcile/application/WalletReconcileService.kt new file mode 100644 index 0000000..a16e203 --- /dev/null +++ b/reconciler/src/main/kotlin/com/openremit/reconcile/application/WalletReconcileService.kt @@ -0,0 +1,42 @@ +package com.openremit.reconcile.application + +import org.springframework.stereotype.Service + +/** + * 정산 검증 도메인 서비스. + * + * A-검증: wallet.balance == Σ wallet_transactions.amount + * - "총합 무결성" — 누락/오기록된 ledger row를 잡는다. + * + * B-검증: wallet.balance == 마지막 wallet_transaction.balance_after + * - "마지막 갱신 무결성" — race로 인한 last-write 손실, 트랜잭션 외부에서 일어난 + * balance 직접 수정을 잡는다. 거래가 0건이면 balance 도 0이어야 한다. + * + * 두 검증 중 하나라도 위반하면 mismatch. + */ +@Service +class WalletReconcileService { + + fun reconcile(snapshots: List): ReconcileResult { + val mismatches = snapshots.mapNotNull { evaluate(it) } + return ReconcileResult(totalCount = snapshots.size, mismatches = mismatches) + } + + private fun evaluate(s: WalletSnapshot): WalletMismatch? { + val violatesA = s.balance.compareTo(s.sumOfTransactions) != 0 + val violatesB = if (s.lastBalanceAfter == null) { + s.balance.signum() != 0 + } else { + s.balance.compareTo(s.lastBalanceAfter) != 0 + } + if (!violatesA && !violatesB) return null + return WalletMismatch( + walletId = s.walletId, + balance = s.balance, + sumOfTransactions = s.sumOfTransactions, + lastBalanceAfter = s.lastBalanceAfter, + violatesA = violatesA, + violatesB = violatesB, + ) + } +} diff --git a/reconciler/src/main/kotlin/com/openremit/reconcile/domain/Reconciliation.kt b/reconciler/src/main/kotlin/com/openremit/reconcile/domain/Reconciliation.kt new file mode 100644 index 0000000..adb210f --- /dev/null +++ b/reconciler/src/main/kotlin/com/openremit/reconcile/domain/Reconciliation.kt @@ -0,0 +1,36 @@ +package com.openremit.reconcile.domain + +import jakarta.persistence.Column +import jakarta.persistence.Entity +import jakarta.persistence.GeneratedValue +import jakarta.persistence.GenerationType +import jakarta.persistence.Id +import jakarta.persistence.Table +import org.hibernate.annotations.JdbcTypeCode +import org.hibernate.type.SqlTypes +import java.time.Instant +import java.time.LocalDate + +@Entity +@Table(name = "reconciliations") +class Reconciliation( + @Column(name = "target_date", nullable = false) + val targetDate: LocalDate, + + @Column(name = "total_count", nullable = false) + val totalCount: Int, + + @Column(name = "mismatch_count", nullable = false) + val mismatchCount: Int, + + @JdbcTypeCode(SqlTypes.JSON) + @Column(name = "details", nullable = false, columnDefinition = "json") + val details: String, + + @Column(name = "created_at", nullable = false) + val createdAt: Instant = Instant.now(), +) { + @Id + @GeneratedValue(strategy = GenerationType.IDENTITY) + var id: Long = 0 +} diff --git a/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/batch/ReconcileScheduler.kt b/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/batch/ReconcileScheduler.kt new file mode 100644 index 0000000..1ed83a4 --- /dev/null +++ b/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/batch/ReconcileScheduler.kt @@ -0,0 +1,30 @@ +package com.openremit.reconcile.infrastructure.batch + +import org.slf4j.LoggerFactory +import org.springframework.batch.core.job.Job +import org.springframework.batch.core.job.parameters.JobParametersBuilder +import org.springframework.batch.core.launch.JobOperator +import org.springframework.beans.factory.annotation.Qualifier +import org.springframework.scheduling.annotation.Scheduled +import org.springframework.stereotype.Component + +/** + * 매일 정해진 시각에 정산 잡을 실행한다. + * 같은 잡을 같은 파라미터로는 Spring Batch 가 재실행을 막으므로 timestamp 를 파라미터로 넣어 매 실행을 구분한다. + */ +@Component +class ReconcileScheduler( + private val jobOperator: JobOperator, + @Qualifier("walletReconcileJob") private val walletReconcileJob: Job, +) { + private val log = LoggerFactory.getLogger(javaClass) + + @Scheduled(cron = "\${openremit.reconcile.cron:0 0 4 * * *}") + fun run() { + val params = JobParametersBuilder() + .addLong("runAt", System.currentTimeMillis()) + .toJobParameters() + log.info("triggering walletReconcileJob") + jobOperator.start(walletReconcileJob, params) + } +} diff --git a/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/batch/WalletReconcileJobConfig.kt b/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/batch/WalletReconcileJobConfig.kt new file mode 100644 index 0000000..5e224ca --- /dev/null +++ b/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/batch/WalletReconcileJobConfig.kt @@ -0,0 +1,75 @@ +package com.openremit.reconcile.infrastructure.batch + +import com.openremit.reconcile.application.WalletReconcileService +import com.openremit.reconcile.domain.Reconciliation +import com.openremit.reconcile.infrastructure.persistence.ReconciliationRepository +import com.openremit.reconcile.infrastructure.persistence.WalletReconcileQuery +import org.slf4j.LoggerFactory +import org.springframework.batch.core.job.Job +import org.springframework.batch.core.job.builder.JobBuilder +import org.springframework.batch.core.repository.JobRepository +import org.springframework.batch.core.step.Step +import org.springframework.batch.core.step.builder.StepBuilder +import org.springframework.batch.infrastructure.repeat.RepeatStatus +import org.springframework.context.annotation.Bean +import org.springframework.context.annotation.Configuration +import org.springframework.transaction.PlatformTransactionManager +import tools.jackson.databind.ObjectMapper +import java.time.LocalDate +import java.time.ZoneId + +/** + * 정산 배치 Job 구성. + * + * - 단일 Tasklet Step. wallet 수가 많지 않은 시연 환경에서는 chunk 가 불필요한 오버헤드. + * - mismatch 발견되어도 Step 은 SUCCESS 로 종료한다 (alert 트리거는 mismatch_count > 0 으로 판단). + * Step 을 FAILED 로 만들면 Spring Batch 메타데이터 측면에서 "잡 자체의 결함" 과 구분이 안 된다. + */ +@Configuration +class WalletReconcileJobConfig { + + @Bean + fun walletReconcileJob( + jobRepository: JobRepository, + walletReconcileStep: Step, + ): Job = + JobBuilder("walletReconcileJob", jobRepository) + .start(walletReconcileStep) + .build() + + @Bean + fun walletReconcileStep( + jobRepository: JobRepository, + transactionManager: PlatformTransactionManager, + query: WalletReconcileQuery, + service: WalletReconcileService, + reconciliationRepository: ReconciliationRepository, + objectMapper: ObjectMapper, + ): Step { + val log = LoggerFactory.getLogger("walletReconcileStep") + return StepBuilder("walletReconcileStep", jobRepository) + .tasklet({ _, _ -> + val snapshots = query.loadSnapshots() + val result = service.reconcile(snapshots) + val details = objectMapper.writeValueAsString(result.mismatches) + reconciliationRepository.save( + Reconciliation( + targetDate = LocalDate.now(ZoneId.systemDefault()), + totalCount = result.totalCount, + mismatchCount = result.mismatchCount, + details = details, + ) + ) + if (result.mismatchCount > 0) { + log.warn( + "wallet reconciliation found {} mismatches out of {}: {}", + result.mismatchCount, result.totalCount, details, + ) + } else { + log.info("wallet reconciliation OK ({} wallets)", result.totalCount) + } + RepeatStatus.FINISHED + }, transactionManager) + .build() + } +} diff --git a/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/persistence/ReconciliationRepository.kt b/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/persistence/ReconciliationRepository.kt new file mode 100644 index 0000000..cf41ede --- /dev/null +++ b/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/persistence/ReconciliationRepository.kt @@ -0,0 +1,6 @@ +package com.openremit.reconcile.infrastructure.persistence + +import com.openremit.reconcile.domain.Reconciliation +import org.springframework.data.jpa.repository.JpaRepository + +interface ReconciliationRepository : JpaRepository diff --git a/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/persistence/WalletReconcileQuery.kt b/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/persistence/WalletReconcileQuery.kt new file mode 100644 index 0000000..8beb6aa --- /dev/null +++ b/reconciler/src/main/kotlin/com/openremit/reconcile/infrastructure/persistence/WalletReconcileQuery.kt @@ -0,0 +1,49 @@ +package com.openremit.reconcile.infrastructure.persistence + +import com.openremit.reconcile.application.WalletSnapshot +import jakarta.persistence.EntityManager +import org.springframework.stereotype.Repository +import java.math.BigDecimal + +/** + * 정산 잡이 wallet 정합성 검증을 위해 읽는 native 쿼리. + * + * reconciler 는 JPA 엔티티로 wallets / wallet_transactions 를 매핑하지 않는다 (DB 소유권 분리, ADR-011). + * 매핑하면 엔티티 정의가 reconciler / remittance-api 에 중복되거나 결합되므로, + * 정산 잡은 read-only native 쿼리로만 두 테이블에 접근한다. + */ +@Repository +class WalletReconcileQuery( + private val entityManager: EntityManager, +) { + + fun loadSnapshots(): List { + // wallet 마다 1행. 거래가 없으면 sum=0, last_balance_after=null. + // last_balance_after: id 가 가장 큰 wallet_transactions.balance_after. + val sql = """ + SELECT + w.id AS wallet_id, + w.balance AS balance, + COALESCE(SUM(wt.amount), 0) AS sum_amount, + (SELECT wt2.balance_after + FROM wallet_transactions wt2 + WHERE wt2.wallet_id = w.id + ORDER BY wt2.id DESC + LIMIT 1) AS last_balance_after + FROM wallets w + LEFT JOIN wallet_transactions wt ON wt.wallet_id = w.id + GROUP BY w.id, w.balance + """.trimIndent() + + @Suppress("UNCHECKED_CAST") + val rows = entityManager.createNativeQuery(sql).resultList as List> + return rows.map { row -> + WalletSnapshot( + walletId = (row[0] as Number).toLong(), + balance = row[1] as BigDecimal, + sumOfTransactions = row[2] as BigDecimal, + lastBalanceAfter = row[3] as BigDecimal?, + ) + } + } +} diff --git a/reconciler/src/main/resources/application.yaml b/reconciler/src/main/resources/application.yaml new file mode 100644 index 0000000..32595eb --- /dev/null +++ b/reconciler/src/main/resources/application.yaml @@ -0,0 +1,52 @@ +spring: + application: + name: reconciler + threads: + virtual: + enabled: true + jackson: + property-naming-strategy: SNAKE_CASE + datasource: + url: jdbc:mysql://localhost:3306/openremit?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC + username: openremit + password: openremit + driver-class-name: com.mysql.cj.jdbc.Driver + jpa: + hibernate: + ddl-auto: none + properties: + hibernate: + format_sql: true + show-sql: false + flyway: + enabled: true + locations: classpath:db/migration + # remittance-api와 같은 DB를 공유하지만 history 테이블은 분리 (ADR-011) + table: flyway_schema_history_reconcile + baseline-on-migrate: true + baseline-version: "0" + batch: + jdbc: + # Spring Batch 메타테이블 (BATCH_JOB_INSTANCE 등) 자동 생성 + initialize-schema: always + job: + # @Scheduled 트리거로만 실행. 부팅 시 자동 실행 방지. + enabled: false + +server: + # 8083 은 docker-compose 의 Debezium 이 사용 — 충돌을 피해 8084. + port: 8084 + +management: + endpoints: + web: + exposure: + include: health,info,metrics + endpoint: + health: + show-details: when-authorized + +openremit: + reconcile: + # 매일 04:00 정산 실행 + cron: "0 0 4 * * *" diff --git a/reconciler/src/main/resources/db/migration/V1__reconciliation_tables.sql b/reconciler/src/main/resources/db/migration/V1__reconciliation_tables.sql new file mode 100644 index 0000000..8999f4b --- /dev/null +++ b/reconciler/src/main/resources/db/migration/V1__reconciliation_tables.sql @@ -0,0 +1,13 @@ +-- reconciler 자체 소유 테이블 (ADR-011 패턴) +-- 정산 결과 저장. wallets/wallet_transactions 는 read-only로만 접근. + +CREATE TABLE reconciliations ( + id BIGINT NOT NULL AUTO_INCREMENT, + target_date DATE NOT NULL, + total_count INT NOT NULL, + mismatch_count INT NOT NULL, + details JSON NOT NULL, + created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + PRIMARY KEY (id), + INDEX idx_reconciliations_target_date (target_date) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; diff --git a/reconciler/src/test/kotlin/com/openremit/reconcile/ReconcilerTestcontainersConfig.kt b/reconciler/src/test/kotlin/com/openremit/reconcile/ReconcilerTestcontainersConfig.kt new file mode 100644 index 0000000..199a09e --- /dev/null +++ b/reconciler/src/test/kotlin/com/openremit/reconcile/ReconcilerTestcontainersConfig.kt @@ -0,0 +1,18 @@ +package com.openremit.reconcile + +import org.springframework.boot.test.context.TestConfiguration +import org.springframework.boot.testcontainers.service.connection.ServiceConnection +import org.springframework.context.annotation.Bean +import org.testcontainers.containers.MySQLContainer + +@TestConfiguration(proxyBeanMethods = false) +class ReconcilerTestcontainersConfig { + + @Bean + @ServiceConnection + fun mysqlContainer(): MySQLContainer<*> = + MySQLContainer("mysql:8.0") + .withDatabaseName("openremit") + .withUsername("test") + .withPassword("test") +} diff --git a/reconciler/src/test/kotlin/com/openremit/reconcile/WalletReconcileIntegrationTest.kt b/reconciler/src/test/kotlin/com/openremit/reconcile/WalletReconcileIntegrationTest.kt new file mode 100644 index 0000000..db4a751 --- /dev/null +++ b/reconciler/src/test/kotlin/com/openremit/reconcile/WalletReconcileIntegrationTest.kt @@ -0,0 +1,147 @@ +package com.openremit.reconcile + +import com.openremit.reconcile.infrastructure.persistence.ReconciliationRepository +import org.springframework.batch.core.BatchStatus +import org.springframework.batch.core.job.Job +import org.springframework.batch.core.job.parameters.JobParametersBuilder +import org.springframework.batch.core.launch.JobOperator +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.beans.factory.annotation.Qualifier +import org.springframework.boot.test.context.SpringBootTest +import org.springframework.context.annotation.Import +import org.springframework.jdbc.core.JdbcTemplate +import tools.jackson.databind.ObjectMapper +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * 정산 잡의 end-to-end 검증. + * + * 시나리오: 정합 wallet 1개 + 불일치 wallet 1개를 주입한 뒤 잡을 실행해 + * reconciliations 테이블에 mismatch_count=1 이 기록되는지, details JSON 에 해당 walletId 가 들어가는지 확인. + * + * 외부 (remittance-api) 소유 테이블은 reconciler 의 production schema 에 없으므로 테스트 환경에서 minimal schema 로 직접 생성한다. + * docker-compose 환경에서는 remittance-api 의 Flyway 마이그레이션이 보장한다. + */ +@SpringBootTest +@Import(ReconcilerTestcontainersConfig::class) +class WalletReconcileIntegrationTest @Autowired constructor( + private val jobOperator: JobOperator, + @Qualifier("walletReconcileJob") private val job: Job, + private val reconciliationRepository: ReconciliationRepository, + private val jdbcTemplate: JdbcTemplate, + private val objectMapper: ObjectMapper, +) { + + @BeforeTest + fun setup() { + jdbcTemplate.execute( + """ + CREATE TABLE IF NOT EXISTS wallets ( + id BIGINT NOT NULL AUTO_INCREMENT, + user_id BIGINT NOT NULL, + currency CHAR(3) NOT NULL, + balance DECIMAL(19, 4) NOT NULL DEFAULT 0, + version BIGINT NOT NULL DEFAULT 0, + updated_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + PRIMARY KEY (id) + ) ENGINE=InnoDB + """.trimIndent() + ) + jdbcTemplate.execute( + """ + CREATE TABLE IF NOT EXISTS wallet_transactions ( + id BIGINT NOT NULL AUTO_INCREMENT, + wallet_id BIGINT NOT NULL, + amount DECIMAL(19, 4) NOT NULL, + balance_after DECIMAL(19, 4) NOT NULL, + reference_type VARCHAR(20) NOT NULL, + reference_id BIGINT NOT NULL, + created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + PRIMARY KEY (id), + INDEX idx_wallet_transactions_wallet (wallet_id, id) + ) ENGINE=InnoDB + """.trimIndent() + ) + jdbcTemplate.execute("DELETE FROM wallet_transactions") + jdbcTemplate.execute("DELETE FROM wallets") + reconciliationRepository.deleteAllInBatch() + } + + @Test + fun `consistent wallet plus inconsistent wallet — job records mismatch_count = 1`() { + // 정합 wallet (id=1): balance=100, ledger 합=100, 마지막 balance_after=100 + jdbcTemplate.update( + "INSERT INTO wallets (id, user_id, currency, balance) VALUES (?, ?, ?, ?)", + 1L, 100L, "KRW", 100, + ) + jdbcTemplate.update( + """INSERT INTO wallet_transactions + (wallet_id, amount, balance_after, reference_type, reference_id) + VALUES (?, ?, ?, ?, ?)""", + 1L, 100, 100, "PAYMENT", 1L, + ) + + // 불일치 wallet (id=2): balance=900 인데 ledger 합=800 (1행 누락) → A 위반 + // 누락된 행은 의도적으로 INSERT 하지 않음. 마지막 balance_after=800 → B 도 위반. + jdbcTemplate.update( + "INSERT INTO wallets (id, user_id, currency, balance) VALUES (?, ?, ?, ?)", + 2L, 200L, "KRW", 900, + ) + jdbcTemplate.update( + """INSERT INTO wallet_transactions + (wallet_id, amount, balance_after, reference_type, reference_id) + VALUES (?, ?, ?, ?, ?)""", + 2L, 800, 800, "REMITTANCE", 1L, + ) + + val params = JobParametersBuilder() + .addLong("runAt", System.currentTimeMillis()) + .toJobParameters() + val execution = jobOperator.start(job, params) + + assertEquals(BatchStatus.COMPLETED, execution.status) + + val recs = reconciliationRepository.findAll() + assertEquals(1, recs.size) + val rec = recs.single() + assertEquals(2, rec.totalCount) + assertEquals(1, rec.mismatchCount) + + @Suppress("UNCHECKED_CAST") + val mismatches = objectMapper.readValue(rec.details, List::class.java) as List> + assertEquals(1, mismatches.size) + val m = mismatches.single() + // application.yaml 의 spring.jackson.property-naming-strategy=SNAKE_CASE 가 적용된다. + assertEquals(2L, (m["wallet_id"] as Number).toLong()) + assertEquals(true, m["violates_a"]) + assertEquals(true, m["violates_b"]) + } + + @Test + fun `all wallets consistent — mismatch_count = 0`() { + jdbcTemplate.update( + "INSERT INTO wallets (id, user_id, currency, balance) VALUES (?, ?, ?, ?)", + 10L, 1000L, "KRW", 500, + ) + jdbcTemplate.update( + """INSERT INTO wallet_transactions + (wallet_id, amount, balance_after, reference_type, reference_id) + VALUES (?, ?, ?, ?, ?)""", + 10L, 500, 500, "PAYMENT", 1L, + ) + + val params = JobParametersBuilder() + .addLong("runAt", System.currentTimeMillis()) + .toJobParameters() + val execution = jobOperator.start(job, params) + + assertEquals(BatchStatus.COMPLETED, execution.status) + val rec = reconciliationRepository.findAll().single() + assertEquals(1, rec.totalCount) + assertEquals(0, rec.mismatchCount) + assertTrue(rec.details == "[]") + } +} diff --git a/reconciler/src/test/kotlin/com/openremit/reconcile/application/WalletReconcileServiceTest.kt b/reconciler/src/test/kotlin/com/openremit/reconcile/application/WalletReconcileServiceTest.kt new file mode 100644 index 0000000..d9be9c3 --- /dev/null +++ b/reconciler/src/test/kotlin/com/openremit/reconcile/application/WalletReconcileServiceTest.kt @@ -0,0 +1,108 @@ +package com.openremit.reconcile.application + +import java.math.BigDecimal +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +class WalletReconcileServiceTest { + + private val service = WalletReconcileService() + + @Test + fun `consistent wallet produces no mismatch`() { + val result = service.reconcile( + listOf(snapshot(1L, balance = "900", sum = "900", last = "900")) + ) + assertEquals(1, result.totalCount) + assertEquals(0, result.mismatchCount) + } + + @Test + fun `wallet with zero balance and no transactions is consistent`() { + val result = service.reconcile( + listOf(snapshot(1L, balance = "0", sum = "0", last = null)) + ) + assertEquals(0, result.mismatchCount) + } + + @Test + fun `A-only violation — sum differs from balance but last balance_after matches`() { + // ledger의 중간 row 1건이 누락되어 합은 안 맞지만 마지막 balance_after는 우연히 일치 + val result = service.reconcile( + listOf(snapshot(1L, balance = "900", sum = "800", last = "900")) + ) + assertEquals(1, result.mismatchCount) + val m = result.mismatches.single() + assertTrue(m.violatesA, "A must be violated") + assertFalse(m.violatesB, "B must not be violated") + } + + @Test + fun `B-only violation — sum matches but last balance_after differs`() { + // ledger 합은 맞지만 마지막 row의 balance_after 가 잘못 기록됨 (또는 race로 손실) + val result = service.reconcile( + listOf(snapshot(1L, balance = "900", sum = "900", last = "800")) + ) + assertEquals(1, result.mismatchCount) + val m = result.mismatches.single() + assertFalse(m.violatesA) + assertTrue(m.violatesB) + } + + @Test + fun `both A and B violated — typical data corruption`() { + val result = service.reconcile( + listOf(snapshot(1L, balance = "900", sum = "800", last = "800")) + ) + val m = result.mismatches.single() + assertTrue(m.violatesA) + assertTrue(m.violatesB) + } + + @Test + fun `no transactions but non-zero balance — B violated, A also violated`() { + // ledger 가 0건인데 잔액이 있음 → 둘 다 잡힘 + val result = service.reconcile( + listOf(snapshot(1L, balance = "900", sum = "0", last = null)) + ) + val m = result.mismatches.single() + assertTrue(m.violatesA) + assertTrue(m.violatesB) + } + + @Test + fun `mixed wallets — only inconsistent ones reported`() { + val result = service.reconcile( + listOf( + snapshot(1L, balance = "100", sum = "100", last = "100"), // ok + snapshot(2L, balance = "200", sum = "150", last = "200"), // A violated + snapshot(3L, balance = "300", sum = "300", last = "300"), // ok + snapshot(4L, balance = "400", sum = "400", last = "350"), // B violated + ) + ) + assertEquals(4, result.totalCount) + assertEquals(2, result.mismatchCount) + assertEquals(setOf(2L, 4L), result.mismatches.map { it.walletId }.toSet()) + } + + @Test + fun `decimal scale differences are not treated as mismatch`() { + // 1000.0000 vs 1000 — compareTo는 0이어야 한다 + val result = service.reconcile( + listOf( + snapshot(1L, balance = "1000.0000", sum = "1000", last = "1000.00") + ) + ) + assertEquals(0, result.mismatchCount) + } + + private fun snapshot(id: Long, balance: String, sum: String, last: String?) = + WalletSnapshot( + walletId = id, + balance = BigDecimal(balance), + sumOfTransactions = BigDecimal(sum), + lastBalanceAfter = last?.let { BigDecimal(it) }, + ) +} diff --git a/remittance-api/src/main/kotlin/com/openremit/api/application/remittance/RemittanceCreateUseCase.kt b/remittance-api/src/main/kotlin/com/openremit/api/application/remittance/RemittanceCreateUseCase.kt index 93861df..22ebebe 100644 --- a/remittance-api/src/main/kotlin/com/openremit/api/application/remittance/RemittanceCreateUseCase.kt +++ b/remittance-api/src/main/kotlin/com/openremit/api/application/remittance/RemittanceCreateUseCase.kt @@ -2,11 +2,14 @@ package com.openremit.api.application.remittance import com.openremit.api.domain.Remittance import com.openremit.api.domain.RemittanceEvent +import com.openremit.api.domain.WalletTransaction +import com.openremit.api.domain.WalletTransactionRefType import com.openremit.api.infrastructure.fx.FxRateProvider import com.openremit.api.infrastructure.lock.WalletLockService import com.openremit.api.infrastructure.persistence.RemittanceEventRepository import com.openremit.api.infrastructure.persistence.RemittanceRepository import com.openremit.api.infrastructure.persistence.WalletRepository +import com.openremit.api.infrastructure.persistence.WalletTransactionRepository import com.openremit.common.Currency import com.openremit.common.Money import com.openremit.common.ReceiverInfo @@ -47,6 +50,7 @@ class RemittanceCreator( private val walletRepository: WalletRepository, private val remittanceRepository: RemittanceRepository, private val remittanceEventRepository: RemittanceEventRepository, + private val walletTransactionRepository: WalletTransactionRepository, private val paymentGateway: PaymentGatewayClient, private val fxRateProvider: FxRateProvider, private val objectMapper: ObjectMapper, @@ -86,6 +90,17 @@ class RemittanceCreator( val saved = remittanceRepository.save(remittance) + // 정산 ledger INSERT — 같은 트랜잭션. wallet 잔액 변동의 단일 출처. + walletTransactionRepository.save( + WalletTransaction( + walletId = wallet.id, + amount = fromMoney.amount.negate(), + balanceAfter = wallet.balance.amount, + referenceType = WalletTransactionRefType.REMITTANCE, + referenceId = saved.id, + ) + ) + // Outbox INSERT — 같은 트랜잭션. Debezium이 binlog → Kafka(remittance.paid) 발행. val payload = RemittancePaidEvent( remittanceId = saved.id, diff --git a/remittance-api/src/main/kotlin/com/openremit/api/domain/WalletTransaction.kt b/remittance-api/src/main/kotlin/com/openremit/api/domain/WalletTransaction.kt new file mode 100644 index 0000000..b3947ed --- /dev/null +++ b/remittance-api/src/main/kotlin/com/openremit/api/domain/WalletTransaction.kt @@ -0,0 +1,41 @@ +package com.openremit.api.domain + +import jakarta.persistence.Column +import jakarta.persistence.Entity +import jakarta.persistence.EnumType +import jakarta.persistence.Enumerated +import jakarta.persistence.GeneratedValue +import jakarta.persistence.GenerationType +import jakarta.persistence.Id +import jakarta.persistence.Table +import java.math.BigDecimal +import java.time.Instant + +enum class WalletTransactionRefType { REMITTANCE, PAYMENT, REFUND } + +@Entity +@Table(name = "wallet_transactions") +class WalletTransaction( + @Column(name = "wallet_id", nullable = false) + val walletId: Long, + + @Column(name = "amount", nullable = false, precision = 19, scale = 4) + val amount: BigDecimal, + + @Column(name = "balance_after", nullable = false, precision = 19, scale = 4) + val balanceAfter: BigDecimal, + + @Enumerated(EnumType.STRING) + @Column(name = "reference_type", nullable = false, length = 20) + val referenceType: WalletTransactionRefType, + + @Column(name = "reference_id", nullable = false) + val referenceId: Long, + + @Column(name = "created_at", nullable = false) + val createdAt: Instant = Instant.now(), +) { + @Id + @GeneratedValue(strategy = GenerationType.IDENTITY) + var id: Long = 0 +} diff --git a/remittance-api/src/main/kotlin/com/openremit/api/infrastructure/persistence/WalletTransactionRepository.kt b/remittance-api/src/main/kotlin/com/openremit/api/infrastructure/persistence/WalletTransactionRepository.kt new file mode 100644 index 0000000..9e450c0 --- /dev/null +++ b/remittance-api/src/main/kotlin/com/openremit/api/infrastructure/persistence/WalletTransactionRepository.kt @@ -0,0 +1,8 @@ +package com.openremit.api.infrastructure.persistence + +import com.openremit.api.domain.WalletTransaction +import org.springframework.data.jpa.repository.JpaRepository + +interface WalletTransactionRepository : JpaRepository { + fun findByWalletIdOrderByIdAsc(walletId: Long): List +} diff --git a/remittance-api/src/main/resources/db/migration/V5__wallet_transactions.sql b/remittance-api/src/main/resources/db/migration/V5__wallet_transactions.sql new file mode 100644 index 0000000..ccc365f --- /dev/null +++ b/remittance-api/src/main/resources/db/migration/V5__wallet_transactions.sql @@ -0,0 +1,20 @@ +-- wallet_transactions: 지갑 잔액 변동 이력 (감사 + 정산 ledger) +-- amount: 양수=입금, 음수=출금 +-- balance_after: 변동 직후 잔액 스냅샷 (정산 잡 B-검증용) +-- reference_type / reference_id: 변동 원인 (REMITTANCE/PAYMENT/REFUND) +-- 정산 잡 (reconciler): +-- A-검증: wallet.balance == Σ wallet_transactions.amount +-- B-검증: wallet.balance == 마지막 wallet_transactions.balance_after + +CREATE TABLE wallet_transactions ( + id BIGINT NOT NULL AUTO_INCREMENT, + wallet_id BIGINT NOT NULL, + amount DECIMAL(19, 4) NOT NULL, + balance_after DECIMAL(19, 4) NOT NULL, + reference_type VARCHAR(20) NOT NULL, + reference_id BIGINT NOT NULL, + created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + PRIMARY KEY (id), + CONSTRAINT fk_wallet_transactions_wallet FOREIGN KEY (wallet_id) REFERENCES wallets(id), + INDEX idx_wallet_transactions_wallet_created (wallet_id, created_at DESC) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; diff --git a/remittance-api/src/test/kotlin/com/openremit/api/remittance/RemittanceConcurrencyTest.kt b/remittance-api/src/test/kotlin/com/openremit/api/remittance/RemittanceConcurrencyTest.kt index 9f318db..a39b8d1 100644 --- a/remittance-api/src/test/kotlin/com/openremit/api/remittance/RemittanceConcurrencyTest.kt +++ b/remittance-api/src/test/kotlin/com/openremit/api/remittance/RemittanceConcurrencyTest.kt @@ -34,6 +34,7 @@ class RemittanceConcurrencyTest @Autowired constructor( private val walletRepository: WalletRepository, private val remittanceRepository: RemittanceRepository, private val paymentRepository: PaymentRepository, + private val walletTransactionRepository: com.openremit.api.infrastructure.persistence.WalletTransactionRepository, private val fxRateCache: FxRateCache, ) { @@ -53,6 +54,7 @@ class RemittanceConcurrencyTest @Autowired constructor( @AfterTest fun cleanup() { + walletTransactionRepository.deleteAllInBatch() remittanceRepository.deleteAllInBatch() paymentRepository.deleteAllInBatch() walletRepository.deleteAllInBatch() diff --git a/remittance-api/src/test/kotlin/com/openremit/api/remittance/RemittanceCreateIntegrationTest.kt b/remittance-api/src/test/kotlin/com/openremit/api/remittance/RemittanceCreateIntegrationTest.kt index 9cfe24e..e8e850f 100644 --- a/remittance-api/src/test/kotlin/com/openremit/api/remittance/RemittanceCreateIntegrationTest.kt +++ b/remittance-api/src/test/kotlin/com/openremit/api/remittance/RemittanceCreateIntegrationTest.kt @@ -11,6 +11,8 @@ import com.openremit.api.infrastructure.persistence.RemittanceEventRepository import com.openremit.api.infrastructure.persistence.RemittanceRepository import com.openremit.api.infrastructure.persistence.UserRepository import com.openremit.api.infrastructure.persistence.WalletRepository +import com.openremit.api.infrastructure.persistence.WalletTransactionRepository +import com.openremit.api.domain.WalletTransactionRefType import com.openremit.common.events.RemittanceEventTopics import com.openremit.api.infrastructure.security.JwtTokenProvider import com.openremit.common.Currency @@ -46,6 +48,7 @@ class RemittanceCreateIntegrationTest @Autowired constructor( private val walletRepository: WalletRepository, private val remittanceRepository: RemittanceRepository, private val remittanceEventRepository: RemittanceEventRepository, + private val walletTransactionRepository: WalletTransactionRepository, private val paymentRepository: PaymentRepository, private val idempotencyKeyRepository: IdempotencyKeyRepository, private val jwtTokenProvider: JwtTokenProvider, @@ -84,6 +87,7 @@ class RemittanceCreateIntegrationTest @Autowired constructor( fun cleanup() { idempotencyKeyRepository.deleteAllInBatch() remittanceEventRepository.deleteAllInBatch() + walletTransactionRepository.deleteAllInBatch() remittanceRepository.deleteAllInBatch() paymentRepository.deleteAllInBatch() walletRepository.deleteAllInBatch() @@ -112,6 +116,22 @@ class RemittanceCreateIntegrationTest @Autowired constructor( assertEquals(1, paymentRepository.count()) } + @Test + fun `wallet transaction ledger row is written in same transaction as withdraw`() { + val key = UUID.randomUUID().toString() + val response = postRemittance(key, body = standardBody()) + val remittanceId = (readMap(response.contentAsString)["id"] as Number).toLong() + + val wallet = walletRepository.findByUserId(userId.toLong())!! + val txns = walletTransactionRepository.findByWalletIdOrderByIdAsc(wallet.id) + assertEquals(1, txns.size) + val txn = txns.single() + assertEquals(0, txn.amount.compareTo(BigDecimal("-100000"))) + assertEquals(0, txn.balanceAfter.compareTo(BigDecimal("900000"))) + assertEquals(WalletTransactionRefType.REMITTANCE, txn.referenceType) + assertEquals(remittanceId, txn.referenceId) + } + @Test fun `paid remittance writes one outbox event in same transaction`() { val key = UUID.randomUUID().toString()