openclaw-taskboard/lib/db.ts

import fs from "node:fs";
import path from "node:path";
import sqlite3 from "sqlite3";

const DB_PATH = process.env.DB_PATH || path.join(process.cwd(), "data", "tasks.db");

fs.mkdirSync(path.dirname(DB_PATH), { recursive: true });

let database: sqlite3.Database | null = null;
let databaseReady: Promise<void> | null = null;

function getColumnNames(db: sqlite3.Database, tableName: string) {
  return new Promise<string[]>((resolve, reject) => {
    db.all<{ name: string }>(`PRAGMA table_info(${tableName})`, [], (error, rows) => {
      if (error) {
        reject(error);
        return;
      }
      resolve(rows.map((row) => row.name));
    });
  });
}

async function ensureColumn(
  db: sqlite3.Database,
  tableName: string,
  columnName: string,
  definition: string,
) {
  const columnNames = await getColumnNames(db, tableName);
  if (columnNames.includes(columnName)) {
    return;
  }

  await new Promise<void>((resolve, reject) => {
    db.run(`ALTER TABLE ${tableName} ADD COLUMN ${columnName} ${definition}`, (error) => {
      if (error) {
        reject(error);
        return;
      }
      resolve();
    });
  });
}

function getDatabase() {
  if (database) {
    return database;
  }

  database = new sqlite3.Database(DB_PATH);
  database.serialize(() => {
    database?.run(`
      CREATE TABLE IF NOT EXISTS tasks (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        description TEXT DEFAULT '',
        assignee TEXT DEFAULT '',
        priority TEXT NOT NULL DEFAULT 'Medium',
        status TEXT NOT NULL DEFAULT 'Backlog',
        tags TEXT NOT NULL DEFAULT '[]',
        created_at TEXT NOT NULL DEFAULT (datetime('now')),
        updated_at TEXT NOT NULL DEFAULT (datetime('now')),
        completed_at TEXT
      )
    `);
    database?.run(`
      CREATE TABLE IF NOT EXISTS usage_tracking (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        agent TEXT NOT NULL,
        provider TEXT NOT NULL,
        model TEXT NOT NULL,
        request_type TEXT DEFAULT 'chat',
        tokens_used INTEGER DEFAULT 0,
        cost_estimate REAL DEFAULT 0,
        timestamp TEXT NOT NULL DEFAULT (datetime('now'))
      )
    `);
    database?.run(`
      CREATE TABLE IF NOT EXISTS task_events (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        task_id INTEGER NOT NULL,
        assignee TEXT NOT NULL DEFAULT '',
        family TEXT,
        host TEXT NOT NULL DEFAULT '',
        event_type TEXT NOT NULL,
        state TEXT,
        summary TEXT NOT NULL,
        detail TEXT NOT NULL DEFAULT '',
        created_at TEXT NOT NULL DEFAULT (datetime('now'))
      )
    `);
    database?.run(`
      CREATE INDEX IF NOT EXISTS idx_task_events_task_time
      ON task_events(task_id, created_at DESC)
    `);
    database?.run(`
      CREATE INDEX IF NOT EXISTS idx_task_events_assignee_time
      ON task_events(assignee, created_at DESC)
    `);
  });

  databaseReady = (async () => {
    if (!database) {
      return;
    }

    await ensureColumn(database, "tasks", "family", "TEXT");
    await ensureColumn(database, "tasks", "target_host", "TEXT NOT NULL DEFAULT ''");
    await ensureColumn(database, "tasks", "target_channel", "TEXT NOT NULL DEFAULT ''");
    await ensureColumn(database, "tasks", "dispatch_method", "TEXT NOT NULL DEFAULT 'manual'");
    await ensureColumn(database, "tasks", "dispatch_state", "TEXT NOT NULL DEFAULT 'planned'");
    await ensureColumn(database, "tasks", "template_key", "TEXT");
    await ensureColumn(database, "tasks", "repo_slug", "TEXT");
    await ensureColumn(database, "tasks", "base_branch", "TEXT");
    await ensureColumn(database, "tasks", "preferred_agent", "TEXT");
    await ensureColumn(database, "tasks", "reasoning_effort", "TEXT");
    await ensureColumn(database, "tasks", "model_hint", "TEXT");
    await ensureColumn(database, "tasks", "result_summary", "TEXT");
    await ensureColumn(database, "tasks", "result_detail", "TEXT");
    await ensureColumn(database, "tasks", "completed_by", "TEXT");
    await ensureColumn(database, "tasks", "last_dispatch_at", "TEXT");
    await ensureColumn(database, "tasks", "acknowledged_at", "TEXT");
    await ensureColumn(database, "tasks", "last_error", "TEXT");
  })();

  return database;
}

async function ensureReady() {
  getDatabase();
  if (databaseReady) {
    await databaseReady;
  }
}

export async function all<T>(sql: string, params: unknown[] = []) {
  await ensureReady();
  const db = getDatabase();
  return new Promise<T[]>((resolve, reject) => {
    db.all(sql, params, (error, rows) => {
      if (error) {
        reject(error);
        return;
      }
      resolve(rows as T[]);
    });
  });
}

export async function get<T>(sql: string, params: unknown[] = []) {
  await ensureReady();
  const db = getDatabase();
  return new Promise<T | undefined>((resolve, reject) => {
    db.get(sql, params, (error, row) => {
      if (error) {
        reject(error);
        return;
      }
      resolve(row as T | undefined);
    });
  });
}

export async function run(sql: string, params: unknown[] = []) {
  await ensureReady();
  const db = getDatabase();
  return new Promise<{ lastID: number; changes: number }>((resolve, reject) => {
    db.run(sql, params, function onRun(error) {
      if (error) {
        reject(error);
        return;
      }
      resolve({ lastID: this.lastID, changes: this.changes });
    });
  });
}