Adopt golib/pg for migrations and pool construction
artifactapi's schema was a ~180-line inline DDL blob re-executed on every start, growing an ALTER TABLE ... IF NOT EXISTS line per change with nothing recording what had run. golib/pg already owns that mechanic for the estate, so move the SQL into a versioned, embedded set and let the library apply it. - Depend on git.unkin.net/unkin/golib v0.1.0. - Move the DDL verbatim to migrations/0001_init.sql, embedded via the new migrations package, and build the pool with pg.NewMigrated (LockName "artifactapi-migrations"). The runner adds a cluster-wide advisory lock the old blob never took, so replicas starting together queue instead of racing each other through the DDL. - The live database has the schema but no schema_migrations, so its first start on this build re-runs 0001. Every statement is IF NOT EXISTS-guarded, so that run is a no-op landing only the tracking row; a container-backed test drops the row from a migrated database and asserts exactly that, and a static guard keeps future migrations additive and idempotent. - Keep config.DatabaseDSN as the DSN source rather than pg.DSNFromEnv: the variable names match, but golib has no default user or database name, and artifactapi documents and ships DBUSER/DBNAME defaults of "artifacts". The deployed env var contract is unchanged. - Guard the embedded set against migrations/ and pin the derived advisory key, so neither can drift unnoticed. - Plumb GOPRIVATE=git.unkin.net for the first cross-repo Go dependency: exported by the Makefile, set in the Dockerfile and the woodpecker Go steps, documented in the README.
This commit is contained in:
@@ -0,0 +1,162 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io/fs"
|
||||
"maps"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.unkin.net/unkin/golib/pg"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/migrations"
|
||||
)
|
||||
|
||||
// migrationsDir is the repo's migrations/ directory, relative to this package.
|
||||
const migrationsDir = "../../migrations"
|
||||
|
||||
func readMigrationsFromDisk(t *testing.T) map[string]string {
|
||||
t.Helper()
|
||||
entries, err := os.ReadDir(migrationsDir)
|
||||
if err != nil {
|
||||
t.Fatalf("read %s: %v", migrationsDir, err)
|
||||
}
|
||||
files := map[string]string{}
|
||||
for _, e := range entries {
|
||||
if e.IsDir() || !strings.HasSuffix(e.Name(), ".sql") {
|
||||
continue
|
||||
}
|
||||
b, err := os.ReadFile(filepath.Join(migrationsDir, e.Name()))
|
||||
if err != nil {
|
||||
t.Fatalf("read %s: %v", e.Name(), err)
|
||||
}
|
||||
files[e.Name()] = string(b)
|
||||
}
|
||||
if len(files) == 0 {
|
||||
t.Fatalf("no .sql files in %s", migrationsDir)
|
||||
}
|
||||
return files
|
||||
}
|
||||
|
||||
// The embedded set is the shipped schema, so it must match the migrations/
|
||||
// directory exactly — a file added on disk but not embedded would never run.
|
||||
func TestEmbeddedMigrationsMatchDirectory(t *testing.T) {
|
||||
onDisk := readMigrationsFromDisk(t)
|
||||
entries, err := fs.ReadDir(migrations.FS, ".")
|
||||
if err != nil {
|
||||
t.Fatalf("read embedded migrations: %v", err)
|
||||
}
|
||||
embedded := map[string]string{}
|
||||
for _, e := range entries {
|
||||
if e.IsDir() || !strings.HasSuffix(e.Name(), ".sql") {
|
||||
continue
|
||||
}
|
||||
b, err := migrations.FS.ReadFile(e.Name())
|
||||
if err != nil {
|
||||
t.Fatalf("read embedded %s: %v", e.Name(), err)
|
||||
}
|
||||
embedded[e.Name()] = string(b)
|
||||
}
|
||||
if len(embedded) != len(onDisk) {
|
||||
t.Fatalf("embedded %d files, migrations/ has %d", len(embedded), len(onDisk))
|
||||
}
|
||||
for name, body := range embedded {
|
||||
want, ok := onDisk[name]
|
||||
if !ok {
|
||||
t.Errorf("%s is embedded but not in migrations/", name)
|
||||
continue
|
||||
}
|
||||
if body != want {
|
||||
t.Errorf("%s: embedded body differs from migrations/%s", name, name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var (
|
||||
createTableRe = regexp.MustCompile(`(?i)\bCREATE\s+TABLE\b(\s+IF\s+NOT\s+EXISTS\b)?`)
|
||||
createIndexRe = regexp.MustCompile(`(?i)\bCREATE\s+(UNIQUE\s+)?INDEX\b(\s+IF\s+NOT\s+EXISTS\b)?`)
|
||||
addColumnRe = regexp.MustCompile(`(?i)\bADD\s+COLUMN\b(\s+IF\s+NOT\s+EXISTS\b)?`)
|
||||
destructiveRe = regexp.MustCompile(`(?i)\b(DROP\s+(TABLE|COLUMN|INDEX)|TRUNCATE|DELETE\s+FROM)\b`)
|
||||
)
|
||||
|
||||
// A migration absent from schema_migrations is re-run even when the live
|
||||
// database already has the schema, which is exactly how the deployed database —
|
||||
// migrated for years by an untracked inline DDL blob — picks 0001 up. Every
|
||||
// statement must therefore be idempotent, or that first tracked run would fail
|
||||
// against production.
|
||||
func TestMigrationsAreIdempotent(t *testing.T) {
|
||||
for name, body := range readMigrationsFromDisk(t) {
|
||||
for _, m := range createTableRe.FindAllStringSubmatch(body, -1) {
|
||||
if m[1] == "" {
|
||||
t.Errorf("%s: %q is not IF NOT EXISTS-guarded", name, strings.Join(strings.Fields(m[0]), " "))
|
||||
}
|
||||
}
|
||||
for _, m := range createIndexRe.FindAllStringSubmatch(body, -1) {
|
||||
if m[2] == "" {
|
||||
t.Errorf("%s: %q is not IF NOT EXISTS-guarded", name, strings.Join(strings.Fields(m[0]), " "))
|
||||
}
|
||||
}
|
||||
for _, m := range addColumnRe.FindAllStringSubmatch(body, -1) {
|
||||
if m[1] == "" {
|
||||
t.Errorf("%s: %q is not IF NOT EXISTS-guarded", name, strings.Join(strings.Fields(m[0]), " "))
|
||||
}
|
||||
}
|
||||
if loc := destructiveRe.FindString(body); loc != "" {
|
||||
t.Errorf("%s: destructive statement %q; migrations are additive", name, loc)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The deployed database was built by the untracked inline DDL this runner
|
||||
// replaces, so its very first tracked start runs 0001 against a schema that
|
||||
// already exists. Reproduce that by dropping the tracking row from an
|
||||
// already-migrated database and starting again: it must succeed and re-record
|
||||
// the version, changing nothing else.
|
||||
func TestMigratingAnAlreadyPopulatedSchemaIsANoOp(t *testing.T) {
|
||||
requireDB(t)
|
||||
c := context.Background()
|
||||
|
||||
if _, err := testDB.Pool.Exec(c, "DELETE FROM schema_migrations"); err != nil {
|
||||
t.Fatalf("clear schema_migrations: %v", err)
|
||||
}
|
||||
db, err := New(testDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("migrate over an existing schema: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
var versions []string
|
||||
rows, err := db.Pool.Query(c, "SELECT version FROM schema_migrations ORDER BY version")
|
||||
if err != nil {
|
||||
t.Fatalf("read schema_migrations: %v", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var v string
|
||||
if err := rows.Scan(&v); err != nil {
|
||||
t.Fatalf("scan version: %v", err)
|
||||
}
|
||||
versions = append(versions, v)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatalf("read schema_migrations: %v", err)
|
||||
}
|
||||
|
||||
want := slices.Sorted(maps.Keys(readMigrationsFromDisk(t)))
|
||||
if !slices.Equal(versions, want) {
|
||||
t.Fatalf("schema_migrations = %v, want %v", versions, want)
|
||||
}
|
||||
}
|
||||
|
||||
// The advisory lock key is derived from migrationLockName by golib. Pin it so a
|
||||
// rename cannot silently let two builds migrate the same cluster at once.
|
||||
func TestMigrationLockKeyIsPinned(t *testing.T) {
|
||||
const wantKey int64 = -6981019939451326383
|
||||
if got := pg.LockKey(migrationLockName); got != wantKey {
|
||||
t.Fatalf("LockKey(%q) = %d, want %d", migrationLockName, got, wantKey)
|
||||
}
|
||||
}
|
||||
+15
-230
@@ -2,249 +2,34 @@ package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"git.unkin.net/unkin/golib/pg"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/migrations"
|
||||
)
|
||||
|
||||
// migrationLockName names the cluster-wide advisory lock the migration run
|
||||
// contends for; golib derives the key as FNV-1a/64 of it. Replicas starting
|
||||
// together queue on it instead of racing each other through the DDL.
|
||||
const migrationLockName = "artifactapi-migrations"
|
||||
|
||||
type DB struct {
|
||||
Pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func New(dsn string) (*DB, error) {
|
||||
pool, err := pgxpool.New(context.Background(), dsn)
|
||||
ctx := context.Background()
|
||||
pool, err := pg.NewMigrated(ctx, dsn, migrations.FS, pg.MigrateOptions{
|
||||
LockName: migrationLockName,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("connect to postgres: %w", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := pool.Ping(context.Background()); err != nil {
|
||||
pool.Close()
|
||||
return nil, fmt.Errorf("ping postgres: %w", err)
|
||||
}
|
||||
|
||||
db := &DB{Pool: pool}
|
||||
if err := db.migrate(); err != nil {
|
||||
pool.Close()
|
||||
return nil, fmt.Errorf("run migrations: %w", err)
|
||||
}
|
||||
|
||||
return db, nil
|
||||
return &DB{Pool: pool}, nil
|
||||
}
|
||||
|
||||
func (db *DB) Close() {
|
||||
db.Pool.Close()
|
||||
}
|
||||
|
||||
func (db *DB) migrate() error {
|
||||
ctx := context.Background()
|
||||
|
||||
_, err := db.Pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS remotes (
|
||||
name TEXT PRIMARY KEY,
|
||||
package_type TEXT NOT NULL,
|
||||
repo_type TEXT DEFAULT 'remote',
|
||||
base_url TEXT NOT NULL DEFAULT '',
|
||||
mirrorlist TEXT[] DEFAULT '{}',
|
||||
mirror_strategy TEXT NOT NULL DEFAULT 'round_robin',
|
||||
description TEXT DEFAULT '',
|
||||
username TEXT DEFAULT '',
|
||||
password TEXT DEFAULT '',
|
||||
immutable_ttl INTEGER DEFAULT 0,
|
||||
mutable_ttl INTEGER DEFAULT 3600,
|
||||
check_mutable BOOLEAN DEFAULT TRUE,
|
||||
patterns TEXT[] DEFAULT '{}',
|
||||
blocklist TEXT[] DEFAULT '{}',
|
||||
mutable_patterns TEXT[] DEFAULT '{}',
|
||||
immutable_patterns TEXT[] DEFAULT '{}',
|
||||
ban_tags_enabled BOOLEAN DEFAULT FALSE,
|
||||
ban_tags TEXT[] DEFAULT '{}',
|
||||
quarantine_enabled BOOLEAN DEFAULT FALSE,
|
||||
quarantine_days INTEGER DEFAULT 3,
|
||||
stale_on_error BOOLEAN DEFAULT TRUE,
|
||||
releases_remote TEXT DEFAULT '',
|
||||
managed_by TEXT DEFAULT '',
|
||||
created_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
updated_at TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS virtuals (
|
||||
name TEXT PRIMARY KEY,
|
||||
package_type TEXT NOT NULL,
|
||||
description TEXT DEFAULT '',
|
||||
members TEXT[] NOT NULL,
|
||||
managed_by TEXT DEFAULT '',
|
||||
created_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
updated_at TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS blobs (
|
||||
content_hash TEXT PRIMARY KEY,
|
||||
s3_key TEXT NOT NULL,
|
||||
size_bytes BIGINT NOT NULL,
|
||||
content_type TEXT DEFAULT 'application/octet-stream',
|
||||
created_at TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS artifacts (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
remote_name TEXT NOT NULL REFERENCES remotes(name) ON DELETE CASCADE,
|
||||
path TEXT NOT NULL,
|
||||
content_hash TEXT NOT NULL REFERENCES blobs(content_hash),
|
||||
upstream_etag TEXT DEFAULT '',
|
||||
upstream_last_modified TIMESTAMPTZ,
|
||||
first_seen_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
last_fetched_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
last_accessed_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
fetch_count BIGINT DEFAULT 1,
|
||||
access_count BIGINT DEFAULT 1,
|
||||
UNIQUE(remote_name, path)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_artifacts_remote ON artifacts(remote_name);
|
||||
CREATE INDEX IF NOT EXISTS idx_artifacts_last_accessed ON artifacts(last_accessed_at);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS local_files (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
repo_name TEXT NOT NULL,
|
||||
file_path TEXT NOT NULL,
|
||||
content_hash TEXT NOT NULL REFERENCES blobs(content_hash),
|
||||
created_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
UNIQUE(repo_name, file_path)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS access_log (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
remote_name TEXT NOT NULL,
|
||||
path TEXT NOT NULL,
|
||||
cache_hit BOOLEAN NOT NULL,
|
||||
size_bytes BIGINT DEFAULT 0,
|
||||
upstream_ms INTEGER DEFAULT 0,
|
||||
client_ip TEXT DEFAULT '',
|
||||
created_at TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_access_log_remote_time ON access_log(remote_name, created_at);
|
||||
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS repo_type TEXT DEFAULT 'remote';
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS mirrorlist TEXT[] DEFAULT '{}';
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS mirror_strategy TEXT NOT NULL DEFAULT 'round_robin';
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS upstream_dial_timeout INTEGER DEFAULT 0;
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS upstream_tls_timeout INTEGER DEFAULT 0;
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS upstream_response_header_timeout INTEGER DEFAULT 0;
|
||||
|
||||
CREATE TABLE IF NOT EXISTS rpm_metadata (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
repo_name TEXT NOT NULL,
|
||||
file_path TEXT NOT NULL,
|
||||
content_hash TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
epoch INTEGER DEFAULT 0,
|
||||
version TEXT NOT NULL,
|
||||
release TEXT NOT NULL,
|
||||
arch TEXT NOT NULL,
|
||||
summary TEXT DEFAULT '',
|
||||
description TEXT DEFAULT '',
|
||||
rpm_size BIGINT DEFAULT 0,
|
||||
installed_size BIGINT DEFAULT 0,
|
||||
license TEXT DEFAULT '',
|
||||
vendor TEXT DEFAULT '',
|
||||
build_group TEXT DEFAULT '',
|
||||
build_host TEXT DEFAULT '',
|
||||
source_rpm TEXT DEFAULT '',
|
||||
url TEXT DEFAULT '',
|
||||
packager TEXT DEFAULT '',
|
||||
requires JSONB DEFAULT '[]',
|
||||
provides JSONB DEFAULT '[]',
|
||||
conflicts JSONB DEFAULT '[]',
|
||||
obsoletes JSONB DEFAULT '[]',
|
||||
files JSONB DEFAULT '[]',
|
||||
changelogs JSONB DEFAULT '[]',
|
||||
created_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
UNIQUE(repo_name, file_path)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_rpm_metadata_repo ON rpm_metadata(repo_name);
|
||||
|
||||
ALTER TABLE rpm_metadata ADD COLUMN IF NOT EXISTS conflicts JSONB DEFAULT '[]';
|
||||
ALTER TABLE rpm_metadata ADD COLUMN IF NOT EXISTS obsoletes JSONB DEFAULT '[]';
|
||||
|
||||
CREATE TABLE IF NOT EXISTS deb_metadata (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
repo_name TEXT NOT NULL,
|
||||
file_path TEXT NOT NULL,
|
||||
content_hash TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
version TEXT NOT NULL,
|
||||
architecture TEXT NOT NULL,
|
||||
control TEXT NOT NULL,
|
||||
size BIGINT DEFAULT 0,
|
||||
md5 TEXT DEFAULT '',
|
||||
sha256 TEXT DEFAULT '',
|
||||
created_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
UNIQUE(repo_name, file_path)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_deb_metadata_repo ON deb_metadata(repo_name);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS alpine_metadata (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
repo_name TEXT NOT NULL,
|
||||
file_path TEXT NOT NULL,
|
||||
content_hash TEXT NOT NULL,
|
||||
checksum TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
version TEXT NOT NULL,
|
||||
arch TEXT NOT NULL,
|
||||
download_size BIGINT DEFAULT 0,
|
||||
installed_size BIGINT DEFAULT 0,
|
||||
description TEXT DEFAULT '',
|
||||
url TEXT DEFAULT '',
|
||||
license TEXT DEFAULT '',
|
||||
origin TEXT DEFAULT '',
|
||||
maintainer TEXT DEFAULT '',
|
||||
build_time BIGINT DEFAULT 0,
|
||||
commit_hash TEXT DEFAULT '',
|
||||
provider_priority TEXT DEFAULT '',
|
||||
depends TEXT DEFAULT '',
|
||||
provides TEXT DEFAULT '',
|
||||
install_if TEXT DEFAULT '',
|
||||
created_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
UNIQUE(repo_name, file_path)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_alpine_metadata_repo ON alpine_metadata(repo_name);
|
||||
CREATE INDEX IF NOT EXISTS idx_alpine_metadata_repo_arch ON alpine_metadata(repo_name, arch);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS github_rpm_sync_state (
|
||||
remote_name TEXT PRIMARY KEY,
|
||||
etag TEXT DEFAULT '',
|
||||
last_synced_at TIMESTAMPTZ,
|
||||
sync_lease_owner TEXT DEFAULT '',
|
||||
sync_lease_expires TIMESTAMPTZ
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS github_deb_sync_state (
|
||||
remote_name TEXT PRIMARY KEY,
|
||||
etag TEXT DEFAULT '',
|
||||
last_synced_at TIMESTAMPTZ,
|
||||
sync_lease_owner TEXT DEFAULT '',
|
||||
sync_lease_expires TIMESTAMPTZ
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS github_alpine_sync_state (
|
||||
remote_name TEXT PRIMARY KEY,
|
||||
etag TEXT DEFAULT '',
|
||||
last_synced_at TIMESTAMPTZ,
|
||||
sync_lease_owner TEXT DEFAULT '',
|
||||
sync_lease_expires TIMESTAMPTZ
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS signing_keys (
|
||||
purpose TEXT PRIMARY KEY,
|
||||
private_key_armor TEXT NOT NULL,
|
||||
key_id TEXT NOT NULL,
|
||||
created_at TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
`)
|
||||
return err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user