diff --git a/app/saas/src/main/java/stirling/software/saas/payg/charge/JobChargeService.java b/app/saas/src/main/java/stirling/software/saas/payg/charge/JobChargeService.java index c8b44ad323..3fdb1f19d6 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/charge/JobChargeService.java +++ b/app/saas/src/main/java/stirling/software/saas/payg/charge/JobChargeService.java @@ -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). + * + *

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. + * + *

Skipped paths — no meter event fired: + * + *

+ */ + @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 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 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 diff --git a/app/saas/src/main/java/stirling/software/saas/payg/meter/PaygMeterReportingService.java b/app/saas/src/main/java/stirling/software/saas/payg/meter/PaygMeterReportingService.java new file mode 100644 index 0000000000..432c6fa5e6 --- /dev/null +++ b/app/saas/src/main/java/stirling/software/saas/payg/meter/PaygMeterReportingService.java @@ -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. + * + *

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. + * + *

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. + * + *

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::close"}) + * so a retry, a reconciliation replay, or a double-fire from two pods never charges twice. + * + *

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 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 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(); + } + } +} diff --git a/app/saas/src/main/resources/application-saas.properties b/app/saas/src/main/resources/application-saas.properties index 3676ad7d82..8fae5d891c 100644 --- a/app/saas/src/main/resources/application-saas.properties +++ b/app/saas/src/main/resources/application-saas.properties @@ -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 diff --git a/app/saas/src/test/java/stirling/software/saas/payg/charge/JobChargeServiceTest.java b/app/saas/src/test/java/stirling/software/saas/payg/charge/JobChargeServiceTest.java index 2db6b30dd8..66db85c731 100644 --- a/app/saas/src/test/java/stirling/software/saas/payg/charge/JobChargeServiceTest.java +++ b/app/saas/src/test/java/stirling/software/saas/payg/charge/JobChargeServiceTest.java @@ -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 stepLimits) { diff --git a/app/saas/src/test/java/stirling/software/saas/payg/meter/PaygMeterReportingServiceTest.java b/app/saas/src/test/java/stirling/software/saas/payg/meter/PaygMeterReportingServiceTest.java new file mode 100644 index 0000000000..152f324afd --- /dev/null +++ b/app/saas/src/test/java/stirling/software/saas/payg/meter/PaygMeterReportingServiceTest.java @@ -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>> entityCaptor = + ArgumentCaptor.forClass(HttpEntity.class); + verify(restTemplate) + .exchange( + eq(ENDPOINT), + eq(HttpMethod.POST), + entityCaptor.capture(), + eq(String.class)); + + HttpEntity> sent = entityCaptor.getValue(); + Map 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>> 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(); + } +}