devsam/core2026
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(gateway): stream release progress logs

Browse Source
This commit is contained in:
Hide_D
2026-08-09 09:50:06 +00:00
parent 73edd0230f
commit d01be27828
15 changed files with 619 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/gateway-api/src/adminRouter.ts
+30
View File
@@ -1351,6 +1351,36 @@ export const adminRouter = router({
list: releaseAdminProcedure
.input(z.object({ limit: z.number().int().min(1).max(200).optional() }).optional())
.query(({ ctx, input }) => ctx.releases.listOperations(input?.limit)),
logs: releaseAdminProcedure
.input(
z.object({
id: z.string().uuid(),
afterCursor: z.string().regex(/^\d+$/u).optional(),
limit: z.number().int().min(1).max(500).default(200),
timeoutMs: z.number().int().min(0).max(25_000).default(20_000),
})
)
.query(async ({ ctx, input }) => {
const deadline = Date.now() + input.timeoutMs;
while (true) {
const [operation, entries] = await Promise.all([
ctx.releases.getOperation(input.id),
ctx.releases.listOperationLogs(input.id, input.afterCursor, input.limit),
]);
if (!operation) {
throw new TRPCError({ code: 'NOT_FOUND', message: 'Gateway release operation not found.' });
}
const terminal = ['SUCCEEDED', 'FAILED', 'CANCELLED'].includes(operation.status);
if (entries.length || terminal || Date.now() >= deadline) {
return {
operation,
entries,
nextCursor: entries.at(-1)?.cursor ?? input.afterCursor,
};
}
await new Promise<void>((resolve) => setTimeout(resolve, 250));
}
}),
requestGatewayDeploy: releaseAdminProcedure
.input(
z.object({
app/gateway-api/src/orchestrator/buildRunner.ts
+46 -13
View File
@@ -13,8 +13,15 @@ export interface BuildResult {
output: string;
}
export type BuildProgressEvent =
| { type: 'COMMAND_START'; command: BuildCommand }
| { type: 'OUTPUT'; stream: 'stdout' | 'stderr'; message: string }
| { type: 'COMMAND_END'; command: BuildCommand; exitCode: number | null };
export type BuildProgressObserver = (event: BuildProgressEvent) => void | Promise<void>;
export interface BuildRunner {
run(commands: BuildCommand[]): Promise<BuildResult>;
run(commands: BuildCommand[], onProgress?: BuildProgressObserver): Promise<BuildResult>;
}
export const MAX_BUILD_OUTPUT_CHARS = 64 * 1024;
@@ -22,41 +29,67 @@ export const MAX_BUILD_OUTPUT_CHARS = 64 * 1024;
const appendOutputTail = (current: string, chunk: unknown): string =>
`${current}${String(chunk)}`.slice(-MAX_BUILD_OUTPUT_CHARS);
const runCommand = (command: BuildCommand): Promise<BuildResult> =>
const runCommand = (command: BuildCommand, onProgress?: BuildProgressObserver): Promise<BuildResult> =>
new Promise((resolve) => {
let progressQueue = Promise.resolve();
const emit = (event: BuildProgressEvent) => {
if (!onProgress) return;
progressQueue = progressQueue.then(() => onProgress(event)).catch(() => undefined);
};
emit({ type: 'COMMAND_START', command });
const child = spawn(command.command, command.args, {
cwd: command.cwd,
env: command.env,
stdio: ['ignore', 'pipe', 'pipe'],
});
let output = '';
let spawnFailed = false;
const lineBuffers = { stdout: '', stderr: '' };
const emitOutput = (stream: 'stdout' | 'stderr', chunk: unknown, flush = false) => {
if (flush && !lineBuffers[stream]) return;
lineBuffers[stream] += String(chunk);
const lines = lineBuffers[stream].split(/\r?\n/u);
lineBuffers[stream] = flush ? '' : (lines.pop() ?? '');
if (flush && lineBuffers[stream]) lines.push(lineBuffers[stream]);
for (const line of lines) {
for (let offset = 0; offset < line.length || (offset === 0 && line.length === 0); offset += 2_000) {
emit({ type: 'OUTPUT', stream, message: line.slice(offset, offset + 2_000) });
if (line.length === 0) break;
}
}
};
child.stdout.on('data', (chunk) => {
output = appendOutputTail(output, chunk);
emitOutput('stdout', chunk);
});
child.stderr.on('data', (chunk) => {
output = appendOutputTail(output, chunk);
emitOutput('stderr', chunk);
});
child.on('error', (error) => {
resolve({
ok: false,
exitCode: null,
output: appendOutputTail(output, error.message),
});
spawnFailed = true;
output = appendOutputTail(output, error.message);
});
child.on('close', (code) => {
resolve({
ok: code === 0,
exitCode: code,
output,
emitOutput('stdout', '', true);
emitOutput('stderr', '', true);
const exitCode = spawnFailed ? null : code;
emit({ type: 'COMMAND_END', command, exitCode });
void progressQueue.then(() => {
resolve({
ok: !spawnFailed && code === 0,
exitCode,
output,
});
});
});
});
export class PnpmBuildRunner implements BuildRunner {
async run(commands: BuildCommand[]): Promise<BuildResult> {
async run(commands: BuildCommand[], onProgress?: BuildProgressObserver): Promise<BuildResult> {
let mergedOutput = '';
for (const command of commands) {
const result = await runCommand(command);
const result = await runCommand(command, onProgress);
mergedOutput = appendOutputTail(mergedOutput, result.output);
if (!result.ok) {
return {
app/gateway-api/src/orchestrator/gatewayReleaseRepository.ts
+60
View File
@@ -46,10 +46,30 @@ export interface GatewayReleaseOperationCreateInput {
requestedBy: string;
}
export const GATEWAY_RELEASE_LOG_LEVELS = ['INFO', 'OUTPUT', 'ERROR'] as const;
export type GatewayReleaseLogLevel = (typeof GATEWAY_RELEASE_LOG_LEVELS)[number];
export interface GatewayReleaseLogRecord {
cursor: string;
operationId: string;
level: GatewayReleaseLogLevel;
phase: string;
message: string;
createdAt: string;
}
export interface GatewayReleaseLogInput {
level: GatewayReleaseLogLevel;
phase: string;
message: string;
}
export interface GatewayReleaseRepository {
getState(): Promise<GatewayReleaseStateRecord>;
listOperations(limit?: number): Promise<GatewayReleaseOperationRecord[]>;
getOperation(id: string): Promise<GatewayReleaseOperationRecord | null>;
listOperationLogs(id: string, afterCursor?: string, limit?: number): Promise<GatewayReleaseLogRecord[]>;
appendOperationLog(id: string, input: GatewayReleaseLogInput): Promise<GatewayReleaseLogRecord>;
createOperation(input: GatewayReleaseOperationCreateInput): Promise<GatewayReleaseOperationRecord>;
claimNextOperation(
now: Date,
@@ -135,6 +155,24 @@ const mapOperation = (row: {
updatedAt: row.updatedAt.toISOString(),
});
const mapLog = (row: {
id: bigint;
operationId: string;
level: string;
phase: string;
message: string;
createdAt: Date;
}): GatewayReleaseLogRecord => ({
cursor: row.id.toString(),
operationId: row.operationId,
level: GATEWAY_RELEASE_LOG_LEVELS.includes(row.level as GatewayReleaseLogLevel)
? (row.level as GatewayReleaseLogLevel)
: 'INFO',
phase: row.phase,
message: row.message,
createdAt: row.createdAt.toISOString(),
});
export const createGatewayReleaseRepository = (prisma: GatewayPrismaClient): GatewayReleaseRepository => ({
async getState() {
const row = await prisma.gatewayReleaseState.upsert({
@@ -155,6 +193,28 @@ export const createGatewayReleaseRepository = (prisma: GatewayPrismaClient): Gat
const row = await prisma.gatewayReleaseOperation.findUnique({ where: { id } });
return row ? mapOperation(row) : null;
},
async listOperationLogs(id, afterCursor, limit = 200) {
const rows = await prisma.gatewayReleaseLog.findMany({
where: {
operationId: id,
...(afterCursor ? { id: { gt: BigInt(afterCursor) } } : {}),
},
orderBy: { id: 'asc' },
take: Math.min(Math.max(limit, 1), 500),
});
return rows.map(mapLog);
},
async appendOperationLog(id, input) {
const row = await prisma.gatewayReleaseLog.create({
data: {
operationId: id,
level: input.level,
phase: input.phase.slice(0, 64),
message: input.message.slice(0, 4_000),
},
});
return mapLog(row);
},
async createOperation(input) {
const row = await prisma.gatewayReleaseOperation.create({
data: {