diff --git a/.github/workflows/backend-ci-cd.yml b/.github/workflows/backend-ci-cd.yml index dd9d7f10..557a89ae 100644 --- a/.github/workflows/backend-ci-cd.yml +++ b/.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 diff --git a/backend/.env.example b/backend/.env.example index b7a5ac07..b379eff0 100644 --- a/backend/.env.example +++ b/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 * * * * diff --git a/backend/docker-compose.production.yml b/backend/docker-compose.production.yml index 2627663c..deb68ed2 100644 --- a/backend/docker-compose.production.yml +++ b/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,6 +79,25 @@ 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 @@ -74,3 +105,4 @@ networks: volumes: postgres-data: redis-data: + elasticsearch-data: diff --git a/backend/docker-compose.yml b/backend/docker-compose.yml index bd7f338e..2f13396d 100644 --- a/backend/docker-compose.yml +++ b/backend/docker-compose.yml @@ -13,6 +13,11 @@ 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 @@ -20,6 +25,31 @@ services: 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: diff --git a/backend/docs/search/SEARCH_INDEX_CONSISTENCY.md b/backend/docs/search/SEARCH_INDEX_CONSISTENCY.md new file mode 100644 index 00000000..a29cb370 --- /dev/null +++ b/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": "" } }, + { "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. diff --git a/backend/src/migrations/1790300000000-AddSearchOutbox.ts b/backend/src/migrations/1790300000000-AddSearchOutbox.ts new file mode 100644 index 00000000..115b556d --- /dev/null +++ b/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 { + 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 { + 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;`); + } +} diff --git a/backend/src/modules/audit/entities/audit-log.entity.ts b/backend/src/modules/audit/entities/audit-log.entity.ts index 8a0cde06..d0e43348 100644 --- a/backend/src/modules/audit/entities/audit-log.entity.ts +++ b/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 { diff --git a/backend/src/modules/monitoring/metrics.service.ts b/backend/src/modules/monitoring/metrics.service.ts index c0b7c044..c7e379e7 100644 --- a/backend/src/modules/monitoring/metrics.service.ts +++ b/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 { let output = '# Chioma Backend Metrics\n'; diff --git a/backend/src/modules/properties/properties.module.ts b/backend/src/modules/properties/properties.module.ts index 6290212d..7cd7f7d2 100644 --- a/backend/src/modules/properties/properties.module.ts +++ b/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, diff --git a/backend/src/modules/properties/properties.service.spec.ts b/backend/src/modules/properties/properties.service.spec.ts index 7c96d1dd..0cc6ade2 100644 --- a/backend/src/modules/properties/properties.service.spec.ts +++ b/backend/src/modules/properties/properties.service.spec.ts @@ -1,5 +1,6 @@ import { Test, TestingModule } from '@nestjs/testing'; import { getRepositoryToken } from '@nestjs/typeorm'; +import { DataSource } from 'typeorm'; import { NotFoundException, ForbiddenException, @@ -20,6 +21,7 @@ import { RentalUnit } from './entities/rental-unit.entity'; import { PropertyListingDraft } from './entities/property-listing-draft.entity'; import { User, UserRole, AuthMethod } from '../users/entities/user.entity'; import { KycStatus } from '../kyc/kyc-status.enum'; +import { SearchOutboxService } from '../search/search-outbox.service'; describe('PropertiesService', () => { let service: PropertiesService; @@ -218,6 +220,27 @@ describe('PropertiesService', () => { provide: CacheService, useValue: mockCacheService, }, + { + provide: DataSource, + useValue: { + transaction: jest.fn(async (cb: (m: unknown) => Promise) => + cb({ + save: jest.fn(async (_entity: unknown, value: unknown) => value), + delete: jest.fn(), + remove: jest.fn(), + create: jest.fn((_entity: unknown, value: unknown) => value), + findOne: jest.fn(async () => mockProperty), + }), + ), + }, + }, + { + provide: SearchOutboxService, + useValue: { + enqueueIndex: jest.fn().mockResolvedValue({}), + enqueueDelete: jest.fn().mockResolvedValue({}), + }, + }, ], }).compile(); @@ -515,7 +538,7 @@ describe('PropertiesService', () => { await service.update('property-id', updateDto, mockOwner); - expect(mockPropertyRepository.save).toHaveBeenCalled(); + expect(service['dataSource'].transaction).toHaveBeenCalled(); }); it('should update a property by admin', async () => { @@ -543,7 +566,6 @@ describe('PropertiesService', () => { it('should strip verificationStatus for non-admin owners', async () => { const draft = { ...mockProperty, verificationStatus: null }; mockPropertyRepository.findOne.mockResolvedValue(draft); - mockPropertyRepository.save.mockImplementation((p) => Promise.resolve(p)); await service.update( 'property-id', @@ -551,18 +573,14 @@ describe('PropertiesService', () => { mockOwner, ); - expect(mockPropertyRepository.save).toHaveBeenCalledWith( - expect.objectContaining({ - verificationStatus: null, - title: 'T', - }), - ); + expect(service['dataSource'].transaction).toHaveBeenCalled(); + expect(draft.verificationStatus).toBeNull(); + expect(draft.title).toBe('T'); }); it('should allow admin to set verificationStatus', async () => { const draft = { ...mockProperty, verificationStatus: null }; mockPropertyRepository.findOne.mockResolvedValue(draft); - mockPropertyRepository.save.mockImplementation((p) => Promise.resolve(p)); await service.update( 'property-id', @@ -570,9 +588,8 @@ describe('PropertiesService', () => { mockAdmin, ); - expect(mockPropertyRepository.save).toHaveBeenCalledWith( - expect.objectContaining({ verificationStatus: 'verified' }), - ); + expect(service['dataSource'].transaction).toHaveBeenCalled(); + expect(draft.verificationStatus).toBe('verified'); }); }); @@ -583,7 +600,7 @@ describe('PropertiesService', () => { await service.remove('property-id', mockOwner); - expect(mockPropertyRepository.remove).toHaveBeenCalledWith(mockProperty); + expect(service['dataSource'].transaction).toHaveBeenCalled(); }); it('should throw ForbiddenException for non-owner', async () => { @@ -598,17 +615,13 @@ describe('PropertiesService', () => { describe('publish', () => { it('should publish a draft property', async () => { const draftProperty = { ...mockProperty, status: ListingStatus.DRAFT }; - const publishedProperty = { - ...draftProperty, - status: ListingStatus.PUBLISHED, - }; mockPropertyRepository.findOne.mockResolvedValue(draftProperty); - mockPropertyRepository.save.mockResolvedValue(publishedProperty); const result = await service.publish('property-id', mockOwner); expect(result.status).toBe(ListingStatus.PUBLISHED); + expect(service['dataSource'].transaction).toHaveBeenCalled(); }); it('should throw BadRequestException if already published', async () => { @@ -652,17 +665,15 @@ describe('PropertiesService', () => { describe('archive', () => { it('should archive a property', async () => { - const archivedProperty = { + mockPropertyRepository.findOne.mockResolvedValue({ ...mockProperty, - status: ListingStatus.ARCHIVED, - }; - - mockPropertyRepository.findOne.mockResolvedValue(mockProperty); - mockPropertyRepository.save.mockResolvedValue(archivedProperty); + status: ListingStatus.PUBLISHED, + }); const result = await service.archive('property-id', mockOwner); expect(result.status).toBe(ListingStatus.ARCHIVED); + expect(service['dataSource'].transaction).toHaveBeenCalled(); }); it('should throw ForbiddenException for non-owner', async () => { @@ -676,14 +687,15 @@ describe('PropertiesService', () => { describe('markAsRented', () => { it('should mark a property as rented', async () => { - const rentedProperty = { ...mockProperty, status: ListingStatus.RENTED }; - - mockPropertyRepository.findOne.mockResolvedValue(mockProperty); - mockPropertyRepository.save.mockResolvedValue(rentedProperty); + mockPropertyRepository.findOne.mockResolvedValue({ + ...mockProperty, + status: ListingStatus.PUBLISHED, + }); const result = await service.markAsRented('property-id', mockOwner); expect(result.status).toBe(ListingStatus.RENTED); + expect(service['dataSource'].transaction).toHaveBeenCalled(); }); }); diff --git a/backend/src/modules/properties/properties.service.ts b/backend/src/modules/properties/properties.service.ts index 0c651a16..f958112a 100644 --- a/backend/src/modules/properties/properties.service.ts +++ b/backend/src/modules/properties/properties.service.ts @@ -6,7 +6,7 @@ import { } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; import * as crypto from 'crypto'; -import { Repository } from 'typeorm'; +import { DataSource, Repository } from 'typeorm'; import { Property, ListingStatus, @@ -27,6 +27,7 @@ import { CACHE_PREFIX_PROPERTIES_LIST, TTL_PUBLIC_PROPERTY_LIST_MS, } from '../../common/cache/cache.constants'; +import { SearchOutboxService } from '../search/search-outbox.service'; @Injectable() export class PropertiesService { @@ -42,6 +43,8 @@ export class PropertiesService { @InjectRepository(PropertyListingDraft) private readonly propertyListingDraftRepository: Repository, private readonly cacheService: CacheService, + private readonly dataSource: DataSource, + private readonly searchOutboxService: SearchOutboxService, ) {} private generateCacheKey(query: QueryPropertyDto): string { @@ -240,47 +243,48 @@ export class PropertiesService { delete safePatch.verificationStatus; } - Object.assign(property, safePatch); - await this.propertyRepository.save(property); - - if (images !== undefined) { - await this.imageRepository.delete({ propertyId: id }); - if (images.length > 0) { - const propertyImages = images.map((img) => - this.imageRepository.create({ - ...img, - propertyId: id, - }), - ); - await this.imageRepository.save(propertyImages); + await this.dataSource.transaction(async (manager) => { + Object.assign(property, safePatch); + await manager.save(Property, property); + + if (images !== undefined) { + await manager.delete(PropertyImage, { propertyId: id }); + if (images.length > 0) { + const propertyImages = images.map((img) => + manager.create(PropertyImage, { ...img, propertyId: id }), + ); + await manager.save(PropertyImage, propertyImages); + } } - } - if (amenities !== undefined) { - await this.amenityRepository.delete({ propertyId: id }); - if (amenities.length > 0) { - const propertyAmenities = amenities.map((amenity) => - this.amenityRepository.create({ - ...amenity, - propertyId: id, - }), - ); - await this.amenityRepository.save(propertyAmenities); + if (amenities !== undefined) { + await manager.delete(PropertyAmenity, { propertyId: id }); + if (amenities.length > 0) { + const propertyAmenities = amenities.map((amenity) => + manager.create(PropertyAmenity, { ...amenity, propertyId: id }), + ); + await manager.save(PropertyAmenity, propertyAmenities); + } } - } - if (rentalUnits !== undefined) { - await this.rentalUnitRepository.delete({ propertyId: id }); - if (rentalUnits.length > 0) { - const propertyUnits = rentalUnits.map((unit) => - this.rentalUnitRepository.create({ - ...unit, - propertyId: id, - }), - ); - await this.rentalUnitRepository.save(propertyUnits); + if (rentalUnits !== undefined) { + await manager.delete(RentalUnit, { propertyId: id }); + if (rentalUnits.length > 0) { + const propertyUnits = rentalUnits.map((unit) => + manager.create(RentalUnit, { ...unit, propertyId: id }), + ); + await manager.save(RentalUnit, propertyUnits); + } } - } + + const refreshed = await manager.findOne(Property, { + where: { id }, + relations: ['amenities'], + }); + if (refreshed) { + await this.searchOutboxService.enqueueIndex(manager, refreshed); + } + }); await this.cacheService.invalidatePropertyDomainCaches(id); return this.findOne(id); @@ -289,7 +293,16 @@ export class PropertiesService { async remove(id: string, user: User): Promise { const property = await this.findOne(id); this.verifyOwnership(property, user); - await this.propertyRepository.remove(property); + + await this.dataSource.transaction(async (manager) => { + await this.searchOutboxService.enqueueDelete( + manager, + property.id, + property.ownerId, + ); + await manager.remove(Property, property); + }); + await this.cacheService.invalidatePropertyDomainCaches(id); } @@ -317,8 +330,20 @@ export class PropertiesService { ); } - property.status = ListingStatus.PUBLISHED; - const saved = await this.propertyRepository.save(property); + const saved = await this.dataSource.transaction(async (manager) => { + property.status = ListingStatus.PUBLISHED; + const updated = await manager.save(Property, property); + const withAmenities = await manager.findOne(Property, { + where: { id }, + relations: ['amenities'], + }); + await this.searchOutboxService.enqueueIndex( + manager, + withAmenities ?? updated, + ); + return updated; + }); + await this.cacheService.invalidatePropertyDomainCaches(id); return saved; } @@ -326,8 +351,21 @@ export class PropertiesService { async archive(id: string, user: User): Promise { const property = await this.findOne(id); this.verifyOwnership(property, user); - property.status = ListingStatus.ARCHIVED; - const saved = await this.propertyRepository.save(property); + + const saved = await this.dataSource.transaction(async (manager) => { + property.status = ListingStatus.ARCHIVED; + const updated = await manager.save(Property, property); + const withAmenities = await manager.findOne(Property, { + where: { id }, + relations: ['amenities'], + }); + await this.searchOutboxService.enqueueIndex( + manager, + withAmenities ?? updated, + ); + return updated; + }); + await this.cacheService.invalidatePropertyDomainCaches(id); return saved; } @@ -335,8 +373,21 @@ export class PropertiesService { async markAsRented(id: string, user: User): Promise { const property = await this.findOne(id); this.verifyOwnership(property, user); - property.status = ListingStatus.RENTED; - const saved = await this.propertyRepository.save(property); + + const saved = await this.dataSource.transaction(async (manager) => { + property.status = ListingStatus.RENTED; + const updated = await manager.save(Property, property); + const withAmenities = await manager.findOne(Property, { + where: { id }, + relations: ['amenities'], + }); + await this.searchOutboxService.enqueueIndex( + manager, + withAmenities ?? updated, + ); + return updated; + }); + await this.cacheService.invalidatePropertyDomainCaches(id); return saved; } diff --git a/backend/src/modules/queues/processors/search-index.processor.ts b/backend/src/modules/queues/processors/search-index.processor.ts new file mode 100644 index 00000000..a6b98f19 --- /dev/null +++ b/backend/src/modules/queues/processors/search-index.processor.ts @@ -0,0 +1,85 @@ +import { Process, Processor, OnQueueActive } from '@nestjs/bull'; +import { Logger, OnModuleInit } from '@nestjs/common'; +import { InjectQueue } from '@nestjs/bull'; +import { Job, Queue } from 'bull'; +import { ConfigService } from '@nestjs/config'; +import { SearchOutboxRelay } from '../../search/search-outbox.relay'; +import { SearchReconcileJob } from '../../search/search-reconcile.job'; + +export type SearchIndexJobType = 'relay' | 'reconcile'; + +export interface SearchIndexJobData { + type: SearchIndexJobType; +} + +export const SEARCH_INDEX_QUEUE = 'search-index'; + +@Processor(SEARCH_INDEX_QUEUE) +export class SearchIndexProcessor implements OnModuleInit { + private readonly logger = new Logger(SearchIndexProcessor.name); + + constructor( + @InjectQueue(SEARCH_INDEX_QUEUE) private readonly searchIndexQueue: Queue, + private readonly relay: SearchOutboxRelay, + private readonly reconcileJob: SearchReconcileJob, + private readonly configService: ConfigService, + ) {} + + async onModuleInit(): Promise { + const pollIntervalMs = this.configService.get( + 'SEARCH_OUTBOX_POLL_INTERVAL_MS', + 5000, + ); + const reconcileCron = this.configService.get( + 'SEARCH_RECONCILE_CRON', + '0 */15 * * * *', // every 15 minutes + ); + + // Repeatable relay drain + await this.searchIndexQueue.add( + { type: 'relay' } satisfies SearchIndexJobData, + { + jobId: 'search-outbox-relay', + repeat: { every: pollIntervalMs }, + removeOnComplete: true, + removeOnFail: false, + }, + ); + + // Repeatable reconcile (Bull cron) + await this.searchIndexQueue.add( + { type: 'reconcile' } satisfies SearchIndexJobData, + { + jobId: 'search-reconcile', + repeat: { cron: reconcileCron }, + removeOnComplete: true, + removeOnFail: false, + }, + ); + + this.logger.log( + `Registered search-index repeatables (relay every ${pollIntervalMs}ms, reconcile cron=${reconcileCron})`, + ); + } + + @OnQueueActive() + onActive(job: Job): void { + this.logger.debug(`Processing search-index job ${job.id}: ${job.data.type}`); + } + + @Process() + async handle(job: Job): Promise { + switch (job.data.type) { + case 'relay': + await this.relay.drain(); + break; + case 'reconcile': + await this.reconcileJob.run(); + break; + default: + throw new Error( + `Unknown search-index job type: ${String((job.data as SearchIndexJobData).type)}`, + ); + } + } +} diff --git a/backend/src/modules/queues/queues.module.ts b/backend/src/modules/queues/queues.module.ts index 9e794b1c..91fc91f0 100644 --- a/backend/src/modules/queues/queues.module.ts +++ b/backend/src/modules/queues/queues.module.ts @@ -5,12 +5,17 @@ import { EmailQueueProcessor } from './processors/email.processor'; import { DocumentQueueProcessor } from './processors/document.processor'; import { BlockchainQueueProcessor } from './processors/blockchain.processor'; import { DataSyncQueueProcessor } from './processors/data-sync.processor'; +import { + SearchIndexProcessor, + SEARCH_INDEX_QUEUE, +} from './processors/search-index.processor'; import { QueueMonitoringService } from './services/queue-monitoring.service'; import { QueueManagementService } from './services/queue-management.service'; import { QueuesController } from './controllers/queues.controller'; import { NotificationsModule } from '../notifications/notifications.module'; import { StorageModule } from '../storage/storage.module'; import { StellarModule } from '../stellar/stellar.module'; +import { SearchModule } from '../search/search.module'; @Module({ imports: [ @@ -47,16 +52,19 @@ import { StellarModule } from '../stellar/stellar.module'; { name: 'documents' }, { name: 'blockchain' }, { name: 'data-sync' }, + { name: SEARCH_INDEX_QUEUE }, ), NotificationsModule, StorageModule, StellarModule, + SearchModule, ], providers: [ EmailQueueProcessor, DocumentQueueProcessor, BlockchainQueueProcessor, DataSyncQueueProcessor, + SearchIndexProcessor, QueueMonitoringService, QueueManagementService, ], diff --git a/backend/src/modules/queues/services/queue-management.service.ts b/backend/src/modules/queues/services/queue-management.service.ts index 0de22403..0f1cecd4 100644 --- a/backend/src/modules/queues/services/queue-management.service.ts +++ b/backend/src/modules/queues/services/queue-management.service.ts @@ -27,6 +27,7 @@ export class QueueManagementService { @InjectQueue('documents') private documentsQueue: Queue, @InjectQueue('blockchain') private blockchainQueue: Queue, @InjectQueue('data-sync') private dataSyncQueue: Queue, + @InjectQueue('search-index') private searchIndexQueue: Queue, ) {} /** @@ -104,6 +105,27 @@ export class QueueManagementService { return this.dataSyncQueue.add(data, defaultOptions); } + /** + * Add search-index relay/reconcile job + */ + async addSearchIndexJob( + data: JobData, + options?: QueueJobOptions, + ): Promise { + const defaultOptions = { + attempts: 3, + backoff: { + type: 'exponential' as const, + delay: 2000, + }, + removeOnComplete: true, + ...options, + }; + + this.logger.debug(`Adding search-index job: ${JSON.stringify(data)}`); + return this.searchIndexQueue.add(data, defaultOptions); + } + /** * Get queue statistics */ @@ -126,7 +148,13 @@ export class QueueManagementService { * Get all queue statistics */ async getAllQueueStats(): Promise { - const queues = ['email', 'documents', 'blockchain', 'data-sync']; + const queues = [ + 'email', + 'documents', + 'blockchain', + 'data-sync', + 'search-index', + ]; return Promise.all(queues.map((q) => this.getQueueStats(q))); } @@ -233,6 +261,8 @@ export class QueueManagementService { return this.blockchainQueue; case 'data-sync': return this.dataSyncQueue; + case 'search-index': + return this.searchIndexQueue; default: throw new Error(`Unknown queue: ${queueName}`); } diff --git a/backend/src/modules/queues/services/queue-monitoring.service.ts b/backend/src/modules/queues/services/queue-monitoring.service.ts index 262d8fdd..c79d794c 100644 --- a/backend/src/modules/queues/services/queue-monitoring.service.ts +++ b/backend/src/modules/queues/services/queue-monitoring.service.ts @@ -25,6 +25,7 @@ export class QueueMonitoringService { @InjectQueue('documents') private documentsQueue: Queue, @InjectQueue('blockchain') private blockchainQueue: Queue, @InjectQueue('data-sync') private dataSyncQueue: Queue, + @InjectQueue('search-index') private searchIndexQueue: Queue, ) { this.initializeMetrics(); } @@ -34,6 +35,7 @@ export class QueueMonitoringService { this.metrics.set('documents', []); this.metrics.set('blockchain', []); this.metrics.set('data-sync', []); + this.metrics.set('search-index', []); } /** @@ -46,6 +48,7 @@ export class QueueMonitoringService { { name: 'documents', queue: this.documentsQueue }, { name: 'blockchain', queue: this.blockchainQueue }, { name: 'data-sync', queue: this.dataSyncQueue }, + { name: 'search-index', queue: this.searchIndexQueue }, ]; for (const { name, queue } of queues) { diff --git a/backend/src/modules/search/__tests__/search-outbox.relay.spec.ts b/backend/src/modules/search/__tests__/search-outbox.relay.spec.ts new file mode 100644 index 00000000..dfa2a8d5 --- /dev/null +++ b/backend/src/modules/search/__tests__/search-outbox.relay.spec.ts @@ -0,0 +1,125 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { getRepositoryToken } from '@nestjs/typeorm'; +import { ConfigService } from '@nestjs/config'; +import { SearchOutboxRelay } from '../search-outbox.relay'; +import { SearchOutboxService } from '../search-outbox.service'; +import { ElasticsearchService } from '../elasticsearch.service'; +import { MetricsService } from '../../monitoring/metrics.service'; +import { + SearchOutbox, + SearchOutboxOperation, + SearchOutboxStatus, +} from '../entities/search-outbox.entity'; +import { SearchVisibility } from '../search-visibility'; + +describe('SearchOutboxRelay', () => { + let relay: SearchOutboxRelay; + let outboxService: jest.Mocked; + let elasticsearch: jest.Mocked>; + let metrics: jest.Mocked>; + + const pendingRow: SearchOutbox = { + id: 'outbox-1', + aggregateType: 'property', + aggregateId: 'prop-1', + tenantId: 'tenant-1', + operation: SearchOutboxOperation.INDEX, + payload: { + id: 'prop-1', + tenant_id: 'tenant-1', + visibility: SearchVisibility.LISTED, + title: 'Loft', + description: '', + type: 'apartment', + city: 'Lagos', + state: '', + country: 'NG', + price: 100, + bedrooms: 1, + bathrooms: 1, + area: 40, + amenities: [], + location: { lat: 0, lon: 0 }, + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }, + status: SearchOutboxStatus.PENDING, + attempts: 0, + createdAt: new Date(), + processedAt: null, + }; + + beforeEach(async () => { + outboxService = { + claimPending: jest.fn(), + markProcessing: jest.fn(), + markDone: jest.fn(), + markFailedOrRetry: jest.fn(), + } as unknown as jest.Mocked; + + elasticsearch = { + isEnabled: jest.fn().mockReturnValue(true), + indexProperty: jest.fn().mockResolvedValue(undefined), + removeProperty: jest.fn().mockResolvedValue(undefined), + }; + + metrics = { + recordSearchOutboxRelay: jest.fn(), + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + SearchOutboxRelay, + { provide: SearchOutboxService, useValue: outboxService }, + { provide: ElasticsearchService, useValue: elasticsearch }, + { provide: MetricsService, useValue: metrics }, + { + provide: ConfigService, + useValue: { + get: jest.fn((key: string, fallback?: number) => + key === 'SEARCH_OUTBOX_MAX_ATTEMPTS' ? 3 : fallback, + ), + }, + }, + { + provide: getRepositoryToken(SearchOutbox), + useValue: {}, + }, + ], + }).compile(); + + relay = module.get(SearchOutboxRelay); + }); + + it('drains pending rows to ES and marks done', async () => { + outboxService.claimPending.mockResolvedValue([pendingRow]); + + const result = await relay.drain(); + + expect(elasticsearch.indexProperty).toHaveBeenCalledWith( + pendingRow.payload, + 'outbox-1', + ); + expect(outboxService.markDone).toHaveBeenCalledWith('outbox-1'); + expect(result.succeeded).toBe(1); + expect(metrics.recordSearchOutboxRelay).toHaveBeenCalledWith('success'); + }); + + it('retries transient failures and dead-letters after max attempts', async () => { + const failing = { ...pendingRow, attempts: 2 }; + outboxService.claimPending.mockResolvedValue([failing]); + (elasticsearch.indexProperty as jest.Mock).mockRejectedValue( + new Error('timeout'), + ); + + const result = await relay.drain(); + + expect(outboxService.markFailedOrRetry).toHaveBeenCalledWith( + 'outbox-1', + 2, + 3, + ); + expect(result.deadLetter).toBe(1); + expect(metrics.recordSearchOutboxRelay).toHaveBeenCalledWith('dead_letter'); + }); +}); diff --git a/backend/src/modules/search/__tests__/search-outbox.transaction.spec.ts b/backend/src/modules/search/__tests__/search-outbox.transaction.spec.ts new file mode 100644 index 00000000..3225f7a2 --- /dev/null +++ b/backend/src/modules/search/__tests__/search-outbox.transaction.spec.ts @@ -0,0 +1,64 @@ +import { SearchOutboxService } from '../search-outbox.service'; +import { + SearchOutbox, + SearchOutboxOperation, + SearchOutboxStatus, +} from '../entities/search-outbox.entity'; +import { + ListingStatus, + Property, + PropertyType, +} from '../../properties/entities/property.entity'; +import { EntityManager } from 'typeorm'; + +describe('SearchOutboxService transactional insert', () => { + it('persists the outbox row via the provided EntityManager (same TX)', async () => { + const saved: SearchOutbox[] = []; + const manager = { + create: jest.fn((_entity, data) => data), + save: jest.fn(async (_entity, row) => { + saved.push(row as SearchOutbox); + return { ...row, id: 'outbox-tx-1' }; + }), + } as unknown as EntityManager; + + const service = new SearchOutboxService({} as never); + + const property = { + id: 'prop-1', + ownerId: 'tenant-1', + title: 'Loft', + description: 'Nice', + type: PropertyType.APARTMENT, + status: ListingStatus.PUBLISHED, + price: 100, + city: 'Lagos', + state: '', + country: 'NG', + bedrooms: 1, + bathrooms: 1, + area: 40, + latitude: 0, + longitude: 0, + amenities: [], + createdAt: new Date(), + updatedAt: new Date(), + } as unknown as Property; + + await service.enqueueIndex(manager, property); + + expect(manager.save).toHaveBeenCalled(); + expect(saved[0].aggregateId).toBe('prop-1'); + expect(saved[0].tenantId).toBe('tenant-1'); + expect(saved[0].operation).toBe(SearchOutboxOperation.INDEX); + expect(saved[0].status).toBe(SearchOutboxStatus.PENDING); + }); + + it('documents that rolling back the source TX rolls back the outbox row', () => { + // Invariant: enqueueIndex uses manager.save (not a separate connection). + // If the caller's dataSource.transaction rolls back, the outbox insert is undone. + const source = SearchOutboxService.prototype.enqueueIndex.toString(); + expect(source).toContain('manager.save'); + expect(source).not.toContain('this.outboxRepo.save'); + }); +}); diff --git a/backend/src/modules/search/__tests__/search-reconcile.job.spec.ts b/backend/src/modules/search/__tests__/search-reconcile.job.spec.ts new file mode 100644 index 00000000..487e828b --- /dev/null +++ b/backend/src/modules/search/__tests__/search-reconcile.job.spec.ts @@ -0,0 +1,92 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { getRepositoryToken } from '@nestjs/typeorm'; +import { DataSource } from 'typeorm'; +import { SearchReconcileJob } from '../search-reconcile.job'; +import { SearchOutboxService } from '../search-outbox.service'; +import { ElasticsearchService } from '../elasticsearch.service'; +import { MetricsService } from '../../monitoring/metrics.service'; +import { AuditService } from '../../audit/audit.service'; +import { + ListingStatus, + Property, + PropertyType, +} from '../../properties/entities/property.entity'; +import { SearchVisibility } from '../search-visibility'; + +describe('SearchReconcileJob', () => { + let job: SearchReconcileJob; + let outboxService: { enqueueIndex: jest.Mock; computeChecksum: jest.Mock }; + let elasticsearch: { + isEnabled: jest.Mock; + getDocument: jest.Mock; + scrollAllIds: jest.Mock; + removeProperty: jest.Mock; + }; + + const property = { + id: 'prop-1', + ownerId: 'tenant-1', + title: 'Loft', + status: ListingStatus.ARCHIVED, + type: PropertyType.APARTMENT, + price: 100, + updatedAt: new Date('2026-01-01T00:00:00.000Z'), + amenities: [], + } as unknown as Property; + + beforeEach(async () => { + outboxService = { + enqueueIndex: jest.fn().mockResolvedValue({}), + computeChecksum: jest.fn().mockReturnValue('cabc'), + }; + elasticsearch = { + isEnabled: jest.fn().mockReturnValue(true), + getDocument: jest.fn(), + scrollAllIds: jest.fn().mockResolvedValue(['prop-1']), + removeProperty: jest.fn().mockResolvedValue(undefined), + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + SearchReconcileJob, + { + provide: getRepositoryToken(Property), + useValue: { find: jest.fn().mockResolvedValue([property]) }, + }, + { provide: SearchOutboxService, useValue: outboxService }, + { provide: ElasticsearchService, useValue: elasticsearch }, + { + provide: DataSource, + useValue: { + transaction: jest.fn(async (cb: (m: unknown) => Promise) => + cb({}), + ), + }, + }, + { + provide: MetricsService, + useValue: { recordSearchReconcile: jest.fn() }, + }, + { provide: AuditService, useValue: { log: jest.fn() } }, + ], + }).compile(); + + job = module.get(SearchReconcileJob); + }); + + it('detects a stale ES document and enqueues an outbox index row', async () => { + elasticsearch.getDocument.mockResolvedValue({ + id: 'prop-1', + tenant_id: 'tenant-1', + visibility: SearchVisibility.LISTED, // stale — PG is archived/unlisted + checksum: 'cold', + updatedAt: '2025-01-01T00:00:00.000Z', + }); + + const result = await job.run(); + + expect(result.drifted).toBe(1); + expect(result.enqueued).toBe(1); + expect(outboxService.enqueueIndex).toHaveBeenCalled(); + }); +}); diff --git a/backend/src/modules/search/__tests__/search-scope.spec.ts b/backend/src/modules/search/__tests__/search-scope.spec.ts new file mode 100644 index 00000000..e4874554 --- /dev/null +++ b/backend/src/modules/search/__tests__/search-scope.spec.ts @@ -0,0 +1,99 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { getRepositoryToken } from '@nestjs/typeorm'; +import { SearchService } from '../search.service'; +import { Property } from '../../properties/entities/property.entity'; +import { CacheService } from '../../../common/cache/cache.service'; +import { ElasticsearchService } from '../elasticsearch.service'; +import { SearchScopeError } from '../search-scope.error'; +import { SearchVisibility } from '../search-visibility'; +import { discoveryTenantContext } from '../tenant-context'; +import { ConfigService } from '@nestjs/config'; + +describe('SearchService.query scope', () => { + let service: SearchService; + let elasticsearch: ElasticsearchService; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [ + SearchService, + ElasticsearchService, + { + provide: getRepositoryToken(Property), + useValue: { + createQueryBuilder: jest.fn(() => ({ + leftJoinAndSelect: jest.fn().mockReturnThis(), + where: jest.fn().mockReturnThis(), + andWhere: jest.fn().mockReturnThis(), + skip: jest.fn().mockReturnThis(), + take: jest.fn().mockReturnThis(), + orderBy: jest.fn().mockReturnThis(), + getManyAndCount: jest.fn().mockResolvedValue([[], 0]), + })), + }, + }, + { + provide: CacheService, + useValue: { + getOrSet: jest.fn((_k: string, fn: () => Promise) => fn()), + }, + }, + { + provide: ConfigService, + useValue: { + get: jest.fn((_key: string, fallback?: unknown) => fallback), + }, + }, + ], + }).compile(); + + service = module.get(SearchService); + elasticsearch = module.get(ElasticsearchService); + elasticsearch.setEnabledForTests(false); + }); + + it('throws SearchScopeError when tenant is absent', async () => { + await expect(service.query(null, {})).rejects.toBeInstanceOf( + SearchScopeError, + ); + await expect(service.query(undefined, {})).rejects.toBeInstanceOf( + SearchScopeError, + ); + await expect( + service.query({ tenantId: '', allowedVisibilities: [SearchVisibility.LISTED] }), + ).rejects.toBeInstanceOf(SearchScopeError); + }); + + it('always emits both tenant_id and visibility filters in the ES body', async () => { + const tenant = discoveryTenantContext('tenant-abc'); + const result = await service.query(tenant, { city: 'Lagos', page: 1 }); + + expect(result.searchBody).toBeDefined(); + const filter = ( + result.searchBody as { + query: { bool: { filter: Array> } }; + } + ).query.bool.filter; + + expect(filter).toEqual( + expect.arrayContaining([ + { term: { tenant_id: 'tenant-abc' } }, + { terms: { visibility: [SearchVisibility.LISTED] } }, + ]), + ); + + // Caller-supplied city is additive; scope filters remain. + expect(filter).toEqual( + expect.arrayContaining([{ term: { city: 'Lagos' } }]), + ); + }); + + it('buildScopedSearchBody rejects missing visibility set', () => { + expect(() => + elasticsearch.buildScopedSearchBody( + { tenantId: 't1', allowedVisibilities: [] }, + {}, + ), + ).toThrow(SearchScopeError); + }); +}); diff --git a/backend/src/modules/search/elasticsearch.service.ts b/backend/src/modules/search/elasticsearch.service.ts index b51bb016..8a705939 100644 --- a/backend/src/modules/search/elasticsearch.service.ts +++ b/backend/src/modules/search/elasticsearch.service.ts @@ -1,8 +1,13 @@ import { Injectable, Logger, OnModuleInit } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; +import { SearchVisibility } from './search-visibility'; +import { TenantContext } from './tenant-context'; +import { SearchScopeError } from './search-scope.error'; export interface PropertySearchDocument { id: string; + tenant_id: string; + visibility: SearchVisibility | string; title: string; description: string; type: string; @@ -18,11 +23,13 @@ export interface PropertySearchDocument { lat: number; lon: number; }; + status?: string; + checksum?: string; createdAt: string; updatedAt: string; } -export interface SearchFilters { +export interface EsSearchFilters { query?: string; city?: string; state?: string; @@ -44,26 +51,39 @@ export interface SearchFilters { sortOrder?: 'asc' | 'desc'; } -export interface SearchResult { +export interface EsSearchResult { hits: T[]; total: number; page: number; limit: number; facets: Record>; + /** Exposed for tests / PR attachments — the ES body that was sent. */ + searchBody?: Record; } @Injectable() export class ElasticsearchService implements OnModuleInit { private readonly logger = new Logger(ElasticsearchService.name); private readonly esUrl: string; - private readonly indexName = 'properties'; + private readonly indexPrefix: string; private enabled = false; constructor(private configService: ConfigService) { this.esUrl = this.configService.get( - 'ELASTICSEARCH_URL', - 'http://localhost:9200', + 'ELASTICSEARCH_NODE', + this.configService.get( + 'ELASTICSEARCH_URL', + 'http://localhost:9200', + )!, ); + this.indexPrefix = this.configService.get( + 'SEARCH_INDEX_PREFIX', + 'vespera', + ); + } + + get indexName(): string { + return `${this.indexPrefix}-properties`; } async onModuleInit(): Promise { @@ -85,6 +105,11 @@ export class ElasticsearchService implements OnModuleInit { return this.enabled; } + /** Test helper to force enable/disable without a live cluster. */ + setEnabledForTests(enabled: boolean): void { + this.enabled = enabled; + } + private async ensureIndex(): Promise { const exists = await fetch(`${this.esUrl}/${this.indexName}`); if (exists.status === 404) { @@ -110,6 +135,10 @@ export class ElasticsearchService implements OnModuleInit { mappings: { properties: { id: { type: 'keyword' }, + tenant_id: { type: 'keyword' }, + visibility: { type: 'keyword' }, + status: { type: 'keyword' }, + checksum: { type: 'keyword' }, title: { type: 'text', analyzer: 'property_analyzer', @@ -146,22 +175,59 @@ export class ElasticsearchService implements OnModuleInit { } } - async indexProperty(doc: PropertySearchDocument): Promise { - if (!this.enabled) return; + async indexProperty( + doc: PropertySearchDocument, + idempotencyKey?: string, + ): Promise { + if (!this.enabled) { + throw new Error('Elasticsearch is not enabled'); + } - await fetch(`${this.esUrl}/${this.indexName}/_doc/${doc.id}`, { - method: 'PUT', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify(doc), - }); + const headers: Record = { + 'Content-Type': 'application/json', + }; + if (idempotencyKey) { + headers['Idempotency-Key'] = idempotencyKey; + } + + const response = await fetch( + `${this.esUrl}/${this.indexName}/_doc/${doc.id}`, + { + method: 'PUT', + headers, + body: JSON.stringify(doc), + }, + ); + + if (!response.ok) { + const text = await response.text(); + throw new Error(`ES index failed: ${response.status} ${text}`); + } } - async removeProperty(id: string): Promise { - if (!this.enabled) return; + async removeProperty(id: string, idempotencyKey?: string): Promise { + if (!this.enabled) { + throw new Error('Elasticsearch is not enabled'); + } + + const headers: Record = {}; + if (idempotencyKey) { + headers['Idempotency-Key'] = idempotencyKey; + } - await fetch(`${this.esUrl}/${this.indexName}/_doc/${id}`, { - method: 'DELETE', - }); + const response = await fetch( + `${this.esUrl}/${this.indexName}/_doc/${id}`, + { + method: 'DELETE', + headers, + }, + ); + + // 404 is idempotent success for deletes + if (!response.ok && response.status !== 404) { + const text = await response.text(); + throw new Error(`ES delete failed: ${response.status} ${text}`); + } } async bulkIndex(docs: PropertySearchDocument[]): Promise { @@ -182,25 +248,36 @@ export class ElasticsearchService implements OnModuleInit { }); if (!response.ok) { - this.logger.error('Bulk indexing failed'); + throw new Error('Bulk indexing failed'); } } - async search( - filters: SearchFilters, - ): Promise> { - if (!this.enabled) { - return { hits: [], total: 0, page: 1, limit: 20, facets: {} }; + /** + * Build a scoped ES search body. Always includes non-removable tenant_id + * and visibility filters derived from TenantContext. + */ + buildScopedSearchBody( + tenant: TenantContext, + filters: EsSearchFilters, + ): Record { + if (!tenant?.tenantId) { + throw new SearchScopeError('tenant_id is required for search'); + } + if (!tenant.allowedVisibilities?.length) { + throw new SearchScopeError('visibility scope is required for search'); } const page = filters.page || 1; const limit = filters.limit || 20; const from = (page - 1) * limit; - const must: any[] = []; - const filterClauses: any[] = []; + const must: Record[] = []; + // Mandatory, non-removable scope filters — never accept from caller. + const filterClauses: Record[] = [ + { term: { tenant_id: tenant.tenantId } }, + { terms: { visibility: tenant.allowedVisibilities } }, + ]; - // Full-text search if (filters.query) { must.push({ multi_match: { @@ -211,7 +288,6 @@ export class ElasticsearchService implements OnModuleInit { }); } - // Keyword filters if (filters.city) filterClauses.push({ term: { city: filters.city } }); if (filters.state) filterClauses.push({ term: { state: filters.state } }); if (filters.country) @@ -222,20 +298,17 @@ export class ElasticsearchService implements OnModuleInit { if (filters.bathrooms) filterClauses.push({ term: { bathrooms: filters.bathrooms } }); - // Price range if (filters.minPrice || filters.maxPrice) { - const range: any = {}; + const range: Record = {}; if (filters.minPrice) range.gte = filters.minPrice; if (filters.maxPrice) range.lte = filters.maxPrice; filterClauses.push({ range: { price: range } }); } - // Amenities if (filters.amenities && filters.amenities.length > 0) { filterClauses.push({ terms: { amenities: filters.amenities } }); } - // Geolocation if (filters.location) { filterClauses.push({ geo_distance: { @@ -248,15 +321,13 @@ export class ElasticsearchService implements OnModuleInit { }); } - // Sort - const sort: any[] = []; + const sort: Record[] = []; if (filters.sortBy) { sort.push({ [filters.sortBy]: { order: filters.sortOrder || 'desc' } }); } else { sort.push({ _score: { order: 'desc' } }); } - // Aggregations for faceted search const aggs = { types: { terms: { field: 'type', size: 20 } }, cities: { terms: { field: 'city', size: 50 } }, @@ -275,7 +346,7 @@ export class ElasticsearchService implements OnModuleInit { amenities: { terms: { field: 'amenities', size: 30 } }, }; - const searchBody = { + return { from, size: limit, query: { @@ -287,6 +358,19 @@ export class ElasticsearchService implements OnModuleInit { sort, aggs, }; + } + + async search( + tenant: TenantContext, + filters: EsSearchFilters, + ): Promise> { + const page = filters.page || 1; + const limit = filters.limit || 20; + const searchBody = this.buildScopedSearchBody(tenant, filters); + + if (!this.enabled) { + return { hits: [], total: 0, page, limit, facets: {}, searchBody }; + } const response = await fetch(`${this.esUrl}/${this.indexName}/_search`, { method: 'POST', @@ -296,13 +380,13 @@ export class ElasticsearchService implements OnModuleInit { if (!response.ok) { this.logger.error('Search query failed'); - return { hits: [], total: 0, page, limit, facets: {} }; + return { hits: [], total: 0, page, limit, facets: {}, searchBody }; } const data = await response.json(); const hits = (data.hits?.hits || []).map( - (hit: any) => hit._source as PropertySearchDocument, + (hit: { _source: PropertySearchDocument }) => hit._source, ); const total = typeof data.hits?.total === 'object' @@ -312,15 +396,42 @@ export class ElasticsearchService implements OnModuleInit { const facets: Record> = {}; if (data.aggregations) { for (const [name, agg] of Object.entries( - data.aggregations as Record, + data.aggregations as Record }>, )) { - facets[name] = (agg.buckets || []).map((b: any) => ({ + facets[name] = (agg.buckets || []).map((b) => ({ key: b.key, count: b.doc_count, })); } } - return { hits, total, page, limit, facets }; + return { hits, total, page, limit, facets, searchBody }; + } + + async getDocument(id: string): Promise { + if (!this.enabled) return null; + const response = await fetch( + `${this.esUrl}/${this.indexName}/_doc/${id}`, + ); + if (response.status === 404) return null; + if (!response.ok) return null; + const data = await response.json(); + return (data._source as PropertySearchDocument) ?? null; + } + + async scrollAllIds(): Promise { + if (!this.enabled) return []; + const response = await fetch(`${this.esUrl}/${this.indexName}/_search`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + size: 1000, + _source: false, + query: { match_all: {} }, + }), + }); + if (!response.ok) return []; + const data = await response.json(); + return (data.hits?.hits || []).map((h: { _id: string }) => h._id); } } diff --git a/backend/src/modules/search/entities/search-outbox.entity.ts b/backend/src/modules/search/entities/search-outbox.entity.ts new file mode 100644 index 00000000..5f53324b --- /dev/null +++ b/backend/src/modules/search/entities/search-outbox.entity.ts @@ -0,0 +1,67 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + Index, +} from 'typeorm'; + +export enum SearchOutboxOperation { + INDEX = 'index', + DELETE = 'delete', +} + +export enum SearchOutboxStatus { + PENDING = 'pending', + PROCESSING = 'processing', + DONE = 'done', + FAILED = 'failed', +} + +@Entity('search_outbox') +@Index(['status', 'createdAt']) +@Index(['aggregateType', 'aggregateId']) +@Index(['tenantId']) +export class SearchOutbox { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column({ name: 'aggregate_type', type: 'varchar', length: 64 }) + aggregateType: string; + + @Column({ name: 'aggregate_id', type: 'uuid' }) + aggregateId: string; + + @Column({ name: 'tenant_id', type: 'uuid' }) + tenantId: string; + + @Column({ + type: 'enum', + enum: SearchOutboxOperation, + enumName: 'search_outbox_operation', + }) + operation: SearchOutboxOperation; + + @Column({ + type: process.env.DB_TYPE === 'sqlite' ? 'simple-json' : 'jsonb', + default: {}, + }) + payload: Record; + + @Column({ + type: 'enum', + enum: SearchOutboxStatus, + enumName: 'search_outbox_status', + default: SearchOutboxStatus.PENDING, + }) + status: SearchOutboxStatus; + + @Column({ type: 'int', default: 0 }) + attempts: number; + + @CreateDateColumn({ name: 'created_at', type: 'timestamptz' }) + createdAt: Date; + + @Column({ name: 'processed_at', type: 'timestamptz', nullable: true }) + processedAt: Date | null; +} diff --git a/backend/src/modules/search/search-outbox.relay.ts b/backend/src/modules/search/search-outbox.relay.ts new file mode 100644 index 00000000..ff73837d --- /dev/null +++ b/backend/src/modules/search/search-outbox.relay.ts @@ -0,0 +1,100 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { SearchOutboxService } from './search-outbox.service'; +import { ElasticsearchService, PropertySearchDocument } from './elasticsearch.service'; +import { + SearchOutboxOperation, +} from './entities/search-outbox.entity'; +import { MetricsService } from '../monitoring/metrics.service'; + +@Injectable() +export class SearchOutboxRelay { + private readonly logger = new Logger(SearchOutboxRelay.name); + private readonly maxAttempts: number; + private running = false; + + constructor( + private readonly outboxService: SearchOutboxService, + private readonly elasticsearch: ElasticsearchService, + private readonly configService: ConfigService, + private readonly metricsService: MetricsService, + ) { + this.maxAttempts = this.configService.get( + 'SEARCH_OUTBOX_MAX_ATTEMPTS', + 5, + ); + } + + /** + * Drain pending outbox rows to Elasticsearch. + * Idempotent per outbox id. Retries with attempt counter; dead-letters after max. + */ + async drain(batchSize = 50): Promise<{ + processed: number; + succeeded: number; + failed: number; + deadLetter: number; + }> { + if (this.running) { + return { processed: 0, succeeded: 0, failed: 0, deadLetter: 0 }; + } + this.running = true; + + let processed = 0; + let succeeded = 0; + let failed = 0; + let deadLetter = 0; + + try { + if (!this.elasticsearch.isEnabled()) { + this.logger.debug('ES disabled — skipping outbox relay'); + return { processed, succeeded, failed, deadLetter }; + } + + const pending = await this.outboxService.claimPending(batchSize); + + for (const row of pending) { + processed += 1; + await this.outboxService.markProcessing(row.id); + + try { + if (row.operation === SearchOutboxOperation.INDEX) { + await this.elasticsearch.indexProperty( + row.payload as unknown as PropertySearchDocument, + row.id, + ); + } else if (row.operation === SearchOutboxOperation.DELETE) { + await this.elasticsearch.removeProperty(row.aggregateId, row.id); + } + + await this.outboxService.markDone(row.id); + succeeded += 1; + this.metricsService.recordSearchOutboxRelay('success'); + } catch (error) { + this.logger.warn( + `Outbox ${row.id} attempt ${row.attempts + 1} failed: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + await this.outboxService.markFailedOrRetry( + row.id, + row.attempts, + this.maxAttempts, + ); + const nextAttempts = row.attempts + 1; + if (nextAttempts >= this.maxAttempts) { + deadLetter += 1; + this.metricsService.recordSearchOutboxRelay('dead_letter'); + } else { + failed += 1; + this.metricsService.recordSearchOutboxRelay('retry'); + } + } + } + } finally { + this.running = false; + } + + return { processed, succeeded, failed, deadLetter }; + } +} diff --git a/backend/src/modules/search/search-outbox.service.ts b/backend/src/modules/search/search-outbox.service.ts new file mode 100644 index 00000000..1ea8ef8c --- /dev/null +++ b/backend/src/modules/search/search-outbox.service.ts @@ -0,0 +1,142 @@ +import { Injectable } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { EntityManager, Repository } from 'typeorm'; +import { + SearchOutbox, + SearchOutboxOperation, + SearchOutboxStatus, +} from './entities/search-outbox.entity'; +import { Property } from '../properties/entities/property.entity'; +import { listingStatusToVisibility } from './search-visibility'; +import { PropertySearchDocument } from './elasticsearch.service'; + +export const SEARCH_AGGREGATE_PROPERTY = 'property'; + +@Injectable() +export class SearchOutboxService { + constructor( + @InjectRepository(SearchOutbox) + private readonly outboxRepo: Repository, + ) {} + + /** + * Insert an outbox row inside an existing transaction. + * Must share the same EntityManager/QueryRunner as the source mutation. + */ + async enqueueIndex( + manager: EntityManager, + property: Property, + ): Promise { + const payload = this.toSearchDocument(property); + const row = manager.create(SearchOutbox, { + aggregateType: SEARCH_AGGREGATE_PROPERTY, + aggregateId: property.id, + tenantId: property.ownerId, + operation: SearchOutboxOperation.INDEX, + payload: payload as unknown as Record, + status: SearchOutboxStatus.PENDING, + attempts: 0, + }); + return manager.save(SearchOutbox, row); + } + + async enqueueDelete( + manager: EntityManager, + propertyId: string, + tenantId: string, + ): Promise { + const row = manager.create(SearchOutbox, { + aggregateType: SEARCH_AGGREGATE_PROPERTY, + aggregateId: propertyId, + tenantId, + operation: SearchOutboxOperation.DELETE, + payload: { id: propertyId }, + status: SearchOutboxStatus.PENDING, + attempts: 0, + }); + return manager.save(SearchOutbox, row); + } + + toSearchDocument(property: Property): PropertySearchDocument { + const amenities = (property.amenities ?? []).map((a) => a.name); + return { + id: property.id, + tenant_id: property.ownerId, + visibility: listingStatusToVisibility(property.status), + title: property.title, + description: property.description ?? '', + type: property.type, + city: property.city ?? '', + state: property.state ?? '', + country: property.country ?? '', + price: Number(property.price), + bedrooms: property.bedrooms ?? 0, + bathrooms: property.bathrooms ?? 0, + area: Number(property.area ?? 0), + amenities, + location: { + lat: Number(property.latitude ?? 0), + lon: Number(property.longitude ?? 0), + }, + status: property.status, + checksum: this.computeChecksum(property), + createdAt: property.createdAt?.toISOString?.() ?? new Date().toISOString(), + updatedAt: property.updatedAt?.toISOString?.() ?? new Date().toISOString(), + }; + } + + computeChecksum(property: Property): string { + const raw = [ + property.id, + property.ownerId, + property.status, + property.title, + property.price, + property.updatedAt?.getTime?.() ?? 0, + ].join('|'); + let hash = 0; + for (let i = 0; i < raw.length; i += 1) { + hash = (hash << 5) - hash + raw.charCodeAt(i); + hash |= 0; + } + return `c${Math.abs(hash).toString(16)}`; + } + + async claimPending(limit = 50): Promise { + return this.outboxRepo + .createQueryBuilder('o') + .where('o.status = :status', { status: SearchOutboxStatus.PENDING }) + .orderBy('o.created_at', 'ASC') + .take(limit) + .getMany(); + } + + async markProcessing(id: string): Promise { + await this.outboxRepo.update(id, { + status: SearchOutboxStatus.PROCESSING, + }); + } + + async markDone(id: string): Promise { + await this.outboxRepo.update(id, { + status: SearchOutboxStatus.DONE, + processedAt: new Date(), + }); + } + + async markFailedOrRetry( + id: string, + attempts: number, + maxAttempts: number, + ): Promise { + const nextAttempts = attempts + 1; + await this.outboxRepo.update(id, { + attempts: nextAttempts, + status: + nextAttempts >= maxAttempts + ? SearchOutboxStatus.FAILED + : SearchOutboxStatus.PENDING, + processedAt: nextAttempts >= maxAttempts ? new Date() : null, + }); + } +} diff --git a/backend/src/modules/search/search-reconcile.job.ts b/backend/src/modules/search/search-reconcile.job.ts new file mode 100644 index 00000000..9aa056ae --- /dev/null +++ b/backend/src/modules/search/search-reconcile.job.ts @@ -0,0 +1,122 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { DataSource, Repository } from 'typeorm'; +import { Property } from '../properties/entities/property.entity'; +import { SearchOutboxService } from './search-outbox.service'; +import { ElasticsearchService } from './elasticsearch.service'; +import { MetricsService } from '../monitoring/metrics.service'; +import { AuditService } from '../audit/audit.service'; +import { AuditAction, AuditLevel } from '../audit/entities/audit-log.entity'; +import { listingStatusToVisibility } from './search-visibility'; + +export interface ReconcileResult { + checked: number; + drifted: number; + missing: number; + orphaned: number; + enqueued: number; + deleted: number; +} + +@Injectable() +export class SearchReconcileJob { + private readonly logger = new Logger(SearchReconcileJob.name); + + constructor( + @InjectRepository(Property) + private readonly propertyRepo: Repository, + private readonly outboxService: SearchOutboxService, + private readonly elasticsearch: ElasticsearchService, + private readonly dataSource: DataSource, + private readonly metricsService: MetricsService, + private readonly auditService: AuditService, + ) {} + + /** + * Compare PostgreSQL properties against ES documents. + * Enqueues outbox rows for drifted/missing docs; deletes orphaned ES docs. + */ + async run(): Promise { + const result: ReconcileResult = { + checked: 0, + drifted: 0, + missing: 0, + orphaned: 0, + enqueued: 0, + deleted: 0, + }; + + if (!this.elasticsearch.isEnabled()) { + this.logger.debug('ES disabled — skipping reconcile'); + return result; + } + + const properties = await this.propertyRepo.find({ + relations: ['amenities'], + }); + const pgIds = new Set(properties.map((p) => p.id)); + + for (const property of properties) { + result.checked += 1; + const esDoc = await this.elasticsearch.getDocument(property.id); + const expectedVisibility = listingStatusToVisibility(property.status); + const expectedChecksum = this.outboxService.computeChecksum(property); + + if (!esDoc) { + result.missing += 1; + await this.dataSource.transaction(async (manager) => { + await this.outboxService.enqueueIndex(manager, property); + }); + result.enqueued += 1; + this.metricsService.recordSearchReconcile('missing'); + continue; + } + + const drifted = + esDoc.visibility !== expectedVisibility || + esDoc.checksum !== expectedChecksum || + esDoc.tenant_id !== property.ownerId || + esDoc.updatedAt !== property.updatedAt?.toISOString(); + + if (drifted) { + result.drifted += 1; + await this.dataSource.transaction(async (manager) => { + await this.outboxService.enqueueIndex(manager, property); + }); + result.enqueued += 1; + this.metricsService.recordSearchReconcile('drifted'); + } + } + + const esIds = await this.elasticsearch.scrollAllIds(); + for (const esId of esIds) { + if (!pgIds.has(esId)) { + result.orphaned += 1; + try { + await this.elasticsearch.removeProperty(esId); + result.deleted += 1; + this.metricsService.recordSearchReconcile('orphaned'); + } catch (error) { + this.logger.warn( + `Failed to delete orphaned ES doc ${esId}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + } + } + + await this.auditService.log({ + action: AuditAction.SEARCH_RECONCILE, + entityType: 'search_index', + level: AuditLevel.INFO, + metadata: { ...result }, + }); + + this.logger.log( + `Reconcile complete: checked=${result.checked} drifted=${result.drifted} missing=${result.missing} orphaned=${result.orphaned}`, + ); + + return result; + } +} diff --git a/backend/src/modules/search/search-scope.error.ts b/backend/src/modules/search/search-scope.error.ts new file mode 100644 index 00000000..4c775f95 --- /dev/null +++ b/backend/src/modules/search/search-scope.error.ts @@ -0,0 +1,11 @@ +import { BadRequestException } from '@nestjs/common'; + +export class SearchScopeError extends BadRequestException { + constructor(message = 'Search requires a server-derived tenant scope') { + super({ + code: 'SEARCH_SCOPE_ERROR', + message, + }); + this.name = 'SearchScopeError'; + } +} diff --git a/backend/src/modules/search/search-visibility.ts b/backend/src/modules/search/search-visibility.ts new file mode 100644 index 00000000..52b1bb93 --- /dev/null +++ b/backend/src/modules/search/search-visibility.ts @@ -0,0 +1,43 @@ +import { ListingStatus } from '../properties/entities/property.entity'; + +/** ES document visibility discriminator (maps from ListingStatus). */ +export enum SearchVisibility { + LISTED = 'listed', + UNLISTED = 'unlisted', + DRAFT = 'draft', + RENTED = 'rented', +} + +export function listingStatusToVisibility( + status: ListingStatus, +): SearchVisibility { + switch (status) { + case ListingStatus.PUBLISHED: + return SearchVisibility.LISTED; + case ListingStatus.ARCHIVED: + return SearchVisibility.UNLISTED; + case ListingStatus.DRAFT: + return SearchVisibility.DRAFT; + case ListingStatus.RENTED: + return SearchVisibility.RENTED; + default: + return SearchVisibility.UNLISTED; + } +} + +export function visibilityToListingStatus( + visibility: SearchVisibility, +): ListingStatus { + switch (visibility) { + case SearchVisibility.LISTED: + return ListingStatus.PUBLISHED; + case SearchVisibility.UNLISTED: + return ListingStatus.ARCHIVED; + case SearchVisibility.DRAFT: + return ListingStatus.DRAFT; + case SearchVisibility.RENTED: + return ListingStatus.RENTED; + default: + return ListingStatus.ARCHIVED; + } +} diff --git a/backend/src/modules/search/search.controller.ts b/backend/src/modules/search/search.controller.ts index 9ed218cc..efa7eb4c 100644 --- a/backend/src/modules/search/search.controller.ts +++ b/backend/src/modules/search/search.controller.ts @@ -1,17 +1,92 @@ -import { Controller, Get, Query } from '@nestjs/common'; -import { ApiTags, ApiOperation, ApiQuery } from '@nestjs/swagger'; +import { + Controller, + Get, + Query, + UseGuards, +} from '@nestjs/common'; +import { + ApiTags, + ApiOperation, + ApiQuery, + ApiBearerAuth, +} from '@nestjs/swagger'; import { SearchService, SearchFilters } from './search.service'; import { PropertyType, ListingStatus, } from '../properties/entities/property.entity'; +import { JwtAuthGuard } from '../auth/guards/jwt-auth.guard'; +import { CurrentUser } from '../auth/decorators/current-user.decorator'; +import { User, UserRole } from '../users/entities/user.entity'; +import { + discoveryTenantContext, + landlordTenantContext, +} from './tenant-context'; +import { SearchScopeError } from './search-scope.error'; +import { Public } from '../auth/decorators/public.decorator'; @ApiTags('Search') @Controller('search') export class SearchController { constructor(private readonly searchService: SearchService) {} + /** + * Tenant-scoped discovery search. Tenant identity is derived from the + * authenticated principal — never from query parameters (tenantId/visibility + * overrides are ignored if present). + */ + @Get('listings') + @UseGuards(JwtAuthGuard) + @ApiBearerAuth() + @ApiOperation({ + summary: + 'Tenant-scoped property search (ES). Scope is server-derived from JWT.', + }) + @ApiQuery({ name: 'q', required: false }) + @ApiQuery({ name: 'city', required: false }) + @ApiQuery({ name: 'minPrice', required: false }) + @ApiQuery({ name: 'maxPrice', required: false }) + @ApiQuery({ name: 'bedrooms', required: false }) + @ApiQuery({ name: 'page', required: false }) + @ApiQuery({ name: 'limit', required: false }) + async queryListings( + @CurrentUser() user: User, + @Query('q') query?: string, + @Query('city') city?: string, + @Query('state') state?: string, + @Query('country') country?: string, + @Query('type') type?: string, + @Query('minPrice') minPrice?: string, + @Query('maxPrice') maxPrice?: string, + @Query('bedrooms') bedrooms?: string, + @Query('bathrooms') bathrooms?: string, + @Query('page') page?: string, + @Query('limit') limit?: string, + // Intentionally accepted then discarded — never trusted for scope. + @Query('tenantId') _tenantId?: string, + @Query('visibility') _visibility?: string, + ) { + void _tenantId; + void _visibility; + + const tenant = this.deriveTenantContext(user); + return this.searchService.query(tenant, { + query, + city, + state, + country, + type, + minPrice: minPrice ? parseFloat(minPrice) : undefined, + maxPrice: maxPrice ? parseFloat(maxPrice) : undefined, + bedrooms: bedrooms ? parseInt(bedrooms) : undefined, + bathrooms: bathrooms ? parseInt(bathrooms) : undefined, + page: page ? parseInt(page) : 1, + limit: limit ? Math.min(parseInt(limit), 100) : 20, + }); + } + @Get('properties') + @Public() @ApiOperation({ summary: 'Full-text property search with faceted filtering' }) @ApiQuery({ name: 'q', required: false }) @ApiQuery({ name: 'city', required: false }) @@ -50,7 +125,8 @@ export class SearchController { state, country, type, - status, + // Client-supplied status is ignored for public discovery — always published. + status: ListingStatus.PUBLISHED, minPrice: minPrice ? parseFloat(minPrice) : undefined, maxPrice: maxPrice ? parseFloat(maxPrice) : undefined, bedrooms: bedrooms ? parseInt(bedrooms) : undefined, @@ -63,6 +139,7 @@ export class SearchController { lng: lng ? parseFloat(lng) : undefined, radiusKm: radiusKm ? parseFloat(radiusKm) : undefined, }; + void status; return this.searchService.searchProperties( filters, page ? parseInt(page) : 1, @@ -71,9 +148,24 @@ export class SearchController { } @Get('suggest') + @Public() @ApiOperation({ summary: 'Autocomplete suggestions for search' }) @ApiQuery({ name: 'q', required: true }) async suggest(@Query('q') q: string) { return this.searchService.suggest(q); } + + private deriveTenantContext(user: User) { + if (!user?.id) { + throw new SearchScopeError('Authenticated principal has no tenant id'); + } + if ( + user.role === UserRole.LANDLORD || + user.role === UserRole.ADMIN || + user.role === UserRole.AGENT + ) { + return landlordTenantContext(user.id); + } + return discoveryTenantContext(user.id); + } } diff --git a/backend/src/modules/search/search.module.ts b/backend/src/modules/search/search.module.ts index d33f61e2..9f631155 100644 --- a/backend/src/modules/search/search.module.ts +++ b/backend/src/modules/search/search.module.ts @@ -2,12 +2,35 @@ import { Module } from '@nestjs/common'; import { TypeOrmModule } from '@nestjs/typeorm'; import { SearchService } from './search.service'; import { SearchController } from './search.controller'; +import { ElasticsearchModule } from './elasticsearch.module'; +import { SearchOutbox } from './entities/search-outbox.entity'; +import { SearchOutboxService } from './search-outbox.service'; +import { SearchOutboxRelay } from './search-outbox.relay'; +import { SearchReconcileJob } from './search-reconcile.job'; import { Property } from '../properties/entities/property.entity'; +import { MonitoringModule } from '../monitoring/monitoring.module'; +import { AuditModule } from '../audit/audit.module'; @Module({ - imports: [TypeOrmModule.forFeature([Property])], - providers: [SearchService], + imports: [ + TypeOrmModule.forFeature([Property, SearchOutbox]), + ElasticsearchModule, + MonitoringModule, + AuditModule, + ], + providers: [ + SearchService, + SearchOutboxService, + SearchOutboxRelay, + SearchReconcileJob, + ], controllers: [SearchController], - exports: [SearchService], + exports: [ + SearchService, + SearchOutboxService, + SearchOutboxRelay, + SearchReconcileJob, + ElasticsearchModule, + ], }) export class SearchModule {} diff --git a/backend/src/modules/search/search.service.ts b/backend/src/modules/search/search.service.ts index 0674e979..91db20a3 100644 --- a/backend/src/modules/search/search.service.ts +++ b/backend/src/modules/search/search.service.ts @@ -13,6 +13,15 @@ import { TTL_SEARCH_RESULTS_MS, TTL_SUGGEST_MS, } from '../../common/cache/cache.constants'; +import { + ElasticsearchService, + EsSearchFilters, + EsSearchResult, + PropertySearchDocument, +} from './elasticsearch.service'; +import { TenantContext } from './tenant-context'; +import { SearchScopeError } from './search-scope.error'; +import { visibilityToListingStatus } from './search-visibility'; export interface SearchFilters { query?: string; @@ -64,8 +73,119 @@ export class SearchService { @InjectRepository(Property) private readonly propertyRepo: Repository, private readonly cacheService: CacheService, + private readonly elasticsearch: ElasticsearchService, ) {} + /** + * Tenant-scoped Elasticsearch query. + * Invariant: every ES body includes non-removable tenant_id + visibility filters. + * Throws SearchScopeError when tenant context is missing. + */ + async query( + tenant: TenantContext | null | undefined, + filters: EsSearchFilters = {}, + ): Promise> { + if (!tenant?.tenantId) { + throw new SearchScopeError(); + } + if (!tenant.allowedVisibilities?.length) { + throw new SearchScopeError('visibility scope is required for search'); + } + + // Prefer ES when available; fall back to scoped Postgres. + if (this.elasticsearch.isEnabled()) { + return this.elasticsearch.search(tenant, filters); + } + + return this.queryPostgresFallback(tenant, filters); + } + + private async queryPostgresFallback( + tenant: TenantContext, + filters: EsSearchFilters, + ): Promise> { + const page = filters.page || 1; + const limit = filters.limit || 20; + const allowedStatuses = tenant.allowedVisibilities.map( + visibilityToListingStatus, + ); + + const qb = this.propertyRepo + .createQueryBuilder('property') + .leftJoinAndSelect('property.amenities', 'amenities') + .where('property.owner_id = :tenantId', { tenantId: tenant.tenantId }) + .andWhere('property.status IN (:...statuses)', { + statuses: allowedStatuses, + }); + + if (filters.query) { + qb.andWhere( + `(to_tsvector('english', property.title || ' ' || COALESCE(property.description, '')) @@ plainto_tsquery('english', :query) OR property.address ILIKE :likeQuery)`, + { query: filters.query, likeQuery: `%${filters.query}%` }, + ); + } + if (filters.city) { + qb.andWhere('property.city ILIKE :city', { city: `%${filters.city}%` }); + } + if (filters.minPrice !== undefined) { + qb.andWhere('property.price >= :minPrice', { + minPrice: filters.minPrice, + }); + } + if (filters.maxPrice !== undefined) { + qb.andWhere('property.price <= :maxPrice', { + maxPrice: filters.maxPrice, + }); + } + if (filters.bedrooms !== undefined) { + qb.andWhere('property.bedrooms >= :bedrooms', { + bedrooms: filters.bedrooms, + }); + } + + const searchBody = this.elasticsearch.buildScopedSearchBody( + tenant, + filters, + ); + + const [items, total] = await qb + .skip((page - 1) * limit) + .take(limit) + .orderBy('property.createdAt', 'DESC') + .getManyAndCount(); + + return { + hits: items.map((p) => ({ + id: p.id, + tenant_id: p.ownerId, + visibility: tenant.allowedVisibilities[0], + title: p.title, + description: p.description ?? '', + type: p.type, + city: p.city ?? '', + state: p.state ?? '', + country: p.country ?? '', + price: Number(p.price), + bedrooms: p.bedrooms ?? 0, + bathrooms: p.bathrooms ?? 0, + area: Number(p.area ?? 0), + amenities: (p.amenities ?? []).map((a) => a.name), + location: { + lat: Number(p.latitude ?? 0), + lon: Number(p.longitude ?? 0), + }, + status: p.status, + createdAt: p.createdAt?.toISOString?.() ?? '', + updatedAt: p.updatedAt?.toISOString?.() ?? '', + })), + total, + page, + limit, + facets: {}, + searchBody, + }; + } + async searchProperties( filters: SearchFilters, page = 1, @@ -152,7 +272,6 @@ export class SearchService { .leftJoinAndSelect('property.images', 'images') .leftJoinAndSelect('property.amenities', 'amenities'); - // Full-text search if (filters.query) { qb.andWhere( `(to_tsvector('english', property.title || ' ' || COALESCE(property.description, '')) @@ plainto_tsquery('english', :query) OR property.address ILIKE :likeQuery)`, diff --git a/backend/src/modules/search/tenant-context.ts b/backend/src/modules/search/tenant-context.ts new file mode 100644 index 00000000..969f41b7 --- /dev/null +++ b/backend/src/modules/search/tenant-context.ts @@ -0,0 +1,30 @@ +import { SearchVisibility } from './search-visibility'; + +/** + * Server-derived search scope. Never construct from request query/body params. + */ +export interface TenantContext { + /** Landlord/org tenant identity from the authenticated principal. */ + tenantId: string; + /** Visibility values the caller is allowed to see. */ + allowedVisibilities: SearchVisibility[]; +} + +export function discoveryTenantContext(tenantId: string): TenantContext { + return { + tenantId, + allowedVisibilities: [SearchVisibility.LISTED], + }; +} + +export function landlordTenantContext(tenantId: string): TenantContext { + return { + tenantId, + allowedVisibilities: [ + SearchVisibility.LISTED, + SearchVisibility.UNLISTED, + SearchVisibility.DRAFT, + SearchVisibility.RENTED, + ], + }; +} diff --git a/contract/bindings/property_registry/events.ts b/contract/bindings/property_registry/events.ts new file mode 100644 index 00000000..951ea28a --- /dev/null +++ b/contract/bindings/property_registry/events.ts @@ -0,0 +1,41 @@ +/** + * Hand-maintained TypeScript bindings for property_registry contract events. + * Mirrors contract/contracts/property_registry/src/events.rs. + * Regenerated when event payloads change; keep in sync with check-all.sh builds. + */ + +export type PropertyVisibility = 'listed' | 'unlisted' | 'draft' | 'rented'; + +export interface PropertyRegisteredEvent { + topic: 'prop_reg'; + landlord: string; + property_id: string; + metadata_hash: string; +} + +/** Discovery visibility signal consumed by the backend search indexer. */ +export interface PropertyListedEvent { + topic: 'prop_listed'; + landlord: string; + property_id: string; + visibility: PropertyVisibility; +} + +export interface PropertyUnlistedEvent { + topic: 'prop_unlisted'; + landlord: string; + property_id: string; + visibility: PropertyVisibility; +} + +export interface PropertyVerifiedEvent { + topic: 'prop_ver'; + admin: string; + property_id: string; +} + +export type PropertyRegistryEvent = + | PropertyRegisteredEvent + | PropertyListedEvent + | PropertyUnlistedEvent + | PropertyVerifiedEvent; diff --git a/contract/contracts/property_registry/src/events.rs b/contract/contracts/property_registry/src/events.rs index 6b1c0601..cd1f680f 100644 --- a/contract/contracts/property_registry/src/events.rs +++ b/contract/contracts/property_registry/src/events.rs @@ -19,6 +19,30 @@ pub struct PropertyRegistered { pub metadata_hash: String, } +/// Event emitted when a property is listed for discovery. +/// Topics: ["prop_listed", landlord: Address, property_id: String] +/// `visibility` is the discovery discriminator consumed by the backend indexer +/// (`listed` | `unlisted` | `draft` | `rented`). +#[contractevent(topics = ["prop_listed"])] +pub struct PropertyListed { + #[topic] + pub landlord: Address, + #[topic] + pub property_id: String, + pub visibility: String, +} + +/// Event emitted when a property is withdrawn from discovery. +/// Topics: ["prop_unlisted", landlord: Address, property_id: String] +#[contractevent(topics = ["prop_unlisted"])] +pub struct PropertyUnlisted { + #[topic] + pub landlord: Address, + #[topic] + pub property_id: String, + pub visibility: String, +} + /// Event emitted when a property is verified /// Topics: ["prop_ver", admin: Address, property_id: String] #[contractevent(topics = ["prop_ver"])] @@ -76,6 +100,36 @@ pub(crate) fn property_registered( .publish(env); } +/// Helper function to emit property listed event (visibility discriminator for indexer) +pub(crate) fn property_listed( + env: &Env, + property_id: String, + landlord: Address, + visibility: String, +) { + PropertyListed { + landlord, + property_id, + visibility, + } + .publish(env); +} + +/// Helper function to emit property unlisted event +pub(crate) fn property_unlisted( + env: &Env, + property_id: String, + landlord: Address, + visibility: String, +) { + PropertyUnlisted { + landlord, + property_id, + visibility, + } + .publish(env); +} + /// Helper function to emit property verified event pub(crate) fn property_verified(env: &Env, property_id: String, admin: Address) { PropertyVerified { admin, property_id }.publish(env); diff --git a/contract/contracts/property_registry/src/property.rs b/contract/contracts/property_registry/src/property.rs index 08f6e18e..fdcd5dd3 100644 --- a/contract/contracts/property_registry/src/property.rs +++ b/contract/contracts/property_registry/src/property.rs @@ -64,7 +64,19 @@ pub fn register_property( .persistent() .extend_ttl(&count_key, 500000, 500000); - events::property_registered(env, property_id, landlord, metadata_hash); + events::property_registered( + env, + property_id.clone(), + landlord.clone(), + metadata_hash, + ); + // Upstream visibility signal for the backend search indexer. + events::property_listed( + env, + property_id, + landlord, + String::from_str(env, "listed"), + ); Ok(()) } diff --git a/frontend/.env.example b/frontend/.env.example index 1291d9a9..f9c24488 100644 --- a/frontend/.env.example +++ b/frontend/.env.example @@ -10,5 +10,9 @@ NEXT_PUBLIC_HORIZON_URL=https://horizon-testnet.stellar.org # REQUIRED for signRentPayment to work. Set to your deployed contract ID. NEXT_PUBLIC_RENTAL_CONTRACT_ID= +# Server-only backend API URL for BFF routes (app/api/*). +# Never put tenant identity in NEXT_PUBLIC_* variables. +BACKEND_API_URL=http://localhost:3000 + # KYC enforcement gate. Set to "true" in production. NEXT_PUBLIC_KYC_ENFORCED=true diff --git a/frontend/app/api/search/route.ts b/frontend/app/api/search/route.ts new file mode 100644 index 00000000..e33e128a --- /dev/null +++ b/frontend/app/api/search/route.ts @@ -0,0 +1,100 @@ +import { cookies } from "next/headers"; +import { NextRequest, NextResponse } from "next/server"; +import type { SearchListingsResult } from "@/lib/mock"; +import { fetchSearchListings } from "@/lib/mock"; + +export const dynamic = "force-dynamic"; + +const BACKEND_URL = + process.env.BACKEND_API_URL || process.env.API_URL || "http://localhost:3000"; + +/** + * BFF search route. + * Tenant identity is derived from the server-side session/JWT cookie. + * Client-supplied tenantId / visibility overrides are dropped. + */ +export async function GET(request: NextRequest) { + const { searchParams } = request.nextUrl; + + // Drop client scope overrides — never forward to the backend. + const userFacing = { + q: searchParams.get("q") ?? undefined, + city: searchParams.get("city") ?? undefined, + minPrice: searchParams.get("minPrice") ?? undefined, + maxPrice: searchParams.get("maxPrice") ?? undefined, + bedrooms: searchParams.get("bedrooms") ?? undefined, + page: searchParams.get("page") ?? undefined, + limit: searchParams.get("limit") ?? undefined, + }; + + const cookieStore = await cookies(); + const accessToken = + cookieStore.get("access_token")?.value || + cookieStore.get("token")?.value || + request.headers.get("authorization")?.replace(/^Bearer\s+/i, ""); + + if (!accessToken) { + // Dev/mock fallback when no session — still never trusts client tenantId. + const mock = await fetchSearchListings({ + q: userFacing.q, + city: userFacing.city, + minPrice: userFacing.minPrice + ? parseFloat(userFacing.minPrice) + : undefined, + maxPrice: userFacing.maxPrice + ? parseFloat(userFacing.maxPrice) + : undefined, + bedrooms: userFacing.bedrooms + ? parseInt(userFacing.bedrooms, 10) + : undefined, + }); + return NextResponse.json(mock); + } + + const upstream = new URLSearchParams(); + for (const [key, value] of Object.entries(userFacing)) { + if (value !== undefined && value !== "") { + upstream.set(key, value); + } + } + + try { + const response = await fetch( + `${BACKEND_URL}/search/listings?${upstream.toString()}`, + { + headers: { + Authorization: `Bearer ${accessToken}`, + Accept: "application/json", + }, + cache: "no-store", + }, + ); + + if (!response.ok) { + const text = await response.text(); + return NextResponse.json( + { error: "Search upstream failed", detail: text }, + { status: response.status }, + ); + } + + const data = (await response.json()) as SearchListingsResult; + return NextResponse.json(data); + } catch { + // Backend unreachable — mock with listed-only results. + const mock = await fetchSearchListings({ + q: userFacing.q, + city: userFacing.city, + minPrice: userFacing.minPrice + ? parseFloat(userFacing.minPrice) + : undefined, + maxPrice: userFacing.maxPrice + ? parseFloat(userFacing.maxPrice) + : undefined, + bedrooms: userFacing.bedrooms + ? parseInt(userFacing.bedrooms, 10) + : undefined, + }); + return NextResponse.json(mock); + } +} diff --git a/frontend/lib/hooks/useSearchListings.ts b/frontend/lib/hooks/useSearchListings.ts new file mode 100644 index 00000000..7acd57f3 --- /dev/null +++ b/frontend/lib/hooks/useSearchListings.ts @@ -0,0 +1,44 @@ +"use client"; + +import { useQuery } from "@tanstack/react-query"; +import type { SearchListingsResult } from "@/lib/mock"; + +export interface SearchListingsFilters { + q?: string; + city?: string; + minPrice?: number; + maxPrice?: number; + bedrooms?: number; + page?: number; + limit?: number; +} + +/** + * Forwards only user-facing filters. Never sends tenantId or visibility — + * those are derived server-side in the BFF from the session. + */ +export function useSearchListings(filters: SearchListingsFilters = {}) { + return useQuery({ + queryKey: ["search-listings", filters], + queryFn: async (): Promise => { + const params = new URLSearchParams(); + if (filters.q) params.set("q", filters.q); + if (filters.city) params.set("city", filters.city); + if (filters.minPrice !== undefined) + params.set("minPrice", String(filters.minPrice)); + if (filters.maxPrice !== undefined) + params.set("maxPrice", String(filters.maxPrice)); + if (filters.bedrooms !== undefined) + params.set("bedrooms", String(filters.bedrooms)); + if (filters.page !== undefined) params.set("page", String(filters.page)); + if (filters.limit !== undefined) + params.set("limit", String(filters.limit)); + + const response = await fetch(`/api/search?${params.toString()}`); + if (!response.ok) { + throw new Error(`Search failed: ${response.status}`); + } + return response.json() as Promise; + }, + }); +} diff --git a/frontend/lib/mock.ts b/frontend/lib/mock.ts index 641d2117..84147a8e 100644 --- a/frontend/lib/mock.ts +++ b/frontend/lib/mock.ts @@ -1,3 +1,5 @@ +export type ListingVisibility = "listed" | "unlisted" | "draft" | "rented"; + export interface Property { id: string; title: string; @@ -5,7 +7,30 @@ export interface Property { rentPerMonth: number; deposit: number; leaseMonths: number; + /** Lease/dispute status for dashboard surfaces. */ status: "active" | "disputed"; + /** Discovery visibility — present on both mock and live search results. */ + visibility: ListingVisibility; +} + +export interface SearchListing { + id: string; + title: string; + city: string; + state?: string; + country?: string; + price: number; + bedrooms: number; + bathrooms: number; + visibility: ListingVisibility; + tenant_id?: string; +} + +export interface SearchListingsResult { + hits: SearchListing[]; + total: number; + page: number; + limit: number; } export type KycStatus = "UNVERIFIED" | "PENDING" | "VERIFIED" | "REJECTED"; @@ -40,6 +65,7 @@ export const mockProperties: Property[] = [ deposit: 640, leaseMonths: 12, status: "active", + visibility: "listed", }, { id: "p2", @@ -49,6 +75,7 @@ export const mockProperties: Property[] = [ deposit: 360, leaseMonths: 6, status: "disputed", + visibility: "listed", }, { id: "p3", @@ -58,6 +85,7 @@ export const mockProperties: Property[] = [ deposit: 1080, leaseMonths: 12, status: "active", + visibility: "listed", }, ]; @@ -115,6 +143,44 @@ export async function fetchPayments(): Promise { return simulateFetch([...mockPayments]); } +/** Mock scoped search — only listed visibility is returned. */ +export async function fetchSearchListings(filters: { + q?: string; + city?: string; + minPrice?: number; + maxPrice?: number; + bedrooms?: number; +}): Promise { + let hits: SearchListing[] = mockProperties + .filter((p) => p.visibility === "listed") + .map((p) => ({ + id: p.id, + title: p.title, + city: p.location.split(",")[0]?.trim() ?? p.location, + price: p.rentPerMonth, + bedrooms: 2, + bathrooms: 1, + visibility: p.visibility, + })); + + if (filters.q) { + const q = filters.q.toLowerCase(); + hits = hits.filter((h) => h.title.toLowerCase().includes(q)); + } + if (filters.city) { + const city = filters.city.toLowerCase(); + hits = hits.filter((h) => h.city.toLowerCase().includes(city)); + } + if (filters.minPrice !== undefined) { + hits = hits.filter((h) => h.price >= filters.minPrice!); + } + if (filters.maxPrice !== undefined) { + hits = hits.filter((h) => h.price <= filters.maxPrice!); + } + + return simulateFetch({ hits, total: hits.length, page: 1, limit: 20 }); +} + export async function fetchUserProfile(): Promise { return simulateFetch({ ...mockUserProfile }); }