package repository import ( "bugulma/backend/internal/domain" "context" "encoding/json" "fmt" "time" "github.com/neo4j/neo4j-go-driver/v5/neo4j" ) // GraphMatchRepository manages Match nodes in Neo4j type GraphMatchRepository struct { driver neo4j.DriverWithContext database string } // NewGraphMatchRepository creates a new graph match repository func NewGraphMatchRepository(driver neo4j.DriverWithContext, dbName string) *GraphMatchRepository { return &GraphMatchRepository{ driver: driver, database: dbName, } } // SyncToGraph syncs a match to the graph database and creates MATCHES relationship func (r *GraphMatchRepository) SyncToGraph(ctx context.Context, match *domain.Match) error { session := r.driver.NewSession(ctx, neo4j.SessionConfig{ AccessMode: neo4j.AccessModeWrite, DatabaseName: r.database, }) defer session.Close(ctx) // Use ExecuteWrite for automatic retry logic _, err := session.ExecuteWrite(ctx, func(tx neo4j.ManagedTransaction) (interface{}, error) { // Marshal JSONB fields riskAssessmentJSON, _ := json.Marshal(match.RiskAssessment) historyJSON, _ := json.Marshal(match.History) contractDetailsJSON, _ := json.Marshal(match.ContractDetails) // Create/update Match node matchCypher := ` MERGE (m:Match {id: $id}) SET m.source_resource_id = $source_resource_id, m.target_resource_id = $target_resource_id, m.status = $status, m.compatibility_score = $compatibility_score, m.temporal_overlap_score = $temporal_overlap_score, m.quality_score = $quality_score, m.economic_value = $economic_value, m.distance_km = $distance_km, m.priority = $priority, m.risk_assessment = $risk_assessment, m.history = $history, m.contract_details = $contract_details, m.failure_reason = $failure_reason, m.version = $version, m.created_at = $created_at, m.updated_at = $updated_at RETURN m.id ` params := map[string]interface{}{ "id": match.ID, "source_resource_id": match.SourceResourceID, "target_resource_id": match.TargetResourceID, "status": string(match.Status), "compatibility_score": match.CompatibilityScore, "temporal_overlap_score": match.TemporalOverlapScore, "quality_score": match.QualityScore, "economic_value": match.EconomicValue, "distance_km": match.DistanceKm, "priority": match.Priority, "risk_assessment": string(riskAssessmentJSON), "history": string(historyJSON), "contract_details": string(contractDetailsJSON), "failure_reason": match.FailureReason, "version": match.Version, "created_at": match.CreatedAt.Format(time.RFC3339), "updated_at": match.UpdatedAt.Format(time.RFC3339), } result, err := tx.Run(ctx, matchCypher, params) if err != nil { return nil, fmt.Errorf("failed to execute match merge: %w", err) } if err := result.Err(); err != nil { return nil, fmt.Errorf("failed to consume match result: %w", err) } // Create ResourceFlow -> MATCHES -> ResourceFlow relationship // Using separate query for clarity and error handling relCypher := ` MATCH (m:Match {id: $match_id}) OPTIONAL MATCH (source:ResourceFlow {id: $source_resource_id}) OPTIONAL MATCH (target:ResourceFlow {id: $target_resource_id}) FOREACH (_ IN CASE WHEN source IS NOT NULL AND target IS NOT NULL THEN [1] ELSE [] END | MERGE (source)-[r:MATCHES]->(target) SET r.match_id = $match_id, r.compatibility_score = $compatibility_score, r.status = $status, r.distance_km = $distance_km, r.economic_value = $economic_value, r.created_at = $created_at ) ` relParams := map[string]interface{}{ "match_id": match.ID, "source_resource_id": match.SourceResourceID, "target_resource_id": match.TargetResourceID, "compatibility_score": match.CompatibilityScore, "status": string(match.Status), "distance_km": match.DistanceKm, "economic_value": match.EconomicValue, "created_at": match.CreatedAt.Format(time.RFC3339), } relResult, err := tx.Run(ctx, relCypher, relParams) if err != nil { return nil, fmt.Errorf("failed to create MATCHES relationship: %w", err) } if err := relResult.Err(); err != nil { return nil, fmt.Errorf("failed to consume MATCHES relationship result: %w", err) } return nil, nil }) return err } // DeleteFromGraph deletes a match from the graph database func (r *GraphMatchRepository) DeleteFromGraph(ctx context.Context, id string) error { session := r.driver.NewSession(ctx, neo4j.SessionConfig{ AccessMode: neo4j.AccessModeWrite, DatabaseName: r.database, }) defer session.Close(ctx) _, err := session.ExecuteWrite(ctx, func(tx neo4j.ManagedTransaction) (interface{}, error) { cypher := ` MATCH (m:Match {id: $id}) DETACH DELETE m ` result, err := tx.Run(ctx, cypher, map[string]interface{}{"id": id}) if err != nil { return nil, fmt.Errorf("failed to delete match: %w", err) } if err := result.Err(); err != nil { return nil, fmt.Errorf("failed to consume delete result: %w", err) } return nil, nil }) return err } // GetByID retrieves a match from the graph func (r *GraphMatchRepository) GetByID(ctx context.Context, id string) (*domain.Match, error) { session := r.driver.NewSession(ctx, neo4j.SessionConfig{ AccessMode: neo4j.AccessModeRead, DatabaseName: r.database, }) defer session.Close(ctx) result, err := session.ExecuteRead(ctx, func(tx neo4j.ManagedTransaction) (interface{}, error) { cypher := ` MATCH (m:Match {id: $id}) RETURN m ` result, err := tx.Run(ctx, cypher, map[string]interface{}{"id": id}) if err != nil { return nil, err } if result.Next(ctx) { record := result.Record() node, ok := record.Get("m") if !ok { return nil, fmt.Errorf("match node not found in result") } return r.nodeToMatch(node.(neo4j.Node)) } return nil, fmt.Errorf("match not found: %s", id) }) if err != nil { return nil, err } return result.(*domain.Match), nil } // FindByStatus retrieves matches by status using graph query func (r *GraphMatchRepository) FindByStatus(ctx context.Context, status domain.MatchStatus, limit int) ([]*domain.Match, error) { session := r.driver.NewSession(ctx, neo4j.SessionConfig{ AccessMode: neo4j.AccessModeRead, DatabaseName: r.database, }) defer session.Close(ctx) result, err := session.ExecuteRead(ctx, func(tx neo4j.ManagedTransaction) (interface{}, error) { cypher := ` MATCH (m:Match {status: $status}) RETURN m ORDER BY m.compatibility_score DESC, m.created_at DESC LIMIT $limit ` params := map[string]interface{}{ "status": string(status), "limit": limit, } result, err := tx.Run(ctx, cypher, params) if err != nil { return nil, err } var matches []*domain.Match for result.Next(ctx) { record := result.Record() node, ok := record.Get("m") if !ok { continue } match, err := r.nodeToMatch(node.(neo4j.Node)) if err != nil { return nil, err } matches = append(matches, match) } if err := result.Err(); err != nil { return nil, err } return matches, nil }) if err != nil { return nil, err } return result.([]*domain.Match), nil } // GetMatchPath retrieves the complete path with organizations and sites involved in a match func (r *GraphMatchRepository) GetMatchPath(ctx context.Context, matchID string) (map[string]interface{}, error) { session := r.driver.NewSession(ctx, neo4j.SessionConfig{ AccessMode: neo4j.AccessModeRead, DatabaseName: r.database, }) defer session.Close(ctx) result, err := session.ExecuteRead(ctx, func(tx neo4j.ManagedTransaction) (interface{}, error) { // Get complete context of a match including organizations and sites cypher := ` MATCH (m:Match {id: $match_id}) MATCH (source:ResourceFlow {id: m.source_resource_id}) MATCH (target:ResourceFlow {id: m.target_resource_id}) OPTIONAL MATCH (sourceSite:Site)-[:HOSTS]->(source) OPTIONAL MATCH (targetSite:Site)-[:HOSTS]->(target) OPTIONAL MATCH (sourceOrg:Organization)-[:OPERATES_AT]->(sourceSite) OPTIONAL MATCH (targetOrg:Organization)-[:OPERATES_AT]->(targetSite) RETURN m AS match, source AS source_flow, target AS target_flow, sourceSite AS source_site, targetSite AS target_site, sourceOrg AS source_org, targetOrg AS target_org ` result, err := tx.Run(ctx, cypher, map[string]interface{}{"match_id": matchID}) if err != nil { return nil, err } if result.Next(ctx) { record := result.Record() matchPath := make(map[string]interface{}) // Extract all nodes from the path for _, key := range record.Keys { if val, ok := record.Get(key); ok && val != nil { matchPath[key] = val } } return matchPath, nil } return nil, fmt.Errorf("match path not found: %s", matchID) }) if err != nil { return nil, err } return result.(map[string]interface{}), nil } // nodeToMatch converts a Neo4j node to a Match domain object func (r *GraphMatchRepository) nodeToMatch(node neo4j.Node) (*domain.Match, error) { props := node.Props match := &domain.Match{ ID: props["id"].(string), SourceResourceID: getStringProp(props, "source_resource_id"), TargetResourceID: getStringProp(props, "target_resource_id"), FailureReason: getStringProp(props, "failure_reason"), } // Parse status enum if statusStr, ok := props["status"].(string); ok { match.Status = domain.MatchStatus(statusStr) } // Parse numeric fields if val, ok := props["compatibility_score"].(float64); ok { match.CompatibilityScore = val } if val, ok := props["temporal_overlap_score"].(float64); ok { match.TemporalOverlapScore = val } if val, ok := props["quality_score"].(float64); ok { match.QualityScore = val } if val, ok := props["economic_value"].(float64); ok { match.EconomicValue = val } if val, ok := props["distance_km"].(float64); ok { match.DistanceKm = val } // Parse integer fields if val, ok := props["priority"].(int64); ok { match.Priority = int(val) } if val, ok := props["version"].(int64); ok { match.Version = int(val) } // Parse timestamps if createdAt, ok := props["created_at"].(string); ok { if t, err := time.Parse(time.RFC3339, createdAt); err == nil { match.CreatedAt = t } } if updatedAt, ok := props["updated_at"].(string); ok { if t, err := time.Parse(time.RFC3339, updatedAt); err == nil { match.UpdatedAt = t } } return match, nil }