[codex] add K3s update center (#314)

* add k3s update center

* fix update center status timeouts

* Polish update center settings UI

* Document k3s update center

* Refocus update center docs for users

* Clarify k3s migration path in docs
This commit is contained in:
Cita
2026-03-29 17:43:49 +08:00
committed by GitHub
parent dcfd45ea65
commit 11583c51b9
30 changed files with 3819 additions and 7 deletions
+5 -3
View File
@@ -20,6 +20,7 @@ import { monitorRoutes } from './routes/api/monitor.js';
import { downstreamApiKeysRoutes } from './routes/api/downstreamApiKeys.js';
import { oauthRoutes } from './routes/api/oauth.js';
import { siteAnnouncementsRoutes } from './routes/api/siteAnnouncements.js';
import { updateCenterRoutes } from './routes/api/updateCenter.js';
import { proxyRoutes } from './routes/proxy/router.js';
import { startScheduler } from './services/checkinScheduler.js';
import * as routeRefreshWorkflow from './services/routeRefreshWorkflow.js';
@@ -439,9 +440,10 @@ await app.register(authRoutes);
await app.register(settingsRoutes);
await app.register(accountTokensRoutes);
await app.register(searchRoutes);
await app.register(eventsRoutes);
await app.register(siteAnnouncementsRoutes);
await app.register(taskRoutes);
await app.register(eventsRoutes);
await app.register(siteAnnouncementsRoutes);
await app.register(updateCenterRoutes);
await app.register(taskRoutes);
await app.register(testRoutes);
await app.register(monitorRoutes);
await app.register(downstreamApiKeysRoutes);
+356
View File
@@ -0,0 +1,356 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { mkdtempSync } from 'node:fs';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { eq } from 'drizzle-orm';
const {
fetchLatestStableGitHubReleaseMock,
fetchLatestDockerHubTagMock,
getUpdateCenterHelperStatusMock,
streamUpdateCenterDeployMock,
} = vi.hoisted(() => ({
fetchLatestStableGitHubReleaseMock: vi.fn(),
fetchLatestDockerHubTagMock: vi.fn(),
getUpdateCenterHelperStatusMock: vi.fn(),
streamUpdateCenterDeployMock: vi.fn(),
}));
vi.mock('../../services/updateCenterVersionService.js', async () => {
const actual = await vi.importActual<typeof import('../../services/updateCenterVersionService.js')>('../../services/updateCenterVersionService.js');
return {
...actual,
fetchLatestStableGitHubRelease: (...args: unknown[]) => fetchLatestStableGitHubReleaseMock(...args),
fetchLatestDockerHubTag: (...args: unknown[]) => fetchLatestDockerHubTagMock(...args),
};
});
vi.mock('../../services/updateCenterHelperClient.js', () => ({
getUpdateCenterHelperStatus: (...args: unknown[]) => getUpdateCenterHelperStatusMock(...args),
streamUpdateCenterDeploy: (...args: unknown[]) => streamUpdateCenterDeployMock(...args),
}));
type DbModule = typeof import('../../db/index.js');
describe('update center routes', () => {
let app: FastifyInstance;
let db: DbModule['db'];
let schema: DbModule['schema'];
let dataDir = '';
let resetBackgroundTasks: (() => void) | null = null;
let getBackgroundTask: ((taskId: string) => { status: string; logs?: Array<{ message: string }> } | null) | null = null;
async function saveValidConfig() {
const response = await app.inject({
method: 'PUT',
url: '/api/update-center/config',
payload: {
enabled: true,
helperBaseUrl: 'http://metapi-deploy-helper.ai.svc.cluster.local:9850',
namespace: 'ai',
releaseName: 'metapi',
chartRef: 'oci://ghcr.io/cita-777/charts/metapi',
imageRepository: '1467078763/metapi',
githubReleasesEnabled: true,
dockerHubTagsEnabled: true,
defaultDeploySource: 'github-release',
},
});
expect(response.statusCode).toBe(200);
}
beforeAll(async () => {
dataDir = mkdtempSync(join(tmpdir(), 'metapi-update-center-'));
process.env.DATA_DIR = dataDir;
process.env.DEPLOY_HELPER_TOKEN = 'helper-token';
await import('../../db/migrate.js');
const dbModule = await import('../../db/index.js');
const routesModule = await import('./updateCenter.js');
const backgroundTaskModule = await import('../../services/backgroundTaskService.js');
db = dbModule.db;
schema = dbModule.schema;
resetBackgroundTasks = backgroundTaskModule.__resetBackgroundTasksForTests;
getBackgroundTask = backgroundTaskModule.getBackgroundTask;
app = Fastify();
await app.register(routesModule.updateCenterRoutes);
});
beforeEach(async () => {
fetchLatestStableGitHubReleaseMock.mockReset();
fetchLatestDockerHubTagMock.mockReset();
getUpdateCenterHelperStatusMock.mockReset();
streamUpdateCenterDeployMock.mockReset();
resetBackgroundTasks?.();
await db.delete(schema.events).run();
await db.delete(schema.settings).run();
});
afterAll(async () => {
if (app) {
await app.close();
}
delete process.env.DATA_DIR;
delete process.env.DEPLOY_HELPER_TOKEN;
});
it('persists config and returns status with both version channels and helper summary', async () => {
fetchLatestStableGitHubReleaseMock.mockResolvedValue({
source: 'github-release',
rawVersion: 'v1.3.0',
normalizedVersion: '1.3.0',
url: 'https://github.com/cita-777/metapi/releases/tag/v1.3.0',
});
fetchLatestDockerHubTagMock.mockResolvedValue({
source: 'docker-hub-tag',
rawVersion: '1.3.1',
normalizedVersion: '1.3.1',
url: null,
});
getUpdateCenterHelperStatusMock.mockResolvedValue({
ok: true,
releaseName: 'metapi',
namespace: 'ai',
revision: '12',
imageRepository: '1467078763/metapi',
imageTag: '1.2.3',
healthy: true,
});
const saveResponse = await app.inject({
method: 'PUT',
url: '/api/update-center/config',
payload: {
enabled: true,
helperBaseUrl: 'http://metapi-deploy-helper.ai.svc.cluster.local:9850',
namespace: 'ai',
releaseName: 'metapi',
chartRef: 'oci://ghcr.io/cita-777/charts/metapi',
imageRepository: '1467078763/metapi',
githubReleasesEnabled: true,
dockerHubTagsEnabled: true,
defaultDeploySource: 'github-release',
},
});
expect(saveResponse.statusCode).toBe(200);
const savedRow = await db.select().from(schema.settings).where(eq(schema.settings.key, 'update_center_k3s_config_v1')).get();
expect(savedRow?.value).toContain('metapi-deploy-helper.ai.svc.cluster.local');
const statusResponse = await app.inject({
method: 'GET',
url: '/api/update-center/status',
});
expect(statusResponse.statusCode).toBe(200);
expect(statusResponse.json()).toMatchObject({
currentVersion: '1.2.3',
config: {
enabled: true,
namespace: 'ai',
releaseName: 'metapi',
defaultDeploySource: 'github-release',
},
githubRelease: {
normalizedVersion: '1.3.0',
},
dockerHubTag: {
normalizedVersion: '1.3.1',
},
helper: {
ok: true,
healthy: true,
releaseName: 'metapi',
},
});
});
it('returns partial status when a single version source lookup fails', async () => {
fetchLatestStableGitHubReleaseMock.mockRejectedValue(new Error('GitHub releases lookup timed out'));
fetchLatestDockerHubTagMock.mockResolvedValue({
source: 'docker-hub-tag',
rawVersion: '1.3.1',
normalizedVersion: '1.3.1',
url: null,
});
getUpdateCenterHelperStatusMock.mockResolvedValue({
ok: true,
releaseName: 'metapi',
namespace: 'ai',
revision: '12',
imageRepository: '1467078763/metapi',
imageTag: '1.2.3',
healthy: true,
});
await saveValidConfig();
const statusResponse = await app.inject({
method: 'GET',
url: '/api/update-center/status',
});
expect(statusResponse.statusCode).toBe(200);
expect(statusResponse.json()).toMatchObject({
githubRelease: null,
dockerHubTag: {
normalizedVersion: '1.3.1',
},
helper: {
ok: true,
healthy: true,
},
});
});
it('dedupes deploy requests while a task is already running', async () => {
await saveValidConfig();
let releaseDeploy: (() => void) | null = null;
const deployGate = new Promise<void>((resolve) => {
releaseDeploy = resolve;
});
streamUpdateCenterDeployMock.mockImplementation(async (_input: unknown, onLog?: (message: string) => void) => {
onLog?.('Running helm upgrade');
await deployGate;
onLog?.('Deployment complete');
return {
success: true,
targetSource: 'github-release',
targetTag: '1.3.0',
previousRevision: '12',
finalRevision: '13',
rolledBack: false,
logLines: ['Running helm upgrade', 'Deployment complete'],
};
});
const firstResponse = await app.inject({
method: 'POST',
url: '/api/update-center/deploy',
payload: {
source: 'github-release',
targetVersion: '1.3.0',
},
});
const secondResponse = await app.inject({
method: 'POST',
url: '/api/update-center/deploy',
payload: {
source: 'github-release',
targetVersion: '1.3.0',
},
});
expect(firstResponse.statusCode).toBe(202);
expect(secondResponse.statusCode).toBe(202);
const firstBody = firstResponse.json() as { task?: { id: string }; reused?: boolean };
const secondBody = secondResponse.json() as { task?: { id: string }; reused?: boolean };
expect(firstBody.task?.id).toBeTruthy();
expect(secondBody.task?.id).toBe(firstBody.task?.id);
expect(secondBody.reused).toBe(true);
releaseDeploy?.();
});
it('rejects deploy requests when the update center is disabled', async () => {
const saveResponse = await app.inject({
method: 'PUT',
url: '/api/update-center/config',
payload: {
enabled: false,
helperBaseUrl: 'http://metapi-deploy-helper.ai.svc.cluster.local:9850',
namespace: 'ai',
releaseName: 'metapi',
chartRef: 'oci://ghcr.io/cita-777/charts/metapi',
imageRepository: '1467078763/metapi',
githubReleasesEnabled: true,
dockerHubTagsEnabled: true,
defaultDeploySource: 'github-release',
},
});
expect(saveResponse.statusCode).toBe(200);
const deployResponse = await app.inject({
method: 'POST',
url: '/api/update-center/deploy',
payload: {
source: 'github-release',
targetVersion: '1.3.0',
},
});
expect(deployResponse.statusCode).toBe(400);
expect(deployResponse.json()).toMatchObject({
success: false,
message: 'update center is disabled',
});
});
it('streams deployment logs for known tasks and rejects unknown task ids', async () => {
await saveValidConfig();
const missingResponse = await app.inject({
method: 'GET',
url: '/api/update-center/tasks/missing-task/stream',
});
expect(missingResponse.statusCode).toBe(404);
streamUpdateCenterDeployMock.mockImplementation(async (_input: unknown, onLog?: (message: string) => void) => {
onLog?.('Resolving target version');
onLog?.('Waiting for rollout');
return {
success: true,
targetSource: 'docker-hub-tag',
targetTag: '1.3.1',
previousRevision: '13',
finalRevision: '14',
rolledBack: false,
logLines: ['Resolving target version', 'Waiting for rollout'],
};
});
const deployResponse = await app.inject({
method: 'POST',
url: '/api/update-center/deploy',
payload: {
source: 'docker-hub-tag',
targetVersion: '1.3.1',
},
});
const deployBody = deployResponse.json() as { task: { id: string } };
for (let attempt = 0; attempt < 20; attempt += 1) {
const task = getBackgroundTask?.(deployBody.task.id);
if (task?.status === 'succeeded') break;
await new Promise((resolve) => setTimeout(resolve, 0));
}
expect(getBackgroundTask?.(deployBody.task.id)?.logs).toEqual(expect.arrayContaining([
expect.objectContaining({ message: 'Resolving target version' }),
expect.objectContaining({ message: 'Waiting for rollout' }),
]));
const streamResponse = await app.inject({
method: 'GET',
url: `/api/update-center/tasks/${deployBody.task.id}/stream`,
});
expect(streamResponse.statusCode).toBe(200);
expect(streamResponse.headers['content-type']).toContain('text/event-stream');
expect(streamResponse.body).toContain('event: log');
expect(streamResponse.body).toContain('Resolving target version');
expect(streamResponse.body).toContain('Waiting for rollout');
expect(streamResponse.body).toContain('event: done');
});
});
+267
View File
@@ -0,0 +1,267 @@
import { FastifyInstance } from 'fastify';
import {
appendBackgroundTaskLog,
getBackgroundTask,
listBackgroundTasks,
startBackgroundTask,
subscribeToBackgroundTaskLogs,
} from '../../services/backgroundTaskService.js';
import {
fetchLatestDockerHubTag,
fetchLatestStableGitHubRelease,
getCurrentRuntimeVersion,
type UpdateCenterVersionCandidate,
type UpdateCenterVersionSource,
} from '../../services/updateCenterVersionService.js';
import {
getDefaultUpdateCenterConfig,
loadUpdateCenterConfig,
normalizeUpdateCenterConfig,
saveUpdateCenterConfig,
type UpdateCenterConfig,
} from '../../services/updateCenterConfigService.js';
import {
getUpdateCenterHelperStatus,
streamUpdateCenterDeploy,
} from '../../services/updateCenterHelperClient.js';
type DeployBody = {
source?: UpdateCenterVersionSource;
targetVersion?: string;
};
const UPDATE_CENTER_DEPLOY_TASK_TYPE = 'update-center.deploy';
const UPDATE_CENTER_DEPLOY_DEDUPE_KEY = 'update-center.deploy';
function getUpdateCenterHelperToken(): string {
return String(process.env.DEPLOY_HELPER_TOKEN || process.env.UPDATE_CENTER_HELPER_TOKEN || '').trim();
}
function assertDeployableConfig(config: UpdateCenterConfig) {
if (!config.enabled) throw new Error('update center is disabled');
if (!config.helperBaseUrl) throw new Error('helperBaseUrl is required');
if (!config.namespace) throw new Error('namespace is required');
if (!config.releaseName) throw new Error('releaseName is required');
if (!config.chartRef) throw new Error('chartRef is required');
if (!config.imageRepository) throw new Error('imageRepository is required');
if (!getUpdateCenterHelperToken()) throw new Error('DEPLOY_HELPER_TOKEN is required');
}
function getDeployTasks() {
return listBackgroundTasks(50).filter((task) => task.type === UPDATE_CENTER_DEPLOY_TASK_TYPE);
}
function summarizeHelperError(error: unknown) {
if (error instanceof Error && error.message) return error.message;
return String(error || 'unknown helper error');
}
async function settleOptional<T>(enabled: boolean, loader: () => Promise<T>): Promise<{
value: T | null;
error: string | null;
}> {
if (!enabled) {
return {
value: null,
error: null,
};
}
try {
return {
value: await loader(),
error: null,
};
} catch (error) {
return {
value: null,
error: summarizeHelperError(error),
};
}
}
async function buildUpdateCenterStatus() {
const config = await loadUpdateCenterConfig();
const helperToken = getUpdateCenterHelperToken();
const [githubLookup, dockerLookup, helperLookup] = await Promise.all([
settleOptional(config.githubReleasesEnabled, async () => await fetchLatestStableGitHubRelease()),
settleOptional(config.dockerHubTagsEnabled, async () => await fetchLatestDockerHubTag()),
settleOptional(!!config.helperBaseUrl, async () => {
if (!helperToken) {
throw new Error('DEPLOY_HELPER_TOKEN is required');
}
return await getUpdateCenterHelperStatus(config, helperToken);
}),
]);
const githubRelease = githubLookup.value as UpdateCenterVersionCandidate | null;
const dockerHubTag = dockerLookup.value as UpdateCenterVersionCandidate | null;
const helper: Record<string, unknown> = helperLookup.value || {
ok: false,
healthy: false,
error: helperLookup.error,
};
const tasks = getDeployTasks();
const runningTask = tasks.find((task) => task.status === 'pending' || task.status === 'running') || null;
const lastFinishedTask = tasks.find((task) => task.status === 'succeeded' || task.status === 'failed') || null;
return {
currentVersion: getCurrentRuntimeVersion(),
config,
githubRelease,
dockerHubTag,
helper,
runningTask,
lastFinishedTask,
};
}
function writeSseEvent(reply: { raw: NodeJS.WritableStream & { write: (chunk: string) => void } }, event: string, data: unknown) {
reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
}
export async function updateCenterRoutes(app: FastifyInstance) {
app.get('/api/update-center/status', async () => {
return await buildUpdateCenterStatus();
});
app.post('/api/update-center/check', async () => {
return await buildUpdateCenterStatus();
});
app.put<{ Body: UpdateCenterConfig }>('/api/update-center/config', async (request) => {
const next = normalizeUpdateCenterConfig(request.body || getDefaultUpdateCenterConfig());
const saved = await saveUpdateCenterConfig(next);
return {
success: true,
config: saved,
};
});
app.post<{ Body: DeployBody }>('/api/update-center/deploy', async (request, reply) => {
const config = await loadUpdateCenterConfig();
try {
assertDeployableConfig(config);
} catch (error) {
return reply.code(400).send({
success: false,
message: summarizeHelperError(error),
});
}
const source = request.body?.source === 'docker-hub-tag'
? 'docker-hub-tag'
: request.body?.source === 'github-release'
? 'github-release'
: config.defaultDeploySource;
const targetVersion = String(request.body?.targetVersion || '').trim();
if (!targetVersion) {
return reply.code(400).send({
success: false,
message: 'targetVersion is required',
});
}
let taskId = '';
const { task, reused } = startBackgroundTask(
{
type: UPDATE_CENTER_DEPLOY_TASK_TYPE,
title: '更新中心部署',
dedupeKey: UPDATE_CENTER_DEPLOY_DEDUPE_KEY,
successTitle: '更新中心部署已完成',
failureTitle: '更新中心部署失败',
},
async () => {
await Promise.resolve();
appendBackgroundTaskLog(taskId, `Resolving target version: ${targetVersion}`);
appendBackgroundTaskLog(taskId, `Contacting deploy helper: ${config.helperBaseUrl}`);
const result = await streamUpdateCenterDeploy(
{
config,
helperToken: getUpdateCenterHelperToken(),
source,
targetVersion,
},
(message) => {
appendBackgroundTaskLog(taskId, message);
},
);
if (!result.success) {
throw new Error('deploy helper reported a failed deployment');
}
appendBackgroundTaskLog(taskId, result.rolledBack ? 'Deployment rolled back' : 'Deployment finished successfully');
return result;
},
);
taskId = task.id;
return reply.code(202).send({
success: true,
reused,
task,
});
});
app.get<{ Params: { id: string } }>('/api/update-center/tasks/:id/stream', async (request, reply) => {
const taskId = String(request.params.id || '').trim();
const task = getBackgroundTask(taskId);
if (!task) {
return reply.code(404).send({
success: false,
message: 'task not found',
});
}
reply.hijack();
reply.raw.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
reply.raw.setHeader('Cache-Control', 'no-cache, no-transform');
reply.raw.setHeader('Connection', 'keep-alive');
const writeDone = () => {
const latest = getBackgroundTask(taskId);
writeSseEvent(reply, 'done', {
status: latest?.status || 'unknown',
});
reply.raw.end();
};
for (const entry of task.logs) {
writeSseEvent(reply, 'log', entry);
}
if (task.status !== 'pending' && task.status !== 'running') {
writeDone();
return;
}
const unsubscribe = subscribeToBackgroundTaskLogs(taskId, (entry) => {
writeSseEvent(reply, 'log', entry);
});
const interval = setInterval(() => {
const latest = getBackgroundTask(taskId);
if (!latest) {
clearInterval(interval);
unsubscribe();
writeDone();
return;
}
if (latest.status !== 'pending' && latest.status !== 'running') {
clearInterval(interval);
unsubscribe();
writeDone();
}
}, 25);
interval.unref?.();
request.raw.on('close', () => {
clearInterval(interval);
unsubscribe();
});
});
}
@@ -0,0 +1,113 @@
import { afterEach, describe, expect, it } from 'vitest';
import {
__resetBackgroundTasksForTests,
appendBackgroundTaskLog,
getBackgroundTask,
startBackgroundTask,
subscribeToBackgroundTaskLogs,
} from './backgroundTaskService.js';
describe('background task log streaming', () => {
afterEach(() => {
__resetBackgroundTasksForTests();
});
it('appends log entries in order and exposes them through task lookups', async () => {
let releaseRunner: (() => void) | null = null;
const runnerGate = new Promise<void>((resolve) => {
releaseRunner = resolve;
});
const { task } = startBackgroundTask(
{
type: 'update-center.deploy',
title: '更新部署',
},
async () => {
await runnerGate;
return { success: true };
},
);
appendBackgroundTaskLog(task.id, 'Resolving target version');
appendBackgroundTaskLog(task.id, 'Running helm upgrade');
const currentTask = getBackgroundTask(task.id);
expect(currentTask?.logs).toEqual([
expect.objectContaining({ seq: 1, message: 'Resolving target version' }),
expect.objectContaining({ seq: 2, message: 'Running helm upgrade' }),
]);
releaseRunner?.();
});
it('notifies subscribers when new log entries arrive', async () => {
let releaseRunner: (() => void) | null = null;
const runnerGate = new Promise<void>((resolve) => {
releaseRunner = resolve;
});
const { task } = startBackgroundTask(
{
type: 'update-center.deploy',
title: '更新部署',
},
async () => {
await runnerGate;
return { success: true };
},
);
const received: string[] = [];
const unsubscribe = subscribeToBackgroundTaskLogs(task.id, (entry) => {
received.push(entry.message);
});
appendBackgroundTaskLog(task.id, 'Waiting for rollout');
appendBackgroundTaskLog(task.id, 'Deployment complete');
expect(received).toEqual([
'Waiting for rollout',
'Deployment complete',
]);
unsubscribe();
releaseRunner?.();
});
it('trims old log entries to a bounded buffer', async () => {
let releaseRunner: (() => void) | null = null;
const runnerGate = new Promise<void>((resolve) => {
releaseRunner = resolve;
});
const { task } = startBackgroundTask(
{
type: 'update-center.deploy',
title: '更新部署',
},
async () => {
await runnerGate;
return { success: true };
},
);
for (let index = 1; index <= 250; index += 1) {
appendBackgroundTaskLog(task.id, `line-${index}`);
}
const currentTask = getBackgroundTask(task.id);
expect(currentTask?.logs).toHaveLength(200);
expect(currentTask?.logs[0]).toMatchObject({
seq: 51,
message: 'line-51',
});
expect(currentTask?.logs.at(-1)).toMatchObject({
seq: 250,
message: 'line-250',
});
releaseRunner?.();
});
});
+78 -1
View File
@@ -4,6 +4,12 @@ import { sendNotification } from './notifyService.js';
export type BackgroundTaskStatus = 'pending' | 'running' | 'succeeded' | 'failed';
export type BackgroundTaskLogEntry = {
seq: number;
message: string;
createdAt: string;
};
export type BackgroundTask = {
id: string;
type: string;
@@ -18,6 +24,7 @@ export type BackgroundTask = {
startedAt: string | null;
finishedAt: string | null;
expiresAtMs: number;
logs: BackgroundTaskLogEntry[];
};
type TaskMessageTemplate = string | ((task: BackgroundTask) => string);
@@ -37,9 +44,12 @@ type BackgroundTaskStartOptions = {
const TASK_TTL_MS = 6 * 60 * 60 * 1000;
const TASK_CLEANUP_INTERVAL_MS = 60 * 1000;
const TASK_LOG_LIMIT = 200;
const tasks = new Map<string, BackgroundTask>();
const dedupeTaskIds = new Map<string, string>();
const taskLogSeq = new Map<string, number>();
const taskLogSubscribers = new Map<string, Set<(entry: BackgroundTaskLogEntry) => void>>();
let cleanupTimer: ReturnType<typeof setInterval> | null = null;
function nowIso() {
@@ -72,8 +82,9 @@ function resolveTaskMessage(template: TaskMessageTemplate | undefined, task: Bac
}
function setTaskStatus(task: BackgroundTask, patch: Partial<BackgroundTask>) {
const currentTask = tasks.get(task.id) || task;
const next: BackgroundTask = {
...task,
...currentTask,
...patch,
updatedAt: nowIso(),
};
@@ -81,6 +92,67 @@ function setTaskStatus(task: BackgroundTask, patch: Partial<BackgroundTask>) {
return next;
}
function cleanupTaskInternals(taskId: string) {
taskLogSeq.delete(taskId);
taskLogSubscribers.delete(taskId);
}
export function appendBackgroundTaskLog(taskId: string, message: string): BackgroundTaskLogEntry | null {
const task = tasks.get(taskId);
const normalizedMessage = String(message || '').trim();
if (!task || !normalizedMessage) return null;
const nextSeq = (taskLogSeq.get(taskId) || 0) + 1;
taskLogSeq.set(taskId, nextSeq);
const entry: BackgroundTaskLogEntry = {
seq: nextSeq,
message: normalizedMessage,
createdAt: nowIso(),
};
const nextLogs = [...task.logs, entry];
const trimmedLogs = nextLogs.length > TASK_LOG_LIMIT
? nextLogs.slice(nextLogs.length - TASK_LOG_LIMIT)
: nextLogs;
tasks.set(taskId, {
...task,
logs: trimmedLogs,
updatedAt: nowIso(),
});
const subscribers = taskLogSubscribers.get(taskId);
if (subscribers) {
for (const subscriber of subscribers) {
subscriber(entry);
}
}
return entry;
}
export function subscribeToBackgroundTaskLogs(
taskId: string,
listener: (entry: BackgroundTaskLogEntry) => void,
): () => void {
let subscribers = taskLogSubscribers.get(taskId);
if (!subscribers) {
subscribers = new Set();
taskLogSubscribers.set(taskId, subscribers);
}
subscribers.add(listener);
return () => {
const current = taskLogSubscribers.get(taskId);
if (!current) return;
current.delete(listener);
if (current.size <= 0) {
taskLogSubscribers.delete(taskId);
}
};
}
async function appendTaskEvent(level: 'info' | 'warning' | 'error', title: string, message: string, taskId: string) {
try {
await db.insert(schema.events).values({
@@ -154,6 +226,7 @@ function cleanupExpiredTasks() {
if (task.dedupeKey && dedupeTaskIds.get(task.dedupeKey) === taskId) {
dedupeTaskIds.delete(task.dedupeKey);
}
cleanupTaskInternals(taskId);
}
}
}
@@ -196,9 +269,11 @@ export function startBackgroundTask(
startedAt: null,
finishedAt: null,
expiresAtMs: Date.now() + Math.max(60_000, options.keepMs ?? TASK_TTL_MS),
logs: [],
};
tasks.set(task.id, task);
taskLogSeq.set(task.id, 0);
if (dedupeKey) dedupeTaskIds.set(dedupeKey, task.id);
appendTaskEvent('info', `${task.title}已开始`, `${task.title} 已开始执行`, task.id);
@@ -246,6 +321,8 @@ export function summarizeCheckinResults(results: Array<{ result?: any }>): { tot
export function __resetBackgroundTasksForTests() {
tasks.clear();
dedupeTaskIds.clear();
taskLogSeq.clear();
taskLogSubscribers.clear();
if (cleanupTimer) {
clearInterval(cleanupTimer);
cleanupTimer = null;
@@ -0,0 +1,81 @@
import { eq } from 'drizzle-orm';
import { db, schema } from '../db/index.js';
import { upsertSetting } from '../db/upsertSetting.js';
import type { UpdateCenterVersionSource } from './updateCenterVersionService.js';
export type UpdateCenterConfig = {
enabled: boolean;
helperBaseUrl: string;
namespace: string;
releaseName: string;
chartRef: string;
imageRepository: string;
githubReleasesEnabled: boolean;
dockerHubTagsEnabled: boolean;
defaultDeploySource: UpdateCenterVersionSource;
};
export const UPDATE_CENTER_CONFIG_SETTING_KEY = 'update_center_k3s_config_v1';
export function getDefaultUpdateCenterConfig(): UpdateCenterConfig {
return {
enabled: false,
helperBaseUrl: '',
namespace: 'default',
releaseName: '',
chartRef: '',
imageRepository: '1467078763/metapi',
githubReleasesEnabled: true,
dockerHubTagsEnabled: true,
defaultDeploySource: 'github-release',
};
}
function normalizeString(value: unknown, fallback = ''): string {
if (typeof value !== 'string') return fallback;
return value.trim();
}
function normalizeBoolean(value: unknown, fallback: boolean): boolean {
return typeof value === 'boolean' ? value : fallback;
}
export function normalizeUpdateCenterConfig(input: unknown): UpdateCenterConfig {
const defaults = getDefaultUpdateCenterConfig();
const record = input && typeof input === 'object' ? input as Record<string, unknown> : {};
const defaultDeploySource = record.defaultDeploySource === 'docker-hub-tag'
? 'docker-hub-tag'
: 'github-release';
return {
enabled: normalizeBoolean(record.enabled, defaults.enabled),
helperBaseUrl: normalizeString(record.helperBaseUrl, defaults.helperBaseUrl),
namespace: normalizeString(record.namespace, defaults.namespace) || defaults.namespace,
releaseName: normalizeString(record.releaseName, defaults.releaseName),
chartRef: normalizeString(record.chartRef, defaults.chartRef),
imageRepository: normalizeString(record.imageRepository, defaults.imageRepository) || defaults.imageRepository,
githubReleasesEnabled: normalizeBoolean(record.githubReleasesEnabled, defaults.githubReleasesEnabled),
dockerHubTagsEnabled: normalizeBoolean(record.dockerHubTagsEnabled, defaults.dockerHubTagsEnabled),
defaultDeploySource,
};
}
export async function loadUpdateCenterConfig(): Promise<UpdateCenterConfig> {
const row = await db.select().from(schema.settings).where(eq(schema.settings.key, UPDATE_CENTER_CONFIG_SETTING_KEY)).get();
if (!row?.value) {
return getDefaultUpdateCenterConfig();
}
try {
return normalizeUpdateCenterConfig(JSON.parse(row.value));
} catch {
return getDefaultUpdateCenterConfig();
}
}
export async function saveUpdateCenterConfig(input: unknown): Promise<UpdateCenterConfig> {
const next = normalizeUpdateCenterConfig(input);
await upsertSetting(UPDATE_CENTER_CONFIG_SETTING_KEY, next);
return next;
}
@@ -0,0 +1,39 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
const { fetchMock } = vi.hoisted(() => ({
fetchMock: vi.fn(),
}));
vi.mock('undici', async () => {
const actual = await vi.importActual<typeof import('undici')>('undici');
return {
...actual,
fetch: (...args: unknown[]) => fetchMock(...args),
};
});
import { getUpdateCenterHelperStatus } from './updateCenterHelperClient.js';
describe('update center helper client', () => {
beforeEach(() => {
fetchMock.mockReset();
});
it('maps helper status aborts to a timeout error', async () => {
const abortError = new Error('aborted');
(abortError as Error & { name: string }).name = 'AbortError';
fetchMock.mockRejectedValue(abortError);
await expect(getUpdateCenterHelperStatus({
enabled: true,
helperBaseUrl: 'http://metapi-deploy-helper.ai.svc.cluster.local:9850',
namespace: 'ai',
releaseName: 'metapi',
chartRef: 'oci://ghcr.io/cita-777/charts/metapi',
imageRepository: '1467078763/metapi',
githubReleasesEnabled: true,
dockerHubTagsEnabled: true,
defaultDeploySource: 'github-release',
}, 'helper-token')).rejects.toThrow('deploy helper status timeout');
});
});
@@ -0,0 +1,192 @@
import { fetch, type RequestInit as UndiciRequestInit } from 'undici';
import type { UpdateCenterConfig } from './updateCenterConfigService.js';
import type { UpdateCenterVersionSource } from './updateCenterVersionService.js';
const UPDATE_CENTER_HELPER_STATUS_TIMEOUT_MS = 5_000;
export type UpdateCenterHelperStatus = {
ok: boolean;
releaseName: string | null;
namespace: string | null;
revision: string | null;
imageRepository: string | null;
imageTag: string | null;
healthy: boolean;
error?: string;
};
export type UpdateCenterDeploySummary = {
success: boolean;
targetSource: UpdateCenterVersionSource;
targetTag: string;
previousRevision: string | null;
finalRevision: string | null;
rolledBack: boolean;
logLines: string[];
};
type HelperDeployLogEvent = {
message: string;
};
async function fetchHelperJsonWithTimeout(url: string, init: UndiciRequestInit): Promise<unknown> {
const controller = new AbortController();
let timeoutHandle: ReturnType<typeof setTimeout> | null = setTimeout(() => {
controller.abort();
}, UPDATE_CENTER_HELPER_STATUS_TIMEOUT_MS);
try {
const response = await fetch(url, {
...init,
signal: controller.signal,
});
if (!response.ok) {
throw new Error(`helper status failed with HTTP ${response.status}`);
}
return await response.json();
} catch (error: any) {
if (error?.name === 'AbortError') {
throw new Error(`deploy helper status timeout (${Math.round(UPDATE_CENTER_HELPER_STATUS_TIMEOUT_MS / 1000)}s)`);
}
throw error;
} finally {
if (timeoutHandle) {
clearTimeout(timeoutHandle);
timeoutHandle = null;
}
}
}
function buildHelperHeaders(token: string, accept = 'application/json') {
return {
authorization: `Bearer ${token}`,
accept,
};
}
function encodeSseEvent(event: string, data: unknown) {
return `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
}
type SseChunkHandler = (event: string, data: unknown) => void;
function parseSseBuffer(buffer: string, onChunk: SseChunkHandler) {
const events = buffer.split('\n\n');
const remainder = events.pop() || '';
for (const block of events) {
const lines = block.split('\n');
let eventName = 'message';
const dataLines: string[] = [];
for (const line of lines) {
if (line.startsWith('event:')) {
eventName = line.slice('event:'.length).trim() || 'message';
} else if (line.startsWith('data:')) {
dataLines.push(line.slice('data:'.length).trim());
}
}
if (dataLines.length <= 0) continue;
const rawData = dataLines.join('\n');
try {
onChunk(eventName, JSON.parse(rawData));
} catch {
onChunk(eventName, rawData);
}
}
return remainder;
}
export async function getUpdateCenterHelperStatus(
config: UpdateCenterConfig,
helperToken: string,
): Promise<UpdateCenterHelperStatus> {
const query = new URLSearchParams({
namespace: config.namespace,
releaseName: config.releaseName,
});
return await fetchHelperJsonWithTimeout(`${config.helperBaseUrl.replace(/\/$/, '')}/status?${query.toString()}`, {
headers: buildHelperHeaders(helperToken),
}) as UpdateCenterHelperStatus;
}
export async function streamUpdateCenterDeploy(
input: {
config: UpdateCenterConfig;
helperToken: string;
source: UpdateCenterVersionSource;
targetVersion: string;
},
onLog?: (message: string) => void,
): Promise<UpdateCenterDeploySummary> {
const response = await fetch(`${input.config.helperBaseUrl.replace(/\/$/, '')}/deploy`, {
method: 'POST',
headers: {
...buildHelperHeaders(input.helperToken, 'text/event-stream'),
'content-type': 'application/json',
},
body: JSON.stringify({
namespace: input.config.namespace,
releaseName: input.config.releaseName,
chartRef: input.config.chartRef,
imageRepository: input.config.imageRepository,
source: input.source,
targetVersion: input.targetVersion,
}),
});
if (!response.ok) {
throw new Error(`helper deploy failed with HTTP ${response.status}`);
}
if (!response.body) {
throw new Error('helper deploy response did not include a stream body');
}
const decoder = new TextDecoder();
const reader = response.body.getReader();
let buffer = '';
let finalResult: UpdateCenterDeploySummary | null = null;
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
buffer = parseSseBuffer(buffer, (event, data) => {
if (event === 'log' && data && typeof data === 'object') {
const message = String((data as HelperDeployLogEvent).message || '').trim();
if (message) onLog?.(message);
return;
}
if (event === 'result') {
finalResult = data as UpdateCenterDeploySummary;
}
});
}
if (buffer.trim()) {
parseSseBuffer(`${buffer}\n\n`, (event, data) => {
if (event === 'log' && data && typeof data === 'object') {
const message = String((data as HelperDeployLogEvent).message || '').trim();
if (message) onLog?.(message);
return;
}
if (event === 'result') {
finalResult = data as UpdateCenterDeploySummary;
}
});
}
if (!finalResult) {
throw new Error('helper deploy stream ended without a result event');
}
return finalResult;
}
export const __updateCenterHelperClientTestUtils = {
encodeSseEvent,
};
@@ -0,0 +1,224 @@
import { describe, expect, it, vi, beforeEach } from 'vitest';
const { fetchMock } = vi.hoisted(() => ({
fetchMock: vi.fn(),
}));
vi.mock('undici', async () => {
const actual = await vi.importActual<typeof import('undici')>('undici');
return {
...actual,
fetch: (...args: unknown[]) => fetchMock(...args),
};
});
import {
compareStableSemVer,
fetchLatestDockerHubTag,
fetchLatestStableGitHubRelease,
parseStableSemVer,
resolvePreferredDeploySource,
selectLatestDockerHubTag,
selectLatestStableGitHubRelease,
} from './updateCenterVersionService.js';
describe('update center version service', () => {
beforeEach(() => {
fetchMock.mockReset();
});
describe('parseStableSemVer', () => {
it('accepts stable semver strings with optional leading v', () => {
expect(parseStableSemVer('1.2.3')).toMatchObject({
raw: '1.2.3',
normalized: '1.2.3',
major: 1,
minor: 2,
patch: 3,
});
expect(parseStableSemVer('v2.4.6')).toMatchObject({
raw: 'v2.4.6',
normalized: '2.4.6',
major: 2,
minor: 4,
patch: 6,
});
});
it('rejects prerelease and invalid versions', () => {
expect(parseStableSemVer('v1.2.3-alpha.1')).toBeNull();
expect(parseStableSemVer('latest')).toBeNull();
expect(parseStableSemVer('2026-03-29')).toBeNull();
expect(parseStableSemVer('1.2')).toBeNull();
});
});
describe('compareStableSemVer', () => {
it('compares semver numerically instead of lexicographically', () => {
const v120 = parseStableSemVer('1.2.0');
const v1100 = parseStableSemVer('1.10.0');
const v200 = parseStableSemVer('2.0.0');
expect(v120).not.toBeNull();
expect(v1100).not.toBeNull();
expect(v200).not.toBeNull();
expect(compareStableSemVer(v120!, v1100!)).toBeLessThan(0);
expect(compareStableSemVer(v200!, v1100!)).toBeGreaterThan(0);
});
});
describe('selectLatestStableGitHubRelease', () => {
it('ignores drafts and prereleases and returns the highest stable semver release', () => {
const latest = selectLatestStableGitHubRelease([
{
tag_name: 'v1.2.3-alpha.1',
html_url: 'https://example.com/pre',
draft: false,
prerelease: true,
published_at: '2026-03-29T00:00:00Z',
name: 'v1.2.3-alpha.1',
},
{
tag_name: 'v1.9.9',
html_url: 'https://example.com/older',
draft: false,
prerelease: false,
published_at: '2026-03-28T00:00:00Z',
name: 'v1.9.9',
},
{
tag_name: 'v1.10.0',
html_url: 'https://example.com/latest',
draft: false,
prerelease: false,
published_at: '2026-03-27T00:00:00Z',
name: 'v1.10.0',
},
{
tag_name: 'v9.9.9',
html_url: 'https://example.com/draft',
draft: true,
prerelease: false,
published_at: '2026-03-30T00:00:00Z',
name: 'v9.9.9',
},
]);
expect(latest).toMatchObject({
rawVersion: 'v1.10.0',
normalizedVersion: '1.10.0',
source: 'github-release',
url: 'https://example.com/latest',
});
});
});
describe('selectLatestDockerHubTag', () => {
it('ignores non-semver tags and prereleases', () => {
const latest = selectLatestDockerHubTag([
'latest',
'main',
'v1.2.3-beta.1',
'v1.2.3',
'1.10.0',
'20260329',
]);
expect(latest).toMatchObject({
rawVersion: '1.10.0',
normalizedVersion: '1.10.0',
source: 'docker-hub-tag',
});
});
});
describe('resolvePreferredDeploySource', () => {
it('prefers the configured default source when both channels have valid candidates', () => {
const preferred = resolvePreferredDeploySource({
defaultSource: 'github-release',
githubRelease: {
source: 'github-release',
rawVersion: 'v1.10.0',
normalizedVersion: '1.10.0',
url: 'https://example.com/release',
},
dockerHubTag: {
source: 'docker-hub-tag',
rawVersion: '1.10.1',
normalizedVersion: '1.10.1',
url: null,
},
});
expect(preferred?.source).toBe('github-release');
expect(preferred?.normalizedVersion).toBe('1.10.0');
});
});
describe('fetchLatestStableGitHubRelease', () => {
it('fetches releases and returns the latest stable semver record', async () => {
fetchMock.mockResolvedValue(new Response(JSON.stringify([
{
tag_name: 'v1.2.3-beta.1',
html_url: 'https://example.com/beta',
draft: false,
prerelease: true,
},
{
tag_name: 'v1.2.9',
html_url: 'https://example.com/older',
draft: false,
prerelease: false,
},
{
tag_name: 'v1.10.0',
html_url: 'https://example.com/latest',
draft: false,
prerelease: false,
},
]), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const latest = await fetchLatestStableGitHubRelease();
expect(latest).toMatchObject({
source: 'github-release',
normalizedVersion: '1.10.0',
url: 'https://example.com/latest',
});
expect(String(fetchMock.mock.calls[0]?.[0] || '')).toContain('/repos/cita-777/metapi/releases');
});
it('maps aborted GitHub lookups to a timeout error', async () => {
const abortError = new Error('aborted');
(abortError as Error & { name: string }).name = 'AbortError';
fetchMock.mockRejectedValue(abortError);
await expect(fetchLatestStableGitHubRelease()).rejects.toThrow('GitHub releases lookup timeout');
});
});
describe('fetchLatestDockerHubTag', () => {
it('fetches tags and returns the latest semver tag only', async () => {
fetchMock.mockResolvedValue(new Response(JSON.stringify({
results: [
{ name: 'latest' },
{ name: 'v1.2.3-rc.1' },
{ name: 'v1.2.3' },
{ name: '1.10.0' },
],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const latest = await fetchLatestDockerHubTag();
expect(latest).toMatchObject({
source: 'docker-hub-tag',
normalizedVersion: '1.10.0',
});
expect(String(fetchMock.mock.calls[0]?.[0] || '')).toContain('/v2/repositories/1467078763/metapi/tags');
});
});
});
@@ -0,0 +1,176 @@
import { fetch, type RequestInit as UndiciRequestInit } from 'undici';
import { readFileSync } from 'node:fs';
import { resolve } from 'node:path';
export type StableSemVer = {
raw: string;
normalized: string;
major: number;
minor: number;
patch: number;
};
export type UpdateCenterVersionSource = 'github-release' | 'docker-hub-tag';
export type UpdateCenterVersionCandidate = {
source: UpdateCenterVersionSource;
rawVersion: string;
normalizedVersion: string;
url: string | null;
};
export type GitHubReleaseRecord = {
tag_name?: string | null;
html_url?: string | null;
draft?: boolean;
prerelease?: boolean;
published_at?: string | null;
name?: string | null;
};
const STABLE_SEMVER_PATTERN = /^v?(\d+)\.(\d+)\.(\d+)(?:\+[\w.-]+)?$/i;
const GITHUB_RELEASES_URL = 'https://api.github.com/repos/cita-777/metapi/releases';
const DOCKER_HUB_TAGS_URL = 'https://hub.docker.com/v2/repositories/1467078763/metapi/tags?page_size=100';
const UPDATE_CENTER_VERSION_FETCH_TIMEOUT_MS = 5_000;
async function fetchJsonWithTimeout(url: string, init: UndiciRequestInit, timeoutLabel: string): Promise<unknown> {
const controller = new AbortController();
let timeoutHandle: ReturnType<typeof setTimeout> | null = setTimeout(() => {
controller.abort();
}, UPDATE_CENTER_VERSION_FETCH_TIMEOUT_MS);
try {
const response = await fetch(url, {
...init,
signal: controller.signal,
});
if (!response.ok) {
throw new Error(`${timeoutLabel} failed with HTTP ${response.status}`);
}
return await response.json();
} catch (error: any) {
if (error?.name === 'AbortError') {
throw new Error(`${timeoutLabel} timeout (${Math.round(UPDATE_CENTER_VERSION_FETCH_TIMEOUT_MS / 1000)}s)`);
}
throw error;
} finally {
if (timeoutHandle) {
clearTimeout(timeoutHandle);
timeoutHandle = null;
}
}
}
export function parseStableSemVer(input: string | null | undefined): StableSemVer | null {
const raw = String(input || '').trim();
if (!raw) return null;
const match = raw.match(STABLE_SEMVER_PATTERN);
if (!match) return null;
const major = Number.parseInt(match[1], 10);
const minor = Number.parseInt(match[2], 10);
const patch = Number.parseInt(match[3], 10);
if (![major, minor, patch].every(Number.isFinite)) return null;
return {
raw,
normalized: `${major}.${minor}.${patch}`,
major,
minor,
patch,
};
}
export function compareStableSemVer(a: StableSemVer, b: StableSemVer): number {
if (a.major !== b.major) return a.major - b.major;
if (a.minor !== b.minor) return a.minor - b.minor;
return a.patch - b.patch;
}
export function selectLatestStableGitHubRelease(
releases: GitHubReleaseRecord[],
): UpdateCenterVersionCandidate | null {
let selected: { semver: StableSemVer; release: GitHubReleaseRecord } | null = null;
for (const release of releases) {
if (release?.draft || release?.prerelease) continue;
const semver = parseStableSemVer(release?.tag_name);
if (!semver) continue;
if (!selected || compareStableSemVer(semver, selected.semver) > 0) {
selected = { semver, release };
}
}
if (!selected) return null;
return {
source: 'github-release',
rawVersion: selected.release.tag_name || selected.semver.raw,
normalizedVersion: selected.semver.normalized,
url: selected.release.html_url || null,
};
}
export function selectLatestDockerHubTag(tags: string[]): UpdateCenterVersionCandidate | null {
let selected: StableSemVer | null = null;
for (const tag of tags) {
const semver = parseStableSemVer(tag);
if (!semver) continue;
if (!selected || compareStableSemVer(semver, selected) > 0) {
selected = semver;
}
}
if (!selected) return null;
return {
source: 'docker-hub-tag',
rawVersion: selected.raw,
normalizedVersion: selected.normalized,
url: null,
};
}
export function resolvePreferredDeploySource(input: {
defaultSource: UpdateCenterVersionSource;
githubRelease: UpdateCenterVersionCandidate | null;
dockerHubTag: UpdateCenterVersionCandidate | null;
}): UpdateCenterVersionCandidate | null {
if (input.defaultSource === 'github-release') {
return input.githubRelease || input.dockerHubTag;
}
return input.dockerHubTag || input.githubRelease;
}
export async function fetchLatestStableGitHubRelease(): Promise<UpdateCenterVersionCandidate | null> {
const releases = await fetchJsonWithTimeout(GITHUB_RELEASES_URL, {
headers: {
accept: 'application/vnd.github+json',
'user-agent': 'metapi-update-center/1.0',
},
}, 'GitHub releases lookup') as GitHubReleaseRecord[];
return selectLatestStableGitHubRelease(Array.isArray(releases) ? releases : []);
}
export async function fetchLatestDockerHubTag(): Promise<UpdateCenterVersionCandidate | null> {
const payload = await fetchJsonWithTimeout(DOCKER_HUB_TAGS_URL, {
headers: {
accept: 'application/json',
'user-agent': 'metapi-update-center/1.0',
},
}, 'Docker Hub tag lookup') as { results?: Array<{ name?: string | null }> };
const tags = Array.isArray(payload?.results)
? payload.results.map((item) => String(item?.name || '')).filter(Boolean)
: [];
return selectLatestDockerHubTag(tags);
}
export function getCurrentRuntimeVersion(): string {
try {
const packageJsonPath = resolve(process.cwd(), 'package.json');
const payload = JSON.parse(readFileSync(packageJsonPath, 'utf8')) as { version?: string };
const version = String(payload?.version || '').trim();
return version || '0.0.0';
} catch {
return '0.0.0';
}
}
+129
View File
@@ -0,0 +1,129 @@
import { afterEach, describe, expect, it, vi } from 'vitest';
import { buildUpdateHelperApp } from './app.js';
describe('update helper app', () => {
const apps: Array<Awaited<ReturnType<typeof buildUpdateHelperApp>>> = [];
afterEach(async () => {
await Promise.all(apps.splice(0).map((app) => app.close()));
});
it('requires bearer auth for protected helper routes', async () => {
const app = await buildUpdateHelperApp({
token: 'helper-token',
getStatus: vi.fn(),
deploy: vi.fn(),
});
apps.push(app);
const response = await app.inject({
method: 'GET',
url: '/status?namespace=ai&releaseName=metapi',
});
expect(response.statusCode).toBe(401);
});
it('allows unauthenticated health checks', async () => {
const app = await buildUpdateHelperApp({
token: 'helper-token',
getStatus: vi.fn(),
deploy: vi.fn(),
});
apps.push(app);
const response = await app.inject({
method: 'GET',
url: '/health',
});
expect(response.statusCode).toBe(200);
expect(response.json()).toEqual({ ok: true });
});
it('returns helper status for authorized requests', async () => {
const getStatus = vi.fn().mockResolvedValue({
ok: true,
releaseName: 'metapi',
namespace: 'ai',
revision: '17',
imageRepository: '1467078763/metapi',
imageTag: '1.3.1',
healthy: true,
});
const app = await buildUpdateHelperApp({
token: 'helper-token',
getStatus,
deploy: vi.fn(),
});
apps.push(app);
const response = await app.inject({
method: 'GET',
url: '/status?namespace=ai&releaseName=metapi',
headers: {
authorization: 'Bearer helper-token',
},
});
expect(response.statusCode).toBe(200);
expect(response.json()).toMatchObject({
healthy: true,
releaseName: 'metapi',
namespace: 'ai',
});
expect(getStatus).toHaveBeenCalledWith({
namespace: 'ai',
releaseName: 'metapi',
});
});
it('streams deploy logs and final result as SSE when requested', async () => {
const deploy = vi.fn().mockImplementation(async (input, onLog) => {
onLog?.('Running helm upgrade');
onLog?.('Waiting for rollout');
return {
success: true,
targetSource: input.targetSource,
targetTag: input.targetTag,
previousRevision: '17',
finalRevision: '18',
rolledBack: false,
logLines: ['Running helm upgrade', 'Waiting for rollout'],
};
});
const app = await buildUpdateHelperApp({
token: 'helper-token',
getStatus: vi.fn(),
deploy,
});
apps.push(app);
const response = await app.inject({
method: 'POST',
url: '/deploy',
headers: {
authorization: 'Bearer helper-token',
accept: 'text/event-stream',
},
payload: {
namespace: 'ai',
releaseName: 'metapi',
chartRef: 'oci://ghcr.io/cita-777/charts/metapi',
imageRepository: '1467078763/metapi',
source: 'github-release',
targetVersion: '1.3.0',
},
});
expect(response.statusCode).toBe(200);
expect(response.headers['content-type']).toContain('text/event-stream');
expect(response.body).toContain('event: log');
expect(response.body).toContain('Running helm upgrade');
expect(response.body).toContain('event: result');
expect(response.body).toContain('"targetTag":"1.3.0"');
});
});
+100
View File
@@ -0,0 +1,100 @@
import Fastify, { type FastifyInstance } from 'fastify';
import {
executeUpdateHelperDeploy,
getUpdateHelperStatus,
type UpdateHelperDeployInput,
type UpdateHelperDeploySummary,
type UpdateHelperStatus,
} from './service.js';
type BuildUpdateHelperAppOptions = {
token: string;
getStatus?: (input: { namespace: string; releaseName: string }) => Promise<UpdateHelperStatus>;
deploy?: (
input: UpdateHelperDeployInput,
onLog?: (message: string) => void,
) => Promise<UpdateHelperDeploySummary>;
};
function requireBearerToken(requestAuthHeader: string | undefined, expectedToken: string) {
const raw = String(requestAuthHeader || '').trim();
if (!raw.startsWith('Bearer ')) return false;
return raw.slice('Bearer '.length).trim() === expectedToken;
}
function writeSseEvent(reply: { raw: NodeJS.WritableStream & { write: (chunk: string) => void } }, event: string, data: unknown) {
reply.raw.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
}
function toDeployInput(body: Record<string, unknown>): UpdateHelperDeployInput {
return {
namespace: String(body.namespace || '').trim(),
releaseName: String(body.releaseName || '').trim(),
chartRef: String(body.chartRef || '').trim(),
imageRepository: String(body.imageRepository || '').trim(),
targetSource: body.source === 'docker-hub-tag' ? 'docker-hub-tag' : 'github-release',
targetTag: String(body.targetVersion || '').trim(),
};
}
export async function buildUpdateHelperApp(options: BuildUpdateHelperAppOptions): Promise<FastifyInstance> {
const app = Fastify();
const getStatus = options.getStatus || (async (input) => await getUpdateHelperStatus(input));
const deploy = options.deploy || (async (input, onLog) => await executeUpdateHelperDeploy(input, undefined, onLog));
app.addHook('onRequest', async (request, reply) => {
const path = String(request.raw.url || '').split('?')[0];
if (path === '/health') {
return;
}
if (!requireBearerToken(String(request.headers.authorization || ''), options.token)) {
return reply.code(401).send({
success: false,
message: 'unauthorized',
});
}
});
app.get('/health', async () => ({ ok: true }));
app.get<{ Querystring: { namespace?: string; releaseName?: string } }>('/status', async (request, reply) => {
const namespace = String(request.query.namespace || '').trim();
const releaseName = String(request.query.releaseName || '').trim();
if (!namespace || !releaseName) {
return reply.code(400).send({
success: false,
message: 'namespace and releaseName are required',
});
}
return await getStatus({ namespace, releaseName });
});
app.post<{ Body: Record<string, unknown> }>('/deploy', async (request, reply) => {
const input = toDeployInput(request.body || {});
if (!input.namespace || !input.releaseName || !input.chartRef || !input.imageRepository || !input.targetTag) {
return reply.code(400).send({
success: false,
message: 'namespace, releaseName, chartRef, imageRepository, and targetVersion are required',
});
}
const wantsSse = String(request.headers.accept || '').includes('text/event-stream');
if (!wantsSse) {
return await deploy(input);
}
reply.hijack();
reply.raw.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
reply.raw.setHeader('Cache-Control', 'no-cache, no-transform');
reply.raw.setHeader('Connection', 'keep-alive');
const result = await deploy(input, (message) => {
writeSseEvent(reply, 'log', { message });
});
writeSseEvent(reply, 'result', result);
reply.raw.end();
});
return app;
}
+19
View File
@@ -0,0 +1,19 @@
import { buildUpdateHelperApp } from './app.js';
const port = Number.parseInt(process.env.DEPLOY_HELPER_PORT || '9850', 10) || 9850;
const host = (process.env.DEPLOY_HELPER_HOST || '0.0.0.0').trim() || '0.0.0.0';
const token = String(process.env.DEPLOY_HELPER_TOKEN || '').trim();
if (!token) {
throw new Error('DEPLOY_HELPER_TOKEN is required');
}
const app = await buildUpdateHelperApp({ token });
try {
await app.listen({ port, host });
app.log.info(`Deploy helper listening on http://${host}:${port}`);
} catch (error) {
app.log.error(error);
process.exit(1);
}
+167
View File
@@ -0,0 +1,167 @@
import { describe, expect, it } from 'vitest';
import {
executeUpdateHelperDeploy,
getUpdateHelperStatus,
} from './service.js';
describe('update helper service', () => {
it('builds helm upgrade and rollout verification commands with the configured release metadata', async () => {
const calls: Array<{ command: string; args: string[] }> = [];
const summary = await executeUpdateHelperDeploy(
{
namespace: 'ai',
releaseName: 'metapi',
chartRef: 'oci://ghcr.io/cita-777/charts/metapi',
imageRepository: '1467078763/metapi',
targetSource: 'github-release',
targetTag: '1.3.0',
},
{
runCommand: async ({ command, args }) => {
calls.push({ command, args });
if (command === 'helm' && args[0] === 'history') {
return {
stdout: JSON.stringify([{ revision: '12' }]),
};
}
if (command === 'helm' && args[0] === 'status') {
return {
stdout: JSON.stringify({
version: 13,
}),
};
}
return { stdout: '' };
},
},
);
expect(calls).toEqual([
{
command: 'helm',
args: ['history', 'metapi', '-n', 'ai', '-o', 'json'],
},
{
command: 'helm',
args: [
'upgrade',
'metapi',
'oci://ghcr.io/cita-777/charts/metapi',
'--namespace',
'ai',
'--reuse-values',
'--set',
'image.repository=1467078763/metapi',
'--set',
'image.tag=1.3.0',
],
},
{
command: 'kubectl',
args: [
'rollout',
'status',
'deployment',
'-n',
'ai',
'-l',
'app.kubernetes.io/instance=metapi',
'--timeout=300s',
],
},
{
command: 'helm',
args: ['status', 'metapi', '-n', 'ai', '-o', 'json'],
},
]);
expect(summary).toMatchObject({
success: true,
targetSource: 'github-release',
targetTag: '1.3.0',
previousRevision: '12',
finalRevision: '13',
rolledBack: false,
});
});
it('rolls back to the previous revision when rollout verification fails', async () => {
const calls: Array<{ command: string; args: string[] }> = [];
await expect(() => executeUpdateHelperDeploy(
{
namespace: 'ai',
releaseName: 'metapi',
chartRef: 'oci://ghcr.io/cita-777/charts/metapi',
imageRepository: '1467078763/metapi',
targetSource: 'docker-hub-tag',
targetTag: '1.3.1',
},
{
runCommand: async ({ command, args }) => {
calls.push({ command, args });
if (command === 'helm' && args[0] === 'history') {
return {
stdout: JSON.stringify([{ revision: '22' }]),
};
}
if (command === 'kubectl' && args[0] === 'rollout') {
throw new Error('rollout status timed out');
}
return { stdout: '' };
},
},
)).rejects.toThrow('rollout status timed out');
expect(calls).toEqual(expect.arrayContaining([
{
command: 'helm',
args: ['rollback', 'metapi', '22', '-n', 'ai'],
},
]));
});
it('parses helper status from helm status and values output', async () => {
const status = await getUpdateHelperStatus(
{
namespace: 'ai',
releaseName: 'metapi',
},
{
runCommand: async ({ command, args }) => {
if (command === 'helm' && args[0] === 'status') {
return {
stdout: JSON.stringify({
version: 17,
info: {
status: 'deployed',
},
}),
};
}
if (command === 'helm' && args[0] === 'get') {
return {
stdout: JSON.stringify({
image: {
repository: '1467078763/metapi',
tag: '1.3.1',
},
}),
};
}
throw new Error('unexpected command');
},
},
);
expect(status).toEqual({
ok: true,
releaseName: 'metapi',
namespace: 'ai',
revision: '17',
imageRepository: '1467078763/metapi',
imageTag: '1.3.1',
healthy: true,
});
});
});
+232
View File
@@ -0,0 +1,232 @@
import { spawn } from 'node:child_process';
import { createInterface } from 'node:readline';
import type { UpdateCenterVersionSource } from '../services/updateCenterVersionService.js';
type RunCommandInput = {
command: string;
args: string[];
onLine?: (line: string) => void;
};
type RunCommandResult = {
stdout: string;
};
type RunCommand = (input: RunCommandInput) => Promise<RunCommandResult>;
export type UpdateHelperDeployInput = {
namespace: string;
releaseName: string;
chartRef: string;
imageRepository: string;
targetSource: UpdateCenterVersionSource;
targetTag: string;
};
export type UpdateHelperStatusInput = {
namespace: string;
releaseName: string;
};
export type UpdateHelperDeploySummary = {
success: boolean;
targetSource: UpdateCenterVersionSource;
targetTag: string;
previousRevision: string | null;
finalRevision: string | null;
rolledBack: boolean;
logLines: string[];
};
export type UpdateHelperStatus = {
ok: boolean;
releaseName: string;
namespace: string;
revision: string | null;
imageRepository: string | null;
imageTag: string | null;
healthy: boolean;
};
async function runCommand(input: RunCommandInput): Promise<RunCommandResult> {
return await new Promise<RunCommandResult>((resolve, reject) => {
const child = spawn(input.command, input.args, {
stdio: ['ignore', 'pipe', 'pipe'],
});
let stdout = '';
let stderr = '';
const handleLine = (line: string) => {
const normalized = line.trim();
if (!normalized) return;
input.onLine?.(normalized);
};
if (child.stdout) {
const stdoutReader = createInterface({ input: child.stdout });
stdoutReader.on('line', (line) => {
stdout += `${line}\n`;
handleLine(line);
});
}
if (child.stderr) {
const stderrReader = createInterface({ input: child.stderr });
stderrReader.on('line', (line) => {
stderr += `${line}\n`;
handleLine(line);
});
}
child.once('error', reject);
child.once('close', (code) => {
if (code === 0) {
resolve({ stdout: stdout.trim() });
return;
}
reject(new Error(stderr.trim() || stdout.trim() || `${input.command} exited with code ${code}`));
});
});
}
function parseLatestRevision(stdout: string): string | null {
try {
const rows = JSON.parse(stdout) as Array<{ revision?: string | number }>;
const last = Array.isArray(rows) && rows.length > 0 ? rows[rows.length - 1] : null;
if (!last?.revision) return null;
return String(last.revision);
} catch {
return null;
}
}
function parseStatusRevision(stdout: string): string | null {
try {
const payload = JSON.parse(stdout) as { version?: string | number };
if (payload?.version === undefined || payload?.version === null) return null;
return String(payload.version);
} catch {
return null;
}
}
export async function executeUpdateHelperDeploy(
input: UpdateHelperDeployInput,
deps: {
runCommand?: RunCommand;
} = {},
onLog?: (line: string) => void,
): Promise<UpdateHelperDeploySummary> {
const commandRunner = deps.runCommand || runCommand;
const logLines: string[] = [];
const captureLine = (line: string) => {
logLines.push(line);
onLog?.(line);
};
const history = await commandRunner({
command: 'helm',
args: ['history', input.releaseName, '-n', input.namespace, '-o', 'json'],
});
const previousRevision = parseLatestRevision(history.stdout);
await commandRunner({
command: 'helm',
args: [
'upgrade',
input.releaseName,
input.chartRef,
'--namespace',
input.namespace,
'--reuse-values',
'--set',
`image.repository=${input.imageRepository}`,
'--set',
`image.tag=${input.targetTag}`,
],
onLine: captureLine,
});
try {
await commandRunner({
command: 'kubectl',
args: [
'rollout',
'status',
'deployment',
'-n',
input.namespace,
'-l',
`app.kubernetes.io/instance=${input.releaseName}`,
'--timeout=300s',
],
onLine: captureLine,
});
} catch (error) {
if (previousRevision) {
await commandRunner({
command: 'helm',
args: ['rollback', input.releaseName, previousRevision, '-n', input.namespace],
onLine: captureLine,
});
}
throw error;
}
const status = await commandRunner({
command: 'helm',
args: ['status', input.releaseName, '-n', input.namespace, '-o', 'json'],
});
return {
success: true,
targetSource: input.targetSource,
targetTag: input.targetTag,
previousRevision,
finalRevision: parseStatusRevision(status.stdout),
rolledBack: false,
logLines,
};
}
export async function getUpdateHelperStatus(
input: UpdateHelperStatusInput,
deps: {
runCommand?: RunCommand;
} = {},
): Promise<UpdateHelperStatus> {
const commandRunner = deps.runCommand || runCommand;
const status = await commandRunner({
command: 'helm',
args: ['status', input.releaseName, '-n', input.namespace, '-o', 'json'],
});
const values = await commandRunner({
command: 'helm',
args: ['get', 'values', input.releaseName, '-n', input.namespace, '--all', '-o', 'json'],
});
const statusPayload = JSON.parse(status.stdout || '{}') as {
version?: string | number;
info?: {
status?: string;
};
};
const valuesPayload = JSON.parse(values.stdout || '{}') as {
image?: {
repository?: string;
tag?: string;
};
};
return {
ok: true,
releaseName: input.releaseName,
namespace: input.namespace,
revision: statusPayload.version === undefined || statusPayload.version === null ? null : String(statusPayload.version),
imageRepository: valuesPayload.image?.repository || null,
imageTag: valuesPayload.image?.tag || null,
healthy: String(statusPayload.info?.status || '').toLowerCase() === 'deployed',
};
}