{

/** @var PDO */

private $pdo;

private const TABLE = 'queue';

private const SCHEMA = "

CREATE TABLE IF NOT EXISTS " . self::TABLE . " (

ON " . self::TABLE . "(processed, in_progress, status);

";

private function __construct(PDO $pdo)

{

$this->pdo = $pdo;

}

/**

public static function open(string $dbPath): self

{

$pdo = new PDO('sqlite:' . $dbPath, null, null, [

PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,

PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,

PDO::ATTR_EMULATE_PREPARES => false,

]);

// Performance pragmas suitable for a single-writer CLI job

$pdo->exec("PRAGMA journal_mode=WAL;");

$pdo->exec("PRAGMA synchronous=NORMAL;");

$pdo->exec("PRAGMA temp_store=MEMORY;");

$pdo->exec("PRAGMA foreign_keys=ON;");

// Schema

foreach (array_filter(array_map('trim', explode(';', self::SCHEMA))) as $stmt) {

if ($stmt !== '') {

$pdo->exec($stmt . ';');

}

}

$self = new self($pdo);

// Single worker: clear any half-done rows

$self->clearInProgressOnStart();

return $self;

}

/**

public function fetchNext(): ?array

{

// BEGIN IMMEDIATE prevents write races if you ever parallelize

$this->pdo->beginTransaction();

$id = $this->pdo->query("

FROM " . self::TABLE . "

")->fetchColumn();

if ($id === false) {

$this->pdo->commit();

return null;

}

$now = $this->now();

$upd = $this->pdo->prepare("

UPDATE " . self::TABLE . "

");

$upd->execute([':id' => $id, ':now' => $now]);

if ($upd->rowCount() !== 1) {

// Lost a race; nothing to do

$this->pdo->commit();

return null;

}

$stmt = $this->pdo->prepare("SELECT * FROM " . self::TABLE . " WHERE id=:id");

$stmt->execute([':id' => $id]);

$row = $stmt->fetch();

$this->pdo->commit();

return $row ?: null;

}

public function markSuccess(string $id): void

{

$stmt = $this->pdo->prepare("

UPDATE " . self::TABLE . "

");

$stmt->execute([':id' => $id, ':now' => $this->now()]);

}

public function markFailure(string $id, string $error): void

{

$stmt = $this->pdo->prepare("

UPDATE " . self::TABLE . "

");

$stmt->execute([

':id' => $id,

':now' => $this->now(),

':err' => str_replace(["\r", "\n"], ' ', $error),

]);

}

/**

public function updateById(string $id, array $fields): void

{

// core columns we store as columns

$allowed = ['processed', 'in_progress', 'status', 'result', 'updated_at', 'last_error', 'extras'];

// Split into column updates and extras updates

$extrasDelta = [];

foreach ($fields as $k => $v) {

if (!in_array($k, $allowed, true)) {

$extrasDelta[$k] = $v;

unset($fields[$k]);

}

}

// If caller passed an 'extras' JSON string/array, normalize and merge with delta

if (array_key_exists('extras', $fields)) {

$base = $fields['extras'];

$baseArr = is_array($base) ? $base : (json_decode((string)$base, true) ?: []);

$fields['extras'] = $baseArr; // keep as array for now

} else {

$fields['extras'] = []; // start empty, maybe merge delta below

}

if ($extrasDelta) {

// Load current extras and merge

$stmt = $this->pdo->prepare("SELECT extras FROM " . self::TABLE . " WHERE id=:id");

$stmt->execute([':id' => $id]);

$cur = $stmt->fetchColumn();

$curArr = $cur ? (json_decode((string)$cur, true) ?: []) : [];

$fields['extras'] = array_replace($curArr, (array)$fields['extras'], $extrasDelta);

}

// Build SQL

$set = [];

$params = [':id' => $id];

foreach ($fields as $k => $v) {

if (!in_array($k, $allowed, true)) {

continue;

}

if ($k === 'extras') {

$params[':extras'] = empty($v) ? null : json_encode($v, JSON_UNESCAPED_UNICODE);

$set[] = "extras = :extras";

} else {

$set[] = "$k = :$k";

$params[":$k"] = $v;

}

}

if (!$set) {

return;

}

$sql = "UPDATE " . self::TABLE . " SET " . implode(', ', $set) . " WHERE id=:id";

$stmt = $this->pdo->prepare($sql);

$stmt->execute($params);

}

/**

public function resetAll(): void

{

$this->pdo->exec("

UPDATE " . self::TABLE . "

");

}

/**

public function retryErrors(): int

{

$stmt = $this->pdo->prepare("

UPDATE " . self::TABLE . "

");

$stmt->execute();

return $stmt->rowCount();

}

/**

public function counts(): array

{

$q = $this->pdo->query("

FROM " . self::TABLE . "

");

$r = $q->fetch();

return array_map('intval', $r ?: ['done' => 0, 'queued' => 0, 'errors' => 0, 'in_progress' => 0]);

}

public function progress(): array

{

$row = $this->pdo->query("

FROM " . self::TABLE . "

")->fetch(PDO::FETCH_ASSOC);

$row = array_map('intval', $row);

$row['remaining'] = $row['total'] - $row['done'] - $row['errors'] - $row['in_progress'];

return $row;

}

public function totalItems(): int

{

return (int)$this->pdo->query("SELECT COUNT(*) FROM " . self::TABLE . "")->fetchColumn();

}

public function vacuum(bool $truncateWal = true): void

{

// If a transaction is open, commit it so VACUUM can run

if ($this->pdo->inTransaction()) {

$this->pdo->commit();

}

// In WAL mode, checkpoint first so the main DB can be compacted

$mode = strtolower((string)$this->pdo->query("PRAGMA journal_mode;")->fetchColumn());

if ($mode === 'wal') {

$this->pdo->exec(

$truncateWal

? "PRAGMA wal_checkpoint(TRUNCATE);"

: "PRAGMA wal_checkpoint(FULL);"

);

}

// Compact main database and refresh stats

$this->pdo->exec("VACUUM;");

$this->pdo->exec("ANALYZE;"); // optional but helpful after big changes

}

/**

public function deleteDatabase(): void

{

// Must commit any open transaction first

if ($this->pdo->inTransaction()) {

$this->pdo->commit();

}

// Determine the physical file path

$file = $this->pdo->query("PRAGMA database_list;")->fetch(PDO::FETCH_ASSOC)['file'] ?? null;

if ($file && is_file($file)) {

// Close the PDO handle explicitly

$this->pdo = null;

// Remove WAL and SHM side files if present

@unlink($file . '-wal');

@unlink($file . '-shm');

@unlink($file);

} else {

throw new RuntimeException("SQLite database file not found or already deleted.");

}

}

/**

public function dropAndRecreate(): void

{

if ($this->pdo->inTransaction()) {

$this->pdo->commit();

}

// Drop old data table

$this->pdo->exec("DROP TABLE IF EXISTS queue;");

// Recreate schema exactly as at startup

foreach (array_filter(array_map('trim', explode(';', self::SCHEMA))) as $stmt) {

if ($stmt !== '') {

$this->pdo->exec($stmt . ';');

}

}

// Reset journal metadata for compactness

$this->pdo->exec("PRAGMA wal_checkpoint(TRUNCATE);");

$this->pdo->exec("VACUUM;");

$this->pdo->exec("ANALYZE;");

}

// internals

private function clearInProgressOnStart(): void

{

$this->pdo->exec("

UPDATE " . self::TABLE . "

");

}

private function now(): string

{

return date('Y-m-d H:i:s');

}

public function setExtras(string $id, array $extras): void

{

$this->updateById($id, ['extras' => $extras]);

}

public function getExtras(string $id): array

{

$stmt = $this->pdo->prepare("SELECT extras FROM " . self::TABLE . " WHERE id=:id");

$stmt->execute([':id' => $id]);

$json = $stmt->fetchColumn();

return $json ? (json_decode((string)$json, true) ?: []) : [];

}

/**

public function transaction(\Closure $callback)

{

if ($this->pdo->inTransaction()) {

return $callback($this->pdo);

}

$this->pdo->beginTransaction();

try {

$result = $callback($this->pdo);

$this->pdo->commit();

return $result;

} catch (\Throwable $e) {

$this->pdo->rollBack();

throw $e; // Re-throw the exception

}

}

public function upsert(string $id, array $coreData, array $extrasData, bool $updateCore): void

{

$ins = $this->pdo->prepare("INSERT OR IGNORE INTO " . self::TABLE . "(id) VALUES(:id)");

$ins->execute([':id' => $id]);

// This part for handling extras is correct.

if (!empty($extrasData)) {

$curStmt = $this->pdo->prepare("SELECT extras FROM " . self::TABLE . " WHERE id=:id");

$curStmt->execute([':id' => $id]);

$curJson = $curStmt->fetchColumn();

if ($curJson) {

$curArr = json_decode((string)$curJson, true) ?: [];

$extrasData = array_replace($curArr, $extrasData);

}

}

$extrasJson = empty($extrasData) ? null : json_encode($extrasData, JSON_UNESCAPED_UNICODE);

$params = [':id' => $id];

$setClauses = [];

// Always include extras in the update, even if null, to clear it if needed.

$setClauses[] = 'extras = :extras';

$params[':extras'] = $extrasJson;

if ($updateCore) {

// Check if the value is null before adding it to the UPDATE statement.

foreach ($coreData as $key => $value) {

if ($value !== null) {

$setClauses[] = "$key = :$key";

$params[":$key"] = $value;

}

}

}

// Only run the UPDATE statement if there is something to update.

// (We always update extras, so the count will usually be > 0)

if (!empty($setClauses)) {

$sql = "UPDATE " . self::TABLE . " SET " . implode(', ', $setClauses) . " WHERE id=:id";

$upd = $this->pdo->prepare($sql);

$upd->execute($params);

}

}

public function performBatch(\Closure $callback)

{

if ($this->pdo->inTransaction()) {

return $callback($this);

}

$this->pdo->beginTransaction();

try {

// We pass the queue instance itself to the callback

$result = $callback($this);

$this->pdo->commit();

return $result;

} catch (\Throwable $e) {

$this->pdo->rollBack();

throw $e;

}

}

public function fetchAllForExport(): iterable

{

$stmt = $this->pdo->query("SELECT * FROM " . self::TABLE . " ORDER BY rowid");

while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {

$extras = [];

if (!empty($row['extras'])) {

$extras = json_decode((string)$row['extras'], true) ?: [];

}

unset($row['extras']); // Avoid duplication

// Yield a single, merged array for each job

yield array_replace($extras, $row);

}

}