agentic-coding-workflow/internal/workspace/repository.go

// 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)
	// GetWorktreeByIDForUpdate 在事务内对 id 行加 FOR UPDATE — Acquire/Release 必须用它，
	// GetWorktreeByID 不带行锁，并发时会 TOCTOU。
	GetWorktreeByIDForUpdate(ctx context.Context, id uuid.UUID) (*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
	ListWorktreesNeedingPruneWarning(ctx context.Context, lastUsedBefore time.Time) ([]*Worktree, error)
	ListWorktreesNeedingPrune(ctx context.Context, lastUsedBefore time.Time) ([]*Worktree, error)
	MarkPruneWarning(ctx context.Context, id uuid.UUID) error
	TryStartPrune(ctx context.Context, id uuid.UUID) (*Worktree, error)
	FinishPruneSuccess(ctx context.Context, id uuid.UUID) error
	FinishPruneRollback(ctx context.Context, id uuid.UUID) error

	// Workspace fetch (jobs runner)
	ListFetchTargets(ctx context.Context) ([]FetchTarget, error)
	MarkSyncing(ctx context.Context, id uuid.UUID) (bool, error)
	MarkFetchResult(ctx context.Context, id uuid.UUID, success bool, errMsg string) 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
}

// ===== Workspace fetch (jobs runner) =====

func (r *pgRepo) ListFetchTargets(ctx context.Context) ([]FetchTarget, error) {
	rows, err := r.q.ListFetchTargets(ctx)
	if err != nil {
		return nil, errs.Wrap(err, errs.CodeInternal, "list fetch targets")
	}
	out := make([]FetchTarget, 0, len(rows))
	for _, row := range rows {
		out = append(out, FetchTarget{
			ID:       fromPgUUID(row.ID),
			OwnerID:  fromPgUUID(row.OwnerID),
			Name:     row.Name,
			MainPath: row.MainPath,
		})
	}
	return out, nil
}

func (r *pgRepo) MarkSyncing(ctx context.Context, id uuid.UUID) (bool, error) {
	n, err := r.q.MarkSyncingCAS(ctx, toPgUUID(id))
	if err != nil {
		return false, errs.Wrap(err, errs.CodeInternal, "mark syncing")
	}
	return n > 0, nil
}

func (r *pgRepo) MarkFetchResult(ctx context.Context, id uuid.UUID, success bool, errMsg string) error {
	var lastErr *string
	if errMsg != "" {
		lastErr = &errMsg
	}
	err := r.q.MarkFetchResult(ctx, workspacesqlc.MarkFetchResultParams{
		ID:            toPgUUID(id),
		Column2:       success,
		LastSyncError: lastErr,
	})
	if err != nil {
		return errs.Wrap(err, errs.CodeInternal, "mark fetch result")
	}
	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,
		PruneWarningAt: fromPgTimePtr(r.PruneWarningAt),
		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
}

func (r *pgRepo) ListWorktreesNeedingPruneWarning(ctx context.Context, lastUsedBefore time.Time) ([]*Worktree, error) {
	ts := pgtype.Timestamptz{Time: lastUsedBefore, Valid: true}
	rows, err := r.q.ListWorktreesNeedingPruneWarning(ctx, ts)
	if err != nil {
		return nil, errs.Wrap(err, errs.CodeInternal, "list worktrees needing prune warning")
	}
	out := make([]*Worktree, 0, len(rows))
	for _, row := range rows {
		out = append(out, rowToWorktree(row))
	}
	return out, nil
}

func (r *pgRepo) ListWorktreesNeedingPrune(ctx context.Context, lastUsedBefore time.Time) ([]*Worktree, error) {
	ts := pgtype.Timestamptz{Time: lastUsedBefore, Valid: true}
	rows, err := r.q.ListWorktreesNeedingPrune(ctx, ts)
	if err != nil {
		return nil, errs.Wrap(err, errs.CodeInternal, "list worktrees needing prune")
	}
	out := make([]*Worktree, 0, len(rows))
	for _, row := range rows {
		out = append(out, rowToWorktree(row))
	}
	return out, nil
}

func (r *pgRepo) MarkPruneWarning(ctx context.Context, id uuid.UUID) error {
	if err := r.q.MarkPruneWarning(ctx, toPgUUID(id)); err != nil {
		return errs.Wrap(err, errs.CodeInternal, "mark prune warning")
	}
	return nil
}

func (r *pgRepo) TryStartPrune(ctx context.Context, id uuid.UUID) (*Worktree, error) {
	row, err := r.q.TryStartPrune(ctx, toPgUUID(id))
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, errs.New(errs.CodeWorktreeAlreadyActive, "worktree no longer idle")
	}
	if err != nil {
		return nil, errs.Wrap(err, errs.CodeInternal, "try start prune")
	}
	return rowToWorktree(row), nil
}

func (r *pgRepo) FinishPruneSuccess(ctx context.Context, id uuid.UUID) error {
	if err := r.q.FinishPruneSuccess(ctx, toPgUUID(id)); err != nil {
		return errs.Wrap(err, errs.CodeInternal, "finish prune success")
	}
	return nil
}

func (r *pgRepo) FinishPruneRollback(ctx context.Context, id uuid.UUID) error {
	if err := r.q.FinishPruneRollback(ctx, toPgUUID(id)); err != nil {
		return errs.Wrap(err, errs.CodeInternal, "finish prune rollback")
	}
	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) GetWorktreeByIDForUpdate(ctx context.Context, id uuid.UUID) (*Worktree, error) {
	row, err := t.q.GetWorktreeByIDForUpdate(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, "select for update")
	}
	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)})
	})
}