mirror of
https://github.com/hoernschen/dendrite.git
synced 2024-12-27 23:48:27 +00:00
roomserver: start refactoring storage layer (#1061)
* Begin refactoring out roomserver/storage/*/storage.go * Convert event_types table
This commit is contained in:
parent
6d50212f29
commit
737c83e0ae
10 changed files with 162 additions and 121 deletions
|
@ -21,6 +21,7 @@ import (
|
||||||
|
|
||||||
"github.com/lib/pq"
|
"github.com/lib/pq"
|
||||||
"github.com/matrix-org/dendrite/internal"
|
"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/dendrite/roomserver/types"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -74,12 +75,13 @@ type eventStateKeyStatements struct {
|
||||||
bulkSelectEventStateKeyStmt *sql.Stmt
|
bulkSelectEventStateKeyStmt *sql.Stmt
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventStateKeyStatements) prepare(db *sql.DB) (err error) {
|
func NewPostgresEventStateKeysTable(db *sql.DB) (tables.EventStateKeys, error) {
|
||||||
_, err = db.Exec(eventStateKeysSchema)
|
s := &eventStateKeyStatements{}
|
||||||
|
_, err := db.Exec(eventStateKeysSchema)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
return nil, err
|
||||||
}
|
}
|
||||||
return statementList{
|
return s, statementList{
|
||||||
{&s.insertEventStateKeyNIDStmt, insertEventStateKeyNIDSQL},
|
{&s.insertEventStateKeyNIDStmt, insertEventStateKeyNIDSQL},
|
||||||
{&s.selectEventStateKeyNIDStmt, selectEventStateKeyNIDSQL},
|
{&s.selectEventStateKeyNIDStmt, selectEventStateKeyNIDSQL},
|
||||||
{&s.bulkSelectEventStateKeyNIDStmt, bulkSelectEventStateKeyNIDSQL},
|
{&s.bulkSelectEventStateKeyNIDStmt, bulkSelectEventStateKeyNIDSQL},
|
||||||
|
@ -87,7 +89,7 @@ func (s *eventStateKeyStatements) prepare(db *sql.DB) (err error) {
|
||||||
}.prepare(db)
|
}.prepare(db)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventStateKeyStatements) insertEventStateKeyNID(
|
func (s *eventStateKeyStatements) InsertEventStateKeyNID(
|
||||||
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
||||||
) (types.EventStateKeyNID, error) {
|
) (types.EventStateKeyNID, error) {
|
||||||
var eventStateKeyNID int64
|
var eventStateKeyNID int64
|
||||||
|
@ -96,7 +98,7 @@ func (s *eventStateKeyStatements) insertEventStateKeyNID(
|
||||||
return types.EventStateKeyNID(eventStateKeyNID), err
|
return types.EventStateKeyNID(eventStateKeyNID), err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventStateKeyStatements) selectEventStateKeyNID(
|
func (s *eventStateKeyStatements) SelectEventStateKeyNID(
|
||||||
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
||||||
) (types.EventStateKeyNID, error) {
|
) (types.EventStateKeyNID, error) {
|
||||||
var eventStateKeyNID int64
|
var eventStateKeyNID int64
|
||||||
|
@ -105,7 +107,7 @@ func (s *eventStateKeyStatements) selectEventStateKeyNID(
|
||||||
return types.EventStateKeyNID(eventStateKeyNID), err
|
return types.EventStateKeyNID(eventStateKeyNID), err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventStateKeyStatements) bulkSelectEventStateKeyNID(
|
func (s *eventStateKeyStatements) BulkSelectEventStateKeyNID(
|
||||||
ctx context.Context, eventStateKeys []string,
|
ctx context.Context, eventStateKeys []string,
|
||||||
) (map[string]types.EventStateKeyNID, error) {
|
) (map[string]types.EventStateKeyNID, error) {
|
||||||
rows, err := s.bulkSelectEventStateKeyNIDStmt.QueryContext(
|
rows, err := s.bulkSelectEventStateKeyNIDStmt.QueryContext(
|
||||||
|
@ -128,7 +130,7 @@ func (s *eventStateKeyStatements) bulkSelectEventStateKeyNID(
|
||||||
return result, rows.Err()
|
return result, rows.Err()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventStateKeyStatements) bulkSelectEventStateKey(
|
func (s *eventStateKeyStatements) BulkSelectEventStateKey(
|
||||||
ctx context.Context, eventStateKeyNIDs []types.EventStateKeyNID,
|
ctx context.Context, eventStateKeyNIDs []types.EventStateKeyNID,
|
||||||
) (map[types.EventStateKeyNID]string, error) {
|
) (map[types.EventStateKeyNID]string, error) {
|
||||||
nIDs := make(pq.Int64Array, len(eventStateKeyNIDs))
|
nIDs := make(pq.Int64Array, len(eventStateKeyNIDs))
|
||||||
|
|
|
@ -22,6 +22,7 @@ import (
|
||||||
"github.com/matrix-org/dendrite/internal"
|
"github.com/matrix-org/dendrite/internal"
|
||||||
|
|
||||||
"github.com/lib/pq"
|
"github.com/lib/pq"
|
||||||
|
"github.com/matrix-org/dendrite/roomserver/storage/tables"
|
||||||
"github.com/matrix-org/dendrite/roomserver/types"
|
"github.com/matrix-org/dendrite/roomserver/types"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -98,36 +99,37 @@ type eventTypeStatements struct {
|
||||||
bulkSelectEventTypeNIDStmt *sql.Stmt
|
bulkSelectEventTypeNIDStmt *sql.Stmt
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventTypeStatements) prepare(db *sql.DB) (err error) {
|
func NewPostgresEventTypesTable(db *sql.DB) (tables.EventTypes, error) {
|
||||||
_, err = db.Exec(eventTypesSchema)
|
s := &eventTypeStatements{}
|
||||||
|
_, err := db.Exec(eventTypesSchema)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
return statementList{
|
return s, statementList{
|
||||||
{&s.insertEventTypeNIDStmt, insertEventTypeNIDSQL},
|
{&s.insertEventTypeNIDStmt, insertEventTypeNIDSQL},
|
||||||
{&s.selectEventTypeNIDStmt, selectEventTypeNIDSQL},
|
{&s.selectEventTypeNIDStmt, selectEventTypeNIDSQL},
|
||||||
{&s.bulkSelectEventTypeNIDStmt, bulkSelectEventTypeNIDSQL},
|
{&s.bulkSelectEventTypeNIDStmt, bulkSelectEventTypeNIDSQL},
|
||||||
}.prepare(db)
|
}.prepare(db)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventTypeStatements) insertEventTypeNID(
|
func (s *eventTypeStatements) InsertEventTypeNID(
|
||||||
ctx context.Context, eventType string,
|
ctx context.Context, txn *sql.Tx, eventType string,
|
||||||
) (types.EventTypeNID, error) {
|
) (types.EventTypeNID, error) {
|
||||||
var eventTypeNID int64
|
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
|
return types.EventTypeNID(eventTypeNID), err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventTypeStatements) selectEventTypeNID(
|
func (s *eventTypeStatements) SelectEventTypeNID(
|
||||||
ctx context.Context, eventType string,
|
ctx context.Context, txn *sql.Tx, eventType string,
|
||||||
) (types.EventTypeNID, error) {
|
) (types.EventTypeNID, error) {
|
||||||
var eventTypeNID int64
|
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
|
return types.EventTypeNID(eventTypeNID), err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventTypeStatements) bulkSelectEventTypeNID(
|
func (s *eventTypeStatements) BulkSelectEventTypeNID(
|
||||||
ctx context.Context, eventTypes []string,
|
ctx context.Context, eventTypes []string,
|
||||||
) (map[string]types.EventTypeNID, error) {
|
) (map[string]types.EventTypeNID, error) {
|
||||||
rows, err := s.bulkSelectEventTypeNIDStmt.QueryContext(ctx, pq.StringArray(eventTypes))
|
rows, err := s.bulkSelectEventTypeNIDStmt.QueryContext(ctx, pq.StringArray(eventTypes))
|
||||||
|
|
|
@ -38,8 +38,6 @@ func (s *statements) prepare(db *sql.DB) error {
|
||||||
var err error
|
var err error
|
||||||
|
|
||||||
for _, prepare := range []func(db *sql.DB) error{
|
for _, prepare := range []func(db *sql.DB) error{
|
||||||
s.eventTypeStatements.prepare,
|
|
||||||
s.eventStateKeyStatements.prepare,
|
|
||||||
s.roomStatements.prepare,
|
s.roomStatements.prepare,
|
||||||
s.eventStatements.prepare,
|
s.eventStatements.prepare,
|
||||||
s.eventJSONStatements.prepare,
|
s.eventJSONStatements.prepare,
|
||||||
|
|
|
@ -26,13 +26,18 @@ import (
|
||||||
// Import the postgres database driver.
|
// Import the postgres database driver.
|
||||||
_ "github.com/lib/pq"
|
_ "github.com/lib/pq"
|
||||||
"github.com/matrix-org/dendrite/roomserver/api"
|
"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/dendrite/roomserver/types"
|
||||||
"github.com/matrix-org/gomatrixserverlib"
|
"github.com/matrix-org/gomatrixserverlib"
|
||||||
)
|
)
|
||||||
|
|
||||||
// A Database is used to store room events and stream offsets.
|
// A Database is used to store room events and stream offsets.
|
||||||
type Database struct {
|
type Database struct {
|
||||||
|
shared.Database
|
||||||
statements statements
|
statements statements
|
||||||
|
eventTypes tables.EventTypes
|
||||||
|
eventStateKeys tables.EventStateKeys
|
||||||
db *sql.DB
|
db *sql.DB
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -46,6 +51,18 @@ func Open(dataSourceName string, dbProperties internal.DbProperties) (*Database,
|
||||||
if err = d.statements.prepare(d.db); err != nil {
|
if err = d.statements.prepare(d.db); err != nil {
|
||||||
return nil, err
|
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.Database = shared.Database{
|
||||||
|
EventTypesTable: d.eventTypes,
|
||||||
|
EventStateKeysTable: d.eventStateKeys,
|
||||||
|
}
|
||||||
return &d, nil
|
return &d, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -180,17 +197,20 @@ func (d *Database) assignRoomNID(
|
||||||
|
|
||||||
func (d *Database) assignEventTypeNID(
|
func (d *Database) assignEventTypeNID(
|
||||||
ctx context.Context, eventType string,
|
ctx context.Context, eventType string,
|
||||||
) (types.EventTypeNID, error) {
|
) (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.
|
// Check if we already have a numeric ID in the database.
|
||||||
eventTypeNID, err := d.statements.selectEventTypeNID(ctx, eventType)
|
eventTypeNID, err = d.eventTypes.SelectEventTypeNID(ctx, txn, eventType)
|
||||||
if err == sql.ErrNoRows {
|
if err == sql.ErrNoRows {
|
||||||
// We don't have a numeric ID so insert one into the database.
|
// We don't have a numeric ID so insert one into the database.
|
||||||
eventTypeNID, err = d.statements.insertEventTypeNID(ctx, eventType)
|
eventTypeNID, err = d.eventTypes.InsertEventTypeNID(ctx, txn, eventType)
|
||||||
if err == sql.ErrNoRows {
|
if err == sql.ErrNoRows {
|
||||||
// We raced with another insert so run the select again.
|
// We raced with another insert so run the select again.
|
||||||
eventTypeNID, err = d.statements.selectEventTypeNID(ctx, eventType)
|
eventTypeNID, err = d.eventTypes.SelectEventTypeNID(ctx, txn, eventType)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
return err
|
||||||
|
})
|
||||||
return eventTypeNID, err
|
return eventTypeNID, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -198,13 +218,13 @@ func (d *Database) assignStateKeyNID(
|
||||||
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
||||||
) (types.EventStateKeyNID, error) {
|
) (types.EventStateKeyNID, error) {
|
||||||
// Check if we already have a numeric ID in the database.
|
// 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 {
|
if err == sql.ErrNoRows {
|
||||||
// We don't have a numeric ID so insert one into the database.
|
// 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 {
|
if err == sql.ErrNoRows {
|
||||||
// We raced with another insert so run the select again.
|
// 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
|
return eventStateKeyNID, err
|
||||||
|
@ -217,27 +237,6 @@ func (d *Database) StateEntriesForEventIDs(
|
||||||
return d.statements.bulkSelectStateEventByID(ctx, eventIDs)
|
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
|
// EventNIDs implements query.RoomserverQueryAPIDatabase
|
||||||
func (d *Database) EventNIDs(
|
func (d *Database) EventNIDs(
|
||||||
ctx context.Context, eventIDs []string,
|
ctx context.Context, eventIDs []string,
|
||||||
|
|
34
roomserver/storage/shared/storage.go
Normal file
34
roomserver/storage/shared/storage.go
Normal file
|
@ -0,0 +1,34 @@
|
||||||
|
package shared
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
|
||||||
|
"github.com/matrix-org/dendrite/roomserver/storage/tables"
|
||||||
|
"github.com/matrix-org/dendrite/roomserver/types"
|
||||||
|
)
|
||||||
|
|
||||||
|
type Database struct {
|
||||||
|
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)
|
||||||
|
}
|
|
@ -21,6 +21,7 @@ import (
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"github.com/matrix-org/dendrite/internal"
|
"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/dendrite/roomserver/types"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -67,13 +68,14 @@ type eventStateKeyStatements struct {
|
||||||
bulkSelectEventStateKeyStmt *sql.Stmt
|
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
|
s.db = db
|
||||||
_, err = db.Exec(eventStateKeysSchema)
|
_, err := db.Exec(eventStateKeysSchema)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
return nil, err
|
||||||
}
|
}
|
||||||
return statementList{
|
return s, statementList{
|
||||||
{&s.insertEventStateKeyNIDStmt, insertEventStateKeyNIDSQL},
|
{&s.insertEventStateKeyNIDStmt, insertEventStateKeyNIDSQL},
|
||||||
{&s.selectEventStateKeyNIDStmt, selectEventStateKeyNIDSQL},
|
{&s.selectEventStateKeyNIDStmt, selectEventStateKeyNIDSQL},
|
||||||
{&s.bulkSelectEventStateKeyNIDStmt, bulkSelectEventStateKeyNIDSQL},
|
{&s.bulkSelectEventStateKeyNIDStmt, bulkSelectEventStateKeyNIDSQL},
|
||||||
|
@ -81,7 +83,7 @@ func (s *eventStateKeyStatements) prepare(db *sql.DB) (err error) {
|
||||||
}.prepare(db)
|
}.prepare(db)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventStateKeyStatements) insertEventStateKeyNID(
|
func (s *eventStateKeyStatements) InsertEventStateKeyNID(
|
||||||
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
||||||
) (types.EventStateKeyNID, error) {
|
) (types.EventStateKeyNID, error) {
|
||||||
var eventStateKeyNID int64
|
var eventStateKeyNID int64
|
||||||
|
@ -94,7 +96,7 @@ func (s *eventStateKeyStatements) insertEventStateKeyNID(
|
||||||
return types.EventStateKeyNID(eventStateKeyNID), err
|
return types.EventStateKeyNID(eventStateKeyNID), err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventStateKeyStatements) selectEventStateKeyNID(
|
func (s *eventStateKeyStatements) SelectEventStateKeyNID(
|
||||||
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
||||||
) (types.EventStateKeyNID, error) {
|
) (types.EventStateKeyNID, error) {
|
||||||
var eventStateKeyNID int64
|
var eventStateKeyNID int64
|
||||||
|
@ -103,8 +105,8 @@ func (s *eventStateKeyStatements) selectEventStateKeyNID(
|
||||||
return types.EventStateKeyNID(eventStateKeyNID), err
|
return types.EventStateKeyNID(eventStateKeyNID), err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventStateKeyStatements) bulkSelectEventStateKeyNID(
|
func (s *eventStateKeyStatements) BulkSelectEventStateKeyNID(
|
||||||
ctx context.Context, txn *sql.Tx, eventStateKeys []string,
|
ctx context.Context, eventStateKeys []string,
|
||||||
) (map[string]types.EventStateKeyNID, error) {
|
) (map[string]types.EventStateKeyNID, error) {
|
||||||
iEventStateKeys := make([]interface{}, len(eventStateKeys))
|
iEventStateKeys := make([]interface{}, len(eventStateKeys))
|
||||||
for k, v := range eventStateKeys {
|
for k, v := range eventStateKeys {
|
||||||
|
@ -112,7 +114,7 @@ func (s *eventStateKeyStatements) bulkSelectEventStateKeyNID(
|
||||||
}
|
}
|
||||||
selectOrig := strings.Replace(bulkSelectEventStateKeySQL, "($1)", internal.QueryVariadic(len(eventStateKeys)), 1)
|
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 {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -129,8 +131,8 @@ func (s *eventStateKeyStatements) bulkSelectEventStateKeyNID(
|
||||||
return result, nil
|
return result, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventStateKeyStatements) bulkSelectEventStateKey(
|
func (s *eventStateKeyStatements) BulkSelectEventStateKey(
|
||||||
ctx context.Context, txn *sql.Tx, eventStateKeyNIDs []types.EventStateKeyNID,
|
ctx context.Context, eventStateKeyNIDs []types.EventStateKeyNID,
|
||||||
) (map[types.EventStateKeyNID]string, error) {
|
) (map[types.EventStateKeyNID]string, error) {
|
||||||
iEventStateKeyNIDs := make([]interface{}, len(eventStateKeyNIDs))
|
iEventStateKeyNIDs := make([]interface{}, len(eventStateKeyNIDs))
|
||||||
for k, v := range eventStateKeyNIDs {
|
for k, v := range eventStateKeyNIDs {
|
||||||
|
@ -138,7 +140,7 @@ func (s *eventStateKeyStatements) bulkSelectEventStateKey(
|
||||||
}
|
}
|
||||||
selectOrig := strings.Replace(bulkSelectEventStateKeyNIDSQL, "($1)", internal.QueryVariadic(len(eventStateKeyNIDs)), 1)
|
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 {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
|
@ -21,6 +21,7 @@ import (
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"github.com/matrix-org/dendrite/internal"
|
"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/dendrite/roomserver/types"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -81,14 +82,15 @@ type eventTypeStatements struct {
|
||||||
bulkSelectEventTypeNIDStmt *sql.Stmt
|
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
|
s.db = db
|
||||||
_, err = db.Exec(eventTypesSchema)
|
_, err := db.Exec(eventTypesSchema)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
return statementList{
|
return s, statementList{
|
||||||
{&s.insertEventTypeNIDStmt, insertEventTypeNIDSQL},
|
{&s.insertEventTypeNIDStmt, insertEventTypeNIDSQL},
|
||||||
{&s.insertEventTypeNIDResultStmt, insertEventTypeNIDResultSQL},
|
{&s.insertEventTypeNIDResultStmt, insertEventTypeNIDResultSQL},
|
||||||
{&s.selectEventTypeNIDStmt, selectEventTypeNIDSQL},
|
{&s.selectEventTypeNIDStmt, selectEventTypeNIDSQL},
|
||||||
|
@ -96,7 +98,7 @@ func (s *eventTypeStatements) prepare(db *sql.DB) (err error) {
|
||||||
}.prepare(db)
|
}.prepare(db)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventTypeStatements) insertEventTypeNID(
|
func (s *eventTypeStatements) InsertEventTypeNID(
|
||||||
ctx context.Context, tx *sql.Tx, eventType string,
|
ctx context.Context, tx *sql.Tx, eventType string,
|
||||||
) (types.EventTypeNID, error) {
|
) (types.EventTypeNID, error) {
|
||||||
var eventTypeNID int64
|
var eventTypeNID int64
|
||||||
|
@ -109,7 +111,7 @@ func (s *eventTypeStatements) insertEventTypeNID(
|
||||||
return types.EventTypeNID(eventTypeNID), err
|
return types.EventTypeNID(eventTypeNID), err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventTypeStatements) selectEventTypeNID(
|
func (s *eventTypeStatements) SelectEventTypeNID(
|
||||||
ctx context.Context, tx *sql.Tx, eventType string,
|
ctx context.Context, tx *sql.Tx, eventType string,
|
||||||
) (types.EventTypeNID, error) {
|
) (types.EventTypeNID, error) {
|
||||||
var eventTypeNID int64
|
var eventTypeNID int64
|
||||||
|
@ -118,8 +120,8 @@ func (s *eventTypeStatements) selectEventTypeNID(
|
||||||
return types.EventTypeNID(eventTypeNID), err
|
return types.EventTypeNID(eventTypeNID), err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *eventTypeStatements) bulkSelectEventTypeNID(
|
func (s *eventTypeStatements) BulkSelectEventTypeNID(
|
||||||
ctx context.Context, tx *sql.Tx, eventTypes []string,
|
ctx context.Context, eventTypes []string,
|
||||||
) (map[string]types.EventTypeNID, error) {
|
) (map[string]types.EventTypeNID, error) {
|
||||||
///////////////
|
///////////////
|
||||||
iEventTypes := make([]interface{}, len(eventTypes))
|
iEventTypes := make([]interface{}, len(eventTypes))
|
||||||
|
@ -133,8 +135,7 @@ func (s *eventTypeStatements) bulkSelectEventTypeNID(
|
||||||
}
|
}
|
||||||
///////////////
|
///////////////
|
||||||
|
|
||||||
selectStmt := internal.TxStmt(tx, selectPrep)
|
rows, err := selectPrep.QueryContext(ctx, iEventTypes...)
|
||||||
rows, err := selectStmt.QueryContext(ctx, iEventTypes...)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
|
@ -38,8 +38,6 @@ func (s *statements) prepare(db *sql.DB) error {
|
||||||
var err error
|
var err error
|
||||||
|
|
||||||
for _, prepare := range []func(db *sql.DB) error{
|
for _, prepare := range []func(db *sql.DB) error{
|
||||||
s.eventTypeStatements.prepare,
|
|
||||||
s.eventStateKeyStatements.prepare,
|
|
||||||
s.roomStatements.prepare,
|
s.roomStatements.prepare,
|
||||||
s.eventStatements.prepare,
|
s.eventStatements.prepare,
|
||||||
s.eventJSONStatements.prepare,
|
s.eventJSONStatements.prepare,
|
||||||
|
|
|
@ -26,6 +26,8 @@ import (
|
||||||
|
|
||||||
"github.com/matrix-org/dendrite/internal"
|
"github.com/matrix-org/dendrite/internal"
|
||||||
"github.com/matrix-org/dendrite/roomserver/api"
|
"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/dendrite/roomserver/types"
|
||||||
"github.com/matrix-org/gomatrixserverlib"
|
"github.com/matrix-org/gomatrixserverlib"
|
||||||
_ "github.com/mattn/go-sqlite3"
|
_ "github.com/mattn/go-sqlite3"
|
||||||
|
@ -33,7 +35,10 @@ import (
|
||||||
|
|
||||||
// A Database is used to store room events and stream offsets.
|
// A Database is used to store room events and stream offsets.
|
||||||
type Database struct {
|
type Database struct {
|
||||||
|
shared.Database
|
||||||
statements statements
|
statements statements
|
||||||
|
eventTypes tables.EventTypes
|
||||||
|
eventStateKeys tables.EventStateKeys
|
||||||
db *sql.DB
|
db *sql.DB
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -66,6 +71,18 @@ func Open(dataSourceName string) (*Database, error) {
|
||||||
if err = d.statements.prepare(d.db); err != nil {
|
if err = d.statements.prepare(d.db); err != nil {
|
||||||
return nil, err
|
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.Database = shared.Database{
|
||||||
|
EventTypesTable: d.eventTypes,
|
||||||
|
EventStateKeysTable: d.eventStateKeys,
|
||||||
|
}
|
||||||
return &d, nil
|
return &d, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -210,13 +227,13 @@ func (d *Database) assignEventTypeNID(
|
||||||
ctx context.Context, txn *sql.Tx, eventType string,
|
ctx context.Context, txn *sql.Tx, eventType string,
|
||||||
) (eventTypeNID types.EventTypeNID, err error) {
|
) (eventTypeNID types.EventTypeNID, err error) {
|
||||||
// Check if we already have a numeric ID in the database.
|
// 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 {
|
if err == sql.ErrNoRows {
|
||||||
// We don't have a numeric ID so insert one into the database.
|
// 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 {
|
if err == sql.ErrNoRows {
|
||||||
// We raced with another insert so run the select again.
|
// 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
|
return
|
||||||
|
@ -226,13 +243,13 @@ func (d *Database) assignStateKeyNID(
|
||||||
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
||||||
) (eventStateKeyNID types.EventStateKeyNID, err error) {
|
) (eventStateKeyNID types.EventStateKeyNID, err error) {
|
||||||
// Check if we already have a numeric ID in the database.
|
// 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 {
|
if err == sql.ErrNoRows {
|
||||||
// We don't have a numeric ID so insert one into the database.
|
// 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 {
|
if err == sql.ErrNoRows {
|
||||||
// We raced with another insert so run the select again.
|
// 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
|
return
|
||||||
|
@ -249,39 +266,6 @@ func (d *Database) StateEntriesForEventIDs(
|
||||||
return
|
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
|
// EventNIDs implements query.RoomserverQueryAPIDatabase
|
||||||
func (d *Database) EventNIDs(
|
func (d *Database) EventNIDs(
|
||||||
ctx context.Context, eventIDs []string,
|
ctx context.Context, eventIDs []string,
|
||||||
|
|
21
roomserver/storage/tables/interface.go
Normal file
21
roomserver/storage/tables/interface.go
Normal file
|
@ -0,0 +1,21 @@
|
||||||
|
package tables
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"database/sql"
|
||||||
|
|
||||||
|
"github.com/matrix-org/dendrite/roomserver/types"
|
||||||
|
)
|
||||||
|
|
||||||
|
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)
|
||||||
|
}
|
Loading…
Reference in a new issue