pn-new-crm/src/lib/services/search.service.ts

/**
 * Global search service — drives the topbar `CommandSearch` dropdown.
 *
 * Buckets covered: clients (with email/phone via client_contacts JOIN),
 * residential clients, yachts, companies, interests (federated when a
 * berth/yacht/client matches), berths (with linked-interest count),
 * invoices, expenses, documents (by title or signer name/email), files,
 * reminders, brochures, tags (as meta-rows pointing at filtered lists),
 * and a static navigation/settings catalog.
 *
 * Matching strategy per column type:
 *   - Long text fields (full_name, company name, yacht name, descriptions)
 *     use `to_tsvector('simple', col) @@ to_tsquery('simple', "joh:* & smi:*")`
 *     so partial words match mid-typing — `joh smi` finds "John Smith".
 *     A trigram (`similarity()`) fallback is unioned in for typo tolerance
 *     on names ("Jhon" → "John").
 *   - Identifier fields (mooring numbers, hull/registration, tax IDs,
 *     invoice numbers) use `ILIKE '%query%'` with a prefix-anchored bonus
 *     in the ORDER BY.
 *   - Phones are matched by stripping the input down to digits and `+`
 *     and ILIKE-ing against `value_e164` (the canonical normalized form
 *     populated by the i18n PhoneInput pipeline).
 *
 * Permissions: the caller passes `isSuperAdmin` + `permissions`. Each
 * bucket gates itself — viewers don't see invoice/expense rows they
 * couldn't open. The query for the bucket is skipped entirely (cheaper
 * than running it and filtering empty results out).
 *
 * Cross-port (super-admin only): when `includeOtherPorts` is set, a
 * second pass runs the same queries against ports the super-admin can
 * see other than `portId`. Returned in a separate `otherPorts` field
 * so the UI can present them as a dimmed secondary section.
 *
 * Affinity ranking: callers may pass a `recentlyTouchedIds` Set —
 * matching rows whose id is in the set get sorted to the top of their
 * bucket. The id set is derived from the user's last 30 days of
 * `audit_logs` writes (see `getRecentlyTouchedIds`). This is the cheap
 * "your John" vs "some John" boost; we don't try to do per-bucket
 * audit-log JOINs because the boost only matters for at most 5 visible
 * rows and the post-sort is O(n log n) on a tiny n.
 */

import { sql } from 'drizzle-orm';
import { match } from 'ts-pattern';

import { db } from '@/lib/db';
import { redis } from '@/lib/redis';
import type { RolePermissions } from '@/lib/db/schema/users';

// ─── Types ────────────────────────────────────────────────────────────────────

/**
 * Provenance hint for a result row that was surfaced via graph expansion
 * rather than a direct query match. The frontend renders this as a
 * subtitle, e.g. "via Berth A10". `null` (or absent) means the row is
 * a direct match against the user's query.
 */
export interface RelatedVia {
  type: 'berth' | 'interest' | 'client' | 'yacht' | 'company';
  id: string;
  label: string;
}

export interface ClientResult {
  id: string;
  fullName: string;
  matchedContact: string | null;
  matchedContactChannel: 'email' | 'phone' | 'whatsapp' | null;
  archivedAt: string | null;
  relatedVia?: RelatedVia | null;
  /** Short label for which field matched ("name", "email", "phone", "trigram", "expansion"). Used by the dropdown to render "matched on X". */
  matchedOn?: string | null;
}

export interface ResidentialClientResult {
  id: string;
  fullName: string;
  email: string | null;
  phone: string | null;
  status: string;
  archivedAt: string | null;
}

export interface InterestResult {
  id: string;
  clientName: string;
  berthMooringNumber: string | null;
  pipelineStage: string;
  outcome: string | null;
  relatedVia?: RelatedVia | null;
}

export interface ResidentialInterestResult {
  id: string;
  clientName: string;
  pipelineStage: string;
}

export interface BerthResult {
  id: string;
  mooringNumber: string;
  area: string | null;
  status: string;
  linkedInterestCount: number;
  relatedVia?: RelatedVia | null;
}

export interface YachtResult {
  id: string;
  name: string;
  hullNumber: string | null;
  registration: string | null;
  archivedAt: string | null;
  relatedVia?: RelatedVia | null;
}

export interface CompanyResult {
  id: string;
  name: string;
  legalName: string | null;
  taxId: string | null;
  matchedField: 'name' | 'legalName' | 'taxId' | 'billingEmail' | 'registrationNumber' | null;
  archivedAt: string | null;
  relatedVia?: RelatedVia | null;
}

export interface InvoiceResult {
  id: string;
  invoiceNumber: string;
  clientName: string;
  status: string;
  paymentStatus: string | null;
  totalAmount: string | null;
  currency: string;
}

export interface ExpenseResult {
  id: string;
  description: string | null;
  vendor: string | null;
  tripLabel: string | null;
  amount: string;
  currency: string;
  paymentStatus: string | null;
}

export interface DocumentResult {
  id: string;
  title: string;
  documentType: string;
  status: string;
  matchedSignerName: string | null;
}

export interface FileResult {
  id: string;
  filename: string;
  category: string | null;
  /** "client:<name>" | "yacht:<name>" | "company:<name>" — best owner label. */
  ownerLabel: string | null;
}

export interface ReminderResult {
  id: string;
  title: string;
  dueAt: string;
  priority: string;
  status: string;
}

export interface BrochureResult {
  id: string;
  label: string;
  isDefault: boolean;
  archivedAt: string | null;
}

export interface TagResult {
  id: string;
  name: string;
  color: string;
  /** Sum of clients + interests + yachts + companies tagged with this tag. */
  totalCount: number;
}

export interface NavResult {
  /** Stable ID = href, since the catalog is a static set. */
  id: string;
  href: string;
  label: string;
  category: 'settings' | 'admin' | 'dashboard';
}

/**
 * Note-fragment match. Polymorphic across the four note tables
 * (client / interest / yacht / company). Each row carries enough
 * context for the dropdown to show a snippet + parent-entity link
 * without a second round-trip.
 */
export interface NoteResult {
  id: string;
  /** Trimmed snippet of the matching note content. */
  snippet: string;
  /** Source entity type — drives the link target + chip label. */
  source: 'client' | 'interest' | 'yacht' | 'company';
  sourceId: string;
  /** Friendly label for the source (e.g. "Mary Smith", "B17", "Sea Breeze"). */
  sourceLabel: string;
  createdAt: Date;
}

export interface SearchResults {
  clients: ClientResult[];
  residentialClients: ResidentialClientResult[];
  yachts: YachtResult[];
  companies: CompanyResult[];
  interests: InterestResult[];
  residentialInterests: ResidentialInterestResult[];
  berths: BerthResult[];
  invoices: InvoiceResult[];
  expenses: ExpenseResult[];
  documents: DocumentResult[];
  files: FileResult[];
  reminders: ReminderResult[];
  brochures: BrochureResult[];
  tags: TagResult[];
  navigation: NavResult[];
  notes: NoteResult[];
  /**
   * Total count BEFORE per-bucket cap. Lets the UI render
   * "Show 12 more clients" links into the dedicated /search page.
   */
  totals: Record<keyof Omit<SearchResults, 'totals' | 'otherPorts'>, number>;
  /**
   * Cross-port matches (super-admin only when `includeOtherPorts` is set).
   * Each row is annotated with the originating port so the UI can show
   * "Port Amador · Client · Jane Smith".
   */
  otherPorts?: OtherPortResult[];
}

export interface OtherPortResult {
  portId: string;
  portSlug: string;
  portName: string;
  type: 'client' | 'yacht' | 'company' | 'berth' | 'interest';
  id: string;
  label: string;
  sub: string | null;
}

export interface SearchOptions {
  /** Permission shape used to gate buckets. null = super_admin (see all). */
  permissions: RolePermissions | null;
  isSuperAdmin: boolean;
  /** Limit per bucket (default 5 for dropdown, 25 for /search page). */
  limit?: number;
  /** When set, only this bucket's query runs (used by /search?type=clients). */
  type?: keyof Omit<SearchResults, 'totals' | 'otherPorts'>;
  /** Super-admin only — also search ports the user can access other than `portId`. */
  includeOtherPorts?: boolean;
  /** Set of entity ids the user has recently touched (for affinity boost). */
  recentlyTouchedIds?: Set<string>;
}

// ─── Helpers ──────────────────────────────────────────────────────────────────

/**
 * Build a `to_tsquery('simple', $1)` argument from free-text input that
 * does prefix matching per token. Returns null if no usable token is
 * present after sanitization.
 *
 * Sanitization is critical — `to_tsquery` raises a syntax error on
 * unescaped `&`, `|`, `:`, `!`, `(`, `)` etc., and we don't want a query
 * for "AT&T" to fail loudly when the user just wants the obvious match.
 */
export function buildPrefixTsquery(input: string): string | null {
  const tokens = input
    .toLowerCase()
    .split(/\s+/)
    .map((t) => t.replace(/[^a-z0-9_]/g, ''))
    .filter((t) => t.length > 0);
  if (tokens.length === 0) return null;
  return tokens.map((t) => `${t}:*`).join(' & ');
}

/**
 * Normalize a phone-like query to digits-and-plus only so it can be
 * matched against `client_contacts.value_e164` (which stores `+447700…`
 * without spaces or punctuation). Returns null if the result is too
 * short to be meaningfully unique.
 */
export function normalizePhoneQuery(input: string): string | null {
  const digits = input.replace(/[^0-9+]/g, '');
  return digits.length >= 3 ? digits : null;
}

/** Permissions check used to skip buckets the user can't see. */
function can(opts: Pick<SearchOptions, 'permissions' | 'isSuperAdmin'>, dotPath: string): boolean {
  if (opts.isSuperAdmin) return true;
  if (!opts.permissions) return false;
  const parts = dotPath.split('.');
  let cur: unknown = opts.permissions;
  for (const p of parts) {
    if (typeof cur !== 'object' || cur === null) return false;
    cur = (cur as Record<string, unknown>)[p];
  }
  return cur === true;
}

/**
 * Sort matched rows so entries the current user has recently touched
 * float to the top of their bucket. Stable wrt original order otherwise.
 */
function applyAffinity<T extends { id: string }>(rows: T[], touched?: Set<string>): T[] {
  if (!touched || touched.size === 0) return rows;
  const indexed = rows.map((row, idx) => ({ row, idx }));
  indexed.sort((a, b) => {
    const aHit = touched.has(a.row.id) ? 1 : 0;
    const bHit = touched.has(b.row.id) ? 1 : 0;
    if (aHit !== bHit) return bHit - aHit;
    return a.idx - b.idx;
  });
  return indexed.map((x) => x.row);
}

// ─── Affinity source ─────────────────────────────────────────────────────────

/**
 * Returns the set of entity ids the user has interacted with in the
 * last `days` days, capped at `limit` rows. Used to boost ranking so
 * "John" means *your* John, not a stranger.
 *
 * Reads from `audit_logs` which records create/update/delete; this misses
 * pure read-only views, but that's fine — read-only "I just looked at
 * this client" tracking is handled separately by `recently-viewed.service`
 * (different signal, different surface).
 */
export async function getRecentlyTouchedIds(
  userId: string,
  portId: string,
  opts: { days?: number; limit?: number } = {},
): Promise<Set<string>> {
  const days = opts.days ?? 30;
  const limit = opts.limit ?? 200;

  // Order by most-recent-touch so when the cap kicks in we keep the
  // entries with the freshest signal (rather than alphabetically-first).
  const rows = await db.execute<{ entity_id: string }>(sql`
    SELECT entity_id
    FROM (
      SELECT entity_id, MAX(created_at) AS last_at
      FROM audit_logs
      WHERE user_id = ${userId}
        AND port_id = ${portId}
        AND entity_id IS NOT NULL
        AND created_at >= NOW() - (${days}::int * INTERVAL '1 day')
      GROUP BY entity_id
      ORDER BY last_at DESC
      LIMIT ${limit}
    ) recent
  `);

  const set = new Set<string>();
  for (const row of Array.from(rows)) {
    if (row.entity_id) set.add(row.entity_id);
  }
  return set;
}

// ─── Per-bucket queries ──────────────────────────────────────────────────────

const DEFAULT_LIMIT = 5;

// Safe sentinels so we never bind NULL into to_tsquery/ILIKE — Postgres
// evaluation order is unspecified, so a NULL guard in WHERE may not
// reliably short-circuit the function call. These strings are valid in
// every context they're used and never realistically match content.
const NEVER_TSQUERY = 'zzznomatchzzz';
const NEVER_PHONE = '~~no_phone_match~~';

async function searchClients(
  portId: string,
  query: string,
  limit: number,
): Promise<ClientResult[]> {
  const tsQ = buildPrefixTsquery(query) ?? NEVER_TSQUERY;
  const phoneQ = normalizePhoneQuery(query) ?? NEVER_PHONE;
  const ilikePattern = `%${query}%`;

  // Two paths unioned by the OR:
  //   (a) match on full_name via tsvector (prefix per token) OR trigram
  //   (b) match on a contact row (email, phone) via JOIN
  // The DISTINCT ON keeps one row per client even when both name and
  // contact match.
  const rows = await db.execute<{
    id: string;
    full_name: string;
    matched_value: string | null;
    matched_channel: 'email' | 'phone' | 'whatsapp' | null;
    archived_at: Date | null;
    rank: number;
  }>(sql`
    SELECT * FROM (
      SELECT DISTINCT ON (c.id)
        c.id,
        c.full_name,
        -- Only surface the contact value when it actually matched the
        -- query. Otherwise the LEFT JOIN's first-found contact row
        -- would be shown as a misleading "matched on" subtitle.
        CASE
          WHEN cc.value ILIKE ${ilikePattern}
            OR (cc.value_e164 IS NOT NULL AND cc.value_e164 ILIKE ${'%' + phoneQ + '%'})
          THEN cc.value
          ELSE NULL
        END AS matched_value,
        CASE
          WHEN cc.value ILIKE ${ilikePattern}
            OR (cc.value_e164 IS NOT NULL AND cc.value_e164 ILIKE ${'%' + phoneQ + '%'})
          THEN cc.channel
          ELSE NULL
        END AS matched_channel,
        c.archived_at,
        CASE
          WHEN c.full_name ILIKE ${query + '%'} THEN 100
          WHEN c.full_name ILIKE ${ilikePattern} THEN 80
          WHEN to_tsvector('simple', coalesce(c.full_name, ''))
              @@ to_tsquery('simple', ${tsQ}) THEN 70
          WHEN cc.value ILIKE ${ilikePattern} THEN 60
          WHEN cc.value_e164 IS NOT NULL
            AND cc.value_e164 ILIKE ${'%' + phoneQ + '%'} THEN 55
          WHEN similarity(c.full_name, ${query}) > 0.3 THEN 30
          ELSE 0
        END AS rank
      FROM clients c
      LEFT JOIN client_contacts cc ON cc.client_id = c.id
      WHERE c.port_id = ${portId}
        AND c.archived_at IS NULL
        AND (
          c.full_name ILIKE ${ilikePattern}
          OR to_tsvector('simple', coalesce(c.full_name, ''))
              @@ to_tsquery('simple', ${tsQ})
          OR similarity(c.full_name, ${query}) > 0.3
          OR cc.value ILIKE ${ilikePattern}
          OR (
            cc.value_e164 IS NOT NULL
            AND cc.value_e164 ILIKE ${'%' + phoneQ + '%'}
          )
        )
      ORDER BY c.id, rank DESC
    ) sub
    ORDER BY rank DESC, full_name
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => {
    // Tag the rank tier we picked back as a human-readable label so the
    // dropdown can render "matched on name" / "matched on email" without
    // the UI re-doing the comparison. Mirrors the CASE in `rank` above.
    let matchedOn: string | null = null;
    if (r.rank >= 80) matchedOn = 'name';
    else if (r.rank >= 70) matchedOn = 'name';
    else if (r.rank >= 60) matchedOn = r.matched_channel ?? 'contact';
    else if (r.rank >= 55) matchedOn = 'phone';
    else if (r.rank >= 30) matchedOn = 'similar name';
    return {
      id: r.id,
      fullName: r.full_name,
      matchedContact: r.matched_value ?? null,
      matchedContactChannel: r.matched_channel ?? null,
      archivedAt: r.archived_at ? r.archived_at.toISOString() : null,
      matchedOn,
    };
  });
}

async function searchResidentialClients(
  portId: string,
  query: string,
  limit: number,
): Promise<ResidentialClientResult[]> {
  const tsQ = buildPrefixTsquery(query) ?? NEVER_TSQUERY;
  const phoneQ = normalizePhoneQuery(query) ?? NEVER_PHONE;
  const ilikePattern = `%${query}%`;

  const rows = await db.execute<{
    id: string;
    full_name: string;
    email: string | null;
    phone: string | null;
    status: string;
    archived_at: Date | null;
  }>(sql`
    SELECT id, full_name, email, phone, status, archived_at
    FROM residential_clients
    WHERE port_id = ${portId}
      AND archived_at IS NULL
      AND (
        full_name ILIKE ${ilikePattern}
        OR email ILIKE ${ilikePattern}
        OR phone ILIKE ${ilikePattern}
        OR (
          phone_e164 IS NOT NULL
          AND phone_e164 ILIKE ${'%' + phoneQ + '%'}
        )
        OR place_of_residence ILIKE ${ilikePattern}
        OR to_tsvector('simple', coalesce(full_name, ''))
            @@ to_tsquery('simple', ${tsQ})
        OR similarity(full_name, ${query}) > 0.3
      )
    ORDER BY
      CASE
        WHEN full_name ILIKE ${query + '%'} THEN 1
        WHEN full_name ILIKE ${ilikePattern} THEN 2
        WHEN email ILIKE ${query + '%'} THEN 3
        ELSE 4
      END,
      full_name
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    fullName: r.full_name,
    email: r.email ?? null,
    phone: r.phone ?? null,
    status: r.status,
    archivedAt: r.archived_at ? r.archived_at.toISOString() : null,
  }));
}

async function searchYachts(portId: string, query: string, limit: number): Promise<YachtResult[]> {
  const tsQ = buildPrefixTsquery(query) ?? NEVER_TSQUERY;
  const ilikePattern = `%${query}%`;

  const rows = await db.execute<{
    id: string;
    name: string;
    hull_number: string | null;
    registration: string | null;
    archived_at: Date | null;
  }>(sql`
    SELECT id, name, hull_number, registration, archived_at
    FROM yachts
    WHERE port_id = ${portId}
      AND archived_at IS NULL
      AND (
        name ILIKE ${ilikePattern}
        OR hull_number ILIKE ${ilikePattern}
        OR registration ILIKE ${ilikePattern}
        OR flag ILIKE ${ilikePattern}
        OR builder ILIKE ${ilikePattern}
        OR to_tsvector('simple', coalesce(name, '') || ' ' || coalesce(builder, ''))
            @@ to_tsquery('simple', ${tsQ})
        OR similarity(name, ${query}) > 0.3
      )
    ORDER BY
      CASE
        WHEN name ILIKE ${query + '%'} THEN 1
        WHEN name ILIKE ${ilikePattern} THEN 2
        WHEN hull_number ILIKE ${query + '%'} THEN 3
        ELSE 4
      END,
      name
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    name: r.name,
    hullNumber: r.hull_number ?? null,
    registration: r.registration ?? null,
    archivedAt: r.archived_at ? r.archived_at.toISOString() : null,
  }));
}

async function searchCompanies(
  portId: string,
  query: string,
  limit: number,
): Promise<CompanyResult[]> {
  const ilikePattern = `%${query}%`;

  const rows = await db.execute<{
    id: string;
    name: string;
    legal_name: string | null;
    tax_id: string | null;
    matched_field: CompanyResult['matchedField'];
    archived_at: Date | null;
  }>(sql`
    SELECT
      id,
      name,
      legal_name,
      tax_id,
      CASE
        WHEN name ILIKE ${ilikePattern} THEN 'name'
        WHEN legal_name ILIKE ${ilikePattern} THEN 'legalName'
        WHEN tax_id ILIKE ${ilikePattern} THEN 'taxId'
        WHEN registration_number ILIKE ${ilikePattern} THEN 'registrationNumber'
        WHEN billing_email ILIKE ${ilikePattern} THEN 'billingEmail'
      END AS matched_field,
      archived_at
    FROM companies
    WHERE port_id = ${portId}
      AND archived_at IS NULL
      AND (
        name ILIKE ${ilikePattern}
        OR legal_name ILIKE ${ilikePattern}
        OR tax_id ILIKE ${ilikePattern}
        OR registration_number ILIKE ${ilikePattern}
        OR billing_email ILIKE ${ilikePattern}
      )
    ORDER BY
      CASE
        WHEN name ILIKE ${query + '%'} THEN 1
        WHEN name ILIKE ${ilikePattern} THEN 2
        ELSE 3
      END,
      name
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    name: r.name,
    legalName: r.legal_name ?? null,
    taxId: r.tax_id ?? null,
    matchedField: r.matched_field ?? null,
    archivedAt: r.archived_at ? r.archived_at.toISOString() : null,
  }));
}

async function searchInterests(
  portId: string,
  query: string,
  limit: number,
): Promise<InterestResult[]> {
  const ilikePattern = `%${query}%`;

  // Federate: an interest matches if the client name OR the primary
  // berth's mooring number OR the linked yacht's name matches the query.
  // This is the relational expansion the user explicitly asked for —
  // type "A12" and the linked interests show up alongside the berth.
  // Two-step query: DISTINCT ON in the inner subquery dedupes the row
  // explosion from the LEFT JOIN to interest_berths (an interest with
  // 3 non-primary berths would otherwise produce 3 rows). The outer
  // SELECT then applies the human-friendly ordering — open-before-closed,
  // then by pipeline stage. Done as a wrapping subquery because Postgres
  // requires DISTINCT-ON's ORDER BY to lead with the DISTINCT key, but
  // we want the *final* sort to be by outcome.
  const rows = await db.execute<{
    id: string;
    full_name: string;
    mooring_number: string | null;
    pipeline_stage: string;
    outcome: string | null;
  }>(sql`
    SELECT id, full_name, mooring_number, pipeline_stage, outcome
    FROM (
      SELECT DISTINCT ON (i.id)
        i.id,
        c.full_name,
        b.mooring_number,
        i.pipeline_stage,
        i.outcome
      FROM interests i
      JOIN clients c ON i.client_id = c.id
      LEFT JOIN interest_berths ib ON ib.interest_id = i.id AND ib.is_primary = true
      LEFT JOIN berths b ON ib.berth_id = b.id
      LEFT JOIN yachts y ON i.yacht_id = y.id
      WHERE i.port_id = ${portId}
        AND i.archived_at IS NULL
        AND (
          c.full_name ILIKE ${ilikePattern}
          OR b.mooring_number ILIKE ${ilikePattern}
          OR y.name ILIKE ${ilikePattern}
          OR y.hull_number ILIKE ${ilikePattern}
          OR EXISTS (
            SELECT 1 FROM interest_berths ib2
            JOIN berths b2 ON ib2.berth_id = b2.id
            WHERE ib2.interest_id = i.id
              AND b2.mooring_number ILIKE ${ilikePattern}
          )
        )
      ORDER BY i.id
    ) deduped
    ORDER BY
      CASE WHEN outcome IS NULL THEN 0 ELSE 1 END,
      pipeline_stage,
      full_name
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    clientName: r.full_name,
    berthMooringNumber: r.mooring_number ?? null,
    pipelineStage: r.pipeline_stage,
    outcome: r.outcome ?? null,
  }));
}

async function searchResidentialInterests(
  portId: string,
  query: string,
  limit: number,
): Promise<ResidentialInterestResult[]> {
  const ilikePattern = `%${query}%`;

  const rows = await db.execute<{
    id: string;
    full_name: string;
    pipeline_stage: string;
  }>(sql`
    SELECT
      ri.id,
      rc.full_name,
      ri.pipeline_stage
    FROM residential_interests ri
    JOIN residential_clients rc ON ri.residential_client_id = rc.id
    WHERE ri.port_id = ${portId}
      AND ri.archived_at IS NULL
      AND (
        rc.full_name ILIKE ${ilikePattern}
        OR rc.email ILIKE ${ilikePattern}
        OR rc.place_of_residence ILIKE ${ilikePattern}
      )
    ORDER BY rc.full_name
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    clientName: r.full_name,
    pipelineStage: r.pipeline_stage,
  }));
}

async function searchBerths(portId: string, query: string, limit: number): Promise<BerthResult[]> {
  // Mooring numbers are short alphanumeric codes (A1, B12, E18) where
  // prefix-on-number expansion produces confusing UX — typing "A1"
  // when A1 exists shouldn't *also* surface A10, A11, A12. Reps know
  // mooring numbers and almost always type them in full.
  //
  // Strategy: if an exact mooring-number match exists for the query,
  // return ONLY that one row. Otherwise fall back to letter-prefix +
  // number-prefix matching (so typing "A" returns the whole A dock,
  // typing "A1" with no A1 in the DB returns A10/A11/A12, etc.).
  // Area-name matches are also folded into the fallback.
  const trimmed = query.trim();
  const m = /^([A-Za-z]*)(\d*)$/.exec(trimmed);
  const letterPart = (m?.[1] ?? '').toUpperCase();
  const numberPart = m?.[2] ?? '';
  const isStructured = letterPart.length > 0 || numberPart.length > 0;

  const ilikePattern = `%${trimmed}%`;
  const prefixPattern = `${trimmed}%`;

  // First: try for an exact match. Cheap — uses the unique-index on
  // (port_id, mooring_number).
  const exact = await db.execute<{
    id: string;
    mooring_number: string;
    area: string | null;
    status: string;
    linked_interest_count: string;
  }>(sql`
    SELECT
      b.id, b.mooring_number, b.area, b.status,
      (
        SELECT COUNT(*)::text FROM interest_berths ib
        JOIN interests i ON ib.interest_id = i.id
        WHERE ib.berth_id = b.id AND i.archived_at IS NULL
      ) AS linked_interest_count
    FROM berths b
    WHERE b.port_id = ${portId}
      AND UPPER(b.mooring_number) = ${trimmed.toUpperCase()}
    LIMIT 1
  `);

  const exactRows = Array.from(exact);
  if (exactRows.length > 0) {
    return exactRows.map((r) => ({
      id: r.id,
      mooringNumber: r.mooring_number,
      area: r.area ?? null,
      status: r.status,
      linkedInterestCount: Number(r.linked_interest_count) || 0,
    }));
  }

  // No exact match — fall back to letter+number-prefix matching plus
  // a generic area/ILIKE fallback for non-structured queries.
  const rows = await db.execute<{
    id: string;
    mooring_number: string;
    area: string | null;
    status: string;
    linked_interest_count: string;
  }>(sql`
    SELECT
      b.id, b.mooring_number, b.area, b.status,
      (
        SELECT COUNT(*)::text FROM interest_berths ib
        JOIN interests i ON ib.interest_id = i.id
        WHERE ib.berth_id = b.id AND i.archived_at IS NULL
      ) AS linked_interest_count
    FROM berths b
    WHERE b.port_id = ${portId}
      AND (
        ${
          isStructured
            ? sql`(
                regexp_replace(b.mooring_number, '[0-9]+$', '') = ${letterPart}
                AND regexp_replace(b.mooring_number, '^[A-Za-z]+', '') LIKE ${numberPart + '%'}
              )`
            : sql`FALSE`
        }
        OR b.mooring_number ILIKE ${prefixPattern}
        OR b.area ILIKE ${ilikePattern}
      )
    ORDER BY
      CASE
        WHEN b.mooring_number ILIKE ${prefixPattern} THEN 1
        WHEN b.area ILIKE ${prefixPattern} THEN 2
        ELSE 3
      END,
      length(b.mooring_number),
      b.mooring_number
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    mooringNumber: r.mooring_number,
    area: r.area ?? null,
    status: r.status,
    linkedInterestCount: Number(r.linked_interest_count) || 0,
  }));
}

async function searchInvoices(
  portId: string,
  query: string,
  limit: number,
): Promise<InvoiceResult[]> {
  const ilikePattern = `%${query}%`;

  const rows = await db.execute<{
    id: string;
    invoice_number: string;
    client_name: string;
    status: string;
    payment_status: string | null;
    total: string | null;
    currency: string;
  }>(sql`
    SELECT id, invoice_number, client_name, status, payment_status, total, currency
    FROM invoices
    WHERE port_id = ${portId}
      AND (
        invoice_number ILIKE ${ilikePattern}
        OR client_name ILIKE ${ilikePattern}
        OR billing_email ILIKE ${ilikePattern}
      )
    ORDER BY
      CASE
        WHEN invoice_number ILIKE ${query + '%'} THEN 1
        WHEN invoice_number ILIKE ${ilikePattern} THEN 2
        WHEN client_name ILIKE ${query + '%'} THEN 3
        ELSE 4
      END,
      created_at DESC
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    invoiceNumber: r.invoice_number,
    clientName: r.client_name,
    status: r.status,
    paymentStatus: r.payment_status ?? null,
    totalAmount: r.total ?? null,
    currency: r.currency,
  }));
}

async function searchExpenses(
  portId: string,
  query: string,
  limit: number,
): Promise<ExpenseResult[]> {
  const ilikePattern = `%${query}%`;

  const rows = await db.execute<{
    id: string;
    description: string | null;
    establishment_name: string | null;
    trip_label: string | null;
    amount: string;
    currency: string;
    payment_status: string | null;
  }>(sql`
    SELECT id, description, establishment_name, trip_label, amount, currency, payment_status
    FROM expenses
    WHERE port_id = ${portId}
      AND (
        description ILIKE ${ilikePattern}
        OR establishment_name ILIKE ${ilikePattern}
        OR trip_label ILIKE ${ilikePattern}
        OR payment_reference ILIKE ${ilikePattern}
      )
    ORDER BY
      CASE
        WHEN description ILIKE ${query + '%'} THEN 1
        WHEN establishment_name ILIKE ${query + '%'} THEN 2
        ELSE 3
      END,
      expense_date DESC
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    description: r.description ?? null,
    vendor: r.establishment_name ?? null,
    tripLabel: r.trip_label ?? null,
    amount: r.amount,
    currency: r.currency,
    paymentStatus: r.payment_status ?? null,
  }));
}

async function searchDocuments(
  portId: string,
  query: string,
  limit: number,
): Promise<DocumentResult[]> {
  const ilikePattern = `%${query}%`;

  const rows = await db.execute<{
    id: string;
    title: string;
    document_type: string;
    status: string;
    matched_signer_name: string | null;
  }>(sql`
    SELECT DISTINCT ON (d.id)
      d.id,
      d.title,
      d.document_type,
      d.status,
      ds.signer_name AS matched_signer_name
    FROM documents d
    LEFT JOIN document_signers ds ON ds.document_id = d.id
    WHERE d.port_id = ${portId}
      AND (
        d.title ILIKE ${ilikePattern}
        OR ds.signer_name ILIKE ${ilikePattern}
        OR ds.signer_email ILIKE ${ilikePattern}
      )
    ORDER BY d.id, d.created_at DESC
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    title: r.title,
    documentType: r.document_type,
    status: r.status,
    matchedSignerName: r.matched_signer_name ?? null,
  }));
}

async function searchFiles(portId: string, query: string, limit: number): Promise<FileResult[]> {
  const ilikePattern = `%${query}%`;

  const rows = await db.execute<{
    id: string;
    filename: string;
    original_name: string;
    category: string | null;
    client_name: string | null;
    yacht_name: string | null;
    company_name: string | null;
  }>(sql`
    SELECT
      f.id,
      f.filename,
      f.original_name,
      f.category,
      c.full_name AS client_name,
      y.name AS yacht_name,
      co.name AS company_name
    FROM files f
    LEFT JOIN clients c ON f.client_id = c.id
    LEFT JOIN yachts y ON f.yacht_id = y.id
    LEFT JOIN companies co ON f.company_id = co.id
    WHERE f.port_id = ${portId}
      AND (
        f.filename ILIKE ${ilikePattern}
        OR f.original_name ILIKE ${ilikePattern}
      )
    ORDER BY f.created_at DESC
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => {
    let ownerLabel: string | null = null;
    if (r.client_name) ownerLabel = `Client: ${r.client_name}`;
    else if (r.yacht_name) ownerLabel = `Yacht: ${r.yacht_name}`;
    else if (r.company_name) ownerLabel = `Company: ${r.company_name}`;

    return {
      id: r.id,
      filename: r.original_name || r.filename,
      category: r.category ?? null,
      ownerLabel,
    };
  });
}

async function searchReminders(
  portId: string,
  query: string,
  limit: number,
): Promise<ReminderResult[]> {
  const ilikePattern = `%${query}%`;

  const rows = await db.execute<{
    id: string;
    title: string;
    due_at: Date;
    priority: string;
    status: string;
  }>(sql`
    SELECT id, title, due_at, priority, status
    FROM reminders
    WHERE port_id = ${portId}
      AND status NOT IN ('dismissed', 'completed')
      AND (
        title ILIKE ${ilikePattern}
        OR note ILIKE ${ilikePattern}
      )
    ORDER BY due_at ASC
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    title: r.title,
    dueAt: r.due_at.toISOString(),
    priority: r.priority,
    status: r.status,
  }));
}

async function searchBrochures(
  portId: string,
  query: string,
  limit: number,
): Promise<BrochureResult[]> {
  const ilikePattern = `%${query}%`;

  const rows = await db.execute<{
    id: string;
    label: string;
    is_default: boolean;
    archived_at: Date | null;
  }>(sql`
    SELECT id, label, is_default, archived_at
    FROM brochures
    WHERE port_id = ${portId}
      AND archived_at IS NULL
      AND (label ILIKE ${ilikePattern} OR description ILIKE ${ilikePattern})
    ORDER BY is_default DESC, label
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    label: r.label,
    isDefault: r.is_default,
    archivedAt: r.archived_at ? r.archived_at.toISOString() : null,
  }));
}

async function searchTags(portId: string, query: string, limit: number): Promise<TagResult[]> {
  const ilikePattern = `%${query}%`;

  // Tags are meta-rows: clicking one navigates to a filtered list.
  // Show the total count of tagged entities so the user can gauge
  // whether the tag is busy or basically unused.
  const rows = await db.execute<{
    id: string;
    name: string;
    color: string;
    total_count: string;
  }>(sql`
    SELECT
      t.id,
      t.name,
      t.color,
      (
        COALESCE((SELECT COUNT(*) FROM client_tags WHERE tag_id = t.id), 0)
        + COALESCE((SELECT COUNT(*) FROM interest_tags WHERE tag_id = t.id), 0)
        + COALESCE((SELECT COUNT(*) FROM yacht_tags WHERE tag_id = t.id), 0)
        + COALESCE((SELECT COUNT(*) FROM company_tags WHERE tag_id = t.id), 0)
      )::text AS total_count
    FROM tags t
    WHERE t.port_id = ${portId}
      AND t.name ILIKE ${ilikePattern}
    ORDER BY
      CASE WHEN t.name ILIKE ${query + '%'} THEN 1 ELSE 2 END,
      t.name
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    name: r.name,
    color: r.color,
    totalCount: Number(r.total_count) || 0,
  }));
}

async function searchNotes(portId: string, query: string, limit: number): Promise<NoteResult[]> {
  const ilikePattern = `%${query}%`;

  // UNION across the four note tables — keeps the result shape uniform
  // and lets Postgres pick its own join plan per branch. Each branch
  // resolves its parent label inline:
  //   client → client.full_name
  //   interest → primary berth's mooring (falls back to "Interest")
  //   yacht → yacht.name
  //   company → company.name
  //
  // Snippet is hard-trimmed at 140 chars so the dropdown row stays
  // single-line; the full content is one click away on the parent
  // entity's Notes tab.
  const rows = await db.execute<{
    id: string;
    snippet: string;
    source: 'client' | 'interest' | 'yacht' | 'company';
    source_id: string;
    source_label: string | null;
    created_at: Date;
  }>(sql`
    SELECT id, snippet, source, source_id, source_label, created_at
    FROM (
      SELECT
        cn.id,
        SUBSTRING(cn.content FROM 1 FOR 140) AS snippet,
        'client'::text AS source,
        cn.client_id AS source_id,
        c.full_name AS source_label,
        cn.created_at
      FROM client_notes cn
      INNER JOIN clients c ON c.id = cn.client_id
      WHERE c.port_id = ${portId}
        AND cn.content ILIKE ${ilikePattern}

      UNION ALL

      SELECT
        i_n.id,
        SUBSTRING(i_n.content FROM 1 FOR 140) AS snippet,
        'interest'::text AS source,
        i_n.interest_id AS source_id,
        b.mooring_number AS source_label,
        i_n.created_at
      FROM interest_notes i_n
      INNER JOIN interests i ON i.id = i_n.interest_id
      LEFT JOIN interest_berths ib ON ib.interest_id = i.id AND ib.is_primary = true
      LEFT JOIN berths b ON b.id = ib.berth_id
      WHERE i.port_id = ${portId}
        AND i_n.content ILIKE ${ilikePattern}

      UNION ALL

      SELECT
        yn.id,
        SUBSTRING(yn.content FROM 1 FOR 140) AS snippet,
        'yacht'::text AS source,
        yn.yacht_id AS source_id,
        y.name AS source_label,
        yn.created_at
      FROM yacht_notes yn
      INNER JOIN yachts y ON y.id = yn.yacht_id
      WHERE y.port_id = ${portId}
        AND yn.content ILIKE ${ilikePattern}

      UNION ALL

      SELECT
        co_n.id,
        SUBSTRING(co_n.content FROM 1 FOR 140) AS snippet,
        'company'::text AS source,
        co_n.company_id AS source_id,
        co.name AS source_label,
        co_n.created_at
      FROM company_notes co_n
      INNER JOIN companies co ON co.id = co_n.company_id
      WHERE co.port_id = ${portId}
        AND co_n.content ILIKE ${ilikePattern}
    ) t
    ORDER BY created_at DESC
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    id: r.id,
    snippet: r.snippet ?? '',
    source: r.source,
    sourceId: r.source_id,
    sourceLabel: r.source_label ?? labelForSource(r.source),
    createdAt: r.created_at,
  }));
}

function labelForSource(source: 'client' | 'interest' | 'yacht' | 'company'): string {
  return match(source)
    .with('client', () => 'Client')
    .with('interest', () => 'Interest')
    .with('yacht', () => 'Yacht')
    .with('company', () => 'Company')
    .exhaustive();
}

// ─── Cross-port (super admin) ────────────────────────────────────────────────

async function searchOtherPorts(
  excludePortId: string,
  query: string,
  limit: number,
): Promise<OtherPortResult[]> {
  const ilikePattern = `%${query}%`;
  const tsQ = buildPrefixTsquery(query);

  // One UNION query touching the high-signal entities only — clients,
  // yachts, companies, interests, berths. Capped tight (limit applies to
  // the total, not per-bucket) so super-admin cross-port noise stays out
  // of the way.
  const rows = await db.execute<{
    port_id: string;
    port_slug: string;
    port_name: string;
    type: OtherPortResult['type'];
    id: string;
    label: string;
    sub: string | null;
  }>(sql`
    WITH port_lookup AS (
      SELECT id, slug, name FROM ports WHERE id != ${excludePortId}
    )
    SELECT * FROM (
      SELECT p.id AS port_id, p.slug AS port_slug, p.name AS port_name,
             'client'::text AS type, c.id, c.full_name AS label, NULL::text AS sub
      FROM clients c
      JOIN port_lookup p ON c.port_id = p.id
      WHERE c.archived_at IS NULL
        AND (
          c.full_name ILIKE ${ilikePattern}
          OR (
            ${tsQ}::text IS NOT NULL
            AND to_tsvector('simple', coalesce(c.full_name, ''))
                @@ to_tsquery('simple', ${tsQ})
          )
        )
      LIMIT ${limit}
    ) clients_section
    UNION ALL
    SELECT * FROM (
      SELECT p.id AS port_id, p.slug AS port_slug, p.name AS port_name,
             'yacht'::text AS type, y.id, y.name AS label,
             y.hull_number AS sub
      FROM yachts y
      JOIN port_lookup p ON y.port_id = p.id
      WHERE y.archived_at IS NULL
        AND (y.name ILIKE ${ilikePattern} OR y.hull_number ILIKE ${ilikePattern})
      LIMIT ${limit}
    ) yachts_section
    UNION ALL
    SELECT * FROM (
      SELECT p.id AS port_id, p.slug AS port_slug, p.name AS port_name,
             'company'::text AS type, co.id, co.name AS label, co.tax_id AS sub
      FROM companies co
      JOIN port_lookup p ON co.port_id = p.id
      WHERE co.archived_at IS NULL
        AND (co.name ILIKE ${ilikePattern} OR co.tax_id ILIKE ${ilikePattern})
      LIMIT ${limit}
    ) companies_section
    UNION ALL
    SELECT * FROM (
      SELECT p.id AS port_id, p.slug AS port_slug, p.name AS port_name,
             'berth'::text AS type, b.id, b.mooring_number AS label, b.area AS sub
      FROM berths b
      JOIN port_lookup p ON b.port_id = p.id
      WHERE b.mooring_number ILIKE ${ilikePattern} OR b.mooring_number % ${query}
      LIMIT ${limit}
    ) berths_section
    LIMIT ${limit}
  `);

  return Array.from(rows).map((r) => ({
    portId: r.port_id,
    portSlug: r.port_slug,
    portName: r.port_name,
    type: r.type,
    id: r.id,
    label: r.label,
    sub: r.sub ?? null,
  }));
}

// ─── Public entrypoint ──────────────────────────────────────────────────────

/**
 * Graph expansion — for every direct match in a search, fetch the
 * 1-hop related entities and add them to the appropriate bucket.
 *
 *   Berth match    → its interests + their clients + their yachts
 *   Interest match → its berth + client + yacht
 *   Client match   → their interests + their owned yachts + companies
 *                    they're members of
 *   Yacht match    → its interests + its owner (client/company)
 *   Company match  → its members (clients) + their interests
 *
 * Depth limited to 1 hop to avoid quadratic fan-out. Each expansion row
 * carries a `relatedVia` hint so the UI can show "via Berth A10" beneath
 * the row's title.
 *
 * Rows that are already a direct match are NOT duplicated — the dedupe
 * runs on `id`. Direct matches always take precedence (their relatedVia
 * stays unset).
 */
async function expandGraph(
  portId: string,
  direct: {
    berthIds: string[];
    interestIds: string[];
    clientIds: string[];
    yachtIds: string[];
    companyIds: string[];
  },
  perBucketCap: number,
): Promise<{
  interests: InterestResult[];
  clients: ClientResult[];
  yachts: YachtResult[];
  companies: CompanyResult[];
  berths: BerthResult[];
}> {
  // Helper: SQL-safe ANY() needs a non-empty array; bail early otherwise.
  const hasAny = (arr: string[]) => arr.length > 0;

  // ─── Berth → Interests (and their clients + yachts) ─────────────────
  const interestsFromBerths = hasAny(direct.berthIds)
    ? await db.execute<{
        id: string;
        client_name: string;
        mooring_number: string;
        pipeline_stage: string;
        outcome: string | null;
        via_berth_id: string;
        via_berth_label: string;
      }>(sql`
        SELECT
          i.id,
          c.full_name AS client_name,
          b.mooring_number,
          i.pipeline_stage,
          i.outcome,
          b.id AS via_berth_id,
          b.mooring_number AS via_berth_label
        FROM interest_berths ib
        JOIN interests i ON ib.interest_id = i.id
        JOIN clients c ON i.client_id = c.id
        JOIN berths b ON ib.berth_id = b.id
        WHERE ib.berth_id IN (${sql.join(
          direct.berthIds.map((id) => sql`${id}`),
          sql`, `,
        )})
          AND i.port_id = ${portId}
          AND i.archived_at IS NULL
        ORDER BY ib.is_primary DESC, i.created_at DESC
        LIMIT ${perBucketCap * direct.berthIds.length}
      `)
    : [];

  // ─── Interest → Berth, Client, Yacht ─────────────────────────────────
  // For interests that matched directly, surface their connected berth +
  // client + yacht as related entries in those buckets.
  const fromInterests = hasAny(direct.interestIds)
    ? await db.execute<{
        interest_id: string;
        client_id: string;
        client_name: string;
        yacht_id: string | null;
        yacht_name: string | null;
        berth_id: string | null;
        mooring_number: string | null;
        berth_area: string | null;
        berth_status: string | null;
      }>(sql`
        SELECT
          i.id AS interest_id,
          c.id AS client_id,
          c.full_name AS client_name,
          y.id AS yacht_id,
          y.name AS yacht_name,
          b.id AS berth_id,
          b.mooring_number,
          b.area AS berth_area,
          b.status AS berth_status
        FROM interests i
        JOIN clients c ON i.client_id = c.id
        LEFT JOIN yachts y ON i.yacht_id = y.id
        LEFT JOIN LATERAL (
          SELECT b.* FROM interest_berths ib2
          JOIN berths b ON ib2.berth_id = b.id
          WHERE ib2.interest_id = i.id
          ORDER BY ib2.is_primary DESC
          LIMIT 1
        ) b ON TRUE
        WHERE i.id IN (${sql.join(
          direct.interestIds.map((id) => sql`${id}`),
          sql`, `,
        )})
          AND i.port_id = ${portId}
      `)
    : [];

  // ─── Client → Interests, Owned Yachts, Member Companies ──────────────
  const fromClients = hasAny(direct.clientIds)
    ? await Promise.all([
        // Their interests
        db.execute<{
          id: string;
          client_id: string;
          client_name: string;
          mooring_number: string | null;
          pipeline_stage: string;
          outcome: string | null;
        }>(sql`
          SELECT i.id, i.client_id, c.full_name AS client_name,
            b.mooring_number, i.pipeline_stage, i.outcome
          FROM interests i
          JOIN clients c ON i.client_id = c.id
          LEFT JOIN LATERAL (
            SELECT b.mooring_number FROM interest_berths ib
            JOIN berths b ON ib.berth_id = b.id
            WHERE ib.interest_id = i.id
            ORDER BY ib.is_primary DESC LIMIT 1
          ) b ON TRUE
          WHERE i.client_id IN (${sql.join(
            direct.clientIds.map((id) => sql`${id}`),
            sql`, `,
          )})
            AND i.port_id = ${portId}
            AND i.archived_at IS NULL
          ORDER BY i.created_at DESC
          LIMIT ${perBucketCap * direct.clientIds.length}
        `),
        // Yachts they own (current_owner_type='client')
        db.execute<{
          id: string;
          name: string;
          hull_number: string | null;
          registration: string | null;
          archived_at: string | null;
          owner_id: string;
          owner_name: string;
        }>(sql`
          SELECT y.id, y.name, y.hull_number, y.registration, y.archived_at::text,
            c.id AS owner_id, c.full_name AS owner_name
          FROM yachts y
          JOIN clients c ON y.current_owner_id = c.id
          WHERE y.current_owner_type = 'client'
            AND y.current_owner_id IN (${sql.join(
              direct.clientIds.map((id) => sql`${id}`),
              sql`, `,
            )})
            AND y.port_id = ${portId}
          ORDER BY y.name
          LIMIT ${perBucketCap * direct.clientIds.length}
        `),
        // Companies they're members of
        db.execute<{
          id: string;
          name: string;
          legal_name: string | null;
          tax_id: string | null;
          archived_at: string | null;
          via_client_id: string;
          via_client_name: string;
        }>(sql`
          SELECT co.id, co.name, co.legal_name, co.tax_id, co.archived_at::text,
            c.id AS via_client_id, c.full_name AS via_client_name
          FROM company_memberships cm
          JOIN companies co ON cm.company_id = co.id
          JOIN clients c ON cm.client_id = c.id
          WHERE cm.client_id IN (${sql.join(
            direct.clientIds.map((id) => sql`${id}`),
            sql`, `,
          )})
            AND cm.end_date IS NULL
            AND co.port_id = ${portId}
          ORDER BY co.name
          LIMIT ${perBucketCap * direct.clientIds.length}
        `),
      ])
    : [[], [], []];

  // ─── Yacht → Interests, Owner ───────────────────────────────────────
  const fromYachts = hasAny(direct.yachtIds)
    ? await Promise.all([
        // Interests on these yachts
        db.execute<{
          id: string;
          client_name: string;
          mooring_number: string | null;
          pipeline_stage: string;
          outcome: string | null;
          via_yacht_id: string;
          via_yacht_name: string;
        }>(sql`
          SELECT i.id, c.full_name AS client_name,
            b.mooring_number, i.pipeline_stage, i.outcome,
            y.id AS via_yacht_id, y.name AS via_yacht_name
          FROM interests i
          JOIN clients c ON i.client_id = c.id
          JOIN yachts y ON i.yacht_id = y.id
          LEFT JOIN LATERAL (
            SELECT b.mooring_number FROM interest_berths ib
            JOIN berths b ON ib.berth_id = b.id
            WHERE ib.interest_id = i.id
            ORDER BY ib.is_primary DESC LIMIT 1
          ) b ON TRUE
          WHERE i.yacht_id IN (${sql.join(
            direct.yachtIds.map((id) => sql`${id}`),
            sql`, `,
          )})
            AND i.port_id = ${portId}
            AND i.archived_at IS NULL
          ORDER BY i.created_at DESC
          LIMIT ${perBucketCap * direct.yachtIds.length}
        `),
        // Owners (client + company variants via polymorphic FKs)
        db.execute<{
          yacht_id: string;
          yacht_name: string;
          owner_type: string;
          owner_id: string;
          owner_label: string;
        }>(sql`
          SELECT y.id AS yacht_id, y.name AS yacht_name,
            y.current_owner_type AS owner_type,
            COALESCE(c.id, co.id) AS owner_id,
            COALESCE(c.full_name, co.name) AS owner_label
          FROM yachts y
          LEFT JOIN clients c
            ON y.current_owner_type = 'client' AND y.current_owner_id = c.id
          LEFT JOIN companies co
            ON y.current_owner_type = 'company' AND y.current_owner_id = co.id
          WHERE y.id IN (${sql.join(
            direct.yachtIds.map((id) => sql`${id}`),
            sql`, `,
          )})
            AND y.port_id = ${portId}
            AND y.current_owner_id IS NOT NULL
        `),
      ])
    : [[], []];

  // ─── Company → Members (Clients), their Interests ────────────────────
  const fromCompanies = hasAny(direct.companyIds)
    ? await Promise.all([
        db.execute<{
          id: string;
          full_name: string;
          archived_at: string | null;
          via_company_id: string;
          via_company_name: string;
        }>(sql`
          SELECT c.id, c.full_name, c.archived_at::text,
            co.id AS via_company_id, co.name AS via_company_name
          FROM company_memberships cm
          JOIN clients c ON cm.client_id = c.id
          JOIN companies co ON cm.company_id = co.id
          WHERE cm.company_id IN (${sql.join(
            direct.companyIds.map((id) => sql`${id}`),
            sql`, `,
          )})
            AND cm.end_date IS NULL
            AND c.port_id = ${portId}
          ORDER BY c.full_name
          LIMIT ${perBucketCap * direct.companyIds.length}
        `),
      ])
    : [[]];

  // ─── Marshal into bucket-shaped result rows ──────────────────────────
  const expandedInterests: InterestResult[] = [];
  const expandedClients: ClientResult[] = [];
  const expandedYachts: YachtResult[] = [];
  const expandedCompanies: CompanyResult[] = [];
  const expandedBerths: BerthResult[] = [];

  // From berths
  for (const r of Array.from(interestsFromBerths)) {
    expandedInterests.push({
      id: r.id,
      clientName: r.client_name,
      berthMooringNumber: r.mooring_number,
      pipelineStage: r.pipeline_stage,
      outcome: r.outcome,
      relatedVia: { type: 'berth', id: r.via_berth_id, label: `Berth ${r.via_berth_label}` },
    });
  }

  // From interests (the matched row's client, yacht, berth)
  for (const r of Array.from(fromInterests)) {
    if (r.client_id) {
      expandedClients.push({
        id: r.client_id,
        fullName: r.client_name,
        matchedContact: null,
        matchedContactChannel: null,
        archivedAt: null,
        relatedVia: { type: 'interest', id: r.interest_id, label: `Interest · ${r.client_name}` },
      });
    }
    if (r.yacht_id) {
      expandedYachts.push({
        id: r.yacht_id,
        name: r.yacht_name ?? '(unnamed yacht)',
        hullNumber: null,
        registration: null,
        archivedAt: null,
        relatedVia: { type: 'interest', id: r.interest_id, label: `Interest · ${r.client_name}` },
      });
    }
    if (r.berth_id) {
      expandedBerths.push({
        id: r.berth_id,
        mooringNumber: r.mooring_number ?? '',
        area: r.berth_area,
        status: r.berth_status ?? 'available',
        linkedInterestCount: 0,
        relatedVia: { type: 'interest', id: r.interest_id, label: `Interest · ${r.client_name}` },
      });
    }
  }

  // From clients
  const [clientInterests, clientYachts, clientCompanies] = fromClients;
  for (const r of Array.from(clientInterests)) {
    expandedInterests.push({
      id: r.id,
      clientName: r.client_name,
      berthMooringNumber: r.mooring_number,
      pipelineStage: r.pipeline_stage,
      outcome: r.outcome,
      relatedVia: { type: 'client', id: r.client_id, label: r.client_name },
    });
  }
  for (const r of Array.from(clientYachts)) {
    expandedYachts.push({
      id: r.id,
      name: r.name,
      hullNumber: r.hull_number,
      registration: r.registration,
      archivedAt: r.archived_at,
      relatedVia: { type: 'client', id: r.owner_id, label: r.owner_name },
    });
  }
  for (const r of Array.from(clientCompanies)) {
    expandedCompanies.push({
      id: r.id,
      name: r.name,
      legalName: r.legal_name,
      taxId: r.tax_id,
      matchedField: null,
      archivedAt: r.archived_at,
      relatedVia: { type: 'client', id: r.via_client_id, label: r.via_client_name },
    });
  }

  // From yachts
  const [yachtInterests, yachtOwners] = fromYachts;
  for (const r of Array.from(yachtInterests)) {
    expandedInterests.push({
      id: r.id,
      clientName: r.client_name,
      berthMooringNumber: r.mooring_number,
      pipelineStage: r.pipeline_stage,
      outcome: r.outcome,
      relatedVia: { type: 'yacht', id: r.via_yacht_id, label: r.via_yacht_name },
    });
  }
  for (const r of Array.from(yachtOwners)) {
    if (!r.owner_id) continue;
    if (r.owner_type === 'client') {
      expandedClients.push({
        id: r.owner_id,
        fullName: r.owner_label,
        matchedContact: null,
        matchedContactChannel: null,
        archivedAt: null,
        relatedVia: { type: 'yacht', id: r.yacht_id, label: r.yacht_name },
      });
    } else if (r.owner_type === 'company') {
      expandedCompanies.push({
        id: r.owner_id,
        name: r.owner_label,
        legalName: null,
        taxId: null,
        matchedField: null,
        archivedAt: null,
        relatedVia: { type: 'yacht', id: r.yacht_id, label: r.yacht_name },
      });
    }
  }

  // From companies
  const [companyMembers] = fromCompanies;
  for (const r of Array.from(companyMembers)) {
    expandedClients.push({
      id: r.id,
      fullName: r.full_name,
      matchedContact: null,
      matchedContactChannel: null,
      archivedAt: r.archived_at,
      relatedVia: { type: 'company', id: r.via_company_id, label: r.via_company_name },
    });
  }

  return {
    interests: expandedInterests,
    clients: expandedClients,
    yachts: expandedYachts,
    companies: expandedCompanies,
    berths: expandedBerths,
  };
}

/**
 * Merge direct-match rows with graph-expansion rows. Direct matches
 * (those without `relatedVia` set) take precedence — if a row appears
 * in both, the direct version wins. Direct matches sort before
 * related matches.
 */
function mergeWithExpansion<T extends { id: string; relatedVia?: RelatedVia | null }>(
  direct: T[],
  expansion: T[],
  cap: number,
): T[] {
  const seen = new Set(direct.map((r) => r.id));
  const merged = [
    ...direct.map((r) => ({ ...r, relatedVia: null as RelatedVia | null })),
    ...expansion.filter((r) => !seen.has(r.id) && (seen.add(r.id), true)),
  ];
  return merged.slice(0, cap);
}

/**
 * Returns a populated `SearchResults` for the given port + query. All
 * unrequested or permission-denied buckets come back as empty arrays so
 * the UI can render uniformly.
 *
 * Per-bucket queries are run in parallel via `Promise.all` — total
 * latency is bounded by the single slowest bucket.
 *
 * Graph expansion: after the direct-match phase, related entities are
 * fetched in a single second pass (`expandGraph`) so reps searching for
 * one entity see everything connected to it. See expandGraph docstring.
 */
export async function search(
  portId: string,
  query: string,
  opts: SearchOptions,
): Promise<SearchResults> {
  const limit = opts.limit ?? DEFAULT_LIMIT;
  const empty = emptyResults();
  if (!query || query.trim().length < 1) return empty;

  // Single-bucket mode (used by /search?type=clients) — skip everything
  // else for speed. Graph-expansion buckets (clients, yachts, companies,
  // interests, berths) fall through to the full pipeline below so that
  // related-via matches survive the chip narrow — otherwise typing
  // "carlos vega" with the Yachts chip selected would return zero rows
  // even though the All chip shows "Yachts (1)" (the yacht owned by
  // Carlos, surfaced through expandGraph). We trim to the requested
  // bucket at the end.
  type GraphBucket = 'clients' | 'yachts' | 'companies' | 'interests' | 'berths';
  const GRAPH_BUCKETS: GraphBucket[] = ['clients', 'yachts', 'companies', 'interests', 'berths'];
  const narrowTo: GraphBucket | null =
    opts.type && (GRAPH_BUCKETS as readonly string[]).includes(opts.type)
      ? (opts.type as GraphBucket)
      : null;
  if (opts.type && !narrowTo) return runSingleBucket(portId, query, limit, opts);

  // We always run the name-bearing buckets even for email/phone-shaped
  // queries — a client named "test+marketing" is rare but real.

  const [
    clients,
    residentialClients,
    yachts,
    companies,
    interests,
    residentialInterests,
    berths,
    invoices,
    expenses,
    documents,
    files,
    reminders,
    brochures,
    tags,
    notes,
    otherPorts,
  ] = await Promise.all([
    can(opts, 'clients.view') ? searchClients(portId, query, limit) : Promise.resolve([]),
    can(opts, 'residential_clients.view')
      ? searchResidentialClients(portId, query, limit)
      : Promise.resolve([]),
    can(opts, 'yachts.view') ? searchYachts(portId, query, limit) : Promise.resolve([]),
    can(opts, 'companies.view') ? searchCompanies(portId, query, limit) : Promise.resolve([]),
    can(opts, 'interests.view') ? searchInterests(portId, query, limit) : Promise.resolve([]),
    can(opts, 'residential_interests.view') || can(opts, 'residential_clients.view')
      ? searchResidentialInterests(portId, query, limit)
      : Promise.resolve([]),
    can(opts, 'berths.view') ? searchBerths(portId, query, limit) : Promise.resolve([]),
    can(opts, 'invoices.view') ? searchInvoices(portId, query, limit) : Promise.resolve([]),
    can(opts, 'expenses.view') ? searchExpenses(portId, query, limit) : Promise.resolve([]),
    can(opts, 'documents.view') ? searchDocuments(portId, query, limit) : Promise.resolve([]),
    can(opts, 'files.view') || can(opts, 'documents.view')
      ? searchFiles(portId, query, limit)
      : Promise.resolve([]),
    can(opts, 'reminders.view') || can(opts, 'clients.view')
      ? searchReminders(portId, query, limit)
      : Promise.resolve([]),
    can(opts, 'admin.manage_settings')
      ? searchBrochures(portId, query, limit)
      : Promise.resolve([]),
    searchTags(portId, query, limit),
    // Notes search runs whenever the user can read any note-bearing
    // entity. Reads are gated by the JOINs in searchNotes itself —
    // a note's row only surfaces when its parent entity is in this
    // port. The dropdown UI sticks notes at the bottom (per the
    // user's "low-noise" preference).
    can(opts, 'clients.view') ||
    can(opts, 'interests.view') ||
    can(opts, 'yachts.view') ||
    can(opts, 'companies.view')
      ? searchNotes(portId, query, limit)
      : Promise.resolve([]),
    opts.includeOtherPorts && opts.isSuperAdmin
      ? searchOtherPorts(portId, query, limit)
      : Promise.resolve([]),
  ]);

  const navigation = await Promise.resolve(
    (await import('@/lib/services/search-nav-catalog')).searchNavCatalog(query, {
      isSuperAdmin: opts.isSuperAdmin,
      permissions: opts.permissions,
      limit,
    }),
  ).then((entries) =>
    entries.map((e) => ({
      id: e.href,
      href: e.href,
      label: e.label,
      category: e.category,
    })),
  );

  // ─── Phase 2: graph expansion ───────────────────────────────────────
  // For every direct match, fetch its 1-hop related entities so reps
  // who search "A10" see the linked interests/clients/yachts/companies
  // surface alongside the berth. See `expandGraph` docstring for the
  // relationship map and per-bucket caps.
  //
  // Latency optimization: when every relationship-bearing bucket already
  // has the maximum number of direct matches the dropdown will render,
  // graph expansion only adds rows that get truncated downstream — skip
  // the (cross-table-heavy) expansion query entirely. Saves the biggest
  // single SQL call in the search path on common-term queries.
  const allBucketsFull =
    clients.length >= limit &&
    yachts.length >= limit &&
    companies.length >= limit &&
    interests.length >= limit &&
    berths.length >= limit;

  const expanded = allBucketsFull
    ? {
        interests: [] as InterestResult[],
        clients: [] as ClientResult[],
        yachts: [] as YachtResult[],
        companies: [] as CompanyResult[],
        berths: [] as BerthResult[],
      }
    : await expandGraph(
        portId,
        {
          berthIds: berths.map((b) => b.id),
          interestIds: interests.map((i) => i.id),
          clientIds: clients.map((c) => c.id),
          yachtIds: yachts.map((y) => y.id),
          companyIds: companies.map((c) => c.id),
        },
        limit,
      );

  const apply = <T extends { id: string }>(rows: T[]) =>
    applyAffinity(rows, opts.recentlyTouchedIds);

  // Merge direct matches with expansion rows; direct rows always win
  // ties and sort first. Each bucket caps at `limit * 2` so reps still
  // see the full direct-match set plus a healthy expansion tail.
  //
  // SECURITY (search-auditor H1): expandGraph runs unconditionally,
  // but its results MUST be re-gated by the destination bucket's view
  // permission. A user with berths.view but no interests.view searching
  // "A12" was previously getting interest rows (client name + stage)
  // surfaced via expansion. Gate each merge call so the expansion
  // contributes empty rows for any bucket the caller can't see.
  const mergedClients = mergeWithExpansion(
    clients,
    can(opts, 'clients.view') ? expanded.clients : [],
    limit * 2,
  );
  const mergedInterests = mergeWithExpansion(
    interests,
    can(opts, 'interests.view') ? expanded.interests : [],
    limit * 2,
  );
  const mergedYachts = mergeWithExpansion(
    yachts,
    can(opts, 'yachts.view') ? expanded.yachts : [],
    limit * 2,
  );
  const mergedCompanies = mergeWithExpansion(
    companies,
    can(opts, 'companies.view') ? expanded.companies : [],
    limit * 2,
  );
  const mergedBerths = mergeWithExpansion(
    berths,
    can(opts, 'berths.view') ? expanded.berths : [],
    limit * 2,
  );

  const result: SearchResults = {
    clients: apply(mergedClients),
    residentialClients: apply(residentialClients),
    yachts: apply(mergedYachts),
    companies: apply(mergedCompanies),
    interests: apply(mergedInterests),
    residentialInterests: apply(residentialInterests),
    berths: apply(mergedBerths),
    invoices: apply(invoices),
    expenses: apply(expenses),
    documents: apply(documents),
    files: apply(files),
    reminders: apply(reminders),
    brochures: apply(brochures),
    tags,
    navigation,
    notes,
    totals: {
      clients: mergedClients.length,
      residentialClients: residentialClients.length,
      yachts: mergedYachts.length,
      companies: mergedCompanies.length,
      interests: mergedInterests.length,
      residentialInterests: residentialInterests.length,
      berths: mergedBerths.length,
      invoices: invoices.length,
      expenses: expenses.length,
      documents: documents.length,
      files: files.length,
      reminders: reminders.length,
      brochures: brochures.length,
      tags: tags.length,
      navigation: navigation.length,
      notes: notes.length,
    },
    otherPorts: otherPorts.length > 0 ? otherPorts : undefined,
  };

  // When narrowing to a graph bucket, zero out every other bucket so the
  // dropdown only renders the chosen one. Totals for the other buckets
  // stay populated so the chip row still shows their counts — the client
  // already snapshots the last "all" totals separately, but keeping them
  // here means a direct API hit with ?type=yachts still sees all chip
  // counts for free.
  if (narrowTo) {
    const keep = narrowTo;
    return {
      ...emptyResults(),
      [keep]: result[keep],
      totals: result.totals,
      otherPorts: result.otherPorts,
    } as SearchResults;
  }

  return result;
}

async function runSingleBucket(
  portId: string,
  query: string,
  limit: number,
  opts: SearchOptions,
): Promise<SearchResults> {
  const empty = emptyResults();
  // Defensive: callers always pass a defined type here (the dispatch above
  // narrows it), but the SearchOptions type leaves it optional so we
  // explicitly handle undefined for the type-checker.
  if (!opts.type) return empty;
  return match(opts.type)
    .with('clients', async () => {
      if (!can(opts, 'clients.view')) return empty;
      empty.clients = applyAffinity(
        await searchClients(portId, query, limit),
        opts.recentlyTouchedIds,
      );
      empty.totals.clients = empty.clients.length;
      return empty;
    })
    .with('residentialClients', async () => {
      if (!can(opts, 'residential_clients.view')) return empty;
      empty.residentialClients = applyAffinity(
        await searchResidentialClients(portId, query, limit),
        opts.recentlyTouchedIds,
      );
      empty.totals.residentialClients = empty.residentialClients.length;
      return empty;
    })
    .with('yachts', async () => {
      if (!can(opts, 'yachts.view')) return empty;
      empty.yachts = applyAffinity(
        await searchYachts(portId, query, limit),
        opts.recentlyTouchedIds,
      );
      empty.totals.yachts = empty.yachts.length;
      return empty;
    })
    .with('companies', async () => {
      if (!can(opts, 'companies.view')) return empty;
      empty.companies = applyAffinity(
        await searchCompanies(portId, query, limit),
        opts.recentlyTouchedIds,
      );
      empty.totals.companies = empty.companies.length;
      return empty;
    })
    .with('interests', async () => {
      if (!can(opts, 'interests.view')) return empty;
      empty.interests = applyAffinity(
        await searchInterests(portId, query, limit),
        opts.recentlyTouchedIds,
      );
      empty.totals.interests = empty.interests.length;
      return empty;
    })
    .with('residentialInterests', async () => {
      if (!can(opts, 'residential_clients.view')) return empty;
      empty.residentialInterests = applyAffinity(
        await searchResidentialInterests(portId, query, limit),
        opts.recentlyTouchedIds,
      );
      empty.totals.residentialInterests = empty.residentialInterests.length;
      return empty;
    })
    .with('berths', async () => {
      if (!can(opts, 'berths.view')) return empty;
      empty.berths = applyAffinity(
        await searchBerths(portId, query, limit),
        opts.recentlyTouchedIds,
      );
      empty.totals.berths = empty.berths.length;
      return empty;
    })
    .with('invoices', async () => {
      if (!can(opts, 'invoices.view')) return empty;
      empty.invoices = applyAffinity(
        await searchInvoices(portId, query, limit),
        opts.recentlyTouchedIds,
      );
      empty.totals.invoices = empty.invoices.length;
      return empty;
    })
    .with('expenses', async () => {
      if (!can(opts, 'expenses.view')) return empty;
      empty.expenses = applyAffinity(
        await searchExpenses(portId, query, limit),
        opts.recentlyTouchedIds,
      );
      empty.totals.expenses = empty.expenses.length;
      return empty;
    })
    .with('documents', async () => {
      if (!can(opts, 'documents.view')) return empty;
      empty.documents = applyAffinity(
        await searchDocuments(portId, query, limit),
        opts.recentlyTouchedIds,
      );
      empty.totals.documents = empty.documents.length;
      return empty;
    })
    .with('files', async () => {
      if (!can(opts, 'files.view') && !can(opts, 'documents.view')) return empty;
      empty.files = applyAffinity(await searchFiles(portId, query, limit), opts.recentlyTouchedIds);
      empty.totals.files = empty.files.length;
      return empty;
    })
    .with('reminders', async () => {
      empty.reminders = applyAffinity(
        await searchReminders(portId, query, limit),
        opts.recentlyTouchedIds,
      );
      empty.totals.reminders = empty.reminders.length;
      return empty;
    })
    .with('brochures', async () => {
      if (!can(opts, 'admin.manage_settings')) return empty;
      empty.brochures = await searchBrochures(portId, query, limit);
      empty.totals.brochures = empty.brochures.length;
      return empty;
    })
    .with('tags', async () => {
      empty.tags = await searchTags(portId, query, limit);
      empty.totals.tags = empty.tags.length;
      return empty;
    })
    .with('navigation', async () => {
      const { searchNavCatalog } = await import('@/lib/services/search-nav-catalog');
      empty.navigation = searchNavCatalog(query, {
        isSuperAdmin: opts.isSuperAdmin,
        permissions: opts.permissions,
        limit,
      }).map((e) => ({ id: e.href, href: e.href, label: e.label, category: e.category }));
      empty.totals.navigation = empty.navigation.length;
      return empty;
    })
    .with('notes', async () => {
      // Previously this case silently fell through to the default (which
      // returned empty). The exhaustive-match conversion surfaced the
      // missing dispatch: searchNotes() already exists, so wire it up.
      empty.notes = applyAffinity(await searchNotes(portId, query, limit), opts.recentlyTouchedIds);
      empty.totals.notes = empty.notes.length;
      return empty;
    })
    .exhaustive();
}

function emptyResults(): SearchResults {
  return {
    clients: [],
    residentialClients: [],
    yachts: [],
    companies: [],
    interests: [],
    residentialInterests: [],
    berths: [],
    invoices: [],
    expenses: [],
    documents: [],
    files: [],
    reminders: [],
    brochures: [],
    tags: [],
    navigation: [],
    notes: [],
    totals: {
      clients: 0,
      residentialClients: 0,
      yachts: 0,
      companies: 0,
      interests: 0,
      residentialInterests: 0,
      berths: 0,
      invoices: 0,
      expenses: 0,
      documents: 0,
      files: 0,
      reminders: 0,
      brochures: 0,
      tags: 0,
      navigation: 0,
      notes: 0,
    },
  };
}

// ─── Recent search-term history ──────────────────────────────────────────────

const RECENT_SEARCH_TTL = 2592000; // 30 days
const RECENT_SEARCH_MAX = 10;

function recentSearchKey(userId: string, portId: string): string {
  return `recent-search:${userId}:${portId}`;
}

/** Fire-and-forget — saves a search term to the user's recent searches. */
export function saveRecentSearch(userId: string, portId: string, searchTerm: string): void {
  const key = recentSearchKey(userId, portId);
  redis
    .zadd(key, Date.now(), searchTerm)
    .then(() => redis.zremrangebyrank(key, 0, -(RECENT_SEARCH_MAX + 1)))
    .then(() => redis.expire(key, RECENT_SEARCH_TTL))
    .catch(() => {
      // intentionally swallowed
    });
}

/** Returns the user's most recent search terms, newest first. */
export async function getRecentSearches(userId: string, portId: string): Promise<string[]> {
  const key = recentSearchKey(userId, portId);
  const items = await redis.zrevrange(key, 0, RECENT_SEARCH_MAX - 1);
  return items;
}