2020-02-13 18:27:33 +01:00
|
|
|
// Copyright 2017-2018 New Vector Ltd
|
|
|
|
// Copyright 2019-2020 The Matrix.org Foundation C.I.C.
|
|
|
|
//
|
|
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
// you may not use this file except in compliance with the License.
|
|
|
|
// You may obtain a copy of the License at
|
|
|
|
//
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
//
|
|
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
// See the License for the specific language governing permissions and
|
|
|
|
// limitations under the License.
|
|
|
|
|
|
|
|
package sqlite3
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"database/sql"
|
|
|
|
"encoding/json"
|
2021-01-19 19:00:42 +01:00
|
|
|
"fmt"
|
2020-02-20 10:28:03 +01:00
|
|
|
"strings"
|
2020-02-13 18:27:33 +01:00
|
|
|
|
2020-05-21 15:40:13 +02:00
|
|
|
"github.com/matrix-org/dendrite/internal"
|
2020-06-12 15:55:57 +02:00
|
|
|
"github.com/matrix-org/dendrite/internal/sqlutil"
|
2022-07-25 11:39:22 +02:00
|
|
|
"github.com/matrix-org/dendrite/syncapi/storage/sqlite3/deltas"
|
2020-05-14 17:11:37 +02:00
|
|
|
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
2020-02-13 18:27:33 +01:00
|
|
|
"github.com/matrix-org/dendrite/syncapi/types"
|
|
|
|
"github.com/matrix-org/gomatrixserverlib"
|
|
|
|
)
|
|
|
|
|
|
|
|
const currentRoomStateSchema = `
|
|
|
|
-- Stores the current room state for every room.
|
|
|
|
CREATE TABLE IF NOT EXISTS syncapi_current_room_state (
|
|
|
|
room_id TEXT NOT NULL,
|
|
|
|
event_id TEXT NOT NULL,
|
|
|
|
type TEXT NOT NULL,
|
|
|
|
sender TEXT NOT NULL,
|
|
|
|
contains_url BOOL NOT NULL DEFAULT false,
|
|
|
|
state_key TEXT NOT NULL,
|
2020-03-19 13:07:01 +01:00
|
|
|
headered_event_json TEXT NOT NULL,
|
2020-02-13 18:27:33 +01:00
|
|
|
membership TEXT,
|
|
|
|
added_at BIGINT,
|
2022-07-18 14:46:15 +02:00
|
|
|
history_visibility SMALLINT NOT NULL DEFAULT 2, -- The history visibility before this event (1 - world_readable; 2 - shared; 3 - invited; 4 - joined)
|
2020-02-13 18:27:33 +01:00
|
|
|
UNIQUE (room_id, type, state_key)
|
|
|
|
);
|
|
|
|
-- for event deletion
|
|
|
|
CREATE UNIQUE INDEX IF NOT EXISTS syncapi_event_id_idx ON syncapi_current_room_state(event_id, room_id, type, sender, contains_url);
|
|
|
|
-- for querying membership states of users
|
|
|
|
-- CREATE INDEX IF NOT EXISTS syncapi_membership_idx ON syncapi_current_room_state(type, state_key, membership) WHERE membership IS NOT NULL AND membership != 'leave';
|
2020-12-16 19:16:39 +01:00
|
|
|
-- for querying state by event IDs
|
|
|
|
CREATE UNIQUE INDEX IF NOT EXISTS syncapi_current_room_state_eventid_idx ON syncapi_current_room_state(event_id);
|
Improve selectRoomIDsWithAnyMembershipSQL performance (#2738)
Recently I have observed that dendrite spends a lot of time (~390s) in
`selectRoomIDsWithAnyMembershipSQL` query
```
dendrite_syncapi=# select total_exec_time, left(query,100) from pg_stat_statements order by total_exec_time desc limit 5 ;
total_exec_time | left
--------------------+------------------------------------------------------------------------------------------------------
747826.5800519128 | SELECT event_id, id, headered_event_json, session_id, exclude_from_sync, transaction_id, history_vis
389130.5490339942 | SELECT DISTINCT room_id, membership FROM syncapi_current_room_state WHERE type = $2 AND state_key =
376104.17514700035 | SELECT psd.datname, xact_commit, xact_rollback, blks_read, blks_hit, tup_returned, tup_fetched, tup_
363644.164092031 | SELECT event_type_nid, event_state_key_nid, event_nid FROM roomserver_events WHERE event_nid = ANY($
58570.48104699995 | SELECT event_id, headered_event_json FROM syncapi_current_room_state WHERE room_id = $1 AND ( $2::te
(5 rows)
```
Explain analyze showed correct usage of `syncapi_room_state_unique`
index:
```
dendrite_syncapi=#
explain analyze SELECT distinct room_id, membership FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = '@qjfl:dendrite.stg.globekeeper.com';
QUERY PLAN
------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Unique (cost=2749.38..2749.56 rows=24 width=52) (actual time=2.933..2.956 rows=65 loops=1)
-> Sort (cost=2749.38..2749.44 rows=24 width=52) (actual time=2.932..2.937 rows=65 loops=1)
Sort Key: room_id, membership
Sort Method: quicksort Memory: 34kB
-> Index Scan using syncapi_room_state_unique on syncapi_current_room_state (cost=0.41..2748.83 rows=24 width=52) (actual time=0.030..2.890 rows=65 loops=1)
Index Cond: ((type = 'm.room.member'::text) AND (state_key = '@qjfl:dendrite.stg.globekeeper.com'::text))
Planning Time: 0.140 ms
Execution Time: 2.990 ms
(8 rows)
```
Multi-column indexes in Postgres shall perform well for leftmost
columns, but I gave it a try and created
`syncapi_current_room_state_type_state_key_idx` index. I could observe
significant performance improvement. Execution time dropped from 2.9 ms
to 0.24 ms:
```
explain analyze SELECT distinct room_id, membership FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = '@qjfl:dendrite.stg.globekeeper.com';
QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------------------------------------------
Unique (cost=96.46..96.64 rows=24 width=52) (actual time=0.199..0.218 rows=65 loops=1)
-> Sort (cost=96.46..96.52 rows=24 width=52) (actual time=0.199..0.202 rows=65 loops=1)
Sort Key: room_id, membership
Sort Method: quicksort Memory: 34kB
-> Bitmap Heap Scan on syncapi_current_room_state (cost=4.53..95.91 rows=24 width=52) (actual time=0.048..0.139 rows=65 loops=1)
Recheck Cond: ((type = 'm.room.member'::text) AND (state_key = '@qjfl:dendrite.stg.globekeeper.com'::text))
Heap Blocks: exact=59
-> Bitmap Index Scan on syncapi_current_room_state_type_state_key_idx (cost=0.00..4.53 rows=24 width=0) (actual time=0.037..0.037 rows=65 loops=1)
Index Cond: ((type = 'm.room.member'::text) AND (state_key = '@qjfl:dendrite.stg.globekeeper.com'::text))
Planning Time: 0.236 ms
Execution Time: 0.242 ms
(11 rows)
```
Next improvement is skipping DISTINCT and rely on map assignment in
`SelectRoomIDsWithAnyMembership`. Execution time drops by almost half:
```
explain analyze SELECT room_id, membership FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = '@qjfl:dendrite.stg.globekeeper.com';
QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------------------------------
Bitmap Heap Scan on syncapi_current_room_state (cost=4.53..95.91 rows=24 width=52) (actual time=0.032..0.113 rows=65 loops=1)
Recheck Cond: ((type = 'm.room.member'::text) AND (state_key = '@qjfl:dendrite.stg.globekeeper.com'::text))
Heap Blocks: exact=59
-> Bitmap Index Scan on syncapi_current_room_state_type_state_key_idx (cost=0.00..4.53 rows=24 width=0) (actual time=0.021..0.021 rows=65 loops=1)
Index Cond: ((type = 'm.room.member'::text) AND (state_key = '@qjfl:dendrite.stg.globekeeper.com'::text))
Planning Time: 0.087 ms
Execution Time: 0.136 ms
(7 rows)
```
In our env we spend only 1s on inserting to table, so the write penalty
of creating an index should be small.
```
dendrite_syncapi=# select total_exec_time, left(query,100) from pg_stat_statements where query like '%INSERT%syncapi_current_room_state%' order by total_exec_time desc;
total_exec_time | left
--------------------+------------------------------------------------------------------------------------------------------
1139.9057619999971 | INSERT INTO syncapi_current_room_state (room_id, event_id, type, sender, contains_url, state_key, he
(1 row)
```
This PR does not require test modifications.
### Pull Request Checklist
<!-- Please read docs/CONTRIBUTING.md before submitting your pull
request -->
* [x] I have added added tests for PR _or_ I have justified why this PR
doesn't need tests.
* [x] Pull request includes a [sign
off](https://github.com/matrix-org/dendrite/blob/main/docs/CONTRIBUTING.md#sign-off)
Signed-off-by: `Piotr Kozimor <p1996k@gmail.com>`
2022-09-27 10:41:36 +02:00
|
|
|
-- for improving selectRoomIDsWithAnyMembershipSQL
|
|
|
|
CREATE INDEX IF NOT EXISTS syncapi_current_room_state_type_state_key_idx ON syncapi_current_room_state(type, state_key);
|
2020-02-13 18:27:33 +01:00
|
|
|
`
|
|
|
|
|
|
|
|
const upsertRoomStateSQL = "" +
|
2022-07-18 14:46:15 +02:00
|
|
|
"INSERT INTO syncapi_current_room_state (room_id, event_id, type, sender, contains_url, state_key, headered_event_json, membership, added_at, history_visibility)" +
|
|
|
|
" VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)" +
|
2020-09-15 12:17:46 +02:00
|
|
|
" ON CONFLICT (room_id, type, state_key)" +
|
2020-03-19 13:07:01 +01:00
|
|
|
" DO UPDATE SET event_id = $2, sender=$4, contains_url=$5, headered_event_json = $7, membership = $8, added_at = $9"
|
2020-02-13 18:27:33 +01:00
|
|
|
|
|
|
|
const deleteRoomStateByEventIDSQL = "" +
|
|
|
|
"DELETE FROM syncapi_current_room_state WHERE event_id = $1"
|
|
|
|
|
2022-03-16 11:25:50 +01:00
|
|
|
const deleteRoomStateForRoomSQL = "" +
|
|
|
|
"DELETE FROM syncapi_current_room_state WHERE room_id = $1"
|
2020-09-15 12:17:46 +02:00
|
|
|
|
2020-02-13 18:27:33 +01:00
|
|
|
const selectRoomIDsWithMembershipSQL = "" +
|
2020-08-21 10:57:52 +02:00
|
|
|
"SELECT DISTINCT room_id FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = $1 AND membership = $2"
|
2020-02-13 18:27:33 +01:00
|
|
|
|
2022-03-11 13:48:45 +01:00
|
|
|
const selectRoomIDsWithAnyMembershipSQL = "" +
|
Improve selectRoomIDsWithAnyMembershipSQL performance (#2738)
Recently I have observed that dendrite spends a lot of time (~390s) in
`selectRoomIDsWithAnyMembershipSQL` query
```
dendrite_syncapi=# select total_exec_time, left(query,100) from pg_stat_statements order by total_exec_time desc limit 5 ;
total_exec_time | left
--------------------+------------------------------------------------------------------------------------------------------
747826.5800519128 | SELECT event_id, id, headered_event_json, session_id, exclude_from_sync, transaction_id, history_vis
389130.5490339942 | SELECT DISTINCT room_id, membership FROM syncapi_current_room_state WHERE type = $2 AND state_key =
376104.17514700035 | SELECT psd.datname, xact_commit, xact_rollback, blks_read, blks_hit, tup_returned, tup_fetched, tup_
363644.164092031 | SELECT event_type_nid, event_state_key_nid, event_nid FROM roomserver_events WHERE event_nid = ANY($
58570.48104699995 | SELECT event_id, headered_event_json FROM syncapi_current_room_state WHERE room_id = $1 AND ( $2::te
(5 rows)
```
Explain analyze showed correct usage of `syncapi_room_state_unique`
index:
```
dendrite_syncapi=#
explain analyze SELECT distinct room_id, membership FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = '@qjfl:dendrite.stg.globekeeper.com';
QUERY PLAN
------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Unique (cost=2749.38..2749.56 rows=24 width=52) (actual time=2.933..2.956 rows=65 loops=1)
-> Sort (cost=2749.38..2749.44 rows=24 width=52) (actual time=2.932..2.937 rows=65 loops=1)
Sort Key: room_id, membership
Sort Method: quicksort Memory: 34kB
-> Index Scan using syncapi_room_state_unique on syncapi_current_room_state (cost=0.41..2748.83 rows=24 width=52) (actual time=0.030..2.890 rows=65 loops=1)
Index Cond: ((type = 'm.room.member'::text) AND (state_key = '@qjfl:dendrite.stg.globekeeper.com'::text))
Planning Time: 0.140 ms
Execution Time: 2.990 ms
(8 rows)
```
Multi-column indexes in Postgres shall perform well for leftmost
columns, but I gave it a try and created
`syncapi_current_room_state_type_state_key_idx` index. I could observe
significant performance improvement. Execution time dropped from 2.9 ms
to 0.24 ms:
```
explain analyze SELECT distinct room_id, membership FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = '@qjfl:dendrite.stg.globekeeper.com';
QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------------------------------------------
Unique (cost=96.46..96.64 rows=24 width=52) (actual time=0.199..0.218 rows=65 loops=1)
-> Sort (cost=96.46..96.52 rows=24 width=52) (actual time=0.199..0.202 rows=65 loops=1)
Sort Key: room_id, membership
Sort Method: quicksort Memory: 34kB
-> Bitmap Heap Scan on syncapi_current_room_state (cost=4.53..95.91 rows=24 width=52) (actual time=0.048..0.139 rows=65 loops=1)
Recheck Cond: ((type = 'm.room.member'::text) AND (state_key = '@qjfl:dendrite.stg.globekeeper.com'::text))
Heap Blocks: exact=59
-> Bitmap Index Scan on syncapi_current_room_state_type_state_key_idx (cost=0.00..4.53 rows=24 width=0) (actual time=0.037..0.037 rows=65 loops=1)
Index Cond: ((type = 'm.room.member'::text) AND (state_key = '@qjfl:dendrite.stg.globekeeper.com'::text))
Planning Time: 0.236 ms
Execution Time: 0.242 ms
(11 rows)
```
Next improvement is skipping DISTINCT and rely on map assignment in
`SelectRoomIDsWithAnyMembership`. Execution time drops by almost half:
```
explain analyze SELECT room_id, membership FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = '@qjfl:dendrite.stg.globekeeper.com';
QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------------------------------
Bitmap Heap Scan on syncapi_current_room_state (cost=4.53..95.91 rows=24 width=52) (actual time=0.032..0.113 rows=65 loops=1)
Recheck Cond: ((type = 'm.room.member'::text) AND (state_key = '@qjfl:dendrite.stg.globekeeper.com'::text))
Heap Blocks: exact=59
-> Bitmap Index Scan on syncapi_current_room_state_type_state_key_idx (cost=0.00..4.53 rows=24 width=0) (actual time=0.021..0.021 rows=65 loops=1)
Index Cond: ((type = 'm.room.member'::text) AND (state_key = '@qjfl:dendrite.stg.globekeeper.com'::text))
Planning Time: 0.087 ms
Execution Time: 0.136 ms
(7 rows)
```
In our env we spend only 1s on inserting to table, so the write penalty
of creating an index should be small.
```
dendrite_syncapi=# select total_exec_time, left(query,100) from pg_stat_statements where query like '%INSERT%syncapi_current_room_state%' order by total_exec_time desc;
total_exec_time | left
--------------------+------------------------------------------------------------------------------------------------------
1139.9057619999971 | INSERT INTO syncapi_current_room_state (room_id, event_id, type, sender, contains_url, state_key, he
(1 row)
```
This PR does not require test modifications.
### Pull Request Checklist
<!-- Please read docs/CONTRIBUTING.md before submitting your pull
request -->
* [x] I have added added tests for PR _or_ I have justified why this PR
doesn't need tests.
* [x] Pull request includes a [sign
off](https://github.com/matrix-org/dendrite/blob/main/docs/CONTRIBUTING.md#sign-off)
Signed-off-by: `Piotr Kozimor <p1996k@gmail.com>`
2022-09-27 10:41:36 +02:00
|
|
|
"SELECT room_id, membership FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = $1"
|
2022-03-11 13:48:45 +01:00
|
|
|
|
2020-02-13 18:27:33 +01:00
|
|
|
const selectCurrentStateSQL = "" +
|
2021-01-19 19:00:42 +01:00
|
|
|
"SELECT event_id, headered_event_json FROM syncapi_current_room_state WHERE room_id = $1"
|
2022-02-21 17:12:22 +01:00
|
|
|
|
|
|
|
// WHEN, ORDER BY and LIMIT will be added by prepareWithFilter
|
2020-02-13 18:27:33 +01:00
|
|
|
|
|
|
|
const selectJoinedUsersSQL = "" +
|
|
|
|
"SELECT room_id, state_key FROM syncapi_current_room_state WHERE type = 'm.room.member' AND membership = 'join'"
|
|
|
|
|
2022-04-28 18:53:28 +02:00
|
|
|
const selectJoinedUsersInRoomSQL = "" +
|
|
|
|
"SELECT room_id, state_key FROM syncapi_current_room_state WHERE type = 'm.room.member' AND membership = 'join' AND room_id IN ($1)"
|
|
|
|
|
2020-02-13 18:27:33 +01:00
|
|
|
const selectStateEventSQL = "" +
|
2020-03-19 13:07:01 +01:00
|
|
|
"SELECT headered_event_json FROM syncapi_current_room_state WHERE room_id = $1 AND type = $2 AND state_key = $3"
|
2020-02-13 18:27:33 +01:00
|
|
|
|
|
|
|
const selectEventsWithEventIDsSQL = "" +
|
2022-07-18 14:46:15 +02:00
|
|
|
// TODO: The session_id and transaction_id blanks are here because
|
|
|
|
// the rowsToStreamEvents expects there to be exactly seven columns. We need to
|
2020-02-13 18:27:33 +01:00
|
|
|
// figure out if these really need to be in the DB, and if so, we need a
|
|
|
|
// better permanent fix for this. - neilalexander, 2 Jan 2020
|
2022-07-18 14:46:15 +02:00
|
|
|
"SELECT event_id, added_at, headered_event_json, 0 AS session_id, false AS exclude_from_sync, '' AS transaction_id, history_visibility" +
|
2020-02-13 18:27:33 +01:00
|
|
|
" FROM syncapi_current_room_state WHERE event_id IN ($1)"
|
|
|
|
|
2022-07-15 17:25:26 +02:00
|
|
|
const selectSharedUsersSQL = "" +
|
2022-08-03 18:35:17 +02:00
|
|
|
"SELECT state_key FROM syncapi_current_room_state WHERE room_id IN(" +
|
2022-09-09 14:50:50 +02:00
|
|
|
" SELECT DISTINCT room_id FROM syncapi_current_room_state WHERE state_key = $1 AND membership='join'" +
|
2022-09-09 15:18:45 +02:00
|
|
|
") AND type = 'm.room.member' AND state_key IN ($2) AND membership IN ('join', 'invite');"
|
2022-07-15 17:25:26 +02:00
|
|
|
|
2020-02-13 18:27:33 +01:00
|
|
|
type currentRoomStateStatements struct {
|
2022-03-11 13:48:45 +01:00
|
|
|
db *sql.DB
|
2022-04-08 18:53:24 +02:00
|
|
|
streamIDStatements *StreamIDStatements
|
2022-03-11 13:48:45 +01:00
|
|
|
upsertRoomStateStmt *sql.Stmt
|
|
|
|
deleteRoomStateByEventIDStmt *sql.Stmt
|
2022-03-16 11:25:50 +01:00
|
|
|
deleteRoomStateForRoomStmt *sql.Stmt
|
2022-03-11 13:48:45 +01:00
|
|
|
selectRoomIDsWithMembershipStmt *sql.Stmt
|
|
|
|
selectRoomIDsWithAnyMembershipStmt *sql.Stmt
|
|
|
|
selectJoinedUsersStmt *sql.Stmt
|
2022-07-15 17:25:26 +02:00
|
|
|
//selectJoinedUsersInRoomStmt *sql.Stmt - prepared at runtime due to variadic
|
2022-04-28 18:53:28 +02:00
|
|
|
selectStateEventStmt *sql.Stmt
|
2022-07-15 17:25:26 +02:00
|
|
|
//selectSharedUsersSQL *sql.Stmt - prepared at runtime due to variadic
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
|
|
|
|
2022-04-08 18:53:24 +02:00
|
|
|
func NewSqliteCurrentRoomStateTable(db *sql.DB, streamID *StreamIDStatements) (tables.CurrentRoomState, error) {
|
2020-05-14 17:11:37 +02:00
|
|
|
s := ¤tRoomStateStatements{
|
2020-07-21 16:48:21 +02:00
|
|
|
db: db,
|
2020-05-14 17:11:37 +02:00
|
|
|
streamIDStatements: streamID,
|
|
|
|
}
|
|
|
|
_, err := db.Exec(currentRoomStateSchema)
|
2020-02-13 18:27:33 +01:00
|
|
|
if err != nil {
|
2020-05-14 17:11:37 +02:00
|
|
|
return nil, err
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
2022-07-25 11:39:22 +02:00
|
|
|
|
|
|
|
m := sqlutil.NewMigrator(db)
|
|
|
|
m.AddMigrations(sqlutil.Migration{
|
|
|
|
Version: "syncapi: add history visibility column (current_room_state)",
|
|
|
|
Up: deltas.UpAddHistoryVisibilityColumnCurrentRoomState,
|
|
|
|
})
|
|
|
|
err = m.Up(context.Background())
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2020-02-13 18:27:33 +01:00
|
|
|
if s.upsertRoomStateStmt, err = db.Prepare(upsertRoomStateSQL); err != nil {
|
2020-05-14 17:11:37 +02:00
|
|
|
return nil, err
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
|
|
|
if s.deleteRoomStateByEventIDStmt, err = db.Prepare(deleteRoomStateByEventIDSQL); err != nil {
|
2020-05-14 17:11:37 +02:00
|
|
|
return nil, err
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
2022-03-16 11:25:50 +01:00
|
|
|
if s.deleteRoomStateForRoomStmt, err = db.Prepare(deleteRoomStateForRoomSQL); err != nil {
|
2020-09-15 12:17:46 +02:00
|
|
|
return nil, err
|
|
|
|
}
|
2020-02-13 18:27:33 +01:00
|
|
|
if s.selectRoomIDsWithMembershipStmt, err = db.Prepare(selectRoomIDsWithMembershipSQL); err != nil {
|
2020-05-14 17:11:37 +02:00
|
|
|
return nil, err
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
2022-03-11 13:48:45 +01:00
|
|
|
if s.selectRoomIDsWithAnyMembershipStmt, err = db.Prepare(selectRoomIDsWithAnyMembershipSQL); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2020-02-13 18:27:33 +01:00
|
|
|
if s.selectJoinedUsersStmt, err = db.Prepare(selectJoinedUsersSQL); err != nil {
|
2020-05-14 17:11:37 +02:00
|
|
|
return nil, err
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
2022-04-28 18:53:28 +02:00
|
|
|
//if s.selectJoinedUsersInRoomStmt, err = db.Prepare(selectJoinedUsersInRoomSQL); err != nil {
|
|
|
|
// return nil, err
|
|
|
|
//}
|
2020-02-13 18:27:33 +01:00
|
|
|
if s.selectStateEventStmt, err = db.Prepare(selectStateEventSQL); err != nil {
|
2020-05-14 17:11:37 +02:00
|
|
|
return nil, err
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
2020-05-14 17:11:37 +02:00
|
|
|
return s, nil
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
|
|
|
|
2022-04-28 18:53:28 +02:00
|
|
|
// SelectJoinedUsers returns a map of room ID to a list of joined user IDs.
|
2020-05-14 17:11:37 +02:00
|
|
|
func (s *currentRoomStateStatements) SelectJoinedUsers(
|
2020-02-13 18:27:33 +01:00
|
|
|
ctx context.Context,
|
|
|
|
) (map[string][]string, error) {
|
|
|
|
rows, err := s.selectJoinedUsersStmt.QueryContext(ctx)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2020-05-21 15:40:13 +02:00
|
|
|
defer internal.CloseAndLogIfError(ctx, rows, "selectJoinedUsers: rows.close() failed")
|
2020-02-13 18:27:33 +01:00
|
|
|
|
|
|
|
result := make(map[string][]string)
|
2022-04-28 18:53:28 +02:00
|
|
|
var roomID string
|
|
|
|
var userID string
|
2020-02-13 18:27:33 +01:00
|
|
|
for rows.Next() {
|
|
|
|
if err := rows.Scan(&roomID, &userID); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
users := result[roomID]
|
|
|
|
users = append(users, userID)
|
|
|
|
result[roomID] = users
|
|
|
|
}
|
|
|
|
return result, nil
|
|
|
|
}
|
|
|
|
|
2022-04-28 18:53:28 +02:00
|
|
|
// SelectJoinedUsersInRoom returns a map of room ID to a list of joined user IDs for a given room.
|
|
|
|
func (s *currentRoomStateStatements) SelectJoinedUsersInRoom(
|
|
|
|
ctx context.Context, roomIDs []string,
|
|
|
|
) (map[string][]string, error) {
|
|
|
|
query := strings.Replace(selectJoinedUsersInRoomSQL, "($1)", sqlutil.QueryVariadic(len(roomIDs)), 1)
|
|
|
|
params := make([]interface{}, 0, len(roomIDs))
|
|
|
|
for _, roomID := range roomIDs {
|
|
|
|
params = append(params, roomID)
|
|
|
|
}
|
|
|
|
stmt, err := s.db.Prepare(query)
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("SelectJoinedUsersInRoom s.db.Prepare: %w", err)
|
|
|
|
}
|
|
|
|
defer internal.CloseAndLogIfError(ctx, stmt, "SelectJoinedUsersInRoom: stmt.close() failed")
|
|
|
|
|
|
|
|
rows, err := stmt.QueryContext(ctx, params...)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
defer internal.CloseAndLogIfError(ctx, rows, "SelectJoinedUsersInRoom: rows.close() failed")
|
|
|
|
|
|
|
|
result := make(map[string][]string)
|
|
|
|
var userID, roomID string
|
|
|
|
for rows.Next() {
|
|
|
|
if err := rows.Scan(&roomID, &userID); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
users := result[roomID]
|
|
|
|
users = append(users, userID)
|
|
|
|
result[roomID] = users
|
|
|
|
}
|
|
|
|
return result, rows.Err()
|
|
|
|
}
|
|
|
|
|
2020-02-13 18:27:33 +01:00
|
|
|
// SelectRoomIDsWithMembership returns the list of room IDs which have the given user in the given membership state.
|
2020-05-14 17:11:37 +02:00
|
|
|
func (s *currentRoomStateStatements) SelectRoomIDsWithMembership(
|
2020-02-13 18:27:33 +01:00
|
|
|
ctx context.Context,
|
|
|
|
txn *sql.Tx,
|
|
|
|
userID string,
|
|
|
|
membership string, // nolint: unparam
|
|
|
|
) ([]string, error) {
|
2020-06-12 15:55:57 +02:00
|
|
|
stmt := sqlutil.TxStmt(txn, s.selectRoomIDsWithMembershipStmt)
|
2020-02-13 18:27:33 +01:00
|
|
|
rows, err := stmt.QueryContext(ctx, userID, membership)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2020-05-21 15:40:13 +02:00
|
|
|
defer internal.CloseAndLogIfError(ctx, rows, "selectRoomIDsWithMembership: rows.close() failed")
|
2020-02-13 18:27:33 +01:00
|
|
|
|
|
|
|
var result []string
|
|
|
|
for rows.Next() {
|
|
|
|
var roomID string
|
|
|
|
if err := rows.Scan(&roomID); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
result = append(result, roomID)
|
|
|
|
}
|
|
|
|
return result, nil
|
|
|
|
}
|
|
|
|
|
2022-03-11 13:48:45 +01:00
|
|
|
// SelectRoomIDsWithAnyMembership returns a map of all memberships for the given user.
|
|
|
|
func (s *currentRoomStateStatements) SelectRoomIDsWithAnyMembership(
|
|
|
|
ctx context.Context,
|
|
|
|
txn *sql.Tx,
|
|
|
|
userID string,
|
|
|
|
) (map[string]string, error) {
|
|
|
|
stmt := sqlutil.TxStmt(txn, s.selectRoomIDsWithAnyMembershipStmt)
|
|
|
|
rows, err := stmt.QueryContext(ctx, userID)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
defer internal.CloseAndLogIfError(ctx, rows, "selectRoomIDsWithAnyMembership: rows.close() failed")
|
|
|
|
|
|
|
|
result := map[string]string{}
|
|
|
|
for rows.Next() {
|
|
|
|
var roomID string
|
|
|
|
var membership string
|
|
|
|
if err := rows.Scan(&roomID, &membership); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
result[roomID] = membership
|
|
|
|
}
|
|
|
|
return result, rows.Err()
|
|
|
|
}
|
|
|
|
|
2020-02-13 18:27:33 +01:00
|
|
|
// CurrentState returns all the current state events for the given room.
|
2020-05-14 17:11:37 +02:00
|
|
|
func (s *currentRoomStateStatements) SelectCurrentState(
|
2020-02-13 18:27:33 +01:00
|
|
|
ctx context.Context, txn *sql.Tx, roomID string,
|
2021-01-19 19:00:42 +01:00
|
|
|
stateFilter *gomatrixserverlib.StateFilter,
|
2021-02-04 13:20:37 +01:00
|
|
|
excludeEventIDs []string,
|
2020-11-16 16:44:53 +01:00
|
|
|
) ([]*gomatrixserverlib.HeaderedEvent, error) {
|
2021-01-19 19:00:42 +01:00
|
|
|
stmt, params, err := prepareWithFilters(
|
|
|
|
s.db, txn, selectCurrentStateSQL,
|
|
|
|
[]interface{}{
|
|
|
|
roomID,
|
|
|
|
},
|
|
|
|
stateFilter.Senders, stateFilter.NotSenders,
|
|
|
|
stateFilter.Types, stateFilter.NotTypes,
|
2022-04-13 13:16:02 +02:00
|
|
|
excludeEventIDs, stateFilter.ContainsURL, stateFilter.Limit, FilterOrderNone,
|
2020-02-13 18:27:33 +01:00
|
|
|
)
|
2021-01-19 19:00:42 +01:00
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("s.prepareWithFilters: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
rows, err := stmt.QueryContext(ctx, params...)
|
2020-02-13 18:27:33 +01:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2020-05-21 15:40:13 +02:00
|
|
|
defer internal.CloseAndLogIfError(ctx, rows, "selectCurrentState: rows.close() failed")
|
2020-02-13 18:27:33 +01:00
|
|
|
|
|
|
|
return rowsToEvents(rows)
|
|
|
|
}
|
|
|
|
|
2020-05-14 17:11:37 +02:00
|
|
|
func (s *currentRoomStateStatements) DeleteRoomStateByEventID(
|
2020-02-13 18:27:33 +01:00
|
|
|
ctx context.Context, txn *sql.Tx, eventID string,
|
|
|
|
) error {
|
2020-08-21 11:42:08 +02:00
|
|
|
stmt := sqlutil.TxStmt(txn, s.deleteRoomStateByEventIDStmt)
|
|
|
|
_, err := stmt.ExecContext(ctx, eventID)
|
|
|
|
return err
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
|
|
|
|
2020-09-15 12:17:46 +02:00
|
|
|
func (s *currentRoomStateStatements) DeleteRoomStateForRoom(
|
|
|
|
ctx context.Context, txn *sql.Tx, roomID string,
|
|
|
|
) error {
|
2022-03-16 11:25:50 +01:00
|
|
|
stmt := sqlutil.TxStmt(txn, s.deleteRoomStateForRoomStmt)
|
2020-09-15 12:17:46 +02:00
|
|
|
_, err := stmt.ExecContext(ctx, roomID)
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2020-05-14 17:11:37 +02:00
|
|
|
func (s *currentRoomStateStatements) UpsertRoomState(
|
2020-02-13 18:27:33 +01:00
|
|
|
ctx context.Context, txn *sql.Tx,
|
2020-11-16 16:44:53 +01:00
|
|
|
event *gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition,
|
2020-02-13 18:27:33 +01:00
|
|
|
) error {
|
|
|
|
// Parse content as JSON and search for an "url" key
|
|
|
|
containsURL := false
|
|
|
|
var content map[string]interface{}
|
|
|
|
if json.Unmarshal(event.Content(), &content) != nil {
|
|
|
|
// Set containsURL to true if url is present
|
|
|
|
_, containsURL = content["url"]
|
|
|
|
}
|
|
|
|
|
2020-03-19 13:07:01 +01:00
|
|
|
headeredJSON, err := json.Marshal(event)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2020-02-13 18:27:33 +01:00
|
|
|
// upsert state event
|
2020-08-21 11:42:08 +02:00
|
|
|
stmt := sqlutil.TxStmt(txn, s.upsertRoomStateStmt)
|
|
|
|
_, err = stmt.ExecContext(
|
|
|
|
ctx,
|
|
|
|
event.RoomID(),
|
|
|
|
event.EventID(),
|
|
|
|
event.Type(),
|
|
|
|
event.Sender(),
|
|
|
|
containsURL,
|
|
|
|
*event.StateKey(),
|
|
|
|
headeredJSON,
|
|
|
|
membership,
|
|
|
|
addedAt,
|
2022-07-18 14:46:15 +02:00
|
|
|
event.Visibility,
|
2020-08-21 11:42:08 +02:00
|
|
|
)
|
|
|
|
return err
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
|
|
|
|
2020-07-27 14:19:30 +02:00
|
|
|
func minOfInts(a, b int) int {
|
|
|
|
if a <= b {
|
|
|
|
return a
|
|
|
|
}
|
|
|
|
return b
|
|
|
|
}
|
|
|
|
|
2020-05-14 17:11:37 +02:00
|
|
|
func (s *currentRoomStateStatements) SelectEventsWithEventIDs(
|
2020-02-13 18:27:33 +01:00
|
|
|
ctx context.Context, txn *sql.Tx, eventIDs []string,
|
|
|
|
) ([]types.StreamEvent, error) {
|
2020-02-20 10:28:03 +01:00
|
|
|
iEventIDs := make([]interface{}, len(eventIDs))
|
|
|
|
for k, v := range eventIDs {
|
|
|
|
iEventIDs[k] = v
|
|
|
|
}
|
2020-07-27 14:19:30 +02:00
|
|
|
res := make([]types.StreamEvent, 0, len(eventIDs))
|
|
|
|
var start int
|
|
|
|
for start < len(eventIDs) {
|
|
|
|
n := minOfInts(len(eventIDs)-start, 999)
|
|
|
|
query := strings.Replace(selectEventsWithEventIDsSQL, "($1)", sqlutil.QueryVariadic(n), 1)
|
|
|
|
rows, err := txn.QueryContext(ctx, query, iEventIDs[start:start+n]...)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
start = start + n
|
|
|
|
events, err := rowsToStreamEvents(rows)
|
|
|
|
internal.CloseAndLogIfError(ctx, rows, "selectEventsWithEventIDs: rows.close() failed")
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
res = append(res, events...)
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
2020-07-27 14:19:30 +02:00
|
|
|
return res, nil
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
|
|
|
|
2020-11-16 16:44:53 +01:00
|
|
|
func rowsToEvents(rows *sql.Rows) ([]*gomatrixserverlib.HeaderedEvent, error) {
|
|
|
|
result := []*gomatrixserverlib.HeaderedEvent{}
|
2020-02-13 18:27:33 +01:00
|
|
|
for rows.Next() {
|
2020-12-09 19:07:17 +01:00
|
|
|
var eventID string
|
2020-02-13 18:27:33 +01:00
|
|
|
var eventBytes []byte
|
2020-12-09 19:07:17 +01:00
|
|
|
if err := rows.Scan(&eventID, &eventBytes); err != nil {
|
2020-02-13 18:27:33 +01:00
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
// TODO: Handle redacted events
|
2020-03-19 13:07:01 +01:00
|
|
|
var ev gomatrixserverlib.HeaderedEvent
|
2020-12-09 19:07:17 +01:00
|
|
|
if err := ev.UnmarshalJSONWithEventID(eventBytes, eventID); err != nil {
|
2020-02-13 18:27:33 +01:00
|
|
|
return nil, err
|
|
|
|
}
|
2020-11-16 16:44:53 +01:00
|
|
|
result = append(result, &ev)
|
2020-02-13 18:27:33 +01:00
|
|
|
}
|
|
|
|
return result, nil
|
|
|
|
}
|
|
|
|
|
2020-05-14 17:11:37 +02:00
|
|
|
func (s *currentRoomStateStatements) SelectStateEvent(
|
2020-02-13 18:27:33 +01:00
|
|
|
ctx context.Context, roomID, evType, stateKey string,
|
2020-03-19 13:07:01 +01:00
|
|
|
) (*gomatrixserverlib.HeaderedEvent, error) {
|
2020-02-13 18:27:33 +01:00
|
|
|
stmt := s.selectStateEventStmt
|
|
|
|
var res []byte
|
|
|
|
err := stmt.QueryRowContext(ctx, roomID, evType, stateKey).Scan(&res)
|
|
|
|
if err == sql.ErrNoRows {
|
|
|
|
return nil, nil
|
|
|
|
}
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2020-03-19 13:07:01 +01:00
|
|
|
var ev gomatrixserverlib.HeaderedEvent
|
|
|
|
if err = json.Unmarshal(res, &ev); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2020-02-13 18:27:33 +01:00
|
|
|
return &ev, err
|
|
|
|
}
|
2022-07-15 17:25:26 +02:00
|
|
|
|
|
|
|
func (s *currentRoomStateStatements) SelectSharedUsers(
|
|
|
|
ctx context.Context, txn *sql.Tx, userID string, otherUserIDs []string,
|
|
|
|
) ([]string, error) {
|
2022-08-03 18:35:17 +02:00
|
|
|
|
|
|
|
params := make([]interface{}, len(otherUserIDs)+1)
|
|
|
|
params[0] = userID
|
|
|
|
for k, v := range otherUserIDs {
|
|
|
|
params[k+1] = v
|
2022-07-15 17:25:26 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
result := make([]string, 0, len(otherUserIDs))
|
2022-08-03 18:35:17 +02:00
|
|
|
query := strings.Replace(selectSharedUsersSQL, "($2)", sqlutil.QueryVariadicOffset(len(otherUserIDs), 1), 1)
|
|
|
|
err := sqlutil.RunLimitedVariablesQuery(
|
|
|
|
ctx, query, s.db, params, sqlutil.SQLite3MaxVariables,
|
|
|
|
func(rows *sql.Rows) error {
|
|
|
|
var stateKey string
|
|
|
|
for rows.Next() {
|
|
|
|
if err := rows.Scan(&stateKey); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
result = append(result, stateKey)
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
return result, err
|
2022-07-15 17:25:26 +02:00
|
|
|
}
|