|
|
import type { Database } from 'bun:sqlite';
|
|
|
import { db, ensureUserExists } from '../db';
|
|
|
import { normalizeWhatsAppId } from '../utils/whatsapp';
|
|
|
import { Metrics } from './metrics';
|
|
|
import { IdentityService } from './identity';
|
|
|
import { AllowedGroups } from './allowed-groups';
|
|
|
|
|
|
// In-memory cache for active groups
|
|
|
// const activeGroupsCache = new Map<string, string>(); // groupId -> groupName
|
|
|
|
|
|
/**
|
|
|
* Represents a group from the Evolution API response
|
|
|
*
|
|
|
* API returns an array of groups in this format:
|
|
|
* [
|
|
|
* {
|
|
|
* id: string, // Group ID in @g.us format (primary key)
|
|
|
* subject: string, // Group name (displayed to users)
|
|
|
* linkedParent?: string, // Parent community ID if group belongs to one
|
|
|
* size?: number, // Current member count (unused in our system)
|
|
|
* creation?: number, // Unix timestamp of group creation (unused)
|
|
|
* desc?: string, // Group description text (unused)
|
|
|
* // ...other fields exist but are ignored by our implementation
|
|
|
* }
|
|
|
* ]
|
|
|
*
|
|
|
* Required fields for our implementation:
|
|
|
* - id (used as database primary key)
|
|
|
* - subject (used as group display name)
|
|
|
* - linkedParent (used for community filtering)
|
|
|
*/
|
|
|
type EvolutionGroup = {
|
|
|
id: string;
|
|
|
subject: string;
|
|
|
linkedParent?: string;
|
|
|
};
|
|
|
|
|
|
export class GroupSyncService {
|
|
|
// Static property for DB instance injection (defaults to global db)
|
|
|
static dbInstance: Database = db;
|
|
|
|
|
|
// In-memory cache for active groups (made public for tests)
|
|
|
public static readonly activeGroupsCache = new Map<string, string>(); // groupId -> groupName
|
|
|
|
|
|
/**
|
|
|
* Gets the sync interval duration in milliseconds.
|
|
|
*
|
|
|
* Priority:
|
|
|
* 1. GROUP_SYNC_INTERVAL_MS environment variable if set
|
|
|
* 2. Default 24 hour interval
|
|
|
*
|
|
|
* In development mode, enforces minimum 10 second interval
|
|
|
* to prevent accidental excessive API calls.
|
|
|
*
|
|
|
* @returns {number} Sync interval in milliseconds
|
|
|
*/
|
|
|
private static get SYNC_INTERVAL_MS(): number {
|
|
|
const interval = process.env.GROUP_SYNC_INTERVAL_MS
|
|
|
? Number(process.env.GROUP_SYNC_INTERVAL_MS)
|
|
|
: 24 * 60 * 60 * 1000; // Default 24 hours
|
|
|
|
|
|
// Ensure minimum 10 second interval in development
|
|
|
if (process.env.NODE_ENV === 'development' && interval < 10000) {
|
|
|
console.warn(`Sync interval too low (${interval}ms), using 10s minimum`);
|
|
|
return 10000;
|
|
|
}
|
|
|
return interval;
|
|
|
}
|
|
|
private static lastSyncAttempt = 0;
|
|
|
private static _groupsTimer: any = null;
|
|
|
private static _groupsSchedulerRunning = false;
|
|
|
private static _membersTimer: any = null;
|
|
|
private static _membersSchedulerRunning = false;
|
|
|
private static _groupsIntervalMs: number | null = null;
|
|
|
private static _groupsNextTickAt: number | null = null;
|
|
|
|
|
|
static async syncGroups(force: boolean = false): Promise<{ added: number; updated: number }> {
|
|
|
if (!this.shouldSync(force)) {
|
|
|
return { added: 0, updated: 0 };
|
|
|
}
|
|
|
const startedAt = Date.now();
|
|
|
Metrics.inc('sync_runs_total');
|
|
|
|
|
|
try {
|
|
|
const groups = await this.fetchGroupsFromAPI();
|
|
|
console.log('ℹ️ Grupos crudos de la API:', JSON.stringify(groups, null, 2));
|
|
|
console.log('ℹ️ Sin filtrar por comunidad (modo multicomunidad). Total grupos:', groups.length);
|
|
|
|
|
|
const dbGroupsBefore = this.dbInstance.prepare('SELECT id, active FROM groups').all();
|
|
|
console.log('ℹ️ Grupos en DB antes de upsert:', dbGroupsBefore);
|
|
|
|
|
|
const result = await this.upsertGroups(groups);
|
|
|
|
|
|
const dbGroupsAfter = this.dbInstance.prepare('SELECT id, active FROM groups').all();
|
|
|
console.log('ℹ️ Grupos en DB después de upsert:', dbGroupsAfter);
|
|
|
|
|
|
// Completar labels faltantes en allowed_groups usando todos los grupos devueltos por la API
|
|
|
try { (AllowedGroups as any).dbInstance = this.dbInstance; this.fillMissingAllowedGroupLabels(groups); } catch {}
|
|
|
|
|
|
// Actualizar métricas
|
|
|
this.cacheActiveGroups();
|
|
|
Metrics.set('active_groups', this.activeGroupsCache.size);
|
|
|
const rowM = this.dbInstance.prepare(`SELECT COUNT(*) AS c FROM group_members WHERE is_active = 1`).get() as any;
|
|
|
Metrics.set('active_members', Number(rowM?.c || 0));
|
|
|
Metrics.set('last_sync_timestamp_seconds', Math.floor(Date.now() / 1000));
|
|
|
Metrics.set('last_sync_ok', 1);
|
|
|
// Duración opcional
|
|
|
Metrics.set('last_sync_duration_ms', Date.now() - (typeof startedAt !== 'undefined' ? startedAt : Date.now()));
|
|
|
|
|
|
return result;
|
|
|
} catch (error) {
|
|
|
console.error('Group sync failed:', error);
|
|
|
Metrics.inc('sync_errors_total');
|
|
|
Metrics.set('last_sync_ok', 0);
|
|
|
throw error;
|
|
|
} finally {
|
|
|
this.lastSyncAttempt = Date.now();
|
|
|
}
|
|
|
}
|
|
|
|
|
|
private static shouldSync(force: boolean = false): boolean {
|
|
|
if (force) return true;
|
|
|
const timeSinceLastSync = Date.now() - this.lastSyncAttempt;
|
|
|
const shouldSync = timeSinceLastSync > this.SYNC_INTERVAL_MS;
|
|
|
|
|
|
if (!shouldSync) {
|
|
|
const nextSyncIn = this.SYNC_INTERVAL_MS - timeSinceLastSync;
|
|
|
console.debug(`Next sync available in ${Math.round(nextSyncIn / 1000)} seconds`);
|
|
|
}
|
|
|
|
|
|
return shouldSync;
|
|
|
}
|
|
|
|
|
|
private static async fetchGroupsFromAPI(): Promise<EvolutionGroup[]> {
|
|
|
const url = `${process.env.EVOLUTION_API_URL}/group/fetchAllGroups/${process.env.EVOLUTION_API_INSTANCE}?getParticipants=false`;
|
|
|
console.log('ℹ️ Fetching groups from API:', {
|
|
|
url: `${url}...`, // Log partial URL for security
|
|
|
communityId: process.env.WHATSAPP_COMMUNITY_ID,
|
|
|
time: new Date().toISOString()
|
|
|
});
|
|
|
|
|
|
try {
|
|
|
const response = await fetch(url, {
|
|
|
method: 'GET',
|
|
|
headers: {
|
|
|
apikey: process.env.EVOLUTION_API_KEY,
|
|
|
},
|
|
|
httpVersion: '2',
|
|
|
timeout: 320000 // 120 second timeout
|
|
|
});
|
|
|
|
|
|
if (!response.ok) {
|
|
|
const errorBody = await response.text().catch(() => 'Unable to read error body');
|
|
|
console.error('❌ API request failed:', {
|
|
|
status: response.status,
|
|
|
statusText: response.statusText,
|
|
|
headers: Object.fromEntries(response.headers.entries()),
|
|
|
body: errorBody
|
|
|
});
|
|
|
throw new Error(`API request failed: ${response.status} ${response.statusText}`);
|
|
|
}
|
|
|
|
|
|
const rawResponse = await response.text();
|
|
|
console.log('ℹ️ Raw API response length:', rawResponse.length);
|
|
|
|
|
|
// Parse response which could be either:
|
|
|
// 1. Direct array of groups: [{group1}, {group2}]
|
|
|
// 2. Or wrapped response: {status, message, response}
|
|
|
let groups;
|
|
|
try {
|
|
|
const parsed = JSON.parse(rawResponse);
|
|
|
if (Array.isArray(parsed)) {
|
|
|
// Case 1: Direct array response
|
|
|
groups = parsed;
|
|
|
console.log('ℹ️ Received direct array of', groups.length, 'groups');
|
|
|
} else if (parsed.response && Array.isArray(parsed.response)) {
|
|
|
// Case 2: Wrapped response
|
|
|
if (parsed.status !== 'success') {
|
|
|
throw new Error(`API error: ${parsed.message || 'Unknown error'}`);
|
|
|
}
|
|
|
groups = parsed.response;
|
|
|
console.log('ℹ️ Received wrapped response with', groups.length, 'groups');
|
|
|
} else {
|
|
|
throw new Error('Invalid API response format - expected array or wrapped response');
|
|
|
}
|
|
|
} catch (e) {
|
|
|
console.error('❌ Failed to parse API response:', {
|
|
|
error: e instanceof Error ? e.message : String(e),
|
|
|
responseSample: rawResponse.substring(0, 100) + '...'
|
|
|
});
|
|
|
throw e;
|
|
|
}
|
|
|
|
|
|
if (!groups.length) {
|
|
|
console.warn('⚠️ API returned empty group list');
|
|
|
}
|
|
|
return groups;
|
|
|
} catch (error) {
|
|
|
console.error('❌ Failed to fetch groups:', {
|
|
|
error: error instanceof Error ? error.message : String(error),
|
|
|
stack: error instanceof Error ? error.stack : undefined
|
|
|
});
|
|
|
throw error;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
private static cacheActiveGroups(): void {
|
|
|
const groups = this.dbInstance.prepare('SELECT id, name FROM groups WHERE active = TRUE').all();
|
|
|
this.activeGroupsCache.clear();
|
|
|
for (const group of groups) {
|
|
|
this.activeGroupsCache.set(group.id, group.name);
|
|
|
}
|
|
|
console.log(`Cached ${this.activeGroupsCache.size} active groups`);
|
|
|
}
|
|
|
|
|
|
// Rellena labels faltantes en allowed_groups a partir de los grupos devueltos por la API.
|
|
|
private static fillMissingAllowedGroupLabels(allGroups: EvolutionGroup[]): number {
|
|
|
try {
|
|
|
if (!Array.isArray(allGroups) || allGroups.length === 0) return 0;
|
|
|
const nameById = new Map<string, string>();
|
|
|
for (const g of allGroups) {
|
|
|
if (!g?.id) continue;
|
|
|
const name = String(g.subject || '').trim();
|
|
|
if (!name) continue;
|
|
|
nameById.set(String(g.id), name);
|
|
|
}
|
|
|
if (nameById.size === 0) return 0;
|
|
|
|
|
|
const rows = this.dbInstance.prepare(`
|
|
|
SELECT group_id AS id
|
|
|
FROM allowed_groups
|
|
|
WHERE label IS NULL OR TRIM(label) = ''
|
|
|
`).all() as any[];
|
|
|
if (!rows || rows.length === 0) return 0;
|
|
|
|
|
|
let filled = 0;
|
|
|
for (const r of rows) {
|
|
|
const id = r?.id ? String(r.id) : null;
|
|
|
if (!id) continue;
|
|
|
const label = nameById.get(id);
|
|
|
if (label) {
|
|
|
try { (AllowedGroups as any).dbInstance = this.dbInstance; AllowedGroups.upsertPending(id, label, null); } catch {}
|
|
|
filled++;
|
|
|
}
|
|
|
}
|
|
|
if (filled > 0) {
|
|
|
try { Metrics.inc('allowed_groups_labels_filled_total', filled); } catch {}
|
|
|
console.log(`ℹ️ Rellenadas ${filled} labels faltantes en allowed_groups`);
|
|
|
}
|
|
|
return filled;
|
|
|
} catch (e) {
|
|
|
console.warn('⚠️ No se pudieron rellenar labels faltantes en allowed_groups:', e);
|
|
|
return 0;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
private static getActiveGroupsCount(): number {
|
|
|
const result = this.dbInstance.prepare('SELECT COUNT(*) as count FROM groups WHERE active = TRUE').get();
|
|
|
return result?.count || 0;
|
|
|
}
|
|
|
|
|
|
static async checkInitialGroups(): Promise<void> {
|
|
|
const count = this.getActiveGroupsCount();
|
|
|
if (count > 0) {
|
|
|
this.cacheActiveGroups();
|
|
|
console.log(`✅ Using ${count} existing groups from database`);
|
|
|
return;
|
|
|
}
|
|
|
|
|
|
const communityId = process.env.WHATSAPP_COMMUNITY_ID;
|
|
|
if (!communityId) {
|
|
|
console.log('ℹ️ WHATSAPP_COMMUNITY_ID no definido - mostrando comunidades disponibles');
|
|
|
try {
|
|
|
const allGroups = await this.fetchGroupsFromAPI();
|
|
|
const communities = allGroups.filter(g => g.linkedParent);
|
|
|
|
|
|
if (communities.length === 0) {
|
|
|
console.log('❌ No se encontraron comunidades (grupos con linkedParent)');
|
|
|
} else {
|
|
|
console.log('\n📋 Comunidades disponibles (copia el ID completo):');
|
|
|
console.log('='.repeat(80));
|
|
|
console.log('Nombre'.padEnd(30), 'ID Comunidad');
|
|
|
console.log('-'.repeat(30), '-'.repeat(48));
|
|
|
communities.forEach(c => {
|
|
|
console.log(c.subject.padEnd(30), c.id);
|
|
|
});
|
|
|
console.log('='.repeat(80));
|
|
|
console.log('⚠️ ATENCIÓN: Estos IDs son sensibles. No los compartas públicamente.');
|
|
|
console.log(`\n⏳ El proceso terminará automáticamente en 120 segundos...`);
|
|
|
|
|
|
// Cuenta regresiva de 120 segundos
|
|
|
await new Promise(resolve => {
|
|
|
setTimeout(resolve, 120000);
|
|
|
const interval = setInterval(() => {
|
|
|
const remaining = Math.ceil((120000 - (Date.now() - startTime)) / 1000);
|
|
|
process.stdout.write(`\r⏳ Tiempo restante: ${remaining}s `);
|
|
|
}, 1000);
|
|
|
const startTime = Date.now();
|
|
|
});
|
|
|
|
|
|
console.log('\n\n✅ Listado completado. Por favor configura WHATSAPP_COMMUNITY_ID');
|
|
|
}
|
|
|
process.exit(0);
|
|
|
} catch (error) {
|
|
|
console.error('❌ Error al obtener comunidades:', error instanceof Error ? error.message : error);
|
|
|
process.exit(1);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
console.log('⚠️ No groups found in database - performing initial sync');
|
|
|
try {
|
|
|
const { added } = await this.syncGroups();
|
|
|
if (added === 0) {
|
|
|
throw new Error('Initial group sync completed but no groups were added');
|
|
|
}
|
|
|
this.cacheActiveGroups();
|
|
|
console.log(`✅ Initial group sync completed - added ${added} groups`);
|
|
|
} catch (error) {
|
|
|
console.error('❌ Critical: Initial group sync failed - no groups available');
|
|
|
console.error(error instanceof Error ? error.message : 'Unknown error');
|
|
|
process.exit(1);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
private static async upsertGroups(groups: EvolutionGroup[]): Promise<{ added: number; updated: number }> {
|
|
|
let added = 0;
|
|
|
let updated = 0;
|
|
|
|
|
|
const transactionResult = this.dbInstance.transaction(() => {
|
|
|
// First mark all groups as inactive and update verification timestamp
|
|
|
const inactiveResult = this.dbInstance.prepare(`
|
|
|
UPDATE groups
|
|
|
SET active = FALSE,
|
|
|
last_verified = CURRENT_TIMESTAMP
|
|
|
WHERE active = TRUE
|
|
|
`).run();
|
|
|
console.log('ℹ️ Grupos marcados como inactivos:', {
|
|
|
count: inactiveResult.changes,
|
|
|
lastId: inactiveResult.lastInsertRowid
|
|
|
});
|
|
|
|
|
|
for (const group of groups) {
|
|
|
const existing = this.dbInstance.prepare('SELECT 1 FROM groups WHERE id = ?').get(group.id);
|
|
|
console.log('Checking group:', group.id, 'exists:', !!existing);
|
|
|
|
|
|
if (existing) {
|
|
|
const updateResult = this.dbInstance.prepare(
|
|
|
'UPDATE groups SET name = ?, community_id = COALESCE(?, community_id), active = TRUE, last_verified = CURRENT_TIMESTAMP WHERE id = ?'
|
|
|
).run(group.subject, group.linkedParent || null, group.id);
|
|
|
console.log('Updated group:', group.id, 'result:', updateResult);
|
|
|
updated++;
|
|
|
} else {
|
|
|
const insertResult = this.dbInstance.prepare(
|
|
|
'INSERT INTO groups (id, community_id, name, active) VALUES (?, ?, ?, TRUE)'
|
|
|
).run(group.id, (group.linkedParent ?? ''), group.subject);
|
|
|
console.log('Added group:', group.id, 'result:', insertResult);
|
|
|
added++;
|
|
|
}
|
|
|
// Propagar subject como label a allowed_groups (no degrada estado; actualiza label si cambia)
|
|
|
try { (AllowedGroups as any).dbInstance = this.dbInstance; AllowedGroups.upsertPending(group.id, group.subject, null); } catch {}
|
|
|
}
|
|
|
|
|
|
return { added, updated };
|
|
|
});
|
|
|
|
|
|
try {
|
|
|
const result = transactionResult();
|
|
|
console.log(`Group sync completed: ${result.added} added, ${result.updated} updated`);
|
|
|
return result;
|
|
|
} catch (error) {
|
|
|
console.error('Error in upsertGroups:', error);
|
|
|
throw error;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Checks if a given group ID is active based on the in-memory cache.
|
|
|
*
|
|
|
* @param groupId The group ID to check (e.g., '123456789@g.us').
|
|
|
* @returns True if the group is active, false otherwise.
|
|
|
*/
|
|
|
static isGroupActive(groupId: string): boolean {
|
|
|
return this.activeGroupsCache.has(groupId);
|
|
|
}
|
|
|
|
|
|
// Fetch members for a single group from Evolution API. Uses a robust parser to accept multiple payload shapes.
|
|
|
private static async fetchGroupMembersFromAPI(groupId: string): Promise<Array<{ userId: string; isAdmin: boolean }>> {
|
|
|
// En tests se recomienda simular fetch; no retornamos temprano para permitir validar el parser
|
|
|
|
|
|
// 1) Intento preferente: endpoint de Evolution "Find Group Members"
|
|
|
// Documentación provista: GET /group/participants/{instance}
|
|
|
// Suponemos soporte de query param groupJid
|
|
|
try {
|
|
|
const url1 = `${process.env.EVOLUTION_API_URL}/group/participants/${process.env.EVOLUTION_API_INSTANCE}?groupJid=${encodeURIComponent(groupId)}`;
|
|
|
console.log('ℹ️ Fetching members via /group/participants:', { groupId });
|
|
|
|
|
|
const r1 = await fetch(url1, {
|
|
|
method: 'GET',
|
|
|
headers: { apikey: process.env.EVOLUTION_API_KEY },
|
|
|
httpVersion: '2',
|
|
|
timeout: 320000
|
|
|
});
|
|
|
|
|
|
if (r1.ok) {
|
|
|
const raw1 = await r1.text();
|
|
|
let parsed1: any;
|
|
|
try {
|
|
|
parsed1 = JSON.parse(raw1);
|
|
|
} catch (e) {
|
|
|
console.error('❌ Failed to parse /group/participants JSON:', String(e));
|
|
|
throw e;
|
|
|
}
|
|
|
|
|
|
const participantsArr = Array.isArray(parsed1?.participants) ? parsed1.participants : null;
|
|
|
if (participantsArr) {
|
|
|
const result: Array<{ userId: string; isAdmin: boolean }> = [];
|
|
|
for (const p of participantsArr) {
|
|
|
let jid: string | null = null;
|
|
|
let isAdmin = false;
|
|
|
|
|
|
if (typeof p === 'string') {
|
|
|
jid = p;
|
|
|
} else if (p && typeof p === 'object') {
|
|
|
const rawId = p.id || p?.user?.id || p.user || null;
|
|
|
const rawJid = p.jid || null; // preferir .jid cuando exista
|
|
|
jid = rawJid || rawId || null;
|
|
|
|
|
|
// Aprender mapping alias→número si vienen ambos
|
|
|
if (rawId && rawJid) {
|
|
|
try { IdentityService.upsertAlias(String(rawId), String(rawJid), 'group.participants'); } catch {}
|
|
|
}
|
|
|
|
|
|
if (typeof p.isAdmin === 'boolean') {
|
|
|
isAdmin = p.isAdmin;
|
|
|
} else if (typeof p.admin === 'string') {
|
|
|
isAdmin = p.admin === 'admin' || p.admin === 'superadmin';
|
|
|
} else if (typeof p.role === 'string') {
|
|
|
isAdmin = p.role.toLowerCase().includes('admin');
|
|
|
}
|
|
|
}
|
|
|
|
|
|
let norm = normalizeWhatsAppId(jid);
|
|
|
if (!norm) {
|
|
|
const digits = (jid || '').replace(/\D+/g, '');
|
|
|
norm = digits || null;
|
|
|
}
|
|
|
if (!norm) continue;
|
|
|
result.push({ userId: norm, isAdmin });
|
|
|
}
|
|
|
let resolved: Array<{ userId: string; isAdmin: boolean }>;
|
|
|
try {
|
|
|
const map = IdentityService.resolveMany(result.map(r => r.userId));
|
|
|
resolved = result.map(r => ({ userId: map.get(r.userId) || r.userId, isAdmin: r.isAdmin }));
|
|
|
} catch {
|
|
|
resolved = result;
|
|
|
}
|
|
|
return resolved;
|
|
|
}
|
|
|
// Si no viene en el formato esperado, caemos al plan B
|
|
|
console.warn('⚠️ /group/participants responded without participants array, falling back to fetchAllGroups');
|
|
|
} else {
|
|
|
const body = await r1.text().catch(() => '');
|
|
|
console.warn(`⚠️ /group/participants failed: ${r1.status} ${r1.statusText} - ${body.slice(0, 200)}. Falling back to fetchAllGroups`);
|
|
|
}
|
|
|
} catch (e) {
|
|
|
console.warn('⚠️ Error calling /group/participants, falling back to fetchAllGroups:', e instanceof Error ? e.message : String(e));
|
|
|
}
|
|
|
|
|
|
// 2) Fallback robusto: fetchAllGroups(getParticipants=true) y filtrar por groupId
|
|
|
const url = `${process.env.EVOLUTION_API_URL}/group/fetchAllGroups/${process.env.EVOLUTION_API_INSTANCE}?getParticipants=true`;
|
|
|
console.log('ℹ️ Fetching members via fetchAllGroups (participants=true):', { groupId });
|
|
|
|
|
|
const response = await fetch(url, {
|
|
|
method: 'GET',
|
|
|
headers: { apikey: process.env.EVOLUTION_API_KEY },
|
|
|
httpVersion: '2',
|
|
|
timeout: 320000
|
|
|
});
|
|
|
if (!response.ok) {
|
|
|
const body = await response.text().catch(() => '');
|
|
|
throw new Error(`Failed to fetch groups with participants: ${response.status} ${response.statusText} - ${body.slice(0,200)}`);
|
|
|
}
|
|
|
const raw = await response.text();
|
|
|
let parsed: any;
|
|
|
try {
|
|
|
parsed = JSON.parse(raw);
|
|
|
} catch (e) {
|
|
|
console.error('❌ Failed to parse members response JSON:', String(e));
|
|
|
throw e;
|
|
|
}
|
|
|
|
|
|
let groups: any[] = [];
|
|
|
if (Array.isArray(parsed)) {
|
|
|
groups = parsed;
|
|
|
} else if (parsed && Array.isArray(parsed.response)) {
|
|
|
groups = parsed.response;
|
|
|
} else {
|
|
|
throw new Error('Invalid response format for groups with participants');
|
|
|
}
|
|
|
|
|
|
const g = groups.find((g: any) => g?.id === groupId);
|
|
|
if (!g) {
|
|
|
console.warn(`⚠️ Group ${groupId} not present in fetchAllGroups(getParticipants=true) response`);
|
|
|
return [];
|
|
|
}
|
|
|
const participants = Array.isArray(g.participants) ? g.participants : [];
|
|
|
|
|
|
const result: Array<{ userId: string; isAdmin: boolean }> = [];
|
|
|
for (const p of participants) {
|
|
|
let jid: string | null = null;
|
|
|
let isAdmin = false;
|
|
|
|
|
|
if (typeof p === 'string') {
|
|
|
jid = p;
|
|
|
} else if (p && typeof p === 'object') {
|
|
|
const rawId = p.id || p?.user?.id || p.user || null;
|
|
|
const rawJid = p.jid || null; // preferir .jid cuando exista
|
|
|
jid = rawJid || rawId || null;
|
|
|
|
|
|
// Aprender mapping alias→número si vienen ambos
|
|
|
if (rawId && rawJid) {
|
|
|
try { IdentityService.upsertAlias(String(rawId), String(rawJid), 'group.participants'); } catch {}
|
|
|
}
|
|
|
|
|
|
if (typeof p.isAdmin === 'boolean') {
|
|
|
isAdmin = p.isAdmin;
|
|
|
} else if (typeof p.admin === 'string') {
|
|
|
// common shapes: 'admin', 'superadmin' or null
|
|
|
isAdmin = p.admin === 'admin' || p.admin === 'superadmin';
|
|
|
} else if (typeof p.role === 'string') {
|
|
|
isAdmin = p.role.toLowerCase().includes('admin');
|
|
|
}
|
|
|
}
|
|
|
|
|
|
let norm = normalizeWhatsAppId(jid);
|
|
|
if (!norm) {
|
|
|
const digits = (jid || '').replace(/\D+/g, '');
|
|
|
norm = digits || null;
|
|
|
}
|
|
|
if (!norm) continue;
|
|
|
result.push({ userId: norm, isAdmin });
|
|
|
}
|
|
|
let resolved: Array<{ userId: string; isAdmin: boolean }>;
|
|
|
try {
|
|
|
const map = IdentityService.resolveMany(result.map(r => r.userId));
|
|
|
resolved = result.map(r => ({ userId: map.get(r.userId) || r.userId, isAdmin: r.isAdmin }));
|
|
|
} catch {
|
|
|
resolved = result;
|
|
|
}
|
|
|
return resolved;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Reconciles current DB membership state for a group with a fresh snapshot.
|
|
|
* Idempotente y atómico por grupo.
|
|
|
*/
|
|
|
static reconcileGroupMembers(groupId: string, snapshot: Array<{ userId: string; isAdmin: boolean }>, nowIso?: string): { added: number; updated: number; deactivated: number } {
|
|
|
if (!groupId || !Array.isArray(snapshot)) {
|
|
|
throw new Error('Invalid arguments for reconcileGroupMembers');
|
|
|
}
|
|
|
const now = nowIso || new Date().toISOString().replace('T', ' ').replace('Z', '');
|
|
|
let added = 0, updated = 0, deactivated = 0;
|
|
|
|
|
|
// Build quick lookup from snapshot
|
|
|
const incoming = new Map<string, { isAdmin: boolean }>();
|
|
|
for (const m of snapshot) {
|
|
|
if (!m?.userId) continue;
|
|
|
incoming.set(m.userId, { isAdmin: !!m.isAdmin });
|
|
|
}
|
|
|
|
|
|
this.dbInstance.transaction(() => {
|
|
|
// Load existing membership for group
|
|
|
const existingRows = this.dbInstance.prepare(`
|
|
|
SELECT user_id, is_admin, is_active
|
|
|
FROM group_members
|
|
|
WHERE group_id = ?
|
|
|
`).all(groupId) as Array<{ user_id: string; is_admin: number; is_active: number }>;
|
|
|
|
|
|
const existing = new Map(existingRows.map(r => [r.user_id, { isAdmin: !!r.is_admin, isActive: !!r.is_active }]));
|
|
|
|
|
|
// Upsert present members
|
|
|
for (const [userId, { isAdmin }] of incoming.entries()) {
|
|
|
// Ensure user exists (FK)
|
|
|
ensureUserExists(userId, this.dbInstance);
|
|
|
|
|
|
const row = existing.get(userId);
|
|
|
if (!row) {
|
|
|
// insert
|
|
|
this.dbInstance.prepare(`
|
|
|
INSERT INTO group_members (group_id, user_id, is_admin, is_active, first_seen_at, last_seen_at)
|
|
|
VALUES (?, ?, ?, 1, ?, ?)
|
|
|
`).run(groupId, userId, isAdmin ? 1 : 0, now, now);
|
|
|
added++;
|
|
|
} else {
|
|
|
// update if needed
|
|
|
let roleChanged = row.isAdmin !== isAdmin;
|
|
|
if (!row.isActive || roleChanged) {
|
|
|
this.dbInstance.prepare(`
|
|
|
UPDATE group_members
|
|
|
SET is_active = 1,
|
|
|
is_admin = ?,
|
|
|
last_seen_at = ?,
|
|
|
last_role_change_at = CASE WHEN ? THEN ? ELSE last_role_change_at END
|
|
|
WHERE group_id = ? AND user_id = ?
|
|
|
`).run(isAdmin ? 1 : 0, now, roleChanged ? 1 : 0, roleChanged ? now : null, groupId, userId);
|
|
|
updated++;
|
|
|
} else {
|
|
|
// still update last_seen_at to reflect presence
|
|
|
this.dbInstance.prepare(`
|
|
|
UPDATE group_members
|
|
|
SET last_seen_at = ?
|
|
|
WHERE group_id = ? AND user_id = ?
|
|
|
`).run(now, groupId, userId);
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
// Deactivate absent members
|
|
|
for (const [userId, state] of existing.entries()) {
|
|
|
if (!incoming.has(userId) && state.isActive) {
|
|
|
this.dbInstance.prepare(`
|
|
|
UPDATE group_members
|
|
|
SET is_active = 0,
|
|
|
last_seen_at = ?
|
|
|
WHERE group_id = ? AND user_id = ?
|
|
|
`).run(now, groupId, userId);
|
|
|
deactivated++;
|
|
|
}
|
|
|
}
|
|
|
})();
|
|
|
|
|
|
return { added, updated, deactivated };
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Sync members for all active groups by calling Evolution API and reconciling.
|
|
|
* Devuelve contadores agregados.
|
|
|
*/
|
|
|
static async syncMembersForActiveGroups(): Promise<{ groups: number; added: number; updated: number; deactivated: number }> {
|
|
|
if (process.env.NODE_ENV === 'test') {
|
|
|
return { groups: 0, added: 0, updated: 0, deactivated: 0 };
|
|
|
}
|
|
|
// ensure cache is populated
|
|
|
if (this.activeGroupsCache.size === 0) {
|
|
|
this.cacheActiveGroups();
|
|
|
}
|
|
|
|
|
|
// Etapa 3: gating también en el scheduler masivo
|
|
|
const mode = String(process.env.GROUP_GATING_MODE || 'off').toLowerCase();
|
|
|
const enforce = mode === 'enforce';
|
|
|
if (enforce) {
|
|
|
try { (AllowedGroups as any).dbInstance = this.dbInstance; } catch {}
|
|
|
}
|
|
|
|
|
|
let groups = 0, added = 0, updated = 0, deactivated = 0;
|
|
|
for (const [groupId] of this.activeGroupsCache.entries()) {
|
|
|
try {
|
|
|
if (enforce) {
|
|
|
try {
|
|
|
if (!AllowedGroups.isAllowed(groupId)) {
|
|
|
// Saltar grupos no permitidos en modo enforce
|
|
|
try { Metrics.inc('sync_skipped_group_total'); } catch {}
|
|
|
continue;
|
|
|
}
|
|
|
} catch {
|
|
|
// Si falla el check, no bloquear el grupo
|
|
|
}
|
|
|
}
|
|
|
|
|
|
const snapshot = await this.fetchGroupMembersFromAPI(groupId);
|
|
|
const res = this.reconcileGroupMembers(groupId, snapshot);
|
|
|
groups++;
|
|
|
added += res.added;
|
|
|
updated += res.updated;
|
|
|
deactivated += res.deactivated;
|
|
|
} catch (e) {
|
|
|
console.error(`❌ Failed to sync members for group ${groupId}:`, e instanceof Error ? e.message : String(e));
|
|
|
}
|
|
|
}
|
|
|
console.log('ℹ️ Members sync summary:', { groups, added, updated, deactivated });
|
|
|
return { groups, added, updated, deactivated };
|
|
|
}
|
|
|
|
|
|
public static refreshActiveGroupsCache(): void {
|
|
|
this.cacheActiveGroups();
|
|
|
}
|
|
|
|
|
|
public static startGroupsScheduler(): void {
|
|
|
if (process.env.NODE_ENV === 'test') return;
|
|
|
if (this._groupsSchedulerRunning) return;
|
|
|
this._groupsSchedulerRunning = true;
|
|
|
|
|
|
// Intervalo de grupos configurable; mínimo 10s en desarrollo
|
|
|
let interval = Number(process.env.GROUP_SYNC_INTERVAL_MS);
|
|
|
if (!Number.isFinite(interval) || interval <= 0) {
|
|
|
interval = 24 * 60 * 60 * 1000; // 24h por defecto
|
|
|
}
|
|
|
if (process.env.NODE_ENV === 'development' && interval < 10000) {
|
|
|
interval = 10000;
|
|
|
}
|
|
|
this._groupsIntervalMs = interval;
|
|
|
this._groupsNextTickAt = Date.now() + interval;
|
|
|
|
|
|
this._groupsTimer = setInterval(() => {
|
|
|
// Programar el siguiente tick antes de ejecutar la sincronización
|
|
|
this._groupsNextTickAt = Date.now() + (this._groupsIntervalMs ?? interval);
|
|
|
this.syncGroups().catch(err => {
|
|
|
console.error('❌ Groups scheduler run error:', err);
|
|
|
});
|
|
|
}, interval);
|
|
|
}
|
|
|
|
|
|
public static stopGroupsScheduler(): void {
|
|
|
this._groupsSchedulerRunning = false;
|
|
|
if (this._groupsTimer) {
|
|
|
clearInterval(this._groupsTimer);
|
|
|
this._groupsTimer = null;
|
|
|
}
|
|
|
this._groupsIntervalMs = null;
|
|
|
this._groupsNextTickAt = null;
|
|
|
}
|
|
|
|
|
|
public static getSecondsUntilNextGroupSync(nowMs: number = Date.now()): number | null {
|
|
|
const next = this._groupsNextTickAt;
|
|
|
if (next == null) return null;
|
|
|
const secs = (next - nowMs) / 1000;
|
|
|
return secs > 0 ? secs : 0;
|
|
|
}
|
|
|
|
|
|
public static startMembersScheduler(): void {
|
|
|
if (process.env.NODE_ENV === 'test') return;
|
|
|
if (this._membersSchedulerRunning) return;
|
|
|
this._membersSchedulerRunning = true;
|
|
|
|
|
|
// Intervalo por defecto 6h; configurable por env; mínimo 10s en desarrollo
|
|
|
const raw = process.env.GROUP_MEMBERS_SYNC_INTERVAL_MS;
|
|
|
let interval = Number.isFinite(Number(raw)) && Number(raw) > 0 ? Number(raw) : 6 * 60 * 60 * 1000;
|
|
|
if (process.env.NODE_ENV === 'development' && interval < 10000) {
|
|
|
interval = 10000;
|
|
|
}
|
|
|
|
|
|
this._membersTimer = setInterval(() => {
|
|
|
this.syncMembersForActiveGroups().catch(err => {
|
|
|
console.error('❌ Members scheduler run error:', err);
|
|
|
});
|
|
|
}, interval);
|
|
|
}
|
|
|
|
|
|
public static stopMembersScheduler(): void {
|
|
|
this._membersSchedulerRunning = false;
|
|
|
if (this._membersTimer) {
|
|
|
clearInterval(this._membersTimer);
|
|
|
this._membersTimer = null;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
// ===== Helpers de membresía y snapshot (Etapa 3) =====
|
|
|
|
|
|
private static get MAX_SNAPSHOT_AGE_MS(): number {
|
|
|
const raw = Number(process.env.MAX_MEMBERS_SNAPSHOT_AGE_MS);
|
|
|
return Number.isFinite(raw) && raw > 0 ? raw : 24 * 60 * 60 * 1000; // 24h por defecto
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Devuelve true si la snapshot de un grupo es "fresca" según MAX_SNAPSHOT_AGE_MS.
|
|
|
* Considera no fresca si no hay registro/fecha.
|
|
|
*/
|
|
|
public static isSnapshotFresh(groupId: string, nowMs: number = Date.now()): boolean {
|
|
|
try {
|
|
|
const row = this.dbInstance.prepare(`SELECT last_verified FROM groups WHERE id = ?`).get(groupId) as any;
|
|
|
const lv = row?.last_verified ? String(row.last_verified) : null;
|
|
|
if (!lv) return false;
|
|
|
// Persistimos 'YYYY-MM-DD HH:MM:SS[.mmm]'. Convertimos a ISO-like para Date.parse
|
|
|
const iso = lv.includes('T') ? lv : (lv.replace(' ', 'T') + 'Z');
|
|
|
const ms = Date.parse(iso);
|
|
|
if (!Number.isFinite(ms)) return false;
|
|
|
return (nowMs - ms) <= this.MAX_SNAPSHOT_AGE_MS;
|
|
|
} catch {
|
|
|
return false;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* ¿El usuario figura como miembro activo del grupo?
|
|
|
*/
|
|
|
public static isUserActiveInGroup(userId: string, groupId: string): boolean {
|
|
|
if (!userId || !groupId) return false;
|
|
|
const row = this.dbInstance.prepare(`
|
|
|
SELECT 1
|
|
|
FROM group_members
|
|
|
WHERE group_id = ? AND user_id = ? AND is_active = 1
|
|
|
LIMIT 1
|
|
|
`).get(groupId, userId);
|
|
|
return !!row;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Devuelve todos los group_ids activos donde el usuario figura activo.
|
|
|
* Filtra también por grupos activos en la tabla groups.
|
|
|
*/
|
|
|
public static getActiveGroupIdsForUser(userId: string): string[] {
|
|
|
if (!userId) return [];
|
|
|
const rows = this.dbInstance.prepare(`
|
|
|
SELECT gm.group_id AS id
|
|
|
FROM group_members gm
|
|
|
JOIN groups g ON g.id = gm.group_id
|
|
|
WHERE gm.user_id = ? AND gm.is_active = 1 AND g.active = 1
|
|
|
`).all(userId) as any[];
|
|
|
const set = new Set<string>();
|
|
|
for (const r of rows) {
|
|
|
if (r?.id) set.add(String(r.id));
|
|
|
}
|
|
|
return Array.from(set);
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Devuelve los group_ids donde el usuario es miembro activo y cuya snapshot es fresca.
|
|
|
*/
|
|
|
public static getFreshMemberGroupsForUser(userId: string): string[] {
|
|
|
const gids = this.getActiveGroupIdsForUser(userId);
|
|
|
return gids.filter(gid => this.isSnapshotFresh(gid));
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Asegura un registro de grupo activo en la base de datos (upsert idempotente).
|
|
|
* Si no existe, lo crea con active=1. Si existe y estaba inactivo, lo reactiva.
|
|
|
* Puede actualizar el nombre si se proporciona.
|
|
|
*/
|
|
|
public static ensureGroupExists(groupId: string, name?: string | null): { created: boolean; reactivated: boolean; updatedName: boolean } {
|
|
|
if (!groupId) return { created: false, reactivated: false, updatedName: false };
|
|
|
let created = false, reactivated = false, updatedName = false;
|
|
|
|
|
|
this.dbInstance.transaction(() => {
|
|
|
const row = this.dbInstance.prepare(`SELECT id, active, name FROM groups WHERE id = ?`).get(groupId) as any;
|
|
|
if (!row) {
|
|
|
const community = process.env.WHATSAPP_COMMUNITY_ID || '';
|
|
|
this.dbInstance.prepare(`
|
|
|
INSERT INTO groups (id, community_id, name, active, last_verified)
|
|
|
VALUES (?, ?, ?, 1, CURRENT_TIMESTAMP)
|
|
|
`).run(groupId, community, name || null);
|
|
|
created = true;
|
|
|
} else {
|
|
|
// Reactivar si estaba inactivo y opcionalmente actualizar nombre
|
|
|
const shouldUpdateName = (typeof name === 'string' && name.trim().length > 0 && name !== row.name);
|
|
|
if (row.active !== 1 || shouldUpdateName) {
|
|
|
this.dbInstance.prepare(`
|
|
|
UPDATE groups
|
|
|
SET active = 1,
|
|
|
name = COALESCE(?, name),
|
|
|
last_verified = CURRENT_TIMESTAMP
|
|
|
WHERE id = ?
|
|
|
`).run(shouldUpdateName ? name : null, groupId);
|
|
|
reactivated = row.active !== 1;
|
|
|
updatedName = shouldUpdateName;
|
|
|
}
|
|
|
}
|
|
|
})();
|
|
|
|
|
|
// Actualizar caché
|
|
|
this.cacheActiveGroups();
|
|
|
Metrics.set('active_groups', this.activeGroupsCache.size);
|
|
|
|
|
|
return { created, reactivated, updatedName };
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Asegura tener el nombre/label de un grupo (cache/DB/API) y lo persiste tanto en groups como en allowed_groups.
|
|
|
* Devuelve el nombre si se pudo resolver, o null en caso contrario.
|
|
|
*/
|
|
|
public static async ensureGroupLabelAndName(groupId: string): Promise<string | null> {
|
|
|
try {
|
|
|
if (!groupId) return null;
|
|
|
|
|
|
// 1) Cache en memoria
|
|
|
const cached = this.activeGroupsCache.get(groupId);
|
|
|
if (cached && cached.trim()) {
|
|
|
try { this.ensureGroupExists(groupId, cached); } catch {}
|
|
|
try { (AllowedGroups as any).dbInstance = this.dbInstance; AllowedGroups.upsertPending(groupId, cached, null); } catch {}
|
|
|
this.cacheActiveGroups();
|
|
|
return cached;
|
|
|
}
|
|
|
|
|
|
// 2) DB (tabla groups)
|
|
|
try {
|
|
|
const row = this.dbInstance.prepare('SELECT name FROM groups WHERE id = ?').get(groupId) as any;
|
|
|
const name = row?.name ? String(row.name).trim() : '';
|
|
|
if (name) {
|
|
|
try { this.ensureGroupExists(groupId, name); } catch {}
|
|
|
try { (AllowedGroups as any).dbInstance = this.dbInstance; AllowedGroups.upsertPending(groupId, name, null); } catch {}
|
|
|
this.cacheActiveGroups();
|
|
|
return name;
|
|
|
}
|
|
|
} catch {}
|
|
|
|
|
|
// 3) API (evitar en tests)
|
|
|
if (process.env.NODE_ENV !== 'test') {
|
|
|
const groups = await this.fetchGroupsFromAPI();
|
|
|
const g = groups.find((gg) => gg?.id === groupId);
|
|
|
const subject = g?.subject ? String(g.subject).trim() : '';
|
|
|
if (subject) {
|
|
|
try { this.ensureGroupExists(groupId, subject); } catch {}
|
|
|
try { (AllowedGroups as any).dbInstance = this.dbInstance; AllowedGroups.upsertPending(groupId, subject, null); } catch {}
|
|
|
this.cacheActiveGroups();
|
|
|
return subject;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
return null;
|
|
|
} catch {
|
|
|
return null;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* Sincroniza miembros para un grupo concreto (útil tras detectar un grupo nuevo).
|
|
|
*/
|
|
|
public static async syncMembersForGroup(groupId: string): Promise<{ added: number; updated: number; deactivated: number }> {
|
|
|
// Gating en modo 'enforce': solo sincronizar miembros para grupos permitidos
|
|
|
try {
|
|
|
const mode = String(process.env.GROUP_GATING_MODE || 'off').toLowerCase();
|
|
|
if (mode === 'enforce') {
|
|
|
try {
|
|
|
(AllowedGroups as any).dbInstance = this.dbInstance;
|
|
|
if (!AllowedGroups.isAllowed(groupId)) {
|
|
|
try { Metrics.inc('sync_skipped_group_total'); } catch {}
|
|
|
return { added: 0, updated: 0, deactivated: 0 };
|
|
|
}
|
|
|
} catch {
|
|
|
// Si el check falla, seguimos sin bloquear
|
|
|
}
|
|
|
}
|
|
|
} catch {}
|
|
|
|
|
|
try {
|
|
|
// Asegurar existencia del grupo en DB (FKs) antes de reconciliar
|
|
|
this.ensureGroupExists(groupId);
|
|
|
const snapshot = await (this as any).fetchGroupMembersFromAPI(groupId);
|
|
|
return this.reconcileGroupMembers(groupId, snapshot);
|
|
|
} catch (e) {
|
|
|
console.error(`❌ Failed to sync members for group ${groupId}:`, e instanceof Error ? e.message : String(e));
|
|
|
return { added: 0, updated: 0, deactivated: 0 };
|
|
|
}
|
|
|
}
|
|
|
}
|