mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2026-01-16 18:43:10 -06:00
Code Issues Packages Projects Releases Wiki Activity

PostgreSQL works I think

Browse source
This commit is contained in:
Neil Alexander 2022-10-17 14:18:48 +01:00
parent 6225a706c6
commit a679a62c9f
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
6 changed files with 63 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
userapi/storage/postgres/deltas/20200929203058_is_active.go
View file

@ -7,7 +7,7 @@ import (
) )
func UpIsActive(ctx context.Context, tx *sql.Tx) error { func UpIsActive(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, "ALTER TABLE account_accounts ADD COLUMN IF NOT EXISTS is_deactivated BOOLEAN DEFAULT FALSE;") _, err := tx.ExecContext(ctx, "ALTER TABLE userapi_accounts ADD COLUMN IF NOT EXISTS is_deactivated BOOLEAN DEFAULT FALSE;")
if err != nil { if err != nil {
return fmt.Errorf("failed to execute upgrade: %w", err) return fmt.Errorf("failed to execute upgrade: %w", err)
} }
@ -15,7 +15,7 @@ func UpIsActive(ctx context.Context, tx *sql.Tx) error {
} }
func DownIsActive(ctx context.Context, tx *sql.Tx) error { func DownIsActive(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, "ALTER TABLE account_accounts DROP COLUMN is_deactivated;") _, err := tx.ExecContext(ctx, "ALTER TABLE userapi_accounts DROP COLUMN is_deactivated;")
if err != nil { if err != nil {
return fmt.Errorf("failed to execute downgrade: %w", err) return fmt.Errorf("failed to execute downgrade: %w", err)
} }

12
userapi/storage/postgres/deltas/20201001204705_last_seen_ts_ip.go
View file

@ -8,9 +8,9 @@ import (
func UpLastSeenTSIP(ctx context.Context, tx *sql.Tx) error { func UpLastSeenTSIP(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, ` _, err := tx.ExecContext(ctx, `
ALTER TABLE device_devices ADD COLUMN IF NOT EXISTS last_seen_ts BIGINT NOT NULL DEFAULT EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)*1000; ALTER TABLE userapi_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 userapi_devices ADD COLUMN IF NOT EXISTS ip TEXT;
ALTER TABLE device_devices ADD COLUMN IF NOT EXISTS user_agent TEXT;`) ALTER TABLE userapi_devices ADD COLUMN IF NOT EXISTS user_agent TEXT;`)
if err != nil { if err != nil {
return fmt.Errorf("failed to execute upgrade: %w", err) return fmt.Errorf("failed to execute upgrade: %w", err)
} }
@ -19,9 +19,9 @@ ALTER TABLE device_devices ADD COLUMN IF NOT EXISTS user_agent TEXT;`)
func DownLastSeenTSIP(ctx context.Context, tx *sql.Tx) error { func DownLastSeenTSIP(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, ` _, err := tx.ExecContext(ctx, `
ALTER TABLE device_devices DROP COLUMN last_seen_ts; ALTER TABLE userapi_devices DROP COLUMN last_seen_ts;
ALTER TABLE device_devices DROP COLUMN ip; ALTER TABLE userapi_devices DROP COLUMN ip;
ALTER TABLE device_devices DROP COLUMN user_agent;`) ALTER TABLE userapi_devices DROP COLUMN user_agent;`)
if err != nil { if err != nil {
return fmt.Errorf("failed to execute downgrade: %w", err) return fmt.Errorf("failed to execute downgrade: %w", err)
} }

10
userapi/storage/postgres/deltas/2022021013023800_add_account_type.go
View file

@ -9,10 +9,10 @@ import (
func UpAddAccountType(ctx context.Context, tx *sql.Tx) error { func UpAddAccountType(ctx context.Context, tx *sql.Tx) error {
// initially set every account to useraccount, change appservice and guest accounts afterwards // initially set every account to useraccount, change appservice and guest accounts afterwards
// (user = 1, guest = 2, admin = 3, appservice = 4) // (user = 1, guest = 2, admin = 3, appservice = 4)
_, err := tx.ExecContext(ctx, `ALTER TABLE account_accounts ADD COLUMN IF NOT EXISTS account_type SMALLINT NOT NULL DEFAULT 1; _, err := tx.ExecContext(ctx, `ALTER TABLE userapi_accounts ADD COLUMN IF NOT EXISTS account_type SMALLINT NOT NULL DEFAULT 1;
UPDATE account_accounts SET account_type = 4 WHERE appservice_id <> ''; UPDATE userapi_accounts SET account_type = 4 WHERE appservice_id <> '';
UPDATE account_accounts SET account_type = 2 WHERE localpart ~ '^[0-9]+$'; UPDATE userapi_accounts SET account_type = 2 WHERE localpart ~ '^[0-9]+$';
ALTER TABLE account_accounts ALTER COLUMN account_type DROP DEFAULT;`, ALTER TABLE userapi_accounts ALTER COLUMN account_type DROP DEFAULT;`,
) )
if err != nil { if err != nil {
return fmt.Errorf("failed to execute upgrade: %w", err) return fmt.Errorf("failed to execute upgrade: %w", err)
@ -21,7 +21,7 @@ ALTER TABLE account_accounts ALTER COLUMN account_type DROP DEFAULT;`,
} }
func DownAddAccountType(ctx context.Context, tx *sql.Tx) error { func DownAddAccountType(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, "ALTER TABLE account_accounts DROP COLUMN account_type;") _, err := tx.ExecContext(ctx, "ALTER TABLE userapi_accounts DROP COLUMN account_type;")
if err != nil { if err != nil {
return fmt.Errorf("failed to execute downgrade: %w", err) return fmt.Errorf("failed to execute downgrade: %w", err)
} }

51
userapi/storage/postgres/deltas/2022101711000000_rename_tables.go
View file

@ -4,6 +4,8 @@ import (
"context" "context"
"database/sql" "database/sql"
"fmt" "fmt"
"github.com/lib/pq"
) )
var renameTableMappings = map[string]string{ var renameTableMappings = map[string]string{
@ -33,19 +35,36 @@ var renameIndicesMappings = map[string]string{
} }
func UpRenameTables(ctx context.Context, tx *sql.Tx) error { func UpRenameTables(ctx context.Context, tx *sql.Tx) error {
// I know what you're thinking: you're wondering "why doesn't this use $1
// and pass variadic parameters to ExecContext?" — the answer is because
// PostgreSQL doesn't expect the table name to be specified as a substituted
// argument in that way so it results in a syntax error in the query.
for old, new := range renameTableMappings { for old, new := range renameTableMappings {
if _, err := tx.ExecContext(ctx, "ALTER TABLE IF EXISTS $1 RENAME TO $2;", old, new); err != nil { q := fmt.Sprintf(
"ALTER TABLE IF EXISTS %s RENAME TO %s;",
pq.QuoteIdentifier(old), pq.QuoteIdentifier(new),
)
if _, err := tx.ExecContext(ctx, q); err != nil {
return fmt.Errorf("rename table %q to %q error: %w", old, new, err) return fmt.Errorf("rename table %q to %q error: %w", old, new, err)
} }
} }
for old, new := range renameSequenceMappings { for old, new := range renameSequenceMappings {
if _, err := tx.ExecContext(ctx, "ALTER SEQUENCE IF EXISTS $1 RENAME TO $2;", old, new); err != nil { q := fmt.Sprintf(
return fmt.Errorf("rename sequence %q to %q error: %w", old, new, err) "ALTER SEQUENCE IF EXISTS %s RENAME TO %s;",
pq.QuoteIdentifier(old), pq.QuoteIdentifier(new),
)
if _, err := tx.ExecContext(ctx, q); err != nil {
return fmt.Errorf("rename table %q to %q error: %w", old, new, err)
} }
} }
for old, new := range renameIndicesMappings { for old, new := range renameIndicesMappings {
if _, err := tx.ExecContext(ctx, "ALTER INDEX IF EXISTS $1 RENAME TO $2;", old, new); err != nil { q := fmt.Sprintf(
return fmt.Errorf("rename index %q to %q error: %w", old, new, err) "ALTER INDEX IF EXISTS %s RENAME TO %s;",
pq.QuoteIdentifier(old), pq.QuoteIdentifier(new),
)
if _, err := tx.ExecContext(ctx, q); err != nil {
return fmt.Errorf("rename table %q to %q error: %w", old, new, err)
} }
} }
return nil return nil
@ -53,18 +72,30 @@ func UpRenameTables(ctx context.Context, tx *sql.Tx) error {
func DownRenameTables(ctx context.Context, tx *sql.Tx) error { func DownRenameTables(ctx context.Context, tx *sql.Tx) error {
for old, new := range renameTableMappings { for old, new := range renameTableMappings {
if _, err := tx.ExecContext(ctx, "ALTER TABLE IF EXISTS $1 RENAME TO $2;", new, old); err != nil { q := fmt.Sprintf(
"ALTER TABLE IF EXISTS %s RENAME TO %s;",
pq.QuoteIdentifier(new), pq.QuoteIdentifier(old),
)
if _, err := tx.ExecContext(ctx, q); err != nil {
return fmt.Errorf("rename table %q to %q error: %w", new, old, err) return fmt.Errorf("rename table %q to %q error: %w", new, old, err)
} }
} }
for old, new := range renameSequenceMappings { for old, new := range renameSequenceMappings {
if _, err := tx.ExecContext(ctx, "ALTER SEQUENCE IF EXISTS $1 RENAME TO $2;", new, old); err != nil { q := fmt.Sprintf(
return fmt.Errorf("rename sequence %q to %q error: %w", new, old, err) "ALTER SEQUENCE IF EXISTS %s RENAME TO %s;",
pq.QuoteIdentifier(new), pq.QuoteIdentifier(old),
)
if _, err := tx.ExecContext(ctx, q); err != nil {
return fmt.Errorf("rename table %q to %q error: %w", new, old, err)
} }
} }
for old, new := range renameIndicesMappings { for old, new := range renameIndicesMappings {
if _, err := tx.ExecContext(ctx, "ALTER INDEX IF EXISTS $1 RENAME TO $2;", new, old); err != nil { q := fmt.Sprintf(
return fmt.Errorf("rename index %q to %q error: %w", new, old, err) "ALTER INDEX IF EXISTS %s RENAME TO %s;",
pq.QuoteIdentifier(new), pq.QuoteIdentifier(old),
)
if _, err := tx.ExecContext(ctx, q); err != nil {
return fmt.Errorf("rename table %q to %q error: %w", new, old, err)
} }
} }
return nil return nil

16
userapi/storage/postgres/stats_table.go
View file

@ -45,7 +45,7 @@ CREATE INDEX IF NOT EXISTS userapi_daily_visits_localpart_timestamp_idx ON usera
const countUsersLastSeenAfterSQL = "" + const countUsersLastSeenAfterSQL = "" +
"SELECT COUNT(*) FROM (" + "SELECT COUNT(*) FROM (" +
" SELECT localpart FROM device_devices WHERE last_seen_ts > $1 " + " SELECT localpart FROM userapi_devices WHERE last_seen_ts > $1 " +
" GROUP BY localpart" + " GROUP BY localpart" +
" ) u" " ) u"
@ -62,7 +62,7 @@ R30Users counts the number of 30 day retained users, defined as:
const countR30UsersSQL = ` const countR30UsersSQL = `
SELECT platform, COUNT(*) FROM ( SELECT platform, COUNT(*) FROM (
SELECT users.localpart, platform, users.created_ts, MAX(uip.last_seen_ts) SELECT users.localpart, platform, users.created_ts, MAX(uip.last_seen_ts)
FROM account_accounts users FROM userapi_accounts users
INNER JOIN INNER JOIN
(SELECT (SELECT
localpart, last_seen_ts, localpart, last_seen_ts,
@ -75,7 +75,7 @@ SELECT platform, COUNT(*) FROM (
ELSE 'unknown' ELSE 'unknown'
END END
AS platform AS platform
FROM device_devices FROM userapi_devices
) uip ) uip
ON users.localpart = uip.localpart ON users.localpart = uip.localpart
AND users.account_type <> 4 AND users.account_type <> 4
@ -121,7 +121,7 @@ GROUP BY client_type
` `
const countUserByAccountTypeSQL = ` const countUserByAccountTypeSQL = `
SELECT COUNT(*) FROM account_accounts WHERE account_type = ANY($1) SELECT COUNT(*) FROM userapi_accounts WHERE account_type = ANY($1)
` `
// $1 = All non guest AccountType IDs // $1 = All non guest AccountType IDs
@ -134,7 +134,7 @@ SELECT user_type, COUNT(*) AS count FROM (
WHEN account_type = $2 AND appservice_id IS NULL THEN 'guest' WHEN account_type = $2 AND appservice_id IS NULL THEN 'guest'
WHEN account_type = ANY($1) AND appservice_id IS NOT NULL THEN 'bridged' WHEN account_type = ANY($1) AND appservice_id IS NOT NULL THEN 'bridged'
END AS user_type END AS user_type
FROM account_accounts FROM userapi_accounts
WHERE created_ts > $3 WHERE created_ts > $3
) AS t GROUP BY user_type ) AS t GROUP BY user_type
` `
@ -143,14 +143,14 @@ SELECT user_type, COUNT(*) AS count FROM (
const updateUserDailyVisitsSQL = ` const updateUserDailyVisitsSQL = `
INSERT INTO userapi_daily_visits(localpart, device_id, timestamp, user_agent) INSERT INTO userapi_daily_visits(localpart, device_id, timestamp, user_agent)
SELECT u.localpart, u.device_id, $1, MAX(u.user_agent) SELECT u.localpart, u.device_id, $1, MAX(u.user_agent)
FROM device_devices AS u FROM userapi_devices AS u
LEFT JOIN ( LEFT JOIN (
SELECT localpart, device_id, timestamp FROM userapi_daily_visits SELECT localpart, device_id, timestamp FROM userapi_daily_visits
WHERE timestamp = $1 WHERE timestamp = $1
) udv ) udv
ON u.localpart = udv.localpart AND u.device_id = udv.device_id ON u.localpart = udv.localpart AND u.device_id = udv.device_id
INNER JOIN device_devices d ON d.localpart = u.localpart INNER JOIN userapi_devices d ON d.localpart = u.localpart
INNER JOIN account_accounts a ON a.localpart = u.localpart INNER JOIN userapi_accounts a ON a.localpart = u.localpart
WHERE $2 <= d.last_seen_ts AND d.last_seen_ts < $3 WHERE $2 <= d.last_seen_ts AND d.last_seen_ts < $3
AND a.account_type in (1, 3) AND a.account_type in (1, 3)
GROUP BY u.localpart, u.device_id GROUP BY u.localpart, u.device_id

2
userapi/storage/postgres/storage.go
View file

@ -23,8 +23,8 @@ import (
"github.com/matrix-org/dendrite/internal/sqlutil" "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup/base" "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config" "github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/userapi/storage/postgres/deltas"
"github.com/matrix-org/dendrite/userapi/storage/shared" "github.com/matrix-org/dendrite/userapi/storage/shared"
"github.com/matrix-org/dendrite/userapi/storage/sqlite3/deltas"
// Import the postgres database driver. // Import the postgres database driver.
_ "github.com/lib/pq" _ "github.com/lib/pq"