// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial // SPDX-FileCopyrightText: 2025-2026 Christian Galo package workflows import ( "context" "database/sql" "fmt" "log/slog" "time" "git.coopcloud.tech/wiki-cafe/member-console/internal/domains" "git.coopcloud.tech/wiki-cafe/member-console/internal/entitlements" fwmod "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/fedwiki/store" siteusage "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/fedwiki/usage" "git.coopcloud.tech/wiki-cafe/member-console/internal/systemtenant" "go.temporal.io/sdk/temporal" "go.temporal.io/sdk/workflow" ) // normalizeFarmStatus maps a FarmManager status onto our lifecycle states, // folding the legacy soft-delete value `inactive` into `archived` and treating // anything unexpected as `active` (so an unknown state never hides a site). func normalizeFarmStatus(s string) string { switch s { case "active", "readonly", "archived": return s case "inactive": return "archived" default: return "active" } } // SyncFedWikiSitesWorkflowInput is the input for the SyncFedWikiSitesWorkflow. // It carries no workspace identity: the System tenant workspace that owns // ownerless sites is resolved per execution by ResolveSystemWorkspaceActivity, // never from configuration or persisted schedule args. The struct is retained as // an empty type for backward compatibility — schedules created under the // previous code still carry a {"DefaultWorkspaceID": ""} Args JSON, which // Go's default DataConverter silently ignores until the next Update rewrites it. type SyncFedWikiSitesWorkflowInput struct{} // SyncFedWikiSitesWorkflowOutput is the output for the SyncFedWikiSitesWorkflow. type SyncFedWikiSitesWorkflowOutput struct { Success bool SitesFound int SitesAdded int SitesRemoved int ErrorMessage string SyncTimestamp time.Time } // SyncActivityOptions returns activity options for sync operations. func SyncActivityOptions() workflow.ActivityOptions { return workflow.ActivityOptions{ StartToCloseTimeout: 5 * time.Minute, RetryPolicy: &temporal.RetryPolicy{ InitialInterval: time.Second, BackoffCoefficient: 2.0, MaximumInterval: time.Minute, MaximumAttempts: 5, }, } } // SyncFedWikiSitesWorkflow synchronizes sites from the FedWiki farm to the local database. // It fetches all sites from the FarmManager API and ensures the local database reflects // the current state of the farm. func SyncFedWikiSitesWorkflow(ctx workflow.Context, input SyncFedWikiSitesWorkflowInput) (*SyncFedWikiSitesWorkflowOutput, error) { logger := workflow.GetLogger(ctx) logger.Info("SyncFedWikiSitesWorkflow started") var activities *Activities activityCtx := workflow.WithActivityOptions(ctx, SyncActivityOptions()) // Step 0: Resolve the System tenant workspace by natural key at execution // time, so the sync never trusts a workspace ID captured once at // schedule-creation time. On failure, return the same friendly output shape // Step 1 uses rather than a workflow-level error, keeping both pre-steps // uniform. var resolveResult *ResolveSystemWorkspaceOutput if err := workflow.ExecuteActivity(activityCtx, activities.ResolveSystemWorkspaceActivity, ResolveSystemWorkspaceInput{}).Get(ctx, &resolveResult); err != nil { logger.Error("Failed to resolve System tenant workspace", "error", err) return &SyncFedWikiSitesWorkflowOutput{ Success: false, ErrorMessage: "Failed to resolve the System tenant workspace. Please try again later.", SyncTimestamp: workflow.Now(ctx), }, nil } logger.Info("Resolved System tenant workspace", "workspaceID", resolveResult.WorkspaceID) // Step 1: Fetch all sites from the FedWiki farm var listResult *ListFedWikiSitesOutput err := workflow.ExecuteActivity(activityCtx, activities.ListFedWikiSitesActivity, ListFedWikiSitesInput{}).Get(ctx, &listResult) if err != nil { logger.Error("Failed to list sites from FedWiki farm", "error", err) return &SyncFedWikiSitesWorkflowOutput{ Success: false, ErrorMessage: "Failed to fetch sites from FedWiki farm. Please try again later.", SyncTimestamp: workflow.Now(ctx), }, nil } logger.Info("Fetched sites from FedWiki farm", "count", len(listResult.Sites)) // Step 2: Sync the sites to the local database var syncResult *SyncSitesToDBOutput err = workflow.ExecuteActivity(activityCtx, activities.SyncSitesToDBActivity, SyncSitesToDBInput{ FarmSites: listResult.Sites, DefaultWorkspaceID: resolveResult.WorkspaceID, }).Get(ctx, &syncResult) if err != nil { logger.Error("Failed to sync sites to database", "error", err) return &SyncFedWikiSitesWorkflowOutput{ Success: false, SitesFound: len(listResult.Sites), ErrorMessage: "Failed to sync sites to local database.", SyncTimestamp: workflow.Now(ctx), }, nil } // Step 3: Enforce "quota counts active sites" (force_reduce park / re-upgrade // reactivate) across every workspace that owns sites. Idempotent and cheap // when already compliant, so running it each tick covers all downgrade and // upgrade sources without per-trigger wiring. Non-fatal: the sync itself // already succeeded; the next tick retries. var reconcileResult *ReconcileFedWikiQuotaOutput if err := workflow.ExecuteActivity(activityCtx, activities.ReconcileFedWikiQuotaActivity, ReconcileFedWikiQuotaInput{}).Get(ctx, &reconcileResult); err != nil { logger.Error("FedWiki quota reconcile failed (non-fatal)", "error", err) } else if reconcileResult != nil && (reconcileResult.Parked > 0 || reconcileResult.Reactivated > 0) { logger.Info("FedWiki quota reconcile applied", "parked", reconcileResult.Parked, "reactivated", reconcileResult.Reactivated) } // Step 4: Retention purge — permanently delete archived sites past the // retention window. Non-fatal; the next tick retries. var purgeResult *PurgeExpiredArchivedOutput if err := workflow.ExecuteActivity(activityCtx, activities.PurgeExpiredArchivedActivity, struct{}{}).Get(ctx, &purgeResult); err != nil { logger.Error("FedWiki retention purge failed (non-fatal)", "error", err) } else if purgeResult != nil && purgeResult.Purged > 0 { logger.Info("FedWiki retention purge applied", "purged", purgeResult.Purged) } logger.Info("SyncFedWikiSitesWorkflow completed successfully", "sitesFound", len(listResult.Sites), "sitesAdded", syncResult.SitesAdded, "sitesRemoved", syncResult.SitesRemoved) return &SyncFedWikiSitesWorkflowOutput{ Success: true, SitesFound: len(listResult.Sites), SitesAdded: syncResult.SitesAdded, SitesRemoved: syncResult.SitesRemoved, SyncTimestamp: workflow.Now(ctx), }, nil } // ResolveSystemWorkspaceInput is the input for ResolveSystemWorkspaceActivity. type ResolveSystemWorkspaceInput struct{} // ResolveSystemWorkspaceOutput carries the resolved System tenant workspace ID. type ResolveSystemWorkspaceOutput struct { WorkspaceID string } // ResolveSystemWorkspaceActivity resolves the System tenant workspace by natural // key (org_type='system') via systemtenant.Ensure, which is idempotent and cheap // in steady state (one indexed SELECT). Run as Step 0 of the sync workflow so the // workspace ID is derived fresh per execution rather than carried in workflow // input or schedule args, where it would go stale on an app-database wipe. func (a *Activities) ResolveSystemWorkspaceActivity(ctx context.Context, input ResolveSystemWorkspaceInput) (*ResolveSystemWorkspaceOutput, error) { workspaceID, err := systemtenant.Ensure(ctx, a.Database) if err != nil { return nil, fmt.Errorf("resolve system tenant workspace: %w", err) } return &ResolveSystemWorkspaceOutput{WorkspaceID: workspaceID}, nil } // SyncSitesToDBInput is the input for the SyncSitesToDB activity. type SyncSitesToDBInput struct { FarmSites []SiteInfo DefaultWorkspaceID string } // SyncSitesToDBOutput is the output for the SyncSitesToDB activity. type SyncSitesToDBOutput struct { SitesAdded int SitesUpdated int SitesRemoved int } // SyncSitesToDBActivity reconciles local site rows with the FedWiki farm. It // PROJECTS each farm site's observed lifecycle status (active/readonly/archived) // plus storage/last-modified onto the local row — a non-active site is retained // with its status updated, NOT deleted. A row is deleted only when the site is // genuinely absent from the farm, and only for the default (sync-managed) // workspace; user-created sites are never deleted. As a usage drift backstop, a // site whose status crossed the active boundary out-of-band (local vs farm) has // its workspace's active-usage counter adjusted to match. // // Each projection also reaches the domains registry (syncPlacement): placement // servability is re-asserted from the observed status — which is what makes an // out-of-band archive on the farm stop /domains/ask answering for the name — // and an operator placement is ensured for System-tenant sites sitting inside // an operator root and inside no deeper claim. Deleting a farm-absent site // releases its placement. func (a *Activities) SyncSitesToDBActivity(ctx context.Context, input SyncSitesToDBInput) (*SyncSitesToDBOutput, error) { a.Logger.Info("syncing sites to database", slog.Int("farmSiteCount", len(input.FarmSites)), slog.String("defaultWorkspaceID", input.DefaultWorkspaceID)) // Defensive: the System tenant workspace is ensured at boot, so this should // never be empty. Skip rather than FK-error if it somehow is. if input.DefaultWorkspaceID == "" { a.Logger.Warn("system tenant workspace unresolved, skipping site sync") return &SyncSitesToDBOutput{}, nil } localSites, err := a.SiteQ.ListAllSites(ctx) if err != nil { a.Logger.Error("failed to get local sites", slog.Any("error", err)) return nil, fmt.Errorf("failed to get local sites: %w", err) } localSiteMap := make(map[string]fwmod.ListAllSitesRow, len(localSites)) for _, site := range localSites { localSiteMap[site.Domain] = site } farmDomainMap := make(map[string]SiteInfo, len(input.FarmSites)) for _, site := range input.FarmSites { farmDomainMap[site.Name] = site } sitesAdded := 0 sitesUpdated := 0 sitesRemoved := 0 // Farm sites at names no operator root covers. The console never // authorized those, so they get no placement and stay unservable through // /domains/ask (a configured ask fallback covers them during migration). // Counted and logged once per run rather than per site: on a farm carrying // a census of foreign domains, per-site logging is noise. sitesOutsideRoots := 0 // Project every farm site: insert new (default workspace) or update the // observed status/storage/last-modified on an existing row. for _, farmSite := range input.FarmSites { status := normalizeFarmStatus(farmSite.Status) var storage sql.NullInt64 if farmSite.StorageBytes > 0 { storage = sql.NullInt64{Int64: farmSite.StorageBytes, Valid: true} } var lastMod sql.NullTime if farmSite.LastModified != nil { lastMod = sql.NullTime{Time: *farmSite.LastModified, Valid: true} } local, exists := localSiteMap[farmSite.Name] site, err := a.SiteQ.UpsertSiteFromFarm(ctx, fwmod.UpsertSiteFromFarmParams{ WorkspaceID: input.DefaultWorkspaceID, Domain: farmSite.Name, Status: status, StorageBytes: storage, LastModifiedAt: lastMod, }) if err != nil { a.Logger.Error("failed to project farm site", slog.String("domain", farmSite.Name), slog.Any("error", err)) continue } // Project the name into the registry alongside the row: servability // follows the observed status wherever the site's placement lives, and // an operator placement is ensured only for names the console may // authoritatively own (see syncPlacement). if a.syncPlacement(ctx, site, input.DefaultWorkspaceID, IsServableStatus(status)) { sitesOutsideRoots++ } if !exists { sitesAdded++ a.Logger.Info("added site from farm", slog.String("domain", farmSite.Name)) continue } sitesUpdated++ // Drift backstop: if the observed status crossed the active boundary // relative to what we had locally, the event-path counter missed an // out-of-band change — adjust the owning workspace's active usage. a.reconcileUsageOnStatusDrift(ctx, local, normalizeFarmStatus(local.Status), status) } // Rows the farm brought in never passed the create workflow's reservation, // so the counter is now understated. One raise-only call covers the whole // pass: it reads the workspace's active row count, so per-insert calls // would only repeat the same result (design D2). Best-effort — a counter // still understated is repaired at the next boot. if sitesAdded > 0 { if _, err := siteusage.ReconcileWorkspace(ctx, a.EntitlementsQ, a.SiteQ, input.DefaultWorkspaceID); err != nil { a.Logger.Warn("usage reconcile after sync failed", slog.String("workspaceID", input.DefaultWorkspaceID), slog.Any("error", err)) } } // Delete only genuinely farm-absent sites, scoped to the default workspace. for _, localSite := range localSites { if localSite.WorkspaceID != input.DefaultWorkspaceID { continue // user-created site, never delete } if _, present := farmDomainMap[localSite.Domain]; present { continue // present on the farm (in whatever state) — projected above } if err := a.SiteQ.DeleteSiteByDomain(ctx, localSite.Domain); err != nil { a.Logger.Error("failed to delete site", slog.String("domain", localSite.Domain), slog.Any("error", err)) continue } // Deleting a sync-managed site frees its name too: the placement goes // with the row (and an emptied carved claim with the placement). if a.Registry != nil { if _, err := ReleaseSiteDomain(ctx, a.Registry, localSite.SiteID); err != nil { a.Logger.Warn("failed to release placement for farm-absent site", slog.String("domain", localSite.Domain), slog.Any("error", err)) } } sitesRemoved++ a.Logger.Info("removed farm-absent site from default workspace", slog.String("domain", localSite.Domain)) } if sitesOutsideRoots > 0 { a.Logger.Info("farm sites outside every operator root were projected without a placement", slog.Int("count", sitesOutsideRoots)) } a.Logger.Info("site sync completed", slog.Int("sitesAdded", sitesAdded), slog.Int("sitesUpdated", sitesUpdated), slog.Int("sitesRemoved", sitesRemoved)) return &SyncSitesToDBOutput{ SitesAdded: sitesAdded, SitesUpdated: sitesUpdated, SitesRemoved: sitesRemoved, }, nil } // syncPlacement projects one farm-discovered site into the domains registry, // and reports whether the name fell outside every operator root (the caller // counts those for a single end-of-run log line). // // Two distinct jobs, in one registry transaction: // // - Servability follows the resource. Whatever claim the site's placement // lives in — the member's own carved claim, their external claim, or an // operator root — an observed status crossing the archived boundary must // flip it. Addressing the placement by resource ref makes this a no-op // (zero rows) for a site that has none, which is exactly right. // - Placement CREATION is operator-scoped only (design D8, and the sync // requirement in the fedwiki-sites spec). The name must sit inside a live // operator root and inside no deeper live claim; a name inside a member or // external claim is the claim owner's to place, never the sync's, and a // name outside every root was never authorized by the console at all. // // The row's own workspace is checked as well: only System-tenant rows get an // operator placement here. A member's pre-registry site inside a root is the // boot reconciliation's to adopt into a member claim — placing it under the // operator root would silently transfer ownership of the member's name and // then make the backfill skip it as "already placed". func (a *Activities) syncPlacement(ctx context.Context, site fwmod.Site, systemWorkspaceID string, servable bool) (outsideRoots bool) { if a.Registry == nil || site.SiteID == "" { return false } err := a.Registry.WithLock(ctx, func(ctx context.Context, tx *domains.Tx) error { placed, err := tx.SetServableByResource(ctx, fwmod.ModuleName, site.SiteID, servable) if err != nil { return err } if placed { return nil // already has a placement; servability is all sync owns } claim, err := tx.DeepestLiveClaim(ctx, site.Domain) if err != nil { return err } switch { case claim == nil: outsideRoots = true return nil case claim.Kind != domains.KindOperatorRoot: return nil // a member's or an external claim's name — not ours to place case site.WorkspaceID != systemWorkspaceID: return nil // member-owned row inside a root; boot reconciliation adopts it } // Never steal a name another resource already serves: a placement at // this fqdn pointing elsewhere is a genuine collision to surface, not // something to overwrite. if existing, err := tx.PlacementAt(ctx, site.Domain); err != nil { return err } else if existing != nil { a.Logger.Warn("farm site's name is already placed by another resource; skipping", slog.String("domain", site.Domain), slog.String("provider", existing.Provider), slog.String("resourceRef", existing.ResourceRef)) return nil } if _, err := tx.Place(ctx, domains.PlaceParams{ WorkspaceID: systemWorkspaceID, FQDN: site.Domain, Provider: fwmod.ModuleName, ResourceRef: site.SiteID, Servable: servable, }); err != nil { return err } a.Logger.Info("placed farm site under operator root", slog.String("domain", site.Domain), slog.String("root", claim.RootFqdn)) return nil }) if err != nil { // Best-effort: the row is projected either way, and the next tick (or // the boot reconciliation) retries. A registry failure must not abort // the sync of the remaining sites. a.Logger.Warn("failed to project farm site into the domains registry", slog.String("domain", site.Domain), slog.Any("error", err)) return false } return outsideRoots } // reconcileUsageOnStatusDrift adjusts a workspace's fedwiki_sites active-usage // counter when a site's observed status crossed the active boundary out-of-band // (i.e. the local status, set by our own transitions, disagreed with the farm). // In normal operation our SetSiteStatusWorkflow keeps both in step, so this only // fires for changes made directly on the farm. Best-effort: failures are logged. func (a *Activities) reconcileUsageOnStatusDrift(ctx context.Context, site fwmod.ListAllSitesRow, oldStatus, newStatus string) { const active = "active" if oldStatus == newStatus { return } entQ := entitlements.New(a.Database) switch { case oldStatus == active && newStatus != active: if _, err := entQ.AtomicDecrementUsage(ctx, entitlements.AtomicDecrementUsageParams{ WorkspaceID: site.WorkspaceID, ResourceKey: "fedwiki_sites", }); err != nil { a.Logger.Warn("usage drift decrement failed", slog.String("domain", site.Domain), slog.Any("error", err)) } case oldStatus != active && newStatus == active: // Reflect farm reality; if at limit the gated increment is a no-op and we // simply log the genuine over-limit state. res, err := entQ.AtomicIncrementUsage(ctx, entitlements.AtomicIncrementUsageParams{ WorkspaceID: site.WorkspaceID, ResourceKey: "fedwiki_sites", }) if err != nil { a.Logger.Warn("usage drift increment failed", slog.String("domain", site.Domain), slog.Any("error", err)) return } if n, _ := res.RowsAffected(); n == 0 { a.Logger.Warn("site active on farm but workspace is at its site limit (over-limit reality)", slog.String("domain", site.Domain)) } } }