moshing-mammut/src/lib/server/db.ts

import { env } from '$env/dynamic/private';
import { IGNORE_USERS, MASTODON_INSTANCE } from '$env/static/private';
import type { Account, Post, Tag } from '$lib/mastodon/response';
import { isTruthy } from '$lib/truthyString';
import sqlite3 from 'sqlite3';
const { DEV } = import.meta.env;

const db: sqlite3.Database = new sqlite3.Database('moshingmammut.db');
// for the local masto instance, the instance name is *not* saved
// as part of the username or acct, so it needs to be stripped
const ignoredUsers: string[] =
  IGNORE_USERS === undefined
    ? []
    : IGNORE_USERS.split(',')
        .map((u) => (u.startsWith('@') ? u.substring(1) : u))
        .map((u) =>
          u.endsWith('@' + MASTODON_INSTANCE)
            ? u.substring(0, u.length - ('@' + MASTODON_INSTANCE).length)
            : u
        );
let databaseReady = false;

if (DEV && isTruthy(env.VERBOSE)) {
  sqlite3.verbose();
  db.on('change', (t, d, table, rowid) => {
    console.debug('DB change event', t, d, table, rowid);
  });

  db.on('trace', (sql) => {
    console.debug('Running', sql);
  });

  db.on('profile', (sql) => {
    console.debug('Finished', sql);
  });
}

interface Migration {
  id: number;
  name: string;
  statement: string;
}

db.on('open', () => {
  console.log('Opened database');
  db.serialize();
  db.run('CREATE TABLE IF NOT EXISTS "migrations" ("id" integer,"name" TEXT, PRIMARY KEY (id))');
  db.all('SELECT id FROM migrations', (err, rows) => {
    if (err !== null) {
      console.error('Could not fetch existing migrations', err);
      databaseReady = true;
      return;
    }
    console.debug('Already applied migrations', rows);
    const appliedMigrations: Set<number> = new Set(rows.map((row: any) => row['id']));
    const toApply = getMigrations().filter((m) => !appliedMigrations.has(m.id));
    let remaining = toApply.length;
    if (remaining === 0) {
      databaseReady = true;
      return;
    }
    for (const migration of toApply) {
      db.exec(migration.statement, (err) => {
        remaining--;
        // This will set databaseReady to true before the migration has been inserted as applies,
        // but that doesn't matter. It's only important that is has been applied
        if (remaining === 0) {
          databaseReady = true;
        }
        if (err !== null) {
          console.error(`Failed to apply migration ${migration.name}`, err);
          return;
        }
        db.run(
          'INSERT INTO migrations (id, name) VALUES(?, ?)',
          [migration.id, migration.name],
          (e: Error) => {
            if (e !== null) {
              console.error(`Failed to mark migration ${migration.name} as applied`, e);
              return;
            }
            console.info(`Applied migration ${migration.name}`);
          }
        );
      });
    }
  });
});
db.on('error', (err) => {
  console.error('Error opening database', err);
});

function getMigrations(): Migration[] {
  return [
    {
      id: 1,
      name: 'initial',
      statement: `
      CREATE TABLE accounts (
        id TEXT NOT NULL PRIMARY KEY,
        acct TEXT,
        username TEXT,
        display_name TEXT,
        url TEXT,
        avatar TEXT,
        avatar_static TEXT
      );
      CREATE TABLE tags (url TEXT NOT NULL PRIMARY KEY, tag TEXT NOT NULL);
      CREATE TABLE posts (
        id TEXT NOT NULL PRIMARY KEY,
        content TEXT,
        created_at TEXT,
        url TEXT NOT NULL,
        account_id TEXT NOT NULL,
        FOREIGN KEY (account_id) REFERENCES accounts(id)
      );
      CREATE TABLE poststags (
        id integer PRIMARY KEY,
        post_id TEXT NOT NULL,
        tag_url TEXT NOT NULL,
        FOREIGN KEY (post_id) REFERENCES posts(id),
        FOREIGN KEY (tag_url) REFERENCES tags(url)
      )`
    },
    {
      id: 2,
      name: 'urls as keys',
      statement: `
      CREATE TABLE accounts_new (
        id TEXT NOT NULL,
        acct TEXT,
        username TEXT,
        display_name TEXT,
        url TEXT NOT NULL PRIMARY KEY,
        avatar TEXT
      );
      INSERT INTO accounts_new (id, acct, username, display_name, url, avatar)
         SELECT id, acct, username, display_name, url, avatar
         FROM accounts;
      DROP TABLE accounts;
      ALTER TABLE accounts_new RENAME TO accounts;

      CREATE TABLE posts_new (
        id TEXT NOT NULL,
        content TEXT,
        created_at TEXT,
        url TEXT NOT NULL PRIMARY KEY,
        account_id TEXT NOT NULL,
        FOREIGN KEY (account_id) REFERENCES accounts(url)
      );
      INSERT INTO posts_new (id, content, created_at, url, account_id)
         SELECT p.id, p.content, p.created_at, p.url, accounts.url
         FROM posts as p
         JOIN accounts ON accounts.id = p.account_id;
      DROP TABLE posts;
      ALTER TABLE posts_new RENAME TO posts;

      CREATE TABLE poststags_new (
        id integer PRIMARY KEY,
        post_id TEXT NOT NULL,
        tag_url TEXT NOT NULL,
        FOREIGN KEY (post_id) REFERENCES posts(url),
        FOREIGN KEY (tag_url) REFERENCES tags(url)
      );

      INSERT INTO poststags_new (id, post_id, tag_url)
         SELECT pt.id, posts.url, pt.tag_url
         FROM poststags as pt
         JOIN posts ON posts.id = pt.post_id;
      DROP TABLE poststags;
      ALTER TABLE poststags_new RENAME TO poststags;
      `
    }
  ];
}

async function waitReady(): Promise<undefined> {
  // Simpler than a semaphore and is really only needed on startup
  return new Promise((resolve) => {
    const interval = setInterval(() => {
      if (DEV) {
        console.debug('Waiting for database to be ready');
      }
      if (databaseReady) {
        if (DEV) {
          console.debug('DB is ready');
        }
        clearInterval(interval);
        resolve(undefined);
      }
    }, 100);
  });
}

export async function savePost(post: Post): Promise<undefined> {
  if (!databaseReady) {
    await waitReady();
  }
  return await new Promise<undefined>((resolve, reject) => {
    console.debug(`Saving post ${post.url}`);
    const account = post.account;
    db.run(
      `
      INSERT INTO accounts (id, acct, username, display_name, url, avatar)
      VALUES(?, ?, ?, ?, ?, ?)
      ON CONFLICT(url)
      DO UPDATE SET
      acct=excluded.acct,
      username=excluded.username,
      display_name=excluded.display_name,
      id=excluded.id,
      avatar=excluded.avatar;`,
      [
        account.id,
        account.acct,
        account.username,
        account.display_name,
        account.url,
        account.avatar
      ],
      (err) => {
        if (err !== null) {
          console.error(`Could not insert/update account ${account.id}`, err);
          reject(err);
          return;
        }
        db.run(
          `
          INSERT INTO posts (id, content, created_at, url, account_id)
          VALUES (?, ?, ?, ?, ?) ON CONFLICT(url) DO UPDATE SET
          content=excluded.content,
          created_at=excluded.created_at,
          id=excluded.id,
          account_id=excluded.account_id;`,
          [post.id, post.content, post.created_at, post.url, post.account.url],
          (postErr) => {
            if (postErr !== null) {
              console.error(`Could not insert post ${post.url}`, postErr);
              reject(postErr);
              return;
            }

            db.parallelize(() => {
              let remaining = post.tags.length;
              for (const tag of post.tags) {
                db.run(
                  `
                  INSERT INTO tags (url, tag) VALUES (?, ?)
                  ON CONFLICT(url) DO UPDATE SET
                  tag=excluded.tag;`,
                  [tag.url, tag.name],
                  (tagErr) => {
                    if (tagErr !== null) {
                      console.error(`Could not insert/update tag ${tag.url}`, tagErr);
                      reject(tagErr);
                      return;
                    }
                    db.run(
                      'INSERT INTO poststags (post_id, tag_url) VALUES (?, ?)',
                      [post.url, tag.url],
                      (posttagserr) => {
                        if (posttagserr !== null) {
                          console.error(
                            `Could not insert poststags ${tag.url}, ${post.url}`,
                            posttagserr
                          );
                          reject(posttagserr);
                          return;
                        }
                        // Don't decrease on fail
                        remaining--;
                        // Only resolve after all have been inserted
                        if (remaining === 0) {
                          resolve(undefined);
                        }
                      }
                    );
                  }
                );
              }
            });
          }
        );
      }
    );
  });
}

export async function getPosts(since: string | null, before: string | null, limit: number) {
  if (!databaseReady) {
    await waitReady();
  }
  const promise = await new Promise<Post[]>((resolve, reject) => {
    let filter_query = '';
    const params: any = { $limit: limit };
    if (since === null && before === null) {
      filter_query = '';
    } else if (since !== null) {
      filter_query = 'WHERE posts.created_at > $since';
      params.$since = since;
    } else if (before !== null) {
      // Setting both, before and since doesn't make sense, so this case is not explicitly handled
      filter_query = 'WHERE posts.created_at < $before';
      params.$before = before;
    }

    ignoredUsers.forEach((ignoredUser, index) => {
      const userParam = `$user_${index}`;
      const acctParam = userParam + 'a';
      const usernameParam = userParam + 'u';
      const prefix = filter_query === '' ? ' WHERE' : ' AND';
      filter_query += `${prefix} acct != ${acctParam} AND username != ${usernameParam} `;
      params[acctParam] = ignoredUser;
      params[usernameParam] = ignoredUser;
    });

    const sql = `SELECT posts.id, posts.content, posts.created_at, posts.url,
      accounts.id AS account_id, accounts.acct, accounts.username, accounts.display_name,
      accounts.url AS account_url, accounts.avatar
      FROM posts
      JOIN accounts ON posts.account_id = accounts.url
      ${filter_query}
      ORDER BY created_at DESC
      LIMIT $limit`;
    db.all(sql, params, (err, rows: any[]) => {
      if (err != null) {
        console.error('Error loading posts', err);
        reject(err);
        return;
      }
      if (rows.length === 0) {
        // No need to check for tags
        resolve([]);
        return;
      }
      const postIdsParams = rows.map(() => '?').join(', ');
      db.all(
        `SELECT post_id, tags.url, tags.tag
          FROM poststags
          JOIN tags ON poststags.tag_url = tags.url
          WHERE post_id IN (${postIdsParams});`,
        rows.map((r: any) => r.url),
        (tagErr, tagRows: any[]) => {
          if (tagErr != null) {
            console.error('Error loading post tags', tagErr);
            reject(tagErr);
            return;
          }
          const tagMap: Map<string, Tag[]> = tagRows.reduce((result: Map<string, Tag[]>, item) => {
            const tag: Tag = {
              url: item.url,
              name: item.tag
            };
            result.set(item.post_id, [...(result.get(item.post_id) || []), tag]);
            return result;
          }, new Map());
          const posts = rows.map((row) => {
            return {
              id: row.id,
              content: row.content,
              created_at: row.created_at,
              url: row.url,
              tags: tagMap.get(row.id) || [],
              account: {
                id: row.account_id,
                acct: row.acct,
                username: row.username,
                display_name: row.display_name,
                url: row.account_url,
                avatar: row.avatar
              } as Account
            } as Post;
          });
          resolve(posts);
        }
      );
    });
  });
  return promise;
}