package engine

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"path/filepath"
	"sync"

	"tangled.org/core/notifier"
	"tangled.org/core/spindle/config"
	"tangled.org/core/spindle/db"
	"tangled.org/core/spindle/models"
	"tangled.org/core/spindle/secrets"
)

var (
	ErrTimedOut       = errors.New("timed out")
	ErrWorkflowFailed = errors.New("workflow failed")
	ErrCancelled      = errors.New("workflow cancelled")
)

// workflowLoggerProvider lets an engine supply the WorkflowLogger StartWorkflows
// uses, instead of the default on-disk one. Engines that don't implement it get
// a file logger under cfg.Server.LogDir. The mill engine supplies a no-op logger
// because it doesn't run steps locally — it relays the executor's real log file
// into place itself, so a second logger here would only write competing lines.
type workflowLoggerProvider interface {
	WorkflowLogger(wid models.WorkflowId) models.WorkflowLogger
}

func reportWorkflowStatusError(l *slog.Logger, database *db.DB, n *notifier.Notifier, wid models.WorkflowId, err error) {
	if errors.Is(err, ErrTimedOut) {
		dbErr := database.StatusTimeout(wid, n)
		if dbErr != nil {
			l.Error("failed to set workflow status to timeout", "wid", wid, "err", dbErr)
		}
	} else if errors.Is(err, ErrCancelled) {
		dbErr := database.StatusCancelled(wid, err.Error(), -1, n)
		if dbErr != nil {
			l.Error("failed to set workflow status to cancelled", "wid", wid, "err", dbErr)
		}
	} else {
		dbErr := database.StatusFailed(wid, err.Error(), -1, n)
		if dbErr != nil {
			l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
		}
	}
}

func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, n *notifier.Notifier, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) {
	l.Info("starting all workflows in parallel", "pipeline", pipelineId)

	var allSecrets []secrets.UnlockedSecret
	// never pass secrets to pipelines that run untrusted (e.g. fork) code
	if pipeline.TrustedSource && pipeline.RepoDid != "" {
		if res, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(pipeline.RepoDid.String())); err == nil {
			allSecrets = res
		}
	} else if !pipeline.TrustedSource {
		l.Info("skipping secrets for untrusted pipeline source", "pipeline", pipelineId)
	}

	secretValues := make([]string, len(allSecrets))
	for i, s := range allSecrets {
		secretValues[i] = s.Value
	}

	s3, err := NewS3(cfg.S3.LogBucket)
	if err != nil {
		l.Error("error creating s3 client", "err", err)
	}

	var wg sync.WaitGroup
	for eng, wfs := range pipeline.Workflows {
		workflowTimeout := eng.WorkflowTimeout()
		l.Info("using workflow timeout", "timeout", workflowTimeout)

		for _, w := range wfs {
			wg.Go(func() {
				wid := models.WorkflowId{
					PipelineId: pipelineId,
					Name:       w.Name,
				}

				defer func() {
					if s3 != nil {
						logFile := filepath.Join(cfg.Server.LogDir, fmt.Sprintf("%s.log", wid.String()))
						if err := s3.WriteFile(ctx, logFile); err != nil {
							l.Error("error uploading logs", "err", err)
						}
					}
				}()

				var wfLogger models.WorkflowLogger
				if p, ok := eng.(workflowLoggerProvider); ok {
					wfLogger = p.WorkflowLogger(wid)
				} else if fileLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues); err != nil {
					l.Warn("failed to setup step logger; logs will not be persisted", "error", err)
					wfLogger = models.NullLogger{}
				} else {
					l.Info("setup step logger; logs will be persisted", "logDir", cfg.Server.LogDir, "wid", wid)
					wfLogger = fileLogger
					defer fileLogger.Close()
				}

				l.Info("waiting for slot", "wid", wid)
				slot := WorkflowSlot(NoopSlot{})
				if s, ok := eng.(WorkflowSlotter); ok {
					var err error
					slot, err = s.AcquireWorkflowSlot(ctx, wid, &w, Wait)
					if err != nil {
						l.Error("failed to acquire slot", "wid", wid, "err", err)
						reportWorkflowStatusError(l, db, n, wid, err)
						return
					}
				}
				defer slot.Release()

				err := db.StatusRunning(wid, n)
				if err != nil {
					l.Error("failed to set workflow status to running", "wid", wid, "err", err)
					return
				}

				err = eng.SetupWorkflow(ctx, wid, &w, wfLogger)
				if err != nil {
					// TODO(winter): Should this always set StatusFailed?
					// In the original, we only do in a subset of cases.
					l.Error("setting up workflow", "wid", wid, "err", err)

					destroyErr := eng.DestroyWorkflow(ctx, wid)
					if destroyErr != nil {
						l.Error("failed to destroy workflow after setup failure", "error", destroyErr)
					}

					reportWorkflowStatusError(l, db, n, wid, err)
					return
				}
				defer eng.DestroyWorkflow(ctx, wid)

				ctx, cancel := context.WithTimeout(ctx, workflowTimeout)
				defer cancel()

				for stepIdx, step := range w.Steps {
					// log start of step
					if wfLogger != nil {
						wfLogger.
							ControlWriter(stepIdx, step, models.StepStatusStart).
							Write([]byte{0})
					}

					err = eng.RunStep(ctx, wid, &w, stepIdx, allSecrets, wfLogger)

					// log end of step
					if wfLogger != nil {
						wfLogger.
							ControlWriter(stepIdx, step, models.StepStatusEnd).
							Write([]byte{0})
					}

					if err != nil {
						reportWorkflowStatusError(l, db, n, wid, err)
						return
					}
				}

				err = db.StatusSuccess(wid, n)
				if err != nil {
					l.Error("failed to set workflow status to success", "wid", wid, "err", err)
				}
			})
		}
	}

	wg.Wait()
	l.Info("all workflows completed")
}