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