import { ProgramSession, ZoomAnalyticsAttendeeSummary, ZoomGeneratedRegistrantLink } from 'src/common/entities';
import InifniBadRequestException from 'src/common/exceptions/infini-badrequest-exception';
import { ERROR_CODES } from 'src/common/constants/error-string-constants';
import { OnlineTypeEnum } from 'src/common/enum/online-type.enum';
import { ZoomAnalyticsResourceType } from 'src/common/enum/zoom-analytics-resource-type.enum';
import { RegistrationOnlineSessionActivationStatus } from 'src/common/enum/registration-online-session-activation-status.enum';
import { SessionKpiFilter } from 'src/common/enum/session-kpi.enum';
import { AttendanceStatus } from 'src/common/enum/attendance-status.enum';
import { AppLoggerService } from 'src/common/services/logger.service';
import {
  ZOOM_ANALYTICS_LOG,
  ZOOM_ANALYTICS_DEFAULTS,
  ZOOM_ANALYTICS_DROP_OFF_REASONS,
  ZOOM_ANALYTICS_FOLLOW_UP,
  ZOOM_ANALYTICS_LATE_COMER_BUCKETS,
  ZOOM_SINCERITY_MAP_BAND_LABELS,
} from '../constants/zoom-analytics.constants';
import { ZoomWebinarRepository } from '../repositories/zoom-webinar.repository';
import { ZoomRegistrationRepository } from '../repositories/zoom-registration.repository';
import { ZoomAnalyticsSessionReport } from './zoom-report-api.client';
import {
  ZoomAnalyticsRosterRepository,
  ZoomAnalyticsRosterEntry,
} from '../repositories/zoom-analytics-roster.repository';
import { ZoomAnalyticsSessionSummaryRepository } from '../repositories/zoom-analytics-session-summary.repository';
import { ZoomAnalyticsAttendeeSummaryRepository } from '../repositories/zoom-analytics-attendee-summary.repository';
import { ZoomAnalyticsLiveEventRepository } from '../repositories/zoom-analytics-live-event.repository';
import { ZoomGeneratedRegistrantLinkRepository } from '../repositories/zoom-generated-registrant-link.repository';
import { ZoomAnalyticsConfigService } from './zoom-analytics-config.service';
import { OnlineAttendanceService } from 'src/online-attendance/online-attendance.service';
import { RegistrationManualMarks } from 'src/online-attendance/interfaces/online-attendance.interface';
import { ZoomAnalyticsProvider } from '../interfaces/zoom-analytics-provider.interface';
import {
  ZoomSessionKpis,
  ZoomGeneralAttendeeKpis,
  ZoomLiveStatus,
  ZoomAttendeeQuery,
  PaginatedZoomAttendeeRows,
  ZoomAttendeeRow,
  ZoomSessionDashboard,
  ZoomSessionAttendanceSummary,
  ZoomJourneyPoint,
  ZoomDropOffAnalysis,
  ZoomMultipleLoginsBreakdown,
  ZoomLateComerAnalysis,
  ZoomSincerityMap,
  ZoomFollowUpRow,
  PaginatedZoomFollowUpRows,
} from '../interfaces/zoom-analytics.interface';
import { ZoomLiveEventType } from 'src/common/enum/zoom-live-event-type.enum';
import {
  buildZoomRegistrantEmail,
  parseZoomRegistrantRegistrationId,
} from '../utils/zoom-registrant-email.util';

interface LiveEventRecord {
  email: string;
  eventType: ZoomLiveEventType;
  occurredAt: Date;
  zoomDisplayName?: string | null;
  zoomParticipantId?: string | null;
  leaveReason?: string | null;
}

/** The slice of a live-event row the drop-off reason fallback needs. */
interface LeaveEventRecord {
  eventType: ZoomLiveEventType;
  occurredAt: Date;
  leaveReason?: string | null;
}

/** Per-email rollup of the append-only live event log, as of a given instant. */
interface LiveEventAggregate {
  firstJoinAt: Date | null;
  lastJoinAt: Date | null;
  lastLeaveAt: Date | null;
  /** True while at least one device is connected — the union across all concurrently open devices, not a single interval. */
  isCurrentlyJoined: boolean;
  /** Raw count of JOINED events (one per device connection, including concurrent ones) — use `rejoinCount`/presence transitions for "did they leave and come back." */
  joinCount: number;
  /** Raw count of LEFT events (one per device disconnection) — use `dropoffCount` for "did every device disconnect." */
  leaveCount: number;
  /**
   * Total wall-clock seconds with at least one device connected (union across concurrent devices),
   * counted from the session's actual start time onward — a seeker with two devices open for the
   * same 10 minutes counts as 10 minutes present, not 20. Time present before the session started is
   * excluded here and tracked separately in `preSessionDurationSeconds`.
   */
  durationSeconds: number;
  /** Wall-clock seconds present before the session's actual start time (e.g. joined early/waiting room) — excluded from `durationSeconds`. */
  preSessionDurationSeconds: number;
  /** The most recent non-null `zoomDisplayName` seen across this seeker's events, chronologically. */
  latestDisplayName: string | null;
  /**
   * Best-known count of distinct devices/connections used to join this seeker's link — the larger of
   * (a) distinct `zoomParticipantId`s seen (only populated when a participant is logged into a Zoom
   * account — empty for the common case of an unauthenticated registrant) and (b) the peak number of
   * concurrently-open connections inferred from JOIN/LEFT interleaving alone, which needs no id: two
   * JOINs with no LEFT between them proves two connections were open. Both are lower bounds, so
   * whichever is higher is the more accurate one. 1 once we know they joined at all but neither signal
   * proves more than one device; 0 if they never joined.
   */
  deviceCount: number;
  /** Number of times every open device disconnected at once (presence went from >0 to 0) — the true "drop off" count, unlike raw leaveCount which also fires for a single device leaving while another stays connected. */
  fullDropoffCount: number;
  /** Number of times presence went from 0 back to >0, excluding the very first join — the true "rejoin" count, unlike raw joinCount which also counts a second concurrent device joining as if it were a rejoin. */
  rejoinTransitionCount: number;
  /** When presence last went from >0 to 0 (last full drop-off), or null if it never has. */
  lastFullDropoffAt: Date | null;
  /** When presence last went from 0 to >0 on a rejoin (excludes the first join), or null if there was no rejoin. */
  lastRejoinAt: Date | null;
  /** How many devices/connections are open right now (0 if `isCurrentlyJoined` is false) — how many self-heal LEFT corrections it'd take to fully clear this seeker's presence. */
  openConnectionCount: number;
  /** Every full drop-off (presence went from >0 to 0), chronologically, each carrying the closing LEFT event's own `leave_reason` (null when Zoom sent none). */
  dropoffEvents: { at: Date; reason: string | null }[];
  /** Every instant presence went from 0 back to >0 excluding the first join, chronologically — the per-event detail behind `rejoinTransitionCount`. */
  rejoinTimes: Date[];
  /** Closed presence intervals (union across concurrent devices); a still-open interval is closed at the aggregation's `asOf`. */
  presenceIntervals: { start: Date; end: Date }[];
}

/**
 * Shared engine behind every "live-event-log-backed" Zoom analytics provider
 * (Webinar, Meeting). `zoom_analytics_live_event` (our own webhook-driven
 * log) is the source of truth for every per-attendee and per-seeker
 * join/leave/duration figure — Zoom's external Report API is used only for
 * the session-level start time and duration, which aren't observable from
 * webhooks alone. Subclasses only need to say how to fetch the resource's
 * roster and its post-session report (see `fetchRoster`/`fetchReport`) — the
 * rest (reconcile, KPIs, attendee table, live status, join/leave recording,
 * self-healing) is identical for any resource keyed by
 * (sessionId, onlineSessionId, email).
 */
export abstract class ZoomLiveEventAnalyticsProviderBase implements ZoomAnalyticsProvider {
  abstract readonly key: ZoomAnalyticsResourceType;

  /** The `OnlineTypeEnum` this provider is valid for — `reconcile()` refuses any session whose onlineType differs. */
  protected abstract readonly expectedOnlineType: OnlineTypeEnum;

  constructor(
    protected readonly webinarRepository: ZoomWebinarRepository,
    protected readonly rosterRepository: ZoomAnalyticsRosterRepository,
    protected readonly sessionSummaryRepository: ZoomAnalyticsSessionSummaryRepository,
    protected readonly attendeeSummaryRepository: ZoomAnalyticsAttendeeSummaryRepository,
    protected readonly liveEventRepository: ZoomAnalyticsLiveEventRepository,
    protected readonly config: ZoomAnalyticsConfigService,
    protected readonly logger: AppLoggerService,
    protected readonly onlineAttendanceService: OnlineAttendanceService,
    protected readonly registrationRepository: ZoomRegistrationRepository,
    protected readonly generatedLinkRepository: ZoomGeneratedRegistrantLinkRepository,
  ) {}

  /**
   * Fetches the resource-specific roster (webinar panelists, meeting registrants, ...).
   * `rmContactId`, when passed, scopes the roster to that RM's own seekers — only
   * `getKpis`/`getAttendeeTable` (the two user-facing reads) ever pass it; every
   * internal/bookkeeping caller (reconcile, webhook matching, self-heal) omits it
   * and always operates on the full roster.
   */
  protected abstract fetchRoster(onlineSessionId: number, rmContactId?: number): Promise<ZoomAnalyticsRosterEntry[]>;

  /** Fetches the resource-specific post-session report (start time + duration), or null if not generated yet. */
  protected abstract fetchReport(extId: string): Promise<ZoomAnalyticsSessionReport | null>;


  /**
   * Attendee-level fields (joinedAt, dropoffCount, rejoinCount,
   * lastDropoffAt, lastRejoinedAt, durationSeconds) are always recomputed
   * from the live event log — never gated on the Report API. Only the
   * session-level `actualStartAt`/`actualDurationMinutes` and the final
   * `reconciledAt` stamp wait on Zoom's Report API, since that's the only
   * place those two figures come from. There is no scheduled retry: the one
   * post-session Zoom pull happens here, triggered by the admin's
   * "Mark session as complete" click (and the `*.ended` webhook) — re-running
   * it is safe and refreshes anything the report hadn't generated yet.
   */
  async reconcile(session: ProgramSession): Promise<void> {
    const onlineSession = session.onlineSession;
    if (session.onlineType !== this.expectedOnlineType || !onlineSession?.externalId) {
      throw new InifniBadRequestException(
        ERROR_CODES.ZOOM_ANALYTICS_RESOURCE_TYPE_NOT_SUPPORTED,
        null,
        null,
        String(session.onlineType),
      );
    }
    const extId = onlineSession.externalId;

    this.logger.log(ZOOM_ANALYTICS_LOG.RECONCILIATION_STARTED, { sessionId: session.id, extId });

    const [report, events, roster, existingSummary, generatedLinks] = await Promise.all([
      this.fetchReport(extId),
      this.liveEventRepository.findAllByOnlineSessionId(onlineSession.id),
      this.fetchRoster(onlineSession.id),
      this.sessionSummaryRepository.findBySessionId(session.id),
      this.generatedLinkRepository.findActiveMapBySession(session.id),
    ]);

    const reportReady = !!report?.start_time && !!report?.duration;
    const sessionStart = report?.start_time
      ? new Date(report.start_time)
      : (existingSummary?.actualStartAt ?? session.startsAt ?? null);
    const actualDurationMinutes =
      report?.duration ?? existingSummary?.actualDurationMinutes ?? null;
    const sessionEnd =
      sessionStart && actualDurationMinutes
        ? new Date(sessionStart.getTime() + actualDurationMinutes * 60_000)
        : (existingSummary?.actualEndAt ?? null);
    const asOf = sessionEnd ?? new Date();

    const aggregates = this.aggregateLiveEvents(events, asOf, sessionStart);
    const aggregatesByRegistrationId = this.indexAggregatesByRegistrationId(aggregates);
    const dropoffGraceMs = ZOOM_ANALYTICS_DEFAULTS.DROPOFF_GRACE_SECONDS * 1000;
    const lateThresholdMs = this.config.getLateJoinThresholdSeconds() * 1000;

    let totalJoined = 0;
    let joinedLate = 0;
    let dropped = 0;
    let rejoined = 0;
    const reconciledAt = new Date();
    const attendeePayloads: Array<
      Partial<ZoomAnalyticsAttendeeSummary> & {
        sessionId: number;
        registrationId: number | null;
        email: string;
      }
    > = [];

    for (const seeker of roster) {
      const aggregate = this.resolveSeekerAggregate(seeker, aggregatesByRegistrationId);
      const payload = this.attendeeUpsertPayload(session, onlineSession.id, seeker, aggregate, sessionEnd);

      if (aggregate && aggregate.joinCount > 0) {
        totalJoined++;
        const isLateJoiner =
          !!sessionStart &&
          aggregate.firstJoinAt!.getTime() - sessionStart.getTime() > lateThresholdMs;
        // "Dropped" (the coarser session-level KPI) means they left well before the session actually
        // ended — not simply that their last event is a leave, which is also true of everyone once
        // Zoom disconnects them at the natural end.
        const isDropped =
          !aggregate.isCurrentlyJoined &&
          (!sessionEnd ||
            !aggregate.lastLeaveAt ||
            sessionEnd.getTime() - aggregate.lastLeaveAt.getTime() > dropoffGraceMs);
        if (isLateJoiner) joinedLate++;
        if (isDropped) dropped++;
        if (payload.rejoinCount! > 0) rejoined++;
      }

      attendeePayloads.push({ ...payload, reconciledAt });
    }

    // General attendees (no registration to attribute to) never count toward the KPI figures above —
    // only toward their own summary row, kept out of the normal attendee table/KPIs by
    // `applyKpiFilter`'s registration_id guard and only surfaced via the dedicated GENERAL kpiFilter.
    for (const payload of this.buildGeneralAttendeePayloads(session, onlineSession.id, aggregates, generatedLinks, sessionEnd)) {
      attendeePayloads.push({ ...payload, reconciledAt });
    }

    await this.attendeeSummaryRepository.upsertMany(attendeePayloads);

    await this.sessionSummaryRepository.upsert({
      programId: session.programId,
      sessionId: session.id,
      onlineSessionId: onlineSession.id,
      resourceType: this.key,
      actualStartAt: sessionStart,
      actualDurationMinutes,
      totalPanelists: roster.length,
      totalJoined,
      notJoined: roster.length - totalJoined,
      joinedLate,
      dropped,
      rejoined,
      // Only stamped once the Report API has actually confirmed start/duration — a null reconciledAt
      // just means another "Mark session as complete" click (or ended-webhook reconcile) should re-pull
      // once Zoom has generated the report; attendee data above is already fresh either way.
      ...(reportReady ? { reconciledAt: new Date() } : {}),
    });

    this.logger.log(ZOOM_ANALYTICS_LOG.RECONCILIATION_COMPLETED, {
      sessionId: session.id,
      extId,
      totalPanelists: roster.length,
      totalJoined,
      reportReady,
    });
  }


  /**
   * All 8 KPIs except duration are computed live from the local roster +
   * event log — not from the post-session summary row — so they're available
   * immediately, without waiting on Zoom's Report API to have data (that API
   * only returns data for sessions that have actually already run).
   * `durationMinutes` is the one exception: it genuinely only exists once
   * Zoom's Report API has it, via `reconcile()`.
   *
   * `totalSeekersJoined` always means "joined at any point during the
   * session" (matches `seekersNotJoined`'s complement) — never "currently
   * connected right now." A seeker who joined and later left (or is between
   * a leave and a reconnect) still counts; use getLiveStatus for a snapshot
   * of who's connected at this exact instant.
   */
  async getKpis(session: ProgramSession, rmContactId?: number): Promise<ZoomSessionKpis> {
    const onlineSession = session.onlineSession;
    const summary = await this.sessionSummaryRepository.findBySessionId(session.id);

    if (!onlineSession) {
      return {
        sessionId: session.id,
        startTime: session.startsAt ?? null,
        durationMinutes: summary?.actualDurationMinutes ?? null,
        totalPanelists: 0,
        totalSeekersJoined: 0,
        seekersNotJoined: 0,
        seekersJoinedLate: 0,
        seekersDropped: 0,
        seekersRejoined: 0,
        finalPresent: 0,
        finalAbsent: 0,
        reconciledAt: summary?.reconciledAt ?? null,
      };
    }

    const [roster, events, finalPresentRegistrationIds] = await Promise.all([
      this.fetchRoster(onlineSession.id, rmContactId),
      this.liveEventRepository.findAllByOnlineSessionId(onlineSession.id),
      this.resolveFinalPresentRegistrationIds(session.id),
    ]);

    const sessionStart = session.startsAt ?? null;
    const sessionEnd = this.resolveKnownSessionEnd(session, summary, sessionStart);
    const aggregates = this.aggregateLiveEvents(
      events,
      this.resolveDurationAsOf(session, summary, sessionStart),
      sessionStart,
    );
    const aggregatesByRegistrationId = this.indexAggregatesByRegistrationId(aggregates);
    const lateThresholdMs = this.config.getLateJoinThresholdSeconds() * 1000;
    const dropoffGraceMs = ZOOM_ANALYTICS_DEFAULTS.DROPOFF_GRACE_SECONDS * 1000;

    let everJoined = 0;
    let joinedLate = 0;
    let dropped = 0;
    let rejoined = 0;

    for (const seeker of roster) {
      const aggregate = this.resolveSeekerAggregate(seeker, aggregatesByRegistrationId);
      if (!aggregate || aggregate.joinCount === 0) continue; // Only registered seekers count toward seeker KPIs.

      everJoined++;

      if (
        sessionStart &&
        aggregate.firstJoinAt &&
        aggregate.firstJoinAt.getTime() - sessionStart.getTime() > lateThresholdMs
      ) {
        joinedLate++;
      }

      // Independent flags, not mutually exclusive: "dropped" means they left well before the
      // session actually ended (same grace-window rule as reconcile()/computeAttendanceSummary) —
      // not simply "not currently connected," which is also true of everyone once Zoom disconnects
      // them at the natural end. "Rejoined" is a lifetime flag (have they ever left and come back).
      // Someone who rejoined 3 times and is currently offline again counts as both.
      const isDropped =
        !aggregate.isCurrentlyJoined &&
        (!sessionEnd || !aggregate.lastLeaveAt || sessionEnd.getTime() - aggregate.lastLeaveAt.getTime() > dropoffGraceMs);
      if (isDropped) dropped++;
      if (aggregate.rejoinTransitionCount > 0) rejoined++;
    }

    const finalPresent = roster.filter((seeker) => finalPresentRegistrationIds.has(seeker.registrationId)).length;

    return {
      sessionId: session.id,
      startTime: session.startsAt ?? null,
      durationMinutes: summary?.actualDurationMinutes ?? null,
      totalPanelists: roster.length,
      totalSeekersJoined: everJoined,
      // "Not joined" means never showed up, ever — a seeker who joined then left still "showed up," so
      // this never recovers once someone has shown up at least once.
      seekersNotJoined: Math.max(roster.length - everJoined, 0),
      seekersJoinedLate: joinedLate,
      seekersDropped: dropped,
      seekersRejoined: rejoined,
      finalPresent,
      finalAbsent: Math.max(roster.length - finalPresent, 0),
      reconciledAt: summary?.reconciledAt ?? null,
    };
  }

  /**
   * The general-attendee counterpart to `getKpis` — computed live from the event log alone, same as
   * `getKpis`. `knownAttendees` counts every active `zoom_generated_registrant_link` row for this
   * session regardless of join status (mirrors how `totalPanelists` counts the whole roster in
   * `getKpis`); `unknownAttendees` counts general-pool joiners matching none of them — true
   * anonymous walk-ins. `generalLoggedIn`/`generalJoinedLate`/`generalDropped`/`generalRejoined`
   * describe join OUTCOMES, so — same as before this known/unknown split existed — they only count
   * people who actually joined (known-and-joined plus unknown), never a known registrant who hasn't
   * joined yet.
   */
  async getGeneralKpis(session: ProgramSession): Promise<ZoomGeneralAttendeeKpis> {
    const onlineSession = session.onlineSession;
    if (!onlineSession) {
      return {
        sessionId: session.id,
        totalGeneralAttendees: 0,
        knownAttendees: 0,
        unknownAttendees: 0,
        generalLoggedIn: 0,
        generalJoinedLate: 0,
        generalDropped: 0,
        generalRejoined: 0,
      };
    }

    const [events, generatedLinks, summary] = await Promise.all([
      this.liveEventRepository.findAllByOnlineSessionId(onlineSession.id),
      this.generatedLinkRepository.findActiveMapBySession(session.id),
      this.sessionSummaryRepository.findBySessionId(session.id),
    ]);

    const sessionStart = session.startsAt ?? null;
    const sessionEnd = this.resolveKnownSessionEnd(session, summary, sessionStart);
    const aggregates = this.aggregateLiveEvents(
      events,
      this.resolveDurationAsOf(session, summary, sessionStart),
      sessionStart,
    );
    const lateThresholdMs = this.config.getLateJoinThresholdSeconds() * 1000;
    const dropoffGraceMs = ZOOM_ANALYTICS_DEFAULTS.DROPOFF_GRACE_SECONDS * 1000;

    let unknownAttendees = 0;
    let knownJoined = 0;
    let joinedLate = 0;
    let dropped = 0;
    let rejoined = 0;

    for (const [normalizedEmail, aggregate] of this.selectGeneralAttendeeAggregates(aggregates)) {
      if (generatedLinks.has(normalizedEmail)) {
        knownJoined++;
      } else {
        unknownAttendees++;
      }
      if (
        sessionStart &&
        aggregate.firstJoinAt &&
        aggregate.firstJoinAt.getTime() - sessionStart.getTime() > lateThresholdMs
      ) {
        joinedLate++;
      }
      // Same session-end grace-window rule as getKpis' seekersDropped — see that comment.
      const isDropped =
        !aggregate.isCurrentlyJoined &&
        (!sessionEnd || !aggregate.lastLeaveAt || sessionEnd.getTime() - aggregate.lastLeaveAt.getTime() > dropoffGraceMs);
      if (isDropped) dropped++;
      if (aggregate.rejoinTransitionCount > 0) rejoined++;
    }

    const knownAttendees = generatedLinks.size;

    return {
      sessionId: session.id,
      totalGeneralAttendees: knownAttendees + unknownAttendees,
      knownAttendees,
      unknownAttendees,
      generalLoggedIn: knownJoined + unknownAttendees,
      generalJoinedLate: joinedLate,
      generalDropped: dropped,
      generalRejoined: rejoined,
    };
  }

  /**
   * Always fresh from `zoom_analytics_live_event` — syncs every roster
   * member's attendee-summary row from the live event log before reading, so
   * the seeker table doesn't require an explicit `reconcile()` call first.
   * Pagination/search still runs at the DB layer via the summary table,
   * which this sync keeps as a live-derived cache rather than a
   * reconcile-only snapshot.
   */
  async getAttendeeTable(
    session: ProgramSession,
    query: ZoomAttendeeQuery,
    rmContactId?: number,
  ): Promise<PaginatedZoomAttendeeRows> {
    await this.syncAttendeeSummaries(session, rmContactId);
    // Only resolved when "joined late" is needed — either the kpiFilter tile itself, or the
    // attendanceOutcome side filter's own joinedLate option. `undefined`, not `null`, when not
    // applicable — the filter object is spread verbatim into the repository call, and an explicit
    // `null` would show up there even when no v1 param was passed.
    const needsLateCutoff =
      query.kpiFilter === SessionKpiFilter.JOINED_LATE || query.attendanceOutcome?.includes(SessionKpiFilter.JOINED_LATE);
    const lateCutoff = needsLateCutoff ? this.resolveLateCutoff(session) : undefined;
    // Only resolved when "dropped" is actually needed — same lazy pattern as lateCutoff above. A
    // seeker whose last leave falls before this instant (the session's resolved end minus the
    // configured grace period) is a real drop-off; one who left at/after it just left when the
    // session naturally ended — see `ZoomAnalyticsAttendeeSummaryRepository.applyKpiFilter`'s DROPPED case.
    const needsDropoffCutoff =
      query.kpiFilter === SessionKpiFilter.DROPPED || query.attendanceOutcome?.includes(SessionKpiFilter.DROPPED);
    const dropoffCutoff =
      needsDropoffCutoff && session.onlineSession ? await this.resolveDropoffCutoff(session) : undefined;

    // rmAttendance/coordinatorAttendance/finalAttendance live in program_user_attendance (a different
    // table, resolved via the precedence engine), not as columns here — so when one of those v1
    // filters is active, manual marks must be fetched FIRST and reduced to the matching registration
    // ids, which then narrow the paginated DB query itself (otherwise a filter like
    // "finalAttendance=absent" would only ever narrow the current page's 20 rows, not the true
    // total/pagination). `systemAttendance` is excluded from this check on purpose — it's a plain
    // `is_system_attended` column on THIS table, filtered directly by the repository, so it never
    // needs the manual-marks detour. No qualifying filter → back to the original parallel fetch, so
    // the common (v0) path pays no extra sequencing cost.
    // JOINED/NOT_JOINED (the Present/Absent tiles) are FINAL-attendance-based, not raw
    // is_system_attended — same precedence as `getKpis`' own `finalPresent`/`finalAbsent`, so the
    // rows behind a clicked tile always match that tile's count. Excluded when `generalOnly`: a
    // general row reuses `SessionKpiFilter.JOINED` for an unrelated tile ("Logged In" — see
    // `buildGeneralAttendanceTiles`) that IS raw Zoom-join based, and has no registration to resolve
    // final attendance against in the first place.
    const needsFinalPresenceFilter =
      !query.generalOnly && (query.kpiFilter === SessionKpiFilter.JOINED || query.kpiFilter === SessionKpiFilter.NOT_JOINED);
    const hasSourceStateFilter =
      query.rmAttendance !== undefined || query.coordinatorAttendance !== undefined || query.finalAttendance !== undefined;
    // The "Present" and "Currently active" tiles (JOINED/ACTIVE) sort by join time — every other
    // tile/tab (including NOT_JOINED, which never has a join time at all) sorts by registration
    // date. Captured from the query's ORIGINAL kpiFilter, before the JOINED/NOT_JOINED branch below
    // blanks it to undefined for the repository's own narrowing (that branch resolves rows via
    // registrationIds instead).
    const sortByJoinTime =
      query.kpiFilter === SessionKpiFilter.JOINED || query.kpiFilter === SessionKpiFilter.ACTIVE ? true : undefined;
    let dataResult: { data: ZoomAnalyticsAttendeeSummary[]; total: number };
    let manualMarks: RegistrationManualMarks[];
    if (needsFinalPresenceFilter) {
      const onlineSessionId = session.onlineSession?.id;
      const [finalPresentRegistrationIds, roster, marks] = await Promise.all([
        this.resolveFinalPresentRegistrationIds(session.id),
        onlineSessionId ? this.fetchRoster(onlineSessionId, rmContactId) : Promise.resolve([]),
        this.onlineAttendanceService.getManualMarksBySession(session.id),
      ]);
      manualMarks = marks;
      const wantsPresent = query.kpiFilter === SessionKpiFilter.JOINED;
      const registrationIds = roster
        .filter((seeker) => finalPresentRegistrationIds.has(seeker.registrationId) === wantsPresent)
        .map((seeker) => seeker.registrationId);

      // TEMP DEBUG — remove once the notJoined/joined final-presence filter is verified.
      this.logger.log('TEMP DEBUG: getAttendeeTable final-presence filter', {
        sessionId: session.id,
        onlineSessionId,
        kpiFilter: query.kpiFilter,
        wantsPresent,
        rosterRegistrationIds: roster.map((seeker) => seeker.registrationId),
        finalPresentRegistrationIds: [...finalPresentRegistrationIds],
        resolvedRegistrationIds: registrationIds,
      });

      dataResult = await this.attendeeSummaryRepository.listBySession(session.id, {
        ...query,
        // Resolved via registrationIds above instead — bypasses the repository's own raw
        // is_system_attended-based JOINED/NOT_JOINED case (see `applyKpiFilter`).
        kpiFilter: undefined,
        rmContactId,
        lateCutoff,
        dropoffCutoff,
        registrationIds,
        sortByJoinTime,
      });
    } else if (hasSourceStateFilter) {
      manualMarks = await this.onlineAttendanceService.getManualMarksBySession(session.id);
      const registrationIds = manualMarks
        .filter((m) => this.matchesSourceStateFilter(m, query))
        .map((m) => m.registrationId);
      dataResult = await this.attendeeSummaryRepository.listBySession(session.id, {
        ...query,
        rmContactId,
        lateCutoff,
        dropoffCutoff,
        registrationIds,
        sortByJoinTime,
      });
    } else {
      [dataResult, manualMarks] = await Promise.all([
        this.attendeeSummaryRepository.listBySession(session.id, {
          ...query,
          rmContactId,
          lateCutoff,
          dropoffCutoff,
          sortByJoinTime,
        }),
        this.onlineAttendanceService.getManualMarksBySession(session.id),
      ]);
    }
    const { data, total } = dataResult;
    const manualMarksByRegistrationId = new Map(manualMarks.map((m) => [m.registrationId, m]));

    const registrationIds = data
      .map((row) => row.registrationId)
      .filter((id): id is number => id != null);
    const onlineSessionId = session.onlineSession?.id;
    const hasRegistrationRows = registrationIds.length > 0;
    const [
      joinUrlsByRegistrationId,
      activationStatusByRegistrationId,
      profileImagesByRegistrationId,
      rmContactsByRegistrationId,
    ] = await Promise.all([
      hasRegistrationRows && onlineSessionId
        ? this.registrationRepository.findActiveExtensionJoinUrls(registrationIds, onlineSessionId)
        : Promise.resolve(
            new Map<
              string,
              { joinUrl: string | null; activationStatus: RegistrationOnlineSessionActivationStatus | null }
            >(),
          ),
      // activationStatus on the attendees table is the registration-level rollup
      // (hdb_program_registration.activation_status), not any single session's extension row.
      hasRegistrationRows
        ? this.registrationRepository.findActivationStatusesByIds(registrationIds)
        : Promise.resolve(new Map<string, RegistrationOnlineSessionActivationStatus | null>()),
      hasRegistrationRows && onlineSessionId
        ? this.registrationRepository.findProfileImagesByIds(registrationIds)
        : Promise.resolve(new Map<string, string | null>()),
      hasRegistrationRows
        ? this.registrationRepository.findRmContactsByIds(registrationIds)
        : Promise.resolve(new Map<string, string | null>()),
    ]);
    // Only fetched when the page actually has general (registrationId null) rows — a known general
    // attendee's joinUrl/sourceTag are resolved the same way joinUrlsByRegistrationId resolves a
    // seeker's, by matching THIS session's active zoom_generated_registrant_link rows by email.
    const generatedLinksByEmail = data.some((row) => row.registrationId === null)
      ? await this.generatedLinkRepository.findActiveMapBySession(session.id)
      : new Map<string, ZoomGeneratedRegistrantLink>();

    return {
      data: data.map((row): ZoomAttendeeRow => {
        const marks = row.registrationId
          ? manualMarksByRegistrationId.get(row.registrationId)
          : undefined;
        const generatedLink =
          row.registrationId === null ? generatedLinksByEmail.get(this.normalizeEmail(row.email) ?? '') : undefined;
        return {
          attendeeId: row.id,
          userId: row.userId,
          registrationId: row.registrationId,
          // A general row's own `fullName` column is only ever set by `notJoinedGeneralAttendeePayload`
          // (a known registrant who hasn't joined yet) — everyone else on this row shape has it null, so
          // fall back to the matched generated link's `displayName` (a known registrant who HAS joined),
          // then to Zoom's own reported name (a true unmatched "Unidentified" walk-in). A registered
          // seeker's `fullName` is always sourced from their registration and never falls back.
          fullName:
            row.registrationId === null
              ? row.fullName ?? generatedLink?.displayName ?? row.zoomDisplayName
              : row.fullName,
          zoomDisplayName: row.zoomDisplayName,
          // Zoom's own report only ever knows the uniquely-tagged address we registered a general
          // attendee with (e.g. "vinod+gen9562p1474@yopmail.com"), never their real one, and never
          // collects a phone number for a registrant at all — so for a known general row, prefer
          // the matched generated link's real sourceEmail/sourceMobile over Zoom's row data.
          email: row.registrationId === null ? generatedLink?.sourceEmail ?? row.email : row.email,
          mobile: row.registrationId === null ? generatedLink?.sourceMobile ?? row.mobile : row.mobile,
          // General rows (registrationId null) have no registration, and therefore no RM, to resolve this from.
          rmContact: row.registrationId ? rmContactsByRegistrationId.get(String(row.registrationId)) ?? null : null,
          // General rows (registrationId null) have no registration to source a photo from.
          profileImage: row.registrationId
            ? profileImagesByRegistrationId.get(String(row.registrationId)) ?? null
            : null,
          joinedAt: row.joinedAt,
          noOfDevices: row.noOfDevices,
          dropoffCount: row.dropoffCount,
          rejoinCount: row.rejoinCount,
          lastDropoffAt: row.lastDropoffAt,
          lastRejoinedAt: row.lastRejoinedAt,
          durationSeconds: row.durationSeconds,
          preSessionDurationSeconds: row.preSessionDurationSeconds,
          attendance: {
            system: row.isSystemAttended,
            // RM/Coordinator marks live in program_user_attendance (the one canonical attendance
            // table), not on this row — null when that source never marked this registrant.
            rm: marks?.rm ?? null,
            coordinator: marks?.coordinator ?? null,
            // The resolved effective status (precedence engine) — the seeker table's "Final" column.
            final: this.booleanFromStatus(marks?.attendanceStatus),
          },
          joinUrl: row.registrationId
            ? joinUrlsByRegistrationId.get(String(row.registrationId))?.joinUrl ?? null
            : generatedLink?.joinUrl ?? null,
          activationStatus: row.registrationId
            ? activationStatusByRegistrationId.get(String(row.registrationId)) ?? null
            : null,
          sourceTag: generatedLink
            ? {
                sourceType: generatedLink.sourceType,
                label: generatedLink.roleKey,
              }
            : null,
          generatedLinkId: generatedLink ? Number(generatedLink.id) : null,
        };
      }),
      total,
      page: query.page,
      limit: query.limit,
    };
  }

  async getLiveStatus(session: ProgramSession): Promise<ZoomLiveStatus> {
    const onlineSession = session.onlineSession;
    const summary = await this.sessionSummaryRepository.findBySessionId(session.id);
    const isLive = !!summary?.actualStartAt && !summary?.reconciledAt;

    const events = onlineSession
      ? await this.liveEventRepository.findAllByOnlineSessionId(onlineSession.id)
      : [];
    const sessionStart = session.startsAt ?? null;
    const aggregates = this.aggregateLiveEvents(events, new Date(), sessionStart);

    let currentlyJoined = 0;
    let totalJoinedSoFar = 0;
    let rejoinsSoFar = 0;
    for (const aggregate of aggregates.values()) {
      if (aggregate.joinCount === 0) continue;
      totalJoinedSoFar++;
      if (aggregate.rejoinTransitionCount > 0) rejoinsSoFar++;
      if (aggregate.isCurrentlyJoined) currentlyJoined++;
    }
    const lastEventAt = events.reduce<Date | null>(
      (latest, e) => (!latest || e.occurredAt > latest ? e.occurredAt : latest),
      null,
    );

    return {
      sessionId: session.id,
      isLive,
      currentlyJoined,
      totalJoinedSoFar,
      rejoinsSoFar,
      lastEventAt,
    };
  }

  /**
   * The post-session analytics dashboard (attendance journey, drop-off
   * reasons, participation activity). Served entirely from locally stored
   * data — the live-event log, the session summary, and the participant-
   * device snapshot reconcile() persisted — Zoom is never called on a read.
   */
  async getDashboard(session: ProgramSession): Promise<ZoomSessionDashboard> {
    const onlineSession = session.onlineSession;
    const summaryRow = await this.sessionSummaryRepository.findBySessionId(session.id);
    if (!onlineSession) {
      return {
        sessionId: session.id,
        summary: this.emptyAttendanceSummary(),
        journey: [],
        dropOff: { total: 0, reasons: [], bands: [] },
        multipleLogins: { total: 0, buckets: [] },
        lateComers: { total: 0, buckets: [] },
        sincerityMap: this.computeSincerityMap([], new Map(), null, null),
        reconciledAt: summaryRow?.reconciledAt ?? null,
      };
    }

    const [roster, events, finalPresentRegistrationIds] = await Promise.all([
      this.fetchRoster(onlineSession.id),
      this.liveEventRepository.findAllByOnlineSessionId(onlineSession.id),
      // The same is_attended-backed source getKpis()'s finalPresent/finalAbsent reads —
      // Coordinator > RM > Zoom-system precedence, kept current on every mark/webhook/undo.
      // "Attended" below follows THIS, not the raw event log, so a manual RM/Coordinator
      // override is reflected everywhere on this dashboard, not just the KPI strip.
      this.resolveFinalPresentRegistrationIds(session.id),
    ]);

    const sessionStart = session.startsAt ?? null;
    const sessionEnd = this.resolveSessionEnd(session, summaryRow, sessionStart, events);
    const asOf = sessionEnd ?? new Date();

    const aggregates = this.aggregateLiveEvents(events, asOf, sessionStart);
    const aggregatesByRegistrationId = this.indexAggregatesByRegistrationId(aggregates);

    return {
      sessionId: session.id,
      summary: this.computeAttendanceSummary(
        roster,
        aggregatesByRegistrationId,
        finalPresentRegistrationIds,
        sessionEnd,
      ),
      journey: this.computeJourney(
        roster,
        aggregatesByRegistrationId,
        finalPresentRegistrationIds,
        sessionStart,
        asOf,
      ),
      dropOff: this.computeDropOffAnalysis(roster, aggregatesByRegistrationId, sessionEnd),
      multipleLogins: this.computeMultipleLogins(roster, aggregatesByRegistrationId),
      lateComers: this.computeLateComers(roster, aggregatesByRegistrationId, sessionStart),
      sincerityMap: this.computeSincerityMap(roster, aggregatesByRegistrationId, sessionStart, sessionEnd),
      reconciledAt: summaryRow?.reconciledAt ?? null,
    };
  }

  /**
   * The "Seekers Who Need Attention" follow-up table: roster members who never
   * joined, plus attendees with mid-session drop-offs. Locally stored data
   * only — same sources as getDashboard. Pagination/search run in memory: the
   * candidate set is one session's roster, and severity ordering (not-joined
   * before drop-offs) doesn't map to a single SQL sort.
   */
  async getFollowUps(
    session: ProgramSession,
    query: ZoomAttendeeQuery,
  ): Promise<PaginatedZoomFollowUpRows> {
    const onlineSession = session.onlineSession;
    if (!onlineSession) {
      return { data: [], total: 0, page: query.page, limit: query.limit };
    }

    const [roster, events, attendeeRows, summaryRow] = await Promise.all([
      this.fetchRoster(onlineSession.id),
      this.liveEventRepository.findAllByOnlineSessionId(onlineSession.id),
      this.attendeeSummaryRepository.findAllBySessionId(session.id),
      this.sessionSummaryRepository.findBySessionId(session.id),
    ]);

    const sessionStart = session.startsAt ?? null;
    const sessionEnd = this.resolveSessionEnd(session, summaryRow, sessionStart, events);
    const asOf = sessionEnd ?? new Date();

    const aggregates = this.aggregateLiveEvents(events, asOf, sessionStart);
    const aggregatesByRegistrationId = this.indexAggregatesByRegistrationId(aggregates);
    const attendeeIdByRegistrationId = new Map(
      attendeeRows
        .filter((row) => row.registrationId !== null)
        .map((row) => [Number(row.registrationId), row.id]),
    );
    const rows: ZoomFollowUpRow[] = [];
    for (const seeker of roster) {
      const aggregate = this.resolveSeekerAggregate(seeker, aggregatesByRegistrationId);
      const row = this.buildFollowUpRow(
        seeker,
        aggregate,
        attendeeIdByRegistrationId.get(seeker.registrationId) ?? null,
        sessionEnd,
      );
      if (row) rows.push(row);
    }

    const searched = this.filterFollowUpRows(rows, query.search);
    const start = (query.page - 1) * query.limit;
    return {
      data: searched.slice(start, start + query.limit),
      total: searched.length,
      page: query.page,
      limit: query.limit,
    };
  }

  /**
   * Same roster/event classification reconcile() uses for the session-level KPI counters, packaged
   * as the Final Attendance widget's stat pills. `attended` (and therefore `absent`) follows
   * `finalPresentRegistrationIds` — the Coordinator > RM > Zoom-system precedence — rather than the
   * raw event log, so a manual attendance override is reflected here too, not just the KPI strip.
   * Deliberately just present/absent, no separate "late comers" bucket: a late joiner ends up
   * finally present or finally absent like anyone else, so carving them out here would either
   * double-count them or require a third mutually-exclusive bucket that a manual override could
   * silently invalidate. "How late" is the Late Comers widget's own job (computeLateComers),
   * unconditionally on the raw event log — see its own doc comment for why.
   * `droppedOff`/`rejoined` are timing facts that only make sense for someone the event log
   * actually saw join, so those stay gated on the log AND on final presence.
   */
  private computeAttendanceSummary(
    roster: ZoomAnalyticsRosterEntry[],
    aggregatesByRegistrationId: Map<number, LiveEventAggregate>,
    finalPresentRegistrationIds: Set<number>,
    sessionEnd: Date | null,
  ): ZoomSessionAttendanceSummary {
    const dropoffGraceMs = ZOOM_ANALYTICS_DEFAULTS.DROPOFF_GRACE_SECONDS * 1000;

    let attended = 0;
    let droppedOff = 0;
    let rejoined = 0;

    for (const seeker of roster) {
      if (!finalPresentRegistrationIds.has(seeker.registrationId)) continue;
      attended++;
      const aggregate = this.resolveSeekerAggregate(seeker, aggregatesByRegistrationId);
      if (!aggregate || aggregate.joinCount === 0) continue;
      const isDropped =
        !aggregate.isCurrentlyJoined &&
        (!sessionEnd ||
          !aggregate.lastLeaveAt ||
          sessionEnd.getTime() - aggregate.lastLeaveAt.getTime() > dropoffGraceMs);
      if (isDropped) droppedOff++;
      if (aggregate.rejoinTransitionCount > 0) rejoined++;
    }

    return {
      total: roster.length,
      attended,
      absent: Math.max(roster.length - attended, 0),
      droppedOff,
      rejoined,
    };
  }

  private emptyAttendanceSummary(): ZoomSessionAttendanceSummary {
    return { total: 0, attended: 0, absent: 0, droppedOff: 0, rejoined: 0 };
  }

  /**
   * Best-known instant the session ended. Zoom's own signals first (`actualEndAt` from the
   * ended-webhook, else start + the Report API's duration); when neither exists — both are
   * routinely missing on this account (no ended-webhook configured for meetings, Report API
   * scope-gated) — falls back to the later of the scheduled end and the last observed
   * join/leave event. Without this, downstream windows (journey buckets, drop-off grace)
   * would stretch to "now", rendering a days-wide chart where the session is one blip.
   *
   * Both Zoom-reported signals are capped at the scheduled `session.endsAt`: an admin routinely
   * tells attendees to leave at the scheduled time but only clicks "End meeting for all" (or lets
   * it time out) much later, so an uncapped `actualEndAt`/duration would stretch the drop-off
   * window into that dead air and misclassify attendees who left on time as dropped off late.
   */
  private resolveSessionEnd(
    session: ProgramSession,
    summaryRow: { actualEndAt?: Date | null; actualDurationMinutes?: number | null } | null,
    sessionStart: Date | null,
    events: { occurredAt: Date }[],
  ): Date | null {
    const capAtScheduledEnd = (candidate: Date): Date =>
      session.endsAt && candidate > session.endsAt ? session.endsAt : candidate;
    if (summaryRow?.actualEndAt) return capAtScheduledEnd(summaryRow.actualEndAt);
    if (sessionStart && summaryRow?.actualDurationMinutes) {
      return capAtScheduledEnd(new Date(sessionStart.getTime() + summaryRow.actualDurationMinutes * 60_000));
    }
    const lastEventAt = events.reduce<Date | null>(
      (latest, event) => (!latest || event.occurredAt > latest ? event.occurredAt : latest),
      null,
    );
    const candidates = [lastEventAt, session.endsAt ?? null].filter(
      (candidate): candidate is Date => !!candidate,
    );
    if (candidates.length === 0) return null;
    const rawEstimate = new Date(Math.max(...candidates.map((candidate) => candidate.getTime())));
    // The dashboard serves LIVE sessions too — a scheduled end still in the future must not
    // stretch the window past this instant (the journey would trail off into empty future
    // buckets); clamp to "now" so a live view always ends at the present moment.
    const estimate = rawEstimate.getTime() > Date.now() ? new Date() : rawEstimate;
    // A session can't end before it started — a bad estimate is worth less than none.
    return sessionStart && estimate.getTime() <= sessionStart.getTime() ? null : estimate;
  }

  /**
   * The session's end, trusting only genuine end-of-session signals — never the "last observed live
   * event" fallback `resolveSessionEnd` uses for chart windowing. That fallback is unsound for the
   * live `seekersDropped`/`generalDropped`/DROPPED-filter grace check below: for a seeker whose own
   * final leave happens to BE the last event in the whole log (the common case for an unreconciled
   * session with no scheduled end), it would resolve to their own leave instant, making
   * `sessionEnd - lastLeaveAt` collapse to ~0 and every real drop-off look like a natural end. So this
   * only trusts Zoom's actual end/duration (from a `reconcile()` pass) or the scheduled end, and
   * returns `null` — falling back to the plain "not currently connected" rule — until one exists.
   *
   * Same scheduled-end cap as `resolveSessionEnd`: an admin routinely ends the meeting well after
   * telling attendees to leave, so an uncapped `actualEndAt`/duration would stretch the grace window
   * into that dead air and let attendees who left on time slip past the DROPPED cutoff.
   */
  private resolveKnownSessionEnd(
    session: ProgramSession,
    summaryRow: { actualEndAt?: Date | null; actualDurationMinutes?: number | null } | null,
    sessionStart: Date | null,
  ): Date | null {
    const capAtScheduledEnd = (candidate: Date): Date =>
      session.endsAt && candidate > session.endsAt ? session.endsAt : candidate;
    if (summaryRow?.actualEndAt) return capAtScheduledEnd(summaryRow.actualEndAt);
    if (sessionStart && summaryRow?.actualDurationMinutes) {
      return capAtScheduledEnd(new Date(sessionStart.getTime() + summaryRow.actualDurationMinutes * 60_000));
    }
    return session.endsAt ?? null;
  }

  /**
   * The `asOf` cutoff `aggregateLiveEvents` uses to close a still-open presence interval — deliberately
   * NOT `resolveKnownSessionEnd`'s value (that one's for the dropoff-grace check, and caps a real
   * recorded end DOWN to the schedule). Duration needs the opposite bias: `actualEndAt` is written once
   * and never revisited (see `markEnded`), so if a meeting gets reconciled as ending early and someone
   * then rejoins (crash + restart, premature Zoom report), a stale early `actualEndAt` must not stay the
   * ceiling forever — it would zero out every minute of the rejoin instead of counting it. So this takes
   * whichever of the scheduled end and the real recorded end is LATER: the schedule acts as a floor when
   * the real end looks premature, and the real end wins outright when it genuinely overruns the
   * schedule. Still never allowed to stretch into the future for a session that's currently live.
   */
  private resolveDurationAsOf(
    session: ProgramSession,
    summaryRow: { actualEndAt?: Date | null; actualDurationMinutes?: number | null } | null,
    sessionStart: Date | null,
  ): Date {
    const scheduledEnd = session.endsAt ?? null;
    const actualEnd =
      summaryRow?.actualEndAt ??
      (sessionStart && summaryRow?.actualDurationMinutes
        ? new Date(sessionStart.getTime() + summaryRow.actualDurationMinutes * 60_000)
        : null);
    const candidate =
      actualEnd && scheduledEnd
        ? actualEnd.getTime() >= scheduledEnd.getTime()
          ? actualEnd
          : scheduledEnd
        : (actualEnd ?? scheduledEnd);
    const now = new Date();
    return candidate && candidate.getTime() < now.getTime() ? candidate : now;
  }

  /**
   * Buckets the session's real activity window into adaptive steps.
   *
   * Window: starts a fixed JOURNEY_LEAD_IN_MINUTES before the session's own
   * recorded start (so early joiners' arrival is always visible on the
   * chart), through the resolved end.
   *
   * Step width: the smallest of 1/2/5/10/15/30/60 minutes that keeps the
   * chart at or under JOURNEY_TARGET_BUCKETS points — a 10-minute session
   * gets 1-minute buckets instead of two useless 15-minute samples.
   *
   * `attended` counts seekers present at ANY point inside the bucket's
   * interval (not sampled at an instant, which misses short stays entirely);
   * `absent` is the roster minus everyone who has joined by the bucket's end;
   * `dropOff`/`rejoins`/`lateComers` count the events inside each bucket.
   *
   * Every figure here follows `finalPresentRegistrationIds` — the Coordinator > RM >
   * Zoom-system precedence — not the raw event log alone:
   *  - A seeker whose final attendance is ABSENT never counts as attended at any point on
   *    the chart, even if the event log shows them joining — a manual override supersedes
   *    the webhook data entirely, so a since-corrected "present" reading can't linger here.
   *  - A seeker whose final attendance is PRESENT but who has no real join event at all is
   *    still plotted as present for the WHOLE session, using the session's own start as
   *    their synthesized join instant — there's no better instant to attribute a manual
   *    mark to, since the override itself carries no timestamp.
   *  - A seeker whose final attendance is PRESENT and who DID actually join keeps their
   *    real event-log presence intervals/timing untouched (the override just confirms what
   *    already happened).
   */
  private computeJourney(
    roster: ZoomAnalyticsRosterEntry[],
    aggregatesByRegistrationId: Map<number, LiveEventAggregate>,
    finalPresentRegistrationIds: Set<number>,
    sessionStart: Date | null,
    sessionEnd: Date,
  ): ZoomJourneyPoint[] {
    if (!sessionStart) return [];

    const finalPresentRoster = roster.filter((seeker) =>
      finalPresentRegistrationIds.has(seeker.registrationId),
    );
    const seekerAggregates: LiveEventAggregate[] = [];
    for (const seeker of finalPresentRoster) {
      const aggregate = this.resolveSeekerAggregate(seeker, aggregatesByRegistrationId);
      if (aggregate && aggregate.joinCount > 0) {
        seekerAggregates.push(aggregate);
      } else {
        seekerAggregates.push(this.syntheticFinalPresentAggregate(sessionStart, sessionEnd));
      }
    }

    const windowStartMs =
      sessionStart.getTime() - ZOOM_ANALYTICS_DEFAULTS.JOURNEY_LEAD_IN_MINUTES * 60_000;
    if (sessionEnd.getTime() <= windowStartMs) return [];

    const windowMs = sessionEnd.getTime() - windowStartMs;
    const stepMinutes =
      ZOOM_ANALYTICS_DEFAULTS.JOURNEY_STEP_LADDER_MINUTES.find(
        (minutes) => windowMs / (minutes * 60_000) <= ZOOM_ANALYTICS_DEFAULTS.JOURNEY_TARGET_BUCKETS,
      ) ?? ZOOM_ANALYTICS_DEFAULTS.JOURNEY_STEP_LADDER_MINUTES.at(-1)!;
    const stepMs = stepMinutes * 60_000;
    const stepCount = Math.min(
      Math.max(1, Math.ceil(windowMs / stepMs)),
      ZOOM_ANALYTICS_DEFAULTS.JOURNEY_MAX_BUCKETS,
    );
    const lateThresholdMs = this.config.getLateJoinThresholdSeconds() * 1000;

    const points: ZoomJourneyPoint[] = [];
    for (let step = 0; step <= stepCount; step++) {
      const bucketAtMs = windowStartMs + step * stepMs;
      // The first bucket swallows anything at/before the window start, so no event can
      // fall off the front of the chart.
      const bucketStartMs = step === 0 ? Number.NEGATIVE_INFINITY : bucketAtMs - stepMs;
      const inBucket = (at: Date) => at.getTime() > bucketStartMs && at.getTime() <= bucketAtMs;

      let attended = 0;
      let everJoined = 0;
      let dropOff = 0;
      let rejoins = 0;
      let lateComers = 0;
      for (const aggregate of seekerAggregates) {
        // Present during the bucket = any presence interval overlapping (bucketStart, bucketAt].
        if (
          aggregate.presenceIntervals.some(
            (interval) =>
              interval.start.getTime() <= bucketAtMs && interval.end.getTime() > bucketStartMs,
          )
        ) {
          attended++;
        }
        if (aggregate.firstJoinAt && aggregate.firstJoinAt.getTime() <= bucketAtMs) {
          everJoined++;
          if (
            inBucket(aggregate.firstJoinAt) &&
            // Late is measured against the RECORDED start, not the widened window start.
            aggregate.firstJoinAt.getTime() - sessionStart.getTime() > lateThresholdMs
          ) {
            lateComers++;
          }
        }
        dropOff += this.midSessionDropoffs(aggregate, null).filter((drop) => inBucket(drop.at)).length;
        rejoins += aggregate.rejoinTimes.filter(inBucket).length;
      }

      points.push({
        bucketAt: new Date(bucketAtMs),
        attended,
        absent: Math.max(roster.length - everJoined, 0),
        dropOff,
        rejoins,
        lateComers,
      });
    }
    return points;
  }

  /**
   * The seeker's REAL mid-session drop-offs: full drops before the end-grace
   * cutoff whose closing leave reason is NOT a host-ended one. When the host
   * ends/closes the meeting, everyone disconnects at once — that's the session
   * ending, not that seeker dropping, so it must never count as a drop-off
   * (the end-grace window alone can't catch it when a host ends a session
   * mid-way through its scheduled slot).
   */
  private midSessionDropoffs(
    aggregate: LiveEventAggregate,
    cutoffMs: number | null,
  ): { at: Date; reason: string | null }[] {
    return aggregate.dropoffEvents.filter(
      (drop) =>
        (cutoffMs === null || drop.at.getTime() < cutoffMs) &&
        (drop.reason === null ||
          this.normalizeDropOffReason(drop.reason) !== ZOOM_ANALYTICS_DROP_OFF_REASONS.HOST_ENDED),
    );
  }

  /**
   * The Drop-off widget: seekers bucketed by how many times they fully dropped
   * mid-session ("1 drop-off" / "2 drop-offs" / "3+ drop-offs"), each band
   * stacked by the seeker's last real drop's OWN leave reason (normalized —
   * 'Unknown' when Zoom sent none). Host-ended disconnects and drops inside
   * the end-grace window are the session ending, not seeker drop-offs — see
   * midSessionDropoffs (the follow-up table uses the same rule).
   */
  private computeDropOffAnalysis(
    roster: ZoomAnalyticsRosterEntry[],
    aggregatesByRegistrationId: Map<number, LiveEventAggregate>,
    sessionEnd: Date | null,
  ): ZoomDropOffAnalysis {
    const cutoffMs = sessionEnd
      ? sessionEnd.getTime() - ZOOM_ANALYTICS_DEFAULTS.DROPOFF_GRACE_SECONDS * 1000
      : null;

    let total = 0;
    const bandReasonCounts = new Map<string, Map<string, number>>();
    for (const seeker of roster) {
      const aggregate = this.resolveSeekerAggregate(seeker, aggregatesByRegistrationId);
      if (!aggregate || aggregate.joinCount === 0) continue;
      const drops = this.midSessionDropoffs(aggregate, cutoffMs);
      if (drops.length === 0) continue;

      total += drops.length;
      const band = drops.length >= 3 ? '3+ drop-offs' : drops.length === 2 ? '2 drop-offs' : '1 drop-off';
      // The seeker's segment carries their LAST real drop's own reason.
      const lastReason = drops[drops.length - 1].reason;
      const reason = lastReason
        ? this.normalizeDropOffReason(lastReason)
        : ZOOM_ANALYTICS_DROP_OFF_REASONS.UNKNOWN;

      const reasonCounts = bandReasonCounts.get(band) ?? new Map<string, number>();
      reasonCounts.set(reason, (reasonCounts.get(reason) ?? 0) + 1);
      bandReasonCounts.set(band, reasonCounts);
    }

    // Legend order: the known families first, then any dynamic reasons, Unknown last.
    const present = new Set<string>();
    for (const reasonCounts of bandReasonCounts.values()) {
      for (const reason of reasonCounts.keys()) present.add(reason);
    }
    const canonical: string[] = [
      ZOOM_ANALYTICS_DROP_OFF_REASONS.USER_LEFT,
      ZOOM_ANALYTICS_DROP_OFF_REASONS.REMOVED_BY_HOST,
      ZOOM_ANALYTICS_DROP_OFF_REASONS.NETWORK_ISSUE,
      ZOOM_ANALYTICS_DROP_OFF_REASONS.HOST_ENDED,
    ];
    const reasons = [
      ...canonical.filter((reason) => present.has(reason)),
      ...Array.from(present)
        .filter(
          (reason) =>
            !canonical.includes(reason) && reason !== ZOOM_ANALYTICS_DROP_OFF_REASONS.UNKNOWN,
        )
        .sort(),
      ...(present.has(ZOOM_ANALYTICS_DROP_OFF_REASONS.UNKNOWN)
        ? [ZOOM_ANALYTICS_DROP_OFF_REASONS.UNKNOWN]
        : []),
    ];

    const bands = ['1 drop-off', '2 drop-offs', '3+ drop-offs']
      .filter((label) => bandReasonCounts.has(label))
      .map((label) => ({
        label,
        segments: reasons
          .filter((reason) => (bandReasonCounts.get(label)!.get(reason) ?? 0) > 0)
          .map((reason) => ({ reason, count: bandReasonCounts.get(label)!.get(reason)! })),
      }));

    return { total, reasons, bands };
  }

  /**
   * The Multiple Logins widget: how many seekers joined from more than one
   * device/connection, bucketed as "2 devices" / "3 devices" / "4 devices" /
   * "5+ devices". `deviceCount` is the live-event aggregate's best-known
   * figure — the larger of distinct zoom_participant_ids seen and the peak
   * number of concurrently open connections (see LiveEventAggregate). Only
   * buckets that actually occur are returned, largest device-count last.
   */
  private computeMultipleLogins(
    roster: ZoomAnalyticsRosterEntry[],
    aggregatesByRegistrationId: Map<number, LiveEventAggregate>,
  ): ZoomMultipleLoginsBreakdown {
    const bucketCounts = new Map<string, number>();
    let total = 0;
    for (const seeker of roster) {
      const aggregate = this.resolveSeekerAggregate(seeker, aggregatesByRegistrationId);
      const devices = aggregate?.deviceCount ?? 0;
      if (devices < 2) continue;
      total++;
      const label = devices >= 5 ? '5+ devices' : `${devices} devices`;
      bucketCounts.set(label, (bucketCounts.get(label) ?? 0) + 1);
    }
    const bucketOrder = ['2 devices', '3 devices', '4 devices', '5+ devices'];
    return {
      total,
      buckets: bucketOrder
        .filter((label) => (bucketCounts.get(label) ?? 0) > 0)
        .map((label) => ({
          label,
          count: bucketCounts.get(label)!,
          percentage: Math.round((bucketCounts.get(label)! / total) * 100),
        })),
    };
  }

  /**
   * The Late Comers widget: every seeker whose first join landed past the late-join
   * cutoff, how far PAST that cutoff it landed, bucketed to seconds/minutes granularity
   * (see ZOOM_ANALYTICS_LATE_COMER_BUCKETS). This is lateness relative to the CUTOFF, not
   * the raw distance from session start — the cutoff itself is already several minutes
   * past start by config, so measuring from start would put every late-comer in the
   * coarsest bucket and defeat the point of a fine-grained breakdown.
   *
   * Deliberately NOT gated on final attendance: whether a late joiner is later marked
   * finally present or finally absent doesn't change that they showed up late — that's a
   * fact about the join event itself, not something a manual attendance override can
   * retroactively erase. (Final Attendance's own present/absent split is where the
   * override actually matters.)
   */
  private computeLateComers(
    roster: ZoomAnalyticsRosterEntry[],
    aggregatesByRegistrationId: Map<number, LiveEventAggregate>,
    sessionStart: Date | null,
  ): ZoomLateComerAnalysis {
    if (!sessionStart) return { total: 0, buckets: [] };

    const lateCutoffMs = sessionStart.getTime() + this.config.getLateJoinThresholdSeconds() * 1000;
    const bucketCounts = new Map<string, number>();
    let total = 0;
    for (const seeker of roster) {
      const aggregate = this.resolveSeekerAggregate(seeker, aggregatesByRegistrationId);
      if (!aggregate?.firstJoinAt || aggregate.firstJoinAt.getTime() <= lateCutoffMs) continue;

      total++;
      const lateBeyondCutoffMs = aggregate.firstJoinAt.getTime() - lateCutoffMs;
      const bucket = ZOOM_ANALYTICS_LATE_COMER_BUCKETS.find((candidate) => lateBeyondCutoffMs < candidate.maxMs)!;
      bucketCounts.set(bucket.label, (bucketCounts.get(bucket.label) ?? 0) + 1);
    }

    return {
      total,
      // Every band shows, even at 0 — the widget is a fixed 5-row list, not a dynamic one.
      buckets: ZOOM_ANALYTICS_LATE_COMER_BUCKETS.map((bucket) => ({
        label: bucket.label,
        count: bucketCounts.get(bucket.label) ?? 0,
      })),
    };
  }

  /** Clamps a raw 0-100 score and maps it to one of ZOOM_SINCERITY_MAP_BAND_LABELS' 5 indices. */
  private sincerityBandIndex(score: number): number {
    const clamped = Math.min(100, Math.max(0, score));
    return Math.min(ZOOM_SINCERITY_MAP_BAND_LABELS.length - 1, Math.floor(clamped / 20));
  }

  /**
   * The Sincerity Map: every roster seeker plotted by punctuality (how on-time their
   * first join was) × endurance (how much of the session they actually stayed for) —
   * both scored 0-100 against THIS session's own duration, per the fixed-window model:
   *   lateness    = max(0, firstJoin − sessionStart)
   *   punctuality = 100 × (1 − lateness / sessionDurationMinutes)
   *   endurance   = 100 × presentMinutes / sessionDurationMinutes
   * A seeker who never joined at all scores 0 on both — same as being maximally late
   * and present for none of it, since there's nothing to distinguish that from a no-show.
   */
  private computeSincerityMap(
    roster: ZoomAnalyticsRosterEntry[],
    aggregatesByRegistrationId: Map<number, LiveEventAggregate>,
    sessionStart: Date | null,
    sessionEnd: Date | null,
  ): ZoomSincerityMap {
    const punctualityBandLabels = [...ZOOM_SINCERITY_MAP_BAND_LABELS].reverse();
    const enduranceBandLabels = [...ZOOM_SINCERITY_MAP_BAND_LABELS];
    const counts = punctualityBandLabels.map(() => enduranceBandLabels.map(() => 0));

    const sessionDurationMinutes = sessionStart && sessionEnd ? (sessionEnd.getTime() - sessionStart.getTime()) / 60_000 : 0;
    if (sessionDurationMinutes <= 0) {
      return { total: roster.length, punctualityBandLabels, enduranceBandLabels, counts };
    }

    for (const seeker of roster) {
      const aggregate = this.resolveSeekerAggregate(seeker, aggregatesByRegistrationId);
      const firstJoinAt = aggregate?.firstJoinAt ?? null;
      const punctuality = firstJoinAt
        ? 100 * (1 - Math.max(0, firstJoinAt.getTime() - sessionStart!.getTime()) / 60_000 / sessionDurationMinutes)
        : 0;
      const presentMinutes = (aggregate?.durationSeconds ?? 0) / 60;
      const endurance = (100 * presentMinutes) / sessionDurationMinutes;

      // punctualityBandLabels is high-to-low, so its index is the mirror of the raw band index.
      const punctualityRowIndex = punctualityBandLabels.length - 1 - this.sincerityBandIndex(punctuality);
      const enduranceColIndex = this.sincerityBandIndex(endurance);
      counts[punctualityRowIndex][enduranceColIndex]++;
    }

    return { total: roster.length, punctualityBandLabels, enduranceBandLabels, counts };
  }

  /**
   * Zoom's live-webhook `leave_reason` is a sentence shaped
   * `"<name> left the meeting. Reason : <detail>."` (observed on this account
   * 2026-07-16) — so classification must run on the DETAIL after "Reason :",
   * never the whole string (every value contains "left the meeting").
   * Known detail families map to the design's canonical labels; anything
   * unrecognized keeps its own (cleaned) detail text as a dynamic bucket
   * rather than collapsing into a lossy "Other".
   */
  private normalizeDropOffReason(reason: string): string {
    const reasonSeparatorMatch = /reason\s*:\s*(.+)$/i.exec(reason);
    const detail = (reasonSeparatorMatch?.[1] ?? reason).trim().replace(/\.+$/, '');
    const normalized = detail.toLowerCase();

    if (/remove|eject|kick/.test(normalized)) return ZOOM_ANALYTICS_DROP_OFF_REASONS.REMOVED_BY_HOST;
    if (/network|connection|disconnect|failover/.test(normalized))
      return ZOOM_ANALYTICS_DROP_OFF_REASONS.NETWORK_ISSUE;
    if (/host (closed|ended)|(closed|ended) the meeting/.test(normalized))
      return ZOOM_ANALYTICS_DROP_OFF_REASONS.HOST_ENDED;
    if (/left|leave/.test(normalized)) return ZOOM_ANALYTICS_DROP_OFF_REASONS.USER_LEFT;
    // Unrecognized detail — surface it as-is (first letter capitalized) so new Zoom
    // reason variants show up in the widget instead of vanishing into a catch-all.
    return detail.charAt(0).toUpperCase() + detail.slice(1);
  }

  /**
   * One seeker's follow-up row, or null when they don't need attention.
   * Includes: roster members who never joined; attendees with mid-session
   * drop-offs (a drop-off inside the grace window of the actual end is the
   * natural end-of-session disconnect and doesn't count — without a known
   * end, drop-offs that were followed by a rejoin are the provable mid-session
   * ones).
   */
  private buildFollowUpRow(
    seeker: ZoomAnalyticsRosterEntry,
    aggregate: LiveEventAggregate | undefined,
    attendeeId: number | null,
    sessionEnd: Date | null,
  ): ZoomFollowUpRow | null {
    const base = {
      attendeeId,
      registrationId: seeker.registrationId,
      userId: seeker.userId,
      fullName: seeker.fullName,
      email: seeker.email,
      mobile: seeker.mobile,
      rmName: seeker.rmName,
    };

    if (!aggregate || aggregate.joinCount === 0) {
      return {
        ...base,
        joinedAt: null,
        durationSeconds: null,
        deviceCount: null,
        reason: ZOOM_ANALYTICS_FOLLOW_UP.REASON_NOT_JOINED,
        engagement: 'LOW',
      };
    }

    const cutoffMs = sessionEnd
      ? sessionEnd.getTime() - ZOOM_ANALYTICS_DEFAULTS.DROPOFF_GRACE_SECONDS * 1000
      : null;
    const midSessionDropoffs = this.midSessionDropoffs(aggregate, cutoffMs).length;
    if (midSessionDropoffs === 0) return null;

    return {
      ...base,
      joinedAt: aggregate.firstJoinAt,
      durationSeconds: aggregate.durationSeconds,
      deviceCount: aggregate.deviceCount,
      reason:
        midSessionDropoffs >= 2
          ? ZOOM_ANALYTICS_FOLLOW_UP.REASON_FREQUENT_DROPOFFS
          : ZOOM_ANALYTICS_FOLLOW_UP.REASON_DROPPED_ONCE,
      engagement: midSessionDropoffs >= 2 ? 'MEDIUM' : 'LOW',
    };
  }

  /** Severity order (never-joined first, then frequent drop-offs, then single drop-offs), name-sorted within each band; optional name/email/mobile search. */
  private filterFollowUpRows(rows: ZoomFollowUpRow[], search?: string): ZoomFollowUpRow[] {
    const severity = (row: ZoomFollowUpRow): number => {
      if (row.reason === ZOOM_ANALYTICS_FOLLOW_UP.REASON_NOT_JOINED) return 0;
      if (row.reason === ZOOM_ANALYTICS_FOLLOW_UP.REASON_FREQUENT_DROPOFFS) return 1;
      return 2;
    };
    const needle = search?.trim().toLowerCase();
    const filtered = needle
      ? rows.filter((row) =>
          [row.fullName, row.email, row.mobile].some((value) =>
            value?.toLowerCase().includes(needle),
          ),
        )
      : rows;
    return filtered.sort(
      (a, b) =>
        severity(a) - severity(b) ||
        (a.fullName ?? '').localeCompare(b.fullName ?? '') ||
        a.registrationId - b.registrationId,
    );
  }

  async recordParticipantJoined(
    externalId: string,
    occurrenceId: string | null,
    email: string,
    occurredAt: Date,
    participantId: string | null,
    zoomDisplayName: string | null,
    meetingStartTime?: Date | null,
  ): Promise<void> {
    const session = await this.webinarRepository.findByExtIdForOccurrence(
      externalId,
      occurrenceId,
      occurredAt,
      meetingStartTime,
    );
    if (!session?.onlineSession) {
      this.logger.warn(ZOOM_ANALYTICS_LOG.WEBHOOK_SESSION_NOT_FOUND, {
        event: 'participant_joined',
        externalId,
        occurrenceId,
      });
      return;
    }
    const match = await this.resolveRosterMatch(session.onlineSession.id, email);
    await this.insertLiveEvent(
      session,
      session.onlineSession.id,
      externalId,
      email,
      match?.userId ?? null,
      ZoomLiveEventType.JOINED,
      occurredAt,
      participantId,
      zoomDisplayName,
    );

    // Also lands in program_user_attendance's attendance_events log (the one canonical attendance
    // table) as a ZOOM_WEBHOOK mark — the precedence engine there decides whether it's the "active"
    // one alongside any RM/Coordinator/Admin/QR/join-click mark for this same registrant.
    if (match) {
      await this.onlineAttendanceService.recordZoomJoin({
        sessionId: session.id,
        programId: session.programId,
        registrationId: match.registrationId,
        userId: match.userId,
        fullName: match.fullName,
        email: match.email,
        mobile: match.mobile,
        occurredAt,
      });
    }
  }

  async recordParticipantLeft(
    externalId: string,
    occurrenceId: string | null,
    email: string,
    occurredAt: Date,
    participantId: string | null,
    zoomDisplayName: string | null,
    leaveReason: string | null,
    meetingStartTime?: Date | null,
  ): Promise<void> {
    const session = await this.webinarRepository.findByExtIdForOccurrence(
      externalId,
      occurrenceId,
      occurredAt,
      meetingStartTime,
    );
    if (!session?.onlineSession) {
      this.logger.warn(ZOOM_ANALYTICS_LOG.WEBHOOK_SESSION_NOT_FOUND, {
        event: 'participant_left',
        externalId,
        occurrenceId,
      });
      return;
    }
    const match = await this.resolveRosterMatch(session.onlineSession.id, email);
    await this.insertLiveEvent(
      session,
      session.onlineSession.id,
      externalId,
      email,
      match?.userId ?? null,
      ZoomLiveEventType.LEFT,
      occurredAt,
      participantId,
      zoomDisplayName,
      leaveReason,
    );
  }

  /**
   * Self-heal pass: reconciles our locally-tracked join/leave state against the
   * Dashboard API's current "who's present" snapshot, and separately refreshes
   * a present seeker's display name if Zoom now reports one that differs from
   * what we've recorded — the only way an in-session rename (no webhook exists
   * for it) can ever be observed. Resolved per roster seeker by registrationId
   * (like reconcile()/getKpis()), not by grouping the event log by raw email —
   * grouping by email would collapse sibling registrations that share one real
   * email (proxy/child regs) into a single correction, misattributing it via
   * first-match. A present email with no parseable `+reg<registrationId>` tag
   * can't be attributed to anyone and is simply ignored (matchRosterSeeker
   * requires the exact tag — see its doc).
   */
  async correctLiveState(
    session: ProgramSession,
    currentlyPresent: Map<string, string | null>,
  ): Promise<void> {
    const onlineSession = session.onlineSession;
    if (!onlineSession) return;
    const [events, roster] = await Promise.all([
      this.liveEventRepository.findAllByOnlineSessionId(onlineSession.id),
      this.fetchRoster(onlineSession.id),
    ]);
    // Only presence (isCurrentlyJoined/openConnectionCount) and latestDisplayName are read below —
    // duration isn't, so no session start is needed to split it correctly here.
    const aggregates = this.aggregateLiveEvents(events, new Date(), null);
    const aggregatesByRegistrationId = this.indexAggregatesByRegistrationId(aggregates);

    const presentByRegistrationId = new Map<number, string | null>();
    for (const [email, name] of currentlyPresent) {
      const match = this.matchRosterSeeker(email, roster);
      if (match) presentByRegistrationId.set(match.registrationId, name);
    }

    const now = new Date();
    for (const seeker of roster) {
      const aggregate = this.resolveSeekerAggregate(seeker, aggregatesByRegistrationId);
      const locallyJoined = !!aggregate?.isCurrentlyJoined;
      const dashboardSaysPresent = presentByRegistrationId.has(seeker.registrationId);

      // Tagged with this seeker's own registrationId so later aggregation always
      // attributes the correction to them, even if their real email is shared
      // with a sibling registration.
      const correctionEmail = seeker.email
        ? buildZoomRegistrantEmail(seeker.email, seeker.registrationId)
        : `unknown-reg${seeker.registrationId}`;

      if (locallyJoined !== dashboardSaysPresent) {
        // Forcing "present" only ever needs one JOIN correction (any nonzero open count is present).
        // Forcing "absent" needs one LEFT correction PER currently-open device — presence is only fully
        // cleared once every open connection is closed, and a single correction can't say "close all";
        // it can only close one (see aggregateLiveEvents' anonymous-leave handling).
        const correctionCount = locallyJoined
          ? Math.max(aggregate?.openConnectionCount ?? 1, 1)
          : 1;

        for (let i = 0; i < correctionCount; i++) {
          await this.insertLiveEvent(
            session,
            onlineSession.id,
            onlineSession.externalId ?? '',
            correctionEmail,
            seeker.userId,
            locallyJoined ? ZoomLiveEventType.LEFT : ZoomLiveEventType.JOINED,
            now,
            null,
          );
        }
      }

      // Name refresh — independent of the presence correction above, and checked even when presence
      // already matched: a still-joined seeker who renamed mid-session is exactly the case with no
      // join/leave event to carry the update.
      const reportedName = presentByRegistrationId.get(seeker.registrationId);
      if (reportedName && reportedName !== aggregate?.latestDisplayName) {
        await this.insertLiveEvent(
          session,
          onlineSession.id,
          onlineSession.externalId ?? '',
          correctionEmail,
          seeker.userId,
          ZoomLiveEventType.RENAMED,
          now,
          null,
          reportedName,
        );
      }
    }
  }

  /** Marks the session as having started, for getLiveStatus's isLive check — no reconciliation happens yet. */
  async markStarted(session: ProgramSession, startedAt: Date): Promise<void> {
    if (!session.onlineSession) return;
    const existing = await this.sessionSummaryRepository.findBySessionId(session.id);
    if (existing?.actualStartAt) return; // idempotent — only set once
    await this.sessionSummaryRepository.upsert({
      programId: session.programId,
      sessionId: session.id,
      onlineSessionId: session.onlineSession.id,
      resourceType: this.key,
      actualStartAt: startedAt,
    });
  }

  /**
   * Marks the session as ended, synchronously — independent of reconcile()'s
   * Report API lag, so getKpis can immediately switch totalSeekersJoined
   * from "currently joined" to the frozen "ever joined" figure. Also stamps
   * `hdb_online_session.actual_meeting_ends_at` — a separate, join-window-
   * facing timestamp consumed outside the analytics module (join-window
   * cutoff, attendance reports) — from this same `*.ended` webhook, since
   * that column previously had no writer at all. Both stamps are
   * independently idempotent (first webhook for this occurrence wins), so
   * one already being set never blocks the other from being written.
   */
  async markEnded(session: ProgramSession, endedAt: Date): Promise<void> {
    if (!session.onlineSession) return;

    const existingSummary = await this.sessionSummaryRepository.findBySessionId(session.id);
    if (!existingSummary?.actualEndAt) {
      await this.sessionSummaryRepository.upsert({
        programId: session.programId,
        sessionId: session.id,
        onlineSessionId: session.onlineSession.id,
        resourceType: this.key,
        actualEndAt: endedAt,
      });
    }

    if (!session.onlineSession.actualMeetingEndsAt) {
      session.onlineSession.actualMeetingEndsAt = endedAt;
      await this.webinarRepository.save(session);
    }
  }

  /** Resolves the roster user_id for a single webhook-reported email — null if they're not a registered seeker. */
  /** Resolves the roster entry (userId, registrationId, name, email, mobile) for a webhook-reported email — null if they're not a registered seeker. */
  private async resolveRosterMatch(
    onlineSessionId: number,
    email: string,
  ): Promise<ZoomAnalyticsRosterEntry | null> {
    const roster = await this.fetchRoster(onlineSessionId);
    const match = this.matchRosterSeeker(email, roster);
    if (!match) {
      this.logger.warn(ZOOM_ANALYTICS_LOG.WEBHOOK_PARTICIPANT_UNMATCHED, {
        onlineSessionId,
        email,
        rosterSize: roster.length,
      });
    }
    return match;
  }

  /**
   * Zoom reports back the derived `+reg<registrationId>` address (see
   * buildZoomRegistrantEmail) — the registration id is the only basis for
   * attribution. One real email can legitimately back several sibling
   * registrations (proxy/child regs), so an untagged or raw-email match would
   * risk guessing between them; an email with no parseable tag, or a tag that
   * doesn't match anyone on the current roster, is simply unmatched rather
   * than attributed by best effort.
   */
  private matchRosterSeeker(
    email: string,
    roster: ZoomAnalyticsRosterEntry[],
  ): ZoomAnalyticsRosterEntry | null {
    const registrationId = parseZoomRegistrantRegistrationId(email);
    if (registrationId === null) return null;
    return roster.find((seeker) => seeker.registrationId === registrationId) ?? null;
  }

  private async insertLiveEvent(
    session: ProgramSession,
    onlineSessionId: number,
    webinarExtId: string,
    email: string,
    userId: number | null,
    eventType: ZoomLiveEventType,
    occurredAt: Date,
    zoomParticipantId: string | null,
    zoomDisplayName: string | null = null,
    leaveReason: string | null = null,
  ): Promise<void> {
    const row = this.liveEventRepository.create({
      programId: session.programId,
      sessionId: session.id,
      onlineSessionId,
      webinarExtId,
      email,
      userId,
      eventType,
      occurredAt,
      zoomParticipantId,
      zoomDisplayName,
      leaveReason,
    });
    await this.liveEventRepository.insert(row);
  }

  /**
   * Re-syncs every roster member's attendee-summary row, plus every general attendee's, from the live
   * event log alone — the read-time counterpart to `reconcile()`'s two attendee loops, minus the Zoom
   * Report API / session-summary bookkeeping. Lets the seeker table (durationSeconds, dropoffCount,
   * rejoinCount, isSystemAttended, ...) stay live on every read without waiting on an explicit
   * `reconcile()` — an insert-only sync would freeze those figures at whatever they were on the row's
   * first read, which is exactly the "duration/dropped count never updates" bug this upserts to avoid.
   * Public (not just `getAttendeeTable`'s own internal first step) so a caller reading the
   * attendee-summary table directly — e.g. the program-wide overall-analytics aggregate — can also
   * force it live first, instead of risking a stale read for any session nobody has opened the
   * Attendees tab for recently (see ZoomAnalyticsProvider.syncAttendeeSummaries's own doc comment).
   */
  async syncAttendeeSummaries(session: ProgramSession, rmContactId?: number): Promise<void> {
    const onlineSession = session.onlineSession;
    if (!onlineSession) return;

    const [events, roster, generatedLinks, summary] = await Promise.all([
      this.liveEventRepository.findAllByOnlineSessionId(onlineSession.id),
      this.fetchRoster(onlineSession.id, rmContactId),
      this.generatedLinkRepository.findActiveMapBySession(session.id),
      this.sessionSummaryRepository.findBySessionId(session.id),
    ]);

    const sessionStart = session.startsAt ?? null;
    // Same trusted-signals-only resolution the DROPPED filter/KPI grace check uses (never the
    // last-observed-event fallback `resolveSessionEnd` uses for chart windowing) — see
    // `resolveKnownSessionEnd`'s own doc comment for why that fallback would be unsound here.
    const sessionEnd = this.resolveKnownSessionEnd(session, summary, sessionStart);
    const aggregates = this.aggregateLiveEvents(
      events,
      this.resolveDurationAsOf(session, summary, sessionStart),
      sessionStart,
    );
    const aggregatesByRegistrationId = this.indexAggregatesByRegistrationId(aggregates);

    const attendeePayloads: Array<
      Partial<ZoomAnalyticsAttendeeSummary> & {
        sessionId: number;
        registrationId: number | null;
        email: string;
      }
    > = roster.map((seeker) => {
      const aggregate = this.resolveSeekerAggregate(seeker, aggregatesByRegistrationId);
      return this.attendeeUpsertPayload(session, onlineSession.id, seeker, aggregate, sessionEnd);
    });
    attendeePayloads.push(
      ...this.buildGeneralAttendeePayloads(session, onlineSession.id, aggregates, generatedLinks, sessionEnd),
    );
    await this.attendeeSummaryRepository.upsertMany(attendeePayloads);
  }

  /**
   * The instant after which a first join counts as "late" — session's actual/scheduled start plus the
   * configured grace period, the same threshold `getKpis` uses for its own `seekersJoinedLate` count, so
   * the v1 `joinedLate` filter narrows the attendee table to exactly the rows behind that KPI. Null when
   * no start time is known yet (unreconciled session with no scheduled start either).
   */
  private resolveLateCutoff(session: ProgramSession): Date | undefined {
    const sessionStart = session.startsAt ?? null;
    if (!sessionStart) return undefined;
    return new Date(sessionStart.getTime() + this.config.getLateJoinThresholdSeconds() * 1000);
  }

  /**
   * The instant before which a seeker's `last_dropoff_at` must fall to count as a real drop-off —
   * the session's known end (`resolveKnownSessionEnd`, the same instant `getKpis`' `seekersDropped`
   * grace-window check uses) minus the configured grace period. `undefined` when the session end
   * can't be resolved at all (unreconciled live session, no scheduled end either) — the repository
   * falls back to the raw "currently not connected" rule in that case.
   */
  private async resolveDropoffCutoff(session: ProgramSession): Promise<Date | undefined> {
    const summary = await this.sessionSummaryRepository.findBySessionId(session.id);
    const sessionStart = session.startsAt ?? null;
    const sessionEnd = this.resolveKnownSessionEnd(session, summary, sessionStart);
    if (!sessionEnd) return undefined;
    return new Date(sessionEnd.getTime() - ZOOM_ANALYTICS_DEFAULTS.DROPOFF_GRACE_SECONDS * 1000);
  }

  /** True/false for a decided present/absent, null for "never marked" (UNKNOWN or no mark at all). */
  private booleanFromStatus(status: AttendanceStatus | undefined): boolean | null {
    if (status === AttendanceStatus.PRESENT) return true;
    if (status === AttendanceStatus.ABSENT) return false;
    return null;
  }

  /** The tri-state a source's own boolean|null mark corresponds to — PRESENT/ABSENT/UNKNOWN, matching AttendanceStatus. */
  private statusFromBoolean(mark: boolean | null): AttendanceStatus {
    if (mark === true) return AttendanceStatus.PRESENT;
    if (mark === false) return AttendanceStatus.ABSENT;
    return AttendanceStatus.UNKNOWN;
  }

  /**
   * Registration ids credited FINAL-attendance present, for this session — a direct
   * `is_attended` column read (`OnlineAttendanceService.getAttendedRegistrationIds`), not a fresh
   * precedence re-derivation: `is_attended` is already kept current on every mark/webhook/undo (see
   * that service's own `pushAndResolve`/`recomputeSummary`). Session-wide, not roster/RM-scoped —
   * callers intersect against their own roster/rmContactId scope themselves (a registrant with no
   * row at all here — never marked, never joined Zoom — is absent from this set, i.e. FINAL ABSENT,
   * same as `getKpis`' `finalAbsent`).
   */
  private async resolveFinalPresentRegistrationIds(sessionId: number): Promise<Set<number>> {
    return this.onlineAttendanceService.getAttendedRegistrationIds(sessionId);
  }

  /** AND-matches every v1 source-state filter that's actually set (rmAttendance/coordinatorAttendance/finalAttendance) against one registrant's manual marks. */
  private matchesSourceStateFilter(marks: RegistrationManualMarks, query: ZoomAttendeeQuery): boolean {
    if (query.rmAttendance !== undefined && this.statusFromBoolean(marks.rm) !== query.rmAttendance) return false;
    if (
      query.coordinatorAttendance !== undefined &&
      this.statusFromBoolean(marks.coordinator) !== query.coordinatorAttendance
    ) {
      return false;
    }
    if (query.finalAttendance !== undefined && marks.attendanceStatus !== query.finalAttendance) return false;
    return true;
  }

  /** Builds one seeker's attendee-summary upsert payload from their live-event aggregate, if any. */
  private attendeeUpsertPayload(
    session: ProgramSession,
    onlineSessionId: number,
    seeker: ZoomAnalyticsRosterEntry,
    aggregate: LiveEventAggregate | undefined,
    sessionEnd: Date | null,
  ): Partial<ZoomAnalyticsAttendeeSummary> & {
    sessionId: number;
    registrationId: number;
    email: string;
  } {
    const key = this.normalizeEmail(seeker.email);
    const base = {
      programId: session.programId,
      sessionId: session.id,
      onlineSessionId,
      userId: seeker.userId,
      registrationId: seeker.registrationId,
      fullName: seeker.fullName,
      mobile: seeker.mobile,
    };

    if (!aggregate || aggregate.joinCount === 0) {
      return {
        ...base,
        email: seeker.email ?? `unknown-${seeker.registrationId}`,
        zoomDisplayName: null,
        joinedAt: null,
        noOfDevices: 0,
        dropoffCount: 0,
        rejoinCount: 0,
        lastDropoffAt: null,
        lastRejoinedAt: null,
        durationSeconds: 0,
        preSessionDurationSeconds: 0,
        isSystemAttended: false,
      };
    }

    // Same "real drop-off" rule the Drop-off widget/Follow-up table already apply via
    // `midSessionDropoffs` — excludes drops past the end-grace cutoff and host-ended disconnects, so
    // the persisted dropoffCount/lastDropoffAt agree with the `dropped` outcome filter/KPI instead of
    // counting "the meeting ended while they were still connected" as a drop-off.
    const cutoffMs = sessionEnd
      ? sessionEnd.getTime() - ZOOM_ANALYTICS_DEFAULTS.DROPOFF_GRACE_SECONDS * 1000
      : null;
    const genuineDropoffs = this.midSessionDropoffs(aggregate, cutoffMs);
    const lastGenuineDropoffAt = genuineDropoffs.length
      ? genuineDropoffs[genuineDropoffs.length - 1].at
      : null;

    return {
      ...base,
      email: seeker.email ?? key ?? '',
      zoomDisplayName: aggregate.latestDisplayName,
      joinedAt: aggregate.firstJoinAt,
      noOfDevices: aggregate.deviceCount,
      dropoffCount: genuineDropoffs.length,
      // Counts presence transitions from fully-absent back to present, not raw joins — a second
      // concurrent device joining isn't a rejoin.
      rejoinCount: aggregate.rejoinTransitionCount,
      // Last real drop-off is shown whenever the seeker isn't currently in the call.
      lastDropoffAt: !aggregate.isCurrentlyJoined ? lastGenuineDropoffAt : null,
      lastRejoinedAt: aggregate.rejoinTransitionCount > 0 ? aggregate.lastRejoinAt : null,
      durationSeconds: aggregate.durationSeconds,
      preSessionDurationSeconds: aggregate.preSessionDurationSeconds,
      isSystemAttended: true,
    };
  }

  /** Indexes an aggregate map (keyed by raw Zoom email) by the registration id parsed out of each key, where parseable. */
  private indexAggregatesByRegistrationId(
    aggregates: Map<string, LiveEventAggregate>,
  ): Map<number, LiveEventAggregate> {
    const byRegistrationId = new Map<number, LiveEventAggregate>();
    for (const [email, aggregate] of aggregates) {
      const registrationId = parseZoomRegistrantRegistrationId(email);
      if (registrationId !== null) byRegistrationId.set(registrationId, aggregate);
    }
    return byRegistrationId;
  }

  /**
   * The complement of `indexAggregatesByRegistrationId` — every email with NO parseable
   * `+reg<registrationId>` tag, i.e. joined via the shared/common link rather than a personalized
   * registration link. These are "general" attendees: real join/leave activity that can't be
   * attributed to any registration, so they get their own summary row (registrationId null) instead
   * of being silently dropped.
   */
  private selectGeneralAttendeeAggregates(
    aggregates: Map<string, LiveEventAggregate>,
  ): Map<string, LiveEventAggregate> {
    const general = new Map<string, LiveEventAggregate>();
    for (const [email, aggregate] of aggregates) {
      if (parseZoomRegistrantRegistrationId(email) === null) general.set(email, aggregate);
    }
    return general;
  }

  /**
   * Builds one general attendee's attendee-summary upsert payload — the registrationId-less
   * counterpart to `attendeeUpsertPayload`. Unlike a roster seeker, a general attendee only exists in
   * the first place because they have real join activity (see `selectGeneralAttendeeAggregates`), so
   * there's no "never joined" branch to handle, and no registration to source fullName/mobile/userId
   * from.
   */
  private generalAttendeeUpsertPayload(
    session: ProgramSession,
    onlineSessionId: number,
    email: string,
    aggregate: LiveEventAggregate,
    sessionEnd: Date | null,
  ): Partial<ZoomAnalyticsAttendeeSummary> & {
    sessionId: number;
    registrationId: null;
    email: string;
  } {
    // Same "real drop-off" rule attendeeUpsertPayload uses — see its own comment.
    const cutoffMs = sessionEnd
      ? sessionEnd.getTime() - ZOOM_ANALYTICS_DEFAULTS.DROPOFF_GRACE_SECONDS * 1000
      : null;
    const genuineDropoffs = this.midSessionDropoffs(aggregate, cutoffMs);
    const lastGenuineDropoffAt = genuineDropoffs.length
      ? genuineDropoffs[genuineDropoffs.length - 1].at
      : null;

    return {
      programId: session.programId,
      sessionId: session.id,
      onlineSessionId,
      userId: null,
      registrationId: null,
      fullName: null,
      mobile: null,
      email,
      zoomDisplayName: aggregate.latestDisplayName,
      joinedAt: aggregate.firstJoinAt,
      noOfDevices: aggregate.deviceCount,
      dropoffCount: genuineDropoffs.length,
      rejoinCount: aggregate.rejoinTransitionCount,
      lastDropoffAt: !aggregate.isCurrentlyJoined ? lastGenuineDropoffAt : null,
      lastRejoinedAt: aggregate.rejoinTransitionCount > 0 ? aggregate.lastRejoinAt : null,
      durationSeconds: aggregate.durationSeconds,
      preSessionDurationSeconds: aggregate.preSessionDurationSeconds,
      isSystemAttended: true,
    };
  }

  /**
   * Builds every general-attendee (registrationId null) upsert payload for one sync/reconcile pass —
   * both "known" registrants pre-generated in `zoom_generated_registrant_link` (a role-matched staff
   * user or a placeholder batch slot — see that table's own doc) and "unknown" true walk-ins with no
   * matching record at all. `generatedLinks` is keyed by normalized `registrantEmail` (see
   * `ZoomGeneratedRegistrantLinkRepository.findActiveMapBySession`), the same normalization
   * `aggregateLiveEvents` already applies to every live-event email, so the two maps line up by key
   * without re-normalizing.
   *
   * Every known registrant gets a row regardless of whether they've joined yet — mirrors how a
   * roster seeker gets a "never joined" row (`attendeeUpsertPayload`'s no-aggregate branch) — so the
   * general-attendees list shows the full eligible set up front, not just people who've already
   * connected. A truly unknown walk-in, by contrast, only ever gets a row because they joined (same
   * as before this change — there's no "eligible" list for them to appear on ahead of time).
   */
  private buildGeneralAttendeePayloads(
    session: ProgramSession,
    onlineSessionId: number,
    aggregates: Map<string, LiveEventAggregate>,
    generatedLinks: Map<string, ZoomGeneratedRegistrantLink>,
    sessionEnd: Date | null,
  ): Array<Partial<ZoomAnalyticsAttendeeSummary> & { sessionId: number; registrationId: null; email: string }> {
    const generalAggregates = this.selectGeneralAttendeeAggregates(aggregates);
    const payloads: Array<
      Partial<ZoomAnalyticsAttendeeSummary> & { sessionId: number; registrationId: null; email: string }
    > = [];

    for (const [normalizedEmail, link] of generatedLinks) {
      const aggregate = generalAggregates.get(normalizedEmail);
      payloads.push(
        aggregate
          ? this.generalAttendeeUpsertPayload(session, onlineSessionId, link.registrantEmail, aggregate, sessionEnd)
          : this.notJoinedGeneralAttendeePayload(session, onlineSessionId, link),
      );
    }
    for (const [normalizedEmail, aggregate] of generalAggregates) {
      if (!generatedLinks.has(normalizedEmail)) {
        payloads.push(
          this.generalAttendeeUpsertPayload(session, onlineSessionId, normalizedEmail, aggregate, sessionEnd),
        );
      }
    }
    return payloads;
  }

  /**
   * A known (pre-generated) general attendee who hasn't joined yet — the general-attendee
   * counterpart to `attendeeUpsertPayload`'s no-aggregate ("never joined") branch. `fullName` is the
   * generated link's own `displayName` (its ROLE user's name, or the placeholder slot's label, e.g.
   * "Staff 3") since there's no registration to source it from.
   */
  private notJoinedGeneralAttendeePayload(
    session: ProgramSession,
    onlineSessionId: number,
    link: ZoomGeneratedRegistrantLink,
  ): Partial<ZoomAnalyticsAttendeeSummary> & { sessionId: number; registrationId: null; email: string } {
    return {
      programId: session.programId,
      sessionId: session.id,
      onlineSessionId,
      userId: link.userId ?? null,
      registrationId: null,
      fullName: link.displayName,
      mobile: null,
      email: link.registrantEmail,
      zoomDisplayName: null,
      joinedAt: null,
      noOfDevices: 0,
      dropoffCount: 0,
      rejoinCount: 0,
      lastDropoffAt: null,
      lastRejoinedAt: null,
      durationSeconds: 0,
      preSessionDurationSeconds: 0,
      isSystemAttended: false,
    };
  }

  /**
   * Finds a roster seeker's own aggregate, by registration id alone. Zoom
   * reports back the derived `+reg<registrationId>` address (see
   * buildZoomRegistrantEmail) — one real email can legitimately back several
   * sibling registrations (proxy/child regs), so an event with no parseable
   * tag can't be safely attributed to any specific seeker and simply doesn't
   * count toward anyone's join/attendance figures.
   */
  private resolveSeekerAggregate(
    seeker: ZoomAnalyticsRosterEntry,
    aggregatesByRegistrationId: Map<number, LiveEventAggregate>,
  ): LiveEventAggregate | undefined {
    return aggregatesByRegistrationId.get(seeker.registrationId);
  }

  /**
   * Stand-in aggregate for a seeker whose final attendance is PRESENT (an RM/Coordinator
   * override) but who has no real join event at all — the Journey chart plots them as
   * present for the entire session, using the session's own start as their join instant,
   * since a manual attendance mark carries no timestamp of its own to attribute one from.
   */
  private syntheticFinalPresentAggregate(sessionStart: Date, sessionEnd: Date): LiveEventAggregate {
    return {
      firstJoinAt: sessionStart,
      lastJoinAt: sessionStart,
      lastLeaveAt: null,
      isCurrentlyJoined: true,
      joinCount: 1,
      leaveCount: 0,
      durationSeconds: Math.max(0, (sessionEnd.getTime() - sessionStart.getTime()) / 1000),
      preSessionDurationSeconds: 0,
      latestDisplayName: null,
      deviceCount: 1,
      fullDropoffCount: 0,
      rejoinTransitionCount: 0,
      lastFullDropoffAt: null,
      lastRejoinAt: null,
      openConnectionCount: 1,
      dropoffEvents: [],
      rejoinTimes: [],
      presenceIntervals: [{ start: sessionStart, end: sessionEnd }],
    };
  }

  /**
   * Splits a presence interval at `sessionStart`, so time before the session actually started is
   * tracked separately from counted duration. Returns [preSessionSeconds, countedSeconds]. With no
   * known `sessionStart`, the whole interval counts (no basis to exclude anything).
   */
  private splitIntervalAtSessionStart(
    intervalStart: Date,
    intervalEnd: Date,
    sessionStart: Date | null,
  ): [number, number] {
    const totalSeconds = Math.max(0, (intervalEnd.getTime() - intervalStart.getTime()) / 1000);
    if (!sessionStart || sessionStart.getTime() <= intervalStart.getTime()) {
      return [0, totalSeconds];
    }
    if (sessionStart.getTime() >= intervalEnd.getTime()) {
      return [totalSeconds, 0];
    }
    const preSeconds = (sessionStart.getTime() - intervalStart.getTime()) / 1000;
    return [preSeconds, totalSeconds - preSeconds];
  }

  /** Rolls up the append-only live event log per email, as of `asOf` (closes any still-open join interval at that instant). `sessionStart` (the session's actual/scheduled start) splits each presence interval so pre-session presence is excluded from `durationSeconds` and reported separately. */
  private aggregateLiveEvents(
    events: LiveEventRecord[],
    asOf: Date,
    sessionStart: Date | null,
  ): Map<string, LiveEventAggregate> {
    const byEmail = new Map<string, LiveEventRecord[]>();
    for (const event of events) {
      const email = this.normalizeEmail(event.email);
      if (!email) continue;
      const list = byEmail.get(email);
      if (list) list.push(event);
      else byEmail.set(email, [event]);
    }

    const result = new Map<string, LiveEventAggregate>();
    for (const [email, emailEvents] of byEmail) {
      let firstJoinAt: Date | null = null;
      let lastJoinAt: Date | null = null;
      let lastLeaveAt: Date | null = null;
      let joinCount = 0;
      let leaveCount = 0;
      let durationSeconds = 0;
      let preSessionDurationSeconds = 0;
      let latestDisplayName: string | null = null;
      const deviceIds = new Set<string>();

      // Presence is tracked as the UNION across concurrently open devices — not a single interval —
      // so a second device joining while one is already connected doesn't register as a rejoin, and
      // the first of several open devices leaving doesn't register as a full drop-off. `openDevices`
      // holds currently-connected devices we can identify by zoomParticipantId; `anonymousOpenCount`
      // covers connections whose events didn't carry one (self-heal corrections, or legacy rows from
      // before leave events carried a participant id) — those can't be matched by id, so an anonymous
      // leave just closes one arbitrary open slot (prefers another anonymous one) as a best effort.
      const openDevices = new Set<string>();
      let anonymousOpenCount = 0;
      let presenceOpenSince: Date | null = null;
      let hasEverJoined = false;
      let fullDropoffCount = 0;
      let rejoinTransitionCount = 0;
      let lastFullDropoffAt: Date | null = null;
      let lastRejoinAt: Date | null = null;
      const dropoffEvents: { at: Date; reason: string | null }[] = [];
      const rejoinTimes: Date[] = [];
      const presenceIntervals: { start: Date; end: Date }[] = [];
      // Zoom only reports zoomParticipantId when the participant is logged into a Zoom account — for
      // seekers joining via a registration link (unauthenticated), it's an empty string, so most real
      // events carry none. Peak concurrent open connections is an id-independent lower bound on device
      // count: two JOINs with no LEFT between them prove two connections were open, regardless of ids.
      let peakConcurrentOpen = 0;

      for (const event of emailEvents) {
        // Events are chronologically ordered, so the last non-null name we see is the latest.
        if (event.zoomDisplayName) latestDisplayName = event.zoomDisplayName;
        // A RENAMED correction only carries a name update — it's not a real join/leave and must not
        // affect presence, device counts, or duration.
        if (event.eventType === ZoomLiveEventType.RENAMED) continue;
        const wasPresent = openDevices.size > 0 || anonymousOpenCount > 0;

        if (event.eventType === ZoomLiveEventType.JOINED) {
          joinCount++;
          lastJoinAt = event.occurredAt;
          if (!firstJoinAt) firstJoinAt = event.occurredAt;
          if (event.zoomParticipantId) {
            deviceIds.add(event.zoomParticipantId);
            openDevices.add(event.zoomParticipantId);
          } else {
            anonymousOpenCount++;
          }
        } else {
          leaveCount++;
          lastLeaveAt = event.occurredAt;
          if (event.zoomParticipantId && openDevices.has(event.zoomParticipantId)) {
            openDevices.delete(event.zoomParticipantId);
          } else if (anonymousOpenCount > 0) {
            anonymousOpenCount--;
          } else if (openDevices.size > 0) {
            const oldest = openDevices.values().next().value;
            if (oldest) openDevices.delete(oldest);
          }
        }

        peakConcurrentOpen = Math.max(peakConcurrentOpen, openDevices.size + anonymousOpenCount);
        const nowPresent = openDevices.size > 0 || anonymousOpenCount > 0;
        if (!wasPresent && nowPresent) {
          presenceOpenSince = event.occurredAt;
          if (hasEverJoined) {
            rejoinTransitionCount++;
            lastRejoinAt = event.occurredAt;
            rejoinTimes.push(event.occurredAt);
          }
          hasEverJoined = true;
        } else if (wasPresent && !nowPresent) {
          fullDropoffCount++;
          lastFullDropoffAt = event.occurredAt;
          dropoffEvents.push({ at: event.occurredAt, reason: event.leaveReason ?? null });
          if (presenceOpenSince) {
            const [preSeconds, countedSeconds] = this.splitIntervalAtSessionStart(
              presenceOpenSince,
              event.occurredAt,
              sessionStart,
            );
            preSessionDurationSeconds += preSeconds;
            durationSeconds += countedSeconds;
            presenceIntervals.push({ start: presenceOpenSince, end: event.occurredAt });
            presenceOpenSince = null;
          }
        }
      }
      const isCurrentlyJoined = presenceOpenSince !== null;
      if (presenceOpenSince) {
        const [preSeconds, countedSeconds] = this.splitIntervalAtSessionStart(
          presenceOpenSince,
          asOf,
          sessionStart,
        );
        preSessionDurationSeconds += preSeconds;
        durationSeconds += countedSeconds;
        presenceIntervals.push({ start: presenceOpenSince, end: asOf });
      }

      result.set(email, {
        firstJoinAt,
        lastJoinAt,
        lastLeaveAt,
        isCurrentlyJoined,
        joinCount,
        leaveCount,
        durationSeconds: Math.round(durationSeconds),
        preSessionDurationSeconds: Math.round(preSessionDurationSeconds),
        latestDisplayName,
        // Distinct ids are only a lower bound too (rare — only logged-in Zoom users get one), so take
        // whichever signal proves the higher count; default to 1 once we know they joined at all.
        deviceCount: joinCount > 0 ? Math.max(peakConcurrentOpen, deviceIds.size, 1) : 0,
        fullDropoffCount,
        rejoinTransitionCount,
        openConnectionCount: openDevices.size + anonymousOpenCount,
        lastFullDropoffAt,
        lastRejoinAt,
        dropoffEvents,
        rejoinTimes,
        presenceIntervals,
      });
    }
    return result;
  }

  private normalizeEmail(email: string | null | undefined): string | null {
    return email ? email.trim().toLowerCase() : null;
  }
}