member-console/internal/integrations/discourse/workflows/workflow.go

package workflows

import (
	"time"

	dcmod "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/discourse/store"
	"go.temporal.io/sdk/temporal"
	"go.temporal.io/sdk/workflow"
)

// ReconcilePersonWorkflowIDPrefix builds the deterministic per-person
// workflow ID: concurrent triggers for one person (webhook + link
// establishment racing) serialize on Temporal's workflow-ID uniqueness
// instead of interleaving their group calls (spec: "Targeted reconcile with
// per-person serialization").
const ReconcilePersonWorkflowIDPrefix = "discourse-reconcile-person-"

// ReconcilePersonWorkflowID returns the workflow ID for a person's targeted
// reconcile.
func ReconcilePersonWorkflowID(personID string) string {
	return ReconcilePersonWorkflowIDPrefix + personID
}

// SweepInput is the sweep workflow's (empty) input — managed groups and
// desired state are resolved fresh each execution, never baked into
// schedule args (FedWiki's stale-args lesson).
type SweepInput struct{}

// SweepResult summarizes one sweep execution.
type SweepResult struct {
	Groups      int
	Linked      int
	Conflicts   int
	Added       int
	Removed     int
	Quarantined int
}

func defaultActivityOptions(ctx workflow.Context) workflow.Context {
	return workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
		// Generous ceiling: a converge pass over a large group at the
		// client's throttled rate can legitimately take minutes.
		StartToCloseTimeout: 15 * time.Minute,
		RetryPolicy: &temporal.RetryPolicy{
			InitialInterval:    5 * time.Second,
			BackoffCoefficient: 2.0,
			MaximumInterval:    2 * time.Minute,
			MaximumAttempts:    5,
		},
	})
}

// DiscourseGroupSyncWorkflow is the periodic sweep: link entitled-but-
// unlinked persons, then converge every managed group to desired state.
// It alone guarantees eventual correctness — webhooks and targeted
// reconciles only reduce latency (design.md D3).
func DiscourseGroupSyncWorkflow(ctx workflow.Context, input SweepInput) (SweepResult, error) {
	ctx = defaultActivityOptions(ctx)
	logger := workflow.GetLogger(ctx)
	var result SweepResult

	var a *Activities

	// A dead admin key answers 404 on /admin/* routes, which downstream
	// lookups read as "record missing" — probe it first so key revocation
	// fails the sweep loudly instead of degrading into "everyone looks
	// absent" (findings #9).
	if err := workflow.ExecuteActivity(ctx, a.VerifyAPIKeyActivity).Get(ctx, nil); err != nil {
		return result, err
	}

	var mappings []dcmod.GroupMapping
	if err := workflow.ExecuteActivity(ctx, a.ListGroupMappingsActivity).Get(ctx, &mappings); err != nil {
		return result, err
	}
	result.Groups = len(mappings)
	if len(mappings) == 0 {
		logger.Info("discourse sweep: no managed groups configured")
		return result, nil
	}

	keySet := make(map[string]bool)
	var keys []string
	for _, m := range mappings {
		if !keySet[m.ResourceKey] {
			keySet[m.ResourceKey] = true
			keys = append(keys, m.ResourceKey)
		}
	}

	var linkStats LinkStats
	if err := workflow.ExecuteActivity(ctx, a.LinkEntitledPersonsActivity, keys).Get(ctx, &linkStats); err != nil {
		return result, err
	}
	result.Linked = linkStats.Linked
	result.Conflicts = linkStats.Conflicts

	// Converge groups sequentially: the client's shared token bucket is the
	// real throughput bound, so parallel activities would only interleave
	// waits while complicating failure attribution.
	for _, mapping := range mappings {
		var stats ConvergeStats
		if err := workflow.ExecuteActivity(ctx, a.ConvergeGroupActivity, mapping).Get(ctx, &stats); err != nil {
			return result, err
		}
		result.Added += stats.Added
		result.Removed += stats.Removed
		result.Quarantined += stats.Quarantined
	}

	logger.Info("discourse sweep complete",
		"groups", result.Groups, "linked", result.Linked,
		"added", result.Added, "removed", result.Removed,
		"quarantined", result.Quarantined)
	return result, nil
}

// ReconcilePersonWorkflow is the targeted fast path for one person,
// started with ReconcilePersonWorkflowID for per-person serialization.
func ReconcilePersonWorkflow(ctx workflow.Context, personID string) error {
	ctx = defaultActivityOptions(ctx)
	var a *Activities
	return workflow.ExecuteActivity(ctx, a.ReconcilePersonActivity, personID).Get(ctx, nil)
}