mirror of
https://github.com/hoernschen/dendrite.git
synced 2025-08-01 05:42:46 +00:00
* Fix #897 and shuffle directory around * Update find-lint * goimports Co-authored-by: Neil Alexander <neilalexander@users.noreply.github.com>
This commit is contained in:
parent
3fdb045116
commit
24d8df664c
260 changed files with 901 additions and 901 deletions
|
@ -19,15 +19,15 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/eduserver/cache"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
type Database interface {
|
||||
common.PartitionStorer
|
||||
internal.PartitionStorer
|
||||
// AllJoinedUsersInRooms returns a map of room ID to a list of all joined user IDs.
|
||||
AllJoinedUsersInRooms(ctx context.Context) (map[string][]string, error)
|
||||
// Events lookups a list of event by their event ID.
|
||||
|
|
|
@ -20,7 +20,7 @@ import (
|
|||
"database/sql"
|
||||
|
||||
"github.com/lib/pq"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
@ -113,7 +113,7 @@ func (s *accountDataStatements) SelectAccountDataInRange(
|
|||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectAccountDataInRange: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectAccountDataInRange: rows.close() failed")
|
||||
|
||||
for rows.Next() {
|
||||
var dataType string
|
||||
|
@ -136,7 +136,7 @@ func (s *accountDataStatements) SelectMaxAccountDataID(
|
|||
ctx context.Context, txn *sql.Tx,
|
||||
) (id int64, err error) {
|
||||
var nullableID sql.NullInt64
|
||||
stmt := common.TxStmt(txn, s.selectMaxAccountDataIDStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectMaxAccountDataIDStmt)
|
||||
err = stmt.QueryRowContext(ctx).Scan(&nullableID)
|
||||
if nullableID.Valid {
|
||||
id = nullableID.Int64
|
||||
|
|
|
@ -18,7 +18,7 @@ import (
|
|||
"context"
|
||||
"database/sql"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
)
|
||||
|
||||
|
@ -84,7 +84,7 @@ func (s *backwardExtremitiesStatements) SelectBackwardExtremitiesForRoom(
|
|||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectBackwardExtremitiesForRoom: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectBackwardExtremitiesForRoom: rows.close() failed")
|
||||
|
||||
bwExtrems = make(map[string][]string)
|
||||
for rows.Next() {
|
||||
|
|
|
@ -21,7 +21,7 @@ import (
|
|||
"encoding/json"
|
||||
|
||||
"github.com/lib/pq"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
@ -142,7 +142,7 @@ func (s *currentRoomStateStatements) SelectJoinedUsers(
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectJoinedUsers: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectJoinedUsers: rows.close() failed")
|
||||
|
||||
result := make(map[string][]string)
|
||||
for rows.Next() {
|
||||
|
@ -165,12 +165,12 @@ func (s *currentRoomStateStatements) SelectRoomIDsWithMembership(
|
|||
userID string,
|
||||
membership string, // nolint: unparam
|
||||
) ([]string, error) {
|
||||
stmt := common.TxStmt(txn, s.selectRoomIDsWithMembershipStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectRoomIDsWithMembershipStmt)
|
||||
rows, err := stmt.QueryContext(ctx, userID, membership)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectRoomIDsWithMembership: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectRoomIDsWithMembership: rows.close() failed")
|
||||
|
||||
var result []string
|
||||
for rows.Next() {
|
||||
|
@ -188,7 +188,7 @@ func (s *currentRoomStateStatements) SelectCurrentState(
|
|||
ctx context.Context, txn *sql.Tx, roomID string,
|
||||
stateFilter *gomatrixserverlib.StateFilter,
|
||||
) ([]gomatrixserverlib.HeaderedEvent, error) {
|
||||
stmt := common.TxStmt(txn, s.selectCurrentStateStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectCurrentStateStmt)
|
||||
rows, err := stmt.QueryContext(ctx, roomID,
|
||||
pq.StringArray(stateFilter.Senders),
|
||||
pq.StringArray(stateFilter.NotSenders),
|
||||
|
@ -200,7 +200,7 @@ func (s *currentRoomStateStatements) SelectCurrentState(
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectCurrentState: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectCurrentState: rows.close() failed")
|
||||
|
||||
return rowsToEvents(rows)
|
||||
}
|
||||
|
@ -208,7 +208,7 @@ func (s *currentRoomStateStatements) SelectCurrentState(
|
|||
func (s *currentRoomStateStatements) DeleteRoomStateByEventID(
|
||||
ctx context.Context, txn *sql.Tx, eventID string,
|
||||
) error {
|
||||
stmt := common.TxStmt(txn, s.deleteRoomStateByEventIDStmt)
|
||||
stmt := internal.TxStmt(txn, s.deleteRoomStateByEventIDStmt)
|
||||
_, err := stmt.ExecContext(ctx, eventID)
|
||||
return err
|
||||
}
|
||||
|
@ -231,7 +231,7 @@ func (s *currentRoomStateStatements) UpsertRoomState(
|
|||
}
|
||||
|
||||
// upsert state event
|
||||
stmt := common.TxStmt(txn, s.upsertRoomStateStmt)
|
||||
stmt := internal.TxStmt(txn, s.upsertRoomStateStmt)
|
||||
_, err = stmt.ExecContext(
|
||||
ctx,
|
||||
event.RoomID(),
|
||||
|
@ -250,12 +250,12 @@ func (s *currentRoomStateStatements) UpsertRoomState(
|
|||
func (s *currentRoomStateStatements) SelectEventsWithEventIDs(
|
||||
ctx context.Context, txn *sql.Tx, eventIDs []string,
|
||||
) ([]types.StreamEvent, error) {
|
||||
stmt := common.TxStmt(txn, s.selectEventsWithEventIDsStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectEventsWithEventIDsStmt)
|
||||
rows, err := stmt.QueryContext(ctx, pq.StringArray(eventIDs))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectEventsWithEventIDs: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectEventsWithEventIDs: rows.close() failed")
|
||||
return rowsToStreamEvents(rows)
|
||||
}
|
||||
|
||||
|
|
|
@ -20,7 +20,7 @@ import (
|
|||
"database/sql"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
@ -119,12 +119,12 @@ func (s *inviteEventsStatements) DeleteInviteEvent(
|
|||
func (s *inviteEventsStatements) SelectInviteEventsInRange(
|
||||
ctx context.Context, txn *sql.Tx, targetUserID string, r types.Range,
|
||||
) (map[string]gomatrixserverlib.HeaderedEvent, error) {
|
||||
stmt := common.TxStmt(txn, s.selectInviteEventsInRangeStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectInviteEventsInRangeStmt)
|
||||
rows, err := stmt.QueryContext(ctx, targetUserID, r.Low(), r.High())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectInviteEventsInRange: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectInviteEventsInRange: rows.close() failed")
|
||||
result := map[string]gomatrixserverlib.HeaderedEvent{}
|
||||
for rows.Next() {
|
||||
var (
|
||||
|
@ -149,7 +149,7 @@ func (s *inviteEventsStatements) SelectMaxInviteID(
|
|||
ctx context.Context, txn *sql.Tx,
|
||||
) (id int64, err error) {
|
||||
var nullableID sql.NullInt64
|
||||
stmt := common.TxStmt(txn, s.selectMaxInviteIDStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectMaxInviteIDStmt)
|
||||
err = stmt.QueryRowContext(ctx).Scan(&nullableID)
|
||||
if nullableID.Valid {
|
||||
id = nullableID.Int64
|
||||
|
|
|
@ -26,7 +26,7 @@ import (
|
|||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
|
||||
"github.com/lib/pq"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
@ -158,7 +158,7 @@ func (s *outputRoomEventsStatements) SelectStateInRange(
|
|||
ctx context.Context, txn *sql.Tx, r types.Range,
|
||||
stateFilter *gomatrixserverlib.StateFilter,
|
||||
) (map[string]map[string]bool, map[string]types.StreamEvent, error) {
|
||||
stmt := common.TxStmt(txn, s.selectStateInRangeStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectStateInRangeStmt)
|
||||
|
||||
rows, err := stmt.QueryContext(
|
||||
ctx, r.Low(), r.High(),
|
||||
|
@ -172,7 +172,7 @@ func (s *outputRoomEventsStatements) SelectStateInRange(
|
|||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectStateInRange: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectStateInRange: rows.close() failed")
|
||||
// Fetch all the state change events for all rooms between the two positions then loop each event and:
|
||||
// - Keep a cache of the event by ID (99% of state change events are for the event itself)
|
||||
// - For each room ID, build up an array of event IDs which represents cumulative adds/removes
|
||||
|
@ -239,7 +239,7 @@ func (s *outputRoomEventsStatements) SelectMaxEventID(
|
|||
ctx context.Context, txn *sql.Tx,
|
||||
) (id int64, err error) {
|
||||
var nullableID sql.NullInt64
|
||||
stmt := common.TxStmt(txn, s.selectMaxEventIDStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectMaxEventIDStmt)
|
||||
err = stmt.QueryRowContext(ctx).Scan(&nullableID)
|
||||
if nullableID.Valid {
|
||||
id = nullableID.Int64
|
||||
|
@ -275,7 +275,7 @@ func (s *outputRoomEventsStatements) InsertEvent(
|
|||
return
|
||||
}
|
||||
|
||||
stmt := common.TxStmt(txn, s.insertEventStmt)
|
||||
stmt := internal.TxStmt(txn, s.insertEventStmt)
|
||||
err = stmt.QueryRowContext(
|
||||
ctx,
|
||||
event.RoomID(),
|
||||
|
@ -303,15 +303,15 @@ func (s *outputRoomEventsStatements) SelectRecentEvents(
|
|||
) ([]types.StreamEvent, error) {
|
||||
var stmt *sql.Stmt
|
||||
if onlySyncEvents {
|
||||
stmt = common.TxStmt(txn, s.selectRecentEventsForSyncStmt)
|
||||
stmt = internal.TxStmt(txn, s.selectRecentEventsForSyncStmt)
|
||||
} else {
|
||||
stmt = common.TxStmt(txn, s.selectRecentEventsStmt)
|
||||
stmt = internal.TxStmt(txn, s.selectRecentEventsStmt)
|
||||
}
|
||||
rows, err := stmt.QueryContext(ctx, roomID, r.Low(), r.High(), limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectRecentEvents: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectRecentEvents: rows.close() failed")
|
||||
events, err := rowsToStreamEvents(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
@ -333,12 +333,12 @@ func (s *outputRoomEventsStatements) SelectEarlyEvents(
|
|||
ctx context.Context, txn *sql.Tx,
|
||||
roomID string, r types.Range, limit int,
|
||||
) ([]types.StreamEvent, error) {
|
||||
stmt := common.TxStmt(txn, s.selectEarlyEventsStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectEarlyEventsStmt)
|
||||
rows, err := stmt.QueryContext(ctx, roomID, r.Low(), r.High(), limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectEarlyEvents: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectEarlyEvents: rows.close() failed")
|
||||
events, err := rowsToStreamEvents(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
@ -357,12 +357,12 @@ func (s *outputRoomEventsStatements) SelectEarlyEvents(
|
|||
func (s *outputRoomEventsStatements) SelectEvents(
|
||||
ctx context.Context, txn *sql.Tx, eventIDs []string,
|
||||
) ([]types.StreamEvent, error) {
|
||||
stmt := common.TxStmt(txn, s.selectEventsStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectEventsStmt)
|
||||
rows, err := stmt.QueryContext(ctx, pq.StringArray(eventIDs))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectEvents: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectEvents: rows.close() failed")
|
||||
return rowsToStreamEvents(rows)
|
||||
}
|
||||
|
||||
|
|
|
@ -18,7 +18,7 @@ import (
|
|||
"context"
|
||||
"database/sql"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
|
@ -139,7 +139,7 @@ func (s *outputRoomEventsTopologyStatements) SelectEventIDsInRange(
|
|||
} else if err != nil {
|
||||
return
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectEventIDsInRange: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectEventIDsInRange: rows.close() failed")
|
||||
|
||||
// Return the IDs.
|
||||
var eventID string
|
||||
|
|
|
@ -22,8 +22,8 @@ import (
|
|||
|
||||
// Import the postgres database driver.
|
||||
_ "github.com/lib/pq"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/eduserver/cache"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/shared"
|
||||
)
|
||||
|
||||
|
@ -32,11 +32,11 @@ import (
|
|||
type SyncServerDatasource struct {
|
||||
shared.Database
|
||||
db *sql.DB
|
||||
common.PartitionOffsetStatements
|
||||
internal.PartitionOffsetStatements
|
||||
}
|
||||
|
||||
// NewDatabase creates a new sync server database
|
||||
func NewDatabase(dbDataSourceName string, dbProperties common.DbProperties) (*SyncServerDatasource, error) {
|
||||
func NewDatabase(dbDataSourceName string, dbProperties internal.DbProperties) (*SyncServerDatasource, error) {
|
||||
var d SyncServerDatasource
|
||||
var err error
|
||||
if d.db, err = sqlutil.Open("postgres", dbDataSourceName, dbProperties); err != nil {
|
||||
|
|
|
@ -8,8 +8,8 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/eduserver/cache"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
|
@ -106,7 +106,7 @@ func (d *Database) GetStateEvent(
|
|||
func (d *Database) GetStateEventsForRoom(
|
||||
ctx context.Context, roomID string, stateFilter *gomatrixserverlib.StateFilter,
|
||||
) (stateEvents []gomatrixserverlib.HeaderedEvent, err error) {
|
||||
err = common.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
err = internal.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
stateEvents, err = d.CurrentRoomState.SelectCurrentState(ctx, txn, roomID, stateFilter)
|
||||
return err
|
||||
})
|
||||
|
@ -116,7 +116,7 @@ func (d *Database) GetStateEventsForRoom(
|
|||
func (d *Database) SyncStreamPosition(ctx context.Context) (types.StreamPosition, error) {
|
||||
var maxID int64
|
||||
var err error
|
||||
err = common.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
err = internal.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
maxID, err = d.OutputEvents.SelectMaxEventID(ctx, txn)
|
||||
if err != nil {
|
||||
return err
|
||||
|
@ -148,7 +148,7 @@ func (d *Database) SyncStreamPosition(ctx context.Context) (types.StreamPosition
|
|||
func (d *Database) AddInviteEvent(
|
||||
ctx context.Context, inviteEvent gomatrixserverlib.HeaderedEvent,
|
||||
) (sp types.StreamPosition, err error) {
|
||||
err = common.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
err = internal.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
sp, err = d.Invites.InsertInviteEvent(ctx, txn, inviteEvent)
|
||||
return err
|
||||
})
|
||||
|
@ -187,7 +187,7 @@ func (d *Database) GetAccountDataInRange(
|
|||
func (d *Database) UpsertAccountData(
|
||||
ctx context.Context, userID, roomID, dataType string,
|
||||
) (sp types.StreamPosition, err error) {
|
||||
err = common.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
err = internal.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
sp, err = d.AccountData.InsertAccountData(ctx, txn, userID, roomID, dataType)
|
||||
return err
|
||||
})
|
||||
|
@ -255,7 +255,7 @@ func (d *Database) WriteEvent(
|
|||
addStateEventIDs, removeStateEventIDs []string,
|
||||
transactionID *api.TransactionID, excludeFromSync bool,
|
||||
) (pduPosition types.StreamPosition, returnErr error) {
|
||||
returnErr = common.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
returnErr = internal.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
var err error
|
||||
pos, err := d.OutputEvents.InsertEvent(
|
||||
ctx, txn, ev, addStateEventIDs, removeStateEventIDs, transactionID, excludeFromSync,
|
||||
|
@ -355,7 +355,7 @@ func (d *Database) GetEventsInTopologicalRange(
|
|||
}
|
||||
|
||||
func (d *Database) SyncPosition(ctx context.Context) (tok types.StreamingToken, err error) {
|
||||
err = common.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
err = internal.WithTransaction(d.DB, func(txn *sql.Tx) error {
|
||||
pos, err := d.syncPositionTx(ctx, txn)
|
||||
if err != nil {
|
||||
return err
|
||||
|
@ -434,7 +434,7 @@ func (d *Database) addPDUDeltaToResponse(
|
|||
}
|
||||
var succeeded bool
|
||||
defer func() {
|
||||
txerr := common.EndTransaction(txn, &succeeded)
|
||||
txerr := internal.EndTransaction(txn, &succeeded)
|
||||
if err == nil && txerr != nil {
|
||||
err = txerr
|
||||
}
|
||||
|
@ -588,7 +588,7 @@ func (d *Database) getResponseWithPDUsForCompleteSync(
|
|||
}
|
||||
var succeeded bool
|
||||
defer func() {
|
||||
txerr := common.EndTransaction(txn, &succeeded)
|
||||
txerr := internal.EndTransaction(txn, &succeeded)
|
||||
if err == nil && txerr != nil {
|
||||
err = txerr
|
||||
}
|
||||
|
|
|
@ -19,7 +19,7 @@ import (
|
|||
"context"
|
||||
"database/sql"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
|
@ -100,7 +100,7 @@ func (s *accountDataStatements) SelectAccountDataInRange(
|
|||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectAccountDataInRange: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectAccountDataInRange: rows.close() failed")
|
||||
|
||||
var entries int
|
||||
|
||||
|
|
|
@ -18,7 +18,7 @@ import (
|
|||
"context"
|
||||
"database/sql"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
)
|
||||
|
||||
|
@ -84,7 +84,7 @@ func (s *backwardExtremitiesStatements) SelectBackwardExtremitiesForRoom(
|
|||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectBackwardExtremitiesForRoom: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectBackwardExtremitiesForRoom: rows.close() failed")
|
||||
|
||||
bwExtrems = make(map[string][]string)
|
||||
for rows.Next() {
|
||||
|
|
|
@ -21,7 +21,7 @@ import (
|
|||
"encoding/json"
|
||||
"strings"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
@ -129,7 +129,7 @@ func (s *currentRoomStateStatements) SelectJoinedUsers(
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectJoinedUsers: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectJoinedUsers: rows.close() failed")
|
||||
|
||||
result := make(map[string][]string)
|
||||
for rows.Next() {
|
||||
|
@ -152,12 +152,12 @@ func (s *currentRoomStateStatements) SelectRoomIDsWithMembership(
|
|||
userID string,
|
||||
membership string, // nolint: unparam
|
||||
) ([]string, error) {
|
||||
stmt := common.TxStmt(txn, s.selectRoomIDsWithMembershipStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectRoomIDsWithMembershipStmt)
|
||||
rows, err := stmt.QueryContext(ctx, userID, membership)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectRoomIDsWithMembership: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectRoomIDsWithMembership: rows.close() failed")
|
||||
|
||||
var result []string
|
||||
for rows.Next() {
|
||||
|
@ -175,7 +175,7 @@ func (s *currentRoomStateStatements) SelectCurrentState(
|
|||
ctx context.Context, txn *sql.Tx, roomID string,
|
||||
stateFilterPart *gomatrixserverlib.StateFilter,
|
||||
) ([]gomatrixserverlib.HeaderedEvent, error) {
|
||||
stmt := common.TxStmt(txn, s.selectCurrentStateStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectCurrentStateStmt)
|
||||
rows, err := stmt.QueryContext(ctx, roomID,
|
||||
nil, // FIXME: pq.StringArray(stateFilterPart.Senders),
|
||||
nil, // FIXME: pq.StringArray(stateFilterPart.NotSenders),
|
||||
|
@ -187,7 +187,7 @@ func (s *currentRoomStateStatements) SelectCurrentState(
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectCurrentState: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectCurrentState: rows.close() failed")
|
||||
|
||||
return rowsToEvents(rows)
|
||||
}
|
||||
|
@ -195,7 +195,7 @@ func (s *currentRoomStateStatements) SelectCurrentState(
|
|||
func (s *currentRoomStateStatements) DeleteRoomStateByEventID(
|
||||
ctx context.Context, txn *sql.Tx, eventID string,
|
||||
) error {
|
||||
stmt := common.TxStmt(txn, s.deleteRoomStateByEventIDStmt)
|
||||
stmt := internal.TxStmt(txn, s.deleteRoomStateByEventIDStmt)
|
||||
_, err := stmt.ExecContext(ctx, eventID)
|
||||
return err
|
||||
}
|
||||
|
@ -218,7 +218,7 @@ func (s *currentRoomStateStatements) UpsertRoomState(
|
|||
}
|
||||
|
||||
// upsert state event
|
||||
stmt := common.TxStmt(txn, s.upsertRoomStateStmt)
|
||||
stmt := internal.TxStmt(txn, s.upsertRoomStateStmt)
|
||||
_, err = stmt.ExecContext(
|
||||
ctx,
|
||||
event.RoomID(),
|
||||
|
@ -241,12 +241,12 @@ func (s *currentRoomStateStatements) SelectEventsWithEventIDs(
|
|||
for k, v := range eventIDs {
|
||||
iEventIDs[k] = v
|
||||
}
|
||||
query := strings.Replace(selectEventsWithEventIDsSQL, "($1)", common.QueryVariadic(len(iEventIDs)), 1)
|
||||
query := strings.Replace(selectEventsWithEventIDsSQL, "($1)", internal.QueryVariadic(len(iEventIDs)), 1)
|
||||
rows, err := txn.QueryContext(ctx, query, iEventIDs...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectEventsWithEventIDs: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectEventsWithEventIDs: rows.close() failed")
|
||||
return rowsToStreamEvents(rows)
|
||||
}
|
||||
|
||||
|
|
|
@ -20,7 +20,7 @@ import (
|
|||
"database/sql"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
@ -123,12 +123,12 @@ func (s *inviteEventsStatements) DeleteInviteEvent(
|
|||
func (s *inviteEventsStatements) SelectInviteEventsInRange(
|
||||
ctx context.Context, txn *sql.Tx, targetUserID string, r types.Range,
|
||||
) (map[string]gomatrixserverlib.HeaderedEvent, error) {
|
||||
stmt := common.TxStmt(txn, s.selectInviteEventsInRangeStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectInviteEventsInRangeStmt)
|
||||
rows, err := stmt.QueryContext(ctx, targetUserID, r.Low(), r.High())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectInviteEventsInRange: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectInviteEventsInRange: rows.close() failed")
|
||||
result := map[string]gomatrixserverlib.HeaderedEvent{}
|
||||
for rows.Next() {
|
||||
var (
|
||||
|
@ -153,7 +153,7 @@ func (s *inviteEventsStatements) SelectMaxInviteID(
|
|||
ctx context.Context, txn *sql.Tx,
|
||||
) (id int64, err error) {
|
||||
var nullableID sql.NullInt64
|
||||
stmt := common.TxStmt(txn, s.selectMaxInviteIDStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectMaxInviteIDStmt)
|
||||
err = stmt.QueryRowContext(ctx).Scan(&nullableID)
|
||||
if nullableID.Valid {
|
||||
id = nullableID.Int64
|
||||
|
|
|
@ -25,7 +25,7 @@ import (
|
|||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
log "github.com/sirupsen/logrus"
|
||||
)
|
||||
|
@ -149,7 +149,7 @@ func (s *outputRoomEventsStatements) SelectStateInRange(
|
|||
ctx context.Context, txn *sql.Tx, r types.Range,
|
||||
stateFilterPart *gomatrixserverlib.StateFilter,
|
||||
) (map[string]map[string]bool, map[string]types.StreamEvent, error) {
|
||||
stmt := common.TxStmt(txn, s.selectStateInRangeStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectStateInRangeStmt)
|
||||
|
||||
rows, err := stmt.QueryContext(
|
||||
ctx, r.Low(), r.High(),
|
||||
|
@ -236,7 +236,7 @@ func (s *outputRoomEventsStatements) SelectMaxEventID(
|
|||
ctx context.Context, txn *sql.Tx,
|
||||
) (id int64, err error) {
|
||||
var nullableID sql.NullInt64
|
||||
stmt := common.TxStmt(txn, s.selectMaxEventIDStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectMaxEventIDStmt)
|
||||
err = stmt.QueryRowContext(ctx).Scan(&nullableID)
|
||||
if nullableID.Valid {
|
||||
id = nullableID.Int64
|
||||
|
@ -286,7 +286,7 @@ func (s *outputRoomEventsStatements) InsertEvent(
|
|||
return
|
||||
}
|
||||
|
||||
insertStmt := common.TxStmt(txn, s.insertEventStmt)
|
||||
insertStmt := internal.TxStmt(txn, s.insertEventStmt)
|
||||
_, err = insertStmt.ExecContext(
|
||||
ctx,
|
||||
streamPos,
|
||||
|
@ -313,16 +313,16 @@ func (s *outputRoomEventsStatements) SelectRecentEvents(
|
|||
) ([]types.StreamEvent, error) {
|
||||
var stmt *sql.Stmt
|
||||
if onlySyncEvents {
|
||||
stmt = common.TxStmt(txn, s.selectRecentEventsForSyncStmt)
|
||||
stmt = internal.TxStmt(txn, s.selectRecentEventsForSyncStmt)
|
||||
} else {
|
||||
stmt = common.TxStmt(txn, s.selectRecentEventsStmt)
|
||||
stmt = internal.TxStmt(txn, s.selectRecentEventsStmt)
|
||||
}
|
||||
|
||||
rows, err := stmt.QueryContext(ctx, roomID, r.Low(), r.High(), limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectRecentEvents: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectRecentEvents: rows.close() failed")
|
||||
events, err := rowsToStreamEvents(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
@ -342,12 +342,12 @@ func (s *outputRoomEventsStatements) SelectEarlyEvents(
|
|||
ctx context.Context, txn *sql.Tx,
|
||||
roomID string, r types.Range, limit int,
|
||||
) ([]types.StreamEvent, error) {
|
||||
stmt := common.TxStmt(txn, s.selectEarlyEventsStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectEarlyEventsStmt)
|
||||
rows, err := stmt.QueryContext(ctx, roomID, r.Low(), r.High(), limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer common.CloseAndLogIfError(ctx, rows, "selectEarlyEvents: rows.close() failed")
|
||||
defer internal.CloseAndLogIfError(ctx, rows, "selectEarlyEvents: rows.close() failed")
|
||||
events, err := rowsToStreamEvents(rows)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
@ -367,7 +367,7 @@ func (s *outputRoomEventsStatements) SelectEvents(
|
|||
ctx context.Context, txn *sql.Tx, eventIDs []string,
|
||||
) ([]types.StreamEvent, error) {
|
||||
var returnEvents []types.StreamEvent
|
||||
stmt := common.TxStmt(txn, s.selectEventsStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectEventsStmt)
|
||||
for _, eventID := range eventIDs {
|
||||
rows, err := stmt.QueryContext(ctx, eventID)
|
||||
if err != nil {
|
||||
|
@ -376,7 +376,7 @@ func (s *outputRoomEventsStatements) SelectEvents(
|
|||
if streamEvents, err := rowsToStreamEvents(rows); err == nil {
|
||||
returnEvents = append(returnEvents, streamEvents...)
|
||||
}
|
||||
common.CloseAndLogIfError(ctx, rows, "selectEvents: rows.close() failed")
|
||||
internal.CloseAndLogIfError(ctx, rows, "selectEvents: rows.close() failed")
|
||||
}
|
||||
return returnEvents, nil
|
||||
}
|
||||
|
|
|
@ -18,7 +18,7 @@ import (
|
|||
"context"
|
||||
"database/sql"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/tables"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
@ -102,7 +102,7 @@ func NewSqliteTopologyTable(db *sql.DB) (tables.Topology, error) {
|
|||
func (s *outputRoomEventsTopologyStatements) InsertEventInTopology(
|
||||
ctx context.Context, txn *sql.Tx, event *gomatrixserverlib.HeaderedEvent, pos types.StreamPosition,
|
||||
) (err error) {
|
||||
stmt := common.TxStmt(txn, s.insertEventInTopologyStmt)
|
||||
stmt := internal.TxStmt(txn, s.insertEventInTopologyStmt)
|
||||
_, err = stmt.ExecContext(
|
||||
ctx, event.EventID(), event.Depth(), event.RoomID(), pos,
|
||||
)
|
||||
|
@ -118,9 +118,9 @@ func (s *outputRoomEventsTopologyStatements) SelectEventIDsInRange(
|
|||
// is requested or not.
|
||||
var stmt *sql.Stmt
|
||||
if chronologicalOrder {
|
||||
stmt = common.TxStmt(txn, s.selectEventIDsInRangeASCStmt)
|
||||
stmt = internal.TxStmt(txn, s.selectEventIDsInRangeASCStmt)
|
||||
} else {
|
||||
stmt = common.TxStmt(txn, s.selectEventIDsInRangeDESCStmt)
|
||||
stmt = internal.TxStmt(txn, s.selectEventIDsInRangeDESCStmt)
|
||||
}
|
||||
|
||||
// Query the event IDs.
|
||||
|
@ -149,7 +149,7 @@ func (s *outputRoomEventsTopologyStatements) SelectEventIDsInRange(
|
|||
func (s *outputRoomEventsTopologyStatements) SelectPositionInTopology(
|
||||
ctx context.Context, txn *sql.Tx, eventID string,
|
||||
) (pos types.StreamPosition, spos types.StreamPosition, err error) {
|
||||
stmt := common.TxStmt(txn, s.selectPositionInTopologyStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectPositionInTopologyStmt)
|
||||
err = stmt.QueryRowContext(ctx, eventID).Scan(&pos, &spos)
|
||||
return
|
||||
}
|
||||
|
@ -157,7 +157,7 @@ func (s *outputRoomEventsTopologyStatements) SelectPositionInTopology(
|
|||
func (s *outputRoomEventsTopologyStatements) SelectMaxPositionInTopology(
|
||||
ctx context.Context, txn *sql.Tx, roomID string,
|
||||
) (pos types.StreamPosition, spos types.StreamPosition, err error) {
|
||||
stmt := common.TxStmt(txn, s.selectMaxPositionInTopologyStmt)
|
||||
stmt := internal.TxStmt(txn, s.selectMaxPositionInTopologyStmt)
|
||||
err = stmt.QueryRowContext(ctx, roomID).Scan(&pos, &spos)
|
||||
return
|
||||
}
|
||||
|
|
|
@ -4,7 +4,7 @@ import (
|
|||
"context"
|
||||
"database/sql"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/types"
|
||||
)
|
||||
|
||||
|
@ -46,8 +46,8 @@ func (s *streamIDStatements) prepare(db *sql.DB) (err error) {
|
|||
}
|
||||
|
||||
func (s *streamIDStatements) nextStreamID(ctx context.Context, txn *sql.Tx) (pos types.StreamPosition, err error) {
|
||||
increaseStmt := common.TxStmt(txn, s.increaseStreamIDStmt)
|
||||
selectStmt := common.TxStmt(txn, s.selectStreamIDStmt)
|
||||
increaseStmt := internal.TxStmt(txn, s.increaseStreamIDStmt)
|
||||
selectStmt := internal.TxStmt(txn, s.selectStreamIDStmt)
|
||||
if _, err = increaseStmt.ExecContext(ctx, "global"); err != nil {
|
||||
return
|
||||
}
|
||||
|
|
|
@ -25,8 +25,8 @@ import (
|
|||
// Import the sqlite3 package
|
||||
_ "github.com/mattn/go-sqlite3"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/eduserver/cache"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/shared"
|
||||
)
|
||||
|
||||
|
@ -35,7 +35,7 @@ import (
|
|||
type SyncServerDatasource struct {
|
||||
shared.Database
|
||||
db *sql.DB
|
||||
common.PartitionOffsetStatements
|
||||
internal.PartitionOffsetStatements
|
||||
streamID streamIDStatements
|
||||
}
|
||||
|
||||
|
@ -55,7 +55,7 @@ func NewDatabase(dataSourceName string) (*SyncServerDatasource, error) {
|
|||
} else {
|
||||
return nil, errors.New("no filename or path in connect string")
|
||||
}
|
||||
if d.db, err = sqlutil.Open(common.SQLiteDriverName(), cs, nil); err != nil {
|
||||
if d.db, err = sqlutil.Open(internal.SQLiteDriverName(), cs, nil); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err = d.prepare(); err != nil {
|
||||
|
|
|
@ -19,13 +19,13 @@ package storage
|
|||
import (
|
||||
"net/url"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/postgres"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/sqlite3"
|
||||
)
|
||||
|
||||
// NewSyncServerDatasource opens a database connection.
|
||||
func NewSyncServerDatasource(dataSourceName string, dbProperties common.DbProperties) (Database, error) {
|
||||
func NewSyncServerDatasource(dataSourceName string, dbProperties internal.DbProperties) (Database, error) {
|
||||
uri, err := url.Parse(dataSourceName)
|
||||
if err != nil {
|
||||
return postgres.NewDatabase(dataSourceName, dbProperties)
|
||||
|
|
|
@ -18,14 +18,14 @@ import (
|
|||
"fmt"
|
||||
"net/url"
|
||||
|
||||
"github.com/matrix-org/dendrite/common"
|
||||
"github.com/matrix-org/dendrite/internal"
|
||||
"github.com/matrix-org/dendrite/syncapi/storage/sqlite3"
|
||||
)
|
||||
|
||||
// NewPublicRoomsServerDatabase opens a database connection.
|
||||
func NewSyncServerDatasource(
|
||||
dataSourceName string,
|
||||
dbProperties common.DbProperties, // nolint:unparam
|
||||
dbProperties internal.DbProperties, // nolint:unparam
|
||||
) (Database, error) {
|
||||
uri, err := url.Parse(dataSourceName)
|
||||
if err != nil {
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue