fix: Elasticsearch index consistency and tenant-scoped search (#245)

.github/workflows/backend-ci-cd.yml

@@ -95,6 +95,30 @@ jobs:
           --health-timeout 5s
           --health-retries 5
 
+      redis:
+        image: redis:7-alpine
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+      elasticsearch:
+        image: docker.elastic.co/elasticsearch/elasticsearch:8.15.0
+        env:
+          discovery.type: single-node
+          xpack.security.enabled: 'false'
+          ES_JAVA_OPTS: -Xms256m -Xmx256m
+        ports:
+          - 9200:9200
+        options: >-
+          --health-cmd "curl -sf http://localhost:9200 >/dev/null"
+          --health-interval 10s
+          --health-timeout 10s
+          --health-retries 20
+
     env:
       NODE_ENV: test
       DB_TYPE: postgres
@@ -112,6 +136,13 @@ jobs:
       RATE_LIMIT_AUTH_MAX: 5
       RATE_LIMIT_STRICT_TTL: 60000
       RATE_LIMIT_STRICT_MAX: 10
+      REDIS_HOST: localhost
+      REDIS_PORT: 6379
+      ELASTICSEARCH_NODE: http://localhost:9200
+      SEARCH_INDEX_PREFIX: vespera_test
+      SEARCH_OUTBOX_POLL_INTERVAL_MS: 1000
+      SEARCH_OUTBOX_MAX_ATTEMPTS: 3
+      SEARCH_RECONCILE_CRON: '0 */15 * * * *'
       CHIOMA_CONTRACT_ID: CA3D5KRYM6CB7OWQ6TWYRR3Z4T7GNZLKERYNZGGA5SOAOPIFY6YQGAXE
       ESCROW_CONTRACT_ID: CA3D5KRYM6CB7OWQ6TWYRR3Z4T7GNZLKERYNZGGA5SOAOPIFY6YQGAXE
       DISPUTE_CONTRACT_ID: CA3D5KRYM6CB7OWQ6TWYRR3Z4T7GNZLKERYNZGGA5SOAOPIFY6YQGAXE

backend/.env.example

@@ -136,3 +136,13 @@ REDIS_TLS=false
 # Bull Job Queue Configuration
 # Queue retry and backoff values are currently hardcoded in
 # QueueManagementService. Override per job via QueueJobOptions when needed.
+
+# Elasticsearch / Search Index
+ELASTICSEARCH_NODE=http://localhost:9200
+# Alias kept for backwards compatibility with ElasticsearchService fallbacks
+ELASTICSEARCH_URL=http://localhost:9200
+SEARCH_INDEX_PREFIX=vespera
+SEARCH_OUTBOX_POLL_INTERVAL_MS=5000
+SEARCH_OUTBOX_MAX_ATTEMPTS=5
+# Bull repeatable cron for SearchReconcileJob (every 15 minutes)
+SEARCH_RECONCILE_CRON=0 */15 * * * *

backend/docker-compose.production.yml

@@ -20,6 +20,11 @@ services:
       - STELLAR_NETWORK=${STELLAR_NETWORK:-mainnet}
       - SOROBAN_RPC_URL=${SOROBAN_RPC_URL}
       - SENTRY_DSN=${SENTRY_DSN}
+      - ELASTICSEARCH_NODE=${ELASTICSEARCH_NODE:-http://elasticsearch:9200}
+      - SEARCH_INDEX_PREFIX=${SEARCH_INDEX_PREFIX:-vespera}
+      - SEARCH_OUTBOX_POLL_INTERVAL_MS=${SEARCH_OUTBOX_POLL_INTERVAL_MS:-5000}
+      - SEARCH_OUTBOX_MAX_ATTEMPTS=${SEARCH_OUTBOX_MAX_ATTEMPTS:-5}
+      - SEARCH_RECONCILE_CRON=${SEARCH_RECONCILE_CRON:-0 */15 * * * *}
     networks:
       - chioma-network
     healthcheck:
@@ -28,6 +33,13 @@ services:
       timeout: 10s
       retries: 3
       start_period: 40s
+    depends_on:
+      postgres:
+        condition: service_healthy
+      redis:
+        condition: service_healthy
+      elasticsearch:
+        condition: service_healthy
     logging:
       driver: "json-file"
       options:
@@ -67,10 +79,30 @@ services:
       timeout: 3s
       retries: 3
 
+  elasticsearch:
+    image: docker.elastic.co/elasticsearch/elasticsearch:8.15.0
+    container_name: chioma-elasticsearch-${ENVIRONMENT:-production}
+    restart: unless-stopped
+    environment:
+      - discovery.type=single-node
+      - xpack.security.enabled=false
+      - ES_JAVA_OPTS=-Xms512m -Xmx512m
+      - cluster.name=vespera-search
+    volumes:
+      - elasticsearch-data:/usr/share/elasticsearch/data
+    networks:
+      - chioma-network
+    healthcheck:
+      test: ["CMD-SHELL", "curl -sf http://localhost:9200 >/dev/null"]
+      interval: 10s
+      timeout: 10s
+      retries: 20
+
 networks:
   chioma-network:
     driver: bridge
 
 volumes:
   postgres-data:
   redis-data:
+  elasticsearch-data:

backend/docker-compose.yml

@@ -13,13 +13,43 @@ services:
       - '5433:5432'
     volumes:
       - chioma_postgres_data:/var/lib/postgresql/data
+    healthcheck:
+      test: ['CMD-SHELL', 'pg_isready -U postgres -d chioma_db']
+      interval: 5s
+      timeout: 5s
+      retries: 10
 
   redis:
     image: redis:7-alpine
     container_name: chioma_redis
     restart: always
     ports:
       - '6379:6379'
+    healthcheck:
+      test: ['CMD', 'redis-cli', 'ping']
+      interval: 5s
+      timeout: 3s
+      retries: 10
+
+  elasticsearch:
+    image: docker.elastic.co/elasticsearch/elasticsearch:8.15.0
+    container_name: chioma_elasticsearch
+    restart: always
+    environment:
+      - discovery.type=single-node
+      - xpack.security.enabled=false
+      - ES_JAVA_OPTS=-Xms512m -Xmx512m
+      - cluster.name=vespera-search
+    ports:
+      - '9200:9200'
+    volumes:
+      - chioma_es_data:/usr/share/elasticsearch/data
+    healthcheck:
+      test: ['CMD-SHELL', 'curl -sf http://localhost:9200 >/dev/null']
+      interval: 10s
+      timeout: 10s
+      retries: 20
 
 volumes:
   chioma_postgres_data:
+  chioma_es_data:

backend/docs/search/SEARCH_INDEX_CONSISTENCY.md

@@ -0,0 +1,65 @@
+# Search Index Consistency — PR Attachments (#245)
+
+## 1. Migration SQL (`AddSearchOutbox1790300000000`)
+
+See `backend/src/migrations/1790300000000-AddSearchOutbox.ts`.
+
+Creates `search_outbox` with operation (`index|delete`) and status (`pending|processing|done|failed`).
+
+## 2. ES query body — before / after
+
+### Before (unscoped)
+
+```json
+{
+  "query": {
+    "bool": {
+      "must": [{ "match_all": {} }],
+      "filter": [{ "term": { "city": "Lagos" } }]
+    }
+  }
+}
+```
+
+### After (mandatory tenant + visibility)
+
+```json
+{
+  "query": {
+    "bool": {
+      "must": [{ "match_all": {} }],
+      "filter": [
+        { "term": { "tenant_id": "<server-derived-tenant>" } },
+        { "terms": { "visibility": ["listed"] } },
+        { "term": { "city": "Lagos" } }
+      ]
+    }
+  }
+}
+```
+
+`tenant_id` and `visibility` are injected by `ElasticsearchService.buildScopedSearchBody` from `TenantContext`. Client `tenantId` / `visibility` query params are discarded.
+
+## 3. Outbox state across drift-and-reconcile
+
+| Phase | status | operation | attempts | notes |
+|-------|--------|-----------|----------|-------|
+| After archive TX commit | pending | index | 0 | Row written in same TX as `status=archived` |
+| Relay pick-up | processing | index | 0 | |
+| ES timeout | pending | index | 1 | Retry; exponential backoff via Bull |
+| Max attempts exceeded | failed | index | 5 | Dead-letter; metric `search_outbox_relay_dead_letter` |
+| Reconcile detects stale ES (`visibility=listed` vs PG `archived`) | pending | index | 0 | New outbox row enqueued |
+| Relay success | done | index | 0 | ES now `visibility=unlisted` |
+
+## 4. Reconciliation metrics
+
+Emitted via `MetricsService`:
+
+- `search_reconcile_drifted`
+- `search_reconcile_missing`
+- `search_reconcile_orphaned`
+- `search_outbox_relay_success`
+- `search_outbox_relay_retry`
+- `search_outbox_relay_dead_letter`
+
+Audit action: `SEARCH_RECONCILE` with result payload in metadata.

backend/src/migrations/1790300000000-AddSearchOutbox.ts

@@ -0,0 +1,70 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class AddSearchOutbox1790300000000 implements MigrationInterface {
+  name = 'AddSearchOutbox1790300000000';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `CREATE EXTENSION IF NOT EXISTS "uuid-ossp";`,
+    );
+
+    await queryRunner.query(`
+      DO $$ BEGIN
+        CREATE TYPE search_outbox_operation AS ENUM ('index', 'delete');
+      EXCEPTION
+        WHEN duplicate_object THEN null;
+      END $$;
+    `);
+
+    await queryRunner.query(`
+      DO $$ BEGIN
+        CREATE TYPE search_outbox_status AS ENUM ('pending', 'processing', 'done', 'failed');
+      EXCEPTION
+        WHEN duplicate_object THEN null;
+      END $$;
+    `);
+
+    await queryRunner.query(`
+      CREATE TABLE IF NOT EXISTS search_outbox (
+        id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+        aggregate_type VARCHAR(64) NOT NULL,
+        aggregate_id UUID NOT NULL,
+        tenant_id UUID NOT NULL,
+        operation search_outbox_operation NOT NULL,
+        payload JSONB NOT NULL DEFAULT '{}'::jsonb,
+        status search_outbox_status NOT NULL DEFAULT 'pending',
+        attempts INT NOT NULL DEFAULT 0,
+        created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+        processed_at TIMESTAMPTZ NULL
+      );
+    `);
+
+    await queryRunner.query(`
+      CREATE INDEX IF NOT EXISTS idx_search_outbox_status_created
+        ON search_outbox (status, created_at);
+    `);
+    await queryRunner.query(`
+      CREATE INDEX IF NOT EXISTS idx_search_outbox_aggregate
+        ON search_outbox (aggregate_type, aggregate_id);
+    `);
+    await queryRunner.query(`
+      CREATE INDEX IF NOT EXISTS idx_search_outbox_tenant
+        ON search_outbox (tenant_id);
+    `);
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `DROP INDEX IF EXISTS idx_search_outbox_tenant;`,
+    );
+    await queryRunner.query(
+      `DROP INDEX IF EXISTS idx_search_outbox_aggregate;`,
+    );
+    await queryRunner.query(
+      `DROP INDEX IF EXISTS idx_search_outbox_status_created;`,
+    );
+    await queryRunner.query(`DROP TABLE IF EXISTS search_outbox;`);
+    await queryRunner.query(`DROP TYPE IF EXISTS search_outbox_status;`);
+    await queryRunner.query(`DROP TYPE IF EXISTS search_outbox_operation;`);
+  }
+}

backend/src/modules/audit/entities/audit-log.entity.ts

@@ -52,6 +52,9 @@ export enum AuditAction {
   SECURITY_INCIDENT = 'SECURITY_INCIDENT',
   DATA_EXPORT = 'DATA_EXPORT',
   BULK_OPERATION = 'BULK_OPERATION',
+  // Search index consistency
+  SEARCH_RECONCILE = 'SEARCH_RECONCILE',
+  SEARCH_OUTBOX_DEAD_LETTER = 'SEARCH_OUTBOX_DEAD_LETTER',
 }
 
 export enum AuditStatus {

backend/src/modules/monitoring/metrics.service.ts

@@ -71,6 +71,19 @@ export class MetricsService {
     this.incrementMetric(key);
   }
 
+  // Search index consistency metrics
+  recordSearchOutboxRelay(
+    outcome: 'success' | 'retry' | 'dead_letter',
+  ): void {
+    this.incrementMetric(`search_outbox_relay_${outcome}`);
+  }
+
+  recordSearchReconcile(
+    kind: 'drifted' | 'missing' | 'orphaned' | 'ok',
+  ): void {
+    this.incrementMetric(`search_reconcile_${kind}`);
+  }
+
   // Get metrics in Prometheus format
   async getMetrics(): Promise<string> {
     let output = '# Chioma Backend Metrics\n';

backend/src/modules/properties/properties.module.ts

@@ -17,6 +17,7 @@ import { PropertyListingDraft } from './entities/property-listing-draft.entity';
 import { PropertyAvailability } from './entities/property-availability.entity';
 import { AvailabilityService } from './availability.service';
 import { AvailabilityController } from './availability.controller';
+import { SearchModule } from '../search/search.module';
 
 @Module({
   imports: [
@@ -29,6 +30,7 @@ import { AvailabilityController } from './availability.controller';
       PropertyListingDraft,
       PropertyAvailability,
     ]),
+    SearchModule,
   ],
   controllers: [
     PropertiesController,