/** * [Ariweb Suite 변경] 발송 속도 조절·제출 제한·발송 대기열(Dev-Order §17.2). * * 세 가지를 맡습니다. * 1) 안전 발송 속도(pace): 실제 외부 전달은 도메인마다, 그리고 서버 전체로 초당 5건을 넘지 않습니다(토큰 버킷 두 개를 모두 통과해야 발송). * 버킷이 비어 있으면 그 자리에서 기다립니다(최대 WAIT_MAX_MS, 넘으면 mail_rate_limited 로 실패시켜 메일 프로그램이 나중에 다시 보내게 함). * 2) 제출 제한(submission): 계정·접속 IP 가 초당 10건 이상 제출하면 자동 제한(mail_submission_throttled)·경고 기록·악용 확인 표(abuse_flags)에 남깁니다. * 제한 창(THROTTLE_WINDOW_MS)이 지나면 자동으로 풀리고, 관리자 화면(/api/admin/rates)이 현재 창의 제출량과 버킷 상태를 보여 줍니다. * 3) 발송 대기열(outbound_queue): 발송 모듈에 넘기기 직전 행을 넣고(state sending) 끝나면 sent/failed 로 바꿉니다. 재시작 때 sending 으로 남은 행은 * interrupted 로 표시해 관리자가 다시 보낼 수 있게 합니다(메일을 잃지 않음). 같은 send_key 가 이미 sent 면 다시 보내지 않습니다(멱등). * * 시계는 clock() 으로 바꿀 수 있어 시험이 실제 시간을 기다리지 않습니다. */ import { now, uid } from './util.mjs'; export const DELIVERY_RATE = 5; // 초당 외부 전달 상한(도메인·서버 각각) export const SUBMISSION_RATE = 10; // 초당 제출 상한(계정·IP 각각). 이 이상이면 제한·경고·악용 확인 export const THROTTLE_WINDOW_MS = 60 * 1000; // 제한이 유지되는 시간 export const WAIT_MAX_MS = 15 * 1000; // 버킷을 기다리는 최대 시간 let clock = () => Date.now(); /** 시험용: 시계를 바꿉니다(ms 를 돌려주는 함수). null 이면 실제 시계. */ export function setClock(fn) { clock = fn || (() => Date.now()); } /** 시험용: 대기 함수(기본 setTimeout). 가짜 시계와 함께 바꿉니다. */ let sleep = (ms) => new Promise((r) => setTimeout(r, ms)); export function setSleep(fn) { sleep = fn || ((ms) => new Promise((r) => setTimeout(r, ms))); } // 토큰 버킷: { tokens, updatedAt }. 초당 rate 만큼 채워지고 최대 rate 개까지만 쌓입니다(순간 폭주 허용 폭 = 1초 분량). const buckets = new Map(); function bucket(key, rate) { let b = buckets.get(key); if (!b) { b = { tokens: rate, updatedAt: clock(), rate }; buckets.set(key, b); } return b; } function refill(b) { const t = clock(); const dt = Math.max(0, t - b.updatedAt) / 1000; b.tokens = Math.min(b.rate, b.tokens + dt * b.rate); b.updatedAt = t; } /** 버킷에서 토큰 하나를 빼려고 시도. 성공하면 0, 아니면 다음 토큰까지 기다릴 ms. */ function take(key, rate) { const b = bucket(key, rate); refill(b); if (b.tokens >= 1) { b.tokens -= 1; return 0; } return Math.ceil(((1 - b.tokens) / b.rate) * 1000); } /** * 외부 전달 전 속도 조절. 도메인 버킷과 서버 버킷을 모두 통과할 때까지 기다립니다. * @param {string} domain 발신 도메인 * @throws {Error} mail_rate_limited — WAIT_MAX_MS 안에 차례가 오지 않음 */ export async function pace(domain) { const started = clock(); for (;;) { const w1 = take('domain:' + domain, DELIVERY_RATE); if (w1 === 0) { const w2 = take('server', DELIVERY_RATE); if (w2 === 0) return; // 서버 버킷이 비었으면 도메인 토큰을 돌려주고(공정성) 기다립니다. bucket('domain:' + domain, DELIVERY_RATE).tokens += 1; if (clock() - started + w2 > WAIT_MAX_MS) { const e = new Error('mail_rate_limited'); e.providerMessage = 'delivery rate limit (server)'; throw e; } await sleep(w2); continue; } if (clock() - started + w1 > WAIT_MAX_MS) { const e = new Error('mail_rate_limited'); e.providerMessage = 'delivery rate limit (domain)'; throw e; } await sleep(w1); } } // 제출 창: key → { count, windowStart, throttledUntil }. 1초 창에서 SUBMISSION_RATE 이상이면 제한. const submissions = new Map(); /** * 제출 1건을 세고 제한 여부를 돌려줍니다. 제한되면 abuse_flags 에 한 번 기록(같은 키·같은 제한 창에서는 한 번만). * @param {object} env DB 가 있으면 악용 표에 기록 * @param {string[]} keys 예: ['account:kim@x.com', 'ip:1.2.3.4'] * @returns {Promise<{throttled:boolean, key?:string, until?:string, rate?:number}>} */ export async function submit(env, keys) { const t = clock(); for (const key of keys) { let s = submissions.get(key); if (!s) { s = { count: 0, windowStart: t, throttledUntil: 0, flagged: false }; submissions.set(key, s); } if (s.throttledUntil > t) return { throttled: true, key, until: new Date(s.throttledUntil).toISOString(), rate: s.lastRate || 0 }; if (t - s.windowStart >= 1000) { s.count = 0; s.windowStart = t; s.flagged = false; } s.count += 1; if (s.count >= SUBMISSION_RATE) { s.throttledUntil = t + THROTTLE_WINDOW_MS; s.lastRate = s.count; if (!s.flagged && env?.DB) { s.flagged = true; try { await env.DB.prepare('INSERT INTO abuse_flags(id,kind,subject,detail,rate,created_at) VALUES(?,?,?,?,?,?)').bind(uid(), 'submission_rate', key, `초당 ${s.count}건 제출 — ${THROTTLE_WINDOW_MS / 1000}초 제한`, s.count, now()).run(); } catch { /* 표가 없는 옛 DB */ } } console.warn(`발송 제출 제한: ${key} 초당 ${s.count}건 (${THROTTLE_WINDOW_MS / 1000}초 동안 거부)`); return { throttled: true, key, until: new Date(s.throttledUntil).toISOString(), rate: s.count }; } } return { throttled: false }; } /** 관리자 화면용 현재 상태: 버킷(남은 토큰)·제출 창(최근 1초 제출량·제한 해제 시각). 비밀값 없음. */ export function rateSnapshot() { const t = clock(); const b = [...buckets.entries()].map(([key, x]) => { refill(x); return { key, tokens: Math.round(x.tokens * 100) / 100, rate: x.rate }; }); const s = [...submissions.entries()].map(([key, x]) => ({ key, countInWindow: t - x.windowStart >= 1000 ? 0 : x.count, throttledUntil: x.throttledUntil > t ? new Date(x.throttledUntil).toISOString() : null, lastRate: x.lastRate || 0 })); return { deliveryRate: DELIVERY_RATE, submissionRate: SUBMISSION_RATE, throttleWindowMs: THROTTLE_WINDOW_MS, buckets: b, submissions: s, at: new Date(t).toISOString() }; } /** 시험·운영자용: 상태 초기화 */ export function resetRates() { buckets.clear(); submissions.clear(); } // ------------------------------------------------------------------ 발송 대기열 /** 발송 직전 기록. 같은 send_key 가 이미 sent 면 null(= 이미 보냄). */ export async function enqueue(env, { domain, mailbox = '', key = null, payload }) { if (!env?.DB) return { id: null }; try { if (key) { const done = await env.DB.prepare("SELECT id, provider_id FROM outbound_queue WHERE send_key=? AND state='sent'").bind(key).first(); if (done) return { id: done.id, alreadySent: true, providerId: done.provider_id }; } const id = uid(); const stamp = now(); // 메일 본문·첨부는 넣지 않습니다(저장소에 이미 있음): 받는 사람 수·제목 길이·첨부 수만 두어 재시도 때 저장소의 원본으로 다시 만듭니다. const slim = { from: payload.from, to: payload.to, cc: payload.cc, bcc: payload.bcc, subject: String(payload.subject || '').slice(0, 200), attachments: (payload.attachments || []).length, messageId: payload.headers?.['Message-ID'] || null, ref: payload.__ref || null }; await env.DB.prepare('INSERT INTO outbound_queue(id,domain,mailbox,send_key,payload_json,state,attempts,created_at,updated_at) VALUES(?,?,?,?,?,\'sending\',1,?,?)').bind(id, domain, mailbox, key, JSON.stringify(slim), stamp, stamp).run(); return { id }; } catch { return { id: null }; } // 0009 마이그레이션 전의 DB 에서도 발송은 계속됩니다 } export async function finish(env, id, { providerId = null, error = null } = {}) { if (!id || !env?.DB) return; try { await env.DB.prepare('UPDATE outbound_queue SET state=?, provider_id=?, error=?, sent_at=?, updated_at=? WHERE id=?').bind(error ? 'failed' : 'sent', providerId, error ? String(error).slice(0, 300) : null, error ? null : now(), now(), id).run(); } catch { /* 무시 */ } } /** 시작 때: 지난 실행에서 'sending' 으로 남은 행을 interrupted 로 표시합니다. 돌아온 값은 그 수. */ export async function recoverQueue(env) { if (!env?.DB) return 0; try { const r = await env.DB.prepare("UPDATE outbound_queue SET state='interrupted', updated_at=? WHERE state='sending'").bind(now()).run(); return Number(r.meta?.changes || 0); } catch { return 0; } } /** 관리자 화면: 최근 대기열 행(실패·끊김 먼저). */ export async function queueList(env, { limit = 50 } = {}) { try { return (await env.DB.prepare("SELECT id, domain, mailbox, state, attempts, provider_id, error, created_at, updated_at, sent_at, payload_json FROM outbound_queue ORDER BY CASE state WHEN 'interrupted' THEN 0 WHEN 'failed' THEN 1 WHEN 'sending' THEN 2 ELSE 3 END, created_at DESC LIMIT ?").bind(limit).all()).results.map((r) => ({ ...r, payload: JSON.parse(r.payload_json || '{}'), payload_json: undefined })); } catch { return []; } } export async function abuseList(env, { open = true, limit = 100 } = {}) { try { return (await env.DB.prepare(`SELECT * FROM abuse_flags ${open ? 'WHERE reviewed_at IS NULL' : ''} ORDER BY created_at DESC LIMIT ?`).bind(limit).all()).results; } catch { return []; } } export async function reviewAbuse(env, id, by) { try { await env.DB.prepare('UPDATE abuse_flags SET reviewed_at=?, reviewed_by=? WHERE id=?').bind(now(), by, id).run(); } catch { /* 무시 */ } }