true ], 'hub:reconcile', DAY_IN_SECONDS ); } /** * Run a pass. * * @param array $payload `daily`, or `entity` and `since` (Unix time) from a directive. * * @internal Hooked to {@see Hooks::HUB_RECONCILE}. */ public function run( array $payload ): void { if ( null === Credentials::load() ) { return; } $daily = (bool) ( $payload['daily'] ?? false ); $since = isset( $payload['since'] ) ? (int) $payload['since'] : time() - self::WINDOW; $entities = $daily ? [ 'product', 'coupon', 'order' ] : array_intersect( [ (string) ( $payload['entity'] ?? '' ) ], [ 'product', 'coupon', 'order', 'customer' ] ); foreach ( $entities as $entity ) { foreach ( $this->changedSince( $entity, $since ) as $id ) { $event = 'order' === $entity ? Backfill::orderEvent( $id ) : CatalogueWatcher::snapshot( $entity, $id ); if ( null !== $event ) { Outbox::add( $entity . '.upserted', $event, Outbox::ORIGIN_RESYNC ); } } } if ( $daily ) { Heartbeat::emit(); } Outbox::write(); if ( $daily ) { // `push()`: this job is reserved while it runs, and `pushUnique()` would // count it as the duplicate and drop tomorrow's run. Queue::push( Hooks::HUB_RECONCILE, [ 'daily' => true ], DAY_IN_SECONDS, 30, 'hub:reconcile' ); } } /** * Ids of an entity modified since a time. * * @param string $entity Entity. * @param int $since Unix time. * * @return list */ private function changedSince( string $entity, int $since ): array { $after = gmdate( 'Y-m-d H:i:s', $since ); $ids = []; for ( $page = 1; $page <= 50; $page++ ) { $batch = $this->page( $entity, $after, $page ); $ids = array_merge( $ids, $batch ); if ( count( $batch ) < self::PAGE ) { break; } } return $ids; } /** * One page of changed ids. * * @param string $entity Entity. * @param string $after UTC datetime. * @param int $page 1-based page. * * @return list */ private function page( string $entity, string $after, int $page ): array { if ( 'order' === $entity ) { $ids = wc_get_orders( [ 'type' => 'shop_order', 'status' => Backfill::orderStatuses(), 'date_modified' => '>' . strtotime( $after . ' UTC' ), 'limit' => self::PAGE, 'paged' => $page, 'return' => 'ids', ] ); return is_array( $ids ) ? Backfill::toIds( $ids ) : []; } if ( 'customer' === $entity ) { // Users have no modified date; a customer resync re-sends them all. return Backfill::ids( 'customer', $page ); } $posts = get_posts( [ 'post_type' => 'product' === $entity ? [ 'product', 'product_variation' ] : 'shop_coupon', 'post_status' => [ 'publish', 'draft', 'pending', 'private', 'future' ], 'date_query' => [ [ 'column' => 'post_modified_gmt', 'after' => $after, ], ], 'orderby' => 'ID', 'order' => 'ASC', 'posts_per_page' => self::PAGE, 'paged' => $page, 'fields' => 'ids', ] ); return array_map( 'intval', $posts ); } }