Skip to content

Commit bef2e24

Browse files
bjcoombsbjcoombs
authored
feat(gateway-arch): move Stripe webhook handling to financial-gateway (#1391)
* feat(financial-gateway): define PaymentCapturedEvent and PaymentFailedEvent protos Adds financial_gateway_events/v1 proto package with PaymentCapturedEvent and PaymentFailedEvent message types. Registers corresponding Kafka topics and HTTP port constant. Part of 033-gateway-architecture task 9: Move webhook handling to financial-gateway. * feat(gateway-arch): move Stripe webhook handling from payment-order to financial-gateway Moves Stripe webhook reception from payment-order to financial-gateway, the correct architectural home for payment provider integration. financial-gateway changes: - Add HTTP webhook handler (adapters/http/webhook_handler.go) that validates Stripe-Signature, maps payment_intent.succeeded → PaymentCapturedEvent and payment_intent.payment_failed → PaymentFailedEvent, and publishes via transactional outbox - Extend ParsedWebhookEvent with AmountMinorUnits and Currency fields - Add ManifestTenantConfigProvider and envTenantConfigProvider for per-tenant webhook secrets (control-plane or env-var fallback) - Wire HTTP server on port 8095 (/webhooks/stripe) alongside existing gRPC server - Add outbox worker for Kafka publish (KAFKA_BOOTSTRAP_SERVERS or KAFKA_BROKERS) - Add event_outbox migration and Atlas config payment-order changes: - Remove StripeWebhookHandler, StripeEventProcessor, and related tests - Remove /webhook/stripe route from HTTP server - Add PaymentEventConsumer that subscribes to financial-gateway Kafka topics and calls UpdatePaymentOrder to transition orders to SETTLED or REJECTED * fix(payment-order): rename unused ctx parameter in consumer test stub Renames unused ctx parameter to _ to satisfy golangci-lint revive rule. * fix: remove payment-order Stripe webhook e2e test referencing deleted types The stripe e2e test file referenced StripeWebhookHandler, NewStripeEventProcessor, and StripeWebhookHandlerConfig which were deleted as part of moving webhook handling to financial-gateway. Since the Stripe webhook E2E tests no longer apply to payment-order (which now receives events via Kafka from financial-gateway), the file is removed. * fix: remove unused withSagaOrchestration helper from payment-order e2e tests withSagaOrchestration was only used by the deleted stripe_e2e_test.go. Removing to fix golangci-lint unused function error. * fix: register financial-gateway database mapping in migration runner The migration runner's ServiceDatabases map did not include financial-gateway, causing the unified binary --migrate flag to fail with "unknown service: no database mapping" when encountering the new financial-gateway migrations. * fix: add financial-gateway topics to topics.yaml and topics_test.go TestAll_ContainsAllConstants and TestTopicsYAML_ConsistentWithGoConstants both enforce that topics.go constants, topics.yaml, and topics.All() are consistent. The new financial-gateway topics were added to the Go constants and All() slice but were missing from topics.yaml and the test's knownConstants list. * fix: address CodeRabbit review comments for webhook migration - Fix silent proto deserialization corruption: use two separate kafka.ProtoConsumer instances (one per topic) with the correct typed msgFactory, preventing PaymentFailedEvent bytes from being decoded as PaymentCapturedEvent - Fix missing tenant context: change route to POST /webhooks/stripe/{tenantID} and extract tenant via r.PathValue("tenantID") instead of relying on middleware that was never wired; inject into ctx via tenant.WithTenant - Fix nil OutboxPublisher not validated: panic in NewWebhookHandler when OutboxPublisher is nil, consistent with ClientFactory guard - Fix gRPC listener resource leak: add listenerClosed flag with deferred close so listener is not abandoned on startup failures after bind - Fix serverErrors channel buffer too small: increase from 2 to 3 to accommodate gRPC + HTTP + payment event consumer goroutines - Use static sentinel errors (ErrUnexpectedCapturedMessageType, ErrUnexpectedFailedMessageType) for type assertion failures in consumers to satisfy err113 linter - Update webhook_handler_test.go to use r.SetPathValue("tenantID") instead of tenant.WithTenant context injection, matching new handler behaviour; remove unused tenant import - Update payment-order/cmd/main.go to use new Start(capturedTopic, failedTopic string) signature for PaymentEventConsumer * fix: address remaining CodeRabbit review comments - Validate payment_order_id before publishing to outbox: events missing this metadata field (non-Meridian Stripe payments) are now acknowledged with a warning log instead of returning 500 and causing infinite retries - Add HTTP server timeouts (ReadHeaderTimeout/ReadTimeout/WriteTimeout/ IdleTimeout) to financial-gateway webhook server to prevent slowloris connection exhaustion attacks - Guard against nil PaymentOrderUpdater in both PaymentEventConsumer constructors; NewPaymentEventConsumer panics, NewPaymentEventConsumerWithKafka returns ErrNilPaymentOrderUpdater - Move NewPaymentEventConsumerWithKafka construction before server start goroutines so initialization failures return cleanly without leaving active listeners/goroutines behind * docs: clarify intentional insecure gRPC credentials for inter-service calls Inter-service gRPC connections use insecure.NewCredentials() by design across the Meridian platform. Network-layer security (mTLS via service mesh) handles encryption for cluster-internal traffic. This matches the pattern used in position-keeping, reconciliation, tenant, and other services. --------- Co-authored-by: Ben Coombs <bjcoombs@users.noreply.github.com>
1 parent 580b37b commit bef2e24

25 files changed

Lines changed: 1587 additions & 1996 deletions

File tree

‎api/proto/meridian/financial_gateway_events/v1/events.proto‎

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
syntax = "proto3";
2+
3+
package meridian.financial_gateway_events.v1;
4+
5+
import "buf/validate/validate.proto";
6+
import "google/protobuf/timestamp.proto";
7+
8+
option go_package = "github.com/meridianhub/meridian/api/proto/meridian/financial_gateway_events/v1;financialgatewayeventsv1";
9+
10+
// PaymentCapturedEvent is published when a Stripe payment_intent.succeeded webhook
11+
// is received and validated. It signals that a payment has been successfully captured
12+
// by the payment provider.
13+
message PaymentCapturedEvent {
14+
// event_id uniquely identifies this event instance (UUID).
15+
string event_id = 1 [(buf.validate.field).string.uuid = true];
16+
17+
// correlation_id links all events across services for a single user request.
18+
string correlation_id = 2 [(buf.validate.field).string.max_len = 255];
19+
20+
// causation_id identifies the provider event that caused this domain event.
21+
string causation_id = 3 [(buf.validate.field).string.max_len = 255];
22+
23+
// version is the event schema version for forward compatibility.
24+
int32 version = 4 [(buf.validate.field).int32.gte = 1];
25+
26+
// payment_order_id is the Meridian payment order associated with this capture.
27+
// May be empty if the Stripe payment was not created via Meridian (no payment_order_id in metadata).
28+
string payment_order_id = 5 [(buf.validate.field).string.max_len = 255];
29+
30+
// provider_reference_id is the payment provider's identifier (e.g., Stripe PaymentIntent ID).
31+
string provider_reference_id = 6 [(buf.validate.field).string = {
32+
min_len: 1
33+
max_len: 255
34+
}];
35+
36+
// amount_minor_units is the captured amount in the smallest currency unit (e.g., cents for USD).
37+
int64 amount_minor_units = 7 [(buf.validate.field).int64.gte = 0];
38+
39+
// currency is the ISO 4217 currency code (e.g., "USD", "GBP").
40+
// May be empty if the provider did not include currency in the webhook payload.
41+
string currency = 8 [(buf.validate.field).string.max_len = 3];
42+
43+
// captured_at is when the payment was captured by the provider.
44+
google.protobuf.Timestamp captured_at = 9;
45+
46+
// provider_event_id is the payment provider's webhook event ID (e.g., Stripe evt_ ID).
47+
// Used for idempotency and deduplication.
48+
string provider_event_id = 10 [(buf.validate.field).string = {
49+
min_len: 1
50+
max_len: 255
51+
}];
52+
53+
// metadata contains additional key-value pairs from the provider event.
54+
map<string, string> metadata = 11 [(buf.validate.field).map = {
55+
max_pairs: 64
56+
keys: {string: {max_len: 128}}
57+
values: {string: {max_len: 1024}}
58+
}];
59+
}
60+
61+
// PaymentFailedEvent is published when a Stripe payment_intent.payment_failed webhook
62+
// is received and validated. It signals that a payment attempt was rejected by the provider.
63+
message PaymentFailedEvent {
64+
// event_id uniquely identifies this event instance (UUID).
65+
string event_id = 1 [(buf.validate.field).string.uuid = true];
66+
67+
// correlation_id links all events across services for a single user request.
68+
string correlation_id = 2 [(buf.validate.field).string.max_len = 255];
69+
70+
// causation_id identifies the provider event that caused this domain event.
71+
string causation_id = 3 [(buf.validate.field).string.max_len = 255];
72+
73+
// version is the event schema version for forward compatibility.
74+
int32 version = 4 [(buf.validate.field).int32.gte = 1];
75+
76+
// payment_order_id is the Meridian payment order associated with this failure.
77+
// May be empty if the Stripe payment was not created via Meridian (no payment_order_id in metadata).
78+
string payment_order_id = 5 [(buf.validate.field).string.max_len = 255];
79+
80+
// provider_reference_id is the payment provider's identifier (e.g., Stripe PaymentIntent ID).
81+
string provider_reference_id = 6 [(buf.validate.field).string = {
82+
min_len: 1
83+
max_len: 255
84+
}];
85+
86+
// failure_reason is a human-readable description of why the payment failed.
87+
string failure_reason = 7 [(buf.validate.field).string.max_len = 1024];
88+
89+
// failure_code is the provider's machine-readable failure code (e.g., Stripe decline code).
90+
string failure_code = 8 [(buf.validate.field).string.max_len = 255];
91+
92+
// provider_event_id is the payment provider's webhook event ID (e.g., Stripe evt_ ID).
93+
// Used for idempotency and deduplication.
94+
string provider_event_id = 9 [(buf.validate.field).string = {
95+
min_len: 1
96+
max_len: 255
97+
}];
98+
99+
// failed_at is when the payment failure occurred.
100+
google.protobuf.Timestamp failed_at = 10;
101+
}

‎internal/migrations/runner.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ var ServiceDatabases = map[string]ServiceDatabase{
7979
"forecasting": {Database: "meridian_forecasting", User: "meridian_forecasting_user", Password: ""},
8080
"reference-data": {Database: "meridian_reference_data", User: "meridian_reference_data_user", Password: ""},
8181
"operational-gateway": {Database: "meridian_operational_gateway", User: "meridian_operational_gateway_user", Password: ""},
82+
"financial-gateway": {Database: "meridian_financial_gateway", User: "meridian_financial_gateway_user", Password: ""},
8283
}
8384

8485
// serviceMigration holds a single migration file for a service.

0 commit comments

Comments
 (0)