OmniAI/omniai-server
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

merge into: OmniAI/omniai-server:codex/beta-task-reliability
Branches Tags
OmniAI/omniai-server:master OmniAI/omniai-server:codex/task-progress-contract OmniAI/omniai-server:codex/beta-task-reliability OmniAI/omniai-server:codex/task-reliability
..
pull from: OmniAI/omniai-server:codex/task-reliability
Branches Tags
OmniAI/omniai-server:codex/task-progress-contract OmniAI/omniai-server:codex/beta-task-reliability OmniAI/omniai-server:master OmniAI/omniai-server:codex/task-reliability

1 Commits
codex/beta-task-reliability .. codex/task-reliability

Author SHA1 Message Date
stringadmin 2249f7ce36 Harden generation task lifecycle 2026-06-04 21:08:14 +08:00
26 changed files with 452 additions and 2005 deletions
Expand all files Collapse all files
.env.example
-7
View File
@@ -20,13 +20,6 @@ JWT_EXPIRES_IN=7d
# Connection pool # Connection pool
PG_POOL_MAX=10 PG_POOL_MAX=10
# Provider polling reliability
# Shared across PM2 workers through Postgres-backed poll slots.
TASK_PROVIDER_POLL_MAX_CONCURRENCY=8
TASK_PROVIDER_POLL_SLOT_TTL_MS=30000
TASK_PROVIDER_POLL_REQUEST_TIMEOUT_MS=25000
GRSAI_IMAGE_SUBMIT_TIMEOUT_MS=30000
# CORS (comma separated allowed origins, * for all) # CORS (comma separated allowed origins, * for all)
CORS_ORIGINS=* CORS_ORIGINS=*
package.json
+1 -5
View File
@@ -14,11 +14,7 @@
"audit-routes": "node src/cli/auditModelRoutes.js", "audit-routes": "node src/cli/auditModelRoutes.js",
"import-config": "node src/cli/importConfig.js", "import-config": "node src/cli/importConfig.js",
"init-pools": "node src/cli/initPools.js", "init-pools": "node src/cli/initPools.js",
"test:community-routes": "node scripts/communityRouteContract.test.js", "test:community-routes": "node scripts/communityRouteContract.test.js"
"test:enterprise-video-pricing": "node scripts/enterpriseVideoPricingContract.test.js",
"test:key-manager": "node scripts/keyManagerReleaseContract.test.js",
"test:provider-poll-limiter": "node scripts/providerPollLimiterContract.test.js",
"test": "npm run test:community-routes && npm run test:enterprise-video-pricing && npm run test:key-manager && npm run test:provider-poll-limiter"
}, },
"dependencies": { "dependencies": {
"alipay-sdk": "^4.14.0", "alipay-sdk": "^4.14.0",
scripts/enterpriseVideoPricingContract.test.js
-54
View File
@@ -1,54 +0,0 @@
const assert = require("node:assert/strict");
const {
calculateEnterpriseVideoCredits,
getEnterpriseVideoCreditRate,
getEnterpriseVideoPricingConfig,
} = require("../src/enterpriseVideoBilling");
function getRule(config, id) {
const rule = config.rules.find((item) => item.id === id);
assert(rule, `missing enterprise video pricing rule: ${id}`);
return rule;
}
const config = getEnterpriseVideoPricingConfig();
assert.equal(config.currency, "CNY");
assert.equal(config.creditsPerCny, 100);
assert.equal(config.billingUnit, "per_second");
assert.deepEqual(config.resolutions, ["720P", "1080P"]);
assert.equal(getRule(config, "happyhorse").rates["720P"], 0.72);
assert.equal(getRule(config, "happyhorse").rates["1080P"], 1.28);
assert.equal(getRule(config, "wanxiang-i2v").rates["720P"], 0.6);
assert.equal(getRule(config, "kling-muted").rates["1080P"], 0.8);
assert.equal(
getEnterpriseVideoCreditRate({
model: "happyhorse-1.0",
resolution: "1080P",
}),
getRule(config, "happyhorse").rates["1080P"],
);
assert.equal(
getEnterpriseVideoCreditRate({
model: "kling-3.0-dashscope",
resolution: "720P",
muted: true,
hasReferenceVideo: false,
}),
getRule(config, "kling-muted").rates["720P"],
);
assert.equal(
calculateEnterpriseVideoCredits({
model: "vidu-q3-turbo",
resolution: "1080P",
durationSeconds: 5,
}),
500,
);
console.log("enterprise video pricing contract tests passed");
scripts/keyManagerReleaseContract.test.js
-73
View File
@@ -1,73 +0,0 @@
const assert = require("node:assert/strict");
const { createRequire } = require("node:module");
const nodeRequire = createRequire(__filename);
function loadKeyManagerWithPool(pool) {
const dbPath = nodeRequire.resolve("../src/db");
const keyManagerPath = nodeRequire.resolve("../src/keyManager");
const originalDbModule = nodeRequire.cache[dbPath];
const originalKeyManagerModule = nodeRequire.cache[keyManagerPath];
delete nodeRequire.cache[keyManagerPath];
nodeRequire.cache[dbPath] = {
id: dbPath,
filename: dbPath,
loaded: true,
exports: {
pool,
withTransaction: async (fn) => fn(pool),
},
};
return {
keyManager: nodeRequire("../src/keyManager"),
restore() {
delete nodeRequire.cache[keyManagerPath];
if (originalKeyManagerModule) nodeRequire.cache[keyManagerPath] = originalKeyManagerModule;
if (originalDbModule) nodeRequire.cache[dbPath] = originalDbModule;
else delete nodeRequire.cache[dbPath];
},
};
}
function createReleasePool() {
const calls = [];
return {
calls,
async query(sql, params) {
calls.push({ sql, params });
if (/WITH candidate AS/i.test(sql)) {
return {
rows: [{
id: 10,
key_id: 20,
lease_user_id: 30,
lease_enterprise_id: 40,
provider: "dashscope",
}],
};
}
if (/UPDATE api_keys SET active_count/i.test(sql)) return { rows: [] };
if (/INSERT INTO usage_logs/i.test(sql)) return { rows: [] };
throw new Error(`Unexpected SQL: ${sql}`);
},
};
}
(async () => {
const pool = createReleasePool();
const { keyManager, restore } = loadKeyManagerWithPool(pool);
try {
const result = await keyManager.releaseKey("lease-token-without-user-context");
assert.equal(result.released, true);
const usageLogCall = pool.calls.find((call) => /INSERT INTO usage_logs/i.test(call.sql));
assert.deepEqual(usageLogCall.params, [30, 40, 20, 20, "release"]);
} finally {
restore();
}
})().catch((error) => {
console.error(error);
process.exitCode = 1;
});
scripts/providerPollLimiterContract.test.js
-96
View File
@@ -1,96 +0,0 @@
const assert = require("node:assert/strict");
const { createRequire } = require("node:module");
const nodeRequire = createRequire(__filename);
function loadLimiterWithPool(pool) {
const dbPath = nodeRequire.resolve("../src/db");
const limiterPath = nodeRequire.resolve("../src/providerPollLimiter");
const originalDbModule = nodeRequire.cache[dbPath];
const originalLimiterModule = nodeRequire.cache[limiterPath];
delete nodeRequire.cache[limiterPath];
nodeRequire.cache[dbPath] = {
id: dbPath,
filename: dbPath,
loaded: true,
exports: { pool },
};
return {
limiter: nodeRequire("../src/providerPollLimiter"),
restore() {
delete nodeRequire.cache[limiterPath];
if (originalLimiterModule) nodeRequire.cache[limiterPath] = originalLimiterModule;
if (originalDbModule) nodeRequire.cache[dbPath] = originalDbModule;
else delete nodeRequire.cache[dbPath];
},
};
}
function createPool(options = {}) {
const calls = [];
return {
calls,
async query(sql, params = []) {
calls.push({ sql, params });
if (/CREATE TABLE IF NOT EXISTS generation_provider_poll_slots/i.test(sql)) return { rows: [] };
if (/WITH candidate AS/i.test(sql)) {
if (options.noAvailableSlot) return { rows: [] };
return { rows: [{ scope: params[0], slot_no: 2 }] };
}
if (/DELETE FROM generation_provider_poll_slots/i.test(sql)) return { rows: [] };
throw new Error(`Unexpected SQL: ${sql}`);
},
};
}
(async () => {
const previousLimit = process.env.TASK_PROVIDER_POLL_MAX_CONCURRENCY;
process.env.TASK_PROVIDER_POLL_MAX_CONCURRENCY = "3";
const pool = createPool();
const { limiter, restore } = loadLimiterWithPool(pool);
try {
const outcome = await limiter.withProviderPollSlot(101, async () => "polled");
assert.equal(outcome.acquired, true);
assert.equal(outcome.value, "polled");
const acquireCall = pool.calls.find((call) => /WITH candidate AS/i.test(call.sql));
assert.equal(acquireCall.params[1], 3);
assert.equal(acquireCall.params[3], 101);
const releaseCall = pool.calls.find((call) => /DELETE FROM generation_provider_poll_slots/i.test(call.sql));
assert.equal(releaseCall.params[0], acquireCall.params[0]);
assert.equal(releaseCall.params[1], 2);
assert.equal(releaseCall.params[2], acquireCall.params[2]);
} finally {
if (previousLimit === undefined) delete process.env.TASK_PROVIDER_POLL_MAX_CONCURRENCY;
else process.env.TASK_PROVIDER_POLL_MAX_CONCURRENCY = previousLimit;
restore();
}
const saturatedPool = createPool({ noAvailableSlot: true });
const { limiter: saturatedLimiter, restore: restoreSaturated } = loadLimiterWithPool(saturatedPool);
try {
let called = false;
const outcome = await saturatedLimiter.withProviderPollSlot(202, async () => {
called = true;
return "should-not-run";
});
assert.equal(outcome.acquired, false);
assert.equal(outcome.value, undefined);
assert.equal(called, false);
assert.equal(
saturatedPool.calls.some((call) => /DELETE FROM generation_provider_poll_slots/i.test(call.sql)),
false,
);
} finally {
restoreSaturated();
}
})().catch((error) => {
console.error(error);
process.exitCode = 1;
});
src/aiTaskWorker.js
+260 -489
View File
@@ -5,7 +5,8 @@ const { EventEmitter } = require("node:events");
const { pool } = require("./db"); const { pool } = require("./db");
const { refundTaskBillingOnFailure } = require("./billing"); const { refundTaskBillingOnFailure } = require("./billing");
const { putObject, isOssConfigured } = require("./ossClient"); const { putObject, isOssConfigured } = require("./ossClient");
const { withProviderPollSlot } = require("./providerPollLimiter"); const keyManager = require("./keyManager");
const { resolveImageProviderCandidates, resolveVideoProvider } = require("./aiProviderRouter");
const taskEvents = new EventEmitter(); const taskEvents = new EventEmitter();
taskEvents.setMaxListeners(200); taskEvents.setMaxListeners(200);
@@ -14,17 +15,10 @@ const activePollers = new Map();
const POLL_INTERVAL_MS = 3000; const POLL_INTERVAL_MS = 3000;
const MAX_POLL_ATTEMPTS = 120; const MAX_POLL_ATTEMPTS = 120;
const GRS_IMAGE_MAX_POLL_ATTEMPTS = Number(process.env.GRSAI_IMAGE_MAX_POLL_ATTEMPTS || 60); const GRS_IMAGE_MAX_POLL_ATTEMPTS = Number(process.env.GRSAI_IMAGE_MAX_POLL_ATTEMPTS || 60);
const TASK_EVENT_CHANNEL = "generation_task_events"; const STALE_TASK_TIMEOUT_MINUTES = Math.max(10, Number(process.env.STALE_GENERATION_TASK_MINUTES || 120));
const TASK_EVENT_ORIGIN = `${process.pid}-${crypto.randomUUID()}`; const RESULT_PERSIST_RETRY_LIMIT = Math.max(1, Number(process.env.RESULT_PERSIST_RETRY_LIMIT || 5));
const POLLER_OWNER_ID = `${process.pid}-${crypto.randomUUID()}`; const RESULT_PERSIST_RETRY_BATCH_SIZE = Math.max(1, Number(process.env.RESULT_PERSIST_RETRY_BATCH_SIZE || 25));
const POLLER_OWNER_STALE_MS = Number(process.env.TASK_POLLER_OWNER_STALE_MS || 20_000); const TASK_STARTUP_RECOVERY_LIMIT = Math.max(1, Number(process.env.TASK_STARTUP_RECOVERY_LIMIT || 50));
const POLLER_RECOVERY_INTERVAL_MS = Number(process.env.TASK_POLLER_RECOVERY_INTERVAL_MS || 30_000);
const PROVIDER_POLL_REQUEST_TIMEOUT_MS = Number(process.env.TASK_PROVIDER_POLL_REQUEST_TIMEOUT_MS || 25_000);
let taskEventListenerClient = null;
let taskEventListenerStarting = null;
let pollerStoreReady = null;
let pollerRecoveryTimer = null;
let staleTaskCleanupStartupTimer = null;
function normalizeTaskProgress(value) { function normalizeTaskProgress(value) {
const numeric = Number(value); const numeric = Number(value);
@@ -42,173 +36,6 @@ function formatTaskEvent(row) {
}; };
} }
function emitTaskEvent(event) {
if (!event?.taskId) return;
taskEvents.emit(`task:${event.taskId}`, event);
}
async function publishTaskEvent(event) {
if (!event?.taskId) return;
emitTaskEvent(event);
try {
await pool.query("SELECT pg_notify($1, $2)", [
TASK_EVENT_CHANNEL,
JSON.stringify({ origin: TASK_EVENT_ORIGIN, event }),
]);
} catch (err) {
console.error(`[aiTaskWorker] task event publish failed for task ${event.taskId}:`, err.message);
}
}
function serializeProviderConfig(providerConfig) {
if (!providerConfig || typeof providerConfig !== "object") return {};
const allowedKeys = [
"provider",
"transport",
"protocol",
"baseUrl",
"endpoint",
"resultEndpoint",
"model",
"requestedModel",
];
const result = {};
for (const key of allowedKeys) {
if (providerConfig[key] !== undefined) result[key] = providerConfig[key];
}
return result;
}
function parseProviderConfig(value) {
if (!value) return {};
if (typeof value === "object") return value;
try {
const parsed = JSON.parse(value);
return parsed && typeof parsed === "object" ? parsed : {};
} catch {
return {};
}
}
async function ensureTaskPollerStore() {
if (pollerStoreReady) return pollerStoreReady;
pollerStoreReady = pool.query(`
CREATE TABLE IF NOT EXISTS generation_task_pollers (
task_id INTEGER PRIMARY KEY REFERENCES generation_tasks(id) ON DELETE CASCADE,
provider_task_id TEXT NOT NULL,
task_type TEXT NOT NULL,
provider_config_json TEXT NOT NULL,
lease_token TEXT,
owner_id TEXT,
owner_heartbeat_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_generation_task_pollers_owner
ON generation_task_pollers(owner_heartbeat_at);
`).catch((err) => {
pollerStoreReady = null;
throw err;
});
return pollerStoreReady;
}
async function persistPollerState(taskDbId, { providerTaskId, type, providerConfig, leaseToken }) {
await ensureTaskPollerStore();
await pool.query(
`
INSERT INTO generation_task_pollers (
task_id, provider_task_id, task_type, provider_config_json, lease_token,
owner_id, owner_heartbeat_at, updated_at
)
VALUES ($1, $2, $3, $4, $5, $6, NOW(), NOW())
ON CONFLICT (task_id) DO UPDATE SET
provider_task_id = EXCLUDED.provider_task_id,
task_type = EXCLUDED.task_type,
provider_config_json = EXCLUDED.provider_config_json,
lease_token = EXCLUDED.lease_token,
owner_id = EXCLUDED.owner_id,
owner_heartbeat_at = NOW(),
updated_at = NOW()
`,
[
taskDbId,
providerTaskId,
type,
JSON.stringify(serializeProviderConfig(providerConfig)),
leaseToken || null,
POLLER_OWNER_ID,
],
);
}
async function refreshPollerHeartbeat(taskDbId) {
await ensureTaskPollerStore();
await pool.query(
"UPDATE generation_task_pollers SET owner_id = $1, owner_heartbeat_at = NOW(), updated_at = NOW() WHERE task_id = $2",
[POLLER_OWNER_ID, taskDbId],
);
}
async function clearPollerState(taskDbId) {
await ensureTaskPollerStore();
await pool.query("DELETE FROM generation_task_pollers WHERE task_id = $1", [taskDbId]);
}
async function orphanOwnedPollerState() {
await ensureTaskPollerStore();
await pool.query(
"UPDATE generation_task_pollers SET owner_id = NULL, owner_heartbeat_at = NULL, updated_at = NOW() WHERE owner_id = $1",
[POLLER_OWNER_ID],
);
}
async function getPersistedLeaseToken(taskDbId) {
await ensureTaskPollerStore();
const { rows } = await pool.query(
"SELECT lease_token FROM generation_task_pollers WHERE task_id = $1 LIMIT 1",
[taskDbId],
);
return rows[0]?.lease_token || null;
}
async function getLeaseKey(leaseToken) {
if (!leaseToken) return null;
const { rows } = await pool.query(
`
SELECT k.api_key
FROM key_leases l
JOIN api_keys k ON k.id = l.key_id
WHERE l.lease_token = $1
AND l.released_at IS NULL
AND k.enabled = 1
LIMIT 1
`,
[leaseToken],
);
const apiKey = rows[0]?.api_key;
return apiKey === "pool-slot" ? "" : apiKey || null;
}
async function claimPoller(taskId) {
await ensureTaskPollerStore();
const staleInterval = `${Math.max(5, Math.ceil(POLLER_OWNER_STALE_MS / 1000))} seconds`;
const { rows } = await pool.query(
`
UPDATE generation_task_pollers
SET owner_id = $1, owner_heartbeat_at = NOW(), updated_at = NOW()
WHERE task_id = $2
AND (
owner_heartbeat_at IS NULL
OR owner_heartbeat_at < NOW() - ($3::text)::interval
)
RETURNING *
`,
[POLLER_OWNER_ID, taskId, staleInterval],
);
return rows[0] || null;
}
async function createTaskLifecycleNotification(task) { async function createTaskLifecycleNotification(task) {
if (!task || !task.user_id || !task.id) return; if (!task || !task.user_id || !task.id) return;
@@ -271,14 +98,15 @@ async function updateTaskInDb(taskId, updates) {
if (fields.length === 0) return; if (fields.length === 0) return;
values.push(taskId); values.push(taskId);
const protectCancelled = nextUpdates.status !== "cancelled" ? " AND status <> 'cancelled'" : "";
const { rows } = await pool.query( const { rows } = await pool.query(
`UPDATE generation_tasks SET ${fields.join(", ")} WHERE id = $${idx} RETURNING *`, `UPDATE generation_tasks SET ${fields.join(", ")} WHERE id = $${idx}${protectCancelled} RETURNING *`,
values, values,
); );
let updatedTask = rows[0]; let updatedTask = rows[0];
if (updatedTask) { if (updatedTask) {
await publishTaskEvent(formatTaskEvent(updatedTask)); taskEvents.emit(`task:${taskId}`, formatTaskEvent(updatedTask));
} }
if (nextUpdates.status === "completed" && updatedTask?.result_url) { if (nextUpdates.status === "completed" && updatedTask?.result_url) {
@@ -291,12 +119,6 @@ async function updateTaskInDb(taskId, updates) {
}); });
} }
if (nextUpdates.status === "completed") {
await markTaskBillingAccepted(taskId).catch((err) => {
console.error(`[aiTaskWorker] billing accept error for task ${taskId}:`, err.message);
});
}
if (nextUpdates.status === "failed") { if (nextUpdates.status === "failed") {
await refundTaskBillingOnFailure(taskId).catch((err) => { await refundTaskBillingOnFailure(taskId).catch((err) => {
console.error(`[aiTaskWorker] refund error for task ${taskId}:`, err.message); console.error(`[aiTaskWorker] refund error for task ${taskId}:`, err.message);
@@ -304,32 +126,71 @@ async function updateTaskInDb(taskId, updates) {
} }
} }
async function markTaskBillingAccepted(taskId) {
await pool.query(
"UPDATE credit_ledger SET status = 'charged', updated_at = NOW() WHERE task_id = $1 AND status = 'reserved'",
[taskId],
);
}
function persistTaskResultUrlToOssInBackground(task) { function persistTaskResultUrlToOssInBackground(task) {
if (!task?.id || !task?.result_url) return; if (!task?.id || !task?.result_url) return;
Promise.resolve() Promise.resolve()
.then(async () => { .then(async () => {
const durableUrl = await persistResultUrlToOss(task); await persistTaskResultUrlToOss(task);
if (!durableUrl || durableUrl === task.result_url) return;
await pool.query(
"UPDATE generation_tasks SET result_url = $1, updated_at = NOW() WHERE id = $2 AND result_url = $3",
[durableUrl, task.id, task.result_url],
);
console.info(`[aiTaskWorker] task ${task.id} result persisted to OSS after completion`);
}) })
.catch((error) => { .catch((error) => {
console.warn(`[aiTaskWorker] background result persistence failed for task ${task.id}:`, error.message); console.warn(`[aiTaskWorker] background result persistence failed for task ${task.id}:`, error.message);
}); });
} }
async function markResultPersistence(taskId, status, error = null, durableUrl = null, previousUrl = null) {
const fields = [
"result_persist_status = $1",
"result_persist_attempts = result_persist_attempts + 1",
"result_persist_error = $2",
"updated_at = NOW()",
];
const values = [status, error ? String(error).slice(0, 1000) : null];
let idx = values.length + 1;
if (status === "succeeded") {
fields.push("result_persisted_at = NOW()");
}
if (durableUrl) {
fields.push(`result_url = $${idx++}`);
values.push(durableUrl);
}
values.push(taskId);
let where = `id = $${idx}`;
if (previousUrl) {
idx += 1;
values.push(previousUrl);
where += ` AND result_url = $${idx}`;
}
await pool.query(`UPDATE generation_tasks SET ${fields.join(", ")} WHERE ${where}`, values);
}
async function persistTaskResultUrlToOss(task) {
if (!task?.id || !task?.result_url) return null;
if (isOwnPersistedResultUrl(task.result_url)) {
await markResultPersistence(task.id, "succeeded", null, null);
return task.result_url;
}
if (!isOssConfigured()) {
await markResultPersistence(task.id, "failed", "OSS is not configured");
return null;
}
const durableUrl = await persistResultUrlToOss(task);
if (!durableUrl) {
await markResultPersistence(task.id, "failed", "Result URL could not be copied to OSS");
return null;
}
await markResultPersistence(task.id, "succeeded", null, durableUrl, task.result_url);
console.info(`[aiTaskWorker] task ${task.id} result persisted to OSS after completion`);
return durableUrl;
}
function asObject(value) { function asObject(value) {
return value && typeof value === "object" && !Array.isArray(value) ? value : undefined; return value && typeof value === "object" && !Array.isArray(value) ? value : undefined;
} }
@@ -665,22 +526,9 @@ function extractErrorMessage(json, fallback) {
} }
async function fetchJson(url, headers) { async function fetchJson(url, headers) {
const controller = new AbortController(); const res = await fetch(url, { method: "GET", headers });
const timeoutMs = Number.isFinite(PROVIDER_POLL_REQUEST_TIMEOUT_MS) && PROVIDER_POLL_REQUEST_TIMEOUT_MS > 0
? PROVIDER_POLL_REQUEST_TIMEOUT_MS
: 25_000;
const timer = setTimeout(() => controller.abort(), timeoutMs);
timer.unref?.();
try {
const res = await fetch(url, { method: "GET", headers, signal: controller.signal });
if (!res.ok) return { ok: false, json: null }; if (!res.ok) return { ok: false, json: null };
return { ok: true, json: await res.json() }; return { ok: true, json: await res.json() };
} catch (err) {
return { ok: false, json: null, error: err };
} finally {
clearTimeout(timer);
}
} }
async function pollGrsaiImage(_taskId, providerTaskId, apiKey, baseUrl, resultEndpoint) { async function pollGrsaiImage(_taskId, providerTaskId, apiKey, baseUrl, resultEndpoint) {
@@ -841,29 +689,28 @@ function getMaxPollAttempts(type, providerConfig) {
return MAX_POLL_ATTEMPTS; return MAX_POLL_ATTEMPTS;
} }
function startPolling(taskDbId, { providerTaskId, apiKey, type, providerConfig, leaseToken, keyManager, onTaskFailed, skipPersist = false }) { async function releasePollingLease(poller) {
if (activePollers.has(taskDbId)) return; if (!poller?.leaseToken || !poller?.keyManager) return;
if (!skipPersist) { await poller.keyManager.releaseKey(poller.leaseToken).catch((err) => {
persistPollerState(taskDbId, { providerTaskId, type, providerConfig, leaseToken }).catch((err) => { console.warn(`[aiTaskWorker] release lease failed for task ${poller.taskDbId}:`, err.message);
console.error(`[aiTaskWorker] failed to persist poller state for task ${taskDbId}:`, err.message);
}); });
} }
function startPolling(taskDbId, { providerTaskId, apiKey, type, providerConfig, leaseToken, keyManager, onTaskFailed }) {
if (activePollers.has(taskDbId)) return;
let attempts = 0; let attempts = 0;
let polling = false; let polling = false;
let skippedPolls = 0;
const maxPollAttempts = getMaxPollAttempts(type, providerConfig); const maxPollAttempts = getMaxPollAttempts(type, providerConfig);
const interval = setInterval(async () => { const interval = setInterval(async () => {
if (polling) return; if (polling) return;
polling = true; polling = true;
attempts++;
try { try {
if (attempts >= maxPollAttempts) { if (attempts > maxPollAttempts) {
clearInterval(interval); await stopPolling(taskDbId, { releaseLease: true });
activePollers.delete(taskDbId);
if (leaseToken && keyManager) await keyManager.releaseKey(leaseToken).catch(() => {});
if (typeof onTaskFailed === "function") { if (typeof onTaskFailed === "function") {
await clearPollerState(taskDbId).catch(() => {});
const handled = await onTaskFailed("Task timed out").catch((fallbackErr) => { const handled = await onTaskFailed("Task timed out").catch((fallbackErr) => {
console.error(`[aiTaskWorker] fallback error for task ${taskDbId}:`, fallbackErr.message); console.error(`[aiTaskWorker] fallback error for task ${taskDbId}:`, fallbackErr.message);
return false; return false;
@@ -871,52 +718,29 @@ function startPolling(taskDbId, { providerTaskId, apiKey, type, providerConfig,
if (handled) return; if (handled) return;
} }
await updateTaskInDb(taskDbId, { status: "failed", error: "Task timed out" }); await updateTaskInDb(taskDbId, { status: "failed", error: "Task timed out" });
await clearPollerState(taskDbId).catch(() => {});
return; return;
} }
// Check if task was cancelled by user
const { rows: [taskRow] } = await pool.query("SELECT status FROM generation_tasks WHERE id = $1", [taskDbId]); const { rows: [taskRow] } = await pool.query("SELECT status FROM generation_tasks WHERE id = $1", [taskDbId]);
if (!taskRow || taskRow.status === "cancelled") { if (!taskRow || taskRow.status === "cancelled") {
clearInterval(interval); await stopPolling(taskDbId, { releaseLease: true });
activePollers.delete(taskDbId);
await clearPollerState(taskDbId).catch(() => {});
if (leaseToken && keyManager) await keyManager.releaseKey(leaseToken).catch(() => {});
return; return;
} }
await refreshPollerHeartbeat(taskDbId).catch(() => {});
const pollOutcome = await withProviderPollSlot(taskDbId, async () => { let result;
attempts++;
if (type === "image") { if (type === "image") {
if (providerConfig.transport === "dashscope-image") { if (providerConfig.transport === "dashscope-image") {
return pollDashscopeImage(taskDbId, providerTaskId, apiKey); result = await pollDashscopeImage(taskDbId, providerTaskId, apiKey);
} else {
result = await pollGrsaiImage(taskDbId, providerTaskId, apiKey, providerConfig.baseUrl, providerConfig.resultEndpoint || "/result");
} }
return pollGrsaiImage(taskDbId, providerTaskId, apiKey, providerConfig.baseUrl, providerConfig.resultEndpoint || "/result"); } else {
} result = await pollVideoTask(taskDbId, providerTaskId, apiKey, providerConfig);
return pollVideoTask(taskDbId, providerTaskId, apiKey, providerConfig);
});
if (!pollOutcome.acquired) {
skippedPolls++;
if (skippedPolls % 20 === 0) {
console.info(`[aiTaskWorker] task ${taskDbId} waiting for provider poll slot (skipped=${skippedPolls})`);
}
return;
}
skippedPolls = 0;
const result = pollOutcome.value;
if (!result) {
return;
} }
if (result.status === "completed" || result.status === "failed") { if (result.status === "completed" || result.status === "failed") {
clearInterval(interval); await stopPolling(taskDbId, { releaseLease: true });
activePollers.delete(taskDbId);
if (leaseToken && keyManager) await keyManager.releaseKey(leaseToken).catch(() => {});
if (result.status === "failed" && typeof onTaskFailed === "function") { if (result.status === "failed" && typeof onTaskFailed === "function") {
await clearPollerState(taskDbId).catch(() => {});
const handled = await onTaskFailed(result.error || "Task failed").catch((fallbackErr) => { const handled = await onTaskFailed(result.error || "Task failed").catch((fallbackErr) => {
console.error(`[aiTaskWorker] fallback error for task ${taskDbId}:`, fallbackErr.message); console.error(`[aiTaskWorker] fallback error for task ${taskDbId}:`, fallbackErr.message);
return false; return false;
@@ -926,9 +750,6 @@ function startPolling(taskDbId, { providerTaskId, apiKey, type, providerConfig,
} }
await updateTaskInDb(taskDbId, result); await updateTaskInDb(taskDbId, result);
if (result.status === "completed" || result.status === "failed") {
await clearPollerState(taskDbId).catch(() => {});
}
} catch (err) { } catch (err) {
console.error(`[aiTaskWorker] poll error for task ${taskDbId}:`, err.message); console.error(`[aiTaskWorker] poll error for task ${taskDbId}:`, err.message);
} finally { } finally {
@@ -936,288 +757,242 @@ function startPolling(taskDbId, { providerTaskId, apiKey, type, providerConfig,
} }
}, POLL_INTERVAL_MS); }, POLL_INTERVAL_MS);
activePollers.set(taskDbId, { interval, leaseToken }); activePollers.set(taskDbId, { taskDbId, interval, leaseToken, keyManager });
} }
function stopPolling(taskDbId) { async function stopPolling(taskDbId, options = {}) {
const poller = activePollers.get(taskDbId); const poller = activePollers.get(taskDbId);
if (poller) { if (!poller) return;
clearInterval(poller.interval); clearInterval(poller.interval);
activePollers.delete(taskDbId); activePollers.delete(taskDbId);
if (options.releaseLease) {
await releasePollingLease(poller);
} }
clearPollerState(taskDbId).catch(() => {});
} }
async function cancelTaskRuntimeState(taskDbId, keyManager) { async function cancelTask(taskId, userId) {
const poller = activePollers.get(taskDbId); const { rows } = await pool.query(
if (poller) { `UPDATE generation_tasks
clearInterval(poller.interval); SET status = 'cancelled', completed_at = NOW(), updated_at = NOW()
activePollers.delete(taskDbId); WHERE id = $1 AND user_id = $2 AND status IN ('pending', 'running')
} RETURNING *`,
[taskId, userId],
);
const task = rows[0];
if (!task) return null;
const leaseToken = poller?.leaseToken || await getPersistedLeaseToken(taskDbId).catch(() => null); await stopPolling(task.id, { releaseLease: true });
await clearPollerState(taskDbId).catch(() => {}); taskEvents.emit(`task:${task.id}`, formatTaskEvent(task));
if (leaseToken && keyManager) { return task;
await keyManager.releaseKey(leaseToken).catch((err) => {
console.error(`[aiTaskWorker] failed to release lease for cancelled task ${taskDbId}:`, err.message);
});
}
await publishTaskEvent({
taskId: taskDbId,
status: "cancelled",
progress: 100,
resultUrl: null,
error: "任务已取消",
});
} }
function getActiveCount() { function getActiveCount() {
return activePollers.size; return activePollers.size;
} }
async function recoverRunnablePollers() {
await ensureTaskPollerStore();
const staleInterval = `${Math.max(5, Math.ceil(POLLER_OWNER_STALE_MS / 1000))} seconds`;
const { rows } = await pool.query(
`
SELECT p.task_id, p.updated_at
FROM generation_task_pollers p
JOIN generation_tasks t ON t.id = p.task_id
WHERE t.status IN ('pending', 'running')
AND (
p.owner_heartbeat_at IS NULL
OR p.owner_heartbeat_at < NOW() - ($1::text)::interval
)
ORDER BY p.owner_heartbeat_at NULLS FIRST, p.updated_at ASC
LIMIT 20
`,
[staleInterval],
);
for (const row of rows) {
const taskId = row.task_id;
if (activePollers.has(taskId)) continue;
const poller = await claimPoller(taskId);
if (!poller || activePollers.has(taskId)) continue;
const apiKey = await getLeaseKey(poller.lease_token);
if (apiKey == null) {
console.warn(`[aiTaskWorker] cannot recover task ${taskId}: active lease not found`);
await releaseUnrecoverableTask(taskId, "任务执行状态已失效，已自动释放");
continue;
}
console.info(`[aiTaskWorker] recovering poller for task ${taskId}`);
startPolling(taskId, {
providerTaskId: poller.provider_task_id,
apiKey,
type: poller.task_type,
providerConfig: parseProviderConfig(poller.provider_config_json),
leaseToken: poller.lease_token,
keyManager: require("./keyManager"),
onTaskFailed: async (failureMessage) => {
await updateTaskInDb(taskId, { status: "failed", error: failureMessage || "Task failed" });
return true;
},
skipPersist: true,
});
}
}
async function releaseUnrecoverableTask(taskId, message) {
const { rows } = await pool.query(
`
UPDATE generation_tasks t
SET status = 'failed', error = $2, completed_at = NOW(), updated_at = NOW()
FROM generation_task_pollers p
WHERE t.id = $1
AND p.task_id = t.id
AND p.owner_id = $3
AND t.status IN ('pending', 'running')
RETURNING t.*
`,
[taskId, message, POLLER_OWNER_ID],
);
const task = rows[0];
if (!task) return false;
const leaseToken = await getPersistedLeaseToken(taskId).catch(() => null);
await clearPollerState(taskId).catch(() => {});
if (leaseToken) {
await require("./keyManager").releaseKey(leaseToken).catch((err) => {
console.error(`[aiTaskWorker] failed to release lease for unrecoverable task ${taskId}:`, err.message);
});
}
await publishTaskEvent(formatTaskEvent(task));
await createTaskLifecycleNotification(task).catch((err) => {
console.error(`[aiTaskWorker] notification error for unrecoverable task ${taskId}:`, err.message);
});
await refundTaskBillingOnFailure(taskId).catch((err) => {
console.error(`[aiTaskWorker] refund error for unrecoverable task ${taskId}:`, err.message);
});
console.warn(`[aiTaskWorker] released unrecoverable task ${taskId}: ${message}`);
return true;
}
// --- Periodic stale task cleanup ---
// Runs every 5 minutes, marks tasks stuck in 'pending'/'running' for too long as 'failed'.
// This catches cases where the worker crashed, the provider API never responded,
// or the cancel request failed silently on the client side.
const STALE_TASK_CLEANUP_INTERVAL_MS = 5 * 60 * 1000; const STALE_TASK_CLEANUP_INTERVAL_MS = 5 * 60 * 1000;
let staleTaskCleanupTimer = null; let staleTaskCleanupTimer = null;
const TASK_RESULT_PERSIST_RETRY_INTERVAL_MS = 5 * 60 * 1000;
let taskResultPersistenceRetryTimer = null;
let taskStartupRecoveryTimer = null;
let taskStaleCleanupRunning = false;
let taskResultPersistenceRetryRunning = false;
let taskStartupRecoveryRunning = false;
function parseTaskParams(paramsJson) {
if (!paramsJson) return {};
if (typeof paramsJson === "object") return paramsJson;
try {
const parsed = JSON.parse(paramsJson);
return parsed && typeof parsed === "object" ? parsed : {};
} catch {
return {};
}
}
function resolveProviderConfigForRecovery(task) {
const params = parseTaskParams(task.params_json);
if (task.type === "video") {
if (params.model === "video-style-transform" || params.operation === "video-style-super-resolution") {
return { provider: "dashscope", protocol: "wan-i2v", baseUrl: "https://dashscope.aliyuncs.com" };
}
if (params.model === "aliyun-video-super-resolve" || params.model === "aliyun-erase-subtitles") {
return null;
}
return resolveVideoProvider(params.model);
}
if (task.type === "image") {
if (params.operation === "image-super-resolution" || params.operation === "image-edit") {
return { provider: "dashscope", transport: "dashscope-image" };
}
const candidates = resolveImageProviderCandidates(params.model);
return candidates[0] || null;
}
return null;
}
function normalizeRecoveryUser(task) {
return {
id: task.user_id,
enterpriseId: task.enterprise_id ?? null,
accountType: task.enterprise_id ? "enterprise" : "personal",
};
}
async function runStaleTaskCleanup() { async function runStaleTaskCleanup() {
if (taskStaleCleanupRunning) return;
taskStaleCleanupRunning = true;
try { try {
const { rows } = await pool.query( const { rows } = await pool.query(
`UPDATE generation_tasks `UPDATE generation_tasks
SET status = 'failed', error = '任务超时自动释放', completed_at = NOW(), updated_at = NOW() SET status = 'failed',
error = 'Task timed out and was released automatically',
completed_at = NOW(),
updated_at = NOW()
WHERE status IN ('pending', 'running') WHERE status IN ('pending', 'running')
AND GREATEST(updated_at, COALESCE(last_poll_at, created_at)) < NOW() - INTERVAL '10 minutes' AND GREATEST(updated_at, COALESCE(last_poll_at, created_at)) < NOW() - ($1::int * INTERVAL '1 minute')
RETURNING *`, RETURNING *`,
[STALE_TASK_TIMEOUT_MINUTES],
); );
for (const row of rows) { for (const row of rows) {
// Also stop any active poller for this task await stopPolling(row.id, { releaseLease: true });
const poller = activePollers.get(row.id); taskEvents.emit(`task:${row.id}`, formatTaskEvent(row));
if (poller) {
clearInterval(poller.interval);
activePollers.delete(row.id);
}
const leaseToken = poller?.leaseToken || await getPersistedLeaseToken(row.id).catch(() => null);
await clearPollerState(row.id).catch(() => {});
if (leaseToken) {
await require("./keyManager").releaseKey(leaseToken).catch((err) => {
console.error(`[aiTaskWorker] failed to release lease for stale task ${row.id}:`, err.message);
});
}
await publishTaskEvent(formatTaskEvent(row));
await createTaskLifecycleNotification(row).catch((err) => {
console.error(`[aiTaskWorker] notification error for stale task ${row.id}:`, err.message);
});
await refundTaskBillingOnFailure(row.id).catch((err) => { await refundTaskBillingOnFailure(row.id).catch((err) => {
console.error(`[aiTaskWorker] refund error for stale task ${row.id}:`, err.message); console.error(`[aiTaskWorker] stale task refund error for task ${row.id}:`, err.message);
}); });
} }
if (rows.length > 0) { if (rows.length > 0) {
console.log(`[aiTaskWorker] Cleaned up ${rows.length} stale task(s)`); console.log(`[aiTaskWorker] Cleaned up ${rows.length} stale task(s)`);
} }
} catch (err) { } catch (err) {
console.error("[aiTaskWorker] Stale task cleanup failed:", err.message); console.error("[aiTaskWorker] Stale task cleanup failed:", err.message);
}
}
async function startTaskEventListener() {
if (taskEventListenerClient) return;
if (taskEventListenerStarting) return taskEventListenerStarting;
taskEventListenerStarting = (async () => {
const client = await pool.connect();
let released = false;
const releaseClient = () => {
if (released) return;
released = true;
taskEventListenerClient = null;
try {
client.release();
} catch {}
};
client.on("notification", (message) => {
if (message.channel !== TASK_EVENT_CHANNEL || !message.payload) return;
try {
const payload = JSON.parse(message.payload);
if (payload?.origin === TASK_EVENT_ORIGIN) return;
emitTaskEvent(payload?.event || payload);
} catch (err) {
console.error("[aiTaskWorker] task event notification parse failed:", err.message);
}
});
client.on("error", (err) => {
console.error("[aiTaskWorker] task event listener error:", err.message);
releaseClient();
setTimeout(() => {
startTaskEventListener().catch((restartErr) => {
console.error("[aiTaskWorker] task event listener restart failed:", restartErr.message);
});
}, 5000).unref?.();
});
await client.query(`LISTEN ${TASK_EVENT_CHANNEL}`);
taskEventListenerClient = client;
console.log(`[aiTaskWorker] listening for task events on ${TASK_EVENT_CHANNEL}`);
})();
try {
await taskEventListenerStarting;
} finally { } finally {
taskEventListenerStarting = null; taskStaleCleanupRunning = false;
} }
} }
async function stopTaskEventListener() { async function runResultPersistenceRetry() {
const client = taskEventListenerClient; if (taskResultPersistenceRetryRunning) return;
taskEventListenerClient = null; taskResultPersistenceRetryRunning = true;
if (!client) return;
try { try {
await client.query(`UNLISTEN ${TASK_EVENT_CHANNEL}`); const { rows } = await pool.query(
} catch {} `SELECT *
client.release(); FROM generation_tasks
WHERE status = 'completed'
AND result_url IS NOT NULL
AND result_url ~* '^https?://'
AND result_url !~* '/users/[^/]+/generation-results/'
AND result_persist_status IN ('pending', 'failed')
AND result_persist_attempts < $1
ORDER BY updated_at ASC
LIMIT $2`,
[RESULT_PERSIST_RETRY_LIMIT, RESULT_PERSIST_RETRY_BATCH_SIZE],
);
for (const row of rows) {
await persistTaskResultUrlToOss(row);
}
if (rows.length > 0) {
console.log(`[aiTaskWorker] Retried OSS result persistence for ${rows.length} task(s)`);
}
} catch (err) {
console.error("[aiTaskWorker] Result persistence retry failed:", err.message);
} finally {
taskResultPersistenceRetryRunning = false;
}
}
async function runTaskStartupRecovery() {
if (taskStartupRecoveryRunning) return;
taskStartupRecoveryRunning = true;
try {
const { rows } = await pool.query(
`SELECT gt.*, u.enterprise_id
FROM generation_tasks gt
JOIN users u ON u.id = gt.user_id
WHERE gt.status = 'running'
AND gt.provider_task_id IS NOT NULL
AND GREATEST(gt.updated_at, COALESCE(gt.last_poll_at, gt.created_at)) >= NOW() - ($1::int * INTERVAL '1 minute')
ORDER BY gt.updated_at DESC
LIMIT $2`,
[STALE_TASK_TIMEOUT_MINUTES, TASK_STARTUP_RECOVERY_LIMIT],
);
let recovered = 0;
for (const task of rows) {
if (activePollers.has(task.id)) continue;
let providerConfig;
try {
providerConfig = resolveProviderConfigForRecovery(task);
} catch (err) {
console.warn(`[aiTaskWorker] task ${task.id} recovery skipped: ${err.message}`);
continue;
}
if (!providerConfig?.provider) continue;
const slotResult = await keyManager.acquireKey(providerConfig.provider, normalizeRecoveryUser(task), null, { waitTimeoutMs: 0 });
if (!slotResult) {
console.warn(`[aiTaskWorker] task ${task.id} recovery waiting for provider capacity: ${providerConfig.provider}`);
continue;
}
startPolling(task.id, {
providerTaskId: task.provider_task_id,
apiKey: slotResult.apiKey,
type: task.type,
providerConfig,
leaseToken: slotResult.leaseToken,
keyManager,
});
recovered += 1;
}
if (recovered > 0) {
console.log(`[aiTaskWorker] Recovered ${recovered} running task poller(s) after startup`);
}
} catch (err) {
console.error("[aiTaskWorker] Startup task recovery failed:", err.message);
} finally {
taskStartupRecoveryRunning = false;
}
} }
function startStaleTaskCleanup() { function startStaleTaskCleanup() {
if (staleTaskCleanupTimer) return; if (staleTaskCleanupTimer) return;
staleTaskCleanupTimer = setInterval(runStaleTaskCleanup, STALE_TASK_CLEANUP_INTERVAL_MS); staleTaskCleanupTimer = setInterval(runStaleTaskCleanup, STALE_TASK_CLEANUP_INTERVAL_MS);
// Run once shortly after startup taskResultPersistenceRetryTimer = setInterval(runResultPersistenceRetry, TASK_RESULT_PERSIST_RETRY_INTERVAL_MS);
staleTaskCleanupStartupTimer = setTimeout(runStaleTaskCleanup, 10_000); taskStartupRecoveryTimer = setTimeout(runTaskStartupRecovery, 5_000);
setTimeout(runStaleTaskCleanup, 10_000);
setTimeout(runResultPersistenceRetry, 15_000);
} }
function stopStaleTaskCleanup() { function stopStaleTaskCleanup() {
if (staleTaskCleanupStartupTimer) {
clearTimeout(staleTaskCleanupStartupTimer);
staleTaskCleanupStartupTimer = null;
}
if (staleTaskCleanupTimer) { if (staleTaskCleanupTimer) {
clearInterval(staleTaskCleanupTimer); clearInterval(staleTaskCleanupTimer);
staleTaskCleanupTimer = null; staleTaskCleanupTimer = null;
} }
if (taskResultPersistenceRetryTimer) {
clearInterval(taskResultPersistenceRetryTimer);
taskResultPersistenceRetryTimer = null;
} }
if (taskStartupRecoveryTimer) {
function startPollerRecovery() { clearTimeout(taskStartupRecoveryTimer);
if (pollerRecoveryTimer) return; taskStartupRecoveryTimer = null;
ensureTaskPollerStore()
.then(() => recoverRunnablePollers())
.catch((err) => console.error("[aiTaskWorker] initial poller recovery failed:", err.message));
pollerRecoveryTimer = setInterval(() => {
recoverRunnablePollers().catch((err) => {
console.error("[aiTaskWorker] poller recovery failed:", err.message);
});
}, POLLER_RECOVERY_INTERVAL_MS);
} }
function stopPollerRecovery() {
if (pollerRecoveryTimer) {
clearInterval(pollerRecoveryTimer);
pollerRecoveryTimer = null;
}
}
async function stopAllPollers() {
for (const [taskId, poller] of activePollers.entries()) {
clearInterval(poller.interval);
activePollers.delete(taskId);
}
await orphanOwnedPollerState().catch((err) => {
console.error("[aiTaskWorker] failed to orphan owned poller state:", err.message);
});
} }
module.exports = { module.exports = {
startPolling, startPolling,
stopPolling, stopPolling,
stopAllPollers, cancelTask,
cancelTaskRuntimeState,
updateTaskInDb, updateTaskInDb,
getActiveCount, getActiveCount,
extractProviderTaskId, extractProviderTaskId,
@@ -1227,10 +1002,6 @@ module.exports = {
parseKlingCredential, parseKlingCredential,
createKlingJwt, createKlingJwt,
taskEvents, taskEvents,
startTaskEventListener,
stopTaskEventListener,
startPollerRecovery,
stopPollerRecovery,
startStaleTaskCleanup, startStaleTaskCleanup,
stopStaleTaskCleanup, stopStaleTaskCleanup,
}; };
src/auth.js
+14 -14
View File
@@ -6,6 +6,7 @@ const { getJwtSecret } = require("./securityConfig");
const JWT_SECRET = getJwtSecret(); const JWT_SECRET = getJwtSecret();
const JWT_EXPIRES_IN = process.env.JWT_EXPIRES_IN || "7d"; const JWT_EXPIRES_IN = process.env.JWT_EXPIRES_IN || "7d";
const MAX_CONCURRENT_SESSIONS = 2;
const USER_CONTEXT_SELECT = ` const USER_CONTEXT_SELECT = `
SELECT SELECT
@@ -169,26 +170,25 @@ function verifyToken(token) {
async function startUserSession(userId, userAgent) { async function startUserSession(userId, userAgent) {
const sessionId = crypto.randomUUID(); const sessionId = crypto.randomUUID();
const client = await pool.connect(); await pool.query(
try {
await client.query("BEGIN");
await client.query("SELECT id FROM users WHERE id = $1 FOR UPDATE", [userId]);
await client.query("DELETE FROM user_sessions WHERE user_id = $1", [userId]);
await client.query(
"INSERT INTO user_sessions (id, user_id, user_agent, created_at) VALUES ($1, $2, $3, NOW())", "INSERT INTO user_sessions (id, user_id, user_agent, created_at) VALUES ($1, $2, $3, NOW())",
[sessionId, userId, userAgent || null], [sessionId, userId, userAgent || null],
); );
await client.query( await pool.query(
`DELETE FROM user_sessions
WHERE user_id = $1
AND id NOT IN (
SELECT id FROM user_sessions
WHERE user_id = $1
ORDER BY created_at DESC
LIMIT $2
)`,
[userId, MAX_CONCURRENT_SESSIONS],
);
await pool.query(
"UPDATE users SET current_session_id = $1, current_session_started_at = NOW(), updated_at = NOW() WHERE id = $2", "UPDATE users SET current_session_id = $1, current_session_started_at = NOW(), updated_at = NOW() WHERE id = $2",
[sessionId, userId], [sessionId, userId],
); );
await client.query("COMMIT");
} catch (error) {
await client.query("ROLLBACK");
throw error;
} finally {
client.release();
}
return sessionId; return sessionId;
} }
src/billing.js
+25 -40
View File
@@ -1,46 +1,29 @@
/** /**
* Billing module — handles balance deduction, package quotas, * Billing module — handles balance deduction (in cents), package quotas,
* transactions, and key-lease pre-authorization. * transactions, and key-lease pre-authorization.
* *
* Unit conventions: * Money conventions:
* - payment_orders.amount_cents / packages.price_cents: cash amount in CNY cents. * - balance: cents (分, 1/100 CNY) — stored in users.balance_cents and enterprises.balance_cents
* - users.balance_cents / enterprises.balance_cents / transactions.amount_cents: * - prices: mills (厘, 1/1000 CNY) — stored in model_prices.*_mills
* credit units, where 100 units = 1 platform credit. * - cost calculation: mills → convert to cents at deduction time (divide by 10, floor)
* - model_prices.*_mills: CNY mills. 1 CNY = 100 credits, so 1 mill = 10 credit units. * - transactions: cents — amount_cents, balance_after_cents
* *
* Flow: * Flow:
* - Enterprise admin recharges enterprise pool → distributes to employee users * - Enterprise admin recharges enterprise pool → distributes to employee users
* - API deductions come from users.balance_cents (per-user credit balance) * - API deductions come from users.balance_cents (per-user)
* - Personal users recharge their own users.balance_cents directly * - Personal users recharge their own users.balance_cents directly
*/ */
const { pool, withTransaction } = require("./db"); const { pool, withTransaction } = require("./db");
const { calculateCostMills, getModelPrice } = require("./pricing"); const { calculateCostMills, getModelPrice } = require("./pricing");
const CREDIT_UNITS_PER_CREDIT = 100; const IMAGE_GENERATION_FLAT_COST_CENTS = 20;
const CREDITS_PER_CNY = 100;
const CREDIT_UNITS_PER_CNY_CENT = 100;
const CREDIT_UNITS_PER_CNY_MILL = 10;
const IMAGE_GENERATION_FLAT_COST_CREDITS = 20;
const IMAGE_GENERATION_FLAT_COST_CENTS = IMAGE_GENERATION_FLAT_COST_CREDITS * CREDIT_UNITS_PER_CREDIT;
function creditsToCreditUnits(credits) {
return Math.max(0, Math.round(Number(credits || 0) * CREDIT_UNITS_PER_CREDIT));
}
function formatCreditsFromCents(amountCents) { function formatCreditsFromCents(amountCents) {
const value = Number(amountCents || 0) / CREDIT_UNITS_PER_CREDIT; const value = Number(amountCents || 0) / 100;
return Number.isInteger(value) ? String(value) : String(Number(value.toFixed(2))); return Number.isInteger(value) ? String(value) : String(Number(value.toFixed(2)));
} }
function cashCentsToCreditUnits(amountCents) {
return Math.max(0, Math.round(Number(amountCents || 0) * CREDIT_UNITS_PER_CNY_CENT));
}
function millsToCreditUnits(mills) {
return Math.max(0, Math.round(Number(mills || 0) * CREDIT_UNITS_PER_CNY_MILL));
}
async function recordEnterpriseCreditLedger(client, entry) { async function recordEnterpriseCreditLedger(client, entry) {
const enterpriseId = entry?.enterpriseId || null; const enterpriseId = entry?.enterpriseId || null;
const userId = entry?.userId || null; const userId = entry?.userId || null;
@@ -131,6 +114,10 @@ async function getEnterpriseName(enterpriseId) {
return rows[0] ? rows[0].name : null; return rows[0] ? rows[0].name : null;
} }
function millsToCents(mills) {
return Math.floor(mills / 10);
}
// ── Atomic balance helpers ─────────────────────────────────────────── // ── Atomic balance helpers ───────────────────────────────────────────
async function atomicDeductUserBalance(client, userId, amountCents) { async function atomicDeductUserBalance(client, userId, amountCents) {
@@ -180,7 +167,7 @@ async function preauthorizeCall(userId, provider) {
const { rows } = await pool.query( const { rows } = await pool.query(
` `
SELECT COALESCE(CAST(ROUND(AVG(cost_estimate * 10000)::numeric) AS INTEGER), 0) AS avg_cents SELECT COALESCE(CAST(ROUND(AVG(cost_estimate * 100)::numeric) AS INTEGER), 0) AS avg_cents
FROM api_call_logs FROM api_call_logs
WHERE provider = $1 WHERE provider = $1
AND status = 'success' AND status = 'success'
@@ -198,9 +185,10 @@ async function preauthorizeCall(userId, provider) {
const bufferedEstimate = Math.ceil(estimatedCostCents * 1.2); const bufferedEstimate = Math.ceil(estimatedCostCents * 1.2);
if (balanceCents < bufferedEstimate) { if (balanceCents < bufferedEstimate) {
const credits = Math.floor(balanceCents / 100);
return { return {
authorized: false, authorized: false,
message: `账户积分不足，请充值 (当前 ${formatCreditsFromCents(balanceCents)} 积分，预估需要 ${formatCreditsFromCents(bufferedEstimate)} 积分)`, message: `账户积分不足，请充值 (当前 ${credits} 积分，预估需要 ${Math.ceil(bufferedEstimate / 100)} 积分)`,
}; };
} }
@@ -217,9 +205,9 @@ async function deductForApiCall(userId, model, promptTokens, completionTokens) {
return { success: true, costCents: 0, deductionType: "none", message: "No pricing" }; return { success: true, costCents: 0, deductionType: "none", message: "No pricing" };
} }
const costCents = millsToCreditUnits(costMills); const costCents = millsToCents(costMills);
if (costCents <= 0) { if (costCents <= 0) {
return { success: true, costCents: 0, deductionType: "none", message: "Cost below minimum credit unit" }; return { success: true, costCents: 0, deductionType: "none", message: "Cost below 1 cent" };
} }
const billingState = await getUserBillingState(userId); const billingState = await getUserBillingState(userId);
@@ -420,7 +408,7 @@ async function tryDeductFromUserBalance(userId, enterpriseId, amountCents, ledge
userId, userId,
-amountCents, -amountCents,
newBal, newBal,
`API 调用扣费 ${formatCreditsFromCents(amountCents)} 积分`, `API 调用扣费 ${Math.ceil(amountCents / 100)} 积分`,
], ],
); );
@@ -441,15 +429,16 @@ async function tryDeductFromUserBalance(userId, enterpriseId, amountCents, ledge
if (newBalanceCents == null) { if (newBalanceCents == null) {
const currentBalance = await getUserBalanceCents(userId); const currentBalance = await getUserBalanceCents(userId);
const credits = Math.floor((currentBalance || 0) / 100);
return { return {
success: false, success: false,
message: `积分不足 (当前 ${formatCreditsFromCents(currentBalance || 0)} 积分，需要 ${formatCreditsFromCents(amountCents)} 积分)`, message: `积分不足 (当前 ${credits} 积分，需要 ${Math.ceil(amountCents / 100)} 积分)`,
}; };
} }
return { return {
success: true, success: true,
message: `Deducted ${formatCreditsFromCents(amountCents)} credits, balance: ${formatCreditsFromCents(newBalanceCents)} credits`, message: `Deducted ${Math.ceil(amountCents / 100)} credits, balance: ${Math.floor(newBalanceCents / 100)} credits`,
}; };
} }
@@ -495,7 +484,7 @@ async function settleLease(leaseId, actualCostCents) {
userId, userId,
-diffCents, -diffCents,
newBal, newBal,
`API 预估差额扣费 ${formatCreditsFromCents(diffCents)} 积分`, `API 预估差额扣费 ${Math.ceil(diffCents / 100)} 积分`,
], ],
); );
} }
@@ -514,7 +503,7 @@ async function settleLease(leaseId, actualCostCents) {
userId, userId,
refundCents, refundCents,
newBal, newBal,
`API 预估差额退回 ${formatCreditsFromCents(refundCents)} 积分`, `API 预估差额退回 ${Math.ceil(refundCents / 100)} 积分`,
], ],
); );
} }
@@ -639,7 +628,7 @@ async function distributeCredits(enterpriseId, targetUserId, amountCents, adminU
targetUserId, targetUserId,
amountCents, amountCents,
newUserBal, newUserBal,
`从企业池获得 ${formatCreditsFromCents(amountCents)} 积分`, `从企业池获得 ${Math.floor(amountCents / 100)} 积分`,
adminUserId, adminUserId,
], ],
); );
@@ -772,8 +761,4 @@ module.exports = {
preauthorizeCall, preauthorizeCall,
settleLease, settleLease,
forceSettleLease, forceSettleLease,
creditsToCreditUnits,
cashCentsToCreditUnits,
millsToCreditUnits,
formatCreditsFromCents,
}; };
src/dbSetup.js
+13
View File
@@ -353,6 +353,18 @@ async function migrateGenerationTasksBillingColumns(client) {
); );
} }
async function migrateGenerationTaskResultPersistence(client) {
await addColumnIfMissing("generation_tasks", "result_persist_status TEXT NOT NULL DEFAULT 'pending'");
await addColumnIfMissing("generation_tasks", "result_persist_attempts INTEGER NOT NULL DEFAULT 0");
await addColumnIfMissing("generation_tasks", "result_persist_error TEXT");
await addColumnIfMissing("generation_tasks", "result_persisted_at TIMESTAMPTZ");
await client.query(`
CREATE INDEX IF NOT EXISTS idx_generation_tasks_result_persist_retry
ON generation_tasks(result_persist_status, updated_at)
WHERE status = 'completed' AND result_url IS NOT NULL
`);
}
async function ensureModelPriceSeed() { async function ensureModelPriceSeed() {
const columns = await getColumnNames("model_prices"); const columns = await getColumnNames("model_prices");
const useMills = columns.includes("input_price_mills"); const useMills = columns.includes("input_price_mills");
@@ -959,6 +971,7 @@ async function ensureSchema() {
await runMigration("030_generation_tasks_user_status_index", migrateGenerationTasksUserStatusIndex); await runMigration("030_generation_tasks_user_status_index", migrateGenerationTasksUserStatusIndex);
await runMigration("031_generation_tasks_billing_columns", migrateGenerationTasksBillingColumns); await runMigration("031_generation_tasks_billing_columns", migrateGenerationTasksBillingColumns);
await runMigration("032_ecommerce_video_history", migrateEcommerceVideoHistorySchema); await runMigration("032_ecommerce_video_history", migrateEcommerceVideoHistorySchema);
await runMigration("033_generation_task_result_persistence", migrateGenerationTaskResultPersistence);
await ensureModelPriceSeed(); await ensureModelPriceSeed();
} }
src/enterpriseVideoBilling.js
+29 -88
View File
@@ -19,56 +19,6 @@ const ENTERPRISE_VIDEO_ALLOWED_MODELS = new Set([
"pixverse-c1-i2v", "pixverse-c1-i2v",
]); ]);
const CREDITS_PER_CNY = 100;
const CREDIT_UNITS_PER_CREDIT = 100;
const ENTERPRISE_VIDEO_RESOLUTIONS = ["720P", "1080P"];
const ENTERPRISE_VIDEO_DEFAULT_RESOLUTION = "1080P";
const ENTERPRISE_VIDEO_PRICING_RULES = [
{
id: "happyhorse",
modelIncludes: ["happyhorse"],
rates: { "720P": 0.72, "1080P": 1.28 },
},
{
id: "wanxiang-i2v",
modelIncludes: ["wan2.7-i2v", "wanxiang"],
rates: { "720P": 0.6, "1080P": 1 },
},
{
id: "wan-animate-s2v",
modelIncludes: ["animate-mix", "s2v"],
rates: { "720P": 0.6, "1080P": 1 },
},
{
id: "kling-muted-reference",
modelIncludes: ["kling"],
when: { muted: true, hasReferenceVideo: true },
rates: { "720P": 0.9, "1080P": 1.2 },
},
{
id: "kling-muted",
modelIncludes: ["kling"],
when: { muted: true, hasReferenceVideo: false },
rates: { "720P": 0.6, "1080P": 0.8 },
},
{
id: "kling-default",
modelIncludes: ["kling"],
rates: { "720P": 0.9, "1080P": 1.2 },
},
{
id: "vidu",
modelIncludes: ["vidu"],
rates: { "720P": 0.6, "1080P": 1 },
},
{
id: "pixverse",
modelIncludes: ["pixverse"],
rates: { "720P": 0.6, "1080P": 1 },
},
];
function normalizeModel(value) { function normalizeModel(value) {
return String(value || "").trim().toLowerCase(); return String(value || "").trim().toLowerCase();
} }
@@ -83,21 +33,6 @@ function normalizeEnterpriseVideoDuration(value) {
return Math.max(1, Math.ceil(numeric)); return Math.max(1, Math.ceil(numeric));
} }
function enterpriseVideoPricingRuleMatches(rule, input, model) {
if (!rule.modelIncludes.some((pattern) => model.includes(pattern))) return false;
if (!rule.when) return true;
if (Object.prototype.hasOwnProperty.call(rule.when, "muted") && Boolean(input.muted) !== rule.when.muted) {
return false;
}
if (
Object.prototype.hasOwnProperty.call(rule.when, "hasReferenceVideo") &&
Boolean(input.hasReferenceVideo) !== rule.when.hasReferenceVideo
) {
return false;
}
return true;
}
function isEnterpriseVideoBillingUser(user) { function isEnterpriseVideoBillingUser(user) {
return Boolean(user?.enterpriseId); return Boolean(user?.enterpriseId);
} }
@@ -131,10 +66,33 @@ function getEnterpriseVideoCreditRate(input) {
const resolution = normalizeEnterpriseVideoResolution(input.resolution || input.quality); const resolution = normalizeEnterpriseVideoResolution(input.resolution || input.quality);
const model = normalizeModel(input.model || input.requestedModel); const model = normalizeModel(input.model || input.requestedModel);
const rule = ENTERPRISE_VIDEO_PRICING_RULES.find((candidate) => if (model.includes("happyhorse")) {
enterpriseVideoPricingRuleMatches(candidate, input, model), return resolution === "720P" ? 0.72 : 1.28;
); }
if (rule) return rule.rates[resolution] ?? rule.rates[ENTERPRISE_VIDEO_DEFAULT_RESOLUTION];
if (model.includes("wan2.7-i2v") || model.includes("wanxiang")) {
return resolution === "720P" ? 0.6 : 1;
}
if (model.includes("animate-mix") || model.includes("s2v")) {
return resolution === "720P" ? 0.6 : 1;
}
if (model.includes("kling")) {
if (input.muted) {
if (input.hasReferenceVideo) return resolution === "720P" ? 0.9 : 1.2;
return resolution === "720P" ? 0.6 : 0.8;
}
return resolution === "720P" ? 0.9 : 1.2;
}
if (model.includes("vidu")) {
return resolution === "720P" ? 0.6 : 1.0;
}
if (model.includes("pixverse")) {
return resolution === "720P" ? 0.6 : 1.0;
}
const error = new Error(`Unsupported enterprise video model: ${input.model || input.requestedModel}`); const error = new Error(`Unsupported enterprise video model: ${input.model || input.requestedModel}`);
error.status = 403; error.status = 403;
@@ -142,25 +100,9 @@ function getEnterpriseVideoCreditRate(input) {
throw error; throw error;
} }
function getEnterpriseVideoPricingConfig() {
return {
currency: "CNY",
creditsPerCny: CREDITS_PER_CNY,
billingUnit: "per_second",
defaultResolution: ENTERPRISE_VIDEO_DEFAULT_RESOLUTION,
resolutions: [...ENTERPRISE_VIDEO_RESOLUTIONS],
rules: ENTERPRISE_VIDEO_PRICING_RULES.map((rule) => ({
id: rule.id,
modelIncludes: [...rule.modelIncludes],
when: rule.when ? { ...rule.when } : undefined,
rates: { ...rule.rates },
})),
};
}
function calculateEnterpriseVideoCredits(input) { function calculateEnterpriseVideoCredits(input) {
const duration = normalizeEnterpriseVideoDuration(input.durationSeconds || input.duration); const duration = normalizeEnterpriseVideoDuration(input.durationSeconds || input.duration);
return Number((getEnterpriseVideoCreditRate(input) * duration * CREDITS_PER_CNY).toFixed(2)); return Number((getEnterpriseVideoCreditRate(input) * duration).toFixed(2));
} }
function calculateEnterpriseVideoCost(input) { function calculateEnterpriseVideoCost(input) {
@@ -171,7 +113,7 @@ function calculateEnterpriseVideoCost(input) {
resolution, resolution,
durationSeconds, durationSeconds,
}); });
const rateCentsPerSecond = Math.round(rateCreditsPerSecond * CREDITS_PER_CNY * CREDIT_UNITS_PER_CREDIT); const rateCentsPerSecond = Math.round(rateCreditsPerSecond * 100);
return { return {
resolution, resolution,
durationSeconds, durationSeconds,
@@ -288,7 +230,6 @@ module.exports = {
assertEnterpriseVideoModelAllowed, assertEnterpriseVideoModelAllowed,
calculateEnterpriseVideoCost, calculateEnterpriseVideoCost,
calculateEnterpriseVideoCredits, calculateEnterpriseVideoCredits,
getEnterpriseVideoPricingConfig,
getEnterpriseVideoCreditRate, getEnterpriseVideoCreditRate,
isEnterpriseVideoBillingUser, isEnterpriseVideoBillingUser,
isEnterpriseVideoModelAllowed, isEnterpriseVideoModelAllowed,
src/index.js
+19 -49
View File
@@ -3,17 +3,8 @@ const express = require('express')
const rateLimit = require('express-rate-limit') const rateLimit = require('express-rate-limit')
const cors = require('cors') const cors = require('cors')
const helmet = require('helmet') const helmet = require('helmet')
const { startSettlementWorker, stopSettlementWorker } = require('./settlementWorker') const { startSettlementWorker } = require('./settlementWorker')
const { startProviderHealthMonitor, stopProviderHealthMonitor } = require('./providerHealthMonitor') const { startProviderHealthMonitor } = require('./providerHealthMonitor')
const {
startStaleTaskCleanup,
startTaskEventListener,
startPollerRecovery,
stopStaleTaskCleanup,
stopTaskEventListener,
stopPollerRecovery,
stopAllPollers,
} = require('./aiTaskWorker')
const { ensureDatabase } = require('./dbSetup') const { ensureDatabase } = require('./dbSetup')
const { assertRuntimeSecurityConfig } = require('./securityConfig') const { assertRuntimeSecurityConfig } = require('./securityConfig')
const { loadPriceCache } = require('./pricing') const { loadPriceCache } = require('./pricing')
@@ -26,7 +17,6 @@ const PORT = Number(process.env.PORT) || 3600
const HOST = process.env.HOST || '0.0.0.0' const HOST = process.env.HOST || '0.0.0.0'
const IS_PRODUCTION = process.env.NODE_ENV === 'production' const IS_PRODUCTION = process.env.NODE_ENV === 'production'
let server = null let server = null
let staleLeaseCleanupTimer = null
// CORS: in production, require explicit allowlist; in dev, allow all with credentials // CORS: in production, require explicit allowlist; in dev, allow all with credentials
function buildCorsOptions() { function buildCorsOptions() {
@@ -143,20 +133,18 @@ async function main() {
// Periodic stale lease cleanup (every 5 min) // Periodic stale lease cleanup (every 5 min)
const { cleanStaleLeases } = require('./keyManager') const { cleanStaleLeases } = require('./keyManager')
staleLeaseCleanupTimer = setInterval(() => { setInterval(() => {
cleanStaleLeases().then((cleaned) => { cleanStaleLeases().then((cleaned) => {
if (cleaned > 0) console.log(`[cleanup] Released ${cleaned} stale lease(s)`) if (cleaned > 0) console.log(`[cleanup] Released ${cleaned} stale lease(s)`)
}).catch((err) => { }).catch((err) => {
console.error('[cleanup] error:', err) console.error('[cleanup] error:', err)
}) })
}, 5 * 60 * 1000) }, 5 * 60 * 1000)
if (staleLeaseCleanupTimer.unref) staleLeaseCleanupTimer.unref()
startSettlementWorker() startSettlementWorker()
startProviderHealthMonitor() startProviderHealthMonitor()
await startTaskEventListener() const { startStaleTaskCleanup } = require('./aiTaskWorker')
startPollerRecovery()
startStaleTaskCleanup() startStaleTaskCleanup()
server = app.listen(PORT, HOST, () => { server = app.listen(PORT, HOST, () => {
@@ -185,47 +173,29 @@ process.on('uncaughtException', (err) => {
// ── Graceful shutdown ─────────────────────────────────────────────────── // ── Graceful shutdown ───────────────────────────────────────────────────
let shuttingDown = false let shuttingDown = false
async function shutdownRuntimeState() { function gracefulShutdown(signal) {
if (staleLeaseCleanupTimer) {
clearInterval(staleLeaseCleanupTimer)
staleLeaseCleanupTimer = null
}
stopSettlementWorker()
stopProviderHealthMonitor()
stopPollerRecovery()
stopStaleTaskCleanup()
await Promise.allSettled([stopTaskEventListener(), stopAllPollers()])
}
function closeServer() {
if (!server || !server.listening) return Promise.resolve()
return new Promise((resolve) => {
server.close(() => {
console.log('[shutdown] Server closed, cleaning up...')
resolve()
})
})
}
async function gracefulShutdown(signal) {
if (shuttingDown) return if (shuttingDown) return
shuttingDown = true shuttingDown = true
console.log('[shutdown] Received ' + signal + ', draining connections...') console.log('[shutdown] Received ' + signal + ', draining connections...')
if (server && server.listening) {
server.close(() => {
console.log('[shutdown] Server closed, cleaning up...')
const { stopProviderHealthMonitor } = require('./providerHealthMonitor')
stopProviderHealthMonitor()
const { pool } = require('./db')
pool.end().then(() => {
console.log('[shutdown] Database pool closed')
process.exit(0)
}).catch(() => process.exit(0))
})
// Force exit after timeout
setTimeout(() => { setTimeout(() => {
console.error('[shutdown] Forced exit after timeout') console.error('[shutdown] Forced exit after timeout')
process.exit(1) process.exit(1)
}, 15000).unref() }, 15000).unref()
} else {
try {
await shutdownRuntimeState()
await closeServer()
const { pool } = require('./db')
await pool.end()
console.log('[shutdown] Database pool closed')
process.exit(0)
} catch (err) {
console.error('[shutdown] error:', err)
process.exit(0) process.exit(0)
} }
} }
src/keyManager.js
+3 -8
View File
@@ -297,7 +297,8 @@ async function releaseLeaseInternal(leaseToken, user, options = {}) {
WHERE id = (SELECT id FROM candidate) WHERE id = (SELECT id FROM candidate)
RETURNING id, key_id RETURNING id, key_id
) )
SELECT r.id, r.key_id, c.user_id AS lease_user_id, c.enterprise_id AS lease_enterprise_id, c.provider SELECT r.id, r.key_id, c.provider
, c.user_id, c.enterprise_id
FROM released r FROM released r
JOIN candidate c ON c.key_id = r.key_id JOIN candidate c ON c.key_id = r.key_id
`, `,
@@ -339,13 +340,7 @@ async function releaseLeaseInternal(leaseToken, user, options = {}) {
INSERT INTO usage_logs (user_id, enterprise_id, provider, key_id, action) INSERT INTO usage_logs (user_id, enterprise_id, provider, key_id, action)
VALUES ($1, $2, (SELECT provider FROM api_keys WHERE id = $3), $4, $5) VALUES ($1, $2, (SELECT provider FROM api_keys WHERE id = $3), $4, $5)
`, `,
[ [userId ?? lease.user_id, enterpriseId ?? lease.enterprise_id, lease.key_id, lease.key_id, "release"],
userId || lease.lease_user_id,
enterpriseId || lease.lease_enterprise_id,
lease.key_id,
lease.key_id,
"release",
],
); );
return { return {
src/paymentAlipay.js
+5 -7
View File
@@ -1,7 +1,7 @@
const fs = require("node:fs"); const fs = require("node:fs");
const { AlipaySdk } = require("alipay-sdk"); const { AlipaySdk } = require("alipay-sdk");
const { pool, withTransaction } = require("./db"); const { pool, withTransaction } = require("./db");
const { cashCentsToCreditUnits, creditBalance, creditUserBalance, activatePackage, formatCreditsFromCents } = require("./billing"); const { creditBalance, creditUserBalance, activatePackage } = require("./billing");
let alipayInstance = null; let alipayInstance = null;
@@ -130,19 +130,17 @@ async function handlePaymentSuccess(orderNo, tradeNo) {
); );
if (order.type === "personal_recharge" && order.user_id) { if (order.type === "personal_recharge" && order.user_id) {
const creditUnits = cashCentsToCreditUnits(order.amount_cents);
await creditUserBalance( await creditUserBalance(
order.user_id, order.user_id,
creditUnits, order.amount_cents,
`支付宝充值 ${formatCreditsFromCents(creditUnits)} 积分`, `支付宝充值 ${Math.floor(order.amount_cents / 100)} 积分`,
orderNo, orderNo,
); );
} else if (order.type === "recharge") { } else if (order.type === "recharge") {
const creditUnits = cashCentsToCreditUnits(order.amount_cents);
await creditBalance( await creditBalance(
order.enterprise_id, order.enterprise_id,
creditUnits, order.amount_cents,
`支付宝充值 ${formatCreditsFromCents(creditUnits)} 积分`, `支付宝充值 ${Math.floor(order.amount_cents / 100)} 积分`,
orderNo, orderNo,
); );
} else if (order.type === "package" && order.package_id) { } else if (order.type === "package" && order.package_id) {
src/paymentWechat.js
+5 -7
View File
@@ -2,7 +2,7 @@ const _crypto = require("node:crypto");
const fs = require("node:fs"); const fs = require("node:fs");
const WxPay = require("wechatpay-node-v3"); const WxPay = require("wechatpay-node-v3");
const { pool, withTransaction } = require("./db"); const { pool, withTransaction } = require("./db");
const { cashCentsToCreditUnits, creditBalance, creditUserBalance, activatePackage, formatCreditsFromCents } = require("./billing"); const { creditBalance, creditUserBalance, activatePackage } = require("./billing");
let wxPayInstance = null; let wxPayInstance = null;
@@ -140,19 +140,17 @@ async function handlePaymentSuccess(orderNo, transactionId) {
); );
if (order.type === "personal_recharge" && order.user_id) { if (order.type === "personal_recharge" && order.user_id) {
const creditUnits = cashCentsToCreditUnits(order.amount_cents);
await creditUserBalance( await creditUserBalance(
order.user_id, order.user_id,
creditUnits, order.amount_cents,
`微信充值 ${formatCreditsFromCents(creditUnits)} 积分`, `微信充值 ${Math.floor(order.amount_cents / 100)} 积分`,
orderNo, orderNo,
); );
} else if (order.type === "recharge") { } else if (order.type === "recharge") {
const creditUnits = cashCentsToCreditUnits(order.amount_cents);
await creditBalance( await creditBalance(
order.enterprise_id, order.enterprise_id,
creditUnits, order.amount_cents,
`微信充值 ${formatCreditsFromCents(creditUnits)} 积分`, `微信充值 ${Math.floor(order.amount_cents / 100)} 积分`,
orderNo, orderNo,
); );
} else if (order.type === "package" && order.package_id) { } else if (order.type === "package" && order.package_id) {
src/pricing.js
+1 -1
View File
@@ -176,7 +176,7 @@ async function getAverageCostCents(provider) {
const { rows } = await pool.query( const { rows } = await pool.query(
` `
SELECT CAST(ROUND(AVG(CASE SELECT CAST(ROUND(AVG(CASE
WHEN cost_estimate IS NOT NULL THEN cost_estimate * 10000 WHEN cost_estimate IS NOT NULL THEN cost_estimate * 100
ELSE 0 ELSE 0
END)::numeric) AS INTEGER) AS avg_cents END)::numeric) AS INTEGER) AS avg_cents
FROM api_call_logs FROM api_call_logs
src/providerHealthMonitor.js
+4 -17
View File
@@ -7,7 +7,6 @@
*/ */
const { pool } = require("./db"); const { pool } = require("./db");
const { recordProviderSuccess, recordProviderFailure, getAllBreakerStats } = require("./providerCircuitBreaker");
const CHECK_INTERVAL_MS = 5 * 60 * 1000; const CHECK_INTERVAL_MS = 5 * 60 * 1000;
const DASHSCOPE_TEST_MODEL = "qwen-max"; const DASHSCOPE_TEST_MODEL = "qwen-max";
@@ -22,15 +21,6 @@ const providerHealthCache = {
grsai: { status: "unknown", lastCheck: null, lastError: null, details: null }, grsai: { status: "unknown", lastCheck: null, lastError: null, details: null },
}; };
function recordProbeOutcome(provider, result, latencyMs) {
if (!provider) return;
if (result?.ok) {
recordProviderSuccess(provider, latencyMs);
} else {
recordProviderFailure(provider);
}
}
async function getDashScopeKey() { async function getDashScopeKey() {
const { rows } = await pool.query( const { rows } = await pool.query(
"SELECT id, api_key FROM api_keys WHERE provider LIKE '%dashscope%' AND enabled = 1 ORDER BY id LIMIT 1" "SELECT id, api_key FROM api_keys WHERE provider LIKE '%dashscope%' AND enabled = 1 ORDER BY id LIMIT 1"
@@ -130,10 +120,8 @@ async function runHealthCheck() {
// ── DashScope ── // ── DashScope ──
const dashKey = await getDashScopeKey(); const dashKey = await getDashScopeKey();
if (dashKey) { if (dashKey) {
const startedAt = Date.now();
try { try {
const result = await probeDashScope(dashKey); const result = await probeDashScope(dashKey);
recordProbeOutcome("dashscope", result, Date.now() - startedAt);
const prev = providerHealthCache.dashscope.status; const prev = providerHealthCache.dashscope.status;
providerHealthCache.dashscope = { providerHealthCache.dashscope = {
status: result.status, status: result.status,
@@ -156,7 +144,6 @@ async function runHealthCheck() {
} }
} }
} catch (err) { } catch (err) {
recordProviderFailure("dashscope");
providerHealthCache.dashscope = { providerHealthCache.dashscope = {
status: "timeout", status: "timeout",
lastCheck: new Date().toISOString(), lastCheck: new Date().toISOString(),
@@ -177,10 +164,8 @@ async function runHealthCheck() {
// ── GrsAI ── // ── GrsAI ──
const grsaiKey = await getGrsaiKey(); const grsaiKey = await getGrsaiKey();
if (grsaiKey) { if (grsaiKey) {
const startedAt = Date.now();
try { try {
const result = await probeGrsai(grsaiKey); const result = await probeGrsai(grsaiKey);
recordProbeOutcome("grsai", result, Date.now() - startedAt);
const prev = providerHealthCache.grsai.status; const prev = providerHealthCache.grsai.status;
providerHealthCache.grsai = { providerHealthCache.grsai = {
status: result.status, status: result.status,
@@ -201,7 +186,6 @@ async function runHealthCheck() {
} }
} }
} catch (err) { } catch (err) {
recordProviderFailure("grsai");
providerHealthCache.grsai = { providerHealthCache.grsai = {
status: "timeout", status: "timeout",
lastCheck: new Date().toISOString(), lastCheck: new Date().toISOString(),
@@ -220,7 +204,10 @@ async function runHealthCheck() {
} }
// ── Circuit breaker summary ── // ── Circuit breaker summary ──
providerHealthCache.circuitBreaker = getAllBreakerStats(); try {
const cb = require("./providerCircuitBreaker");
providerHealthCache.circuitBreaker = cb.getProviderStatusMap ? cb.getProviderStatusMap() : null;
} catch {}
// ── Admin low-balance alert ── // ── Admin low-balance alert ──
try { try {
src/providerPollLimiter.js
-120
View File
@@ -1,120 +0,0 @@
"use strict";
const crypto = require("node:crypto");
const { pool } = require("./db");
const DEFAULT_MAX_CONCURRENCY = 8;
const DEFAULT_SLOT_TTL_MS = 30_000;
const POLL_SCOPE = "generation-provider-poll:global";
const OWNER_ID = `${process.pid}-${crypto.randomUUID()}`;
let storeReady = null;
function normalizePositiveInteger(value, fallback) {
const numeric = Number(value);
if (!Number.isFinite(numeric) || numeric <= 0) return fallback;
return Math.max(1, Math.trunc(numeric));
}
function getMaxConcurrency() {
return normalizePositiveInteger(process.env.TASK_PROVIDER_POLL_MAX_CONCURRENCY, DEFAULT_MAX_CONCURRENCY);
}
function getSlotTtlInterval() {
const ttlMs = normalizePositiveInteger(process.env.TASK_PROVIDER_POLL_SLOT_TTL_MS, DEFAULT_SLOT_TTL_MS);
return `${Math.max(1, Math.ceil(ttlMs / 1000))} seconds`;
}
async function ensureProviderPollLimiterStore() {
if (storeReady) return storeReady;
storeReady = pool.query(`
CREATE TABLE IF NOT EXISTS generation_provider_poll_slots (
scope TEXT NOT NULL,
slot_no INTEGER NOT NULL,
owner_id TEXT NOT NULL,
task_id INTEGER,
expires_at TIMESTAMPTZ NOT NULL,
acquired_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (scope, slot_no)
);
CREATE INDEX IF NOT EXISTS idx_generation_provider_poll_slots_expires
ON generation_provider_poll_slots(expires_at);
`).catch((err) => {
storeReady = null;
throw err;
});
return storeReady;
}
async function acquireProviderPollSlot(taskId, options = {}) {
await ensureProviderPollLimiterStore();
const scope = options.scope || POLL_SCOPE;
const maxConcurrency = normalizePositiveInteger(options.maxConcurrency, getMaxConcurrency());
const ttlInterval = options.ttlInterval || getSlotTtlInterval();
const { rows } = await pool.query(
`
WITH candidate AS (
SELECT s.slot_no
FROM generate_series(1, $2::integer) AS s(slot_no)
LEFT JOIN generation_provider_poll_slots l
ON l.scope = $1 AND l.slot_no = s.slot_no
WHERE l.scope IS NULL OR l.expires_at < NOW()
ORDER BY s.slot_no ASC
LIMIT 1
),
claimed AS (
INSERT INTO generation_provider_poll_slots (
scope, slot_no, owner_id, task_id, expires_at, acquired_at, updated_at
)
SELECT $1, slot_no, $3, $4, NOW() + ($5::text)::interval, NOW(), NOW()
FROM candidate
ON CONFLICT (scope, slot_no) DO UPDATE SET
owner_id = EXCLUDED.owner_id,
task_id = EXCLUDED.task_id,
expires_at = EXCLUDED.expires_at,
acquired_at = NOW(),
updated_at = NOW()
WHERE generation_provider_poll_slots.expires_at < NOW()
RETURNING scope, slot_no
)
SELECT scope, slot_no FROM claimed
`,
[scope, maxConcurrency, OWNER_ID, taskId || null, ttlInterval],
);
const slot = rows[0];
return slot ? { scope: slot.scope, slotNo: slot.slot_no, ownerId: OWNER_ID } : null;
}
async function releaseProviderPollSlot(slot) {
if (!slot?.scope || !slot?.slotNo) return;
await ensureProviderPollLimiterStore();
await pool.query(
"DELETE FROM generation_provider_poll_slots WHERE scope = $1 AND slot_no = $2 AND owner_id = $3",
[slot.scope, slot.slotNo, slot.ownerId || OWNER_ID],
);
}
async function withProviderPollSlot(taskId, fn, options = {}) {
const slot = await acquireProviderPollSlot(taskId, options);
if (!slot) return { acquired: false, value: undefined };
try {
return { acquired: true, value: await fn() };
} finally {
await releaseProviderPollSlot(slot).catch((err) => {
console.error(`[providerPollLimiter] failed to release poll slot ${slot.scope}:${slot.slotNo}:`, err.message);
});
}
}
module.exports = {
acquireProviderPollSlot,
ensureProviderPollLimiterStore,
getMaxConcurrency,
normalizePositiveInteger,
releaseProviderPollSlot,
withProviderPollSlot,
};
src/routes/admin.js
+6 -21
View File
@@ -6,8 +6,6 @@ const {
listModelPrices, listModelPrices,
loadPriceCache, loadPriceCache,
creditUserBalance, creditUserBalance,
creditsToCreditUnits,
formatCreditsFromCents,
pool, pool,
validateUsername, validateUsername,
validatePassword, validatePassword,
@@ -158,18 +156,14 @@ function registerAdminRoutes(router) {
router.post("/admin/users/:id/credit", requireAuth, requireAdmin, async (req, res) => { router.post("/admin/users/:id/credit", requireAuth, requireAdmin, async (req, res) => {
const targetUserId = Number(req.params.id); const targetUserId = Number(req.params.id);
const { amountCredits, amountCents } = req.body; const { amountCents } = req.body;
const creditUnits = if (!amountCents || amountCents <= 0) return res.status(400).json({ error: "积分必须大于 0" });
amountCredits !== undefined && amountCredits !== null && amountCredits !== ""
? creditsToCreditUnits(amountCredits)
: Number(amountCents);
if (!creditUnits || creditUnits <= 0) return res.status(400).json({ error: "积分必须大于 0" });
try { try {
const newBalance = await creditUserBalance( const newBalance = await creditUserBalance(
targetUserId, targetUserId,
creditUnits, amountCents,
`管理员 ${req.user.username} 发放 ${formatCreditsFromCents(creditUnits)} 积分`, `管理员 ${req.user.username} 发放 ${Math.floor(amountCents / 100)} 积分`,
); );
res.json({ success: true, newBalanceCents: newBalance }); res.json({ success: true, newBalanceCents: newBalance });
} catch (err) { } catch (err) {
@@ -553,8 +547,6 @@ function registerAdminRoutes(router) {
name, name,
description = "", description = "",
priceCents, priceCents,
credits,
amountCredits,
creditsCents = 0, creditsCents = 0,
imageQuota = 0, imageQuota = 0,
videoQuota = 0, videoQuota = 0,
@@ -580,9 +572,7 @@ function registerAdminRoutes(router) {
name, name,
description, description,
Number(priceCents), Number(priceCents),
credits !== undefined || amountCredits !== undefined Number(creditsCents || 0),
? creditsToCreditUnits(credits ?? amountCredits)
: Number(creditsCents || 0),
Number(imageQuota || 0), Number(imageQuota || 0),
Number(videoQuota || 0), Number(videoQuota || 0),
Number(textQuota || 0), Number(textQuota || 0),
@@ -609,8 +599,6 @@ function registerAdminRoutes(router) {
name, name,
description, description,
priceCents, priceCents,
credits,
amountCredits,
creditsCents, creditsCents,
imageQuota, imageQuota,
videoQuota, videoQuota,
@@ -635,10 +623,7 @@ function registerAdminRoutes(router) {
updates.push(`price_cents = $${idx++}`); updates.push(`price_cents = $${idx++}`);
params.push(Number(priceCents)); params.push(Number(priceCents));
} }
if (credits !== undefined || amountCredits !== undefined) { if (creditsCents !== undefined) {
updates.push(`credits_cents = $${idx++}`);
params.push(creditsToCreditUnits(credits ?? amountCredits));
} else if (creditsCents !== undefined) {
updates.push(`credits_cents = $${idx++}`); updates.push(`credits_cents = $${idx++}`);
params.push(Number(creditsCents)); params.push(Number(creditsCents));
} }
src/routes/ai.js
+20 -130
View File
@@ -4,7 +4,7 @@ const crypto = require("node:crypto");
const { requireAuth, keyManager, preauthorizeCall, pool, withTransaction, deductImageGenerationCredits } = require("./context"); const { requireAuth, keyManager, preauthorizeCall, pool, withTransaction, deductImageGenerationCredits } = require("./context");
const { putObject, isOssConfigured } = require("../ossClient"); const { putObject, isOssConfigured } = require("../ossClient");
const { buildImageProviderDebug, resolveImageProviderCandidates, resolveVideoProvider, resolveTextProvider, getPostUrl } = require("../aiProviderRouter"); const { buildImageProviderDebug, resolveImageProviderCandidates, resolveVideoProvider, resolveTextProvider, getPostUrl } = require("../aiProviderRouter");
const { shouldSkipProvider, recordProviderSuccess, recordProviderFailure, getAdaptiveTimeout } = require("../providerCircuitBreaker"); const { shouldSkipProvider, recordProviderSuccess, recordProviderFailure } = require("../providerCircuitBreaker");
const { const {
isEnterpriseVideoBillingUser, isEnterpriseVideoBillingUser,
markEnterpriseVideoCreditsAccepted, markEnterpriseVideoCreditsAccepted,
@@ -16,7 +16,7 @@ const {
} = require("../enterpriseVideoBilling"); } = require("../enterpriseVideoBilling");
const { const {
startPolling, startPolling,
cancelTaskRuntimeState, cancelTask,
updateTaskInDb, updateTaskInDb,
extractProviderTaskId, extractProviderTaskId,
extractImageUrl, extractImageUrl,
@@ -60,7 +60,6 @@ function toViapiAccessibleUrl(url) {
const SUPER_RESOLVE_POLL_INTERVAL_MS = 3000; const SUPER_RESOLVE_POLL_INTERVAL_MS = 3000;
const SUPER_RESOLVE_MAX_POLL_ATTEMPTS = 200; const SUPER_RESOLVE_MAX_POLL_ATTEMPTS = 200;
const IMAGE_PROVIDER_SUBMIT_TIMEOUT_MS = 90_000; const IMAGE_PROVIDER_SUBMIT_TIMEOUT_MS = 90_000;
const GRSAI_IMAGE_SUBMIT_TIMEOUT_MS = Number(process.env.GRSAI_IMAGE_SUBMIT_TIMEOUT_MS || 30_000);
const GEMINI_IMAGE_SUBMIT_TIMEOUT_MS = 180_000; const GEMINI_IMAGE_SUBMIT_TIMEOUT_MS = 180_000;
const DASHSCOPE_VIDEO_STYLE_ENDPOINT = "https://dashscope.aliyuncs.com/api/v1/services/aigc/video-generation/video-synthesis"; const DASHSCOPE_VIDEO_STYLE_ENDPOINT = "https://dashscope.aliyuncs.com/api/v1/services/aigc/video-generation/video-synthesis";
const DASHSCOPE_IMAGE_EDIT_ENDPOINT = "https://dashscope.aliyuncs.com/api/v1/services/aigc/image2image/image-synthesis"; const DASHSCOPE_IMAGE_EDIT_ENDPOINT = "https://dashscope.aliyuncs.com/api/v1/services/aigc/image2image/image-synthesis";
@@ -99,18 +98,6 @@ function clampImageQualityForModel(model, quality) {
return normalized; return normalized;
} }
function isDashscopeWan27Limited2KScene(params) {
const model = String(params?.model || "").toLowerCase();
if (model !== "wan2.7-image-pro") return false;
const hasReferenceImages = Array.isArray(params.referenceUrls) && params.referenceUrls.some(Boolean);
return hasReferenceImages || getGridCount(params.gridMode) > 1;
}
function resolveDashscopeImageQuality(params) {
const quality = clampImageQualityForModel(params.model, params.quality);
return isDashscopeWan27Limited2KScene(params) && quality === "4K" ? "2K" : quality;
}
function clampGrsaiImageQualityForModel(model, quality) { function clampGrsaiImageQualityForModel(model, quality) {
const normalized = normalizeQuality(quality, "1K"); const normalized = normalizeQuality(quality, "1K");
const maxQuality = GRSAI_IMAGE_MAX_QUALITY.get(String(model || "").toLowerCase()); const maxQuality = GRSAI_IMAGE_MAX_QUALITY.get(String(model || "").toLowerCase());
@@ -348,25 +335,18 @@ async function assertUserGenerationConcurrencyLimit(userId, client = pool) {
[userId], [userId],
); );
const { rows: limitRows } = await client.query(
"SELECT max_concurrency FROM users WHERE id = $1",
[userId],
);
const rawLimit = Number(limitRows[0]?.max_concurrency);
const concurrencyLimit = Number.isFinite(rawLimit) && rawLimit > 0 ? rawLimit : MAX_USER_ACTIVE_GENERATION_TASKS;
const { rows } = await client.query( const { rows } = await client.query(
"SELECT COUNT(*)::int AS active_count FROM generation_tasks WHERE user_id = $1 AND status IN ('pending', 'running')", "SELECT COUNT(*)::int AS active_count FROM generation_tasks WHERE user_id = $1 AND status IN ('pending', 'running')",
[userId], [userId],
); );
const activeCount = Number(rows[0]?.active_count ?? rows[0]?.count ?? 0); const activeCount = Number(rows[0]?.active_count ?? rows[0]?.count ?? 0);
if (activeCount < concurrencyLimit) return; if (activeCount < MAX_USER_ACTIVE_GENERATION_TASKS) return;
const error = new Error(`最多只能同时进行${concurrencyLimit}个任务`); const error = new Error(GENERATION_CONCURRENCY_LIMIT_MESSAGE);
error.status = 429; error.status = 429;
error.code = "GENERATION_CONCURRENCY_LIMIT"; error.code = "GENERATION_CONCURRENCY_LIMIT";
error.activeCount = activeCount; error.activeCount = activeCount;
error.maxActiveTasks = concurrencyLimit; error.maxActiveTasks = MAX_USER_ACTIVE_GENERATION_TASKS;
throw error; throw error;
} }
@@ -490,22 +470,17 @@ function buildDashscopeImageBody(params) {
if (url) content.push({ image: url }); if (url) content.push({ image: url });
} }
content.push({ text: params.prompt }); content.push({ text: params.prompt });
const quality = resolveDashscopeImageQuality(params); const quality = clampImageQualityForModel(params.model, params.quality);
const gridCount = getGridCount(params.gridMode);
const parameters = {
size: mapAspectRatioToDashscopeSize(params.ratio, quality),
n: gridCount,
watermark: false,
};
if (gridCount > 1) {
parameters.enable_sequential = true;
}
return { return {
model: params.model, model: params.model,
input: { input: {
messages: [{ role: "user", content }], messages: [{ role: "user", content }],
}, },
parameters, parameters: {
size: mapAspectRatioToDashscopeSize(params.ratio, quality),
n: params.gridMode === "grid-4" ? 4 : params.gridMode === "grid-9" ? 9 : 1,
watermark: false,
},
}; };
} }
@@ -1073,16 +1048,6 @@ function registerAiRoutes(router) {
error.costCents = billingResult.costCents; error.costCents = billingResult.costCents;
throw error; throw error;
} }
if (billingResult.costCents > 0) {
await client.query(
"UPDATE generation_tasks SET cost_cents = $1, billing_target = $2, billing_refunded = 0, updated_at = NOW() WHERE id = $3",
[
billingResult.costCents,
billingResult.deductionType === "enterprise_image_flat" ? "enterprise_image" : "user",
nextTaskRow.id,
],
);
}
return { taskRow: nextTaskRow, imageBilling: billingResult }; return { taskRow: nextTaskRow, imageBilling: billingResult };
}); });
const preauth = { authorized: true, estimatedCostCents: 0, billingMode: imageBilling.deductionType }; const preauth = { authorized: true, estimatedCostCents: 0, billingMode: imageBilling.deductionType };
@@ -1097,11 +1062,9 @@ function registerAiRoutes(router) {
}, },
providerDebug: buildImageProviderDebug(model), providerDebug: buildImageProviderDebug(model),
}); });
submitImageWithProviderFallback(taskRow.id, providerCandidates, req.user, preauth, params).catch(async (err) => { submitImageWithProviderFallback(taskRow.id, providerCandidates, req.user, preauth, params).catch((err) => {
console.error("[ai/image] submit error:", err.message); console.error("[ai/image] submit error:", err.message);
await updateTaskInDb(taskRow.id, { status: "failed", error: err.message }).catch((updateErr) => { updateTaskInDb(taskRow.id, { status: "failed", error: err.message });
console.error(`[ai/image] failed to persist task ${taskRow.id} failure:`, updateErr.message);
});
}); });
} catch (err) { } catch (err) {
console.error("[ai/image] error:", err.message); console.error("[ai/image] error:", err.message);
@@ -1213,10 +1176,6 @@ function registerAiRoutes(router) {
...enterpriseBilling, ...enterpriseBilling,
taskId: nextTaskRow.id, taskId: nextTaskRow.id,
}); });
await client.query(
"UPDATE generation_tasks SET cost_cents = $1, billing_target = 'enterprise_video', billing_refunded = 0, updated_at = NOW() WHERE id = $2",
[nextBilling.amountCents, nextTaskRow.id],
);
return { taskRow: nextTaskRow, reservedBilling: nextBilling, regularBilling: null }; return { taskRow: nextTaskRow, reservedBilling: nextBilling, regularBilling: null };
} }
// Regular user: deduct from personal balance // Regular user: deduct from personal balance
@@ -1239,10 +1198,6 @@ function registerAiRoutes(router) {
"INSERT INTO transactions (user_id, type, amount_cents, balance_after_cents, description) VALUES ($1, 'deduct', $2, $3, $4)", "INSERT INTO transactions (user_id, type, amount_cents, balance_after_cents, description) VALUES ($1, 'deduct', $2, $3, $4)",
[req.user.id, -costCents, deducted.balance_cents, `视频生成扣费 ${credits} 积分`], [req.user.id, -costCents, deducted.balance_cents, `视频生成扣费 ${credits} 积分`],
); );
await client.query(
"UPDATE generation_tasks SET cost_cents = $1, billing_target = 'user', billing_refunded = 0, updated_at = NOW() WHERE id = $2",
[costCents, nextTaskRow.id],
);
return { taskRow: nextTaskRow, reservedBilling: null, regularBilling: { costCents, balanceAfterCents: deducted.balance_cents, credits } }; return { taskRow: nextTaskRow, reservedBilling: null, regularBilling: { costCents, balanceAfterCents: deducted.balance_cents, credits } };
}); });
@@ -1764,35 +1719,6 @@ function registerAiRoutes(router) {
} }
}); });
const streamTaskStatusPoll = async (taskId, userId, emit) => {
const { rows } = await pool.query(
"SELECT * FROM generation_tasks WHERE id = $1 AND user_id = $2",
[taskId, userId],
);
const row = rows[0];
if (!row) return { found: false, terminal: true };
if (row.status === "pending" || row.status === "running") {
pool.query(
"UPDATE generation_tasks SET last_poll_at = NOW() WHERE id = $1",
[taskId],
).catch(() => {});
}
const event = {
taskId: row.id,
status: row.status,
progress: Number(row.progress || 0),
resultUrl: row.result_url || null,
error: row.error || null,
};
emit(event);
return {
found: true,
terminal: ["completed", "failed", "cancelled"].includes(row.status),
};
};
router.get("/ai/tasks/:taskId/stream", requireAuth, async (req, res) => { router.get("/ai/tasks/:taskId/stream", requireAuth, async (req, res) => {
const { taskId } = req.params; const { taskId } = req.params;
try { try {
@@ -1824,43 +1750,16 @@ function registerAiRoutes(router) {
return; return;
} }
let closed = false;
let lastSnapshot = JSON.stringify(initial);
let dbPollTimer = null;
const endStream = () => {
if (closed) return;
closed = true;
if (dbPollTimer) clearInterval(dbPollTimer);
taskEvents.off(`task:${taskId}`, onUpdate);
res.end();
};
const emitIfChanged = (evt) => {
if (closed) return;
const snapshot = JSON.stringify(evt);
if (snapshot === lastSnapshot) return;
lastSnapshot = snapshot;
res.write(`data: ${snapshot}\n\n`);
};
const onUpdate = (evt) => { const onUpdate = (evt) => {
emitIfChanged(evt); res.write(`data: ${JSON.stringify(evt)}\n\n`);
if (["completed", "failed", "cancelled"].includes(evt.status)) { if (["completed", "failed", "cancelled"].includes(evt.status)) {
endStream(); res.end();
} }
}; };
taskEvents.on(`task:${taskId}`, onUpdate); taskEvents.on(`task:${taskId}`, onUpdate);
dbPollTimer = setInterval(() => {
streamTaskStatusPoll(taskId, req.user.id, emitIfChanged)
.then((result) => {
if (!result.found || result.terminal) endStream();
})
.catch((pollErr) => {
console.error(`[ai/task-stream] db poll failed for task ${taskId}:`, pollErr.message);
});
}, 3000);
req.on("close", () => { req.on("close", () => {
endStream(); taskEvents.off(`task:${taskId}`, onUpdate);
}); });
} catch (err) { } catch (err) {
if (!res.headersSent) res.status(err.name === "AbortError" ? 504 : 500).json({ error: err.name === "AbortError" ? "AI 上游响应超时，请重试" : err.message }); if (!res.headersSent) res.status(err.name === "AbortError" ? 504 : 500).json({ error: err.name === "AbortError" ? "AI 上游响应超时，请重试" : err.message });
@@ -1872,13 +1771,9 @@ function registerAiRoutes(router) {
if (!Number.isFinite(taskId)) return res.status(400).json({ error: "Invalid task id" }); if (!Number.isFinite(taskId)) return res.status(400).json({ error: "Invalid task id" });
try { try {
const { rows } = await pool.query( const task = await cancelTask(taskId, req.user.id);
"UPDATE generation_tasks SET status = 'cancelled', updated_at = NOW() WHERE id = $1 AND user_id = $2 AND status IN ('pending', 'running') RETURNING id, status", if (!task) return res.status(404).json({ error: "Task not found or not in active state" });
[taskId, req.user.id], res.json({ id: task.id, status: task.status });
);
if (rows.length === 0) return res.status(404).json({ error: "Task not found or not in active state" });
await cancelTaskRuntimeState(taskId, keyManager);
res.json({ id: rows[0].id, status: rows[0].status });
} catch (err) { } catch (err) {
console.error("[ai/task-cancel] error:", err.message); console.error("[ai/task-cancel] error:", err.message);
res.status(500).json({ error: "取消任务失败" }); res.status(500).json({ error: "取消任务失败" });
@@ -2036,12 +1931,7 @@ async function submitImageToProvider(taskDbId, providerConfig, slotResult, param
const { headers, body } = buildImageRequest(providerConfig, params, slotResult.apiKey); const { headers, body } = buildImageRequest(providerConfig, params, slotResult.apiKey);
await updateTaskInDb(taskDbId, { status: "running", progress: 10 }); await updateTaskInDb(taskDbId, { status: "running", progress: 10 });
const defaultSubmitTimeout = providerConfig.transport === "gemini-image" const submitTimeout = providerConfig.transport === "gemini-image" ? GEMINI_IMAGE_SUBMIT_TIMEOUT_MS : IMAGE_PROVIDER_SUBMIT_TIMEOUT_MS;
? GEMINI_IMAGE_SUBMIT_TIMEOUT_MS
: providerConfig.transport === "grsai-image"
? GRSAI_IMAGE_SUBMIT_TIMEOUT_MS
: IMAGE_PROVIDER_SUBMIT_TIMEOUT_MS;
const submitTimeout = getAdaptiveTimeout(providerConfig.provider, defaultSubmitTimeout);
const response = await fetchWithTimeout(url, { method: "POST", headers, body: JSON.stringify(body) }, submitTimeout); const response = await fetchWithTimeout(url, { method: "POST", headers, body: JSON.stringify(body) }, submitTimeout);
if (!response.ok) { if (!response.ok) {
const errText = await response.text().catch(() => "provider error"); const errText = await response.text().catch(() => "provider error");
src/routes/betaApplications.js
-519
View File
@@ -1,519 +0,0 @@
"use strict";
const { getUserContextById, requireAuth, verifyToken } = require("../auth");
const { pool, withTransaction } = require("../db");
const { loadBetaInviteCodes, normalizeBetaInviteCode } = require("../betaInviteCodes");
const REVIEW_USERNAMES = new Set(["xqy1912"]);
const EMAIL_PATTERN = /^[^\s@]+@[^\s@]+\.[^\s@]+$/;
function cleanText(value, maxLength) {
return String(value || "").trim().slice(0, maxLength);
}
function cleanTextArray(value, maxItems = 20, maxLength = 200) {
if (!Array.isArray(value)) return [];
return value.map((item) => cleanText(item, maxLength)).filter(Boolean).slice(0, maxItems);
}
function normalizeEmail(email) {
return String(email || "").trim().toLowerCase();
}
function validateEmail(email) {
const normalized = normalizeEmail(email);
if (!normalized) return "请填写用于接收内测码的邮箱";
if (!EMAIL_PATTERN.test(normalized)) return "邮箱格式不正确";
return null;
}
function parseJson(value, fallback) {
if (!value || typeof value !== "string") return fallback;
try {
return JSON.parse(value);
} catch {
return fallback;
}
}
function safeJsonString(value, fallback) {
try {
return JSON.stringify(value ?? fallback);
} catch {
return JSON.stringify(fallback);
}
}
function buildSmtpTransportOptions(scope) {
const prefix = scope ? `${scope}_` : "";
return {
host: process.env[`${prefix}SMTP_HOST`] || process.env.SMTP_HOST,
port: Number(process.env[`${prefix}SMTP_PORT`] || process.env.SMTP_PORT) || 587,
secure: String(process.env[`${prefix}SMTP_SECURE`] || process.env.SMTP_SECURE || "") === "1",
auth: {
user: process.env[`${prefix}SMTP_USER`] || process.env.SMTP_USER,
pass: process.env[`${prefix}SMTP_PASS`] || process.env.SMTP_PASS,
},
};
}
function formatEmailAddress(address, displayName) {
const email = String(address || "").trim();
const name = String(displayName || "").trim();
if (!name) return email;
const escapedName = name.replace(/"/g, '\\"');
return `"${escapedName}" <${email}>`;
}
function getRequestIp(req) {
const forwardedFor = String(req.headers["x-forwarded-for"] || "").split(",")[0].trim();
return forwardedFor || req.socket?.remoteAddress || "";
}
async function optionalAuth(req, _res, next) {
const authHeader = req.headers.authorization;
if (!authHeader?.startsWith("Bearer ")) {
next();
return;
}
try {
const payload = verifyToken(authHeader.slice(7));
const user = await getUserContextById(payload.userId);
if (user?.enabled) req.user = user;
} catch {
// Public application submission should still work without a valid session.
}
next();
}
function canReviewBetaApplications(user) {
if (!user) return false;
const role = String(user.role || "").trim().toLowerCase();
const username = String(user.username || "").trim().toLowerCase();
return role === "admin" || REVIEW_USERNAMES.has(username);
}
function requireBetaApplicationReviewer(req, res, next) {
if (!canReviewBetaApplications(req.user)) {
return res.status(403).json({ error: "无权审核内测申请" });
}
next();
}
async function ensureBetaApplicationSchema() {
await pool.query(`
CREATE TABLE IF NOT EXISTS beta_applications (
id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
name TEXT,
email TEXT,
phone TEXT,
wechat TEXT,
industry TEXT,
company TEXT,
city TEXT,
ai_tools TEXT,
ai_duration TEXT,
ai_track TEXT,
ai_direction_json TEXT NOT NULL DEFAULT '[]',
weekly_usage TEXT,
feedback_willing TEXT,
want_feature_json TEXT NOT NULL DEFAULT '[]',
self_statement TEXT,
signature TEXT,
application_date TEXT,
agree_rules INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'pending',
invite_code TEXT,
review_note TEXT,
reviewed_by INTEGER REFERENCES users(id) ON DELETE SET NULL,
reviewed_at TIMESTAMPTZ,
ip_address TEXT,
user_agent TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_beta_applications_status_created
ON beta_applications(status, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_beta_applications_user_created
ON beta_applications(user_id, created_at DESC);
ALTER TABLE beta_applications
ADD COLUMN IF NOT EXISTS email TEXT;
ALTER TABLE beta_applications
ADD COLUMN IF NOT EXISTS application_date TEXT;
CREATE INDEX IF NOT EXISTS idx_beta_applications_email
ON beta_applications(LOWER(email));
`);
}
function normalizeApplicationBody(body) {
return {
name: cleanText(body?.name, 120),
email: normalizeEmail(body?.email),
phone: cleanText(body?.phone, 60),
wechat: cleanText(body?.wechat, 120),
industry: cleanText(body?.industry, 160),
company: cleanText(body?.company, 200),
city: cleanText(body?.city, 120),
aiTools: cleanText(body?.aiTools ?? body?.ai_tools, 1000),
aiDuration: cleanText(body?.aiDuration ?? body?.ai_duration, 120),
aiTrack: cleanText(body?.aiTrack ?? body?.ai_track, 160),
aiDirection: cleanTextArray(body?.aiDirection ?? body?.ai_direction),
weeklyUsage: cleanText(body?.weeklyUsage ?? body?.weekly_usage, 120),
feedbackWilling: cleanText(body?.feedbackWilling ?? body?.feedback_willing, 160),
wantFeature: cleanTextArray(body?.wantFeature ?? body?.want_feature),
selfStatement: cleanText(body?.selfStatement ?? body?.self_statement, 5000),
signature: cleanText(body?.signature, 120),
applicationDate: cleanText(body?.applicationDate ?? body?.application_date, 120),
agreeRules: body?.agreeRules === true || body?.agree_rules === true || body?.agreeRules === 1 || body?.agree_rules === 1,
};
}
function formatApplication(row) {
return {
id: Number(row.id),
userId: row.user_id == null ? null : Number(row.user_id),
username: row.username || null,
name: row.name || "",
email: row.email || "",
phone: row.phone || "",
wechat: row.wechat || "",
industry: row.industry || "",
company: row.company || "",
city: row.city || "",
aiTools: row.ai_tools || "",
aiDuration: row.ai_duration || "",
aiTrack: row.ai_track || "",
aiDirection: parseJson(row.ai_direction_json, []),
weeklyUsage: row.weekly_usage || "",
feedbackWilling: row.feedback_willing || "",
wantFeature: parseJson(row.want_feature_json, []),
selfStatement: row.self_statement || "",
signature: row.signature || "",
applicationDate: row.application_date || "",
agreeRules: Boolean(row.agree_rules),
status: row.status || "pending",
inviteCode: row.invite_code || null,
reviewNote: row.review_note || null,
reviewedBy: row.reviewed_by == null ? null : Number(row.reviewed_by),
reviewerUsername: row.reviewer_username || null,
reviewedAt: row.reviewed_at || null,
ipAddress: row.ip_address || null,
userAgent: row.user_agent || null,
createdAt: row.created_at,
updatedAt: row.updated_at,
};
}
async function selectApplicationById(client, id) {
const { rows } = await client.query(
`
SELECT a.*, u.username, reviewer.username AS reviewer_username
FROM beta_applications a
LEFT JOIN users u ON u.id = a.user_id
LEFT JOIN users reviewer ON reviewer.id = a.reviewed_by
WHERE a.id = $1
LIMIT 1
`,
[id],
);
return rows[0] || null;
}
async function issueNextBetaInviteCode(client) {
const codes = Array.from(loadBetaInviteCodes()).map(normalizeBetaInviteCode).filter(Boolean).sort();
for (const code of codes) {
const { rows } = await client.query(
`
SELECT 1
FROM beta_invite_code_uses
WHERE code = $1
UNION ALL
SELECT 1
FROM beta_applications
WHERE invite_code = $1 AND status = 'approved'
LIMIT 1
`,
[code],
);
if (rows.length === 0) return code;
}
return null;
}
async function createNotification(client, userId, input) {
if (!userId) return;
await client.query(
`
INSERT INTO web_notifications (
user_id, type, title, description, target_type, target_id, metadata_json
)
VALUES ($1, $2, $3, $4, $5, $6, $7)
`,
[
userId,
input.type,
input.title,
input.description || null,
input.targetType || "beta_application",
input.targetId ? String(input.targetId) : null,
safeJsonString(input.metadata, {}),
],
);
}
function buildReviewEmailContent(application, action, inviteCode, reviewNote) {
const name = application.name || "内测申请人";
if (action === "approve") {
const text = [
`${name}，您好：`,
"",
"您的 OmniAI 内测申请已通过。",
`内测码：${inviteCode}`,
"",
"请在注册页面填写该内测码完成账号注册。内测码仅限本人使用，请勿转发。",
"",
"OmniAI 团队",
].join("\n");
const html = `
<div style="font-family:Arial,'Microsoft YaHei',sans-serif;max-width:560px;margin:0 auto;padding:24px;color:#222">
<h2 style="margin:0 0 16px;color:#166534">OmniAI 内测申请已通过</h2>
<p>${name}，您好：</p>
<p>您的 OmniAI 内测申请已通过。</p>
<p style="padding:14px 16px;background:#f0fdf4;border:1px solid #bbf7d0;border-radius:8px;font-size:20px;font-weight:700;letter-spacing:1px;color:#166534">内测码：${inviteCode}</p>
<p>请在注册页面填写该内测码完成账号注册。内测码仅限本人使用，请勿转发。</p>
<p style="margin-top:24px;color:#666">OmniAI 团队</p>
</div>
`;
return { subject: "[OmniAI] 内测申请已通过", text, html };
}
const reason = reviewNote || "很遗憾，您的内测申请暂未通过。";
const text = [
`${name}，您好：`,
"",
"您未通过 OmniAI 内测申请。",
`审核备注：${reason}`,
"",
"感谢您的关注。",
"",
"OmniAI 团队",
].join("\n");
const html = `
<div style="font-family:Arial,'Microsoft YaHei',sans-serif;max-width:560px;margin:0 auto;padding:24px;color:#222">
<h2 style="margin:0 0 16px;color:#991b1b">OmniAI 内测申请未通过</h2>
<p>${name}，您好：</p>
<p>您未通过 OmniAI 内测申请。</p>
<p style="padding:12px 14px;background:#fef2f2;border:1px solid #fecaca;border-radius:8px;color:#7f1d1d">审核备注：${reason}</p>
<p>感谢您的关注。</p>
<p style="margin-top:24px;color:#666">OmniAI 团队</p>
</div>
`;
return { subject: "[OmniAI] 内测申请未通过", text, html };
}
async function sendBetaApplicationReviewEmail(application, action, inviteCode, reviewNote) {
const email = normalizeEmail(application.email);
const emailError = validateEmail(email);
if (emailError) {
const err = new Error(`申请邮箱无效，无法发送审核结果：${emailError}`);
err.status = 409;
throw err;
}
const provider = String(process.env.EMAIL_PROVIDER || "mock").trim().toLowerCase();
const content = buildReviewEmailContent(application, action, inviteCode, reviewNote);
if (provider === "smtp") {
const nodemailer = require("nodemailer");
const smtpOptions = buildSmtpTransportOptions("BETA");
const transporter = nodemailer.createTransport(smtpOptions);
const fromAddress = process.env.BETA_SMTP_FROM || process.env.SMTP_FROM || smtpOptions.auth.user;
const fromName = process.env.BETA_SMTP_FROM_NAME || process.env.SMTP_FROM_NAME || "万物可爱";
await transporter.sendMail({
from: formatEmailAddress(fromAddress, fromName),
to: email,
subject: content.subject,
text: content.text,
html: content.html,
});
return { provider: "smtp" };
}
console.log(`[beta-application-email:${action}] ${email} ${content.subject}`);
return { provider: "mock" };
}
function registerBetaApplicationRoutes(router) {
router.post("/beta-applications", optionalAuth, async (req, res) => {
try {
await ensureBetaApplicationSchema();
const app = normalizeApplicationBody(req.body);
const emailError = validateEmail(app.email);
if (!app.name || emailError || !app.phone || !app.wechat || !app.selfStatement || !app.signature || !app.applicationDate || !app.agreeRules) {
return res.status(400).json({ error: emailError || "请填写姓名、手机号、微信、申请自述、签名、申请日期并同意内测规则" });
}
const { rows } = await pool.query(
`
INSERT INTO beta_applications (
user_id, name, email, phone, wechat, industry, company, city,
ai_tools, ai_duration, ai_track, ai_direction_json,
weekly_usage, feedback_willing, want_feature_json,
self_statement, signature, application_date, agree_rules, ip_address, user_agent
)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21)
RETURNING id, status, created_at
`,
[
req.user?.id || null,
app.name,
app.email,
app.phone,
app.wechat,
app.industry || null,
app.company || null,
app.city || null,
app.aiTools || null,
app.aiDuration || null,
app.aiTrack || null,
safeJsonString(app.aiDirection, []),
app.weeklyUsage || null,
app.feedbackWilling || null,
safeJsonString(app.wantFeature, []),
app.selfStatement,
app.signature,
app.applicationDate,
app.agreeRules ? 1 : 0,
getRequestIp(req),
cleanText(req.headers["user-agent"], 1000) || null,
],
);
res.status(201).json({
application: {
id: rows[0].id,
status: rows[0].status,
createdAt: rows[0].created_at,
},
});
} catch (err) {
console.error("[beta-applications] create failed:", err.message);
res.status(500).json({ error: "提交内测申请失败" });
}
});
router.get("/admin/beta-applications", requireAuth, requireBetaApplicationReviewer, async (req, res) => {
try {
await ensureBetaApplicationSchema();
const status = cleanText(req.query.status, 32);
const params = [];
const where = [];
if (status) {
params.push(status);
where.push(`a.status = $${params.length}`);
}
const { rows } = await pool.query(
`
SELECT a.*, u.username, reviewer.username AS reviewer_username
FROM beta_applications a
LEFT JOIN users u ON u.id = a.user_id
LEFT JOIN users reviewer ON reviewer.id = a.reviewed_by
${where.length ? `WHERE ${where.join(" AND ")}` : ""}
ORDER BY
CASE a.status WHEN 'pending' THEN 0 WHEN 'approved' THEN 1 ELSE 2 END,
a.created_at DESC
LIMIT 300
`,
params,
);
res.json({ applications: rows.map(formatApplication) });
} catch (err) {
console.error("[admin/beta-applications] list failed:", err.message);
res.status(500).json({ error: "读取内测申请失败" });
}
});
router.patch("/admin/beta-applications/:id", requireAuth, requireBetaApplicationReviewer, async (req, res) => {
const id = Number(req.params.id);
const action = cleanText(req.body?.action, 32);
const reviewNote = cleanText(req.body?.reviewNote ?? req.body?.review_note, 1000) || null;
if (!Number.isFinite(id)) return res.status(400).json({ error: "申请 ID 不正确" });
if (action !== "approve" && action !== "reject") return res.status(400).json({ error: "审核动作不正确" });
try {
await ensureBetaApplicationSchema();
const application = await withTransaction(async (client) => {
const current = await selectApplicationById(client, id);
if (!current) {
const err = new Error("申请不存在");
err.status = 404;
throw err;
}
if (current.status !== "pending") {
const err = new Error("该申请已审核");
err.status = 409;
throw err;
}
let inviteCode = null;
if (action === "approve") {
inviteCode = await issueNextBetaInviteCode(client);
if (!inviteCode) {
const err = new Error("暂无可用内测码，请先补充内测码");
err.status = 409;
throw err;
}
}
const { rows } = await client.query(
`
UPDATE beta_applications
SET status = $1,
invite_code = $2,
review_note = $3,
reviewed_by = $4,
reviewed_at = NOW(),
updated_at = NOW()
WHERE id = $5
RETURNING *
`,
[action === "approve" ? "approved" : "rejected", inviteCode, reviewNote, req.user.id, id],
);
const updated = rows[0];
await sendBetaApplicationReviewEmail(updated, action, inviteCode, reviewNote);
if (updated.user_id) {
if (action === "approve") {
await createNotification(client, updated.user_id, {
type: "review_passed",
title: "内测申请已通过",
description: `您的内测申请已通过，内测码：${inviteCode}`,
targetId: updated.id,
metadata: { inviteCode },
});
} else {
await createNotification(client, updated.user_id, {
type: "review_rejected",
title: "您未通过内测申请",
description: reviewNote || "很遗憾，您的内测申请暂未通过。",
targetId: updated.id,
});
}
}
return selectApplicationById(client, id);
});
res.json({ application: formatApplication(application) });
} catch (err) {
const status = Number(err.status || 500);
if (status >= 400 && status < 500) return res.status(status).json({ error: err.message });
console.error("[admin/beta-applications] review failed:", err.message);
res.status(500).json({ error: "审核内测申请失败" });
}
});
}
module.exports = { registerBetaApplicationRoutes, canReviewBetaApplications };
src/routes/bugFeedback.js
-96
View File
@@ -1,96 +0,0 @@
const express = require("express");
const { requireAuth, requireAdmin } = require("../auth");
const { pool } = require("../db");
const { creditUserBalance } = require("../billing");
const router = express.Router();
router.post("/bug-feedback", requireAuth, async (req, res) => {
const userId = req.user.id;
const { title, description, screenshotUrl } = req.body;
if (!title || String(title).trim().length === 0) return res.status(400).json({ error: "标题不能为空" });
if (!description || String(description).trim().length === 0) return res.status(400).json({ error: "描述不能为空" });
if (String(title).length > 200) return res.status(400).json({ error: "标题不能超过200字" });
if (String(description).length > 5000) return res.status(400).json({ error: "描述不能超过5000字" });
try {
const result = await pool.query(
`INSERT INTO bug_feedback (user_id, title, description, screenshot_url) VALUES ($1, $2, $3, $4) RETURNING id, status, created_at`,
[userId, String(title).trim(), String(description).trim(), screenshotUrl || null]
);
res.json({ feedback: { id: result.rows[0].id, status: result.rows[0].status, createdAt: result.rows[0].created_at } });
} catch (err) {
console.error("[bug-feedback] submit failed:", err.message);
res.status(500).json({ error: "提交失败，请稍后重试" });
}
});
router.get("/bug-feedback/mine", requireAuth, async (req, res) => {
const userId = req.user.id;
try {
const result = await pool.query(
`SELECT id, title, description, screenshot_url, status, admin_note, created_at FROM bug_feedback WHERE user_id = $1 ORDER BY created_at DESC LIMIT 50`,
[userId]
);
res.json({ feedbacks: result.rows.map(r => ({ id: r.id, title: r.title, description: r.description, screenshotUrl: r.screenshot_url, status: r.status, adminNote: r.admin_note, createdAt: r.created_at })) });
} catch (err) {
console.error("[bug-feedback] list mine failed:", err.message);
res.status(500).json({ error: "获取反馈列表失败" });
}
});
router.get("/admin/bug-feedback", requireAuth, requireAdmin, async (req, res) => {
const status = req.query.status || null;
const limit = Math.min(Number(req.query.limit) || 20, 100);
const offset = Number(req.query.offset) || 0;
try {
const where = status ? "WHERE bf.status = $1" : "";
const params = status ? [status, limit, offset] : [limit, offset];
const countWhere = status ? "WHERE status = $1" : "";
const countParams = status ? [status] : [];
const [dataRes, countRes] = await Promise.all([
pool.query(`SELECT bf.id, bf.title, bf.description, bf.screenshot_url, bf.status, bf.admin_note, bf.reward_credited, bf.created_at, u.username FROM bug_feedback bf JOIN users u ON u.id = bf.user_id ${where} ORDER BY bf.created_at DESC LIMIT $${status ? 2 : 1} OFFSET $${status ? 3 : 2}`, params),
pool.query(`SELECT COUNT(*)::int AS total FROM bug_feedback ${countWhere}`, countParams),
]);
res.json({
feedbacks: dataRes.rows.map(r => ({ id: r.id, title: r.title, description: r.description, screenshotUrl: r.screenshot_url, status: r.status, adminNote: r.admin_note, rewardCredited: r.reward_credited, username: r.username, createdAt: r.created_at })),
total: countRes.rows[0].total,
});
} catch (err) {
console.error("[admin/bug-feedback] list failed:", err.message);
res.status(500).json({ error: "获取反馈列表失败" });
}
});
router.patch("/admin/bug-feedback/:id", requireAuth, requireAdmin, async (req, res) => {
const feedbackId = Number(req.params.id);
const { status, adminNote } = req.body;
if (!["approved", "rejected"].includes(status)) return res.status(400).json({ error: "状态只能是 approved 或 rejected" });
const client = await pool.connect();
try {
await client.query("BEGIN");
const existing = await client.query("SELECT id, user_id, status, reward_credited FROM bug_feedback WHERE id = $1 FOR UPDATE", [feedbackId]);
if (existing.rows.length === 0) { await client.query("ROLLBACK"); return res.status(404).json({ error: "反馈不存在" }); }
const row = existing.rows[0];
await client.query("UPDATE bug_feedback SET status = $1, admin_note = $2, updated_at = NOW() WHERE id = $3", [status, adminNote || null, feedbackId]);
let rewardCredited = row.reward_credited;
if (status === "approved" && !row.reward_credited) {
await creditUserBalance(row.user_id, 100, "Bug反馈奖励 1 积分");
await client.query("UPDATE bug_feedback SET reward_credited = TRUE WHERE id = $1", [feedbackId]);
rewardCredited = true;
}
await client.query("COMMIT");
res.json({ success: true, rewardCredited });
} catch (err) {
await client.query("ROLLBACK");
console.error("[admin/bug-feedback] patch failed:", err.message);
res.status(500).json({ error: "操作失败" });
} finally {
client.release();
}
});
module.exports = router;
src/routes/context.js
+14 -113
View File
@@ -32,8 +32,6 @@ const {
getUserEnterpriseId, getUserEnterpriseId,
getEnterpriseName, getEnterpriseName,
preauthorizeCall, preauthorizeCall,
creditsToCreditUnits,
formatCreditsFromCents,
} = require("../billing"); } = require("../billing");
const wechatPay = require("../paymentWechat"); const wechatPay = require("../paymentWechat");
const alipay = require("../paymentAlipay"); const alipay = require("../paymentAlipay");
@@ -212,123 +210,28 @@ function hashEmailCode(email, code) {
return crypto.createHash("sha256").update(email + ":" + code + ":" + secret).digest("hex"); return crypto.createHash("sha256").update(email + ":" + code + ":" + secret).digest("hex");
} }
function buildSmtpTransportOptions(scope) {
const prefix = scope ? `${scope}_` : "";
return {
host: process.env[`${prefix}SMTP_HOST`] || process.env.SMTP_HOST,
port: Number(process.env[`${prefix}SMTP_PORT`] || process.env.SMTP_PORT) || 587,
secure: String(process.env[`${prefix}SMTP_SECURE`] || process.env.SMTP_SECURE || "") === "1",
auth: {
user: process.env[`${prefix}SMTP_USER`] || process.env.SMTP_USER,
pass: process.env[`${prefix}SMTP_PASS`] || process.env.SMTP_PASS,
},
};
}
function formatEmailAddress(address, displayName) {
const email = String(address || "").trim();
const name = String(displayName || "").trim();
if (!name) return email;
const escapedName = name.replace(/"/g, '\\"');
return `"${escapedName}" <${email}>`;
}
function escapeEmailHtml(value) {
return String(value || "")
.replace(/&/g, "&amp;")
.replace(/</g, "&lt;")
.replace(/>/g, "&gt;")
.replace(/"/g, "&quot;");
}
function buildEmailCodeContent(code, purpose) {
const purposeText = purpose === "register" ? "注册" : purpose === "login" ? "登录" : "重置密码";
const ttlText = String(EMAIL_CODE_TTL_MINUTES);
const safeCode = escapeEmailHtml(code);
const safePurposeText = escapeEmailHtml(purposeText);
const preheader = `您的 OmniAI ${purposeText}验证码是 ${code}，${ttlText} 分钟内有效。`;
return {
subject: "[OmniAI] 邮箱验证码",
text:
`您的验证码是：${code}\n` +
`用途：${purposeText}\n` +
`有效期：${ttlText} 分钟\n` +
"请勿将验证码转发给他人。如非本人操作，请忽略此邮件。",
html: `<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>OmniAI 邮箱验证码</title>
</head>
<body style="margin:0;padding:0;background:#f4f7fb;color:#1f2937;font-family:-apple-system,BlinkMacSystemFont,'Segoe UI','PingFang SC','Microsoft YaHei',Arial,sans-serif;">
<div style="display:none;max-height:0;overflow:hidden;opacity:0;color:transparent;">${escapeEmailHtml(preheader)}</div>
<table role="presentation" width="100%" cellspacing="0" cellpadding="0" style="width:100%;background:#f4f7fb;margin:0;padding:28px 12px;">
<tr>
<td align="center">
<table role="presentation" width="100%" cellspacing="0" cellpadding="0" style="width:100%;max-width:560px;background:#ffffff;border-radius:16px;overflow:hidden;border:1px solid #e5ebf3;box-shadow:0 18px 45px rgba(31,41,55,0.08);">
<tr>
<td style="padding:28px 28px 20px;background:#101827;color:#ffffff;">
<div style="font-size:13px;letter-spacing:2px;text-transform:uppercase;color:#a7f3d0;font-weight:700;">OmniAI</div>
<h1 style="margin:10px 0 0;font-size:24px;line-height:1.35;font-weight:800;color:#ffffff;">万物可爱邮箱验证</h1>
<p style="margin:10px 0 0;font-size:14px;line-height:1.8;color:#cbd5e1;">请使用下方验证码完成${safePurposeText}操作。</p>
</td>
</tr>
<tr>
<td style="padding:28px;">
<div style="border:1px solid #dbe6f4;background:#f8fbff;border-radius:14px;padding:22px 18px;text-align:center;">
<div style="font-size:13px;color:#64748b;margin-bottom:10px;">验证码</div>
<div style="font-size:38px;line-height:1.2;letter-spacing:8px;font-weight:800;color:#0f766e;font-family:'SFMono-Regular',Consolas,'Liberation Mono',monospace;">${safeCode}</div>
<div style="font-size:13px;color:#64748b;margin-top:14px;">${ttlText} 分钟内有效</div>
</div>
<table role="presentation" width="100%" cellspacing="0" cellpadding="0" style="margin-top:22px;border-collapse:collapse;">
<tr>
<td style="padding:12px 0;border-bottom:1px solid #edf2f7;color:#64748b;font-size:14px;">用途</td>
<td align="right" style="padding:12px 0;border-bottom:1px solid #edf2f7;color:#111827;font-size:14px;font-weight:700;">${safePurposeText}</td>
</tr>
<tr>
<td style="padding:12px 0;border-bottom:1px solid #edf2f7;color:#64748b;font-size:14px;">有效期</td>
<td align="right" style="padding:12px 0;border-bottom:1px solid #edf2f7;color:#111827;font-size:14px;font-weight:700;">${ttlText} 分钟</td>
</tr>
</table>
<div style="margin-top:22px;padding:14px 16px;border-radius:12px;background:#fff7ed;border:1px solid #fed7aa;color:#9a3412;font-size:13px;line-height:1.8;">
请勿将验证码转发给他人。万物可爱工作人员不会向您索要邮箱验证码。
</div>
<p style="margin:22px 0 0;color:#64748b;font-size:13px;line-height:1.8;">如果不是您本人操作，可以直接忽略此邮件。</p>
</td>
</tr>
<tr>
<td style="padding:18px 28px;background:#f8fafc;border-top:1px solid #edf2f7;color:#94a3b8;font-size:12px;line-height:1.7;text-align:center;">
此邮件由系统自动发送，请勿直接回复。<br>OmniAI · 万物可爱
</td>
</tr>
</table>
</td>
</tr>
</table>
</body>
</html>`,
};
}
async function sendEmailCode(email, code, purpose) { async function sendEmailCode(email, code, purpose) {
const provider = String(process.env.EMAIL_PROVIDER || "mock").trim().toLowerCase(); const provider = String(process.env.EMAIL_PROVIDER || "mock").trim().toLowerCase();
if (provider === "smtp") { if (provider === "smtp") {
const nodemailer = require("nodemailer"); const nodemailer = require("nodemailer");
const smtpOptions = buildSmtpTransportOptions("SYSTEM"); const transporter = nodemailer.createTransport({
const transporter = nodemailer.createTransport(smtpOptions); host: process.env.SMTP_HOST,
const fromAddress = process.env.SYSTEM_SMTP_FROM || process.env.SMTP_FROM || smtpOptions.auth.user; port: Number(process.env.SMTP_PORT) || 587,
const fromName = process.env.SYSTEM_SMTP_FROM_NAME || process.env.SMTP_FROM_NAME || "万物可爱"; secure: process.env.SMTP_SECURE === "1",
auth: {
user: process.env.SMTP_USER,
pass: process.env.SMTP_PASS,
},
});
const content = buildEmailCodeContent(code, purpose); const purposeText = purpose === "register" ? "注册" : purpose === "login" ? "登录" : "重置密码";
await transporter.sendMail({ await transporter.sendMail({
from: formatEmailAddress(fromAddress, fromName), from: process.env.SMTP_FROM || process.env.SMTP_USER,
to: email, to: email,
subject: content.subject, subject: "[OmniAI] \u90ae\u7bb1\u9a8c\u8bc1\u7801",
text: content.text, text: "\u60a8\u7684\u9a8c\u8bc1\u7801\u662f\uff1a" + code + "\n\u7528\u9014\uff1a" + purposeText + "\n\u6709\u6548\u671f\uff1a" + String(process.env.EMAIL_CODE_TTL_MINUTES || 10) + " \u5206\u949f\n\u5982\u679c\u4e0d\u662f\u60a8\u672c\u4eba\u64cd\u4f5c\uff0c\u8bf7\u5ffd\u7565\u6b64\u90ae\u4ef6\u3002",
html: content.html, html: "<div style=\"font-family:sans-serif;max-width:480px;margin:0 auto;padding:24px\"><h2 style=\"color:#333\">OmniAI \u90ae\u7bb1\u9a8c\u8bc1</h2><p style=\"font-size:16px;color:#555\">\u60a8\u7684\u9a8c\u8bc1\u7801\u662f\uff1a</p><p style=\"font-size:32px;font-weight:bold;letter-spacing:6px;color:#1677ff;margin:16px 0\">" + code + "</p><p style=\"color:#888\">\u7528\u9014\uff1a" + purposeText + "</p><p style=\"color:#888\">\u6709\u6548\u671f\uff1a" + String(process.env.EMAIL_CODE_TTL_MINUTES || 10) + " \u5206\u949f</p><hr style=\"border:none;border-top:1px solid #eee;margin:24px 0\"><p style=\"color:#aaa;font-size:13px\">\u5982\u679c\u4e0d\u662f\u60a8\u672c\u4eba\u64cd\u4f5c\uff0c\u8bf7\u5ffd\u7565\u6b64\u90ae\u4ef6\u3002</p></div>",
}); });
return { provider: "smtp" }; return { provider: "smtp" };
} }
@@ -890,8 +793,6 @@ module.exports = {
getUserEnterpriseId, getUserEnterpriseId,
getEnterpriseName, getEnterpriseName,
preauthorizeCall, preauthorizeCall,
creditsToCreditUnits,
formatCreditsFromCents,
wechatPay, wechatPay,
alipay, alipay,
crypto, crypto,
src/routes/enterprise.js
+8 -17
View File
@@ -2,7 +2,6 @@ const {
requireAuth, requireAuth,
requireEnterpriseAdmin, requireEnterpriseAdmin,
distributeCredits, distributeCredits,
creditsToCreditUnits,
getEnterpriseFinancials, getEnterpriseFinancials,
getEnterpriseName, getEnterpriseName,
pool, pool,
@@ -303,33 +302,25 @@ function registerEnterpriseRoutes(router) {
}); });
router.post("/enterprise/distribute", requireAuth, requireEnterpriseAdmin, async (req, res) => { router.post("/enterprise/distribute", requireAuth, requireEnterpriseAdmin, async (req, res) => {
const { userId, amountCredits, amountCents, distributions } = req.body; const { userId, amountCents, distributions } = req.body;
try { try {
if (distributions && Array.isArray(distributions)) { if (distributions && Array.isArray(distributions)) {
for (const d of distributions) { for (const d of distributions) {
const creditUnits = if (!d.userId || !d.amountCents || d.amountCents <= 0) {
d.amountCredits !== undefined && d.amountCredits !== null && d.amountCredits !== ""
? creditsToCreditUnits(d.amountCredits)
: Number(d.amountCents);
if (!d.userId || !creditUnits || creditUnits <= 0) {
return res return res
.status(400) .status(400)
.json({ error: "每条分发记录必须包含有效的 userId 和 amountCredits" }); .json({ error: "每条分发记录必须包含有效的 userId 和 amountCents" });
} }
await distributeCredits(req.user.enterpriseId, d.userId, creditUnits, req.user.id); await distributeCredits(req.user.enterpriseId, d.userId, d.amountCents, req.user.id);
} }
res.json({ success: true, count: distributions.length }); res.json({ success: true, count: distributions.length });
} else if (userId && (amountCredits || amountCents)) { } else if (userId && amountCents) {
const creditUnits = if (amountCents <= 0) return res.status(400).json({ error: "分发积分必须大于0" });
amountCredits !== undefined && amountCredits !== null && amountCredits !== ""
? creditsToCreditUnits(amountCredits)
: Number(amountCents);
if (!creditUnits || creditUnits <= 0) return res.status(400).json({ error: "分发积分必须大于0" });
const result = await distributeCredits( const result = await distributeCredits(
req.user.enterpriseId, req.user.enterpriseId,
userId, userId,
creditUnits, amountCents,
req.user.id, req.user.id,
); );
res.json({ success: true, ...result }); res.json({ success: true, ...result });
@@ -358,7 +349,7 @@ function registerEnterpriseRoutes(router) {
u.username, u.username,
u.balance_cents AS current_balance_cents, u.balance_cents AS current_balance_cents,
COUNT(acl.id) AS total_calls, COUNT(acl.id) AS total_calls,
COALESCE(SUM(CASE WHEN acl.cost_estimate IS NOT NULL THEN CAST(ROUND((acl.cost_estimate * 10000)::numeric) AS INTEGER) ELSE 0 END), 0) AS total_cost_cents, COALESCE(SUM(CASE WHEN acl.cost_estimate IS NOT NULL THEN CAST(ROUND((acl.cost_estimate * 100)::numeric) AS INTEGER) ELSE 0 END), 0) AS total_cost_cents,
MAX(acl.created_at) AS last_active MAX(acl.created_at) AS last_active
FROM users u FROM users u
LEFT JOIN api_call_logs acl ON acl.user_id = u.id AND acl.status = 'success' LEFT JOIN api_call_logs acl ON acl.user_id = u.id AND acl.status = 'success'
src/routes/index.js
-4
View File
@@ -17,11 +17,9 @@ const { registerConversationRoutes } = require('./conversations')
const { registerReportRoutes } = require('./reports') const { registerReportRoutes } = require('./reports')
const { registerAssetRoutes } = require('./assets') const { registerAssetRoutes } = require('./assets')
const { registerNotificationRoutes } = require('./notifications') const { registerNotificationRoutes } = require('./notifications')
const { registerBetaApplicationRoutes } = require('./betaApplications')
const { registerDraftRoutes } = require('./drafts'); const { registerDraftRoutes } = require('./drafts');
const { registerFileExtractRoutes } = require('./fileExtract'); const { registerFileExtractRoutes } = require('./fileExtract');
const mountClientErrorRoutes = require('./clientErrors') const mountClientErrorRoutes = require('./clientErrors')
const bugFeedbackRouter = require("./bugFeedback")
const router = express.Router() const router = express.Router()
@@ -50,10 +48,8 @@ registerConversationRoutes(router)
registerReportRoutes(router) registerReportRoutes(router)
registerAssetRoutes(router) registerAssetRoutes(router)
registerNotificationRoutes(router) registerNotificationRoutes(router)
registerBetaApplicationRoutes(router)
registerDraftRoutes(router) registerDraftRoutes(router)
registerFileExtractRoutes(router) registerFileExtractRoutes(router)
router.use(bugFeedbackRouter)
registerHealthRoutes(router) registerHealthRoutes(router)
module.exports = router module.exports = router
src/routes/public.js
+1 -6
View File
@@ -1,16 +1,11 @@
const { keyManager, listModelPrices, pool } = require("./context"); const { keyManager, listModelPrices, pool } = require("./context");
const { getEnterpriseVideoPricingConfig } = require("../enterpriseVideoBilling");
function registerPriceRoutes(router) { function registerPriceRoutes(router) {
// ── Public ─────────────────────────────────────────────────────────── // ── Public ───────────────────────────────────────────────────────────
router.get("/prices", async (_req, res) => { router.get("/prices", async (_req, res) => {
const prices = await listModelPrices({ enabledOnly: true }); const prices = await listModelPrices({ enabledOnly: true });
res.json({ res.json(prices);
prices,
modelPrices: prices,
enterpriseVideoPricing: getEnterpriseVideoPricingConfig(),
});
}); });
} }
src/routes/user.js
+6 -6
View File
@@ -136,8 +136,8 @@ function registerUserRoutes(router) {
CASE CASE
WHEN billing_refunded = 1 THEN 0 WHEN billing_refunded = 1 THEN 0
WHEN cost_cents > 0 THEN cost_cents WHEN cost_cents > 0 THEN cost_cents
WHEN status = 'completed' AND type = 'image' THEN 2000 WHEN status = 'completed' AND type = 'image' THEN 20
WHEN status = 'completed' AND type = 'video' THEN 50000 WHEN status = 'completed' AND type = 'video' THEN 500
ELSE 0 ELSE 0
END END
), 0) AS used_cents ), 0) AS used_cents
@@ -162,7 +162,7 @@ function registerUserRoutes(router) {
resolution = params.resolution || params.quality || params.ratio || null; resolution = params.resolution || params.quality || params.ratio || null;
if (row.status === "completed") { if (row.status === "completed") {
if (row.type === "image") { if (row.type === "image") {
estimatedCents = 2000; estimatedCents = 20;
} else if (row.type === "video") { } else if (row.type === "video") {
const dur = params.duration || 5; const dur = params.duration || 5;
const res = String(params.resolution || params.quality || "").toUpperCase(); const res = String(params.resolution || params.quality || "").toUpperCase();
@@ -172,7 +172,7 @@ function registerUserRoutes(router) {
else if (model.includes("wan2.7-i2v") || model.includes("wanxiang")) rate = res === "720P" ? 0.6 : 1; else if (model.includes("wan2.7-i2v") || model.includes("wanxiang")) rate = res === "720P" ? 0.6 : 1;
else if (model.includes("animate-mix") || model.includes("s2v")) rate = res === "720P" ? 0.6 : 1; else if (model.includes("animate-mix") || model.includes("s2v")) rate = res === "720P" ? 0.6 : 1;
else if (model.includes("kling")) rate = res === "720P" ? 0.6 : 0.8; else if (model.includes("kling")) rate = res === "720P" ? 0.6 : 0.8;
estimatedCents = Math.ceil(rate * dur * 10000); estimatedCents = Math.ceil(rate * dur * 100);
} }
} }
} catch { } catch {
@@ -209,8 +209,8 @@ function registerUserRoutes(router) {
CASE CASE
WHEN billing_refunded = 1 THEN 0 WHEN billing_refunded = 1 THEN 0
WHEN cost_cents > 0 THEN cost_cents WHEN cost_cents > 0 THEN cost_cents
WHEN status = 'completed' AND type = 'image' THEN 2000 WHEN status = 'completed' AND type = 'image' THEN 20
WHEN status = 'completed' AND type = 'video' THEN 50000 WHEN status = 'completed' AND type = 'video' THEN 500
ELSE 0 ELSE 0
END END
), 0) AS used_cents ), 0) AS used_cents