CREATE TABLE IF NOT EXISTS t2_sync_membro (
 raiz_origem VARCHAR(64) NOT NULL, entidade VARCHAR(64) NOT NULL, raiz VARCHAR(191) NOT NULL,
 tabela VARCHAR(64) NOT NULL, origem VARCHAR(64) NOT NULL, chave VARCHAR(191) NOT NULL,
 local_id VARCHAR(191) NOT NULL, hash_local CHAR(64) NULL,
 hash_recebido CHAR(64) NULL, ultima_origem VARCHAR(64) NULL,
 atualizado DATETIME(3) NOT NULL DEFAULT (UTC_TIMESTAMP(3)),
 PRIMARY KEY(raiz_origem,entidade,raiz,tabela,origem,chave)
) ENGINE=InnoDB;
CREATE TABLE IF NOT EXISTS t2_sync_revisao (
 origem VARCHAR(64) NOT NULL, entidade VARCHAR(64) NOT NULL, chave VARCHAR(191) NOT NULL,
 revisao BIGINT NOT NULL, hash_payload CHAR(64) NULL, PRIMARY KEY(origem,entidade,chave)
) ENGINE=InnoDB;
DELIMITER $$
DROP PROCEDURE IF EXISTS t2_sync_hash_linha$$
CREATE PROCEDURE t2_sync_hash_linha(IN p_t VARCHAR(64),IN p_id VARCHAR(191),OUT p_hash CHAR(64))
BEGIN
 DECLARE v_pk VARCHAR(64); DECLARE v_cols JSON; DECLARE v_expr LONGTEXT;
 SELECT pk,colunas INTO v_pk,v_cols FROM t2_sync_tabelas WHERE nome=p_t;
 SELECT GROUP_CONCAT(CONCAT(QUOTE(j.c),',CONVERT(t.`',IF(p_t='estoqueprodutos' AND j.c='qtd',t2_sync_estoque_coluna(),j.c),'` USING utf8mb4)') ORDER BY j.n SEPARATOR ',')
  INTO v_expr FROM JSON_TABLE(v_cols,'$[*]' COLUMNS(n FOR ORDINALITY,c VARCHAR(64) PATH '$')) j;
 SET @sync_k=p_id; SET @sync_h=NULL;
 SET @sync_sql=CONCAT('SELECT SHA2(CAST(JSON_OBJECT(',v_expr,') AS CHAR CHARACTER SET utf8mb4),256) INTO @sync_h FROM `',p_t,'` t WHERE `',v_pk,'`=? FOR UPDATE');
 PREPARE sync_stmt FROM @sync_sql; EXECUTE sync_stmt USING @sync_k; DEALLOCATE PREPARE sync_stmt;
 SET p_hash=@sync_h;
END$$
DROP PROCEDURE IF EXISTS t2_sync_estoque$$
CREATE PROCEDURE t2_sync_estoque(IN p_orig VARCHAR(64),IN p_raiz VARCHAR(191),IN p_doc INT)
BEGIN
 DECLARE fim BOOLEAN DEFAULT FALSE; DECLARE v_sku VARCHAR(191); DECLARE v_novo,v_antigo DECIMAL(18,4);
 DECLARE v_col VARCHAR(64); DECLARE v_empresa INT;
 DECLARE c CURSOR FOR
  SELECT sku,SUM(qtd) FROM (
   SELECT i.produto sku,SUM(IF(i.cancelado='N' AND m.situacao<>'C',i.quantidade,0)) qtd
    FROM itmovimentacao i JOIN movimentacao m ON m.documento=i.documento WHERE i.documento=p_doc GROUP BY i.produto
   UNION ALL SELECT sku,0 FROM t2_sync_estoque_venda WHERE origem=p_orig AND venda=p_raiz
  ) z GROUP BY sku;
 DECLARE CONTINUE HANDLER FOR NOT FOUND SET fim=TRUE;
 SET v_empresa=(SELECT empresa FROM t2_sync_config WHERE id=1);
 SET v_col=CONCAT('qtd',v_empresa);
 IF NOT EXISTS(SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='estoqueprodutos' AND column_name=v_col) THEN SET v_col='qtd'; END IF;
 OPEN c;
 la: LOOP
  FETCH c INTO v_sku,v_novo; IF fim THEN LEAVE la; END IF;
  SET v_antigo=COALESCE((SELECT quantidade FROM t2_sync_estoque_venda WHERE origem=p_orig AND venda=p_raiz AND sku=v_sku),0);
  IF v_novo<>v_antigo THEN
   IF (SELECT COUNT(*) FROM estoqueprodutos WHERE codproduto=v_sku)<>1 THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_SKU_ESTOQUE_AMBIGUO_OU_AUSENTE'; END IF;
   SET @sync_delta=v_novo-v_antigo; SET @sync_sku=v_sku;
   SET @sync_sql=CONCAT('UPDATE estoqueprodutos SET `',v_col,'`=COALESCE(`',v_col,'`,0)-? WHERE codproduto=?');
   PREPARE sync_stmt FROM @sync_sql; EXECUTE sync_stmt USING @sync_delta,@sync_sku; DEALLOCATE PREPARE sync_stmt;
   INSERT INTO t2_sync_estoque_venda VALUES(p_orig,p_raiz,v_sku,v_novo)
    ON DUPLICATE KEY UPDATE quantidade=v_novo;
  END IF;
 END LOOP;
 CLOSE c;
END$$
DROP PROCEDURE IF EXISTS t2_sync_aplicar$$
CREATE PROCEDURE t2_sync_aplicar(IN p_op CHAR(36),IN p_origem VARCHAR(64),IN p_hash CHAR(64),IN p_doc LONGTEXT)
proc: BEGIN
 DECLARE v_lado,v_dono VARCHAR(8); DECLARE v_self,v_peer,v_ent,v_root_t,v_o,v_t,v_pk,v_c,v_ref_t,v_ref_o VARCHAR(64);
 DECLARE v_root_o VARCHAR(64); DECLARE v_root_k,v_key,v_local,v_rootlocal,v_ref_k,v_ref_local VARCHAR(191);
 DECLARE v_linhas,v_members,v_r,v_d,v_rf,v_cols,v_desc_refs,v_nat JSON;
 DECLARE v_expr,v_values,v_where,v_code LONGTEXT; DECLARE v_hash,v_anterior,v_sourcehash,v_revhash CHAR(64); DECLARE v_revisao,v_ultima BIGINT; DECLARE v_stockcol VARCHAR(64);
 DECLARE v_emp,v_emp_peer,v_i,v_j,v_n,v_auto,v_exist,v_inseriu INT;
 DECLARE EXIT HANDLER FOR SQLEXCEPTION BEGIN ROLLBACK; SET @t2_sync_aplicando=0; RESIGNAL; END;
 IF p_op NOT REGEXP '^[0-9a-fA-F-]{36}$' OR JSON_VALID(p_doc)=0 OR OCTET_LENGTH(p_doc)>8388608 OR SHA2(p_doc,256)<>p_hash THEN
  SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_ENVELOPE_INVALIDO'; END IF;
 SELECT lado,origem,par,empresa,empresa_par INTO v_lado,v_self,v_peer,v_emp,v_emp_peer FROM t2_sync_config WHERE id=1;
 IF p_origem<>v_peer OR JSON_UNQUOTE(JSON_EXTRACT(p_doc,'$.origem'))<>v_peer OR JSON_EXTRACT(p_doc,'$.empresa')<>v_emp_peer THEN
  SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_ORIGEM_OU_EMPRESA_INVALIDA'; END IF;
 SET v_stockcol=t2_sync_estoque_coluna();
 SET v_ent=JSON_UNQUOTE(JSON_EXTRACT(p_doc,'$.entidade'));
 SELECT dono,tabela_raiz,tabelas INTO v_dono,v_root_t,v_members FROM t2_sync_entidades WHERE nome=v_ent;
 IF v_dono IS NULL OR (v_dono<>'BOTH' AND v_dono=v_lado) THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_DIRECAO_NAO_AUTORIZADA'; END IF;
 SET v_linhas=JSON_EXTRACT(p_doc,'$.linhas'); SET v_n=JSON_LENGTH(v_linhas);
 IF JSON_TYPE(v_linhas)<>'ARRAY' OR v_n IS NULL OR v_n<1 OR v_n>10000 OR JSON_UNQUOTE(JSON_EXTRACT(v_linhas,'$[0].tabela'))<>v_root_t THEN
  SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_AGREGADO_INVALIDO'; END IF;
 SET v_root_o=JSON_UNQUOTE(JSON_EXTRACT(v_linhas,'$[0].origem'));
 SET v_root_k=JSON_UNQUOTE(JSON_EXTRACT(v_linhas,'$[0].chave'));
 IF JSON_UNQUOTE(JSON_EXTRACT(p_doc,'$.revisao')) NOT REGEXP '^[1-9][0-9]{0,17}$' OR JSON_EXTRACT(p_doc,'$.versao')<>1 THEN
  SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_REVISAO_INVALIDA'; END IF;
 SET v_revisao=CAST(JSON_UNQUOTE(JSON_EXTRACT(p_doc,'$.revisao')) AS UNSIGNED);
 SET SESSION group_concat_max_len=16777216;
 SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
 START TRANSACTION;
 INSERT IGNORE INTO t2_sync_recibo(operacao,origem,hash_payload) VALUES(p_op,p_origem,p_hash);
 SET v_inseriu=ROW_COUNT();
 IF v_inseriu=0 THEN
  IF NOT EXISTS(SELECT 1 FROM t2_sync_recibo WHERE operacao=p_op AND origem=p_origem AND hash_payload=p_hash) THEN
   SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_IDEMPOTENCIA_CONTEUDO_DIFERENTE'; END IF;
  COMMIT; SELECT p_op AS operacao,p_hash AS hash_payload,'confirmado' AS resultado; LEAVE proc;
 END IF;
 INSERT IGNORE INTO t2_sync_revisao(origem,entidade,chave,revisao) VALUES(p_origem,v_ent,JSON_UNQUOTE(JSON_EXTRACT(p_doc,'$.chave')),0);
 SELECT revisao,hash_payload INTO v_ultima,v_revhash FROM t2_sync_revisao
  WHERE origem=p_origem AND entidade=v_ent AND chave=JSON_UNQUOTE(JSON_EXTRACT(p_doc,'$.chave')) FOR UPDATE;
 IF v_revisao<v_ultima THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_REVISAO_SUPERADA'; END IF;
 IF v_revisao=v_ultima THEN
  IF v_revhash<>p_hash THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_REVISAO_CONTEUDO_DIFERENTE'; END IF;
  COMMIT; SELECT p_op AS operacao,p_hash AS hash_payload,'confirmado' AS resultado; LEAVE proc;
 END IF;
 SET @t2_sync_aplicando=1;
 DROP TEMPORARY TABLE IF EXISTS t2_sync_recebidos;
 CREATE TEMPORARY TABLE t2_sync_recebidos(tabela VARCHAR(64),origem VARCHAR(64),chave VARCHAR(191),PRIMARY KEY(tabela,origem,chave));
 SET v_i=0;
 WHILE v_i<v_n DO
  SET v_r=JSON_EXTRACT(v_linhas,CONCAT('$[',v_i,']'));
  SET v_t=JSON_UNQUOTE(JSON_EXTRACT(v_r,'$.tabela')); SET v_o=JSON_UNQUOTE(JSON_EXTRACT(v_r,'$.origem'));
  SET v_key=JSON_UNQUOTE(JSON_EXTRACT(v_r,'$.chave')); SET v_d=JSON_EXTRACT(v_r,'$.dados'); SET v_rf=JSON_EXTRACT(v_r,'$.refs');
  SET v_sourcehash=SHA2(CAST(v_d AS CHAR CHARACTER SET utf8mb4),256);
  IF (v_o<>v_self AND v_o<>v_peer) OR v_key IS NULL OR LENGTH(v_key)>191 OR
   JSON_SEARCH(v_members,'one',v_t,NULL,'$[*].table') IS NULL THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_LINHA_NAO_AUTORIZADA'; END IF;
  SELECT pk,automatico,colunas,referencias,campos_naturais INTO v_pk,v_auto,v_cols,v_desc_refs,v_nat FROM t2_sync_tabelas WHERE nome=v_t;
  IF JSON_TYPE(v_d)<>'OBJECT' OR JSON_TYPE(v_rf)<>'OBJECT' OR JSON_LENGTH(v_d)<>JSON_LENGTH(v_cols) THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_SCHEMA_DIVERGENTE'; END IF;
  SET v_j=0;
  WHILE v_j<JSON_LENGTH(v_cols) DO
   SET v_c=JSON_UNQUOTE(JSON_EXTRACT(v_cols,CONCAT('$[',v_j,']')));
   IF JSON_CONTAINS_PATH(v_d,'one',CONCAT('$.',v_c))=0 THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_COLUNA_AUSENTE'; END IF;
   SET v_j=v_j+1;
  END WHILE;
  SET v_j=0;
  WHILE v_j<JSON_LENGTH(v_desc_refs) DO
   SET v_c=JSON_UNQUOTE(JSON_EXTRACT(JSON_KEYS(v_desc_refs),CONCAT('$[',v_j,']')));
   SET v_ref_t=JSON_UNQUOTE(JSON_EXTRACT(v_desc_refs,CONCAT('$.',v_c)));
   IF v_ref_t='@empresa' THEN
    IF IF(JSON_TYPE(JSON_EXTRACT(v_d,CONCAT('$.',v_c)))='NULL','0',COALESCE(JSON_UNQUOTE(JSON_EXTRACT(v_d,CONCAT('$.',v_c))),'0')) NOT IN ('0','',CAST(v_emp_peer AS CHAR)) THEN
     SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_ESCOPO_EMPRESA_INVALIDO'; END IF;
    SET v_d=JSON_SET(v_d,CONCAT('$.',v_c),CAST(v_emp AS CHAR));
   ELSE
    SET v_ref_k=JSON_UNQUOTE(JSON_EXTRACT(v_rf,CONCAT('$.',v_c,'.chave')));
    IF v_ref_k IS NOT NULL THEN
     SET v_ref_o=JSON_UNQUOTE(JSON_EXTRACT(v_rf,CONCAT('$.',v_c,'.origem')));
     IF JSON_UNQUOTE(JSON_EXTRACT(v_rf,CONCAT('$.',v_c,'.tabela')))<>v_ref_t THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_REFERENCIA_INVALIDA'; END IF;
     -- NF-e/caixa criados no ERP podem nao ter venda local correspondente.
     IF v_lado='PDV' AND v_ent IN('nfce','caixa') AND v_c='documento' AND
       NOT EXISTS(SELECT 1 FROM t2_sync_mapa WHERE origem=v_ref_o AND tabela=v_ref_t AND chave=v_ref_k) AND v_ref_o<>v_self THEN
      SET v_ref_local=NULL;
     ELSE CALL t2_sync_resolver(v_ref_o,v_ref_t,v_ref_k,v_ref_local); END IF;
     SET v_d=JSON_SET(v_d,CONCAT('$.',v_c),v_ref_local);
    ELSEIF IF(JSON_TYPE(JSON_EXTRACT(v_d,CONCAT('$.',v_c)))='NULL','0',COALESCE(JSON_UNQUOTE(JSON_EXTRACT(v_d,CONCAT('$.',v_c))),'0')) NOT IN('0','') THEN
     SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_REFERENCIA_OMITIDA';
    END IF;
   END IF;
   SET v_j=v_j+1;
  END WHILE;
  SET v_local=NULL;
  IF v_o=v_self THEN SET v_local=v_key;
  ELSE SET v_local=(SELECT local_id FROM t2_sync_mapa WHERE origem=v_o AND tabela=v_t AND chave=v_key); END IF;
  SET @sync_data=CAST(v_d AS CHAR CHARACTER SET utf8mb4);
  IF v_local IS NULL AND JSON_LENGTH(v_nat)>0 THEN
   SET v_j=0; SET v_where='';
   WHILE v_j<JSON_LENGTH(v_nat) DO
    SET v_c=JSON_UNQUOTE(JSON_EXTRACT(v_nat,CONCAT('$[',v_j,']')));
    IF JSON_TYPE(JSON_EXTRACT(v_d,CONCAT('$.',v_c)))='NULL' OR COALESCE(JSON_UNQUOTE(JSON_EXTRACT(v_d,CONCAT('$.',v_c))),'')='' THEN SET v_where=NULL;
    ELSE SET v_where=CONCAT(v_where,IF(v_j>0,' AND ',''),'`',v_c,'`=JSON_UNQUOTE(JSON_EXTRACT(@sync_data,',QUOTE(CONCAT('$.',v_c)),'))'); END IF;
    SET v_j=v_j+1;
   END WHILE;
   IF v_where IS NOT NULL THEN
    SET @sync_found=NULL; SET @sync_count=0;
    SET @sync_sql=CONCAT('SELECT COUNT(*),MIN(CAST(`',v_pk,'` AS CHAR)) INTO @sync_count,@sync_found FROM `',v_t,'` WHERE ',v_where);
    PREPARE sync_stmt FROM @sync_sql; EXECUTE sync_stmt; DEALLOCATE PREPARE sync_stmt;
    IF @sync_count>1 THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_CHAVE_NATURAL_AMBIGUA'; END IF;
    IF @sync_count=1 THEN SET v_local=@sync_found; END IF;
   END IF;
  END IF;
  SET v_exist=0;
  IF v_local IS NOT NULL THEN
   CALL t2_sync_hash_linha(v_t,v_local,v_hash); SET v_exist=(v_hash IS NOT NULL);
   SET v_anterior=(SELECT hash_local FROM t2_sync_membro WHERE raiz_origem=v_root_o AND entidade=v_ent AND raiz=v_root_k AND tabela=v_t AND origem=v_o AND chave=v_key);
   IF v_lado='ERP' AND v_exist=1 AND (v_dono='BOTH' OR v_t='caixa') AND
     ((v_anterior IS NOT NULL AND v_anterior<>v_hash) OR (v_anterior IS NULL AND v_o=v_self)) AND
     COALESCE(JSON_UNQUOTE(JSON_EXTRACT(v_r,'$.base_hash')),'')<>v_hash THEN
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_ERP_PREVALECE'; END IF;
  END IF;
  IF v_t='estoqueprodutos' AND v_lado='PDV' THEN
   IF EXISTS(SELECT 1 FROM itmovimentacao i JOIN movimentacao mv ON mv.documento=i.documento
     WHERE i.produto=JSON_UNQUOTE(JSON_EXTRACT(v_d,'$.codproduto')) AND i.cancelado='N' AND
     ((COALESCE(mv.finalizado,'N')<>'S' AND mv.situacao<>'C') OR
      EXISTS(SELECT 1 FROM t2_sync_fila f WHERE f.entidade='venda' AND f.chave=CAST(mv.documento AS CHAR) AND f.estado<>'concluido'))) THEN
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_ESTOQUE_COM_VENDAS_LOCAIS_PENDENTES'; END IF;
  END IF;
  IF v_t='nfe' THEN
   IF EXISTS(SELECT 1 FROM nfe nn WHERE nn.emitente=v_emp AND nn.modelo='65'
    AND nn.serie=JSON_UNQUOTE(JSON_EXTRACT(v_d,'$.serie')) AND nn.nNF=JSON_UNQUOTE(JSON_EXTRACT(v_d,'$.nNF'))
    AND nn.tpAmb=JSON_UNQUOTE(JSON_EXTRACT(v_d,'$.tpAmb')) AND nn.id<>COALESCE(v_local,0)
    AND COALESCE(nn.chavenfe,'')<>JSON_UNQUOTE(JSON_EXTRACT(v_d,'$.chavenfe'))) THEN
     SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_NUMERACAO_FISCAL_CONFLITANTE'; END IF;
   IF v_exist=1 AND EXISTS(SELECT 1 FROM nfe nn WHERE nn.id=v_local AND nn.status='101')
    AND JSON_UNQUOTE(JSON_EXTRACT(v_d,'$.status'))<>'101' THEN
     SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_STATUS_FISCAL_REGRESSIVO'; END IF;
  END IF;
  IF v_t='nfe' AND v_exist=1 THEN
   SET @sync_k=v_local; SET @sync_xml=NULL;
   SELECT CONVERT(xml USING utf8mb4) INTO @sync_xml FROM nfe WHERE id=v_local FOR UPDATE;
   IF LENGTH(COALESCE(@sync_xml,''))>0 AND @sync_xml<>JSON_UNQUOTE(JSON_EXTRACT(v_d,'$.xml')) THEN
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_XML_ORIGINAL_IMUTAVEL'; END IF;
  END IF;
  IF v_i=0 THEN
   SET v_rootlocal=v_local;
   IF v_lado='PDV' AND v_local IS NOT NULL AND EXISTS(SELECT 1 FROM t2_sync_fila WHERE entidade=v_ent AND chave=v_local AND estado='pendente') THEN
    CALL t2_sync_snapshot(v_ent,v_local,v_code);
    UPDATE t2_sync_fila SET estado='conflito',erro_codigo='SYNC_ERP_PREVALECE',conflito_payload=v_code WHERE entidade=v_ent AND chave=v_local;
   END IF;
  END IF;
  SET @sync_local=v_local;
  IF v_exist=1 THEN
   SELECT GROUP_CONCAT(CONCAT('`',IF(v_t='estoqueprodutos' AND j.c='qtd',v_stockcol,j.c),'`=IF(JSON_TYPE(JSON_EXTRACT(@sync_data,',QUOTE(CONCAT('$.',j.c)),
    '))=''NULL'',NULL,JSON_UNQUOTE(JSON_EXTRACT(@sync_data,',QUOTE(CONCAT('$.',j.c)),')))') ORDER BY j.n SEPARATOR ',')
    INTO v_expr FROM JSON_TABLE(v_cols,'$[*]' COLUMNS(n FOR ORDINALITY,c VARCHAR(64) PATH '$')) j WHERE j.c<>v_pk;
   SET @sync_sql=CONCAT('UPDATE `',v_t,'` SET ',v_expr,' WHERE `',v_pk,'`=?');
   PREPARE sync_stmt FROM @sync_sql; EXECUTE sync_stmt USING @sync_local; DEALLOCATE PREPARE sync_stmt;
  ELSE
   IF v_local IS NOT NULL THEN SET v_d=JSON_SET(v_d,CONCAT('$.',v_pk),v_local); END IF;
   SET @sync_data=CAST(v_d AS CHAR CHARACTER SET utf8mb4);
   SELECT GROUP_CONCAT(CONCAT('`',IF(v_t='estoqueprodutos' AND j.c='qtd',v_stockcol,j.c),'`') ORDER BY j.n SEPARATOR ','),
    GROUP_CONCAT(CONCAT('IF(JSON_TYPE(JSON_EXTRACT(@sync_data,',QUOTE(CONCAT('$.',j.c)),
    '))=''NULL'',NULL,JSON_UNQUOTE(JSON_EXTRACT(@sync_data,',QUOTE(CONCAT('$.',j.c)),')))') ORDER BY j.n SEPARATOR ',') INTO v_expr,v_values
    FROM JSON_TABLE(v_cols,'$[*]' COLUMNS(n FOR ORDINALITY,c VARCHAR(64) PATH '$')) j
    WHERE j.c<>v_pk OR v_auto=0 OR v_local IS NOT NULL;
   SET @sync_sql=CONCAT('INSERT INTO `',v_t,'`(',v_expr,') VALUES(',v_values,')');
   PREPARE sync_stmt FROM @sync_sql; EXECUTE sync_stmt; DEALLOCATE PREPARE sync_stmt;
   IF v_auto=1 AND v_local IS NULL THEN SET v_local=CAST(LAST_INSERT_ID() AS CHAR);
   ELSEIF v_local IS NULL THEN SET v_local=JSON_UNQUOTE(JSON_EXTRACT(v_d,CONCAT('$.',v_pk))); END IF;
  END IF;
  IF v_i=0 THEN SET v_rootlocal=v_local; END IF;
  IF v_o<>v_self THEN
   INSERT INTO t2_sync_mapa VALUES(v_o,v_t,v_key,v_local)
    ON DUPLICATE KEY UPDATE local_id=IF(local_id=VALUES(local_id),local_id,NULL);
  END IF;
  CALL t2_sync_hash_linha(v_t,v_local,v_hash);
  INSERT INTO t2_sync_membro(raiz_origem,entidade,raiz,tabela,origem,chave,local_id,hash_local,hash_recebido,ultima_origem)
   VALUES(v_root_o,v_ent,v_root_k,v_t,v_o,v_key,v_local,v_hash,v_sourcehash,p_origem)
   ON DUPLICATE KEY UPDATE local_id=v_local,hash_local=v_hash,hash_recebido=v_sourcehash,ultima_origem=p_origem,atualizado=UTC_TIMESTAMP(3);
  INSERT INTO t2_sync_recebidos VALUES(v_t,v_o,v_key);
  SET v_i=v_i+1;
 END WHILE;
 -- Exclusoes de filhos financeiros sao espelhadas somente se ja vinculados e inalterados.
 SET v_j=JSON_LENGTH(v_members)-1;
 WHILE v_j>0 DO
  SET v_t=JSON_UNQUOTE(JSON_EXTRACT(v_members,CONCAT('$[',v_j,'].table')));
  SET v_pk=(SELECT pk FROM t2_sync_tabelas WHERE nome=v_t);
  WHILE EXISTS(SELECT 1 FROM t2_sync_membro mm WHERE mm.raiz_origem=v_root_o AND mm.entidade=v_ent AND mm.raiz=v_root_k AND mm.tabela=v_t
   AND NOT EXISTS(SELECT 1 FROM t2_sync_recebidos rr WHERE rr.tabela=mm.tabela AND rr.origem=mm.origem AND rr.chave=mm.chave)) DO
   SELECT mm.origem,mm.chave,mm.local_id,mm.hash_local INTO v_o,v_key,v_local,v_anterior FROM t2_sync_membro mm
    WHERE mm.raiz_origem=v_root_o AND mm.entidade=v_ent AND mm.raiz=v_root_k AND mm.tabela=v_t
    AND NOT EXISTS(SELECT 1 FROM t2_sync_recebidos rr WHERE rr.tabela=mm.tabela AND rr.origem=mm.origem AND rr.chave=mm.chave) LIMIT 1;
   IF v_ent<>'venda' THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_EXCLUSAO_FILHO_REQUER_REVISAO'; END IF;
   CALL t2_sync_hash_linha(v_t,v_local,v_hash);
   IF v_hash<>v_anterior THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_ERP_PREVALECE'; END IF;
   SET @sync_k=v_local; SET @sync_sql=CONCAT('DELETE FROM `',v_t,'` WHERE `',v_pk,'`=?');
   PREPARE sync_stmt FROM @sync_sql; EXECUTE sync_stmt USING @sync_k; DEALLOCATE PREPARE sync_stmt;
   DELETE FROM t2_sync_membro WHERE raiz_origem=v_root_o AND entidade=v_ent AND raiz=v_root_k AND tabela=v_t AND origem=v_o AND chave=v_key;
  END WHILE;
  SET v_j=v_j-1;
 END WHILE;
 IF v_ent='venda' AND v_lado='ERP' THEN CALL t2_sync_estoque(v_root_o,v_root_k,CAST(v_rootlocal AS UNSIGNED)); END IF;
 INSERT INTO t2_sync_log(operacao,entidade,codigo) VALUES(p_op,v_ent,'CONFIRMADO');
 UPDATE t2_sync_revisao SET revisao=v_revisao,hash_payload=p_hash WHERE origem=p_origem AND entidade=v_ent AND chave=JSON_UNQUOTE(JSON_EXTRACT(p_doc,'$.chave'));
 SET @t2_sync_aplicando=0; COMMIT;
 SELECT p_op AS operacao,p_hash AS hash_payload,'confirmado' AS resultado;
END$$
DELIMITER ;