<?php
$lane=getenv('SS_LANE');$root=getenv('SS_ROOT');$cfg=(static function($file){return require $file;})($root.'/application/database.php');
if($lane==='B'){$local=require '/workspace/zhct/zhctproject/store/application/config/local/database.php';foreach(['hostname','database','username','password']as $key)if((string)$cfg[$key]!==$local[$key])throw new RuntimeException('Not code local');$dsn='mysql:host='.$cfg['hostname'].';port='.($cfg['hostport']?:3306).';dbname='.$cfg['database'].';charset=utf8mb4';}
else{if($cfg['hostname']!=='127.0.0.1'||(string)$cfg['hostport']!=='5237'||$cfg['database']!=='ZHCT_APP')throw new RuntimeException('Not nonprod5237/ZHCT_APP');$dsn='dm:host=127.0.0.1;port=5237;schema=ZHCT_APP';}
$db=new PDO($dsn,$cfg['username'],$cfg['password'],[PDO::ATTR_ERRMODE=>PDO::ERRMODE_EXCEPTION,PDO::ATTR_CASE=>PDO::CASE_LOWER,PDO::ATTR_STRINGIFY_FETCHES=>false,PDO::ATTR_EMULATE_PREPARES=>false]);$journal=[];
$q=static function($sql,$args=[])use($db,&$journal){if(!preg_match('/^SELECT\b/i',$sql))throw new RuntimeException('SELECT only');$item=['SQL'=>$sql,'bindings'=>$args];try{$s=$db->prepare($sql);$s->execute($args);$rows=$s->fetchAll(PDO::FETCH_ASSOC);if(count($rows)>1000)throw new RuntimeException('Metadata/aggregate cap1000');foreach($rows as &$row){$row=array_change_key_case($row,CASE_LOWER);foreach($row as &$v)if(is_resource($v))$v=stream_get_contents($v);unset($v);}unset($row);$item['ok']=true;$journal[]=$item;return $rows;}catch(Throwable $e){$item['error']=$e->getMessage();$journal[]=$item;return ['read_error'=>$e->getMessage()];}};

$reads=0;$ddl=0;$dml=0;$created=false;$indexCreated=false;$events=[];
$emit=static function($kind,$data){echo json_encode(['kind'=>$kind,'data'=>$data],JSON_UNESCAPED_UNICODE|JSON_PRESERVE_ZERO_FRACTION),"\n";flush();};
$check=static function($ok,$reason){if(!$ok)throw new RuntimeException($reason);};
$read=static function($sql,$args=[])use($db,&$reads,$check){
 $check(preg_match('/^SELECT\\b/i',$sql)||$sql==='SHOW WARNINGS','SELECT/diagnostic only');
 $reads++;$check($reads<=120,'SELECT cap120');
 $s=$db->prepare($sql);$s->execute($args);$rows=$s->fetchAll(PDO::FETCH_ASSOC);$check(count($rows)<=1000,'Metadata/own/aggregate cap');
 foreach($rows as &$row){$row=array_change_key_case($row,CASE_LOWER);foreach($row as &$v)if(is_resource($v))$v=stream_get_contents($v);unset($v);}unset($row);return $rows;
};
$spec=$plan['lanes'][$lane];$table=$spec['table'];$short=$lane==='B'?'cdx_ss7_261011':'CDX_SS7_261011';$index='UX_CDX_SS7_USERID';
$check($table===($lane==='B'?$short:'ZHCT_APP.'.$short),'Approved exact target');
$collision=static function()use($read,$lane,$short,$index){
 return $lane==='B'?['table'=>$read('SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=?',[$short]),'index'=>[]]:['table'=>$read('SELECT TABLE_NAME FROM USER_TABLES WHERE TABLE_NAME=?',[$short]),'index'=>$read('SELECT INDEX_NAME,TABLE_NAME FROM USER_INDEXES WHERE INDEX_NAME=?',[$index])];
};
$business=static function()use($read,$lane){
 $t=$lane==='B'?'ydy_staff':'YDY_STAFF';$out=[];
 if($lane==='B'){
  $out['columns']=$read('SELECT * FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? ORDER BY ORDINAL_POSITION',[$t]);
  $out['indexes']=$read('SELECT * FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? ORDER BY INDEX_NAME,SEQ_IN_INDEX',[$t]);
  $out['constraints']=$read('SELECT * FROM information_schema.TABLE_CONSTRAINTS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? ORDER BY CONSTRAINT_NAME',[$t]);
  $out['keys']=$read('SELECT * FROM information_schema.KEY_COLUMN_USAGE WHERE TABLE_SCHEMA=DATABASE() AND (TABLE_NAME=? OR REFERENCED_TABLE_NAME=?) ORDER BY TABLE_NAME,CONSTRAINT_NAME,ORDINAL_POSITION',[$t,$t]);
  $out['triggers']=$read('SELECT * FROM information_schema.TRIGGERS WHERE EVENT_OBJECT_SCHEMA=DATABASE() AND EVENT_OBJECT_TABLE=?',[$t]);
 }else{
  $out['columns']=$read('SELECT * FROM USER_TAB_COLUMNS WHERE TABLE_NAME=? ORDER BY COLUMN_ID',[$t]);
  $out['indexes']=$read('SELECT INDEX_NAME,INDEX_TYPE,UNIQUENESS,TABLE_NAME FROM USER_INDEXES WHERE TABLE_NAME=? ORDER BY INDEX_NAME',[$t]);
  $out['index_columns']=$read('SELECT * FROM USER_IND_COLUMNS WHERE TABLE_NAME=? ORDER BY INDEX_NAME,COLUMN_POSITION',[$t]);
  $out['constraints']=$read('SELECT * FROM USER_CONSTRAINTS WHERE TABLE_NAME=? ORDER BY CONSTRAINT_NAME',[$t]);
  $out['constraint_columns']=$read('SELECT * FROM USER_CONS_COLUMNS WHERE TABLE_NAME=? ORDER BY CONSTRAINT_NAME,POSITION',[$t]);
  $out['triggers']=$read('SELECT * FROM USER_TRIGGERS WHERE TABLE_NAME=?',[$t]);
 }
 $names=array_map('strtolower',array_column($out['columns'],'column_name'));$fields=['COUNT(*) AS TOTAL'];
 foreach(['dingtalk_avatar','update_time','dingtalk_userid','primary_department_uuid','dingtalk_last_seen_full_batch_id','dingtalk_sync_time','dingtalk_sync_md5']as $name){
  if(!in_array($name,$names,true))continue;$col=$lane==='B'?chr(96).$name.chr(96):'"'.strtoupper($name).'"';
  $fields[]='SUM(CASE WHEN '.$col.' IS NULL THEN 1 ELSE 0 END) AS '.$name.'_NULLS';
  if(in_array($name,['update_time','dingtalk_sync_time'],true)){
   $e=$lane==='B'?"DATE_FORMAT(".$col.",'%Y-%m-%d %H:%i:%s.%f')":"TO_CHAR(".$col.",'YYYY-MM-DD HH24:MI:SS.FF6')";
   $fields[]='MIN('.$e.') AS '.$name.'_MIN';$fields[]='MAX('.$e.') AS '.$name.'_MAX';
  }elseif($name==='dingtalk_last_seen_full_batch_id'){
   $cast=$lane==='B'?'CHAR':'VARCHAR(40)';$fields[]='CAST(MIN('.$col.') AS '.$cast.') AS '.$name.'_MIN';$fields[]='CAST(MAX('.$col.') AS '.$cast.') AS '.$name.'_MAX';
  }else{
   $chars=$lane==='B'?'CHAR_LENGTH':'LENGTH';$bytes=$lane==='B'?'OCTET_LENGTH':'LENGTHB';
   $fields[]='MAX('.$chars.'('.$col.')) AS '.$name.'_CHARS';$fields[]='MAX('.$bytes.'('.$col.')) AS '.$name.'_BYTES';$fields[]='COUNT(DISTINCT '.$col.') AS '.$name.'_DISTINCT';
  }
 }
 $out['aggregate']=$read('SELECT '.implode(',',$fields).' FROM '.$t);
 return $out;
};
$meta=static function()use($read,$lane,$short){
 if($lane==='B')return ['columns'=>$read('SELECT COLUMN_NAME,COLUMN_TYPE,IS_NULLABLE,COLUMN_DEFAULT,EXTRA,COLLATION_NAME,CHARACTER_MAXIMUM_LENGTH FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? ORDER BY ORDINAL_POSITION',[$short]),'indexes'=>$read('SELECT INDEX_NAME,NON_UNIQUE,COLUMN_NAME,SEQ_IN_INDEX FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? ORDER BY INDEX_NAME,SEQ_IN_INDEX',[$short])];
 return ['columns'=>$read('SELECT COLUMN_NAME,DATA_TYPE,DATA_LENGTH,DATA_PRECISION,DATA_SCALE,NULLABLE,DATA_DEFAULT,CHAR_LENGTH,CHAR_USED FROM USER_TAB_COLUMNS WHERE TABLE_NAME=? ORDER BY COLUMN_ID',[$short]),'indexes'=>$read('SELECT INDEX_NAME,INDEX_TYPE,UNIQUENESS,TABLE_NAME FROM USER_INDEXES WHERE TABLE_NAME=? ORDER BY INDEX_NAME',[$short]),'expressions'=>$read('SELECT * FROM USER_IND_EXPRESSIONS WHERE TABLE_NAME=?',[$short])];
};
$own=static function()use($read,$lane,$table){
 $hex=$lane==='B'?'HEX':'RAWTOHEX';$binary=$lane==='B'?'':'CHARTOBIN';
 $h=static function($v)use($hex,$binary){return $hex.'('.($binary?$binary.'('.$v.')':$v).')';};
 $cast=$lane==='B'?'CHAR':'VARCHAR(40)';
 return $read('SELECT T.*,CAST(DINGTALK_LAST_SEEN_FULL_BATCH_ID AS '.$cast.') AS BATCH_EXACT,'.$h('DINGTALK_AVATAR').' AS AVATAR_HEX,'.$h('DINGTALK_USERID').' AS USERID_HEX,'.$h('PRIMARY_DEPARTMENT_UUID').' AS PRIMARY_HEX,'.$h('DINGTALK_SYNC_MD5').' AS MD5_HEX FROM '.$table.' T ORDER BY ID');
};
$allowedDDL=array_values(array_filter(array_merge([$spec['create_table'],$spec['create_unique_index']],$spec['cleanup'])));
$execDDL=static function($sql)use($db,$allowedDDL,$check,$emit,&$ddl,$lane){$check(in_array($sql,$allowedDDL,true),'Exact approved DDL');$ddl++;$check($ddl<=($lane==='B'?2:4),'DDL cap');$emit('DDL-start',['SQL'=>$sql,'sha256'=>hash('sha256',$sql)]);$s=$db->prepare($sql);$s->execute();$emit('DDL-success',['SQL'=>$sql]);};
try{
 $baseline=$business();$emit('business-before',$baseline);
 $objects=$collision();$emit('collision-before',$objects);$check($objects['table']===[]&&$objects['index']===[],'Objects collision0');
 $execDDL($spec['create_table']);$created=true;$initial=$meta();$emit('metadata-created-table',$initial);
 $expectedColumns=['id','null_v','null_n','null_t','lob_value','dingtalk_avatar','update_time','dingtalk_userid','primary_department_uuid','dingtalk_last_seen_full_batch_id','dingtalk_sync_time','dingtalk_sync_md5'];
 $check(array_map('strtolower',array_column($initial['columns'],'column_name'))===$expectedColumns,'Exact12 approved columns');
 if($spec['create_unique_index']){$execDDL($spec['create_unique_index']);$indexCreated=true;}
 $fixedMeta=$meta();$emit('metadata-created-final',$fixedMeta);$last=[];
 foreach($spec['steps']as $step){
  $pk=(int)(preg_match('/^INSERT/',$step['SQL'])?$step['binds'][0]:$step['binds'][count($step['binds'])-1]);
  $existing=array_values(array_filter($last,static function($row)use($pk){return (int)$row['id']===$pk;}));
  if(preg_match('/^UPDATE/',$step['SQL'])&&count($existing)!==1){$emit('skipped',['label'=>$step['label'],'reason'=>'Dependent own ID missing','id'=>$pk]);continue;}
  if(preg_match('/^INSERT/',$step['SQL'])&&$existing!==[]){$emit('skipped',['label'=>$step['label'],'reason'=>'Exact planned PK already exists after prior observed divergence','id'=>$pk]);continue;}
  $dml++;$check($dml<=19,'DML attempt cap19');$entry=['label'=>$step['label'],'SQL'=>$step['SQL'],'bindings'=>[],'hypothesis'=>$step['hypothesis'],'before'=>$last];
  $stmt=$db->prepare($step['SQL']);foreach($step['binds']as $i=>$value){$type=$value===null?PDO::PARAM_NULL:(is_int($value)?PDO::PARAM_INT:PDO::PARAM_STR);if(strpos($step['label'],'uint64')!==false&&$i===0)$check($value===null||is_string($value),'UINT64 never float/int intermediate');$stmt->bindValue($i+1,$value,$type);$entry['bindings'][]=['value'=>$value,'type'=>$type];}
  $emit('DML-start',$entry);
  try{$ok=$stmt->execute();$entry['ok']=$ok;$entry['rowCount']=$stmt->rowCount();}catch(Throwable $e){$entry['ok']=false;$entry['error']=['class'=>get_class($e),'code'=>$e->getCode(),'message'=>$e->getMessage()];}
  if($lane==='B'&&$step['label']==='S18-md5-char')$entry['warnings']=$read('SHOW WARNINGS');
  $last=$own();$entry['after']=$last;$entry['storage_checks']=[];
  $check(count($last)<=6,'Own max6');foreach($last as $row)$check((int)$row['id']>=1&&(int)$row['id']<=6&&$row['null_v']==='CDXSS7','Own marker/ID');
  if($entry['ok']&&$pk===6){
   $row=array_values(array_filter($last,static function($row){return (int)$row['id']===6;}))[0];
   if($step['label']==='S09-max-capacities'||$step['label']==='S19-avatar-last-byte'){
    $expectedAvatar=$step['label']==='S09-max-capacities'?$step['binds'][4]:$step['binds'][0];$expectedUser=$step['label']==='S09-max-capacities'?$step['binds'][3]:$step['binds'][1];
    $entry['storage_checks']['avatar_full_native']=$row['dingtalk_avatar']===$expectedAvatar;
    $entry['storage_checks']['avatar_full_HEX']=strtoupper((string)$row['avatar_hex'])===strtoupper(bin2hex($expectedAvatar));
    $entry['storage_checks']['userid_full_native']=$row['dingtalk_userid']===$expectedUser;
    $entry['storage_checks']['userid_full_HEX']=strtoupper((string)$row['userid_hex'])===strtoupper(bin2hex($expectedUser));
    $entry['avatar_bytes']=strlen($row['dingtalk_avatar']);$entry['avatar_sha256']=hash('sha256',$row['dingtalk_avatar']);$entry['userid_bytes']=strlen($row['dingtalk_userid']);$entry['userid_sha256']=hash('sha256',$row['dingtalk_userid']);
   }
   if(strpos($step['label'],'uint64')!==false)$entry['storage_checks']['batch_exact_text']=$row['batch_exact']===$step['binds'][0];
   if($step['label']==='S09-max-capacities')$entry['storage_checks']['batch_exact_text']=$row['batch_exact']==='18446744073709551615';
   $entry['native_batch_type']=gettype($row['dingtalk_last_seen_full_batch_id']);
  }
  $emit('DML-result',$entry);
  foreach($entry['storage_checks']as $valid)$check($valid,'Full native/HEX/exact-value failure; retain unknown state');
  if(!$entry['ok']&&$last===[]&&strpos($entry['error']['message'],'LOB')!==false){$emit('dependency-stop',['label'=>$step['label'],'reason'=>'Empty LOB blocked all prescribed seeds']);break;}
 }
 $predicate=[];foreach(['A','a','A ','',' ']as $value){
  $predicate[$value]=['plain'=>$read('SELECT ID FROM '.$table.' WHERE DINGTALK_USERID=? ORDER BY ID',[$value])];
  if($lane==='C')$predicate[$value]['binary']=$read('SELECT ID FROM '.$table.' WHERE BINARY(DINGTALK_USERID)=BINARY(?) ORDER BY ID',[$value]);
 }
 $emit('predicate-final',$predicate);
 $check($meta()===$fixedMeta,'Own schema/index fingerprint before cleanup');
 $emit('own-final',$last);
 foreach($spec['cleanup']as $sql)$execDDL($sql);$created=false;$indexCreated=false;
 $objects=$collision();$emit('collision-after',$objects);$check($objects['table']===[]&&$objects['index']===[],'Objects0 after exact cleanup');
 $final=$business();$emit('business-after',$final);
 $emit('summary',['lane'=>$lane,'completed'=>true,'DML_attempts'=>$dml,'DDL_attempts'=>$ddl,'SELECT'=>$reads,'HTTP'=>0,'business_table_writes'=>0,'staff_metadata_aggregate_exact'=>$final===$baseline,'objects_remaining'=>0,'hypotheses_may_have_observed_divergence'=>true]);
}catch(Throwable $e){
 $emit('stopped',['lane'=>$lane,'class'=>get_class($e),'message'=>$e->getMessage(),'table_created'=>$created,'index_created'=>$indexCreated,'DML_attempts'=>$dml,'DDL_attempts'=>$ddl,'SELECT'=>$reads]);exit(1);
}