diff options
Diffstat (limited to 'roomserver/storage/postgres/storage.go')
| -rw-r--r-- | roomserver/storage/postgres/storage.go | 281 |
1 files changed, 18 insertions, 263 deletions
diff --git a/roomserver/storage/postgres/storage.go b/roomserver/storage/postgres/storage.go index 0022c617..d44da858 100644 --- a/roomserver/storage/postgres/storage.go +++ b/roomserver/storage/postgres/storage.go @@ -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" @@ -40,6 +38,8 @@ type Database struct { eventTypes tables.EventTypes eventStateKeys tables.EventStateKeys eventJSON tables.EventJSON + rooms tables.Rooms + transactions tables.Transactions db *sql.DB } @@ -69,164 +69,43 @@ func Open(dataSourceName string, dbProperties internal.DbProperties) (*Database, if err != nil { return nil, err } + d.rooms, err = NewPostgresRoomsTable(d.db) + if err != nil { + return nil, err + } + d.transactions, err = NewPostgresTransactionsTable(d.db) + if err != nil { + return nil, err + } d.Database = shared.Database{ + DB: d.db, EventTypesTable: d.eventTypes, EventStateKeysTable: d.eventStateKeys, EventJSONTable: d.eventJSON, EventsTable: d.events, + RoomsTable: d.rooms, + TransactionsTable: d.transactions, } 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.statements.insertTransaction( - ctx, 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, ) (types.RoomNID, error) { // Check if we already have a numeric ID in the database. - roomNID, err := d.statements.selectRoomNID(ctx, txn, roomID) + roomNID, err := d.rooms.SelectRoomNID(ctx, txn, roomID) if err == sql.ErrNoRows { // We don't have a numeric ID so insert one into the database. - roomNID, err = d.statements.insertRoomNID(ctx, txn, roomID, roomVersion) + roomNID, err = d.rooms.InsertRoomNID(ctx, txn, roomID, roomVersion) if err == sql.ErrNoRows { // We raced with another insert so run the select again. - roomNID, err = d.statements.selectRoomNID(ctx, txn, roomID) + roomNID, err = d.rooms.SelectRoomNID(ctx, txn, roomID) } } 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) { @@ -243,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.statements.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, @@ -319,7 +166,7 @@ func (d *Database) GetLatestEventsForUpdate( return nil, err } eventNIDs, lastEventNIDSent, currentStateSnapshotNID, err := - d.statements.selectLatestEventsNIDsForUpdate(ctx, txn, roomNID) + d.rooms.SelectLatestEventsNIDsForUpdate(ctx, txn, roomNID) if err != nil { txn.Rollback() // nolint: errcheck return nil, err @@ -342,18 +189,6 @@ func (d *Database) GetLatestEventsForUpdate( }, nil } -// GetTransactionEventID implements input.EventDatabase -func (d *Database) GetTransactionEventID( - ctx context.Context, transactionID string, - sessionID int64, userID string, -) (string, error) { - eventID, err := d.statements.selectTransactionEventID(ctx, transactionID, sessionID, userID) - if err == sql.ErrNoRows { - return "", nil - } - return eventID, err -} - type roomRecentEventsUpdater struct { transaction d *Database @@ -415,7 +250,7 @@ func (u *roomRecentEventsUpdater) SetLatestEvents( for i := range latest { eventNIDs[i] = latest[i].EventNID } - return u.d.statements.updateLatestEventNIDs(u.ctx, u.txn, roomNID, eventNIDs, lastEventNIDSent, currentStateSnapshotNID) + return u.d.rooms.UpdateLatestEventNIDs(u.ctx, u.txn, roomNID, eventNIDs, lastEventNIDSent, currentStateSnapshotNID) } // HasEventBeenSent implements types.RoomRecentEventsUpdater @@ -432,55 +267,6 @@ func (u *roomRecentEventsUpdater) MembershipUpdater(targetUserNID types.EventSta return u.d.membershipUpdaterTxn(u.ctx, u.txn, u.roomNID, targetUserNID, targetLocal) } -// RoomNID implements query.RoomserverQueryAPIDB -func (d *Database) RoomNID(ctx context.Context, roomID string) (types.RoomNID, error) { - roomNID, err := d.statements.selectRoomNID(ctx, nil, roomID) - if err == sql.ErrNoRows { - return 0, nil - } - return roomNID, err -} - -// RoomNIDExcludingStubs implements query.RoomserverQueryAPIDB -func (d *Database) RoomNIDExcludingStubs(ctx context.Context, roomID string) (roomNID types.RoomNID, err error) { - roomNID, err = d.RoomNID(ctx, roomID) - if err != nil { - return - } - latestEvents, _, err := d.statements.selectLatestEventNIDs(ctx, roomNID) - if err != nil { - return - } - if len(latestEvents) == 0 { - roomNID = 0 - return - } - return -} - -// LatestEventIDs implements query.RoomserverQueryAPIDatabase -func (d *Database) LatestEventIDs( - ctx context.Context, roomNID types.RoomNID, -) (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 func (d *Database) GetInvitesForUser( ctx context.Context, @@ -733,37 +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) -} - -func (d *Database) GetRoomVersionForRoom( - ctx context.Context, roomID string, -) (gomatrixserverlib.RoomVersion, error) { - return d.statements.selectRoomVersionForRoomID( - ctx, nil, roomID, - ) -} - -func (d *Database) GetRoomVersionForRoomNID( - ctx context.Context, roomNID types.RoomNID, -) (gomatrixserverlib.RoomVersion, error) { - return d.statements.selectRoomVersionForRoomNID( - ctx, nil, roomNID, - ) -} - type transaction struct { ctx context.Context txn *sql.Tx |