diff --git a/diode-server/dbstore/postgres/migrations/20260121000001_graph_tables.sql b/diode-server/dbstore/postgres/migrations/20260121000001_graph_tables.sql index 75ffe755..8f7e6209 100644 --- a/diode-server/dbstore/postgres/migrations/20260121000001_graph_tables.sql +++ b/diode-server/dbstore/postgres/migrations/20260121000001_graph_tables.sql @@ -8,8 +8,8 @@ CREATE TABLE IF NOT EXISTS graph_nodes ( data JSONB NOT NULL, duplicate_count INTEGER DEFAULT 1 NOT NULL, matching_schema_version INTEGER DEFAULT 1 NOT NULL, - created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), - updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL, + updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL, CONSTRAINT unique_entity UNIQUE (node_type, external_id) ); @@ -19,8 +19,8 @@ CREATE TABLE IF NOT EXISTS graph_node_snapshots ( id BIGSERIAL PRIMARY KEY, node_id BIGINT NOT NULL REFERENCES graph_nodes(id) ON DELETE CASCADE, snapshot_data JSONB NOT NULL, - sequence_number INTEGER NOT NULL, - created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + sequence_number INTEGER NOT NULL CHECK (sequence_number > 0), + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL, CONSTRAINT unique_node_sequence UNIQUE (node_id, sequence_number) ); @@ -33,10 +33,10 @@ CREATE TABLE IF NOT EXISTS graph_edges ( edge_type TEXT NOT NULL, edge_subtype TEXT, properties JSONB DEFAULT '{}', - confidence_score REAL DEFAULT 1.0 NOT NULL, + confidence_score REAL DEFAULT 1.0 NOT NULL CHECK (confidence_score >= 0.0 AND confidence_score <= 1.0), match_reason TEXT, matching_fields JSONB DEFAULT '[]', - created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL, CONSTRAINT unique_edge UNIQUE (source_node_id, target_node_id, edge_type) ); diff --git a/diode-server/dbstore/postgres/queries/graph.sql b/diode-server/dbstore/postgres/queries/graph.sql new file mode 100644 index 00000000..177051b4 --- /dev/null +++ b/diode-server/dbstore/postgres/queries/graph.sql @@ -0,0 +1,228 @@ +-- name: UpsertGraphNode :one +INSERT INTO graph_nodes (external_id, node_type, data, duplicate_count, matching_schema_version) +VALUES ($1, $2, $3, 1, $4) +ON CONFLICT (node_type, external_id) +DO UPDATE SET + data = EXCLUDED.data, + duplicate_count = graph_nodes.duplicate_count + 1, + matching_schema_version = EXCLUDED.matching_schema_version, + updated_at = NOW() +RETURNING id, external_id, node_type, data, duplicate_count, matching_schema_version; + +-- name: UpdateGraphNodeData :one +UPDATE graph_nodes +SET data = $3, matching_schema_version = $4, updated_at = NOW() +WHERE node_type = $1 AND external_id = $2 +RETURNING id, external_id, node_type, data, duplicate_count, matching_schema_version; + +-- name: FindGraphNode :one +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version +FROM graph_nodes +WHERE node_type = $1 AND external_id = $2; + +-- name: UpsertGraphEdge :exec +INSERT INTO graph_edges (source_node_id, target_node_id, edge_type, properties) +VALUES ($1, $2, $3, $4) +ON CONFLICT (source_node_id, target_node_id, edge_type) +DO UPDATE SET properties = EXCLUDED.properties; + +-- name: UpsertGraphEdgeWithConfidence :exec +INSERT INTO graph_edges (source_node_id, target_node_id, edge_type, properties, confidence_score, match_reason, matching_fields, edge_subtype) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8) +ON CONFLICT (source_node_id, target_node_id, edge_type) +DO UPDATE SET + properties = EXCLUDED.properties, + confidence_score = EXCLUDED.confidence_score, + match_reason = EXCLUDED.match_reason, + matching_fields = EXCLUDED.matching_fields, + edge_subtype = EXCLUDED.edge_subtype; + +-- name: GetGraphNodesByType :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE node_type = $1 +ORDER BY updated_at DESC +LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset'); + +-- name: GetConnectedNodes :many +SELECT + target.id, + target.external_id, + target.node_type, + target.data, + target.duplicate_count, + target.matching_schema_version, + edges.edge_type, + edges.properties as edge_properties +FROM graph_edges edges +JOIN graph_nodes target ON edges.target_node_id = target.id +WHERE edges.source_node_id = $1; + +-- name: GetNodesByFrequency :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE duplicate_count >= $1 +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset'); + +-- name: SearchGraphNodes :many +-- search_term is REQUIRED (non-null, non-empty string) to prevent full table scans. +-- Empty string returns no results. Use GetGraphNodesByType for listing without filter. +-- node_type is optional - pass NULL to search across all types. +-- Note: Very long search_term values may cause slow scans on large datasets. +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE (node_type = sqlc.narg('node_type')::text OR sqlc.narg('node_type')::text IS NULL) + AND LENGTH(sqlc.arg('search_term')::text) > 0 + AND ( + external_id ILIKE '%' || sqlc.arg('search_term')::text || '%' + OR data::text ILIKE '%' || sqlc.arg('search_term')::text || '%' + ) +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset'); + +-- name: GetGraphStats :one +SELECT + COUNT(*) as total_nodes, + COUNT(DISTINCT node_type) as unique_node_types, + SUM(duplicate_count) as total_observations, + AVG(duplicate_count) as avg_observations_per_node +FROM graph_nodes; + +-- name: GetGraphStatsByType :many +SELECT + node_type, + COUNT(*) as node_count, + SUM(duplicate_count) as total_observations, + AVG(duplicate_count) as avg_observations_per_node, + MAX(duplicate_count) as max_observations, + MIN(created_at) as first_seen, + MAX(updated_at) as last_seen +FROM graph_nodes +GROUP BY node_type +ORDER BY node_count DESC; + +-- Entity matching queries for confidence-based matching + +-- name: FindNodesByFieldMatch :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE node_type = $1 + AND ( + -- Support matching by single JSON field (only if json_field is provided) + (sqlc.narg('json_field')::text IS NOT NULL AND data->>sqlc.narg('json_field')::text = sqlc.narg('field_value')::text) + -- Support matching by nested JSON path (only if nested_path is provided) + OR (sqlc.narg('nested_path')::text[] IS NOT NULL AND data#>>sqlc.narg('nested_path')::text[] = sqlc.narg('nested_value')::text) + ) +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset'); + +-- name: FindNodesByMultiFieldMatch :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE node_type = $1 + AND ( + -- Primary field match (required) + data->>sqlc.arg('primary_field')::text = sqlc.arg('primary_value')::text + -- Secondary field match (optional) + AND (sqlc.narg('secondary_field')::text IS NULL + OR data->>sqlc.narg('secondary_field')::text = sqlc.narg('secondary_value')::text) + -- Tertiary field match (optional) + AND (sqlc.narg('tertiary_field')::text IS NULL + OR data->>sqlc.narg('tertiary_field')::text = sqlc.narg('tertiary_value')::text) + ) +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset'); + +-- name: FindNodesByExactFieldMatch :many +-- Uses GIN index (idx_graph_nodes_data_gin) for fast exact field matching. +-- Callers pass match_filter as JSONB, e.g., '{"name": "exact_value"}' or '{"name": "x", "serial": "y"}'. +-- Fuzzy/pattern matching should be done in the application layer (matching package). +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE node_type = $1 + AND data @> sqlc.arg('match_filter')::jsonb +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset'); + +-- name: FindNodesByComplexMatch :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE node_type = $1 + AND ( + -- Support JSONB containment queries + (sqlc.narg('contains_filter')::jsonb IS NULL OR data @> sqlc.narg('contains_filter')::jsonb) + -- Support JSONB path existence + AND (sqlc.narg('path_exists')::text IS NULL OR data ? sqlc.narg('path_exists')::text) + -- Support regex matching on text fields + AND (sqlc.narg('regex_field')::text IS NULL OR sqlc.narg('regex_pattern')::text IS NULL + OR data->>sqlc.narg('regex_field')::text ~ sqlc.narg('regex_pattern')::text) + ) +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset'); + +-- Snapshot management queries + +-- name: InsertSnapshot :one +-- Atomically inserts a snapshot with next sequence number using advisory lock to prevent races. +-- Uses namespaced lock (1=snapshot_insert, node_id) to avoid clashes with other advisory locks. +WITH lock AS ( + SELECT pg_advisory_xact_lock(1, $1::int) +) +INSERT INTO graph_node_snapshots (node_id, snapshot_data, sequence_number) +SELECT $1, $2, COALESCE(MAX(sequence_number), 0) + 1 +FROM graph_node_snapshots, lock +WHERE node_id = $1 +RETURNING id, node_id, snapshot_data, sequence_number, created_at; + +-- name: GetLatestSnapshot :one +SELECT id, node_id, snapshot_data, sequence_number, created_at +FROM graph_node_snapshots +WHERE node_id = $1 +ORDER BY sequence_number DESC +LIMIT 1; + +-- name: GetSnapshotsByNode :many +SELECT id, node_id, snapshot_data, sequence_number, created_at +FROM graph_node_snapshots +WHERE node_id = $1 +ORDER BY sequence_number DESC +LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset'); + +-- name: CleanupOldSnapshots :exec +-- Deletes old snapshots keeping only the most recent $2 snapshots for a node. +-- WARNING: If $2 is 0, all snapshots for the node will be deleted. +DELETE FROM graph_node_snapshots outer_snap +WHERE outer_snap.node_id = $1 + AND outer_snap.id NOT IN ( + SELECT s.id + FROM graph_node_snapshots s + WHERE s.node_id = $1 + ORDER BY s.sequence_number DESC + LIMIT $2 + ); + +-- name: GetNodeWithLatestSnapshot :one +-- Note: snapshot fields use COALESCE defaults for NULL safety when no snapshot exists +-- sequence_number = 0 and empty snapshot_data indicate no snapshot +SELECT + n.id, n.external_id, n.node_type, n.data as matching_data, n.duplicate_count, n.matching_schema_version, n.created_at, n.updated_at, + COALESCE(s.snapshot_data, '{}'::jsonb) as snapshot_data, + COALESCE(s.sequence_number, 0) as sequence_number, + s.created_at as snapshot_created_at +FROM graph_nodes n +LEFT JOIN LATERAL ( + SELECT snapshot_data, sequence_number, created_at + FROM graph_node_snapshots + WHERE node_id = n.id + ORDER BY sequence_number DESC + LIMIT 1 +) s ON true +WHERE n.node_type = $1 AND n.external_id = $2; + +-- name: FindNodesNeedingSchemaUpdate :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE matching_schema_version < $1 +ORDER BY updated_at DESC +LIMIT sqlc.arg('limit') OFFSET sqlc.arg('offset'); \ No newline at end of file diff --git a/diode-server/gen/dbstore/postgres/change_sets.sql.go b/diode-server/gen/dbstore/postgres/change_sets.sql.go index 4408fe00..f67d1925 100644 --- a/diode-server/gen/dbstore/postgres/change_sets.sql.go +++ b/diode-server/gen/dbstore/postgres/change_sets.sql.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.28.0 +// sqlc v1.30.0 // source: change_sets.sql package postgres diff --git a/diode-server/gen/dbstore/postgres/db.go b/diode-server/gen/dbstore/postgres/db.go index fd4248fe..87370773 100644 --- a/diode-server/gen/dbstore/postgres/db.go +++ b/diode-server/gen/dbstore/postgres/db.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.28.0 +// sqlc v1.30.0 package postgres diff --git a/diode-server/gen/dbstore/postgres/deviations.sql.go b/diode-server/gen/dbstore/postgres/deviations.sql.go index b21c49f2..5f5a48a1 100644 --- a/diode-server/gen/dbstore/postgres/deviations.sql.go +++ b/diode-server/gen/dbstore/postgres/deviations.sql.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.28.0 +// sqlc v1.30.0 // source: deviations.sql package postgres diff --git a/diode-server/gen/dbstore/postgres/graph.sql.go b/diode-server/gen/dbstore/postgres/graph.sql.go new file mode 100644 index 00000000..735c6685 --- /dev/null +++ b/diode-server/gen/dbstore/postgres/graph.sql.go @@ -0,0 +1,942 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.30.0 +// source: graph.sql + +package postgres + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const cleanupOldSnapshots = `-- name: CleanupOldSnapshots :exec +DELETE FROM graph_node_snapshots outer_snap +WHERE outer_snap.node_id = $1 + AND outer_snap.id NOT IN ( + SELECT s.id + FROM graph_node_snapshots s + WHERE s.node_id = $1 + ORDER BY s.sequence_number DESC + LIMIT $2 + ) +` + +type CleanupOldSnapshotsParams struct { + NodeID int64 `json:"node_id"` + Limit int32 `json:"limit"` +} + +// Deletes old snapshots keeping only the most recent $2 snapshots for a node. +// WARNING: If $2 is 0, all snapshots for the node will be deleted. +func (q *Queries) CleanupOldSnapshots(ctx context.Context, arg CleanupOldSnapshotsParams) error { + _, err := q.db.Exec(ctx, cleanupOldSnapshots, arg.NodeID, arg.Limit) + return err +} + +const findGraphNode = `-- name: FindGraphNode :one +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version +FROM graph_nodes +WHERE node_type = $1 AND external_id = $2 +` + +type FindGraphNodeParams struct { + NodeType string `json:"node_type"` + ExternalID string `json:"external_id"` +} + +type FindGraphNodeRow struct { + ID int64 `json:"id"` + ExternalID string `json:"external_id"` + NodeType string `json:"node_type"` + Data []byte `json:"data"` + DuplicateCount int32 `json:"duplicate_count"` + MatchingSchemaVersion int32 `json:"matching_schema_version"` +} + +func (q *Queries) FindGraphNode(ctx context.Context, arg FindGraphNodeParams) (FindGraphNodeRow, error) { + row := q.db.QueryRow(ctx, findGraphNode, arg.NodeType, arg.ExternalID) + var i FindGraphNodeRow + err := row.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + ) + return i, err +} + +const findNodesByComplexMatch = `-- name: FindNodesByComplexMatch :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE node_type = $1 + AND ( + -- Support JSONB containment queries + ($2::jsonb IS NULL OR data @> $2::jsonb) + -- Support JSONB path existence + AND ($3::text IS NULL OR data ? $3::text) + -- Support regex matching on text fields + AND ($4::text IS NULL OR $5::text IS NULL + OR data->>$4::text ~ $5::text) + ) +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT $7 OFFSET $6 +` + +type FindNodesByComplexMatchParams struct { + NodeType string `json:"node_type"` + ContainsFilter []byte `json:"contains_filter"` + PathExists pgtype.Text `json:"path_exists"` + RegexField pgtype.Text `json:"regex_field"` + RegexPattern pgtype.Text `json:"regex_pattern"` + Offset int32 `json:"offset"` + Limit int32 `json:"limit"` +} + +func (q *Queries) FindNodesByComplexMatch(ctx context.Context, arg FindNodesByComplexMatchParams) ([]GraphNode, error) { + rows, err := q.db.Query(ctx, findNodesByComplexMatch, + arg.NodeType, + arg.ContainsFilter, + arg.PathExists, + arg.RegexField, + arg.RegexPattern, + arg.Offset, + arg.Limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphNode + for rows.Next() { + var i GraphNode + if err := rows.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const findNodesByExactFieldMatch = `-- name: FindNodesByExactFieldMatch :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE node_type = $1 + AND data @> $2::jsonb +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT $4 OFFSET $3 +` + +type FindNodesByExactFieldMatchParams struct { + NodeType string `json:"node_type"` + MatchFilter []byte `json:"match_filter"` + Offset int32 `json:"offset"` + Limit int32 `json:"limit"` +} + +// Uses GIN index (idx_graph_nodes_data_gin) for fast exact field matching. +// Callers pass match_filter as JSONB, e.g., '{"name": "exact_value"}' or '{"name": "x", "serial": "y"}'. +// Fuzzy/pattern matching should be done in the application layer (matching package). +func (q *Queries) FindNodesByExactFieldMatch(ctx context.Context, arg FindNodesByExactFieldMatchParams) ([]GraphNode, error) { + rows, err := q.db.Query(ctx, findNodesByExactFieldMatch, + arg.NodeType, + arg.MatchFilter, + arg.Offset, + arg.Limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphNode + for rows.Next() { + var i GraphNode + if err := rows.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const findNodesByFieldMatch = `-- name: FindNodesByFieldMatch :many + +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE node_type = $1 + AND ( + -- Support matching by single JSON field (only if json_field is provided) + ($2::text IS NOT NULL AND data->>$2::text = $3::text) + -- Support matching by nested JSON path (only if nested_path is provided) + OR ($4::text[] IS NOT NULL AND data#>>$4::text[] = $5::text) + ) +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT $7 OFFSET $6 +` + +type FindNodesByFieldMatchParams struct { + NodeType string `json:"node_type"` + JsonField pgtype.Text `json:"json_field"` + FieldValue pgtype.Text `json:"field_value"` + NestedPath []string `json:"nested_path"` + NestedValue pgtype.Text `json:"nested_value"` + Offset int32 `json:"offset"` + Limit int32 `json:"limit"` +} + +// Entity matching queries for confidence-based matching +func (q *Queries) FindNodesByFieldMatch(ctx context.Context, arg FindNodesByFieldMatchParams) ([]GraphNode, error) { + rows, err := q.db.Query(ctx, findNodesByFieldMatch, + arg.NodeType, + arg.JsonField, + arg.FieldValue, + arg.NestedPath, + arg.NestedValue, + arg.Offset, + arg.Limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphNode + for rows.Next() { + var i GraphNode + if err := rows.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const findNodesByMultiFieldMatch = `-- name: FindNodesByMultiFieldMatch :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE node_type = $1 + AND ( + -- Primary field match (required) + data->>$2::text = $3::text + -- Secondary field match (optional) + AND ($4::text IS NULL + OR data->>$4::text = $5::text) + -- Tertiary field match (optional) + AND ($6::text IS NULL + OR data->>$6::text = $7::text) + ) +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT $9 OFFSET $8 +` + +type FindNodesByMultiFieldMatchParams struct { + NodeType string `json:"node_type"` + PrimaryField string `json:"primary_field"` + PrimaryValue string `json:"primary_value"` + SecondaryField pgtype.Text `json:"secondary_field"` + SecondaryValue pgtype.Text `json:"secondary_value"` + TertiaryField pgtype.Text `json:"tertiary_field"` + TertiaryValue pgtype.Text `json:"tertiary_value"` + Offset int32 `json:"offset"` + Limit int32 `json:"limit"` +} + +func (q *Queries) FindNodesByMultiFieldMatch(ctx context.Context, arg FindNodesByMultiFieldMatchParams) ([]GraphNode, error) { + rows, err := q.db.Query(ctx, findNodesByMultiFieldMatch, + arg.NodeType, + arg.PrimaryField, + arg.PrimaryValue, + arg.SecondaryField, + arg.SecondaryValue, + arg.TertiaryField, + arg.TertiaryValue, + arg.Offset, + arg.Limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphNode + for rows.Next() { + var i GraphNode + if err := rows.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const findNodesNeedingSchemaUpdate = `-- name: FindNodesNeedingSchemaUpdate :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE matching_schema_version < $1 +ORDER BY updated_at DESC +LIMIT $3 OFFSET $2 +` + +type FindNodesNeedingSchemaUpdateParams struct { + MatchingSchemaVersion int32 `json:"matching_schema_version"` + Offset int32 `json:"offset"` + Limit int32 `json:"limit"` +} + +func (q *Queries) FindNodesNeedingSchemaUpdate(ctx context.Context, arg FindNodesNeedingSchemaUpdateParams) ([]GraphNode, error) { + rows, err := q.db.Query(ctx, findNodesNeedingSchemaUpdate, arg.MatchingSchemaVersion, arg.Offset, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphNode + for rows.Next() { + var i GraphNode + if err := rows.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getConnectedNodes = `-- name: GetConnectedNodes :many +SELECT + target.id, + target.external_id, + target.node_type, + target.data, + target.duplicate_count, + target.matching_schema_version, + edges.edge_type, + edges.properties as edge_properties +FROM graph_edges edges +JOIN graph_nodes target ON edges.target_node_id = target.id +WHERE edges.source_node_id = $1 +` + +type GetConnectedNodesRow struct { + ID int64 `json:"id"` + ExternalID string `json:"external_id"` + NodeType string `json:"node_type"` + Data []byte `json:"data"` + DuplicateCount int32 `json:"duplicate_count"` + MatchingSchemaVersion int32 `json:"matching_schema_version"` + EdgeType string `json:"edge_type"` + EdgeProperties []byte `json:"edge_properties"` +} + +func (q *Queries) GetConnectedNodes(ctx context.Context, sourceNodeID int64) ([]GetConnectedNodesRow, error) { + rows, err := q.db.Query(ctx, getConnectedNodes, sourceNodeID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetConnectedNodesRow + for rows.Next() { + var i GetConnectedNodesRow + if err := rows.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + &i.EdgeType, + &i.EdgeProperties, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getGraphNodesByType = `-- name: GetGraphNodesByType :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE node_type = $1 +ORDER BY updated_at DESC +LIMIT $3 OFFSET $2 +` + +type GetGraphNodesByTypeParams struct { + NodeType string `json:"node_type"` + Offset int32 `json:"offset"` + Limit int32 `json:"limit"` +} + +func (q *Queries) GetGraphNodesByType(ctx context.Context, arg GetGraphNodesByTypeParams) ([]GraphNode, error) { + rows, err := q.db.Query(ctx, getGraphNodesByType, arg.NodeType, arg.Offset, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphNode + for rows.Next() { + var i GraphNode + if err := rows.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getGraphStats = `-- name: GetGraphStats :one +SELECT + COUNT(*) as total_nodes, + COUNT(DISTINCT node_type) as unique_node_types, + SUM(duplicate_count) as total_observations, + AVG(duplicate_count) as avg_observations_per_node +FROM graph_nodes +` + +type GetGraphStatsRow struct { + TotalNodes int64 `json:"total_nodes"` + UniqueNodeTypes int64 `json:"unique_node_types"` + TotalObservations int64 `json:"total_observations"` + AvgObservationsPerNode float64 `json:"avg_observations_per_node"` +} + +func (q *Queries) GetGraphStats(ctx context.Context) (GetGraphStatsRow, error) { + row := q.db.QueryRow(ctx, getGraphStats) + var i GetGraphStatsRow + err := row.Scan( + &i.TotalNodes, + &i.UniqueNodeTypes, + &i.TotalObservations, + &i.AvgObservationsPerNode, + ) + return i, err +} + +const getGraphStatsByType = `-- name: GetGraphStatsByType :many +SELECT + node_type, + COUNT(*) as node_count, + SUM(duplicate_count) as total_observations, + AVG(duplicate_count) as avg_observations_per_node, + MAX(duplicate_count) as max_observations, + MIN(created_at) as first_seen, + MAX(updated_at) as last_seen +FROM graph_nodes +GROUP BY node_type +ORDER BY node_count DESC +` + +type GetGraphStatsByTypeRow struct { + NodeType string `json:"node_type"` + NodeCount int64 `json:"node_count"` + TotalObservations int64 `json:"total_observations"` + AvgObservationsPerNode float64 `json:"avg_observations_per_node"` + MaxObservations interface{} `json:"max_observations"` + FirstSeen interface{} `json:"first_seen"` + LastSeen interface{} `json:"last_seen"` +} + +func (q *Queries) GetGraphStatsByType(ctx context.Context) ([]GetGraphStatsByTypeRow, error) { + rows, err := q.db.Query(ctx, getGraphStatsByType) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetGraphStatsByTypeRow + for rows.Next() { + var i GetGraphStatsByTypeRow + if err := rows.Scan( + &i.NodeType, + &i.NodeCount, + &i.TotalObservations, + &i.AvgObservationsPerNode, + &i.MaxObservations, + &i.FirstSeen, + &i.LastSeen, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getLatestSnapshot = `-- name: GetLatestSnapshot :one +SELECT id, node_id, snapshot_data, sequence_number, created_at +FROM graph_node_snapshots +WHERE node_id = $1 +ORDER BY sequence_number DESC +LIMIT 1 +` + +func (q *Queries) GetLatestSnapshot(ctx context.Context, nodeID int64) (GraphNodeSnapshot, error) { + row := q.db.QueryRow(ctx, getLatestSnapshot, nodeID) + var i GraphNodeSnapshot + err := row.Scan( + &i.ID, + &i.NodeID, + &i.SnapshotData, + &i.SequenceNumber, + &i.CreatedAt, + ) + return i, err +} + +const getNodeWithLatestSnapshot = `-- name: GetNodeWithLatestSnapshot :one +SELECT + n.id, n.external_id, n.node_type, n.data as matching_data, n.duplicate_count, n.matching_schema_version, n.created_at, n.updated_at, + COALESCE(s.snapshot_data, '{}'::jsonb) as snapshot_data, + COALESCE(s.sequence_number, 0) as sequence_number, + s.created_at as snapshot_created_at +FROM graph_nodes n +LEFT JOIN LATERAL ( + SELECT snapshot_data, sequence_number, created_at + FROM graph_node_snapshots + WHERE node_id = n.id + ORDER BY sequence_number DESC + LIMIT 1 +) s ON true +WHERE n.node_type = $1 AND n.external_id = $2 +` + +type GetNodeWithLatestSnapshotParams struct { + NodeType string `json:"node_type"` + ExternalID string `json:"external_id"` +} + +type GetNodeWithLatestSnapshotRow struct { + ID int64 `json:"id"` + ExternalID string `json:"external_id"` + NodeType string `json:"node_type"` + MatchingData []byte `json:"matching_data"` + DuplicateCount int32 `json:"duplicate_count"` + MatchingSchemaVersion int32 `json:"matching_schema_version"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + SnapshotData []byte `json:"snapshot_data"` + SequenceNumber int32 `json:"sequence_number"` + SnapshotCreatedAt pgtype.Timestamptz `json:"snapshot_created_at"` +} + +// Note: snapshot fields use COALESCE defaults for NULL safety when no snapshot exists +// sequence_number = 0 and empty snapshot_data indicate no snapshot +func (q *Queries) GetNodeWithLatestSnapshot(ctx context.Context, arg GetNodeWithLatestSnapshotParams) (GetNodeWithLatestSnapshotRow, error) { + row := q.db.QueryRow(ctx, getNodeWithLatestSnapshot, arg.NodeType, arg.ExternalID) + var i GetNodeWithLatestSnapshotRow + err := row.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.MatchingData, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + &i.CreatedAt, + &i.UpdatedAt, + &i.SnapshotData, + &i.SequenceNumber, + &i.SnapshotCreatedAt, + ) + return i, err +} + +const getNodesByFrequency = `-- name: GetNodesByFrequency :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE duplicate_count >= $1 +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT $3 OFFSET $2 +` + +type GetNodesByFrequencyParams struct { + DuplicateCount int32 `json:"duplicate_count"` + Offset int32 `json:"offset"` + Limit int32 `json:"limit"` +} + +func (q *Queries) GetNodesByFrequency(ctx context.Context, arg GetNodesByFrequencyParams) ([]GraphNode, error) { + rows, err := q.db.Query(ctx, getNodesByFrequency, arg.DuplicateCount, arg.Offset, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphNode + for rows.Next() { + var i GraphNode + if err := rows.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getSnapshotsByNode = `-- name: GetSnapshotsByNode :many +SELECT id, node_id, snapshot_data, sequence_number, created_at +FROM graph_node_snapshots +WHERE node_id = $1 +ORDER BY sequence_number DESC +LIMIT $3 OFFSET $2 +` + +type GetSnapshotsByNodeParams struct { + NodeID int64 `json:"node_id"` + Offset int32 `json:"offset"` + Limit int32 `json:"limit"` +} + +func (q *Queries) GetSnapshotsByNode(ctx context.Context, arg GetSnapshotsByNodeParams) ([]GraphNodeSnapshot, error) { + rows, err := q.db.Query(ctx, getSnapshotsByNode, arg.NodeID, arg.Offset, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphNodeSnapshot + for rows.Next() { + var i GraphNodeSnapshot + if err := rows.Scan( + &i.ID, + &i.NodeID, + &i.SnapshotData, + &i.SequenceNumber, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const insertSnapshot = `-- name: InsertSnapshot :one + +WITH lock AS ( + SELECT pg_advisory_xact_lock(1, $1::int) +) +INSERT INTO graph_node_snapshots (node_id, snapshot_data, sequence_number) +SELECT $1, $2, COALESCE(MAX(sequence_number), 0) + 1 +FROM graph_node_snapshots, lock +WHERE node_id = $1 +RETURNING id, node_id, snapshot_data, sequence_number, created_at +` + +type InsertSnapshotParams struct { + NodeID int64 `json:"node_id"` + SnapshotData []byte `json:"snapshot_data"` +} + +// Snapshot management queries +// Atomically inserts a snapshot with next sequence number using advisory lock to prevent races. +// Uses namespaced lock (1=snapshot_insert, node_id) to avoid clashes with other advisory locks. +func (q *Queries) InsertSnapshot(ctx context.Context, arg InsertSnapshotParams) (GraphNodeSnapshot, error) { + row := q.db.QueryRow(ctx, insertSnapshot, arg.NodeID, arg.SnapshotData) + var i GraphNodeSnapshot + err := row.Scan( + &i.ID, + &i.NodeID, + &i.SnapshotData, + &i.SequenceNumber, + &i.CreatedAt, + ) + return i, err +} + +const searchGraphNodes = `-- name: SearchGraphNodes :many +SELECT id, external_id, node_type, data, duplicate_count, matching_schema_version, created_at, updated_at +FROM graph_nodes +WHERE (node_type = $1::text OR $1::text IS NULL) + AND LENGTH($2::text) > 0 + AND ( + external_id ILIKE '%' || $2::text || '%' + OR data::text ILIKE '%' || $2::text || '%' + ) +ORDER BY duplicate_count DESC, updated_at DESC +LIMIT $4 OFFSET $3 +` + +type SearchGraphNodesParams struct { + NodeType pgtype.Text `json:"node_type"` + SearchTerm string `json:"search_term"` + Offset int32 `json:"offset"` + Limit int32 `json:"limit"` +} + +// search_term is REQUIRED (non-null, non-empty string) to prevent full table scans. +// Empty string returns no results. Use GetGraphNodesByType for listing without filter. +// node_type is optional - pass NULL to search across all types. +// Note: Very long search_term values may cause slow scans on large datasets. +func (q *Queries) SearchGraphNodes(ctx context.Context, arg SearchGraphNodesParams) ([]GraphNode, error) { + rows, err := q.db.Query(ctx, searchGraphNodes, + arg.NodeType, + arg.SearchTerm, + arg.Offset, + arg.Limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphNode + for rows.Next() { + var i GraphNode + if err := rows.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const updateGraphNodeData = `-- name: UpdateGraphNodeData :one +UPDATE graph_nodes +SET data = $3, matching_schema_version = $4, updated_at = NOW() +WHERE node_type = $1 AND external_id = $2 +RETURNING id, external_id, node_type, data, duplicate_count, matching_schema_version +` + +type UpdateGraphNodeDataParams struct { + NodeType string `json:"node_type"` + ExternalID string `json:"external_id"` + Data []byte `json:"data"` + MatchingSchemaVersion int32 `json:"matching_schema_version"` +} + +type UpdateGraphNodeDataRow struct { + ID int64 `json:"id"` + ExternalID string `json:"external_id"` + NodeType string `json:"node_type"` + Data []byte `json:"data"` + DuplicateCount int32 `json:"duplicate_count"` + MatchingSchemaVersion int32 `json:"matching_schema_version"` +} + +func (q *Queries) UpdateGraphNodeData(ctx context.Context, arg UpdateGraphNodeDataParams) (UpdateGraphNodeDataRow, error) { + row := q.db.QueryRow(ctx, updateGraphNodeData, + arg.NodeType, + arg.ExternalID, + arg.Data, + arg.MatchingSchemaVersion, + ) + var i UpdateGraphNodeDataRow + err := row.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + ) + return i, err +} + +const upsertGraphEdge = `-- name: UpsertGraphEdge :exec +INSERT INTO graph_edges (source_node_id, target_node_id, edge_type, properties) +VALUES ($1, $2, $3, $4) +ON CONFLICT (source_node_id, target_node_id, edge_type) +DO UPDATE SET properties = EXCLUDED.properties +` + +type UpsertGraphEdgeParams struct { + SourceNodeID int64 `json:"source_node_id"` + TargetNodeID int64 `json:"target_node_id"` + EdgeType string `json:"edge_type"` + Properties []byte `json:"properties"` +} + +func (q *Queries) UpsertGraphEdge(ctx context.Context, arg UpsertGraphEdgeParams) error { + _, err := q.db.Exec(ctx, upsertGraphEdge, + arg.SourceNodeID, + arg.TargetNodeID, + arg.EdgeType, + arg.Properties, + ) + return err +} + +const upsertGraphEdgeWithConfidence = `-- name: UpsertGraphEdgeWithConfidence :exec +INSERT INTO graph_edges (source_node_id, target_node_id, edge_type, properties, confidence_score, match_reason, matching_fields, edge_subtype) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8) +ON CONFLICT (source_node_id, target_node_id, edge_type) +DO UPDATE SET + properties = EXCLUDED.properties, + confidence_score = EXCLUDED.confidence_score, + match_reason = EXCLUDED.match_reason, + matching_fields = EXCLUDED.matching_fields, + edge_subtype = EXCLUDED.edge_subtype +` + +type UpsertGraphEdgeWithConfidenceParams struct { + SourceNodeID int64 `json:"source_node_id"` + TargetNodeID int64 `json:"target_node_id"` + EdgeType string `json:"edge_type"` + Properties []byte `json:"properties"` + ConfidenceScore float32 `json:"confidence_score"` + MatchReason pgtype.Text `json:"match_reason"` + MatchingFields []byte `json:"matching_fields"` + EdgeSubtype pgtype.Text `json:"edge_subtype"` +} + +func (q *Queries) UpsertGraphEdgeWithConfidence(ctx context.Context, arg UpsertGraphEdgeWithConfidenceParams) error { + _, err := q.db.Exec(ctx, upsertGraphEdgeWithConfidence, + arg.SourceNodeID, + arg.TargetNodeID, + arg.EdgeType, + arg.Properties, + arg.ConfidenceScore, + arg.MatchReason, + arg.MatchingFields, + arg.EdgeSubtype, + ) + return err +} + +const upsertGraphNode = `-- name: UpsertGraphNode :one +INSERT INTO graph_nodes (external_id, node_type, data, duplicate_count, matching_schema_version) +VALUES ($1, $2, $3, 1, $4) +ON CONFLICT (node_type, external_id) +DO UPDATE SET + data = EXCLUDED.data, + duplicate_count = graph_nodes.duplicate_count + 1, + matching_schema_version = EXCLUDED.matching_schema_version, + updated_at = NOW() +RETURNING id, external_id, node_type, data, duplicate_count, matching_schema_version +` + +type UpsertGraphNodeParams struct { + ExternalID string `json:"external_id"` + NodeType string `json:"node_type"` + Data []byte `json:"data"` + MatchingSchemaVersion int32 `json:"matching_schema_version"` +} + +type UpsertGraphNodeRow struct { + ID int64 `json:"id"` + ExternalID string `json:"external_id"` + NodeType string `json:"node_type"` + Data []byte `json:"data"` + DuplicateCount int32 `json:"duplicate_count"` + MatchingSchemaVersion int32 `json:"matching_schema_version"` +} + +func (q *Queries) UpsertGraphNode(ctx context.Context, arg UpsertGraphNodeParams) (UpsertGraphNodeRow, error) { + row := q.db.QueryRow(ctx, upsertGraphNode, + arg.ExternalID, + arg.NodeType, + arg.Data, + arg.MatchingSchemaVersion, + ) + var i UpsertGraphNodeRow + err := row.Scan( + &i.ID, + &i.ExternalID, + &i.NodeType, + &i.Data, + &i.DuplicateCount, + &i.MatchingSchemaVersion, + ) + return i, err +} diff --git a/diode-server/gen/dbstore/postgres/ingestion_logs.sql.go b/diode-server/gen/dbstore/postgres/ingestion_logs.sql.go index ab8df19a..74809c2c 100644 --- a/diode-server/gen/dbstore/postgres/ingestion_logs.sql.go +++ b/diode-server/gen/dbstore/postgres/ingestion_logs.sql.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.28.0 +// sqlc v1.30.0 // source: ingestion_logs.sql package postgres diff --git a/diode-server/gen/dbstore/postgres/types.go b/diode-server/gen/dbstore/postgres/types.go index 7c76b80e..9db45ca9 100644 --- a/diode-server/gen/dbstore/postgres/types.go +++ b/diode-server/gen/dbstore/postgres/types.go @@ -1,6 +1,6 @@ // Code generated by sqlc. DO NOT EDIT. // versions: -// sqlc v1.28.0 +// sqlc v1.30.0 package postgres @@ -38,6 +38,45 @@ type ChangeSet struct { UpdatedAt pgtype.Timestamptz `json:"updated_at"` } +type GraphEdge struct { + ID int64 `json:"id"` + SourceNodeID int64 `json:"source_node_id"` + TargetNodeID int64 `json:"target_node_id"` + EdgeType string `json:"edge_type"` + // Subtype of edge based on confidence: high_confidence, medium_confidence, low_confidence, possible_duplicate + EdgeSubtype pgtype.Text `json:"edge_subtype"` + Properties []byte `json:"properties"` + // Confidence score (0.0-1.0) for the relationship match + ConfidenceScore float32 `json:"confidence_score"` + // Human-readable explanation of why entities were matched + MatchReason pgtype.Text `json:"match_reason"` + // JSON array of field names that contributed to the match + MatchingFields []byte `json:"matching_fields"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type GraphNode struct { + ID int64 `json:"id"` + ExternalID string `json:"external_id"` + NodeType string `json:"node_type"` + Data []byte `json:"data"` + DuplicateCount int32 `json:"duplicate_count"` + // Version of the matching configuration used to extract matching attributes + MatchingSchemaVersion int32 `json:"matching_schema_version"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type GraphNodeSnapshot struct { + ID int64 `json:"id"` + NodeID int64 `json:"node_id"` + // Complete entity protojson snapshot for historical tracking + SnapshotData []byte `json:"snapshot_data"` + // Sequential number for ordering snapshots, higher is newer + SequenceNumber int32 `json:"sequence_number"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + type IngestionLog struct { ID int32 `json:"id"` ExternalID string `json:"external_id"`