feat(master): store service inventory snapshots and web sessions

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()
+}

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) }

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
+}

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)
+}