Merge both user API databases into one (#2186)

* Merge user API databases into one

* Remove DeviceDatabase from config

* Fix tests

* Try that again

* Clean up keyserver device keys when the devices no longer exist in the user API

* Tweak ordering

* Fix UserExists flag, device check

* Allow including empty entries so we can clean them up

* Remove logging

userapi/storage/postgres/account_data_table.go

@@ -0,0 +1,130 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+)
+
+const accountDataSchema = `
+-- Stores data about accounts data.
+CREATE TABLE IF NOT EXISTS account_data (
+    -- The Matrix user ID localpart for this account
+    localpart TEXT NOT NULL,
+    -- The room ID for this data (empty string if not specific to a room)
+    room_id TEXT,
+    -- The account data type
+    type TEXT NOT NULL,
+    -- The account data content
+    content TEXT NOT NULL,
+
+    PRIMARY KEY(localpart, room_id, type)
+);
+`
+
+const insertAccountDataSQL = `
+	INSERT INTO account_data(localpart, room_id, type, content) VALUES($1, $2, $3, $4)
+	ON CONFLICT (localpart, room_id, type) DO UPDATE SET content = EXCLUDED.content
+`
+
+const selectAccountDataSQL = "" +
+	"SELECT room_id, type, content FROM account_data WHERE localpart = $1"
+
+const selectAccountDataByTypeSQL = "" +
+	"SELECT content FROM account_data WHERE localpart = $1 AND room_id = $2 AND type = $3"
+
+type accountDataStatements struct {
+	insertAccountDataStmt       *sql.Stmt
+	selectAccountDataStmt       *sql.Stmt
+	selectAccountDataByTypeStmt *sql.Stmt
+}
+
+func (s *accountDataStatements) prepare(db *sql.DB) (err error) {
+	_, err = db.Exec(accountDataSchema)
+	if err != nil {
+		return
+	}
+	return sqlutil.StatementList{
+		{&s.insertAccountDataStmt, insertAccountDataSQL},
+		{&s.selectAccountDataStmt, selectAccountDataSQL},
+		{&s.selectAccountDataByTypeStmt, selectAccountDataByTypeSQL},
+	}.Prepare(db)
+}
+
+func (s *accountDataStatements) insertAccountData(
+	ctx context.Context, txn *sql.Tx, localpart, roomID, dataType string, content json.RawMessage,
+) (err error) {
+	stmt := sqlutil.TxStmt(txn, s.insertAccountDataStmt)
+	_, err = stmt.ExecContext(ctx, localpart, roomID, dataType, content)
+	return
+}
+
+func (s *accountDataStatements) selectAccountData(
+	ctx context.Context, localpart string,
+) (
+	/* global */ map[string]json.RawMessage,
+	/* rooms */ map[string]map[string]json.RawMessage,
+	error,
+) {
+	rows, err := s.selectAccountDataStmt.QueryContext(ctx, localpart)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer internal.CloseAndLogIfError(ctx, rows, "selectAccountData: rows.close() failed")
+
+	global := map[string]json.RawMessage{}
+	rooms := map[string]map[string]json.RawMessage{}
+
+	for rows.Next() {
+		var roomID string
+		var dataType string
+		var content []byte
+
+		if err = rows.Scan(&roomID, &dataType, &content); err != nil {
+			return nil, nil, err
+		}
+
+		if roomID != "" {
+			if _, ok := rooms[roomID]; !ok {
+				rooms[roomID] = map[string]json.RawMessage{}
+			}
+			rooms[roomID][dataType] = content
+		} else {
+			global[dataType] = content
+		}
+	}
+
+	return global, rooms, rows.Err()
+}
+
+func (s *accountDataStatements) selectAccountDataByType(
+	ctx context.Context, localpart, roomID, dataType string,
+) (data json.RawMessage, err error) {
+	var bytes []byte
+	stmt := s.selectAccountDataByTypeStmt
+	if err = stmt.QueryRowContext(ctx, localpart, roomID, dataType).Scan(&bytes); err != nil {
+		if err == sql.ErrNoRows {
+			return nil, nil
+		}
+		return
+	}
+	data = json.RawMessage(bytes)
+	return
+}

userapi/storage/postgres/accounts_table.go

@@ -0,0 +1,180 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"time"
+
+	"github.com/matrix-org/gomatrixserverlib"
+
+	"github.com/matrix-org/dendrite/clientapi/userutil"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/userapi/api"
+
+	log "github.com/sirupsen/logrus"
+)
+
+const accountsSchema = `
+-- Stores data about accounts.
+CREATE TABLE IF NOT EXISTS account_accounts (
+    -- The Matrix user ID localpart for this account
+    localpart TEXT NOT NULL PRIMARY KEY,
+    -- When this account was first created, as a unix timestamp (ms resolution).
+    created_ts BIGINT NOT NULL,
+    -- The password hash for this account. Can be NULL if this is a passwordless account.
+    password_hash TEXT,
+    -- Identifies which application service this account belongs to, if any.
+    appservice_id TEXT,
+    -- If the account is currently active
+    is_deactivated BOOLEAN DEFAULT FALSE,
+	-- The account_type (user = 1, guest = 2, admin = 3, appservice = 4)
+	account_type SMALLINT NOT NULL
+    -- TODO:
+    -- upgraded_ts, devices, any email reset stuff?
+);
+-- Create sequence for autogenerated numeric usernames
+CREATE SEQUENCE IF NOT EXISTS numeric_username_seq START 1;
+`
+
+const insertAccountSQL = "" +
+	"INSERT INTO account_accounts(localpart, created_ts, password_hash, appservice_id, account_type) VALUES ($1, $2, $3, $4, $5)"
+
+const updatePasswordSQL = "" +
+	"UPDATE account_accounts SET password_hash = $1 WHERE localpart = $2"
+
+const deactivateAccountSQL = "" +
+	"UPDATE account_accounts SET is_deactivated = TRUE WHERE localpart = $1"
+
+const selectAccountByLocalpartSQL = "" +
+	"SELECT localpart, appservice_id, account_type FROM account_accounts WHERE localpart = $1"
+
+const selectPasswordHashSQL = "" +
+	"SELECT password_hash FROM account_accounts WHERE localpart = $1 AND is_deactivated = FALSE"
+
+const selectNewNumericLocalpartSQL = "" +
+	"SELECT nextval('numeric_username_seq')"
+
+type accountsStatements struct {
+	insertAccountStmt             *sql.Stmt
+	updatePasswordStmt            *sql.Stmt
+	deactivateAccountStmt         *sql.Stmt
+	selectAccountByLocalpartStmt  *sql.Stmt
+	selectPasswordHashStmt        *sql.Stmt
+	selectNewNumericLocalpartStmt *sql.Stmt
+	serverName                    gomatrixserverlib.ServerName
+}
+
+func (s *accountsStatements) execSchema(db *sql.DB) error {
+	_, err := db.Exec(accountsSchema)
+	return err
+}
+
+func (s *accountsStatements) prepare(db *sql.DB, server gomatrixserverlib.ServerName) (err error) {
+	s.serverName = server
+	return sqlutil.StatementList{
+		{&s.insertAccountStmt, insertAccountSQL},
+		{&s.updatePasswordStmt, updatePasswordSQL},
+		{&s.deactivateAccountStmt, deactivateAccountSQL},
+		{&s.selectAccountByLocalpartStmt, selectAccountByLocalpartSQL},
+		{&s.selectPasswordHashStmt, selectPasswordHashSQL},
+		{&s.selectNewNumericLocalpartStmt, selectNewNumericLocalpartSQL},
+	}.Prepare(db)
+}
+
+// insertAccount creates a new account. 'hash' should be the password hash for this account. If it is missing,
+// this account will be passwordless. Returns an error if this account already exists. Returns the account
+// on success.
+func (s *accountsStatements) insertAccount(
+	ctx context.Context, txn *sql.Tx, localpart, hash, appserviceID string, accountType api.AccountType,
+) (*api.Account, error) {
+	createdTimeMS := time.Now().UnixNano() / 1000000
+	stmt := sqlutil.TxStmt(txn, s.insertAccountStmt)
+
+	var err error
+	if accountType != api.AccountTypeAppService {
+		_, err = stmt.ExecContext(ctx, localpart, createdTimeMS, hash, nil, accountType)
+	} else {
+		_, err = stmt.ExecContext(ctx, localpart, createdTimeMS, hash, appserviceID, accountType)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return &api.Account{
+		Localpart:    localpart,
+		UserID:       userutil.MakeUserID(localpart, s.serverName),
+		ServerName:   s.serverName,
+		AppServiceID: appserviceID,
+		AccountType:  accountType,
+	}, nil
+}
+
+func (s *accountsStatements) updatePassword(
+	ctx context.Context, localpart, passwordHash string,
+) (err error) {
+	_, err = s.updatePasswordStmt.ExecContext(ctx, passwordHash, localpart)
+	return
+}
+
+func (s *accountsStatements) deactivateAccount(
+	ctx context.Context, localpart string,
+) (err error) {
+	_, err = s.deactivateAccountStmt.ExecContext(ctx, localpart)
+	return
+}
+
+func (s *accountsStatements) selectPasswordHash(
+	ctx context.Context, localpart string,
+) (hash string, err error) {
+	err = s.selectPasswordHashStmt.QueryRowContext(ctx, localpart).Scan(&hash)
+	return
+}
+
+func (s *accountsStatements) selectAccountByLocalpart(
+	ctx context.Context, localpart string,
+) (*api.Account, error) {
+	var appserviceIDPtr sql.NullString
+	var acc api.Account
+
+	stmt := s.selectAccountByLocalpartStmt
+	err := stmt.QueryRowContext(ctx, localpart).Scan(&acc.Localpart, &appserviceIDPtr, &acc.AccountType)
+	if err != nil {
+		if err != sql.ErrNoRows {
+			log.WithError(err).Error("Unable to retrieve user from the db")
+		}
+		return nil, err
+	}
+	if appserviceIDPtr.Valid {
+		acc.AppServiceID = appserviceIDPtr.String
+	}
+
+	acc.UserID = userutil.MakeUserID(localpart, s.serverName)
+	acc.ServerName = s.serverName
+
+	return &acc, nil
+}
+
+func (s *accountsStatements) selectNewNumericLocalpart(
+	ctx context.Context, txn *sql.Tx,
+) (id int64, err error) {
+	stmt := s.selectNewNumericLocalpartStmt
+	if txn != nil {
+		stmt = sqlutil.TxStmt(txn, stmt)
+	}
+	err = stmt.QueryRowContext(ctx).Scan(&id)
+	return
+}

userapi/storage/postgres/deltas/20200929203058_is_active.go

@@ -0,0 +1,35 @@
+package deltas
+
+import (
+	"database/sql"
+	"fmt"
+
+	"github.com/pressly/goose"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+)
+
+func LoadFromGoose() {
+	goose.AddMigration(UpIsActive, DownIsActive)
+	goose.AddMigration(UpAddAccountType, DownAddAccountType)
+}
+
+func LoadIsActive(m *sqlutil.Migrations) {
+	m.AddMigration(UpIsActive, DownIsActive)
+}
+
+func UpIsActive(tx *sql.Tx) error {
+	_, err := tx.Exec("ALTER TABLE account_accounts ADD COLUMN IF NOT EXISTS is_deactivated BOOLEAN DEFAULT FALSE;")
+	if err != nil {
+		return fmt.Errorf("failed to execute upgrade: %w", err)
+	}
+	return nil
+}
+
+func DownIsActive(tx *sql.Tx) error {
+	_, err := tx.Exec("ALTER TABLE account_accounts DROP COLUMN is_deactivated;")
+	if err != nil {
+		return fmt.Errorf("failed to execute downgrade: %w", err)
+	}
+	return nil
+}

userapi/storage/postgres/deltas/20201001204705_last_seen_ts_ip.go

@@ -0,0 +1,34 @@
+package deltas
+
+import (
+	"database/sql"
+	"fmt"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+)
+
+func LoadLastSeenTSIP(m *sqlutil.Migrations) {
+	m.AddMigration(UpLastSeenTSIP, DownLastSeenTSIP)
+}
+
+func UpLastSeenTSIP(tx *sql.Tx) error {
+	_, err := tx.Exec(`
+ALTER TABLE device_devices ADD COLUMN IF NOT EXISTS last_seen_ts BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)*1000;
+ALTER TABLE device_devices ADD COLUMN IF NOT EXISTS ip TEXT;
+ALTER TABLE device_devices ADD COLUMN IF NOT EXISTS user_agent TEXT;`)
+	if err != nil {
+		return fmt.Errorf("failed to execute upgrade: %w", err)
+	}
+	return nil
+}
+
+func DownLastSeenTSIP(tx *sql.Tx) error {
+	_, err := tx.Exec(`
+	ALTER TABLE device_devices DROP COLUMN last_seen_ts;
+	ALTER TABLE device_devices DROP COLUMN ip;
+	ALTER TABLE device_devices DROP COLUMN user_agent;`)
+	if err != nil {
+		return fmt.Errorf("failed to execute downgrade: %w", err)
+	}
+	return nil
+}

userapi/storage/postgres/deltas/2022021013023800_add_account_type.go

@@ -0,0 +1,34 @@
+package deltas
+
+import (
+	"database/sql"
+	"fmt"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+)
+
+func LoadAddAccountType(m *sqlutil.Migrations) {
+	m.AddMigration(UpAddAccountType, DownAddAccountType)
+}
+
+func UpAddAccountType(tx *sql.Tx) error {
+	// initially set every account to useraccount, change appservice and guest accounts afterwards
+	// (user = 1, guest = 2, admin = 3, appservice = 4)
+	_, err := tx.Exec(`ALTER TABLE account_accounts ADD COLUMN IF NOT EXISTS account_type SMALLINT NOT NULL DEFAULT 1;
+UPDATE account_accounts SET account_type = 4 WHERE appservice_id <> '';
+UPDATE account_accounts SET account_type = 2 WHERE localpart ~ '^[0-9]+$';
+ALTER TABLE account_accounts ALTER COLUMN account_type DROP DEFAULT;`,
+	)
+	if err != nil {
+		return fmt.Errorf("failed to execute upgrade: %w", err)
+	}
+	return nil
+}
+
+func DownAddAccountType(tx *sql.Tx) error {
+	_, err := tx.Exec("ALTER TABLE account_accounts DROP COLUMN account_type;")
+	if err != nil {
+		return fmt.Errorf("failed to execute downgrade: %w", err)
+	}
+	return nil
+}

userapi/storage/postgres/devices_table.go

@@ -0,0 +1,321 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"time"
+
+	"github.com/lib/pq"
+	"github.com/matrix-org/dendrite/clientapi/userutil"
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/userapi/api"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+const devicesSchema = `
+-- This sequence is used for automatic allocation of session_id.
+CREATE SEQUENCE IF NOT EXISTS device_session_id_seq START 1;
+
+-- Stores data about devices.
+CREATE TABLE IF NOT EXISTS device_devices (
+    -- The access token granted to this device. This has to be the primary key
+    -- so we can distinguish which device is making a given request.
+    access_token TEXT NOT NULL PRIMARY KEY,
+    -- The auto-allocated unique ID of the session identified by the access token.
+    -- This can be used as a secure substitution of the access token in situations
+    -- where data is associated with access tokens (e.g. transaction storage),
+    -- so we don't have to store users' access tokens everywhere.
+    session_id BIGINT NOT NULL DEFAULT nextval('device_session_id_seq'),
+    -- The device identifier. This only needs to uniquely identify a device for a given user, not globally.
+    -- access_tokens will be clobbered based on the device ID for a user.
+    device_id TEXT NOT NULL,
+    -- The Matrix user ID localpart for this device. This is preferable to storing the full user_id
+    -- as it is smaller, makes it clearer that we only manage devices for our own users, and may make
+    -- migration to different domain names easier.
+    localpart TEXT NOT NULL,
+    -- When this devices was first recognised on the network, as a unix timestamp (ms resolution).
+    created_ts BIGINT NOT NULL,
+    -- The display name, human friendlier than device_id and updatable
+    display_name TEXT,
+	-- The time the device was last used, as a unix timestamp (ms resolution).
+	last_seen_ts BIGINT NOT NULL,
+	-- The last seen IP address of this device
+	ip TEXT,
+	-- User agent of this device
+	user_agent TEXT
+                                          
+    -- TODO: device keys, device display names, token restrictions (if 3rd-party OAuth app)
+);
+
+-- Device IDs must be unique for a given user.
+CREATE UNIQUE INDEX IF NOT EXISTS device_localpart_id_idx ON device_devices(localpart, device_id);
+`
+
+const insertDeviceSQL = "" +
+	"INSERT INTO device_devices(device_id, localpart, access_token, created_ts, display_name, last_seen_ts, ip, user_agent) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)" +
+	" RETURNING session_id"
+
+const selectDeviceByTokenSQL = "" +
+	"SELECT session_id, device_id, localpart FROM device_devices WHERE access_token = $1"
+
+const selectDeviceByIDSQL = "" +
+	"SELECT display_name FROM device_devices WHERE localpart = $1 and device_id = $2"
+
+const selectDevicesByLocalpartSQL = "" +
+	"SELECT device_id, display_name, last_seen_ts, ip, user_agent FROM device_devices WHERE localpart = $1 AND device_id != $2"
+
+const updateDeviceNameSQL = "" +
+	"UPDATE device_devices SET display_name = $1 WHERE localpart = $2 AND device_id = $3"
+
+const deleteDeviceSQL = "" +
+	"DELETE FROM device_devices WHERE device_id = $1 AND localpart = $2"
+
+const deleteDevicesByLocalpartSQL = "" +
+	"DELETE FROM device_devices WHERE localpart = $1 AND device_id != $2"
+
+const deleteDevicesSQL = "" +
+	"DELETE FROM device_devices WHERE localpart = $1 AND device_id = ANY($2)"
+
+const selectDevicesByIDSQL = "" +
+	"SELECT device_id, localpart, display_name FROM device_devices WHERE device_id = ANY($1)"
+
+const updateDeviceLastSeen = "" +
+	"UPDATE device_devices SET last_seen_ts = $1, ip = $2 WHERE localpart = $3 AND device_id = $4"
+
+type devicesStatements struct {
+	insertDeviceStmt             *sql.Stmt
+	selectDeviceByTokenStmt      *sql.Stmt
+	selectDeviceByIDStmt         *sql.Stmt
+	selectDevicesByLocalpartStmt *sql.Stmt
+	selectDevicesByIDStmt        *sql.Stmt
+	updateDeviceNameStmt         *sql.Stmt
+	updateDeviceLastSeenStmt     *sql.Stmt
+	deleteDeviceStmt             *sql.Stmt
+	deleteDevicesByLocalpartStmt *sql.Stmt
+	deleteDevicesStmt            *sql.Stmt
+	serverName                   gomatrixserverlib.ServerName
+}
+
+func (s *devicesStatements) execSchema(db *sql.DB) error {
+	_, err := db.Exec(devicesSchema)
+	return err
+}
+
+func (s *devicesStatements) prepare(db *sql.DB, server gomatrixserverlib.ServerName) (err error) {
+	if err = s.execSchema(db); err != nil {
+		return
+	}
+	if s.insertDeviceStmt, err = db.Prepare(insertDeviceSQL); err != nil {
+		return
+	}
+	if s.selectDeviceByTokenStmt, err = db.Prepare(selectDeviceByTokenSQL); err != nil {
+		return
+	}
+	if s.selectDeviceByIDStmt, err = db.Prepare(selectDeviceByIDSQL); err != nil {
+		return
+	}
+	if s.selectDevicesByLocalpartStmt, err = db.Prepare(selectDevicesByLocalpartSQL); err != nil {
+		return
+	}
+	if s.updateDeviceNameStmt, err = db.Prepare(updateDeviceNameSQL); err != nil {
+		return
+	}
+	if s.deleteDeviceStmt, err = db.Prepare(deleteDeviceSQL); err != nil {
+		return
+	}
+	if s.deleteDevicesByLocalpartStmt, err = db.Prepare(deleteDevicesByLocalpartSQL); err != nil {
+		return
+	}
+	if s.deleteDevicesStmt, err = db.Prepare(deleteDevicesSQL); err != nil {
+		return
+	}
+	if s.selectDevicesByIDStmt, err = db.Prepare(selectDevicesByIDSQL); err != nil {
+		return
+	}
+	if s.updateDeviceLastSeenStmt, err = db.Prepare(updateDeviceLastSeen); err != nil {
+		return
+	}
+	s.serverName = server
+	return
+}
+
+// insertDevice creates a new device. Returns an error if any device with the same access token already exists.
+// Returns an error if the user already has a device with the given device ID.
+// Returns the device on success.
+func (s *devicesStatements) insertDevice(
+	ctx context.Context, txn *sql.Tx, id, localpart, accessToken string,
+	displayName *string, ipAddr, userAgent string,
+) (*api.Device, error) {
+	createdTimeMS := time.Now().UnixNano() / 1000000
+	var sessionID int64
+	stmt := sqlutil.TxStmt(txn, s.insertDeviceStmt)
+	if err := stmt.QueryRowContext(ctx, id, localpart, accessToken, createdTimeMS, displayName, createdTimeMS, ipAddr, userAgent).Scan(&sessionID); err != nil {
+		return nil, err
+	}
+	return &api.Device{
+		ID:          id,
+		UserID:      userutil.MakeUserID(localpart, s.serverName),
+		AccessToken: accessToken,
+		SessionID:   sessionID,
+		LastSeenTS:  createdTimeMS,
+		LastSeenIP:  ipAddr,
+		UserAgent:   userAgent,
+	}, nil
+}
+
+// deleteDevice removes a single device by id and user localpart.
+func (s *devicesStatements) deleteDevice(
+	ctx context.Context, txn *sql.Tx, id, localpart string,
+) error {
+	stmt := sqlutil.TxStmt(txn, s.deleteDeviceStmt)
+	_, err := stmt.ExecContext(ctx, id, localpart)
+	return err
+}
+
+// deleteDevices removes a single or multiple devices by ids and user localpart.
+// Returns an error if the execution failed.
+func (s *devicesStatements) deleteDevices(
+	ctx context.Context, txn *sql.Tx, localpart string, devices []string,
+) error {
+	stmt := sqlutil.TxStmt(txn, s.deleteDevicesStmt)
+	_, err := stmt.ExecContext(ctx, localpart, pq.Array(devices))
+	return err
+}
+
+// deleteDevicesByLocalpart removes all devices for the
+// given user localpart.
+func (s *devicesStatements) deleteDevicesByLocalpart(
+	ctx context.Context, txn *sql.Tx, localpart, exceptDeviceID string,
+) error {
+	stmt := sqlutil.TxStmt(txn, s.deleteDevicesByLocalpartStmt)
+	_, err := stmt.ExecContext(ctx, localpart, exceptDeviceID)
+	return err
+}
+
+func (s *devicesStatements) updateDeviceName(
+	ctx context.Context, txn *sql.Tx, localpart, deviceID string, displayName *string,
+) error {
+	stmt := sqlutil.TxStmt(txn, s.updateDeviceNameStmt)
+	_, err := stmt.ExecContext(ctx, displayName, localpart, deviceID)
+	return err
+}
+
+func (s *devicesStatements) selectDeviceByToken(
+	ctx context.Context, accessToken string,
+) (*api.Device, error) {
+	var dev api.Device
+	var localpart string
+	stmt := s.selectDeviceByTokenStmt
+	err := stmt.QueryRowContext(ctx, accessToken).Scan(&dev.SessionID, &dev.ID, &localpart)
+	if err == nil {
+		dev.UserID = userutil.MakeUserID(localpart, s.serverName)
+		dev.AccessToken = accessToken
+	}
+	return &dev, err
+}
+
+// selectDeviceByID retrieves a device from the database with the given user
+// localpart and deviceID
+func (s *devicesStatements) selectDeviceByID(
+	ctx context.Context, localpart, deviceID string,
+) (*api.Device, error) {
+	var dev api.Device
+	var displayName sql.NullString
+	stmt := s.selectDeviceByIDStmt
+	err := stmt.QueryRowContext(ctx, localpart, deviceID).Scan(&displayName)
+	if err == nil {
+		dev.ID = deviceID
+		dev.UserID = userutil.MakeUserID(localpart, s.serverName)
+		if displayName.Valid {
+			dev.DisplayName = displayName.String
+		}
+	}
+	return &dev, err
+}
+
+func (s *devicesStatements) selectDevicesByID(ctx context.Context, deviceIDs []string) ([]api.Device, error) {
+	rows, err := s.selectDevicesByIDStmt.QueryContext(ctx, pq.StringArray(deviceIDs))
+	if err != nil {
+		return nil, err
+	}
+	defer internal.CloseAndLogIfError(ctx, rows, "selectDevicesByID: rows.close() failed")
+	var devices []api.Device
+	for rows.Next() {
+		var dev api.Device
+		var localpart string
+		var displayName sql.NullString
+		if err := rows.Scan(&dev.ID, &localpart, &displayName); err != nil {
+			return nil, err
+		}
+		if displayName.Valid {
+			dev.DisplayName = displayName.String
+		}
+		dev.UserID = userutil.MakeUserID(localpart, s.serverName)
+		devices = append(devices, dev)
+	}
+	return devices, rows.Err()
+}
+
+func (s *devicesStatements) selectDevicesByLocalpart(
+	ctx context.Context, txn *sql.Tx, localpart, exceptDeviceID string,
+) ([]api.Device, error) {
+	devices := []api.Device{}
+	rows, err := sqlutil.TxStmt(txn, s.selectDevicesByLocalpartStmt).QueryContext(ctx, localpart, exceptDeviceID)
+
+	if err != nil {
+		return devices, err
+	}
+	defer internal.CloseAndLogIfError(ctx, rows, "selectDevicesByLocalpart: rows.close() failed")
+
+	for rows.Next() {
+		var dev api.Device
+		var lastseents sql.NullInt64
+		var id, displayname, ip, useragent sql.NullString
+		err = rows.Scan(&id, &displayname, &lastseents, &ip, &useragent)
+		if err != nil {
+			return devices, err
+		}
+		if id.Valid {
+			dev.ID = id.String
+		}
+		if displayname.Valid {
+			dev.DisplayName = displayname.String
+		}
+		if lastseents.Valid {
+			dev.LastSeenTS = lastseents.Int64
+		}
+		if ip.Valid {
+			dev.LastSeenIP = ip.String
+		}
+		if useragent.Valid {
+			dev.UserAgent = useragent.String
+		}
+
+		dev.UserID = userutil.MakeUserID(localpart, s.serverName)
+		devices = append(devices, dev)
+	}
+
+	return devices, rows.Err()
+}
+
+func (s *devicesStatements) updateDeviceLastSeen(ctx context.Context, txn *sql.Tx, localpart, deviceID, ipAddr string) error {
+	lastSeenTs := time.Now().UnixNano() / 1000000
+	stmt := sqlutil.TxStmt(txn, s.updateDeviceLastSeenStmt)
+	_, err := stmt.ExecContext(ctx, lastSeenTs, ipAddr, localpart, deviceID)
+	return err
+}

userapi/storage/postgres/key_backup_table.go

@@ -0,0 +1,164 @@
+// Copyright 2021 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/userapi/api"
+)
+
+const keyBackupTableSchema = `
+CREATE TABLE IF NOT EXISTS account_e2e_room_keys (
+    user_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    session_id TEXT NOT NULL,
+
+    version TEXT NOT NULL,
+    first_message_index INTEGER NOT NULL,
+    forwarded_count INTEGER NOT NULL,
+    is_verified BOOLEAN NOT NULL,
+    session_data TEXT NOT NULL
+);
+CREATE UNIQUE INDEX IF NOT EXISTS e2e_room_keys_idx ON account_e2e_room_keys(user_id, room_id, session_id, version);
+CREATE INDEX IF NOT EXISTS e2e_room_keys_versions_idx ON account_e2e_room_keys(user_id, version);
+`
+
+const insertBackupKeySQL = "" +
+	"INSERT INTO account_e2e_room_keys(user_id, room_id, session_id, version, first_message_index, forwarded_count, is_verified, session_data) " +
+	"VALUES ($1, $2, $3, $4, $5, $6, $7, $8)"
+
+const updateBackupKeySQL = "" +
+	"UPDATE account_e2e_room_keys SET first_message_index=$1, forwarded_count=$2, is_verified=$3, session_data=$4 " +
+	"WHERE user_id=$5 AND room_id=$6 AND session_id=$7 AND version=$8"
+
+const countKeysSQL = "" +
+	"SELECT COUNT(*) FROM account_e2e_room_keys WHERE user_id = $1 AND version = $2"
+
+const selectKeysSQL = "" +
+	"SELECT room_id, session_id, first_message_index, forwarded_count, is_verified, session_data FROM account_e2e_room_keys " +
+	"WHERE user_id = $1 AND version = $2"
+
+const selectKeysByRoomIDSQL = "" +
+	"SELECT room_id, session_id, first_message_index, forwarded_count, is_verified, session_data FROM account_e2e_room_keys " +
+	"WHERE user_id = $1 AND version = $2 AND room_id = $3"
+
+const selectKeysByRoomIDAndSessionIDSQL = "" +
+	"SELECT room_id, session_id, first_message_index, forwarded_count, is_verified, session_data FROM account_e2e_room_keys " +
+	"WHERE user_id = $1 AND version = $2 AND room_id = $3 AND session_id = $4"
+
+type keyBackupStatements struct {
+	insertBackupKeyStmt                *sql.Stmt
+	updateBackupKeyStmt                *sql.Stmt
+	countKeysStmt                      *sql.Stmt
+	selectKeysStmt                     *sql.Stmt
+	selectKeysByRoomIDStmt             *sql.Stmt
+	selectKeysByRoomIDAndSessionIDStmt *sql.Stmt
+}
+
+func (s *keyBackupStatements) prepare(db *sql.DB) (err error) {
+	_, err = db.Exec(keyBackupTableSchema)
+	if err != nil {
+		return
+	}
+	return sqlutil.StatementList{
+		{&s.insertBackupKeyStmt, insertBackupKeySQL},
+		{&s.updateBackupKeyStmt, updateBackupKeySQL},
+		{&s.countKeysStmt, countKeysSQL},
+		{&s.selectKeysStmt, selectKeysSQL},
+		{&s.selectKeysByRoomIDStmt, selectKeysByRoomIDSQL},
+		{&s.selectKeysByRoomIDAndSessionIDStmt, selectKeysByRoomIDAndSessionIDSQL},
+	}.Prepare(db)
+}
+
+func (s keyBackupStatements) countKeys(
+	ctx context.Context, txn *sql.Tx, userID, version string,
+) (count int64, err error) {
+	err = txn.Stmt(s.countKeysStmt).QueryRowContext(ctx, userID, version).Scan(&count)
+	return
+}
+
+func (s *keyBackupStatements) insertBackupKey(
+	ctx context.Context, txn *sql.Tx, userID, version string, key api.InternalKeyBackupSession,
+) (err error) {
+	_, err = txn.Stmt(s.insertBackupKeyStmt).ExecContext(
+		ctx, userID, key.RoomID, key.SessionID, version, key.FirstMessageIndex, key.ForwardedCount, key.IsVerified, string(key.SessionData),
+	)
+	return
+}
+
+func (s *keyBackupStatements) updateBackupKey(
+	ctx context.Context, txn *sql.Tx, userID, version string, key api.InternalKeyBackupSession,
+) (err error) {
+	_, err = txn.Stmt(s.updateBackupKeyStmt).ExecContext(
+		ctx, key.FirstMessageIndex, key.ForwardedCount, key.IsVerified, string(key.SessionData), userID, key.RoomID, key.SessionID, version,
+	)
+	return
+}
+
+func (s *keyBackupStatements) selectKeys(
+	ctx context.Context, txn *sql.Tx, userID, version string,
+) (map[string]map[string]api.KeyBackupSession, error) {
+	rows, err := txn.Stmt(s.selectKeysStmt).QueryContext(ctx, userID, version)
+	if err != nil {
+		return nil, err
+	}
+	return unpackKeys(ctx, rows)
+}
+
+func (s *keyBackupStatements) selectKeysByRoomID(
+	ctx context.Context, txn *sql.Tx, userID, version, roomID string,
+) (map[string]map[string]api.KeyBackupSession, error) {
+	rows, err := txn.Stmt(s.selectKeysByRoomIDStmt).QueryContext(ctx, userID, version, roomID)
+	if err != nil {
+		return nil, err
+	}
+	return unpackKeys(ctx, rows)
+}
+
+func (s *keyBackupStatements) selectKeysByRoomIDAndSessionID(
+	ctx context.Context, txn *sql.Tx, userID, version, roomID, sessionID string,
+) (map[string]map[string]api.KeyBackupSession, error) {
+	rows, err := txn.Stmt(s.selectKeysByRoomIDAndSessionIDStmt).QueryContext(ctx, userID, version, roomID, sessionID)
+	if err != nil {
+		return nil, err
+	}
+	return unpackKeys(ctx, rows)
+}
+
+func unpackKeys(ctx context.Context, rows *sql.Rows) (map[string]map[string]api.KeyBackupSession, error) {
+	result := make(map[string]map[string]api.KeyBackupSession)
+	defer internal.CloseAndLogIfError(ctx, rows, "selectKeysStmt.Close failed")
+	for rows.Next() {
+		var key api.InternalKeyBackupSession
+		// room_id, session_id, first_message_index, forwarded_count, is_verified, session_data
+		var sessionDataStr string
+		if err := rows.Scan(&key.RoomID, &key.SessionID, &key.FirstMessageIndex, &key.ForwardedCount, &key.IsVerified, &sessionDataStr); err != nil {
+			return nil, err
+		}
+		key.SessionData = json.RawMessage(sessionDataStr)
+		roomData := result[key.RoomID]
+		if roomData == nil {
+			roomData = make(map[string]api.KeyBackupSession)
+		}
+		roomData[key.SessionID] = key.KeyBackupSession
+		result[key.RoomID] = roomData
+	}
+	return result, nil
+}

userapi/storage/postgres/key_backup_version_table.go

@@ -0,0 +1,161 @@
+// Copyright 2021 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"strconv"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+)
+
+const keyBackupVersionTableSchema = `
+CREATE SEQUENCE IF NOT EXISTS account_e2e_room_keys_versions_seq;
+
+-- the metadata for each generation of encrypted e2e session backups
+CREATE TABLE IF NOT EXISTS account_e2e_room_keys_versions (
+    user_id TEXT NOT NULL,
+	-- this means no 2 users will ever have the same version of e2e session backups which strictly
+	-- isn't necessary, but this is easy to do rather than SELECT MAX(version)+1.
+    version BIGINT DEFAULT nextval('account_e2e_room_keys_versions_seq'),
+    algorithm TEXT NOT NULL,
+    auth_data TEXT NOT NULL,
+	etag TEXT NOT NULL,
+    deleted SMALLINT DEFAULT 0 NOT NULL
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS account_e2e_room_keys_versions_idx ON account_e2e_room_keys_versions(user_id, version);
+`
+
+const insertKeyBackupSQL = "" +
+	"INSERT INTO account_e2e_room_keys_versions(user_id, algorithm, auth_data, etag) VALUES ($1, $2, $3, $4) RETURNING version"
+
+const updateKeyBackupAuthDataSQL = "" +
+	"UPDATE account_e2e_room_keys_versions SET auth_data = $1 WHERE user_id = $2 AND version = $3"
+
+const updateKeyBackupETagSQL = "" +
+	"UPDATE account_e2e_room_keys_versions SET etag = $1 WHERE user_id = $2 AND version = $3"
+
+const deleteKeyBackupSQL = "" +
+	"UPDATE account_e2e_room_keys_versions SET deleted=1 WHERE user_id = $1 AND version = $2"
+
+const selectKeyBackupSQL = "" +
+	"SELECT algorithm, auth_data, etag, deleted FROM account_e2e_room_keys_versions WHERE user_id = $1 AND version = $2"
+
+const selectLatestVersionSQL = "" +
+	"SELECT MAX(version) FROM account_e2e_room_keys_versions WHERE user_id = $1"
+
+type keyBackupVersionStatements struct {
+	insertKeyBackupStmt         *sql.Stmt
+	updateKeyBackupAuthDataStmt *sql.Stmt
+	deleteKeyBackupStmt         *sql.Stmt
+	selectKeyBackupStmt         *sql.Stmt
+	selectLatestVersionStmt     *sql.Stmt
+	updateKeyBackupETagStmt     *sql.Stmt
+}
+
+func (s *keyBackupVersionStatements) prepare(db *sql.DB) (err error) {
+	_, err = db.Exec(keyBackupVersionTableSchema)
+	if err != nil {
+		return
+	}
+	return sqlutil.StatementList{
+		{&s.insertKeyBackupStmt, insertKeyBackupSQL},
+		{&s.updateKeyBackupAuthDataStmt, updateKeyBackupAuthDataSQL},
+		{&s.deleteKeyBackupStmt, deleteKeyBackupSQL},
+		{&s.selectKeyBackupStmt, selectKeyBackupSQL},
+		{&s.selectLatestVersionStmt, selectLatestVersionSQL},
+		{&s.updateKeyBackupETagStmt, updateKeyBackupETagSQL},
+	}.Prepare(db)
+}
+
+func (s *keyBackupVersionStatements) insertKeyBackup(
+	ctx context.Context, txn *sql.Tx, userID, algorithm string, authData json.RawMessage, etag string,
+) (version string, err error) {
+	var versionInt int64
+	err = txn.Stmt(s.insertKeyBackupStmt).QueryRowContext(ctx, userID, algorithm, string(authData), etag).Scan(&versionInt)
+	return strconv.FormatInt(versionInt, 10), err
+}
+
+func (s *keyBackupVersionStatements) updateKeyBackupAuthData(
+	ctx context.Context, txn *sql.Tx, userID, version string, authData json.RawMessage,
+) error {
+	versionInt, err := strconv.ParseInt(version, 10, 64)
+	if err != nil {
+		return fmt.Errorf("invalid version")
+	}
+	_, err = txn.Stmt(s.updateKeyBackupAuthDataStmt).ExecContext(ctx, string(authData), userID, versionInt)
+	return err
+}
+
+func (s *keyBackupVersionStatements) updateKeyBackupETag(
+	ctx context.Context, txn *sql.Tx, userID, version, etag string,
+) error {
+	versionInt, err := strconv.ParseInt(version, 10, 64)
+	if err != nil {
+		return fmt.Errorf("invalid version")
+	}
+	_, err = txn.Stmt(s.updateKeyBackupETagStmt).ExecContext(ctx, etag, userID, versionInt)
+	return err
+}
+
+func (s *keyBackupVersionStatements) deleteKeyBackup(
+	ctx context.Context, txn *sql.Tx, userID, version string,
+) (bool, error) {
+	versionInt, err := strconv.ParseInt(version, 10, 64)
+	if err != nil {
+		return false, fmt.Errorf("invalid version")
+	}
+	result, err := txn.Stmt(s.deleteKeyBackupStmt).ExecContext(ctx, userID, versionInt)
+	if err != nil {
+		return false, err
+	}
+	ra, err := result.RowsAffected()
+	if err != nil {
+		return false, err
+	}
+	return ra == 1, nil
+}
+
+func (s *keyBackupVersionStatements) selectKeyBackup(
+	ctx context.Context, txn *sql.Tx, userID, version string,
+) (versionResult, algorithm string, authData json.RawMessage, etag string, deleted bool, err error) {
+	var versionInt int64
+	if version == "" {
+		var v *int64 // allows nulls
+		if err = txn.Stmt(s.selectLatestVersionStmt).QueryRowContext(ctx, userID).Scan(&v); err != nil {
+			return
+		}
+		if v == nil {
+			err = sql.ErrNoRows
+			return
+		}
+		versionInt = *v
+	} else {
+		if versionInt, err = strconv.ParseInt(version, 10, 64); err != nil {
+			return
+		}
+	}
+	versionResult = strconv.FormatInt(versionInt, 10)
+	var deletedInt int
+	var authDataStr string
+	err = txn.Stmt(s.selectKeyBackupStmt).QueryRowContext(ctx, userID, versionInt).Scan(&algorithm, &authDataStr, &etag, &deletedInt)
+	deleted = deletedInt == 1
+	authData = json.RawMessage(authDataStr)
+	return
+}

userapi/storage/postgres/logintoken_table.go

@@ -0,0 +1,96 @@
+// Copyright 2021 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"time"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/userapi/api"
+	"github.com/matrix-org/util"
+)
+
+type loginTokenStatements struct {
+	insertStmt        *sql.Stmt
+	deleteStmt        *sql.Stmt
+	selectByTokenStmt *sql.Stmt
+}
+
+// execSchema ensures tables and indices exist.
+func (s *loginTokenStatements) execSchema(db *sql.DB) error {
+	_, err := db.Exec(`
+CREATE TABLE IF NOT EXISTS login_tokens (
+	-- The random value of the token issued to a user
+	token TEXT NOT NULL PRIMARY KEY,
+	-- When the token expires
+	token_expires_at TIMESTAMP NOT NULL,
+
+    -- The mxid for this account
+	user_id TEXT NOT NULL
+);
+
+-- This index allows efficient garbage collection of expired tokens.
+CREATE INDEX IF NOT EXISTS login_tokens_expiration_idx ON login_tokens(token_expires_at);
+`)
+	return err
+}
+
+// prepare runs statement preparation.
+func (s *loginTokenStatements) prepare(db *sql.DB) error {
+	if err := s.execSchema(db); err != nil {
+		return err
+	}
+	return sqlutil.StatementList{
+		{&s.insertStmt, "INSERT INTO login_tokens(token, token_expires_at, user_id) VALUES ($1, $2, $3)"},
+		{&s.deleteStmt, "DELETE FROM login_tokens WHERE token = $1 OR token_expires_at <= $2"},
+		{&s.selectByTokenStmt, "SELECT user_id FROM login_tokens WHERE token = $1 AND token_expires_at > $2"},
+	}.Prepare(db)
+}
+
+// insert adds an already generated token to the database.
+func (s *loginTokenStatements) insert(ctx context.Context, txn *sql.Tx, metadata *api.LoginTokenMetadata, data *api.LoginTokenData) error {
+	stmt := sqlutil.TxStmt(txn, s.insertStmt)
+	_, err := stmt.ExecContext(ctx, metadata.Token, metadata.Expiration.UTC(), data.UserID)
+	return err
+}
+
+// deleteByToken removes the named token.
+//
+// As a simple way to garbage-collect stale tokens, we also remove all expired tokens.
+// The login_tokens_expiration_idx index should make that efficient.
+func (s *loginTokenStatements) deleteByToken(ctx context.Context, txn *sql.Tx, token string) error {
+	stmt := sqlutil.TxStmt(txn, s.deleteStmt)
+	res, err := stmt.ExecContext(ctx, token, time.Now().UTC())
+	if err != nil {
+		return err
+	}
+	if n, err := res.RowsAffected(); err == nil && n > 1 {
+		util.GetLogger(ctx).WithField("num_deleted", n).Infof("Deleted %d login tokens (%d likely additional expired token)", n, n-1)
+	}
+	return nil
+}
+
+// selectByToken returns the data associated with the given token. May return sql.ErrNoRows.
+func (s *loginTokenStatements) selectByToken(ctx context.Context, token string) (*api.LoginTokenData, error) {
+	var data api.LoginTokenData
+	err := s.selectByTokenStmt.QueryRowContext(ctx, token, time.Now().UTC()).Scan(&data.UserID)
+	if err != nil {
+		return nil, err
+	}
+
+	return &data, nil
+}

userapi/storage/postgres/openid_table.go

@@ -0,0 +1,81 @@
+package postgres
+
+import (
+	"context"
+	"database/sql"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/userapi/api"
+	"github.com/matrix-org/gomatrixserverlib"
+	log "github.com/sirupsen/logrus"
+)
+
+const openIDTokenSchema = `
+-- Stores data about openid tokens issued for accounts.
+CREATE TABLE IF NOT EXISTS open_id_tokens (
+	-- The value of the token issued to a user
+	token TEXT NOT NULL PRIMARY KEY,
+    -- The Matrix user ID for this account
+	localpart TEXT NOT NULL,
+	-- When the token expires, as a unix timestamp (ms resolution).
+	token_expires_at_ms BIGINT NOT NULL
+);
+`
+
+const insertTokenSQL = "" +
+	"INSERT INTO open_id_tokens(token, localpart, token_expires_at_ms) VALUES ($1, $2, $3)"
+
+const selectTokenSQL = "" +
+	"SELECT localpart, token_expires_at_ms FROM open_id_tokens WHERE token = $1"
+
+type tokenStatements struct {
+	insertTokenStmt *sql.Stmt
+	selectTokenStmt *sql.Stmt
+	serverName      gomatrixserverlib.ServerName
+}
+
+func (s *tokenStatements) prepare(db *sql.DB, server gomatrixserverlib.ServerName) (err error) {
+	_, err = db.Exec(openIDTokenSchema)
+	if err != nil {
+		return
+	}
+	s.serverName = server
+	return sqlutil.StatementList{
+		{&s.insertTokenStmt, insertTokenSQL},
+		{&s.selectTokenStmt, selectTokenSQL},
+	}.Prepare(db)
+}
+
+// insertToken inserts a new OpenID Connect token to the DB.
+// Returns new token, otherwise returns error if the token already exists.
+func (s *tokenStatements) insertToken(
+	ctx context.Context,
+	txn *sql.Tx,
+	token, localpart string,
+	expiresAtMS int64,
+) (err error) {
+	stmt := sqlutil.TxStmt(txn, s.insertTokenStmt)
+	_, err = stmt.ExecContext(ctx, token, localpart, expiresAtMS)
+	return
+}
+
+// selectOpenIDTokenAtrributes gets the attributes associated with an OpenID token from the DB
+// Returns the existing token's attributes, or err if no token is found
+func (s *tokenStatements) selectOpenIDTokenAtrributes(
+	ctx context.Context,
+	token string,
+) (*api.OpenIDTokenAttributes, error) {
+	var openIDTokenAttrs api.OpenIDTokenAttributes
+	err := s.selectTokenStmt.QueryRowContext(ctx, token).Scan(
+		&openIDTokenAttrs.UserID,
+		&openIDTokenAttrs.ExpiresAtMS,
+	)
+	if err != nil {
+		if err != sql.ErrNoRows {
+			log.WithError(err).Error("Unable to retrieve token from the db")
+		}
+		return nil, err
+	}
+
+	return &openIDTokenAttrs, nil
+}

userapi/storage/postgres/profile_table.go

@@ -0,0 +1,130 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+)
+
+const profilesSchema = `
+-- Stores data about accounts profiles.
+CREATE TABLE IF NOT EXISTS account_profiles (
+    -- The Matrix user ID localpart for this account
+    localpart TEXT NOT NULL PRIMARY KEY,
+    -- The display name for this account
+    display_name TEXT,
+    -- The URL of the avatar for this account
+    avatar_url TEXT
+);
+`
+
+const insertProfileSQL = "" +
+	"INSERT INTO account_profiles(localpart, display_name, avatar_url) VALUES ($1, $2, $3)"
+
+const selectProfileByLocalpartSQL = "" +
+	"SELECT localpart, display_name, avatar_url FROM account_profiles WHERE localpart = $1"
+
+const setAvatarURLSQL = "" +
+	"UPDATE account_profiles SET avatar_url = $1 WHERE localpart = $2"
+
+const setDisplayNameSQL = "" +
+	"UPDATE account_profiles SET display_name = $1 WHERE localpart = $2"
+
+const selectProfilesBySearchSQL = "" +
+	"SELECT localpart, display_name, avatar_url FROM account_profiles WHERE localpart LIKE $1 OR display_name LIKE $1 LIMIT $2"
+
+type profilesStatements struct {
+	insertProfileStmt            *sql.Stmt
+	selectProfileByLocalpartStmt *sql.Stmt
+	setAvatarURLStmt             *sql.Stmt
+	setDisplayNameStmt           *sql.Stmt
+	selectProfilesBySearchStmt   *sql.Stmt
+}
+
+func (s *profilesStatements) prepare(db *sql.DB) (err error) {
+	_, err = db.Exec(profilesSchema)
+	if err != nil {
+		return
+	}
+	return sqlutil.StatementList{
+		{&s.insertProfileStmt, insertProfileSQL},
+		{&s.selectProfileByLocalpartStmt, selectProfileByLocalpartSQL},
+		{&s.setAvatarURLStmt, setAvatarURLSQL},
+		{&s.setDisplayNameStmt, setDisplayNameSQL},
+		{&s.selectProfilesBySearchStmt, selectProfilesBySearchSQL},
+	}.Prepare(db)
+}
+
+func (s *profilesStatements) insertProfile(
+	ctx context.Context, txn *sql.Tx, localpart string,
+) (err error) {
+	_, err = sqlutil.TxStmt(txn, s.insertProfileStmt).ExecContext(ctx, localpart, "", "")
+	return
+}
+
+func (s *profilesStatements) selectProfileByLocalpart(
+	ctx context.Context, localpart string,
+) (*authtypes.Profile, error) {
+	var profile authtypes.Profile
+	err := s.selectProfileByLocalpartStmt.QueryRowContext(ctx, localpart).Scan(
+		&profile.Localpart, &profile.DisplayName, &profile.AvatarURL,
+	)
+	if err != nil {
+		return nil, err
+	}
+	return &profile, nil
+}
+
+func (s *profilesStatements) setAvatarURL(
+	ctx context.Context, localpart string, avatarURL string,
+) (err error) {
+	_, err = s.setAvatarURLStmt.ExecContext(ctx, avatarURL, localpart)
+	return
+}
+
+func (s *profilesStatements) setDisplayName(
+	ctx context.Context, localpart string, displayName string,
+) (err error) {
+	_, err = s.setDisplayNameStmt.ExecContext(ctx, displayName, localpart)
+	return
+}
+
+func (s *profilesStatements) selectProfilesBySearch(
+	ctx context.Context, searchString string, limit int,
+) ([]authtypes.Profile, error) {
+	var profiles []authtypes.Profile
+	// The fmt.Sprintf directive below is building a parameter for the
+	// "LIKE" condition in the SQL query. %% escapes the % char, so the
+	// statement in the end will look like "LIKE %searchString%".
+	rows, err := s.selectProfilesBySearchStmt.QueryContext(ctx, fmt.Sprintf("%%%s%%", searchString), limit)
+	if err != nil {
+		return nil, err
+	}
+	defer internal.CloseAndLogIfError(ctx, rows, "selectProfilesBySearch: rows.close() failed")
+	for rows.Next() {
+		var profile authtypes.Profile
+		if err := rows.Scan(&profile.Localpart, &profile.DisplayName, &profile.AvatarURL); err != nil {
+			return nil, err
+		}
+		profiles = append(profiles, profile)
+	}
+	return profiles, nil
+}

userapi/storage/postgres/storage.go

@@ -0,0 +1,729 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package postgres
+
+import (
+	"context"
+	"crypto/rand"
+	"database/sql"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/matrix-org/gomatrixserverlib"
+	"golang.org/x/crypto/bcrypt"
+
+	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/userapi/api"
+	"github.com/matrix-org/dendrite/userapi/storage/postgres/deltas"
+
+	// Import the postgres database driver.
+	_ "github.com/lib/pq"
+)
+
+// Database represents an account database
+type Database struct {
+	db     *sql.DB
+	writer sqlutil.Writer
+	sqlutil.PartitionOffsetStatements
+	accounts              accountsStatements
+	profiles              profilesStatements
+	accountDatas          accountDataStatements
+	threepids             threepidStatements
+	openIDTokens          tokenStatements
+	keyBackupVersions     keyBackupVersionStatements
+	devices               devicesStatements
+	loginTokens           loginTokenStatements
+	loginTokenLifetime    time.Duration
+	keyBackups            keyBackupStatements
+	serverName            gomatrixserverlib.ServerName
+	bcryptCost            int
+	openIDTokenLifetimeMS int64
+}
+
+const (
+	// The length of generated device IDs
+	deviceIDByteLength   = 6
+	loginTokenByteLength = 32
+)
+
+// NewDatabase creates a new accounts and profiles database
+func NewDatabase(dbProperties *config.DatabaseOptions, serverName gomatrixserverlib.ServerName, bcryptCost int, openIDTokenLifetimeMS int64, loginTokenLifetime time.Duration) (*Database, error) {
+	db, err := sqlutil.Open(dbProperties)
+	if err != nil {
+		return nil, err
+	}
+	d := &Database{
+		serverName:            serverName,
+		db:                    db,
+		writer:                sqlutil.NewDummyWriter(),
+		loginTokenLifetime:    loginTokenLifetime,
+		bcryptCost:            bcryptCost,
+		openIDTokenLifetimeMS: openIDTokenLifetimeMS,
+	}
+
+	// Create tables before executing migrations so we don't fail if the table is missing,
+	// and THEN prepare statements so we don't fail due to referencing new columns
+	if err = d.accounts.execSchema(db); err != nil {
+		return nil, err
+	}
+	m := sqlutil.NewMigrations()
+	deltas.LoadIsActive(m)
+	//deltas.LoadLastSeenTSIP(m)
+	deltas.LoadAddAccountType(m)
+	if err = m.RunDeltas(db, dbProperties); err != nil {
+		return nil, err
+	}
+
+	if err = d.PartitionOffsetStatements.Prepare(db, d.writer, "account"); err != nil {
+		return nil, err
+	}
+	if err = d.accounts.prepare(db, serverName); err != nil {
+		return nil, err
+	}
+	if err = d.profiles.prepare(db); err != nil {
+		return nil, err
+	}
+	if err = d.accountDatas.prepare(db); err != nil {
+		return nil, err
+	}
+	if err = d.threepids.prepare(db); err != nil {
+		return nil, err
+	}
+	if err = d.openIDTokens.prepare(db, serverName); err != nil {
+		return nil, err
+	}
+	if err = d.keyBackupVersions.prepare(db); err != nil {
+		return nil, err
+	}
+	if err = d.keyBackups.prepare(db); err != nil {
+		return nil, err
+	}
+	if err = d.devices.prepare(db, serverName); err != nil {
+		return nil, err
+	}
+	if err = d.loginTokens.prepare(db); err != nil {
+		return nil, err
+	}
+
+	return d, nil
+}
+
+// GetAccountByPassword returns the account associated with the given localpart and password.
+// Returns sql.ErrNoRows if no account exists which matches the given localpart.
+func (d *Database) GetAccountByPassword(
+	ctx context.Context, localpart, plaintextPassword string,
+) (*api.Account, error) {
+	hash, err := d.accounts.selectPasswordHash(ctx, localpart)
+	if err != nil {
+		return nil, err
+	}
+	if err := bcrypt.CompareHashAndPassword([]byte(hash), []byte(plaintextPassword)); err != nil {
+		return nil, err
+	}
+	return d.accounts.selectAccountByLocalpart(ctx, localpart)
+}
+
+// GetProfileByLocalpart returns the profile associated with the given localpart.
+// Returns sql.ErrNoRows if no profile exists which matches the given localpart.
+func (d *Database) GetProfileByLocalpart(
+	ctx context.Context, localpart string,
+) (*authtypes.Profile, error) {
+	return d.profiles.selectProfileByLocalpart(ctx, localpart)
+}
+
+// SetAvatarURL updates the avatar URL of the profile associated with the given
+// localpart. Returns an error if something went wrong with the SQL query
+func (d *Database) SetAvatarURL(
+	ctx context.Context, localpart string, avatarURL string,
+) error {
+	return d.profiles.setAvatarURL(ctx, localpart, avatarURL)
+}
+
+// SetDisplayName updates the display name of the profile associated with the given
+// localpart. Returns an error if something went wrong with the SQL query
+func (d *Database) SetDisplayName(
+	ctx context.Context, localpart string, displayName string,
+) error {
+	return d.profiles.setDisplayName(ctx, localpart, displayName)
+}
+
+// SetPassword sets the account password to the given hash.
+func (d *Database) SetPassword(
+	ctx context.Context, localpart, plaintextPassword string,
+) error {
+	hash, err := d.hashPassword(plaintextPassword)
+	if err != nil {
+		return err
+	}
+	return d.accounts.updatePassword(ctx, localpart, hash)
+}
+
+// CreateAccount makes a new account with the given login name and password, and creates an empty profile
+// for this account. If no password is supplied, the account will be a passwordless account. If the
+// account already exists, it will return nil, sqlutil.ErrUserExists.
+func (d *Database) CreateAccount(
+	ctx context.Context, localpart, plaintextPassword, appserviceID string, accountType api.AccountType,
+) (acc *api.Account, err error) {
+	err = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		// For guest accounts, we create a new numeric local part
+		if accountType == api.AccountTypeGuest {
+			var numLocalpart int64
+			numLocalpart, err = d.accounts.selectNewNumericLocalpart(ctx, txn)
+			if err != nil {
+				return err
+			}
+			localpart = strconv.FormatInt(numLocalpart, 10)
+			plaintextPassword = ""
+			appserviceID = ""
+		}
+		acc, err = d.createAccount(ctx, txn, localpart, plaintextPassword, appserviceID, accountType)
+		return err
+	})
+	return
+}
+
+func (d *Database) createAccount(
+	ctx context.Context, txn *sql.Tx, localpart, plaintextPassword, appserviceID string, accountType api.AccountType,
+) (*api.Account, error) {
+	var account *api.Account
+	var err error
+	// Generate a password hash if this is not a password-less user
+	hash := ""
+	if plaintextPassword != "" {
+		hash, err = d.hashPassword(plaintextPassword)
+		if err != nil {
+			return nil, err
+		}
+	}
+	if account, err = d.accounts.insertAccount(ctx, txn, localpart, hash, appserviceID, accountType); err != nil {
+		if sqlutil.IsUniqueConstraintViolationErr(err) {
+			return nil, sqlutil.ErrUserExists
+		}
+		return nil, err
+	}
+	if err = d.profiles.insertProfile(ctx, txn, localpart); err != nil {
+		return nil, err
+	}
+	if err = d.accountDatas.insertAccountData(ctx, txn, localpart, "", "m.push_rules", json.RawMessage(`{
+		"global": {
+			"content": [],
+			"override": [],
+			"room": [],
+			"sender": [],
+			"underride": []
+		}
+	}`)); err != nil {
+		return nil, err
+	}
+	return account, nil
+}
+
+// SaveAccountData saves new account data for a given user and a given room.
+// If the account data is not specific to a room, the room ID should be an empty string
+// If an account data already exists for a given set (user, room, data type), it will
+// update the corresponding row with the new content
+// Returns a SQL error if there was an issue with the insertion/update
+func (d *Database) SaveAccountData(
+	ctx context.Context, localpart, roomID, dataType string, content json.RawMessage,
+) error {
+	return sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		return d.accountDatas.insertAccountData(ctx, txn, localpart, roomID, dataType, content)
+	})
+}
+
+// GetAccountData returns account data related to a given localpart
+// If no account data could be found, returns an empty arrays
+// Returns an error if there was an issue with the retrieval
+func (d *Database) GetAccountData(ctx context.Context, localpart string) (
+	global map[string]json.RawMessage,
+	rooms map[string]map[string]json.RawMessage,
+	err error,
+) {
+	return d.accountDatas.selectAccountData(ctx, localpart)
+}
+
+// GetAccountDataByType returns account data matching a given
+// localpart, room ID and type.
+// If no account data could be found, returns nil
+// Returns an error if there was an issue with the retrieval
+func (d *Database) GetAccountDataByType(
+	ctx context.Context, localpart, roomID, dataType string,
+) (data json.RawMessage, err error) {
+	return d.accountDatas.selectAccountDataByType(
+		ctx, localpart, roomID, dataType,
+	)
+}
+
+// GetNewNumericLocalpart generates and returns a new unused numeric localpart
+func (d *Database) GetNewNumericLocalpart(
+	ctx context.Context,
+) (int64, error) {
+	return d.accounts.selectNewNumericLocalpart(ctx, nil)
+}
+
+func (d *Database) hashPassword(plaintext string) (hash string, err error) {
+	hashBytes, err := bcrypt.GenerateFromPassword([]byte(plaintext), d.bcryptCost)
+	return string(hashBytes), err
+}
+
+// Err3PIDInUse is the error returned when trying to save an association involving
+// a third-party identifier which is already associated to a local user.
+var Err3PIDInUse = errors.New("this third-party identifier is already in use")
+
+// SaveThreePIDAssociation saves the association between a third party identifier
+// and a local Matrix user (identified by the user's ID's local part).
+// If the third-party identifier is already part of an association, returns Err3PIDInUse.
+// Returns an error if there was a problem talking to the database.
+func (d *Database) SaveThreePIDAssociation(
+	ctx context.Context, threepid, localpart, medium string,
+) (err error) {
+	return sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		user, err := d.threepids.selectLocalpartForThreePID(
+			ctx, txn, threepid, medium,
+		)
+		if err != nil {
+			return err
+		}
+
+		if len(user) > 0 {
+			return Err3PIDInUse
+		}
+
+		return d.threepids.insertThreePID(ctx, txn, threepid, medium, localpart)
+	})
+}
+
+// RemoveThreePIDAssociation removes the association involving a given third-party
+// identifier.
+// If no association exists involving this third-party identifier, returns nothing.
+// If there was a problem talking to the database, returns an error.
+func (d *Database) RemoveThreePIDAssociation(
+	ctx context.Context, threepid string, medium string,
+) (err error) {
+	return d.threepids.deleteThreePID(ctx, threepid, medium)
+}
+
+// GetLocalpartForThreePID looks up the localpart associated with a given third-party
+// identifier.
+// If no association involves the given third-party idenfitier, returns an empty
+// string.
+// Returns an error if there was a problem talking to the database.
+func (d *Database) GetLocalpartForThreePID(
+	ctx context.Context, threepid string, medium string,
+) (localpart string, err error) {
+	return d.threepids.selectLocalpartForThreePID(ctx, nil, threepid, medium)
+}
+
+// GetThreePIDsForLocalpart looks up the third-party identifiers associated with
+// a given local user.
+// If no association is known for this user, returns an empty slice.
+// Returns an error if there was an issue talking to the database.
+func (d *Database) GetThreePIDsForLocalpart(
+	ctx context.Context, localpart string,
+) (threepids []authtypes.ThreePID, err error) {
+	return d.threepids.selectThreePIDsForLocalpart(ctx, localpart)
+}
+
+// CheckAccountAvailability checks if the username/localpart is already present
+// in the database.
+// If the DB returns sql.ErrNoRows the Localpart isn't taken.
+func (d *Database) CheckAccountAvailability(ctx context.Context, localpart string) (bool, error) {
+	_, err := d.accounts.selectAccountByLocalpart(ctx, localpart)
+	if err == sql.ErrNoRows {
+		return true, nil
+	}
+	return false, err
+}
+
+// GetAccountByLocalpart returns the account associated with the given localpart.
+// This function assumes the request is authenticated or the account data is used only internally.
+// Returns sql.ErrNoRows if no account exists which matches the given localpart.
+func (d *Database) GetAccountByLocalpart(ctx context.Context, localpart string,
+) (*api.Account, error) {
+	return d.accounts.selectAccountByLocalpart(ctx, localpart)
+}
+
+// SearchProfiles returns all profiles where the provided localpart or display name
+// match any part of the profiles in the database.
+func (d *Database) SearchProfiles(ctx context.Context, searchString string, limit int,
+) ([]authtypes.Profile, error) {
+	return d.profiles.selectProfilesBySearch(ctx, searchString, limit)
+}
+
+// DeactivateAccount deactivates the user's account, removing all ability for the user to login again.
+func (d *Database) DeactivateAccount(ctx context.Context, localpart string) (err error) {
+	return d.accounts.deactivateAccount(ctx, localpart)
+}
+
+// CreateOpenIDToken persists a new token that was issued through OpenID Connect
+func (d *Database) CreateOpenIDToken(
+	ctx context.Context,
+	token, localpart string,
+) (int64, error) {
+	expiresAtMS := time.Now().UnixNano()/int64(time.Millisecond) + d.openIDTokenLifetimeMS
+	err := sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		return d.openIDTokens.insertToken(ctx, txn, token, localpart, expiresAtMS)
+	})
+	return expiresAtMS, err
+}
+
+// GetOpenIDTokenAttributes gets the attributes of issued an OIDC auth token
+func (d *Database) GetOpenIDTokenAttributes(
+	ctx context.Context,
+	token string,
+) (*api.OpenIDTokenAttributes, error) {
+	return d.openIDTokens.selectOpenIDTokenAtrributes(ctx, token)
+}
+
+func (d *Database) CreateKeyBackup(
+	ctx context.Context, userID, algorithm string, authData json.RawMessage,
+) (version string, err error) {
+	err = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		version, err = d.keyBackupVersions.insertKeyBackup(ctx, txn, userID, algorithm, authData, "")
+		return err
+	})
+	return
+}
+
+func (d *Database) UpdateKeyBackupAuthData(
+	ctx context.Context, userID, version string, authData json.RawMessage,
+) (err error) {
+	err = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		return d.keyBackupVersions.updateKeyBackupAuthData(ctx, txn, userID, version, authData)
+	})
+	return
+}
+
+func (d *Database) DeleteKeyBackup(
+	ctx context.Context, userID, version string,
+) (exists bool, err error) {
+	err = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		exists, err = d.keyBackupVersions.deleteKeyBackup(ctx, txn, userID, version)
+		return err
+	})
+	return
+}
+
+func (d *Database) GetKeyBackup(
+	ctx context.Context, userID, version string,
+) (versionResult, algorithm string, authData json.RawMessage, etag string, deleted bool, err error) {
+	err = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		versionResult, algorithm, authData, etag, deleted, err = d.keyBackupVersions.selectKeyBackup(ctx, txn, userID, version)
+		return err
+	})
+	return
+}
+
+func (d *Database) GetBackupKeys(
+	ctx context.Context, version, userID, filterRoomID, filterSessionID string,
+) (result map[string]map[string]api.KeyBackupSession, err error) {
+	err = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		if filterSessionID != "" {
+			result, err = d.keyBackups.selectKeysByRoomIDAndSessionID(ctx, txn, userID, version, filterRoomID, filterSessionID)
+			return err
+		}
+		if filterRoomID != "" {
+			result, err = d.keyBackups.selectKeysByRoomID(ctx, txn, userID, version, filterRoomID)
+			return err
+		}
+		result, err = d.keyBackups.selectKeys(ctx, txn, userID, version)
+		return err
+	})
+	return
+}
+
+func (d *Database) CountBackupKeys(
+	ctx context.Context, version, userID string,
+) (count int64, err error) {
+	err = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		count, err = d.keyBackups.countKeys(ctx, txn, userID, version)
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+	return
+}
+
+// nolint:nakedret
+func (d *Database) UpsertBackupKeys(
+	ctx context.Context, version, userID string, uploads []api.InternalKeyBackupSession,
+) (count int64, etag string, err error) {
+	// wrap the following logic in a txn to ensure we atomically upload keys
+	err = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		_, _, _, oldETag, deleted, err := d.keyBackupVersions.selectKeyBackup(ctx, txn, userID, version)
+		if err != nil {
+			return err
+		}
+		if deleted {
+			return fmt.Errorf("backup was deleted")
+		}
+		// pull out all keys for this (user_id, version)
+		existingKeys, err := d.keyBackups.selectKeys(ctx, txn, userID, version)
+		if err != nil {
+			return err
+		}
+
+		changed := false
+		// loop over all the new keys (which should be smaller than the set of backed up keys)
+		for _, newKey := range uploads {
+			// if we have a matching (room_id, session_id), we may need to update the key if it meets some rules, check them.
+			existingRoom := existingKeys[newKey.RoomID]
+			if existingRoom != nil {
+				existingSession, ok := existingRoom[newKey.SessionID]
+				if ok {
+					if existingSession.ShouldReplaceRoomKey(&newKey.KeyBackupSession) {
+						err = d.keyBackups.updateBackupKey(ctx, txn, userID, version, newKey)
+						changed = true
+						if err != nil {
+							return fmt.Errorf("d.keyBackups.updateBackupKey: %w", err)
+						}
+					}
+					// if we shouldn't replace the key we do nothing with it
+					continue
+				}
+			}
+			// if we're here, either the room or session are new, either way, we insert
+			err = d.keyBackups.insertBackupKey(ctx, txn, userID, version, newKey)
+			changed = true
+			if err != nil {
+				return fmt.Errorf("d.keyBackups.insertBackupKey: %w", err)
+			}
+		}
+
+		count, err = d.keyBackups.countKeys(ctx, txn, userID, version)
+		if err != nil {
+			return err
+		}
+		if changed {
+			// update the etag
+			var newETag string
+			if oldETag == "" {
+				newETag = "1"
+			} else {
+				oldETagInt, err := strconv.ParseInt(oldETag, 10, 64)
+				if err != nil {
+					return fmt.Errorf("failed to parse old etag: %s", err)
+				}
+				newETag = strconv.FormatInt(oldETagInt+1, 10)
+			}
+			etag = newETag
+			return d.keyBackupVersions.updateKeyBackupETag(ctx, txn, userID, version, newETag)
+		} else {
+			etag = oldETag
+		}
+		return nil
+	})
+	return
+}
+
+// GetDeviceByAccessToken returns the device matching the given access token.
+// Returns sql.ErrNoRows if no matching device was found.
+func (d *Database) GetDeviceByAccessToken(
+	ctx context.Context, token string,
+) (*api.Device, error) {
+	return d.devices.selectDeviceByToken(ctx, token)
+}
+
+// GetDeviceByID returns the device matching the given ID.
+// Returns sql.ErrNoRows if no matching device was found.
+func (d *Database) GetDeviceByID(
+	ctx context.Context, localpart, deviceID string,
+) (*api.Device, error) {
+	return d.devices.selectDeviceByID(ctx, localpart, deviceID)
+}
+
+// GetDevicesByLocalpart returns the devices matching the given localpart.
+func (d *Database) GetDevicesByLocalpart(
+	ctx context.Context, localpart string,
+) ([]api.Device, error) {
+	return d.devices.selectDevicesByLocalpart(ctx, nil, localpart, "")
+}
+
+func (d *Database) GetDevicesByID(ctx context.Context, deviceIDs []string) ([]api.Device, error) {
+	return d.devices.selectDevicesByID(ctx, deviceIDs)
+}
+
+// CreateDevice makes a new device associated with the given user ID localpart.
+// If there is already a device with the same device ID for this user, that access token will be revoked
+// and replaced with the given accessToken. If the given accessToken is already in use for another device,
+// an error will be returned.
+// If no device ID is given one is generated.
+// Returns the device on success.
+func (d *Database) CreateDevice(
+	ctx context.Context, localpart string, deviceID *string, accessToken string,
+	displayName *string, ipAddr, userAgent string,
+) (dev *api.Device, returnErr error) {
+	if deviceID != nil {
+		returnErr = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+			var err error
+			// Revoke existing tokens for this device
+			if err = d.devices.deleteDevice(ctx, txn, *deviceID, localpart); err != nil {
+				return err
+			}
+
+			dev, err = d.devices.insertDevice(ctx, txn, *deviceID, localpart, accessToken, displayName, ipAddr, userAgent)
+			return err
+		})
+	} else {
+		// We generate device IDs in a loop in case its already taken.
+		// We cap this at going round 5 times to ensure we don't spin forever
+		var newDeviceID string
+		for i := 1; i <= 5; i++ {
+			newDeviceID, returnErr = generateDeviceID()
+			if returnErr != nil {
+				return
+			}
+
+			returnErr = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+				var err error
+				dev, err = d.devices.insertDevice(ctx, txn, newDeviceID, localpart, accessToken, displayName, ipAddr, userAgent)
+				return err
+			})
+			if returnErr == nil {
+				return
+			}
+		}
+	}
+	return
+}
+
+// generateDeviceID creates a new device id. Returns an error if failed to generate
+// random bytes.
+func generateDeviceID() (string, error) {
+	b := make([]byte, deviceIDByteLength)
+	_, err := rand.Read(b)
+	if err != nil {
+		return "", err
+	}
+	// url-safe no padding
+	return base64.RawURLEncoding.EncodeToString(b), nil
+}
+
+// UpdateDevice updates the given device with the display name.
+// Returns SQL error if there are problems and nil on success.
+func (d *Database) UpdateDevice(
+	ctx context.Context, localpart, deviceID string, displayName *string,
+) error {
+	return sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		return d.devices.updateDeviceName(ctx, txn, localpart, deviceID, displayName)
+	})
+}
+
+// RemoveDevice revokes a device by deleting the entry in the database
+// matching with the given device ID and user ID localpart.
+// If the device doesn't exist, it will not return an error
+// If something went wrong during the deletion, it will return the SQL error.
+func (d *Database) RemoveDevice(
+	ctx context.Context, deviceID, localpart string,
+) error {
+	return sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		if err := d.devices.deleteDevice(ctx, txn, deviceID, localpart); err != sql.ErrNoRows {
+			return err
+		}
+		return nil
+	})
+}
+
+// RemoveDevices revokes one or more devices by deleting the entry in the database
+// matching with the given device IDs and user ID localpart.
+// If the devices don't exist, it will not return an error
+// If something went wrong during the deletion, it will return the SQL error.
+func (d *Database) RemoveDevices(
+	ctx context.Context, localpart string, devices []string,
+) error {
+	return sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		if err := d.devices.deleteDevices(ctx, txn, localpart, devices); err != sql.ErrNoRows {
+			return err
+		}
+		return nil
+	})
+}
+
+// RemoveAllDevices revokes devices by deleting the entry in the
+// database matching the given user ID localpart.
+// If something went wrong during the deletion, it will return the SQL error.
+func (d *Database) RemoveAllDevices(
+	ctx context.Context, localpart, exceptDeviceID string,
+) (devices []api.Device, err error) {
+	err = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		devices, err = d.devices.selectDevicesByLocalpart(ctx, txn, localpart, exceptDeviceID)
+		if err != nil {
+			return err
+		}
+		if err := d.devices.deleteDevicesByLocalpart(ctx, txn, localpart, exceptDeviceID); err != sql.ErrNoRows {
+			return err
+		}
+		return nil
+	})
+	return
+}
+
+// UpdateDeviceLastSeen updates a the last seen timestamp and the ip address
+func (d *Database) UpdateDeviceLastSeen(ctx context.Context, localpart, deviceID, ipAddr string) error {
+	return sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		return d.devices.updateDeviceLastSeen(ctx, txn, localpart, deviceID, ipAddr)
+	})
+}
+
+// CreateLoginToken generates a token, stores and returns it. The lifetime is
+// determined by the loginTokenLifetime given to the Database constructor.
+func (d *Database) CreateLoginToken(ctx context.Context, data *api.LoginTokenData) (*api.LoginTokenMetadata, error) {
+	tok, err := generateLoginToken()
+	if err != nil {
+		return nil, err
+	}
+	meta := &api.LoginTokenMetadata{
+		Token:      tok,
+		Expiration: time.Now().Add(d.loginTokenLifetime),
+	}
+
+	err = sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		return d.loginTokens.insert(ctx, txn, meta, data)
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return meta, nil
+}
+
+func generateLoginToken() (string, error) {
+	b := make([]byte, loginTokenByteLength)
+	_, err := rand.Read(b)
+	if err != nil {
+		return "", err
+	}
+	return base64.RawURLEncoding.EncodeToString(b), nil
+}
+
+// RemoveLoginToken removes the named token (and may clean up other expired tokens).
+func (d *Database) RemoveLoginToken(ctx context.Context, token string) error {
+	return sqlutil.WithTransaction(d.db, func(txn *sql.Tx) error {
+		return d.loginTokens.deleteByToken(ctx, txn, token)
+	})
+}
+
+// GetLoginTokenDataByToken returns the data associated with the given token.
+// May return sql.ErrNoRows.
+func (d *Database) GetLoginTokenDataByToken(ctx context.Context, token string) (*api.LoginTokenData, error) {
+	return d.loginTokens.selectByToken(ctx, token)
+}

userapi/storage/postgres/threepid_table.go

@@ -0,0 +1,121 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package postgres
+
+import (
+	"context"
+	"database/sql"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+
+	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
+)
+
+const threepidSchema = `
+-- Stores data about third party identifiers
+CREATE TABLE IF NOT EXISTS account_threepid (
+	-- The third party identifier
+	threepid TEXT NOT NULL,
+	-- The 3PID medium
+	medium TEXT NOT NULL DEFAULT 'email',
+	-- The localpart of the Matrix user ID associated to this 3PID
+	localpart TEXT NOT NULL,
+
+	PRIMARY KEY(threepid, medium)
+);
+
+CREATE INDEX IF NOT EXISTS account_threepid_localpart ON account_threepid(localpart);
+`
+
+const selectLocalpartForThreePIDSQL = "" +
+	"SELECT localpart FROM account_threepid WHERE threepid = $1 AND medium = $2"
+
+const selectThreePIDsForLocalpartSQL = "" +
+	"SELECT threepid, medium FROM account_threepid WHERE localpart = $1"
+
+const insertThreePIDSQL = "" +
+	"INSERT INTO account_threepid (threepid, medium, localpart) VALUES ($1, $2, $3)"
+
+const deleteThreePIDSQL = "" +
+	"DELETE FROM account_threepid WHERE threepid = $1 AND medium = $2"
+
+type threepidStatements struct {
+	selectLocalpartForThreePIDStmt  *sql.Stmt
+	selectThreePIDsForLocalpartStmt *sql.Stmt
+	insertThreePIDStmt              *sql.Stmt
+	deleteThreePIDStmt              *sql.Stmt
+}
+
+func (s *threepidStatements) prepare(db *sql.DB) (err error) {
+	_, err = db.Exec(threepidSchema)
+	if err != nil {
+		return
+	}
+	return sqlutil.StatementList{
+		{&s.selectLocalpartForThreePIDStmt, selectLocalpartForThreePIDSQL},
+		{&s.selectThreePIDsForLocalpartStmt, selectThreePIDsForLocalpartSQL},
+		{&s.insertThreePIDStmt, insertThreePIDSQL},
+		{&s.deleteThreePIDStmt, deleteThreePIDSQL},
+	}.Prepare(db)
+}
+
+func (s *threepidStatements) selectLocalpartForThreePID(
+	ctx context.Context, txn *sql.Tx, threepid string, medium string,
+) (localpart string, err error) {
+	stmt := sqlutil.TxStmt(txn, s.selectLocalpartForThreePIDStmt)
+	err = stmt.QueryRowContext(ctx, threepid, medium).Scan(&localpart)
+	if err == sql.ErrNoRows {
+		return "", nil
+	}
+	return
+}
+
+func (s *threepidStatements) selectThreePIDsForLocalpart(
+	ctx context.Context, localpart string,
+) (threepids []authtypes.ThreePID, err error) {
+	rows, err := s.selectThreePIDsForLocalpartStmt.QueryContext(ctx, localpart)
+	if err != nil {
+		return
+	}
+
+	threepids = []authtypes.ThreePID{}
+	for rows.Next() {
+		var threepid string
+		var medium string
+		if err = rows.Scan(&threepid, &medium); err != nil {
+			return
+		}
+		threepids = append(threepids, authtypes.ThreePID{
+			Address: threepid,
+			Medium:  medium,
+		})
+	}
+
+	return
+}
+
+func (s *threepidStatements) insertThreePID(
+	ctx context.Context, txn *sql.Tx, threepid, medium, localpart string,
+) (err error) {
+	stmt := sqlutil.TxStmt(txn, s.insertThreePIDStmt)
+	_, err = stmt.ExecContext(ctx, threepid, medium, localpart)
+	return
+}
+
+func (s *threepidStatements) deleteThreePID(
+	ctx context.Context, threepid string, medium string) (err error) {
+	_, err = s.deleteThreePIDStmt.ExecContext(ctx, threepid, medium)
+	return
+}