package repository import ( "bugulma/backend/internal/domain" "context" "encoding/json" "fmt" "time" "github.com/neo4j/neo4j-go-driver/v5/neo4j" ) // GraphSiteRepository manages Site nodes in Neo4j type GraphSiteRepository struct { driver neo4j.DriverWithContext database string } // NewGraphSiteRepository creates a new graph site repository func NewGraphSiteRepository(driver neo4j.DriverWithContext, dbName string) *GraphSiteRepository { return &GraphSiteRepository{ driver: driver, database: dbName, } } // SyncToGraph syncs a site to the graph database and creates OPERATES_AT relationship func (r *GraphSiteRepository) SyncToGraph(ctx context.Context, site *domain.Site) 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 availableUtilitiesJSON, _ := json.Marshal(site.AvailableUtilities) wasteManagementJSON, _ := json.Marshal(site.WasteManagement) sourcesJSON, _ := json.Marshal(site.Sources) // Create/update Site node siteCypher := ` MERGE (s:Site {id: $id}) SET s.name = $name, s.site_type = $site_type, s.floor_area_m2 = $floor_area_m2, s.ownership = $ownership, s.latitude = $latitude, s.longitude = $longitude, s.owner_organization_id = $owner_organization_id, s.available_utilities = $available_utilities, s.parking_spaces = $parking_spaces, s.loading_docks = $loading_docks, s.crane_capacity_tonnes = $crane_capacity_tonnes, s.energy_rating = $energy_rating, s.waste_management = $waste_management, s.environmental_impact = $environmental_impact, s.year_built = $year_built, s.builder_owner = $builder_owner, s.architect = $architect, s.original_purpose = $original_purpose, s.current_use = $current_use, s.style = $style, s.materials = $materials, s.storeys = $storeys, s.heritage_status = $heritage_status, s.notes = $notes, s.sources = $sources, s.created_at = $created_at, s.updated_at = $updated_at ` // Add location point if coordinates are available if site.Latitude != 0 && site.Longitude != 0 { siteCypher += `, s.location = point({latitude: $latitude, longitude: $longitude})` } siteCypher += ` RETURN s.id ` siteParams := map[string]interface{}{ "id": site.ID, "name": site.Name, "site_type": site.SiteType, "floor_area_m2": site.FloorAreaM2, "ownership": site.Ownership, "latitude": site.Latitude, "longitude": site.Longitude, "owner_organization_id": site.OwnerOrganizationID, "available_utilities": string(availableUtilitiesJSON), "parking_spaces": site.ParkingSpaces, "loading_docks": site.LoadingDocks, "crane_capacity_tonnes": site.CraneCapacityTonnes, "energy_rating": site.EnergyRating, "waste_management": string(wasteManagementJSON), "environmental_impact": site.EnvironmentalImpact, "year_built": site.YearBuilt, "builder_owner": site.BuilderOwner, "architect": site.Architect, "original_purpose": site.OriginalPurpose, "current_use": site.CurrentUse, "style": site.Style, "materials": site.Materials, "storeys": site.Storeys, "heritage_status": site.HeritageStatus, "notes": site.Notes, "sources": string(sourcesJSON), "created_at": site.CreatedAt.Format(time.RFC3339), "updated_at": site.UpdatedAt.Format(time.RFC3339), } result, err := tx.Run(ctx, siteCypher, siteParams) if err != nil { return nil, fmt.Errorf("failed to execute site merge: %w", err) } if err := result.Err(); err != nil { return nil, fmt.Errorf("failed to consume site result: %w", err) } // Create OPERATES_AT relationship if owner organization exists if site.OwnerOrganizationID != "" { relCypher := ` MATCH (s:Site {id: $site_id}) OPTIONAL MATCH (o:Organization {id: $org_id}) FOREACH (_ IN CASE WHEN o IS NOT NULL THEN [1] ELSE [] END | MERGE (o)-[:OPERATES_AT]->(s) ) ` relParams := map[string]interface{}{ "site_id": site.ID, "org_id": site.OwnerOrganizationID, } relResult, err := tx.Run(ctx, relCypher, relParams) if err != nil { return nil, fmt.Errorf("failed to create OPERATES_AT relationship: %w", err) } if err := relResult.Err(); err != nil { return nil, fmt.Errorf("failed to consume relationship result: %w", err) } } return nil, nil }) return err } // DeleteFromGraph deletes a site from the graph database func (r *GraphSiteRepository) 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 (s:Site {id: $id}) DETACH DELETE s ` result, err := tx.Run(ctx, cypher, map[string]interface{}{"id": id}) if err != nil { return nil, fmt.Errorf("failed to delete site: %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 site from the graph func (r *GraphSiteRepository) GetByID(ctx context.Context, id string) (*domain.Site, 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 (s:Site {id: $id}) RETURN s ` 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("s") if !ok { return nil, fmt.Errorf("site node not found in result") } return r.nodeToSite(node.(neo4j.Node)) } return nil, fmt.Errorf("site not found: %s", id) }) if err != nil { return nil, err } return result.(*domain.Site), nil } // GetNearby finds sites within a radius using spatial queries func (r *GraphSiteRepository) GetNearby(ctx context.Context, lat, lng, radiusKm float64, limit int) ([]*domain.Site, 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) { // Use point.distance for spatial queries cypher := ` MATCH (s:Site) WHERE s.latitude IS NOT NULL AND s.longitude IS NOT NULL WITH s, point.distance( point({latitude: $lat, longitude: $lng}), point({latitude: s.latitude, longitude: s.longitude}) ) / 1000.0 AS distance_km WHERE distance_km <= $radius_km RETURN s, distance_km ORDER BY distance_km LIMIT $limit ` params := map[string]interface{}{ "lat": lat, "lng": lng, "radius_km": radiusKm, "limit": limit, } result, err := tx.Run(ctx, cypher, params) if err != nil { return nil, err } var sites []*domain.Site for result.Next(ctx) { record := result.Record() node, ok := record.Get("s") if !ok { continue } site, err := r.nodeToSite(node.(neo4j.Node)) if err != nil { return nil, err } sites = append(sites, site) } if err := result.Err(); err != nil { return nil, err } return sites, nil }) if err != nil { return nil, err } return result.([]*domain.Site), nil } // nodeToSite converts a Neo4j node to a Site domain object func (r *GraphSiteRepository) nodeToSite(node neo4j.Node) (*domain.Site, error) { props := node.Props site := &domain.Site{ ID: props["id"].(string), Name: getStringProp(props, "name"), OwnerOrganizationID: getStringProp(props, "owner_organization_id"), EnergyRating: getStringProp(props, "energy_rating"), EnvironmentalImpact: getStringProp(props, "environmental_impact"), YearBuilt: getStringProp(props, "year_built"), BuilderOwner: getStringProp(props, "builder_owner"), Architect: getStringProp(props, "architect"), OriginalPurpose: getStringProp(props, "original_purpose"), CurrentUse: getStringProp(props, "current_use"), Style: getStringProp(props, "style"), Materials: getStringProp(props, "materials"), HeritageStatus: getStringProp(props, "heritage_status"), Notes: getStringProp(props, "notes"), } // Parse enum fields if siteTypeStr, ok := props["site_type"].(string); ok { site.SiteType = domain.SiteType(siteTypeStr) } if ownershipStr, ok := props["ownership"].(string); ok { site.Ownership = domain.Ownership(ownershipStr) } // Parse numeric fields if val, ok := props["floor_area_m2"].(float64); ok { site.FloorAreaM2 = val } if val, ok := props["latitude"].(float64); ok { site.Latitude = val } if val, ok := props["longitude"].(float64); ok { site.Longitude = val } if val, ok := props["parking_spaces"].(int64); ok { site.ParkingSpaces = int(val) } if val, ok := props["loading_docks"].(int64); ok { site.LoadingDocks = int(val) } if val, ok := props["crane_capacity_tonnes"].(float64); ok { site.CraneCapacityTonnes = val } if val, ok := props["storeys"].(int64); ok { site.Storeys = int(val) } // Parse timestamps if createdAt, ok := props["created_at"].(string); ok { if t, err := time.Parse(time.RFC3339, createdAt); err == nil { site.CreatedAt = t } } if updatedAt, ok := props["updated_at"].(string); ok { if t, err := time.Parse(time.RFC3339, updatedAt); err == nil { site.UpdatedAt = t } } return site, nil }