-- 0003: import layer — batch tracking, loosely-typed staging, aliases,
-- ADP payroll code map. Staging is loose ON PURPOSE (CLAUDE.md §12):
-- rejecting garbage is validation's job, not the loader's.

CREATE TABLE import_batches (
  company_id uuid NOT NULL REFERENCES companies (id),
  id bigint GENERATED ALWAYS AS IDENTITY,
  source text NOT NULL,                  -- 'quickbooks_journal' | 'adp_gl' | ...
  filename text NOT NULL,
  file_sha256 text NOT NULL,
  -- deliberate vocabulary: batches use 'validated' (a completed pipeline
  -- step); staging rows below use 'valid' (a per-row verdict). Different
  -- words on purpose, so a query against the wrong table fails loudly
  -- instead of silently matching the other table's status string.
  status text NOT NULL DEFAULT 'loaded'
    CHECK (status IN ('loaded', 'resolved', 'validated', 'posted',
                      'rolled_back', 'failed')),
  created_by uuid NOT NULL REFERENCES users (id),
  created_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (id),
  UNIQUE (company_id, id),
  -- §8: re-importing an identical file is rejected
  UNIQUE (company_id, file_sha256)
);

CREATE TABLE staging_journal (
  company_id uuid NOT NULL REFERENCES companies (id),
  id bigint GENERATED ALWAYS AS IDENTITY,
  batch_id bigint NOT NULL,
  entry_key text,          -- QuickBooks "Trans #": groups rows into one entry
  txn_type text,
  txn_date text,           -- loose: text until validation parses it
  doc_num text,
  name text,
  memo text,
  account_raw text,
  debit_raw text,
  credit_raw text,
  raw jsonb NOT NULL,      -- §12: raw source row kept forever
  resolved_account_id bigint,
  parsed_date date,
  parsed_debit numeric(19,4),
  parsed_credit numeric(19,4),
  posted_entry_id bigint,
  -- deliberate vocabulary: staging rows use 'valid' (a per-row verdict);
  -- import_batches above uses 'validated' (a completed pipeline step).
  -- Different words on purpose — see the note on import_batches.status.
  -- 'void': a well-formed zero-amount line (voided txn or capped payroll
  -- tax); kept forever, never posted.
  status text NOT NULL DEFAULT 'loaded'
    CHECK (status IN ('loaded', 'resolved', 'valid', 'error', 'posted', 'void')),
  error text,
  PRIMARY KEY (id),
  UNIQUE (company_id, id),
  FOREIGN KEY (company_id, batch_id) REFERENCES import_batches (company_id, id),
  FOREIGN KEY (company_id, resolved_account_id) REFERENCES accounts (company_id, id),
  FOREIGN KEY (company_id, posted_entry_id) REFERENCES journal_entries (company_id, id),
  -- a row cannot claim to be posted without the ledger link that proves it
  CHECK (status <> 'posted' OR posted_entry_id IS NOT NULL),
  -- a row cannot claim to be valid/posted without the resolution that got it there
  CHECK (status NOT IN ('valid', 'posted')
         OR (resolved_account_id IS NOT NULL AND parsed_date IS NOT NULL))
);
CREATE INDEX staging_journal_by_batch
  ON staging_journal (company_id, batch_id, entry_key);

CREATE TABLE account_aliases (
  company_id uuid NOT NULL REFERENCES companies (id),
  id bigint GENERATED ALWAYS AS IDENTITY,
  alias text NOT NULL,
  account_id bigint NOT NULL,
  PRIMARY KEY (id),
  UNIQUE (company_id, id),
  UNIQUE (company_id, alias),
  FOREIGN KEY (company_id, account_id) REFERENCES accounts (company_id, id)
);

-- §8: every ADP earnings/tax/deduction code maps to a GL account and a
-- normal side. Unmapped codes FAIL validation loudly — never a suspense sweep.
CREATE TABLE payroll_code_map (
  company_id uuid NOT NULL REFERENCES companies (id),
  id bigint GENERATED ALWAYS AS IDENTITY,
  code text NOT NULL,
  description text,
  account_id bigint NOT NULL,
  normal_side char(1) NOT NULL CHECK (normal_side IN ('D', 'C')),
  PRIMARY KEY (id),
  UNIQUE (company_id, id),
  UNIQUE (company_id, code),
  FOREIGN KEY (company_id, account_id) REFERENCES accounts (company_id, id)
);

DO $$
DECLARE t text;
BEGIN
  FOREACH t IN ARRAY ARRAY[
    'import_batches', 'staging_journal', 'account_aliases', 'payroll_code_map'
  ] LOOP
    EXECUTE format('ALTER TABLE %I ENABLE ROW LEVEL SECURITY', t);
    EXECUTE format('ALTER TABLE %I FORCE ROW LEVEL SECURITY', t);
    EXECUTE format(
      'CREATE POLICY %I_isolation ON %I USING (company_id = current_company_id())',
      t, t);
  END LOOP;
END $$;

-- column-restricted grants: file_sha256, filename, raw, and the loose
-- source columns are immutable for app — the raw row is kept forever,
-- per CLAUDE.md §12. Only the pipeline's working columns are updatable.
-- Batch/staging state advances ONLY through the pipeline functions
-- (SECURITY DEFINER); app writes rows in, but cannot move the state
-- machine by hand.
GRANT SELECT, INSERT ON import_batches, staging_journal TO app;
GRANT SELECT, INSERT, UPDATE, DELETE ON account_aliases, payroll_code_map TO app;
GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO app;

-- import history is part of the audit trail; no bulk wipes, owner included
CREATE TRIGGER import_batches_no_truncate
  BEFORE TRUNCATE ON import_batches
  FOR EACH STATEMENT EXECUTE FUNCTION forbid_truncate();
CREATE TRIGGER staging_journal_no_truncate
  BEFORE TRUNCATE ON staging_journal
  FOR EACH STATEMENT EXECUTE FUNCTION forbid_truncate();

-- Resolve: map account_raw -> accounts (by exact name or code) or alias.
CREATE FUNCTION resolve_batch(p_company_id uuid, p_batch_id bigint) RETURNS int
LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$
DECLARE
  v_unresolved int;
  v_status text;
BEGIN
  IF current_company_id() IS DISTINCT FROM p_company_id THEN
    RAISE EXCEPTION 'session company does not match batch company';
  END IF;

  SELECT status INTO v_status FROM import_batches
   WHERE company_id = p_company_id AND id = p_batch_id
   FOR UPDATE;
  IF NOT FOUND THEN
    RAISE EXCEPTION 'batch % not found', p_batch_id;
  END IF;
  IF v_status IN ('posted', 'rolled_back') THEN
    RAISE EXCEPTION 'batch % is already % — load a new batch instead',
      p_batch_id, v_status;
  END IF;

  UPDATE staging_journal s
     SET resolved_account_id = coalesce(
           (SELECT min(a.id) FROM accounts a
             WHERE a.company_id = s.company_id
               AND (a.name = s.account_raw OR a.code = s.account_raw)
            HAVING count(*) = 1),
           (SELECT al.account_id FROM account_aliases al
             WHERE al.company_id = s.company_id AND al.alias = s.account_raw)),
         status = 'resolved',
         error = NULL
   WHERE s.company_id = p_company_id AND s.batch_id = p_batch_id
     AND s.status IN ('loaded', 'resolved', 'error');

  SELECT count(*) INTO v_unresolved FROM staging_journal
   WHERE company_id = p_company_id AND batch_id = p_batch_id
     AND resolved_account_id IS NULL;

  UPDATE import_batches SET status = 'resolved'
   WHERE company_id = p_company_id AND id = p_batch_id;
  RETURN v_unresolved;
END $$;

-- Validate: parse the loose text, then mark every defect loudly.
CREATE FUNCTION validate_batch(p_company_id uuid, p_batch_id bigint) RETURNS int
LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$
DECLARE
  v_errors int;
  v_lock date;
  v_status text;
BEGIN
  IF current_company_id() IS DISTINCT FROM p_company_id THEN
    RAISE EXCEPTION 'session company does not match batch company';
  END IF;

  SELECT status INTO v_status FROM import_batches
   WHERE company_id = p_company_id AND id = p_batch_id
   FOR UPDATE;
  IF NOT FOUND THEN
    RAISE EXCEPTION 'batch % not found', p_batch_id;
  END IF;
  IF v_status IN ('posted', 'rolled_back') THEN
    RAISE EXCEPTION 'batch % is already % — load a new batch instead',
      p_batch_id, v_status;
  END IF;

  v_lock := effective_locked_through(p_company_id);

  UPDATE staging_journal s SET
    parsed_date = CASE
      WHEN btrim(s.txn_date) ~ '^\d{4}-\d{2}-\d{2}$' THEN btrim(s.txn_date)::date
      WHEN btrim(s.txn_date) ~ '^\d{1,2}/\d{1,2}/\d{4}$'
        THEN to_date(btrim(s.txn_date), 'FMMM/FMDD/YYYY')
      ELSE NULL END,
    parsed_debit = CASE
      WHEN nullif(replace(btrim(s.debit_raw), ',', ''), '') ~ '^\d+(\.\d+)?$'
      THEN replace(btrim(s.debit_raw), ',', '')::numeric(19,4) ELSE NULL END,
    parsed_credit = CASE
      WHEN nullif(replace(btrim(s.credit_raw), ',', ''), '') ~ '^\d+(\.\d+)?$'
      THEN replace(btrim(s.credit_raw), ',', '')::numeric(19,4) ELSE NULL END
   WHERE s.company_id = p_company_id AND s.batch_id = p_batch_id;

  -- per-row defects
  UPDATE staging_journal s SET status = 'error', error = CASE
      WHEN s.resolved_account_id IS NULL
        THEN 'unmapped account (or ambiguous name): ' || coalesce(s.account_raw, '(blank)')
      WHEN s.entry_key IS NULL OR s.entry_key = '' THEN 'missing entry key'
      WHEN s.parsed_date IS NULL
        THEN 'unparseable date: ' || coalesce(s.txn_date, '(blank)')
      WHEN s.parsed_debit IS NULL AND s.parsed_credit IS NULL THEN
        CASE WHEN nullif(btrim(coalesce(s.debit_raw, '')), '') IS NULL
              AND nullif(btrim(coalesce(s.credit_raw, '')), '') IS NULL
             THEN 'no amount'
             ELSE 'unparseable amount: '
                  || coalesce(nullif(btrim(s.debit_raw), ''), btrim(s.credit_raw))
        END
      WHEN coalesce(s.parsed_debit, 0) > 0 AND coalesce(s.parsed_credit, 0) > 0
        THEN 'both debit and credit set'
      WHEN v_lock IS NOT NULL AND s.parsed_date <= v_lock
        THEN 'date falls in a locked period (locked through ' || v_lock || ')'
    END
   WHERE s.company_id = p_company_id AND s.batch_id = p_batch_id
     AND (s.resolved_account_id IS NULL
       OR s.entry_key IS NULL OR s.entry_key = ''
       OR s.parsed_date IS NULL
       OR (s.parsed_debit IS NULL AND s.parsed_credit IS NULL)
       OR (coalesce(s.parsed_debit, 0) > 0 AND coalesce(s.parsed_credit, 0) > 0)
       OR (v_lock IS NOT NULL AND s.parsed_date <= v_lock));

  -- well-formed zero-amount lines (voided txn or capped payroll tax): kept
  -- forever, never posted. Defect marking already ran, so unmapped/bad-date
  -- rows are already 'error' and keep precedence.
  UPDATE staging_journal s SET status = 'void', error = NULL
   WHERE s.company_id = p_company_id AND s.batch_id = p_batch_id
     AND s.status <> 'error'
     AND coalesce(s.parsed_debit, 0) = 0 AND coalesce(s.parsed_credit, 0) = 0
     AND (nullif(btrim(coalesce(s.debit_raw, '')), '') IS NOT NULL
       OR nullif(btrim(coalesce(s.credit_raw, '')), '') IS NOT NULL);

  -- unbalanced entry groups (void rows excluded: a fully-void entry_key has
  -- no rows left in the balance group and is correctly not flagged)
  UPDATE staging_journal s SET status = 'error',
         error = 'entry ' || s.entry_key || ' does not balance'
   WHERE s.company_id = p_company_id AND s.batch_id = p_batch_id
     AND s.status NOT IN ('error', 'void')
     AND s.entry_key IN (
       SELECT entry_key FROM staging_journal
        WHERE company_id = p_company_id AND batch_id = p_batch_id
          AND status NOT IN ('error', 'void')
        GROUP BY entry_key
        HAVING coalesce(sum(parsed_debit), 0) <> coalesce(sum(parsed_credit), 0)
            OR count(*) < 2);

  UPDATE staging_journal s SET status = 'valid', error = NULL
   WHERE s.company_id = p_company_id AND s.batch_id = p_batch_id
     AND s.status NOT IN ('error', 'void');

  SELECT count(*) INTO v_errors FROM staging_journal
   WHERE company_id = p_company_id AND batch_id = p_batch_id AND status = 'error';

  UPDATE import_batches
     SET status = CASE WHEN v_errors = 0 THEN 'validated' ELSE 'failed' END
   WHERE company_id = p_company_id AND id = p_batch_id;
  RETURN v_errors;
END $$;

-- Post: one journal entry per entry_key; external_ref = entry_key so the
-- (company_id, source, external_ref) unique index makes re-posting impossible.
CREATE FUNCTION post_batch(p_company_id uuid, p_batch_id bigint, p_user_id uuid)
RETURNS int
LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$
DECLARE
  v_batch import_batches%ROWTYPE;
  rec record;
  v_entry_id bigint;
  v_count int := 0;
BEGIN
  IF current_company_id() IS DISTINCT FROM p_company_id THEN
    RAISE EXCEPTION 'session company does not match batch company';
  END IF;

  SELECT * INTO v_batch FROM import_batches
   WHERE company_id = p_company_id AND id = p_batch_id FOR UPDATE;
  IF NOT FOUND THEN
    RAISE EXCEPTION 'batch % not found', p_batch_id;
  END IF;
  IF v_batch.status <> 'validated' THEN
    RAISE EXCEPTION 'batch % is not validated (status: %)', p_batch_id, v_batch.status;
  END IF;

  FOR rec IN
    -- min(memo) is a deliberate simplification: QuickBooks journal rows in
    -- one Trans # share the memo in practice.
    -- Only 'valid' rows post. A fully-void entry_key has no 'valid' rows and
    -- never appears in this cursor — zero financial effect, nothing to post.
    SELECT entry_key,
           min(parsed_date) AS entry_date,
           min(memo) AS memo,
           min(id) AS first_row
      FROM staging_journal
     WHERE company_id = p_company_id AND batch_id = p_batch_id
       AND status = 'valid'
     GROUP BY entry_key
     ORDER BY first_row
  LOOP
    INSERT INTO journal_entries
      (company_id, entry_date, memo, source, external_ref, created_by)
    VALUES
      (p_company_id, rec.entry_date, rec.memo, v_batch.source, rec.entry_key, p_user_id)
    RETURNING id INTO v_entry_id;

    INSERT INTO journal_lines
      (company_id, entry_id, account_id, debit, credit, memo)
    SELECT company_id, v_entry_id, resolved_account_id,
           coalesce(parsed_debit, 0), coalesce(parsed_credit, 0), memo
      FROM staging_journal
     WHERE company_id = p_company_id AND batch_id = p_batch_id
       AND entry_key = rec.entry_key
       AND status = 'valid';

    PERFORM post_entry(p_company_id, v_entry_id);

    -- void rows in this entry_key (e.g. a capped payroll tax line) keep
    -- status 'void' and NULL posted_entry_id — they are never posted.
    UPDATE staging_journal
       SET status = 'posted', posted_entry_id = v_entry_id
     WHERE company_id = p_company_id AND batch_id = p_batch_id
       AND entry_key = rec.entry_key
       AND status = 'valid';

    v_count := v_count + 1;
  END LOOP;

  UPDATE import_batches SET status = 'posted'
   WHERE company_id = p_company_id AND id = p_batch_id;
  RETURN v_count;
END $$;

-- §8: a bad batch is undone by REVERSING, never deleting.
CREATE FUNCTION rollback_import_batch(
  p_company_id uuid, p_batch_id bigint, p_user_id uuid,
  p_reversal_date date DEFAULT NULL
) RETURNS int
LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$
DECLARE
  rec record;
  v_count int := 0;
  v_status text;
BEGIN
  IF current_company_id() IS DISTINCT FROM p_company_id THEN
    RAISE EXCEPTION 'session company does not match batch company';
  END IF;

  SELECT status INTO v_status FROM import_batches
   WHERE company_id = p_company_id AND id = p_batch_id
   FOR UPDATE;
  IF NOT FOUND THEN
    RAISE EXCEPTION 'batch % not found', p_batch_id;
  END IF;
  IF v_status <> 'posted' THEN
    RAISE EXCEPTION 'batch % is not posted (status: %)', p_batch_id, v_status;
  END IF;

  FOR rec IN
    SELECT DISTINCT sj.posted_entry_id, e.entry_date
      FROM staging_journal sj
      JOIN journal_entries e
        ON e.company_id = sj.company_id AND e.id = sj.posted_entry_id
     WHERE sj.company_id = p_company_id AND sj.batch_id = p_batch_id
       AND sj.posted_entry_id IS NOT NULL
  LOOP
    PERFORM reverse_entry(p_company_id, rec.posted_entry_id,
      coalesce(p_reversal_date, rec.entry_date), p_user_id);
    v_count := v_count + 1;
  END LOOP;

  UPDATE import_batches SET status = 'rolled_back'
   WHERE company_id = p_company_id AND id = p_batch_id;
  RETURN v_count;
END $$;

-- Reconciliation helper: staged totals must tie to posted totals.
CREATE FUNCTION batch_reconciliation(p_company_id uuid, p_batch_id bigint)
RETURNS TABLE (
  batch_status text,
  staged_debits numeric(19,4), staged_credits numeric(19,4),
  posted_debits numeric(19,4), posted_credits numeric(19,4),
  matches boolean
)
LANGUAGE sql STABLE AS $$
  WITH b AS (
    SELECT status FROM import_batches
     WHERE company_id = p_company_id AND id = p_batch_id
  ), staged AS (
    SELECT coalesce(sum(parsed_debit), 0) AS d, coalesce(sum(parsed_credit), 0) AS c
      FROM staging_journal
     WHERE company_id = p_company_id AND batch_id = p_batch_id
  ), posted AS (
    SELECT coalesce(sum(l.debit), 0) AS d, coalesce(sum(l.credit), 0) AS c
      FROM journal_lines l
     WHERE l.company_id = p_company_id
       AND l.entry_id IN (
         SELECT posted_entry_id FROM staging_journal
          WHERE company_id = p_company_id AND batch_id = p_batch_id
            AND posted_entry_id IS NOT NULL)
  )
  -- matches is only claimed for a batch that is currently posted; a
  -- rolled-back batch's gross totals still tie but its net effect is zero.
  SELECT b.status,
         staged.d::numeric(19,4), staged.c::numeric(19,4),
         posted.d::numeric(19,4), posted.c::numeric(19,4),
         (b.status = 'posted' AND staged.d = posted.d AND staged.c = posted.c)
    FROM b, staged, posted
$$;
