diff --git a/backend/notifications/app/run.go b/backend/notifications/app/run.go index 118b0bb..a4c0034 100644 --- a/backend/notifications/app/run.go +++ b/backend/notifications/app/run.go @@ -45,8 +45,6 @@ func Run(ctx context.Context, c Config, logger mlogger.Logger) error { defer cancel() _ = client.Disconnect(cleanup) }() - startup, cancel := context.WithTimeout(ctx, 20*time.Second) - defer cancel() repo, err := store.New(logger, client.Database()) if err != nil { return err @@ -57,7 +55,7 @@ func Run(ctx context.Context, c Config, logger mlogger.Logger) error { } defer rt.Close() svc := service.New(repo, catalog, sender, string(c.Email.Driver), logger) - broker, err := bus.New(startup, rt.Durable, rt.DurableAdmin, repo, svc, logger, c.Queue) + broker, err := bus.New(ctx, rt.Durable, rt.DurableAdmin, repo, svc, logger, c.Queue) if err != nil { return err } diff --git a/backend/notifications/bus/dispatch.go b/backend/notifications/bus/dispatch.go index 37482c0..f62613a 100644 --- a/backend/notifications/bus/dispatch.go +++ b/backend/notifications/bus/dispatch.go @@ -35,9 +35,13 @@ func New(ctx context.Context, transport reliable.Transport, admin reliable.Admin } settings := reliable.DurableSettings{Enabled: true, Stream: config.Stream, Provisioning: config.Provisioning} consumer := reliable.DurableConsumerSettings{Durable: "email", Subject: subject, AckWaitSeconds: 60, MaxDeliver: -1, DeliverPolicy: reliable.DurableDeliverAll} - if err := admin.EnsureConsumerResources(ctx, settings, consumer); err != nil { + startup, cancel := context.WithTimeout(ctx, 20*time.Second) + defer cancel() + if err := admin.EnsureConsumerResources(startup, settings, consumer); err != nil { return nil, merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "configure notification messaging") } + // NATS retains this context and unsubscribes when it ends. The worker must + // live until service shutdown, independently of the provisioning deadline. subscription, err := transport.OpenConsumer(ctx, settings, consumer) if err != nil { return nil, merrors.OpaqueWrap(merrors.ErrServiceUnavailable, err, "subscribe notification worker") diff --git a/backend/notifications/bus/dispatch_test.go b/backend/notifications/bus/dispatch_test.go index 418b2fc..d4d65ae 100644 --- a/backend/notifications/bus/dispatch_test.go +++ b/backend/notifications/bus/dispatch_test.go @@ -10,6 +10,7 @@ import ( merrors "git.sendico.io/tech/sxchnge/shared/errors" "git.sendico.io/tech/sxchnge/shared/messaging/reliable" "go.mongodb.org/mongo-driver/v2/bson" + "go.uber.org/zap" ) type outbox struct{ jobs []model.Notification } @@ -23,6 +24,50 @@ type publisher struct { records []reliable.Record } +type consumerTransport struct { + reliable.Transport + reliable.Subscription + ctx context.Context +} + +func (t *consumerTransport) OpenConsumer(ctx context.Context, _ reliable.DurableSettings, _ reliable.DurableConsumerSettings) (reliable.Subscription, error) { + t.ctx = ctx + return t.Subscription, nil +} + +type consumerAdmin struct { + reliable.Admin + ctx context.Context +} + +func (a *consumerAdmin) EnsureConsumerResources(ctx context.Context, _ reliable.DurableSettings, _ reliable.DurableConsumerSettings) error { + a.ctx = ctx + return nil +} + +func TestSubscriptionOutlivesProvisioning(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + transport := &consumerTransport{} + admin := &consumerAdmin{} + if _, err := New(ctx, transport, admin, &outbox{}, nil, zap.NewNop(), DefaultConfig()); err != nil { + t.Fatal(err) + } + if admin.ctx.Err() != context.Canceled { + t.Fatal("provisioning context was not released") + } + if _, ok := admin.ctx.Deadline(); !ok { + t.Fatal("provisioning has no deadline") + } + if err := transport.ctx.Err(); err != nil { + t.Fatalf("subscription ended with provisioning: %v", err) + } + cancel() + if transport.ctx.Err() != context.Canceled { + t.Fatal("subscription does not follow service shutdown") + } +} + func (p *publisher) Publish(_ context.Context, r reliable.Record) error { if p.fail { return merrors.ServiceUnavailable("offline") diff --git a/backend/product/providers/sendico/client.go b/backend/product/providers/sendico/client.go index 0b15617..5d79d9e 100644 --- a/backend/product/providers/sendico/client.go +++ b/backend/product/providers/sendico/client.go @@ -12,7 +12,7 @@ import ( "go.mongodb.org/mongo-driver/v2/bson" ) -func (c *Client) post(ctx context.Context, resource, requestRef string, payload any) (receiptDTO, error) { +func (c *Client) post(ctx context.Context, resource, requestRef string, payload any, expectedStatus int) (receiptDTO, error) { var receipt receiptDTO token, err := c.token() if err != nil { @@ -38,7 +38,7 @@ func (c *Client) post(ctx context.Context, resource, requestRef string, payload } defer func() { _ = response.Body.Close() }() // Never decode or expose upstream error bodies, which can contain assertions. - if response.StatusCode != http.StatusOK { + if response.StatusCode != expectedStatus { return receipt, merrors.ServiceUnavailable("Sendico request was not accepted") } decoder := json.NewDecoder(io.LimitReader(response.Body, 16385)) @@ -54,14 +54,14 @@ func (c *Client) post(ctx context.Context, resource, requestRef string, payload } func (c *Client) Register(ctx context.Context, requestRef string) (string, error) { - dto, err := c.post(ctx, "individuals", requestRef, nil) + dto, err := c.post(ctx, "individuals", requestRef, nil, http.StatusCreated) if err != nil { return "", err } return mapSubject(dto) } func (c *Client) Submit(ctx context.Context, subject, source string, facts identity.Facts) (identity.Receipt, error) { - dto, err := c.post(ctx, "identity_facts", bson.NewObjectID().Hex(), mapFacts(subject, source, facts)) + dto, err := c.post(ctx, "identity_facts", bson.NewObjectID().Hex(), mapFacts(subject, source, facts), http.StatusOK) if err != nil { return identity.Receipt{}, err } diff --git a/backend/product/providers/sendico/client_test.go b/backend/product/providers/sendico/client_test.go index c466418..c776db9 100644 --- a/backend/product/providers/sendico/client_test.go +++ b/backend/product/providers/sendico/client_test.go @@ -48,6 +48,7 @@ func TestSignupWireContractAndCredentialRotation(t *testing.T) { if strings.Contains(r.URL.Path, "/individuals/") { require.Equal(t, "registration-request", r.Header.Get("X-Request-ID")) require.Empty(t, body) + w.WriteHeader(http.StatusCreated) _, _ = fmt.Fprintf(w, `{"status":"success","data":{"subject_ref":%q,"registered_at":"2026-10-07T10:30:00.123Z"}}`, subject) } else { require.JSONEq(t, `{"subject_ref":"`+subject+`","source_submission_ref":"signup-1","facts":{"full_name":"Ada Lovelace","email":"ada@example.com","external_customer_ref":"local-user"}}`, string(body)) @@ -95,6 +96,36 @@ func TestRejectInvalidConfiguration(t *testing.T) { require.Error(t, err) } +func TestSignupReceiptsRequireOperationStatus(t *testing.T) { + for _, status := range []int{200, 201, 202, 204, 206, 300, 403, 409, 503} { + t.Run(http.StatusText(status), func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(status) + _, _ = fmt.Fprintf(w, `{"status":"success","data":{"subject_ref":%q,"registered_at":"2026-10-07T10:30:00Z","submission_ref":%q,"accepted_at":"2026-10-07T10:30:00Z"}}`, subject, submission) + })) + defer server.Close() + client, err := Open(config(t, server.URL)) + require.NoError(t, err) + ref, err := client.Register(context.Background(), "registration-request") + if status == http.StatusCreated { + require.NoError(t, err) + require.Equal(t, subject, ref) + } else { + require.Error(t, err) + require.Empty(t, ref) + } + receipt, err := client.Submit(context.Background(), subject, "key", identity.Facts{}) + if status == http.StatusOK { + require.NoError(t, err) + require.Equal(t, submission, receipt.SubmissionRef) + } else { + require.Error(t, err) + require.Empty(t, receipt) + } + }) + } +} + func TestUpstreamFailuresNeverProduceReceiptsOrLeakBodies(t *testing.T) { for _, tc := range []struct { name string @@ -113,7 +144,11 @@ func TestUpstreamFailuresNeverProduceReceiptsOrLeakBodies(t *testing.T) { } { t.Run(tc.name, func(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(tc.status) + status := tc.status + if status == http.StatusOK && strings.Contains(r.URL.Path, "/individuals/") { + status = http.StatusCreated + } + w.WriteHeader(status) _, _ = io.WriteString(w, tc.body) })) defer server.Close() diff --git a/docs/signup-facts.md b/docs/signup-facts.md index bac4c8e..5e0ebe3 100644 --- a/docs/signup-facts.md +++ b/docs/signup-facts.md @@ -5,6 +5,58 @@ Task [#46](https://git.sendico.io/tech/sxchnge/issues/46), branch customer creation, runtime, shared database source and Flutter error feedback. Delivery: [PR #47](https://git.sendico.io/tech/sxchnge/pulls/47), targeting `main`. +## Registration response correction + +Follow-up [#57](https://git.sendico.io/tech/sxchnge/issues/57), branch +`identity-facts-57`, corrects the registration success status to HTTP 201 Created. +Sendico's individual creation handler returns 201; the original adapter and test +fixture incorrectly required 200. A successful registration therefore left a +pending customer with an unresolved claim and `signup_attention_required`. +The adapter now checks the exact status per operation before validating receipts: +201 for registration and 200 for Facts intake. Other statuses, including other +2xx responses, remain failures. Malformed receipts and uncertain registrations +still require the existing recovery rules. + +On 2026-10-17, the affected local signup was reconciled using its registration +correlation in Sendico logs and an authenticated organization-scoped individual +GET. A conditional majority write bound that confirmed individual to the existing +local claim, preserving the saved facts, namespace and registration request. +The rebuilt BFF resumed the public activation-resend flow: Facts intake returned +a persisted receipt and the endpoint returned 202 after activation-email queue +acceptance. No replacement individual was registered. The account remains +pending until the customer confirms the email; queue acceptance does not prove +inbox delivery. + +Recovery also exposed a notification worker lifetime issue: its NATS subscription +retained the 20-second startup context and was automatically unsubscribed after +that deadline. Resource provisioning now has its own bounded context, while the +subscription follows the service lifetime and closes at shutdown. A regression +test checks that releasing provisioning leaves the subscription context active, +and service cancellation still ends it. + +The rebuilt worker processed the queued activation message. The live Resend +provider rejected it with `provider_authentication_failed` (HTTP 401 or 403), +leaving delivery in `failed` after one attempt. The configured API key/sender +authorization must be corrected before sending another activation email. Rotate +the credential securely with `make email`, wait for Vault Agent to refresh it, +restart notifications so it reloads its environment, then use activation resend +with a new idempotency key. Do not replay the failed notification as a success or +mark the customer active on their behalf. + +Validation passed: `GOFLAGS=-p=2 make test` (all backend race tests and canonical +frontend suites), `make lint-backend`, focused provider/identity race tests and +notification bus/app race tests. The HTTP 201 regression first failed against the +old adapter and passed after correction. An earlier full run hit a database +credential-rotation timing failure at `rotation_test.go:249` during the Docker +build; the isolated race-test retry and both subsequent full runs passed. +Local BFF and notifications images were rebuilt with fresh upstream manifests +and resolved stable digest pins; the BFF, session sidecar and notifications +services are healthy. NATS shows the worker still waiting for deliveries beyond +the startup deadline, with no pending or unacknowledged jobs. Live email delivery +is blocked by provider authorization as described above. No security scans were +requested or run. Unrelated notifications runtime settings are excluded from +the task commits. + ## Behavior and boundaries With `api.sendico` configured, the product BFF creates the local pending customer @@ -18,7 +70,7 @@ backfilled or blocked at login. The adapter consumes these existing Sendico contracts: -- `POST /individuals/{organization_ref}`, empty body, HTTP 200 registration +- `POST /individuals/{organization_ref}`, empty body, HTTP 201 registration receipt containing `subject_ref` and `registered_at`. - `POST /identity_facts/{organization_ref}`, HTTP 200 immutable intake receipt containing `submission_ref`, `subject_ref` and `accepted_at`. @@ -73,9 +125,10 @@ the adapter rereads it on every request. Startup rejects missing, expired or mismatched credentials and rejects Sendico configuration without notifications. HTTPS is required except for loopback HTTP test/development endpoints. Redirects -are rejected and each HTTP call has a four-second timeout. Non-200 responses, -malformed/truncated receipts, invalid identifiers, wrong subjects and unsuccessful -envelopes never count as acceptance. Upstream error bodies are not surfaced. +are rejected and each HTTP call has a four-second timeout. Unexpected response +statuses (registration requires 201; Facts intake requires 200), malformed/truncated +receipts, invalid identifiers, wrong subjects and unsuccessful envelopes never +count as acceptance. Upstream error bodies are not surfaced. Omitting `api.sendico` keeps existing local-only behavior. Enabling it also gates activation for previously pending customers; use activation resend to submit