relayQueue.js

'use strict';

const DEFAULT_STALE_MS = 5 * 60 * 1000;

// 以單一 UPDATE ... RETURNING 取得所有權；多行程同時掃描時，同一列只會被一個工作者 claim。
// 先從最早到期的 25 則中隨機選一則，保留原有「同釋放窗打散順序」的隱私性質。
async function claimDueRelay(db, now = Date.now()) {
    return db.get(`
        UPDATE anon_messages
           SET status='delivering',
               delivery_claimed_at=?,
               delivery_started_at=0,
               delivery_attempts=COALESCE(delivery_attempts, 0) + 1
         WHERE id = (
             SELECT id FROM (
                 SELECT id
                   FROM anon_messages
                  WHERE status='queued' AND release_at<=?
                  ORDER BY release_at ASC
                  LIMIT 25
             )
             ORDER BY random()
             LIMIT 1
         )
           AND status='queued'
        RETURNING *
    `, [now, now]);
}

// Claim one known row without exposing a check-then-act race to the periodic
// drainer. This is used by the interactive "instant" path after moderation:
// the moderation transition first makes the row queued, then exactly one
// caller can move it to delivering.
async function claimRelayById(db, id, now = Date.now()) {
    return db.get(`
        UPDATE anon_messages
           SET status='delivering',
               delivery_claimed_at=?,
               delivery_started_at=0,
               delivery_attempts=COALESCE(delivery_attempts, 0) + 1
         WHERE id=?
           AND status='queued'
           AND release_at<=?
        RETURNING *
    `, [now, id, now]);
}

// 只有在第一個對外 send 緊貼之前才記錄。此時之後崩潰，結果就可能不明，不可自動重送。
async function markRelaySendStarted(db, id, now = Date.now()) {
    const result = await db.run(`
        UPDATE anon_messages
           SET delivery_started_at=?
         WHERE id=? AND status='delivering' AND COALESCE(delivery_started_at, 0)=0
    `, [now, id]);
    return result.changes === 1;
}

async function finishRelay(db, id, status, fields = {}) {
    const allowedStatuses = new Set(['posted', 'rejected', 'error', 'delivery_unknown']);
    if (!allowedStatuses.has(status)) throw new Error(`invalid relay terminal status: ${status}`);

    const assignments = ['status=?'];
    const params = [status];
    if (Object.hasOwn(fields, 'messageId')) {
        assignments.push('message_id=?');
        params.push(fields.messageId);
    }
    if (Object.hasOwn(fields, 'attachmentUrl')) {
        assignments.push('attachment_url=?');
        params.push(fields.attachmentUrl);
    }
    params.push(id);
    const result = await db.run(`
        UPDATE anon_messages
           SET ${assignments.join(', ')}
         WHERE id=? AND status='delivering'
    `, params);
    return result.changes === 1;
}

// 啟動復原區分兩種：還沒有進入 send 的 claim 可安全重排；開始 send 後的結果不明，只能標記人工確認。
async function recoverStaleRelays(db, now = Date.now(), staleMs = DEFAULT_STALE_MS) {
    const cutoff = now - staleMs;
    // Image payload bytes only live in the original interaction process. If a
    // process dies before beginning Discord delivery, the row is terminally
    // failed rather than queued for a worker that cannot reconstruct the file.
    const abandoned = await db.run(`
        UPDATE anon_messages
           SET status='error'
         WHERE status='delivering'
           AND attachment_hash IS NOT NULL
           AND COALESCE(delivery_started_at, 0)=0
           AND delivery_claimed_at>0
           AND delivery_claimed_at<=?
    `, [cutoff]);
    const requeued = await db.run(`
        UPDATE anon_messages
           SET status='queued', delivery_claimed_at=0
         WHERE status='delivering'
           AND attachment_hash IS NULL
           AND COALESCE(delivery_started_at, 0)=0
           AND delivery_claimed_at>0
           AND delivery_claimed_at<=?
    `, [cutoff]);
    const unknown = await db.run(`
        UPDATE anon_messages
           SET status='delivery_unknown'
         WHERE status='delivering'
           AND delivery_started_at>0
           AND delivery_started_at<=?
    `, [cutoff]);
    return { requeued: requeued.changes, unknown: unknown.changes, abandoned: abandoned.changes };
}

module.exports = {
    DEFAULT_STALE_MS,
    claimDueRelay,
    claimRelayById,
    markRelaySendStarted,
    finishRelay,
    recoverStaleRelays,
};