import type { Pool } from 'pg';
import { pgGet, pgAll } from '@/shared/db';
import { notFound } from '@/shared/http/errors';
import { logger } from '@/shared/logger';
import { buildCacheKey, getSearchCache, setSearchCache, invalidateSearchCache } from '@/entities/search-cache/repo';

import type {
  Message,
  MessageDraft,
  MessageFilter,
  MessageSearchFilter,
  MessageSearchHit,
  SearchMode,
  SearchResult,
} from './types';

interface Row {
  id: string;
  device_id: string;
  conversation_id: string;
  external_id: string;
  provider: Message['provider'];
  from_handle: string;
  body: string;
  is_from_me: boolean;
  has_attachments: boolean;
  attachment_count: number;
  attachment_mime_types: string[] | null;
  sent_at: string;
  delivered_at: string | null;
  read_at: string | null;
  synced_at: string;
  attributed_body: Buffer | null;
  body_decoded: string | null;
  associated_message_type: number | null;
  associated_message_guid: string | null;
  expressive_send_style_id: string | null;
  balloon_bundle_id: string | null;
  is_audio_message: boolean;
  reply_to_guid: string | null;
  thread_originator_guid: string | null;
  group_title: string | null;
  service: string | null;
  sender_identifier: string | null;
  sender_display_name: string | null;
  sender_phone_number: string | null;
  sender_email: string | null;
}

const hydrate = (r: Row): Message => ({
  id: r.id,
  deviceId: r.device_id,
  conversationId: r.conversation_id,
  externalId: r.external_id,
  provider: r.provider,
  fromHandle: r.from_handle,
  body: r.body,
  isFromMe: r.is_from_me,
  hasAttachments: r.has_attachments,
  attachmentCount: r.attachment_count,
  attachmentMimeTypes: r.attachment_mime_types ?? [],
  sentAt: r.sent_at,
  deliveredAt: r.delivered_at,
  readAt: r.read_at,
  syncedAt: r.synced_at,
  attributedBody: r.attributed_body ? r.attributed_body.toString('base64') : null,
  bodyDecoded: r.body_decoded,
  associatedMessageType: r.associated_message_type,
  associatedMessageGuid: r.associated_message_guid,
  expressiveSendStyleId: r.expressive_send_style_id,
  balloonBundleId: r.balloon_bundle_id,
  isAudioMessage: r.is_audio_message,
  replyToGuid: r.reply_to_guid,
  threadOriginatorGuid: r.thread_originator_guid,
  groupTitle: r.group_title,
  service: r.service,
  senderIdentifier: r.sender_identifier,
  senderDisplayName: r.sender_display_name,
  senderPhoneNumber: r.sender_phone_number,
  senderEmail: r.sender_email,
});

export async function upsertMessage(pool: Pool, draft: MessageDraft): Promise<Message> {
  const attributedBodyBuf = draft.attributedBody
    ? Buffer.from(draft.attributedBody, 'base64')
    : null;
  await pgAll(
    pool,
    `INSERT INTO macsync.messages
     (id, device_id, conversation_id, external_id, provider, from_handle, body, is_from_me,
      has_attachments, attachment_count, sent_at, delivered_at, read_at,
      attributed_body, body_decoded, associated_message_type, associated_message_guid,
      expressive_send_style_id, balloon_bundle_id, is_audio_message,
      reply_to_guid, thread_originator_guid, group_title, service,
      sender_identifier, sender_display_name, sender_phone_number, sender_email)
     VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13,
             $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28)
     ON CONFLICT(device_id, external_id, provider) DO UPDATE SET
       body = excluded.body,
       delivered_at = COALESCE(excluded.delivered_at, messages.delivered_at),
       read_at = COALESCE(excluded.read_at, messages.read_at),
       attributed_body = COALESCE(excluded.attributed_body, messages.attributed_body),
       body_decoded = COALESCE(excluded.body_decoded, messages.body_decoded),
       associated_message_type = COALESCE(excluded.associated_message_type, messages.associated_message_type),
       associated_message_guid = COALESCE(excluded.associated_message_guid, messages.associated_message_guid),
       expressive_send_style_id = COALESCE(excluded.expressive_send_style_id, messages.expressive_send_style_id),
       balloon_bundle_id = COALESCE(excluded.balloon_bundle_id, messages.balloon_bundle_id),
       is_audio_message = excluded.is_audio_message,
       reply_to_guid = COALESCE(excluded.reply_to_guid, messages.reply_to_guid),
       thread_originator_guid = COALESCE(excluded.thread_originator_guid, messages.thread_originator_guid),
       group_title = COALESCE(excluded.group_title, messages.group_title),
       service = COALESCE(excluded.service, messages.service),
       sender_identifier = COALESCE(excluded.sender_identifier, messages.sender_identifier),
       sender_display_name = COALESCE(excluded.sender_display_name, messages.sender_display_name),
       sender_phone_number = COALESCE(excluded.sender_phone_number, messages.sender_phone_number),
       sender_email = COALESCE(excluded.sender_email, messages.sender_email),
       synced_at = now()`,
    [
      draft.id,
      draft.deviceId,
      draft.conversationId,
      draft.externalId,
      draft.provider,
      draft.fromHandle,
      draft.body,
      draft.isFromMe,
      draft.hasAttachments,
      draft.attachmentCount ?? 0,
      draft.sentAt,
      draft.deliveredAt ?? null,
      draft.readAt ?? null,
      attributedBodyBuf,
      draft.bodyDecoded ?? null,
      draft.associatedMessageType ?? null,
      draft.associatedMessageGuid ?? null,
      draft.expressiveSendStyleId ?? null,
      draft.balloonBundleId ?? null,
      draft.isAudioMessage ?? false,
      draft.replyToGuid ?? null,
      draft.threadOriginatorGuid ?? null,
      draft.groupTitle ?? null,
      draft.service ?? null,
      draft.senderIdentifier ?? null,
      draft.senderDisplayName ?? null,
      draft.senderPhoneNumber ?? null,
      draft.senderEmail ?? null,
    ],
  );
  return getMessage(pool, draft.id);
}

/**
 * Upserts messages in bulk. Returns a map of externalId → actual DB id.
 * On conflict the existing row's id is preserved (not the draft id), so
 * callers that need to FK-reference messages (e.g. attachments) must use
 * the returned map rather than draft.id.
 */
export async function bulkUpsertMessages(
  pool: Pool,
  drafts: MessageDraft[],
): Promise<{ count: number; idByExternalId: Map<string, string> }> {
  if (drafts.length === 0) return { count: 0, idByExternalId: new Map() };

  // 28 cols per row; chunk at 2000 to stay under pg's 65535-param limit
  const COLS = 28;
  const CHUNK = 2000;
  let total = 0;
  const idByExternalId = new Map<string, string>();
  for (let i = 0; i < drafts.length; i += CHUNK) {
    const chunk = drafts.slice(i, i + CHUNK);
    const placeholders = chunk
      .map((_, ri) => {
        const base = ri * COLS;
        const nums = Array.from({ length: COLS }, (_, ci) => `$${base + ci + 1}`).join(', ');
        return `(${nums})`;
      })
      .join(',\n');

    const params: unknown[] = [];
    for (const d of chunk) {
      params.push(
        d.id,
        d.deviceId,
        d.conversationId,
        d.externalId,
        d.provider,
        d.fromHandle,
        d.body,
        d.isFromMe,
        d.hasAttachments,
        d.attachmentCount ?? 0,
        d.sentAt,
        d.deliveredAt ?? null,
        d.readAt ?? null,
        d.attributedBody ? Buffer.from(d.attributedBody, 'base64') : null,
        d.bodyDecoded ?? null,
        d.associatedMessageType ?? null,
        d.associatedMessageGuid ?? null,
        d.expressiveSendStyleId ?? null,
        d.balloonBundleId ?? null,
        d.isAudioMessage ?? false,
        d.replyToGuid ?? null,
        d.threadOriginatorGuid ?? null,
        d.groupTitle ?? null,
        d.service ?? null,
        d.senderIdentifier ?? null,
        d.senderDisplayName ?? null,
        d.senderPhoneNumber ?? null,
        d.senderEmail ?? null,
      );
    }

    const rows = await pgAll<{ id: string; external_id: string }>(
      pool,
      `INSERT INTO macsync.messages
       (id, device_id, conversation_id, external_id, provider, from_handle, body, is_from_me,
        has_attachments, attachment_count, sent_at, delivered_at, read_at,
        attributed_body, body_decoded, associated_message_type, associated_message_guid,
        expressive_send_style_id, balloon_bundle_id, is_audio_message,
        reply_to_guid, thread_originator_guid, group_title, service,
        sender_identifier, sender_display_name, sender_phone_number, sender_email)
       VALUES ${placeholders}
       ON CONFLICT(device_id, external_id, provider) DO UPDATE SET
         body = excluded.body,
         delivered_at = COALESCE(excluded.delivered_at, messages.delivered_at),
         read_at = COALESCE(excluded.read_at, messages.read_at),
         attributed_body = COALESCE(excluded.attributed_body, messages.attributed_body),
         body_decoded = COALESCE(excluded.body_decoded, messages.body_decoded),
         associated_message_type = COALESCE(excluded.associated_message_type, messages.associated_message_type),
         associated_message_guid = COALESCE(excluded.associated_message_guid, messages.associated_message_guid),
         expressive_send_style_id = COALESCE(excluded.expressive_send_style_id, messages.expressive_send_style_id),
         balloon_bundle_id = COALESCE(excluded.balloon_bundle_id, messages.balloon_bundle_id),
         is_audio_message = excluded.is_audio_message,
         reply_to_guid = COALESCE(excluded.reply_to_guid, messages.reply_to_guid),
         thread_originator_guid = COALESCE(excluded.thread_originator_guid, messages.thread_originator_guid),
         group_title = COALESCE(excluded.group_title, messages.group_title),
         service = COALESCE(excluded.service, messages.service),
         sender_identifier = COALESCE(excluded.sender_identifier, messages.sender_identifier),
         sender_display_name = COALESCE(excluded.sender_display_name, messages.sender_display_name),
         sender_phone_number = COALESCE(excluded.sender_phone_number, messages.sender_phone_number),
         sender_email = COALESCE(excluded.sender_email, messages.sender_email),
         synced_at = now()
       RETURNING id, external_id`,
      params,
    );
    for (const row of rows) idByExternalId.set(row.external_id, row.id);
    total += chunk.length;
  }

  // Bump epoch so all pre-sync cached searches become stale
  invalidateSearchCache(pool).catch((err) => {
    logger.warn('search cache invalidation failed', { err: String(err) });
  });

  return { count: total, idByExternalId };
}

export async function getMessage(pool: Pool, id: string): Promise<Message> {
  const row = await pgGet<Row>(pool, 'SELECT * FROM macsync.messages WHERE id = $1', [id]);
  if (!row) throw notFound('message_not_found', `message ${id} not found`);
  return hydrate(row);
}

export async function getLastSyncedAt(
  pool: Pool,
  { deviceId, provider }: { deviceId: string; provider?: string },
): Promise<string | null> {
  const conditions = ['device_id = $1'];
  const params: unknown[] = [deviceId];
  if (provider) {
    conditions.push(`provider = $${params.length + 1}`);
    params.push(provider);
  }
  const where = `WHERE ${conditions.join(' AND ')}`;
  const row = await pgGet<{ last: string | null }>(
    pool,
    `SELECT MAX(synced_at) as last FROM macsync.messages ${where}`,
    params,
  );
  return row?.last ?? null;
}

export async function listMessages(pool: Pool, filter?: MessageFilter): Promise<readonly Message[]> {
  const conditions: string[] = [];
  const params: unknown[] = [];

  if (filter?.conversationId) {
    conditions.push(`conversation_id = $${params.length + 1}`);
    params.push(filter.conversationId);
  }
  if (filter?.deviceId) {
    conditions.push(`device_id = $${params.length + 1}`);
    params.push(filter.deviceId);
  }
  if (filter?.provider) {
    conditions.push(`provider = $${params.length + 1}`);
    params.push(filter.provider);
  }
  if (filter?.since) {
    conditions.push(`sent_at > $${params.length + 1}`);
    params.push(filter.since);
  }

  const where = conditions.length ? `WHERE ${conditions.join(' AND ')}` : '';
  const limit = filter?.limit ?? 100;
  const rows = await pgAll<Row>(
    pool,
    `SELECT m.*,
       (SELECT array_agg(a.mime_type) FILTER (WHERE a.mime_type IS NOT NULL)
        FROM macsync.attachments a WHERE a.message_id = m.id) AS attachment_mime_types
     FROM macsync.messages m ${where} ORDER BY m.sent_at DESC LIMIT $${params.length + 1}`,
    [...params, limit],
  );
  return rows.map(hydrate);
}

const SEARCH_LIMIT_DEFAULT = 50;
/** RRF constant — standard value from Cormack et al. */
const RRF_K = 60;

/**
 * Max pgvector cosine distance for a semantic hit to be considered relevant.
 * Distance ranges 0 (identical) … 2 (opposite). Below ~0.55 typically means
 * "actually similar" for sentence-embedding models; above that is noise that
 * pollutes total counts and pushes irrelevant rows to the top via RRF.
 *
 * Set via SEARCH_SEMANTIC_MAX_DISTANCE; default tuned to suppress the
 * "every message is a match" failure mode of unbounded ANN ordering.
 */
const SEMANTIC_MAX_DISTANCE: number = (() => {
  const raw = process.env['SEARCH_SEMANTIC_MAX_DISTANCE'];
  if (!raw) return 0.55;
  const parsed = Number.parseFloat(raw);
  if (!Number.isFinite(parsed) || parsed <= 0 || parsed > 2) {
    logger.warn('SEARCH_SEMANTIC_MAX_DISTANCE invalid, falling back to 0.55', { raw });
    return 0.55;
  }
  return parsed;
})();

/**
 * Maximum candidate pool to materialize when warming the search cache on a
 * cache-miss page-1 request. Previously this used Number.MAX_SAFE_INTEGER,
 * which caused semanticSearchFull to ANN-rank every embedded message and
 * report `total = corpus size` on no-match queries.
 */
const CACHE_FILL_LIMIT = 500;

type SearchRow = Row & { rank: number; snippet: string };

async function lexicalSearch(
  pool: Pool,
  filter: MessageSearchFilter,
  limit: number,
): Promise<SearchResult> {
  // Base conditions for COUNT (no since/offset)
  const baseConditions: string[] = ['m.body_tsv @@ q'];
  const baseParams: unknown[] = [filter.query];

  if (filter.conversationId) {
    baseConditions.push(`m.conversation_id = $${baseParams.length + 1}`);
    baseParams.push(filter.conversationId);
  }
  if (filter.deviceId) {
    baseConditions.push(`m.device_id = $${baseParams.length + 1}`);
    baseParams.push(filter.deviceId);
  }
  if (filter.provider) {
    baseConditions.push(`m.provider = $${baseParams.length + 1}`);
    baseParams.push(filter.provider);
  }

  const baseWhere = `WHERE ${baseConditions.join(' AND ')}`;

  // Result conditions — optionally add synced_at > since
  const resultConditions = [...baseConditions];
  const resultParams = [...baseParams];
  if (filter.since) {
    resultConditions.push(`m.synced_at > $${resultParams.length + 1}`);
    resultParams.push(filter.since);
  }
  const resultWhere = `WHERE ${resultConditions.join(' AND ')}`;
  const offset = filter.offset ?? 0;

  // Single-token queries get a prefix wildcard (`shannon:*`) but only if the
  // token is long enough to be meaningful — `:*` on 1–2 char tokens
  // matches huge swaths of the lexicon and pollutes results. Multi-word
  // queries go through websearch_to_tsquery for natural-language parsing.
  const tsQueryCte = `WITH q AS (
    SELECT CASE
      WHEN array_length(string_to_array(trim($1), ' '), 1) = 1
        AND char_length(trim($1)) >= 3
        THEN to_tsquery('simple', trim($1) || ':*')
      WHEN array_length(string_to_array(trim($1), ' '), 1) = 1
        THEN plainto_tsquery('simple', trim($1))
      ELSE websearch_to_tsquery('english', $1)
    END AS q
  )`;

  const [countRow, rows] = await Promise.all([
    pgGet<{ n: string }>(
      pool,
      `${tsQueryCte}
       SELECT COUNT(*) AS n FROM macsync.messages m CROSS JOIN q ${baseWhere}`,
      baseParams,
    ),
    pgAll<SearchRow>(
      pool,
      `${tsQueryCte}
       SELECT m.*,
         (SELECT array_agg(a.mime_type) FILTER (WHERE a.mime_type IS NOT NULL)
          FROM macsync.attachments a WHERE a.message_id = m.id) AS attachment_mime_types,
         ts_rank(m.body_tsv, q.q) AS rank,
         ts_headline('english', coalesce(m.body, ''), q.q,
           'StartSel=«,StopSel=»,MaxFragments=2,MaxWords=12,MinWords=3'
         ) AS snippet
       FROM macsync.messages m CROSS JOIN q
       ${resultWhere}
       ORDER BY rank DESC, m.sent_at DESC
       LIMIT $${resultParams.length + 1} OFFSET $${resultParams.length + 2}`,
      [...resultParams, limit, offset],
    ),
  ]);

  return {
    hits: rows.map((r) => ({
      message: hydrate(r),
      rank: Number(r.rank),
      snippet: r.snippet,
      mode: 'lexical' as SearchMode,
    })),
    total: Number(countRow?.n ?? 0),
  };
}

async function semanticSearchFull(
  pool: Pool,
  filter: MessageSearchFilter,
  limit: number,
): Promise<SearchResult> {
  if (!filter.queryEmbedding || filter.queryEmbedding.length === 0) {
    throw new Error('semanticSearchFull requires queryEmbedding in filter');
  }

  const vec = `[${filter.queryEmbedding.join(',')}]`;

  // Shared conv/device/provider conditions used in both count and results.
  // Both queries also filter by SEMANTIC_MAX_DISTANCE so "total" reflects
  // actually-relevant neighbors, not the entire embedded corpus.
  const sharedFilters: Array<{ col: string; val: unknown }> = [];
  if (filter.conversationId) sharedFilters.push({ col: 'm.conversation_id', val: filter.conversationId });
  if (filter.deviceId) sharedFilters.push({ col: 'm.device_id', val: filter.deviceId });
  if (filter.provider) sharedFilters.push({ col: 'm.provider', val: filter.provider });

  // COUNT params: [vec, threshold, ...sharedFilters]
  const countParams: unknown[] = [vec, SEMANTIC_MAX_DISTANCE, ...sharedFilters.map((f) => f.val)];
  const countSharedWhere = sharedFilters.length
    ? `AND ${sharedFilters.map((f, i) => `${f.col} = $${i + 3}`).join(' AND ')}`
    : '';

  // Results params: [vec, threshold, ...sharedFilters, (since?), limit, offset]
  const resultParams: unknown[] = [vec, SEMANTIC_MAX_DISTANCE];
  const resultConditions: string[] = [];
  for (const f of sharedFilters) {
    resultConditions.push(`${f.col} = $${resultParams.length + 1}`);
    resultParams.push(f.val);
  }
  if (filter.since) {
    resultConditions.push(`m.synced_at > $${resultParams.length + 1}`);
    resultParams.push(filter.since);
  }
  const resultExtra = resultConditions.length ? `AND ${resultConditions.join(' AND ')}` : '';
  const offset = filter.offset ?? 0;

  const [countRow, rows] = await Promise.all([
    pgGet<{ n: string }>(
      pool,
      `SELECT COUNT(*) AS n
       FROM macsync.message_embeddings e
       JOIN macsync.messages m ON m.id = e.message_id
       WHERE (e.embedding <=> $1::vector) <= $2 ${countSharedWhere}`,
      countParams,
    ),
    pgAll<Row & { distance: string }>(
      pool,
      `SELECT m.*,
         (SELECT array_agg(a.mime_type) FILTER (WHERE a.mime_type IS NOT NULL)
          FROM macsync.attachments a WHERE a.message_id = m.id) AS attachment_mime_types,
         (e.embedding <=> $1::vector) AS distance
       FROM macsync.message_embeddings e
       JOIN macsync.messages m ON m.id = e.message_id
       WHERE (e.embedding <=> $1::vector) <= $2 ${resultExtra}
       ORDER BY e.embedding <=> $1::vector
       LIMIT $${resultParams.length + 1} OFFSET $${resultParams.length + 2}`,
      [...resultParams, limit, offset],
    ),
  ]);

  return {
    hits: rows.map((r) => ({
      message: hydrate(r),
      rank: 1 - Number(r.distance) / 2,
      snippet: r.body.slice(0, 120),
      mode: 'semantic' as SearchMode,
    })),
    total: Number(countRow?.n ?? 0),
  };
}

/**
 * Hybrid search via Reciprocal Rank Fusion.
 *
 * Runs lexical and semantic searches in parallel, fuses via RRF, applies
 * offset in-memory. Never uses filter.since — hybrid always full-refreshes
 * to preserve RRF rank integrity.
 */
async function hybridSearch(
  pool: Pool,
  filter: MessageSearchFilter,
  limit: number,
): Promise<SearchResult> {
  // Strip since/offset from arm fetches — hybrid re-ranks the full candidate pool
  const armFilter: MessageSearchFilter = { ...filter, since: undefined, offset: undefined };
  // Multiply small limits to widen the candidate pool; don't overflow for cache-fill (MAX_SAFE_INTEGER)
  const armLimit = limit > Number.MAX_SAFE_INTEGER / 4 ? limit : limit * 4;

  const [lexResult, semResult] = await Promise.all([
    lexicalSearch(pool, armFilter, armLimit),
    filter.queryEmbedding
      ? semanticSearchFull(pool, armFilter, armLimit)
      : Promise.resolve<SearchResult>({ hits: [], total: 0 }),
  ]);

  const lexResults = lexResult.hits;
  const semResults = semResult.hits;

  // Build position maps: messageId → 1-based rank
  const lexRank = new Map<string, number>(lexResults.map((h, i) => [h.message.id, i + 1]));
  const semRank = new Map<string, number>(semResults.map((h, i) => [h.message.id, i + 1]));

  // Union of all candidate message IDs
  const allIds = new Set<string>([
    ...lexResults.map((h) => h.message.id),
    ...semResults.map((h) => h.message.id),
  ]);

  // Index hits by id (semantic hits carry a better snippet for shared ids)
  const byId = new Map<string, MessageSearchHit>();
  for (const h of [...lexResults, ...semResults]) byId.set(h.message.id, h);

  // Compute RRF scores
  const scored = Array.from(allIds).map((id) => {
    const lRank = lexRank.get(id);
    const sRank = semRank.get(id);
    const rrfScore =
      (lRank !== undefined ? 1 / (RRF_K + lRank) : 0) +
      (sRank !== undefined ? 1 / (RRF_K + sRank) : 0);
    return { id, rrfScore };
  });
  scored.sort((a, b) => b.rrfScore - a.rrfScore);

  const offset = filter.offset ?? 0;
  const page = scored.slice(offset, offset + limit);

  return {
    hits: page.map(({ id, rrfScore }) => {
      const hit = byId.get(id)!;
      return { message: hit.message, rank: rrfScore, snippet: hit.snippet, mode: 'hybrid' as SearchMode };
    }),
    total: allIds.size,
  };
}

async function runSearch(
  pool: Pool,
  filter: MessageSearchFilter,
  limit: number,
  mode: SearchMode,
  hasEmbedding: boolean,
): Promise<SearchResult> {
  if (mode === 'lexical' || !hasEmbedding) return lexicalSearch(pool, filter, limit);
  if (mode === 'semantic') return semanticSearchFull(pool, filter, limit);
  return hybridSearch(pool, filter, limit);
}

export async function listMessagesByHandle(
  pool: Pool,
  filter: { deviceId?: string; handle: string; since?: string; limit: number },
): Promise<readonly Message[]> {
  const conditions: string[] = ['from_handle = $1'];
  const params: unknown[] = [filter.handle];

  if (filter.deviceId) {
    conditions.push(`device_id = $${params.length + 1}`);
    params.push(filter.deviceId);
  }
  if (filter.since) {
    conditions.push(`sent_at > $${params.length + 1}`);
    params.push(filter.since);
  }

  const where = `WHERE ${conditions.join(' AND ')}`;
  const rows = await pgAll<Row>(
    pool,
    `SELECT * FROM macsync.messages ${where} ORDER BY sent_at DESC LIMIT $${params.length + 1}`,
    [...params, filter.limit],
  );
  return rows.map(hydrate);
}

export async function searchMessages(
  pool: Pool,
  filter: MessageSearchFilter,
): Promise<SearchResult> {
  const limit = filter.limit ?? SEARCH_LIMIT_DEFAULT;
  const mode = filter.mode ?? 'hybrid';
  const hasEmbedding = (filter.queryEmbedding?.length ?? 0) > 0;
  const offset = filter.offset ?? 0;

  // Incremental fetches (since is set) always bypass the cache to get fresh results.
  if (!filter.since) {
    const key = buildCacheKey(filter);
    const cached = await getSearchCache(pool, key);

    if (cached !== null && offset < cached.hits.length) {
      // Serve from cache: slice to the requested page
      return { hits: cached.hits.slice(offset, offset + limit), total: cached.total };
    }

    if (offset === 0) {
      // Cache miss — materialize a bounded candidate pool so the cache row
      // doesn't store the entire corpus and `total` stays meaningful.
      const fillLimit = Math.max(limit, CACHE_FILL_LIMIT);
      const result = await runSearch(pool, { ...filter, limit: fillLimit, offset: 0 }, fillLimit, mode, hasEmbedding);
      setSearchCache(pool, key, result).catch((err) => {
        logger.warn('search cache write failed', { err: String(err) });
      });
      return { hits: result.hits.slice(0, limit), total: result.total };
    }
  }

  // Incremental path (since set) or pagination beyond cached window
  return runSearch(pool, filter, limit, mode, hasEmbedding);
}

const INBOUND_SINCE_DEFAULT_LIMIT = 50;
const INBOUND_SINCE_MAX_LIMIT = 200;

export async function listInboundSince(
  pool: Pool,
  filter: { sinceIso: string; limit?: number },
): Promise<readonly Message[]> {
  const limit = Math.min(
    filter.limit ?? INBOUND_SINCE_DEFAULT_LIMIT,
    INBOUND_SINCE_MAX_LIMIT,
  );
  const rows = await pgAll<Row>(
    pool,
    `SELECT * FROM macsync.messages
     WHERE is_from_me = false AND sent_at > $1
     ORDER BY sent_at ASC
     LIMIT $2`,
    [filter.sinceIso, limit],
  );
  return rows.map(hydrate);
}
