anonMessageState.js

'use strict';

const MODERATION_ACTION_STATUS = Object.freeze({
    reject: 'rejected',
    queue: 'held',
    pass: 'queued',
});
const IMAGE_STATUSES = new Set(['rejected', 'delivering']);
const ACTIVE_DELIVERY_STATUSES = "'moderating','held','queued','delivering','posted','delivery_unknown'";

function requiredText(value, name) {
    const text = String(value == null ? '' : value);
    if (!text) throw new TypeError(`${name} is required`);
    return text;
}

function optionalText(value) {
    return value == null || value === '' ? null : String(value);
}

function finiteTimestamp(value, name) {
    const number = Number(value);
    if (!Number.isSafeInteger(number) || number < 0) throw new TypeError(`${name} must be a non-negative safe integer`);
    return number;
}

/**
 * Persist the original text before any moderation provider or admission wait.
 * A crash can therefore lose an automated decision, but never the intercepted
 * message or its audit identity.
 */
async function createModerationRecord(db, record, now = Date.now(), { quota = {} } = {}) {
    if (!db) throw new TypeError('db is required');
    const createdAt = finiteTimestamp(record.createdAt ?? now, 'createdAt');
    const dayStart = finiteTimestamp(quota.dayStart ?? 0, 'quota.dayStart');
    const normalizeLimit = value => {
        const number = Number(value);
        return Number.isSafeInteger(number) && number > 0 ? number : 0;
    };
    const attemptsLimit = normalizeLimit(quota.attemptsLimit);
    const userLimit = normalizeLimit(quota.userLimit);
    const guildLimit = normalizeLimit(quota.guildLimit);
    const guildId = requiredText(record.guildId, 'guildId');
    const userId = requiredText(record.userId, 'userId');
    const result = await db.get(`
        WITH quota(attempts_limit, user_limit, guild_limit, day_start) AS (
            VALUES (?,?,?,?)
        )
        INSERT INTO anon_messages (
            guild_id, channel_id, user_id, user_tag, alias, content,
            reply_to_message_id, created_at, status, avatar_seed,
            moderation_started_at, moderation_completed_at, decision_reason,
            release_at, delivery_claimed_at, delivery_started_at
        )
        SELECT ?,?,?,?,?,?,?,?, 'moderating', ?, ?, 0, NULL, 0, 0, 0
          FROM quota
         WHERE (attempts_limit<=0 OR (
                   SELECT COUNT(*) FROM anon_messages
                    WHERE guild_id=? AND user_id=? AND created_at>=day_start
               ) < attempts_limit)
           AND (user_limit<=0 OR (
                   SELECT COUNT(*) FROM anon_messages
                    WHERE guild_id=? AND user_id=? AND created_at>=day_start
                      AND status IN ('moderating','held','queued','delivering','posted','delivery_unknown')
               ) < user_limit)
           AND (guild_limit<=0 OR (
                   SELECT COUNT(*) FROM anon_messages
                    WHERE guild_id=? AND created_at>=day_start
                      AND status IN ('moderating','held','queued','delivering','posted','delivery_unknown')
               ) < guild_limit)
        RETURNING id
    `, [
        attemptsLimit,
        userLimit,
        guildLimit,
        dayStart,
        guildId,
        requiredText(record.channelId, 'channelId'),
        userId,
        optionalText(record.userTag),
        requiredText(record.alias, 'alias'),
        String(record.content == null ? '' : record.content),
        optionalText(record.replyToMessageId),
        createdAt,
        optionalText(record.avatarSeed),
        createdAt,
        guildId,
        userId,
        guildId,
        userId,
        guildId,
    ]);
    return result ? Number(result.id) : null;
}

/**
 * Atomically records an already-sanitized image and reserves every applicable
 * daily budget. attachment_mime is the durable image discriminator; unlike
 * attachment_hash it survives privacy unlinking and cannot collide with text.
 */
async function createImageRecord(db, record, now = Date.now(), { quota = {} } = {}) {
    if (!db) throw new TypeError('db is required');
    if (!IMAGE_STATUSES.has(record.status)) throw new TypeError('image status must be rejected or delivering');
    const createdAt = finiteTimestamp(record.createdAt ?? now, 'createdAt');
    const dayStart = finiteTimestamp(quota.dayStart ?? 0, 'quota.dayStart');
    const normalizeLimit = value => {
        const number = Number(value);
        return Number.isSafeInteger(number) && number > 0 ? number : 0;
    };
    const attemptsLimit = normalizeLimit(quota.attemptsLimit);
    const userLimit = normalizeLimit(quota.userLimit);
    const guildLimit = normalizeLimit(quota.guildLimit);
    const imageLimit = normalizeLimit(quota.imageLimit);
    const guildId = requiredText(record.guildId, 'guildId');
    const userId = requiredText(record.userId, 'userId');
    const activeDelivery = record.status === 'delivering' ? 1 : 0;
    const result = await db.get(`
        WITH quota(attempts_limit, user_limit, guild_limit, image_limit, day_start, active_delivery) AS (
            VALUES (?,?,?,?,?,?)
        )
        INSERT INTO anon_messages (
            guild_id, channel_id, user_id, user_tag, alias, content,
            reply_to_message_id, created_at, status, avatar_seed,
            attachment_hash, attachment_bytes, attachment_mime, attachment_name,
            delivery_claimed_at, delivery_started_at, delivery_attempts
        )
        SELECT ?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,0,?
          FROM quota
         WHERE (attempts_limit<=0 OR (
                   SELECT COUNT(*) FROM anon_messages
                    WHERE guild_id=? AND user_id=? AND created_at>=day_start
               ) < attempts_limit)
           AND (image_limit<=0 OR (
                   SELECT COUNT(*) FROM anon_messages
                    WHERE guild_id=? AND user_id=? AND created_at>=day_start
                      AND attachment_mime IS NOT NULL
               ) < image_limit)
           AND (active_delivery=0 OR user_limit<=0 OR (
                   SELECT COUNT(*) FROM anon_messages
                    WHERE guild_id=? AND user_id=? AND created_at>=day_start
                      AND status IN (${ACTIVE_DELIVERY_STATUSES})
               ) < user_limit)
           AND (active_delivery=0 OR guild_limit<=0 OR (
                   SELECT COUNT(*) FROM anon_messages
                    WHERE guild_id=? AND created_at>=day_start
                      AND status IN (${ACTIVE_DELIVERY_STATUSES})
               ) < guild_limit)
        RETURNING id
    `, [
        attemptsLimit, userLimit, guildLimit, imageLimit, dayStart, activeDelivery,
        guildId,
        requiredText(record.channelId, 'channelId'),
        userId,
        optionalText(record.userTag),
        requiredText(record.alias, 'alias'),
        String(record.content == null ? '' : record.content),
        optionalText(record.replyToMessageId),
        createdAt,
        record.status,
        optionalText(record.avatarSeed),
        requiredText(record.attachmentHash, 'attachmentHash'),
        finiteTimestamp(record.attachmentBytes, 'attachmentBytes'),
        requiredText(record.attachmentMime, 'attachmentMime'),
        requiredText(record.attachmentName, 'attachmentName'),
        record.status === 'delivering' ? createdAt : 0,
        record.status === 'delivering' ? 1 : 0,
        guildId, userId,
        guildId, userId,
        guildId, userId,
        guildId,
    ]);
    return result ? Number(result.id) : null;
}

/**
 * Compare-and-swap the result of one moderation attempt. Every passing row
 * first becomes queued; interactive callers may immediately claim that row,
 * while the background drainer safely owns anything left behind by a crash.
 */
async function completeModeration(db, id, {
    action, reason = null, releaseAt = 0, heldLimit = 0,
} = {}, now = Date.now()) {
    const status = MODERATION_ACTION_STATUS[action];
    if (!status) throw new TypeError(`unsupported moderation action: ${action}`);
    const completedAt = finiteTimestamp(now, 'now');
    const normalizedReleaseAt = action === 'pass'
        ? finiteTimestamp(releaseAt, 'releaseAt')
        : 0;
    const normalizedHeldLimit = Number.isSafeInteger(Number(heldLimit)) && Number(heldLimit) > 0
        ? Math.min(5000, Number(heldLimit)) : 0;
    const updateSql = `
        UPDATE anon_messages
           SET status=?,
               decision_reason=?,
               moderation_completed_at=?,
               release_at=?,
               review_notice_status=CASE WHEN ?='held' THEN 'pending' ELSE 'none' END,
               review_notice_next_at=CASE WHEN ?='held' THEN ? ELSE 0 END
         WHERE id=? AND status='moderating'
           AND (?<=0 OR ?<>'held' OR (
                SELECT COUNT(*) FROM anon_messages queued
                 WHERE queued.guild_id=anon_messages.guild_id AND queued.status='held'
           ) < ?)
        RETURNING *
    `;
    const params = [status, optionalText(reason), completedAt, normalizedReleaseAt,
        status, status, completedAt, id, normalizedHeldLimit, status, normalizedHeldLimit];
    if (status !== 'held' || normalizedHeldLimit <= 0) return db.get(updateSql, params);
    return db.transaction(raw => {
        const accepted = raw.prepare(updateSql).get(...params);
        if (accepted) return accepted;
        const capacityReason = 'Human review queue reached its per-guild safety capacity; message was rejected without delivery.';
        const rejected = raw.prepare(`
            UPDATE anon_messages
               SET status='rejected', decision_reason=?, moderation_completed_at=?,
                   release_at=0, review_notice_status='none', review_notice_next_at=0
             WHERE id=? AND status='moderating'
            RETURNING *
        `).get(capacityReason, completedAt, id);
        return rejected ? { ...rejected, queueCapacityRejected: true } : null;
    }, 'immediate');
}

/**
 * On boot, interrupted moderation is held for a human instead of being
 * silently released or automatically re-billed to an external AI provider.
 * The transaction returns exactly the rows whose state it won.
 */
async function recoverInterruptedModeration(db, now = Date.now(), {
    staleMs = 0, limit = 100, heldLimit = 0,
} = {}) {
    const completedAt = finiteTimestamp(now, 'now');
    const normalizedStaleMs = finiteTimestamp(staleMs, 'staleMs');
    if (!Number.isSafeInteger(limit) || limit <= 0 || limit > 1000) throw new TypeError('limit must be an integer from 1 to 1000');
    const cutoff = completedAt - normalizedStaleMs;
    const normalizedHeldLimit = Number.isSafeInteger(Number(heldLimit)) && Number(heldLimit) > 0
        ? Math.min(5000, Number(heldLimit)) : 0;
    const reason = 'Automated moderation was interrupted or exceeded its processing lease; human review required.';
    return db.transaction(raw => {
        const rows = raw.prepare(`
            SELECT * FROM anon_messages
             WHERE status='moderating'
               AND COALESCE(moderation_started_at, 0)<=?
             ORDER BY id
             LIMIT ?
        `).all(cutoff, limit);
        if (rows.length === 0) return [];
        const update = raw.prepare(`
            UPDATE anon_messages
               SET status='held', decision_reason=?, moderation_completed_at=?,
                   review_notice_status='pending', review_notice_next_at=?
             WHERE id=? AND status='moderating'
        `);
        const reject = raw.prepare(`
            UPDATE anon_messages
               SET status='rejected', decision_reason=?, moderation_completed_at=?,
                   review_notice_status='none', review_notice_next_at=0
             WHERE id=? AND status='moderating'
        `);
        const countHeld = raw.prepare(`SELECT COUNT(*) AS count FROM anon_messages WHERE guild_id=? AND status='held'`);
        const heldByGuild = new Map();
        const recovered = [];
        for (const row of rows) {
            if (!heldByGuild.has(row.guild_id)) {
                heldByGuild.set(row.guild_id, Number(countHeld.get(row.guild_id)?.count || 0));
            }
            if (normalizedHeldLimit > 0 && heldByGuild.get(row.guild_id) >= normalizedHeldLimit) {
                const capacityReason = 'Human review queue capacity was full during interrupted moderation recovery.';
                if (reject.run(capacityReason, completedAt, row.id).changes === 1) {
                    recovered.push({ ...row, status: 'rejected', decision_reason: capacityReason, queueCapacityRejected: true });
                }
                continue;
            }
            if (update.run(reason, completedAt, completedAt, row.id).changes === 1) {
                recovered.push({ ...row, status: 'held', decision_reason: reason, moderation_completed_at: completedAt });
                heldByGuild.set(row.guild_id, heldByGuild.get(row.guild_id) + 1);
            }
        }
        return recovered;
    }, 'immediate');
}

module.exports = {
    MODERATION_ACTION_STATUS,
    createImageRecord,
    createModerationRecord,
    completeModeration,
    recoverInterruptedModeration,
};