// Returns a map following the format data[roomID] = []dataTypes

// If no data is retrieved, returns an empty map

// If there was an issue with the retrieval, returns an error

// UpsertAccountData keeps track of new or updated account data, by saving the type

// of the new/updated data, and the user ID and room ID the data is related to (empty)

// room ID means the data isn't specific to any room)

func (s *accountDataStatements) SelectAccountDataInRange(

ctx context.Context,

userID string,

accountDataEventFilter *gomatrixserverlib.EventFilter,

) (data map[string][]string, err error) {

data = make(map[string][]string)

pq.StringArray(filterConvertTypeWildcardToSQL(accountDataEventFilter.Types)),

pq.StringArray(filterConvertTypeWildcardToSQL(accountDataEventFilter.NotTypes)),

accountDataEventFilter.Limit,

// selectInviteEventsInRange returns a map of room ID to invite event for the

// active invites for the target user ID in the supplied range.

func (s *inviteEventsStatements) SelectInviteEventsInRange(

) (map[string]gomatrixserverlib.HeaderedEvent, error) {

stmt := common.TxStmt(txn, s.selectInviteEventsInRangeStmt)

if err != nil {

return nil, err

}

// Results are bucketed based on the room ID. If the same state is overwritten multiple times between the

// two positions, only the most recent state is returned.

func (s *outputRoomEventsStatements) SelectStateInRange(

stateFilter *gomatrixserverlib.StateFilter,

) (map[string]map[string]bool, map[string]types.StreamEvent, error) {

stmt := common.TxStmt(txn, s.selectStateInRangeStmt)

rows, err := stmt.QueryContext(

pq.StringArray(stateFilter.Senders),

pq.StringArray(stateFilter.NotSenders),

pq.StringArray(filterConvertTypeWildcardToSQL(stateFilter.Types)),

// since it'll just mark the event as not being needed.

if len(addIDs) < len(delIDs) {

log.WithFields(log.Fields{

"adds": addIDs,

"dels": delIDs,

}).Warn("StateBetween: ignoring deleted state")

// from sync.

func (s *outputRoomEventsStatements) SelectRecentEvents(

ctx context.Context, txn *sql.Tx,

chronologicalOrder bool, onlySyncEvents bool,

) ([]types.StreamEvent, error) {

var stmt *sql.Stmt

} else {

stmt = common.TxStmt(txn, s.selectRecentEventsStmt)

}

if err != nil {

return nil, err

}

// from a given position, up to a maximum of 'limit'.

func (s *outputRoomEventsStatements) SelectEarlyEvents(

ctx context.Context, txn *sql.Tx,

) ([]types.StreamEvent, error) {

stmt := common.TxStmt(txn, s.selectEarlyEventsStmt)

if err != nil {

return nil, err

}

roomID string, limit int,

backwardOrdering bool,

) (events []types.StreamEvent, err error) {

if backwardOrdering {

// When using backward ordering, we want the most recent events first.

if events, err = d.OutputEvents.SelectRecentEvents(

); err != nil {

return

}

} else {

// When using forward ordering, we want the least recent events first.

if events, err = d.OutputEvents.SelectEarlyEvents(

); err != nil {

return

}

// If no data is retrieved, returns an empty map

// If there was an issue with the retrieval, returns an error

func (d *Database) GetAccountDataInRange(

accountDataFilterPart *gomatrixserverlib.EventFilter,

) (map[string][]string, error) {

}

// UpsertAccountData keeps track of new or updated account data, by saving the type

func (d *Database) addPDUDeltaToResponse(

ctx context.Context,

device authtypes.Device,

numRecentEventsPerRoom int,

wantFullState bool,

res *types.Response,

var deltas []stateDelta

if !wantFullState {

deltas, joinedRoomIDs, err = d.getStateDeltas(

)

} else {

deltas, joinedRoomIDs, err = d.getStateDeltasForFullStateSync(

)

}

if err != nil {

}

for _, delta := range deltas {

if err != nil {

return nil, err

}

}

// TODO: This should be done in getStateDeltas

return nil, err

}

var joinedRoomIDs []string

var err error

if fromPos.PDUPosition() != toPos.PDUPosition() || wantFullState {

joinedRoomIDs, err = d.addPDUDeltaToResponse(

)

} else {

joinedRoomIDs, err = d.CurrentRoomState.SelectRoomIDsWithMembership(

if err != nil {

return

}

res = types.NewResponse(toPos)

// See: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L316

var recentStreamEvents []types.StreamEvent

recentStreamEvents, err = d.OutputEvents.SelectRecentEvents(

)

if err != nil {

return

res.Rooms.Join[roomID] = *jr

}

return

}

func (d *Database) addInvitesToResponse(

ctx context.Context, txn *sql.Tx,

userID string,

res *types.Response,

) error {

invites, err := d.Invites.SelectInviteEventsInRange(

)

if err != nil {

return err

ctx context.Context,

device *authtypes.Device,

txn *sql.Tx,

delta stateDelta,

numRecentEventsPerRoom int,

res *types.Response,

) error {

if delta.membershipPos > 0 && delta.membership == gomatrixserverlib.Leave {

// make sure we don't leak recent events after the leave event.

// TODO: History visibility makes this somewhat complex to handle correctly. For example:

// TODO: This will fail on join -> leave -> sensitive msg -> join -> leave

// in a single /sync request

// This is all "okay" assuming history_visibility == "shared" which it is by default.

}

recentStreamEvents, err := d.OutputEvents.SelectRecentEvents(

numRecentEventsPerRoom, true, true,

)

if err != nil {

// A list of joined room IDs is also returned in case the caller needs it.

func (d *Database) getStateDeltas(

ctx context.Context, device *authtypes.Device, txn *sql.Tx,

stateFilter *gomatrixserverlib.StateFilter,

) ([]stateDelta, []string, error) {

// Implement membership change algorithm: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L821

var deltas []stateDelta

// get all the state events ever between these two positions

if err != nil {

return nil, nil, err

}

// updates for other rooms.

func (d *Database) getStateDeltasForFullStateSync(

ctx context.Context, device *authtypes.Device, txn *sql.Tx,

stateFilter *gomatrixserverlib.StateFilter,

) ([]stateDelta, []string, error) {

joinedRoomIDs, err := d.CurrentRoomState.SelectRoomIDsWithMembership(ctx, txn, userID, gomatrixserverlib.Join)

}

// Get all the state events ever between these two positions

if err != nil {

return nil, nil, err

}

func (s *accountDataStatements) SelectAccountDataInRange(

ctx context.Context,

userID string,

accountDataFilterPart *gomatrixserverlib.EventFilter,

) (data map[string][]string, err error) {

data = make(map[string][]string)

if err != nil {

return

}

// selectInviteEventsInRange returns a map of room ID to invite event for the

// active invites for the target user ID in the supplied range.

func (s *inviteEventsStatements) SelectInviteEventsInRange(

) (map[string]gomatrixserverlib.HeaderedEvent, error) {

stmt := common.TxStmt(txn, s.selectInviteEventsInRangeStmt)

if err != nil {

return nil, err

}

// Results are bucketed based on the room ID. If the same state is overwritten multiple times between the

// two positions, only the most recent state is returned.

func (s *outputRoomEventsStatements) SelectStateInRange(

stateFilterPart *gomatrixserverlib.StateFilter,

) (map[string]map[string]bool, map[string]types.StreamEvent, error) {

stmt := common.TxStmt(txn, s.selectStateInRangeStmt)

rows, err := stmt.QueryContext(

/*pq.StringArray(stateFilterPart.Senders),

pq.StringArray(stateFilterPart.NotSenders),

pq.StringArray(filterConvertTypeWildcardToSQL(stateFilterPart.Types)),

// since it'll just mark the event as not being needed.

if len(addIDs) < len(delIDs) {

log.WithFields(log.Fields{

"adds": addIDsJSON,

"dels": delIDsJSON,

}).Warn("StateBetween: ignoring deleted state")

func (s *outputRoomEventsStatements) SelectRecentEvents(

ctx context.Context, txn *sql.Tx,

chronologicalOrder bool, onlySyncEvents bool,

) ([]types.StreamEvent, error) {

var stmt *sql.Stmt

stmt = common.TxStmt(txn, s.selectRecentEventsStmt)

}

if err != nil {

return nil, err

}

func (s *outputRoomEventsStatements) SelectEarlyEvents(

ctx context.Context, txn *sql.Tx,

) ([]types.StreamEvent, error) {

stmt := common.TxStmt(txn, s.selectEarlyEventsStmt)

if err != nil {

return nil, err

}

type AccountData interface {

InsertAccountData(ctx context.Context, txn *sql.Tx, userID, roomID, dataType string) (pos types.StreamPosition, err error)

SelectMaxAccountDataID(ctx context.Context, txn *sql.Tx) (id int64, err error)

}

type Invites interface {

InsertInviteEvent(ctx context.Context, txn *sql.Tx, inviteEvent gomatrixserverlib.HeaderedEvent) (streamPos types.StreamPosition, err error)

DeleteInviteEvent(ctx context.Context, inviteEventID string) error

SelectMaxInviteID(ctx context.Context, txn *sql.Tx) (id int64, err error)

}

type Events interface {

SelectMaxEventID(ctx context.Context, txn *sql.Tx) (id int64, err error)

InsertEvent(ctx context.Context, txn *sql.Tx, event *gomatrixserverlib.HeaderedEvent, addState, removeState []string, transactionID *api.TransactionID, excludeFromSync bool) (streamPos types.StreamPosition, err error)

// If onlySyncEvents has a value of true, only returns the events that aren't marked as to exclude from sync.

// Returns up to `limit` events.

SelectEvents(ctx context.Context, txn *sql.Tx, eventIDs []string) ([]types.StreamEvent, error)

}

return data, nil

}

// Sync is not initial, get all account data since the latest sync

dataTypes, err := rp.db.GetAccountDataInRange(

)

if err != nil {

return nil, err

ExcludeFromSync bool

}

// SyncTokenType represents the type of a sync token.

// It can be either "s" (representing a position in the whole stream of events)

// or "t" (representing a position in a room's topology/depth).