From 99c553b7c6fb0ffb5ea4ac19df49613de93c1ca7 Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Mon, 12 Oct 2026 14:43:34 +0200 Subject: [PATCH] feat(master): postgres store with migrations, master identity and agents --- go.mod | 6 ++ go.sum | 15 +++ internal/master/model/model.go | 88 ++++++++++++++++++ internal/master/model/model_test.go | 41 +++++++++ internal/master/store/agents.go | 91 +++++++++++++++++++ .../master/store/agents_integration_test.go | 84 +++++++++++++++++ .../master/store/helpers_integration_test.go | 43 +++++++++ internal/master/store/identity.go | 42 +++++++++ internal/master/store/migrate.go | 68 ++++++++++++++ .../master/store/migrations/0001_init.sql | 48 ++++++++++ internal/master/store/store.go | 41 +++++++++ 11 files changed, 567 insertions(+) create mode 100644 internal/master/model/model.go create mode 100644 internal/master/model/model_test.go create mode 100644 internal/master/store/agents.go create mode 100644 internal/master/store/agents_integration_test.go create mode 100644 internal/master/store/helpers_integration_test.go create mode 100644 internal/master/store/identity.go create mode 100644 internal/master/store/migrate.go create mode 100644 internal/master/store/migrations/0001_init.sql create mode 100644 internal/master/store/store.go diff --git a/go.mod b/go.mod index 99445dc..db55330 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.26.3 require ( github.com/coder/websocket v1.8.15 github.com/docker/docker v28.5.2+incompatible + github.com/jackc/pgx/v5 v5.10.0 github.com/prometheus/client_golang v1.23.2 github.com/prometheus/client_model v0.6.2 github.com/stretchr/testify v1.11.1 @@ -24,6 +25,9 @@ require ( github.com/felixge/httpsnoop v1.0.4 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect github.com/moby/sys/atomicwriter v0.1.0 // indirect github.com/moby/term v0.5.2 // indirect @@ -42,7 +46,9 @@ require ( go.opentelemetry.io/otel/metric v1.43.0 // indirect go.opentelemetry.io/otel/trace v1.43.0 // indirect go.yaml.in/yaml/v2 v2.4.2 // indirect + golang.org/x/sync v0.20.0 // indirect golang.org/x/sys v0.42.0 // indirect + golang.org/x/text v0.35.0 // indirect golang.org/x/time v0.15.0 // indirect google.golang.org/protobuf v1.36.11 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index e8146cc..a1d0591 100644 --- a/go.sum +++ b/go.sum @@ -16,6 +16,7 @@ github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151X github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I= github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= @@ -39,6 +40,14 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= +github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= @@ -79,6 +88,9 @@ github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0t github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= @@ -107,6 +119,8 @@ go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0= golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= @@ -124,6 +138,7 @@ google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q= diff --git a/internal/master/model/model.go b/internal/master/model/model.go new file mode 100644 index 0000000..ed376be --- /dev/null +++ b/internal/master/model/model.go @@ -0,0 +1,88 @@ +// Package model holds the master's domain types shared by storage, the agent +// hub and the web UI. +package model + +import ( + "errors" + "strconv" + "strings" + "time" + + "github.com/shcizo/package-updater/internal/protocol" +) + +// Sentinel errors returned by storage. +var ( + ErrNotFound = errors.New("not found") + ErrConflict = errors.New("already exists") +) + +// Agent is a registered agent. Whether it is connected right now is held in +// memory by the hub, not here. +type Agent struct { + ID int64 + Name string + Hostname string + AgentVersion string + SwarmManager bool + LastSeenAt *time.Time + RevokedAt *time.Time + CreatedAt time.Time +} + +// Service is one service as last reported by an agent. +type Service struct { + ID int64 + Kind protocol.Kind + LocationKey string + AgentID *int64 + AgentName string + ComposeProject string + ComposeService string + WorkingDir string + SwarmClusterID string + SwarmService string + ContainerName string + ImageRepo string + ActualRef string + ActualDigest string + Status string + Self bool + UpdateMode string + FirstSeenAt time.Time + UpdatedAt time.Time +} + +// DisplayName is the human-facing name of the service. +func (s Service) DisplayName() string { + switch s.Kind { + case protocol.KindCompose: + return s.ComposeProject + "/" + s.ComposeService + case protocol.KindSwarm: + return s.SwarmService + default: + return s.ContainerName + } +} + +// Session is a logged-in browser session. +type Session struct { + ID string + Subject string + Name string + Email string + CSRFToken string + ExpiresAt time.Time +} + +// LocationKey turns an agent-relative protocol key into the fleet-wide key +// stored in services.location_key. Compose services and containers belong to +// the agent's machine; Swarm services belong to the cluster, so two manager +// agents in one cluster report the same row. +func LocationKey(agentID int64, clusterID, agentKey string) string { + kind, rest, _ := strings.Cut(agentKey, "|") + if kind == string(protocol.KindSwarm) { + return kind + "|" + clusterID + "|" + rest + } + return kind + "|" + strconv.FormatInt(agentID, 10) + "|" + rest +} diff --git a/internal/master/model/model_test.go b/internal/master/model/model_test.go new file mode 100644 index 0000000..0705f69 --- /dev/null +++ b/internal/master/model/model_test.go @@ -0,0 +1,41 @@ +package model + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/shcizo/package-updater/internal/protocol" +) + +func TestLocationKey(t *testing.T) { + cases := []struct { + name, agentKey, want string + }{ + {"compose is per agent", "compose|slusk|web", "compose|7|slusk|web"}, + {"container is per agent", "container|adhoc", "container|7|adhoc"}, + {"swarm is per cluster", "swarm|gitea_gitea", "swarm|cluster-1|gitea_gitea"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + assert.Equal(t, c.want, LocationKey(7, "cluster-1", c.agentKey)) + }) + } +} + +func TestDisplayName(t *testing.T) { + cases := []struct { + name string + svc Service + want string + }{ + {"compose", Service{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web"}, "slusk/web"}, + {"swarm", Service{Kind: protocol.KindSwarm, SwarmService: "gitea_gitea"}, "gitea_gitea"}, + {"container", Service{Kind: protocol.KindContainer, ContainerName: "adhoc"}, "adhoc"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + assert.Equal(t, c.want, c.svc.DisplayName()) + }) + } +} diff --git a/internal/master/store/agents.go b/internal/master/store/agents.go new file mode 100644 index 0000000..00a1e8a --- /dev/null +++ b/internal/master/store/agents.go @@ -0,0 +1,91 @@ +package store + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/protocol" +) + +const agentColumns = `id, name, hostname, agent_version, swarm_manager, last_seen_at, revoked_at, created_at` + +func scanAgent(row pgx.Row) (model.Agent, error) { + var a model.Agent + err := row.Scan(&a.ID, &a.Name, &a.Hostname, &a.AgentVersion, &a.SwarmManager, &a.LastSeenAt, &a.RevokedAt, &a.CreatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return model.Agent{}, model.ErrNotFound + } + return a, err +} + +// CreateAgent registers an agent. Returns model.ErrConflict if the name is taken. +func (s *Store) CreateAgent(ctx context.Context, name, tokenHash string) (model.Agent, error) { + a, err := scanAgent(s.pool.QueryRow(ctx, + `INSERT INTO agents (name, token_hash) VALUES ($1, $2) RETURNING `+agentColumns, name, tokenHash)) + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == "23505" { + return model.Agent{}, model.ErrConflict + } + if err != nil { + return model.Agent{}, fmt.Errorf("create agent: %w", err) + } + return a, nil +} + +// ListAgents returns all agents, including revoked ones, ordered by name. +func (s *Store) ListAgents(ctx context.Context) ([]model.Agent, error) { + rows, err := s.pool.Query(ctx, `SELECT `+agentColumns+` FROM agents ORDER BY name`) + if err != nil { + return nil, fmt.Errorf("list agents: %w", err) + } + defer rows.Close() + var out []model.Agent + for rows.Next() { + a, err := scanAgent(rows) + if err != nil { + return nil, fmt.Errorf("scan agent: %w", err) + } + out = append(out, a) + } + return out, rows.Err() +} + +// AgentByTokenHash finds the agent a token belongs to. Revoked agents are +// returned too; the caller decides. Returns model.ErrNotFound if none. +func (s *Store) AgentByTokenHash(ctx context.Context, hash string) (model.Agent, error) { + return scanAgent(s.pool.QueryRow(ctx, `SELECT `+agentColumns+` FROM agents WHERE token_hash = $1`, hash)) +} + +// RevokeAgent marks the agent revoked. Its token stops working immediately; +// closing a live connection is the hub's job. +func (s *Store) RevokeAgent(ctx context.Context, id int64, at time.Time) error { + if _, err := s.pool.Exec(ctx, + `UPDATE agents SET revoked_at = $2 WHERE id = $1 AND revoked_at IS NULL`, id, at); err != nil { + return fmt.Errorf("revoke agent: %w", err) + } + return nil +} + +// RecordHello stores what the agent reported about itself on connect. +func (s *Store) RecordHello(ctx context.Context, id int64, h protocol.Hello, at time.Time) error { + if _, err := s.pool.Exec(ctx, + `UPDATE agents SET hostname = $2, agent_version = $3, swarm_manager = $4, last_seen_at = $5 WHERE id = $1`, + id, h.Hostname, h.AgentVersion, h.SwarmManager, at); err != nil { + return fmt.Errorf("record hello: %w", err) + } + return nil +} + +// MarkSeen updates last_seen_at, called when an agent disconnects. +func (s *Store) MarkSeen(ctx context.Context, id int64, at time.Time) error { + if _, err := s.pool.Exec(ctx, `UPDATE agents SET last_seen_at = $2 WHERE id = $1`, id, at); err != nil { + return fmt.Errorf("mark agent seen: %w", err) + } + return nil +} diff --git a/internal/master/store/agents_integration_test.go b/internal/master/store/agents_integration_test.go new file mode 100644 index 0000000..eee4d9d --- /dev/null +++ b/internal/master/store/agents_integration_test.go @@ -0,0 +1,84 @@ +//go:build integration + +package store + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/pin" + "github.com/shcizo/package-updater/internal/protocol" +) + +func TestIdentity_CreatedOnceThenReused(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + k1, err := s.LoadOrCreateIdentity(ctx, pin.GenerateKey) + require.NoError(t, err) + k2, err := s.LoadOrCreateIdentity(ctx, pin.GenerateKey) + require.NoError(t, err) + assert.True(t, k1.Equal(k2), "a restarted master must keep its key or every agent breaks") +} + +func TestMigrationsAreIdempotent(t *testing.T) { + s := openTestStore(t) + require.NoError(t, migrate(context.Background(), s.pool)) +} + +func TestAgents_CreateLookupRevoke(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + + created, err := s.CreateAgent(ctx, "node-2", "hash-2") + require.NoError(t, err) + assert.Equal(t, "node-2", created.Name) + assert.Nil(t, created.RevokedAt) + + found, err := s.AgentByTokenHash(ctx, "hash-2") + require.NoError(t, err) + assert.Equal(t, created.ID, found.ID) + + _, err = s.AgentByTokenHash(ctx, "nope") + assert.ErrorIs(t, err, model.ErrNotFound) + + require.NoError(t, s.RevokeAgent(ctx, created.ID, time.Now())) + found, err = s.AgentByTokenHash(ctx, "hash-2") + require.NoError(t, err) + assert.NotNil(t, found.RevokedAt) +} + +func TestAgents_DuplicateNameIsConflict(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + _, err := s.CreateAgent(ctx, "node-2", "h1") + require.NoError(t, err) + _, err = s.CreateAgent(ctx, "node-2", "h2") + assert.ErrorIs(t, err, model.ErrConflict) +} + +func TestAgents_RecordHelloAndMarkSeen(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + a, err := s.CreateAgent(ctx, "swarm-mgr", "h") + require.NoError(t, err) + + at := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) + require.NoError(t, s.RecordHello(ctx, a.ID, protocol.Hello{ + Hostname: "swarm-mgr", AgentVersion: "v1", SwarmManager: true, SwarmClusterID: "c1", + }, at)) + require.NoError(t, s.MarkSeen(ctx, a.ID, at.Add(time.Minute))) + + agents, err := s.ListAgents(ctx) + require.NoError(t, err) + require.Len(t, agents, 1) + assert.Equal(t, "swarm-mgr", agents[0].Hostname) + assert.Equal(t, "v1", agents[0].AgentVersion) + assert.True(t, agents[0].SwarmManager) + require.NotNil(t, agents[0].LastSeenAt) + assert.True(t, agents[0].LastSeenAt.Equal(at.Add(time.Minute))) +} diff --git a/internal/master/store/helpers_integration_test.go b/internal/master/store/helpers_integration_test.go new file mode 100644 index 0000000..3cd08a4 --- /dev/null +++ b/internal/master/store/helpers_integration_test.go @@ -0,0 +1,43 @@ +//go:build integration + +package store + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/require" +) + +// openTestStore returns a Store migrated into a fresh schema that is dropped +// when the test ends, so tests never see each other's rows. +func openTestStore(t *testing.T) *Store { + t.Helper() + url := os.Getenv("DATABASE_URL") + if url == "" { + t.Skip("DATABASE_URL not set") + } + ctx := context.Background() + admin, err := pgx.Connect(ctx, url) + require.NoError(t, err) + schema := fmt.Sprintf("test_%d", time.Now().UnixNano()) + _, err = admin.Exec(ctx, "CREATE SCHEMA "+schema) + require.NoError(t, err) + t.Cleanup(func() { + _, _ = admin.Exec(ctx, "DROP SCHEMA "+schema+" CASCADE") + _ = admin.Close(ctx) + }) + + cfg, err := pgxpool.ParseConfig(url) + require.NoError(t, err) + cfg.ConnConfig.RuntimeParams["search_path"] = schema + s, err := openWithConfig(ctx, cfg) + require.NoError(t, err) + t.Cleanup(s.Close) + return s +} diff --git a/internal/master/store/identity.go b/internal/master/store/identity.go new file mode 100644 index 0000000..fb4e28d --- /dev/null +++ b/internal/master/store/identity.go @@ -0,0 +1,42 @@ +package store + +import ( + "context" + "crypto/ed25519" + "crypto/x509" + "errors" + "fmt" +) + +// LoadOrCreateIdentity returns the master's private key, creating and +// storing one with generate on first start. Keeping it in Postgres means it +// survives container recreation and is covered by database backups; losing +// it would force every agent to be reconfigured with a new MASTER_KEY. +func (s *Store) LoadOrCreateIdentity(ctx context.Context, generate func() (ed25519.PrivateKey, error)) (ed25519.PrivateKey, error) { + fresh, err := generate() + if err != nil { + return nil, err + } + der, err := x509.MarshalPKCS8PrivateKey(fresh) + if err != nil { + return nil, fmt.Errorf("marshal master key: %w", err) + } + if _, err := s.pool.Exec(ctx, + `INSERT INTO master_identity (id, private_key) VALUES (1, $1) ON CONFLICT (id) DO NOTHING`, der, + ); err != nil { + return nil, fmt.Errorf("store master key: %w", err) + } + var stored []byte + if err := s.pool.QueryRow(ctx, `SELECT private_key FROM master_identity WHERE id = 1`).Scan(&stored); err != nil { + return nil, fmt.Errorf("load master key: %w", err) + } + parsed, err := x509.ParsePKCS8PrivateKey(stored) + if err != nil { + return nil, fmt.Errorf("parse master key: %w", err) + } + key, ok := parsed.(ed25519.PrivateKey) + if !ok { + return nil, errors.New("stored master key is not ed25519") + } + return key, nil +} diff --git a/internal/master/store/migrate.go b/internal/master/store/migrate.go new file mode 100644 index 0000000..4be77e8 --- /dev/null +++ b/internal/master/store/migrate.go @@ -0,0 +1,68 @@ +package store + +import ( + "context" + "embed" + "fmt" + "io/fs" + + "github.com/jackc/pgx/v5/pgxpool" +) + +//go:embed migrations/*.sql +var migrationsFS embed.FS + +// migrate applies every embedded migration not yet recorded in +// schema_migrations, in file-name order, each in its own transaction. +func migrate(ctx context.Context, pool *pgxpool.Pool) error { + if _, err := pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations ( + version text PRIMARY KEY, + applied_at timestamptz NOT NULL DEFAULT now() + )`); err != nil { + return fmt.Errorf("create schema_migrations: %w", err) + } + entries, err := fs.ReadDir(migrationsFS, "migrations") + if err != nil { + return fmt.Errorf("read migrations: %w", err) + } + for _, e := range entries { + name := e.Name() + var applied bool + if err := pool.QueryRow(ctx, + `SELECT EXISTS (SELECT 1 FROM schema_migrations WHERE version = $1)`, name, + ).Scan(&applied); err != nil { + return fmt.Errorf("check migration %s: %w", name, err) + } + if applied { + continue + } + sql, err := migrationsFS.ReadFile("migrations/" + name) + if err != nil { + return fmt.Errorf("read migration %s: %w", name, err) + } + if err := applyMigration(ctx, pool, name, string(sql)); err != nil { + return err + } + } + return nil +} + +func applyMigration(ctx context.Context, pool *pgxpool.Pool, name, sql string) error { + tx, err := pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin migration %s: %w", name, err) + } + defer func() { _ = tx.Rollback(ctx) }() + // Exec without arguments uses the simple protocol, which allows the + // several statements a migration file contains. + if _, err := tx.Exec(ctx, sql); err != nil { + return fmt.Errorf("migration %s: %w", name, err) + } + if _, err := tx.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ($1)`, name); err != nil { + return fmt.Errorf("record migration %s: %w", name, err) + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit migration %s: %w", name, err) + } + return nil +} diff --git a/internal/master/store/migrations/0001_init.sql b/internal/master/store/migrations/0001_init.sql new file mode 100644 index 0000000..9acdea5 --- /dev/null +++ b/internal/master/store/migrations/0001_init.sql @@ -0,0 +1,48 @@ +CREATE TABLE master_identity ( + id int PRIMARY KEY CHECK (id = 1), + private_key bytea NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE TABLE agents ( + id bigserial PRIMARY KEY, + name text NOT NULL UNIQUE, + token_hash text NOT NULL UNIQUE, + hostname text NOT NULL DEFAULT '', + agent_version text NOT NULL DEFAULT '', + swarm_manager boolean NOT NULL DEFAULT false, + last_seen_at timestamptz, + revoked_at timestamptz, + created_at timestamptz NOT NULL DEFAULT now() +); + +CREATE TABLE services ( + id bigserial PRIMARY KEY, + kind text NOT NULL CHECK (kind IN ('compose', 'swarm', 'container')), + location_key text NOT NULL UNIQUE, + agent_id bigint REFERENCES agents (id), + compose_project text NOT NULL DEFAULT '', + compose_service text NOT NULL DEFAULT '', + working_dir text NOT NULL DEFAULT '', + swarm_cluster_id text NOT NULL DEFAULT '', + swarm_service text NOT NULL DEFAULT '', + container_name text NOT NULL DEFAULT '', + image_repo text NOT NULL, + actual_ref text NOT NULL, + actual_digest text NOT NULL DEFAULT '', + status text NOT NULL, + self boolean NOT NULL DEFAULT false, + update_mode text NOT NULL DEFAULT 'manual' CHECK (update_mode IN ('manual', 'auto')), + first_seen_at timestamptz NOT NULL, + updated_at timestamptz NOT NULL, + removed_at timestamptz +); + +CREATE TABLE sessions ( + id text PRIMARY KEY, + subject text NOT NULL, + name text NOT NULL DEFAULT '', + email text NOT NULL DEFAULT '', + csrf_token text NOT NULL, + expires_at timestamptz NOT NULL +); diff --git a/internal/master/store/store.go b/internal/master/store/store.go new file mode 100644 index 0000000..6ab2329 --- /dev/null +++ b/internal/master/store/store.go @@ -0,0 +1,41 @@ +// Package store persists the master's state in Postgres. +package store + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Store is the Postgres-backed storage for the master. +type Store struct { + pool *pgxpool.Pool +} + +// Open connects to Postgres and applies pending migrations. +func Open(ctx context.Context, url string) (*Store, error) { + cfg, err := pgxpool.ParseConfig(url) + if err != nil { + return nil, fmt.Errorf("parse DATABASE_URL: %w", err) + } + return openWithConfig(ctx, cfg) +} + +func openWithConfig(ctx context.Context, cfg *pgxpool.Config) (*Store, error) { + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + return nil, fmt.Errorf("connect postgres: %w", err) + } + if err := migrate(ctx, pool); err != nil { + pool.Close() + return nil, err + } + return &Store{pool: pool}, nil +} + +// Close releases the connection pool. +func (s *Store) Close() { s.pool.Close() } + +// Ping checks that Postgres answers. Used by /healthz. +func (s *Store) Ping(ctx context.Context) error { return s.pool.Ping(ctx) }