create table if not exists public.sun_chat_threads (
  id uuid primary key default gen_random_uuid(),
  workspace_id uuid not null references public.sun_workspaces(id) on delete cascade,
  kind text not null check (kind in ('company','direct','order')),
  title text,
  order_id text,
  direct_key text,
  created_by uuid references auth.users(id) on delete set null,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now(),
  last_message_at timestamptz
);
create unique index if not exists sun_chat_threads_company_uq on public.sun_chat_threads(workspace_id) where kind='company';
create unique index if not exists sun_chat_threads_order_uq on public.sun_chat_threads(workspace_id,order_id) where kind='order';
create unique index if not exists sun_chat_threads_direct_uq on public.sun_chat_threads(workspace_id,direct_key) where kind='direct';
create index if not exists sun_chat_threads_workspace_idx on public.sun_chat_threads(workspace_id,coalesce(last_message_at,created_at) desc);

create table if not exists public.sun_chat_participants (
  thread_id uuid not null references public.sun_chat_threads(id) on delete cascade,
  workspace_id uuid not null references public.sun_workspaces(id) on delete cascade,
  user_id uuid not null references auth.users(id) on delete cascade,
  created_at timestamptz not null default now(),
  primary key(thread_id,user_id)
);
create index if not exists sun_chat_participants_user_idx on public.sun_chat_participants(user_id,workspace_id);

create table if not exists public.sun_chat_messages (
  id uuid primary key default gen_random_uuid(),
  workspace_id uuid not null references public.sun_workspaces(id) on delete cascade,
  thread_id uuid not null references public.sun_chat_threads(id) on delete cascade,
  sender_user_id uuid not null references auth.users(id) on delete cascade,
  body text not null default '',
  attachments jsonb not null default '[]'::jsonb,
  created_at timestamptz not null default now(),
  edited_at timestamptz,
  deleted_at timestamptz
);
create index if not exists sun_chat_messages_thread_idx on public.sun_chat_messages(thread_id,created_at desc);
create index if not exists sun_chat_messages_workspace_idx on public.sun_chat_messages(workspace_id,created_at desc);

create table if not exists public.sun_chat_reads (
  thread_id uuid not null references public.sun_chat_threads(id) on delete cascade,
  workspace_id uuid not null references public.sun_workspaces(id) on delete cascade,
  user_id uuid not null references auth.users(id) on delete cascade,
  last_read_at timestamptz not null default now(),
  updated_at timestamptz not null default now(),
  primary key(thread_id,user_id)
);
create index if not exists sun_chat_reads_user_idx on public.sun_chat_reads(user_id,workspace_id);

alter table public.sun_chat_threads enable row level security;
alter table public.sun_chat_participants enable row level security;
alter table public.sun_chat_messages enable row level security;
alter table public.sun_chat_reads enable row level security;

create or replace function public.sun_chat_is_member_v29(p_workspace uuid,p_user uuid default null)
returns boolean
language sql
stable
security definer
set search_path='public','auth'
as $$
  select exists(
    select 1 from public.sun_workspace_members m
    where m.workspace_id=p_workspace
      and m.user_id=coalesce(p_user,auth.uid())
      and m.is_active=true
  )
$$;

create or replace function public.sun_chat_can_access_thread_as_v29(p_thread uuid,p_user uuid)
returns boolean
language plpgsql
stable
security definer
set search_path='public','auth'
as $$
declare v public.sun_chat_threads%rowtype;
begin
  if p_user is null then return false; end if;
  select * into v from public.sun_chat_threads where id=p_thread;
  if not found then return false; end if;
  if not public.sun_chat_is_member_v29(v.workspace_id,p_user) then return false; end if;
  if v.kind='direct' then
    return exists(select 1 from public.sun_chat_participants p where p.thread_id=v.id and p.user_id=p_user);
  end if;
  return true;
end;
$$;

create or replace function public.sun_chat_can_access_thread_v29(p_thread uuid)
returns boolean
language sql
stable
security definer
set search_path='public','auth'
as $$ select public.sun_chat_can_access_thread_as_v29(p_thread,auth.uid()) $$;

create or replace function public.sun_chat_realtime_topic_access_v29(p_topic text)
returns boolean
language plpgsql
stable
security definer
set search_path='public','auth'
as $$
declare v_id uuid;
begin
  if p_topic like 'sun-chat-workspace:%' then
    begin v_id:=substring(p_topic from length('sun-chat-workspace:')+1)::uuid; exception when others then return false; end;
    return public.sun_chat_is_member_v29(v_id,auth.uid());
  elsif p_topic like 'sun-chat-thread:%' then
    begin v_id:=substring(p_topic from length('sun-chat-thread:')+1)::uuid; exception when others then return false; end;
    return public.sun_chat_can_access_thread_v29(v_id);
  end if;
  return false;
end;
$$;

create or replace function public.sun_chat_storage_access_v29(p_name text)
returns boolean
language plpgsql
stable
security definer
set search_path='public','auth','storage'
as $$
declare parts text[]; v_ws uuid; v_thread uuid; v_thread_ws uuid;
begin
  parts:=storage.foldername(p_name);
  if array_length(parts,1)<2 then return false; end if;
  begin v_ws:=parts[1]::uuid; v_thread:=parts[2]::uuid; exception when others then return false; end;
  select workspace_id into v_thread_ws from public.sun_chat_threads where id=v_thread;
  if v_thread_ws is null or v_thread_ws<>v_ws then return false; end if;
  return public.sun_chat_can_access_thread_as_v29(v_thread,auth.uid());
end;
$$;

-- Read-only table access for authenticated users; writes go through RPCs.
revoke all on public.sun_chat_threads,public.sun_chat_participants,public.sun_chat_messages,public.sun_chat_reads from anon,authenticated;
grant select on public.sun_chat_threads,public.sun_chat_participants,public.sun_chat_messages,public.sun_chat_reads to authenticated;

-- RLS read policies.
drop policy if exists sun_chat_threads_read_v29 on public.sun_chat_threads;
create policy sun_chat_threads_read_v29 on public.sun_chat_threads for select to authenticated using (public.sun_chat_can_access_thread_v29(id));
drop policy if exists sun_chat_participants_read_v29 on public.sun_chat_participants;
create policy sun_chat_participants_read_v29 on public.sun_chat_participants for select to authenticated using (public.sun_chat_can_access_thread_v29(thread_id));
drop policy if exists sun_chat_messages_read_v29 on public.sun_chat_messages;
create policy sun_chat_messages_read_v29 on public.sun_chat_messages for select to authenticated using (public.sun_chat_can_access_thread_v29(thread_id));
drop policy if exists sun_chat_reads_read_v29 on public.sun_chat_reads;
create policy sun_chat_reads_read_v29 on public.sun_chat_reads for select to authenticated using (user_id=auth.uid() and public.sun_chat_can_access_thread_v29(thread_id));

create or replace function public.sun_chat_get_company_thread_v29(p_workspace uuid)
returns uuid
language plpgsql
security definer
set search_path='public','auth'
as $$
declare v_id uuid;
begin
  if not public.sun_chat_is_member_v29(p_workspace,auth.uid()) then raise exception 'Нет доступа к чату компании'; end if;
  select id into v_id from public.sun_chat_threads where workspace_id=p_workspace and kind='company' limit 1;
  if v_id is null then
    insert into public.sun_chat_threads(workspace_id,kind,title,created_by) values(p_workspace,'company','Общий чат',auth.uid())
    on conflict (workspace_id) where kind='company' do update set updated_at=excluded.updated_at returning id into v_id;
  end if;
  insert into public.sun_chat_reads(thread_id,workspace_id,user_id,last_read_at,updated_at)
  values(v_id,p_workspace,auth.uid(),'epoch'::timestamptz,now()) on conflict(thread_id,user_id) do nothing;
  return v_id;
end;
$$;

create or replace function public.sun_chat_get_order_thread_v29(p_workspace uuid,p_order_id text)
returns uuid
language plpgsql
security definer
set search_path='public','auth'
as $$
declare v_id uuid; v_order text:=trim(coalesce(p_order_id,''));
begin
  if not public.sun_chat_is_member_v29(p_workspace,auth.uid()) then raise exception 'Нет доступа к обсуждениям заказов'; end if;
  if v_order='' then raise exception 'Номер заказа не указан'; end if;
  select id into v_id from public.sun_chat_threads where workspace_id=p_workspace and kind='order' and order_id=v_order limit 1;
  if v_id is null then
    insert into public.sun_chat_threads(workspace_id,kind,title,order_id,created_by)
    values(p_workspace,'order','Заказ № '||v_order,v_order,auth.uid())
    on conflict (workspace_id,order_id) where kind='order' do update set updated_at=excluded.updated_at returning id into v_id;
  end if;
  insert into public.sun_chat_reads(thread_id,workspace_id,user_id,last_read_at,updated_at)
  values(v_id,p_workspace,auth.uid(),'epoch'::timestamptz,now()) on conflict(thread_id,user_id) do nothing;
  return v_id;
end;
$$;

create or replace function public.sun_chat_get_direct_thread_v29(p_workspace uuid,p_other_user uuid)
returns uuid
language plpgsql
security definer
set search_path='public','auth'
as $$
declare v_me uuid:=auth.uid(); v_key text; v_id uuid;
begin
  if v_me is null or not public.sun_chat_is_member_v29(p_workspace,v_me) then raise exception 'Нет доступа к чату компании'; end if;
  if p_other_user is null or p_other_user=v_me then raise exception 'Выберите другого сотрудника'; end if;
  if not public.sun_chat_is_member_v29(p_workspace,p_other_user) then raise exception 'Сотрудник больше не состоит в компании'; end if;
  v_key:=least(v_me::text,p_other_user::text)||':'||greatest(v_me::text,p_other_user::text);
  select id into v_id from public.sun_chat_threads where workspace_id=p_workspace and kind='direct' and direct_key=v_key limit 1;
  if v_id is null then
    insert into public.sun_chat_threads(workspace_id,kind,direct_key,created_by)
    values(p_workspace,'direct',v_key,v_me)
    on conflict (workspace_id,direct_key) where kind='direct' do update set updated_at=excluded.updated_at returning id into v_id;
  end if;
  insert into public.sun_chat_participants(thread_id,workspace_id,user_id) values(v_id,p_workspace,v_me) on conflict do nothing;
  insert into public.sun_chat_participants(thread_id,workspace_id,user_id) values(v_id,p_workspace,p_other_user) on conflict do nothing;
  insert into public.sun_chat_reads(thread_id,workspace_id,user_id,last_read_at,updated_at)
  values(v_id,p_workspace,v_me,'epoch'::timestamptz,now()) on conflict(thread_id,user_id) do nothing;
  return v_id;
end;
$$;

create or replace function public.sun_chat_list_members_v29(p_workspace uuid)
returns table(user_id uuid,display_name text,email text,role text)
language plpgsql
stable
security definer
set search_path='public','auth'
as $$
begin
  if not public.sun_chat_is_member_v29(p_workspace,auth.uid()) then raise exception 'Нет доступа к сотрудникам компании'; end if;
  return query select m.user_id,coalesce(nullif(m.display_name,''),split_part(coalesce(u.email,''),'@',1)),u.email::text,m.role
  from public.sun_workspace_members m join auth.users u on u.id=m.user_id
  where m.workspace_id=p_workspace and m.is_active=true
  order by (m.user_id=auth.uid()) desc,coalesce(nullif(m.display_name,''),u.email::text);
end;
$$;

create or replace function public.sun_chat_list_threads_v29(p_workspace uuid)
returns table(thread_id uuid,kind text,order_id text,title text,other_user_id uuid,last_message text,last_message_at timestamptz,unread_count bigint)
language plpgsql
stable
security definer
set search_path='public','auth'
as $$
begin
  if not public.sun_chat_is_member_v29(p_workspace,auth.uid()) then raise exception 'Нет доступа к чатам компании'; end if;
  return query
  select t.id,t.kind,t.order_id,
    case when t.kind='direct' then coalesce(nullif(om.display_name,''),split_part(coalesce(ou.email,''),'@',1),'Сотрудник') else coalesce(t.title,case when t.kind='company' then 'Общий чат' else 'Обсуждение' end) end,
    case when t.kind='direct' then op.user_id else null end,
    case when lm.id is null then '' when nullif(trim(lm.body),'') is not null then left(lm.body,110) when jsonb_array_length(coalesce(lm.attachments,'[]'::jsonb))>0 then 'Вложение' else '' end,
    lm.created_at,
    (select count(*) from public.sun_chat_messages um where um.thread_id=t.id and um.deleted_at is null and um.sender_user_id<>auth.uid() and um.created_at>coalesce(r.last_read_at,'epoch'::timestamptz))
  from public.sun_chat_threads t
  left join public.sun_chat_reads r on r.thread_id=t.id and r.user_id=auth.uid()
  left join lateral (select m.* from public.sun_chat_messages m where m.thread_id=t.id and m.deleted_at is null order by m.created_at desc limit 1) lm on true
  left join lateral (select p.user_id from public.sun_chat_participants p where p.thread_id=t.id and p.user_id<>auth.uid() limit 1) op on t.kind='direct'
  left join public.sun_workspace_members om on om.workspace_id=t.workspace_id and om.user_id=op.user_id
  left join auth.users ou on ou.id=op.user_id
  where t.workspace_id=p_workspace and public.sun_chat_can_access_thread_as_v29(t.id,auth.uid())
  order by case t.kind when 'company' then 0 when 'direct' then 1 else 2 end,coalesce(t.last_message_at,t.created_at) desc;
end;
$$;

create or replace function public.sun_chat_list_messages_v29(p_thread uuid,p_limit integer default 100,p_before timestamptz default null)
returns table(message_id uuid,sender_user_id uuid,sender_name text,sender_email text,body text,attachments jsonb,created_at timestamptz,read_by_count bigint)
language plpgsql
stable
security definer
set search_path='public','auth'
as $$
begin
  if not public.sun_chat_can_access_thread_as_v29(p_thread,auth.uid()) then raise exception 'Нет доступа к переписке'; end if;
  return query
  with recent as (
    select m.* from public.sun_chat_messages m
    where m.thread_id=p_thread and m.deleted_at is null and (p_before is null or m.created_at<p_before)
    order by m.created_at desc limit greatest(1,least(coalesce(p_limit,100),200))
  )
  select m.id,m.sender_user_id,coalesce(nullif(sm.display_name,''),split_part(coalesce(su.email,''),'@',1),'Сотрудник'),su.email::text,m.body,m.attachments,m.created_at,
    (select count(*) from public.sun_chat_reads rr where rr.thread_id=m.thread_id and rr.user_id<>m.sender_user_id and rr.last_read_at>=m.created_at)
  from recent m
  left join public.sun_workspace_members sm on sm.workspace_id=m.workspace_id and sm.user_id=m.sender_user_id
  left join auth.users su on su.id=m.sender_user_id
  order by m.created_at asc;
end;
$$;

create or replace function public.sun_chat_send_message_v29(p_thread uuid,p_body text default '',p_attachments jsonb default '[]'::jsonb)
returns uuid
language plpgsql
security definer
set search_path='public','auth','realtime'
as $$
declare v public.sun_chat_threads%rowtype; v_id uuid; v_body text:=trim(coalesce(p_body,'')); v_att jsonb:=coalesce(p_attachments,'[]'::jsonb);
begin
  if not public.sun_chat_can_access_thread_as_v29(p_thread,auth.uid()) then raise exception 'Нет доступа к переписке'; end if;
  select * into v from public.sun_chat_threads where id=p_thread;
  if length(v_body)>4000 then raise exception 'Сообщение слишком длинное'; end if;
  if jsonb_typeof(v_att)<>'array' then raise exception 'Некорректные вложения'; end if;
  if jsonb_array_length(v_att)>5 then raise exception 'Можно отправить не более 5 файлов за раз'; end if;
  if v_body='' and jsonb_array_length(v_att)=0 then raise exception 'Введите сообщение или добавьте файл'; end if;
  insert into public.sun_chat_messages(workspace_id,thread_id,sender_user_id,body,attachments)
  values(v.workspace_id,v.id,auth.uid(),v_body,v_att) returning id into v_id;
  update public.sun_chat_threads set last_message_at=now(),updated_at=now() where id=v.id;
  insert into public.sun_chat_reads(thread_id,workspace_id,user_id,last_read_at,updated_at)
  values(v.id,v.workspace_id,auth.uid(),now(),now())
  on conflict(thread_id,user_id) do update set last_read_at=excluded.last_read_at,updated_at=now();
  perform realtime.send('{}'::jsonb,'chat_changed','sun-chat-workspace:'||v.workspace_id::text,true);
  perform realtime.send(jsonb_build_object('message_id',v_id),'message','sun-chat-thread:'||v.id::text,true);
  return v_id;
end;
$$;

create or replace function public.sun_chat_mark_read_v29(p_thread uuid)
returns void
language plpgsql
security definer
set search_path='public','auth','realtime'
as $$
declare v_ws uuid; v_old timestamptz; v_changed boolean:=false;
begin
  if not public.sun_chat_can_access_thread_as_v29(p_thread,auth.uid()) then raise exception 'Нет доступа к переписке'; end if;
  select workspace_id into v_ws from public.sun_chat_threads where id=p_thread;
  select last_read_at into v_old from public.sun_chat_reads where thread_id=p_thread and user_id=auth.uid();
  select exists(
    select 1 from public.sun_chat_messages m
    where m.thread_id=p_thread and m.deleted_at is null and m.sender_user_id<>auth.uid()
      and m.created_at>coalesce(v_old,'epoch'::timestamptz)
  ) into v_changed;
  insert into public.sun_chat_reads(thread_id,workspace_id,user_id,last_read_at,updated_at)
  values(p_thread,v_ws,auth.uid(),now(),now())
  on conflict(thread_id,user_id) do update set last_read_at=excluded.last_read_at,updated_at=now();
  if v_changed then
    perform realtime.send(jsonb_build_object('thread_id',p_thread),'read','sun-chat-thread:'||p_thread::text,true);
    perform realtime.send('{}'::jsonb,'chat_changed','sun-chat-workspace:'||v_ws::text,true);
  end if;
end;
$$;

create or replace function public.sun_chat_unread_total_v29(p_workspace uuid)
returns bigint
language plpgsql
stable
security definer
set search_path='public','auth'
as $$
declare v_total bigint;
begin
  if not public.sun_chat_is_member_v29(p_workspace,auth.uid()) then return 0; end if;
  select count(*) into v_total
  from public.sun_chat_messages m
  join public.sun_chat_threads t on t.id=m.thread_id
  left join public.sun_chat_reads r on r.thread_id=t.id and r.user_id=auth.uid()
  where t.workspace_id=p_workspace and m.deleted_at is null and m.sender_user_id<>auth.uid()
    and public.sun_chat_can_access_thread_as_v29(t.id,auth.uid())
    and m.created_at>coalesce(r.last_read_at,'epoch'::timestamptz);
  return coalesce(v_total,0);
end;
$$;

-- Private Storage bucket for chat files.
insert into storage.buckets(id,name,public,file_size_limit)
values('sun-chat','sun-chat',false,15728640)
on conflict(id) do update set public=false,file_size_limit=15728640;

drop policy if exists sun_chat_storage_read_v29 on storage.objects;
create policy sun_chat_storage_read_v29 on storage.objects for select to authenticated
using(bucket_id='sun-chat' and public.sun_chat_storage_access_v29(name));
drop policy if exists sun_chat_storage_insert_v29 on storage.objects;
create policy sun_chat_storage_insert_v29 on storage.objects for insert to authenticated
with check(bucket_id='sun-chat' and public.sun_chat_storage_access_v29(name));

-- Realtime private-channel authorization for chat workspace/thread topics.
drop policy if exists sun_chat_realtime_read_v29 on realtime.messages;
create policy sun_chat_realtime_read_v29 on realtime.messages for select to authenticated
using(realtime.messages.extension in ('broadcast','presence') and public.sun_chat_realtime_topic_access_v29((select realtime.topic())));
drop policy if exists sun_chat_realtime_write_v29 on realtime.messages;
create policy sun_chat_realtime_write_v29 on realtime.messages for insert to authenticated
with check(realtime.messages.extension in ('broadcast','presence') and public.sun_chat_realtime_topic_access_v29((select realtime.topic())));

revoke all on function public.sun_chat_is_member_v29(uuid,uuid) from public,anon;
revoke all on function public.sun_chat_can_access_thread_as_v29(uuid,uuid) from public,anon;
revoke all on function public.sun_chat_can_access_thread_v29(uuid) from public,anon;
revoke all on function public.sun_chat_realtime_topic_access_v29(text) from public,anon;
revoke all on function public.sun_chat_storage_access_v29(text) from public,anon;
revoke all on function public.sun_chat_get_company_thread_v29(uuid) from public,anon;
revoke all on function public.sun_chat_get_order_thread_v29(uuid,text) from public,anon;
revoke all on function public.sun_chat_get_direct_thread_v29(uuid,uuid) from public,anon;
revoke all on function public.sun_chat_list_members_v29(uuid) from public,anon;
revoke all on function public.sun_chat_list_threads_v29(uuid) from public,anon;
revoke all on function public.sun_chat_list_messages_v29(uuid,integer,timestamptz) from public,anon;
revoke all on function public.sun_chat_send_message_v29(uuid,text,jsonb) from public,anon;
revoke all on function public.sun_chat_mark_read_v29(uuid) from public,anon;
revoke all on function public.sun_chat_unread_total_v29(uuid) from public,anon;

grant execute on function public.sun_chat_realtime_topic_access_v29(text) to authenticated;
grant execute on function public.sun_chat_storage_access_v29(text) to authenticated;
grant execute on function public.sun_chat_get_company_thread_v29(uuid) to authenticated;
grant execute on function public.sun_chat_get_order_thread_v29(uuid,text) to authenticated;
grant execute on function public.sun_chat_get_direct_thread_v29(uuid,uuid) to authenticated;
grant execute on function public.sun_chat_list_members_v29(uuid) to authenticated;
grant execute on function public.sun_chat_list_threads_v29(uuid) to authenticated;
grant execute on function public.sun_chat_list_messages_v29(uuid,integer,timestamptz) to authenticated;
grant execute on function public.sun_chat_send_message_v29(uuid,text,jsonb) to authenticated;
grant execute on function public.sun_chat_mark_read_v29(uuid) to authenticated;
grant execute on function public.sun_chat_unread_total_v29(uuid) to authenticated;