Preview
Open Original
I spent WAYYY too long trying to build a more accurate RAG retrieval system.
With Context Mesh Lite, I managed to combine hybrid vector search with SQL search (agentic text-to-sql) with graph search (shallow graph using dependent tables).
The results were a significantly more accurate (albeit slower) RAG system.
How does it work?
- SQL Functions do most of the heavy lifting, creating tables and table dependencies.
- Then Edge Functions call Gemini (embeddings 001 and 2.5 flash) to create vector embeddings and graph entity/predicate extraction.
REQUIREMENTS: This system was built to exist within a Supabase instance. It also requires a Gemini API key (set in your Edge Functions window).
I also connected the system to n8n workflows and it works like a charm. Anyway, I…
I spent WAYYY too long trying to build a more accurate RAG retrieval system.
With Context Mesh Lite, I managed to combine hybrid vector search with SQL search (agentic text-to-sql) with graph search (shallow graph using dependent tables).
The results were a significantly more accurate (albeit slower) RAG system.
How does it work?
- SQL Functions do most of the heavy lifting, creating tables and table dependencies.
- Then Edge Functions call Gemini (embeddings 001 and 2.5 flash) to create vector embeddings and graph entity/predicate extraction.
REQUIREMENTS: This system was built to exist within a Supabase instance. It also requires a Gemini API key (set in your Edge Functions window).
I also connected the system to n8n workflows and it works like a charm. Anyway, I’m gonna give it to you. Maybe it’ll be useful. Maybe you can improve on it.
So, first, go to your Supabase (the entire end-to-end system exists there...only the interface for document upsert and chat are external).
Step 1. Go to the SQL editor and paste this master query:
-- ===============================================================
-- CONTEXT MESH V9.0: GOLDEN MASTER (COMPOSITE RETRIEVAL)
-- UPDATED: Nov 25, 2025
-- ===============================================================
-- PART 1: EXTENSIONS
-- PART 2: STORAGE CONFIGURATION
-- PART 3: CORE TABLES (Docs, Graph, Queue, Config, Registry)
-- PART 4: INDEXES
-- PART 5: HELPER FUNCTIONS & TRIGGERS
-- PART 6: INGESTION FUNCTIONS (Universal V8 Logic)
-- PART 7: SEARCH FUNCTIONS (V9: FTS, Enrichment, Detection, Peek)
-- PART 8: CONFIGURATION LOGIC (V9 Weights)
-- PART 9: GRAPH CONTEXT RERANKER
-- PART 10: WORKER SETUP
-- PART 11: PERMISSIONS & REPAIRS
-- ===============================================================
-- ===============================================================
-- PART 1: EXTENSIONS
-- ===============================================================
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE EXTENSION IF NOT EXISTS pg_net; -- Required for Async Worker
-- ===============================================================
-- PART 2: STORAGE CONFIGURATION
-- ===============================================================
INSERT INTO storage.buckets (id, name, public)
VALUES ('raw_uploads', 'raw_uploads', false)
ON CONFLICT (id) DO NOTHING;
DROP POLICY IF EXISTS "Service Role Full Access" ON storage.objects;
CREATE POLICY "Service Role Full Access"
ON storage.objects FOR ALL
USING ( auth.role() = 'service_role' )
WITH CHECK ( auth.role() = 'service_role' );
-- ===============================================================
-- PART 3: CORE TABLES
-- ===============================================================
-- 3a. The Async Queue
CREATE TABLE IF NOT EXISTS public.ingestion_queue (
id uuid default gen_random_uuid() primary key,
uri text not null,
title text not null,
chunk_index int not null,
chunk_text text not null,
status text default 'pending',
error_log text,
created_at timestamptz default now()
);
-- 3b. RAG Tables
CREATE TABLE IF NOT EXISTS public.document (
id BIGSERIAL PRIMARY KEY,
uri TEXT NOT NULL UNIQUE,
title TEXT NOT NULL,
doc_type TEXT NOT NULL DEFAULT 'document',
meta JSONB NOT NULL DEFAULT '{}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS public.chunk (
id BIGSERIAL PRIMARY KEY,
document_id BIGINT NOT NULL REFERENCES public.document(id) ON DELETE CASCADE,
ordinal INT NOT NULL,
text TEXT NOT NULL,
tsv TSVECTOR,
UNIQUE (document_id, ordinal)
);
-- PERFORMANCE: Using halfvec(768) for Gemini embeddings
CREATE TABLE IF NOT EXISTS public.chunk_embedding (
chunk_id BIGINT PRIMARY KEY REFERENCES public.chunk(id) ON DELETE CASCADE,
embedding halfvec(768) NOT NULL
);
-- 3c. Graph Tables
CREATE TABLE IF NOT EXISTS public.node (
id BIGSERIAL PRIMARY KEY,
key TEXT UNIQUE NOT NULL,
labels TEXT[] NOT NULL DEFAULT '{}',
props JSONB NOT NULL DEFAULT '{}'::jsonb
);
CREATE TABLE IF NOT EXISTS public.edge (
src BIGINT NOT NULL REFERENCES public.node(id) ON DELETE CASCADE,
dst BIGINT NOT NULL REFERENCES public.node(id) ON DELETE CASCADE,
type TEXT NOT NULL,
props JSONB NOT NULL DEFAULT '{}'::jsonb,
PRIMARY KEY (src, dst, type)
);
CREATE TABLE IF NOT EXISTS public.chunk_node (
chunk_id BIGINT NOT NULL REFERENCES public.chunk(id) ON DELETE CASCADE,
node_id BIGINT NOT NULL REFERENCES public.node(id) ON DELETE CASCADE,
rel TEXT NOT NULL DEFAULT 'MENTIONS',
PRIMARY KEY (chunk_id, node_id, rel)
);
-- 3d. Structured Data Registry (V8 Updated)
CREATE TABLE IF NOT EXISTS public.structured_table (
id BIGSERIAL PRIMARY KEY,
table_name TEXT NOT NULL UNIQUE,
document_id BIGINT REFERENCES public.document(id) ON DELETE CASCADE,
schema_def JSONB NOT NULL,
row_count INT DEFAULT 0,
-- V8 Metadata Columns
description TEXT,
column_semantics JSONB DEFAULT '{}'::jsonb,
graph_hints JSONB DEFAULT '[]'::jsonb,
sample_row JSONB,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- 3e. Configuration Table
CREATE TABLE IF NOT EXISTS public.app_config (
id INT PRIMARY KEY DEFAULT 1,
settings JSONB NOT NULL,
updated_at TIMESTAMPTZ DEFAULT now(),
CONSTRAINT single_row CHECK (id = 1)
);
-- ===============================================================
-- PART 4: INDEXES
-- ===============================================================
CREATE INDEX IF NOT EXISTS idx_queue_status ON public.ingestion_queue(status);
CREATE INDEX IF NOT EXISTS document_type_idx ON public.document(doc_type);
CREATE INDEX IF NOT EXISTS document_uri_idx ON public.document(uri);
CREATE INDEX IF NOT EXISTS chunk_tsv_gin ON public.chunk USING GIN (tsv);
CREATE INDEX IF NOT EXISTS chunk_doc_idx ON public.chunk(document_id);
-- Embedding Index (HNSW)
CREATE INDEX IF NOT EXISTS emb_hnsw_cos ON public.chunk_embedding USING HNSW (embedding halfvec_cosine_ops);
-- Graph Indexes
CREATE INDEX IF NOT EXISTS edge_src_idx ON public.edge (src);
CREATE INDEX IF NOT EXISTS edge_dst_idx ON public.edge (dst);
CREATE INDEX IF NOT EXISTS node_labels_gin ON public.node USING GIN (labels);
CREATE INDEX IF NOT EXISTS node_props_gin ON public.node USING GIN (props);
CREATE INDEX IF NOT EXISTS chunknode_node_idx ON public.chunk_node (node_id);
CREATE INDEX IF NOT EXISTS chunknode_chunk_idx ON public.chunk_node (chunk_id);
-- Registry Index
CREATE INDEX IF NOT EXISTS idx_structured_table_active ON public.structured_table(table_name) WHERE row_count > 0;
-- ===============================================================
-- PART 5: HELPER FUNCTIONS & TRIGGERS
-- ===============================================================
-- 5a. Full Text Search Update Trigger
CREATE OR REPLACE FUNCTION public.chunk_tsv_update()
RETURNS trigger LANGUAGE plpgsql AS $$
DECLARE doc_title text;
BEGIN
SELECT d.title INTO doc_title FROM public.document d WHERE d.id = NEW.document_id;
NEW.tsv :=
setweight(to_tsvector('english', coalesce(doc_title, '')), 'D') ||
setweight(to_tsvector('english', coalesce(NEW.text, '')), 'A');
RETURN NEW;
END $$;
DROP TRIGGER IF EXISTS chunk_tsv_trg ON public.chunk;
CREATE TRIGGER chunk_tsv_trg
BEFORE INSERT OR UPDATE OF text, document_id ON public.chunk
FOR EACH ROW EXECUTE FUNCTION public.chunk_tsv_update();
-- 5b. Sanitize Table Names
CREATE OR REPLACE FUNCTION public.sanitize_table_name(name TEXT)
RETURNS TEXT LANGUAGE sql IMMUTABLE AS $$
SELECT 'tbl_' || regexp_replace(lower(trim(name)), '[^a-z0-9_]', '_', 'g');
$$;
-- 5c. Data Extraction Helpers
CREATE OR REPLACE FUNCTION public.extract_numeric(text TEXT, key TEXT)
RETURNS NUMERIC LANGUAGE sql IMMUTABLE AS $$
SELECT (regexp_match(text, key || '\s*:\s*\$?([0-9,]+\.?[0-9]*)', 'i'))[1]::text::numeric;
$$;
-- 5d. Polymorphic Date Extraction (Supports Excel, ISO, US formats)
-- 1. Text
CREATE OR REPLACE FUNCTION public.extract_date(text TEXT)
RETURNS DATE LANGUAGE plpgsql IMMUTABLE AS $$
BEGIN
IF text ~ '^\d{5}$' THEN RETURN '1899-12-30'::date + (text::int); END IF;
IF text ~ '\d{4}-\d{2}-\d{2}' THEN RETURN (regexp_match(text, '(\d{4}-\d{2}-\d{2})'))[1]::date; END IF;
IF text ~ '\d{1,2}/\d{1,2}/\d{4}' THEN RETURN to_date((regexp_match(text, '(\d{1,2}/\d{1,2}/\d{4})'))[1], 'MM/DD/YYYY'); END IF;
RETURN NULL;
EXCEPTION WHEN OTHERS THEN RETURN NULL;
END $$;
-- 2. Numeric (Excel Serial)
CREATE OR REPLACE FUNCTION public.extract_date(val NUMERIC)
RETURNS DATE LANGUAGE plpgsql IMMUTABLE AS $$
BEGIN RETURN '1899-12-30'::date + (val::int); EXCEPTION WHEN OTHERS THEN RETURN NULL; END $$;
-- 3. Integer
CREATE OR REPLACE FUNCTION public.extract_date(val INTEGER)
RETURNS DATE LANGUAGE plpgsql IMMUTABLE AS $$
BEGIN RETURN '1899-12-30'::date + val; EXCEPTION WHEN OTHERS THEN RETURN NULL; END $$;
-- 4. Date (Pass-through)
CREATE OR REPLACE FUNCTION public.extract_date(val DATE)
RETURNS DATE LANGUAGE sql IMMUTABLE AS $$ SELECT val; $$;
CREATE OR REPLACE FUNCTION public.extract_keywords(p_text TEXT)
RETURNS TEXT[] LANGUAGE sql IMMUTABLE AS $$
SELECT array_agg(DISTINCT word) FROM (
SELECT unnest(tsvector_to_array(to_tsvector('english', p_text))) AS word
) words WHERE length(word) > 2 LIMIT 10;
$$;
-- 5e. Column Type Inference
CREATE OR REPLACE FUNCTION public.infer_column_type(sample_values TEXT[])
RETURNS TEXT LANGUAGE plpgsql IMMUTABLE AS $$
DECLARE
val TEXT;
numeric_count INT := 0;
date_count INT := 0;
boolean_count INT := 0;
total_non_null INT := 0;
BEGIN
FOR val IN SELECT unnest(sample_values) LOOP
IF val IS NOT NULL AND val != '' THEN
total_non_null := total_non_null + 1;
IF lower(val) IN ('true', 'false', 'yes', 'no', 't', 'f', 'y', 'n', '1', '0') THEN boolean_count := boolean_count + 1; END IF;
IF val ~ '\d+\s*x\s*\d+' THEN RETURN 'TEXT'; END IF;
IF val ~ '\d+\s*(cm|mm|m|km|in|ft|yd|kg|g|mg|lb|oz|ml|l|gal)' THEN RETURN 'TEXT'; END IF;
IF val ~ '^\$?-?[0-9,]+\.?[0-9]*$' THEN numeric_count := numeric_count + 1; END IF;
IF val ~ '^\d{4}-\d{2}-\d{2}' OR val ~ '^\d{1,2}/\d{1,2}/\d{4}' OR val ~ '^\d{5}$' THEN date_count := date_count + 1; END IF;
END IF;
END LOOP;
IF total_non_null = 0 THEN RETURN 'TEXT'; END IF;
IF boolean_count::float / total_non_null > 0.8 THEN RETURN 'BOOLEAN'; END IF;
IF numeric_count::float / total_non_null > 0.8 THEN RETURN 'NUMERIC'; END IF;
IF date_count::float / total_non_null > 0.8 THEN RETURN 'DATE'; END IF;
RETURN 'TEXT';
END $$;
-- ===============================================================
-- PART 6: INGESTION FUNCTIONS (RPC)
-- ===============================================================
-- 6a. Document Ingest (Standard)
CREATE OR REPLACE FUNCTION public.ingest_document_chunk(
p_uri TEXT, p_title TEXT, p_doc_meta JSONB,
p_chunk JSONB, p_nodes JSONB, p_edges JSONB, p_mentions JSONB
)
RETURNS JSONB LANGUAGE plpgsql SECURITY DEFINER SET search_path = public, pg_temp AS $$
DECLARE
v_doc_id BIGINT; v_chunk_id BIGINT; v_node JSONB; v_edge JSONB; v_mention JSONB;
v_src_id BIGINT; v_dst_id BIGINT;
BEGIN
INSERT INTO public.document(uri, title, doc_type, meta)
VALUES (p_uri, p_title, 'document', COALESCE(p_doc_meta, '{}'::jsonb))
ON CONFLICT (uri) DO UPDATE SET title = EXCLUDED.title, meta = public.document.meta || EXCLUDED.meta
RETURNING id INTO v_doc_id;
INSERT INTO public.chunk(document_id, ordinal, text)
VALUES (v_doc_id, (p_chunk->>'ordinal')::INT, p_chunk->>'text')
ON CONFLICT (document_id, ordinal) DO UPDATE SET text = EXCLUDED.text
RETURNING id INTO v_chunk_id;
IF (p_chunk ? 'embedding') THEN
INSERT INTO public.chunk_embedding(chunk_id, embedding)
VALUES (v_chunk_id, (SELECT array_agg((e)::float4 ORDER BY ord) FROM jsonb_array_elements_text(p_chunk->'embedding') WITH ORDINALITY t(e, ord))::halfvec(768))
ON CONFLICT (chunk_id) DO UPDATE SET embedding = EXCLUDED.embedding;
END IF;
FOR v_node IN SELECT * FROM jsonb_array_elements(COALESCE(p_nodes, '[]'::jsonb)) LOOP
INSERT INTO public.node(key, labels, props)
VALUES (v_node->>'key', COALESCE((SELECT array_agg(l::TEXT) FROM jsonb_array_elements_text(v_node->'labels') l), '{}'), COALESCE(v_node->'props', '{}'::jsonb))
ON CONFLICT (key) DO UPDATE SET props = public.node.props || EXCLUDED.props;
END LOOP;
FOR v_edge IN SELECT * FROM jsonb_array_elements(COALESCE(p_edges, '[]'::jsonb)) LOOP
SELECT id INTO v_src_id FROM public.node WHERE key = v_edge->>'src_key';
SELECT id INTO v_dst_id FROM public.node WHERE key = v_edge->>'dst_key';
IF v_src_id IS NOT NULL AND v_dst_id IS NOT NULL THEN
INSERT INTO public.edge(src, dst, type, props)
VALUES (v_src_id, v_dst_id, v_edge->>'type', COALESCE(v_edge->'props', '{}'::jsonb))
ON CONFLICT (src, dst, type) DO UPDATE SET props = public.edge.props || EXCLUDED.props;
END IF;
END LOOP;
FOR v_mention IN SELECT * FROM jsonb_array_elements(COALESCE(p_mentions, '[]'::jsonb)) LOOP
SELECT id INTO v_src_id FROM public.node WHERE key = v_mention->>'node_key';
IF v_chunk_id IS NOT NULL AND v_src_id IS NOT NULL THEN
INSERT INTO public.chunk_node(chunk_id, node_id, rel)
VALUES (v_chunk_id, v_src_id, COALESCE(v_mention->>'rel', 'MENTIONS'))
ON CONFLICT (chunk_id, node_id, rel) DO NOTHING;
END IF;
END LOOP;
RETURN jsonb_build_object('ok', true);
END $$;
-- 6b. Universal Spreadsheet Ingest (V8 Updated: Description Support)
DROP FUNCTION IF EXISTS public.ingest_spreadsheet(text, text, text, jsonb, jsonb, jsonb, jsonb);
CREATE OR REPLACE FUNCTION public.ingest_spreadsheet(
p_uri TEXT, p_title TEXT, p_table_name TEXT,
p_description TEXT, -- V8 Addition
p_rows JSONB, p_schema JSONB, p_nodes JSONB, p_edges JSONB
)
RETURNS JSONB LANGUAGE plpgsql SECURITY DEFINER SET search_path = public, pg_temp AS $$
DECLARE
v_doc_id BIGINT; v_safe_name TEXT; v_col_name TEXT; v_inferred_type TEXT;
v_cols TEXT[]; v_sample_values TEXT[]; v_row JSONB; v_node JSONB; v_edge JSONB;
v_src_id BIGINT; v_dst_id BIGINT; v_table_exists BOOLEAN; v_all_columns TEXT[];
v_schema_def JSONB;
BEGIN
INSERT INTO public.document(uri, title, doc_type, meta)
VALUES (p_uri, p_title, 'spreadsheet', jsonb_build_object('table_name', p_table_name))
ON CONFLICT (uri) DO UPDATE SET title = EXCLUDED.title RETURNING id INTO v_doc_id;
v_safe_name := public.sanitize_table_name(p_table_name);
SELECT EXISTS (SELECT FROM pg_tables WHERE schemaname = 'public' AND tablename = v_safe_name) INTO v_table_exists;
IF NOT v_table_exists THEN
-- Table Creation Logic
SELECT array_agg(DISTINCT key ORDER BY key) INTO v_all_columns FROM jsonb_array_elements(p_rows) AS r, jsonb_object_keys(r) AS key;
v_cols := ARRAY['id BIGSERIAL PRIMARY KEY'];
FOREACH v_col_name IN ARRAY v_all_columns LOOP
SELECT array_agg(kv.value::text) INTO v_sample_values FROM jsonb_array_elements(p_rows) r, jsonb_each_text(r) kv WHERE kv.key = v_col_name LIMIT 100;
v_inferred_type := public.infer_column_type(COALESCE(v_sample_values, ARRAY[]::TEXT[]));
v_cols := v_cols || format('%I %s', v_col_name, v_inferred_type);
END LOOP;
EXECUTE format('CREATE TABLE public.%I (%s)', v_safe_name, array_to_string(v_cols, ', '));
-- Permissions
EXECUTE format('GRANT ALL ON TABLE public.%I TO service_role', v_safe_name);
EXECUTE format('GRANT SELECT ON TABLE public.%I TO authenticated, anon', v_safe_name);
SELECT jsonb_object_agg(col_name, 'TEXT') INTO v_schema_def FROM unnest(v_all_columns) AS col_name;
END IF;
-- Insert Rows
FOR v_row IN SELECT * FROM jsonb_array_elements(p_rows) LOOP
DECLARE v_k TEXT; v_v TEXT; v_cl TEXT[] := ARRAY[]::TEXT[]; v_vl TEXT[] := ARRAY[]::TEXT[];
BEGIN
FOR v_k, v_v IN SELECT * FROM jsonb_each_text(v_row) LOOP v_cl := v_cl || quote_ident(v_k); v_vl := v_vl || quote_literal(v_v); END LOOP;
IF array_length(v_cl, 1) > 0 THEN EXECUTE format('INSERT INTO public.%I (%s) VALUES (%s)', v_safe_name, array_to_string(v_cl, ', '), array_to_string(v_vl, ', ')); END IF;
END;
END LOOP;
-- Upsert Registry (V8 with Description)
INSERT INTO public.structured_table(table_name, document_id, schema_def, row_count, description)
VALUES (v_safe_name, v_doc_id, COALESCE(v_schema_def, '{}'::jsonb), jsonb_array_length(p_rows), p_description)
ON CONFLICT (table_name) DO UPDATE SET updated_at = now(), description = EXCLUDED.description;
-- Upsert Graph Nodes
FOR v_node IN SELECT * FROM jsonb_array_elements(COALESCE(p_nodes, '[]'::jsonb)) LOOP
INSERT INTO public.node(key, labels, props)
VALUES (v_node->>'key', COALESCE((SELECT array_agg(l::TEXT) FROM jsonb_array_elements_text(v_node->'labels') l), '{}'), COALESCE(v_node->'props', '{}'::jsonb))
ON CONFLICT (key) DO UPDATE SET props = public.node.props || EXCLUDED.props;
END LOOP;
-- Upsert Graph Edges
FOR v_edge IN SELECT * FROM jsonb_array_elements(COALESCE(p_edges, '[]'::jsonb)) LOOP
SELECT id INTO v_src_id FROM public.node WHERE key = v_edge->>'src_key';
SELECT id INTO v_dst_id FROM public.node WHERE key = v_edge->>'dst_key';
IF v_src_id IS NOT NULL AND v_dst_id IS NOT NULL THEN
INSERT INTO public.edge(src, dst, type, props)
VALUES (v_src_id, v_dst_id, v_edge->>'type', COALESCE(v_edge->'props', '{}'::jsonb))
ON CONFLICT (src, dst, type) DO UPDATE SET props = public.edge.props || EXCLUDED.props;
END IF;
END LOOP;
RETURN jsonb_build_object('ok', true, 'table_name', v_safe_name);
END $$;
-- ===============================================================
-- PART 7: SEARCH FUNCTIONS (UPDATED V9 - COMPOSITE RETRIEVAL)
-- ===============================================================
-- 7a. Vector Search
CREATE OR REPLACE FUNCTION public.search_vector(
p_embedding VECTOR(768),
p_limit INT,
p_threshold FLOAT8 DEFAULT 0.65
)
RETURNS TABLE(chunk_id BIGINT, content TEXT, score FLOAT8)
LANGUAGE sql STABLE AS $$
SELECT
ce.chunk_id,
c.text as content,
1.0 / (1.0 + (ce.embedding <=> p_embedding)) AS score
FROM public.chunk_embedding ce
JOIN public.chunk c ON c.id = ce.chunk_id
WHERE (1.0 / (1.0 + (ce.embedding <=> p_embedding))) >= p_threshold
ORDER BY score DESC
LIMIT p_limit;
$$;
-- 7b. Multi-Strategy Full-Text Search (V9)
CREATE OR REPLACE FUNCTION public.search_fulltext(
p_query text,
p_limit integer
)
RETURNS TABLE(
chunk_id bigint,
content text,
score double precision
)
LANGUAGE sql STABLE AS $$
WITH query_variants AS (
SELECT
websearch_to_tsquery('english', p_query) AS tsq_websearch,
plainto_tsquery('english', p_query) AS tsq_plain,
to_tsquery('english', regexp_replace(p_query, '\s+', ' | ', 'g')) AS tsq_or
),
results AS (
-- Strategy 1: Websearch (most precise)
SELECT
c.id AS chunk_id,
c.text AS content,
ts_rank_cd(c.tsv, q.tsq_websearch)::float8 AS score,
1 as strategy
FROM public.chunk c
CROSS JOIN query_variants q
WHERE c.tsv @@ q.tsq_websearch
UNION ALL
-- Strategy 2: Plain text (more flexible)
SELECT
c.id AS chunk_id,
c.text AS content,
ts_rank_cd(c.tsv, q.tsq_plain)::float8 * 0.8 AS score,
2 as strategy
FROM public.chunk c
CROSS JOIN query_variants q
WHERE c.tsv @@ q.tsq_plain
AND NOT EXISTS (
SELECT 1 FROM public.chunk c2
CROSS JOIN query_variants q2
WHERE c2.id = c.id AND c2.tsv @@ q2.tsq_websearch
)
UNION ALL
-- Strategy 3: OR query (most flexible)
SELECT
c.id AS chunk_id,
c.text AS content,
ts_rank_cd(c.tsv, q.tsq_or)::float8 * 0.6 AS score,
3 as strategy
FROM public.chunk c
CROSS JOIN query_variants q
WHERE c.tsv @@ q.tsq_or
AND NOT EXISTS (
SELECT 1 FROM public.chunk c2
CROSS JOIN query_variants q2
WHERE c2.id = c.id
AND (c2.tsv @@ q2.tsq_websearch OR c2.tsv @@ q2.tsq_plain)
)
)
SELECT chunk_id, content, score
FROM results
ORDER BY score DESC
LIMIT p_limit;
$$;
GRANT EXECUTE ON FUNCTION public.search_fulltext TO anon, authenticated, service_role;
-- 7c. Table Peeking (V9: GENERIC FK DETECTION + REVERSE LOOKUP)
CREATE OR REPLACE FUNCTION public.get_table_context(p_table_name TEXT)
RETURNS JSONB
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
DECLARE
v_safe_name TEXT;
v_columns TEXT;
v_sample_row JSONB;
v_description TEXT;
v_semantics JSONB;
v_categorical_values JSONB := '{}'::jsonb;
v_related_tables JSONB := '[]'::jsonb;
v_cat_col TEXT;
v_cat_values JSONB;
v_distinct_count INT;
v_fk_col TEXT;
v_ref_table TEXT;
v_ref_table_exists BOOLEAN;
v_join_col TEXT;
v_ref_has_id BOOLEAN;
v_ref_has_name BOOLEAN;
v_reverse_rec RECORD;
BEGIN
v_safe_name := quote_ident(p_table_name);
-- Get schema
SELECT string_agg(column_name || ' (' || data_type || ')', ', ')
INTO v_columns
FROM information_schema.columns
WHERE table_name = p_table_name AND table_schema = 'public';
-- Get sample row
EXECUTE format('SELECT to_jsonb(t) FROM (SELECT * FROM public.%I LIMIT 1) t', v_safe_name)
INTO v_sample_row;
-- Get semantic metadata from structured_table
SELECT
description,
column_semantics
INTO v_description, v_semantics
FROM public.structured_table
WHERE table_name = p_table_name;
-- Get categorical values for columns with status/type/category in name
FOR v_cat_col IN
SELECT column_name
FROM information_schema.columns
WHERE table_schema = 'public'
AND table_name = p_table_name
AND (
column_name LIKE '%status%' OR
column_name LIKE '%type%' OR
column_name LIKE '%category%' OR
column_name LIKE '%state%' OR
column_name LIKE '%priority%' OR
column_name LIKE '%level%'
)
LOOP
BEGIN
-- Check if column has reasonable number of distinct values
EXECUTE format('SELECT COUNT(DISTINCT %I) FROM public.%I', v_cat_col, v_safe_name)
INTO v_distinct_count;
-- Only include if 20 or fewer distinct values
IF v_distinct_count <= 20 THEN
EXECUTE format('SELECT jsonb_agg(DISTINCT %I ORDER BY %I) FROM public.%I',
v_cat_col, v_cat_col, v_safe_name)
INTO v_cat_values;
-- Add to categorical_values object
v_categorical_values := v_categorical_values || jsonb_build_object(v_cat_col, v_cat_values);
END IF;
EXCEPTION WHEN OTHERS THEN
-- Skip this column if any error
CONTINUE;
END;
END LOOP;
-- ========================================================================
-- PART 1: FORWARD FK DETECTION (this table references other tables)
-- ========================================================================
FOR v_fk_col IN
SELECT column_name
FROM information_schema.columns
WHERE table_schema = 'public'
AND table_name = p_table_name
AND (
column_name LIKE '%\_id' OR -- customer_id, warehouse_id, manager_id
column_name LIKE '%\_name' -- manager_name, customer_name
)
AND column_name != 'id' -- Skip primary key
LOOP
-- Infer referenced table name
v_ref_table := regexp_replace(v_fk_col, '_(id|name)$', '');
-- Handle pluralization
-- carriers → carrier, employees → employee, warehouses → warehouse
IF v_ref_table ~ '(ss|us|ch|sh|x|z)es$' THEN
v_ref_table := regexp_replace(v_ref_table, 'es$', '');
ELSIF v_ref_table ~ 'ies$' THEN
v_ref_table := regexp_replace(v_ref_table, 'ies$', 'y');
ELSIF v_ref_table ~ 's$' THEN
v_ref_table := regexp_replace(v_ref_table, 's$', '');
END IF;
-- Add tbl_ prefix
v_ref_table := 'tbl_' || v_ref_table;
-- Check if referenced table exists
SELECT EXISTS (
SELECT FROM pg_tables
WHERE schemaname = 'public' AND tablename = v_ref_table
) INTO v_ref_table_exists;
IF v_ref_table_exists THEN
-- Determine join column in referenced table
v_join_col := NULL;
v_ref_has_id := FALSE;
v_ref_has_name := FALSE;
-- Check what columns the referenced table has
IF v_fk_col LIKE '%\_id' THEN
-- For FK columns ending in _id, look for matching ID column in ref table
-- customer_id → look for customer_id in tbl_customers
-- manager_id → look for employee_id in tbl_employees (special case)
SELECT bool_or(
column_name = v_fk_col OR
column_name = regexp_replace(v_ref_table, '^tbl_', '') || '_id'
)
INTO v_ref_has_id
FROM information_schema.columns
WHERE table_schema = 'public' AND table_name = v_ref_table;
IF v_ref_has_id THEN
-- Find the actual ID column name
SELECT column_name INTO v_join_col
FROM information_schema.columns
WHERE table_schema = 'public' AND table_name = v_ref_table
AND (column_name = v_fk_col OR
column_name = regexp_replace(v_ref_table, '^tbl_', '') || '_id')
LIMIT 1;
END IF;
END IF;
IF v_fk_col LIKE '%\_name' THEN
-- For FK columns ending in _name, look for 'name' column in ref table
SELECT bool_or(column_name = 'name')
INTO v_ref_has_name
FROM information_schema.columns
WHERE table_schema = 'public' AND table_name = v_ref_table;
IF v_ref_has_name THEN
v_join_col := 'name';
END IF;
END IF;
-- If we found a valid join column, add to related tables
IF v_join_col IS NOT NULL THEN
v_related_tables := v_related_tables || jsonb_build_array(
jsonb_build_object(
'table', v_ref_table,
'fk_column', v_fk_col,
'join_on', format('%I.%I = %I.%I',
p_table_name, v_fk_col,
v_ref_table, v_join_col),
'useful_columns', 'Details from ' || v_ref_table,
'use_when', format('Query mentions %s or asks about %s details',
regexp_replace(v_ref_table, '^tbl_', ''),
regexp_replace(v_fk_col, '_(id|name)$', ''))
)
);
END IF;
END IF;
END LOOP;
-- ========================================================================
-- PART 2: REVERSE FK DETECTION (other tables reference this table)
-- ========================================================================
-- Example: tbl_employees should know that tbl_warehouses.manager_name references it
FOR v_reverse_rec IN
SELECT DISTINCT
c.table_name,
c.column_name,
c.data_type
FROM information_schema.columns c
WHERE c.table_schema = 'public'
AND c.table_name LIKE 'tbl_%'
AND c.table_name != p_table_name
AND (
c.column_name LIKE '%\_id' OR
c.column_name LIKE '%\_name'
)
LOOP
v_ref_table := v_reverse_rec.table_name;
v_fk_col := v_reverse_rec.column_name;
v_join_col := NULL;
-- Extract the base entity name from the foreign key column
-- manager_id → manager, customer_name → customer
DECLARE
v_base_entity TEXT;
v_current_table_entity TEXT;
BEGIN
v_base_entity := regexp_replace(v_fk_col, '_(id|name)$', '');
v_current_table_entity := regexp_replace(p_table_name, '^tbl_', '');
-- Normalize pluralization for comparison
IF v_current_table_entity ~ 's$' THEN
v_current_table_entity := regexp_replace(v_current_table_entity, 's$', '');
END IF;
-- Check if this FK might reference the current table
-- Examples:
-- manager → employee (tbl_warehouses.manager_name → tbl_employees.name)
-- customer → customer (tbl_orders.customer_id → tbl_customers.customer_id)
-- employee → employee (tbl_employees.manager_id → tbl_employees.employee_id)
IF v_base_entity = v_current_table_entity OR
(v_base_entity = 'manager' AND v_current_table_entity = 'employee') OR
(v_base_entity = 'employee' AND v_current_table_entity = 'employee') THEN
-- Determine what column in current table this FK should join to
IF v_fk_col LIKE '%\_id' THEN
-- Look for matching ID column in current table
SELECT column_name INTO v_join_col
FROM information_schema.columns
WHERE table_schema = 'public'
AND table_name = p_table_name
AND (
column_name = v_fk_col OR
column_name = p_table_name || '_id' OR
column_name = regexp_replace(p_table_name, '^tbl_', '') || '_id'
)
LIMIT 1;
ELSIF v_fk_col LIKE '%\_name' THEN
-- Look for 'name' column in current table
SELECT column_name INTO v_join_col
FROM information_schema.columns
WHERE table_schema = 'public'
AND table_name = p_table_name
AND column_name = 'name'
LIMIT 1;
END IF;
-- If we found a matching join column, add to related tables
IF v_join_col IS NOT NULL THEN
-- Check if this relationship already exists (avoid duplicates from forward pass)
IF NOT EXISTS (
SELECT 1 FROM jsonb_array_elements(v_related_tables) elem
WHERE elem->>'table' = v_ref_table
AND elem->>'fk_column' = v_fk_col
) THEN
v_related_tables := v_related_tables || jsonb_build_array(
jsonb_build_object(
'table', v_ref_table,
'fk_column', v_fk_col,
'join_on', format('%I.%I = %I.%I',
p_table_name, v_join_col,
v_ref_table, v_fk_col),
'useful_columns', 'Details from ' || v_ref_table,
'use_when', format('Query asks about %s that reference %s',
regexp_replace(v_ref_table, '^tbl_', ''),
regexp_replace(p_table_name, '^tbl_', ''))
)
);
END IF;
END IF;
END IF;
END;
END LOOP;
RETURN jsonb_build_object(
'table', p_table_name,
'schema', v_columns,
'sample', COALESCE(v_sample_row, '{}'::jsonb),
'description', v_description,
'column_semantics', COALESCE(v_semantics, '{}'::jsonb),
'categorical_values', v_categorical_values,
'related_tables', v_related_tables
);
EXCEPTION WHEN OTHERS THEN
RETURN jsonb_build_object('error', SQLERRM);
END $$;
GRANT EXECUTE ON FUNCTION public.get_table_context(text) TO service_role, authenticated, anon;
-- 7d. Hybrid Graph Search
CREATE OR REPLACE FUNCTION public.search_graph_hybrid(
p_entities TEXT[],
p_actions TEXT[],
p_limit INT DEFAULT 20
)
RETURNS TABLE(chunk_id BIGINT, content TEXT, score FLOAT8, strategy TEXT)
LANGUAGE plpgsql STABLE AS $$
DECLARE
v_has_actions BOOLEAN;
BEGIN
v_has_actions := (array_length(p_actions, 1) > 0);
RETURN QUERY
WITH relevant_nodes AS (
SELECT id FROM public.node
WHERE EXISTS (
SELECT 1 FROM unnest(p_entities) entity
WHERE public.node.key ILIKE '%' || entity || '%'
OR public.node.props->>'name' ILIKE '%' || entity || '%'
)
),
relevant_edges AS (
SELECT e.src, e.dst, e.type,
CASE
WHEN s.id IS NOT NULL AND d.id IS NOT NULL THEN 'entity-entity'
ELSE 'entity-action'
END as match_strategy,
CASE
WHEN s.id IS NOT NULL AND d.id IS NOT NULL THEN 2.0
ELSE 1.5
END as base_score
FROM public.edge e
LEFT JOIN relevant_nodes s ON e.src = s.id
LEFT JOIN relevant_nodes d ON e.dst = d.id
WHERE
(s.id IS NOT NULL AND d.id IS NOT NULL)
OR
(v_has_actions AND (s.id IS NOT NULL OR d.id IS NOT NULL) AND
EXISTS (SELECT 1 FROM unnest(p_actions) act WHERE e.type ILIKE act || '%')
)
),
hits AS (
SELECT
cn.chunk_id,
count(*) as mention_count,
max(base_score) as max_strategy_score,
string_agg(DISTINCT match_strategy, ', ') as strategies
FROM relevant_edges re
JOIN public.chunk_node cn ON cn.node_id = re.src
GROUP BY cn.chunk_id
)
SELECT
h.chunk_id,
c.text as content,
(log(h.mention_count + 1) * h.max_strategy_score)::float8 AS score,
h.strategies::text as strategy
FROM hits h
JOIN public.chunk c ON c.id = h.chunk_id
ORDER BY score DESC
LIMIT p_limit;
END $$;
-- 7e. Legacy Targeted Graph Search (RESTORED FOR COMPLETENESS)
CREATE OR REPLACE FUNCTION public.search_graph_targeted(
p_entities TEXT[],
p_actions TEXT[],
p_limit INT DEFAULT 20
)
RETURNS TABLE(chunk_id BIGINT, content TEXT, score FLOAT8)
LANGUAGE plpgsql STABLE AS $$
DECLARE
v_has_actions BOOLEAN;
BEGIN
v_has_actions := (array_length(p_actions, 1) > 0);
RETURN QUERY
WITH relevant_nodes AS (
SELECT id, props->>'name' as name
FROM public.node
WHERE EXISTS (
SELECT 1 FROM unnest(p_entities) entity
WHERE public.node.key ILIKE '%' || entity || '%'
OR public.node.props->>'name' ILIKE '%' || entity || '%'
)
),
relevant_edges AS (
SELECT e.src, e.dst, e.type
FROM public.edge e
JOIN relevant_nodes rn ON e.src = rn.id
),
hits AS (
SELECT
cn.chunk_id,
count(*) as mention_count
FROM relevant_edges re
JOIN public.chunk_node cn ON cn.node_id = re.src
GROUP BY cn.chunk_id
)
SELECT
h.chunk_id,
c.text as content,
(log(h.mention_count + 1) * 1.5)::float8 AS score
FROM hits h
JOIN public.chunk c ON c.id = h.chunk_id
ORDER BY score DESC
LIMIT p_limit;
END $$;
-- 7f. Graph Neighborhood (Update 4: Security Definer / Case Insensitive)
DROP FUNCTION IF EXISTS public.get_graph_neighborhood(text[]);
CREATE OR REPLACE FUNCTION public.get_graph_neighborhood(
p_entity_names TEXT[]
)
RETURNS TABLE(subject TEXT, action TEXT, object TEXT, context JSONB)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
BEGIN
RETURN QUERY
WITH target_nodes AS (
SELECT id, key, props->>'name' as name
FROM public.node
WHERE
-- 1. Direct Key Match
key = ANY(p_entity_names)
-- 2. Name Match (Case Insensitive, Trimmed)
OR lower(trim(props->>'name')) = ANY(
SELECT lower(trim(x)) FROM unnest(p_entity_names) x
)
-- 3. Fuzzy Match (Substring)
OR EXISTS (
SELECT 1 FROM unnest(p_entity_names) term
WHERE length(term) > 3
AND public.node.props->>'name' ILIKE '%' || term || '%'
)
)
-- Outgoing Edges
SELECT n1.props->>'name', e.type, n2.props->>'name', e.props
FROM target_nodes tn
JOIN public.edge e ON tn.id = e.src
JOIN public.node n1 ON e.src = n1.id
JOIN public.node n2 ON e.dst = n2.id
UNION ALL
-- Incoming Edges
SELECT n1.props->>'name', e.type, n2.props->>'name', e.props
FROM target_nodes tn
JOIN public.edge e ON tn.id = e.dst
JOIN public.node n1 ON e.src = n1.id
JOIN public.node n2 ON e.dst = n2.id;
END $$;
GRANT EXECUTE ON FUNCTION public.get_graph_neighborhood(text[]) TO service_role, authenticated, anon;
-- 7g. Structured Search (Safe V6.1)
DROP FUNCTION IF EXISTS public.search_structured(text, int);
CREATE OR REPLACE FUNCTION public.search_structured(p_query_sql TEXT, p_limit INT DEFAULT 20)
RETURNS TABLE(table_name TEXT, row_data JSONB, score FLOAT8, rank INT)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
DECLARE
v_sql TEXT;
BEGIN
IF p_query_sql IS NULL OR length(trim(p_query_sql)) = 0 THEN RETURN; END IF;
v_sql := p_query_sql;
-- Sanitization
v_sql := regexp_replace(v_sql, '(\W)to\.([a-zA-Z0-9_]+)', '\1t_orders.\2', 'g');
v_sql := regexp_replace(v_sql, '\s+to\s+ON\s+', ' t_orders ON ', 'gi');
v_sql := regexp_replace(v_sql, '\s+AS\s+to\s+', ' AS t_orders ', 'gi');
v_sql := regexp_replace(v_sql, 'tbl_orders\s+to\s+', 'tbl_orders t_orders ', 'gi');
v_sql := regexp_replace(v_sql, '[;\s]+$', '');
RETURN QUERY EXECUTE format(
'WITH user_query AS (%s)
SELECT
''result''::text AS table_name,
to_jsonb(user_query.*) AS row_data,
1.0::float8 AS score,
(row_number() OVER ())::int AS rank
FROM user_query LIMIT %s',
v_sql, p_limit
);
EXCEPTION WHEN OTHERS THEN
RETURN QUERY SELECT 'ERROR'::text, jsonb_build_object('msg', SQLERRM, 'sql', v_sql), 1.0, 1;
END $$;
GRANT EXECUTE ON FUNCTION public.search_structured(text, int) TO service_role, authenticated, anon;
-- 7h. Smart Entity Detection (V9 - Composite Retrieval)
CREATE OR REPLACE FUNCTION public.detect_query_entities(
p_query TEXT
)
RETURNS TABLE(
entity_type TEXT,
table_name TEXT,
key_column TEXT,
key_value TEXT
)
LANGUAGE plpgsql
AS $$
BEGIN
-- Detect ORDER IDs (O followed by 5 digits)
IF p_query ~* 'O\d{5}' THEN
RETURN QUERY
SELECT
'order'::TEXT,
'tbl_orders'::TEXT,
'order_id'::TEXT,
(regexp_match(p_query, '(O\d{5})', 'i'))[1]::TEXT;
END IF;
-- Detect CUSTOMER IDs (CU followed by 3 digits)
IF p_query ~* 'CU\d{3}' THEN
RETURN QUERY
SELECT
'customer'::TEXT,
'tbl_customers'::TEXT,
'customer_id'::TEXT,
(regexp_match(p_query, '(CU\d{3})', 'i'))[1]::TEXT;
END IF;
-- Detect EMPLOYEE IDs (E followed by 3 digits)
IF p_query ~* 'E\d{3}' THEN
RETURN QUERY
SELECT
'employee'::TEXT,
'tbl_employees'::TEXT,
'employee_id'::TEXT,
(regexp_match(p_query, '(E\d{3})', 'i'))[1]::TEXT;
END IF;
-- Detect WAREHOUSE IDs (WH followed by 3 digits)
IF p_query ~* 'WH\d{3}' THEN
RETURN QUERY
SELECT
'warehouse'::TEXT,
'tbl_warehouses'::TEXT,
'warehouse_id'::TEXT,
(regexp_match(p_query, '(WH\d{3})', 'i'))[1]::TEXT;
END IF;
-- Detect CARRIER IDs (CR followed by 3 digits)
IF p_query ~* 'CR\d{3}' THEN
RETURN QUERY
SELECT
'carrier'::TEXT,
'tbl_carriers'::TEXT,
'carrier_id'::TEXT,
(regexp_match(p_query, '(CR\d{3})', 'i'))[1]::TEXT;
END IF;
RETURN;
END;
$$;
GRANT EXECUTE ON FUNCTION public.detect_query_entities TO anon, authenticated, service_role;
-- 7i. Context Enrichment (V9 - Composite Retrieval)
CREATE OR REPLACE FUNCTION public.enrich_query_context(
p_primary_table TEXT,
p_primary_key TEXT,
p_primary_value TEXT
)
RETURNS TABLE(
enrichment_type TEXT,
table_name TEXT,
row_data JSONB,
relationship TEXT
)
LANGUAGE plpgsql
AS $$
DECLARE
v_customer_id TEXT;
v_warehouse_id TEXT;
v_employee_id TEXT;
v_carrier_id TEXT;
v_primary_row JSONB;
BEGIN
-- ====================================================
-- STEP 1: GET PRIMARY ROW
-- ====================================================
EXECUTE format(
'SELECT to_jsonb(t.*) FROM public.%I t WHERE %I = $1 LIMIT 1',
p_primary_table,
p_primary_key
) INTO v_primary_row USING p_primary_value;
IF v_primary_row IS NULL THEN
RETURN;
END IF;
RETURN QUERY SELECT
'primary'::TEXT,
p_primary_table,
v_primary_row,
'direct_match'::TEXT;
-- ====================================================
-- STEP 2: FOLLOW FOREIGN KEYS (ENRICH!)
-- ====================================================
-- ORDERS TABLE
IF p_primary_table = 'tbl_orders' THEN
v_customer_id := v_primary_row->>'customer_id';
v_warehouse_id := v_primary_row->>'warehouse_id';
v_carrier_id := v_primary_row->>'carrier_id';
-- Customer
IF v_customer_id IS NOT NULL THEN
RETURN QUERY
SELECT 'enrichment'::TEXT, 'tbl_customers'::TEXT, to_jsonb(c.*), 'order_customer'::TEXT
FROM public.tbl_customers c WHERE c.customer_id = v_customer_id;
-- Customer History
RETURN QUERY
SELECT 'related_orders'::TEXT, 'tbl_orders'::TEXT, to_jsonb(o.*), 'customer_history'::TEXT
FROM public.tbl_orders o WHERE o.customer_id = v_customer_id AND o.order_id != p_primary_value
ORDER BY o.order_date DESC LIMIT 5;
END IF;
-- Warehouse
IF v_warehouse_id IS NOT NULL THEN
RETURN QUERY
SELECT 'enrichment'::TEXT, 'tbl_warehouses'::TEXT, to_jsonb(w.*), 'order_warehouse'::TEXT
FROM public.tbl_warehouses w WHERE w.warehouse_id = v_warehouse_id;
END IF;
-- Carrier
IF v_carrier_id IS NOT NULL THEN
RETURN QUERY
SELECT 'enrichment'::TEXT, 'tbl_carriers'::TEXT, to_jsonb(cr.*), 'order_carrier'::TEXT
FROM public.tbl_carriers cr WHERE cr.carrier_id = v_carrier_id;
END IF;
END IF;
-- CUSTOMERS TABLE
IF p_primary_table = 'tbl_customers' THEN
RETURN QUERY
SELECT 'enrichment'::TEXT, 'tbl_orders'::TEXT, to_jsonb(o.*), 'customer_orders'::TEXT
FROM public.tbl_orders o WHERE o.customer_id = p_primary_value
ORDER BY o.order_date DESC LIMIT 10;
END IF;
-- EMPLOYEES TABLE
IF p_primary_table = 'tbl_employees' THEN
v_employee_id := v_primary_row->>'employee_id';
RETURN QUERY
SELECT 'enrichment'::TEXT, 'tbl_employees'::TEXT, to_jsonb(e.*), 'direct_reports'::TEXT
FROM public.tbl_employees e WHERE e.manager_id = v_employee_id;
RETURN QUERY
SELECT 'enrichment'::TEXT, 'tbl_employees'::TEXT, to_jsonb(m.*), 'manager'::TEXT
FROM public.tbl_employees m WHERE m.employee_id = (v_primary_row->>'manager_id');
END IF;
-- WAREHOUSES TABLE
IF p_primary_table = 'tbl_warehouses' THEN
RETURN QUERY
SELECT 'enrichment'::TEXT, 'tbl_orders'::TEXT, to_jsonb(o.*), 'warehouse_orders'::TEXT
FROM public.tbl_orders o WHERE o.warehouse_id = p_primary_value
ORDER BY o.order_date DESC LIMIT 10;
END IF;
RETURN;
END;
$$;
GRANT EXECUTE ON FUNCTION public.enrich_query_context TO anon, authenticated, service_role;
-- ===============================================================
-- PART 8: CONFIGURATION SYSTEM
-- ===============================================================
INSERT INTO public.app_config (id, settings)
VALUES (1, '{
"chunk_size": 500,
"chunk_overlap": 100,
"graph_sample_rate": 5,
"worker_batch_size": 5,
"model_router": "gemini-2.5-flash",
"model_reranker": "gemini-2.5-flash-lite",
"model_sql": "gemini-2.5-flash",
"model_extraction": "gemini-2.5-flash",
"rrf_weight_enrichment": 15.0,
"rrf_weight_sql": 10.0,
"rrf_weight_graph": 5.0,
"rrf_weight_fts": 3.0,
"rrf_weight_vector": 1.0,
"rerank_depth": 15,
"min_vector_score": 0.01,
"search_limit": 10
}'::jsonb)
ON CONFLICT (id) DO NOTHING;
CREATE OR REPLACE FUNCTION public.configure_system(
p_chunk_size INT DEFAULT NULL,
p_chunk_overlap INT DEFAULT NULL,
p_graph_sample_rate INT DEFAULT NULL,
p_worker_batch_size INT DEFAULT NULL,
p_model_router TEXT DEFAULT NULL,
p_model_reranker TEXT DEFAULT NULL,
p_model_extraction TEXT DEFAULT NULL,
p_search_limit INT DEFAULT NULL,
p_rerank_depth INT DEFAULT NULL,
p_rrf_weight_vector NUMERIC DEFAULT NULL,
p_rrf_weight_graph NUMERIC DEFAULT NULL,
p_rrf_weight_sql NUMERIC DEFAULT NULL
)
RETURNS TEXT LANGUAGE plpgsql SECURITY DEFINER AS $$
DECLARE
current_settings JSONB;
new_settings JSONB;
BEGIN
SELECT settings INTO current_settings FROM public.app_config WHERE id = 1;
new_settings := current_settings;
IF p_chunk_size IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{chunk_size}', to_jsonb(p_chunk_size)); END IF;
IF p_chunk_overlap IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{chunk_overlap}', to_jsonb(p_chunk_overlap)); END IF;
IF p_graph_sample_rate IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{graph_sample_rate}', to_jsonb(p_graph_sample_rate)); END IF;
IF p_worker_batch_size IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{worker_batch_size}', to_jsonb(p_worker_batch_size)); END IF;
IF p_model_router IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{model_router}', to_jsonb(p_model_router)); END IF;
IF p_model_reranker IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{model_reranker}', to_jsonb(p_model_reranker)); END IF;
IF p_model_extraction IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{model_extraction}', to_jsonb(p_model_extraction)); END IF;
IF p_search_limit IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{search_limit}', to_jsonb(p_search_limit)); END IF;
IF p_rerank_depth IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{rerank_depth}', to_jsonb(p_rerank_depth)); END IF;
IF p_rrf_weight_vector IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{rrf_weight_vector}', to_jsonb(p_rrf_weight_vector)); END IF;
IF p_rrf_weight_graph IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{rrf_weight_graph}', to_jsonb(p_rrf_weight_graph)); END IF;
IF p_rrf_weight_sql IS NOT NULL THEN new_settings := jsonb_set(new_settings, '{rrf_weight_sql}', to_jsonb(p_rrf_weight_sql)); END IF;
UPDATE public.app_config SET settings = new_settings, updated_at = now() WHERE id = 1;
RETURN 'System configuration updated successfully.';
END;
$$;
-- ===============================================================
-- PART 9: GRAPH CONTEXT RERANKER
-- ===============================================================
DROP FUNCTION IF EXISTS public.get_graph_context(bigint[], text[]);
CREATE OR REPLACE FUNCTION public.get_graph_context(
p_chunk_ids BIGINT[],
p_keywords TEXT[] DEFAULT '{}',
p_actions TEXT[] DEFAULT '{}'
)
RETURNS TABLE (chunk_id BIGINT, graph_data JSONB)
LANGUAGE sql STABLE AS $$
WITH raw_edges AS (
SELECT
cn_src.chunk_id,
jsonb_build_object(
'subject', n1.props->>'name',
'action', e.type,
'object', n2.props->>'name',
'context', e.props
) as edge_json,
(
0
+ (SELECT COALESCE(MAX(CASE WHEN n1.props->>'name' ILIKE '%' || kw || '%' THEN 10 ELSE 0 END),0) FROM unnest(p_keywords) kw)
+ (SELECT COALESCE(MAX(CASE WHEN n2.props->>'name' ILIKE '%' || kw || '%' THEN 10 ELSE 0 END),0) FROM unnest(p_keywords) kw)
+ (SELECT COALESCE(MAX(CASE WHEN array_length(p_actions, 1) > 0 AND e.type ILIKE act || '%' THEN 20 ELSE 0 END),0) FROM unnest(p_actions) act)
) as relevance_score
FROM public.chunk_node cn_src
JOIN public.edge e ON cn_src.node_id = e.src
JOIN public.chunk_node cn_tgt ON e.dst = cn_tgt.node_id AND cn_src.chunk_id = cn_tgt.chunk_id
JOIN public.node n1 ON e.src = n1.id
JOIN public.node n2 ON e.dst = n2.id
WHERE cn_src.chunk_id = ANY(p_chunk_ids)
),
ranked_edges AS (
SELECT
chunk_id,
edge_json,
relevance_score,
ROW_NUMBER() OVER (PARTITION BY chunk_id ORDER BY relevance_score DESC, length(edge_json::text) ASC) as rn
FROM raw_edges
)
SELECT
chunk_id,
jsonb_agg(edge_json) as graph_data
FROM ranked_edges
WHERE rn <= 5
GROUP BY chunk_id;
$$;
-- ===============================================================
-- PART 10: WORKER SETUP
-- ===============================================================
CREATE OR REPLACE FUNCTION public.setup_worker(project_url TEXT, service_role_key TEXT)
RETURNS TEXT LANGUAGE plpgsql SECURITY DEFINER AS $$
BEGIN
EXECUTE format(
$f$
CREATE OR REPLACE FUNCTION public.trigger_ingest_worker()
RETURNS TRIGGER LANGUAGE plpgsql SECURITY DEFINER AS $func$
BEGIN
PERFORM net.http_post(
url := '%s/functions/v1/ingest-worker',
headers := jsonb_build_object('Content-Type', 'application/json', 'Authorization', 'Bearer %s'),
body := jsonb_build_object('type', 'INSERT', 'table', 'objects', 'schema', 'storage', 'record', row_to_json(NEW))
);
RETURN NEW;
END;
$func$;
$f$,
project_url, service_role_key
);
DROP TRIGGER IF EXISTS "trigger-ingest-worker" ON storage.objects;
CREATE TRIGGER "trigger-ingest-worker"
AFTER INSERT ON storage.objects
FOR EACH ROW
WHEN (NEW.bucket_id = 'raw_uploads')
EXECUTE FUNCTION public.trigger_ingest_worker();
RETURN 'Worker configured successfully!';
END;
$$;
-- ===============================================================
-- PART 11: PERMISSIONS & REPAIRS
-- ===============================================================
GRANT USAGE ON SCHEMA public TO service_role, authenticated, anon;
GRANT ALL ON ALL TABLES IN SCHEMA public TO service_role, authenticated;
GRANT ALL ON ALL SEQUENCES IN SCHEMA public TO service_role, authenticated;
GRANT ALL ON TABLE public.ingestion_queue TO service_role, postgres, anon, authenticated;
GRANT SELECT ON TABLE public.app_config TO anon, authenticated, service_role;
GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA public TO authenticated, service_role;
-- 🚨 REPAIR SCRIPT: Fix permissions for any EXISTING "tbl_" spreadsheets
DO $$
DECLARE
r RECORD;
BEGIN
FOR r IN (SELECT tablename FROM pg_tables WHERE schemaname = 'public' AND tablename LIKE 'tbl_%')
LOOP
EXECUTE format('GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE public.%I TO service_role', r.tablename);
EXECUTE format('GRANT SELECT ON TABLE public.%I TO anon, authenticated', r.tablename);
END LOOP;
END $$;
Step 2. This is minor but important. If you upload a BIG file, it needs to be able to process in the background. So in order to enable background functions you need to set up a worker. In a new query, paste this and then fill in your information before running it:
-- insert your url and service worker secret
SELECT setup_worker(
'https://YOUR URL.supabase.co',
'ey**YOUR SERVICE WORKER SERET***Tc'
);
Step 3. If you want to adjust weighting and other parameters, run any of these queries with your desired adjustments:
-- ===============================================================
-- CONTEXT MESH: SIMPLE CONFIGURATION SCRIPT
-- ===============================================================
-- Instructions: Replace the default values below and run any line
-- Each parameter can be updated independently
-- ===============================================================
-- ============================================================
-- INGESTION PARAMETERS
-- ============================================================
SELECT public.configure_system(p_chunk_size => 500);
-- What it does: Number of characters per document chunk
-- Suggestions:
-- Higher (800-1000) = Better context, slower processing
-- Lower (300-400) = Faster processing, more precise retrieval
-- Default (500) = Balanced performance
SELECT public.configure_system(p_chunk_overlap => 100);
-- What it does: Character overlap between adjacent chunks
-- Suggestions:
-- Higher (150-200) = Better continuity, more redundancy
-- Lower (50-75) = Less redundancy, faster processing
-- Default (100) = Balanced overlap
SELECT public.configure_system(p_graph_sample_rate => 5);
-- What it does: Extract graph relationships from every Nth chunk
-- Suggestions:
-- Higher (10-15) = Faster ingestion, sparser graph
-- Lower (1-3) = Dense graph, slower ingestion
-- Default (5) = Balanced graph density
SELECT public.configure_system(p_worker_batch_size => 5);
-- What it does: Queue items processed per worker batch
-- Suggestions:
-- Higher (10-20) = Faster bulk ingestion, higher memory
-- Lower (1-3) = Slower ingestion, lower memory
-- Default (5) = Balanced throughput
-- ============================================================
-- MODEL SELECTION
-- ============================================================
SELECT public.configure_system(p_model_router => 'gemini-2.5-flash');
-- What it does: LLM for query routing and entity extraction
-- Suggestions:
-- 'gemini-2.5-flash' = Fast, cost-effective (default)
-- 'gemini-2.0-flash-exp' = Experimental, free tier
-- 'gemini-2.5-pro' = Most accurate, expensive
SELECT public.configure_system(p_model_reranker => 'gemini-2.5-flash-lite');
-- What it does: LLM for reranking documents by relevance
-- Suggestions:
-- 'gemini-2.5-flash-lite' = Ultra-fast, cheap (default)
-- 'gemini-2.5-flash' = More accurate, slightly slower
-- 'gemini-2.0-flash-exp' = Free tier option
SELECT public.configure_system(p_model_sql => 'gemini-2.5-flash');
-- What it does: LLM for generating SQL queries
-- Suggestions:
-- 'gemini-2.5-flash' = Good balance (default)
-- 'gemini-2.5-pro' = Complex queries, better accuracy
-- 'gemini-2.0-flash-exp' = Free tier option
-- ============================================================
-- SEARCH WEIGHTS (RRF Fusion)
-- ============================================================
-- Higher weight = More influence in final ranking
-- ============================================================
SELECT public.configure_system(p_rrf_weight_enrichment => 15.0);
-- What it does: Weight for composite enrichment (SQL + FK relationships)
-- Suggestions:
-- Higher (20-30) = Prioritize structured data context
-- Lower (10-12) = Balance with documents
-- Default (15.0) = Highest priority (recommended)
SELECT public.configure_system(p_rrf_weight_sql => 10.0);
-- What it does: Weight for direct SQL query results
-- Suggestions:
-- Higher (15-20) = Prioritize exact matches
-- Lower (5-8) = More exploratory results
-- Default (10.0) = Strong influence
SELECT public.configure_system(p_rrf_weight_graph => 5.0);
-- What it does: Weight for knowledge graph relationships
-- Suggestions:
-- Higher (8-12) = More relationship context
-- Lower (2-4) = Focus on direct matches
-- Default (5.0) = Moderate context
SELECT public.configure_system(p_rrf_weight_fts => 3.0);
-- What it does: Weight for full-text keyword search
-- Suggestions:
-- Higher (5-7) = Better keyword matching
-- Lower (1