feat: switch OSS download quota to reservation plus log reconcile
This commit is contained in:
@@ -380,6 +380,41 @@ function initDatabase() {
|
||||
)
|
||||
`);
|
||||
|
||||
// 下载流量预扣保留表（直连下载签发时占用额度，不计入已用）
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS user_download_traffic_reservations (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
user_id INTEGER NOT NULL,
|
||||
source TEXT NOT NULL DEFAULT 'direct', -- direct/share_direct
|
||||
object_key TEXT,
|
||||
reserved_bytes INTEGER NOT NULL DEFAULT 0,
|
||||
remaining_bytes INTEGER NOT NULL DEFAULT 0,
|
||||
status TEXT NOT NULL DEFAULT 'pending', -- pending/confirmed/expired/cancelled
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
expires_at DATETIME NOT NULL,
|
||||
finalized_at DATETIME,
|
||||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
FOREIGN KEY (user_id) REFERENCES users (id) ON DELETE CASCADE
|
||||
)
|
||||
`);
|
||||
|
||||
// OSS 日志已处理记录（避免重复入账）
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS download_traffic_ingested_logs (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
bucket TEXT NOT NULL,
|
||||
log_key TEXT NOT NULL,
|
||||
etag TEXT,
|
||||
file_size INTEGER DEFAULT 0,
|
||||
line_count INTEGER DEFAULT 0,
|
||||
bytes_count INTEGER DEFAULT 0,
|
||||
status TEXT NOT NULL DEFAULT 'success',
|
||||
error_message TEXT,
|
||||
processed_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||||
UNIQUE(bucket, log_key, etag)
|
||||
)
|
||||
`);
|
||||
|
||||
// 日志表索引
|
||||
db.exec(`
|
||||
CREATE INDEX IF NOT EXISTS idx_logs_created_at ON system_logs(created_at);
|
||||
@@ -395,6 +430,14 @@ function initDatabase() {
|
||||
|
||||
-- 下载流量报表索引（按用户+日期查询）
|
||||
CREATE INDEX IF NOT EXISTS idx_download_traffic_user_date ON user_download_traffic_daily(user_id, date_key);
|
||||
|
||||
-- 下载预扣索引（按用户+状态+到期时间）
|
||||
CREATE INDEX IF NOT EXISTS idx_download_reservation_user_status_expires
|
||||
ON user_download_traffic_reservations(user_id, status, expires_at);
|
||||
|
||||
-- 已处理日志索引（按处理时间回溯）
|
||||
CREATE INDEX IF NOT EXISTS idx_ingested_logs_processed_at
|
||||
ON download_traffic_ingested_logs(processed_at);
|
||||
`);
|
||||
|
||||
console.log('[数据库性能优化] ✓ 日志表复合索引已创建');
|
||||
@@ -1537,6 +1580,154 @@ const DownloadTrafficReportDB = {
|
||||
}
|
||||
};
|
||||
|
||||
const DownloadTrafficReservationDB = {
|
||||
create({ userId, source = 'direct', objectKey = null, reservedBytes, expiresAt }) {
|
||||
const uid = Number(userId);
|
||||
const bytes = Math.floor(Number(reservedBytes));
|
||||
const expiresAtValue = typeof expiresAt === 'string' ? expiresAt : null;
|
||||
|
||||
if (!Number.isFinite(uid) || uid <= 0 || !Number.isFinite(bytes) || bytes <= 0 || !expiresAtValue) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const result = db.prepare(`
|
||||
INSERT INTO user_download_traffic_reservations (
|
||||
user_id, source, object_key, reserved_bytes, remaining_bytes,
|
||||
status, expires_at, created_at, updated_at
|
||||
) VALUES (?, ?, ?, ?, ?, 'pending', ?, datetime('now', 'localtime'), datetime('now', 'localtime'))
|
||||
`).run(uid, source, objectKey, bytes, bytes, expiresAtValue);
|
||||
|
||||
return db.prepare('SELECT * FROM user_download_traffic_reservations WHERE id = ?').get(result.lastInsertRowid);
|
||||
},
|
||||
|
||||
getPendingReservedBytes(userId) {
|
||||
const uid = Number(userId);
|
||||
if (!Number.isFinite(uid) || uid <= 0) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const row = db.prepare(`
|
||||
SELECT COALESCE(SUM(remaining_bytes), 0) AS reserved
|
||||
FROM user_download_traffic_reservations
|
||||
WHERE user_id = ?
|
||||
AND status = 'pending'
|
||||
AND expires_at > datetime('now', 'localtime')
|
||||
`).get(uid);
|
||||
|
||||
return Number(row?.reserved || 0);
|
||||
},
|
||||
|
||||
expirePendingReservations() {
|
||||
return db.prepare(`
|
||||
UPDATE user_download_traffic_reservations
|
||||
SET status = 'expired',
|
||||
remaining_bytes = 0,
|
||||
finalized_at = datetime('now', 'localtime'),
|
||||
updated_at = datetime('now', 'localtime')
|
||||
WHERE status = 'pending'
|
||||
AND expires_at <= datetime('now', 'localtime')
|
||||
`).run();
|
||||
},
|
||||
|
||||
consumePendingBytes(userId, bytes) {
|
||||
const uid = Number(userId);
|
||||
let remaining = Math.floor(Number(bytes));
|
||||
if (!Number.isFinite(uid) || uid <= 0 || !Number.isFinite(remaining) || remaining <= 0) {
|
||||
return { consumed: 0, finalizedCount: 0 };
|
||||
}
|
||||
|
||||
const pendingRows = db.prepare(`
|
||||
SELECT id, remaining_bytes
|
||||
FROM user_download_traffic_reservations
|
||||
WHERE user_id = ?
|
||||
AND status = 'pending'
|
||||
AND remaining_bytes > 0
|
||||
ORDER BY created_at ASC, id ASC
|
||||
`).all(uid);
|
||||
|
||||
let consumed = 0;
|
||||
let finalizedCount = 0;
|
||||
|
||||
for (const row of pendingRows) {
|
||||
if (remaining <= 0) break;
|
||||
|
||||
const rowRemaining = Number(row.remaining_bytes || 0);
|
||||
if (!Number.isFinite(rowRemaining) || rowRemaining <= 0) continue;
|
||||
|
||||
const useBytes = Math.min(remaining, rowRemaining);
|
||||
const nextRemaining = rowRemaining - useBytes;
|
||||
const nextStatus = nextRemaining <= 0 ? 'confirmed' : 'pending';
|
||||
|
||||
db.prepare(`
|
||||
UPDATE user_download_traffic_reservations
|
||||
SET remaining_bytes = ?,
|
||||
status = ?,
|
||||
finalized_at = CASE WHEN ? = 'confirmed' THEN datetime('now', 'localtime') ELSE finalized_at END,
|
||||
updated_at = datetime('now', 'localtime')
|
||||
WHERE id = ?
|
||||
`).run(nextRemaining, nextStatus, nextStatus, row.id);
|
||||
|
||||
consumed += useBytes;
|
||||
remaining -= useBytes;
|
||||
if (nextStatus === 'confirmed') {
|
||||
finalizedCount += 1;
|
||||
}
|
||||
}
|
||||
|
||||
return { consumed, finalizedCount };
|
||||
}
|
||||
};
|
||||
|
||||
const DownloadTrafficIngestDB = {
|
||||
isProcessed(bucket, logKey, etag = null) {
|
||||
const row = db.prepare(`
|
||||
SELECT id
|
||||
FROM download_traffic_ingested_logs
|
||||
WHERE bucket = ?
|
||||
AND log_key = ?
|
||||
AND COALESCE(etag, '') = COALESCE(?, '')
|
||||
LIMIT 1
|
||||
`).get(bucket, logKey, etag);
|
||||
|
||||
return !!row;
|
||||
},
|
||||
|
||||
markProcessed({
|
||||
bucket,
|
||||
logKey,
|
||||
etag = null,
|
||||
fileSize = 0,
|
||||
lineCount = 0,
|
||||
bytesCount = 0,
|
||||
status = 'success',
|
||||
errorMessage = null
|
||||
}) {
|
||||
return db.prepare(`
|
||||
INSERT INTO download_traffic_ingested_logs (
|
||||
bucket, log_key, etag, file_size, line_count, bytes_count,
|
||||
status, error_message, processed_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now', 'localtime'))
|
||||
ON CONFLICT(bucket, log_key, etag)
|
||||
DO UPDATE SET
|
||||
file_size = excluded.file_size,
|
||||
line_count = excluded.line_count,
|
||||
bytes_count = excluded.bytes_count,
|
||||
status = excluded.status,
|
||||
error_message = excluded.error_message,
|
||||
processed_at = datetime('now', 'localtime')
|
||||
`).run(
|
||||
bucket,
|
||||
logKey,
|
||||
etag || '',
|
||||
Number(fileSize) || 0,
|
||||
Number(lineCount) || 0,
|
||||
Number(bytesCount) || 0,
|
||||
status || 'success',
|
||||
errorMessage || null
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
// 系统日志操作
|
||||
const SystemLogDB = {
|
||||
// 日志级别常量
|
||||
@@ -1733,6 +1924,8 @@ module.exports = {
|
||||
VerificationDB,
|
||||
PasswordResetTokenDB,
|
||||
DownloadTrafficReportDB,
|
||||
DownloadTrafficReservationDB,
|
||||
DownloadTrafficIngestDB,
|
||||
SystemLogDB,
|
||||
TransactionDB,
|
||||
WalManager
|
||||
|
||||
Reference in New Issue
Block a user