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

package run

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"
	runsqlc "github.com/yan1h/agent-coding-workflow/internal/run/sqlc"
)

// Repository 是 run 模块对 PG 的全部依赖。Service / Supervisor 单测通过手写
// fakeRepo 满足。
type Repository interface {
	CreateRunProfile(ctx context.Context, p *RunProfile) (*RunProfile, error)
	GetRunProfileByID(ctx context.Context, id uuid.UUID) (*RunProfile, error)
	ListRunProfilesByWorkspace(ctx context.Context, wsID uuid.UUID) ([]*RunProfile, error)
	UpdateRunProfile(ctx context.Context, p *RunProfile) (*RunProfile, error)
	DeleteRunProfile(ctx context.Context, id uuid.UUID) error
	MarkRunProfileStarted(ctx context.Context, id uuid.UUID) (*RunProfile, error)
	UpdateRunProfileExit(ctx context.Context, id uuid.UUID, exitCode *int32, lastErr string) error
}

// IsUniqueViolation reports whether err is a PG unique constraint violation
// （service 层据此把 (workspace_id, slug) 冲突翻成 409）。
func IsUniqueViolation(err error) bool {
	var pg *pgconn.PgError
	return errors.As(err, &pg) && pg.Code == pgerrcode.UniqueViolation
}

type pgRepo struct {
	q *runsqlc.Queries
}

// NewPostgresRepository 用 pgxpool 构造 Repository。
func NewPostgresRepository(pool *pgxpool.Pool) Repository {
	return &pgRepo{q: runsqlc.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 fromPgTimePtr(t pgtype.Timestamptz) *time.Time {
	if !t.Valid {
		return nil
	}
	v := t.Time
	return &v
}

// normalizeStrSlice 把 nil 归一为非 nil 空切片，避免 pgx 把 nil []string 写成
// SQL NULL 违反 TEXT[] NOT NULL 约束（同 acp/repository.go 的同名 helper）。
func normalizeStrSlice(s []string) []string {
	if s == nil {
		return []string{}
	}
	return s
}

func rowToRunProfile(row runsqlc.WorkspaceRunProfile) *RunProfile {
	return &RunProfile{
		ID:            fromPgUUID(row.ID),
		WorkspaceID:   fromPgUUID(row.WorkspaceID),
		Slug:          row.Slug,
		Name:          row.Name,
		Description:   row.Description,
		Command:       row.Command,
		Args:          row.Args,
		EncryptedEnv:  row.EncryptedEnv,
		Enabled:       row.Enabled,
		LastStartedAt: fromPgTimePtr(row.LastStartedAt),
		LastExitCode:  row.LastExitCode,
		LastError:     row.LastError,
		CreatedBy:     fromPgUUID(row.CreatedBy),
		CreatedAt:     row.CreatedAt.Time,
		UpdatedAt:     row.UpdatedAt.Time,
	}
}

func (r *pgRepo) CreateRunProfile(ctx context.Context, p *RunProfile) (*RunProfile, error) {
	row, err := r.q.CreateRunProfile(ctx, runsqlc.CreateRunProfileParams{
		ID:           toPgUUID(p.ID),
		WorkspaceID:  toPgUUID(p.WorkspaceID),
		Slug:         p.Slug,
		Name:         p.Name,
		Description:  p.Description,
		Command:      p.Command,
		Args:         normalizeStrSlice(p.Args),
		EncryptedEnv: p.EncryptedEnv,
		Enabled:      p.Enabled,
		CreatedBy:    toPgUUID(p.CreatedBy),
	})
	if err != nil {
		if IsUniqueViolation(err) {
			return nil, errs.New(errs.CodeRunProfileSlugTaken, "run profile slug already taken in this workspace")
		}
		return nil, errs.Wrap(err, errs.CodeInternal, "create run profile")
	}
	return rowToRunProfile(row), nil
}

func (r *pgRepo) GetRunProfileByID(ctx context.Context, id uuid.UUID) (*RunProfile, error) {
	row, err := r.q.GetRunProfileByID(ctx, toPgUUID(id))
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, errs.New(errs.CodeRunProfileNotFound, "run profile not found")
		}
		return nil, errs.Wrap(err, errs.CodeInternal, "get run profile")
	}
	return rowToRunProfile(row), nil
}

func (r *pgRepo) ListRunProfilesByWorkspace(ctx context.Context, wsID uuid.UUID) ([]*RunProfile, error) {
	rows, err := r.q.ListRunProfilesByWorkspace(ctx, toPgUUID(wsID))
	if err != nil {
		return nil, errs.Wrap(err, errs.CodeInternal, "list run profiles")
	}
	out := make([]*RunProfile, 0, len(rows))
	for _, row := range rows {
		out = append(out, rowToRunProfile(row))
	}
	return out, nil
}

func (r *pgRepo) UpdateRunProfile(ctx context.Context, p *RunProfile) (*RunProfile, error) {
	row, err := r.q.UpdateRunProfile(ctx, runsqlc.UpdateRunProfileParams{
		ID:           toPgUUID(p.ID),
		Slug:         p.Slug,
		Name:         p.Name,
		Description:  p.Description,
		Command:      p.Command,
		Args:         normalizeStrSlice(p.Args),
		EncryptedEnv: p.EncryptedEnv,
		Enabled:      p.Enabled,
	})
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, errs.New(errs.CodeRunProfileNotFound, "run profile not found")
		}
		if IsUniqueViolation(err) {
			return nil, errs.New(errs.CodeRunProfileSlugTaken, "run profile slug already taken in this workspace")
		}
		return nil, errs.Wrap(err, errs.CodeInternal, "update run profile")
	}
	return rowToRunProfile(row), nil
}

func (r *pgRepo) DeleteRunProfile(ctx context.Context, id uuid.UUID) error {
	if err := r.q.DeleteRunProfile(ctx, toPgUUID(id)); err != nil {
		return errs.Wrap(err, errs.CodeInternal, "delete run profile")
	}
	return nil
}

func (r *pgRepo) MarkRunProfileStarted(ctx context.Context, id uuid.UUID) (*RunProfile, error) {
	row, err := r.q.MarkRunProfileStarted(ctx, toPgUUID(id))
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, errs.New(errs.CodeRunProfileNotFound, "run profile not found")
		}
		return nil, errs.Wrap(err, errs.CodeInternal, "mark run profile started")
	}
	return rowToRunProfile(row), nil
}

func (r *pgRepo) UpdateRunProfileExit(ctx context.Context, id uuid.UUID, exitCode *int32, lastErr string) error {
	if err := r.q.UpdateRunProfileExit(ctx, runsqlc.UpdateRunProfileExitParams{
		ID:           toPgUUID(id),
		LastExitCode: exitCode,
		LastError:    lastErr,
	}); err != nil {
		return errs.Wrap(err, errs.CodeInternal, "update run profile exit")
	}
	return nil
}