<?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]);
$reads=0;$ddl=0;$dml=0;$created=false;$indexCreated=false;$triggerCreated=false;$readJournal=[];
$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,&$readJournal,$check){
 $check(preg_match('/^SELECT\\b/i',$sql),'SELECT only');$reads++;$check($reads<=160,'SELECT cap160');
 $entry=['SQL'=>$sql,'bindings'=>array_map(static function($v){return ['value'=>$v,'type'=>$v===null?0:(is_int($v)?1:2)];},$args)];
 try{$s=$db->prepare($sql);foreach($args as $i=>$v)$s->bindValue($i+1,$v,$v===null?PDO::PARAM_NULL:(is_int($v)?PDO::PARAM_INT:PDO::PARAM_STR));$s->execute();$rows=$s->fetchAll(PDO::FETCH_ASSOC);$check(count($rows)<=1000,'Metadata/own/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);$entry['ok']=true;$readJournal[]=$entry;return $rows;}catch(Throwable $e){$entry['error']=$e->getMessage();$readJournal[]=$entry;throw $e;}
};
$spec=$plan['lanes'][$lane];$table=$spec['table'];$short=$lane==='B'?'cdx_ss7_2a261011':'CDX_SS7_2A261011';$index='UX_CDX_SS7_2A_USERID';$trigger='TRG_CDX_SS7_2A_CHAR';
$check($table===($lane==='B'?$short:'ZHCT_APP.'.$short),'Approved exact2A target');
$check(count($spec['steps'])===33,'Exact33 planned steps');
$collision=static function()use($read,$lane,$short,$index,$trigger){
 if($lane==='B')return ['table'=>$read('SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=?',[$short]),'index'=>[],'trigger'=>[]];
 return ['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]),'trigger'=>$read('SELECT TRIGGER_NAME,TABLE_NAME FROM USER_TRIGGERS WHERE TRIGGER_NAME=?',[$trigger])];
};
$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]),'triggers'=>$read('SELECT * FROM USER_TRIGGERS WHERE TABLE_NAME=?',[$short])];
};
$own=static function()use($read,$spec){return $read($spec['readback_SQL']);};
$allowedDDL=array_values(array_filter(array_merge([$spec['create_table'],$spec['create_unique_index'],$spec['create_trigger']],$spec['cleanup'])));
$execDDL=static function($sql)use($db,$allowedDDL,$check,$emit,&$ddl,$lane){$check(in_array($sql,$allowedDDL,true),'Exact approved2A DDL');$ddl++;$check($ddl<=($lane==='B'?2:6),'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);
 $check(count($baseline['columns'])===($lane==='B'?77:70),'Fresh known Staff column count');
 $check((int)$baseline['aggregate'][0]['total']===($lane==='B'?40930:2103),'Fresh known Staff count');
 $objects=$collision();$emit('collision-before',$objects);$check($objects===['table'=>[],'index'=>[],'trigger'=>[]],'Objects collision0');
 $execDDL($spec['create_table']);$created=true;$initial=$meta();$emit('metadata-created-table',$initial);
 $check(array_map('strtolower',array_column($initial['columns'],'column_name'))===['id','marker','dingtalk_userid','primary_department_uuid','dingtalk_sync_md5'],'Exact5 approved columns');
 if($spec['create_unique_index']){$execDDL($spec['create_unique_index']);$indexCreated=true;$emit('metadata-created-index',$meta());}
 if($spec['create_trigger']){$execDDL($spec['create_trigger']);$triggerCreated=true;$emit('metadata-created-trigger',$meta());}
 $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'=>'Planned PK already present from observed divergence','id'=>$pk]);continue;}
  $dml++;$check($dml<=33,'DML cap33');$entry=['label'=>$step['label'],'SQL'=>$step['SQL'],'bindings'=>[],'hypothesis'=>$step['hypothesis'],'expect_reject'=>$step['expect_reject'],'before'=>$last];
  foreach($step['binds']as $i=>$value){$type=$value===null?PDO::PARAM_NULL:(is_int($value)?PDO::PARAM_INT:PDO::PARAM_STR);$check($type===$step['PDO_types'][$i],'Exact planned PDO type');$entry['bindings'][]=['value'=>$value,'type'=>$type];}
  $emit('DML-start',$entry);
  try{$stmt=$db->prepare($step['SQL']);foreach($entry['bindings']as $i=>$bind)$stmt->bindValue($i+1,$bind['value'],$bind['type']);$entry['ok']=$stmt->execute();$entry['rowCount']=$stmt->rowCount();}
  catch(Throwable $e){$entry['ok']=false;$entry['error']=['class'=>get_class($e),'code'=>$e->getCode(),'message'=>$e->getMessage()];}
  $last=$own();$entry['after']=$last;$entry['observed_hypothesis_match']=$entry['ok']===!$step['expect_reject'];
  $check(count($last)<=6,'Own max6');foreach($last as $row)$check((int)$row['id']>=1&&(int)$row['id']<=6&&$row['marker']==='CDXSS2A','Own marker/ID');
  $entry['byte_checks']=[];foreach($last as $row){$item=['id'=>$row['id']];foreach(['dingtalk_userid'=>'userid_hex','primary_department_uuid'=>'primary_hex','dingtalk_sync_md5'=>'md5_hex']as $field=>$alias){$v=$row[$field];$item[$field]=['native_type'=>gettype($v),'bytes'=>$v===null?null:strlen($v),'sha256'=>$v===null?null:hash('sha256',$v),'HEX_exact'=>$v===null?$row[$alias]===null:($v===''?($row[$alias]===null||$row[$alias]===''):strtoupper(bin2hex($v))===strtoupper($row[$alias]))];}
   $v=$row['dingtalk_userid'];$trim=$v===null?null:rtrim($v,' ');$expected=$v===null?null:($trim===''?'#':strtoupper(bin2hex($trim)));$item['key_expected']=$expected;$item['key_actual']=$row['userid_key'];$item['key_exact']=$row['userid_key']===$expected;$entry['byte_checks'][]=$item;}
  $emit('DML-result',$entry);
 }
 $predicates=[];foreach($spec['predicate_queries']as $query){$check($query['PDO_types']===[2],'Predicate string bind');$predicates[]=['kind'=>$query['kind'],'SQL'=>$query['SQL'],'bindings'=>[['value'=>$query['binds'][0],'type'=>2]],'rows'=>$read($query['SQL'],$query['binds'])];}
 $emit('predicate-final',$predicates);
 $check($meta()===$fixedMeta,'Own exact metadata fingerprint before cleanup');$check($own()===$last,'Own exact final rows before cleanup');$emit('own-final',$last);
 foreach($spec['cleanup']as $sql)$execDDL($sql);$created=false;$indexCreated=false;$triggerCreated=false;
 $objects=$collision();$emit('collision-after',$objects);$check($objects===['table'=>[],'index'=>[],'trigger'=>[]],'Objects0 after exact cleanup');
 $final=$business();$emit('business-after',$final);$emit('read-journal',$readJournal);
 $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('read-journal',$readJournal);$emit('stopped',['lane'=>$lane,'class'=>get_class($e),'message'=>$e->getMessage(),'table_created'=>$created,'index_created'=>$indexCreated,'trigger_created'=>$triggerCreated,'DML_attempts'=>$dml,'DDL_attempts'=>$ddl,'SELECT'=>$reads]);exit(1);}