-- =========================================
-- 1) MOTIE hist에 open/close run 귀속 컬럼 추가
-- =========================================
alter table if exists public.motie_org_hist
  add column if not exists open_run_id  uuid null,
  add column if not exists close_run_id uuid null;

create index if not exists idx_motie_org_hist_open_run  on public.motie_org_hist (open_run_id);
create index if not exists idx_motie_org_hist_close_run on public.motie_org_hist (close_run_id);

-- MOEF 도 동일 정책이면 주석 해제
-- alter table if exists public.moef_org_hist
--   add column if not exists open_run_id  uuid null,
--   add column if not exists close_run_id uuid null;
-- create index if not exists idx_moef_org_hist_open_run  on public.moef_org_hist (open_run_id);
-- create index if not exists idx_moef_org_hist_close_run on public.moef_org_hist (close_run_id);

-- =========================================
-- 2) motie_org_raw의 run 타임윈도우로 open/close_run_id 백필
--    (LATERAL 제거, 상관 서브쿼리로 재작성)
--    ※ motie_org_raw.run_id 가 uuid 라는 가정. (text면 ::uuid 캐스팅 조정)
-- =========================================

-- Open 귀속: valid_from 이 포함되는 run_id를 가장 근접하게 매핑
with runs as (
  select
    r.run_id,                     -- uuid
    min(r.fetched_at) as start_at,
    max(r.fetched_at) as end_at
  from public.motie_org_raw r
  group by r.run_id
)
update public.motie_org_hist h
set open_run_id = (
  select rr.run_id
  from runs rr
  where h.valid_from between rr.start_at - interval '2 hours'
                        and     rr.end_at   + interval '2 hours'
  order by abs(extract(epoch from (h.valid_from - rr.end_at))) asc
  limit 1
)
where h.open_run_id is null;

-- Close 귀속: valid_to 가 포함되는 run_id를 가장 근접하게 매핑
with runs as (
  select
    r.run_id,
    min(r.fetched_at) as start_at,
    max(r.fetched_at) as end_at
  from public.motie_org_raw r
  group by r.run_id
)
update public.motie_org_hist h
set close_run_id = (
  select rr.run_id
  from runs rr
  where h.valid_to is not null
    and h.valid_to between rr.start_at - interval '2 hours'
                      and     rr.end_at   + interval '2 hours'
  order by abs(extract(epoch from (h.valid_to - rr.end_at))) asc
  limit 1
)
where h.close_run_id is null
  and h.valid_to is not null;

-- =========================================
-- 3) 동일 run 내 close+add를 한 건의 'moved'로 접합하는 이벤트 뷰
--    컬럼 셋(16개): key_hash, name, department, "position", phone, task,
--                   valid_from, valid_to, changed_fields, event_type, run_id,
--                   dep_before, pos_before, phone_before, same_run, pair_delta_sec
-- =========================================
create or replace view public.motie_org_events_v3 as
with base as (
  select
    h.key_hash,
    h.name,
    h.department,
    h."position",
    h.phone,
    h.task,
    h.valid_from,
    h.valid_to,
    h.open_run_id,
    h.close_run_id
  from public.motie_org_hist h
),
closed as (
  select * from base where valid_to is not null
),
added as (
  select * from base where valid_to is null
),
pairs as (
  select
    c.key_hash   as key_hash_before,
    a.key_hash   as key_hash_after,
    c.name,
    c.department as dep_before,
    a.department as dep_after,
    c."position" as pos_before,
    a."position" as pos_after,
    c.phone      as phone_before,
    a.phone      as phone_after,
    coalesce(a.task, c.task) as task_after,
    a.valid_from as valid_from,
    c.valid_to   as valid_to,
    a.open_run_id  as run_after,
    c.close_run_id as run_before,
    (a.open_run_id is not null and a.open_run_id = c.close_run_id) as same_run,
    abs(extract(epoch from (a.valid_from - c.valid_to)))::integer  as pair_delta_sec,
    array_remove(array[
      case when a.department is distinct from c.department then 'department' end,
      case when a."position" is distinct from c."position" then 'position'   end,
      case when a.phone      is distinct from c.phone      then 'phone'      end,
      case when a.task       is distinct from c.task       then 'task'       end
    ], null) as changed_fields
  from closed c
  join lateral (
    select a.*
    from added a
    where a.name = c.name
      and (
        (a.open_run_id is not null and a.open_run_id = c.close_run_id)
        or (a.valid_from between c.valid_to - interval '3 minutes'
                           and     c.valid_to + interval '3 minutes')
      )
    order by
      (a.open_run_id is not null and a.open_run_id = c.close_run_id) desc,
      abs(extract(epoch from (a.valid_from - c.valid_to))) asc
    limit 1
  ) a on true
),
moved as (
  select
    coalesce(p.key_hash_after, p.key_hash_before) as key_hash,
    p.name,
    p.dep_after   as department,
    p.pos_after   as "position",
    p.phone_after as phone,
    p.task_after  as task,
    p.valid_from,
    p.valid_to,
    p.changed_fields,
    'updated'::text as event_type,
    p.run_after   as run_id,
    p.dep_before,
    p.pos_before,
    p.phone_before,
    p.same_run,
    p.pair_delta_sec
  from pairs p
  where coalesce(array_length(p.changed_fields, 1), 0) > 0
),
closed_only as (
  select
    c.key_hash      as key_hash,
    c.name          as name,
    c.department    as department,
    c."position"    as "position",
    c.phone         as phone,
    c.task          as task,
    c.valid_from    as valid_from,
    c.valid_to      as valid_to,
    array[]::text[] as changed_fields,
    'closed'::text  as event_type,
    c.close_run_id  as run_id,
    c.department    as dep_before,
    c."position"    as pos_before,
    c.phone         as phone_before,
    false           as same_run,
    null::integer   as pair_delta_sec
  from closed c
  where not exists (select 1 from pairs p where p.key_hash_before = c.key_hash)
),
added_only as (
  select
    a.key_hash                                      as key_hash,
    a.name                                          as name,
    a.department                                    as department,
    a."position"                                    as "position",
    a.phone                                         as phone,
    a.task                                          as task,
    a.valid_from                                    as valid_from,
    a.valid_to                                      as valid_to,
    array['department','position','phone','task']::text[] as changed_fields,
    'added'::text                                   as event_type,
    a.open_run_id                                   as run_id,
    null::text                                      as dep_before,
    null::text                                      as pos_before,
    null::text                                      as phone_before,
    false                                           as same_run,
    null::integer                                   as pair_delta_sec
  from added a
  where not exists (select 1 from pairs p where p.key_hash_after = a.key_hash)
)
select * from moved
union all
select * from closed_only
union all
select * from added_only;
