mirror of
https://github.com/harness/drone.git
synced 2025-05-06 02:40:52 +08:00
467 lines
12 KiB
Go
467 lines
12 KiB
Go
// Copyright 2022 Harness Inc. All rights reserved.
|
|
// Use of this source code is governed by the Polyform Free Trial License
|
|
// that can be found in the LICENSE.md file for this repository.
|
|
|
|
package database
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/harness/gitness/internal/paths"
|
|
"github.com/harness/gitness/internal/store"
|
|
gitness_store "github.com/harness/gitness/store"
|
|
"github.com/harness/gitness/store/database"
|
|
"github.com/harness/gitness/store/database/dbtx"
|
|
"github.com/harness/gitness/types"
|
|
"github.com/harness/gitness/types/enum"
|
|
|
|
"github.com/jmoiron/sqlx"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
var _ store.RepoStore = (*RepoStore)(nil)
|
|
|
|
// NewRepoStore returns a new RepoStore.
|
|
func NewRepoStore(
|
|
db *sqlx.DB,
|
|
spacePathCache store.SpacePathCache,
|
|
spacePathStore store.SpacePathStore,
|
|
) *RepoStore {
|
|
return &RepoStore{
|
|
db: db,
|
|
spacePathCache: spacePathCache,
|
|
spacePathStore: spacePathStore,
|
|
}
|
|
}
|
|
|
|
// RepoStore implements a store.RepoStore backed by a relational database.
|
|
type RepoStore struct {
|
|
db *sqlx.DB
|
|
spacePathCache store.SpacePathCache
|
|
spacePathStore store.SpacePathStore
|
|
}
|
|
|
|
type repository struct {
|
|
// TODO: int64 ID doesn't match DB
|
|
ID int64 `db:"repo_id"`
|
|
Version int64 `db:"repo_version"`
|
|
ParentID int64 `db:"repo_parent_id"`
|
|
UID string `db:"repo_uid"`
|
|
Description string `db:"repo_description"`
|
|
IsPublic bool `db:"repo_is_public"`
|
|
CreatedBy int64 `db:"repo_created_by"`
|
|
Created int64 `db:"repo_created"`
|
|
Updated int64 `db:"repo_updated"`
|
|
|
|
GitUID string `db:"repo_git_uid"`
|
|
DefaultBranch string `db:"repo_default_branch"`
|
|
ForkID int64 `db:"repo_fork_id"`
|
|
PullReqSeq int64 `db:"repo_pullreq_seq"`
|
|
|
|
NumForks int `db:"repo_num_forks"`
|
|
NumPulls int `db:"repo_num_pulls"`
|
|
NumClosedPulls int `db:"repo_num_closed_pulls"`
|
|
NumOpenPulls int `db:"repo_num_open_pulls"`
|
|
NumMergedPulls int `db:"repo_num_merged_pulls"`
|
|
|
|
Importing bool `db:"repo_importing"`
|
|
}
|
|
|
|
const (
|
|
repoColumnsForJoin = `
|
|
repo_id
|
|
,repo_version
|
|
,repo_parent_id
|
|
,repo_uid
|
|
,repo_description
|
|
,repo_is_public
|
|
,repo_created_by
|
|
,repo_created
|
|
,repo_updated
|
|
,repo_git_uid
|
|
,repo_default_branch
|
|
,repo_pullreq_seq
|
|
,repo_fork_id
|
|
,repo_num_forks
|
|
,repo_num_pulls
|
|
,repo_num_closed_pulls
|
|
,repo_num_open_pulls
|
|
,repo_num_merged_pulls
|
|
,repo_importing`
|
|
|
|
repoSelectBase = `
|
|
SELECT` + repoColumnsForJoin + `
|
|
FROM repositories`
|
|
)
|
|
|
|
// Find finds the repo by id.
|
|
func (s *RepoStore) Find(ctx context.Context, id int64) (*types.Repository, error) {
|
|
const sqlQuery = repoSelectBase + `
|
|
WHERE repo_id = $1`
|
|
|
|
db := dbtx.GetAccessor(ctx, s.db)
|
|
|
|
dst := new(repository)
|
|
if err := db.GetContext(ctx, dst, sqlQuery, id); err != nil {
|
|
return nil, database.ProcessSQLErrorf(err, "Failed to find repo")
|
|
}
|
|
|
|
return s.mapToRepo(ctx, dst)
|
|
}
|
|
|
|
// Find finds the repo with the given UID in the given space ID.
|
|
func (s *RepoStore) FindByUID(ctx context.Context, spaceID int64, uid string) (*types.Repository, error) {
|
|
const sqlQuery = repoSelectBase + `
|
|
WHERE repo_parent_id = $1 AND LOWER(repo_uid) = $2`
|
|
|
|
db := dbtx.GetAccessor(ctx, s.db)
|
|
|
|
dst := new(repository)
|
|
if err := db.GetContext(ctx, dst, sqlQuery, spaceID, strings.ToLower(uid)); err != nil {
|
|
return nil, database.ProcessSQLErrorf(err, "Failed to find repo")
|
|
}
|
|
|
|
return s.mapToRepo(ctx, dst)
|
|
}
|
|
|
|
// FindByRef finds the repo using the repoRef as either the id or the repo path.
|
|
func (s *RepoStore) FindByRef(ctx context.Context, repoRef string) (*types.Repository, error) {
|
|
// ASSUMPTION: digits only is not a valid repo path
|
|
id, err := strconv.ParseInt(repoRef, 10, 64)
|
|
if err != nil {
|
|
spacePath, repoUID, err := paths.DisectLeaf(repoRef)
|
|
pathObject, err := s.spacePathCache.Get(ctx, spacePath)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get space path: %w", err)
|
|
}
|
|
|
|
return s.FindByUID(ctx, pathObject.SpaceID, repoUID)
|
|
}
|
|
|
|
return s.Find(ctx, id)
|
|
}
|
|
|
|
// Create creates a new repository.
|
|
func (s *RepoStore) Create(ctx context.Context, repo *types.Repository) error {
|
|
const sqlQuery = `
|
|
INSERT INTO repositories (
|
|
repo_version
|
|
,repo_parent_id
|
|
,repo_uid
|
|
,repo_description
|
|
,repo_is_public
|
|
,repo_created_by
|
|
,repo_created
|
|
,repo_updated
|
|
,repo_git_uid
|
|
,repo_default_branch
|
|
,repo_fork_id
|
|
,repo_pullreq_seq
|
|
,repo_num_forks
|
|
,repo_num_pulls
|
|
,repo_num_closed_pulls
|
|
,repo_num_open_pulls
|
|
,repo_num_merged_pulls
|
|
,repo_importing
|
|
) values (
|
|
:repo_version
|
|
,:repo_parent_id
|
|
,:repo_uid
|
|
,:repo_description
|
|
,:repo_is_public
|
|
,:repo_created_by
|
|
,:repo_created
|
|
,:repo_updated
|
|
,:repo_git_uid
|
|
,:repo_default_branch
|
|
,:repo_fork_id
|
|
,:repo_pullreq_seq
|
|
,:repo_num_forks
|
|
,:repo_num_pulls
|
|
,:repo_num_closed_pulls
|
|
,:repo_num_open_pulls
|
|
,:repo_num_merged_pulls
|
|
,:repo_importing
|
|
) RETURNING repo_id`
|
|
|
|
db := dbtx.GetAccessor(ctx, s.db)
|
|
|
|
// insert repo first so we get id
|
|
query, arg, err := db.BindNamed(sqlQuery, mapToInternalRepo(repo))
|
|
if err != nil {
|
|
return database.ProcessSQLErrorf(err, "Failed to bind repo object")
|
|
}
|
|
|
|
if err = db.QueryRowContext(ctx, query, arg...).Scan(&repo.ID); err != nil {
|
|
return database.ProcessSQLErrorf(err, "Insert query failed")
|
|
}
|
|
|
|
repo.Path, err = s.getRepoPath(ctx, repo.ParentID, repo.UID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Update updates the repo details.
|
|
func (s *RepoStore) Update(ctx context.Context, repo *types.Repository) error {
|
|
const sqlQuery = `
|
|
UPDATE repositories
|
|
SET
|
|
repo_version = :repo_version
|
|
,repo_updated = :repo_updated
|
|
,repo_parent_id = :repo_parent_id
|
|
,repo_uid = :repo_uid
|
|
,repo_git_uid = :repo_git_uid
|
|
,repo_description = :repo_description
|
|
,repo_is_public = :repo_is_public
|
|
,repo_default_branch = :repo_default_branch
|
|
,repo_pullreq_seq = :repo_pullreq_seq
|
|
,repo_num_forks = :repo_num_forks
|
|
,repo_num_pulls = :repo_num_pulls
|
|
,repo_num_closed_pulls = :repo_num_closed_pulls
|
|
,repo_num_open_pulls = :repo_num_open_pulls
|
|
,repo_num_merged_pulls = :repo_num_merged_pulls
|
|
,repo_importing = :repo_importing
|
|
WHERE repo_id = :repo_id AND repo_version = :repo_version - 1`
|
|
|
|
dbRepo := mapToInternalRepo(repo)
|
|
|
|
// update Version (used for optimistic locking) and Updated time
|
|
dbRepo.Version++
|
|
dbRepo.Updated = time.Now().UnixMilli()
|
|
|
|
db := dbtx.GetAccessor(ctx, s.db)
|
|
|
|
query, arg, err := db.BindNamed(sqlQuery, dbRepo)
|
|
if err != nil {
|
|
return database.ProcessSQLErrorf(err, "Failed to bind repo object")
|
|
}
|
|
|
|
result, err := db.ExecContext(ctx, query, arg...)
|
|
if err != nil {
|
|
return database.ProcessSQLErrorf(err, "Failed to update repository")
|
|
}
|
|
|
|
count, err := result.RowsAffected()
|
|
if err != nil {
|
|
return database.ProcessSQLErrorf(err, "Failed to get number of updated rows")
|
|
}
|
|
|
|
if count == 0 {
|
|
return gitness_store.ErrVersionConflict
|
|
}
|
|
|
|
repo.Version = dbRepo.Version
|
|
repo.Updated = dbRepo.Updated
|
|
|
|
// update path in case parent/uid changed (its most likely cached anyway)
|
|
repo.Path, err = s.getRepoPath(ctx, repo.ParentID, repo.UID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// UpdateOptLock updates the repository using the optimistic locking mechanism.
|
|
func (s *RepoStore) UpdateOptLock(ctx context.Context,
|
|
repo *types.Repository,
|
|
mutateFn func(repository *types.Repository) error,
|
|
) (*types.Repository, error) {
|
|
for {
|
|
dup := *repo
|
|
|
|
err := mutateFn(&dup)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = s.Update(ctx, &dup)
|
|
if err == nil {
|
|
return &dup, nil
|
|
}
|
|
if !errors.Is(err, gitness_store.ErrVersionConflict) {
|
|
return nil, err
|
|
}
|
|
|
|
repo, err = s.Find(ctx, repo.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
|
|
// Delete the repository.
|
|
func (s *RepoStore) Delete(ctx context.Context, id int64) error {
|
|
const repoDelete = `
|
|
DELETE FROM repositories
|
|
WHERE repo_id = $1`
|
|
|
|
db := dbtx.GetAccessor(ctx, s.db)
|
|
|
|
if _, err := db.ExecContext(ctx, repoDelete, id); err != nil {
|
|
return database.ProcessSQLErrorf(err, "the delete query failed")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Count of repos in a space. if parentID (space) is zero then it will count all repositories in the system.
|
|
func (s *RepoStore) Count(ctx context.Context, parentID int64, opts *types.RepoFilter) (int64, error) {
|
|
stmt := database.Builder.
|
|
Select("count(*)").
|
|
From("repositories")
|
|
|
|
if parentID > 0 {
|
|
stmt = stmt.Where("repo_parent_id = ?", parentID)
|
|
}
|
|
|
|
if opts.Query != "" {
|
|
stmt = stmt.Where("LOWER(repo_uid) LIKE ?", fmt.Sprintf("%%%s%%", strings.ToLower(opts.Query)))
|
|
}
|
|
|
|
sql, args, err := stmt.ToSql()
|
|
if err != nil {
|
|
return 0, errors.Wrap(err, "Failed to convert query to sql")
|
|
}
|
|
|
|
db := dbtx.GetAccessor(ctx, s.db)
|
|
|
|
var count int64
|
|
err = db.QueryRowContext(ctx, sql, args...).Scan(&count)
|
|
if err != nil {
|
|
return 0, database.ProcessSQLErrorf(err, "Failed executing count query")
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
// List returns a list of repos in a space.
|
|
func (s *RepoStore) List(ctx context.Context, parentID int64, opts *types.RepoFilter) ([]*types.Repository, error) {
|
|
stmt := database.Builder.
|
|
Select(repoColumnsForJoin).
|
|
From("repositories").
|
|
Where("repo_parent_id = ?", fmt.Sprint(parentID))
|
|
|
|
if opts.Query != "" {
|
|
stmt = stmt.Where("LOWER(repo_uid) LIKE ?", fmt.Sprintf("%%%s%%", strings.ToLower(opts.Query)))
|
|
}
|
|
|
|
stmt = stmt.Limit(database.Limit(opts.Size))
|
|
stmt = stmt.Offset(database.Offset(opts.Page, opts.Size))
|
|
|
|
switch opts.Sort {
|
|
case enum.RepoAttrUID, enum.RepoAttrNone:
|
|
// NOTE: string concatenation is safe because the
|
|
// order attribute is an enum and is not user-defined,
|
|
// and is therefore not subject to injection attacks.
|
|
stmt = stmt.OrderBy("repo_importing desc, repo_uid " + opts.Order.String())
|
|
case enum.RepoAttrCreated:
|
|
stmt = stmt.OrderBy("repo_created " + opts.Order.String())
|
|
case enum.RepoAttrUpdated:
|
|
stmt = stmt.OrderBy("repo_updated " + opts.Order.String())
|
|
}
|
|
|
|
sql, args, err := stmt.ToSql()
|
|
if err != nil {
|
|
return nil, errors.Wrap(err, "Failed to convert query to sql")
|
|
}
|
|
|
|
db := dbtx.GetAccessor(ctx, s.db)
|
|
|
|
dst := []*repository{}
|
|
if err = db.SelectContext(ctx, &dst, sql, args...); err != nil {
|
|
return nil, database.ProcessSQLErrorf(err, "Failed executing custom list query")
|
|
}
|
|
|
|
return s.mapToRepos(ctx, dst)
|
|
}
|
|
|
|
func (s *RepoStore) mapToRepo(
|
|
ctx context.Context,
|
|
in *repository,
|
|
) (*types.Repository, error) {
|
|
var err error
|
|
res := &types.Repository{
|
|
ID: in.ID,
|
|
Version: in.Version,
|
|
ParentID: in.ParentID,
|
|
UID: in.UID,
|
|
Description: in.Description,
|
|
IsPublic: in.IsPublic,
|
|
Created: in.Created,
|
|
CreatedBy: in.CreatedBy,
|
|
Updated: in.Updated,
|
|
GitUID: in.GitUID,
|
|
DefaultBranch: in.DefaultBranch,
|
|
ForkID: in.ForkID,
|
|
PullReqSeq: in.PullReqSeq,
|
|
NumForks: in.NumForks,
|
|
NumPulls: in.NumPulls,
|
|
NumClosedPulls: in.NumClosedPulls,
|
|
NumOpenPulls: in.NumOpenPulls,
|
|
NumMergedPulls: in.NumMergedPulls,
|
|
Importing: in.Importing,
|
|
// Path: is set below
|
|
}
|
|
|
|
res.Path, err = s.getRepoPath(ctx, in.ParentID, in.UID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return res, nil
|
|
}
|
|
|
|
func (s *RepoStore) getRepoPath(ctx context.Context, parentID int64, repoUID string) (string, error) {
|
|
spacePath, err := s.spacePathStore.FindPrimaryBySpaceID(ctx, parentID)
|
|
if err != nil {
|
|
return "", fmt.Errorf("failed to get primary path for space %d: %w", parentID, err)
|
|
}
|
|
return paths.Concatinate(spacePath.Value, repoUID), nil
|
|
}
|
|
|
|
func (s *RepoStore) mapToRepos(
|
|
ctx context.Context,
|
|
repos []*repository,
|
|
) ([]*types.Repository, error) {
|
|
var err error
|
|
res := make([]*types.Repository, len(repos))
|
|
for i := range repos {
|
|
res[i], err = s.mapToRepo(ctx, repos[i])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
func mapToInternalRepo(in *types.Repository) *repository {
|
|
return &repository{
|
|
ID: in.ID,
|
|
Version: in.Version,
|
|
ParentID: in.ParentID,
|
|
UID: in.UID,
|
|
Description: in.Description,
|
|
IsPublic: in.IsPublic,
|
|
Created: in.Created,
|
|
CreatedBy: in.CreatedBy,
|
|
Updated: in.Updated,
|
|
GitUID: in.GitUID,
|
|
DefaultBranch: in.DefaultBranch,
|
|
ForkID: in.ForkID,
|
|
PullReqSeq: in.PullReqSeq,
|
|
NumForks: in.NumForks,
|
|
NumPulls: in.NumPulls,
|
|
NumClosedPulls: in.NumClosedPulls,
|
|
NumOpenPulls: in.NumOpenPulls,
|
|
NumMergedPulls: in.NumMergedPulls,
|
|
Importing: in.Importing,
|
|
}
|
|
}
|