dingtalk-workspace-cli/internal/wait/wait.go

// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// Package wait is the framework terminal-state wait engine behind the
// reviewed contract.WaitSpec capability. It owns polling cadence, status
// extraction, and status→outcome mapping; it knows nothing about Cobra, MCP,
// or any product backend. How one poll executes is supplied by the leaf's
// WaitPoll hook (corecmd), so "poll = an existing read command" stays a leaf
// decision rather than a framework assumption.
package wait

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/corecmd/contract"
)

// DefaultPollInterval is the cadence between polls when LoopSpec.Interval is
// zero. The first poll runs immediately so an already-terminal resource does
// not pay a sleep tax.
const DefaultPollInterval = 2 * time.Second

// MaxPollInterval caps the exponential backoff growth between polls so a long
// wait cannot degenerate into effectively-blind polling.
const MaxPollInterval = 30 * time.Second

// PollDoc is one decoded poll response document (typically the unified-output
// envelope data of the poll command).
type PollDoc map[string]any

// Poller executes one poll. Returning an error fails the wait phase; the
// engine never retries a poller error because read commands failing is a real
// failure, not a "not yet" signal.
type Poller func(ctx context.Context) (PollDoc, error)

// LoopSpec is the runtime-resolved projection of contract.WaitSpec plus the
// caller-provided timeout.
type LoopSpec struct {
	StatusQuery string
	Terminal    map[string]contract.ResultOutcome
	Pending     []string
	Timeout     time.Duration
	Interval    time.Duration
}

// Outcome is the closed result of a wait loop. TimedOut reports deadline
// exhaustion (Outcome is then pending — an accepted-but-not-terminal state is
// not a process failure per the exit-code contract); Status is the last
// observed status value.
type Outcome struct {
	Status   string
	Outcome  contract.ResultOutcome
	Attempts int
	TimedOut bool
}

// ErrUnknownStatus reports a status value that is neither declared terminal
// nor declared pending. Unknown fails closed: mapping it to pending could
// hide a real state change until timeout, mapping it to success is worse.
type ErrUnknownStatus struct {
	Status string
	Query  string
}

func (e *ErrUnknownStatus) Error() string {
	return fmt.Sprintf("wait: status %q (from %q) is neither terminal nor pending", e.Status, e.Query)
}

// Run polls poller until a declared terminal status, deadline exhaustion, or
// a poller error. The first poll is immediate; subsequent polls back off
// exponentially (×1.5) from Interval, capped at MaxPollInterval. Deadline
// exhaustion anywhere — before a poll, during a poll (a context-aware poller
// returns ctx.Err()), or during the wait between polls — always closes as
// timed-out pending with the last observed status, never as a poll failure.
func Run(ctx context.Context, spec LoopSpec, poll Poller) (Outcome, error) {
	if spec.Interval <= 0 {
		spec.Interval = DefaultPollInterval
	}
	if spec.Timeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, spec.Timeout)
		defer cancel()
	}
	pending := make(map[string]bool, len(spec.Pending))
	for _, value := range spec.Pending {
		pending[value] = true
	}
	timedOut := func(status string, attempts int) Outcome {
		return Outcome{Status: status, Outcome: contract.ResultOutcomePending, Attempts: attempts, TimedOut: true}
	}
	interval := spec.Interval
	attempts := 0
	lastStatus := ""
	for {
		if ctx.Err() != nil {
			return timedOut(lastStatus, attempts), nil
		}
		doc, err := poll(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return timedOut(lastStatus, attempts), nil
			}
			return Outcome{Attempts: attempts}, fmt.Errorf("wait: poll failed: %w", err)
		}
		attempts++
		status, ok := ExtractStatus(doc, spec.StatusQuery)
		if !ok {
			return Outcome{Attempts: attempts}, fmt.Errorf(
				"wait: status query %q not found in poll result", spec.StatusQuery)
		}
		lastStatus = status
		if outcome, ok := spec.Terminal[status]; ok {
			return Outcome{Status: status, Outcome: outcome, Attempts: attempts}, nil
		}
		if !pending[status] {
			return Outcome{Status: status, Attempts: attempts}, &ErrUnknownStatus{Status: status, Query: spec.StatusQuery}
		}
		timer := time.NewTimer(interval)
		select {
		case <-ctx.Done():
			timer.Stop()
			return timedOut(status, attempts), nil
		case <-timer.C:
		}
		interval = nextInterval(interval)
	}
}

func nextInterval(current time.Duration) time.Duration {
	next := current * 3 / 2
	if next > MaxPollInterval {
		next = MaxPollInterval
	}
	return next
}

// ExtractStatus resolves a dotted status query against a poll document. Each
// segment walks one map level; array indexes are not supported because wait
// targets a single resource. Numeric segments are stringified, so a document
// decoded with json.Number keys still resolves.
func ExtractStatus(doc PollDoc, query string) (string, bool) {
	query = strings.TrimSpace(query)
	if query == "" {
		return "", false
	}
	// PollDoc is a defined type, so its dynamic type does not satisfy a
	// map[string]any assertion — convert once at the boundary; nested values
	// from JSON decoding are plain maps.
	var current any = map[string]any(doc)
	for _, segment := range strings.Split(query, ".") {
		segment = strings.TrimSpace(segment)
		if segment == "" {
			return "", false
		}
		node, ok := current.(map[string]any)
		if !ok {
			return "", false
		}
		value, ok := node[segment]
		if !ok {
			return "", false
		}
		current = value
	}
	switch value := current.(type) {
	case string:
		return value, true
	case fmt.Stringer:
		return value.String(), true
	case bool:
		return strconv.FormatBool(value), true
	case int:
		return strconv.Itoa(value), true
	case int64:
		return strconv.FormatInt(value, 10), true
	case float64:
		return strconv.FormatFloat(value, 'f', -1, 64), true
	default:
		return "", false
	}
}

// IsUnknownStatus reports whether err is the closed fail-on-unknown error.
func IsUnknownStatus(err error) bool {
	var unknown *ErrUnknownStatus
	return errors.As(err, &unknown)
}

// EventStream is the leaf-owned push subscription consumed by the event
// phase (the WaitEvents hook in corecmd). Recv delivers the next decoded
// event document; it returns an error or io.EOF-style termination when the
// stream ends — the engine treats non-terminal termination as a stream
// failure the caller (auto mode) may fall back from.
type EventStream interface {
	Recv(ctx context.Context) (PollDoc, error)
}

// EventLoopSpec is the event-phase projection of contract.WaitSpec.
type EventLoopSpec struct {
	StatusQuery string
	MatchField  string
	Terminal    map[string]contract.ResultOutcome
	Pending     []string
	Timeout     time.Duration
}

// ErrEventStreamEnded reports a stream that terminated before a terminal
// status. Auto mode uses it to fall back to polling; strict event mode
// surfaces it as a wait failure.
var ErrEventStreamEnded = errors.New("wait: event stream ended before a terminal status")

// RunEvent consumes stream until a correlated event reaches a declared
// terminal status, the deadline exhausts, or the stream ends. Events whose
// MatchField value does not equal resource are ignored (other resources on
// the same channel); a correlated event with an unknown status fails closed
// exactly like a poll would.
func RunEvent(ctx context.Context, spec EventLoopSpec, resource string, stream EventStream) (Outcome, error) {
	if spec.Timeout > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, spec.Timeout)
		defer cancel()
	}
	pending := make(map[string]bool, len(spec.Pending))
	for _, value := range spec.Pending {
		pending[value] = true
	}
	attempts := 0
	lastStatus := ""
	for {
		doc, err := stream.Recv(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return Outcome{Status: lastStatus, Outcome: contract.ResultOutcomePending, Attempts: attempts, TimedOut: true}, nil
			}
			// Wrap with ErrEventStreamEnded so auto mode can fall back to
			// polling while correlated-status failures (unknown status,
			// missing status query) stay non-recoverable.
			return Outcome{Attempts: attempts}, fmt.Errorf("%w: %v", ErrEventStreamEnded, err)
		}
		attempts++
		correlated, ok := ExtractStatus(doc, spec.MatchField)
		if !ok || correlated != resource {
			continue
		}
		status, ok := ExtractStatus(doc, spec.StatusQuery)
		if !ok {
			return Outcome{Attempts: attempts}, fmt.Errorf(
				"wait: status query %q not found in event document", spec.StatusQuery)
		}
		lastStatus = status
		if outcome, ok := spec.Terminal[status]; ok {
			return Outcome{Status: status, Outcome: outcome, Attempts: attempts}, nil
		}
		if !pending[status] {
			return Outcome{Status: status, Attempts: attempts}, &ErrUnknownStatus{Status: status, Query: spec.StatusQuery}
		}
	}
}