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)