Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 stringadmin
|
2249f7ce36 |
+262
-337
@@ -5,6 +5,8 @@ const { EventEmitter } = require("node:events");
|
||||
const { pool } = require("./db");
|
||||
const { refundTaskBillingOnFailure } = require("./billing");
|
||||
const { putObject, isOssConfigured } = require("./ossClient");
|
||||
const keyManager = require("./keyManager");
|
||||
const { resolveImageProviderCandidates, resolveVideoProvider } = require("./aiProviderRouter");
|
||||
|
||||
const taskEvents = new EventEmitter();
|
||||
taskEvents.setMaxListeners(200);
|
||||
@@ -13,15 +15,10 @@ const activePollers = new Map();
|
||||
const POLL_INTERVAL_MS = 3000;
|
||||
const MAX_POLL_ATTEMPTS = 120;
|
||||
const GRS_IMAGE_MAX_POLL_ATTEMPTS = Number(process.env.GRSAI_IMAGE_MAX_POLL_ATTEMPTS || 60);
|
||||
const TASK_EVENT_CHANNEL = "generation_task_events";
|
||||
const TASK_EVENT_ORIGIN = `${process.pid}-${crypto.randomUUID()}`;
|
||||
const POLLER_OWNER_ID = `${process.pid}-${crypto.randomUUID()}`;
|
||||
const POLLER_OWNER_STALE_MS = Number(process.env.TASK_POLLER_OWNER_STALE_MS || 20_000);
|
||||
const POLLER_RECOVERY_INTERVAL_MS = Number(process.env.TASK_POLLER_RECOVERY_INTERVAL_MS || 30_000);
|
||||
let taskEventListenerClient = null;
|
||||
let taskEventListenerStarting = null;
|
||||
let pollerStoreReady = null;
|
||||
let pollerRecoveryTimer = null;
|
||||
const STALE_TASK_TIMEOUT_MINUTES = Math.max(10, Number(process.env.STALE_GENERATION_TASK_MINUTES || 120));
|
||||
const RESULT_PERSIST_RETRY_LIMIT = Math.max(1, Number(process.env.RESULT_PERSIST_RETRY_LIMIT || 5));
|
||||
const RESULT_PERSIST_RETRY_BATCH_SIZE = Math.max(1, Number(process.env.RESULT_PERSIST_RETRY_BATCH_SIZE || 25));
|
||||
const TASK_STARTUP_RECOVERY_LIMIT = Math.max(1, Number(process.env.TASK_STARTUP_RECOVERY_LIMIT || 50));
|
||||
|
||||
function normalizeTaskProgress(value) {
|
||||
const numeric = Number(value);
|
||||
@@ -39,156 +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 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) {
|
||||
if (!task || !task.user_id || !task.id) return;
|
||||
|
||||
@@ -251,14 +98,15 @@ async function updateTaskInDb(taskId, updates) {
|
||||
|
||||
if (fields.length === 0) return;
|
||||
values.push(taskId);
|
||||
const protectCancelled = nextUpdates.status !== "cancelled" ? " AND status <> 'cancelled'" : "";
|
||||
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,
|
||||
);
|
||||
let updatedTask = rows[0];
|
||||
|
||||
if (updatedTask) {
|
||||
await publishTaskEvent(formatTaskEvent(updatedTask));
|
||||
taskEvents.emit(`task:${taskId}`, formatTaskEvent(updatedTask));
|
||||
}
|
||||
|
||||
if (nextUpdates.status === "completed" && updatedTask?.result_url) {
|
||||
@@ -283,20 +131,66 @@ function persistTaskResultUrlToOssInBackground(task) {
|
||||
|
||||
Promise.resolve()
|
||||
.then(async () => {
|
||||
const durableUrl = await persistResultUrlToOss(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`);
|
||||
await persistTaskResultUrlToOss(task);
|
||||
})
|
||||
.catch((error) => {
|
||||
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) {
|
||||
return value && typeof value === "object" && !Array.isArray(value) ? value : undefined;
|
||||
}
|
||||
@@ -795,22 +689,27 @@ function getMaxPollAttempts(type, providerConfig) {
|
||||
return MAX_POLL_ATTEMPTS;
|
||||
}
|
||||
|
||||
function startPolling(taskDbId, { providerTaskId, apiKey, type, providerConfig, leaseToken, keyManager, onTaskFailed, skipPersist = false }) {
|
||||
if (activePollers.has(taskDbId)) return;
|
||||
if (!skipPersist) {
|
||||
persistPollerState(taskDbId, { providerTaskId, type, providerConfig, leaseToken }).catch((err) => {
|
||||
console.error(`[aiTaskWorker] failed to persist poller state for task ${taskDbId}:`, err.message);
|
||||
async function releasePollingLease(poller) {
|
||||
if (!poller?.leaseToken || !poller?.keyManager) return;
|
||||
await poller.keyManager.releaseKey(poller.leaseToken).catch((err) => {
|
||||
console.warn(`[aiTaskWorker] release lease failed for task ${poller.taskDbId}:`, err.message);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function startPolling(taskDbId, { providerTaskId, apiKey, type, providerConfig, leaseToken, keyManager, onTaskFailed }) {
|
||||
if (activePollers.has(taskDbId)) return;
|
||||
|
||||
let attempts = 0;
|
||||
let polling = false;
|
||||
const maxPollAttempts = getMaxPollAttempts(type, providerConfig);
|
||||
const interval = setInterval(async () => {
|
||||
if (polling) return;
|
||||
polling = true;
|
||||
attempts++;
|
||||
|
||||
try {
|
||||
if (attempts > maxPollAttempts) {
|
||||
clearInterval(interval);
|
||||
activePollers.delete(taskDbId);
|
||||
if (leaseToken && keyManager) await keyManager.releaseKey(leaseToken).catch(() => {});
|
||||
await stopPolling(taskDbId, { releaseLease: true });
|
||||
if (typeof onTaskFailed === "function") {
|
||||
const handled = await onTaskFailed("Task timed out").catch((fallbackErr) => {
|
||||
console.error(`[aiTaskWorker] fallback error for task ${taskDbId}:`, fallbackErr.message);
|
||||
@@ -819,21 +718,14 @@ function startPolling(taskDbId, { providerTaskId, apiKey, type, providerConfig,
|
||||
if (handled) return;
|
||||
}
|
||||
await updateTaskInDb(taskDbId, { status: "failed", error: "Task timed out" });
|
||||
await clearPollerState(taskDbId).catch(() => {});
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
// Check if task was cancelled by user
|
||||
const { rows: [taskRow] } = await pool.query("SELECT status FROM generation_tasks WHERE id = $1", [taskDbId]);
|
||||
if (!taskRow || taskRow.status === "cancelled") {
|
||||
clearInterval(interval);
|
||||
activePollers.delete(taskDbId);
|
||||
await clearPollerState(taskDbId).catch(() => {});
|
||||
if (leaseToken && keyManager) await keyManager.releaseKey(leaseToken).catch(() => {});
|
||||
await stopPolling(taskDbId, { releaseLease: true });
|
||||
return;
|
||||
}
|
||||
await refreshPollerHeartbeat(taskDbId).catch(() => {});
|
||||
|
||||
let result;
|
||||
if (type === "image") {
|
||||
@@ -847,9 +739,7 @@ function startPolling(taskDbId, { providerTaskId, apiKey, type, providerConfig,
|
||||
}
|
||||
|
||||
if (result.status === "completed" || result.status === "failed") {
|
||||
clearInterval(interval);
|
||||
activePollers.delete(taskDbId);
|
||||
if (leaseToken && keyManager) await keyManager.releaseKey(leaseToken).catch(() => {});
|
||||
await stopPolling(taskDbId, { releaseLease: true });
|
||||
if (result.status === "failed" && typeof onTaskFailed === "function") {
|
||||
const handled = await onTaskFailed(result.error || "Task failed").catch((fallbackErr) => {
|
||||
console.error(`[aiTaskWorker] fallback error for task ${taskDbId}:`, fallbackErr.message);
|
||||
@@ -860,179 +750,228 @@ function startPolling(taskDbId, { providerTaskId, apiKey, type, providerConfig,
|
||||
}
|
||||
|
||||
await updateTaskInDb(taskDbId, result);
|
||||
if (result.status === "completed" || result.status === "failed") {
|
||||
await clearPollerState(taskDbId).catch(() => {});
|
||||
}
|
||||
} catch (err) {
|
||||
console.error(`[aiTaskWorker] poll error for task ${taskDbId}:`, err.message);
|
||||
} finally {
|
||||
polling = false;
|
||||
}
|
||||
}, 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);
|
||||
if (poller) {
|
||||
if (!poller) return;
|
||||
|
||||
clearInterval(poller.interval);
|
||||
activePollers.delete(taskDbId);
|
||||
if (options.releaseLease) {
|
||||
await releasePollingLease(poller);
|
||||
}
|
||||
clearPollerState(taskDbId).catch(() => {});
|
||||
}
|
||||
|
||||
async function cancelTask(taskId, userId) {
|
||||
const { rows } = await pool.query(
|
||||
`UPDATE generation_tasks
|
||||
SET status = 'cancelled', completed_at = NOW(), updated_at = NOW()
|
||||
WHERE id = $1 AND user_id = $2 AND status IN ('pending', 'running')
|
||||
RETURNING *`,
|
||||
[taskId, userId],
|
||||
);
|
||||
const task = rows[0];
|
||||
if (!task) return null;
|
||||
|
||||
await stopPolling(task.id, { releaseLease: true });
|
||||
taskEvents.emit(`task:${task.id}`, formatTaskEvent(task));
|
||||
return task;
|
||||
}
|
||||
|
||||
function getActiveCount() {
|
||||
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
|
||||
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],
|
||||
);
|
||||
const STALE_TASK_CLEANUP_INTERVAL_MS = 5 * 60 * 1000;
|
||||
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;
|
||||
|
||||
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`);
|
||||
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"),
|
||||
skipPersist: true,
|
||||
});
|
||||
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 {};
|
||||
}
|
||||
}
|
||||
|
||||
// --- 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;
|
||||
let staleTaskCleanupTimer = null;
|
||||
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() {
|
||||
if (taskStaleCleanupRunning) return;
|
||||
taskStaleCleanupRunning = true;
|
||||
try {
|
||||
const { rows } = await pool.query(
|
||||
`UPDATE generation_tasks
|
||||
SET status = 'failed', error = '任务超时自动释放', 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')
|
||||
AND GREATEST(updated_at, COALESCE(last_poll_at, created_at)) < NOW() - INTERVAL '10 minutes'
|
||||
RETURNING id`,
|
||||
AND GREATEST(updated_at, COALESCE(last_poll_at, created_at)) < NOW() - ($1::int * INTERVAL '1 minute')
|
||||
RETURNING *`,
|
||||
[STALE_TASK_TIMEOUT_MINUTES],
|
||||
);
|
||||
|
||||
for (const row of rows) {
|
||||
await publishTaskEvent({
|
||||
taskId: row.id,
|
||||
status: "failed",
|
||||
progress: null,
|
||||
resultUrl: null,
|
||||
error: "任务超时自动释放",
|
||||
await stopPolling(row.id, { releaseLease: true });
|
||||
taskEvents.emit(`task:${row.id}`, formatTaskEvent(row));
|
||||
await refundTaskBillingOnFailure(row.id).catch((err) => {
|
||||
console.error(`[aiTaskWorker] stale task refund error for task ${row.id}:`, err.message);
|
||||
});
|
||||
// Also stop any active poller for this task
|
||||
const poller = activePollers.get(row.id);
|
||||
if (poller) {
|
||||
clearInterval(poller.interval);
|
||||
activePollers.delete(row.id);
|
||||
}
|
||||
await clearPollerState(row.id).catch(() => {});
|
||||
}
|
||||
|
||||
if (rows.length > 0) {
|
||||
console.log(`[aiTaskWorker] Cleaned up ${rows.length} stale task(s)`);
|
||||
}
|
||||
} catch (err) {
|
||||
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 {
|
||||
taskEventListenerStarting = null;
|
||||
taskStaleCleanupRunning = false;
|
||||
}
|
||||
}
|
||||
|
||||
async function stopTaskEventListener() {
|
||||
const client = taskEventListenerClient;
|
||||
taskEventListenerClient = null;
|
||||
if (!client) return;
|
||||
async function runResultPersistenceRetry() {
|
||||
if (taskResultPersistenceRetryRunning) return;
|
||||
taskResultPersistenceRetryRunning = true;
|
||||
try {
|
||||
await client.query(`UNLISTEN ${TASK_EVENT_CHANNEL}`);
|
||||
} catch {}
|
||||
client.release();
|
||||
const { rows } = await pool.query(
|
||||
`SELECT *
|
||||
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() {
|
||||
if (staleTaskCleanupTimer) return;
|
||||
staleTaskCleanupTimer = setInterval(runStaleTaskCleanup, STALE_TASK_CLEANUP_INTERVAL_MS);
|
||||
// Run once shortly after startup
|
||||
taskResultPersistenceRetryTimer = setInterval(runResultPersistenceRetry, TASK_RESULT_PERSIST_RETRY_INTERVAL_MS);
|
||||
taskStartupRecoveryTimer = setTimeout(runTaskStartupRecovery, 5_000);
|
||||
setTimeout(runStaleTaskCleanup, 10_000);
|
||||
setTimeout(runResultPersistenceRetry, 15_000);
|
||||
}
|
||||
|
||||
function stopStaleTaskCleanup() {
|
||||
@@ -1040,30 +979,20 @@ function stopStaleTaskCleanup() {
|
||||
clearInterval(staleTaskCleanupTimer);
|
||||
staleTaskCleanupTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
function startPollerRecovery() {
|
||||
if (pollerRecoveryTimer) return;
|
||||
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;
|
||||
if (taskResultPersistenceRetryTimer) {
|
||||
clearInterval(taskResultPersistenceRetryTimer);
|
||||
taskResultPersistenceRetryTimer = null;
|
||||
}
|
||||
if (taskStartupRecoveryTimer) {
|
||||
clearTimeout(taskStartupRecoveryTimer);
|
||||
taskStartupRecoveryTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
startPolling,
|
||||
stopPolling,
|
||||
cancelTask,
|
||||
updateTaskInDb,
|
||||
getActiveCount,
|
||||
extractProviderTaskId,
|
||||
@@ -1073,10 +1002,6 @@ module.exports = {
|
||||
parseKlingCredential,
|
||||
createKlingJwt,
|
||||
taskEvents,
|
||||
startTaskEventListener,
|
||||
stopTaskEventListener,
|
||||
startPollerRecovery,
|
||||
stopPollerRecovery,
|
||||
startStaleTaskCleanup,
|
||||
stopStaleTaskCleanup,
|
||||
};
|
||||
|
||||
+14
-14
@@ -6,6 +6,7 @@ const { getJwtSecret } = require("./securityConfig");
|
||||
|
||||
const JWT_SECRET = getJwtSecret();
|
||||
const JWT_EXPIRES_IN = process.env.JWT_EXPIRES_IN || "7d";
|
||||
const MAX_CONCURRENT_SESSIONS = 2;
|
||||
|
||||
const USER_CONTEXT_SELECT = `
|
||||
SELECT
|
||||
@@ -169,26 +170,25 @@ function verifyToken(token) {
|
||||
|
||||
async function startUserSession(userId, userAgent) {
|
||||
const sessionId = crypto.randomUUID();
|
||||
const client = await pool.connect();
|
||||
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(
|
||||
await pool.query(
|
||||
"INSERT INTO user_sessions (id, user_id, user_agent, created_at) VALUES ($1, $2, $3, NOW())",
|
||||
[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",
|
||||
[sessionId, userId],
|
||||
);
|
||||
await client.query("COMMIT");
|
||||
} catch (error) {
|
||||
await client.query("ROLLBACK");
|
||||
throw error;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
return sessionId;
|
||||
}
|
||||
|
||||
|
||||
+25
-40
@@ -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.
|
||||
*
|
||||
* Unit conventions:
|
||||
* - payment_orders.amount_cents / packages.price_cents: cash amount in CNY cents.
|
||||
* - users.balance_cents / enterprises.balance_cents / transactions.amount_cents:
|
||||
* credit units, where 100 units = 1 platform credit.
|
||||
* - model_prices.*_mills: CNY mills. 1 CNY = 100 credits, so 1 mill = 10 credit units.
|
||||
* Money conventions:
|
||||
* - balance: cents (分, 1/100 CNY) — stored in users.balance_cents and enterprises.balance_cents
|
||||
* - prices: mills (厘, 1/1000 CNY) — stored in model_prices.*_mills
|
||||
* - cost calculation: mills → convert to cents at deduction time (divide by 10, floor)
|
||||
* - transactions: cents — amount_cents, balance_after_cents
|
||||
*
|
||||
* Flow:
|
||||
* - 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
|
||||
*/
|
||||
|
||||
const { pool, withTransaction } = require("./db");
|
||||
const { calculateCostMills, getModelPrice } = require("./pricing");
|
||||
|
||||
const CREDIT_UNITS_PER_CREDIT = 100;
|
||||
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));
|
||||
}
|
||||
const IMAGE_GENERATION_FLAT_COST_CENTS = 20;
|
||||
|
||||
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)));
|
||||
}
|
||||
|
||||
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) {
|
||||
const enterpriseId = entry?.enterpriseId || null;
|
||||
const userId = entry?.userId || null;
|
||||
@@ -131,6 +114,10 @@ async function getEnterpriseName(enterpriseId) {
|
||||
return rows[0] ? rows[0].name : null;
|
||||
}
|
||||
|
||||
function millsToCents(mills) {
|
||||
return Math.floor(mills / 10);
|
||||
}
|
||||
|
||||
// ── Atomic balance helpers ───────────────────────────────────────────
|
||||
|
||||
async function atomicDeductUserBalance(client, userId, amountCents) {
|
||||
@@ -180,7 +167,7 @@ async function preauthorizeCall(userId, provider) {
|
||||
|
||||
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
|
||||
WHERE provider = $1
|
||||
AND status = 'success'
|
||||
@@ -198,9 +185,10 @@ async function preauthorizeCall(userId, provider) {
|
||||
const bufferedEstimate = Math.ceil(estimatedCostCents * 1.2);
|
||||
|
||||
if (balanceCents < bufferedEstimate) {
|
||||
const credits = Math.floor(balanceCents / 100);
|
||||
return {
|
||||
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" };
|
||||
}
|
||||
|
||||
const costCents = millsToCreditUnits(costMills);
|
||||
const costCents = millsToCents(costMills);
|
||||
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);
|
||||
@@ -420,7 +408,7 @@ async function tryDeductFromUserBalance(userId, enterpriseId, amountCents, ledge
|
||||
userId,
|
||||
-amountCents,
|
||||
newBal,
|
||||
`API 调用扣费 ${formatCreditsFromCents(amountCents)} 积分`,
|
||||
`API 调用扣费 ${Math.ceil(amountCents / 100)} 积分`,
|
||||
],
|
||||
);
|
||||
|
||||
@@ -441,15 +429,16 @@ async function tryDeductFromUserBalance(userId, enterpriseId, amountCents, ledge
|
||||
|
||||
if (newBalanceCents == null) {
|
||||
const currentBalance = await getUserBalanceCents(userId);
|
||||
const credits = Math.floor((currentBalance || 0) / 100);
|
||||
return {
|
||||
success: false,
|
||||
message: `积分不足 (当前 ${formatCreditsFromCents(currentBalance || 0)} 积分，需要 ${formatCreditsFromCents(amountCents)} 积分)`,
|
||||
message: `积分不足 (当前 ${credits} 积分，需要 ${Math.ceil(amountCents / 100)} 积分)`,
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
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,
|
||||
-diffCents,
|
||||
newBal,
|
||||
`API 预估差额扣费 ${formatCreditsFromCents(diffCents)} 积分`,
|
||||
`API 预估差额扣费 ${Math.ceil(diffCents / 100)} 积分`,
|
||||
],
|
||||
);
|
||||
}
|
||||
@@ -514,7 +503,7 @@ async function settleLease(leaseId, actualCostCents) {
|
||||
userId,
|
||||
refundCents,
|
||||
newBal,
|
||||
`API 预估差额退回 ${formatCreditsFromCents(refundCents)} 积分`,
|
||||
`API 预估差额退回 ${Math.ceil(refundCents / 100)} 积分`,
|
||||
],
|
||||
);
|
||||
}
|
||||
@@ -639,7 +628,7 @@ async function distributeCredits(enterpriseId, targetUserId, amountCents, adminU
|
||||
targetUserId,
|
||||
amountCents,
|
||||
newUserBal,
|
||||
`从企业池获得 ${formatCreditsFromCents(amountCents)} 积分`,
|
||||
`从企业池获得 ${Math.floor(amountCents / 100)} 积分`,
|
||||
adminUserId,
|
||||
],
|
||||
);
|
||||
@@ -772,8 +761,4 @@ module.exports = {
|
||||
preauthorizeCall,
|
||||
settleLease,
|
||||
forceSettleLease,
|
||||
creditsToCreditUnits,
|
||||
cashCentsToCreditUnits,
|
||||
millsToCreditUnits,
|
||||
formatCreditsFromCents,
|
||||
};
|
||||
|
||||
@@ -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() {
|
||||
const columns = await getColumnNames("model_prices");
|
||||
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("031_generation_tasks_billing_columns", migrateGenerationTasksBillingColumns);
|
||||
await runMigration("032_ecommerce_video_history", migrateEcommerceVideoHistorySchema);
|
||||
await runMigration("033_generation_task_result_persistence", migrateGenerationTaskResultPersistence);
|
||||
await ensureModelPriceSeed();
|
||||
}
|
||||
|
||||
|
||||
@@ -19,9 +19,6 @@ const ENTERPRISE_VIDEO_ALLOWED_MODELS = new Set([
|
||||
"pixverse-c1-i2v",
|
||||
]);
|
||||
|
||||
const CREDITS_PER_CNY = 100;
|
||||
const CREDIT_UNITS_PER_CREDIT = 100;
|
||||
|
||||
function normalizeModel(value) {
|
||||
return String(value || "").trim().toLowerCase();
|
||||
}
|
||||
@@ -105,7 +102,7 @@ function getEnterpriseVideoCreditRate(input) {
|
||||
|
||||
function calculateEnterpriseVideoCredits(input) {
|
||||
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) {
|
||||
@@ -116,7 +113,7 @@ function calculateEnterpriseVideoCost(input) {
|
||||
resolution,
|
||||
durationSeconds,
|
||||
});
|
||||
const rateCentsPerSecond = Math.round(rateCreditsPerSecond * CREDITS_PER_CNY * CREDIT_UNITS_PER_CREDIT);
|
||||
const rateCentsPerSecond = Math.round(rateCreditsPerSecond * 100);
|
||||
return {
|
||||
resolution,
|
||||
durationSeconds,
|
||||
|
||||
+1
-6
@@ -144,9 +144,7 @@ async function main() {
|
||||
startSettlementWorker()
|
||||
startProviderHealthMonitor()
|
||||
|
||||
const { startStaleTaskCleanup, startTaskEventListener, startPollerRecovery } = require('./aiTaskWorker')
|
||||
await startTaskEventListener()
|
||||
startPollerRecovery()
|
||||
const { startStaleTaskCleanup } = require('./aiTaskWorker')
|
||||
startStaleTaskCleanup()
|
||||
|
||||
server = app.listen(PORT, HOST, () => {
|
||||
@@ -185,9 +183,6 @@ function gracefulShutdown(signal) {
|
||||
console.log('[shutdown] Server closed, cleaning up...')
|
||||
const { stopProviderHealthMonitor } = require('./providerHealthMonitor')
|
||||
stopProviderHealthMonitor()
|
||||
const { stopTaskEventListener, stopPollerRecovery } = require('./aiTaskWorker')
|
||||
stopPollerRecovery()
|
||||
void stopTaskEventListener()
|
||||
const { pool } = require('./db')
|
||||
pool.end().then(() => {
|
||||
console.log('[shutdown] Database pool closed')
|
||||
|
||||
+3
-2
@@ -284,7 +284,7 @@ async function releaseLeaseInternal(leaseToken, user, options = {}) {
|
||||
const { rows } = await client.query(
|
||||
`
|
||||
WITH candidate AS (
|
||||
SELECT l.id, l.key_id, k.provider
|
||||
SELECT l.id, l.key_id, l.user_id, l.enterprise_id, k.provider
|
||||
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
|
||||
@@ -298,6 +298,7 @@ async function releaseLeaseInternal(leaseToken, user, options = {}) {
|
||||
RETURNING id, key_id
|
||||
)
|
||||
SELECT r.id, r.key_id, c.provider
|
||||
, c.user_id, c.enterprise_id
|
||||
FROM released r
|
||||
JOIN candidate c ON c.key_id = r.key_id
|
||||
`,
|
||||
@@ -339,7 +340,7 @@ async function releaseLeaseInternal(leaseToken, user, options = {}) {
|
||||
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)
|
||||
`,
|
||||
[userId, enterpriseId, lease.key_id, lease.key_id, "release"],
|
||||
[userId ?? lease.user_id, enterpriseId ?? lease.enterprise_id, lease.key_id, lease.key_id, "release"],
|
||||
);
|
||||
|
||||
return {
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
const fs = require("node:fs");
|
||||
const { AlipaySdk } = require("alipay-sdk");
|
||||
const { pool, withTransaction } = require("./db");
|
||||
const { cashCentsToCreditUnits, creditBalance, creditUserBalance, activatePackage, formatCreditsFromCents } = require("./billing");
|
||||
const { creditBalance, creditUserBalance, activatePackage } = require("./billing");
|
||||
|
||||
let alipayInstance = null;
|
||||
|
||||
@@ -130,19 +130,17 @@ async function handlePaymentSuccess(orderNo, tradeNo) {
|
||||
);
|
||||
|
||||
if (order.type === "personal_recharge" && order.user_id) {
|
||||
const creditUnits = cashCentsToCreditUnits(order.amount_cents);
|
||||
await creditUserBalance(
|
||||
order.user_id,
|
||||
creditUnits,
|
||||
`支付宝充值 ${formatCreditsFromCents(creditUnits)} 积分`,
|
||||
order.amount_cents,
|
||||
`支付宝充值 ${Math.floor(order.amount_cents / 100)} 积分`,
|
||||
orderNo,
|
||||
);
|
||||
} else if (order.type === "recharge") {
|
||||
const creditUnits = cashCentsToCreditUnits(order.amount_cents);
|
||||
await creditBalance(
|
||||
order.enterprise_id,
|
||||
creditUnits,
|
||||
`支付宝充值 ${formatCreditsFromCents(creditUnits)} 积分`,
|
||||
order.amount_cents,
|
||||
`支付宝充值 ${Math.floor(order.amount_cents / 100)} 积分`,
|
||||
orderNo,
|
||||
);
|
||||
} else if (order.type === "package" && order.package_id) {
|
||||
|
||||
@@ -2,7 +2,7 @@ const _crypto = require("node:crypto");
|
||||
const fs = require("node:fs");
|
||||
const WxPay = require("wechatpay-node-v3");
|
||||
const { pool, withTransaction } = require("./db");
|
||||
const { cashCentsToCreditUnits, creditBalance, creditUserBalance, activatePackage, formatCreditsFromCents } = require("./billing");
|
||||
const { creditBalance, creditUserBalance, activatePackage } = require("./billing");
|
||||
|
||||
let wxPayInstance = null;
|
||||
|
||||
@@ -140,19 +140,17 @@ async function handlePaymentSuccess(orderNo, transactionId) {
|
||||
);
|
||||
|
||||
if (order.type === "personal_recharge" && order.user_id) {
|
||||
const creditUnits = cashCentsToCreditUnits(order.amount_cents);
|
||||
await creditUserBalance(
|
||||
order.user_id,
|
||||
creditUnits,
|
||||
`微信充值 ${formatCreditsFromCents(creditUnits)} 积分`,
|
||||
order.amount_cents,
|
||||
`微信充值 ${Math.floor(order.amount_cents / 100)} 积分`,
|
||||
orderNo,
|
||||
);
|
||||
} else if (order.type === "recharge") {
|
||||
const creditUnits = cashCentsToCreditUnits(order.amount_cents);
|
||||
await creditBalance(
|
||||
order.enterprise_id,
|
||||
creditUnits,
|
||||
`微信充值 ${formatCreditsFromCents(creditUnits)} 积分`,
|
||||
order.amount_cents,
|
||||
`微信充值 ${Math.floor(order.amount_cents / 100)} 积分`,
|
||||
orderNo,
|
||||
);
|
||||
} else if (order.type === "package" && order.package_id) {
|
||||
|
||||
+1
-1
@@ -176,7 +176,7 @@ async function getAverageCostCents(provider) {
|
||||
const { rows } = await pool.query(
|
||||
`
|
||||
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
|
||||
END)::numeric) AS INTEGER) AS avg_cents
|
||||
FROM api_call_logs
|
||||
|
||||
+6
-21
@@ -6,8 +6,6 @@ const {
|
||||
listModelPrices,
|
||||
loadPriceCache,
|
||||
creditUserBalance,
|
||||
creditsToCreditUnits,
|
||||
formatCreditsFromCents,
|
||||
pool,
|
||||
validateUsername,
|
||||
validatePassword,
|
||||
@@ -158,18 +156,14 @@ function registerAdminRoutes(router) {
|
||||
|
||||
router.post("/admin/users/:id/credit", requireAuth, requireAdmin, async (req, res) => {
|
||||
const targetUserId = Number(req.params.id);
|
||||
const { amountCredits, amountCents } = req.body;
|
||||
const creditUnits =
|
||||
amountCredits !== undefined && amountCredits !== null && amountCredits !== ""
|
||||
? creditsToCreditUnits(amountCredits)
|
||||
: Number(amountCents);
|
||||
if (!creditUnits || creditUnits <= 0) return res.status(400).json({ error: "积分必须大于 0" });
|
||||
const { amountCents } = req.body;
|
||||
if (!amountCents || amountCents <= 0) return res.status(400).json({ error: "积分必须大于 0" });
|
||||
|
||||
try {
|
||||
const newBalance = await creditUserBalance(
|
||||
targetUserId,
|
||||
creditUnits,
|
||||
`管理员 ${req.user.username} 发放 ${formatCreditsFromCents(creditUnits)} 积分`,
|
||||
amountCents,
|
||||
`管理员 ${req.user.username} 发放 ${Math.floor(amountCents / 100)} 积分`,
|
||||
);
|
||||
res.json({ success: true, newBalanceCents: newBalance });
|
||||
} catch (err) {
|
||||
@@ -553,8 +547,6 @@ function registerAdminRoutes(router) {
|
||||
name,
|
||||
description = "",
|
||||
priceCents,
|
||||
credits,
|
||||
amountCredits,
|
||||
creditsCents = 0,
|
||||
imageQuota = 0,
|
||||
videoQuota = 0,
|
||||
@@ -580,9 +572,7 @@ function registerAdminRoutes(router) {
|
||||
name,
|
||||
description,
|
||||
Number(priceCents),
|
||||
credits !== undefined || amountCredits !== undefined
|
||||
? creditsToCreditUnits(credits ?? amountCredits)
|
||||
: Number(creditsCents || 0),
|
||||
Number(creditsCents || 0),
|
||||
Number(imageQuota || 0),
|
||||
Number(videoQuota || 0),
|
||||
Number(textQuota || 0),
|
||||
@@ -609,8 +599,6 @@ function registerAdminRoutes(router) {
|
||||
name,
|
||||
description,
|
||||
priceCents,
|
||||
credits,
|
||||
amountCredits,
|
||||
creditsCents,
|
||||
imageQuota,
|
||||
videoQuota,
|
||||
@@ -635,10 +623,7 @@ function registerAdminRoutes(router) {
|
||||
updates.push(`price_cents = $${idx++}`);
|
||||
params.push(Number(priceCents));
|
||||
}
|
||||
if (credits !== undefined || amountCredits !== undefined) {
|
||||
updates.push(`credits_cents = $${idx++}`);
|
||||
params.push(creditsToCreditUnits(credits ?? amountCredits));
|
||||
} else if (creditsCents !== undefined) {
|
||||
if (creditsCents !== undefined) {
|
||||
updates.push(`credits_cents = $${idx++}`);
|
||||
params.push(Number(creditsCents));
|
||||
}
|
||||
|
||||
+13
-39
@@ -16,6 +16,7 @@ const {
|
||||
} = require("../enterpriseVideoBilling");
|
||||
const {
|
||||
startPolling,
|
||||
cancelTask,
|
||||
updateTaskInDb,
|
||||
extractProviderTaskId,
|
||||
extractImageUrl,
|
||||
@@ -97,18 +98,6 @@ function clampImageQualityForModel(model, quality) {
|
||||
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) {
|
||||
const normalized = normalizeQuality(quality, "1K");
|
||||
const maxQuality = GRSAI_IMAGE_MAX_QUALITY.get(String(model || "").toLowerCase());
|
||||
@@ -346,25 +335,18 @@ async function assertUserGenerationConcurrencyLimit(userId, client = pool) {
|
||||
[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(
|
||||
"SELECT COUNT(*)::int AS active_count FROM generation_tasks WHERE user_id = $1 AND status IN ('pending', 'running')",
|
||||
[userId],
|
||||
);
|
||||
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.code = "GENERATION_CONCURRENCY_LIMIT";
|
||||
error.activeCount = activeCount;
|
||||
error.maxActiveTasks = concurrencyLimit;
|
||||
error.maxActiveTasks = MAX_USER_ACTIVE_GENERATION_TASKS;
|
||||
throw error;
|
||||
}
|
||||
|
||||
@@ -488,22 +470,17 @@ function buildDashscopeImageBody(params) {
|
||||
if (url) content.push({ image: url });
|
||||
}
|
||||
content.push({ text: params.prompt });
|
||||
const quality = resolveDashscopeImageQuality(params);
|
||||
const gridCount = getGridCount(params.gridMode);
|
||||
const parameters = {
|
||||
size: mapAspectRatioToDashscopeSize(params.ratio, quality),
|
||||
n: gridCount,
|
||||
watermark: false,
|
||||
};
|
||||
if (gridCount > 1) {
|
||||
parameters.enable_sequential = true;
|
||||
}
|
||||
const quality = clampImageQualityForModel(params.model, params.quality);
|
||||
return {
|
||||
model: params.model,
|
||||
input: {
|
||||
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,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -1794,12 +1771,9 @@ function registerAiRoutes(router) {
|
||||
if (!Number.isFinite(taskId)) return res.status(400).json({ error: "Invalid task id" });
|
||||
|
||||
try {
|
||||
const { rows } = await pool.query(
|
||||
"UPDATE generation_tasks SET status = 'cancelled', updated_at = NOW() WHERE id = $1 AND user_id = $2 AND status IN ('pending', 'running') RETURNING id, status",
|
||||
[taskId, req.user.id],
|
||||
);
|
||||
if (rows.length === 0) return res.status(404).json({ error: "Task not found or not in active state" });
|
||||
res.json({ id: rows[0].id, status: rows[0].status });
|
||||
const task = await cancelTask(taskId, req.user.id);
|
||||
if (!task) return res.status(404).json({ error: "Task not found or not in active state" });
|
||||
res.json({ id: task.id, status: task.status });
|
||||
} catch (err) {
|
||||
console.error("[ai/task-cancel] error:", err.message);
|
||||
res.status(500).json({ error: "取消任务失败" });
|
||||
|
||||
@@ -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 };
|
||||
@@ -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;
|
||||
+14
-113
@@ -32,8 +32,6 @@ const {
|
||||
getUserEnterpriseId,
|
||||
getEnterpriseName,
|
||||
preauthorizeCall,
|
||||
creditsToCreditUnits,
|
||||
formatCreditsFromCents,
|
||||
} = require("../billing");
|
||||
const wechatPay = require("../paymentWechat");
|
||||
const alipay = require("../paymentAlipay");
|
||||
@@ -212,123 +210,28 @@ function hashEmailCode(email, code) {
|
||||
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, "&")
|
||||
.replace(/</g, "<")
|
||||
.replace(/>/g, ">")
|
||||
.replace(/"/g, """);
|
||||
}
|
||||
|
||||
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) {
|
||||
const provider = String(process.env.EMAIL_PROVIDER || "mock").trim().toLowerCase();
|
||||
|
||||
if (provider === "smtp") {
|
||||
const nodemailer = require("nodemailer");
|
||||
const smtpOptions = buildSmtpTransportOptions("SYSTEM");
|
||||
const transporter = nodemailer.createTransport(smtpOptions);
|
||||
const fromAddress = process.env.SYSTEM_SMTP_FROM || process.env.SMTP_FROM || smtpOptions.auth.user;
|
||||
const fromName = process.env.SYSTEM_SMTP_FROM_NAME || process.env.SMTP_FROM_NAME || "万物可爱";
|
||||
const transporter = nodemailer.createTransport({
|
||||
host: process.env.SMTP_HOST,
|
||||
port: Number(process.env.SMTP_PORT) || 587,
|
||||
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({
|
||||
from: formatEmailAddress(fromAddress, fromName),
|
||||
from: process.env.SMTP_FROM || process.env.SMTP_USER,
|
||||
to: email,
|
||||
subject: content.subject,
|
||||
text: content.text,
|
||||
html: content.html,
|
||||
subject: "[OmniAI] \u90ae\u7bb1\u9a8c\u8bc1\u7801",
|
||||
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: "<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" };
|
||||
}
|
||||
@@ -890,8 +793,6 @@ module.exports = {
|
||||
getUserEnterpriseId,
|
||||
getEnterpriseName,
|
||||
preauthorizeCall,
|
||||
creditsToCreditUnits,
|
||||
formatCreditsFromCents,
|
||||
wechatPay,
|
||||
alipay,
|
||||
crypto,
|
||||
|
||||
@@ -2,7 +2,6 @@ const {
|
||||
requireAuth,
|
||||
requireEnterpriseAdmin,
|
||||
distributeCredits,
|
||||
creditsToCreditUnits,
|
||||
getEnterpriseFinancials,
|
||||
getEnterpriseName,
|
||||
pool,
|
||||
@@ -303,33 +302,25 @@ function registerEnterpriseRoutes(router) {
|
||||
});
|
||||
|
||||
router.post("/enterprise/distribute", requireAuth, requireEnterpriseAdmin, async (req, res) => {
|
||||
const { userId, amountCredits, amountCents, distributions } = req.body;
|
||||
const { userId, amountCents, distributions } = req.body;
|
||||
|
||||
try {
|
||||
if (distributions && Array.isArray(distributions)) {
|
||||
for (const d of distributions) {
|
||||
const creditUnits =
|
||||
d.amountCredits !== undefined && d.amountCredits !== null && d.amountCredits !== ""
|
||||
? creditsToCreditUnits(d.amountCredits)
|
||||
: Number(d.amountCents);
|
||||
if (!d.userId || !creditUnits || creditUnits <= 0) {
|
||||
if (!d.userId || !d.amountCents || d.amountCents <= 0) {
|
||||
return res
|
||||
.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 });
|
||||
} else if (userId && (amountCredits || amountCents)) {
|
||||
const creditUnits =
|
||||
amountCredits !== undefined && amountCredits !== null && amountCredits !== ""
|
||||
? creditsToCreditUnits(amountCredits)
|
||||
: Number(amountCents);
|
||||
if (!creditUnits || creditUnits <= 0) return res.status(400).json({ error: "分发积分必须大于0" });
|
||||
} else if (userId && amountCents) {
|
||||
if (amountCents <= 0) return res.status(400).json({ error: "分发积分必须大于0" });
|
||||
const result = await distributeCredits(
|
||||
req.user.enterpriseId,
|
||||
userId,
|
||||
creditUnits,
|
||||
amountCents,
|
||||
req.user.id,
|
||||
);
|
||||
res.json({ success: true, ...result });
|
||||
@@ -358,7 +349,7 @@ function registerEnterpriseRoutes(router) {
|
||||
u.username,
|
||||
u.balance_cents AS current_balance_cents,
|
||||
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
|
||||
FROM users u
|
||||
LEFT JOIN api_call_logs acl ON acl.user_id = u.id AND acl.status = 'success'
|
||||
|
||||
@@ -17,11 +17,9 @@ const { registerConversationRoutes } = require('./conversations')
|
||||
const { registerReportRoutes } = require('./reports')
|
||||
const { registerAssetRoutes } = require('./assets')
|
||||
const { registerNotificationRoutes } = require('./notifications')
|
||||
const { registerBetaApplicationRoutes } = require('./betaApplications')
|
||||
const { registerDraftRoutes } = require('./drafts');
|
||||
const { registerFileExtractRoutes } = require('./fileExtract');
|
||||
const mountClientErrorRoutes = require('./clientErrors')
|
||||
const bugFeedbackRouter = require("./bugFeedback")
|
||||
|
||||
const router = express.Router()
|
||||
|
||||
@@ -50,10 +48,8 @@ registerConversationRoutes(router)
|
||||
registerReportRoutes(router)
|
||||
registerAssetRoutes(router)
|
||||
registerNotificationRoutes(router)
|
||||
registerBetaApplicationRoutes(router)
|
||||
registerDraftRoutes(router)
|
||||
registerFileExtractRoutes(router)
|
||||
router.use(bugFeedbackRouter)
|
||||
registerHealthRoutes(router)
|
||||
|
||||
module.exports = router
|
||||
|
||||
+6
-6
@@ -136,8 +136,8 @@ function registerUserRoutes(router) {
|
||||
CASE
|
||||
WHEN billing_refunded = 1 THEN 0
|
||||
WHEN cost_cents > 0 THEN cost_cents
|
||||
WHEN status = 'completed' AND type = 'image' THEN 2000
|
||||
WHEN status = 'completed' AND type = 'video' THEN 50000
|
||||
WHEN status = 'completed' AND type = 'image' THEN 20
|
||||
WHEN status = 'completed' AND type = 'video' THEN 500
|
||||
ELSE 0
|
||||
END
|
||||
), 0) AS used_cents
|
||||
@@ -162,7 +162,7 @@ function registerUserRoutes(router) {
|
||||
resolution = params.resolution || params.quality || params.ratio || null;
|
||||
if (row.status === "completed") {
|
||||
if (row.type === "image") {
|
||||
estimatedCents = 2000;
|
||||
estimatedCents = 20;
|
||||
} else if (row.type === "video") {
|
||||
const dur = params.duration || 5;
|
||||
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("animate-mix") || model.includes("s2v")) rate = res === "720P" ? 0.6 : 1;
|
||||
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 {
|
||||
@@ -209,8 +209,8 @@ function registerUserRoutes(router) {
|
||||
CASE
|
||||
WHEN billing_refunded = 1 THEN 0
|
||||
WHEN cost_cents > 0 THEN cost_cents
|
||||
WHEN status = 'completed' AND type = 'image' THEN 2000
|
||||
WHEN status = 'completed' AND type = 'video' THEN 50000
|
||||
WHEN status = 'completed' AND type = 'image' THEN 20
|
||||
WHEN status = 'completed' AND type = 'video' THEN 500
|
||||
ELSE 0
|
||||
END
|
||||
), 0) AS used_cents
|
||||
|
||||
Reference in New Issue
Block a user