*/ private array $records = array(); /** * Next auto-increment id to allocate on insert. * * Starts at 1 and advances monotonically with every successful * insert. Mirrors the MySQL `AUTO_INCREMENT` convention: ids are * never reused, even after a delete. */ private int $nextId = 1; /** * Construct the repository, optionally seeded with a set of * already-persisted records. * * Seeding is primarily useful for tests that need to exercise * "existing environment" code paths without driving a sequence of * {@see self::save()} calls first. Seed records MUST carry a * non-null id; their ids are honored verbatim, and the repository's * auto-increment cursor advances past the highest seeded id so * subsequent inserts cannot collide with a seeded row. * * @param list $seed Records to preload. * * @throws InvalidArgumentException When a seed record has no id or * when two seeds share an id. */ public function __construct( array $seed = array() ) { foreach ( $seed as $record ) { $id = $record->id(); if ( $id === null ) { $message = 'InMemorySyncJobRepository seed records must carry a non-null id.'; // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML. throw new InvalidArgumentException( $message ); } if ( array_key_exists( $id, $this->records ) ) { $message = sprintf( 'InMemorySyncJobRepository seed contains duplicate id %d.', $id ); // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML. throw new InvalidArgumentException( $message ); } $this->records[ $id ] = $record; if ( $id >= $this->nextId ) { $this->nextId = $id + 1; } } } /** * Persist the record and return its post-save form. * * See {@see SyncJobRepository::save()} for the full contract. This * implementation: * * - Treats a null id as an insert, allocates the next * auto-increment id, and stores the record under the new id. * * - Treats a non-null id as an update. If no row currently * matches the id, raises {@see \RuntimeException} per the * "update-after-delete is a visible error" rule. * * @param SyncJobRecord $record Record to persist. * * @throws RuntimeException When an update targets a missing id. */ public function save( SyncJobRecord $record ): SyncJobRecord { $id = $record->id(); if ( $id === null ) { return $this->insert( $record ); } return $this->update( $id, $record ); } /** * Insert a fresh (unpersisted) record under a newly-allocated id. * * @param SyncJobRecord $record Record to insert. Must have a null id. */ private function insert( SyncJobRecord $record ): SyncJobRecord { $id = $this->nextId; ++$this->nextId; $persisted = $record->withId( $id ); $this->records[ $id ] = $persisted; return $persisted; } /** * Overwrite an existing row with the supplied record. * * @param int $id Primary key of the row being updated. * @param SyncJobRecord $record Record carrying the new column values. * * @throws RuntimeException When no row currently matches `$id`. */ private function update( int $id, SyncJobRecord $record ): SyncJobRecord { if ( ! array_key_exists( $id, $this->records ) ) { $message = sprintf( 'InMemorySyncJobRepository: cannot update missing id %d; row may have been deleted.', $id ); // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML. throw new RuntimeException( $message ); } $this->records[ $id ] = $record; return $record; } /** * Look up a record by its primary key. * * Returns null when no row matches; raises * {@see InvalidArgumentException} for non-positive ids so a * programming error at the call site surfaces loudly instead of * silently missing. * * @param int $id Primary key to look up. * * @throws InvalidArgumentException When `$id` is non-positive. */ public function findById( int $id ): ?SyncJobRecord { if ( $id <= 0 ) { $message = sprintf( 'InMemorySyncJobRepository::findById() requires a positive id, got %d.', $id ); // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML. throw new InvalidArgumentException( $message ); } return $this->records[ $id ] ?? null; } /** * Remove the row identified by the given id. * * Idempotent: returns true when a row was removed, false when no * row matched. Non-positive ids raise * {@see InvalidArgumentException} so a programming error at the * call site surfaces loudly instead of silently missing. * * @param int $id Primary key of the row to delete. * * @throws InvalidArgumentException When `$id` is non-positive. */ public function delete( int $id ): bool { if ( $id <= 0 ) { $message = sprintf( 'InMemorySyncJobRepository::delete() requires a positive id, got %d.', $id ); // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML. throw new InvalidArgumentException( $message ); } if ( ! array_key_exists( $id, $this->records ) ) { return false; } unset( $this->records[ $id ] ); return true; } /** * Total number of rows currently stored. */ public function count(): int { return count( $this->records ); } /** * Return every stored record, ordered by ascending primary key. * * Exposed purely as a convenience for test assertions that want * to enumerate the repository contents without driving a * paginated listing. Production code MUST NOT rely on this method * because it has no counterpart on the {@see SyncJobRepository} * contract and the Wpdb-backed implementation does not offer it. * * @return list */ public function all(): array { $ordered = $this->records; ksort( $ordered, SORT_NUMERIC ); return array_values( $ordered ); } /** * Next auto-increment id the repository would allocate on the * following insert. * * Exposed purely as a convenience for test assertions that want to * probe the internal cursor without issuing a dummy insert. Like * {@see self::all()}, production code MUST NOT rely on this. */ public function nextId(): int { return $this->nextId; } /** * Look up a paginated, filtered slice of stored rows. * * See {@see SyncJobRepository::findPaginated()} for the contract. * Filtering is applied per non-null axis on `$query`; the * remaining rows are sorted by descending id (a proxy for * `created_at DESC, id DESC` since this double allocates ids * monotonically) and sliced to the requested page. * * @param SyncJobListQuery $query Validated list-query inputs. */ public function findPaginated( SyncJobListQuery $query ): SyncJobListResult { $matching = array(); foreach ( $this->records as $record ) { if ( ! $this->matchesQuery( $record, $query ) ) { continue; } $matching[] = $record; } usort( $matching, static function ( SyncJobRecord $a, SyncJobRecord $b ): int { return ( $b->id() ?? 0 ) <=> ( $a->id() ?? 0 ); } ); $total = count( $matching ); $offset = $query->offset(); $perPage = $query->perPage(); $slice = array_slice( $matching, $offset, $perPage ); /** @var list $slice */ return new SyncJobListResult( $slice, $total, $query->page(), $perPage ); } /** * Group every stored row by DB-vocabulary status and return the * per-status counts. * * See {@see SyncJobRepository::countByStatus()} for the contract; * in particular, statuses with zero occurrences are omitted so * callers default-fill against * {@see SyncJobListQuery::VALID_STATUSES}. * * @return array */ public function countByStatus(): array { $counts = array(); foreach ( $this->records as $record ) { $status = $record->status(); $counts[ $status ] = ( $counts[ $status ] ?? 0 ) + 1; } return $counts; } /** * Find the oldest queued/running sync_product row for a product id. * * @param int $productId Local product id carried in payload_json. */ public function findOpenProductSync( int $productId ): ?SyncJobRecord { if ( $productId <= 0 ) { $message = sprintf( 'InMemorySyncJobRepository::findOpenProductSync() requires a positive productId, got %d.', $productId ); // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML. throw new InvalidArgumentException( $message ); } $matches = array(); foreach ( $this->records as $record ) { if ( $record->jobType() !== 'sync_product' ) { continue; } if ( ! in_array( $record->status(), array( SyncJobRecord::DEFAULT_STATUS, SyncJobRecord::STATUS_RUNNING ), true ) ) { continue; } $payload = json_decode( $record->payloadJson(), true ); if ( ! is_array( $payload ) || (int) ( $payload['product_id'] ?? 0 ) !== $productId ) { continue; } $matches[] = $record; } if ( $matches === array() ) { return null; } usort( $matches, static function ( SyncJobRecord $a, SyncJobRecord $b ): int { $cmp = $a->createdAt()->getTimestamp() <=> $b->createdAt()->getTimestamp(); if ( $cmp !== 0 ) { return $cmp; } return ( $a->id() ?? 0 ) <=> ( $b->id() ?? 0 ); } ); return $matches[0]; } /** * Whether `$record` satisfies every non-null filter axis on * `$query`. * * @param SyncJobRecord $record Record under test. * @param SyncJobListQuery $query Filter axes. */ private function matchesQuery( SyncJobRecord $record, SyncJobListQuery $query ): bool { if ( $query->status() !== null && $record->status() !== $query->status() ) { return false; } if ( $query->jobType() !== null && $record->jobType() !== $query->jobType() ) { return false; } return true; } /** * Atomically claim the next runnable job per * {@see SyncJobRepository::claimNext()}. * * Pure in-memory single-threaded implementation: scans every * stored record, filters by §8.5 eligibility (pending + * `available_at <= now`, or running + `locked_at < * lockTimeoutCutoff`), sorts by §13.2 ordering (priority asc, * available_at asc, id asc), and transitions the head of the * sorted list via {@see SyncJobRecord::withClaim()}. Returns null * when no candidate qualifies. * * @param string $workerId Lock-holder identifier. * @param DateTimeImmutable $now Current UTC instant. * @param DateTimeImmutable $lockTimeoutCutoff Reclaim threshold. * * @throws InvalidArgumentException When `$workerId` is empty or > 64 bytes. * @throws RuntimeException When a candidate record has a null id * (storage invariant violated). */ public function claimNext( string $workerId, DateTimeImmutable $now, DateTimeImmutable $lockTimeoutCutoff ): ?SyncJobRecord { $length = strlen( $workerId ); if ( $length === 0 || $length > 64 ) { $message = sprintf( 'InMemorySyncJobRepository::claimNext() requires workerId of 1-64 bytes, got %d.', $length ); // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML. throw new InvalidArgumentException( $message ); } $nowTs = $now->getTimestamp(); $cutoffTs = $lockTimeoutCutoff->getTimestamp(); $candidates = array(); foreach ( $this->records as $record ) { $status = $record->status(); if ( $status === SyncJobRecord::DEFAULT_STATUS ) { if ( $record->availableAt()->getTimestamp() <= $nowTs ) { $candidates[] = $record; } continue; } if ( $status === SyncJobRecord::STATUS_RUNNING ) { $lockedAt = $record->lockedAt(); if ( $lockedAt !== null && $lockedAt->getTimestamp() < $cutoffTs ) { $candidates[] = $record; } } } if ( $candidates === array() ) { return null; } usort( $candidates, static function ( SyncJobRecord $a, SyncJobRecord $b ): int { $cmp = $a->priority() <=> $b->priority(); if ( $cmp !== 0 ) { return $cmp; } $cmp = $a->availableAt()->getTimestamp() <=> $b->availableAt()->getTimestamp(); if ( $cmp !== 0 ) { return $cmp; } return ( $a->id() ?? 0 ) <=> ( $b->id() ?? 0 ); } ); $head = $candidates[0]; $id = $head->id(); if ( $id === null ) { $message = 'InMemorySyncJobRepository::claimNext(): candidate had a null id; storage invariant violated.'; // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped -- Developer-facing message, never rendered to HTML. throw new RuntimeException( $message ); } $claimed = $head->withClaim( $workerId, $now ); $this->records[ $id ] = $claimed; return $claimed; } }