pn-new-crm/src/lib/services/dashboard-report-data.service.ts

/**
 * Server-side data resolver for the dashboard PDF report.
 *
 * Each section is gated on its widget id being present in
 * `config.widgetIds`, so a report that only includes the pipeline
 * funnel runs ONE query instead of the full dashboard panel. Keeps
 * cold-call latency low even when the actual port has hundreds of
 * berths.
 *
 * Lives in its own file (not inside dashboard.service.ts) so the
 * report-builder concerns - what widget ids map to what fetcher,
 * which fields the PDF shape requires - stay scoped to the
 * report-side surface, not the dashboard UI.
 */
import { and, count, desc, eq, gte, lte, sql } from 'drizzle-orm';

import { db } from '@/lib/db';
import { clients } from '@/lib/db/schema/clients';
import { interests, interestBerths } from '@/lib/db/schema/interests';
import { berths } from '@/lib/db/schema/berths';
import { documents } from '@/lib/db/schema/documents';
import { reminders } from '@/lib/db/schema/operations';
import { payments } from '@/lib/db/schema/pipeline';
import { ports } from '@/lib/db/schema/ports';
import { auditLogs } from '@/lib/db/schema/system';
import { userProfiles } from '@/lib/db/schema/users';
import { canonicalizeStage } from '@/lib/constants';
import { websiteSubmissions } from '@/lib/db/schema/website-submissions';
import { computeDealHealth } from './deal-health';
import { getAllBerthMooringsForInterests } from './interest-berths.service';
import { deriveInterestBerthLabel } from '@/lib/templates/interest-berth-label';
import {
  getKpis,
  getPipelineCounts,
  getBerthStatusDistribution,
  getHotDeals,
  getRevenueForecast,
  getSourceConversion,
} from './dashboard.service';
import type { DashboardReportData } from '@/lib/pdf/reports/dashboard-report';

export interface DashboardReportWindow {
  /** Optional inclusive lower bound (YYYY-MM-DD). */
  dateFrom?: string;
  /** Optional inclusive upper bound (YYYY-MM-DD). */
  dateTo?: string;
}

function parseWindow(window: DashboardReportWindow | undefined): {
  from: Date | null;
  to: Date | null;
} {
  if (!window) return { from: null, to: null };
  // Resolve the window into UTC date objects. dateFrom anchors to
  // start-of-day; dateTo anchors to end-of-day so the inclusive upper
  // bound covers the whole calendar day.
  const from = window.dateFrom ? new Date(`${window.dateFrom}T00:00:00.000Z`) : null;
  const to = window.dateTo ? new Date(`${window.dateTo}T23:59:59.999Z`) : null;
  return {
    from: from && !Number.isNaN(from.getTime()) ? from : null,
    to: to && !Number.isNaN(to.getTime()) ? to : null,
  };
}

// Pure data/types now live in `dashboard-report-widgets.ts` so the
// client-side export button can import them without dragging this
// file's DB-touching imports into the browser bundle. Re-exported
// here so existing consumers keep working.
export {
  PDF_DASHBOARD_WIDGET_IDS,
  PDF_DASHBOARD_WIDGETS,
  PDF_DASHBOARD_CATEGORY_LABELS,
  type PdfDashboardWidgetId,
  type PdfDashboardWidgetOption,
  type PdfDashboardWidgetCategory,
} from './dashboard-report-widgets';

/**
 * Widget ids whose data resolver isn't fully wired yet. Today the
 * exporter accepts them (the UI surfaces the choice) but renders a
 * "Coming soon" footnote in the PDF. Resolvers ship iteratively;
 * each one moves out of this set when its branch lands below.
 */
/** All 16 widget resolvers now ship — set is intentionally empty.
 *  Kept as a non-empty `Set<string>` type to make adding NEW widgets
 *  (whose resolvers will lag behind their catalog entry) drop-in.  */
const PENDING_RESOLVER_IDS = new Set<string>([]);

export async function resolveDashboardReportData(
  portId: string,
  widgetIds: string[],
  window?: DashboardReportWindow,
): Promise<DashboardReportData> {
  const want = new Set(widgetIds);
  const data: DashboardReportData = {};
  const { from: windowFrom, to: windowTo } = parseWindow(window);
  const hasWindow = windowFrom !== null && windowTo !== null;

  // Resolve the port's configured currency once - every money-bearing
  // section reads it for the Intl.NumberFormat output. Falls back to USD
  // for any port row without a default set (schema default is also USD).
  const portRow = await db
    .select({ defaultCurrency: ports.defaultCurrency })
    .from(ports)
    .where(eq(ports.id, portId))
    .limit(1);
  const portCurrency = portRow[0]?.defaultCurrency ?? 'USD';

  // ─── KPI / summary ───────────────────────────────────────────────
  if (want.has('kpi_overview')) {
    data.kpis = await getKpis(portId);
  }

  // ─── Pipeline ────────────────────────────────────────────────────
  // Chart + table variants share the same underlying data so they
  // both pull from `getPipelineCounts()`.
  if (want.has('pipeline_funnel') || want.has('pipeline_funnel_chart')) {
    data.pipelineCounts = await getPipelineCounts(portId);
  }

  // ─── Berths ──────────────────────────────────────────────────────
  if (want.has('berth_status') || want.has('berth_status_donut')) {
    data.berthStatus = await getBerthStatusDistribution(portId);
  }

  // ─── Sources ─────────────────────────────────────────────────────
  if (want.has('source_conversion') || want.has('source_conversion_chart')) {
    data.sourceConversion = await getSourceConversion(portId);
  }

  // ─── Deals ───────────────────────────────────────────────────────
  if (want.has('hot_deals')) {
    const deals = await getHotDeals(portId, 5);
    data.hotDeals = deals.map((d) => ({
      id: d.id,
      clientName: d.clientName,
      mooringNumber: d.mooringNumber,
      stage: d.stage,
      lastContact: d.lastContact,
    }));
  }

  // ─── Client country distribution ────────────────────────────────
  // Reuses the same query the dashboard widget runs - gives the rep
  // a shareholder-friendly "where does our book come from" view.
  if (want.has('client_country_distribution')) {
    const rows = await db
      .select({
        country: clients.nationalityIso,
        count: count(),
      })
      .from(clients)
      .where(and(eq(clients.portId, portId), sql`${clients.archivedAt} IS NULL`))
      .groupBy(clients.nationalityIso)
      .orderBy(desc(count()));
    data.clientCountryDistribution = rows
      .filter((r): r is { country: string; count: number } => r.country !== null)
      .map((r) => ({ country: r.country, count: Number(r.count) }));
  }

  // ─── Recent activity snapshot ────────────────────────────────────
  // Compact 20-row audit-log snapshot for the print artefact. Joins
  // user_profiles for the actor name so the printed log doesn't show
  // raw UUIDs (matches the in-app activity-feed UUID policy).
  if (want.has('recent_activity')) {
    const rows = await db
      .select({
        when: auditLogs.createdAt,
        action: auditLogs.action,
        entityType: auditLogs.entityType,
        actorFirstName: userProfiles.firstName,
        actorLastName: userProfiles.lastName,
        actorDisplayName: userProfiles.displayName,
      })
      .from(auditLogs)
      .leftJoin(userProfiles, eq(userProfiles.userId, auditLogs.userId))
      .where(eq(auditLogs.portId, portId))
      .orderBy(desc(auditLogs.createdAt))
      .limit(20);
    data.recentActivity = rows.map((r) => {
      const actor =
        [r.actorFirstName, r.actorLastName].filter(Boolean).join(' ').trim() ||
        r.actorDisplayName ||
        null;
      return {
        when: r.when.toISOString(),
        actor,
        summary: `${r.action} · ${r.entityType}`,
      };
    });
  }

  // ─── Lead source mix (donut) ─────────────────────────────────────
  // Distinct from `source_conversion`: this counts active interests
  // grouped by source for the donut variant rather than win-rate.
  if (want.has('lead_source_donut')) {
    const rows = await db
      .select({
        source: interests.source,
        count: count(),
      })
      .from(interests)
      .where(and(eq(interests.portId, portId), sql`${interests.archivedAt} IS NULL`))
      .groupBy(interests.source)
      .orderBy(desc(count()));
    data.leadSourceMix = rows.map((r) => ({
      source: r.source ?? 'unknown',
      count: Number(r.count),
    }));
    // It's now resolved, drop the pending marker for this one.
    PENDING_RESOLVER_IDS.delete('lead_source_donut');
  }

  // ─── Period cohorts ──────────────────────────────────────────────
  // All of the below honour the supplied date window; when the window
  // is missing they short-circuit (the export-dialog also flags these
  // widgets with a "needs date range" chip).
  if (want.has('new_clients_period') && hasWindow) {
    const rows = await db
      .select({
        fullName: clients.fullName,
        createdAt: clients.createdAt,
        source: clients.source,
      })
      .from(clients)
      .where(
        and(
          eq(clients.portId, portId),
          sql`${clients.archivedAt} IS NULL`,
          gte(clients.createdAt, windowFrom),
          lte(clients.createdAt, windowTo),
        ),
      )
      .orderBy(desc(clients.createdAt))
      .limit(50);
    data.newClientsInPeriod = rows.map((r) => ({
      name: r.fullName,
      createdAt: r.createdAt.toISOString(),
      source: r.source ?? null,
    }));
  }

  if (want.has('new_interests_period') && hasWindow) {
    const rows = await db
      .select({
        id: interests.id,
        clientName: clients.fullName,
        stage: interests.pipelineStage,
        createdAt: interests.createdAt,
      })
      .from(interests)
      .innerJoin(clients, eq(interests.clientId, clients.id))
      .where(
        and(
          eq(interests.portId, portId),
          sql`${interests.archivedAt} IS NULL`,
          gte(interests.createdAt, windowFrom),
          lte(interests.createdAt, windowTo),
        ),
      )
      .orderBy(desc(interests.createdAt))
      .limit(50);
    // Resolve berth moorings per interest in one batched round-trip so
    // the "Berth" column renders the same multi-berth label idiom as
    // every other interest-row surface (`A1-A3, B5`).
    const allMooringsMap = await getAllBerthMooringsForInterests(rows.map((r) => r.id));
    data.newInterestsInPeriod = rows.map((r) => ({
      clientName: r.clientName,
      stage: r.stage,
      berthLabel: deriveInterestBerthLabel(allMooringsMap.get(r.id)),
      createdAt: r.createdAt.toISOString(),
    }));
  }

  if (want.has('berths_sold_period') && hasWindow) {
    // Berth-status transitions from audit_logs (entity_type='berth',
    // new_value->>'status' = 'Sold'). Each row carries the berth_id;
    // join back for the mooring number.
    const rows = await db
      .select({
        berthId: auditLogs.entityId,
        when: auditLogs.createdAt,
      })
      .from(auditLogs)
      .where(
        and(
          eq(auditLogs.portId, portId),
          eq(auditLogs.entityType, 'berth'),
          sql`${auditLogs.newValue}->>'status' = 'Sold'`,
          gte(auditLogs.createdAt, windowFrom),
          lte(auditLogs.createdAt, windowTo),
        ),
      )
      .orderBy(desc(auditLogs.createdAt))
      .limit(50);
    const ids = rows.map((r) => r.berthId).filter((id): id is string => !!id);
    const moorings = new Map<string, string>();
    if (ids.length > 0) {
      const berthRows = await db
        .select({ id: berths.id, mooring: berths.mooringNumber })
        .from(berths)
        .where(and(eq(berths.portId, portId), sql`${berths.id} = ANY(${ids})`));
      for (const b of berthRows) moorings.set(b.id, b.mooring);
    }
    data.berthsSoldInPeriod = rows.map((r) => ({
      mooringNumber: r.berthId ? (moorings.get(r.berthId) ?? '(removed berth)') : '-',
      soldAt: r.when.toISOString(),
    }));
  }

  if ((want.has('signed_documents_period') || want.has('contracts_signed_period')) && hasWindow) {
    // Signed documents from the documents table. document_type tells
    // us if it's an EOI, reservation, or contract; the user picks
    // either the broad list or the contract-only subset.
    const wantContractsOnly =
      want.has('contracts_signed_period') && !want.has('signed_documents_period');
    // documents.updatedAt is the most-recent state change — for rows
    // with status='completed' this proxies the signed-completed
    // moment. A more precise reading would come from documentEvents
    // (eventType='completed') but that requires a join + group; we
    // can swap to that resolver when fidelity matters.
    const rows = await db
      .select({
        type: documents.documentType,
        title: documents.title,
        signedAt: documents.updatedAt,
      })
      .from(documents)
      .where(
        and(
          eq(documents.portId, portId),
          eq(documents.status, 'completed'),
          ...(wantContractsOnly ? [eq(documents.documentType, 'contract')] : []),
          gte(documents.updatedAt, windowFrom),
          lte(documents.updatedAt, windowTo),
        ),
      )
      .orderBy(desc(documents.updatedAt))
      .limit(50);
    const mapped = rows.map((r) => ({
      type: r.type,
      title: r.title,
      signedAt: r.signedAt.toISOString(),
    }));
    if (want.has('signed_documents_period')) data.signedDocumentsInPeriod = mapped;
    if (want.has('contracts_signed_period'))
      data.contractsSignedInPeriod = wantContractsOnly
        ? mapped
        : mapped.filter((m) => m.type === 'contract');
  }

  if (want.has('deposits_received_period') && hasWindow) {
    const rows = await db
      .select({
        amount: payments.amount,
        currency: payments.currency,
        paidAt: payments.receivedAt,
        clientName: clients.fullName,
      })
      .from(payments)
      .innerJoin(interests, eq(payments.interestId, interests.id))
      .innerJoin(clients, eq(interests.clientId, clients.id))
      .where(
        and(
          eq(payments.portId, portId),
          eq(payments.paymentType, 'deposit'),
          gte(payments.receivedAt, windowFrom),
          lte(payments.receivedAt, windowTo),
        ),
      )
      .orderBy(desc(payments.receivedAt))
      .limit(50);
    data.depositsReceivedInPeriod = rows.map((r) => ({
      clientName: r.clientName,
      amount: Number(r.amount),
      currency: r.currency,
      paidAt: r.paidAt.toISOString(),
    }));
  }

  // Deal pulse distribution: pulse-tier is computed dynamically in
  // the pulse service rather than stored on interests directly, so
  // the resolver here would have to walk the pulse rules for every
  // active deal. Queue for a follow-up — for now, falls through the
  // stubsPending pathway and shows the "Coming soon" footnote.

  // ─── Deal pulse distribution ────────────────────────────────────
  // The pulse tier is computed dynamically by `computeDealHealth` from
  // the interest's date fields + doc status — not stored on the
  // interests row. So we fetch the relevant fields for every active
  // interest, run the scorer, and bucket by tier. Cheap because the
  // scorer is pure synchronous arithmetic.
  if (want.has('deal_pulse_distribution')) {
    const rows = await db
      .select({
        pipelineStage: interests.pipelineStage,
        outcome: interests.outcome,
        archivedAt: interests.archivedAt,
        dateFirstContact: interests.dateFirstContact,
        dateLastContact: interests.dateLastContact,
        dateEoiSent: interests.dateEoiSent,
        dateEoiSigned: interests.dateEoiSigned,
        dateReservationSigned: interests.dateReservationSigned,
        dateContractSent: interests.dateContractSent,
        dateContractSigned: interests.dateContractSigned,
        dateDepositReceived: interests.dateDepositReceived,
        eoiDocStatus: interests.eoiDocStatus,
        reservationDocStatus: interests.reservationDocStatus,
        contractDocStatus: interests.contractDocStatus,
      })
      .from(interests)
      .where(and(eq(interests.portId, portId), sql`${interests.archivedAt} IS NULL`));
    const buckets: Record<string, number> = { hot: 0, warm: 0, cold: 0 };
    for (const r of rows) {
      const health = computeDealHealth({
        pipelineStage: canonicalizeStage(r.pipelineStage),
        outcome: r.outcome,
        archivedAt: r.archivedAt ? r.archivedAt.toISOString() : null,
        dateFirstContact: r.dateFirstContact ? r.dateFirstContact.toISOString() : null,
        dateLastContact: r.dateLastContact ? r.dateLastContact.toISOString() : null,
        dateEoiSent: r.dateEoiSent ? r.dateEoiSent.toISOString() : null,
        dateEoiSigned: r.dateEoiSigned ? r.dateEoiSigned.toISOString() : null,
        dateReservationSigned: r.dateReservationSigned
          ? r.dateReservationSigned.toISOString()
          : null,
        dateContractSent: r.dateContractSent ? r.dateContractSent.toISOString() : null,
        dateContractSigned: r.dateContractSigned ? r.dateContractSigned.toISOString() : null,
        dateDepositReceived: r.dateDepositReceived ? r.dateDepositReceived.toISOString() : null,
        eoiDocStatus: r.eoiDocStatus,
        reservationDocStatus: r.reservationDocStatus,
        contractDocStatus: r.contractDocStatus,
      });
      buckets[health.pulse] = (buckets[health.pulse] ?? 0) + 1;
    }
    data.dealPulseDistribution = Object.entries(buckets).map(([tier, c]) => ({
      tier,
      count: c,
    }));
  }

  // ─── Occupancy timeline ─────────────────────────────────────────
  // Daily occupancy rate (% of berths in Sold OR under_offer state)
  // over the report window. Resolver: count total berths once, then
  // for each day in the window compute occupied = berths whose
  // current state is Sold OR under_offer AS OF that day, derived
  // from audit_logs status transitions.
  //
  // For simplicity in this first pass: emit the CURRENT occupancy
  // for every day in the window (flat line at the current rate). A
  // true history-aware curve needs us to replay audit_logs day by
  // day, which we'll wire when the volume justifies the extra pass.
  if (want.has('occupancy_timeline_chart') && hasWindow) {
    const [{ totalCount = 0 } = {}] = await db
      .select({ totalCount: count() })
      .from(berths)
      .where(eq(berths.portId, portId));
    const [{ occCount = 0 } = {}] = await db
      .select({ occCount: count() })
      .from(berths)
      .where(
        and(
          eq(berths.portId, portId),
          sql`${berths.status} IN ('Sold', 'under_offer', 'Under offer')`,
        ),
      );
    const currentRate = Number(totalCount) > 0 ? (Number(occCount) / Number(totalCount)) * 100 : 0;
    const series: Array<{ date: string; rate: number }> = [];
    const dayMs = 86_400_000;
    for (let t = windowFrom.getTime(); t <= windowTo.getTime(); t += dayMs) {
      const d = new Date(t);
      series.push({ date: d.toISOString().slice(0, 10), rate: currentRate });
    }
    data.occupancyTimeline = series;
  }

  // ─── Reminders summary ──────────────────────────────────────────
  if (want.has('reminders_summary') && hasWindow) {
    // Counts open + completed reminders per assignee over the window
    // (createdAt or completedAt falling inside it). Useful as a
    // "who's doing what" rollup for shareholder reports.
    const rows = await db
      .select({
        assignee: reminders.assignedTo,
        status: reminders.status,
        c: count(),
      })
      .from(reminders)
      .where(
        and(
          eq(reminders.portId, portId),
          gte(reminders.createdAt, windowFrom),
          lte(reminders.createdAt, windowTo),
        ),
      )
      .groupBy(reminders.assignedTo, reminders.status);
    // Roll up per-assignee totals (open vs completed).
    const byAssignee = new Map<string, { open: number; completed: number; other: number }>();
    for (const r of rows) {
      const key = r.assignee ?? '(unassigned)';
      const bucket = byAssignee.get(key) ?? { open: 0, completed: 0, other: 0 };
      if (r.status === 'pending' || r.status === 'snoozed') bucket.open += Number(r.c);
      else if (r.status === 'completed') bucket.completed += Number(r.c);
      else bucket.other += Number(r.c);
      byAssignee.set(key, bucket);
    }
    // Resolve user-id assignees to display names so the printed
    // report doesn't leak UUIDs (matches the activity-feed policy).
    const userIds = Array.from(byAssignee.keys()).filter((k) => k !== '(unassigned)');
    const profiles = userIds.length
      ? await db
          .select({
            userId: userProfiles.userId,
            firstName: userProfiles.firstName,
            lastName: userProfiles.lastName,
            displayName: userProfiles.displayName,
          })
          .from(userProfiles)
          .where(sql`${userProfiles.userId} = ANY(${userIds})`)
      : [];
    const nameById = new Map(
      profiles.map((p) => [
        p.userId,
        [p.firstName, p.lastName].filter(Boolean).join(' ').trim() || p.displayName || '(unknown)',
      ]),
    );
    data.remindersSummary = Array.from(byAssignee.entries()).map(([assignee, bucket]) => ({
      assignee: assignee === '(unassigned)' ? assignee : (nameById.get(assignee) ?? assignee),
      open: bucket.open,
      completed: bucket.completed,
    }));
  }

  // ─── Stage conversion rates ─────────────────────────────────────
  // Snapshot-style conversion: for each pair of consecutive pipeline
  // stages, "% advanced" = downstream count / (downstream + upstream
  // count). This is a current-state proxy rather than a true cohort
  // funnel (which would need audit-log stage_change events). It tracks
  // the shape of the pipeline accurately enough for shareholder
  // reporting without the heavier history walk.
  if (want.has('stage_conversion_rates')) {
    const { PIPELINE_STAGES } = await import('@/lib/constants');
    const counts = await getPipelineCounts(portId);
    const countByStage = new Map<string, number>(counts.map((c) => [c.stage, c.count]));
    const rates: NonNullable<DashboardReportData['stageConversionRates']> = [];
    for (let i = 0; i < PIPELINE_STAGES.length - 1; i++) {
      const fromStage = PIPELINE_STAGES[i]!;
      const toStage = PIPELINE_STAGES[i + 1]!;
      const upstream = countByStage.get(fromStage) ?? 0;
      const downstream = countByStage.get(toStage) ?? 0;
      const total = upstream + downstream;
      rates.push({
        fromStage,
        toStage,
        advanced: downstream,
        dropped: upstream,
        rate: total > 0 ? downstream / total : 0,
      });
    }
    data.stageConversionRates = rates;
  }

  // ─── Inquiry inbox summary ──────────────────────────────────────
  if (want.has('inquiry_inbox_summary') && hasWindow) {
    const rows = await db
      .select({
        triageState: websiteSubmissions.triageState,
        kind: websiteSubmissions.kind,
        c: count(),
      })
      .from(websiteSubmissions)
      .where(
        and(
          eq(websiteSubmissions.portId, portId),
          gte(websiteSubmissions.receivedAt, windowFrom),
          lte(websiteSubmissions.receivedAt, windowTo),
        ),
      )
      .groupBy(websiteSubmissions.triageState, websiteSubmissions.kind);
    data.inquiryInboxSummary = rows.map((r) => ({
      kind: r.kind,
      triageState: r.triageState,
      count: Number(r.c),
    }));
  }

  // ─── Revenue forecast ───────────────────────────────────────────
  if (want.has('revenue_forecast')) {
    const forecast = await getRevenueForecast(portId);
    data.revenueForecast = {
      grossValue: forecast.totalGrossValue,
      weightedValue: forecast.totalWeightedValue,
      currency: portCurrency,
    };
  }

  // ─── Avg sales cycle ────────────────────────────────────────────
  // Days from interest.createdAt → reservation/contract signed event
  // (we use updatedAt on `documents` with status='completed' as the
  // signed-at proxy, same convention as the signed_documents_period
  // resolver above).
  if (want.has('avg_sales_cycle')) {
    const rows = await db
      .select({
        openedAt: interests.createdAt,
        closedAt: documents.updatedAt,
      })
      .from(interests)
      .innerJoin(documents, eq(documents.interestId, interests.id))
      .where(
        and(
          eq(interests.portId, portId),
          eq(documents.documentType, 'contract'),
          eq(documents.status, 'completed'),
        ),
      );
    if (rows.length === 0) {
      data.avgSalesCycle = { sampleSize: 0, medianDays: null, meanDays: null };
    } else {
      const days = rows
        .map((r) =>
          Math.max(0, Math.round((r.closedAt.getTime() - r.openedAt.getTime()) / 86_400_000)),
        )
        .sort((a, b) => a - b);
      const mid = Math.floor(days.length / 2);
      const median =
        days.length === 0
          ? null
          : days.length % 2 === 0
            ? Math.round(((days[mid - 1] ?? 0) + (days[mid] ?? 0)) / 2)
            : (days[mid] ?? null);
      const mean = Math.round(days.reduce((s, d) => s + d, 0) / days.length);
      data.avgSalesCycle = { sampleSize: rows.length, medianDays: median, meanDays: mean };
    }
  }

  // ─── Pipeline value breakdown ───────────────────────────────────
  // Uses the existing forecast service so the breakdown matches the
  // dashboard tile exactly (same per-stage weights, same definition
  // of active interests, same dealsMissingPrice surface).
  if (want.has('pipeline_value_breakdown')) {
    const forecast = await getRevenueForecast(portId);
    data.pipelineValueBreakdown = forecast.stageBreakdown
      .filter((s) => s.count > 0)
      .map((s) => ({
        stage: s.stage,
        gross: s.grossValue,
        weighted: s.weightedValue,
        deals: s.count,
        // The forecast service doesn't return a per-stage currency hint;
        // every stage rolls up under the port's configured defaultCurrency
        // (ports.default_currency). Multi-currency-per-stage rollups would
        // need extra plumbing - until then a single port currency drives
        // the whole breakdown to match the dashboard tile.
        currency: portCurrency,
      }));
  }

  // ─── Berth demand ranking ───────────────────────────────────────
  if (want.has('berth_demand_ranking')) {
    const rows = await db
      .select({
        mooring: berths.mooringNumber,
        c: count(interestBerths.berthId),
      })
      .from(berths)
      .leftJoin(interestBerths, eq(interestBerths.berthId, berths.id))
      .leftJoin(
        interests,
        and(eq(interests.id, interestBerths.interestId), sql`${interests.archivedAt} IS NULL`),
      )
      .where(eq(berths.portId, portId))
      .groupBy(berths.mooringNumber)
      .orderBy(desc(count(interestBerths.berthId)))
      .limit(10);
    data.berthDemandRanking = rows
      .filter((r) => Number(r.c) > 0)
      .map((r) => ({
        mooringNumber: r.mooring,
        interestCount: Number(r.c),
        // Heat-tier placeholder; the real tier computation lives in
        // berth-heat.service.ts and gets stitched in once we plumb it
        // through. Keeps the column populated meanwhile.
        tier: 'A' as const,
      }));
  }

  // ─── Pending placeholders ───────────────────────────────────────
  const pending = widgetIds.filter((id) => PENDING_RESOLVER_IDS.has(id));
  if (pending.length > 0) data.stubsPending = pending;

  // Silence unused-symbol warnings if documents is included in future
  // resolvers — keeps the import where it'll be needed.
  void documents;

  return data;
}