panels-origin/db/core/changesets/014-rpc-import.sql

--liquibase formatted sql
-- PANELS · core · RPC importación de personal y metadatos documentales

--changeset panel:core-014a-import-helpers splitStatements:false
CREATE OR REPLACE FUNCTION core._import_tenant_id(payload jsonb)
RETURNS integer
LANGUAGE sql
STABLE
AS $$
  SELECT COALESCE(
    NULLIF(payload->>'tenant_id', '')::integer,
    NULLIF(current_setting('app.tenant_id', true), '')::integer,
    0
  );
$$;

CREATE OR REPLACE FUNCTION core._import_norm_upper(p_value text)
RETURNS text
LANGUAGE sql
IMMUTABLE
AS $$
  SELECT upper(regexp_replace(btrim(COALESCE(p_value, '')), '\s+', '', 'g'));
$$;

CREATE OR REPLACE FUNCTION core._import_norm_name(p_value text)
RETURNS text
LANGUAGE sql
IMMUTABLE
AS $$
  SELECT regexp_replace(btrim(COALESCE(p_value, '')), '\s+', ' ', 'g');
$$;

CREATE OR REPLACE FUNCTION core._import_format_nss(p_value text)
RETURNS text
LANGUAGE sql
IMMUTABLE
AS $$
  SELECT regexp_replace(COALESCE(p_value, ''), '\D', '', 'g');
$$;

CREATE OR REPLACE FUNCTION core._import_canonical_risk(p_value text)
RETURNS text
LANGUAGE sql
IMMUTABLE
AS $$
  SELECT CASE lower(btrim(COALESCE(p_value, '')))
    WHEN 'alto' THEN 'rojo'
    WHEN 'medio' THEN 'amarillo'
    WHEN 'bajo' THEN 'verde'
    ELSE lower(btrim(COALESCE(p_value, '')))
  END;
$$;

CREATE OR REPLACE FUNCTION core._import_validate_worker_row(p_row jsonb)
RETURNS text[]
LANGUAGE plpgsql
IMMUTABLE
AS $$
DECLARE
  v_errors text[] := ARRAY[]::text[];
  v_email text := lower(btrim(COALESCE(p_row->>'email', '')));
  v_wt text := upper(btrim(COALESCE(p_row->>'work_type', '')));
  v_nss text := core._import_format_nss(p_row->>'nss');
BEGIN
  IF core._import_norm_name(p_row->>'first_name') = '' THEN
    v_errors := array_append(v_errors, 'Nombre obligatorio');
  END IF;
  IF core._import_norm_name(p_row->>'last_name_p') = '' THEN
    v_errors := array_append(v_errors, 'Apellido paterno obligatorio');
  END IF;
  IF core._import_norm_name(p_row->>'last_name_m') = '' THEN
    v_errors := array_append(v_errors, 'Apellido materno obligatorio');
  END IF;
  IF length(core._import_norm_upper(p_row->>'curp')) <> 18 THEN
    v_errors := array_append(v_errors, 'CURP inválida (18 caracteres, formato oficial)');
  END IF;
  IF length(core._import_norm_upper(p_row->>'rfc')) NOT IN (12, 13) THEN
    v_errors := array_append(v_errors, 'RFC inválido (12 o 13 caracteres, formato SAT)');
  END IF;
  IF length(v_nss) <> 11 THEN
    v_errors := array_append(v_errors, 'NSS inválido (11 dígitos)');
  END IF;
  IF core._import_norm_name(p_row->>'phone') = '' THEN
    v_errors := array_append(v_errors, 'Teléfono obligatorio');
  END IF;
  IF v_email = '' THEN
    v_errors := array_append(v_errors, 'Correo obligatorio');
  ELSIF v_email !~ '^[^\s@]+@[^\s@]+\.[^\s@]+$' THEN
    v_errors := array_append(v_errors, 'Correo inválido');
  END IF;
  IF core._import_norm_name(p_row->>'address') = '' THEN
    v_errors := array_append(v_errors, 'Dirección obligatoria');
  END IF;
  IF core._import_norm_name(p_row->>'position') = '' THEN
    v_errors := array_append(v_errors, 'Cargo obligatorio');
  END IF;
  IF core._import_canonical_risk(p_row->>'risk_code') = '' THEN
    v_errors := array_append(v_errors, 'Riesgo obligatorio');
  END IF;
  IF v_wt NOT IN ('N', 'D') THEN
    v_errors := array_append(v_errors, 'Tipo de trabajo N o D');
  END IF;
  IF p_row->>'daily_wage' IS NULL OR (p_row->>'daily_wage')::numeric IS NULL THEN
    v_errors := array_append(v_errors, 'Jornal obligatorio');
  END IF;
  RETURN v_errors;
END;
$$;

CREATE OR REPLACE FUNCTION core._import_find_worker(p_curp text, p_rfc text, p_nss text)
RETURNS TABLE(id bigint, curp text, rfc text, nss text)
LANGUAGE sql
STABLE
SET search_path = core
AS $$
  SELECT w.id, w.curp::text, w.rfc::text, w.nss
    FROM workers w
   WHERE w.curp = p_curp OR w.rfc = p_rfc OR w.nss = p_nss
   LIMIT 1;
$$;

CREATE OR REPLACE FUNCTION core._import_assign_worker(p_worker_id bigint, p_project_id bigint)
RETURNS void
LANGUAGE plpgsql
SET search_path = core
AS $$
BEGIN
  IF p_project_id IS NULL THEN
    RETURN;
  END IF;
  INSERT INTO assignments (worker_id, project_id, active, start_date)
  VALUES (p_worker_id, p_project_id, true, current_date)
  ON CONFLICT (worker_id, project_id) DO UPDATE SET
    active = true,
    start_date = CASE
      WHEN assignments.active = false THEN excluded.start_date
      ELSE assignments.start_date
    END,
    end_date = NULL;
END;
$$;

CREATE OR REPLACE FUNCTION core._import_upsert_worker(
  p_row jsonb,
  p_company_id bigint,
  p_tenant_id integer
)
RETURNS TABLE(worker_id bigint, action text, matched text)
LANGUAGE plpgsql
SET search_path = core
AS $$
DECLARE
  v_curp text := core._import_norm_upper(p_row->>'curp');
  v_rfc text := core._import_norm_upper(p_row->>'rfc');
  v_nss text := core._import_format_nss(p_row->>'nss');
  v_existing record;
  v_status text := CASE WHEN lower(btrim(COALESCE(p_row->>'status', 'activo'))) = 'baja' THEN 'baja' ELSE 'activo' END;
  v_needs_badge boolean := CASE
    WHEN lower(COALESCE(p_row->>'needs_badge', 'true')) IN ('false', '0', 'no') THEN false
    ELSE true
  END;
  v_id bigint;
BEGIN
  SELECT * INTO v_existing FROM core._import_find_worker(v_curp, v_rfc, v_nss);
  IF FOUND THEN
    UPDATE workers SET
      first_name = initcap(lower(core._import_norm_name(p_row->>'first_name'))),
      middle_name = NULLIF(initcap(lower(core._import_norm_name(p_row->>'middle_name'))), ''),
      last_name_p = initcap(lower(core._import_norm_name(p_row->>'last_name_p'))),
      last_name_m = initcap(lower(core._import_norm_name(p_row->>'last_name_m'))),
      curp = v_curp,
      rfc = v_rfc,
      nss = v_nss,
      phone = core._import_norm_name(p_row->>'phone'),
      email = lower(btrim(COALESCE(p_row->>'email', ''))),
      address = core._import_norm_name(p_row->>'address'),
      blood_type = NULLIF(core._import_norm_upper(p_row->>'blood_type'), ''),
      hire_type = upper(btrim(COALESCE(p_row->>'hire_type', ''))),
      company_id = p_company_id,
      tenant_id = COALESCE(p_tenant_id, tenant_id),
      position = initcap(lower(core._import_norm_name(p_row->>'position'))),
      risk_code = core._import_canonical_risk(p_row->>'risk_code'),
      work_type = upper(btrim(COALESCE(p_row->>'work_type', ''))),
      daily_wage = COALESCE((p_row->>'daily_wage')::numeric, 0),
      needs_badge = v_needs_badge,
      status = v_status,
      updated_at = now()
    WHERE id = v_existing.id;
    worker_id := v_existing.id;
    action := 'existed';
    matched := CASE
      WHEN v_existing.curp = v_curp THEN 'CURP'
      WHEN v_existing.rfc = v_rfc THEN 'RFC'
      ELSE 'NSS'
    END;
    RETURN NEXT;
    RETURN;
  END IF;
  INSERT INTO workers (
    first_name, middle_name, last_name_p, last_name_m, curp, rfc, nss,
    phone, email, address, blood_type, hire_type, company_id, position,
    risk_code, work_type, daily_wage, needs_badge, status, tenant_id
  ) VALUES (
    initcap(lower(core._import_norm_name(p_row->>'first_name'))),
    NULLIF(initcap(lower(core._import_norm_name(p_row->>'middle_name'))), ''),
    initcap(lower(core._import_norm_name(p_row->>'last_name_p'))),
    initcap(lower(core._import_norm_name(p_row->>'last_name_m'))),
    v_curp, v_rfc, v_nss,
    core._import_norm_name(p_row->>'phone'),
    lower(btrim(COALESCE(p_row->>'email', ''))),
    core._import_norm_name(p_row->>'address'),
    NULLIF(core._import_norm_upper(p_row->>'blood_type'), ''),
    upper(btrim(COALESCE(p_row->>'hire_type', ''))),
    p_company_id,
    initcap(lower(core._import_norm_name(p_row->>'position'))),
    core._import_canonical_risk(p_row->>'risk_code'),
    upper(btrim(COALESCE(p_row->>'work_type', ''))),
    COALESCE((p_row->>'daily_wage')::numeric, 0),
    v_needs_badge,
    v_status,
    p_tenant_id
  )
  RETURNING id INTO v_id;
  worker_id := v_id;
  action := 'inserted';
  matched := NULL;
  RETURN NEXT;
END;
$$;

--changeset panel:core-014b-fn-workers-import-batch splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_workers_import_batch(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_tid integer := core._import_tenant_id(payload);
  v_default_project_id bigint := NULLIF(payload->>'project_id', '')::bigint;
  v_rows jsonb := COALESCE(payload->'rows', '[]'::jsonb);
  v_row jsonb;
  v_idx integer := 0;
  v_sheet text;
  v_row_no integer;
  v_nombre text;
  v_errors text[];
  v_company record;
  v_company_code text;
  v_company_id bigint;
  v_project_id bigint;
  v_status text;
  v_curp text;
  v_rfc text;
  v_nss text;
  v_seen_curp jsonb := '{}'::jsonb;
  v_seen_rfc jsonb := '{}'::jsonb;
  v_seen_nss jsonb := '{}'::jsonb;
  v_dup_row integer;
  v_upsert record;
  v_inserted integer := 0;
  v_existed integer := 0;
  v_error_count integer := 0;
  v_error_rows jsonb := '[]'::jsonb;
  v_existed_rows jsonb := '[]'::jsonb;
BEGIN
  IF jsonb_typeof(v_rows) <> 'array' THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_workers_import_batch: rows debe ser un arreglo JSON',
      jsonb_build_object('fn', 'fn_workers_import_batch', 'field', 'rows')
    );
  END IF;
  IF jsonb_array_length(v_rows) = 0 THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_workers_import_batch: el arreglo rows está vacío',
      jsonb_build_object('fn', 'fn_workers_import_batch', 'field', 'rows')
    );
  END IF;

  FOR v_row IN SELECT value FROM jsonb_array_elements(v_rows) LOOP
    v_idx := v_idx + 1;
    v_sheet := COALESCE(NULLIF(btrim(v_row->>'sheet'), ''), 'IMPORT');
    v_row_no := COALESCE((v_row->>'row')::integer, v_idx);
    v_nombre := trim(concat_ws(' ', v_row->>'first_name', v_row->>'last_name_p', v_row->>'last_name_m'));
    v_errors := core._import_validate_worker_row(v_row);

    v_company_code := upper(btrim(COALESCE(v_row->>'hire_type', v_row->>'company_code', '')));
    v_company_id := NULLIF(v_row->>'company_id', '')::bigint;
    IF v_company_id IS NOT NULL THEN
      SELECT id, code, tenant_id INTO v_company FROM companies WHERE id = v_company_id;
    ELSIF v_company_code <> '' THEN
      SELECT id, code, tenant_id INTO v_company FROM companies WHERE upper(code) = v_company_code;
    ELSE
      v_company := NULL;
    END IF;
    IF v_company.id IS NULL THEN
      v_errors := array_append(
        v_errors,
        format('Empresa no encontrada (%s)', COALESCE(NULLIF(v_company_code, ''), '—'))
      );
    ELSE
      v_company_id := v_company.id;
      IF v_tid = 0 AND v_company.tenant_id IS NOT NULL THEN
        v_tid := v_company.tenant_id;
      END IF;
    END IF;

    IF NOT EXISTS (
      SELECT 1 FROM risk_levels WHERE code = core._import_canonical_risk(v_row->>'risk_code')
    ) THEN
      v_errors := array_append(
        v_errors,
        format('Riesgo inválido (%s). Use alto, medio o bajo', COALESCE(v_row->>'risk_code', '—'))
      );
    END IF;

    v_curp := core._import_norm_upper(v_row->>'curp');
    v_rfc := core._import_norm_upper(v_row->>'rfc');
    v_nss := core._import_format_nss(v_row->>'nss');

    IF cardinality(v_errors) = 0 THEN
      IF v_seen_curp ? v_curp THEN
        v_dup_row := (v_seen_curp->>v_curp)::integer;
        v_errors := array_append(v_errors, format('CURP duplicada en el lote (fila %s)', v_dup_row));
      END IF;
      IF v_seen_rfc ? v_rfc THEN
        v_dup_row := (v_seen_rfc->>v_rfc)::integer;
        v_errors := array_append(v_errors, format('RFC duplicado en el lote (fila %s)', v_dup_row));
      END IF;
      IF v_seen_nss ? v_nss THEN
        v_dup_row := (v_seen_nss->>v_nss)::integer;
        v_errors := array_append(v_errors, format('NSS duplicado en el lote (fila %s)', v_dup_row));
      END IF;
    END IF;

    IF cardinality(v_errors) > 0 THEN
      v_error_count := v_error_count + 1;
      v_error_rows := v_error_rows || jsonb_build_array(jsonb_build_object(
        'sheet', v_sheet,
        'row', v_row_no,
        'nombre', v_nombre,
        'curp', COALESCE(v_row->>'curp', ''),
        'messages', to_jsonb(v_errors)
      ));
      CONTINUE;
    END IF;

    v_seen_curp := v_seen_curp || jsonb_build_object(v_curp, v_row_no);
    v_seen_rfc := v_seen_rfc || jsonb_build_object(v_rfc, v_row_no);
    v_seen_nss := v_seen_nss || jsonb_build_object(v_nss, v_row_no);

    v_status := CASE WHEN lower(btrim(COALESCE(v_row->>'status', 'activo'))) = 'baja' THEN 'baja' ELSE 'activo' END;
    v_project_id := COALESCE(NULLIF(v_row->>'project_id', '')::bigint, v_default_project_id);
    IF v_status <> 'activo' THEN
      v_project_id := NULL;
    END IF;

    SELECT * INTO v_upsert
      FROM core._import_upsert_worker(
        v_row || jsonb_build_object('hire_type', v_company.code),
        v_company_id,
        COALESCE(v_tid, v_company.tenant_id, 1)
      );

    IF v_project_id IS NOT NULL THEN
      PERFORM core._import_assign_worker(v_upsert.worker_id, v_project_id);
    END IF;

    IF v_status = 'baja' THEN
      UPDATE workers SET pipeline_status = 'baja' WHERE id = v_upsert.worker_id;
    END IF;

    IF v_upsert.action = 'inserted' THEN
      v_inserted := v_inserted + 1;
    ELSE
      v_existed := v_existed + 1;
      v_existed_rows := v_existed_rows || jsonb_build_array(jsonb_build_object(
        'sheet', v_sheet,
        'row', v_row_no,
        'nombre', v_nombre,
        'curp', v_curp,
        'matched', COALESCE(v_upsert.matched, 'CURP')
      ));
    END IF;
  END LOOP;

  RETURN core.rpc_ok(
    jsonb_build_object(
      'inserted', v_inserted,
      'existed', v_existed,
      'errors', v_error_count,
      'existed_rows', v_existed_rows,
      'error_rows', v_error_rows
    ),
    format('Importación por lote: %s nuevo(s), %s existente(s), %s error(es)',
      v_inserted, v_existed, v_error_count),
    jsonb_build_object(
      'fn', 'fn_workers_import_batch',
      'tenant_id', v_tid,
      'inserted', v_inserted,
      'existed', v_existed,
      'errors', v_error_count
    )
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_workers_import_batch', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-014c-fn-worker-document-store splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_worker_document_store(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_worker_id bigint := NULLIF(payload->>'worker_id', '')::bigint;
  v_type_code text := btrim(COALESCE(payload->>'type_code', payload->>'type', ''));
  v_original_name text := btrim(COALESCE(payload->>'original_name', payload->>'filename', ''));
  v_mime text := btrim(COALESCE(payload->>'mime', ''));
  v_size_bytes bigint := NULLIF(payload->>'size_bytes', '')::bigint;
  v_sha256 text := btrim(COALESCE(payload->>'sha256', ''));
  v_iv text := btrim(COALESCE(payload->>'iv', ''));
  v_storage_name text := btrim(COALESCE(payload->>'storage_name', ''));
  v_uploaded_by_id integer := NULLIF(payload->>'uploaded_by_id', '')::integer;
  v_uploaded_by_name text := COALESCE(NULLIF(btrim(payload->>'uploaded_by_name'), ''), '');
  v_issued_at date := NULLIF(btrim(payload->>'issued_at'), '')::date;
  v_expires_at date := NULLIF(btrim(payload->>'expires_at'), '')::date;
  v_imss_company_id integer := NULLIF(payload->>'imss_company_id', '')::integer;
  v_imss_alta_at date := NULLIF(btrim(payload->>'imss_alta_at'), '')::date;
  v_imss_baja_at date := NULLIF(btrim(payload->>'imss_baja_at'), '')::date;
  v_today date := current_date;
  v_movement_date date;
  v_doc_id bigint;
  v_company record;
BEGIN
  IF v_worker_id IS NULL THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_document_store: worker_id es obligatorio',
      jsonb_build_object('fn', 'fn_worker_document_store', 'field', 'worker_id')
    );
  END IF;
  IF NOT EXISTS (SELECT 1 FROM workers WHERE id = v_worker_id) THEN
    RETURN core.rpc_err(
      'NOT_FOUND',
      format('fn_worker_document_store: trabajador id=%s no encontrado', v_worker_id),
      jsonb_build_object('fn', 'fn_worker_document_store', 'worker_id', v_worker_id)
    );
  END IF;
  IF v_type_code = '' THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_document_store: type_code es obligatorio',
      jsonb_build_object('fn', 'fn_worker_document_store', 'field', 'type_code')
    );
  END IF;
  IF NOT EXISTS (SELECT 1 FROM document_types WHERE code = v_type_code) THEN
    RETURN core.rpc_err(
      'VALIDATION',
      format('fn_worker_document_store: tipo de documento no válido (%s)', v_type_code),
      jsonb_build_object('fn', 'fn_worker_document_store', 'type_code', v_type_code)
    );
  END IF;
  IF v_original_name = '' OR v_mime = '' OR v_sha256 = '' OR v_iv = '' OR v_storage_name = '' THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_document_store: original_name, mime, sha256, iv y storage_name son obligatorios tras la carga a almacenamiento',
      jsonb_build_object('fn', 'fn_worker_document_store')
    );
  END IF;
  IF v_size_bytes IS NULL OR v_size_bytes < 0 THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_document_store: size_bytes debe ser un entero no negativo',
      jsonb_build_object('fn', 'fn_worker_document_store', 'field', 'size_bytes')
    );
  END IF;

  v_imss_alta_at := COALESCE(v_imss_alta_at, CASE WHEN v_type_code = 'alta_imss' THEN v_today END);
  v_imss_baja_at := COALESCE(v_imss_baja_at, CASE WHEN v_type_code = 'baja_imss' THEN v_today END);
  v_movement_date := CASE WHEN v_type_code = 'baja_imss' THEN v_imss_baja_at ELSE v_imss_alta_at END;

  UPDATE documents
     SET is_current = false
   WHERE worker_id = v_worker_id
     AND type_code = v_type_code
     AND is_current = true;

  INSERT INTO documents (
    worker_id, type_code, original_name, mime, size_bytes, sha256, iv, storage_name,
    is_current, parse_status, issued_at, expires_at, imss_company_id, imss_alta_at,
    uploaded_by_id, uploaded_by_name
  ) VALUES (
    v_worker_id, v_type_code, v_original_name, v_mime, v_size_bytes, v_sha256, v_iv, v_storage_name,
    true, 'manual',
    COALESCE(v_issued_at, v_movement_date),
    v_expires_at,
    v_imss_company_id,
    v_movement_date,
    v_uploaded_by_id,
    v_uploaded_by_name
  )
  RETURNING id INTO v_doc_id;

  IF v_type_code = 'alta_imss' THEN
    v_imss_company_id := COALESCE(
      v_imss_company_id,
      (SELECT company_id FROM workers WHERE id = v_worker_id)
    );
    SELECT id, code INTO v_company FROM companies WHERE id = v_imss_company_id;
    IF NOT FOUND THEN
      RETURN core.rpc_err(
        'VALIDATION',
        'fn_worker_document_store: empresa patrón no válida para alta IMSS',
        jsonb_build_object('fn', 'fn_worker_document_store', 'worker_id', v_worker_id)
      );
    END IF;
    UPDATE workers SET
      imss_status = 'alta',
      imss_company_id = v_company.id,
      imss_alta_at = v_imss_alta_at,
      imss_baja_at = NULL,
      company_id = v_company.id,
      hire_type = v_company.code,
      updated_at = now()
    WHERE id = v_worker_id;
  END IF;

  IF v_type_code = 'baja_imss' THEN
    UPDATE workers SET
      imss_status = 'baja_imss',
      imss_baja_at = v_imss_baja_at,
      imss_company_id = NULL,
      company_id = NULL,
      hire_type = '',
      updated_at = now()
    WHERE id = v_worker_id;
  END IF;

  IF (SELECT status FROM workers WHERE id = v_worker_id) = 'baja' THEN
    UPDATE workers SET pipeline_status = 'baja' WHERE id = v_worker_id;
  END IF;

  RETURN core.rpc_created(
    jsonb_build_object(
      'id', v_doc_id,
      'worker_id', v_worker_id,
      'type_code', v_type_code,
      'is_current', true,
      'storage_name', v_storage_name
    ),
    format('Metadatos del documento %s guardados para trabajador %s (versión actual)', v_type_code, v_worker_id),
    jsonb_build_object('fn', 'fn_worker_document_store', 'id', v_doc_id, 'worker_id', v_worker_id)
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_worker_document_store', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-014d-fn-import-grants endDelimiter:; splitStatements:true
GRANT EXECUTE ON FUNCTION core.fn_workers_import_batch(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core.fn_worker_document_store(jsonb) TO panels_core_app;