import { Injectable, OnModuleDestroy } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { Program, ProgramSession } from 'src/common/entities';
import { AppLoggerService } from 'src/common/services/logger.service';
import { handleKnownErrors } from 'src/common/utils/handle-error.util';
import { ERROR_CODES } from 'src/common/constants/error-string-constants';
import { InifniNotFoundException } from 'src/common/exceptions/infini-notfound-exception';
import InifniBadRequestException from 'src/common/exceptions/infini-badrequest-exception';
import InifniConflictException from 'src/common/exceptions/infini-conflict-exception';
import { zeptoEmailCreadentials } from 'src/common/constants/strings-constants';
import { PROGRAM_TYPE_KEYS } from 'src/common/constants/string-constants';
import { ROLE_KEYS } from 'src/common/constants/strings-constants';
import { UserRepository } from 'src/user/user.repository';
import { CommunicationTypeEnum } from 'src/common/enum/communication-type.enum';
import { ModeOfOperationEnum } from 'src/common/enum/mode-of-operation.enum';
import { SessionCommunicationPurposeEnum } from 'src/common/enum/session-communication-purpose.enum';
import { SessionCommunicationStatusEnum } from 'src/common/enum/session-communication-status.enum';
import { BulkCommunicationSelectionModeEnum } from 'src/common/enum/bulk-communication-selection-mode.enum';
import { EmailQueueService } from 'src/queue/services/email-queue.service';
import { WhatsAppQueueService } from 'src/queue/services/whatsapp-queue.service';
import {
  CommunicationMergeDataService,
  SendMergeCache,
  createSendMergeCache,
} from 'src/communication/service/communication-merge-data.service';
import { CommunicationTemplatesRepository } from 'src/communication/repositories/communication-templates.repository';
import { CommunicationService } from 'src/communication/communication.service';
import { SendBulkEmailDto, EmailAttachments } from 'src/communication/dto/email-communication.dto';
import { AwsS3Service } from 'src/common/services/awsS3.service';
import { fetchPdfAsBase64, delay } from 'src/common/utils/common.util';
import { QueuedAttachment } from 'src/queue/utils/queue-attachment.util';
import { WHATSAPP_BULK_SETTINGS } from 'src/common/constants/constants';
import { SendTemplateMessageDto } from 'src/communication/dto/whatsapp-communication.dto';
import {
  SessionCommunicationRepository,
  SessionCommunicationRecipient,
  CommonInviteRecipient,
  SystemGeneratedLink,
} from './session-communication.repository';
import {
  SendBulkSessionCommunicationDto,
  SendGeneralLinkBulkDto,
  SendGeneralLinkSingleDto,
  SendSingleSessionCommunicationDto,
  SendValueCardBulkDto,
  SendValueCardSingleDto,
  SendGeneralLinkValueCardBulkDto,
  SendGeneralLinkValueCardSingleDto,
} from './dto/send-session-communication.dto';
import { SendResult, SessionCommunicationStatusSummary } from './session-communication.types';
import {
  resolveChannels,
  resolveCommonInviteChannels,
  resolveGeneralLinkChannels,
  resolveSystemLinksChannels,
  purposeNeedsOccurrence,
  purposeSupportsOccurrence,
  isProgramLevelPurpose,
  GENERAL_LINK_TO_SEEKER_PURPOSE,
  SessionOccurrence,
  ChannelConfig,
} from './session-communication.constants';

/**
 * A channel + its access key + the resolved hdb_communication_templates id (null when the
 * template is not configured for the program).
 */
interface ChannelTemplate extends ChannelConfig {
  templateId: number | null;
}

/**
 * The email-sender identity + optional target session threaded through the dispatch
 * pipeline. Session-scoped sends (Invite/Absent/Value Card) build it from the loaded
 * ProgramSession; program-level sends (Welcome/Program completion) build it from the
 * Program with sessionId = null, so no session is required.
 */
interface SendContext {
  programId: number;
  sessionId: number | null;
  programTypeKey: string | null;
  emailSenderAddress: string | null;
  emailSenderName: string | null;
  /** Extra merge values supplied by the request (e.g. Value Card `description`), not from the DB. */
  extraContext?: Record<string, any>;
  /**
   * Attachments applied to every recipient of this send. ONE field rather than a per-transport
   * pair, so the queued and direct forms of the same attachment cannot disagree — each transport
   * derives what it needs (see toQueueAttachments / resolveDirectAttachments).
   *
   * Prefer `s3Key`: it is the only form that can travel on a queue message, and it means the bytes
   * are read once, by the consumer, immediately before sending. `content`/`sourceUrl` are for files
   * we hold no key for; they still send correctly, but only via a direct send.
   */
  attachments?: SendAttachment[];
}

/**
 * One attachment as it travels through this module — the union of what the two transports accept.
 * At least one of `content`/`s3Key` must be set; an item with neither is dropped (and logged) at
 * conversion time rather than silently producing an empty attachment.
 */
interface SendAttachment {
  name: string;
  contentType: string;
  /** Base64 bytes. */
  content?: string;
  /** S3 object key, resolvable to bytes by the consumer or the direct-send path. */
  s3Key?: string;
  /**
   * A non-S3 URL to fetch the bytes from, for a file we hold no key for. Not referenceable, so its
   * presence forces the send off the queue and onto the direct path, which fetches it in-process.
   * Kept as a third option rather than pre-fetching everywhere so the common all-S3 case still
   * downloads nothing.
   */
  sourceUrl?: string;
}

/**
 * Which audience a Value Card send is for. The two share the template and the whole send pipeline
 * but keep their own stored description + files on the session, because the material differs:
 * SEEKER goes to registered seekers, GENERAL_LINK ("pre-test") to generated-general-link recipients
 * who hold no registration.
 */
type ValueCardAudience = 'SEEKER' | 'GENERAL_LINK';

/** The program_session column each audience's Value Card details live in. */
const VALUE_CARD_DETAILS_COLUMN: Record<
  ValueCardAudience,
  'valueCardDetails' | 'pretestValueCardDetails'
> = {
  SEEKER: 'valueCardDetails',
  GENERAL_LINK: 'pretestValueCardDetails',
};

/** The S3 key prefix each audience's Value Card files are re-hosted under. */
const VALUE_CARD_S3_PREFIX: Record<ValueCardAudience, string> = {
  SEEKER: 'valuecard',
  GENERAL_LINK: 'pretestvaluecard',
};

@Injectable()
export class SessionCommunicationService implements OnModuleDestroy {
  /**
   * Work running off-request (bulk send processing, direct provider sends) — see runInBackground.
   * Held so shutdown can drain them instead of killing a send mid-flight.
   */
  private readonly pendingBackgroundWork = new Set<Promise<void>>();

  constructor(
    @InjectRepository(ProgramSession)
    private readonly sessionRepo: Repository<ProgramSession>,
    @InjectRepository(Program)
    private readonly programRepo: Repository<Program>,
    private readonly repository: SessionCommunicationRepository,
    private readonly templatesRepository: CommunicationTemplatesRepository,
    private readonly mergeDataService: CommunicationMergeDataService,
    private readonly emailQueueService: EmailQueueService,
    private readonly whatsAppQueueService: WhatsAppQueueService,
    private readonly communicationService: CommunicationService,
    private readonly awsS3Service: AwsS3Service,
    private readonly userRepository: UserRepository,
    private readonly logger: AppLoggerService,
  ) {}

  /**
   * Bulk send for a session. Resolves recipients per selection mode, then dispatches.
   *
   * Returns as soon as the audience is known — the dispatch itself runs off-request. Everything
   * that can legitimately fail the REQUEST stays inline and still answers with a real HTTP error:
   * the session/program lookup, the template check, and the recipient query (so "no eligible
   * recipients" is still a 400, and the response still carries a real `requested` count). What
   * moves to the background is the part that scales with the audience — resolving merge info per
   * recipient and enqueuing — which on a few hundred registrations took long enough for the
   * gateway to give up and return a 502.
   *
   * The returned SendResult therefore carries `accepted: true` with zeroed counters: the send has
   * been taken on, not completed. Its outcome is written to hdb_session_communication_status when
   * the background pass finishes, which is what the online-session GET responses already surface.
   */
  async sendBulk(
    dto: SendBulkSessionCommunicationDto,
    createdBy: number,
    options?: { background?: boolean },
  ): Promise<SendResult> {
    try {
      this.assertSelectionIds(dto);
      // Program-level purposes (Welcome/Program completion) carry no sessionId and resolve
      // their sender/merge context from the program; session-scoped purposes load the session.
      const { context, occurrence } = await this.resolveSendContext(dto);

      // Step 1 — fetch the communication template id(s) for this purpose/occurrence.
      const channelTemplates = await this.fetchChannelTemplates(
        dto.programId,
        dto.purpose,
        occurrence,
      );

      // Step 2 — filter the registrations via the per-purpose recipient function.
      const recipients = await this.filterRecipients(dto, context, occurrence);

      if (!recipients || recipients.length === 0) {
        throw new InifniBadRequestException(ERROR_CODES.SESSION_COMMUNICATION_NO_RECIPIENTS);
      }

      // Steps 3-4, inline. Used by the SQS-triggered path (triggerProgramCommunication), which is
      // already off any request and MUST keep awaiting + throwing: the processor marks the queue
      // message handled on return, so a swallowed failure would delete the message and lose the
      // scheduled send instead of retrying it.
      if (!options?.background) {
        const result = await this.processCommunications(
          context,
          channelTemplates,
          recipients,
          dto.purpose,
          createdBy,
          true,
        );
        await this.recordSendStatus(context, dto.purpose, occurrence, result, createdBy);
        return result;
      }

      // Steps 3-4 run OFF the request — see this method's doc comment. Everything they need is
      // already resolved above, so the task closes over plain values and touches no request state.
      this.runInBackground(
        `bulk ${dto.purpose} send, program ${dto.programId} (session ${dto.sessionId ?? 'n/a'}), ${recipients.length} recipient(s)`,
        async () => {
          try {
            // Step 3 — process (resolve merge info + enqueue) via the common function.
            const result = await this.processCommunications(
              context,
              channelTemplates,
              recipients,
              dto.purpose,
              createdBy,
              true,
            );
            // Step 4 — record the outcome (append-only history, per program + session + purpose).
            await this.recordSendStatus(context, dto.purpose, occurrence, result, createdBy);
            this.logger.log(
              `Bulk ${dto.purpose} send completed for program ${dto.programId} (session ${dto.sessionId ?? 'n/a'}): ${result.requested} requested, ${result.enqueued.email} enqueued, ${result.skipped.length} skipped`
            );
          } catch (error) {
            // The response is long gone, so a throw here would vanish. Record the failed run as a
            // SKIPPED row (nothing went out) so the status history shows the attempt instead of
            // looking like the send was never triggered at all.
            this.logger.error(
              `Bulk ${dto.purpose} send failed after acceptance for program ${dto.programId} (session ${dto.sessionId ?? 'n/a'}): ${(error as Error)?.message}`,
              (error as Error)?.stack,
            );
            await this.recordSendStatus(
              context,
              dto.purpose,
              occurrence,
              { requested: recipients.length, enqueued: { email: 0, whatsapp: 0 }, skipped: [] },
              createdBy,
            );
          }
        },
      );

      return {
        requested: recipients.length,
        enqueued: { email: 0, whatsapp: 0 },
        skipped: [],
        accepted: true,
      };
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SEND_FAILED, error);
    }
  }

  /**
   * Bulk Value Card send (email only). Like sendBulk, but the admin-supplied `description` is
   * threaded into the merge context (resolved by the `description` merge field) and the
   * `attachments` are attached to every email — which forces a direct send (the bulk-email
   * queue message carries no attachments).
   */
  async sendValueCardBulk(dto: SendValueCardBulkDto, createdBy: number): Promise<SendResult> {
    try {
      const purpose = SessionCommunicationPurposeEnum.VALUE_CARD;
      this.assertSelectionIds(dto as unknown as SendBulkSessionCommunicationDto);
      const { context, occurrence } = await this.resolveSendContext({
        programId: dto.programId,
        sessionId: dto.sessionId,
        purpose,
      });

      // Fetch each incoming file URL, re-upload it under valuecard/sessions/<id>/<filename>,
      // and keep both the email attachments and the uploaded S3 URLs.
      const { attachments, uploadedUrls } = await this.prepareValueCardFiles(
        dto.sessionId,
        dto.attachmentUrls,
      );
      // Persist the description + uploaded file URLs on the session.
      await this.storeValueCardDetails(dto.sessionId, dto.description, uploadedUrls);

      // Enrich the context with the request-supplied description + the resolved attachments.
      // Each attachment carries BOTH an s3Key (so the batch can be queued by reference) and the
      // bytes (already in hand from the re-upload, so a direct-send fallback costs no extra fetch).
      const sendContext: SendContext = {
        ...context,
        extraContext: { description: dto.description },
        attachments,
      };

      const channelTemplates = await this.fetchChannelTemplates(dto.programId, purpose, occurrence);
      const recipients = await this.repository.getValueCardRecipients(
        dto.programId,
        dto.sessionId,
        dto.selectionMode,
        dto.registrationIds,
      );
      if (!recipients || recipients.length === 0) {
        throw new InifniBadRequestException(ERROR_CODES.SESSION_COMMUNICATION_NO_RECIPIENTS);
      }

      const result = await this.processCommunications(
        sendContext,
        channelTemplates,
        recipients,
        purpose,
        createdBy,
        true,
      );
      await this.recordSendStatus(sendContext, purpose, occurrence, result, createdBy);
      return result;
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SEND_FAILED, error);
    }
  }

  /**
   * Single Value Card send (email only) — a top-up for ONE registration, for the late registrant
   * or the corrected address the bulk run couldn't reach.
   *
   * Two things must hold, both checked here rather than trusted from the request:
   *
   *   1. The bulk send for this session must already have gone out (a TRIGGERED row in
   *      hdb_session_communication_status). The value card is a session-wide artefact; a single
   *      send is a follow-up to a delivery that happened, never the first delivery.
   *   2. The session must carry stored `valueCardDetails` — both a non-empty description AND at
   *      least one document URL. These are written by the bulk send, and are the ONLY source of
   *      this email's body text and attachments, which is why the DTO has no description or
   *      attachment fields: every recipient of a session's value card receives identical content.
   *
   * Condition 1 nearly implies condition 2, but they are checked separately on purpose. The bulk
   * path persists the details on a best-effort basis (`storeValueCardDetails` swallows its own
   * errors so a storage hiccup can't fail a send that already went out), so a TRIGGERED run with
   * missing details is genuinely reachable — and produces a distinct, actionable error rather than
   * an email with an empty body and no attachment.
   *
   * Unlike the BULK value card — which must always send directly, because the bulk-email queue
   * message has no attachment field at all — this goes through the normal single-email queue: that
   * message DOES carry attachments as S3 references, which EmailProcessor downloads before sending.
   * The stored files are already in S3, so nothing is re-uploaded and no file bytes ride in the SQS
   * message.
   *
   * Like every other single send, this is NOT recorded in hdb_session_communication_status — that
   * table tracks bulk runs. The per-registration record lands in hdb_communication_track via
   * processCommunications, which is what the /summary endpoint counts.
   */
  async sendValueCardSingle(dto: SendValueCardSingleDto, createdBy: number): Promise<SendResult> {
    try {
      const purpose = SessionCommunicationPurposeEnum.VALUE_CARD;
      const { context, occurrence } = await this.resolveSendContext({
        programId: dto.programId,
        sessionId: dto.sessionId,
        purpose,
      });

      // Gate 1 — the bulk must have actually dispatched for this session.
      const bulkSent = await this.repository.hasTriggeredBulkSend(
        dto.programId,
        dto.sessionId,
        purpose,
      );
      if (!bulkSent) {
        throw new InifniBadRequestException(
          ERROR_CODES.VALUE_CARD_BULK_NOT_SENT,
          null,
          null,
          String(dto.sessionId),
        );
      }

      // Gate 2 — the session must hold the description + file URLs the bulk send stored.
      const details = await this.loadStoredValueCardDetails(dto.sessionId);

      // The stored files already live in S3 (the bulk run put them under
      // valuecard/sessions/<id>/), so these are pure references — no bytes are read here at all.
      // The single queue accepts references, so nothing is downloaded unless this send ends up
      // going direct; a stored URL that isn't an S3 object simply travels inline instead.
      const sendContext: SendContext = {
        ...context,
        extraContext: { description: details.description },
        attachments: this.buildSendAttachmentsFromUrls(details.documentUrls),
      };

      const channelTemplates = await this.fetchChannelTemplates(dto.programId, purpose, occurrence);

      // Same applicability filter the bulk send uses, scoped to this one registration — so a
      // single send can never reach someone the bulk audience rule would have excluded.
      const recipients = await this.repository.getValueCardRecipients(
        dto.programId,
        dto.sessionId,
        BulkCommunicationSelectionModeEnum.SELECTED,
        [dto.registrationId],
      );
      const recipient = recipients?.find((r) => r.id === dto.registrationId) ?? null;
      if (!recipient) {
        throw new InifniBadRequestException(
          ERROR_CODES.SESSION_COMMUNICATION_NOT_APPLICABLE,
          null,
          null,
          String(dto.registrationId),
        );
      }

      return await this.processCommunications(
        sendContext,
        channelTemplates,
        [recipient],
        purpose,
        createdBy,
        false,
      );
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SEND_FAILED, error);
    }
  }

  /**
   * Reads back the `valueCardDetails` the bulk send stored on the session, requiring BOTH a
   * non-empty description and at least one document URL — a value card email with no body text or
   * no attachment is not worth sending, so this fails loudly instead of degrading.
   */
  private async loadStoredValueCardDetails(
    sessionId: number,
    audience: ValueCardAudience = 'SEEKER',
  ): Promise<{ description: string; documentUrls: string[] }> {
    const column = VALUE_CARD_DETAILS_COLUMN[audience];
    const session = await this.sessionRepo.findOne({
      where: { id: sessionId },
      select: ['id', column],
    });
    const details = session?.[column] ?? null;
    const description = details?.description?.trim() ?? '';
    const documentUrls = (details?.documentUrls ?? []).filter(
      (url) => typeof url === 'string' && url.trim().length > 0,
    );
    if (!description || documentUrls.length === 0) {
      throw new InifniBadRequestException(
        ERROR_CODES.VALUE_CARD_DETAILS_MISSING,
        null,
        null,
        String(sessionId),
      );
    }
    return { description, documentUrls };
  }

  /**
   * Attachments for a queue message, or null when this send cannot be queued at all.
   *
   * References ONLY — never inline bytes, on either queue. That is the convention every other
   * queueing caller in the app follows (registration/payment/invoice all strip `content` and pass
   * `{ name, s3Key, contentType }`), and for good reason: base64 inflates a file by 4/3 and a queue
   * message is capped at 256 KB (enforced in SqsClientService.sendMessage), so inlining bytes both
   * wastes queue space and risks failing the enqueue outright. The bytes belong on the consumer
   * side, fetched from S3 immediately before the provider call — see resolveQueuedAttachments.
   *
   * An attachment with no `s3Key` therefore cannot travel on a message. Since sending a partial set
   * would deliver an email with a file silently missing, one such attachment makes the whole send
   * un-queueable: null means "send this directly instead", where the bytes can be fetched in-process
   * from any source (an S3 key OR an arbitrary URL — see resolveDirectAttachments).
   */
  private toQueueAttachments(attachments: SendAttachment[]): QueuedAttachment[] | null {
    const queued: QueuedAttachment[] = [];
    for (const attachment of attachments) {
      if (!attachment.s3Key) {
        this.logger.log(
          `Attachment ${attachment.name} has no s3Key, so this send cannot be queued ` +
            `(queue messages carry references only) — sending directly instead.`,
        );
        return null;
      }
      queued.push({
        name: attachment.name,
        contentType: attachment.contentType,
        s3Key: attachment.s3Key,
      });
    }
    return queued;
  }

  /**
   * Resolves attachments to the base64 form a DIRECT provider send needs, downloading only those
   * that arrived as a bare `s3Key`. Callers memoize the result, so on a queued send — where this is
   * never reached — nothing is downloaded at all.
   *
   * An attachment that cannot be fetched fails the send rather than being dropped: unlike the queue
   * consumer (which drops one bad file to save the rest of a batch), here the caller is still
   * inside the request and can report the problem.
   */
  private async resolveDirectAttachments(
    attachments: SendAttachment[],
  ): Promise<EmailAttachments[]> {
    const resolved: EmailAttachments[] = [];
    for (const attachment of attachments) {
      if (attachment.content) {
        resolved.push({
          content: attachment.content,
          mime_type: attachment.contentType,
          name: attachment.name,
        });
        continue;
      }
      // fetchFileBuffer reads an S3 key when the url is an S3 url and falls back to HTTP, so one
      // call covers both the s3Key and sourceUrl cases.
      const source = attachment.s3Key ?? attachment.sourceUrl;
      if (!source) {
        this.logger.warn(
          `Attachment ${attachment.name} has no content, s3Key or sourceUrl — dropping.`,
        );
        continue;
      }
      const buffer = await this.requireFileBuffer(source, attachment.name);
      resolved.push({
        content: buffer.toString('base64'),
        mime_type: attachment.contentType,
        name: attachment.name,
      });
    }
    return resolved;
  }

  /**
   * Reads one attachment's bytes, from an S3 key or any URL, and fails the send if they can't be
   * read. Unlike the queue consumer — which drops one bad file to save the rest of a batch — the
   * caller here is still inside the request, so a missing file is worth surfacing rather than
   * sending an email that quietly lacks its attachment.
   */
  private async requireFileBuffer(source: string, name: string): Promise<Buffer> {
    try {
      const buffer = await this.fetchFileBuffer(source);
      if (!buffer || buffer.length === 0) {
        throw new Error('empty file content');
      }
      return buffer;
    } catch (error) {
      this.logger.error(
        `Failed to load attachment ${name} (${source}): ${(error as Error)?.message}`,
        (error as Error)?.stack,
      );
      throw new InifniBadRequestException(
        ERROR_CODES.SESSION_COMMUNICATION_ATTACHMENT_FETCH_FAILED,
      );
    }
  }

  /**
   * Turns stored value-card URLs into unified {@link SendAttachment}s — references, with no bytes
   * read here at all.
   *
   * A URL that resolves to an S3 key becomes a reference, which is what lets the send stay on the
   * queue and defers the download to the consumer. One that doesn't (an external link, say) gets
   * neither `content` nor `s3Key` here: it is resolved lazily by
   * {@link SessionCommunicationService.resolveDirectAttachments} if a direct send happens, or
   * travels inline once fetched. Either way nothing is silently dropped, because the attachment
   * still appears in the list.
   */
  private buildSendAttachmentsFromUrls(urls: string[]): SendAttachment[] {
    return urls.map((url) => {
      const s3Key = this.awsS3Service.extractS3KeyFromUrl(url);
      if (!s3Key) {
        this.logger.warn(
          `Value card file is not an S3 object, so it cannot be sent by reference: ${url}`,
        );
      }
      return {
        name: this.fileNameFromUrl(url),
        contentType: this.mimeTypeFromUrl(url),
        ...(s3Key ? { s3Key } : { sourceUrl: url }),
      };
    });
  }


  /**
   * Bulk Common-Invite send (email + whatsapp). Notifies the staff who were issued a generated
   * common Zoom link (zoom_generated_registrant_link, ROLE) of the shared link + meeting
   * id/passcode. Scoped to the whole program by default (SHARED link type — one resource backs
   * every session); pass `sessionId` to narrow to just that session's recipients instead (the
   * PER_SESSION case, where each session holds its own distinct link). These recipients have no
   * program registration of their own, so this uses its own recipient set and per-recipient merge
   * context (common_user_name / common_zoom_join_link / common_meeting_id /
   * common_meeting_passcode) — the program-level fields (program_name / session_dates /
   * session_days_and_time) resolve via the shared merge service. The run is recorded once in
   * hdb_session_communication_status like the other bulk sends.
   */
  async sendCommonInviteBulk(
    programId: number,
    createdBy: number,
    sessionId?: number,
  ): Promise<SendResult> {
    try {
      const purpose = SessionCommunicationPurposeEnum.COMMON_INVITE;
      // Once-only: both of these announce a link that does not change, so a second run just mails
      // the same thing again. The scope is part of the identity — a program-wide run does not block
      // a per-session one, or vice versa (see hasTriggeredScopedBulkSend). TRIGGERED rows only, so
      // a run that resolved recipients but dispatched nothing (SKIPPED) can still be retried.
      if (await this.repository.hasTriggeredScopedBulkSend(programId, sessionId ?? null, purpose)) {
        this.logger.warn(
          `[COMMON-INVITE] program ${programId}${sessionId != null ? ` session ${sessionId}` : ''}: ` +
            'already triggered — send refused',
        );
        throw new InifniConflictException(ERROR_CODES.SESSION_COMMUNICATION_ALREADY_TRIGGERED);
      }

      const context: SendContext = {
        ...(await this.buildProgramContext(programId)),
        sessionId: sessionId ?? null,
      };
      // A sessionId narrows the send to one session and selects the per-session template variant
      // (session_name / session_date / session_time resolve for that session); otherwise the
      // program-level common invite is used.
      const channelTemplates = await this.fetchTemplatesForChannels(
        programId,
        resolveCommonInviteChannels(sessionId != null),
      );
      const recipients = await this.repository.getCommonInviteRecipients(programId, sessionId);
      this.logger.log(
        `[COMMON-INVITE] program ${programId}${sessionId != null ? ` session ${sessionId}` : ''}: ${recipients.length} recipient(s)`,
      );
      if (!recipients || recipients.length === 0) {
        throw new InifniBadRequestException(ERROR_CODES.SESSION_COMMUNICATION_NO_RECIPIENTS);
      }
      this.logger.log(
        `[COMMON-INVITE] program ${programId}: dispatching ${recipients.length} recipient(s) and context ${JSON.stringify(context)} and recipients ${JSON.stringify(recipients)}`,
      );
      const result = await this.processCommonInvite(
        context,
        channelTemplates,
        recipients,
        purpose,
        createdBy,
      );
      await this.recordSendStatus(context, purpose, null, result, createdBy);
      return result;
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SEND_FAILED, error);
    }
  }

  /**
   * Bulk System-Links send (email only). Notifies the ADMIN users who were issued a generated
   * link of the system/placeholder Zoom links, listed as an HTML table (systemJoiningDetails).
   * Scoped to the whole program by default (SHARED link type); pass `sessionId` to narrow the
   * table to just that session's placeholder links instead (the PER_SESSION case). admin_user_name
   * is per recipient; the table + program-level fields (program_name / session_dates /
   * session_days_and_time) are shared. Recorded once in hdb_session_communication_status like the
   * other bulk sends.
   */
  async sendSystemLinksBulk(
    programId: number,
    createdBy: number,
    sessionId?: number,
  ): Promise<SendResult> {
    try {
      const purpose = SessionCommunicationPurposeEnum.SYSTEM_LINKS;
      // Once-only: both of these announce a link that does not change, so a second run just mails
      // the same thing again. The scope is part of the identity — a program-wide run does not block
      // a per-session one, or vice versa (see hasTriggeredScopedBulkSend). TRIGGERED rows only, so
      // a run that resolved recipients but dispatched nothing (SKIPPED) can still be retried.
      if (await this.repository.hasTriggeredScopedBulkSend(programId, sessionId ?? null, purpose)) {
        this.logger.warn(
          `[SYSTEM-LINKS] program ${programId}${sessionId != null ? ` session ${sessionId}` : ''}: ` +
            'already triggered — send refused',
        );
        throw new InifniConflictException(ERROR_CODES.SESSION_COMMUNICATION_ALREADY_TRIGGERED);
      }

      const context: SendContext = {
        ...(await this.buildProgramContext(programId)),
        sessionId: sessionId ?? null,
      };
      // A sessionId narrows the send (and the system-links table) to one session and selects the
      // per-session template variant (session_name / session_date / session_time); otherwise the
      // program-level system-links template is used.
      const channelTemplates = await this.fetchTemplatesForChannels(
        programId,
        resolveSystemLinksChannels(sessionId != null),
      );
      const recipients = await this.resolveAdminRecipients();
      if (!recipients || recipients.length === 0) {
        this.logger.warn(`[SYSTEM-LINKS] program ${programId}: no admin recipients found`);
        throw new InifniBadRequestException(ERROR_CODES.SESSION_COMMUNICATION_NO_RECIPIENTS);
      }

      // Cumulative table of the program's (or, when scoped, just this session's) system/placeholder
      // links (name + link) — same for every admin recipient, so build it once.
      const systemLinks = await this.repository.getSystemPlaceholderLinks(programId, sessionId);
      const systemJoiningDetails = this.buildSystemJoiningDetailsTable(systemLinks);

      // The links ARE the message: this email carries nothing else of substance. With none
      // generated yet (or none for the requested session), buildSystemJoiningDetailsTable returns
      // '' and every admin would get a mail whose table is simply blank — which reads as "the
      // links are gone" rather than "they were never created". Refuse instead, so the caller is
      // told to generate the links first.
      if (systemJoiningDetails === '') {
        this.logger.warn(
          `[SYSTEM-LINKS] program ${programId}${sessionId != null ? ` session ${sessionId}` : ''}: ` +
            `no usable system links (${systemLinks?.length ?? 0} row(s) found) — send refused`,
        );
        throw new InifniBadRequestException(ERROR_CODES.SESSION_COMMUNICATION_NO_SYSTEM_LINKS);
      }

      const result = await this.processSystemLinks(
        context,
        channelTemplates,
        recipients,
        systemJoiningDetails,
        purpose,
        createdBy,
      );
      await this.recordSendStatus(context, purpose, null, result, createdBy);
      return result;
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SEND_FAILED, error);
    }
  }

  /**
   * Bulk General-Link send (Welcome / Invite / Absent / Program completion, email + whatsapp).
   * Recipients are every zoom_generated_registrant_link row for the program (or, when `sessionId`
   * is given, just that session). Reuses the SAME templates as the seeker purposes (see
   * GENERAL_LINK_TO_SEEKER_PURPOSE / resolveGeneralLinkChannels), including the real REGULAR/
   * PRE_FINAL/FINAL occurrence resolved by resolveOccurrence — exactly like a seeker send.
   * Eligibility always requires status = REGISTERED, and always gates GENERAL_LINK_ABSENT to real
   * absentees (read from zoom_analytics_attendee_summary, matched by session + email — the
   * general-attendee half of that table), the same way getAbsentRecipients never lets a caller
   * bypass it. GENERAL_LINK_INVITE at the final session is deliberately NOT gated by prior-session
   * attendance the way the seeker getFinalInviteRecipients rule is — these recipients are staff/
   * admin (ROLE-type generated links), not seekers on a structured multi-session journey, so
   * `occurrence` here only picks the regular- vs final-session template content. Reuses the
   * Common-Invite dispatch pipeline (processCommonInvite/resolveProgramLevelSend) since it
   * already handles registration-less recipients with per-recipient common_* merge fields.
   */
  async sendGeneralLinkBulk(dto: SendGeneralLinkBulkDto, createdBy: number): Promise<SendResult> {
    try {
      const { programId, sessionId, purpose } = dto;
      const seekerPurpose = GENERAL_LINK_TO_SEEKER_PURPOSE[purpose];
      if (!seekerPurpose) {
        // Unreachable in practice — the DTO's IsIn(GENERAL_LINK_PURPOSES) already rejects any
        // other value — but narrows the type for resolveOccurrence below.
        throw new InifniBadRequestException(ERROR_CODES.SESSION_COMMUNICATION_NOT_APPLICABLE);
      }
      const context: SendContext = {
        ...(await this.buildProgramContext(programId)),
        sessionId: sessionId ?? null,
      };

      let occurrence: SessionOccurrence | null = null;
      if (sessionId != null) {
        const session = await this.loadOnlineSession(sessionId);
        occurrence = await this.resolveOccurrence(session, seekerPurpose);
      }

      const channelTemplates = await this.fetchTemplatesForChannels(
        programId,
        resolveGeneralLinkChannels(purpose, occurrence),
      );
      const recipients = await this.repository.getGeneralLinkRecipients(
        programId,
        sessionId ?? null,
        purpose,
      );
      this.logger.log(
        `[GENERAL-LINK] program ${programId}${sessionId != null ? ` session ${sessionId}` : ''}: ` +
          `purpose=${purpose}, occurrence=${occurrence ?? 'n/a'}, ${recipients.length} recipient(s)`,
      );
      if (!recipients || recipients.length === 0) {
        throw new InifniBadRequestException(ERROR_CODES.SESSION_COMMUNICATION_NO_RECIPIENTS);
      }
      const result = await this.processCommonInvite(
        context,
        channelTemplates,
        recipients,
        purpose,
        createdBy,
      );
      await this.recordSendStatus(context, purpose, occurrence, result, createdBy);
      return result;
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SEND_FAILED, error);
    }
  }

  /**
   * Single send to one general-link recipient (one zoom_generated_registrant_link row, identified
   * by its own id — these have no registrationId). `occurrence` picks the REGULAR/FINAL template
   * for GENERAL_LINK_INVITE/GENERAL_LINK_ABSENT; omitted for the occurrence-independent Welcome/
   * Program-completion. Eligibility always requires status = REGISTERED, and additionally — for
   * GENERAL_LINK_ABSENT, that the row's email did NOT attend its own session (read from
   * zoom_analytics_attendee_summary, matched by email — see getGeneralLinkRecipientById).
   * GENERAL_LINK_INVITE at occurrence FINAL is NOT gated by prior-session attendance — these
   * recipients are staff/admin, not seekers on a structured multi-session journey, so `occurrence`
   * only picks the template here. There is no explicit sessionId on this endpoint (see the DTO) —
   * the recipient's own program_session_id (from getGeneralLinkRecipientById) is threaded into the
   * context so session-level merge fields (session name/date/time, meeting id/passcode) resolve
   * against THIS row's actual session, not the program's first session (the fallback every
   * session-level computed field uses when no target session is supplied). Reuses the
   * Common-Invite dispatch pipeline like the bulk send. Not recorded in
   * hdb_session_communication_status — consistent with sendSingle, that table tracks bulk-
   * communication runs only.
   */
  /**
   * Bulk general-link ("pre-test") Value Card send — email only, session-scoped.
   *
   * The general-link counterpart of {@link sendValueCardBulk}: same VALUE_CARD template, same
   * upload-then-reference handling of the files, same "only those who attended the session" audience
   * rule (resolved from the general-attendee rows of zoom_analytics_attendee_summary, since these
   * recipients have no registration to read program_user_attendance by) — but a different audience
   * (generated-general-link recipients, who hold no program registration) and its own storage column
   * (program_session.pretest_value_card_details), so a seeker send and a general-link send for the
   * same session never overwrite each other's material.
   *
   * Like the seeker bulk send, THIS is the only writer of those stored details — the single send
   * below reads them back.
   */
  async sendGeneralLinkValueCardBulk(
    dto: SendGeneralLinkValueCardBulkDto,
    createdBy: number,
  ): Promise<SendResult> {
    try {
      const purpose = SessionCommunicationPurposeEnum.GENERAL_LINK_VALUE_CARD;
      // Validates the session is online and resolves the occurrence the template pair is chosen by.
      const session = await this.loadOnlineSession(dto.sessionId);
      const occurrence = await this.resolveOccurrence(
        session,
        SessionCommunicationPurposeEnum.VALUE_CARD,
      );

      const { attachments, uploadedUrls } = await this.prepareValueCardFiles(
        dto.sessionId,
        dto.attachmentUrls,
        'GENERAL_LINK',
      );
      await this.storeValueCardDetails(
        dto.sessionId,
        dto.description,
        uploadedUrls,
        'GENERAL_LINK',
      );

      const context: SendContext = {
        ...(await this.buildProgramContext(dto.programId)),
        sessionId: dto.sessionId,
        extraContext: { description: dto.description },
        attachments,
      };

      const channelTemplates = await this.fetchTemplatesForChannels(
        dto.programId,
        resolveGeneralLinkChannels(purpose, occurrence),
      );
      const recipients = await this.repository.getGeneralLinkRecipients(
        dto.programId,
        dto.sessionId,
        purpose,
      );
      if (!recipients || recipients.length === 0) {
        throw new InifniBadRequestException(ERROR_CODES.SESSION_COMMUNICATION_NO_RECIPIENTS);
      }

      const result = await this.processCommonInvite(
        context,
        channelTemplates,
        recipients,
        purpose,
        createdBy,
      );
      await this.recordSendStatus(context, purpose, occurrence, result, createdBy);
      return result;
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SEND_FAILED, error);
    }
  }

  /**
   * Single general-link ("pre-test") Value Card send — a top-up for ONE generated-link recipient.
   *
   * Mirrors {@link sendValueCardSingle} exactly, against the general-link audience and the
   * pretest_value_card_details column: no description/attachment fields on the request (they come
   * from the stored details, so every recipient of a session gets identical material), the bulk send
   * must already have gone out, and those stored details must exist. The recipient must also have
   * attended the session — enforced by the same recipient query the bulk audience uses, so a single
   * send can never reach someone bulk would have excluded. Writes nothing.
   */
  async sendGeneralLinkValueCardSingle(
    dto: SendGeneralLinkValueCardSingleDto,
    createdBy: number,
  ): Promise<SendResult> {
    try {
      const purpose = SessionCommunicationPurposeEnum.GENERAL_LINK_VALUE_CARD;
      const recipient = await this.repository.getGeneralLinkRecipientById(
        dto.programId,
        dto.generatedLinkId,
        purpose,
      );
      if (!recipient) {
        throw new InifniBadRequestException(
          ERROR_CODES.SESSION_COMMUNICATION_NOT_APPLICABLE,
          null,
          null,
          String(dto.generatedLinkId),
        );
      }

      // The value card is session material, so the session comes from the link itself rather than
      // the request — a general link is always issued against one session.
      const sessionId = recipient.programSessionId ?? null;
      if (sessionId == null) {
        throw new InifniBadRequestException(
          ERROR_CODES.SESSION_COMMUNICATION_IDS_REQUIRED,
          null,
          null,
          String(dto.generatedLinkId),
        );
      }
      const session = await this.loadOnlineSession(sessionId);
      const occurrence = await this.resolveOccurrence(
        session,
        SessionCommunicationPurposeEnum.VALUE_CARD,
      );

      // Gate 1 — the general-link bulk must have actually dispatched for this session.
      const bulkSent = await this.repository.hasTriggeredBulkSend(
        dto.programId,
        sessionId,
        purpose,
      );
      if (!bulkSent) {
        throw new InifniBadRequestException(
          ERROR_CODES.VALUE_CARD_BULK_NOT_SENT,
          null,
          null,
          String(sessionId),
        );
      }

      // Gate 2 — the session must hold the description + file URLs that bulk send stored.
      const details = await this.loadStoredValueCardDetails(sessionId, 'GENERAL_LINK');

      const context: SendContext = {
        ...(await this.buildProgramContext(dto.programId)),
        sessionId,
        extraContext: { description: details.description },
        attachments: this.buildSendAttachmentsFromUrls(details.documentUrls),
      };

      const channelTemplates = await this.fetchTemplatesForChannels(
        dto.programId,
        resolveGeneralLinkChannels(purpose, occurrence),
      );

      // Not recorded in hdb_session_communication_status — that table tracks bulk runs only.
      return await this.processCommonInvite(
        context,
        channelTemplates,
        [recipient],
        purpose,
        createdBy,
      );
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SEND_FAILED, error);
    }
  }

  async sendGeneralLinkSingle(
    dto: SendGeneralLinkSingleDto,
    createdBy: number,
  ): Promise<SendResult> {
    try {
      const { programId, generatedLinkId, purpose, occurrence } = dto;
      const context = await this.buildProgramContext(programId);
      const channelTemplates = await this.fetchTemplatesForChannels(
        programId,
        resolveGeneralLinkChannels(purpose, occurrence ?? null),
      );
      const recipient = await this.repository.getGeneralLinkRecipientById(
        programId,
        generatedLinkId,
        purpose,
      );
      if (!recipient) {
        throw new InifniBadRequestException(
          ERROR_CODES.SESSION_COMMUNICATION_NOT_APPLICABLE,
          null,
          null,
          String(generatedLinkId),
        );
      }
      context.sessionId = recipient.programSessionId ?? null;
      return await this.processCommonInvite(
        context,
        channelTemplates,
        [recipient],
        purpose,
        createdBy,
      );
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SEND_FAILED, error);
    }
  }

  /**
   * System-links audience: every user holding the ADMIN role (from the user↔role tables, the
   * authoritative role source — not the generated-link role_key, which stores only one role per
   * row and would miss an admin who also holds another role). Users with no email are dropped,
   * and the set is deduplicated by user id. Mapped to the shared program-level recipient shape
   * (email only; no link/meeting fields).
   */
  private async resolveAdminRecipients(): Promise<CommonInviteRecipient[]> {
    const adminUsers = await this.userRepository.getUsersByRoleKeys([ROLE_KEYS.ADMIN]);
    const byId = new Map<number, CommonInviteRecipient>();
    for (const user of adminUsers) {
      if (!user.email || byId.has(user.id)) {
        continue;
      }
      byId.set(user.id, {
        userId: user.id,
        displayName: user.fullName || user.email,
        emailAddress: user.email,
        mobileNumber: null,
        joinUrl: null,
        meetingId: null,
        meetingPasscode: null,
      });
    }
    return Array.from(byId.values());
  }

  /**
   * For every incoming value-card file URL: fetch the bytes (private-S3 via the S3 client, any
   * other URL over HTTP), re-upload under `valuecard/sessions/<sessionId>/<filename>` (filename
   * taken from the incoming URL), and build the email attachment. Returns the email attachments
   * plus the re-hosted S3 URLs. A URL that can't be fetched/uploaded aborts the send — the value
   * card is meaningless without its file(s).
   */
  private async prepareValueCardFiles(
    sessionId: number,
    urls: string[],
    audience: ValueCardAudience = 'SEEKER',
  ): Promise<{ attachments: SendAttachment[]; uploadedUrls: string[] }> {
    // Each entry carries both forms: the s3Key we just uploaded to (no URL parsing needed) and the
    // bytes we had to read anyway in order to upload them.
    const attachments: SendAttachment[] = [];
    const uploadedUrls: string[] = [];
    for (const url of urls) {
      try {
        const buffer = await this.fetchFileBuffer(url);
        if (!buffer || buffer.length === 0) {
          throw new Error('empty file content');
        }
        const name = this.fileNameFromUrl(url);
        const mimeType = this.mimeTypeFromUrl(url);
        const s3Key = `${VALUE_CARD_S3_PREFIX[audience]}/sessions/${sessionId}/${name}`;
        const uploadedUrl = await this.awsS3Service.uploadToS3(s3Key, buffer, mimeType);
        uploadedUrls.push(uploadedUrl);
        attachments.push({
          name,
          contentType: mimeType,
          content: buffer.toString('base64'),
          s3Key,
        });
      } catch (error) {
        this.logger.error(
          `Failed to prepare value card attachment ${url}: ${(error as Error)?.message}`,
          (error as Error)?.stack,
        );
        throw new InifniBadRequestException(
          ERROR_CODES.SESSION_COMMUNICATION_ATTACHMENT_FETCH_FAILED,
        );
      }
    }
    return { attachments, uploadedUrls };
  }

  /** Fetch a file's bytes: private-S3 URLs via the S3 client, any other URL over HTTP. */
  private async fetchFileBuffer(url: string): Promise<Buffer | null> {
    const key = this.awsS3Service.extractS3KeyFromUrl(url);
    if (key) {
      return this.awsS3Service.getS3ObjectAsBuffer(key);
    }
    const base64 = await fetchPdfAsBase64(url);
    return base64 ? Buffer.from(base64, 'base64') : null;
  }

  /** Persist the value card's description + uploaded file URLs on the session. Best-effort. */
  private async storeValueCardDetails(
    sessionId: number,
    description: string,
    documentUrls: string[],
    audience: ValueCardAudience = 'SEEKER',
  ): Promise<void> {
    try {
      await this.sessionRepo.update(
        { id: sessionId },
        { [VALUE_CARD_DETAILS_COLUMN[audience]]: { description, documentUrls } },
      );
    } catch (error) {
      this.logger.error(
        `Failed to store ${audience} value card details for session ${sessionId}: ${(error as Error)?.message}`,
        (error as Error)?.stack,
      );
    }
  }

  /** File name from a URL (last path segment, query stripped); falls back to "attachment". */
  private fileNameFromUrl(url: string): string {
    const path = url.split('?')[0];
    const name = decodeURIComponent(path.substring(path.lastIndexOf('/') + 1));
    return name || 'attachment';
  }

  /** Best-effort content type from a URL's file extension. */
  private mimeTypeFromUrl(url: string): string {
    const ext = url.split('?')[0].split('.').pop()?.toLowerCase();
    switch (ext) {
      case 'ppt':
        return 'application/vnd.ms-powerpoint';
      case 'pptx':
        return 'application/vnd.openxmlformats-officedocument.presentationml.presentation';
      case 'pdf':
        return 'application/pdf';
      default:
        return 'application/octet-stream';
    }
  }

  /**
   * Single send for one registration in a session.
   */
  async sendSingle(dto: SendSingleSessionCommunicationDto, createdBy: number): Promise<SendResult> {
    try {
      const { context, occurrence } = await this.resolveSendContext(dto);

      // Step 1 — fetch the communication template id(s) for this purpose/occurrence.
      const channelTemplates = await this.fetchChannelTemplates(
        dto.programId,
        dto.purpose,
        occurrence,
      );

      // Step 2 — resolve the recipient through the SAME per-purpose applicability filter bulk
      // uses, scoped to just this registration (selection mode SELECTED). This enforces the
      // purpose's audience rule for a single send too — Absent must be absent + have a link,
      // Invite must have a link (final Invite also attended every prior session), etc. — rather
      // than sending to any eligible registration. An empty result means the registration is not
      // an applicable recipient for this communication.
      const recipients = await this.filterRecipients(
        {
          programId: dto.programId,
          sessionId: dto.sessionId,
          purpose: dto.purpose,
          selectionMode: BulkCommunicationSelectionModeEnum.SELECTED,
          registrationIds: [dto.registrationId],
        } as SendBulkSessionCommunicationDto,
        context,
        occurrence,
      );
      const recipient = recipients?.find((r) => r.id === dto.registrationId) ?? null;
      if (!recipient) {
        throw new InifniBadRequestException(
          ERROR_CODES.SESSION_COMMUNICATION_NOT_APPLICABLE,
          null,
          null,
          String(dto.registrationId),
        );
      }

      // Step 3 — process (resolve merge info + enqueue) via the common function. Single send.
      // Single sends are NOT recorded in hdb_session_communication_status — that table tracks
      // bulk-communication runs only.
      return await this.processCommunications(
        context,
        channelTemplates,
        [recipient],
        dto.purpose,
        createdBy,
        false,
      );
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SEND_FAILED, error);
    }
  }

  /**
   * Run a DIRECT (non-queued) provider send off the request.
   *
   * The queued path returns as soon as the batch is accepted by SQS; the direct fallback used to
   * be the one place where the caller waited for the provider itself — and that wait is long by
   * construction (WATI paces 10 recipients per call with a 2s gap; ZeptoMail 500 per call with a
   * 1s gap), so a large audience could hold the HTTP request open for minutes. Starting the send
   * here and returning immediately makes both paths behave the same way from the caller's side:
   * a successful response means "accepted for sending", never "delivered".
   *
   * The trade-off is deliberate: a failure inside the task can no longer be reported in the
   * response's `skipped` list, because the response is already gone. It surfaces where per-send
   * outcomes already live — this log, and the per-recipient FAILED rows in hdb_communication_track
   * that sendBulkEmail / sendBulkTemplateMessage write either way.
   *
   * The promise is tracked so onModuleDestroy can drain in-flight sends rather than let a deploy
   * kill one halfway through its provider chunks.
   */
  private runInBackground(label: string, task: () => Promise<void>): void {
    const pending = task()
      .catch((error) => {
        this.logger.error(
          `Background direct send failed (${label}): ${(error as Error)?.message}`,
          (error as Error)?.stack,
        );
      })
      .finally(() => {
        this.pendingBackgroundWork.delete(pending);
      });
    this.pendingBackgroundWork.add(pending);
  }

  /**
   * Wait for every in-flight background direct send to settle. Called on shutdown; also what
   * tests await before asserting on a direct-send fallback.
   */
  async whenBackgroundWorkSettles(): Promise<void> {
    while (this.pendingBackgroundWork.size > 0) {
      // A settling task can start no new work, but awaiting the snapshot rather than the live set
      // keeps this correct even if that ever changes.
      await Promise.all([...this.pendingBackgroundWork]);
    }
  }

  async onModuleDestroy(): Promise<void> {
    if (this.pendingBackgroundWork.size === 0) {
      return;
    }
    this.logger.log(
      `Waiting for ${this.pendingBackgroundWork.size} in-flight direct communication send(s) before shutdown`,
    );
    await this.whenBackgroundWorkSettles();
  }

  /**
   * Record the outcome of a completed send as an append-only status row, keyed by program +
   * session + purpose (both channels aggregated). TRIGGERED when at least one recipient was
   * dispatched on any channel; otherwise SKIPPED. Best-effort — a status-write failure is
   * logged but never fails the send that already went out.
   */
  private async recordSendStatus(
    context: SendContext,
    purpose: SessionCommunicationPurposeEnum,
    occurrence: SessionOccurrence | null,
    result: SendResult,
    createdBy: number,
  ): Promise<void> {
    const dispatched = result.enqueued.email + result.enqueued.whatsapp;
    try {
      await this.repository.recordSendStatus({
        programId: context.programId,
        sessionId: context.sessionId,
        purpose,
        occurrence,
        status:
          dispatched > 0
            ? SessionCommunicationStatusEnum.TRIGGERED
            : SessionCommunicationStatusEnum.SKIPPED,
        requestedCount: result.requested,
        emailEnqueuedCount: result.enqueued.email,
        whatsappEnqueuedCount: result.enqueued.whatsapp,
        skippedCount: result.skipped.length,
        createdBy: createdBy || null,
      });
    } catch (error) {
      this.logger.error(
        `Failed to record session communication status for program ${context.programId} (session ${context.sessionId ?? 'n/a'}, purpose ${purpose}): ${error?.message}`,
        error?.stack,
      );
    }
  }

  /**
   * Build the send context + occurrence for a request. Program-level purposes
   * (Welcome/Program completion) resolve the program directly and require no session id;
   * session-scoped purposes load the online session and derive its occurrence.
   */
  private async resolveSendContext(dto: {
    programId: number;
    sessionId?: number;
    purpose: SessionCommunicationPurposeEnum;
  }): Promise<{ context: SendContext; occurrence: SessionOccurrence | null }> {
    if (isProgramLevelPurpose(dto.purpose)) {
      return { context: await this.buildProgramContext(dto.programId), occurrence: null };
    }
    if (!dto.sessionId) {
      throw new InifniBadRequestException(ERROR_CODES.SESSION_COMMUNICATION_IDS_REQUIRED);
    }
    const session = await this.loadOnlineSession(dto.sessionId);
    const occurrence = await this.resolveOccurrence(session, dto.purpose);
    return { context: this.buildSessionContext(session), occurrence };
  }

  /**
   * Program-level context: sender identity from the program, no target session.
   */
  private async buildProgramContext(programId: number): Promise<SendContext> {
    const program = await this.programRepo.findOne({
      where: { id: programId },
      relations: ['type'],
    });
    if (!program) {
      throw new InifniNotFoundException(
        ERROR_CODES.PROGRAM_NOTFOUND,
        null,
        null,
        String(programId),
      );
    }
    return {
      programId,
      sessionId: null,
      programTypeKey: program.type?.key ?? null,
      emailSenderAddress: program.emailSenderAddress ?? null,
      emailSenderName: program.emailSenderName ?? null,
    };
  }

  /**
   * Session-scoped context: sender identity + target session from the loaded session.
   */
  private buildSessionContext(session: ProgramSession): SendContext {
    return {
      programId: session.programId,
      sessionId: session.id,
      programTypeKey: session.program?.type?.key ?? null,
      emailSenderAddress: session.emailSenderAddress ?? null,
      emailSenderName: session.emailSenderName ?? null,
    };
  }

  /**
   * Auto-trigger entry point (from the EventBridge-scheduled processor) for the two
   * program-level communications (Welcome, Program completion). These target the whole
   * program — no session id — so the send goes to the entire eligible audience via sendBulk.
   */
  async triggerProgramCommunication(
    programId: number,
    purpose: SessionCommunicationPurposeEnum,
    createdBy: number,
  ): Promise<SendResult | void> {
    try {
      return await this.sendBulk(
        {
          programId,
          purpose,
          selectionMode: BulkCommunicationSelectionModeEnum.ALL,
        } as SendBulkSessionCommunicationDto,
        createdBy,
      );
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SEND_FAILED, error);
    }
  }

  /**
   * Per-registration counts grouped by purpose and channel for a program.
   */
  async getSummary(programId: number, registrationIds?: number[]) {
    try {
      const rows = await this.repository.getSummary(programId, registrationIds);

      // Reshape into a stable nested structure keyed by registration then communication.
      // The communication key distinguishes first/final variants, e.g. "ABSENT_FIRST",
      // "INVITE_FINAL", "VALUE_CARD".
      const byRegistration = new Map<
        number,
        Record<string, { email: number; whatsapp: number; lastSentAt: Date | string | null }>
      >();

      for (const row of rows) {
        let communicationMap = byRegistration.get(row.registrationId);
        if (!communicationMap) {
          communicationMap = {};
          byRegistration.set(row.registrationId, communicationMap);
        }
        const key = row.occurrence ? `${row.purpose}_${row.occurrence}` : row.purpose;
        if (!communicationMap[key]) {
          communicationMap[key] = { email: 0, whatsapp: 0, lastSentAt: null };
        }
        const bucket = communicationMap[key];
        if (row.channel === CommunicationTypeEnum.EMAIL) {
          bucket.email = row.count;
        } else if (row.channel === CommunicationTypeEnum.WHATSAPP) {
          bucket.whatsapp = row.count;
        }
        if (!bucket.lastSentAt || new Date(row.lastSentAt) > new Date(bucket.lastSentAt)) {
          bucket.lastSentAt = row.lastSentAt;
        }
      }

      return Array.from(byRegistration.entries()).map(([registrationId, byCommunication]) => ({
        registrationId,
        byCommunication,
      }));
    } catch (error) {
      handleKnownErrors(ERROR_CODES.SESSION_COMMUNICATION_SUMMARY_FAILED, error);
    }
  }

  /**
   * Which session-scoped purposes actually went out for each of the given sessions —
   * the Overall Analytics timeline's "comms checklist" source. Thin passthrough to the
   * repository's bulk query.
   */
  async getTriggeredPurposesForSessions(
    programId: number,
    sessionIds: number[],
  ): Promise<Map<number, SessionCommunicationPurposeEnum[]>> {
    return this.repository.findTriggeredPurposesBySessions(programId, sessionIds);
  }

  /**
   * Latest outcome of one session-scoped bulk send, as the summary shape other domains surface on
   * their GET responses (the general-attendees analytics response uses this for the general-link
   * Value Card). null when that purpose has never run for the session.
   */
  async getLatestSessionStatus(
    sessionId: number,
    purpose: SessionCommunicationPurposeEnum,
  ): Promise<SessionCommunicationStatusSummary | null> {
    const row = await this.repository.findLatestSessionStatus(sessionId, purpose);
    if (!row) {
      return null;
    }
    return {
      status: row.status,
      requested: row.requestedCount,
      emailSent: row.emailEnqueuedCount,
      whatsappSent: row.whatsappEnqueuedCount,
      skipped: row.skippedCount,
      lastTriggeredAt: row.createdAt,
    };
  }

  /**
   * Step 2 — filter/get the registrations for the communication. Each purpose has its own
   * recipient function so its audience rule (incl. future attendance-based narrowing) is
   * isolated from the rest of the flow.
   */
  private filterRecipients(
    dto: SendBulkSessionCommunicationDto,
    context: SendContext,
    occurrence: SessionOccurrence | null,
  ): Promise<SessionCommunicationRecipient[]> {
    const { programId, sessionId, selectionMode, registrationIds } = dto;
    switch (dto.purpose) {
      case SessionCommunicationPurposeEnum.WELCOME:
        return this.repository.getWelcomeRecipients(programId, selectionMode, registrationIds);
      case SessionCommunicationPurposeEnum.INVITE:
        // The FINAL-session invite of a multi-session TAT program (occurrence === FINAL) goes only
        // to seekers who attended EVERY earlier session — anyone absent from a prior session is
        // dropped. Every other case (regular session, single-session or non-TAT program) invites
        // the full eligible set.
        if (occurrence === SessionOccurrence.FINAL && sessionId) {
          return this.repository.getFinalInviteRecipients(
            programId,
            Number(sessionId),
            selectionMode,
            registrationIds,
          );
        }
        return this.repository.getInviteRecipients(
          programId,
          Number(sessionId),
          selectionMode,
          registrationIds,
        );
      case SessionCommunicationPurposeEnum.ABSENT:
        // Absentees only: eligible registrations who did NOT attend the target session.
        // ABSENT is session-scoped, so sessionId is guaranteed present (validated upstream).
        return this.repository.getAbsentRecipients(
          programId,
          Number(sessionId),
          selectionMode,
          registrationIds,
        );
      case SessionCommunicationPurposeEnum.VALUE_CARD:
        // Attendees only: value card recaps a session the seeker was present for.
        // VALUE_CARD is session-scoped, so sessionId is guaranteed present (validated upstream).
        return this.repository.getValueCardRecipients(
          programId,
          Number(sessionId),
          selectionMode,
          registrationIds,
        );
      case SessionCommunicationPurposeEnum.PROGRAM_COMPLETION:
        return this.filterProgramCompletionRecipients(dto, context);
      default:
        return Promise.resolve([]);
    }
  }

  /**
   * Program-completion audience by program type:
   * - Normal programs: every eligible registration (seat allocated + pending/completed).
   * - TAT programs: only registrants who attended the FINAL session. With no final session
   *   there is no attendance to gate on, so nobody qualifies.
   */
  private async filterProgramCompletionRecipients(
    dto: SendBulkSessionCommunicationDto,
    context: SendContext,
  ): Promise<SessionCommunicationRecipient[]> {
    const { programId, selectionMode, registrationIds } = dto;
    if (context.programTypeKey !== PROGRAM_TYPE_KEYS.TAT) {
      return this.repository.getProgramCompletionRecipients(
        programId,
        selectionMode,
        registrationIds,
      );
    }
    const finalSessionId = await this.resolveFinalSessionId(programId);
    if (!finalSessionId) {
      return [];
    }
    return this.repository.getProgramCompletionRecipients(
      programId,
      selectionMode,
      registrationIds,
      finalSessionId,
    );
  }

  /**
   * The program's final session id — the last session ordered chronologically
   * (startsAt, then displayOrder, then id), matching the FINAL-occurrence resolution.
   */
  private async resolveFinalSessionId(programId: number): Promise<number | null> {
    const sessions = await this.sessionRepo.find({
      where: { programId },
      order: { startsAt: 'ASC', displayOrder: 'ASC', id: 'ASC' },
      select: ['id'],
    });
    const last = sessions[sessions.length - 1];
    return last ? Number(last.id) : null;
  }

  /**
   * Step 1 — fetch the communication template id per channel for this purpose/occurrence.
   * The template id is constant for the program and is later recorded on each track row.
   * templateId is null when no template row is configured (the send is skipped downstream).
   */
  private async fetchChannelTemplates(
    programId: number,
    purpose: SessionCommunicationPurposeEnum,
    occurrence: SessionOccurrence | null,
  ): Promise<ChannelTemplate[]> {
    return this.fetchTemplatesForChannels(programId, resolveChannels(purpose, occurrence));
  }

  /**
   * Resolve the program's configured template id for each of the given channel/access-key pairs.
   * Shared by the purpose/occurrence path (fetchChannelTemplates) and callers that select access
   * keys directly (e.g. the common invite, which picks its program-level vs per-session pair).
   */
  private async fetchTemplatesForChannels(
    programId: number,
    channels: ChannelConfig[],
  ): Promise<ChannelTemplate[]> {
    const channelTemplates: ChannelTemplate[] = [];

    for (const cfg of channels) {
      const templateRow = await this.templatesRepository.findByProgramAndAccessKey(
        programId,
        cfg.accessKey,
        cfg.channel,
      );
      channelTemplates.push({
        ...cfg,
        templateId: templateRow?.id ? Number(templateRow.id) : null,
      });
    }

    return channelTemplates;
  }

  /**
   * Step 3 — common processing: for each resolved channel-template, resolve per-recipient
   * merge info and enqueue (email as a batch, whatsapp per recipient). The template id is
   * threaded through the queue metadata so the processor records it on the shared
   * hdb_communication_track row. Never throws on a single bad recipient — records a skip.
   */
  private async processCommunications(
    context: SendContext,
    channelTemplates: ChannelTemplate[],
    recipients: SessionCommunicationRecipient[],
    purpose: SessionCommunicationPurposeEnum,
    createdBy: number,
    isBulk: boolean,
  ): Promise<SendResult> {
    const result: SendResult = {
      requested: recipients.length,
      enqueued: { email: 0, whatsapp: 0 },
      skipped: [],
    };

    for (const channelTemplate of channelTemplates) {
      const channelEnabled =
        channelTemplate.channel === CommunicationTypeEnum.EMAIL
          ? this.emailQueueService.isQueueEnabled()
          : this.whatsAppQueueService.isQueueEnabled();

      // Bulk sends use the bulk provider methods (batch email / WATI bulk); single sends use the
      // single methods. Both prefer the queue and fall back to a direct send.
      if (channelTemplate.channel === CommunicationTypeEnum.EMAIL) {
        if (isBulk) {
          await this.dispatchEmailBatch(
            context,
            recipients,
            purpose,
            channelTemplate,
            channelEnabled,
            createdBy,
            result,
          );
        } else {
          await this.dispatchEmailSingle(
            context,
            recipients,
            purpose,
            channelTemplate,
            channelEnabled,
            createdBy,
            result,
          );
        }
      } else if (isBulk) {
        await this.dispatchWhatsAppBulk(
          context,
          recipients,
          purpose,
          channelTemplate,
          channelEnabled,
          createdBy,
          result,
        );
      } else {
        await this.dispatchWhatsApp(
          context,
          recipients,
          purpose,
          channelTemplate,
          channelEnabled,
          createdBy,
          result,
        );
      }
    }

    this.logger.log(
      `Session communication dispatched: purpose=${purpose}, requested=${result.requested}, ` +
        `email=${result.enqueued.email}, whatsapp=${result.enqueued.whatsapp}, skipped=${result.skipped.length}`,
    );
    return result;
  }

  /**
   * Resolve a recipient for a channel: validates contact presence and template/merge
   * availability, recording a skip and returning null on any miss. Independent of the
   * queue state — the dispatcher decides whether to enqueue or send directly.
   */
  private async resolveForSend(
    context: SendContext,
    cfg: ChannelConfig,
    recipient: SessionCommunicationRecipient,
    result: SendResult,
    /**
     * One per (send, channel) — see SendMergeCache. Lets the template row, its merge-field map and
     * every send-constant (`isCommon`) field resolve once for the whole batch instead of once per
     * recipient. Omitted by single sends, where there is nothing to amortise.
     */
    cache?: SendMergeCache,
  ): Promise<{ contact: string; templateKey: string; mergeInfo: Record<string, any> } | null> {
    const contact =
      cfg.channel === CommunicationTypeEnum.EMAIL ? recipient.emailAddress : recipient.mobileNumber;

    if (!contact) {
      result.skipped.push({
        registrationId: recipient.id,
        channel: cfg.channel,
        reason: 'recipient has no contact for this channel',
      });
      return null;
    }

    // The target session is threaded through extraMergeContext so occurrence-dependent
    // templates (Invite/Absent) resolve THIS session's meeting/date fields rather than the
    // program's first session. Program-level sends (Welcome/Program completion) carry no
    // sessionId, so their program-level merge fields resolve without a target session.
    // extraContext carries request-supplied values (e.g. Value Card `description`).
    // `manager` is left to default (the merge service's own).
    const template = await this.mergeDataService.getTemplateWithMergeInfo(
      context.programId,
      cfg.accessKey,
      cfg.channel,
      recipient.id,
      undefined,
      {
        ...(context.sessionId ? { sessionId: context.sessionId } : {}),
        ...(context.extraContext || {}),
      },
      cache ? { cache } : undefined,
    );

    if (!template?.templateKey) {
      result.skipped.push({
        registrationId: recipient.id,
        channel: cfg.channel,
        reason: 'template not configured for program',
      });
      return null;
    }

    return { contact, templateKey: template.templateKey, mergeInfo: template.mergeInfo || {} };
  }

  /**
   * Email channel: resolve every recipient into a single provider-side batch
   * (ZeptoMail /batch). When the communication queue is enabled the batch is enqueued
   * (the bulk processor sends + tracks it); when the queue is disabled — or an enqueue
   * fails/queues nothing — it falls back to sending the batch DIRECTLY via
   * CommunicationService.sendBulkEmail (which writes the same track rows). Per-recipient
   * tracking uses the templateId + registrationId either way.
   */
  private async dispatchEmailBatch(
    context: SendContext,
    recipients: SessionCommunicationRecipient[],
    purpose: SessionCommunicationPurposeEnum,
    channelTemplate: ChannelTemplate,
    channelEnabled: boolean,
    createdBy: number,
    result: SendResult,
  ): Promise<void> {
    const batch: Array<{
      emailAddress: string;
      name: string;
      templateData: Record<string, any>;
      registrationId: number;
    }> = [];
    let templateKey: string | null = null;
    // One cache for the whole batch: the template, its merge map and every send-constant field
    // (session name/date/time, meeting details, portal link, guidelines PDF + its signed URL)
    // resolve on the first recipient and are reused for the rest.
    const cache = createSendMergeCache();

    for (const recipient of recipients) {
      const prepared = await this.resolveForSend(
        context,
        channelTemplate,
        recipient,
        result,
        cache,
      );
      if (!prepared) {
        continue;
      }
      templateKey = prepared.templateKey;
      batch.push({
        emailAddress: prepared.contact,
        name: recipient.fullName || '',
        templateData: prepared.mergeInfo,
        registrationId: recipient.id,
      });
    }

    if (batch.length === 0 || !templateKey) {
      return;
    }

    // Prefer the queue; fall back to a direct send when it's disabled or the enqueue
    // fails (queueBulkEmail swallows SQS errors and returns [], so treat "no message ids"
    // as a failed enqueue too).
    //
    // Attachments travel as references only — see toQueueAttachments, which returns null when some
    // attachment has no S3 key and the batch therefore has to go direct. That is a fact about the
    // attachments, not a mismatch between context fields.
    const queueAttachments = context.attachments?.length
      ? this.toQueueAttachments(context.attachments)
      : undefined;
    const canEnqueue = channelEnabled && queueAttachments !== null;
    if (
      canEnqueue &&
      (await this.tryEnqueueEmailBatch(
        context,
        purpose,
        channelTemplate,
        templateKey,
        batch,
        createdBy,
        queueAttachments ?? undefined,
      ))
    ) {
      result.enqueued.email += batch.length;
      return;
    }

    // Queue unavailable — send directly, but OFF the request (see runInBackground), so the caller
    // returns just as promptly as on the queued path. Counted as dispatched here for the same
    // reason the queued path counts on a queued message id: both mean "accepted". Resolving the
    // attachment bytes moves into the task too — it's an S3 download per file, and only the
    // fallback needs it at all.
    const directTemplateKey = templateKey;
    result.enqueued.email += batch.length;
    this.runInBackground(
      `bulk email, program ${context.programId} (session ${context.sessionId ?? 'n/a'}), ${batch.length} recipient(s)`,
      async () => {
        await this.directSendEmailBatch(
          context,
          channelTemplate,
          directTemplateKey,
          batch,
          createdBy,
          context.attachments?.length
            ? await this.resolveDirectAttachments(context.attachments)
            : undefined,
        );
      },
    );
  }

  /**
   * Enqueue the email batch. Returns true only when at least one bulk-email message was
   * queued; false on a disabled/failed enqueue so the caller can fall back to a direct send.
   */
  private async tryEnqueueEmailBatch(
    context: SendContext,
    purpose: SessionCommunicationPurposeEnum,
    channelTemplate: ChannelTemplate,
    templateKey: string,
    batch: Array<{
      emailAddress: string;
      name: string;
      templateData: Record<string, any>;
      registrationId: number;
    }>,
    createdBy: number,
    queueAttachments?: QueuedAttachment[],
  ): Promise<boolean> {
    try {
      const messageIds = await this.emailQueueService.queueBulkEmail({
        from: {
          address: context.emailSenderAddress || zeptoEmailCreadentials.ZEPTO_EMAIL,
          name: context.emailSenderName || zeptoEmailCreadentials.ZEPTO_EMAIL_NAME,
        },
        subject: '',
        templateKey,
        recipients: batch,
        // References only (guaranteed by toQueueAttachments) — BulkEmailProcessor downloads them
        // once per batch before sending.
        attachments: queueAttachments?.length
          ? queueAttachments.map(({ name, contentType, s3Key }) => ({
              name,
              contentType,
              s3Key: s3Key as string,
            }))
          : undefined,
        templateId: channelTemplate.templateId,
        createdBy,
        userId: createdBy ? String(createdBy) : undefined,
        metadata: { sessionId: context.sessionId ?? undefined, purpose },
      });
      return (messageIds?.length ?? 0) > 0;
    } catch (error) {
      this.logger.error(
        `Failed to enqueue bulk session email for program ${context.programId} (session ${context.sessionId ?? 'n/a'}): ${error?.message}`,
        error?.stack,
      );
      return false;
    }
  }

  /**
   * Direct fallback: send the batch in one provider call (ZeptoMail /batch) via
   * CommunicationService.sendBulkEmail, which writes the per-recipient track rows using
   * the same templateId/registrationId the queue path would. A failure skips the batch.
   */
  private async directSendEmailBatch(
    context: SendContext,
    channelTemplate: ChannelTemplate,
    templateKey: string,
    batch: Array<{
      emailAddress: string;
      name: string;
      templateData: Record<string, any>;
      registrationId: number;
    }>,
    createdBy: number,
    /** Already resolved to bytes by the caller — see dispatchEmailBatch's memoized resolver. */
    directAttachments?: EmailAttachments[],
  ): Promise<void> {
    const dto: SendBulkEmailDto = {
      to: batch.map((recipient) => ({
        emailAddress: recipient.emailAddress,
        name: recipient.name,
        mergeInfo: recipient.templateData,
        registrationId: recipient.registrationId,
      })),
      from: {
        address: context.emailSenderAddress || zeptoEmailCreadentials.ZEPTO_EMAIL,
        name: context.emailSenderName || zeptoEmailCreadentials.ZEPTO_EMAIL_NAME,
      },
      subject: '',
      templateKey,
      attachments: directAttachments,
      trackinfo: {
        templateId: channelTemplate.templateId ?? undefined,
        createdBy,
        updatedBy: createdBy,
      },
    };

    try {
      await this.communicationService.sendBulkEmail(dto);
      this.logger.log(
        `Sent session email batch directly (queue fallback) for program ${context.programId} (session ${context.sessionId ?? 'n/a'}): ${batch.length} recipient(s)`,
      );
    } catch (error) {
      this.logger.error(
        `Failed to direct-send bulk session email for program ${context.programId} (session ${context.sessionId ?? 'n/a'}): ${error?.message}`,
        error?.stack,
      );
    }
  }

  /**
   * Email channel, SINGLE send: resolve the (single) recipient and send via the single-email
   * method — queued (queueEmail → EmailProcessor → sendSingleEmail) when the queue is enabled,
   * else a direct sendSingleEmail. Both write the per-recipient track row.
   */
  private async dispatchEmailSingle(
    context: SendContext,
    recipients: SessionCommunicationRecipient[],
    purpose: SessionCommunicationPurposeEnum,
    channelTemplate: ChannelTemplate,
    channelEnabled: boolean,
    createdBy: number,
    result: SendResult,
  ): Promise<void> {
    // Queue messages carry references only (see toQueueAttachments), so a send whose attachments
    // aren't all in S3 has to go direct — the same rule, and the same one-liner, as the bulk
    // dispatcher below.
    const queueAttachments = context.attachments?.length
      ? this.toQueueAttachments(context.attachments)
      : undefined;
    const canEnqueue = channelEnabled && queueAttachments !== null;

    // Resolved at most once per dispatch, and only if a direct send actually happens — so the
    // queued path downloads nothing.
    let directAttachments: EmailAttachments[] | undefined;
    const directAttachmentsFor = async (): Promise<EmailAttachments[] | undefined> => {
      if (!context.attachments?.length) return undefined;
      directAttachments ??= await this.resolveDirectAttachments(context.attachments);
      return directAttachments;
    };

    for (const recipient of recipients) {
      const prepared = await this.resolveForSend(context, channelTemplate, recipient, result);
      if (!prepared) {
        continue;
      }
      if (
        canEnqueue &&
        (await this.tryEnqueueSingleEmail(
          context,
          purpose,
          channelTemplate,
          prepared,
          recipient,
          createdBy,
          queueAttachments ?? undefined,
        ))
      ) {
        result.enqueued.email += 1;
        continue;
      }
      await this.directSendSingleEmail(
        context,
        channelTemplate,
        prepared,
        recipient,
        createdBy,
        result,
        await directAttachmentsFor(),
      );
    }
  }

  /** Enqueue one single-email message. Returns true only when a message id came back. */
  private async tryEnqueueSingleEmail(
    context: SendContext,
    purpose: SessionCommunicationPurposeEnum,
    channelTemplate: ChannelTemplate,
    prepared: { contact: string; templateKey: string; mergeInfo: Record<string, any> },
    recipient: SessionCommunicationRecipient,
    createdBy: number,
    queueAttachments?: QueuedAttachment[],
  ): Promise<boolean> {
    try {
      const messageId = await this.emailQueueService.queueEmail({
        to: { emailAddress: prepared.contact, name: recipient.fullName || '' },
        from: {
          address: context.emailSenderAddress || zeptoEmailCreadentials.ZEPTO_EMAIL,
          name: context.emailSenderName || zeptoEmailCreadentials.ZEPTO_EMAIL_NAME,
        },
        subject: '',
        templateKey: prepared.templateKey,
        templateData: prepared.mergeInfo,
        // References only, so the message stays small regardless of file size; EmailProcessor
        // downloads each object just before sending.
        attachments: queueAttachments?.length ? queueAttachments : undefined,
        userId: createdBy ? String(createdBy) : undefined,
        metadata: {
          registrationId: recipient.id,
          templateId: channelTemplate.templateId,
          createdBy,
          updatedBy: createdBy,
          sessionId: context.sessionId ?? undefined,
          purpose,
        },
      });
      return !!messageId;
    } catch (error) {
      this.logger.error(
        `Failed to enqueue single session email for registration ${recipient.id}: ${error?.message}`,
        error?.stack,
      );
      return false;
    }
  }

  /** Direct fallback: send one email via CommunicationService.sendSingleEmail (writes the track row). */
  private async directSendSingleEmail(
    context: SendContext,
    channelTemplate: ChannelTemplate,
    prepared: { contact: string; templateKey: string; mergeInfo: Record<string, any> },
    recipient: SessionCommunicationRecipient,
    createdBy: number,
    result: SendResult,
    /** Already resolved to bytes by the caller — see dispatchEmailSingle's memoized resolver. */
    directAttachments?: EmailAttachments[],
  ): Promise<void> {
    try {
      await this.communicationService.sendSingleEmail({
        to: { emailAddress: prepared.contact, name: recipient.fullName || '' },
        from: {
          address: context.emailSenderAddress || zeptoEmailCreadentials.ZEPTO_EMAIL,
          name: context.emailSenderName || zeptoEmailCreadentials.ZEPTO_EMAIL_NAME,
        },
        subject: '',
        templateKey: prepared.templateKey,
        mergeInfo: prepared.mergeInfo,
        attachments: directAttachments,
        trackinfo: {
          registrationId: recipient.id,
          templateId: channelTemplate.templateId ?? undefined,
          createdBy,
          updatedBy: createdBy,
        },
      });
      result.enqueued.email += 1;
    } catch (error) {
      this.logger.error(
        `Failed to direct-send single session email for registration ${recipient.id}: ${error?.message}`,
        error?.stack,
      );
      result.skipped.push({
        registrationId: recipient.id,
        channel: channelTemplate.channel,
        reason: 'failed to send',
      });
    }
  }

  /**
   * WhatsApp channel, BULK send: resolve all recipients, then dispatch as a WATI bulk template
   * send — enqueued (queueBulkWhatsApp → BulkWhatsAppProcessor → sendBulkTemplateMessage) when the
   * queue is enabled, else a direct sendBulkTemplateMessage. The queue path chunks the recipients
   * into several messages (QUEUE_CONSTANTS.BULK_WHATSAPP_BATCH_SIZE), like the email batch does.
   * Both write per-recipient track rows via the registrationId + templateId.
   */
  private async dispatchWhatsAppBulk(
    context: SendContext,
    recipients: SessionCommunicationRecipient[],
    purpose: SessionCommunicationPurposeEnum,
    channelTemplate: ChannelTemplate,
    channelEnabled: boolean,
    createdBy: number,
    result: SendResult,
  ): Promise<void> {
    const batch: Array<{
      whatsappNumber: string;
      customParams: Array<{ name: string; value: string }>;
      registrationId: number;
    }> = [];
    let templateKey: string | null = null;
    // Same per-batch cache as the email path — see dispatchEmailBatch.
    const cache = createSendMergeCache();

    for (const recipient of recipients) {
      const prepared = await this.resolveForSend(
        context,
        channelTemplate,
        recipient,
        result,
        cache,
      );
      if (!prepared) {
        continue;
      }
      templateKey = prepared.templateKey;
      batch.push({
        whatsappNumber: prepared.contact,
        customParams: Object.entries(prepared.mergeInfo).map(([name, value]) => ({
          name,
          value: String(value),
        })),
        registrationId: recipient.id,
      });
    }

    if (batch.length === 0 || !templateKey) {
      return;
    }

    if (
      channelEnabled &&
      (await this.tryEnqueueBulkWhatsApp(
        context,
        purpose,
        channelTemplate,
        templateKey,
        batch,
        createdBy,
      ))
    ) {
      result.enqueued.whatsapp += batch.length;
      return;
    }

    // Direct fallback, off-request — same reasoning as the email batch above. This is the path
    // where waiting hurt most: sendBulkTemplateMessage walks the recipients 10 at a time with a
    // 2s pause between calls.
    const directTemplateKey = templateKey;
    result.enqueued.whatsapp += batch.length;
    this.runInBackground(
      `bulk whatsapp, program ${context.programId} (session ${context.sessionId ?? 'n/a'}), ${batch.length} recipient(s)`,
      () => this.directSendBulkWhatsApp(channelTemplate, directTemplateKey, batch, createdBy),
    );
  }

  /**
   * Enqueue the batch as bulk-whatsapp messages (chunked by the queue service, exactly like the
   * email batch). Returns true when at least one chunk was queued; false on a disabled/failed
   * enqueue so the caller can fall back to a direct send.
   */
  private async tryEnqueueBulkWhatsApp(
    context: SendContext,
    purpose: SessionCommunicationPurposeEnum,
    channelTemplate: ChannelTemplate,
    templateKey: string,
    batch: Array<{
      whatsappNumber: string;
      customParams: Array<{ name: string; value: string }>;
      registrationId: number;
    }>,
    createdBy: number,
  ): Promise<boolean> {
    try {
      const messageIds = await this.whatsAppQueueService.queueBulkWhatsApp({
        templateName: templateKey,
        broadcastName: templateKey,
        recipients: batch,
        templateId: channelTemplate.templateId,
        createdBy,
        userId: createdBy ? String(createdBy) : undefined,
        metadata: { sessionId: context.sessionId ?? undefined, purpose },
      });
      return (messageIds?.length ?? 0) > 0;
    } catch (error) {
      this.logger.error(
        `Failed to enqueue bulk session whatsapp for program ${context.programId}: ${(error as Error)?.message}`,
        (error as Error)?.stack,
      );
      return false;
    }
  }

  /** Direct fallback: WATI bulk template send via CommunicationService.sendBulkTemplateMessage. */
  private async directSendBulkWhatsApp(
    channelTemplate: ChannelTemplate,
    templateKey: string,
    batch: Array<{
      whatsappNumber: string;
      customParams: Array<{ name: string; value: string }>;
      registrationId: number;
    }>,
    createdBy: number,
  ): Promise<void> {
    try {
      await this.communicationService.sendBulkTemplateMessage({
        recipients: batch,
        templateName: templateKey,
        broadcastName: templateKey,
        trackinfo: {
          templateId: channelTemplate.templateId ?? undefined,
          createdBy,
          updatedBy: createdBy,
        },
      });
      this.logger.log(
        `Sent session whatsapp bulk directly (queue fallback): ${batch.length} recipient(s)`,
      );
    } catch (error) {
      this.logger.error(
        `Failed to direct-send bulk session whatsapp: ${(error as Error)?.message}`,
        (error as Error)?.stack,
      );
    }
  }

  /**
   * WhatsApp channel, SINGLE send: WATI has no batch API, so send one message per recipient.
   */
  private async dispatchWhatsApp(
    context: SendContext,
    recipients: SessionCommunicationRecipient[],
    purpose: SessionCommunicationPurposeEnum,
    channelTemplate: ChannelTemplate,
    channelEnabled: boolean,
    createdBy: number,
    result: SendResult,
  ): Promise<void> {
    // Count of direct WATI sends so far — used to throttle the fallback path (each direct send
    // is a live WATI call; the queue path is paced by SQS polling and needs no delay).
    let directSentCount = 0;
    for (const recipient of recipients) {
      const prepared = await this.resolveForSend(context, channelTemplate, recipient, result);
      if (!prepared) {
        continue;
      }

      const parameters = Object.entries(prepared.mergeInfo).map(([name, value]) => ({
        name,
        value: String(value),
      }));

      // Prefer the queue; fall back to a direct WATI send when it's disabled or the
      // enqueue fails/returns no id.
      const enqueued =
        channelEnabled &&
        (await this.tryEnqueueWhatsApp(
          context,
          recipient,
          purpose,
          channelTemplate,
          prepared,
          parameters,
          createdBy,
        ));
      if (enqueued) {
        result.enqueued.whatsapp += 1;
        continue;
      }

      // Direct fallback: WATI is one call per recipient (no batch API), so throttle between
      // calls with the shared bulk delay to respect rate limits. Per-recipient send keeps the
      // communication-track row the summary/status rely on.
      if (directSentCount > 0) {
        await delay(WHATSAPP_BULK_SETTINGS.DELAY_BETWEEN_REQUESTS);
      }
      await this.directSendWhatsApp(
        recipient,
        channelTemplate,
        prepared,
        parameters,
        createdBy,
        result,
      );
      directSentCount += 1;
    }
  }

  /**
   * Enqueue one WhatsApp message. Returns true only when a message id came back; false on a
   * disabled/failed enqueue so the caller can fall back to a direct send.
   */
  private async tryEnqueueWhatsApp(
    context: SendContext,
    recipient: SessionCommunicationRecipient,
    purpose: SessionCommunicationPurposeEnum,
    channelTemplate: ChannelTemplate,
    prepared: { contact: string; templateKey: string; mergeInfo: Record<string, any> },
    parameters: Array<{ name: string; value: string }>,
    createdBy: number,
  ): Promise<boolean> {
    try {
      const messageId = await this.whatsAppQueueService.queueWhatsAppMessage({
        phoneNumber: prepared.contact,
        templateName: prepared.templateKey,
        broadcastName: prepared.templateKey,
        parameters,
        metadata: {
          registrationId: recipient.id,
          sessionId: context.sessionId ?? undefined,
          purpose,
          templateId: channelTemplate.templateId,
          createdBy,
          updatedBy: createdBy,
        },
        userId: createdBy ? String(createdBy) : undefined,
      });
      return !!messageId;
    } catch (error) {
      this.logger.error(
        `Failed to enqueue whatsapp session communication for registration ${recipient.id} (program ${context.programId}): ${(error as Error)?.message}`,
        (error as Error)?.stack,
      );
      return false;
    }
  }

  /**
   * Direct fallback: send one WATI template message via CommunicationService, which writes
   * the track row using the same registrationId/templateId the queue path would. A failure
   * skips the recipient.
   */
  private async directSendWhatsApp(
    recipient: SessionCommunicationRecipient,
    channelTemplate: ChannelTemplate,
    prepared: { contact: string; templateKey: string; mergeInfo: Record<string, any> },
    parameters: Array<{ name: string; value: string }>,
    createdBy: number,
    result: SendResult,
  ): Promise<void> {
    const dto: SendTemplateMessageDto = {
      whatsappNumber: prepared.contact,
      templateName: prepared.templateKey,
      broadcastName: prepared.templateKey,
      parameters,
      trackinfo: {
        registrationId: recipient.id,
        templateId: channelTemplate.templateId ?? undefined,
        createdBy,
        updatedBy: createdBy,
      },
    };

    try {
      await this.communicationService.sendTemplateMessage(dto);
      result.enqueued.whatsapp += 1;
      this.logger.log(
        `Sent session whatsapp directly (queue fallback) for registration ${recipient.id}`,
      );
    } catch (error) {
      this.logger.error(
        `Failed to direct-send whatsapp session communication for registration ${recipient.id}: ${(error as Error)?.message}`,
        (error as Error)?.stack,
      );
      result.skipped.push({
        registrationId: recipient.id,
        channel: channelTemplate.channel,
        reason: 'failed to send',
      });
    }
  }

  // ==================================================================================
  // Common invite (program-level, generated-link staff recipients) dispatch.
  //
  // Kept separate from processCommunications because the recipients are
  // zoom_generated_registrant_link rows (no ProgramRegistration) and the merge values
  // (common_user_name / common_zoom_join_link / common_meeting_id / common_meeting_passcode)
  // vary per recipient and are supplied via extraMergeContext rather than resolved from a
  // registration. Both channels use the bulk provider methods (batch email / WATI bulk).
  // ==================================================================================

  /**
   * Fire each configured channel for the common invite. Mirrors processCommunications but
   * over CommonInviteRecipient (no registrationId threaded into tracking).
   */
  private async processCommonInvite(
    context: SendContext,
    channelTemplates: ChannelTemplate[],
    recipients: CommonInviteRecipient[],
    purpose: SessionCommunicationPurposeEnum,
    createdBy: number,
  ): Promise<SendResult> {
    const result: SendResult = {
      requested: recipients.length,
      enqueued: { email: 0, whatsapp: 0 },
      skipped: [],
    };

    for (const channelTemplate of channelTemplates) {
      const channelEnabled =
        channelTemplate.channel === CommunicationTypeEnum.EMAIL
          ? this.emailQueueService.isQueueEnabled()
          : this.whatsAppQueueService.isQueueEnabled();
      this.logger.log(
        `Dispatching common invite: program=${context.programId}, channel=${channelTemplate.channel}, ` +
          `templateId=${channelTemplate.templateId}, enabled=${channelEnabled}, recipients=${JSON.stringify(recipients)}`,
      );
      if (channelTemplate.channel === CommunicationTypeEnum.EMAIL) {
        await this.dispatchCommonInviteEmail(
          context,
          recipients,
          purpose,
          channelTemplate,
          channelEnabled,
          createdBy,
          result,
        );
      } else {
        this.logger.log(
          `Dispatching common invite WhatsApp: program=${context.programId}, channel=${channelTemplate.channel}, ` +
            `templateId=${channelTemplate.templateId}, enabled=${channelEnabled}, requested=${recipients.length}`,
        );
        await this.dispatchCommonInviteWhatsApp(
          context,
          recipients,
          purpose,
          channelTemplate,
          channelEnabled,
          createdBy,
          result,
        );
      }
    }

    this.logger.log(
      `Common invite dispatched: program=${context.programId}, requested=${result.requested}, ` +
        `email=${result.enqueued.email}, whatsapp=${result.enqueued.whatsapp}, skipped=${result.skipped.length}`,
    );
    return result;
  }

  /**
   * Resolve one program-level (registration-less) recipient for a channel: validates the channel
   * contact and resolves the template + merge info, with the caller-supplied per-recipient values
   * passed through extraMergeContext. Records a skip and returns null on any miss. No
   * registrationId — every field resolves via the program-level (common) path regardless of each
   * row's is_common flag. Shared by Common Invite and System Links.
   */
  private async resolveProgramLevelSend(
    context: SendContext,
    cfg: ChannelConfig,
    recipient: CommonInviteRecipient,
    extraMergeContext: Record<string, any>,
    result: SendResult,
  ): Promise<{ contact: string; templateKey: string; mergeInfo: Record<string, any> } | null> {
    const contact =
      cfg.channel === CommunicationTypeEnum.EMAIL ? recipient.emailAddress : recipient.mobileNumber;

    if (!contact) {
      result.skipped.push({
        registrationId: recipient.userId ?? 0,
        channel: cfg.channel,
        reason: 'recipient has no contact for this channel',
      });
      return null;
    }

    const template = await this.mergeDataService.getTemplateWithMergeInfo(
      context.programId,
      cfg.accessKey,
      cfg.channel,
      undefined,
      undefined,
      extraMergeContext,
      // These recipients have no registration — resolve every field via the program-level
      // (common) path regardless of each row's is_common flag.
      { resolveAllFieldsAsCommon: true },
    );

    if (!template?.templateKey) {
      result.skipped.push({
        registrationId: recipient.userId ?? 0,
        channel: cfg.channel,
        reason: 'template not configured for program',
      });
      return null;
    }

    return { contact, templateKey: template.templateKey, mergeInfo: template.mergeInfo || {} };
  }

  /**
   * Per-recipient merge context for a Common Invite send (link + meeting details vary per staff
   * member). When the send is scoped to one session (per-session variant), sessionId is included
   * so the target-session merge fields (session_name / session_date / session_time) resolve that
   * session rather than falling back to the program's first session.
   */
  private commonInviteMergeContext(
    recipient: CommonInviteRecipient,
    sessionId?: number | null,
    extraContext?: Record<string, any>,
  ): Record<string, any> {
    return {
      common_user_name: recipient.displayName ?? '',
      common_zoom_join_link: recipient.joinUrl ?? '',
      common_meeting_id: recipient.meetingId ?? '',
      common_meeting_passcode: recipient.meetingPasscode ?? '',
      // Lets registration-less merge fields that key off email rather than registrationId resolve
      // for this recipient — e.g. resolveAbsentSessions' zoom_analytics_attendee_summary fallback
      // for GENERAL_LINK_ABSENT's absentDataTabel/absent_session content.
      common_user_email: recipient.emailAddress ?? '',
      ...(sessionId != null ? { sessionId } : {}),
      // Request-supplied values (the general-link Value Card's `description`). Spread last so the
      // caller can also override a derived field if it ever needs to. Without this, a purpose whose
      // template takes an admin-entered field would render it empty on this path — the seeker
      // pipeline threads extraContext through resolveForSend, and this one has to do the same.
      ...(extraContext ?? {}),
    };
  }

  /**
   * Common invite, EMAIL: resolve every recipient into one ZeptoMail /batch send — enqueued
   * (bulk-email queue) when enabled, else sent directly. No per-recipient registrationId.
   */
  private async dispatchCommonInviteEmail(
    context: SendContext,
    recipients: CommonInviteRecipient[],
    purpose: SessionCommunicationPurposeEnum,
    channelTemplate: ChannelTemplate,
    channelEnabled: boolean,
    createdBy: number,
    result: SendResult,
  ): Promise<void> {
    const batch: Array<{ emailAddress: string; name: string; templateData: Record<string, any> }> =
      [];
    let templateKey: string | null = null;

    for (const recipient of recipients) {
      const prepared = await this.resolveProgramLevelSend(
        context,
        channelTemplate,
        recipient,
        this.commonInviteMergeContext(recipient, context.sessionId, context.extraContext),
        result,
      );
      if (!prepared) {
        continue;
      }
      templateKey = prepared.templateKey;
      batch.push({
        emailAddress: prepared.contact,
        name: recipient.displayName || '',
        templateData: prepared.mergeInfo,
      });
    }

    if (batch.length === 0 || !templateKey) {
      return;
    }

    // Attachments (general-link Value Card) travel as S3 references only, same rule as every other
    // send here — see toQueueAttachments, which returns null when some file has no key and the send
    // therefore has to go direct.
    const queueAttachments = context.attachments?.length
      ? this.toQueueAttachments(context.attachments)
      : undefined;

    if (channelEnabled && queueAttachments !== null) {
      try {
        this.logger.log(
          `Enqueuing common-invite email for program ${context.programId}: ${JSON.stringify(batch)} recipient(s)`,
        );
        const messageIds = await this.emailQueueService.queueBulkEmail({
          from: {
            address: context.emailSenderAddress || zeptoEmailCreadentials.ZEPTO_EMAIL,
            name: context.emailSenderName || zeptoEmailCreadentials.ZEPTO_EMAIL_NAME,
          },
          subject: '',
          templateKey,
          recipients: batch,
          attachments: queueAttachments?.length
            ? queueAttachments.map(({ name, contentType, s3Key }) => ({
                name,
                contentType,
                s3Key: s3Key as string,
              }))
            : undefined,
          templateId: channelTemplate.templateId,
          createdBy,
          userId: createdBy ? String(createdBy) : undefined,
          metadata: { purpose },
        });
        if ((messageIds?.length ?? 0) > 0) {
          result.enqueued.email += batch.length;
          return;
        }
      } catch (error) {
        this.logger.error(
          `Failed to enqueue common-invite email for program ${context.programId}: ${(error as Error)?.message}`,
          (error as Error)?.stack,
        );
      }
    }

    // Direct fallback — one ZeptoMail /batch send (writes the same track rows), off-request like
    // the seeker batch: counted as dispatched now, outcome recorded in the log + track rows.
    const directTemplateKey = templateKey;
    result.enqueued.email += batch.length;
    this.runInBackground(
      `common-invite email, program ${context.programId}, ${batch.length} recipient(s)`,
      async () => {
        const dto: SendBulkEmailDto = {
          to: batch.map((recipient) => ({
            emailAddress: recipient.emailAddress,
            name: recipient.name,
            mergeInfo: recipient.templateData,
          })),
          from: {
            address: context.emailSenderAddress || zeptoEmailCreadentials.ZEPTO_EMAIL,
            name: context.emailSenderName || zeptoEmailCreadentials.ZEPTO_EMAIL_NAME,
          },
          subject: '',
          templateKey: directTemplateKey,
          // Only the fallback needs real bytes — the queued path above sends references and
          // downloads nothing.
          attachments: context.attachments?.length
            ? await this.resolveDirectAttachments(context.attachments)
            : undefined,
          trackinfo: {
            templateId: channelTemplate.templateId ?? undefined,
            createdBy,
            updatedBy: createdBy,
          },
        };
        try {
          await this.communicationService.sendBulkEmail(dto);
          this.logger.log(
            `Sent common-invite email batch directly (queue fallback) for program ${context.programId}: ${batch.length} recipient(s)`,
          );
        } catch (error) {
          this.logger.error(
            `Failed to direct-send common-invite email for program ${context.programId}: ${(error as Error)?.message}`,
            (error as Error)?.stack,
          );
        }
      },
    );
  }

  /**
   * Common invite, WHATSAPP: resolve every recipient into one WATI bulk template send —
   * enqueued (bulk-whatsapp queue) when enabled, else sent directly.
   */
  private async dispatchCommonInviteWhatsApp(
    context: SendContext,
    recipients: CommonInviteRecipient[],
    purpose: SessionCommunicationPurposeEnum,
    channelTemplate: ChannelTemplate,
    channelEnabled: boolean,
    createdBy: number,
    result: SendResult,
  ): Promise<void> {
    const batch: Array<{
      whatsappNumber: string;
      customParams: Array<{ name: string; value: string }>;
    }> = [];
    let templateKey: string | null = null;

    for (const recipient of recipients) {
      const prepared = await this.resolveProgramLevelSend(
        context,
        channelTemplate,
        recipient,
        this.commonInviteMergeContext(recipient, context.sessionId, context.extraContext),
        result,
      );
      if (!prepared) {
        continue;
      }
      templateKey = prepared.templateKey;
      batch.push({
        whatsappNumber: prepared.contact,
        customParams: Object.entries(prepared.mergeInfo).map(([name, value]) => ({
          name,
          value: String(value),
        })),
      });
    }

    if (batch.length === 0 || !templateKey) {
      return;
    }

    if (channelEnabled) {
      try {
        const messageIds = await this.whatsAppQueueService.queueBulkWhatsApp({
          templateName: templateKey,
          broadcastName: templateKey,
          recipients: batch,
          templateId: channelTemplate.templateId,
          createdBy,
          userId: createdBy ? String(createdBy) : undefined,
          metadata: { purpose },
        });
        if (messageIds?.length) {
          result.enqueued.whatsapp += batch.length;
          return;
        }
      } catch (error) {
        this.logger.error(
          `Failed to enqueue common-invite whatsapp for program ${context.programId}: ${(error as Error)?.message}`,
          (error as Error)?.stack,
        );
      }
    }

    // Direct fallback — WATI bulk template send, off-request like the seeker bulk.
    const directTemplateKey = templateKey;
    result.enqueued.whatsapp += batch.length;
    this.runInBackground(
      `common-invite whatsapp, program ${context.programId}, ${batch.length} recipient(s)`,
      async () => {
        try {
          await this.communicationService.sendBulkTemplateMessage({
            recipients: batch,
            templateName: directTemplateKey,
            broadcastName: directTemplateKey,
            trackinfo: {
              templateId: channelTemplate.templateId ?? undefined,
              createdBy,
              updatedBy: createdBy,
            },
          });
          this.logger.log(
            `Sent common-invite whatsapp bulk directly (queue fallback) for program ${context.programId}: ${batch.length} recipient(s)`,
          );
        } catch (error) {
          this.logger.error(
            `Failed to direct-send common-invite whatsapp for program ${context.programId}: ${(error as Error)?.message}`,
            (error as Error)?.stack,
          );
        }
      },
    );
  }

  // ==================================================================================
  // System links (program-level, ADMIN recipients, email only) dispatch.
  // ==================================================================================

  /**
   * Fire the (email) System Links send. Mirrors processCommonInvite but email-only and with a
   * shared systemJoiningDetails table threaded into every recipient's merge context.
   */
  private async processSystemLinks(
    context: SendContext,
    channelTemplates: ChannelTemplate[],
    recipients: CommonInviteRecipient[],
    systemJoiningDetails: string,
    purpose: SessionCommunicationPurposeEnum,
    createdBy: number,
  ): Promise<SendResult> {
    const result: SendResult = {
      requested: recipients.length,
      enqueued: { email: 0, whatsapp: 0 },
      skipped: [],
    };

    for (const channelTemplate of channelTemplates) {
      // System Links is email only; ignore any other channel defensively.
      if (channelTemplate.channel !== CommunicationTypeEnum.EMAIL) {
        continue;
      }
      const channelEnabled = this.emailQueueService.isQueueEnabled();
      await this.dispatchSystemLinksEmail(
        context,
        recipients,
        systemJoiningDetails,
        purpose,
        channelTemplate,
        channelEnabled,
        createdBy,
        result,
      );
    }

    this.logger.log(
      `System links dispatched: program=${context.programId}, requested=${result.requested}, ` +
        `email=${result.enqueued.email}, skipped=${result.skipped.length}`,
    );
    return result;
  }

  /**
   * System Links, EMAIL: resolve every admin recipient into one ZeptoMail /batch send — enqueued
   * (bulk-email queue) when enabled, else sent directly. admin_user_name is per recipient; the
   * systemJoiningDetails table is shared across the batch.
   */
  private async dispatchSystemLinksEmail(
    context: SendContext,
    recipients: CommonInviteRecipient[],
    systemJoiningDetails: string,
    purpose: SessionCommunicationPurposeEnum,
    channelTemplate: ChannelTemplate,
    channelEnabled: boolean,
    createdBy: number,
    result: SendResult,
  ): Promise<void> {
    const batch: Array<{ emailAddress: string; name: string; templateData: Record<string, any> }> =
      [];
    let templateKey: string | null = null;

    for (const recipient of recipients) {
      const prepared = await this.resolveProgramLevelSend(
        context,
        channelTemplate,
        recipient,
        {
          admin_user_name: recipient.displayName ?? '',
          systemJoiningDetails,
          // Per-session send: include the sessionId so session_name/date/time resolve that
          // session rather than falling back to the program's first session.
          ...(context.sessionId != null ? { sessionId: context.sessionId } : {}),
        },
        result,
      );
      if (!prepared) {
        continue;
      }
      templateKey = prepared.templateKey;
      batch.push({
        emailAddress: prepared.contact,
        name: recipient.displayName || '',
        templateData: prepared.mergeInfo,
      });
    }

    if (batch.length === 0 || !templateKey) {
      return;
    }

    if (channelEnabled) {
      try {
        const messageIds = await this.emailQueueService.queueBulkEmail({
          from: {
            address: context.emailSenderAddress || zeptoEmailCreadentials.ZEPTO_EMAIL,
            name: context.emailSenderName || zeptoEmailCreadentials.ZEPTO_EMAIL_NAME,
          },
          subject: '',
          templateKey,
          recipients: batch,
          templateId: channelTemplate.templateId,
          createdBy,
          userId: createdBy ? String(createdBy) : undefined,
          metadata: { purpose },
        });
        if ((messageIds?.length ?? 0) > 0) {
          result.enqueued.email += batch.length;
          return;
        }
      } catch (error) {
        this.logger.error(
          `Failed to enqueue system-links email for program ${context.programId}: ${(error as Error)?.message}`,
          (error as Error)?.stack,
        );
      }
    }

    // Direct fallback — one ZeptoMail /batch send (writes the same track rows).
    const dto: SendBulkEmailDto = {
      to: batch.map((recipient) => ({
        emailAddress: recipient.emailAddress,
        name: recipient.name,
        mergeInfo: recipient.templateData,
      })),
      from: {
        address: context.emailSenderAddress || zeptoEmailCreadentials.ZEPTO_EMAIL,
        name: context.emailSenderName || zeptoEmailCreadentials.ZEPTO_EMAIL_NAME,
      },
      subject: '',
      templateKey,
      trackinfo: {
        templateId: channelTemplate.templateId ?? undefined,
        createdBy,
        updatedBy: createdBy,
      },
    };
    try {
      await this.communicationService.sendBulkEmail(dto);
      result.enqueued.email += batch.length;
      this.logger.log(
        `Sent system-links email batch directly (queue fallback) for program ${context.programId}: ${batch.length} recipient(s)`,
      );
    } catch (error) {
      this.logger.error(
        `Failed to direct-send system-links email for program ${context.programId}: ${(error as Error)?.message}`,
        (error as Error)?.stack,
      );
      batch.forEach(() =>
        result.skipped.push({
          registrationId: 0,
          channel: channelTemplate.channel,
          reason: 'failed to send',
        }),
      );
    }
  }

  /**
   * Build the systemJoiningDetails HTML table (Name | Link) from the program's generated
   * system/placeholder links — same bordered style as the absent-sessions table. Returns '' when
   * there are no system links (so the template can hide the block).
   */
  private buildSystemJoiningDetailsTable(links: SystemGeneratedLink[]): string {
    const usable = links.filter((link) => link.joinUrl);
    if (usable.length === 0) {
      return '';
    }
    const cell = 'border:1px solid #d0d0d0;padding:8px 12px;';
    const header = `${cell}background:#f2f2f2;font-weight:bold;text-align:left;`;
    const rows = usable
      .map((link) => {
        const name = this.escapeHtml(link.displayName ?? '');
        const url = this.escapeHtml(link.joinUrl ?? '');
        return (
          `<tr><td style="${cell}">${name}</td>` +
          `<td style="${cell}"><a href="${url}">${url}</a></td></tr>`
        );
      })
      .join('');
    return (
      `<table style="border-collapse:collapse;width:100%;font-family:Cambria, serif;">` +
      `<thead><tr><th style="${header}">Name</th><th style="${header}">Link</th></tr></thead>` +
      `<tbody>${rows}</tbody></table>`
    );
  }

  /** Minimal HTML escape for values interpolated into the systemJoiningDetails table. */
  private escapeHtml(value: string): string {
    return String(value ?? '')
      .replace(/&/g, '&amp;')
      .replace(/</g, '&lt;')
      .replace(/>/g, '&gt;')
      .replace(/"/g, '&quot;');
  }

  /**
   * Load a session and enforce that it exists and is ONLINE.
   */
  private async loadOnlineSession(sessionId: number): Promise<ProgramSession> {
    const session = await this.sessionRepo.findOne({
      where: { id: sessionId },
      relations: ['program', 'program.type'],
    });
    if (!session) {
      throw new InifniNotFoundException(
        ERROR_CODES.PROGRAM_SESSION_NOTFOUND,
        null,
        null,
        String(sessionId),
      );
    }
    if (session.modeOfOperation !== ModeOfOperationEnum.ONLINE) {
      throw new InifniBadRequestException(ERROR_CODES.SESSION_COMMUNICATION_NOT_ONLINE);
    }
    return session;
  }

  /**
   * Auto-derive the template variant for the target session. Purposes that don't depend on
   * session position (Welcome, Value Card) short-circuit to null.
   *
   * The distinct FINAL- and PRE_FINAL-session templates apply ONLY to the TAT program type.
   * For TAT, the last session of a multi-session program resolves to FINAL, the last-but-one
   * (penultimate) session resolves to PRE_FINAL (for purposes that have a pre-final template —
   * only ABSENT today; INVITE falls back to REGULAR), and every earlier session resolves to
   * REGULAR. For all other program types (and single-session programs) every session uses the
   * REGULAR communication.
   *
   * Ordering: chronological by startsAt (nulls last), then displayOrder, then id.
   */
  private async resolveOccurrence(
    session: ProgramSession,
    purpose: SessionCommunicationPurposeEnum,
  ): Promise<SessionOccurrence | null> {
    if (!purposeNeedsOccurrence(purpose)) {
      return null;
    }

    // Only the TAT program type distinguishes the final / pre-final sessions.
    const isTat = session.program?.type?.key === PROGRAM_TYPE_KEYS.TAT;
    if (!isTat) {
      return SessionOccurrence.REGULAR;
    }

    const sessions = await this.sessionRepo.find({
      where: { programId: session.programId },
      order: { startsAt: 'ASC', displayOrder: 'ASC', id: 'ASC' },
      select: ['id'],
    });

    const finalId = sessions[sessions.length - 1]?.id;
    const preFinalId = sessions.length >= 2 ? sessions[sessions.length - 2]?.id : undefined;

    if (sessions.length > 1 && session.id === finalId) {
      return SessionOccurrence.FINAL;
    }
    // Penultimate session — only when the purpose actually has a PRE_FINAL template
    // (ABSENT); otherwise it falls back to REGULAR so channel resolution still finds a template.
    if (
      preFinalId !== undefined &&
      session.id === preFinalId &&
      purposeSupportsOccurrence(purpose, SessionOccurrence.PRE_FINAL)
    ) {
      return SessionOccurrence.PRE_FINAL;
    }
    return SessionOccurrence.REGULAR;
  }

  /**
   * Selection modes SELECTED / EXCLUDED require a non-empty registrationIds list.
   * (class-validator also enforces this; this guards service-level callers.)
   */
  private assertSelectionIds(dto: SendBulkSessionCommunicationDto): void {
    const needsIds =
      dto.selectionMode === BulkCommunicationSelectionModeEnum.SELECTED ||
      dto.selectionMode === BulkCommunicationSelectionModeEnum.EXCLUDED;
    if (needsIds && (!dto.registrationIds || dto.registrationIds.length === 0)) {
      throw new InifniBadRequestException(ERROR_CODES.SESSION_COMMUNICATION_IDS_REQUIRED);
    }
  }
}