panels-origin/db/core/changesets/010-rpc-workers.sql

--liquibase formatted sql
-- PANELS · core · RPC trabajadores (personal)

--changeset panel:core-010a-worker-helpers splitStatements:false
CREATE OR REPLACE FUNCTION core._worker_trim_text(p_value text)
RETURNS text
LANGUAGE sql
IMMUTABLE
AS $$
  SELECT btrim(COALESCE(p_value, ''));
$$;

CREATE OR REPLACE FUNCTION core._worker_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._worker_format_nss(p_value text)
RETURNS text
LANGUAGE sql
IMMUTABLE
AS $$
  SELECT regexp_replace(COALESCE(p_value, ''), '\D', '', 'g');
$$;

CREATE OR REPLACE FUNCTION core._worker_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._worker_validate_curp(p_raw text)
RETURNS text
LANGUAGE plpgsql
IMMUTABLE
AS $$
DECLARE
  v_curp text := core._worker_norm_upper(p_raw);
BEGIN
  IF v_curp !~ '^[A-Z][AEIOUX][A-Z]{2}\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])[HMX](?:AS|BC|BS|CC|CS|CH|CL|CM|DF|DG|GT|GR|HG|JC|MC|MN|MS|NT|NL|OC|PL|QT|QR|SP|SL|SR|TC|TS|TL|VZ|YN|ZS|NE)[B-DF-HJ-NP-TV-Z]{3}[A-Z\d]\d$' THEN
    RETURN 'CURP inválida (18 caracteres, formato oficial)';
  END IF;
  RETURN NULL;
END;
$$;

CREATE OR REPLACE FUNCTION core._worker_validate_rfc(p_raw text)
RETURNS text
LANGUAGE plpgsql
IMMUTABLE
AS $$
DECLARE
  v_rfc text := core._worker_norm_upper(p_raw);
BEGIN
  IF v_rfc !~ '^[A-ZÑ&]{4}\d{6}[A-Z0-9]{3}$'
     AND v_rfc !~ '^[A-ZÑ&]{3}\d{6}[A-Z0-9]{3}$' THEN
    RETURN 'RFC inválido (12 o 13 caracteres, formato SAT)';
  END IF;
  RETURN NULL;
END;
$$;

CREATE OR REPLACE FUNCTION core._worker_validate_nss(p_raw text)
RETURNS text
LANGUAGE plpgsql
IMMUTABLE
AS $$
DECLARE
  v_nss text := core._worker_format_nss(p_raw);
  v_sum integer := 0;
  v_n integer;
  v_check integer;
  i integer;
BEGIN
  IF length(v_nss) <> 11 THEN
    RETURN 'NSS inválido (11 dígitos)';
  END IF;
  FOR i IN 1..10 LOOP
    v_n := substring(v_nss, i, 1)::integer * CASE WHEN (i - 1) % 2 = 1 THEN 2 ELSE 1 END;
    IF v_n > 9 THEN
      v_n := v_n - 9;
    END IF;
    v_sum := v_sum + v_n;
  END LOOP;
  v_check := (10 - (v_sum % 10)) % 10;
  IF v_check <> substring(v_nss, 11, 1)::integer THEN
    RETURN 'NSS inválido (dígito verificador)';
  END IF;
  RETURN NULL;
END;
$$;

CREATE OR REPLACE FUNCTION core._worker_validate_fields(p_payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
IMMUTABLE
AS $$
DECLARE
  v_errors jsonb := '{}'::jsonb;
  v_err text;
  v_wt text := upper(core._worker_trim_text(p_payload->>'work_type'));
BEGIN
  IF core._worker_trim_text(p_payload->>'first_name') = '' THEN
    v_errors := v_errors || jsonb_build_object('first_name', 'Nombre obligatorio');
  END IF;
  IF core._worker_trim_text(p_payload->>'last_name_p') = '' THEN
    v_errors := v_errors || jsonb_build_object('last_name_p', 'Apellido paterno obligatorio');
  END IF;
  IF core._worker_trim_text(p_payload->>'last_name_m') = '' THEN
    v_errors := v_errors || jsonb_build_object('last_name_m', 'Apellido materno obligatorio');
  END IF;
  v_err := core._worker_validate_curp(p_payload->>'curp');
  IF v_err IS NOT NULL THEN
    v_errors := v_errors || jsonb_build_object('curp', v_err);
  END IF;
  v_err := core._worker_validate_rfc(p_payload->>'rfc');
  IF v_err IS NOT NULL THEN
    v_errors := v_errors || jsonb_build_object('rfc', v_err);
  END IF;
  v_err := core._worker_validate_nss(p_payload->>'nss');
  IF v_err IS NOT NULL THEN
    v_errors := v_errors || jsonb_build_object('nss', v_err);
  END IF;
  IF core._worker_trim_text(p_payload->>'phone') = '' THEN
    v_errors := v_errors || jsonb_build_object('phone', 'Teléfono obligatorio');
  END IF;
  IF core._worker_trim_text(p_payload->>'email') = '' THEN
    v_errors := v_errors || jsonb_build_object('email', 'Correo obligatorio');
  END IF;
  IF core._worker_trim_text(p_payload->>'address') = '' THEN
    v_errors := v_errors || jsonb_build_object('address', 'Dirección obligatoria');
  END IF;
  IF core._worker_trim_text(p_payload->>'position') = '' THEN
    v_errors := v_errors || jsonb_build_object('position', 'Cargo obligatorio');
  END IF;
  IF core._worker_trim_text(p_payload->>'risk_code') = '' THEN
    v_errors := v_errors || jsonb_build_object('risk_code', 'Riesgo obligatorio');
  END IF;
  IF v_wt NOT IN ('N', 'D') THEN
    v_errors := v_errors || jsonb_build_object('work_type', 'Tipo de trabajo N o D');
  END IF;
  IF p_payload->>'daily_wage' IS NULL OR btrim(p_payload->>'daily_wage') = '' THEN
    v_errors := v_errors || jsonb_build_object('daily_wage', 'Jornal obligatorio');
  ELSIF (p_payload->>'daily_wage')::numeric IS NULL THEN
    v_errors := v_errors || jsonb_build_object('daily_wage', 'Jornal obligatorio');
  END IF;
  RETURN v_errors;
END;
$$;

CREATE OR REPLACE FUNCTION core._worker_normalize(p_payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
IMMUTABLE
AS $$
DECLARE
  v_middle text := core._worker_trim_text(p_payload->>'middle_name');
  v_blood text := core._worker_norm_upper(p_payload->>'blood_type');
  v_needs_badge boolean := NOT (
    p_payload->>'needs_badge' = 'false'
    OR p_payload->>'needs_badge' = '0'
    OR p_payload->'needs_badge' = 'false'::jsonb
    OR p_payload->'needs_badge' = '0'::jsonb
  );
BEGIN
  RETURN jsonb_build_object(
    'first_name', initcap(lower(core._worker_trim_text(p_payload->>'first_name'))),
    'middle_name', CASE WHEN v_middle = '' THEN NULL ELSE initcap(lower(v_middle)) END,
    'last_name_p', initcap(lower(core._worker_trim_text(p_payload->>'last_name_p'))),
    'last_name_m', initcap(lower(core._worker_trim_text(p_payload->>'last_name_m'))),
    'curp', core._worker_norm_upper(p_payload->>'curp'),
    'rfc', core._worker_norm_upper(p_payload->>'rfc'),
    'nss', core._worker_format_nss(p_payload->>'nss'),
    'phone', core._worker_trim_text(p_payload->>'phone'),
    'email', lower(core._worker_trim_text(p_payload->>'email')),
    'address', core._worker_trim_text(p_payload->>'address'),
    'blood_type', CASE WHEN v_blood = '' THEN NULL ELSE v_blood END,
    'hire_type', upper(core._worker_trim_text(p_payload->>'hire_type')),
    'position', initcap(lower(core._worker_trim_text(p_payload->>'position'))),
    'risk_code', core._worker_canonical_risk(p_payload->>'risk_code'),
    'work_type', upper(core._worker_trim_text(p_payload->>'work_type')),
    'daily_wage', COALESCE((p_payload->>'daily_wage')::numeric(12,2), 0),
    'needs_badge', v_needs_badge,
    'status', CASE WHEN p_payload->>'status' = 'baja' THEN 'baja' ELSE 'activo' END
  );
END;
$$;

CREATE OR REPLACE FUNCTION core._worker_full_name(
  p_first_name text,
  p_middle_name text,
  p_last_name_p text,
  p_last_name_m text
)
RETURNS text
LANGUAGE sql
IMMUTABLE
AS $$
  SELECT array_to_string(
    array_remove(ARRAY[p_first_name, p_middle_name, p_last_name_p, p_last_name_m], NULL),
  ' ');
$$;

CREATE OR REPLACE FUNCTION core._worker_tenant_timezone(p_tenant_id integer)
RETURNS text
LANGUAGE plpgsql
STABLE
SET search_path = core
AS $$
DECLARE
  v_tz text;
BEGIN
  IF p_tenant_id IS NULL THEN
    RETURN 'America/Mexico_City';
  END IF;
  SELECT timezone INTO v_tz
    FROM tenant_settings
   WHERE tenant_id = p_tenant_id;
  RETURN COALESCE(v_tz, 'America/Mexico_City');
END;
$$;

CREATE OR REPLACE FUNCTION core._worker_today(p_tenant_id integer)
RETURNS date
LANGUAGE sql
STABLE
SET search_path = core
AS $$
  SELECT (timezone(core._worker_tenant_timezone(p_tenant_id), now()))::date;
$$;

CREATE OR REPLACE FUNCTION core._worker_imss_context(p_worker_id bigint)
RETURNS jsonb
LANGUAGE plpgsql
STABLE
SET search_path = core
AS $$
DECLARE
  v_ctx jsonb;
BEGIN
  SELECT jsonb_build_object(
    'imss_status', COALESCE(w.imss_status, 'sin_alta'),
    'last_rehire_at', w.last_rehire_at,
    'current_alta_at', COALESCE(d.imss_alta_at, w.imss_alta_at, d.uploaded_at::date)
  )
    INTO v_ctx
    FROM workers w
    LEFT JOIN LATERAL (
      SELECT doc.imss_alta_at, doc.uploaded_at
        FROM documents doc
       WHERE doc.worker_id = w.id
         AND doc.type_code = 'alta_imss'
         AND doc.is_current = true
       LIMIT 1
    ) d ON true
   WHERE w.id = p_worker_id;
  RETURN COALESCE(v_ctx, jsonb_build_object(
    'imss_status', 'sin_alta',
    'last_rehire_at', NULL,
    'current_alta_at', NULL
  ));
END;
$$;

CREATE OR REPLACE FUNCTION core._worker_freshness_required(p_ctx jsonb)
RETURNS boolean
LANGUAGE plpgsql
IMMUTABLE
AS $$
DECLARE
  v_rehire date;
  v_alta date;
BEGIN
  IF COALESCE(p_ctx->>'imss_status', 'sin_alta') <> 'alta' THEN
    RETURN true;
  END IF;
  IF p_ctx->>'last_rehire_at' IS NULL OR btrim(p_ctx->>'last_rehire_at') = '' THEN
    RETURN false;
  END IF;
  v_rehire := (p_ctx->>'last_rehire_at')::date;
  IF p_ctx->>'current_alta_at' IS NULL OR btrim(p_ctx->>'current_alta_at') = '' THEN
    RETURN true;
  END IF;
  v_alta := (p_ctx->>'current_alta_at')::date;
  RETURN v_alta < v_rehire;
END;
$$;

CREATE OR REPLACE FUNCTION core._worker_eval_doc_validity(
  p_policy jsonb,
  p_doc jsonb,
  p_ctx jsonb,
  p_today date
)
RETURNS jsonb
LANGUAGE plpgsql
IMMUTABLE
AS $$
DECLARE
  v_mode text := COALESCE(p_policy->>'validity_mode', 'none');
  v_issued date;
  v_expires date;
  v_need_fresh boolean;
  v_days integer;
BEGIN
  IF p_doc IS NULL THEN
    RETURN jsonb_build_object(
      'present', false,
      'valid', false,
      'validity_status', 'missing',
      'issued_at', NULL,
      'expires_at', NULL
    );
  END IF;
  v_issued := NULLIF(p_doc->>'issued_at', '')::date;
  v_expires := NULLIF(p_doc->>'expires_at', '')::date;
  v_need_fresh := core._worker_freshness_required(p_ctx);
  IF v_mode = 'expiry' THEN
    IF COALESCE((p_policy->>'requires_expires_at')::boolean, false) AND v_expires IS NULL THEN
      RETURN jsonb_build_object(
        'present', true,
        'valid', false,
        'validity_status', 'missing_dates',
        'issued_at', v_issued,
        'expires_at', v_expires
      );
    END IF;
    IF v_expires IS NOT NULL AND v_expires < p_today THEN
      RETURN jsonb_build_object(
        'present', true,
        'valid', false,
        'validity_status', 'expired',
        'issued_at', v_issued,
        'expires_at', v_expires
      );
    END IF;
    RETURN jsonb_build_object(
      'present', true,
      'valid', true,
      'validity_status', 'ok',
      'issued_at', v_issued,
      'expires_at', v_expires
    );
  END IF;
  IF v_mode = 'freshness' THEN
    IF COALESCE((p_policy->>'requires_issued_at')::boolean, false) AND v_issued IS NULL THEN
      RETURN jsonb_build_object(
        'present', true,
        'valid', false,
        'validity_status', 'missing_dates',
        'issued_at', v_issued,
        'expires_at', v_expires
      );
    END IF;
    IF v_need_fresh AND v_issued IS NOT NULL THEN
      v_days := COALESCE((p_policy->>'freshness_days')::integer, 90);
      IF (p_today - v_issued) > v_days THEN
        RETURN jsonb_build_object(
          'present', true,
          'valid', false,
          'validity_status', 'stale_for_alta',
          'issued_at', v_issued,
          'expires_at', v_expires
        );
      END IF;
    END IF;
    RETURN jsonb_build_object(
      'present', true,
      'valid', true,
      'validity_status', 'ok',
      'issued_at', v_issued,
      'expires_at', v_expires
    );
  END IF;
  RETURN jsonb_build_object(
    'present', true,
    'valid', true,
    'validity_status', 'ok',
    'issued_at', v_issued,
    'expires_at', v_expires
  );
END;
$$;

CREATE OR REPLACE FUNCTION core._worker_checklist_items(
  p_worker_id bigint,
  p_tenant_id integer DEFAULT NULL
)
RETURNS jsonb
LANGUAGE plpgsql
STABLE
SET search_path = core
AS $$
DECLARE
  v_today date := core._worker_today(p_tenant_id);
  v_ctx jsonb := core._worker_imss_context(p_worker_id);
  v_items jsonb := '[]'::jsonb;
  v_freshness_required boolean := core._worker_freshness_required(v_ctx);
  v_rec record;
  v_doc jsonb;
  v_policy jsonb;
  v_evaled jsonb;
BEGIN
  FOR v_rec IN
    SELECT code, label, required, validity_mode, freshness_days,
           requires_issued_at, requires_expires_at, category
      FROM document_types
     ORDER BY required DESC, label
  LOOP
    SELECT jsonb_build_object(
      'type_code', d.type_code,
      'issued_at', d.issued_at,
      'expires_at', d.expires_at,
      'uploaded_at', d.uploaded_at,
      'imss_alta_at', d.imss_alta_at
    )
      INTO v_doc
      FROM documents d
     WHERE d.worker_id = p_worker_id
       AND d.type_code = v_rec.code
       AND d.is_current = true
     LIMIT 1;
    v_policy := jsonb_build_object(
      'code', v_rec.code,
      'label', v_rec.label,
      'required', v_rec.required,
      'validity_mode', COALESCE(v_rec.validity_mode, 'none'),
      'freshness_days', v_rec.freshness_days,
      'requires_issued_at', v_rec.requires_issued_at,
      'requires_expires_at', v_rec.requires_expires_at
    );
    v_evaled := core._worker_eval_doc_validity(v_policy, v_doc, v_ctx, v_today);
    v_items := v_items || jsonb_build_array(
      jsonb_build_object(
        'code', v_rec.code,
        'label', v_rec.label,
        'required', v_rec.required,
        'validity_mode', COALESCE(v_rec.validity_mode, 'none'),
        'freshness_days', v_rec.freshness_days,
        'requires_issued_at', v_rec.requires_issued_at,
        'requires_expires_at', v_rec.requires_expires_at,
        'category', COALESCE(v_rec.category, 'otro'),
        'present', COALESCE((v_evaled->>'present')::boolean, false),
        'valid', COALESCE((v_evaled->>'valid')::boolean, false),
        'validity_status', v_evaled->>'validity_status',
        'issued_at', v_evaled->>'issued_at',
        'expires_at', v_evaled->>'expires_at'
      )
    );
  END LOOP;
  RETURN jsonb_build_object(
    'ctx', v_ctx,
    'freshness_required', v_freshness_required,
    'items', v_items
  );
END;
$$;

CREATE OR REPLACE FUNCTION core._worker_imss_flags(
  p_worker_id bigint,
  p_tenant_id integer DEFAULT NULL
)
RETURNS jsonb
LANGUAGE plpgsql
STABLE
SET search_path = core
AS $$
DECLARE
  v_pack jsonb := core._worker_checklist_items(p_worker_id, p_tenant_id);
  v_items jsonb := v_pack->'items';
  v_ctx jsonb := v_pack->'ctx';
  v_required_ok boolean;
  v_in_project boolean;
  v_has_imss boolean;
BEGIN
  SELECT COALESCE(bool_and(
    COALESCE((elem->>'present')::boolean, false)
    AND COALESCE((elem->>'valid')::boolean, false)
  ), false)
    INTO v_required_ok
    FROM jsonb_array_elements(v_items) AS elem
   WHERE COALESCE((elem->>'required')::boolean, false);
  SELECT EXISTS (
    SELECT 1 FROM assignments a
     WHERE a.worker_id = p_worker_id AND a.active = true
     LIMIT 1
  ) INTO v_in_project;
  v_has_imss := COALESCE(v_ctx->>'imss_status', 'sin_alta') = 'alta';
  RETURN jsonb_build_object(
    'imss_status', COALESCE(v_ctx->>'imss_status', 'sin_alta'),
    'freshness_required', COALESCE((v_pack->>'freshness_required')::boolean, false),
    'imss_ready', v_required_ok AND NOT v_has_imss,
    'in_project_without_imss', v_in_project AND NOT v_has_imss,
    'expediente_ok', v_required_ok
  );
END;
$$;

CREATE OR REPLACE FUNCTION core._worker_conflict_message(
  p_row workers,
  p_curp text,
  p_rfc text,
  p_nss text
)
RETURNS text
LANGUAGE plpgsql
IMMUTABLE
AS $$
DECLARE
  v_field text;
BEGIN
  IF p_row.curp::text = p_curp THEN
    v_field := 'CURP';
  ELSIF p_row.rfc::text = p_rfc THEN
    v_field := 'RFC';
  ELSE
    v_field := 'NSS';
  END IF;
  RETURN format(
    'Este %s ya pertenece a %s %s',
    v_field, p_row.first_name, p_row.last_name_p
  );
END;
$$;

--changeset panel:core-010b-fn-worker-list splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_worker_list(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_tid integer := COALESCE(
    NULLIF(payload->>'tenant_id', '')::integer,
    NULLIF(current_setting('app.tenant_id', true), '')::integer
  );
  v_status text := nullif(btrim(payload->>'status'), '');
  v_project_id bigint := NULLIF(payload->>'project_id', '')::bigint;
  v_q text := nullif(btrim(payload->>'q'), '');
  v_like text;
  v_workers jsonb;
  v_count integer;
  v_imss_stats jsonb;
BEGIN
  IF v_q IS NOT NULL THEN
    v_like := '%' || v_q || '%';
  END IF;
  SELECT COALESCE(jsonb_agg(row_to_json(x)::jsonb ORDER BY x.last_name_p, x.first_name), '[]'::jsonb),
         COUNT(*)::integer
    INTO v_workers, v_count
    FROM (
      SELECT
        w.*,
        r.label AS risk_label,
        r.color AS risk_color,
        r.text_color AS risk_text,
        c.name AS company_name,
        c.kind AS company_kind,
        ic.name AS imss_company_name,
        ic.code AS imss_company_code,
        ic.registro_patronal AS imss_registro_patronal,
        (SELECT string_agg(p.name, ', ')
           FROM assignments a
           JOIN projects p ON p.id = a.project_id
          WHERE a.worker_id = w.id AND a.active = true) AS proyectos,
        (SELECT string_agg(p.code || '|' || replace(p.name, '|', '/'), ';;')
           FROM assignments a
           JOIN projects p ON p.id = a.project_id
          WHERE a.worker_id = w.id AND a.active = true) AS proyecto_pairs,
        (SELECT string_agg(p.id::text, ',')
           FROM assignments a
           JOIN projects p ON p.id = a.project_id
          WHERE a.worker_id = w.id AND a.active = true) AS project_ids,
        (SELECT COUNT(*)
           FROM document_types t
          WHERE t.required = true
            AND NOT EXISTS (
              SELECT 1 FROM documents d
               WHERE d.worker_id = w.id
                 AND d.type_code = t.code
                 AND d.is_current = true
            )) AS missing_docs,
        (SELECT COALESCE(SUM(l.balance), 0)
           FROM loans l
          WHERE l.worker_id = w.id AND l.balance > 0) AS loan_balance,
        EXISTS (
          SELECT 1 FROM assignments a
           WHERE a.worker_id = w.id AND a.active = true
        ) AS in_project,
        flags.data->>'imss_status' AS imss_status,
        (flags.data->>'imss_ready')::boolean AS imss_ready,
        (flags.data->>'in_project_without_imss')::boolean AS in_project_without_imss,
        (flags.data->>'expediente_ok')::boolean AS expediente_ok,
        (flags.data->>'freshness_required')::boolean AS freshness_required
      FROM workers w
      JOIN risk_levels r ON r.code = w.risk_code
      LEFT JOIN companies c ON c.id = w.company_id
      LEFT JOIN companies ic ON ic.id = w.imss_company_id
      LEFT JOIN LATERAL (
        SELECT core._worker_imss_flags(w.id, v_tid) AS data
      ) flags ON true
     WHERE (v_tid IS NULL OR w.tenant_id = v_tid)
       AND (v_status IS NULL OR w.status = v_status)
       AND (v_status IS DISTINCT FROM 'activo' OR w.pipeline_status <> 'baja')
       AND (v_project_id IS NULL OR EXISTS (
         SELECT 1 FROM assignments a
          WHERE a.worker_id = w.id
            AND a.project_id = v_project_id
            AND a.active = true
       ))
       AND (v_q IS NULL OR (
         w.first_name ILIKE v_like
         OR w.last_name_p ILIKE v_like
         OR w.curp::text ILIKE v_like
         OR w.rfc::text ILIKE v_like
         OR w.nss ILIKE v_like
       ))
    ) x;
  SELECT jsonb_build_object(
    'with_imss', COUNT(*) FILTER (
      WHERE elem->>'status' = 'activo'
        AND elem->>'pipeline_status' <> 'baja'
        AND elem->>'imss_status' = 'alta'
    ),
    'without_imss', COUNT(*) FILTER (
      WHERE elem->>'status' = 'activo'
        AND elem->>'pipeline_status' <> 'baja'
        AND elem->>'imss_status' IS DISTINCT FROM 'alta'
    ),
    'in_project_without_imss', COUNT(*) FILTER (
      WHERE elem->>'status' = 'activo'
        AND elem->>'pipeline_status' <> 'baja'
        AND COALESCE((elem->>'in_project_without_imss')::boolean, false)
    ),
    'imss_ready_count', COUNT(*) FILTER (
      WHERE elem->>'status' = 'activo'
        AND elem->>'pipeline_status' <> 'baja'
        AND COALESCE((elem->>'imss_ready')::boolean, false)
    )
  )
    INTO v_imss_stats
    FROM jsonb_array_elements(v_workers) AS elem;
  RETURN core.rpc_ok(
    jsonb_build_object(
      'workers', v_workers,
      'imss_stats', v_imss_stats
    ),
    format('Listado de trabajadores: %s registro(s)%s',
      v_count,
      CASE WHEN v_tid IS NOT NULL THEN format(' para tenant_id=%s', v_tid) ELSE '' END),
    jsonb_build_object('fn', 'fn_worker_list', 'tenant_id', v_tid, 'count', v_count)
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_worker_list', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-010c-fn-worker-get splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_worker_get(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_id bigint := NULLIF(payload->>'id', '')::bigint;
  v_tid integer := COALESCE(
    NULLIF(payload->>'tenant_id', '')::integer,
    NULLIF(current_setting('app.tenant_id', true), '')::integer
  );
  v_worker jsonb;
  v_documents jsonb;
  v_assignments jsonb;
  v_loans jsonb;
  v_checklist_pack jsonb;
  v_flags jsonb;
  v_doc_types jsonb;
BEGIN
  IF v_id IS NULL THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_get: id es obligatorio para consultar un trabajador',
      jsonb_build_object('fn', 'fn_worker_get', 'field', 'id')
    );
  END IF;
  SELECT row_to_json(x)::jsonb
    INTO v_worker
    FROM (
      SELECT
        w.*,
        r.label AS risk_label,
        r.color AS risk_color,
        r.text_color AS risk_text,
        c.name AS company_name,
        c.kind AS company_kind,
        ic.name AS imss_company_name,
        ic.code AS imss_company_code,
        ic.registro_patronal AS imss_registro_patronal,
        core._worker_full_name(w.first_name, w.middle_name, w.last_name_p, w.last_name_m) AS full_name
      FROM workers w
      JOIN risk_levels r ON r.code = w.risk_code
      LEFT JOIN companies c ON c.id = w.company_id
      LEFT JOIN companies ic ON ic.id = w.imss_company_id
     WHERE w.id = v_id
       AND (v_tid IS NULL OR w.tenant_id = v_tid)
    ) x;
  IF v_worker IS NULL THEN
    RETURN core.rpc_err(
      'NOT_FOUND',
      format('fn_worker_get: trabajador id=%s no encontrado%s', v_id,
        CASE WHEN v_tid IS NOT NULL THEN format(' en tenant_id=%s', v_tid) ELSE '' END),
      jsonb_build_object('fn', 'fn_worker_get', 'id', v_id, 'tenant_id', v_tid)
    );
  END IF;
  SELECT COALESCE(jsonb_agg(to_jsonb(d) ORDER BY d.uploaded_at DESC), '[]'::jsonb)
    INTO v_documents
    FROM (
      SELECT id, type_code, original_name, mime, size_bytes, is_current, uploaded_at,
             issued_at, expires_at, imss_company_id, imss_alta_at
        FROM documents
       WHERE worker_id = v_id
    ) d;
  SELECT COALESCE(jsonb_agg(row_to_json(a)::jsonb ORDER BY a.active DESC, a.start_date DESC, a.id DESC), '[]'::jsonb)
    INTO v_assignments
    FROM (
      SELECT asn.*, p.name AS project_name, p.code AS project_code
        FROM assignments asn
        JOIN projects p ON p.id = asn.project_id
       WHERE asn.worker_id = v_id
    ) a;
  SELECT COALESCE(jsonb_agg(to_jsonb(l) ORDER BY l.id DESC), '[]'::jsonb)
    INTO v_loans
    FROM loans l
   WHERE l.worker_id = v_id;
  v_checklist_pack := core._worker_checklist_items(v_id, v_tid);
  v_flags := core._worker_imss_flags(v_id, v_tid);
  v_worker := v_worker || v_flags || jsonb_build_object('full_name', v_worker->>'full_name');
  SELECT COALESCE(jsonb_agg(to_jsonb(t) ORDER BY t.required DESC, t.label), '[]'::jsonb)
    INTO v_doc_types
    FROM document_types t;
  RETURN core.rpc_ok(
    jsonb_build_object(
      'worker', v_worker,
      'documents', v_documents,
      'assignments', v_assignments,
      'loans', v_loans,
      'checklist', v_checklist_pack->'items',
      'document_types', v_doc_types,
      'freshness_required', v_checklist_pack->'freshness_required'
    ),
    format('Trabajador id=%s cargado (%s)', v_id, v_worker->>'full_name'),
    jsonb_build_object('fn', 'fn_worker_get', 'id', v_id)
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_worker_get', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-010d-fn-worker-create splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_worker_create(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_tid integer := COALESCE(
    NULLIF(payload->>'tenant_id', '')::integer,
    NULLIF(current_setting('app.tenant_id', true), '')::integer
  );
  v_project_id bigint := NULLIF(payload->>'project_id', '')::bigint;
  v_errors jsonb;
  v_norm jsonb;
  v_existing workers%ROWTYPE;
  v_row workers%ROWTYPE;
  v_project_status text;
  v_refresh jsonb;
BEGIN
  v_errors := core._worker_validate_fields(payload);
  IF v_errors <> '{}'::jsonb THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_create: datos del trabajador incompletos o inválidos',
      jsonb_build_object('fn', 'fn_worker_create', 'field', 'payload'),
      v_errors
    );
  END IF;
  v_norm := core._worker_normalize(payload);
  SELECT * INTO v_existing
    FROM workers
   WHERE curp = v_norm->>'curp'
      OR rfc = v_norm->>'rfc'
      OR nss = v_norm->>'nss'
   LIMIT 1;
  IF FOUND THEN
    RETURN core.rpc_err(
      'CONFLICT',
      core._worker_conflict_message(v_existing, v_norm->>'curp', v_norm->>'rfc', v_norm->>'nss'),
      jsonb_build_object('fn', 'fn_worker_create', 'worker_id', v_existing.id)
    );
  END IF;
  IF v_project_id IS NOT NULL THEN
    SELECT status INTO v_project_status FROM projects WHERE id = v_project_id;
    IF NOT FOUND THEN
      RETURN core.rpc_err(
        'NOT_FOUND',
        format('fn_worker_create: proyecto id=%s no encontrado', v_project_id),
        jsonb_build_object('fn', 'fn_worker_create', 'project_id', v_project_id)
      );
    END IF;
    IF v_project_status <> 'activo' THEN
      RETURN core.rpc_err(
        'VALIDATION',
        'fn_worker_create: solo se asigna personal a proyectos activos',
        jsonb_build_object('fn', 'fn_worker_create', 'project_id', v_project_id, 'status', v_project_status)
      );
    END IF;
  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 (
    v_norm->>'first_name',
    v_norm->>'middle_name',
    v_norm->>'last_name_p',
    v_norm->>'last_name_m',
    v_norm->>'curp',
    v_norm->>'rfc',
    v_norm->>'nss',
    v_norm->>'phone',
    v_norm->>'email',
    v_norm->>'address',
    v_norm->>'blood_type',
    '',
    NULL,
    v_norm->>'position',
    v_norm->>'risk_code',
    v_norm->>'work_type',
    (v_norm->>'daily_wage')::numeric(12,2),
    (v_norm->>'needs_badge')::boolean,
    v_norm->>'status',
    v_tid
  )
  RETURNING * INTO v_row;
  IF v_project_id IS NOT NULL THEN
    INSERT INTO assignments (worker_id, project_id, active, start_date)
    VALUES (v_row.id, v_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 IF;
  v_refresh := core.fn_worker_refresh_pipeline(jsonb_build_object(
    'worker_id', v_row.id,
    'tenant_id', v_tid
  ));
  IF NOT COALESCE((v_refresh->>'ok')::boolean, false) THEN
    RETURN v_refresh;
  END IF;
  SELECT pipeline_status INTO v_row.pipeline_status FROM workers WHERE id = v_row.id;
  RETURN core.rpc_created(
    jsonb_build_object('id', v_row.id, 'status', v_row.status, 'pipeline_status', v_row.pipeline_status),
    format('Trabajador %s %s registrado (id=%s)',
      v_row.first_name, v_row.last_name_p, v_row.id),
    jsonb_build_object('fn', 'fn_worker_create', 'id', v_row.id)
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_worker_create', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-010e-fn-worker-update splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_worker_update(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_id bigint := NULLIF(payload->>'id', '')::bigint;
  v_tid integer := COALESCE(
    NULLIF(payload->>'tenant_id', '')::integer,
    NULLIF(current_setting('app.tenant_id', true), '')::integer
  );
  v_cur workers%ROWTYPE;
  v_merged jsonb;
  v_errors jsonb;
  v_norm jsonb;
  v_existing workers%ROWTYPE;
  v_refresh jsonb;
BEGIN
  IF v_id IS NULL THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_update: id es obligatorio para actualizar un trabajador',
      jsonb_build_object('fn', 'fn_worker_update', 'field', 'id')
    );
  END IF;
  SELECT * INTO v_cur FROM workers WHERE id = v_id;
  IF NOT FOUND THEN
    RETURN core.rpc_err(
      'NOT_FOUND',
      format('fn_worker_update: trabajador id=%s no encontrado', v_id),
      jsonb_build_object('fn', 'fn_worker_update', 'id', v_id)
    );
  END IF;
  v_merged := to_jsonb(v_cur) || payload;
  v_errors := core._worker_validate_fields(v_merged);
  IF v_errors <> '{}'::jsonb THEN
    RETURN core.rpc_err(
      'VALIDATION',
      format('fn_worker_update: datos inválidos para trabajador id=%s', v_id),
      jsonb_build_object('fn', 'fn_worker_update', 'id', v_id),
      v_errors
    );
  END IF;
  v_norm := core._worker_normalize(v_merged);
  SELECT * INTO v_existing
    FROM workers
   WHERE (curp = v_norm->>'curp' OR rfc = v_norm->>'rfc' OR nss = v_norm->>'nss')
     AND id <> v_id
   LIMIT 1;
  IF FOUND THEN
    RETURN core.rpc_err(
      'CONFLICT',
      core._worker_conflict_message(v_existing, v_norm->>'curp', v_norm->>'rfc', v_norm->>'nss'),
      jsonb_build_object('fn', 'fn_worker_update', 'id', v_id, 'worker_id', v_existing.id)
    );
  END IF;
  UPDATE workers SET
    first_name = v_norm->>'first_name',
    middle_name = v_norm->>'middle_name',
    last_name_p = v_norm->>'last_name_p',
    last_name_m = v_norm->>'last_name_m',
    curp = v_norm->>'curp',
    rfc = v_norm->>'rfc',
    nss = v_norm->>'nss',
    phone = v_norm->>'phone',
    email = v_norm->>'email',
    address = v_norm->>'address',
    blood_type = v_norm->>'blood_type',
    hire_type = v_cur.hire_type,
    company_id = v_cur.company_id,
    position = v_norm->>'position',
    risk_code = v_norm->>'risk_code',
    work_type = v_norm->>'work_type',
    daily_wage = (v_norm->>'daily_wage')::numeric(12,2),
    needs_badge = (v_norm->>'needs_badge')::boolean,
    status = v_norm->>'status',
    updated_at = now()
  WHERE id = v_id;
  v_refresh := core.fn_worker_refresh_pipeline(jsonb_build_object(
    'worker_id', v_id,
    'tenant_id', v_tid
  ));
  IF NOT COALESCE((v_refresh->>'ok')::boolean, false) THEN
    RETURN v_refresh;
  END IF;
  RETURN core.rpc_ok(
    jsonb_build_object('id', v_id, 'status', v_norm->>'status'),
    format('Trabajador id=%s actualizado correctamente', v_id),
    jsonb_build_object('fn', 'fn_worker_update', 'id', v_id)
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_worker_update', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-010f-fn-worker-set-pipeline splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_worker_set_pipeline(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_id bigint := COALESCE(
    NULLIF(payload->>'id', '')::bigint,
    NULLIF(payload->>'worker_id', '')::bigint
  );
  v_tid integer := COALESCE(
    NULLIF(payload->>'tenant_id', '')::integer,
    NULLIF(current_setting('app.tenant_id', true), '')::integer
  );
  v_pipeline text := nullif(btrim(payload->>'pipeline_status'), '');
  v_prev workers%ROWTYPE;
  v_worker jsonb;
  v_refresh jsonb;
BEGIN
  IF v_id IS NULL THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_set_pipeline: id o worker_id es obligatorio',
      jsonb_build_object('fn', 'fn_worker_set_pipeline', 'field', 'id')
    );
  END IF;
  IF v_pipeline IS NULL OR v_pipeline NOT IN ('incompleto', 'listo_gafete', 'impreso', 'activo', 'baja') THEN
    RETURN core.rpc_err(
      'VALIDATION',
      format('fn_worker_set_pipeline: estado de pipeline inválido (%s)', COALESCE(v_pipeline, 'NULL')),
      jsonb_build_object('fn', 'fn_worker_set_pipeline', 'field', 'pipeline_status', 'value', v_pipeline)
    );
  END IF;
  SELECT * INTO v_prev FROM workers WHERE id = v_id;
  IF NOT FOUND THEN
    RETURN core.rpc_err(
      'NOT_FOUND',
      format('fn_worker_set_pipeline: trabajador id=%s no encontrado', v_id),
      jsonb_build_object('fn', 'fn_worker_set_pipeline', 'id', v_id)
    );
  END IF;
  IF v_pipeline = 'baja' THEN
    UPDATE workers
       SET status = 'baja',
           pipeline_status = 'baja',
           updated_at = now()
     WHERE id = v_id;
  ELSE
    IF v_prev.status = 'baja' THEN
      UPDATE workers
         SET status = 'activo',
             last_rehire_at = current_date,
             updated_at = now()
       WHERE id = v_id;
    ELSE
      UPDATE workers
         SET status = 'activo',
             updated_at = now()
       WHERE id = v_id;
    END IF;
    v_refresh := core.fn_worker_refresh_pipeline(jsonb_build_object(
      'worker_id', v_id,
      'tenant_id', v_tid
    ));
    IF NOT COALESCE((v_refresh->>'ok')::boolean, false) THEN
      RETURN v_refresh;
    END IF;
  END IF;
  SELECT to_jsonb(w) INTO v_worker
    FROM (
      SELECT status, pipeline_status, last_rehire_at, imss_status
        FROM workers
       WHERE id = v_id
    ) w;
  RETURN core.rpc_ok(
    jsonb_build_object('worker', v_worker),
    format('Pipeline del trabajador id=%s actualizado a %s', v_id, v_worker->>'pipeline_status'),
    jsonb_build_object('fn', 'fn_worker_set_pipeline', 'id', v_id, 'pipeline_status', v_worker->>'pipeline_status')
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_worker_set_pipeline', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-010g-fn-worker-assign splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_worker_assign(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_id bigint := COALESCE(
    NULLIF(payload->>'id', '')::bigint,
    NULLIF(payload->>'worker_id', '')::bigint
  );
  v_tid integer := COALESCE(
    NULLIF(payload->>'tenant_id', '')::integer,
    NULLIF(current_setting('app.tenant_id', true), '')::integer
  );
  v_project_id bigint := NULLIF(payload->>'project_id', '')::bigint;
  v_project_status text;
  v_refresh jsonb;
BEGIN
  IF v_id IS NULL THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_assign: id o worker_id es obligatorio',
      jsonb_build_object('fn', 'fn_worker_assign', 'field', 'worker_id')
    );
  END IF;
  IF v_project_id IS NULL THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_assign: project_id es obligatorio',
      jsonb_build_object('fn', 'fn_worker_assign', 'field', 'project_id')
    );
  END IF;
  IF NOT EXISTS (SELECT 1 FROM workers WHERE id = v_id) THEN
    RETURN core.rpc_err(
      'NOT_FOUND',
      format('fn_worker_assign: trabajador id=%s no encontrado', v_id),
      jsonb_build_object('fn', 'fn_worker_assign', 'worker_id', v_id)
    );
  END IF;
  SELECT status INTO v_project_status FROM projects WHERE id = v_project_id;
  IF NOT FOUND THEN
    RETURN core.rpc_err(
      'NOT_FOUND',
      format('fn_worker_assign: proyecto id=%s no encontrado', v_project_id),
      jsonb_build_object('fn', 'fn_worker_assign', 'project_id', v_project_id)
    );
  END IF;
  IF v_project_status <> 'activo' THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_assign: solo se asigna personal a proyectos activos',
      jsonb_build_object('fn', 'fn_worker_assign', 'project_id', v_project_id, 'status', v_project_status)
    );
  END IF;
  INSERT INTO assignments (worker_id, project_id, active, start_date)
  VALUES (v_id, v_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;
  v_refresh := core.fn_worker_refresh_pipeline(jsonb_build_object(
    'worker_id', v_id,
    'tenant_id', v_tid
  ));
  IF NOT COALESCE((v_refresh->>'ok')::boolean, false) THEN
    RETURN v_refresh;
  END IF;
  RETURN core.rpc_ok(
    jsonb_build_object('worker_id', v_id, 'project_id', v_project_id, 'active', true),
    format('Trabajador id=%s asignado al proyecto id=%s', v_id, v_project_id),
    jsonb_build_object('fn', 'fn_worker_assign', 'worker_id', v_id, 'project_id', v_project_id)
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_worker_assign', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-010h-fn-worker-unassign splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_worker_unassign(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_id bigint := COALESCE(
    NULLIF(payload->>'id', '')::bigint,
    NULLIF(payload->>'worker_id', '')::bigint
  );
  v_tid integer := COALESCE(
    NULLIF(payload->>'tenant_id', '')::integer,
    NULLIF(current_setting('app.tenant_id', true), '')::integer
  );
  v_project_id bigint := NULLIF(payload->>'project_id', '')::bigint;
  v_updated integer;
  v_refresh jsonb;
BEGIN
  IF v_id IS NULL THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_unassign: id o worker_id es obligatorio',
      jsonb_build_object('fn', 'fn_worker_unassign', 'field', 'worker_id')
    );
  END IF;
  IF v_project_id IS NULL THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_unassign: project_id es obligatorio',
      jsonb_build_object('fn', 'fn_worker_unassign', 'field', 'project_id')
    );
  END IF;
  UPDATE assignments
     SET active = false,
         end_date = current_date
   WHERE worker_id = v_id
     AND project_id = v_project_id
     AND active = true;
  GET DIAGNOSTICS v_updated = ROW_COUNT;
  IF v_updated = 0 THEN
    RETURN core.rpc_err(
      'NOT_FOUND',
      format('fn_worker_unassign: no hay asignación activa del trabajador id=%s al proyecto id=%s', v_id, v_project_id),
      jsonb_build_object('fn', 'fn_worker_unassign', 'worker_id', v_id, 'project_id', v_project_id)
    );
  END IF;
  v_refresh := core.fn_worker_refresh_pipeline(jsonb_build_object(
    'worker_id', v_id,
    'tenant_id', v_tid
  ));
  IF NOT COALESCE((v_refresh->>'ok')::boolean, false) THEN
    RETURN v_refresh;
  END IF;
  RETURN core.rpc_ok(
    jsonb_build_object('worker_id', v_id, 'project_id', v_project_id, 'active', false),
    format('Trabajador id=%s desasignado del proyecto id=%s', v_id, v_project_id),
    jsonb_build_object('fn', 'fn_worker_unassign', 'worker_id', v_id, 'project_id', v_project_id)
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_worker_unassign', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-010i-fn-worker-find-existing splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_worker_find_existing(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_curp text := core._worker_norm_upper(payload->>'curp');
  v_rfc text := core._worker_norm_upper(payload->>'rfc');
  v_nss text := core._worker_format_nss(payload->>'nss');
  v_exclude bigint := COALESCE(NULLIF(payload->>'exclude_id', '')::bigint, 0);
  v_worker jsonb;
BEGIN
  IF v_curp = '' AND v_rfc = '' AND v_nss = '' THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_find_existing: indique al menos curp, rfc o nss para buscar duplicados',
      jsonb_build_object('fn', 'fn_worker_find_existing', 'field', 'curp')
    );
  END IF;
  SELECT to_jsonb(w) INTO v_worker
    FROM (
      SELECT id, first_name, last_name_p, curp, rfc, nss
        FROM workers
       WHERE id <> v_exclude
         AND (
           (v_curp <> '' AND curp = v_curp)
           OR (v_rfc <> '' AND rfc = v_rfc)
           OR (v_nss <> '' AND nss = v_nss)
         )
       LIMIT 1
    ) w;
  IF v_worker IS NULL THEN
    RETURN core.rpc_ok(
      jsonb_build_object('worker', NULL, 'found', false),
      'No se encontró trabajador con la misma CURP, RFC o NSS',
      jsonb_build_object('fn', 'fn_worker_find_existing', 'found', false)
    );
  END IF;
  RETURN core.rpc_ok(
    jsonb_build_object('worker', v_worker, 'found', true),
    format('Trabajador existente encontrado: %s %s (id=%s)',
      v_worker->>'first_name', v_worker->>'last_name_p', v_worker->>'id'),
    jsonb_build_object('fn', 'fn_worker_find_existing', 'found', true, 'id', v_worker->>'id')
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_worker_find_existing', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-010j-fn-worker-refresh-pipeline splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_worker_refresh_pipeline(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_id bigint := COALESCE(
    NULLIF(payload->>'worker_id', '')::bigint,
    NULLIF(payload->>'id', '')::bigint
  );
  v_tid integer := COALESCE(
    NULLIF(payload->>'tenant_id', '')::integer,
    NULLIF(current_setting('app.tenant_id', true), '')::integer
  );
  v_status text;
  v_pack jsonb;
  v_items jsonb;
  v_photo jsonb;
  v_photo_ok boolean;
  v_required_ok boolean;
  v_printed boolean;
  v_assigned boolean;
  v_pipeline text := 'incompleto';
BEGIN
  IF v_id IS NULL THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_refresh_pipeline: worker_id es obligatorio',
      jsonb_build_object('fn', 'fn_worker_refresh_pipeline', 'field', 'worker_id')
    );
  END IF;
  SELECT status INTO v_status FROM workers WHERE id = v_id;
  IF NOT FOUND THEN
    RETURN core.rpc_err(
      'NOT_FOUND',
      format('fn_worker_refresh_pipeline: trabajador id=%s no encontrado', v_id),
      jsonb_build_object('fn', 'fn_worker_refresh_pipeline', 'worker_id', v_id)
    );
  END IF;
  IF v_status = 'baja' THEN
    UPDATE workers SET pipeline_status = 'baja', updated_at = now() WHERE id = v_id;
    RETURN core.rpc_ok(
      jsonb_build_object('worker_id', v_id, 'pipeline_status', 'baja'),
      format('Pipeline del trabajador id=%s fijado en baja', v_id),
      jsonb_build_object('fn', 'fn_worker_refresh_pipeline', 'worker_id', v_id, 'pipeline_status', 'baja')
    );
  END IF;
  v_pack := core._worker_checklist_items(v_id, v_tid);
  v_items := v_pack->'items';
  SELECT elem INTO v_photo
    FROM jsonb_array_elements(v_items) AS elem
   WHERE elem->>'code' = 'foto'
   LIMIT 1;
  v_photo_ok := COALESCE((v_photo->>'present')::boolean, false)
    AND COALESCE((v_photo->>'valid')::boolean, false);
  SELECT COALESCE(bool_and(
    COALESCE((elem->>'present')::boolean, false)
    AND COALESCE((elem->>'valid')::boolean, false)
  ), false)
    INTO v_required_ok
    FROM jsonb_array_elements(v_items) AS elem
   WHERE COALESCE((elem->>'required')::boolean, false);
  SELECT EXISTS (
    SELECT 1
      FROM badge_job_people p
      JOIN badge_jobs j ON j.id = p.job_id
     WHERE p.worker_id = v_id
     LIMIT 1
  ) INTO v_printed;
  IF v_required_ok AND v_photo_ok THEN
    v_pipeline := CASE WHEN v_printed THEN 'impreso' ELSE 'listo_gafete' END;
  END IF;
  SELECT EXISTS (
    SELECT 1 FROM assignments a
     WHERE a.worker_id = v_id AND a.active = true
     LIMIT 1
  ) INTO v_assigned;
  IF v_pipeline <> 'incompleto' AND v_assigned THEN
    v_pipeline := CASE WHEN v_printed THEN 'activo' ELSE 'listo_gafete' END;
  END IF;
  UPDATE workers
     SET pipeline_status = v_pipeline,
         updated_at = now()
   WHERE id = v_id;
  RETURN core.rpc_ok(
    jsonb_build_object('worker_id', v_id, 'pipeline_status', v_pipeline),
    format('Pipeline del trabajador id=%s recalculado: %s', v_id, v_pipeline),
    jsonb_build_object('fn', 'fn_worker_refresh_pipeline', 'worker_id', v_id, 'pipeline_status', v_pipeline)
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_worker_refresh_pipeline', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-010k-fn-worker-checklist-raw splitStatements:false
CREATE OR REPLACE FUNCTION core.fn_worker_checklist_raw(payload jsonb)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY INVOKER
SET search_path = core
AS $$
DECLARE
  v_id bigint := COALESCE(
    NULLIF(payload->>'worker_id', '')::bigint,
    NULLIF(payload->>'id', '')::bigint
  );
  v_tid integer := COALESCE(
    NULLIF(payload->>'tenant_id', '')::integer,
    NULLIF(current_setting('app.tenant_id', true), '')::integer
  );
  v_doc_types jsonb;
  v_current_docs jsonb;
  v_pack jsonb;
BEGIN
  IF v_id IS NULL THEN
    RETURN core.rpc_err(
      'VALIDATION',
      'fn_worker_checklist_raw: worker_id es obligatorio',
      jsonb_build_object('fn', 'fn_worker_checklist_raw', 'field', 'worker_id')
    );
  END IF;
  IF NOT EXISTS (SELECT 1 FROM workers WHERE id = v_id) THEN
    RETURN core.rpc_err(
      'NOT_FOUND',
      format('fn_worker_checklist_raw: trabajador id=%s no encontrado', v_id),
      jsonb_build_object('fn', 'fn_worker_checklist_raw', 'worker_id', v_id)
    );
  END IF;
  SELECT COALESCE(jsonb_agg(to_jsonb(t) ORDER BY t.required DESC, t.label), '[]'::jsonb)
    INTO v_doc_types
    FROM document_types t;
  SELECT COALESCE(jsonb_agg(to_jsonb(d)), '[]'::jsonb)
    INTO v_current_docs
    FROM (
      SELECT type_code, issued_at, expires_at, uploaded_at, imss_alta_at
        FROM documents
       WHERE worker_id = v_id AND is_current = true
    ) d;
  v_pack := core._worker_checklist_items(v_id, v_tid);
  RETURN core.rpc_ok(
    jsonb_build_object(
      'document_types', v_doc_types,
      'current_documents', v_current_docs,
      'imss_context', v_pack->'ctx',
      'freshness_required', v_pack->'freshness_required',
      'checklist', v_pack->'items'
    ),
    format('Checklist documental del trabajador id=%s: %s tipo(s)',
      v_id, jsonb_array_length(v_pack->'items')),
    jsonb_build_object('fn', 'fn_worker_checklist_raw', 'worker_id', v_id)
  );
EXCEPTION WHEN OTHERS THEN
  RETURN core.rpc_from_exception('fn_worker_checklist_raw', SQLSTATE, SQLERRM);
END;
$$;

--changeset panel:core-010l-fn-workers-grants endDelimiter:; splitStatements:true
GRANT EXECUTE ON FUNCTION core._worker_trim_text(text) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_norm_upper(text) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_format_nss(text) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_canonical_risk(text) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_validate_curp(text) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_validate_rfc(text) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_validate_nss(text) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_validate_fields(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_normalize(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_full_name(text, text, text, text) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_tenant_timezone(integer) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_today(integer) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_imss_context(bigint) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_freshness_required(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_eval_doc_validity(jsonb, jsonb, jsonb, date) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_checklist_items(bigint, integer) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_imss_flags(bigint, integer) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core._worker_conflict_message(workers, text, text, text) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core.fn_worker_list(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core.fn_worker_get(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core.fn_worker_create(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core.fn_worker_update(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core.fn_worker_set_pipeline(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core.fn_worker_assign(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core.fn_worker_unassign(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core.fn_worker_find_existing(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core.fn_worker_refresh_pipeline(jsonb) TO panels_core_app;
GRANT EXECUTE ON FUNCTION core.fn_worker_checklist_raw(jsonb) TO panels_core_app;