Hanson.xyz Dev
fc018ca604
Property Map: - Replace ACF-based property display with MLS database queries - Use real lat/lng coordinates from MLS (100% coverage) - Create property-card-mls.php template for MLS property cards - Update AJAX handler to filter MLS properties MLS Plugin Enhancements: - Add 'wp mls run' smart sync command (auto-detects full/incremental/resume) - Add database index migrations for lat/lng and composite search indexes - Add comprehensive README.md documentation Documentation: - Update site README.md with sysadmin quick reference - Add FEATURES_PENDING_12_15.md tracking client feature requests 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
1050 lines
37 KiB
PHP
1050 lines
37 KiB
PHP
<?php
|
|
/**
|
|
* MLS Sync Engine
|
|
*
|
|
* Handles synchronization of MLS data including:
|
|
* - Full initial import
|
|
* - Incremental updates
|
|
* - Delete handling (MlgCanView = false)
|
|
* - Sync state tracking for resume capability
|
|
*/
|
|
|
|
if (!defined('ABSPATH')) {
|
|
exit;
|
|
}
|
|
|
|
class MLS_Sync_Engine {
|
|
|
|
/**
|
|
* Sync types
|
|
*/
|
|
const TYPE_FULL = 'full';
|
|
const TYPE_INCREMENTAL = 'incremental';
|
|
const TYPE_MEDIA = 'media';
|
|
|
|
/**
|
|
* Sync statuses
|
|
*/
|
|
const STATUS_PENDING = 'pending';
|
|
const STATUS_RUNNING = 'running';
|
|
const STATUS_COMPLETED = 'completed';
|
|
const STATUS_FAILED = 'failed';
|
|
const STATUS_PAUSED = 'paused';
|
|
|
|
/**
|
|
* Database instance
|
|
*/
|
|
private $db;
|
|
|
|
/**
|
|
* API client instance
|
|
*/
|
|
private $api_client;
|
|
|
|
/**
|
|
* Media handler instance
|
|
*/
|
|
private $media_handler;
|
|
|
|
/**
|
|
* Logger instance
|
|
*/
|
|
private $logger;
|
|
|
|
/**
|
|
* Current sync state ID
|
|
*/
|
|
private $sync_state_id = null;
|
|
|
|
/**
|
|
* Stats for current sync
|
|
*/
|
|
private $stats = array(
|
|
'processed' => 0,
|
|
'created' => 0,
|
|
'updated' => 0,
|
|
'deleted' => 0,
|
|
'errors' => 0,
|
|
);
|
|
|
|
/**
|
|
* Constructor
|
|
*/
|
|
public function __construct(MLS_DB $db, MLS_API_Client $api_client, MLS_Media_Handler $media_handler, MLS_Logger $logger) {
|
|
$this->db = $db;
|
|
$this->api_client = $api_client;
|
|
$this->media_handler = $media_handler;
|
|
$this->logger = $logger;
|
|
}
|
|
|
|
/**
|
|
* Run full sync (Active/Pending properties only)
|
|
*
|
|
* Initial import fetches only Active and Pending properties.
|
|
* Use incremental sync (replication) for ongoing updates.
|
|
*
|
|
* @param bool $dry_run If true, don't make changes
|
|
* @param int|null $limit Max records to process
|
|
* @param callable|null $progress_callback Callback for progress updates
|
|
* @return array Sync results
|
|
*/
|
|
public function run_full_sync($dry_run = false, $limit = null, $progress_callback = null) {
|
|
$this->logger->info('Starting full sync (Active/Pending only)', array('dry_run' => $dry_run, 'limit' => $limit));
|
|
|
|
// Store progress callback for use in process_property
|
|
$this->progress_callback = $progress_callback;
|
|
|
|
// Create sync state record
|
|
if (!$dry_run) {
|
|
$this->sync_state_id = $this->create_sync_state(self::TYPE_FULL);
|
|
$this->logger->set_sync_state($this->sync_state_id);
|
|
}
|
|
|
|
$this->stats = array(
|
|
'processed' => 0,
|
|
'created' => 0,
|
|
'updated' => 0,
|
|
'deleted' => 0,
|
|
'errors' => 0,
|
|
);
|
|
|
|
try {
|
|
// Get first page of Active/Pending properties with media
|
|
$start_time = microtime(true);
|
|
$this->emit_progress('api_request', array(
|
|
'method' => 'GET',
|
|
'url' => 'Property',
|
|
'params' => array('type' => 'initial_sync', 'filter' => 'Active/Pending', 'limit' => $limit),
|
|
));
|
|
|
|
$response = $this->api_client->get_properties_for_initial_sync('Media', $limit ? min($limit, 1000) : null);
|
|
$elapsed = round((microtime(true) - $start_time) * 1000);
|
|
|
|
if (is_wp_error($response)) {
|
|
$this->emit_progress('api_response', array(
|
|
'success' => false,
|
|
'status_code' => 0,
|
|
'error' => $response->get_error_message(),
|
|
'record_count' => 0,
|
|
'response_time' => $elapsed,
|
|
));
|
|
throw new Exception($response->get_error_message());
|
|
}
|
|
|
|
$this->emit_progress('api_response', array(
|
|
'success' => true,
|
|
'status_code' => 200,
|
|
'record_count' => isset($response['value']) ? count($response['value']) : 0,
|
|
'response_time' => $elapsed,
|
|
'has_more' => isset($response['@odata.nextLink']),
|
|
));
|
|
|
|
// Process pages
|
|
$continue = true;
|
|
while ($continue && isset($response['value'])) {
|
|
foreach ($response['value'] as $property) {
|
|
if ($limit && $this->stats['processed'] >= $limit) {
|
|
$continue = false;
|
|
break;
|
|
}
|
|
|
|
$this->process_property($property, $dry_run);
|
|
}
|
|
|
|
// Emit page complete event
|
|
$this->emit_progress('page_complete', array('processed' => $this->stats['processed']));
|
|
|
|
// Check for next page
|
|
if ($continue && isset($response['@odata.nextLink'])) {
|
|
// Save progress
|
|
if (!$dry_run) {
|
|
$this->update_sync_state(array(
|
|
'last_next_link' => $response['@odata.nextLink'],
|
|
'records_processed' => $this->stats['processed'],
|
|
'records_created' => $this->stats['created'],
|
|
'records_updated' => $this->stats['updated'],
|
|
));
|
|
}
|
|
|
|
$start_time = microtime(true);
|
|
$this->emit_progress('api_request', array(
|
|
'method' => 'GET',
|
|
'url' => 'Property (next page)',
|
|
'params' => array('page' => 'next'),
|
|
));
|
|
|
|
$response = $this->api_client->get_next_page($response['@odata.nextLink']);
|
|
$elapsed = round((microtime(true) - $start_time) * 1000);
|
|
|
|
if (is_wp_error($response)) {
|
|
$this->emit_progress('api_response', array(
|
|
'success' => false,
|
|
'status_code' => 0,
|
|
'error' => $response->get_error_message(),
|
|
'record_count' => 0,
|
|
'response_time' => $elapsed,
|
|
));
|
|
throw new Exception($response->get_error_message());
|
|
}
|
|
|
|
$this->emit_progress('api_response', array(
|
|
'success' => true,
|
|
'status_code' => 200,
|
|
'record_count' => isset($response['value']) ? count($response['value']) : 0,
|
|
'response_time' => $elapsed,
|
|
'has_more' => isset($response['@odata.nextLink']),
|
|
));
|
|
} else {
|
|
$continue = false;
|
|
}
|
|
}
|
|
|
|
// Mark sync as completed
|
|
if (!$dry_run) {
|
|
$this->update_sync_state(array(
|
|
'status' => self::STATUS_COMPLETED,
|
|
'completed_at' => current_time('mysql'),
|
|
'records_processed' => $this->stats['processed'],
|
|
'records_created' => $this->stats['created'],
|
|
'records_updated' => $this->stats['updated'],
|
|
'records_deleted' => $this->stats['deleted'],
|
|
));
|
|
|
|
// Update last sync time
|
|
$options = mls_plugin()->get_options();
|
|
$options->update_last_sync('full');
|
|
}
|
|
|
|
$this->logger->info('Full sync completed', $this->stats);
|
|
|
|
} catch (Exception $e) {
|
|
$this->logger->error('Full sync failed', array('error' => $e->getMessage()));
|
|
|
|
if (!$dry_run) {
|
|
$this->update_sync_state(array(
|
|
'status' => self::STATUS_FAILED,
|
|
'last_error' => $e->getMessage(),
|
|
'error_count' => $this->stats['errors'] + 1,
|
|
));
|
|
}
|
|
|
|
return array(
|
|
'success' => false,
|
|
'error' => $e->getMessage(),
|
|
'stats' => $this->stats,
|
|
);
|
|
}
|
|
|
|
return array(
|
|
'success' => true,
|
|
'stats' => $this->stats,
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Run incremental sync (replication)
|
|
*
|
|
* Fetches all properties modified since last sync, including those that:
|
|
* - Became unavailable (MlgCanView=false)
|
|
* - Changed status (Active -> Sold)
|
|
*
|
|
* Properties are deleted from local DB if MlgCanView=false or status not Active/Pending.
|
|
*
|
|
* @param bool $dry_run If true, don't make changes
|
|
* @param callable|null $progress_callback Callback for progress updates
|
|
* @return array Sync results
|
|
*/
|
|
public function run_incremental_sync($dry_run = false, $progress_callback = null) {
|
|
// Get last modification timestamp
|
|
$last_timestamp = $this->get_last_modification_timestamp();
|
|
|
|
if (!$last_timestamp) {
|
|
$this->logger->info('No previous sync found, running full sync instead');
|
|
return $this->run_full_sync($dry_run, null, $progress_callback);
|
|
}
|
|
|
|
$this->logger->info('Starting replication sync', array(
|
|
'since' => $last_timestamp,
|
|
'dry_run' => $dry_run,
|
|
));
|
|
|
|
// Store progress callback for use in process_property
|
|
$this->progress_callback = $progress_callback;
|
|
|
|
if (!$dry_run) {
|
|
$this->sync_state_id = $this->create_sync_state(self::TYPE_INCREMENTAL);
|
|
$this->logger->set_sync_state($this->sync_state_id);
|
|
}
|
|
|
|
$this->stats = array(
|
|
'processed' => 0,
|
|
'created' => 0,
|
|
'updated' => 0,
|
|
'deleted' => 0,
|
|
'errors' => 0,
|
|
);
|
|
|
|
try {
|
|
// Get ALL modified properties (no MlgCanView or status filter for replication)
|
|
$start_time = microtime(true);
|
|
$this->emit_progress('api_request', array(
|
|
'method' => 'GET',
|
|
'url' => 'Property',
|
|
'params' => array('type' => 'replication', 'since' => $last_timestamp),
|
|
));
|
|
|
|
$response = $this->api_client->get_properties_for_replication($last_timestamp, 'Media');
|
|
$elapsed = round((microtime(true) - $start_time) * 1000);
|
|
|
|
if (is_wp_error($response)) {
|
|
$this->emit_progress('api_response', array(
|
|
'success' => false,
|
|
'status_code' => 0,
|
|
'error' => $response->get_error_message(),
|
|
'record_count' => 0,
|
|
'response_time' => $elapsed,
|
|
));
|
|
throw new Exception($response->get_error_message());
|
|
}
|
|
|
|
$this->emit_progress('api_response', array(
|
|
'success' => true,
|
|
'status_code' => 200,
|
|
'record_count' => isset($response['value']) ? count($response['value']) : 0,
|
|
'response_time' => $elapsed,
|
|
'has_more' => isset($response['@odata.nextLink']),
|
|
));
|
|
|
|
// Process pages
|
|
while (isset($response['value'])) {
|
|
foreach ($response['value'] as $property) {
|
|
$this->process_property($property, $dry_run);
|
|
}
|
|
|
|
// Emit page complete event
|
|
$this->emit_progress('page_complete', array('processed' => $this->stats['processed']));
|
|
|
|
// Check for next page
|
|
if (isset($response['@odata.nextLink'])) {
|
|
// Save progress for resume capability
|
|
if (!$dry_run) {
|
|
$this->update_sync_state(array(
|
|
'last_next_link' => $response['@odata.nextLink'],
|
|
'records_processed' => $this->stats['processed'],
|
|
'records_created' => $this->stats['created'],
|
|
'records_updated' => $this->stats['updated'],
|
|
'records_deleted' => $this->stats['deleted'],
|
|
));
|
|
}
|
|
|
|
$start_time = microtime(true);
|
|
$this->emit_progress('api_request', array(
|
|
'method' => 'GET',
|
|
'url' => 'Property (next page)',
|
|
'params' => array('page' => 'next'),
|
|
));
|
|
|
|
$response = $this->api_client->get_next_page($response['@odata.nextLink']);
|
|
$elapsed = round((microtime(true) - $start_time) * 1000);
|
|
|
|
if (is_wp_error($response)) {
|
|
$this->emit_progress('api_response', array(
|
|
'success' => false,
|
|
'status_code' => 0,
|
|
'error' => $response->get_error_message(),
|
|
'record_count' => 0,
|
|
'response_time' => $elapsed,
|
|
));
|
|
throw new Exception($response->get_error_message());
|
|
}
|
|
|
|
$this->emit_progress('api_response', array(
|
|
'success' => true,
|
|
'status_code' => 200,
|
|
'record_count' => isset($response['value']) ? count($response['value']) : 0,
|
|
'response_time' => $elapsed,
|
|
'has_more' => isset($response['@odata.nextLink']),
|
|
));
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
|
|
// Mark sync as completed
|
|
if (!$dry_run) {
|
|
$this->update_sync_state(array(
|
|
'status' => self::STATUS_COMPLETED,
|
|
'completed_at' => current_time('mysql'),
|
|
'records_processed' => $this->stats['processed'],
|
|
'records_created' => $this->stats['created'],
|
|
'records_updated' => $this->stats['updated'],
|
|
'records_deleted' => $this->stats['deleted'],
|
|
));
|
|
|
|
$options = mls_plugin()->get_options();
|
|
$options->update_last_sync('incremental');
|
|
}
|
|
|
|
$this->logger->info('Incremental sync completed', $this->stats);
|
|
|
|
} catch (Exception $e) {
|
|
$this->logger->error('Incremental sync failed', array('error' => $e->getMessage()));
|
|
|
|
if (!$dry_run) {
|
|
$this->update_sync_state(array(
|
|
'status' => self::STATUS_FAILED,
|
|
'last_error' => $e->getMessage(),
|
|
));
|
|
}
|
|
|
|
return array(
|
|
'success' => false,
|
|
'error' => $e->getMessage(),
|
|
'stats' => $this->stats,
|
|
);
|
|
}
|
|
|
|
return array(
|
|
'success' => true,
|
|
'stats' => $this->stats,
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Resume an interrupted sync
|
|
*
|
|
* @param int $sync_state_id Sync state ID to resume
|
|
* @param callable|null $progress_callback Progress callback
|
|
* @return array Sync results
|
|
*/
|
|
public function resume_sync($sync_state_id, $progress_callback = null) {
|
|
global $wpdb;
|
|
|
|
$state = $wpdb->get_row($wpdb->prepare(
|
|
"SELECT * FROM {$this->db->sync_state_table()} WHERE id = %d",
|
|
$sync_state_id
|
|
));
|
|
|
|
if (!$state) {
|
|
return array(
|
|
'success' => false,
|
|
'error' => 'Sync state not found',
|
|
);
|
|
}
|
|
|
|
if ($state->status === self::STATUS_COMPLETED) {
|
|
return array(
|
|
'success' => false,
|
|
'error' => 'Sync already completed',
|
|
);
|
|
}
|
|
|
|
$this->sync_state_id = $sync_state_id;
|
|
$this->progress_callback = $progress_callback;
|
|
$this->logger->set_sync_state($sync_state_id);
|
|
$this->logger->info('Resuming sync', array('sync_state_id' => $sync_state_id));
|
|
|
|
// Load existing stats
|
|
$this->stats = array(
|
|
'processed' => (int) $state->records_processed,
|
|
'created' => (int) $state->records_created,
|
|
'updated' => (int) $state->records_updated,
|
|
'deleted' => (int) $state->records_deleted,
|
|
'errors' => (int) $state->error_count,
|
|
);
|
|
|
|
// Update status to running
|
|
$this->update_sync_state(array('status' => self::STATUS_RUNNING));
|
|
|
|
try {
|
|
// Resume from last next_link
|
|
if ($state->last_next_link) {
|
|
$response = $this->api_client->get_next_page($state->last_next_link);
|
|
} else {
|
|
// Start fresh
|
|
$response = $this->api_client->get_properties_for_sync(
|
|
$state->last_modification_timestamp,
|
|
'Media'
|
|
);
|
|
}
|
|
|
|
if (is_wp_error($response)) {
|
|
throw new Exception($response->get_error_message());
|
|
}
|
|
|
|
// Process remaining pages
|
|
while (isset($response['value'])) {
|
|
foreach ($response['value'] as $property) {
|
|
$this->process_property($property, false);
|
|
}
|
|
|
|
// Emit page complete event
|
|
$this->emit_progress('page_complete', array('processed' => $this->stats['processed']));
|
|
|
|
if (isset($response['@odata.nextLink'])) {
|
|
$this->update_sync_state(array(
|
|
'last_next_link' => $response['@odata.nextLink'],
|
|
'records_processed' => $this->stats['processed'],
|
|
));
|
|
|
|
$response = $this->api_client->get_next_page($response['@odata.nextLink']);
|
|
|
|
if (is_wp_error($response)) {
|
|
throw new Exception($response->get_error_message());
|
|
}
|
|
} else {
|
|
break;
|
|
}
|
|
}
|
|
|
|
$this->update_sync_state(array(
|
|
'status' => self::STATUS_COMPLETED,
|
|
'completed_at' => current_time('mysql'),
|
|
'records_processed' => $this->stats['processed'],
|
|
'records_created' => $this->stats['created'],
|
|
'records_updated' => $this->stats['updated'],
|
|
'records_deleted' => $this->stats['deleted'],
|
|
));
|
|
|
|
$this->logger->info('Resume sync completed', $this->stats);
|
|
|
|
} catch (Exception $e) {
|
|
$this->logger->error('Resume sync failed', array('error' => $e->getMessage()));
|
|
|
|
$this->update_sync_state(array(
|
|
'status' => self::STATUS_FAILED,
|
|
'last_error' => $e->getMessage(),
|
|
));
|
|
|
|
return array(
|
|
'success' => false,
|
|
'error' => $e->getMessage(),
|
|
'stats' => $this->stats,
|
|
);
|
|
}
|
|
|
|
return array(
|
|
'success' => true,
|
|
'stats' => $this->stats,
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Progress callback reference
|
|
*/
|
|
private $progress_callback = null;
|
|
|
|
/**
|
|
* Allowed statuses for our database (Active/Pending only)
|
|
*/
|
|
const ALLOWED_STATUSES = array('Active', 'Pending');
|
|
|
|
/**
|
|
* Process a single property record
|
|
*
|
|
* During replication, properties are deleted if:
|
|
* - MlgCanView = false (removed from feed)
|
|
* - StandardStatus not in (Active, Pending)
|
|
*
|
|
* @param array $property Property data from API
|
|
* @param bool $dry_run If true, don't make changes
|
|
*/
|
|
private function process_property($property, $dry_run = false) {
|
|
global $wpdb;
|
|
|
|
$this->stats['processed']++;
|
|
|
|
$listing_key = $property['ListingKey'] ?? null;
|
|
if (!$listing_key) {
|
|
$this->stats['errors']++;
|
|
$this->logger->warning('Property missing ListingKey', array('property' => $property));
|
|
return;
|
|
}
|
|
|
|
// Check MlgCanView and StandardStatus
|
|
$can_view = $property['MlgCanView'] ?? true;
|
|
$status = $property['StandardStatus'] ?? null;
|
|
|
|
// Delete if: not viewable OR status is not Active/Pending
|
|
$should_delete = !$can_view || !in_array($status, self::ALLOWED_STATUSES);
|
|
|
|
if ($should_delete) {
|
|
// Check if we have this record locally before attempting delete
|
|
$exists_locally = $wpdb->get_var($wpdb->prepare(
|
|
"SELECT id FROM {$this->db->properties_table()} WHERE listing_key = %s",
|
|
$listing_key
|
|
));
|
|
|
|
if ($exists_locally) {
|
|
if (!$dry_run) {
|
|
$this->delete_property($listing_key);
|
|
}
|
|
$this->stats['deleted']++;
|
|
$this->emit_progress('property_deleted', array(
|
|
'listing_key' => $listing_key,
|
|
'reason' => !$can_view ? 'MlgCanView=false' : "Status={$status}",
|
|
));
|
|
}
|
|
// If not in our DB, just skip silently (e.g., Sold property we never had)
|
|
return;
|
|
}
|
|
|
|
// Check if property exists
|
|
$existing = $wpdb->get_var($wpdb->prepare(
|
|
"SELECT id FROM {$this->db->properties_table()} WHERE listing_key = %s",
|
|
$listing_key
|
|
));
|
|
|
|
// Prepare data for insert/update
|
|
$data = $this->map_property_data($property);
|
|
|
|
if ($dry_run) {
|
|
if ($existing) {
|
|
$this->stats['updated']++;
|
|
$this->emit_progress('property_skipped', array('listing_key' => $listing_key));
|
|
} else {
|
|
$this->stats['created']++;
|
|
$this->emit_progress('property_skipped', array('listing_key' => $listing_key));
|
|
}
|
|
return;
|
|
}
|
|
|
|
if ($existing) {
|
|
// Update existing
|
|
$wpdb->update(
|
|
$this->db->properties_table(),
|
|
$data,
|
|
array('listing_key' => $listing_key)
|
|
);
|
|
$this->stats['updated']++;
|
|
$this->emit_progress('property_updated', array('listing_key' => $listing_key));
|
|
} else {
|
|
// Insert new
|
|
$data['listing_key'] = $listing_key;
|
|
$data['created_at'] = current_time('mysql');
|
|
$wpdb->insert($this->db->properties_table(), $data);
|
|
$this->stats['created']++;
|
|
$this->emit_progress('property_created', array('listing_key' => $listing_key));
|
|
}
|
|
|
|
// Process media if present
|
|
if (isset($property['Media']) && is_array($property['Media'])) {
|
|
$this->media_handler->sync_property_media($listing_key, $property['Media'], false, $this->progress_callback);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Emit progress event
|
|
*
|
|
* @param string $event Event name
|
|
* @param array $data Event data
|
|
*/
|
|
private function emit_progress($event, $data = array()) {
|
|
if ($this->progress_callback) {
|
|
call_user_func($this->progress_callback, $event, $data);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Map API property data to database columns
|
|
*
|
|
* @param array $property API property data
|
|
* @return array Mapped data for database
|
|
*/
|
|
private function map_property_data($property) {
|
|
return array(
|
|
'listing_id' => $property['ListingId'] ?? null,
|
|
'originating_system' => $property['OriginatingSystemName'] ?? 'northstar',
|
|
'standard_status' => $property['StandardStatus'] ?? null,
|
|
'mls_status' => $property['MlsStatus'] ?? null,
|
|
'mlg_can_view' => isset($property['MlgCanView']) ? ($property['MlgCanView'] ? 1 : 0) : 1,
|
|
|
|
'list_price' => $property['ListPrice'] ?? null,
|
|
'original_list_price' => $property['OriginalListPrice'] ?? null,
|
|
'close_price' => $property['ClosePrice'] ?? null,
|
|
|
|
'street_number' => $property['StreetNumber'] ?? null,
|
|
'street_name' => $property['StreetName'] ?? null,
|
|
'street_suffix' => $property['StreetSuffix'] ?? null,
|
|
'unit_number' => $property['UnitNumber'] ?? null,
|
|
'city' => $property['City'] ?? null,
|
|
'state_or_province' => $property['StateOrProvince'] ?? 'MN',
|
|
'postal_code' => $property['PostalCode'] ?? null,
|
|
'county' => $property['CountyOrParish'] ?? null,
|
|
'latitude' => $property['Latitude'] ?? null,
|
|
'longitude' => $property['Longitude'] ?? null,
|
|
|
|
'property_type' => $property['PropertyType'] ?? null,
|
|
'property_sub_type' => $property['PropertySubType'] ?? null,
|
|
'bedrooms_total' => $property['BedroomsTotal'] ?? null,
|
|
'bathrooms_total' => $property['BathroomsTotalInteger'] ?? null,
|
|
'bathrooms_full' => $property['BathroomsFull'] ?? null,
|
|
'bathrooms_half' => $property['BathroomsHalf'] ?? null,
|
|
'living_area' => $property['LivingArea'] ?? null,
|
|
'lot_size_area' => $property['LotSizeArea'] ?? null,
|
|
'lot_size_units' => $property['LotSizeUnits'] ?? null,
|
|
'year_built' => $property['YearBuilt'] ?? null,
|
|
'garage_spaces' => $property['GarageSpaces'] ?? null,
|
|
|
|
'public_remarks' => $property['PublicRemarks'] ?? null,
|
|
'directions' => $property['Directions'] ?? null,
|
|
|
|
'list_agent_key' => $property['ListAgentKey'] ?? null,
|
|
'list_agent_mls_id' => $property['ListAgentMlsId'] ?? null,
|
|
'list_office_key' => $property['ListOfficeKey'] ?? null,
|
|
'list_office_mls_id' => $property['ListOfficeMlsId'] ?? null,
|
|
'list_office_name' => $property['ListOfficeName'] ?? null,
|
|
|
|
'photos_count' => $property['PhotosCount'] ?? 0,
|
|
'modification_timestamp' => $this->format_timestamp($property['ModificationTimestamp'] ?? null),
|
|
'photos_change_timestamp' => $this->format_timestamp($property['PhotosChangeTimestamp'] ?? null),
|
|
'listing_contract_date' => $this->format_date($property['ListingContractDate'] ?? null),
|
|
'close_date' => $this->format_date($property['CloseDate'] ?? null),
|
|
'days_on_market' => $property['DaysOnMarket'] ?? null,
|
|
|
|
'raw_data' => wp_json_encode($property),
|
|
'updated_at' => current_time('mysql'),
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Format ISO 8601 timestamp to MySQL datetime
|
|
*
|
|
* @param string|null $timestamp ISO 8601 timestamp
|
|
* @return string|null MySQL datetime
|
|
*/
|
|
private function format_timestamp($timestamp) {
|
|
if (!$timestamp) {
|
|
return null;
|
|
}
|
|
$dt = new DateTime($timestamp);
|
|
return $dt->format('Y-m-d H:i:s');
|
|
}
|
|
|
|
/**
|
|
* Format date string to MySQL date
|
|
*
|
|
* @param string|null $date Date string
|
|
* @return string|null MySQL date
|
|
*/
|
|
private function format_date($date) {
|
|
if (!$date) {
|
|
return null;
|
|
}
|
|
return date('Y-m-d', strtotime($date));
|
|
}
|
|
|
|
/**
|
|
* Delete a property and its media
|
|
*
|
|
* @param string $listing_key Listing key
|
|
*/
|
|
private function delete_property($listing_key) {
|
|
global $wpdb;
|
|
|
|
// Delete media files
|
|
$this->media_handler->delete_property_media($listing_key);
|
|
|
|
// Delete from database
|
|
$wpdb->delete(
|
|
$this->db->properties_table(),
|
|
array('listing_key' => $listing_key)
|
|
);
|
|
|
|
$this->logger->debug('Deleted property', array('listing_key' => $listing_key));
|
|
}
|
|
|
|
/**
|
|
* Create a sync state record
|
|
*
|
|
* @param string $type Sync type
|
|
* @return int Sync state ID
|
|
*/
|
|
private function create_sync_state($type) {
|
|
global $wpdb;
|
|
|
|
$wpdb->insert(
|
|
$this->db->sync_state_table(),
|
|
array(
|
|
'sync_type' => $type,
|
|
'entity_type' => 'Property',
|
|
'status' => self::STATUS_RUNNING,
|
|
'started_at' => current_time('mysql'),
|
|
'created_at' => current_time('mysql'),
|
|
)
|
|
);
|
|
|
|
return $wpdb->insert_id;
|
|
}
|
|
|
|
/**
|
|
* Update sync state record
|
|
*
|
|
* @param array $data Data to update
|
|
*/
|
|
private function update_sync_state($data) {
|
|
global $wpdb;
|
|
|
|
if (!$this->sync_state_id) {
|
|
return;
|
|
}
|
|
|
|
$data['updated_at'] = current_time('mysql');
|
|
|
|
$wpdb->update(
|
|
$this->db->sync_state_table(),
|
|
$data,
|
|
array('id' => $this->sync_state_id)
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Get last modification timestamp from synced data
|
|
*
|
|
* @return string|null ISO 8601 timestamp
|
|
*/
|
|
private function get_last_modification_timestamp() {
|
|
global $wpdb;
|
|
|
|
$timestamp = $wpdb->get_var(
|
|
"SELECT MAX(modification_timestamp) FROM {$this->db->properties_table()}"
|
|
);
|
|
|
|
if ($timestamp) {
|
|
$dt = new DateTime($timestamp);
|
|
return $dt->format('Y-m-d\TH:i:s.v\Z');
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
/**
|
|
* Get sync status
|
|
*
|
|
* @return array Sync status
|
|
*/
|
|
public function get_status() {
|
|
global $wpdb;
|
|
|
|
$last_sync = $wpdb->get_row(
|
|
"SELECT * FROM {$this->db->sync_state_table()}
|
|
WHERE status = 'completed'
|
|
ORDER BY completed_at DESC
|
|
LIMIT 1"
|
|
);
|
|
|
|
$running_sync = $wpdb->get_row(
|
|
"SELECT * FROM {$this->db->sync_state_table()}
|
|
WHERE status = 'running'
|
|
ORDER BY started_at DESC
|
|
LIMIT 1"
|
|
);
|
|
|
|
$failed_sync = $wpdb->get_row(
|
|
"SELECT * FROM {$this->db->sync_state_table()}
|
|
WHERE status = 'failed'
|
|
ORDER BY updated_at DESC
|
|
LIMIT 1"
|
|
);
|
|
|
|
return array(
|
|
'last_sync' => $last_sync,
|
|
'running_sync' => $running_sync,
|
|
'last_failed' => $failed_sync,
|
|
'stats' => $this->db->get_stats(),
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Get resumable (failed or interrupted) syncs
|
|
*
|
|
* @return array List of resumable sync states
|
|
*/
|
|
public function get_resumable_syncs() {
|
|
global $wpdb;
|
|
|
|
// Find failed syncs that have a next_link (can be resumed)
|
|
// Also find "running" syncs older than 1 hour (likely interrupted)
|
|
$one_hour_ago = date('Y-m-d H:i:s', strtotime('-1 hour'));
|
|
|
|
return $wpdb->get_results($wpdb->prepare(
|
|
"SELECT * FROM {$this->db->sync_state_table()}
|
|
WHERE (status = 'failed' AND last_next_link IS NOT NULL)
|
|
OR (status = 'running' AND updated_at < %s)
|
|
ORDER BY updated_at DESC",
|
|
$one_hour_ago
|
|
));
|
|
}
|
|
|
|
/**
|
|
* Get the most recent resumable sync
|
|
*
|
|
* @return object|null Sync state or null
|
|
*/
|
|
public function get_latest_resumable() {
|
|
$resumable = $this->get_resumable_syncs();
|
|
return !empty($resumable) ? $resumable[0] : null;
|
|
}
|
|
|
|
/**
|
|
* Mark stale running syncs as failed
|
|
* Call this on startup to clean up interrupted syncs
|
|
*
|
|
* @param int $hours_threshold Hours after which running sync is considered stale
|
|
* @return int Number of syncs marked as failed
|
|
*/
|
|
public function cleanup_stale_syncs($hours_threshold = 1) {
|
|
global $wpdb;
|
|
|
|
$threshold_time = date('Y-m-d H:i:s', strtotime("-{$hours_threshold} hour"));
|
|
|
|
$updated = $wpdb->query($wpdb->prepare(
|
|
"UPDATE {$this->db->sync_state_table()}
|
|
SET status = 'failed', last_error = 'Sync interrupted (stale)'
|
|
WHERE status = 'running' AND updated_at < %s",
|
|
$threshold_time
|
|
));
|
|
|
|
if ($updated > 0) {
|
|
$this->logger->info('Cleaned up stale syncs', array('count' => $updated));
|
|
}
|
|
|
|
return $updated;
|
|
}
|
|
|
|
/**
|
|
* Auto-resume the most recent failed/interrupted sync
|
|
*
|
|
* @param callable|null $progress_callback Progress callback
|
|
* @return array|null Sync results or null if nothing to resume
|
|
*/
|
|
public function auto_resume($progress_callback = null) {
|
|
// First clean up any stale syncs
|
|
$this->cleanup_stale_syncs();
|
|
|
|
// Find the most recent resumable sync
|
|
$resumable = $this->get_latest_resumable();
|
|
|
|
if (!$resumable) {
|
|
return null;
|
|
}
|
|
|
|
$this->logger->info('Auto-resuming sync', array(
|
|
'sync_id' => $resumable->id,
|
|
'type' => $resumable->sync_type,
|
|
'processed' => $resumable->records_processed,
|
|
));
|
|
|
|
return $this->resume_sync($resumable->id, $progress_callback);
|
|
}
|
|
|
|
/**
|
|
* Smart sync - autonomous self-healing sync that handles all scenarios
|
|
*
|
|
* Decision logic:
|
|
* 1. If a sync is currently running (and not stale), abort
|
|
* 2. If there's a resumable failed/interrupted sync, resume it
|
|
* 3. If no data exists, run full sync
|
|
* 4. Otherwise, run incremental sync
|
|
*
|
|
* On failure, the sync state is preserved for future resume.
|
|
*
|
|
* @param callable|null $progress_callback Progress callback
|
|
* @param callable|null $status_callback Callback for status messages: function(string $message, string $level)
|
|
* @return array Sync results with 'action' key indicating what was done
|
|
*/
|
|
public function smart_sync($progress_callback = null, $status_callback = null) {
|
|
// Helper to emit status messages
|
|
$status = function($message, $level = 'info') use ($status_callback) {
|
|
if ($status_callback) {
|
|
call_user_func($status_callback, $message, $level);
|
|
}
|
|
$this->logger->log($level, $message);
|
|
};
|
|
|
|
// Step 1: Clean up stale syncs (running > 1 hour = probably dead)
|
|
$stale_cleaned = $this->cleanup_stale_syncs();
|
|
if ($stale_cleaned > 0) {
|
|
$status("Cleaned up {$stale_cleaned} stale sync(s)", 'info');
|
|
}
|
|
|
|
// Step 2: Check if a sync is actively running
|
|
$running = $this->get_running_sync();
|
|
if ($running) {
|
|
$status("Sync #{$running->id} is already running (started {$running->started_at})", 'warning');
|
|
return array(
|
|
'success' => false,
|
|
'action' => 'aborted',
|
|
'reason' => 'Sync already running',
|
|
'running_sync' => $running,
|
|
);
|
|
}
|
|
|
|
// Step 3: Check for resumable syncs
|
|
$resumable = $this->get_latest_resumable();
|
|
if ($resumable) {
|
|
$status("Found resumable sync #{$resumable->id} ({$resumable->sync_type}), processed {$resumable->records_processed} records", 'info');
|
|
$status("Resuming...", 'info');
|
|
|
|
$result = $this->resume_sync($resumable->id, $progress_callback);
|
|
$result['action'] = 'resumed';
|
|
$result['resumed_sync_id'] = $resumable->id;
|
|
return $result;
|
|
}
|
|
|
|
// Step 4: Check if we have any data
|
|
$has_data = $this->has_synced_data();
|
|
|
|
if (!$has_data) {
|
|
// No data - need full sync
|
|
$status("No existing data found, starting full sync", 'info');
|
|
|
|
$result = $this->run_full_sync(false, null, $progress_callback);
|
|
$result['action'] = 'full';
|
|
return $result;
|
|
}
|
|
|
|
// Step 5: We have data - run incremental sync
|
|
$last_timestamp = $this->get_last_modification_timestamp();
|
|
$status("Running incremental sync (changes since {$last_timestamp})", 'info');
|
|
|
|
$result = $this->run_incremental_sync(false, $progress_callback);
|
|
$result['action'] = 'incremental';
|
|
return $result;
|
|
}
|
|
|
|
/**
|
|
* Check if there's a currently running sync (not stale)
|
|
*
|
|
* @return object|null Running sync state or null
|
|
*/
|
|
public function get_running_sync() {
|
|
global $wpdb;
|
|
|
|
$one_hour_ago = date('Y-m-d H:i:s', strtotime('-1 hour'));
|
|
|
|
return $wpdb->get_row($wpdb->prepare(
|
|
"SELECT * FROM {$this->db->sync_state_table()}
|
|
WHERE status = 'running' AND updated_at >= %s
|
|
ORDER BY started_at DESC
|
|
LIMIT 1",
|
|
$one_hour_ago
|
|
));
|
|
}
|
|
|
|
/**
|
|
* Check if we have any synced property data
|
|
*
|
|
* @return bool
|
|
*/
|
|
public function has_synced_data() {
|
|
global $wpdb;
|
|
|
|
$count = $wpdb->get_var(
|
|
"SELECT COUNT(*) FROM {$this->db->properties_table()} WHERE mlg_can_view = 1"
|
|
);
|
|
|
|
return (int) $count > 0;
|
|
}
|
|
}
|