From 8a1fb45971db241b17beaa5e18cb707da5fdeeac Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Mon, 12 Oct 2026 14:45:46 +0200 Subject: [PATCH] feat(master): store service inventory snapshots and web sessions --- internal/master/store/services.go | 136 +++++++++++++++ .../master/store/services_integration_test.go | 162 ++++++++++++++++++ internal/master/store/sessions.go | 50 ++++++ .../master/store/sessions_integration_test.go | 34 ++++ 4 files changed, 382 insertions(+) create mode 100644 internal/master/store/services.go create mode 100644 internal/master/store/services_integration_test.go create mode 100644 internal/master/store/sessions.go create mode 100644 internal/master/store/sessions_integration_test.go diff --git a/internal/master/store/services.go b/internal/master/store/services.go new file mode 100644 index 0000000..1a6c6e5 --- /dev/null +++ b/internal/master/store/services.go @@ -0,0 +1,136 @@ +package store + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgconn" + + "github.com/shcizo/package-updater/internal/imageref" + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/protocol" +) + +type execer interface { + Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) +} + +const upsertServiceSQL = ` +INSERT INTO services ( + kind, location_key, agent_id, compose_project, compose_service, working_dir, + swarm_cluster_id, swarm_service, container_name, image_repo, actual_ref, actual_digest, + status, self, first_seen_at, updated_at) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $15) +ON CONFLICT (location_key) DO UPDATE SET + agent_id = EXCLUDED.agent_id, + working_dir = EXCLUDED.working_dir, + image_repo = EXCLUDED.image_repo, + actual_ref = EXCLUDED.actual_ref, + actual_digest = EXCLUDED.actual_digest, + status = EXCLUDED.status, + self = EXCLUDED.self, + updated_at = EXCLUDED.updated_at, + removed_at = NULL` + +func upsertService(ctx context.Context, db execer, agentID int64, clusterID string, svc protocol.Service, at time.Time) error { + clusterCol := "" + if svc.Kind == protocol.KindSwarm { + clusterCol = clusterID + } + _, err := db.Exec(ctx, upsertServiceSQL, + string(svc.Kind), model.LocationKey(agentID, clusterID, svc.Key()), agentID, + svc.ComposeProject, svc.ComposeService, svc.WorkingDir, + clusterCol, svc.SwarmService, svc.ContainerName, + imageref.Parse(svc.Image).Repo, svc.Image, svc.Digest, + svc.Status, svc.Self, at) + if err != nil { + return fmt.Errorf("upsert service %s: %w", svc.Key(), err) + } + return nil +} + +// ApplySnapshot makes the stored inventory for this agent's scope equal to +// services: everything listed is upserted, everything in scope but not +// listed is marked removed. Scope is the agent's own Compose services and +// containers, plus the cluster's Swarm services if the agent is a manager. +// A manager without a cluster ID is treated as a non-manager, since its +// Swarm rows could not be told apart from other agents' rows. +func (s *Store) ApplySnapshot(ctx context.Context, agentID int64, h protocol.Hello, services []protocol.Service, at time.Time) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin snapshot: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + isManager := h.SwarmManager && h.SwarmClusterID != "" + keys := make([]string, 0, len(services)) + for _, svc := range services { + if svc.Kind == protocol.KindSwarm && !isManager { + continue + } + if err := upsertService(ctx, tx, agentID, h.SwarmClusterID, svc, at); err != nil { + return err + } + keys = append(keys, model.LocationKey(agentID, h.SwarmClusterID, svc.Key())) + } + if _, err := tx.Exec(ctx, ` + UPDATE services SET removed_at = $1 + WHERE removed_at IS NULL + AND NOT (location_key = ANY($2)) + AND ((kind IN ('compose', 'container') AND agent_id = $3) + OR (kind = 'swarm' AND $4 AND swarm_cluster_id = $5))`, + at, keys, agentID, isManager, h.SwarmClusterID); err != nil { + return fmt.Errorf("mark removed services: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit snapshot: %w", err) + } + return nil +} + +// UpsertService stores one changed or new service. +func (s *Store) UpsertService(ctx context.Context, agentID int64, clusterID string, svc protocol.Service, at time.Time) error { + return upsertService(ctx, s.pool, agentID, clusterID, svc, at) +} + +// RemoveService marks the service with the agent-relative key as removed. +func (s *Store) RemoveService(ctx context.Context, agentID int64, clusterID, agentKey string, at time.Time) error { + if _, err := s.pool.Exec(ctx, + `UPDATE services SET removed_at = $2 WHERE location_key = $1 AND removed_at IS NULL`, + model.LocationKey(agentID, clusterID, agentKey), at); err != nil { + return fmt.Errorf("remove service: %w", err) + } + return nil +} + +// ListServices returns all services not marked removed: machine-bound +// services grouped by agent name first, Swarm services last. +func (s *Store) ListServices(ctx context.Context) ([]model.Service, error) { + rows, err := s.pool.Query(ctx, ` + SELECT s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.name, ''), + s.compose_project, s.compose_service, s.working_dir, s.swarm_cluster_id, + s.swarm_service, s.container_name, s.image_repo, s.actual_ref, s.actual_digest, + s.status, s.self, s.update_mode, s.first_seen_at, s.updated_at + FROM services s LEFT JOIN agents a ON a.id = s.agent_id + WHERE s.removed_at IS NULL + ORDER BY s.kind = 'swarm', COALESCE(a.name, ''), s.location_key`) + if err != nil { + return nil, fmt.Errorf("list services: %w", err) + } + defer rows.Close() + var out []model.Service + for rows.Next() { + var m model.Service + var kind string + if err := rows.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName, + &m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.SwarmClusterID, + &m.SwarmService, &m.ContainerName, &m.ImageRepo, &m.ActualRef, &m.ActualDigest, + &m.Status, &m.Self, &m.UpdateMode, &m.FirstSeenAt, &m.UpdatedAt); err != nil { + return nil, fmt.Errorf("scan service: %w", err) + } + m.Kind = protocol.Kind(kind) + out = append(out, m) + } + return out, rows.Err() +} diff --git a/internal/master/store/services_integration_test.go b/internal/master/store/services_integration_test.go new file mode 100644 index 0000000..87d8013 --- /dev/null +++ b/internal/master/store/services_integration_test.go @@ -0,0 +1,162 @@ +//go:build integration + +package store + +import ( + "context" + "strconv" + "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/protocol" +) + +var ( + t0 = time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) + slusk = protocol.Service{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", WorkingDir: "/srv/slusk", Image: "gitea.shcizo.se/shcizo/slusk:1.4.2", Digest: "sha256:d1", Status: "running"} + adhoc = protocol.Service{Kind: protocol.KindContainer, ContainerName: "adhoc", Image: "redis:alpine", Status: "stopped"} + gitea = protocol.Service{Kind: protocol.KindSwarm, SwarmService: "gitea_gitea", Image: "gitea/gitea:1.27.3@sha256:87a6", Digest: "sha256:87a6", Status: "running"} + worker = protocol.Hello{Hostname: "node-2"} + manager = protocol.Hello{Hostname: "swarm-mgr", SwarmManager: true, SwarmClusterID: "c1"} +) + +func names(svcs []model.Service) []string { + var out []string + for _, s := range svcs { + out = append(out, s.DisplayName()) + } + return out +} + +func newAgent(t *testing.T, s *Store, name string) int64 { + t.Helper() + a, err := s.CreateAgent(context.Background(), name, "hash-"+name) + require.NoError(t, err) + return a.ID +} + +func TestApplySnapshot_InsertsAsManualWithAgentName(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + id := newAgent(t, s, "node-2") + + require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk}, t0)) + + got, err := s.ListServices(ctx) + require.NoError(t, err) + require.Len(t, got, 1) + assert.Equal(t, "manual", got[0].UpdateMode) + assert.Equal(t, "node-2", got[0].AgentName) + assert.Equal(t, "gitea.shcizo.se/shcizo/slusk", got[0].ImageRepo) + assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:1.4.2", got[0].ActualRef) + assert.Equal(t, "sha256:d1", got[0].ActualDigest) + assert.Equal(t, "compose|"+itoa(id)+"|slusk|web", got[0].LocationKey) +} + +func TestApplySnapshot_RemovedWhileOfflineThenBack(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + id := newAgent(t, s, "node-2") + + require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk, adhoc}, t0)) + before, err := s.ListServices(ctx) + require.NoError(t, err) + + require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk}, t0.Add(time.Hour))) + got, err := s.ListServices(ctx) + require.NoError(t, err) + assert.Equal(t, []string{"slusk/web"}, names(got)) + + require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk, adhoc}, t0.Add(2*time.Hour))) + after, err := s.ListServices(ctx) + require.NoError(t, err) + require.Len(t, after, 2) + for i := range before { + assert.Equal(t, before[i].ID, after[i].ID, "a returning service keeps its row") + assert.True(t, before[i].FirstSeenAt.Equal(after[i].FirstSeenAt)) + } +} + +func TestApplySnapshot_TwoManagersShareSwarmRowsAndKeepOwnCompose(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + a := newAgent(t, s, "swarm-mgr") + b := newAgent(t, s, "swarm-mgr-2") + + require.NoError(t, s.ApplySnapshot(ctx, a, manager, []protocol.Service{gitea, slusk}, t0)) + require.NoError(t, s.ApplySnapshot(ctx, b, manager, []protocol.Service{gitea}, t0)) + + got, err := s.ListServices(ctx) + require.NoError(t, err) + assert.ElementsMatch(t, []string{"slusk/web", "gitea_gitea"}, names(got), + "one swarm row, and b's snapshot must not remove a's compose service") +} + +func TestApplySnapshot_IgnoresSwarmFromWorker(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + id := newAgent(t, s, "node-2") + require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{gitea}, t0)) + got, err := s.ListServices(ctx) + require.NoError(t, err) + assert.Empty(t, got) +} + +func TestApplySnapshot_IgnoresSwarmFromManagerWithoutClusterID(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + a := newAgent(t, s, "mgr-a") + b := newAgent(t, s, "mgr-b") + noCluster := protocol.Hello{Hostname: "mgr", SwarmManager: true} + + require.NoError(t, s.ApplySnapshot(ctx, a, manager, []protocol.Service{gitea}, t0)) + require.NoError(t, s.ApplySnapshot(ctx, b, noCluster, []protocol.Service{gitea}, t0)) + + got, err := s.ListServices(ctx) + require.NoError(t, err) + require.Len(t, got, 1, "the cluster-less manager adds no swarm row") + assert.Equal(t, "swarm|c1|gitea_gitea", got[0].LocationKey) + + require.NoError(t, s.ApplySnapshot(ctx, b, noCluster, nil, t0.Add(time.Hour))) + got, err = s.ListServices(ctx) + require.NoError(t, err) + assert.Len(t, got, 1, "the cluster-less manager removes no swarm rows") +} + +func TestUpsertAndRemoveService(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + id := newAgent(t, s, "swarm-mgr") + + require.NoError(t, s.UpsertService(ctx, id, "c1", gitea, t0)) + changed := gitea + changed.Status = "degraded" + require.NoError(t, s.UpsertService(ctx, id, "c1", changed, t0.Add(time.Minute))) + + got, err := s.ListServices(ctx) + require.NoError(t, err) + require.Len(t, got, 1) + assert.Equal(t, "degraded", got[0].Status) + assert.Equal(t, "swarm|c1|gitea_gitea", got[0].LocationKey) + + require.NoError(t, s.RemoveService(ctx, id, "c1", gitea.Key(), t0.Add(2*time.Minute))) + got, err = s.ListServices(ctx) + require.NoError(t, err) + assert.Empty(t, got) +} + +func TestListServices_SwarmLast(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + id := newAgent(t, s, "swarm-mgr") + require.NoError(t, s.ApplySnapshot(ctx, id, manager, []protocol.Service{gitea, slusk}, t0)) + got, err := s.ListServices(ctx) + require.NoError(t, err) + assert.Equal(t, []string{"slusk/web", "gitea_gitea"}, names(got)) +} + +func itoa(n int64) string { return strconv.FormatInt(n, 10) } diff --git a/internal/master/store/sessions.go b/internal/master/store/sessions.go new file mode 100644 index 0000000..90e7ef3 --- /dev/null +++ b/internal/master/store/sessions.go @@ -0,0 +1,50 @@ +package store + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + + "github.com/shcizo/package-updater/internal/master/model" +) + +// CreateSession stores a new session and deletes expired ones, so the table +// never grows without bound. +func (s *Store) CreateSession(ctx context.Context, sess model.Session) error { + if _, err := s.pool.Exec(ctx, `DELETE FROM sessions WHERE expires_at < now()`); err != nil { + return fmt.Errorf("delete expired sessions: %w", err) + } + if _, err := s.pool.Exec(ctx, + `INSERT INTO sessions (id, subject, name, email, csrf_token, expires_at) VALUES ($1, $2, $3, $4, $5, $6)`, + sess.ID, sess.Subject, sess.Name, sess.Email, sess.CSRFToken, sess.ExpiresAt); err != nil { + return fmt.Errorf("create session: %w", err) + } + return nil +} + +// GetSession returns the session unless it is missing or expired at now, +// in which case it returns model.ErrNotFound. +func (s *Store) GetSession(ctx context.Context, id string, now time.Time) (model.Session, error) { + var m model.Session + err := s.pool.QueryRow(ctx, + `SELECT id, subject, name, email, csrf_token, expires_at FROM sessions WHERE id = $1 AND expires_at > $2`, + id, now).Scan(&m.ID, &m.Subject, &m.Name, &m.Email, &m.CSRFToken, &m.ExpiresAt) + if errors.Is(err, pgx.ErrNoRows) { + return model.Session{}, model.ErrNotFound + } + if err != nil { + return model.Session{}, fmt.Errorf("get session: %w", err) + } + return m, nil +} + +// DeleteSession removes a session (logout). +func (s *Store) DeleteSession(ctx context.Context, id string) error { + if _, err := s.pool.Exec(ctx, `DELETE FROM sessions WHERE id = $1`, id); err != nil { + return fmt.Errorf("delete session: %w", err) + } + return nil +} diff --git a/internal/master/store/sessions_integration_test.go b/internal/master/store/sessions_integration_test.go new file mode 100644 index 0000000..b8d8a15 --- /dev/null +++ b/internal/master/store/sessions_integration_test.go @@ -0,0 +1,34 @@ +//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" +) + +func TestSessions(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) + sess := model.Session{ID: "s1", Subject: "sub", Name: "Samuel", Email: "s@example.com", CSRFToken: "c1", ExpiresAt: now.Add(time.Hour)} + + require.NoError(t, s.CreateSession(ctx, sess)) + got, err := s.GetSession(ctx, "s1", now) + require.NoError(t, err) + assert.Equal(t, "Samuel", got.Name) + assert.Equal(t, "c1", got.CSRFToken) + + _, err = s.GetSession(ctx, "s1", now.Add(2*time.Hour)) + assert.ErrorIs(t, err, model.ErrNotFound, "expired sessions are not returned") + + require.NoError(t, s.DeleteSession(ctx, "s1")) + _, err = s.GetSession(ctx, "s1", now) + assert.ErrorIs(t, err, model.ErrNotFound) +}