-- Esquema convergente completo de API-NLP para pgAdmin/DataGrid. -- Sirve para una BD vacia o una BD existente que ya use VECTOR(300). -- NO convierte VECTOR(16): usa la migracion FastText separada para ese caso. DO $bootstrap$ DECLARE actual_type TEXT; BEGIN IF to_regclass('public.post_embeddings') IS NOT NULL THEN SELECT format_type(attribute.atttypid, attribute.atttypmod) INTO actual_type FROM pg_attribute attribute WHERE attribute.attrelid = 'public.post_embeddings'::regclass AND attribute.attname = 'embedding' AND NOT attribute.attisdropped; IF actual_type IS DISTINCT FROM 'vector(300)' THEN RAISE EXCEPTION 'Se requiere vector(300), actual=%', actual_type; END IF; END IF; IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'nlp_owner') THEN EXECUTE 'CREATE ROLE nlp_owner NOLOGIN'; END IF; IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'nlp_reader') THEN EXECUTE 'CREATE ROLE nlp_reader LOGIN'; END IF; IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'nlp_writer') THEN EXECUTE 'CREATE ROLE nlp_writer LOGIN'; END IF; END $bootstrap$; SET search_path = public; -- Reference contract for AWS RDS/Aurora PostgreSQL + pgvector. -- Run this once with an admin/DBA role, not from Hugging Face. CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE IF NOT EXISTS place_embeddings ( external_id TEXT PRIMARY KEY, document TEXT NOT NULL, metadata JSONB NOT NULL DEFAULT '{}'::jsonb, embedding VECTOR(300) NOT NULL, content_hash TEXT NOT NULL, embedding_model TEXT NOT NULL, embedding_version TEXT NOT NULL, is_active BOOLEAN NOT NULL DEFAULT true, updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE TABLE IF NOT EXISTS post_embeddings ( external_id TEXT PRIMARY KEY, document TEXT NOT NULL, metadata JSONB NOT NULL DEFAULT '{}'::jsonb, embedding VECTOR(300) NOT NULL, content_hash TEXT NOT NULL, embedding_model TEXT NOT NULL, embedding_version TEXT NOT NULL, is_active BOOLEAN NOT NULL DEFAULT true, updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); -- Post-feed derived storage. The versioned companion migration defines the -- controlled functions, revokes and grants used by the adapters. CREATE TABLE IF NOT EXISTS post_cluster_runs ( id UUID PRIMARY KEY, algorithm TEXT NOT NULL CHECK (algorithm = 'kmeans'), algorithm_version TEXT NOT NULL, embedding_model TEXT NOT NULL, embedding_version TEXT NOT NULL, embedding_dimension INTEGER NOT NULL CHECK (embedding_dimension = 300), k INTEGER NOT NULL CHECK (k > 1), sample_size INTEGER NOT NULL CHECK (sample_size >= k), inertia DOUBLE PRECISION, silhouette_score DOUBLE PRECISION, status TEXT NOT NULL CHECK (status IN ('running', 'validated', 'active', 'failed', 'superseded')), parameters JSONB NOT NULL DEFAULT '{}'::jsonb, started_at TIMESTAMPTZ NOT NULL, completed_at TIMESTAMPTZ, activated_at TIMESTAMPTZ, error_message TEXT ); CREATE UNIQUE INDEX IF NOT EXISTS post_cluster_runs_one_active_idx ON post_cluster_runs ((status)) WHERE status = 'active'; CREATE TABLE IF NOT EXISTS post_clusters ( run_id UUID NOT NULL REFERENCES post_cluster_runs(id) ON DELETE CASCADE, cluster_id INTEGER NOT NULL, centroid VECTOR(300) NOT NULL, label TEXT, size INTEGER NOT NULL CHECK (size >= 0), representative_post_ids TEXT[] NOT NULL DEFAULT '{}', metadata JSONB NOT NULL DEFAULT '{}'::jsonb, PRIMARY KEY (run_id, cluster_id) ); CREATE TABLE IF NOT EXISTS post_cluster_memberships ( run_id UUID NOT NULL, post_id TEXT NOT NULL REFERENCES post_embeddings(external_id) ON DELETE CASCADE, cluster_id INTEGER NOT NULL, distance_to_centroid DOUBLE PRECISION NOT NULL CHECK (distance_to_centroid >= 0), assigned_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (run_id, post_id), FOREIGN KEY (run_id, cluster_id) REFERENCES post_clusters(run_id, cluster_id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS user_interest_embeddings ( user_id TEXT PRIMARY KEY, embedding VECTOR(300) NOT NULL, weighted_sum VECTOR(300) NOT NULL, total_weight DOUBLE PRECISION NOT NULL DEFAULT 0 CHECK (total_weight >= 0), source_hash TEXT NOT NULL, profile_version TEXT NOT NULL, interaction_count INTEGER NOT NULL DEFAULT 0 CHECK (interaction_count >= 0), last_event_id BIGINT NOT NULL DEFAULT 0 CHECK (last_event_id >= 0), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE TABLE IF NOT EXISTS post_sync_checkpoints ( consumer TEXT PRIMARY KEY, last_event_id BIGINT NOT NULL DEFAULT 0 CHECK (last_event_id >= 0), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE TABLE IF NOT EXISTS post_embedding_tombstones ( post_id TEXT PRIMARY KEY, source_version BIGINT NOT NULL CHECK (source_version >= 0), deleted_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE TABLE IF NOT EXISTS user_embeddings ( external_id TEXT PRIMARY KEY, document TEXT NOT NULL, metadata JSONB NOT NULL DEFAULT '{}'::jsonb, embedding VECTOR(300) NOT NULL, content_hash TEXT NOT NULL, embedding_model TEXT NOT NULL, embedding_version TEXT NOT NULL, is_active BOOLEAN NOT NULL DEFAULT true, author_type TEXT, author_id TEXT, published_at TIMESTAMPTZ, source_version BIGINT, indexed_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE TABLE IF NOT EXISTS club_embeddings (LIKE user_embeddings INCLUDING ALL); CREATE TABLE IF NOT EXISTS group_embeddings (LIKE user_embeddings INCLUDING ALL); CREATE TABLE IF NOT EXISTS event_embeddings (LIKE user_embeddings INCLUDING ALL); ALTER TABLE place_embeddings ADD COLUMN IF NOT EXISTS textsearch TSVECTOR GENERATED ALWAYS AS (to_tsvector('simple', document)) STORED; ALTER TABLE post_embeddings ADD COLUMN IF NOT EXISTS textsearch TSVECTOR GENERATED ALWAYS AS (to_tsvector('simple', document)) STORED; ALTER TABLE user_embeddings ADD COLUMN IF NOT EXISTS textsearch TSVECTOR GENERATED ALWAYS AS (to_tsvector('simple', document)) STORED; ALTER TABLE club_embeddings ADD COLUMN IF NOT EXISTS textsearch TSVECTOR GENERATED ALWAYS AS (to_tsvector('simple', document)) STORED; ALTER TABLE group_embeddings ADD COLUMN IF NOT EXISTS textsearch TSVECTOR GENERATED ALWAYS AS (to_tsvector('simple', document)) STORED; ALTER TABLE event_embeddings ADD COLUMN IF NOT EXISTS textsearch TSVECTOR GENERATED ALWAYS AS (to_tsvector('simple', document)) STORED; CREATE INDEX IF NOT EXISTS place_embeddings_embedding_hnsw_idx ON place_embeddings USING hnsw (embedding vector_cosine_ops); CREATE INDEX IF NOT EXISTS post_embeddings_embedding_hnsw_idx ON post_embeddings USING hnsw (embedding vector_cosine_ops); CREATE INDEX IF NOT EXISTS place_embeddings_metadata_gin_idx ON place_embeddings USING gin (metadata); CREATE INDEX IF NOT EXISTS post_embeddings_metadata_gin_idx ON post_embeddings USING gin (metadata); CREATE INDEX IF NOT EXISTS user_embeddings_embedding_hnsw_idx ON user_embeddings USING hnsw (embedding vector_cosine_ops); CREATE INDEX IF NOT EXISTS club_embeddings_embedding_hnsw_idx ON club_embeddings USING hnsw (embedding vector_cosine_ops); CREATE INDEX IF NOT EXISTS group_embeddings_embedding_hnsw_idx ON group_embeddings USING hnsw (embedding vector_cosine_ops); CREATE INDEX IF NOT EXISTS event_embeddings_embedding_hnsw_idx ON event_embeddings USING hnsw (embedding vector_cosine_ops); CREATE INDEX IF NOT EXISTS user_embeddings_metadata_gin_idx ON user_embeddings USING gin (metadata); CREATE INDEX IF NOT EXISTS club_embeddings_metadata_gin_idx ON club_embeddings USING gin (metadata); CREATE INDEX IF NOT EXISTS group_embeddings_metadata_gin_idx ON group_embeddings USING gin (metadata); CREATE INDEX IF NOT EXISTS event_embeddings_metadata_gin_idx ON event_embeddings USING gin (metadata); CREATE INDEX IF NOT EXISTS place_embeddings_textsearch_gin_idx ON place_embeddings USING gin (textsearch); CREATE INDEX IF NOT EXISTS post_embeddings_textsearch_gin_idx ON post_embeddings USING gin (textsearch); CREATE INDEX IF NOT EXISTS user_embeddings_textsearch_gin_idx ON user_embeddings USING gin (textsearch); CREATE INDEX IF NOT EXISTS club_embeddings_textsearch_gin_idx ON club_embeddings USING gin (textsearch); CREATE INDEX IF NOT EXISTS group_embeddings_textsearch_gin_idx ON group_embeddings USING gin (textsearch); CREATE INDEX IF NOT EXISTS event_embeddings_textsearch_gin_idx ON event_embeddings USING gin (textsearch); CREATE OR REPLACE FUNCTION match_places( query_embedding VECTOR(300), match_count INTEGER, filters JSONB DEFAULT '{}'::jsonb ) RETURNS TABLE ( external_id TEXT, document TEXT, metadata JSONB, score DOUBLE PRECISION ) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ SELECT p.external_id, p.document, p.metadata, 1 - (p.embedding <=> query_embedding) AS score FROM public.place_embeddings p WHERE p.is_active = true AND COALESCE((filters->>'is_active')::boolean, true) = true AND ((filters ? 'city') IS FALSE OR lower(p.metadata->>'city') = lower(filters->>'city')) AND ((filters ? 'state') IS FALSE OR lower(p.metadata->>'state') = lower(filters->>'state')) AND ((filters ? 'category') IS FALSE OR lower(p.metadata->>'category') = lower(filters->>'category')) AND ( (filters ? 'categories') IS FALSE OR EXISTS ( SELECT 1 FROM jsonb_array_elements_text(filters->'categories') AS allowed(value) WHERE lower(p.metadata->>'category') = lower(allowed.value) ) ) AND ((filters ? 'price_range') IS FALSE OR p.metadata->>'price_range' = filters->>'price_range') AND ( (filters ? 'price_ranges') IS FALSE OR EXISTS ( SELECT 1 FROM jsonb_array_elements_text(filters->'price_ranges') AS allowed(value) WHERE p.metadata->>'price_range' = allowed.value ) ) AND ( (filters ? 'tags') IS FALSE OR EXISTS ( SELECT 1 FROM jsonb_array_elements_text(filters->'tags') AS allowed(value) WHERE lower(COALESCE(p.metadata->>'tags', '')) LIKE ('%' || lower(allowed.value) || '%') ) ) AND ((filters ? 'occasion') IS FALSE OR p.metadata->>'occasion' ILIKE ('%' || (filters->>'occasion') || '%')) AND ( (filters ? 'place_ids') IS FALSE OR p.external_id IN (SELECT jsonb_array_elements_text(filters->'place_ids')) ) ORDER BY p.embedding <=> query_embedding, p.external_id ASC LIMIT match_count; $$; CREATE OR REPLACE FUNCTION match_posts( query_embedding VECTOR(300), match_count INTEGER, filters JSONB DEFAULT '{}'::jsonb ) RETURNS TABLE ( external_id TEXT, document TEXT, metadata JSONB, score DOUBLE PRECISION ) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ SELECT p.external_id, p.document, p.metadata, 1 - (p.embedding <=> query_embedding) AS score FROM public.post_embeddings p WHERE p.is_active = true AND COALESCE((filters->>'is_active')::boolean, true) = true AND ((filters ? 'city') IS FALSE OR lower(p.metadata->>'city') = lower(filters->>'city')) ORDER BY p.embedding <=> query_embedding LIMIT match_count; $$; CREATE OR REPLACE FUNCTION upsert_place_embedding( p_external_id TEXT, p_document TEXT, p_metadata JSONB, p_embedding VECTOR(300), p_content_hash TEXT, p_embedding_model TEXT, p_embedding_version TEXT, p_is_active BOOLEAN ) RETURNS VOID LANGUAGE sql SECURITY DEFINER SET search_path = public AS $$ INSERT INTO public.place_embeddings ( external_id, document, metadata, embedding, content_hash, embedding_model, embedding_version, is_active, updated_at ) VALUES ( p_external_id, p_document, p_metadata, p_embedding, p_content_hash, p_embedding_model, p_embedding_version, p_is_active, now() ) ON CONFLICT (external_id) DO UPDATE SET document = EXCLUDED.document, metadata = EXCLUDED.metadata, embedding = EXCLUDED.embedding, content_hash = EXCLUDED.content_hash, embedding_model = EXCLUDED.embedding_model, embedding_version = EXCLUDED.embedding_version, is_active = EXCLUDED.is_active, updated_at = now(); $$; CREATE OR REPLACE FUNCTION upsert_post_embedding( p_external_id TEXT, p_document TEXT, p_metadata JSONB, p_embedding VECTOR(300), p_content_hash TEXT, p_embedding_model TEXT, p_embedding_version TEXT, p_is_active BOOLEAN, p_author_type TEXT, p_author_id TEXT, p_published_at TIMESTAMPTZ, p_source_version BIGINT ) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN IF EXISTS ( SELECT 1 FROM public.post_embedding_tombstones tombstone WHERE tombstone.post_id = p_external_id AND (p_source_version IS NULL OR tombstone.source_version >= p_source_version) ) THEN RETURN; END IF; IF p_source_version IS NOT NULL THEN DELETE FROM public.post_embedding_tombstones WHERE post_id = p_external_id AND source_version < p_source_version; END IF; INSERT INTO public.post_embeddings ( external_id, document, metadata, embedding, content_hash, embedding_model, embedding_version, is_active, author_type, author_id, published_at, source_version, indexed_at, updated_at ) VALUES ( p_external_id, p_document, p_metadata, p_embedding, p_content_hash, p_embedding_model, p_embedding_version, p_is_active, p_author_type, p_author_id, p_published_at, p_source_version, now(), now() ) ON CONFLICT (external_id) DO UPDATE SET document = EXCLUDED.document, metadata = EXCLUDED.metadata, embedding = EXCLUDED.embedding, content_hash = EXCLUDED.content_hash, embedding_model = EXCLUDED.embedding_model, embedding_version = EXCLUDED.embedding_version, is_active = EXCLUDED.is_active, author_type = EXCLUDED.author_type, author_id = EXCLUDED.author_id, published_at = EXCLUDED.published_at, source_version = EXCLUDED.source_version, indexed_at = now(), updated_at = now() WHERE EXCLUDED.source_version IS NULL OR post_embeddings.source_version IS NULL OR EXCLUDED.source_version >= post_embeddings.source_version; END; $$; CREATE OR REPLACE FUNCTION get_place_content_hashes(p_external_ids TEXT[]) RETURNS TABLE ( external_id TEXT, content_hash TEXT ) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ SELECT p.external_id, p.content_hash FROM public.place_embeddings p WHERE p.external_id = ANY(p_external_ids); $$; CREATE OR REPLACE FUNCTION get_post_content_hashes(p_external_ids TEXT[]) RETURNS TABLE ( external_id TEXT, content_hash TEXT ) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ SELECT p.external_id, p.content_hash FROM public.post_embeddings p WHERE p.external_id = ANY(p_external_ids); $$; CREATE OR REPLACE FUNCTION search_resource_embeddings( p_resource_type TEXT, p_query_text TEXT, p_query_embedding VECTOR(300), p_match_count INTEGER, p_filters JSONB DEFAULT '{}'::jsonb ) RETURNS TABLE ( external_id TEXT, document TEXT, metadata JSONB, score DOUBLE PRECISION, semantic_score DOUBLE PRECISION, lexical_score DOUBLE PRECISION ) LANGUAGE plpgsql STABLE SECURITY DEFINER SET search_path = public AS $$ DECLARE target_table TEXT; BEGIN target_table := CASE p_resource_type WHEN 'places' THEN 'place_embeddings' WHEN 'posts' THEN 'post_embeddings' WHEN 'users' THEN 'user_embeddings' WHEN 'clubs' THEN 'club_embeddings' WHEN 'groups' THEN 'group_embeddings' WHEN 'events' THEN 'event_embeddings' ELSE NULL END; IF target_table IS NULL THEN RAISE EXCEPTION 'Unsupported search resource type: %', p_resource_type; END IF; RETURN QUERY EXECUTE format($query$ WITH eligible AS NOT MATERIALIZED ( SELECT e.* FROM public.%I e WHERE e.is_active = true AND COALESCE(($4->>'is_active')::boolean, true) = true AND ( ($4 ? 'city') IS FALSE OR lower(e.metadata->>'city') = lower($4->>'city') ) AND ( ($4 ? 'state') IS FALSE OR lower(e.metadata->>'state') = lower($4->>'state') ) AND ( ($4 ? 'categories') IS FALSE OR EXISTS ( SELECT 1 FROM jsonb_array_elements_text($4->'categories') AS allowed(value) WHERE lower(e.metadata->>'category') = lower(allowed.value) ) ) AND ( ($4 ? 'tags') IS FALSE OR EXISTS ( SELECT 1 FROM jsonb_array_elements_text($4->'tags') AS allowed(value) WHERE lower(COALESCE(e.metadata->>'tags', '')) LIKE ('%%' || lower(allowed.value) || '%%') ) ) AND ( ($4 ? 'published_from') IS FALSE OR NULLIF(e.metadata->>'published_at', '')::timestamptz >= ($4->>'published_from')::timestamptz ) AND ( ($4 ? 'published_to') IS FALSE OR NULLIF(e.metadata->>'published_at', '')::timestamptz <= ($4->>'published_to')::timestamptz ) AND ( ($4 ? 'event_from') IS FALSE OR NULLIF(e.metadata->>'start_time', '')::timestamptz >= ($4->>'event_from')::timestamptz ) AND ( ($4 ? 'event_to') IS FALSE OR NULLIF(e.metadata->>'start_time', '')::timestamptz <= ($4->>'event_to')::timestamptz ) AND ( ($4 ? 'is_online') IS FALSE OR COALESCE((e.metadata->>'is_online')::boolean, false) = ($4->>'is_online')::boolean ) AND ( ($4 ? 'user_roles') IS FALSE OR EXISTS ( SELECT 1 FROM jsonb_array_elements_text($4->'user_roles') AS allowed(value) WHERE lower(e.metadata->>'role') = lower(allowed.value) ) ) AND ( ($4 ? 'place_ids') IS FALSE OR e.metadata->>'place_id' IN ( SELECT jsonb_array_elements_text($4->'place_ids') ) ) AND ( ( COALESCE((e.metadata->>'is_private')::boolean, false) = false AND COALESCE((e.metadata->>'is_public')::boolean, true) = true ) OR ( $4 ? 'requester_id' AND ( e.metadata->>'creator_id' = $4->>'requester_id' OR COALESCE(e.metadata->'authorized_user_ids', '[]'::jsonb) ? ($4->>'requester_id') ) ) ) ), vector_results AS ( SELECT e.external_id, 1 - (e.embedding <=> $1) AS semantic_score, row_number() OVER ( ORDER BY e.embedding <=> $1, e.external_id ASC ) AS vector_rank FROM eligible e ORDER BY e.embedding <=> $1, e.external_id ASC LIMIT GREATEST($3 * 3, 30) ), lexical_results AS ( SELECT e.external_id, ts_rank_cd(e.textsearch, query.value) AS lexical_score, row_number() OVER ( ORDER BY ts_rank_cd(e.textsearch, query.value) DESC, e.external_id ASC ) AS lexical_rank FROM eligible e CROSS JOIN websearch_to_tsquery('simple', $2) AS query(value) WHERE e.textsearch @@ query.value ORDER BY lexical_score DESC, e.external_id ASC LIMIT GREATEST($3 * 3, 30) ), fused AS ( SELECT COALESCE(v.external_id, l.external_id) AS external_id, v.semantic_score, l.lexical_score, COALESCE(1.0 / (60.0 + v.vector_rank), 0.0) + COALESCE(1.0 / (60.0 + l.lexical_rank), 0.0) AS rrf_score FROM vector_results v FULL OUTER JOIN lexical_results l USING (external_id) ) SELECT e.external_id, e.document, e.metadata, LEAST( 1.0, f.rrf_score / (2.0 / 61.0) + CASE WHEN lower(COALESCE( e.metadata->>'search_title', e.metadata->>'name', e.metadata->>'title', e.metadata->>'username', '' )) = lower($2) THEN 0.15 ELSE 0.0 END )::double precision AS score, f.semantic_score::double precision, f.lexical_score::double precision FROM fused f JOIN eligible e USING (external_id) ORDER BY score DESC, f.semantic_score DESC NULLS LAST, e.external_id ASC LIMIT GREATEST($3, 1) $query$, target_table) USING p_query_embedding, p_query_text, p_match_count, p_filters; END; $$; CREATE OR REPLACE FUNCTION get_resource_content_hashes( p_resource_type TEXT, p_external_ids TEXT[] ) RETURNS TABLE (external_id TEXT, content_hash TEXT) LANGUAGE plpgsql STABLE SECURITY DEFINER SET search_path = public AS $$ DECLARE target_table TEXT; BEGIN target_table := CASE p_resource_type WHEN 'users' THEN 'user_embeddings' WHEN 'clubs' THEN 'club_embeddings' WHEN 'groups' THEN 'group_embeddings' WHEN 'events' THEN 'event_embeddings' ELSE NULL END; IF target_table IS NULL THEN RAISE EXCEPTION 'Unsupported sync resource type: %', p_resource_type; END IF; RETURN QUERY EXECUTE format( 'SELECT e.external_id, e.content_hash FROM public.%I e WHERE e.external_id = ANY($1)', target_table ) USING p_external_ids; END; $$; CREATE OR REPLACE FUNCTION upsert_resource_embedding( p_resource_type TEXT, p_external_id TEXT, p_document TEXT, p_metadata JSONB, p_embedding VECTOR(300), p_content_hash TEXT, p_embedding_model TEXT, p_embedding_version TEXT, p_is_active BOOLEAN ) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ DECLARE target_table TEXT; BEGIN target_table := CASE p_resource_type WHEN 'users' THEN 'user_embeddings' WHEN 'clubs' THEN 'club_embeddings' WHEN 'groups' THEN 'group_embeddings' WHEN 'events' THEN 'event_embeddings' ELSE NULL END; IF target_table IS NULL THEN RAISE EXCEPTION 'Unsupported sync resource type: %', p_resource_type; END IF; EXECUTE format($query$ INSERT INTO public.%I ( external_id, document, metadata, embedding, content_hash, embedding_model, embedding_version, is_active, updated_at ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now()) ON CONFLICT (external_id) DO UPDATE SET document = EXCLUDED.document, metadata = EXCLUDED.metadata, embedding = EXCLUDED.embedding, content_hash = EXCLUDED.content_hash, embedding_model = EXCLUDED.embedding_model, embedding_version = EXCLUDED.embedding_version, is_active = EXCLUDED.is_active, updated_at = now() $query$, target_table) USING p_external_id, p_document, p_metadata, p_embedding, p_content_hash, p_embedding_model, p_embedding_version, p_is_active; END; $$; -- Example least-privilege grants. Adjust schema/database/user creation to your setup. -- GRANT EXECUTE ON FUNCTION match_places(VECTOR, INTEGER, JSONB) TO nlp_reader; -- GRANT EXECUTE ON FUNCTION match_posts(VECTOR, INTEGER, JSONB) TO nlp_reader; -- GRANT EXECUTE ON FUNCTION get_place_content_hashes(TEXT[]) TO nlp_writer; -- GRANT EXECUTE ON FUNCTION get_post_content_hashes(TEXT[]) TO nlp_writer; -- GRANT EXECUTE ON FUNCTION upsert_place_embedding(TEXT, TEXT, JSONB, VECTOR, TEXT, TEXT, TEXT, BOOLEAN) TO nlp_writer; -- GRANT EXECUTE ON FUNCTION upsert_post_embedding(TEXT, TEXT, JSONB, VECTOR, TEXT, TEXT, TEXT, BOOLEAN, TEXT, TEXT, TIMESTAMPTZ, BIGINT) TO nlp_writer; -- GRANT EXECUTE ON FUNCTION search_resource_embeddings(TEXT, TEXT, VECTOR, INTEGER, JSONB) TO nlp_reader; -- GRANT EXECUTE ON FUNCTION get_resource_content_hashes(TEXT, TEXT[]) TO nlp_writer; -- GRANT EXECUTE ON FUNCTION upsert_resource_embedding(TEXT, TEXT, TEXT, JSONB, VECTOR, TEXT, TEXT, TEXT, BOOLEAN) TO nlp_writer; -- The feed-specific tables and controlled functions are versioned in -- sql/migrate_post_feed_v1.sql. Apply that companion contract immediately -- after this base contract. Keeping the feed extension versioned prevents a -- fresh setup and an upgraded database from drifting apart; the full psql -- setup includes it automatically with \ir. BEGIN; -- Derived post data. Source-of-truth permissions remain in the Go API. ALTER TABLE public.post_embeddings ADD COLUMN IF NOT EXISTS author_type TEXT, ADD COLUMN IF NOT EXISTS author_id TEXT, ADD COLUMN IF NOT EXISTS published_at TIMESTAMPTZ, ADD COLUMN IF NOT EXISTS source_version BIGINT, ADD COLUMN IF NOT EXISTS indexed_at TIMESTAMPTZ NOT NULL DEFAULT now(); CREATE INDEX IF NOT EXISTS post_embeddings_active_published_idx ON public.post_embeddings (published_at DESC, external_id DESC) WHERE is_active = TRUE; CREATE INDEX IF NOT EXISTS post_embeddings_author_idx ON public.post_embeddings (author_type, author_id) WHERE is_active = TRUE; CREATE TABLE IF NOT EXISTS public.post_cluster_runs ( id UUID PRIMARY KEY, algorithm TEXT NOT NULL CHECK (algorithm = 'kmeans'), algorithm_version TEXT NOT NULL, embedding_model TEXT NOT NULL, embedding_version TEXT NOT NULL, embedding_dimension INTEGER NOT NULL CHECK (embedding_dimension = 300), k INTEGER NOT NULL CHECK (k > 1), sample_size INTEGER NOT NULL CHECK (sample_size >= k), inertia DOUBLE PRECISION, silhouette_score DOUBLE PRECISION, status TEXT NOT NULL CHECK (status IN ('running', 'validated', 'active', 'failed', 'superseded')), parameters JSONB NOT NULL DEFAULT '{}'::jsonb, started_at TIMESTAMPTZ NOT NULL, completed_at TIMESTAMPTZ, activated_at TIMESTAMPTZ, error_message TEXT ); CREATE UNIQUE INDEX IF NOT EXISTS post_cluster_runs_one_active_idx ON public.post_cluster_runs ((status)) WHERE status = 'active'; CREATE TABLE IF NOT EXISTS public.post_clusters ( run_id UUID NOT NULL REFERENCES public.post_cluster_runs(id) ON DELETE CASCADE, cluster_id INTEGER NOT NULL, centroid VECTOR(300) NOT NULL, label TEXT, size INTEGER NOT NULL CHECK (size >= 0), representative_post_ids TEXT[] NOT NULL DEFAULT '{}', metadata JSONB NOT NULL DEFAULT '{}'::jsonb, PRIMARY KEY (run_id, cluster_id) ); CREATE TABLE IF NOT EXISTS public.post_cluster_memberships ( run_id UUID NOT NULL, post_id TEXT NOT NULL REFERENCES public.post_embeddings(external_id) ON DELETE CASCADE, cluster_id INTEGER NOT NULL, distance_to_centroid DOUBLE PRECISION NOT NULL CHECK (distance_to_centroid >= 0), assigned_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (run_id, post_id), FOREIGN KEY (run_id, cluster_id) REFERENCES public.post_clusters(run_id, cluster_id) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS post_cluster_memberships_cluster_idx ON public.post_cluster_memberships (run_id, cluster_id, distance_to_centroid); CREATE TABLE IF NOT EXISTS public.user_interest_embeddings ( user_id TEXT PRIMARY KEY, embedding VECTOR(300) NOT NULL, weighted_sum VECTOR(300) NOT NULL, total_weight DOUBLE PRECISION NOT NULL DEFAULT 0 CHECK (total_weight >= 0), source_hash TEXT NOT NULL, profile_version TEXT NOT NULL, interaction_count INTEGER NOT NULL DEFAULT 0 CHECK (interaction_count >= 0), last_event_id BIGINT NOT NULL DEFAULT 0 CHECK (last_event_id >= 0), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); ALTER TABLE public.user_interest_embeddings ADD COLUMN IF NOT EXISTS weighted_sum VECTOR(300), ADD COLUMN IF NOT EXISTS total_weight DOUBLE PRECISION NOT NULL DEFAULT 0, ADD COLUMN IF NOT EXISTS last_event_id BIGINT NOT NULL DEFAULT 0; UPDATE public.user_interest_embeddings SET weighted_sum = embedding WHERE weighted_sum IS NULL; ALTER TABLE public.user_interest_embeddings ALTER COLUMN weighted_sum SET NOT NULL; CREATE TABLE IF NOT EXISTS public.post_sync_checkpoints ( consumer TEXT PRIMARY KEY, last_event_id BIGINT NOT NULL DEFAULT 0 CHECK (last_event_id >= 0), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE TABLE IF NOT EXISTS public.post_embedding_tombstones ( post_id TEXT PRIMARY KEY, source_version BIGINT NOT NULL CHECK (source_version >= 0), deleted_at TIMESTAMPTZ NOT NULL DEFAULT now() ); -- Replace the legacy eight-argument function with the version-aware contract. DROP FUNCTION IF EXISTS public.upsert_post_embedding( TEXT, TEXT, JSONB, VECTOR, TEXT, TEXT, TEXT, BOOLEAN ); CREATE OR REPLACE FUNCTION public.upsert_post_embedding( p_external_id TEXT, p_document TEXT, p_metadata JSONB, p_embedding VECTOR(300), p_content_hash TEXT, p_embedding_model TEXT, p_embedding_version TEXT, p_is_active BOOLEAN, p_author_type TEXT, p_author_id TEXT, p_published_at TIMESTAMPTZ, p_source_version BIGINT ) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN IF EXISTS ( SELECT 1 FROM public.post_embedding_tombstones tombstone WHERE tombstone.post_id = p_external_id AND (p_source_version IS NULL OR tombstone.source_version >= p_source_version) ) THEN RETURN; END IF; IF p_source_version IS NOT NULL THEN DELETE FROM public.post_embedding_tombstones WHERE post_id = p_external_id AND source_version < p_source_version; END IF; INSERT INTO public.post_embeddings ( external_id, document, metadata, embedding, content_hash, embedding_model, embedding_version, is_active, author_type, author_id, published_at, source_version, indexed_at, updated_at ) VALUES ( p_external_id, p_document, p_metadata, p_embedding, p_content_hash, p_embedding_model, p_embedding_version, p_is_active, p_author_type, p_author_id, p_published_at, p_source_version, now(), now() ) ON CONFLICT (external_id) DO UPDATE SET document = EXCLUDED.document, metadata = EXCLUDED.metadata, embedding = EXCLUDED.embedding, content_hash = EXCLUDED.content_hash, embedding_model = EXCLUDED.embedding_model, embedding_version = EXCLUDED.embedding_version, is_active = EXCLUDED.is_active, author_type = EXCLUDED.author_type, author_id = EXCLUDED.author_id, published_at = EXCLUDED.published_at, source_version = EXCLUDED.source_version, indexed_at = now(), updated_at = now() WHERE EXCLUDED.source_version IS NULL OR post_embeddings.source_version IS NULL OR EXCLUDED.source_version >= post_embeddings.source_version; END; $$; CREATE OR REPLACE FUNCTION public.deactivate_post_embedding( p_external_id TEXT, p_source_version BIGINT ) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN INSERT INTO public.post_embedding_tombstones (post_id, source_version, deleted_at) VALUES (p_external_id, p_source_version, now()) ON CONFLICT (post_id) DO UPDATE SET source_version = GREATEST( post_embedding_tombstones.source_version, EXCLUDED.source_version ), deleted_at = CASE WHEN EXCLUDED.source_version >= post_embedding_tombstones.source_version THEN now() ELSE post_embedding_tombstones.deleted_at END; UPDATE public.post_embeddings SET is_active = FALSE, source_version = GREATEST(COALESCE(source_version, 0), p_source_version), indexed_at = now(), updated_at = now() WHERE external_id = p_external_id AND (source_version IS NULL OR p_source_version >= source_version); END; $$; CREATE OR REPLACE FUNCTION public.get_post_sync_checkpoint(p_consumer TEXT) RETURNS BIGINT LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ SELECT COALESCE( (SELECT last_event_id FROM public.post_sync_checkpoints WHERE consumer = p_consumer), 0 ); $$; CREATE OR REPLACE FUNCTION public.save_post_sync_checkpoint( p_consumer TEXT, p_last_event_id BIGINT ) RETURNS VOID LANGUAGE sql SECURITY DEFINER SET search_path = public AS $$ INSERT INTO public.post_sync_checkpoints (consumer, last_event_id, updated_at) VALUES (p_consumer, p_last_event_id, now()) ON CONFLICT (consumer) DO UPDATE SET last_event_id = GREATEST(post_sync_checkpoints.last_event_id, EXCLUDED.last_event_id), updated_at = now(); $$; CREATE OR REPLACE FUNCTION public.reset_post_sync_checkpoint(p_consumer TEXT) RETURNS VOID LANGUAGE sql SECURITY DEFINER SET search_path = public AS $$ DELETE FROM public.post_sync_checkpoints WHERE consumer = p_consumer; $$; CREATE OR REPLACE FUNCTION public.get_post_feed_features( p_user_id TEXT, p_post_ids TEXT[] ) RETURNS TABLE ( external_id TEXT, semantic_similarity DOUBLE PRECISION, cluster_id INTEGER, cluster_run_id TEXT, cold_start BOOLEAN, post_embedding TEXT ) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ WITH requested AS ( SELECT post_id, ordinality FROM unnest(p_post_ids) WITH ORDINALITY AS ids(post_id, ordinality) ), active_run AS ( SELECT id FROM public.post_cluster_runs WHERE status = 'active' ORDER BY activated_at DESC NULLS LAST LIMIT 1 ), profile AS ( SELECT embedding FROM public.user_interest_embeddings WHERE user_id = p_user_id ) SELECT requested.post_id, CASE WHEN profile.embedding IS NULL OR post.embedding IS NULL THEN 0.0 ELSE 1 - (post.embedding <=> profile.embedding) END, membership.cluster_id, active_run.id::text, profile.embedding IS NULL, post.embedding::text FROM requested LEFT JOIN public.post_embeddings post ON post.external_id = requested.post_id AND post.is_active = TRUE LEFT JOIN profile ON TRUE LEFT JOIN active_run ON TRUE LEFT JOIN public.post_cluster_memberships membership ON membership.post_id = post.external_id AND membership.run_id = active_run.id ORDER BY requested.ordinality; $$; CREATE OR REPLACE FUNCTION public.get_post_embeddings_for_profile(p_post_ids TEXT[]) RETURNS TABLE (post_id TEXT, embedding VECTOR(300)) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ SELECT external_id, embedding FROM public.post_embeddings WHERE external_id = ANY(p_post_ids) AND is_active = TRUE; $$; CREATE OR REPLACE FUNCTION public.get_user_interest_states(p_user_ids TEXT[]) RETURNS TABLE ( user_id TEXT, embedding VECTOR(300), weighted_sum VECTOR(300), total_weight DOUBLE PRECISION, interaction_count INTEGER, last_event_id BIGINT ) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ SELECT profile.user_id, profile.embedding, profile.weighted_sum, profile.total_weight, profile.interaction_count, profile.last_event_id FROM public.user_interest_embeddings profile WHERE profile.user_id = ANY(p_user_ids); $$; CREATE OR REPLACE FUNCTION public.upsert_user_interest_state( p_user_id TEXT, p_embedding VECTOR(300), p_weighted_sum VECTOR(300), p_total_weight DOUBLE PRECISION, p_interaction_count INTEGER, p_last_event_id BIGINT, p_profile_version TEXT, p_source_hash TEXT ) RETURNS VOID LANGUAGE sql SECURITY DEFINER SET search_path = public AS $$ INSERT INTO public.user_interest_embeddings ( user_id, embedding, weighted_sum, total_weight, source_hash, profile_version, interaction_count, last_event_id, updated_at ) VALUES ( p_user_id, p_embedding, p_weighted_sum, p_total_weight, p_source_hash, p_profile_version, p_interaction_count, p_last_event_id, now() ) ON CONFLICT (user_id) DO UPDATE SET embedding = EXCLUDED.embedding, weighted_sum = EXCLUDED.weighted_sum, total_weight = EXCLUDED.total_weight, source_hash = EXCLUDED.source_hash, profile_version = EXCLUDED.profile_version, interaction_count = EXCLUDED.interaction_count, last_event_id = EXCLUDED.last_event_id, updated_at = now() WHERE EXCLUDED.last_event_id >= user_interest_embeddings.last_event_id; $$; CREATE OR REPLACE FUNCTION public.reset_user_interest_profiles(p_user_id TEXT) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN IF p_user_id IS NULL THEN DELETE FROM public.user_interest_embeddings; ELSE DELETE FROM public.user_interest_embeddings WHERE user_id = p_user_id; END IF; END; $$; CREATE OR REPLACE FUNCTION public.list_post_embeddings_for_clustering( p_lookback_days INTEGER, p_embedding_model TEXT, p_embedding_version TEXT ) RETURNS TABLE (post_id TEXT, embedding VECTOR(300)) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ SELECT external_id, embedding FROM public.post_embeddings WHERE is_active = TRUE AND published_at >= now() - (p_lookback_days * interval '1 day') AND embedding_model = p_embedding_model AND embedding_version = p_embedding_version ORDER BY external_id; $$; CREATE OR REPLACE FUNCTION public.create_post_cluster_run( p_id UUID, p_algorithm_version TEXT, p_embedding_model TEXT, p_embedding_version TEXT, p_embedding_dimension INTEGER, p_k INTEGER, p_sample_size INTEGER, p_parameters JSONB ) RETURNS VOID LANGUAGE sql SECURITY DEFINER SET search_path = public AS $$ INSERT INTO public.post_cluster_runs ( id, algorithm, algorithm_version, embedding_model, embedding_version, embedding_dimension, k, sample_size, parameters, status, started_at ) VALUES ( p_id, 'kmeans', p_algorithm_version, p_embedding_model, p_embedding_version, p_embedding_dimension, p_k, p_sample_size, p_parameters, 'running', now() ); $$; CREATE OR REPLACE FUNCTION public.validate_post_cluster_run( p_run_id UUID, p_k INTEGER, p_inertia DOUBLE PRECISION, p_silhouette_score DOUBLE PRECISION, p_parameters JSONB ) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN UPDATE public.post_cluster_runs SET k = p_k, inertia = p_inertia, silhouette_score = p_silhouette_score, parameters = parameters || p_parameters, status = 'validated', completed_at = now(), error_message = NULL WHERE id = p_run_id AND status = 'running'; IF NOT FOUND THEN RAISE EXCEPTION 'cluster run % no existe o no esta running', p_run_id; END IF; END; $$; CREATE OR REPLACE FUNCTION public.fail_post_cluster_run( p_run_id UUID, p_error_message TEXT ) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN UPDATE public.post_cluster_runs SET status = 'failed', completed_at = now(), error_message = LEFT(p_error_message, 4000) WHERE id = p_run_id AND status IN ('running', 'validated'); END; $$; CREATE OR REPLACE FUNCTION public.replace_post_cluster_artifacts( p_run_id UUID, p_clusters JSONB, p_memberships JSONB ) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN DELETE FROM public.post_cluster_memberships WHERE run_id = p_run_id; DELETE FROM public.post_clusters WHERE run_id = p_run_id; IF NOT EXISTS ( SELECT 1 FROM public.post_cluster_runs WHERE id = p_run_id AND status = 'validated' ) THEN RAISE EXCEPTION 'cluster run % no esta validado', p_run_id; END IF; INSERT INTO public.post_clusters ( run_id, cluster_id, centroid, size, representative_post_ids ) SELECT p_run_id, item.cluster_id, item.centroid::vector(300), item.size, COALESCE(item.representative_post_ids, '{}') FROM jsonb_to_recordset(p_clusters) AS item( cluster_id INTEGER, centroid TEXT, size INTEGER, representative_post_ids TEXT[] ); INSERT INTO public.post_cluster_memberships ( run_id, post_id, cluster_id, distance_to_centroid ) SELECT p_run_id, item.post_id, item.cluster_id, item.distance FROM jsonb_to_recordset(p_memberships) AS item( post_id TEXT, cluster_id INTEGER, distance DOUBLE PRECISION ); END; $$; CREATE OR REPLACE FUNCTION public.activate_post_cluster_run(p_run_id UUID) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN IF EXISTS ( SELECT 1 FROM public.post_cluster_runs WHERE id = p_run_id AND status = 'active' ) THEN RETURN; END IF; IF NOT EXISTS ( SELECT 1 FROM public.post_cluster_runs WHERE id = p_run_id AND status IN ('validated', 'superseded') ) THEN RAISE EXCEPTION 'cluster run % no existe o no es activable', p_run_id; END IF; UPDATE public.post_cluster_runs SET status = 'superseded' WHERE status = 'active'; UPDATE public.post_cluster_runs SET status = 'active', activated_at = now() WHERE id = p_run_id; END; $$; CREATE OR REPLACE FUNCTION public.get_post_cluster_run(p_run_id UUID) RETURNS TABLE ( run_id UUID, status TEXT, k INTEGER, sample_size INTEGER, inertia DOUBLE PRECISION, silhouette_score DOUBLE PRECISION, started_at TIMESTAMPTZ, completed_at TIMESTAMPTZ, activated_at TIMESTAMPTZ, error_message TEXT ) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ SELECT id, post_cluster_runs.status, post_cluster_runs.k, post_cluster_runs.sample_size, post_cluster_runs.inertia, post_cluster_runs.silhouette_score, post_cluster_runs.started_at, post_cluster_runs.completed_at, post_cluster_runs.activated_at, post_cluster_runs.error_message FROM public.post_cluster_runs WHERE id = p_run_id; $$; CREATE OR REPLACE FUNCTION public.assign_posts_to_active_cluster(p_post_ids TEXT[]) RETURNS INTEGER LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ DECLARE affected INTEGER; BEGIN WITH active_run AS ( SELECT id, embedding_model, embedding_version FROM public.post_cluster_runs WHERE status = 'active' ORDER BY activated_at DESC NULLS LAST LIMIT 1 ), assignments AS ( SELECT active_run.id AS run_id, post.external_id AS post_id, nearest.cluster_id, nearest.distance FROM active_run JOIN public.post_embeddings post ON post.external_id = ANY(p_post_ids) AND post.is_active = TRUE AND post.embedding_model = active_run.embedding_model AND post.embedding_version = active_run.embedding_version CROSS JOIN LATERAL ( SELECT cluster.cluster_id, (post.embedding <=> cluster.centroid) AS distance FROM public.post_clusters cluster WHERE cluster.run_id = active_run.id ORDER BY post.embedding <=> cluster.centroid LIMIT 1 ) nearest ) INSERT INTO public.post_cluster_memberships ( run_id, post_id, cluster_id, distance_to_centroid, assigned_at ) SELECT run_id, post_id, cluster_id, distance, now() FROM assignments ON CONFLICT (run_id, post_id) DO UPDATE SET cluster_id = EXCLUDED.cluster_id, distance_to_centroid = EXCLUDED.distance_to_centroid, assigned_at = now(); GET DIAGNOSTICS affected = ROW_COUNT; RETURN affected; END; $$; CREATE OR REPLACE FUNCTION public.get_post_cluster_status() RETURNS TABLE ( active_run_id UUID, k INTEGER, sample_size INTEGER, silhouette_score DOUBLE PRECISION, activated_at TIMESTAMPTZ ) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ SELECT id, post_cluster_runs.k, post_cluster_runs.sample_size, post_cluster_runs.silhouette_score, post_cluster_runs.activated_at FROM public.post_cluster_runs WHERE status = 'active' ORDER BY activated_at DESC NULLS LAST LIMIT 1; $$; -- Least-privilege grants when the expected roles exist. REVOKE ALL ON FUNCTION public.upsert_post_embedding(TEXT, TEXT, JSONB, VECTOR, TEXT, TEXT, TEXT, BOOLEAN, TEXT, TEXT, TIMESTAMPTZ, BIGINT) FROM PUBLIC; REVOKE ALL ON FUNCTION public.deactivate_post_embedding(TEXT, BIGINT) FROM PUBLIC; REVOKE ALL ON FUNCTION public.get_post_sync_checkpoint(TEXT) FROM PUBLIC; REVOKE ALL ON FUNCTION public.save_post_sync_checkpoint(TEXT, BIGINT) FROM PUBLIC; REVOKE ALL ON FUNCTION public.reset_post_sync_checkpoint(TEXT) FROM PUBLIC; REVOKE ALL ON FUNCTION public.get_post_feed_features(TEXT, TEXT[]) FROM PUBLIC; REVOKE ALL ON FUNCTION public.get_post_embeddings_for_profile(TEXT[]) FROM PUBLIC; REVOKE ALL ON FUNCTION public.get_user_interest_states(TEXT[]) FROM PUBLIC; REVOKE ALL ON FUNCTION public.upsert_user_interest_state(TEXT, VECTOR, VECTOR, DOUBLE PRECISION, INTEGER, BIGINT, TEXT, TEXT) FROM PUBLIC; REVOKE ALL ON FUNCTION public.reset_user_interest_profiles(TEXT) FROM PUBLIC; REVOKE ALL ON FUNCTION public.list_post_embeddings_for_clustering(INTEGER, TEXT, TEXT) FROM PUBLIC; REVOKE ALL ON FUNCTION public.create_post_cluster_run(UUID, TEXT, TEXT, TEXT, INTEGER, INTEGER, INTEGER, JSONB) FROM PUBLIC; REVOKE ALL ON FUNCTION public.validate_post_cluster_run(UUID, INTEGER, DOUBLE PRECISION, DOUBLE PRECISION, JSONB) FROM PUBLIC; REVOKE ALL ON FUNCTION public.fail_post_cluster_run(UUID, TEXT) FROM PUBLIC; REVOKE ALL ON FUNCTION public.replace_post_cluster_artifacts(UUID, JSONB, JSONB) FROM PUBLIC; REVOKE ALL ON FUNCTION public.activate_post_cluster_run(UUID) FROM PUBLIC; REVOKE ALL ON FUNCTION public.assign_posts_to_active_cluster(TEXT[]) FROM PUBLIC; REVOKE ALL ON FUNCTION public.get_post_cluster_status() FROM PUBLIC; REVOKE ALL ON FUNCTION public.get_post_cluster_run(UUID) FROM PUBLIC; DO $$ BEGIN IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'nlp_reader') THEN GRANT EXECUTE ON FUNCTION public.get_post_feed_features(TEXT, TEXT[]) TO nlp_reader; GRANT EXECUTE ON FUNCTION public.get_post_cluster_status() TO nlp_reader; GRANT EXECUTE ON FUNCTION public.get_post_cluster_run(UUID) TO nlp_reader; END IF; IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'nlp_writer') THEN GRANT EXECUTE ON FUNCTION public.upsert_post_embedding(TEXT, TEXT, JSONB, VECTOR, TEXT, TEXT, TEXT, BOOLEAN, TEXT, TEXT, TIMESTAMPTZ, BIGINT) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.deactivate_post_embedding(TEXT, BIGINT) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.get_post_sync_checkpoint(TEXT) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.save_post_sync_checkpoint(TEXT, BIGINT) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.reset_post_sync_checkpoint(TEXT) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.get_post_embeddings_for_profile(TEXT[]) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.get_user_interest_states(TEXT[]) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.upsert_user_interest_state(TEXT, VECTOR, VECTOR, DOUBLE PRECISION, INTEGER, BIGINT, TEXT, TEXT) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.reset_user_interest_profiles(TEXT) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.list_post_embeddings_for_clustering(INTEGER, TEXT, TEXT) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.create_post_cluster_run(UUID, TEXT, TEXT, TEXT, INTEGER, INTEGER, INTEGER, JSONB) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.validate_post_cluster_run(UUID, INTEGER, DOUBLE PRECISION, DOUBLE PRECISION, JSONB) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.fail_post_cluster_run(UUID, TEXT) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.replace_post_cluster_artifacts(UUID, JSONB, JSONB) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.activate_post_cluster_run(UUID) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.assign_posts_to_active_cluster(TEXT[]) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.get_post_cluster_status() TO nlp_writer; GRANT EXECUTE ON FUNCTION public.get_post_cluster_run(UUID) TO nlp_writer; END IF; END $$; COMMIT; BEGIN; DO $$ DECLARE actual_type TEXT; BEGIN SELECT format_type(attribute.atttypid, attribute.atttypmod) INTO actual_type FROM pg_attribute attribute WHERE attribute.attrelid = 'public.post_embeddings'::regclass AND attribute.attname = 'embedding' AND NOT attribute.attisdropped; IF actual_type IS DISTINCT FROM 'vector(300)' THEN RAISE EXCEPTION 'post_embeddings.embedding debe ser vector(300), actual=%', actual_type; END IF; IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'nlp_owner') OR NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'nlp_reader') OR NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'nlp_writer') THEN RAISE EXCEPTION 'se requieren los roles nlp_owner, nlp_reader y nlp_writer'; END IF; END $$; CREATE OR REPLACE FUNCTION public.upsert_post_embedding( p_external_id TEXT, p_document TEXT, p_metadata JSONB, p_embedding VECTOR(300), p_content_hash TEXT, p_embedding_model TEXT, p_embedding_version TEXT, p_is_active BOOLEAN, p_author_type TEXT, p_author_id TEXT, p_published_at TIMESTAMPTZ, p_source_version BIGINT ) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ DECLARE current_version BIGINT; current_hash TEXT; BEGIN IF p_source_version IS NULL OR p_source_version < 0 THEN RAISE EXCEPTION 'source_version es obligatorio y no negativo para %', p_external_id; END IF; SELECT source_version, content_hash INTO current_version, current_hash FROM public.post_embeddings WHERE external_id = p_external_id; IF FOUND AND current_version IS NOT NULL THEN IF current_version > p_source_version THEN RETURN; END IF; IF current_version = p_source_version THEN IF current_hash = p_content_hash THEN RETURN; END IF; RAISE EXCEPTION 'conflicto: misma source_version con contenido distinto para %', p_external_id; END IF; END IF; IF EXISTS ( SELECT 1 FROM public.post_embedding_tombstones tombstone WHERE tombstone.post_id = p_external_id AND tombstone.source_version >= p_source_version ) THEN RETURN; END IF; DELETE FROM public.post_embedding_tombstones WHERE post_id = p_external_id AND source_version < p_source_version; INSERT INTO public.post_embeddings ( external_id, document, metadata, embedding, content_hash, embedding_model, embedding_version, is_active, author_type, author_id, published_at, source_version, indexed_at, updated_at ) VALUES ( p_external_id, p_document, p_metadata, p_embedding, p_content_hash, p_embedding_model, p_embedding_version, p_is_active, p_author_type, p_author_id, p_published_at, p_source_version, now(), now() ) ON CONFLICT (external_id) DO UPDATE SET document = EXCLUDED.document, metadata = EXCLUDED.metadata, embedding = EXCLUDED.embedding, content_hash = EXCLUDED.content_hash, embedding_model = EXCLUDED.embedding_model, embedding_version = EXCLUDED.embedding_version, is_active = EXCLUDED.is_active, author_type = EXCLUDED.author_type, author_id = EXCLUDED.author_id, published_at = EXCLUDED.published_at, source_version = EXCLUDED.source_version, indexed_at = now(), updated_at = now() WHERE post_embeddings.source_version IS NULL OR EXCLUDED.source_version > post_embeddings.source_version; END; $$; CREATE OR REPLACE FUNCTION public.deactivate_post_embedding( p_external_id TEXT, p_source_version BIGINT ) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ BEGIN IF p_source_version IS NULL OR p_source_version < 0 THEN RAISE EXCEPTION 'source_version es obligatorio y no negativo para %', p_external_id; END IF; INSERT INTO public.post_embedding_tombstones (post_id, source_version, deleted_at) VALUES (p_external_id, p_source_version, now()) ON CONFLICT (post_id) DO UPDATE SET source_version = GREATEST(post_embedding_tombstones.source_version, EXCLUDED.source_version), deleted_at = CASE WHEN EXCLUDED.source_version >= post_embedding_tombstones.source_version THEN now() ELSE post_embedding_tombstones.deleted_at END; UPDATE public.post_embeddings SET is_active = FALSE, source_version = GREATEST(COALESCE(source_version, 0), p_source_version), indexed_at = now(), updated_at = now() WHERE external_id = p_external_id AND (source_version IS NULL OR p_source_version >= source_version); END; $$; CREATE OR REPLACE FUNCTION public.get_post_content_hashes(p_external_ids TEXT[]) RETURNS TABLE (external_id TEXT, content_hash TEXT) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ SELECT post.external_id, post.content_hash FROM public.post_embeddings post WHERE post.external_id = ANY(p_external_ids) AND post.is_active = TRUE AND NOT EXISTS ( SELECT 1 FROM public.post_embedding_tombstones tombstone WHERE tombstone.post_id = post.external_id AND tombstone.source_version >= COALESCE(post.source_version, 0) ); $$; CREATE OR REPLACE FUNCTION public.get_post_feed_features( p_user_id TEXT, p_post_ids TEXT[] ) RETURNS TABLE ( external_id TEXT, semantic_similarity DOUBLE PRECISION, cluster_id INTEGER, cluster_run_id TEXT, cold_start BOOLEAN, post_embedding TEXT ) LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$ WITH requested AS ( SELECT post_id, ordinality FROM unnest(p_post_ids) WITH ORDINALITY AS ids(post_id, ordinality) ), active_run AS ( SELECT id FROM public.post_cluster_runs WHERE status = 'active' ORDER BY activated_at DESC NULLS LAST LIMIT 1 ), profile AS ( SELECT embedding FROM public.user_interest_embeddings WHERE user_id = p_user_id ) SELECT requested.post_id, CASE WHEN profile.embedding IS NULL OR post.embedding IS NULL THEN 0.0 ELSE 1 - (post.embedding <=> profile.embedding) END, membership.cluster_id, active_run.id::text, (profile.embedding IS NULL OR post.embedding IS NULL), post.embedding::text FROM requested LEFT JOIN public.post_embeddings post ON post.external_id = requested.post_id AND post.is_active = TRUE LEFT JOIN profile ON TRUE LEFT JOIN active_run ON TRUE LEFT JOIN public.post_cluster_memberships membership ON membership.post_id = post.external_id AND membership.run_id = active_run.id ORDER BY requested.ordinality; $$; CREATE OR REPLACE FUNCTION public.activate_post_cluster_run(p_run_id UUID) RETURNS VOID LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ DECLARE expected_k INTEGER; expected_sample_size INTEGER; cluster_count INTEGER; membership_count INTEGER; declared_size INTEGER; BEGIN PERFORM pg_advisory_xact_lock(hashtext('frimeet-post-cluster-activation')); SELECT k, sample_size INTO expected_k, expected_sample_size FROM public.post_cluster_runs WHERE id = p_run_id AND status IN ('validated', 'superseded', 'active') FOR UPDATE; IF NOT FOUND THEN RAISE EXCEPTION 'cluster run % no existe o no es activable', p_run_id; END IF; SELECT count(*), COALESCE(sum(size), 0) INTO cluster_count, declared_size FROM public.post_clusters WHERE run_id = p_run_id; SELECT count(*) INTO membership_count FROM public.post_cluster_memberships WHERE run_id = p_run_id; IF cluster_count <> expected_k OR membership_count <> expected_sample_size OR declared_size <> expected_sample_size THEN RAISE EXCEPTION 'cluster run % tiene artefactos incompletos: clusters=%/% memberships=%/% size=%/%', p_run_id, cluster_count, expected_k, membership_count, expected_sample_size, declared_size, expected_sample_size; END IF; IF EXISTS (SELECT 1 FROM public.post_cluster_runs WHERE id = p_run_id AND status = 'active') THEN RETURN; END IF; UPDATE public.post_cluster_runs SET status = 'superseded' WHERE status = 'active'; UPDATE public.post_cluster_runs SET status = 'active', activated_at = now() WHERE id = p_run_id; END; $$; ALTER FUNCTION public.upsert_post_embedding(text, text, jsonb, vector, text, text, text, boolean, text, text, timestamptz, bigint) OWNER TO nlp_owner; ALTER FUNCTION public.deactivate_post_embedding(text, bigint) OWNER TO nlp_owner; ALTER FUNCTION public.get_post_content_hashes(text[]) OWNER TO nlp_owner; ALTER FUNCTION public.get_post_feed_features(text, text[]) OWNER TO nlp_owner; ALTER FUNCTION public.activate_post_cluster_run(uuid) OWNER TO nlp_owner; REVOKE ALL ON FUNCTION public.upsert_post_embedding(text, text, jsonb, vector, text, text, text, boolean, text, text, timestamptz, bigint) FROM PUBLIC; REVOKE ALL ON FUNCTION public.deactivate_post_embedding(text, bigint) FROM PUBLIC; REVOKE ALL ON FUNCTION public.get_post_content_hashes(text[]) FROM PUBLIC; REVOKE ALL ON FUNCTION public.get_post_feed_features(text, text[]) FROM PUBLIC; REVOKE ALL ON FUNCTION public.activate_post_cluster_run(uuid) FROM PUBLIC; GRANT EXECUTE ON FUNCTION public.get_post_feed_features(text, text[]) TO nlp_reader; GRANT EXECUTE ON FUNCTION public.upsert_post_embedding(text, text, jsonb, vector, text, text, text, boolean, text, text, timestamptz, bigint) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.deactivate_post_embedding(text, bigint) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.get_post_content_hashes(text[]) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.activate_post_cluster_run(uuid) TO nlp_writer; COMMIT; -- Reconciliacion final de propietarios y minimo privilegio. DO $security$ DECLARE item RECORD; BEGIN FOR item IN SELECT format('%I.%I', table_schema, table_name) AS object_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name = ANY(ARRAY[ 'place_embeddings','post_embeddings','user_embeddings', 'club_embeddings','group_embeddings','event_embeddings', 'post_cluster_runs','post_clusters','post_cluster_memberships', 'user_interest_embeddings','post_sync_checkpoints', 'post_embedding_tombstones' ]) LOOP EXECUTE format('ALTER TABLE %s OWNER TO nlp_owner', item.object_name); EXECUTE format('REVOKE ALL ON TABLE %s FROM PUBLIC', item.object_name); END LOOP; FOR item IN SELECT procedure.oid::regprocedure AS signature FROM pg_proc procedure JOIN pg_namespace namespace ON namespace.oid = procedure.pronamespace WHERE namespace.nspname = 'public' AND procedure.proname = ANY(ARRAY[ 'match_places','match_posts','upsert_place_embedding', 'upsert_post_embedding','get_place_content_hashes', 'get_post_content_hashes','search_resource_embeddings', 'get_resource_content_hashes','upsert_resource_embedding', 'deactivate_post_embedding','get_post_sync_checkpoint', 'save_post_sync_checkpoint','reset_post_sync_checkpoint', 'get_post_feed_features','get_post_embeddings_for_profile', 'get_user_interest_states','upsert_user_interest_state', 'reset_user_interest_profiles','create_post_cluster_run', 'validate_post_cluster_run','replace_post_cluster_artifacts', 'fail_post_cluster_run','get_post_cluster_status', 'get_post_cluster_run','activate_post_cluster_run', 'assign_posts_to_active_cluster','list_post_embeddings_for_clustering' ]) LOOP EXECUTE format('ALTER FUNCTION %s OWNER TO nlp_owner', item.signature); EXECUTE format('REVOKE ALL ON FUNCTION %s FROM PUBLIC', item.signature); END LOOP; END $security$; GRANT USAGE ON SCHEMA public TO nlp_reader, nlp_writer; GRANT EXECUTE ON FUNCTION public.match_places(vector, integer, jsonb) TO nlp_reader; GRANT EXECUTE ON FUNCTION public.match_posts(vector, integer, jsonb) TO nlp_reader; GRANT EXECUTE ON FUNCTION public.search_resource_embeddings(text, text, vector, integer, jsonb) TO nlp_reader; GRANT EXECUTE ON FUNCTION public.get_post_feed_features(text, text[]) TO nlp_reader; GRANT EXECUTE ON FUNCTION public.upsert_place_embedding(text, text, jsonb, vector, text, text, text, boolean) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.upsert_post_embedding(text, text, jsonb, vector, text, text, text, boolean, text, text, timestamptz, bigint) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.get_place_content_hashes(text[]) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.get_post_content_hashes(text[]) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.get_resource_content_hashes(text, text[]) TO nlp_writer; GRANT EXECUTE ON FUNCTION public.upsert_resource_embedding(text, text, text, jsonb, vector, text, text, text, boolean) TO nlp_writer; -- Configura contrasenas fuera de este archivo: -- ALTER ROLE nlp_reader PASSWORD ''; -- ALTER ROLE nlp_writer PASSWORD '';