refactor(store): wave B.2 — settings + identity_link + logs repos (#247) (#248)

Continues the per-domain migration started by Wave B.1 (#246). Three
small handlers move from Store.Queries() to domain-specific repos
under store.Repos(); cross-domain reads (GetUserByID, GetDeviceByID)
stay on Queries until those domains migrate in later sub-waves.

Domains migrated:

- SettingsRepo (GetServer) — settings_handler.go.
- IdentityLinkRepo (Get / ListForUser / CountForUser) — identity_link_handler.go
  + populateUserIdentityLinks in user_handler.go. ListForUser returns
  the join-shape IdentityLinkWithProvider so handlers can't confuse
  it with the basic Get shape.
- LogsRepo (CreateQueryResult / GetQueryResult / ExpirePendingQueryResult) —
  logs_handler.go. First write-side migration in Wave B; the :exec
  signatures stay opaque to handlers, matching the read-side pattern.

Each new repo's Postgres impl translates pgx.ErrNoRows to
store.ErrNotFound at the boundary, matching the contract established
in Wave A (#244) and applied in Wave B.1.

Non-goals:

- User / device repos — cross-domain reads in these handlers
  remain on Store.Queries() until those domains migrate.
- Compliance-policy queries — separate sub-wave.

Closes #247.

Author: PaulDotterer

internal/api/identity_link_handler.go

@@ -12,7 +12,6 @@ import (
 	"github.com/manchtools/power-manage/server/internal/eventtypes"
 	"github.com/manchtools/power-manage/server/internal/eventtypes/payloads"
 	"github.com/manchtools/power-manage/server/internal/store"
-	db "github.com/manchtools/power-manage/server/internal/store/generated"
 )
 
 // IdentityLinkHandler handles self-service identity linking RPCs.
@@ -33,7 +32,7 @@ func (h *IdentityLinkHandler) ListIdentityLinks(ctx context.Context, req *connec
 		return nil, err
 	}
 
-	links, err := h.store.Queries().ListIdentityLinksForUser(ctx, userCtx.ID)
+	links, err := h.store.Repos().IdentityLink.ListForUser(ctx, userCtx.ID)
 	if err != nil {
 		return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to list identity links")
 	}
@@ -60,7 +59,7 @@ func (h *IdentityLinkHandler) UnlinkIdentity(ctx context.Context, req *connect.R
 	}
 
 	// Get the link to verify ownership
-	link, err := h.store.Queries().GetIdentityLinkByID(ctx, req.Msg.LinkId)
+	link, err := h.store.Repos().IdentityLink.Get(ctx, req.Msg.LinkId)
 	if err != nil {
 		return nil, handleGetError(ctx, err, ErrIdentityLinkNotFound, "identity link not found")
 	}
@@ -78,7 +77,7 @@ func (h *IdentityLinkHandler) UnlinkIdentity(ctx context.Context, req *connect.R
 		return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to get user")
 	}
 
-	linkCount, err := h.store.Queries().CountIdentityLinksForUser(ctx, targetUserID)
+	linkCount, err := h.store.Repos().IdentityLink.CountForUser(ctx, targetUserID)
 	if err != nil {
 		return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to count identity links")
 	}
@@ -106,7 +105,7 @@ func (h *IdentityLinkHandler) UnlinkIdentity(ctx context.Context, req *connect.R
 }
 
 // identityLinkRowToProto converts a joined identity link row to a proto message.
-func identityLinkRowToProto(link db.ListIdentityLinksForUserRow) *pm.IdentityLink {
+func identityLinkRowToProto(link store.IdentityLinkWithProvider) *pm.IdentityLink {
 	protoLink := &pm.IdentityLink{
 		Id:            link.ID,
 		UserId:        link.UserID,

internal/api/logs_handler.go

@@ -56,10 +56,7 @@ func (h *LogsHandler) QueryDeviceLogs(ctx context.Context, req *connect.Request[
 	queryID := ulid.Make().String()
 
 	// Create pending result row
-	if err := h.store.Queries().CreateLogQueryResult(ctx, generated.CreateLogQueryResultParams{
-		QueryID:  queryID,
-		DeviceID: msg.DeviceId,
-	}); err != nil {
+	if err := h.store.Repos().Logs.CreateQueryResult(ctx, queryID, msg.DeviceId); err != nil {
 		return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to create log query result")
 	}
 
@@ -83,10 +80,7 @@ func (h *LogsHandler) QueryDeviceLogs(ctx context.Context, req *connect.Request[
 	); err != nil {
 		h.logger.Error("log query enqueue failed; marking result expired",
 			"query_id", queryID, "device_id", msg.DeviceId, "error", err)
-		if expireErr := h.store.Queries().ExpirePendingLogQueryResult(ctx, generated.ExpirePendingLogQueryResultParams{
-			QueryID: queryID,
-			Error:   fmt.Sprintf("dispatch enqueue failed: %v", err),
-		}); expireErr != nil {
+		if expireErr := h.store.Repos().Logs.ExpirePendingQueryResult(ctx, queryID, fmt.Sprintf("dispatch enqueue failed: %v", err)); expireErr != nil {
 			h.logger.Error("failed to mark enqueue-failed log query result as expired",
 				"query_id", queryID, "error", expireErr)
 		}
@@ -104,18 +98,15 @@ func (h *LogsHandler) QueryDeviceLogs(ctx context.Context, req *connect.Request[
 
 // GetDeviceLogResult polls for the result of a dispatched log query.
 func (h *LogsHandler) GetDeviceLogResult(ctx context.Context, req *connect.Request[pm.GetDeviceLogResultRequest]) (*connect.Response[pm.GetDeviceLogResultResponse], error) {
-	result, err := h.store.Queries().GetLogQueryResult(ctx, req.Msg.QueryId)
+	result, err := h.store.Repos().Logs.GetQueryResult(ctx, req.Msg.QueryId)
 	if err != nil {
 		return nil, apiErrorCtx(ctx, ErrQueryResultNotFound, connect.CodeNotFound, "log query result not found")
 	}
 
 	// Auto-expire pending results that have been waiting too long
 	if !result.Completed && time.Since(result.CreatedAt) > logQueryResultTimeout {
 		timeoutErr := "log query timed out: device did not respond within 5 minutes"
-		if err := h.store.Queries().ExpirePendingLogQueryResult(ctx, generated.ExpirePendingLogQueryResultParams{
-			QueryID: result.QueryID,
-			Error:   timeoutErr,
-		}); err != nil {
+		if err := h.store.Repos().Logs.ExpirePendingQueryResult(ctx, result.QueryID, timeoutErr); err != nil {
 			h.logger.Warn("failed to expire pending log query result", "query_id", result.QueryID, "error", err)
 		}
 		result.Completed = true

internal/api/settings_handler.go

@@ -32,7 +32,7 @@ func NewSettingsHandler(st *store.Store, logger *slog.Logger, systemActions *Sys
 
 // GetServerSettings returns the current server settings.
 func (h *SettingsHandler) GetServerSettings(ctx context.Context, req *connect.Request[pm.GetServerSettingsRequest]) (*connect.Response[pm.GetServerSettingsResponse], error) {
-	settings, err := h.store.Queries().GetServerSettings(ctx)
+	settings, err := h.store.Repos().Settings.GetServer(ctx)
 	if err != nil {
 		return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to get server settings")
 	}
@@ -62,7 +62,7 @@ func (h *SettingsHandler) UpdateServerSettings(ctx context.Context, req *connect
 	}
 
 	// Read back from projection
-	settings, err := h.store.Queries().GetServerSettings(ctx)
+	settings, err := h.store.Repos().Settings.GetServer(ctx)
 	if err != nil {
 		return nil, apiErrorCtx(ctx, ErrInternal, connect.CodeInternal, "failed to read server settings")
 	}

internal/api/user_handler.go

@@ -799,7 +799,7 @@ func (h *UserHandler) UpdateUserSshSettings(ctx context.Context, req *connect.Re
 
 // populateUserIdentityLinks loads identity links for a user and attaches them to the proto User.
 func (h *UserHandler) populateUserIdentityLinks(ctx context.Context, user *pm.User) {
-	links, err := h.store.Queries().ListIdentityLinksForUser(ctx, user.Id)
+	links, err := h.store.Repos().IdentityLink.ListForUser(ctx, user.Id)
 	if err != nil {
 		return
 	}

internal/store/identity_link.go

@@ -0,0 +1,49 @@
+package store
+
+import (
+	"context"
+	"time"
+)
+
+// IdentityLink is the per-user SSO-link projection row.
+type IdentityLink struct {
+	ID            string
+	UserID        string
+	ProviderID    string
+	ExternalID    string
+	ExternalEmail string
+	ExternalName  string
+	LinkedAt      time.Time
+	LastLoginAt   *time.Time
+}
+
+// IdentityLinkWithProvider is the join shape used by the
+// ListForUser query, which augments each link with display fields
+// from identity_providers_projection. Kept distinct from IdentityLink
+// so callers can't mistakenly assume provider info is populated on a
+// basic Get.
+type IdentityLinkWithProvider struct {
+	IdentityLink
+	ProviderName string
+	ProviderSlug string
+}
+
+// IdentityLinkRepo reads SSO-link projection rows for self-service
+// identity management. Writes happen via the IdentityUnlinked /
+// IdentityLinked event types, not through this interface.
+type IdentityLinkRepo interface {
+	// Get returns the identity link with the given ID. Returns
+	// ErrNotFound if the link does not exist (unlinked or never
+	// existed).
+	Get(ctx context.Context, id string) (IdentityLink, error)
+
+	// ListForUser returns all identity links owned by the user,
+	// joined with provider display info, ordered newest-linked
+	// first. Returns an empty slice when the user has no links.
+	ListForUser(ctx context.Context, userID string) ([]IdentityLinkWithProvider, error)
+
+	// CountForUser returns the number of identity links for the
+	// user. Used by the "cannot unlink last auth method"
+	// pre-condition in UnlinkIdentity.
+	CountForUser(ctx context.Context, userID string) (int64, error)
+}

internal/store/logs.go

@@ -0,0 +1,45 @@
+package store
+
+import (
+	"context"
+	"time"
+)
+
+// LogQueryResult is the device-log-query result row. Backed by
+// log_query_results, NOT by the event store — this is a derived
+// scratchpad for the async dispatch/poll dance, so the shape can
+// evolve without event-schema implications.
+type LogQueryResult struct {
+	QueryID     string
+	DeviceID    string
+	Completed   bool
+	Success     bool
+	Error       string
+	Logs        string
+	CreatedAt   time.Time
+	CompletedAt *time.Time
+}
+
+// LogsRepo manages the device-log-query result rows used by the
+// QueryDeviceLogs / GetDeviceLogResult RPC pair. The agent writes
+// the completed result via a separate gateway path; this repo
+// covers the creation, expiry, and read sites the control handler
+// owns.
+type LogsRepo interface {
+	// CreateQueryResult inserts a fresh pending row for the given
+	// query+device pair. Called when the control handler dispatches
+	// a log query to the agent.
+	CreateQueryResult(ctx context.Context, queryID, deviceID string) error
+
+	// GetQueryResult returns the row for the given query ID.
+	// Returns ErrNotFound when no such row exists (caller used a
+	// bogus query ID, or the row was already aged out by the
+	// periodic cleanup).
+	GetQueryResult(ctx context.Context, queryID string) (LogQueryResult, error)
+
+	// ExpirePendingQueryResult marks a pending row as failed with
+	// the supplied error message. Used by the auto-expiry path
+	// (5-minute poll timeout) and by the dispatch-failure recovery
+	// in QueryDeviceLogs.
+	ExpirePendingQueryResult(ctx context.Context, queryID, errMsg string) error
+}

internal/store/postgres/identity_link.go

@@ -0,0 +1,80 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/manchtools/power-manage/server/internal/store"
+	"github.com/manchtools/power-manage/server/internal/store/generated"
+)
+
+// IdentityLink implements store.IdentityLinkRepo against the
+// identity_links_projection via sqlc-generated queries.
+type IdentityLink struct {
+	q *generated.Queries
+}
+
+// NewIdentityLink returns an IdentityLink repo bound to the given
+// sqlc handle.
+func NewIdentityLink(q *generated.Queries) *IdentityLink {
+	return &IdentityLink{q: q}
+}
+
+// Get returns a single identity link by its ID. pgx.ErrNoRows is
+// translated to store.ErrNotFound.
+func (i *IdentityLink) Get(ctx context.Context, id string) (store.IdentityLink, error) {
+	row, err := i.q.GetIdentityLinkByID(ctx, id)
+	if err != nil {
+		return store.IdentityLink{}, fmt.Errorf("identity_link: get: %w", translateNotFound(err))
+	}
+	return store.IdentityLink{
+		ID:            row.ID,
+		UserID:        row.UserID,
+		ProviderID:    row.ProviderID,
+		ExternalID:    row.ExternalID,
+		ExternalEmail: row.ExternalEmail,
+		ExternalName:  row.ExternalName,
+		LinkedAt:      row.LinkedAt,
+		LastLoginAt:   row.LastLoginAt,
+	}, nil
+}
+
+// ListForUser returns the user's links joined with provider display
+// fields. The underlying :many query returns an empty slice (not an
+// error) when the user has no links.
+func (i *IdentityLink) ListForUser(ctx context.Context, userID string) ([]store.IdentityLinkWithProvider, error) {
+	rows, err := i.q.ListIdentityLinksForUser(ctx, userID)
+	if err != nil {
+		return nil, fmt.Errorf("identity_link: list for user: %w", err)
+	}
+	out := make([]store.IdentityLinkWithProvider, len(rows))
+	for j, r := range rows {
+		out[j] = store.IdentityLinkWithProvider{
+			IdentityLink: store.IdentityLink{
+				ID:            r.ID,
+				UserID:        r.UserID,
+				ProviderID:    r.ProviderID,
+				ExternalID:    r.ExternalID,
+				ExternalEmail: r.ExternalEmail,
+				ExternalName:  r.ExternalName,
+				LinkedAt:      r.LinkedAt,
+				LastLoginAt:   r.LastLoginAt,
+			},
+			ProviderName: r.ProviderName,
+			ProviderSlug: r.ProviderSlug,
+		}
+	}
+	return out, nil
+}
+
+// CountForUser returns the number of identity links the user owns.
+// The COUNT(*) :one query always returns a row, so the
+// pgx.ErrNoRows path is unreachable today — the translateNotFound
+// wrap stays as future-proofing if the query shape ever changes.
+func (i *IdentityLink) CountForUser(ctx context.Context, userID string) (int64, error) {
+	n, err := i.q.CountIdentityLinksForUser(ctx, userID)
+	if err != nil {
+		return 0, fmt.Errorf("identity_link: count for user: %w", translateNotFound(err))
+	}
+	return n, nil
+}

internal/store/postgres/logs.go

@@ -0,0 +1,64 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/manchtools/power-manage/server/internal/store"
+	"github.com/manchtools/power-manage/server/internal/store/generated"
+)
+
+// Logs implements store.LogsRepo against log_query_results.
+type Logs struct {
+	q *generated.Queries
+}
+
+// NewLogs returns a Logs repo bound to the given sqlc handle.
+func NewLogs(q *generated.Queries) *Logs {
+	return &Logs{q: q}
+}
+
+// CreateQueryResult inserts a pending row. The underlying :exec
+// query has no return value beyond errors; backend constraint
+// violations (e.g. duplicate query_id) surface unchanged.
+func (l *Logs) CreateQueryResult(ctx context.Context, queryID, deviceID string) error {
+	if err := l.q.CreateLogQueryResult(ctx, generated.CreateLogQueryResultParams{
+		QueryID:  queryID,
+		DeviceID: deviceID,
+	}); err != nil {
+		return fmt.Errorf("logs: create query result: %w", err)
+	}
+	return nil
+}
+
+// GetQueryResult returns the result row. pgx.ErrNoRows is
+// translated to store.ErrNotFound.
+func (l *Logs) GetQueryResult(ctx context.Context, queryID string) (store.LogQueryResult, error) {
+	row, err := l.q.GetLogQueryResult(ctx, queryID)
+	if err != nil {
+		return store.LogQueryResult{}, fmt.Errorf("logs: get query result: %w", translateNotFound(err))
+	}
+	return store.LogQueryResult{
+		QueryID:     row.QueryID,
+		DeviceID:    row.DeviceID,
+		Completed:   row.Completed,
+		Success:     row.Success,
+		Error:       row.Error,
+		Logs:        row.Logs,
+		CreatedAt:   row.CreatedAt,
+		CompletedAt: row.CompletedAt,
+	}, nil
+}
+
+// ExpirePendingQueryResult marks a pending row as failed. The :exec
+// query's WHERE clause includes `completed = FALSE` so calling on
+// an already-completed row is a no-op and not an error.
+func (l *Logs) ExpirePendingQueryResult(ctx context.Context, queryID, errMsg string) error {
+	if err := l.q.ExpirePendingLogQueryResult(ctx, generated.ExpirePendingLogQueryResultParams{
+		QueryID: queryID,
+		Error:   errMsg,
+	}); err != nil {
+		return fmt.Errorf("logs: expire pending: %w", err)
+	}
+	return nil
+}

internal/store/postgres/settings.go

@@ -0,0 +1,35 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/manchtools/power-manage/server/internal/store"
+	"github.com/manchtools/power-manage/server/internal/store/generated"
+)
+
+// Settings implements store.SettingsRepo against the Postgres
+// server_settings_projection via sqlc-generated queries.
+type Settings struct {
+	q *generated.Queries
+}
+
+// NewSettings returns a Settings repo bound to the given sqlc handle.
+func NewSettings(q *generated.Queries) *Settings {
+	return &Settings{q: q}
+}
+
+// GetServer returns the global settings row, translating
+// pgx.ErrNoRows to store.ErrNotFound at the repo boundary so callers
+// rely solely on store.IsNotFound.
+func (s *Settings) GetServer(ctx context.Context) (store.ServerSettings, error) {
+	row, err := s.q.GetServerSettings(ctx)
+	if err != nil {
+		return store.ServerSettings{}, fmt.Errorf("settings: get server: %w", translateNotFound(err))
+	}
+	return store.ServerSettings{
+		UserProvisioningEnabled: row.UserProvisioningEnabled,
+		SshAccessForAll:         row.SshAccessForAll,
+		UpdatedAt:               row.UpdatedAt,
+	}, nil
+}