aboutsummaryrefslogtreecommitdiff
path: root/syncapi/storage/sqlite3
diff options
context:
space:
mode:
authorNeil Alexander <neilalexander@users.noreply.github.com>2020-06-01 17:50:19 +0100
committerGitHub <noreply@github.com>2020-06-01 17:50:19 +0100
commita5d822004dd93d6f6a7ed73371aeb4bfb163b5ba (patch)
tree76b10a79094415e45cfe5f2db9dfdf58cf0d2837 /syncapi/storage/sqlite3
parent1f43c24f8602dfbc95620e9d34fac78a7b449c11 (diff)
Send-to-device support (#1072)
* Groundwork for send-to-device messaging * Update sample config * Add unstable routing for now * Send to device consumer in sync API * Start the send-to-device consumer * fix indentation in dendrite-config.yaml * Create send-to-device database tables, other tweaks * Add some logic for send-to-device messages, add them into sync stream * Handle incoming send-to-device messages, count them with EDU stream pos * Undo changes to test * pq.Array * Fix sync * Logging * Fix a couple of transaction things, fix client API * Add send-to-device test, hopefully fix bugs * Comments * Refactor a bit * Fix schema * Fix queries * Debug logging * Fix storing and retrieving of send-to-device messages * Try to avoid database locks * Update sync position * Use latest sync position * Jiggle about sync a bit * Fix tests * Break out the retrieval from the update/delete behaviour * Comments * nolint on getResponseWithPDUsForCompleteSync * Try to line up sync tokens again * Implement wildcard * Add all send-to-device tests to whitelist, what could possibly go wrong? * Only care about wildcard when targeted locally * Deduplicate transactions * Handle tokens properly, return immediately if waiting send-to-device messages * Fix sync * Update sytest-whitelist * Fix copyright notice (need to do more of this) * Comments, copyrights * Return errors from Do, fix dendritejs * Review comments * Comments * Constructor for TransactionWriter * defletions * Update gomatrixserverlib, sytest-blacklist
Diffstat (limited to 'syncapi/storage/sqlite3')
-rw-r--r--syncapi/storage/sqlite3/send_to_device_table.go172
-rw-r--r--syncapi/storage/sqlite3/syncserver.go6
2 files changed, 178 insertions, 0 deletions
diff --git a/syncapi/storage/sqlite3/send_to_device_table.go b/syncapi/storage/sqlite3/send_to_device_table.go
new file mode 100644
index 00000000..0d03f23e
--- /dev/null
+++ b/syncapi/storage/sqlite3/send_to_device_table.go
@@ -0,0 +1,172 @@
+// 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"
+ "strings"
+
+ "github.com/matrix-org/dendrite/internal"
+ "github.com/matrix-org/dendrite/syncapi/storage/tables"
+ "github.com/matrix-org/dendrite/syncapi/types"
+)
+
+const sendToDeviceSchema = `
+-- Stores send-to-device messages.
+CREATE TABLE IF NOT EXISTS syncapi_send_to_device (
+ -- The ID that uniquely identifies this message.
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ -- The user ID to send the message to.
+ user_id TEXT NOT NULL,
+ -- The device ID to send the message to.
+ device_id TEXT NOT NULL,
+ -- The event content JSON.
+ content TEXT NOT NULL,
+ -- The token that was supplied to the /sync at the time that this
+ -- message was included in a sync response, or NULL if we haven't
+ -- included it in a /sync response yet.
+ sent_by_token TEXT
+);
+`
+
+const insertSendToDeviceMessageSQL = `
+ INSERT INTO syncapi_send_to_device (user_id, device_id, content)
+ VALUES ($1, $2, $3)
+`
+
+const countSendToDeviceMessagesSQL = `
+ SELECT COUNT(*)
+ FROM syncapi_send_to_device
+ WHERE user_id = $1 AND device_id = $2
+`
+
+const selectSendToDeviceMessagesSQL = `
+ SELECT id, user_id, device_id, content, sent_by_token
+ FROM syncapi_send_to_device
+ WHERE user_id = $1 AND device_id = $2
+ ORDER BY id DESC
+`
+
+const updateSentSendToDeviceMessagesSQL = `
+ UPDATE syncapi_send_to_device SET sent_by_token = $1
+ WHERE id IN ($2)
+`
+
+const deleteSendToDeviceMessagesSQL = `
+ DELETE FROM syncapi_send_to_device WHERE id IN ($1)
+`
+
+type sendToDeviceStatements struct {
+ insertSendToDeviceMessageStmt *sql.Stmt
+ selectSendToDeviceMessagesStmt *sql.Stmt
+ countSendToDeviceMessagesStmt *sql.Stmt
+}
+
+func NewSqliteSendToDeviceTable(db *sql.DB) (tables.SendToDevice, error) {
+ s := &sendToDeviceStatements{}
+ _, err := db.Exec(sendToDeviceSchema)
+ if err != nil {
+ return nil, err
+ }
+ if s.countSendToDeviceMessagesStmt, err = db.Prepare(countSendToDeviceMessagesSQL); err != nil {
+ return nil, err
+ }
+ if s.insertSendToDeviceMessageStmt, err = db.Prepare(insertSendToDeviceMessageSQL); err != nil {
+ return nil, err
+ }
+ if s.selectSendToDeviceMessagesStmt, err = db.Prepare(selectSendToDeviceMessagesSQL); err != nil {
+ return nil, err
+ }
+ return s, nil
+}
+
+func (s *sendToDeviceStatements) InsertSendToDeviceMessage(
+ ctx context.Context, txn *sql.Tx, userID, deviceID, content string,
+) (err error) {
+ _, err = internal.TxStmt(txn, s.insertSendToDeviceMessageStmt).ExecContext(ctx, userID, deviceID, content)
+ return
+}
+
+func (s *sendToDeviceStatements) CountSendToDeviceMessages(
+ ctx context.Context, txn *sql.Tx, userID, deviceID string,
+) (count int, err error) {
+ row := internal.TxStmt(txn, s.countSendToDeviceMessagesStmt).QueryRowContext(ctx, userID, deviceID)
+ if err = row.Scan(&count); err != nil {
+ return
+ }
+ return count, nil
+}
+
+func (s *sendToDeviceStatements) SelectSendToDeviceMessages(
+ ctx context.Context, txn *sql.Tx, userID, deviceID string,
+) (events []types.SendToDeviceEvent, err error) {
+ rows, err := internal.TxStmt(txn, s.selectSendToDeviceMessagesStmt).QueryContext(ctx, userID, deviceID)
+ if err != nil {
+ return
+ }
+ defer internal.CloseAndLogIfError(ctx, rows, "SelectSendToDeviceMessages: rows.close() failed")
+
+ for rows.Next() {
+ var id types.SendToDeviceNID
+ var userID, deviceID, content string
+ var sentByToken *string
+ if err = rows.Scan(&id, &userID, &deviceID, &content, &sentByToken); err != nil {
+ return
+ }
+ event := types.SendToDeviceEvent{
+ ID: id,
+ UserID: userID,
+ DeviceID: deviceID,
+ }
+ if err = json.Unmarshal([]byte(content), &event.SendToDeviceEvent); err != nil {
+ return
+ }
+ if sentByToken != nil {
+ if token, err := types.NewStreamTokenFromString(*sentByToken); err == nil {
+ event.SentByToken = &token
+ }
+ }
+ events = append(events, event)
+ }
+
+ return events, rows.Err()
+}
+
+func (s *sendToDeviceStatements) UpdateSentSendToDeviceMessages(
+ ctx context.Context, txn *sql.Tx, token string, nids []types.SendToDeviceNID,
+) (err error) {
+ query := strings.Replace(updateSentSendToDeviceMessagesSQL, "($2)", internal.QueryVariadic(1+len(nids)), 1)
+ params := make([]interface{}, 1+len(nids))
+ params[0] = token
+ for k, v := range nids {
+ params[k+1] = v
+ }
+ _, err = txn.ExecContext(ctx, query, params...)
+ return
+}
+
+func (s *sendToDeviceStatements) DeleteSendToDeviceMessages(
+ ctx context.Context, txn *sql.Tx, nids []types.SendToDeviceNID,
+) (err error) {
+ query := strings.Replace(deleteSendToDeviceMessagesSQL, "($1)", internal.QueryVariadic(len(nids)), 1)
+ params := make([]interface{}, 1+len(nids))
+ for k, v := range nids {
+ params[k] = v
+ }
+ _, err = txn.ExecContext(ctx, query, params...)
+ return
+}
diff --git a/syncapi/storage/sqlite3/syncserver.go b/syncapi/storage/sqlite3/syncserver.go
index 8ab1d404..5ba07617 100644
--- a/syncapi/storage/sqlite3/syncserver.go
+++ b/syncapi/storage/sqlite3/syncserver.go
@@ -95,6 +95,10 @@ func (d *SyncServerDatasource) prepare() (err error) {
if err != nil {
return err
}
+ sendToDevice, err := NewSqliteSendToDeviceTable(d.db)
+ if err != nil {
+ return err
+ }
d.Database = shared.Database{
DB: d.db,
Invites: invites,
@@ -103,6 +107,8 @@ func (d *SyncServerDatasource) prepare() (err error) {
BackwardExtremities: bwExtrem,
CurrentRoomState: roomState,
Topology: topology,
+ SendToDevice: sendToDevice,
+ SendToDeviceWriter: internal.NewTransactionWriter(),
EDUCache: cache.New(),
}
return nil