mirror of
https://github.com/matrix-org/dendrite.git
synced 2026-01-01 03:03:10 -06:00
Use filter and limit presence count
This commit is contained in:
parent
34221938cc
commit
9f7ab9c608
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -72,7 +72,7 @@ const selectMaxPresenceSQL = "" +
|
|||
const selectPresenceAfter = "" +
|
||||
" SELECT id, user_id, presence, status_msg, last_active_ts" +
|
||||
" FROM syncapi_presence" +
|
||||
" WHERE id > $1"
|
||||
" WHERE id > $1 ORDER BY last_active_ts DESC LIMIT $2"
|
||||
|
||||
type presenceStatements struct {
|
||||
upsertPresenceStmt *sql.Stmt
|
||||
|
|
@ -144,11 +144,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)
|
||||
rows, err := stmt.QueryContext(ctx, after, filter.Limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -71,7 +71,7 @@ const selectMaxPresenceSQL = "" +
|
|||
const selectPresenceAfter = "" +
|
||||
" SELECT id, user_id, presence, status_msg, last_active_ts" +
|
||||
" FROM syncapi_presence" +
|
||||
" WHERE id > $1"
|
||||
" WHERE id > $1 ORDER BY last_active_ts DESC LIMIT $2"
|
||||
|
||||
type presenceStatements struct {
|
||||
db *sql.DB
|
||||
|
|
@ -158,12 +158,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)
|
||||
rows, err := stmt.QueryContext(ctx, after, filter.Limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ func (p *PresenceStreamProvider) IncrementalSync(
|
|||
req *types.SyncRequest,
|
||||
from, to types.StreamPosition,
|
||||
) types.StreamPosition {
|
||||
presences, err := p.DB.PresenceAfter(ctx, from)
|
||||
presences, err := p.DB.PresenceAfter(ctx, from, req.Filter.Presence)
|
||||
if err != nil {
|
||||
req.Log.WithError(err).Error("p.DB.PresenceAfter failed")
|
||||
return from
|
||||
|
|
|
|||
Loading…
Reference in a new issue