payg(c10): PaygMeterReportingService + JobChargeService.close() afterCommit hook
Push billable usage to Stripe via Supabase meter-payg-units edge function after the
ledger DEBIT durably commits. Closes the "we tracked it" → "Stripe knows" loop.
PaygMeterReportingService POSTs {team_id, stripe_customer_id, units, idempotency_key,
metadata.category} to a config-driven endpoint. Both payg.meter.endpoint and
payg.meter.service-role-token default empty → no-op in local dev / tests. Never
throws; non-2xx and exceptions bump payg.meter.errors counter and log WARN.
JobChargeService.close() delegates to JobService.close() and registers an
afterCommit synchronization that fires the meter event for paid teams. Skips
when: shadow row is REFUNDED, category is BYPASSED, team has no stripeCustomerId
(free tier — ledger entry suffices), or no synchronization is active (defensive).
Idempotency key: "process:<jobId>:close" — deterministic across pod retries and
the reconciliation backfill so Stripe never bills twice for one close.
Tests:
- PaygMeterReportingServiceTest: happy path body+headers, 5xx, ConnectException,
RuntimeException, blank/null endpoint, zero units, blank service-role-token.
- JobChargeServiceTest extended: subscribed posts; free-tier skips; refunded skips;
bypassed skips; no shadow row skips; no team-ext row skips; meter exception is
swallowed inside afterCommit; close() without an active transaction skips the
meter post but still closes the job.
Free-tier detection uses stripeCustomerId == null as the established pre-#6532
stand-in (see PaygWalletController for the same pattern). Tightens to
paygSubscriptionId != null when PR #6532's payg_team_extensions column lands.
This commit is contained in:
@@ -11,6 +11,8 @@ import java.util.UUID;
|
||||
import org.springframework.context.annotation.Profile;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import org.springframework.transaction.support.TransactionSynchronization;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
import org.springframework.web.multipart.MultipartFile;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
@@ -21,12 +23,16 @@ import stirling.software.saas.payg.job.JobContext;
|
||||
import stirling.software.saas.payg.job.JobService;
|
||||
import stirling.software.saas.payg.job.JoinOrOpenResult;
|
||||
import stirling.software.saas.payg.job.ProcessingJob;
|
||||
import stirling.software.saas.payg.meter.PaygMeterReportingService;
|
||||
import stirling.software.saas.payg.model.BillingCategory;
|
||||
import stirling.software.saas.payg.model.JobSource;
|
||||
import stirling.software.saas.payg.model.JobStatus;
|
||||
import stirling.software.saas.payg.model.ShadowChargeStatus;
|
||||
import stirling.software.saas.payg.policy.PaygTeamExtensions;
|
||||
import stirling.software.saas.payg.policy.PricingPolicy;
|
||||
import stirling.software.saas.payg.policy.PricingPolicyService;
|
||||
import stirling.software.saas.payg.repository.PaygShadowChargeRepository;
|
||||
import stirling.software.saas.payg.repository.PaygTeamExtensionsRepository;
|
||||
import stirling.software.saas.payg.repository.ProcessingJobRepository;
|
||||
import stirling.software.saas.payg.shadow.PaygShadowCharge;
|
||||
|
||||
@@ -55,18 +61,26 @@ public class JobChargeService {
|
||||
private final DocumentClassifier classifier;
|
||||
private final PaygShadowChargeRepository shadowRepository;
|
||||
private final ProcessingJobRepository jobRepository;
|
||||
private final PaygTeamExtensionsRepository teamExtensionsRepository;
|
||||
private final PaygMeterReportingService meterReportingService;
|
||||
|
||||
public JobChargeService(
|
||||
JobService jobService,
|
||||
PricingPolicyService policyService,
|
||||
DocumentClassifier classifier,
|
||||
PaygShadowChargeRepository shadowRepository,
|
||||
ProcessingJobRepository jobRepository) {
|
||||
ProcessingJobRepository jobRepository,
|
||||
PaygTeamExtensionsRepository teamExtensionsRepository,
|
||||
PaygMeterReportingService meterReportingService) {
|
||||
this.jobService = Objects.requireNonNull(jobService, "jobService");
|
||||
this.policyService = Objects.requireNonNull(policyService, "policyService");
|
||||
this.classifier = Objects.requireNonNull(classifier, "classifier");
|
||||
this.shadowRepository = Objects.requireNonNull(shadowRepository, "shadowRepository");
|
||||
this.jobRepository = Objects.requireNonNull(jobRepository, "jobRepository");
|
||||
this.teamExtensionsRepository =
|
||||
Objects.requireNonNull(teamExtensionsRepository, "teamExtensionsRepository");
|
||||
this.meterReportingService =
|
||||
Objects.requireNonNull(meterReportingService, "meterReportingService");
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -201,6 +215,105 @@ public class JobChargeService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Closes a process and — for paid teams — pushes a Stripe meter event for the units captured on
|
||||
* the originating shadow row. Idempotent w.r.t. process state (delegates to {@link
|
||||
* JobService#close(UUID)}, which silently no-ops on an already-closed row).
|
||||
*
|
||||
* <p>The meter POST runs in an {@code afterCommit} hook so we only tell Stripe about work that
|
||||
* actually committed to our ledger (the customer's bill is authoritative — Stripe is the
|
||||
* downstream invoice). A failed POST does not roll back the close; the reconciliation backfill
|
||||
* (separate chunk) is the durability mechanism.
|
||||
*
|
||||
* <p>Skipped paths — no meter event fired:
|
||||
*
|
||||
* <ul>
|
||||
* <li>{@code BillingCategory.BYPASSED} on the shadow row (manual UI tool — defensive; the
|
||||
* interceptor never opens a process for these).
|
||||
* <li>Refunded shadow row (the customer was credited; nothing to bill).
|
||||
* <li>Team has no {@link PaygTeamExtensions#getStripeCustomerId() stripe_customer_id} (free
|
||||
* tier; ledger entry suffices).
|
||||
* <li>No shadow row at all (job opened outside the shadow path, e.g. legacy import).
|
||||
* </ul>
|
||||
*/
|
||||
@Transactional
|
||||
public ProcessingJob close(UUID jobId) {
|
||||
Objects.requireNonNull(jobId, "jobId");
|
||||
ProcessingJob closed = jobService.close(jobId);
|
||||
|
||||
// The afterCommit hook only fires if there's an active transaction (Spring's
|
||||
// @Transactional ensures that). If we're called outside one — e.g. a test using the raw
|
||||
// bean — fall through with a debug log: the close() above already happened in a
|
||||
// sub-transaction created by JobService, but the surrounding scope has no synchronization.
|
||||
if (!TransactionSynchronizationManager.isSynchronizationActive()) {
|
||||
log.debug("close({}): no active synchronization; skipping meter POST", jobId);
|
||||
return closed;
|
||||
}
|
||||
|
||||
TransactionSynchronizationManager.registerSynchronization(
|
||||
new TransactionSynchronization() {
|
||||
@Override
|
||||
public void afterCommit() {
|
||||
try {
|
||||
postMeterEventForClose(jobId);
|
||||
} catch (RuntimeException e) {
|
||||
// PaygMeterReportingService should already swallow; defence in depth so
|
||||
// a thrown exception out of afterCommit doesn't leak past the
|
||||
// synchronization boundary and bubble into the caller.
|
||||
log.warn(
|
||||
"afterCommit meter post for job {} threw unexpectedly: {}",
|
||||
jobId,
|
||||
e.getMessage());
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
return closed;
|
||||
}
|
||||
|
||||
private void postMeterEventForClose(UUID jobId) {
|
||||
Optional<PaygShadowCharge> rowOpt = shadowRepository.findFirstByJobIdOrderByIdAsc(jobId);
|
||||
if (rowOpt.isEmpty()) {
|
||||
// No shadow row → not a PAYG-tracked job; nothing to meter.
|
||||
return;
|
||||
}
|
||||
PaygShadowCharge row = rowOpt.get();
|
||||
if (row.getStatus() == ShadowChargeStatus.REFUNDED) {
|
||||
// Refunded rows are zero-net charges; do not emit a meter event.
|
||||
return;
|
||||
}
|
||||
BillingCategory category = row.getBillingCategory();
|
||||
if (category == null || category == BillingCategory.BYPASSED) {
|
||||
// Defensive: BYPASSED rows shouldn't exist (interceptor short-circuits before
|
||||
// openProcess), but tolerate if a future caller writes one.
|
||||
log.debug("close({}): shadow row category={} → no meter event", jobId, category);
|
||||
return;
|
||||
}
|
||||
Integer units = row.getPaygUnits();
|
||||
if (units == null || units <= 0) {
|
||||
return;
|
||||
}
|
||||
Long teamId = row.getTeamId();
|
||||
if (teamId == null) {
|
||||
return;
|
||||
}
|
||||
Optional<PaygTeamExtensions> ext = teamExtensionsRepository.findById(teamId);
|
||||
String stripeCustomerId = ext.map(PaygTeamExtensions::getStripeCustomerId).orElse(null);
|
||||
if (stripeCustomerId == null || stripeCustomerId.isBlank()) {
|
||||
// Free-tier team (no Stripe identity) — ledger entry is enough. When PR #6532 lands
|
||||
// this check tightens to ext.getPaygSubscriptionId() != null, but on this branch the
|
||||
// presence of stripe_customer_id is the established stand-in for "is subscribed."
|
||||
log.debug(
|
||||
"close({}): team {} has no stripeCustomerId → free-tier, no meter event",
|
||||
jobId,
|
||||
teamId);
|
||||
return;
|
||||
}
|
||||
String idempotencyKey = "process:" + jobId + ":close";
|
||||
meterReportingService.recordUsage(
|
||||
teamId, stripeCustomerId, units, category, idempotencyKey);
|
||||
}
|
||||
|
||||
/**
|
||||
* Mid-chain 5xx on a JOINED step: return the step slot. The {@code lastStepAt} timestamp stays
|
||||
* advanced (workflow window intentionally remains active for the next retry). No shadow-row
|
||||
|
||||
+142
@@ -0,0 +1,142 @@
|
||||
package stirling.software.saas.payg.meter;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.context.annotation.Profile;
|
||||
import org.springframework.http.HttpEntity;
|
||||
import org.springframework.http.HttpHeaders;
|
||||
import org.springframework.http.HttpMethod;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.http.ResponseEntity;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.web.client.RestTemplate;
|
||||
|
||||
import io.micrometer.core.instrument.Counter;
|
||||
import io.micrometer.core.instrument.MeterRegistry;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import stirling.software.saas.payg.model.BillingCategory;
|
||||
|
||||
/**
|
||||
* POSTs PAYG billable usage to the Supabase {@code meter-payg-units} edge function. Called from
|
||||
* {@code JobChargeService.close()} in an {@code afterCommit} hook, so the wallet ledger DEBIT (the
|
||||
* customer's authoritative bill) is already durable before we tell Stripe about it.
|
||||
*
|
||||
* <p>Stripe is on a single flat-priced meter forever. {@link BillingCategory} ships as metadata for
|
||||
* analytics — pricing never reads it. Free-tier teams (no Stripe subscription) skip this call
|
||||
* entirely; the ledger entry is the only record needed.
|
||||
*
|
||||
* <p>Failure mode: we owe Stripe an event but the customer's bill via the ledger is correct. Log
|
||||
* WARN, bump {@code payg.meter.errors}, and swallow — the meter event log table on Supabase plus
|
||||
* the reconciliation job (separate chunk) is the durability story, not retries here. Caller's
|
||||
* {@code close()} must not roll back because Stripe wobbled.
|
||||
*
|
||||
* <p>Both config keys default to empty so unit tests / local dev never crash on missing env. When
|
||||
* blank, this service no-ops at WARN-debug level — useful for SaaS smoke tests that don't want to
|
||||
* touch the real edge function.
|
||||
*/
|
||||
@Service
|
||||
@Profile("saas")
|
||||
@Slf4j
|
||||
public class PaygMeterReportingService {
|
||||
|
||||
private final String endpoint;
|
||||
private final String serviceRoleToken;
|
||||
private final RestTemplate restTemplate;
|
||||
private final Counter errorsCounter;
|
||||
|
||||
public PaygMeterReportingService(
|
||||
@Value("${payg.meter.endpoint:}") String endpoint,
|
||||
@Value("${payg.meter.service-role-token:}") String serviceRoleToken,
|
||||
RestTemplate saasRestTemplate,
|
||||
MeterRegistry meterRegistry) {
|
||||
this.endpoint = endpoint;
|
||||
this.serviceRoleToken = serviceRoleToken;
|
||||
this.restTemplate = saasRestTemplate;
|
||||
this.errorsCounter =
|
||||
Counter.builder("payg.meter.errors")
|
||||
.description("Failures POSTing PAYG meter events to Supabase edge function")
|
||||
.register(meterRegistry);
|
||||
}
|
||||
|
||||
/**
|
||||
* Best-effort POST of a single billable event. Idempotency on the Supabase side is keyed on
|
||||
* {@code idempotency_key} — supply a deterministic value (e.g. {@code "process:<uuid>:close"})
|
||||
* so a retry, a reconciliation replay, or a double-fire from two pods never charges twice.
|
||||
*
|
||||
* <p>Never throws. The wallet ledger entry is the source of truth for what the customer is
|
||||
* billed; if this method fails the only loss is that Stripe doesn't see this event until the
|
||||
* reconciliation backfill runs.
|
||||
*/
|
||||
public void recordUsage(
|
||||
Long teamId,
|
||||
String stripeCustomerId,
|
||||
int units,
|
||||
BillingCategory category,
|
||||
String idempotencyKey) {
|
||||
if (endpoint == null || endpoint.isBlank()) {
|
||||
log.debug(
|
||||
"payg.meter.endpoint not configured; skipping meter event for team {} key {}",
|
||||
teamId,
|
||||
idempotencyKey);
|
||||
return;
|
||||
}
|
||||
if (units <= 0) {
|
||||
// Zero-unit events would inflate event count without changing the bill — defensive.
|
||||
log.debug(
|
||||
"Skipping meter event with units={} for team {} key {}",
|
||||
units,
|
||||
teamId,
|
||||
idempotencyKey);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
HttpHeaders headers = new HttpHeaders();
|
||||
if (serviceRoleToken != null && !serviceRoleToken.isBlank()) {
|
||||
headers.setBearerAuth(serviceRoleToken);
|
||||
}
|
||||
headers.setContentType(MediaType.APPLICATION_JSON);
|
||||
|
||||
Map<String, Object> body =
|
||||
Map.of(
|
||||
"team_id",
|
||||
teamId == null ? "" : teamId.toString(),
|
||||
"stripe_customer_id",
|
||||
stripeCustomerId == null ? "" : stripeCustomerId,
|
||||
"units",
|
||||
units,
|
||||
"idempotency_key",
|
||||
idempotencyKey,
|
||||
"metadata",
|
||||
Map.of("category", category == null ? "UNKNOWN" : category.name()));
|
||||
|
||||
ResponseEntity<String> response =
|
||||
restTemplate.exchange(
|
||||
endpoint,
|
||||
HttpMethod.POST,
|
||||
new HttpEntity<>(body, headers),
|
||||
String.class);
|
||||
|
||||
if (!response.getStatusCode().is2xxSuccessful()) {
|
||||
log.warn(
|
||||
"Meter event POST returned {} for team {} key {}: {}",
|
||||
response.getStatusCode(),
|
||||
teamId,
|
||||
idempotencyKey,
|
||||
response.getBody());
|
||||
errorsCounter.increment();
|
||||
}
|
||||
} catch (Exception e) {
|
||||
// Catch-all by design: this method MUST NOT propagate. The customer's bill via the
|
||||
// ledger is correct; we just owe Stripe an event we'll backfill via reconciliation.
|
||||
log.warn(
|
||||
"Meter event POST failed for team {} key {}: {}",
|
||||
teamId,
|
||||
idempotencyKey,
|
||||
e.getMessage());
|
||||
errorsCounter.increment();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -35,6 +35,15 @@ app.supabase.clock-skew-seconds=${app.jwt.clock-skew-seconds:120}
|
||||
app.supabase.edge-function-url=https://${app.supabase.project-ref}.supabase.co/functions/v1
|
||||
app.supabase.edge-function-secret=${SUPABASE_EDGE_FUNCTION_SECRET:}
|
||||
|
||||
# ---------- PAYG meter reporting ----------
|
||||
# Posts billable usage to the Supabase `meter-payg-units` edge function in the JobChargeService
|
||||
# close() afterCommit hook. Defaults to empty so unit tests / local dev are no-ops; set
|
||||
# PAYG_METER_ENDPOINT + SUPABASE_SERVICE_ROLE_TOKEN in deployed envs to enable. Compose with the
|
||||
# Supabase functions base via env (e.g. PAYG_METER_ENDPOINT=$SUPABASE_FUNCTIONS_URL/meter-payg-units)
|
||||
# rather than templating here — concatenating an empty default would produce a half-valid URL.
|
||||
payg.meter.endpoint=${PAYG_METER_ENDPOINT:}
|
||||
payg.meter.service-role-token=${SUPABASE_SERVICE_ROLE_TOKEN:}
|
||||
|
||||
supabase.url=https://${app.supabase.project-ref}.supabase.co
|
||||
|
||||
spring.security.oauth2.resourceserver.jwt.jwk-set-uri=https://${app.supabase.project-ref}.supabase.co/auth/v1/.well-known/jwks.json
|
||||
|
||||
+221
-1
@@ -17,14 +17,18 @@ import java.time.LocalDateTime;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.UUID;
|
||||
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.io.TempDir;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.Mockito;
|
||||
import org.springframework.mock.web.MockMultipartFile;
|
||||
import org.springframework.transaction.support.TransactionSynchronization;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
import org.springframework.web.multipart.MultipartFile;
|
||||
|
||||
import stirling.software.saas.payg.docs.DocumentClassifier;
|
||||
@@ -33,14 +37,17 @@ import stirling.software.saas.payg.job.JobContext;
|
||||
import stirling.software.saas.payg.job.JobService;
|
||||
import stirling.software.saas.payg.job.JoinOrOpenResult;
|
||||
import stirling.software.saas.payg.job.ProcessingJob;
|
||||
import stirling.software.saas.payg.meter.PaygMeterReportingService;
|
||||
import stirling.software.saas.payg.model.BillingCategory;
|
||||
import stirling.software.saas.payg.model.JobSource;
|
||||
import stirling.software.saas.payg.model.JobStatus;
|
||||
import stirling.software.saas.payg.model.ProcessType;
|
||||
import stirling.software.saas.payg.model.ShadowChargeStatus;
|
||||
import stirling.software.saas.payg.policy.PaygTeamExtensions;
|
||||
import stirling.software.saas.payg.policy.PricingPolicy;
|
||||
import stirling.software.saas.payg.policy.PricingPolicyService;
|
||||
import stirling.software.saas.payg.repository.PaygShadowChargeRepository;
|
||||
import stirling.software.saas.payg.repository.PaygTeamExtensionsRepository;
|
||||
import stirling.software.saas.payg.repository.ProcessingJobRepository;
|
||||
import stirling.software.saas.payg.shadow.PaygShadowCharge;
|
||||
|
||||
@@ -55,6 +62,8 @@ class JobChargeServiceTest {
|
||||
private DocumentClassifier classifier;
|
||||
private PaygShadowChargeRepository shadowRepo;
|
||||
private ProcessingJobRepository jobRepo;
|
||||
private PaygTeamExtensionsRepository teamExtRepo;
|
||||
private PaygMeterReportingService meterReporter;
|
||||
private JobChargeService service;
|
||||
|
||||
@BeforeEach
|
||||
@@ -64,7 +73,26 @@ class JobChargeServiceTest {
|
||||
classifier = Mockito.mock(DocumentClassifier.class);
|
||||
shadowRepo = Mockito.mock(PaygShadowChargeRepository.class);
|
||||
jobRepo = Mockito.mock(ProcessingJobRepository.class);
|
||||
service = new JobChargeService(jobService, policyService, classifier, shadowRepo, jobRepo);
|
||||
teamExtRepo = Mockito.mock(PaygTeamExtensionsRepository.class);
|
||||
meterReporter = Mockito.mock(PaygMeterReportingService.class);
|
||||
service =
|
||||
new JobChargeService(
|
||||
jobService,
|
||||
policyService,
|
||||
classifier,
|
||||
shadowRepo,
|
||||
jobRepo,
|
||||
teamExtRepo,
|
||||
meterReporter);
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void tearDown() {
|
||||
// Defensive: a previous test could have left a fake synchronization registered. Clearing
|
||||
// ensures isolation when tests run in any order.
|
||||
if (TransactionSynchronizationManager.isSynchronizationActive()) {
|
||||
TransactionSynchronizationManager.clear();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -396,6 +424,198 @@ class JobChargeServiceTest {
|
||||
verify(jobRepo, never()).save(any());
|
||||
}
|
||||
|
||||
// --- close() — meter reporting in afterCommit -----------------------------------------------
|
||||
|
||||
@Test
|
||||
void close_subscribedTeam_postsMeterEventAfterCommit() {
|
||||
UUID jobId = UUID.randomUUID();
|
||||
ProcessingJob job = openJob(jobId);
|
||||
when(jobService.close(jobId)).thenReturn(job);
|
||||
|
||||
PaygShadowCharge row = chargedShadowRow(jobId, 100L, 4, BillingCategory.API);
|
||||
when(shadowRepo.findFirstByJobIdOrderByIdAsc(jobId)).thenReturn(Optional.of(row));
|
||||
|
||||
PaygTeamExtensions ext = new PaygTeamExtensions();
|
||||
ext.setTeamId(100L);
|
||||
ext.setStripeCustomerId("cus_subscribed");
|
||||
when(teamExtRepo.findById(100L)).thenReturn(Optional.of(ext));
|
||||
|
||||
withTransactionSynchronization(
|
||||
() -> {
|
||||
service.close(jobId);
|
||||
Mockito.verifyNoInteractions(meterReporter);
|
||||
});
|
||||
|
||||
// afterCommit ran on tearDown of withTransactionSynchronization → meter posted now.
|
||||
verify(meterReporter)
|
||||
.recordUsage(
|
||||
100L,
|
||||
"cus_subscribed",
|
||||
4,
|
||||
BillingCategory.API,
|
||||
"process:" + jobId + ":close");
|
||||
}
|
||||
|
||||
@Test
|
||||
void close_freeTierTeam_doesNotPostMeterEvent() {
|
||||
UUID jobId = UUID.randomUUID();
|
||||
ProcessingJob job = openJob(jobId);
|
||||
when(jobService.close(jobId)).thenReturn(job);
|
||||
|
||||
PaygShadowCharge row = chargedShadowRow(jobId, 100L, 4, BillingCategory.API);
|
||||
when(shadowRepo.findFirstByJobIdOrderByIdAsc(jobId)).thenReturn(Optional.of(row));
|
||||
|
||||
// No stripe_customer_id → treated as free-tier on this branch (pre-#6532).
|
||||
PaygTeamExtensions ext = new PaygTeamExtensions();
|
||||
ext.setTeamId(100L);
|
||||
ext.setStripeCustomerId(null);
|
||||
when(teamExtRepo.findById(100L)).thenReturn(Optional.of(ext));
|
||||
|
||||
withTransactionSynchronization(() -> service.close(jobId));
|
||||
|
||||
Mockito.verifyNoInteractions(meterReporter);
|
||||
}
|
||||
|
||||
@Test
|
||||
void close_noTeamExtensionsRow_doesNotPostMeterEvent() {
|
||||
UUID jobId = UUID.randomUUID();
|
||||
ProcessingJob job = openJob(jobId);
|
||||
when(jobService.close(jobId)).thenReturn(job);
|
||||
|
||||
PaygShadowCharge row = chargedShadowRow(jobId, 100L, 4, BillingCategory.API);
|
||||
when(shadowRepo.findFirstByJobIdOrderByIdAsc(jobId)).thenReturn(Optional.of(row));
|
||||
when(teamExtRepo.findById(100L)).thenReturn(Optional.empty());
|
||||
|
||||
withTransactionSynchronization(() -> service.close(jobId));
|
||||
|
||||
Mockito.verifyNoInteractions(meterReporter);
|
||||
}
|
||||
|
||||
@Test
|
||||
void close_refundedShadowRow_doesNotPostMeterEvent() {
|
||||
UUID jobId = UUID.randomUUID();
|
||||
ProcessingJob job = openJob(jobId);
|
||||
when(jobService.close(jobId)).thenReturn(job);
|
||||
|
||||
PaygShadowCharge row = chargedShadowRow(jobId, 100L, 4, BillingCategory.API);
|
||||
row.setStatus(ShadowChargeStatus.REFUNDED);
|
||||
when(shadowRepo.findFirstByJobIdOrderByIdAsc(jobId)).thenReturn(Optional.of(row));
|
||||
|
||||
withTransactionSynchronization(() -> service.close(jobId));
|
||||
|
||||
Mockito.verifyNoInteractions(meterReporter);
|
||||
Mockito.verifyNoInteractions(teamExtRepo);
|
||||
}
|
||||
|
||||
@Test
|
||||
void close_noShadowRow_doesNotPostMeterEvent() {
|
||||
UUID jobId = UUID.randomUUID();
|
||||
ProcessingJob job = openJob(jobId);
|
||||
when(jobService.close(jobId)).thenReturn(job);
|
||||
|
||||
when(shadowRepo.findFirstByJobIdOrderByIdAsc(jobId)).thenReturn(Optional.empty());
|
||||
|
||||
withTransactionSynchronization(() -> service.close(jobId));
|
||||
|
||||
Mockito.verifyNoInteractions(meterReporter);
|
||||
Mockito.verifyNoInteractions(teamExtRepo);
|
||||
}
|
||||
|
||||
@Test
|
||||
void close_bypassedCategoryOnShadowRow_doesNotPostMeterEvent() {
|
||||
// Defensive: BYPASSED rows shouldn't normally exist (the interceptor short-circuits
|
||||
// before openProcess), but if one slips through we must not meter it.
|
||||
UUID jobId = UUID.randomUUID();
|
||||
ProcessingJob job = openJob(jobId);
|
||||
when(jobService.close(jobId)).thenReturn(job);
|
||||
|
||||
PaygShadowCharge row = chargedShadowRow(jobId, 100L, 4, BillingCategory.BYPASSED);
|
||||
when(shadowRepo.findFirstByJobIdOrderByIdAsc(jobId)).thenReturn(Optional.of(row));
|
||||
|
||||
withTransactionSynchronization(() -> service.close(jobId));
|
||||
|
||||
Mockito.verifyNoInteractions(meterReporter);
|
||||
Mockito.verifyNoInteractions(teamExtRepo);
|
||||
}
|
||||
|
||||
@Test
|
||||
void close_meterReporterThrowsRuntimeException_doesNotPropagate() {
|
||||
// PaygMeterReportingService is documented to swallow; defence-in-depth in
|
||||
// JobChargeService catches a misbehaving impl so the afterCommit hook can't poison the
|
||||
// close flow.
|
||||
UUID jobId = UUID.randomUUID();
|
||||
ProcessingJob job = openJob(jobId);
|
||||
when(jobService.close(jobId)).thenReturn(job);
|
||||
|
||||
PaygShadowCharge row = chargedShadowRow(jobId, 100L, 4, BillingCategory.AUTOMATION);
|
||||
when(shadowRepo.findFirstByJobIdOrderByIdAsc(jobId)).thenReturn(Optional.of(row));
|
||||
|
||||
PaygTeamExtensions ext = new PaygTeamExtensions();
|
||||
ext.setTeamId(100L);
|
||||
ext.setStripeCustomerId("cus_subscribed");
|
||||
when(teamExtRepo.findById(100L)).thenReturn(Optional.of(ext));
|
||||
|
||||
Mockito.doThrow(new RuntimeException("simulated meter failure"))
|
||||
.when(meterReporter)
|
||||
.recordUsage(
|
||||
Mockito.anyLong(),
|
||||
Mockito.anyString(),
|
||||
Mockito.anyInt(),
|
||||
Mockito.any(BillingCategory.class),
|
||||
Mockito.anyString());
|
||||
|
||||
// Should not throw — afterCommit's defence-in-depth wraps the call.
|
||||
withTransactionSynchronization(() -> service.close(jobId));
|
||||
verify(meterReporter)
|
||||
.recordUsage(
|
||||
100L,
|
||||
"cus_subscribed",
|
||||
4,
|
||||
BillingCategory.AUTOMATION,
|
||||
"process:" + jobId + ":close");
|
||||
}
|
||||
|
||||
@Test
|
||||
void close_noActiveTransactionSync_skipsMeterPostButStillClosesJob() {
|
||||
// Direct call without an outer @Transactional → no sync to register against. close()
|
||||
// must still close the job; the meter post is implicitly deferred to whatever async path
|
||||
// eventually wraps the call (or is never made, which is fine for ledger-only flows).
|
||||
UUID jobId = UUID.randomUUID();
|
||||
ProcessingJob job = openJob(jobId);
|
||||
when(jobService.close(jobId)).thenReturn(job);
|
||||
|
||||
assertThat(TransactionSynchronizationManager.isSynchronizationActive()).isFalse();
|
||||
service.close(jobId);
|
||||
|
||||
Mockito.verifyNoInteractions(meterReporter);
|
||||
verify(jobService).close(jobId);
|
||||
}
|
||||
|
||||
private static void withTransactionSynchronization(Runnable body) {
|
||||
TransactionSynchronizationManager.initSynchronization();
|
||||
try {
|
||||
body.run();
|
||||
// Drain registered synchronizations to simulate a successful commit.
|
||||
for (TransactionSynchronization sync :
|
||||
TransactionSynchronizationManager.getSynchronizations()) {
|
||||
sync.afterCommit();
|
||||
}
|
||||
} finally {
|
||||
TransactionSynchronizationManager.clear();
|
||||
}
|
||||
}
|
||||
|
||||
private static PaygShadowCharge chargedShadowRow(
|
||||
UUID jobId, Long teamId, int units, BillingCategory category) {
|
||||
PaygShadowCharge row = new PaygShadowCharge();
|
||||
row.setJobId(jobId);
|
||||
row.setTeamId(teamId);
|
||||
row.setPaygUnits(units);
|
||||
row.setStatus(ShadowChargeStatus.CHARGED);
|
||||
row.setBillingCategory(category);
|
||||
return row;
|
||||
}
|
||||
|
||||
// --- helpers --------------------------------------------------------------------------------
|
||||
|
||||
private static PricingPolicy stubPolicy(int minCharge, Map<JobSource, Integer> stepLimits) {
|
||||
|
||||
+196
@@ -0,0 +1,196 @@
|
||||
package stirling.software.saas.payg.meter;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatCode;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.net.ConnectException;
|
||||
import java.util.Map;
|
||||
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.Mockito;
|
||||
import org.springframework.http.HttpEntity;
|
||||
import org.springframework.http.HttpHeaders;
|
||||
import org.springframework.http.HttpMethod;
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.http.ResponseEntity;
|
||||
import org.springframework.web.client.ResourceAccessException;
|
||||
import org.springframework.web.client.RestTemplate;
|
||||
|
||||
import io.micrometer.core.instrument.Counter;
|
||||
import io.micrometer.core.instrument.MeterRegistry;
|
||||
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
|
||||
|
||||
import stirling.software.saas.payg.model.BillingCategory;
|
||||
|
||||
/**
|
||||
* Covers the contract documented on {@link PaygMeterReportingService#recordUsage}: never throws,
|
||||
* skips when endpoint is blank, counts non-2xx and exceptions on {@code payg.meter.errors}.
|
||||
*/
|
||||
class PaygMeterReportingServiceTest {
|
||||
|
||||
private static final String ENDPOINT =
|
||||
"https://example.supabase.co/functions/v1/meter-payg-units";
|
||||
private static final String TOKEN = "test-service-role-token";
|
||||
|
||||
private RestTemplate restTemplate;
|
||||
private MeterRegistry meterRegistry;
|
||||
private Counter errorsCounter;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
restTemplate = Mockito.mock(RestTemplate.class);
|
||||
meterRegistry = new SimpleMeterRegistry();
|
||||
errorsCounter = meterRegistry.counter("payg.meter.errors");
|
||||
}
|
||||
|
||||
private PaygMeterReportingService newService(String endpoint, String token) {
|
||||
return new PaygMeterReportingService(endpoint, token, restTemplate, meterRegistry);
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordUsage_happyPath_postsBodyAndDoesNotIncrementErrorCounter() {
|
||||
when(restTemplate.exchange(eq(ENDPOINT), eq(HttpMethod.POST), any(), eq(String.class)))
|
||||
.thenReturn(new ResponseEntity<>("{\"ok\":true}", HttpStatus.OK));
|
||||
|
||||
PaygMeterReportingService service = newService(ENDPOINT, TOKEN);
|
||||
service.recordUsage(100L, "cus_abc", 5, BillingCategory.API, "process:job1:close");
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
ArgumentCaptor<HttpEntity<Map<String, Object>>> entityCaptor =
|
||||
ArgumentCaptor.forClass(HttpEntity.class);
|
||||
verify(restTemplate)
|
||||
.exchange(
|
||||
eq(ENDPOINT),
|
||||
eq(HttpMethod.POST),
|
||||
entityCaptor.capture(),
|
||||
eq(String.class));
|
||||
|
||||
HttpEntity<Map<String, Object>> sent = entityCaptor.getValue();
|
||||
Map<String, Object> body = sent.getBody();
|
||||
assertThat(body).isNotNull();
|
||||
assertThat(body.get("team_id")).isEqualTo("100");
|
||||
assertThat(body.get("stripe_customer_id")).isEqualTo("cus_abc");
|
||||
assertThat(body.get("units")).isEqualTo(5);
|
||||
assertThat(body.get("idempotency_key")).isEqualTo("process:job1:close");
|
||||
assertThat(body.get("metadata")).isEqualTo(Map.of("category", "API"));
|
||||
|
||||
HttpHeaders headers = sent.getHeaders();
|
||||
assertThat(headers.getFirst("Authorization")).isEqualTo("Bearer " + TOKEN);
|
||||
assertThat(headers.getContentType()).isNotNull();
|
||||
assertThat(headers.getContentType().toString()).startsWith("application/json");
|
||||
|
||||
assertThat(errorsCounter.count()).isZero();
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordUsage_5xxResponse_incrementsErrorCounterAndDoesNotThrow() {
|
||||
when(restTemplate.exchange(eq(ENDPOINT), eq(HttpMethod.POST), any(), eq(String.class)))
|
||||
.thenReturn(new ResponseEntity<>("oops", HttpStatus.INTERNAL_SERVER_ERROR));
|
||||
|
||||
PaygMeterReportingService service = newService(ENDPOINT, TOKEN);
|
||||
assertThatCode(
|
||||
() ->
|
||||
service.recordUsage(
|
||||
100L,
|
||||
"cus_abc",
|
||||
3,
|
||||
BillingCategory.AUTOMATION,
|
||||
"process:job2:close"))
|
||||
.doesNotThrowAnyException();
|
||||
|
||||
assertThat(errorsCounter.count()).isEqualTo(1.0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordUsage_connectionRefused_incrementsErrorCounterAndDoesNotThrow() {
|
||||
when(restTemplate.exchange(eq(ENDPOINT), eq(HttpMethod.POST), any(), eq(String.class)))
|
||||
.thenThrow(new ResourceAccessException("connect refused", new ConnectException()));
|
||||
|
||||
PaygMeterReportingService service = newService(ENDPOINT, TOKEN);
|
||||
assertThatCode(
|
||||
() ->
|
||||
service.recordUsage(
|
||||
100L,
|
||||
"cus_abc",
|
||||
7,
|
||||
BillingCategory.AI,
|
||||
"process:job3:close"))
|
||||
.doesNotThrowAnyException();
|
||||
|
||||
assertThat(errorsCounter.count()).isEqualTo(1.0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordUsage_runtimeException_swallowed() {
|
||||
when(restTemplate.exchange(eq(ENDPOINT), eq(HttpMethod.POST), any(), eq(String.class)))
|
||||
.thenThrow(new RuntimeException("boom"));
|
||||
|
||||
PaygMeterReportingService service = newService(ENDPOINT, TOKEN);
|
||||
assertThatCode(
|
||||
() ->
|
||||
service.recordUsage(
|
||||
100L,
|
||||
"cus_abc",
|
||||
7,
|
||||
BillingCategory.AI,
|
||||
"process:job4:close"))
|
||||
.doesNotThrowAnyException();
|
||||
|
||||
assertThat(errorsCounter.count()).isEqualTo(1.0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordUsage_blankEndpoint_noopsAndDoesNotCallRestTemplate() {
|
||||
PaygMeterReportingService service = newService("", TOKEN);
|
||||
service.recordUsage(100L, "cus_abc", 5, BillingCategory.API, "process:job5:close");
|
||||
|
||||
verify(restTemplate, never()).exchange(any(String.class), any(), any(), any(Class.class));
|
||||
assertThat(errorsCounter.count()).isZero();
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordUsage_nullEndpoint_noopsAndDoesNotCallRestTemplate() {
|
||||
PaygMeterReportingService service = newService(null, TOKEN);
|
||||
service.recordUsage(100L, "cus_abc", 5, BillingCategory.API, "process:job6:close");
|
||||
|
||||
verify(restTemplate, never()).exchange(any(String.class), any(), any(), any(Class.class));
|
||||
assertThat(errorsCounter.count()).isZero();
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordUsage_zeroUnits_noopsAndDoesNotCallRestTemplate() {
|
||||
PaygMeterReportingService service = newService(ENDPOINT, TOKEN);
|
||||
service.recordUsage(100L, "cus_abc", 0, BillingCategory.API, "process:job7:close");
|
||||
|
||||
verify(restTemplate, never()).exchange(any(String.class), any(), any(), any(Class.class));
|
||||
assertThat(errorsCounter.count()).isZero();
|
||||
}
|
||||
|
||||
@Test
|
||||
void recordUsage_blankServiceRoleToken_postsWithoutAuthorizationHeader() {
|
||||
when(restTemplate.exchange(eq(ENDPOINT), eq(HttpMethod.POST), any(), eq(String.class)))
|
||||
.thenReturn(new ResponseEntity<>("{}", HttpStatus.OK));
|
||||
|
||||
PaygMeterReportingService service = newService(ENDPOINT, "");
|
||||
service.recordUsage(100L, "cus_abc", 5, BillingCategory.API, "process:job8:close");
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
ArgumentCaptor<HttpEntity<Map<String, Object>>> entityCaptor =
|
||||
ArgumentCaptor.forClass(HttpEntity.class);
|
||||
verify(restTemplate, times(1))
|
||||
.exchange(
|
||||
eq(ENDPOINT),
|
||||
eq(HttpMethod.POST),
|
||||
entityCaptor.capture(),
|
||||
eq(String.class));
|
||||
assertThat(entityCaptor.getValue().getHeaders().getFirst("Authorization")).isNull();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user