clawdie-ai/setup/groups.ts

/**
 * Step: groups — Sync available groups/chats into the DB.
 */
import { execSync } from 'child_process';
import fs from 'fs';
import path from 'path';

import pg from 'pg';

import { OPS_DB_URL, STORE_DIR } from '../src/config.js';
import { logger } from '../src/logger.js';
import { emitStatus } from './status.js';

function parseArgs(args: string[]): { list: boolean; limit: number } {
  let list = false;
  let limit = 30;
  for (let i = 0; i < args.length; i++) {
    if (args[i] === '--list') list = true;
    if (args[i] === '--limit' && args[i + 1]) {
      limit = parseInt(args[i + 1], 10);
      i++;
    }
  }
  return { list, limit };
}

export async function run(args: string[]): Promise<void> {
  const projectRoot = process.cwd();
  const { list, limit } = parseArgs(args);

  if (list) {
    await listGroups(limit);
    return;
  }

  await syncGroups(projectRoot);
}

async function listGroups(limit: number): Promise<void> {
  try {
    const pool = new pg.Pool({ connectionString: OPS_DB_URL, max: 3 });
    const { rows } = await pool.query(
      `SELECT jid, name FROM chats
       WHERE jid <> '__group_sync__' AND name <> jid
       ORDER BY last_message_time DESC
       LIMIT $1`,
      [limit],
    );
    await pool.end();
    for (const row of rows as Array<{ jid: string; name: string }>) {
      console.log(`${row.jid}|${row.name}`);
    }
  } catch {
    console.error('ERROR: database not found or connection failed');
    process.exit(1);
  }
}

async function syncGroups(projectRoot: string): Promise<void> {
  // Build TypeScript first
  logger.info('Building TypeScript');
  let buildOk = false;
  try {
    execSync('npm run build', {
      cwd: projectRoot,
      stdio: ['ignore', 'pipe', 'pipe'],
    });
    buildOk = true;
    logger.info('Build succeeded');
  } catch {
    logger.error('Build failed');
    emitStatus('SYNC_GROUPS', {
      BUILD: 'failed',
      SYNC: 'skipped',
      GROUPS_IN_DB: 0,
      STATUS: 'failed',
      ERROR: 'build_failed',
      LOG: 'logs/setup.log',
    });
    process.exit(1);
  }

  // Run inline sync script via node
  logger.info('Fetching group metadata');
  let syncOk = false;
  try {
    const syncScript = `
import makeWASocket, { useMultiFileAuthState, makeCacheableSignalKeyStore, Browsers } from '@whiskeysockets/baileys';
import pg from 'pg';
import pino from 'pino';
import path from 'path';
import fs from 'fs';

const logger = pino({ level: 'silent' });
const authDir = path.join('store', 'auth');

const OPS_DB_URL = process.env.OPS_DB_URL;
if (!OPS_DB_URL) { console.error('NO_DB_URL'); process.exit(1); }

if (!fs.existsSync(authDir)) {
  console.error('NO_AUTH');
  process.exit(1);
}

const pool = new pg.Pool({ connectionString: OPS_DB_URL, max: 3 });

await pool.query('CREATE TABLE IF NOT EXISTS chats (jid TEXT PRIMARY KEY, name TEXT, last_message_time TEXT, channel TEXT, is_group INTEGER DEFAULT 0)');

const { state, saveCreds } = await useMultiFileAuthState(authDir);

const sock = makeWASocket({
  auth: { creds: state.creds, keys: makeCacheableSignalKeyStore(state.keys, logger) },
  printQRInTerminal: false,
  logger,
  browser: Browsers.macOS('Chrome'),
});

const timeout = setTimeout(() => {
  console.error('TIMEOUT');
  pool.end();
  process.exit(1);
}, 30000);

sock.ev.on('creds.update', saveCreds);

sock.ev.on('connection.update', async (update) => {
  if (update.connection === 'open') {
    try {
      const groups = await sock.groupFetchAllParticipating();
      const now = new Date().toISOString();
      let count = 0;
      for (const [jid, metadata] of Object.entries(groups)) {
        if (metadata.subject) {
          await pool.query(
            'INSERT INTO chats (jid, name, last_message_time) VALUES ($1, $2, $3) ON CONFLICT(jid) DO UPDATE SET name = EXCLUDED.name',
            [jid, metadata.subject, now]
          );
          count++;
        }
      }
      console.log('SYNCED:' + count);
    } catch (err) {
      console.error('FETCH_ERROR:' + err.message);
    } finally {
      clearTimeout(timeout);
      sock.end(undefined);
      await pool.end();
      process.exit(0);
    }
  } else if (update.connection === 'close') {
    clearTimeout(timeout);
    console.error('CONNECTION_CLOSED');
    await pool.end();
    process.exit(1);
  }
});
`;

    const output = execSync(
      `OPS_DB_URL=${process.env.OPS_DB_URL || ''} node --input-type=module -e ${JSON.stringify(syncScript)}`,
      {
        cwd: projectRoot,
        encoding: 'utf-8',
        timeout: 45000,
        stdio: ['ignore', 'pipe', 'pipe'],
      },
    );
    syncOk = output.includes('SYNCED:');
    logger.info({ output: output.trim() }, 'Sync output');
  } catch (err) {
    logger.error({ err }, 'Sync failed');
  }

  // Count groups in DB via Postgres
  let groupsInDb = 0;
  try {
    const pool = new pg.Pool({ connectionString: OPS_DB_URL, max: 3 });
    const { rows } = await pool.query(
      "SELECT COUNT(*) as count FROM chats WHERE jid <> '__group_sync__'",
    );
    groupsInDb = parseInt((rows[0] as { count: string }).count, 10);
    await pool.end();
  } catch {
    // DB may not exist yet
  }

  const status = syncOk ? 'success' : 'failed';

  emitStatus('SYNC_GROUPS', {
    BUILD: buildOk ? 'success' : 'failed',
    SYNC: syncOk ? 'success' : 'failed',
    GROUPS_IN_DB: groupsInDb,
    STATUS: status,
    LOG: 'logs/setup.log',
  });

  if (status === 'failed') process.exit(1);
}