fix(customer-health-drizzle): join caller transactions

.changeset/healthy-transactions-join.md

@@ -0,0 +1,7 @@
+---
+"@croco/customer-health-drizzle": patch
+"@croco/customer-health-core": patch
+---
+
+Join caller-owned transactions when storing customer health transitions and defer their external
+events until the transaction commits.

.github/workflows/ci.yml

@@ -1,3 +1,3 @@
 name: CI
 
 on:
@@ -662,6 +662,8 @@
               - 'packages/auth-drizzle/**'
               - 'packages/credits-core/**'
               - 'packages/credits-drizzle/**'
+              - 'packages/customer-health-core/**'
+              - 'packages/customer-health-drizzle/**'
               - 'packages/engagement-core/**'
               - 'packages/engagement-drizzle/**'
               - 'packages/entitlements-core/**'
@@ -1495,6 +1497,7 @@
       - name: Verify persistence concurrency against PostgreSQL
         env:
           CREDITS_POSTGRES_URL: postgresql://postgres:postgres@127.0.0.1:5432/croco_membership
+          CUSTOMER_HEALTH_POSTGRES_URL: postgresql://postgres:postgres@127.0.0.1:5432/croco_membership
           ENGAGEMENT_POSTGRES_URL: postgresql://postgres:postgres@127.0.0.1:5432/croco_membership
           ENTITLEMENTS_POSTGRES_URL: postgresql://postgres:postgres@127.0.0.1:5432/croco_membership
           EXECUTION_POSTGRES_URL: postgresql://postgres:postgres@127.0.0.1:5432/croco_membership
@@ -1503,6 +1506,8 @@
         run: |
           pnpm build --filter=@croco/credits-drizzle...
           pnpm --filter @croco/credits-drizzle test:postgres
+          pnpm build --filter=@croco/customer-health-drizzle...
+          pnpm --filter @croco/customer-health-drizzle test:postgres
           pnpm build --filter=@croco/engagement-drizzle...
           pnpm --filter @croco/engagement-drizzle test:postgres
           pnpm build --filter=@croco/entitlements-drizzle...

packages/customer-health-core/README.md

@@ -122,6 +122,9 @@ const score = await service.calculateAndStore("tenant-1", profile);
 시도하고, 계속 커밋되지 않으면 `HealthTransitionPersistenceRetryExhaustedProblem`을 던집니다. 별도 복구
 작업에서는 `publishPendingEvents(tenantId)`를 호출할 수 있습니다.
 
+저장소가 호출자 트랜잭션에 참여하면 `calculateAndStore`는 커밋 전 외부 발행을 건너뛰고 이벤트 의도를
+pending 상태로 남깁니다. 호출자 커밋 후 `publishPendingEvents` 또는 별도 outbox worker로 발행하세요.
+
 ### 추세 분석
 
 ```typescript

packages/customer-health-core/src/libs/CustomerHealthService.ts

@@ -1,3 +1,3 @@
 import { Component, Container, Inject } from "@croco/framework-context";
 import { createHealthTransitionEventIntents } from "./eventIntent";
 import type { HealthTransitionEventIntent } from "./eventIntent";
@@ -44,10 +44,10 @@
     score.tenantId = tenantId;
 
     const previous = await this.store.findLatest(tenantId);
-    await this.persistTransition(score, previous);
+    const eventPublicationDeferred = await this.persistTransition(score, previous);
 
     const eventPublisher = this.getEventPublisher();
-    if (eventPublisher) {
+    if (eventPublisher && !eventPublicationDeferred) {
       const intents = await this.store.listPendingEventIntents(tenantId, 100);
       await this.publishEventIntents(intents, eventPublisher);
     }
@@ -58,14 +58,14 @@
   private async persistTransition(
     score: TenantHealthScore,
     initialPrevious: TenantHealthScore | null,
-  ): Promise<void> {
+  ): Promise<boolean> {
     let previous = initialPrevious;
 
     for (let attempt = 1; attempt <= MAX_TRANSITION_PERSISTENCE_ATTEMPTS; attempt += 1) {
       this.applyPreviousScore(score, previous);
       const eventIntents = createHealthTransitionEventIntents(previous, score);
       const commit = await this.store.saveTransition(score, previous, eventIntents);
-      if (commit.committed) return;
+      if (commit.committed) return commit.eventPublicationDeferred === true;
 
       previous = commit.latest;
       if (attempt < MAX_TRANSITION_PERSISTENCE_ATTEMPTS) {

packages/customer-health-core/src/libs/InMemoryHealthScoreStore.ts

@@ -3,6 +3,7 @@ import { cloneTenantHealthScore } from "./healthScoreSnapshot";
 import { HealthScoreStore } from "./interfaces";
 import { cloneHealthTransitionEventIntent } from "./eventIntent";
 import type { HealthTransitionEventIntent } from "./eventIntent";
+import type { HealthTransitionCommitResult } from "./interfaces";
 import type { TenantHealthScore, TrendPeriod } from "./types";
 import { HealthEventIntentConflictProblem } from "./problems/HealthProblems";
 
@@ -16,10 +17,7 @@ export class InMemoryHealthScoreStore extends HealthScoreStore {
     score: TenantHealthScore,
     previous: TenantHealthScore | null,
     eventIntents: readonly HealthTransitionEventIntent[],
-  ): Promise<
-    | { readonly committed: true }
-    | { readonly committed: false; readonly latest: TenantHealthScore | null }
-  > {
+  ): Promise<HealthTransitionCommitResult> {
     const { tenantId } = score;
     const history = this.store.get(tenantId) ?? [];
     const latest = history.at(-1) ?? null;

packages/customer-health-core/src/libs/interfaces.ts

@@ -9,6 +9,14 @@ import type {
   TrendPeriod,
 } from "./types";
 
+export type HealthTransitionCommitResult =
+  | {
+      readonly committed: true;
+      /** The transition joined a caller-owned transaction and its events must remain pending. */
+      readonly eventPublicationDeferred?: true;
+    }
+  | { readonly committed: false; readonly latest: TenantHealthScore | null };
+
 export abstract class SignalProvider {
   static readonly token = new Token<SignalProvider>("SignalProvider");
   abstract readonly category: SignalCategory;
@@ -17,14 +25,17 @@ export abstract class SignalProvider {
 
 export abstract class HealthScoreStore {
   static readonly token = new Token<HealthScoreStore>("HealthScoreStore");
+  /**
+   * Persists one optimistic transition and assigns its CAS version to `score`.
+   * A version assigned inside a caller-owned transaction is provisional until that transaction
+   * commits. Discard affected snapshots after rollback or an unknown transaction outcome, then
+   * reload the latest committed score before retrying.
+   */
   abstract saveTransition(
     score: TenantHealthScore,
     previous: TenantHealthScore | null,
     eventIntents: readonly HealthTransitionEventIntent[],
-  ): Promise<
-    | { readonly committed: true }
-    | { readonly committed: false; readonly latest: TenantHealthScore | null }
-  >;
+  ): Promise<HealthTransitionCommitResult>;
   abstract listPendingEventIntents(
     tenantId: string,
     limit?: number,

packages/customer-health-core/src/libs/types.ts

@@ -56,6 +56,11 @@ export type HealthScoreProfile = {
 
 export type TenantHealthScore = {
   tenantId: string;
+  /**
+   * Optimistic concurrency token assigned by the store. A token obtained inside a caller-owned
+   * transaction is provisional until commit and must not be reused after rollback or an unknown
+   * transaction outcome.
+   */
   transitionVersion?: string;
   overallScore: number;
   status: HealthStatus;

packages/customer-health-core/src/tests/CustomerHealthService.spec.ts

@@ -11,6 +11,7 @@ import {
   HealthScoreStore,
   HealthSignalRegistry,
 } from "../libs/interfaces";
+import type { HealthTransitionCommitResult } from "../libs/interfaces";
 import type { HealthScoreProfile, HealthSignal, SignalCategory } from "../libs/types";
 
 class MockSignalProvider implements HealthSignalRegistry {
@@ -34,6 +35,15 @@ class MockSignalProvider implements HealthSignalRegistry {
   }
 }
 
+class DeferredPublicationHealthScoreStore extends InMemoryHealthScoreStore {
+  override async saveTransition(
+    ...args: Parameters<InMemoryHealthScoreStore["saveTransition"]>
+  ): Promise<HealthTransitionCommitResult> {
+    const result = await super.saveTransition(...args);
+    return result.committed ? { ...result, eventPublicationDeferred: true } : result;
+  }
+}
+
 describe("CustomerHealthService", () => {
   let service!: CustomerHealthService;
   let store!: InMemoryHealthScoreStore;
@@ -273,6 +283,32 @@ describe("CustomerHealthService", () => {
     });
   });
 
+  it("keeps joined-transaction events pending until publication runs after commit", async () => {
+    const deferredStore = new DeferredPublicationHealthScoreStore();
+    const profile: HealthScoreProfile = {
+      id: "profile-1",
+      name: "Default Profile",
+      weights: { usage: 1, business: 1, engagement: 1 },
+      thresholds: { healthy: 80, atRisk: 60 },
+    };
+    mockRegistry.addProvider("usage", [healthSignal(90, "2026-03-15T10:00:00Z")]);
+    service = new CustomerHealthService(mockRegistry, deferredStore, calculator);
+    await service.calculateAndStore("tenant-1", profile);
+
+    mockRegistry = new MockSignalProvider();
+    mockRegistry.addProvider("usage", [healthSignal(50, "2026-03-15T11:00:00Z")]);
+    service = new CustomerHealthService(mockRegistry, deferredStore, calculator);
+
+    await service.calculateAndStore("tenant-1", profile);
+
+    expect(mockEventPublisher.publishIdempotently).not.toHaveBeenCalled();
+    await expect(deferredStore.listPendingEventIntents("tenant-1")).resolves.toHaveLength(2);
+
+    await expect(service.publishPendingEvents("tenant-1")).resolves.toBe(2);
+    expect(mockEventPublisher.publishIdempotently).toHaveBeenCalledTimes(2);
+    await expect(deferredStore.listPendingEventIntents("tenant-1")).resolves.toHaveLength(0);
+  });
+
   it("should retry the persisted transition without deriving events from the stored score", async () => {
     const profile: HealthScoreProfile = {
       id: "profile-1",

packages/customer-health-drizzle/README.md

@@ -5,7 +5,7 @@
 ## 설치
 
 ```bash
-pnpm add @croco/customer-health-drizzle @croco/customer-health-core drizzle-orm
+pnpm add @croco/customer-health-drizzle @croco/customer-health-core @croco/tx-core @croco/tx-drizzle drizzle-orm
 ```
 
 ## 사용법
@@ -18,9 +18,12 @@ import {
   DrizzleHealthSignalRegistry,
   MeteringSignalProvider,
 } from "@croco/customer-health-drizzle";
+import { TxManager } from "@croco/tx-core";
+import { createDrizzleTxAdapter } from "@croco/tx-drizzle";
 
 await addHealthEventIntents(db);
-const scoreStore = new DrizzleHealthScoreStore(db);
+const txManager = new TxManager(createDrizzleTxAdapter(db));
+const scoreStore = new DrizzleHealthScoreStore(db, txManager);
 const usageProvider = new MeteringSignalProvider(usageStorage);
 const billingProvider = new BillingSignalProvider(subscriptionStorage);
 const registry = new DrizzleHealthSignalRegistry(usageProvider, billingProvider);
@@ -85,6 +88,14 @@ COMMIT;
 - `findHistory(tenantId, limit)`, 최근 점수 이력을 조회합니다.
 - `findHistoryByPeriod(tenantId, period, startDate, endDate)`, 기간별 이력을 조회합니다.
 
+`saveTransition`이 호출자 트랜잭션에 참여하면 결과에 `eventPublicationDeferred: true`가 포함됩니다.
+이때 이벤트 의도는 커밋 전 외부로 발행하지 않으며, 호출자 커밋 후 `publishPendingEvents` 또는 outbox worker가
+발행해야 합니다.
+
+호출자 트랜잭션 안에서 `saveTransition`이 점수에 부여한 `transitionVersion`은 같은 트랜잭션의 후속 CAS에 사용할
+수 있지만, 호출자 트랜잭션이 커밋되기 전까지는 잠정 값입니다. 트랜잭션이나 savepoint가 롤백되었거나 결과를
+확정할 수 없다면 해당 점수 스냅샷을 버리고 `findLatest`로 커밋된 최신 점수를 다시 읽은 뒤 재시도하세요.
+
 ### 신호 제공자
 
 - `BillingSignalProvider`, 구독 상태를 business 신호로 변환합니다.

packages/customer-health-drizzle/package.json

@@ -1,3 +1,3 @@
 {
   "name": "@croco/customer-health-drizzle",
   "version": "0.0.4",
@@ -32,20 +32,24 @@
   "scripts": {
     "build": "tsup src/index.ts --format esm,cjs --minify --clean --dts",
     "lint": "oxlint .",
-    "test": "vitest run",
+    "test": "vitest run --exclude src/tests/DrizzleHealthScoreStorePostgres.spec.ts",
     "test:evidence": "pnpm run test --maxWorkers=1 --reporter=json --outputFile=.turbo/croco-test-evidence.json",
+    "test:postgres": "vitest run src/tests/DrizzleHealthScoreStorePostgres.spec.ts",
     "typecheck": "tsc --noEmit",
     "docs:api:model": "node --experimental-strip-types ../docs/scripts/generate-package-api-model.mts"
   },
   "dependencies": {
     "@croco/customer-health-core": "workspace:*",
     "@croco/framework-context": "workspace:*",
     "@croco/problems-core": "workspace:*",
+    "@croco/tx-core": "workspace:*",
     "@croco/tx-drizzle": "workspace:*"
   },
   "devDependencies": {
     "@croco/testing": "workspace:*",
+    "@types/pg": "8.20.0",
     "drizzle-orm": "catalog:",
+    "pg": "8.22.0",
     "reflect-metadata": "^0.2.2"
   },
   "peerDependencies": {