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

Begin refactoring out roomserver/storage/*/storage.go

Browse source
This commit is contained in:
Kegan Dougal 2020-05-26 12:45:26 +01:00
parent 492af0f2ec
commit fe70da53ef
8 changed files with 95 additions and 69 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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))

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

@ -39,7 +39,6 @@ func (s *statements) prepare(db *sql.DB) error {
for _, prepare := range []func(db *sql.DB) error{
s.eventTypeStatements.prepare,
s.eventStateKeyStatements.prepare,
s.roomStatements.prepare,
s.eventStatements.prepare,
s.eventJSONStatements.prepare,

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

@ -26,13 +26,17 @@ 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 {
shared.Database
statements statements
eventStateKeys tables.EventStateKeys
db *sql.DB
}
@ -46,6 +50,13 @@ 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.Database = shared.Database{
EventStateKeysTable: d.eventStateKeys,
}
return &d, nil
}
@ -198,13 +209,13 @@ 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
@ -224,20 +235,6 @@ func (d *Database) EventTypeNIDs(
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,

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

@ -0,0 +1,26 @@
package shared
import (
"context"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
)
type Database struct {
EventStateKeysTable tables.EventStateKeys
}
// 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)
}

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
}

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

@ -39,7 +39,6 @@ func (s *statements) prepare(db *sql.DB) error {
for _, prepare := range []func(db *sql.DB) error{
s.eventTypeStatements.prepare,
s.eventStateKeyStatements.prepare,
s.roomStatements.prepare,
s.eventStatements.prepare,
s.eventJSONStatements.prepare,

38
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,7 +35,9 @@ import (
// A Database is used to store room events and stream offsets.
type Database struct {
shared.Database
statements statements
eventStateKeys tables.EventStateKeys
db *sql.DB
}
@ -66,7 +70,11 @@ func Open(dataSourceName string) (*Database, error) {
if err = d.statements.prepare(d.db); err != nil {
return nil, err
}
return &d, nil
d.eventStateKeys, err = NewSqliteEventStateKeysTable(d.db)
d.Database = shared.Database{
EventStateKeysTable: d.eventStateKeys,
}
return &d, err
}
// StoreEvent implements input.EventDatabase
@ -226,13 +234,13 @@ 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
@ -260,28 +268,6 @@ func (d *Database) EventTypeNIDs(
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,

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

@ -0,0 +1,15 @@
package tables
import (
"context"
"database/sql"
"github.com/matrix-org/dendrite/roomserver/types"
)
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)
}