mirror of
https://github.com/matrix-org/dendrite.git
synced 2025-12-19 21:03:09 -06:00
Convert rooms table and factor out lots of functions
This commit is contained in:
parent
0cbb13b934
commit
7b331bcf37
|
|
@ -426,10 +426,9 @@ func (s *eventStatements) SelectMaxEventDepth(ctx context.Context, txn *sql.Tx,
|
|||
}
|
||||
|
||||
func (s *eventStatements) SelectRoomNIDForEventNID(
|
||||
ctx context.Context, txn *sql.Tx, eventNID types.EventNID,
|
||||
ctx context.Context, eventNID types.EventNID,
|
||||
) (roomNID types.RoomNID, err error) {
|
||||
selectStmt := internal.TxStmt(txn, s.selectRoomNIDForEventNIDStmt)
|
||||
err = selectStmt.QueryRowContext(ctx, int64(eventNID)).Scan(&roomNID)
|
||||
err = s.selectRoomNIDForEventNIDStmt.QueryRowContext(ctx, int64(eventNID)).Scan(&roomNID)
|
||||
return
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -186,11 +186,10 @@ func (s *roomStatements) SelectRoomVersionForRoomID(
|
|||
}
|
||||
|
||||
func (s *roomStatements) SelectRoomVersionForRoomNID(
|
||||
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
|
||||
ctx context.Context, roomNID types.RoomNID,
|
||||
) (gomatrixserverlib.RoomVersion, error) {
|
||||
var roomVersion gomatrixserverlib.RoomVersion
|
||||
stmt := internal.TxStmt(txn, s.selectRoomVersionForRoomNIDStmt)
|
||||
err := stmt.QueryRowContext(ctx, roomNID).Scan(&roomVersion)
|
||||
err := s.selectRoomVersionForRoomNIDStmt.QueryRowContext(ctx, roomNID).Scan(&roomVersion)
|
||||
if err == sql.ErrNoRows {
|
||||
return roomVersion, errors.New("room not found")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,14 +18,12 @@ package postgres
|
|||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/internal/sqlutil"
|
||||
|
||||
// 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"
|
||||
|
|
@ -91,119 +89,6 @@ func Open(dataSourceName string, dbProperties internal.DbProperties) (*Database,
|
|||
return &d, nil
|
||||
}
|
||||
|
||||
// StoreEvent implements input.EventDatabase
|
||||
func (d *Database) StoreEvent(
|
||||
ctx context.Context, event gomatrixserverlib.Event,
|
||||
txnAndSessionID *api.TransactionID, authEventNIDs []types.EventNID,
|
||||
) (types.RoomNID, types.StateAtEvent, error) {
|
||||
var (
|
||||
roomNID types.RoomNID
|
||||
eventTypeNID types.EventTypeNID
|
||||
eventStateKeyNID types.EventStateKeyNID
|
||||
eventNID types.EventNID
|
||||
stateNID types.StateSnapshotNID
|
||||
err error
|
||||
)
|
||||
|
||||
if txnAndSessionID != nil {
|
||||
if err = d.transactions.InsertTransaction(
|
||||
ctx, nil, txnAndSessionID.TransactionID,
|
||||
txnAndSessionID.SessionID, event.Sender(), event.EventID(),
|
||||
); err != nil {
|
||||
return 0, types.StateAtEvent{}, err
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: Here we should aim to have two different code paths for new rooms
|
||||
// vs existing ones.
|
||||
|
||||
// Get the default room version. If the client doesn't supply a room_version
|
||||
// then we will use our configured default to create the room.
|
||||
// https://matrix.org/docs/spec/client_server/r0.6.0#post-matrix-client-r0-createroom
|
||||
// Note that the below logic depends on the m.room.create event being the
|
||||
// first event that is persisted to the database when creating or joining a
|
||||
// room.
|
||||
var roomVersion gomatrixserverlib.RoomVersion
|
||||
if roomVersion, err = extractRoomVersionFromCreateEvent(event); err != nil {
|
||||
return 0, types.StateAtEvent{}, err
|
||||
}
|
||||
|
||||
if roomNID, err = d.assignRoomNID(ctx, nil, event.RoomID(), roomVersion); err != nil {
|
||||
return 0, types.StateAtEvent{}, err
|
||||
}
|
||||
|
||||
if eventTypeNID, err = d.assignEventTypeNID(ctx, event.Type()); err != nil {
|
||||
return 0, types.StateAtEvent{}, err
|
||||
}
|
||||
|
||||
eventStateKey := event.StateKey()
|
||||
// Assigned a numeric ID for the state_key if there is one present.
|
||||
// Otherwise set the numeric ID for the state_key to 0.
|
||||
if eventStateKey != nil {
|
||||
if eventStateKeyNID, err = d.assignStateKeyNID(ctx, nil, *eventStateKey); err != nil {
|
||||
return 0, types.StateAtEvent{}, err
|
||||
}
|
||||
}
|
||||
|
||||
if eventNID, stateNID, err = d.events.InsertEvent(
|
||||
ctx,
|
||||
nil,
|
||||
roomNID,
|
||||
eventTypeNID,
|
||||
eventStateKeyNID,
|
||||
event.EventID(),
|
||||
event.EventReference().EventSHA256,
|
||||
authEventNIDs,
|
||||
event.Depth(),
|
||||
); err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
// We've already inserted the event so select the numeric event ID
|
||||
eventNID, stateNID, err = d.events.SelectEvent(ctx, nil, event.EventID())
|
||||
}
|
||||
if err != nil {
|
||||
return 0, types.StateAtEvent{}, err
|
||||
}
|
||||
}
|
||||
|
||||
if err = d.eventJSON.InsertEventJSON(ctx, nil, eventNID, event.JSON()); err != nil {
|
||||
return 0, types.StateAtEvent{}, err
|
||||
}
|
||||
|
||||
return roomNID, types.StateAtEvent{
|
||||
BeforeStateSnapshotNID: stateNID,
|
||||
StateEntry: types.StateEntry{
|
||||
StateKeyTuple: types.StateKeyTuple{
|
||||
EventTypeNID: eventTypeNID,
|
||||
EventStateKeyNID: eventStateKeyNID,
|
||||
},
|
||||
EventNID: eventNID,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func extractRoomVersionFromCreateEvent(event gomatrixserverlib.Event) (
|
||||
gomatrixserverlib.RoomVersion, error,
|
||||
) {
|
||||
var err error
|
||||
var roomVersion gomatrixserverlib.RoomVersion
|
||||
// Look for m.room.create events.
|
||||
if event.Type() != gomatrixserverlib.MRoomCreate {
|
||||
return gomatrixserverlib.RoomVersion(""), nil
|
||||
}
|
||||
roomVersion = gomatrixserverlib.RoomVersionV1
|
||||
var createContent gomatrixserverlib.CreateContent
|
||||
// The m.room.create event contains an optional "room_version" key in
|
||||
// the event content, so we need to unmarshal that first.
|
||||
if err = json.Unmarshal(event.Content(), &createContent); err != nil {
|
||||
return gomatrixserverlib.RoomVersion(""), err
|
||||
}
|
||||
// A room version was specified in the event content?
|
||||
if createContent.RoomVersion != nil {
|
||||
roomVersion = gomatrixserverlib.RoomVersion(*createContent.RoomVersion)
|
||||
}
|
||||
return roomVersion, err
|
||||
}
|
||||
|
||||
func (d *Database) assignRoomNID(
|
||||
ctx context.Context, txn *sql.Tx,
|
||||
roomID string, roomVersion gomatrixserverlib.RoomVersion,
|
||||
|
|
@ -221,25 +106,6 @@ func (d *Database) assignRoomNID(
|
|||
return roomNID, err
|
||||
}
|
||||
|
||||
func (d *Database) assignEventTypeNID(
|
||||
ctx context.Context, eventType string,
|
||||
) (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 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
|
||||
}
|
||||
|
||||
func (d *Database) assignStateKeyNID(
|
||||
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
||||
) (types.EventStateKeyNID, error) {
|
||||
|
|
@ -256,38 +122,6 @@ func (d *Database) assignStateKeyNID(
|
|||
return eventStateKeyNID, err
|
||||
}
|
||||
|
||||
// Events implements input.EventDatabase
|
||||
func (d *Database) Events(
|
||||
ctx context.Context, eventNIDs []types.EventNID,
|
||||
) ([]types.Event, error) {
|
||||
eventJSONs, err := d.eventJSON.BulkSelectEventJSON(ctx, eventNIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
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.events.SelectRoomNIDForEventNID(ctx, nil, eventJSON.EventNID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
roomVersion, err = d.rooms.SelectRoomVersionForRoomNID(ctx, nil, roomNID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result.Event, err = gomatrixserverlib.NewEventFromTrustedJSON(
|
||||
eventJSON.EventJSON, false, roomVersion,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// AddState implements input.EventDatabase
|
||||
func (d *Database) AddState(
|
||||
ctx context.Context,
|
||||
|
|
@ -685,21 +519,6 @@ func (d *Database) GetMembershipEventNIDsForRoom(
|
|||
return d.statements.selectMembershipsFromRoom(ctx, roomNID, localOnly)
|
||||
}
|
||||
|
||||
// EventsFromIDs implements query.RoomserverQueryAPIEventDB
|
||||
func (d *Database) EventsFromIDs(ctx context.Context, eventIDs []string) ([]types.Event, error) {
|
||||
nidMap, err := d.EventNIDs(ctx, eventIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var nids []types.EventNID
|
||||
for _, nid := range nidMap {
|
||||
nids = append(nids, nid)
|
||||
}
|
||||
|
||||
return d.Events(ctx, nids)
|
||||
}
|
||||
|
||||
type transaction struct {
|
||||
ctx context.Context
|
||||
txn *sql.Tx
|
||||
|
|
|
|||
|
|
@ -3,8 +3,10 @@ package shared
|
|||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/roomserver/storage/tables"
|
||||
"github.com/matrix-org/dendrite/roomserver/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
|
@ -84,6 +86,21 @@ func (d *Database) EventIDs(
|
|||
return d.EventsTable.BulkSelectEventID(ctx, eventNIDs)
|
||||
}
|
||||
|
||||
// EventsFromIDs implements query.RoomserverQueryAPIEventDB
|
||||
func (d *Database) EventsFromIDs(ctx context.Context, eventIDs []string) ([]types.Event, error) {
|
||||
nidMap, err := d.EventNIDs(ctx, eventIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var nids []types.EventNID
|
||||
for _, nid := range nidMap {
|
||||
nids = append(nids, nid)
|
||||
}
|
||||
|
||||
return d.Events(ctx, nids)
|
||||
}
|
||||
|
||||
// RoomNID implements query.RoomserverQueryAPIDB
|
||||
func (d *Database) RoomNID(ctx context.Context, roomID string) (types.RoomNID, error) {
|
||||
roomNID, err := d.RoomsTable.SelectRoomNID(ctx, nil, roomID)
|
||||
|
|
@ -145,10 +162,42 @@ func (d *Database) GetRoomVersionForRoomNID(
|
|||
ctx context.Context, roomNID types.RoomNID,
|
||||
) (gomatrixserverlib.RoomVersion, error) {
|
||||
return d.RoomsTable.SelectRoomVersionForRoomNID(
|
||||
ctx, nil, roomNID,
|
||||
ctx, roomNID,
|
||||
)
|
||||
}
|
||||
|
||||
// Events implements input.EventDatabase
|
||||
func (d *Database) Events(
|
||||
ctx context.Context, eventNIDs []types.EventNID,
|
||||
) ([]types.Event, error) {
|
||||
eventJSONs, err := d.EventJSONTable.BulkSelectEventJSON(ctx, eventNIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
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.EventsTable.SelectRoomNIDForEventNID(ctx, eventJSON.EventNID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
roomVersion, err = d.RoomsTable.SelectRoomVersionForRoomNID(ctx, roomNID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result.Event, err = gomatrixserverlib.NewEventFromTrustedJSON(
|
||||
eventJSON.EventJSON, false, roomVersion,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// GetTransactionEventID implements input.EventDatabase
|
||||
func (d *Database) GetTransactionEventID(
|
||||
ctx context.Context, transactionID string,
|
||||
|
|
@ -160,3 +209,172 @@ func (d *Database) GetTransactionEventID(
|
|||
}
|
||||
return eventID, err
|
||||
}
|
||||
|
||||
// StoreEvent implements input.EventDatabase
|
||||
func (d *Database) StoreEvent(
|
||||
ctx context.Context, event gomatrixserverlib.Event,
|
||||
txnAndSessionID *api.TransactionID, authEventNIDs []types.EventNID,
|
||||
) (types.RoomNID, types.StateAtEvent, error) {
|
||||
var (
|
||||
roomNID types.RoomNID
|
||||
eventTypeNID types.EventTypeNID
|
||||
eventStateKeyNID types.EventStateKeyNID
|
||||
eventNID types.EventNID
|
||||
stateNID types.StateSnapshotNID
|
||||
err error
|
||||
)
|
||||
|
||||
err = internal.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
if txnAndSessionID != nil {
|
||||
if err = d.TransactionsTable.InsertTransaction(
|
||||
ctx, txn, txnAndSessionID.TransactionID,
|
||||
txnAndSessionID.SessionID, event.Sender(), event.EventID(),
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: Here we should aim to have two different code paths for new rooms
|
||||
// vs existing ones.
|
||||
|
||||
// Get the default room version. If the client doesn't supply a room_version
|
||||
// then we will use our configured default to create the room.
|
||||
// https://matrix.org/docs/spec/client_server/r0.6.0#post-matrix-client-r0-createroom
|
||||
// Note that the below logic depends on the m.room.create event being the
|
||||
// first event that is persisted to the database when creating or joining a
|
||||
// room.
|
||||
var roomVersion gomatrixserverlib.RoomVersion
|
||||
if roomVersion, err = extractRoomVersionFromCreateEvent(event); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if roomNID, err = d.assignRoomNID(ctx, txn, event.RoomID(), roomVersion); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if eventTypeNID, err = d.assignEventTypeNID(ctx, txn, event.Type()); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
eventStateKey := event.StateKey()
|
||||
// Assigned a numeric ID for the state_key if there is one present.
|
||||
// Otherwise set the numeric ID for the state_key to 0.
|
||||
if eventStateKey != nil {
|
||||
if eventStateKeyNID, err = d.assignStateKeyNID(ctx, txn, *eventStateKey); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if eventNID, stateNID, err = d.EventsTable.InsertEvent(
|
||||
ctx,
|
||||
txn,
|
||||
roomNID,
|
||||
eventTypeNID,
|
||||
eventStateKeyNID,
|
||||
event.EventID(),
|
||||
event.EventReference().EventSHA256,
|
||||
authEventNIDs,
|
||||
event.Depth(),
|
||||
); err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
// We've already inserted the event so select the numeric event ID
|
||||
eventNID, stateNID, err = d.EventsTable.SelectEvent(ctx, txn, event.EventID())
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if err = d.EventJSONTable.InsertEventJSON(ctx, txn, eventNID, event.JSON()); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return 0, types.StateAtEvent{}, err
|
||||
}
|
||||
|
||||
return roomNID, types.StateAtEvent{
|
||||
BeforeStateSnapshotNID: stateNID,
|
||||
StateEntry: types.StateEntry{
|
||||
StateKeyTuple: types.StateKeyTuple{
|
||||
EventTypeNID: eventTypeNID,
|
||||
EventStateKeyNID: eventStateKeyNID,
|
||||
},
|
||||
EventNID: eventNID,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (d *Database) assignRoomNID(
|
||||
ctx context.Context, txn *sql.Tx,
|
||||
roomID string, roomVersion gomatrixserverlib.RoomVersion,
|
||||
) (types.RoomNID, error) {
|
||||
// Check if we already have a numeric ID in the database.
|
||||
roomNID, err := d.RoomsTable.SelectRoomNID(ctx, txn, roomID)
|
||||
if err == sql.ErrNoRows {
|
||||
// We don't have a numeric ID so insert one into the database.
|
||||
roomNID, err = d.RoomsTable.InsertRoomNID(ctx, txn, roomID, roomVersion)
|
||||
if err == sql.ErrNoRows {
|
||||
// We raced with another insert so run the select again.
|
||||
roomNID, err = d.RoomsTable.SelectRoomNID(ctx, txn, roomID)
|
||||
}
|
||||
}
|
||||
return roomNID, err
|
||||
}
|
||||
|
||||
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.EventTypesTable.SelectEventTypeNID(ctx, txn, eventType)
|
||||
if err == sql.ErrNoRows {
|
||||
// We don't have a numeric ID so insert one into the database.
|
||||
eventTypeNID, err = d.EventTypesTable.InsertEventTypeNID(ctx, txn, eventType)
|
||||
if err == sql.ErrNoRows {
|
||||
// We raced with another insert so run the select again.
|
||||
eventTypeNID, err = d.EventTypesTable.SelectEventTypeNID(ctx, txn, eventType)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
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.EventStateKeysTable.SelectEventStateKeyNID(ctx, txn, eventStateKey)
|
||||
if err == sql.ErrNoRows {
|
||||
// We don't have a numeric ID so insert one into the database.
|
||||
eventStateKeyNID, err = d.EventStateKeysTable.InsertEventStateKeyNID(ctx, txn, eventStateKey)
|
||||
if err == sql.ErrNoRows {
|
||||
// We raced with another insert so run the select again.
|
||||
eventStateKeyNID, err = d.EventStateKeysTable.SelectEventStateKeyNID(ctx, txn, eventStateKey)
|
||||
}
|
||||
}
|
||||
return eventStateKeyNID, err
|
||||
}
|
||||
|
||||
func extractRoomVersionFromCreateEvent(event gomatrixserverlib.Event) (
|
||||
gomatrixserverlib.RoomVersion, error,
|
||||
) {
|
||||
var err error
|
||||
var roomVersion gomatrixserverlib.RoomVersion
|
||||
// Look for m.room.create events.
|
||||
if event.Type() != gomatrixserverlib.MRoomCreate {
|
||||
return gomatrixserverlib.RoomVersion(""), nil
|
||||
}
|
||||
roomVersion = gomatrixserverlib.RoomVersionV1
|
||||
var createContent gomatrixserverlib.CreateContent
|
||||
// The m.room.create event contains an optional "room_version" key in
|
||||
// the event content, so we need to unmarshal that first.
|
||||
if err = json.Unmarshal(event.Content(), &createContent); err != nil {
|
||||
return gomatrixserverlib.RoomVersion(""), err
|
||||
}
|
||||
// A room version was specified in the event content?
|
||||
if createContent.RoomVersion != nil {
|
||||
roomVersion = gomatrixserverlib.RoomVersion(*createContent.RoomVersion)
|
||||
}
|
||||
return roomVersion, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -469,10 +469,9 @@ func (s *eventStatements) SelectMaxEventDepth(ctx context.Context, txn *sql.Tx,
|
|||
}
|
||||
|
||||
func (s *eventStatements) SelectRoomNIDForEventNID(
|
||||
ctx context.Context, txn *sql.Tx, eventNID types.EventNID,
|
||||
ctx context.Context, eventNID types.EventNID,
|
||||
) (roomNID types.RoomNID, err error) {
|
||||
selectStmt := internal.TxStmt(txn, s.selectRoomNIDForEventNIDStmt)
|
||||
err = selectStmt.QueryRowContext(ctx, int64(eventNID)).Scan(&roomNID)
|
||||
err = s.selectRoomNIDForEventNIDStmt.QueryRowContext(ctx, int64(eventNID)).Scan(&roomNID)
|
||||
return
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -178,11 +178,10 @@ func (s *roomStatements) SelectRoomVersionForRoomID(
|
|||
}
|
||||
|
||||
func (s *roomStatements) SelectRoomVersionForRoomNID(
|
||||
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
|
||||
ctx context.Context, roomNID types.RoomNID,
|
||||
) (gomatrixserverlib.RoomVersion, error) {
|
||||
var roomVersion gomatrixserverlib.RoomVersion
|
||||
stmt := internal.TxStmt(txn, s.selectRoomVersionForRoomNIDStmt)
|
||||
err := stmt.QueryRowContext(ctx, roomNID).Scan(&roomVersion)
|
||||
err := s.selectRoomVersionForRoomNIDStmt.QueryRowContext(ctx, roomNID).Scan(&roomVersion)
|
||||
if err == sql.ErrNoRows {
|
||||
return roomVersion, errors.New("room not found")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,14 +18,12 @@ package sqlite3
|
|||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/url"
|
||||
|
||||
"github.com/matrix-org/dendrite/internal/sqlutil"
|
||||
|
||||
"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"
|
||||
|
|
@ -110,126 +108,6 @@ func Open(dataSourceName string) (*Database, error) {
|
|||
return &d, nil
|
||||
}
|
||||
|
||||
// StoreEvent implements input.EventDatabase
|
||||
func (d *Database) StoreEvent(
|
||||
ctx context.Context, event gomatrixserverlib.Event,
|
||||
txnAndSessionID *api.TransactionID, authEventNIDs []types.EventNID,
|
||||
) (types.RoomNID, types.StateAtEvent, error) {
|
||||
var (
|
||||
roomNID types.RoomNID
|
||||
eventTypeNID types.EventTypeNID
|
||||
eventStateKeyNID types.EventStateKeyNID
|
||||
eventNID types.EventNID
|
||||
stateNID types.StateSnapshotNID
|
||||
err error
|
||||
)
|
||||
|
||||
err = internal.WithTransaction(d.db, func(txn *sql.Tx) error {
|
||||
if txnAndSessionID != nil {
|
||||
if err = d.transactions.InsertTransaction(
|
||||
ctx, txn, txnAndSessionID.TransactionID,
|
||||
txnAndSessionID.SessionID, event.Sender(), event.EventID(),
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: Here we should aim to have two different code paths for new rooms
|
||||
// vs existing ones.
|
||||
|
||||
// Get the default room version. If the client doesn't supply a room_version
|
||||
// then we will use our configured default to create the room.
|
||||
// https://matrix.org/docs/spec/client_server/r0.6.0#post-matrix-client-r0-createroom
|
||||
// Note that the below logic depends on the m.room.create event being the
|
||||
// first event that is persisted to the database when creating or joining a
|
||||
// room.
|
||||
var roomVersion gomatrixserverlib.RoomVersion
|
||||
if roomVersion, err = extractRoomVersionFromCreateEvent(event); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if roomNID, err = d.assignRoomNID(ctx, txn, event.RoomID(), roomVersion); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if eventTypeNID, err = d.assignEventTypeNID(ctx, txn, event.Type()); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
eventStateKey := event.StateKey()
|
||||
// Assigned a numeric ID for the state_key if there is one present.
|
||||
// Otherwise set the numeric ID for the state_key to 0.
|
||||
if eventStateKey != nil {
|
||||
if eventStateKeyNID, err = d.assignStateKeyNID(ctx, txn, *eventStateKey); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if eventNID, stateNID, err = d.events.InsertEvent(
|
||||
ctx,
|
||||
txn,
|
||||
roomNID,
|
||||
eventTypeNID,
|
||||
eventStateKeyNID,
|
||||
event.EventID(),
|
||||
event.EventReference().EventSHA256,
|
||||
authEventNIDs,
|
||||
event.Depth(),
|
||||
); err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
// We've already inserted the event so select the numeric event ID
|
||||
eventNID, stateNID, err = d.events.SelectEvent(ctx, txn, event.EventID())
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if err = d.eventJSON.InsertEventJSON(ctx, txn, eventNID, event.JSON()); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return 0, types.StateAtEvent{}, err
|
||||
}
|
||||
|
||||
return roomNID, types.StateAtEvent{
|
||||
BeforeStateSnapshotNID: stateNID,
|
||||
StateEntry: types.StateEntry{
|
||||
StateKeyTuple: types.StateKeyTuple{
|
||||
EventTypeNID: eventTypeNID,
|
||||
EventStateKeyNID: eventStateKeyNID,
|
||||
},
|
||||
EventNID: eventNID,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func extractRoomVersionFromCreateEvent(event gomatrixserverlib.Event) (
|
||||
gomatrixserverlib.RoomVersion, error,
|
||||
) {
|
||||
var err error
|
||||
var roomVersion gomatrixserverlib.RoomVersion
|
||||
// Look for m.room.create events.
|
||||
if event.Type() != gomatrixserverlib.MRoomCreate {
|
||||
return gomatrixserverlib.RoomVersion(""), nil
|
||||
}
|
||||
roomVersion = gomatrixserverlib.RoomVersionV1
|
||||
var createContent gomatrixserverlib.CreateContent
|
||||
// The m.room.create event contains an optional "room_version" key in
|
||||
// the event content, so we need to unmarshal that first.
|
||||
if err = json.Unmarshal(event.Content(), &createContent); err != nil {
|
||||
return gomatrixserverlib.RoomVersion(""), err
|
||||
}
|
||||
// A room version was specified in the event content?
|
||||
if createContent.RoomVersion != nil {
|
||||
roomVersion = gomatrixserverlib.RoomVersion(*createContent.RoomVersion)
|
||||
}
|
||||
return roomVersion, err
|
||||
}
|
||||
|
||||
func (d *Database) assignRoomNID(
|
||||
ctx context.Context, txn *sql.Tx,
|
||||
roomID string, roomVersion gomatrixserverlib.RoomVersion,
|
||||
|
|
@ -247,22 +125,6 @@ func (d *Database) assignRoomNID(
|
|||
return
|
||||
}
|
||||
|
||||
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.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.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
|
||||
}
|
||||
|
||||
func (d *Database) assignStateKeyNID(
|
||||
ctx context.Context, txn *sql.Tx, eventStateKey string,
|
||||
) (eventStateKeyNID types.EventStateKeyNID, err error) {
|
||||
|
|
@ -279,47 +141,6 @@ func (d *Database) assignStateKeyNID(
|
|||
return
|
||||
}
|
||||
|
||||
// Events implements input.EventDatabase
|
||||
func (d *Database) Events(
|
||||
ctx context.Context, eventNIDs []types.EventNID,
|
||||
) ([]types.Event, error) {
|
||||
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 {
|
||||
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.events.SelectRoomNIDForEventNID(ctx, txn, eventJSON.EventNID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
roomVersion, err = d.rooms.SelectRoomVersionForRoomNID(ctx, txn, roomNID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
result.Event, err = gomatrixserverlib.NewEventFromTrustedJSON(
|
||||
eventJSON.EventJSON, false, roomVersion,
|
||||
)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return []types.Event{}, err
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
|
||||
// AddState implements input.EventDatabase
|
||||
func (d *Database) AddState(
|
||||
ctx context.Context,
|
||||
|
|
@ -791,21 +612,6 @@ func (d *Database) GetMembershipEventNIDsForRoom(
|
|||
return
|
||||
}
|
||||
|
||||
// EventsFromIDs implements query.RoomserverQueryAPIEventDB
|
||||
func (d *Database) EventsFromIDs(ctx context.Context, eventIDs []string) ([]types.Event, error) {
|
||||
nidMap, err := d.EventNIDs(ctx, eventIDs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var nids []types.EventNID
|
||||
for _, nid := range nidMap {
|
||||
nids = append(nids, nid)
|
||||
}
|
||||
|
||||
return d.Events(ctx, nids)
|
||||
}
|
||||
|
||||
type transaction struct {
|
||||
ctx context.Context
|
||||
txn *sql.Tx
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ type Events interface {
|
|||
// 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)
|
||||
SelectRoomNIDForEventNID(ctx context.Context, eventNID types.EventNID) (roomNID types.RoomNID, err error)
|
||||
}
|
||||
|
||||
type Rooms interface {
|
||||
|
|
@ -63,7 +63,7 @@ type Rooms interface {
|
|||
SelectLatestEventsNIDsForUpdate(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) ([]types.EventNID, types.EventNID, types.StateSnapshotNID, error)
|
||||
UpdateLatestEventNIDs(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, eventNIDs []types.EventNID, lastEventSentNID types.EventNID, stateSnapshotNID types.StateSnapshotNID) error
|
||||
SelectRoomVersionForRoomID(ctx context.Context, txn *sql.Tx, roomID string) (gomatrixserverlib.RoomVersion, error)
|
||||
SelectRoomVersionForRoomNID(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) (gomatrixserverlib.RoomVersion, error)
|
||||
SelectRoomVersionForRoomNID(ctx context.Context, roomNID types.RoomNID) (gomatrixserverlib.RoomVersion, error)
|
||||
}
|
||||
|
||||
type Transactions interface {
|
||||
|
|
|
|||
Loading…
Reference in a new issue