member-console/internal/workflows/domains/verify.go

// Package domains runs the core domain-registry workflows. Today that is
// external-claim verification: proving domain control for a pending
// `domains.claims` row and activating it. The workflow creates no provider
// resource — placing a name inside a verified claim is a separate, instant
// member action (design D4).
package domains

import (
	"context"
	"errors"
	"net"
	"strings"
	"time"

	dommod "git.coopcloud.tech/wiki-cafe/member-console/internal/domains"
	"git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/common"
	"go.temporal.io/sdk/temporal"
	"go.temporal.io/sdk/workflow"
)

// CheckNowSignal asks a running verification workflow to probe immediately
// instead of waiting out its current backoff interval.
const CheckNowSignal = "check-now"

// VerifyClaimWorkflowIDPrefix + claim ID is the deterministic workflow ID of
// a claim's verification poller — shared by the initiation (start),
// check-now (signal), and cancel paths.
const VerifyClaimWorkflowIDPrefix = "verify-claim-"

// Challenge record shape: a TXT record at
// _member-console-challenge.<root> whose value is
// member-console-verify=<token>. Deliberately console-branded, not
// provider-branded: any provider's placements can live under the claim.
const (
	challengeRecordLabel = "_member-console-challenge"
	challengeValuePrefix = "member-console-verify="
)

// Observed-value caps: DNS answers are member-controlled input headed for a
// UI, so bound what a probe will carry.
const (
	maxObservedRecords  = 8
	maxObservedValueLen = 255
	initialPollInterval = 30 * time.Second
	// maxPollInterval caps detection latency at three minutes: the ramp
	// 30s → 1m → 2m → 3m puts five probes inside the first seven minutes,
	// which is where a member fixing DNS actually is. It must stay ≥
	// initialPollInterval, which the check-now reset returns to.
	maxPollInterval = 3 * time.Minute
)

// ChallengeRecordName returns the DNS name the member publishes the TXT
// challenge at.
func ChallengeRecordName(root string) string {
	return challengeRecordLabel + "." + root
}

// ChallengeRecordValue returns the exact TXT record value expected for token.
func ChallengeRecordValue(token string) string {
	return challengeValuePrefix + token
}

// matchesChallengeToken reports whether any published TXT record equals the
// expected challenge value for token.
func matchesChallengeToken(records []string, token string) bool {
	want := ChallengeRecordValue(token)
	for _, record := range records {
		if strings.TrimSpace(record) == want {
			return true
		}
	}
	return false
}

// hasChallengeEvidence reports whether any observed record at the challenge
// name carries OUR challenge value prefix — the evidence latch's predicate
// (design D2). The token need not match: publishing `member-console-verify=…`
// at `_member-console-challenge.<root>` already requires write access to the
// zone, and that is the whole signal.
//
// The two looser predicates are both exploitable and must never be
// substituted:
//
//   - a matching connect record is free for any name already pointed at the
//     deployment, including one the claimant does not control;
//   - "some record exists at the challenge name" (which is exactly what
//     classifyTXT reports as `mismatch`) would let a wildcard TXT in a
//     victim's own zone permanently exempt every name a squatter aims at them.
func hasChallengeEvidence(observed []string) bool {
	for _, record := range observed {
		if strings.HasPrefix(strings.TrimSpace(record), challengeValuePrefix) {
			return true
		}
	}
	return false
}

// capObserved bounds a probe's observed values for storage/display.
//
// The values are whatever bytes a member's DNS provider serves, and they end
// up in a `text` column: anything that is not valid UTF-8 must be repaired
// here, because Postgres rejects the INSERT outright and the recording
// activity would then burn its retries and leave the probe unchecked forever.
// The length cap counts bytes (the column's own budget) but cuts on a rune
// boundary, so truncation cannot manufacture the invalid sequence the repair
// just removed.
func capObserved(values []string) []string {
	capped := make([]string, 0, min(len(values), maxObservedRecords))
	for _, v := range values {
		v = strings.TrimSpace(strings.ToValidUTF8(v, "�"))
		if v == "" {
			continue
		}
		if len(v) > maxObservedValueLen {
			v = truncateRunes(v, maxObservedValueLen) + "…"
		}
		capped = append(capped, v)
		if len(capped) == maxObservedRecords {
			break
		}
	}
	return capped
}

// truncateRunes returns the longest prefix of s that is at most maxBytes long
// and ends on a rune boundary. s must already be valid UTF-8.
func truncateRunes(s string, maxBytes int) string {
	if len(s) <= maxBytes {
		return s
	}
	end := 0
	for i := range s {
		if i > maxBytes {
			break
		}
		end = i
	}
	return s[:end]
}

// classifyTXT maps a successful TXT lookup to a probe state: the expected
// value present is a match, other records at the challenge name are a
// mismatch (typo — show what we saw), none at all is missing.
//
// evidence reports the latch predicate (design D2) and is read from the RAW
// records, exactly like matchesChallengeToken. `observed` is display data and
// is capped at maxObservedRecords, so a zone serving nine TXT records with the
// challenge-prefixed one last would prove control for verification and lose it
// for the ledger if the latch read the capped list.
func classifyTXT(records []string, token string) (state string, observed []string, evidence bool) {
	observed = capObserved(records)
	evidence = hasChallengeEvidence(records)
	switch {
	case matchesChallengeToken(records, token):
		return dommod.ProbeStateMatch, observed, evidence
	case len(observed) > 0:
		return dommod.ProbeStateMismatch, observed, evidence
	default:
		return dommod.ProbeStateMissing, observed, evidence
	}
}

// normalizeDNSName lowercases and strips the trailing dot resolver answers
// carry, so canonical names compare against configured targets.
func normalizeDNSName(name string) string {
	return strings.TrimSuffix(strings.ToLower(strings.TrimSpace(name)), ".")
}

// hostsIntersect reports whether the two address sets share any member.
func hostsIntersect(a, b []string) bool {
	set := make(map[string]struct{}, len(a))
	for _, addr := range a {
		set[addr] = struct{}{}
	}
	for _, addr := range b {
		if _, ok := set[addr]; ok {
			return true
		}
	}
	return false
}

// classifyConnect maps the connect-record observations to a probe state.
// canonical is the chain-followed CNAME answer for the name (normalized;
// equal to the name when no CNAME exists). A CNAME to the target matches
// outright; otherwise address intersection covers A-record and
// ALIAS/flattened setups.
func classifyConnect(name, target, canonical string, nameAddrs, targetAddrs []string) (state string, observed []string) {
	hasCNAME := canonical != "" && canonical != normalizeDNSName(name)
	if hasCNAME && canonical == target {
		return dommod.ProbeStateMatch, capObserved([]string{canonical})
	}
	if hostsIntersect(nameAddrs, targetAddrs) {
		if hasCNAME {
			return dommod.ProbeStateMatch, capObserved([]string{canonical})
		}
		return dommod.ProbeStateMatch, capObserved(nameAddrs)
	}
	if hasCNAME {
		return dommod.ProbeStateMismatch, capObserved([]string{canonical})
	}
	if len(nameAddrs) == 0 {
		return dommod.ProbeStateMissing, nil
	}
	return dommod.ProbeStateMismatch, capObserved(nameAddrs)
}

// DNSActivityOptions: lookups are quick and the verification workflow's poll
// loop provides the long-horizon retry, so in-activity retries stay shorter
// than common.DefaultActivityOptions.
func DNSActivityOptions() workflow.ActivityOptions {
	return workflow.ActivityOptions{
		StartToCloseTimeout: 30 * time.Second,
		RetryPolicy: &temporal.RetryPolicy{
			InitialInterval:    2 * time.Second,
			BackoffCoefficient: 2.0,
			MaximumInterval:    10 * time.Second,
			MaximumAttempts:    3,
		},
	}
}

// isDNSNotFound reports whether err is the resolver telling us the name (or
// record) simply is not published yet — the normal pre-propagation state.
func isDNSNotFound(err error) bool {
	var dnsErr *net.DNSError
	return errors.As(err, &dnsErr) && dnsErr.IsNotFound
}

// CheckTXTInput is the input for CheckTXTActivity.
type CheckTXTInput struct {
	Root  string // the claim's root FQDN
	Token string
}

// CheckTXTOutput is the output for CheckTXTActivity.
type CheckTXTOutput struct {
	State    string // dommod.ProbeState*
	Observed []string
	// Evidence reports the evidence latch's predicate (design D2) as computed
	// from the RAW resolver answer. It travels beside Observed rather than
	// being recomputed from it, because Observed is capped for storage and the
	// latch must see everything the resolver returned.
	Evidence bool
}

// CheckTXTActivity looks up the challenge TXT record for the claim root and
// classifies what it finds. It resolves through the process resolver
// deliberately — the same vantage point the deployment serves from.
// Transient resolver trouble (SERVFAIL, timeouts) is a probe *state*, not an
// activity failure: the poll loop is the long-horizon retry.
func (a *Activities) CheckTXTActivity(ctx context.Context, input CheckTXTInput) (*CheckTXTOutput, error) {
	records, err := net.DefaultResolver.LookupTXT(ctx, ChallengeRecordName(input.Root))
	if err != nil {
		if isDNSNotFound(err) {
			return &CheckTXTOutput{State: dommod.ProbeStateMissing}, nil
		}
		var dnsErr *net.DNSError
		if errors.As(err, &dnsErr) {
			return &CheckTXTOutput{State: dommod.ProbeStateError}, nil
		}
		return nil, err
	}
	state, observed, evidence := classifyTXT(records, input.Token)
	return &CheckTXTOutput{State: state, Observed: observed, Evidence: evidence}, nil
}

// CheckConnectInput is the input for CheckConnectActivity. Target empty
// falls back to the worker's configured connect target.
type CheckConnectInput struct {
	Root   string
	Target string
}

// CheckConnectOutput is the output for CheckConnectActivity.
type CheckConnectOutput struct {
	State    string // dommod.ProbeState*
	Observed []string
}

// CheckConnectActivity checks whether the claim root points at the
// deployment's connect target. Diagnostic only — the TXT challenge alone
// drives verification (design D4); this exists so the member can see the
// routing record's health before placing a name.
func (a *Activities) CheckConnectActivity(ctx context.Context, input CheckConnectInput) (*CheckConnectOutput, error) {
	target := normalizeDNSName(input.Target)
	if target == "" {
		target = normalizeDNSName(a.ConnectTarget)
	}
	if target == "" {
		return &CheckConnectOutput{State: dommod.ProbeStateUnchecked}, nil
	}

	// IP-literal target: the member is expected to publish an A/AAAA record.
	if net.ParseIP(target) != nil {
		addrs, err := net.DefaultResolver.LookupHost(ctx, input.Root)
		if err != nil {
			if isDNSNotFound(err) {
				return &CheckConnectOutput{State: dommod.ProbeStateMissing}, nil
			}
			return &CheckConnectOutput{State: dommod.ProbeStateError}, nil
		}
		state, observed := classifyConnect(input.Root, target, "", addrs, []string{target})
		return &CheckConnectOutput{State: state, Observed: observed}, nil
	}

	// Hostname target: prefer the CNAME chain; fall back to address
	// intersection for A-record/ALIAS setups.
	canonical := ""
	if cname, err := net.DefaultResolver.LookupCNAME(ctx, input.Root); err == nil {
		canonical = normalizeDNSName(cname)
	} else if isDNSNotFound(err) {
		return &CheckConnectOutput{State: dommod.ProbeStateMissing}, nil
	}
	if canonical != "" && canonical == target {
		return &CheckConnectOutput{State: dommod.ProbeStateMatch, Observed: capObserved([]string{canonical})}, nil
	}

	rootAddrs, err := net.DefaultResolver.LookupHost(ctx, input.Root)
	if err != nil {
		if isDNSNotFound(err) {
			return &CheckConnectOutput{State: dommod.ProbeStateMissing}, nil
		}
		return &CheckConnectOutput{State: dommod.ProbeStateError}, nil
	}
	targetAddrs, err := net.DefaultResolver.LookupHost(ctx, target)
	if err != nil {
		// The operator's own target failing to resolve is not the member's
		// record being wrong — surface as a probe error, with what we did see.
		return &CheckConnectOutput{State: dommod.ProbeStateError, Observed: capObserved(rootAddrs)}, nil
	}
	state, observed := classifyConnect(input.Root, target, canonical, rootAddrs, targetAddrs)
	return &CheckConnectOutput{State: state, Observed: observed}, nil
}

// RecordProbeInput is the input for RecordProbeActivity.
type RecordProbeInput struct {
	ClaimID         string
	TXTState        string
	TXTObserved     []string
	ConnectState    string
	ConnectObserved []string
	// Evidence reports that this probe saw the challenge value prefix at the
	// challenge name. The workflow computes it (hasChallengeEvidence) because
	// the probe STATE cannot: `match` implies it, but `mismatch` covers any
	// foreign record at that name too.
	Evidence bool
}

// RecordProbeActivity persists the latest per-record probe result onto the
// claim row. The query is guarded on status = 'pending' (zero rows is fine —
// the claim went terminal while we probed) and latches evidence_at with a
// COALESCE, so re-running this after a lost write neither clears nor moves an
// existing latch.
func (a *Activities) RecordProbeActivity(ctx context.Context, input RecordProbeInput) error {
	_, err := a.Q.RecordClaimProbe(ctx, dommod.RecordClaimProbeParams{
		ClaimID:         input.ClaimID,
		TxtState:        input.TXTState,
		TxtObserved:     strings.Join(input.TXTObserved, "\n"),
		ConnectState:    input.ConnectState,
		ConnectObserved: strings.Join(input.ConnectObserved, "\n"),
		Evidence:        input.Evidence,
	})
	return err
}

// MarkClaimInput identifies the claim a mark activity operates on.
type MarkClaimInput struct {
	ClaimID string
	// Evidence carries the final probe's evidence verdict into the expiry
	// write (design D2/D3). Ignored by the activation mark, which needs no
	// latch: a claim that verified is never charged an abandonment.
	Evidence bool
}

// MarkClaimActiveActivity moves a pending claim to active with verified_at
// set, and in the same statement clears this workspace's counted
// abandonments at the verified name and beneath it — proving control of a
// name wipes the slate for that name (design D3). Zero rows updated means the
// claim is no longer pending (expired, canceled, or already active) — a
// non-retryable state, not a transient failure.
func (a *Activities) MarkClaimActiveActivity(ctx context.Context, input MarkClaimInput) error {
	rows, err := a.Q.MarkClaimActive(ctx, input.ClaimID)
	if err != nil {
		return err
	}
	if rows == 0 {
		return temporal.NewNonRetryableApplicationError("claim is no longer pending", "ClaimNotPending", nil)
	}
	return nil
}

// MarkClaimExpiredActivity moves a pending claim to expired, carrying the
// final probe's evidence verdict so a lost probe write cannot cost the member
// an abandonment they did not earn. Zero rows updated is fine (already
// expired, canceled, or activated in a race).
func (a *Activities) MarkClaimExpiredActivity(ctx context.Context, input MarkClaimInput) error {
	_, err := a.Q.MarkClaimExpired(ctx, dommod.MarkClaimExpiredParams{
		ClaimID:  input.ClaimID,
		Evidence: input.Evidence,
	})
	return err
}

// VerifyClaimWorkflowInput is the input for VerifyClaimWorkflow.
type VerifyClaimWorkflowInput struct {
	ClaimID     string
	WorkspaceID string
	Root        string // the claim's root FQDN
	Token       string
	// ConnectTarget pins the deployment's connect target for this run's
	// diagnostic probe; empty falls back to the worker's configuration.
	ConnectTarget string
	ExpiresAt     time.Time
	// PollInterval carries the current backoff across ContinueAsNew; zero
	// (every fresh start) means initialPollInterval.
	PollInterval time.Duration
}

// VerifyClaimWorkflowOutput is the output for VerifyClaimWorkflow.
type VerifyClaimWorkflowOutput struct {
	Success      bool
	Verified     bool
	Root         string
	ErrorMessage string
}

// VerifyClaimWorkflow polls for the domain-control TXT challenge with
// backoff until it is found or the claim's window elapses. Each poll also
// probes the connect record and records both records' states on the claim
// row for the member-facing status view; only the TXT probe gates
// verification. A check-now signal probes immediately and resets the
// backoff; workflow cancellation (the member canceled the claim — the HTTP
// handler owns that terminal write) just exits. On TXT match the claim
// becomes active and the workflow stops: verification is decoupled from
// creation, so no provider resource is created here (design D4).
func VerifyClaimWorkflow(ctx workflow.Context, input VerifyClaimWorkflowInput) (*VerifyClaimWorkflowOutput, error) {
	logger := workflow.GetLogger(ctx)
	logger.Info("VerifyClaimWorkflow started",
		"root", input.Root,
		"claimID", input.ClaimID,
		"workspaceID", input.WorkspaceID)

	var activities *Activities
	dnsCtx := workflow.WithActivityOptions(ctx, DNSActivityOptions())
	storeCtx := workflow.WithActivityOptions(ctx, common.DefaultActivityOptions())
	signalCh := workflow.GetSignalChannel(ctx, CheckNowSignal)

	pollInterval := input.PollInterval
	if pollInterval <= 0 {
		pollInterval = initialPollInterval
	}
	for {
		if ctx.Err() != nil {
			return nil, temporal.NewCanceledError("claim verification canceled")
		}

		// Both probes in parallel; a failed probe activity degrades to an
		// error state rather than failing the loop.
		txtFuture := workflow.ExecuteActivity(dnsCtx, activities.CheckTXTActivity, CheckTXTInput{
			Root:  input.Root,
			Token: input.Token,
		})
		connectFuture := workflow.ExecuteActivity(dnsCtx, activities.CheckConnectActivity, CheckConnectInput{
			Root:   input.Root,
			Target: input.ConnectTarget,
		})
		var txt CheckTXTOutput
		if err := txtFuture.Get(ctx, &txt); err != nil {
			logger.Warn("TXT probe attempt failed", "error", err)
			txt = CheckTXTOutput{State: dommod.ProbeStateError}
		}
		var connect CheckConnectOutput
		if err := connectFuture.Get(ctx, &connect); err != nil {
			logger.Warn("connect probe attempt failed", "error", err)
			connect = CheckConnectOutput{State: dommod.ProbeStateError}
		}
		if ctx.Err() != nil {
			return nil, temporal.NewCanceledError("claim verification canceled")
		}

		// Evidence is read off the raw TXT answer by the probe activity, never
		// off the probe state, the capped observations, or the connect result
		// — see hasChallengeEvidence for the looser predicates and why each is
		// exploitable.
		if err := workflow.ExecuteActivity(storeCtx, activities.RecordProbeActivity, RecordProbeInput{
			ClaimID:         input.ClaimID,
			TXTState:        txt.State,
			TXTObserved:     txt.Observed,
			ConnectState:    connect.State,
			ConnectObserved: connect.Observed,
			Evidence:        txt.Evidence,
		}).Get(ctx, nil); err != nil {
			// Mostly display data, but the write also carries the evidence
			// latch — a policy fact. Verification still proceeds: the latch is
			// idempotent, so the next probe re-stamps what this one lost, and
			// nothing about proving control depends on it. The exception is
			// the last probe before the deadline, which has no next probe;
			// the expiry mark below carries txt.Evidence for exactly that.
			logger.Warn("failed to record probe result", "error", err)
		}

		if txt.State == dommod.ProbeStateMatch {
			break
		}

		// Expiry is checked AFTER probing, so a record published inside the
		// window always gets one final probe at the deadline (the wait below
		// is clamped to ExpiresAt) instead of dying unprobed in the last
		// backoff interval.
		now := workflow.Now(ctx)
		if !now.Before(input.ExpiresAt) {
			if err := workflow.ExecuteActivity(storeCtx, activities.MarkClaimExpiredActivity, MarkClaimInput{
				ClaimID: input.ClaimID,
				// What the probe immediately above saw. The latch's usual
				// recovery is "the next probe re-stamps it", and there is no
				// next probe: if that RecordProbeActivity write failed, this
				// is the only thing standing between a member who published
				// our challenge record and a ledger entry they did not earn.
				Evidence: txt.Evidence,
			}).Get(ctx, nil); err != nil {
				logger.Error("failed to mark claim expired", "error", err)
			}
			return &VerifyClaimWorkflowOutput{
				Success:      false,
				Root:         input.Root,
				ErrorMessage: "The verification window elapsed before the DNS record was found.",
			}, nil
		}

		// ~483 polls over a full 24-hour window (30s, 60s, 120s, then 180s
		// flat) × 3 activities + a timer each crosses the server's
		// suggest-continue-as-new history threshold, so roll into a fresh run
		// when asked, carrying the current backoff.
		if workflow.GetInfo(ctx).GetContinueAsNewSuggested() {
			next := input
			next.PollInterval = pollInterval
			return nil, workflow.NewContinueAsNewError(ctx, VerifyClaimWorkflow, next)
		}

		// Wait for the backoff timer, a check-now signal, or cancellation.
		// The timer is clamped so the final wake lands on the deadline.
		wait := pollInterval
		if until := input.ExpiresAt.Sub(now); until < wait {
			wait = until
		}
		timerCtx, cancelTimer := workflow.WithCancel(ctx)
		timer := workflow.NewTimer(timerCtx, wait)
		checkNow := false
		canceled := false
		selector := workflow.NewSelector(ctx)
		selector.AddFuture(timer, func(f workflow.Future) {})
		selector.AddReceive(signalCh, func(c workflow.ReceiveChannel, more bool) {
			c.Receive(ctx, nil)
			checkNow = true
		})
		selector.AddReceive(ctx.Done(), func(c workflow.ReceiveChannel, more bool) {
			canceled = true
		})
		selector.Select(ctx)
		cancelTimer()
		if canceled {
			return nil, temporal.NewCanceledError("claim verification canceled")
		}
		if checkNow {
			// Coalesce burst signals into this one probe, and tighten the
			// cadence — a member actively fixing DNS wants fast feedback.
			for signalCh.ReceiveAsync(nil) {
			}
			pollInterval = initialPollInterval
			continue
		}
		if pollInterval < maxPollInterval {
			pollInterval *= 2
			if pollInterval > maxPollInterval {
				pollInterval = maxPollInterval
			}
		}
	}

	if err := workflow.ExecuteActivity(storeCtx, activities.MarkClaimActiveActivity, MarkClaimInput{
		ClaimID: input.ClaimID,
	}).Get(ctx, nil); err != nil {
		// Includes the non-retryable "no longer pending" case: a cancel won
		// the race, so the terminal row stands and the workflow just stops.
		logger.Error("failed to mark claim active", "error", err)
		return &VerifyClaimWorkflowOutput{
			Success:      false,
			Root:         input.Root,
			ErrorMessage: "Verification could not be recorded. Please try again.",
		}, nil
	}

	logger.Info("VerifyClaimWorkflow completed", "root", input.Root)

	return &VerifyClaimWorkflowOutput{
		Success:  true,
		Verified: true,
		Root:     input.Root,
	}, nil
}