From cfc6ea67136311fe6ab9d6a1709cf346ab5761cc Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Mon, 12 Oct 2026 20:40:04 +0200 Subject: [PATCH] feat(store): available ref, update mode and services-by-repo for the webhook (#8) --- internal/master/model/job.go | 2 +- internal/master/model/model.go | 34 +++- internal/master/store/jobs.go | 16 +- .../master/store/migrations/0003_webhook.sql | 6 + internal/master/store/services.go | 48 +++++- .../master/store/webhook_integration_test.go | 149 ++++++++++++++++++ 6 files changed, 240 insertions(+), 15 deletions(-) create mode 100644 internal/master/store/migrations/0003_webhook.sql create mode 100644 internal/master/store/webhook_integration_test.go diff --git a/internal/master/model/job.go b/internal/master/model/job.go index afd2106..5c1f8b9 100644 --- a/internal/master/model/job.go +++ b/internal/master/model/job.go @@ -46,7 +46,7 @@ const ( JobFailed JobStatus = "failed" ) -// JobTrigger says what created a job. Only TriggerManual is used in phase 2. +// JobTrigger says what created a job. TriggerManual comes from the UI, TriggerWebhook from POST /update. type JobTrigger string // Job triggers. diff --git a/internal/master/model/model.go b/internal/master/model/model.go index 086402d..b17e810 100644 --- a/internal/master/model/model.go +++ b/internal/master/model/model.go @@ -51,11 +51,35 @@ type Service struct { ActualDigest string Status string Self bool - Built bool // Compose service with a build: section; never updated - UpdateMode string - FirstSeenAt time.Time - UpdatedAt time.Time - RemovedAt *time.Time + Built bool // Compose service with a build: section; never updated + UpdateMode string // ModeManual or ModeAuto + // AvailableRef is the newest ref a webhook or poll reported for this + // service; "" if none, or if a later job already deployed it. + AvailableRef string + AvailableSource string // SourceWebhook or SourcePoll; "" when AvailableRef is empty + AvailableAt *time.Time + FirstSeenAt time.Time + UpdatedAt time.Time + RemovedAt *time.Time +} + +// Update modes of a service. +const ( + ModeManual = "manual" + ModeAuto = "auto" +) + +// Sources of an available ref. +const ( + SourceWebhook = "webhook" + SourcePoll = "poll" +) + +// Available is a newer ref reported for a service. +type Available struct { + Ref string + Source string // SourceWebhook or SourcePoll + At time.Time } // DisplayName is the human-facing name of the service. diff --git a/internal/master/store/jobs.go b/internal/master/store/jobs.go index a118c43..7933ed8 100644 --- a/internal/master/store/jobs.go +++ b/internal/master/store/jobs.go @@ -187,7 +187,9 @@ func (s *Store) JobProgress(ctx context.Context, agentID, jobID int64, chunk str // agentID: status, error, output (replacing what progress assembled), // started_at (the agent's, or the existing one if the agent sent none), // finished_at; to_ref = f.Ref when set. On success it also sets the -// service's desired_ref = f.Ref, in the same transaction. Returns false if no +// service's desired_ref = f.Ref and clears its available ref if that was +// reported no later than the job was created (the job then deployed the +// newest version), in the same transaction. Returns false if no // such active job belongs to agentID. func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.JobFinish) (bool, error) { tx, err := s.pool.Begin(ctx) @@ -203,14 +205,15 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job startedAt = &f.StartedAt } var serviceID int64 + var createdAt time.Time err = tx.QueryRow(ctx, ` UPDATE update_jobs SET status = $3, error = $4, output = $5, output_truncated = $6, to_ref = CASE WHEN $7::text <> '' THEN $7::text ELSE to_ref END, started_at = COALESCE($8, started_at), finished_at = $9 WHERE id = $1 AND agent_id = $2 AND status IN ('queued', 'running') - RETURNING service_id`, - jobID, agentID, string(f.Status), f.Error, f.Output, f.OutputTruncated, f.Ref, startedAt, f.FinishedAt).Scan(&serviceID) + RETURNING service_id, created_at`, + jobID, agentID, string(f.Status), f.Error, f.Output, f.OutputTruncated, f.Ref, startedAt, f.FinishedAt).Scan(&serviceID, &createdAt) if errors.Is(err, pgx.ErrNoRows) { return false, nil } @@ -218,7 +221,12 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job return false, fmt.Errorf("finish job: %w", err) } if f.Status == model.JobSucceeded { - if _, err := tx.Exec(ctx, `UPDATE services SET desired_ref = $2 WHERE id = $1`, serviceID, f.Ref); err != nil { + if _, err := tx.Exec(ctx, ` + UPDATE services SET desired_ref = $2, + available_ref = CASE WHEN available_at <= $3 THEN '' ELSE available_ref END, + available_source = CASE WHEN available_at <= $3 THEN '' ELSE available_source END, + available_at = CASE WHEN available_at <= $3 THEN NULL ELSE available_at END + WHERE id = $1`, serviceID, f.Ref, createdAt); err != nil { return false, fmt.Errorf("finish job: desired ref: %w", err) } } diff --git a/internal/master/store/migrations/0003_webhook.sql b/internal/master/store/migrations/0003_webhook.sql new file mode 100644 index 0000000..ed7114b --- /dev/null +++ b/internal/master/store/migrations/0003_webhook.sql @@ -0,0 +1,6 @@ +ALTER TABLE services ADD COLUMN available_ref text NOT NULL DEFAULT ''; +ALTER TABLE services ADD COLUMN available_source text NOT NULL DEFAULT '' + CHECK (available_source IN ('', 'webhook', 'poll')); +ALTER TABLE services ADD COLUMN available_at timestamptz; + +CREATE INDEX services_by_image_repo ON services (image_repo) WHERE removed_at IS NULL; diff --git a/internal/master/store/services.go b/internal/master/store/services.go index 816d2c5..a71f6d5 100644 --- a/internal/master/store/services.go +++ b/internal/master/store/services.go @@ -149,7 +149,8 @@ func (s *Store) RemoveService(ctx context.Context, agentID int64, clusterID, age const serviceColumns = `s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.name, ''), s.compose_project, s.compose_service, s.working_dir, s.config_files, s.built, s.swarm_cluster_id, s.swarm_service, s.container_name, s.image_repo, s.desired_ref, s.actual_ref, s.actual_digest, - s.status, s.self, s.update_mode, s.first_seen_at, s.updated_at, s.removed_at` + s.status, s.self, s.update_mode, s.available_ref, s.available_source, s.available_at, + s.first_seen_at, s.updated_at, s.removed_at` func scanService(row pgx.Row) (model.Service, error) { var m model.Service @@ -157,7 +158,8 @@ func scanService(row pgx.Row) (model.Service, error) { err := row.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName, &m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.ConfigFiles, &m.Built, &m.SwarmClusterID, &m.SwarmService, &m.ContainerName, &m.ImageRepo, &m.DesiredRef, &m.ActualRef, &m.ActualDigest, - &m.Status, &m.Self, &m.UpdateMode, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) + &m.Status, &m.Self, &m.UpdateMode, &m.AvailableRef, &m.AvailableSource, &m.AvailableAt, + &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) m.Kind = protocol.Kind(kind) return m, err } @@ -165,13 +167,23 @@ func scanService(row pgx.Row) (model.Service, error) { // 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) { + return s.queryServices(ctx, "list services", "", nil) +} + +// ServicesByRepo returns the services not marked removed whose image repo +// equals repo (case-sensitive), in the same order as ListServices. +func (s *Store) ServicesByRepo(ctx context.Context, repo string) ([]model.Service, error) { + return s.queryServices(ctx, "services by repo", "AND s.image_repo = $1", []any{repo}) +} + +func (s *Store) queryServices(ctx context.Context, what, extraWhere string, args []any) ([]model.Service, error) { rows, err := s.pool.Query(ctx, ` SELECT `+serviceColumns+` 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`) + WHERE s.removed_at IS NULL `+extraWhere+` + ORDER BY s.kind = 'swarm', COALESCE(a.name, ''), s.location_key`, args...) if err != nil { - return nil, fmt.Errorf("list services: %w", err) + return nil, fmt.Errorf("%s: %w", what, err) } defer rows.Close() var out []model.Service @@ -185,6 +197,32 @@ func (s *Store) ListServices(ctx context.Context) ([]model.Service, error) { return out, rows.Err() } +// SetAvailable records a newer ref for the service, replacing any earlier +// suggestion. +func (s *Store) SetAvailable(ctx context.Context, id int64, a model.Available) error { + if _, err := s.pool.Exec(ctx, ` + UPDATE services SET available_ref = $2, available_source = $3, available_at = $4 + WHERE id = $1`, id, a.Ref, a.Source, a.At); err != nil { + return fmt.Errorf("set available: %w", err) + } + return nil +} + +// SetUpdateMode sets the service's update mode (model.ModeManual or +// model.ModeAuto). Returns model.ErrNotFound if the service does not exist +// or is marked removed. +func (s *Store) SetUpdateMode(ctx context.Context, id int64, mode string) error { + tag, err := s.pool.Exec(ctx, + `UPDATE services SET update_mode = $2 WHERE id = $1 AND removed_at IS NULL`, id, mode) + if err != nil { + return fmt.Errorf("set update mode: %w", err) + } + if tag.RowsAffected() == 0 { + return model.ErrNotFound + } + return nil +} + // GetService returns one service, including a removed one (RemovedAt set). // Returns model.ErrNotFound if the ID does not exist. func (s *Store) GetService(ctx context.Context, id int64) (model.Service, error) { diff --git a/internal/master/store/webhook_integration_test.go b/internal/master/store/webhook_integration_test.go new file mode 100644 index 0000000..a8b4891 --- /dev/null +++ b/internal/master/store/webhook_integration_test.go @@ -0,0 +1,149 @@ +//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/protocol" +) + +func TestServicesByRepo(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + id := newAgent(t, s, "node-2") + other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "o", ComposeService: "o", Image: "gitea.shcizo.se/shcizo/Slusk:1", Status: "running"} + gone := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "g", ComposeService: "g", Image: "gitea.shcizo.se/shcizo/slusk:0.9", Status: "running"} + require.NoError(t, s.ApplySnapshot(ctx, id, worker, []protocol.Service{slusk, other, gone, adhoc}, t0)) + require.NoError(t, s.RemoveService(ctx, id, "", gone.Key(), t0)) + + got, err := s.ServicesByRepo(ctx, "gitea.shcizo.se/shcizo/slusk") + + require.NoError(t, err) + assert.Equal(t, []string{"slusk/web"}, names(got), "removed rows and rows of other (case-sensitive) repos are excluded") + assert.Equal(t, "node-2", got[0].AgentName) + + none, err := s.ServicesByRepo(ctx, "nope") + require.NoError(t, err) + assert.Empty(t, none) +} + +func TestSetAvailable_RoundTrip(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + _, svc := jobFixture(t, s, "a") + + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: "r:2", Source: model.SourceWebhook, At: t0})) + + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, "r:2", got.AvailableRef) + assert.Equal(t, model.SourceWebhook, got.AvailableSource) + require.NotNil(t, got.AvailableAt) + assert.True(t, t0.Equal(*got.AvailableAt)) +} + +func TestSetUpdateMode(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + + require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeAuto)) + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, model.ModeAuto, got.UpdateMode) + + require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeManual)) + got, _ = s.GetService(ctx, svc) + assert.Equal(t, model.ModeManual, got.UpdateMode) + + assert.ErrorIs(t, s.SetUpdateMode(ctx, 99999, model.ModeAuto), model.ErrNotFound) + + require.NoError(t, s.RemoveService(ctx, agent, "", slusk.Key(), t0)) + assert.ErrorIs(t, s.SetUpdateMode(ctx, svc, model.ModeAuto), model.ErrNotFound) +} + +func TestUpsertKeepsUpdateMode(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, model.ModeManual, got.UpdateMode, "a new service starts manual") + + require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeAuto)) + changed := slusk + changed.Image = "gitea.shcizo.se/shcizo/slusk:1.5.0" + require.NoError(t, s.UpsertService(ctx, agent, "", changed, t0.Add(time.Minute))) + require.NoError(t, s.ApplySnapshot(ctx, agent, worker, []protocol.Service{changed}, t0.Add(2*time.Minute))) + + got, err = s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, model.ModeAuto, got.UpdateMode) +} + +func TestFinishJob_ClearsAvailableOnlyWhenJobIsNewer(t *testing.T) { + succeed := model.JobFinish{Status: model.JobSucceeded, Ref: "r:2@sha256:d", StartedAt: t0, FinishedAt: t0} + tests := []struct { + name string + availableAt time.Time // job is created at t0 + finish model.JobFinish + wantCleared bool + }{ + {"suggestion before the job", t0.Add(-time.Minute), succeed, true}, + {"suggestion at the same instant", t0, succeed, true}, + {"suggestion after the job was created", t0.Add(time.Minute), succeed, false}, + {"failed job keeps the suggestion", t0.Add(-time.Minute), + model.JobFinish{Status: model.JobFailed, Error: "boom", StartedAt: t0, FinishedAt: t0}, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: "r:2", Source: model.SourceWebhook, At: tt.availableAt})) + j, err := s.CreateJob(ctx, newJob(svc, agent)) + require.NoError(t, err) + + ok, err := s.FinishJob(ctx, agent, j.ID, tt.finish) + require.NoError(t, err) + require.True(t, ok) + + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + if tt.wantCleared { + assert.Empty(t, got.AvailableRef) + assert.Empty(t, got.AvailableSource) + assert.Nil(t, got.AvailableAt) + } else { + assert.Equal(t, "r:2", got.AvailableRef) + assert.Equal(t, model.SourceWebhook, got.AvailableSource) + assert.NotNil(t, got.AvailableAt) + } + }) + } +} + +func TestFinishJob_WithoutAvailableLeavesItEmpty(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + j, err := s.CreateJob(ctx, newJob(svc, agent)) + require.NoError(t, err) + + ok, err := s.FinishJob(ctx, agent, j.ID, model.JobFinish{Status: model.JobSucceeded, Ref: "r:1@sha256:d", StartedAt: t0, FinishedAt: t0}) + require.NoError(t, err) + require.True(t, ok) + + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Empty(t, got.AvailableRef) + assert.Nil(t, got.AvailableAt) + assert.Equal(t, "r:1@sha256:d", got.DesiredRef) +}