68 lines
5.3 KiB
TypeScript
68 lines
5.3 KiB
TypeScript
import {hash} from 'node:crypto';
|
|
import {z} from 'zod';
|
|
import type {BatchLeaseContext} from '@/lib/dn-pipeline/lock';
|
|
import type {BatchMutation,MutationRecord} from '@/lib/dn-pipeline/batch-mutations';
|
|
import type {PurchasingContext,OperationResult,MutationInput} from './types';
|
|
import {PurchasingError} from './errors';
|
|
import {withResourceLease} from './resource-lease';
|
|
import {withBatchLease} from '@/lib/dn-pipeline/lock';
|
|
const normalize=(v:unknown):unknown=>Array.isArray(v)?v.map(normalize):v&&typeof v==='object'?Object.fromEntries(Object.entries(v).filter(([k])=>k!=='operationId').sort(([a],[b])=>a.localeCompare(b)).map(([k,x])=>[k,normalize(x)])):v;
|
|
export function operationPayloadHash(tool:string,input:Record<string,unknown>):string{return hash('sha256',JSON.stringify([tool,normalize(input)]));}
|
|
export function operationRecordId(credential:string,tool:string,id:string):string{return hash('sha256',JSON.stringify([credential,tool,id])).slice(0,15);}
|
|
export const MutationSchema=z.object({operationId:z.string().uuid(),reason:z.string().max(2000).optional()});
|
|
export async function getOperation(ctx:PurchasingContext,operationId:string):Promise<OperationResult>{
|
|
z.string().uuid().parse(operationId);
|
|
const rows=await ctx.pb.collection('mcp_operations').getFullList({filter:ctx.pb.filter('credential={:credential} && operation_id={:id}',{credential:ctx.principal.id,id:operationId}),fields:'id,operation_id,status,result'});
|
|
if(rows.length!==1)throw new PurchasingError(rows.length?'operation_id_ambiguous':'operation_not_found',404);
|
|
return {operationId,status:rows[0].status,result:rows[0].result??null};
|
|
}
|
|
export async function runOperation(ctx:PurchasingContext,tool:string,input:MutationInput & Record<string,unknown>,commit:(lease:BatchLeaseContext)=>Promise<Record<string,unknown>>,resourceKey?:string,options?:{scopeKey?:string;resourceKey?:string;commitFence?:BatchMutation}):Promise<OperationResult>{
|
|
MutationSchema.parse(input);
|
|
if(options?.resourceKey){
|
|
const {resourceKey:extraKey,...rest}=options;
|
|
const outer=await withResourceLease(ctx.pb,extraKey,ctx.principal.id,lease=>runOperation(ctx,tool,input,commit,resourceKey,{...rest,commitFence:lease.fence()}));
|
|
if(!outer.acquired)throw new PurchasingError('resource_locked',409);
|
|
return outer.value;
|
|
}
|
|
const id=operationRecordId(ctx.principal.id,tool,input.operationId),payloadHash=operationPayloadHash(tool,input);
|
|
const acquire=resourceKey?.startsWith('batch:')?withBatchLease:withResourceLease;
|
|
const key=resourceKey?.startsWith('batch:')?resourceKey.slice(6):resourceKey??`operation:${id}`;
|
|
const result=await acquire(ctx.pb,key,ctx.principal.id,async lease=>{
|
|
let operation;
|
|
try{operation=await ctx.pb.collection('mcp_operations').getOne(id);}catch(error){if((error as {status?:number}).status!==404)throw new PurchasingError('mcp_schema_required',503);}
|
|
if(operation){
|
|
if(operation.payload_hash!==payloadHash)throw new PurchasingError('operation_payload_conflict',409);
|
|
if(operation.status==='committed')return {operationId:input.operationId,status:'committed' as const,result:operation.result??null};
|
|
// Failed or uncertain writes require reconciliation rather than a blind duplicate retry.
|
|
if(operation.status==='uncertain')throw new PurchasingError('operation_reconciliation_required',409);
|
|
}else{
|
|
await lease.create('mcp_operations',{id,credential:ctx.principal.id,tool,operation_id:input.operationId,payload_hash:payloadHash,status:'pending'});
|
|
}
|
|
const pending:BatchMutation[]=[];
|
|
const buffered:BatchLeaseContext={...lease,
|
|
batch:async mutations=>{
|
|
if(pending.length+mutations.length>(options?.commitFence?47:48))throw new PurchasingError('transaction_too_large');
|
|
for(const mutation of mutations){if(mutation.method==='delete')throw new PurchasingError('business_delete_forbidden',403);pending.push(mutation);}
|
|
return mutations.map(m=>({...(m.method!=='delete'&&!(m.data instanceof FormData)?m.data:{}),id:m.method==='create'?String((m.data as Record<string,unknown>).id??''):m.id}) as MutationRecord);
|
|
},
|
|
create:async(collection,data)=>(await buffered.batch([{collection,method:'create',data}]))[0]!,
|
|
update:async(collection,recordId,data)=>(await buffered.batch([{collection,method:'update',id:recordId,data}]))[0]!,
|
|
delete:async()=>{throw new PurchasingError('business_delete_forbidden',403);},
|
|
};
|
|
let submitted=false;
|
|
try{
|
|
const value=await commit(buffered);
|
|
submitted=true;
|
|
await lease.batch([...(options?.commitFence?[options.commitFence]:[]),...pending,{collection:'mcp_operations',method:'update',id,data:{status:'committed',result:value}}]);
|
|
return {operationId:input.operationId,status:'committed' as const,result:value};
|
|
}catch(error){
|
|
// A lost response may hide a successful atomic commit. Preserve committed state if readable.
|
|
const saved=await ctx.pb.collection('mcp_operations').getOne(id).catch(()=>null);
|
|
if(saved?.status==='committed')return {operationId:input.operationId,status:'committed' as const,result:saved.result??null};
|
|
await lease.update('mcp_operations',id,{status:submitted&&![400,401,403,409,422].includes(Number((error as {status?:number})?.status))?'uncertain':'failed',error_code:error instanceof PurchasingError?error.code:'operation_failed'}).catch(()=>undefined);
|
|
throw error;
|
|
}
|
|
},options);
|
|
if(!result.acquired)throw new PurchasingError('resource_locked',409);
|
|
return result.value;
|
|
}
|