/* * Copyright (c) 2026 by Christian Kellner. * Licensed under Apache-2.0 with Commons Clause and Attribution/Naming Clause */ import { NoNewListingsWarning } from './errors.js'; import { storeListings, getKnownListingHashesForJobAndProvider, deleteListingsById, } from './services/storage/listingsStorage.js'; import { getJob } from './services/storage/jobStorage.js'; import * as notify from './notification/notify.js'; import Extractor from './services/extractor/extractor.js'; import urlModifier from './services/queryStringMutator.js'; import logger from './services/logger.js'; import { geocodeAddress } from './services/geocoding/geoCodingService.js'; import { distanceMeters } from './services/listings/distanceCalculator.js'; import { getUserSettings } from './services/storage/settingsStorage.js'; import { updateListingDistance } from './services/storage/listingsStorage.js'; import booleanPointInPolygon from '@turf/boolean-point-in-polygon'; /** * @typedef {Object} Listing * @property {string} id Stable unique identifier (hash) of the listing. * @property {string} title Title or headline of the listing. * @property {string} [address] Optional address/location text. * @property {string} [price] Optional price text/value. * @property {string} [url] Link to the listing detail page. * @property {any} [meta] Provider-specific additional metadata. */ /** * @typedef {Object} SimilarityCache * @property {(title:string, address?:string)=>boolean} hasSimilarEntries Returns true if a similar entry is known. * @property {(title:string, address?:string)=>void} addCacheEntry Adds a new entry to the similarity cache. */ /** * Runtime orchestrator for fetching, normalizing, filtering, deduplicating, storing, * and notifying about new listings from a configured provider. * * The execution flow is: * 1) Prepare provider URL (sorting, etc.) * 2) Extract raw listings from the provider * 3) Normalize listings to the provider schema * 4) Filter out incomplete/blacklisted listings * 5) Identify new listings (vs. previously stored hashes) * 6) Persist new listings * 7) Filter out entries similar to already seen ones * 8) Dispatch notifications */ class FredyPipelineExecutioner { /** * Create a new runtime instance for a single provider/job execution. * * @param {Object} providerConfig Provider configuration. * @param {string} providerConfig.url Base URL to crawl. * @param {string} [providerConfig.sortByDateParam] Query parameter used to enforce sorting by date (provider-specific). * @param {string} [providerConfig.waitForSelector] CSS selector to wait for before parsing content. * @param {Object.} providerConfig.crawlFields Mapping of field names to selectors/paths to extract. * @param {string} providerConfig.crawlContainer CSS selector for the container holding listing items. * @param {(raw:any)=>Listing} providerConfig.normalize Function to convert raw scraped data into a Listing shape. * @param {(listing:Listing)=>boolean} providerConfig.filter Function to filter out unwanted listings. * @param {(url:string, waitForSelector?:string)=>Promise|Promise} [providerConfig.getListings] Optional override to fetch listings. * @param {(listing:Listing, browser:any)=>Promise} [providerConfig.fetchDetails] Optional per-listing detail enrichment. Called in parallel for each new listing after deduplication. Receives the shared browser instance. Must always resolve (never reject). * @param {Object} notificationConfig Notification configuration passed to notification adapters. * @param {Object} spatialFilter Optional spatial filter configuration. * @param {string} providerId The ID of the provider currently in use. * @param {string} jobKey Key of the job that is currently running (from within the config). * @param {SimilarityCache} similarityCache Cache instance for checking similar entries. * @param browser */ constructor(providerConfig, notificationConfig, spatialFilter, providerId, jobKey, similarityCache, browser) { this._providerConfig = providerConfig; this._notificationConfig = notificationConfig; this._spatialFilter = spatialFilter; this._providerId = providerId; this._jobKey = jobKey; this._similarityCache = similarityCache; this._browser = browser; } /** * Execute the end-to-end pipeline for a single provider run. * * @returns {Promise} Resolves to the list of new (and similarity-filtered) listings * after notifications have been sent; resolves to void when there are no new listings. */ execute() { return Promise.resolve(urlModifier(this._providerConfig.url, this._providerConfig.sortByDateParam)) .then(this._providerConfig.getListings?.bind(this) ?? this._getListings.bind(this)) .then(this._normalize.bind(this)) .then(this._filter.bind(this)) .then(this._findNew.bind(this)) .then(this._fetchDetails.bind(this)) .then(this._geocode.bind(this)) .then(this._save.bind(this)) .then(this._calculateDistance.bind(this)) .then(this._filterBySimilarListings.bind(this)) .then(this._filterByArea.bind(this)) .then(this._notify.bind(this)) .catch(this._handleError.bind(this)); } /** * Optionally enrich new listings with data from their detail pages. * Only called when the provider config defines a `fetchDetails` function. * Runs all fetches in parallel. Each individual fetch must handle its own errors * and always resolve (never reject) to avoid aborting other listings. * * @param {Listing[]} newListings New listings to enrich. * @returns {Promise} Resolves with enriched listings. */ async _fetchDetails(newListings) { if (typeof this._providerConfig.fetchDetails !== 'function') { return newListings; } const userId = getJob(this._jobKey)?.userId; const enabledProviders = getUserSettings(userId)?.provider_details ?? []; if (!userId || !Array.isArray(enabledProviders) || !enabledProviders.includes(this._providerId)) { return newListings; } const listingsToEnrich = process.env.NODE_ENV === 'test' ? newListings.slice(0, 1) : newListings; const enriched = []; for (const listing of listingsToEnrich) { enriched.push(await this._providerConfig.fetchDetails(listing, this._browser)); } return enriched; } /** * Geocode new listings. * * @param {Listing[]} newListings New listings to geocode. * @returns {Promise} Resolves with the listings (potentially with added coordinates). */ async _geocode(newListings) { for (const listing of newListings) { if (listing.address) { const coords = await geocodeAddress(listing.address); if (coords) { listing.latitude = coords.lat; listing.longitude = coords.lng; } } } return newListings; } /** * Filter listings by area using the provider's area filter if available. * Only filters if areaFilter is set on the provider AND the listing has coordinates. * * @param {Listing[]} newListings New listings to filter by area. * @returns {Promise} Resolves with listings that are within the area (or not filtered if no area is set). */ _filterByArea(newListings) { const polygonFeatures = this._spatialFilter?.features?.filter((f) => f.geometry?.type === 'Polygon'); // If no area filter is set, return all listings if (!polygonFeatures?.length) { return newListings; } const filteredIds = []; // Filter listings by area - keep only those within the polygon const keptListings = newListings.filter((listing) => { // If listing doesn't have coordinates, keep it (don't filter out) if (listing.latitude == null || listing.longitude == null) { return true; } // Check if the point is inside the polygons const point = [listing.longitude, listing.latitude]; // GeoJSON format: [lon, lat] const isInPolygon = polygonFeatures.some((feature) => booleanPointInPolygon(point, feature)); if (!isInPolygon) { filteredIds.push(listing.id); } return isInPolygon; }); if (filteredIds.length > 0) { deleteListingsById(filteredIds); } return keptListings; } /** * Fetch listings from the provider, using the default Extractor flow unless * a provider-specific getListings override is supplied. * * @param {string} url The provider URL to fetch from. * @returns {Promise} Resolves with an array of listings (empty when none found). */ _getListings(url) { const extractor = new Extractor({ ...this._providerConfig.puppeteerOptions, browser: this._browser }); return new Promise((resolve, reject) => { extractor .execute(url, this._providerConfig.waitForSelector) .then(() => { const listings = extractor.parseResponseText( this._providerConfig.crawlContainer, this._providerConfig.crawlFields, url, ); resolve(listings == null ? [] : listings); }) .catch((err) => { reject(err); logger.error(err); }); }); } /** * Normalize raw listings into the provider-specific Listing shape. * * @param {any[]} listings Raw listing entries from the extractor or override. * @returns {Listing[]} Normalized listings. */ _normalize(listings) { return listings.map(this._providerConfig.normalize); } /** * Filter out listings that are missing required fields and those rejected by the * provider's blacklist/filter function. * * @param {Listing[]} listings Listings to filter. * @returns {Listing[]} Filtered listings that pass validation and provider filter. */ _filter(listings) { const keys = Object.keys(this._providerConfig.crawlFields); const filteredListings = listings.filter((item) => keys.every((key) => key in item)); return filteredListings.filter(this._providerConfig.filter); } /** * Determine which listings are new by comparing their IDs against stored hashes. * * @param {Listing[]} listings Listings to evaluate for novelty. * @returns {Listing[]} New listings not seen before. * @throws {NoNewListingsWarning} When no new listings are found. */ _findNew(listings) { logger.debug(`Checking ${listings.length} listings for new entries (Provider: '${this._providerId}')`); const hashes = getKnownListingHashesForJobAndProvider(this._jobKey, this._providerId) || []; const newListings = listings.filter((o) => !hashes.includes(o.id)); if (newListings.length === 0) { throw new NoNewListingsWarning(); } return newListings; } /** * Send notifications for new listings using the configured notification adapter(s). * * @param {Listing[]} newListings New listings to notify about. * @returns {Promise} Resolves to the provided listings after notifications complete. * @throws {NoNewListingsWarning} When there are no listings to notify about. */ _notify(newListings) { if (newListings.length === 0) { throw new NoNewListingsWarning(); } const sendNotifications = notify.send(this._providerId, newListings, this._notificationConfig, this._jobKey); return Promise.all(sendNotifications).then(() => newListings); } /** * Persist new listings and pass them through. * * @param {Listing[]} newListings Listings to store. * @returns {Listing[]} The same listings, unchanged. */ _save(newListings) { logger.debug(`Storing ${newListings.length} new listings (Provider: '${this._providerId}')`); storeListings(this._jobKey, this._providerId, newListings); return newListings; } /** * Calculate distance for new listings. * * @param {Listing[]} listings * @returns {Listing[]} * @private */ _calculateDistance(listings) { if (listings.length === 0) return []; const job = getJob(this._jobKey); const userId = job?.userId; if (userId == null || typeof userId !== 'string') { logger.debug('Skipping distance calculation: userId is missing or invalid'); return listings; } const userSettings = getUserSettings(userId); const homeAddress = userSettings?.home_address; if (!homeAddress || !homeAddress.coords) { return listings; } const { lat, lng } = homeAddress.coords; for (const listing of listings) { if (listing.latitude != null && listing.longitude != null) { const dist = distanceMeters(lat, lng, listing.latitude, listing.longitude); updateListingDistance(listing.id, dist); listing.distance_to_destination = dist; } } return listings; } /** * Remove listings that are similar to already known entries according to the similarity cache. * Adds the remaining listings to the cache. * * @param {Listing[]} listings Listings to filter by similarity. * @returns {Listing[]} Listings considered unique enough to keep. */ _filterBySimilarListings(listings) { const filteredIds = []; const keptListings = listings.filter((listing) => { const similar = this._similarityCache.checkAndAddEntry({ title: listing.title, address: listing.address, price: listing.price, }); if (similar) { logger.debug( `Filtering similar entry for title '${listing.title}' and address '${listing.address}' (Provider: '${this._providerId}')`, ); filteredIds.push(listing.id); } return !similar; }); if (filteredIds.length > 0) { deleteListingsById(filteredIds); } return keptListings; } /** * Handle errors occurring in the pipeline, logging levels depending on type. * * @param {Error} err Error instance thrown by previous steps. * @returns {void} */ _handleError(err) { if (err.name === 'NoNewListingsWarning') { logger.debug(`No new listings found (Provider: '${this._providerId}').`); } else { logger.error(err); } } } export default FredyPipelineExecutioner;