// Copyright (C) 2026 Yota Hamada
// SPDX-License-Identifier: GPL-3.0-or-later

package cmd

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"os/signal"
	"path/filepath"
	"runtime/pprof"
	"strings"
	"syscall"
	"time"

	"golang.org/x/term"

	"github.com/dagucloud/dagu/internal/clicontext"
	cmdprocess "github.com/dagucloud/dagu/internal/cmd/process"
	"github.com/dagucloud/dagu/internal/cmn/config"
	"github.com/dagucloud/dagu/internal/cmn/crypto"
	"github.com/dagucloud/dagu/internal/cmn/fileutil"
	"github.com/dagucloud/dagu/internal/cmn/logger"
	"github.com/dagucloud/dagu/internal/cmn/logger/tag"
	"github.com/dagucloud/dagu/internal/cmn/logpath"
	"github.com/dagucloud/dagu/internal/cmn/signalctx"
	"github.com/dagucloud/dagu/internal/cmn/stringutil"
	"github.com/dagucloud/dagu/internal/core"
	"github.com/dagucloud/dagu/internal/core/exec"
	"github.com/dagucloud/dagu/internal/license"
	"github.com/dagucloud/dagu/internal/persis/filebaseconfig"
	"github.com/dagucloud/dagu/internal/persis/filedagrun"
	"github.com/dagucloud/dagu/internal/persis/filedistributed"
	"github.com/dagucloud/dagu/internal/persis/fileeventstore"
	"github.com/dagucloud/dagu/internal/persis/filelicense"
	"github.com/dagucloud/dagu/internal/persis/fileproc"
	"github.com/dagucloud/dagu/internal/persis/filequeue"
	"github.com/dagucloud/dagu/internal/persis/fileserviceregistry"
	"github.com/dagucloud/dagu/internal/runtime"
	"github.com/dagucloud/dagu/internal/runtime/transform"
	"github.com/dagucloud/dagu/internal/service/coordinator"
	"github.com/dagucloud/dagu/internal/service/eventstore"
	"github.com/dagucloud/dagu/internal/service/frontend"
	"github.com/dagucloud/dagu/internal/service/resource"
	"github.com/dagucloud/dagu/internal/service/scheduler"
	"github.com/google/uuid"
	"github.com/spf13/cobra"
	"github.com/spf13/viper"
)

// Context holds the configuration for a command.
type Context struct {
	context.Context

	Command *cobra.Command
	Flags   []commandLineFlag
	Config  *config.Config
	Quiet   bool
	Scope   commandScope

	EventService              *eventstore.Service
	EventSourceInstance       string
	DAGRunStore               exec.DAGRunStore
	DAGRunMgr                 runtime.Manager
	ProcStore                 exec.ProcStore
	QueueStore                exec.QueueStore
	ServiceRegistry           exec.ServiceRegistry
	DispatchTaskStore         exec.DispatchTaskStore
	WorkerHeartbeatStore      exec.WorkerHeartbeatStore
	DAGRunLeaseStore          exec.DAGRunLeaseStore
	ActiveDistributedRunStore exec.ActiveDistributedRunStore

	Proc           exec.ProcHandle
	LicenseManager *license.Manager
	ContextStore   *clicontext.Store
	CLIContext     *clicontext.Context
	ContextName    string
	Remote         *remoteClient
}

// WithContext returns a new Context with a different underlying context.Context.
// This is useful for creating a signal-aware context for service operations.
func (c *Context) WithContext(ctx context.Context) *Context {
	return &Context{
		Context:                   ctx,
		Command:                   c.Command,
		Flags:                     c.Flags,
		Config:                    c.Config,
		Quiet:                     c.Quiet,
		EventService:              c.EventService,
		EventSourceInstance:       c.EventSourceInstance,
		DAGRunStore:               c.DAGRunStore,
		DAGRunMgr:                 c.DAGRunMgr,
		ProcStore:                 c.ProcStore,
		QueueStore:                c.QueueStore,
		ServiceRegistry:           c.ServiceRegistry,
		DispatchTaskStore:         c.DispatchTaskStore,
		WorkerHeartbeatStore:      c.WorkerHeartbeatStore,
		DAGRunLeaseStore:          c.DAGRunLeaseStore,
		ActiveDistributedRunStore: c.ActiveDistributedRunStore,
		Proc:                      c.Proc,
		LicenseManager:            c.LicenseManager,
		ContextStore:              c.ContextStore,
		CLIContext:                c.CLIContext,
		ContextName:               c.ContextName,
		Remote:                    c.Remote,
		Scope:                     c.Scope,
	}
}

// WithEventSource returns a shallow copy whose context carries the given event source.
// If the event store is not configured, the original context is preserved.
func (c *Context) WithEventSource(service string) *Context {
	if c == nil || c.EventService == nil {
		return c
	}
	return c.WithContext(eventstore.WithContext(c.Context, c.EventService, eventstore.Source{
		Service:  service,
		Instance: c.EventSourceInstance,
	}))
}

// LogToFile creates a new logger context with a file writer.
func (c *Context) LogToFile(f *os.File) {
	var opts []logger.Option
	if c.Config.Core.Debug {
		opts = append(opts, logger.WithDebug())
	}
	if c.Quiet {
		opts = append(opts, logger.WithQuiet())
	}
	if c.Config.Core.LogFormat != "" {
		opts = append(opts, logger.WithFormat(c.Config.Core.LogFormat))
	}
	if f != nil {
		opts = append(opts, logger.WithWriter(f))
	}
	c.Context = logger.WithLogger(c.Context, logger.NewLogger(opts...))
}

// NewContext creates and initializes an application Context for the given Cobra command.
// It binds command flags, loads configuration scoped to the command, configures logging
// (respecting debug, quiet, and log format settings), logs any configuration warnings,
// and initializes history, DAG run, proc, queue, and service registry stores and managers.
// Returns an initialized Context or an error if flag retrieval, configuration loading,
// or other initialization steps fail.
func NewContext(cmd *cobra.Command, flags []commandLineFlag) (*Context, error) {
	ctx := cmd.Context()
	commandName := commandFamilyName(cmd)
	scope := scopeForCommand(commandName)

	v := viper.New()
	bindFlags(v, cmd, flags...)

	quiet, err := cmd.Flags().GetBool("quiet")
	if err != nil {
		return nil, fmt.Errorf("failed to get quiet flag: %w", err)
	}
	daguHome, err := cmd.Flags().GetString("dagu-home")
	if err != nil {
		return nil, fmt.Errorf("failed to get dagu-home flag: %w", err)
	}

	var configLoaderOpts []config.ConfigLoaderOption
	if daguHome != "" {
		if resolvedHome := fileutil.ResolvePathOrBlank(daguHome); resolvedHome != "" {
			configLoaderOpts = append(configLoaderOpts, config.WithAppHomeDir(resolvedHome))
		}
	}

	// Use a custom config file if provided via the command flag "config"
	cfgPath, err := cmd.Flags().GetString("config")
	if err != nil {
		return nil, fmt.Errorf("failed to get config flag: %w", err)
	}
	if cfgPath != "" {
		configLoaderOpts = append(configLoaderOpts, config.WithConfigFile(cfgPath))
	}

	// Set service type based on command to load only necessary config sections
	configLoaderOpts = append(configLoaderOpts, config.WithService(serviceForCommand(commandName)))

	loader := config.NewConfigLoader(v, configLoaderOpts...)
	cfg, err := loader.Load()
	if err != nil {
		return nil, err
	}
	ctx = config.WithConfig(ctx, cfg)

	requestedContextName, err := requestedCLIContextName(cmd)
	if err != nil {
		return nil, err
	}
	selectedContextName := clicontext.LocalContextName
	selectedContext := &clicontext.Context{Name: clicontext.LocalContextName}
	var (
		contextStore        *clicontext.Store
		contextStoreWarning error
	)

	if isContextCommand(cmd) || scope != commandScopeStatic {
		contextStore, err = newCLIContextStore(cfg.Paths.DataDir, cfg.Paths.ContextsDir)
		if err != nil {
			if shouldFailForContextStoreError(cmd, scope, requestedContextName) {
				return nil, fmt.Errorf("failed to initialize context store: %w", err)
			}
			contextStoreWarning = fmt.Errorf("failed to initialize context store, using local context: %w", err)
		} else if !isContextCommand(cmd) {
			selectedContextName, selectedContext, err = resolveCLIContext(cmd, contextStore, requestedContextName)
			if err != nil {
				if shouldFailForContextResolutionError(scope, requestedContextName) {
					return nil, err
				}
				contextStoreWarning = fmt.Errorf("failed to resolve context selection, using local context: %w", err)
				selectedContextName = clicontext.LocalContextName
				selectedContext = &clicontext.Context{Name: clicontext.LocalContextName}
			}
		}
	}
	if scope == commandScopeLocalOnly && selectedContextName != clicontext.LocalContextName {
		return nil, fmt.Errorf("command %q only supports the local context", cmd.Name())
	}

	// Create a logger context based on config and quiet mode
	var opts []logger.Option
	if cfg.Core.Debug || os.Getenv("DEBUG") != "" {
		opts = append(opts, logger.WithDebug())
	}
	if quiet {
		opts = append(opts, logger.WithQuiet())
	}
	// For agent commands running in a terminal, suppress console output early
	// to avoid debug logs cluttering the progress display or tree output
	if !quiet && isAgentCommand(cmd.Name()) && term.IsTerminal(int(os.Stderr.Fd())) && os.Getenv("DISABLE_PROGRESS") == "" {
		opts = append(opts, logger.WithQuiet())
	}
	if cfg.Core.LogFormat != "" {
		opts = append(opts, logger.WithFormat(cfg.Core.LogFormat))
	}
	ctx = logger.WithLogger(ctx, logger.NewLogger(opts...))

	// Log any warnings collected during configuration loading
	for _, notice := range cfg.Notices {
		logger.Info(ctx, notice)
	}
	for _, warning := range cfg.Warnings {
		logger.Warn(ctx, warning)
	}
	if contextStoreWarning != nil {
		logger.Warn(ctx, contextStoreWarning.Error())
	}

	baseCtx := ctx
	eventSourceInstance := eventstore.DefaultSourceInstance()
	var eventSvc *eventstore.Service
	sharedNothingWorker := isSharedNothingWorker(cmd, cfg)
	if !sharedNothingWorker && cfg.EventStore.Enabled {
		store, eventErr := fileeventstore.New(cfg.Paths.EventStoreDir)
		if eventErr != nil {
			logger.Warn(ctx, "Failed to initialize event store; continuing without event persistence", tag.Error(eventErr))
		} else {
			eventSvc = eventstore.New(store)
			ctx = eventstore.WithContext(ctx, eventSvc, eventstore.Source{
				Service:  eventSourceServiceForCommand(cmd.Name()),
				Instance: eventSourceInstance,
			})
		}
	}

	if scope == commandScopeContextAware && selectedContextName != clicontext.LocalContextName {
		remote, err := newRemoteClient(selectedContext)
		if err != nil {
			return nil, fmt.Errorf("failed to initialize remote context %q: %w", selectedContextName, err)
		}
		return &Context{
			Context:             ctx,
			Command:             cmd,
			Config:              cfg,
			Quiet:               quiet,
			Flags:               flags,
			EventService:        eventSvc,
			EventSourceInstance: eventSourceInstance,
			ContextStore:        contextStore,
			CLIContext:          selectedContext,
			ContextName:         selectedContextName,
			Remote:              remote,
			Scope:               scope,
		}, nil
	}

	// For shared-nothing workers, skip creating file-based stores
	// as they only use temporary directories and push status to coordinator
	if sharedNothingWorker {
		logger.Debug(ctx, "Shared-nothing worker mode: skipping file-based stores",
			slog.Any("coordinators", cfg.Worker.Coordinators),
		)
		return &Context{
			Context:             baseCtx,
			Command:             cmd,
			Config:              cfg,
			Quiet:               quiet,
			Flags:               flags,
			EventService:        nil,
			EventSourceInstance: eventSourceInstance,
			ContextStore:        contextStore,
			CLIContext:          selectedContext,
			ContextName:         selectedContextName,
			Scope:               scope,
			// All stores are nil - shared-nothing workers don't need local storage
			// Status is pushed to coordinator, DAG definitions come from task payload
		}, nil
	}

	// Initialize history repository and history manager
	hrOpts := []filedagrun.DAGRunStoreOption{
		filedagrun.WithArtifactDir(cfg.Paths.ArtifactDir),
		filedagrun.WithLatestStatusToday(cfg.Server.LatestStatusToday),
		filedagrun.WithLocation(cfg.Core.Location),
	}

	switch cmd.Name() {
	case "server", "scheduler", "start-all", "coordinator":
		// For long-running process, we setup file cache for better performance
		limits := cfg.Cache.Limits()
		hc := fileutil.NewCache[*exec.DAGRunStatus]("dag_run_status", limits.DAGRun.Limit, limits.DAGRun.TTL)
		hc.StartEviction(ctx)
		hrOpts = append(hrOpts, filedagrun.WithHistoryFileCache(hc))
	}

	ps := fileproc.New(cfg.Paths.ProcDir,
		fileproc.WithStaleThreshold(cfg.Proc.StaleThreshold),
		fileproc.WithHeartbeatInterval(cfg.Proc.HeartbeatInterval),
		fileproc.WithHeartbeatSyncInterval(cfg.Proc.HeartbeatSyncInterval),
	)
	if err := ps.Validate(ctx); err != nil {
		return nil, fmt.Errorf("failed to validate proc directory %s: %w", cfg.Paths.ProcDir, err)
	}
	drs := filedagrun.New(cfg.Paths.DAGRunsDir, hrOpts...)
	distributedDir := filepath.Join(cfg.Paths.DataDir, "distributed")
	dagRunLeaseStore := filedistributed.NewDAGRunLeaseStore(distributedDir)
	activeDistributedRunStore := filedistributed.NewActiveDistributedRunStore(distributedDir)
	drm := runtime.NewManager(drs, ps, cfg)
	qs := filequeue.New(cfg.Paths.QueueDir)
	sm := fileserviceregistry.New(cfg.Paths.ServiceRegistryDir)
	dispatchTaskStore := filedistributed.NewDispatchTaskStore(distributedDir)
	workerHeartbeatStore := filedistributed.NewWorkerHeartbeatStore(distributedDir)

	// Initialize license manager for server commands
	var licMgr *license.Manager
	switch cmd.Name() {
	case "server", "start-all":
		pubKey, pubKeyErr := license.PublicKey()
		if pubKeyErr != nil {
			logger.Warn(ctx, "Failed to load license public key", tag.Error(pubKeyErr))
			break
		}
		licenseDir := filepath.Join(cfg.Paths.DataDir, "license")
		licStore := filelicense.New(licenseDir)
		licMgr = license.NewManager(license.ManagerConfig{
			LicenseDir: licenseDir,
			ConfigKey:  cfg.License.Key,
			CloudURL:   cfg.License.CloudURL,
		}, pubKey, licStore, slog.Default())
		if err := licMgr.Start(ctx); err != nil {
			logger.Warn(ctx, "License manager initialization failed", tag.Error(err))
		}
	}

	// Log key configuration settings for debugging
	logger.Debug(ctx, "Configuration loaded",
		tag.Config(cfg.Paths.ConfigFileUsed),
		tag.Dir(cfg.Paths.DAGsDir),
	)
	logger.Debug(ctx, "Paths configuration",
		slog.String("log-dir", cfg.Paths.LogDir),
		slog.String("data-dir", cfg.Paths.DataDir),
		slog.String("dag-runs-dir", cfg.Paths.DAGRunsDir),
	)

	// Initialize default base config if it doesn't exist
	if cfg.Paths.BaseConfig != "" {
		bcStore, bcErr := filebaseconfig.New(cfg.Paths.BaseConfig,
			filebaseconfig.WithSkipDefault(cfg.Core.SkipExamples),
		)
		if bcErr != nil {
			logger.Warn(ctx, "Failed to create base config store", tag.Error(bcErr))
		} else {
			if initErr := bcStore.Initialize(); initErr != nil {
				logger.Warn(ctx, "Failed to initialize default base config", tag.Error(initErr))
			}
		}
	}

	return &Context{
		Context:                   ctx,
		Command:                   cmd,
		Config:                    cfg,
		Quiet:                     quiet,
		EventService:              eventSvc,
		EventSourceInstance:       eventSourceInstance,
		DAGRunStore:               drs,
		DAGRunMgr:                 drm,
		Flags:                     flags,
		ProcStore:                 ps,
		QueueStore:                qs,
		ServiceRegistry:           sm,
		DispatchTaskStore:         dispatchTaskStore,
		WorkerHeartbeatStore:      workerHeartbeatStore,
		DAGRunLeaseStore:          dagRunLeaseStore,
		ActiveDistributedRunStore: activeDistributedRunStore,
		LicenseManager:            licMgr,
		ContextStore:              contextStore,
		CLIContext:                selectedContext,
		ContextName:               selectedContextName,
		Scope:                     scope,
	}, nil
}

func newCLIContextStore(dataDir, contextsDir string) (*clicontext.Store, error) {
	encKey, err := crypto.ResolveKey(dataDir)
	if err != nil {
		return nil, err
	}
	enc, err := crypto.NewEncryptor(encKey)
	if err != nil {
		return nil, err
	}
	return clicontext.NewStore(contextsDir, enc)
}

func commandFamilyName(cmd *cobra.Command) string {
	if isContextCommand(cmd) {
		return "context"
	}
	if isAgentCLICommand(cmd) {
		return "agent"
	}
	return cmd.Name()
}

func isContextCommand(cmd *cobra.Command) bool {
	for current := cmd; current != nil; current = current.Parent() {
		if current.Name() == "context" {
			return true
		}
	}
	return false
}

func isAgentCLICommand(cmd *cobra.Command) bool {
	for current := cmd; current != nil; current = current.Parent() {
		if current.Name() == "agent" {
			return true
		}
	}
	return false
}

func requestedCLIContextName(cmd *cobra.Command) (string, error) {
	if cmd.Flags().Lookup("context") == nil {
		return "", nil
	}
	contextName, err := cmd.Flags().GetString("context")
	if err != nil {
		return "", fmt.Errorf("failed to get context flag: %w", err)
	}
	return strings.TrimSpace(contextName), nil
}

func resolveCLIContext(cmd *cobra.Command, store *clicontext.Store, requested string) (string, *clicontext.Context, error) {
	contextName := strings.TrimSpace(requested)
	var err error
	if contextName == "" {
		contextName, err = store.Current(cmd.Context())
		if err != nil {
			return "", nil, fmt.Errorf("failed to resolve current context: %w", err)
		}
	}
	if contextName == "" {
		contextName = clicontext.LocalContextName
	}
	ctx, err := store.Get(cmd.Context(), contextName)
	if err != nil {
		return "", nil, fmt.Errorf("failed to resolve context %q: %w", contextName, err)
	}
	return contextName, ctx, nil
}

func shouldFailForContextStoreError(cmd *cobra.Command, scope commandScope, requested string) bool {
	if isContextCommand(cmd) {
		return true
	}
	if scope == commandScopeStatic {
		return false
	}
	return requested != "" && requested != clicontext.LocalContextName
}

func shouldFailForContextResolutionError(scope commandScope, requested string) bool {
	if requested == "" {
		return false
	}
	if requested == clicontext.LocalContextName {
		return false
	}
	return scope != commandScopeStatic
}

func (c *Context) IsRemote() bool {
	return c != nil && c.Remote != nil && c.ContextName != clicontext.LocalContextName
}

func eventSourceServiceForCommand(cmdName string) string {
	switch cmdName {
	case "scheduler":
		return eventstore.SourceServiceScheduler
	case "server":
		return eventstore.SourceServiceServer
	case "coordinator":
		return eventstore.SourceServiceCoordinator
	default:
		return eventstore.SourceServiceCLI
	}
}

// serviceForCommand determines which config.Service to load for a given command name.
// Returns the appropriate service type for the command, or ServiceNone to load all config.
func serviceForCommand(cmdName string) config.Service {
	switch cmdName {
	case "server":
		return config.ServiceServer
	case "scheduler":
		return config.ServiceScheduler
	case "worker":
		return config.ServiceWorker
	case "coordinator":
		return config.ServiceCoordinator
	case "start", "restart", "retry", "dry", "exec", "agent":
		return config.ServiceAgent
	default:
		// For all other commands (status, stop, validate, etc.), load all config
		return config.ServiceNone
	}
}

// isAgentCommand returns true if the command name is an agent command
// that displays progress or tree output.
func isAgentCommand(cmdName string) bool {
	switch cmdName {
	case "start", "restart", "retry", "dry", "exec", "agent":
		return true
	default:
		return false
	}
}

// isSharedNothingWorker checks if the current command is a worker with static coordinators
// configured, indicating shared-nothing mode where no local storage is needed.
func isSharedNothingWorker(cmd *cobra.Command, cfg *config.Config) bool {
	if cmd.Name() != "worker" {
		return false
	}
	return len(cfg.Worker.Coordinators) > 0
}

// NewServer creates and returns a new web UI server for this command context.
func (c *Context) NewServer(rs *resource.Service, opts ...frontend.ServerOption) (*frontend.Server, error) {
	return cmdprocess.NewServer(cmdprocess.ServerConfig{
		Context:              c.Context,
		Config:               c.Config,
		DAGRunStore:          c.DAGRunStore,
		QueueStore:           c.QueueStore,
		ProcStore:            c.ProcStore,
		DAGRunManager:        c.DAGRunMgr,
		ServiceRegistry:      c.ServiceRegistry,
		DAGRunLeaseStore:     c.DAGRunLeaseStore,
		WorkerHeartbeatStore: c.WorkerHeartbeatStore,
		LicenseManager:       c.LicenseManager,
		ResourceService:      rs,
	}, opts...)
}

// NewCoordinatorClient creates a new coordinator client using the global peer configuration.
// Returns nil when the coordinator is disabled via configuration.
func (c *Context) NewCoordinatorClient() coordinator.Client {
	return cmdprocess.NewCoordinatorClient(c.Context, c.Config, c.ServiceRegistry)
}

// NewScheduler creates a scheduler for this command context.
func (c *Context) NewScheduler() (*scheduler.Scheduler, error) {
	return cmdprocess.NewScheduler(cmdprocess.SchedulerConfig{
		Context:           c.Context,
		Config:            c.Config,
		QueueStore:        c.QueueStore,
		ProcStore:         c.ProcStore,
		ServiceRegistry:   c.ServiceRegistry,
		DispatchTaskStore: c.DispatchTaskStore,
		DAGRunLeaseStore:  c.DAGRunLeaseStore,
		EventService:      c.EventService,
		LicenseManager:    c.LicenseManager,
	})
}

// StringParam retrieves a string parameter from the command line flags.
// It checks if the parameter is wrapped in quotes and removes them if necessary.
func (c *Context) StringParam(name string) (string, error) {
	val, err := c.Command.Flags().GetString(name)
	if err != nil {
		return "", fmt.Errorf("failed to get flag %s: %w", name, err)
	}

	// If it's wrapped in quotes, remove them
	val = stringutil.RemoveQuotes(val)
	return val, nil
}

// getWorkerID retrieves the worker ID from context, defaulting to "local" if not set or on error.
func getWorkerID(ctx *Context) string {
	workerID, err := ctx.StringParam("worker-id")
	if err != nil {
		logger.Warn(ctx, "Failed to read worker-id flag, defaulting to 'local'", tag.Error(err))
		return "local"
	}
	if workerID == "" {
		return "local"
	}
	return workerID
}

// dagStoreConfig contains options for creating a DAG store.
type dagStoreConfig struct {
	Cache                 *fileutil.Cache[*core.DAG] // Optional cache for DAG objects
	SearchPaths           []string                   // Additional search paths for DAG files
	SkipDirectoryCreation bool                       // Skip directory creation (for distributed worker execution)
}

// dagStore returns a new DAGRepository instance.
func (c *Context) dagStore(cfg dagStoreConfig) (exec.DAGStore, error) {
	return cmdprocess.NewDAGStore(c.Config, cmdprocess.DAGStoreConfig{
		Cache:                 cfg.Cache,
		SearchPaths:           cfg.SearchPaths,
		SkipDirectoryCreation: cfg.SkipDirectoryCreation,
	})
}

// agentStoresResult holds the agent stores created by agentStores().
type agentStoresResult = cmdprocess.AgentStores

// agentStores creates the agent store bundle for this command context.
func (c *Context) agentStores() agentStoresResult {
	return cmdprocess.NewAgentStores(c.Context, c.Config, c.ContextStore)
}

// OpenLogFile creates and opens a log file for a given dag-run.
// It evaluates the log directory, validates settings, creates the log directory,
// builds a filename using the current timestamp and dag-run ID, and then opens the file.
func (c *Context) OpenLogFile(
	dag *core.DAG,
	dagRunID string,
) (*os.File, error) {
	logPath, err := c.GenLogFileName(dag, dagRunID)
	if err != nil {
		return nil, fmt.Errorf("failed to generate log file name: %w", err)
	}
	return fileutil.OpenOrCreateFile(logPath)
}

// GenLogFileName generates a log file name based on the DAG and dag-run ID.
func (c *Context) GenLogFileName(dag *core.DAG, dagRunID string) (string, error) {
	return logpath.Generate(c, c.Config.Paths.LogDir, dag.LogDir, dag.Name, dagRunID)
}

// GenArtifactDir generates an artifact directory path for the DAG run when artifacts are enabled.
func (c *Context) GenArtifactDir(dag *core.DAG, dagRunID string) (string, error) {
	if dag == nil || !dag.ArtifactsEnabled() {
		return "", nil
	}

	dagArtifactDir := ""
	if dag.Artifacts != nil {
		dagArtifactDir = dag.Artifacts.Dir
	}

	return logpath.GenerateDir(c, c.Config.Paths.ArtifactDir, dagArtifactDir, dag.Name, dagRunID)
}

// NewCommand creates a new command instance with the given cobra command and run function.
func NewCommand(cmd *cobra.Command, flags []commandLineFlag, runFunc func(cmd *Context, args []string) error) *cobra.Command {
	initFlags(cmd, flags...)

	cmd.SilenceUsage = true

	cmd.RunE = func(cmd *cobra.Command, args []string) error {
		// Setup cpu profiling if enabled.
		cpuProfileEnabled, err := cmd.Flags().GetBool("cpu-profile")
		if err != nil {
			return fmt.Errorf("failed to read cpu-profile flag: %w", err)
		}
		if cpuProfileEnabled {
			f, err := os.Create("cpu.prof")
			if err != nil {
				return fmt.Errorf("failed to create CPU profile file: %w", err)
			}
			_ = pprof.StartCPUProfile(f)
			defer func() {
				pprof.StopCPUProfile()
				if err := f.Close(); err != nil {
					fmt.Printf("Failed to close CPU profile file: %v\n", err)
				}
			}()
		}

		ctx, err := NewContext(cmd, flags)

		if err != nil {
			return fmt.Errorf("initialization error: %w", err)
		}
		return runFunc(ctx, args)
	}

	return cmd
}

// genRunID creates a new UUID string to be used as a dag-run IDentifier.
func genRunID() (string, error) {
	id, err := uuid.NewV7()
	if err != nil {
		return "", err
	}
	return id.String(), nil
}

// validateRunID checks if the dag-run ID is valid and not empty.
func validateRunID(dagRunID string) error {
	return exec.ValidateDAGRunID(dagRunID)
}

// signalListener is an interface for types that can receive OS signals.
type signalListener interface {
	Signal(context.Context, os.Signal)
}

// listenSignals subscribes to SIGINT and SIGTERM signals and forwards them to the provided listener.
// It also listens for context cancellation and signals the listener with an os.Interrupt.
func listenSignals(ctx context.Context, listener signalListener) {
	go func() {
		if signalctx.OSSignalsDisabled(ctx) {
			<-ctx.Done()
			listener.Signal(ctx, os.Interrupt)
			return
		}

		signalChan := make(chan os.Signal, 1)
		signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
		defer signal.Stop(signalChan)

		select {
		// If context is cancelled, signal with os.Interrupt.
		case <-ctx.Done():
			listener.Signal(ctx, os.Interrupt)
		// Forward the received signal.
		case sig := <-signalChan:
			listener.Signal(ctx, sig)
		}
	}()
}

// LogConfig defines configuration for log file creation.
type LogConfig = logpath.Config

// RecordEarlyFailure records a failure in the execution history before the DAG has fully started.
// This is used for infrastructure errors like singleton conflicts or process acquisition failures.
func (c *Context) RecordEarlyFailure(dag *core.DAG, dagRunID string, err error) error {
	if dag == nil || dagRunID == "" {
		return fmt.Errorf("DAG and dag-run ID are required to record failure")
	}

	// 1. Check if a DAGRunAttempt already exists for the given run-id.
	ref := exec.NewDAGRunRef(dag.Name, dagRunID)
	attempt, findErr := c.DAGRunStore.FindAttempt(c, ref)
	if findErr != nil && !errors.Is(findErr, exec.ErrDAGRunIDNotFound) {
		return fmt.Errorf("failed to check for existing attempt: %w", findErr)
	}

	if attempt == nil {
		// 2. Create the attempt if not exists
		att, createErr := c.DAGRunStore.CreateAttempt(c, dag, time.Now(), dagRunID, exec.NewDAGRunAttemptOptions{})
		if createErr != nil {
			return fmt.Errorf("failed to create run to record failure: %w", createErr)
		}
		attempt = att
	}

	// 3. Construct the "Failed" status
	statusBuilder := transform.NewStatusBuilder(dag)
	logPath, logPathErr := c.GenLogFileName(dag, dagRunID)
	if logPathErr != nil {
		logger.Warn(c, "Failed to generate log file path for early failure status",
			tag.Error(logPathErr),
			tag.DAG(dag.Name),
			tag.RunID(dagRunID),
		)
	}
	artifactDir, artifactDirErr := c.GenArtifactDir(dag, dagRunID)
	if artifactDirErr != nil {
		logger.Warn(c, "Failed to generate artifact directory for early failure status",
			tag.Error(artifactDirErr),
			tag.DAG(dag.Name),
			tag.RunID(dagRunID),
		)
	}
	status := statusBuilder.Create(dagRunID, core.Failed, 0, time.Now(),
		transform.WithLogFilePath(logPath),
		transform.WithArchiveDir(artifactDir),
		transform.WithFinishedAt(time.Now()),
		transform.WithError(err.Error()),
	)

	// 4. Write the status
	if err := attempt.Open(c); err != nil {
		return fmt.Errorf("failed to open attempt for recording failure: %w", err)
	}
	defer func() {
		_ = attempt.Close(c)
	}()

	if err := attempt.Write(c, status); err != nil {
		return fmt.Errorf("failed to write failed status: %w", err)
	}

	return nil
}
