MonLogement Ads Importer
A queue-driven pipeline that ingests real-estate listings from agency FTP drops (ZIP + pipe-delimited CSV), validates and geocodes them, processes their photos, and keeps a search index in sync.
Role: Backend Engineer
Overview
Real-estate agencies publish listings in a vendor feed format: a ZIP (Annonces.csv, Photos.cfg, optional images folder) and an Agences.csv uploaded over FTP. This asynchronous, fault-tolerant pipeline turns those feeds into live, searchable listings without manual work. It is the import service behind the MonLogement platform.
Tech Stack
backend
queue
search
storage
data
libraries
external Apis
testing
The Problem
Real-estate agencies publish their listings in a vendor feed format: they upload a ZIP (Annonces.csv, Photos.cfg, optional images folder) and an Agences.csv over FTP. The platform needs these listings live and searchable without manual work. Feeds are large, messy and inconsistent: mixed encodings, a custom '!#' delimiter, 300+ columns, free-text localities that don't match official names, missing or wrong coordinates, inconsistent floor data, and photos delivered in several different ways. Listings also disappear from feeds and must be removed.
- Processing very large feeds without exhausting the queue or memory
- Two imports for the same agency running at the same time
- Removing listings that vanished from a feed without a delete flag
- Unreliable coordinates and no geocoder budget
- Keeping search consistent while writing many records
- Dirty input: mixed encodings and a non-standard delimiter
The Solution
- 1An asynchronous, fault-tolerant pipeline split into small single-purpose jobs on dedicated queues. Each stage validates its input, logs failures without blocking the rest of the feed, and hands off to the next stage.
Multi-mode photo ingestion
Supports four photo delivery modes declared in Photos.cfg: FULL (images shipped in the ZIP), DIFF (timestamped file names so only changed photos are re-uploaded), URL and AR (remote download with timeout).
Image optimisation
MIME and 40 MB size validation, scale-down to 1440px, JPEG re-encode at quality 80, and a 341x256 thumbnail generated from the already-resized image. Temporary files are always cleaned in a finally block.
Fuzzy locality matching
Normalises free-text city names (digit/letter spacing, zero-padded arrondissements, œ encoding issue), searches a Meilisearch index filtered by postal code, and falls back to SQL matching with saint/sainte to st/ste alias handling.
Geocoding and coordinate sanity checks
Supplied coordinates are verified against the official commune boundary (geo.api.gouv.fr, with special handling for Paris/Lyon/Marseille arrondissements). Bad or missing ones are replaced by a random point inside the commune polygon (ray-casting; largest polygon for MultiPolygons) with a small jitter, and flagged as auto-filled.
Domain mapping
Maps agency vocabulary to internal types (e.g. 'maison-villa' to house, 'vente de prestige' to sell), converts kWh and GES values to A-G energy and gas classes, derives facilities (lift, pool, cellar, parking count...), and resolves French administrative levels from postal codes including Corsica 2A/2B and overseas departments.
Data-quality rules
Rows are skipped with a precise log line when postal code, locality, reference, price or agency is invalid; inconsistent floor / total-floor pairs are nulled; construction year is range-checked; HTML descriptions are sanitised on write with Purifier.
Error observability
Row and image failures are persisted to an AdsImportingErrorLog table (type, message, stack trace, file, line, batch id, agency and technical id context) and pruned by a daily command.
Agency hierarchy support
Listings are attributed to the child agency named in the row, falling back to the parent feed owner, and the user account falls back along the same chain. Cleanup covers both the agency and its sub-agencies.
Architecture
Event-driven job pipeline with scheduled polling
- 1Scan
- 2Agencies sync
- 3Unpack & validate
- 4Self-scheduling batches
- 5Row processing
- 6Image processing
- 7Reconciliation cleanup
Components
Scan
ScanJob (scheduled every 5 min)Walks the FTP incoming directory, maps each folder to an agency via its ftp_username, and chains an agencies import and a ZIP processing job. Capped at 10 files per run and unique-locked to avoid overlapping runs.
Agencies sync
AgenciesImportJob → AgenciesImportChunkJobCleans the Agences.csv, streams it with Spatie Simple Excel and fans it out as a batch of 1000-row chunk jobs.
Unpack & validate
CompressedFileProcessingJobArchives the ZIP under a dated processed folder, extracts it to a timestamped directory, validates required files, reads the photo mode from Photos.cfg, and normalises the CSV (encoding detection, unsupported chars, custom delimiter).
Self-scheduling batches
ProcessCsvBatchJobCreates batches of 5 chunk jobs x 500 rows, then re-dispatches itself 30s later until the file is exhausted, so the queue is never flooded. The last iteration schedules cleanup.
Row processing
ProcessCsvJobPer row: validates, maps types, resolves region/department from postal code, matches the locality, resolves coordinates, then upserts the listing and its facilities inside a DB transaction and indexes it.
Image processing
ProcessImageJob (images-processing queue)Resizes, recompresses and thumbnails up to 10 photos per listing and uploads them to object storage.
Reconciliation cleanup
CleanupJob → CleanupImagesJobDeletes listings no longer present in the feed (with their images and index entries) and resets the import flag for the next run.
Queues
Scheduled Maintenance
- ScanJob every 5 minutes
- Cleanup of extracted directories older than a week (daily)
- Cleanup of old processed files (daily)
- Pruning of import error logs (daily)
- Orphaned object-storage file cleanup command
Technical Challenges
Processing very large feeds without exhausting the queue or memory
Solution: Streamed CSV reading with fgetcsv, 500-row chunk jobs grouped into small batches, and a self-rescheduling batch dispatcher with a 30s delay instead of enqueueing everything at once.
Outcome: Predictable memory use and back-pressure regardless of file size; failures are isolated to a chunk (allowFailures).
Two imports for the same agency running at the same time
Solution: ShouldBeUnique on ScanJob and on the ZIP job (keyed by agency UUID, 1h lock), WithoutOverlapping middleware keyed by agency on chunk jobs, and chained jobs to preserve agencies-before-listings order.
Outcome: No duplicate or interleaved writes per agency.
Removing listings that vanished from a feed without a delete flag
Solution: Mark-and-sweep: every upsert sets is_imported = true; the final cleanup job deletes records still false for that agency tree (chunks of 1000, images deleted from storage in batches of 100 via dedicated jobs, DB rows and search index removed in a transaction), then resets the flag.
Outcome: Catalogue always mirrors the latest feed, and storage does not accumulate orphaned photos.
Unreliable coordinates and no geocoder budget
Solution: Validate against official boundaries, otherwise generate a point inside the polygon so map pins never land outside the right town.
Outcome: All listings are map-ready and consistently placed; approximate points are marked with coordinates_auto_fill.
Keeping search consistent while writing many records
Solution: Wrap writes in withoutSyncingToSearch, then trigger a single searchable() call after the listing and its facilities are saved.
Outcome: One index update per listing instead of several.
Dirty input: mixed encodings and a non-standard delimiter
Solution: Line-by-line encoding detection (UTF-8 / ISO-8859-1 / Windows-1252), conversion to UTF-8, stripping of invalid bytes, and delimiter rewriting to a single character before parsing.
Outcome: Accents and special characters survive end to end.
Engineering Practices
- Single-responsibility jobs with explicit constructors and typed parameters
- Dedicated queues per workload type so heavy image work cannot starve CSV processing
- Idempotent upserts keyed on the agency technical id
- DB transactions around listing writes
- External HTTP calls with timeouts and retries
- Scheduled housekeeping commands for disk and log growth
- Configuration-driven city code mapping (composite cities enum + config)