wulf-pulse/lib/services/engagement-sync-service.ts

/**
 * 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<Period, number> = { D7: 7, D30: 30, D90: 90 };

interface CalendarBucket {
  meetingCount: number;
  durationSeconds: number;
  externalCount: number;
}

function computeCalendarBuckets(
  events: CalendarEvent[],
  userEmail: string,
  internalDomains: Set<string>,
  now: Date
): Record<Period, CalendarBucket> {
  const cutoffs: Record<Period, number> = {
    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<Period, CalendarBucket> = {
    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<string, Map<Period, SnapshotData>>();

      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<string, { contactId: number; companyId: number | null }>();
      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<Period, number> = {
        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;
}