aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--syncapi/storage/interface.go2
-rw-r--r--syncapi/storage/postgres/presence_table.go9
-rw-r--r--syncapi/storage/shared/syncserver.go4
-rw-r--r--syncapi/storage/sqlite3/presence_table.go10
-rw-r--r--syncapi/storage/tables/interface.go2
-rw-r--r--syncapi/streams/stream_presence.go12
-rw-r--r--syncapi/sync/requestpool_test.go2
7 files changed, 27 insertions, 14 deletions
diff --git a/syncapi/storage/interface.go b/syncapi/storage/interface.go
index 43aaa358..48697859 100644
--- a/syncapi/storage/interface.go
+++ b/syncapi/storage/interface.go
@@ -159,6 +159,6 @@ type Database interface {
type Presence interface {
UpdatePresence(ctx context.Context, userID string, presence types.Presence, statusMsg *string, lastActiveTS gomatrixserverlib.Timestamp, fromSync bool) (types.StreamPosition, error)
GetPresence(ctx context.Context, userID string) (*types.PresenceInternal, error)
- PresenceAfter(ctx context.Context, after types.StreamPosition) (map[string]*types.PresenceInternal, error)
+ PresenceAfter(ctx context.Context, after types.StreamPosition, filter gomatrixserverlib.EventFilter) (map[string]*types.PresenceInternal, error)
MaxStreamPositionForPresence(ctx context.Context) (types.StreamPosition, error)
}
diff --git a/syncapi/storage/postgres/presence_table.go b/syncapi/storage/postgres/presence_table.go
index 9f1e37f7..7194afea 100644
--- a/syncapi/storage/postgres/presence_table.go
+++ b/syncapi/storage/postgres/presence_table.go
@@ -17,6 +17,7 @@ package postgres
import (
"context"
"database/sql"
+ "time"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
@@ -72,7 +73,8 @@ const selectMaxPresenceSQL = "" +
const selectPresenceAfter = "" +
" SELECT id, user_id, presence, status_msg, last_active_ts" +
" FROM syncapi_presence" +
- " WHERE id > $1"
+ " WHERE id > $1 AND last_active_ts >= $2" +
+ " ORDER BY id ASC LIMIT $3"
type presenceStatements struct {
upsertPresenceStmt *sql.Stmt
@@ -144,11 +146,12 @@ func (p *presenceStatements) GetMaxPresenceID(ctx context.Context, txn *sql.Tx)
func (p *presenceStatements) GetPresenceAfter(
ctx context.Context, txn *sql.Tx,
after types.StreamPosition,
+ filter gomatrixserverlib.EventFilter,
) (presences map[string]*types.PresenceInternal, err error) {
presences = make(map[string]*types.PresenceInternal)
stmt := sqlutil.TxStmt(txn, p.selectPresenceAfterStmt)
-
- rows, err := stmt.QueryContext(ctx, after)
+ afterTS := gomatrixserverlib.AsTimestamp(time.Now().Add(time.Minute * -5))
+ rows, err := stmt.QueryContext(ctx, after, afterTS, filter.Limit)
if err != nil {
return nil, err
}
diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go
index 25aca50a..b7d2d3a2 100644
--- a/syncapi/storage/shared/syncserver.go
+++ b/syncapi/storage/shared/syncserver.go
@@ -1056,8 +1056,8 @@ func (s *Database) GetPresence(ctx context.Context, userID string) (*types.Prese
return s.Presence.GetPresenceForUser(ctx, nil, userID)
}
-func (s *Database) PresenceAfter(ctx context.Context, after types.StreamPosition) (map[string]*types.PresenceInternal, error) {
- return s.Presence.GetPresenceAfter(ctx, nil, after)
+func (s *Database) PresenceAfter(ctx context.Context, after types.StreamPosition, filter gomatrixserverlib.EventFilter) (map[string]*types.PresenceInternal, error) {
+ return s.Presence.GetPresenceAfter(ctx, nil, after, filter)
}
func (s *Database) MaxStreamPositionForPresence(ctx context.Context) (types.StreamPosition, error) {
diff --git a/syncapi/storage/sqlite3/presence_table.go b/syncapi/storage/sqlite3/presence_table.go
index 177a01bf..b61a825d 100644
--- a/syncapi/storage/sqlite3/presence_table.go
+++ b/syncapi/storage/sqlite3/presence_table.go
@@ -17,6 +17,7 @@ package sqlite3
import (
"context"
"database/sql"
+ "time"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
@@ -71,7 +72,8 @@ const selectMaxPresenceSQL = "" +
const selectPresenceAfter = "" +
" SELECT id, user_id, presence, status_msg, last_active_ts" +
" FROM syncapi_presence" +
- " WHERE id > $1"
+ " WHERE id > $1 AND last_active_ts >= $2" +
+ " ORDER BY id ASC LIMIT $3"
type presenceStatements struct {
db *sql.DB
@@ -158,12 +160,12 @@ func (p *presenceStatements) GetMaxPresenceID(ctx context.Context, txn *sql.Tx)
// GetPresenceAfter returns the changes presences after a given stream id
func (p *presenceStatements) GetPresenceAfter(
ctx context.Context, txn *sql.Tx,
- after types.StreamPosition,
+ after types.StreamPosition, filter gomatrixserverlib.EventFilter,
) (presences map[string]*types.PresenceInternal, err error) {
presences = make(map[string]*types.PresenceInternal)
stmt := sqlutil.TxStmt(txn, p.selectPresenceAfterStmt)
-
- rows, err := stmt.QueryContext(ctx, after)
+ afterTS := gomatrixserverlib.AsTimestamp(time.Now().Add(time.Minute * -5))
+ rows, err := stmt.QueryContext(ctx, after, afterTS, filter.Limit)
if err != nil {
return nil, err
}
diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go
index 4ff4689e..6fdd9483 100644
--- a/syncapi/storage/tables/interface.go
+++ b/syncapi/storage/tables/interface.go
@@ -188,5 +188,5 @@ type Presence interface {
UpsertPresence(ctx context.Context, txn *sql.Tx, userID string, statusMsg *string, presence types.Presence, lastActiveTS gomatrixserverlib.Timestamp, fromSync bool) (pos types.StreamPosition, err error)
GetPresenceForUser(ctx context.Context, txn *sql.Tx, userID string) (presence *types.PresenceInternal, err error)
GetMaxPresenceID(ctx context.Context, txn *sql.Tx) (pos types.StreamPosition, err error)
- GetPresenceAfter(ctx context.Context, txn *sql.Tx, after types.StreamPosition) (presences map[string]*types.PresenceInternal, err error)
+ GetPresenceAfter(ctx context.Context, txn *sql.Tx, after types.StreamPosition, filter gomatrixserverlib.EventFilter) (presences map[string]*types.PresenceInternal, err error)
}
diff --git a/syncapi/streams/stream_presence.go b/syncapi/streams/stream_presence.go
index 614b88d4..675a7a17 100644
--- a/syncapi/streams/stream_presence.go
+++ b/syncapi/streams/stream_presence.go
@@ -53,7 +53,8 @@ func (p *PresenceStreamProvider) IncrementalSync(
req *types.SyncRequest,
from, to types.StreamPosition,
) types.StreamPosition {
- presences, err := p.DB.PresenceAfter(ctx, from)
+ // We pull out a larger number than the filter asks for, since we're filtering out events later
+ presences, err := p.DB.PresenceAfter(ctx, from, gomatrixserverlib.EventFilter{Limit: 1000})
if err != nil {
req.Log.WithError(err).Error("p.DB.PresenceAfter failed")
return from
@@ -72,6 +73,7 @@ func (p *PresenceStreamProvider) IncrementalSync(
req.Log.WithError(err).Error("unable to refresh notifier lists")
return from
}
+ NewlyJoinedLoop:
for _, roomID := range newlyJoined {
roomUsers := p.notifier.JoinedUsers(roomID)
for i := range roomUsers {
@@ -86,11 +88,14 @@ func (p *PresenceStreamProvider) IncrementalSync(
req.Log.WithError(err).Error("unable to query presence for user")
return from
}
+ if len(presences) > req.Filter.Presence.Limit {
+ break NewlyJoinedLoop
+ }
}
}
}
- lastPos := to
+ lastPos := from
for _, presence := range presences {
if presence == nil {
continue
@@ -135,6 +140,9 @@ func (p *PresenceStreamProvider) IncrementalSync(
if presence.StreamPos > lastPos {
lastPos = presence.StreamPos
}
+ if len(req.Response.Presence.Events) == req.Filter.Presence.Limit {
+ break
+ }
p.cache.Store(cacheKey, presence)
}
diff --git a/syncapi/sync/requestpool_test.go b/syncapi/sync/requestpool_test.go
index a8008994..5e52bc7c 100644
--- a/syncapi/sync/requestpool_test.go
+++ b/syncapi/sync/requestpool_test.go
@@ -30,7 +30,7 @@ func (d dummyDB) GetPresence(ctx context.Context, userID string) (*types.Presenc
return &types.PresenceInternal{}, nil
}
-func (d dummyDB) PresenceAfter(ctx context.Context, after types.StreamPosition) (map[string]*types.PresenceInternal, error) {
+func (d dummyDB) PresenceAfter(ctx context.Context, after types.StreamPosition, filter gomatrixserverlib.EventFilter) (map[string]*types.PresenceInternal, error) {
return map[string]*types.PresenceInternal{}, nil
}
generated by cgit v1.2.3 (git 2.26.2) at 2024-11-28 23:07:17 +0000