mnote/supabase/migrations/20260105214122_20260105_add_rag_index_sources.sql

-- 说明：为自动入库（Supabase -> LightRAG）提供队列表与触发器/函数。
-- 编码：UTF-8

create table public.rag_index_sources (
  id uuid primary key default gen_random_uuid(),
  workspace_id uuid not null references public.workspaces(id) on delete cascade,
  source_type text not null,
  source_id uuid not null,
  document_id uuid references public.documents(id) on delete set null,
  user_id uuid references public.profiles(id) on delete set null,
  status text not null default 'pending',
  attempts integer not null default 0,
  last_error text,
  source_updated_at timestamp with time zone,
  last_enqueued_at timestamp with time zone not null default timezone('utc', now()),
  last_processed_at timestamp with time zone,
  created_at timestamp with time zone not null default timezone('utc', now()),
  updated_at timestamp with time zone not null default timezone('utc', now()),
  constraint rag_index_sources_source_type_check check (
    source_type = any (array['document'::text, 'mindmap'::text, 'media_asset'::text])
  ),
  constraint rag_index_sources_status_check check (
    status = any (array['pending'::text, 'processing'::text, 'completed'::text, 'failed'::text, 'skipped'::text])
  )
);

create unique index rag_index_sources_unique_source
  on public.rag_index_sources (source_type, source_id);

create index rag_index_sources_status_idx
  on public.rag_index_sources (status, last_enqueued_at);

create index rag_index_sources_workspace_status_idx
  on public.rag_index_sources (workspace_id, status, last_enqueued_at desc);

alter table public.rag_index_sources enable row level security;

create policy "Select rag index within workspace"
  on public.rag_index_sources
  as permissive
  for select
  to public
  using (
    exists (
      select 1
      from public.workspace_members wm
      where wm.workspace_id = rag_index_sources.workspace_id
        and wm.user_id = uid()
    )
  );

create or replace function public.set_rag_index_sources_updated_at()
returns trigger
language plpgsql
as $function$
begin
  new.updated_at := timezone('utc', now());
  return new;
end;
$function$;

create trigger set_rag_index_sources_updated_at
before update on public.rag_index_sources
for each row execute function public.set_rag_index_sources_updated_at();

create or replace function public.upsert_rag_index_source(
  p_workspace_id uuid,
  p_source_type text,
  p_source_id uuid,
  p_document_id uuid,
  p_user_id uuid,
  p_source_updated_at timestamp with time zone
)
returns void
language plpgsql
security definer
set search_path to 'public'
as $function$
begin
  insert into public.rag_index_sources (
    workspace_id,
    source_type,
    source_id,
    document_id,
    user_id,
    status,
    source_updated_at,
    last_enqueued_at,
    last_error
  )
  values (
    p_workspace_id,
    p_source_type,
    p_source_id,
    p_document_id,
    p_user_id,
    'pending',
    p_source_updated_at,
    timezone('utc', now()),
    null
  )
  on conflict (source_type, source_id)
  do update set
    workspace_id = excluded.workspace_id,
    document_id = excluded.document_id,
    user_id = excluded.user_id,
    status = 'pending',
    source_updated_at = excluded.source_updated_at,
    last_enqueued_at = timezone('utc', now()),
    last_error = null,
    updated_at = timezone('utc', now());
end;
$function$;

create or replace function public.enqueue_rag_for_document()
returns trigger
language plpgsql
as $function$
begin
  perform public.upsert_rag_index_source(
    new.workspace_id,
    'document',
    new.id,
    new.id,
    new.user_id,
    new.updated_at
  );
  return new;
end;
$function$;

create or replace function public.enqueue_rag_for_media_asset()
returns trigger
language plpgsql
as $function$
declare
  v_user_id uuid;
begin
  select user_id
    into v_user_id
    from public.documents
   where id = new.document_id;

  perform public.upsert_rag_index_source(
    new.workspace_id,
    'media_asset',
    new.id,
    new.document_id,
    coalesce(new.created_by, v_user_id),
    new.updated_at
  );
  return new;
end;
$function$;

create or replace function public.enqueue_rag_for_mindmap_node()
returns trigger
language plpgsql
as $function$
declare
  v_user_id uuid;
begin
  select user_id
    into v_user_id
    from public.documents
   where id = new.document_id;

  perform public.upsert_rag_index_source(
    new.workspace_id,
    'mindmap',
    new.mindmap_id,
    new.document_id,
    v_user_id,
    new.updated_at
  );
  return new;
end;
$function$;

create trigger enqueue_rag_for_document_insert
after insert on public.documents
for each row execute function public.enqueue_rag_for_document();

create trigger enqueue_rag_for_document_update
after update on public.documents
for each row
when (
  old.title is distinct from new.title
  or old.content is distinct from new.content
  or old.raw_text is distinct from new.raw_text
  or old.mindmap_data is distinct from new.mindmap_data
  or old.deleted_at is distinct from new.deleted_at
)
execute function public.enqueue_rag_for_document();

create trigger enqueue_rag_for_media_asset_insert
after insert on public.media_assets
for each row execute function public.enqueue_rag_for_media_asset();

create trigger enqueue_rag_for_media_asset_update
after update on public.media_assets
for each row
when (
  old.file_url is distinct from new.file_url
  or old.storage_path is distinct from new.storage_path
  or old.bucket is distinct from new.bucket
  or old.ocr_text is distinct from new.ocr_text
  or old.ocr_status is distinct from new.ocr_status
  or old.mime_type is distinct from new.mime_type
)
execute function public.enqueue_rag_for_media_asset();

create trigger enqueue_rag_for_mindmap_node_insert
after insert on public.mindmap_nodes
for each row execute function public.enqueue_rag_for_mindmap_node();

create trigger enqueue_rag_for_mindmap_node_update
after update on public.mindmap_nodes
for each row
when (
  old.data is distinct from new.data
  or old.parent_id is distinct from new.parent_id
  or old.block_id is distinct from new.block_id
  or old.order_index is distinct from new.order_index
)
execute function public.enqueue_rag_for_mindmap_node();

create or replace function public.backfill_rag_index_sources()
returns integer
language plpgsql
security definer
set search_path to 'public'
as $function$
declare
  v_count integer := 0;
begin
  -- 仅允许 service_role 调用，避免被普通用户滥用造成全量重索引
  if auth.role() <> 'service_role' then
    raise exception 'forbidden';
  end if;

  insert into public.rag_index_sources (
    workspace_id,
    source_type,
    source_id,
    document_id,
    user_id,
    status,
    source_updated_at,
    last_enqueued_at,
    last_error
  )
  select
    d.workspace_id,
    'document',
    d.id,
    d.id,
    d.user_id,
    'pending',
    d.updated_at,
    timezone('utc', now()),
    null
  from public.documents d
  on conflict (source_type, source_id)
  do update set
    workspace_id = excluded.workspace_id,
    document_id = excluded.document_id,
    user_id = excluded.user_id,
    status = 'pending',
    source_updated_at = excluded.source_updated_at,
    last_enqueued_at = timezone('utc', now()),
    last_error = null,
    updated_at = timezone('utc', now());

  get diagnostics v_count = row_count;
  return v_count;
end;
$function$;

grant execute on function public.backfill_rag_index_sources() to service_role;