diff --git a/internal/mcp/repository.go b/internal/mcp/repository.go new file mode 100644 index 0000000..793c30f --- /dev/null +++ b/internal/mcp/repository.go @@ -0,0 +1,202 @@ +package mcp + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/yan1h/agent-coding-workflow/internal/infra/errs" + mcpsqlc "github.com/yan1h/agent-coding-workflow/internal/mcp/sqlc" +) + +// Repository 是 mcp 模块对 PG 的全部依赖。Service 单测通过手写 fakeRepo 满足。 +type Repository interface { + // CRUD + Create(ctx context.Context, t *MCPToken) (*MCPToken, error) + GetByHash(ctx context.Context, hash []byte) (*MCPToken, error) + GetByID(ctx context.Context, id uuid.UUID) (*MCPToken, error) + List(ctx context.Context, callerUserID *uuid.UUID, activeOnly bool) ([]*MCPToken, error) + + // 状态变更 + UpdateLastUsed(ctx context.Context, id uuid.UUID) error + Revoke(ctx context.Context, id uuid.UUID) (uuid.UUID, error) + RevokeBySession(ctx context.Context, sessionID uuid.UUID) ([]uuid.UUID, error) + ReapStaleSystemTokens(ctx context.Context) ([]uuid.UUID, error) + PurgeExpired(ctx context.Context, before time.Time) (int64, error) + + // 测试辅助:直接物理删(生产不暴露) + DeleteByID(ctx context.Context, id uuid.UUID) error +} + +// NewPostgresRepository 用 pgxpool 构造 Repository。 +func NewPostgresRepository(pool *pgxpool.Pool) Repository { + return &pgRepo{pool: pool, q: mcpsqlc.New(pool)} +} + +type pgRepo struct { + pool *pgxpool.Pool + q *mcpsqlc.Queries +} + +// ===== 类型转换 helper ===== + +func toPgUUID(u uuid.UUID) pgtype.UUID { + return pgtype.UUID{Bytes: u, Valid: true} +} +func toPgUUIDPtr(u *uuid.UUID) pgtype.UUID { + if u == nil { + return 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 fromPgUUIDPtr(p pgtype.UUID) *uuid.UUID { + if !p.Valid { + return nil + } + id := uuid.UUID(p.Bytes) + return &id +} +func toPgTimestamp(t *time.Time) pgtype.Timestamptz { + if t == nil { + return pgtype.Timestamptz{} + } + return pgtype.Timestamptz{Time: *t, Valid: true} +} +func fromPgTimestamp(t pgtype.Timestamptz) *time.Time { + if !t.Valid { + return nil + } + out := t.Time + return &out +} + +// scopeToJSON 把 Scope 序列化为 jsonb 入参。空 Scope 也写 `{}`,与 DB DEFAULT 一致。 +func scopeToJSON(s Scope) ([]byte, error) { + return json.Marshal(s) +} + +// scopeFromJSON 反序列化 DB 中 jsonb 字段。空对象 `{}` → 零值 Scope(Tools/ProjectIDs 都为 nil)。 +func scopeFromJSON(raw []byte) (Scope, error) { + if len(raw) == 0 { + return Scope{}, nil + } + var s Scope + if err := json.Unmarshal(raw, &s); err != nil { + return Scope{}, err + } + return s, nil +} + +// rowToMCPToken 把 sqlc 生成的 row 翻译为业务结构。 +func rowToMCPToken(r mcpsqlc.McpToken) (*MCPToken, error) { + scope, err := scopeFromJSON(r.Scope) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "decode mcp_token.scope") + } + return &MCPToken{ + ID: fromPgUUID(r.ID), + TokenHash: r.TokenHash, + UserID: fromPgUUID(r.UserID), + Name: r.Name, + Issuer: Issuer(r.Issuer), + Scope: scope, + ACPSessionID: fromPgUUIDPtr(r.AcpSessionID), + ExpiresAt: fromPgTimestamp(r.ExpiresAt), + LastUsedAt: fromPgTimestamp(r.LastUsedAt), + RevokedAt: fromPgTimestamp(r.RevokedAt), + CreatedBy: fromPgUUID(r.CreatedBy), + CreatedAt: r.CreatedAt.Time, + }, nil +} + +// pgxNoRowsToErrNotFound 把 pgx.ErrNoRows 翻译为 errs.AppError(NotFound)。 +func pgxNoRowsToErrNotFound(err error, code errs.Code, msg string) error { + if errors.Is(err, pgx.ErrNoRows) { + return errs.New(code, msg) + } + return errs.Wrap(err, errs.CodeInternal, msg) +} + +// ===== Create ===== + +func (r *pgRepo) Create(ctx context.Context, t *MCPToken) (*MCPToken, error) { + scopeJSON, err := scopeToJSON(t.Scope) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "encode mcp_token.scope") + } + row, err := r.q.CreateMcpToken(ctx, mcpsqlc.CreateMcpTokenParams{ + ID: toPgUUID(t.ID), + TokenHash: t.TokenHash, + UserID: toPgUUID(t.UserID), + Name: t.Name, + Issuer: string(t.Issuer), + Scope: scopeJSON, + AcpSessionID: toPgUUIDPtr(t.ACPSessionID), + ExpiresAt: toPgTimestamp(t.ExpiresAt), + CreatedBy: toPgUUID(t.CreatedBy), + }) + if err != nil { + return nil, errs.Wrap(err, errs.CodeInternal, "insert mcp_token") + } + return rowToMCPToken(row) +} + +// ===== Get ===== + +func (r *pgRepo) GetByHash(ctx context.Context, hash []byte) (*MCPToken, error) { + row, err := r.q.GetMcpTokenByHash(ctx, hash) + if err != nil { + return nil, pgxNoRowsToErrNotFound(err, errs.CodeMcpTokenInvalid, "mcp token not found") + } + return rowToMCPToken(row) +} + +func (r *pgRepo) GetByID(ctx context.Context, id uuid.UUID) (*MCPToken, error) { + row, err := r.q.GetMcpTokenByID(ctx, toPgUUID(id)) + if err != nil { + return nil, pgxNoRowsToErrNotFound(err, errs.CodeMcpTokenNotFound, "mcp token not found") + } + return rowToMCPToken(row) +} + +// ===== Stub methods (implemented in subsequent tasks) ===== + +func (r *pgRepo) List(_ context.Context, _ *uuid.UUID, _ bool) ([]*MCPToken, error) { + panic("not implemented: List") +} + +func (r *pgRepo) UpdateLastUsed(_ context.Context, _ uuid.UUID) error { + panic("not implemented: UpdateLastUsed") +} + +func (r *pgRepo) Revoke(_ context.Context, _ uuid.UUID) (uuid.UUID, error) { + panic("not implemented: Revoke") +} + +func (r *pgRepo) RevokeBySession(_ context.Context, _ uuid.UUID) ([]uuid.UUID, error) { + panic("not implemented: RevokeBySession") +} + +func (r *pgRepo) ReapStaleSystemTokens(_ context.Context) ([]uuid.UUID, error) { + panic("not implemented: ReapStaleSystemTokens") +} + +func (r *pgRepo) PurgeExpired(_ context.Context, _ time.Time) (int64, error) { + panic("not implemented: PurgeExpired") +} + +func (r *pgRepo) DeleteByID(_ context.Context, _ uuid.UUID) error { + panic("not implemented: DeleteByID") +}