update( [ 'backfill_state' => 'running', 'backfill_entity' => $entity, 'backfill_done' => 0, ] ); Queue::push( Hooks::HUB_BACKFILL, [ 'entity' => $entity, 'page' => 1, 'only' => null !== $only ], 5, 30, 'hub:backfill' ); } /** * Run one page. * * @param array $payload `entity`, `page` (1-based), `only`. * * @internal Hooked to {@see Hooks::HUB_BACKFILL}. */ public function run( array $payload ): void { if ( null === Credentials::load() || ! function_exists( 'wc_get_orders' ) ) { return; } $entity = in_array( $payload['entity'] ?? null, self::ENTITIES, true ) ? (string) $payload['entity'] : 'order'; $page = max( 1, (int) ( $payload['page'] ?? 1 ) ); $only = (bool) ( $payload['only'] ?? false ); $state = new State(); if ( Outbox::counts()['pending'] > self::BACKPRESSURE ) { Queue::push( Hooks::HUB_BACKFILL, $payload, 60, 30, 'hub:backfill' ); return; } $ids = self::ids( $entity, $page ); foreach ( $ids as $id ) { $event = 'order' === $entity ? self::orderEvent( $id ) : CatalogueWatcher::snapshot( $entity, $id ); if ( null !== $event ) { Outbox::add( $entity . '.upserted', $event, Outbox::ORIGIN_BACKFILL ); } } Outbox::write(); $state->update( [ 'backfill_entity' => $entity, 'backfill_done' => (int) $state->get( 'backfill_done', 0 ) + count( $ids ), ] ); if ( count( $ids ) === self::PAGE ) { Queue::push( Hooks::HUB_BACKFILL, [ 'entity' => $entity, 'page' => $page + 1, 'only' => $only ], 2, 30, 'hub:backfill' ); return; } $next = array_search( $entity, self::ENTITIES, true ); $next = false === $next ? null : ( self::ENTITIES[ $next + 1 ] ?? null ); if ( ! $only && null !== $next ) { Queue::push( Hooks::HUB_BACKFILL, [ 'entity' => $next, 'page' => 1, 'only' => false ], 2, 30, 'hub:backfill' ); return; } $state->update( [ 'backfill_state' => 'done', 'backfill_entity' => null, ] ); // The hub compares its counts with these once the upload is done. Heartbeat::emit(); Outbox::write(); } /** * One page of ids of an entity, oldest first. * * @param string $entity Entity. * @param int $page 1-based page. * * @return list */ public static function ids( string $entity, int $page ): array { $offset = ( $page - 1 ) * self::PAGE; switch ( $entity ) { case 'category': $terms = get_terms( [ 'taxonomy' => 'product_cat', 'hide_empty' => false, 'orderby' => 'term_id', 'order' => 'ASC', 'number' => self::PAGE, 'offset' => $offset, 'fields' => 'ids', ] ); return is_array( $terms ) ? array_map( 'intval', $terms ) : []; case 'product': $ids = wc_get_products( [ 'type' => array_merge( array_keys( wc_get_product_types() ), [ 'variation' ] ), 'status' => [ 'publish', 'draft', 'pending', 'private' ], 'orderby' => 'ID', 'order' => 'ASC', 'limit' => self::PAGE, 'page' => $page, 'return' => 'ids', ] ); return is_array( $ids ) ? array_map( 'intval', $ids ) : []; case 'customer': $users = get_users( [ 'role' => 'customer', 'orderby' => 'ID', 'order' => 'ASC', 'number' => self::PAGE, 'offset' => $offset, 'fields' => 'ID', ] ); return array_map( 'intval', $users ); case 'coupon': $posts = get_posts( [ 'post_type' => 'shop_coupon', 'post_status' => [ 'publish', 'draft', 'pending', 'private', 'future' ], 'orderby' => 'ID', 'order' => 'ASC', 'posts_per_page' => self::PAGE, 'offset' => $offset, 'fields' => 'ids', ] ); return array_map( 'intval', $posts ); default: $ids = wc_get_orders( [ 'type' => 'shop_order', 'status' => self::orderStatuses(), 'orderby' => 'ID', 'order' => 'ASC', 'limit' => self::PAGE, 'paged' => $page, 'return' => 'ids', 'paginate' => false, ] ); return is_array( $ids ) ? self::toIds( $ids ) : []; } } /** * Ids from a WooCommerce query that may hand back ids or objects. * * @param array $items Query result. * * @return list */ public static function toIds( array $items ): array { $ids = []; foreach ( $items as $item ) { $ids[] = is_object( $item ) && method_exists( $item, 'get_id' ) ? (int) $item->get_id() : (int) $item; } return $ids; } /** * Every order status except the block checkout's draft. * * @return list */ public static function orderStatuses(): array { return array_values( array_filter( array_keys( wc_get_order_statuses() ), static fn( string $status ): bool => 'wc-checkout-draft' !== $status ) ); } /** * Snapshot of one order, or null. * * @param int $id Order id. * * @return array|null */ public static function orderEvent( int $id ): ?array { $order = wc_get_order( $id ); return $order instanceof WC_Order ? OrderSnapshot::build( $order ) : null; } }