/** * Engagement Sync Service * Orchestrates Microsoft Graph → PostgreSQL sync for employee engagement data */ import { getMsgraphClient } from './msgraph-factory'; import type { CalendarEvent, UserMessage } from './msgraph-client'; import { postgresClient } from './postgres-client'; const PERIODS = ['D7', 'D30', 'D90'] as const; type Period = typeof PERIODS[number]; const PERIOD_DAYS: Record = { D7: 7, D30: 30, D90: 90 }; interface CalendarBucket { meetingCount: number; durationSeconds: number; externalCount: number; } function computeCalendarBuckets( events: CalendarEvent[], userEmail: string, internalDomains: Set, now: Date ): Record { const cutoffs: Record = { D7: now.getTime() - 7 * 24 * 60 * 60 * 1000, D30: now.getTime() - 30 * 24 * 60 * 60 * 1000, D90: now.getTime() - 90 * 24 * 60 * 60 * 1000, }; const result: Record = { D7: { meetingCount: 0, durationSeconds: 0, externalCount: 0 }, D30: { meetingCount: 0, durationSeconds: 0, externalCount: 0 }, D90: { meetingCount: 0, durationSeconds: 0, externalCount: 0 }, }; for (const event of events) { const eventTime = new Date(event.start.dateTime).getTime(); const durationSec = Math.max( 0, Math.round( (new Date(event.end.dateTime).getTime() - new Date(event.start.dateTime).getTime()) / 1000 ) ); // An attendee is external if their domain is not in the org's verified domains // (filter out the user themselves to avoid false positives) const hasExternal = event.attendees.some(a => { const email = (a.emailAddress?.address ?? '').toLowerCase(); if (email === userEmail.toLowerCase()) return false; const domain = email.split('@')[1]; return domain && !internalDomains.has(domain); }); for (const period of PERIODS) { if (eventTime >= cutoffs[period]) { result[period].meetingCount++; result[period].durationSeconds += durationSec; if (hasExternal) result[period].externalCount++; } } } return result; } export class EngagementSyncService { private syncInProgress = false; isSyncInProgress(): boolean { return this.syncInProgress; } async sync(): Promise<{ usersUpserted: number; snapshotsUpserted: number }> { if (this.syncInProgress) { throw new Error('Engagement sync already in progress'); } this.syncInProgress = true; const startTime = Date.now(); console.log('[ENGAGEMENT-SYNC] Starting sync...'); try { const client = getMsgraphClient(); // 1. Fetch org's verified domains (to identify external attendees) const orgDomains = await client.getOrganizationDomains(); const internalDomains = new Set(orgDomains); console.log(`[ENGAGEMENT-SYNC] Internal domains: ${[...internalDomains].join(', ')}`); // 2. Sync users console.log('[ENGAGEMENT-SYNC] Fetching Graph users...'); const graphUsers = await client.getUsers(); const licencedUsers = graphUsers.filter(u => u.mail || u.userPrincipalName); for (const user of licencedUsers) { const email = (user.mail || user.userPrincipalName || '').toLowerCase(); if (!email) continue; await postgresClient.query( `INSERT INTO graph_users (id, display_name, email, job_title, department, account_enabled, synced_at) VALUES ($1, $2, $3, $4, $5, $6, NOW()) ON CONFLICT (id) DO UPDATE SET display_name = EXCLUDED.display_name, email = EXCLUDED.email, job_title = EXCLUDED.job_title, department = EXCLUDED.department, account_enabled = EXCLUDED.account_enabled, synced_at = NOW()`, [user.id, user.displayName, email, user.jobTitle, user.department, user.accountEnabled ?? true] ); } console.log(`[ENGAGEMENT-SYNC] Upserted ${licencedUsers.length} graph users`); // 3. Fetch Teams + Email activity reports for each period const today = new Date().toISOString().split('T')[0]; const now = new Date(); // Build snapshot data map: email → period → data type SnapshotData = { teamsChatMessages: number; teamsPrivateMessages: number; teamsCalls: number; teamsMeetingsAttended: number; teamsMeetingsOrganized: number; audioDurationSeconds: number; emailsSent: number; emailsReceived: number; emailsRead: number; meetingDurationSeconds: number; meetingsWithExternal: number; lastActivityDate: string | null; afterHoursMessages: number; }; const snapshotMap = new Map>(); for (const period of PERIODS) { console.log(`[ENGAGEMENT-SYNC] Fetching activity for period ${period}...`); const [teamsRows, emailRows] = await Promise.all([ client.getTeamsActivity(period).catch(err => { console.warn(`[ENGAGEMENT-SYNC] Teams activity failed for ${period}:`, err.message); return []; }), client.getEmailActivity(period).catch(err => { console.warn(`[ENGAGEMENT-SYNC] Email activity failed for ${period}:`, err.message); return []; }), ]); const teamsMap = new Map(teamsRows.map(r => [r.userPrincipalName.toLowerCase(), r])); const emailMap = new Map(emailRows.map(r => [r.userPrincipalName.toLowerCase(), r])); const allEmails = new Set([ ...teamsRows.map(r => r.userPrincipalName.toLowerCase()), ...emailRows.map(r => r.userPrincipalName.toLowerCase()), ]); for (const email of allEmails) { const teams = teamsMap.get(email); const mail = emailMap.get(email); const dates = [teams?.lastActivityDate, mail?.lastActivityDate].filter(Boolean) as string[]; const lastActivity = dates.length > 0 ? dates.sort().reverse()[0] : null; if (!snapshotMap.has(email)) snapshotMap.set(email, new Map()); snapshotMap.get(email)!.set(period, { teamsChatMessages: teams?.teamChatMessageCount ?? 0, teamsPrivateMessages: teams?.privateChatMessageCount ?? 0, teamsCalls: teams?.callCount ?? 0, teamsMeetingsAttended: teams?.meetingsAttendedCount ?? 0, teamsMeetingsOrganized: teams?.meetingsOrganizedCount ?? 0, audioDurationSeconds: teams?.audioDurationSeconds ?? 0, emailsSent: mail?.sendCount ?? 0, emailsReceived: mail?.receiveCount ?? 0, emailsRead: mail?.readCount ?? 0, meetingDurationSeconds: 0, meetingsWithExternal: 0, lastActivityDate: lastActivity, afterHoursMessages: 0, }); } } // 4. Fetch calendar events per user (90 days, aggregate into all period buckets) console.log('[ENGAGEMENT-SYNC] Fetching calendar events...'); const activeUsers = licencedUsers.filter(u => { const email = (u.mail || u.userPrincipalName || '').toLowerCase(); return snapshotMap.has(email); }); // Build contact email → { contactId, companyId } index for attendee matching const contactRows = await postgresClient.query( `SELECT id, company_id, LOWER(email_address) as e1, LOWER(email_address2) as e2, LOWER(email_address3) as e3 FROM contacts WHERE (is_deleted = false OR is_deleted IS NULL)` ); const contactEmailIndex = new Map(); for (const row of contactRows.rows) { for (const e of [row.e1, row.e2, row.e3]) { if (e && !contactEmailIndex.has(e)) { contactEmailIndex.set(e, { contactId: row.id, companyId: row.company_id }); } } } const calStart = new Date(now.getTime() - 90 * 24 * 60 * 60 * 1000); let calFetched = 0; let calSkipped = 0; for (const user of activeUsers) { const email = (user.mail || user.userPrincipalName || '').toLowerCase(); try { const events = await client.getUserCalendarEvents(user.id, calStart, now); if (events.length > 0) { const buckets = computeCalendarBuckets(events, email, internalDomains, now); for (const period of PERIODS) { const snap = snapshotMap.get(email)?.get(period); if (snap) { snap.meetingDurationSeconds = buckets[period].durationSeconds; snap.meetingsWithExternal = buckets[period].externalCount; } } // Persist individual meeting records for (const event of events) { if (!event.id) continue; try { const startTime = new Date(event.start.dateTime); const endTime = new Date(event.end.dateTime); const durationMinutes = Math.max( 0, Math.round((endTime.getTime() - startTime.getTime()) / 60000) ); const externalAttendees = event.attendees.filter(a => { const aEmail = (a.emailAddress?.address ?? '').toLowerCase(); if (aEmail === email) return false; const domain = aEmail.split('@')[1]; return domain && !internalDomains.has(domain); }); const attendeeCount = event.attendees.length; const meetingResult = await postgresClient.query( `INSERT INTO teams_meetings (graph_event_id, user_email, subject, start_time, end_time, duration_minutes, is_online_meeting, attendee_count, synced_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW()) ON CONFLICT (user_email, graph_event_id) DO UPDATE SET subject = EXCLUDED.subject, start_time = EXCLUDED.start_time, end_time = EXCLUDED.end_time, duration_minutes = EXCLUDED.duration_minutes, is_online_meeting = EXCLUDED.is_online_meeting, attendee_count = EXCLUDED.attendee_count, synced_at = NOW() RETURNING id`, [event.id, email, event.subject, startTime, endTime, durationMinutes, event.isOnlineMeeting, attendeeCount] ); const meetingId = meetingResult.rows[0]?.id; if (!meetingId) continue; // Re-sync attendees clean await postgresClient.query( `DELETE FROM teams_meeting_attendees WHERE meeting_id = $1`, [meetingId] ); let clientCount = 0; for (const att of externalAttendees) { const attEmail = (att.emailAddress?.address ?? '').toLowerCase(); const attName = att.emailAddress?.name ?? null; const match = attEmail ? contactEmailIndex.get(attEmail) : undefined; await postgresClient.query( `INSERT INTO teams_meeting_attendees (meeting_id, attendee_email, attendee_name, matched_contact_id, matched_company_id) VALUES ($1, $2, $3, $4, $5)`, [meetingId, attEmail || null, attName, match?.contactId ?? null, match?.companyId ?? null] ); if (match?.companyId) clientCount++; } await postgresClient.query( `UPDATE teams_meetings SET client_attendee_count = $1, has_client_attendees = $2 WHERE id = $3`, [clientCount, clientCount > 0, meetingId] ); } catch (meetingErr) { const msg = meetingErr instanceof Error ? meetingErr.message : String(meetingErr); console.warn(`[ENGAGEMENT-SYNC] Meeting persist failed for event ${event.id}: ${msg}`); } } } calFetched++; } catch (err) { calSkipped++; const msg = err instanceof Error ? err.message : String(err); console.warn(`[ENGAGEMENT-SYNC] Calendar fetch failed for ${email}: ${msg}`); } } console.log(`[ENGAGEMENT-SYNC] Calendar: ${calFetched} fetched, ${calSkipped} skipped`); // 5. Count after-hours messages per user (requires Chat.Read.All permission) // After-hours window: 5:30 PM – 7:00 AM America/New_York const AFTER_HOURS_START_MIN = 17 * 60 + 30; // 1050 — 5:30 PM local const AFTER_HOURS_END_MIN = 7 * 60; // 420 — 7:00 AM local // Returns minutes since midnight in America/New_York (handles EST/EDT automatically) function toEasternMinutes(dt: Date): number { const parts = new Intl.DateTimeFormat('en-US', { timeZone: 'America/New_York', hour: 'numeric', minute: 'numeric', hour12: false, }).formatToParts(dt); const h = parseInt(parts.find(p => p.type === 'hour')?.value ?? '0'); const m = parseInt(parts.find(p => p.type === 'minute')?.value ?? '0'); return h * 60 + m; } const periodCutoffs: Record = { D7: now.getTime() - 7 * 24 * 60 * 60 * 1000, D30: now.getTime() - 30 * 24 * 60 * 60 * 1000, D90: now.getTime() - 90 * 24 * 60 * 60 * 1000, }; let msgFetched = 0; let msgSkipped = 0; console.log(`[ENGAGEMENT-SYNC] Fetching after-hours messages for ${activeUsers.length} users...`); for (const user of activeUsers) { const email = (user.mail || user.userPrincipalName || '').toLowerCase(); if (!snapshotMap.has(email)) continue; try { const messages: UserMessage[] = await client.getUserMessages(user.id, calStart, now); const afterD90 = messages.filter(msg => { const estMin = toEasternMinutes(new Date(msg.createdDateTime)); return estMin >= AFTER_HOURS_START_MIN || estMin < AFTER_HOURS_END_MIN; }).length; console.log(`[ENGAGEMENT-SYNC] ${email}: ${messages.length} msgs total, ${afterD90} after-hours`); // Bucket by period for (const period of PERIODS) { const snap = snapshotMap.get(email)?.get(period); if (!snap) continue; snap.afterHoursMessages = messages.filter(msg => { const msgMs = new Date(msg.createdDateTime).getTime(); if (msgMs < periodCutoffs[period]) return false; const estMin = toEasternMinutes(new Date(msg.createdDateTime)); return estMin >= AFTER_HOURS_START_MIN || estMin < AFTER_HOURS_END_MIN; }).length; } msgFetched++; } catch (err) { const errMsg = err instanceof Error ? err.message : String(err); console.warn(`[ENGAGEMENT-SYNC] Message fetch failed for ${email}: ${errMsg.slice(0, 200)}`); msgSkipped++; } // Small pause between users to stay under Graph API rate limits (10 req/10 sec per app) await new Promise(r => setTimeout(r, 500)); } console.log(`[ENGAGEMENT-SYNC] After-hours messages: ${msgFetched} fetched, ${msgSkipped} skipped`); // 6. Upsert all snapshots let totalSnapshots = 0; for (const [email, periodMap] of snapshotMap) { for (const [period, snap] of periodMap) { await postgresClient.query( `INSERT INTO engagement_snapshots ( user_email, period_type, period_end, teams_chat_messages, teams_private_messages, teams_calls, teams_meetings_attended, teams_meetings_organized, emails_sent, emails_received, emails_read, audio_duration_seconds, meeting_duration_seconds, meetings_with_external, after_hours_messages, last_activity_date, synced_at ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,NOW()) ON CONFLICT (user_email, period_type, period_end) DO UPDATE SET teams_chat_messages = EXCLUDED.teams_chat_messages, teams_private_messages = EXCLUDED.teams_private_messages, teams_calls = EXCLUDED.teams_calls, teams_meetings_attended = EXCLUDED.teams_meetings_attended, teams_meetings_organized = EXCLUDED.teams_meetings_organized, emails_sent = EXCLUDED.emails_sent, emails_received = EXCLUDED.emails_received, emails_read = EXCLUDED.emails_read, audio_duration_seconds = EXCLUDED.audio_duration_seconds, meeting_duration_seconds = EXCLUDED.meeting_duration_seconds, meetings_with_external = EXCLUDED.meetings_with_external, after_hours_messages = EXCLUDED.after_hours_messages, last_activity_date = EXCLUDED.last_activity_date, synced_at = NOW()`, [ email, period, today, snap.teamsChatMessages, snap.teamsPrivateMessages, snap.teamsCalls, snap.teamsMeetingsAttended, snap.teamsMeetingsOrganized, snap.emailsSent, snap.emailsReceived, snap.emailsRead, snap.audioDurationSeconds, snap.meetingDurationSeconds, snap.meetingsWithExternal, snap.afterHoursMessages, snap.lastActivityDate, ] ); totalSnapshots++; } } const duration = Date.now() - startTime; console.log(`[ENGAGEMENT-SYNC] Done in ${duration}ms. Users: ${licencedUsers.length}, Snapshots: ${totalSnapshots}`); return { usersUpserted: licencedUsers.length, snapshotsUpserted: totalSnapshots }; } finally { this.syncInProgress = false; } } } let _instance: EngagementSyncService | null = null; export function getEngagementSyncService(): EngagementSyncService { if (!_instance) { _instance = new EngagementSyncService(); } return _instance; }