from concurrent import futures
from pathlib import Path
import grpc, time, threading, json, queue
ROOT=Path(__file__).resolve().parents[1]
SERVICE='maknow.dat016.InventoryFeed'
WATCH=f'/{SERVICE}/WatchUpdates'
SNAP=f'/{SERVICE}/GetSnapshot'
PAYLOAD_PREFIX=b'update:'
class State:
    def __init__(self):
        self.cleanup_seen=threading.Event(); self.write_samples=[]; self.messages_attempted=0

def _unary(req, context):
    return b'snapshot:ok'

def make_watch(state, cfg):
    payload=PAYLOAD_PREFIX + b'x'*int(cfg['payload_bytes'])
    def watch(req, context):
        try:
            for i in range(int(cfg['message_count'])):
                if not context.is_active():
                    state.cleanup_seen.set(); break
                state.messages_attempted += 1
                t=time.monotonic()
                yield payload + str(i).encode()
                state.write_samples.append(round((time.monotonic()-t)*1000,3))
                time.sleep(float(cfg.get('server_work_delay_ms',40))/1000)
        finally:
            if not context.is_active(): state.cleanup_seen.set()
    return watch

def start_server(cfg):
    state=State(); server=grpc.server(futures.ThreadPoolExecutor(max_workers=4))
    handlers={
      'GetSnapshot':grpc.unary_unary_rpc_method_handler(_unary,request_deserializer=lambda b:b,response_serializer=lambda b:b),
      'WatchUpdates':grpc.unary_stream_rpc_method_handler(make_watch(state,cfg),request_deserializer=lambda b:b,response_serializer=lambda b:b)
    }
    server.add_generic_rpc_handlers((grpc.method_handlers_generic_handler(SERVICE,handlers),))
    port=server.add_insecure_port('127.0.0.1:0');
    if port<=0: raise RuntimeError('loopback bind failed')
    server.start(); return server,state,port

def run_stream_case(cfg, deadline_ms, receiver_delay_ms):
    server,state,port=start_server(cfg)
    status='OK'; received=0; started=time.monotonic()
    try:
        with grpc.insecure_channel(f'127.0.0.1:{port}') as ch:
            call=ch.unary_stream(WATCH,request_serializer=lambda b:b,response_deserializer=lambda b:b)
            try:
                for msg in call(b'watch',timeout=deadline_ms/1000):
                    received+=1
                    if receiver_delay_ms: time.sleep(receiver_delay_ms/1000)
            except grpc.RpcError as e:
                status=e.code().name
        time.sleep(0.08)
        return {'status':status,'received':received,'elapsed_ms':round((time.monotonic()-started)*1000,3),'server_cleanup_seen':state.cleanup_seen.is_set(),'server_yield_wait_ms':state.write_samples,'messages_attempted':state.messages_attempted}
    finally:
        server.stop(grace=0).wait()

def deadlock_risk_simulation(cfg):
    q_ab=queue.Queue(maxsize=1); q_ba=queue.Queue(maxsize=1)
    q_ab.put('already-full'); q_ba.put('already-full')
    progress=[]
    def writer(q,label):
        try: q.put('burst',timeout=cfg['deadlock_watchdog_ms']/1000); progress.append(label)
        except queue.Full: pass
    a=threading.Thread(target=writer,args=(q_ab,'A_WRITE')); b=threading.Thread(target=writer,args=(q_ba,'B_WRITE'))
    a.start(); b.start(); a.join(cfg['deadlock_watchdog_ms']/1000+0.1); b.join(cfg['deadlock_watchdog_ms']/1000+0.1)
    return {'classification':'DEADLOCK_RISK_SIMULATION_ONLY','progress':progress,'bounded':True,'watchdog_ms':cfg['deadlock_watchdog_ms'],'note':'Application-level bounded simulation; not a claim about live HTTP/2 transport internals.'}

def publisher_selftest():
    cfg=json.loads((ROOT/'fixture/service-contract.json').read_text())
    a=run_stream_case(cfg,900,0)
    b=run_stream_case(cfg,160,70)
    c=deadlock_risk_simulation(cfg)
    checks=[a['received']==int(cfg['message_count']),a['status']=='OK',b['status']=='DEADLINE_EXCEEDED',b['server_cleanup_seen'] is True,c['bounded'] is True,c['classification']=='DEADLOCK_RISK_SIMULATION_ONLY',c['progress']==[]]
    return {'grpcio':grpc.__version__,'checks':checks,'pass':all(checks),'baseline':a,'deadline_case':b,'deadlock_sim':c}
if __name__=='__main__': print(json.dumps(publisher_selftest(),indent=2))