You've already forked agentic-coding-workflow
feat(workspace): WorkspaceService + async clone runner
This commit is contained in:
@@ -0,0 +1,88 @@
|
||||
package workspace
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/yan1h/agent-coding-workflow/internal/audit"
|
||||
"github.com/yan1h/agent-coding-workflow/internal/infra/git"
|
||||
)
|
||||
|
||||
// CloneRunner 用 goroutine 异步跑 git clone。Schedule 是非阻塞的：
|
||||
// 调用方立即返回，clone 进度通过 sync_status 列轮询。
|
||||
//
|
||||
// Plan #5（Job Runner）实现后会替换 Schedule 内部为 job dispatcher 入队，
|
||||
// 调用点不需要改动。
|
||||
type CloneRunner struct {
|
||||
repo Repository
|
||||
rec audit.Recorder
|
||||
gitr git.Runner
|
||||
log *slog.Logger
|
||||
timeout time.Duration
|
||||
|
||||
wg sync.WaitGroup
|
||||
resolveCred func(ctx context.Context, wsID uuid.UUID) (*git.Credential, error)
|
||||
}
|
||||
|
||||
// NewCloneRunner 构造 CloneRunner。timeout==0 时使用默认 30 分钟超时。
|
||||
func NewCloneRunner(
|
||||
repo Repository, rec audit.Recorder, gitr git.Runner, log *slog.Logger,
|
||||
resolveCred func(context.Context, uuid.UUID) (*git.Credential, error),
|
||||
timeout time.Duration,
|
||||
) *CloneRunner {
|
||||
if timeout == 0 {
|
||||
timeout = 30 * time.Minute
|
||||
}
|
||||
return &CloneRunner{
|
||||
repo: repo, rec: rec, gitr: gitr, log: log,
|
||||
timeout: timeout, resolveCred: resolveCred,
|
||||
}
|
||||
}
|
||||
|
||||
// Schedule 非阻塞地排程一次 clone；调用方立即返回。
|
||||
func (r *CloneRunner) Schedule(wsID uuid.UUID) {
|
||||
r.wg.Add(1)
|
||||
go func() {
|
||||
defer r.wg.Done()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), r.timeout)
|
||||
defer cancel()
|
||||
r.run(ctx, wsID)
|
||||
}()
|
||||
}
|
||||
|
||||
// Wait 阻塞直到所有已排程的 clone 任务完成；测试与 graceful shutdown 用。
|
||||
func (r *CloneRunner) Wait() { r.wg.Wait() }
|
||||
|
||||
func (r *CloneRunner) run(ctx context.Context, wsID uuid.UUID) {
|
||||
ws, err := r.repo.GetWorkspaceByID(ctx, wsID)
|
||||
if err != nil {
|
||||
r.log.Error("clone runner: get workspace", "ws_id", wsID, "err", err.Error())
|
||||
return
|
||||
}
|
||||
cred, err := r.resolveCred(ctx, wsID)
|
||||
if err != nil {
|
||||
r.markError(ctx, wsID, "resolve credential: "+err.Error())
|
||||
return
|
||||
}
|
||||
if cloneErr := r.gitr.Clone(ctx, ws.MainPath, ws.GitRemoteURL, ws.DefaultBranch, cred); cloneErr != nil {
|
||||
r.markError(ctx, wsID, summarizeErr(cloneErr))
|
||||
_ = r.rec.Record(ctx, audit.Entry{Action: "workspace.clone_failed", TargetType: "workspace", TargetID: wsID.String(), Metadata: map[string]any{"err": summarizeErr(cloneErr)}})
|
||||
return
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
if _, err := r.repo.UpdateWorkspaceSyncStatus(ctx, wsID, SyncStatusIdle, &now, ""); err != nil {
|
||||
r.log.Error("clone runner: update sync status idle", "ws_id", wsID, "err", err.Error())
|
||||
return
|
||||
}
|
||||
_ = r.rec.Record(ctx, audit.Entry{Action: "workspace.clone_ok", TargetType: "workspace", TargetID: wsID.String()})
|
||||
}
|
||||
|
||||
func (r *CloneRunner) markError(ctx context.Context, wsID uuid.UUID, msg string) {
|
||||
if _, err := r.repo.UpdateWorkspaceSyncStatus(ctx, wsID, SyncStatusError, nil, msg); err != nil {
|
||||
r.log.Error("clone runner: update sync status error", "ws_id", wsID, "err", err.Error())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,370 @@
|
||||
// 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})
|
||||
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 {
|
||||
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")
|
||||
}
|
||||
if ws.SyncStatus == SyncStatusCloning || ws.SyncStatus == SyncStatusSyncing {
|
||||
return nil, errs.New(errs.CodeWorkspaceSyncInProgress, "sync already in progress")
|
||||
}
|
||||
if _, err := s.repo.UpdateWorkspaceSyncStatus(ctx, wsID, SyncStatusSyncing, ws.LastSyncedAt, ""); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cred, err := s.loadDecryptedCredential(ctx, wsID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
fetchCtx, cancel := context.WithTimeout(ctx, s.fetchTimout)
|
||||
defer cancel()
|
||||
fetchErr := s.gitr.Fetch(fetchCtx, ws.MainPath, cred)
|
||||
if fetchErr != nil {
|
||||
errMsg := summarizeErr(fetchErr)
|
||||
_, _ = s.repo.UpdateWorkspaceSyncStatus(ctx, wsID, SyncStatusError, ws.LastSyncedAt, errMsg)
|
||||
s.audit(ctx, &c.UserID, "workspace.sync_failed", "workspace", wsID.String(), map[string]any{"err": errMsg})
|
||||
return nil, errs.Wrap(fetchErr, errs.CodeGitCmdFailed, "git fetch failed")
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
updated, err := s.repo.UpdateWorkspaceSyncStatus(ctx, wsID, SyncStatusIdle, &now, "")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s.audit(ctx, &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) 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
|
||||
}
|
||||
|
||||
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
|
||||
@@ -0,0 +1,269 @@
|
||||
package workspace
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"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"
|
||||
)
|
||||
|
||||
type fakeRepo struct {
|
||||
mu sync.Mutex
|
||||
wss map[uuid.UUID]*Workspace
|
||||
creds map[uuid.UUID]*Credential
|
||||
wts map[uuid.UUID]*Worktree
|
||||
errOn map[string]error
|
||||
}
|
||||
|
||||
func newFakeRepo() *fakeRepo {
|
||||
return &fakeRepo{
|
||||
wss: map[uuid.UUID]*Workspace{}, creds: map[uuid.UUID]*Credential{},
|
||||
wts: map[uuid.UUID]*Worktree{}, errOn: map[string]error{},
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fakeRepo) CreateWorkspace(_ context.Context, w *Workspace) (*Workspace, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if e := f.errOn["create"]; e != nil {
|
||||
return nil, e
|
||||
}
|
||||
cp := *w
|
||||
f.wss[w.ID] = &cp
|
||||
return &cp, nil
|
||||
}
|
||||
func (f *fakeRepo) GetWorkspaceByID(_ context.Context, id uuid.UUID) (*Workspace, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if w, ok := f.wss[id]; ok {
|
||||
cp := *w
|
||||
return &cp, nil
|
||||
}
|
||||
return nil, errors.New("not found")
|
||||
}
|
||||
func (f *fakeRepo) GetWorkspaceBySlug(_ context.Context, _, _ string) (*Workspace, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (f *fakeRepo) ListWorkspacesByProject(_ context.Context, _ uuid.UUID) ([]*Workspace, error) {
|
||||
out := make([]*Workspace, 0, len(f.wss))
|
||||
for _, w := range f.wss {
|
||||
out = append(out, w)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (f *fakeRepo) UpdateWorkspaceCore(_ context.Context, id uuid.UUID, name, desc, br string) (*Workspace, error) {
|
||||
w := f.wss[id]
|
||||
w.Name, w.Description, w.DefaultBranch = name, desc, br
|
||||
return w, nil
|
||||
}
|
||||
func (f *fakeRepo) UpdateWorkspaceSyncStatus(_ context.Context, id uuid.UUID, st SyncStatus, _ *time.Time, msg string) (*Workspace, error) {
|
||||
w := f.wss[id]
|
||||
w.SyncStatus, w.LastSyncError = st, msg
|
||||
return w, nil
|
||||
}
|
||||
func (f *fakeRepo) DeleteWorkspace(_ context.Context, id uuid.UUID) error {
|
||||
delete(f.wss, id)
|
||||
return nil
|
||||
}
|
||||
func (f *fakeRepo) ResetStuckSyncStatuses(_ context.Context) error { return nil }
|
||||
func (f *fakeRepo) GetWorktreeByID(_ context.Context, id uuid.UUID) (*Worktree, error) {
|
||||
if w, ok := f.wts[id]; ok {
|
||||
return w, nil
|
||||
}
|
||||
return nil, errors.New("not found")
|
||||
}
|
||||
func (f *fakeRepo) ListWorktreesByWorkspace(_ context.Context, _ uuid.UUID) ([]*Worktree, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (f *fakeRepo) DeleteWorktree(_ context.Context, _ uuid.UUID) error { return nil }
|
||||
func (f *fakeRepo) UpsertCredential(_ context.Context, c *Credential) (*Credential, error) {
|
||||
cp := *c
|
||||
f.creds[c.WorkspaceID] = &cp
|
||||
return &cp, nil
|
||||
}
|
||||
func (f *fakeRepo) GetCredentialByWorkspace(_ context.Context, ws uuid.UUID) (*Credential, error) {
|
||||
if c, ok := f.creds[ws]; ok {
|
||||
cp := *c
|
||||
return &cp, nil
|
||||
}
|
||||
return nil, errs.New(errs.CodeNotFound, "credential not found")
|
||||
}
|
||||
func (f *fakeRepo) InTx(_ context.Context, fn func(Tx) error) error { return fn(&fakeTx{f: f}) }
|
||||
|
||||
type fakeTx struct{ f *fakeRepo }
|
||||
|
||||
func (t *fakeTx) GetWorktreeForUpdate(_ context.Context, ws uuid.UUID, br string) (*Worktree, error) {
|
||||
for _, w := range t.f.wts {
|
||||
if w.WorkspaceID == ws && w.Branch == br {
|
||||
cp := *w
|
||||
return &cp, nil
|
||||
}
|
||||
}
|
||||
return nil, &errsNotFound{}
|
||||
}
|
||||
func (t *fakeTx) GetWorktreeByID(_ context.Context, id uuid.UUID) (*Worktree, error) {
|
||||
if w, ok := t.f.wts[id]; ok {
|
||||
cp := *w
|
||||
return &cp, nil
|
||||
}
|
||||
return nil, &errsNotFound{}
|
||||
}
|
||||
func (t *fakeTx) InsertWorktree(_ context.Context, w *Worktree) (*Worktree, error) {
|
||||
cp := *w
|
||||
t.f.wts[w.ID] = &cp
|
||||
return &cp, nil
|
||||
}
|
||||
func (t *fakeTx) SetWorktreeActive(_ context.Context, id uuid.UUID, holder string) (*Worktree, error) {
|
||||
w := t.f.wts[id]
|
||||
w.Status, w.ActiveHolder = WorktreeStatusActive, holder
|
||||
return w, nil
|
||||
}
|
||||
func (t *fakeTx) SetWorktreeIdle(_ context.Context, id uuid.UUID) (*Worktree, error) {
|
||||
w := t.f.wts[id]
|
||||
w.Status, w.ActiveHolder = WorktreeStatusIdle, ""
|
||||
return w, nil
|
||||
}
|
||||
func (t *fakeTx) SetWorktreePruning(_ context.Context, id uuid.UUID) (*Worktree, error) {
|
||||
w := t.f.wts[id]
|
||||
w.Status = WorktreeStatusPruning
|
||||
return w, nil
|
||||
}
|
||||
|
||||
type errsNotFound struct{}
|
||||
|
||||
func (e *errsNotFound) Error() string { return "not found" }
|
||||
|
||||
type fakePA struct {
|
||||
pid uuid.UUID
|
||||
canRead bool
|
||||
canWrite bool
|
||||
}
|
||||
|
||||
func (f *fakePA) Resolve(_ context.Context, _ uuid.UUID, _ bool, _ string) (uuid.UUID, bool, bool, error) {
|
||||
return f.pid, f.canRead, f.canWrite, nil
|
||||
}
|
||||
func (f *fakePA) ResolveByID(_ context.Context, _ uuid.UUID, _ bool, _ uuid.UUID) (bool, bool, error) {
|
||||
return f.canRead, f.canWrite, nil
|
||||
}
|
||||
|
||||
type fakeSched struct{ called []uuid.UUID }
|
||||
|
||||
func (s *fakeSched) Schedule(id uuid.UUID) { s.called = append(s.called, id) }
|
||||
|
||||
type fakeGit struct {
|
||||
cloneErr error
|
||||
fetchErr error
|
||||
cloneSeen []string
|
||||
}
|
||||
|
||||
func (f *fakeGit) Clone(_ context.Context, dst, _, _ string, _ *git.Credential) error {
|
||||
f.cloneSeen = append(f.cloneSeen, dst)
|
||||
return f.cloneErr
|
||||
}
|
||||
func (f *fakeGit) Fetch(_ context.Context, _ string, _ *git.Credential) error { return f.fetchErr }
|
||||
func (f *fakeGit) Push(_ context.Context, _, _ string, _ *git.Credential) error { return nil }
|
||||
func (f *fakeGit) Commit(_ context.Context, _, _ string, _ bool) (string, error) {
|
||||
return "deadbeef", nil
|
||||
}
|
||||
func (f *fakeGit) Status(_ context.Context, _ string) ([]git.FileStatus, error) { return nil, nil }
|
||||
func (f *fakeGit) WorktreeAdd(_ context.Context, _, _, _ string) error { return nil }
|
||||
func (f *fakeGit) WorktreeRemove(_ context.Context, _, _ string) error { return nil }
|
||||
func (f *fakeGit) WorktreeList(_ context.Context, _ string) ([]git.Worktree, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
type noopAudit struct{}
|
||||
|
||||
func (noopAudit) Record(_ context.Context, _ audit.Entry) error { return nil }
|
||||
|
||||
func newSvc(t *testing.T, pa *fakePA, sched CloneScheduler, gitr git.Runner) (*workspaceService, *fakeRepo) {
|
||||
t.Helper()
|
||||
enc, err := crypto.NewEncryptor(make([]byte, 32))
|
||||
require.NoError(t, err)
|
||||
repo := newFakeRepo()
|
||||
svc := NewWorkspaceService(repo, noopAudit{}, enc, gitr, pa, sched, "/data").(*workspaceService)
|
||||
return svc, repo
|
||||
}
|
||||
|
||||
func TestWorkspaceService_Create_HappyPath(t *testing.T) {
|
||||
t.Parallel()
|
||||
pa := &fakePA{pid: uuid.New(), canRead: true, canWrite: true}
|
||||
sched := &fakeSched{}
|
||||
svc, repo := newSvc(t, pa, sched, &fakeGit{})
|
||||
caller := Caller{UserID: uuid.New()}
|
||||
ws, err := svc.Create(context.Background(), caller, "p", CreateWorkspaceInput{
|
||||
Slug: "ws1", Name: "W1", GitRemoteURL: "https://x", DefaultBranch: "main",
|
||||
Credential: UpsertCredentialInput{Kind: CredKindNone},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, SyncStatusCloning, ws.SyncStatus)
|
||||
require.Len(t, sched.called, 1)
|
||||
require.Equal(t, ws.ID, sched.called[0])
|
||||
_, ok := repo.creds[ws.ID]
|
||||
require.True(t, ok)
|
||||
}
|
||||
|
||||
func TestWorkspaceService_Create_BadSlug(t *testing.T) {
|
||||
t.Parallel()
|
||||
pa := &fakePA{pid: uuid.New(), canWrite: true, canRead: true}
|
||||
svc, _ := newSvc(t, pa, &fakeSched{}, &fakeGit{})
|
||||
_, err := svc.Create(context.Background(), Caller{UserID: uuid.New()}, "p", CreateWorkspaceInput{
|
||||
Slug: "BAD-SLUG", Credential: UpsertCredentialInput{Kind: CredKindNone},
|
||||
})
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
func TestWorkspaceService_Sync_BlockedWhenCloning(t *testing.T) {
|
||||
t.Parallel()
|
||||
pa := &fakePA{pid: uuid.New(), canWrite: true, canRead: true}
|
||||
svc, repo := newSvc(t, pa, &fakeSched{}, &fakeGit{})
|
||||
id := uuid.New()
|
||||
repo.wss[id] = &Workspace{ID: id, ProjectID: pa.pid, MainPath: "/x", SyncStatus: SyncStatusCloning}
|
||||
_, err := svc.Sync(context.Background(), Caller{UserID: uuid.New()}, id)
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
func TestCloneRunner_HappyPath(t *testing.T) {
|
||||
t.Parallel()
|
||||
pa := &fakePA{pid: uuid.New(), canWrite: true, canRead: true}
|
||||
gitr := &fakeGit{}
|
||||
svc, repo := newSvc(t, pa, &fakeSched{}, gitr)
|
||||
id := uuid.New()
|
||||
repo.wss[id] = &Workspace{ID: id, MainPath: "/data/main", DefaultBranch: "main", SyncStatus: SyncStatusCloning}
|
||||
runner := NewCloneRunner(repo, noopAudit{}, gitr, slog.Default(),
|
||||
func(ctx context.Context, wsID uuid.UUID) (*git.Credential, error) {
|
||||
return svc.loadDecryptedCredential(ctx, wsID)
|
||||
}, 0)
|
||||
runner.Schedule(id)
|
||||
runner.Wait()
|
||||
got, _ := repo.GetWorkspaceByID(context.Background(), id)
|
||||
require.Equal(t, SyncStatusIdle, got.SyncStatus)
|
||||
}
|
||||
|
||||
func TestCloneRunner_Failure(t *testing.T) {
|
||||
t.Parallel()
|
||||
pa := &fakePA{pid: uuid.New(), canWrite: true, canRead: true}
|
||||
gitr := &fakeGit{cloneErr: errors.New("boom")}
|
||||
svc, repo := newSvc(t, pa, &fakeSched{}, gitr)
|
||||
id := uuid.New()
|
||||
repo.wss[id] = &Workspace{ID: id, MainPath: "/x", DefaultBranch: "main", SyncStatus: SyncStatusCloning}
|
||||
runner := NewCloneRunner(repo, noopAudit{}, gitr, slog.Default(),
|
||||
func(ctx context.Context, wsID uuid.UUID) (*git.Credential, error) {
|
||||
return svc.loadDecryptedCredential(ctx, wsID)
|
||||
}, 0)
|
||||
runner.Schedule(id)
|
||||
runner.Wait()
|
||||
got, _ := repo.GetWorkspaceByID(context.Background(), id)
|
||||
require.Equal(t, SyncStatusError, got.SyncStatus)
|
||||
require.Contains(t, got.LastSyncError, "boom")
|
||||
}
|
||||
Reference in New Issue
Block a user