-- Recria somente rotinas tecnicas t2_sync_ com contexto de collation do banco.
-- Nao altera colunas, tabelas comerciais, filas, recibos ou configuracoes.
-- Executar com integrador parado; DDL MySQL confirma implicitamente.
SET SESSION collation_connection=@@collation_database;
DELIMITER $$
DROP PROCEDURE IF EXISTS t2_sync_marcar$$
CREATE PROCEDURE t2_sync_marcar(IN p_ent VARCHAR(64), IN p_chave VARCHAR(191))
proc: BEGIN
 DECLARE v_lado VARCHAR(8); DECLARE v_dono VARCHAR(8);
 IF COALESCE(@t2_sync_aplicando,0)=1 OR p_chave IS NULL OR p_chave='' THEN LEAVE proc; END IF;
 SET v_lado=(SELECT lado FROM t2_sync_config WHERE id=1);
 SET v_dono=(SELECT dono FROM t2_sync_entidades WHERE nome=p_ent);
 IF v_lado IS NULL OR (v_dono<>'BOTH' AND v_lado<>v_dono) THEN LEAVE proc; END IF;
 INSERT INTO t2_sync_fila(entidade,chave) VALUES(p_ent,p_chave)
 ON DUPLICATE KEY UPDATE versao=versao+1,
 estado=IF(estado='conflito','conflito','pendente'), atualizado=UTC_TIMESTAMP(3),
 proxima_tentativa=IF(operacao IS NULL,UTC_TIMESTAMP(3),proxima_tentativa);
END$$


DELIMITER ;

DELIMITER $$
DROP FUNCTION IF EXISTS t2_sync_estoque_coluna$$
CREATE FUNCTION t2_sync_estoque_coluna() RETURNS VARCHAR(64)
READS SQL DATA
BEGIN
 DECLARE c VARCHAR(64);
 SET c=CONCAT('qtd',(SELECT empresa FROM t2_sync_config WHERE id=1));
 IF EXISTS(SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='estoqueprodutos' AND column_name=c) THEN RETURN c; END IF;
 RETURN 'qtd';
END$$
DROP FUNCTION IF EXISTS t2_sync_ref$$
CREATE FUNCTION t2_sync_ref(p_t VARCHAR(64),p_id VARCHAR(191)) RETURNS JSON
READS SQL DATA
BEGIN
 DECLARE v_o VARCHAR(64); DECLARE v_k VARCHAR(191);
 IF p_id IS NULL OR p_id='' OR p_id='0' THEN RETURN NULL; END IF;
 IF p_t='movimentacao' AND (SELECT lado FROM t2_sync_config WHERE id=1)='PDV' AND
   EXISTS(SELECT 1 FROM movimentacao WHERE documento=p_id AND doc_interno>0) THEN
  RETURN JSON_OBJECT('origem',(SELECT par FROM t2_sync_config WHERE id=1),'tabela',p_t,
   'chave',CAST((SELECT doc_interno FROM movimentacao WHERE documento=p_id) AS CHAR));
 END IF;
 SELECT origem,chave INTO v_o,v_k FROM t2_sync_mapa WHERE tabela=p_t AND local_id=p_id LIMIT 1;
 IF v_o IS NULL THEN SET v_o=(SELECT origem FROM t2_sync_config WHERE id=1); SET v_k=p_id; END IF;
 RETURN JSON_OBJECT('origem',v_o,'tabela',p_t,'chave',v_k);
END$$
DROP PROCEDURE IF EXISTS t2_sync_resolver$$
CREATE PROCEDURE t2_sync_resolver(IN p_o VARCHAR(64),IN p_t VARCHAR(64),IN p_k VARCHAR(191),OUT p_local VARCHAR(191))
BEGIN
 DECLARE v_pk VARCHAR(64);
 SET p_local=NULL;
 IF p_k IS NULL OR p_k='' OR p_k='0' THEN SET p_local=p_k;
 ELSE
  IF p_o=(SELECT origem FROM t2_sync_config WHERE id=1) THEN
   SET v_pk=(SELECT pk FROM t2_sync_tabelas WHERE nome=p_t);
   IF v_pk IS NULL THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_TABELA_INVALIDA'; END IF;
   SET @sync_k=p_k; SET @sync_count=0;
   SET @sync_sql=CONCAT('SELECT COUNT(*) INTO @sync_count FROM `',p_t,'` WHERE `',v_pk,'`=?');
   PREPARE sync_stmt FROM @sync_sql; EXECUTE sync_stmt USING @sync_k; DEALLOCATE PREPARE sync_stmt;
   IF @sync_count=1 THEN SET p_local=p_k; END IF;
  ELSE
   SELECT local_id INTO p_local FROM t2_sync_mapa WHERE origem=p_o AND tabela=p_t AND chave=p_k LIMIT 1;
  END IF;
  IF p_local IS NULL THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_DEPENDENCIA_SEM_VINCULO'; END IF;
 END IF;
END$$
DROP PROCEDURE IF EXISTS t2_sync_snapshot$$
CREATE PROCEDURE t2_sync_snapshot(IN p_e VARCHAR(64),IN p_k VARCHAR(191),OUT p_doc LONGTEXT)
BEGIN
 DECLARE v_t,v_pk,v_root,v_rootkey,v_empcol VARCHAR(64);
 DECLARE v_members,v_cols,v_refs,v_rows,v_all JSON;
 DECLARE v_expr,v_refexpr,v_where LONGTEXT;
 DECLARE v_i INT DEFAULT 0; DECLARE v_empresa INT; DECLARE v_origem VARCHAR(64);
 SET v_all=JSON_ARRAY();
 SELECT tabela_raiz,chave_raiz,tabelas INTO v_root,v_rootkey,v_members FROM t2_sync_entidades WHERE nome=p_e;
 SELECT origem,empresa INTO v_origem,v_empresa FROM t2_sync_config WHERE id=1;
 IF v_root IS NULL OR v_origem IS NULL THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_CONFIGURACAO_AUSENTE'; END IF;
 SET v_empcol=(SELECT campo_empresa FROM t2_sync_tabelas WHERE nome=v_root);
 SET @sync_k=p_k; SET @sync_emp=v_empresa; SET @sync_count=0;
 SET @sync_sql=CONCAT('SELECT COUNT(*) INTO @sync_count FROM `',v_root,'` WHERE `',v_rootkey,'`=?',
  IF(v_empcol IS NULL,'',CONCAT(' AND `',v_empcol,'`=',v_empresa)));
 PREPARE sync_stmt FROM @sync_sql; EXECUTE sync_stmt USING @sync_k; DEALLOCATE PREPARE sync_stmt;
 IF @sync_count<>1 THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_EXCLUSAO_FISICA_OU_ESCOPO_INVALIDO'; END IF;
 IF p_e='venda' THEN
  IF EXISTS(SELECT 1 FROM movimentacao WHERE documento=p_k AND doc_interno>0) THEN
   SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_VENDA_IMPORTADA_ERP_REQUER_CONCILIACAO'; END IF;
  IF NOT EXISTS(SELECT 1 FROM movimentacao WHERE documento=p_k AND finalizado='S' AND empresa=v_empresa) THEN
   SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_VENDA_NAO_FINALIZADA'; END IF;
 END IF;
 IF p_e='nfce' THEN
  IF NOT EXISTS(SELECT 1 FROM nfe WHERE id=p_k AND modelo='65' AND status IN('100','150','101') AND LENGTH(xml)>0) THEN
   SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_AGUARDANDO_XML_AUTORIZADO'; END IF;
 END IF;
 SET SESSION group_concat_max_len=16777216;
 WHILE v_i<JSON_LENGTH(v_members) DO
  SET v_t=JSON_UNQUOTE(JSON_EXTRACT(v_members,CONCAT('$[',v_i,'].table')));
  SELECT pk,colunas,referencias INTO v_pk,v_cols,v_refs FROM t2_sync_tabelas WHERE nome=v_t;
  SELECT GROUP_CONCAT(CONCAT(QUOTE(j.c),',CONVERT(t.`',IF(v_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;
  SELECT GROUP_CONCAT(CONCAT(QUOTE(j.c),',t2_sync_ref(',QUOTE(JSON_UNQUOTE(JSON_EXTRACT(v_refs,CONCAT('$.',j.c)))),
   ',CAST(t.`',j.c,'` AS CHAR))') SEPARATOR ',') INTO v_refexpr
   FROM JSON_TABLE(JSON_KEYS(v_refs),'$[*]' COLUMNS(c VARCHAR(64) PATH '$')) j
   WHERE JSON_UNQUOTE(JSON_EXTRACT(v_refs,CONCAT('$.',j.c)))<>'@empresa';
  SET v_where=CONCAT('t.`',JSON_UNQUOTE(JSON_EXTRACT(v_members,CONCAT('$[',v_i,'].field'))),'`=?');
  SET v_empcol=(SELECT campo_empresa FROM t2_sync_tabelas WHERE nome=v_t);
  IF v_empcol IS NOT NULL THEN SET v_where=CONCAT(v_where,' AND t.`',v_empcol,'`=',v_empresa); END IF;
  SET @sync_rows=NULL;
  SET @sync_sql=CONCAT('SELECT CONCAT(''['',GROUP_CONCAT(JSON_OBJECT(''tabela'',',QUOTE(v_t),
   ',''origem'',COALESCE(mp.origem,',QUOTE(v_origem),'),''chave'',COALESCE(mp.chave,CAST(t.`',v_pk,'` AS CHAR)),',
   '''base_hash'',(SELECT mm.hash_recebido FROM t2_sync_membro mm WHERE mm.tabela=',QUOTE(v_t),' AND mm.local_id=CAST(t.`',v_pk,'` AS CHAR) AND mm.ultima_origem=(SELECT par FROM t2_sync_config WHERE id=1) ORDER BY mm.atualizado DESC LIMIT 1),''dados'',JSON_OBJECT(',v_expr,'),''refs'',JSON_OBJECT(',COALESCE(v_refexpr,''),')) ORDER BY t.`',v_pk,'` SEPARATOR '',''),'']'') INTO @sync_rows FROM `',v_t,
   '` t LEFT JOIN t2_sync_mapa mp ON mp.tabela=',QUOTE(v_t),' AND mp.local_id=CAST(t.`',v_pk,'` AS CHAR) WHERE ',v_where);
  PREPARE sync_stmt FROM @sync_sql; EXECUTE sync_stmt USING @sync_k; DEALLOCATE PREPARE sync_stmt;
  SET v_rows=COALESCE(CAST(@sync_rows AS JSON),JSON_ARRAY());
  SET v_all=JSON_MERGE_PRESERVE(v_all,v_rows); SET v_i=v_i+1;
 END WHILE;
 SET p_doc=CAST(JSON_OBJECT('versao',1,'origem',v_origem,'empresa',v_empresa,'entidade',p_e,'chave',p_k,'linhas',v_all) AS CHAR CHARACTER SET utf8mb4);
 IF OCTET_LENGTH(p_doc)>8388608 OR JSON_LENGTH(v_all)>10000 THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_LOTE_EXCESSIVO'; END IF;
END$$
DROP PROCEDURE IF EXISTS t2_sync_preparar$$
CREATE PROCEDURE t2_sync_preparar(IN p_e VARCHAR(64),IN p_k VARCHAR(191))
BEGIN
 DECLARE v_op CHAR(36); DECLARE v_doc LONGTEXT; DECLARE v_hash CHAR(64); DECLARE v_ver BIGINT;
 DECLARE EXIT HANDLER FOR SQLEXCEPTION BEGIN ROLLBACK; RESIGNAL; END;
 START TRANSACTION;
 SELECT operacao,payload,hash_payload,versao INTO v_op,v_doc,v_hash,v_ver FROM t2_sync_fila
  WHERE entidade=p_e AND chave=p_k AND estado='pendente' FOR UPDATE;
 IF v_ver IS NULL THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_PENDENCIA_INEXISTENTE'; END IF;
 IF v_op IS NULL THEN
  CALL t2_sync_snapshot(p_e,p_k,v_doc);
  SET v_doc=CAST(JSON_SET(CAST(v_doc AS JSON),'$.revisao',v_ver) AS CHAR CHARACTER SET utf8mb4); SET v_op=UUID(); SET v_hash=SHA2(v_doc,256);
  UPDATE t2_sync_fila SET operacao=v_op,payload=v_doc,hash_payload=v_hash,versao_envio=v_ver WHERE entidade=p_e AND chave=p_k;
 END IF;
 COMMIT;
 SELECT v_op AS operacao,v_doc AS payload,v_hash AS hash_payload;
END$$
DROP PROCEDURE IF EXISTS t2_sync_confirmar$$
CREATE PROCEDURE t2_sync_confirmar(IN p_op CHAR(36),IN p_hash CHAR(64))
BEGIN
 UPDATE t2_sync_fila SET estado=IF(versao=versao_envio,'concluido','pendente'),
  operacao=NULL,payload=NULL,hash_payload=NULL,versao_envio=NULL,tentativas=0,
  proxima_tentativa=UTC_TIMESTAMP(3),erro_codigo=NULL
 WHERE operacao=p_op AND hash_payload=p_hash AND estado='pendente';
 IF ROW_COUNT()<>1 THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_CONFIRMACAO_INVALIDA'; END IF;
END$$
DROP PROCEDURE IF EXISTS t2_sync_falhar$$
CREATE PROCEDURE t2_sync_falhar(IN p_e VARCHAR(64),IN p_k VARCHAR(191),IN p_codigo VARCHAR(120),IN p_permanente BOOLEAN)
BEGIN
 UPDATE t2_sync_fila SET tentativas=tentativas+1,erro_codigo=p_codigo,
  estado=IF(p_permanente,'conflito','pendente'),
  proxima_tentativa=TIMESTAMPADD(SECOND,LEAST(300,POW(2,LEAST(tentativas+1,8))),UTC_TIMESTAMP(3))
 WHERE entidade=p_e AND chave=p_k;
 -- Uma rejeicao explicita por dependencia/estoque pendente significa rollback no destino.
 -- So nesses casos e seguro recapturar referencias/saldo; falha de rede mantem UUID e payload.
 IF p_codigo IN('SYNC_DEPENDENCIA_SEM_VINCULO','SYNC_ESTOQUE_COM_VENDAS_LOCAIS_PENDENTES') THEN
  UPDATE t2_sync_fila SET conflito_payload=COALESCE(payload,conflito_payload),versao=versao+1,
   operacao=NULL,payload=NULL,hash_payload=NULL,versao_envio=NULL WHERE entidade=p_e AND chave=p_k;
 END IF;
 INSERT INTO t2_sync_log(entidade,codigo) VALUES(p_e,p_codigo);
END$$
DELIMITER ;

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 ;

DELIMITER $$
DROP PROCEDURE IF EXISTS t2_sync_confirmar$$
CREATE PROCEDURE t2_sync_confirmar(IN p_op CHAR(36),IN p_hash CHAR(64))
proc: BEGIN
 DECLARE EXIT HANDLER FOR SQLEXCEPTION BEGIN ROLLBACK; RESIGNAL; END;
 START TRANSACTION;
 IF EXISTS(SELECT 1 FROM t2_sync_ack WHERE operacao=p_op AND hash_payload=p_hash) THEN COMMIT; LEAVE proc; END IF;
 UPDATE t2_sync_fila SET estado=IF(versao=versao_envio,'concluido','pendente'),
  operacao=NULL,payload=NULL,hash_payload=NULL,versao_envio=NULL,tentativas=0,
  proxima_tentativa=UTC_TIMESTAMP(3),erro_codigo=NULL
 WHERE operacao=p_op AND hash_payload=p_hash AND estado='pendente';
 IF ROW_COUNT()<>1 THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='SYNC_CONFIRMACAO_INVALIDA'; END IF;
 INSERT INTO t2_sync_ack(operacao,hash_payload) VALUES(p_op,p_hash);
 COMMIT;
END$$
DELIMITER ;