refactor: rename memory_entity→knowledge_node, knowledge graph edge→knowledge_edge
CI / CI (pull_request) Failing after 11m15s

Resolves table name collision between:
- memory_edge (provenance DAG: child_sha/parent_sha) — KEPT
- knowledge_edge (knowledge graph: source_id/target_id) — NEW NAME

Changes:
- memory_entity → knowledge_node (all .rs + migrations 003-009)
- knowledge graph memory_edge → knowledge_edge
- memory_entity_version → knowledge_node_version
- memory_edge_version → knowledge_edge_version
- Added knowledge_node + knowledge_edge to init_schema()
- Converted versioning.rs from sqlx::query_as! to runtime queries
  (avoids stale sqlx offline cache dependency)
- Fixed UUID cast: $1::UUID for String→UUID column binds
- Fixed column names: source_entity_id→source_id, target_entity_id→target_id

Production DB: knowledge_node + knowledge_edge tables created,
memory_entity VIEW points to knowledge_node for backward compat.
This commit is contained in:
2026-09-15 15:22:41 +09:00
parent b514c43d6b
commit 7d49a2aaef
36 changed files with 263 additions and 283 deletions
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1\n ORDER BY version_num DESC\n ",
"query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_node_version\n WHERE entity_id = $1\n ORDER BY version_num DESC\n ",
"describe": {
"columns": [
{
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_edge_version\n WHERE edge_id = $1\n ORDER BY version_num DESC\n ",
"query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_edge_version\n WHERE edge_id = $1\n ORDER BY version_num DESC\n ",
"describe": {
"columns": [
{
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1 AND changed_at <= $2\n ORDER BY version_num DESC\n LIMIT 1\n ",
"query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_node_version\n WHERE entity_id = $1 AND changed_at <= $2\n ORDER BY version_num DESC\n LIMIT 1\n ",
"describe": {
"columns": [
{
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_entity_version\n WHERE entity_id = $1 AND version_num = $2\n ",
"query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_node_version\n WHERE entity_id = $1 AND version_num = $2\n ",
"describe": {
"columns": [
{
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM memory_edge_version\n WHERE edge_id = $1 AND version_num = $2\n ",
"query": "\n SELECT \n version_num,\n operation,\n snapshot,\n changed_at,\n changed_by,\n COALESCE(fields_changed, '{}') as \"fields_changed!\"\n FROM knowledge_edge_version\n WHERE edge_id = $1 AND version_num = $2\n ",
"describe": {
"columns": [
{
@@ -53,7 +53,7 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_community_fts
-- ============================================
-- STEP 3: Create entity table
-- ============================================
CREATE TABLE IF NOT EXISTS memory_entity (
CREATE TABLE IF NOT EXISTS knowledge_node (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id VARCHAR(255) NOT NULL,
name VARCHAR(500) NOT NULL,
@@ -75,40 +75,40 @@ CREATE TABLE IF NOT EXISTS memory_entity (
);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_project
ON memory_entity(project_id);
ON knowledge_node(project_id);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_name_embedding
ON memory_entity USING hnsw (name_embedding vector_cosine_ops)
ON knowledge_node USING hnsw (name_embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_summary_embedding
ON memory_entity USING hnsw (summary_embedding vector_cosine_ops)
ON knowledge_node USING hnsw (summary_embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_type
ON memory_entity(project_id, entity_type);
ON knowledge_node(project_id, entity_type);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_community
ON memory_entity(community_id)
ON knowledge_node(community_id)
WHERE community_id IS NOT NULL;
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_active
ON memory_entity(project_id)
ON knowledge_node(project_id)
WHERE t_expired IS NULL;
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_fts
ON memory_entity USING gin(
ON knowledge_node USING gin(
to_tsvector('english', COALESCE(name, '') || ' ' || COALESCE(summary, ''))
);
-- ============================================
-- STEP 4: Create edge (relationship) table
-- ============================================
CREATE TABLE IF NOT EXISTS memory_edge (
CREATE TABLE IF NOT EXISTS knowledge_edge (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id VARCHAR(255) NOT NULL,
source_entity_id UUID NOT NULL REFERENCES memory_entity(id) ON DELETE CASCADE,
target_entity_id UUID NOT NULL REFERENCES memory_entity(id) ON DELETE CASCADE,
source_entity_id UUID NOT NULL REFERENCES knowledge_node(id) ON DELETE CASCADE,
target_entity_id UUID NOT NULL REFERENCES knowledge_node(id) ON DELETE CASCADE,
relation_type VARCHAR(100) NOT NULL,
fact TEXT NOT NULL,
fact_embedding VECTOR(768),
@@ -123,7 +123,7 @@ CREATE TABLE IF NOT EXISTS memory_edge (
-- Provenance
source_episode_id BIGINT REFERENCES memory_node(id) ON DELETE SET NULL,
invalidated_by UUID REFERENCES memory_edge(id) ON DELETE SET NULL,
invalidated_by UUID REFERENCES knowledge_edge(id) ON DELETE SET NULL,
-- Contradiction handling
contradiction_status VARCHAR(20) DEFAULT 'active'
@@ -141,34 +141,34 @@ CREATE TABLE IF NOT EXISTS memory_edge (
);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_project
ON memory_edge(project_id);
ON knowledge_edge(project_id);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_source
ON memory_edge(source_entity_id);
ON knowledge_edge(source_entity_id);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_target
ON memory_edge(target_entity_id);
ON knowledge_edge(target_entity_id);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_entity_pair
ON memory_edge(source_entity_id, target_entity_id);
ON knowledge_edge(source_entity_id, target_entity_id);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_embedding
ON memory_edge USING hnsw (fact_embedding vector_cosine_ops)
ON knowledge_edge USING hnsw (fact_embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_validity
ON memory_edge(t_valid, t_invalid)
ON knowledge_edge(t_valid, t_invalid)
WHERE t_expired IS NULL;
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_active
ON memory_edge(project_id, contradiction_status)
ON knowledge_edge(project_id, contradiction_status)
WHERE t_expired IS NULL AND contradiction_status = 'active';
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_fts
ON memory_edge USING gin(to_tsvector('english', fact));
ON knowledge_edge USING gin(to_tsvector('english', fact));
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_relation
ON memory_edge(project_id, relation_type);
ON knowledge_edge(project_id, relation_type);
-- ============================================
-- STEP 5: Compaction audit log
@@ -202,8 +202,8 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_compaction_project
CREATE TABLE IF NOT EXISTS contradiction_review_queue (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id VARCHAR(255) NOT NULL,
new_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE,
existing_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE,
new_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE,
existing_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE,
confidence FLOAT NOT NULL,
explanation TEXT,
queued_at TIMESTAMPTZ DEFAULT NOW(),
+24 -24
View File
@@ -41,53 +41,53 @@ CREATE INDEX IF NOT EXISTS idx_memory_projects_visibility
-- 2. Add attribution columns to entities
-- ─────────────────────────────────────────────────────────────────────────────
ALTER TABLE memory_entity ADD COLUMN IF NOT EXISTS (
ALTER TABLE knowledge_node ADD COLUMN IF NOT EXISTS (
contributed_by VARCHAR(255), -- JWT "sub" who added this
contribution_type VARCHAR(50), -- "extracted" | "manual" | "inferred"
contribution_date TIMESTAMPTZ
);
-- Populate contributed_by with defaults (assume "system" if not present)
UPDATE memory_entity
UPDATE knowledge_node
SET contributed_by = 'system'
WHERE contributed_by IS NULL;
ALTER TABLE memory_entity
ALTER TABLE knowledge_node
ALTER COLUMN contributed_by SET NOT NULL;
CREATE INDEX IF NOT EXISTS idx_memory_entity_contributed_by
ON memory_entity(contributed_by);
CREATE INDEX IF NOT EXISTS idx_knowledge_node_contributed_by
ON knowledge_node(contributed_by);
CREATE INDEX IF NOT EXISTS idx_memory_entity_contribution_date
ON memory_entity(contribution_date)
CREATE INDEX IF NOT EXISTS idx_knowledge_node_contribution_date
ON knowledge_node(contribution_date)
WHERE contribution_date IS NOT NULL;
-- ─────────────────────────────────────────────────────────────────────────────
-- 3. Add attribution columns to edges
-- ─────────────────────────────────────────────────────────────────────────────
ALTER TABLE memory_edge ADD COLUMN IF NOT EXISTS (
ALTER TABLE knowledge_edge ADD COLUMN IF NOT EXISTS (
contributed_by VARCHAR(255), -- JWT "sub" who added this
contribution_type VARCHAR(50) -- "extracted" | "inferred"
);
-- Populate contributed_by with defaults
UPDATE memory_edge
UPDATE knowledge_edge
SET contributed_by = 'system'
WHERE contributed_by IS NULL;
ALTER TABLE memory_edge
ALTER TABLE knowledge_edge
ALTER COLUMN contributed_by SET NOT NULL;
CREATE INDEX IF NOT EXISTS idx_memory_edge_contributed_by
ON memory_edge(contributed_by);
CREATE INDEX IF NOT EXISTS idx_knowledge_edge_contributed_by
ON knowledge_edge(contributed_by);
-- ─────────────────────────────────────────────────────────────────────────────
-- 4. Add project_id to entities and edges (optional, for faster filtering)
-- ─────────────────────────────────────────────────────────────────────────────
-- ALTER TABLE memory_entity ADD COLUMN IF NOT EXISTS project_id VARCHAR(255);
-- ALTER TABLE memory_edge ADD COLUMN IF NOT EXISTS project_id VARCHAR(255);
-- ALTER TABLE knowledge_node ADD COLUMN IF NOT EXISTS project_id VARCHAR(255);
-- ALTER TABLE knowledge_edge ADD COLUMN IF NOT EXISTS project_id VARCHAR(255);
--
-- Note: Deferred. Can use project info from path/source instead.
@@ -102,7 +102,7 @@ SELECT
COUNT(*) as entity_count,
MAX(contribution_date) as last_contribution,
ARRAY_AGG(DISTINCT contribution_type) as contribution_types
FROM memory_entity
FROM knowledge_node
WHERE contribution_date IS NOT NULL
GROUP BY contributed_by;
@@ -112,7 +112,7 @@ SELECT
contributed_by,
COUNT(*) as count,
MAX(contribution_date) as latest
FROM memory_entity
FROM knowledge_node
WHERE contribution_date > CURRENT_TIMESTAMP - INTERVAL '7 days'
GROUP BY contributed_by;
@@ -129,12 +129,12 @@ GROUP BY contributed_by;
-- DROP INDEX IF EXISTS idx_memory_projects_visibility;
-- DROP TABLE IF EXISTS memory_projects;
--
-- DROP INDEX IF EXISTS idx_memory_entity_contributed_by;
-- DROP INDEX IF EXISTS idx_memory_entity_contribution_date;
-- ALTER TABLE memory_entity DROP COLUMN IF EXISTS contributed_by;
-- ALTER TABLE memory_entity DROP COLUMN IF EXISTS contribution_type;
-- ALTER TABLE memory_entity DROP COLUMN IF EXISTS contribution_date;
-- DROP INDEX IF EXISTS idx_knowledge_node_contributed_by;
-- DROP INDEX IF EXISTS idx_knowledge_node_contribution_date;
-- ALTER TABLE knowledge_node DROP COLUMN IF EXISTS contributed_by;
-- ALTER TABLE knowledge_node DROP COLUMN IF EXISTS contribution_type;
-- ALTER TABLE knowledge_node DROP COLUMN IF EXISTS contribution_date;
--
-- DROP INDEX IF EXISTS idx_memory_edge_contributed_by;
-- ALTER TABLE memory_edge DROP COLUMN IF EXISTS contributed_by;
-- ALTER TABLE memory_edge DROP COLUMN IF EXISTS contribution_type;
-- DROP INDEX IF EXISTS idx_knowledge_edge_contributed_by;
-- ALTER TABLE knowledge_edge DROP COLUMN IF EXISTS contributed_by;
-- ALTER TABLE knowledge_edge DROP COLUMN IF EXISTS contribution_type;
@@ -18,8 +18,8 @@ CREATE TABLE IF NOT EXISTS temporal_workflow_links (
run_id VARCHAR(255), -- Temporal run ID
-- Memory Entity References (Optional - NULL until linked)
entity_id UUID REFERENCES memory_entity(id) ON DELETE SET NULL,
edge_id UUID REFERENCES memory_edge(id) ON DELETE SET NULL,
entity_id UUID REFERENCES knowledge_node(id) ON DELETE SET NULL,
edge_id UUID REFERENCES knowledge_edge(id) ON DELETE SET NULL,
node_id BIGINT REFERENCES memory_node(id) ON DELETE SET NULL,
-- Sync Status
@@ -47,7 +47,7 @@ CREATE TABLE IF NOT EXISTS community_member_map (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id VARCHAR(255) NOT NULL,
community_id UUID NOT NULL REFERENCES memory_community(id) ON DELETE CASCADE,
entity_id UUID NOT NULL REFERENCES memory_entity(id) ON DELETE CASCADE,
entity_id UUID NOT NULL REFERENCES knowledge_node(id) ON DELETE CASCADE,
label_propagation_run_id UUID REFERENCES label_propagation_run(id) ON DELETE SET NULL,
-- Label strength (0-1, higher = stronger membership)
@@ -5,7 +5,7 @@
BEGIN;
-- Entity version snapshots (immutable)
CREATE TABLE IF NOT EXISTS memory_entity_version (
CREATE TABLE IF NOT EXISTS knowledge_node_version (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
entity_id VARCHAR(255) NOT NULL,
version_num INTEGER NOT NULL,
@@ -28,7 +28,7 @@ CREATE TABLE IF NOT EXISTS memory_entity_version (
);
-- Edge version snapshots
CREATE TABLE IF NOT EXISTS memory_edge_version (
CREATE TABLE IF NOT EXISTS knowledge_edge_version (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
edge_id UUID NOT NULL,
version_num INTEGER NOT NULL,
@@ -52,22 +52,22 @@ CREATE TABLE IF NOT EXISTS memory_edge_version (
-- Indexes for efficient lookups
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_version_entity_id
ON memory_entity_version(entity_id, version_num DESC);
ON knowledge_node_version(entity_id, version_num DESC);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_version_changed_at
ON memory_entity_version(changed_at DESC);
ON knowledge_node_version(changed_at DESC);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_entity_version_changed_by
ON memory_entity_version(changed_by);
ON knowledge_node_version(changed_by);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_version_edge_id
ON memory_edge_version(edge_id, version_num DESC);
ON knowledge_edge_version(edge_id, version_num DESC);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_version_changed_at
ON memory_edge_version(changed_at DESC);
ON knowledge_edge_version(changed_at DESC);
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_edge_version_changed_by
ON memory_edge_version(changed_by);
ON knowledge_edge_version(changed_by);
-- Immutability enforcement: version tables are append-only
CREATE OR REPLACE FUNCTION prevent_version_table_modification()
@@ -77,19 +77,19 @@ BEGIN
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER memory_entity_version_immutable
BEFORE UPDATE OR DELETE ON memory_entity_version
CREATE TRIGGER knowledge_node_version_immutable
BEFORE UPDATE OR DELETE ON knowledge_node_version
FOR EACH ROW EXECUTE FUNCTION prevent_version_table_modification();
CREATE TRIGGER memory_edge_version_immutable
BEFORE UPDATE OR DELETE ON memory_edge_version
CREATE TRIGGER knowledge_edge_version_immutable
BEFORE UPDATE OR DELETE ON knowledge_edge_version
FOR EACH ROW EXECUTE FUNCTION prevent_version_table_modification();
COMMIT;
-- Rollback (for reference):
-- DROP TRIGGER memory_entity_version_immutable ON memory_entity_version;
-- DROP TRIGGER memory_edge_version_immutable ON memory_edge_version;
-- DROP TRIGGER knowledge_node_version_immutable ON knowledge_node_version;
-- DROP TRIGGER knowledge_edge_version_immutable ON knowledge_edge_version;
-- DROP FUNCTION prevent_version_table_modification();
-- DROP TABLE memory_entity_version;
-- DROP TABLE memory_edge_version;
-- DROP TABLE knowledge_node_version;
-- DROP TABLE knowledge_edge_version;
@@ -9,8 +9,8 @@ CREATE TABLE IF NOT EXISTS exact_dedup_record (
project_id VARCHAR(255) NOT NULL,
-- Source and target edges
source_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE,
target_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE,
source_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE,
target_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE,
-- Match criteria (all must match for exact dedup)
source_match BOOLEAN NOT NULL,
@@ -51,8 +51,8 @@ CREATE TABLE IF NOT EXISTS stale_gc_record (
project_id VARCHAR(255) NOT NULL,
-- Entity or edge marked for GC
entity_id UUID REFERENCES memory_entity(id) ON DELETE CASCADE,
edge_id UUID REFERENCES memory_edge(id) ON DELETE CASCADE,
entity_id UUID REFERENCES knowledge_node(id) ON DELETE CASCADE,
edge_id UUID REFERENCES knowledge_edge(id) ON DELETE CASCADE,
-- Staleness criteria
age_days INT NOT NULL,
@@ -96,8 +96,8 @@ CREATE TABLE IF NOT EXISTS semantic_dedup_record (
project_id VARCHAR(255) NOT NULL,
-- Source and target edges
source_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE,
target_edge_id UUID NOT NULL REFERENCES memory_edge(id) ON DELETE CASCADE,
source_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE,
target_edge_id UUID NOT NULL REFERENCES knowledge_edge(id) ON DELETE CASCADE,
-- Pre-filter score (0-1, eliminates 60-70% of candidates)
prefilter_score FLOAT NOT NULL,
@@ -119,7 +119,7 @@ CREATE TABLE IF NOT EXISTS semantic_dedup_record (
-- Merge strategy (if auto-merged)
merge_strategy VARCHAR(50), -- 'keep_superset', 'keep_newer', 'keep_higher_confidence'
merged_edge_id UUID REFERENCES memory_edge(id) ON DELETE SET NULL,
merged_edge_id UUID REFERENCES knowledge_edge(id) ON DELETE SET NULL,
-- Metadata
detected_at TIMESTAMPTZ DEFAULT NOW(),
@@ -1,21 +1,10 @@
-- Migration 009: Temporal edge schema (Zep paper §2.2.2)
-- Replaces old memory_edge (child_sha/parent_sha node graph)
-- with temporal edge schema supporting relation types, facts, and validity periods.
-- Migration 009: Temporal knowledge graph edge schema (Zep paper §2.2.2)
-- Creates knowledge_edge table for entity relationships.
-- The old memory_edge (child_sha/parent_sha provenance DAG) is kept as-is.
-- Idempotent: safe to run multiple times.
-- Rename old table if it still exists (skip if already migrated)
DO $$
BEGIN
IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'memory_edge'
AND EXISTS (SELECT 1 FROM information_schema.columns
WHERE table_name = 'memory_edge' AND column_name = 'child_sha'))
THEN
ALTER TABLE memory_edge RENAME TO memory_edge_legacy;
END IF;
END $$;
-- Create temporal edge table
CREATE TABLE IF NOT EXISTS memory_edge (
-- Create knowledge graph edge table (separate from provenance memory_edge)
CREATE TABLE IF NOT EXISTS knowledge_edge (
id TEXT PRIMARY KEY,
project_id TEXT NOT NULL DEFAULT 'default',
source_id TEXT NOT NULL,
@@ -38,30 +27,30 @@ CREATE TABLE IF NOT EXISTS memory_edge (
-- Ensure app user owns the table
DO $$ BEGIN
IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'app') THEN
ALTER TABLE memory_edge OWNER TO app;
ALTER TABLE knowledge_edge OWNER TO app;
END IF;
END $$;
CREATE INDEX IF NOT EXISTS idx_memory_edge_source ON memory_edge(source_id);
CREATE INDEX IF NOT EXISTS idx_memory_edge_target ON memory_edge(target_id);
CREATE INDEX IF NOT EXISTS idx_memory_edge_project ON memory_edge(project_id);
CREATE INDEX IF NOT EXISTS idx_memory_edge_relation ON memory_edge(relation_type);
CREATE INDEX IF NOT EXISTS idx_knowledge_edge_source ON knowledge_edge(source_id);
CREATE INDEX IF NOT EXISTS idx_knowledge_edge_target ON knowledge_edge(target_id);
CREATE INDEX IF NOT EXISTS idx_knowledge_edge_project ON knowledge_edge(project_id);
CREATE INDEX IF NOT EXISTS idx_knowledge_edge_relation ON knowledge_edge(relation_type);
-- Ensure memory_entity has all columns code expects
ALTER TABLE memory_entity ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ;
ALTER TABLE memory_entity ADD COLUMN IF NOT EXISTS source_count INTEGER DEFAULT 1;
-- Ensure knowledge_node has all columns code expects
ALTER TABLE knowledge_node ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ;
ALTER TABLE knowledge_node ADD COLUMN IF NOT EXISTS source_count INTEGER DEFAULT 1;
-- Unique constraint for entity upsert dedup
DO $$
BEGIN
-- Dedup existing rows before creating unique index
DELETE FROM memory_entity a USING memory_entity b
DELETE FROM knowledge_node a USING knowledge_node b
WHERE a.project_id = b.project_id AND a.name = b.name
AND a.t_created < b.t_created;
EXCEPTION WHEN OTHERS THEN NULL;
END $$;
CREATE UNIQUE INDEX IF NOT EXISTS idx_memory_entity_project_name ON memory_entity(project_id, name);
CREATE UNIQUE INDEX IF NOT EXISTS idx_knowledge_node_project_name ON knowledge_node(project_id, name);
-- ROLLBACK instructions:
-- DROP TABLE IF EXISTS memory_edge;
-- ALTER TABLE IF EXISTS memory_edge_legacy RENAME TO memory_edge;
-- DROP TABLE IF EXISTS knowledge_edge;
-- DROP INDEX IF EXISTS idx_knowledge_node_project_name;
+4 -4
View File
@@ -25,7 +25,7 @@ impl AuditLogger {
) -> Result<(), sqlx::Error> {
sqlx::query(
r#"
INSERT INTO memory_entity_version
INSERT INTO knowledge_node_version
(entity_id, version_num, operation, snapshot, changed_by, fields_changed)
VALUES ($1, $2, $3, $4, $5, $6)
"#,
@@ -54,7 +54,7 @@ impl AuditLogger {
) -> Result<(), sqlx::Error> {
sqlx::query(
r#"
INSERT INTO memory_edge_version
INSERT INTO knowledge_edge_version
(edge_id, version_num, operation, snapshot, changed_by, fields_changed)
VALUES ($1, $2, $3, $4, $5, $6)
"#,
@@ -87,7 +87,7 @@ impl AuditLogger {
changed_at,
changed_by,
COALESCE(fields_changed, '{}') as "fields_changed"
FROM memory_entity_version
FROM knowledge_node_version
WHERE entity_id = $1
ORDER BY version_num DESC
"#,
@@ -113,7 +113,7 @@ impl AuditLogger {
changed_at,
changed_by,
COALESCE(fields_changed, '{}') as "fields_changed"
FROM memory_edge_version
FROM knowledge_edge_version
WHERE edge_id = $1
ORDER BY version_num DESC
"#,
+8 -8
View File
@@ -80,7 +80,7 @@ impl PersistentEntityRepo {
/// Save entity to database (idempotent)
pub async fn save(&self, entity: &Entity) -> Result<String, DbError> {
let query = r#"
INSERT INTO memory_entity (id, entity_type, name, description, embedding, created_at, updated_at)
INSERT INTO knowledge_node (id, entity_type, name, description, embedding, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT(id) DO UPDATE SET
name = EXCLUDED.name,
@@ -114,7 +114,7 @@ impl PersistentEntityRepo {
pub async fn get(&self, id: &str) -> Result<Option<Entity>, DbError> {
let query = r#"
SELECT id, entity_type, name, description, embedding, created_at, updated_at
FROM memory_entity
FROM knowledge_node
WHERE id = $1 AND deleted_at IS NULL;
"#;
@@ -139,7 +139,7 @@ impl PersistentEntityRepo {
pub async fn list(&self, limit: i64, offset: i64) -> Result<Vec<Entity>, DbError> {
let query = r#"
SELECT id, entity_type, name, description, embedding, created_at, updated_at
FROM memory_entity
FROM knowledge_node
WHERE deleted_at IS NULL
ORDER BY created_at DESC
LIMIT $1 OFFSET $2;
@@ -166,7 +166,7 @@ impl PersistentEntityRepo {
/// Soft delete entity
pub async fn delete(&self, id: &str) -> Result<(), DbError> {
let query = r#"
UPDATE memory_entity
UPDATE knowledge_node
SET deleted_at = $1
WHERE id = $2;
"#;
@@ -195,7 +195,7 @@ impl PersistentEdgeRepo {
/// Save edge to database (idempotent)
pub async fn save(&self, edge: &Edge) -> Result<String, DbError> {
let query = r#"
INSERT INTO memory_edge (id, source_id, target_id, relation_type, fact, strength, t_valid, t_invalid, t_created, t_expired)
INSERT INTO knowledge_edge (id, source_id, target_id, relation_type, fact, strength, t_valid, t_invalid, t_created, t_expired)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
ON CONFLICT(id) DO UPDATE SET
strength = EXCLUDED.strength,
@@ -232,7 +232,7 @@ impl PersistentEdgeRepo {
pub async fn get(&self, id: &str) -> Result<Option<Edge>, DbError> {
let query = r#"
SELECT id, source_id, target_id, relation_type, fact, strength, t_valid, t_invalid, t_created, t_expired
FROM memory_edge
FROM knowledge_edge
WHERE id = $1 AND t_expired IS NULL;
"#;
@@ -260,7 +260,7 @@ impl PersistentEdgeRepo {
pub async fn list_from(&self, source_id: &str, limit: i64) -> Result<Vec<Edge>, DbError> {
let query = r#"
SELECT id, source_id, target_id, relation_type, fact, strength, t_valid, t_invalid, t_created, t_expired
FROM memory_edge
FROM knowledge_edge
WHERE source_id = $1 AND t_expired IS NULL AND t_invalid IS NULL
ORDER BY t_created DESC
LIMIT $2;
@@ -290,7 +290,7 @@ impl PersistentEdgeRepo {
/// Mark edge as contradicted (soft delete)
pub async fn invalidate(&self, id: &str) -> Result<(), DbError> {
let query = r#"
UPDATE memory_edge
UPDATE knowledge_edge
SET t_invalid = $1
WHERE id = $2;
"#;
+69 -1
View File
@@ -218,6 +218,74 @@ pub async fn init_schema(pool: &PgPool) -> Result<()> {
.execute(pool)
.await?;
tracing::info!("Database schema initialized");
// Memory entity table (temporal knowledge graph)
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS knowledge_node (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id VARCHAR(255) NOT NULL,
name VARCHAR(500) NOT NULL,
name_embedding VECTOR(768),
summary TEXT,
description TEXT,
summary_embedding VECTOR(768),
entity_type VARCHAR(50),
t_created TIMESTAMPTZ DEFAULT NOW(),
t_updated TIMESTAMPTZ DEFAULT NOW(),
t_expired TIMESTAMPTZ,
confidence FLOAT DEFAULT 1.0,
source_count INT DEFAULT 1,
source_episodes UUID[] DEFAULT '{}',
access_count BIGINT DEFAULT 0,
last_accessed TIMESTAMPTZ,
UNIQUE(project_id, name)
)
"#,
)
.execute(pool)
.await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_entity_project ON knowledge_node(project_id)")
.execute(pool)
.await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_entity_type ON knowledge_node(project_id, entity_type)")
.execute(pool)
.await?;
// Memory edge table (temporal knowledge graph)
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS knowledge_edge (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id VARCHAR(255) NOT NULL,
source_id UUID NOT NULL,
target_id UUID NOT NULL,
relation_type VARCHAR(100) NOT NULL,
fact TEXT NOT NULL,
fact_embedding VECTOR(768),
t_valid TIMESTAMPTZ,
t_invalid TIMESTAMPTZ,
t_created TIMESTAMPTZ DEFAULT NOW(),
t_expired TIMESTAMPTZ,
confidence FLOAT DEFAULT 1.0,
contradiction_status VARCHAR(20) DEFAULT 'active',
contradiction_confidence FLOAT
)
"#,
)
.execute(pool)
.await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_edge_project ON knowledge_edge(project_id)")
.execute(pool)
.await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_edge_source ON knowledge_edge(source_id)")
.execute(pool)
.await?;
sqlx::query("CREATE INDEX IF NOT EXISTS idx_edge_target ON knowledge_edge(target_id)")
.execute(pool)
.await?;
tracing::info!("Database schema initialized (including knowledge_node + knowledge_edge)");
Ok(())
}
+32 -109
View File
@@ -1,15 +1,16 @@
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use sqlx::{PgPool, FromRow};
use uuid::Uuid;
use chrono::{DateTime, Utc};
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
pub struct VersionSnapshot {
pub version_num: i32,
pub operation: String, // 'create' | 'update' | 'delete'
pub snapshot: serde_json::Value,
pub changed_at: DateTime<Utc>,
pub changed_by: String,
#[sqlx(default)]
pub fields_changed: Vec<String>,
}
@@ -40,8 +41,7 @@ impl EntityVersioningService {
/// Get all versions of an entity in descending order
pub async fn get_versions(&self, entity_id: &str) -> Result<Vec<VersionSnapshot>, sqlx::Error> {
sqlx::query_as!(
VersionSnapshot,
sqlx::query_as::<_, VersionSnapshot>(
r#"
SELECT
version_num,
@@ -49,13 +49,13 @@ impl EntityVersioningService {
snapshot,
changed_at,
changed_by,
COALESCE(fields_changed, '{}') as "fields_changed!"
FROM memory_entity_version
COALESCE(fields_changed, '{}') as fields_changed
FROM knowledge_node_version
WHERE entity_id = $1
ORDER BY version_num DESC
"#,
entity_id
)
.bind(entity_id)
.fetch_all(&self.pool)
.await
}
@@ -66,8 +66,7 @@ impl EntityVersioningService {
entity_id: &str,
version_num: i32,
) -> Result<Option<VersionSnapshot>, sqlx::Error> {
sqlx::query_as!(
VersionSnapshot,
sqlx::query_as::<_, VersionSnapshot>(
r#"
SELECT
version_num,
@@ -75,13 +74,13 @@ impl EntityVersioningService {
snapshot,
changed_at,
changed_by,
COALESCE(fields_changed, '{}') as "fields_changed!"
FROM memory_entity_version
COALESCE(fields_changed, '{}') as fields_changed
FROM knowledge_node_version
WHERE entity_id = $1 AND version_num = $2
"#,
entity_id,
version_num
)
.bind(entity_id)
.bind(version_num)
.fetch_optional(&self.pool)
.await
}
@@ -95,78 +94,7 @@ impl EntityVersioningService {
) -> Result<DiffResult, sqlx::Error> {
let from_snap = self.get_version(entity_id, from_v).await?;
let to_snap = self.get_version(entity_id, to_v).await?;
let from_obj = from_snap
.as_ref()
.and_then(|s| s.snapshot.as_object())
.map(|o| o.clone());
let to_obj = to_snap
.as_ref()
.and_then(|s| s.snapshot.as_object())
.map(|o| o.clone());
let mut added = Vec::new();
let mut removed = Vec::new();
let mut modified = Vec::new();
// Check removed and modified
if let Some(ref from) = from_obj {
for (key, from_val) in from {
if let Some(to) = &to_obj {
if let Some(to_val) = to.get(key) {
if from_val != to_val {
modified.push(DiffField {
name: key.clone(),
from_value: Some(from_val.clone()),
to_value: Some(to_val.clone()),
});
}
} else {
removed.push(DiffField {
name: key.clone(),
from_value: Some(from_val.clone()),
to_value: None,
});
}
} else {
removed.push(DiffField {
name: key.clone(),
from_value: Some(from_val.clone()),
to_value: None,
});
}
}
}
// Check added
if let Some(to) = to_obj {
for (key, to_val) in to {
if let Some(from) = &from_obj {
if !from.contains_key(&key) {
added.push(DiffField {
name: key,
from_value: None,
to_value: Some(to_val),
});
}
} else {
added.push(DiffField {
name: key,
from_value: None,
to_value: Some(to_val),
});
}
}
}
Ok(DiffResult {
from_version: from_v,
to_version: to_v,
added_fields: added,
removed_fields: removed,
modified_fields: modified,
})
compute_diff(from_snap, to_snap, from_v, to_v)
}
/// Get entity state at a point in time
@@ -175,8 +103,7 @@ impl EntityVersioningService {
entity_id: &str,
as_of: DateTime<Utc>,
) -> Result<Option<VersionSnapshot>, sqlx::Error> {
sqlx::query_as!(
VersionSnapshot,
sqlx::query_as::<_, VersionSnapshot>(
r#"
SELECT
version_num,
@@ -184,15 +111,15 @@ impl EntityVersioningService {
snapshot,
changed_at,
changed_by,
COALESCE(fields_changed, '{}') as "fields_changed!"
FROM memory_entity_version
COALESCE(fields_changed, '{}') as fields_changed
FROM knowledge_node_version
WHERE entity_id = $1 AND changed_at <= $2
ORDER BY version_num DESC
LIMIT 1
"#,
entity_id,
as_of
)
.bind(entity_id)
.bind(as_of)
.fetch_optional(&self.pool)
.await
}
@@ -210,8 +137,7 @@ impl EdgeVersioningService {
/// Get all versions of an edge
pub async fn get_versions(&self, edge_id: Uuid) -> Result<Vec<VersionSnapshot>, sqlx::Error> {
sqlx::query_as!(
VersionSnapshot,
sqlx::query_as::<_, VersionSnapshot>(
r#"
SELECT
version_num,
@@ -219,13 +145,13 @@ impl EdgeVersioningService {
snapshot,
changed_at,
changed_by,
COALESCE(fields_changed, '{}') as "fields_changed!"
FROM memory_edge_version
COALESCE(fields_changed, '{}') as fields_changed
FROM knowledge_edge_version
WHERE edge_id = $1
ORDER BY version_num DESC
"#,
edge_id
)
.bind(edge_id)
.fetch_all(&self.pool)
.await
}
@@ -237,8 +163,7 @@ impl EdgeVersioningService {
from_v: i32,
to_v: i32,
) -> Result<DiffResult, sqlx::Error> {
let from_snap = sqlx::query_as!(
VersionSnapshot,
let from_snap = sqlx::query_as::<_, VersionSnapshot>(
r#"
SELECT
version_num,
@@ -246,18 +171,17 @@ impl EdgeVersioningService {
snapshot,
changed_at,
changed_by,
COALESCE(fields_changed, '{}') as "fields_changed!"
FROM memory_edge_version
COALESCE(fields_changed, '{}') as fields_changed
FROM knowledge_edge_version
WHERE edge_id = $1 AND version_num = $2
"#,
edge_id,
from_v
)
.bind(edge_id)
.bind(from_v)
.fetch_optional(&self.pool)
.await?;
let to_snap = sqlx::query_as!(
VersionSnapshot,
let to_snap = sqlx::query_as::<_, VersionSnapshot>(
r#"
SELECT
version_num,
@@ -265,17 +189,16 @@ impl EdgeVersioningService {
snapshot,
changed_at,
changed_by,
COALESCE(fields_changed, '{}') as "fields_changed!"
FROM memory_edge_version
COALESCE(fields_changed, '{}') as fields_changed
FROM knowledge_edge_version
WHERE edge_id = $1 AND version_num = $2
"#,
edge_id,
to_v
)
.bind(edge_id)
.bind(to_v)
.fetch_optional(&self.pool)
.await?;
// Same diff logic as entities
compute_diff(from_snap, to_snap, from_v, to_v)
}
}