pn-new-crm/src/lib/services/document-sends.service.ts

/**
 * Sales send-out flow (Phase 7 — see plan §4.8 / §11.1 / §14.7).
 *
 * Sends per-berth PDFs and brochures to a client recipient, attaching the
 * file when it's at-or-below the configured threshold or falling back to a
 * 24h signed-URL link when it's larger. Every send writes one row to
 * `document_sends` (success OR failure) so the rep can see the outcome in
 * the timeline.
 *
 * §14.7 critical mitigations implemented here:
 *
 *   - **Body XSS** — bodies go through `renderEmailBody()` (HTML-escape +
 *     allowlist of markdown rules) before reaching nodemailer.
 *   - **Recipient typo** — recipient email validated against a strict regex
 *     before the SMTP transaction.
 *   - **Unresolved merge fields** — `findUnresolvedTokens()` is exported
 *     for the dry-run UI; the service blocks sends with unresolved tokens
 *     unless `allowUnresolved: true` is explicitly passed (test-only).
 *   - **SMTP failure** — every transport rejection writes a `failedAt` row
 *     with `errorReason` and surfaces a typed error to the API.
 *   - **Hourly rate limit** — 50 sends/user/hour individual.
 *   - **Size threshold fallback** — files larger than the per-port
 *     `email_attach_threshold_mb` go as a signed-URL link in the body
 *     instead of an attachment (§11.1).
 */

import { Readable } from 'node:stream';

import { and, desc, eq } from 'drizzle-orm';
import type { SentMessageInfo } from 'nodemailer';

import { db } from '@/lib/db';
import {
  brochures,
  brochureVersions,
  documentSends,
  berths,
  berthPdfVersions,
  clients,
  clientContacts,
  customFieldDefinitions,
  customFieldValues,
  interests,
  ports,
} from '@/lib/db/schema';
import { inArray } from 'drizzle-orm';
import type { DocumentSend } from '@/lib/db/schema';
import { ForbiddenError, NotFoundError, ValidationError } from '@/lib/errors';
import { logger } from '@/lib/logger';
import { checkRateLimit } from '@/lib/rate-limit';
import { getStorageBackend } from '@/lib/storage';
import {
  EMAIL_BODY_MAX_BYTES,
  expandMergeTokens,
  findUnresolvedTokens,
  renderEmailBody,
} from '@/lib/utils/markdown-email';
import { getDefaultBrochure } from '@/lib/services/brochures.service';
import {
  createSalesTransporter,
  getSalesContentConfig,
} from '@/lib/services/sales-email-config.service';

// ─── Public types ────────────────────────────────────────────────────────────

export interface SendRecipientInput {
  /** Existing client ID (resolves the primary email automatically). */
  clientId?: string;
  /** Optional explicit address override (for cases where a client has multiple). */
  email?: string;
  /** Optional interest pin so the audit row links into the interest timeline. */
  interestId?: string;
}

export interface SendBerthPdfInput {
  portId: string;
  berthId: string;
  recipient: SendRecipientInput;
  /** When provided, replaces the per-port template. Still passes through
   *  merge expansion + sanitization. */
  customBodyMarkdown?: string;
  sentBy: string;
  ipAddress: string;
  userAgent: string;
  /** Test-only: skip the unresolved-merge-field block. */
  allowUnresolved?: boolean;
}

export interface SendBrochureInput {
  portId: string;
  /** Defaults to the port's default brochure when omitted. */
  brochureId?: string;
  recipient: SendRecipientInput;
  customBodyMarkdown?: string;
  sentBy: string;
  ipAddress: string;
  userAgent: string;
  allowUnresolved?: boolean;
}

export interface SendResult {
  send: DocumentSend;
  /** True when the file was attached; false when a signed-URL link was used. */
  deliveredAsAttachment: boolean;
  /** Set when the transport rejected — the row carries `failedAt`. */
  error?: string;
}

// ─── Public dry-run / preview helpers (used by the modal) ────────────────────

/**
 * Compute the merge-value bag for a given send context. The same map is used
 * by the dry-run preview AND the actual send so the rep sees exactly what
 * gets posted.
 */
export async function buildMergeValues(
  portId: string,
  recipient: SendRecipientInput,
  context: { berthId?: string; brochureLabel?: string } = {},
): Promise<Record<string, string>> {
  const values: Record<string, string> = {};
  values['{{date.today}}'] = new Date().toISOString().slice(0, 10);
  values['{{date.year}}'] = String(new Date().getFullYear());

  const port = await db.query.ports.findFirst({ where: eq(ports.id, portId) });
  if (port) {
    values['{{port.name}}'] = port.name;
    if (port.defaultCurrency) values['{{port.defaultCurrency}}'] = port.defaultCurrency;
  }

  if (recipient.clientId) {
    const client = await db.query.clients.findFirst({
      where: and(eq(clients.id, recipient.clientId), eq(clients.portId, portId)),
    });
    if (client) {
      if (client.fullName) values['{{client.fullName}}'] = client.fullName;
      if (client.nationalityIso) values['{{client.nationality}}'] = client.nationalityIso;
      if (client.source) values['{{client.source}}'] = client.source;
      const contacts = await db.query.clientContacts.findMany({
        where: eq(clientContacts.clientId, client.id),
      });
      const primaryEmail =
        contacts.find((c) => c.channel === 'email' && c.isPrimary)?.value ??
        contacts.find((c) => c.channel === 'email')?.value;
      const primaryPhone =
        contacts.find((c) => c.channel === 'phone' && c.isPrimary)?.value ??
        contacts.find((c) => c.channel === 'phone')?.value;
      if (primaryEmail) values['{{client.email}}'] = primaryEmail;
      if (primaryPhone) values['{{client.phone}}'] = primaryPhone;
    }
  }

  if (context.berthId) {
    const berth = await db.query.berths.findFirst({
      where: and(eq(berths.id, context.berthId), eq(berths.portId, portId)),
    });
    if (berth) {
      values['{{berth.mooringNumber}}'] = berth.mooringNumber;
      if (berth.area) values['{{berth.area}}'] = berth.area;
      if (berth.status) values['{{berth.status}}'] = berth.status;
      if (berth.lengthFt) values['{{berth.lengthFt}}'] = String(berth.lengthFt);
      if (berth.widthFt) values['{{berth.widthFt}}'] = String(berth.widthFt);
      if (berth.price) values['{{berth.price}}'] = String(berth.price);
      if (berth.priceCurrency) values['{{berth.priceCurrency}}'] = berth.priceCurrency;
    }
  }

  // Custom-field tokens (`{{custom.<fieldName>}}`). The validator allows
  // any matching shape; the resolver here looks up real values per-port,
  // per-entity and substitutes them. Unknown field names stay
  // unresolved — `findUnresolvedTokens` flags them at preview time so
  // the rep can edit the template before sending.
  await mergeCustomFieldValues(values, portId, recipient, context);

  return values;
}

interface CustomMergeContext {
  berthId?: string;
  brochureLabel?: string;
}

/**
 * Resolve `{{custom.<fieldName>}}` tokens. Reads every per-port custom
 * field definition for the entity types currently in scope (client,
 * interest, berth) and joins to the actual stored value for each entity
 * id we have on hand. Boolean values render as 'true' / 'false', dates
 * as ISO yyyy-mm-dd, numbers as plain numerics, selects/text verbatim.
 */
async function mergeCustomFieldValues(
  values: Record<string, string>,
  portId: string,
  recipient: SendRecipientInput,
  context: CustomMergeContext,
): Promise<void> {
  // Build the (entityType → entityId) map for the current send context.
  const entityIdsByType = new Map<string, string>();
  if (recipient.clientId) entityIdsByType.set('client', recipient.clientId);
  if (recipient.interestId) entityIdsByType.set('interest', recipient.interestId);
  if (context.berthId) entityIdsByType.set('berth', context.berthId);
  if (entityIdsByType.size === 0) return;

  const definitions = await db
    .select()
    .from(customFieldDefinitions)
    .where(
      and(
        eq(customFieldDefinitions.portId, portId),
        inArray(customFieldDefinitions.entityType, Array.from(entityIdsByType.keys())),
      ),
    );
  if (definitions.length === 0) return;

  const fieldIds = definitions.map((d) => d.id);
  const entityIds = Array.from(entityIdsByType.values());
  const valueRows = await db
    .select()
    .from(customFieldValues)
    .where(
      and(
        inArray(customFieldValues.fieldId, fieldIds),
        inArray(customFieldValues.entityId, entityIds),
      ),
    );

  const valueByFieldEntity = new Map<string, unknown>();
  for (const row of valueRows) {
    valueByFieldEntity.set(`${row.fieldId}|${row.entityId}`, row.value);
  }

  for (const def of definitions) {
    const entityId = entityIdsByType.get(def.entityType);
    if (!entityId) continue;
    const raw = valueByFieldEntity.get(`${def.id}|${entityId}`);
    if (raw === undefined || raw === null) continue;
    const token = `{{custom.${def.fieldName}}}`;
    values[token] = stringifyCustomValue(raw, def.fieldType);
  }
}

function stringifyCustomValue(raw: unknown, fieldType: string): string {
  if (raw === null || raw === undefined) return '';
  switch (fieldType) {
    case 'boolean':
      return raw ? 'true' : 'false';
    case 'date':
      return typeof raw === 'string' ? raw.slice(0, 10) : String(raw);
    case 'number':
      return String(raw);
    default:
      return typeof raw === 'string' ? raw : JSON.stringify(raw);
  }
}

/**
 * Render a body for the dry-run UI. Returns `{ html, unresolved }`. The UI
 * uses `unresolved` to populate the warning chip; the rep can't submit
 * until the list is empty.
 */
export async function previewBody(
  portId: string,
  documentKind: 'berth_pdf' | 'brochure',
  recipient: SendRecipientInput,
  customBody: string | null,
  ctx: { berthId?: string; brochureLabel?: string } = {},
): Promise<{ html: string; markdown: string; unresolved: string[] }> {
  const content = await getSalesContentConfig(portId);
  const template = customBody?.trim()?.length
    ? customBody
    : documentKind === 'berth_pdf'
      ? content.templateBerthPdfBody
      : content.templateBrochureBody;
  const values = await buildMergeValues(portId, recipient, ctx);
  const expanded = expandMergeTokens(template, values);
  const unresolved = findUnresolvedTokens(template, values);
  const html = renderEmailBody(expanded);
  return { html, markdown: expanded, unresolved };
}

// ─── Internal helpers ────────────────────────────────────────────────────────

const RFC5322_EMAIL = /^[^\s@]+@[^\s@]+\.[^\s@]+$/;

function assertEmailValid(email: string): void {
  if (!email || email.length > 254 || !RFC5322_EMAIL.test(email)) {
    throw new ValidationError(`Invalid recipient email: ${email}`);
  }
}

async function resolveRecipientEmail(
  portId: string,
  recipient: SendRecipientInput,
): Promise<string> {
  if (recipient.email) {
    assertEmailValid(recipient.email);
    return recipient.email;
  }
  if (!recipient.clientId) {
    throw new ValidationError('Recipient must include either clientId or email');
  }
  const client = await db.query.clients.findFirst({
    where: and(eq(clients.id, recipient.clientId), eq(clients.portId, portId)),
  });
  if (!client) throw new NotFoundError('Client');

  const contacts = await db.query.clientContacts.findMany({
    where: eq(clientContacts.clientId, client.id),
  });
  const emails = contacts.filter((c) => c.channel === 'email');
  const primary = emails.find((c) => c.isPrimary) ?? emails[0];
  if (!primary) throw new ValidationError('Client has no email on file');
  assertEmailValid(primary.value);
  return primary.value;
}

/**
 * Verify a caller-supplied `interestId` belongs to the authenticated port
 * before it lands on the `document_sends` audit row. Without this, an
 * attacker who knows a foreign-port interest UUID can pollute another
 * tenant's audit history (the surrounding `clientId` lookup is already
 * port-scoped, so data isn't exposed — but the audit trail would be).
 */
async function assertInterestInPort(portId: string, interestId: string): Promise<void> {
  const row = await db.query.interests.findFirst({
    where: and(eq(interests.id, interestId), eq(interests.portId, portId)),
    columns: { id: true },
  });
  if (!row) throw new NotFoundError('Interest');
}

async function checkSendRateLimit(portId: string, userId: string): Promise<void> {
  // Per-(port, user) so a multi-port rep can't be DoS'd by another tenant
  // burning their global cap. Audit caught this — the original
  // single-key version locked a user out across every port they touched.
  const result = await checkRateLimit(`${portId}:${userId}`, {
    windowMs: 60 * 60 * 1000,
    max: 50,
    keyPrefix: 'docsend',
  });
  if (!result.allowed) {
    throw new ForbiddenError(
      `Hit hourly send limit (${result.limit}). Retry after ${new Date(
        result.resetAt,
      ).toISOString()}.`,
    );
  }
}

interface ResolvedAttachment {
  /** Object key in the active storage backend. */
  storageKey: string;
  fileName: string;
  fileSizeBytes: number;
}

async function streamAttachmentOrLink(
  portId: string,
  attachment: ResolvedAttachment,
): Promise<{
  attachments?: Array<{ filename: string; content: Readable }>;
  bodySuffixHtml?: string;
  deliveredAsAttachment: boolean;
}> {
  const content = await getSalesContentConfig(portId);
  const thresholdBytes = content.emailAttachThresholdMb * 1024 * 1024;

  if (attachment.fileSizeBytes <= thresholdBytes) {
    // Stream from storage directly into nodemailer to avoid buffering 20MB+.
    const storage = await getStorageBackend();
    const stream = await storage.get(attachment.storageKey);
    // The storage abstraction returns NodeJS.ReadableStream; nodemailer's
    // Attachment.content type wants `Readable`. The two are compatible —
    // both stream backends expose a Readable. Cast to keep types tight.
    const readable = stream as unknown as Readable;
    return {
      deliveredAsAttachment: true,
      attachments: [{ filename: attachment.fileName, content: readable }],
    };
  }

  // Above threshold: generate a 24h signed download URL and append a link
  // to the body. Per §11.1 the size decision is made BEFORE the SMTP relay,
  // so we never produce duplicate sends.
  const storage = await getStorageBackend();
  // Bind the proxy token to the issuing port slug. The storage key is
  // already structured `${portSlug}/...` via generateStorageKey() — this
  // closes the loop so a buggy future call site that hands us a key from
  // a different port can't mint a valid 24h URL for it.
  const portRow = await db.query.ports.findFirst({
    where: eq(ports.id, portId),
    columns: { slug: true },
  });
  const { url } = await storage.presignDownload(attachment.storageKey, {
    expirySeconds: 24 * 60 * 60,
    filename: attachment.fileName,
    portSlug: portRow?.slug,
  });
  // HTML-escape the filename: brochure filenames are admin-supplied and
  // could in theory carry markup (e.g. `"><script>...`). Even a benign
  // ampersand or angle bracket would break the rendered link otherwise.
  const safeFileName = attachment.fileName
    .replace(/&/g, '&amp;')
    .replace(/</g, '&lt;')
    .replace(/>/g, '&gt;')
    .replace(/"/g, '&quot;')
    .replace(/'/g, '&#39;');
  const html = `<p>The file is large enough that we're sending it as a download link rather than an attachment:</p>
<p><a href="${url}" target="_blank" rel="noopener noreferrer">Download ${safeFileName}</a> (link expires in 24 hours)</p>`;
  return { deliveredAsAttachment: false, bodySuffixHtml: html };
}

async function performSend(args: {
  portId: string;
  recipientEmail: string;
  subject: string;
  bodyHtml: string;
  attachment: ResolvedAttachment;
  recordSeed: Omit<typeof documentSends.$inferInsert, 'id' | 'sentAt' | 'createdAt'>;
}): Promise<SendResult> {
  // 1. Build attachment vs link preamble.
  const delivery = await streamAttachmentOrLink(args.portId, args.attachment);
  const finalHtml = delivery.bodySuffixHtml
    ? `${args.bodyHtml}\n${delivery.bodySuffixHtml}`
    : args.bodyHtml;

  // 2. Create the transporter (per-port sales account).
  let transporter, fromAddress;
  try {
    ({ transporter, fromAddress } = await createSalesTransporter(args.portId));
  } catch (configErr) {
    const msg = configErr instanceof Error ? configErr.message : String(configErr);
    const [row] = await db
      .insert(documentSends)
      .values({
        ...args.recordSeed,
        fromAddress: args.recordSeed.fromAddress || 'unknown',
        bodyMarkdown: args.recordSeed.bodyMarkdown ?? null,
        failedAt: new Date(),
        errorReason: msg,
      })
      .returning();
    return {
      send: row!,
      deliveredAsAttachment: false,
      error: msg,
    };
  }

  // 3. Send.
  try {
    const info: SentMessageInfo = await transporter.sendMail({
      from: fromAddress,
      to: args.recipientEmail,
      subject: args.subject,
      html: finalHtml,
      ...(delivery.attachments ? { attachments: delivery.attachments } : {}),
    });
    const [row] = await db
      .insert(documentSends)
      .values({
        ...args.recordSeed,
        fromAddress,
        messageId: info.messageId ?? null,
        fallbackToLinkReason: delivery.deliveredAsAttachment ? null : 'size_above_threshold',
      })
      .returning();
    return { send: row!, deliveredAsAttachment: delivery.deliveredAsAttachment };
  } catch (sendErr) {
    const msg = sendErr instanceof Error ? sendErr.message : String(sendErr);
    logger.error({ err: sendErr, portId: args.portId }, 'Sales send failed');
    const [row] = await db
      .insert(documentSends)
      .values({
        ...args.recordSeed,
        fromAddress,
        failedAt: new Date(),
        errorReason: msg,
      })
      .returning();
    return { send: row!, deliveredAsAttachment: false, error: msg };
  }
}

// ─── Public sender: berth PDF ────────────────────────────────────────────────

export async function sendBerthPdf(input: SendBerthPdfInput): Promise<SendResult> {
  // Rate-limit AFTER validation so a typo'd recipient or missing-PDF rep
  // doesn't burn a slot on a send that would have failed anyway.
  const recipientEmail = await resolveRecipientEmail(input.portId, input.recipient);
  if (input.recipient.interestId) {
    await assertInterestInPort(input.portId, input.recipient.interestId);
  }

  // Resolve berth + active version.
  const berth = await db.query.berths.findFirst({
    where: and(eq(berths.id, input.berthId), eq(berths.portId, input.portId)),
  });
  if (!berth) throw new NotFoundError('Berth');
  if (!berth.currentPdfVersionId) {
    throw new ValidationError(
      'No PDF uploaded for this berth yet. Upload one in the berth detail page first.',
    );
  }
  const version = await db.query.berthPdfVersions.findFirst({
    where: eq(berthPdfVersions.id, berth.currentPdfVersionId),
  });
  if (!version) throw new NotFoundError('Berth PDF version');

  // Build body.
  const content = await getSalesContentConfig(input.portId);
  const template = input.customBodyMarkdown?.trim()?.length
    ? input.customBodyMarkdown
    : content.templateBerthPdfBody;
  if (Buffer.byteLength(template, 'utf8') > EMAIL_BODY_MAX_BYTES) {
    throw new ValidationError('Email body exceeds maximum length');
  }
  const values = await buildMergeValues(input.portId, input.recipient, { berthId: berth.id });
  const unresolved = findUnresolvedTokens(template, values);
  if (unresolved.length > 0 && !input.allowUnresolved) {
    throw new ValidationError(`Unresolved merge tokens: ${unresolved.join(', ')}`);
  }
  const expanded = expandMergeTokens(template, values);
  const bodyHtml = renderEmailBody(expanded);

  // Subject pulls in the mooring number for inbox triage.
  const subject = `Berth ${berth.mooringNumber} — spec sheet`;

  await checkSendRateLimit(input.portId, input.sentBy);

  return performSend({
    portId: input.portId,
    recipientEmail,
    subject,
    bodyHtml,
    attachment: {
      storageKey: version.storageKey,
      fileName: version.fileName,
      fileSizeBytes: version.fileSizeBytes,
    },
    recordSeed: {
      portId: input.portId,
      clientId: input.recipient.clientId ?? null,
      interestId: input.recipient.interestId ?? null,
      recipientEmail,
      documentKind: 'berth_pdf',
      berthId: berth.id,
      berthPdfVersionId: version.id,
      brochureId: null,
      brochureVersionId: null,
      bodyMarkdown: expanded,
      sentByUserId: input.sentBy,
      fromAddress: '',
    },
  });
}

// ─── Public sender: brochure ─────────────────────────────────────────────────

export async function sendBrochure(input: SendBrochureInput): Promise<SendResult> {
  // Rate-limit AFTER validation (audit finding); typos shouldn't burn slots.
  const recipientEmail = await resolveRecipientEmail(input.portId, input.recipient);
  if (input.recipient.interestId) {
    await assertInterestInPort(input.portId, input.recipient.interestId);
  }

  // Resolve brochure + most-recent version.
  let brochureRow;
  if (input.brochureId) {
    brochureRow = await db.query.brochures.findFirst({
      where: and(eq(brochures.id, input.brochureId), eq(brochures.portId, input.portId)),
    });
    if (!brochureRow) throw new NotFoundError('Brochure');
    if (brochureRow.archivedAt) {
      throw new ValidationError('Brochure is archived');
    }
  } else {
    const def = await getDefaultBrochure(input.portId);
    if (!def || !def.currentVersion) {
      throw new ValidationError(
        'No default brochure configured for this port. Upload one in /admin/brochures.',
      );
    }
    // The partial unique index on `is_default` only enforces uniqueness when
    // archived_at IS NULL — an archived row can still carry is_default=true
    // and would silently be returned here without this guard.
    if (def.archivedAt) {
      throw new ValidationError(
        'Default brochure is archived. Choose a non-archived brochure as the default first.',
      );
    }
    brochureRow = def;
  }

  const versions = await db.query.brochureVersions.findMany({
    where: eq(brochureVersions.brochureId, brochureRow.id),
    orderBy: [desc(brochureVersions.uploadedAt)],
    limit: 1,
  });
  const version = versions[0];
  if (!version) {
    throw new ValidationError('Brochure has no uploaded version yet');
  }

  // Build body.
  const content = await getSalesContentConfig(input.portId);
  const template = input.customBodyMarkdown?.trim()?.length
    ? input.customBodyMarkdown
    : content.templateBrochureBody;
  if (Buffer.byteLength(template, 'utf8') > EMAIL_BODY_MAX_BYTES) {
    throw new ValidationError('Email body exceeds maximum length');
  }
  const values = await buildMergeValues(input.portId, input.recipient, {
    brochureLabel: brochureRow.label,
  });
  const unresolved = findUnresolvedTokens(template, values);
  if (unresolved.length > 0 && !input.allowUnresolved) {
    throw new ValidationError(`Unresolved merge tokens: ${unresolved.join(', ')}`);
  }
  const expanded = expandMergeTokens(template, values);
  const bodyHtml = renderEmailBody(expanded);
  const subject = `${brochureRow.label} — brochure`;

  await checkSendRateLimit(input.portId, input.sentBy);

  return performSend({
    portId: input.portId,
    recipientEmail,
    subject,
    bodyHtml,
    attachment: {
      storageKey: version.storageKey,
      fileName: version.fileName,
      fileSizeBytes: version.fileSizeBytes,
    },
    recordSeed: {
      portId: input.portId,
      clientId: input.recipient.clientId ?? null,
      interestId: input.recipient.interestId ?? null,
      recipientEmail,
      documentKind: 'brochure',
      berthId: null,
      berthPdfVersionId: null,
      brochureId: brochureRow.id,
      brochureVersionId: version.id,
      bodyMarkdown: expanded,
      sentByUserId: input.sentBy,
      fromAddress: '',
    },
  });
}

// ─── Audit query ─────────────────────────────────────────────────────────────

export interface ListSendsFilters {
  portId: string;
  clientId?: string;
  interestId?: string;
  berthId?: string;
  limit?: number;
}

export async function listSends(filters: ListSendsFilters): Promise<DocumentSend[]> {
  const conds = [eq(documentSends.portId, filters.portId)];
  if (filters.clientId) conds.push(eq(documentSends.clientId, filters.clientId));
  if (filters.interestId) conds.push(eq(documentSends.interestId, filters.interestId));
  if (filters.berthId) conds.push(eq(documentSends.berthId, filters.berthId));
  const rows = await db
    .select()
    .from(documentSends)
    .where(and(...conds))
    .orderBy(desc(documentSends.sentAt))
    .limit(filters.limit ?? 100);
  return rows;
}