实现客户端埋点上传清理与后台查询
新增客户端每十五分钟上传及入库确认后批次清理 新增私有事件表、原子幂等入库及鉴权查询接口 新增后台客户端埋点明细、筛选、分页与权限 补充定向测试、真实数据库联调脚本及工程验收文档
This commit is contained in:
@@ -7,6 +7,7 @@ import {
|
||||
getAdminFeatureGateConfig,
|
||||
getAdminUserDetail,
|
||||
importAdminAgcTemplates,
|
||||
listAdminAgcTrackingEvents,
|
||||
listAdminRechargeOrders,
|
||||
reconcileAdminUserConsumption,
|
||||
resolveAdminRechargeRefundManualReview,
|
||||
@@ -21,6 +22,30 @@ afterEach(() => {
|
||||
vi.unstubAllGlobals();
|
||||
});
|
||||
|
||||
test('客户端埋点查询传递筛选和游标并复用后台认证', async () => {
|
||||
const payload = { entries: [], nextCursor: null };
|
||||
const fetchMock = vi.fn().mockResolvedValue(
|
||||
new Response(JSON.stringify({ ok: true, data: payload }), {
|
||||
status: 200,
|
||||
}),
|
||||
);
|
||||
vi.stubGlobal('fetch', fetchMock);
|
||||
expect(
|
||||
await listAdminAgcTrackingEvents('token', {
|
||||
userId: 'user+1',
|
||||
projectId: 'project-1',
|
||||
cursor: 'page/2',
|
||||
limit: 50,
|
||||
}),
|
||||
).toEqual(payload);
|
||||
expect(fetchMock).toHaveBeenCalledWith(
|
||||
'/admin/api/agc/tracking-events?userId=user%2B1&projectId=project-1&cursor=page%2F2&limit=50',
|
||||
expect.objectContaining({
|
||||
headers: expect.objectContaining({ Authorization: 'Bearer token' }),
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
test('模板管理读取和更新复用认证封装，提交 revision 和封面但不提交 ZIP 或版本', async () => {
|
||||
const library = { revision: 'revision-new', writable: true, templates: [] };
|
||||
const fetchMock = vi.fn().mockImplementation(
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import type {
|
||||
AdminAccountListResponse,
|
||||
AdminAgcTemplateLibraryResponse,
|
||||
AdminAgcTrackingEventListResponse,
|
||||
AdminAgcTrackingEventQuery,
|
||||
AdminConfirmEditorShowcaseCampaignImageUploadRequest,
|
||||
AdminCreateAccountRequest,
|
||||
AdminCreateAccountResponse,
|
||||
@@ -406,6 +408,31 @@ export function listAdminTrackingEventKeys(token: string) {
|
||||
);
|
||||
}
|
||||
|
||||
export function listAdminAgcTrackingEvents(
|
||||
token: string,
|
||||
query: AdminAgcTrackingEventQuery = {},
|
||||
) {
|
||||
return request<AdminAgcTrackingEventListResponse>(
|
||||
`/admin/api/agc/tracking-events${buildQueryString((params) => {
|
||||
for (const key of [
|
||||
'userId',
|
||||
'projectId',
|
||||
'creativeTaskId',
|
||||
'agentRunId',
|
||||
'eventName',
|
||||
'clientVersion',
|
||||
'startTime',
|
||||
'endTime',
|
||||
'cursor',
|
||||
] as const) {
|
||||
appendQueryParam(params, key, query[key]);
|
||||
}
|
||||
appendNumericQueryParam(params, 'limit', query.limit);
|
||||
})}`,
|
||||
{ token },
|
||||
);
|
||||
}
|
||||
|
||||
export function listAdminErrorReports(
|
||||
token: string,
|
||||
query: {
|
||||
|
||||
@@ -820,6 +820,44 @@ export interface AdminTrackingEventListResponse {
|
||||
entries: AdminTrackingEventEntryPayload[];
|
||||
}
|
||||
|
||||
export interface AdminAgcTrackingEventQuery {
|
||||
userId?: string;
|
||||
projectId?: string;
|
||||
creativeTaskId?: string;
|
||||
agentRunId?: string;
|
||||
eventName?: string;
|
||||
clientVersion?: string;
|
||||
startTime?: string;
|
||||
endTime?: string;
|
||||
cursor?: string;
|
||||
limit?: number;
|
||||
}
|
||||
|
||||
export interface AdminAgcTrackingEventEntry {
|
||||
eventId: string;
|
||||
schemaVersion: number;
|
||||
eventName: string;
|
||||
eventTime: string;
|
||||
userId: string;
|
||||
editorSessionId: string;
|
||||
projectId: string | null;
|
||||
creativeTaskId: string | null;
|
||||
agentRunId: string | null;
|
||||
agentTurnId: string | null;
|
||||
status: string | null;
|
||||
errorCode: string | null;
|
||||
source: string;
|
||||
clientVersion: string;
|
||||
properties: Record<string, unknown>;
|
||||
batchId: string;
|
||||
receivedAt: string;
|
||||
}
|
||||
|
||||
export interface AdminAgcTrackingEventListResponse {
|
||||
entries: AdminAgcTrackingEventEntry[];
|
||||
nextCursor: string | null;
|
||||
}
|
||||
|
||||
export interface AdminTrackingEventKeyPayload {
|
||||
eventKey: string;
|
||||
eventTitle: string;
|
||||
|
||||
@@ -20,6 +20,7 @@ import {
|
||||
import { AdminAccountsPage } from '../pages/AdminAccountsPage';
|
||||
import { AdminAgcModelsPage } from '../pages/AdminAgcModelsPage';
|
||||
import { AdminAgcTemplatesPage } from '../pages/AdminAgcTemplatesPage';
|
||||
import { AdminAgcTrackingPage } from '../pages/AdminAgcTrackingPage';
|
||||
import { AdminDashboardPage } from '../pages/AdminDashboardPage';
|
||||
import { AdminDatabaseTablesPage } from '../pages/AdminDatabaseTablesPage';
|
||||
import { AdminDebugHttpPage } from '../pages/AdminDebugHttpPage';
|
||||
@@ -232,6 +233,12 @@ export function AdminApp() {
|
||||
onUnauthorized={handleUnauthorized}
|
||||
/>
|
||||
) : null}
|
||||
{activeRouteId === 'agc-tracking' ? (
|
||||
<AdminAgcTrackingPage
|
||||
token={token}
|
||||
onUnauthorized={handleUnauthorized}
|
||||
/>
|
||||
) : null}
|
||||
{activeRouteId === 'error-reports' ? (
|
||||
<AdminErrorReportsPage
|
||||
token={token}
|
||||
|
||||
@@ -39,6 +39,7 @@ const routeIcons = {
|
||||
tables: Database,
|
||||
debug: Bug,
|
||||
tracking: Table2,
|
||||
'agc-tracking': Table2,
|
||||
'error-reports': Bug,
|
||||
'gray-release': GitBranch,
|
||||
redeem: TicketPercent,
|
||||
|
||||
@@ -8,6 +8,22 @@ import {
|
||||
routeHash,
|
||||
} from './adminRoutes';
|
||||
|
||||
test('客户端埋点路由遵守成员页签权限', () => {
|
||||
expect(resolveAdminRoute('#agc-tracking')).toBe('agc-tracking');
|
||||
expect(
|
||||
getAccessibleAdminRoutes({
|
||||
accountRole: 'member',
|
||||
tabPermissions: ['agc-tracking'],
|
||||
}).map((route) => route.id),
|
||||
).toEqual(['agc-tracking']);
|
||||
expect(
|
||||
getAccessibleAdminRoutes({
|
||||
accountRole: 'member',
|
||||
tabPermissions: ['tracking'],
|
||||
}).some((route) => route.id === 'agc-tracking'),
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
test('后台默认进入 Dashboard', () => {
|
||||
expect(adminRoutes[0]).toEqual({
|
||||
id: 'dashboard',
|
||||
|
||||
@@ -5,6 +5,7 @@ export type AdminRouteId =
|
||||
| 'tables'
|
||||
| 'debug'
|
||||
| 'tracking'
|
||||
| 'agc-tracking'
|
||||
| 'error-reports'
|
||||
| 'gray-release'
|
||||
| 'redeem'
|
||||
@@ -40,6 +41,7 @@ export const adminRoutes: AdminRouteDefinition[] = [
|
||||
{ id: 'tables', label: '表查询', hash: '#tables' },
|
||||
{ id: 'debug', label: 'API 调试', hash: '#debug' },
|
||||
{ id: 'tracking', label: '埋点数据', hash: '#tracking' },
|
||||
{ id: 'agc-tracking', label: '客户端埋点', hash: '#agc-tracking' },
|
||||
{ id: 'error-reports', label: '错误报告', hash: '#error-reports' },
|
||||
{ id: 'gray-release', label: '灰度发布', hash: '#gray-release' },
|
||||
{ id: 'redeem', label: '兑换码', hash: '#redeem' },
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
// @vitest-environment jsdom
|
||||
import {
|
||||
cleanup,
|
||||
fireEvent,
|
||||
render,
|
||||
screen,
|
||||
waitFor,
|
||||
within,
|
||||
} from '@testing-library/react';
|
||||
import { afterEach, beforeEach, expect, test, vi } from 'vitest';
|
||||
|
||||
import {
|
||||
AdminApiError,
|
||||
listAdminAgcTrackingEvents,
|
||||
} from '../api/adminApiClient';
|
||||
import type { AdminAgcTrackingEventEntry } from '../api/adminApiTypes';
|
||||
import { AdminAgcTrackingPage } from './AdminAgcTrackingPage';
|
||||
|
||||
vi.mock('../api/adminApiClient', async (original) => ({
|
||||
...(await original<typeof import('../api/adminApiClient')>()),
|
||||
listAdminAgcTrackingEvents: vi.fn(),
|
||||
}));
|
||||
vi.mock('../components/AdminUserReferenceButton', () => ({
|
||||
AdminUserReferenceButton: () => null,
|
||||
}));
|
||||
|
||||
const entry: AdminAgcTrackingEventEntry = {
|
||||
eventId: 'event-1',
|
||||
schemaVersion: 1,
|
||||
eventName: 'project_save',
|
||||
eventTime: '2026-09-21T12:00:00.123Z',
|
||||
userId: 'user-1',
|
||||
editorSessionId: 'session-1',
|
||||
projectId: 'project-1',
|
||||
creativeTaskId: 'project-1',
|
||||
agentRunId: null,
|
||||
agentTurnId: null,
|
||||
status: 'success',
|
||||
errorCode: null,
|
||||
source: 'gui',
|
||||
clientVersion: '1.0',
|
||||
properties: { reason: 'manual' },
|
||||
batchId: 'batch-1',
|
||||
receivedAt: '2026-09-21T12:15:00.000Z',
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
vi.mocked(listAdminAgcTrackingEvents).mockReset();
|
||||
});
|
||||
afterEach(cleanup);
|
||||
|
||||
test('翻页沿用游标，筛选和刷新从第一页重新查询，详情保留 null 和事件属性', async () => {
|
||||
vi.mocked(listAdminAgcTrackingEvents)
|
||||
.mockResolvedValueOnce({ entries: [entry], nextCursor: 'cursor-page-2' })
|
||||
.mockResolvedValueOnce({
|
||||
entries: [{ ...entry, eventId: 'event-2' }],
|
||||
nextCursor: null,
|
||||
})
|
||||
.mockResolvedValue({ entries: [entry], nextCursor: 'cursor-new' });
|
||||
render(<AdminAgcTrackingPage token="admin-token" onUnauthorized={vi.fn()} />);
|
||||
await screen.findByText('保存项目', { selector: 'td' });
|
||||
expect(listAdminAgcTrackingEvents).toHaveBeenLastCalledWith('admin-token', {
|
||||
cursor: undefined,
|
||||
limit: 50,
|
||||
});
|
||||
fireEvent.click(screen.getByText('查看详情'));
|
||||
const dialog = screen.getByRole('dialog');
|
||||
expect(within(dialog).getAllByText('—').length).toBe(3);
|
||||
expect(within(dialog).getByText(/"reason": "manual"/)).toBeTruthy();
|
||||
expect(within(dialog).getByText(entry.eventTime)).toBeTruthy();
|
||||
fireEvent.click(within(dialog).getByLabelText('关闭详情'));
|
||||
fireEvent.click(screen.getByText('下一页'));
|
||||
await waitFor(() =>
|
||||
expect(listAdminAgcTrackingEvents).toHaveBeenLastCalledWith('admin-token', {
|
||||
cursor: 'cursor-page-2',
|
||||
limit: 50,
|
||||
}),
|
||||
);
|
||||
await waitFor(() =>
|
||||
expect(screen.getByText('查询').closest('button')?.disabled).toBe(false),
|
||||
);
|
||||
fireEvent.click(screen.getByText('上一页'));
|
||||
await screen.findByText('第 1 页');
|
||||
expect(listAdminAgcTrackingEvents).toHaveBeenCalledTimes(2);
|
||||
fireEvent.change(screen.getByLabelText('项目 ID'), {
|
||||
target: { value: 'project-2' },
|
||||
});
|
||||
fireEvent.click(screen.getByText('查询'));
|
||||
await waitFor(() =>
|
||||
expect(listAdminAgcTrackingEvents).toHaveBeenLastCalledWith(
|
||||
'admin-token',
|
||||
expect.objectContaining({
|
||||
projectId: 'project-2',
|
||||
cursor: undefined,
|
||||
limit: 50,
|
||||
}),
|
||||
),
|
||||
);
|
||||
expect(screen.getByText('第 1 页')).toBeTruthy();
|
||||
await waitFor(() =>
|
||||
expect(screen.getByText('刷新').closest('button')?.disabled).toBe(false),
|
||||
);
|
||||
fireEvent.click(screen.getByText('刷新'));
|
||||
await waitFor(() =>
|
||||
expect(listAdminAgcTrackingEvents).toHaveBeenCalledTimes(4),
|
||||
);
|
||||
});
|
||||
|
||||
test('空结果、权限失败与登录失效沿用后台反馈', async () => {
|
||||
const unauthorized = vi.fn();
|
||||
vi.mocked(listAdminAgcTrackingEvents)
|
||||
.mockResolvedValueOnce({ entries: [], nextCursor: null })
|
||||
.mockRejectedValueOnce(
|
||||
new AdminApiError({ status: 403, message: '无权访问客户端埋点' }),
|
||||
)
|
||||
.mockRejectedValueOnce(
|
||||
new AdminApiError({ status: 401, message: '登录失效' }),
|
||||
);
|
||||
render(<AdminAgcTrackingPage token="token" onUnauthorized={unauthorized} />);
|
||||
await screen.findByText('暂无客户端埋点数据');
|
||||
fireEvent.click(screen.getByText('刷新'));
|
||||
expect((await screen.findByRole('alert')).textContent).toContain(
|
||||
'无权访问客户端埋点',
|
||||
);
|
||||
fireEvent.click(screen.getByText('刷新'));
|
||||
await waitFor(() =>
|
||||
expect(unauthorized).toHaveBeenCalledWith('登录状态已失效'),
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,379 @@
|
||||
import { Modal } from '@genarrative/shared/components';
|
||||
import { FormEvent, useEffect, useRef, useState } from 'react';
|
||||
|
||||
import { listAdminAgcTrackingEvents } from '../api/adminApiClient';
|
||||
import type {
|
||||
AdminAgcTrackingEventEntry,
|
||||
AdminAgcTrackingEventListResponse,
|
||||
AdminAgcTrackingEventQuery,
|
||||
} from '../api/adminApiTypes';
|
||||
import { AdminUserReferenceButton } from '../components/AdminUserReferenceButton';
|
||||
import { handlePageError } from './pageUtils';
|
||||
|
||||
const eventLabels: Record<string, string> = {
|
||||
editor_session_start: '编辑器会话开始',
|
||||
editor_session_end: '编辑器会话结束',
|
||||
editor_focus_start: '编辑器获得焦点',
|
||||
editor_focus_end: '编辑器失去焦点',
|
||||
project_create_success: '项目创建成功',
|
||||
project_open: '打开项目',
|
||||
creative_task_submit: '首次提交创作目标',
|
||||
agent_run_completed: 'Agent 运行完成',
|
||||
agent_run_failed: 'Agent 运行失败',
|
||||
project_revision_created: '项目产生修改',
|
||||
preview_ready: '预览就绪',
|
||||
project_save: '保存项目',
|
||||
};
|
||||
|
||||
const fieldLabels: Record<keyof AdminAgcTrackingEventEntry, string> = {
|
||||
eventId: '事件 ID',
|
||||
schemaVersion: '事件版本',
|
||||
eventName: '事件类型',
|
||||
eventTime: '发生时间',
|
||||
userId: '用户 ID',
|
||||
editorSessionId: '编辑器会话 ID',
|
||||
projectId: '项目 ID',
|
||||
creativeTaskId: '创作目标 ID',
|
||||
agentRunId: 'Agent run ID',
|
||||
agentTurnId: 'Agent turn ID',
|
||||
status: '结果',
|
||||
errorCode: '错误码',
|
||||
source: '来源',
|
||||
clientVersion: '客户端版本',
|
||||
properties: '事件属性',
|
||||
batchId: '批次 ID',
|
||||
receivedAt: '入库时间',
|
||||
};
|
||||
|
||||
function formatTime(value: string) {
|
||||
const date = new Date(value);
|
||||
return Number.isNaN(date.getTime()) ? value : date.toLocaleString('zh-CN');
|
||||
}
|
||||
|
||||
export function AdminAgcTrackingPage({
|
||||
token,
|
||||
onUnauthorized,
|
||||
}: {
|
||||
token: string;
|
||||
onUnauthorized: (message?: string) => void;
|
||||
}) {
|
||||
const [filters, setFilters] = useState({
|
||||
userId: '',
|
||||
projectId: '',
|
||||
eventName: '',
|
||||
startTime: '',
|
||||
endTime: '',
|
||||
});
|
||||
const [query, setQuery] = useState<AdminAgcTrackingEventQuery>({});
|
||||
const [cursors, setCursors] = useState<Array<string | undefined>>([
|
||||
undefined,
|
||||
]);
|
||||
const [page, setPage] = useState(0);
|
||||
const [refresh, setRefresh] = useState(0);
|
||||
const [entries, setEntries] = useState<AdminAgcTrackingEventEntry[]>([]);
|
||||
const [nextCursor, setNextCursor] = useState<string | null>(null);
|
||||
const [loading, setLoading] = useState(true);
|
||||
const [error, setError] = useState('');
|
||||
const [detail, setDetail] = useState<AdminAgcTrackingEventEntry | null>(null);
|
||||
const cursor = cursors[page];
|
||||
// 首页没有入站游标，返回首页时保留原快照，刷新才获取新数据。
|
||||
const firstPage = useRef<{
|
||||
token: string;
|
||||
query: AdminAgcTrackingEventQuery;
|
||||
response: AdminAgcTrackingEventListResponse;
|
||||
} | null>(null);
|
||||
|
||||
useEffect(() => {
|
||||
const saved = firstPage.current;
|
||||
if (!cursor && saved?.token === token && saved.query === query) {
|
||||
setEntries(saved.response.entries);
|
||||
setNextCursor(saved.response.nextCursor);
|
||||
setError('');
|
||||
setLoading(false);
|
||||
return;
|
||||
}
|
||||
let active = true;
|
||||
setLoading(true);
|
||||
setError('');
|
||||
setEntries([]);
|
||||
setNextCursor(null);
|
||||
void listAdminAgcTrackingEvents(token, { ...query, cursor, limit: 50 })
|
||||
.then((response) => {
|
||||
if (!active) return;
|
||||
if (!cursor) firstPage.current = { token, query, response };
|
||||
setEntries(response.entries);
|
||||
setNextCursor(response.nextCursor);
|
||||
})
|
||||
.catch((failure: unknown) => {
|
||||
if (active) handlePageError(failure, onUnauthorized, setError);
|
||||
})
|
||||
.finally(() => {
|
||||
if (active) setLoading(false);
|
||||
});
|
||||
return () => {
|
||||
active = false;
|
||||
};
|
||||
}, [token, query, cursor, refresh, onUnauthorized]);
|
||||
|
||||
function resetPages() {
|
||||
firstPage.current = null;
|
||||
setCursors([undefined]);
|
||||
setPage(0);
|
||||
setDetail(null);
|
||||
}
|
||||
|
||||
function search(event: FormEvent<HTMLFormElement>) {
|
||||
event.preventDefault();
|
||||
if (
|
||||
filters.startTime &&
|
||||
filters.endTime &&
|
||||
new Date(filters.startTime) >= new Date(filters.endTime)
|
||||
) {
|
||||
setError('结束时间必须晚于开始时间');
|
||||
return;
|
||||
}
|
||||
resetPages();
|
||||
setQuery({
|
||||
userId: filters.userId.trim(),
|
||||
projectId: filters.projectId.trim(),
|
||||
eventName: filters.eventName,
|
||||
startTime: filters.startTime
|
||||
? new Date(filters.startTime).toISOString()
|
||||
: undefined,
|
||||
endTime: filters.endTime
|
||||
? new Date(filters.endTime).toISOString()
|
||||
: undefined,
|
||||
});
|
||||
}
|
||||
|
||||
function related(
|
||||
key: 'creativeTaskId' | 'agentRunId' | 'clientVersion',
|
||||
value: string,
|
||||
) {
|
||||
resetPages();
|
||||
setFilters({
|
||||
userId: '',
|
||||
projectId: '',
|
||||
eventName: '',
|
||||
startTime: '',
|
||||
endTime: '',
|
||||
});
|
||||
setQuery({ [key]: value });
|
||||
}
|
||||
|
||||
return (
|
||||
<section className="admin-page admin-page-wide">
|
||||
<div className="admin-page-heading">
|
||||
<div>
|
||||
<h2>客户端埋点</h2>
|
||||
<p>按入库时间倒序展示</p>
|
||||
</div>
|
||||
<button
|
||||
type="button"
|
||||
className="admin-secondary-button"
|
||||
disabled={loading}
|
||||
onClick={() => {
|
||||
resetPages();
|
||||
setRefresh((value) => value + 1);
|
||||
}}
|
||||
>
|
||||
刷新
|
||||
</button>
|
||||
</div>
|
||||
<form className="admin-panel admin-form" onSubmit={search}>
|
||||
<div className="admin-filter-grid">
|
||||
{(['userId', 'projectId'] as const).map((key) => (
|
||||
<label key={key} className="admin-field">
|
||||
<span>{fieldLabels[key]}</span>
|
||||
<input
|
||||
value={filters[key]}
|
||||
onChange={(event) =>
|
||||
setFilters({ ...filters, [key]: event.target.value })
|
||||
}
|
||||
/>
|
||||
</label>
|
||||
))}
|
||||
<label className="admin-field">
|
||||
<span>事件类型</span>
|
||||
<select
|
||||
value={filters.eventName}
|
||||
onChange={(event) =>
|
||||
setFilters({ ...filters, eventName: event.target.value })
|
||||
}
|
||||
>
|
||||
<option value="">全部</option>
|
||||
{Object.entries(eventLabels).map(([value, label]) => (
|
||||
<option key={value} value={value}>
|
||||
{label}
|
||||
</option>
|
||||
))}
|
||||
</select>
|
||||
</label>
|
||||
{(['startTime', 'endTime'] as const).map((key) => (
|
||||
<label key={key} className="admin-field">
|
||||
<span>
|
||||
{key === 'startTime'
|
||||
? '发生时间起点（含）'
|
||||
: '发生时间终点（不含）'}
|
||||
</span>
|
||||
<input
|
||||
type="datetime-local"
|
||||
value={filters[key]}
|
||||
onChange={(event) =>
|
||||
setFilters({ ...filters, [key]: event.target.value })
|
||||
}
|
||||
/>
|
||||
</label>
|
||||
))}
|
||||
</div>
|
||||
<div className="admin-action-row">
|
||||
<button
|
||||
type="submit"
|
||||
className="admin-primary-button"
|
||||
disabled={loading}
|
||||
>
|
||||
查询
|
||||
</button>
|
||||
</div>
|
||||
{(['creativeTaskId', 'agentRunId', 'clientVersion'] as const)
|
||||
.filter((key) => query[key])
|
||||
.map((key) => (
|
||||
<p key={key}>
|
||||
{fieldLabels[key]}：{query[key]}（重新查询可清除）
|
||||
</p>
|
||||
))}
|
||||
</form>
|
||||
{error ? (
|
||||
<p role="alert" className="admin-error-message">
|
||||
{error}
|
||||
</p>
|
||||
) : null}
|
||||
<div className="admin-panel">
|
||||
<div className="admin-table-wrap">
|
||||
<table className="admin-table">
|
||||
<thead>
|
||||
<tr>
|
||||
{[
|
||||
'入库时间',
|
||||
'发生时间',
|
||||
'用户',
|
||||
'事件名称',
|
||||
'项目',
|
||||
'来源',
|
||||
'结果',
|
||||
'客户端版本',
|
||||
'详情',
|
||||
].map((label) => (
|
||||
<th key={label}>{label}</th>
|
||||
))}
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{entries.map((entry) => (
|
||||
<tr key={entry.eventId}>
|
||||
<td>{formatTime(entry.receivedAt)}</td>
|
||||
<td>{formatTime(entry.eventTime)}</td>
|
||||
<td>
|
||||
{entry.userId}
|
||||
<AdminUserReferenceButton
|
||||
token={token}
|
||||
userId={entry.userId}
|
||||
onUnauthorized={onUnauthorized}
|
||||
/>
|
||||
</td>
|
||||
<td>{eventLabels[entry.eventName] ?? entry.eventName}</td>
|
||||
<td>{entry.projectId ?? '—'}</td>
|
||||
<td>{entry.source}</td>
|
||||
<td>{entry.status ?? '—'}</td>
|
||||
<td>{entry.clientVersion}</td>
|
||||
<td>
|
||||
<button
|
||||
type="button"
|
||||
className="admin-ghost-button"
|
||||
onClick={() => setDetail(entry)}
|
||||
>
|
||||
查看详情
|
||||
</button>
|
||||
</td>
|
||||
</tr>
|
||||
))}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
{loading ? (
|
||||
<p role="status">加载中…</p>
|
||||
) : !error && entries.length === 0 ? (
|
||||
<p>暂无客户端埋点数据</p>
|
||||
) : null}
|
||||
<div className="admin-action-row">
|
||||
<button
|
||||
type="button"
|
||||
className="admin-secondary-button"
|
||||
disabled={loading || page === 0}
|
||||
onClick={() => setPage((value) => value - 1)}
|
||||
>
|
||||
上一页
|
||||
</button>
|
||||
<span>第 {page + 1} 页</span>
|
||||
<button
|
||||
type="button"
|
||||
className="admin-secondary-button"
|
||||
disabled={loading || !nextCursor}
|
||||
onClick={() => {
|
||||
if (!nextCursor) return;
|
||||
setCursors([...cursors.slice(0, page + 1), nextCursor]);
|
||||
setPage(page + 1);
|
||||
}}
|
||||
>
|
||||
下一页
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
{detail ? (
|
||||
<Modal
|
||||
open
|
||||
title="客户端埋点详情"
|
||||
closeLabel="关闭详情"
|
||||
onClose={() => setDetail(null)}
|
||||
className="genarrative-ui"
|
||||
>
|
||||
<dl>
|
||||
{(
|
||||
Object.keys(fieldLabels) as Array<
|
||||
keyof AdminAgcTrackingEventEntry
|
||||
>
|
||||
)
|
||||
.filter((key) => key !== 'properties')
|
||||
.map((key) => (
|
||||
<div key={key}>
|
||||
<dt>{fieldLabels[key]}</dt>
|
||||
<dd style={{ overflowWrap: 'anywhere' }}>
|
||||
{detail[key] == null ? '—' : String(detail[key])}
|
||||
</dd>
|
||||
</div>
|
||||
))}
|
||||
</dl>
|
||||
<div className="admin-action-row">
|
||||
{(['creativeTaskId', 'agentRunId', 'clientVersion'] as const).map(
|
||||
(key) =>
|
||||
detail[key] ? (
|
||||
<button
|
||||
type="button"
|
||||
key={key}
|
||||
className="admin-secondary-button"
|
||||
onClick={() => related(key, detail[key]!)}
|
||||
>
|
||||
查询同一{fieldLabels[key]}
|
||||
</button>
|
||||
) : null,
|
||||
)}
|
||||
</div>
|
||||
<h3>事件属性</h3>
|
||||
<pre style={{ whiteSpace: 'pre-wrap', overflowWrap: 'anywhere' }}>
|
||||
{JSON.stringify(detail.properties, null, 2)}
|
||||
</pre>
|
||||
</Modal>
|
||||
) : null}
|
||||
</section>
|
||||
);
|
||||
}
|
||||
@@ -393,7 +393,7 @@ fn initialize_with_identity(config_dir: PathBuf, version: String, route: Route,
|
||||
return;
|
||||
}
|
||||
let session_id = Uuid::new_v4().to_string();
|
||||
let writer = AnalyticsWriter::start(config_dir, session_id.clone());
|
||||
let writer = AnalyticsWriter::start(config_dir.clone(), session_id.clone());
|
||||
let context = Context {
|
||||
route,
|
||||
editor_session_id: session_id.clone(),
|
||||
@@ -410,6 +410,7 @@ fn initialize_with_identity(config_dir: PathBuf, version: String, route: Route,
|
||||
.is_ok()
|
||||
{
|
||||
with_state(|state| state.start());
|
||||
super::upload::start(config_dir);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
//! 客户端产品埋点：本地采集与持久化，不发起上传。
|
||||
//! 客户端产品埋点：本地采集、持久化与确认后上传清理。
|
||||
pub(crate) mod contract;
|
||||
pub(crate) mod design;
|
||||
pub(crate) mod goal;
|
||||
@@ -8,3 +8,4 @@ pub(crate) mod project;
|
||||
pub(crate) mod run;
|
||||
pub(crate) mod session;
|
||||
pub(crate) mod store;
|
||||
pub(crate) mod upload;
|
||||
|
||||
@@ -18,6 +18,7 @@ const RETENTION: Duration = Duration::from_secs(7 * 24 * 60 * 60);
|
||||
const SEAL_AFTER: Duration = Duration::from_secs(5 * 60);
|
||||
const QUEUE_CAPACITY: usize = 1024;
|
||||
const MAX_META_BYTES: usize = 256 * 1024;
|
||||
const MAX_LIVE_OBSERVATIONS: usize = 16_384;
|
||||
|
||||
#[derive(Clone, Copy, Debug, Default)]
|
||||
pub(crate) struct StoreCounters {
|
||||
@@ -338,6 +339,8 @@ struct Store {
|
||||
events: Vec<(Event, String, Vec<u8>)>,
|
||||
bytes: usize,
|
||||
started: Option<Instant>,
|
||||
live_facts: VecDeque<(Instant, String, String)>,
|
||||
live_projects: HashMap<String, (Instant, bool)>,
|
||||
known: HashMap<String, String>,
|
||||
known_ids: HashSet<String>,
|
||||
counters: Arc<Counters>,
|
||||
@@ -369,6 +372,8 @@ impl Store {
|
||||
events: Vec::new(),
|
||||
bytes: 0,
|
||||
started: None,
|
||||
live_facts: VecDeque::new(),
|
||||
live_projects: HashMap::new(),
|
||||
known: HashMap::new(),
|
||||
known_ids: HashSet::new(),
|
||||
counters,
|
||||
@@ -615,7 +620,19 @@ impl Store {
|
||||
if result.is_ok() {
|
||||
for (event, _, _) in &events {
|
||||
observe_project(&mut self.projects, &meta.route, event);
|
||||
if let Some(project) = &event.project_id {
|
||||
let key = fact_digest(&meta.route, project);
|
||||
if let Some(opened) = self.projects.get(&key) {
|
||||
self.live_projects.insert(key, (Instant::now(), *opened));
|
||||
}
|
||||
}
|
||||
}
|
||||
self.live_facts.extend(
|
||||
meta.facts
|
||||
.iter()
|
||||
.map(|(key, id)| (Instant::now(), key.clone(), id.clone())),
|
||||
);
|
||||
self.trim_live();
|
||||
self.known_ids.extend(meta.event_ids);
|
||||
self.known.extend(meta.facts);
|
||||
} else {
|
||||
@@ -629,14 +646,52 @@ impl Store {
|
||||
|
||||
fn prune(&mut self, reserve: u64) -> io::Result<()> {
|
||||
let result = self.prune_files(reserve);
|
||||
// 其他实例也能清理封存批次；内存索引不能比持久批次活得更久。
|
||||
// 重建磁盘索引，但保留当前实例已观测事实，上传删除不能重授回调资格。
|
||||
self.projects.clear();
|
||||
self.known.clear();
|
||||
self.known_ids.clear();
|
||||
self.recover();
|
||||
self.trim_live();
|
||||
for (_, key, id) in &self.live_facts {
|
||||
self.known.insert(key.clone(), id.clone());
|
||||
self.known_ids.insert(id.clone());
|
||||
}
|
||||
for (key, (_, opened)) in &self.live_projects {
|
||||
self.projects
|
||||
.entry(key.clone())
|
||||
.and_modify(|known| *known |= *opened)
|
||||
.or_insert(*opened);
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
fn trim_live(&mut self) {
|
||||
self.trim_live_at(Instant::now());
|
||||
}
|
||||
|
||||
fn trim_live_at(&mut self, now: Instant) {
|
||||
// 仅当前进程内保留近期观测，固定条数上限；不形成永久历史账本。
|
||||
while self.live_facts.len() > MAX_LIVE_OBSERVATIONS
|
||||
|| self
|
||||
.live_facts
|
||||
.front()
|
||||
.is_some_and(|(time, _, _)| now.saturating_duration_since(*time) >= RETENTION)
|
||||
{
|
||||
self.live_facts.pop_front();
|
||||
}
|
||||
self.live_projects
|
||||
.retain(|_, (time, _)| now.saturating_duration_since(*time) < RETENTION);
|
||||
while self.live_projects.len() > MAX_LIVE_OBSERVATIONS {
|
||||
let oldest = self
|
||||
.live_projects
|
||||
.iter()
|
||||
.min_by_key(|(_, (time, _))| *time)
|
||||
.map(|(key, _)| key.clone())
|
||||
.unwrap();
|
||||
self.live_projects.remove(&oldest);
|
||||
}
|
||||
}
|
||||
|
||||
fn prune_files(&mut self, reserve: u64) -> io::Result<()> {
|
||||
session::prune(
|
||||
&self.root.join("instances"),
|
||||
@@ -914,3 +969,59 @@ fn observe_project(projects: &mut HashMap<String, bool>, route: &Route, event: &
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
/// 已验证的封存批次；路径只来自本地扫描，不接受服务端输入。
|
||||
pub(super) struct UploadBatch {
|
||||
path: PathBuf,
|
||||
pub request: serde_json::Value,
|
||||
pub batch_id: String,
|
||||
pub route: Route,
|
||||
pub event_count: usize,
|
||||
}
|
||||
|
||||
pub(super) fn upload_candidates(root: &Path) -> io::Result<Vec<PathBuf>> {
|
||||
let mut batches = list_batches(root)?;
|
||||
batches.sort_by_key(|batch| batch.modified);
|
||||
Ok(batches.into_iter().map(|batch| batch.path).collect())
|
||||
}
|
||||
|
||||
pub(super) fn load_upload_batch(path: PathBuf) -> io::Result<UploadBatch> {
|
||||
let meta = read_batch(&path)?;
|
||||
let events = read_events(&path)?;
|
||||
// 同一份内存副本再次核对，避免保留清理或外部修改造成两次读取不一致。
|
||||
if events.len() != meta.event_count
|
||||
|| events.iter().zip(&meta.event_ids).any(|(event, id)| {
|
||||
&event.event_id != id
|
||||
|| event.validate().is_err()
|
||||
|| event.user_id != meta.route.user_id
|
||||
|| event.editor_session_id != meta.editor_session_id
|
||||
})
|
||||
{
|
||||
return Err(invalid_data());
|
||||
}
|
||||
let request = serde_json::json!({
|
||||
"schema_version": 1, "batch_id": meta.batch_id,
|
||||
"destination_origin": meta.route.destination_origin, "user_id": meta.route.user_id,
|
||||
"events": events,
|
||||
});
|
||||
Ok(UploadBatch {
|
||||
path,
|
||||
request,
|
||||
batch_id: meta.batch_id,
|
||||
route: meta.route,
|
||||
event_count: meta.event_count,
|
||||
})
|
||||
}
|
||||
|
||||
impl UploadBatch {
|
||||
pub(super) fn exists(&self) -> bool {
|
||||
safe_metadata(&self.path).is_ok_and(|m| m.is_dir())
|
||||
}
|
||||
|
||||
pub(super) fn acknowledge(&self) -> io::Result<()> {
|
||||
match remove_batch(&self.path) {
|
||||
Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
|
||||
result => result,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,157 @@
|
||||
use super::*;
|
||||
use serde_json::json;
|
||||
|
||||
/// 由 API 测试启动隔离主站与数据库；本进程走实际封存、上传和确认清理。
|
||||
#[tokio::test]
|
||||
#[ignore = "requires isolated local API and database smoke bridge"]
|
||||
async fn analytics_live_api_upload_cleans_confirmed_batch() {
|
||||
#[derive(serde::Deserialize)]
|
||||
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
||||
struct Bridge {
|
||||
origin: String,
|
||||
user_id: String,
|
||||
access_token: String,
|
||||
result_path: PathBuf,
|
||||
}
|
||||
let bridge_path = std::env::var_os("AGC_ANALYTICS_SMOKE_BRIDGE")
|
||||
.expect("isolated smoke bridge path required");
|
||||
let bridge: Bridge = serde_json::from_slice(
|
||||
&read_bounded(Path::new(&bridge_path), 16 * 1024).expect("read smoke bridge"),
|
||||
)
|
||||
.expect("valid smoke bridge");
|
||||
let url = url::Url::parse(&bridge.origin).expect("valid smoke origin");
|
||||
assert!(
|
||||
url.scheme() == "http"
|
||||
&& url.origin().ascii_serialization() == bridge.origin
|
||||
&& url.host_str().is_some_and(|host| {
|
||||
host == "localhost"
|
||||
|| host
|
||||
.trim_matches(['[', ']'])
|
||||
.parse::<std::net::IpAddr>()
|
||||
.is_ok_and(|ip| ip.is_loopback())
|
||||
}),
|
||||
"smoke requires a loopback origin"
|
||||
);
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let mut store = open(dir.path());
|
||||
let event = event(&store.session_id, &bridge.user_id);
|
||||
let event_id = event.event_id.clone();
|
||||
store.record(
|
||||
Route::from_identity(Some(bridge.user_id.clone()), Some(&bridge.origin)),
|
||||
event,
|
||||
"live-api-session-start",
|
||||
);
|
||||
store.seal();
|
||||
let candidates = upload_candidates(&store.root).unwrap();
|
||||
assert_eq!(candidates.len(), 1, "real local batch must be sealed");
|
||||
let batch = load_upload_batch(candidates.into_iter().next().unwrap()).unwrap();
|
||||
let _identity = crate::platform_session::install_test_platform_session(
|
||||
&bridge.user_id,
|
||||
&bridge.access_token,
|
||||
&bridge.origin,
|
||||
);
|
||||
let client = reqwest::Client::builder()
|
||||
.no_proxy()
|
||||
.redirect(reqwest::redirect::Policy::none())
|
||||
.timeout(Duration::from_secs(30))
|
||||
.build()
|
||||
.unwrap();
|
||||
super::super::upload::cycle(&store.root, &client).await;
|
||||
assert!(
|
||||
!batch.exists(),
|
||||
"only a matching server acknowledgement permits cleanup"
|
||||
);
|
||||
assert!(upload_candidates(&store.root).unwrap().is_empty());
|
||||
fs::write(
|
||||
&bridge.result_path,
|
||||
serde_json::to_vec(&json!({"eventId": event_id, "batchId": batch.batch_id})).unwrap(),
|
||||
)
|
||||
.expect("write non-secret smoke result");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn upload_keeps_failures_then_deletes_only_acknowledged_batch_without_regranting_fact() {
|
||||
use std::net::TcpListener;
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let mut store = open(dir.path());
|
||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
let origin = format!("http://{}", listener.local_addr().unwrap());
|
||||
let route = Route::from_identity(Some("A".into()), Some(&origin));
|
||||
store.record(route.clone(), event(&store.session_id, "A"), "single-fact");
|
||||
store.seal();
|
||||
let batch = load_upload_batch(upload_candidates(&store.root).unwrap().remove(0)).unwrap();
|
||||
store.record(
|
||||
Route::from_identity(Some("B".into()), Some(&origin)),
|
||||
event(&store.session_id, "B"),
|
||||
"other-account",
|
||||
);
|
||||
store.seal();
|
||||
let id = batch.batch_id.clone();
|
||||
let server = std::thread::spawn(move || {
|
||||
for (status, body) in [
|
||||
(
|
||||
200,
|
||||
json!({"ok": true, "data": {"acknowledged_batch_ids": ["wrong"], "event_count": 1}, "error": null}),
|
||||
),
|
||||
(503, json!({"ok": false})),
|
||||
(
|
||||
200,
|
||||
json!({"ok": true, "data": {"acknowledged_batch_ids": [id], "event_count": 1}, "error": null}),
|
||||
),
|
||||
] {
|
||||
let (mut stream, _) = listener.accept().unwrap();
|
||||
stream
|
||||
.set_read_timeout(Some(Duration::from_secs(5)))
|
||||
.unwrap();
|
||||
let mut bytes = Vec::new();
|
||||
let mut chunk = [0; 4096];
|
||||
loop {
|
||||
let count = stream.read(&mut chunk).unwrap();
|
||||
assert!(count > 0);
|
||||
bytes.extend_from_slice(&chunk[..count]);
|
||||
if let Some(end) = bytes.windows(4).position(|w| w == b"\r\n\r\n") {
|
||||
let headers = String::from_utf8_lossy(&bytes[..end]).to_lowercase();
|
||||
let length: usize = headers
|
||||
.lines()
|
||||
.find_map(|line| line.strip_prefix("content-length: "))
|
||||
.unwrap()
|
||||
.parse()
|
||||
.unwrap();
|
||||
if bytes.len() >= end + 4 + length {
|
||||
assert!(headers.contains("authorization: bearer test-token"));
|
||||
assert!(headers.contains("x-genarrative-response-envelope: 1"));
|
||||
let request: serde_json::Value =
|
||||
serde_json::from_slice(&bytes[end + 4..]).unwrap();
|
||||
assert_eq!(request["user_id"], "A");
|
||||
assert_eq!(request["events"].as_array().unwrap().len(), 1);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
let body = body.to_string();
|
||||
write!(stream, "HTTP/1.1 {status} Test\r\nContent-Length: {}\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n{body}", body.len()).unwrap();
|
||||
}
|
||||
});
|
||||
let _identity =
|
||||
crate::platform_session::install_test_platform_session("A", "test-token", &origin);
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(Duration::from_secs(5))
|
||||
.build()
|
||||
.unwrap();
|
||||
for _ in 0..2 {
|
||||
super::super::upload::cycle(&store.root, &client).await;
|
||||
assert!(batch.exists());
|
||||
}
|
||||
super::super::upload::cycle(&store.root, &client).await;
|
||||
assert!(!batch.exists());
|
||||
assert_eq!(upload_candidates(&store.root).unwrap().len(), 1);
|
||||
server.join().unwrap();
|
||||
store.prune(0).unwrap();
|
||||
store.record(route, event(&store.session_id, "A"), "single-fact");
|
||||
assert!(store.events.is_empty());
|
||||
assert_eq!(store.counters.duplicate.load(Ordering::Relaxed), 1);
|
||||
}
|
||||
|
||||
fn route(user: &str) -> Route {
|
||||
Route {
|
||||
user_id: Some(user.into()),
|
||||
@@ -154,6 +305,7 @@ fn retention_counts_corrupt_files_and_removes_expired_sealed_batches() {
|
||||
let old = list_batches(&store.root).unwrap().pop().unwrap().path;
|
||||
// 发布目录 mtime 是封存时间；不需要等待一周。
|
||||
let old_time = SystemTime::now() - RETENTION - Duration::from_secs(1);
|
||||
store.trim_live_at(Instant::now() + RETENTION + Duration::from_secs(1));
|
||||
#[cfg(not(windows))]
|
||||
File::open(&old)
|
||||
.unwrap()
|
||||
@@ -261,7 +413,7 @@ fn real_background_writer_flushes_when_last_sender_closes() {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn retention_rebuilds_index_after_another_instance_removed_a_batch() {
|
||||
fn retention_preserves_recent_live_facts_after_another_instance_removed_a_batch() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let mut store = open(dir.path());
|
||||
store.record(
|
||||
@@ -274,6 +426,10 @@ fn retention_rebuilds_index_after_another_instance_removed_a_batch() {
|
||||
remove_batch(&batch.path).unwrap();
|
||||
assert_eq!(store.known.len(), 1);
|
||||
store.prune(0).unwrap();
|
||||
assert_eq!(store.known.len(), 1);
|
||||
assert_eq!(store.known_ids.len(), 1);
|
||||
store.trim_live_at(Instant::now() + RETENTION);
|
||||
store.prune(0).unwrap();
|
||||
assert!(store.known.is_empty());
|
||||
assert!(store.known_ids.is_empty());
|
||||
}
|
||||
|
||||
@@ -0,0 +1,228 @@
|
||||
//! 独立后台上传轮次；网络等待不占用 writer 或业务锁。
|
||||
use super::{contract::Route, store};
|
||||
use crate::platform_session::{current_platform_session, PlatformSessionSnapshot};
|
||||
use std::fs::{File, OpenOptions};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
const INTERVAL: Duration = Duration::from_secs(15 * 60);
|
||||
const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
const CYCLE_BUDGET: Duration = Duration::from_secs(120);
|
||||
const MAX_BATCHES: usize = 20;
|
||||
static FAILURES: AtomicU64 = AtomicU64::new(0);
|
||||
|
||||
fn failure() {
|
||||
let _ = FAILURES.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| {
|
||||
Some(n.saturating_add(1))
|
||||
});
|
||||
}
|
||||
|
||||
pub(super) fn start(config_dir: PathBuf) {
|
||||
tauri::async_runtime::spawn(async move {
|
||||
let Ok(client) = reqwest::Client::builder()
|
||||
.redirect(reqwest::redirect::Policy::none())
|
||||
.timeout(REQUEST_TIMEOUT)
|
||||
.build()
|
||||
else {
|
||||
return;
|
||||
};
|
||||
let root = config_dir.join("analytics");
|
||||
let mut next = tokio::time::Instant::now() + INTERVAL;
|
||||
loop {
|
||||
tokio::time::sleep_until(next).await;
|
||||
// 休眠恢复只跑一次，下一轮从当前时间重新计时。
|
||||
if advance_schedule(tokio::time::Instant::now(), &mut next) {
|
||||
cycle(&root, &client).await;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
fn advance_schedule(now: tokio::time::Instant, next: &mut tokio::time::Instant) -> bool {
|
||||
if now < *next {
|
||||
return false;
|
||||
}
|
||||
*next = now + INTERVAL;
|
||||
true
|
||||
}
|
||||
|
||||
fn claim(root: &Path) -> std::io::Result<File> {
|
||||
if !store::safe_metadata(root)?.is_dir() {
|
||||
return Err(store::invalid_data());
|
||||
}
|
||||
let path = root.join("upload.lock");
|
||||
if path.exists() {
|
||||
store::safe_metadata(&path)?;
|
||||
}
|
||||
let lock = OpenOptions::new()
|
||||
.read(true)
|
||||
.write(true)
|
||||
.create(true)
|
||||
.truncate(false)
|
||||
.open(path)?;
|
||||
lock.try_lock().map_err(|_| store::invalid_data())?;
|
||||
Ok(lock)
|
||||
}
|
||||
|
||||
fn matches_identity(route: &Route, session: &PlatformSessionSnapshot) -> bool {
|
||||
route.user_id.is_some()
|
||||
&& route.destination_origin.is_some()
|
||||
&& *route
|
||||
== Route::from_identity(Some(session.user_id.clone()), Some(&session.api_base_url))
|
||||
}
|
||||
|
||||
fn acknowledged(value: &serde_json::Value, batch: &store::UploadBatch) -> bool {
|
||||
value.get("ok") == Some(&serde_json::Value::Bool(true))
|
||||
&& value.get("error") == Some(&serde_json::Value::Null)
|
||||
&& value.pointer("/data/acknowledged_batch_ids")
|
||||
== Some(&serde_json::json!([batch.batch_id]))
|
||||
&& value
|
||||
.pointer("/data/event_count")
|
||||
.and_then(serde_json::Value::as_u64)
|
||||
== Some(batch.event_count as u64)
|
||||
}
|
||||
|
||||
fn stop_after(status: u16) -> bool {
|
||||
matches!(status, 401 | 403 | 429 | 500..=599)
|
||||
}
|
||||
|
||||
pub(super) async fn cycle(root: &Path, client: &reqwest::Client) {
|
||||
let Ok(_lock) = claim(root) else { return };
|
||||
let started = Instant::now();
|
||||
let Ok(candidates) = store::upload_candidates(root) else {
|
||||
failure();
|
||||
return;
|
||||
};
|
||||
let mut attempted = 0;
|
||||
for path in candidates {
|
||||
if attempted >= MAX_BATCHES || started.elapsed() >= CYCLE_BUDGET {
|
||||
break;
|
||||
}
|
||||
let Ok(batch) = store::load_upload_batch(path) else {
|
||||
continue;
|
||||
};
|
||||
let Some(session) = current_platform_session() else {
|
||||
break;
|
||||
};
|
||||
if !matches_identity(&batch.route, &session) || !batch.exists() {
|
||||
continue;
|
||||
}
|
||||
let Some(origin) = &batch.route.destination_origin else {
|
||||
continue;
|
||||
};
|
||||
attempted += 1;
|
||||
// 身份只在发起前读取；在途请求保持原凭据，切换账号后仍可清理原批次。
|
||||
let response = client
|
||||
.post(format!("{origin}/api/agc/analytics/batches"))
|
||||
.bearer_auth(&session.access_token)
|
||||
.header("x-genarrative-response-envelope", "1")
|
||||
.timeout(REQUEST_TIMEOUT.min(CYCLE_BUDGET.saturating_sub(started.elapsed())))
|
||||
.json(&batch.request)
|
||||
.send()
|
||||
.await;
|
||||
let Ok(mut response) = response else {
|
||||
failure();
|
||||
break;
|
||||
};
|
||||
let status = response.status().as_u16();
|
||||
if status != 200 {
|
||||
failure();
|
||||
if stop_after(status) {
|
||||
break;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
let mut bytes = Vec::new();
|
||||
let mut valid = true;
|
||||
let mut unavailable = false;
|
||||
loop {
|
||||
match response.chunk().await {
|
||||
Ok(Some(chunk)) if bytes.len() + chunk.len() <= 16 * 1024 => {
|
||||
bytes.extend_from_slice(&chunk)
|
||||
}
|
||||
Ok(None) => break,
|
||||
Err(_) => {
|
||||
valid = false;
|
||||
unavailable = true;
|
||||
break;
|
||||
}
|
||||
_ => {
|
||||
valid = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
if valid && serde_json::from_slice(&bytes).is_ok_and(|value| acknowledged(&value, &batch)) {
|
||||
if batch.acknowledge().is_err() {
|
||||
failure();
|
||||
}
|
||||
} else {
|
||||
failure();
|
||||
}
|
||||
if unavailable {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn first_cycle_waits_fifteen_minutes_and_resume_does_not_catch_up() {
|
||||
let start = tokio::time::Instant::now();
|
||||
let mut next = start + INTERVAL;
|
||||
assert!(!advance_schedule(start, &mut next));
|
||||
assert!(!advance_schedule(
|
||||
start + INTERVAL - Duration::from_secs(1),
|
||||
&mut next
|
||||
));
|
||||
assert!(advance_schedule(start + INTERVAL, &mut next));
|
||||
let resumed = start + INTERVAL * 8;
|
||||
assert!(advance_schedule(resumed, &mut next));
|
||||
assert!(!advance_schedule(resumed, &mut next));
|
||||
assert_eq!(next, resumed + INTERVAL);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn identity_does_not_claim_anonymous_or_other_account_or_platform() {
|
||||
let session = PlatformSessionSnapshot {
|
||||
user_id: "A".into(),
|
||||
access_token: "test".into(),
|
||||
api_base_url: "https://example.com/api".into(),
|
||||
identity_generation: 1,
|
||||
revision: 1,
|
||||
};
|
||||
let route = Route::from_identity(Some("A".into()), Some("https://example.com"));
|
||||
assert!(matches_identity(&route, &session));
|
||||
for route in [
|
||||
Route::from_identity(None, Some("https://example.com")),
|
||||
Route::from_identity(Some("B".into()), Some("https://example.com")),
|
||||
Route::from_identity(Some("A".into()), Some("https://other.example")),
|
||||
] {
|
||||
assert!(!matches_identity(&route, &session));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn upload_lock_is_nonblocking_and_single_owner() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let lock = claim(dir.path()).unwrap();
|
||||
assert!(claim(dir.path()).is_err());
|
||||
drop(lock);
|
||||
assert!(claim(dir.path()).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn toxic_batch_can_be_skipped_but_auth_and_service_failures_end_cycle() {
|
||||
for status in [400, 409, 413] {
|
||||
assert!(!stop_after(status));
|
||||
}
|
||||
for status in [401, 403, 429, 500, 503] {
|
||||
assert!(stop_after(status));
|
||||
}
|
||||
assert_eq!(INTERVAL, Duration::from_secs(900));
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -25,7 +25,7 @@
|
||||
|
||||
## AI 游戏创作与 Agent Runtime
|
||||
|
||||
- [客户端本地埋点与主站入库契约](./technical/【技术方案】客户端本地埋点与主站入库契约-2026-09-21.md)：本期本地采集已验收，合同与本地队列、会话窗口与项目、策划阶段成果、首次提交、两类 Agent run、Direct 文件/补丁成果、GUI 人工编辑、checkpoint 与 UI 保存，以及正式 Web 预览 ready 已验收；同一宿主组件链路关联已验证，不要求逐项覆盖全部资源操作，合计 12 类启用事件；本期只做事件采集和明文 JSONL 持久化，不做加密或上传。
|
||||
- [客户端本地埋点与主站入库契约](./technical/【技术方案】客户端本地埋点与主站入库契约-2026-09-21.md)：本地 12 类事件采集、每 15 分钟上传、私有事件表、确认后清理与后台明细查询已完成隔离环境验收；不扩充采集范围、不做加密，未部署生产。配置要求及验证边界见第 13 节。
|
||||
|
||||
- [AGC 资源 kind 枚举化契约](./technical/【技术方案】AI游戏创作智能体App实施计划-2026-06-24.md#2026-09-15-gamecreationapp-资源-kind-枚举化当前权威口径)：GameCreationApp 资源 kind 的 Rust enum、ts-rs 绑定、Unknown 可观测性和 shell 内重构边界。
|
||||
|
||||
|
||||
@@ -11,7 +11,8 @@
|
||||
- 首次提交按本地观测口径每个新项目最多一条：真实受理候选在后台持久消费 `.agent/analytics-goal.json` 资格后投递。资格跨批次清理保留，旧项目不初始化；此前候选丢失时允许后续真实受理消费，使用后者自己的用户与时间，不宣称绝对首次。消费后事件丢失可零条，重放与恢复不补历史；不得为埋点扫描双 Agent 完整历史或阻塞业务写盘。
|
||||
|
||||
- 当前合同唯一维护入口为[客户端本地埋点与主站入库契约](../../technical/【技术方案】客户端本地埋点与主站入库契约-2026-09-21.md)，原始需求作为仓库内历史来源保存；后续里程碑规范与实施计划放在 `docs/project-memory/plans/`。
|
||||
- 已确认本期只做本地事件采集和明文 JSONL 持久化，不做加密与上传；一个项目对应一个目标，事件按业务节点采集，5 分钟封存，7 天或 20 MiB 清理。后续上传周期为 15 分钟。
|
||||
- 本地采集阶段已验收明文 JSONL 持久化；当前仍不做加密。一个项目对应一个目标，事件按业务节点采集，5 分钟封存，7 天或 20 MiB 清理；上传失败也受保留上限约束。
|
||||
- 当前上传实现已完成隔离环境验收，证据见同一主规范第 13 节：每 15 分钟上传匹配当前账号与平台的封存批次，新增一张客户端事件私有表、批次原子入库与幂等确认、成功清理及独立后台明细栏目；失败静默留待下周期重试。真实客户端文件、HTTP、数据库与后台查询已关联同一事件验证，浏览器列表/筛选/详情通过；未部署生产。保持原 12 类事件和原采集边界。上线须配置 `GENARRATIVE_AGC_ANALYTICS_ORIGIN`，按数据库、API/后台、客户端顺序发布。
|
||||
- 已按技术负责人授权开始实施：合同与本地队列、会话窗口与项目接入、策划阶段成果、首次提交及两类 Agent run 已实现并经独立审查；定向测试、生产编译和前序 GUI 启停证据统一见主规范第 12 节。不得宣称完整产品采集已上线。
|
||||
## 2026-09-21 渠道进安装身份：不同渠道的 AGC 包体在同一台设备并存
|
||||
|
||||
|
||||
@@ -1,15 +1,17 @@
|
||||
# 客户端本地埋点与主站入库契约
|
||||
|
||||
Version: 0.23
|
||||
Status: 本期本地采集已实现并通过验收；不加密、不上传，证据与未验证范围见第 12 节
|
||||
Version: 0.25
|
||||
Status: 本地采集及上传、入库、成功清理、后台查询已实现并完成本地隔离环境验收；未部署生产
|
||||
Date: 2026-09-21
|
||||
需求来源：[Game Agent 埋点设计原始方案](./【需求来源】GameAgent埋点设计原始方案-2026-09-05.md)；原始方案与后续已确认决策有差异时，以本文为准。
|
||||
|
||||
## 1. 交付目标与范围
|
||||
|
||||
交付结果：在 Game Agent 与现役 Design Agent 客户端的真实业务节点产生有明确语义的产品事件，以明文 JSONL 批次持久化在应用级目录；后续主站按统一事件合同入库。
|
||||
阅读说明：第 1 至 12 节记录本地采集合同及其已完成的验收证据；当前上传、入库、清理与后台查询的实施合同统一见第 13 节。第 13 节细化第 8 节，不改变现有 12 类事件和采集触发点。当前整体仍在实施验收中，不表示已经上线。
|
||||
|
||||
本阶段验收以本地记录为准：字段来源可追溯、事件不会冒充业务成功、正常重放不重复、跨账号不串归属、埋点故障不影响正常操作。没有上传成功、服务端去重或报表已可用的承诺。
|
||||
交付结果：在 Game Agent 与现役 Design Agent 客户端的真实业务节点产生有明确语义的产品事件，以明文 JSONL 批次持久化在应用级目录；当前实施增加主站入库、确认后清理与后台明细查询，验收要求见第 13 节。
|
||||
|
||||
本地采集阶段的既有验收以本地记录为准：字段来源可追溯、事件不会冒充业务成功、正常重放不重复、跨账号不串归属、埋点故障不影响正常操作。这些证据不替代当前上传成功、服务端去重与后台查询的验收。
|
||||
|
||||
优先级：
|
||||
|
||||
@@ -17,7 +19,7 @@ Date: 2026-09-21
|
||||
- 风险项：项目目标身份稳定性、跨账号在途操作、多窗口前台时间、崩溃未闭合区间、revision 与保存事实、重复回调。
|
||||
- 可选项：无；本阶段不扩充分析维度或细粒度点击事件。
|
||||
|
||||
本阶段不实现文件加密、上传定时器、HTTP 接口、数据库表、服务端聚合或后台看板。不读取或批量转换已有对话历史为埋点，不改变现有 Agent 产物与恢复账本的职责。
|
||||
当前实施包括上传定时器、HTTP 接口、一张私有事件表与后台明细栏目；不实现文件加密、服务端聚合或统计看板。不读取或批量转换已有对话历史为埋点，不改变现有 Agent 产物与恢复账本的职责。
|
||||
|
||||
2026-09-21 存储范围确认：用户决定本阶段先不做加密，直接保存明文事件批次；无公钥配置、密钥生成或轮换前置要求。以后若启用加密，另行定义文件格式与接收兼容合同。
|
||||
|
||||
@@ -27,13 +29,13 @@ Date: 2026-09-21
|
||||
|
||||
2026-09-21 执行终态口径确认：等待澄清或审批算本次 run 正常完成，通过 end_reason 区分；用户取消和进程崩溃不计入 agent_run_failed，不进入已知成功/失败终态的分母。
|
||||
|
||||
2026-09-21 本地保留策略确认：埋点队列及相关索引最多保留 7 天、总量上限 20 MiB。任一条件达到即按第 7.3 节清理，未上传的数据也适用；不影响项目、Agent 对话或创作成果。本版尚不上传，数据不会无限保留等待上传功能上线。
|
||||
2026-09-21 本地保留策略确认：埋点队列及相关索引最多保留 7 天、总量上限 20 MiB。任一条件达到即按第 7.3 节清理，未上传的数据也适用；不影响项目、Agent 对话或创作成果。上传失败的数据同样不会无限保留。
|
||||
|
||||
2026-09-21 本地封存时间确认：从当前批次第一条事件进入开始，5 分钟到期后封存落盘；没有事件不生成空批次。该时间与后续每 15 分钟上传的周期独立，本轮实施评审采用 500 条及 1 MiB 的提前封存上限，作为内部常量。
|
||||
|
||||
2026-09-21 策划成果采集确认：审批通过后实际进入下一阶段，且该阶段变化成功持久化，视为一次 project_revision_created；不严格校验文档版本或文件差异，不为埋点新增文件 revision 机制。阶段内随意修改文件不逐次采集，打开、重开或关闭项目不补历史阶段事件。同策划会话同目标阶段幂等；不独立采集策划进度事件或审批、澄清状态字段。5 分钟是封存落盘周期，后续 15 分钟是上传周期，均不是采集周期。
|
||||
|
||||
用户已确定的后续要求：上传至主站数据库；客户端每 15 分钟上传一次；绑定真实用户与业务标识；成功后删除对应待上传副本；失败无弹窗、不阻塞正常进程；失败是否重试可选。本草案建议后续失败保留至下一个周期重试，不做立即重试；这是建议，不是已实现行为。
|
||||
用户已确定并授权实施的上传要求：上传至主站数据库；客户端每 15 分钟上传一次；绑定真实用户与业务标识；成功后删除对应待上传副本；失败无弹窗、不阻塞正常进程。当前实现采用失败保留至下一个周期重试，不做立即重试，仍受本地保留上限约束；整体完成情况以第 13 节验收为准。
|
||||
|
||||
本文是团队共享的埋点技术方案唯一维护入口，原始需求副本仅用于追溯。正式编码前，按仓库规范驱动工作流在 `docs/project-memory/plans/` 补齐里程碑规范及当前里程碑实施计划，并完成对应评审；本文的建议项不因迁入仓库而自动视为已确认。本次按技术负责人授权推进实施，里程碑仍按规范逐项评审和验收。
|
||||
|
||||
@@ -334,7 +336,7 @@ session.json 是本地恢复元数据，不是待上传事件；事件文件不
|
||||
|
||||
实例生命周期标记、打开历史索引与批次幂等映射均需有界清理；索引过期后不承诺识别无限久远的重放或首次打开。不可因此将未知 is_first_open 填 true。项目级首次创作提交标记跟随项目业务记录，不随埋点批次保留期清理。
|
||||
|
||||
本阶段没有成功上传清理，正常数据只因容量/保留期策略清理；不得启动一个空上传器并将数据标成“已上传”。这也意味着本阶段积累的数据不保证一直保留到上传功能上线。
|
||||
容量/保留期清理与上传成功清理独立：前者不表示已入库，后者必须收到主站匹配的整批确认，详见第 13.7 节。任何批次均不保证无限保留至上传成功。
|
||||
|
||||
### 7.4 数据最小化与文件边界
|
||||
|
||||
@@ -342,21 +344,21 @@ session.json 是本地恢复元数据，不是待上传事件；事件文件不
|
||||
- 事件只包含本方案白名单字段；session.json 与 meta.json 只保存路由、实例身份、生命周期、幂等和清理所需元数据。
|
||||
- 不采集 Prompt、对话正文、工具参数/结果、代码、文件正文、Token、Cookie、API Key 或异常堆栈；不因取消加密扩大采集范围。
|
||||
- 正式事件写入 analytics 队列，不在普通应用日志、诊断包中重复输出完整事件正文。目录沿用现有应用私有目录权限。
|
||||
- 后续网络传输仍使用 HTTPS，并由主站验证用户身份与字段。文件不加密不等于取消传输安全或服务端校验；本阶段无上传行为。
|
||||
- 网络传输使用 HTTPS，本地开发仅允许显式配置的 loopback HTTP；由主站验证用户身份与字段。文件不加密不等于取消传输安全或服务端校验，上传实现见第 13 节。
|
||||
|
||||
## 8. 为后续主站入库预留的合同
|
||||
## 8. 主站入库的事件与传输合同
|
||||
|
||||
### 8.1 保持事件与传输分离
|
||||
|
||||
本地事件 envelope 保持第 4 至 6 节合同，不添加 uploaded、retry_count、last_upload_error 等会随传输变化的业务字段。批次目标、发送进度与确认信息留在传输元数据。
|
||||
|
||||
后续批量请求固定包含 schema_version（整数 1）、batch_id（批次 UUID）、destination_origin（规范平台 origin）、user_id（真实用户 ID 字符串或 JSON null）、events（从 JSONL 解析得到的事件对象数组）。全批事件用户必须一致；本地路由只作发送提示，不能作为服务端归属的唯一证据。
|
||||
批量请求固定包含 schema_version（整数 1）、batch_id（批次 UUID）、destination_origin（规范平台 origin）、user_id（真实用户 ID 字符串）、events（从 JSONL 解析得到的事件对象数组）。本地匿名事件仍可保留 null，但本次上传不接收匿名批次。全批事件用户必须一致；本地路由只作发送提示，不能作为服务端归属的唯一证据。
|
||||
|
||||
后续客户端通过 HTTPS 上传 JSON 批次，传输体类型为 application/json；保持各事件的原 ID、时间和字段值。本阶段不实现该请求，不虚构 URL 或物理表名。
|
||||
客户端通过 HTTPS 上传 JSON 批次，传输体类型为 application/json；保持各事件的原 ID、时间和字段值。具体接口、物理表与本地开发例外以第 13.3 至 13.4 节为准。
|
||||
|
||||
为简化客户端文件管理，后续保留整批确认语义：服务端只有在全批事件已可靠保存，或按 event_id 去重确认内容相同时，才返回 `acknowledged_batch_ids`。客户端核对响应来自预期主站且 batch_id 属于本次请求后，删除对应整个批次目录。
|
||||
保持整批确认语义：服务端只有在全批事件已可靠保存，或按 event_id 去重确认内容相同时，才返回 `acknowledged_batch_ids`。客户端核对预期主站、成功 envelope、批次 ID 与事件数均匹配本次请求后，删除对应整个批次目录。
|
||||
|
||||
部分入库后失败时不确认整批；客户端保留原事件批次并在后续重发全批，由服务端按 event_id 去重。无需为了全批确认强制采用一个数据库大事务，但必须可靠判断全批完成。相同 event_id 不同内容为冲突，不能覆盖原记录或谎报确认。
|
||||
本次使用单批原子事务，任一事件冲突导致整批回滚且不确认；客户端保留原批次并在下个周期重发，由服务端按 event_id 去重。相同 event_id 不同内容为冲突，不能覆盖原记录或谎报确认。
|
||||
|
||||
本版不实现逐事件确认后的部分文件重写。永久无效批次的拒收/过期清理不得记为上传成功，也不能因为 HTTP 200、收到字节或仅解析成功就删除本地数据。
|
||||
|
||||
@@ -369,9 +371,9 @@ session.json 是本地恢复元数据，不是待上传事件；事件文件不
|
||||
- event_id 是全局唯一去重键。导入时保留原 ID、发生时间、关联字段和客户端版本，不重生成事件身份。
|
||||
- 服务端单独生成 received_at，用于观察延迟；不覆盖 event_time。
|
||||
- 不要求会话、任务、run 与 revision 按顺序到达，不因父事件丢失或批次乱序拒绝所有子事件；关联不完整需可识别。
|
||||
- 查询维度需要支持 user_id、project_id、creative_task_id、agent_run_id、event_name、event_time 和 client_version。主站物理表、索引和现有 tracking 适配在后续阶段核查后确定。
|
||||
- 查询维度支持 user_id、project_id、creative_task_id、agent_run_id、event_name、event_time 和 client_version。主站独立事件表、索引与后台查询以第 13 节为准，不进入现有主站 tracking 聚合链路。
|
||||
- 本地匿名事件、未知平台、未知 schema 版本不能静默改写归属或格式后入库。
|
||||
- 未来每 15 分钟触发后台上传、同一客户端同时最多一个上传任务；成功清理、失败静默、不阻塞退出。重试策略、请求上限和超时在上传里程碑定义。
|
||||
- 每 15 分钟触发后台上传、同一应用数据目录同时最多一个上传轮次；成功清理、失败静默、不阻塞退出。重试策略、请求上限和超时以第 13.6 节为准。
|
||||
|
||||
### 8.3 可用指标的限制
|
||||
|
||||
@@ -428,7 +430,7 @@ session.json 是本地恢复元数据，不是待上传事件；事件文件不
|
||||
| A 数据合同与本地队列 | 强类型事件、捕获身份、JSONL 批次、有界异步写入与恢复、静默失败 | 字段校验、JSONL 读取、跨账号、不阻塞、损坏隔离、容量与幂等测试通过 |
|
||||
| B 会话、窗口与项目 | start/end/focus/create/open，以及下次启动的本地 incomplete 标记 | 多窗口、重载、最小化、正常退出、旧实例存活识别、异常未闭合、项目重复打开测试与桌面 smoke |
|
||||
| C 两类 Agent 与成果链路 | submit/run、策划阶段推进成果、已有真实 revision、preview/save | 成功失败与等待、阶段实际推进且持久化成功与幂等、重试关联、无变化、版本漂移、真实访问与保存边界全部可核对；不新建策划文件 revision 机制 |
|
||||
| 后续独立阶段 | 上传主站与查询 | 另立规范；不在本版实施范围 |
|
||||
| 当前上传阶段 | 上传主站、确认后清理与后台查询 | 按第 13 节及配套里程碑实施，尚待完成整体验收 |
|
||||
|
||||
实施前为选定里程碑生成单独实现计划，前一个里程碑评审/验收后再推进下一项。
|
||||
|
||||
@@ -485,7 +487,7 @@ session.json 是本地恢复元数据，不是待上传事件；事件文件不
|
||||
| output_change_detected / duration_ms | 证据不足允许 null | 原方案未明确可空性，不能用 false/0 伪造确定值 |
|
||||
| 策划阶段成果（已确认） | 审批通过并实际进入下一阶段，成功持久化后复用 project_revision_created；同会话同目标阶段幂等 | 不采独立进度与审批字段；不严格校验文档版本/差异；阶段内修改不逐次采集，重开不补历史 |
|
||||
| revision | Game Agent 与其他已有正式 revision 路径按真实变化采集；策划使用 design:<session_id>:<target_phase> 阶段事实标识 | 策划标识不是文件版本，不据此伪造 run 文件变化；其他来源剔除纯元数据同步 |
|
||||
| 本地保留（已确认） | 最多 7 天或总量 20 MiB，任一条件达到即清理；超量先清最旧批次 | 未上传数据也会清理，本版不保证保留至上传功能上线；不删除项目及 Agent 业务数据 |
|
||||
| 本地保留（已确认） | 最多 7 天或总量 20 MiB，任一条件达到即清理；超量先清最旧批次 | 未上传数据也会清理，不保证保留至上传成功；不删除项目及 Agent 业务数据 |
|
||||
| 本地封存时间（已确认） | 当前批次首条事件进入后 5 分钟封存；无事件不生成空批次 | 与 15 分钟上传独立；强退可能丢失尚未落盘的一批事件 |
|
||||
| 本地批量上限（实施采用） | 达到 500 条或 1 MiB 序列化事件大小提前封存；超大单事件静默拒绝 | 内部常量，与 5 分钟条件并行；不新增用户设置 |
|
||||
| 字段补充 | schema_version、focus_interval_id、run end_reason；incomplete 仅是本地生命周期状态 | 需与后续接收端按同版合同实现，不任意扩展 properties，不将本地状态冒充产品事件 |
|
||||
@@ -530,3 +532,194 @@ session.json 是本地恢复元数据，不是待上传事件；事件文件不
|
||||
| 预览前端来源、生产编译及条款验收 | 合并 master 后 `appSurface.test.ts --threads=false -t 'preview\|预览\|play request\|run local\|local game'` 104 项、shell tsc、生产 `cargo +stable check`、fmt/编码/索引/diff及独立验收 | 全部通过；旧草稿自动预览不传 user，不据任意外部 URL 推断项目成果。Direct 原 run capture 与临时 guard 经独立接线审查；实测 host HTTP 可访问性不等价于完整 Chrome 双端验证。无上传、外部探测或新增业务版本；临时计划已融合删除 |
|
||||
|
||||
环境限制：本机固定 1.98.1 工具链缺可用 cargo，实际显式使用 stable（1.96.0）完成编译测试，未更改仓库固定版本。真实 GUI smoke 使用隐藏窗口，只证明启动与正常退出，不证明实际焦点操作；窗口并集、最小化和重复通知以状态测试验证，锁屏/休眠为已知观测限制。真实付费 Provider 和完整 GUI run smoke 未执行，Runtime 模拟 Provider 与前端 helper 测试不等价于真实模型验证。Direct 未完成交付合同的 guard 释放可能将账本中断，本版不改变其业务恢复行为，不承诺所有认证刷新都能继续生成；埋点不会将此类中断报告记为成功。人工和 UI 保存已有真实业务函数/文件/JSONL及前端 hook 验证，尚无完整 GUI 跨层操作 smoke；预览已有真实宿主 HTTP 和 JSONL 验证，未运行完整 Chrome 双端验证。
|
||||
|
||||
## 13. 上传、入库、成功清理与后台查询工程方案
|
||||
|
||||
### 13.1 交付目标与边界
|
||||
|
||||
状态：implemented / verified；技术负责人已授权开工，本地隔离环境验收完成，证据与限制见 §13.11，未核验或变更线上 schema。客户端定期上传已有封存批次，主站可靠保存并确认后清理对应本地副本，管理员能够查询真实入库的事件明细。
|
||||
|
||||
- 必须项：一张私有持久表、批量接收 API、后台上传器、确认后删除、一个后台明细栏目、身份和幂等验证。
|
||||
- 风险项：服务端已提交但响应丢失、跨账号/平台串传、保留清理与上传并发、删除导致观测资格重置、查询截断冒充最新数据。
|
||||
- 可选项：本版无。暂不做加密、匿名接收、聚合表、统计看板、Excel 导出、后台编辑/删除、立即重试或新增采集事件。
|
||||
- 采集仍为现有 12 类事件；不上传对话、提示词、文件内容、策划快照、本地路径、Token、项目资格账本或会话 incomplete 元数据。
|
||||
- 5 分钟 / 500 条 / 1 MiB 封存和 7 天 / 20 MiB 本地保留合同继续生效。上传成功清理与保留上限清理是两种独立原因，不能把后者统计为上传成功。
|
||||
|
||||
### 13.2 现状与接入边界
|
||||
|
||||
现有 `tracking_event` 保存主站的 event_key、scope、用户及 metadata_json，`tracking_daily_stat` 保存每日聚合；后台已有 `#tracking` 和 `#tables`。本方案新增 `agc_tracking_event`，不改主站埋点表、不进入个人任务奖励或主站每日统计链路。
|
||||
|
||||
正式路径为：客户端 Rust 上传器 → api-server 用户鉴权与 DTO 校验 → module-* 中的事件规则 → spacetime-client typed facade → spacetime-module 原子事务。后台通过管理端鉴权 BFF 和同一 facade 读取私有表，浏览器不直连数据库。
|
||||
|
||||
主站接受的是客户端观测数据，不因此确认项目所有权、成果真实性或奖励资格。允许父事件缺失、乱序、迟到，不给本地项目 ID 添加主站项目外键约束。
|
||||
|
||||
### 13.3 数据表与字段映射
|
||||
|
||||
新增普通私有持久表 `agc_tracking_event`，不是仅供实时订阅的 SpacetimeDB event table。一条事件一行。
|
||||
|
||||
| 字段 | 数据库类型 | 来源与约束 |
|
||||
| --- | --- | --- |
|
||||
| event_id | String，主键 | 原事件 UUID；唯一去重键 |
|
||||
| schema_version | u32 | 原事件版本，首版只接收 1 |
|
||||
| event_name | String | 原事件名，现有 12 类白名单 |
|
||||
| event_time | Timestamp | 原 UTC 毫秒时间，无损转换，不用接收时间替代 |
|
||||
| user_id | String | 本版只接收非匿名用户；与鉴权主体及批次 user_id 相同 |
|
||||
| editor_session_id | String | 原编辑器会话 ID |
|
||||
| project_id | Option<String> | 原值，保留 null |
|
||||
| creative_task_id | Option<String> | 原值，按事件合同与 project_id 一致 |
|
||||
| agent_run_id | Option<String> | 原值，保留 null |
|
||||
| agent_turn_id | Option<String> | 原值，保留 null |
|
||||
| status | Option<String> | 原枚举 success / failed 或 null |
|
||||
| error_code | Option<String> | 原错误码白名单或 null |
|
||||
| source | String | 原来源枚举 |
|
||||
| client_version | String | 原客户端版本 |
|
||||
| properties_json | String | 按事件类型验证后的专属属性 JSON；不接受任意字段 |
|
||||
| batch_id | String | 首次接收入库时的批次 UUID；重复事件不改写 |
|
||||
| received_at | Timestamp | 首次插入事务的 ctx.timestamp；重传不刷新 |
|
||||
|
||||
字段和 nullable 语义以第 4 至 6 节为准，不能用空串或零替代未知。传输 origin 不作为事件新增采集字段或表内分析维度；部署环境由服务端配置及批次 origin 校验隔离。
|
||||
|
||||
索引：received_at、(user_id, received_at)、(event_name, received_at)。主键承担去重。SpacetimeDB 2.8.3 的索引过滤不支持 Option<String> 项目字段（已由编译核验），项目条件使用上述时间索引候选进行精确筛选，不增加空串哨兵或重复项目字段。其余关联字段先作为受限查询条件，不为了预想报表遍建索引。
|
||||
|
||||
本版只新增表，同步 migration 注册、架构表目录、生成绑定和 schema 检查；不迁移旧主站埋点，不改动已有表字段。服务端暂不设置自动到期删除；本地 7 天 / 20 MiB 不适用于数据库。数据库容量与后续留存另行评估，不在本次加入自动清库。
|
||||
|
||||
### 13.4 上传接口与确认合同
|
||||
|
||||
新增 `POST /api/agc/analytics/batches`，使用现有用户 Bearer 认证与平台会话能力，HTTPS、`Content-Type: application/json`。本地开发仅复用既有显式配置的开发服务地址。该路径不是 external v1，不引入开发者 API Key。
|
||||
|
||||
请求保持第 8 节合同，snake_case：
|
||||
|
||||
| 字段 | 类型与来源 |
|
||||
| --- | --- |
|
||||
| schema_version | 整数 1，上传协议版本 |
|
||||
| batch_id | 原批次 UUID 字符串 |
|
||||
| destination_origin | 批次原目标平台 origin |
|
||||
| user_id | 批次原真实用户 ID 字符串 |
|
||||
| events | 从原 JSONL 解析的事件对象数组，完整对象示例见第 9 节 |
|
||||
|
||||
- 一次请求只发一个已封存批次，1 至 500 条。事件数据沿用 1 MiB 上限，JSON 请求体上限 2 MiB，容纳数组及 envelope 开销；服务端不依赖 Content-Length 自报大小。
|
||||
- 批次与所有事件的 user_id 必须相同且等于鉴权主体；匿名、未知 origin 批次不上传，不补认领到后来登录的账号。
|
||||
- destination_origin 必须匹配当前服务部署的可信公开 origin 配置，不使用未经信任的 Host 头作为判据。
|
||||
- 接收端配置 `GENARRATIVE_AGC_ANALYTICS_ORIGIN` 为规范 origin（无路径和尾部斜杠）：正式站 `https://www.genarrative.world`，测试站 `https://dev.genarrative.world`，本地联调填写客户端实际连接的 loopback origin。未配置/非法配置时接口返回 503，不猜测默认站点；允许 HTTPS 和本地 loopback HTTP。
|
||||
- 复用现有事件版本、枚举、字段长度与必填 nullable 校验；批内重复 event_id、未知字段或非法事件整批拒绝。
|
||||
- 本接口不进入普通成功路由 tracking 映射，不为每次上传另生成主站产品事件。
|
||||
|
||||
成功响应使用现有 API 成功 envelope，其 payload 为：
|
||||
|
||||
```json
|
||||
{
|
||||
"acknowledged_batch_ids": ["<本次请求的 batch_id>"],
|
||||
"event_count": 12
|
||||
}
|
||||
```
|
||||
|
||||
`event_count` 是本次确认的事件总数，包含内容一致的已有事件，不是新增行数。客户端只在成功 envelope、批次 ID 和本次事件数均匹配时删除。HTTP 200 本身不足以证明入库。
|
||||
|
||||
| 返回 | 意义 | 客户端处理 |
|
||||
| --- | --- | --- |
|
||||
| 200 且确认匹配 | 整批已提交或已存在且内容相同 | 删除该批次待上传副本 |
|
||||
| 400 | 格式、事件合同或 schema 不支持 | 本轮跳过，保留至后续周期或保留清理 |
|
||||
| 401 / 403 | 凭据失效、用户或 origin 不匹配 | 本轮停止该身份上传，不弹登录窗口、不替换归属 |
|
||||
| 409 | event_id 已存在但内容不同 | 不覆盖、不确认，保留原批次 |
|
||||
| 413 | 请求超限 | 不拆分或改写原批次，保留 |
|
||||
| 429 / 5xx / 网络超时 | 本轮不可确认 | 保留，下周期重试，不立即循环重试 |
|
||||
| 响应缺字段、ID/数量不符 | 确认无效 | 保留 |
|
||||
|
||||
沿用现有错误 envelope；响应不回显事件正文、凭据或其他用户数据。
|
||||
|
||||
### 13.5 原子入库与幂等
|
||||
|
||||
首版单批最多 500 条 / 1 MiB，采用一个原子事务完成，不新增批次回执表或服务端任务队列。
|
||||
|
||||
1. API 先鉴权并校验完整请求，再通过受信任服务身份调用数据库写路径；普通客户端身份不能直接写表。
|
||||
2. 事务核验全批事件。event_id 不存在则插入；已存在时比较原事件全部业务字段，properties 按解析后的结构比较，JSON 对象键顺序不影响等价。
|
||||
3. 相同事件视为已接收，保留第一次的 batch_id 和 received_at；事件 ID 相同但用户或任意事件内容不同则冲突，整批回滚。
|
||||
4. 只在确认数据库事务已提交后响应。Reducer 如被采用，facade 必须等待其已提交结果，不能将“已发送调用”当作完成。
|
||||
5. 服务端已提交但响应丢失、客户端确认后崩溃或本地删除失败，均可用原批次重发；不增加事件行数。
|
||||
|
||||
batch_id 是传输关联标识，event_id 是持久幂等依据。一张表不提供独立批次审计、批次内容哈希登记或永久批次状态机；本版不承诺这些能力。
|
||||
|
||||
### 13.6 客户端调度与身份
|
||||
|
||||
- 从客户端后台上传器启动时计时，每 15 分钟触发一次；首次不立即上传。休眠恢复最多执行一次到期轮次，不补跑错过的全部轮次。运行不足 15 分钟的数据留待后续启动并达到上传周期；退出不强制联网。
|
||||
- 每轮串行处理当时已封存且身份匹配的批次，优先旧批次；轮次开始后新封存的批次留待下一轮。
|
||||
- 同一应用数据目录通过一个非阻塞上传锁只允许一个实例执行上传轮次；竞争者本轮跳过。不使用长时间文件锁包住正常采集或网络等待。
|
||||
- 单请求总超时 30 秒；每轮最多处理 20 批、最多 2 分钟，不重叠执行。下一轮重新扫描；以上均为内部常量，不新增设置界面。
|
||||
- 仅使用与批次 user_id、destination_origin 完全一致的当前认证会话。账号或平台切换后不发起旧身份的新请求；已经发出的请求仍绑定原身份，收到有效确认可以清理原批次。
|
||||
- Token 只在内存请求中使用，不写入批次、表或日志。复用现有凭据获取能力，不新增埋点专属登录、刷新循环或登录提示。
|
||||
- 未登录、匿名、未知平台数据保留到原有保留策略清理；不主动登录所有历史账号来上传。
|
||||
- 400/409/413 等单批错误跳过该批继续本轮剩余候选，避免毒批次独占队列；认证错误或暂时服务故障结束本轮。下一周期仍按相同有界规则处理，不无限立即重试。
|
||||
|
||||
### 13.7 本地成功清理与现有状态边界
|
||||
|
||||
上传只读取已发布批次的 meta.json 和 events.jsonl，使用现有安全路径、大小、事件清单及一致性校验；不扫描项目内容。不得根据服务端返回字符串直接拼出待删路径，删除目标只能是本次本地选中的批次目录。
|
||||
|
||||
上传器持有一个有界批次内存副本再执行网络请求，不在网络等待期间占用 writer。保留清理可能先删原目录：发送前已缺失就跳过；发送后已缺失则确认清理视为无事可做。允许既有保留策略丢弃未上传数据，不建立第二套 durable outbox。
|
||||
|
||||
有效确认后调用本地存储层的批次删除入口，只删除对应目录，并同步/失效化有关批次缓存。删除失败保留，下一轮允许重传。上传异常只增加有界诊断计数，不弹窗、不阻塞 Agent、项目操作或退出。
|
||||
|
||||
不能删除项目 `.agent/analytics-goal.json`、`.agent/analytics-runs.json`、正式会话、对话或成果。保留会话恢复所需的控制元数据，空实例目录交给既有保留清理。
|
||||
|
||||
现有 store 的去重索引会随批次清理重建。上传更早删除批次不能重授首次提交资格或重置 run 身份：项目资格/run 账本保持原生命周期。当前实例另保留有界内存中的近期事实及项目观测，分别最多 16,384 项、最多 7 天，超限淘汰最旧项而非停止后续采集，避免最近已确认批次的重复回调重新生成 ID。内存缓存不承诺无限历史去重，不新增永久历史索引，也不从服务器或完整对话回填历史。
|
||||
|
||||
### 13.8 后台“客户端埋点”栏目
|
||||
|
||||
新增路由 `#agc-tracking`，显示名“客户端埋点”，与现有主站“埋点数据”并列；复用后台布局、筛选表单、用户引用组件及详情弹窗。owner 按现有规则可访问，member 必须具备新增页签权限 `agc-tracking`，接口同时执行权限校验。
|
||||
|
||||
新增 `GET /admin/api/agc/tracking-events`，管理端 DTO 沿用 camelCase：
|
||||
|
||||
- 筛选参数：`userId`、`projectId`、`creativeTaskId`、`agentRunId`、`eventName`、`clientVersion`、`startTime`、`endTime`、`cursor`、`limit`。
|
||||
- UI 首版展示发生时间范围、用户、项目和事件类型；其他关联筛选可通过详情入口携带，均使用精确匹配，不引入全文搜索。
|
||||
- 时间范围按 event_time，起点包含、终点不包含；时间格式为带时区 RFC3339。默认不设时间过滤，默认每页 50 条，最大 200 条。
|
||||
- 返回 payload：`entries`、`nextCursor`。条目包括表内字段的 camelCase 映射，`properties` 返回解析后的 JSON 对象；时间转为 UTC 字符串，Option 转 null，不向 UI 暴露 SATS 原始值。
|
||||
- 列表按首次入库时间 received_at 倒序，同时间按 event_id 稳定排序；明确列标题为“入库时间”和“发生时间”，不混称最新发生的事件。
|
||||
- cursor 固定查询快照上界、最后一行的 (received_at, event_id) 和筛选摘要；切换筛选重置游标，非法游标返回 400。新增数据点击刷新后出现，分页不依赖会随插入漂移的 offset。
|
||||
- 由数据库查询路径按索引及条件确定排序候选与页边界；若当前 SQL 不支持排序，通过受限 typed procedure 完成。不能任取 LIMIT N 行后排序并宣称为全表最新；实现计划须在现役 2.8.3 上验证此点，不照搬旧页面的截断查询方式。
|
||||
|
||||
列表列：入库时间、发生时间、用户、事件中文名、项目、来源、结果、客户端版本。空值显示“—”。详情展示全部既有字段与格式化 properties，包括会话/run/批次 ID；不新增编辑或删除按钮。空结果、加载失败、无权限沿用后台现有反馈，移动端表格支持横向滚动。
|
||||
|
||||
### 13.9 兼容、上线与回退
|
||||
|
||||
- 本地事件 schema_version 继续为 1，不改写已封存批次；升级后只处理仍在保留范围内且身份匹配的数据，不恢复已过期数据。
|
||||
- 发布顺序：新增数据库表/事务 → API 与后台 → 带上传器的客户端。旧客户端继续只写本地，旧批次无须迁移。
|
||||
- API 未部署或服务不可用时客户端静默保留并按周期重试；现有本地保留上限仍有效。
|
||||
- 回退客户端上传器或接收路由时保留已入库表，不删除正式数据；本地清理只对真实确认发生。没有新增远程开关或复杂灰度框架。
|
||||
- 新增事件版本必须显式扩展接收合同，不能靠忽略未知字段兼容。首次接收字段校验必须与客户端合同做对应测试。
|
||||
|
||||
### 13.10 实施边界与验收
|
||||
|
||||
技术负责人已授权按本方案开工；里程碑验收证据已融合至 §13.11，完成的临时里程碑规范与实施计划已删除。实现可在正确配置的部署环境使用，不代表已发布上线。
|
||||
|
||||
实现涉及 analytics 存储/上传器与平台会话、shared-contracts、现有 module-* 领域、spacetime-module 表/事务/migration、spacetime-client facade/绑定、api-server 用户及后台路由、admin-web 页面/权限。不更改 Game/Design Agent 埋点触发逻辑。
|
||||
|
||||
| 验收项 | 必须取得的证据 |
|
||||
| --- | --- |
|
||||
| 正常闭环 | 真实本地批次 → 测试主站数据库 → 确认 → 本地删除 → 后台可查相同 event_id |
|
||||
| 原子与去重 | 首传、原批重传不增行；任一冲突导致本次新行全部不落库；确认丢失后重传成功 |
|
||||
| 身份 | A 批次不能用 B 凭据或另一平台发送；匿名不认领；后台未授权读被拒绝 |
|
||||
| 静默失败 | 断网、超时、非成功/无效确认时保留；业务调用和退出不等待上传；下一周期可恢复 |
|
||||
| 清理边界 | 删除失败可重传；保留清理并发不误删其他批次；首次提交资格/run 身份保持；现有 7 天/20 MiB 不变 |
|
||||
| 调度 | 15 分钟、单目录互斥、有界轮次、休眠不积压补跑、毒批次不阻塞本轮其他数据 |
|
||||
| 后台查询 | 筛选、null/时间/属性显示、跨页稳定顺序及新数据刷新；数据量超过一页，不用截断结果冒充全量 |
|
||||
| 回归 | 现有事件合同、store/goal/run 定向测试及主站 tracking 不受影响 |
|
||||
| 工程检查 | schema 生成/检查、相关 Rust 测试与编译、后台类型检查、文档索引、编码和 diff 检查 |
|
||||
|
||||
运行时 smoke 必须包含真实 SpacetimeDB 提交及后台页面；mock 请求成功不算完成入库验收。不需要为本轮重新调用付费模型或扩大 Agent 创作测试。
|
||||
|
||||
本版实施口径为：只接收已登录用户、一张事件表、原子整批提交、首轮等待 15 分钟、30 秒请求超时、20 批/2 分钟轮次预算、失败下周期重试、后台按入库时间分页、服务端暂不自动过期。已确认的采集范围、上传周期、静默非阻塞和成功删除原则不变。
|
||||
|
||||
### 13.11 实现与验证记录（2026-09-21）
|
||||
|
||||
实现沿用既有 12 类事件和采集入口。新增 `analytics/upload.rs`、`agc_tracking_event` 私有持久表、批量接收与后台查询接口，以及后台“客户端埋点”页；未部署生产。
|
||||
|
||||
| 验证 | 结果与边界 |
|
||||
| --- | --- |
|
||||
| 客户端 | `cargo +stable test --locked --manifest-path apps/ai-game-creator-shell/src-tauri/Cargo.toml --bin genarrative-ai-game-creator-shell analytics -- --test-threads=1`：56 项通过，真实服务联调用例默认 ignored；普通客户端程序同时编译通过。覆盖调度、锁、身份匹配、错误确认/503 保留、成功清理、近期事实去重和既有 store/goal/run 回归 |
|
||||
| 领域与 API | module-runtime `agc_analytics` 4 项、api-server `agc_analytics` 3 项和后台权限 2 项通过；覆盖 12 类事件合同、非法字段、身份/origin、请求大小和后台权限 |
|
||||
| 真实数据库 | `scripts/agc-analytics-smoke.mjs` 在独立临时 SpacetimeDB 2.8.3 上使用最终源码 WASM 通过首传、重传、属性键序等价、冲突整批回滚、保留首次入库值、用户筛选、稳定分页、分页期间插入与刷新、UTC/null 和服务身份限制 |
|
||||
| 同一事件完整链路 | 手工执行 ignored `app::tests::agc_analytics_real_database_http_roundtrip`：真实客户端 Store 封存 JSONL → 生产上传器 → Axum HTTP → 真实数据库事务 → 确认后删除 → 后台接口查到相同 event_id/batch_id；同一事件重传只保留一行。用户认证使用测试夹具，未通过完整客户端 GUI 登录 |
|
||||
| 后台静态验证 | 28 项定向测试、admin-web 类型检查与作用域 ESLint 通过 |
|
||||
| 后台运行时 | 已编译 API 以临时 test 配置连接同一隔离数据库，`/healthz` 返回 200；真实浏览器登录、进入客户端埋点页、查看上传器生成记录、用户筛选及详情成功，时间、null 与 properties 显示正常。分页及刷新由真实数据库多页测试和页面定向测试验证 |
|
||||
| 数据契约 | migration 注册、表目录与生成绑定已同步；schema 检查通过 145 张表，DDD 边界检查通过；未改变既有表字段 |
|
||||
|
||||
编码、文档索引、作用域 Rust 格式与 diff 检查通过。`npm run dev:api-server` 编译通过，但本机启动脚本以既有短信配置覆盖测试环境变量，因缺少阿里云短信凭据而退出；页面 smoke 改为直接启动同一已编译 API，使用进程级 mock 认证/test 配置，未改本地配置文件。未验证完整客户端 GUI 登录、15 分钟墙钟等待、操作系统休眠及生产部署；调度周期与恢复通过定向状态测试验证。本轮不调用付费 Provider，不扩大 Agent 创作验收。
|
||||
|
||||
@@ -1362,6 +1362,14 @@ RPG 创作入口的配置 ID 是 `rpg`，当前 `visible=true`、`open=true`；
|
||||
- 源码：`server-rs/crates/spacetime-module/src/runtime/profile.rs`
|
||||
- 写入：由单条或批量 tracking procedure 在同一事务中随 `tracking_event` 更新，作为运营查询和个人任务进度的聚合投影。
|
||||
|
||||
### `agc_tracking_event`
|
||||
|
||||
- 普通私有持久表，保存客户端既有 12 类事件；以 `event_id` 为主键，不进入主站任务奖励或每日聚合。
|
||||
- 字段保留本地事件合同：schema_version、event_name、event_time、user_id、editor_session_id、project_id、creative_task_id、agent_run_id、agent_turn_id、status、error_code、source、client_version、properties_json；另存首次接收的 batch_id 和 received_at。
|
||||
- `upload_agc_analytics_batch` 只允许受信任服务身份调用，整批原子提交；相同事件重传不改首次批次和接收时间，内容冲突则整批回滚。
|
||||
- `list_agc_tracking_events` 经后台 BFF 调用，以 received_at 索引及用户/事件复合索引扫描候选（nullable 项目字段不建索引），在数据库事务中保留有界最新页；按 (received_at, event_id) 倒序及固定快照游标分页。
|
||||
- 不上传正文、快照、路径或凭据。字段/接口详见 [客户端本地埋点与主站入库契约](technical/【技术方案】客户端本地埋点与主站入库契约-2026-09-21.md) 第 13 节。
|
||||
|
||||
### `tracking_event`
|
||||
|
||||
- Rust 结构体：`TrackingEvent`
|
||||
|
||||
@@ -0,0 +1,407 @@
|
||||
#!/usr/bin/env node
|
||||
// 仅在临时 standalone 验证客户端埋点；显式传入带测试 bootstrap hash 的 WASM。
|
||||
import assert from 'node:assert/strict';
|
||||
import { spawn } from 'node:child_process';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { once } from 'node:events';
|
||||
import { access, chmod, mkdtemp, rm, writeFile } from 'node:fs/promises';
|
||||
import net from 'node:net';
|
||||
import os from 'node:os';
|
||||
import path from 'node:path';
|
||||
import { setTimeout as delay } from 'node:timers/promises';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
|
||||
import { createSpacetimeWebIdentity } from './spacetime-migration-common.mjs';
|
||||
|
||||
const root = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');
|
||||
const database = 'agc-analytics-smoke';
|
||||
const testSecret = 'a'.repeat(64); // 公开测试值，绝不能用于正式部署。
|
||||
const sensitive = [testSecret];
|
||||
const redact = (value) =>
|
||||
sensitive.reduce(
|
||||
(text, secret) => text.replaceAll(secret, '[REDACTED]'),
|
||||
String(value),
|
||||
);
|
||||
|
||||
function command(args) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const child = spawn('spacetime', args, {
|
||||
cwd: root,
|
||||
windowsHide: true,
|
||||
shell: false,
|
||||
stdio: ['ignore', 'pipe', 'pipe'],
|
||||
});
|
||||
let output = '';
|
||||
for (const stream of [child.stdout, child.stderr])
|
||||
stream.on('data', (chunk) => {
|
||||
output = (output + chunk).slice(-16000);
|
||||
});
|
||||
const timer = setTimeout(() => {
|
||||
child.kill();
|
||||
reject(new Error('SpacetimeDB command timed out'));
|
||||
}, 120000);
|
||||
child.once('error', (error) => {
|
||||
clearTimeout(timer);
|
||||
reject(error);
|
||||
});
|
||||
child.once('exit', (code) => {
|
||||
clearTimeout(timer);
|
||||
if (code === 0) resolve(output);
|
||||
else reject(new Error(redact(output)));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function localPort() {
|
||||
const listener = net.createServer();
|
||||
listener.listen(0, '127.0.0.1');
|
||||
await once(listener, 'listening');
|
||||
const port = listener.address().port;
|
||||
await new Promise((resolve) => listener.close(resolve));
|
||||
return port;
|
||||
}
|
||||
|
||||
async function call(url, token, name, input) {
|
||||
const response = await fetch(`${url}/v1/database/${database}/call/${name}`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
Authorization: `Bearer ${token}`,
|
||||
'Content-Type': 'application/json',
|
||||
},
|
||||
body: JSON.stringify([input]),
|
||||
signal: AbortSignal.timeout(30000),
|
||||
});
|
||||
const text = await response.text();
|
||||
assert(response.ok, `${name}: HTTP ${response.status}: ${redact(text)}`);
|
||||
return JSON.parse(text);
|
||||
}
|
||||
|
||||
function ok(result) {
|
||||
assert.equal(result[0], 0, `Expected Ok, got ${JSON.stringify(result)}`);
|
||||
return JSON.parse(result[1]);
|
||||
}
|
||||
|
||||
const event = (user = 'smoke-user-a') => ({
|
||||
schema_version: 1,
|
||||
event_id: randomUUID(),
|
||||
event_name: 'editor_session_start',
|
||||
event_time: '2026-09-21T12:00:00.123Z',
|
||||
user_id: user,
|
||||
editor_session_id: randomUUID(),
|
||||
project_id: null,
|
||||
creative_task_id: null,
|
||||
agent_run_id: null,
|
||||
agent_turn_id: null,
|
||||
status: 'success',
|
||||
error_code: null,
|
||||
source: 'editor',
|
||||
client_version: 'smoke-1',
|
||||
properties: { entry_source: 'direct_launch', first_project_id: null },
|
||||
});
|
||||
const batch = (events) => ({
|
||||
schema_version: 1,
|
||||
batch_id: randomUUID(),
|
||||
destination_origin: 'https://analytics-smoke.invalid',
|
||||
user_id: events[0].user_id,
|
||||
events,
|
||||
});
|
||||
|
||||
function pricingInput() {
|
||||
const tiered = (model, unit, keys) => ({
|
||||
model,
|
||||
unit,
|
||||
price: [1, []],
|
||||
prices: keys.map((key) => ({ key, price: 1 })),
|
||||
});
|
||||
return {
|
||||
admin_user_id: 'smoke-bootstrap',
|
||||
updated_at_micros: 1,
|
||||
bootstrap_secret: testSecret,
|
||||
models: [
|
||||
tiered('gemini-3.1-flash-image-preview', 'perGeneration', [
|
||||
'0.5K',
|
||||
'1K',
|
||||
'2K',
|
||||
]),
|
||||
tiered('gpt-image-2', 'perGeneration', ['1K', '2K']),
|
||||
...['audio1.0', 'eleven_text_to_sound_v2', 'chirp-v5'].map((model) => ({
|
||||
model,
|
||||
unit: 'perGeneration',
|
||||
price: [0, 1],
|
||||
prices: [],
|
||||
})),
|
||||
...[
|
||||
'seedance2.0-fast',
|
||||
'seedance2.0',
|
||||
'kling3.0',
|
||||
'kling3.0-omni',
|
||||
'veo3.1',
|
||||
'veo3.1-fast',
|
||||
].map((model) => tiered(model, 'perSecond', ['480p', '720p', '1080p'])),
|
||||
],
|
||||
};
|
||||
}
|
||||
|
||||
async function verify(url, serviceToken, outsiderToken) {
|
||||
const upload = async (payload) =>
|
||||
call(
|
||||
url,
|
||||
serviceToken,
|
||||
'upload_agc_analytics_batch',
|
||||
JSON.stringify(payload),
|
||||
);
|
||||
const list = async (query = {}) =>
|
||||
ok(
|
||||
await call(
|
||||
url,
|
||||
serviceToken,
|
||||
'list_agc_tracking_events',
|
||||
JSON.stringify(query),
|
||||
),
|
||||
);
|
||||
const first = batch([event(), event(), event()]);
|
||||
assert.deepEqual(ok(await upload(first)), {
|
||||
acknowledged_batch_ids: [first.batch_id],
|
||||
event_count: 3,
|
||||
});
|
||||
const initial = await list();
|
||||
assert.equal(initial.entries.length, 3);
|
||||
ok(await upload(first));
|
||||
assert.deepEqual(
|
||||
await list(),
|
||||
initial,
|
||||
'Replay must preserve rows, first batch ID and receipt time',
|
||||
);
|
||||
const reordered = structuredClone(first);
|
||||
reordered.events[0].properties = {
|
||||
first_project_id: null,
|
||||
entry_source: 'direct_launch',
|
||||
};
|
||||
ok(await upload(reordered));
|
||||
assert.deepEqual(
|
||||
await list(),
|
||||
initial,
|
||||
'JSON property order must not cause conflict',
|
||||
);
|
||||
|
||||
const rollbackCandidate = event();
|
||||
const conflict = batch([
|
||||
rollbackCandidate,
|
||||
{ ...first.events[0], client_version: 'changed' },
|
||||
]);
|
||||
const rejected = await upload(conflict);
|
||||
assert.equal(rejected[0], 1);
|
||||
assert.equal(rejected[1], 'agc_event_conflict');
|
||||
assert.deepEqual(
|
||||
await list(),
|
||||
initial,
|
||||
'Earlier insert in failed batch must roll back',
|
||||
);
|
||||
const userB = batch([event('smoke-user-b')]);
|
||||
ok(await upload(userB));
|
||||
assert.equal(
|
||||
(await list({ userId: 'smoke-user-b' })).entries[0].eventId,
|
||||
userB.events[0].event_id,
|
||||
);
|
||||
|
||||
const page1 = await list({ limit: 2 });
|
||||
assert(page1.nextCursor);
|
||||
const late = batch([event()]);
|
||||
ok(await upload(late));
|
||||
const seen = [...page1.entries];
|
||||
let cursor = page1.nextCursor;
|
||||
while (cursor) {
|
||||
const page = await list({ limit: 2, cursor });
|
||||
seen.push(...page.entries);
|
||||
cursor = page.nextCursor;
|
||||
}
|
||||
assert.equal(seen.length, 4);
|
||||
assert.equal(new Set(seen.map((row) => row.eventId)).size, 4);
|
||||
assert(
|
||||
!seen.some((row) => row.eventId === late.events[0].event_id),
|
||||
'Cursor snapshot must exclude later insert',
|
||||
);
|
||||
for (let index = 1; index < seen.length; index++) {
|
||||
const previous = seen[index - 1];
|
||||
const current = seen[index];
|
||||
assert(
|
||||
previous.receivedAt > current.receivedAt ||
|
||||
(previous.receivedAt === current.receivedAt &&
|
||||
previous.eventId > current.eventId),
|
||||
'Stable receipt/event ID descending order',
|
||||
);
|
||||
}
|
||||
const refreshed = await list();
|
||||
assert.equal(refreshed.entries.length, 5);
|
||||
assert.equal(refreshed.entries[0].eventId, late.events[0].event_id);
|
||||
assert.equal(refreshed.entries[0].projectId, null);
|
||||
assert.equal(refreshed.entries[0].eventTime, '2026-09-21T12:00:00.123Z');
|
||||
for (const [name, input] of [
|
||||
['upload_agc_analytics_batch', first],
|
||||
['list_agc_tracking_events', {}],
|
||||
]) {
|
||||
const unauthorized = await call(
|
||||
url,
|
||||
outsiderToken,
|
||||
name,
|
||||
JSON.stringify(input),
|
||||
);
|
||||
assert.equal(unauthorized[0], 1, 'Nonservice identity must be rejected');
|
||||
assert.match(unauthorized[1], /无权/);
|
||||
}
|
||||
console.log(
|
||||
'[agc-analytics-smoke] PASS: first/replay, JSON order, atomic rollback, user filter, stable snapshot pagination, refresh, null/time, service authorization. 5 rows persisted.',
|
||||
);
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const wasm = process.env.GENARRATIVE_AGC_ANALYTICS_SMOKE_WASM;
|
||||
assert(
|
||||
wasm,
|
||||
'Set GENARRATIVE_AGC_ANALYTICS_SMOKE_WASM to an isolated test WASM compiled with SHA256(a repeated 64 times) bootstrap hash.',
|
||||
);
|
||||
await access(wasm);
|
||||
const version = await command(['--version']);
|
||||
assert(
|
||||
version.includes('version 2.8.3') &&
|
||||
version.includes('8e410d2842147bd8e5a32a9589cc00c19f7478e2'),
|
||||
);
|
||||
const temp = await mkdtemp(
|
||||
path.join(os.tmpdir(), 'genarrative-agc-analytics-smoke-'),
|
||||
);
|
||||
let standalone;
|
||||
try {
|
||||
const port = await localPort();
|
||||
const url = `http://127.0.0.1:${port}`;
|
||||
standalone = spawn(
|
||||
'spacetime',
|
||||
[
|
||||
'start',
|
||||
'--data-dir',
|
||||
path.join(temp, 'data'),
|
||||
'--listen-addr',
|
||||
`127.0.0.1:${port}`,
|
||||
'--non-interactive',
|
||||
],
|
||||
{ windowsHide: true, shell: false, stdio: ['ignore', 'pipe', 'pipe'] },
|
||||
);
|
||||
let logs = '';
|
||||
for (const stream of [standalone.stdout, standalone.stderr])
|
||||
stream.on('data', (chunk) => {
|
||||
logs = (logs + chunk).slice(-8000);
|
||||
});
|
||||
const deadline = Date.now() + 30000;
|
||||
for (;;) {
|
||||
assert(standalone.exitCode === null, redact(logs));
|
||||
try {
|
||||
if (
|
||||
(await fetch(`${url}/v1/ping`, { signal: AbortSignal.timeout(1000) }))
|
||||
.ok
|
||||
)
|
||||
break;
|
||||
} catch {
|
||||
/* 等待启动 */
|
||||
}
|
||||
assert(
|
||||
Date.now() < deadline,
|
||||
`Standalone startup timeout: ${redact(logs)}`,
|
||||
);
|
||||
await delay(200);
|
||||
}
|
||||
const owner = await createSpacetimeWebIdentity({
|
||||
database,
|
||||
serverUrl: url,
|
||||
});
|
||||
const service = await createSpacetimeWebIdentity({
|
||||
database,
|
||||
serverUrl: url,
|
||||
});
|
||||
sensitive.push(owner.token, service.token);
|
||||
const config = path.join(temp, 'cli.toml');
|
||||
await command(['--config-path', config, 'login', '--token', owner.token]);
|
||||
await chmod(config, 0o600);
|
||||
await command([
|
||||
'--config-path',
|
||||
config,
|
||||
'publish',
|
||||
database,
|
||||
'--server',
|
||||
url,
|
||||
'--yes=all',
|
||||
'--no-config',
|
||||
'--bin-path',
|
||||
path.resolve(wasm),
|
||||
]);
|
||||
const bootstrap = await call(
|
||||
url,
|
||||
service.token,
|
||||
'initialize_editor_generation_pricing_config_if_missing_and_return',
|
||||
pricingInput(),
|
||||
);
|
||||
assert.equal(
|
||||
bootstrap[0],
|
||||
true,
|
||||
`Pricing service bootstrap failed: ${redact(JSON.stringify(bootstrap))}`,
|
||||
);
|
||||
await verify(url, service.token, owner.token);
|
||||
if (process.env.AGC_ANALYTICS_SMOKE_KEEP === '1') {
|
||||
const stopFile = path.join(temp, 'stop');
|
||||
const contextPath = path.join(temp, 'connection.json');
|
||||
await writeFile(
|
||||
contextPath,
|
||||
JSON.stringify({
|
||||
serverUrl: url,
|
||||
database,
|
||||
token: service.token,
|
||||
operatorToken: owner.token,
|
||||
configPath: config,
|
||||
stopFile,
|
||||
}),
|
||||
{ mode: 0o600 },
|
||||
);
|
||||
console.log(
|
||||
`[agc-analytics-smoke] Integration connection file: ${contextPath}`,
|
||||
);
|
||||
const stopDeadline = Date.now() + 45 * 60 * 1000;
|
||||
while (Date.now() < stopDeadline) {
|
||||
try {
|
||||
await access(stopFile);
|
||||
break;
|
||||
} catch {
|
||||
/* 联调结束后由调用方创建 stop 文件 */
|
||||
}
|
||||
await delay(1000);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (standalone && standalone.exitCode === null) {
|
||||
if (process.platform === 'win32') {
|
||||
const killer = spawn(
|
||||
'taskkill',
|
||||
['/PID', String(standalone.pid), '/T', '/F'],
|
||||
{ windowsHide: true, stdio: 'ignore' },
|
||||
);
|
||||
await once(killer, 'exit');
|
||||
} else {
|
||||
standalone.kill('SIGTERM');
|
||||
}
|
||||
if (standalone.exitCode === null) await once(standalone, 'exit');
|
||||
}
|
||||
// 仅删除本脚本 mkdtemp 创建的系统临时子目录。
|
||||
assert(
|
||||
path.dirname(path.resolve(temp)) === path.resolve(os.tmpdir()) &&
|
||||
path.basename(temp).startsWith('genarrative-agc-analytics-smoke-'),
|
||||
);
|
||||
await rm(temp, {
|
||||
recursive: true,
|
||||
force: true,
|
||||
maxRetries: 5,
|
||||
retryDelay: 200,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
main().catch((error) => {
|
||||
console.error(redact(error.stack ?? error));
|
||||
process.exitCode = 1;
|
||||
});
|
||||
Generated
+3
@@ -2904,9 +2904,12 @@ version = "0.1.0"
|
||||
dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
"shared-contracts",
|
||||
"shared-kernel",
|
||||
"spacetimedb",
|
||||
"time",
|
||||
"url",
|
||||
"uuid",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -2195,6 +2195,7 @@ fn admin_permission_requirement(_method: &Method, path: &str) -> AdminPermission
|
||||
"/admin/api/external-api-keys" => AnyTab(&["tables"]),
|
||||
"/admin/api/debug/http" => AnyTab(&["debug"]),
|
||||
"/admin/api/tracking/events" => AnyTab(&["tracking"]),
|
||||
"/admin/api/agc/tracking-events" => AnyTab(&["agc-tracking"]),
|
||||
"/admin/api/tracking/event-keys" => AnyTab(&["tracking", "tasks"]),
|
||||
"/admin/api/error-reports" => AnyTab(&["error-reports"]),
|
||||
"/admin/api/project-snapshots" => AnyTab(&["project-snapshots"]),
|
||||
@@ -2221,6 +2222,7 @@ fn admin_permission_requirement(_method: &Method, path: &str) -> AdminPermission
|
||||
"/admin/api/profile/users/detail" => AnyTab(&[
|
||||
"tables",
|
||||
"tracking",
|
||||
"agc-tracking",
|
||||
"recharge-orders",
|
||||
"editor-showcase",
|
||||
"editor-assets",
|
||||
@@ -6882,6 +6884,26 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn admin_tab_permissions_cover_shared_and_sensitive_routes() {
|
||||
assert!(
|
||||
enforce_admin_request_permission(
|
||||
"member",
|
||||
&["agc-tracking".into()],
|
||||
&[],
|
||||
&Method::GET,
|
||||
"/admin/api/profile/users/detail"
|
||||
)
|
||||
.is_ok()
|
||||
);
|
||||
assert!(
|
||||
enforce_admin_request_permission(
|
||||
"member",
|
||||
&["tracking".into()],
|
||||
&[],
|
||||
&Method::GET,
|
||||
"/admin/api/agc/tracking-events"
|
||||
)
|
||||
.is_err()
|
||||
);
|
||||
assert!(
|
||||
enforce_admin_request_permission(
|
||||
"member",
|
||||
@@ -6912,6 +6934,11 @@ mod tests {
|
||||
("tables", Method::GET, "/admin/api/database/tables"),
|
||||
("debug", Method::POST, "/admin/api/debug/http"),
|
||||
("tracking", Method::GET, "/admin/api/tracking/events"),
|
||||
(
|
||||
"agc-tracking",
|
||||
Method::GET,
|
||||
"/admin/api/agc/tracking-events",
|
||||
),
|
||||
("error-reports", Method::GET, "/admin/api/error-reports"),
|
||||
(
|
||||
"project-snapshots",
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
//! 客户端埋点只在数据库提交后确认；不使用普通路由 tracking outbox。
|
||||
use axum::{
|
||||
Json, Router,
|
||||
extract::{DefaultBodyLimit, Extension, Query, State},
|
||||
http::StatusCode,
|
||||
middleware,
|
||||
routing::{get, post},
|
||||
};
|
||||
use serde_json::Value;
|
||||
use shared_contracts::{admin::AdminAgcTrackingEventListQuery, agc_analytics::AgcAnalyticsBatch};
|
||||
use spacetime_client::SpacetimeClientError;
|
||||
|
||||
use crate::{
|
||||
admin::{AuthenticatedAdmin, require_admin_auth},
|
||||
api_response::json_success_body,
|
||||
auth::{AuthenticatedAccessToken, require_bearer_auth},
|
||||
http_error::AppError,
|
||||
request_context::RequestContext,
|
||||
state::AppState,
|
||||
};
|
||||
|
||||
const MAX_REQUEST_BYTES: usize = 2 * 1024 * 1024;
|
||||
|
||||
pub fn router(state: AppState) -> Router<AppState> {
|
||||
Router::new()
|
||||
.route(
|
||||
"/api/agc/analytics/batches",
|
||||
post(upload_batch)
|
||||
.route_layer(middleware::from_fn_with_state(
|
||||
state.clone(),
|
||||
require_bearer_auth,
|
||||
))
|
||||
.layer(DefaultBodyLimit::max(MAX_REQUEST_BYTES)),
|
||||
)
|
||||
.route(
|
||||
"/admin/api/agc/tracking-events",
|
||||
get(list_events).route_layer(middleware::from_fn_with_state(state, require_admin_auth)),
|
||||
)
|
||||
}
|
||||
|
||||
async fn upload_batch(
|
||||
State(state): State<AppState>,
|
||||
Extension(context): Extension<RequestContext>,
|
||||
Extension(auth): Extension<AuthenticatedAccessToken>,
|
||||
payload: Result<Json<Value>, axum::extract::rejection::JsonRejection>,
|
||||
) -> Result<Json<Value>, AppError> {
|
||||
let Json(raw) = payload.map_err(|error| {
|
||||
let status = if error.status() == StatusCode::PAYLOAD_TOO_LARGE {
|
||||
StatusCode::PAYLOAD_TOO_LARGE
|
||||
} else {
|
||||
StatusCode::BAD_REQUEST
|
||||
};
|
||||
AppError::from_status(status).with_message("客户端埋点请求格式无效")
|
||||
})?;
|
||||
// serde 也能将位置数组解成 struct；上传合同只接受命名字段的 JSON 对象。
|
||||
if !raw.is_object()
|
||||
|| !raw
|
||||
.get("events")
|
||||
.and_then(Value::as_array)
|
||||
.is_some_and(|events| events.iter().all(Value::is_object))
|
||||
{
|
||||
return Err(AppError::from_status(StatusCode::BAD_REQUEST)
|
||||
.with_message("客户端埋点必须使用事件对象"));
|
||||
}
|
||||
let batch: AgcAnalyticsBatch = serde_json::from_value(raw).map_err(|_| {
|
||||
AppError::from_status(StatusCode::BAD_REQUEST).with_message("客户端埋点字段不符合合同")
|
||||
})?;
|
||||
validate_subject_and_origin(
|
||||
&batch,
|
||||
auth.claims().user_id(),
|
||||
&state.config.agc_analytics_origin,
|
||||
)?;
|
||||
module_runtime::agc_analytics::validate_agc_analytics_batch(&batch).map_err(|error| {
|
||||
let status = if error == "events_too_large" {
|
||||
StatusCode::PAYLOAD_TOO_LARGE
|
||||
} else {
|
||||
StatusCode::BAD_REQUEST
|
||||
};
|
||||
AppError::from_status(status).with_message("客户端埋点批次格式不符合合同")
|
||||
})?;
|
||||
let acknowledgement = state
|
||||
.spacetime_client()
|
||||
.upload_agc_analytics_batch(batch)
|
||||
.await
|
||||
.map_err(map_database_error)?;
|
||||
Ok(json_success_body(Some(&context), acknowledgement))
|
||||
}
|
||||
|
||||
async fn list_events(
|
||||
State(state): State<AppState>,
|
||||
Extension(context): Extension<RequestContext>,
|
||||
Extension(_admin): Extension<AuthenticatedAdmin>,
|
||||
Query(query): Query<AdminAgcTrackingEventListQuery>,
|
||||
) -> Result<Json<Value>, AppError> {
|
||||
module_runtime::agc_analytics::validate_agc_tracking_query(&query).map_err(|_| {
|
||||
AppError::from_status(StatusCode::BAD_REQUEST).with_message("客户端埋点查询参数无效")
|
||||
})?;
|
||||
let payload = state
|
||||
.spacetime_client()
|
||||
.list_agc_tracking_events(query)
|
||||
.await
|
||||
.map_err(map_database_error)?;
|
||||
Ok(json_success_body(Some(&context), payload))
|
||||
}
|
||||
|
||||
fn validate_subject_and_origin(
|
||||
batch: &AgcAnalyticsBatch,
|
||||
user_id: &str,
|
||||
expected_origin: &str,
|
||||
) -> Result<(), AppError> {
|
||||
// 公开地址必须来自部署配置；客户端 body 和 Host 头均不是配置来源。
|
||||
let origin = url::Url::parse(expected_origin).ok().filter(|url| {
|
||||
let loopback = url.host_str().is_some_and(|host| {
|
||||
host == "localhost"
|
||||
|| host
|
||||
.trim_matches(['[', ']'])
|
||||
.parse::<std::net::IpAddr>()
|
||||
.is_ok_and(|ip| ip.is_loopback())
|
||||
});
|
||||
(url.scheme() == "https" || (url.scheme() == "http" && loopback))
|
||||
&& url.host_str().is_some()
|
||||
&& url.username().is_empty()
|
||||
&& url.password().is_none()
|
||||
&& url.origin().ascii_serialization() == expected_origin
|
||||
});
|
||||
if origin.is_none() {
|
||||
return Err(AppError::from_status(StatusCode::SERVICE_UNAVAILABLE)
|
||||
.with_message("客户端埋点接收地址尚未配置"));
|
||||
}
|
||||
if batch.user_id != user_id
|
||||
|| batch.destination_origin != expected_origin
|
||||
|| batch
|
||||
.events
|
||||
.iter()
|
||||
.any(|event| event.user_id.as_deref() != Some(user_id))
|
||||
{
|
||||
return Err(
|
||||
AppError::from_status(StatusCode::FORBIDDEN).with_message("客户端埋点身份或平台不匹配")
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn map_database_error(error: SpacetimeClientError) -> AppError {
|
||||
match error {
|
||||
SpacetimeClientError::Procedure(message) if message == "agc_event_conflict" => {
|
||||
AppError::from_status(StatusCode::CONFLICT).with_message("客户端埋点事件 ID 内容冲突")
|
||||
}
|
||||
SpacetimeClientError::Procedure(message)
|
||||
if matches!(message.as_str(), "invalid_agc_query" | "invalid_agc_cursor") =>
|
||||
{
|
||||
AppError::from_status(StatusCode::BAD_REQUEST).with_message("客户端埋点查询参数无效")
|
||||
}
|
||||
// 不回显数据库或上传内容，超时/未知提交结果不产生确认。
|
||||
_ => AppError::from_status(StatusCode::SERVICE_UNAVAILABLE)
|
||||
.with_message("客户端埋点数据服务暂不可用"),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn batch() -> AgcAnalyticsBatch {
|
||||
AgcAnalyticsBatch {
|
||||
schema_version: 1,
|
||||
batch_id: uuid::Uuid::new_v4().to_string(),
|
||||
destination_origin: "https://dev.genarrative.world".into(),
|
||||
user_id: "user-a".into(),
|
||||
events: vec![],
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn agc_analytics_subject_and_deployment_are_both_required() {
|
||||
let batch = batch();
|
||||
assert!(validate_subject_and_origin(&batch, "user-a", &batch.destination_origin).is_ok());
|
||||
for (user, origin, expected) in [
|
||||
(
|
||||
"user-b",
|
||||
"https://dev.genarrative.world",
|
||||
StatusCode::FORBIDDEN,
|
||||
),
|
||||
(
|
||||
"user-a",
|
||||
"https://www.genarrative.world",
|
||||
StatusCode::FORBIDDEN,
|
||||
),
|
||||
("user-a", "", StatusCode::SERVICE_UNAVAILABLE),
|
||||
(
|
||||
"user-a",
|
||||
"https://dev.genarrative.world/path",
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
),
|
||||
(
|
||||
"user-a",
|
||||
"http://dev.genarrative.world",
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
),
|
||||
] {
|
||||
assert_eq!(
|
||||
validate_subject_and_origin(&batch, user, origin)
|
||||
.unwrap_err()
|
||||
.status_code(),
|
||||
expected
|
||||
);
|
||||
}
|
||||
let mut local = batch;
|
||||
local.destination_origin = "http://127.0.0.1:8082".into();
|
||||
assert!(validate_subject_and_origin(&local, "user-a", &local.destination_origin).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn agc_analytics_database_failure_never_becomes_acknowledgement() {
|
||||
for (message, status) in [
|
||||
("agc_event_conflict", StatusCode::CONFLICT),
|
||||
("invalid_agc_cursor", StatusCode::BAD_REQUEST),
|
||||
("unknown", StatusCode::SERVICE_UNAVAILABLE),
|
||||
] {
|
||||
assert_eq!(
|
||||
map_database_error(SpacetimeClientError::Procedure(message.into())).status_code(),
|
||||
status
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -56,6 +56,7 @@ pub fn build_router(state: AppState) -> Router {
|
||||
.merge(modules::raw::router(state.clone()))
|
||||
.merge(modules::project_snapshots::router(state.clone()))
|
||||
.merge(crate::error_reports::router(state.clone()))
|
||||
.merge(crate::agc_analytics::router(state.clone()))
|
||||
.route(
|
||||
"/api/profile/recharge/wechat/notify",
|
||||
post(handle_wechat_pay_notify),
|
||||
@@ -256,6 +257,234 @@ mod tests {
|
||||
const TEST_PASSWORD: &str = "secret123";
|
||||
const INTERNAL_TEST_SECRET: &str = "test-internal-secret";
|
||||
|
||||
/// 需 scripts/agc-analytics-smoke.mjs 创建的隔离数据库；认证用户为测试夹具，
|
||||
/// HTTP、facade、数据库事务和后台查询均走正式实现。
|
||||
#[tokio::test]
|
||||
#[ignore = "requires isolated AGC analytics smoke database"]
|
||||
async fn agc_analytics_real_database_http_roundtrip() {
|
||||
let path =
|
||||
std::env::var("AGC_ANALYTICS_SMOKE_CONNECTION").expect("isolated connection file");
|
||||
let connection: Value = serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap();
|
||||
let db_url = connection["serverUrl"].as_str().unwrap();
|
||||
assert!(
|
||||
db_url.starts_with("http://127.0.0.1:"),
|
||||
"only isolated local database"
|
||||
);
|
||||
assert!(
|
||||
connection["database"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.starts_with("agc-analytics-smoke")
|
||||
);
|
||||
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let origin = format!("http://{}", listener.local_addr().unwrap());
|
||||
let state = AppState::new(AppConfig {
|
||||
spacetime_server_url: db_url.into(),
|
||||
spacetime_database: connection["database"].as_str().unwrap().into(),
|
||||
spacetime_token: Some(connection["token"].as_str().unwrap().into()),
|
||||
agc_analytics_origin: origin.clone(),
|
||||
admin_username: Some("analytics-smoke-owner".into()),
|
||||
admin_password: Some("analytics-smoke-fixture-password".into()),
|
||||
..AppConfig::default()
|
||||
})
|
||||
.unwrap();
|
||||
let user = seed_phone_user_with_password(&state, "13800138001", TEST_PASSWORD).await;
|
||||
let token = sign_test_user_token(&state, &user, "analytics-live-db");
|
||||
let app = build_router(state);
|
||||
let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
|
||||
let client = Client::builder()
|
||||
.no_proxy()
|
||||
.timeout(std::time::Duration::from_secs(30))
|
||||
.build()
|
||||
.unwrap();
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
let batch_id = uuid::Uuid::new_v4().to_string();
|
||||
let payload = serde_json::json!({
|
||||
"schema_version":1, "batch_id":batch_id, "destination_origin":origin, "user_id":user.id,
|
||||
"events":[{
|
||||
"schema_version":1, "event_id":id, "event_name":"editor_session_start",
|
||||
"event_time":"2026-09-21T00:00:00.000Z", "user_id":user.id,
|
||||
"editor_session_id":uuid::Uuid::new_v4().to_string(), "project_id":null,
|
||||
"creative_task_id":null, "agent_run_id":null, "agent_turn_id":null,
|
||||
"status":"success", "error_code":null, "source":"editor", "client_version":"smoke",
|
||||
"properties":{"entry_source":"direct_launch", "first_project_id":null}
|
||||
}]
|
||||
});
|
||||
for _ in 0..2 {
|
||||
let response = client
|
||||
.post(format!("{origin}/api/agc/analytics/batches"))
|
||||
.bearer_auth(&token)
|
||||
.header("x-genarrative-response-envelope", "1")
|
||||
.json(&payload)
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.status().as_u16(), 200);
|
||||
let ack: Value = response.json().await.unwrap();
|
||||
assert_eq!(ack["ok"], true);
|
||||
assert_eq!(
|
||||
ack["data"]["acknowledged_batch_ids"],
|
||||
serde_json::json!([batch_id])
|
||||
);
|
||||
assert_eq!(ack["data"]["event_count"], 1);
|
||||
}
|
||||
let response = client.post(format!("{origin}/admin/api/login"))
|
||||
.json(&serde_json::json!({"username":"analytics-smoke-owner", "password":"analytics-smoke-fixture-password"}))
|
||||
.send().await.unwrap();
|
||||
assert_eq!(response.status().as_u16(), 200);
|
||||
let login: Value = response.json().await.unwrap();
|
||||
let admin_token = login["token"].as_str().unwrap();
|
||||
let response = client
|
||||
.get(format!("{origin}/admin/api/agc/tracking-events"))
|
||||
.bearer_auth(admin_token)
|
||||
.query(&[("userId", user.id.as_str())])
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.status().as_u16(), 200);
|
||||
let rows: Value = response.json().await.unwrap();
|
||||
let matches = rows["entries"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|row| row["eventId"] == id)
|
||||
.count();
|
||||
assert_eq!(
|
||||
matches, 1,
|
||||
"two uploads produce one persisted row in admin read"
|
||||
);
|
||||
// 启动客户端现役 store/upload 集成用例，关联实际磁盘批次与同一个主站数据库。
|
||||
let bridge_dir = tempfile::tempdir().unwrap();
|
||||
let bridge = bridge_dir.path().join("bridge.json");
|
||||
let result_path = bridge_dir.path().join("result.json");
|
||||
std::fs::write(
|
||||
&bridge,
|
||||
serde_json::to_vec(&serde_json::json!({
|
||||
"origin":origin, "userId":user.id, "accessToken":token, "resultPath":result_path
|
||||
}))
|
||||
.unwrap(),
|
||||
)
|
||||
.unwrap();
|
||||
let executable = std::env::var("AGC_ANALYTICS_SMOKE_CLIENT_BINARY")
|
||||
.expect("compiled client test binary");
|
||||
let mut child = tokio::process::Command::new(executable);
|
||||
child
|
||||
.args([
|
||||
"analytics::store::tests::analytics_live_api_upload_cleans_confirmed_batch",
|
||||
"--ignored",
|
||||
"--exact",
|
||||
"--test-threads=1",
|
||||
])
|
||||
.env("AGC_ANALYTICS_SMOKE_BRIDGE", &bridge)
|
||||
.kill_on_drop(true);
|
||||
#[cfg(windows)]
|
||||
child.creation_flags(0x08000000);
|
||||
let output = tokio::time::timeout(std::time::Duration::from_secs(90), child.output())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"client live upload fixture failed: {}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
let uploaded: Value = serde_json::from_slice(
|
||||
&std::fs::read(result_path)
|
||||
.expect("client result required, zero matched tests is not success"),
|
||||
)
|
||||
.unwrap();
|
||||
let response = client
|
||||
.get(format!("{origin}/admin/api/agc/tracking-events"))
|
||||
.bearer_auth(admin_token)
|
||||
.query(&[("userId", user.id.as_str())])
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.status().as_u16(), 200);
|
||||
let rows: Value = response.json().await.unwrap();
|
||||
assert_eq!(
|
||||
rows["entries"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|row| row["eventId"] == uploaded["eventId"]
|
||||
&& row["batchId"] == uploaded["batchId"])
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
server.abort();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn agc_analytics_upload_enforces_auth_identity_contract_and_body_limit() {
|
||||
let state = AppState::new(AppConfig {
|
||||
agc_analytics_origin: "https://dev.genarrative.world".into(),
|
||||
admin_username: Some("analytics-smoke-owner".into()),
|
||||
admin_password: Some("analytics-smoke-fixture-password".into()),
|
||||
..AppConfig::default()
|
||||
})
|
||||
.expect("state");
|
||||
let user = seed_phone_user_with_password(&state, "13800138001", TEST_PASSWORD).await;
|
||||
let token = sign_test_user_token(&state, &user, "analytics-upload-test");
|
||||
let app = build_router(state);
|
||||
let payload = serde_json::json!({
|
||||
"schema_version":1, "batch_id":uuid::Uuid::new_v4().to_string(),
|
||||
"destination_origin":"https://dev.genarrative.world", "user_id":user.id,
|
||||
"events": [{
|
||||
"schema_version":1, "event_id":uuid::Uuid::new_v4().to_string(),
|
||||
"event_name":"editor_session_start", "event_time":"2026-09-21T00:00:00.000Z",
|
||||
"user_id":user.id, "editor_session_id":uuid::Uuid::new_v4().to_string(),
|
||||
"project_id":null, "creative_task_id":null, "agent_run_id":null,
|
||||
"agent_turn_id":null, "status":"success", "error_code":null,
|
||||
"source":"editor", "client_version":"1.0.0",
|
||||
"properties":{"entry_source":"direct_launch", "first_project_id":null}
|
||||
}]
|
||||
});
|
||||
let mut wrong_user = payload.clone();
|
||||
wrong_user["user_id"] = "other-user".into();
|
||||
let mut wrong_origin = payload.clone();
|
||||
wrong_origin["destination_origin"] = "https://www.genarrative.world".into();
|
||||
let mut bad_contract = payload.clone();
|
||||
bad_contract["events"][0]["properties"]["prompt"] = "not collected".into();
|
||||
for (body, authenticated, expected) in [
|
||||
(payload.to_string(), false, StatusCode::UNAUTHORIZED),
|
||||
(wrong_user.to_string(), true, StatusCode::FORBIDDEN),
|
||||
(wrong_origin.to_string(), true, StatusCode::FORBIDDEN),
|
||||
(bad_contract.to_string(), true, StatusCode::BAD_REQUEST),
|
||||
("{".into(), true, StatusCode::BAD_REQUEST),
|
||||
(
|
||||
" ".repeat(2 * 1024 * 1024 + 1),
|
||||
true,
|
||||
StatusCode::PAYLOAD_TOO_LARGE,
|
||||
),
|
||||
] {
|
||||
let mut request = Request::builder()
|
||||
.method("POST")
|
||||
.uri("/api/agc/analytics/batches")
|
||||
.header("content-type", "application/json")
|
||||
.header("x-genarrative-response-envelope", "1");
|
||||
if authenticated {
|
||||
request = request.header("authorization", format!("Bearer {token}"));
|
||||
}
|
||||
let response = app
|
||||
.clone()
|
||||
.oneshot(request.body(Body::from(body)).unwrap())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.status(), expected);
|
||||
}
|
||||
let response = app
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.uri("/admin/api/agc/tracking-events")
|
||||
.body(Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
|
||||
}
|
||||
|
||||
mod http_tracing {
|
||||
use std::{
|
||||
collections::BTreeMap,
|
||||
|
||||
@@ -92,6 +92,8 @@ pub struct AppConfig {
|
||||
pub editor_bgfilter_circuit_cooldown: Duration,
|
||||
pub image_editor_agent_sidebar_enabled: bool,
|
||||
pub client_download_channel: String,
|
||||
/// 客户端埋点接收所绑定的公开 origin；未配置时拒绝上传。
|
||||
pub agc_analytics_origin: String,
|
||||
/// AGC 项目快照的部署渠道：上传与后台默认查询都按它分区。
|
||||
pub project_snapshot_channel: String,
|
||||
pub log_filter: String,
|
||||
@@ -404,6 +406,7 @@ impl Default for AppConfig {
|
||||
),
|
||||
image_editor_agent_sidebar_enabled: false,
|
||||
client_download_channel: "dev".to_string(),
|
||||
agc_analytics_origin: String::new(),
|
||||
project_snapshot_channel: "dev".to_string(),
|
||||
log_filter: "info,tower_http=info".to_string(),
|
||||
otel_enabled: false,
|
||||
@@ -727,6 +730,9 @@ impl AppConfig {
|
||||
// 显式空值或非法值也保留，由下载入口失败关闭，不能悄悄改读 dev。
|
||||
config.client_download_channel = channel.trim().to_string();
|
||||
}
|
||||
if let Ok(origin) = std::env::var("GENARRATIVE_AGC_ANALYTICS_ORIGIN") {
|
||||
config.agc_analytics_origin = origin.trim().to_string();
|
||||
}
|
||||
// 快照渠道缺省沿用同一个部署渠道（本部署的客户端渠道），显式配置优先；
|
||||
// 显式空值或非法值同样保留，由快照入口失败关闭。
|
||||
config.project_snapshot_channel = config.client_download_channel.clone();
|
||||
|
||||
@@ -5,6 +5,7 @@ mod admin_accounts;
|
||||
mod admin_project_snapshots;
|
||||
mod admin_recharge;
|
||||
mod admin_templates;
|
||||
mod agc_analytics;
|
||||
mod agc_models;
|
||||
mod ai_tasks;
|
||||
mod aliyun_matting;
|
||||
|
||||
Some files were not shown because too many files have changed in this diff Show More
Reference in New Issue
Block a user