/** * Data Migration: sub_agents JSONB → conversation_participants * ADR-091 Phase 2 / Ticket #41163 * * Context: * Existing conversations store agent references in the sub_agents JSONB column. * New code (ADR-091 dual-write) writes agents to BOTH sub_agents AND * conversation_participants. This one-time script back-fills conversation_participant * records for all historical conversations that only have sub_agents entries. * * Strategy: * 1. Find every conversation where sub_agents is non-empty. * 2. For each element in the sub_agents array (plain row_id or { row_id, response_mode }), * call resolveAgentUser() to find-or-create the agent user account. * 3. Insert a conversation_participant row (user_type = 'agent', role = 'member') * preserving the response_mode as agent_response_mode. * 4. Use ON CONFLICT DO NOTHING so the script is safe to re-run (idempotent). * 5. sub_agents JSONB is NOT modified (backward-compat preserved). * * Usage: * node backend/scripts/migrate-sub-agents-to-participants.js * # or with dotenv * DATABASE_TYPE=postgres PGPASSWORD=... node backend/scripts/migrate-sub-agents-to-participants.js */ import { dbAll, dbRun, isPostgres, safeJsonParse } from '../database/connection.js'; import { resolveAgentUser } from '../services/agent-users.js'; import { apiLogger } from '../utils/logger.js'; // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- /** * Parse a sub_agents value from a conversations row. * Handles: * - Already a JS array (when PG returns parsed JSONB) * - A JSON string (SQLite TEXT column) * - null / empty → returns [] * * Normalised entries always have shape: { row_id: number, response_mode?: string } * * @param {*} raw * @returns {{ row_id: number, response_mode?: string }[]} */ function parseSubAgents(raw) { if (!raw) return []; const arr = Array.isArray(raw) ? raw : safeJsonParse(raw, []); if (!Array.isArray(arr) || arr.length === 0) return []; return arr .map((entry) => { if (typeof entry === 'number' && Number.isFinite(entry) && entry > 0) { return { row_id: entry }; } if (typeof entry === 'object' && entry !== null && typeof entry.row_id === 'number') { return { row_id: entry.row_id, response_mode: entry.response_mode ?? null }; } return null; }) .filter(Boolean); } // --------------------------------------------------------------------------- // Main migration // --------------------------------------------------------------------------- async function migrate() { const log = apiLogger.child({ script: 'migrate-sub-agents-to-participants' }); log.info('Starting migration: sub_agents → conversation_participants'); // ------------------------------------------------------------------ // 1. Fetch all conversations that have non-empty sub_agents // ------------------------------------------------------------------ const pg = isPostgres(); const conversations = await dbAll( pg ? `SELECT id, sub_agents FROM conversations WHERE sub_agents IS NOT NULL AND sub_agents != '[]'::jsonb AND jsonb_array_length(sub_agents) > 0` : `SELECT id, sub_agents FROM conversations WHERE sub_agents IS NOT NULL AND sub_agents != '[]' AND sub_agents != ''` ); log.info({ count: conversations.length }, 'Conversations with non-empty sub_agents found'); if (conversations.length === 0) { log.info('Nothing to migrate — exiting'); return { migrated: 0, skipped: 0, errors: 0 }; } let migrated = 0; let skipped = 0; let errors = 0; // ------------------------------------------------------------------ // 2. For each conversation × sub_agent, find/create user + insert participant // ------------------------------------------------------------------ for (const conv of conversations) { const entries = parseSubAgents(conv.sub_agents); if (entries.length === 0) { log.debug({ conversationId: conv.id }, 'sub_agents array is empty after parsing — skipping'); skipped++; continue; } for (const entry of entries) { try { // 2a. Find or create agent user via the unified resolver const resolved = await resolveAgentUser(entry.row_id); if (!resolved) { log.warn( { conversationId: conv.id, agentRowId: entry.row_id }, 'resolveAgentUser returned null — agent row may be deleted or inactive, skipping' ); skipped++; continue; } const { userId } = resolved; const responseMode = entry.response_mode ?? null; // 2b. Upsert conversation_participant (ON CONFLICT DO NOTHING = idempotent) if (pg) { await dbRun( `INSERT INTO conversation_participants (conversation_id, user_id, role, user_type, agent_response_mode, joined_at) VALUES ($1, $2, 'member', 'agent', $3, NOW()) ON CONFLICT (conversation_id, user_id) DO NOTHING`, [conv.id, userId, responseMode] ); } else { // SQLite: INSERT OR IGNORE is the idempotent equivalent await dbRun( `INSERT OR IGNORE INTO conversation_participants (conversation_id, user_id, role, user_type, agent_response_mode, joined_at) VALUES (?, ?, 'member', 'agent', ?, datetime('now'))`, [conv.id, userId, responseMode] ); } log.info( { conversationId: conv.id, agentRowId: entry.row_id, userId, responseMode, }, 'Inserted/confirmed conversation_participant for agent' ); migrated++; } catch (err) { log.error( { conversationId: conv.id, agentRowId: entry.row_id, err }, 'Error processing sub_agent entry' ); errors++; } } } const summary = { migrated, skipped, errors }; log.info(summary, 'Migration complete'); return summary; } // --------------------------------------------------------------------------- // CLI entry point // --------------------------------------------------------------------------- migrate() .then(({ migrated, skipped, errors }) => { console.log(`\n✅ Migration complete — processed: ${migrated}, skipped: ${skipped}, errors: ${errors}`); if (errors > 0) { console.error(`⚠️ ${errors} error(s) occurred — check logs above`); process.exit(1); } process.exit(0); }) .catch((err) => { console.error('❌ Migration failed with unexpected error:', err); process.exit(1); });