table = $wpdb->prefix . 'shiftora_file_progress'; } /** * Persist every file whose state changed this chunk. * * The map may be sparse: each entry carries its manifest `sequence` so ordering * survives. `ensure_seeded` passes the full snapshot and lets key position stand in for it. * * @param array> $file_offsets Sparse live entries (authoritative). * @param string[] $changed_paths Paths mutated since the last write. * * @return int Number of paths written. */ public function upsert_changed( int $migration_id, string $stage, array $file_offsets, array $changed_paths ): int { $positions = array_flip( array_keys( $file_offsets ) ); $written = 0; foreach ( $changed_paths as $path ) { if ( ! isset( $file_offsets[ $path ] ) ) { continue; } $meta = $file_offsets[ $path ]; $sequence = isset( $meta['sequence'] ) ? (int) $meta['sequence'] : (int) ( $positions[ $path ] ?? 0 ); // A numeric array key becomes an int in PHP, so a root-level file named "123" needs the cast below for strict types. $this->upsert_row( $migration_id, $stage, $sequence, (string) $path, $meta ); $written++; } return $written; } /** * Convert a progress row (object from next_pending(), array from next_pending_batch()) * into the engine's offsets-entry shape, keeping the sequence so a sparse upsert preserves ordering. * * @param object|array $row * * @return array */ public function row_to_meta( object|array $row ): array { $row = (array) $row; $status = (string) ( $row['status'] ?? 'pending' ); return [ 'size' => (int) ( $row['byte_total'] ?? 0 ), 'hash' => (string) ( $row['content_hash'] ?? '' ), 'offset' => (int) ( $row['byte_done'] ?? 0 ), 'done' => in_array( $status, [ 'done', 'failed', 'skipped' ], true ), 'failed' => 'failed' === $status, 'skipped' => 'skipped' === $status, 'errors' => (int) ( $row['errors'] ?? 0 ), 'skip_batch' => (bool) ( $row['skip_batch'] ?? 0 ), 'sequence' => (int) ( $row['sequence'] ?? 0 ), ]; } /** * Roll one stage's rows up into the counters the status payload and migration row consume. * * @return array{total_files: int, transferred_files: int, skipped_files: int, bytes_total: int, bytes_transferred: int, progress: int} */ public function summarize( int $migration_id, string $stage ): array { $wpdb = $this->wpdb; // phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required $row = $wpdb->get_row( $wpdb->prepare( "SELECT SUM(status <> 'skipped') AS total_files, SUM(status = 'done') AS transferred_files, SUM(status = 'skipped') AS skipped_files, SUM(CASE WHEN status = 'skipped' THEN 0 ELSE byte_total END) AS bytes_total, SUM(CASE WHEN status = 'done' THEN byte_total WHEN status = 'skipped' THEN 0 ELSE LEAST(byte_done, byte_total) END) AS bytes_transferred FROM %i WHERE migration_id = %d AND stage = %s", $this->table, $migration_id, $stage ), ARRAY_A ); $total = (int) ( $row['total_files'] ?? 0 ); $transferred = (int) ( $row['transferred_files'] ?? 0 ); $bytes_total = (int) ( $row['bytes_total'] ?? 0 ); $bytes_moved = (int) ( $row['bytes_transferred'] ?? 0 ); if ( $bytes_total > 0 ) { $bytes_moved = min( $bytes_moved, $bytes_total ); } return [ 'total_files' => $total, 'transferred_files' => min( $transferred, $total ), 'skipped_files' => (int) ( $row['skipped_files'] ?? 0 ), 'bytes_total' => $bytes_total, 'bytes_transferred' => max( 0, $bytes_moved ), 'progress' => $total > 0 ? min( 100, (int) round( $transferred / $total * 100 ) ) : 0, ]; } /** * The lowest manifest sequence still pending. * * @return object{sequence: int, virtual_path: string, status: string}|null */ public function next_pending( int $migration_id, string $stage ): ?object { $wpdb = $this->wpdb; // phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required return $wpdb->get_row( $wpdb->prepare( "SELECT sequence, virtual_path, status, byte_total, byte_done, content_hash, errors, skip_batch FROM %i WHERE migration_id = %d AND stage = %s AND status = 'pending' ORDER BY sequence ASC LIMIT 1", $this->table, $migration_id, $stage ) ) ?: null; } /** * A bounded slice of pending files, lowest sequence first. * * Bulk work pages through these instead of loading the full offsets map each * chunk. The (migration_id, stage, status, sequence) index serves the lookup. * * @return array> */ public function next_pending_batch( int $migration_id, string $stage, int $limit = 250 ): array { $wpdb = $this->wpdb; // phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required $rows = $wpdb->get_results( $wpdb->prepare( "SELECT sequence, virtual_path, status, byte_total, byte_done, content_hash, errors, skip_batch FROM %i WHERE migration_id = %d AND stage = %s AND status = 'pending' ORDER BY sequence ASC LIMIT %d", $this->table, $migration_id, $stage, max( 1, min( $limit, 1000 ) ) ), ARRAY_A ); return is_array( $rows ) ? $rows : []; } /** * A bounded page of files this run marked transferred, for checking against the * filesystem, the only way to tell a promoted file from a row that says it was. * * @return array> */ public function sample_done( int $migration_id, string $stage, int $limit = 25 ): array { $wpdb = $this->wpdb; // phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required $rows = $wpdb->get_results( $wpdb->prepare( "SELECT sequence, virtual_path, byte_total, content_hash FROM %i WHERE migration_id = %d AND stage = %s AND status = 'done' ORDER BY sequence ASC LIMIT %d", $this->table, $migration_id, $stage, max( 1, min( $limit, 200 ) ) ), ARRAY_A ); return is_array( $rows ) ? $rows : []; } /** * Whether rows exist for a stage. Guards the one-time conversion of the older JSON offsets format. */ public function has_rows( int $migration_id, string $stage ): bool { $wpdb = $this->wpdb; // phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required $count = $wpdb->get_var( $wpdb->prepare( 'SELECT COUNT(*) FROM %i WHERE migration_id = %d AND stage = %s', $this->table, $migration_id, $stage ) ); return (int) $count > 0; } /** * Mirror a freshly built (or decoded) offsets map into the table. * * Idempotent: existing rows are refreshed to the snapshot's statuses, never * duplicated, so a stage that gained files after a re-plan gets new rows without losing state. * * @param array> $file_offsets * * @return int Number of paths written. */ public function ensure_seeded( int $migration_id, string $stage, array $file_offsets ): int { return $this->upsert_changed( $migration_id, $stage, $file_offsets, array_keys( $file_offsets ) ); } /** * Bring a stage's rows back into agreement with a freshly fetched manifest. * * The manifest decides which files exist; the table decides how far each got. New * files get rows, vanished files lose theirs, survivors keep offset, status and * error count. Dropping matters because a row left by an earlier manifest stays * 'pending' forever and the remote refuses it with shiftora_file_not_allowed. * * @param array> $file_offsets The manifest as offsets. * @param bool $preserve_progress False resets surviving files to pending (a stage restart). * * @return array{written: int, dropped: int} Rows added and rows removed. */ public function reconcile_stage( int $migration_id, string $stage, array $file_offsets, bool $preserve_progress = true ): array { $wpdb = $this->wpdb; $keep = array_keys( $file_offsets ); // phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required $existing = $wpdb->get_col( $wpdb->prepare( 'SELECT virtual_path FROM %i WHERE migration_id = %d AND stage = %s', $this->table, $migration_id, $stage ) ); $existing = array_map( static fn( mixed $path ): string => (string) $path, (array) $existing ); $surplus = array_values( array_diff( $existing, $keep ) ); $missing = array_values( array_diff( $keep, $existing ) ); // A preserving reconcile on a table already in step is a pure read. A non-preserving one is a rewrite and never short-circuits. $in_step = [] === $surplus && [] === $missing; if ( $in_step && $preserve_progress ) { return [ 'written' => 0, 'dropped' => 0, ]; } $dropped = 0; foreach ( array_chunk( $surplus, self::PRUNE_CHUNK ) as $chunk ) { // phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required $dropped += (int) $wpdb->query( $wpdb->prepare( 'DELETE FROM %i WHERE migration_id = %d AND stage = %s AND virtual_path IN (' . implode( ', ', array_fill( 0, count( $chunk ), '%s' ) ) . ')', array_merge( [ $this->table, $migration_id, $stage ], array_values( $chunk ) ) ) ); } // An existing row records how far that file got, so a preserving reconcile writes only new paths. if ( ! $preserve_progress ) { $written = $this->ensure_seeded( $migration_id, $stage, $file_offsets ); } elseif ( [] !== $missing ) { $written = $this->upsert_changed( $migration_id, $stage, $file_offsets, $missing ); } else { $written = 0; } return [ 'written' => $written, 'dropped' => $dropped, ]; } /** * Pending/failed counts for one stage, so the worker can stop or advance without paging the manifest. * * @return array{pending: int, failed: int} */ public function stage_blocking( int $migration_id, string $stage ): array { $wpdb = $this->wpdb; // phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required $row = $wpdb->get_row( $wpdb->prepare( "SELECT SUM(status = 'pending') AS pending_count, SUM(status = 'failed') AS failed_count FROM %i WHERE migration_id = %d AND stage = %s", $this->table, $migration_id, $stage ), ARRAY_A ); return [ 'pending' => (int) ( $row['pending_count'] ?? 0 ), 'failed' => (int) ( $row['failed_count'] ?? 0 ), ]; } /** * Pending/failed counts across every stage, for the completion check. * * @return array{pending: int, failed: int} */ public function blocking( int $migration_id ): array { $wpdb = $this->wpdb; // phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required $row = $wpdb->get_row( $wpdb->prepare( "SELECT SUM(status = 'pending') AS pending_count, SUM(status = 'failed') AS failed_count FROM %i WHERE migration_id = %d", $this->table, $migration_id ), ARRAY_A ); return [ 'pending' => (int) ( $row['pending_count'] ?? 0 ), 'failed' => (int) ( $row['failed_count'] ?? 0 ), ]; } /** * Write or update one file's row, keyed on (migration_id, stage, virtual_path) so sibling rows are never rewritten. * * @param array $meta Engine offsets entry for the file. */ private function upsert_row( int $migration_id, string $stage, int $sequence, string $path, array $meta ): void { $wpdb = $this->wpdb; // Refuse by name rather than truncate: a truncated key is a different path the source won't serve, so the stage would finish silently without the file. if ( mb_strlen( $path ) > self::MAX_PATH_CHARS ) { // Exception messages go to the migration log, which escapes on output. // phpcs:disable WordPress.Security.EscapeOutput.ExceptionNotEscaped throw new \RuntimeException( sprintf( 'The path "%s" is %d characters long, more than the %d a transfer can track. Exclude it and run the migration again.', $path, mb_strlen( $path ), self::MAX_PATH_CHARS ) ); // phpcs:enable WordPress.Security.EscapeOutput.ExceptionNotEscaped } $size = max( 0, (int) ( $meta['size'] ?? 0 ) ); $offset = max( 0, (int) ( $meta['offset'] ?? 0 ) ); $done = ! empty( $meta['done'] ); $failed = ! empty( $meta['failed'] ); $status = $done && $failed ? 'failed' : ( $done ? ( ! empty( $meta['skipped'] ) ? 'skipped' : 'done' ) : 'pending' ); // phpcs:ignore WordPress.DB.DirectDatabaseQuery -- custom table, fresh read required $result = $wpdb->query( $wpdb->prepare( 'INSERT INTO %i (migration_id, stage, sequence, virtual_path, status, byte_total, byte_done, content_hash, errors, skip_batch, updated_at) VALUES (%d, %s, %d, %s, %s, %d, %d, %s, %d, %d, %s) ON DUPLICATE KEY UPDATE sequence = VALUES(sequence), status = VALUES(status), byte_total = VALUES(byte_total), byte_done = VALUES(byte_done), content_hash = VALUES(content_hash), errors = VALUES(errors), skip_batch = VALUES(skip_batch), updated_at = VALUES(updated_at)', $this->table, $migration_id, $stage, $sequence, $path, $status, $size, min( $offset, $size ), (string) ( $meta['hash'] ?? '' ), (int) ( $meta['errors'] ?? 0 ), ! empty( $meta['skip_batch'] ) ? 1 : 0, current_time( 'mysql', true ) ) ); // An unwritten row is a file the worker will never ask for again. if ( false === $result ) { // Escaped on output, never echoed raw. // phpcs:disable WordPress.Security.EscapeOutput.ExceptionNotEscaped throw new \RuntimeException( sprintf( 'Could not record transfer progress for "%s": %s', $path, (string) $wpdb->last_error ) ); // phpcs:enable WordPress.Security.EscapeOutput.ExceptionNotEscaped } } }