Merge commit '45c604f56367708d8f1916a5598b1df0973ba33e'

This commit is contained in:
Brian Meek 2022-09-09 17:17:50 -07:00
commit 976e06eb24
59 changed files with 1461 additions and 338 deletions

View file

@ -1,5 +1,26 @@
# Changelog
## Dendrite 0.9.7 (2022-09-09)
### Features
* Initial supporting code to enable full-text search has been merged (although not ready for use yet)
* Newly created rooms now have higher default power levels for enabling encryption, setting server ACLs or sending tombstone events
* Incoming signing key updates over federation are now queued in JetStream for processing, so that they cannot be dropped accidentally
### Fixes
* A race condition between the roomserver output events being generated, forward extremities being updated and room info being updated has been fixed
* Appservices will no longer receive invite events which they are not interested in, which caused heavy load in some cases or excessive request sizes in others
* A bug in state resolution v2 where events could incorrectly be classified as control events has been fixed
* A bug in state resolution v2 where some specific events with unexpected non-empty state keys are dropped has been fixed
* A bug in state resolution v2 when fetching auth events vs partial state has been fixed
* Stale device lists should now be handled correctly for all user IDs, which may help with E2EE reliability
* A number of database writer issues have been fixed in the user API and sync API, which should help to reduce `database is locked` errors with SQLite databases
* Database migrations should now be detected more reliably to prevent unexpected errors at startup
* A number of minor database transaction issues have been fixed, particularly for assigning NIDs in the roomserver, cleaning up device keys and cleaning up notifications
* The database query for finding shared users in the sync API has been optimised, using significantly less CPU time as a result
## Dendrite 0.9.6 (2022-09-01)
### Features

View file

@ -76,9 +76,14 @@ func main() {
panic(err)
}
var eventNIDs []types.EventNID
eventNIDMap := map[types.EventNID]struct{}{}
for _, entry := range append(removed, added...) {
eventNIDs = append(eventNIDs, entry.EventNID)
eventNIDMap[entry.EventNID] = struct{}{}
}
eventNIDs := make([]types.EventNID, 0, len(eventNIDMap))
for eventNID := range eventNIDMap {
eventNIDs = append(eventNIDs, eventNID)
}
var eventEntries []types.Event
@ -129,12 +134,17 @@ func main() {
stateEntries = append(stateEntries, entries...)
}
var eventNIDs []types.EventNID
eventNIDMap := map[types.EventNID]struct{}{}
for _, entry := range stateEntries {
eventNIDs = append(eventNIDs, entry.EventNID)
eventNIDMap[entry.EventNID] = struct{}{}
}
fmt.Println("Fetching", len(eventNIDs), "state events")
eventNIDs := make([]types.EventNID, 0, len(eventNIDMap))
for eventNID := range eventNIDMap {
eventNIDs = append(eventNIDs, eventNID)
}
fmt.Println("Fetching", len(eventNIDMap), "state events")
eventEntries, err := roomserverDB.Events(ctx, eventNIDs)
if err != nil {
panic(err)

View file

@ -278,6 +278,10 @@ sync_api:
# address of the client. This is likely required if Dendrite is running behind
# a reverse proxy server.
# real_ip_header: X-Real-IP
fulltext:
enabled: false
index_path: "./fulltextindex"
language: "en" # more possible languages can be found at https://github.com/blevesearch/bleve/tree/master/analysis/lang
# Configuration for the User API.
user_api:

View file

@ -329,6 +329,10 @@ sync_api:
max_open_conns: 10
max_idle_conns: 2
conn_max_lifetime: -1
fulltext:
enabled: false
index_path: "./fulltextindex"
language: "en" # more possible languages can be found at https://github.com/blevesearch/bleve/tree/master/analysis/lang
# This option controls which HTTP header to inspect to find the real remote IP
# address of the client. This is likely required if Dendrite is running behind

View file

@ -138,6 +138,19 @@ room_server:
conn_max_lifetime: -1
```
## Fulltext search
Dendrite supports experimental fulltext indexing using [Bleve](https://github.com/blevesearch/bleve), it is configured in the `sync_api` section as follows. Depending on the language most likely to be used on the server, it might make sense to change the `language` used when indexing, to ensure the returned results match the expections. A full list of possible languages can be found [here](https://github.com/blevesearch/bleve/tree/master/analysis/lang).
```yaml
sync_api:
# ...
fulltext:
enabled: false
index_path: "./fulltextindex"
language: "en"
```
## Other sections
There are other options which may be useful so review them all. In particular, if you are

View file

@ -5,9 +5,10 @@ import (
"fmt"
"time"
"github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/gomatrix"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/federationapi/types"
)
// FederationInternalAPI is used to query information from the federation sender.
@ -108,6 +109,7 @@ type FederationClientError struct {
Err string
RetryAfter time.Duration
Blacklisted bool
Code int // HTTP Status code from the remote server
}
func (e FederationClientError) Error() string {

View file

@ -18,6 +18,8 @@ import (
"time"
"github.com/gorilla/mux"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/federationapi/api"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/federationapi/consumers"
@ -33,10 +35,10 @@ import (
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/jetstream"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/sirupsen/logrus"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/federationapi/routing"
"github.com/matrix-org/gomatrixserverlib"
)
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
@ -66,6 +68,7 @@ func AddPublicRoutes(
TopicTypingEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent),
TopicPresenceEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent),
TopicDeviceListUpdate: cfg.Matrix.JetStream.Prefixed(jetstream.InputDeviceListUpdate),
TopicSigningKeyUpdate: cfg.Matrix.JetStream.Prefixed(jetstream.InputSigningKeyUpdate),
ServerName: cfg.Matrix.ServerName,
UserAPI: userAPI,
}

View file

@ -4,12 +4,15 @@ import (
"context"
"encoding/json"
"net/http"
"net/url"
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/gomatrix"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/httputil"
)
// AddRoutes adds the FederationInternalAPI handlers to the http.ServeMux.
@ -229,9 +232,21 @@ func federationClientError(err error) error {
return &ferr
case *api.FederationClientError:
return ferr
default:
case gomatrix.HTTPError:
return &api.FederationClientError{
Err: err.Error(),
Code: ferr.Code,
}
case *url.Error: // e.g. certificate error, unable to connect
return &api.FederationClientError{
Err: ferr.Error(),
Code: 400,
}
default:
// We don't know what exactly failed, but we probably don't
// want to retry the request immediately in the device list updater
return &api.FederationClientError{
Err: err.Error(),
Code: 400,
}
}
}

View file

@ -21,12 +21,13 @@ import (
"strconv"
"time"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/syncapi/types"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/nats-io/nats.go"
log "github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/syncapi/types"
userapi "github.com/matrix-org/dendrite/userapi/api"
)
// SyncAPIProducer produces events for the sync API server to consume
@ -36,6 +37,7 @@ type SyncAPIProducer struct {
TopicTypingEvent string
TopicPresenceEvent string
TopicDeviceListUpdate string
TopicSigningKeyUpdate string
JetStream nats.JetStreamContext
ServerName gomatrixserverlib.ServerName
UserAPI userapi.UserInternalAPI
@ -165,16 +167,24 @@ func (p *SyncAPIProducer) SendPresence(
}
func (p *SyncAPIProducer) SendDeviceListUpdate(
ctx context.Context, deviceListUpdate *gomatrixserverlib.DeviceListUpdateEvent,
ctx context.Context, deviceListUpdate gomatrixserverlib.RawJSON, origin string,
) (err error) {
m := nats.NewMsg(p.TopicDeviceListUpdate)
m.Header.Set(jetstream.UserID, deviceListUpdate.UserID)
m.Data, err = json.Marshal(deviceListUpdate)
if err != nil {
return fmt.Errorf("json.Marshal: %w", err)
}
m.Header.Set("origin", origin)
m.Data = deviceListUpdate
log.Debugf("Sending device list update: %+v", m.Header)
_, err = p.JetStream.PublishMsg(m, nats.Context(ctx))
return err
}
func (p *SyncAPIProducer) SendSigningKeyUpdate(
ctx context.Context, data gomatrixserverlib.RawJSON, origin gomatrixserverlib.ServerName,
) (err error) {
m := nats.NewMsg(p.TopicSigningKeyUpdate)
m.Header.Set("origin", string(origin))
m.Data = data
log.Debugf("Sending signing key update")
_, err = p.JetStream.PublishMsg(m, nats.Context(ctx))
return err
}

View file

@ -22,6 +22,11 @@ import (
"sync"
"time"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"github.com/prometheus/client_golang/prometheus"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/federationapi/producers"
@ -31,10 +36,6 @@ import (
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/config"
syncTypes "github.com/matrix-org/dendrite/syncapi/types"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"github.com/prometheus/client_golang/prometheus"
"github.com/sirupsen/logrus"
)
const (
@ -358,7 +359,9 @@ func (t *txnReq) processEDUs(ctx context.Context) {
}
}
case gomatrixserverlib.MDeviceListUpdate:
t.processDeviceListUpdate(ctx, e)
if err := t.producer.SendDeviceListUpdate(ctx, e.Content, e.Origin); err != nil {
util.GetLogger(ctx).WithError(err).Error("failed to InputDeviceListUpdate")
}
case gomatrixserverlib.MReceipt:
// https://matrix.org/docs/spec/server_server/r0.1.4#receipts
payload := map[string]types.FederationReceiptMRead{}
@ -391,7 +394,7 @@ func (t *txnReq) processEDUs(ctx context.Context) {
}
}
case types.MSigningKeyUpdate:
if err := t.processSigningKeyUpdate(ctx, e); err != nil {
if err := t.producer.SendSigningKeyUpdate(ctx, e.Content, t.Origin); err != nil {
logrus.WithError(err).Errorf("Failed to process signing key update")
}
case gomatrixserverlib.MPresence:
@ -431,42 +434,6 @@ func (t *txnReq) processPresence(ctx context.Context, e gomatrixserverlib.EDU) e
return nil
}
func (t *txnReq) processSigningKeyUpdate(ctx context.Context, e gomatrixserverlib.EDU) error {
var updatePayload keyapi.CrossSigningKeyUpdate
if err := json.Unmarshal(e.Content, &updatePayload); err != nil {
util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{
"user_id": updatePayload.UserID,
}).Debug("Failed to unmarshal signing key update")
return err
}
if _, serverName, err := gomatrixserverlib.SplitID('@', updatePayload.UserID); err != nil {
return nil
} else if serverName == t.ourServerName {
return nil
} else if serverName != t.Origin {
return nil
}
keys := gomatrixserverlib.CrossSigningKeys{}
if updatePayload.MasterKey != nil {
keys.MasterKey = *updatePayload.MasterKey
}
if updatePayload.SelfSigningKey != nil {
keys.SelfSigningKey = *updatePayload.SelfSigningKey
}
uploadReq := &keyapi.PerformUploadDeviceKeysRequest{
CrossSigningKeys: keys,
UserID: updatePayload.UserID,
}
uploadRes := &keyapi.PerformUploadDeviceKeysResponse{}
if err := t.keyAPI.PerformUploadDeviceKeys(ctx, uploadReq, uploadRes); err != nil {
return err
}
if uploadRes.Error != nil {
return uploadRes.Error
}
return nil
}
// processReceiptEvent sends receipt events to JetStream
func (t *txnReq) processReceiptEvent(ctx context.Context,
userID, roomID, receiptType string,
@ -489,21 +456,3 @@ func (t *txnReq) processReceiptEvent(ctx context.Context,
return nil
}
func (t *txnReq) processDeviceListUpdate(ctx context.Context, e gomatrixserverlib.EDU) {
var payload gomatrixserverlib.DeviceListUpdateEvent
if err := json.Unmarshal(e.Content, &payload); err != nil {
util.GetLogger(ctx).WithError(err).Error("Failed to unmarshal device list update event")
return
}
if _, serverName, err := gomatrixserverlib.SplitID('@', payload.UserID); err != nil {
return
} else if serverName == t.ourServerName {
return
} else if serverName != t.Origin {
return
}
if err := t.producer.SendDeviceListUpdate(ctx, &payload); err != nil {
util.GetLogger(ctx).WithError(err).WithField("user_id", payload.UserID).Error("failed to InputDeviceListUpdate")
}
}

View file

@ -53,9 +53,9 @@ func (d *Database) AssociateEDUWithDestination(
// Keep EDUs for at least x minutes before deleting them
expiresAt = gomatrixserverlib.AsTimestamp(time.Now().Add(duration))
}
// We forcibly set m.direct_to_device events to 0, as we always want them
// to be delivered. (required for E2EE)
if eduType == gomatrixserverlib.MDirectToDevice {
// We forcibly set m.direct_to_device and m.device_list_update events
// to 0, as we always want them to be delivered. (required for E2EE)
if eduType == gomatrixserverlib.MDirectToDevice || eduType == gomatrixserverlib.MDeviceListUpdate {
expiresAt = 0
}
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {

33
go.mod
View file

@ -6,6 +6,7 @@ require (
github.com/DATA-DOG/go-sqlmock v1.5.0
github.com/MFAshby/stdemuxerhook v1.0.0
github.com/Masterminds/semver/v3 v3.1.1
github.com/blevesearch/bleve/v2 v2.3.4
github.com/codeclysm/extract v2.2.0+incompatible
github.com/dgraph-io/ristretto v0.1.1-0.20220403145359-8e850b710d6d
github.com/docker/docker v20.10.16+incompatible
@ -16,12 +17,11 @@ require (
github.com/google/uuid v1.3.0
github.com/gorilla/mux v1.8.0
github.com/gorilla/websocket v1.5.0
github.com/kardianos/minwinsvc v1.0.0
github.com/lib/pq v1.10.5
github.com/matrix-org/dugong v0.0.0-20210921133753-66e6b1c67e2e
github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91
github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16
github.com/matrix-org/gomatrixserverlib v0.0.0-20220830164018-c71e518537a2
github.com/matrix-org/gomatrixserverlib v0.0.0-20220907081047-637a173a3661
github.com/matrix-org/pinecone v0.0.0-20220901133433-565beccfebed
github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4
github.com/mattn/go-sqlite3 v1.14.13
@ -54,13 +54,33 @@ require (
nhooyr.io/websocket v1.8.7
)
require github.com/ethereum/go-ethereum v1.10.15
require (
github.com/ethereum/go-ethereum v1.10.15
github.com/kardianos/minwinsvc v1.0.0
)
require (
github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
github.com/HdrHistogram/hdrhistogram-go v1.1.2 // indirect
github.com/Microsoft/go-winio v0.5.1 // indirect
github.com/RoaringBitmap/roaring v0.9.4 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/bits-and-blooms/bitset v1.2.0 // indirect
github.com/blevesearch/bleve_index_api v1.0.3 // indirect
github.com/blevesearch/geo v0.1.13 // indirect
github.com/blevesearch/go-porterstemmer v1.0.3 // indirect
github.com/blevesearch/gtreap v0.1.1 // indirect
github.com/blevesearch/mmap-go v1.0.4 // indirect
github.com/blevesearch/scorch_segment_api/v2 v2.1.2 // indirect
github.com/blevesearch/segment v0.9.0 // indirect
github.com/blevesearch/snowballstem v0.9.0 // indirect
github.com/blevesearch/upsidedown_store_api v1.0.1 // indirect
github.com/blevesearch/vellum v1.0.8 // indirect
github.com/blevesearch/zapx/v11 v11.3.5 // indirect
github.com/blevesearch/zapx/v12 v12.3.5 // indirect
github.com/blevesearch/zapx/v13 v13.3.5 // indirect
github.com/blevesearch/zapx/v14 v14.3.5 // indirect
github.com/blevesearch/zapx/v15 v15.3.5 // indirect
github.com/btcsuite/btcd v0.20.1-beta // indirect
github.com/cespare/xxhash/v2 v2.1.2 // indirect
github.com/cheekybits/genny v1.0.0 // indirect
@ -73,9 +93,12 @@ require (
github.com/fsnotify/fsnotify v1.5.4 // indirect
github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang/geo v0.0.0-20210211234256-740aa86cb551 // indirect
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b // indirect
github.com/golang/protobuf v1.5.2 // indirect
github.com/golang/snappy v0.0.4 // indirect
github.com/h2non/filetype v1.1.3 // indirect
github.com/json-iterator/go v1.1.12 // indirect
github.com/juju/errors v0.0.0-20220203013757-bd733f3c86b9 // indirect
github.com/juju/testing v0.0.0-20220203020004-a0ff61f03494 // indirect
github.com/klauspost/compress v1.15.9 // indirect
@ -88,7 +111,10 @@ require (
github.com/miekg/dns v1.1.49 // indirect
github.com/minio/highwayhash v1.0.2 // indirect
github.com/moby/term v0.0.0-20210610120745-9d4ed1856297 // indirect
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
github.com/modern-go/reflect2 v1.0.2 // indirect
github.com/morikuni/aec v1.0.0 // indirect
github.com/mschoch/smat v0.2.0 // indirect
github.com/nats-io/jwt/v2 v2.3.0 // indirect
github.com/nats-io/nkeys v0.3.0 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
@ -104,6 +130,7 @@ require (
github.com/relvacode/iso8601 v1.1.0 // indirect
github.com/tidwall/match v1.1.1 // indirect
github.com/tidwall/pretty v1.2.0 // indirect
go.etcd.io/bbolt v1.3.5 // indirect
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
golang.org/x/sys v0.0.0-20220731174439-a90be440212d // indirect
golang.org/x/text v0.3.8-0.20211004125949-5bd84dd9b33b // indirect

75
go.sum
View file

@ -77,6 +77,8 @@ github.com/Microsoft/go-winio v0.5.1 h1:aPJp2QD7OOrhO5tQXqQoGSJc+DjDtWTGLOmNyAm6
github.com/Microsoft/go-winio v0.5.1/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84=
github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
github.com/RoaringBitmap/roaring v0.4.7/go.mod h1:8khRDP4HmeXns4xIj9oGrKSz7XTQiJx2zgh7AcNke4w=
github.com/RoaringBitmap/roaring v0.9.4 h1:ckvZSX5gwCRaJYBNe7syNawCU5oruY9gQmjXlp4riwo=
github.com/RoaringBitmap/roaring v0.9.4/go.mod h1:icnadbWcNyfEHlYdr+tDlOTih1Bf/h+rzPpv4sbomAA=
github.com/RyanCarrier/dijkstra v1.1.0/go.mod h1:5agGUBNEtUAGIANmbw09fuO3a2htPEkc1jNH01qxCWA=
github.com/RyanCarrier/dijkstra-1 v0.0.0-20170512020943-0e5801a26345/go.mod h1:OK4EvWJ441LQqGzed5NGB6vKBAE34n3z7iayPcEwr30=
github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
@ -104,6 +106,7 @@ github.com/anacrolix/tagflag v0.0.0-20180109131632-2146c8d41bf0/go.mod h1:1m2U/K
github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883/go.mod h1:rCTlJbsFo29Kk6CurOXKm700vrz8f0KW0JNfpkRJY/8=
github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
github.com/apache/arrow/go/arrow v0.0.0-20191024131854-af6fa24be0db/go.mod h1:VTxUBvSJ3s3eHAg65PNgrsn5BtqCRPdmyXh6rAfdxN0=
github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
github.com/aws/aws-sdk-go-v2 v1.2.0/go.mod h1:zEQs02YRBw1DjK0PoJv3ygDYOFTre1ejlJWl8FwAuQo=
github.com/aws/aws-sdk-go-v2/config v1.1.1/go.mod h1:0XsVy9lBI/BCXm+2Tuvt39YmdHwS5unDQmxZOYe8F5Y=
github.com/aws/aws-sdk-go-v2/credentials v1.1.1/go.mod h1:mM2iIjwl7LULWtS6JCACyInboHirisUUdkBPoTHMOUo=
@ -117,6 +120,44 @@ github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24
github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/bits-and-blooms/bitset v1.2.0 h1:Kn4yilvwNtMACtf1eYDlG8H77R07mZSPbMjLyS07ChA=
github.com/bits-and-blooms/bitset v1.2.0/go.mod h1:gIdJ4wp64HaoK2YrL1Q5/N7Y16edYb8uY+O0FJTyyDA=
github.com/blevesearch/bleve/v2 v2.3.4 h1:SSb7/cwGzo85LWX1jchIsXM8ZiNNMX3shT5lROM63ew=
github.com/blevesearch/bleve/v2 v2.3.4/go.mod h1:Ot0zYum8XQRfPcwhae8bZmNyYubynsoMjVvl1jPqL30=
github.com/blevesearch/bleve_index_api v1.0.3 h1:DDSWaPXOZZJ2BB73ZTWjKxydAugjwywcqU+91AAqcAg=
github.com/blevesearch/bleve_index_api v1.0.3/go.mod h1:fiwKS0xLEm+gBRgv5mumf0dhgFr2mDgZah1pqv1c1M4=
github.com/blevesearch/geo v0.1.13 h1:RsY1vfFm81iv1g+uoCQtsOFvKAhZnpOdTOK8JRA6pqw=
github.com/blevesearch/geo v0.1.13/go.mod h1:cRIvqCdk3cgMhGeHNNe6yPzb+w56otxbfo1FBJfR2Pc=
github.com/blevesearch/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:9eJDeqxJ3E7WnLebQUlPD7ZjSce7AnDb9vjGmMCbD0A=
github.com/blevesearch/go-porterstemmer v1.0.3 h1:GtmsqID0aZdCSNiY8SkuPJ12pD4jI+DdXTAn4YRcHCo=
github.com/blevesearch/go-porterstemmer v1.0.3/go.mod h1:angGc5Ht+k2xhJdZi511LtmxuEf0OVpvUUNrwmM1P7M=
github.com/blevesearch/goleveldb v1.0.1/go.mod h1:WrU8ltZbIp0wAoig/MHbrPCXSOLpe79nz5lv5nqfYrQ=
github.com/blevesearch/gtreap v0.1.1 h1:2JWigFrzDMR+42WGIN/V2p0cUvn4UP3C4Q5nmaZGW8Y=
github.com/blevesearch/gtreap v0.1.1/go.mod h1:QaQyDRAT51sotthUWAH4Sj08awFSSWzgYICSZ3w0tYk=
github.com/blevesearch/mmap-go v1.0.2/go.mod h1:ol2qBqYaOUsGdm7aRMRrYGgPvnwLe6Y+7LMvAB5IbSA=
github.com/blevesearch/mmap-go v1.0.4 h1:OVhDhT5B/M1HNPpYPBKIEJaD0F3Si+CrEKULGCDPWmc=
github.com/blevesearch/mmap-go v1.0.4/go.mod h1:EWmEAOmdAS9z/pi/+Toxu99DnsbhG1TIxUoRmJw/pSs=
github.com/blevesearch/scorch_segment_api/v2 v2.1.2 h1:TAte9VZLWda5WAVlZTTZ+GCzEHqGJb4iB2aiZSA6Iv8=
github.com/blevesearch/scorch_segment_api/v2 v2.1.2/go.mod h1:rvoQXZGq8drq7vXbNeyiRzdEOwZkjkiYGf1822i6CRA=
github.com/blevesearch/segment v0.9.0 h1:5lG7yBCx98or7gK2cHMKPukPZ/31Kag7nONpoBt22Ac=
github.com/blevesearch/segment v0.9.0/go.mod h1:9PfHYUdQCgHktBgvtUOF4x+pc4/l8rdH0u5spnW85UQ=
github.com/blevesearch/snowball v0.6.1/go.mod h1:ZF0IBg5vgpeoUhnMza2v0A/z8m1cWPlwhke08LpNusg=
github.com/blevesearch/snowballstem v0.9.0 h1:lMQ189YspGP6sXvZQ4WZ+MLawfV8wOmPoD/iWeNXm8s=
github.com/blevesearch/snowballstem v0.9.0/go.mod h1:PivSj3JMc8WuaFkTSRDW2SlrulNWPl4ABg1tC/hlgLs=
github.com/blevesearch/upsidedown_store_api v1.0.1 h1:1SYRwyoFLwG3sj0ed89RLtM15amfX2pXlYbFOnF8zNU=
github.com/blevesearch/upsidedown_store_api v1.0.1/go.mod h1:MQDVGpHZrpe3Uy26zJBf/a8h0FZY6xJbthIMm8myH2Q=
github.com/blevesearch/vellum v1.0.8 h1:iMGh4lfxza4BnWO/UJTMPlI3HsK9YawjPv+TteVa9ck=
github.com/blevesearch/vellum v1.0.8/go.mod h1:+cpRi/tqq49xUYSQN2P7A5zNSNrS+MscLeeaZ3J46UA=
github.com/blevesearch/zapx/v11 v11.3.5 h1:eBQWQ7huA+mzm0sAGnZDwgGGli7S45EO+N+ObFWssbI=
github.com/blevesearch/zapx/v11 v11.3.5/go.mod h1:5UdIa/HRMdeRCiLQOyFESsnqBGiip7vQmYReA9toevU=
github.com/blevesearch/zapx/v12 v12.3.5 h1:5pX2hU+R1aZihT7ac1dNWh1n4wqkIM9pZzWp0ANED9s=
github.com/blevesearch/zapx/v12 v12.3.5/go.mod h1:ANcthYRZQycpbRut/6ArF5gP5HxQyJqiFcuJCBju/ss=
github.com/blevesearch/zapx/v13 v13.3.5 h1:eJ3gbD+Nu8p36/O6lhfdvWQ4pxsGYSuTOBrLLPVWJ74=
github.com/blevesearch/zapx/v13 v13.3.5/go.mod h1:FV+dRnScFgKnRDIp08RQL4JhVXt1x2HE3AOzqYa6fjo=
github.com/blevesearch/zapx/v14 v14.3.5 h1:hEvVjZaagFCvOUJrlFQ6/Z6Jjy0opM3g7TMEo58TwP4=
github.com/blevesearch/zapx/v14 v14.3.5/go.mod h1:954A/eKFb+pg/ncIYWLWCKY+mIjReM9FGTGIO2Wu1cU=
github.com/blevesearch/zapx/v15 v15.3.5 h1:NVD0qq8vRk66ImJn1KloXT5ckqPDUZT7VbVJs9jKlac=
github.com/blevesearch/zapx/v15 v15.3.5/go.mod h1:QMUh2hXCaYIWFKPYGavq/Iga2zbHWZ9DZAa9uFbWyvg=
github.com/bmizerany/pat v0.0.0-20170815010413-6226ea591a40/go.mod h1:8rLXio+WjiTceGBHIoTvn60HIbs7Hm7bcHjyrSqYB9c=
github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
github.com/bradfitz/go-smtpd v0.0.0-20170404230938-deb6d6237625/go.mod h1:HYsPBTaaSFSlLx/70C2HPIMNZpVV8+vt/A+FMnYP11g=
@ -153,7 +194,13 @@ github.com/codeclysm/extract v2.2.0+incompatible h1:q3wyckoA30bhUSiwdQezMqVhwd8+
github.com/codeclysm/extract v2.2.0+incompatible/go.mod h1:2nhFMPHiU9At61hz+12bfrlpXSUrOnK+wR+KlGO4Uks=
github.com/consensys/bavard v0.1.8-0.20210406032232-f3452dc9b572/go.mod h1:Bpd0/3mZuaj6Sj+PqrmIquiOKy397AKGThQPaGzNXAQ=
github.com/consensys/gnark-crypto v0.4.1-0.20210426202927-39ac3d4b3f1f/go.mod h1:815PAHg3wvysy0SyIqanF8gZ0Y1wjk/hrDHD/iT88+Q=
github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
github.com/coreos/go-systemd v0.0.0-20181012123002-c6f51f82210d/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
github.com/couchbase/ghistogram v0.1.0/go.mod h1:s1Jhy76zqfEecpNWJfWUiKZookAFaiGOEoyzgHt9i7k=
github.com/couchbase/moss v0.2.0/go.mod h1:9MaHIaRuy9pvLPUJxB8sh8OrLfyDczECVL37grCIubs=
github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwcJI5acqYI6dE=
github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
@ -266,6 +313,8 @@ github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0/go.mod h1:E/TSTwGwJL78qG/PmXZO1EjYhfJinVAhrmmHX6Z8B9k=
github.com/golang/geo v0.0.0-20190916061304-5b978397cfec/go.mod h1:QZ0nwyI2jOfgRAoBvP+ab5aRr7c9x7lhGEJrKvBwjWI=
github.com/golang/geo v0.0.0-20210211234256-740aa86cb551 h1:gtexQ/VGyN+VVFRXSFiguSNcXmS6rkKT+X7FdIrTtfo=
github.com/golang/geo v0.0.0-20210211234256-740aa86cb551/go.mod h1:QZ0nwyI2jOfgRAoBvP+ab5aRr7c9x7lhGEJrKvBwjWI=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
@ -299,7 +348,9 @@ github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaS
github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw=
github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/golangci/lint-1 v0.0.0-20181222135242-d2cdd8c08219/go.mod h1:/X8TswGSh1pIozq4ZwCfxS0WA5JGXguxk94ar/4c87Y=
github.com/gologme/log v1.3.0 h1:l781G4dE+pbigClDSDzSaaYKtiueHCILUa/qSDsmHAo=
@ -324,6 +375,7 @@ github.com/google/go-github v17.0.0+incompatible/go.mod h1:zLgOLi98H3fifZn+44m+u
github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
github.com/google/gofuzz v1.1.1-0.20200604201612-c04b05f3adfa/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
@ -360,6 +412,7 @@ github.com/hashicorp/go-bexpr v0.1.10/go.mod h1:oxlubA2vC/gFVfX1A6JGp7ls7uCDlfJn
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
github.com/holiman/bloomfilter/v2 v2.0.3/go.mod h1:zpoh+gs7qcpqrHr3dB55AMiJwo0iURXE7ZOP9L9hSkA=
github.com/holiman/uint256 v1.2.0/go.mod h1:y4ga/t+u+Xwd7CpDgZESaRcWy0I7XMlTMA25ApIH5Jw=
github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
@ -388,6 +441,7 @@ github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHW
github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlTRt3OuAQ=
github.com/json-iterator/go v0.0.0-20171115153421-f7279a603ede/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
@ -449,6 +503,7 @@ github.com/lib/pq v1.10.5/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
github.com/lucas-clemente/quic-go v0.28.1 h1:Uo0lvVxWg5la9gflIF9lwa39ONq85Xq2D91YNEIslzU=
github.com/lucas-clemente/quic-go v0.28.1/go.mod h1:oGz5DKK41cJt5+773+BSO9BXDsREY4HLf7+0odGAPO0=
github.com/lunixbochs/vtclean v1.0.0/go.mod h1:pHhQNgMf3btfWnGBVipUOjRYhoOsdGqdm/+2c2E2WMI=
github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
github.com/mailru/easyjson v0.0.0-20190312143242-1de009706dbe/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
@ -468,8 +523,8 @@ github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91/go.mod h1
github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0=
github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16 h1:ZtO5uywdd5dLDCud4r0r55eP4j9FuUNpl60Gmntcop4=
github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16/go.mod h1:/gBX06Kw0exX1HrwmoBibFA98yBk/jxKpGVeyQbff+s=
github.com/matrix-org/gomatrixserverlib v0.0.0-20220830164018-c71e518537a2 h1:esbNn9hg//tAStA6TogatAJAursw23A+yfVRQsdiv70=
github.com/matrix-org/gomatrixserverlib v0.0.0-20220830164018-c71e518537a2/go.mod h1:jX38yp3SSLJNftBg3PXU1ayd0PCLIiDHQ4xAc9DIixk=
github.com/matrix-org/gomatrixserverlib v0.0.0-20220907081047-637a173a3661 h1:dww9rH0HVfAO9JOBD1nxq26GHKbEw07thAJTu1DrAQs=
github.com/matrix-org/gomatrixserverlib v0.0.0-20220907081047-637a173a3661/go.mod h1:jX38yp3SSLJNftBg3PXU1ayd0PCLIiDHQ4xAc9DIixk=
github.com/matrix-org/pinecone v0.0.0-20220901133433-565beccfebed h1:YMcCnrmTbT5M1LtTiagiFFaj9vEgvC6iVEzWsIb0tQQ=
github.com/matrix-org/pinecone v0.0.0-20220901133433-565beccfebed/go.mod h1:K0N1ixHQxXoCyqolDqVxPM3ArrDtcMs8yegOx2Lfv9k=
github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7/go.mod h1:vVQlW/emklohkZnOPwD3LrZUBqdfsbiyO3p1lNV8F6U=
@ -503,6 +558,8 @@ github.com/miekg/dns v1.1.49 h1:qe0mQU3Z/XpFeE+AEBo2rqaS1IPBJ3anmqZ4XiZJVG8=
github.com/miekg/dns v1.1.49/go.mod h1:e3IlAVfNqAllflbibAZEWOXOQ+Ynzk/dDozDxY7XnME=
github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
github.com/mitchellh/mapstructure v1.4.1/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
github.com/mitchellh/pointerstructure v1.2.0/go.mod h1:BRAsLI5zgXmw97Lf6s25bs8ohIXc3tViBH44KcwB2g4=
github.com/moby/term v0.0.0-20210610120745-9d4ed1856297 h1:yH0SvLzcbZxcJXho2yh7CqdENGMQe73Cw3woZBpPli0=
@ -517,6 +574,8 @@ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjY
github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A=
github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc=
github.com/mschoch/smat v0.0.0-20160514031455-90eadee771ae/go.mod h1:qAyveg+e4CE+eKJXWVjKXM4ck2QobLqTDytGJbLLhJg=
github.com/mschoch/smat v0.2.0 h1:8imxQsjDm8yFEAVBe7azKmKSgzSkZXDuKkSq9374khM=
github.com/mschoch/smat v0.2.0/go.mod h1:kc9mz7DoBKqDyiRL7VZN8KvXQMWeTaVnttLRXOlotKw=
github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
github.com/naoina/go-stringutil v0.1.0/go.mod h1:XJ2SJL9jCtBh+P9q5btrd/Ylo8XwT/h1USek5+NqSA0=
@ -573,6 +632,7 @@ github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTm
github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
github.com/paulbellamy/ratecounter v0.2.0/go.mod h1:Hfx1hDpSGoqxkVVpBi/IlYD7kChlfo5C6hzIHwPqfFE=
github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
github.com/peterh/liner v1.0.1-0.20180619022028-8c1271fcf47f/go.mod h1:xIteQHvHuaLYG9IFj6mSxM0fCKrs34IrEQUhOYuGPHc=
github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0=
github.com/philhofer/fwd v1.0.0/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU=
@ -663,9 +723,13 @@ github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9
github.com/sourcegraph/annotate v0.0.0-20160123013949-f4cad6c6324d/go.mod h1:UdhH50NIW0fCiwBSr0co2m7BnFLdv4fQTgdqdJTHFeE=
github.com/sourcegraph/syntaxhighlight v0.0.0-20170531221838-bd320f5d308e/go.mod h1:HuIsMU8RRBOtsCgI77wP899iHVBQpCmg4ErYMZB+2IA=
github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
github.com/spf13/cobra v0.0.5/go.mod h1:3K3wKZymM7VvHMDS9+Akkh4K60UwM26emMESw8tLCHU=
github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=
github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
github.com/spruceid/siwe-go v0.2.0 h1:MkBZ/TpPlh1mBhul3h/XLSNZJAbbaHF587Q/VQbhPI0=
github.com/spruceid/siwe-go v0.2.0/go.mod h1:rvV+8/z/ryBKqdw9RcexFgtcsrDlESOGR38sPdVWbSI=
github.com/status-im/keycard-go v0.0.0-20190316090335-8537d3370df4/go.mod h1:RZLeN1LMWmRsyYjvAu+I6Dm9QmlDaIIt+Y+4Kd7Tp+Q=
@ -703,6 +767,7 @@ github.com/uber/jaeger-lib v2.4.1+incompatible h1:td4jdvLcExb4cBISKIpHuGoVXh+dVK
github.com/uber/jaeger-lib v2.4.1+incompatible/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
github.com/ugorji/go v1.1.7 h1:/68gy2h+1mWMrwZFeD1kQialdSzAb432dtpeJ42ovdo=
github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw=
github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs=
github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI=
@ -717,6 +782,7 @@ github.com/vishvananda/netns v0.0.0-20211101163701-50045581ed74/go.mod h1:DD4vA1
github.com/willf/bitset v1.1.3/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4=
github.com/willf/bitset v1.1.9/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4=
github.com/xlab/treeprint v0.0.0-20180616005107-d6fb6747feb6/go.mod h1:ce1O1j6UtZfjr22oyGxGLbauSBp2YVXpARAosm7dHBg=
github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
github.com/yggdrasil-network/yggdrasil-go v0.4.5-0.20220901155642-4f2abece817c h1:/cTmA6pV2Z20BT/FGSmnb5BmJ8eRbDP0HbCB5IO1aKw=
github.com/yggdrasil-network/yggdrasil-go v0.4.5-0.20220901155642-4f2abece817c/go.mod h1:cIwhYwX9yT9Bcei59O0oOBSaj+kQP+9aVQUMWHh5R00=
github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
@ -726,6 +792,8 @@ github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9dec
github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
github.com/yuin/goldmark v1.4.0/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
go.etcd.io/bbolt v1.3.5 h1:XAzx9gjCb0Rxj7EoqcClPD1d5ZBxZJk0jbuoPHenBt0=
go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
go.opencensus.io v0.18.0/go.mod h1:vKdFvxhtzZ9onBp9VKHK8z/sRpBMnKAsufL7wlDrCOA=
go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
@ -743,6 +811,7 @@ golang.org/x/crypto v0.0.0-20170930174604-9419663f5a44/go.mod h1:6SG95UA2DQfeDnf
golang.org/x/crypto v0.0.0-20180723164146-c126467f60eb/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
golang.org/x/crypto v0.0.0-20181030102418-4d3f4d9ffa16/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20190313024323-a1f597ede03a/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
@ -881,6 +950,8 @@ golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5h
golang.org/x/sys v0.0.0-20181029174526-d69651ed3497/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20181221143128-b4a75ba826a6/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

View file

@ -53,6 +53,9 @@ func InitialPowerLevelsContent(roomCreator string) (c gomatrixserverlib.PowerLev
"m.room.history_visibility": 100,
"m.room.canonical_alias": 50,
"m.room.avatar": 50,
"m.room.tombstone": 100,
"m.room.encryption": 100,
"m.room.server_acl": 100,
}
c.Users = map[string]int64{roomCreator: 100}
return c

164
internal/fulltext/bleve.go Normal file
View file

@ -0,0 +1,164 @@
// Copyright 2022 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.
//go:build !wasm
// +build !wasm
package fulltext
import (
"strings"
"github.com/blevesearch/bleve/v2"
"github.com/blevesearch/bleve/v2/mapping"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrixserverlib"
)
// Search contains all existing bleve.Index
type Search struct {
FulltextIndex bleve.Index
}
// IndexElement describes the layout of an element to index
type IndexElement struct {
EventID string
RoomID string
Content string
ContentType string
StreamPosition int64
}
// SetContentType sets i.ContentType given an identifier
func (i *IndexElement) SetContentType(v string) {
switch v {
case "m.room.message":
i.ContentType = "content.body"
case gomatrixserverlib.MRoomName:
i.ContentType = "content.name"
case gomatrixserverlib.MRoomTopic:
i.ContentType = "content.topic"
}
}
// New opens a new/existing fulltext index
func New(cfg config.Fulltext) (fts *Search, err error) {
fts = &Search{}
fts.FulltextIndex, err = openIndex(cfg)
if err != nil {
return nil, err
}
return fts, nil
}
// Close closes the fulltext index
func (f *Search) Close() error {
return f.FulltextIndex.Close()
}
// Index indexes the given elements
func (f *Search) Index(elements ...IndexElement) error {
batch := f.FulltextIndex.NewBatch()
for _, element := range elements {
err := batch.Index(element.EventID, element)
if err != nil {
return err
}
}
return f.FulltextIndex.Batch(batch)
}
// Delete deletes an indexed element by the eventID
func (f *Search) Delete(eventID string) error {
return f.FulltextIndex.Delete(eventID)
}
// Search searches the index given a search term, roomIDs and keys.
func (f *Search) Search(term string, roomIDs, keys []string, limit, from int, orderByStreamPos bool) (*bleve.SearchResult, error) {
qry := bleve.NewConjunctionQuery()
termQuery := bleve.NewBooleanQuery()
terms := strings.Split(term, " ")
for _, term := range terms {
matchQuery := bleve.NewMatchQuery(term)
matchQuery.SetField("Content")
termQuery.AddMust(matchQuery)
}
qry.AddQuery(termQuery)
roomQuery := bleve.NewBooleanQuery()
for _, roomID := range roomIDs {
roomSearch := bleve.NewMatchQuery(roomID)
roomSearch.SetField("RoomID")
roomQuery.AddShould(roomSearch)
}
if len(roomIDs) > 0 {
qry.AddQuery(roomQuery)
}
keyQuery := bleve.NewBooleanQuery()
for _, key := range keys {
keySearch := bleve.NewMatchQuery(key)
keySearch.SetField("ContentType")
keyQuery.AddShould(keySearch)
}
if len(keys) > 0 {
qry.AddQuery(keyQuery)
}
s := bleve.NewSearchRequestOptions(qry, limit, from, false)
s.Fields = []string{"*"}
s.SortBy([]string{"_score"})
if orderByStreamPos {
s.SortBy([]string{"-StreamPosition"})
}
return f.FulltextIndex.Search(s)
}
func openIndex(cfg config.Fulltext) (bleve.Index, error) {
m := getMapping(cfg)
if cfg.InMemory {
return bleve.NewMemOnly(m)
}
if index, err := bleve.Open(string(cfg.IndexPath)); err == nil {
return index, nil
}
index, err := bleve.New(string(cfg.IndexPath), m)
if err != nil {
return nil, err
}
return index, nil
}
func getMapping(cfg config.Fulltext) *mapping.IndexMappingImpl {
enFieldMapping := bleve.NewTextFieldMapping()
enFieldMapping.Analyzer = cfg.Language
eventMapping := bleve.NewDocumentMapping()
eventMapping.AddFieldMappingsAt("Content", enFieldMapping)
eventMapping.AddFieldMappingsAt("StreamPosition", bleve.NewNumericFieldMapping())
// Index entries as is
idFieldMapping := bleve.NewKeywordFieldMapping()
eventMapping.AddFieldMappingsAt("ContentType", idFieldMapping)
eventMapping.AddFieldMappingsAt("RoomID", idFieldMapping)
eventMapping.AddFieldMappingsAt("EventID", idFieldMapping)
indexMapping := bleve.NewIndexMapping()
indexMapping.AddDocumentMapping("Event", eventMapping)
indexMapping.DefaultType = "Event"
return indexMapping
}

View file

@ -0,0 +1,250 @@
// Copyright 2022 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 fulltext_test
import (
"reflect"
"testing"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"github.com/matrix-org/dendrite/internal/fulltext"
"github.com/matrix-org/dendrite/setup/config"
)
func mustOpenIndex(t *testing.T, tempDir string) *fulltext.Search {
t.Helper()
cfg := config.Fulltext{}
cfg.Defaults(config.DefaultOpts{
Generate: true,
Monolithic: true,
})
if tempDir != "" {
cfg.IndexPath = config.Path(tempDir)
cfg.InMemory = false
}
fts, err := fulltext.New(cfg)
if err != nil {
t.Fatal("failed to open fulltext index:", err)
}
return fts
}
func mustAddTestData(t *testing.T, fts *fulltext.Search, firstStreamPos int64) (eventIDs, roomIDs []string) {
t.Helper()
// create some more random data
var batchItems []fulltext.IndexElement
streamPos := firstStreamPos
wantRoomID := util.RandomString(16)
for i := 0; i < 30; i++ {
streamPos++
eventID := util.RandomString(16)
// Create more data for the first room
if i > 15 {
wantRoomID = util.RandomString(16)
}
e := fulltext.IndexElement{
EventID: eventID,
RoomID: wantRoomID,
Content: "lorem ipsum",
StreamPosition: streamPos,
}
e.SetContentType("m.room.message")
batchItems = append(batchItems, e)
roomIDs = append(roomIDs, wantRoomID)
eventIDs = append(eventIDs, eventID)
}
e := fulltext.IndexElement{
EventID: util.RandomString(16),
RoomID: wantRoomID,
Content: "Roomname testing",
StreamPosition: streamPos,
}
e.SetContentType(gomatrixserverlib.MRoomName)
batchItems = append(batchItems, e)
e = fulltext.IndexElement{
EventID: util.RandomString(16),
RoomID: wantRoomID,
Content: "Room topic fulltext",
StreamPosition: streamPos,
}
e.SetContentType(gomatrixserverlib.MRoomTopic)
batchItems = append(batchItems, e)
if err := fts.Index(batchItems...); err != nil {
t.Fatalf("failed to batch insert elements: %v", err)
}
return eventIDs, roomIDs
}
func TestOpen(t *testing.T) {
dataDir := t.TempDir()
fts := mustOpenIndex(t, dataDir)
if err := fts.Close(); err != nil {
t.Fatal("unable to close fulltext index", err)
}
// open existing index
fts = mustOpenIndex(t, dataDir)
defer fts.Close()
}
func TestIndex(t *testing.T) {
fts := mustOpenIndex(t, "")
defer fts.Close()
// add some data
var streamPos int64 = 1
roomID := util.RandomString(8)
eventID := util.RandomString(16)
e := fulltext.IndexElement{
EventID: eventID,
RoomID: roomID,
Content: "lorem ipsum",
StreamPosition: streamPos,
}
e.SetContentType("m.room.message")
if err := fts.Index(e); err != nil {
t.Fatal("failed to index element", err)
}
// create some more random data
mustAddTestData(t, fts, streamPos)
}
func TestDelete(t *testing.T) {
fts := mustOpenIndex(t, "")
defer fts.Close()
eventIDs, roomIDs := mustAddTestData(t, fts, 0)
res1, err := fts.Search("lorem", roomIDs[:1], nil, 50, 0, false)
if err != nil {
t.Fatal(err)
}
if err = fts.Delete(eventIDs[0]); err != nil {
t.Fatal(err)
}
res2, err := fts.Search("lorem", roomIDs[:1], nil, 50, 0, false)
if err != nil {
t.Fatal(err)
}
if res1.Total <= res2.Total {
t.Fatalf("got unexpected result: %d <= %d", res1.Total, res2.Total)
}
}
func TestSearch(t *testing.T) {
type args struct {
term string
keys []string
limit int
from int
orderByStreamPos bool
roomIndex []int
}
tests := []struct {
name string
args args
wantCount int
wantErr bool
}{
{
name: "Can search for many results in one room",
wantCount: 16,
args: args{
term: "lorem",
roomIndex: []int{0},
limit: 20,
},
},
{
name: "Can search for one result in one room",
wantCount: 1,
args: args{
term: "lorem",
roomIndex: []int{16},
limit: 20,
},
},
{
name: "Can search for many results in multiple rooms",
wantCount: 17,
args: args{
term: "lorem",
roomIndex: []int{0, 16},
limit: 20,
},
},
{
name: "Can search for many results in all rooms, reversed",
wantCount: 30,
args: args{
term: "lorem",
limit: 30,
orderByStreamPos: true,
},
},
{
name: "Can search for specific search room name",
wantCount: 1,
args: args{
term: "testing",
roomIndex: []int{},
limit: 20,
keys: []string{"content.name"},
},
},
{
name: "Can search for specific search room topic",
wantCount: 1,
args: args{
term: "fulltext",
roomIndex: []int{},
limit: 20,
keys: []string{"content.topic"},
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
f := mustOpenIndex(t, "")
eventIDs, roomIDs := mustAddTestData(t, f, 0)
var searchRooms []string
for _, x := range tt.args.roomIndex {
searchRooms = append(searchRooms, roomIDs[x])
}
t.Logf("searching in rooms: %v - %v\n", searchRooms, tt.args.keys)
got, err := f.Search(tt.args.term, searchRooms, tt.args.keys, tt.args.limit, tt.args.from, tt.args.orderByStreamPos)
if (err != nil) != tt.wantErr {
t.Errorf("Search() error = %v, wantErr %v", err, tt.wantErr)
return
}
if !reflect.DeepEqual(len(got.Hits), tt.wantCount) {
t.Errorf("Search() got = %v, want %v", len(got.Hits), tt.wantCount)
}
if tt.args.orderByStreamPos {
if got.Hits[0].ID != eventIDs[29] {
t.Fatalf("expected ID %s, got %s", eventIDs[29], got.Hits[0].ID)
}
}
})
}
}

View file

@ -0,0 +1,65 @@
// Copyright 2022 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 fulltext
import (
"github.com/matrix-org/dendrite/setup/config"
"time"
)
type Search struct{}
type IndexElement struct {
EventID string
RoomID string
Content string
ContentType string
StreamPosition int64
}
type SearchResult struct {
Status interface{} `json:"status"`
Request *interface{} `json:"request"`
Hits []interface{} `json:"hits"`
Total uint64 `json:"total_hits"`
MaxScore float64 `json:"max_score"`
Took time.Duration `json:"took"`
Facets interface{} `json:"facets"`
}
func (i *IndexElement) SetContentType(v string) {}
func New(cfg config.Fulltext) (fts *Search, err error) {
return &Search{}, nil
}
func (f *Search) Close() error {
return nil
}
func (f *Search) Index(e IndexElement) error {
return nil
}
func (f *Search) BatchIndex(elements []IndexElement) error {
return nil
}
func (f *Search) Delete(eventID string) error {
return nil
}
func (f *Search) Search(term string, roomIDs, keys []string, limit, from int, orderByStreamPos bool) (SearchResult, error) {
return SearchResult{}, nil
}

View file

@ -24,7 +24,7 @@ import (
"net/url"
"strings"
opentracing "github.com/opentracing/opentracing-go"
"github.com/opentracing/opentracing-go"
"github.com/opentracing/opentracing-go/ext"
)
@ -81,8 +81,8 @@ func PostJSON[reqtype, restype any, errtype error](
return fmt.Errorf("HTTP %d from %s (no response body)", res.StatusCode, apiURL)
}
var reserr errtype
if err = json.Unmarshal(body, reserr); err != nil {
return fmt.Errorf("HTTP %d from %s", res.StatusCode, apiURL)
if err = json.Unmarshal(body, &reserr); err != nil {
return fmt.Errorf("HTTP %d from %s - %w", res.StatusCode, apiURL, err)
}
return reserr
}

View file

@ -60,6 +60,9 @@ func MakeAuthAPI(
// add the user to Sentry, if enabled
hub := sentry.GetHubFromContext(req.Context())
if hub != nil {
hub.Scope().SetUser(sentry.User{
Username: device.UserID,
})
hub.Scope().SetTag("user_id", device.UserID)
hub.Scope().SetTag("device_id", device.ID)
}

View file

@ -16,6 +16,12 @@ func mRuleContainsUserNameDefinition(localpart string) *Rule {
Default: true,
Enabled: true,
Pattern: localpart,
Conditions: []*Condition{
{
Kind: EventMatchCondition,
Key: "content.body",
},
},
Actions: []*Action{
{Kind: NotifyAction},
{

View file

@ -7,8 +7,9 @@ func defaultOverrideRules(userID string) []*Rule {
mRuleInviteForMeDefinition(userID),
&mRuleMemberEventDefinition,
&mRuleContainsDisplayNameDefinition,
&mRuleTombstoneDefinition,
&mRuleRoomNotifDefinition,
&mRuleTombstoneDefinition,
&mRuleReactionDefinition,
}
}
@ -20,6 +21,7 @@ const (
MRuleContainsDisplayName = ".m.rule.contains_display_name"
MRuleTombstone = ".m.rule.tombstone"
MRuleRoomNotif = ".m.rule.roomnotif"
MRuleReaction = ".m.rule.reaction"
)
var (
@ -96,7 +98,7 @@ var (
{
Kind: SetTweakAction,
Tweak: HighlightTweak,
Value: false,
Value: true,
},
},
}
@ -120,10 +122,25 @@ var (
{
Kind: SetTweakAction,
Tweak: HighlightTweak,
Value: false,
Value: true,
},
},
}
mRuleReactionDefinition = Rule{
RuleID: MRuleReaction,
Default: true,
Enabled: true,
Conditions: []*Condition{
{
Kind: EventMatchCondition,
Key: "type",
Pattern: "m.reaction",
},
},
Actions: []*Action{
{Kind: DontNotifyAction},
},
}
)
func mRuleInviteForMeDefinition(userID string) *Rule {

View file

@ -10,8 +10,8 @@ const (
var defaultUnderrideRules = []*Rule{
&mRuleCallDefinition,
&mRuleEncryptedRoomOneToOneDefinition,
&mRuleRoomOneToOneDefinition,
&mRuleEncryptedRoomOneToOneDefinition,
&mRuleMessageDefinition,
&mRuleEncryptedDefinition,
}
@ -59,6 +59,11 @@ var (
},
Actions: []*Action{
{Kind: NotifyAction},
{
Kind: SetTweakAction,
Tweak: SoundTweak,
Value: "default",
},
{
Kind: SetTweakAction,
Tweak: HighlightTweak,
@ -88,6 +93,11 @@ var (
Tweak: HighlightTweak,
Value: false,
},
{
Kind: SetTweakAction,
Tweak: HighlightTweak,
Value: false,
},
},
}
mRuleMessageDefinition = Rule{
@ -101,7 +111,14 @@ var (
Pattern: "m.room.message",
},
},
Actions: []*Action{{Kind: NotifyAction}},
Actions: []*Action{
{Kind: NotifyAction},
{
Kind: SetTweakAction,
Tweak: HighlightTweak,
Value: false,
},
},
}
mRuleEncryptedDefinition = Rule{
RuleID: MRuleEncrypted,
@ -114,6 +131,13 @@ var (
Pattern: "m.room.encrypted",
},
},
Actions: []*Action{{Kind: NotifyAction}},
Actions: []*Action{
{Kind: NotifyAction},
{
Kind: SetTweakAction,
Tweak: HighlightTweak,
Value: false,
},
},
}
)

View file

@ -24,24 +24,28 @@ func TestRuleSetEvaluatorMatchEvent(t *testing.T) {
Default: false,
Enabled: true,
}
defaultRuleset := DefaultGlobalRuleSet("test", "test")
tsts := []struct {
Name string
RuleSet RuleSet
Want *Rule
Event *gomatrixserverlib.Event
}{
{"empty", RuleSet{}, nil},
{"defaultCanWin", RuleSet{Override: []*Rule{defaultEnabled}}, defaultEnabled},
{"userWins", RuleSet{Override: []*Rule{defaultEnabled, userEnabled}}, userEnabled},
{"defaultOverrideWins", RuleSet{Override: []*Rule{defaultEnabled}, Underride: []*Rule{userEnabled}}, defaultEnabled},
{"overrideContent", RuleSet{Override: []*Rule{userEnabled}, Content: []*Rule{userEnabled2}}, userEnabled},
{"overrideRoom", RuleSet{Override: []*Rule{userEnabled}, Room: []*Rule{userEnabled2}}, userEnabled},
{"overrideSender", RuleSet{Override: []*Rule{userEnabled}, Sender: []*Rule{userEnabled2}}, userEnabled},
{"overrideUnderride", RuleSet{Override: []*Rule{userEnabled}, Underride: []*Rule{userEnabled2}}, userEnabled},
{"empty", RuleSet{}, nil, ev},
{"defaultCanWin", RuleSet{Override: []*Rule{defaultEnabled}}, defaultEnabled, ev},
{"userWins", RuleSet{Override: []*Rule{defaultEnabled, userEnabled}}, userEnabled, ev},
{"defaultOverrideWins", RuleSet{Override: []*Rule{defaultEnabled}, Underride: []*Rule{userEnabled}}, defaultEnabled, ev},
{"overrideContent", RuleSet{Override: []*Rule{userEnabled}, Content: []*Rule{userEnabled2}}, userEnabled, ev},
{"overrideRoom", RuleSet{Override: []*Rule{userEnabled}, Room: []*Rule{userEnabled2}}, userEnabled, ev},
{"overrideSender", RuleSet{Override: []*Rule{userEnabled}, Sender: []*Rule{userEnabled2}}, userEnabled, ev},
{"overrideUnderride", RuleSet{Override: []*Rule{userEnabled}, Underride: []*Rule{userEnabled2}}, userEnabled, ev},
{"reactions don't notify", *defaultRuleset, &mRuleReactionDefinition, mustEventFromJSON(t, `{"type":"m.reaction"}`)},
{"receipts don't notify", *defaultRuleset, nil, mustEventFromJSON(t, `{"type":"m.receipt"}`)},
}
for _, tst := range tsts {
t.Run(tst.Name, func(t *testing.T) {
rse := NewRuleSetEvaluator(nil, &tst.RuleSet)
got, err := rse.MatchEvent(ev)
rse := NewRuleSetEvaluator(fakeEvaluationContext{3}, &tst.RuleSet)
got, err := rse.MatchEvent(tst.Event)
if err != nil {
t.Fatalf("MatchEvent failed: %v", err)
}
@ -128,7 +132,7 @@ func TestConditionMatches(t *testing.T) {
}
for _, tst := range tsts {
t.Run(tst.Name, func(t *testing.T) {
got, err := conditionMatches(&tst.Cond, mustEventFromJSON(t, tst.EventJSON), &fakeEvaluationContext{})
got, err := conditionMatches(&tst.Cond, mustEventFromJSON(t, tst.EventJSON), &fakeEvaluationContext{2})
if err != nil {
t.Fatalf("conditionMatches failed: %v", err)
}
@ -139,10 +143,10 @@ func TestConditionMatches(t *testing.T) {
}
}
type fakeEvaluationContext struct{}
type fakeEvaluationContext struct{ memberCount int }
func (fakeEvaluationContext) UserDisplayName() string { return "Dear User" }
func (fakeEvaluationContext) RoomMemberCount() (int, error) { return 2, nil }
func (fakeEvaluationContext) UserDisplayName() string { return "Dear User" }
func (f fakeEvaluationContext) RoomMemberCount() (int, error) { return f.memberCount, nil }
func (fakeEvaluationContext) HasPowerLevel(userID, levelKey string) (bool, error) {
return userID == "@poweruser:example.com" && levelKey == "powerlevel", nil
}

View file

@ -11,7 +11,7 @@ import (
// kind and a tweaks map. Returns a nil map if it would have been
// empty.
func ActionsToTweaks(as []*Action) (ActionKind, map[string]interface{}, error) {
var kind ActionKind
kind := UnknownAction
tweaks := map[string]interface{}{}
for _, a := range as {

View file

@ -21,8 +21,9 @@ import (
"sync"
"time"
"github.com/matrix-org/dendrite/internal"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/internal"
)
const createDBMigrationsSQL = "" +
@ -95,11 +96,11 @@ func (m *Migrator) Up(ctx context.Context) error {
for i := range m.migrations {
now := time.Now().UTC().Format(time.RFC3339)
migration := m.migrations[i]
logrus.Debugf("Executing database migration '%s'", migration.Version)
// Skip migration if it was already executed
if _, ok := executedMigrations[migration.Version]; ok {
continue
}
logrus.Debugf("Executing database migration '%s'", migration.Version)
err = migration.Up(ctx, txn)
if err != nil {
return fmt.Errorf("unable to execute migration '%s': %w", migration.Version, err)
@ -140,3 +141,19 @@ func (m *Migrator) ExecutedMigrations(ctx context.Context) (map[string]struct{},
return result, rows.Err()
}
// InsertMigration creates the migrations table if it doesn't exist and
// inserts a migration given their name to the database.
// This should only be used when manually inserting migrations.
func InsertMigration(ctx context.Context, db *sql.DB, migrationName string) error {
_, err := db.ExecContext(ctx, createDBMigrationsSQL)
if err != nil {
return fmt.Errorf("unable to create db_migrations: %w", err)
}
_, err = db.ExecContext(ctx, insertVersionSQL,
migrationName,
time.Now().Format(time.RFC3339),
internal.VersionString(),
)
return err
}

View file

@ -17,7 +17,7 @@ var build string
const (
VersionMajor = 0
VersionMinor = 9
VersionPatch = 6
VersionPatch = 7
VersionTag = "" // example: "rc1"
)

View file

@ -18,22 +18,24 @@ import (
"context"
"encoding/json"
"github.com/matrix-org/gomatrixserverlib"
"github.com/nats-io/nats.go"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/keyserver/internal"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
"github.com/nats-io/nats.go"
"github.com/sirupsen/logrus"
)
// DeviceListUpdateConsumer consumes device list updates that came in over federation.
type DeviceListUpdateConsumer struct {
ctx context.Context
jetstream nats.JetStreamContext
durable string
topic string
updater *internal.DeviceListUpdater
ctx context.Context
jetstream nats.JetStreamContext
durable string
topic string
updater *internal.DeviceListUpdater
serverName gomatrixserverlib.ServerName
}
// NewDeviceListUpdateConsumer creates a new DeviceListConsumer. Call Start() to begin consuming from key servers.
@ -44,11 +46,12 @@ func NewDeviceListUpdateConsumer(
updater *internal.DeviceListUpdater,
) *DeviceListUpdateConsumer {
return &DeviceListUpdateConsumer{
ctx: process.Context(),
jetstream: js,
durable: cfg.Matrix.JetStream.Prefixed("KeyServerInputDeviceListConsumer"),
topic: cfg.Matrix.JetStream.Prefixed(jetstream.InputDeviceListUpdate),
updater: updater,
ctx: process.Context(),
jetstream: js,
durable: cfg.Matrix.JetStream.Prefixed("KeyServerInputDeviceListConsumer"),
topic: cfg.Matrix.JetStream.Prefixed(jetstream.InputDeviceListUpdate),
updater: updater,
serverName: cfg.Matrix.ServerName,
}
}
@ -69,6 +72,15 @@ func (t *DeviceListUpdateConsumer) onMessage(ctx context.Context, msgs []*nats.M
logrus.WithError(err).Errorf("Failed to read from device list update input topic")
return true
}
origin := gomatrixserverlib.ServerName(msg.Header.Get("origin"))
if _, serverName, err := gomatrixserverlib.SplitID('@', m.UserID); err != nil {
return true
} else if serverName == t.serverName {
return true
} else if serverName != origin {
return true
}
err := t.updater.Update(ctx, m)
if err != nil {
logrus.WithFields(logrus.Fields{

View file

@ -0,0 +1,110 @@
// Copyright 2022 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 consumers
import (
"context"
"encoding/json"
"github.com/matrix-org/gomatrixserverlib"
"github.com/nats-io/nats.go"
"github.com/sirupsen/logrus"
keyapi "github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/dendrite/keyserver/internal"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
)
// SigningKeyUpdateConsumer consumes signing key updates that came in over federation.
type SigningKeyUpdateConsumer struct {
ctx context.Context
jetstream nats.JetStreamContext
durable string
topic string
keyAPI *internal.KeyInternalAPI
cfg *config.KeyServer
}
// NewSigningKeyUpdateConsumer creates a new SigningKeyUpdateConsumer. Call Start() to begin consuming from key servers.
func NewSigningKeyUpdateConsumer(
process *process.ProcessContext,
cfg *config.KeyServer,
js nats.JetStreamContext,
keyAPI *internal.KeyInternalAPI,
) *SigningKeyUpdateConsumer {
return &SigningKeyUpdateConsumer{
ctx: process.Context(),
jetstream: js,
durable: cfg.Matrix.JetStream.Prefixed("KeyServerSigningKeyConsumer"),
topic: cfg.Matrix.JetStream.Prefixed(jetstream.InputSigningKeyUpdate),
keyAPI: keyAPI,
cfg: cfg,
}
}
// Start consuming from key servers
func (t *SigningKeyUpdateConsumer) Start() error {
return jetstream.JetStreamConsumer(
t.ctx, t.jetstream, t.topic, t.durable, 1,
t.onMessage, nats.DeliverAll(), nats.ManualAck(),
)
}
// onMessage is called in response to a message received on the
// signing key update events topic from the key server.
func (t *SigningKeyUpdateConsumer) onMessage(ctx context.Context, msgs []*nats.Msg) bool {
msg := msgs[0] // Guaranteed to exist if onMessage is called
var updatePayload keyapi.CrossSigningKeyUpdate
if err := json.Unmarshal(msg.Data, &updatePayload); err != nil {
logrus.WithError(err).Errorf("Failed to read from signing key update input topic")
return true
}
origin := gomatrixserverlib.ServerName(msg.Header.Get("origin"))
if _, serverName, err := gomatrixserverlib.SplitID('@', updatePayload.UserID); err != nil {
logrus.WithError(err).Error("failed to split user id")
return true
} else if serverName == t.cfg.Matrix.ServerName {
logrus.Warn("dropping device key update from ourself")
return true
} else if serverName != origin {
logrus.Warnf("dropping device key update, %s != %s", serverName, origin)
return true
}
keys := gomatrixserverlib.CrossSigningKeys{}
if updatePayload.MasterKey != nil {
keys.MasterKey = *updatePayload.MasterKey
}
if updatePayload.SelfSigningKey != nil {
keys.SelfSigningKey = *updatePayload.SelfSigningKey
}
uploadReq := &keyapi.PerformUploadDeviceKeysRequest{
CrossSigningKeys: keys,
UserID: updatePayload.UserID,
}
uploadRes := &keyapi.PerformUploadDeviceKeysResponse{}
if err := t.keyAPI.PerformUploadDeviceKeys(ctx, uploadReq, uploadRes); err != nil {
logrus.WithError(err).Error("failed to upload device keys")
return false
}
if uploadRes.Error != nil {
logrus.WithError(uploadRes.Error).Error("failed to upload device keys")
return true
}
return true
}

View file

@ -19,9 +19,11 @@ import (
"encoding/json"
"fmt"
"hash/fnv"
"net"
"sync"
"time"
"github.com/matrix-org/gomatrix"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"github.com/prometheus/client_golang/prometheus"
@ -165,6 +167,7 @@ func (u *DeviceListUpdater) Start() error {
step = (time.Second * 120) / time.Duration(max)
}
for _, userID := range staleLists {
userID := userID // otherwise we are only sending the last entry
time.AfterFunc(offset, func() {
u.notifyWorkers(userID)
})
@ -388,28 +391,59 @@ func (u *DeviceListUpdater) processServer(serverName gomatrixserverlib.ServerNam
return waitTime, true
}
failCount := 0
userLoop:
for _, userID := range userIDs {
if ctx.Err() != nil {
// we've timed out, give up and go to the back of the queue to let another server be processed.
failCount += 1
waitTime = time.Minute * 10
break
}
res, err := u.fedClient.GetUserDevices(ctx, serverName, userID)
if err != nil {
failCount += 1
fcerr, ok := err.(*fedsenderapi.FederationClientError)
if ok {
if fcerr.RetryAfter > 0 {
waitTime = fcerr.RetryAfter
} else if fcerr.Blacklisted {
select {
case <-ctx.Done():
// we've timed out, give up and go to the back of the queue to let another server be processed.
waitTime = time.Minute * 10
break userLoop
default:
}
switch e := err.(type) {
case *fedsenderapi.FederationClientError:
if e.RetryAfter > 0 {
waitTime = e.RetryAfter
} else if e.Blacklisted {
waitTime = time.Hour * 8
} else {
// For all other errors (DNS resolution, network etc.) wait 1 hour.
break userLoop
} else if e.Code >= 300 {
// We didn't get a real FederationClientError (e.g. in polylith mode, where gomatrix.HTTPError
// are "converted" to FederationClientError), but we probably shouldn't hit them every $waitTime seconds.
waitTime = time.Hour
break userLoop
}
} else {
waitTime = time.Hour
logger.WithError(err).WithField("user_id", userID).Debug("GetUserDevices returned unknown error type")
case net.Error:
// Use the default waitTime, if it's a timeout.
// It probably doesn't make sense to try further users.
if !e.Timeout() {
waitTime = time.Minute * 10
logger.WithError(e).Error("GetUserDevices returned net.Error")
break userLoop
}
case gomatrix.HTTPError:
// The remote server returned an error, give it some time to recover.
// This is to avoid spamming remote servers, which may not be Matrix servers anymore.
if e.Code >= 300 {
waitTime = time.Hour
logger.WithError(e).Error("GetUserDevices returned gomatrix.HTTPError")
break userLoop
}
default:
// Something else failed
waitTime = time.Minute * 10
logger.WithError(err).WithField("user_id", userID).Debugf("GetUserDevices returned unknown error type: %T", err)
break userLoop
}
continue
}
@ -437,7 +471,12 @@ func (u *DeviceListUpdater) processServer(serverName gomatrixserverlib.ServerNam
}
}
if failCount > 0 {
logger.WithField("total", len(userIDs)).WithField("failed", failCount).WithField("wait", waitTime).Warn("Failed to query device keys for some users")
logger.WithFields(logrus.Fields{
"total": len(userIDs),
"failed": failCount,
"skipped": len(userIDs) - failCount,
"waittime": waitTime,
}).Warn("Failed to query device keys for some users")
}
for _, userID := range userIDs {
// always clear the channel to unblock Update calls regardless of success/failure

View file

@ -16,6 +16,8 @@ package keyserver
import (
"github.com/gorilla/mux"
"github.com/sirupsen/logrus"
fedsenderapi "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/dendrite/keyserver/consumers"
@ -26,7 +28,6 @@ import (
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/sirupsen/logrus"
)
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
@ -72,5 +73,12 @@ func NewInternalAPI(
logrus.WithError(err).Panic("failed to start device list consumer")
}
sigConsumer := consumers.NewSigningKeyUpdateConsumer(
base.ProcessContext, cfg, js, ap,
)
if err := sigConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start signing key consumer")
}
return ap
}

View file

@ -17,8 +17,9 @@ package postgres
import (
"context"
"database/sql"
"errors"
"github.com/lib/pq"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
@ -63,30 +64,37 @@ func NewPostgresKeyChangesTable(db *sql.DB) (tables.KeyChanges, error) {
return s, err
}
if err = executeMigration(context.Background(), db); err != nil {
return nil, err
}
return s, nil
}
func executeMigration(ctx context.Context, db *sql.DB) error {
// TODO: Remove when we are sure we are not having goose artefacts in the db
// This forces an error, which indicates the migration is already applied, since the
// column partition was removed from the table
var count int
err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan(&count)
if err == nil {
m := sqlutil.NewMigrator(db)
m.AddMigrations(sqlutil.Migration{
Version: "keyserver: refactor key changes",
Up: deltas.UpRefactorKeyChanges,
})
return s, m.Up(context.Background())
} else {
switch e := err.(type) {
case *pq.Error:
// ignore undefined_column (42703) errors, as this is expected at this point
if e.Code != "42703" {
return nil, err
migrationName := "keyserver: refactor key changes"
var cName string
err := db.QueryRowContext(ctx, "select column_name from information_schema.columns where table_name = 'keyserver_key_changes' AND column_name = 'partition'").Scan(&cName)
if err != nil {
if errors.Is(err, sql.ErrNoRows) { // migration was already executed, as the column was removed
if err = sqlutil.InsertMigration(ctx, db, migrationName); err != nil {
// not a fatal error, log and continue
logrus.WithError(err).Warnf("unable to manually insert migration '%s'", migrationName)
}
default:
return nil, err
return nil
}
return err
}
return s, nil
m := sqlutil.NewMigrator(db)
m.AddMigrations(sqlutil.Migration{
Version: migrationName,
Up: deltas.UpRefactorKeyChanges,
})
return m.Up(ctx)
}
func (s *keyChangesStatements) Prepare() (err error) {

View file

@ -158,7 +158,7 @@ func (d *Database) MarkDeviceListStale(ctx context.Context, userID string, isSta
// DeleteDeviceKeys removes the device keys for a given user/device, and any accompanying
// cross-signing signatures relating to that device.
func (d *Database) DeleteDeviceKeys(ctx context.Context, userID string, deviceIDs []gomatrixserverlib.KeyID) error {
return d.Writer.Do(nil, nil, func(txn *sql.Tx) error {
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
for _, deviceID := range deviceIDs {
if err := d.CrossSigningSigsTable.DeleteCrossSigningSigsForTarget(ctx, txn, userID, deviceID); err != nil && err != sql.ErrNoRows {
return fmt.Errorf("d.CrossSigningSigsTable.DeleteCrossSigningSigsForTarget: %w", err)

View file

@ -17,6 +17,9 @@ package sqlite3
import (
"context"
"database/sql"
"errors"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
@ -58,23 +61,40 @@ func NewSqliteKeyChangesTable(db *sql.DB) (tables.KeyChanges, error) {
if err != nil {
return s, err
}
// TODO: Remove when we are sure we are not having goose artefacts in the db
// This forces an error, which indicates the migration is already applied, since the
// column partition was removed from the table
var count int
err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan(&count)
if err == nil {
m := sqlutil.NewMigrator(db)
m.AddMigrations(sqlutil.Migration{
Version: "keyserver: refactor key changes",
Up: deltas.UpRefactorKeyChanges,
})
return s, m.Up(context.Background())
if err = executeMigration(context.Background(), db); err != nil {
return nil, err
}
return s, nil
}
func executeMigration(ctx context.Context, db *sql.DB) error {
// TODO: Remove when we are sure we are not having goose artefacts in the db
// This forces an error, which indicates the migration is already applied, since the
// column partition was removed from the table
migrationName := "keyserver: refactor key changes"
var cName string
err := db.QueryRowContext(ctx, `SELECT p.name FROM sqlite_master AS m JOIN pragma_table_info(m.name) AS p WHERE m.name = 'keyserver_key_changes' AND p.name = 'partition'`).Scan(&cName)
if err != nil {
if errors.Is(err, sql.ErrNoRows) { // migration was already executed, as the column was removed
if err = sqlutil.InsertMigration(ctx, db, migrationName); err != nil {
// not a fatal error, log and continue
logrus.WithError(err).Warnf("unable to manually insert migration '%s'", migrationName)
}
return nil
}
return err
}
m := sqlutil.NewMigrator(db)
m.AddMigrations(sqlutil.Migration{
Version: migrationName,
Up: deltas.UpRefactorKeyChanges,
})
return m.Up(ctx)
}
func (s *keyChangesStatements) Prepare() (err error) {
if s.upsertKeyChangeStmt, err = s.db.Prepare(upsertKeyChangeSQL); err != nil {
return err

View file

@ -188,6 +188,9 @@ func (w *worker) _next() {
// Look up what the next event is that's waiting to be processed.
ctx, cancel := context.WithTimeout(w.r.ProcessContext.Context(), time.Minute)
defer cancel()
if scope := sentry.CurrentHub().Scope(); scope != nil {
scope.SetTag("room_id", w.roomID)
}
msgs, err := w.subscription.Fetch(1, nats.Context(ctx))
switch err {
case nil:
@ -239,6 +242,9 @@ func (w *worker) _next() {
return
}
if scope := sentry.CurrentHub().Scope(); scope != nil {
scope.SetTag("event_id", inputRoomEvent.Event.EventID())
}
roomserverInputBackpressure.With(prometheus.Labels{"room_id": w.roomID}).Inc()
defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": w.roomID}).Dec()

View file

@ -20,6 +20,7 @@ import (
"context"
"fmt"
"github.com/getsentry/sentry-go"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"github.com/opentracing/opentracing-go"
@ -178,6 +179,10 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
u.newStateNID = u.oldStateNID
}
if err = u.updater.SetLatestEvents(u.roomInfo.RoomNID, u.latest, u.stateAtEvent.EventNID, u.newStateNID); err != nil {
return fmt.Errorf("u.updater.SetLatestEvents: %w", err)
}
update, err := u.makeOutputNewRoomEvent()
if err != nil {
return fmt.Errorf("u.makeOutputNewRoomEvent: %w", err)
@ -196,10 +201,6 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
return fmt.Errorf("u.api.WriteOutputEvents: %w", err)
}
if err = u.updater.SetLatestEvents(u.roomInfo.RoomNID, u.latest, u.stateAtEvent.EventNID, u.newStateNID); err != nil {
return fmt.Errorf("u.updater.SetLatestEvents: %w", err)
}
if err = u.updater.MarkEventAsSent(u.stateAtEvent.EventNID); err != nil {
return fmt.Errorf("u.updater.MarkEventAsSent: %w", err)
}
@ -275,7 +276,7 @@ func (u *latestEventsUpdater) latestState() error {
return fmt.Errorf("roomState.DifferenceBetweenStateSnapshots: %w", err)
}
if removed := len(u.removed) - len(u.added); removed > 0 {
if removed := len(u.removed) - len(u.added); !u.rewritesState && removed > 0 {
logrus.WithFields(logrus.Fields{
"event_id": u.event.EventID(),
"room_id": u.event.RoomID(),
@ -283,7 +284,19 @@ func (u *latestEventsUpdater) latestState() error {
"new_state_nid": u.newStateNID,
"old_latest": u.oldLatest.EventIDs(),
"new_latest": u.latest.EventIDs(),
}).Errorf("Unexpected state deletion (removing %d events)", removed)
}).Warnf("State reset detected (removing %d events)", removed)
sentry.WithScope(func(scope *sentry.Scope) {
scope.SetLevel("warning")
scope.SetContext("State reset", map[string]interface{}{
"Event ID": u.event.EventID(),
"Old state NID": fmt.Sprintf("%d", u.oldStateNID),
"New state NID": fmt.Sprintf("%d", u.newStateNID),
"Old latest": u.oldLatest.EventIDs(),
"New latest": u.latest.EventIDs(),
"State removed": removed,
})
sentry.CaptureMessage("State reset detected")
})
}
// Also work out the state before the event removes and the event

View file

@ -16,10 +16,13 @@
package postgres
import (
"context"
"database/sql"
"errors"
"fmt"
"github.com/lib/pq"
"github.com/sirupsen/logrus"
// Import the postgres database driver.
_ "github.com/lib/pq"
@ -52,30 +55,8 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
// Special case, since this migration uses several tables, so it needs to
// be sure that all tables are created first.
// TODO: Remove when we are sure we are not having goose artefacts in the db
// This forces an error, which indicates the migration is already applied, since the
// column event_nid was removed from the table
var eventNID int
err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan(&eventNID)
if err == nil {
m := sqlutil.NewMigrator(db)
m.AddMigrations(sqlutil.Migration{
Version: "roomserver: state blocks refactor",
Up: deltas.UpStateBlocksRefactor,
})
if err = m.Up(base.Context()); err != nil {
return nil, err
}
} else {
switch e := err.(type) {
case *pq.Error:
// ignore undefined_column (42703) errors, as this is expected at this point
if e.Code != "42703" {
return nil, err
}
default:
return nil, err
}
if err = executeMigration(base.Context(), db); err != nil {
return nil, err
}
// Then prepare the statements. Now that the migrations have run, any columns referred
@ -87,6 +68,33 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
return &d, nil
}
func executeMigration(ctx context.Context, db *sql.DB) error {
// TODO: Remove when we are sure we are not having goose artefacts in the db
// This forces an error, which indicates the migration is already applied, since the
// column event_nid was removed from the table
migrationName := "roomserver: state blocks refactor"
var cName string
err := db.QueryRowContext(ctx, "select column_name from information_schema.columns where table_name = 'roomserver_state_block' AND column_name = 'event_nid'").Scan(&cName)
if err != nil {
if errors.Is(err, sql.ErrNoRows) { // migration was already executed, as the column was removed
if err = sqlutil.InsertMigration(ctx, db, migrationName); err != nil {
// not a fatal error, log and continue
logrus.WithError(err).Warnf("unable to manually insert migration '%s'", migrationName)
}
return nil
}
return err
}
m := sqlutil.NewMigrator(db)
m.AddMigrations(sqlutil.Migration{
Version: migrationName,
Up: deltas.UpStateBlocksRefactor,
})
return m.Up(ctx)
}
func (d *Database) create(db *sql.DB) error {
if err := CreateEventStateKeysTable(db); err != nil {
return err

View file

@ -26,11 +26,11 @@ func NewMembershipUpdater(
var targetUserNID types.EventStateKeyNID
var err error
err = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error {
roomNID, err = d.assignRoomNID(ctx, roomID, roomVersion)
roomNID, err = d.assignRoomNID(ctx, txn, roomID, roomVersion)
if err != nil {
return err
}
targetUserNID, err = d.assignStateKeyNID(ctx, targetUserID)
targetUserNID, err = d.assignStateKeyNID(ctx, txn, targetUserID)
if err != nil {
return err
}
@ -101,7 +101,7 @@ func (u *MembershipUpdater) Update(newMembership tables.MembershipState, event *
var inserted bool // Did the query result in a membership change?
var retired []string // Did we retire any updates in the process?
return inserted, retired, u.d.Writer.Do(u.d.DB, u.txn, func(txn *sql.Tx) error {
senderUserNID, err := u.d.assignStateKeyNID(u.ctx, event.Sender())
senderUserNID, err := u.d.assignStateKeyNID(u.ctx, u.txn, event.Sender())
if err != nil {
return fmt.Errorf("u.d.AssignStateKeyNID: %w", err)
}

View file

@ -402,7 +402,7 @@ func (d *Database) RemoveRoomAlias(ctx context.Context, alias string) error {
func (d *Database) GetMembership(ctx context.Context, roomNID types.RoomNID, requestSenderUserID string) (membershipEventNID types.EventNID, stillInRoom, isRoomforgotten bool, err error) {
var requestSenderUserNID types.EventStateKeyNID
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
requestSenderUserNID, err = d.assignStateKeyNID(ctx, requestSenderUserID)
requestSenderUserNID, err = d.assignStateKeyNID(ctx, txn, requestSenderUserID)
return err
})
if err != nil {
@ -596,7 +596,9 @@ func (d *Database) storeEvent(
if updater != nil && updater.txn != nil {
txn = updater.txn
}
err = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error {
// First writer is with a database-provided transaction, so that NIDs are assigned
// globally outside of the updater context, to help avoid races.
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
// TODO: Here we should aim to have two different code paths for new rooms
// vs existing ones.
@ -611,11 +613,11 @@ func (d *Database) storeEvent(
return fmt.Errorf("extractRoomVersionFromCreateEvent: %w", err)
}
if roomNID, err = d.assignRoomNID(ctx, event.RoomID(), roomVersion); err != nil {
if roomNID, err = d.assignRoomNID(ctx, txn, event.RoomID(), roomVersion); err != nil {
return fmt.Errorf("d.assignRoomNID: %w", err)
}
if eventTypeNID, err = d.assignEventTypeNID(ctx, event.Type()); err != nil {
if eventTypeNID, err = d.assignEventTypeNID(ctx, txn, event.Type()); err != nil {
return fmt.Errorf("d.assignEventTypeNID: %w", err)
}
@ -623,11 +625,19 @@ func (d *Database) storeEvent(
// Assigned a numeric ID for the state_key if there is one present.
// Otherwise set the numeric ID for the state_key to 0.
if eventStateKey != nil {
if eventStateKeyNID, err = d.assignStateKeyNID(ctx, *eventStateKey); err != nil {
if eventStateKeyNID, err = d.assignStateKeyNID(ctx, txn, *eventStateKey); err != nil {
return fmt.Errorf("d.assignStateKeyNID: %w", err)
}
}
return nil
})
if err != nil {
return 0, 0, types.StateAtEvent{}, nil, "", fmt.Errorf("d.Writer.Do: %w", err)
}
// Second writer is using the database-provided transaction, probably from the
// room updater, for easy roll-back if required.
err = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error {
if eventNID, stateNID, err = d.EventsTable.InsertEvent(
ctx,
txn,
@ -749,48 +759,48 @@ func (d *Database) MissingAuthPrevEvents(
}
func (d *Database) assignRoomNID(
ctx context.Context, roomID string, roomVersion gomatrixserverlib.RoomVersion,
ctx context.Context, txn *sql.Tx, roomID string, roomVersion gomatrixserverlib.RoomVersion,
) (types.RoomNID, error) {
// Check if we already have a numeric ID in the database.
roomNID, err := d.RoomsTable.SelectRoomNID(ctx, nil, roomID)
roomNID, err := d.RoomsTable.SelectRoomNID(ctx, txn, roomID)
if err == sql.ErrNoRows {
// We don't have a numeric ID so insert one into the database.
roomNID, err = d.RoomsTable.InsertRoomNID(ctx, nil, roomID, roomVersion)
roomNID, err = d.RoomsTable.InsertRoomNID(ctx, txn, roomID, roomVersion)
if err == sql.ErrNoRows {
// We raced with another insert so run the select again.
roomNID, err = d.RoomsTable.SelectRoomNID(ctx, nil, roomID)
roomNID, err = d.RoomsTable.SelectRoomNID(ctx, txn, roomID)
}
}
return roomNID, err
}
func (d *Database) assignEventTypeNID(
ctx context.Context, eventType string,
ctx context.Context, txn *sql.Tx, eventType string,
) (types.EventTypeNID, error) {
// Check if we already have a numeric ID in the database.
eventTypeNID, err := d.EventTypesTable.SelectEventTypeNID(ctx, nil, eventType)
eventTypeNID, err := d.EventTypesTable.SelectEventTypeNID(ctx, txn, eventType)
if err == sql.ErrNoRows {
// We don't have a numeric ID so insert one into the database.
eventTypeNID, err = d.EventTypesTable.InsertEventTypeNID(ctx, nil, eventType)
eventTypeNID, err = d.EventTypesTable.InsertEventTypeNID(ctx, txn, eventType)
if err == sql.ErrNoRows {
// We raced with another insert so run the select again.
eventTypeNID, err = d.EventTypesTable.SelectEventTypeNID(ctx, nil, eventType)
eventTypeNID, err = d.EventTypesTable.SelectEventTypeNID(ctx, txn, eventType)
}
}
return eventTypeNID, err
}
func (d *Database) assignStateKeyNID(
ctx context.Context, eventStateKey string,
ctx context.Context, txn *sql.Tx, eventStateKey string,
) (types.EventStateKeyNID, error) {
// Check if we already have a numeric ID in the database.
eventStateKeyNID, err := d.EventStateKeysTable.SelectEventStateKeyNID(ctx, nil, eventStateKey)
eventStateKeyNID, err := d.EventStateKeysTable.SelectEventStateKeyNID(ctx, txn, eventStateKey)
if err == sql.ErrNoRows {
// We don't have a numeric ID so insert one into the database.
eventStateKeyNID, err = d.EventStateKeysTable.InsertEventStateKeyNID(ctx, nil, eventStateKey)
eventStateKeyNID, err = d.EventStateKeysTable.InsertEventStateKeyNID(ctx, txn, eventStateKey)
if err == sql.ErrNoRows {
// We raced with another insert so run the select again.
eventStateKeyNID, err = d.EventStateKeysTable.SelectEventStateKeyNID(ctx, nil, eventStateKey)
eventStateKeyNID, err = d.EventStateKeysTable.SelectEventStateKeyNID(ctx, txn, eventStateKey)
}
}
return eventStateKeyNID, err

View file

@ -18,9 +18,11 @@ package sqlite3
import (
"context"
"database/sql"
"errors"
"fmt"
"github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/sqlutil"
@ -61,20 +63,8 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
// Special case, since this migration uses several tables, so it needs to
// be sure that all tables are created first.
// TODO: Remove when we are sure we are not having goose artefacts in the db
// This forces an error, which indicates the migration is already applied, since the
// column event_nid was removed from the table
var eventNID int
err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan(&eventNID)
if err == nil {
m := sqlutil.NewMigrator(db)
m.AddMigrations(sqlutil.Migration{
Version: "roomserver: state blocks refactor",
Up: deltas.UpStateBlocksRefactor,
})
if err = m.Up(base.Context()); err != nil {
return nil, err
}
if err = executeMigration(base.Context(), db); err != nil {
return nil, err
}
// Then prepare the statements. Now that the migrations have run, any columns referred
@ -86,6 +76,32 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
return &d, nil
}
func executeMigration(ctx context.Context, db *sql.DB) error {
// TODO: Remove when we are sure we are not having goose artefacts in the db
// This forces an error, which indicates the migration is already applied, since the
// column event_nid was removed from the table
migrationName := "roomserver: state blocks refactor"
var cName string
err := db.QueryRowContext(ctx, `SELECT p.name FROM sqlite_master AS m JOIN pragma_table_info(m.name) AS p WHERE m.name = 'roomserver_state_block' AND p.name = 'event_nid'`).Scan(&cName)
if err != nil {
if errors.Is(err, sql.ErrNoRows) { // migration was already executed, as the column was removed
if err = sqlutil.InsertMigration(ctx, db, migrationName); err != nil {
// not a fatal error, log and continue
logrus.WithError(err).Warnf("unable to manually insert migration '%s'", migrationName)
}
return nil
}
return err
}
m := sqlutil.NewMigrator(db)
m.AddMigrations(sqlutil.Migration{
Version: migrationName,
Up: deltas.UpStateBlocksRefactor,
})
return m.Up(ctx)
}
func (d *Database) create(db *sql.DB) error {
if err := CreateEventStateKeysTable(db); err != nil {
return err

View file

@ -38,6 +38,7 @@ import (
"golang.org/x/net/http2/h2c"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/fulltext"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/pushgateway"
"github.com/matrix-org/dendrite/internal/sqlutil"
@ -90,6 +91,7 @@ type BaseDendrite struct {
Database *sql.DB
DatabaseWriter sqlutil.Writer
EnableMetrics bool
Fulltext *fulltext.Search
startupLock sync.Mutex
}
@ -150,6 +152,15 @@ func NewBaseDendrite(cfg *config.Dendrite, componentName string, options ...Base
logrus.WithError(err).Panicf("failed to start opentracing")
}
var fts *fulltext.Search
isSyncOrMonolith := componentName == "syncapi" || isMonolith
if cfg.SyncAPI.Fulltext.Enabled && isSyncOrMonolith {
fts, err = fulltext.New(cfg.SyncAPI.Fulltext)
if err != nil {
logrus.WithError(err).Panicf("failed to create full text")
}
}
if cfg.Global.Sentry.Enabled {
logrus.Info("Setting up Sentry for debugging...")
err = sentry.Init(sentry.ClientOptions{
@ -247,6 +258,7 @@ func NewBaseDendrite(cfg *config.Dendrite, componentName string, options ...Base
Database: db, // set if monolith with global connection pool only
DatabaseWriter: writer, // set if monolith with global connection pool only
EnableMetrics: enableMetrics,
Fulltext: fts,
}
}

View file

@ -9,6 +9,8 @@ type SyncAPI struct {
Database DatabaseOptions `yaml:"database,omitempty"`
RealIPHeader string `yaml:"real_ip_header"`
Fulltext Fulltext `yaml:"fulltext"`
}
func (c *SyncAPI) Defaults(opts DefaultOpts) {
@ -18,6 +20,7 @@ func (c *SyncAPI) Defaults(opts DefaultOpts) {
c.ExternalAPI.Listen = "http://localhost:8073"
c.Database.Defaults(20)
}
c.Fulltext.Defaults(opts)
if opts.Generate {
if !opts.Monolithic {
c.Database.ConnectionString = "file:syncapi.db"
@ -26,6 +29,7 @@ func (c *SyncAPI) Defaults(opts DefaultOpts) {
}
func (c *SyncAPI) Verify(configErrs *ConfigErrors, isMonolith bool) {
c.Fulltext.Verify(configErrs, isMonolith)
if isMonolith { // polylith required configs below
return
}
@ -36,3 +40,28 @@ func (c *SyncAPI) Verify(configErrs *ConfigErrors, isMonolith bool) {
checkURL(configErrs, "sync_api.internal_api.connect", string(c.InternalAPI.Connect))
checkURL(configErrs, "sync_api.external_api.listen", string(c.ExternalAPI.Listen))
}
type Fulltext struct {
Enabled bool `yaml:"enabled"`
IndexPath Path `yaml:"index_path"`
InMemory bool `yaml:"in_memory"` // only useful in tests
Language string `yaml:"language"` // the language to use when analysing content
}
func (f *Fulltext) Defaults(opts DefaultOpts) {
f.Enabled = false
f.IndexPath = "./fulltextindex"
f.Language = "en"
if opts.Generate {
f.Enabled = true
f.InMemory = true
}
}
func (f *Fulltext) Verify(configErrs *ConfigErrors, isMonolith bool) {
if !f.Enabled {
return
}
checkNotEmpty(configErrs, "syncapi.fulltext.index_path", string(f.IndexPath))
checkNotEmpty(configErrs, "syncapi.fulltext.language", f.Language)
}

View file

@ -17,6 +17,7 @@ const (
var (
InputRoomEvent = "InputRoomEvent"
InputDeviceListUpdate = "InputDeviceListUpdate"
InputSigningKeyUpdate = "InputSigningKeyUpdate"
OutputRoomEvent = "OutputRoomEvent"
OutputSendToDeviceEvent = "OutputSendToDeviceEvent"
OutputKeyChangeEvent = "OutputKeyChangeEvent"
@ -28,6 +29,7 @@ var (
OutputReadUpdate = "OutputReadUpdate"
RequestPresence = "GetPresence"
OutputPresenceEvent = "OutputPresenceEvent"
InputFulltextReindex = "InputFulltextReindex"
)
var safeCharacters = regexp.MustCompile("[^A-Za-z0-9$]+")
@ -51,6 +53,11 @@ var streams = []*nats.StreamConfig{
Retention: nats.InterestPolicy,
Storage: nats.FileStorage,
},
{
Name: InputSigningKeyUpdate,
Retention: nats.InterestPolicy,
Storage: nats.FileStorage,
},
{
Name: OutputRoomEvent,
Retention: nats.InterestPolicy,

View file

@ -111,8 +111,8 @@ const selectEventsWithEventIDsSQL = "" +
const selectSharedUsersSQL = "" +
"SELECT state_key FROM syncapi_current_room_state WHERE room_id = ANY(" +
" SELECT room_id FROM syncapi_current_room_state WHERE state_key = $1 AND membership='join'" +
") AND state_key = ANY($2) AND membership IN ('join', 'invite');"
" SELECT DISTINCT room_id FROM syncapi_current_room_state WHERE state_key = $1 AND membership='join'" +
") AND type = 'm.room.member' AND state_key = ANY($2) AND membership IN ('join', 'invite');"
type currentRoomStateStatements struct {
upsertRoomStateStmt *sql.Stmt

View file

@ -19,6 +19,7 @@ import (
"database/sql"
"encoding/json"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/syncapi/storage/tables"
"github.com/matrix-org/dendrite/syncapi/types"
)
@ -61,10 +62,10 @@ func NewPostgresIgnoresTable(db *sql.DB) (tables.Ignores, error) {
}
func (s *ignoresStatements) SelectIgnores(
ctx context.Context, userID string,
ctx context.Context, txn *sql.Tx, userID string,
) (*types.IgnoredUsers, error) {
var ignoresData []byte
err := s.selectIgnoresStmt.QueryRowContext(ctx, userID).Scan(&ignoresData)
err := sqlutil.TxStmt(txn, s.selectIgnoresStmt).QueryRowContext(ctx, userID).Scan(&ignoresData)
if err != nil {
return nil, err
}
@ -76,12 +77,12 @@ func (s *ignoresStatements) SelectIgnores(
}
func (s *ignoresStatements) UpsertIgnores(
ctx context.Context, userID string, ignores *types.IgnoredUsers,
ctx context.Context, txn *sql.Tx, userID string, ignores *types.IgnoredUsers,
) error {
ignoresJSON, err := json.Marshal(ignores)
if err != nil {
return err
}
_, err = s.upsertIgnoresStmt.ExecContext(ctx, userID, ignoresJSON)
_, err = sqlutil.TxStmt(txn, s.upsertIgnoresStmt).ExecContext(ctx, userID, ignoresJSON)
return err
}

View file

@ -70,13 +70,13 @@ const selectUserUnreadNotificationCountsSQL = `SELECT
const selectMaxNotificationIDSQL = `SELECT CASE COUNT(*) WHEN 0 THEN 0 ELSE MAX(id) END FROM syncapi_notification_data`
func (r *notificationDataStatements) UpsertRoomUnreadCounts(ctx context.Context, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) {
err = r.upsertRoomUnreadCounts.QueryRowContext(ctx, userID, roomID, notificationCount, highlightCount).Scan(&pos)
func (r *notificationDataStatements) UpsertRoomUnreadCounts(ctx context.Context, txn *sql.Tx, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) {
err = sqlutil.TxStmt(txn, r.upsertRoomUnreadCounts).QueryRowContext(ctx, userID, roomID, notificationCount, highlightCount).Scan(&pos)
return
}
func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error) {
rows, err := r.selectUserUnreadCounts.QueryContext(ctx, userID, fromExcl, toIncl)
func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context, txn *sql.Tx, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error) {
rows, err := sqlutil.TxStmt(txn, r.selectUserUnreadCounts).QueryContext(ctx, userID, fromExcl, toIncl)
if err != nil {
return nil, err
}
@ -101,8 +101,8 @@ func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context,
return roomCounts, rows.Err()
}
func (r *notificationDataStatements) SelectMaxID(ctx context.Context) (int64, error) {
func (r *notificationDataStatements) SelectMaxID(ctx context.Context, txn *sql.Tx) (int64, error) {
var id int64
err := r.selectMaxID.QueryRowContext(ctx).Scan(&id)
err := sqlutil.TxStmt(txn, r.selectMaxID).QueryRowContext(ctx).Scan(&id)
return id, err
}

View file

@ -108,7 +108,7 @@ func (d *Database) MaxStreamPositionForAccountData(ctx context.Context) (types.S
}
func (d *Database) MaxStreamPositionForNotificationData(ctx context.Context) (types.StreamPosition, error) {
id, err := d.NotificationData.SelectMaxID(ctx)
id, err := d.NotificationData.SelectMaxID(ctx, nil)
if err != nil {
return 0, fmt.Errorf("d.NotificationData.SelectMaxID: %w", err)
}
@ -1029,15 +1029,15 @@ func (d *Database) GetRoomReceipts(ctx context.Context, roomIDs []string, stream
}
func (d *Database) UpsertRoomUnreadNotificationCounts(ctx context.Context, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) {
err = d.Writer.Do(nil, nil, func(_ *sql.Tx) error {
pos, err = d.NotificationData.UpsertRoomUnreadCounts(ctx, userID, roomID, notificationCount, highlightCount)
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
pos, err = d.NotificationData.UpsertRoomUnreadCounts(ctx, txn, userID, roomID, notificationCount, highlightCount)
return err
})
return
}
func (d *Database) GetUserUnreadNotificationCounts(ctx context.Context, userID string, from, to types.StreamPosition) (map[string]*eventutil.NotificationData, error) {
return d.NotificationData.SelectUserUnreadCounts(ctx, userID, from, to)
return d.NotificationData.SelectUserUnreadCounts(ctx, nil, userID, from, to)
}
func (d *Database) SelectContextEvent(ctx context.Context, roomID, eventID string) (int, gomatrixserverlib.HeaderedEvent, error) {
@ -1052,15 +1052,23 @@ func (d *Database) SelectContextAfterEvent(ctx context.Context, id int, roomID s
}
func (d *Database) IgnoresForUser(ctx context.Context, userID string) (*types.IgnoredUsers, error) {
return d.Ignores.SelectIgnores(ctx, userID)
return d.Ignores.SelectIgnores(ctx, nil, userID)
}
func (d *Database) UpdateIgnoresForUser(ctx context.Context, userID string, ignores *types.IgnoredUsers) error {
return d.Ignores.UpsertIgnores(ctx, userID, ignores)
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
return d.Ignores.UpsertIgnores(ctx, txn, userID, ignores)
})
}
func (d *Database) UpdatePresence(ctx context.Context, userID string, presence types.Presence, statusMsg *string, lastActiveTS gomatrixserverlib.Timestamp, fromSync bool) (types.StreamPosition, error) {
return d.Presence.UpsertPresence(ctx, nil, userID, statusMsg, presence, lastActiveTS, fromSync)
var pos types.StreamPosition
var err error
_ = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
pos, err = d.Presence.UpsertPresence(ctx, txn, userID, statusMsg, presence, lastActiveTS, fromSync)
return nil
})
return pos, err
}
func (d *Database) GetPresence(ctx context.Context, userID string) (*types.PresenceInternal, error) {

View file

@ -95,8 +95,8 @@ const selectEventsWithEventIDsSQL = "" +
const selectSharedUsersSQL = "" +
"SELECT state_key FROM syncapi_current_room_state WHERE room_id IN(" +
" SELECT room_id FROM syncapi_current_room_state WHERE state_key = $1 AND membership='join'" +
") AND state_key IN ($2) AND membership IN ('join', 'invite');"
" SELECT DISTINCT room_id FROM syncapi_current_room_state WHERE state_key = $1 AND membership='join'" +
") AND type = 'm.room.member' AND state_key IN ($2) AND membership IN ('join', 'invite');"
type currentRoomStateStatements struct {
db *sql.DB

View file

@ -19,6 +19,7 @@ import (
"database/sql"
"encoding/json"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/syncapi/storage/tables"
"github.com/matrix-org/dendrite/syncapi/types"
)
@ -61,10 +62,10 @@ func NewSqliteIgnoresTable(db *sql.DB) (tables.Ignores, error) {
}
func (s *ignoresStatements) SelectIgnores(
ctx context.Context, userID string,
ctx context.Context, txn *sql.Tx, userID string,
) (*types.IgnoredUsers, error) {
var ignoresData []byte
err := s.selectIgnoresStmt.QueryRowContext(ctx, userID).Scan(&ignoresData)
err := sqlutil.TxStmt(txn, s.selectIgnoresStmt).QueryRowContext(ctx, userID).Scan(&ignoresData)
if err != nil {
return nil, err
}
@ -76,12 +77,12 @@ func (s *ignoresStatements) SelectIgnores(
}
func (s *ignoresStatements) UpsertIgnores(
ctx context.Context, userID string, ignores *types.IgnoredUsers,
ctx context.Context, txn *sql.Tx, userID string, ignores *types.IgnoredUsers,
) error {
ignoresJSON, err := json.Marshal(ignores)
if err != nil {
return err
}
_, err = s.upsertIgnoresStmt.ExecContext(ctx, userID, ignoresJSON)
_, err = sqlutil.TxStmt(txn, s.upsertIgnoresStmt).ExecContext(ctx, userID, ignoresJSON)
return err
}

View file

@ -72,7 +72,7 @@ const selectUserUnreadNotificationCountsSQL = `SELECT
const selectMaxNotificationIDSQL = `SELECT CASE COUNT(*) WHEN 0 THEN 0 ELSE MAX(id) END FROM syncapi_notification_data`
func (r *notificationDataStatements) UpsertRoomUnreadCounts(ctx context.Context, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) {
func (r *notificationDataStatements) UpsertRoomUnreadCounts(ctx context.Context, txn *sql.Tx, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) {
pos, err = r.streamIDStatements.nextNotificationID(ctx, nil)
if err != nil {
return
@ -81,8 +81,8 @@ func (r *notificationDataStatements) UpsertRoomUnreadCounts(ctx context.Context,
return
}
func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error) {
rows, err := r.selectUserUnreadCounts.QueryContext(ctx, userID, fromExcl, toIncl)
func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context, txn *sql.Tx, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error) {
rows, err := sqlutil.TxStmt(txn, r.selectUserUnreadCounts).QueryContext(ctx, userID, fromExcl, toIncl)
if err != nil {
return nil, err
}
@ -107,8 +107,8 @@ func (r *notificationDataStatements) SelectUserUnreadCounts(ctx context.Context,
return roomCounts, rows.Err()
}
func (r *notificationDataStatements) SelectMaxID(ctx context.Context) (int64, error) {
func (r *notificationDataStatements) SelectMaxID(ctx context.Context, txn *sql.Tx) (int64, error) {
var id int64
err := r.selectMaxID.QueryRowContext(ctx).Scan(&id)
err := sqlutil.TxStmt(txn, r.selectMaxID).QueryRowContext(ctx).Scan(&id)
return id, err
}

View file

@ -189,14 +189,14 @@ type Memberships interface {
}
type NotificationData interface {
UpsertRoomUnreadCounts(ctx context.Context, userID, roomID string, notificationCount, highlightCount int) (types.StreamPosition, error)
SelectUserUnreadCounts(ctx context.Context, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error)
SelectMaxID(ctx context.Context) (int64, error)
UpsertRoomUnreadCounts(ctx context.Context, txn *sql.Tx, userID, roomID string, notificationCount, highlightCount int) (types.StreamPosition, error)
SelectUserUnreadCounts(ctx context.Context, txn *sql.Tx, userID string, fromExcl, toIncl types.StreamPosition) (map[string]*eventutil.NotificationData, error)
SelectMaxID(ctx context.Context, txn *sql.Tx) (int64, error)
}
type Ignores interface {
SelectIgnores(ctx context.Context, userID string) (*types.IgnoredUsers, error)
UpsertIgnores(ctx context.Context, userID string, ignores *types.IgnoredUsers) error
SelectIgnores(ctx context.Context, txn *sql.Tx, userID string) (*types.IgnoredUsers, error)
UpsertIgnores(ctx context.Context, txn *sql.Tx, userID string, ignores *types.IgnoredUsers) error
}
type Presence interface {

View file

@ -49,3 +49,6 @@ Notifications can be viewed with GET /notifications
If remote user leaves room we no longer receive device updates
Guest users can join guest_access rooms
# This will fail in HTTP API mode, so blacklisted for now
If a device list update goes missing, the server resyncs on the next one

View file

@ -742,4 +742,4 @@ Newly joined room includes presence in incremental sync
User in private room doesn't appear in user directory
User joining then leaving public room appears and dissappears from directory
User in remote room doesn't appear in user directory after server left room
User in shared private room does appear in user directory until leave
User in shared private room does appear in user directory until leave

View file

@ -22,10 +22,11 @@ import (
"strings"
"testing"
"github.com/nats-io/nats.go"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
"github.com/nats-io/nats.go"
)
func CreateBaseDendrite(t *testing.T, dbType test.DBType) (*base.BaseDendrite, func()) {
@ -45,6 +46,10 @@ func CreateBaseDendrite(t *testing.T, dbType test.DBType) (*base.BaseDendrite, f
Generate: true,
Monolithic: true,
})
cfg.SyncAPI.Fulltext.Defaults(config.DefaultOpts{ // use in memory fts
Generate: true,
Monolithic: true,
})
cfg.Global.ServerName = "test"
// use a distinct prefix else concurrent postgres/sqlite runs will clash since NATS will use
// the file system event with InMemory=true :(
@ -100,6 +105,7 @@ func Base(cfg *config.Dendrite) (*base.BaseDendrite, nats.JetStreamContext, *nat
})
}
cfg.Global.JetStream.InMemory = true
cfg.SyncAPI.Fulltext.InMemory = true
base := base.NewBaseDendrite(cfg, "Tests")
js, jc := base.NATS.Prepare(base.ProcessContext, &cfg.Global.JetStream)
return base, js, jc

View file

@ -29,7 +29,6 @@ import (
type OutputStreamEventConsumer struct {
ctx context.Context
cfg *config.UserAPI
userAPI api.UserInternalAPI
rsAPI rsapi.UserRoomserverAPI
jetstream nats.JetStreamContext
durable string
@ -45,7 +44,6 @@ func NewOutputStreamEventConsumer(
js nats.JetStreamContext,
store storage.Database,
pgClient pushgateway.Client,
userAPI api.UserInternalAPI,
rsAPI rsapi.UserRoomserverAPI,
syncProducer *producers.SyncAPI,
) *OutputStreamEventConsumer {
@ -57,7 +55,6 @@ func NewOutputStreamEventConsumer(
durable: cfg.Matrix.JetStream.Durable("UserAPISyncAPIStreamEventConsumer"),
topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputStreamEvent),
pgClient: pgClient,
userAPI: userAPI,
rsAPI: rsAPI,
syncProducer: syncProducer,
}
@ -305,7 +302,7 @@ func (s *OutputStreamEventConsumer) notifyLocal(ctx context.Context, event *goma
"event_id": event.EventID(),
"room_id": event.RoomID(),
"localpart": mem.Localpart,
}).Tracef("Push rule evaluation rejected the event")
}).Debugf("Push rule evaluation rejected the event")
return nil
}
@ -348,7 +345,7 @@ func (s *OutputStreamEventConsumer) notifyLocal(ctx context.Context, event *goma
"localpart": mem.Localpart,
"num_urls": len(devicesByURLAndFormat),
"num_unread": userNumUnreadNotifs,
}).Tracef("Notifying single member")
}).Debugf("Notifying single member")
// Push gateways are out of our control, and we cannot risk
// looking up the server on a misbehaving push gateway. Each user
@ -422,8 +419,8 @@ func (s *OutputStreamEventConsumer) evaluatePushRules(ctx context.Context, event
return nil, fmt.Errorf("user %s is ignored", sender)
}
}
var res api.QueryPushRulesResponse
if err = s.userAPI.QueryPushRules(ctx, &api.QueryPushRulesRequest{UserID: mem.UserID}, &res); err != nil {
ruleSets, err := s.db.QueryPushRules(ctx, mem.Localpart)
if err != nil {
return nil, err
}
@ -434,7 +431,7 @@ func (s *OutputStreamEventConsumer) evaluatePushRules(ctx context.Context, event
roomID: event.RoomID(),
roomSize: roomSize,
}
eval := pushrules.NewRuleSetEvaluator(ec, &res.RuleSets.Global)
eval := pushrules.NewRuleSetEvaluator(ec, &ruleSets.Global)
rule, err := eval.MatchEvent(event.Event)
if err != nil {
return nil, err

View file

@ -0,0 +1,129 @@
package consumers
import (
"context"
"testing"
"github.com/matrix-org/gomatrixserverlib"
"github.com/stretchr/testify/assert"
"github.com/matrix-org/dendrite/internal/pushrules"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/userapi/storage"
)
func mustCreateDatabase(t *testing.T, dbType test.DBType) (storage.Database, func()) {
t.Helper()
connStr, close := test.PrepareDBConnectionString(t, dbType)
db, err := storage.NewUserAPIDatabase(nil, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, "", 4, 0, 0, "")
if err != nil {
t.Fatalf("failed to create new user db: %v", err)
}
return db, close
}
func mustCreateEvent(t *testing.T, content string) *gomatrixserverlib.HeaderedEvent {
t.Helper()
ev, err := gomatrixserverlib.NewEventFromTrustedJSON([]byte(content), false, gomatrixserverlib.RoomVersionV10)
if err != nil {
t.Fatalf("failed to create event: %v", err)
}
return ev.Headered(gomatrixserverlib.RoomVersionV10)
}
func Test_evaluatePushRules(t *testing.T) {
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
db, close := mustCreateDatabase(t, dbType)
defer close()
consumer := OutputStreamEventConsumer{db: db}
testCases := []struct {
name string
eventContent string
wantAction pushrules.ActionKind
wantActions []*pushrules.Action
wantNotify bool
}{
{
name: "m.receipt doesn't notify",
eventContent: `{"type":"m.receipt"}`,
wantAction: pushrules.UnknownAction,
wantActions: nil,
},
{
name: "m.reaction doesn't notify",
eventContent: `{"type":"m.reaction"}`,
wantAction: pushrules.DontNotifyAction,
wantActions: []*pushrules.Action{
{
Kind: pushrules.DontNotifyAction,
},
},
},
{
name: "m.room.message notifies",
eventContent: `{"type":"m.room.message"}`,
wantNotify: true,
wantAction: pushrules.NotifyAction,
wantActions: []*pushrules.Action{
{Kind: pushrules.NotifyAction},
{
Kind: pushrules.SetTweakAction,
Tweak: pushrules.HighlightTweak,
Value: false,
},
},
},
{
name: "m.room.message highlights",
eventContent: `{"type":"m.room.message", "content": {"body": "test"} }`,
wantNotify: true,
wantAction: pushrules.NotifyAction,
wantActions: []*pushrules.Action{
{Kind: pushrules.NotifyAction},
{
Kind: pushrules.SetTweakAction,
Tweak: pushrules.SoundTweak,
Value: "default",
},
{
Kind: pushrules.SetTweakAction,
Tweak: pushrules.HighlightTweak,
Value: true,
},
},
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
actions, err := consumer.evaluatePushRules(ctx, mustCreateEvent(t, tc.eventContent), &localMembership{
UserID: "@test:localhost",
Localpart: "test",
Domain: "localhost",
}, 10)
if err != nil {
t.Fatalf("failed to evaluate push rules: %v", err)
}
assert.Equal(t, tc.wantActions, actions)
gotAction, _, err := pushrules.ActionsToTweaks(actions)
if err != nil {
t.Fatalf("failed to get actions: %v", err)
}
if gotAction != tc.wantAction {
t.Fatalf("expected action to be '%s', got '%s'", tc.wantAction, gotAction)
}
// this is taken from `notifyLocal`
if tc.wantNotify && gotAction != pushrules.NotifyAction && gotAction != pushrules.CoalesceAction {
t.Fatalf("expected to notify but didn't")
}
})
}
})
}

View file

@ -30,7 +30,6 @@ import (
"github.com/matrix-org/dendrite/clientapi/userutil"
"github.com/matrix-org/dendrite/internal/eventutil"
"github.com/matrix-org/dendrite/internal/pushrules"
"github.com/matrix-org/dendrite/internal/sqlutil"
keyapi "github.com/matrix-org/dendrite/keyserver/api"
rsapi "github.com/matrix-org/dendrite/roomserver/api"
@ -760,57 +759,15 @@ func (a *UserInternalAPI) PerformPushRulesPut(
}
func (a *UserInternalAPI) QueryPushRules(ctx context.Context, req *api.QueryPushRulesRequest, res *api.QueryPushRulesResponse) error {
userReq := api.QueryAccountDataRequest{
UserID: req.UserID,
DataType: pushRulesAccountDataType,
localpart, _, err := gomatrixserverlib.SplitID('@', req.UserID)
if err != nil {
return fmt.Errorf("failed to split user ID %q for push rules", req.UserID)
}
var userRes api.QueryAccountDataResponse
if err := a.QueryAccountData(ctx, &userReq, &userRes); err != nil {
return err
pushRules, err := a.DB.QueryPushRules(ctx, localpart)
if err != nil {
return fmt.Errorf("failed to query push rules: %w", err)
}
bs, ok := userRes.GlobalAccountData[pushRulesAccountDataType]
if ok {
// Legacy Dendrite users will have completely empty push rules, so we should
// detect that situation and set some defaults.
var rules struct {
G struct {
Content []json.RawMessage `json:"content"`
Override []json.RawMessage `json:"override"`
Room []json.RawMessage `json:"room"`
Sender []json.RawMessage `json:"sender"`
Underride []json.RawMessage `json:"underride"`
} `json:"global"`
}
if err := json.Unmarshal([]byte(bs), &rules); err == nil {
count := len(rules.G.Content) + len(rules.G.Override) +
len(rules.G.Room) + len(rules.G.Sender) + len(rules.G.Underride)
ok = count > 0
}
}
if !ok {
// If we didn't find any default push rules then we should just generate some
// fresh ones.
localpart, _, err := gomatrixserverlib.SplitID('@', req.UserID)
if err != nil {
return fmt.Errorf("failed to split user ID %q for push rules", req.UserID)
}
pushRuleSets := pushrules.DefaultAccountRuleSets(localpart, a.ServerName)
prbs, err := json.Marshal(pushRuleSets)
if err != nil {
return fmt.Errorf("failed to marshal default push rules: %w", err)
}
if err := a.DB.SaveAccountData(ctx, localpart, "", pushRulesAccountDataType, json.RawMessage(prbs)); err != nil {
return fmt.Errorf("failed to save default push rules: %w", err)
}
res.RuleSets = pushRuleSets
return nil
}
var data pushrules.AccountRuleSets
if err := json.Unmarshal([]byte(bs), &data); err != nil {
util.GetLogger(ctx).WithError(err).Error("json.Unmarshal of push rules failed")
return err
}
res.RuleSets = &data
res.RuleSets = pushRules
return nil
}

View file

@ -20,6 +20,7 @@ import (
"errors"
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
"github.com/matrix-org/dendrite/internal/pushrules"
"github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/dendrite/userapi/storage/tables"
"github.com/matrix-org/dendrite/userapi/types"
@ -53,6 +54,7 @@ type AccountData interface {
// If no account data could be found, returns nil
// Returns an error if there was an issue with the retrieval
GetAccountDataByType(ctx context.Context, localpart, roomID, dataType string) (data json.RawMessage, err error)
QueryPushRules(ctx context.Context, localpart string) (*pushrules.AccountRuleSets, error)
}
type Device interface {

View file

@ -26,10 +26,11 @@ import (
"strings"
"time"
"github.com/matrix-org/dendrite/userapi/types"
"github.com/matrix-org/gomatrixserverlib"
"golang.org/x/crypto/bcrypt"
"github.com/matrix-org/dendrite/userapi/types"
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
"github.com/matrix-org/dendrite/internal/pushrules"
"github.com/matrix-org/dendrite/internal/sqlutil"
@ -177,6 +178,41 @@ func (d *Database) createAccount(
return account, nil
}
func (d *Database) QueryPushRules(
ctx context.Context,
localpart string,
) (*pushrules.AccountRuleSets, error) {
data, err := d.AccountDatas.SelectAccountDataByType(ctx, localpart, "", "m.push_rules")
if err != nil {
return nil, err
}
// If we didn't find any default push rules then we should just generate some
// fresh ones.
if len(data) == 0 {
pushRuleSets := pushrules.DefaultAccountRuleSets(localpart, d.ServerName)
prbs, err := json.Marshal(pushRuleSets)
if err != nil {
return nil, fmt.Errorf("failed to marshal default push rules: %w", err)
}
err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
if dbErr := d.AccountDatas.InsertAccountData(ctx, txn, localpart, "", "m.push_rules", prbs); dbErr != nil {
return fmt.Errorf("failed to save default push rules: %w", dbErr)
}
return nil
})
return pushRuleSets, err
}
var pushRules pushrules.AccountRuleSets
if err := json.Unmarshal(data, &pushRules); err != nil {
return nil, err
}
return &pushRules, nil
}
// SaveAccountData saves new account data for a given user and a given room.
// If the account data is not specific to a room, the room ID should be an empty string
// If an account data already exists for a given set (user, room, data type), it will
@ -699,7 +735,9 @@ func (d *Database) GetRoomNotificationCounts(ctx context.Context, localpart, roo
}
func (d *Database) DeleteOldNotifications(ctx context.Context) error {
return d.Notifications.Clean(ctx, nil)
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
return d.Notifications.Clean(ctx, txn)
})
}
func (d *Database) UpsertPusher(

View file

@ -18,6 +18,8 @@ import (
"time"
"github.com/gorilla/mux"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/internal/pushgateway"
keyapi "github.com/matrix-org/dendrite/keyserver/api"
rsapi "github.com/matrix-org/dendrite/roomserver/api"
@ -31,7 +33,6 @@ import (
"github.com/matrix-org/dendrite/userapi/producers"
"github.com/matrix-org/dendrite/userapi/storage"
"github.com/matrix-org/dendrite/userapi/util"
"github.com/sirupsen/logrus"
)
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
@ -90,7 +91,7 @@ func NewInternalAPI(
}
eventConsumer := consumers.NewOutputStreamEventConsumer(
base.ProcessContext, cfg, js, db, pgClient, userAPI, rsAPI, syncProducer,
base.ProcessContext, cfg, js, db, pgClient, rsAPI, syncProducer,
)
if err := eventConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start user API streamed event consumer")