run.go

package project

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"os/signal"
	"path"
	"path/filepath"
	"strings"
	"syscall"
	"time"

	"github.com/pulumi/pulumi/sdk/v3/go/auto/events"
	"github.com/pulumi/pulumi/sdk/v3/go/common/apitype"
	"github.com/sst/sst/v3/internal/util"
	"github.com/sst/sst/v3/pkg/bus"
	"github.com/sst/sst/v3/pkg/flag"
	"github.com/sst/sst/v3/pkg/global"
	"github.com/sst/sst/v3/pkg/id"
	"github.com/sst/sst/v3/pkg/js"
	"github.com/sst/sst/v3/pkg/process"
	"github.com/sst/sst/v3/pkg/project/provider"
	"github.com/sst/sst/v3/pkg/telemetry"
	"github.com/sst/sst/v3/pkg/types"
	"golang.org/x/exp/slices"
	"golang.org/x/sync/errgroup"
)

func (p *Project) Run(ctx context.Context, input *StackInput) error {
	// if flag.SST_EXPERIMENTAL {
	// 	slog.Info("using next run system")
	// }
	// return p.RunOld(ctx, input)
	return p.RunNext(ctx, input)
}

func (p *Project) RunNext(ctx context.Context, input *StackInput) error {
	log := slog.Default().With("service", "project.run")
	log.Info("running stack command", "cmd", input.Command)

	if p.app.Protect && input.Command == "remove" {
		return ErrProtectedStage
	}

	bus.Publish(&StackCommandEvent{
		App:     p.app.Name,
		Stage:   p.app.Stage,
		Config:  p.PathConfig(),
		Command: input.Command,
		Version: p.Version(),
	})

	update := &provider.Update{
		ID: id.Descending(),
	}
	var err error
	if input.Command != "diff" {
		update, err = p.Lock(input.Command)
		if err != nil {
			if err == provider.ErrLockExists {
				bus.Publish(&ConcurrentUpdateEvent{})
			}
			return err
		}
		log = log.With("updateID", update.ID)
		defer p.Unlock()
	}

	workdir, err := p.NewWorkdir(update.ID)
	if err != nil {
		return err
	}
	defer workdir.Cleanup()

	passphrase, err := provider.Passphrase(p.home, p.app.Name, p.app.Stage)
	if err != nil {
		return err
	}

	outfile := filepath.Join(p.PathPlatformDir(), fmt.Sprintf("sst.config.%v.mjs", time.Now().UnixMilli()))
	os.WriteFile(
		filepath.Join(workdir.path, "Pulumi.yaml"),
		[]byte("name: "+p.app.Name+"\nruntime: nodejs\nmain: "+outfile+"\n"),
		0644,
	)
	pulumiStdout, err := os.Create(p.PathLog("pulumi"))
	if err != nil {
		return err
	}
	defer pulumiStdout.Close()
	pulumiStderr, err := os.Create(p.PathLog("pulumi.err"))
	if err != nil {
		return err
	}
	defer pulumiStderr.Close()
	_, err = workdir.Pull()
	if err != nil {
		if errors.Is(err, provider.ErrStateNotFound) {
			if input.Command != "deploy" {
				return ErrStageNotFound
			}
			cmd := process.Command(global.PulumiPath(), "stack", "init", "organization/"+p.app.Name+"/"+p.app.Stage)
			cmd.Stdout = pulumiStdout
			cmd.Stderr = pulumiStderr
			cmd.Dir = workdir.path
			cmd.Env = os.Environ()
			cmd.Env = append(cmd.Env,
				"PULUMI_BACKEND_URL="+filepath.ToSlash("file://"+workdir.Backend()),
				"PULUMI_CONFIG_PASSPHRASE="+passphrase,
			)
			err := cmd.Run()
			if err != nil {
				return err
			}

		} else {
			return err
		}
	}

	completed, err := getCompletedEvent(ctx, passphrase, workdir)
	if err != nil {
		bus.Publish(&BuildFailedEvent{
			Error: err.Error(),
		})
		log.Info("state file might be corrupted", "err", err)
		return err
	}
	completed.Finished = true
	completed.Old = true
	bus.Publish(completed)
	log.Info("got previous deployment")

	cli := map[string]interface{}{
		"command": input.Command,
		"dev":     input.Dev,
		"paths": map[string]string{
			"home":     global.ConfigDir(),
			"root":     p.PathRoot(),
			"work":     p.PathWorkingDir(),
			"platform": p.PathPlatformDir(),
		},
		"state": map[string]interface{}{
			"version": completed.Versions,
		},
	}
	cliBytes, err := json.Marshal(cli)
	if err != nil {
		return err
	}
	appBytes, err := json.Marshal(p.app)
	if err != nil {
		return err
	}

	providerShim := []string{}
	for _, entry := range p.lock {
		providerShim = append(providerShim, fmt.Sprintf("import * as %s from \"%s\";", entry.Alias, entry.Package))
	}
	providerShim = append(providerShim, fmt.Sprintf("import * as sst from \"%s\";", path.Join(filepath.ToSlash(p.PathPlatformDir()), "src/components")))

	buildResult, err := js.Build(js.EvalOptions{
		Dir:     p.PathRoot(),
		Outfile: outfile,
		Define: map[string]string{
			"$app": string(appBytes),
			"$cli": string(cliBytes),
			"$dev": fmt.Sprintf("%v", input.Dev),
		},
		Inject:  []string{filepath.ToSlash(filepath.Join(p.PathWorkingDir(), "platform/src/shim/run.js"))},
		Globals: strings.Join(providerShim, "\n"),
		Code: fmt.Sprintf(`
      import { run } from "%v";
			import mod from '%s';
      const result = await run(mod.run);
      export default result;
    `,
			filepath.ToSlash(path.Join(p.PathWorkingDir(), "platform/src/auto/run.ts")),
			filepath.ToSlash(p.PathConfig()),
		),
	})
	if err != nil {
		bus.Publish(&BuildFailedEvent{
			Error: err.Error(),
		})
		log.Error("failed to build sst.config.ts", "err", err)
		return err
	}
	log.Info("built sst.config.ts", "to", outfile)
	if !flag.SST_NO_CLEANUP {
		defer js.Cleanup(buildResult)
	}

	// disable for now until we hash env too
	if input.SkipHash != "" && buildResult.OutputFiles[0].Hash == input.SkipHash && false {
		bus.Publish(&SkipEvent{})
		return nil
	}

	var meta = js.Metafile{}
	err = json.Unmarshal([]byte(buildResult.Metafile), &meta)
	if err != nil {
		return err
	}
	files := []string{}

	for key := range meta.Inputs {
		absPath, err := filepath.Abs(key)
		if err != nil {
			continue
		}
		files = append(files, absPath)
	}
	bus.Publish(&BuildSuccessEvent{
		Files: files,
		Hash:  buildResult.OutputFiles[0].Hash,
	})
	log.Info("tracked files")

	secrets := map[string]string{}
	fallback := map[string]string{}

	wg := errgroup.Group{}

	wg.Go(func() error {
		secrets, err = provider.GetSecrets(p.home, p.app.Name, p.app.Stage)
		if err != nil {
			return ErrPassphraseInvalid
		}
		return nil
	})

	wg.Go(func() error {
		fallback, err = provider.GetSecrets(p.home, p.app.Name, "")
		if err != nil {
			return ErrPassphraseInvalid
		}
		return nil
	})

	if err := wg.Wait(); err != nil {
		return err
	}

	env := os.Environ()
	for key, value := range p.Env() {
		env = append(env, fmt.Sprintf("%v=%v", key, value))
	}
	for key, value := range fallback {
		env = append(env, fmt.Sprintf("SST_SECRET_%v=%v", key, value))
	}
	for key, value := range secrets {
		env = append(env, fmt.Sprintf("SST_SECRET_%v=%v", key, value))
	}
	env = append(env,
		"PULUMI_CONFIG_PASSPHRASE="+passphrase,
		"PULUMI_SKIP_UPDATE_CHECK=true",
		"PULUMI_BACKEND_URL=file://"+filepath.ToSlash(workdir.Backend()),
		"PULUMI_DEBUG_COMMANDS=true",
		"PULUMI_IGNORE_AMBIENT_PLUGINS=true",
		// "PULUMI_DISABLE_AUTOMATIC_PLUGIN_ACQUISITION=true",
		"NODE_OPTIONS="+func() string {
			nodeOptions := "--enable-source-maps --no-deprecation"
			if existing := os.Getenv("NODE_OPTIONS"); existing != "" {
				nodeOptions = existing + " " + nodeOptions
			}
			return nodeOptions
		}(),
		"PULUMI_HOME="+global.ConfigDir(),
	)
	if input.ServerPort != 0 {
		env = append(env, "SST_SERVER=http://127.0.0.1:"+fmt.Sprint(input.ServerPort))
	}
	pulumiPath := global.PulumiPath()
	if flag.SST_PULUMI_PATH != "" {
		pulumiPath = flag.SST_PULUMI_PATH
	}

	eventlogPath := workdir.EventLogPath()
	eventlog, err := os.OpenFile(eventlogPath, os.O_RDWR|os.O_CREATE, 0644)
	if err != nil {
		return err
	}
	defer eventlog.Close()

	args := []string{
		"--stack", fmt.Sprintf("organization/%v/%v", p.app.Name, p.app.Stage),
		"--non-interactive",
		"--event-log", eventlogPath,
	}

	if input.Command == "deploy" {
		upgradeMsgs := p.checkProviderUpgrade(completed.Resources)
		if len(upgradeMsgs) > 0 {
			return util.NewReadableError(nil, strings.Join(upgradeMsgs, "\n\n"))
		}
	}

	if input.Command == "deploy" || input.Command == "diff" || input.Command == "refresh" {
		for provider, opts := range p.app.Providers {
			for key, value := range opts.(map[string]interface{}) {
				// Skip SST-only fields that Pulumi doesn't understand
				if key == "package" {
					continue
				}
				switch v := value.(type) {
				case map[string]interface{}:
					bytes, err := json.Marshal(v)
					if err != nil {
						return err
					}
					args = append(args, "--config", fmt.Sprintf("%v:%v=%v", provider, key, string(bytes)))
				case []interface{}:
					bytes, err := json.Marshal(v)
					if err != nil {
						return err
					}
					args = append(args, "--config", fmt.Sprintf("%v:%v=%v", provider, key, string(bytes)))
				case string:
					args = append(args, "--config", fmt.Sprintf("%v:%v=%v", provider, key, v))
				}
			}
		}
	}

	switch input.Command {
	case "diff":
		args = append([]string{"preview"}, args...)
	case "refresh":
		args = append([]string{"refresh", "--yes", "--run-program"}, args...)
	case "deploy":
		args = append([]string{"up", "--yes", "-f"}, args...)
	case "remove":
		args = append([]string{"destroy", "--yes", "-f"}, args...)
	}

	if (input.Command == "diff" || input.Command == "deploy") && input.PolicyPath != "" {
		policyPath, err := p.ResolvePolicyPackPath(input.PolicyPath)
		if err != nil {
			return util.NewReadableError(nil, err.Error())
		}
		args = append(args, "--policy-pack", policyPath)
	}

	if input.Target != nil {
		for _, item := range input.Target {
			index := slices.IndexFunc(completed.Resources, func(res apitype.ResourceV3) bool {
				return res.URN.Name() == item
			})
			if index == -1 {
				return util.NewReadableError(nil, fmt.Sprintf("Target not found: %v", item))
			}
			resource := completed.Resources[index]
			if resource.Type == "sst:sst:Group" {
				for _, memberURN := range resolveGroupMembers(resource, completed.Resources) {
					args = append(args, "--target", memberURN)
				}
			} else {
				args = append(args, "--target", string(resource.URN))
			}
		}
		if len(input.Target) > 0 {
			args = append(args, "--target-dependents")
		}
	}

	if input.Exclude != nil {
		for _, item := range input.Exclude {
			index := slices.IndexFunc(completed.Resources, func(res apitype.ResourceV3) bool {
				return res.URN.Name() == item
			})
			if index == -1 {
				return util.NewReadableError(nil, fmt.Sprintf("Exclude target not found: %v", item))
			}
			resource := completed.Resources[index]
			if resource.Type == "sst:sst:Group" {
				for _, memberURN := range resolveGroupMembers(resource, completed.Resources) {
					args = append(args, "--exclude", memberURN)
				}
			} else {
				args = append(args, "--exclude", string(resource.URN))
			}
		}
		if len(input.Exclude) > 0 {
			args = append(args, "--exclude-dependents")
		}
	}

	cmd := process.Command(pulumiPath, args...)
	process.Detach(cmd)
	cmd.Env = env
	cmd.Stdout = pulumiStdout
	cmd.Stderr = pulumiStderr
	cmd.Dir = workdir.Backend()
	log.Info("starting pulumi", "args", cmd.Args)

	errors := []Error{}
	finished := false
	importDiffs := map[string][]ImportDiff{}
	hasPolicyFlag := input.PolicyPath != ""
	hasPolicyEvents := false
	hasPolicyViolations := false

	partial := make(chan int, 1000)
	partialContext, partialCancel := context.WithCancel(ctx)
	defer partialCancel()
	partialDone := make(chan error)
	go func() {
		if input.Command == "diff" {
			return
		}
		for {
			select {
			case <-partialContext.Done():
				partialDone <- nil
				return
			case <-partial:
				workdir.PushPartial(update.ID)
			case <-time.After(time.Second * 5):
				workdir.PushPartial(update.ID)
				continue
			}
		}
	}()

	err = cmd.Start()
	if err != nil {
		return err
	}
	exited := make(chan struct{})
	go func() {
		cmd.Wait()
		log.Info("pulumi exited", "err", err)
		close(exited)
	}()

	go func() {
		select {
		case <-exited:
			return
		case <-ctx.Done():
			if cmd.Process != nil {
				log.Info("sending interrupt")
				err := cmd.Process.Signal(syscall.SIGINT)
				if err != nil {
					log.Error("failed to send interrupt", "err", err)
				}
				bus.Publish(&CancelledEvent{})
				interruptChannel := make(chan os.Signal, 1)
				signal.Notify(interruptChannel, syscall.SIGINT, syscall.SIGTERM)

				for {
					select {
					case <-exited:
						return
					case <-interruptChannel:
						if cmd.Process != nil {
							log.Info("sending force interrupt")
							err := cmd.Process.Signal(syscall.SIGINT)
							if err != nil {
								log.Error("failed to send interrupt", "err", err)
							}
						}
					}
				}
			}
		}
	}()

	reader := bufio.NewReader(eventlog)

	eofs := 0
loop:
	for {
		bytes, err := reader.ReadBytes('\n')
		if err != nil {
			if err == io.EOF {
				select {
				case <-exited:
					log.Info("eof and exited", "eofs", eofs)
					eofs++
					if eofs < 2 {
						continue
					}
					log.Info("breaking out of tail loop")
					break loop
				case <-time.After(time.Millisecond * 100):
					continue
				}
			}
			continue
		}

		var event events.EngineEvent
		err = json.Unmarshal(bytes, &event)
		if err != nil {
			log.Error("failed to unmarshal event", "err", err)
			continue
		}

		if event.PolicyEvent != nil {
			hasPolicyEvents = true
			message := fmt.Sprintf("Policy: %s\n%s",
				event.PolicyEvent.PolicyName,
				strings.TrimSpace(event.PolicyEvent.Message))

			if event.PolicyEvent.EnforcementLevel == "mandatory" {
				log.Info("policy violation",
					"policy", event.PolicyEvent.PolicyName,
					"urn", event.PolicyEvent.ResourceURN)

				errors = append(errors, Error{
					Message: message,
					URN:     event.PolicyEvent.ResourceURN,
				})
				hasPolicyViolations = true
			} else if event.PolicyEvent.EnforcementLevel == "advisory" {
				log.Info("policy advisory",
					"policy", event.PolicyEvent.PolicyName,
					"urn", event.PolicyEvent.ResourceURN)

				bus.Publish(&PolicyAdvisoryEvent{
					Policy:  event.PolicyEvent.PolicyName,
					Message: strings.TrimSpace(event.PolicyEvent.Message),
					URN:     event.PolicyEvent.ResourceURN,
				})
			}
		}

		if event.DiagnosticEvent != nil && event.DiagnosticEvent.Severity == "error" {
			if strings.HasPrefix(event.DiagnosticEvent.Message, "update failed") || strings.Contains(event.DiagnosticEvent.Message, "failed to register new resource") {
				continue
			}

			// check if the error is a common error
			help := []string{}
			for _, commonError := range CommonErrors {
				if strings.Contains(event.DiagnosticEvent.Message, commonError.Message) {
					help = append(help, commonError.Short...)
				}
			}

			exists := false
			if event.DiagnosticEvent.URN != "" {
				for _, item := range errors {
					if item.URN == event.DiagnosticEvent.URN {
						exists = true
						break
					}
				}
			}

			if exists {
				continue
			}

			if !exists {
				errors = append(errors, Error{
					Message: strings.TrimSpace(event.DiagnosticEvent.Message),
					URN:     event.DiagnosticEvent.URN,
					Help:    help,
				})
				log.Info("telemetry tracking error")
				telemetry.Track("cli.resource.error", map[string]interface{}{
					"error": event.DiagnosticEvent.Message,
					"urn":   event.DiagnosticEvent.URN,
				})
			}
		}

		if event.ResOpFailedEvent != nil {
			if event.ResOpFailedEvent.Metadata.Op == apitype.OpImport {
				for _, name := range event.ResOpFailedEvent.Metadata.Diffs {
					old := event.ResOpFailedEvent.Metadata.Old.Inputs[name]
					next := event.ResOpFailedEvent.Metadata.New.Inputs[name]
					diffs, ok := importDiffs[event.ResOpFailedEvent.Metadata.URN]
					if !ok {
						diffs = []ImportDiff{}
					}
					importDiffs[event.ResOpFailedEvent.Metadata.URN] = append(diffs, ImportDiff{
						URN:   event.ResOpFailedEvent.Metadata.URN,
						Input: name,
						Old:   old,
						New:   next,
					})
				}
			}
		}

		if input.Command != "diff" && (event.ResOutputsEvent != nil || event.CancelEvent != nil || event.SummaryEvent != nil) {
			partial <- 1
		}

		for _, field := range getNotNilFields(event) {
			bus.Publish(field)
		}

		if event.SummaryEvent != nil {
			finished = true
		}
	}

	// fallback: re-read the event log if the tailing loop missed SummaryEvent
	if !finished {
		if data, err := os.ReadFile(eventlogPath); err == nil {
			for _, line := range strings.Split(string(data), "\n") {
				if line == "" {
					continue
				}
				var ev events.EngineEvent
				if json.Unmarshal([]byte(line), &ev) == nil && ev.SummaryEvent != nil {
					finished = true
					break
				}
			}
		}
	}

	log.Info("parsing state")
	complete, err := getCompletedEvent(context.Background(), passphrase, workdir)
	if err != nil {
		return err
	}
	complete.UpdateID = update.ID
	complete.Finished = finished
	complete.Errors = errors
	complete.ImportDiffs = importDiffs
	types.Generate(p.PathConfig(), complete.Links)
	defer bus.Publish(complete)

	if input.Command != "diff" {
		log.Info("canceling partial")
		partialCancel()
		log.Info("waiting for partial to exit")
		<-partialDone

		err = workdir.Push(update.ID)
		if err != nil {
			return err
		}
	}

	outputsFilePath := filepath.Join(p.PathWorkingDir(), "outputs.json")
	outputsFile, _ := os.Create(outputsFilePath)
	defer outputsFile.Close()
	json.NewEncoder(outputsFile).Encode(complete.Outputs)

	if input.Command != "diff " {
		update.TimeCompleted = time.Now().Format(time.RFC3339)
		for _, err := range errors {
			update.Errors = append(update.Errors, provider.SummaryError{
				URN:     err.URN,
				Message: err.Message,
			})
		}
		err = provider.PutUpdate(p.home, p.app.Name, p.app.Stage, update)
		if err != nil {
			return err
		}
	}

	if input.Command == "remove" && len(complete.Resources) == 0 {
		provider.Cleanup(p.home, p.app.Name, p.app.Stage)
	}

	log.Info("done running stack command", "resources", len(complete.Resources))

	if cmd.ProcessState.ExitCode() > 0 {
		if hasPolicyViolations {
			return ErrPolicyViolation
		}
		if hasPolicyFlag && !hasPolicyEvents && len(errors) == 0 {
			return ErrPolicyConfigError
		}
		return ErrStackRunFailed
	}
	return nil
}

func resolveGroupMembers(resource apitype.ResourceV3, allResources []apitype.ResourceV3) []string {
	outputs, ok := parsePlaintext(resource.Outputs).(map[string]interface{})
	if !ok {
		return nil
	}
	group, ok := outputs["_group"].(map[string]interface{})
	if !ok {
		return nil
	}
	members, ok := group["members"].([]interface{})
	if !ok {
		return nil
	}

	// Collect direct member URNs
	var directMembers []string
	memberSet := make(map[string]bool)
	for _, m := range members {
		if urn, ok := m.(string); ok {
			directMembers = append(directMembers, urn)
			memberSet[urn] = true
		}
	}

	// For each direct member, find linked dependencies by walking the
	// children's dependency graph back up to top-level SST components
	for _, urn := range directMembers {
		for _, desc := range findDescendants(urn, allResources) {
			for _, dep := range desc.Dependencies {
				if parent := findTopLevelSSTParent(string(dep), allResources); parent != "" {
					memberSet[parent] = true
				}
			}
			for _, deps := range desc.PropertyDependencies {
				for _, dep := range deps {
					if parent := findTopLevelSSTParent(string(dep), allResources); parent != "" {
						memberSet[parent] = true
					}
				}
			}
		}
	}

	result := make([]string, 0, len(memberSet))
	for urn := range memberSet {
		result = append(result, urn)
	}
	return result
}

func findDescendants(parentURN string, allResources []apitype.ResourceV3) []apitype.ResourceV3 {
	var result []apitype.ResourceV3
	for _, res := range allResources {
		if string(res.Parent) == parentURN {
			result = append(result, res)
			result = append(result, findDescendants(string(res.URN), allResources)...)
		}
	}
	return result
}

func findTopLevelSSTParent(urn string, allResources []apitype.ResourceV3) string {
	idx := slices.IndexFunc(allResources, func(r apitype.ResourceV3) bool {
		return string(r.URN) == urn
	})
	if idx == -1 {
		return ""
	}
	res := allResources[idx]
	if res.Parent == "" || res.Parent.Type() == "pulumi:pulumi:Stack" {
		if strings.HasPrefix(string(res.Type), "sst:") {
			return string(res.URN)
		}
		return ""
	}
	return findTopLevelSSTParent(string(res.Parent), allResources)
}