// 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 consume

import (
	"errors"
	"fmt"
	"io"
	"strings"
)

// ValidationError represents a flag-level user error. It wraps a clear,
// actionable message — the cobra command layer surfaces it to the user
// with exit code 2 (validation error).
type ValidationError struct{ Msg string }

func (e *ValidationError) Error() string { return e.Msg }

// validation sentinels (cobra layer uses errors.Is to set the exit code).
var (
	// ErrForceRequiresForeground is the plan §3.1 contract: --force only
	// makes sense in foreground mode where the bus runs in the current
	// process (then --force skips the single-instance lock so a second
	// foreground bus can co-exist for a brief debug window). Outside of
	// --foreground, --force would silently produce two daemons writing
	// to the same socket — refuse upfront.
	ErrForceRequiresForeground = &ValidationError{
		Msg: "--force is only meaningful with --foreground (in daemon mode it would produce multiple bus instances; cloud events would be randomly split across connections). To restart the bus: preview dws event stop --all --dry-run, confirm with dws event stop --all --yes, then run dws event consume",
	}

	// ErrJSONFormatRequiresBounded is the plan §3.1 contract: --format
	// json renders each event as a multi-line JSON object suitable for
	// human inspection. With an unbounded stream the output mixes events
	// without delimiters. Force the user to bound the run.
	ErrJSONFormatRequiresBounded = &ValidationError{
		Msg: "--format json requires --max-events or --duration (an unbounded JSON stream is not parseable). Use --format ndjson for unbounded streams.",
	}
)

// ValidateConfig performs all pre-flight validation that does not require
// disk / network I/O. Returns a *ValidationError for any rule violation;
// returns nil if the cfg is launchable. The cobra layer calls this BEFORE
// calling Run so the user gets clear errors at parse time.
//
// Rules implemented:
//  1. WorkDir / IPCEndpoint / ClientID non-empty
//  2. --force requires --foreground (plan §3.1)
//  3. --format json requires --max-events OR --duration (bounded)
//  4. Routes already pre-parsed (any parse error is reported by ParseRoutes)
//  5. --output-dir conflict with global --output (caller-supplied flag —
//     we expose ValidateNoOutputConflict separately because global -o is
//     a cobra-layer concern)
//
// Rules NOT enforced here (deferred to caller / Run):
//   - Credentials presence (auth.ResolveAppCredentialsStrict already
//     reports a typed error)
//   - bus availability (busctl.Discover handles)
func ValidateConfig(cfg Config) error {
	if strings.TrimSpace(cfg.WorkDir) == "" {
		return &ValidationError{Msg: "consume: WorkDir is required"}
	}
	if strings.TrimSpace(cfg.IPCEndpoint) == "" {
		return &ValidationError{Msg: "consume: IPCEndpoint is required"}
	}
	if strings.TrimSpace(cfg.ClientID) == "" {
		return &ValidationError{Msg: "consume: ClientID is required"}
	}
	if cfg.Force && !cfg.Foreground {
		return ErrForceRequiresForeground
	}
	if cfg.Format == FormatJSON && cfg.MaxEvents <= 0 && cfg.Duration <= 0 {
		return ErrJSONFormatRequiresBounded
	}
	return nil
}

// ValidateNoOutputConflict ensures --output-dir / --route (event-stream
// sinks) are not combined with the dws global hidden -o/--output flag
// (request-output to file). The cobra layer reads the global output flag
// from inherited flags and passes its value here; an empty globalOutput
// means the flag was unset.
func ValidateNoOutputConflict(cfg Config, globalOutput string) error {
	if globalOutput == "" {
		return nil
	}
	if cfg.OutputDir != "" || len(cfg.Routes) > 0 {
		return &ValidationError{
			Msg: fmt.Sprintf("--output-dir/--route cannot be combined with global -o/--output=%q (event stream sinks are mutually exclusive with single-file output capture)", globalOutput),
		}
	}
	return nil
}

// IsValidationError reports whether err is a flag-level user error.
// Cobra command handlers use this to map validation errors to exit code 2.
func IsValidationError(err error) bool {
	var v *ValidationError
	return errors.As(err, &v)
}

// PrintDryRun writes the resolved configuration to w in a single
// human-readable block. Called by Run when cfg.DryRun is true. Format
// avoids JSON so users can `dws event consume --dry-run | head` cleanly.
//
// RuntimeToken is the only secret-bearing Config field and is deliberately
// not read or rendered here. Keep this function allowlist-based: never switch
// it to generic struct serialization.
func PrintDryRun(w io.Writer, cfg Config) {
	if w == nil {
		return
	}
	fmt.Fprintln(w, "dws event consume — dry run (no bus connection will be made)")
	fmt.Fprintf(w, "  client_id        : %s\n", cfg.ClientID)
	fmt.Fprintf(w, "  workdir          : %s\n", cfg.WorkDir)
	fmt.Fprintf(w, "  ipc_endpoint     : %s\n", cfg.IPCEndpoint)
	if len(cfg.EventTypes) > 0 {
		fmt.Fprintf(w, "  event_types      : %s\n", strings.Join(cfg.EventTypes, ","))
	} else {
		fmt.Fprintln(w, "  event_types      : (catch-all)")
	}
	if cfg.Filter != "" {
		fmt.Fprintf(w, "  filter           : %s\n", cfg.Filter)
	}
	fmt.Fprintf(w, "  format           : %s\n", cfg.Format)
	fmt.Fprintf(w, "  flatten          : %v\n", cfg.Flatten)
	if cfg.OutputDir != "" {
		fmt.Fprintf(w, "  output_dir       : %s\n", cfg.OutputDir)
	}
	for i, r := range cfg.Routes {
		fmt.Fprintf(w, "  route[%d]         : %s\n", i, r.Raw)
	}
	if cfg.MaxEvents > 0 {
		fmt.Fprintf(w, "  max_events       : %d\n", cfg.MaxEvents)
	}
	if cfg.Duration > 0 {
		fmt.Fprintf(w, "  duration         : %s\n", cfg.Duration)
	}
	fmt.Fprintf(w, "  compact          : %v\n", cfg.Compact)
	fmt.Fprintf(w, "  quiet            : %v\n", cfg.Quiet)
	fmt.Fprintf(w, "  foreground       : %v\n", cfg.Foreground)
	fmt.Fprintf(w, "  force            : %v\n", cfg.Force)
}

// PrintDryRunMany renders the shared consume configuration plus every local
// logical consumer without opening the bus.
func PrintDryRunMany(w io.Writer, cfg Config, specs []ConsumerSpec) {
	preview := cfg
	preview.EventTypes = nil
	for _, spec := range specs {
		preview.EventTypes = append(preview.EventTypes, spec.EventTypes...)
	}
	PrintDryRun(w, preview)
	for i, spec := range specs {
		fmt.Fprintf(w, "  consumer[%d]      : event_key=%s subscribe_id=%s event_types=%s\n",
			i, spec.EventKey, displayDryRunValue(spec.SubscribeID), strings.Join(spec.EventTypes, ","))
	}
}

func displayDryRunValue(value string) string {
	if value = strings.TrimSpace(value); value != "" {
		return value
	}
	return "(pending)"
}