vendoo/lib/publishing-jobs.mjs

import { randomUUID, createHash } from 'crypto';
import { db, getListing, createPublishLog, updatePublishLog } from './db.mjs';

const ACTIVE = new Set(['queued', 'running', 'awaiting_user', 'retry_wait']);
const TERMINAL = new Set(['published', 'drafted', 'failed', 'cancelled', 'ended', 'blocked']);
const DIRECT = new Set(['ebay-de', 'etsy']);
let timer = null;
let processing = false;
let processors = null;

function parseJson(value, fallback = {}) {
  try { return JSON.parse(value || ''); } catch { return fallback; }
}
function serialize(row) {
  if (!row) return null;
  return { ...row, payload: parseJson(row.payload_json), result: parseJson(row.result_json) };
}
function modeFor(platform) { return DIRECT.has(platform) ? 'api' : 'extension'; }
function keyFor(listingId, platform, action, nonce = '') {
  return createHash('sha256').update(`${listingId}|${platform}|${action}|${nonce}`).digest('hex');
}
function addEvent(job, eventType, message, level = 'info', details = {}) {
  db.prepare(`INSERT INTO publishing_events (job_id, listing_id, platform, level, event_type, message, details_json)
    VALUES (?, ?, ?, ?, ?, ?, ?)`).run(job?.id || null, job.listing_id, job.platform, level, eventType, message, JSON.stringify(details || {}));
}
function latestPublished(listingId, platform) {
  return db.prepare(`SELECT * FROM publish_log WHERE listing_id = ? AND platform = ? AND status = 'published' ORDER BY id DESC LIMIT 1`).get(listingId, platform);
}
function currentActive(listingId, platform, action) {
  return db.prepare(`SELECT * FROM publishing_jobs WHERE listing_id = ? AND platform = ? AND action = ?
    AND status IN ('queued','running','awaiting_user','retry_wait') ORDER BY created_at DESC LIMIT 1`).get(listingId, platform, action);
}
export function enqueuePublishingJob({ listingId, platform, action = 'publish', force = false, payload = {}, maxRetries = 3 } = {}) {
  const listing = getListing(Number(listingId));
  if (!listing) throw new Error('Artikel nicht gefunden');
  const target = platform || listing.platform;
  if (!target) throw new Error('Plattform fehlt');
  const existing = currentActive(listing.id, target, action);
  if (existing) return { job: serialize(existing), duplicate: true };
  if (action === 'publish' && !force) {
    const live = latestPublished(listing.id, target);
    if (live?.external_id) {
      const error = new Error('Artikel ist auf dieser Plattform bereits als veröffentlicht protokolliert');
      error.code = 'DUPLICATE_PUBLISH';
      error.existing = live;
      throw error;
    }
  }
  const id = randomUUID();
  const mode = modeFor(target);
  const status = mode === 'extension' && action === 'publish' ? 'awaiting_user' : 'queued';
  const nonce = force ? randomUUID() : `${new Date().toISOString()}`;
  const key = keyFor(listing.id, target, action, nonce);
  db.prepare(`INSERT INTO publishing_jobs
    (id, listing_id, platform, mode, action, status, phase, idempotency_key, max_retries, payload_json)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
    .run(id, listing.id, target, mode, action, status, status, key, Math.min(Math.max(Number(maxRetries) || 3, 0), 10), JSON.stringify(payload || {}));
  const job = serialize(db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(id));
  addEvent(job, 'created', mode === 'api' ? 'Publishing-Auftrag eingereiht' : 'Auftrag wartet auf Browser-Extension');
  return { job, duplicate: false };
}
export function getPublishingJob(id) {
  const row = db.prepare(`SELECT j.*, l.title, l.sku, l.photos FROM publishing_jobs j JOIN listings l ON l.id = j.listing_id WHERE j.id = ?`).get(id);
  if (!row) return null;
  const job = serialize(row);
  job.events = db.prepare('SELECT * FROM publishing_events WHERE job_id = ? ORDER BY id DESC LIMIT 100').all(id).map(event => ({ ...event, details: parseJson(event.details_json) }));
  return job;
}
export function paginatePublishingJobs({ page = 1, pageSize = 10, status = '', platform = '', q = '' } = {}) {
  const size = [5, 10, 15, 25, 50].includes(Number(pageSize)) ? Number(pageSize) : 10;
  const current = Math.max(1, Number(page) || 1);
  const where = [];
  const params = [];
  if (status) { where.push('j.status = ?'); params.push(status); }
  if (platform) { where.push('j.platform = ?'); params.push(platform); }
  if (q) { where.push('(l.title LIKE ? OR l.sku LIKE ? OR j.external_id LIKE ? OR j.error_message LIKE ?)'); const needle = `%${q}%`; params.push(needle, needle, needle, needle); }
  const clause = where.length ? `WHERE ${where.join(' AND ')}` : '';
  const total = db.prepare(`SELECT COUNT(*) AS count FROM publishing_jobs j JOIN listings l ON l.id = j.listing_id ${clause}`).get(...params).count;
  const rows = db.prepare(`SELECT j.*, l.title, l.sku, l.photos FROM publishing_jobs j JOIN listings l ON l.id = j.listing_id ${clause}
    ORDER BY CASE j.status WHEN 'running' THEN 0 WHEN 'queued' THEN 1 WHEN 'awaiting_user' THEN 2 WHEN 'retry_wait' THEN 3 WHEN 'failed' THEN 4 ELSE 5 END, j.created_at DESC
    LIMIT ? OFFSET ?`).all(...params, size, (current - 1) * size);
  const summaryRows = db.prepare(`SELECT status, COUNT(*) AS count FROM publishing_jobs GROUP BY status`).all();
  return {
    items: rows.map(serialize),
    pagination: { page: current, page_size: size, total, total_pages: Math.max(1, Math.ceil(total / size)) },
    summary: Object.fromEntries(summaryRows.map(row => [row.status, row.count])),
  };
}
export function listPublishingEvents({ limit = 100, level = '', platform = '' } = {}) {
  const where = [];
  const params = [];
  if (level) { where.push('e.level = ?'); params.push(level); }
  if (platform) { where.push('e.platform = ?'); params.push(platform); }
  const clause = where.length ? `WHERE ${where.join(' AND ')}` : '';
  return db.prepare(`SELECT e.*, l.title, l.sku FROM publishing_events e JOIN listings l ON l.id = e.listing_id ${clause} ORDER BY e.id DESC LIMIT ?`)
    .all(...params, Math.min(Math.max(Number(limit) || 100, 1), 500)).map(row => ({ ...row, details: parseJson(row.details_json) }));
}
export function retryPublishingJob(id) {
  const job = db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(id);
  if (!job) throw new Error('Auftrag nicht gefunden');
  if (!['failed', 'blocked', 'cancelled'].includes(job.status)) throw new Error('Nur fehlgeschlagene, blockierte oder abgebrochene Aufträge können wiederholt werden');
  const next = modeFor(job.platform) === 'extension' && job.action === 'publish' ? 'awaiting_user' : 'queued';
  db.prepare(`UPDATE publishing_jobs SET status = ?, phase = ?, error_code = NULL, error_message = NULL, next_attempt_at = NULL,
    completed_at = NULL, updated_at = datetime('now') WHERE id = ?`).run(next, next, id);
  const updated = serialize(db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(id));
  addEvent(updated, 'retry', 'Auftrag erneut eingeplant');
  return updated;
}
export function cancelPublishingJob(id) {
  const job = db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(id);
  if (!job) throw new Error('Auftrag nicht gefunden');
  if (TERMINAL.has(job.status)) return serialize(job);
  db.prepare(`UPDATE publishing_jobs SET status = 'cancelled', phase = 'cancelled', completed_at = datetime('now'), updated_at = datetime('now') WHERE id = ?`).run(id);
  const updated = serialize(db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(id));
  addEvent(updated, 'cancelled', 'Auftrag abgebrochen', 'warning');
  return updated;
}
export function deletePublishingJobs(ids = []) {
  const safe = [...new Set(ids.map(String).filter(Boolean))];
  if (!safe.length) return 0;
  const placeholders = safe.map(() => '?').join(',');
  const rows = db.prepare(`SELECT id, status FROM publishing_jobs WHERE id IN (${placeholders})`).all(...safe);
  if (rows.some(row => ACTIVE.has(row.status))) throw new Error('Aktive Aufträge können nicht gelöscht werden');
  return db.prepare(`DELETE FROM publishing_jobs WHERE id IN (${placeholders})`).run(...safe).changes;
}
export function claimExtensionJob({ platform = 'vinted', clientId = 'extension' } = {}) {
  const job = db.prepare(`SELECT * FROM publishing_jobs WHERE platform = ? AND mode = 'extension' AND status = 'awaiting_user'
    ORDER BY created_at LIMIT 1`).get(platform);
  if (!job) return null;
  db.prepare(`UPDATE publishing_jobs SET claimed_by = ?, claimed_at = datetime('now'), phase = 'claimed', updated_at = datetime('now') WHERE id = ?`).run(clientId, job.id);
  const updated = serialize(db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(job.id));
  addEvent(updated, 'claimed', 'Auftrag von Browser-Extension übernommen', 'info', { clientId });
  return { ...updated, listing: getListing(updated.listing_id) };
}
export function completeExtensionJob(id, { status = 'published', externalId = null, externalUrl = null, error = null } = {}) {
  const job = db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(id);
  if (!job) throw new Error('Auftrag nicht gefunden');
  const nextStatus = status === 'published' ? 'published' : status === 'draft' ? 'awaiting_user' : 'failed';
  db.prepare(`UPDATE publishing_jobs SET status = ?, phase = ?, external_id = COALESCE(?, external_id), external_url = COALESCE(?, external_url),
    error_message = ?, completed_at = CASE WHEN ? IN ('published','failed') THEN datetime('now') ELSE NULL END, updated_at = datetime('now') WHERE id = ?`)
    .run(nextStatus, nextStatus, externalId, externalUrl, error, nextStatus, id);
  const updated = serialize(db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(id));
  if (nextStatus === 'published') createPublishLog(updated.listing_id, updated.platform, 'published', externalId, externalUrl);
  addEvent(updated, nextStatus, nextStatus === 'published' ? 'Extension-Auftrag als veröffentlicht bestätigt' : (error || 'Extension-Auftrag aktualisiert'), nextStatus === 'failed' ? 'error' : 'info');
  return updated;
}
function markRunning(job) {
  db.prepare(`UPDATE publishing_jobs SET status = 'running', phase = ?, started_at = COALESCE(started_at, datetime('now')), updated_at = datetime('now') WHERE id = ?`).run(job.action, job.id);
  return serialize(db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(job.id));
}
function markSuccess(job, result = {}) {
  const status = job.action === 'end' ? 'ended' : (job.action === 'sync' ? (result.status || 'published') : (String(result.status || '').toLowerCase() === 'draft' ? 'drafted' : 'published'));
  db.prepare(`UPDATE publishing_jobs SET status = ?, phase = 'completed', external_id = COALESCE(?, external_id), external_url = COALESCE(?, external_url),
    offer_id = COALESCE(?, offer_id), result_json = ?, error_code = NULL, error_message = NULL, completed_at = datetime('now'), updated_at = datetime('now') WHERE id = ?`)
    .run(status, result.externalId || result.listingId || null, result.url || null, result.offerId || null, JSON.stringify(result || {}), job.id);
  const updated = serialize(db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(job.id));
  if (job.action === 'publish') {
    createPublishLog(job.listing_id, job.platform, status === 'drafted' ? 'draft' : 'published', updated.external_id, updated.external_url);
    const createdLog = db.prepare('SELECT id FROM publish_log WHERE listing_id = ? AND platform = ? ORDER BY id DESC LIMIT 1').get(job.listing_id, job.platform);
    if (createdLog) updatePublishLog(createdLog.id, { offer_id: updated.offer_id, sync_status: status, last_synced_at: new Date().toISOString(), updated_at: new Date().toISOString() });
  } else {
    const log = latestPublished(job.listing_id, job.platform);
    if (log) updatePublishLog(log.id, {
      status: job.action === 'end' ? 'ended' : (result.status || log.status),
      external_id: updated.external_id,
      external_url: updated.external_url,
      offer_id: updated.offer_id,
      sync_status: result.status || status,
      last_synced_at: new Date().toISOString(),
      ended_at: job.action === 'end' ? new Date().toISOString() : undefined,
      updated_at: new Date().toISOString(),
    });
  }
  addEvent(updated, 'completed', job.action === 'end' ? 'Angebot beendet' : job.action === 'sync' ? 'Status synchronisiert' : 'Veröffentlichung erfolgreich');
  return updated;
}
function markFailure(job, error) {
  const retries = Number(job.retries || 0) + 1;
  const canRetry = retries <= Number(job.max_retries || 3);
  const delayMinutes = Math.min(30, Math.max(1, 2 ** Math.max(0, retries - 1)));
  const status = canRetry ? 'retry_wait' : 'failed';
  db.prepare(`UPDATE publishing_jobs SET status = ?, phase = ?, retries = ?, error_code = ?, error_message = ?,
    next_attempt_at = CASE WHEN ? = 'retry_wait' THEN datetime('now', '+' || ? || ' minutes') ELSE NULL END,
    completed_at = CASE WHEN ? = 'failed' THEN datetime('now') ELSE NULL END, updated_at = datetime('now') WHERE id = ?`)
    .run(status, status, retries, error?.code || 'PUBLISH_ERROR', error?.message || String(error), status, delayMinutes, status, job.id);
  const updated = serialize(db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(job.id));
  addEvent(updated, status, error?.message || String(error), 'error', { retries, nextRetryMinutes: canRetry ? delayMinutes : null });
  return updated;
}
async function processOne() {
  if (!processors) return;
  const row = db.prepare(`SELECT * FROM publishing_jobs WHERE mode = 'api' AND (
    status = 'queued' OR (status = 'retry_wait' AND (next_attempt_at IS NULL OR next_attempt_at <= datetime('now')))
  ) ORDER BY created_at LIMIT 1`).get();
  if (!row) return;
  const job = markRunning(row);
  try {
    const listing = getListing(job.listing_id);
    if (!listing) throw new Error('Artikel nicht gefunden');
    const result = await processors.process(job, listing);
    markSuccess(job, result || {});
  } catch (error) { markFailure(job, error); }
}
export function configurePublishingProcessors(value) { processors = value; }
export function startPublishingJobLoop(intervalMs = 1500) {
  if (timer) return;
  const tick = async () => {
    if (processing) return;
    processing = true;
    try { await processOne(); } finally { processing = false; }
  };
  timer = setInterval(tick, intervalMs);
  timer.unref?.();
  tick();
}

export function stopPublishingJobLoop() {
  if (timer) clearInterval(timer);
  timer = null;
  processing = false;
}

export function migrateLegacyEbayQueue() {
  let rows = [];
  try { rows = db.prepare('SELECT * FROM ebay_queue ORDER BY id').all(); } catch { return { migrated: 0, skipped: 0 }; }
  let migrated = 0;
  let skipped = 0;
  for (const row of rows) {
    const key = `legacy-ebay-queue-${row.id}`;
    const exists = db.prepare('SELECT id FROM publishing_jobs WHERE idempotency_key = ?').get(key);
    if (exists) { skipped += 1; continue; }
    const statusMap = { pending: 'queued', uploading: 'queued', failed: 'failed', published: 'published' };
    const status = statusMap[row.status] || 'queued';
    const id = randomUUID();
    db.prepare(`INSERT OR IGNORE INTO publishing_jobs
      (id, listing_id, platform, mode, action, status, phase, idempotency_key, external_id, external_url,
       retries, max_retries, error_message, created_at, completed_at, updated_at)
      VALUES (?, ?, 'ebay-de', 'api', 'publish', ?, ?, ?, ?, ?, ?, 3, ?, COALESCE(?, datetime('now')),
       CASE WHEN ? IN ('published','failed') THEN COALESCE(?, datetime('now')) ELSE NULL END, datetime('now'))`)
      .run(id, row.listing_id, status, row.status === 'uploading' ? 'recovered' : status, key,
        row.ebay_item_id || null, row.ebay_url || null, Number(row.retries || 0), row.error || null,
        row.created_at || null, status, row.updated_at || row.created_at || null);
    const job = serialize(db.prepare('SELECT * FROM publishing_jobs WHERE id = ?').get(id));
    if (job) {
      addEvent(job, 'legacy_migrated', 'Eintrag aus der bisherigen eBay-Queue übernommen', 'info', { legacyQueueId: row.id, legacyStatus: row.status });
      migrated += 1;
    }
  }
  return { migrated, skipped };
}

export function recoverPublishingJobs() {
  const changes = db.prepare(`UPDATE publishing_jobs SET status = 'queued', phase = 'recovered', updated_at = datetime('now') WHERE status = 'running'`).run().changes;
  return changes;
}
export function getPublishCapabilities() {
  return {
    platforms: {
      'ebay-de': { mode: 'api', publish: true, update: true, end: true, sync: true },
      etsy: { mode: 'api', publish: true, update: false, end: false, sync: true },
      vinted: { mode: 'extension', publish: true, update: false, end: false, sync: false },
      'ebay-ka': { mode: 'extension', publish: true, update: false, end: false, sync: false },
    },
  };
}