member-console/internal/workflows/worker.go

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")
	}
}