xz
410 lines
18 KiB
JavaScript
410 lines
18 KiB
JavaScript
import { dbAll, dbGet, dbRun } from "./db.mjs";
|
|
|
|
const makeId = (prefix) => `${prefix}-${Date.now()}-${Math.random().toString(16).slice(2, 8)}`;
|
|
|
|
function parseJson(value, fallback) {
|
|
try {
|
|
return JSON.parse(value);
|
|
} catch {
|
|
return fallback;
|
|
}
|
|
}
|
|
|
|
function privateHost(hostname) {
|
|
const host = String(hostname || "").toLowerCase();
|
|
if (["localhost", "127.0.0.1", "::1"].includes(host) || host.endsWith(".local")) return true;
|
|
const octets = host.split(".").map(Number);
|
|
if (octets.length !== 4 || octets.some((value) => !Number.isInteger(value) || value < 0 || value > 255)) return false;
|
|
return octets[0] === 10 || octets[0] === 127 || (octets[0] === 172 && octets[1] >= 16 && octets[1] <= 31) || (octets[0] === 192 && octets[1] === 168);
|
|
}
|
|
|
|
async function fetchWithTimeout(url, options, timeoutMs = 8000) {
|
|
const controller = new AbortController();
|
|
const timer = setTimeout(() => controller.abort(), timeoutMs);
|
|
try {
|
|
return await fetch(url, { ...options, signal: controller.signal });
|
|
} finally {
|
|
clearTimeout(timer);
|
|
}
|
|
}
|
|
|
|
function eventMatches(channel, eventKey) {
|
|
const events = parseJson(channel.events_json, []);
|
|
return events.includes("*") || events.includes(eventKey);
|
|
}
|
|
|
|
function notificationError(status, code, message, details = {}) {
|
|
const error = new Error(message);
|
|
error.status = status;
|
|
error.code = code;
|
|
error.details = details;
|
|
return error;
|
|
}
|
|
|
|
const NOTIFICATION_PREFERENCE_CATALOG = [
|
|
{ category: "job", label: "生成任务", description: "本地任务完成、失败和重试状态" },
|
|
{ category: "review", label: "审片流程", description: "质检通过、驳回、修改和新评论" },
|
|
{ category: "delivery", label: "交付运营", description: "交付版本创建、批准和回滚" },
|
|
{ category: "task", label: "协作任务", description: "任务分派、状态变更和截止提醒" },
|
|
{ category: "access", label: "组织访问", description: "邀请、成员和权限相关事件" },
|
|
{ category: "billing", label: "用量与配额", description: "配额预警和成本边界" },
|
|
{ category: "system", label: "系统通知", description: "平台策略和系统范围变更" }
|
|
];
|
|
|
|
function notificationDefinition(eventKey, payload = {}) {
|
|
const kind = String(payload.kind || "生成任务");
|
|
const shot = payload.shotTitle || payload.shotId || "当前镜头";
|
|
const error = String(payload.error || "模型连接器返回了失败结果").slice(0, 180);
|
|
const lane = payload.lane || "连续性检查";
|
|
const version = payload.version || "交付版本";
|
|
const definitions = {
|
|
"job.completed": {
|
|
category: "job",
|
|
severity: "success",
|
|
title: `${kind}已完成`,
|
|
body: `${shot} 已完成本地执行，产出 ${Number(payload.artifactCount || 0)} 个可检查文件。`,
|
|
targetTab: "jobs"
|
|
},
|
|
"job.failed": {
|
|
category: "job",
|
|
severity: "error",
|
|
title: `${kind}生成失败`,
|
|
body: `${shot} 的本地任务失败：${error}`,
|
|
targetTab: "jobs"
|
|
},
|
|
"review.approved": {
|
|
category: "review",
|
|
severity: "success",
|
|
title: `${lane}已通过`,
|
|
body: `${shot} 的审片门已通过，可以继续下一步生产。`,
|
|
targetTab: "qa"
|
|
},
|
|
"review.changes_requested": {
|
|
category: "review",
|
|
severity: "warning",
|
|
title: `${lane}需要修改`,
|
|
body: `${shot} 的审片门要求补充证据或修改后再提交。`,
|
|
targetTab: "qa"
|
|
},
|
|
"review.rejected": {
|
|
category: "review",
|
|
severity: "error",
|
|
title: `${lane}已驳回`,
|
|
body: `${shot} 未通过审片，请查看阻断项并重新提交。`,
|
|
targetTab: "qa"
|
|
},
|
|
"review.comment": {
|
|
category: "review",
|
|
severity: "info",
|
|
title: "审片中心有新评论",
|
|
body: `${shot} 收到新的审片意见，请回到审片中心查看。`,
|
|
targetTab: "qa"
|
|
},
|
|
"delivery.created": {
|
|
category: "delivery",
|
|
severity: "info",
|
|
title: `${version} 已创建`,
|
|
body: "新的内部交付版本已建立，等待批次证据和审片结果。",
|
|
targetTab: "export"
|
|
},
|
|
"delivery.approved": {
|
|
category: "delivery",
|
|
severity: "success",
|
|
title: `${version} 已批准交付`,
|
|
body: "交付版本已通过所有阻断门，可以进入本地发布或导出流程。",
|
|
targetTab: "export"
|
|
},
|
|
"task.assigned": {
|
|
category: "task",
|
|
severity: "info",
|
|
title: "你收到一个协作任务",
|
|
body: `任务“${payload.taskTitle || "未命名任务"}”已分配给你，请在任务中心确认负责人和截止时间。`,
|
|
targetTab: "tasks"
|
|
},
|
|
"task.updated": {
|
|
category: "task",
|
|
severity: payload.status === "blocked" ? "warning" : payload.status === "done" ? "success" : "info",
|
|
title: "协作任务状态已更新",
|
|
body: `任务“${payload.taskTitle || "未命名任务"}”当前状态：${payload.status || "已更新"}。`,
|
|
targetTab: "tasks"
|
|
},
|
|
"task.commented": {
|
|
category: "task",
|
|
severity: "info",
|
|
title: "协作任务有新讨论",
|
|
body: `任务“${payload.taskTitle || "未命名任务"}”收到新的评论或 @提醒，请打开任务详情查看。`,
|
|
targetTab: "tasks"
|
|
},
|
|
"invitation.created": {
|
|
category: "access",
|
|
severity: "info",
|
|
title: "你收到新的组织邀请",
|
|
body: `${payload.organizationName || "一个生产组织"} 邀请你以 ${payload.roleName || payload.roleKey || "成员"} 身份加入。`,
|
|
targetTab: "account"
|
|
},
|
|
"system.changed": {
|
|
category: "system",
|
|
severity: "info",
|
|
title: "系统配置已更新",
|
|
body: payload.detail || "系统范围配置发生了变化，请确认当前生产策略。",
|
|
targetTab: payload.targetTab || "system-overview"
|
|
},
|
|
"quota.warning": {
|
|
category: "billing",
|
|
severity: "warning",
|
|
title: "组织配额即将达到上限",
|
|
body: payload.detail || "请联系组织管理员调整配额或清理不再需要的产物。",
|
|
targetTab: "admin-usage"
|
|
}
|
|
};
|
|
return definitions[eventKey] || {
|
|
category: String(payload.category || "system"),
|
|
severity: String(payload.severity || "info"),
|
|
title: String(payload.title || eventKey || "平台通知"),
|
|
body: String(payload.body || "生产平台收到一条新的事件通知。"),
|
|
targetTab: String(payload.targetTab || "creator-home")
|
|
};
|
|
}
|
|
|
|
function scopedRecipientIds(context, eventKey, payload = {}) {
|
|
const requested = [
|
|
...(Array.isArray(payload.recipientUserIds) ? payload.recipientUserIds : []),
|
|
...(payload.recipientUserId ? [payload.recipientUserId] : [])
|
|
].map((value) => String(value || "").trim()).filter(Boolean);
|
|
if (requested.length) {
|
|
const placeholders = requested.map(() => "?").join(",");
|
|
return dbAll(
|
|
`SELECT DISTINCT u.id
|
|
FROM users u
|
|
JOIN organization_members om ON om.user_id = u.id
|
|
WHERE u.status = 'active' AND om.organization_id = ? AND om.status = 'active'
|
|
AND u.id IN (${placeholders})`,
|
|
[context.organization.id, ...requested]
|
|
).map((row) => row.id);
|
|
}
|
|
|
|
if (!context?.organization?.id || !context?.user?.id) return [];
|
|
if (eventKey.startsWith("system.")) return [context.user.id];
|
|
if (eventKey === "invitation.created") return payload.recipientUserId ? [String(payload.recipientUserId)] : [context.user.id];
|
|
|
|
return dbAll(
|
|
`SELECT DISTINCT u.id
|
|
FROM users u
|
|
JOIN organization_members om ON om.user_id = u.id
|
|
LEFT JOIN workspace_members wm ON wm.user_id = u.id AND wm.workspace_id = ? AND wm.status = 'active'
|
|
LEFT JOIN project_members pm ON pm.user_id = u.id AND pm.project_id = ? AND pm.status = 'active'
|
|
WHERE u.status = 'active' AND om.organization_id = ? AND om.status = 'active'
|
|
AND (om.role_key IN ('org_owner', 'org_admin') OR wm.user_id IS NOT NULL OR pm.user_id IS NOT NULL)`,
|
|
[context.workspace?.id || "", context.project?.id || "", context.organization.id]
|
|
).map((row) => row.id);
|
|
}
|
|
|
|
function userNotificationPayload(row) {
|
|
return {
|
|
id: row.id,
|
|
category: row.category,
|
|
eventKey: row.event_key,
|
|
severity: row.severity,
|
|
title: row.title,
|
|
body: row.body,
|
|
targetTab: row.target_tab,
|
|
targetId: row.target_id || "",
|
|
metadata: parseJson(row.metadata_json, {}),
|
|
createdAt: row.created_at,
|
|
readAt: row.read_at,
|
|
read: Boolean(row.read_at),
|
|
organizationId: row.organization_id,
|
|
workspaceId: row.workspace_id,
|
|
projectId: row.project_id
|
|
};
|
|
}
|
|
|
|
function preferenceCatalogItem(row) {
|
|
const definition = NOTIFICATION_PREFERENCE_CATALOG.find((item) => item.category === row.category) || { category: row.category, label: row.category, description: "" };
|
|
return {
|
|
...definition,
|
|
enabled: row.in_app_enabled === undefined ? true : Boolean(row.in_app_enabled),
|
|
customized: row.in_app_enabled !== undefined
|
|
};
|
|
}
|
|
|
|
function inAppScope(context) {
|
|
if (!context?.user?.id || !context?.organization?.id) throw notificationError(401, "auth_required", "请先登录");
|
|
return {
|
|
where: "user_id = ? AND organization_id = ? AND (workspace_id IS NULL OR workspace_id = ?)",
|
|
params: [context.user.id, context.organization.id, context.workspace?.id || ""]
|
|
};
|
|
}
|
|
|
|
function createUserNotifications({ context, eventKey, payload = {}, definition }) {
|
|
const recipientIds = scopedRecipientIds(context, eventKey, payload);
|
|
if (!recipientIds.length) return [];
|
|
const timestamp = new Date().toISOString();
|
|
const metadata = { ...payload };
|
|
delete metadata.recipientUserId;
|
|
delete metadata.recipientUserIds;
|
|
const created = [];
|
|
for (const userId of recipientIds) {
|
|
const preference = dbGet("SELECT in_app_enabled FROM user_notification_preferences WHERE user_id = ? AND organization_id = ? AND category = ?", [userId, context.organization.id, definition.category]);
|
|
if (preference && !Boolean(preference.in_app_enabled)) continue;
|
|
const id = makeId("user-notification");
|
|
dbRun(
|
|
"INSERT INTO user_notifications(id, user_id, organization_id, workspace_id, project_id, category, event_key, severity, title, body, target_tab, target_id, metadata_json, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
[
|
|
id,
|
|
userId,
|
|
context.organization.id,
|
|
context.workspace?.id || null,
|
|
context.project?.id || null,
|
|
definition.category,
|
|
eventKey,
|
|
definition.severity,
|
|
definition.title,
|
|
definition.body,
|
|
definition.targetTab,
|
|
String(payload.targetId || payload.reviewId || payload.jobId || payload.deliveryId || ""),
|
|
JSON.stringify(metadata),
|
|
timestamp
|
|
]
|
|
);
|
|
created.push(dbGet("SELECT * FROM user_notifications WHERE id = ?", [id]));
|
|
}
|
|
return created;
|
|
}
|
|
|
|
export function listUserNotificationPreferences(context) {
|
|
const scope = inAppScope(context);
|
|
const rows = dbAll("SELECT category, in_app_enabled FROM user_notification_preferences WHERE user_id = ? AND organization_id = ?", [scope.params[0], scope.params[1]]);
|
|
const byCategory = new Map(rows.map((row) => [row.category, row]));
|
|
return {
|
|
preferences: NOTIFICATION_PREFERENCE_CATALOG.map((definition) => preferenceCatalogItem(byCategory.get(definition.category) || { category: definition.category })),
|
|
scope: { organizationId: context.organization.id },
|
|
generatedAt: new Date().toISOString()
|
|
};
|
|
}
|
|
|
|
export function updateUserNotificationPreference(context, category, enabled) {
|
|
inAppScope(context);
|
|
const definition = NOTIFICATION_PREFERENCE_CATALOG.find((item) => item.category === String(category || "").trim());
|
|
if (!definition) throw notificationError(400, "notification_category_invalid", "通知类别不存在", { category });
|
|
dbRun(
|
|
"INSERT INTO user_notification_preferences(user_id, organization_id, category, in_app_enabled, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(user_id, organization_id, category) DO UPDATE SET in_app_enabled = excluded.in_app_enabled, updated_at = excluded.updated_at",
|
|
[context.user.id, context.organization.id, definition.category, enabled ? 1 : 0, new Date().toISOString()]
|
|
);
|
|
return listUserNotificationPreferences(context);
|
|
}
|
|
|
|
export function listUserNotifications(context, { limit = 60, unreadOnly = false } = {}) {
|
|
const scope = inAppScope(context);
|
|
const normalizedLimit = Math.max(1, Math.min(200, Number(limit || 60)));
|
|
const unreadClause = unreadOnly ? " AND read_at IS NULL" : "";
|
|
const rows = dbAll(
|
|
`SELECT * FROM user_notifications
|
|
WHERE ${scope.where}${unreadClause}
|
|
ORDER BY CASE WHEN read_at IS NULL THEN 0 ELSE 1 END, created_at DESC
|
|
LIMIT ?`,
|
|
[...scope.params, normalizedLimit]
|
|
);
|
|
const unreadCount = Number(dbGet(`SELECT COUNT(*) AS count FROM user_notifications WHERE ${scope.where} AND read_at IS NULL`, scope.params)?.count || 0);
|
|
return {
|
|
notifications: rows.map(userNotificationPayload),
|
|
unreadCount,
|
|
total: rows.length,
|
|
scope: { organizationId: context.organization.id, workspaceId: context.workspace?.id || "" },
|
|
generatedAt: new Date().toISOString()
|
|
};
|
|
}
|
|
|
|
export function markUserNotificationRead(context, notificationId, read = true) {
|
|
const scope = inAppScope(context);
|
|
const current = dbGet(`SELECT * FROM user_notifications WHERE id = ? AND ${scope.where}`, [notificationId, ...scope.params]);
|
|
if (!current) throw notificationError(404, "notification_not_found", "通知不存在或不属于当前工作区");
|
|
dbRun(`UPDATE user_notifications SET read_at = ? WHERE id = ? AND ${scope.where}`, [read ? new Date().toISOString() : null, notificationId, ...scope.params]);
|
|
return listUserNotifications(context, { limit: 60 });
|
|
}
|
|
|
|
export function markAllUserNotificationsRead(context) {
|
|
const scope = inAppScope(context);
|
|
dbRun(`UPDATE user_notifications SET read_at = COALESCE(read_at, ?) WHERE ${scope.where} AND read_at IS NULL`, [new Date().toISOString(), ...scope.params]);
|
|
return listUserNotifications(context, { limit: 60 });
|
|
}
|
|
|
|
async function deliver(channel, eventKey, requestBody) {
|
|
const id = makeId("delivery");
|
|
const timestamp = new Date().toISOString();
|
|
dbRun("INSERT INTO notification_deliveries(id, channel_id, event_key, organization_id, workspace_id, project_id, status, attempt_count, request_json, created_at) VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?)", [id, channel.id, eventKey, requestBody.organizationId || null, requestBody.workspaceId || null, requestBody.projectId || null, JSON.stringify(requestBody), timestamp]);
|
|
let status = "delivered";
|
|
let response = { kind: channel.kind };
|
|
let errorMessage = "";
|
|
let httpStatus = null;
|
|
try {
|
|
if (channel.kind === "local-log") {
|
|
response = { logged: true, event: eventKey };
|
|
} else if (channel.kind === "webhook") {
|
|
const endpoint = new URL(String(channel.endpoint || ""));
|
|
if (endpoint.protocol !== "http:" || !privateHost(endpoint.hostname)) {
|
|
status = "blocked";
|
|
errorMessage = "本地策略只允许投递到 HTTP 私有地址";
|
|
} else {
|
|
const result = await fetchWithTimeout(endpoint.toString(), {
|
|
method: "POST",
|
|
headers: { "content-type": "application/json", "x-ai-drama-event": eventKey },
|
|
body: JSON.stringify(requestBody)
|
|
});
|
|
httpStatus = result.status;
|
|
const raw = await result.text();
|
|
response = { httpStatus, body: raw.slice(0, 1000) };
|
|
if (!result.ok) {
|
|
status = "failed";
|
|
errorMessage = `Webhook 返回 HTTP ${result.status}`;
|
|
}
|
|
}
|
|
} else {
|
|
status = "blocked";
|
|
errorMessage = `通知渠道类型 ${channel.kind} 暂不支持`;
|
|
}
|
|
} catch (error) {
|
|
status = "failed";
|
|
errorMessage = error.name === "AbortError" ? "通知投递超时" : String(error.message || error).slice(0, 500);
|
|
}
|
|
const deliveredAt = status === "delivered" ? new Date().toISOString() : null;
|
|
dbRun("UPDATE notification_deliveries SET status = ?, attempt_count = 1, response_json = ?, error_message = ?, delivered_at = ? WHERE id = ?", [status, JSON.stringify({ ...response, httpStatus }), errorMessage, deliveredAt, id]);
|
|
return dbGet("SELECT * FROM notification_deliveries WHERE id = ?", [id]);
|
|
}
|
|
|
|
export async function dispatchNotificationEvent({ context, eventKey, payload = {} }) {
|
|
const definition = notificationDefinition(eventKey, payload);
|
|
const userNotifications = createUserNotifications({ context, eventKey, payload, definition });
|
|
const channels = dbAll("SELECT * FROM notification_channels WHERE enabled = 1 ORDER BY created_at").filter((channel) => eventMatches(channel, eventKey));
|
|
const requestBody = {
|
|
schema: "ai-drama.notification.v1",
|
|
event: eventKey,
|
|
occurredAt: new Date().toISOString(),
|
|
actor: context?.user ? { id: context.user.id, name: context.user.display_name } : null,
|
|
organizationId: context?.organization?.id || null,
|
|
workspaceId: context?.workspace?.id || null,
|
|
projectId: context?.project?.id || null,
|
|
payload
|
|
};
|
|
const deliveries = [];
|
|
for (const channel of channels) deliveries.push(await deliver(channel, eventKey, requestBody));
|
|
return { deliveries, userNotifications: userNotifications.map(userNotificationPayload) };
|
|
}
|
|
|
|
export function notificationDeliveries(context, limit = 80) {
|
|
return dbAll(
|
|
`SELECT d.*, c.name AS channel_name, c.kind AS channel_kind
|
|
FROM notification_deliveries d
|
|
JOIN notification_channels c ON c.id = d.channel_id
|
|
WHERE d.organization_id = ?
|
|
AND (d.workspace_id IS NULL OR d.workspace_id = ?)
|
|
AND (d.project_id IS NULL OR d.project_id = ?)
|
|
ORDER BY d.created_at DESC LIMIT ?`,
|
|
[context.organization.id, context.workspace.id, context.project?.id || "", Math.max(1, Math.min(200, Number(limit || 80)))]
|
|
).map((row) => ({
|
|
...row,
|
|
request: parseJson(row.request_json, {}),
|
|
response: parseJson(row.response_json, {})
|
|
}));
|
|
}
|