mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-20 05:13:11 -06:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into neilalexander/keydb

Browse source
This commit is contained in:
Neil Alexander 2020-05-26 17:53:16 +01:00 committed by GitHub
parent c9f1399ea6 803af87dc4
commit e182cb524c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 388 additions and 357 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
roomserver/storage/postgres/event_json_table.go
View file

@ -21,6 +21,7 @@ import (
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
)
@ -58,32 +59,28 @@ type eventJSONStatements struct {
bulkSelectEventJSONStmt *sql.Stmt
}
func (s *eventJSONStatements) prepare(db *sql.DB) (err error) {
_, err = db.Exec(eventJSONSchema)
func NewPostgresEventJSONTable(db *sql.DB) (tables.EventJSON, error) {
s := &eventJSONStatements{}
_, err := db.Exec(eventJSONSchema)
if err != nil {
return
return nil, err
}
return statementList{
return s, statementList{
{&s.insertEventJSONStmt, insertEventJSONSQL},
{&s.bulkSelectEventJSONStmt, bulkSelectEventJSONSQL},
}.prepare(db)
}
func (s *eventJSONStatements) insertEventJSON(
ctx context.Context, eventNID types.EventNID, eventJSON []byte,
func (s *eventJSONStatements) InsertEventJSON(
ctx context.Context, txn *sql.Tx, eventNID types.EventNID, eventJSON []byte,
) error {
_, err := s.insertEventJSONStmt.ExecContext(ctx, int64(eventNID), eventJSON)
return err
}
type eventJSONPair struct {
EventNID types.EventNID
EventJSON []byte
}
func (s *eventJSONStatements) bulkSelectEventJSON(
func (s *eventJSONStatements) BulkSelectEventJSON(
ctx context.Context, eventNIDs []types.EventNID,
) ([]eventJSONPair, error) {
) ([]tables.EventJSONPair, error) {
rows, err := s.bulkSelectEventJSONStmt.QueryContext(ctx, eventNIDsAsArray(eventNIDs))
if err != nil {
return nil, err
@ -94,7 +91,7 @@ func (s *eventJSONStatements) bulkSelectEventJSON(
// because of the unique constraint on event NIDs.
// So we can allocate an array of the correct size now.
// We might get fewer results than NIDs so we adjust the length of the slice before returning it.
results := make([]eventJSONPair, len(eventNIDs))
results := make([]tables.EventJSONPair, len(eventNIDs))
i := 0
for ; rows.Next(); i++ {
result := &results[i]

18
roomserver/storage/postgres/event_state_keys_table.go
View file

@ -21,6 +21,7 @@ import (
"github.com/lib/pq"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
)
@ -74,12 +75,13 @@ type eventStateKeyStatements struct {
bulkSelectEventStateKeyStmt *sql.Stmt
}
func (s *eventStateKeyStatements) prepare(db *sql.DB) (err error) {
_, err = db.Exec(eventStateKeysSchema)
func NewPostgresEventStateKeysTable(db *sql.DB) (tables.EventStateKeys, error) {
s := &eventStateKeyStatements{}
_, err := db.Exec(eventStateKeysSchema)
if err != nil {
return
return nil, err
}
return statementList{
return s, statementList{
{&s.insertEventStateKeyNIDStmt, insertEventStateKeyNIDSQL},
{&s.selectEventStateKeyNIDStmt, selectEventStateKeyNIDSQL},
{&s.bulkSelectEventStateKeyNIDStmt, bulkSelectEventStateKeyNIDSQL},
@ -87,7 +89,7 @@ func (s *eventStateKeyStatements) prepare(db *sql.DB) (err error) {
}.prepare(db)
}
func (s *eventStateKeyStatements) insertEventStateKeyNID(
func (s *eventStateKeyStatements) InsertEventStateKeyNID(
ctx context.Context, txn *sql.Tx, eventStateKey string,
) (types.EventStateKeyNID, error) {
var eventStateKeyNID int64
@ -96,7 +98,7 @@ func (s *eventStateKeyStatements) insertEventStateKeyNID(
return types.EventStateKeyNID(eventStateKeyNID), err
}
func (s *eventStateKeyStatements) selectEventStateKeyNID(
func (s *eventStateKeyStatements) SelectEventStateKeyNID(
ctx context.Context, txn *sql.Tx, eventStateKey string,
) (types.EventStateKeyNID, error) {
var eventStateKeyNID int64
@ -105,7 +107,7 @@ func (s *eventStateKeyStatements) selectEventStateKeyNID(
return types.EventStateKeyNID(eventStateKeyNID), err
}
func (s *eventStateKeyStatements) bulkSelectEventStateKeyNID(
func (s *eventStateKeyStatements) BulkSelectEventStateKeyNID(
ctx context.Context, eventStateKeys []string,
) (map[string]types.EventStateKeyNID, error) {
rows, err := s.bulkSelectEventStateKeyNIDStmt.QueryContext(
@ -128,7 +130,7 @@ func (s *eventStateKeyStatements) bulkSelectEventStateKeyNID(
return result, rows.Err()
}
func (s *eventStateKeyStatements) bulkSelectEventStateKey(
func (s *eventStateKeyStatements) BulkSelectEventStateKey(
ctx context.Context, eventStateKeyNIDs []types.EventStateKeyNID,
) (map[types.EventStateKeyNID]string, error) {
nIDs := make(pq.Int64Array, len(eventStateKeyNIDs))

24
roomserver/storage/postgres/event_types_table.go
View file

@ -22,6 +22,7 @@ import (
"github.com/matrix-org/dendrite/internal"
"github.com/lib/pq"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
)
@ -98,36 +99,37 @@ type eventTypeStatements struct {
bulkSelectEventTypeNIDStmt *sql.Stmt
}
func (s *eventTypeStatements) prepare(db *sql.DB) (err error) {
_, err = db.Exec(eventTypesSchema)
func NewPostgresEventTypesTable(db *sql.DB) (tables.EventTypes, error) {
s := &eventTypeStatements{}
_, err := db.Exec(eventTypesSchema)
if err != nil {
return
return nil, err
}
return statementList{
return s, statementList{
{&s.insertEventTypeNIDStmt, insertEventTypeNIDSQL},
{&s.selectEventTypeNIDStmt, selectEventTypeNIDSQL},
{&s.bulkSelectEventTypeNIDStmt, bulkSelectEventTypeNIDSQL},
}.prepare(db)
}
func (s *eventTypeStatements) insertEventTypeNID(
ctx context.Context, eventType string,
func (s *eventTypeStatements) InsertEventTypeNID(
ctx context.Context, txn *sql.Tx, eventType string,
) (types.EventTypeNID, error) {
var eventTypeNID int64
err := s.insertEventTypeNIDStmt.QueryRowContext(ctx, eventType).Scan(&eventTypeNID)
err := txn.Stmt(s.insertEventTypeNIDStmt).QueryRowContext(ctx, eventType).Scan(&eventTypeNID)
return types.EventTypeNID(eventTypeNID), err
}
func (s *eventTypeStatements) selectEventTypeNID(
ctx context.Context, eventType string,
func (s *eventTypeStatements) SelectEventTypeNID(
ctx context.Context, txn *sql.Tx, eventType string,
) (types.EventTypeNID, error) {
var eventTypeNID int64
err := s.selectEventTypeNIDStmt.QueryRowContext(ctx, eventType).Scan(&eventTypeNID)
err := txn.Stmt(s.selectEventTypeNIDStmt).QueryRowContext(ctx, eventType).Scan(&eventTypeNID)
return types.EventTypeNID(eventTypeNID), err
}
func (s *eventTypeStatements) bulkSelectEventTypeNID(
func (s *eventTypeStatements) BulkSelectEventTypeNID(
ctx context.Context, eventTypes []string,
) (map[string]types.EventTypeNID, error) {
rows, err := s.bulkSelectEventTypeNIDStmt.QueryContext(ctx, pq.StringArray(eventTypes))

43
roomserver/storage/postgres/events_table.go
View file

@ -22,6 +22,7 @@ import (
"github.com/lib/pq"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
)
@ -136,13 +137,14 @@ type eventStatements struct {
selectRoomNIDForEventNIDStmt *sql.Stmt
}
func (s *eventStatements) prepare(db *sql.DB) (err error) {
_, err = db.Exec(eventsSchema)
func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
s := &eventStatements{}
_, err := db.Exec(eventsSchema)
if err != nil {
return
return nil, err
}
return statementList{
return s, statementList{
{&s.insertEventStmt, insertEventSQL},
{&s.selectEventStmt, selectEventSQL},
{&s.bulkSelectStateEventByIDStmt, bulkSelectStateEventByIDSQL},
@ -160,8 +162,9 @@ func (s *eventStatements) prepare(db *sql.DB) (err error) {
}.prepare(db)
}
func (s *eventStatements) insertEvent(
func (s *eventStatements) InsertEvent(
ctx context.Context,
txn *sql.Tx,
roomNID types.RoomNID,
eventTypeNID types.EventTypeNID,
eventStateKeyNID types.EventStateKeyNID,
@ -179,8 +182,8 @@ func (s *eventStatements) insertEvent(
return types.EventNID(eventNID), types.StateSnapshotNID(stateNID), err
}
func (s *eventStatements) selectEvent(
ctx context.Context, eventID string,
func (s *eventStatements) SelectEvent(
ctx context.Context, txn *sql.Tx, eventID string,
) (types.EventNID, types.StateSnapshotNID, error) {
var eventNID int64
var stateNID int64
@ -190,7 +193,7 @@ func (s *eventStatements) selectEvent(
// bulkSelectStateEventByID lookups a list of state events by event ID.
// If any of the requested events are missing from the database it returns a types.MissingEventError
func (s *eventStatements) bulkSelectStateEventByID(
func (s *eventStatements) BulkSelectStateEventByID(
ctx context.Context, eventIDs []string,
) ([]types.StateEntry, error) {
rows, err := s.bulkSelectStateEventByIDStmt.QueryContext(ctx, pq.StringArray(eventIDs))
@ -233,7 +236,7 @@ func (s *eventStatements) bulkSelectStateEventByID(
// bulkSelectStateAtEventByID lookups the state at a list of events by event ID.
// If any of the requested events are missing from the database it returns a types.MissingEventError.
// If we do not have the state for any of the requested events it returns a types.MissingEventError.
func (s *eventStatements) bulkSelectStateAtEventByID(
func (s *eventStatements) BulkSelectStateAtEventByID(
ctx context.Context, eventIDs []string,
) ([]types.StateAtEvent, error) {
rows, err := s.bulkSelectStateAtEventByIDStmt.QueryContext(ctx, pq.StringArray(eventIDs))
@ -270,14 +273,14 @@ func (s *eventStatements) bulkSelectStateAtEventByID(
return results, nil
}
func (s *eventStatements) updateEventState(
func (s *eventStatements) UpdateEventState(
ctx context.Context, eventNID types.EventNID, stateNID types.StateSnapshotNID,
) error {
_, err := s.updateEventStateStmt.ExecContext(ctx, int64(eventNID), int64(stateNID))
return err
}
func (s *eventStatements) selectEventSentToOutput(
func (s *eventStatements) SelectEventSentToOutput(
ctx context.Context, txn *sql.Tx, eventNID types.EventNID,
) (sentToOutput bool, err error) {
stmt := internal.TxStmt(txn, s.selectEventSentToOutputStmt)
@ -285,13 +288,13 @@ func (s *eventStatements) selectEventSentToOutput(
return
}
func (s *eventStatements) updateEventSentToOutput(ctx context.Context, txn *sql.Tx, eventNID types.EventNID) error {
func (s *eventStatements) UpdateEventSentToOutput(ctx context.Context, txn *sql.Tx, eventNID types.EventNID) error {
stmt := internal.TxStmt(txn, s.updateEventSentToOutputStmt)
_, err := stmt.ExecContext(ctx, int64(eventNID))
return err
}
func (s *eventStatements) selectEventID(
func (s *eventStatements) SelectEventID(
ctx context.Context, txn *sql.Tx, eventNID types.EventNID,
) (eventID string, err error) {
stmt := internal.TxStmt(txn, s.selectEventIDStmt)
@ -299,7 +302,7 @@ func (s *eventStatements) selectEventID(
return
}
func (s *eventStatements) bulkSelectStateAtEventAndReference(
func (s *eventStatements) BulkSelectStateAtEventAndReference(
ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID,
) ([]types.StateAtEventAndReference, error) {
stmt := internal.TxStmt(txn, s.bulkSelectStateAtEventAndReferenceStmt)
@ -341,8 +344,8 @@ func (s *eventStatements) bulkSelectStateAtEventAndReference(
return results, nil
}
func (s *eventStatements) bulkSelectEventReference(
ctx context.Context, eventNIDs []types.EventNID,
func (s *eventStatements) BulkSelectEventReference(
ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID,
) ([]gomatrixserverlib.EventReference, error) {
rows, err := s.bulkSelectEventReferenceStmt.QueryContext(ctx, eventNIDsAsArray(eventNIDs))
if err != nil {
@ -367,7 +370,7 @@ func (s *eventStatements) bulkSelectEventReference(
}
// bulkSelectEventID returns a map from numeric event ID to string event ID.
func (s *eventStatements) bulkSelectEventID(ctx context.Context, eventNIDs []types.EventNID) (map[types.EventNID]string, error) {
func (s *eventStatements) BulkSelectEventID(ctx context.Context, eventNIDs []types.EventNID) (map[types.EventNID]string, error) {
rows, err := s.bulkSelectEventIDStmt.QueryContext(ctx, eventNIDsAsArray(eventNIDs))
if err != nil {
return nil, err
@ -394,7 +397,7 @@ func (s *eventStatements) bulkSelectEventID(ctx context.Context, eventNIDs []typ
// bulkSelectEventNIDs returns a map from string event ID to numeric event ID.
// If an event ID is not in the database then it is omitted from the map.
func (s *eventStatements) bulkSelectEventNID(ctx context.Context, eventIDs []string) (map[string]types.EventNID, error) {
func (s *eventStatements) BulkSelectEventNID(ctx context.Context, eventIDs []string) (map[string]types.EventNID, error) {
rows, err := s.bulkSelectEventNIDStmt.QueryContext(ctx, pq.StringArray(eventIDs))
if err != nil {
return nil, err
@ -412,7 +415,7 @@ func (s *eventStatements) bulkSelectEventNID(ctx context.Context, eventIDs []str
return results, rows.Err()
}
func (s *eventStatements) selectMaxEventDepth(ctx context.Context, eventNIDs []types.EventNID) (int64, error) {
func (s *eventStatements) SelectMaxEventDepth(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (int64, error) {
var result int64
stmt := s.selectMaxEventDepthStmt
err := stmt.QueryRowContext(ctx, eventNIDsAsArray(eventNIDs)).Scan(&result)
@ -422,7 +425,7 @@ func (s *eventStatements) selectMaxEventDepth(ctx context.Context, eventNIDs []t
return result, nil
}
func (s *eventStatements) selectRoomNIDForEventNID(
func (s *eventStatements) SelectRoomNIDForEventNID(
ctx context.Context, txn *sql.Tx, eventNID types.EventNID,
) (roomNID types.RoomNID, err error) {
selectStmt := internal.TxStmt(txn, s.selectRoomNIDForEventNIDStmt)

4
roomserver/storage/postgres/sql.go
View file

@ -38,11 +38,7 @@ func (s *statements) prepare(db *sql.DB) error {
var err error
for _, prepare := range []func(db *sql.DB) error{
s.eventTypeStatements.prepare,
s.eventStateKeyStatements.prepare,
s.roomStatements.prepare,
s.eventStatements.prepare,
s.eventJSONStatements.prepare,
s.stateSnapshotStatements.prepare,
s.stateBlockStatements.prepare,
s.previousEventStatements.prepare,

175
roomserver/storage/postgres/storage.go
View file

@ -26,14 +26,21 @@ import (
// Import the postgres database driver.
_ "github.com/lib/pq"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/storage/shared"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
)
// A Database is used to store room events and stream offsets.
type Database struct {
statements statements
db *sql.DB
shared.Database
statements statements
events tables.Events
eventTypes tables.EventTypes
eventStateKeys tables.EventStateKeys
eventJSON tables.EventJSON
db *sql.DB
}
// Open a postgres database.
@ -46,6 +53,28 @@ func Open(dataSourceName string, dbProperties internal.DbProperties) (*Database,
if err = d.statements.prepare(d.db); err != nil {
return nil, err
}
d.eventStateKeys, err = NewPostgresEventStateKeysTable(d.db)
if err != nil {
return nil, err
}
d.eventTypes, err = NewPostgresEventTypesTable(d.db)
if err != nil {
return nil, err
}
d.eventJSON, err = NewPostgresEventJSONTable(d.db)
if err != nil {
return nil, err
}
d.events, err = NewPostgresEventsTable(d.db)
if err != nil {
return nil, err
}
d.Database = shared.Database{
EventTypesTable: d.eventTypes,
EventStateKeysTable: d.eventStateKeys,
EventJSONTable: d.eventJSON,
EventsTable: d.events,
}
return &d, nil
}
@ -103,8 +132,9 @@ func (d *Database) StoreEvent(
}
}
if eventNID, stateNID, err = d.statements.insertEvent(
if eventNID, stateNID, err = d.events.InsertEvent(
ctx,
nil,
roomNID,
eventTypeNID,
eventStateKeyNID,
@ -115,14 +145,14 @@ func (d *Database) StoreEvent(
); err != nil {
if err == sql.ErrNoRows {
// We've already inserted the event so select the numeric event ID
eventNID, stateNID, err = d.statements.selectEvent(ctx, event.EventID())
eventNID, stateNID, err = d.events.SelectEvent(ctx, nil, event.EventID())
}
if err != nil {
return 0, types.StateAtEvent{}, err
}
}
if err = d.statements.insertEventJSON(ctx, eventNID, event.JSON()); err != nil {
if err = d.eventJSON.InsertEventJSON(ctx, nil, eventNID, event.JSON()); err != nil {
return 0, types.StateAtEvent{}, err
}
@ -180,17 +210,20 @@ func (d *Database) assignRoomNID(
func (d *Database) assignEventTypeNID(
ctx context.Context, eventType string,
) (types.EventTypeNID, error) {
// Check if we already have a numeric ID in the database.
eventTypeNID, err := d.statements.selectEventTypeNID(ctx, eventType)
if err == sql.ErrNoRows {
// We don't have a numeric ID so insert one into the database.
eventTypeNID, err = d.statements.insertEventTypeNID(ctx, eventType)
) (eventTypeNID types.EventTypeNID, err error) {
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
// Check if we already have a numeric ID in the database.
eventTypeNID, err = d.eventTypes.SelectEventTypeNID(ctx, txn, eventType)
if err == sql.ErrNoRows {
// We raced with another insert so run the select again.
eventTypeNID, err = d.statements.selectEventTypeNID(ctx, eventType)
// We don't have a numeric ID so insert one into the database.
eventTypeNID, err = d.eventTypes.InsertEventTypeNID(ctx, txn, eventType)
if err == sql.ErrNoRows {
// We raced with another insert so run the select again.
eventTypeNID, err = d.eventTypes.SelectEventTypeNID(ctx, txn, eventType)
}
}
}
return err
})
return eventTypeNID, err
}
@ -198,58 +231,23 @@ func (d *Database) assignStateKeyNID(
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.statements.selectEventStateKeyNID(ctx, txn, eventStateKey)
eventStateKeyNID, err := d.eventStateKeys.SelectEventStateKeyNID(ctx, txn, eventStateKey)
if err == sql.ErrNoRows {
// We don't have a numeric ID so insert one into the database.
eventStateKeyNID, err = d.statements.insertEventStateKeyNID(ctx, txn, eventStateKey)
eventStateKeyNID, err = d.eventStateKeys.InsertEventStateKeyNID(ctx, txn, eventStateKey)
if err == sql.ErrNoRows {
// We raced with another insert so run the select again.
eventStateKeyNID, err = d.statements.selectEventStateKeyNID(ctx, txn, eventStateKey)
eventStateKeyNID, err = d.eventStateKeys.SelectEventStateKeyNID(ctx, txn, eventStateKey)
}
}
return eventStateKeyNID, err
}
// StateEntriesForEventIDs implements input.EventDatabase
func (d *Database) StateEntriesForEventIDs(
ctx context.Context, eventIDs []string,
) ([]types.StateEntry, error) {
return d.statements.bulkSelectStateEventByID(ctx, eventIDs)
}
// EventTypeNIDs implements state.RoomStateDatabase
func (d *Database) EventTypeNIDs(
ctx context.Context, eventTypes []string,
) (map[string]types.EventTypeNID, error) {
return d.statements.bulkSelectEventTypeNID(ctx, eventTypes)
}
// EventStateKeyNIDs implements state.RoomStateDatabase
func (d *Database) EventStateKeyNIDs(
ctx context.Context, eventStateKeys []string,
) (map[string]types.EventStateKeyNID, error) {
return d.statements.bulkSelectEventStateKeyNID(ctx, eventStateKeys)
}
// EventStateKeys implements query.RoomserverQueryAPIDatabase
func (d *Database) EventStateKeys(
ctx context.Context, eventStateKeyNIDs []types.EventStateKeyNID,
) (map[types.EventStateKeyNID]string, error) {
return d.statements.bulkSelectEventStateKey(ctx, eventStateKeyNIDs)
}
// EventNIDs implements query.RoomserverQueryAPIDatabase
func (d *Database) EventNIDs(
ctx context.Context, eventIDs []string,
) (map[string]types.EventNID, error) {
return d.statements.bulkSelectEventNID(ctx, eventIDs)
}
// Events implements input.EventDatabase
func (d *Database) Events(
ctx context.Context, eventNIDs []types.EventNID,
) ([]types.Event, error) {
eventJSONs, err := d.statements.bulkSelectEventJSON(ctx, eventNIDs)
eventJSONs, err := d.eventJSON.BulkSelectEventJSON(ctx, eventNIDs)
if err != nil {
return nil, err
}
@ -259,7 +257,7 @@ func (d *Database) Events(
var roomVersion gomatrixserverlib.RoomVersion
result := &results[i]
result.EventNID = eventJSON.EventNID
roomNID, err = d.statements.selectRoomNIDForEventNID(ctx, nil, eventJSON.EventNID)
roomNID, err = d.events.SelectRoomNIDForEventNID(ctx, nil, eventJSON.EventNID)
if err != nil {
return nil, err
}
@ -298,20 +296,6 @@ func (d *Database) AddState(
return d.statements.insertState(ctx, roomNID, stateBlockNIDs)
}
// SetState implements input.EventDatabase
func (d *Database) SetState(
ctx context.Context, eventNID types.EventNID, stateNID types.StateSnapshotNID,
) error {
return d.statements.updateEventState(ctx, eventNID, stateNID)
}
// StateAtEventIDs implements input.EventDatabase
func (d *Database) StateAtEventIDs(
ctx context.Context, eventIDs []string,
) ([]types.StateAtEvent, error) {
return d.statements.bulkSelectStateAtEventByID(ctx, eventIDs)
}
// StateBlockNIDs implements state.RoomStateDatabase
func (d *Database) StateBlockNIDs(
ctx context.Context, stateNIDs []types.StateSnapshotNID,
@ -326,21 +310,6 @@ func (d *Database) StateEntries(
return d.statements.bulkSelectStateBlockEntries(ctx, stateBlockNIDs)
}
// SnapshotNIDFromEventID implements state.RoomStateDatabase
func (d *Database) SnapshotNIDFromEventID(
ctx context.Context, eventID string,
) (types.StateSnapshotNID, error) {
_, stateNID, err := d.statements.selectEvent(ctx, eventID)
return stateNID, err
}
// EventIDs implements input.RoomEventDatabase
func (d *Database) EventIDs(
ctx context.Context, eventNIDs []types.EventNID,
) (map[types.EventNID]string, error) {
return d.statements.bulkSelectEventID(ctx, eventNIDs)
}
// GetLatestEventsForUpdate implements input.EventDatabase
func (d *Database) GetLatestEventsForUpdate(
ctx context.Context, roomNID types.RoomNID,
@ -355,14 +324,14 @@ func (d *Database) GetLatestEventsForUpdate(
txn.Rollback() // nolint: errcheck
return nil, err
}
stateAndRefs, err := d.statements.bulkSelectStateAtEventAndReference(ctx, txn, eventNIDs)
stateAndRefs, err := d.events.BulkSelectStateAtEventAndReference(ctx, txn, eventNIDs)
if err != nil {
txn.Rollback() // nolint: errcheck
return nil, err
}
var lastEventIDSent string
if lastEventNIDSent != 0 {
lastEventIDSent, err = d.statements.selectEventID(ctx, txn, lastEventNIDSent)
lastEventIDSent, err = d.events.SelectEventID(ctx, txn, lastEventNIDSent)
if err != nil {
txn.Rollback() // nolint: errcheck
return nil, err
@ -451,12 +420,12 @@ func (u *roomRecentEventsUpdater) SetLatestEvents(
// HasEventBeenSent implements types.RoomRecentEventsUpdater
func (u *roomRecentEventsUpdater) HasEventBeenSent(eventNID types.EventNID) (bool, error) {
return u.d.statements.selectEventSentToOutput(u.ctx, u.txn, eventNID)
return u.d.events.SelectEventSentToOutput(u.ctx, u.txn, eventNID)
}
// MarkEventAsSent implements types.RoomRecentEventsUpdater
func (u *roomRecentEventsUpdater) MarkEventAsSent(eventNID types.EventNID) error {
return u.d.statements.updateEventSentToOutput(u.ctx, u.txn, eventNID)
return u.d.events.UpdateEventSentToOutput(u.ctx, u.txn, eventNID)
}
func (u *roomRecentEventsUpdater) MembershipUpdater(targetUserNID types.EventStateKeyNID, targetLocal bool) (types.MembershipUpdater, error) {
@ -492,20 +461,24 @@ func (d *Database) RoomNIDExcludingStubs(ctx context.Context, roomID string) (ro
// LatestEventIDs implements query.RoomserverQueryAPIDatabase
func (d *Database) LatestEventIDs(
ctx context.Context, roomNID types.RoomNID,
) ([]gomatrixserverlib.EventReference, types.StateSnapshotNID, int64, error) {
eventNIDs, currentStateSnapshotNID, err := d.statements.selectLatestEventNIDs(ctx, roomNID)
if err != nil {
return nil, 0, 0, err
}
references, err := d.statements.bulkSelectEventReference(ctx, eventNIDs)
if err != nil {
return nil, 0, 0, err
}
depth, err := d.statements.selectMaxEventDepth(ctx, eventNIDs)
if err != nil {
return nil, 0, 0, err
}
return references, currentStateSnapshotNID, depth, nil
) (references []gomatrixserverlib.EventReference, currentStateSnapshotNID types.StateSnapshotNID, depth int64, err error) {
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
var eventNIDs []types.EventNID
eventNIDs, currentStateSnapshotNID, err = d.statements.selectLatestEventNIDs(ctx, roomNID)
if err != nil {
return err
}
references, err = d.events.BulkSelectEventReference(ctx, txn, eventNIDs)
if err != nil {
return err
}
depth, err = d.events.SelectMaxEventDepth(ctx, txn, eventNIDs)
if err != nil {
return err
}
return nil
})
return
}
// GetInvitesForUser implements query.RoomserverQueryAPIDatabase

79
roomserver/storage/shared/storage.go Normal file
View file

@ -0,0 +1,79 @@
package shared
import (
"context"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
)
type Database struct {
EventsTable tables.Events
EventJSONTable tables.EventJSON
EventTypesTable tables.EventTypes
EventStateKeysTable tables.EventStateKeys
}
// EventTypeNIDs implements state.RoomStateDatabase
func (d *Database) EventTypeNIDs(
ctx context.Context, eventTypes []string,
) (map[string]types.EventTypeNID, error) {
return d.EventTypesTable.BulkSelectEventTypeNID(ctx, eventTypes)
}
// EventStateKeys implements query.RoomserverQueryAPIDatabase
func (d *Database) EventStateKeys(
ctx context.Context, eventStateKeyNIDs []types.EventStateKeyNID,
) (map[types.EventStateKeyNID]string, error) {
return d.EventStateKeysTable.BulkSelectEventStateKey(ctx, eventStateKeyNIDs)
}
// EventStateKeyNIDs implements state.RoomStateDatabase
func (d *Database) EventStateKeyNIDs(
ctx context.Context, eventStateKeys []string,
) (map[string]types.EventStateKeyNID, error) {
return d.EventStateKeysTable.BulkSelectEventStateKeyNID(ctx, eventStateKeys)
}
// StateEntriesForEventIDs implements input.EventDatabase
func (d *Database) StateEntriesForEventIDs(
ctx context.Context, eventIDs []string,
) ([]types.StateEntry, error) {
return d.EventsTable.BulkSelectStateEventByID(ctx, eventIDs)
}
// EventNIDs implements query.RoomserverQueryAPIDatabase
func (d *Database) EventNIDs(
ctx context.Context, eventIDs []string,
) (map[string]types.EventNID, error) {
return d.EventsTable.BulkSelectEventNID(ctx, eventIDs)
}
// SetState implements input.EventDatabase
func (d *Database) SetState(
ctx context.Context, eventNID types.EventNID, stateNID types.StateSnapshotNID,
) error {
return d.EventsTable.UpdateEventState(ctx, eventNID, stateNID)
}
// StateAtEventIDs implements input.EventDatabase
func (d *Database) StateAtEventIDs(
ctx context.Context, eventIDs []string,
) ([]types.StateAtEvent, error) {
return d.EventsTable.BulkSelectStateAtEventByID(ctx, eventIDs)
}
// SnapshotNIDFromEventID implements state.RoomStateDatabase
func (d *Database) SnapshotNIDFromEventID(
ctx context.Context, eventID string,
) (types.StateSnapshotNID, error) {
_, stateNID, err := d.EventsTable.SelectEvent(ctx, nil, eventID)
return stateNID, err
}
// EventIDs implements input.RoomEventDatabase
func (d *Database) EventIDs(
ctx context.Context, eventNIDs []types.EventNID,
) (map[types.EventNID]string, error) {
return d.EventsTable.BulkSelectEventID(ctx, eventNIDs)
}

27
roomserver/storage/sqlite3/event_json_table.go
View file

@ -21,6 +21,7 @@ import (
"strings"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
)
@ -51,40 +52,36 @@ type eventJSONStatements struct {
bulkSelectEventJSONStmt *sql.Stmt
}
func (s *eventJSONStatements) prepare(db *sql.DB) (err error) {
func NewSqliteEventJSONTable(db *sql.DB) (tables.EventJSON, error) {
s := &eventJSONStatements{}
s.db = db
_, err = db.Exec(eventJSONSchema)
_, err := db.Exec(eventJSONSchema)
if err != nil {
return
return nil, err
}
return statementList{
return s, statementList{
{&s.insertEventJSONStmt, insertEventJSONSQL},
{&s.bulkSelectEventJSONStmt, bulkSelectEventJSONSQL},
}.prepare(db)
}
func (s *eventJSONStatements) insertEventJSON(
func (s *eventJSONStatements) InsertEventJSON(
ctx context.Context, txn *sql.Tx, eventNID types.EventNID, eventJSON []byte,
) error {
_, err := internal.TxStmt(txn, s.insertEventJSONStmt).ExecContext(ctx, int64(eventNID), eventJSON)
return err
}
type eventJSONPair struct {
EventNID types.EventNID
EventJSON []byte
}
func (s *eventJSONStatements) bulkSelectEventJSON(
ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID,
) ([]eventJSONPair, error) {
func (s *eventJSONStatements) BulkSelectEventJSON(
ctx context.Context, eventNIDs []types.EventNID,
) ([]tables.EventJSONPair, error) {
iEventNIDs := make([]interface{}, len(eventNIDs))
for k, v := range eventNIDs {
iEventNIDs[k] = v
}
selectOrig := strings.Replace(bulkSelectEventJSONSQL, "($1)", internal.QueryVariadic(len(iEventNIDs)), 1)
rows, err := txn.QueryContext(ctx, selectOrig, iEventNIDs...)
rows, err := s.db.QueryContext(ctx, selectOrig, iEventNIDs...)
if err != nil {
return nil, err
}
@ -94,7 +91,7 @@ func (s *eventJSONStatements) bulkSelectEventJSON(
// because of the unique constraint on event NIDs.
// So we can allocate an array of the correct size now.
// We might get fewer results than NIDs so we adjust the length of the slice before returning it.
results := make([]eventJSONPair, len(eventNIDs))
results := make([]tables.EventJSONPair, len(eventNIDs))
i := 0
for ; rows.Next(); i++ {
result := &results[i]

26
roomserver/storage/sqlite3/event_state_keys_table.go
View file

@ -21,6 +21,7 @@ import (
"strings"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
)
@ -67,13 +68,14 @@ type eventStateKeyStatements struct {
bulkSelectEventStateKeyStmt *sql.Stmt
}
func (s *eventStateKeyStatements) prepare(db *sql.DB) (err error) {
func NewSqliteEventStateKeysTable(db *sql.DB) (tables.EventStateKeys, error) {
s := &eventStateKeyStatements{}
s.db = db
_, err = db.Exec(eventStateKeysSchema)
_, err := db.Exec(eventStateKeysSchema)
if err != nil {
return
return nil, err
}
return statementList{
return s, statementList{
{&s.insertEventStateKeyNIDStmt, insertEventStateKeyNIDSQL},
{&s.selectEventStateKeyNIDStmt, selectEventStateKeyNIDSQL},
{&s.bulkSelectEventStateKeyNIDStmt, bulkSelectEventStateKeyNIDSQL},
@ -81,7 +83,7 @@ func (s *eventStateKeyStatements) prepare(db *sql.DB) (err error) {
}.prepare(db)
}
func (s *eventStateKeyStatements) insertEventStateKeyNID(
func (s *eventStateKeyStatements) InsertEventStateKeyNID(
ctx context.Context, txn *sql.Tx, eventStateKey string,
) (types.EventStateKeyNID, error) {
var eventStateKeyNID int64
@ -94,7 +96,7 @@ func (s *eventStateKeyStatements) insertEventStateKeyNID(
return types.EventStateKeyNID(eventStateKeyNID), err
}
func (s *eventStateKeyStatements) selectEventStateKeyNID(
func (s *eventStateKeyStatements) SelectEventStateKeyNID(
ctx context.Context, txn *sql.Tx, eventStateKey string,
) (types.EventStateKeyNID, error) {
var eventStateKeyNID int64
@ -103,8 +105,8 @@ func (s *eventStateKeyStatements) selectEventStateKeyNID(
return types.EventStateKeyNID(eventStateKeyNID), err
}
func (s *eventStateKeyStatements) bulkSelectEventStateKeyNID(
ctx context.Context, txn *sql.Tx, eventStateKeys []string,
func (s *eventStateKeyStatements) BulkSelectEventStateKeyNID(
ctx context.Context, eventStateKeys []string,
) (map[string]types.EventStateKeyNID, error) {
iEventStateKeys := make([]interface{}, len(eventStateKeys))
for k, v := range eventStateKeys {
@ -112,7 +114,7 @@ func (s *eventStateKeyStatements) bulkSelectEventStateKeyNID(
}
selectOrig := strings.Replace(bulkSelectEventStateKeySQL, "($1)", internal.QueryVariadic(len(eventStateKeys)), 1)
rows, err := txn.QueryContext(ctx, selectOrig, iEventStateKeys...)
rows, err := s.db.QueryContext(ctx, selectOrig, iEventStateKeys...)
if err != nil {
return nil, err
}
@ -129,8 +131,8 @@ func (s *eventStateKeyStatements) bulkSelectEventStateKeyNID(
return result, nil
}
func (s *eventStateKeyStatements) bulkSelectEventStateKey(
ctx context.Context, txn *sql.Tx, eventStateKeyNIDs []types.EventStateKeyNID,
func (s *eventStateKeyStatements) BulkSelectEventStateKey(
ctx context.Context, eventStateKeyNIDs []types.EventStateKeyNID,
) (map[types.EventStateKeyNID]string, error) {
iEventStateKeyNIDs := make([]interface{}, len(eventStateKeyNIDs))
for k, v := range eventStateKeyNIDs {
@ -138,7 +140,7 @@ func (s *eventStateKeyStatements) bulkSelectEventStateKey(
}
selectOrig := strings.Replace(bulkSelectEventStateKeyNIDSQL, "($1)", internal.QueryVariadic(len(eventStateKeyNIDs)), 1)
rows, err := txn.QueryContext(ctx, selectOrig, iEventStateKeyNIDs...)
rows, err := s.db.QueryContext(ctx, selectOrig, iEventStateKeyNIDs...)
if err != nil {
return nil, err
}

21
roomserver/storage/sqlite3/event_types_table.go
View file

@ -21,6 +21,7 @@ import (
"strings"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
)
@ -81,14 +82,15 @@ type eventTypeStatements struct {
bulkSelectEventTypeNIDStmt *sql.Stmt
}
func (s *eventTypeStatements) prepare(db *sql.DB) (err error) {
func NewSqliteEventTypesTable(db *sql.DB) (tables.EventTypes, error) {
s := &eventTypeStatements{}
s.db = db
_, err = db.Exec(eventTypesSchema)
_, err := db.Exec(eventTypesSchema)
if err != nil {
return
return nil, err
}
return statementList{
return s, statementList{
{&s.insertEventTypeNIDStmt, insertEventTypeNIDSQL},
{&s.insertEventTypeNIDResultStmt, insertEventTypeNIDResultSQL},
{&s.selectEventTypeNIDStmt, selectEventTypeNIDSQL},
@ -96,7 +98,7 @@ func (s *eventTypeStatements) prepare(db *sql.DB) (err error) {
}.prepare(db)
}
func (s *eventTypeStatements) insertEventTypeNID(
func (s *eventTypeStatements) InsertEventTypeNID(
ctx context.Context, tx *sql.Tx, eventType string,
) (types.EventTypeNID, error) {
var eventTypeNID int64
@ -109,7 +111,7 @@ func (s *eventTypeStatements) insertEventTypeNID(
return types.EventTypeNID(eventTypeNID), err
}
func (s *eventTypeStatements) selectEventTypeNID(
func (s *eventTypeStatements) SelectEventTypeNID(
ctx context.Context, tx *sql.Tx, eventType string,
) (types.EventTypeNID, error) {
var eventTypeNID int64
@ -118,8 +120,8 @@ func (s *eventTypeStatements) selectEventTypeNID(
return types.EventTypeNID(eventTypeNID), err
}
func (s *eventTypeStatements) bulkSelectEventTypeNID(
ctx context.Context, tx *sql.Tx, eventTypes []string,
func (s *eventTypeStatements) BulkSelectEventTypeNID(
ctx context.Context, eventTypes []string,
) (map[string]types.EventTypeNID, error) {
///////////////
iEventTypes := make([]interface{}, len(eventTypes))
@ -133,8 +135,7 @@ func (s *eventTypeStatements) bulkSelectEventTypeNID(
}
///////////////
selectStmt := internal.TxStmt(tx, selectPrep)
rows, err := selectStmt.QueryContext(ctx, iEventTypes...)
rows, err := selectPrep.QueryContext(ctx, iEventTypes...)
if err != nil {
return nil, err
}

69
roomserver/storage/sqlite3/events_table.go
View file

@ -23,6 +23,7 @@ import (
"strings"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
)
@ -111,14 +112,15 @@ type eventStatements struct {
selectRoomNIDForEventNIDStmt *sql.Stmt
}
func (s *eventStatements) prepare(db *sql.DB) (err error) {
func NewSqliteEventsTable(db *sql.DB) (tables.Events, error) {
s := &eventStatements{}
s.db = db
_, err = db.Exec(eventsSchema)
_, err := db.Exec(eventsSchema)
if err != nil {
return
return nil, err
}
return statementList{
return s, statementList{
{&s.insertEventStmt, insertEventSQL},
{&s.selectEventStmt, selectEventSQL},
{&s.bulkSelectStateEventByIDStmt, bulkSelectStateEventByIDSQL},
@ -135,7 +137,7 @@ func (s *eventStatements) prepare(db *sql.DB) (err error) {
}.prepare(db)
}
func (s *eventStatements) insertEvent(
func (s *eventStatements) InsertEvent(
ctx context.Context,
txn *sql.Tx,
roomNID types.RoomNID,
@ -145,7 +147,7 @@ func (s *eventStatements) insertEvent(
referenceSHA256 []byte,
authEventNIDs []types.EventNID,
depth int64,
) (types.EventNID, error) {
) (types.EventNID, types.StateSnapshotNID, error) {
// attempt to insert: the last_row_id is the event NID
insertStmt := internal.TxStmt(txn, s.insertEventStmt)
result, err := insertStmt.ExecContext(
@ -153,17 +155,17 @@ func (s *eventStatements) insertEvent(
eventID, referenceSHA256, eventNIDsAsArray(authEventNIDs), depth,
)
if err != nil {
return 0, err
return 0, 0, err
}
modified, err := result.RowsAffected()
if modified == 0 && err == nil {
return 0, sql.ErrNoRows
return 0, 0, sql.ErrNoRows
}
eventNID, err := result.LastInsertId()
return types.EventNID(eventNID), err
return types.EventNID(eventNID), 0, err
}
func (s *eventStatements) selectEvent(
func (s *eventStatements) SelectEvent(
ctx context.Context, txn *sql.Tx, eventID string,
) (types.EventNID, types.StateSnapshotNID, error) {
var eventNID int64
@ -175,8 +177,8 @@ func (s *eventStatements) selectEvent(
// bulkSelectStateEventByID lookups a list of state events by event ID.
// If any of the requested events are missing from the database it returns a types.MissingEventError
func (s *eventStatements) bulkSelectStateEventByID(
ctx context.Context, txn *sql.Tx, eventIDs []string,
func (s *eventStatements) BulkSelectStateEventByID(
ctx context.Context, eventIDs []string,
) ([]types.StateEntry, error) {
///////////////
iEventIDs := make([]interface{}, len(eventIDs))
@ -184,13 +186,12 @@ func (s *eventStatements) bulkSelectStateEventByID(
iEventIDs[k] = v
}
selectOrig := strings.Replace(bulkSelectStateEventByIDSQL, "($1)", internal.QueryVariadic(len(iEventIDs)), 1)
selectPrep, err := txn.Prepare(selectOrig)
selectStmt, err := s.db.Prepare(selectOrig)
if err != nil {
return nil, err
}
///////////////
selectStmt := internal.TxStmt(txn, selectPrep)
rows, err := selectStmt.QueryContext(ctx, iEventIDs...)
if err != nil {
return nil, err
@ -228,8 +229,8 @@ func (s *eventStatements) bulkSelectStateEventByID(
// bulkSelectStateAtEventByID lookups the state at a list of events by event ID.
// If any of the requested events are missing from the database it returns a types.MissingEventError.
// If we do not have the state for any of the requested events it returns a types.MissingEventError.
func (s *eventStatements) bulkSelectStateAtEventByID(
ctx context.Context, txn *sql.Tx, eventIDs []string,
func (s *eventStatements) BulkSelectStateAtEventByID(
ctx context.Context, eventIDs []string,
) ([]types.StateAtEvent, error) {
///////////////
iEventIDs := make([]interface{}, len(eventIDs))
@ -237,13 +238,11 @@ func (s *eventStatements) bulkSelectStateAtEventByID(
iEventIDs[k] = v
}
selectOrig := strings.Replace(bulkSelectStateAtEventByIDSQL, "($1)", internal.QueryVariadic(len(iEventIDs)), 1)
selectPrep, err := txn.Prepare(selectOrig)
selectStmt, err := s.db.Prepare(selectOrig)
if err != nil {
return nil, err
}
///////////////
selectStmt := internal.TxStmt(txn, selectPrep)
rows, err := selectStmt.QueryContext(ctx, iEventIDs...)
if err != nil {
return nil, err
@ -275,15 +274,14 @@ func (s *eventStatements) bulkSelectStateAtEventByID(
return results, err
}
func (s *eventStatements) updateEventState(
ctx context.Context, txn *sql.Tx, eventNID types.EventNID, stateNID types.StateSnapshotNID,
func (s *eventStatements) UpdateEventState(
ctx context.Context, eventNID types.EventNID, stateNID types.StateSnapshotNID,
) error {
updateStmt := internal.TxStmt(txn, s.updateEventStateStmt)
_, err := updateStmt.ExecContext(ctx, int64(stateNID), int64(eventNID))
_, err := s.updateEventStateStmt.ExecContext(ctx, int64(stateNID), int64(eventNID))
return err
}
func (s *eventStatements) selectEventSentToOutput(
func (s *eventStatements) SelectEventSentToOutput(
ctx context.Context, txn *sql.Tx, eventNID types.EventNID,
) (sentToOutput bool, err error) {
selectStmt := internal.TxStmt(txn, s.selectEventSentToOutputStmt)
@ -294,14 +292,14 @@ func (s *eventStatements) selectEventSentToOutput(
return
}
func (s *eventStatements) updateEventSentToOutput(ctx context.Context, txn *sql.Tx, eventNID types.EventNID) error {
func (s *eventStatements) UpdateEventSentToOutput(ctx context.Context, txn *sql.Tx, eventNID types.EventNID) error {
updateStmt := internal.TxStmt(txn, s.updateEventSentToOutputStmt)
_, err := updateStmt.ExecContext(ctx, int64(eventNID))
//_, err := s.updateEventSentToOutputStmt.ExecContext(ctx, int64(eventNID))
return err
}
func (s *eventStatements) selectEventID(
func (s *eventStatements) SelectEventID(
ctx context.Context, txn *sql.Tx, eventNID types.EventNID,
) (eventID string, err error) {
selectStmt := internal.TxStmt(txn, s.selectEventIDStmt)
@ -309,7 +307,7 @@ func (s *eventStatements) selectEventID(
return
}
func (s *eventStatements) bulkSelectStateAtEventAndReference(
func (s *eventStatements) BulkSelectStateAtEventAndReference(
ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID,
) ([]types.StateAtEventAndReference, error) {
///////////////
@ -355,7 +353,7 @@ func (s *eventStatements) bulkSelectStateAtEventAndReference(
return results, nil
}
func (s *eventStatements) bulkSelectEventReference(
func (s *eventStatements) BulkSelectEventReference(
ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID,
) ([]gomatrixserverlib.EventReference, error) {
///////////////
@ -391,20 +389,19 @@ func (s *eventStatements) bulkSelectEventReference(
}
// bulkSelectEventID returns a map from numeric event ID to string event ID.
func (s *eventStatements) bulkSelectEventID(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (map[types.EventNID]string, error) {
func (s *eventStatements) BulkSelectEventID(ctx context.Context, eventNIDs []types.EventNID) (map[types.EventNID]string, error) {
///////////////
iEventNIDs := make([]interface{}, len(eventNIDs))
for k, v := range eventNIDs {
iEventNIDs[k] = v
}
selectOrig := strings.Replace(bulkSelectEventIDSQL, "($1)", internal.QueryVariadic(len(iEventNIDs)), 1)
selectPrep, err := txn.Prepare(selectOrig)
selectStmt, err := s.db.Prepare(selectOrig)
if err != nil {
return nil, err
}
///////////////
selectStmt := internal.TxStmt(txn, selectPrep)
rows, err := selectStmt.QueryContext(ctx, iEventNIDs...)
if err != nil {
return nil, err
@ -428,20 +425,18 @@ func (s *eventStatements) bulkSelectEventID(ctx context.Context, txn *sql.Tx, ev
// bulkSelectEventNIDs returns a map from string event ID to numeric event ID.
// If an event ID is not in the database then it is omitted from the map.
func (s *eventStatements) bulkSelectEventNID(ctx context.Context, txn *sql.Tx, eventIDs []string) (map[string]types.EventNID, error) {
func (s *eventStatements) BulkSelectEventNID(ctx context.Context, eventIDs []string) (map[string]types.EventNID, error) {
///////////////
iEventIDs := make([]interface{}, len(eventIDs))
for k, v := range eventIDs {
iEventIDs[k] = v
}
selectOrig := strings.Replace(bulkSelectEventNIDSQL, "($1)", internal.QueryVariadic(len(iEventIDs)), 1)
selectPrep, err := txn.Prepare(selectOrig)
selectStmt, err := s.db.Prepare(selectOrig)
if err != nil {
return nil, err
}
///////////////
selectStmt := internal.TxStmt(txn, selectPrep)
rows, err := selectStmt.QueryContext(ctx, iEventIDs...)
if err != nil {
return nil, err
@ -459,7 +454,7 @@ func (s *eventStatements) bulkSelectEventNID(ctx context.Context, txn *sql.Tx, e
return results, nil
}
func (s *eventStatements) selectMaxEventDepth(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (int64, error) {
func (s *eventStatements) SelectMaxEventDepth(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (int64, error) {
var result int64
iEventIDs := make([]interface{}, len(eventNIDs))
for i, v := range eventNIDs {
@ -473,7 +468,7 @@ func (s *eventStatements) selectMaxEventDepth(ctx context.Context, txn *sql.Tx,
return result, nil
}
func (s *eventStatements) selectRoomNIDForEventNID(
func (s *eventStatements) SelectRoomNIDForEventNID(
ctx context.Context, txn *sql.Tx, eventNID types.EventNID,
) (roomNID types.RoomNID, err error) {
selectStmt := internal.TxStmt(txn, s.selectRoomNIDForEventNIDStmt)

4
roomserver/storage/sqlite3/sql.go
View file

@ -38,11 +38,7 @@ func (s *statements) prepare(db *sql.DB) error {
var err error
for _, prepare := range []func(db *sql.DB) error{
s.eventTypeStatements.prepare,
s.eventStateKeyStatements.prepare,
s.roomStatements.prepare,
s.eventStatements.prepare,
s.eventJSONStatements.prepare,
s.stateSnapshotStatements.prepare,
s.stateBlockStatements.prepare,
s.previousEventStatements.prepare,

173
roomserver/storage/sqlite3/storage.go
View file

@ -26,6 +26,8 @@ import (
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/storage/shared"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
_ "github.com/mattn/go-sqlite3"
@ -33,8 +35,13 @@ import (
// A Database is used to store room events and stream offsets.
type Database struct {
statements statements
db *sql.DB
shared.Database
statements statements
events tables.Events
eventJSON tables.EventJSON
eventTypes tables.EventTypes
eventStateKeys tables.EventStateKeys
db *sql.DB
}
// Open a sqlite database.
@ -66,6 +73,28 @@ func Open(dataSourceName string) (*Database, error) {
if err = d.statements.prepare(d.db); err != nil {
return nil, err
}
d.eventStateKeys, err = NewSqliteEventStateKeysTable(d.db)
if err != nil {
return nil, err
}
d.eventTypes, err = NewSqliteEventTypesTable(d.db)
if err != nil {
return nil, err
}
d.eventJSON, err = NewSqliteEventJSONTable(d.db)
if err != nil {
return nil, err
}
d.events, err = NewSqliteEventsTable(d.db)
if err != nil {
return nil, err
}
d.Database = shared.Database{
EventsTable: d.events,
EventTypesTable: d.eventTypes,
EventStateKeysTable: d.eventStateKeys,
EventJSONTable: d.eventJSON,
}
return &d, nil
}
@ -124,7 +153,7 @@ func (d *Database) StoreEvent(
}
}
if eventNID, err = d.statements.insertEvent(
if eventNID, stateNID, err = d.events.InsertEvent(
ctx,
txn,
roomNID,
@ -137,14 +166,14 @@ func (d *Database) StoreEvent(
); err != nil {
if err == sql.ErrNoRows {
// We've already inserted the event so select the numeric event ID
eventNID, stateNID, err = d.statements.selectEvent(ctx, txn, event.EventID())
eventNID, stateNID, err = d.events.SelectEvent(ctx, txn, event.EventID())
}
if err != nil {
return err
}
}
if err = d.statements.insertEventJSON(ctx, txn, eventNID, event.JSON()); err != nil {
if err = d.eventJSON.InsertEventJSON(ctx, txn, eventNID, event.JSON()); err != nil {
return err
}
@ -210,13 +239,13 @@ func (d *Database) assignEventTypeNID(
ctx context.Context, txn *sql.Tx, eventType string,
) (eventTypeNID types.EventTypeNID, err error) {
// Check if we already have a numeric ID in the database.
eventTypeNID, err = d.statements.selectEventTypeNID(ctx, txn, eventType)
eventTypeNID, err = d.eventTypes.SelectEventTypeNID(ctx, txn, eventType)
if err == sql.ErrNoRows {
// We don't have a numeric ID so insert one into the database.
eventTypeNID, err = d.statements.insertEventTypeNID(ctx, txn, eventType)
eventTypeNID, err = d.eventTypes.InsertEventTypeNID(ctx, txn, eventType)
if err == sql.ErrNoRows {
// We raced with another insert so run the select again.
eventTypeNID, err = d.statements.selectEventTypeNID(ctx, txn, eventType)
eventTypeNID, err = d.eventTypes.SelectEventTypeNID(ctx, txn, eventType)
}
}
return
@ -226,92 +255,37 @@ func (d *Database) assignStateKeyNID(
ctx context.Context, txn *sql.Tx, eventStateKey string,
) (eventStateKeyNID types.EventStateKeyNID, err error) {
// Check if we already have a numeric ID in the database.
eventStateKeyNID, err = d.statements.selectEventStateKeyNID(ctx, txn, eventStateKey)
eventStateKeyNID, err = d.eventStateKeys.SelectEventStateKeyNID(ctx, txn, eventStateKey)
if err == sql.ErrNoRows {
// We don't have a numeric ID so insert one into the database.
eventStateKeyNID, err = d.statements.insertEventStateKeyNID(ctx, txn, eventStateKey)
eventStateKeyNID, err = d.eventStateKeys.InsertEventStateKeyNID(ctx, txn, eventStateKey)
if err == sql.ErrNoRows {
// We raced with another insert so run the select again.
eventStateKeyNID, err = d.statements.selectEventStateKeyNID(ctx, txn, eventStateKey)
eventStateKeyNID, err = d.eventStateKeys.SelectEventStateKeyNID(ctx, txn, eventStateKey)
}
}
return
}
// StateEntriesForEventIDs implements input.EventDatabase
func (d *Database) StateEntriesForEventIDs(
ctx context.Context, eventIDs []string,
) (se []types.StateEntry, err error) {
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
se, err = d.statements.bulkSelectStateEventByID(ctx, txn, eventIDs)
return err
})
return
}
// EventTypeNIDs implements state.RoomStateDatabase
func (d *Database) EventTypeNIDs(
ctx context.Context, eventTypes []string,
) (etnids map[string]types.EventTypeNID, err error) {
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
etnids, err = d.statements.bulkSelectEventTypeNID(ctx, txn, eventTypes)
return err
})
return
}
// EventStateKeyNIDs implements state.RoomStateDatabase
func (d *Database) EventStateKeyNIDs(
ctx context.Context, eventStateKeys []string,
) (esknids map[string]types.EventStateKeyNID, err error) {
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
esknids, err = d.statements.bulkSelectEventStateKeyNID(ctx, txn, eventStateKeys)
return err
})
return
}
// EventStateKeys implements query.RoomserverQueryAPIDatabase
func (d *Database) EventStateKeys(
ctx context.Context, eventStateKeyNIDs []types.EventStateKeyNID,
) (out map[types.EventStateKeyNID]string, err error) {
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
out, err = d.statements.bulkSelectEventStateKey(ctx, txn, eventStateKeyNIDs)
return err
})
return
}
// EventNIDs implements query.RoomserverQueryAPIDatabase
func (d *Database) EventNIDs(
ctx context.Context, eventIDs []string,
) (out map[string]types.EventNID, err error) {
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
out, err = d.statements.bulkSelectEventNID(ctx, txn, eventIDs)
return err
})
return
}
// Events implements input.EventDatabase
func (d *Database) Events(
ctx context.Context, eventNIDs []types.EventNID,
) ([]types.Event, error) {
var eventJSONs []eventJSONPair
var eventJSONs []tables.EventJSONPair
var err error
var results []types.Event
eventJSONs, err = d.eventJSON.BulkSelectEventJSON(ctx, eventNIDs)
if err != nil || len(eventJSONs) == 0 {
return nil, nil
}
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
eventJSONs, err = d.statements.bulkSelectEventJSON(ctx, txn, eventNIDs)
if err != nil || len(eventJSONs) == 0 {
return nil
}
results = make([]types.Event, len(eventJSONs))
for i, eventJSON := range eventJSONs {
var roomNID types.RoomNID
var roomVersion gomatrixserverlib.RoomVersion
result := &results[i]
result.EventNID = eventJSON.EventNID
roomNID, err = d.statements.selectRoomNIDForEventNID(ctx, txn, eventJSON.EventNID)
roomNID, err = d.events.SelectRoomNIDForEventNID(ctx, txn, eventJSON.EventNID)
if err != nil {
return err
}
@ -359,27 +333,6 @@ func (d *Database) AddState(
return
}
// SetState implements input.EventDatabase
func (d *Database) SetState(
ctx context.Context, eventNID types.EventNID, stateNID types.StateSnapshotNID,
) error {
e := internal.WithTransaction(d.db, func(txn *sql.Tx) error {
return d.statements.updateEventState(ctx, txn, eventNID, stateNID)
})
return e
}
// StateAtEventIDs implements input.EventDatabase
func (d *Database) StateAtEventIDs(
ctx context.Context, eventIDs []string,
) (se []types.StateAtEvent, err error) {
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
se, err = d.statements.bulkSelectStateAtEventByID(ctx, txn, eventIDs)
return err
})
return
}
// StateBlockNIDs implements state.RoomStateDatabase
func (d *Database) StateBlockNIDs(
ctx context.Context, stateNIDs []types.StateSnapshotNID,
@ -402,28 +355,6 @@ func (d *Database) StateEntries(
return
}
// SnapshotNIDFromEventID implements state.RoomStateDatabase
func (d *Database) SnapshotNIDFromEventID(
ctx context.Context, eventID string,
) (stateNID types.StateSnapshotNID, err error) {
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
_, stateNID, err = d.statements.selectEvent(ctx, txn, eventID)
return err
})
return
}
// EventIDs implements input.RoomEventDatabase
func (d *Database) EventIDs(
ctx context.Context, eventNIDs []types.EventNID,
) (out map[types.EventNID]string, err error) {
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
out, err = d.statements.bulkSelectEventID(ctx, txn, eventNIDs)
return err
})
return
}
// GetLatestEventsForUpdate implements input.EventDatabase
func (d *Database) GetLatestEventsForUpdate(
ctx context.Context, roomNID types.RoomNID,
@ -438,14 +369,14 @@ func (d *Database) GetLatestEventsForUpdate(
txn.Rollback() // nolint: errcheck
return nil, err
}
stateAndRefs, err := d.statements.bulkSelectStateAtEventAndReference(ctx, txn, eventNIDs)
stateAndRefs, err := d.events.BulkSelectStateAtEventAndReference(ctx, txn, eventNIDs)
if err != nil {
txn.Rollback() // nolint: errcheck
return nil, err
}
var lastEventIDSent string
if lastEventNIDSent != 0 {
lastEventIDSent, err = d.statements.selectEventID(ctx, txn, lastEventNIDSent)
lastEventIDSent, err = d.events.SelectEventID(ctx, txn, lastEventNIDSent)
if err != nil {
txn.Rollback() // nolint: errcheck
return nil, err
@ -555,7 +486,7 @@ func (u *roomRecentEventsUpdater) SetLatestEvents(
// HasEventBeenSent implements types.RoomRecentEventsUpdater
func (u *roomRecentEventsUpdater) HasEventBeenSent(eventNID types.EventNID) (res bool, err error) {
err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error {
res, err = u.d.statements.selectEventSentToOutput(u.ctx, txn, eventNID)
res, err = u.d.events.SelectEventSentToOutput(u.ctx, txn, eventNID)
return err
})
return
@ -564,7 +495,7 @@ func (u *roomRecentEventsUpdater) HasEventBeenSent(eventNID types.EventNID) (res
// MarkEventAsSent implements types.RoomRecentEventsUpdater
func (u *roomRecentEventsUpdater) MarkEventAsSent(eventNID types.EventNID) error {
err := internal.WithTransaction(u.d.db, func(txn *sql.Tx) error {
return u.d.statements.updateEventSentToOutput(u.ctx, txn, eventNID)
return u.d.events.UpdateEventSentToOutput(u.ctx, txn, eventNID)
})
return err
}
@ -617,11 +548,11 @@ func (d *Database) LatestEventIDs(
if err != nil {
return err
}
references, err = d.statements.bulkSelectEventReference(ctx, txn, eventNIDs)
references, err = d.events.BulkSelectEventReference(ctx, txn, eventNIDs)
if err != nil {
return err
}
depth, err = d.statements.selectMaxEventDepth(ctx, txn, eventNIDs)
depth, err = d.events.SelectMaxEventDepth(ctx, txn, eventNIDs)
if err != nil {
return err
}

57
roomserver/storage/tables/interface.go Normal file
View file

@ -0,0 +1,57 @@
package tables
import (
"context"
"database/sql"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
)
type EventJSONPair struct {
EventNID types.EventNID
EventJSON []byte
}
type EventJSON interface {
InsertEventJSON(ctx context.Context, tx *sql.Tx, eventNID types.EventNID, eventJSON []byte) error
BulkSelectEventJSON(ctx context.Context, eventNIDs []types.EventNID) ([]EventJSONPair, error)
}
type EventTypes interface {
InsertEventTypeNID(ctx context.Context, tx *sql.Tx, eventType string) (types.EventTypeNID, error)
SelectEventTypeNID(ctx context.Context, tx *sql.Tx, eventType string) (types.EventTypeNID, error)
BulkSelectEventTypeNID(ctx context.Context, eventTypes []string) (map[string]types.EventTypeNID, error)
}
type EventStateKeys interface {
InsertEventStateKeyNID(ctx context.Context, txn *sql.Tx, eventStateKey string) (types.EventStateKeyNID, error)
SelectEventStateKeyNID(ctx context.Context, txn *sql.Tx, eventStateKey string) (types.EventStateKeyNID, error)
BulkSelectEventStateKeyNID(ctx context.Context, eventStateKeys []string) (map[string]types.EventStateKeyNID, error)
BulkSelectEventStateKey(ctx context.Context, eventStateKeyNIDs []types.EventStateKeyNID) (map[types.EventStateKeyNID]string, error)
}
type Events interface {
InsertEvent(c context.Context, txn *sql.Tx, i types.RoomNID, j types.EventTypeNID, k types.EventStateKeyNID, eventID string, referenceSHA256 []byte, authEventNIDs []types.EventNID, depth int64) (types.EventNID, types.StateSnapshotNID, error)
SelectEvent(ctx context.Context, txn *sql.Tx, eventID string) (types.EventNID, types.StateSnapshotNID, error)
// bulkSelectStateEventByID lookups a list of state events by event ID.
// If any of the requested events are missing from the database it returns a types.MissingEventError
BulkSelectStateEventByID(ctx context.Context, eventIDs []string) ([]types.StateEntry, error)
// BulkSelectStateAtEventByID lookups the state at a list of events by event ID.
// If any of the requested events are missing from the database it returns a types.MissingEventError.
// If we do not have the state for any of the requested events it returns a types.MissingEventError.
BulkSelectStateAtEventByID(ctx context.Context, eventIDs []string) ([]types.StateAtEvent, error)
UpdateEventState(ctx context.Context, eventNID types.EventNID, stateNID types.StateSnapshotNID) error
SelectEventSentToOutput(ctx context.Context, txn *sql.Tx, eventNID types.EventNID) (sentToOutput bool, err error)
UpdateEventSentToOutput(ctx context.Context, txn *sql.Tx, eventNID types.EventNID) error
SelectEventID(ctx context.Context, txn *sql.Tx, eventNID types.EventNID) (eventID string, err error)
BulkSelectStateAtEventAndReference(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) ([]types.StateAtEventAndReference, error)
BulkSelectEventReference(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) ([]gomatrixserverlib.EventReference, error)
// BulkSelectEventID returns a map from numeric event ID to string event ID.
BulkSelectEventID(ctx context.Context, eventNIDs []types.EventNID) (map[types.EventNID]string, error)
// BulkSelectEventNIDs returns a map from string event ID to numeric event ID.
// If an event ID is not in the database then it is omitted from the map.
BulkSelectEventNID(ctx context.Context, eventIDs []string) (map[string]types.EventNID, error)
SelectMaxEventDepth(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (int64, error)
SelectRoomNIDForEventNID(ctx context.Context, txn *sql.Tx, eventNID types.EventNID) (roomNID types.RoomNID, err error)
}