Add support for lobby filtering

Using https://github.com/poki/mongodb-filter-to-postgres

Co-authored-by: Koen Bollen <[email protected]>

cmd/testproxy/main.go

@@ -2,12 +2,15 @@ package main
 
 import (
 	"context"
+	"io"
 	"net"
 	"net/http"
+	"os"
 	"os/signal"
 	"syscall"
 	"time"
 
+	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/koenbollen/logging"
 	"github.com/poki/netlib/internal/util"
 	"go.uber.org/zap"
@@ -23,6 +26,15 @@ func main() {
 	defer logger.Info("fin")
 	ctx = logging.WithLogger(ctx, logger)
 
+	db, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
+	if err != nil {
+		logger.Fatal("failed to connect", zap.Error(err))
+	}
+
+	if err := db.Ping(ctx); err != nil {
+		logger.Fatal("failed to ping db", zap.Error(err))
+	}
+
 	connections := make(map[string]net.Conn)
 	interrupts := make(map[string]bool)
 	http.HandleFunc("/create", func(w http.ResponseWriter, r *http.Request) {
@@ -85,6 +97,18 @@ func main() {
 			delete(connections, id)
 		}
 	})
+	http.HandleFunc("/sql", func(w http.ResponseWriter, r *http.Request) {
+		sql, err := io.ReadAll(r.Body)
+		if err != nil {
+			panic(err)
+		}
+
+		// This process is only ran during tests.
+		_, err = db.Exec(ctx, string(sql))
+		if err != nil {
+			panic(err)
+		}
+	})
 
 	addr := util.Getenv("ADDR", ":8080")
 	server := &http.Server{

features/lobbies.feature

@@ -2,6 +2,7 @@ Feature: Lobby Discovery
 
   Background:
     Given the "signaling" backend is running
+    And the "testproxy" backend is running
 
   Scenario: List empty lobby set
     Given "green" is connected and ready for game "f666036d-d9e1-4d70-b0c3-4a68b24a9884"
@@ -21,15 +22,15 @@ Feature: Lobby Discovery
     And "blue" is connected and ready for game "f666036d-d9e1-4d70-b0c3-4a68b24a9884"
 
     When "blue" creates a lobby with these settings:
-      """
+      """json
       {
         "public": true
       }
       """
     And "blue" receives the network event "lobby" with the argument "prb67ouj837u"
 
     When "green" requests all lobbies
-    Then "green" should have received only these lobbies
+    Then "green" should have received only these lobbies:
       | code         | playerCount |
       | prb67ouj837u | 1           |
 
@@ -39,26 +40,68 @@ Feature: Lobby Discovery
     And "yellow" is connected and ready for game "f666036d-d9e1-4d70-b0c3-4a68b24a9884"
 
     When "blue" creates a lobby with these settings:
-      """
+      """json
       {
         "public": true
       }
       """
     And "blue" receives the network event "lobby" with the argument "dhgp75mn2bll"
     And "yellow" creates a lobby with these settings:
-      """
+      """json
       {
         "public": false
       }
       """
     And "yellow" receives the network event "lobby" with the argument "1qva9vyurwbbl"
 
     When "green" requests all lobbies
-    Then "green" should have received only these lobbies
+    Then "green" should have received only these lobbies:
       | code         | playerCount | public |
       | dhgp75mn2bll | 1           | true   |
 
+  Scenario: Filter on playerCount
+    Given "green" is connected and ready for game "f666036d-d9e1-4d70-b0c3-4a68b24a9884"
+    And these lobbies exist:
+      | code          | game                                 | playerCount | public |
+      | 0qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 1           | true   |
+      | 1qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 2           | false  |
+      | 2qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 3           | true   |
+      | 3qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 4           | true   |
+      | 4qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 5           | true   |
+      | 5qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 6           | true   |
+      | 6qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 7           | false  |
+      | 7qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 8           | true   |
+      | 8qva9vyurwbbl | 54fa57d5-b4bd-401d-981d-2c13de99be27 | 9           | true   |
+      | 9qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 10          | true   |
 
+    When "green" requests lobbies with this filter:
+      """json
+      {
+        "playerCount": {"$gte": 5}
+      }
+      """
+    Then "green" should have received only these lobbies:
+      | code          | playerCount | public |
+      | 4qva9vyurwbbl | 5           | true   |
+      | 5qva9vyurwbbl | 6           | true   |
+      | 7qva9vyurwbbl | 8           | true   |
+      | 9qva9vyurwbbl | 10          | true   |
 
+  Scenario: Filter on customData
+    Given "green" is connected and ready for game "f666036d-d9e1-4d70-b0c3-4a68b24a9884"
+    And these lobbies exist:
+      | code          | game                                 | playerCount | meta               | public | created_at |
+      | 0qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 1           | {"map": "de_nuke"} | true   | 2020-01-01 |
+      | 1qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 1           | {"map": "de_dust"} | true   | 2020-01-02 | 
+      | 2qva9vyurwbbl | f666036d-d9e1-4d70-b0c3-4a68b24a9884 | 1           | {"map": "de_nuke"} | true   | 2020-01-03 |
 
-
+    When "green" requests lobbies with this filter:
+      """json
+      {
+        "map": "de_nuke",
+        "createdAt": {"$gte": "2020-01-02"}
+      }
+      """
+    Then "green" should have received only these lobbies:
+      | code          |
+      | 2qva9vyurwbbl |

features/support/steps/backend.ts

@@ -5,21 +5,29 @@ import { World } from '../world'
 Given('the {string} backend is running', async function (this: World, backend: string) {
   return await new Promise(resolve => {
     const port = 10000 + Math.ceil(Math.random() * 1000)
+    const env: NodeJS.ProcessEnv = {
+      ...process.env,
+      ADDR: `127.0.0.1:${port}`,
+      ENV: 'test'
+    }
+
+    if (this.databaseURL === undefined) {
+      env.DATABASE_URL = this.databaseURL
+    }
+
     const prc = spawn(`/tmp/netlib-cucumber-${backend}`, [], {
       windowsHide: true,
-      env: {
-        ...process.env,
-        ADDR: `127.0.0.1:${port}`,
-        ENV: 'test'
-      }
+      env
     })
     prc.stderr.setEncoding('utf8')
     prc.stderr.on('data', (data: string) => {
       const lines = data.split('\n')
       lines.forEach(line => {
         try {
           const entry = JSON.parse(line)
-          if (entry.message === 'listening') {
+          if (entry.message === 'using database') {
+            this.databaseURL = entry.url
+          } else if (entry.message === 'listening') {
             resolve(undefined)
           }
         } catch (_) {

features/support/steps/network.ts

@@ -52,6 +52,52 @@ Given('{string} are joined in a lobby', async function (this: World, playerNames
   }
 })
 
+Given('these lobbies exist:', async function (this: World, lobbies: DataTable) {
+  if (this.testproxyURL === undefined) {
+    throw new Error('testproxy not active')
+  }
+
+  const columns: string[] = []
+  const values: string[] = []
+
+  lobbies.hashes().forEach(row => {
+    const v: string[] = []
+
+    Object.keys(row).forEach(key => {
+      const value = row[key] as string
+      if (key === 'playerCount') {
+        if (!columns.includes('peers')) {
+          columns.push('peers')
+        }
+
+        const n = parseInt(value, 10)
+        const peers: string[] = []
+
+        for (let i = 0; i < n; i++) {
+          peers.push(`'peer${i}'`)
+        }
+
+        v.push(`ARRAY [${peers.join(', ')}]`)
+      } else {
+        if (!columns.includes(key)) {
+          columns.push(key)
+        }
+
+        v.push(`'${value}'`)
+      }
+    })
+
+    values.push(`(${v.join(', ')})`)
+  })
+
+  console.log('INSERT INTO lobbies (' + columns.join(', ') + ') VALUES ' + values.join(', '))
+
+  await fetch(`${this.testproxyURL}/sql`, {
+    method: 'POST',
+    body: 'INSERT INTO lobbies (' + columns.join(', ') + ') VALUES ' + values.join(', ')
+  })
+})
+
 When('{string} creates a network for game {string}', function (this: World, playerName: string, gameID: string) {
   this.createPlayer(playerName, gameID)
 })
@@ -106,6 +152,15 @@ When('{string} requests all lobbies', async function (this: World, playerName: s
   player.lastReceivedLobbies = lobbies
 })
 
+When('{string} requests lobbies with this filter:', async function (this: World, playerName: string, filter: string) {
+  const player = this.players.get(playerName)
+  if (player == null) {
+    return 'no such player'
+  }
+  const lobbies = await player.network.list(filter)
+  player.lastReceivedLobbies = lobbies
+})
+
 Then('{string} receives the network event {string}', async function (this: World, playerName: string, eventName: string) {
   const player = this.players.get(playerName)
   if (player == null) {
@@ -160,7 +215,7 @@ Then('{string} should receive {int} lobbies', function (this: World, playerName:
   return player.lastReceivedLobbies?.length === expectedLobbyCount
 })
 
-Then('{string} should have received only these lobbies', function (this: World, playerName: string, expectedLobbies: DataTable) {
+Then('{string} should have received only these lobbies:', function (this: World, playerName: string, expectedLobbies: DataTable) {
   const player = this.players.get(playerName)
   if (player == null) {
     throw new Error('no such player')

features/support/world.ts

@@ -32,6 +32,7 @@ export class World extends CucumberWorld {
   public signalingURL?: string
   public testproxyURL?: string
   public useTestProxy: boolean = false
+  public databaseURL?: string
 
   public players: Map<string, Player> = new Map<string, Player>()
 

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/jackc/pgx/v5 v5.4.2
 	github.com/koenbollen/logging v0.0.0-20230520102501-e01d64214504
 	github.com/ory/dockertest/v3 v3.10.0
+	github.com/poki/mongodb-filter-to-postgres v0.0.0-20240503105833-0b1842cffb65
 	github.com/rs/cors v1.9.0
 	github.com/rs/xid v1.5.0
 	go.uber.org/zap v1.24.0

go.sum

@@ -124,6 +124,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/poki/mongodb-filter-to-postgres v0.0.0-20240503105833-0b1842cffb65 h1:3NaeAVswO0NlXp/kbQRg1WRQog3ma5CuktKnn214ekI=
+github.com/poki/mongodb-filter-to-postgres v0.0.0-20240503105833-0b1842cffb65/go.mod h1:euoY8HWNMnOQRo+yKhMPKwMUKW6Z6sNmmsmSJzhURTk=
 github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
 github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
 github.com/rs/cors v1.9.0 h1:l9HGsTsHJcvW14Nk7J9KFz8bzeAWXn3CG6bgt7LsrAE=

internal/signaling/stores/postgres.go

@@ -12,27 +12,28 @@ import (
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/koenbollen/logging"
+	"github.com/poki/mongodb-filter-to-postgres/filter"
 	"github.com/poki/netlib/internal/util"
 	"go.uber.org/zap"
 )
 
-type notificationPayload struct {
-	Topic string `json:"t"`
-	Data  []byte `json:"d"`
-}
-
 type PostgresStore struct {
 	DB *pgxpool.Pool
 
 	mutex             sync.Mutex
 	callbacks         map[string]map[uint64]SubscriptionCallback
 	nextCallbackIndex uint64
+	filterConverter   *filter.Converter
 }
 
 func NewPostgresStore(ctx context.Context, db *pgxpool.Pool) (*PostgresStore, error) {
 	s := &PostgresStore{
 		DB:        db,
 		callbacks: make(map[string]map[uint64]SubscriptionCallback),
+		filterConverter: filter.NewConverter(
+			filter.WithNestedJSONB("meta", "code", "playerCount", "createdAt", "updatedAt"),
+			filter.WithEmptyCondition("TRUE"), // No filter returns all lobbies.
+		),
 	}
 	go s.run(ctx)
 	return s, nil
@@ -274,31 +275,49 @@ func (s *PostgresStore) GetLobby(ctx context.Context, game, lobbyCode string) ([
 }
 
 func (s *PostgresStore) ListLobbies(ctx context.Context, game, filter string) ([]Lobby, error) {
+	// TODO: Remove this.
+	if filter == "" {
+		filter = "{}"
+	}
 
-	// TODO: Filters
+	where, values, err := s.filterConverter.Convert([]byte(filter), 2)
+	if err != nil {
+		logger := logging.GetLogger(ctx)
+		logger.Warn("failed to convert filter", zap.String("filter", filter), zap.Error(err))
+		return nil, fmt.Errorf("invalid filter: %w", err)
+	}
 
 	var lobbies []Lobby
 	rows, err := s.DB.Query(ctx, `
-		SELECT code, peers, public, meta
+		WITH lobbies AS (
+			SELECT
+				code,
+				ARRAY_LENGTH(peers, 1) AS "playerCount",
+				public,
+				meta,
+				created_at AS "createdAt",
+				updated_at AS "updatedAt"
+			FROM lobbies
+			WHERE game = $1
+			AND public = true
+		)
+		SELECT *
 		FROM lobbies
-		WHERE game = $1
-		AND public = true
-		ORDER BY created_at DESC
+		WHERE `+where+`
+		ORDER BY "createdAt" DESC
 		LIMIT 50
-	`, game)
+	`, append([]any{game}, values...)...)
 	if err != nil {
 		return nil, err
 	}
 	defer rows.Close() //nolint:errcheck
 
 	for rows.Next() {
 		var lobby Lobby
-		var peers []string
-		err = rows.Scan(&lobby.Code, &peers, &lobby.Public, &lobby.CustomData)
+		err = rows.Scan(&lobby.Code, &lobby.PlayerCount, &lobby.Public, &lobby.CustomData, &lobby.CreatedAt, &lobby.UpdatedAt)
 		if err != nil {
 			return nil, err
 		}
-		lobby.PlayerCount = len(peers)
 		lobbies = append(lobbies, lobby)
 	}
 	if err = rows.Err(); err != nil {