schema.sql

-- =============================================================================
-- Polza Agency test task — database schema
--
-- Design notes
--   * `companies` holds the CLEAN, deduplicated business records. One row per
--     real-world company. Every column that survived validation is typed
--     properly (numeric / int), so SQL analytics need no casting or guessing.
--   * `import_issues` is the audit trail: every value the importer had to
--     repair, reject or flag is recorded here with the original raw text.
--     Nothing is silently dropped — Task 3's ANOMALIES.md is generated from
--     exactly this table.
--   * `import_runs` records each execution so re-imports are traceable and the
--     idempotency claim is verifiable from inside the DB.
--
-- Idempotency
--   Imports are UPSERTs keyed on `source_id`. Running any importer twice
--   produces the same number of rows and the same content; only `updated_at`
--   and the import-run bookkeeping change.
--
-- This script is safe to run repeatedly (IF NOT EXISTS everywhere).
-- =============================================================================

-- Trigram index support for fast case-insensitive substring search on names.
CREATE EXTENSION IF NOT EXISTS pg_trgm;

-- -----------------------------------------------------------------------------
-- Import runs
-- -----------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS import_runs (
    id              bigserial PRIMARY KEY,
    source          text        NOT NULL,   -- 'json_pack' | 'review_csv'
    started_at      timestamptz NOT NULL DEFAULT now(),
    finished_at     timestamptz,
    rows_read       integer     NOT NULL DEFAULT 0,  -- raw records seen in the files
    rows_inserted   integer     NOT NULL DEFAULT 0,  -- new companies created
    rows_updated    integer     NOT NULL DEFAULT 0,  -- existing companies refreshed
    rows_duplicate  integer     NOT NULL DEFAULT 0,  -- collapsed by dedup before hitting the DB
    rows_rejected   integer     NOT NULL DEFAULT 0,  -- unusable rows (e.g. no id/name)
    issues_logged   integer     NOT NULL DEFAULT 0
);

COMMENT ON TABLE import_runs IS 'One row per importer execution; makes re-imports auditable.';

-- -----------------------------------------------------------------------------
-- Companies (clean layer)
-- -----------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS companies (
    id             bigserial PRIMARY KEY,

    -- Natural key from the source system ("c_000123"). Unique => idempotent upserts.
    source_id      text        NOT NULL UNIQUE,

    name           text        NOT NULL,
    -- Normalized forms are what dedup and search actually compare against:
    -- lowercased, whitespace-collapsed, legal-form prefixes and quotes removed.
    name_norm      text        NOT NULL,

    category       text,
    city           text,
    -- NOT NULL (empty string instead of NULL) so the natural-key unique index
    -- below actually constrains rows: in Postgres, NULLs never conflict.
    city_norm      text        NOT NULL DEFAULT '',
    address        text,
    address_norm   text        NOT NULL DEFAULT '',

    -- Ratings are 0..5 by definition of the source product; anything outside
    -- that range was rejected at import time and logged in import_issues.
    rating         numeric(2,1) CHECK (rating IS NULL OR (rating >= 0 AND rating <= 5)),
    reviews_count  integer      CHECK (reviews_count IS NULL OR reviews_count >= 0),

    site           text CHECK (site IS NULL OR site ~ '^https?://'),
    phone          text,          -- as displayed, e.g. "+7 (495) 248-44-40"
    phone_e164     text CHECK (phone_e164 IS NULL OR phone_e164 ~ '^\+7[0-9]{10}$'),

    -- Provenance
    source         text        NOT NULL,   -- 'json_pack' | 'review_csv'
    source_file    text        NOT NULL,   -- e.g. 'page_003.json'
    -- True when the record needed repairs or carries a data-quality flag.
    is_suspect     boolean     NOT NULL DEFAULT false,

    first_seen_at  timestamptz NOT NULL DEFAULT now(),
    updated_at     timestamptz NOT NULL DEFAULT now()
);

COMMENT ON COLUMN companies.source_id  IS 'Upsert key: guarantees import idempotency.';
COMMENT ON COLUMN companies.is_suspect IS 'Row required repair/flagging; see import_issues.';

-- Cross-id deduplication guard.
-- Two records with different source_ids but the same normalized name + city +
-- address are the same business. A UNIQUE INDEX makes that impossible to insert
-- twice, so a future re-export with renumbered ids cannot silently duplicate.
CREATE UNIQUE INDEX IF NOT EXISTS companies_natural_key_uidx
    ON companies (name_norm, city_norm, address_norm);

-- Indexes chosen for the three analytical queries + the /companies page.
CREATE INDEX IF NOT EXISTS companies_category_idx      ON companies (category);
CREATE INDEX IF NOT EXISTS companies_city_idx          ON companies (city);
CREATE INDEX IF NOT EXISTS companies_rating_idx        ON companies (rating);
CREATE INDEX IF NOT EXISTS companies_reviews_count_idx ON companies (reviews_count);
-- Covers "average rating per city among companies with 10+ reviews".
CREATE INDEX IF NOT EXISTS companies_city_reviews_idx  ON companies (city, reviews_count)
    WHERE rating IS NOT NULL;
-- Substring search by name (ILIKE '%...%') without a sequential scan.
CREATE INDEX IF NOT EXISTS companies_name_trgm_idx     ON companies USING gin (name_norm gin_trgm_ops);

-- -----------------------------------------------------------------------------
-- Import issues (audit trail / anomaly evidence)
-- -----------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS import_issues (
    id           bigserial PRIMARY KEY,
    run_id       bigint      NOT NULL REFERENCES import_runs(id) ON DELETE CASCADE,
    source       text        NOT NULL,
    source_file  text        NOT NULL,
    row_number   integer,                 -- 1-based record index within the file
    source_id    text,                    -- may be NULL for rows that had no id
    field        text,                    -- offending column, NULL for row-level issues
    issue_code   text        NOT NULL,    -- machine-readable, e.g. 'rating_out_of_range'
    severity     text        NOT NULL CHECK (severity IN ('info', 'warning', 'error')),
    raw_value    text,                    -- exactly what was in the file
    action       text        NOT NULL,    -- what the importer did about it
    detail       text,
    created_at   timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS import_issues_run_idx   ON import_issues (run_id);
CREATE INDEX IF NOT EXISTS import_issues_code_idx  ON import_issues (issue_code);
CREATE INDEX IF NOT EXISTS import_issues_src_idx   ON import_issues (source, source_id);

-- The most recent import run for each source.
CREATE OR REPLACE VIEW latest_import_runs AS
SELECT DISTINCT ON (source) id, source, started_at, finished_at
FROM import_runs
ORDER BY source, id DESC;

-- Convenience view: the anomaly summary used to write ANOMALIES.md.
--
-- Scoped to the LATEST run per source. `import_issues` deliberately keeps the
-- history of every run, so aggregating over the whole table would double the
-- counts as soon as an import is repeated — which, since imports are designed
-- to be re-runnable, is exactly what happens in normal use.
CREATE OR REPLACE VIEW anomaly_summary AS
SELECT i.source,
       i.issue_code,
       i.severity,
       count(*)                                     AS occurrences,
       min(i.action)                                AS action,
       (array_agg(i.raw_value ORDER BY i.id))[1:3]  AS sample_values
FROM import_issues i
JOIN latest_import_runs r ON r.id = i.run_id
GROUP BY i.source, i.issue_code, i.severity
ORDER BY i.source, occurrences DESC, i.issue_code;