Laan Tungir
50653dc86a
- New caching_custom_backfill_jobs and caching_custom_backfill_batches tables - Admin API (POST create/cancel, GET list) at admin/api/custom_backfill.php - Admin UI form with preset buttons and job status table on Backfill page - Caching daemon module (custom_backfill.c) processes jobs independently - pg_inbox functions for job/batch claim, progress update, and cancel - Three batch modes: author-batched, id-batched, time-window scan - Fixed done_batches overcount on job completion - Admin config now environment-variable driven (C_RELAY_DB_*) - admin/serve.sh supports separate instances for different databases - make_and_restart_relay.sh only kills relay on target port, not all relays
316 lines
13 KiB
PHP
316 lines
13 KiB
PHP
<?php
|
|
/**
|
|
* admin/api/custom_backfill.php — Custom backfill job management.
|
|
*
|
|
* POST actions:
|
|
* - create: Submit a new custom backfill job with a NIP-01 filter.
|
|
* - cancel: Cancel a running/pending job.
|
|
*
|
|
* GET:
|
|
* - List recent jobs with progress.
|
|
*
|
|
* The caching daemon polls caching_custom_backfill_jobs for pending jobs
|
|
* and executes them independently of the followed-set backfill.
|
|
*/
|
|
require_once __DIR__ . '/../lib/helpers.php';
|
|
$pdo = db();
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// POST: create or cancel
|
|
// ---------------------------------------------------------------------------
|
|
if ($_SERVER['REQUEST_METHOD'] === 'POST') {
|
|
$input = json_decode(file_get_contents('php://input'), true) ?: [];
|
|
$action = $input['action'] ?? '';
|
|
|
|
if ($action === 'cancel') {
|
|
$job_id = (int)($input['job_id'] ?? 0);
|
|
if ($job_id <= 0) {
|
|
json_response(['ok' => false, 'error' => 'Invalid job_id'], 400);
|
|
}
|
|
try {
|
|
$stmt = $pdo->prepare(
|
|
"UPDATE caching_custom_backfill_jobs
|
|
SET status = 'cancelled',
|
|
completed_at = EXTRACT(EPOCH FROM NOW())::BIGINT,
|
|
error_message = 'Cancelled by user'
|
|
WHERE job_id = ? AND status IN ('pending', 'running')"
|
|
);
|
|
$stmt->execute([$job_id]);
|
|
if ($stmt->rowCount() === 0) {
|
|
json_response(['ok' => false, 'error' => 'Job not found or not cancellable'], 404);
|
|
}
|
|
json_response(['ok' => true, 'message' => 'Job cancelled']);
|
|
} catch (PDOException $e) {
|
|
json_response(['ok' => false, 'error' => $e->getMessage()], 500);
|
|
}
|
|
}
|
|
|
|
if ($action === 'create') {
|
|
$label = trim($input['label'] ?? '');
|
|
$kinds = trim($input['kinds'] ?? '');
|
|
$authors_source = $input['authors_source'] ?? 'none';
|
|
$authors_list = trim($input['authors_list'] ?? '');
|
|
$authors_missing_kind0 = !empty($input['authors_missing_kind0']);
|
|
$ids_list = trim($input['ids_list'] ?? '');
|
|
$since_ts = $input['since_ts'] ?? null;
|
|
$until_ts = $input['until_ts'] ?? null;
|
|
$limit = (int)($input['limit'] ?? 0);
|
|
$tag_filters = trim($input['tag_filters'] ?? '');
|
|
$batch_size = max(1, (int)($input['batch_size'] ?? 100));
|
|
$relay_mode = $input['relay_mode'] ?? 'bootstrap';
|
|
$relay_list = trim($input['relay_list'] ?? '');
|
|
|
|
// --- Resolve authors ---
|
|
$authors_resolved = '';
|
|
$ids_resolved = '';
|
|
|
|
if ($authors_source === 'manual') {
|
|
// Parse manual list: one hex pubkey per line
|
|
$lines = preg_split('/[\s,]+/', $authors_list);
|
|
$pubkeys = [];
|
|
foreach ($lines as $line) {
|
|
$line = trim($line);
|
|
if (preg_match('/^[0-9a-fA-F]{64}$/', $line)) {
|
|
$pubkeys[] = strtolower($line);
|
|
}
|
|
}
|
|
$authors_resolved = implode(',', array_unique($pubkeys));
|
|
} elseif ($authors_source === 'all_in_db') {
|
|
$rows = $pdo->query("SELECT DISTINCT pubkey FROM events")->fetchAll(PDO::FETCH_COLUMN);
|
|
$authors_resolved = implode(',', $rows);
|
|
} elseif ($authors_source === 'kind1_authors') {
|
|
$rows = $pdo->query("SELECT DISTINCT pubkey FROM events WHERE kind = 1")->fetchAll(PDO::FETCH_COLUMN);
|
|
$authors_resolved = implode(',', $rows);
|
|
} elseif ($authors_source === 'followed') {
|
|
$rows = $pdo->query("SELECT pubkey FROM caching_followed_pubkeys")->fetchAll(PDO::FETCH_COLUMN);
|
|
$authors_resolved = implode(',', $rows);
|
|
}
|
|
// authors_source === 'none' → empty list (time-window mode)
|
|
|
|
// Apply "only missing kind-0" filter
|
|
if ($authors_missing_kind0 && $authors_resolved !== '') {
|
|
$pks = explode(',', $authors_resolved);
|
|
$placeholders = implode(',', array_fill(0, count($pks), '?'));
|
|
$sql = "SELECT pk FROM (VALUES " .
|
|
implode(',', array_map(function($i) {
|
|
return "(:pk$i)";
|
|
}, array_keys($pks))) . ") AS t(pk)
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM events WHERE kind = 0 AND pubkey = t.pk
|
|
)";
|
|
// Simpler: use a temp table approach via IN clause
|
|
$sql = "SELECT t.pk FROM (SELECT unnest(ARRAY[:pks]::text[]) AS pk) t
|
|
WHERE NOT EXISTS (SELECT 1 FROM events WHERE kind = 0 AND pubkey = t.pk)";
|
|
try {
|
|
$stmt = $pdo->prepare("SELECT pk FROM unnest(:pks::text[]) AS pk
|
|
WHERE NOT EXISTS (SELECT 1 FROM events WHERE kind = 0 AND pubkey = pk)");
|
|
$stmt->execute([':pks' => '{' . implode(',', $pks) . '}']);
|
|
$filtered = $stmt->fetchAll(PDO::FETCH_COLUMN);
|
|
$authors_resolved = implode(',', $filtered);
|
|
} catch (PDOException $e) {
|
|
// Fallback: use NOT IN with a subquery
|
|
$stmt = $pdo->prepare("SELECT DISTINCT pubkey FROM events
|
|
WHERE pubkey IN (" . str_repeat('?,', count($pks) - 1) . "?)
|
|
AND NOT EXISTS (SELECT 1 FROM events e2 WHERE e2.kind = 0 AND e2.pubkey = events.pubkey)");
|
|
$stmt->execute($pks);
|
|
$filtered = $stmt->fetchAll(PDO::FETCH_COLUMN);
|
|
$authors_resolved = implode(',', $filtered);
|
|
}
|
|
}
|
|
|
|
// --- Parse IDs ---
|
|
if ($ids_list !== '') {
|
|
$lines = preg_split('/[\s,]+/', $ids_list);
|
|
$ids = [];
|
|
foreach ($lines as $line) {
|
|
$line = trim($line);
|
|
if (preg_match('/^[0-9a-fA-F]{64}$/', $line)) {
|
|
$ids[] = strtolower($line);
|
|
}
|
|
}
|
|
$ids_resolved = implode(',', array_unique($ids));
|
|
}
|
|
|
|
// --- Build filter_json ---
|
|
$filter = [];
|
|
if ($kinds !== '') {
|
|
$kinds_arr = array_values(array_filter(explode(',', $kinds), 'is_numeric'));
|
|
if ($kinds_arr) $filter['kinds'] = array_map('intval', $kinds_arr);
|
|
}
|
|
if ($since_ts !== null && $since_ts !== '') {
|
|
$filter['since'] = (int)$since_ts;
|
|
}
|
|
if ($until_ts !== null && $until_ts !== '') {
|
|
$filter['until'] = (int)$until_ts;
|
|
}
|
|
if ($limit > 0) {
|
|
$filter['limit'] = $limit;
|
|
}
|
|
if ($tag_filters !== '') {
|
|
$tags = json_decode($tag_filters, true);
|
|
if (is_array($tags)) {
|
|
foreach ($tags as $key => $val) {
|
|
if (strpos($key, '#') === 0 && is_array($val)) {
|
|
$filter[$key] = $val;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// --- Determine batch mode ---
|
|
$batch_mode = 'none';
|
|
if ($authors_resolved !== '') {
|
|
$batch_mode = 'authors';
|
|
} elseif ($ids_resolved !== '') {
|
|
$batch_mode = 'ids';
|
|
}
|
|
|
|
// --- Resolve relay list ---
|
|
$relays = [];
|
|
if ($relay_mode === 'custom' && $relay_list !== '') {
|
|
$lines = preg_split('/[\s,]+/', $relay_list);
|
|
foreach ($lines as $line) {
|
|
$line = trim($line);
|
|
if ($line !== '' && preg_match('/^wss?:\/\//', $line)) {
|
|
$relays[] = $line;
|
|
}
|
|
}
|
|
} elseif ($relay_mode === 'bootstrap') {
|
|
// Get bootstrap relays from config
|
|
$cfg = $pdo->query("SELECT value FROM config WHERE key = 'caching_bootstrap_relays'")->fetchColumn();
|
|
if ($cfg) {
|
|
$relays = array_map('trim', explode(',', $cfg));
|
|
}
|
|
}
|
|
if (empty($relays)) {
|
|
// Fallback: use enabled relays from caching_relays
|
|
$rows = $pdo->query("SELECT relay_url FROM caching_relays WHERE live_enabled OR backfill_enabled")->fetchAll(PDO::FETCH_COLUMN);
|
|
$relays = $rows ?: ['wss://relay.damus.io', 'wss://nos.lol'];
|
|
}
|
|
|
|
// --- Create batches ---
|
|
$batch_items_list = [];
|
|
if ($batch_mode === 'authors') {
|
|
$items = explode(',', $authors_resolved);
|
|
$batch_items_list = array_chunk($items, $batch_size);
|
|
} elseif ($batch_mode === 'ids') {
|
|
$items = explode(',', $ids_resolved);
|
|
$batch_items_list = array_chunk($items, $batch_size);
|
|
} else {
|
|
// Time-window mode: one "batch" per relay (empty batch_items)
|
|
$batch_items_list = array_fill(0, count($relays), '');
|
|
}
|
|
|
|
// For author/id-batched mode, create one batch per chunk per relay
|
|
$batches = [];
|
|
if ($batch_mode === 'none') {
|
|
foreach ($relays as $relay) {
|
|
$batches[] = ['items' => '', 'relay' => $relay];
|
|
}
|
|
} else {
|
|
foreach ($batch_items_list as $chunk) {
|
|
foreach ($relays as $relay) {
|
|
$batches[] = ['items' => implode(',', $chunk), 'relay' => $relay];
|
|
}
|
|
}
|
|
}
|
|
|
|
$filter_json = json_encode($filter);
|
|
$total_batches = count($batches);
|
|
|
|
if ($total_batches === 0) {
|
|
json_response(['ok' => false, 'error' => 'No batches to process (empty filter or no relays)'], 400);
|
|
}
|
|
|
|
try {
|
|
$pdo->beginTransaction();
|
|
$stmt = $pdo->prepare(
|
|
"INSERT INTO caching_custom_backfill_jobs
|
|
(label, filter_json, authors_source, authors_resolved, ids_resolved,
|
|
authors_missing_kind0, batch_mode, batch_size, relay_mode, relay_list,
|
|
status, total_batches, created_by)
|
|
VALUES
|
|
(:label, :filter_json::jsonb, :authors_source, :authors_resolved, :ids_resolved,
|
|
:authors_missing_kind0, :batch_mode, :batch_size, :relay_mode, :relay_list,
|
|
'pending', :total_batches, 'admin')
|
|
RETURNING job_id"
|
|
);
|
|
$stmt->execute([
|
|
':label' => $label,
|
|
':filter_json' => $filter_json,
|
|
':authors_source' => $authors_source,
|
|
':authors_resolved' => $authors_resolved,
|
|
':ids_resolved' => $ids_resolved,
|
|
':authors_missing_kind0' => $authors_missing_kind0 ? 1 : 0,
|
|
':batch_mode' => $batch_mode,
|
|
':batch_size' => $batch_size,
|
|
':relay_mode' => $relay_mode,
|
|
':relay_list' => $relay_list,
|
|
':total_batches' => $total_batches,
|
|
]);
|
|
$job_id = $stmt->fetchColumn();
|
|
|
|
// Insert batches
|
|
$batch_stmt = $pdo->prepare(
|
|
"INSERT INTO caching_custom_backfill_batches
|
|
(job_id, batch_index, batch_items, relay_url, status)
|
|
VALUES (?, ?, ?, ?, 'pending')"
|
|
);
|
|
foreach ($batches as $i => $batch) {
|
|
$batch_stmt->execute([$job_id, $i, $batch['items'], $batch['relay']]);
|
|
}
|
|
|
|
$pdo->commit();
|
|
json_response([
|
|
'ok' => true,
|
|
'job_id' => (int)$job_id,
|
|
'total_batches' => $total_batches,
|
|
'batch_mode' => $batch_mode,
|
|
'message' => "Job created with $total_batches batches"
|
|
]);
|
|
} catch (PDOException $e) {
|
|
$pdo->rollBack();
|
|
json_response(['ok' => false, 'error' => $e->getMessage()], 500);
|
|
}
|
|
}
|
|
|
|
json_response(['ok' => false, 'error' => 'Unknown action'], 400);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// GET: list jobs
|
|
// ---------------------------------------------------------------------------
|
|
$limit = min(50, max(1, (int)($_GET['limit'] ?? 20)));
|
|
try {
|
|
$jobs = $pdo->prepare(
|
|
"SELECT job_id, label, filter_json, authors_source, batch_mode,
|
|
batch_size, relay_mode, status, total_batches, done_batches,
|
|
events_fetched, error_message,
|
|
created_at, started_at, completed_at
|
|
FROM caching_custom_backfill_jobs
|
|
ORDER BY created_at DESC
|
|
LIMIT ?"
|
|
);
|
|
$jobs->execute([$limit]);
|
|
$jobs = $jobs->fetchAll();
|
|
|
|
// Include batch details for running jobs
|
|
foreach ($jobs as &$job) {
|
|
$job['filter_json'] = json_decode($job['filter_json'], true);
|
|
if ($job['status'] === 'running') {
|
|
$bstmt = $pdo->prepare(
|
|
"SELECT batch_index, relay_url, status, events_fetched, error_message
|
|
FROM caching_custom_backfill_batches
|
|
WHERE job_id = ?
|
|
ORDER BY batch_index"
|
|
);
|
|
$bstmt->execute([$job['job_id']]);
|
|
$job['batches'] = $bstmt->fetchAll();
|
|
}
|
|
}
|
|
|
|
json_response(['ok' => true, 'jobs' => $jobs]);
|
|
} catch (PDOException $e) {
|
|
json_response(['ok' => false, 'error' => $e->getMessage()], 500);
|
|
}
|