feat(search): Meilisearch message search — indexer + permission-scoped query

New search module: a message index (Meilisearch, MEILI_URL-gated; no-op when unset),
a SearchProjector that indexes on message.sent, and a SearchService whose permission
fence runs server-side — a query only touches the caller's own scope AND the threads
they currently belong to (resolved live from participant rows, so membership changes
reflect immediately). Every conversation kind (chat/mail/sms) is a message with a TEXT
part, so all are searchable through one index. POST /v1/search + /reindex (backfill).
5 fence tests green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

packages/iios-service/package.json

@@ -29,6 +29,7 @@
     "ioredis": "^5.11.1",
     "jsonwebtoken": "^9.0.3",
     "jwks-rsa": "^4.1.0",
+    "meilisearch": "^0.45.0",
     "nodemailer": "^9.0.3",
     "prisma": "^6.2.1",
     "reflect-metadata": "^0.2.2",

packages/iios-service/src/app.module.ts

@@ -10,6 +10,7 @@ import { OutboxModule } from './outbox/outbox.module';
 import { ThreadsModule } from './threads/threads.module';
 import { MessageModule } from './messaging/message.module';
 import { InboxModule } from './inbox/inbox.module';
+import { SearchModule } from './search/search.module';
 import { TemplateModule } from './templates/template.module';
 import { MailModule } from './mail/mail.module';
 import { MediaModule } from './media/media.module';
@@ -39,6 +40,7 @@ import { DevController } from './dev/dev.controller';
     ThreadsModule,
     MessageModule,
     InboxModule,
+    SearchModule,
     TemplateModule,
     MailModule,
     MediaModule,

packages/iios-service/src/search/meili.adapter.ts

@@ -0,0 +1,101 @@
+import { Logger } from '@nestjs/common';
+import { MeiliSearch } from 'meilisearch';
+import type { MessageSearchPort, SearchDoc, SearchHit } from './message-search.port';
+
+const INDEX = 'iios_messages';
+
+/** A no-op search port — bound when Meilisearch isn't configured, so search degrades to empty. */
+export const noopMessageSearch: MessageSearchPort = {
+  ready: () => false,
+  index: async () => undefined,
+  remove: async () => undefined,
+  search: async () => [],
+};
+
+/**
+ * Meilisearch-backed message search. Configured via MEILI_URL (+ optional MEILI_KEY). The index is
+ * created + configured on first use (searchable text/subject; filterable scopeId/threadId/source;
+ * sortable at). Filtering by scopeId AND threadId is the permission fence — the caller only ever
+ * passes thread ids they belong to.
+ */
+export class MeiliMessageSearch implements MessageSearchPort {
+  private readonly client: MeiliSearch;
+  private readonly logger = new Logger(MeiliMessageSearch.name);
+  private settingsReady?: Promise<void>;
+
+  constructor(host: string, apiKey?: string) {
+    this.client = new MeiliSearch({ host, ...(apiKey ? { apiKey } : {}) });
+  }
+
+  ready(): boolean {
+    return true;
+  }
+
+  /** Idempotently ensure the index + its attribute settings exist (runs once, memoised). */
+  private ensure(): Promise<void> {
+    if (!this.settingsReady) {
+      this.settingsReady = (async () => {
+        await this.client.createIndex(INDEX, { primaryKey: 'id' }).catch(() => undefined);
+        await this.client.index(INDEX).updateSettings({
+          searchableAttributes: ['text', 'subject'],
+          filterableAttributes: ['scopeId', 'threadId', 'source'],
+          sortableAttributes: ['at'],
+        });
+      })().catch((err) => {
+        this.settingsReady = undefined; // let a later call retry
+        throw err;
+      });
+    }
+    return this.settingsReady;
+  }
+
+  async index(docs: SearchDoc[]): Promise<void> {
+    if (docs.length === 0) return;
+    await this.ensure();
+    await this.client.index(INDEX).addDocuments(docs, { primaryKey: 'id' });
+  }
+
+  async remove(ids: string[]): Promise<void> {
+    if (ids.length === 0) return;
+    await this.client.index(INDEX).deleteDocuments(ids);
+  }
+
+  async search(scopeId: string, threadIds: string[], query: string, limit: number): Promise<SearchHit[]> {
+    if (threadIds.length === 0 || !query.trim()) return [];
+    await this.ensure();
+    const quoted = threadIds.map((t) => JSON.stringify(t)).join(', ');
+    const res = await this.client.index(INDEX).search(query, {
+      filter: `scopeId = ${JSON.stringify(scopeId)} AND threadId IN [${quoted}]`,
+      limit,
+      sort: ['at:desc'],
+      attributesToHighlight: ['text'],
+      highlightPreTag: '<em>',
+      highlightPostTag: '</em>',
+      attributesToCrop: ['text'],
+      cropLength: 30,
+    });
+    return res.hits.map((h) => {
+      const doc = h as unknown as SearchDoc & { _formatted?: { text?: string } };
+      return {
+        id: doc.id,
+        threadId: doc.threadId,
+        text: doc.text,
+        subject: doc.subject ?? null,
+        source: doc.source ?? null,
+        at: doc.at,
+        snippet: doc._formatted?.text ?? doc.text,
+      };
+    });
+  }
+}
+
+/** Build the search port from env: MEILI_URL set → Meilisearch; else the no-op. */
+export function messageSearchFromEnv(env: NodeJS.ProcessEnv = process.env): MessageSearchPort {
+  const host = env.MEILI_URL?.trim();
+  if (!host) {
+    new Logger('MessageSearch').log('MEILI_URL unset — message search disabled (no-op)');
+    return noopMessageSearch;
+  }
+  new Logger('MessageSearch').log(`using Meilisearch @ ${host}`);
+  return new MeiliMessageSearch(host, env.MEILI_KEY?.trim() || undefined);
+}

packages/iios-service/src/search/message-search.port.ts

@@ -0,0 +1,39 @@
+/** One indexed message document. `at` is epoch-ms so Meilisearch can sort recency-first. */
+export interface SearchDoc {
+  id: string; // interactionId
+  scopeId: string;
+  threadId: string;
+  text: string;
+  subject: string | null;
+  source: string | null; // opaque app source tag (e.g. crm-messenger / crm-mail) — drives the surface
+  actorId: string | null;
+  at: number;
+}
+
+/** A search hit with a highlighted snippet. */
+export interface SearchHit {
+  id: string;
+  threadId: string;
+  text: string;
+  subject: string | null;
+  source: string | null;
+  at: number;
+  /** The match with `<em>…</em>` around the query terms (from the engine's highlighter). */
+  snippet: string;
+}
+
+/**
+ * The message-search seam. IIOS owns egress-style search: a message index + a permission-scoped
+ * query. The concrete engine (Meilisearch) sits behind this; an unconfigured deployment binds a
+ * no-op so search degrades to empty rather than erroring.
+ */
+export interface MessageSearchPort {
+  /** True only when a real engine is configured (else index/search are inert). */
+  ready(): boolean;
+  index(docs: SearchDoc[]): Promise<void>;
+  remove(ids: string[]): Promise<void>;
+  /** Search WITHIN the given scope and thread set only (the permission fence — enforced here). */
+  search(scopeId: string, threadIds: string[], query: string, limit: number): Promise<SearchHit[]>;
+}
+
+export const MESSAGE_SEARCH_PORT = Symbol('MESSAGE_SEARCH_PORT');

packages/iios-service/src/search/search.controller.ts

@@ -0,0 +1,36 @@
+import { BadRequestException, Body, Controller, Headers, Post } from '@nestjs/common';
+import { SessionVerifier } from '../platform/session.verifier';
+import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver';
+import { SearchService } from './search.service';
+
+interface SearchDto { query?: string; limit?: number }
+
+/** Message search (session-auth). Results are permission-scoped inside the service to the caller's
+ *  own scope + threads. Reindex backfills the caller's scope (idempotent). */
+@Controller('v1/search')
+export class SearchController {
+  constructor(
+    private readonly search: SearchService,
+    private readonly session: SessionVerifier,
+    private readonly actors: ActorResolver,
+  ) {}
+
+  @Post()
+  async run(@Body() body: SearchDto, @Headers('authorization') authorization?: string) {
+    const principal = this.principal(authorization);
+    return this.search.search(principal, { query: body.query ?? '', ...(body.limit != null ? { limit: body.limit } : {}) });
+  }
+
+  @Post('reindex')
+  async reindex(@Headers('authorization') authorization?: string) {
+    const principal = this.principal(authorization);
+    const scope = await this.actors.findScope(principal);
+    return { indexed: await this.search.reindexAll(scope?.id) };
+  }
+
+  private principal(authorization?: string): MessagePrincipal {
+    const token = (authorization ?? '').replace(/^Bearer\s+/i, '');
+    if (!token) throw new BadRequestException('Authorization bearer token is required');
+    return this.session.verify(token);
+  }
+}

packages/iios-service/src/search/search.module.ts

@@ -0,0 +1,24 @@
+import { Module } from '@nestjs/common';
+import { OutboxModule } from '../outbox/outbox.module';
+import { SearchService } from './search.service';
+import { SearchProjector } from './search.projector';
+import { SearchController } from './search.controller';
+import { MESSAGE_SEARCH_PORT } from './message-search.port';
+import { messageSearchFromEnv } from './meili.adapter';
+
+/**
+ * Message search (Meilisearch). The engine binds from MEILI_URL; unset → a no-op port so search
+ * returns empty rather than erroring. The projector indexes on `message.sent`; the service enforces
+ * the permission fence (scope + caller's threads). SessionVerifier + ActorResolver are global.
+ */
+@Module({
+  imports: [OutboxModule],
+  controllers: [SearchController],
+  providers: [
+    SearchService,
+    SearchProjector,
+    { provide: MESSAGE_SEARCH_PORT, useFactory: () => messageSearchFromEnv() },
+  ],
+  exports: [SearchService],
+})
+export class SearchModule {}

packages/iios-service/src/search/search.projector.ts

@@ -0,0 +1,35 @@
+import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
+import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
+import { OutboxBus } from '../outbox/outbox.bus';
+import { DlqService } from '../outbox/dlq.service';
+import { ProjectionCursorService } from '../projection/projection-cursor.service';
+import { SearchService } from './search.service';
+
+/**
+ * Indexes messages into the search engine as they are sent. Reacts to the `message.sent` event on
+ * the outbox bus (same pattern as the inbox projector). Indexing is idempotent (upsert by id), so
+ * no per-event claim is needed — a redelivered event just re-indexes the same doc.
+ */
+@Injectable()
+export class SearchProjector implements OnModuleInit {
+  private readonly consumer = 'search-projector';
+  private readonly logger = new Logger(SearchProjector.name);
+
+  constructor(
+    private readonly search: SearchService,
+    private readonly bus: OutboxBus,
+    private readonly dlq: DlqService,
+    private readonly cursor: ProjectionCursorService,
+  ) {}
+
+  onModuleInit(): void {
+    this.dlq.registerHandler(this.consumer, (e) => this.onMessageSent(e));
+    this.bus.on(IIOS_EVENTS.messageSent, (p) => void this.onMessageSent(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err)));
+  }
+
+  async onMessageSent(event: CloudEvent): Promise<void> {
+    const data = event.data as { interactionId?: string };
+    if (data.interactionId) await this.search.indexInteraction(data.interactionId);
+    await this.cursor.advance(this.consumer, event);
+  }
+}

packages/iios-service/src/search/search.service.spec.ts

@@ -0,0 +1,57 @@
+import { describe, it, expect, vi } from 'vitest';
+import { SearchService } from './search.service';
+import type { MessageSearchPort, SearchHit } from './message-search.port';
+import type { ActorResolver, MessagePrincipal } from '../identity/actor.resolver';
+import type { PrismaService } from '../prisma/prisma.service';
+
+const principal: MessagePrincipal = { userId: 'u1', orgId: 'org', appId: 'app' };
+
+function make(opts: { ready?: boolean; threadIds?: string[]; hits?: SearchHit[]; scope?: { id: string } | null } = {}) {
+  const engine: MessageSearchPort = {
+    ready: () => opts.ready ?? true,
+    index: vi.fn(async () => undefined),
+    remove: vi.fn(async () => undefined),
+    search: vi.fn(async () => opts.hits ?? []),
+  };
+  const actors = {
+    findScope: vi.fn(async () => (opts.scope === undefined ? { id: 'scope_1' } : opts.scope)),
+    resolveActor: vi.fn(async () => ({ id: 'actor_1' })),
+  } as unknown as ActorResolver;
+  const prisma = {
+    iiosThreadParticipant: { findMany: vi.fn(async () => (opts.threadIds ?? ['t1', 't2']).map((threadId) => ({ threadId }))) },
+  } as unknown as PrismaService;
+  return { svc: new SearchService(engine, prisma, actors), engine };
+}
+
+describe('SearchService (permission fence)', () => {
+  it('searches only within the caller scope + their thread ids', async () => {
+    const { svc, engine } = make({ threadIds: ['t1', 't2', 't3'], hits: [{ id: 'i1', threadId: 't2', text: 'hello world', subject: 'General', source: 'crm-messenger', at: 1, snippet: '<em>hello</em>' }] });
+    const res = await svc.search(principal, { query: 'hello' });
+    expect(res[0]!.id).toBe('i1');
+    expect(engine.search).toHaveBeenCalledWith('scope_1', ['t1', 't2', 't3'], 'hello', 20);
+  });
+
+  it('returns empty (no engine hit) for a blank query', async () => {
+    const { svc, engine } = make();
+    expect(await svc.search(principal, { query: '   ' })).toEqual([]);
+    expect(engine.search).not.toHaveBeenCalled();
+  });
+
+  it('returns empty when the caller belongs to no threads', async () => {
+    const { svc, engine } = make({ threadIds: [] });
+    expect(await svc.search(principal, { query: 'hello' })).toEqual([]);
+    expect(engine.search).not.toHaveBeenCalled();
+  });
+
+  it('returns empty when search is not configured (no-op engine)', async () => {
+    const { svc, engine } = make({ ready: false });
+    expect(await svc.search(principal, { query: 'hello' })).toEqual([]);
+    expect(engine.search).not.toHaveBeenCalled();
+  });
+
+  it('caps the limit at 50', async () => {
+    const { svc, engine } = make();
+    await svc.search(principal, { query: 'x', limit: 999 });
+    expect((engine.search as ReturnType<typeof vi.fn>).mock.calls[0]![3]).toBe(50);
+  });
+});

packages/iios-service/src/search/search.service.ts

@@ -0,0 +1,83 @@
+import { Inject, Injectable } from '@nestjs/common';
+import { PrismaService } from '../prisma/prisma.service';
+import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver';
+import { MESSAGE_SEARCH_PORT, type MessageSearchPort, type SearchDoc, type SearchHit } from './message-search.port';
+
+/**
+ * Message search. The permission fence lives HERE: a query only ever runs against the caller's own
+ * scope AND the set of threads the caller currently belongs to (resolved live from participant
+ * rows, so membership changes are reflected immediately). The engine never sees a thread the caller
+ * isn't in. Text is drawn from the message's TEXT part; every conversation kind (chat, mail, sms…)
+ * is a message with parts, so all are searchable through one index.
+ */
+@Injectable()
+export class SearchService {
+  constructor(
+    @Inject(MESSAGE_SEARCH_PORT) private readonly engine: MessageSearchPort,
+    private readonly prisma: PrismaService,
+    private readonly actors: ActorResolver,
+  ) {}
+
+  async search(principal: MessagePrincipal, opts: { query: string; limit?: number }): Promise<SearchHit[]> {
+    const q = (opts.query ?? '').trim();
+    if (!q || !this.engine.ready()) return [];
+    const scope = await this.actors.findScope(principal);
+    if (!scope) return [];
+    const actor = await this.actors.resolveActor(scope.id, principal);
+    const memberships = await this.prisma.iiosThreadParticipant.findMany({ where: { actorId: actor.id }, select: { threadId: true } });
+    const threadIds = memberships.map((m) => m.threadId);
+    if (threadIds.length === 0) return [];
+    return this.engine.search(scope.id, threadIds, q, Math.min(opts.limit ?? 20, 50));
+  }
+
+  /** Build + index the search doc for one interaction (called by the projector on message.sent). */
+  async indexInteraction(interactionId: string): Promise<void> {
+    if (!this.engine.ready()) return;
+    const doc = await this.buildDoc(interactionId);
+    if (doc) await this.engine.index([doc]);
+  }
+
+  /** Remove an interaction from the index (retention / redaction hook). */
+  async removeInteraction(interactionId: string): Promise<void> {
+    if (this.engine.ready()) await this.engine.remove([interactionId]);
+  }
+
+  /** Backfill: (re)index every text message, optionally for one scope. Returns the count indexed. */
+  async reindexAll(scopeId?: string, batch = 500): Promise<number> {
+    if (!this.engine.ready()) return 0;
+    const interactions = await this.prisma.iiosInteraction.findMany({
+      where: { ...(scopeId ? { scopeId } : {}), threadId: { not: null }, parts: { some: { kind: 'TEXT' } } },
+      select: { id: true },
+    });
+    let indexed = 0;
+    for (let i = 0; i < interactions.length; i += batch) {
+      const docs = (await Promise.all(interactions.slice(i, i + batch).map((x) => this.buildDoc(x.id)))).filter((d): d is SearchDoc => d !== null);
+      if (docs.length > 0) {
+        await this.engine.index(docs);
+        indexed += docs.length;
+      }
+    }
+    return indexed;
+  }
+
+  private async buildDoc(interactionId: string): Promise<SearchDoc | null> {
+    const i = await this.prisma.iiosInteraction.findUnique({
+      where: { id: interactionId },
+      include: { parts: { where: { kind: 'TEXT' }, take: 1 }, thread: true },
+    });
+    if (!i || !i.threadId || !i.thread) return null;
+    const text = i.parts[0]?.bodyText?.trim();
+    if (!text) return null; // nothing searchable
+    const meta = (i.thread.metadata as { source?: string } | null) ?? {};
+    return {
+      id: i.id,
+      scopeId: i.scopeId,
+      threadId: i.threadId,
+      text,
+      subject: i.thread.subject ?? null,
+      source: meta.source ?? null,
+      actorId: i.actorId ?? null,
+      at: i.occurredAt.getTime(),
+    };
+  }
+}

pnpm-lock.yaml

@@ -415,6 +415,9 @@ importers:
       jwks-rsa:
         specifier: ^4.1.0
         version: 4.1.0
+      meilisearch:
+        specifier: ^0.45.0
+        version: 0.45.0
       nodemailer:
         specifier: ^9.0.3
         version: 9.0.3
@@ -2715,6 +2718,9 @@ packages:
     resolution: {integrity: sha512-aisnrDP4GNe06UcKFnV5bfMNPBUw4jsLGaWwWfnH3v02GnBuXX2MCVn5RbrWo0j3pczUilYblq7fQ7Nw2t5XKw==}
     engines: {node: '>= 0.8'}
 
+  meilisearch@0.45.0:
+    resolution: {integrity: sha512-+zCzEqE+CumY4icB0Vox180adZqaNtnr60hJWGiEdmol5eWmksfY8rYsTcz87styXC2ZOg+2yF56gdH6oyIBTA==}
+
   memfs@3.5.3:
     resolution: {integrity: sha512-UERzLsxzllchadvbPs5aolHh65ISpKpM+ccLbOJ8/vvpBKmAWf+la7dXFy7Mr0ySHbdHrFv5kGFCUHHe6GFEmw==}
     engines: {node: '>= 4.0.0'}
@@ -5902,6 +5908,8 @@ snapshots:
 
   media-typer@1.1.0: {}
 
+  meilisearch@0.45.0: {}
+
   memfs@3.5.3:
     dependencies:
       fs-monkey: 1.1.0