mirror of
https://github.com/thelounge/thelounge.git
synced 2026-03-14 14:35:50 +01:00
sqlite: add migrations support and introduce primary key
Add the ability to migrate our db in the upwards direction. Use the facility to add primary keys to our messages table. This should allow work like jumping to messages and the likes. This also introduces the framework for rollback, without actually hooking it up. This should be easy enough to do when the need arises.
This commit is contained in:
Reto Brunner
commit
2ef8b37009
2 changed files with 422 additions and 57 deletions
|
|
@ -23,14 +23,74 @@ try {
|
|||
);
|
||||
}
|
||||
|
||||
const currentSchemaVersion = 1520239200;
|
||||
type Migration = {version: number; stmts: string[]};
|
||||
type Rollback = {version: number; rollback_forbidden?: boolean; stmts: string[]};
|
||||
|
||||
export const currentSchemaVersion = 1679743888000; // use `new Date().getTime()`
|
||||
|
||||
// Desired schema, adapt to the newest version and add migrations to the array below
|
||||
const schema = [
|
||||
// Schema version #1
|
||||
"CREATE TABLE IF NOT EXISTS options (name TEXT, value TEXT, CONSTRAINT name_unique UNIQUE (name))",
|
||||
"CREATE TABLE IF NOT EXISTS messages (network TEXT, channel TEXT, time INTEGER, type TEXT, msg TEXT)",
|
||||
"CREATE INDEX IF NOT EXISTS network_channel ON messages (network, channel)",
|
||||
"CREATE INDEX IF NOT EXISTS time ON messages (time)",
|
||||
"CREATE TABLE options (name TEXT, value TEXT, CONSTRAINT name_unique UNIQUE (name))",
|
||||
"CREATE TABLE messages (id INTEGER PRIMARY KEY AUTOINCREMENT, network TEXT, channel TEXT, time INTEGER, type TEXT, msg TEXT)",
|
||||
`CREATE TABLE migrations (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
version INTEGER NOT NULL UNIQUE,
|
||||
rollback_forbidden INTEGER DEFAULT 0 NOT NULL
|
||||
)`,
|
||||
`CREATE TABLE rollback_steps (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
migration_id INTEGER NOT NULL REFERENCES migrations ON DELETE CASCADE,
|
||||
step INTEGER NOT NULL,
|
||||
statement TEXT NOT NULL
|
||||
)`,
|
||||
"CREATE INDEX network_channel ON messages (network, channel)",
|
||||
"CREATE INDEX time ON messages (time)",
|
||||
];
|
||||
|
||||
// the migrations will be executed in an exclusive transaction as a whole
|
||||
// add new migrations to the end, with the version being the new 'currentSchemaVersion'
|
||||
// write a corresponding down migration into rollbacks
|
||||
export const migrations: Migration[] = [
|
||||
{
|
||||
version: 1672236339873,
|
||||
stmts: [
|
||||
"CREATE TABLE messages_new (id INTEGER PRIMARY KEY AUTOINCREMENT, network TEXT, channel TEXT, time INTEGER, type TEXT, msg TEXT)",
|
||||
"INSERT INTO messages_new(network, channel, time, type, msg) select network, channel, time, type, msg from messages order by time asc",
|
||||
"DROP TABLE messages",
|
||||
"ALTER TABLE messages_new RENAME TO messages",
|
||||
"CREATE INDEX network_channel ON messages (network, channel)",
|
||||
"CREATE INDEX time ON messages (time)",
|
||||
],
|
||||
},
|
||||
{
|
||||
version: 1679743888000,
|
||||
stmts: [
|
||||
`CREATE TABLE IF NOT EXISTS migrations (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
version INTEGER NOT NULL UNIQUE,
|
||||
rollback_forbidden INTEGER DEFAULT 0 NOT NULL
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS rollback_steps (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
migration_id INTEGER NOT NULL REFERENCES migrations ON DELETE CASCADE,
|
||||
step INTEGER NOT NULL,
|
||||
statement TEXT NOT NULL
|
||||
)`,
|
||||
],
|
||||
},
|
||||
];
|
||||
|
||||
// down migrations need to restore the state of the prior version.
|
||||
// rollback can be disallowed by adding rollback_forbidden: true to it
|
||||
export const rollbacks: Rollback[] = [
|
||||
{
|
||||
version: 1672236339873,
|
||||
stmts: [], // changes aren't visible, left empty on purpose
|
||||
},
|
||||
{
|
||||
version: 1679743888000,
|
||||
stmts: [], // here we can't drop the tables, as we use them in the code, so just leave those in
|
||||
},
|
||||
];
|
||||
|
||||
class Deferred {
|
||||
|
|
@ -71,6 +131,7 @@ class SqliteMessageStorage implements SearchableMessageStorage {
|
|||
this.database = new sqlite3.Database(sqlitePath);
|
||||
|
||||
try {
|
||||
await this.run_pragmas(); // must be done outside of a transaction
|
||||
await this.run_migrations();
|
||||
} catch (e) {
|
||||
this.isEnabled = false;
|
||||
|
|
@ -86,41 +147,89 @@ class SqliteMessageStorage implements SearchableMessageStorage {
|
|||
}
|
||||
}
|
||||
|
||||
async run_migrations() {
|
||||
async setup_new_db() {
|
||||
for (const stmt of schema) {
|
||||
await this.serialize_run(stmt, []);
|
||||
}
|
||||
|
||||
await this.serialize_run("INSERT INTO options (name, value) VALUES ('schema_version', ?)", [
|
||||
currentSchemaVersion.toString(),
|
||||
]);
|
||||
}
|
||||
|
||||
async current_version(): Promise<number> {
|
||||
const have_options = await this.serialize_get(
|
||||
"select 1 from sqlite_master where type = 'table' and name = 'options'"
|
||||
);
|
||||
|
||||
if (!have_options) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const version = await this.serialize_get(
|
||||
"SELECT value FROM options WHERE name = 'schema_version'"
|
||||
);
|
||||
|
||||
if (version === undefined) {
|
||||
// new table
|
||||
await this.serialize_run(
|
||||
"INSERT INTO options (name, value) VALUES ('schema_version', ?)",
|
||||
[currentSchemaVersion]
|
||||
);
|
||||
return;
|
||||
// technically shouldn't happen, means something created a schema but didn't populate it
|
||||
// we'll try our best to recover
|
||||
return 0;
|
||||
}
|
||||
|
||||
const storedSchemaVersion = parseInt(version.value, 10);
|
||||
return storedSchemaVersion;
|
||||
}
|
||||
|
||||
if (storedSchemaVersion === currentSchemaVersion) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (storedSchemaVersion > currentSchemaVersion) {
|
||||
throw `sqlite messages schema version is higher than expected (${storedSchemaVersion} > ${currentSchemaVersion}). Is The Lounge out of date?`;
|
||||
}
|
||||
async update_version_in_db() {
|
||||
return this.serialize_run("UPDATE options SET value = ? WHERE name = 'schema_version'", [
|
||||
currentSchemaVersion.toString(),
|
||||
]);
|
||||
}
|
||||
|
||||
async _run_migrations(dbVersion: number) {
|
||||
log.info(
|
||||
`sqlite messages schema version is out of date (${storedSchemaVersion} < ${currentSchemaVersion}). Running migrations if any.`
|
||||
`sqlite messages schema version is out of date (${dbVersion} < ${currentSchemaVersion}). Running migrations.`
|
||||
);
|
||||
|
||||
await this.serialize_run("UPDATE options SET value = ? WHERE name = 'schema_version'", [
|
||||
currentSchemaVersion,
|
||||
]);
|
||||
const to_execute = necessaryMigrations(dbVersion);
|
||||
|
||||
for (const stmt of to_execute.map((m) => m.stmts).flat()) {
|
||||
await this.serialize_run(stmt, []);
|
||||
}
|
||||
|
||||
await this.update_version_in_db();
|
||||
}
|
||||
|
||||
async run_pragmas() {
|
||||
await this.serialize_run("PRAGMA foreign_keys = ON;", []);
|
||||
}
|
||||
|
||||
async run_migrations() {
|
||||
const version = await this.current_version();
|
||||
|
||||
if (version > currentSchemaVersion) {
|
||||
throw `sqlite messages schema version is higher than expected (${version} > ${currentSchemaVersion}). Is The Lounge out of date?`;
|
||||
} else if (version === currentSchemaVersion) {
|
||||
return; // nothing to do
|
||||
}
|
||||
|
||||
await this.serialize_run("BEGIN EXCLUSIVE TRANSACTION", []);
|
||||
|
||||
try {
|
||||
if (version === 0) {
|
||||
await this.setup_new_db();
|
||||
} else {
|
||||
await this._run_migrations(version);
|
||||
}
|
||||
|
||||
await this.insert_rollback_since(version);
|
||||
} catch (err) {
|
||||
await this.serialize_run("ROLLBACK", []);
|
||||
throw err;
|
||||
}
|
||||
|
||||
await this.serialize_run("COMMIT", []);
|
||||
await this.serialize_run("VACUUM", []);
|
||||
}
|
||||
|
||||
async close() {
|
||||
|
|
@ -142,6 +251,116 @@ class SqliteMessageStorage implements SearchableMessageStorage {
|
|||
});
|
||||
}
|
||||
|
||||
async fetch_rollbacks(since_version: number) {
|
||||
const res = await this.serialize_fetchall(
|
||||
`select version, rollback_forbidden, statement
|
||||
from rollback_steps
|
||||
join migrations on migrations.id=rollback_steps.migration_id
|
||||
where version > ?
|
||||
order by version desc, step asc`,
|
||||
since_version
|
||||
);
|
||||
const result: Rollback[] = [];
|
||||
|
||||
// convert to Rollback[]
|
||||
// requires ordering in the sql statement
|
||||
for (const raw of res) {
|
||||
const last = result.at(-1);
|
||||
|
||||
if (!last || raw.version !== last.version) {
|
||||
result.push({
|
||||
version: raw.version,
|
||||
rollback_forbidden: Boolean(raw.rollback_forbidden),
|
||||
stmts: [raw.statement],
|
||||
});
|
||||
} else {
|
||||
last.stmts.push(raw.statment);
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
async delete_migrations_older_than(version: number) {
|
||||
return this.serialize_run("delete from migrations where migrations.version > ?", [version]);
|
||||
}
|
||||
|
||||
async _downgrade_to(version: number) {
|
||||
const _rollbacks = await this.fetch_rollbacks(version);
|
||||
|
||||
if (_rollbacks.length === 0) {
|
||||
return version;
|
||||
}
|
||||
|
||||
const forbidden = _rollbacks.find((item) => item.rollback_forbidden);
|
||||
|
||||
if (forbidden) {
|
||||
throw Error(`can't downgrade past ${forbidden.version}`);
|
||||
}
|
||||
|
||||
for (const rollback of _rollbacks) {
|
||||
for (const stmt of rollback.stmts) {
|
||||
await this.serialize_run(stmt, []);
|
||||
}
|
||||
}
|
||||
|
||||
await this.delete_migrations_older_than(version);
|
||||
await this.update_version_in_db();
|
||||
|
||||
return _rollbacks.at(-1)!.version; // assert valid due to length guard above
|
||||
}
|
||||
|
||||
async downgrade_to(version: number) {
|
||||
if (version <= 0) {
|
||||
throw Error(`${version} is not a valid version to downgrade to`);
|
||||
}
|
||||
|
||||
await this.serialize_run("BEGIN EXCLUSIVE TRANSACTION", []);
|
||||
|
||||
let new_version: number;
|
||||
|
||||
try {
|
||||
new_version = await this._downgrade_to(version);
|
||||
} catch (err) {
|
||||
await this.serialize_run("ROLLBACK", []);
|
||||
throw err;
|
||||
}
|
||||
|
||||
await this.serialize_run("COMMIT", []);
|
||||
return new_version;
|
||||
}
|
||||
|
||||
async downgrade() {
|
||||
const res = await this.downgrade_to(currentSchemaVersion);
|
||||
return res;
|
||||
}
|
||||
|
||||
async insert_rollback_since(version: number) {
|
||||
const missing = newRollbacks(version);
|
||||
|
||||
for (const rollback of missing) {
|
||||
const migration = await this.serialize_get(
|
||||
`insert into migrations
|
||||
(version, rollback_forbidden)
|
||||
values (?, ?)
|
||||
returning id`,
|
||||
rollback.version,
|
||||
rollback.rollback_forbidden || 0
|
||||
);
|
||||
|
||||
for (const stmt of rollback.stmts) {
|
||||
let step = 0;
|
||||
await this.serialize_run(
|
||||
`insert into rollback_steps
|
||||
(migration_id, step, statement)
|
||||
values (?, ?, ?)`,
|
||||
[migration.id, step, stmt]
|
||||
);
|
||||
step++;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async index(network: Network, channel: Chan, msg: Msg) {
|
||||
await this.initDone.promise;
|
||||
|
||||
|
|
@ -326,4 +545,12 @@ function parseSearchRowsToMessages(id: number, rows: any[]) {
|
|||
return messages;
|
||||
}
|
||||
|
||||
export function necessaryMigrations(since: number): Migration[] {
|
||||
return migrations.filter((m) => m.version > since);
|
||||
}
|
||||
|
||||
export function newRollbacks(since: number): Rollback[] {
|
||||
return rollbacks.filter((r) => r.version > since);
|
||||
}
|
||||
|
||||
export default SqliteMessageStorage;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue