import fs from 'node:fs';
import http from 'node:http';
import { DatabaseSync } from 'node:sqlite';
import * as Effect from 'effect/Effect';
import * as Layer from 'effect/Layer';
import * as Stream from 'effect/Stream';
import * as RpcClient from 'effect/rpc/RpcClient';
import * as RpcSerialization from 'effect/rpc/RpcSerialization';
import * as Socket from 'effect/socket/Socket';
import * as NodeSocket from '@effect/platform-node/NodeSocket';
import { WsRpcGroup } from '@t3tools/contracts';
const [side, dir, portText, controlText] = process.argv.slice(2);
const port=Number(portText), control=Number(controlText);
const log=(kind:string,data:unknown)=>fs.appendFileSync(`${dir}/transcript.ndjson`, JSON.stringify({at:new Date().toISOString(),side,kind,data})+'\n');
const token=fs.readFileSync(`${dir}/credential`,'utf8');
const socket=Socket.layerWebSocket(`ws://127.0.0.1:${port}/ws?orchestrationProtocol=2`,{headers:{Authorization:`Bearer ${token}`}}).pipe(Layer.provide(NodeSocket.layerWebSocketConstructor));
const protocol=RpcClient.layerProtocolSocket().pipe(Layer.provide(socket),Layer.provide(RpcSerialization.layerJson));
const program=Effect.scoped(Effect.gen(function*(){
 const client=yield* RpcClient.make(WsRpcGroup);
 yield* Stream.runForEach(client['server.getStorageCleanupReport']({}),r=>Effect.sync(()=>log('report',r))).pipe(Effect.forkScoped);
 yield* Stream.runForEach(client['orchestration.subscribeShell']({}),r=>Effect.sync(()=>log('shell',r))).pipe(Effect.forkScoped);
 yield* Stream.runForEach(client['subscribeTerminalMetadata']({}),r=>Effect.sync(()=>log('terminal-metadata',r))).pipe(Effect.forkScoped);
 const db=new DatabaseSync(`${dir}/home/userdata/statev2.sqlite`,{readOnly:true});let seq=0;
 const collect=()=>{for(const row of db.prepare('SELECT sequence,event_type,stream_id,occurred_at,payload_json FROM orchestration_events WHERE sequence > ? ORDER BY sequence').all(seq)){seq=Number(row.sequence);log('domain-event',{...row,payload:JSON.parse(String(row.payload_json)),payload_json:undefined});}};
 const timer=setInterval(collect,250);
 log('driver-ready',{port,control,observations:'Read-only domain-event observer at 250ms; event occurred_at is persisted server time. RPC uses WsRpcGroup and effect RpcSerialization.layerJson.'});
 yield* Effect.promise(()=>new Promise<void>(resolve=>{
 const server=http.createServer(async(req,res)=>{let body='';for await(const chunk of req)body+=chunk;
 try{const action=JSON.parse(body);if(action.marker){collect();log('marker',action.marker);res.end(JSON.stringify({ok:true}));return;}
 if(action.finish){collect();log('finish',{});res.end('{}');server.close();clearInterval(timer);db.close();resolve();return;}
 log('rpc-request',action);const out=await Effect.runPromise(client[action.method](action.payload??{}));log('rpc-response',{method:action.method,result:out});res.end(JSON.stringify({ok:true,result:out}));}
 catch(e){log('rpc-error',String(e));res.statusCode=500;res.end(JSON.stringify({ok:false,error:String(e)}));}});server.listen(control,'127.0.0.1');
 }));
})).pipe(Effect.provide(protocol));
await Effect.runPromise(program);