You've already forked agentic-coding-workflow
462 lines
15 KiB
Go
462 lines
15 KiB
Go
// workspace_service.go 实现 WorkspaceService。Service 不直接调 git CLI；
|
|
// 写路径上调 cloneRunner.Run（异步）/ Runner.Fetch（同步）。
|
|
// 鉴权：private project 写要求 owner+admin；internal project 写要求 owner+admin；
|
|
// 读权限委托给 ProjectAccess（窄接口，避免对 internal/project 的强依赖）。
|
|
package workspace
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"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/infra/git"
|
|
)
|
|
|
|
// ProjectAccess 是一个窄接口，让 workspace.Service 不依赖 internal/project 全部。
|
|
type ProjectAccess interface {
|
|
Resolve(ctx context.Context, callerID uuid.UUID, isAdmin bool, projectSlug string) (uuid.UUID, bool, bool, error)
|
|
ResolveByID(ctx context.Context, callerID uuid.UUID, isAdmin bool, projectID uuid.UUID) (bool, bool, error)
|
|
}
|
|
|
|
// CloneScheduler 抽象 clone 的异步调度。
|
|
type CloneScheduler interface {
|
|
Schedule(wsID uuid.UUID)
|
|
}
|
|
|
|
type workspaceService struct {
|
|
repo Repository
|
|
rec audit.Recorder
|
|
enc *crypto.Encryptor
|
|
gitr git.Runner
|
|
pa ProjectAccess
|
|
clones CloneScheduler
|
|
dataDir string
|
|
fetchTimout time.Duration
|
|
}
|
|
|
|
// NewWorkspaceService 构造 WorkspaceService 实现。
|
|
func NewWorkspaceService(
|
|
repo Repository,
|
|
rec audit.Recorder,
|
|
enc *crypto.Encryptor,
|
|
gitr git.Runner,
|
|
pa ProjectAccess,
|
|
clones CloneScheduler,
|
|
dataDir string,
|
|
) WorkspaceService {
|
|
return &workspaceService{
|
|
repo: repo,
|
|
rec: rec,
|
|
enc: enc,
|
|
gitr: gitr,
|
|
pa: pa,
|
|
clones: clones,
|
|
dataDir: dataDir,
|
|
fetchTimout: 5 * time.Minute,
|
|
}
|
|
}
|
|
|
|
func (s *workspaceService) Create(ctx context.Context, c Caller, projectSlug string, in CreateWorkspaceInput) (*Workspace, error) {
|
|
pid, _, canWrite, err := s.pa.Resolve(ctx, c.UserID, c.IsAdmin, projectSlug)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !canWrite {
|
|
return nil, errs.New(errs.CodeForbidden, "no write access to project")
|
|
}
|
|
if err := ValidateSlug(in.Slug); err != nil {
|
|
return nil, errs.Wrap(err, errs.CodeInvalidInput, "invalid slug")
|
|
}
|
|
if err := ValidateRemoteURL(in.GitRemoteURL); err != nil {
|
|
return nil, errs.Wrap(err, errs.CodeWorkspaceRemoteInvalid, "invalid remote url")
|
|
}
|
|
if !in.Credential.Kind.IsValid() {
|
|
return nil, errs.New(errs.CodeGitCredentialInvalid, "credential kind invalid")
|
|
}
|
|
defaultBr := in.DefaultBranch
|
|
if defaultBr == "" {
|
|
defaultBr = "main"
|
|
}
|
|
if err := ValidateBranch(defaultBr); err != nil {
|
|
return nil, errs.Wrap(err, errs.CodeInvalidInput, "invalid default branch")
|
|
}
|
|
wsID := uuid.New()
|
|
ws := &Workspace{
|
|
ID: wsID,
|
|
ProjectID: pid,
|
|
Slug: in.Slug,
|
|
Name: firstNonEmpty(in.Name, in.Slug),
|
|
Description: in.Description,
|
|
GitRemoteURL: in.GitRemoteURL,
|
|
DefaultBranch: defaultBr,
|
|
MainPath: MainPath(s.dataDir, wsID),
|
|
SyncStatus: SyncStatusCloning,
|
|
}
|
|
created, err := s.repo.CreateWorkspace(ctx, ws)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cred, err := s.encryptForUpsert(wsID, in.Credential)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if _, err := s.repo.UpsertCredential(ctx, cred); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
s.audit(ctx, &c.UserID, "workspace.create", "workspace", wsID.String(), map[string]any{"slug": in.Slug})
|
|
if s.clones != nil {
|
|
s.clones.Schedule(wsID)
|
|
}
|
|
return created, nil
|
|
}
|
|
|
|
func (s *workspaceService) Get(ctx context.Context, c Caller, wsID uuid.UUID) (*Workspace, error) {
|
|
ws, err := s.repo.GetWorkspaceByID(ctx, wsID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
canRead, _, err := s.pa.ResolveByID(ctx, c.UserID, c.IsAdmin, ws.ProjectID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !canRead {
|
|
return nil, errs.New(errs.CodeForbidden, "no read access")
|
|
}
|
|
return ws, nil
|
|
}
|
|
|
|
func (s *workspaceService) List(ctx context.Context, c Caller, projectSlug string) ([]*Workspace, error) {
|
|
pid, canRead, _, err := s.pa.Resolve(ctx, c.UserID, c.IsAdmin, projectSlug)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !canRead {
|
|
return nil, errs.New(errs.CodeForbidden, "no read access")
|
|
}
|
|
return s.repo.ListWorkspacesByProject(ctx, pid)
|
|
}
|
|
|
|
func (s *workspaceService) Update(ctx context.Context, c Caller, wsID uuid.UUID, in UpdateWorkspaceInput) (*Workspace, error) {
|
|
ws, err := s.repo.GetWorkspaceByID(ctx, wsID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if _, canWrite, err := s.pa.ResolveByID(ctx, c.UserID, c.IsAdmin, ws.ProjectID); err != nil {
|
|
return nil, err
|
|
} else if !canWrite {
|
|
return nil, errs.New(errs.CodeForbidden, "no write access")
|
|
}
|
|
name := ws.Name
|
|
desc := ws.Description
|
|
defaultBr := ws.DefaultBranch
|
|
if in.Name != nil {
|
|
name = *in.Name
|
|
}
|
|
if in.Description != nil {
|
|
desc = *in.Description
|
|
}
|
|
if in.DefaultBranch != nil && *in.DefaultBranch != ws.DefaultBranch {
|
|
// 切换 default_branch：先校验→拉远端→切到新分支，再写 DB。
|
|
// 任一 git 步骤失败则不改库，让调用方感知并自行重试/回滚。
|
|
if err := ValidateBranch(*in.DefaultBranch); err != nil {
|
|
return nil, errs.Wrap(err, errs.CodeInvalidInput, "invalid default branch")
|
|
}
|
|
cred, credErr := s.loadDecryptedCredential(ctx, wsID)
|
|
if credErr != nil {
|
|
return nil, credErr
|
|
}
|
|
if fetchErr := s.gitr.FetchRemoteBranch(ctx, ws.MainPath, *in.DefaultBranch, cred); fetchErr != nil {
|
|
return nil, errs.Wrap(fetchErr, errs.CodeGitCmdFailed, "fetch remote branch")
|
|
}
|
|
if coErr := s.gitr.Checkout(ctx, ws.MainPath, *in.DefaultBranch); coErr != nil {
|
|
return nil, errs.Wrap(coErr, errs.CodeGitCmdFailed, "checkout branch")
|
|
}
|
|
defaultBr = *in.DefaultBranch
|
|
} else if in.DefaultBranch != nil {
|
|
if err := ValidateBranch(*in.DefaultBranch); err != nil {
|
|
return nil, errs.Wrap(err, errs.CodeInvalidInput, "invalid default branch")
|
|
}
|
|
defaultBr = *in.DefaultBranch
|
|
}
|
|
updated, err := s.repo.UpdateWorkspaceCore(ctx, wsID, name, desc, defaultBr)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
s.audit(ctx, &c.UserID, "workspace.update", "workspace", wsID.String(), nil)
|
|
return updated, nil
|
|
}
|
|
|
|
func (s *workspaceService) Delete(ctx context.Context, c Caller, wsID uuid.UUID) error {
|
|
ws, err := s.repo.GetWorkspaceByID(ctx, wsID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, canWrite, err := s.pa.ResolveByID(ctx, c.UserID, c.IsAdmin, ws.ProjectID); err != nil {
|
|
return err
|
|
} else if !canWrite {
|
|
return errs.New(errs.CodeForbidden, "no write access")
|
|
}
|
|
if err := s.repo.DeleteWorkspace(ctx, wsID); err != nil {
|
|
return err
|
|
}
|
|
// MainPath 是 ${dataDir}/<wsID>/main；删 workspace 时连父目录（含 .worktrees/）一起清掉。
|
|
go func(path string) {
|
|
_ = os.RemoveAll(path)
|
|
}(filepath.Dir(ws.MainPath))
|
|
s.audit(ctx, &c.UserID, "workspace.delete", "workspace", wsID.String(), nil)
|
|
return nil
|
|
}
|
|
|
|
func (s *workspaceService) Sync(ctx context.Context, c Caller, wsID uuid.UUID) (*Workspace, error) {
|
|
ws, err := s.repo.GetWorkspaceByID(ctx, wsID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if _, canWrite, err := s.pa.ResolveByID(ctx, c.UserID, c.IsAdmin, ws.ProjectID); err != nil {
|
|
return nil, err
|
|
} else if !canWrite {
|
|
return nil, errs.New(errs.CodeForbidden, "no write access")
|
|
}
|
|
// 先解密凭据，再抢锁置 syncing：避免置 syncing 后 load 失败留下残留。
|
|
cred, err := s.loadDecryptedCredential(ctx, wsID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// 原子 CAS 抢锁（WHERE sync_status IN ('idle','error')），消除"读后写"TOCTOU。
|
|
// 抢到才继续，否则说明已有 clone/sync 在进行中。
|
|
acquired, err := s.repo.MarkSyncing(ctx, wsID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !acquired {
|
|
return nil, errs.New(errs.CodeWorkspaceSyncInProgress, "sync already in progress")
|
|
}
|
|
// 状态回写与请求生命周期解耦：客户端断开/请求超时也要能写回状态，避免永久卡 syncing。
|
|
writeCtx := context.WithoutCancel(ctx)
|
|
// main 仓库目录缺失（历史 clone 从未成功 / 数据卷丢失）时 fetch 注定失败，
|
|
// 转为重新 clone 自愈：状态切回 cloning 并调度 CloneRunner，由其落 idle/error。
|
|
if _, statErr := os.Stat(ws.MainPath); errors.Is(statErr, os.ErrNotExist) {
|
|
if s.clones == nil {
|
|
_, _ = s.repo.UpdateWorkspaceSyncStatus(writeCtx, wsID, SyncStatusError, ws.LastSyncedAt, "main repo missing on disk; clone scheduler unavailable")
|
|
return nil, errs.New(errs.CodeGitCmdFailed, "main repo missing on disk")
|
|
}
|
|
updated, err := s.repo.UpdateWorkspaceSyncStatus(writeCtx, wsID, SyncStatusCloning, ws.LastSyncedAt, "")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
s.audit(writeCtx, &c.UserID, "workspace.reclone", "workspace", wsID.String(), map[string]any{"reason": "main path missing"})
|
|
s.clones.Schedule(wsID)
|
|
return updated, nil
|
|
}
|
|
fetchCtx, cancel := context.WithTimeout(ctx, s.fetchTimout)
|
|
defer cancel()
|
|
if fetchErr := s.gitr.Fetch(fetchCtx, ws.MainPath, cred); fetchErr != nil {
|
|
errMsg := summarizeErr(fetchErr)
|
|
_, _ = s.repo.UpdateWorkspaceSyncStatus(writeCtx, wsID, SyncStatusError, ws.LastSyncedAt, errMsg)
|
|
s.audit(writeCtx, &c.UserID, "workspace.sync_failed", "workspace", wsID.String(), map[string]any{"err": errMsg})
|
|
return nil, errs.Wrap(fetchErr, errs.CodeGitCmdFailed, "git fetch failed")
|
|
}
|
|
// fetch 只更新远端引用，需 ff-only merge 把 main 工作区当前分支快进到远端。
|
|
if mergeErr := s.gitr.MergeFFOnly(fetchCtx, ws.MainPath); mergeErr != nil {
|
|
errMsg := summarizeErr(mergeErr)
|
|
_, _ = s.repo.UpdateWorkspaceSyncStatus(writeCtx, wsID, SyncStatusError, ws.LastSyncedAt, errMsg)
|
|
s.audit(writeCtx, &c.UserID, "workspace.sync_failed", "workspace", wsID.String(), map[string]any{"err": errMsg})
|
|
return nil, errs.Wrap(mergeErr, errs.CodeGitCmdFailed, "git merge --ff-only failed")
|
|
}
|
|
now := time.Now().UTC()
|
|
updated, err := s.repo.UpdateWorkspaceSyncStatus(writeCtx, wsID, SyncStatusIdle, &now, "")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
s.audit(writeCtx, &c.UserID, "workspace.sync", "workspace", wsID.String(), nil)
|
|
return updated, nil
|
|
}
|
|
|
|
func (s *workspaceService) UpsertCredential(ctx context.Context, c Caller, wsID uuid.UUID, in UpsertCredentialInput) (*Credential, error) {
|
|
ws, err := s.repo.GetWorkspaceByID(ctx, wsID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if _, canWrite, err := s.pa.ResolveByID(ctx, c.UserID, c.IsAdmin, ws.ProjectID); err != nil {
|
|
return nil, err
|
|
} else if !canWrite {
|
|
return nil, errs.New(errs.CodeForbidden, "no write access")
|
|
}
|
|
if !in.Kind.IsValid() {
|
|
return nil, errs.New(errs.CodeGitCredentialInvalid, "kind invalid")
|
|
}
|
|
cred, err := s.encryptForUpsert(wsID, in)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
saved, err := s.repo.UpsertCredential(ctx, cred)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
s.audit(ctx, &c.UserID, "credential.update", "workspace", wsID.String(), map[string]any{"kind": string(in.Kind)})
|
|
saved.Secret = nil
|
|
return saved, nil
|
|
}
|
|
|
|
func (s *workspaceService) GetCredentialMeta(ctx context.Context, c Caller, wsID uuid.UUID) (*Credential, error) {
|
|
ws, err := s.repo.GetWorkspaceByID(ctx, wsID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if canRead, _, err := s.pa.ResolveByID(ctx, c.UserID, c.IsAdmin, ws.ProjectID); err != nil {
|
|
return nil, err
|
|
} else if !canRead {
|
|
return nil, errs.New(errs.CodeForbidden, "no read access")
|
|
}
|
|
cred, err := s.repo.GetCredentialByWorkspace(ctx, wsID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
cred.Secret = nil
|
|
return cred, nil
|
|
}
|
|
|
|
func (s *workspaceService) ListBranches(ctx context.Context, c Caller, wsID uuid.UUID) ([]string, error) {
|
|
ws, err := s.repo.GetWorkspaceByID(ctx, wsID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if canRead, _, err := s.pa.ResolveByID(ctx, c.UserID, c.IsAdmin, ws.ProjectID); err != nil {
|
|
return nil, err
|
|
} else if !canRead {
|
|
return nil, errs.New(errs.CodeForbidden, "no read access")
|
|
}
|
|
if ws.MainPath == "" {
|
|
return nil, errs.New(errs.CodeInvalidInput, "workspace not yet cloned")
|
|
}
|
|
if _, statErr := os.Stat(ws.MainPath); errors.Is(statErr, os.ErrNotExist) {
|
|
return nil, errs.New(errs.CodeInvalidInput, "workspace not yet cloned")
|
|
}
|
|
cred, err := s.loadDecryptedCredential(ctx, wsID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return s.gitr.ListRemoteBranches(ctx, ws.MainPath, cred)
|
|
}
|
|
|
|
func (s *workspaceService) encryptForUpsert(wsID uuid.UUID, in UpsertCredentialInput) (*Credential, error) {
|
|
cred := &Credential{
|
|
ID: uuid.New(),
|
|
WorkspaceID: wsID,
|
|
Kind: in.Kind,
|
|
Username: in.Username,
|
|
}
|
|
if in.Kind == CredKindNone {
|
|
return cred, nil
|
|
}
|
|
if len(in.Secret) == 0 {
|
|
return nil, errs.New(errs.CodeGitCredentialInvalid, "secret required")
|
|
}
|
|
enc, err := s.enc.Encrypt(in.Secret)
|
|
if err != nil {
|
|
return nil, errs.Wrap(err, errs.CodeInternal, "encrypt secret")
|
|
}
|
|
cred.Secret = enc
|
|
cred.Fingerprint = fingerprint(in.Kind, in.Secret)
|
|
return cred, nil
|
|
}
|
|
|
|
// CredentialAccessor 让 app 装配代码不重复实现凭据加载。
|
|
// CloneRunner 与 GitOpsService 都通过这个窄接口拿明文 git.Credential。
|
|
type CredentialAccessor interface {
|
|
LoadDecryptedCredential(ctx context.Context, wsID uuid.UUID) (*git.Credential, error)
|
|
}
|
|
|
|
// SchedulerSetter 让 app 装配代码在创建 cloneRunner 后回填到 service。
|
|
// NewWorkspaceService 阶段允许传 nil scheduler；构造完 cloneRunner 再 SetScheduler。
|
|
type SchedulerSetter interface {
|
|
SetScheduler(s CloneScheduler)
|
|
}
|
|
|
|
// LoadDecryptedCredential 暴露 loadDecryptedCredential 给 app 装配代码（CloneRunner、GitOps 共用）。
|
|
func (s *workspaceService) LoadDecryptedCredential(ctx context.Context, wsID uuid.UUID) (*git.Credential, error) {
|
|
return s.loadDecryptedCredential(ctx, wsID)
|
|
}
|
|
|
|
// SetScheduler 在装配链路把后构造的 CloneRunner 回填到 service（NewWorkspaceService 阶段允许 nil）。
|
|
func (s *workspaceService) SetScheduler(sched CloneScheduler) {
|
|
s.clones = sched
|
|
}
|
|
|
|
func (s *workspaceService) loadDecryptedCredential(ctx context.Context, wsID uuid.UUID) (*git.Credential, error) {
|
|
cred, err := s.repo.GetCredentialByWorkspace(ctx, wsID)
|
|
if err != nil {
|
|
var ae *errs.AppError
|
|
if errors.As(err, &ae) && ae.Code == errs.CodeNotFound {
|
|
return &git.Credential{Kind: git.CredentialNone}, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
if cred.Kind == CredKindNone || len(cred.Secret) == 0 {
|
|
return &git.Credential{Kind: git.CredentialNone}, nil
|
|
}
|
|
plain, err := s.enc.Decrypt(cred.Secret)
|
|
if err != nil {
|
|
return nil, errs.Wrap(err, errs.CodeInternal, "decrypt secret")
|
|
}
|
|
return &git.Credential{
|
|
Kind: git.CredentialKind(cred.Kind),
|
|
Username: cred.Username,
|
|
Secret: plain,
|
|
}, nil
|
|
}
|
|
|
|
func (s *workspaceService) audit(ctx context.Context, uid *uuid.UUID, action, ttype, tid string, meta map[string]any) {
|
|
_ = s.rec.Record(ctx, audit.Entry{
|
|
UserID: uid, Action: action, TargetType: ttype, TargetID: tid, Metadata: meta,
|
|
})
|
|
}
|
|
|
|
func firstNonEmpty(a, b string) string {
|
|
if a != "" {
|
|
return a
|
|
}
|
|
return b
|
|
}
|
|
|
|
func summarizeErr(err error) string {
|
|
if err == nil {
|
|
return ""
|
|
}
|
|
s := err.Error()
|
|
if len(s) > 1000 {
|
|
return s[:1000] + "...[truncated]"
|
|
}
|
|
return s
|
|
}
|
|
|
|
func fingerprint(kind CredentialKind, secret []byte) string {
|
|
switch kind {
|
|
case CredKindPAT:
|
|
if len(secret) <= 4 {
|
|
return "****"
|
|
}
|
|
return "****" + string(secret[len(secret)-4:])
|
|
case CredKindSSHKey:
|
|
sum := sha256.Sum256(secret)
|
|
return "SHA256:" + hex.EncodeToString(sum[:])[:16]
|
|
default:
|
|
return ""
|
|
}
|
|
}
|
|
|
|
var _ = errors.Is
|
|
var _ = fmt.Sprintf
|