toki
c6102d33d0
- Update control plane phase roadmap - Move repo-registry milestone to archive - Implement git engine command execution - Add postgres repository storage layer - Update contracts documentation - Add test cases for storage and git engine
626 lines
18 KiB
Go
626 lines
18 KiB
Go
package storage
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"git.toki-labs.com/toki/gito/services/core/internal/core"
|
|
)
|
|
|
|
// PgStore is a pgxpool-backed Store implementation.
|
|
type PgStore struct {
|
|
pool *pgxpool.Pool
|
|
repos *pgRepoStore
|
|
leases *pgWorkspaceLeaseStore
|
|
watches *pgBranchWatchStore
|
|
cursors *pgRevisionCursorStore
|
|
deliveries *pgProviderDeliveryStore
|
|
}
|
|
|
|
// NewPgStore opens a pgxpool connection and optionally applies a migration.
|
|
// migrationSQL is the raw SQL content (e.g. from //go:embed); pass an empty
|
|
// string to skip migration.
|
|
func NewPgStore(ctx context.Context, dsn string, migrationSQL string) (*PgStore, error) {
|
|
pool, err := pgxpool.New(ctx, dsn)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("pgstore: open pool: %w", err)
|
|
}
|
|
if err := pool.Ping(ctx); err != nil {
|
|
pool.Close()
|
|
return nil, fmt.Errorf("pgstore: ping: %w", err)
|
|
}
|
|
if migrationSQL != "" {
|
|
if err := applyMigrationSQL(ctx, pool, migrationSQL); err != nil {
|
|
pool.Close()
|
|
return nil, fmt.Errorf("pgstore: migration: %w", err)
|
|
}
|
|
}
|
|
s := &PgStore{pool: pool}
|
|
s.repos = &pgRepoStore{pool: pool}
|
|
s.leases = &pgWorkspaceLeaseStore{pool: pool}
|
|
s.watches = &pgBranchWatchStore{pool: pool}
|
|
s.cursors = &pgRevisionCursorStore{pool: pool}
|
|
s.deliveries = &pgProviderDeliveryStore{pool: pool}
|
|
return s, nil
|
|
}
|
|
|
|
func (s *PgStore) Close() { s.pool.Close() }
|
|
|
|
func (s *PgStore) Ping(ctx context.Context) error { return s.pool.Ping(ctx) }
|
|
|
|
func (s *PgStore) Repos() RepoStore { return s.repos }
|
|
func (s *PgStore) WorkspaceLeases() WorkspaceLeaseStore { return s.leases }
|
|
func (s *PgStore) Operations() OperationStore { return nil }
|
|
func (s *PgStore) OperationEvents() OperationEventStore { return nil }
|
|
func (s *PgStore) BranchWatches() BranchWatchStore { return s.watches }
|
|
func (s *PgStore) RevisionCursors() RevisionCursorStore { return s.cursors }
|
|
func (s *PgStore) ProviderDeliveries() ProviderDeliveryStore { return s.deliveries }
|
|
|
|
// applyMigrationSQL executes the Up block extracted from goose-style SQL content.
|
|
func applyMigrationSQL(ctx context.Context, pool *pgxpool.Pool, content string) error {
|
|
sql := extractUpBlock(content)
|
|
if sql == "" {
|
|
return nil
|
|
}
|
|
_, err := pool.Exec(ctx, sql)
|
|
return err
|
|
}
|
|
|
|
func extractUpBlock(content string) string {
|
|
const beginMarker = "-- +goose StatementBegin"
|
|
const endMarker = "-- +goose StatementEnd"
|
|
const upMarker = "-- +goose Up"
|
|
|
|
upIdx := strings.Index(content, upMarker)
|
|
if upIdx < 0 {
|
|
return ""
|
|
}
|
|
after := content[upIdx:]
|
|
beginIdx := strings.Index(after, beginMarker)
|
|
endIdx := strings.Index(after, endMarker)
|
|
if beginIdx < 0 || endIdx < 0 || endIdx <= beginIdx {
|
|
return ""
|
|
}
|
|
return strings.TrimSpace(after[beginIdx+len(beginMarker) : endIdx])
|
|
}
|
|
|
|
// pgRepoStore
|
|
|
|
type pgRepoStore struct{ pool *pgxpool.Pool }
|
|
|
|
func (s *pgRepoStore) CreateRepo(ctx context.Context, repo core.Repo) error {
|
|
repo, err := normalizeRepo(repo)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
const q = `
|
|
INSERT INTO repos (id, name, remote_url, default_branch, workspace_root, credential_ref)
|
|
VALUES ($1, $2, $3, $4, $5, $6)
|
|
`
|
|
_, err = s.pool.Exec(ctx, q,
|
|
repo.ID,
|
|
repo.Name,
|
|
repo.RemoteURL,
|
|
repo.DefaultBranch,
|
|
repo.WorkspaceRoot,
|
|
nullableString(repo.CredentialRef),
|
|
)
|
|
if err != nil {
|
|
if isUniqueViolation(err) {
|
|
return fmt.Errorf("%w: %s", ErrRepoAlreadyExists, repo.ID)
|
|
}
|
|
return fmt.Errorf("create repo: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *pgRepoStore) GetRepo(ctx context.Context, id string) (core.Repo, error) {
|
|
id = strings.TrimSpace(id)
|
|
if id == "" {
|
|
return core.Repo{}, fmt.Errorf("%w: id is required", ErrInvalidRepo)
|
|
}
|
|
|
|
const q = `
|
|
SELECT id, name, remote_url, default_branch, workspace_root, credential_ref
|
|
FROM repos
|
|
WHERE id=$1
|
|
`
|
|
repo, err := scanRepo(s.pool.QueryRow(ctx, q, id))
|
|
if err != nil {
|
|
if isNoRows(err) {
|
|
return core.Repo{}, fmt.Errorf("%w: %s", ErrRepoNotFound, id)
|
|
}
|
|
return core.Repo{}, fmt.Errorf("get repo: %w", err)
|
|
}
|
|
return repo, nil
|
|
}
|
|
|
|
func (s *pgRepoStore) ListRepos(ctx context.Context) ([]core.Repo, error) {
|
|
const q = `
|
|
SELECT id, name, remote_url, default_branch, workspace_root, credential_ref
|
|
FROM repos
|
|
ORDER BY id
|
|
`
|
|
rows, err := s.pool.Query(ctx, q)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list repos: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
var out []core.Repo
|
|
for rows.Next() {
|
|
repo, err := scanRepo(rows)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("scan repo: %w", err)
|
|
}
|
|
out = append(out, repo)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *pgRepoStore) UpdateRepo(ctx context.Context, repo core.Repo) error {
|
|
repo, err := normalizeRepo(repo)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
const q = `
|
|
UPDATE repos
|
|
SET name=$2,
|
|
remote_url=$3,
|
|
default_branch=$4,
|
|
workspace_root=$5,
|
|
credential_ref=$6,
|
|
updated_at=now()
|
|
WHERE id=$1
|
|
`
|
|
tag, err := s.pool.Exec(ctx, q,
|
|
repo.ID,
|
|
repo.Name,
|
|
repo.RemoteURL,
|
|
repo.DefaultBranch,
|
|
repo.WorkspaceRoot,
|
|
nullableString(repo.CredentialRef),
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("update repo: %w", err)
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return fmt.Errorf("%w: %s", ErrRepoNotFound, repo.ID)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func normalizeRepo(repo core.Repo) (core.Repo, error) {
|
|
repo.ID = strings.TrimSpace(repo.ID)
|
|
repo.Name = strings.TrimSpace(repo.Name)
|
|
repo.RemoteURL = strings.TrimSpace(repo.RemoteURL)
|
|
repo.DefaultBranch = strings.TrimSpace(repo.DefaultBranch)
|
|
repo.WorkspaceRoot = strings.TrimSpace(repo.WorkspaceRoot)
|
|
repo.CredentialRef = strings.TrimSpace(repo.CredentialRef)
|
|
|
|
var missing []string
|
|
if repo.ID == "" {
|
|
missing = append(missing, "id")
|
|
}
|
|
if repo.Name == "" {
|
|
missing = append(missing, "name")
|
|
}
|
|
if repo.RemoteURL == "" {
|
|
missing = append(missing, "remote_url")
|
|
}
|
|
if repo.DefaultBranch == "" {
|
|
missing = append(missing, "default_branch")
|
|
}
|
|
if repo.WorkspaceRoot == "" {
|
|
missing = append(missing, "workspace_root")
|
|
}
|
|
if len(missing) > 0 {
|
|
return core.Repo{}, fmt.Errorf("%w: required fields missing: %s", ErrInvalidRepo, strings.Join(missing, ", "))
|
|
}
|
|
return repo, nil
|
|
}
|
|
|
|
type repoScanner interface {
|
|
Scan(dest ...any) error
|
|
}
|
|
|
|
func scanRepo(scanner repoScanner) (core.Repo, error) {
|
|
var repo core.Repo
|
|
var credentialRef sql.NullString
|
|
if err := scanner.Scan(
|
|
&repo.ID,
|
|
&repo.Name,
|
|
&repo.RemoteURL,
|
|
&repo.DefaultBranch,
|
|
&repo.WorkspaceRoot,
|
|
&credentialRef,
|
|
); err != nil {
|
|
return core.Repo{}, err
|
|
}
|
|
if credentialRef.Valid {
|
|
repo.CredentialRef = credentialRef.String
|
|
}
|
|
return repo, nil
|
|
}
|
|
|
|
func nullableString(value string) any {
|
|
if value == "" {
|
|
return nil
|
|
}
|
|
return value
|
|
}
|
|
|
|
// pgWorkspaceLeaseStore
|
|
|
|
type pgWorkspaceLeaseStore struct{ pool *pgxpool.Pool }
|
|
|
|
func (s *pgWorkspaceLeaseStore) SaveLease(ctx context.Context, lease core.WorkspaceLease) error {
|
|
lease, err := normalizeLease(lease)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
const q = `
|
|
INSERT INTO workspace_leases (id, repo_id, slot, path, state, expires_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6)
|
|
`
|
|
_, err = s.pool.Exec(ctx, q,
|
|
lease.ID,
|
|
lease.RepoID,
|
|
lease.Slot,
|
|
lease.Path,
|
|
lease.State,
|
|
nullableTime(lease.ExpiresAt),
|
|
)
|
|
if err != nil {
|
|
if isUniqueViolation(err) {
|
|
return fmt.Errorf("%w: %s", ErrLeaseAlreadyExists, lease.ID)
|
|
}
|
|
return fmt.Errorf("save lease: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *pgWorkspaceLeaseStore) GetLease(ctx context.Context, id string) (core.WorkspaceLease, error) {
|
|
id = strings.TrimSpace(id)
|
|
if id == "" {
|
|
return core.WorkspaceLease{}, fmt.Errorf("%w: id is required", ErrInvalidLease)
|
|
}
|
|
|
|
const q = `
|
|
SELECT id, repo_id, slot, path, state, expires_at
|
|
FROM workspace_leases
|
|
WHERE id=$1
|
|
`
|
|
lease, err := scanLease(s.pool.QueryRow(ctx, q, id))
|
|
if err != nil {
|
|
if isNoRows(err) {
|
|
return core.WorkspaceLease{}, fmt.Errorf("%w: %s", ErrLeaseNotFound, id)
|
|
}
|
|
return core.WorkspaceLease{}, fmt.Errorf("get lease: %w", err)
|
|
}
|
|
return lease, nil
|
|
}
|
|
|
|
func (s *pgWorkspaceLeaseStore) ListLeases(ctx context.Context, repoID string) ([]core.WorkspaceLease, error) {
|
|
repoID = strings.TrimSpace(repoID)
|
|
if repoID == "" {
|
|
return nil, fmt.Errorf("%w: repo_id is required", ErrInvalidLease)
|
|
}
|
|
|
|
const q = `
|
|
SELECT id, repo_id, slot, path, state, expires_at
|
|
FROM workspace_leases
|
|
WHERE repo_id=$1
|
|
ORDER BY slot
|
|
`
|
|
rows, err := s.pool.Query(ctx, q, repoID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list leases: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
var out []core.WorkspaceLease
|
|
for rows.Next() {
|
|
lease, err := scanLease(rows)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("scan lease: %w", err)
|
|
}
|
|
out = append(out, lease)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *pgWorkspaceLeaseStore) AcquireAvailableLease(ctx context.Context, repoID string, now time.Time, ttl time.Duration) (core.WorkspaceLease, bool, error) {
|
|
repoID = strings.TrimSpace(repoID)
|
|
if repoID == "" {
|
|
return core.WorkspaceLease{}, false, fmt.Errorf("%w: repo_id is required", ErrInvalidLease)
|
|
}
|
|
if ttl <= 0 {
|
|
return core.WorkspaceLease{}, false, fmt.Errorf("%w: ttl must be positive", ErrInvalidLease)
|
|
}
|
|
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return core.WorkspaceLease{}, false, fmt.Errorf("begin lease acquire: %w", err)
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
expiresAt := now.UTC().Add(ttl)
|
|
const q = `
|
|
WITH candidate AS (
|
|
SELECT id
|
|
FROM workspace_leases
|
|
WHERE repo_id=$1 AND state=$2
|
|
ORDER BY slot
|
|
FOR UPDATE SKIP LOCKED
|
|
LIMIT 1
|
|
)
|
|
UPDATE workspace_leases AS lease
|
|
SET state=$3,
|
|
expires_at=$4,
|
|
updated_at=now()
|
|
FROM candidate
|
|
WHERE lease.id=candidate.id
|
|
RETURNING lease.id, lease.repo_id, lease.slot, lease.path, lease.state, lease.expires_at
|
|
`
|
|
lease, err := scanLease(tx.QueryRow(ctx, q, repoID, core.WorkspaceAvailable, core.WorkspaceLeased, expiresAt))
|
|
if err != nil {
|
|
if isNoRows(err) {
|
|
return core.WorkspaceLease{}, false, nil
|
|
}
|
|
return core.WorkspaceLease{}, false, fmt.Errorf("acquire lease: %w", err)
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return core.WorkspaceLease{}, false, fmt.Errorf("commit lease acquire: %w", err)
|
|
}
|
|
return lease, true, nil
|
|
}
|
|
|
|
func (s *pgWorkspaceLeaseStore) ReleaseLease(ctx context.Context, id string, release LeaseRelease) (core.WorkspaceLease, error) {
|
|
id = strings.TrimSpace(id)
|
|
if id == "" {
|
|
return core.WorkspaceLease{}, fmt.Errorf("%w: id is required", ErrInvalidLease)
|
|
}
|
|
nextState, err := leaseStateForRelease(release)
|
|
if err != nil {
|
|
return core.WorkspaceLease{}, err
|
|
}
|
|
|
|
const q = `
|
|
UPDATE workspace_leases
|
|
SET state=$2,
|
|
expires_at=NULL,
|
|
updated_at=now()
|
|
WHERE id=$1 AND state=$3
|
|
RETURNING id, repo_id, slot, path, state, expires_at
|
|
`
|
|
lease, err := scanLease(s.pool.QueryRow(ctx, q, id, nextState, core.WorkspaceLeased))
|
|
if err != nil {
|
|
if isNoRows(err) {
|
|
existing, getErr := s.GetLease(ctx, id)
|
|
if getErr != nil {
|
|
return core.WorkspaceLease{}, getErr
|
|
}
|
|
return core.WorkspaceLease{}, fmt.Errorf("%w: lease %s is %s, not %s", ErrInvalidLease, existing.ID, existing.State, core.WorkspaceLeased)
|
|
}
|
|
return core.WorkspaceLease{}, fmt.Errorf("release lease: %w", err)
|
|
}
|
|
return lease, nil
|
|
}
|
|
|
|
func normalizeLease(lease core.WorkspaceLease) (core.WorkspaceLease, error) {
|
|
lease.ID = strings.TrimSpace(lease.ID)
|
|
lease.RepoID = strings.TrimSpace(lease.RepoID)
|
|
lease.Slot = strings.TrimSpace(lease.Slot)
|
|
lease.Path = strings.TrimSpace(lease.Path)
|
|
|
|
var missing []string
|
|
if lease.ID == "" {
|
|
missing = append(missing, "id")
|
|
}
|
|
if lease.RepoID == "" {
|
|
missing = append(missing, "repo_id")
|
|
}
|
|
if lease.Slot == "" {
|
|
missing = append(missing, "slot")
|
|
}
|
|
if lease.Path == "" {
|
|
missing = append(missing, "path")
|
|
}
|
|
if len(missing) > 0 {
|
|
return core.WorkspaceLease{}, fmt.Errorf("%w: required fields missing: %s", ErrInvalidLease, strings.Join(missing, ", "))
|
|
}
|
|
if !isValidWorkspaceState(lease.State) {
|
|
return core.WorkspaceLease{}, fmt.Errorf("%w: unsupported state %q", ErrInvalidLease, lease.State)
|
|
}
|
|
return lease, nil
|
|
}
|
|
|
|
func isValidWorkspaceState(state core.WorkspaceState) bool {
|
|
switch state {
|
|
case core.WorkspaceAvailable, core.WorkspaceLeased, core.WorkspaceDirty, core.WorkspaceError:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func leaseStateForRelease(release LeaseRelease) (core.WorkspaceState, error) {
|
|
if release.Dirty {
|
|
return core.WorkspaceDirty, nil
|
|
}
|
|
switch release.OperationState {
|
|
case core.OperationSucceeded, core.OperationCancelled:
|
|
return core.WorkspaceAvailable, nil
|
|
case core.OperationFailed:
|
|
return core.WorkspaceError, nil
|
|
default:
|
|
return "", fmt.Errorf("%w: unsupported release operation state %q", ErrInvalidLease, release.OperationState)
|
|
}
|
|
}
|
|
|
|
func scanLease(scanner repoScanner) (core.WorkspaceLease, error) {
|
|
var lease core.WorkspaceLease
|
|
var expiresAt sql.NullTime
|
|
if err := scanner.Scan(
|
|
&lease.ID,
|
|
&lease.RepoID,
|
|
&lease.Slot,
|
|
&lease.Path,
|
|
&lease.State,
|
|
&expiresAt,
|
|
); err != nil {
|
|
return core.WorkspaceLease{}, err
|
|
}
|
|
if expiresAt.Valid {
|
|
value := expiresAt.Time
|
|
lease.ExpiresAt = &value
|
|
}
|
|
return lease, nil
|
|
}
|
|
|
|
func nullableTime(value *time.Time) any {
|
|
if value == nil {
|
|
return nil
|
|
}
|
|
return value.UTC()
|
|
}
|
|
|
|
// pgBranchWatchStore
|
|
|
|
type pgBranchWatchStore struct{ pool *pgxpool.Pool }
|
|
|
|
func (s *pgBranchWatchStore) UpsertBranchWatch(ctx context.Context, watch core.BranchWatch) (core.BranchWatch, error) {
|
|
const q = `
|
|
INSERT INTO branch_watches (id, provider, repo_id, branch, created_at)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
ON CONFLICT (provider, repo_id, branch) DO UPDATE
|
|
SET id = branch_watches.id, created_at = branch_watches.created_at
|
|
RETURNING id, provider, repo_id, branch, created_at
|
|
`
|
|
var out core.BranchWatch
|
|
row := s.pool.QueryRow(ctx, q, watch.ID, watch.Provider, watch.RepoID, watch.Branch, watch.CreatedAt)
|
|
if err := row.Scan(&out.ID, &out.Provider, &out.RepoID, &out.Branch, &out.CreatedAt); err != nil {
|
|
return core.BranchWatch{}, fmt.Errorf("upsert branch watch: %w", err)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (s *pgBranchWatchStore) FindBranchWatch(ctx context.Context, provider, repoID, branch string) (core.BranchWatch, bool, error) {
|
|
const q = `SELECT id, provider, repo_id, branch, created_at FROM branch_watches WHERE provider=$1 AND repo_id=$2 AND branch=$3`
|
|
var out core.BranchWatch
|
|
row := s.pool.QueryRow(ctx, q, provider, repoID, branch)
|
|
if err := row.Scan(&out.ID, &out.Provider, &out.RepoID, &out.Branch, &out.CreatedAt); err != nil {
|
|
if isNoRows(err) {
|
|
return core.BranchWatch{}, false, nil
|
|
}
|
|
return core.BranchWatch{}, false, fmt.Errorf("find branch watch: %w", err)
|
|
}
|
|
return out, true, nil
|
|
}
|
|
|
|
func (s *pgBranchWatchStore) ListBranchWatches(ctx context.Context) ([]core.BranchWatch, error) {
|
|
const q = `SELECT id, provider, repo_id, branch, created_at FROM branch_watches ORDER BY id`
|
|
rows, err := s.pool.Query(ctx, q)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list branch watches: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []core.BranchWatch
|
|
for rows.Next() {
|
|
var w core.BranchWatch
|
|
if err := rows.Scan(&w.ID, &w.Provider, &w.RepoID, &w.Branch, &w.CreatedAt); err != nil {
|
|
return nil, fmt.Errorf("scan branch watch: %w", err)
|
|
}
|
|
out = append(out, w)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// pgRevisionCursorStore
|
|
|
|
type pgRevisionCursorStore struct{ pool *pgxpool.Pool }
|
|
|
|
func (s *pgRevisionCursorStore) UpsertRevisionCursor(ctx context.Context, cursor core.RevisionCursor) error {
|
|
const q = `
|
|
INSERT INTO revision_cursors (repo_id, branch, revision, observed_at)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (repo_id, branch) DO UPDATE
|
|
SET revision = EXCLUDED.revision, observed_at = EXCLUDED.observed_at
|
|
`
|
|
_, err := s.pool.Exec(ctx, q, cursor.RepoID, cursor.Branch, cursor.Revision, cursor.ObservedAt)
|
|
if err != nil {
|
|
return fmt.Errorf("upsert revision cursor: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *pgRevisionCursorStore) GetRevisionCursor(ctx context.Context, repoID, branch string) (core.RevisionCursor, bool, error) {
|
|
const q = `SELECT repo_id, branch, revision, observed_at FROM revision_cursors WHERE repo_id=$1 AND branch=$2`
|
|
var out core.RevisionCursor
|
|
row := s.pool.QueryRow(ctx, q, repoID, branch)
|
|
if err := row.Scan(&out.RepoID, &out.Branch, &out.Revision, &out.ObservedAt); err != nil {
|
|
if isNoRows(err) {
|
|
return core.RevisionCursor{}, false, nil
|
|
}
|
|
return core.RevisionCursor{}, false, fmt.Errorf("get revision cursor: %w", err)
|
|
}
|
|
return out, true, nil
|
|
}
|
|
|
|
// pgProviderDeliveryStore
|
|
|
|
type pgProviderDeliveryStore struct{ pool *pgxpool.Pool }
|
|
|
|
func (s *pgProviderDeliveryStore) RecordOnce(ctx context.Context, delivery core.ProviderDelivery) (DeliveryResult, error) {
|
|
const insert = `
|
|
INSERT INTO provider_deliveries
|
|
(id, provider, delivery_id, dedupe_key, event_id, repo_id, branch, revision, created_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
|
|
ON CONFLICT (provider, dedupe_key) DO NOTHING
|
|
`
|
|
tag, err := s.pool.Exec(ctx, insert,
|
|
delivery.ID, delivery.Provider, delivery.DeliveryID, delivery.DedupeKey,
|
|
delivery.EventID, delivery.RepoID, delivery.Branch, delivery.Revision,
|
|
delivery.CreatedAt,
|
|
)
|
|
if err != nil {
|
|
return DeliveryResult{}, fmt.Errorf("record delivery: %w", err)
|
|
}
|
|
if tag.RowsAffected() == 1 {
|
|
return DeliveryResult{First: true}, nil
|
|
}
|
|
// Duplicate: fetch existing event_id and created_at for idempotent response.
|
|
const sel = `SELECT event_id, created_at FROM provider_deliveries WHERE provider=$1 AND dedupe_key=$2`
|
|
var existingEventID string
|
|
var existingCreatedAt time.Time
|
|
row := s.pool.QueryRow(ctx, sel, delivery.Provider, delivery.DedupeKey)
|
|
if err := row.Scan(&existingEventID, &existingCreatedAt); err != nil {
|
|
return DeliveryResult{}, fmt.Errorf("fetch existing delivery: %w", err)
|
|
}
|
|
return DeliveryResult{
|
|
First: false,
|
|
ExistingEventID: existingEventID,
|
|
ExistingCreatedAt: existingCreatedAt,
|
|
}, nil
|
|
}
|
|
|
|
func isNoRows(err error) bool {
|
|
return err != nil && (errors.Is(err, pgx.ErrNoRows) || strings.Contains(err.Error(), "no rows"))
|
|
}
|
|
|
|
func isUniqueViolation(err error) bool {
|
|
var pgErr *pgconn.PgError
|
|
return errors.As(err, &pgErr) && pgErr.Code == "23505"
|
|
}
|