Files
member-console/cmd/start.go
T
cgalo5758 bfc24e2c74 Serve before Temporal connects, and connect it in the background
Sixth section of #141, which closes the issue's defect: start no longer
waits on Temporal before it serves. Start derives the Stripe key's mode
and fingerprint, sets the SDK's key, opens the database, applies the
operator overrides and the boot data state, binds the listener, and only
then starts the Temporal connector and the identity provider's discovery
beside the server. A bind error returns before any dial or discovery
request. While Temporal is unreachable the console serves: sign-in, the
domain ask, the static files and every page that needs no workflow
answer as usual, and the Temporal-backed actions answer as the previous
section made them.

The connector runs the retry loop. Each attempt dials once (NewClient
no longer retries on its own; dialWithRetry and its budget are gone),
builds the worker and runs its precondition. A precondition that
refuses keeps the client and retries at the cap, so FedWiki's refusal
over open runs of a removed workflow type is read again on the same
connection; one that cannot read what it checks, a worker that fails to
build or start, closes the client and dials again. The attempt ends when
the worker starts: "connected to Temporal server" is logged with the
attempts and the time it took, then the holder is published and the
integrations' Startup hooks and the schedules run as before. "Temporal not ready yet, retrying" is a Warn until
temporal-connect-timeout (90 seconds by default), whose help now says
so, and an Error from then on, under the loop's decay. A Temporal OAuth
configuration that cannot be built still refuses at start.

start exits 1 when serving fails, at any point, and returns only after
the connector and discovery have returned. A failure before the
listener keeps the message it had (the overrides, the boot data state,
the integration connections); a serving failure logs "server failed to
start".

The Stripe webhook activity marks an event processing only from
received, processing or failed, so a run of an event already completed,
skipped or dead-lettered returns at once and dispatches nothing.

The handlers and drains that found no Temporal client logged that it
was not configured; with Temporal still connecting that was false, so
they log "Temporal is not connected" (the six plan-act goldens carry the
new drain line). Copies 13, 15 and 16 also carry the host and namespace.

Checked by hand on the test stack: with Temporal stopped, the console
served /login and the ask at once; the connector logged Warn, then
Error from the first attempt after 90 seconds; when Temporal came back
the same process logged "connected to Temporal server" after 14
attempts. With the issuer refusing, /login answered the styled 503 with
Retry-After: 5 and the ask answered.

Full suite on reset databases: 4513 pass, 15 skip, none failed. go test -race -p 1 over cmd,
auth, server, retryloop, workflows and the three integrations after
reset-test-db.sh: every package passes, no data race. make lint passes. Breaks, each caught
(section 1's UseKey break among them). Two depart from item 6.10's
first wording, which the item now carries: the synchronous-fetch break
sits between listen and the server's start (before listen the row would
fail waiting for the address, not at its one-second assertions), and the
"returns without waiting" break is caught by the fourth boot row, whose
fake dial holds for 200 ms of wall-clock time past the context's end,
the least that makes an early return observable.
- serveConsole without UseKey
- UseKey after the bound server's Serve
- Serve without go Discover
- Serve fetches the discovery document before serving
- serveConsole waits for connectTemporal before Serve
- connectTemporal started before Bind returns
- serveConsole returns without waiting for its goroutines
- serveConsole returns without waiting for its goroutines (race)
- skipSession without the ask's path
- connectTemporal publishes after afterConnect
- connectTemporal publishes before worker.Start
- connectTemporal returns after the escalation time
- the open-runs refusal closes the client
- ErrCannotCheck treated as a refusal
- worker.Start failure leaves the client open
- listRunsError without its Is method
- worker.Start failure logged from inside the attempt
- ProcessWebhookEvent without its status guard
- the guard without failed
- the guard without processing
- NewClient without the deadline
- connectTemporal under serveConsole's own context
- the attempt runs the hooks before copy 14
- Run logs copy 21 for a step failure
- the overrides failure is a plain error
- a Temporal authorization that cannot be built is not refused
2026-10-23 20:45:22 -04:00

1016 lines
47 KiB
Go

// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo
package cmd
import (
"context"
"database/sql"
"errors"
"fmt"
"log/slog"
"net"
"os"
"strings"
"sync"
"time"
"git.coopcloud.tech/wiki-cafe/member-console/internal/billing"
"git.coopcloud.tech/wiki-cafe/member-console/internal/config"
"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/identity"
"git.coopcloud.tech/wiki-cafe/member-console/internal/instance"
"git.coopcloud.tech/wiki-cafe/member-console/internal/integration"
"git.coopcloud.tech/wiki-cafe/member-console/internal/integrations"
stripeintegration "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe"
stripemod "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/store"
stripewf "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/workflows"
"git.coopcloud.tech/wiki-cafe/member-console/internal/logging"
"git.coopcloud.tech/wiki-cafe/member-console/internal/organization"
"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"
wfMaintenance "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/maintenance"
"github.com/spf13/cobra"
"github.com/spf13/viper"
enumspb "go.temporal.io/api/enums/v1"
"go.temporal.io/sdk/client"
)
// workerRefusedError is a provider's precondition refusing the start, as
// opposed to the worker failing to build.
type workerRefusedError struct{ err error }
func (e *workerRefusedError) Error() string { return e.err.Error() }
func (e *workerRefusedError) Unwrap() error { return e.err }
// newCheckedWorker runs every provider's WorkerPrecondition, in registry
// order, and builds the worker only when all of them pass: an error from one
// is returned, wrapped in a workerRefusedError, without calling newWorker. It
// is generic in the worker's type so that the connector builds a worker behind
// an interface and the tests build none.
func newCheckedWorker[W any](
ctx context.Context,
c client.Client,
cfg workflows.WorkerConfig,
newWorker func(client.Client, workflows.WorkerConfig) (W, error),
) (W, error) {
for _, wp := range cfg.WorkflowProviders {
if pre, ok := wp.(workflows.WorkerPrecondition); ok {
if err := pre.CheckBeforeWorker(ctx, c); err != nil {
var none W
return none, &workerRefusedError{err: err}
}
}
}
return newWorker(c, cfg)
}
// domainsReconciler is the optional capability an installed integration
// implements to reconcile the domains registry against its own state at boot
// (design D8): ensuring the operator roots its configuration implies, and
// backfilling registry rows for resources that predate the registry.
//
// Declared HERE, at its point of use, rather than in internal/integrations:
// the composition root is the only place that holds both the integration
// registry and the live *sql.DB, and an integration satisfies this
// structurally without importing anything to say so — the same pattern
// server.RouteProvider and workflows.WorkflowProvider follow, for the same
// import-cycle reason.
//
// Implementations MUST be idempotent: the loop below runs on every boot, with
// no gate of any kind, and doubles as the backstop for allocation orphans.
type domainsReconciler interface {
ReconcileDomains(ctx context.Context, database *sql.DB, logger *slog.Logger) error
}
var startCmd = &cobra.Command{
Use: "start",
Short: "Start serving the member-console web application",
Long: `The start command starts an HTTP server that serves the member-console web
application from the components directory in the current directory.
The server listens on port 8080 by default, unless a different port is specified using the --port flag.`,
Args: cobra.NoArgs,
Run: func(cmd *cobra.Command, args []string) {
// Create base context for the application
ctx := context.Background()
// Set up structured logging
env := viper.GetString("env")
logger := logging.SetupLogger(env)
// Store logger in context
ctx = logging.WithContext(ctx, logger)
// Installed integrations (FedWiki, Stripe, ...) are read from the
// registry once and reused for every registry-driven concern below:
// secret/file resolution, config validation, and (further down,
// after the database is up) route/workflow wiring.
integs := integrations.All()
var integrationConfigSpecs []config.ConfigKey
for _, integ := range integs {
if cp, ok := integ.(config.ConfigProvider); ok {
integrationConfigSpecs = append(integrationConfigSpecs, cp.ConfigSpec()...)
}
}
// Check every installed integration's own declarations before
// reading a single configured value: a Type that disagrees with
// its default, an Enum with the wrong default shape, or a
// Positive on a key that cannot carry it is a programming error
// in that integration, not a bad deployment value, so it fails
// boot in development regardless of what any environment sets
// (design D1, D6, typed-config-keys).
if err := config.CheckSpecs(integrationConfigSpecs); err != nil {
logger.Error("invalid integration config declaration", slog.Any("error", err))
os.Exit(1)
}
// Resolve *-secret / *-secret-file pairs into Viper before validating or
// starting any service, so file-backed secrets (e.g. stripe-api-key-file)
// are present when ValidateStart reads them. Shared with config validate
// (D10, typed-config-keys), so both read a deployment's secret-file
// wiring identically.
if err := resolveSecretFiles(integrationConfigSpecs); err != nil {
logger.Error("configuration error", slog.Any("error", err))
os.Exit(1)
}
// Validate required configuration before initializing any service, so a
// misconfiguration fails fast with an actionable, aggregated error rather
// than a late, cryptic downstream failure. integrationConfigSpecs drives
// the generic required-together check (e.g. Stripe's api-key/webhook-
// secret pair) instead of a hardcoded per-integration conditional.
if err := config.ValidateStart(integrationConfigSpecs); err != nil {
logger.Error("invalid configuration", slog.Any("error", err))
os.Exit(1)
}
// The Stripe mode is a property of the API key, derived here beside
// the other configuration checks so a key with an unrecognized
// prefix fails boot rather than mislabeling live money as test
// data downstream (design D7).
stripeMode, err := stripeintegration.ModeForKey(viper.GetString("stripe-api-key"))
if err != nil {
logger.Error("invalid configuration", slog.Any("error", err))
os.Exit(1)
}
// The key's fingerprint, derived here beside its mode: a digest,
// never the key. The environment check records which key it ran
// under, and this is what the next boot compares against that
// record to learn the key moved (stripe-environment-stamp D3).
stripeKeyFingerprint := stripeintegration.FingerprintForKey(viper.GetString("stripe-api-key"))
// Database Setup
database, closeDB, err := openDB(ctx, logger, dbRuntime)
if err != nil {
logOpenFailure(logger, err)
os.Exit(1)
}
defer closeDB()
// Everything after the database steps is the serving half. It returns
// only when serving stops, which in production means it failed; a
// process that stopped serving exits non-zero, so an orchestrator that
// restarts only on failure restarts it (design D18).
boot := bootState{stripeMode: stripeMode, stripeKeyFingerprint: stripeKeyFingerprint, specs: integrationConfigSpecs}
reportServeFailure(logger, serveConsole(ctx, logger, boot, database, integs, listenTCP(":"+viper.GetString("port"))))
os.Exit(1)
},
}
// startStepError is a failure of a step before the listener opens, carrying
// the message the step logged before the serving half existed, so that the
// operator reads the same line for the same refusal. Run logs message with err;
// any other error reads as the server failing to start (design D18).
type startStepError struct {
message string
err error
}
func (e *startStepError) Error() string { return e.message + ": " + e.err.Error() }
func (e *startStepError) Unwrap() error { return e.err }
// reportServeFailure logs what ended serveConsole, which Run calls before it
// exits 1: a step's own message for a startStepError, and copy 21 otherwise.
func reportServeFailure(logger *slog.Logger, err error) {
var step *startStepError
switch {
case errors.As(err, &step):
logger.Error(step.message, slog.Any("error", step.err))
case err != nil:
logger.Error("server failed to start", slog.Any("error", err))
default:
logger.Error("server failed to start")
}
}
// bootState carries what Run derived before the database opened and the
// serving half needs: the Stripe mode, the fingerprint of the key, and the
// integrations' configuration declarations.
type bootState struct {
stripeMode string
stripeKeyFingerprint string
specs []config.ConfigKey
}
// listenTCP returns the function that opens the listener on addr, which is
// what Run passes to serveConsole and a test replaces with a loopback listener
// on a free port.
func listenTCP(addr string) func() (net.Listener, error) {
return func() (net.Listener, error) { return net.Listen("tcp", addr) }
}
// serveConsole is the serving half of start (design D1, D19): it runs the
// steps that need the database, binds the listener and serves. It does not need
// Temporal, the identity provider or the session store to answer. It returns
// the error of the first step that failed, having started no background work (a
// *startStepError for the steps before the listener), or the error that stopped
// serving. When ctx ends, or serving fails, it stops connectTemporal and the
// discovery loop, waits for both to return, and returns: nil for the context's
// end, so a test's cleanup restores what it changed with nothing still
// running.
func serveConsole(
ctx context.Context,
logger *slog.Logger,
boot bootState,
database *sql.DB,
integs []integrations.Integration,
listen func() (net.Listener, error),
) error {
ctx = logging.WithContext(ctx, logger)
// The Stripe SDK's key is set here, once, after the mode and the
// fingerprint are derived and before the listener and before any goroutine
// that reads the key, so the worker's activities, Checkout and the plan
// moves read one key that no later write races; nothing assigns it again.
stripeintegration.UseKey(viper.GetString("stripe-api-key"))
// Layer operator-set config overrides (core.integration_config_
// overrides) over the environment. Must run here — after migrations
// (the table exists), before any route/workflow registration (that's
// when integrations capture their config) and before goroutines
// (viper.Set is not synchronized). Overrides apply on restart by
// design; see internal/config.ApplyOverlay.
if err := applyConfigOverrides(ctx, database, boot.specs); err != nil {
return err
}
// Register providers from code into the registry (parallel to how
// migration sources are enumerated). Runs after migrations, so the
// resource keys each provider manifest names already exist for stamping.
// Data state a booted deployment guarantees (providers, the system
// tenant, the expired-claim sweep, webhook partitions, each
// integration's domains reconciliation with its operator roots).
// Shared with seed-demo so a reset-and-reseeded database matches a
// booted one (bootstate.go).
if err := ensureBootDataState(ctx, database, logger, integs); err != nil {
return &startStepError{message: "boot data state", err: err}
}
// Each key-owning provider's connection, ensured by registration
// above, is looked up once and cached by the integrations that act
// on it, before their routes and workflows are registered below.
if err := bindIntegrationConnections(ctx, database, integs); err != nil {
return &startStepError{message: "bind integration connections", err: err}
}
m := collectMounts(integs)
// The holder the router's Temporal-backed handlers read per request.
// It is empty until connectTemporal publishes a client, which happens once
// the worker is polling; without temporal-host it stays empty.
temporalHolder := clientholder.New()
// Create server config. stripeDashboardURL is Stripe-specific plumbing
// (the derived mode mapped to a dashboard base URL) owned by the
// Stripe integration and only consumed
// downstream by the payments-seam operator billing UI (internal/
// server/operator_billing.go, operator_partials.go), which stays in
// core per design.md Decision 6 (design D7).
if boot.stripeMode != "" {
logger.Info("stripe mode derived from the API key", slog.String("mode", boot.stripeMode))
}
serverConfig := server.Config{
Env: viper.GetString("env"),
Logger: logger,
Database: database,
IdentityQ: identity.New(database),
OrgQ: organization.New(database),
EntitlementsQ: entitlements.New(database),
BillingQ: billing.New(database),
StripeQ: stripemod.New(database),
DomainsQ: domains.New(database),
Temporal: temporalHolder,
StripeWebhookSecret: viper.GetString("stripe-webhook-secret"),
StripeAPIKey: viper.GetString("stripe-api-key"),
StripeDashboardURL: stripeintegration.DashboardURL(boot.stripeMode),
StripeMode: boot.stripeMode,
StripeKeyFingerprint: boot.stripeKeyFingerprint,
BaseURL: viper.GetString("base-url"),
AskFallbackURL: viper.GetString("domains-ask-fallback-url"),
DomainsPolicy: domainsPolicyFromConfig(),
DomainsConnectTarget: viper.GetString("domains-connect-target"),
RouteMounts: m.routes,
UIMounts: m.ui,
DashboardCards: m.cards,
IntegrationConfigs: m.configs,
}
// Every configuration write is above this point, and the first goroutine
// starts below it: viper is not safe for a write concurrent with reads. The
// connector and the discovery loop take their own keys once, here; the
// worker's registrations and the integrations' Startup hooks read theirs
// inside the connector, after the last write (design D13). A Temporal
// authorization that cannot be built refuses here, before the listener.
var connection temporalConnection
temporal, withTemporal, err := newTemporalConnect(logger, database, boot, m.workflowProviders, temporalHolder)
if err != nil {
return &startStepError{message: "failed to connect to Temporal", err: err}
}
// The listener opens before any background work starts, so a port in use
// or a session store that refuses the console exits having started
// nothing (design D19).
bound, err := server.Bind(serverConfig, listen)
if err != nil {
return err
}
// The connector runs on a context that ends when serving ends, so a server
// that fails while Temporal is unreachable still returns (design D18).
background, stopBackground := context.WithCancel(ctx)
defer stopBackground()
var connector sync.WaitGroup
if withTemporal {
connector.Add(1)
go func() {
defer connector.Done()
connection.c, connection.w = connectTemporal(background, temporal)
}()
} else {
logger.Warn("Temporal not configured - integration workflows and scheduled jobs will be unavailable")
// The desired-state triggers keep writing rows; without Temporal
// no relay runs, so boot reports the backlog it leaves.
if err := desiredstate.LogBacklog(ctx, database, logger); err != nil {
logger.Error("failed to read the desired-state outbox backlog", slog.Any("error", err))
}
}
// Serve starts the discovery loop and returns once ctx ends or the server
// fails, having stopped and waited for the loop. The connector is stopped
// and waited on here.
err = bound.Serve(ctx)
stopBackground()
connector.Wait()
connection.stop()
return err
}
// applyConfigOverrides layers the operator's overrides over the environment.
func applyConfigOverrides(ctx context.Context, database *sql.DB, specs []config.ConfigKey) error {
overrideRows, err := integration.New(database).ListConfigOverrides(ctx)
if err != nil {
return &startStepError{message: "failed to load integration config overrides", err: err}
}
overrides := make([]config.Override, 0, len(overrideRows))
for _, row := range overrideRows {
overrides = append(overrides, config.Override{Key: row.Key, Value: row.Value})
}
if err := config.ApplyOverlay(specs, overrides); err != nil {
return &startStepError{message: "invalid integration config override", err: err}
}
return nil
}
// mounts are the route, workflow, UI and card mounts of the installed
// integrations.
type mounts struct {
routes []server.RouteMount
workflowProviders []workflows.WorkflowProvider
ui []server.UIMount
cards []server.DashboardCard
configs []server.IntegrationConfigInfo
}
// collectMounts collects HTTP route mounts, Temporal workflow/activity
// registrations, and UI (template/static) mounts from installed
// integrations. server.RouteProvider and workflows.WorkflowProvider are
// declared at their point of use (not in internal/integrations) to avoid
// import cycles — see those interfaces' doc comments.
// integrations.UIProvider has no such cycle (it references no
// server/workflows types) and is declared alongside Integration
// itself. cmd/start.go is the only place that imports
// internal/integrations, internal/server, and internal/workflows
// together, so it's the only place these type assertions can
// happen.
func collectMounts(integs []integrations.Integration) mounts {
var m mounts
for _, integ := range integs {
if cp, ok := integ.(config.ConfigProvider); ok {
manifest := integ.Provider().ProviderManifest()
m.configs = append(m.configs, server.IntegrationConfigInfo{
Key: integ.Key(),
DisplayName: manifest.DisplayName,
SurfacePath: manifest.OperatorSurfacePath,
Keys: cp.ConfigSpec(),
})
}
if rp, ok := integ.(server.RouteProvider); ok {
m.routes = append(m.routes, server.RouteMount{
Register: rp.RegisterRoutes,
CSRFExemptPaths: rp.CSRFExemptPaths(),
})
}
if wp, ok := integ.(workflows.WorkflowProvider); ok {
m.workflowProviders = append(m.workflowProviders, wp)
}
if uip, ok := integ.(integrations.UIProvider); ok {
m.ui = append(m.ui, server.UIMount{
Key: integ.Key(),
Templates: uip.Templates(),
Static: uip.Static(),
})
}
if dp, ok := integ.(server.DashboardCardProvider); ok {
m.cards = append(m.cards, dp.DashboardCards()...)
}
}
return m
}
// domainsPolicyFromConfig reads the claim lifecycle policy (design D6) once,
// to be carried on server.Config → server.Deps, so core's Domains page and
// every integration surface that claims names allocate under one policy.
// The Temporal worker needs none of it — every policy decision is made
// in the registry at allocation time. Unset keys leave the matching
// field zero, which the registry resolves to its shipped default.
func domainsPolicyFromConfig() domains.Policy {
return domains.Policy{
ClaimWindow: viper.GetDuration("domains-claim-window"),
PendingCap: viper.GetInt("domains-pending-cap"),
AbandonBudget: configuredBudget(viper.GetInt("domains-abandon-budget")),
AbandonWindow: viper.GetDuration("domains-abandon-window"),
ScopeLabels: viper.GetInt("domains-scope-labels"),
InitiationBudget: configuredBudget(viper.GetInt("domains-initiation-budget")),
}
}
// temporalConnection is what connectTemporal connected, kept so serveConsole
// can stop the worker and close the client once the background work has
// returned. It is written by the connector's goroutine and read after
// serveConsole has waited for it.
type temporalConnection struct {
c client.Client
w temporalWorker
}
// stop stops the worker and closes the client, if the connector got that far.
func (t temporalConnection) stop() {
if t.w != nil {
t.w.Stop()
}
if t.c != nil {
t.c.Close()
}
}
// newTemporalConnect builds the connector's inputs from configuration, and
// reports false when temporal-host is not set. It reads every key the
// connector itself needs, so its attempts read none; the worker's registrations
// and the Startup hooks read their own. It returns an error for a Temporal
// authorization that cannot be built, which no attempt could fix.
func newTemporalConnect(
logger *slog.Logger,
database *sql.DB,
boot bootState,
providers []workflows.WorkflowProvider,
holder *clientholder.Holder,
) (temporalConnect, bool, error) {
host := temporalHost()
if host == "" {
return temporalConnect{}, false, nil
}
clientCfg := workflows.ClientConfig{
HostPort: host,
Namespace: viper.GetString("temporal-namespace"),
Logger: logger,
}
tokenURL := viper.GetString("temporal-oauth-token-url")
clientID := viper.GetString("temporal-oauth-client-id")
clientSecret := viper.GetString("temporal-oauth-client-secret")
scopes := viper.GetStringSlice("temporal-oauth-scopes")
if tokenURL != "" || clientID != "" || clientSecret != "" || len(scopes) > 0 {
clientCfg.OAuthTokenProvider = &workflows.OAuthTokenProviderConfig{
TokenURL: tokenURL,
ClientID: clientID,
ClientSecret: clientSecret,
Scopes: scopes,
}
if _, err := workflows.NewOAuthTokenProvider(*clientCfg.OAuthTokenProvider); err != nil {
return temporalConnect{}, false, fmt.Errorf("failed to configure Temporal auth: %w", err)
}
}
workerCfg := workflows.DefaultWorkerConfig(database, logger)
workerCfg.WorkflowProviders = providers
workerCfg.DomainsConnectTarget = viper.GetString("domains-connect-target")
workerCfg.StripeMode = boot.stripeMode
// No transport serves a connection until integration-contract-wire
// adds one: records are computed and stored, nothing is pushed.
workerCfg.Transports = integration.NoTransports{}
work := newTemporalWork(logger, database, boot, providers, workerCfg)
return temporalConnect{
clientConfig: clientCfg,
workerConfig: workerCfg,
holder: holder,
logger: logger,
escalation: viper.GetDuration("temporal-connect-timeout"),
firstWait: temporalFirstWait,
maxWait: temporalMaxWait,
dial: temporalDial,
newWorker: func(c client.Client, cfg workflows.WorkerConfig) (temporalWorker, error) {
w, err := workflows.NewWorker(c, cfg)
if err != nil {
return nil, err
}
return w, nil
},
afterConnect: func(ctx context.Context, c client.Client, _ temporalWorker) {
startTemporalWork(ctx, c, work)
},
}, true, nil
}
// temporalWork is what the work after a connection needs, read from
// configuration before the listener opens.
type temporalWork struct {
logger *slog.Logger
database *sql.DB
providers []workflows.WorkflowProvider
workerConfig workflows.WorkerConfig
stripeMode string
stripeKeyFingerprint string
billingSweepInterval time.Duration
claimSweepInterval time.Duration
webhookPartitionPeriod time.Duration
}
func newTemporalWork(
logger *slog.Logger,
database *sql.DB,
boot bootState,
providers []workflows.WorkflowProvider,
workerCfg workflows.WorkerConfig,
) temporalWork {
w := temporalWork{
logger: logger,
database: database,
providers: providers,
workerConfig: workerCfg,
stripeMode: boot.stripeMode,
stripeKeyFingerprint: boot.stripeKeyFingerprint,
billingSweepInterval: viper.GetDuration("billing-sweep-interval"),
claimSweepInterval: viper.GetDuration("domains-expiry-sweep-interval"),
webhookPartitionPeriod: viper.GetDuration("webhook-partition-ensure-interval"),
}
if w.billingSweepInterval == 0 {
w.billingSweepInterval = wfBilling.DefaultSweepInterval
}
if w.claimSweepInterval == 0 {
w.claimSweepInterval = wfDomains.DefaultExpirySweepInterval
}
if w.webhookPartitionPeriod == 0 {
w.webhookPartitionPeriod = wfMaintenance.DefaultWebhookPartitionEnsureInterval
}
return w
}
// startTemporalWork runs, once a connection is published, today's block of
// start: each integration's Startup hook, the pollers, the relay and the
// read-back schedules, the Stripe environment check and public-details read,
// and the sweep schedules, in that order (design D6). Every step logs its own
// failure and goes on: none of them stops the console.
func startTemporalWork(ctx context.Context, temporalClient client.Client, w temporalWork) {
logger, database := w.logger, w.database
workerCfg := w.workerConfig
// 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
// its own failures (see workflows.WorkflowProvider's doc comment); this
// loop only logs defensively for a well-behaved implementation that
// does return an error.
for _, wp := range w.providers {
if err := wp.Startup(ctx, temporalClient, workerCfg.TaskQueue, database, logger); err != nil {
logger.Error("integration workflow startup failed", slog.Any("error", err))
}
}
// Start the entitlement recompute poller, the drain's backstop:
// a rule change above the synchronous cap leaves one pending
// obligation per pool it owes, the commit starts that change's
// drain itself, and this poller covers a start that failed, a
// process that died mid-drain and an operator Retry that
// requeued failed rows. Conflict policy USE_EXISTING, so a
// second console process joins the running poller.
if _, err := temporalClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
ID: "entitlement-recompute-poller",
TaskQueue: workerCfg.TaskQueue,
WorkflowIDConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
}, wfEnt.PollEntitlementRecompute, wfEnt.PollEntitlementRecomputeInput{}); err != nil {
logger.Error("failed to start entitlement recompute poller workflow", slog.Any("error", err))
} else {
logger.Info("started entitlement recompute poller workflow")
}
// Start the plan act poller, the plan act drains' backstop: a
// default change leaves its listing and its moves to a drain the
// request starts, and this poller covers a start that failed, a
// process that died mid-drain and a Retry that requeued failed
// obligations. USE_EXISTING, as above.
if _, err := temporalClient.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
ID: "plan-act-poller",
TaskQueue: workerCfg.TaskQueue,
WorkflowIDConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
}, wfEnt.PollPlanActs, wfEnt.PollPlanActsInput{}); err != nil {
logger.Error("failed to start plan act poller workflow", slog.Any("error", err))
} else {
logger.Info("started plan act poller workflow")
}
// Start the desired-state relay, which turns the triggers' outbox
// rows into records and pushes them where a transport serves the
// connection. USE_EXISTING, so a second console process joins the
// running relay.
if err := desiredstate.StartRelay(ctx, temporalClient, workerCfg.TaskQueue); err != nil {
logger.Error("failed to start the desired-state relay", slog.Any("error", err))
} else {
logger.Info("started the desired-state relay")
}
// Read-back schedules exist only for a connection a transport
// serves; a leftover one of any other connection is paused.
if err := desiredstate.EnsureReadBackSchedules(ctx, database, workerCfg.Transports,
desiredstate.TemporalReadBackSchedules{
Client: temporalClient,
Logger: logger,
PauseLeftover: workflows.PauseLeftoverSchedule,
ResumeDormant: workflows.ResumeDormantSchedule,
}, desiredstate.ReadBackInterval, logger); err != nil {
logger.Error("failed to set up desired-state read-back schedules", slog.Any("error", err))
}
startStripeWork(ctx, temporalClient, w)
startSweepSchedules(ctx, temporalClient, w)
}
// startStripeWork starts the Stripe environment check and the public-details
// read, when a Stripe key is configured.
func startStripeWork(ctx context.Context, temporalClient client.Client, w temporalWork) {
if w.stripeMode == "" {
return
}
// The Stripe environment check (stripe-environment-stamp D3).
// A Stripe id resolves in the environment that made it and in
// no other, so a key that moved leaves every mapping pointing
// at an object this deployment can no longer read. Boot
// compares the key's fingerprint with the one the last check
// ran under; when they differ, or no check has ever run, every
// product and price id is unverified again and the read-back
// is started. Started, not awaited, like the poller above: a
// deployment whose Stripe is unreachable still comes up, and a
// start that fails is logged and left to the provider page's
// control.
startStripeEnvironmentCheck(ctx, temporalClient, w.workerConfig.TaskQueue, w.database, w.stripeKeyFingerprint, w.logger)
// The Stripe public-details read (merchant-disclosure D1): the
// business name, statement descriptor and icon members meet in
// Stripe's pages live in Stripe's settings and change without
// the console knowing, so every connection reads them again. At most
// four calls, started and not awaited, for the reason above.
if _, err := stripewf.StartPublicDetailsRead(ctx, temporalClient, w.workerConfig.TaskQueue, stripewf.PublicDetailsTriggerBoot); err != nil {
w.logger.Error("failed to start the stripe public details read", slog.Any("error", err))
}
}
// startSweepSchedules sets up the three backstop schedules.
func startSweepSchedules(ctx context.Context, temporalClient client.Client, w temporalWork) {
logger := w.logger
// Set up the billing scheduled-change sweep — the backstop firing
// path for due subscription scheduled changes. Period-end
// cancellations also fire via the Stripe deletion webhook, so a
// failure here is non-fatal.
sweepSchedule := wfBilling.NewScheduleManager(temporalClient, logger)
if err := sweepSchedule.EnsureSweepSchedule(ctx, wfBilling.SweepScheduleConfig{
Interval: w.billingSweepInterval,
}); err != nil {
logger.Error("failed to set up billing sweep schedule", slog.Any("error", err))
}
// Set up the domain claim expiry sweep — the backstop firing path
// for pending claims past their deadline. A claim's own
// verification workflow normally expires it; this catches the ones
// whose workflow is gone. The console also sweeps once per boot
// (ensureBootDataState), which is what covers deployments with no Temporal at
// all, so a failure here is non-fatal.
claimSweepSchedule := wfDomains.NewScheduleManager(temporalClient, logger)
if err := claimSweepSchedule.EnsureExpirySweepSchedule(ctx, wfDomains.ExpirySweepScheduleConfig{
Interval: w.claimSweepInterval,
}); err != nil {
logger.Error("failed to set up domain claim expiry sweep schedule", slog.Any("error", err))
}
// Set up the webhook_events partition ensure schedule (design.md D3,
// webhook-partition-maintenance): the backstop for deployments that
// run longer than the partition lookahead window without
// restarting. The console also ensures partitions once per boot
// (ensureBootDataState), which is what covers deployments with no Temporal at
// all, so a failure here is non-fatal. Unlike the sweeps above, the
// spec requires triggering an immediate run on registration.
webhookPartitionSchedule := wfMaintenance.NewScheduleManager(temporalClient, logger)
if err := webhookPartitionSchedule.EnsureWebhookPartitionSchedule(ctx, wfMaintenance.WebhookPartitionScheduleConfig{
Interval: w.webhookPartitionPeriod,
TriggerImmediately: true,
}); err != nil {
logger.Error("failed to set up webhook partition ensure schedule", slog.Any("error", err))
}
}
// startStripeEnvironmentCheck runs boot's half of the environment check
// (stripe-environment-stamp D3): read the record of the key the last check
// ran under, and when it names another key, or names none, clear every
// product and price id's verification and start the read-back.
//
// Nothing here waits for Stripe. The clears are two statements against the
// console's own tables, and the workflow is started and abandoned, so a
// deployment whose Stripe is unreachable is not held up by it. It runs once
// Temporal is connected, after the connection is published (startTemporalWork),
// never before the listener opens. A failed clear is logged and the check still
// starts: the read-back is what settles each row, and a stale
// verification timestamp on a row the check is about to rewrite is worth
// less than the run.
func startStripeEnvironmentCheck(ctx context.Context, c client.Client, taskQueue string, database *sql.DB, fingerprint string, logger *slog.Logger) {
raw, found, err := instance.NewStore(database).GetJSON(ctx, instance.StripeEnvironmentCheck)
if err != nil {
logger.Error("could not read the stripe environment check record", slog.Any("error", err))
return
}
if !stripemod.EnvironmentCheckNeeded(raw, found, fingerprint) {
return
}
q := stripemod.New(database)
if err := q.ClearProductMappingVerifiedAt(ctx); err != nil {
logger.Error("could not clear the product mappings' verification", slog.Any("error", err))
}
if err := q.ClearPriceMappingVerifiedAt(ctx); err != nil {
logger.Error("could not clear the price mappings' verification", slog.Any("error", err))
}
if _, err := stripewf.StartEnvironmentCheck(ctx, c, taskQueue, stripewf.EnvironmentCheckTriggerBoot); err != nil {
logger.Error("failed to start the stripe environment check workflow", slog.Any("error", err))
return
}
logger.Info("started the stripe environment check workflow",
slog.String("trigger", stripewf.EnvironmentCheckTriggerBoot),
slog.String("key_fingerprint", stripemod.FingerprintPrefix(fingerprint)))
}
func init() {
// Register flags with Cobra
// DO NOT SET DEFAULT VALUES HERE. Use viper.SetDefault() instead. https://github.com/spf13/viper/issues/671
// General configuration
startCmd.Flags().StringP("port", "p", "", "Port to listen on")
startCmd.Flags().String("base-url", "", "Address at which the server is exposed")
startCmd.Flags().String("env", "", "Environment label; development selects text logs, anything else JSON")
startCmd.Flags().String("deployment-name", "", "Name this deployment presents for itself on the member and operator mastheads, page titles, and OpenGraph tags (default \"Member Console\")")
startCmd.Flags().String("db-dsn", "", "PostgreSQL connection string of the runtime user (e.g., postgres://user:pass@localhost:5432/dbname?sslmode=disable; required, or db-dsn-file)")
startCmd.Flags().String("db-dsn-file", "", "Path to file containing the runtime user's PostgreSQL connection string")
startCmd.Flags().String("db-migrate-dsn", "", "PostgreSQL connection string of the migration user; start runs migrations only when it is set")
startCmd.Flags().String("db-migrate-dsn-file", "", "Path to file containing the migration user's PostgreSQL connection string")
startCmd.Flags().String("valkey-addr", "", "Valkey/Redis address for session storage (host:port)")
// OIDC configuration
startCmd.Flags().String("oidc-sp-client-id", "", "OIDC Client ID")
startCmd.Flags().String("oidc-idp-issuer-url", "", "OIDC Identity Provider Issuer URL")
startCmd.Flags().String("valkey-username", "", "Username for the session store (ACL auth; optional)")
startCmd.Flags().String("valkey-password", "", "Password for the session store (required; or valkey-password-file)")
startCmd.Flags().String("valkey-password-file", "", "Path to file containing the session store password")
startCmd.Flags().Bool("valkey-tls", false, "Connect to the session store over TLS")
startCmd.Flags().Bool("valkey-tls-skip-verify", false, "Do not verify the session store's TLS certificate (self-signed stores only)")
startCmd.Flags().String("oidc-idp-account-url", "", "URL of the identity provider's self-service account console, opened by the shell's Identity and Access control (default: issuer URL + /account, Keycloak's shape)")
startCmd.Flags().String("oidc-sp-client-secret", "", "OIDC Client Secret")
startCmd.Flags().String("oidc-sp-client-secret-file", "", "Path to file containing OIDC Client Secret")
startCmd.Flags().Duration("billing-sweep-interval", time.Hour, "Interval between billing scheduled-change sweeps (backstop firing path)")
// Domains registry configuration. The fallback URL is a migration-window
// seam: registry misses (and only misses) are forwarded to a legacy
// on-demand-TLS answerer, so a deployment can strangle a filesystem-based
// answerer incrementally. Empty — the default — refuses unknown names.
startCmd.Flags().String("domains-ask-fallback-url", "", "Legacy on-demand-TLS answerer consulted for names unknown to the domains registry (empty disables)")
startCmd.Flags().String("domains-connect-target", "", "DNS target members point external custom domains at (CNAME host or A-record IP); empty disables external domain claims deployment-wide")
// Claim lifecycle policy (design D6). Flag-only core keys: they appear on
// no operator settings page, so the operator domains surface prints the
// effective set. Defaults are internal/domains' own constants, applied via
// viper.SetDefault below — a single-tenant coop and a public multi-tenant
// host want different numbers, and the defaults suit the latter.
startCmd.Flags().Duration("domains-claim-window", 0, "How long a member has to publish the TXT challenge before a pending external domain claim expires (default 24h)")
startCmd.Flags().Int("domains-pending-cap", 0, "Concurrent pending external domain verifications allowed per workspace (default 5)")
startCmd.Flags().Int("domains-abandon-budget", 0, "Verifications a workspace may start and abandon under one domain scope within the abandonment window before further claims are refused (default 3; 0 disables the ledger)")
startCmd.Flags().Duration("domains-abandon-window", 0, "Rolling window abandoned verifications are counted over (default 168h)")
startCmd.Flags().Int("domains-scope-labels", 0, "Trailing DNS labels forming the scope abandonments are counted under, so sibling names share one scope (default 2)")
startCmd.Flags().Int("domains-initiation-budget", 0, "New external domain claims allowed per workspace per rolling 24h (default 10; 0 disables the cap)")
startCmd.Flags().Duration("domains-expiry-sweep-interval", wfDomains.DefaultExpirySweepInterval, "Interval between domain claim expiry sweeps (backstop for claims whose verification workflow is gone)")
startCmd.Flags().Duration("webhook-partition-ensure-interval", wfMaintenance.DefaultWebhookPartitionEnsureInterval, "Interval between core.webhook_events partition ensure runs (backstop keeping monthly partitions provisioned ahead of need)")
// Support configuration
startCmd.Flags().String("support-url", "", "URL for users to get support (shown in error messages)")
// Integration configuration: every installed integration (FedWiki,
// Stripe, ...) declares its own flags/defaults via ConfigSpec instead
// of each key being hand-declared here — see internal/config.ConfigKey
// and internal/config.ConfigProvider. This replaces what used to be
// separate "FedWiki configuration" (farm API URL, allowed domains,
// site scheme, admin token) and "Stripe configuration" (api key,
// webhook secret, mode) blocks.
registerIntegrationConfigFlags(startCmd)
// Temporal configuration
startCmd.Flags().String("temporal-host", "", "Temporal server host:port (e.g., localhost:7233)")
startCmd.Flags().String("temporal-namespace", "", "Temporal namespace")
startCmd.Flags().String("temporal-oauth-token-url", "", "OAuth2 token endpoint for Temporal authentication")
startCmd.Flags().String("temporal-oauth-client-id", "", "OAuth2 client ID for Temporal authentication")
startCmd.Flags().String("temporal-oauth-client-secret", "", "OAuth2 client secret for Temporal authentication")
startCmd.Flags().String("temporal-oauth-client-secret-file", "", "Path to file containing Temporal OAuth2 client secret")
startCmd.Flags().StringSlice("temporal-oauth-scopes", nil, "OAuth2 scopes for Temporal authentication (comma-separated)")
startCmd.Flags().Duration("temporal-connect-timeout", 0, "Time after startup from which failed Temporal connection attempts log at Error; retrying continues (0 = 90s)")
// Bind all flags to Viper
if err := viper.BindPFlags(startCmd.Flags()); err != nil {
panic(fmt.Sprintf("binding the start flags: %v", err))
}
// Set default values
viper.SetDefault("port", "8080")
viper.SetDefault("valkey-addr", "localhost:6379")
viper.SetDefault("env", "development")
viper.SetDefault("deployment-name", config.DefaultDeploymentName)
viper.SetDefault("temporal-namespace", "default")
viper.SetDefault("temporal-connect-timeout", 90*time.Second)
viper.SetDefault("billing-sweep-interval", time.Hour)
viper.SetDefault("domains-claim-window", domains.ExternalClaimWindow)
viper.SetDefault("domains-pending-cap", domains.DefaultPendingCap)
viper.SetDefault("domains-abandon-budget", domains.DefaultAbandonBudget)
viper.SetDefault("domains-abandon-window", domains.DefaultAbandonWindow)
viper.SetDefault("domains-scope-labels", domains.DefaultScopeLabels)
viper.SetDefault("domains-initiation-budget", domains.DefaultInitiationBudget)
viper.SetDefault("domains-expiry-sweep-interval", wfDomains.DefaultExpirySweepInterval)
viper.SetDefault("webhook-partition-ensure-interval", wfMaintenance.DefaultWebhookPartitionEnsureInterval)
// fedwiki-site-scheme and the other integration defaults are set by
// registerIntegrationConfigFlags above, from each integration's
// declared ConfigSpec.
// Add the command to the root command
rootCmd.AddCommand(startCmd)
}
// registerIntegrationConfigFlags binds a cobra flag — and, for secrets, its
// "<name>-file" companion — for every ConfigKey each installed integration
// declares via ConfigProvider, and sets the key's default via
// viper.SetDefault. This replaces what used to be per-integration flag
// literals and SetDefault calls hand-maintained in this file (see
// internal/config.ConfigKey's doc comment for the declarations this reads).
// Flags always register with a zero-value default (matching the
// "DO NOT SET DEFAULT VALUES HERE" convention above): the real default
// flows through viper.SetDefault instead.
func registerIntegrationConfigFlags(cmd *cobra.Command) {
for _, integ := range integrations.All() {
cp, ok := integ.(config.ConfigProvider)
if !ok {
continue
}
for _, key := range cp.ConfigSpec() {
// A key's value type (declared, or inferred from its Default;
// design D1, typed-config-keys) decides its flag constructor:
// bool, duration, int and list keys get their matching pflag
// type; a url key is a string in Go, so it registers as a
// string flag like a bare string or enum key does (spec
// integration-config-declaration, "A URL key's flag and value
// go through the seam"). No integration knob needs
// hand-declaring in core.
switch key.TypeOf() {
case config.TypeList:
cmd.Flags().StringSlice(key.Name, nil, key.Usage)
case config.TypeBool:
cmd.Flags().Bool(key.Name, false, key.Usage)
case config.TypeDuration:
cmd.Flags().Duration(key.Name, 0, key.Usage)
case config.TypeInt:
cmd.Flags().Int(key.Name, 0, key.Usage)
default: // string, url, enum
cmd.Flags().String(key.Name, "", key.Usage)
}
if key.Secret {
cmd.Flags().String(key.Name+"-file", "", "Path to file containing the "+key.Name+" value")
}
if key.Default != nil {
viper.SetDefault(key.Name, key.Default)
}
}
}
}
// configuredBudget maps a configured claim budget onto domains.Policy's
// encoding. A configured 0 means "never refuse for this reason", which the
// struct's zero value cannot carry — there it means "unconfigured, use the
// default" — so it becomes domains.PolicyDisabled.
func configuredBudget(configured int) int {
if configured == 0 {
return domains.PolicyDisabled
}
return configured
}
// loadFromFile reads a file and returns its contents as a trimmed string
func loadFromFile(path string) (string, error) {
data, err := os.ReadFile(path)
if err != nil {
return "", err
}
return strings.TrimSpace(string(data)), nil
}
// databaseSecretPairs are the database's connection-string pairs. Every
// command that opens the database resolves them (resolveDatabaseSecretFiles),
// and resolveSecretFiles includes them, so a container that mounts the string
// as a file runs migrate, config and seed-demo as it runs start.
var databaseSecretPairs = []config.SecretPair{
{Name: "db-dsn", FileName: "db-dsn-file"},
{Name: "db-migrate-dsn", FileName: "db-migrate-dsn-file"},
}
// resolveSecretFiles loads every "<name>-file" secret variant (core's own
// small, hand-listed set, the database pairs and every installed
// integration's declared Secret keys, config.SecretPairsFrom) into Viper, so a
// file-backed secret (e.g. stripe-api-key-file) is present before any
// validation reads it. Both a direct value and a file path for the same key is
// a fatal misconfiguration, reported as an error rather than exiting here:
// shared between start's own boot sequence and config validate (design D10,
// typed-config-keys), so a deployment's secret-file wiring is read
// identically by both, and each caller decides how to report the failure and
// exit. A command resolves once: a second call after this one would see both
// the value and the file set and report the pair as ambiguous.
func resolveSecretFiles(integrationConfigSpecs []config.ConfigKey) error {
secretPairs := []config.SecretPair{
{Name: "oidc-sp-client-secret", FileName: "oidc-sp-client-secret-file"},
{Name: "temporal-oauth-client-secret", FileName: "temporal-oauth-client-secret-file"},
{Name: "valkey-password", FileName: "valkey-password-file"},
}
secretPairs = append(secretPairs, databaseSecretPairs...)
secretPairs = append(secretPairs, config.SecretPairsFrom(integrationConfigSpecs)...)
return resolveSecretPairs(secretPairs)
}
// resolveDatabaseSecretFiles resolves the database's pairs alone. migrate,
// config list|set|clear and seed-demo call it before openDB; they do not
// resolve the other secrets, because a one-shot migration container may mount
// only the database's secret while its environment names other secret files.
// start and config validate call resolveSecretFiles instead, which includes
// these pairs.
func resolveDatabaseSecretFiles() error {
return resolveSecretPairs(databaseSecretPairs)
}
// resolveSecretPairs loads the file of each pair into the pair's key, and
// reports a pair with both a value and a file as ambiguous.
func resolveSecretPairs(pairs []config.SecretPair) error {
for _, pair := range pairs {
value := viper.GetString(pair.Name)
file := viper.GetString(pair.FileName)
// Both a direct value and a file path is ambiguous.
if value != "" && file != "" {
return fmt.Errorf("%s: both a direct value and a file path are set; use only one", pair.Name)
}
// If a file path is provided, load the value from the file.
if file != "" {
loaded, err := loadFromFile(file)
if err != nil {
return fmt.Errorf("%s: failed to load configuration from file %s: %w", pair.Name, file, err)
}
viper.Set(pair.Name, loaded)
}
}
return nil
}