initial implementation: encapi ENC server + CLI

Postgres-backed External Node Classifier for Puppet, replacing Cobbler. - encapi HTTP server (chi + pgx): read/write API + two ENC document shapes (reshaped for the exec terminus; cobbler-wire for enc_direct_facts.rb) - encapi-cli: classify/node/role/status CRUD + import-cobbler seeder - pkg/client Go SDK; unit tests across all packages (DB via testcontainers) - Dockerfile (distroless), Makefile, nfpm RPM (encapi-cli + encapi-enc wrapper), Woodpecker CI, docs/cutover.md
2026-07-04 23:45:15 +10:00
parent b83d214e9a
commit 373d21a744
42 changed files with 3575 additions and 1 deletions
@@ -0,0 +1,165 @@
+package database
+
+import (
+	"context"
+	"errors"
+	"os"
+	"testing"
+
+	"git.unkin.net/unkin/encapi/internal/testsupport"
+	"git.unkin.net/unkin/encapi/pkg/models"
+)
+
+var testDB *DB
+
+func TestMain(m *testing.M) {
+	ctx := context.Background()
+	dsn, terminate, err := testsupport.StartPostgres(ctx)
+	if err != nil {
+		// Docker unavailable: run so tests self-skip via requireDB.
+		os.Exit(m.Run())
+	}
+	db, err := New(dsn)
+	if err != nil {
+		terminate()
+		panic(err)
+	}
+	testDB = db
+
+	code := m.Run()
+	db.Close()
+	terminate()
+	if code != 0 {
+		os.Exit(code)
+	}
+}
+
+func requireDB(t *testing.T) {
+	t.Helper()
+	if testDB == nil {
+		t.Skip("Docker unavailable; skipping database integration test")
+	}
+}
+
+// clean truncates all tables between tests for isolation.
+func clean(t *testing.T) {
+	t.Helper()
+	_, err := testDB.Pool.Exec(context.Background(), `TRUNCATE nodes, roles, statuses CASCADE`)
+	if err != nil {
+		t.Fatalf("truncate: %v", err)
+	}
+}
+
+func seed(t *testing.T) {
+	t.Helper()
+	ctx := context.Background()
+	if err := testDB.UpsertStatus(ctx, &models.Status{Name: "testing"}); err != nil {
+		t.Fatalf("seed status: %v", err)
+	}
+	if err := testDB.UpsertRole(ctx, &models.Role{Name: "roles::base"}); err != nil {
+		t.Fatalf("seed role: %v", err)
+	}
+}
+
+func TestStatusCRUD(t *testing.T) {
+	requireDB(t)
+	clean(t)
+	ctx := context.Background()
+
+	if err := testDB.UpsertStatus(ctx, &models.Status{Name: "production", Description: "prod"}); err != nil {
+		t.Fatal(err)
+	}
+	got, err := testDB.GetStatus(ctx, "production")
+	if err != nil || got.Description != "prod" {
+		t.Fatalf("GetStatus = %+v, %v", got, err)
+	}
+	// upsert updates description
+	if err := testDB.UpsertStatus(ctx, &models.Status{Name: "production", Description: "changed"}); err != nil {
+		t.Fatal(err)
+	}
+	got, _ = testDB.GetStatus(ctx, "production")
+	if got.Description != "changed" {
+		t.Errorf("description = %q, want changed", got.Description)
+	}
+	list, err := testDB.ListStatuses(ctx)
+	if err != nil || len(list) != 1 {
+		t.Fatalf("ListStatuses = %v, %v", list, err)
+	}
+	if err := testDB.DeleteStatus(ctx, "production"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := testDB.GetStatus(ctx, "production"); !errors.Is(err, ErrNotFound) {
+		t.Errorf("GetStatus after delete = %v, want ErrNotFound", err)
+	}
+}
+
+func TestRoleCRUDWithParams(t *testing.T) {
+	requireDB(t)
+	clean(t)
+	ctx := context.Background()
+
+	r := &models.Role{Name: "roles::infra::storage::minio", Description: "minio", DefaultParams: map[string]any{"minio_pool": "pool1", "replicas": float64(3)}}
+	if err := testDB.UpsertRole(ctx, r); err != nil {
+		t.Fatal(err)
+	}
+	got, err := testDB.GetRole(ctx, r.Name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.DefaultParams["minio_pool"] != "pool1" || got.DefaultParams["replicas"] != float64(3) {
+		t.Errorf("default_params = %#v", got.DefaultParams)
+	}
+	if _, err := testDB.GetRole(ctx, "nope"); !errors.Is(err, ErrNotFound) {
+		t.Errorf("GetRole(nope) = %v, want ErrNotFound", err)
+	}
+}
+
+func TestNodeCRUDAndForeignKeys(t *testing.T) {
+	requireDB(t)
+	clean(t)
+	seed(t)
+	ctx := context.Background()
+
+	// node referencing an unknown role must fail the FK
+	badRole := &models.Node{Certname: "h1", Role: "roles::ghost", Environment: "testing"}
+	if err := testDB.UpsertNode(ctx, badRole); err == nil {
+		t.Error("expected FK violation for unknown role")
+	}
+	// node referencing an unknown environment must fail the FK
+	badEnv := &models.Node{Certname: "h1", Role: "roles::base", Environment: "ghost"}
+	if err := testDB.UpsertNode(ctx, badEnv); err == nil {
+		t.Error("expected FK violation for unknown environment")
+	}
+
+	n := &models.Node{Certname: "h1", Role: "roles::base", Environment: "testing", Params: map[string]any{"x": "y"}}
+	if err := testDB.UpsertNode(ctx, n); err != nil {
+		t.Fatal(err)
+	}
+	got, err := testDB.GetNode(ctx, "h1")
+	if err != nil || got.Role != "roles::base" || got.Params["x"] != "y" {
+		t.Fatalf("GetNode = %+v, %v", got, err)
+	}
+
+	// role in use cannot be deleted
+	if err := testDB.DeleteRole(ctx, "roles::base"); err == nil {
+		t.Error("expected error deleting role in use")
+	}
+	// status in use cannot be deleted
+	if err := testDB.DeleteStatus(ctx, "testing"); err == nil {
+		t.Error("expected error deleting status in use")
+	}
+
+	list, err := testDB.ListNodes(ctx)
+	if err != nil || len(list) != 1 {
+		t.Fatalf("ListNodes = %v, %v", list, err)
+	}
+	if err := testDB.DeleteNode(ctx, "h1"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := testDB.GetNode(ctx, "h1"); !errors.Is(err, ErrNotFound) {
+		t.Errorf("GetNode after delete = %v, want ErrNotFound", err)
+	}
+	if err := testDB.DeleteNode(ctx, "h1"); !errors.Is(err, ErrNotFound) {
+		t.Errorf("DeleteNode missing = %v, want ErrNotFound", err)
+	}
+}
@@ -0,0 +1,91 @@
+package database
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+
+	"git.unkin.net/unkin/encapi/pkg/models"
+)
+
+// UpsertNode creates or updates a host-to-role assignment. The referenced role
+// and environment must already exist (enforced by foreign keys).
+func (db *DB) UpsertNode(ctx context.Context, n *models.Node) error {
+	params, err := marshalParams(n.Params)
+	if err != nil {
+		return fmt.Errorf("marshal params for node %q: %w", n.Certname, err)
+	}
+	_, err = db.Pool.Exec(ctx, `
+		INSERT INTO nodes (certname, role, environment, params)
+		VALUES ($1, $2, $3, $4)
+		ON CONFLICT (certname) DO UPDATE
+		SET role = EXCLUDED.role,
+		    environment = EXCLUDED.environment,
+		    params = EXCLUDED.params,
+		    updated_at = NOW()
+	`, n.Certname, n.Role, n.Environment, params)
+	if err != nil {
+		return fmt.Errorf("upsert node %q: %w", n.Certname, err)
+	}
+	return nil
+}
+
+// GetNode returns a single node or ErrNotFound.
+func (db *DB) GetNode(ctx context.Context, certname string) (*models.Node, error) {
+	var (
+		n      models.Node
+		params []byte
+	)
+	err := db.Pool.QueryRow(ctx,
+		`SELECT certname, role, environment, params FROM nodes WHERE certname = $1`, certname,
+	).Scan(&n.Certname, &n.Role, &n.Environment, &params)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, fmt.Errorf("get node %q: %w", certname, err)
+	}
+	if n.Params, err = unmarshalParams(params); err != nil {
+		return nil, fmt.Errorf("decode params for node %q: %w", certname, err)
+	}
+	return &n, nil
+}
+
+// ListNodes returns all nodes ordered by certname.
+func (db *DB) ListNodes(ctx context.Context) ([]models.Node, error) {
+	rows, err := db.Pool.Query(ctx, `SELECT certname, role, environment, params FROM nodes ORDER BY certname`)
+	if err != nil {
+		return nil, fmt.Errorf("list nodes: %w", err)
+	}
+	defer rows.Close()
+
+	out := []models.Node{}
+	for rows.Next() {
+		var (
+			n      models.Node
+			params []byte
+		)
+		if err := rows.Scan(&n.Certname, &n.Role, &n.Environment, &params); err != nil {
+			return nil, fmt.Errorf("scan node: %w", err)
+		}
+		if n.Params, err = unmarshalParams(params); err != nil {
+			return nil, fmt.Errorf("decode params: %w", err)
+		}
+		out = append(out, n)
+	}
+	return out, rows.Err()
+}
+
+// DeleteNode removes a host assignment.
+func (db *DB) DeleteNode(ctx context.Context, certname string) error {
+	tag, err := db.Pool.Exec(ctx, `DELETE FROM nodes WHERE certname = $1`, certname)
+	if err != nil {
+		return fmt.Errorf("delete node %q: %w", certname, err)
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
@@ -0,0 +1,71 @@
+// Package database is the Postgres persistence layer for encapi. It stores
+// three entities — statuses (Puppet environments), roles (class assignment
+// targets with inheritable default params), and nodes (host-to-role
+// assignments) — and enforces referential integrity between them.
+package database
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// DB wraps a pgx connection pool.
+type DB struct {
+	Pool *pgxpool.Pool
+}
+
+// New connects to Postgres, verifies the connection, and runs migrations.
+func New(dsn string) (*DB, error) {
+	pool, err := pgxpool.New(context.Background(), dsn)
+	if err != nil {
+		return nil, fmt.Errorf("connect to postgres: %w", 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
+}
+
+// Close releases the pool.
+func (db *DB) Close() { db.Pool.Close() }
+
+func (db *DB) migrate() error {
+	_, err := db.Pool.Exec(context.Background(), `
+		CREATE TABLE IF NOT EXISTS statuses (
+			name        TEXT PRIMARY KEY,
+			description TEXT NOT NULL DEFAULT '',
+			created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+			updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
+		);
+
+		CREATE TABLE IF NOT EXISTS roles (
+			name           TEXT PRIMARY KEY,
+			description    TEXT NOT NULL DEFAULT '',
+			default_params JSONB NOT NULL DEFAULT '{}',
+			created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+			updated_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
+		);
+
+		CREATE TABLE IF NOT EXISTS nodes (
+			certname    TEXT PRIMARY KEY,
+			role        TEXT NOT NULL REFERENCES roles(name) ON UPDATE CASCADE,
+			environment TEXT NOT NULL REFERENCES statuses(name) ON UPDATE CASCADE,
+			params      JSONB NOT NULL DEFAULT '{}',
+			created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+			updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
+		);
+	`)
+	if err != nil {
+		return err
+	}
+	return nil
+}
@@ -0,0 +1,110 @@
+package database
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+
+	"git.unkin.net/unkin/encapi/pkg/models"
+)
+
+// UpsertRole creates or updates a role and its inheritable default params.
+func (db *DB) UpsertRole(ctx context.Context, r *models.Role) error {
+	params, err := marshalParams(r.DefaultParams)
+	if err != nil {
+		return fmt.Errorf("marshal default_params for role %q: %w", r.Name, err)
+	}
+	_, err = db.Pool.Exec(ctx, `
+		INSERT INTO roles (name, description, default_params)
+		VALUES ($1, $2, $3)
+		ON CONFLICT (name) DO UPDATE
+		SET description = EXCLUDED.description,
+		    default_params = EXCLUDED.default_params,
+		    updated_at = NOW()
+	`, r.Name, r.Description, params)
+	if err != nil {
+		return fmt.Errorf("upsert role %q: %w", r.Name, err)
+	}
+	return nil
+}
+
+// GetRole returns a single role or ErrNotFound.
+func (db *DB) GetRole(ctx context.Context, name string) (*models.Role, error) {
+	var (
+		r      models.Role
+		params []byte
+	)
+	err := db.Pool.QueryRow(ctx,
+		`SELECT name, description, default_params FROM roles WHERE name = $1`, name,
+	).Scan(&r.Name, &r.Description, &params)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, fmt.Errorf("get role %q: %w", name, err)
+	}
+	if r.DefaultParams, err = unmarshalParams(params); err != nil {
+		return nil, fmt.Errorf("decode default_params for role %q: %w", name, err)
+	}
+	return &r, nil
+}
+
+// ListRoles returns all roles ordered by name.
+func (db *DB) ListRoles(ctx context.Context) ([]models.Role, error) {
+	rows, err := db.Pool.Query(ctx, `SELECT name, description, default_params FROM roles ORDER BY name`)
+	if err != nil {
+		return nil, fmt.Errorf("list roles: %w", err)
+	}
+	defer rows.Close()
+
+	out := []models.Role{}
+	for rows.Next() {
+		var (
+			r      models.Role
+			params []byte
+		)
+		if err := rows.Scan(&r.Name, &r.Description, &params); err != nil {
+			return nil, fmt.Errorf("scan role: %w", err)
+		}
+		if r.DefaultParams, err = unmarshalParams(params); err != nil {
+			return nil, fmt.Errorf("decode default_params: %w", err)
+		}
+		out = append(out, r)
+	}
+	return out, rows.Err()
+}
+
+// DeleteRole removes a role. It fails if any node still references it.
+func (db *DB) DeleteRole(ctx context.Context, name string) error {
+	tag, err := db.Pool.Exec(ctx, `DELETE FROM roles WHERE name = $1`, name)
+	if err != nil {
+		return fmt.Errorf("delete role %q: %w", name, err)
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+// marshalParams renders a params map to JSONB bytes, treating nil as {}.
+func marshalParams(m map[string]any) ([]byte, error) {
+	if m == nil {
+		return []byte("{}"), nil
+	}
+	return json.Marshal(m)
+}
+
+// unmarshalParams decodes JSONB bytes into a params map, treating empty as {}.
+func unmarshalParams(b []byte) (map[string]any, error) {
+	m := map[string]any{}
+	if len(b) == 0 {
+		return m, nil
+	}
+	if err := json.Unmarshal(b, &m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
@@ -0,0 +1,75 @@
+package database
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+
+	"git.unkin.net/unkin/encapi/pkg/models"
+)
+
+// ErrNotFound is returned when a requested entity does not exist.
+var ErrNotFound = errors.New("not found")
+
+// UpsertStatus creates or updates a status (Puppet environment).
+func (db *DB) UpsertStatus(ctx context.Context, s *models.Status) error {
+	_, err := db.Pool.Exec(ctx, `
+		INSERT INTO statuses (name, description)
+		VALUES ($1, $2)
+		ON CONFLICT (name) DO UPDATE
+		SET description = EXCLUDED.description, updated_at = NOW()
+	`, s.Name, s.Description)
+	if err != nil {
+		return fmt.Errorf("upsert status %q: %w", s.Name, err)
+	}
+	return nil
+}
+
+// GetStatus returns a single status or ErrNotFound.
+func (db *DB) GetStatus(ctx context.Context, name string) (*models.Status, error) {
+	var s models.Status
+	err := db.Pool.QueryRow(ctx,
+		`SELECT name, description FROM statuses WHERE name = $1`, name,
+	).Scan(&s.Name, &s.Description)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, fmt.Errorf("get status %q: %w", name, err)
+	}
+	return &s, nil
+}
+
+// ListStatuses returns all statuses ordered by name.
+func (db *DB) ListStatuses(ctx context.Context) ([]models.Status, error) {
+	rows, err := db.Pool.Query(ctx, `SELECT name, description FROM statuses ORDER BY name`)
+	if err != nil {
+		return nil, fmt.Errorf("list statuses: %w", err)
+	}
+	defer rows.Close()
+
+	out := []models.Status{}
+	for rows.Next() {
+		var s models.Status
+		if err := rows.Scan(&s.Name, &s.Description); err != nil {
+			return nil, fmt.Errorf("scan status: %w", err)
+		}
+		out = append(out, s)
+	}
+	return out, rows.Err()
+}
+
+// DeleteStatus removes a status. It fails if any node still references it
+// (enforced by the nodes.environment foreign key).
+func (db *DB) DeleteStatus(ctx context.Context, name string) error {
+	tag, err := db.Pool.Exec(ctx, `DELETE FROM statuses WHERE name = $1`, name)
+	if err != nil {
+		return fmt.Errorf("delete status %q: %w", name, err)
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrNotFound
+	}
+	return nil
+}