use sea_orm_migration::prelude::*;
use sea_orm_migration::sea_orm::ConnectionTrait;
use super::FTS_CONFIG;
pub const EMBEDDING_DIMENSION: u32 = 384;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
let backend = manager.get_database_backend();
if backend != sea_orm::DatabaseBackend::Postgres {
return Err(DbErr::Custom(format!(
"graph-storage requires PostgreSQL; refusing to migrate {backend:?}"
)));
}
manager
.get_connection()
.execute_unprepared(&format!(
r"
CREATE EXTENSION IF NOT EXISTS vector SCHEMA public;
-- Interned GTS types: a cache of the authoritative registry with a foreign
-- identity. Types are rows, not enums: registration is an API call.
CREATE TABLE IF NOT EXISTS gts_type (
tenant_id UUID NOT NULL,
id INTEGER NOT NULL GENERATED BY DEFAULT AS IDENTITY,
gts_type_uuid UUID NOT NULL,
gts_type_id TEXT NOT NULL,
kind TEXT NOT NULL CHECK (kind IN ('node', 'edge', 'attribute')),
type_schema JSONB NOT NULL DEFAULT '{{}}'::jsonb,
effective_traits JSONB NOT NULL DEFAULT '{{}}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (tenant_id, id),
UNIQUE (tenant_id, gts_type_uuid),
UNIQUE (tenant_id, gts_type_id)
);
CREATE TABLE IF NOT EXISTS node (
tenant_id UUID NOT NULL,
id BIGINT NOT NULL GENERATED BY DEFAULT AS IDENTITY,
node_key TEXT NOT NULL,
gts_node_type_id INTEGER NOT NULL,
name TEXT NOT NULL DEFAULT '',
payload JSONB NOT NULL DEFAULT '{{}}'::jsonb,
search_text TEXT NOT NULL DEFAULT '',
search TSVECTOR GENERATED ALWAYS AS (to_tsvector('{FTS_CONFIG}', search_text)) STORED,
embedding VECTOR({EMBEDDING_DIMENSION}),
embedding_epoch BIGINT,
embedding_input_hash TEXT,
source_namespace TEXT,
owner_principal TEXT NOT NULL DEFAULT '',
created_by TEXT NOT NULL DEFAULT '',
version BIGINT NOT NULL DEFAULT 1,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
deleted_at TIMESTAMPTZ,
PRIMARY KEY (tenant_id, id),
UNIQUE (tenant_id, node_key),
FOREIGN KEY (tenant_id, gts_node_type_id) REFERENCES gts_type (tenant_id, id)
);
-- Endpoint foreign keys are RESTRICT, never CASCADE: deleting a static node
-- must not silently destroy analysis edges attached to it. Note: PostgreSQL
-- 18+ reports a RESTRICT refusal as SQLSTATE 23001 (restrict_violation),
-- not 23503 — the error classifier accepts both.
CREATE TABLE IF NOT EXISTS edge (
tenant_id UUID NOT NULL,
id BIGINT NOT NULL GENERATED BY DEFAULT AS IDENTITY,
edge_key TEXT NOT NULL,
gts_edge_type_id INTEGER NOT NULL,
src_node_id BIGINT NOT NULL,
dst_node_id BIGINT NOT NULL,
discriminator TEXT,
payload JSONB NOT NULL DEFAULT '{{}}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
deleted_at TIMESTAMPTZ,
PRIMARY KEY (tenant_id, id),
UNIQUE (tenant_id, edge_key),
FOREIGN KEY (tenant_id, gts_edge_type_id) REFERENCES gts_type (tenant_id, id),
FOREIGN KEY (tenant_id, src_node_id) REFERENCES node (tenant_id, id) ON DELETE RESTRICT,
FOREIGN KEY (tenant_id, dst_node_id) REFERENCES node (tenant_id, id) ON DELETE RESTRICT
);
-- Per-tenant metadata. Two normative keys: graph_revision (per tenant) and
-- source_epoch (deployment-wide, stored under the nil tenant).
CREATE TABLE IF NOT EXISTS graph_meta (
tenant_id UUID NOT NULL,
key TEXT NOT NULL,
value JSONB NOT NULL,
PRIMARY KEY (tenant_id, key)
);
-- Idempotency receipts commit in the same transaction as the batch.
CREATE TABLE IF NOT EXISTS ingest_idempotency (
tenant_id UUID NOT NULL,
producer TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
request_hash TEXT NOT NULL,
source_epoch BIGINT NOT NULL,
graph_revision BIGINT NOT NULL,
response JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (tenant_id, producer, idempotency_key)
);
-- Scope replacement registry: the single-writer lock row and the generation
-- fence per canonical scope identity.
CREATE TABLE IF NOT EXISTS scope_registry (
tenant_id UUID NOT NULL,
scope_attribute TEXT NOT NULL,
scope_value TEXT NOT NULL,
owner_producer TEXT NOT NULL,
generation BIGINT NOT NULL,
request_hash TEXT NOT NULL,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (tenant_id, scope_attribute, scope_value)
);
-- Traversal backbone: one composite index per direction, live rows only.
CREATE INDEX IF NOT EXISTS idx_edge_src
ON edge (tenant_id, src_node_id) WHERE deleted_at IS NULL;
CREATE INDEX IF NOT EXISTS idx_edge_dst
ON edge (tenant_id, dst_node_id) WHERE deleted_at IS NULL;
CREATE INDEX IF NOT EXISTS idx_edge_type
ON edge (tenant_id, gts_edge_type_id) WHERE deleted_at IS NULL;
CREATE INDEX IF NOT EXISTS idx_node_type
ON node (tenant_id, gts_node_type_id) WHERE deleted_at IS NULL;
-- Lexical arm: GIN over the generated tsvector, live rows only.
CREATE INDEX IF NOT EXISTS idx_node_search
ON node USING gin (search) WHERE deleted_at IS NULL;
-- Vector arm: HNSW cosine, live embedded rows only.
CREATE INDEX IF NOT EXISTS idx_node_embedding
ON node USING hnsw (embedding vector_cosine_ops)
WHERE deleted_at IS NULL AND embedding IS NOT NULL;
",
))
.await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager
.get_connection()
.execute_unprepared(
"DROP TABLE IF EXISTS scope_registry; \
DROP TABLE IF EXISTS ingest_idempotency; \
DROP TABLE IF EXISTS graph_meta; \
DROP TABLE IF EXISTS edge; \
DROP TABLE IF EXISTS node; \
DROP TABLE IF EXISTS gts_type;",
)
.await?;
Ok(())
}
}