|
|
@@ -0,0 +1,142 @@
|
|
|
+import {mkdir,open,rm,stat,writeFile} from 'node:fs/promises'
|
|
|
+import {join,dirname,resolve} from 'node:path'
|
|
|
+import {existsSync} from 'node:fs'
|
|
|
+import {createHash} from 'node:crypto'
|
|
|
+import type {DatabaseSync} from 'node:sqlite'
|
|
|
+import {z} from 'zod'
|
|
|
+import {ComputerImages,imageManifest} from './images.js'
|
|
|
+import {ComputerError} from './container.js'
|
|
|
+import type {ComputerRuntime} from '../worker/gateway.js'
|
|
|
+import type {ImageJob,ImageInventory,ImageOperation} from '../../shared/imageManagement.js'
|
|
|
+const chunkSize=512*1024,maxSize=20*1024**3
|
|
|
+const digest=(v:unknown)=>createHash('sha256').update(JSON.stringify(v)).digest('hex')
|
|
|
+interface Upload {id:string;owner:string;size:number;offset:number;manifest:Record<string,unknown>;createdAt:number}
|
|
|
+interface StoredJob extends ImageJob {owner:string;fingerprint:string}
|
|
|
+/** Only administrator RPCs reach this manager; no host paths come from a model or browser. */
|
|
|
+export class ImageManagement {
|
|
|
+ private pending?:Promise<void>
|
|
|
+ private closed=false
|
|
|
+ private blocked?:{ids:string[];release:()=>void}
|
|
|
+ private repair?:Promise<void>
|
|
|
+ private directory:string
|
|
|
+ readonly images:ComputerImages
|
|
|
+ readonly ready:Promise<void>
|
|
|
+ constructor(readonly runtime:ComputerRuntime,readonly db:DatabaseSync,readonly home:string){
|
|
|
+ this.directory=join(home,'image-management');this.images=new ComputerImages(runtime.config.runtime)
|
|
|
+ db.exec('CREATE TABLE IF NOT EXISTS managed_image_jobs(id TEXT PRIMARY KEY,value TEXT NOT NULL); CREATE TABLE IF NOT EXISTS managed_image_uploads(id TEXT PRIMARY KEY,value TEXT NOT NULL); CREATE TABLE IF NOT EXISTS managed_image_selection(id INTEGER PRIMARY KEY,value TEXT NOT NULL)')
|
|
|
+ const selected=this.selection();if(selected?.base===runtime.config.imageId)runtime.config.imageId=selected.active;else if(selected)db.prepare('DELETE FROM managed_image_selection').run()
|
|
|
+ const release=runtime.pool.holdMaintenance();this.ready=runtime.ready.then(()=>this.recover()).then(()=>{if(this.blocked)this.blocked.release=release;else release()});void this.ready.catch(()=>{})
|
|
|
+ }
|
|
|
+ private selection():{base:string;active:string;previous?:string}|undefined{const row=this.db.prepare('SELECT value FROM managed_image_selection WHERE id=1').get() as any;return row?JSON.parse(row.value):undefined}
|
|
|
+ private job(id:string):StoredJob|undefined {const row=this.db.prepare('SELECT value FROM managed_image_jobs WHERE id=?').get(id) as any;return row?JSON.parse(row.value):undefined}
|
|
|
+ private jobs():StoredJob[]{return (this.db.prepare('SELECT value FROM managed_image_jobs ORDER BY rowid DESC LIMIT 40').all() as any[]).map(row=>JSON.parse(row.value))}
|
|
|
+ private save(job:StoredJob){job.updatedAt=Date.now();this.db.prepare('INSERT OR REPLACE INTO managed_image_jobs VALUES(?,?)').run(job.id,JSON.stringify(job))}
|
|
|
+ private async recover(){
|
|
|
+ await mkdir(this.directory,{recursive:true,mode:0o700})
|
|
|
+ for(const job of this.jobs())if(job.state==='running'||job.state==='uncertain'){
|
|
|
+ job.state='uncertain';job.message='节点曾重启，原操作不会重复执行。请刷新列表或重新校验镜像。';this.save(job)
|
|
|
+ try{await this.cleanupCheck(job.id)}catch{this.blocked??={ids:[],release:()=>{}};this.blocked.ids.push(job.id);job.message='测试电脑停止尚未确认，刷新状态重试清理';this.save(job)}
|
|
|
+ }
|
|
|
+ for(const row of this.db.prepare('SELECT value FROM managed_image_uploads').all() as any[]){const u:Upload=JSON.parse(row.value);if(Date.now()-u.createdAt>2*3600000){await rm(this.uploadPath(u.id),{force:true});await rm(this.uploadPath(u.id)+'.json',{force:true});this.db.prepare('DELETE FROM managed_image_uploads WHERE id=?').run(u.id)}}
|
|
|
+ }
|
|
|
+ private checkSpec(id:string,imageId=this.runtime.config.imageId){return {id,ownerKey:'image-management',imageId}}
|
|
|
+ private async cleanupCheck(id:string){const spec=this.checkSpec(id);await this.runtime.provider.remove(spec);await this.runtime.provider.deleteWorkspace(spec)}
|
|
|
+ private async retryCleanup(){
|
|
|
+ if(this.repair)return this.repair
|
|
|
+ const blocked=this.blocked;if(!blocked)return
|
|
|
+ this.repair=(async()=>{for(const id of blocked.ids)await this.cleanupCheck(id);if(this.blocked===blocked){this.blocked=undefined;blocked.release();for(const id of blocked.ids){const job=this.job(id);if(job){job.state='failed';job.message='测试电脑已清理，可重新校验镜像';this.save(job)}}}})().finally(()=>{this.repair=undefined})
|
|
|
+ return this.repair
|
|
|
+ }
|
|
|
+ async inventory(owner:string):Promise<ImageInventory>{
|
|
|
+ await this.ready;await this.retryCleanup().catch(()=>{})
|
|
|
+ let output:string
|
|
|
+ try{output=await this.images.run(this.images.runtime,['image','ls','--no-trunc','--quiet','--filter','label=com.openmausbot.cua-driver=0.20.0'])}catch{throw new ComputerError('image_runtime_unavailable','无法读取镜像，请确认执行节点上的 Docker/Podman 服务已启动')}
|
|
|
+ const images=[]
|
|
|
+ for(const id of [...new Set(output.trim().split(/\s+/).filter(Boolean))].slice(0,100))try{images.push(await this.images.inspect(id))}catch{}
|
|
|
+ return {configured:true,activeImageId:this.runtime.config.imageId,previousImageId:this.selection()?.previous,images,jobs:this.jobs().filter(j=>j.owner===owner).map(({owner:_o,fingerprint:_f,...j})=>j),busy:!!this.pending||!!this.blocked,uploads:(this.db.prepare('SELECT value FROM managed_image_uploads').all() as any[]).map(row=>JSON.parse(row.value) as Upload).filter(u=>u.owner===owner).map(u=>({id:u.id,size:u.size,offset:u.offset}))}
|
|
|
+ }
|
|
|
+ private uploadPath(id:string){return join(this.directory,id+'.tar')}
|
|
|
+ async beginUpload(owner:string,id:string,size:number,manifest:unknown){
|
|
|
+ z.string().uuid().parse(id);z.number().int().positive().max(maxSize).parse(size);const value=imageManifest.parse(manifest)
|
|
|
+ if(!value.archiveSha256)throw new ComputerError('image_manifest_invalid','请选择包含 SHA-256 的镜像校验清单')
|
|
|
+ await this.ready
|
|
|
+ const old=this.db.prepare('SELECT value FROM managed_image_uploads WHERE id=?').get(id) as any
|
|
|
+ if(old){const entry:Upload=JSON.parse(old.value);if(entry.owner!==owner||entry.size!==size||digest(entry.manifest)!==digest(value))throw new ComputerError('image_upload_conflict','上传编号已用于其他文件');return {id,offset:entry.offset}}
|
|
|
+ if(Number(this.db.prepare('SELECT count(*) AS n FROM managed_image_uploads').get()!.n)>=2)throw new ComputerError('image_upload_limit','请先完成或取消已有上传')
|
|
|
+ await writeFile(this.uploadPath(id),'',{flag:'wx',mode:0o600})
|
|
|
+ const entry:Upload={id,owner,size,manifest:value,offset:0,createdAt:Date.now()};this.db.prepare('INSERT INTO managed_image_uploads VALUES(?,?)').run(id,JSON.stringify(entry));return {id,offset:0}
|
|
|
+ }
|
|
|
+ private upload(owner:string,id:string):Upload{z.string().uuid().parse(id);const row=this.db.prepare('SELECT value FROM managed_image_uploads WHERE id=?').get(id) as any;const u:Upload|undefined=row?JSON.parse(row.value):undefined;if(!u||u.owner!==owner)throw new ComputerError('image_upload_missing','上传不存在或不属于当前账号');return u}
|
|
|
+ private uploading=new Set<string>()
|
|
|
+ async appendUpload(owner:string,id:string,offset:number,data:string){
|
|
|
+ const u=this.upload(owner,id);if(this.uploading.has(id))throw new ComputerError('image_upload_busy','上一块仍在写入')
|
|
|
+ const bytes=Buffer.from(z.string().max(750000).parse(data),'base64');z.number().int().nonnegative().parse(offset)
|
|
|
+ if(!bytes.length||bytes.length>chunkSize||offset+bytes.length>u.size)throw new ComputerError('image_upload_invalid','镜像数据块无效')
|
|
|
+ this.uploading.add(id)
|
|
|
+ try{
|
|
|
+ const file=await open(this.uploadPath(id),'r+')
|
|
|
+ try{
|
|
|
+ if(offset<u.offset){const old=Buffer.alloc(bytes.length);await file.read(old,0,old.length,offset);if(offset+bytes.length>u.offset||!old.equals(bytes))throw new Error('上传内容冲突');return {offset:u.offset}}
|
|
|
+ if(offset!==u.offset)throw new ComputerError('image_upload_offset','请从已接收的位置继续上传')
|
|
|
+ let written=0;while(written<bytes.length){const result=await file.write(bytes,written,bytes.length-written,offset+written);if(!result.bytesWritten)throw new Error('写入未完成');written+=result.bytesWritten}await file.sync();u.offset+=bytes.length
|
|
|
+ this.db.prepare('UPDATE managed_image_uploads SET value=? WHERE id=?').run(JSON.stringify(u),id);return {offset:u.offset}
|
|
|
+ }finally{await file.close()}
|
|
|
+ }finally{this.uploading.delete(id)}
|
|
|
+ }
|
|
|
+ async cancelUpload(owner:string,id:string){this.upload(owner,id);if(this.uploading.has(id)||this.pending)throw new ComputerError('image_upload_busy','请等待当前操作结束');await rm(this.uploadPath(id),{force:true});this.db.prepare('DELETE FROM managed_image_uploads WHERE id=?').run(id);return {ok:true}}
|
|
|
+ async start(owner:string,id:string,operation:ImageOperation,args:{imageId?:string;uploadId?:string},check:()=>Promise<void>){
|
|
|
+ await this.ready;await check();z.string().uuid().parse(id)
|
|
|
+ const fingerprint=digest({owner,operation,args}),old=this.job(id)
|
|
|
+ if(old){if(old.fingerprint!==fingerprint)throw new ComputerError('idempotency_conflict','操作编号与内容不一致');return {job:old}}
|
|
|
+ if(this.closed||this.pending||this.blocked)throw new ComputerError('image_job_busy','已有镜像操作正在执行')
|
|
|
+ z.enum(['prepare','verify','activate','rollback','import','export']).parse(operation)
|
|
|
+ if(operation==='export'&&this.jobs().filter(j=>j.archiveSize).length>=3)throw new ComputerError('image_export_limit','请先清理之前的导出文件，最多保留三份')
|
|
|
+ const release=this.runtime.pool.beginMaintenance()
|
|
|
+ const job:StoredJob={id,owner,operation,fingerprint,state:'running',message:'正在准备',createdAt:Date.now(),updatedAt:Date.now()};this.save(job)
|
|
|
+ const progress=(message:string)=>{job.message=message;this.save(job)}
|
|
|
+ this.pending=(async()=>{
|
|
|
+ try{
|
|
|
+ await this.runtime.provider.verifyRuntime();await check()
|
|
|
+ const manager=this.images
|
|
|
+ const verify=async(manifest:any)=>{
|
|
|
+ progress('正在启动测试电脑并校验桌面');await check()
|
|
|
+ const spec=this.checkSpec(id,manifest.imageId)
|
|
|
+ try{await this.runtime.provider.ensure(spec,()=>{if(this.closed)throw new Error('节点正在关闭')});const end=Date.now()+60000;for(;;){await check();try{await this.runtime.provider.health(spec,()=>{});await this.runtime.provider.capture(spec,()=>{});break}catch(e){if(Date.now()>end)throw e;await new Promise(r=>setTimeout(r,1000))}}}
|
|
|
+ finally{try{await this.runtime.provider.remove(spec);await this.runtime.provider.deleteWorkspace(spec)}catch{this.blocked={ids:[id],release};throw new ComputerError('computer_stop_uncertain','测试电脑停止尚未确认，请刷新状态重试清理')}}
|
|
|
+ }
|
|
|
+ const previousVerify=manager.verify.bind(manager);manager.verify=verify
|
|
|
+ try{
|
|
|
+ let manifest:any
|
|
|
+ if(operation==='prepare'){progress('正在构建兼容镜像，首次准备可能需要数分钟');manifest=await manager.prepare([join(dirname(this.runtime.script),'computer-image'),join(dirname(this.runtime.script),'../computers/computer-image'),resolve('deploy/computer')].find(path=>existsSync(join(path,'Dockerfile')))??'missing-computer-recipe')}
|
|
|
+ else if(operation==='import'){
|
|
|
+ const u=this.upload(owner,args.uploadId!);if(u.offset!==u.size)throw new ComputerError('image_upload_incomplete','镜像尚未上传完成')
|
|
|
+ progress('正在校验归档并导入镜像');await writeFile(this.uploadPath(u.id)+'.json',JSON.stringify(u.manifest),{mode:0o600})
|
|
|
+ manifest=await manager.import(this.uploadPath(u.id));await rm(this.uploadPath(u.id),{force:true});await rm(this.uploadPath(u.id)+'.json',{force:true});this.db.prepare('DELETE FROM managed_image_uploads WHERE id=?').run(u.id)
|
|
|
+ }else{
|
|
|
+ const selected=operation==='rollback'?this.selection()?.previous:args.imageId
|
|
|
+ if(!selected)throw new ComputerError('image_missing','没有可用的镜像或回退版本')
|
|
|
+ manifest=await manager.inspect(selected)
|
|
|
+ if(operation==='export'){progress('正在导出镜像');manifest=await manager.export(selected,join(this.directory,id+'.export.tar'));job.archiveSize=(await stat(join(this.directory,id+'.export.tar'))).size;job.manifest=manifest}
|
|
|
+ else await verify(manifest)
|
|
|
+ }
|
|
|
+ await check();job.imageId=manifest.imageId
|
|
|
+ if((operation==='activate'||operation==='rollback')&&manifest.imageId!==this.runtime.config.imageId){
|
|
|
+ const old=this.selection();this.db.prepare('INSERT OR REPLACE INTO managed_image_selection VALUES(1,?)').run(JSON.stringify({base:old?.base??this.runtime.config.imageId,active:manifest.imageId,previous:this.runtime.config.imageId}))
|
|
|
+ this.runtime.config.imageId=manifest.imageId
|
|
|
+ }
|
|
|
+ job.state='complete';progress(operation==='activate'||operation==='rollback'?'已应用，下次启动电脑使用此镜像':'操作完成')
|
|
|
+ }finally{manager.verify=previousVerify}
|
|
|
+ }catch(error){job.state=this.blocked?'uncertain':'failed';progress(error instanceof Error&&/^(镜像|电脑|上传|请|没有)/.test(error.message)?error.message.slice(0,240):'镜像操作失败，请检查运行时、网络和镜像清单后重试')}
|
|
|
+ finally{if(!this.blocked)release();this.pending=undefined}
|
|
|
+ })()
|
|
|
+ return {job:{...job}}
|
|
|
+ }
|
|
|
+ async download(owner:string,id:string,offset:number){
|
|
|
+ z.string().uuid().parse(id);z.number().int().nonnegative().parse(offset)
|
|
|
+ const job=this.job(id)
|
|
|
+ if(!job||job.owner!==owner||job.state!=='complete'||job.operation!=='export'||!job.archiveSize||offset>job.archiveSize)throw new ComputerError('image_export_missing','导出文件不存在')
|
|
|
+ const file=await open(join(this.directory,id+'.export.tar'),'r');try{const bytes=Buffer.alloc(Math.min(chunkSize,job.archiveSize-offset));let received=0;while(received<bytes.length){const result=await file.read(bytes,received,bytes.length-received,offset+received);if(!result.bytesRead)throw new ComputerError('image_export_incomplete','镜像导出不完整');received+=result.bytesRead}return {data:bytes.toString('base64'),size:job.archiveSize,offset,manifest:job.manifest}}finally{await file.close()}
|
|
|
+ }
|
|
|
+ async removeExport(owner:string,id:string){const job=this.job(id);if(!job||job.owner!==owner||job.state!=='complete'||job.operation!=='export')throw new ComputerError('image_export_missing','导出不存在');await rm(join(this.directory,id+'.export.tar'),{force:true});await rm(join(this.directory,id+'.export.tar.json'),{force:true});delete job.archiveSize;job.message='导出文件已清理，可重新导出';this.save(job);return {ok:true}}
|
|
|
+ async close(){this.closed=true;await this.ready;await this.pending;await this.retryCleanup()}
|
|
|
+}
|
LSamien