# SPDX-License-Identifier: MIT
"""Loopback-only, read-only native-ledger observer; no trading or login endpoint."""
import asyncio
import json
from pathlib import Path
import time
from aiohttp import web, WSMsgType
from audit_viewer import snapshot

ROOT=Path(__file__).resolve().parent


class Observer:
    def __init__(self, project, port=8080):
        self.path=(Path(project)/'native-live.sqlite3').resolve()
        self.port=port
        self.peers={}
        self.pending_ws=0
        self.runner=None
        self.site=None
        self.endpoint=None
        self.requests=asyncio.Semaphore(8)

    def attested(self, project):
        return self.path==(Path(project)/'native-live.sqlite3').resolve() and any(
            time.monotonic()-seen<5 for seen in list(self.peers.values()))

    async def start(self):
        @web.middleware
        async def boundary(request,handler):
            permitted={f'127.0.0.1:{self.port}',f'localhost:{self.port}'}
            origin=request.headers.get('Origin')
            if request.host not in permitted or (origin and origin not in {f'http://{h}' for h in permitted}):
                raise web.HTTPForbidden()
            if request.method!='GET': raise web.HTTPMethodNotAllowed(request.method,['GET'])
            if request.path=='/ws':
                response=await handler(request)
            else:
                if self.requests.locked(): raise web.HTTPServiceUnavailable()
                async with self.requests:
                    response=await handler(request)
            response.headers.update({'Content-Security-Policy':"default-src 'none'; script-src 'self'; style-src 'self'; connect-src 'self'; img-src 'self'; base-uri 'none'; frame-ancestors 'none'",
                'X-Content-Type-Options':'nosniff','Cache-Control':'no-store','Referrer-Policy':'no-referrer'})
            return response
        app=web.Application(middlewares=[boundary],client_max_size=32768)
        async def index(request): return web.FileResponse(ROOT/'web'/'index.html')
        async def static(request):
            names={'app.js':'application/javascript','styles.css':'text/css'}
            name=request.match_info['name']
            if name not in names: raise web.HTTPNotFound()
            return web.Response(text=(ROOT/'web'/name).read_text(),content_type=names[name])
        async def api(request):
            try:
                eid=int(request.query['event_id']) if 'event_id' in request.query else None
                milestone=request.query.get('milestone')
                if eid is not None and eid<0: raise ValueError
                if len(str(milestone or ''))>40 or len(request.query.get('date',''))>40: raise ValueError
                data=await asyncio.to_thread(snapshot,self.path,eid,request.query.get('date'),milestone)
                return self.response(data)
            except (ValueError,KeyError,ArithmeticError): raise web.HTTPBadRequest()
        async def health(request):
            return self.response({'database':str(self.path),'read_only':True,'connected_clients':len(self.peers)})
        async def stream(request):
            if len(self.peers)+self.pending_ws>=8: raise web.HTTPServiceUnavailable()
            self.pending_ws+=1
            ws=web.WebSocketResponse(heartbeat=2,max_msg_size=32768)
            # Set security headers before prepare(), since streamed headers are final.
            ws.headers.update({'Cache-Control':'no-store','X-Content-Type-Options':'nosniff'})
            try:
                await ws.prepare(request)
                self.peers[ws]=time.monotonic()
            finally:
                self.pending_ws-=1
            cursor=0
            async def send():
                nonlocal cursor
                while not ws.closed:
                    data=await asyncio.to_thread(snapshot,self.path,None,None,None,cursor,128)
                    await asyncio.wait_for(ws.send_str(self.payload(data)),3)
                    if data['stream']: cursor=data['stream'][-1]['id']
                    await asyncio.sleep(1)
            sender=asyncio.create_task(send())
            try:
                async for message in ws:
                    if sender.done():
                        sender.result()
                        break
                    if message.type==WSMsgType.TEXT:
                        value=json.loads(message.data)
                        if value.get('type')!='heartbeat': break
                        self.peers[ws]=time.monotonic()
                    elif message.type==WSMsgType.ERROR: break
            except Exception: pass
            finally:
                self.peers.pop(ws,None)
                sender.cancel()
                await asyncio.gather(sender,return_exceptions=True)
                await ws.close()
            return ws
        app.router.add_get('/',index)
        app.router.add_get('/static/{name}',static)
        app.router.add_get('/api/window',api)
        app.router.add_get('/api/health',health)
        app.router.add_get('/ws',stream)
        self.runner=web.AppRunner(app,access_log=None,shutdown_timeout=2)
        await self.runner.setup()
        self.site=web.TCPSite(self.runner,'127.0.0.1',self.port)
        await self.site.start()
        self.port=self.site._server.sockets[0].getsockname()[1]
        self.endpoint=f'http://127.0.0.1:{self.port}'
        return self

    def payload(self,data):
        payload=json.dumps(data,separators=(',',':'),allow_nan=False)
        if len(payload.encode())>2*1024*1024:
            raise ValueError('observer_response_limit')
        return payload

    def response(self,data):
        return web.Response(text=self.payload(data),content_type='application/json')

    async def close(self):
        for ws in list(self.peers): await ws.close()
        if self.runner: await self.runner.cleanup()
        self.peers.clear()