source/teaql-ts/src/core/object-storage.ts

import type { UserContext } from './context';
import { RuntimeTelemetry, startRuntimeOperation } from './telemetry';

export type StorageErrorCode = 'InvalidInput' | 'NotConfigured' | 'NotFound' | 'AccessDenied' | 'Conflict' | 'Timeout' | 'Unavailable' | 'Unsupported' | 'TooLarge' | 'Io';
export class StorageError extends Error {
  constructor(public readonly code: StorageErrorCode) { super(`Object storage: ${code}`); this.name = 'StorageError'; }
}
const inspect = Symbol.for('nodejs.util.inspect.custom');
export const MAX_OBJECT_PUT_SIZE = 5 * 1024 ** 3;
const invalid = (): never => { throw new StorageError('InvalidInput'); };
function header(value: string): string { if (typeof value !== 'string' || !value || /[^\x20-\x7e]/.test(value)) invalid(); return value; }
export class ObjectKey {
  #value: string;
  constructor(value: string) {
    if (typeof value !== 'string' || !value || new TextEncoder().encode(value).length > 1024 || /[\x00-\x1f\x7f-\x9f]/.test(value)) invalid();
    // Reject isolated UTF-16 surrogates; TextEncoder would otherwise replace them.
    for (let i=0;i<value.length;i++) { const c=value.charCodeAt(i); if(c>=0xd800&&c<=0xdbff){const n=value.charCodeAt(++i);if(!(n>=0xdc00&&n<=0xdfff))invalid();} else if(c>=0xdc00&&c<=0xdfff)invalid(); }
    this.#value = value;
  }
  get value(): string { return this.#value; }
  toString(): string { return 'ObjectKey[masked]'; }
  [inspect](): string { return this.toString(); }
}
export type ObjectWriteMode = 'create_only' | 'replace';
export class ObjectPutOptions {
  #size: number; #mode: ObjectWriteMode; #contentType: string; #metadata: Readonly<Record<string,string>>;
  constructor(size: number, mode: ObjectWriteMode, contentType='application/octet-stream', metadata: Record<string,string>={}) {
    if (!Number.isSafeInteger(size) || size<0 || !['create_only','replace'].includes(mode)) invalid();
    if(size>MAX_OBJECT_PUT_SIZE)throw new StorageError('TooLarge');
    header(contentType);
    for(const [key,value] of Object.entries(metadata)){if(!/^[a-z0-9][a-z0-9-]*$/.test(key))invalid();header(value);}
    this.#size=size;this.#mode=mode;this.#contentType=contentType;this.#metadata=Object.freeze({...metadata});
  }
  get size(): number{return this.#size;}
  get mode(): ObjectWriteMode{return this.#mode;}
  get contentType(): string{return this.#contentType;}
  get metadata(): Readonly<Record<string,string>>{return this.#metadata;}
  toString(): string{return `ObjectPutOptions[size=${this.size}, mode=${this.mode}, metadata=masked]`;}
  [inspect](): string{return this.toString();}
}
export class ObjectMetadata {
  #key: ObjectKey;#size:number;#contentType?:string;#etag?:string;#lastModified?:Date;#metadata:Readonly<Record<string,string>>;
  constructor(key:ObjectKey,size:number,contentType?:string,etag?:string,lastModified?:Date,metadata:Record<string,string>={}){
    if(!Number.isSafeInteger(size)||size<0)throw new StorageError('Unavailable');
    this.#key=key;this.#size=size;this.#contentType=contentType;this.#etag=etag;this.#lastModified=lastModified&&new Date(lastModified);this.#metadata=Object.freeze({...metadata});
  }
  get key():ObjectKey{return this.#key;}get size():number{return this.#size;}get contentType():string|undefined{return this.#contentType;}
  get etag():string|undefined{return this.#etag;}get lastModified():Date|undefined{return this.#lastModified&&new Date(this.#lastModified);}
  get customMetadata():Readonly<Record<string,string>>{return this.#metadata;}
  toString():string{return `ObjectMetadata[size=${this.size}, details=masked]`;}[inspect]():string{return this.toString();}
}
export class ObjectReadUrlOptions {
  #seconds:number;#disposition?:string;
  constructor(seconds=300,disposition?:'inline'|'attachment',filename?:string){
    if(!Number.isInteger(seconds)||seconds<=0||seconds>604800||disposition!==undefined&&!['inline','attachment'].includes(disposition))invalid();
    if(filename!==undefined){header(filename);if(!disposition||/[\/\\"]/.test(filename))invalid();}
    this.#seconds=seconds;this.#disposition=disposition+(filename===undefined?'':`; filename="${filename}"`);
    if(disposition===undefined)this.#disposition=undefined;
  }
  get seconds():number{return this.#seconds;}get contentDisposition():string|undefined{return this.#disposition;}
  toString():string{return 'ObjectReadUrlOptions[details=masked]';}[inspect]():string{return this.toString();}
}
export class ObjectReadUrl {
  #url:string;#expiresAt:Date;
  constructor(url:string,expiresAt:Date){this.#url=url;this.#expiresAt=new Date(expiresAt);}
  get url():string{return this.#url;}get expiresAt():Date{return new Date(this.#expiresAt);}
  toString():string{return 'ObjectReadUrl[url=masked]';}[inspect]():string{return this.toString();}
}
export type ObjectByteStream = AsyncIterable<Uint8Array>;
export type StorageInvocation = Readonly<{context:UserContext;comment:string}>;
export interface ObjectStorageProvider {
  put(invocation:StorageInvocation,key:ObjectKey,content:ObjectByteStream,options:ObjectPutOptions):Promise<ObjectMetadata>;
  get(invocation:StorageInvocation,key:ObjectKey):Promise<ObjectContent>;
  head(invocation:StorageInvocation,key:ObjectKey):Promise<ObjectMetadata|undefined>;
  delete(invocation:StorageInvocation,key:ObjectKey):Promise<void>;
  getReadUrl(invocation:StorageInvocation,key:ObjectKey,options:ObjectReadUrlOptions):Promise<ObjectReadUrl>;
}
export type ObjectStorageEvent = Readonly<{operation:string;comment:string;traceId:string;outcome:string;elapsedMs:number}>;
export type ObjectStorageObserver = (event:ObjectStorageEvent)=>void;
function sanitized(error:unknown):StorageError{return error instanceof StorageError?error:new StorageError('Unavailable');}
const safe=(text:unknown):string=>String(text??'').replace(/[\x00-\x1f\x7f-\x9f]/g,' ');
class Receipt {
  #ended=false;#started=performance.now();readonly scope;
  constructor(private invocation:StorageInvocation,private operation:string){this.scope=startRuntimeOperation(invocation.context.getResource<RuntimeTelemetry>('runtimeTelemetry'),{family:'object_storage',name:operation});}
  finish(outcome:string):void{
    if(this.#ended)return;this.#ended=true;
    if(outcome==='success')this.scope.success();else this.scope.failure(new StorageError(outcome==='cancelled'?'Io':outcome as StorageErrorCode));
    const event=Object.freeze({operation:this.operation,comment:safe(this.invocation.comment),traceId:safe(this.invocation.context.getResource('traceId')),outcome,elapsedMs:performance.now()-this.#started});
    try{const observer=this.invocation.context.getResource<ObjectStorageObserver>('objectStorageObserver');if(observer)observer(event);else console.info(`ObjectStorage operation=${event.operation} comment=${event.comment} trace=${event.traceId} outcome=${event.outcome} elapsedMs=${event.elapsedMs.toFixed(3)}`);}catch{/* fail open */}
  }
}
export class ObjectContent {
  #metadata:ObjectMetadata;#stream:ObjectByteStream;#close:()=>Promise<void>;
  constructor(metadata:ObjectMetadata,stream:ObjectByteStream,close:()=>Promise<void>){this.#metadata=metadata;this.#stream=stream;this.#close=close;}
  get metadata():ObjectMetadata{return this.#metadata;}get stream():ObjectByteStream{return this.#stream;}
  close():Promise<void>{return this.#close();}
  toString():string{return 'ObjectContent[masked]';}[inspect]():string{return this.toString();}
}
export class ObjectStorage {
  #invocation:StorageInvocation;#provider:ObjectStorageProvider;
  constructor(context:UserContext,comment:string){
    if(typeof comment!=='string'||!comment.trim())invalid();const provider=context.getResource<ObjectStorageProvider>('objectStorageProvider');if(!provider)throw new StorageError('NotConfigured');
    this.#invocation=Object.freeze({context,comment});this.#provider=provider;
  }
  async #call<T>(operation:string,work:()=>Promise<T>):Promise<T>{const receipt=new Receipt(this.#invocation,operation);try{const result=await (receipt.scope.run?receipt.scope.run(work):work());receipt.finish('success');return result;}catch(error){const safeError=sanitized(error);receipt.finish(safeError.code);throw safeError;}}
  async put(key:ObjectKey,content:ObjectByteStream,options:ObjectPutOptions):Promise<ObjectMetadata>{
    const iterator=content[Symbol.asyncIterator]();let done=false;
    const owned:AsyncIterableIterator<Uint8Array>={next:async()=>{const result=await iterator.next();if(result.done)done=true;return result;},return:async()=>{if(!done){done=true;await iterator.return?.();}return {done:true,value:undefined};},[Symbol.asyncIterator](){return this;}};
    return this.#call('put',async()=>{try{return await this.#provider.put(this.#invocation,key,owned,options);}finally{await owned.return!();}});
  }
  async putBytes(key:ObjectKey,bytes:Uint8Array,options:ObjectPutOptions):Promise<ObjectMetadata>{return this.put(key,(async function*(){yield bytes;})(),options);}
  async get(key:ObjectKey):Promise<ObjectContent>{
    const receipt=new Receipt(this.#invocation,'get');let response:ObjectContent;
    try{response=await (receipt.scope.run?receipt.scope.run(()=>this.#provider.get(this.#invocation,key)):this.#provider.get(this.#invocation,key));}catch(error){const safeError=sanitized(error);receipt.finish(safeError.code);throw safeError;}
    let closed=false,started=false;
    const close=async():Promise<void>=>{if(closed)return;closed=true;try{await response.close();receipt.finish('cancelled');}catch{receipt.finish('Io');throw new StorageError('Io');}};
    const stream: ObjectByteStream = {[Symbol.asyncIterator]:async function*(){
      if(started||closed)throw new StorageError('Io');started=true;let remaining=response.metadata.size;
      try{for await(const bytes of response.stream){if(!(bytes instanceof Uint8Array)||bytes.byteLength>remaining)throw new StorageError('Io');remaining-=bytes.byteLength;yield bytes;}if(remaining!==0)throw new StorageError('Io');receipt.finish('success');}
      catch{receipt.finish('Io');throw new StorageError('Io');}finally{await close();}
    }};
    return new ObjectContent(response.metadata,stream,close);
  }
  async getBytes(key:ObjectKey,maximumBytes:number):Promise<Uint8Array>{
    if(!Number.isSafeInteger(maximumBytes)||maximumBytes<0)invalid();const response=await this.get(key);
    try{if(response.metadata.size>maximumBytes)throw new StorageError('TooLarge');const chunks:Uint8Array[]=[];let size=0;for await(const block of response.stream){if(block.byteLength>maximumBytes-size)throw new StorageError('TooLarge');size+=block.byteLength;chunks.push(block);}const result=new Uint8Array(size);let offset=0;for(const block of chunks){result.set(block,offset);offset+=block.byteLength;}return result;}finally{await response.close();}
  }
  head(key:ObjectKey):Promise<ObjectMetadata|undefined>{return this.#call('head',()=>this.#provider.head(this.#invocation,key));}
  async exists(key:ObjectKey):Promise<boolean>{return (await this.head(key))!==undefined;}
  delete(key:ObjectKey):Promise<void>{return this.#call('delete',()=>this.#provider.delete(this.#invocation,key));}
  getReadUrl(key:ObjectKey,options=new ObjectReadUrlOptions()):Promise<ObjectReadUrl>{return this.#call('read_url',()=>this.#provider.getReadUrl(this.#invocation,key,options));}
}