<?php
/* Shared ingest helpers for vendor clients. */
function ingest_reading(string $streamCode, string $ts, float $value, string $rawPath='', string $rawBody='', string $srcUrl=''): void {
  $s=stream_by_code($streamCode); if(!$s) return;
  if($rawBody!==''){
    $dir=__DIR__.'/../../data/raw/'.date('Y/m/d').'/'.$streamCode;
    @mkdir($dir,0775,true);
    $rawPath='data/raw/'.date('Y/m/d').'/'.$streamCode.'/'.date('His').'.json';
    file_put_contents(__DIR__.'/../../'.$rawPath,$rawBody);
    db()->prepare('INSERT INTO files_raw(stream_id,path,sha256,bytes,pulled_at,source_url,is_placeholder) VALUES(?,?,?,?,NOW(),?,0)')
      ->execute([$s['id'],$rawPath,hash('sha256',$rawBody),strlen($rawBody),$srcUrl]);
  }
  db()->prepare('INSERT INTO readings(stream_id,ts,value,is_placeholder) VALUES(?,?,?,0)')->execute([$s['id'],$ts,$value]);
  if($s['mode']!=='live'){
    db()->prepare("UPDATE streams SET mode='live' WHERE id=?")->execute([$s['id']]);
    db()->prepare('INSERT INTO events(instrument_id,ts,class,detail,is_public) VALUES(?,NOW(),?,?,1)')
      ->execute([$s['instrument_id'],'info',$s['label'].': first live data received; placeholder retired']);
    log_activity('cron','stream_went_live',$streamCode);
  }
}