Add filtering database API to syncapi account data table (#513)

syncapi/storage/account_data_table.go

@@ -18,7 +18,9 @@ import (
 	"context"
 	"database/sql"
 
+	"github.com/lib/pq"
 	"github.com/matrix-org/dendrite/common"
+	"github.com/matrix-org/gomatrixserverlib"
 )
 
 const accountDataSchema = `
@@ -41,7 +43,7 @@ CREATE TABLE IF NOT EXISTS syncapi_account_data_type (
     CONSTRAINT syncapi_account_data_unique UNIQUE (user_id, room_id, type)
 );
 
-CREATE UNIQUE INDEX IF NOT EXISTS syncapi_account_data_id_idx ON syncapi_account_data_type(id);
+CREATE UNIQUE INDEX IF NOT EXISTS syncapi_account_data_id_idx ON syncapi_account_data_type(id, type);
 `
 
 const insertAccountDataSQL = "" +
@@ -53,7 +55,9 @@ const insertAccountDataSQL = "" +
 const selectAccountDataInRangeSQL = "" +
 	"SELECT room_id, type FROM syncapi_account_data_type" +
 	" WHERE user_id = $1 AND id > $2 AND id <= $3" +
-	" ORDER BY id ASC"
+	" AND ( $4::text[] IS NULL OR     type LIKE ANY($4)  )" +
+	" AND ( $5::text[] IS NULL OR NOT(type LIKE ANY($5)) )" +
+	" ORDER BY id ASC LIMIT $6"
 
 const selectMaxAccountDataIDSQL = "" +
 	"SELECT MAX(id) FROM syncapi_account_data_type"
@@ -93,6 +97,7 @@ func (s *accountDataStatements) selectAccountDataInRange(
 	ctx context.Context,
 	userID string,
 	oldPos, newPos int64,
+	accountDataFilterPart *gomatrixserverlib.FilterPart,
 ) (data map[string][]string, err error) {
 	data = make(map[string][]string)
 
@@ -103,7 +108,11 @@ func (s *accountDataStatements) selectAccountDataInRange(
 		oldPos--
 	}
 
-	rows, err := s.selectAccountDataInRangeStmt.QueryContext(ctx, userID, oldPos, newPos)
+	rows, err := s.selectAccountDataInRangeStmt.QueryContext(ctx, userID, oldPos, newPos,
+		pq.StringArray(filterConvertTypeWildcardToSQL(accountDataFilterPart.Types)),
+		pq.StringArray(filterConvertTypeWildcardToSQL(accountDataFilterPart.NotTypes)),
+		accountDataFilterPart.Limit,
+	)
 	if err != nil {
 		return
 	}

syncapi/storage/syncserver.go

@@ -495,8 +495,9 @@ var txReadOnlySnapshot = sql.TxOptions{
 // If there was an issue with the retrieval, returns an error
 func (d *SyncServerDatasource) GetAccountDataInRange(
 	ctx context.Context, userID string, oldPos, newPos int64,
+	accountDataFilterPart *gomatrixserverlib.FilterPart,
 ) (map[string][]string, error) {
-	return d.accountData.selectAccountDataInRange(ctx, userID, oldPos, newPos)
+	return d.accountData.selectAccountDataInRange(ctx, userID, oldPos, newPos, accountDataFilterPart)
 }
 
 // UpsertAccountData keeps track of new or updated account data, by saving the type

syncapi/sync/requestpool.go

@@ -141,12 +141,14 @@ func (rp *RequestPool) currentSyncForUser(req syncRequest, latestPos types.SyncP
 		return
 	}
 
-	res, err = rp.appendAccountData(res, req.device.UserID, req, latestPos.PDUPosition)
+	accountDataFilter := gomatrixserverlib.DefaultFilterPart() // TODO: use filter provided in req instead
+	res, err = rp.appendAccountData(res, req.device.UserID, req, latestPos.PDUPosition, &accountDataFilter)
 	return
 }
 
 func (rp *RequestPool) appendAccountData(
 	data *types.Response, userID string, req syncRequest, currentPos int64,
+	accountDataFilter *gomatrixserverlib.FilterPart,
 ) (*types.Response, error) {
 	// TODO: Account data doesn't have a sync position of its own, meaning that
 	// account data might be sent multiple time to the client if multiple account
@@ -180,7 +182,7 @@ func (rp *RequestPool) appendAccountData(
 	}
 
 	// Sync is not initial, get all account data since the latest sync
-	dataTypes, err := rp.db.GetAccountDataInRange(req.ctx, userID, req.since.PDUPosition, currentPos)
+	dataTypes, err := rp.db.GetAccountDataInRange(req.ctx, userID, req.since.PDUPosition, currentPos, accountDataFilter)
 	if err != nil {
 		return nil, err
 	}