create extension if not exists pgcrypto;

create schema if not exists eventhub;

-- Acquisition source definition. This represents where the imported source
-- record came from. It is tenant-scoped because the same provider/source keys
-- may be reused by different customers.
create table if not exists eventhub.event_source (
    id integer generated always as identity primary key,
    tenant_key text not null,
    provider_key text not null,
    source_kind text not null,
    source_key text not null,
    source_instance_key text not null default 'default',
    tenant_provider_setting_key text,
    external_fleet_key text,
    created_at timestamptz not null default now(),
    constraint ux_event_source unique (tenant_key, provider_key, source_kind, source_key, source_instance_key)
);

-- One execution of a provider acquisition job. A run may create many EventHub
-- extraction packages. Original provider packages are preserved separately as
-- source_package_* references, not used as EventHub package identity.
create table if not exists eventhub.import_run (
    id uuid primary key,
    tenant_key text not null,
    event_source_id integer not null references eventhub.event_source(id),
    mode text not null,
    status text not null,
    refresh_master_data_first boolean not null default true,

    source_group_type text,
    source_group_entity_id text,
    source_group_code text,
    source_group_name text,

    import_scope_type text not null,
    root_source_org_entity_id text,
    root_source_org_code text,
    root_source_org_name text,
    include_children boolean not null default false,
    occurred_from timestamptz,
    occurred_to timestamptz,

    requested_event_families text[] not null default '{}',
    acquisition_strategy text,
    metadata jsonb not null default '{}'::jsonb,
    planned_package_count integer not null default 0,
    started_at timestamptz not null default now(),
    finished_at timestamptz,
    error_message text,
    constraint chk_import_run_occ_time_order check (occurred_from is null or occurred_to is null or occurred_from < occurred_to)
);

-- Optional cursor table for scheduled/difference imports. For regular updates,
-- prefer source-package watermarks because new source packages may contain older
-- occurredAt events. Fallback strategies can use source-row updatedAt or occurredAt
-- overlap windows. Cursor scope is separated by scope_hash and event family/source kind.
create table if not exists eventhub.import_cursor (
    id uuid primary key,
    tenant_key text not null,
    event_source_id integer not null references eventhub.event_source(id),
    scope_hash text not null,
    event_family text not null,
    source_kind text not null,
    cursor_type text not null,
    last_source_package_imported_at timestamptz,
    last_source_package_id text,
    last_source_row_updated_at timestamptz,
    last_occurred_to timestamptz,
    updated_at timestamptz not null default now(),
    constraint ux_import_cursor unique (tenant_key, event_source_id, scope_hash, event_family, source_kind, cursor_type)
);

-- One EventHub acquisition package. It represents an extraction batch, not
-- necessarily one original provider package. A single extraction batch can
-- contain rows from multiple original provider packages; those are preserved on
-- acquired_event.source_package_* where available.
create table if not exists eventhub.data_package (
    id uuid primary key,
    event_source_id integer not null references eventhub.event_source(id),
    import_run_id uuid references eventhub.import_run(id),
    tenant_key text not null,
    package_key text not null,
    package_type text not null,
    status text not null,

    source_group_type text,
    source_group_entity_id text,
    source_group_code text,
    source_group_name text,

    import_scope_type text,
    root_source_org_entity_id text,
    root_source_org_code text,
    root_source_org_name text,
    include_children boolean not null default false,
    occurred_from timestamptz,
    occurred_to timestamptz,

    event_family text,
    business_date date,
    external_package_id text,

    extraction_code text,
    extraction_source_kind text,
    entity_axis text,
    batch_no integer,
    chunk_from timestamptz,
    chunk_to timestamptz,

    -- Optional package-level source package reference. Usually null for extraction
    -- batches, because the original source package is stored per acquired event.
    source_package_kind text,
    source_package_id text,
    source_package_entity_id text,
    source_package_period_from timestamptz,
    source_package_period_to timestamptz,
    source_package_imported_at timestamptz,

    received_at timestamptz not null default now(),
    completed_at timestamptz,
    event_count integer not null default 0,
    metadata jsonb not null default '{}'::jsonb,
    error_message text,
    constraint ux_data_package_package_key unique (tenant_key, event_source_id, package_key),
    constraint chk_data_package_occ_time_order check (occurred_from is null or occurred_to is null or occurred_from < occurred_to),
    constraint chk_data_package_chunk_time_order check (chunk_from is null or chunk_to is null or chunk_from < chunk_to)
);

-- Temporary acquisition-stage point-event store.
-- It keeps acquired point events with EventSource context, externalSourceEventId,
-- one occurredAt timestamp, source-side driver/vehicle refs, normalized event details,
-- optional original source-package reference, and raw JSON payload. Organisation is
-- intentionally not stored per event; it belongs to master-data relations for
-- driver/vehicle and is represented in importScope/sourceGroup.
create table if not exists eventhub.acquired_event (
    id uuid not null,
    event_source_id integer not null references eventhub.event_source(id),
    data_package_id uuid not null references eventhub.data_package(id),

    external_source_event_id text not null,

    driver_source_entity_id text,
    driver_card_nation text,
    driver_card_number text,

    vehicle_source_entity_id text,
    vehicle_vin text,
    vehicle_registration_nation text,
    vehicle_registration_number text,

    source_package_kind text,
    source_package_id text,
    source_package_entity_id text,
    source_package_period_from timestamptz,
    source_package_period_to timestamptz,
    source_package_imported_at timestamptz,

    occurred_at timestamptz not null,
    received_partner_at timestamptz,
    received_hub_at timestamptz not null default now(),

    event_domain text not null,
    event_type text not null,
    lifecycle text not null,

    odometer_m bigint,
    latitude numeric(10, 7),
    longitude numeric(10, 7),

    event_details jsonb not null default '{}'::jsonb,
    payload jsonb not null default '{}'::jsonb,
    manual_entry boolean not null default false,

    -- Includes tenant + EventSource + externalSourceEventId. Used only for source-record import idempotency.
    -- It does not merge equivalent events from different providers/sources.
    source_record_key_hash text not null,

    -- Non-unique semantic acquisition signature. It intentionally excludes EventSource and
    -- externalSourceEventId and is only a later query/projection hint for source comparison,
    -- gap filling, and merged timelines. It must not be used for import deduplication.
    event_signature_hash text,

    created_at timestamptz not null default now(),

    constraint pk_acquired_event primary key (occurred_at, id),
    constraint chk_acquired_event_driver_or_vehicle_ref check (
        driver_source_entity_id is not null
        or driver_card_number is not null
        or vehicle_source_entity_id is not null
        or vehicle_vin is not null
        or vehicle_registration_number is not null
    ),
    constraint chk_acquired_event_position_pair check ((latitude is null and longitude is null) or (latitude is not null and longitude is not null)),
    constraint chk_driver_card_nation_when_number check (driver_card_number is null or driver_card_nation is not null),
    constraint chk_vehicle_registration_nation_when_number check (vehicle_registration_number is null or vehicle_registration_nation is not null),
    constraint chk_acquired_event_source_package_time_order check (source_package_period_from is null or source_package_period_to is null or source_package_period_from < source_package_period_to)
);

create unique index if not exists ux_acquired_event_source_record
    on eventhub.acquired_event(source_record_key_hash);

create index if not exists idx_acquired_event_signature
    on eventhub.acquired_event(event_signature_hash)
    where event_signature_hash is not null;

create index if not exists idx_acquired_event_source_package
    on eventhub.acquired_event(source_package_kind, source_package_id, source_package_imported_at)
    where source_package_id is not null;

create index if not exists idx_acquired_event_vehicle_vin_time
    on eventhub.acquired_event(vehicle_vin, occurred_at desc)
    where vehicle_vin is not null;

create index if not exists idx_acquired_event_vehicle_registration_time
    on eventhub.acquired_event(vehicle_registration_nation, vehicle_registration_number, occurred_at desc)
    where vehicle_registration_number is not null;

create index if not exists idx_acquired_event_driver_card_time
    on eventhub.acquired_event(driver_card_nation, driver_card_number, occurred_at desc)
    where driver_card_number is not null;

create index if not exists idx_acquired_event_domain_type_time
    on eventhub.acquired_event(event_domain, event_type, occurred_at desc);

create index if not exists idx_acquired_event_details_gin
    on eventhub.acquired_event using gin(event_details);

create index if not exists idx_acquired_event_payload_gin
    on eventhub.acquired_event using gin(payload);

create index if not exists idx_data_package_source_time
    on eventhub.data_package(tenant_key, event_source_id, received_at desc);

create index if not exists idx_data_package_scope
    on eventhub.data_package(tenant_key, import_scope_type, root_source_org_entity_id, occurred_from, occurred_to);

create index if not exists idx_data_package_extraction
    on eventhub.data_package(tenant_key, event_source_id, import_run_id, event_family, extraction_source_kind, extraction_code, batch_no);

create index if not exists idx_import_run_source_status
    on eventhub.import_run(tenant_key, event_source_id, status, started_at desc);
