<?php
/**
* SeekQuarry/Yioop --
* Open Source Pure PHP Search Engine, Crawler, and Indexer
*
* Copyright (C) 2009 - 2026 Chris Pollett chris@pollett.org
*
* LICENSE:
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*
* END LICENSE
*
* @author Chris Pollett chris@pollett.org
* @license https://www.gnu.org/licenses/ GPL3
* @link https://www.seekquarry.com/
* @copyright 2009 - 2026
* @filesource
*/
namespace seekquarry\yioop\models;
use seekquarry\yioop\configs as C;
/**
* Persists MailSite clone jobs and their per-message dedup state.
* Backs the IMAP-clone-to-local feature (item 48 follow-up, v10
* todo): a Yioop user submits a clone of one of their configured
* IMAP accounts into their MailSite mailbox; the MediaUpdater-
* driven MailCloneJob picks up the row and walks the source
* IMAP server folder by folder, message by message, calling
* appendMessage on the destination FileMailStorage. State that
* needs to survive between MediaUpdater ticks lives in
* MAIL_CLONE_JOB; the per-message imported set that prevents
* double-import on resume / re-run lives in MAIL_CLONE_SEEN.
*
* Two tables managed here:
*
* MAIL_CLONE_JOB
* One row per clone submission. STATUS transitions: pending
* -> running -> done (or failed or cancelled). FOLDERS_DONE
* is a JSON-encoded list of folder names already fully
* drained; CURRENT_FOLDER + CURRENT_UID together form the
* resume cursor inside the current folder, and the two
* CURRENT_UID_PARTIAL_* columns extend the cursor across
* ticks for very large messages (>2 MB) that ratchet through
* in BODY.PEEK[<offset.size>] chunks instead of fetching
* whole.
*
* MAIL_CLONE_SEEN
* Composite key (JOB_ID, SRC_UIDVALIDITY, SRC_UID). Records
* every message the job has appended to the destination
* successfully. The job consults this set before fetching
* each source UID; presence means skip. UIDVALIDITY plus
* UID is the RFC 3501 stable identifier (unchanged across
* IMAP sessions unless the server rebuilds the mailbox),
* which makes the dedup key cheap and exact. Rows are
* garbage-collected once the parent job's row has been in
* STATUS='done' for more than 30 days.
*
* Access control: every public method takes $user_id and
* refuses to surface rows belonging to a different user. The
* one exception is listClaimable(), which crosses the user
* boundary because the MediaUpdater process needs to enumerate
* runnable jobs across all users; only daemon code is expected
* to call it, never a request handler.
*
* @author Chris Pollett
*/
class MailCloneModel extends Model
{
/**
* Inserts a fresh pending clone-job row and returns its
* generated ID. Callers (the userMailStartClone handler)
* have already validated the source-account ownership and
* destination-user match; this method does not re-check.
* STARTED_AT and UPDATED_AT are stamped to the current
* wall-clock time so the MediaUpdater's claim ordering sees
* the row immediately. FOLDERS_DONE is initialized to the
* empty JSON array '[]' so the job loop's
* !in_array(folder, decode(FOLDERS_DONE)) test is well-
* defined on the very first folder.
*
* @param int $user_id Yioop USER_ID submitting the clone
* @param int $src_account MAIL_ACCOUNT.ID of the IMAP source
* @param string $destination_user MailSite local-part to
* write into (typically the user's own Yioop username)
* @param string $mode 'add' (preserve existing destination
* mail) or 'wipe' (purge destination first)
* @param bool $dry_run true to simulate without actually
* writing to the destination FileMailStorage
* @param int $folder_cap newest-N-per-folder limit, or 0 for
* no limit (clone the entire folder)
* @return int the new MAIL_CLONE_JOB.ID. The PdoManager's
* insertID() returns a string (PDO::lastInsertId
* always returns a string per upstream contract); this
* method casts to int so callers can compare with
* strict-mode in_array against int-cast row IDs.
*/
public function insertJob($user_id, $src_account,
$destination_user, $mode, $dry_run = false,
$folder_cap = 0)
{
$now = time();
$sql = "INSERT INTO MAIL_CLONE_JOB " .
"(USER_ID, SRC_ACCOUNT, DESTINATION_USER, MODE, " .
"STATUS, DRY_RUN, FOLDERS_DONE, CURRENT_FOLDER, " .
"CURRENT_UID, CURRENT_UID_PARTIAL_OFFSET, " .
"CURRENT_UID_PARTIAL_SIZE, IMPORTED, SKIPPED, " .
"FAILED, FOLDER_CAP, STARTED_AT, UPDATED_AT) " .
"VALUES (?, ?, ?, ?, 'pending', ?, '[]', '', 0, 0, " .
"0, 0, 0, 0, ?, ?, ?)";
$this->db->execute($sql, [$user_id, $src_account,
$destination_user, $mode, $dry_run ? 1 : 0,
(int) $folder_cap, $now, $now]);
return (int) $this->db->insertID('MAIL_CLONE_JOB');
}
/**
* Returns the most recent pending or running clone job for
* a user, or null when none exists. Used both by the UI to
* render the per-user "clone in progress" status banner and
* by the start-clone handler to enforce the single-job-per-
* user limit. Sorted by STARTED_AT descending so a brand-new
* pending row wins if the MediaUpdater has not picked it up
* yet.
*
* @param int $user_id Yioop USER_ID to look up for
* @return array|null the row as an associative array, or
* null when the user has no in-flight clone job
*/
public function activeJobForUser($user_id)
{
$sql = "SELECT * FROM MAIL_CLONE_JOB " .
"WHERE USER_ID = ? AND STATUS IN ('pending', " .
"'running') ORDER BY STARTED_AT DESC " .
$this->db->limitOffset(1);
$result = $this->db->execute($sql, [$user_id]);
if ($result === false) {
return null;
}
$row = $this->db->fetchArray($result);
return ($row === false || empty($row)) ? null : $row;
}
/**
* Returns the user's most recent clone job when (and only
* when) that latest job is in the 'failed' state, otherwise
* null. Drives the retry banner: a failed clone would
* otherwise vanish from the UI (activeJobForUser surfaces only
* pending/running jobs, deliberately, so the single-job limit
* and the sidebar cloning badge stay scoped to live work).
* Keying on the single most-recent row means starting a fresh
* clone after a failure -- which inserts a newer pending row --
* immediately stops surfacing the old failed one, and a job
* that later succeeded or was cancelled never shows here.
*
* @param int $user_id Yioop USER_ID to look up for
* @return array|null the failed job row, or null
*/
public function recentFailedJobForUser($user_id)
{
$sql = "SELECT * FROM MAIL_CLONE_JOB WHERE USER_ID = ? " .
"ORDER BY STARTED_AT DESC " . $this->db->limitOffset(1);
$result = $this->db->execute($sql, [$user_id]);
if ($result === false) {
return null;
}
$row = $this->db->fetchArray($result);
if ($row === false || empty($row) ||
$row['STATUS'] !== 'failed') {
return null;
}
return $row;
}
/**
* Returns the row for a specific clone-job id, scoped to the
* supplied user. Returns null if the row does not exist or
* belongs to a different user; the access-control check is
* part of the SELECT predicate, not a separate compare, so
* the caller cannot accidentally leak rows by forgetting to
* verify ownership after fetch.
*
* @param int $job_id MAIL_CLONE_JOB.ID to fetch
* @param int $user_id Yioop USER_ID that should own the row
* @return array|null the row as an associative array, or
* null when not found or not owned by the user
*/
public function getJob($job_id, $user_id)
{
$sql = "SELECT * FROM MAIL_CLONE_JOB " .
"WHERE ID = ? AND USER_ID = ? " .
$this->db->limitOffset(1);
$result = $this->db->execute($sql, [$job_id, $user_id]);
if ($result === false) {
return null;
}
$row = $this->db->fetchArray($result);
return ($row === false || empty($row)) ? null : $row;
}
/**
* Returns the rows the MediaUpdater can pick up on the next
* tick: every job whose STATUS is 'pending' or 'running',
* across all users. Daemon-only entrypoint (the MailCloneJob
* tick loop); never call from a request handler because the
* result crosses the user-scope boundary on purpose. Order
* is STARTED_AT ascending so an older pending job is
* preferred over a newer one when only one tick of work
* fits.
*
* @return array list of clone-job rows; empty when no jobs
* are eligible
*/
public function listClaimable()
{
$sql = "SELECT * FROM MAIL_CLONE_JOB " .
"WHERE STATUS = ? OR STATUS = ? " .
"ORDER BY STARTED_AT ASC";
$result = $this->db->execute($sql,
['pending', 'running']);
if ($result === false) {
return [];
}
$rows = [];
while ($row = $this->db->fetchArray($result)) {
$rows[] = $row;
}
return $rows;
}
/**
* Writes a partial-progress update to an in-flight job. The
* caller is the MailCloneJob tick code, which has already
* decided the next cursor position based on what was just
* processed; this method just stamps it into the row plus
* the running counters and UPDATED_AT. Passing null for any
* counter leaves it unchanged (so a partial-message progress
* tick can advance the cursor without bumping IMPORTED).
*
* @param int $job_id MAIL_CLONE_JOB.ID being updated
* @param array $fields associative array of column => value
* pairs to write; keys must come from the allow-list
* ALLOWED_PROGRESS_FIELDS (this method silently drops
* anything else, so the caller cannot accidentally
* mutate USER_ID, SRC_ACCOUNT, STATUS, or MODE)
* @return bool whether the UPDATE ran
*/
public function updateProgress($job_id, $fields)
{
$allowed = ['CURRENT_FOLDER', 'CURRENT_UID',
'CURRENT_UID_PARTIAL_OFFSET',
'CURRENT_UID_PARTIAL_SIZE', 'IMPORTED', 'SKIPPED',
'FAILED', 'FOLDERS_DONE'];
$assignments = [];
$values = [];
foreach ($fields as $key => $value) {
if (!in_array($key, $allowed, true)) {
continue;
}
$assignments[] = "$key = ?";
$values[] = $value;
}
if (empty($assignments)) {
return false;
}
$assignments[] = "UPDATED_AT = ?";
$values[] = time();
$values[] = $job_id;
$sql = "UPDATE MAIL_CLONE_JOB SET " .
implode(", ", $assignments) . " WHERE ID = ?";
return $this->db->execute($sql, $values) !== false;
}
/**
* Transitions a job's STATUS column to one of the terminal
* or in-flight values: pending, running, done, failed,
* cancelled. Bumps UPDATED_AT in the same statement.
* Separate from updateProgress so the STATUS column is not
* in the progress-fields allow-list -- callers have to
* decide explicitly to change run-state, not slip it in
* with a counter bump.
*
* @param int $job_id MAIL_CLONE_JOB.ID being transitioned
* @param string $status new STATUS value
* @return bool whether the UPDATE ran
*/
public function setStatus($job_id, $status)
{
$sql = "UPDATE MAIL_CLONE_JOB SET STATUS = ?, " .
"UPDATED_AT = ? WHERE ID = ?";
return $this->db->execute($sql,
[$status, time(), $job_id]) !== false;
}
/**
* Cancels a user's job by id. STATUS goes to 'cancelled' so
* the MailCloneJob tick loop drops it on next entry. The
* user-scope predicate in the UPDATE means a cancel cannot
* affect any other user's job by id-guess. Returns true
* even when zero rows match (e.g. the job already finished
* or never existed for this user) -- the caller's intent
* was "make sure this job is not running for me", which the
* post-condition satisfies either way.
*
* @param int $job_id MAIL_CLONE_JOB.ID to cancel
* @param int $user_id Yioop USER_ID that should own the row
* @return bool whether the UPDATE ran (not whether it
* matched a row)
*/
public function cancelJob($job_id, $user_id)
{
$sql = "UPDATE MAIL_CLONE_JOB SET STATUS = 'cancelled', " .
"UPDATED_AT = ? WHERE ID = ? AND USER_ID = ? AND " .
"STATUS IN ('pending', 'running')";
return $this->db->execute($sql,
[time(), $job_id, $user_id]) !== false;
}
/**
* Requeues a failed clone job so the MediaUpdater picks it up
* again. Flips STATUS from 'failed' back to 'pending'; the job
* resumes from its checkpointed progress (CURRENT_FOLDER /
* CURRENT_UID / FOLDERS_DONE) and skips already-imported
* messages via MAIL_CLONE_SEEN, so a retry costs only the work
* not yet done. Scoped to the owning user and gated on the
* current status being 'failed' so it cannot disturb a job
* that is pending, running, done, or cancelled.
*
* @param int $job_id MAIL_CLONE_JOB.ID to requeue
* @param int $user_id owning USER_ID, for authorization
* @return bool whether the requeue UPDATE ran
*/
public function retryJob($job_id, $user_id)
{
$sql = "UPDATE MAIL_CLONE_JOB SET STATUS = 'pending', " .
"UPDATED_AT = ? WHERE ID = ? AND USER_ID = ? AND " .
"STATUS = 'failed'";
return $this->db->execute($sql,
[time(), $job_id, $user_id]) !== false;
}
/**
* Marks one source UID as already-imported under a job, so
* a resume tick skips it. Idempotent: the composite primary
* key on MAIL_CLONE_SEEN (JOB_ID, SRC_UIDVALIDITY, SRC_UID)
* makes a second insert of the same triple harmless under
* the portable insertIgnore wrapper (INSERT IGNORE on MySQL,
* INSERT OR IGNORE on SQLite, INSERT ... ON CONFLICT DO
* NOTHING on Postgres/PDO). The dedicated isSeen precheck
* the older implementation did is no longer needed since
* insertIgnore handles the duplicate case directly; one
* trip to the DB instead of two.
*
* @param int $job_id MAIL_CLONE_JOB.ID this dedup row belongs to
* @param int $uidvalidity source folder's UIDVALIDITY
* @param int $uid source message's UID within that folder
* @return bool whether the row is now present
*/
public function markSeen($job_id, $uidvalidity, $uid)
{
$sql = $this->db->insertIgnore(
"INSERT INTO MAIL_CLONE_SEEN (JOB_ID, " .
"SRC_UIDVALIDITY, SRC_UID) VALUES (?, ?, ?)");
return $this->db->execute($sql,
[$job_id, $uidvalidity, $uid]) !== false;
}
/**
* Bulk version of markSeen. Wraps the inserts in a single
* transaction so the DBMS does one commit (and on storage
* engines that fsync per transaction, one fsync) rather
* than fsync-per-row. Used by MailCloneJob::advanceBatch
* once per pipelined IMAP fetch, where the saved fsyncs
* are the largest single per-message cost on a default
* SQLite configuration.
*
* The caller passes a list of [uidvalidity, uid] tuples;
* the job_id is shared across the batch. Empty input is a
* no-op. Failed inserts inside the transaction are
* tolerated -- insertIgnore makes duplicates harmless --
* but a transaction-level error throws (the caller will
* catch and downgrade to per-row inserts on the next tick).
*
* @param int $job_id MAIL_CLONE_JOB.ID all rows belong to
* @param array $tuples list of [uidvalidity, uid] pairs
* @return bool true on commit, false on failure
*/
public function markSeenBulk($job_id, $tuples)
{
if (empty($tuples)) {
return true;
}
$sql = $this->db->insertIgnore(
"INSERT INTO MAIL_CLONE_SEEN (JOB_ID, " .
"SRC_UIDVALIDITY, SRC_UID) VALUES (?, ?, ?)");
$this->db->beginTransaction();
try {
foreach ($tuples as $tuple) {
$this->db->execute($sql, [$job_id,
(int) $tuple[0], (int) $tuple[1]]);
}
$this->db->commit();
return true;
} catch (\Exception $exception) {
/* No portable rollback in DatasourceManager; PDO
auto-rolls-back the implicit transaction when the
handle is destroyed or the next beginTransaction
starts, so simply not committing is safe. Re-throw
so the caller can fall back to per-row inserts. */
throw $exception;
}
}
/**
* Returns true when the (job, UIDVALIDITY, UID) triple has
* already been recorded as imported. Backs the per-message
* "should I fetch this UID?" gate in the clone tick loop;
* the lookup hits the composite primary key and is O(log n)
* regardless of how many messages a job has already
* imported.
*
* @param int $job_id MAIL_CLONE_JOB.ID to scope the lookup
* @param int $uidvalidity source folder's UIDVALIDITY
* @param int $uid source message's UID within that folder
* @return bool whether the message was already imported
*/
public function isSeen($job_id, $uidvalidity, $uid)
{
$sql = "SELECT JOB_ID FROM MAIL_CLONE_SEEN " .
"WHERE JOB_ID = ? AND SRC_UIDVALIDITY = ? " .
"AND SRC_UID = ? " . $this->db->limitOffset(1);
$result = $this->db->execute($sql,
[$job_id, $uidvalidity, $uid]);
if ($result === false) {
return false;
}
$row = $this->db->fetchArray($result);
return !empty($row);
}
/**
* Garbage-collects MAIL_CLONE_SEEN rows whose parent job
* has been in STATUS='done' for more than the supplied
* retention window. Called opportunistically from the
* MailCloneJob tick code. The default retention is one
* month (C\ONE_MONTH); within that window an operator can
* re-clone the same source with full dedup credit, beyond
* it dedup regresses to "first run wipes and re-imports",
* which is the right trade since the source state has
* presumably drifted enough that a fresh pass is preferable
* to relying on month-old UIDVALIDITY/UID pairs.
*
* Implementation uses the two-step "fetch IDs then delete by
* ID list" pattern Yioop's existing models follow rather
* than a DELETE...WHERE IN (SELECT...) subquery, which keeps
* the SQL flat and portable across SQLite, MySQL, Postgres,
* Oracle, and DB2 (the only forms Yioop targets).
*
* @param int $retention_seconds keep dedup rows newer than
* this many seconds past the parent job's completion;
* default C\ONE_MONTH
* @return bool whether the cleanup ran without error
*/
public function purgeOldSeen($retention_seconds = null)
{
if ($retention_seconds === null) {
$retention_seconds = C\ONE_MONTH;
}
$cutoff = time() - $retention_seconds;
$find_sql = "SELECT ID FROM MAIL_CLONE_JOB WHERE " .
"STATUS = 'done' AND UPDATED_AT < ?";
$find_result = $this->db->execute($find_sql, [$cutoff]);
if ($find_result === false) {
return false;
}
$stale_ids = [];
while ($row = $this->db->fetchArray($find_result)) {
$stale_ids[] = (int) $row['ID'];
}
if (empty($stale_ids)) {
return true;
}
$delete_sql = "DELETE FROM MAIL_CLONE_SEEN " .
"WHERE JOB_ID = ?";
foreach ($stale_ids as $stale_id) {
$this->db->execute($delete_sql, [$stale_id]);
}
return true;
}
/**
* Records a per-message error against a clone job. Used by
* MailCloneJob whenever a single UID's import attempt fails
* in a way the user might want to see (pipelined FETCH read
* error, empty body, partial-fetch socket loss, etc.). The
* row stays in MAIL_CLONE_ERROR with RESOLVED_AT=0; a later
* successful retry can flip RESOLVED_AT and RESOLVED_VIA via
* markErrorsResolved so the UI can surface the rescue rather
* than just letting the error disappear.
*
* @param int $job_id MAIL_CLONE_JOB.ID this error belongs to
* @param string $folder source folder name being processed
* @param int $uidvalidity source folder's UIDVALIDITY, or 0
* when the error is folder-scoped rather than UID-scoped
* @param int $uid source UID the error was attributed to, or
* 0 when the error is folder-scoped
* @param string $kind short snake_case token classifying the
* failure (e.g. 'pipelined_fetch_eof', 'empty_body',
* 'partial_fetch_failed'); used by the UI to group
* similar errors
* @param string $message human-readable detail line; truncated
* to MAIL_LAST_ERROR_LEN on insert
*/
public function recordError($job_id, $folder, $uidvalidity,
$uid, $kind, $message)
{
$sql = "INSERT INTO MAIL_CLONE_ERROR (JOB_ID, FOLDER, " .
"SRC_UIDVALIDITY, SRC_UID, ERROR_KIND, MESSAGE, " .
"OCCURRED_AT) VALUES (?, ?, ?, ?, ?, ?, ?)";
$this->db->execute($sql, [(int) $job_id, $folder,
(int) $uidvalidity, (int) $uid,
substr($kind, 0, C\MAIL_CLONE_ERROR_KIND_LEN),
substr($message, 0, C\MAIL_LAST_ERROR_LEN),
time()]);
}
/**
* Marks every unresolved MAIL_CLONE_ERROR row for the given
* job whose SRC_UID falls inside $resolved_uids as resolved.
* Called when a batch of UIDs succeeds via the fall-back
* single-connection path after their pipelined attempt had
* recorded an error; the UI can then show "previously failed,
* rescued via $via" for those errors and stop counting them
* against the active Failed total.
*
* Empty UID list is a no-op. Already-resolved rows are left
* unchanged (the predicate filters on RESOLVED_AT = 0) so a
* second rescue does not overwrite the first.
*
* @param int $job_id MAIL_CLONE_JOB.ID owning the rows
* @param int $uidvalidity source folder's UIDVALIDITY
* @param array $uids list of integer SRC_UID values that just
* succeeded
* @param string $via short snake_case token describing how
* the rescue happened (e.g. 'single_connection_retry')
*/
public function markErrorsResolved($job_id, $uidvalidity,
$uids, $via)
{
if (empty($uids)) {
return;
}
$now = time();
$via_token = substr($via, 0, C\MAIL_CLONE_ERROR_KIND_LEN);
$sql = "UPDATE MAIL_CLONE_ERROR SET RESOLVED_AT = ?, " .
"RESOLVED_VIA = ? WHERE JOB_ID = ? AND " .
"SRC_UIDVALIDITY = ? AND SRC_UID = ? AND " .
"RESOLVED_AT = 0";
foreach ($uids as $uid) {
$this->db->execute($sql, [$now, $via_token,
(int) $job_id, (int) $uidvalidity, (int) $uid]);
}
}
/**
* Returns the count of still-unresolved per-message
* MAIL_CLONE_ERROR rows for a job. The status payload exposes
* this as the authoritative "Failed" number so the counter
* never lags behind the per-row error list shown beneath it.
* Job-level rows (SRC_UID = 0, e.g. a transient connection
* blip that the job recovered from on the next tick) are
* excluded so the "Failed" count reflects messages that failed
* to clone, not whole-job hiccups.
*
* @param int $job_id MAIL_CLONE_JOB.ID to count for
* @return int number of unresolved per-message errors
*/
public function unresolvedErrorCount($job_id)
{
$sql = "SELECT COUNT(*) AS C FROM MAIL_CLONE_ERROR " .
"WHERE JOB_ID = ? AND RESOLVED_AT = 0 AND SRC_UID > 0";
$result = $this->db->execute($sql, [(int) $job_id]);
if ($result === false) {
return 0;
}
$row = $this->db->fetchArray($result);
return ($row === false || empty($row)) ? 0 :
(int) $row['C'];
}
/**
* Returns the most recent error rows for a clone job, newest
* first, capped at $limit. Both unresolved and resolved rows
* are returned so the UI can show "rescued via X" entries
* alongside still-open failures; the caller filters or
* groups as needed.
*
* @param int $job_id MAIL_CLONE_JOB.ID to list errors for
* @param int $limit maximum rows to return
* @return array list of error rows (associative arrays)
*/
public function recentErrors($job_id, $limit = 50)
{
$sql = "SELECT * FROM MAIL_CLONE_ERROR WHERE JOB_ID = ? " .
"ORDER BY OCCURRED_AT DESC " .
$this->db->limitOffset((int) $limit);
$result = $this->db->execute($sql, [(int) $job_id]);
if ($result === false) {
return [];
}
$rows = [];
while ($row = $this->db->fetchArray($result)) {
$rows[] = $row;
}
return $rows;
}
}
(c) 2026 Seekquarry - Open Source Search Engine Software