member-console/internal/integrations/fedwiki/workflows/sync.go

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

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