-- ARR database MVP v1: OSS/Agent ingestion, booking source facts, and Finance daily facts
-- PostgreSQL 15+
-- Destructive only inside the isolated booking_test database.
-- Apply only through the guarded executor after a verified backup and rollback probe.

BEGIN;

DO $$
BEGIN
    IF current_database() <> 'booking_test' THEN
        RAISE EXCEPTION
            'ARR MVP v1 rebuild is allowed only in booking_test';
    END IF;
END;
$$;

DROP SCHEMA IF EXISTS finance CASCADE;
DROP SCHEMA IF EXISTS booking CASCADE;
DROP SCHEMA IF EXISTS ingestion CASCADE;

CREATE SCHEMA ingestion;
CREATE SCHEMA booking;
CREATE SCHEMA finance;

COMMENT ON SCHEMA ingestion IS
  'ARR-owned OSS artifact identities, Agent processing, delivery validation, and reliable events.';

COMMENT ON SCHEMA booking IS
  'Booking source rows, immutable row-level Agent parses, and queryable original room items.';

COMMENT ON SCHEMA finance IS
  'Opera daily result sets and records used by ordinary programs to generate monthly and channel outputs.';

-- ---------------------------------------------------------------------------
-- Cross-domain ingestion
-- ---------------------------------------------------------------------------

CREATE TABLE ingestion.artifacts (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    artifact_kind text NOT NULL CHECK (artifact_kind IN (
        'booking_source_md',
        'booking_excel',
        'opera_xml',
        'daily_xlsx',
        'monthly_xlsx',
        'channel_detail_xlsx',
        'company_ten_day_xlsx',
        'exception_xlsx',
        'result_json',
        'structured_result_json'
    )),
    storage_provider text NOT NULL CHECK (storage_provider IN (
        'oss', 's3', 'local_fixture'
    )),
    bucket_alias text NOT NULL CHECK (
        btrim(bucket_alias) <> ''
        AND bucket_alias !~ '[/\\]'
    ),
    object_key text NOT NULL CHECK (
        btrim(object_key) <> ''
        AND object_key !~ '^/'
        AND object_key !~ '(^|/)\.\.(/|$)'
        AND object_key !~ '[?#]'
    ),
    object_version_id text,
    etag text,
    original_filename text NOT NULL CHECK (btrim(original_filename) <> ''),
    sha256 character(64) NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'),
    byte_size bigint NOT NULL CHECK (byte_size >= 0),
    mime_type text,
    created_at timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT artifacts_kind_hash_unique UNIQUE (artifact_kind, sha256)
);

CREATE UNIQUE INDEX artifacts_storage_identity_unique
    ON ingestion.artifacts (
        storage_provider,
        bucket_alias,
        object_key,
        COALESCE(object_version_id, '')
    );

COMMENT ON TABLE ingestion.artifacts IS
  'Metadata-only identity for immutable private objects. Never stores bytes, signed URLs, access keys, or Agent-local paths.';

CREATE TABLE ingestion.processing_runs (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    run_key text NOT NULL UNIQUE CHECK (
        run_key ~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$'
    ),
    pipeline_type text NOT NULL CHECK (pipeline_type IN (
        'opera_daily',
        'booking_source_import'
    )),
    source_artifact_id bigint NOT NULL
        REFERENCES ingestion.artifacts(id),
    result_artifact_id bigint
        REFERENCES ingestion.artifacts(id),
    run_status text NOT NULL CHECK (run_status IN (
        'received',
        'queued',
        'running',
        'validating',
        'accepted',
        'rejected',
        'failed',
        'cancelled'
    )),
    requested_processor_version text,
    requested_rule_set_sha256 character(64) CHECK (
        requested_rule_set_sha256 IS NULL
        OR requested_rule_set_sha256 ~ '^[0-9a-f]{64}$'
    ),
    delivered_processor_version text,
    delivered_rule_set_sha256 character(64) CHECK (
        delivered_rule_set_sha256 IS NULL
        OR delivered_rule_set_sha256 ~ '^[0-9a-f]{64}$'
    ),
    result_schema_version text,
    delivery_sha256 character(64) CHECK (
        delivery_sha256 IS NULL
        OR delivery_sha256 ~ '^[0-9a-f]{64}$'
    ),
    delivery_json jsonb NOT NULL DEFAULT '{}'::jsonb CHECK (
        jsonb_typeof(delivery_json) = 'object'
    ),
    business_date date,
    failure_code text,
    failure_message text,
    created_at timestamptz NOT NULL DEFAULT now(),
    updated_at timestamptz NOT NULL DEFAULT now(),
    validated_at timestamptz,
    finished_at timestamptz,
    CONSTRAINT processing_runs_terminal_shape CHECK (
        (
            run_status = 'accepted'
            AND failure_code IS NULL
            AND validated_at IS NOT NULL
            AND finished_at IS NOT NULL
            AND (
                pipeline_type <> 'opera_daily'
                OR (
                    result_artifact_id IS NOT NULL
                    AND business_date IS NOT NULL
                    AND delivered_processor_version IS NOT NULL
                    AND delivered_rule_set_sha256 IS NOT NULL
                    AND result_schema_version IS NOT NULL
                    AND delivery_sha256 IS NOT NULL
                )
            )
        )
        OR (
            run_status IN ('rejected', 'failed')
            AND failure_code IS NOT NULL
            AND finished_at IS NOT NULL
        )
        OR (
            run_status = 'cancelled'
            AND finished_at IS NOT NULL
        )
        OR run_status IN ('received', 'queued', 'running', 'validating')
    )
);

CREATE TABLE ingestion.processing_attempts (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    processing_run_id bigint NOT NULL
        REFERENCES ingestion.processing_runs(id),
    attempt_no integer NOT NULL CHECK (attempt_no > 0),
    attempt_status text NOT NULL CHECK (attempt_status IN (
        'queued',
        'dispatched',
        'running',
        'delivered',
        'succeeded',
        'failed',
        'cancelled'
    )),
    idempotency_key character(64) NOT NULL UNIQUE CHECK (
        idempotency_key ~ '^[0-9a-f]{64}$'
    ),
    remote_run_id text UNIQUE,
    failure_code text,
    failure_message text,
    created_at timestamptz NOT NULL DEFAULT now(),
    started_at timestamptz,
    finished_at timestamptz,
    CONSTRAINT processing_attempts_run_number_unique
        UNIQUE (processing_run_id, attempt_no),
    CONSTRAINT processing_attempts_id_run_unique
        UNIQUE (id, processing_run_id),
    CONSTRAINT processing_attempts_terminal_shape CHECK (
        (
            attempt_status = 'succeeded'
            AND failure_code IS NULL
            AND finished_at IS NOT NULL
        )
        OR (
            attempt_status = 'failed'
            AND failure_code IS NOT NULL
            AND finished_at IS NOT NULL
        )
        OR (
            attempt_status = 'cancelled'
            AND finished_at IS NOT NULL
        )
        OR attempt_status IN ('queued', 'dispatched', 'running', 'delivered')
    )
);

CREATE TABLE ingestion.processing_deliveries (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    delivery_key text NOT NULL UNIQUE CHECK (
        delivery_key ~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$'
    ),
    processing_run_id bigint NOT NULL
        REFERENCES ingestion.processing_runs(id),
    attempt_id bigint,
    envelope_sha256 character(64) NOT NULL CHECK (
        envelope_sha256 ~ '^[0-9a-f]{64}$'
    ),
    envelope_json jsonb NOT NULL CHECK (
        jsonb_typeof(envelope_json) = 'object'
    ),
    delivery_status text NOT NULL CHECK (delivery_status IN (
        'received',
        'validating',
        'committed',
        'recorded_failure',
        'rejected'
    )),
    result_status text NOT NULL CHECK (result_status IN (
        'success', 'failed'
    )),
    processor_version text NOT NULL CHECK (btrim(processor_version) <> ''),
    rule_set_sha256 character(64) NOT NULL CHECK (
        rule_set_sha256 ~ '^[0-9a-f]{64}$'
    ),
    result_schema_version text NOT NULL CHECK (
        btrim(result_schema_version) <> ''
    ),
    business_date date,
    source_artifact_id bigint
        REFERENCES ingestion.artifacts(id),
    daily_report_artifact_id bigint
        REFERENCES ingestion.artifacts(id),
    result_json_artifact_id bigint
        REFERENCES ingestion.artifacts(id),
    structured_result_artifact_id bigint
        REFERENCES ingestion.artifacts(id),
    exception_report_artifact_id bigint
        REFERENCES ingestion.artifacts(id),
    daily_version_id bigint,
    failure_code text,
    failure_message text,
    received_at timestamptz NOT NULL DEFAULT now(),
    validated_at timestamptz,
    committed_at timestamptz,
    CONSTRAINT processing_deliveries_attempt_run_fk
        FOREIGN KEY (attempt_id, processing_run_id)
        REFERENCES ingestion.processing_attempts(id, processing_run_id),
    CONSTRAINT processing_deliveries_attempt_envelope_unique
        UNIQUE (attempt_id, envelope_sha256)
);

CREATE TABLE ingestion.outbox_events (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    event_key text NOT NULL UNIQUE CHECK (btrim(event_key) <> ''),
    aggregate_type text NOT NULL CHECK (aggregate_type IN (
        'processing_run', 'daily_version', 'booking_source_batch'
    )),
    aggregate_id bigint NOT NULL CHECK (aggregate_id > 0),
    event_type text NOT NULL CHECK (btrim(event_type) <> ''),
    payload jsonb NOT NULL CHECK (jsonb_typeof(payload) = 'object'),
    publish_status text NOT NULL DEFAULT 'pending' CHECK (
        publish_status IN ('pending', 'publishing', 'published', 'dead')
    ),
    publish_attempts integer NOT NULL DEFAULT 0 CHECK (
        publish_attempts >= 0
    ),
    available_at timestamptz NOT NULL DEFAULT now(),
    created_at timestamptz NOT NULL DEFAULT now(),
    published_at timestamptz,
    last_error_code text,
    CONSTRAINT outbox_events_published_shape CHECK (
        (publish_status = 'published' AND published_at IS NOT NULL)
        OR publish_status <> 'published'
    )
);

-- ---------------------------------------------------------------------------
-- Booking source and row-level parse facts
-- ---------------------------------------------------------------------------

CREATE TABLE booking.source_batches (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    source_artifact_id bigint NOT NULL UNIQUE
        REFERENCES ingestion.artifacts(id),
    source_kind text NOT NULL CHECK (source_kind IN (
        'expected_fixture',
        'booking_excel'
    )),
    source_format_version text NOT NULL CHECK (
        btrim(source_format_version) <> ''
    ),
    reference_year integer CHECK (
        reference_year IS NULL OR reference_year BETWEEN 2000 AND 2199
    ),
    reference_month integer CHECK (
        reference_month IS NULL OR reference_month BETWEEN 1 AND 12
    ),
    batch_status text NOT NULL CHECK (batch_status IN (
        'processing', 'accepted', 'failed'
    )),
    source_rows integer NOT NULL DEFAULT 0 CHECK (source_rows >= 0),
    accepted_rows integer NOT NULL DEFAULT 0 CHECK (accepted_rows >= 0),
    failed_rows integer NOT NULL DEFAULT 0 CHECK (failed_rows >= 0),
    failure_code text,
    failure_message text,
    created_at timestamptz NOT NULL DEFAULT now(),
    validated_at timestamptz,
    finished_at timestamptz,
    CONSTRAINT source_batches_counts_valid CHECK (
        accepted_rows + failed_rows <= source_rows
    ),
    CONSTRAINT source_batches_terminal_shape CHECK (
        (
            batch_status = 'accepted'
            AND accepted_rows = source_rows
            AND failed_rows = 0
            AND failure_code IS NULL
            AND validated_at IS NOT NULL
            AND finished_at IS NOT NULL
        )
        OR (
            batch_status = 'failed'
            AND failure_code IS NOT NULL
            AND finished_at IS NOT NULL
        )
        OR batch_status = 'processing'
    )
);

CREATE TABLE booking.source_rows (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    source_batch_id bigint NOT NULL
        REFERENCES booking.source_batches(id),
    source_worksheet text NOT NULL CHECK (btrim(source_worksheet) <> ''),
    source_row_no integer NOT NULL CHECK (source_row_no > 0),
    group_code_raw text NOT NULL CHECK (btrim(group_code_raw) <> ''),
    group_code_key text GENERATED ALWAYS AS (
        upper(btrim(group_code_raw))
    ) STORED,
    type_of_room_raw text NOT NULL CHECK (btrim(type_of_room_raw) <> ''),
    no_of_rooms integer NOT NULL CHECK (no_of_rooms > 0),
    source_row_sha256 character(64) NOT NULL CHECK (
        source_row_sha256 ~ '^[0-9a-f]{64}$'
    ),
    created_at timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT source_rows_position_unique
        UNIQUE (source_batch_id, source_worksheet, source_row_no),
    CONSTRAINT source_rows_hash_unique
        UNIQUE (source_batch_id, source_row_sha256)
);

CREATE INDEX source_rows_group_code_idx
    ON booking.source_rows (group_code_key);

CREATE TABLE booking.parse_versions (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    source_row_id bigint NOT NULL
        REFERENCES booking.source_rows(id),
    version_no integer NOT NULL CHECK (version_no > 0),
    parse_status text NOT NULL CHECK (parse_status IN (
        'accepted', 'needs_review', 'failed'
    )),
    result_schema_version text NOT NULL CHECK (
        btrim(result_schema_version) <> ''
    ),
    processor_name text NOT NULL CHECK (btrim(processor_name) <> ''),
    processor_version text NOT NULL CHECK (btrim(processor_version) <> ''),
    rule_set_sha256 character(64) NOT NULL CHECK (
        rule_set_sha256 ~ '^[0-9a-f]{64}$'
    ),
    input_row_sha256 character(64) NOT NULL CHECK (
        input_row_sha256 ~ '^[0-9a-f]{64}$'
    ),
    result_sha256 character(64) CHECK (
        result_sha256 IS NULL OR result_sha256 ~ '^[0-9a-f]{64}$'
    ),
    result_json jsonb NOT NULL CHECK (jsonb_typeof(result_json) = 'object'),
    failure_code text,
    failure_message text,
    created_at timestamptz NOT NULL DEFAULT now(),
    validated_at timestamptz,
    CONSTRAINT parse_versions_source_version_unique
        UNIQUE (source_row_id, version_no),
    CONSTRAINT parse_versions_id_source_unique
        UNIQUE (id, source_row_id),
    CONSTRAINT parse_versions_status_shape CHECK (
        (
            parse_status = 'accepted'
            AND result_sha256 IS NOT NULL
            AND failure_code IS NULL
            AND validated_at IS NOT NULL
        )
        OR parse_status = 'needs_review'
        OR (
            parse_status = 'failed'
            AND failure_code IS NOT NULL
        )
    )
);

CREATE INDEX parse_versions_source_row_idx
    ON booking.parse_versions (source_row_id, version_no DESC);

CREATE TABLE booking.room_items (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    parse_version_id bigint NOT NULL
        REFERENCES booking.parse_versions(id),
    item_no integer NOT NULL CHECK (item_no > 0),
    room_type_raw text NOT NULL CHECK (btrim(room_type_raw) <> ''),
    room_type_code text CHECK (
        room_type_code IS NULL OR btrim(room_type_code) <> ''
    ),
    quantity integer NOT NULL CHECK (quantity > 0),
    unit_price numeric(18,2) CHECK (
        unit_price IS NULL OR unit_price >= 0
    ),
    currency_code character(3) CHECK (
        currency_code IS NULL OR currency_code ~ '^[A-Z]{3}$'
    ),
    price_token_raw text,
    source_fragment text,
    created_at timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT room_items_position_unique
        UNIQUE (parse_version_id, item_no)
);

CREATE INDEX room_items_type_idx
    ON booking.room_items (
        COALESCE(room_type_code, room_type_raw)
    );

CREATE TABLE booking.current_row_parses (
    source_row_id bigint PRIMARY KEY,
    parse_version_id bigint NOT NULL UNIQUE,
    activated_at timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT current_row_parses_version_fk
        FOREIGN KEY (parse_version_id, source_row_id)
        REFERENCES booking.parse_versions(id, source_row_id)
);

CREATE OR REPLACE FUNCTION booking.validate_current_row_parse()
RETURNS trigger
LANGUAGE plpgsql
AS $$
DECLARE
    source_quantity integer;
    parsed_quantity bigint;
BEGIN
    SELECT source.no_of_rooms
    INTO source_quantity
    FROM booking.parse_versions AS parsed
    JOIN booking.source_rows AS source
      ON source.id = parsed.source_row_id
    JOIN booking.source_batches AS batch
      ON batch.id = source.source_batch_id
    WHERE parsed.id = NEW.parse_version_id
      AND parsed.source_row_id = NEW.source_row_id
      AND parsed.parse_status = 'accepted'
      AND batch.batch_status = 'accepted';

    IF source_quantity IS NULL THEN
        RAISE EXCEPTION
            'current row parse must reference an accepted parse in an accepted source batch';
    END IF;

    SELECT COALESCE(sum(item.quantity), 0)
    INTO parsed_quantity
    FROM booking.room_items AS item
    WHERE item.parse_version_id = NEW.parse_version_id;

    IF parsed_quantity <> source_quantity THEN
        RAISE EXCEPTION
            'parsed room quantity % does not equal source room quantity %',
            parsed_quantity,
            source_quantity;
    END IF;
    RETURN NEW;
END;
$$;

CREATE TRIGGER current_row_parses_acceptance_guard
BEFORE INSERT OR UPDATE ON booking.current_row_parses
FOR EACH ROW
EXECUTE FUNCTION booking.validate_current_row_parse();

CREATE VIEW booking.v_current_room_items AS
SELECT
    source.id AS source_row_id,
    source.source_batch_id,
    source.source_worksheet,
    source.source_row_no,
    source.group_code_raw,
    source.group_code_key,
    source.type_of_room_raw,
    source.no_of_rooms AS source_no_of_rooms,
    parsed.id AS parse_version_id,
    parsed.version_no AS parse_version_no,
    item.item_no,
    item.room_type_raw,
    item.room_type_code,
    item.quantity,
    item.unit_price,
    item.currency_code,
    item.price_token_raw,
    item.source_fragment
FROM booking.current_row_parses AS current_parse
JOIN booking.parse_versions AS parsed
  ON parsed.id = current_parse.parse_version_id
JOIN booking.source_rows AS source
  ON source.id = current_parse.source_row_id
JOIN booking.source_batches AS batch
  ON batch.id = source.source_batch_id
JOIN booking.room_items AS item
  ON item.parse_version_id = parsed.id
WHERE batch.batch_status = 'accepted'
  AND parsed.parse_status = 'accepted';

CREATE VIEW booking.v_group_room_item_summary AS
SELECT
    current_item.group_code_key,
    COALESCE(
        NULLIF(btrim(current_item.room_type_code), ''),
        current_item.room_type_raw
    ) AS booking_room_type,
    sum(current_item.quantity)::bigint AS quantity,
    count(DISTINCT current_item.source_row_id)::bigint AS source_row_count
FROM booking.v_current_room_items AS current_item
GROUP BY
    current_item.group_code_key,
    COALESCE(
        NULLIF(btrim(current_item.room_type_code), ''),
        current_item.room_type_raw
    );

CREATE VIEW booking.v_group_booking_rooms AS
WITH room_summary AS (
    SELECT
        summary.group_code_key,
        string_agg(
            summary.booking_room_type || ' × ' || summary.quantity::text,
            ', '
            ORDER BY summary.booking_room_type
        ) AS booking_room,
        sum(summary.quantity)::bigint AS total_booking_rooms
    FROM booking.v_group_room_item_summary AS summary
    GROUP BY summary.group_code_key
),
source_summary AS (
    SELECT
        current_item.group_code_key,
        count(DISTINCT current_item.source_row_id)::bigint
            AS source_row_count
    FROM booking.v_current_room_items AS current_item
    GROUP BY current_item.group_code_key
)
SELECT
    room_summary.group_code_key,
    room_summary.booking_room,
    room_summary.total_booking_rooms,
    source_summary.source_row_count
FROM room_summary
JOIN source_summary
  ON source_summary.group_code_key = room_summary.group_code_key;

COMMENT ON VIEW booking.v_group_booking_rooms IS
  'Query-time Group Code summary. Original source rows remain unmerged and traceable.';

-- ---------------------------------------------------------------------------
-- Finance daily facts
-- ---------------------------------------------------------------------------

CREATE TABLE finance.daily_versions (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    business_date date,
    version_no integer,
    processing_run_id bigint NOT NULL UNIQUE
        REFERENCES ingestion.processing_runs(id),
    source_artifact_id bigint NOT NULL
        REFERENCES ingestion.artifacts(id),
    daily_report_artifact_id bigint
        REFERENCES ingestion.artifacts(id),
    result_json_artifact_id bigint
        REFERENCES ingestion.artifacts(id),
    structured_result_artifact_id bigint
        REFERENCES ingestion.artifacts(id),
    exception_report_artifact_id bigint
        REFERENCES ingestion.artifacts(id),
    version_status text NOT NULL CHECK (version_status IN (
        'validated', 'active', 'superseded', 'rejected'
    )),
    processor_version text NOT NULL CHECK (btrim(processor_version) <> ''),
    rule_set_sha256 character(64) NOT NULL CHECK (
        rule_set_sha256 ~ '^[0-9a-f]{64}$'
    ),
    result_schema_version text NOT NULL CHECK (
        btrim(result_schema_version) <> ''
    ),
    result_sha256 character(64) NOT NULL CHECK (
        result_sha256 ~ '^[0-9a-f]{64}$'
    ),
    source_rows integer NOT NULL CHECK (source_rows >= 0),
    retained_rows integer NOT NULL CHECK (retained_rows >= 0),
    excluded_rate_code_rows integer NOT NULL CHECK (
        excluded_rate_code_rows >= 0
    ),
    duplicate_rows integer NOT NULL CHECK (duplicate_rows >= 0),
    validation_failed_rows integer NOT NULL CHECK (
        validation_failed_rows >= 0
    ),
    price_unmatched_rows integer NOT NULL CHECK (
        price_unmatched_rows >= 0
    ),
    failure_code text,
    failure_message text,
    created_at timestamptz NOT NULL DEFAULT now(),
    validated_at timestamptz,
    activated_at timestamptz,
    superseded_at timestamptz,
    CONSTRAINT daily_versions_date_number_shape CHECK (
        (
            business_date IS NOT NULL
            AND version_no IS NOT NULL
            AND version_no > 0
        )
        OR (
            business_date IS NULL
            AND version_no IS NULL
            AND version_status = 'rejected'
        )
    ),
    CONSTRAINT daily_versions_business_version_unique
        UNIQUE (business_date, version_no),
    CONSTRAINT daily_versions_id_date_unique
        UNIQUE (id, business_date),
    CONSTRAINT daily_versions_counts_reconcile CHECK (
        source_rows =
            retained_rows
            + excluded_rate_code_rows
            + duplicate_rows
            + validation_failed_rows
            + price_unmatched_rows
    ),
    CONSTRAINT daily_versions_status_shape CHECK (
        (
            version_status IN ('validated', 'active', 'superseded')
            AND business_date IS NOT NULL
            AND daily_report_artifact_id IS NOT NULL
            AND result_json_artifact_id IS NOT NULL
            AND structured_result_artifact_id IS NOT NULL
            AND exception_report_artifact_id IS NULL
            AND failure_code IS NULL
            AND validated_at IS NOT NULL
        )
        OR (
            version_status = 'rejected'
            AND failure_code IS NOT NULL
        )
    ),
    CONSTRAINT daily_versions_active_shape CHECK (
        version_status <> 'active'
        OR activated_at IS NOT NULL
    )
);

CREATE UNIQUE INDEX daily_versions_source_rule_unique
    ON finance.daily_versions (
        source_artifact_id,
        business_date,
        processor_version,
        rule_set_sha256
    )
    WHERE business_date IS NOT NULL;

CREATE INDEX daily_versions_business_date_idx
    ON finance.daily_versions (business_date, version_no DESC);

ALTER TABLE ingestion.processing_deliveries
    ADD CONSTRAINT processing_deliveries_daily_version_fk
        FOREIGN KEY (daily_version_id)
        REFERENCES finance.daily_versions(id);

CREATE TABLE finance.daily_records (
    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    daily_version_id bigint NOT NULL
        REFERENCES finance.daily_versions(id),
    source_sequence integer NOT NULL CHECK (source_sequence > 0),
    source_location text NOT NULL CHECK (btrim(source_location) <> ''),
    source_worksheet text,
    source_row_no integer,
    outcome text NOT NULL CHECK (outcome IN (
        'retained',
        'excluded_rate_code',
        'duplicate',
        'validation_failed',
        'price_unmatched'
    )),
    decision_codes text[] NOT NULL CHECK (
        cardinality(decision_codes) > 0
    ),
    duplicate_of_record_id bigint,

    block_code text,
    adults integer,
    children integer,
    company_name text,
    company_key text,
    confirmation_no text,
    disp_room_no text,
    effective_rate_amount numeric(18,2),
    full_name text,
    res_comment text,
    group_code_key text GENERATED ALWAYS AS (
        NULLIF(upper(btrim(res_comment)), '')
    ) STORED,
    trace_text text,
    no_of_rooms integer,
    products text,
    rate_code text,
    normalized_rate_code text GENERATED ALWAYS AS (
        NULLIF(upper(btrim(rate_code)), '')
    ) STORED,
    room_category_label text,
    arrival date,
    departure date,
    nights integer,
    real_price numeric(18,2),
    total_price numeric(18,2),
    kb_amount numeric(18,2),
    channel_key text,
    pricing_method text,

    booking_source_match_status text NOT NULL CHECK (
        booking_source_match_status IN (
            'matched',
            'unmatched',
            'missing_group_code',
            'not_checked'
        )
    ),
    booking_source_match_count integer NOT NULL DEFAULT 0 CHECK (
        booking_source_match_count >= 0
    ),
    created_at timestamptz NOT NULL DEFAULT now(),

    CONSTRAINT daily_records_source_unique
        UNIQUE (daily_version_id, source_sequence),
    CONSTRAINT daily_records_id_version_unique
        UNIQUE (id, daily_version_id),
    CONSTRAINT daily_records_duplicate_version_fk
        FOREIGN KEY (duplicate_of_record_id, daily_version_id)
        REFERENCES finance.daily_records(id, daily_version_id),
    CONSTRAINT daily_records_source_coordinate_shape CHECK (
        (source_worksheet IS NULL AND source_row_no IS NULL)
        OR (
            source_worksheet IS NOT NULL
            AND btrim(source_worksheet) <> ''
            AND source_row_no > 0
        )
    ),
    CONSTRAINT daily_records_duplicate_shape CHECK (
        (outcome = 'duplicate' AND duplicate_of_record_id IS NOT NULL)
        OR (outcome <> 'duplicate' AND duplicate_of_record_id IS NULL)
    ),
    CONSTRAINT daily_records_booking_match_shape CHECK (
        (
            booking_source_match_status = 'matched'
            AND group_code_key IS NOT NULL
            AND booking_source_match_count > 0
        )
        OR (
            booking_source_match_status = 'unmatched'
            AND group_code_key IS NOT NULL
            AND booking_source_match_count = 0
        )
        OR (
            booking_source_match_status = 'missing_group_code'
            AND group_code_key IS NULL
            AND booking_source_match_count = 0
        )
        OR (
            booking_source_match_status = 'not_checked'
            AND booking_source_match_count = 0
        )
    ),
    CONSTRAINT daily_records_retained_checked CHECK (
        outcome <> 'retained'
        OR booking_source_match_status <> 'not_checked'
    ),
    CONSTRAINT daily_records_retained_values CHECK (
        outcome <> 'retained'
        OR (
            block_code IS NOT NULL
            AND adults IS NOT NULL
            AND adults >= 0
            AND children IS NOT NULL
            AND children >= 0
            AND company_name IS NOT NULL
            AND btrim(company_name) <> ''
            AND company_key IS NOT NULL
            AND btrim(company_key) <> ''
            AND confirmation_no IS NOT NULL
            AND btrim(confirmation_no) <> ''
            AND disp_room_no IS NOT NULL
            AND btrim(disp_room_no) <> ''
            AND effective_rate_amount IS NOT NULL
            AND effective_rate_amount >= 0
            AND full_name IS NOT NULL
            AND no_of_rooms IS NOT NULL
            AND no_of_rooms > 0
            AND rate_code IS NOT NULL
            AND normalized_rate_code IS NOT NULL
            AND arrival IS NOT NULL
            AND departure IS NOT NULL
            AND nights IS NOT NULL
            AND nights >= 0
            AND real_price IS NOT NULL
            AND real_price >= 0
            AND total_price IS NOT NULL
            AND total_price >= 0
            AND channel_key IS NOT NULL
            AND btrim(channel_key) <> ''
            AND pricing_method IN (
                'price_reference_exact',
                'zero_price_exception'
            )
        )
    ),
    CONSTRAINT daily_records_dates_nights_valid CHECK (
        outcome <> 'retained'
        OR (
            departure >= arrival
            AND nights = departure - arrival
        )
    ),
    CONSTRAINT daily_records_total_price_valid CHECK (
        outcome <> 'retained'
        OR total_price = real_price * no_of_rooms * nights
    ),
    CONSTRAINT daily_records_kb_valid CHECK (
        outcome <> 'retained'
        OR kb_amount IS NULL
        OR kb_amount = no_of_rooms * 100
    )
);

CREATE UNIQUE INDEX daily_records_retained_dedupe_key
    ON finance.daily_records (
        daily_version_id,
        disp_room_no,
        arrival
    )
    WHERE outcome = 'retained';

CREATE INDEX daily_records_version_outcome_idx
    ON finance.daily_records (daily_version_id, outcome);

CREATE INDEX daily_records_group_code_idx
    ON finance.daily_records (group_code_key)
    WHERE group_code_key IS NOT NULL;

CREATE INDEX daily_records_company_date_idx
    ON finance.daily_records (company_key, arrival);

CREATE TABLE finance.daily_channel_metrics (
    daily_version_id bigint NOT NULL
        REFERENCES finance.daily_versions(id),
    channel_key text NOT NULL CHECK (btrim(channel_key) <> ''),
    channel_order integer NOT NULL CHECK (channel_order > 0),
    row_count integer NOT NULL CHECK (row_count >= 0),
    PRIMARY KEY (daily_version_id, channel_key),
    CONSTRAINT daily_channel_metrics_order_unique
        UNIQUE (daily_version_id, channel_order)
);

CREATE TABLE finance.current_daily_versions (
    business_date date PRIMARY KEY,
    daily_version_id bigint NOT NULL,
    activated_at timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT current_daily_versions_version_fk
        FOREIGN KEY (daily_version_id, business_date)
        REFERENCES finance.daily_versions(id, business_date)
);

CREATE OR REPLACE FUNCTION finance.validate_current_daily_version()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
    IF NOT EXISTS (
        SELECT 1
        FROM finance.daily_versions AS version
        WHERE version.id = NEW.daily_version_id
          AND version.business_date = NEW.business_date
          AND version.version_status = 'active'
    ) THEN
        RAISE EXCEPTION
            'current business date must reference an active daily version';
    END IF;
    RETURN NEW;
END;
$$;

CREATE TRIGGER current_daily_versions_active_guard
BEFORE INSERT OR UPDATE ON finance.current_daily_versions
FOR EACH ROW
EXECUTE FUNCTION finance.validate_current_daily_version();

-- ---------------------------------------------------------------------------
-- Read models. Monthly and channel rows are generated, not duplicated.
-- ---------------------------------------------------------------------------

CREATE VIEW finance.v_active_daily_facts AS
SELECT
    version.business_date,
    record.*
FROM finance.current_daily_versions AS current_version
JOIN finance.daily_versions AS version
  ON version.id = current_version.daily_version_id
JOIN finance.daily_records AS record
  ON record.daily_version_id = current_version.daily_version_id
WHERE record.outcome = 'retained';

CREATE VIEW finance.v_monthly_report_rows AS
SELECT
    fact.business_date,
    fact.id AS daily_record_id,
    fact.arrival,
    fact.departure,
    fact.nights,
    fact.adults,
    fact.children,
    fact.block_code,
    fact.no_of_rooms,
    fact.company_name,
    fact.confirmation_no,
    fact.disp_room_no,
    fact.effective_rate_amount AS rate_amount,
    fact.full_name,
    fact.res_comment,
    fact.trace_text,
    fact.products,
    fact.rate_code,
    fact.room_category_label AS room_category,
    fact.real_price,
    booking_room.booking_room,
    fact.total_price
FROM finance.v_active_daily_facts AS fact
LEFT JOIN booking.v_group_booking_rooms AS booking_room
  ON booking_room.group_code_key = fact.group_code_key;

CREATE VIEW finance.v_channel_details AS
SELECT
    fact.business_date,
    fact.id AS daily_record_id,
    fact.channel_key,
    fact.arrival,
    fact.departure,
    fact.nights,
    fact.block_code,
    fact.res_comment,
    fact.group_code_key,
    booking_room.booking_room,
    fact.total_price AS total_booking_price
FROM finance.v_active_daily_facts AS fact
LEFT JOIN booking.v_group_booking_rooms AS booking_room
  ON booking_room.group_code_key = fact.group_code_key;

CREATE VIEW finance.v_company_report_source AS
SELECT
    fact.business_date,
    fact.id AS daily_record_id,
    fact.channel_key,
    fact.company_key,
    fact.company_name,
    fact.block_code,
    fact.group_code_key,
    fact.res_comment,
    booking_room.booking_room AS type_of_room,
    booking_room.total_booking_rooms,
    fact.no_of_rooms AS actual_room_quantity,
    fact.arrival,
    fact.departure,
    fact.nights,
    fact.real_price,
    fact.total_price
FROM finance.v_active_daily_facts AS fact
LEFT JOIN booking.v_group_booking_rooms AS booking_room
  ON booking_room.group_code_key = fact.group_code_key;

CREATE VIEW finance.v_daily_processing_audit AS
SELECT
    version.business_date,
    version.id AS daily_version_id,
    version.version_no,
    version.version_status,
    run.id AS processing_run_id,
    run.run_key,
    artifact.original_filename AS source_filename,
    artifact.sha256 AS source_sha256,
    record.id AS daily_record_id,
    record.source_sequence,
    record.source_location,
    record.outcome,
    record.decision_codes,
    record.duplicate_of_record_id,
    record.block_code,
    record.res_comment,
    record.group_code_key,
    record.rate_code,
    record.booking_source_match_status,
    record.booking_source_match_count,
    version.processor_version,
    version.rule_set_sha256,
    version.result_schema_version,
    version.validated_at,
    version.activated_at
FROM finance.daily_versions AS version
JOIN ingestion.processing_runs AS run
  ON run.id = version.processing_run_id
JOIN ingestion.artifacts AS artifact
  ON artifact.id = version.source_artifact_id
JOIN finance.daily_records AS record
  ON record.daily_version_id = version.id;

COMMENT ON VIEW finance.v_active_daily_facts IS
  'Current active version and retained rows only; authoritative source for ordinary downstream programs.';

COMMENT ON VIEW finance.v_monthly_report_rows IS
  'Monthly export projection. RATE_AMOUNT/ROOM_CATEGORY aliases and Booking Room enrichment are computed without storing monthly rows.';

COMMENT ON VIEW finance.v_channel_details IS
  'Channel detail projection. total_booking_price is exactly Finance daily total_price.';

COMMENT ON VIEW finance.v_daily_processing_audit IS
  'All outcomes and deterministic lineage for later read-only processing trace APIs.';

COMMIT;