82 lines
8.3 KiB
TypeScript
82 lines
8.3 KiB
TypeScript
import {randomUUID,createHash} from 'node:crypto';
|
|
import {PutObjectCommand} from '@aws-sdk/client-s3';
|
|
import {db,transaction} from './db';
|
|
import {storage,bucket,getObject,removeObject} from './storage';
|
|
import {newToken,tokenDigest} from './account-crypto';
|
|
import {parseCanvasDocument,agentTask,type CanvasDocument} from './canvas-document';
|
|
export class CanvasError extends Error { constructor(public status:number,message:string){super(message)} }
|
|
const inputDocument=(input:unknown)=>{try{return parseCanvasDocument(input)}catch{throw new CanvasError(400,'画布或结果参数无效')}};
|
|
export const uuid=(v:unknown):v is string=>typeof v==='string'&&/^[a-f\d]{8}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{4}-[a-f\d]{12}$/i.test(v);
|
|
export const revision=(v:unknown):v is number=>typeof v==='number'&&Number.isSafeInteger(v)&&v>0&&v<2147483647;
|
|
export async function readCanvasBody(key:string):Promise<CanvasDocument>{const o=await getObject(key);return parseCanvasDocument(await o.Body!.transformToString())}
|
|
async function putJson(canvas:string,value:unknown){const key=`canvases/${canvas}/${randomUUID()}.json`,body=JSON.stringify(value);if(Buffer.byteLength(body)>15000000)throw new CanvasError(413,'画布不能超过 15MB');await storage().send(new PutObjectCommand({Bucket:bucket(),Key:key,Body:body,ContentType:'application/json'}));return key}
|
|
export async function authorizeCanvas(id:string,owner:string|null,bearer:string|null){
|
|
const record=await db().query('SELECT * FROM canvas_documents WHERE id=$1',[id]);const doc=record.rows[0];
|
|
if(!doc)throw new CanvasError(404,'画布不存在');
|
|
if(owner===doc.owner_id)return doc;
|
|
if(bearer&&/^[a-f\d]{64}$/.test(bearer)){
|
|
const auth=await db().query('UPDATE canvas_agent_links SET last_seen_at=now() WHERE canvas_id=$1 AND digest=$2 AND expires_at>now() RETURNING canvas_id',[id,tokenDigest(bearer)]);
|
|
if(auth.rowCount)return doc;
|
|
}
|
|
throw new CanvasError(404,'画布不存在或无访问权限');
|
|
}
|
|
export async function canvasState(record:{id:string;revision:number;name:string;document_key:string}){
|
|
const tasks=await db().query('SELECT id,target_id AS "targetId",status,error,source_revision AS "sourceRevision",created_at AS "createdAt" FROM canvas_tasks WHERE canvas_id=$1 ORDER BY created_at DESC LIMIT 80',[record.id]);
|
|
return {id:record.id,name:record.name,revision:record.revision,document:await readCanvasBody(record.document_key),tasks:tasks.rows};
|
|
}
|
|
export async function createCanvas(owner:string,input:unknown,name='未命名画布'){
|
|
if(typeof name!=='string'||!name.trim()||name.length>200)throw new CanvasError(400,'画布名称无效');
|
|
const document=inputDocument(input),id=randomUUID(),key=await putJson(id,document);
|
|
try{const r=await db().query('INSERT INTO canvas_documents(id,owner_id,name,document_key) VALUES($1,$2,$3,$4) RETURNING *',[id,owner,name.trim(),key]);return canvasState(r.rows[0])}catch(e){await removeObject(key).catch(()=>{});throw e}
|
|
}
|
|
export async function saveCanvas(id:string,expected:number,input:unknown){
|
|
if(!revision(expected))throw new CanvasError(400,'版本无效');const document=inputDocument(input),key=await putJson(id,document);
|
|
try{return await transaction(async client=>{const r=await client.query('UPDATE canvas_documents SET document_key=$3,revision=revision+1,updated_at=now() WHERE id=$1 AND revision=$2 RETURNING *',[id,expected,key]);if(!r.rowCount)throw new CanvasError(409,'在线画布已有更新，请先读取新版本');return r.rows[0]})}catch(e){await removeObject(key).catch(()=>{});throw e}
|
|
}
|
|
export async function createCanvasLink(id:string){
|
|
const token=newToken(),expiresAt=new Date(Date.now()+3600000).toISOString();
|
|
await transaction(async client=>{await client.query('SELECT id FROM canvas_documents WHERE id=$1 FOR UPDATE',[id]);await client.query('DELETE FROM canvas_agent_links WHERE canvas_id=$1',[id]);await client.query('INSERT INTO canvas_agent_links(digest,canvas_id,expires_at) VALUES($1,$2,$3)',[tokenDigest(token),id,expiresAt])});
|
|
return {token,expiresAt};
|
|
}
|
|
export async function queueCanvasTask(id:string,expected:number,target:string){
|
|
if(!revision(expected)||typeof target!=='string'||target.length>100)throw new CanvasError(400,'任务参数无效');
|
|
let key:string|undefined;
|
|
try{return await transaction(async client=>{
|
|
const row=(await client.query('SELECT * FROM canvas_documents WHERE id=$1 FOR UPDATE',[id])).rows[0];if(!row)throw new CanvasError(404,'画布不存在');if(row.revision!==expected)throw new CanvasError(409,'画布已有更新，请先同步');
|
|
const body=await readCanvasBody(row.document_key);let request;try{request=agentTask(body,target)}catch(e){throw new CanvasError(400,e instanceof Error?e.message:'创作要求无效')}key=await putJson(id,request);
|
|
const result=await client.query('INSERT INTO canvas_tasks(id,canvas_id,target_id,source_revision,request_key) VALUES($1,$2,$3,$4,$5) RETURNING id,status,target_id AS "targetId"',[randomUUID(),id,target,expected,key]);return result.rows[0];
|
|
})}catch(e){if(key)await removeObject(key).catch(()=>{});if((e as {code?:string}).code==='23505')throw new CanvasError(409,'该节点已有待处理任务');throw e}
|
|
}
|
|
export async function readCanvasTask(id:string,task:string){
|
|
const r=await db().query('SELECT * FROM canvas_tasks WHERE canvas_id=$1 AND id=$2',[id,task]);if(!r.rowCount)throw new CanvasError(404,'任务不存在');const row=r.rows[0],o=await getObject(row.request_key);return {id:row.id,targetId:row.target_id,status:row.status,sourceRevision:row.source_revision,error:row.error,request:JSON.parse(await o.Body!.transformToString())};
|
|
}
|
|
export async function updateCanvasTask(id:string,task:string,input:{status?:unknown;result?:unknown;error?:unknown}){
|
|
let key:string|undefined;
|
|
try{return await transaction(async client=>{
|
|
const record=(await client.query('SELECT * FROM canvas_documents WHERE id=$1 FOR UPDATE',[id])).rows[0];
|
|
const row=(await client.query('SELECT * FROM canvas_tasks WHERE canvas_id=$1 AND id=$2 FOR UPDATE',[id,task])).rows[0];
|
|
if(!record||!row)throw new CanvasError(404,'任务不存在');
|
|
if(input.status==='running'){
|
|
if(!['queued','running'].includes(row.status))throw new CanvasError(409,'任务已经结束');
|
|
await client.query("UPDATE canvas_tasks SET status='running',updated_at=now() WHERE id=$1",[task]);return {id:task,status:'running',revision:record.revision};
|
|
}
|
|
if(input.status==='failed'){
|
|
if(!['queued','running'].includes(row.status))throw new CanvasError(409,'任务已经结束');
|
|
if(typeof input.error!=='string'||!input.error.trim()||input.error.length>1000)throw new CanvasError(400,'请提供任务失败原因');
|
|
await client.query("UPDATE canvas_tasks SET status='failed',error=$2,updated_at=now() WHERE id=$1",[task,input.error]);return {id:task,status:'failed',revision:record.revision};
|
|
}
|
|
if(input.status!=='succeeded'||!input.result||typeof input.result!=='object'||Array.isArray(input.result))throw new CanvasError(400,'任务更新格式不正确');
|
|
const result=input.result as {text?:unknown;image?:unknown};
|
|
if(result.text===undefined&&result.image===undefined)throw new CanvasError(400,'没有返回结果');
|
|
const digest=createHash('sha256').update(JSON.stringify({text:result.text,image:result.image})).digest('hex');
|
|
if(row.status==='succeeded'&&row.result_digest===digest)return {id:task,status:'succeeded',revision:record.revision};
|
|
if(row.status!=='running')throw new CanvasError(409,'请先领取任务，已结束的任务不能改写');
|
|
if(record.revision!==row.source_revision)throw new CanvasError(409,'画布在任务创建后已修改，请重新派发任务');
|
|
const doc=await readCanvasBody(record.document_key),node=doc.cards.find(n=>n.id===row.target_id);if(!node)throw new CanvasError(409,'目标节点已移除');
|
|
if(result.image!==undefined&&node.kind!=='image')throw new CanvasError(400,'该节点不接收图片结果');
|
|
const next=inputDocument({...doc,cards:doc.cards.map(n=>n.id===node.id?{...n,...(result.text===undefined?{}:{text:result.text}),...(result.image===undefined?{}:{image:result.image})}:n)});
|
|
key=await putJson(id,next);await client.query('UPDATE canvas_documents SET document_key=$2,revision=revision+1,updated_at=now() WHERE id=$1',[id,key]);
|
|
await client.query("UPDATE canvas_tasks SET status='succeeded',result_digest=$2,error=NULL,updated_at=now() WHERE id=$1",[task,digest]);return {id:task,status:'succeeded',revision:record.revision+1};
|
|
})}catch(e){if(key)await removeObject(key).catch(()=>{});throw e}
|
|
}
|