diff --git a/server/internal/store/agents.go b/server/internal/store/agents.go new file mode 100644 index 0000000..22945f4 --- /dev/null +++ b/server/internal/store/agents.go @@ -0,0 +1,293 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package store + +import ( + "context" + "encoding/json" + "net/netip" + "time" + + "github.com/jackc/pgx/v5" +) + +// PKI key purposes (table pki_keys). +const ( + PKIAgentCA = "agent_ca" + PKIBundleSigning = "bundle_signing" +) + +// PKIKey is a row of pki_keys. PrivateKey is sealed with the master key. +type PKIKey struct { + Purpose string + Certificate []byte + PublicKey []byte + PrivateKey []byte + CreatedAt time.Time +} + +// GetPKIKey returns the key material for purpose. +func (t *Tx) GetPKIKey(ctx context.Context, purpose string) (*PKIKey, error) { + var k PKIKey + err := t.q.QueryRow(ctx, `SELECT purpose, certificate, public_key, private_key, created_at FROM pki_keys WHERE purpose = $1`, + purpose).Scan(&k.Purpose, &k.Certificate, &k.PublicKey, &k.PrivateKey, &k.CreatedAt) + if err != nil { + return nil, mapErr(err) + } + return &k, nil +} + +// InsertPKIKeyIfAbsent stores k unless a key for its purpose exists already +// (concurrent first starts: the first insert wins, everyone reads it back). +func (t *Tx) InsertPKIKeyIfAbsent(ctx context.Context, k *PKIKey) (inserted bool, err error) { + tag, err := t.q.Exec(ctx, `INSERT INTO pki_keys (purpose, certificate, public_key, private_key) VALUES ($1, $2, $3, $4) + ON CONFLICT (purpose) DO NOTHING`, k.Purpose, k.Certificate, k.PublicKey, k.PrivateKey) + if err != nil { + return false, mapErr(err) + } + return tag.RowsAffected() == 1, nil +} + +// Agent is a row of the agents table. +type Agent struct { + ID string + Role string + Hostname string + AgentVersion string + ServiceVersion string + Mode string + EnrolledAt time.Time + LastSeenAt *time.Time + RemoteAddress *netip.Addr + CertSerial string + CertNotAfter time.Time + AppliedRevision int64 + LastGoodRevision int64 + Health json.RawMessage + RevokedAt *time.Time + LeaseSequence int64 +} + +const agentColumns = `id, role, hostname, agent_version, service_version, mode, enrolled_at, last_seen_at, + remote_address, cert_serial, cert_not_after, applied_revision, last_good_revision, health, revoked_at, lease_sequence` + +func scanAgent(row pgx.Row) (*Agent, error) { + var a Agent + var health []byte + if err := row.Scan(&a.ID, &a.Role, &a.Hostname, &a.AgentVersion, &a.ServiceVersion, &a.Mode, &a.EnrolledAt, + &a.LastSeenAt, &a.RemoteAddress, &a.CertSerial, &a.CertNotAfter, &a.AppliedRevision, &a.LastGoodRevision, + &health, &a.RevokedAt, &a.LeaseSequence); err != nil { + return nil, mapErr(err) + } + a.Health = health + if a.RemoteAddress != nil { + ip := a.RemoteAddress.Unmap() + a.RemoteAddress = &ip + } + return &a, nil +} + +// InsertAgent creates an agent; the enrollment time is set. +func (t *Tx) InsertAgent(ctx context.Context, a *Agent) error { + row := t.q.QueryRow(ctx, ` + INSERT INTO agents (id, role, hostname, agent_version, remote_address, cert_serial, cert_not_after, last_seen_at) + VALUES ($1, $2::agent_role, $3, $4, $5, $6, $7, now()) + RETURNING `+agentColumns, + a.ID, a.Role, a.Hostname, a.AgentVersion, a.RemoteAddress, a.CertSerial, a.CertNotAfter) + got, err := scanAgent(row) + if err != nil { + return err + } + *a = *got + return nil +} + +// GetAgent returns one agent. +func (t *Tx) GetAgent(ctx context.Context, id string, forUpdate bool) (*Agent, error) { + sql := `SELECT ` + agentColumns + ` FROM agents WHERE id = $1` + if forUpdate { + sql += ` FOR UPDATE` + } + return scanAgent(t.q.QueryRow(ctx, sql, id)) +} + +// ListAgents returns all agents ordered by role and hostname. +func (t *Tx) ListAgents(ctx context.Context) ([]*Agent, error) { + rows, err := t.q.Query(ctx, `SELECT `+agentColumns+` FROM agents ORDER BY role, lower(hostname), id`) + if err != nil { + return nil, mapErr(err) + } + defer rows.Close() + var out []*Agent + for rows.Next() { + a, err := scanAgent(rows) + if err != nil { + return nil, err + } + out = append(out, a) + } + return out, mapErr(rows.Err()) +} + +// RevokeAgent marks the agent revoked (keeps an earlier revocation time). +func (t *Tx) RevokeAgent(ctx context.Context, id string, at time.Time) (*Agent, error) { + return scanAgent(t.q.QueryRow(ctx, `UPDATE agents SET revoked_at = coalesce(revoked_at, $2) WHERE id = $1 RETURNING `+agentColumns, id, at)) +} + +// DeleteAgent removes an agent with its deployments and leases. +func (t *Tx) DeleteAgent(ctx context.Context, id string) error { + tag, err := t.q.Exec(ctx, `DELETE FROM agents WHERE id = $1`, id) + if err != nil { + return mapErr(err) + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + +// UpdateAgentCertificate records the serial of a renewed certificate. Only +// non-revoked agents are updated. +func (t *Tx) UpdateAgentCertificate(ctx context.Context, id, serial string, notAfter time.Time) error { + tag, err := t.q.Exec(ctx, `UPDATE agents SET cert_serial = $2, cert_not_after = $3 WHERE id = $1 AND revoked_at IS NULL`, + id, serial, notAfter) + if err != nil { + return mapErr(err) + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + +// AgentStatus is reported by an agent. Nil fields keep the stored value. +type AgentStatus struct { + Hostname *string + AgentVersion *string + ServiceVersion *string + AppliedRevision *int64 + LastGoodRevision *int64 + Health json.RawMessage + RemoteAddress *netip.Addr + SeenAt time.Time +} + +// UpdateAgentStatus stores reported status and the last contact. It returns +// ErrNotFound for unknown and revoked agents. +func (t *Tx) UpdateAgentStatus(ctx context.Context, id string, s AgentStatus) (*Agent, error) { + return scanAgent(t.q.QueryRow(ctx, ` + UPDATE agents SET + hostname = coalesce($2, hostname), + agent_version = coalesce($3, agent_version), + service_version = coalesce($4, service_version), + applied_revision = coalesce($5, applied_revision), + last_good_revision = coalesce($6, last_good_revision), + health = coalesce($7::jsonb, health), + remote_address = coalesce($8, remote_address), + last_seen_at = $9 + WHERE id = $1 AND revoked_at IS NULL + RETURNING `+agentColumns, + id, s.Hostname, s.AgentVersion, s.ServiceVersion, s.AppliedRevision, s.LastGoodRevision, + nullJSON(s.Health), s.RemoteAddress, s.SeenAt)) +} + +// EnrollmentToken is a row of enrollment_tokens. +type EnrollmentToken struct { + ID string + SecretHash []byte + Role string + Description string + CreatedBy *string + CreatedByName string + CreatedAt time.Time + ExpiresAt time.Time + UsedAt *time.Time + AgentID *string +} + +const tokenColumns = //nolint:gosec // a column list, not a credential +`id, secret_hash, role, description, created_by, created_by_name, created_at, expires_at, used_at, agent_id` + +func scanToken(row pgx.Row) (*EnrollmentToken, error) { + var tk EnrollmentToken + if err := row.Scan(&tk.ID, &tk.SecretHash, &tk.Role, &tk.Description, &tk.CreatedBy, &tk.CreatedByName, + &tk.CreatedAt, &tk.ExpiresAt, &tk.UsedAt, &tk.AgentID); err != nil { + return nil, mapErr(err) + } + return &tk, nil +} + +// InsertEnrollmentToken stores a new token (CreatedAt is taken from tk). +func (t *Tx) InsertEnrollmentToken(ctx context.Context, tk *EnrollmentToken) error { + _, err := t.q.Exec(ctx, ` + INSERT INTO enrollment_tokens (id, secret_hash, role, description, created_by, created_by_name, created_at, expires_at) + VALUES ($1, $2, $3::agent_role, $4, $5, $6, $7, $8)`, + tk.ID, tk.SecretHash, tk.Role, tk.Description, tk.CreatedBy, tk.CreatedByName, tk.CreatedAt, tk.ExpiresAt) + return mapErr(err) +} + +// GetEnrollmentToken returns one token. +func (t *Tx) GetEnrollmentToken(ctx context.Context, id string, forUpdate bool) (*EnrollmentToken, error) { + sql := `SELECT ` + tokenColumns + ` FROM enrollment_tokens WHERE id = $1` + if forUpdate { + sql += ` FOR UPDATE` + } + return scanToken(t.q.QueryRow(ctx, sql, id)) +} + +// ListEnrollmentTokens returns tokens that were used or expired after since +// or are still unused and valid, newest first. +func (t *Tx) ListEnrollmentTokens(ctx context.Context, since time.Time) ([]*EnrollmentToken, error) { + rows, err := t.q.Query(ctx, `SELECT `+tokenColumns+` FROM enrollment_tokens + WHERE (used_at IS NULL AND expires_at > $1) OR used_at > $1 + ORDER BY created_at DESC, id`, since) + if err != nil { + return nil, mapErr(err) + } + defer rows.Close() + var out []*EnrollmentToken + for rows.Next() { + tk, err := scanToken(rows) + if err != nil { + return nil, err + } + out = append(out, tk) + } + return out, mapErr(rows.Err()) +} + +// DeleteEnrollmentToken removes a token. +func (t *Tx) DeleteEnrollmentToken(ctx context.Context, id string) error { + tag, err := t.q.Exec(ctx, `DELETE FROM enrollment_tokens WHERE id = $1`, id) + if err != nil { + return mapErr(err) + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + +// MarkEnrollmentTokenUsed records the enrollment. It fails with ErrNotFound +// if the token was used in the meantime. +func (t *Tx) MarkEnrollmentTokenUsed(ctx context.Context, id, agentID string, at time.Time) error { + tag, err := t.q.Exec(ctx, `UPDATE enrollment_tokens SET used_at = $3, agent_id = $2 WHERE id = $1 AND used_at IS NULL`, + id, agentID, at) + if err != nil { + return mapErr(err) + } + if tag.RowsAffected() == 0 { + return ErrNotFound + } + return nil +} + +// DeleteEnrollmentTokensBefore removes tokens that were used or expired +// before t0 (housekeeping). +func (t *Tx) DeleteEnrollmentTokensBefore(ctx context.Context, t0 time.Time) (int64, error) { + tag, err := t.q.Exec(ctx, `DELETE FROM enrollment_tokens WHERE coalesce(used_at, expires_at) < $1`, t0) + if err != nil { + return 0, mapErr(err) + } + return tag.RowsAffected(), nil +} diff --git a/server/internal/store/agents_test.go b/server/internal/store/agents_test.go new file mode 100644 index 0000000..a5c0043 --- /dev/null +++ b/server/internal/store/agents_test.go @@ -0,0 +1,286 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package store_test + +import ( + "context" + "encoding/json" + "errors" + "net/netip" + "testing" + "time" + + "git.beugsiud.de/beug.k/DDI/server/internal/store" + "git.beugsiud.de/beug.k/DDI/server/internal/store/storetest" +) + +const ( + agentA = "00000000-0000-4000-8000-00000000000a" + agentB = "00000000-0000-4000-8000-00000000000b" +) + +func newAgent(t *testing.T, st *store.Store, id, role, host, serial string) *store.Agent { + t.Helper() + ip := netip.MustParseAddr("192.0.2.7") + a := &store.Agent{ID: id, Role: role, Hostname: host, AgentVersion: "1.0", RemoteAddress: &ip, CertSerial: serial, + CertNotAfter: time.Now().Add(14 * 24 * time.Hour)} + if err := st.Q().InsertAgent(context.Background(), a); err != nil { + t.Fatalf("insert agent: %v", err) + } + return a +} + +func TestPKIKeysInsertOnce(t *testing.T) { + st := storetest.New(t) + ctx := context.Background() + k := &store.PKIKey{Purpose: store.PKIBundleSigning, PublicKey: []byte("pub1"), PrivateKey: []byte("sealed1")} + if ok, err := st.Q().InsertPKIKeyIfAbsent(ctx, k); err != nil || !ok { + t.Fatalf("first insert: %v %v", ok, err) + } + k2 := &store.PKIKey{Purpose: store.PKIBundleSigning, PublicKey: []byte("pub2"), PrivateKey: []byte("sealed2")} + if ok, err := st.Q().InsertPKIKeyIfAbsent(ctx, k2); err != nil || ok { + t.Fatalf("second insert: %v %v", ok, err) + } + got, err := st.Q().GetPKIKey(ctx, store.PKIBundleSigning) + if err != nil || string(got.PublicKey) != "pub1" { + t.Fatalf("get: %+v %v", got, err) + } + if _, err := st.Q().InsertPKIKeyIfAbsent(ctx, &store.PKIKey{Purpose: store.PKIAgentCA, PublicKey: []byte("x"), PrivateKey: []byte("y")}); err == nil { + t.Error("CA without certificate accepted") + } + if _, err := st.Q().GetPKIKey(ctx, store.PKIAgentCA); !errors.Is(err, store.ErrNotFound) { + t.Errorf("missing key: %v", err) + } +} + +func TestAgentsAndTokens(t *testing.T) { + st := storetest.New(t) + ctx := context.Background() + q := st.Q() + now := time.Now().UTC().Truncate(time.Microsecond) + + tk := &store.EnrollmentToken{ID: "0123456789abcdef", SecretHash: make([]byte, 32), Role: "dns", CreatedByName: "admin", + CreatedAt: now, ExpiresAt: now.Add(time.Hour)} + if err := q.InsertEnrollmentToken(ctx, tk); err != nil { + t.Fatal(err) + } + bad := *tk + bad.ID = "XYZ" + if err := q.InsertEnrollmentToken(ctx, &bad); !store.IsConstraint(err, "enrollment_tokens_id_format") { + t.Errorf("bad token id: %v", err) + } + + a := newAgent(t, st, agentA, "dns", "ns1", "0a") + if a.Mode != "auto_pull" || a.EnrolledAt.IsZero() || a.RemoteAddress == nil || a.RemoteAddress.String() != "192.0.2.7" { + t.Errorf("inserted agent %+v", a) + } + var h map[string]any + if err := json.Unmarshal(a.Health, &h); err != nil || h["state"] != "unknown" { + t.Errorf("default health %s", a.Health) + } + if err := st.Q().InsertAgent(ctx, &store.Agent{ID: agentB, Role: "dhcp", CertSerial: "0a", CertNotAfter: now}); !store.IsConstraint(err, "agents_cert_serial_unique") { + t.Errorf("duplicate serial: %v", err) + } + newAgent(t, st, agentB, "dhcp", "dhcp1", "0b") + + if err := q.MarkEnrollmentTokenUsed(ctx, tk.ID, agentA, now); err != nil { + t.Fatal(err) + } + if err := q.MarkEnrollmentTokenUsed(ctx, tk.ID, agentB, now); !errors.Is(err, store.ErrNotFound) { + t.Errorf("second use: %v", err) + } + list, err := q.ListEnrollmentTokens(ctx, now.Add(-time.Minute)) + if err != nil || len(list) != 1 || list[0].AgentID == nil || *list[0].AgentID != agentA { + t.Fatalf("tokens %+v %v", list, err) + } + if list, _ := q.ListEnrollmentTokens(ctx, now.Add(time.Minute)); len(list) != 0 { + t.Errorf("old used token listed: %+v", list) + } + + host := "ns1.example.test" + applied := int64(3) + got, err := q.UpdateAgentStatus(ctx, agentA, store.AgentStatus{Hostname: &host, AppliedRevision: &applied, + Health: json.RawMessage(`{"state":"ok","message":"","checks":[]}`), SeenAt: now}) + if err != nil || got.Hostname != host || got.AppliedRevision != 3 || got.AgentVersion != "1.0" || got.LastSeenAt == nil { + t.Fatalf("status update %+v %v", got, err) + } + if err := q.UpdateAgentCertificate(ctx, agentA, "ff", now.Add(time.Hour)); err != nil { + t.Fatal(err) + } + if _, err := q.RevokeAgent(ctx, agentA, now); err != nil { + t.Fatal(err) + } + if _, err := q.UpdateAgentStatus(ctx, agentA, store.AgentStatus{SeenAt: now}); !errors.Is(err, store.ErrNotFound) { + t.Errorf("status of revoked agent: %v", err) + } + if err := q.UpdateAgentCertificate(ctx, agentA, "fe", now); !errors.Is(err, store.ErrNotFound) { + t.Errorf("renewal of revoked agent: %v", err) + } + agents, err := q.ListAgents(ctx) + if err != nil || len(agents) != 2 || agents[0].ID != agentA || agents[0].RevokedAt == nil || agents[0].CertSerial != "ff" { + t.Fatalf("agents %+v %v", agents, err) + } + + if err := q.DeleteAgent(ctx, agentA); err != nil { + t.Fatal(err) + } + if got, _ := q.GetEnrollmentToken(ctx, tk.ID, false); got == nil || got.AgentID != nil || got.UsedAt == nil { + t.Errorf("token after agent deletion %+v", got) + } + if err := q.DeleteEnrollmentToken(ctx, tk.ID); err != nil { + t.Fatal(err) + } + if err := q.DeleteEnrollmentToken(ctx, tk.ID); !errors.Is(err, store.ErrNotFound) { + t.Errorf("second delete: %v", err) + } +} + +func TestRevisionsAndDeployments(t *testing.T) { + st := storetest.New(t) + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Microsecond) + + if _, err := st.Q().LatestRevision(ctx); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("latest of none: %v", err) + } + files := []store.RevisionFile{ + {Role: "dns", Path: "ddi-suite.conf", Mode: 0o644, Content: []byte("conf")}, + {Role: "dns", Path: "zones/db.a.test", Mode: 0o644, CreateOnly: true, Content: []byte("zone")}, + {Role: "dhcp", Path: "kea-dhcp4.conf", Mode: 0o640, Content: []byte("{}")}, + } + err := st.InTx(ctx, func(tx *store.Tx) error { + if err := tx.LockRevisions(ctx); err != nil { + return err + } + return tx.InsertRevision(ctx, &store.Revision{Number: 1, CreatedAt: now, CreatedByName: "admin", + Warnings: json.RawMessage(`[{"code":"w"}]`)}, files) + }) + if err != nil { + t.Fatal(err) + } + err = st.InTx(ctx, func(tx *store.Tx) error { + return tx.InsertRevision(ctx, &store.Revision{Number: 2, CreatedAt: now, CreatedByName: "admin"}, + []store.RevisionFile{{Role: "dns", Path: "../x", Content: []byte("x")}}) + }) + if !store.IsConstraint(err, "config_revision_files_path") { + t.Errorf("bad path: %v", err) + } + if err := st.Q().InsertRevision(ctx, &store.Revision{Number: 1 << 32, CreatedAt: now, CreatedByName: "admin"}, nil); !store.IsConstraint(err, "config_revisions_number") { + t.Errorf("number out of range: %v", err) + } + if _, err := st.Pool().Exec(ctx, `UPDATE config_revisions SET comment = 'x'`); err == nil { + t.Error("revision is mutable") + } + if _, err := st.Pool().Exec(ctx, `UPDATE config_revision_files SET content = 'x'`); err == nil { + t.Error("revision file is mutable") + } + got, err := st.Q().RevisionFiles(ctx, 1, "dns") + if err != nil || len(got) != 2 || got[1].Path != "zones/db.a.test" || !got[1].CreateOnly || got[0].Mode != 0o644 { + t.Fatalf("dns files %+v %v", got, err) + } + if all, _ := st.Q().RevisionFiles(ctx, 1, ""); len(all) != 3 || all[2].Role != "dhcp" { + t.Errorf("all files %+v", all) + } + revs, err := st.Q().ListRevisions(ctx, 10, 0) + if err != nil || len(revs) != 1 || revs[0].DNSFiles != 2 || revs[0].DHCPFiles != 1 || revs[0].WarningCount != 1 { + t.Fatalf("list %+v %v", revs, err) + } + if revs, _ := st.Q().ListRevisions(ctx, 10, 1); len(revs) != 0 { + t.Errorf("before=1: %+v", revs) + } + + newAgent(t, st, agentA, "dns", "ns1", "01") + for i, outcome := range []string{"rejected", "succeeded"} { + d := &store.Deployment{AgentID: agentA, Revision: 1, Trigger: "pull", Outcome: outcome, ActiveRevision: int64(i), + StartedAt: now, FinishedAt: now.Add(time.Second)} + if err := st.Q().InsertDeployment(ctx, d); err != nil || d.ID == 0 { + t.Fatalf("insert deployment: %v", err) + } + } + bad := &store.Deployment{AgentID: agentA, Revision: 1, Trigger: "pull", Outcome: "succeeded", StartedAt: now, FinishedAt: now.Add(-time.Second)} + if err := st.Q().InsertDeployment(ctx, bad); !store.IsConstraint(err, "deployments_times") { + t.Errorf("finished before started: %v", err) + } + ds, err := st.Q().ListDeployments(ctx, store.DeploymentFilter{AgentID: agentA, Limit: 1}) + if err != nil || len(ds) != 1 || ds[0].Outcome != "succeeded" || ds[0].AgentHostname != "ns1" { + t.Fatalf("deployments %+v %v", ds, err) + } + if older, _ := st.Q().ListDeployments(ctx, store.DeploymentFilter{BeforeID: ds[0].ID}); len(older) != 1 || older[0].Outcome != "rejected" { + t.Errorf("older %+v", older) + } + latest, err := st.Q().LatestDeployments(ctx) + if err != nil || latest[agentA] == nil || latest[agentA].ID != ds[0].ID { + t.Errorf("latest %+v %v", latest, err) + } + if err := st.Q().DeleteAgent(ctx, agentA); err != nil { + t.Fatal(err) + } + if ds, _ := st.Q().ListDeployments(ctx, store.DeploymentFilter{}); len(ds) != 0 { + t.Errorf("deployments survived agent deletion: %+v", ds) + } +} + +func TestLeases(t *testing.T) { + st := storetest.New(t) + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Microsecond) + newAgent(t, st, agentA, "dhcp", "dhcp1", "01") + newAgent(t, st, agentB, "dhcp", "dhcp2", "02") + var netID string + var subnet int32 + if err := st.Pool().QueryRow(ctx, `INSERT INTO networks (cidr, name) VALUES ('192.0.2.0/24', 'Office') RETURNING id, kea_subnet_id`).Scan(&netID, &subnet); err != nil { + t.Fatal(err) + } + lease := func(ip, mac, host, state string) store.Lease { + return store.Lease{IP: netip.MustParseAddr(ip), MAC: mac, Hostname: host, State: state, ExpiresAt: &now, SubnetID: uint32(subnet)} + } + store1 := func(agent string, seq int64, replace bool, ls ...store.Lease) { + t.Helper() + err := st.InTx(ctx, func(tx *store.Tx) error { + got, role, err := tx.LockLeaseSequence(ctx, agent) + if err != nil { + return err + } + if role != "dhcp" || got >= seq { + t.Fatalf("sequence %d role %s", got, role) + } + return tx.StoreLeases(ctx, agent, seq, replace, ls, now) + }) + if err != nil { + t.Fatal(err) + } + } + store1(agentA, 1, true, lease("192.0.2.10", "00:11:22:33:44:55", "pc1", "active"), lease("192.0.2.11", "00:11:22:33:44:56", "pc2", "active")) + store1(agentA, 2, false, lease("192.0.2.11", "00:11:22:33:44:56", "pc2", "released"), lease("192.0.2.12", "", "", "declined")) + store1(agentB, 1, true, lease("192.0.2.10", "00:11:22:33:44:55", "pc1", "active")) + + all, more, err := st.Q().ListLeases(ctx, store.LeaseFilter{}) + if err != nil || more || len(all) != 4 || all[0].NetworkID == nil || *all[0].NetworkID != netID { + t.Fatalf("all leases %+v %v %v", all, more, err) + } + // The page holds two IPs: both agents' leases of .10 and the one of .11. + page, more, err := st.Q().ListLeases(ctx, store.LeaseFilter{Limit: 2}) + if err != nil || !more || len(page) != 3 || page[2].IP.String() != "192.0.2.11" || page[2].State != "released" { + t.Fatalf("page %+v %v %v", page, more, err) + } + after := page[2].IP + rest, more, _ := st.Q().ListLeases(ctx, store.LeaseFilter{Limit: 2, After: &after}) + if more || len(rest) != 1 || rest[0].IP.String() != "192.0.2.12" { + t.Errorf("rest %+v %v", rest, more) + } + if ls, _, _ := st.Q().ListLeases(ctx, store.LeaseFilter{State: "active", Query: "pc1"}); len(ls) != 2 { + t.Errorf("filtered %+v", ls) + } + if ls, _, _ := st.Q().ListLeases(ctx, store.LeaseFilter{NetworkID: netID, Query: "44:56"}); len(ls) != 1 { + t.Errorf("network+mac %+v", ls) + } + if ls, _, _ := st.Q().ListLeases(ctx, store.LeaseFilter{NetworkID: "00000000-0000-4000-8000-000000000000"}); len(ls) != 0 { + t.Errorf("other network %+v", ls) + } + + // A full sync replaces the agent's set. + store1(agentA, 3, true, lease("192.0.2.20", "00:11:22:33:44:57", "pc3", "active")) + if ls, _, _ := st.Q().ListLeases(ctx, store.LeaseFilter{}); len(ls) != 2 { + t.Errorf("after full sync %+v", ls) + } +} diff --git a/server/internal/store/leases.go b/server/internal/store/leases.go new file mode 100644 index 0000000..d080f0a --- /dev/null +++ b/server/internal/store/leases.go @@ -0,0 +1,141 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package store + +import ( + "context" + "fmt" + "net/netip" + "strings" + "time" + + "github.com/jackc/pgx/v5" +) + +// MaxLeasePage bounds a page of ListLeases. +const MaxLeasePage = 1000 + +// Lease is a DHCP lease reported by an agent. NetworkID is resolved through +// the Kea subnet ID by ListLeases. +type Lease struct { + AgentID string + IP netip.Addr + MAC string + Hostname string + State string + ExpiresAt *time.Time + UpdatedAt *time.Time + SubnetID uint32 + ClientID string + ReportedAt time.Time + NetworkID *string +} + +// LockLeaseSequence locks the agent row and returns its lease sequence and +// role. Use it inside a transaction before StoreLeases. +func (t *Tx) LockLeaseSequence(ctx context.Context, agentID string) (seq int64, role string, err error) { + err = t.q.QueryRow(ctx, `SELECT lease_sequence, role FROM agents WHERE id = $1 AND revoked_at IS NULL FOR UPDATE`, + agentID).Scan(&seq, &role) + return seq, role, mapErr(err) +} + +// StoreLeases writes a lease batch of an agent and records its sequence. +// With replace, the batch is the complete lease set of the agent. +func (t *Tx) StoreLeases(ctx context.Context, agentID string, seq int64, replace bool, leases []Lease, at time.Time) error { + if replace { + if _, err := t.q.Exec(ctx, `DELETE FROM dhcp_leases WHERE agent_id = $1`, agentID); err != nil { + return mapErr(err) + } + } + b := &pgx.Batch{} + for _, l := range leases { + b.Queue(` + INSERT INTO dhcp_leases (agent_id, ip, mac, hostname, state, expires_at, updated_at, subnet_id, client_id, reported_at) + VALUES ($1, $2, $3, $4, $5::lease_state, $6, $7, $8, $9, $10) + ON CONFLICT (agent_id, ip) DO UPDATE SET mac = EXCLUDED.mac, hostname = EXCLUDED.hostname, + state = EXCLUDED.state, expires_at = EXCLUDED.expires_at, updated_at = EXCLUDED.updated_at, + subnet_id = EXCLUDED.subnet_id, client_id = EXCLUDED.client_id, reported_at = EXCLUDED.reported_at`, + agentID, l.IP, l.MAC, l.Hostname, l.State, l.ExpiresAt, l.UpdatedAt, int64(l.SubnetID), l.ClientID, at) + } + if err := t.execBatch(ctx, b); err != nil { + return err + } + _, err := t.q.Exec(ctx, `UPDATE agents SET lease_sequence = $2 WHERE id = $1`, agentID, seq) + return mapErr(err) +} + +// LeaseFilter selects leases. Zero values mean "no restriction". +type LeaseFilter struct { + NetworkID string + State string + Query string // substring of IP, MAC or hostname + After *netip.Addr + Limit int +} + +// ListLeases returns leases ordered by IP and agent. Limit counts distinct +// IPs, so leases of the same IP from several agents are never split across +// pages; more reports whether further IPs follow. +func (t *Tx) ListLeases(ctx context.Context, f LeaseFilter) (out []*Lease, more bool, err error) { + var where []string + var args []any + add := func(cond string, v any) { + args = append(args, v) + where = append(where, fmt.Sprintf(cond, len(args))) + } + if f.NetworkID != "" { + add("n.id = $%d::uuid", f.NetworkID) + } + if f.State != "" { + add("l.state = $%d::lease_state", f.State) + } + if f.Query != "" { + args = append(args, "%"+escapeLike(f.Query)+"%") + n := len(args) + where = append(where, fmt.Sprintf("(host(l.ip) ILIKE $%[1]d OR l.mac ILIKE $%[1]d OR l.hostname ILIKE $%[1]d)", n)) + } + if f.After != nil { + add("l.ip > $%d", *f.After) + } + limit := f.Limit + if limit <= 0 || limit > MaxLeasePage { + limit = 100 + } + from := ` FROM dhcp_leases l LEFT JOIN networks n ON n.kea_subnet_id = l.subnet_id AND l.subnet_id > 0` + cond := "" + if len(where) > 0 { + cond = " WHERE " + strings.Join(where, " AND ") + } + // The page holds the first limit+1 distinct IPs; the extra one only + // tells whether there is a next page. + sql := `WITH page AS (SELECT DISTINCT l.ip` + from + cond + fmt.Sprintf(` ORDER BY l.ip LIMIT %d)`, limit+1) + ` + SELECT l.agent_id, l.ip, l.mac, l.hostname, l.state, l.expires_at, l.updated_at, l.subnet_id, l.client_id, + l.reported_at, n.id` + from + ` JOIN page p ON p.ip = l.ip` + cond + ` ORDER BY l.ip, l.agent_id` + rows, err := t.q.Query(ctx, sql, args...) + if err != nil { + return nil, false, mapErr(err) + } + defer rows.Close() + ips := 0 + var last netip.Addr + for rows.Next() { + var l Lease + var subnet int64 + if err := rows.Scan(&l.AgentID, &l.IP, &l.MAC, &l.Hostname, &l.State, &l.ExpiresAt, &l.UpdatedAt, &subnet, + &l.ClientID, &l.ReportedAt, &l.NetworkID); err != nil { + return nil, false, mapErr(err) + } + l.IP = l.IP.Unmap() + l.SubnetID = uint32(subnet) //nolint:gosec // 0..2^32-1 by constraint + if ips == 0 || l.IP != last { + ips++ + last = l.IP + } + if ips > limit { + more = true + continue + } + out = append(out, &l) + } + return out, more, mapErr(rows.Err()) +} diff --git a/server/internal/store/migrations/00003_agents.sql b/server/internal/store/migrations/00003_agents.sql new file mode 100644 index 0000000..a505239 --- /dev/null +++ b/server/internal/store/migrations/00003_agents.sql @@ -0,0 +1,172 @@ +-- SPDX-License-Identifier: LicenseRef-DDI-Proprietary +-- Agents (M3, ADR 0021): PKI keys, enrollment tokens, agents, immutable +-- configuration revisions, deployment reports and DHCP leases. + +-- +goose Up + +-- Key material of the agent PKI: the agent CA (ADR 0008) and the bundle +-- signing key (ADR 0009). Private keys are sealed with the master key; the +-- associated data binds them to their row (ADR 0015). Rows are created once +-- (INSERT ... ON CONFLICT DO NOTHING, so concurrent first starts agree). +CREATE TABLE pki_keys ( + purpose text PRIMARY KEY, + certificate bytea, + public_key bytea NOT NULL, + private_key bytea NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + CONSTRAINT pki_keys_purpose CHECK (purpose IN ('agent_ca', 'bundle_signing')), + CONSTRAINT pki_keys_ca_certificate CHECK (purpose <> 'agent_ca' OR certificate IS NOT NULL) +); + +CREATE TYPE agent_role AS ENUM ('dns', 'dhcp'); +CREATE TYPE agent_mode AS ENUM ('auto_pull', 'manual'); + +CREATE TABLE agents ( + id uuid PRIMARY KEY, + role agent_role NOT NULL, + hostname text NOT NULL DEFAULT '', + agent_version text NOT NULL DEFAULT '', + service_version text NOT NULL DEFAULT '', + mode agent_mode NOT NULL DEFAULT 'auto_pull', + enrolled_at timestamptz NOT NULL DEFAULT now(), + last_seen_at timestamptz, + remote_address inet, + -- Serial number (lowercase hex) of the only certificate the agent may use. + cert_serial text NOT NULL, + cert_not_after timestamptz NOT NULL, + applied_revision bigint NOT NULL DEFAULT 0, + last_good_revision bigint NOT NULL DEFAULT 0, + health jsonb NOT NULL DEFAULT '{"state": "unknown", "message": "", "checks": []}', + revoked_at timestamptz, + -- Highest lease batch sequence processed (ReportLeases). + lease_sequence bigint NOT NULL DEFAULT 0, + CONSTRAINT agents_cert_serial CHECK (cert_serial ~ '^[0-9a-f]{1,40}$'), + CONSTRAINT agents_revisions CHECK (applied_revision BETWEEN 0 AND 4294967295 + AND last_good_revision BETWEEN 0 AND 4294967295), + CONSTRAINT agents_lease_sequence CHECK (lease_sequence >= 0), + CONSTRAINT agents_health_object CHECK (jsonb_typeof(health) = 'object') +); +CREATE UNIQUE INDEX agents_cert_serial_unique ON agents (cert_serial); + +CREATE TABLE enrollment_tokens ( + id text PRIMARY KEY, + -- SHA-256 of the secret part; the secret has 256 bits of entropy. + secret_hash bytea NOT NULL, + role agent_role NOT NULL, + description text NOT NULL DEFAULT '', + created_by uuid REFERENCES users (id) ON DELETE SET NULL, + created_by_name text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + expires_at timestamptz NOT NULL, + used_at timestamptz, + agent_id uuid REFERENCES agents (id) ON DELETE SET NULL, + CONSTRAINT enrollment_tokens_id_format CHECK (id ~ '^[0-9a-f]{16}$'), + CONSTRAINT enrollment_tokens_secret_hash CHECK (length(secret_hash) = 32), + CONSTRAINT enrollment_tokens_expiry CHECK (expires_at > created_at), + CONSTRAINT enrollment_tokens_agent_used CHECK (agent_id IS NULL OR used_at IS NOT NULL) +); +CREATE INDEX enrollment_tokens_created ON enrollment_tokens (created_at DESC); + +-- A revision is immutable; its number is also the SOA serial (uint32). +CREATE TABLE config_revisions ( + number bigint PRIMARY KEY, + created_at timestamptz NOT NULL DEFAULT now(), + created_by uuid REFERENCES users (id) ON DELETE SET NULL, + created_by_name text NOT NULL, + comment text NOT NULL DEFAULT '', + warnings jsonb NOT NULL DEFAULT '[]', + CONSTRAINT config_revisions_number CHECK (number BETWEEN 1 AND 4294967295), + CONSTRAINT config_revisions_warnings_array CHECK (jsonb_typeof(warnings) = 'array') +); + +CREATE TABLE config_revision_files ( + revision bigint NOT NULL REFERENCES config_revisions (number) ON DELETE CASCADE, + role agent_role NOT NULL, + -- Position in the generator output (bundles list files in this order). + ordinal integer NOT NULL, + path text NOT NULL, + mode integer NOT NULL, + create_only boolean NOT NULL DEFAULT false, + content bytea NOT NULL, + PRIMARY KEY (revision, role, ordinal), + CONSTRAINT config_revision_files_path_unique UNIQUE (revision, role, path), + CONSTRAINT config_revision_files_ordinal CHECK (ordinal >= 0), + CONSTRAINT config_revision_files_mode CHECK (mode BETWEEN 0 AND 511), + CONSTRAINT config_revision_files_path CHECK (path <> '' AND length(path) <= 255 + AND left(path, 1) <> '/' AND path !~ '(^|/)\.') +); + +-- Revisions and their files never change once written. +-- +goose StatementBegin +CREATE FUNCTION config_revisions_immutable() RETURNS trigger +LANGUAGE plpgsql AS $$ +BEGIN + RAISE EXCEPTION 'configuration revisions are immutable' + USING ERRCODE = 'check_violation', CONSTRAINT = 'config_revisions_immutable'; +END; +$$; +-- +goose StatementEnd +CREATE TRIGGER config_revisions_immutable + BEFORE UPDATE ON config_revisions + FOR EACH ROW EXECUTE FUNCTION config_revisions_immutable(); +CREATE TRIGGER config_revision_files_immutable + BEFORE UPDATE ON config_revision_files + FOR EACH ROW EXECUTE FUNCTION config_revisions_immutable(); + +CREATE TYPE deployment_trigger AS ENUM ('pull', 'push'); +CREATE TYPE deployment_outcome AS ENUM ('succeeded', 'rejected', 'rolled_back', 'failed'); + +CREATE TABLE deployments ( + id bigserial PRIMARY KEY, + agent_id uuid NOT NULL REFERENCES agents (id) ON DELETE CASCADE, + -- deployment_id of the ApplyRevisionCommand, empty for pulls. + correlation_id text NOT NULL DEFAULT '', + revision bigint NOT NULL, + trigger_kind deployment_trigger NOT NULL, + outcome deployment_outcome NOT NULL, + message text NOT NULL DEFAULT '', + active_revision bigint NOT NULL DEFAULT 0, + started_at timestamptz NOT NULL, + finished_at timestamptz NOT NULL, + reported_at timestamptz NOT NULL DEFAULT now(), + CONSTRAINT deployments_revision CHECK (revision BETWEEN 1 AND 4294967295), + CONSTRAINT deployments_active_revision CHECK (active_revision BETWEEN 0 AND 4294967295), + CONSTRAINT deployments_times CHECK (finished_at >= started_at) +); +CREATE INDEX deployments_agent ON deployments (agent_id, id DESC); +CREATE INDEX deployments_revision ON deployments (revision, id DESC); + +-- Leases as last reported by each DHCP agent. +CREATE TABLE dhcp_leases ( + agent_id uuid NOT NULL REFERENCES agents (id) ON DELETE CASCADE, + ip inet NOT NULL, + mac text NOT NULL DEFAULT '', + hostname text NOT NULL DEFAULT '', + state lease_state NOT NULL, + expires_at timestamptz, + updated_at timestamptz, + subnet_id bigint NOT NULL DEFAULT 0, + client_id text NOT NULL DEFAULT '', + reported_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (agent_id, ip), + CONSTRAINT dhcp_leases_host_ip CHECK (masklen(ip) = CASE family(ip) WHEN 4 THEN 32 ELSE 128 END), + CONSTRAINT dhcp_leases_subnet_id CHECK (subnet_id BETWEEN 0 AND 4294967295) +); +CREATE INDEX dhcp_leases_ip ON dhcp_leases (ip); +CREATE INDEX dhcp_leases_subnet ON dhcp_leases (subnet_id); + +-- +goose Down +DROP TABLE dhcp_leases; +DROP TABLE deployments; +DROP TYPE deployment_outcome; +DROP TYPE deployment_trigger; +DROP TRIGGER config_revision_files_immutable ON config_revision_files; +DROP TRIGGER config_revisions_immutable ON config_revisions; +DROP FUNCTION config_revisions_immutable(); +DROP TABLE config_revision_files; +DROP TABLE config_revisions; +DROP TABLE enrollment_tokens; +DROP TABLE agents; +DROP TYPE agent_mode; +DROP TYPE agent_role; +DROP TABLE pki_keys; diff --git a/server/internal/store/revisions.go b/server/internal/store/revisions.go new file mode 100644 index 0000000..d60b22f --- /dev/null +++ b/server/internal/store/revisions.go @@ -0,0 +1,266 @@ +// SPDX-License-Identifier: LicenseRef-DDI-Proprietary + +package store + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "github.com/jackc/pgx/v5" +) + +// MaxRevision is the highest revision number (it doubles as SOA serial). +const MaxRevision = 1<<32 - 1 + +// revisionLockKey serializes revision creation (pg_advisory_xact_lock). +const revisionLockKey = 0x64646901 + +// Revision is a row of config_revisions. DNSFiles, DHCPFiles and +// WarningCount are filled by ListRevisions. +type Revision struct { + Number int64 + CreatedAt time.Time + CreatedBy *string + CreatedByName string + Comment string + Warnings json.RawMessage + DNSFiles int + DHCPFiles int + WarningCount int +} + +// RevisionFile is one artifact of a revision. +type RevisionFile struct { + Role string + Path string + Mode uint32 + CreateOnly bool + Content []byte +} + +// LockRevisions takes a transaction-scoped lock that serializes revision +// creation. It must run inside a transaction. +func (t *Tx) LockRevisions(ctx context.Context) error { + _, err := t.q.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, int64(revisionLockKey)) + return mapErr(err) +} + +// LatestRevision returns the newest revision or ErrNotFound. +func (t *Tx) LatestRevision(ctx context.Context) (*Revision, error) { + return t.getRevision(ctx, `ORDER BY number DESC LIMIT 1`) +} + +// GetRevision returns one revision without files. +func (t *Tx) GetRevision(ctx context.Context, number int64) (*Revision, error) { + return t.getRevision(ctx, `WHERE number = $1`, number) +} + +func (t *Tx) getRevision(ctx context.Context, cond string, args ...any) (*Revision, error) { + var r Revision + var warnings []byte + err := t.q.QueryRow(ctx, `SELECT number, created_at, created_by, created_by_name, comment, warnings + FROM config_revisions `+cond, args...).Scan(&r.Number, &r.CreatedAt, &r.CreatedBy, &r.CreatedByName, &r.Comment, &warnings) + if err != nil { + return nil, mapErr(err) + } + r.Warnings = warnings + return &r, nil +} + +// InsertRevision stores a revision with its files (in the given order per +// role). CreatedAt is taken from r. +func (t *Tx) InsertRevision(ctx context.Context, r *Revision, files []RevisionFile) error { + warnings := r.Warnings + if len(warnings) == 0 { + warnings = json.RawMessage("[]") + } + if _, err := t.q.Exec(ctx, `INSERT INTO config_revisions (number, created_at, created_by, created_by_name, comment, warnings) + VALUES ($1, $2, $3, $4, $5, $6)`, r.Number, r.CreatedAt, r.CreatedBy, r.CreatedByName, r.Comment, []byte(warnings)); err != nil { + return mapErr(err) + } + b := &pgx.Batch{} + ordinal := map[string]int32{} + for _, f := range files { + n := ordinal[f.Role] + ordinal[f.Role] = n + 1 + b.Queue(`INSERT INTO config_revision_files (revision, role, ordinal, path, mode, create_only, content) + VALUES ($1, $2::agent_role, $3, $4, $5, $6, $7)`, + r.Number, f.Role, n, f.Path, int32(f.Mode), f.CreateOnly, f.Content) //nolint:gosec // modes are at most 0o777 (checked by the database) + } + return t.execBatch(ctx, b) +} + +// RevisionFiles returns the files of a revision ordered by role and +// generator order. role "" returns both roles. +func (t *Tx) RevisionFiles(ctx context.Context, number int64, role string) ([]RevisionFile, error) { + sql := `SELECT role, path, mode, create_only, content FROM config_revision_files WHERE revision = $1` + args := []any{number} + if role != "" { + sql += ` AND role = $2::agent_role` + args = append(args, role) + } + rows, err := t.q.Query(ctx, sql+` ORDER BY role, ordinal`, args...) + if err != nil { + return nil, mapErr(err) + } + defer rows.Close() + out := []RevisionFile{} + for rows.Next() { + var f RevisionFile + var mode int32 + if err := rows.Scan(&f.Role, &f.Path, &mode, &f.CreateOnly, &f.Content); err != nil { + return nil, mapErr(err) + } + f.Mode = uint32(mode) //nolint:gosec // 0..511 by constraint + out = append(out, f) + } + return out, mapErr(rows.Err()) +} + +// ListRevisions returns up to limit revisions with a number below before +// (0 = no bound), newest first. +func (t *Tx) ListRevisions(ctx context.Context, limit int, before int64) ([]*Revision, error) { + sql := `SELECT r.number, r.created_at, r.created_by, r.created_by_name, r.comment, jsonb_array_length(r.warnings), + (SELECT count(*) FROM config_revision_files f WHERE f.revision = r.number AND f.role = 'dns'), + (SELECT count(*) FROM config_revision_files f WHERE f.revision = r.number AND f.role = 'dhcp') + FROM config_revisions r` + args := []any{} + if before > 0 { + sql += ` WHERE r.number < $1` + args = append(args, before) + } + sql += fmt.Sprintf(` ORDER BY r.number DESC LIMIT %d`, limit) + rows, err := t.q.Query(ctx, sql, args...) + if err != nil { + return nil, mapErr(err) + } + defer rows.Close() + var out []*Revision + for rows.Next() { + var r Revision + if err := rows.Scan(&r.Number, &r.CreatedAt, &r.CreatedBy, &r.CreatedByName, &r.Comment, &r.WarningCount, + &r.DNSFiles, &r.DHCPFiles); err != nil { + return nil, mapErr(err) + } + out = append(out, &r) + } + return out, mapErr(rows.Err()) +} + +// Deployment is a row of deployments; AgentHostname comes from the agent. +type Deployment struct { + ID int64 + AgentID string + AgentHostname string + CorrelationID string + Revision int64 + Trigger string + Outcome string + Message string + ActiveRevision int64 + StartedAt time.Time + FinishedAt time.Time + ReportedAt time.Time +} + +const deploymentColumns = `d.id, d.agent_id, a.hostname, d.correlation_id, d.revision, d.trigger_kind, d.outcome, d.message, + d.active_revision, d.started_at, d.finished_at, d.reported_at` + +func scanDeployment(row pgx.Row) (*Deployment, error) { + var d Deployment + if err := row.Scan(&d.ID, &d.AgentID, &d.AgentHostname, &d.CorrelationID, &d.Revision, &d.Trigger, &d.Outcome, + &d.Message, &d.ActiveRevision, &d.StartedAt, &d.FinishedAt, &d.ReportedAt); err != nil { + return nil, mapErr(err) + } + return &d, nil +} + +// InsertDeployment stores a deployment report; ID and ReportedAt are set. +func (t *Tx) InsertDeployment(ctx context.Context, d *Deployment) error { + return mapErr(t.q.QueryRow(ctx, ` + INSERT INTO deployments (agent_id, correlation_id, revision, trigger_kind, outcome, message, active_revision, + started_at, finished_at) + VALUES ($1, $2, $3, $4::deployment_trigger, $5::deployment_outcome, $6, $7, $8, $9) + RETURNING id, reported_at`, + d.AgentID, d.CorrelationID, d.Revision, d.Trigger, d.Outcome, d.Message, d.ActiveRevision, + d.StartedAt, d.FinishedAt).Scan(&d.ID, &d.ReportedAt)) +} + +// DeploymentFilter selects deployments. Zero values mean "no restriction". +type DeploymentFilter struct { + AgentID string + Revision int64 + BeforeID int64 + Limit int +} + +// ListDeployments returns deployments newest first. +func (t *Tx) ListDeployments(ctx context.Context, f DeploymentFilter) ([]*Deployment, error) { + var where []string + var args []any + add := func(cond string, v any) { + args = append(args, v) + where = append(where, fmt.Sprintf(cond, len(args))) + } + if f.AgentID != "" { + add("d.agent_id = $%d", f.AgentID) + } + if f.Revision > 0 { + add("d.revision = $%d", f.Revision) + } + if f.BeforeID > 0 { + add("d.id < $%d", f.BeforeID) + } + limit := f.Limit + if limit <= 0 { + limit = 100 + } + sql := `SELECT ` + deploymentColumns + ` FROM deployments d JOIN agents a ON a.id = d.agent_id` + if len(where) > 0 { + sql += " WHERE " + strings.Join(where, " AND ") + } + sql += fmt.Sprintf(" ORDER BY d.id DESC LIMIT %d", limit) + rows, err := t.q.Query(ctx, sql, args...) + if err != nil { + return nil, mapErr(err) + } + defer rows.Close() + var out []*Deployment + for rows.Next() { + d, err := scanDeployment(rows) + if err != nil { + return nil, err + } + out = append(out, d) + } + return out, mapErr(rows.Err()) +} + +// LatestDeployments returns the newest deployment of every agent that has +// one, keyed by agent ID. +func (t *Tx) LatestDeployments(ctx context.Context) (map[string]*Deployment, error) { + rows, err := t.q.Query(ctx, `SELECT DISTINCT ON (d.agent_id) `+deploymentColumns+` + FROM deployments d JOIN agents a ON a.id = d.agent_id ORDER BY d.agent_id, d.id DESC`) + if err != nil { + return nil, mapErr(err) + } + defer rows.Close() + out := map[string]*Deployment{} + for rows.Next() { + d, err := scanDeployment(rows) + if err != nil { + return nil, err + } + out[d.AgentID] = d + } + return out, mapErr(rows.Err()) +} + +// LatestDeployment returns the newest deployment of an agent or ErrNotFound. +func (t *Tx) LatestDeployment(ctx context.Context, agentID string) (*Deployment, error) { + return scanDeployment(t.q.QueryRow(ctx, `SELECT `+deploymentColumns+` + FROM deployments d JOIN agents a ON a.id = d.agent_id WHERE d.agent_id = $1 ORDER BY d.id DESC LIMIT 1`, agentID)) +} diff --git a/server/internal/store/store.go b/server/internal/store/store.go index b5d2d09..1a8eb46 100644 --- a/server/internal/store/store.go +++ b/server/internal/store/store.go @@ -134,6 +134,15 @@ type querier interface { Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row + SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults +} + +// execBatch sends b and returns the first error. +func (t *Tx) execBatch(ctx context.Context, b *pgx.Batch) error { + if b.Len() == 0 { + return nil + } + return mapErr(t.q.SendBatch(ctx, b).Close()) } // Tx is a repository bound to a transaction.