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

Undo non-roomserver changes for now

Browse source
This commit is contained in:
Neil Alexander 2020-08-18 09:54:51 +01:00
parent 5c1183fba4
commit c9ee7cc269
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
41 changed files with 135 additions and 164 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
appservice/storage/sqlite3/appservice_events_table.go
View file

@ -75,9 +75,9 @@ type eventsStatements struct {
deleteEventsBeforeAndIncludingIDStmt *sql.Stmt
}
func (s *eventsStatements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter) (err error) {
func (s *eventsStatements) prepare(db *sql.DB) (err error) {
s.db = db
s.writer = writer
s.writer = sqlutil.NewTransactionWriter()
_, err = db.Exec(appserviceEventsSchema)
if err != nil {
return

9
appservice/storage/sqlite3/storage.go
View file

@ -41,8 +41,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*Database, error) {
if result.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
writer := sqlutil.NewTransactionWriter()
if err = result.prepare(writer); err != nil {
if err = result.prepare(); err != nil {
return nil, err
}
if err = result.PartitionOffsetStatements.Prepare(result.db, "appservice"); err != nil {
@ -51,12 +50,12 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*Database, error) {
return &result, nil
}
func (d *Database) prepare(writer *sqlutil.TransactionWriter) error {
if err := d.events.prepare(d.db, writer); err != nil {
func (d *Database) prepare() error {
if err := d.events.prepare(d.db); err != nil {
return err
}
return d.txnID.prepare(d.db, writer)
return d.txnID.prepare(d.db)
}
// StoreEvent takes in a gomatrixserverlib.HeaderedEvent and stores it in the database

4
appservice/storage/sqlite3/txn_id_counter_table.go
View file

@ -42,9 +42,9 @@ type txnStatements struct {
selectTxnIDStmt *sql.Stmt
}
func (s *txnStatements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter) (err error) {
func (s *txnStatements) prepare(db *sql.DB) (err error) {
s.db = db
s.writer = writer
s.writer = sqlutil.NewTransactionWriter()
_, err = db.Exec(txnIDSchema)
if err != nil {
return

4
currentstateserver/storage/sqlite3/current_room_state_table.go
View file

@ -93,10 +93,10 @@ type currentRoomStateStatements struct {
selectKnownUsersStmt *sql.Stmt
}
func NewSqliteCurrentRoomStateTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.CurrentRoomState, error) {
func NewSqliteCurrentRoomStateTable(db *sql.DB) (tables.CurrentRoomState, error) {
s := &currentRoomStateStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err := db.Exec(currentRoomStateSchema)
if err != nil {

3
currentstateserver/storage/sqlite3/storage.go
View file

@ -22,11 +22,10 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*Database, error) {
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
writer := sqlutil.NewTransactionWriter()
if err = d.PartitionOffsetStatements.Prepare(d.db, "currentstate"); err != nil {
return nil, err
}
currRoomState, err := NewSqliteCurrentRoomStateTable(d.db, writer)
currRoomState, err := NewSqliteCurrentRoomStateTable(d.db)
if err != nil {
return nil, err
}

4
federationsender/storage/sqlite3/blacklist_table.go
View file

@ -48,10 +48,10 @@ type blacklistStatements struct {
deleteBlacklistStmt *sql.Stmt
}
func NewSQLiteBlacklistTable(db *sql.DB, writer *sqlutil.TransactionWriter) (s *blacklistStatements, err error) {
func NewSQLiteBlacklistTable(db *sql.DB) (s *blacklistStatements, err error) {
s = &blacklistStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err = db.Exec(blacklistSchema)
if err != nil {

4
federationsender/storage/sqlite3/joined_hosts_table.go
View file

@ -73,10 +73,10 @@ type joinedHostsStatements struct {
// selectJoinedHostsForRoomsStmt *sql.Stmt - prepared at runtime due to variadic
}
func NewSQLiteJoinedHostsTable(db *sql.DB, writer *sqlutil.TransactionWriter) (s *joinedHostsStatements, err error) {
func NewSQLiteJoinedHostsTable(db *sql.DB) (s *joinedHostsStatements, err error) {
s = &joinedHostsStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err = db.Exec(joinedHostsSchema)
if err != nil {

4
federationsender/storage/sqlite3/queue_edus_table.go
View file

@ -72,10 +72,10 @@ type queueEDUsStatements struct {
selectQueueEDUServerNamesStmt *sql.Stmt
}
func NewSQLiteQueueEDUsTable(db *sql.DB, writer *sqlutil.TransactionWriter) (s *queueEDUsStatements, err error) {
func NewSQLiteQueueEDUsTable(db *sql.DB) (s *queueEDUsStatements, err error) {
s = &queueEDUsStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err = db.Exec(queueEDUsSchema)
if err != nil {

4
federationsender/storage/sqlite3/queue_json_table.go
View file

@ -56,10 +56,10 @@ type queueJSONStatements struct {
//selectJSONStmt *sql.Stmt - prepared at runtime due to variadic
}
func NewSQLiteQueueJSONTable(db *sql.DB, writer *sqlutil.TransactionWriter) (s *queueJSONStatements, err error) {
func NewSQLiteQueueJSONTable(db *sql.DB) (s *queueJSONStatements, err error) {
s = &queueJSONStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err = db.Exec(queueJSONSchema)
if err != nil {

4
federationsender/storage/sqlite3/queue_pdus_table.go
View file

@ -81,10 +81,10 @@ type queuePDUsStatements struct {
// deleteQueuePDUsStmt *sql.Stmt - prepared at runtime due to variadic
}
func NewSQLiteQueuePDUsTable(db *sql.DB, writer *sqlutil.TransactionWriter) (s *queuePDUsStatements, err error) {
func NewSQLiteQueuePDUsTable(db *sql.DB) (s *queuePDUsStatements, err error) {
s = &queuePDUsStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err = db.Exec(queuePDUsSchema)
if err != nil {

4
federationsender/storage/sqlite3/room_table.go
View file

@ -50,10 +50,10 @@ type roomStatements struct {
updateRoomStmt *sql.Stmt
}
func NewSQLiteRoomsTable(db *sql.DB, writer *sqlutil.TransactionWriter) (s *roomStatements, err error) {
func NewSQLiteRoomsTable(db *sql.DB) (s *roomStatements, err error) {
s = &roomStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err = db.Exec(roomSchema)
if err != nil {

13
federationsender/storage/sqlite3/storage.go
View file

@ -39,28 +39,27 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*Database, error) {
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
writer := sqlutil.NewTransactionWriter()
joinedHosts, err := NewSQLiteJoinedHostsTable(d.db, writer)
joinedHosts, err := NewSQLiteJoinedHostsTable(d.db)
if err != nil {
return nil, err
}
rooms, err := NewSQLiteRoomsTable(d.db, writer)
rooms, err := NewSQLiteRoomsTable(d.db)
if err != nil {
return nil, err
}
queuePDUs, err := NewSQLiteQueuePDUsTable(d.db, writer)
queuePDUs, err := NewSQLiteQueuePDUsTable(d.db)
if err != nil {
return nil, err
}
queueEDUs, err := NewSQLiteQueueEDUsTable(d.db, writer)
queueEDUs, err := NewSQLiteQueueEDUsTable(d.db)
if err != nil {
return nil, err
}
queueJSON, err := NewSQLiteQueueJSONTable(d.db, writer)
queueJSON, err := NewSQLiteQueueJSONTable(d.db)
if err != nil {
return nil, err
}
blacklist, err := NewSQLiteBlacklistTable(d.db, writer)
blacklist, err := NewSQLiteBlacklistTable(d.db)
if err != nil {
return nil, err
}

4
keyserver/storage/sqlite3/device_keys_table.go
View file

@ -71,10 +71,10 @@ type deviceKeysStatements struct {
deleteAllDeviceKeysStmt *sql.Stmt
}
func NewSqliteDeviceKeysTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.DeviceKeys, error) {
func NewSqliteDeviceKeysTable(db *sql.DB) (tables.DeviceKeys, error) {
s := &deviceKeysStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err := db.Exec(deviceKeysSchema)
if err != nil {

4
keyserver/storage/sqlite3/key_changes_table.go
View file

@ -57,10 +57,10 @@ type keyChangesStatements struct {
selectKeyChangesStmt *sql.Stmt
}
func NewSqliteKeyChangesTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.KeyChanges, error) {
func NewSqliteKeyChangesTable(db *sql.DB) (tables.KeyChanges, error) {
s := &keyChangesStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err := db.Exec(keyChangesSchema)
if err != nil {

4
keyserver/storage/sqlite3/one_time_keys_table.go
View file

@ -68,10 +68,10 @@ type oneTimeKeysStatements struct {
deleteOneTimeKeyStmt *sql.Stmt
}
func NewSqliteOneTimeKeysTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.OneTimeKeys, error) {
func NewSqliteOneTimeKeysTable(db *sql.DB) (tables.OneTimeKeys, error) {
s := &oneTimeKeysStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err := db.Exec(oneTimeKeysSchema)
if err != nil {

16
keyserver/storage/sqlite3/stale_device_lists.go
View file

@ -20,7 +20,6 @@ import (
"time"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/keyserver/storage/tables"
"github.com/matrix-org/gomatrixserverlib"
)
@ -50,18 +49,13 @@ const selectStaleDeviceListsSQL = "" +
"SELECT user_id FROM keyserver_stale_device_lists WHERE is_stale = $1"
type staleDeviceListsStatements struct {
db *sql.DB
writer *sqlutil.TransactionWriter
upsertStaleDeviceListStmt *sql.Stmt
selectStaleDeviceListsWithDomainsStmt *sql.Stmt
selectStaleDeviceListsStmt *sql.Stmt
}
func NewSqliteStaleDeviceListsTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.StaleDeviceLists, error) {
s := &staleDeviceListsStatements{
db: db,
writer: writer,
}
func NewSqliteStaleDeviceListsTable(db *sql.DB) (tables.StaleDeviceLists, error) {
s := &staleDeviceListsStatements{}
_, err := db.Exec(staleDeviceListsSchema)
if err != nil {
return nil, err
@ -83,10 +77,8 @@ func (s *staleDeviceListsStatements) InsertStaleDeviceList(ctx context.Context,
if err != nil {
return err
}
return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
_, err = s.upsertStaleDeviceListStmt.ExecContext(ctx, userID, string(domain), isStale, time.Now().Unix())
return err
})
_, err = s.upsertStaleDeviceListStmt.ExecContext(ctx, userID, string(domain), isStale, time.Now().Unix())
return err
}
func (s *staleDeviceListsStatements) SelectUserIDsWithStaleDeviceLists(ctx context.Context, domains []gomatrixserverlib.ServerName) ([]string, error) {

9
keyserver/storage/sqlite3/storage.go
View file

@ -25,20 +25,19 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*shared.Database, error)
if err != nil {
return nil, err
}
writer := sqlutil.NewTransactionWriter()
otk, err := NewSqliteOneTimeKeysTable(db, writer)
otk, err := NewSqliteOneTimeKeysTable(db)
if err != nil {
return nil, err
}
dk, err := NewSqliteDeviceKeysTable(db, writer)
dk, err := NewSqliteDeviceKeysTable(db)
if err != nil {
return nil, err
}
kc, err := NewSqliteKeyChangesTable(db, writer)
kc, err := NewSqliteKeyChangesTable(db)
if err != nil {
return nil, err
}
sdl, err := NewSqliteStaleDeviceListsTable(db, writer)
sdl, err := NewSqliteStaleDeviceListsTable(db)
if err != nil {
return nil, err
}

4
mediaapi/storage/sqlite3/media_repository_table.go
View file

@ -67,9 +67,9 @@ type mediaStatements struct {
selectMediaStmt *sql.Stmt
}
func (s *mediaStatements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter) (err error) {
func (s *mediaStatements) prepare(db *sql.DB) (err error) {
s.db = db
s.writer = writer
s.writer = sqlutil.NewTransactionWriter()
_, err = db.Exec(mediaSchema)
if err != nil {

8
mediaapi/storage/sqlite3/sql.go
View file

@ -17,8 +17,6 @@ package sqlite3
import (
"database/sql"
"github.com/matrix-org/dendrite/internal/sqlutil"
)
type statements struct {
@ -26,11 +24,11 @@ type statements struct {
thumbnail thumbnailStatements
}
func (s *statements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter) (err error) {
if err = s.media.prepare(db, writer); err != nil {
func (s *statements) prepare(db *sql.DB) (err error) {
if err = s.media.prepare(db); err != nil {
return
}
if err = s.thumbnail.prepare(db, writer); err != nil {
if err = s.thumbnail.prepare(db); err != nil {
return
}

4
mediaapi/storage/sqlite3/storage.go
View file

@ -31,7 +31,6 @@ import (
type Database struct {
statements statements
db *sql.DB
writer *sqlutil.TransactionWriter
}
// Open opens a postgres database.
@ -41,8 +40,7 @@ func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
d.writer = sqlutil.NewTransactionWriter()
if err = d.statements.prepare(d.db, d.writer); err != nil {
if err = d.statements.prepare(d.db); err != nil {
return nil, err
}
return &d, nil

34
mediaapi/storage/sqlite3/thumbnail_table.go
View file

@ -21,7 +21,6 @@ import (
"time"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/mediaapi/types"
"github.com/matrix-org/gomatrixserverlib"
)
@ -58,16 +57,12 @@ SELECT content_type, file_size_bytes, creation_ts, width, height, resize_method
`
type thumbnailStatements struct {
db *sql.DB
writer *sqlutil.TransactionWriter
insertThumbnailStmt *sql.Stmt
selectThumbnailStmt *sql.Stmt
selectThumbnailsStmt *sql.Stmt
}
func (s *thumbnailStatements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter) (err error) {
s.db = db
s.writer = writer
func (s *thumbnailStatements) prepare(db *sql.DB) (err error) {
_, err = db.Exec(thumbnailSchema)
if err != nil {
return
@ -84,21 +79,18 @@ func (s *thumbnailStatements) insertThumbnail(
ctx context.Context, thumbnailMetadata *types.ThumbnailMetadata,
) error {
thumbnailMetadata.MediaMetadata.CreationTimestamp = types.UnixMs(time.Now().UnixNano() / 1000000)
return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
stmt := sqlutil.TxStmt(txn, s.insertThumbnailStmt)
_, err := stmt.ExecContext(
ctx,
thumbnailMetadata.MediaMetadata.MediaID,
thumbnailMetadata.MediaMetadata.Origin,
thumbnailMetadata.MediaMetadata.ContentType,
thumbnailMetadata.MediaMetadata.FileSizeBytes,
thumbnailMetadata.MediaMetadata.CreationTimestamp,
thumbnailMetadata.ThumbnailSize.Width,
thumbnailMetadata.ThumbnailSize.Height,
thumbnailMetadata.ThumbnailSize.ResizeMethod,
)
return err
})
_, err := s.insertThumbnailStmt.ExecContext(
ctx,
thumbnailMetadata.MediaMetadata.MediaID,
thumbnailMetadata.MediaMetadata.Origin,
thumbnailMetadata.MediaMetadata.ContentType,
thumbnailMetadata.MediaMetadata.FileSizeBytes,
thumbnailMetadata.MediaMetadata.CreationTimestamp,
thumbnailMetadata.ThumbnailSize.Width,
thumbnailMetadata.ThumbnailSize.Height,
thumbnailMetadata.ThumbnailSize.ResizeMethod,
)
return err
}
func (s *thumbnailStatements) selectThumbnail(

10
serverkeyapi/storage/sqlite3/keydb.go
View file

@ -17,7 +17,6 @@ package sqlite3
import (
"context"
"database/sql"
"golang.org/x/crypto/ed25519"
@ -31,8 +30,6 @@ import (
// A Database implements gomatrixserverlib.KeyDatabase and is used to store
// the public keys for other matrix servers.
type Database struct {
db *sql.DB
writer *sqlutil.TransactionWriter
statements serverKeyStatements
}
@ -50,11 +47,8 @@ func NewDatabase(
if err != nil {
return nil, err
}
d := &Database{
db: db,
writer: sqlutil.NewTransactionWriter(),
}
err = d.statements.prepare(d.db, d.writer)
d := &Database{}
err = d.statements.prepare(db)
if err != nil {
return nil, err
}

4
serverkeyapi/storage/sqlite3/server_key_table.go
View file

@ -68,9 +68,9 @@ type serverKeyStatements struct {
upsertServerKeysStmt *sql.Stmt
}
func (s *serverKeyStatements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter) (err error) {
func (s *serverKeyStatements) prepare(db *sql.DB) (err error) {
s.db = db
s.writer = writer
s.writer = sqlutil.NewTransactionWriter()
_, err = db.Exec(serverKeysSchema)
if err != nil {
return

6
syncapi/storage/shared/syncserver.go
View file

@ -171,7 +171,11 @@ func (d *Database) SyncStreamPosition(ctx context.Context) (types.StreamPosition
func (d *Database) AddInviteEvent(
ctx context.Context, inviteEvent gomatrixserverlib.HeaderedEvent,
) (sp types.StreamPosition, err error) {
return d.Invites.InsertInviteEvent(ctx, nil, inviteEvent)
err = sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error {
sp, err = d.Invites.InsertInviteEvent(ctx, txn, inviteEvent)
return err
})
return
}
// RetireInviteEvent removes an old invite event from the database.

4
syncapi/storage/sqlite3/account_data_table.go
View file

@ -58,10 +58,10 @@ type accountDataStatements struct {
selectAccountDataInRangeStmt *sql.Stmt
}
func NewSqliteAccountDataTable(db *sql.DB, writer *sqlutil.TransactionWriter, streamID *streamIDStatements) (tables.AccountData, error) {
func NewSqliteAccountDataTable(db *sql.DB, streamID *streamIDStatements) (tables.AccountData, error) {
s := &accountDataStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
streamIDStatements: streamID,
}
_, err := db.Exec(accountDataSchema)

4
syncapi/storage/sqlite3/backwards_extremities_table.go
View file

@ -55,10 +55,10 @@ type backwardExtremitiesStatements struct {
deleteBackwardExtremityStmt *sql.Stmt
}
func NewSqliteBackwardsExtremitiesTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.BackwardsExtremities, error) {
func NewSqliteBackwardsExtremitiesTable(db *sql.DB) (tables.BackwardsExtremities, error) {
s := &backwardExtremitiesStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err := db.Exec(backwardExtremitiesSchema)
if err != nil {

4
syncapi/storage/sqlite3/current_room_state_table.go
View file

@ -95,10 +95,10 @@ type currentRoomStateStatements struct {
selectStateEventStmt *sql.Stmt
}
func NewSqliteCurrentRoomStateTable(db *sql.DB, writer *sqlutil.TransactionWriter, streamID *streamIDStatements) (tables.CurrentRoomState, error) {
func NewSqliteCurrentRoomStateTable(db *sql.DB, streamID *streamIDStatements) (tables.CurrentRoomState, error) {
s := &currentRoomStateStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
streamIDStatements: streamID,
}
_, err := db.Exec(currentRoomStateSchema)

4
syncapi/storage/sqlite3/filter_table.go
View file

@ -58,14 +58,14 @@ type filterStatements struct {
insertFilterStmt *sql.Stmt
}
func NewSqliteFilterTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.Filter, error) {
func NewSqliteFilterTable(db *sql.DB) (tables.Filter, error) {
_, err := db.Exec(filterSchema)
if err != nil {
return nil, err
}
s := &filterStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
if s.selectFilterStmt, err = db.Prepare(selectFilterSQL); err != nil {
return nil, err

27
syncapi/storage/sqlite3/invites_table.go
View file

@ -67,10 +67,10 @@ type inviteEventsStatements struct {
selectMaxInviteIDStmt *sql.Stmt
}
func NewSqliteInvitesTable(db *sql.DB, writer *sqlutil.TransactionWriter, streamID *streamIDStatements) (tables.Invites, error) {
func NewSqliteInvitesTable(db *sql.DB, streamID *streamIDStatements) (tables.Invites, error) {
s := &inviteEventsStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
streamIDStatements: streamID,
}
_, err := db.Exec(inviteEventsSchema)
@ -95,12 +95,13 @@ func NewSqliteInvitesTable(db *sql.DB, writer *sqlutil.TransactionWriter, stream
func (s *inviteEventsStatements) InsertInviteEvent(
ctx context.Context, txn *sql.Tx, inviteEvent gomatrixserverlib.HeaderedEvent,
) (streamPos types.StreamPosition, err error) {
streamPos, err = s.streamIDStatements.nextStreamID(ctx, txn)
if err != nil {
return
}
err = s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
var err error
streamPos, err = s.streamIDStatements.nextStreamID(ctx, txn)
if err != nil {
return err
}
var headeredJSON []byte
headeredJSON, err = json.Marshal(inviteEvent)
if err != nil {
@ -123,11 +124,13 @@ func (s *inviteEventsStatements) InsertInviteEvent(
func (s *inviteEventsStatements) DeleteInviteEvent(
ctx context.Context, inviteEventID string,
) (types.StreamPosition, error) {
streamPos, err := s.streamIDStatements.nextStreamID(ctx, nil)
if err != nil {
return streamPos, err
}
err = s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
var streamPos types.StreamPosition
err := s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
var err error
streamPos, err = s.streamIDStatements.nextStreamID(ctx, nil)
if err != nil {
return err
}
_, err = s.deleteInviteEventStmt.ExecContext(ctx, streamPos, inviteEventID)
return err
})

14
syncapi/storage/sqlite3/output_room_events_table.go
View file

@ -117,10 +117,10 @@ type outputRoomEventsStatements struct {
updateEventJSONStmt *sql.Stmt
}
func NewSqliteEventsTable(db *sql.DB, writer *sqlutil.TransactionWriter, streamID *streamIDStatements) (tables.Events, error) {
func NewSqliteEventsTable(db *sql.DB, streamID *streamIDStatements) (tables.Events, error) {
s := &outputRoomEventsStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
streamIDStatements: streamID,
}
_, err := db.Exec(outputRoomEventsSchema)
@ -304,11 +304,13 @@ func (s *outputRoomEventsStatements) InsertEvent(
return 0, err
}
streamPos, err := s.streamIDStatements.nextStreamID(ctx, txn)
if err != nil {
return 0, err
}
var streamPos types.StreamPosition
err = s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
streamPos, err = s.streamIDStatements.nextStreamID(ctx, txn)
if err != nil {
return err
}
insertStmt := sqlutil.TxStmt(txn, s.insertEventStmt)
_, ierr := insertStmt.ExecContext(
ctx,

4
syncapi/storage/sqlite3/output_room_events_topology_table.go
View file

@ -75,10 +75,10 @@ type outputRoomEventsTopologyStatements struct {
selectMaxPositionInTopologyStmt *sql.Stmt
}
func NewSqliteTopologyTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.Topology, error) {
func NewSqliteTopologyTable(db *sql.DB) (tables.Topology, error) {
s := &outputRoomEventsTopologyStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err := db.Exec(outputRoomEventsTopologySchema)
if err != nil {

4
syncapi/storage/sqlite3/send_to_device_table.go
View file

@ -79,10 +79,10 @@ type sendToDeviceStatements struct {
countSendToDeviceMessagesStmt *sql.Stmt
}
func NewSqliteSendToDeviceTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.SendToDevice, error) {
func NewSqliteSendToDeviceTable(db *sql.DB) (tables.SendToDevice, error) {
s := &sendToDeviceStatements{
db: db,
writer: writer,
writer: sqlutil.NewTransactionWriter(),
}
_, err := db.Exec(sendToDeviceSchema)
if err != nil {

4
syncapi/storage/sqlite3/stream_id_table.go
View file

@ -33,9 +33,9 @@ type streamIDStatements struct {
selectStreamIDStmt *sql.Stmt
}
func (s *streamIDStatements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter) (err error) {
func (s *streamIDStatements) prepare(db *sql.DB) (err error) {
s.db = db
s.writer = writer
s.writer = sqlutil.NewTransactionWriter()
_, err = db.Exec(streamIDTableSchema)
if err != nil {
return

22
syncapi/storage/sqlite3/syncserver.go
View file

@ -31,8 +31,7 @@ import (
// both the database for PDUs and caches for EDUs.
type SyncServerDatasource struct {
shared.Database
db *sql.DB
writer *sqlutil.TransactionWriter
db *sql.DB
sqlutil.PartitionOffsetStatements
streamID streamIDStatements
}
@ -45,7 +44,6 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
d.writer = sqlutil.NewTransactionWriter()
if err = d.prepare(); err != nil {
return nil, err
}
@ -56,38 +54,38 @@ func (d *SyncServerDatasource) prepare() (err error) {
if err = d.PartitionOffsetStatements.Prepare(d.db, "syncapi"); err != nil {
return err
}
if err = d.streamID.prepare(d.db, d.writer); err != nil {
if err = d.streamID.prepare(d.db); err != nil {
return err
}
accountData, err := NewSqliteAccountDataTable(d.db, d.writer, &d.streamID)
accountData, err := NewSqliteAccountDataTable(d.db, &d.streamID)
if err != nil {
return err
}
events, err := NewSqliteEventsTable(d.db, d.writer, &d.streamID)
events, err := NewSqliteEventsTable(d.db, &d.streamID)
if err != nil {
return err
}
roomState, err := NewSqliteCurrentRoomStateTable(d.db, d.writer, &d.streamID)
roomState, err := NewSqliteCurrentRoomStateTable(d.db, &d.streamID)
if err != nil {
return err
}
invites, err := NewSqliteInvitesTable(d.db, d.writer, &d.streamID)
invites, err := NewSqliteInvitesTable(d.db, &d.streamID)
if err != nil {
return err
}
topology, err := NewSqliteTopologyTable(d.db, d.writer)
topology, err := NewSqliteTopologyTable(d.db)
if err != nil {
return err
}
bwExtrem, err := NewSqliteBackwardsExtremitiesTable(d.db, d.writer)
bwExtrem, err := NewSqliteBackwardsExtremitiesTable(d.db)
if err != nil {
return err
}
sendToDevice, err := NewSqliteSendToDeviceTable(d.db, d.writer)
sendToDevice, err := NewSqliteSendToDeviceTable(d.db)
if err != nil {
return err
}
filter, err := NewSqliteFilterTable(d.db, d.writer)
filter, err := NewSqliteFilterTable(d.db)
if err != nil {
return err
}

4
userapi/storage/accounts/sqlite3/account_data_table.go
View file

@ -57,9 +57,9 @@ type accountDataStatements struct {
selectAccountDataByTypeStmt *sql.Stmt
}
func (s *accountDataStatements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter) (err error) {
func (s *accountDataStatements) prepare(db *sql.DB) (err error) {
s.db = db
s.writer = writer
s.writer = sqlutil.NewTransactionWriter()
_, err = db.Exec(accountDataSchema)
if err != nil {
return

4
userapi/storage/accounts/sqlite3/accounts_table.go
View file

@ -67,9 +67,9 @@ type accountsStatements struct {
serverName gomatrixserverlib.ServerName
}
func (s *accountsStatements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter, server gomatrixserverlib.ServerName) (err error) {
func (s *accountsStatements) prepare(db *sql.DB, server gomatrixserverlib.ServerName) (err error) {
s.db = db
s.writer = writer
s.writer = sqlutil.NewTransactionWriter()
_, err = db.Exec(accountsSchema)
if err != nil {
return

4
userapi/storage/accounts/sqlite3/profile_table.go
View file

@ -61,9 +61,9 @@ type profilesStatements struct {
selectProfilesBySearchStmt *sql.Stmt
}
func (s *profilesStatements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter) (err error) {
func (s *profilesStatements) prepare(db *sql.DB) (err error) {
s.db = db
s.writer = writer
s.writer = sqlutil.NewTransactionWriter()
_, err = db.Exec(profilesSchema)
if err != nil {
return

14
userapi/storage/accounts/sqlite3/storage.go
View file

@ -57,24 +57,20 @@ func NewDatabase(dbProperties *config.DatabaseOptions, serverName gomatrixserver
if err = partitions.Prepare(db, "account"); err != nil {
return nil, err
}
writer := sqlutil.NewTransactionWriter()
a := accountsStatements{
db: db,
writer: writer,
}
if err = a.prepare(db, writer, serverName); err != nil {
a := accountsStatements{}
if err = a.prepare(db, serverName); err != nil {
return nil, err
}
p := profilesStatements{}
if err = p.prepare(db, writer); err != nil {
if err = p.prepare(db); err != nil {
return nil, err
}
ac := accountDataStatements{}
if err = ac.prepare(db, writer); err != nil {
if err = ac.prepare(db); err != nil {
return nil, err
}
t := threepidStatements{}
if err = t.prepare(db, writer); err != nil {
if err = t.prepare(db); err != nil {
return nil, err
}
return &Database{

4
userapi/storage/accounts/sqlite3/threepid_table.go
View file

@ -61,9 +61,9 @@ type threepidStatements struct {
deleteThreePIDStmt *sql.Stmt
}
func (s *threepidStatements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter) (err error) {
func (s *threepidStatements) prepare(db *sql.DB) (err error) {
s.db = db
s.writer = writer
s.writer = sqlutil.NewTransactionWriter()
_, err = db.Exec(threepidSchema)
if err != nil {
return

4
userapi/storage/devices/sqlite3/devices_table.go
View file

@ -91,9 +91,9 @@ type devicesStatements struct {
serverName gomatrixserverlib.ServerName
}
func (s *devicesStatements) prepare(db *sql.DB, writer *sqlutil.TransactionWriter, server gomatrixserverlib.ServerName) (err error) {
func (s *devicesStatements) prepare(db *sql.DB, server gomatrixserverlib.ServerName) (err error) {
s.db = db
s.writer = writer
s.writer = sqlutil.NewTransactionWriter()
_, err = db.Exec(devicesSchema)
if err != nil {
return

6
userapi/storage/devices/sqlite3/storage.go
View file

@ -34,7 +34,6 @@ var deviceIDByteLength = 6
// Database represents a device database.
type Database struct {
db *sql.DB
writer *sqlutil.TransactionWriter
devices devicesStatements
}
@ -44,12 +43,11 @@ func NewDatabase(dbProperties *config.DatabaseOptions, serverName gomatrixserver
if err != nil {
return nil, err
}
writer := sqlutil.NewTransactionWriter()
d := devicesStatements{}
if err = d.prepare(db, writer, serverName); err != nil {
if err = d.prepare(db, serverName); err != nil {
return nil, err
}
return &Database{db, writer, d}, nil
return &Database{db, d}, nil
}
// GetDeviceByAccessToken returns the device matching the given access token.