Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
100.00%
59 / 59
100.00%
9 / 9
 CRAP
100.00%
1 / 1
Database
100.00%
59 / 59
100.00%
9 / 9
 16
100.00%
1 / 1
 __construct
100.00%
4 / 4
100.00%
1 / 1
 2
 getDb
100.00%
1 / 1
100.00%
1 / 1
 1
 getTable
100.00%
1 / 1
100.00%
1 / 1
 1
 write
100.00%
14 / 14
100.00%
1 / 1
 1
 read
100.00%
7 / 7
100.00%
1 / 1
 2
 all
100.00%
9 / 9
100.00%
1 / 1
 3
 delete
100.00%
5 / 5
100.00%
1 / 1
 1
 purgeUndecodable
100.00%
10 / 10
100.00%
1 / 1
 4
 createTable
100.00%
8 / 8
100.00%
1 / 1
 1
1<?php
2declare(strict_types=1);
3/**
4 * Pop PHP Framework (https://www.popphp.org/)
5 *
6 * @link       https://github.com/popphp/popphp-framework
7 * @author     Nick Sagona, III <nick@popphp.org>
8 * @copyright  Copyright (c) 2009-2026 Nick Sagona, III
9 * @license    https://www.popphp.org/license     New BSD License
10 */
11
12/**
13 * @namespace
14 */
15namespace Pop\Queue\Registry\Adapter;
16
17use Pop\Db\Adapter\AbstractAdapter as DbAdapter;
18use Pop\Queue\Registry\AbstractRegistry;
19use Pop\Queue\Registry\WorkerRecord;
20
21/**
22 * Database registry class
23 *
24 * One row per worker: the ID as primary key, last_seen_at as its own column
25 * so the table is directly queryable by an operator, and the full record as
26 * JSON.
27 *
28 * prune() is inherited as a read-filter-delete loop rather than overridden
29 * with a bulk DELETE: pop-db's string where() parser is only reliable for
30 * simple expressions and this codebase has no precedent for a string '<'
31 * comparison (src/Adapter/Database.php uses the lessThanOrEqualTo()
32 * predicate API wherever it needs one). A registry holds tens of rows, so
33 * this costs nothing real.
34 *
35 * @category   Pop
36 * @package    Pop\Queue
37 * @author     Nick Sagona, III <nick@popphp.org>
38 * @copyright  Copyright (c) 2009-2026 Nick Sagona, III
39 * @license    https://www.popphp.org/license     New BSD License
40 * @version    3.0.0
41 */
42class Database extends AbstractRegistry
43{
44
45    /**
46     * Database adapter
47     * @var ?DbAdapter
48     */
49    protected ?DbAdapter $db = null;
50
51    /**
52     * Table name
53     * @var ?string
54     */
55    protected ?string $table = null;
56
57    /**
58     * Constructor
59     *
60     * @param DbAdapter $db
61     * @param string    $table
62     */
63    public function __construct(DbAdapter $db, string $table = 'pop_worker_registry')
64    {
65        $this->db    = $db;
66        $this->table = $table;
67
68        if (!$this->db->hasTable($table)) {
69            $this->createTable($table);
70        }
71    }
72
73    /**
74     * Get the database adapter
75     *
76     * @return ?DbAdapter
77     */
78    public function getDb(): ?DbAdapter
79    {
80        return $this->db;
81    }
82
83    /**
84     * Get the table name
85     *
86     * @return ?string
87     */
88    public function getTable(): ?string
89    {
90        return $this->table;
91    }
92
93    public function write(WorkerRecord $record): void
94    {
95        // Upsert without relying on dialect-specific ON CONFLICT syntax:
96        // delete any existing row for this ID, then insert. Each worker
97        // writes only its own row, so there is no cross-worker race here.
98        $this->delete($record->getId());
99
100        $sql = $this->db->createSql();
101        $sql->insert($this->table)->values([
102            'id'           => ':id',
103            'last_seen_at' => ':last_seen_at',
104            'record'       => ':record'
105        ]);
106
107        $this->db->prepare($sql);
108        $this->db->bindParams([
109            'id'           => $record->getId(),
110            'last_seen_at' => $record->getLastSeenAt(),
111            'record'       => $this->encode($record)
112        ]);
113        $this->db->execute();
114    }
115
116    public function read(string $id): ?WorkerRecord
117    {
118        $sql = $this->db->createSql();
119        $sql->select('record')->from($this->table)->where('id = :id');
120        $this->db->prepare($sql);
121        $this->db->bindParams(['id' => $id]);
122        $this->db->execute();
123        $rows = $this->db->fetchAll();
124
125        return isset($rows[0]['record']) ? $this->decode($rows[0]['record']) : null;
126    }
127
128    public function all(): array
129    {
130        $sql = $this->db->createSql();
131        $sql->select('record')->from($this->table);
132        $this->db->query($sql);
133
134        $records = [];
135        foreach ($this->db->fetchAll() as $row) {
136            $record = $this->decode($row['record'] ?? null);
137            if ($record !== null) {
138                $records[$record->getId()] = $record;
139            }
140        }
141
142        return $records;
143    }
144
145    public function delete(string $id): void
146    {
147        $sql = $this->db->createSql();
148        $sql->delete()->from($this->table)->where('id = :id');
149        $this->db->prepare($sql);
150        $this->db->bindParams(['id' => $id]);
151        $this->db->execute();
152    }
153
154    protected function purgeUndecodable(): int
155    {
156        $sql = $this->db->createSql();
157        $sql->select(['id', 'record'])->from($this->table);
158        $this->db->query($sql);
159
160        $bad = [];
161        foreach ($this->db->fetchAll() as $row) {
162            if ($this->decode($row['record'] ?? null) === null) {
163                $bad[] = (string)$row['id'];
164            }
165        }
166
167        foreach ($bad as $id) {
168            $this->delete($id);
169        }
170
171        return count($bad);
172    }
173
174    /**
175     * Create the registry table
176     *
177     * @param  string $table
178     * @return Database
179     */
180    public function createTable(string $table): Database
181    {
182        $schema = $this->db->createSchema();
183
184        $schema->create($table)
185            ->varchar('id', 255)
186            ->int('last_seen_at', 16)
187            ->text('record')
188            ->primary('id');
189
190        $this->db->query($schema);
191
192        return $this;
193    }
194
195}