Migrate to SQLite (#174)

* Migrating Fredy from LowDb to SqLite 🎉

* adding new sql migration system for future sql migrations

* adding setting to change  sqlite path for db files

* create migration plan for graceful migration lowdb -> sqlite

* Improving Documentation

* adding test for sqliteconnection

* upgrading dependencies

* making nodejs 22 as min version

* improve scraper

* adding overwrite ability for db migra

lib/services/demoCleanup.js

@@ -1,5 +1,5 @@
 import { setInterval } from 'node:timers';
-import { removeJobsByUserName } from './storage/jobStorage.js';
+import { removeJobsByUserId } from './storage/jobStorage.js';
 import { config } from '../utils.js';
 import { getUsers } from './storage/userStorage.js';
 import logger from './logger.js';
@@ -33,6 +33,6 @@ function cleanup() {
       logger.error('Demo user not found, cannot remove Jobs');
       return;
     }
-    removeJobsByUserName(demoUser.id);
+    removeJobsByUserId(demoUser.id);
   }
 }

lib/services/storage/LowDashAdapter.js

@@ -1,8 +0,0 @@
-import lodash from 'lodash';
-import { LowSync } from 'lowdb';
-export default class LowdashAdapter extends LowSync {
-  constructor(adapter, defaultData = {}) {
-    super(adapter, defaultData);
-    this.chain = lodash.chain(this).get('data');
-  }
-}

lib/services/storage/SqliteConnection.js

@@ -0,0 +1,140 @@
+import fs from 'fs';
+import path from 'path';
+import Database from 'better-sqlite3';
+import logger from '../../services/logger.js';
+import { config } from '../../utils.js';
+
+/**
+ * SqliteConnection
+ * A small, high-performance wrapper around better-sqlite3 that provides a
+ * singleton connection, sensible PRAGMA tuning, and helper methods. This
+ * module is safe to import and reuse.
+ *
+ * Performance notes:
+ * - journal_mode = WAL: allows concurrent readers with a single writer and
+ *   yields better performance for server apps.
+ * - synchronous = NORMAL: trades a bit of durability for significant speed
+ *   while still being safe in most environments.
+ * - cache_size = -64000: ~64MB page cache (negative value sets KB) to improve
+ *   query performance for frequent reads.
+ * - foreign_keys = ON: ensure referential integrity is enforced.
+ * - optimize: runs SQLite's auto-analysis and purges internal caches. It is
+ *   cheap; we call it at startup and before process exit. You can also call
+ *   optimize() manually after large schema changes or bulk operations.
+ */
+class SqliteConnection {
+  static #db = null;
+
+  /**
+   * Returns a singleton instance of better-sqlite3 Database.
+   * Respects env var SQLITE_DB_PATH and defaults to db/listings.db.
+   */
+  static getConnection() {
+    if (this.#db) return this.#db;
+
+    // Interpret config.sqlitepath as a directory relative to project root when it starts with '/'
+    const cfg = typeof config === 'object' && config ? config.sqlitepath : undefined;
+    const rawDir = cfg && cfg.length > 0 ? cfg : '/db';
+    const relDir = rawDir.startsWith('/') ? rawDir.slice(1) : rawDir;
+    const absDir = path.isAbsolute(relDir) ? relDir : path.join(process.cwd(), relDir);
+    const dbPath = path.join(absDir, 'listings.db');
+
+    // Ensure directory exists
+    const dir = path.dirname(dbPath);
+    if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });
+
+    // Open the database synchronously (better-sqlite3 is sync and very fast)
+    this.#db = new Database(dbPath, { verbose: undefined });
+
+    // Apply high-performance PRAGMA's
+    try {
+      this.#db.pragma('journal_mode = WAL');
+      this.#db.pragma('synchronous = NORMAL');
+      this.#db.pragma('cache_size = -64000');
+      this.#db.pragma('foreign_keys = ON');
+      this.#db.pragma('optimize');
+    } catch (e) {
+      logger.warn('Failed to apply one or more PRAGMAs:', e.message);
+    }
+
+    // Run optimize on exit to persist analysis and cleanup internal caches.
+    process.once('beforeExit', () => {
+      try {
+        this.#db?.pragma('optimize');
+      } catch (e) {
+        logger.debug('PRAGMA optimize on exit failed:', e.message);
+      }
+    });
+
+    return this.#db;
+  }
+
+  /**
+   * Execute a write statement (INSERT/UPDATE/DELETE/DDL). Returns better-sqlite3 run info.
+   */
+  static execute(sql, params = {}) {
+    const db = this.getConnection();
+    return db.prepare(sql).run(params);
+  }
+
+  /**
+   * Execute a query and returns all rows.
+   */
+  static query(sql, params = {}) {
+    const db = this.getConnection();
+    return db.prepare(sql).all(params);
+  }
+
+  /**
+   * Check whether a table exists.
+   */
+  static tableExists(tableName) {
+    const db = this.getConnection();
+    const row = db.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?").get(tableName);
+    return !!row;
+  }
+
+  /**
+   * Run the given callback inside a transaction. The callback receives the Database instance.
+   * If the callback throws, the transaction is rolled back and the error re-thrown.
+   */
+  static withTransaction(callback) {
+    const db = this.getConnection();
+    const trx = db.transaction((cb) => cb(db));
+    return trx(callback);
+  }
+
+  /**
+   * Run SQLite PRAGMA optimize. See https://sqlite.org/pragma.html#pragma_optimize
+   *
+   * Explanation: PRAGMA optimize triggers internal housekeeping, such as
+   * recomputing query planner statistics (similar to ANALYZE) when appropriate
+   * and purging unused pages from caches. It is inexpensive and can improve
+   * performance after schema changes or heavy write activity.
+   */
+  static optimize() {
+    const db = this.getConnection();
+    try {
+      db.pragma('optimize');
+    } catch (e) {
+      logger.warn('PRAGMA optimize failed:', e.message);
+    }
+  }
+
+  /**
+   * Close the database connection. Typically not needed for long-running apps.
+   */
+  static close() {
+    if (this.#db) {
+      try {
+        this.#db.pragma('optimize');
+      } catch (e) {
+        logger.debug('PRAGMA optimize before close failed:', e.message);
+      }
+      this.#db.close();
+      this.#db = null;
+    }
+  }
+}
+
+export default SqliteConnection;

lib/services/storage/jobStorage.js

@@ -1,106 +1,144 @@
-import { JSONFileSync } from 'lowdb/node';
 import { nanoid } from 'nanoid';
-import * as listingStorage from './listingsStorage.js';
-import { getDirName } from '../../utils.js';
-import path from 'path';
-import LowdashAdapter from './LowDashAdapter.js';
+import SqliteConnection from './SqliteConnection.js';
 import logger from '../logger.js';
+import { toJson, fromJson } from '../../utils.js';
 
-const file = path.join(getDirName(), '../', 'db/jobs.json');
-const adapter = new JSONFileSync(file);
-const db = new LowdashAdapter(adapter, { jobs: [] });
-
-db.read();
-
+/**
+ * Insert or update a job. Preserves original owner (userId) when updating an existing job.
+ *
+ * @param {Object} params
+ * @param {string} [params.jobId] - Existing job id to update; omit to insert a new job.
+ * @param {string} [params.name] - Job display name.
+ * @param {Array<any>} [params.blacklist] - Blacklist entries; defaults to empty array.
+ * @param {boolean} [params.enabled] - Whether the job is enabled; defaults to true.
+ * @param {Array<any>} params.provider - Provider configuration list.
+ * @param {Array<any>} params.notificationAdapter - Notification adapter configuration list.
+ * @param {string} params.userId - Owner user id for inserts; preserved on updates.
+ * @returns {void}
+ */
 export const upsertJob = ({ jobId, name, blacklist = [], enabled = true, provider, notificationAdapter, userId }) => {
-  const currentJob =
-    jobId == null
-      ? null
-      : db.chain
-          .get('jobs')
-          .find((job) => job.id === jobId)
-          .value();
-  const jobs = db.chain
-    .get('jobs')
-    .filter((job) => job.id !== jobId)
-    .value();
-  jobs.push({
-    id: jobId || nanoid(),
-    //make sure to not overwrite the user id in case an admin changes the job
-    userId: currentJob == null ? userId : currentJob.userId,
-    enabled,
-    name,
-    blacklist,
-    provider,
-    notificationAdapter,
-  });
-  db.chain.set('jobs', jobs).value();
-  db.write();
-};
-export const getJob = (jobId) => {
-  const job = db.chain
-    .get('jobs')
-    .find((job) => job.id === jobId)
-    .value();
-  if (job == null) {
-    return null;
+  const id = jobId || nanoid();
+  const existing = SqliteConnection.query(`SELECT id, user_id FROM jobs WHERE id = @id LIMIT 1`, { id })[0];
+  const ownerId = existing ? existing.user_id : userId;
+  if (existing) {
+    SqliteConnection.execute(
+      `UPDATE jobs
+         SET enabled = @enabled,
+             name = @name,
+             blacklist = @blacklist,
+             provider = @provider,
+             notification_adapter = @notification_adapter
+       WHERE id = @id`,
+      {
+        id,
+        enabled: enabled ? 1 : 0,
+        name: name ?? null,
+        blacklist: toJson(blacklist ?? []),
+        provider: toJson(provider ?? []),
+        notification_adapter: toJson(notificationAdapter ?? []),
+      },
+    );
+  } else {
+    SqliteConnection.execute(
+      `INSERT INTO jobs (id, user_id, enabled, name, blacklist, provider, notification_adapter)
+       VALUES (@id, @user_id, @enabled, @name, @blacklist, @provider, @notification_adapter)`,
+      {
+        id,
+        user_id: ownerId,
+        enabled: enabled ? 1 : 0,
+        name: name ?? null,
+        blacklist: toJson(blacklist ?? []),
+        provider: toJson(provider ?? []),
+        notification_adapter: toJson(notificationAdapter ?? []),
+      },
+    );
   }
+};
+
+/**
+ * Get a single job by id.
+ * @param {string} jobId - Job primary key.
+ * @returns {Job|null} The job or null if not found.
+ */
+export const getJob = (jobId) => {
+  const row = SqliteConnection.query(
+    `SELECT j.id,
+            j.user_id AS userId,
+            j.enabled,
+            j.name,
+            j.blacklist,
+            j.provider,
+            j.notification_adapter AS notificationAdapter,
+            (SELECT COUNT(1) FROM listings l WHERE l.job_id = j.id) AS numberOfFoundListings
+       FROM jobs j
+      WHERE j.id = @id
+      LIMIT 1`,
+    { id: jobId },
+  )[0];
+  if (!row) return null;
   return {
-    ...job,
-    numberOfFoundListings: listingStorage.getNumberOfAllKnownListings(job.id).length,
+    ...row,
+    enabled: !!row.enabled,
+    blacklist: fromJson(row.blacklist, []),
+    provider: fromJson(row.provider, []),
+    notificationAdapter: fromJson(row.notificationAdapter, []),
   };
 };
+
+/**
+ * Update job enabled status.
+ * @param {{jobId: string, status: boolean}} params - Parameters.
+ * @returns {void}
+ */
 export const setJobStatus = ({ jobId, status }) => {
-  db.chain
-    .get('jobs')
-    .find((job) => job.id === jobId)
-    .assign({ enabled: status })
-    .value();
-  db.write();
+  SqliteConnection.execute(`UPDATE jobs SET enabled = @enabled WHERE id = @id`, {
+    id: jobId,
+    enabled: status ? 1 : 0,
+  });
 };
+
+/**
+ * Remove a job by id. Listings are deleted automatically due to FK ON DELETE CASCADE.
+ * @param {string} jobId - Job id.
+ * @returns {void}
+ */
 export const removeJob = (jobId) => {
-  listingStorage.removeListings(jobId);
-  db.chain
-    .get('jobs')
-    .remove((job) => job.id === jobId)
-    .value();
-  db.write();
+  // listings table has FK ON DELETE CASCADE via job_id
+  SqliteConnection.execute(`DELETE FROM jobs WHERE id = @id`, { id: jobId });
 };
+
 export const removeJobsByUserId = (userId) => {
-  db.chain
-    .get('jobs')
-    .filter((job) => job.userId === userId)
-    .forEach((job) => listingStorage.removeListings(job.id));
-  db.chain
-    .get('jobs')
-    .remove((job) => job.userId === userId)
-    .value();
-  db.write();
-};
-export const removeJobsByUserName = (userId) => {
-  let removedDemoJobs = 0;
-  db.chain
-    .get('jobs')
-    .filter((job) => job.userId === userId)
-    .forEach((job) => {
-      removedDemoJobs++;
-      listingStorage.removeListings(job.id);
-    });
-  db.chain
-    .get('jobs')
-    .remove((job) => job.userId === userId)
-    .value();
-  db.write();
-  if (removedDemoJobs > 0) {
-    logger.info(`Removed ${removedDemoJobs} demo jobs`);
+  // Count jobs to log similar to previous behavior
+  const count =
+    SqliteConnection.query(`SELECT COUNT(1) AS c FROM jobs WHERE user_id = @user_id`, { user_id: userId })[0]?.c ?? 0;
+  SqliteConnection.execute(`DELETE FROM jobs WHERE user_id = @user_id`, { user_id: userId });
+  if (count > 0) {
+    logger.info(`Removed ${count} jobs for user ${userId}`);
   }
 };
+
+/**
+ * Get all jobs.
+ * @returns {Job[]} List of jobs ordered by name (NULLs last).
+ */
 export const getJobs = () => {
-  return db.chain
-    .get('jobs')
-    .map((job) => ({
-      ...job,
-      numberOfFoundListings: listingStorage.getNumberOfAllKnownListings(job.id),
-    }))
-    .value();
+  const rows = SqliteConnection.query(
+    `SELECT j.id,
+            j.user_id AS userId,
+            j.enabled,
+            j.name,
+            j.blacklist,
+            j.provider,
+            j.notification_adapter AS notificationAdapter,
+            (SELECT COUNT(1) FROM listings l WHERE l.job_id = j.id) AS numberOfFoundListings
+       FROM jobs j
+       ORDER BY j.name IS NULL, j.name`,
+  );
+  return rows.map((row) => ({
+    ...row,
+    enabled: !!row.enabled,
+    blacklist: fromJson(row.blacklist, []),
+    provider: fromJson(row.provider, []),
+    notificationAdapter: fromJson(row.notificationAdapter, []),
+  }));
 };

lib/services/storage/listingsStorage.js

@@ -1,52 +1,138 @@
-import { JSONFileSync } from 'lowdb/node';
-import { getDirName } from '../../utils.js';
-import path from 'path';
-import LowdashAdapter from './LowDashAdapter.js';
+import { nullOrEmpty } from '../../utils.js';
+import SqliteConnection from './SqliteConnection.js';
+import { nanoid } from 'nanoid';
 
-const file = path.join(getDirName(), '../', 'db/jobListingData.json');
-const adapter = new JSONFileSync(file);
-const db = new LowdashAdapter(adapter, {});
-
-db.read();
-
-const buildKey = (jobKey, providerId, endpoint) => {
-  let key = `${jobKey}`;
-  if (jobKey == null && endpoint == null) {
-    return key;
-  }
-  if (providerId != null) {
-    key += `.${providerId}`;
-  }
-  if (endpoint != null) {
-    key += `.${endpoint}`;
-  }
-  return key;
-};
-export const getNumberOfAllKnownListings = (jobId) => {
-  const data = db.chain.get(`${jobId}.providerData`).value() || {};
-  return Object.values(data)
-    .map((values) => Object.keys(values).length)
-    .reduce((accumulator, currentValue) => accumulator + currentValue, 0);
-};
+/**
+ * Build analytics data for a given job by grouping all listings by provider and
+ * mapping each listing hash to its creation timestamp.
+ *
+ * SQL shape:
+ * SELECT json_group_object(provider, json_object(hash, created_at)) AS result
+ * FROM listings WHERE job_id = @jobId;
+ *
+ * The resulting object has the shape:
+ * {
+ *   providerA: { "<hash1>": <created_at_ms>, "<hash2>": <created_at_ms>, ... },
+ *   providerB: { ... }
+ * }
+ *
+ * @param {string} jobId - ID of the job whose listings should be aggregated.
+ * @returns {Record<string, Record<string, number>>} Object grouped by provider mapping listing-hash -> created_at epoch ms.
+ */
 export const getListingProviderDataForAnalytics = (jobId) => {
-  const key = buildKey(jobId, 'providerData');
-  return db.chain.get(key).value() || {};
+  const row = SqliteConnection.query(
+    `SELECT COALESCE(
+              json_group_object(provider, json(provider_map)),
+              json('{}')
+            ) AS result
+     FROM (SELECT provider,
+                  json_group_object(hash, created_at) AS provider_map
+           FROM listings
+           WHERE job_id = @jobId
+           GROUP BY provider);`,
+    { jobId },
+  );
+
+  return row?.length > 0 ? JSON.parse(row[0].result) : {};
 };
-export const getKnownListings = (jobId, providerId) => {
-  const providerListingsKey = buildKey(jobId, 'providerData', providerId, 'listings');
-  return db.chain.get(providerListingsKey).value() || {};
+
+/**
+ * Return a list of known listing hashes for a given job and provider.
+ * Useful to de-duplicate before inserting new listings.
+ *
+ * @param {string} jobId - The job identifier.
+ * @param {string} providerId - The provider identifier (e.g., 'immoscout').
+ * @returns {string[]} Array of listing hashes.
+ */
+export const getKnownListingHashesForJobAndProvider = (jobId, providerId) => {
+  return SqliteConnection.query(
+    `SELECT hash
+     FROM listings
+     WHERE job_id = @jobId AND provider = @providerId`,
+    { jobId, providerId },
+  ).map((r) => r.hash);
 };
-export const setKnownListings = (jobId, providerId, listings) => {
-  const providerListingsKey = buildKey(jobId, 'providerData', providerId, 'listings');
-  db.chain.set(providerListingsKey, listings).value();
-  return db.write();
-};
-export const setLastJobExecution = (jobId) => {
-  const key = buildKey(jobId, null, 'lastExecution');
-  db.chain.set(key, Date.now()).value();
-  return db.write();
-};
-export const removeListings = (jobId) => {
-  db.chain.unset(jobId).value();
-  db.write();
+
+/**
+ * Persist a batch of scraped listings for a given job and provider.
+ *
+ * - Empty or non-array inputs are ignored.
+ * - Each listing is inserted with ON CONFLICT(hash) DO NOTHING to avoid duplicates.
+ * - Performs inserts in a single transaction for performance.
+ *
+ * Listing input shape (minimal expected):
+ * {
+ *   id: string,            // unique id
+ *   hash: string           // stable hash/id of the listing (used as unique hash)
+ *   price?: string,        // e.g., "1.234 €" or "1,234€"
+ *   size?: string,         // e.g., "70 m²"
+ *   title?: string,
+ *   image?: string,        // image URL
+ *   description?: string,
+ *   address?: string,      // free-text address possibly containing parentheses
+ *   link?: string
+ * }
+ *
+ * @param {string} jobId - The job identifier.
+ * @param {string} providerId - The provider identifier.
+ * @param {Array<Object>} listings - Array of listing objects as described above.
+ * @returns {void}
+ */
+export const storeListings = (jobId, providerId, listings) => {
+  if (!Array.isArray(listings) || listings.length === 0) {
+    return;
+  }
+
+  SqliteConnection.withTransaction((db) => {
+    const stmt = db.prepare(
+      `INSERT INTO listings (id, hash, provider, job_id, price, size, title, image_url, description, address, city,
+                             link, created_at)
+       VALUES (@id, @hash, @provider, @job_id, @price, @size, @title, @image_url, @description, @address, @city, @link,
+               @created_at)
+       ON CONFLICT(hash) DO NOTHING`,
+    );
+
+    for (const item of listings) {
+      const params = {
+        id: nanoid(),
+        hash: item.id,
+        provider: providerId,
+        job_id: jobId,
+        price: extractNumber(item.price),
+        size: extractNumber(item.size),
+        title: item.title,
+        image_url: item.image,
+        description: item.description,
+        address: removeParentheses(item.address),
+        link: item.link,
+        created_at: Date.now(),
+      };
+      stmt.run(params);
+    }
+  });
+
+  /**
+   * Extract the first number from a string like "1.234 €" or "70 m²".
+   * Removes dots/commas before parsing. Returns null on invalid input.
+   * @param {string|undefined|null} str
+   * @returns {number|null}
+   */
+  function extractNumber(str) {
+    if (!str) return null;
+    const match = str.replace(/[.,]/g, '').match(/\d+/);
+    return match ? +match[0] : null;
+  }
+
+  /**
+   * Remove any parentheses segments (including surrounding whitespace) from a string.
+   * Returns null for empty input.
+   * @param {string|undefined|null} str
+   * @returns {string|null}
+   */
+  function removeParentheses(str) {
+    if (nullOrEmpty(str)) {
+      return null;
+    }
+    return str.replace(/\s*\([^)]*\)/g, '');
+  }
 };

lib/services/storage/userStorage.js

@@ -1,123 +1,176 @@
-import { JSONFileSync } from 'lowdb/node';
-import { config, getDirName } from '../../utils.js';
+import { config } from '../../utils.js';
 import * as hasher from '../security/hash.js';
 import { nanoid } from 'nanoid';
-import * as jobStorage from './jobStorage.js';
-import path from 'path';
-import LowdashAdapter from './LowDashAdapter.js';
-
-const defaultData = {
-  user: [
-    //you probably want to change the default password ;)
-    {
-      id: nanoid(),
-      lastLogin: Date.now(),
-      username: 'admin',
-      password: hasher.hash('admin'),
-      isAdmin: true,
-    },
-    {
-      id: nanoid(),
-      lastLogin: Date.now(),
-      username: 'demo',
-      password: hasher.hash('demo'),
-      isAdmin: true,
-    },
-  ],
-};
-
-const file = path.join(getDirName(), '../', 'db/users.json');
-const adapter = new JSONFileSync(file);
-const db = new LowdashAdapter(adapter, defaultData);
-
-db.read();
+import SqliteConnection from './SqliteConnection.js';
 
+/**
+ * Get all users.
+ *
+ * Notes:
+ * - Password hashes are omitted by default to avoid leaking them to callers that don’t need them.
+ * - numberOfJobs is computed via a subquery for each user.
+ *
+ * @param {boolean} withPassword - If true, include the hashed password in the returned objects; otherwise set password to null.
+ * @returns {User[]} Array of users ordered by username.
+ */
 export const getUsers = (withPassword) => {
-  const jobs = jobStorage.getJobs();
-  return db.chain
-    .get('user')
-    .value()
-    .map((user) => ({
-      //we dont want the password in the frontend, even tho it's hashed
-      ...user,
-      password: withPassword ? user.password : null,
-      numberOfJobs: jobs.filter((job) => job.userId === user.id).length,
-    }));
-};
-export const getUser = (id) => {
-  const jobs = jobStorage.getJobs();
-  const user = db.chain
-    .get('user')
-    .find((user) => user.id === id)
-    .value();
-  if (user == null) {
-    return null;
-  }
-  return {
-    ...user,
-    numberOfJobs: jobs.filter((job) => job.userId === user.id).length,
-  };
-};
-export const upsertUser = ({ username, password, userId, isAdmin }) => {
-  const user = db.chain
-    .get('user')
-    .filter((u) => u.id !== userId)
-    .value();
-  user.push({
-    id: userId || nanoid(),
-    username,
-    lastLogin: user.lastLogin,
-    password: hasher.hash(password),
-    isAdmin,
-  });
-  db.chain.set('user', user).value();
-  db.write();
-};
-export const setLastLoginToNow = ({ userId }) => {
-  db.chain
-    .get('user')
-    .find((u) => u.id === userId)
-    .assign({ lastLogin: Date.now() })
-    .value();
-  db.write();
-};
-export const removeUser = (userId) => {
-  const user = db.chain.get('user').value();
-  db.chain
-    .set(
-      'user',
-      user.filter((u) => u.id !== userId),
-    )
-    .value();
-  db.write();
+  const rows = SqliteConnection.query(
+    `SELECT u.id, u.username, u.password, u.last_login AS lastLogin, u.is_admin AS isAdmin,
+            (SELECT COUNT(1) FROM jobs j WHERE j.user_id = u.id) AS numberOfJobs
+       FROM users u
+       ORDER BY u.username`,
+  );
+  return rows.map((u) => ({
+    ...u,
+    password: withPassword ? u.password : null,
+    isAdmin: !!u.isAdmin,
+  }));
 };
 
-export const handleDemoUser = () => {
-  if (!config.demoMode) {
-    const user = db.chain.get('user').value();
-    db.chain
-      .set(
-        'user',
-        user.filter((u) => u.username !== 'demo'),
-      )
-      .value();
-    db.write();
+/**
+ * Get a single user by id.
+ *
+ * @param {string} id - User id (primary key).
+ * @returns {User|null} The user when found; otherwise null. The password field is included but callers should not expose it.
+ */
+export const getUser = (id) => {
+  const rows = SqliteConnection.query(
+    `SELECT u.id, u.username, u.password, u.last_login AS lastLogin, u.is_admin AS isAdmin,
+            (SELECT COUNT(1) FROM jobs j WHERE j.user_id = u.id) AS numberOfJobs
+       FROM users u
+      WHERE u.id = @id
+      LIMIT 1`,
+    { id },
+  );
+  const u = rows[0];
+  if (!u) return null;
+  return { ...u, isAdmin: !!u.isAdmin };
+};
+
+/**
+ * Insert a new user or update an existing one.
+ *
+ * Behavior:
+ * - When userId is provided and exists: updates username and isAdmin. Password is only updated when a non-empty password is provided.
+ * - When userId is missing or does not exist: inserts a new user with a freshly generated id. last_login is initialized to null.
+ * - Passwords are hashed using the same hashing function used for login comparison.
+ *
+ * @param {Object} params
+ * @param {string} params.username - Username (must be unique in DB).
+ * @param {string} [params.password] - Plain text password to set; if omitted on update, existing hash is preserved.
+ * @param {string} [params.userId] - Existing user id to update; if missing, a new id is generated.
+ * @param {boolean} params.isAdmin - Whether the user should have admin privileges.
+ * @returns {void}
+ */
+export const upsertUser = ({ username, password, userId, isAdmin }) => {
+  const id = userId || nanoid();
+  // Check if user exists
+  const exists = SqliteConnection.query(`SELECT 1 FROM users WHERE id = @id LIMIT 1`, { id }).length > 0;
+  if (exists) {
+    // Update existing user. Update password only if provided (non-empty string)
+    if (password && password.length > 0) {
+      SqliteConnection.execute(
+        `UPDATE users SET username = @username, password = @password, is_admin = @is_admin WHERE id = @id`,
+        { id, username, password: hasher.hash(password), is_admin: isAdmin ? 1 : 0 },
+      );
+    } else {
+      SqliteConnection.execute(`UPDATE users SET username = @username, is_admin = @is_admin WHERE id = @id`, {
+        id,
+        username,
+        is_admin: isAdmin ? 1 : 0,
+      });
+    }
   } else {
-    const demoUser = db.chain
-      .get('user')
-      .filter((u) => u.username === 'demo')
-      .value();
-    if (demoUser == null || demoUser.length === 0) {
-      db.chain
-        .get('user')
-        .value()
-        .push({
-          id: nanoid(),
-          username: 'demo',
-          password: hasher.hash('demo'),
-          isAdmin: true,
-        });
-      db.write();
+    SqliteConnection.execute(
+      `INSERT INTO users (id, username, password, last_login, is_admin)
+       VALUES (@id, @username, @password, @last_login, @is_admin)`,
+      {
+        id,
+        username,
+        password: hasher.hash(password || ''),
+        last_login: null,
+        is_admin: isAdmin ? 1 : 0,
+      },
+    );
+  }
+};
+
+/**
+ * Update the last_login timestamp to now for the given user.
+ *
+ * @param {{userId: string}} params - Parameters.
+ * @param {string} params.userId - The user's id.
+ * @returns {void}
+ */
+export const setLastLoginToNow = ({ userId }) => {
+  SqliteConnection.execute(`UPDATE users SET last_login = @now WHERE id = @id`, { id: userId, now: Date.now() });
+};
+
+/**
+ * Remove a user by id.
+ *
+ * Notes:
+ * - In the SQLite schema, jobs reference users with ON DELETE CASCADE, so jobs (and their listings via jobs) are removed automatically.
+ *
+ * @param {string} userId - The id of the user to remove.
+ * @returns {void}
+ */
+export const removeUser = (userId) => {
+  SqliteConnection.execute(`DELETE FROM users WHERE id = @id`, { id: userId });
+};
+
+/**
+ * Ensure the demo user matches the demo mode setting.
+ *
+ * Behavior:
+ * - When config.demoMode is false: remove the demo user (and its cascading data via FKs).
+ * - When config.demoMode is true: ensure a 'demo' user exists with password 'demo' and admin rights.
+ *
+ * Security: The demo user's password is set to a known value ('demo') and should only be enabled in demoMode.
+ * @returns {void}
+ */
+export const ensureDemoUserExists = () => {
+  if (!config.demoMode) {
+    // Remove demo user (and cascade delete their jobs/listings)
+    SqliteConnection.execute(`DELETE FROM users WHERE username = 'demo'`);
+    return;
+  }
+  // Ensure demo user exists when demo mode is on
+  const existing = SqliteConnection.query(`SELECT id FROM users WHERE username = 'demo' LIMIT 1`);
+  if (existing.length === 0) {
+    SqliteConnection.execute(
+      `INSERT INTO users (id, username, password, last_login, is_admin)
+       VALUES (@id, 'demo', @password, NULL, 1)`,
+      { id: nanoid(), password: hasher.hash('demo') },
+    );
+  }
+};
+
+/**
+ * Ensure there is at least one administrator in the system.
+ *
+ * Behavior:
+ * - If there are no users at all, create default 'admin' user with password 'admin'.
+ * - If users exist but none is admin, promote the first existing user to admin.
+ *
+ * Security: On a fresh instance, a default admin/admin is created; change this password immediately.
+ * @returns {void}
+ */
+export const ensureAdminUserExists = () => {
+  const anyUser = SqliteConnection.query(`SELECT id FROM users LIMIT 1`).length > 0;
+  if (!anyUser) {
+    SqliteConnection.execute(
+      `INSERT INTO users (id, username, password, last_login, is_admin)
+       VALUES (@id, 'admin', @password, @last_login, 1)`,
+      { id: nanoid(), password: hasher.hash('admin'), last_login: Date.now() },
+    );
+    return;
+  }
+  const adminCount = SqliteConnection.query(`SELECT COUNT(1) AS c FROM users WHERE is_admin = 1`)[0]?.c ?? 0;
+  if (adminCount === 0) {
+    const firstUser = SqliteConnection.query(`SELECT id FROM users LIMIT 1`)[0];
+    if (firstUser) {
+      SqliteConnection.execute(`UPDATE users SET is_admin = 1 WHERE id = @id`, { id: firstUser.id });
     }
   }
 };