import { Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import {
  ZoomAnalyticsAttendeeSummary,
  ProgramRegistration,
  ProgramRegistrationOnlineSession,
  ZoomGeneratedRegistrantLink,
  ProgramUserAttendance,
  User,
} from 'src/common/entities';
import { RegistrationOnlineSessionActivationStatus } from 'src/common/enum/registration-online-session-activation-status.enum';
import { SessionKpiFilter } from 'src/common/enum/session-kpi.enum';
import { OnlineSessionRegistrationStatus } from 'src/common/enum/online-session-registration-status.enum';
import { AttendanceStatus } from 'src/common/enum/attendance-status.enum';
import { AppLoggerService } from 'src/common/services/logger.service';
import { handleKnownErrors } from 'src/common/utils/handle-error.util';
import {
  UserTypeFilterValue,
  resolveUserTypeFilter,
  userTypeFilterIncludesAccountless,
} from 'src/common/utils/user-type-filter.util';
import { ERROR_CODES } from 'src/common/constants/error-string-constants';

/** Maps a v1 sortKey to its actual column — whitelisted here so a query param can never inject an arbitrary ORDER BY expression. */
const ATTENDEE_SORT_COLUMNS: Record<string, string> = {
  fullName: 'attendee.full_name',
  joinedAt: 'attendee.joined_at',
  durationSeconds: 'attendee.duration_seconds',
  dropoffCount: 'attendee.dropoff_count',
  rejoinCount: 'attendee.rejoin_count',
};

export interface AttendeeSummaryFilter {
  page: number;
  limit: number;
  search?: string;
  /** When set (an RM viewing their own seekers, or an admin/coordinator explicitly scoping to one RM), restricts rows to registrations whose `rm_contact` is this user. */
  rmContactId?: number;
  /** v1 only — the clicked KPI tile's filter; `all`/undefined applies no narrowing. */
  kpiFilter?: SessionKpiFilter;
  /**
   * v1 only, required when `kpiFilter === JOINED_LATE` — the instant after which a first join counts
   * as late (session start + the configured grace period). Resolved by the provider, which knows the
   * session's actual/scheduled start; the repository has no session context of its own.
   */
  lateCutoff?: Date;
  /**
   * v1 only, required when `kpiFilter === DROPPED` (or `attendanceOutcome` includes it) — the instant
   * before which `last_dropoff_at` must fall to count as a real drop-off (the session's resolved
   * end, minus the configured grace period). Without it, `last_dropoff_at IS NOT NULL` alone would
   * also match every seeker who simply stayed until the session's natural end — see
   * `applyKpiFilter`'s DROPPED case. Resolved by the provider (`resolveSessionEnd`); the repository
   * has no session context of its own.
   */
  dropoffCutoff?: Date;
  /**
   * v1 only — restricts rows to exactly these registration ids, pre-computed by the provider from
   * `rmAttendance`/`coordinatorAttendance`/`finalAttendance` attendance-state filters (those states
   * live in `program_user_attendance`, a different table resolved via the precedence engine, not a
   * column here — so the provider resolves the matching ids first and this repository just intersects
   * on them). An empty array (filter matched nobody) short-circuits to an empty page without querying;
   * `undefined` means "no such filter active."
   */
  registrationIds?: number[];
  /**
   * v1 only — filters directly on this table's own `is_system_attended` column (no cross-table
   * lookup needed, unlike rmAttendance/coordinatorAttendance/finalAttendance above). `PRESENT` →
   * `is_system_attended = true`; `ABSENT`/`UNKNOWN` → `= false` (the column is a plain boolean, so
   * there is no third "never marked" state to distinguish from absent).
   */
  systemAttendance?: AttendanceStatus;
  /**
   * v1 only — checkbox side filter (OR'd together), a subset of `dropped`/`rejoined`/`joinedLate`.
   * All three are plain columns/derivations on THIS table (same as `kpiFilter`'s own DROPPED/REJOINED/
   * JOINED_LATE cases — see `applyKpiFilter`), so this needs no cross-table lookup either.
   * `joinedLate` needs `lateCutoff` just like `kpiFilter` does.
   */
  attendanceOutcome?: SessionKpiFilter[];
  /**
   * v1 only — checkbox side filter (OR'd together) over the known/unknown split, independent of
   * clicking a KPI tile. Only meaningful for general attendees — doesn't force the
   * `registration_id IS NULL` scope itself (unlike `kpiFilter`'s own KNOWN/UNKNOWN cases in
   * `applyKpiFilter`), since it's only ever set alongside `generalOnly: true`. Selecting both values
   * applies no extra narrowing — see `applyKnownStatusFilter`.
   */
  knownStatus?: SessionKpiFilter[];
  /**
   * v1 only — checkbox side filter over the registrant's OWN account type
   * (`users.user_type`): Org and/or Seeker, matched as a plain IN over the selected values (see
   * `resolveUserTypeFilter`). A Seeker selection additionally matches registrations with no user
   * account of their own — the "registered for someone else" rows (see
   * `userTypeFilterIncludesAccountless`). Only meaningful for registered seekers — the facade drops
   * it for `generalOnly` callers, which have no registration and therefore no user account.
   */
  userType?: UserTypeFilterValue[];
  /**
   * v1 only — forces the `registration_id IS NULL` scope regardless of `kpiFilter`, so
   * `applyKpiFilter`'s presence/outcome conditions (joined/notJoined/dropped/rejoined/joinedLate)
   * can be layered on top of the general-attendees scope instead of being mutually exclusive with it.
   */
  generalOnly?: boolean;
  /**
   * Resolved by the provider from the query's ORIGINAL kpiFilter (the "Present" tile's own kpiFilter
   * is blanked to undefined by the time it reaches here — see `getAttendeeTable`'s
   * `needsFinalPresenceFilter` branch). True → order by join time (Present tile). False/undefined →
   * order by registration date (every other tile/tab), falling back to this row's own createdAt for
   * general/known/unknown attendees, who have no registration to read a date from.
   */
  sortByJoinTime?: boolean;
  /** Explicit column sort (e.g. a clicked table header) — takes precedence over the per-tab default above when present. */
  sortKey?: string;
  sortOrder?: 'ASC' | 'DESC';
}

export interface PaginatedAttendeeSummaries {
  data: ZoomAnalyticsAttendeeSummary[];
  total: number;
}

/** Owns `zoom_analytics_attendee_summary` — one upserted row per seeker per session. */
@Injectable()
export class ZoomAnalyticsAttendeeSummaryRepository {
  constructor(
    @InjectRepository(ZoomAnalyticsAttendeeSummary)
    private readonly repo: Repository<ZoomAnalyticsAttendeeSummary>,
    private readonly logger: AppLoggerService,
  ) {}

  async findById(id: number): Promise<ZoomAnalyticsAttendeeSummary | null> {
    try {
      return await this.repo.findOne({ where: { id } });
    } catch (error) {
      this.logger.error('Error finding Zoom analytics attendee summary by id', error?.stack, { error, id });
      handleKnownErrors(ERROR_CODES.ZOOM_ANALYTICS_GET_FAILED, error);
    }
  }

  /** Every session's summary row for one registration in one program — the seeker-detail rollup's source. */
  async findAllByProgramAndRegistration(
    programId: number,
    registrationId: number,
  ): Promise<ZoomAnalyticsAttendeeSummary[]> {
    try {
      return await this.repo.find({ where: { programId, registrationId } });
    } catch (error) {
      this.logger.error('Error listing Zoom analytics attendee summaries for registration', error?.stack, {
        error,
        programId,
        registrationId,
      });
      handleKnownErrors(ERROR_CODES.ZOOM_ANALYTICS_GET_FAILED, error);
    }
  }

  /**
   * Every session's summary row for an ENTIRE program (all registrations) in one
   * query — the Overall Analytics aggregate's source, so it can compute every
   * registration's per-session attendance/duration/devices without looping
   * {@link findAllByProgramAndRegistration} once per registration.
   */
  async findAllByProgram(programId: number): Promise<ZoomAnalyticsAttendeeSummary[]> {
    try {
      return await this.repo.find({ where: { programId } });
    } catch (error) {
      this.logger.error('Error listing Zoom analytics attendee summaries for program', error?.stack, {
        error,
        programId,
      });
      handleKnownErrors(ERROR_CODES.ZOOM_ANALYTICS_GET_FAILED, error);
    }
  }

  async findAllBySessionId(sessionId: number): Promise<ZoomAnalyticsAttendeeSummary[]> {
    try {
      return await this.repo.find({ where: { sessionId } });
    } catch (error) {
      this.logger.error('Error listing Zoom analytics attendee summaries', error?.stack, { error, sessionId });
      handleKnownErrors(ERROR_CODES.ZOOM_ANALYTICS_GET_FAILED, error);
    }
  }

  async listBySession(sessionId: number, filter: AttendeeSummaryFilter): Promise<PaginatedAttendeeSummaries> {
    if (filter.registrationIds && filter.registrationIds.length === 0) {
      return { data: [], total: 0 };
    }
    try {
      const query = this.repo.createQueryBuilder('attendee').where('attendee.session_id = :sessionId', { sessionId });
      if (filter.registrationIds) {
        query.andWhere('attendee.registration_id IN (:...registrationIds)', {
          registrationIds: filter.registrationIds,
        });
      }
      // A summary row is upserted from the roster at sync time and never deleted
      // when someone later drops off it (e.g. deactivated) — so re-check
      // activation here at read time too. Must check the per-session extension
      // row (registration_id + online_session_id), not ProgramRegistration's
      // global rollup status: the rollup reflects whichever session was most
      // recently toggled and deactivation cascades only to upcoming sessions,
      // so using the rollup here hid registrants from PAST sessions they
      // actually attended while active. Subquery, not innerJoin, to stay
      // getManyAndCount()-hydration-safe (same pattern as rmContactId below).
      // General attendees (registration_id NULL) have no registration to check
      // activation against — `extension.registration_id = NULL` is never true,
      // so without the OR this silently dropped every general-attendee row.
      // Deliberately does NOT check the registration's own deletedAt/status
      // (cancelled/archived/deleted) — this row is this SESSION's historical
      // attendance record, and it must keep showing exactly as it happened
      // regardless of what later happens to the registration elsewhere. Only
      // this per-session extension's own active/inactive state controls it.
      query.andWhere(
        `(attendee.registration_id IS NULL OR EXISTS ${query
          .subQuery()
          .select('1')
          .from(ProgramRegistrationOnlineSession, 'extension')
          .where('extension.registration_id = attendee.registration_id')
          .andWhere('extension.online_session_id = attendee.online_session_id')
          .andWhere('extension.deleted_at IS NULL')
          .andWhere('extension.activation_status = :activeActivationStatus')
          .getQuery()})`,
        {
          activeActivationStatus: RegistrationOnlineSessionActivationStatus.ACTIVE,
        },
      );
      if (filter.search) {
        query.andWhere(
          '(attendee.full_name ILIKE :search OR attendee.email ILIKE :search OR attendee.mobile ILIKE :search)',
          { search: `%${filter.search}%` },
        );
      }
      if (filter.rmContactId) {
        // A subquery, not an innerJoin: joining ProgramRegistration (not a relation on this entity)
        // directly into a query used with getManyAndCount() breaks TypeORM's order-by/select-expression
        // mapping (throws reading 'databaseName' of undefined) since it can't resolve the joined
        // entity's metadata for hydration. EXISTS keeps the query entity-hydration-safe.
        query.andWhere(
          `EXISTS ${query
            .subQuery()
            .select('1')
            .from(ProgramRegistration, 'registration')
            .withDeleted()
            .where('registration.id = attendee.registration_id')
            .andWhere('registration.rm_contact = :rmContactId')
            .getQuery()}`,
          { rmContactId: filter.rmContactId },
        );
      }
      // `userType` — the registrant's own account type, two tables away from this one
      // (attendee → registration → user). Same EXISTS-not-innerJoin reasoning as `rmContactId`
      // above: joining non-relation entities into a getManyAndCount() query breaks TypeORM's
      // hydration metadata resolution. A general-attendee row (registration_id NULL) can never
      // satisfy this, which is exactly why the facade drops the filter for generalOnly callers.
      const userTypes = resolveUserTypeFilter(filter.userType);
      if (userTypes) {
        // Second arm: a registration with no user account of its own ("registered for someone
        // else" — only owner_user_id is set) counts as a Seeker, so a Seeker selection must keep
        // those rows instead of dropping them with the user_id join. See
        // userTypeFilterIncludesAccountless.
        const typeMatch = `EXISTS ${query
          .subQuery()
          .select('1')
          .from(ProgramRegistration, 'userTypeRegistration')
          .withDeleted()
          .innerJoin(User, 'userTypeUser', 'userTypeUser.id = userTypeRegistration.userId')
          .where('userTypeRegistration.id = attendee.registration_id')
          .andWhere('userTypeUser.userType IN (:...fUserTypes)')
          .getQuery()}`;
        const accountlessMatch = `EXISTS ${query
          .subQuery()
          .select('1')
          .from(ProgramRegistration, 'accountlessRegistration')
          .withDeleted()
          .where('accountlessRegistration.id = attendee.registration_id')
          .andWhere('accountlessRegistration.userId IS NULL')
          .getQuery()}`;
        query.andWhere(
          userTypeFilterIncludesAccountless(userTypes)
            ? `(${typeMatch} OR ${accountlessMatch})`
            : typeMatch,
          { fUserTypes: userTypes },
        );
      }
      this.applyKpiFilter(query, filter.kpiFilter, filter.lateCutoff, filter.dropoffCutoff, filter.generalOnly);
      if (filter.systemAttendance !== undefined) {
        query.andWhere('attendee.is_system_attended = :systemAttendance', {
          systemAttendance: filter.systemAttendance === AttendanceStatus.PRESENT,
        });
      }
      this.applyAttendanceOutcomeFilter(query, filter.attendanceOutcome, filter.lateCutoff, filter.dropoffCutoff);
      this.applyKnownStatusFilter(query, filter.knownStatus);
      // An explicit column sort (clicked table header) always wins. Otherwise, default per tab:
      // Present/Currently-active tiles → join time desc. Every other tile/tab → registration date
      // desc, since that's the seeker's own "when did they sign up" date; general/known/unknown
      // attendees have no registration to read one from, so they fall back to this summary row's
      // own createdAt. NULLS LAST regardless of direction — a row with no value for the sorted
      // column (e.g. joined_at on someone who never joined) belongs at the end, not the top.
      // registration_id is always unique per row, so it's added as a tiebreaker to keep the order
      // fully deterministic across paginated pages (both sort expressions can tie or be null).
      let sortExpression: string;
      let sortOrder: 'ASC' | 'DESC';
      if (filter.sortKey && ATTENDEE_SORT_COLUMNS[filter.sortKey]) {
        sortExpression = ATTENDEE_SORT_COLUMNS[filter.sortKey];
        sortOrder = filter.sortOrder ?? 'ASC';
      } else if (filter.sortByJoinTime) {
        sortExpression = 'attendee.joined_at';
        sortOrder = 'DESC';
      } else {
        const registrationDateSubquery = query
          .subQuery()
          .select('registration.registration_date')
          .from(ProgramRegistration, 'registration')
          .where('registration.id = attendee.registration_id')
          .getQuery();
        sortExpression = `COALESCE((${registrationDateSubquery}), attendee.created_at)`;
        sortOrder = 'DESC';
      }
      const [data, total] = await query
        .orderBy(sortExpression, sortOrder, 'NULLS LAST')
        .addOrderBy('attendee.registration_id', 'ASC', 'NULLS LAST')
        .take(filter.limit)
        .skip((filter.page - 1) * filter.limit)
        .getManyAndCount();
      return { data, total };
    } catch (error) {
      this.logger.error('Error paginating Zoom analytics attendee summaries', error?.stack, { error, sessionId });
      handleKnownErrors(ERROR_CODES.ZOOM_ANALYTICS_GET_FAILED, error);
    }
  }

  /**
   * Narrows `listBySession` to the rows behind one v1 KPI tile — mirrors the tile's own counting rule
   * one-for-one (see `ZoomLiveEventAnalyticsProviderBase.getKpis`): JOINED/NOT_JOINED here read
   * `is_system_attended` (has this seeker ever joined, at all) — still correct for the general-
   * attendee "Logged In" tile (raw Zoom join activity), but for the registered-seeker Present/Absent
   * tiles `getAttendeeTable` intercepts JOINED/NOT_JOINED itself (`needsFinalPresenceFilter`) and
   * narrows by `registrationIds` (FINAL attendance) instead, passing `kpiFilter: undefined` down here
   * so this case never runs for that path. JOINED_LATE needs `lateCutoff` (the
   * provider-computed session-start + grace-period instant — this repository has no session context
   * of its own); REJOINED reads `rejoin_count` (the lifetime rejoin-transition count `getKpis`/
   * `getGeneralKpis` also key off). DROPPED reads `last_dropoff_at IS NOT NULL` — deliberately NOT
   * `dropoff_count > 0 AND last_rejoined_at IS NULL` (that pair would silently exclude anyone who
   * ever rejoined, even if they're dropped again right now, undercounting the list against the tile —
   * `last_dropoff_at` is set to non-null exactly when `!aggregate.isCurrentlyJoined` at sync time (see
   * `attendeeUpsertPayload`/`generalAttendeeUpsertPayload`), independent of `rejoin_count` — dropped
   * and rejoined are documented as not mutually exclusive on the tile side). "Currently not
   * connected" alone would also match everyone who simply stayed until the session's natural end, so
   * DROPPED also requires `last_dropoff_at` to fall before `dropoffCutoff` (the session's resolved
   * end minus the configured grace period) whenever the caller resolved one — the same grace-window
   * rule `getKpis`' `seekersDropped` tile counts by. ACTIVE/INACTIVE (the live-monitor tiles) are built from
   * the same two columns as JOINED/DROPPED: ACTIVE is `is_system_attended = true AND last_dropoff_at
   * IS NULL` (joined and still connected right now); INACTIVE is its complement (never joined, or
   * joined and currently disconnected) — matching `getKpis`' own `currentlyActive`/`currentlyInactive`
   * derivation one-for-one.
   *
   * General attendees (registration_id NULL) are excluded from every case except GENERAL itself,
   * unless `generalOnly` overrides the scope — they have no registration to attribute a KPI/
   * attendee-table row to, so `all`/`joined`/etc. (and the main `.../attendees` list with no
   * kpiFilter at all) must stay registered-seekers-only *unless* the caller is the dedicated
   * `.../general-attendees` endpoint, which passes `generalOnly: true` to keep every row scoped to
   * `registration_id IS NULL` while still layering on the same joined/notJoined/dropped/rejoined/
   * joinedLate condition below (see `AttendeeSummaryFilter.generalOnly`).
   *
   * KNOWN/UNKNOWN (the "External Members"/"Unidentified" tiles — see
   * `ZoomAnalyticsFacadeService.buildGeneralAttendanceTiles`) only ever describe general attendees, so
   * — same as GENERAL — they force the `registration_id IS NULL` scope themselves even if the caller
   * forgot to also pass `generalOnly`. There's no stored known/unknown column on this table (see
   * `ZoomLiveEventAnalyticsProviderBase.getGeneralKpis`'s own comment): both are resolved the same way
   * the KPI figures are, a live per-email EXISTS/NOT EXISTS match against `zoom_generated_registrant_link`,
   * keyed the same way (session + case/whitespace-insensitive `registrant_email`)
   * `ZoomGeneratedRegistrantLinkRepository.findActiveMapBySession` normalizes for.
   */
  private applyKpiFilter(
    query: ReturnType<Repository<ZoomAnalyticsAttendeeSummary>['createQueryBuilder']>,
    kpiFilter: SessionKpiFilter | undefined,
    lateCutoff: Date | undefined,
    dropoffCutoff: Date | undefined,
    generalOnly?: boolean,
  ): void {
    const generalOnlyFilter =
      kpiFilter === SessionKpiFilter.GENERAL || kpiFilter === SessionKpiFilter.KNOWN || kpiFilter === SessionKpiFilter.UNKNOWN;
    if (generalOnly || generalOnlyFilter) {
      query.andWhere('attendee.registration_id IS NULL');
    } else {
      query.andWhere('attendee.registration_id IS NOT NULL');
    }
    switch (kpiFilter) {
      case SessionKpiFilter.JOINED:
        query.andWhere('attendee.is_system_attended = true');
        break;
      case SessionKpiFilter.NOT_JOINED:
        query.andWhere('attendee.is_system_attended = false');
        break;
      case SessionKpiFilter.JOINED_LATE:
        if (lateCutoff) {
          query.andWhere('attendee.joined_at IS NOT NULL AND attendee.joined_at > :lateCutoff', { lateCutoff });
        } else {
          // Session start isn't known yet (unreconciled, no scheduled start either) — no seeker can
          // be classified as "late" without it, so the filter matches nothing rather than everything.
          query.andWhere('1 = 0');
        }
        break;
      case SessionKpiFilter.DROPPED:
        if (dropoffCutoff) {
          query.andWhere('attendee.last_dropoff_at IS NOT NULL AND attendee.last_dropoff_at < :dropoffCutoff', {
            dropoffCutoff,
          });
        } else {
          // No resolved session end (unreconciled, still live, no scheduled end either) — fall back
          // to the raw "currently not connected" rule rather than matching nothing.
          query.andWhere('attendee.last_dropoff_at IS NOT NULL');
        }
        break;
      case SessionKpiFilter.REJOINED:
        query.andWhere('attendee.rejoin_count > 0');
        break;
      case SessionKpiFilter.ACTIVE:
        // Deliberately the raw "currently connected right now" rule, NOT the DROPPED case's
        // grace-window/dropoffCutoff narrowing above — a seeker inside the end-grace window is
        // still correctly "not currently connected" for live-monitor purposes even though they
        // don't count as DROPPED for the coarser KPI. Matches getKpis' currentlyActive tile.
        query.andWhere('attendee.is_system_attended = true AND attendee.last_dropoff_at IS NULL');
        break;
      case SessionKpiFilter.INACTIVE:
        // ACTIVE's complement: never joined, or joined and currently disconnected.
        query.andWhere('(attendee.is_system_attended = false OR attendee.last_dropoff_at IS NOT NULL)');
        break;
      case SessionKpiFilter.KNOWN:
        query.andWhere(`EXISTS ${this.generatedLinkMatchSubQuery(query)}`, {
          generatedLinkStatus: OnlineSessionRegistrationStatus.REGISTERED,
        });
        break;
      case SessionKpiFilter.UNKNOWN:
        query.andWhere(`NOT EXISTS ${this.generatedLinkMatchSubQuery(query)}`, {
          generatedLinkStatus: OnlineSessionRegistrationStatus.REGISTERED,
        });
        break;
      case SessionKpiFilter.GENERAL:
      case SessionKpiFilter.ALL:
      case undefined:
        break;
    }
  }

  /**
   * The active `zoom_generated_registrant_link` row (if any) matching this attendee row's session +
   * email — shared by the KNOWN/UNKNOWN `applyKpiFilter` cases above. Same scope as
   * `ZoomGeneratedRegistrantLinkRepository.findActiveMapBySession` (status=registered, not
   * soft-deleted) so a KNOWN/UNKNOWN table narrowing never disagrees with the KPI tile's own count —
   * a FAILED registration attempt has no real Zoom registrant behind it, so its email must still
   * count as UNKNOWN if it shows up as a live walk-in. `generatedLinkStatus` is bound by the caller's
   * own `andWhere` (same convention as the `rmContactId` subquery above) — a subquery builder's own
   * parameters aren't merged into the parent query automatically.
   */
  private generatedLinkMatchSubQuery(
    query: ReturnType<Repository<ZoomAnalyticsAttendeeSummary>['createQueryBuilder']>,
  ): string {
    return query
      .subQuery()
      .select('1')
      .from(ZoomGeneratedRegistrantLink, 'generatedLink')
      .where('generatedLink.program_session_id = attendee.session_id')
      .andWhere('LOWER(TRIM(generatedLink.registrant_email)) = LOWER(TRIM(attendee.email))')
      .andWhere('generatedLink.status = :generatedLinkStatus')
      .andWhere('generatedLink.deleted_at IS NULL')
      .getQuery();
  }

  /**
   * The `program_user_attendance` row (if any) crediting this attendee row's own registration_id +
   * session_id as attended — `is_attended` there is kept current on every mark/webhook/undo (see
   * `OnlineAttendanceService.recomputeSummary`), so this is FINAL attendance (Coordinator > RM > Zoom
   * webhook), not this table's own Zoom-only `is_system_attended`. Used by
   * `getSessionAttendanceCounts` so a seeker marked present by RM/Coordinator without ever joining
   * Zoom still counts as attended (and lands in the `under60` duration bucket, since their
   * `duration_seconds` stays 0).
   */
  private finalAttendedExistsSubQuery(
    query: ReturnType<Repository<ZoomAnalyticsAttendeeSummary>['createQueryBuilder']>,
  ): string {
    return query
      .subQuery()
      .select('1')
      .from(ProgramUserAttendance, 'attendance')
      .where('attendance.registration_id = attendee.registration_id')
      .andWhere('attendance.session_id = attendee.session_id')
      .andWhere('attendance.is_attended = true')
      .getQuery();
  }

  /**
   * The `attendanceOutcome` side filter — a checkbox multi-select (OR'd together) over the same
   * `dropped`/`rejoined`/`joinedLate` conditions `applyKpiFilter` already knows, but independent of
   * clicking a KPI tile. Empty/undefined `outcomes` applies no narrowing.
   */
  private applyAttendanceOutcomeFilter(
    query: ReturnType<Repository<ZoomAnalyticsAttendeeSummary>['createQueryBuilder']>,
    outcomes: SessionKpiFilter[] | undefined,
    lateCutoff: Date | undefined,
    dropoffCutoff: Date | undefined,
  ): void {
    if (!outcomes?.length) return;
    const conditions: string[] = [];
    const params: Record<string, unknown> = {};
    if (outcomes.includes(SessionKpiFilter.DROPPED)) {
      // Same session-end grace-window rule as applyKpiFilter's DROPPED case — see its comment.
      if (dropoffCutoff) {
        conditions.push('(attendee.last_dropoff_at IS NOT NULL AND attendee.last_dropoff_at < :outcomeDropoffCutoff)');
        params.outcomeDropoffCutoff = dropoffCutoff;
      } else {
        conditions.push('attendee.last_dropoff_at IS NOT NULL');
      }
    }
    if (outcomes.includes(SessionKpiFilter.REJOINED)) {
      conditions.push('attendee.rejoin_count > 0');
    }
    if (outcomes.includes(SessionKpiFilter.JOINED_LATE)) {
      if (lateCutoff) {
        conditions.push('(attendee.joined_at IS NOT NULL AND attendee.joined_at > :outcomeLateCutoff)');
        params.outcomeLateCutoff = lateCutoff;
      } else {
        // Session start isn't known yet — no seeker can be "late" without it. Only this one disjunct
        // matches nothing (not the whole filter) — dropped/rejoined selections still narrow correctly.
        conditions.push('1 = 0');
      }
    }
    if (conditions.length) {
      query.andWhere(`(${conditions.join(' OR ')})`, params);
    }
  }

  /**
   * The `knownStatus` side filter — a checkbox multi-select (OR'd together) over the same
   * `zoom_generated_registrant_link` match `applyKpiFilter`'s KNOWN/UNKNOWN cases use, but
   * independent of clicking a KPI tile. Both selected = every general row already matches one or
   * the other, so no extra narrowing is applied — same as omitting the filter entirely.
   * Empty/undefined `knownStatus` applies no narrowing either.
   */
  private applyKnownStatusFilter(
    query: ReturnType<Repository<ZoomAnalyticsAttendeeSummary>['createQueryBuilder']>,
    knownStatus: SessionKpiFilter[] | undefined,
  ): void {
    if (!knownStatus?.length) return;
    const wantsKnown = knownStatus.includes(SessionKpiFilter.KNOWN);
    const wantsUnknown = knownStatus.includes(SessionKpiFilter.UNKNOWN);
    if (wantsKnown === wantsUnknown) return;
    const exists = `EXISTS ${this.generatedLinkMatchSubQuery(query)}`;
    query.andWhere(wantsKnown ? exists : `NOT ${exists}`, {
      generatedLinkStatus: OnlineSessionRegistrationStatus.REGISTERED,
    });
  }

  /**
   * Duration buckets among attended seekers, in minutes: <60, 60-90, 90+.
   * The last bucket is open-ended (no upper bound) so seekers who stayed for
   * the full session (often 120+ min once a meeting runs long) still land in
   * a bucket instead of being silently excluded from all three.
   */
  private static readonly DURATION_BUCKET_SQL = {
    under60: 'attendee.duration_seconds < 3600',
    from60to90: 'attendee.duration_seconds >= 3600 AND attendee.duration_seconds < 5400',
    from90to120: 'attendee.duration_seconds >= 5400',
  };

  /**
   * Attended/absent + duration-bucket counts for several sessions in one query,
   * GROUP BY session_id — powers the online-session list's per-session
   * attendance summary. "Attended" is FINAL attendance (`finalAttendedExistsSubQuery` —
   * Coordinator > RM > Zoom webhook via `program_user_attendance.is_attended`), not this table's own
   * Zoom-only `is_system_attended` — a seeker marked present without ever joining Zoom still counts
   * as attended, landing in `under60` (their `duration_seconds` stays 0). Reuses the exact
   * activation-status and rm scoping conventions from {@link listBySession} (see its comments for
   * why): a registrant deactivated for this session is excluded, and an RM caller only sees
   * counts for their own contacts.
   */
  async getSessionAttendanceCounts(
    sessionIds: number[],
    rmContactId?: number,
  ): Promise<
    Map<
      number,
      {
        attended: number;
        absent: number;
        durationBuckets: { under60: number; from60to90: number; from90to120: number };
      }
    >
  > {
    const map = new Map<
      number,
      {
        attended: number;
        absent: number;
        durationBuckets: { under60: number; from60to90: number; from90to120: number };
      }
    >();
    if (!sessionIds.length) return map;
    try {
      const query = this.repo.createQueryBuilder('attendee').select('attendee.session_id', 'sessionId');
      const finalAttended = `EXISTS ${this.finalAttendedExistsSubQuery(query)}`;
      query
        .addSelect(`COUNT(*) FILTER (WHERE ${finalAttended})`, 'attended')
        .addSelect(`COUNT(*) FILTER (WHERE NOT (${finalAttended}))`, 'absent')
        .addSelect(
          `COUNT(*) FILTER (WHERE ${finalAttended} AND ${ZoomAnalyticsAttendeeSummaryRepository.DURATION_BUCKET_SQL.under60})`,
          'under60',
        )
        .addSelect(
          `COUNT(*) FILTER (WHERE ${finalAttended} AND ${ZoomAnalyticsAttendeeSummaryRepository.DURATION_BUCKET_SQL.from60to90})`,
          'from60to90',
        )
        .addSelect(
          `COUNT(*) FILTER (WHERE ${finalAttended} AND ${ZoomAnalyticsAttendeeSummaryRepository.DURATION_BUCKET_SQL.from90to120})`,
          'from90to120',
        )
        .where('attendee.session_id IN (:...sessionIds)', { sessionIds })
        .andWhere('attendee.registration_id IS NOT NULL')
        .groupBy('attendee.session_id');

      query.andWhere(
        `EXISTS ${query
          .subQuery()
          .select('1')
          .from(ProgramRegistrationOnlineSession, 'extension')
          .where('extension.registration_id = attendee.registration_id')
          .andWhere('extension.online_session_id = attendee.online_session_id')
          .andWhere('extension.deleted_at IS NULL')
          .andWhere('extension.activation_status = :activeActivationStatus')
          .getQuery()}`,
        {
          activeActivationStatus: RegistrationOnlineSessionActivationStatus.ACTIVE,
        },
      );
      if (rmContactId != null) {
        query.andWhere(
          `EXISTS ${query
            .subQuery()
            .select('1')
            .from(ProgramRegistration, 'registration')
            .withDeleted()
            .where('registration.id = attendee.registration_id')
            .andWhere('registration.rm_contact = :rmContactId')
            .getQuery()}`,
          { rmContactId },
        );
      }

      const rows = await query.getRawMany<{
        sessionId: string;
        attended: string;
        absent: string;
        under60: string;
        from60to90: string;
        from90to120: string;
      }>();
      for (const row of rows) {
        map.set(Number(row.sessionId), {
          attended: Number(row.attended),
          absent: Number(row.absent),
          durationBuckets: {
            under60: Number(row.under60),
            from60to90: Number(row.from60to90),
            from90to120: Number(row.from90to120),
          },
        });
      }
      return map;
    } catch (error) {
      this.logger.error('Error computing session attendance counts', error?.stack, {
        error,
        sessionIds,
      });
      handleKnownErrors(ERROR_CODES.ZOOM_ANALYTICS_GET_FAILED, error);
    }
  }

  create(data: Partial<ZoomAnalyticsAttendeeSummary>): ZoomAnalyticsAttendeeSummary {
    return this.repo.create(new ZoomAnalyticsAttendeeSummary(data));
  }

  async save(row: ZoomAnalyticsAttendeeSummary): Promise<ZoomAnalyticsAttendeeSummary> {
    try {
      return await this.repo.save(row);
    } catch (error) {
      this.logger.error('Error saving Zoom analytics attendee summary', error?.stack, { error });
      handleKnownErrors(ERROR_CODES.ZOOM_ANALYTICS_SAVE_FAILED, error);
    }
  }

  /**
   * Batch upsert for a whole session's worth of attendee rows: one read for every existing row in the
   * session (instead of a find per attendee), then one independent save per attendee — deliberately NOT
   * `this.repo.save(rows[])`, which TypeORM wraps in a single transaction and would roll back every
   * attendee in the batch if even one save fails. Each attendee's save stands alone: a bad row (e.g. a
   * unique-constraint collision) is logged and skipped without blocking or rolling back anyone else's.
   * Keyed the same way the old single-row version was: (sessionId, registrationId) for seekers, since
   * several sibling registrations (proxy/child regs) can share the same real email, so keying on email
   * would collapse them into one row; (sessionId, email) for general attendees (registrationId null),
   * the same key the partial unique index on this table enforces for general rows (see the entity) —
   * every general attendee's `email` is already the normalized one `aggregateLiveEvents` grouped events
   * by, so this lookup and that index agree.
   */
  async upsertMany(
    payloads: Array<
      Partial<ZoomAnalyticsAttendeeSummary> & {
        sessionId: number;
        registrationId: number | null;
        email: string;
      }
    >,
  ): Promise<ZoomAnalyticsAttendeeSummary[]> {
    if (!payloads.length) return [];
    const sessionId = payloads[0].sessionId;
    let existingRows: ZoomAnalyticsAttendeeSummary[];
    try {
      existingRows = await this.repo.find({ where: { sessionId } });
    } catch (error) {
      this.logger.error('Error fetching existing Zoom analytics attendee summaries', error?.stack, {
        error,
        sessionId,
      });
      handleKnownErrors(ERROR_CODES.ZOOM_ANALYTICS_GET_FAILED, error);
    }

    const existingByRegistrationId = new Map(
      existingRows.filter((row) => row.registrationId !== null).map((row) => [row.registrationId as number, row]),
    );
    const existingByEmail = new Map(
      existingRows.filter((row) => row.registrationId === null).map((row) => [row.email, row]),
    );

    const results = await Promise.allSettled(
      payloads.map((data) => {
        const existing =
          data.registrationId === null
            ? existingByEmail.get(data.email)
            : existingByRegistrationId.get(data.registrationId);
        const row = existing ? Object.assign(existing, data) : this.create(data);
        return this.saveWithConflictRetry(row, data);
      }),
    );

    const saved: ZoomAnalyticsAttendeeSummary[] = [];
    results.forEach((result, index) => {
      if (result.status === 'fulfilled') {
        saved.push(result.value);
        return;
      }
      const failed = payloads[index];
      this.logger.error('Error upserting Zoom analytics attendee summary', result.reason?.stack, {
        error: result.reason,
        sessionId,
        registrationId: failed.registrationId,
        email: failed.email,
      });
    });
    return saved;
  }

  /**
   * `upsertMany` decides insert-vs-update from a snapshot of existing rows taken once up front, so two
   * overlapping calls for the same session (e.g. the attendee table and a live poll syncing at the same
   * time) can both miss the same registrant and race to INSERT — one wins, the other hits the
   * (session_id, registration_id)/(session_id, email) unique constraint. Without this retry that losing
   * save was simply dropped (logged, not persisted), silently discarding whichever aggregate lost the
   * race and leaving the seeker's duration/dropoff/rejoin figures stale — the "counts are wrong"
   * symptom. On a unique-violation (Postgres 23505), the other side's insert has just landed, so re-fetch
   * it and retry as an update instead of losing this write.
   */
  private async saveWithConflictRetry(
    row: ZoomAnalyticsAttendeeSummary,
    data: Partial<ZoomAnalyticsAttendeeSummary> & {
      sessionId: number;
      registrationId: number | null;
      email: string;
    },
  ): Promise<ZoomAnalyticsAttendeeSummary> {
    try {
      return await this.repo.save(row);
    } catch (error) {
      if ((error as { code?: string })?.code !== '23505') {
        throw error;
      }
      const conflicting = await this.repo.findOne({
        where:
          data.registrationId === null
            ? { sessionId: data.sessionId, email: data.email }
            : { sessionId: data.sessionId, registrationId: data.registrationId },
      });
      if (!conflicting) {
        throw error;
      }
      return this.repo.save(Object.assign(conflicting, data));
    }
  }
}