feat(workspace): postgres repository with type conversions and tx

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)})
+	})
+}

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" }