PocketBase store: query rows are closed by defer, as lint asks

Three reads closed their rows by hand, early, because the call writes
right after or loops over chunks; any other early return would have
leaked them, and golangci-lint's sqlclosecheck failed CI on it. Each
read is now a function of its own with a deferred Close, as sqlstore
already does, so the rows are still closed before the writes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

server/internal/storage/pocketbase/keys.go

@@ -99,27 +99,38 @@ func (s *Keys) DeviceListChanges(ctx context.Context, userIDs []string, from, to
 			args = append(args, to)
 		}
 
-		rows, err := s.db.QueryContext(ctx, query, args...)
+		changes, err := s.deviceListChanges(ctx, query, args)
 		if err != nil {
-			return nil, fmt.Errorf("read device list changes: %w", err)
-		}
-		for rows.Next() {
-			var c storage.DeviceListChange
-			if err := rows.Scan(&c.UserID, &c.StreamPos); err != nil {
-				_ = rows.Close()
-				return nil, fmt.Errorf("read device list changes: %w", err)
-			}
-			out = append(out, c)
-		}
-		_ = rows.Close()
-		if err := rows.Err(); err != nil {
-			return nil, fmt.Errorf("read device list changes: %w", err)
+			return nil, err
 		}
+		out = append(out, changes...)
 	}
 	slices.SortFunc(out, func(a, b storage.DeviceListChange) int { return cmp.Compare(a.StreamPos, b.StreamPos) })
 	return out, nil
 }
 
+// deviceListChanges runs one chunk of DeviceListChanges.
+func (s *Keys) deviceListChanges(ctx context.Context, query string, args []any) ([]storage.DeviceListChange, error) {
+	rows, err := s.db.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("read device list changes: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var out []storage.DeviceListChange
+	for rows.Next() {
+		var c storage.DeviceListChange
+		if err := rows.Scan(&c.UserID, &c.StreamPos); err != nil {
+			return nil, fmt.Errorf("read device list changes: %w", err)
+		}
+		out = append(out, c)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("read device list changes: %w", err)
+	}
+	return out, nil
+}
+
 // DeviceListPosition implements storage.KeyStore.
 func (s *Keys) DeviceListPosition(ctx context.Context, userID string) (int64, error) {
 	var pos int64
@@ -257,23 +268,9 @@ func (s *Keys) AddOneTimeKeys(ctx context.Context, userID, deviceID string, keys
 		for i, k := range keys {
 			ids[i] = k.KeyID
 		}
-		held := map[string]string{}
-		rows, err := t.query(`SELECT key_id, key_json FROM keyserver_one_time_keys
-			WHERE user_id = ? AND device_id = ? AND key_id IN `+inList, userID, deviceID, jsonList(ids))
+		held, err := heldOneTimeKeys(t, userID, deviceID, ids)
 		if err != nil {
-			return fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err)
-		}
-		for rows.Next() {
-			var id, keyJSON string
-			if err := rows.Scan(&id, &keyJSON); err != nil {
-				_ = rows.Close()
-				return fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err)
-			}
-			held[id] = keyJSON
-		}
-		_ = rows.Close()
-		if err := rows.Err(); err != nil {
-			return fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err)
+			return err
 		}
 
 		for _, k := range keys {
@@ -293,6 +290,30 @@ func (s *Keys) AddOneTimeKeys(ctx context.Context, userID, deviceID string, keys
 	})
 }
 
+// heldOneTimeKeys reads which of ids the device already holds, by key ID.
+// The rows are closed when it returns, before the call writes.
+func heldOneTimeKeys(t *tx, userID, deviceID string, ids []string) (map[string]string, error) {
+	rows, err := t.query(`SELECT key_id, key_json FROM keyserver_one_time_keys
+		WHERE user_id = ? AND device_id = ? AND key_id IN `+inList, userID, deviceID, jsonList(ids))
+	if err != nil {
+		return nil, fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	held := map[string]string{}
+	for rows.Next() {
+		var id, keyJSON string
+		if err := rows.Scan(&id, &keyJSON); err != nil {
+			return nil, fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err)
+		}
+		held[id] = keyJSON
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err)
+	}
+	return held, nil
+}
+
 // OneTimeKeyCounts implements storage.KeyStore.
 func (s *Keys) OneTimeKeyCounts(ctx context.Context, userID, deviceID string) (map[string]int, error) {
 	rows, err := s.db.QueryContext(ctx, `SELECT algorithm, COUNT(*) FROM keyserver_one_time_keys

server/internal/storage/pocketbase/rooms.go

@@ -501,22 +501,9 @@ func (w *roomWrite) applyStateDelta(rw storage.RoomWrite) error {
 func (w *roomWrite) recountMembers(rw storage.RoomWrite) error {
 	joined := map[string]bool{}
 	if !rw.StateDeltaIsWholeState {
-		rows, err := w.query(`SELECT state_key FROM roomserver_current_state
-			WHERE room_id = ? AND event_type = 'm.room.member' AND membership = 'join'`, rw.Event.RoomID)
-		if err != nil {
-			return fmt.Errorf("recount joined members: %w", err)
-		}
-		for rows.Next() {
-			var user string
-			if err := rows.Scan(&user); err != nil {
-				_ = rows.Close()
-				return fmt.Errorf("recount joined members: %w", err)
-			}
-			joined[user] = true
-		}
-		_ = rows.Close()
-		if err := rows.Err(); err != nil {
-			return fmt.Errorf("recount joined members: %w", err)
+		var err error
+		if joined, err = w.storedJoins(rw.Event.RoomID); err != nil {
+			return err
 		}
 	}
 	for _, e := range rw.StateDelta {
@@ -537,6 +524,30 @@ func (w *roomWrite) recountMembers(rw storage.RoomWrite) error {
 	return nil
 }
 
+// storedJoins reads the users the stored current state has joined to the
+// room. The rows are closed when it returns, before the call writes.
+func (w *roomWrite) storedJoins(roomID string) (map[string]bool, error) {
+	rows, err := w.query(`SELECT state_key FROM roomserver_current_state
+		WHERE room_id = ? AND event_type = 'm.room.member' AND membership = 'join'`, roomID)
+	if err != nil {
+		return nil, fmt.Errorf("recount joined members: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	joined := map[string]bool{}
+	for rows.Next() {
+		var user string
+		if err := rows.Scan(&user); err != nil {
+			return nil, fmt.Errorf("recount joined members: %w", err)
+		}
+		joined[user] = true
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("recount joined members: %w", err)
+	}
+	return joined, nil
+}
+
 func (w *roomWrite) replaceExtremities(roomID string, extremities []string) error {
 	if len(extremities) == 0 {
 		return nil