From 3bfd0f67f4eac8dd6aed4aa8cb1a63b986be7f29 Mon Sep 17 00:00:00 2001 From: Christian Galo Date: Fri, 23 Oct 2026 19:11:03 -0400 Subject: [PATCH] Read the Temporal connection per request, and answer for its absence Fifth section of #141. internal/workflows/clientholder holds the Temporal client: start builds an empty holder, hands it to the server and the integrations, and publishes the client once, after the worker starts and before the first Startup hook; a second Publish panics. Every handler and integration that starts a workflow reads the holder when a request arrives, so a client that connects after the listener opens reaches them without a restart; a test scans the handler configurations and fails on any field that still holds a client.Client. While the holder is empty: - POST /webhooks/stripe stores the event as received, then answers 503 with Retry-After: 5 and the body "Temporal is not connected", and logs that it answered 503 so Stripe redelivers it. A redelivery after Temporal connects starts the event's workflow, and the processing guard keeps a finished event from running twice. - A Discourse event is recorded as failed with "start reconcile: Temporal is not connected"; the sweep restores what it missed. - Issuing or managing a grant with an end date is refused under the end-date field with "An end date needs the expiration scheduler, which is not connected.", after the date itself is checked, and writes nothing. The operator page's "no scheduler" warnings, which saved the grant and left it without an expiry, are gone. - FedWiki's site actions and the domain claims answer that they are not available, as they did with no Temporal configured. The operator test harnesses and the plan-management end-to-end harness publish a holder over a fake client whose ExecuteWorkflow succeeds, so every bounded save keeps its result. Full suite on reset databases: 4485 pass, 15 skip, none failed. The console booted on the test stack and connected to Temporal. go test -race -p 1 over clientholder, internal/server and the three integrations after reset-test-db.sh: every package passes, no data race. make lint passes. Breaks, each caught: - 5.10 member domains handler reads the client at construction - 5.10 FedWiki site action handler reads the client at construction - 5.10 Stripe environment-check handler reads the client at construction - 5.10 the holder is a plain field in place of an atomic.Pointer (-race) - 5.10 a client.Client field added back to MemberDomainsConfig - 5.10 the scan ignores import names, so a temporalsdk.Client field would pass - 5.10 Stripe StartProcessing reads the holder at registration - 5.10 the webhook answers 500 for ErrNotConnected - 5.10 the webhook answers 503 before the insert - 5.10 Discourse StartReconcile calls the nil client - 5.10 the issuance refusal after ConferGrant - 5.10 the management refusal removed - 5.10 the issuance refusal applied to a grant without an end date - 5.10 the management refusal applied to a save that clears the end date - 5.10 Publish stores a second client without panicking - 5.10 a temporalsdk.Client field added back to a FedWiki handler configuration - 5.10 Discourse StartReconcile reads the holder at registration - 5.10 the issuance refusal before the parse - 5.10 the management refusal before the parse - 5.10 the issuance refusal is a form error, not under the end date --- cmd/start.go | 15 +- docs/building-an-integration.md | 4 +- internal/integrations/discourse/discourse.go | 15 +- .../integrations/discourse/web/operator.go | 8 +- .../discourse/web/webhook_temporal_db_test.go | 103 ++++++++ internal/integrations/fedwiki/fedwiki.go | 6 +- internal/integrations/fedwiki/web/api.go | 21 +- .../integrations/fedwiki/web/customdomain.go | 8 +- internal/integrations/fedwiki/web/operator.go | 8 +- .../integrations/fedwiki/web/oracle_test.go | 18 +- internal/integrations/fedwiki/web/partials.go | 8 +- .../fedwiki/web/requests_db_test.go | 54 +++- .../integrations/fedwiki/web/site_actions.go | 20 +- internal/integrations/stripe/stripe.go | 20 +- internal/integrations/stripe/web/webhook.go | 20 +- .../stripe/web/webhook_temporal_db_test.go | 176 +++++++++++++ .../server/bounded_grant_temporal_db_test.go | 244 ++++++++++++++++++ .../server/grant_resumption_operator_test.go | 11 +- internal/server/member_domains.go | 54 ++-- .../server/member_domains_rollback_db_test.go | 8 +- internal/server/new_handler_test.go | 15 +- .../no_handler_holds_temporal_client_test.go | 193 ++++++++++++++ internal/server/operator_domains.go | 4 +- internal/server/operator_enrollment.go | 74 ++++-- .../server/operator_entitlement_recompute.go | 5 +- internal/server/operator_grants_list_test.go | 4 + .../server/operator_integrations_stripe.go | 10 +- .../operator_integrations_stripe_test.go | 2 +- .../server/operator_org_type_change_test.go | 38 ++- internal/server/operator_partials.go | 11 +- internal/server/operator_plan_acts.go | 5 +- .../operator_stripe_public_details_test.go | 2 +- internal/server/server.go | 36 +-- internal/server/temporal_holder_test.go | 31 +++ .../server/temporal_per_request_db_test.go | 117 +++++++++ .../workflows/clientholder/clientholder.go | 83 ++++++ .../clientholder/clientholder_test.go | 124 +++++++++ .../serve-before-dependencies/design.md | 1 + .../serve-before-dependencies/tasks.md | 32 ++- test/e2e/plan-management/e2e_test.go | 32 ++- 40 files changed, 1444 insertions(+), 196 deletions(-) create mode 100644 internal/integrations/discourse/web/webhook_temporal_db_test.go create mode 100644 internal/integrations/stripe/web/webhook_temporal_db_test.go create mode 100644 internal/server/bounded_grant_temporal_db_test.go create mode 100644 internal/server/no_handler_holds_temporal_client_test.go create mode 100644 internal/server/temporal_holder_test.go create mode 100644 internal/server/temporal_per_request_db_test.go create mode 100644 internal/workflows/clientholder/clientholder.go create mode 100644 internal/workflows/clientholder/clientholder_test.go diff --git a/cmd/start.go b/cmd/start.go index 1a90b736..6ee58b2f 100644 --- a/cmd/start.go +++ b/cmd/start.go @@ -29,6 +29,7 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/server" "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows" wfBilling "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/billing" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/desiredstate" wfDomains "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/domains" wfEnt "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/entitlements" @@ -285,6 +286,11 @@ var startCmd = &cobra.Command{ // domains at. Empty disables external domain claims deployment-wide. domainsConnectTarget := viper.GetString("domains-connect-target") + // The holder the router's Temporal-backed handlers read per request. + // It is empty until a client is published, which happens below once + // the worker is polling; without temporal-host it stays empty. + temporalHolder := clientholder.New() + // Create Temporal client if configured var temporalClient client.Client temporalHost := viper.GetString("temporal-host") @@ -341,6 +347,13 @@ var startCmd = &cobra.Command{ } defer worker.Stop() + // Publish after the worker has started, so a workflow a handler + // starts has a worker, and before the first Startup hook, so the + // sweeps the hooks run see every row recorded while the holder + // was empty (design D7). Temporal still connects before the + // listener opens in this order. + temporalHolder.Publish(temporalClient) + // Run each installed integration's Startup hook (Stripe's hand-started // webhook/outbox-poller workflows, FedWiki's sync schedule, ...) // instead of naming them here. Each implementation logs and swallows @@ -513,7 +526,7 @@ var startCmd = &cobra.Command{ BillingQ: billing.New(database), StripeQ: stripemod.New(database), DomainsQ: domains.New(database), - TemporalClient: temporalClient, + Temporal: temporalHolder, StripeWebhookSecret: viper.GetString("stripe-webhook-secret"), StripeAPIKey: viper.GetString("stripe-api-key"), StripeDashboardURL: stripeDashboardURL, diff --git a/docs/building-an-integration.md b/docs/building-an-integration.md index 222c75b2..4f902bb7 100644 --- a/docs/building-an-integration.md +++ b/docs/building-an-integration.md @@ -374,7 +374,9 @@ type RouteProvider interface { ``` `RegisterRoutes` constructs your handlers from `Deps` (the DB connection, -core module queriers, the Temporal client, the auth config, the logger — no +core module queriers, the Temporal holder (`Deps.Temporal`, empty until +Temporal connects; a handler calls `Client()` when a request arrives, never +while its routes are registered), the auth config, the logger — no core code ever names an integration handler type) and registers them on the shared mux. A handler that renders an operator page builds its set with `web.OperatorTemplates`, and one that renders a member page with diff --git a/internal/integrations/discourse/discourse.go b/internal/integrations/discourse/discourse.go index 41c35b4b..052fd0f3 100644 --- a/internal/integrations/discourse/discourse.go +++ b/internal/integrations/discourse/discourse.go @@ -46,6 +46,7 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/discourse/workflows" "git.coopcloud.tech/wiki-cafe/member-console/internal/server" corework "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/queues" "github.com/spf13/viper" "go.temporal.io/sdk/client" @@ -194,7 +195,7 @@ func (Adapter) Static() fs.FS { // and the page itself states the unconfigured state (spec: "Unconfigured // integration") — while mutation routes and the webhook need a reachable // forum and so mount only when configured. deps supplies the DB and -// Temporal client; Discourse-specific config is read from viper (this +// Temporal holder; Discourse-specific config is read from viper (this // integration's own concern). func (Adapter) RegisterRoutes(mux *http.ServeMux, deps server.Deps) error { var c *dcclient.Client @@ -208,7 +209,7 @@ func (Adapter) RegisterRoutes(mux *http.ServeMux, deps server.Deps) error { Logger: deps.Logger, AuthConfig: deps.AuthConfig, Client: c, - Temporal: deps.TemporalClient, + Temporal: deps.Temporal, Configured: configured(), Description: dcmod.ProviderSource().ProviderManifest().Description, TemplatesFS: templatesFS(), @@ -252,15 +253,21 @@ func (Adapter) RegisterRoutes(mux *http.ServeMux, deps server.Deps) error { mux.HandleFunc("GET /partials/discourse/forum-access", memberHandler.GetForumAccess) if viper.GetString("discourse-webhook-secret") != "" { - temporalClient := deps.TemporalClient webhookHandler := web.NewWebhookHandler(web.WebhookHandlerConfig{ DB: deps.Database, Logger: deps.Logger, Secret: viper.GetString("discourse-webhook-secret"), Mode: mode, Client: c, + // The holder is read when an event arrives, not when the route + // is built: with no client the event is recorded failed and the + // periodic sweep converges the person (design D9). StartReconcile: func(personID string) error { - _, err := temporalClient.ExecuteWorkflow(context.Background(), client.StartWorkflowOptions{ + tc := deps.Temporal.Client() + if tc == nil { + return clientholder.ErrNotConnected + } + _, err := tc.ExecuteWorkflow(context.Background(), client.StartWorkflowOptions{ ID: workflows.ReconcilePersonWorkflowID(personID), TaskQueue: queues.Main, }, workflows.ReconcilePersonWorkflow, personID) diff --git a/internal/integrations/discourse/web/operator.go b/internal/integrations/discourse/web/operator.go index eb2e74ba..1fa43f0a 100644 --- a/internal/integrations/discourse/web/operator.go +++ b/internal/integrations/discourse/web/operator.go @@ -18,7 +18,7 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/server" "git.coopcloud.tech/wiki-cafe/member-console/internal/web" "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows" - temporalsdk "go.temporal.io/sdk/client" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" ) // groupMappingConstraints translates this form's plausible constraint @@ -93,7 +93,7 @@ type DiscourseOperatorHandler struct { logger *slog.Logger authConfig *auth.Config apiClient *dcclient.Client - temporal temporalsdk.Client + temporal *clientholder.Holder configured bool description string sweepScheduleID string @@ -108,7 +108,7 @@ type DiscourseOperatorHandlerConfig struct { // Client is nil when the integration is unconfigured (the page then // renders only the unconfigured notice; mutation routes are unmounted). Client *dcclient.Client - Temporal temporalsdk.Client + Temporal *clientholder.Holder Configured bool // Description: see OperatorPageData.Description. Description string @@ -231,7 +231,7 @@ func (h *DiscourseOperatorHandler) buildPageData(r *http.Request) OperatorPageDa // sweepHealth reports the most recent schedule-spawned sweep's status and // time; "no sweep yet" when the schedule hasn't fired or Temporal is absent. func (h *DiscourseOperatorHandler) sweepHealth(r *http.Request) (string, string) { - status, when := workflows.ScheduleRunStatus(r.Context(), h.temporal, h.sweepScheduleID) + status, when := workflows.ScheduleRunStatus(r.Context(), h.temporal.Client(), h.sweepScheduleID) if status == "" { return "No sweep has run yet", "" } diff --git a/internal/integrations/discourse/web/webhook_temporal_db_test.go b/internal/integrations/discourse/web/webhook_temporal_db_test.go new file mode 100644 index 00000000..f0d72504 --- /dev/null +++ b/internal/integrations/discourse/web/webhook_temporal_db_test.go @@ -0,0 +1,103 @@ +// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial +// SPDX-FileCopyrightText: 2025-2026 Christian Galo + +package web_test + +import ( + "io" + "log/slog" + "net/http" + "testing" + + "github.com/google/uuid" + "github.com/spf13/viper" + "github.com/stretchr/testify/mock" + "go.temporal.io/sdk/mocks" + + "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/discourse" + "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/discourse/discoursetest" + "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/discourse/linkage" + "git.coopcloud.tech/wiki-cafe/member-console/internal/server" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" +) + +// setViper sets a viper key for the test and restores what it held. +func setViper(t *testing.T, key string, value any) { + t.Helper() + prev, had := viper.Get(key), viper.IsSet(key) + viper.Set(key, value) + t.Cleanup(func() { + if had { + viper.Set(key, prev) + } else { + viper.Set(key, nil) + } + }) +} + +// adapterWebhookFixture is the webhook fixture with its mux replaced by the +// one the Discourse adapter registers, so the start function under test is +// the adapter's own, reading the holder it was given. +func adapterWebhookFixture(t *testing.T, holder *clientholder.Holder) *webhookFixture { + t.Helper() + f := newWebhookFixture(t, linkage.ModeOIDC) + setViper(t, "discourse-base-url", f.fake.URL()) + setViper(t, "discourse-api-key", "k") + setViper(t, "discourse-api-username", "system") + setViper(t, "discourse-linkage-mode", "oidc") + setViper(t, "discourse-webhook-secret", testSecret) + f.mux = http.NewServeMux() + deps := server.Deps{Database: f.database, Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), Temporal: holder} + if err := (discourse.Adapter{}).RegisterRoutes(f.mux, deps); err != nil { + t.Fatalf("RegisterRoutes: %v", err) + } + return f +} + +// groupEvent delivers a user_added_to_group event for a managed group and a +// linked person, and returns the event id. +func (f *webhookFixture) groupEvent(gid, uid int64) string { + f.t.Helper() + eventID := uuid.New().String() + code := f.deliver("user_added_to_group", map[string]any{ + "group_user": map[string]any{"group_id": gid, "user_id": uid, "notification_level": 3, "id": 31, "created_at": "2026-07-20T01:00:00.000Z"}, + }, true, eventID) + if code != http.StatusOK { + f.t.Fatalf("delivery = %d, want 200", code) + } + return eventID +} + +// TestWebhookBeforeTemporalRecordsFailed registers the webhook through the +// Discourse adapter with an empty Temporal holder. An event for a managed +// group and a linked person is acknowledged with 200 and recorded `failed` +// with the sentence copy row 5 names, and nothing panics; after Publish the +// same handler records the next event `processed` and starts the reconcile. +func TestWebhookBeforeTemporalRecordsFailed(t *testing.T) { + holder := clientholder.New() + f := adapterWebhookFixture(t, holder) + personID := f.newPerson("bt-"+uuid.New().String()[:8]+"@example.com", "subj-"+uuid.New().String()[:8]) + groupName := "managed-" + uuid.New().String()[:8] + gid := f.fake.AddGroup(groupName, false) + f.newMapping(groupName, gid) + uid := f.fake.AddUser(discoursetest.FakeUser{Username: "btuser", Email: "bt@example.com", Active: true}) + f.linkPerson(personID, uid, "btuser") + + before := f.groupEvent(gid, uid) + if got := f.eventStatus(before); got != "failed" { + t.Errorf("status with an empty holder = %q, want failed", got) + } + if got, want := f.eventError(before), "start reconcile: Temporal is not connected"; got != want { + t.Errorf("error_message = %q, want %q", got, want) + } + + fake := &mocks.Client{} + fake.On("ExecuteWorkflow", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&mocks.WorkflowRun{}, nil) + holder.Publish(fake) + + after := f.groupEvent(gid, uid) + if got := f.eventStatus(after); got != "processed" { + t.Errorf("status after Publish = %q, want processed", got) + } + fake.AssertNumberOfCalls(t, "ExecuteWorkflow", 1) +} diff --git a/internal/integrations/fedwiki/fedwiki.go b/internal/integrations/fedwiki/fedwiki.go index 24acc112..e0cd1548 100644 --- a/internal/integrations/fedwiki/fedwiki.go +++ b/internal/integrations/fedwiki/fedwiki.go @@ -260,7 +260,7 @@ func (a Adapter) RegisterRoutes(mux *http.ServeMux, deps server.Deps) error { Registry: registry, Database: deps.Database, Logger: deps.Logger, - TemporalClient: deps.TemporalClient, + Temporal: deps.Temporal, AuthConfig: deps.AuthConfig, FedWikiAllowedDomains: allowedDomains, FedWikiSiteScheme: siteScheme, @@ -277,7 +277,7 @@ func (a Adapter) RegisterRoutes(mux *http.ServeMux, deps server.Deps) error { Registry: registry, Database: deps.Database, Logger: deps.Logger, - TemporalClient: deps.TemporalClient, + Temporal: deps.Temporal, AuthConfig: deps.AuthConfig, FedWikiAllowedDomains: allowedDomains, FedWikiSiteScheme: siteScheme, @@ -301,7 +301,7 @@ func (a Adapter) RegisterRoutes(mux *http.ServeMux, deps server.Deps) error { AuthConfig: deps.AuthConfig, FedWikiSiteScheme: siteScheme, TemplatesFS: templatesFS(), - Temporal: deps.TemporalClient, + Temporal: deps.Temporal, SyncScheduleID: workflows.SyncScheduleID, Configured: configured(), Description: fwmod.ProviderSource().ProviderManifest().Description, diff --git a/internal/integrations/fedwiki/web/api.go b/internal/integrations/fedwiki/web/api.go index d320b1ff..f08c5c73 100644 --- a/internal/integrations/fedwiki/web/api.go +++ b/internal/integrations/fedwiki/web/api.go @@ -31,8 +31,8 @@ import ( fwmod "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/fedwiki/store" siteusage "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/fedwiki/usage" "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/fedwiki/workflows" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" "github.com/google/uuid" - "go.temporal.io/sdk/client" ) // FedWikiHandler handles FedWiki site management API endpoints. @@ -44,7 +44,7 @@ type FedWikiHandler struct { Registry *domains.Registry Database *sql.DB Logger *slog.Logger - TemporalClient client.Client + Temporal *clientholder.Holder AuthConfig *auth.Config FedWikiAllowedDomains []string // Domains where users can create sites FedWikiSiteScheme string // http or https @@ -66,7 +66,7 @@ type FedWikiHandlerConfig struct { Registry *domains.Registry Database *sql.DB Logger *slog.Logger - TemporalClient client.Client + Temporal *clientholder.Holder AuthConfig *auth.Config FedWikiAllowedDomains []string FedWikiSiteScheme string @@ -84,7 +84,7 @@ func NewFedWikiHandler(cfg FedWikiHandlerConfig) *FedWikiHandler { Registry: cfg.Registry, Database: cfg.Database, Logger: cfg.Logger, - TemporalClient: cfg.TemporalClient, + Temporal: cfg.Temporal, AuthConfig: cfg.AuthConfig, FedWikiAllowedDomains: cfg.FedWikiAllowedDomains, FedWikiSiteScheme: cfg.FedWikiSiteScheme, @@ -298,6 +298,8 @@ func (h *FedWikiHandler) CreateSite(w http.ResponseWriter, r *http.Request) { respondError(w, http.StatusUnauthorized, "Unauthorized", "") return } + // Read once: the connection may arrive after the routes are registered. + tc := h.Temporal.Client() var req CreateSiteRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { @@ -392,7 +394,7 @@ func (h *FedWikiHandler) CreateSite(w http.ResponseWriter, r *http.Request) { respondError(w, http.StatusForbidden, "Your current plan doesn't include custom domains.", h.SupportURL) return } - initiation, msg := startClaimVerification(r.Context(), h.Registry, h.TemporalClient, h.Logger, + initiation, msg := startClaimVerification(r.Context(), h.Registry, tc, h.Logger, session.WorkspaceID, req.Domain, h.CustomDomainTarget) if msg != "" { respondError(w, http.StatusBadRequest, msg, h.SupportURL) @@ -417,7 +419,7 @@ func (h *FedWikiHandler) CreateSite(w http.ResponseWriter, r *http.Request) { } // Check if Temporal client is available - if h.TemporalClient == nil { + if tc == nil { h.Logger.Error("Temporal client not configured") respondError(w, http.StatusServiceUnavailable, msgServiceAvailable, h.SupportURL) return @@ -488,7 +490,7 @@ func (h *FedWikiHandler) CreateSite(w http.ResponseWriter, r *http.Request) { return } - started := startRequestWorkflow(r.Context(), h.TemporalClient, h.Database, h.Logger, + started := startRequestWorkflow(r.Context(), tc, h.Database, h.Logger, workflowID, workflows.CreateFedWikiSiteRequestWorkflow, workflows.CreateSiteRequestInput{ ConnectionID: h.connectionID(), InstanceID: instance, @@ -522,6 +524,7 @@ func (h *FedWikiHandler) DeleteSite(w http.ResponseWriter, r *http.Request) { respondError(w, http.StatusUnauthorized, "Unauthorized", "") return } + tc := h.Temporal.Client() domain := r.PathValue("domain") if domain == "" { @@ -541,7 +544,7 @@ func (h *FedWikiHandler) DeleteSite(w http.ResponseWriter, r *http.Request) { } } - if h.TemporalClient == nil { + if tc == nil { h.Logger.Error("Temporal client not configured") respondError(w, http.StatusServiceUnavailable, msgServiceAvailable, h.SupportURL) return @@ -625,7 +628,7 @@ func (h *FedWikiHandler) DeleteSite(w http.ResponseWriter, r *http.Request) { return } - started := startRequestWorkflow(r.Context(), h.TemporalClient, h.Database, h.Logger, + started := startRequestWorkflow(r.Context(), tc, h.Database, h.Logger, workflowID, workflows.DeleteFedWikiSiteRequestWorkflow, workflows.DeleteSiteRequestInput{ ConnectionID: h.connectionID(), InstanceID: site.SiteID, diff --git a/internal/integrations/fedwiki/web/customdomain.go b/internal/integrations/fedwiki/web/customdomain.go index 7e1eb25c..a8a90804 100644 --- a/internal/integrations/fedwiki/web/customdomain.go +++ b/internal/integrations/fedwiki/web/customdomain.go @@ -445,8 +445,8 @@ func (h *FedWikiPartialsHandler) CheckCustomDomainNow(w http.ResponseWriter, r * if !ok { return } - if claim.Status == domains.StatusPending && h.TemporalClient != nil { - if err := h.TemporalClient.SignalWorkflow(r.Context(), + if tc := h.Temporal.Client(); claim.Status == domains.StatusPending && tc != nil { + if err := tc.SignalWorkflow(r.Context(), wfdomains.VerifyClaimWorkflowIDPrefix+claim.ClaimID, "", wfdomains.CheckNowSignal, nil); err != nil { h.Logger.Warn("failed to signal check-now", slog.Any("error", err)) @@ -472,8 +472,8 @@ func (h *FedWikiPartialsHandler) CancelCustomDomainVerification(w http.ResponseW return } if canceled { - if h.TemporalClient != nil { - if cErr := h.TemporalClient.CancelWorkflow(r.Context(), + if tc := h.Temporal.Client(); tc != nil { + if cErr := tc.CancelWorkflow(r.Context(), wfdomains.VerifyClaimWorkflowIDPrefix+claim.ClaimID, ""); cErr != nil { // The claim is already terminal; an orphaned poller records // nothing (probe writes are pending-guarded) and expires on diff --git a/internal/integrations/fedwiki/web/operator.go b/internal/integrations/fedwiki/web/operator.go index 2abcb49d..62285a58 100644 --- a/internal/integrations/fedwiki/web/operator.go +++ b/internal/integrations/fedwiki/web/operator.go @@ -15,7 +15,7 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/server" "git.coopcloud.tech/wiki-cafe/member-console/internal/web" "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows" - temporalsdk "go.temporal.io/sdk/client" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" ) // FedWikiSiteViewModel represents a site for the operator template @@ -85,7 +85,7 @@ type FedWikiOperatorHandler struct { AuthConfig *auth.Config FedWikiSiteScheme string // http or https; mirrors FedWikiHandler/FedWikiPartialsHandler's field Templates *server.SafeTemplates - Temporal temporalsdk.Client + Temporal *clientholder.Holder SyncScheduleID string // Configured mirrors DiscourseOperatorHandler's field: whether // FedWiki's required config group resolves, computed once by the @@ -113,7 +113,7 @@ type FedWikiOperatorHandlerConfig struct { // Temporal + SyncScheduleID feed the Sync health panel (nil/"" → // "no sync yet"); see workflows.ScheduleRunStatus for why health is // resolved through the schedule rather than a workflow ID. - Temporal temporalsdk.Client + Temporal *clientholder.Holder SyncScheduleID string // Configured: see FedWikiOperatorHandler.Configured. Configured bool @@ -166,7 +166,7 @@ func (h *FedWikiOperatorHandler) GetSitesPage(w http.ResponseWriter, r *http.Req return } - bodyData.SyncStatus, bodyData.SyncTime = workflows.ScheduleRunStatus(r.Context(), h.Temporal, h.SyncScheduleID) + bodyData.SyncStatus, bodyData.SyncTime = workflows.ScheduleRunStatus(r.Context(), h.Temporal.Client(), h.SyncScheduleID) if bodyData.SyncStatus == "" { bodyData.SyncStatus = "No sync has run yet" } diff --git a/internal/integrations/fedwiki/web/oracle_test.go b/internal/integrations/fedwiki/web/oracle_test.go index f5c17fa6..6fb313e7 100644 --- a/internal/integrations/fedwiki/web/oracle_test.go +++ b/internal/integrations/fedwiki/web/oracle_test.go @@ -92,7 +92,7 @@ func TestAPIDeleteSiteOracle(t *testing.T) { SiteQ: oracleFixtureSiteQ(), Logger: discardLogger(), // Never called: both requests are refused before any start. - TemporalClient: &mocks.Client{}, + Temporal: publishedHolder(&mocks.Client{}), } cfg, sctx := oracleSession(t, oracleCallerWorkspace) h.AuthConfig = cfg @@ -126,10 +126,10 @@ func TestPartialsDeleteSiteOracle(t *testing.T) { // (integration-delivery-core design D31), so it needs the database and // a posted address; no request exists at it, so the site decides. h := &FedWikiPartialsHandler{ - SiteQ: oracleFixtureSiteQ(), - Logger: discardLogger(), - Database: newClaimTestDB(t), - TemporalClient: &mocks.Client{}, + SiteQ: oracleFixtureSiteQ(), + Logger: discardLogger(), + Database: newClaimTestDB(t), + Temporal: publishedHolder(&mocks.Client{}), } cfg, sctx := oracleSession(t, oracleCallerWorkspace) h.AuthConfig = cfg @@ -164,10 +164,10 @@ func TestPartialsStatusChangeOracle(t *testing.T) { // (integration-delivery-core design D31), so it needs the database and // a posted address; no request exists at it, so the site decides. h := &FedWikiPartialsHandler{ - SiteQ: oracleFixtureSiteQ(), - Logger: discardLogger(), - Database: newClaimTestDB(t), - TemporalClient: &mocks.Client{}, + SiteQ: oracleFixtureSiteQ(), + Logger: discardLogger(), + Database: newClaimTestDB(t), + Temporal: publishedHolder(&mocks.Client{}), } cfg, sctx := oracleSession(t, oracleCallerWorkspace) h.AuthConfig = cfg diff --git a/internal/integrations/fedwiki/web/partials.go b/internal/integrations/fedwiki/web/partials.go index fc202680..52a9e186 100644 --- a/internal/integrations/fedwiki/web/partials.go +++ b/internal/integrations/fedwiki/web/partials.go @@ -23,7 +23,7 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/fedwiki/workflows" "git.coopcloud.tech/wiki-cafe/member-console/internal/server" coreweb "git.coopcloud.tech/wiki-cafe/member-console/internal/web" - "go.temporal.io/sdk/client" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" ) // FedWikiPartialsHandler handles HTMX partial requests for FedWiki management. @@ -37,7 +37,7 @@ type FedWikiPartialsHandler struct { Registry *domains.Registry Database *sql.DB Logger *slog.Logger - TemporalClient client.Client + Temporal *clientholder.Holder AuthConfig *auth.Config FedWikiAllowedDomains []string // Domains where users can create sites FedWikiSiteScheme string // http or https @@ -63,7 +63,7 @@ type FedWikiPartialsConfig struct { Registry *domains.Registry Database *sql.DB Logger *slog.Logger - TemporalClient client.Client + Temporal *clientholder.Holder AuthConfig *auth.Config FedWikiAllowedDomains []string FedWikiSiteScheme string @@ -105,7 +105,7 @@ func NewFedWikiPartialsHandler(cfg FedWikiPartialsConfig) (*FedWikiPartialsHandl Registry: cfg.Registry, Database: cfg.Database, Logger: cfg.Logger, - TemporalClient: cfg.TemporalClient, + Temporal: cfg.Temporal, AuthConfig: cfg.AuthConfig, FedWikiAllowedDomains: cfg.FedWikiAllowedDomains, FedWikiSiteScheme: cfg.FedWikiSiteScheme, diff --git a/internal/integrations/fedwiki/web/requests_db_test.go b/internal/integrations/fedwiki/web/requests_db_test.go index aefd4269..d9da3a49 100644 --- a/internal/integrations/fedwiki/web/requests_db_test.go +++ b/internal/integrations/fedwiki/web/requests_db_test.go @@ -20,6 +20,7 @@ import ( "net/http" "net/http/httptest" "net/url" + "os" "regexp" "strconv" "strings" @@ -43,8 +44,17 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/authtest" "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest" "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/htmltest" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" ) +// publishedHolder returns a holder with c published, the shape the console +// has once Temporal is connected. +func publishedHolder(c client.Client) *clientholder.Holder { + h := clientholder.New() + h.Publish(c) + return h +} + // fakeTemporal records every workflow start and answers from a script. type fakeTemporal struct { client.Client @@ -133,13 +143,13 @@ func newSiteEnv(t *testing.T, limit int) *siteEnv { h.EntitlementsQ = entitlements.New(database) h.Registry = e.registry h.Logger = discardLogger() - h.TemporalClient = e.temporal + h.Temporal = publishedHolder(e.temporal) h.FedWikiAllowedDomains = []string{e.root} h.ConnectionID = func() string { return e.conn } e.h = h e.api = &FedWikiHandler{ SiteQ: fwmod.New(database), EntitlementsQ: entitlements.New(database), Registry: e.registry, - Database: database, Logger: discardLogger(), TemporalClient: e.temporal, + Database: database, Logger: discardLogger(), Temporal: publishedHolder(e.temporal), FedWikiAllowedDomains: []string{e.root}, ConnectionID: func() string { return e.conn }, } @@ -1644,3 +1654,43 @@ func TestPendingRowHasNoActions(t *testing.T) { } } } + +// TestSiteActionReadsTheConnectionPerRequest builds the partials handler once +// with an empty Temporal holder. An archive is answered with the service- +// unavailable sentence, records no request and starts nothing; after Publish +// the same handler, not rebuilt, records the request and starts the workflow. +// A handler that copied the client when it was built would never start one. +func TestSiteActionReadsTheConnectionPerRequest(t *testing.T) { + e := newSiteEnv(t, 5) + holder := clientholder.New() + h, err := NewFedWikiPartialsHandler(FedWikiPartialsConfig{ + SiteQ: e.h.SiteQ, EntitlementsQ: e.h.EntitlementsQ, Registry: e.registry, + Database: e.db, Logger: discardLogger(), Temporal: holder, + AuthConfig: e.h.AuthConfig, + FedWikiAllowedDomains: e.h.FedWikiAllowedDomains, + ConnectionID: e.h.ConnectionID, + TemplatesFS: os.DirFS("../templates"), + }) + if err != nil { + t.Fatalf("NewFedWikiPartialsHandler: %v", err) + } + id, domain := e.site("perrequest"+e.uniq, "active") + gen := e.nextGen(id) + + rec := e.rowAction(h.ArchiveSite, "archive", domain, id, gen) + if !strings.Contains(rec.Body.String(), msgServiceAvailable) { + t.Fatalf("archive with an empty holder = %d %q, want the service-unavailable sentence", rec.Code, rec.Body.String()) + } + if n := e.requestCount(id); n != 0 || len(e.temporal.starts()) != 0 { + t.Fatalf("archive with an empty holder recorded %d requests and started %v, want neither", n, e.temporal.starts()) + } + + holder.Publish(e.temporal) + rec = e.rowAction(h.ArchiveSite, "archive", domain, id, gen) + if !accepted(rec) { + t.Fatalf("archive after Publish = %d %q, want accepted", rec.Code, rec.Body.String()) + } + if got := e.temporal.starts(); len(got) != 1 || got[0] != workflows.SetStatusRequestWorkflowID(id, gen) { + t.Errorf("starts after Publish = %v, want the set-status workflow of the request", got) + } +} diff --git a/internal/integrations/fedwiki/web/site_actions.go b/internal/integrations/fedwiki/web/site_actions.go index f131b2fe..9e585cca 100644 --- a/internal/integrations/fedwiki/web/site_actions.go +++ b/internal/integrations/fedwiki/web/site_actions.go @@ -45,6 +45,8 @@ func (h *FedWikiPartialsHandler) CreateSite(w http.ResponseWriter, r *http.Reque http.Error(w, "Unauthorized", http.StatusUnauthorized) return } + // Read once: the connection may arrive after the routes are registered. + tc := h.Temporal.Client() values, errs := siteCreateForm.ParseWith(r, siteCreateOptions(h.FedWikiAllowedDomains)) if errs.Any() { @@ -106,7 +108,7 @@ func (h *FedWikiPartialsHandler) CreateSite(w http.ResponseWriter, r *http.Reque h.renderCreateFormRefusal(w, r, session.WorkspaceID, values, errs, false) return } - initiation, msg := startClaimVerification(r.Context(), h.Registry, h.TemporalClient, h.Logger, + initiation, msg := startClaimVerification(r.Context(), h.Registry, tc, h.Logger, session.WorkspaceID, domain, h.CustomDomainTarget) if msg != "" { errs.Form(msg) @@ -146,7 +148,7 @@ func (h *FedWikiPartialsHandler) CreateSite(w http.ResponseWriter, r *http.Reque } } - if h.TemporalClient == nil { + if tc == nil { h.Logger.Error("Temporal client not configured") errs.Form(msgServiceAvailable) h.renderCreateFormRefusal(w, r, session.WorkspaceID, values, errs, false) @@ -211,7 +213,7 @@ func (h *FedWikiPartialsHandler) CreateSite(w http.ResponseWriter, r *http.Reque return } - started := startRequestWorkflow(r.Context(), h.TemporalClient, h.Database, h.Logger, + started := startRequestWorkflow(r.Context(), tc, h.Database, h.Logger, workflowID, workflows.CreateFedWikiSiteRequestWorkflow, workflows.CreateSiteRequestInput{ ConnectionID: h.connectionID(), InstanceID: addr.Instance, @@ -316,7 +318,9 @@ type rowAction struct { } // beginRowAction runs the checks every row action shares, in design D31's -// order: the session, the posted address and the Temporal client; then the +// order: the session, the posted address and the Temporal client (a holder +// never empties once it holds a client, so the action reads it again where it +// starts its workflow and finds the one this check found); then the // request at the address, before the path's domain is resolved, so a repeat // that arrives after a permanent delete finished is answered from its // record; then the site, which must be the caller's and must be the posted @@ -338,7 +342,7 @@ func (h *FedWikiPartialsHandler) beginRowAction(w http.ResponseWriter, r *http.R h.siteActionConflict(w) return rowAction{}, false } - if h.TemporalClient == nil { + if h.Temporal.Client() == nil { h.Logger.Error("Temporal client not configured") h.siteActionResult(w, msgServiceAvailable) return rowAction{}, false @@ -520,7 +524,7 @@ func (h *FedWikiPartialsHandler) changeStatus(w http.ResponseWriter, r *http.Req return } - started := startRequestWorkflow(ctx, h.TemporalClient, h.Database, h.Logger, + started := startRequestWorkflow(ctx, h.Temporal.Client(), h.Database, h.Logger, workflowID, workflows.SetFedWikiSiteStatusRequestWorkflow, workflows.SetSiteStatusRequestInput{ ConnectionID: h.connectionID(), InstanceID: act.site.SiteID, @@ -648,7 +652,7 @@ func (h *FedWikiPartialsHandler) swapActiveSite(w http.ResponseWriter, r *http.R return } - started := startRequestWorkflow(ctx, h.TemporalClient, h.Database, h.Logger, + started := startRequestWorkflow(ctx, h.Temporal.Client(), h.Database, h.Logger, workflowID, workflows.SwapFedWikiActiveSiteRequestWorkflow, workflows.SwapSiteRequestInput{ ConnectionID: conn, InstanceID: act.site.SiteID, @@ -688,7 +692,7 @@ func (h *FedWikiPartialsHandler) DeleteSite(w http.ResponseWriter, r *http.Reque if !h.answerRecord(w, answer) { return } - started := startRequestWorkflow(ctx, h.TemporalClient, h.Database, h.Logger, + started := startRequestWorkflow(ctx, h.Temporal.Client(), h.Database, h.Logger, workflowID, workflows.DeleteFedWikiSiteRequestWorkflow, workflows.DeleteSiteRequestInput{ ConnectionID: h.connectionID(), InstanceID: act.site.SiteID, diff --git a/internal/integrations/stripe/stripe.go b/internal/integrations/stripe/stripe.go index 4737980d..2deb5846 100644 --- a/internal/integrations/stripe/stripe.go +++ b/internal/integrations/stripe/stripe.go @@ -68,6 +68,7 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/web" "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/workflows" "git.coopcloud.tech/wiki-cafe/member-console/internal/server" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/queues" "github.com/spf13/viper" stripego "github.com/stripe/stripe-go/v81" @@ -135,6 +136,9 @@ func (Adapter) ConfigSpec() []config.ConfigKey { // route is only mounted when a webhook secret is configured. The secret // is read directly from viper — Stripe's own concern, not threaded // through server.Deps. ConfigSpec above declares this key. +// +// The webhook handler's StartProcessing reads deps.Temporal when an event +// arrives; with no client it returns clientholder.ErrNotConnected (design D8). func (Adapter) RegisterRoutes(mux *http.ServeMux, deps server.Deps) error { webhookSecret := viper.GetString("stripe-webhook-secret") if webhookSecret == "" { @@ -153,13 +157,17 @@ func (Adapter) RegisterRoutes(mux *http.ServeMux, deps server.Deps) error { KeyMode: keyMode, Logger: deps.Logger, } - if deps.TemporalClient != nil { - temporalClient := deps.TemporalClient - handler.StartProcessing = func(ctx context.Context, e web.QueuedEvent) error { - return workflows.StartWebhookEvent(ctx, temporalClient, queues.Main, workflows.WebhookEvent{ - ID: e.ID, Provider: "stripe", ProviderEventID: e.ProviderEventID, EventType: e.EventType, - }) + // The holder is read when an event arrives, not when the route is built: + // the connection may come after the listener opens. With none, the event + // is recorded and the handler answers 503 (design D8). + handler.StartProcessing = func(ctx context.Context, e web.QueuedEvent) error { + tc := deps.Temporal.Client() + if tc == nil { + return clientholder.ErrNotConnected } + return workflows.StartWebhookEvent(ctx, tc, queues.Main, workflows.WebhookEvent{ + ID: e.ID, Provider: "stripe", ProviderEventID: e.ProviderEventID, EventType: e.EventType, + }) } mux.Handle("POST "+webhookPath, handler) return nil diff --git a/internal/integrations/stripe/web/webhook.go b/internal/integrations/stripe/web/webhook.go index 5bd36f4f..6baefde9 100644 --- a/internal/integrations/stripe/web/webhook.go +++ b/internal/integrations/stripe/web/webhook.go @@ -16,6 +16,7 @@ import ( "sync" internalstripe "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/store" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" stripe "github.com/stripe/stripe-go/v81" "github.com/stripe/stripe-go/v81/webhook" ) @@ -90,7 +91,10 @@ type StripeWebhookHandler struct { // StartProcessing starts the recorded event's workflow. It must be // idempotent for an execution already running, since a redelivery of an // unfinished event calls it again. Nil records events without starting - // anything, which only a handler wired without Temporal does. + // anything, which only a hand-built handler does. It returns + // an error that errors.Is clientholder.ErrNotConnected while the console + // has no Temporal connection; ServeHTTP answers 503 for that and keeps + // the row received. StartProcessing func(ctx context.Context, e QueuedEvent) error // versionsSeen holds the event API versions other than the SDK's this @@ -149,6 +153,13 @@ func eventEnvironment(livemode bool) string { // event being silently dropped. The redelivery of an unfinished duplicate // starts the workflow again, which is a no-op when it is already running. // +// While Temporal is not connected, StartProcessing returns +// clientholder.ErrNotConnected. The event is recorded and left `received`, +// and the handler answers 503 with Retry-After: 5, so Stripe redelivers it +// (the redelivery starts the workflow once Temporal is connected, and so +// does the boot sweep that runs once Temporal connects); any other start error answers +// 500. +// // An event whose environment disagrees with the key's mode is recorded // `refused` and answered 200 with no workflow started: the row is the // evidence the provider page counts, and Stripe stops redelivering an @@ -273,6 +284,13 @@ func (h *StripeWebhookHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) if h.StartProcessing != nil { if err := h.StartProcessing(r.Context(), QueuedEvent{ID: rowID, ProviderEventID: event.ID, EventType: string(event.Type)}); err != nil { + if errors.Is(err, clientholder.ErrNotConnected) { + h.Logger.Warn("stripe webhook event recorded before Temporal connected; answering 503 so Stripe redelivers it", + slog.String("event_id", event.ID), slog.String("event_type", string(event.Type))) + w.Header().Set("Retry-After", "5") + http.Error(w, clientholder.ErrNotConnected.Error(), http.StatusServiceUnavailable) + return + } h.Logger.Error("failed to start the webhook event's workflow; the row stays recorded", slog.String("event_id", event.ID), slog.Any("error", err)) // Recorded but not started: answer 500 so Stripe redelivers, diff --git a/internal/integrations/stripe/web/webhook_temporal_db_test.go b/internal/integrations/stripe/web/webhook_temporal_db_test.go new file mode 100644 index 00000000..1410db37 --- /dev/null +++ b/internal/integrations/stripe/web/webhook_temporal_db_test.go @@ -0,0 +1,176 @@ +// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial +// SPDX-FileCopyrightText: 2025-2026 Christian Galo + +package web_test + +import ( + "bytes" + "context" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/spf13/viper" + "github.com/stretchr/testify/mock" + "github.com/stripe/stripe-go/v81/webhook" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/mocks" + + "git.coopcloud.tech/wiki-cafe/member-console/internal/server" + "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" + + stripeintegration "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe" + stripewf "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/workflows" +) + +// logLine is one record of a captured logger: its level, message and the +// attributes it carries as text. +type logLine struct { + level slog.Level + msg string + attrs map[string]string +} + +// logCapture is a slog handler that keeps every record, so a test compares +// the level, the message and the named attributes and never formatted text. +type logCapture struct { + mu *sync.Mutex + lines *[]logLine +} + +func newLogCapture() logCapture { return logCapture{mu: new(sync.Mutex), lines: new([]logLine)} } + +func (logCapture) Enabled(context.Context, slog.Level) bool { return true } +func (c logCapture) WithAttrs([]slog.Attr) slog.Handler { return c } +func (c logCapture) WithGroup(string) slog.Handler { return c } + +func (c logCapture) Handle(_ context.Context, r slog.Record) error { + line := logLine{level: r.Level, msg: r.Message, attrs: map[string]string{}} + r.Attrs(func(a slog.Attr) bool { + line.attrs[a.Key] = a.Value.String() + return true + }) + c.mu.Lock() + defer c.mu.Unlock() + *c.lines = append(*c.lines, line) + return nil +} + +// find returns the captured lines with the given message. +func (c logCapture) find(msg string) []logLine { + c.mu.Lock() + defer c.mu.Unlock() + var out []logLine + for _, l := range *c.lines { + if l.msg == msg { + out = append(out, l) + } + } + return out +} + +// setViper sets a viper key for the test and restores what it held. +func setViper(t *testing.T, key string, value any) { + t.Helper() + prev, had := viper.Get(key), viper.IsSet(key) + viper.Set(key, value) + t.Cleanup(func() { + if had { + viper.Set(key, prev) + } else { + viper.Set(key, nil) + } + }) +} + +// signedPost sends a signed event through the mux the adapter registered. +func signedPost(mux *http.ServeMux, payload []byte) *httptest.ResponseRecorder { + now := time.Now() + sig := webhook.ComputeSignature(now, payload, testSecret) + req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe", bytes.NewReader(payload)) + req.Header.Set("Stripe-Signature", fmt.Sprintf("t=%d,v1=%x", now.Unix(), sig)) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + return rec +} + +// TestWebhookBeforeTemporalAnswers503 registers the webhook through the Stripe +// adapter, as the console does, with an empty Temporal holder. A signed event +// is recorded and left received, and answered 503 with Retry-After: 5 and the +// body Stripe shows in its delivery log; the redelivery after Publish answers +// 200 and starts the event's workflow, keyed by the event id. The handler the +// adapter built before Publish is the one that answers both, so a start +// function that read the holder at registration would start nothing. +func TestWebhookBeforeTemporalAnswers503(t *testing.T) { + database := dbtest.Open(t) + setViper(t, "stripe-webhook-secret", testSecret) + setViper(t, "stripe-api-key", "") + + logs := newLogCapture() + holder := clientholder.New() + mux := http.NewServeMux() + deps := server.Deps{Database: database, Logger: slog.New(logs), Temporal: holder} + if err := (stripeintegration.Adapter{}).RegisterRoutes(mux, deps); err != nil { + t.Fatalf("RegisterRoutes: %v", err) + } + + eventID := "evt_before_temporal_" + uuid.New().String()[:12] + t.Cleanup(func() { + _, _ = database.Exec(`DELETE FROM core.webhook_events WHERE provider = 'stripe' AND provider_event_id = $1`, eventID) + }) + payload := []byte(fmt.Sprintf(`{"id": %q, "type": "customer.created", "created": 1757400000, "data": {"object": {"id": "cus_bt"}}}`, eventID)) + + rec := signedPost(mux, payload) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("delivery with an empty holder = %d, want 503", rec.Code) + } + if got := rec.Header().Get("Retry-After"); got != "5" { + t.Errorf("Retry-After = %q, want 5", got) + } + if got := strings.TrimSpace(rec.Body.String()); got != "Temporal is not connected" { + t.Errorf("body = %q, want %q", got, "Temporal is not connected") + } + var status string + if err := database.QueryRow(`SELECT status FROM core.webhook_events WHERE provider = 'stripe' AND provider_event_id = $1`, eventID).Scan(&status); err != nil { + t.Fatalf("the event was not recorded: %v", err) + } + if status != "received" { + t.Errorf("row status = %q, want received", status) + } + assertNotConnectedLine(t, logs, eventID) + + fake := &mocks.Client{} + fake.On("ExecuteWorkflow", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&mocks.WorkflowRun{}, nil) + holder.Publish(fake) + + if rec := signedPost(mux, payload); rec.Code != http.StatusOK { + t.Fatalf("redelivery after Publish = %d, want 200", rec.Code) + } + fake.AssertNumberOfCalls(t, "ExecuteWorkflow", 1) + opts, ok := fake.Calls[0].Arguments.Get(1).(client.StartWorkflowOptions) + if !ok || opts.ID != stripewf.WebhookEventWorkflowID(eventID) { + t.Errorf("start options = %+v, want the workflow keyed by event id %s", fake.Calls[0].Arguments.Get(1), eventID) + } +} + +// assertNotConnectedLine checks the Warn line the handler writes for an event +// recorded before Temporal connected. +func assertNotConnectedLine(t *testing.T, logs logCapture, eventID string) { + t.Helper() + const msg = "stripe webhook event recorded before Temporal connected; answering 503 so Stripe redelivers it" + lines := logs.find(msg) + if len(lines) != 1 { + t.Fatalf("%d lines with message %q, want 1", len(lines), msg) + } + l := lines[0] + if l.level != slog.LevelWarn || l.attrs["event_id"] != eventID || l.attrs["event_type"] != "customer.created" { + t.Errorf("line = level %v attrs %v, want Warn with event_id %s and event_type customer.created", l.level, l.attrs, eventID) + } +} diff --git a/internal/server/bounded_grant_temporal_db_test.go b/internal/server/bounded_grant_temporal_db_test.go new file mode 100644 index 00000000..7ce0c537 --- /dev/null +++ b/internal/server/bounded_grant_temporal_db_test.go @@ -0,0 +1,244 @@ +// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial +// SPDX-FileCopyrightText: 2025-2026 Christian Galo + +package server + +import ( + "database/sql" + "net/http" + "net/url" + "strings" + "testing" + + "github.com/google/uuid" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/mocks" + + "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" +) + +// needsSchedulerMessage is the end date's field error while Temporal is not +// connected (the design's Copy table, row 6). +const needsSchedulerMessage = "An end date needs the expiration scheduler, which is not connected." + +// boundedGrantWorld is one operator, one product and the handler whose Temporal +// holder a row sets. +type boundedGrantWorld struct { + t *testing.T + db *sql.DB + f glFixture + h *glHarness + fake *mocks.Client +} + +func newBoundedGrantWorld(t *testing.T) *boundedGrantWorld { + t.Helper() + database := dbtest.Open(t) + f := newGLFixture(t, database) + return &boundedGrantWorld{t: t, db: database, f: f, h: newGLHarness(t, database, f.operatorID)} +} + +// disconnect gives the handler an empty holder, the state before Temporal +// connects. +func (w *boundedGrantWorld) disconnect() { w.h.handler.Temporal = clientholder.New() } + +// connect gives the handler a published holder over a fake that accepts every +// workflow start and records them. +func (w *boundedGrantWorld) connect() { + w.fake = succeedingTemporal() + w.h.handler.Temporal = holderOver(client.Client(w.fake)) +} + +func (w *boundedGrantWorld) org(label string) string { + w.t.Helper() + return glOrg(w.t, w.db, w.f, label+" "+uuid.New().String()[:8]) +} + +func (w *boundedGrantWorld) grantCount(org string) int { + w.t.Helper() + var n int + if err := w.db.QueryRow(`SELECT count(*) FROM core.grants WHERE granted_to_org_id = $1`, org).Scan(&n); err != nil { + w.t.Fatalf("count grants: %v", err) + } + return n +} + +// issue posts the issuance form for the fixture's product, with the end date +// when validUntil is not empty. +func (w *boundedGrantWorld) issue(org, validUntil string) (int, string, string) { + w.t.Helper() + form := url.Values{"product_id": {w.f.prodA}, "reason": {"manual"}, "quantity": {"1"}} + if validUntil != "" { + form.Set("valid_until", validUntil) + } + return w.h.postOrg(w.h.handler.IssueGrant, org, form) +} + +// grantID reads the id of the org's only grant. +func (w *boundedGrantWorld) grantID(org string) string { + w.t.Helper() + var id string + if err := w.db.QueryRow(`SELECT grant_id FROM core.grants WHERE granted_to_org_id = $1`, org).Scan(&id); err != nil { + w.t.Fatalf("read the org's grant: %v", err) + } + return id +} + +// headBound reads the end date of the org's newest grant, empty when open. +func (w *boundedGrantWorld) headBound(org string) string { + w.t.Helper() + var bound sql.NullString + if err := w.db.QueryRow(`SELECT valid_until::text FROM core.grants WHERE granted_to_org_id = $1 ORDER BY created_at DESC, grant_id DESC LIMIT 1`, org).Scan(&bound); err != nil { + w.t.Fatalf("read the end date: %v", err) + } + return bound.String +} + +func checkIssuanceWithBoundRefused(t *testing.T) { + w := newBoundedGrantWorld(t) + org := w.org("BG refused") + w.disconnect() + code, body, _ := w.issue(org, laterBound(30)) + if code != http.StatusUnprocessableEntity || !strings.Contains(body, needsSchedulerMessage) { + t.Errorf("issuance with an end date and an empty holder: status=%d, want 422 with %q", code, needsSchedulerMessage) + } + errID := `id="form-operator.enrollment.grant.issue-valid_until-error"` + if i := strings.Index(body, errID); i < 0 || !strings.Contains(body[i:min(i+300, len(body))], needsSchedulerMessage) { + t.Errorf("the refusal is not under the end date field: no %s carrying %q", errID, needsSchedulerMessage) + } + if n := w.grantCount(org); n != 0 { + t.Errorf("a refused issuance wrote %d grants", n) + } +} + +func checkIssuancePastBoundKeepsItsError(t *testing.T) { + w := newBoundedGrantWorld(t) + org := w.org("BG past") + w.disconnect() + code, body, _ := w.issue(org, "2020-01-01T00:00") + if code != http.StatusUnprocessableEntity || !strings.Contains(body, "Choose a time in the future.") { + t.Errorf("a past end date with an empty holder: status=%d, want 422 with the past-date error", code) + } + if strings.Contains(body, needsSchedulerMessage) { + t.Errorf("a past end date with an empty holder: the body carries %q, want the past-date error alone", needsSchedulerMessage) + } + if n := w.grantCount(org); n != 0 { + t.Errorf("a past end date with an empty holder: the refused issuance wrote %d grants", n) + } +} + +func checkIssuanceWithoutBoundIssues(t *testing.T) { + w := newBoundedGrantWorld(t) + org := w.org("BG open") + w.disconnect() + code, body, trigger := w.issue(org, "") + if code != http.StatusOK || !strings.Contains(trigger, "issued") { + t.Fatalf("issuance without an end date and an empty holder: status=%d trigger=%q body=%.300s", code, trigger, body) + } + if n := w.grantCount(org); n != 1 { + t.Errorf("grants after the issuance = %d, want 1", n) + } +} + +func checkManagementWithNewBoundRefused(t *testing.T) { + w := newBoundedGrantWorld(t) + org := w.org("BG manage") + w.disconnect() + if code, body, trigger := w.issue(org, ""); code != http.StatusOK || !strings.Contains(trigger, "issued") { + t.Fatalf("issue the open grant: status=%d trigger=%q body=%.300s", code, trigger, body) + } + grant := w.grantID(org) + var ( + code int + body string + ) + func() { + // A handler that reaches the scheduler without a client panics after + // the write; the row reports the write below. + defer func() { + if p := recover(); p != nil { + t.Errorf("ManageGrant panicked: %v", p) + } + }() + code, body, _ = w.h.postOrgGrant(w.h.handler.ManageGrant, org, grant, manageRow(grant, url.Values{"valid_until": {laterBound(30)}})) + }() + if code != http.StatusUnprocessableEntity || !strings.Contains(editingRow(body, grant), needsSchedulerMessage) { + t.Errorf("a new end date with an empty holder: status=%d, want the row editing with %q", code, needsSchedulerMessage) + } + if n, bound := w.grantCount(org), w.headBound(org); n != 1 || bound != "" { + t.Errorf("a refused save left %d grants and an end date of %q, want the grant unchanged", n, bound) + } +} + +func checkManagementPastBoundKeepsItsError(t *testing.T) { + w := newBoundedGrantWorld(t) + org := w.org("BG manage past") + w.disconnect() + if code, body, trigger := w.issue(org, ""); code != http.StatusOK || !strings.Contains(trigger, "issued") { + t.Fatalf("issue the open grant: status=%d trigger=%q body=%.300s", code, trigger, body) + } + grant := w.grantID(org) + code, body, _ := w.h.postOrgGrant(w.h.handler.ManageGrant, org, grant, manageRow(grant, url.Values{"valid_until": {"2020-01-01T00:00"}})) + row := editingRow(body, grant) + if code != http.StatusUnprocessableEntity || !strings.Contains(row, "Choose a time in the future.") { + t.Errorf("a past end date with an empty holder: status=%d, want the row editing with the past-date error", code) + } + if strings.Contains(row, needsSchedulerMessage) { + t.Errorf("a past end date with an empty holder: the row carries %q, want the past-date error alone", needsSchedulerMessage) + } +} + +func checkManagementClearingBoundSaved(t *testing.T) { + w := newBoundedGrantWorld(t) + org := w.org("BG clear") + w.connect() + if code, body, trigger := w.issue(org, laterBound(30)); code != http.StatusOK || !strings.Contains(trigger, "issued") { + t.Fatalf("issue the bounded grant: status=%d trigger=%q body=%.300s", code, trigger, body) + } + grant := w.grantID(org) + w.disconnect() + code, body, trigger := w.h.postOrgGrant(w.h.handler.ManageGrant, org, grant, manageRow(grant, url.Values{"valid_until": {""}})) + if code != http.StatusOK || !strings.Contains(trigger, "Grant saved.") { + t.Fatalf("clearing the end date with an empty holder: status=%d trigger=%q body=%.300s", code, trigger, body) + } + if bound := w.headBound(org); bound != "" { + t.Errorf("the end date after clearing = %q, want none", bound) + } +} + +func checkIssuanceWithBoundStartsExpiration(t *testing.T) { + w := newBoundedGrantWorld(t) + org := w.org("BG published") + w.connect() + code, body, trigger := w.issue(org, laterBound(30)) + if code != http.StatusOK || !strings.Contains(trigger, "issued") { + t.Fatalf("issuance with an end date and a published holder: status=%d trigger=%q body=%.300s", code, trigger, body) + } + w.fake.AssertNumberOfCalls(t, "ExecuteWorkflow", 1) + opts, ok := w.fake.Calls[0].Arguments.Get(1).(client.StartWorkflowOptions) + if !ok || opts.ID != "grant-expiration-"+w.grantID(org) { + t.Errorf("start options = %+v, want the expiration workflow of the issued grant", w.fake.Calls[0].Arguments.Get(1)) + } +} + +// TestBoundedGrantNeedsTemporal: the expiration workflow is the only thing +// that ends a bounded grant, so a bound is refused before anything is written +// while Temporal is not connected, and saving without a bound or clearing one +// needs nothing from it. +func TestBoundedGrantNeedsTemporal(t *testing.T) { + for _, row := range []struct { + name string + check func(*testing.T) + }{ + {"issuance with an end date and an empty holder is refused", checkIssuanceWithBoundRefused}, + {"issuance without an end date issues", checkIssuanceWithoutBoundIssues}, + {"a management save that sets a new end date is refused", checkManagementWithNewBoundRefused}, + {"a past end date keeps its own error on issuance", checkIssuancePastBoundKeepsItsError}, + {"a past end date keeps its own error on a management save", checkManagementPastBoundKeepsItsError}, + {"a management save that clears the end date is saved", checkManagementClearingBoundSaved}, + {"issuance with an end date and a published fake starts the run", checkIssuanceWithBoundStartsExpiration}, + } { + t.Run(row.name, row.check) + } +} diff --git a/internal/server/grant_resumption_operator_test.go b/internal/server/grant_resumption_operator_test.go index faf27b71..9182e5f7 100644 --- a/internal/server/grant_resumption_operator_test.go +++ b/internal/server/grant_resumption_operator_test.go @@ -32,7 +32,6 @@ import ( "time" "github.com/google/uuid" - "github.com/stretchr/testify/mock" "go.temporal.io/sdk/client" "go.temporal.io/sdk/mocks" @@ -709,10 +708,12 @@ func laterBound(days int) string { // production; the returned client records the starts. func (w *grWorld) temporal() *mocks.Client { w.t.Helper() - c := &mocks.Client{} - c.On("ExecuteWorkflow", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&mocks.WorkflowRun{}, nil) - w.h.handler.TemporalClient = c - w.t.Cleanup(func() { w.h.handler.TemporalClient = nil }) + c := succeedingTemporal() + // A holder takes one client for good, so the handler gets a new holder and + // gets the harness's own back when the test ends. + harnessHolder := w.h.handler.Temporal + w.h.handler.Temporal = holderOver(c) + w.t.Cleanup(func() { w.h.handler.Temporal = harnessHolder }) return c } diff --git a/internal/server/member_domains.go b/internal/server/member_domains.go index 45345d59..1a341852 100644 --- a/internal/server/member_domains.go +++ b/internal/server/member_domains.go @@ -31,6 +31,7 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/domains" "git.coopcloud.tech/wiki-cafe/member-console/internal/entitlements" "git.coopcloud.tech/wiki-cafe/member-console/internal/web" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" wfdomains "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/domains" "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/queues" "go.temporal.io/sdk/client" @@ -45,10 +46,12 @@ type MemberDomainsHandler struct { // Registry is the allocation API: every claim read and write on this // surface goes through it, so the ownership and disjointness rules are // enforced in exactly one place. - Registry *domains.Registry - TemporalClient client.Client // nil if Temporal isn't configured - AuthConfig *auth.Config - Logger *slog.Logger + Registry *domains.Registry + // Temporal holds the connection to Temporal. A request reads Client once; + // it is nil when Temporal is not configured or not connected yet. + Temporal *clientholder.Holder + AuthConfig *auth.Config + Logger *slog.Logger // ConnectTarget is the DNS target members point external domains at. // Empty disables external claims deployment-wide (design D4/D7). Still // held here for rendering the connect record's value; the gate below is @@ -63,13 +66,13 @@ type MemberDomainsHandler struct { // MemberDomainsConfig holds configuration for the member domains handler. type MemberDomainsConfig struct { - DomainsQ domains.Querier - Registry *domains.Registry - EntitlementsQ entitlements.Querier - TemporalClient client.Client - AuthConfig *auth.Config - Logger *slog.Logger - ConnectTarget string + DomainsQ domains.Querier + Registry *domains.Registry + EntitlementsQ entitlements.Querier + Temporal *clientholder.Holder + AuthConfig *auth.Config + Logger *slog.Logger + ConnectTarget string } // NewMemberDomainsHandler creates a new MemberDomainsHandler. @@ -80,14 +83,14 @@ func NewMemberDomainsHandler(cfg MemberDomainsConfig) (*MemberDomainsHandler, er } return &MemberDomainsHandler{ - DomainsQ: cfg.DomainsQ, - Registry: cfg.Registry, - TemporalClient: cfg.TemporalClient, - AuthConfig: cfg.AuthConfig, - Logger: cfg.Logger, - ConnectTarget: strings.TrimSpace(cfg.ConnectTarget), - Templates: NewSafeTemplates(tmpl, cfg.Logger), - gate: NewExternalClaimGate(cfg.EntitlementsQ, cfg.ConnectTarget), + DomainsQ: cfg.DomainsQ, + Registry: cfg.Registry, + Temporal: cfg.Temporal, + AuthConfig: cfg.AuthConfig, + Logger: cfg.Logger, + ConnectTarget: strings.TrimSpace(cfg.ConnectTarget), + Templates: NewSafeTemplates(tmpl, cfg.Logger), + gate: NewExternalClaimGate(cfg.EntitlementsQ, cfg.ConnectTarget), }, nil } @@ -427,7 +430,8 @@ func (h *MemberDomainsHandler) renderClaimStatus(w http.ResponseWriter, r *http. // frees the name again instead of parking it for the whole window. Returns // the claim, or a member-facing rejection message. func (h *MemberDomainsHandler) startClaimVerification(ctx context.Context, workspaceID, root string) (*domains.Claim, string) { - if h.TemporalClient == nil { + tc := h.Temporal.Client() + if tc == nil { h.Logger.Error("Temporal client not configured") return nil, "Domain verification is not available right now." } @@ -449,7 +453,7 @@ func (h *MemberDomainsHandler) startClaimVerification(ctx context.Context, works ID: wfdomains.VerifyClaimWorkflowIDPrefix + claim.ClaimID, TaskQueue: queues.Main, } - if _, err := h.TemporalClient.ExecuteWorkflow(ctx, options, wfdomains.VerifyClaimWorkflow, input); err != nil { + if _, err := tc.ExecuteWorkflow(ctx, options, wfdomains.VerifyClaimWorkflow, input); err != nil { h.Logger.Error("failed to start claim verification workflow", slog.Any("error", err)) // Free the name: a pending claim nothing is polling for would hold it // for the full window. `canceled`, not `expired` — the window did not @@ -507,8 +511,8 @@ func (h *MemberDomainsHandler) CheckClaimNow(w http.ResponseWriter, r *http.Requ if !ok { return } - if claim.Status == domains.StatusPending && h.TemporalClient != nil { - if err := h.TemporalClient.SignalWorkflow(r.Context(), + if tc := h.Temporal.Client(); claim.Status == domains.StatusPending && tc != nil { + if err := tc.SignalWorkflow(r.Context(), wfdomains.VerifyClaimWorkflowIDPrefix+claim.ClaimID, "", wfdomains.CheckNowSignal, nil); err != nil { h.Logger.Warn("failed to signal check-now", slog.Any("error", err)) @@ -535,8 +539,8 @@ func (h *MemberDomainsHandler) CancelClaim(w http.ResponseWriter, r *http.Reques return } if canceled { - if h.TemporalClient != nil { - if cErr := h.TemporalClient.CancelWorkflow(r.Context(), + if tc := h.Temporal.Client(); tc != nil { + if cErr := tc.CancelWorkflow(r.Context(), wfdomains.VerifyClaimWorkflowIDPrefix+claim.ClaimID, ""); cErr != nil { // The claim is already terminal; an orphaned poller records // nothing (probe writes are pending-guarded) and expires on diff --git a/internal/server/member_domains_rollback_db_test.go b/internal/server/member_domains_rollback_db_test.go index 1ca5d1ac..f497ba7a 100644 --- a/internal/server/member_domains_rollback_db_test.go +++ b/internal/server/member_domains_rollback_db_test.go @@ -78,9 +78,9 @@ func TestStartClaimVerificationRollbackNeverDebitsTheLedger(t *testing.T) { down.On("ExecuteWorkflow", mock.Anything, mock.Anything, mock.Anything, mock.Anything). Return(nil, errors.New("temporal unavailable")) h := &MemberDomainsHandler{ - Registry: domains.NewRegistry(database), - TemporalClient: down, - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + Registry: domains.NewRegistry(database), + Temporal: holderOver(down), + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), } for i := 0; i <= domains.DefaultInitiationBudget; i++ { @@ -104,7 +104,7 @@ func TestStartClaimVerificationRollbackNeverDebitsTheLedger(t *testing.T) { up := &mocks.Client{} up.On("ExecuteWorkflow", mock.Anything, mock.Anything, mock.Anything, mock.Anything). Return(nil, nil) - h.TemporalClient = up + h.Temporal = holderOver(up) claim, message := h.startClaimVerification(ctx, workspaceID, name) if claim == nil { t.Fatalf("claim after the outage refused with %q, want success", message) diff --git a/internal/server/new_handler_test.go b/internal/server/new_handler_test.go index 7a8ecf70..306cbe9f 100644 --- a/internal/server/new_handler_test.go +++ b/internal/server/new_handler_test.go @@ -51,6 +51,7 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/authtest" "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/golden" "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/htmltest" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" ) const ( @@ -398,11 +399,13 @@ func TestNewHandlerRouteTable(t *testing.T) { // TestNewHandlerRouteTableNotReady is the same table over the console's own // authentication configuration before the identity provider's discovery -// document is published (auth.NewAwaitingDiscovery). Its diff from -// routes.golden is the list of routes that answer differently: the sign-in -// rows. +// document is published (auth.NewAwaitingDiscovery) and before Temporal +// connects (an empty clientholder.Holder). Both tables run without a Temporal +// client (routeConfig sets no holder), so its diff from routes.golden is the +// sign-in rows. func TestNewHandlerRouteTableNotReady(t *testing.T) { - env := newRouteEnvOver(t, nil, auth.NewAwaitingDiscovery(authtest.NewSessionManager())) + env := newRouteEnvOver(t, func(c *Config) { c.Temporal = clientholder.New() }, + auth.NewAwaitingDiscovery(authtest.NewSessionManager())) golden.Compare(t, "testdata/routes-not-ready.golden", routeTable(t, env)) } @@ -497,7 +500,7 @@ type sentinelTemporal struct{ client.Client } // the auth configuration and an Include that dispatches through the router. func TestNewHandlerPassesDepsToMounts(t *testing.T) { env := newRouteEnv(t, func(c *Config) { - c.Database, c.TemporalClient = new(sql.DB), &sentinelTemporal{} + c.Database, c.Temporal = new(sql.DB), holderOver(&sentinelTemporal{}) c.DomainsPolicy = domains.Policy{PendingCap: 7} c.DomainsConnectTarget = "connect.test" }) @@ -508,7 +511,7 @@ func TestNewHandlerPassesDepsToMounts(t *testing.T) { "OrgQ": d.OrgQ == c.OrgQ, "EntitlementsQ": d.EntitlementsQ == c.EntitlementsQ, "BillingQ": d.BillingQ == c.BillingQ, - "TemporalClient": d.TemporalClient == c.TemporalClient, + "Temporal": d.Temporal == c.Temporal, "AuthConfig": d.AuthConfig == env.auth, "Logger": d.Logger == c.Logger, "DomainsPolicy": d.DomainsPolicy == c.DomainsPolicy, diff --git a/internal/server/no_handler_holds_temporal_client_test.go b/internal/server/no_handler_holds_temporal_client_test.go new file mode 100644 index 00000000..3dbd3781 --- /dev/null +++ b/internal/server/no_handler_holds_temporal_client_test.go @@ -0,0 +1,193 @@ +// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial +// SPDX-FileCopyrightText: 2025-2026 Christian Galo + +package server + +import ( + "go/ast" + "go/parser" + "go/token" + "os" + "path/filepath" + "sort" + "strconv" + "testing" +) + +const temporalClientImport = "go.temporal.io/sdk/client" + +// clientField is a struct field whose type is the Temporal SDK's Client. +type clientField struct { + file string + line int + owner string + name string +} + +// temporalClientName returns the name file imports the SDK's client package +// under, or "" when it does not import it. An import with no name is known to +// the compiler by the last element of its path, which is `client`. +func temporalClientName(file *ast.File) string { + for _, spec := range file.Imports { + path, err := strconv.Unquote(spec.Path.Value) + if err != nil || path != temporalClientImport { + continue + } + if spec.Name != nil { + return spec.Name.Name + } + return "client" + } + return "" +} + +// isClientType reports whether expr is `.Client`, or a bare `Client` +// when the file dot-imports the package. +func isClientType(expr ast.Expr, pkg string) bool { + if id, ok := expr.(*ast.Ident); ok { + return pkg == "." && id.Name == "Client" + } + sel, ok := expr.(*ast.SelectorExpr) + if !ok || sel.Sel.Name != "Client" { + return false + } + id, ok := sel.X.(*ast.Ident) + return ok && id.Name == pkg +} + +// fieldNames names a field for a failure message; an embedded field has no +// name of its own. +func fieldNames(f *ast.Field) string { + if len(f.Names) == 0 { + return "(embedded)" + } + return f.Names[0].Name +} + +// clientFieldsIn parses one file's source and returns every struct field +// whose type is the Temporal SDK's Client under whatever name the file +// imports the package by. The scan works on source text so that a test can +// run it on a source it writes itself. +func clientFieldsIn(filename string, src []byte) ([]clientField, error) { + fset := token.NewFileSet() + file, err := parser.ParseFile(fset, filename, src, 0) + if err != nil { + return nil, err + } + pkg := temporalClientName(file) + if pkg == "" || pkg == "_" { + return nil, nil + } + var found []clientField + ast.Inspect(file, func(n ast.Node) bool { + spec, ok := n.(*ast.TypeSpec) + if !ok { + return true + } + st, ok := spec.Type.(*ast.StructType) + if !ok { + return true + } + for _, f := range st.Fields.List { + if isClientType(f.Type, pkg) { + found = append(found, clientField{ + file: filename, line: fset.Position(f.Pos()).Line, + owner: spec.Name.Name, name: fieldNames(f), + }) + } + } + return true + }) + return found, nil +} + +// clientFieldsInDir scans the non-test files of one directory. +func clientFieldsInDir(t *testing.T, dir string) []clientField { + t.Helper() + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("read %s: %v", dir, err) + } + var found []clientField + for _, e := range entries { + name := e.Name() + if e.IsDir() || filepath.Ext(name) != ".go" || len(name) > 8 && name[len(name)-8:] == "_test.go" { + continue + } + path := filepath.Join(dir, name) + src, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + fields, err := clientFieldsIn(path, src) + if err != nil { + t.Fatalf("parse %s: %v", path, err) + } + found = append(found, fields...) + } + return found +} + +// handlerPackageDirs are the packages whose structs must not hold a client: +// this one, every integration's web package, and the three adapters' own +// packages. The workflows packages' schedule managers receive the +// client from start and the startup hooks, and are outside the scan on +// purpose: they are built after the connection exists and serve no request. +func handlerPackageDirs(t *testing.T) []string { + t.Helper() + dirs := []string{"."} + web, err := filepath.Glob(filepath.Join("..", "integrations", "*", "web")) + if err != nil { + t.Fatal(err) + } + dirs = append(dirs, web...) + for _, name := range []string{"fedwiki", "stripe", "discourse"} { + dirs = append(dirs, filepath.Join("..", "integrations", name)) + } + sort.Strings(dirs) + return dirs +} + +// TestNoHandlerHoldsATemporalClient refuses a struct field of type +// client.Client in the packages that build handlers. A handler that copied the +// client when its routes were registered would never see one that arrives +// later (design D7); it holds a *clientholder.Holder and reads Client when a +// request arrives. +func TestNoHandlerHoldsATemporalClient(t *testing.T) { + for _, dir := range handlerPackageDirs(t) { + for _, f := range clientFieldsInDir(t, dir) { + t.Errorf("%s:%d: %s.%s has type client.Client from %s; hold a *clientholder.Holder and read Client per request", + f.file, f.line, f.owner, f.name, temporalClientImport) + } + } +} + +// TestClientFieldScanFindsEveryImportName runs the scan on sources the test +// writes: the package imported unaliased, imported as temporalsdk (the name +// the FedWiki and Discourse operator handlers used), and a source with no +// such field. +func TestClientFieldScanFindsEveryImportName(t *testing.T) { + cases := []struct { + name string + src string + want int + }{ + {"unaliased", "package p\nimport \"go.temporal.io/sdk/client\"\ntype H struct{ C client.Client }\n", 1}, + {"aliased as temporalsdk", "package p\nimport temporalsdk \"go.temporal.io/sdk/client\"\ntype H struct{ C temporalsdk.Client }\n", 1}, + {"dot import", "package p\nimport . \"go.temporal.io/sdk/client\"\ntype H struct{ C Client }\n", 1}, + {"embedded", "package p\nimport \"go.temporal.io/sdk/client\"\ntype H struct{ client.Client }\n", 1}, + {"another package named client", "package p\nimport \"example.test/client\"\ntype H struct{ C client.Client }\n", 0}, + {"a parameter, not a field", "package p\nimport \"go.temporal.io/sdk/client\"\nfunc f(c client.Client) {}\n", 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := clientFieldsIn("p.go", []byte(tc.src)) + if err != nil { + t.Fatal(err) + } + if len(got) != tc.want { + t.Errorf("found %d client fields %v, want %d", len(got), got, tc.want) + } + }) + } +} diff --git a/internal/server/operator_domains.go b/internal/server/operator_domains.go index 855fd2e2..2cf03c7f 100644 --- a/internal/server/operator_domains.go +++ b/internal/server/operator_domains.go @@ -440,8 +440,8 @@ func (h *OperatorPartialsHandler) ForceReleaseClaim(w http.ResponseWriter, r *ht return } - if claim.Status == domains.StatusPending && h.TemporalClient != nil { - if cErr := h.TemporalClient.CancelWorkflow(ctx, + if tc := h.Temporal.Client(); claim.Status == domains.StatusPending && tc != nil { + if cErr := tc.CancelWorkflow(ctx, wfdomains.VerifyClaimWorkflowIDPrefix+claimID, ""); cErr != nil { h.Logger.Warn("failed to cancel verification workflow after force-release", slog.String("claim_id", claimID), slog.Any("error", cErr)) diff --git a/internal/server/operator_enrollment.go b/internal/server/operator_enrollment.go index 54ace5c7..9dad488f 100644 --- a/internal/server/operator_enrollment.go +++ b/internal/server/operator_enrollment.go @@ -657,11 +657,20 @@ func (h *OperatorPartialsHandler) IssueGrant(w http.ResponseWriter, r *http.Requ } // valid_until is optional and allowed on ANY product now (Doc 41 uniform // bounds), not only trials. + // + // The expiration workflow is the only thing that ends a bounded grant, so a + // bound is refused before the grant is written while Temporal is not + // connected (D15). The refusal follows the parse, so a past or malformed + // date keeps its own field error. + tc := h.Temporal.Client() var validUntil sql.NullTime if vu, msg := parseGrantValidUntilValue(values.String("valid_until"), r.PostForm.Get("valid_until_offset")); msg != "" { errs.Field("valid_until", msg) } else { validUntil = vu + if vu.Valid && tc == nil { + errs.Field("valid_until", boundNeedsTemporalMessage) + } } if errs.Any() { h.renderIssueGrantRefusal(w, r, orgID, values, errs) @@ -703,17 +712,11 @@ func (h *OperatorPartialsHandler) IssueGrant(w http.ResponseWriter, r *http.Requ // expiry). GrantExpirationWorkflow is the sole enforcement (no sweep), so a // failed schedule must not report plain success (finding #49). if validUntil.Valid { - if h.TemporalClient == nil { - h.Logger.Warn("no Temporal client configured; grant expiration was not scheduled", slog.String("grant_id", result.Grant.GrantID)) - h.renderOrgEnrollmentPage(w, r, orgID, "", - "Grant issued, but expiration scheduling FAILED (no scheduler); it will not auto-expire. Revoke it manually to end it.") - return - } workflowOptions := client.StartWorkflowOptions{ ID: "grant-expiration-" + result.Grant.GrantID, TaskQueue: queues.Main, } - if _, err := h.TemporalClient.ExecuteWorkflow(r.Context(), workflowOptions, wf.GrantExpirationWorkflow, wf.GrantExpirationInput{ + if _, err := tc.ExecuteWorkflow(r.Context(), workflowOptions, wf.GrantExpirationWorkflow, wf.GrantExpirationInput{ GrantID: result.Grant.GrantID, ValidUntil: validUntil.Time, }); err != nil { @@ -764,6 +767,11 @@ type validUntilRefusal struct{ message string } func (e *validUntilRefusal) Error() string { return "valid until refused: " + e.message } +// boundNeedsTemporalMessage is the end date's field error while Temporal is +// not connected: the expiration workflow is the only thing that ends a +// bounded grant. +const boundNeedsTemporalMessage = "An end date needs the expiration scheduler, which is not connected." + // logBound renders a grant's bound for the Manage log line: RFC 3339, or // "none" for an open-ended grant. func logBound(bound sql.NullTime) string { @@ -773,6 +781,28 @@ func logBound(bound sql.NullTime) string { return bound.Time.Format(time.RFC3339) } +// manageGrantValidUntil is ManageGrantInput.ValidUntil for a Save that +// submitted bound: it compares the submission with the stored bound, parses a +// changed one as issuance does, and refuses a new bound while tc, the client +// the request read, is nil. A new bound is enforced only by the expiration +// workflow, so it is refused before the write while Temporal is not connected +// (D15); clearing the bound starts no workflow and is saved. +func manageGrantValidUntil(submitted string, tc client.Client) func(stored sql.NullTime) (bool, sql.NullTime, error) { + return func(stored sql.NullTime) (bool, sql.NullTime, error) { + if submitted == grantValidUntilFieldValue(stored) { + return false, stored, nil + } + bound, msg := parseGrantValidUntilValue(submitted, "") + if msg != "" { + return true, sql.NullTime{}, &validUntilRefusal{message: msg} + } + if bound.Valid && tc == nil { + return true, sql.NullTime{}, &validUntilRefusal{message: boundNeedsTemporalMessage} + } + return true, bound, nil + } +} + // ManageGrant handles POST /partials/operator/organizations/{orgID}/grants/{grantID}/manage // — the one Manage action (grant-resumption D15), which replaced Edit and // Extend: the Save of the grant's editing row. It saves what the row submits for the head of the named @@ -831,6 +861,7 @@ func (h *OperatorPartialsHandler) ManageGrant(w http.ResponseWriter, r *http.Req submittedBound := values.String("valid_until") description := values.String("description") resumes := values.Bool("resumes_after_supersession") + tc := h.Temporal.Client() // Manage takes the pool lock every materializer takes, so it opens its // transaction the way they do: the shared rendezvous first, and a @@ -848,18 +879,9 @@ func (h *OperatorPartialsHandler) ManageGrant(w http.ResponseWriter, r *http.Req GrantID: grantID, GrantedByPersonID: session.PersonID, ActorID: uuid.NullUUID{UUID: uuid.MustParse(session.PersonID), Valid: true}, - ValidUntil: func(stored sql.NullTime) (bool, sql.NullTime, error) { - if submittedBound == grantValidUntilFieldValue(stored) { - return false, stored, nil - } - bound, msg := parseGrantValidUntilValue(submittedBound, "") - if msg != "" { - return true, sql.NullTime{}, &validUntilRefusal{message: msg} - } - return true, bound, nil - }, - Description: description, - Resumes: resumes, + ValidUntil: manageGrantValidUntil(submittedBound, tc), + Description: description, + Resumes: resumes, }) if err != nil { var refused *validUntilRefusal @@ -953,7 +975,7 @@ func (h *OperatorPartialsHandler) ManageGrant(w http.ResponseWriter, r *http.Req // A failed schedule must not report plain success; the warning says the // grant was saved, as issuance's says it was issued. if result.Extended && result.ValidUntil.Valid { - if msg := h.scheduleGrantExpiration(r, result.Grant.GrantID, result.ValidUntil.Time); msg != "" { + if msg := h.scheduleGrantExpiration(r, tc, result.Grant.GrantID, result.ValidUntil.Time); msg != "" { h.renderOrgEnrollmentPage(w, r, orgID, "", "Grant saved, but "+msg) return } @@ -964,17 +986,15 @@ func (h *OperatorPartialsHandler) ManageGrant(w http.ResponseWriter, r *http.Req // scheduleGrantExpiration starts the expiration workflow of a grant written // with a bound, the sole enforcement of valid_until (finding #49). It // returns the end of the warning the page shows when the schedule failed, -// for the caller to lead with what it saved; empty when it did not. -func (h *OperatorPartialsHandler) scheduleGrantExpiration(r *http.Request, grantID string, validUntil time.Time) string { - if h.TemporalClient == nil { - h.Logger.Warn("no Temporal client configured; grant expiration was not scheduled", slog.String("grant_id", grantID)) - return "expiration scheduling FAILED (no scheduler); it will not auto-expire. Revoke it manually to end it." - } +// for the caller to lead with what it saved; empty when it did not. tc is +// the client the request read before it wrote: ManageGrant refuses a new +// bound while there is none, so a bound that reaches this function has one. +func (h *OperatorPartialsHandler) scheduleGrantExpiration(r *http.Request, tc client.Client, grantID string, validUntil time.Time) string { workflowOptions := client.StartWorkflowOptions{ ID: "grant-expiration-" + grantID, TaskQueue: queues.Main, } - if _, err := h.TemporalClient.ExecuteWorkflow(r.Context(), workflowOptions, wf.GrantExpirationWorkflow, wf.GrantExpirationInput{ + if _, err := tc.ExecuteWorkflow(r.Context(), workflowOptions, wf.GrantExpirationWorkflow, wf.GrantExpirationInput{ GrantID: grantID, ValidUntil: validUntil, }); err != nil { diff --git a/internal/server/operator_entitlement_recompute.go b/internal/server/operator_entitlement_recompute.go index 417818a7..0914b063 100644 --- a/internal/server/operator_entitlement_recompute.go +++ b/internal/server/operator_entitlement_recompute.go @@ -37,12 +37,13 @@ func (h *OperatorPartialsHandler) startEntitlementRecomputeDrain(ctx context.Con if result.SyncPath != "deferred" && result.Failed == 0 && result.Recomputed >= result.PoolCount { return } - if h.TemporalClient == nil { + tc := h.Temporal.Client() + if tc == nil { h.Logger.Warn("no Temporal client configured; the entitlement recompute drain was not started", slog.String("change_id", result.ChangeID)) return } - if _, err := h.TemporalClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{ + if _, err := tc.ExecuteWorkflow(ctx, client.StartWorkflowOptions{ ID: wfEnt.RecomputeDrainWorkflowID(result.ChangeID), TaskQueue: queues.Main, WorkflowIDConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING, diff --git a/internal/server/operator_grants_list_test.go b/internal/server/operator_grants_list_test.go index 68814fba..649a87a3 100644 --- a/internal/server/operator_grants_list_test.go +++ b/internal/server/operator_grants_list_test.go @@ -193,6 +193,10 @@ func newGLHarness(t *testing.T, database *sql.DB, personID string) *glHarness { OrgQ: organization.New(database), Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), AuthConfig: authCfg, + // A published holder over a fake whose workflow starts succeed: a save + // that sets an end date is refused while Temporal is not connected + // (D15), and these tests save bounded grants. + Temporal: holderOver(succeedingTemporal()), }) if err != nil { t.Fatalf("NewOperatorPartialsHandler: %v", err) diff --git a/internal/server/operator_integrations_stripe.go b/internal/server/operator_integrations_stripe.go index 2dedd1d3..74e631e6 100644 --- a/internal/server/operator_integrations_stripe.go +++ b/internal/server/operator_integrations_stripe.go @@ -654,13 +654,14 @@ func (h *OperatorPartialsHandler) PostStripeEnvironmentCheck(w http.ResponseWrit http.Error(w, "Stripe is not configured.", http.StatusNotFound) return } - if h.TemporalClient == nil { + tc := h.Temporal.Client() + if tc == nil { fireErrorToast(w, "Temporal is unavailable.") h.renderStripeEnvironmentCheck(w, r) return } - run, err := stripewf.StartEnvironmentCheck(r.Context(), h.TemporalClient, queues.Main, stripewf.EnvironmentCheckTriggerOperator) + run, err := stripewf.StartEnvironmentCheck(r.Context(), tc, queues.Main, stripewf.EnvironmentCheckTriggerOperator) if err != nil { if temporal.IsWorkflowExecutionAlreadyStartedError(err) { fireSuccessToast(w, "Check running.") @@ -717,13 +718,14 @@ func (h *OperatorPartialsHandler) PostStripePublicDetails(w http.ResponseWriter, http.Error(w, "Stripe is not configured.", http.StatusNotFound) return } - if h.TemporalClient == nil { + tc := h.Temporal.Client() + if tc == nil { fireErrorToast(w, "Temporal is unavailable.") h.renderStripePublicDetails(w, r) return } - run, err := stripewf.StartPublicDetailsRead(r.Context(), h.TemporalClient, queues.Main, stripewf.PublicDetailsTriggerOperator) + run, err := stripewf.StartPublicDetailsRead(r.Context(), tc, queues.Main, stripewf.PublicDetailsTriggerOperator) if err != nil { if temporal.IsWorkflowExecutionAlreadyStartedError(err) { fireSuccessToast(w, "Read running.") diff --git a/internal/server/operator_integrations_stripe_test.go b/internal/server/operator_integrations_stripe_test.go index a9924d3d..dfc6acc5 100644 --- a/internal/server/operator_integrations_stripe_test.go +++ b/internal/server/operator_integrations_stripe_test.go @@ -478,7 +478,7 @@ func TestPostStripeEnvironmentCheck(t *testing.T) { h, err := NewOperatorPartialsHandler(OperatorPartialsConfig{ StripeMode: "test", StripeConfigured: true, - TemporalClient: temporalClient, + Temporal: holderOver(temporalClient), Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), AuthConfig: &auth.Config{SessionManager: scs.New()}, IntegrationConfigs: stripeConfigs, diff --git a/internal/server/operator_org_type_change_test.go b/internal/server/operator_org_type_change_test.go index ffadf5a6..2b03b4d6 100644 --- a/internal/server/operator_org_type_change_test.go +++ b/internal/server/operator_org_type_change_test.go @@ -36,6 +36,9 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/server" "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest" "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/htmltest" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" + "github.com/stretchr/testify/mock" + "go.temporal.io/sdk/mocks" ) // otcHarness bundles the handler under test with an authenticated-operator @@ -52,7 +55,17 @@ type otcHarness struct { func newOtcHarness(t *testing.T, database *sql.DB, personID string) *otcHarness { t.Helper() - return newOtcHarnessActs(t, database, personID, entitlements.PlanActConfig{}) + // A published holder over a fake whose workflow starts succeed: a save that + // sets an end date is refused while Temporal is not connected (D15), and + // the tests built on this harness save bounded grants. + return newOtcHarnessWith(t, database, personID, otcOptions{temporal: succeedingClient()}) +} + +// succeedingClient is a client whose workflow starts all succeed. +func succeedingClient() client.Client { + c := &mocks.Client{} + c.On("ExecuteWorkflow", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&mocks.WorkflowRun{}, nil) + return c } // newOtcHarnessActs is newOtcHarness with the queriers the plan acts build @@ -88,16 +101,21 @@ func newOtcHarnessWith(t *testing.T, database *sql.DB, personID string, o otcOpt if o.orgQ == nil { o.orgQ = organization.New(database) } + // A holder per harness; it stays empty when the test passes no client. + temporal := clientholder.New() + if o.temporal != nil { + temporal.Publish(o.temporal) + } handler, err := server.NewOperatorPartialsHandler(server.OperatorPartialsConfig{ - EntitlementsQ: o.entitlementsQ, - BillingQ: o.billingQ, - Database: database, - IdentityQ: identity.New(database), - OrgQ: o.orgQ, - Logger: logger, - AuthConfig: authCfg, - PlanActs: o.acts, - TemporalClient: o.temporal, + EntitlementsQ: o.entitlementsQ, + BillingQ: o.billingQ, + Database: database, + IdentityQ: identity.New(database), + OrgQ: o.orgQ, + Logger: logger, + AuthConfig: authCfg, + PlanActs: o.acts, + Temporal: temporal, }) if err != nil { t.Fatalf("NewOperatorPartialsHandler: %v", err) diff --git a/internal/server/operator_partials.go b/internal/server/operator_partials.go index 73a14254..c78d3b16 100644 --- a/internal/server/operator_partials.go +++ b/internal/server/operator_partials.go @@ -21,7 +21,7 @@ import ( stripedb "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/store" "git.coopcloud.tech/wiki-cafe/member-console/internal/organization" "git.coopcloud.tech/wiki-cafe/member-console/internal/web" - "go.temporal.io/sdk/client" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" ) // OperatorRole is the OIDC role required to access operator pages @@ -114,7 +114,10 @@ type OperatorPartialsHandler struct { // this key from one that ran under another. StripeKeyFingerprint string StripeConfigured bool - TemporalClient client.Client + // Temporal holds the connection to Temporal; a handler reads Client once + // per request, because the connection may arrive after routes are + // registered (see clientholder). + Temporal *clientholder.Holder // Registry is the domains allocation API, for the operator Domains // surface (operator_domains.go). Nil disables that page's data rather // than the page: it renders the "registry unavailable" banner, the same @@ -147,7 +150,7 @@ type OperatorPartialsConfig struct { // under, which is how the page knows the key moved since. StripeKeyFingerprint string StripeConfigured bool - TemporalClient client.Client + Temporal *clientholder.Holder Registry *domains.Registry IntegrationConfigs []IntegrationConfigInfo } @@ -175,7 +178,7 @@ func NewOperatorPartialsHandler(cfg OperatorPartialsConfig) (*OperatorPartialsHa StripeMode: cfg.StripeMode, StripeKeyFingerprint: cfg.StripeKeyFingerprint, StripeConfigured: cfg.StripeConfigured, - TemporalClient: cfg.TemporalClient, + Temporal: cfg.Temporal, Registry: cfg.Registry, }, nil } diff --git a/internal/server/operator_plan_acts.go b/internal/server/operator_plan_acts.go index 367142ac..5c32d74e 100644 --- a/internal/server/operator_plan_acts.go +++ b/internal/server/operator_plan_acts.go @@ -196,11 +196,12 @@ func (h *OperatorPartialsHandler) loadLadderPlanActLines(r *http.Request, ladder // Temporal client and a failed start are therefore both logged and passed over. func (h *OperatorPartialsHandler) startPlanActDrain(ctx context.Context, actID string) { log := logging.FromContext(ctx).With(slog.String("act_id", actID)) - if h.TemporalClient == nil { + tc := h.Temporal.Client() + if tc == nil { log.Warn("no Temporal client configured; the plan act drain was not started") return } - if _, err := h.TemporalClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{ + if _, err := tc.ExecuteWorkflow(ctx, client.StartWorkflowOptions{ ID: wfEnt.PlanActDrainWorkflowID(actID), TaskQueue: queues.Main, WorkflowIDConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING, diff --git a/internal/server/operator_stripe_public_details_test.go b/internal/server/operator_stripe_public_details_test.go index f86abb18..eaa446fc 100644 --- a/internal/server/operator_stripe_public_details_test.go +++ b/internal/server/operator_stripe_public_details_test.go @@ -218,7 +218,7 @@ func TestPostStripePublicDetails(t *testing.T) { h, err := NewOperatorPartialsHandler(OperatorPartialsConfig{ StripeMode: "test", StripeConfigured: true, - TemporalClient: c, + Temporal: holderOver(c), Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), AuthConfig: &auth.Config{SessionManager: scs.New()}, IntegrationConfigs: stripeConfigs, diff --git a/internal/server/server.go b/internal/server/server.go index e5656e12..80fcd508 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -29,9 +29,9 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/middleware" "git.coopcloud.tech/wiki-cafe/member-console/internal/organization" "git.coopcloud.tech/wiki-cafe/member-console/internal/web" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" "github.com/rs/cors" "github.com/spf13/viper" - "go.temporal.io/sdk/client" ) // Config holds the configuration for the server. @@ -46,7 +46,7 @@ type Config struct { BillingQ billing.Querier // Billing module queries StripeQ stripedb.Querier // Stripe module queries DomainsQ domains.Querier // Domains registry queries - TemporalClient client.Client // nil if not configured + Temporal *clientholder.Holder // empty until Temporal connects; handlers read Client per request StripeWebhookSecret string // Stripe webhook signing secret StripeAPIKey string // Stripe API secret key StripeDashboardURL string // Stripe dashboard base URL for deep links @@ -125,14 +125,16 @@ type IntegrationConfigInfo struct { // (ConfigProvider-driven binding is a later step), so this struct never // grows a field per integration. type Deps struct { - Database *sql.DB - IdentityQ identity.Querier - OrgQ organization.Querier - EntitlementsQ entitlements.Querier - BillingQ billing.Querier - TemporalClient client.Client - AuthConfig *auth.Config - Logger *slog.Logger + Database *sql.DB + IdentityQ identity.Querier + OrgQ organization.Querier + EntitlementsQ entitlements.Querier + BillingQ billing.Querier + // Temporal is the holder an adapter's handlers read per request; it may be + // empty when the adapter registers its routes (see clientholder). + Temporal *clientholder.Holder + AuthConfig *auth.Config + Logger *slog.Logger // DomainsPolicy is core configuration, not any integration's: an // integration that builds its own domains.Registry (to claim names for // its resources) must allocate under the SAME policy core does, and a @@ -476,7 +478,7 @@ func registerIntegrationRoutes(httpRequestRouter *http.ServeMux, cfg Config, aut OrgQ: cfg.OrgQ, EntitlementsQ: cfg.EntitlementsQ, BillingQ: cfg.BillingQ, - TemporalClient: cfg.TemporalClient, + Temporal: cfg.Temporal, AuthConfig: authConfig, Logger: cfg.Logger, DomainsPolicy: cfg.DomainsPolicy, @@ -542,7 +544,7 @@ func registerCoreHandlers(httpRequestRouter *http.ServeMux, cfg Config, authConf StripeMode: cfg.StripeMode, StripeKeyFingerprint: cfg.StripeKeyFingerprint, StripeConfigured: cfg.StripeAPIKey != "" && cfg.StripeWebhookSecret != "", - TemporalClient: cfg.TemporalClient, + Temporal: cfg.Temporal, // The operator Domains surface reads and moderates claims through // the registry, under the deployment's policy — the same policy the // member surface allocates under, since the page prints it as the @@ -612,11 +614,11 @@ func registerCoreHandlers(httpRequestRouter *http.ServeMux, cfg Config, authConf // (centralize-external-claim-gate). Registry: domains.NewRegistry(cfg.Database, domains.WithPolicy(cfg.DomainsPolicy), domains.WithExternalClaimGate(NewExternalClaimGate(cfg.EntitlementsQ, cfg.DomainsConnectTarget))), - EntitlementsQ: cfg.EntitlementsQ, - TemporalClient: cfg.TemporalClient, - AuthConfig: authConfig, - Logger: cfg.Logger, - ConnectTarget: cfg.DomainsConnectTarget, + EntitlementsQ: cfg.EntitlementsQ, + Temporal: cfg.Temporal, + AuthConfig: authConfig, + Logger: cfg.Logger, + ConnectTarget: cfg.DomainsConnectTarget, }) if err != nil { cfg.Logger.Error("failed to set up Member Domains handler", slog.Any("error", err)) diff --git a/internal/server/temporal_holder_test.go b/internal/server/temporal_holder_test.go new file mode 100644 index 00000000..4b870386 --- /dev/null +++ b/internal/server/temporal_holder_test.go @@ -0,0 +1,31 @@ +// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial +// SPDX-FileCopyrightText: 2025-2026 Christian Galo + +package server + +import ( + "github.com/stretchr/testify/mock" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/mocks" + + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" +) + +// holderOver is the holder a connected console hands its handlers: c is +// published, or the holder stays empty when c is nil, which is the state +// before Temporal connects. +func holderOver(c client.Client) *clientholder.Holder { + h := clientholder.New() + if c != nil { + h.Publish(c) + } + return h +} + +// succeedingTemporal is a client whose workflow starts all succeed, so a save +// that schedules a grant's expiration finishes as it does in production. +func succeedingTemporal() *mocks.Client { + c := &mocks.Client{} + c.On("ExecuteWorkflow", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&mocks.WorkflowRun{}, nil) + return c +} diff --git a/internal/server/temporal_per_request_db_test.go b/internal/server/temporal_per_request_db_test.go new file mode 100644 index 00000000..b30a9ea2 --- /dev/null +++ b/internal/server/temporal_per_request_db_test.go @@ -0,0 +1,117 @@ +// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial +// SPDX-FileCopyrightText: 2025-2026 Christian Galo + +package server + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http/httptest" + "testing" + "time" + + "github.com/alexedwards/scs/v2" + "github.com/stretchr/testify/mock" + "go.temporal.io/sdk/mocks" + + "git.coopcloud.tech/wiki-cafe/member-console/internal/auth" + "git.coopcloud.tech/wiki-cafe/member-console/internal/domains" + "git.coopcloud.tech/wiki-cafe/member-console/internal/entitlements" + stripewf "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/workflows" + "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" +) + +// startRecorder is a client that records workflow starts and answers each +// with the given run. +func startRecorder(run *mocks.WorkflowRun) *mocks.Client { + c := &mocks.Client{} + c.On("ExecuteWorkflow", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(run, nil) + return c +} + +// TestDomainClaimReadsTheConnectionPerRequest builds the member domains +// handler once with an empty holder. While the holder is empty a claim is +// refused with the existing sentence and nothing starts; after Publish the +// same handler, not rebuilt, starts the verification workflow. A handler that +// copied the client when it was built would never start one. +func TestDomainClaimReadsTheConnectionPerRequest(t *testing.T) { + database := dbtest.Open(t) + ctx := context.Background() + workspaceID := newRollbackWorkspace(t, database) + name := fmt.Sprintf("perrequest%d.example", time.Now().UnixNano()) + + holder := clientholder.New() + h, err := NewMemberDomainsHandler(MemberDomainsConfig{ + DomainsQ: domains.New(database), + Registry: domains.NewRegistry(database), + EntitlementsQ: entitlements.New(database), + Temporal: holder, + AuthConfig: &auth.Config{SessionManager: scs.New()}, + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + }) + if err != nil { + t.Fatalf("NewMemberDomainsHandler: %v", err) + } + fake := startRecorder(&mocks.WorkflowRun{}) + + claim, message := h.startClaimVerification(ctx, workspaceID, name) + if claim != nil || message != "Domain verification is not available right now." { + t.Fatalf("with an empty holder = (%v, %q), want the not-available sentence", claim, message) + } + fake.AssertNumberOfCalls(t, "ExecuteWorkflow", 0) + + holder.Publish(fake) + claim, message = h.startClaimVerification(ctx, workspaceID, name) + if claim == nil { + t.Fatalf("after Publish the claim was refused with %q, want a verification started", message) + } + fake.AssertNumberOfCalls(t, "ExecuteWorkflow", 1) +} + +// TestStripeEnvironmentCheckReadsTheConnectionPerRequest is the same pattern +// for the provider page's Check now control: an empty holder answers +// "Temporal is unavailable.", and after Publish the handler built before it +// starts the check. +func TestStripeEnvironmentCheckReadsTheConnectionPerRequest(t *testing.T) { + holder := clientholder.New() + h, err := NewOperatorPartialsHandler(OperatorPartialsConfig{ + StripeMode: "test", + StripeConfigured: true, + Temporal: holder, + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + AuthConfig: &auth.Config{SessionManager: scs.New()}, + IntegrationConfigs: stripeConfigs, + }) + if err != nil { + t.Fatalf("NewOperatorPartialsHandler: %v", err) + } + post := func() map[string]string { + w := httptest.NewRecorder() + h.PostStripeEnvironmentCheck(w, httptest.NewRequest("POST", "/partials/operator/integrations/stripe/environment-check", nil)) + var fired map[string]string + if err := json.Unmarshal([]byte(w.Header().Get("HX-Trigger")), &fired); err != nil { + t.Fatalf("HX-Trigger %q: %v", w.Header().Get("HX-Trigger"), err) + } + return fired + } + + if got := post()["showErrorToast"]; got != "Temporal is unavailable." { + t.Fatalf("error toast with an empty holder = %q, want %q", got, "Temporal is unavailable.") + } + + run := &mocks.WorkflowRun{} + run.On("Get", mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { + *(args.Get(1).(*stripewf.EnvironmentCheckResult)) = stripewf.EnvironmentCheckResult{Checked: 3, Stale: 1} + }).Return(nil) + fake := startRecorder(run) + holder.Publish(fake) + if got := post()["showSuccessToast"]; got != "3 checked, 1 stale." { + t.Fatalf("success toast after Publish = %q, want the run's counts", got) + } + fake.AssertNumberOfCalls(t, "ExecuteWorkflow", 1) +} diff --git a/internal/workflows/clientholder/clientholder.go b/internal/workflows/clientholder/clientholder.go new file mode 100644 index 00000000..279ee30f --- /dev/null +++ b/internal/workflows/clientholder/clientholder.go @@ -0,0 +1,83 @@ +// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial +// SPDX-FileCopyrightText: 2025-2026 Christian Galo + +// Package clientholder holds the Temporal client for handlers that run before +// the connection exists. +// +// A handler that copied a client when its routes were registered would never +// see one that arrives later, and a nil copy would stay nil for the life of the +// process. A handler holds a *Holder instead and calls Client when a request +// arrives. +// +// The contract, one row per property: +// +// - Empty means never published: a deployment without temporal-host, a +// connection that has not succeeded yet, or a worker precondition that +// refuses. Client returns nil. +// - Client returns nil until Publish, then the published client for the life +// of the process. A nil *Holder returns nil too, so a hand-built +// configuration without a holder works. +// - Writer: the Temporal connector, once, after the worker has started and +// before the integrations' startup hooks run. A second Publish panics, +// because two publishers are a programming error. +// - Read rule: a handler calls Client once when a request arrives and uses +// that value for the whole request. A request that read nil answers for +// that state even if the holder is filled while it runs. +// - The holder is never emptied and never closed; the client lives until the +// process exits. +// - No reader blocks. Publish and Client are one atomic store and one atomic +// load. +// +// The package imports only the Temporal SDK, so any package that starts a +// workflow can import it. +package clientholder + +import ( + "errors" + "sync/atomic" + + "go.temporal.io/sdk/client" +) + +// ErrNotConnected is what a function that needs the Temporal client returns +// while the holder is empty. +var ErrNotConnected = errors.New("Temporal is not connected") //nolint:staticcheck // ST1005: the text is the sentence Stripe shows and a Discourse event records + +// entry gives the atomic pointer a concrete type. +type entry struct { + c client.Client +} + +// Holder publishes one Temporal client to many readers. New returns an empty +// holder; a Holder must not be copied. +type Holder struct { + p atomic.Pointer[entry] +} + +// New returns an empty holder. +func New() *Holder { return &Holder{} } + +// Publish stores the client. It panics when the holder already holds one or +// when c is a nil interface value. A typed nil pointer inside a non-nil +// interface is not detected. +func (h *Holder) Publish(c client.Client) { + if c == nil { + panic("clientholder: Publish of a nil client") + } + if !h.p.CompareAndSwap(nil, &entry{c: c}) { + panic("clientholder: the Temporal client holder was published twice") + } +} + +// Client returns the published client, or nil before Publish and on a nil +// *Holder. +func (h *Holder) Client() client.Client { + if h == nil { + return nil + } + e := h.p.Load() + if e == nil { + return nil + } + return e.c +} diff --git a/internal/workflows/clientholder/clientholder_test.go b/internal/workflows/clientholder/clientholder_test.go new file mode 100644 index 00000000..9619b691 --- /dev/null +++ b/internal/workflows/clientholder/clientholder_test.go @@ -0,0 +1,124 @@ +// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial +// SPDX-FileCopyrightText: 2025-2026 Christian Galo + +package clientholder + +import ( + "sync" + "testing" + + "go.temporal.io/sdk/client" +) + +// fakeClient is a client whose methods are never called; the holder only +// stores it. +type fakeClient struct{ client.Client } + +func panics(f func()) (didPanic bool) { + defer func() { didPanic = recover() != nil }() + f() + return false +} + +func TestHolder(t *testing.T) { + for _, row := range []struct { + name string + check func(*testing.T) + }{ + {"a nil holder reads nil", func(t *testing.T) { + var h *Holder + if got := h.Client(); got != nil { + t.Errorf("Client() on a nil holder = %v, want nil", got) + } + }}, + {"an empty holder reads nil", func(t *testing.T) { + if got := New().Client(); got != nil { + t.Errorf("Client() before Publish = %v, want nil", got) + } + }}, + {"a published holder reads the client", func(t *testing.T) { + h, c := New(), &fakeClient{} + h.Publish(c) + if got := h.Client(); got != client.Client(c) { + t.Errorf("Client() after Publish = %v, want the published client", got) + } + }}, + {"a second Publish panics", func(t *testing.T) { + h := New() + h.Publish(&fakeClient{}) + if !panics(func() { h.Publish(&fakeClient{}) }) { + t.Error("a second Publish returned; want a panic") + } + if got := h.Client(); got == nil { + t.Error("the first client was lost after the second Publish") + } + }}, + {"a nil client panics", func(t *testing.T) { + h := New() + if !panics(func() { h.Publish(nil) }) { + t.Error("Publish(nil) returned; want a panic") + } + if got := h.Client(); got != nil { + t.Errorf("Client() after a refused Publish = %v, want nil", got) + } + }}, + } { + t.Run(row.name, row.check) + } +} + +func TestErrNotConnectedText(t *testing.T) { + if got, want := ErrNotConnected.Error(), "Temporal is not connected"; got != want { + t.Errorf("ErrNotConnected = %q, want %q", got, want) + } +} + +// TestHolderUnderConcurrentReads reads while one goroutine publishes; run +// under -race, a plain field in place of the atomic pointer is reported. +func TestHolderUnderConcurrentReads(t *testing.T) { + h := New() + c := &fakeClient{} + stop := make(chan struct{}) + var wg, started sync.WaitGroup + bad := make(chan client.Client, 16) + for range 8 { + wg.Add(1) + started.Add(1) + go func() { + defer wg.Done() + for i := 0; ; i++ { + if got := h.Client(); got != nil && got != client.Client(c) { + select { + case bad <- got: + default: + } + return + } + if i == 0 { + // Every reader has read once before Publish runs, and + // reads again whatever stop says: the second read is + // unordered with Publish. + started.Done() + continue + } + select { + case <-stop: + return + default: + } + } + }() + } + started.Wait() + h.Publish(c) + close(stop) + wg.Wait() + select { + case got := <-bad: + t.Fatalf("a reader saw %v, want nil or the published client", got) + default: + } + if got := h.Client(); got != client.Client(c) { + t.Errorf("Client() after the readers stopped = %v, want the published client", got) + } +} diff --git a/openspec/changes/serve-before-dependencies/design.md b/openspec/changes/serve-before-dependencies/design.md index 67f74706..b475e596 100644 --- a/openspec/changes/serve-before-dependencies/design.md +++ b/openspec/changes/serve-before-dependencies/design.md @@ -376,6 +376,7 @@ Removed: `Grant issued, but expiration scheduling FAILED (no scheduler); it will | # | Flag | Text | |---|---|---| | 23 | `--temporal-connect-timeout` (changed) | `Time after startup from which failed Temporal connection attempts log at Error; retrying continues (0 = 90s)` | +| 25 | The handlers and drains that start a workflow, while the holder is empty (changed; their level stays) | `Temporal is not connected` in place of `Temporal client not configured` (FedWiki's site actions, API and custom domain, the member domain claim); `Temporal is not connected; the entitlement recompute drain was not started` and `Temporal is not connected; the plan act drain was not started` in place of `no Temporal client configured; ...`. From section 6 the holder is empty while a configured Temporal is still connecting, so "not configured" would be false. | ## Tests diff --git a/openspec/changes/serve-before-dependencies/tasks.md b/openspec/changes/serve-before-dependencies/tasks.md index 4eb7342a..56477ac5 100644 --- a/openspec/changes/serve-before-dependencies/tasks.md +++ b/openspec/changes/serve-before-dependencies/tasks.md @@ -113,33 +113,37 @@ Sections 1 to 7 are one commit each, and their items are that commit's parts; se ## 5. Temporal-backed actions read the connection per request -- [ ] 5.1 `internal/workflows/clientholder` (new package, design D7): `Holder` with `New()`, `Publish(client.Client)` (stores once; a second call panics with a message naming the holder), and `Client()` (nil before `Publish`, nil on a nil `*Holder`); `ErrNotConnected` with the text `Temporal is not connected`. The package doc comment states the contract of the design's holder table. There is no test-kit helper for a published holder: tests write `h := clientholder.New(); h.Publish(fake)`, because a package reached only by tests needs an entry in `scripts/deadcode-testonly.txt` and the prefix line `internal/testkit.*` does not cover `internal/testkit/temporaltest`. -- [ ] 5.2 `internal/server`: `Config.TemporalClient` and `Deps.TemporalClient` become `Temporal *clientholder.Holder`, and every handler configuration that held a `client.Client` holds the holder under the name `Temporal` (`operator_partials.go`, `member_domains.go`, and the handlers they build: `operator_domains.go`, `operator_enrollment.go`, `operator_entitlement_recompute.go`, `operator_integrations_stripe.go`, `operator_plan_acts.go`). Each handler reads `Client()` once at the start of the request and keeps its nil branch and its copy. `test/e2e/plan-management/e2e_test.go` sets `Temporal` in the same step, because its `TemporalClient: nil` stops compiling (item 5.8). -- [ ] 5.3 FedWiki (`fedwiki.go`, `web/api.go`, `web/partials.go`, `web/site_actions.go`, `web/customdomain.go`, `web/operator.go`) and Discourse (`discourse.go`, `web/operator.go`, its schedule health read): the same change. The functions the adapters build at registration (Discourse's `StartReconcile`, Stripe's `StartProcessing`) read the holder when they are called; with no client they return `clientholder.ErrNotConnected` (design D8, D9). -- [ ] 5.4 `internal/integrations/stripe/web/webhook.go` (design D8): when `StartProcessing` returns an error that `errors.Is` `clientholder.ErrNotConnected`, the handler sets `Retry-After: 5`, answers 503 with copy 4 as its body, logs copy 22 at Warn, and leaves the row `received`; every other start error keeps its 500. The `StartProcessing` field's comment and `ServeHTTP`'s doc comment say so. -- [ ] 5.5 `internal/server/operator_enrollment.go` (design D15): the issuance handler refuses a submission with `valid_until` while `Temporal.Client()` is nil, with copy 6 as the field error on `valid_until`, through the same refusal path as the form's other field errors, after the end date is parsed and checked (so a past or malformed date keeps its own field error, such as `Choose a time in the future.`, as `grant_resumption_operator_test.go:1412-1413` asserts for a save) and before `ConferGrant`. The management handler refuses, before its write and inside the `ValidUntil` callback (`operator_enrollment.go`, where `scheduleGrantExpiration` is chosen from `result.ValidUntil.Valid`), a submission whose end date differs from the grant's current one, whose new bound is set (`bound.Valid`) and while the client is nil, with the same field error, in the same place relative to the parse. A save that clears the bound is not refused. The two "no scheduler" branches and their messages are deleted; `scheduleGrantExpiration` keeps its branch for a start that fails. -- [ ] 5.6 `cmd/start.go`: build `clientholder.New()` and pass it as `serverConfig.Temporal`. When `temporal-host` is set, publish the connected client after `worker.Start` returns and before the first `Startup` hook (design D7); the hooks, pollers and schedules keep receiving the client. Temporal still connects before the listener in this section. -- [ ] 5.7 `TestNoHandlerHoldsATemporalClient` (`internal/server`, a structural test with `go/parser`): in every non-test file of `internal/server`, of `internal/integrations/*/web` and of the three adapters' own packages, no struct field's type is `client.Client` from `go.temporal.io/sdk/client`, under any import name; a failure names the file, the line and the field. The scan is a function over source text, and the test runs it on an in-memory source that imports the package as `temporalsdk` and one that imports it unaliased, each with a field of the package's `Client` type, and expects both found. The workflows packages' schedule managers, which receive a client per call from the hooks, are outside the scan, and the test's comment says why. -- [ ] 5.8 Tests. `internal/workflows/clientholder`: `TestHolder`, one row per case (a nil holder, before and after `Publish`, a second `Publish` panics) and `TestHolderUnderConcurrentReads` (readers loop on `Client()` while one goroutine publishes; every reading is nil or the client). The nine test files that pass a fake client as `TemporalClient` or `Temporal` publish the fake to a holder (`clientholder.New()`, then `Publish`), and their results do not change: `internal/integrations/fedwiki/web/oracle_test.go` and `requests_db_test.go`; in `internal/server`, `grant_resumption_operator_test.go`, `member_domains_rollback_db_test.go`, `new_handler_test.go`, `operator_integrations_stripe_test.go`, `operator_org_type_change_test.go` and `operator_stripe_public_details_test.go`; and `test/e2e/plan-management/e2e_test.go`, which `go test ./...` builds and whose `TemporalClient: nil` (`:220`) no longer compiles after item 5.2. The operator harnesses build their configuration without a client, so under D15 every save that sets an end date would be refused with copy 6 and change the result of tests that do not name the warning. The shared harnesses therefore take a published holder over a fake client whose `ExecuteWorkflow` succeeds, and every bounded save keeps its result: `newGLHarness` (`operator_grants_list_test.go:184-196`, used by `operator_grants_delivery_state_test.go:105` `TestGrantDeliveryState_EndToEnd` and `operator_grants_list_test.go:294` `issueAgreementGrants`), `newOtcHarness`, `newGRWorld` (built on `newGLHarness`, `grant_resumption_operator_test.go:65`, so it needs no change of its own; `:836`, `:1226` and `:1509` save with no `w.temporal()` call before them), and the plan-management `httpHarness`, whose `issueTrial` (`:345-365`) and `extendGrant` (`:389-430`) save a `valid_until` and want the trial conferred and a `transfer` transition. `operator_enrollment_manage_guard_test.go` (`TestManageGrant_SucceedsForSinglePoolOrg` at `:256`, `TestManageGrant_ExtendsExactlyTheNamedGrant` at `:292` and `:392`) takes its holder from the same harness (`newOtcHarness`). `w.temporal()` (`grant_resumption_operator_test.go:710-718`) assigns `w.h.handler.TemporalClient` and resets it to nil; once `newGLHarness` publishes its holder, a second `Publish` on it panics (item 5.1), so `w.temporal()` assigns a new published holder to the handler and its cleanup restores the harness's holder. The e2e harness comment (`:202-205`) is rewritten to say that the harness publishes a fake client. The commit message lists these files. `TestDomainClaimReadsTheConnectionPerRequest` (`internal/server`, a database test): a member domains handler built once with an empty holder answers `Domain verification is not available right now.` and starts nothing; after `Publish` of a fake that records starts, the same handler, not rebuilt, starts the verification workflow. The same pattern for one FedWiki site action (`web/site_actions_test` or its database test) and for the Stripe environment check (`Temporal is unavailable.`, then a start). `TestWebhookBeforeTemporalAnswers503` (Stripe `web`, a database test with the committed-rows fixture of the existing receiver tests): a signed event with an empty holder answers 503, `Retry-After: 5`, the body of copy 4, a row `received`, and copy 22 logged; after `Publish`, the redelivery answers 200 and the fake recorded one start keyed by the event id. `TestWebhookBeforeTemporalRecordsFailed` (Discourse `web`, a database test): a `user_added_to_group` event for a managed group and a linked user with an empty holder answers 200, the row reads `failed` with `error_message` copy 5, and nothing panics; after `Publish`, the same kind of event reads `processed` and the reconcile started. `TestBoundedGrantNeedsTemporal` (`internal/server`, a database test), one row per case: issuance with an end date and an empty holder re-renders with copy 6 on the end date and writes no grant row; issuance without an end date issues; a management save that sets a new end date is refused with copy 6 and the grant is unchanged; a management save that clears the end date is saved with an empty holder; with a published fake, issuance with an end date issues and the fake recorded the expiration start. No test asserts the deleted "no scheduler" warnings (`git grep "no scheduler"` over the tests finds nothing), so the files above are the tests whose results the refusal would change without a published fake, and the commit message names them. -- [ ] 5.9 `TestNewHandlerRouteTableNotReady` builds its environment with an empty Temporal holder as well; `routes-not-ready.golden` is regenerated, and every row that changes from section 3's version is listed in the commit message with the reason it changed (a route table probe that reaches a Temporal-backed answer). -- [ ] 5.10 Breaks: +- [x] 5.1 `internal/workflows/clientholder` (new package, design D7): `Holder` with `New()`, `Publish(client.Client)` (stores once; a second call panics with a message naming the holder), and `Client()` (nil before `Publish`, nil on a nil `*Holder`); `ErrNotConnected` with the text `Temporal is not connected`. The package doc comment states the contract of the design's holder table. There is no test-kit helper for a published holder: tests write `h := clientholder.New(); h.Publish(fake)`, because a package reached only by tests needs an entry in `scripts/deadcode-testonly.txt` and the prefix line `internal/testkit.*` does not cover `internal/testkit/temporaltest`. +- [x] 5.2 `internal/server`: `Config.TemporalClient` and `Deps.TemporalClient` become `Temporal *clientholder.Holder`, and every handler configuration that held a `client.Client` holds the holder under the name `Temporal` (`operator_partials.go`, `member_domains.go`, and the handlers they build: `operator_domains.go`, `operator_enrollment.go`, `operator_entitlement_recompute.go`, `operator_integrations_stripe.go`, `operator_plan_acts.go`). Each handler reads `Client()` once at the start of the request and keeps its nil branch and its copy. `test/e2e/plan-management/e2e_test.go` sets `Temporal` in the same step, because its `TemporalClient: nil` stops compiling (item 5.8). +- [x] 5.3 FedWiki (`fedwiki.go`, `web/api.go`, `web/partials.go`, `web/site_actions.go`, `web/customdomain.go`, `web/operator.go`) and Discourse (`discourse.go`, `web/operator.go`, its schedule health read): the same change. The functions the adapters build at registration (Discourse's `StartReconcile`, Stripe's `StartProcessing`) read the holder when they are called; with no client they return `clientholder.ErrNotConnected` (design D8, D9). `docs/building-an-integration.md` section 5 says that `Deps` carries the Temporal holder, empty until Temporal connects, and that a handler calls `Client()` when a request arrives, never while its routes are registered. +- [x] 5.4 `internal/integrations/stripe/web/webhook.go` (design D8): when `StartProcessing` returns an error that `errors.Is` `clientholder.ErrNotConnected`, the handler sets `Retry-After: 5`, answers 503 with copy 4 as its body, logs copy 22 at Warn, and leaves the row `received`; every other start error keeps its 500. The `StartProcessing` field's comment and `ServeHTTP`'s doc comment say so. +- [x] 5.5 `internal/server/operator_enrollment.go` (design D15): the issuance handler refuses a submission with `valid_until` while `Temporal.Client()` is nil, with copy 6 as the field error on `valid_until`, through the same refusal path as the form's other field errors, after the end date is parsed and checked (so a past or malformed date keeps its own field error, such as `Choose a time in the future.`, as `grant_resumption_operator_test.go:1412-1413` asserts for a save) and before `ConferGrant`. The management handler refuses, before its write and inside the `ValidUntil` callback (`operator_enrollment.go`, where `scheduleGrantExpiration` is chosen from `result.ValidUntil.Valid`), a submission whose end date differs from the grant's current one, whose new bound is set (`bound.Valid`) and while the client is nil, with the same field error, in the same place relative to the parse. A save that clears the bound is not refused. The two "no scheduler" branches and their messages are deleted; `scheduleGrantExpiration` keeps its branch for a start that fails. +- [x] 5.6 `cmd/start.go`: build `clientholder.New()` and pass it as `serverConfig.Temporal`. When `temporal-host` is set, publish the connected client after `worker.Start` returns and before the first `Startup` hook (design D7); the hooks, pollers and schedules keep receiving the client. Temporal still connects before the listener in this section. +- [x] 5.7 `TestNoHandlerHoldsATemporalClient` (`internal/server`, a structural test with `go/parser`): in every non-test file of `internal/server`, of `internal/integrations/*/web` and of the three adapters' own packages, no struct field's type is `client.Client` from `go.temporal.io/sdk/client`, under any import name; a failure names the file, the line and the field. The scan is a function over source text, and the test runs it on an in-memory source that imports the package as `temporalsdk` and one that imports it unaliased, each with a field of the package's `Client` type, and expects both found. A file that dot-imports the package is scanned for a bare `Client` field, and the in-memory rows include it. The workflows packages' schedule managers, which are built by `start` and the startup hooks with the connected client and serve no request, are outside the scan, and the test's comment says why. +- [x] 5.8 Tests. `internal/workflows/clientholder`: `TestHolder`, one row per case (a nil holder, before and after `Publish`, a second `Publish` panics, a nil client panics) and `TestHolderUnderConcurrentReads` (readers loop on `Client()` while one goroutine publishes; every reading is nil or the client; a start barrier makes every reader read once before `Publish` and once more whatever the stop channel says, so a read is always unordered with the store). The nine test files that pass a fake client as `TemporalClient` or `Temporal` publish the fake to a holder (`clientholder.New()`, then `Publish`), and their results do not change: `internal/integrations/fedwiki/web/oracle_test.go` and `requests_db_test.go`; in `internal/server`, `grant_resumption_operator_test.go`, `member_domains_rollback_db_test.go`, `new_handler_test.go`, `operator_integrations_stripe_test.go`, `operator_org_type_change_test.go` and `operator_stripe_public_details_test.go`; and `test/e2e/plan-management/e2e_test.go`, which `go test ./...` builds and whose `TemporalClient: nil` (`:220`) no longer compiles after item 5.2. The operator harnesses build their configuration without a client, so under D15 every save that sets an end date would be refused with copy 6 and change the result of tests that do not name the warning. The shared harnesses therefore take a published holder over a fake client whose `ExecuteWorkflow` succeeds, and every bounded save keeps its result: `newGLHarness` (`operator_grants_list_test.go:184-196`, used by `operator_grants_list_test.go:294` `issueAgreementGrants`), `newOtcHarness` (which `operator_grants_delivery_state_test.go:69` `TestGrantDeliveryState_EndToEnd` builds), `newGRWorld` (built on `newGLHarness`, `grant_resumption_operator_test.go:65`, so it needs no change of its own; `:836`, `:1226` and `:1509` save with no `w.temporal()` call before them), and the plan-management `httpHarness`, whose `issueTrial` (`:345-365`) and `extendGrant` (`:389-430`) save a `valid_until` and want the trial conferred and a `transfer` transition. `operator_enrollment_manage_guard_test.go` (`TestManageGrant_SucceedsForSinglePoolOrg` at `:256`, `TestManageGrant_ExtendsExactlyTheNamedGrant` at `:292` and `:392`) takes its holder from the same harness (`newOtcHarness`). `w.temporal()` (`grant_resumption_operator_test.go:710-718`) assigns `w.h.handler.TemporalClient` and resets it to nil; once `newGLHarness` publishes its holder, a second `Publish` on it panics (item 5.1), so `w.temporal()` assigns a new published holder to the handler and its cleanup restores the harness's holder. The e2e harness comment (`:202-205`) is rewritten to say that the harness publishes a fake client. The commit message lists these files. `TestDomainClaimReadsTheConnectionPerRequest` (`internal/server`, a database test): a member domains handler built once with an empty holder answers `Domain verification is not available right now.` and starts nothing; after `Publish` of a fake that records starts, the same handler, not rebuilt, starts the verification workflow. The same pattern for one FedWiki site action (`web/site_actions_test` or its database test) and for the Stripe environment check (`Temporal is unavailable.`, then a start). `TestWebhookBeforeTemporalAnswers503` (Stripe `web`, a database test with the committed-rows fixture of the existing receiver tests): a signed event with an empty holder answers 503, `Retry-After: 5`, the body of copy 4, a row `received`, and copy 22 logged; after `Publish`, the redelivery answers 200 and the fake recorded one start keyed by the event id. `TestWebhookBeforeTemporalRecordsFailed` (Discourse `web`, a database test): a `user_added_to_group` event for a managed group and a linked user with an empty holder answers 200, the row reads `failed` with `error_message` copy 5, and nothing panics; after `Publish`, the same kind of event reads `processed` and the reconcile started. `TestBoundedGrantNeedsTemporal` (`internal/server`, a database test), one row per case: issuance with an end date and an empty holder re-renders with copy 6 on the end date (the `valid_until` error element carries it) and writes no grant row; issuance and a management save with a past end date and an empty holder keep `Choose a time in the future.` and do not show copy 6; issuance without an end date issues; a management save that sets a new end date is refused with copy 6 and the grant is unchanged; a management save that clears the end date is saved with an empty holder; with a published fake, issuance with an end date issues and the fake recorded the expiration start. No test asserts the deleted "no scheduler" warnings (`git grep "no scheduler"` over the tests finds nothing), so the files above are the tests whose results the refusal would change without a published fake, and the commit message names them. +- [x] 5.9 `TestNewHandlerRouteTableNotReady` builds its environment with an empty Temporal holder as well; `routes-not-ready.golden` is compared, and every row that changes from section 3's version is listed in the commit message with the reason it changed. None changes: `routes.golden`'s environment has no Temporal holder either, and an empty holder answers as a nil one, so the commit message says "No row changed" and why. +- [x] 5.10 Breaks: - the member domains handler reads the client at construction (a field filled in `NewMemberDomainsHandler` from `cfg.Temporal.Client()`): caught by `TestDomainClaimReadsTheConnectionPerRequest` (no start after `Publish`); - the FedWiki site action handler reads the client at construction: caught by the FedWiki site action row of item 5.8 (no start after `Publish`); - the Stripe environment-check handler reads the client at construction: caught by the Stripe environment-check row of item 5.8 (no start after `Publish`); - the holder is a plain field in place of an `atomic.Pointer` (command record, `-race`): caught by `TestHolderUnderConcurrentReads`; - a `client.Client` field added back to `MemberDomainsConfig`: caught by `TestNoHandlerHoldsATemporalClient`; - - a `temporalsdk.Client` field (the import name `fedwiki/web/operator.go` and `discourse/web/operator.go` use) added back to a handler configuration: caught by the same test's alias row; + - a `temporalsdk.Client` field (the import name `fedwiki/web/operator.go` and `discourse/web/operator.go` use) added back to a FedWiki handler configuration: caught by `TestNoHandlerHoldsATemporalClient`; + - the scan ignores import names, so a `temporalsdk.Client` field would pass: caught by `TestClientFieldScanFindsEveryImportName`; - Stripe's `StartProcessing` reads the holder at registration: caught by `TestWebhookBeforeTemporalAnswers503` (no start after `Publish`); - the webhook answers 500 for `ErrNotConnected`: caught by the same test (`want 503`); - the webhook answers 503 before the insert (nothing recorded): caught by the same test's row assertion; - Discourse's `StartReconcile` calls the nil client: caught by `TestWebhookBeforeTemporalRecordsFailed` (a panic); + - Discourse's `StartReconcile` reads the holder at registration: caught by the same test's row after `Publish`; - the issuance refusal after `ConferGrant`: caught by `TestBoundedGrantNeedsTemporal` (a grant row is written); - - the management refusal removed: caught by its row; + - the management refusal removed: caught by its row (the handler panics at the scheduler; the row recovers and reports the write); + - the issuance or management refusal placed before the parse: caught by the past-end-date rows; + - the issuance refusal as a form error instead of a field error: caught by the end date field assertion; - the refusal applied to a grant without an end date: caught by the unbounded row; - the management refusal applied to a save that clears the end date: caught by the clearing row; - `Publish` stores a second client without panicking: caught by `TestHolder`; - `Publish` in `start` after the hooks: no test of this section runs `start`; the order is the connector's from section 6, whose test catches it (item 6.10). -- [ ] 5.11 After `test/reset-test-db.sh`, `go test -race -p 1 -count=1 ./internal/workflows/clientholder/ ./internal/server/ ./internal/integrations/stripe/... ./internal/integrations/discourse/... ./internal/integrations/fedwiki/...` passes. +- [x] 5.11 After `test/reset-test-db.sh`, `go test -race -p 1 -count=1 ./internal/workflows/clientholder/ ./internal/server/ ./internal/integrations/stripe/... ./internal/integrations/discourse/... ./internal/integrations/fedwiki/...` passes. ## 6. Temporal connects in the background, and start serves before it @@ -148,7 +152,7 @@ Sections 1 to 7 are one commit each, and their items are that commit's parts; se - [ ] 6.3 `cmd/temporal_connect.go` (new, design D6, D7, D10): `connectTemporal(ctx, t temporalConnect)`, where `temporalConnect` carries the client configuration, the worker configuration, the holder, the logger, the escalation time (`temporal-connect-timeout`, 90 seconds when zero), the policy's first wait and cap (two package variables, 0.5 and 15 seconds, which the boot test of item 6.8 shortens to 50 milliseconds), and the injected `dial`, `newWorker` and `afterConnect` functions. It runs `retryloop.Run` with D17's numbers and copies 13, 14, 15 and 16, and the attempt logs nothing itself: every failure is returned as a `retryloop.Failure`. One attempt: dial (a plain error, copy 13's message); `newCheckedWorker`; on a refusal that is not `ErrCannotCheck`, keep the client and return a `Failure` with copy 15's message and level Error, marked `AtCap`, so that the next attempt reads the list again on the same client; on `ErrCannotCheck`, a build error or a `worker.Start` error, close the client and return the error (a `Failure` with copy 16's message and level Error for the worker's two failures). On success, `Publish`, then `afterConnect(ctx, c, worker)`. `connectTemporal` and the `temporal-host` test that moves into it read the host trimmed, so a host of only whitespace counts as unset (a row of item 6.8 covers it). `afterConnect` in production is today's block of `cmd/start.go` from the `Startup` hooks to the webhook partition schedule, moved unchanged into `startTemporalWork`. - [ ] 6.4 `cmd/start.go` (design D1, D13, D18, D19): `Run` keeps steps 1 to 5 of the design's boot sequence (the key derivations without `UseKey`), then calls `serveConsole(ctx, logger, boot, database, integs, listenTCP(":"+port))`, where `boot` carries the Stripe mode, the key's fingerprint and the integrations' specs `Run` derived, logs copy 21 when it returns and exits 1. `serveConsole` calls `stripeintegration.UseKey` first, then runs the overrides, the boot data state, the connection binding, the mounts and the holder, then `server.Bind(serverConfig, listen)`, then `go connectTemporal(...)` when `temporal-host` is set (the Warn and the backlog otherwise, as today), then the bound server's `Serve(ctx)`. When `ctx` ends it waits for `connectTemporal` and `Discover` to return (a `sync.WaitGroup`) before it returns. A comment above the first `go` statement says that every configuration write is above it and that the goroutines take their keys once (design D13). The words "before any goroutine starts" are no longer true once `UseKey` runs inside `serveConsole`, after `openDB` has started database/sql's connection opener, so they become "before the listener, and before any goroutine that reads the key" in the `UseKey` doc comment (`stripe.go:288-291`) and in the comment above the call in `cmd/start.go`. The block comments of the old Temporal section and `startStripeEnvironmentCheck`'s "boots at the same speed" paragraph are rewritten to the new order. - [ ] 6.5 `internal/server/server.go` (design D19): `Start(ctx, cfg)` becomes two steps. `Bind(cfg, listen func() (net.Listener, error))` runs `auth.Setup`, `NewHandler` and `listen` and returns the bound server, or the first error having started nothing. `(*Bound).Serve(ctx)` starts the discovery loop (`go authConfig.Discover(ctx)`) and serves on the listener; when `ctx` ends it calls `Shutdown` with a bounded context (five seconds), waits for `Discover` to return, and returns. Their doc comments say that nothing before `listen` waits on Temporal or the identity provider and that a failure of `Bind` starts no background work. -- [ ] 6.6 `--temporal-connect-timeout`'s help is copy 23; the viper default stays 90 seconds. +- [ ] 6.6 `--temporal-connect-timeout`'s help is copy 23; the viper default stays 90 seconds. The log lines of copy 25 change as that row says (`internal/integrations/fedwiki/web/api.go`, `customdomain.go`, `site_actions.go`, `internal/server/member_domains.go`, `operator_entitlement_recompute.go`, `operator_plan_acts.go`), each at its current level; a test that asserts one of the old texts asserts the new one. - [ ] 6.7 `internal/integrations/stripe/workflows/webhook.go` (design F34): `ProcessWebhookEvent`'s first `UPDATE` adds `AND status IN ('received','processing','failed')` and the activity checks the rows affected; when none is, it returns nil without dispatching, with a comment that says the sweep and a redelivery can start a completed event's id again. The guard is in the activity because the sweep's status read and the workflow start are two steps, and the workflow id's reuse policy lets a completed id run again. Test (`TestProcessWebhookEventSkipsAFinishedEvent`, a database test, one row per status through a named function): a row at `completed`, `skipped`, `refused` and `dead_letter`, with an event type whose handler fails on the empty payload, is returned nil, unchanged and not dispatched; a row at `received`, one at `processing` (a worker that died mid-attempt, which Temporal then retries) and one at `failed` are each dispatched: the activity returns the dispatch error, and the row reads `failed` with `retry_count` 1 and the error, because the handler fails on the payload and the row does not rest at `processing`. - [ ] 6.8 Tests. `TestConnectTemporal` (`cmd/temporal_connect_test.go`), injected fakes and a fake clock, one row per case through a named function: three refused dials, then a dial that succeeds: the holder is published, `afterConnect` runs once after `Publish` (the fakes record the order of `Start`, `Publish` and `afterConnect`), and copy 14 carries `attempts` 4; a precondition that refuses twice and then passes: one dial, no worker until it passes, the holder empty while it refuses, copy 15 at Error, the second reading a cap's wait after the first; a precondition that returns `ErrCannotCheck`: the client is closed and the next attempt dials again after the backoff's first wait; `worker.Start` that fails once: the client is closed, the next attempt dials again, copy 16; `worker.Start` that fails on every attempt for an hour of fake time: at most fourteen lines carry copy 16, all at Error, because the attempt logs nothing outside the decay; a context that ends during a wait: `connectTemporal` returns and publishes nothing; failures past the escalation time: Error lines. `TestServeConsoleBeforeDependencies` (`cmd/serve_db_test.go`, a database test that also needs the stack's session store), with `temporal-host` naming a listener that accepts, counts and closes each connection (a closed port records nothing, so a count that stays zero could not fail), the connector's first wait and cap shortened to 50 milliseconds through the package variables of item 6.3, `oidc-idp-issuer-url` naming an `httptest` server that answers 503 until the test switches it to a discovery document whose issuer is its own URL, `stripe-api-key` `sk_test_serve` with a webhook secret, the SDK key empty beforehand, and a servable placement seeded the way the domains registry's own database tests seed one, under the test's suffix. The test clears and reads the SDK key through `stripetest.ClearSDKKey(t)` (which restores it in `Cleanup`) and `stripetest.SDKKey()`, both new in `internal/stripetest`, because `.golangci.yml` lints tests and forbids `stripe.Key` outside `internal/(integrations/stripe|stripetest)/`. `serveConsole` runs in a goroutine with a `listen` that opens `127.0.0.1:0` and sends the address on a channel, which is the happens-before edge the key read needs; within one second of the address being received: the ask for the seeded name answers 200 and for an unknown name 404; `/static/app.css` 200; `/login` 503 with `Retry-After: 5`; a signed Stripe event (`webhook.GenerateTestSignedPayload`) 503 with its row `received`; and `stripetest.SDKKey()`, read after the first answer, is the configured key. After the switch, `/login` answers 302 to the fake's authorization endpoint within five seconds. Cancelling the context makes `serveConsole` return, and it returns only after `connectTemporal` and `Discover` have: the listener's count and the identity provider fake's count are equal when `serveConsole` returns and 500 milliseconds later, so a loop that outlived it and dialled again within its shortened wait is seen. A second row runs `serveConsole` with a `listen` that returns the error of binding an address already in use, and asserts that `serveConsole` returns that error and that, 500 milliseconds after it returned, the counting listener recorded no Temporal dial and the identity provider fake no discovery request. A third row sets `valkey-addr` to a refused port and asserts, with and without a session cookie, that the ask answers 200 and 404, `/static/app.css` 200, and an anonymous `/` 302 to `/login`. `TestServeConsoleUnderLoad` runs the first row with twenty goroutines requesting `/login`, the ask and a signed-in page across the switch, under `-race` (item 6.11). - [ ] 6.9 Start this worktree's application on the stack (`go run . start` with `test/.env` loaded) and check that it serves; stop the stack's Temporal container and start the application again: the ask and `/login` answer while the boot log shows copy 13 at Warn, then Error after ninety seconds; start the container again: copy 14 appears and the process has not restarted. Point `oidc-idp-issuer-url` at an address that refuses and check that `/login` answers the styled 503 page and that the ask answers. diff --git a/test/e2e/plan-management/e2e_test.go b/test/e2e/plan-management/e2e_test.go index dbd2e794..d3085c5e 100644 --- a/test/e2e/plan-management/e2e_test.go +++ b/test/e2e/plan-management/e2e_test.go @@ -24,7 +24,10 @@ import ( "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/authtest" "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/dbtest" "git.coopcloud.tech/wiki-cafe/member-console/internal/testkit/world" + "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/clientholder" "github.com/google/uuid" + "github.com/stretchr/testify/mock" + "go.temporal.io/sdk/mocks" ) // TestMVPPlanManagement exercises the full plan-management checkpoint: @@ -199,25 +202,30 @@ type httpHarness struct { // newHTTPHarness constructs an OperatorPartialsHandler wired to the test // database, plus a session manager seeded with an authenticated operator -// context. TemporalClient and StripeQ are nil. IssueGrant checks for nil -// before scheduling trial expiry, and answers 200 with a banner that says the -// scheduling failed. +// context. StripeQ is nil. The harness publishes a fake Temporal client whose +// workflow starts succeed: a grant with an end date is refused while Temporal +// is not connected, and issueTrial and extendGrant save one. func newHTTPHarness(t *testing.T, database *sql.DB, w *world.World, operator world.Person) *httpHarness { t.Helper() sm := authtest.NewSessionManager() authCfg := &auth.Config{SessionManager: sm} + fakeTemporal := &mocks.Client{} + fakeTemporal.On("ExecuteWorkflow", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&mocks.WorkflowRun{}, nil) + temporal := clientholder.New() + temporal.Publish(fakeTemporal) + handler, err := server.NewOperatorPartialsHandler(server.OperatorPartialsConfig{ - EntitlementsQ: entitlements.New(database), - BillingQ: billing.New(database), - StripeQ: nil, - Database: database, - IdentityQ: identity.New(database), - OrgQ: organization.New(database), - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - AuthConfig: authCfg, - TemporalClient: nil, + EntitlementsQ: entitlements.New(database), + BillingQ: billing.New(database), + StripeQ: nil, + Database: database, + IdentityQ: identity.New(database), + OrgQ: organization.New(database), + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + AuthConfig: authCfg, + Temporal: temporal, }) if err != nil { t.Fatalf("NewOperatorPartialsHandler: %v", err)