package workflows import ( "context" "database/sql" "fmt" "log/slog" wfBilling "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/billing" wfDomains "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/domains" wfEnt "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/entitlements" "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/example" wfMaintenance "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/maintenance" "git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/queues" "go.temporal.io/sdk/client" "go.temporal.io/sdk/worker" ) // WorkerConfig holds configuration for the embedded Temporal worker. type WorkerConfig struct { TaskQueue string Database *sql.DB Logger *slog.Logger MaxConcurrentActivities int MaxConcurrentWorkflows int // DomainsConnectTarget is the DNS target external-domain claimants point // their records at, used by the claim-verification workflow's diagnostic // connect probe. The composition root resolves it from the core // `domains-connect-target` key (empty ⇒ external claims are disabled // deployment-wide). It is threaded through the config rather than read // here because this package deliberately holds no viper dependency — // every other worker input arrives the same way. DomainsConnectTarget string // WorkflowProviders registers each installed integration's Temporal // workflows/activities (see WorkflowProvider below). Populated by the // composition root (cmd/start.go) from the integration registry // (internal/integrations.All, type-asserted against WorkflowProvider); // this file never names an integration workflow package (FedWiki, // Stripe, ...) directly. WorkflowProviders []WorkflowProvider } // WorkflowProvider is implemented by installed integrations that run // Temporal workflows/activities against the shared worker. It is declared // here (internal/workflows), not in internal/integrations, mirroring why // server.RouteProvider is declared in internal/server: the design's // import-direction rule holds that internal/server and internal/workflows // end this change with zero internal/integrations imports (only // cmd/start.go and internal/migrate/sources.go consume the registry), so // this file exposes the capability type at its point of use instead of // requiring internal/integrations to declare it and internal/workflows to // import internal/integrations to reference it. cmd/start.go imports both // packages, so it is the only place integrations from the registry are // type-asserted against this interface. (A related, separate cycle — // internal/integrations/fedwiki importing internal/systemtenant for this // capability's Startup below — is worked around in // internal/systemtenant's test doc comment; it is independent of where // this interface itself is declared.) type WorkflowProvider interface { // RegisterWorkflows registers the integration's workflows and // activities against the shared worker at construction time. // database/logger let the integration construct whatever queriers its // own activities need (e.g. fwmod.New(database)); integration-specific // configuration (FedWiki's farm API URL, Stripe's webhook secret, ...) // is read by the integration itself (from viper), not threaded // through this call — mirroring server.RouteProvider's Deps, which // deliberately excludes integration-specific config for the same // reason. RegisterWorkflows(w worker.Worker, database *sql.DB, logger *slog.Logger) // Startup runs once per boot, after the Temporal client is connected // and the worker has started: one-off workflow starts (Stripe's // webhook/outbox pollers) or schedule creation (FedWiki's sync // schedule). By convention, implementations log their own failures and // return nil — an integration's startup hook failing should degrade // that integration's background functionality, not halt boot — but // the composition root also logs any non-nil error defensively. Startup(ctx context.Context, c client.Client, taskQueue string, database *sql.DB, logger *slog.Logger) error } // DefaultWorkerConfig returns a WorkerConfig with sensible defaults. func DefaultWorkerConfig(database *sql.DB, logger *slog.Logger) WorkerConfig { return WorkerConfig{ TaskQueue: queues.Main, Database: database, Logger: logger, MaxConcurrentActivities: 1000, MaxConcurrentWorkflows: 1000, } } // Worker wraps a Temporal worker with application-specific setup. type Worker struct { worker worker.Worker logger *slog.Logger } // NewWorker creates and configures a new embedded Temporal worker. func NewWorker(c client.Client, cfg WorkerConfig) (*Worker, error) { if cfg.TaskQueue == "" { cfg.TaskQueue = queues.Main } opts := worker.Options{} if cfg.MaxConcurrentActivities > 0 { opts.MaxConcurrentActivityExecutionSize = cfg.MaxConcurrentActivities } if cfg.MaxConcurrentWorkflows > 0 { opts.MaxConcurrentWorkflowTaskExecutionSize = cfg.MaxConcurrentWorkflows } w := worker.New(c, cfg.TaskQueue, opts) // --- Domain Registration --- // This is where we wire up the different domains to this worker. // 1. Example Domain w.RegisterWorkflow(example.Workflow) w.RegisterActivity(example.NewActivities(cfg.Logger)) // 2. Installed integrations (FedWiki, Stripe, ...) — each registers its // own workflows/activities via WorkflowProvider; this file names no // integration workflow package directly. for _, wp := range cfg.WorkflowProviders { wp.RegisterWorkflows(w, cfg.Database, cfg.Logger) } // 3. Entitlements Domain — grant-expiration workflow invokes Transition // at valid_until so trial grants cleanly return the pool to the // configured default (or detach if no default is set). w.RegisterWorkflow(wfEnt.GrantExpirationWorkflow) w.RegisterActivity(wfEnt.NewActivities(cfg.Database, cfg.Logger)) // 4. Billing Domain — periodic sweep that fires due subscription scheduled // changes (backstop to the Stripe-webhook-primary firing path). w.RegisterWorkflow(wfBilling.SweepScheduledChangesWorkflow) w.RegisterActivity(wfBilling.NewActivities(cfg.Database, cfg.Logger)) // 5. Domains Domain — external-claim verification polls DNS for the // domain-control challenge and activates the claim; it creates no // provider resource (placement is a separate member action). The // expiry sweep is the backstop for claims whose verification workflow // is gone, which is the only other thing that ever expires one. w.RegisterWorkflow(wfDomains.VerifyClaimWorkflow) w.RegisterWorkflow(wfDomains.SweepExpiredClaimsWorkflow) w.RegisterActivity(wfDomains.NewActivities(cfg.Database, cfg.Logger, cfg.DomainsConnectTarget)) // 6. Maintenance Domain — recurring ensure pass that keeps // core.webhook_events' monthly RANGE partitions provisioned ahead of // need. The console also runs the same ensure once per boot, // unconditionally, because this schedule exists only where Temporal // is configured. w.RegisterWorkflow(wfMaintenance.EnsureWebhookPartitionsWorkflow) w.RegisterActivity(wfMaintenance.NewActivities(cfg.Database, cfg.Logger)) if cfg.Logger != nil { cfg.Logger.Info("created Temporal worker", slog.String("taskQueue", cfg.TaskQueue)) } return &Worker{ worker: w, logger: cfg.Logger, }, nil } func (w *Worker) Start() error { if err := w.worker.Start(); err != nil { return fmt.Errorf("failed to start Temporal worker: %w", err) } if w.logger != nil { w.logger.Info("Temporal worker started") } return nil } func (w *Worker) Stop() { w.worker.Stop() if w.logger != nil { w.logger.Info("Temporal worker stopped") } } func (w *Worker) GracefulStop(ctx context.Context) { w.worker.Stop() if w.logger != nil { w.logger.Info("Temporal worker stopped gracefully") } }