diff --git a/internal/workspace/repository.go b/internal/workspace/repository.go new file mode 100644 index 0000000..0457c4b --- /dev/null +++ b/internal/workspace/repository.go @@ -0,0 +1,408 @@ +// Package workspace 内的 repository.go 是 sqlc 生成层之上的薄包装:把 +// pgtype.* 与领域类型互转,把 pgx.ErrNoRows 翻译成 errs.NotFound,把 +// pg unique_violation 暴露为可识别的 error 让 service 决定是否 409。 +package workspace + +import ( + "context" + "errors" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgerrcode" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/yan1h/agent-coding-workflow/internal/infra/errs" + workspacesqlc "github.com/yan1h/agent-coding-workflow/internal/workspace/sqlc" +) + +// Tx 是 repo 暴露给 service 的事务句柄;service 只需调 SelectForUpdate 与 +// 状态转移方法,不直接操作 pgx.Tx。 +type Tx interface { + GetWorktreeForUpdate(ctx context.Context, wsID uuid.UUID, branch string) (*Worktree, error) + GetWorktreeByID(ctx context.Context, id uuid.UUID) (*Worktree, error) + InsertWorktree(ctx context.Context, w *Worktree) (*Worktree, error) + SetWorktreeActive(ctx context.Context, id uuid.UUID, holder string) (*Worktree, error) + SetWorktreeIdle(ctx context.Context, id uuid.UUID) (*Worktree, error) + SetWorktreePruning(ctx context.Context, id uuid.UUID) (*Worktree, error) +} + +// Repository 是 workspace 模块对 PG 的全部依赖。所有方法都接受 context 并返回 +// 领域类型。Service 单元测试通过手写 fakeRepo 满足该接口。 +type Repository interface { + // Workspace + CreateWorkspace(ctx context.Context, w *Workspace) (*Workspace, error) + GetWorkspaceByID(ctx context.Context, id uuid.UUID) (*Workspace, error) + GetWorkspaceBySlug(ctx context.Context, projectSlug, wsSlug string) (*Workspace, error) + ListWorkspacesByProject(ctx context.Context, projectID uuid.UUID) ([]*Workspace, error) + UpdateWorkspaceCore(ctx context.Context, id uuid.UUID, name, description, defaultBranch string) (*Workspace, error) + UpdateWorkspaceSyncStatus(ctx context.Context, id uuid.UUID, st SyncStatus, syncedAt *time.Time, errMsg string) (*Workspace, error) + DeleteWorkspace(ctx context.Context, id uuid.UUID) error + ResetStuckSyncStatuses(ctx context.Context) error + + // Worktree (no-tx 部分) + GetWorktreeByID(ctx context.Context, id uuid.UUID) (*Worktree, error) + ListWorktreesByWorkspace(ctx context.Context, wsID uuid.UUID) ([]*Worktree, error) + DeleteWorktree(ctx context.Context, id uuid.UUID) error + + // Credential + UpsertCredential(ctx context.Context, c *Credential) (*Credential, error) + GetCredentialByWorkspace(ctx context.Context, wsID uuid.UUID) (*Credential, error) + + // Tx 是 acquire/release 用的事务入口 + InTx(ctx context.Context, fn func(Tx) error) error +} + +// IsUniqueViolation 报告 err 是否为 PG 唯一约束冲突。 +func IsUniqueViolation(err error) bool { + var pg *pgconn.PgError + return errors.As(err, &pg) && pg.Code == pgerrcode.UniqueViolation +} + +type pgRepo struct { + pool *pgxpool.Pool + q *workspacesqlc.Queries +} + +// NewPostgresRepository 用 pgxpool 构造 Repository。 +func NewPostgresRepository(pool *pgxpool.Pool) Repository { + return &pgRepo{pool: pool, q: workspacesqlc.New(pool)} +} + +// ===== 类型转换 helper ===== + +func toPgUUID(u uuid.UUID) pgtype.UUID { return pgtype.UUID{Bytes: u, Valid: true} } +func fromPgUUID(p pgtype.UUID) uuid.UUID { + if !p.Valid { + return uuid.Nil + } + return uuid.UUID(p.Bytes) +} +func toPgText(s string) *string { + if s == "" { + return nil + } + return &s +} +func fromPtrString(s *string) string { + if s == nil { + return "" + } + return *s +} +func toPgTimePtr(t *time.Time) pgtype.Timestamptz { + if t == nil { + return pgtype.Timestamptz{Valid: false} + } + return pgtype.Timestamptz{Time: *t, Valid: true} +} +func fromPgTimePtr(t pgtype.Timestamptz) *time.Time { + if !t.Valid { + return nil + } + v := t.Time + return &v +} + +// ===== Workspace ===== + +func rowToWorkspace(r workspacesqlc.Workspace) *Workspace { + return &Workspace{ + ID: fromPgUUID(r.ID), + ProjectID: fromPgUUID(r.ProjectID), + Slug: r.Slug, + Name: r.Name, + Description: r.Description, + GitRemoteURL: r.GitRemoteUrl, + DefaultBranch: r.DefaultBranch, + MainPath: r.MainPath, + SyncStatus: SyncStatus(r.SyncStatus), + LastSyncedAt: fromPgTimePtr(r.LastSyncedAt), + LastSyncError: fromPtrString(r.LastSyncError), + CreatedAt: r.CreatedAt.Time, + UpdatedAt: r.UpdatedAt.Time, + } +} + +func (r *pgRepo) CreateWorkspace(ctx context.Context, w *Workspace) (*Workspace, error) { + row, err := r.q.CreateWorkspace(ctx, workspacesqlc.CreateWorkspaceParams{ + ID: toPgUUID(w.ID), + ProjectID: toPgUUID(w.ProjectID), + Slug: w.Slug, + Name: w.Name, + Description: w.Description, + GitRemoteUrl: w.GitRemoteURL, + DefaultBranch: w.DefaultBranch, + MainPath: w.MainPath, + SyncStatus: string(w.SyncStatus), + }) + if err != nil { + if IsUniqueViolation(err) { + return nil, errs.Wrap(err, errs.CodeWorkspaceSlugTaken, "workspace slug already used in project") + } + return nil, errs.Wrap(err, errs.CodeInternal, "create workspace") + } + return rowToWorkspace(row), nil +} + +func (r *pgRepo) GetWorkspaceByID(ctx context.Context, id uuid.UUID) (*Workspace, error) { + row, err := r.q.GetWorkspaceByID(ctx, toPgUUID(id)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, errs.New(errs.CodeNotFound, "workspace not found") + } + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "get workspace") + } + return rowToWorkspace(row), nil +} + +func (r *pgRepo) GetWorkspaceBySlug(ctx context.Context, projectSlug, wsSlug string) (*Workspace, error) { + row, err := r.q.GetWorkspaceBySlug(ctx, workspacesqlc.GetWorkspaceBySlugParams{ + Slug: projectSlug, + Slug_2: wsSlug, + }) + if errors.Is(err, pgx.ErrNoRows) { + return nil, errs.New(errs.CodeNotFound, "workspace not found") + } + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "get workspace by slug") + } + return rowToWorkspace(row), nil +} + +func (r *pgRepo) ListWorkspacesByProject(ctx context.Context, projectID uuid.UUID) ([]*Workspace, error) { + rows, err := r.q.ListWorkspacesByProject(ctx, toPgUUID(projectID)) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "list workspaces") + } + out := make([]*Workspace, 0, len(rows)) + for _, row := range rows { + out = append(out, rowToWorkspace(row)) + } + return out, nil +} + +func (r *pgRepo) UpdateWorkspaceCore(ctx context.Context, id uuid.UUID, name, description, defaultBranch string) (*Workspace, error) { + row, err := r.q.UpdateWorkspaceCore(ctx, workspacesqlc.UpdateWorkspaceCoreParams{ + ID: toPgUUID(id), + Name: name, + Description: description, + DefaultBranch: defaultBranch, + }) + if errors.Is(err, pgx.ErrNoRows) { + return nil, errs.New(errs.CodeNotFound, "workspace not found") + } + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "update workspace") + } + return rowToWorkspace(row), nil +} + +func (r *pgRepo) UpdateWorkspaceSyncStatus(ctx context.Context, id uuid.UUID, st SyncStatus, syncedAt *time.Time, errMsg string) (*Workspace, error) { + row, err := r.q.UpdateWorkspaceSyncStatus(ctx, workspacesqlc.UpdateWorkspaceSyncStatusParams{ + ID: toPgUUID(id), + SyncStatus: string(st), + LastSyncedAt: toPgTimePtr(syncedAt), + LastSyncError: toPgText(errMsg), + }) + if errors.Is(err, pgx.ErrNoRows) { + return nil, errs.New(errs.CodeNotFound, "workspace not found") + } + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "update sync status") + } + return rowToWorkspace(row), nil +} + +// DeleteWorkspace 是 :exec — 不存在时静默返回 nil。需要 404 语义的调用方应先 +// GetWorkspaceByID 校验存在性(service 层鉴权也通常这样做)。 +func (r *pgRepo) DeleteWorkspace(ctx context.Context, id uuid.UUID) error { + if err := r.q.DeleteWorkspace(ctx, toPgUUID(id)); err != nil { + return errs.Wrap(err, errs.CodeInternal, "delete workspace") + } + return nil +} + +func (r *pgRepo) ResetStuckSyncStatuses(ctx context.Context) error { + if err := r.q.ResetStuckSyncStatuses(ctx); err != nil { + return errs.Wrap(err, errs.CodeInternal, "reset stuck sync") + } + return nil +} + +// ===== Worktree ===== + +func rowToWorktree(r workspacesqlc.WorkspaceWorktree) *Worktree { + return &Worktree{ + ID: fromPgUUID(r.ID), + WorkspaceID: fromPgUUID(r.WorkspaceID), + Branch: r.Branch, + Path: r.Path, + Status: WorktreeStatus(r.Status), + ActiveHolder: fromPtrString(r.ActiveHolder), + AcquiredAt: fromPgTimePtr(r.AcquiredAt), + LastUsedAt: r.LastUsedAt.Time, + CreatedAt: r.CreatedAt.Time, + } +} + +func (r *pgRepo) GetWorktreeByID(ctx context.Context, id uuid.UUID) (*Worktree, error) { + row, err := r.q.GetWorktreeByID(ctx, toPgUUID(id)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, errs.New(errs.CodeNotFound, "worktree not found") + } + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "get worktree") + } + return rowToWorktree(row), nil +} + +func (r *pgRepo) ListWorktreesByWorkspace(ctx context.Context, wsID uuid.UUID) ([]*Worktree, error) { + rows, err := r.q.ListWorktreesByWorkspace(ctx, toPgUUID(wsID)) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "list worktrees") + } + out := make([]*Worktree, 0, len(rows)) + for _, row := range rows { + out = append(out, rowToWorktree(row)) + } + return out, nil +} + +// DeleteWorktree 是 :exec — 不存在时静默返回 nil。同 DeleteWorkspace 注释。 +func (r *pgRepo) DeleteWorktree(ctx context.Context, id uuid.UUID) error { + if err := r.q.DeleteWorktree(ctx, toPgUUID(id)); err != nil { + return errs.Wrap(err, errs.CodeInternal, "delete worktree") + } + return nil +} + +// ===== Credential ===== + +func rowToCredential(r workspacesqlc.GitCredential) *Credential { + return &Credential{ + ID: fromPgUUID(r.ID), + WorkspaceID: fromPgUUID(r.WorkspaceID), + Kind: CredentialKind(r.Kind), + Username: fromPtrString(r.Username), + Fingerprint: fromPtrString(r.Fingerprint), + CreatedAt: r.CreatedAt.Time, + UpdatedAt: r.UpdatedAt.Time, + // Secret 永远不读 + } +} + +func (r *pgRepo) UpsertCredential(ctx context.Context, c *Credential) (*Credential, error) { + row, err := r.q.UpsertCredential(ctx, workspacesqlc.UpsertCredentialParams{ + ID: toPgUUID(c.ID), + WorkspaceID: toPgUUID(c.WorkspaceID), + Kind: string(c.Kind), + Username: toPgText(c.Username), + EncryptedSecret: c.Secret, + Fingerprint: toPgText(c.Fingerprint), + }) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "upsert credential") + } + return rowToCredential(row), nil +} + +// GetCredentialByWorkspace 返回的 *Credential 不在普通读路径上暴露 Secret,但 +// 这里保留 EncryptedSecret 给 service 调 crypto.Decrypt 解开。其它读路径 +// (如 GetCredentialMeta)应在 service 层显式清空 Secret 字段。 +func (r *pgRepo) GetCredentialByWorkspace(ctx context.Context, wsID uuid.UUID) (*Credential, error) { + row, err := r.q.GetCredentialByWorkspace(ctx, toPgUUID(wsID)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, errs.New(errs.CodeNotFound, "credential not found") + } + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "get credential") + } + c := rowToCredential(row) + c.Secret = row.EncryptedSecret // 仅密文,service 调 crypto.Decrypt 解开 + return c, nil +} + +// ===== Tx 实现 ===== + +type pgTx struct{ q *workspacesqlc.Queries } + +func (t *pgTx) GetWorktreeForUpdate(ctx context.Context, wsID uuid.UUID, branch string) (*Worktree, error) { + row, err := t.q.GetWorktreeByBranchForUpdate(ctx, workspacesqlc.GetWorktreeByBranchForUpdateParams{ + WorkspaceID: toPgUUID(wsID), + Branch: branch, + }) + if errors.Is(err, pgx.ErrNoRows) { + return nil, errs.New(errs.CodeNotFound, "worktree not found") + } + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "select for update") + } + return rowToWorktree(row), nil +} + +func (t *pgTx) GetWorktreeByID(ctx context.Context, id uuid.UUID) (*Worktree, error) { + row, err := t.q.GetWorktreeByID(ctx, toPgUUID(id)) + if errors.Is(err, pgx.ErrNoRows) { + return nil, errs.New(errs.CodeNotFound, "worktree not found") + } + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "get worktree") + } + return rowToWorktree(row), nil +} + +func (t *pgTx) InsertWorktree(ctx context.Context, w *Worktree) (*Worktree, error) { + row, err := t.q.CreateWorktree(ctx, workspacesqlc.CreateWorktreeParams{ + ID: toPgUUID(w.ID), + WorkspaceID: toPgUUID(w.WorkspaceID), + Branch: w.Branch, + Path: w.Path, + Status: string(w.Status), + ActiveHolder: toPgText(w.ActiveHolder), + AcquiredAt: toPgTimePtr(w.AcquiredAt), + }) + if err != nil { + if IsUniqueViolation(err) { + return nil, errs.Wrap(err, errs.CodeWorktreeBranchConflict, "worktree branch conflict") + } + return nil, errs.Wrap(err, errs.CodeInternal, "insert worktree") + } + return rowToWorktree(row), nil +} + +func (t *pgTx) SetWorktreeActive(ctx context.Context, id uuid.UUID, holder string) (*Worktree, error) { + row, err := t.q.SetWorktreeActive(ctx, workspacesqlc.SetWorktreeActiveParams{ + ID: toPgUUID(id), + ActiveHolder: toPgText(holder), + }) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "set active") + } + return rowToWorktree(row), nil +} + +func (t *pgTx) SetWorktreeIdle(ctx context.Context, id uuid.UUID) (*Worktree, error) { + row, err := t.q.SetWorktreeIdle(ctx, toPgUUID(id)) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "set idle") + } + return rowToWorktree(row), nil +} + +func (t *pgTx) SetWorktreePruning(ctx context.Context, id uuid.UUID) (*Worktree, error) { + row, err := t.q.SetWorktreePruning(ctx, toPgUUID(id)) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "set pruning") + } + return rowToWorktree(row), nil +} + +func (r *pgRepo) InTx(ctx context.Context, fn func(Tx) error) error { + return pgx.BeginFunc(ctx, r.pool, func(tx pgx.Tx) error { + return fn(&pgTx{q: r.q.WithTx(tx)}) + }) +} diff --git a/internal/workspace/repository_test.go b/internal/workspace/repository_test.go new file mode 100644 index 0000000..85be473 --- /dev/null +++ b/internal/workspace/repository_test.go @@ -0,0 +1,138 @@ +//go:build integration + +package workspace + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/require" + tcpg "github.com/testcontainers/testcontainers-go/modules/postgres" + + "github.com/yan1h/agent-coding-workflow/internal/infra/db" +) + +// setupRepo 起一个临时 postgres 容器、跑迁移、注入一条 user/project 行做外键, +// 返回可直接调用的 Repository 与所属 project_id。每个测试用独立容器,互不干扰, +// 便于 Parallel。 +func setupRepo(t *testing.T) (Repository, *pgxpool.Pool, uuid.UUID) { + t.Helper() + ctx := context.Background() + container, err := tcpg.Run(ctx, "postgres:16-alpine", + tcpg.WithDatabase("acw"), + tcpg.WithUsername("acw"), + tcpg.WithPassword("acw"), + ) + require.NoError(t, err) + t.Cleanup(func() { _ = container.Terminate(ctx) }) + + dsn, err := container.ConnectionString(ctx, "sslmode=disable") + require.NoError(t, err) + + require.NoError(t, db.Migrate(ctx, dsn, "file://../../migrations")) + + pool, err := db.NewPool(ctx, dsn) + require.NoError(t, err) + t.Cleanup(pool.Close) + + uid := uuid.New() + _, err = pool.Exec(ctx, `INSERT INTO users (id,email,password_hash,display_name,is_admin,created_at,updated_at) + VALUES ($1,'a@b.c','x','admin',true,now(),now())`, uid) + require.NoError(t, err) + pid := uuid.New() + _, err = pool.Exec(ctx, `INSERT INTO projects (id,slug,name,owner_id,visibility) VALUES ($1,'p','P',$2,'private')`, pid, uid) + require.NoError(t, err) + return NewPostgresRepository(pool), pool, pid +} + +func TestRepo_CreateWorkspaceAndRead(t *testing.T) { + t.Parallel() + repo, _, pid := setupRepo(t) + ctx := context.Background() + id := uuid.New() + ws, err := repo.CreateWorkspace(ctx, &Workspace{ + ID: id, ProjectID: pid, Slug: "ws1", Name: "W1", + GitRemoteURL: "https://x", DefaultBranch: "main", MainPath: "/data/ws1/main", + SyncStatus: SyncStatusCloning, + }) + require.NoError(t, err) + require.Equal(t, "ws1", ws.Slug) + + got, err := repo.GetWorkspaceByID(ctx, id) + require.NoError(t, err) + require.Equal(t, "W1", got.Name) +} + +func TestRepo_SyncStatusRoundtrip(t *testing.T) { + t.Parallel() + repo, _, pid := setupRepo(t) + ctx := context.Background() + id := uuid.New() + _, err := repo.CreateWorkspace(ctx, &Workspace{ + ID: id, ProjectID: pid, Slug: "ws2", Name: "W2", + GitRemoteURL: "https://x", DefaultBranch: "main", MainPath: "/x", SyncStatus: SyncStatusCloning, + }) + require.NoError(t, err) + now := time.Now().UTC() + _, err = repo.UpdateWorkspaceSyncStatus(ctx, id, SyncStatusError, &now, "boom") + require.NoError(t, err) + got, err := repo.GetWorkspaceByID(ctx, id) + require.NoError(t, err) + require.Equal(t, SyncStatusError, got.SyncStatus) + require.Equal(t, "boom", got.LastSyncError) +} + +func TestRepo_AcquireRelease_Concurrent(t *testing.T) { + t.Parallel() + repo, _, pid := setupRepo(t) + ctx := context.Background() + wsID := uuid.New() + _, err := repo.CreateWorkspace(ctx, &Workspace{ + ID: wsID, ProjectID: pid, Slug: "ws3", Name: "W3", + GitRemoteURL: "https://x", DefaultBranch: "main", MainPath: "/x", SyncStatus: SyncStatusIdle, + }) + require.NoError(t, err) + + first := make(chan error, 1) + second := make(chan error, 1) + holderA, holderB := "user:a", "user:b" + go func() { + first <- repo.InTx(ctx, func(tx Tx) error { + wt, err := tx.GetWorktreeForUpdate(ctx, wsID, "feat") + if err == nil && wt.Status == WorktreeStatusActive { + return nil + } + _, err = tx.InsertWorktree(ctx, &Worktree{ + ID: uuid.New(), WorkspaceID: wsID, Branch: "feat", + Path: "/x/.worktrees/feat", Status: WorktreeStatusActive, + ActiveHolder: holderA, + }) + return err + }) + }() + <-first + go func() { + second <- repo.InTx(ctx, func(tx Tx) error { + wt, err := tx.GetWorktreeForUpdate(ctx, wsID, "feat") + if err != nil { + return err + } + if wt.Status == WorktreeStatusActive && wt.ActiveHolder != holderB { + return errsAlreadyActive() + } + return nil + }) + }() + require.Error(t, <-second) +} + +func errsAlreadyActive() error { + return &shimErr{} +} + +type shimErr struct{} + +func (e *shimErr) Error() string { return "already active" }