create table if not exists eventhub.driver_card (
    id uuid primary key,
    driver_id uuid references eventhub.driver(id),
    nation text not null,
    card_number text not null,
    source_updated_at timestamptz,
    payload jsonb not null default '{}'::jsonb,
    created_at timestamptz not null default now(),
    updated_at timestamptz not null default now()
);

create table if not exists eventhub.source_driver_identity (
    id uuid primary key,
    tenant_key text not null,
    event_source_id integer not null references eventhub.event_source(id),
    source_driver_entity_id text not null,
    driver_id uuid not null references eventhub.driver(id) on delete cascade,
    source_updated_at timestamptz,
    payload jsonb not null default '{}'::jsonb,
    created_at timestamptz not null default now(),
    updated_at timestamptz not null default now(),
    constraint ux_source_driver_identity unique (tenant_key, event_source_id, source_driver_entity_id)
);

create table if not exists eventhub.source_driver_card_identity (
    id uuid primary key,
    tenant_key text not null,
    event_source_id integer not null references eventhub.event_source(id),
    source_driver_card_entity_id text not null,
    driver_card_id uuid not null references eventhub.driver_card(id) on delete cascade,
    source_updated_at timestamptz,
    payload jsonb not null default '{}'::jsonb,
    created_at timestamptz not null default now(),
    updated_at timestamptz not null default now(),
    constraint ux_source_driver_card_identity unique (tenant_key, event_source_id, source_driver_card_entity_id)
);

create table if not exists eventhub.source_vehicle_identity (
    id uuid primary key,
    tenant_key text not null,
    event_source_id integer not null references eventhub.event_source(id),
    source_vehicle_entity_id text not null,
    vehicle_id uuid not null references eventhub.vehicle(id) on delete cascade,
    source_updated_at timestamptz,
    payload jsonb not null default '{}'::jsonb,
    created_at timestamptz not null default now(),
    updated_at timestamptz not null default now(),
    constraint ux_source_vehicle_identity unique (tenant_key, event_source_id, source_vehicle_entity_id)
);

create table if not exists eventhub.source_vehicle_registration_identity (
    id uuid primary key,
    tenant_key text not null,
    event_source_id integer not null references eventhub.event_source(id),
    source_registration_entity_id text not null,
    vehicle_registration_id uuid not null references eventhub.vehicle_registration(id) on delete cascade,
    source_updated_at timestamptz,
    payload jsonb not null default '{}'::jsonb,
    created_at timestamptz not null default now(),
    updated_at timestamptz not null default now(),
    constraint ux_source_vehicle_registration_identity unique (tenant_key, event_source_id, source_registration_entity_id)
);

alter table eventhub.event
    add column if not exists driver_card_id uuid references eventhub.driver_card(id);

insert into eventhub.source_driver_identity(
    id, tenant_key, event_source_id, source_driver_entity_id,
    driver_id, source_updated_at, payload, created_at, updated_at
)
select gen_random_uuid(),
       driver.tenant_key,
       driver.event_source_id,
       driver.source_driver_entity_id,
       driver.id,
       driver.source_updated_at,
       driver.payload,
       coalesce(driver.created_at, now()),
       coalesce(driver.updated_at, now())
from eventhub.driver driver
where exists (
    select 1
    from information_schema.columns
    where table_schema = 'eventhub'
      and table_name = 'driver'
      and column_name = 'source_driver_entity_id'
)
  and driver.source_driver_entity_id is not null
on conflict (tenant_key, event_source_id, source_driver_entity_id)
do update set
    driver_id = excluded.driver_id,
    source_updated_at = coalesce(excluded.source_updated_at, eventhub.source_driver_identity.source_updated_at),
    payload = eventhub.source_driver_identity.payload || excluded.payload,
    updated_at = now();

with legacy_driver_cards as (
    select distinct on (driver.card_nation, driver.card_number)
           driver.id as driver_id,
           driver.card_nation as nation,
           driver.card_number,
           driver.source_updated_at,
           driver.payload,
           driver.created_at,
           driver.updated_at
    from eventhub.driver driver
    where exists (
        select 1
        from information_schema.columns
        where table_schema = 'eventhub'
          and table_name = 'driver'
          and column_name = 'card_nation'
    )
      and driver.card_nation is not null
      and driver.card_number is not null
    order by driver.card_nation,
             driver.card_number,
             case when driver.source_driver_entity_id is null then 1 else 0 end,
             driver.updated_at desc,
             driver.created_at desc,
             driver.id
),
existing_driver_cards as (
    select distinct on (card.nation, card.card_number)
           card.id,
           card.nation,
           card.card_number
    from eventhub.driver_card card
    order by card.nation,
             card.card_number,
             case when card.driver_id is null then 1 else 0 end,
             card.updated_at desc,
             card.created_at desc,
             card.id
),
resolved_cards as (
    select legacy.*,
           coalesce(existing.id, gen_random_uuid()) as driver_card_id
    from legacy_driver_cards legacy
    left join existing_driver_cards existing
      on existing.nation = legacy.nation
     and existing.card_number = legacy.card_number
),
inserted_cards as (
    insert into eventhub.driver_card(
        id, driver_id, nation, card_number, source_updated_at, payload, created_at, updated_at
    )
    select distinct on (resolved.driver_card_id)
           resolved.driver_card_id,
           resolved.driver_id,
           resolved.nation,
           resolved.card_number,
           resolved.source_updated_at,
           jsonb_build_object('migrated_from', 'eventhub.driver') || coalesce(resolved.payload, '{}'::jsonb),
           resolved.created_at,
           resolved.updated_at
    from resolved_cards resolved
    where not exists (
        select 1
        from eventhub.driver_card existing
        where existing.id = resolved.driver_card_id
    )
    returning id
),
updated_cards as (
    update eventhub.driver_card card
       set driver_id = coalesce(card.driver_id, resolved.driver_id),
           source_updated_at = coalesce(resolved.source_updated_at, card.source_updated_at),
           payload = card.payload || jsonb_build_object('migrated_from', 'eventhub.driver') || coalesce(resolved.payload, '{}'::jsonb),
           updated_at = now()
      from resolved_cards resolved
     where card.id = resolved.driver_card_id
    returning card.id
)
select count(*) from inserted_cards
union all
select count(*) from updated_cards;

with single_card_per_driver as (
    select driver_id,
           min(id::text)::uuid as driver_card_id
    from eventhub.driver_card
    where driver_id is not null
    group by driver_id
    having count(*) = 1
)
update eventhub.event event
set driver_card_id = card.driver_card_id
from single_card_per_driver card
where event.driver_id = card.driver_id
  and event.driver_card_id is null;

with ranked_driver_cards as (
    select card.id,
           card.driver_id,
           card.source_updated_at,
           first_value(card.id) over (
               partition by card.nation, card.card_number
               order by case when card.driver_id is null then 1 else 0 end,
                        card.updated_at desc,
                        card.created_at desc,
                        card.id
           ) as canonical_id,
           row_number() over (
               partition by card.nation, card.card_number
               order by case when card.driver_id is null then 1 else 0 end,
                        card.updated_at desc,
                        card.created_at desc,
                        card.id
           ) as duplicate_rank
    from eventhub.driver_card card
),
duplicate_driver_cards as (
    select id, canonical_id, driver_id, source_updated_at
    from ranked_driver_cards
    where duplicate_rank > 1
),
duplicate_card_rollup as (
    select duplicate.canonical_id,
           (array_agg(duplicate.driver_id order by case when duplicate.driver_id is null then 1 else 0 end, duplicate.id))[1] as preferred_driver_id,
           max(duplicate.source_updated_at) as latest_source_updated_at
    from duplicate_driver_cards duplicate
    group by duplicate.canonical_id
),
updated_canonical_cards as (
    update eventhub.driver_card card
       set driver_id = coalesce(card.driver_id, rollup.preferred_driver_id),
           source_updated_at = case
               when card.source_updated_at is null then rollup.latest_source_updated_at
               when rollup.latest_source_updated_at is null then card.source_updated_at
               else greatest(card.source_updated_at, rollup.latest_source_updated_at)
           end,
           updated_at = now()
      from duplicate_card_rollup rollup
     where card.id = rollup.canonical_id
    returning card.id
),
relinked_source_driver_card_identity as (
    update eventhub.source_driver_card_identity identity
       set driver_card_id = duplicate.canonical_id,
           updated_at = now()
      from duplicate_driver_cards duplicate
     where identity.driver_card_id = duplicate.id
    returning identity.id
),
relinked_events as (
    update eventhub.event event
       set driver_card_id = duplicate.canonical_id
      from duplicate_driver_cards duplicate
     where event.driver_card_id = duplicate.id
    returning event.id
),
deleted_duplicate_driver_cards as (
    delete from eventhub.driver_card card
    using duplicate_driver_cards duplicate
    where card.id = duplicate.id
    returning card.id
)
select (select count(*) from updated_canonical_cards) as updated_canonical_cards,
       (select count(*) from relinked_source_driver_card_identity) as relinked_source_driver_card_identity,
       (select count(*) from relinked_events) as relinked_events,
       (select count(*) from deleted_duplicate_driver_cards) as deleted_duplicate_driver_cards;

insert into eventhub.source_vehicle_identity(
    id, tenant_key, event_source_id, source_vehicle_entity_id,
    vehicle_id, source_updated_at, payload, created_at, updated_at
)
select gen_random_uuid(),
       vehicle.tenant_key,
       vehicle.event_source_id,
       vehicle.source_vehicle_entity_id,
       vehicle.id,
       null,
       '{}'::jsonb,
       coalesce(vehicle.created_at, now()),
       coalesce(vehicle.updated_at, now())
from eventhub.vehicle vehicle
where exists (
    select 1
    from information_schema.columns
    where table_schema = 'eventhub'
      and table_name = 'vehicle'
      and column_name = 'source_vehicle_entity_id'
)
  and vehicle.source_vehicle_entity_id is not null
on conflict (tenant_key, event_source_id, source_vehicle_entity_id)
do update set
    vehicle_id = excluded.vehicle_id,
    updated_at = now();

insert into eventhub.source_vehicle_registration_identity(
    id, tenant_key, event_source_id, source_registration_entity_id,
    vehicle_registration_id, source_updated_at, payload, created_at, updated_at
)
select gen_random_uuid(),
       registration.tenant_key,
       registration.event_source_id,
       registration.source_registration_entity_id,
       registration.id,
       registration.source_updated_at,
       registration.payload,
       coalesce(registration.created_at, now()),
       coalesce(registration.updated_at, now())
from eventhub.vehicle_registration registration
where exists (
    select 1
    from information_schema.columns
    where table_schema = 'eventhub'
      and table_name = 'vehicle_registration'
      and column_name = 'source_registration_entity_id'
)
  and registration.source_registration_entity_id is not null
on conflict (tenant_key, event_source_id, source_registration_entity_id)
do update set
    vehicle_registration_id = excluded.vehicle_registration_id,
    source_updated_at = coalesce(excluded.source_updated_at, eventhub.source_vehicle_registration_identity.source_updated_at),
    payload = eventhub.source_vehicle_registration_identity.payload || excluded.payload,
    updated_at = now();

create index if not exists idx_driver_card_key
    on eventhub.driver_card(nation, card_number);

create index if not exists idx_driver_card_driver
    on eventhub.driver_card(driver_id)
    where driver_id is not null;

create unique index if not exists ux_driver_card_key
    on eventhub.driver_card(nation, card_number);

create unique index if not exists ux_vehicle_vin
    on eventhub.vehicle(vin)
    where vin is not null;

create unique index if not exists ux_vehicle_registration_plate
    on eventhub.vehicle_registration(nation, registration_number);

create index if not exists idx_source_driver_identity_driver
    on eventhub.source_driver_identity(driver_id);

create index if not exists idx_source_driver_card_identity_card
    on eventhub.source_driver_card_identity(driver_card_id);

create index if not exists idx_source_vehicle_identity_vehicle
    on eventhub.source_vehicle_identity(vehicle_id);

create index if not exists idx_source_vehicle_registration_identity_registration
    on eventhub.source_vehicle_registration_identity(vehicle_registration_id);

create index if not exists idx_event_driver_card_time
    on eventhub.event(driver_card_id, occurred_at desc)
    where driver_card_id is not null;

alter table eventhub.event
    drop constraint if exists chk_event_driver_or_vehicle_ref;

alter table eventhub.event
    add constraint chk_event_driver_or_vehicle_ref
        check (
            driver_id is not null
            or driver_card_id is not null
            or driver_entity_id is not null
            or vehicle_id is not null
            or vehicle_registration_id is not null
        );
