agentic-coding-workflow/internal/run/run_profile_service.go

package run

import (
	"context"
	"encoding/json"
	"errors"
	"log/slog"
	"os"
	"strings"
	"time"

	"github.com/google/uuid"

	"github.com/yan1h/agent-coding-workflow/internal/audit"
	"github.com/yan1h/agent-coding-workflow/internal/infra/crypto"
	"github.com/yan1h/agent-coding-workflow/internal/infra/errs"
	"github.com/yan1h/agent-coding-workflow/internal/workspace"
)

// ServiceConfig 是 Service 的运行参数子集（派生自 config.RunConfig）。
type ServiceConfig struct {
	EnvWhitelist []string
	KillGrace    time.Duration
	OutTailBytes int
}

// Service 编排 run profile 的 CRUD 与进程生命周期（start/stop/restart/status/logs）。
// 鉴权完全继承所属 workspace：读靠 visibility，写靠 owner+admin（经 ProjectAccess）。
type Service struct {
	repo   Repository
	wsRepo workspace.Repository
	sup    *RunSupervisor
	pa     workspace.ProjectAccess
	rec    audit.Recorder
	enc    *crypto.Encryptor
	cfg    ServiceConfig
	log    *slog.Logger
}

// NewService 构造 run.Service。注入 workspace.Repository 以解析 main_path/project_id，
// ProjectAccess 做 owner/admin 写鉴权（复用 workspace 的窄接口）。
func NewService(
	repo Repository,
	wsRepo workspace.Repository,
	sup *RunSupervisor,
	pa workspace.ProjectAccess,
	rec audit.Recorder,
	enc *crypto.Encryptor,
	cfg ServiceConfig,
	log *slog.Logger,
) *Service {
	if log == nil {
		log = slog.Default()
	}
	return &Service{repo: repo, wsRepo: wsRepo, sup: sup, pa: pa, rec: rec, enc: enc, cfg: cfg, log: log}
}

// ===== 鉴权 helper =====

// authByWorkspace 解析 workspace 并校验调用方读/写权限。write=true 要求 owner+admin。
func (s *Service) authByWorkspace(ctx context.Context, c workspace.Caller, wsID uuid.UUID, write bool) (*workspace.Workspace, error) {
	ws, err := s.wsRepo.GetWorkspaceByID(ctx, wsID)
	if err != nil {
		return nil, err
	}
	canRead, canWrite, err := s.pa.ResolveByID(ctx, c.UserID, c.IsAdmin, ws.ProjectID)
	if err != nil {
		return nil, err
	}
	if write {
		if !canWrite {
			return nil, errs.New(errs.CodeForbidden, "no write access to this workspace")
		}
	} else if !canRead {
		return nil, errs.New(errs.CodeForbidden, "no read access to this workspace")
	}
	return ws, nil
}

// authByProfile 取 profile → 其 workspace，并校验权限。返回 profile 与 workspace。
func (s *Service) authByProfile(ctx context.Context, c workspace.Caller, profileID uuid.UUID, write bool) (*RunProfile, *workspace.Workspace, error) {
	p, err := s.repo.GetRunProfileByID(ctx, profileID)
	if err != nil {
		return nil, nil, err
	}
	ws, err := s.authByWorkspace(ctx, c, p.WorkspaceID, write)
	if err != nil {
		return nil, nil, err
	}
	return p, ws, nil
}

// ===== CRUD =====

func (s *Service) List(ctx context.Context, c workspace.Caller, wsID uuid.UUID) ([]RunProfileResp, error) {
	if _, err := s.authByWorkspace(ctx, c, wsID, false); err != nil {
		return nil, err
	}
	profiles, err := s.repo.ListRunProfilesByWorkspace(ctx, wsID)
	if err != nil {
		return nil, err
	}
	out := make([]RunProfileResp, 0, len(profiles))
	for _, p := range profiles {
		out = append(out, s.toResp(p))
	}
	return out, nil
}

func (s *Service) Create(ctx context.Context, c workspace.Caller, wsID uuid.UUID, req CreateRunProfileReq) (RunProfileResp, error) {
	if _, err := s.authByWorkspace(ctx, c, wsID, true); err != nil {
		return RunProfileResp{}, err
	}
	if strings.TrimSpace(req.Command) == "" {
		return RunProfileResp{}, errs.New(errs.CodeRunCommandInvalid, "command must not be empty")
	}
	encEnv, err := s.encryptEnv(req.Env)
	if err != nil {
		return RunProfileResp{}, err
	}
	enabled := true
	if req.Enabled != nil {
		enabled = *req.Enabled
	}
	p := &RunProfile{
		ID:           uuid.New(),
		WorkspaceID:  wsID,
		Slug:         req.Slug,
		Name:         req.Name,
		Description:  req.Description,
		Command:      req.Command,
		Args:         req.Args,
		EncryptedEnv: encEnv,
		Enabled:      enabled,
		CreatedBy:    c.UserID,
	}
	created, err := s.repo.CreateRunProfile(ctx, p)
	if err != nil {
		return RunProfileResp{}, err
	}
	s.audit(ctx, c, "run.profile.created", created.ID, map[string]any{"workspace_id": wsID.String(), "slug": created.Slug})
	return s.toResp(created), nil
}

func (s *Service) Update(ctx context.Context, c workspace.Caller, profileID uuid.UUID, req UpdateRunProfileReq) (RunProfileResp, error) {
	p, _, err := s.authByProfile(ctx, c, profileID, true)
	if err != nil {
		return RunProfileResp{}, err
	}
	if req.Slug != nil {
		p.Slug = *req.Slug
	}
	if req.Name != nil {
		p.Name = *req.Name
	}
	if req.Description != nil {
		p.Description = *req.Description
	}
	if req.Command != nil {
		if strings.TrimSpace(*req.Command) == "" {
			return RunProfileResp{}, errs.New(errs.CodeRunCommandInvalid, "command must not be empty")
		}
		p.Command = *req.Command
	}
	if req.Args != nil {
		p.Args = *req.Args
	}
	if req.Enabled != nil {
		p.Enabled = *req.Enabled
	}
	if req.Env != nil {
		encEnv, encErr := s.encryptEnv(*req.Env)
		if encErr != nil {
			return RunProfileResp{}, encErr
		}
		p.EncryptedEnv = encEnv
	}
	updated, err := s.repo.UpdateRunProfile(ctx, p)
	if err != nil {
		return RunProfileResp{}, err
	}
	s.audit(ctx, c, "run.profile.updated", updated.ID, map[string]any{"workspace_id": updated.WorkspaceID.String()})
	return s.toResp(updated), nil
}

func (s *Service) Delete(ctx context.Context, c workspace.Caller, profileID uuid.UUID) error {
	p, _, err := s.authByProfile(ctx, c, profileID, true)
	if err != nil {
		return err
	}
	// 运行中先整树停掉，避免删行后留下无主进程。
	if _, running := s.sup.Get(profileID); running {
		_ = s.sup.Stop(ctx, profileID, s.cfg.KillGrace)
	}
	if err := s.repo.DeleteRunProfile(ctx, profileID); err != nil {
		return err
	}
	s.audit(ctx, c, "run.profile.deleted", profileID, map[string]any{"workspace_id": p.WorkspaceID.String()})
	return nil
}

// ===== 进程生命周期 =====

func (s *Service) Start(ctx context.Context, c workspace.Caller, profileID uuid.UUID) (RunStatusResp, error) {
	p, ws, err := s.authByProfile(ctx, c, profileID, true)
	if err != nil {
		return RunStatusResp{}, err
	}
	if !p.Enabled {
		return RunStatusResp{}, errs.New(errs.CodeRunProfileDisabled, "run profile is disabled")
	}
	overrides, err := s.decryptEnv(p.EncryptedEnv)
	if err != nil {
		return RunStatusResp{}, err
	}
	_, err = s.sup.Spawn(SpawnParams{
		ProfileID: p.ID,
		Command:   p.Command,
		Args:      p.Args,
		Env:       s.buildEnv(overrides),
		Dir:       ws.MainPath,
	})
	if err != nil {
		if errors.Is(err, ErrAlreadyRunning) {
			return RunStatusResp{}, errs.New(errs.CodeRunProfileAlreadyRunning, "run profile already running")
		}
		return RunStatusResp{}, errs.Wrap(err, errs.CodeRunSpawnFailed, "spawn run profile")
	}
	if _, err := s.repo.MarkRunProfileStarted(ctx, p.ID); err != nil {
		s.log.Error("run.start.mark_started", "profile_id", p.ID, "err", err.Error())
	}
	s.audit(ctx, c, "run.profile.started", p.ID, map[string]any{"workspace_id": p.WorkspaceID.String()})
	return s.statusResp(p.ID), nil
}

func (s *Service) Stop(ctx context.Context, c workspace.Caller, profileID uuid.UUID) (RunStatusResp, error) {
	p, _, err := s.authByProfile(ctx, c, profileID, true)
	if err != nil {
		return RunStatusResp{}, err
	}
	if err := s.sup.Stop(ctx, p.ID, s.cfg.KillGrace); err != nil {
		if errors.Is(err, ErrNotRunning) {
			return RunStatusResp{}, errs.New(errs.CodeRunProfileNotRunning, "run profile is not running")
		}
		return RunStatusResp{}, errs.Wrap(err, errs.CodeInternal, "stop run profile")
	}
	s.audit(ctx, c, "run.profile.stopped", p.ID, map[string]any{"workspace_id": p.WorkspaceID.String()})
	return s.statusResp(p.ID), nil
}

func (s *Service) Restart(ctx context.Context, c workspace.Caller, profileID uuid.UUID) (RunStatusResp, error) {
	p, ws, err := s.authByProfile(ctx, c, profileID, true)
	if err != nil {
		return RunStatusResp{}, err
	}
	if !p.Enabled {
		return RunStatusResp{}, errs.New(errs.CodeRunProfileDisabled, "run profile is disabled")
	}
	if _, running := s.sup.Get(p.ID); running {
		_ = s.sup.Stop(ctx, p.ID, s.cfg.KillGrace)
	}
	overrides, err := s.decryptEnv(p.EncryptedEnv)
	if err != nil {
		return RunStatusResp{}, err
	}
	if _, err := s.sup.Spawn(SpawnParams{
		ProfileID: p.ID,
		Command:   p.Command,
		Args:      p.Args,
		Env:       s.buildEnv(overrides),
		Dir:       ws.MainPath,
	}); err != nil {
		if errors.Is(err, ErrAlreadyRunning) {
			return RunStatusResp{}, errs.New(errs.CodeRunProfileAlreadyRunning, "run profile already running")
		}
		return RunStatusResp{}, errs.Wrap(err, errs.CodeRunSpawnFailed, "spawn run profile")
	}
	if _, err := s.repo.MarkRunProfileStarted(ctx, p.ID); err != nil {
		s.log.Error("run.restart.mark_started", "profile_id", p.ID, "err", err.Error())
	}
	s.audit(ctx, c, "run.profile.restarted", p.ID, map[string]any{"workspace_id": p.WorkspaceID.String()})
	return s.statusResp(p.ID), nil
}

func (s *Service) Status(ctx context.Context, c workspace.Caller, profileID uuid.UUID) (RunStatusResp, error) {
	p, _, err := s.authByProfile(ctx, c, profileID, false)
	if err != nil {
		return RunStatusResp{}, err
	}
	return s.statusRespFrom(p), nil
}

// Logs 返回正在运行进程的内存日志（ID > since，最多 limit 条）。
func (s *Service) Logs(ctx context.Context, c workspace.Caller, profileID uuid.UUID, since int64, limit int) ([]LogLineResp, error) {
	if _, _, err := s.authByProfile(ctx, c, profileID, false); err != nil {
		return nil, err
	}
	envs := s.sup.LogsSince(profileID, since, limit)
	out := make([]LogLineResp, 0, len(envs))
	for _, e := range envs {
		out = append(out, toLogLineResp(e))
	}
	return out, nil
}

// SubscribeLogs 暴露给 WS handler：为运行中 profile 注册订阅，返回订阅句柄与 relay。
// 同时返回鉴权后的 profile（未运行时 relay 为 nil）。
func (s *Service) SubscribeLogs(ctx context.Context, c workspace.Caller, profileID, userID uuid.UUID) (*Subscriber, *runRelay, bool, error) {
	if _, _, err := s.authByProfile(ctx, c, profileID, false); err != nil {
		return nil, nil, false, err
	}
	sub, relay, ok := s.sup.Subscribe(profileID, userID)
	return sub, relay, ok, nil
}

// ===== 内部 helper =====

// buildEnv 构造被托管命令的环境：宿主白名单基础变量 + profile 覆盖。
// 不继承宿主全部环境，避免泄漏 APP_MASTER_KEY / DB DSN / git 凭据等敏感变量。
func (s *Service) buildEnv(overrides map[string]string) []string {
	return buildEnvFromWhitelist(s.cfg.EnvWhitelist, overrides)
}

// buildEnvFromWhitelist 是 env 构造的包级可复用实现：仅透传 whitelist 列出的宿主
// 环境变量，再叠加 overrides。供 run profile（长驻）与 ExecService（一次性 exec）
// 共用，保证两条路径相同的安全取舍——绝不泄漏 APP_MASTER_KEY / DB DSN / git 凭据。
func buildEnvFromWhitelist(whitelist []string, overrides map[string]string) []string {
	base := map[string]string{}
	for _, k := range whitelist {
		if v, ok := os.LookupEnv(k); ok {
			base[k] = v
		}
	}
	for k, v := range overrides {
		base[k] = v
	}
	out := make([]string, 0, len(base))
	for k, v := range base {
		out = append(out, k+"="+v)
	}
	return out
}

func (s *Service) encryptEnv(env map[string]string) ([]byte, error) {
	if len(env) == 0 {
		return nil, nil
	}
	b, err := json.Marshal(env)
	if err != nil {
		return nil, errs.Wrap(err, errs.CodeInternal, "marshal env")
	}
	ct, err := s.enc.Encrypt(b)
	if err != nil {
		return nil, errs.Wrap(err, errs.CodeInternal, "encrypt env")
	}
	return ct, nil
}

func (s *Service) decryptEnv(encrypted []byte) (map[string]string, error) {
	if len(encrypted) == 0 {
		return map[string]string{}, nil
	}
	plain, err := s.enc.Decrypt(encrypted)
	if err != nil {
		return nil, errs.Wrap(err, errs.CodeInternal, "decrypt env")
	}
	var m map[string]string
	if err := json.Unmarshal(plain, &m); err != nil {
		return nil, errs.Wrap(err, errs.CodeInternal, "unmarshal env")
	}
	if m == nil {
		m = map[string]string{}
	}
	return m, nil
}

// envKeys 解密并返回排序后的 env key 列表（仅 key，供前端展示）。解密失败回空列表。
func (s *Service) envKeys(encrypted []byte) []string {
	m, err := s.decryptEnv(encrypted)
	if err != nil {
		s.log.Warn("run.env_keys.decrypt_failed", "err", err.Error())
		return []string{}
	}
	return sortedKeys(m)
}

func (s *Service) toResp(p *RunProfile) RunProfileResp {
	status := s.sup.StatusOf(p.ID)
	var startedAt *time.Time
	if proc, ok := s.sup.Get(p.ID); ok {
		t := proc.StartedAt
		startedAt = &t
	}
	return buildRunProfileResp(p, s.envKeys(p.EncryptedEnv), status, startedAt)
}

func (s *Service) statusResp(profileID uuid.UUID) RunStatusResp {
	p, err := s.repo.GetRunProfileByID(context.Background(), profileID)
	if err != nil {
		return RunStatusResp{Status: string(s.sup.StatusOf(profileID))}
	}
	return s.statusRespFrom(p)
}

func (s *Service) statusRespFrom(p *RunProfile) RunStatusResp {
	status := s.sup.StatusOf(p.ID)
	var startedAt *string
	if proc, ok := s.sup.Get(p.ID); ok {
		t := proc.StartedAt.UTC().Format(time.RFC3339)
		startedAt = &t
	}
	return RunStatusResp{
		Status:       string(status),
		StartedAt:    startedAt,
		LastExitCode: p.LastExitCode,
		LastError:    p.LastError,
	}
}

func (s *Service) audit(ctx context.Context, c workspace.Caller, action string, targetID uuid.UUID, meta map[string]any) {
	if s.rec == nil {
		return
	}
	uid := c.UserID
	_ = s.rec.Record(ctx, audit.Entry{
		UserID:     &uid,
		Action:     action,
		TargetType: "run_profile",
		TargetID:   targetID.String(),
		Metadata:   meta,
	})
}