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