What do you think Dad ?

.gitignore

@@ -88,3 +88,7 @@ src/spark/*.ipynb
 
 # User environment file
 .env
+
+# Mac-specific files
+.DS_Store
+.idea

KillrVideo.Cassandra/KillrVideo.Cassandra.csproj

@@ -0,0 +1,12 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>netcoreapp2.2</TargetFramework>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <PackageReference Include="Dse" Version="2.8.0" />
+      <PackageReference Include="System.ComponentModel.Composition" Version="4.5.0" />
+    </ItemGroup>
+
+</Project>

src/KillrVideo.Cassandra/MicrosecondsSinceEpoch.cs → KillrVideo.Cassandra/MicrosecondsSinceEpoch.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 
 namespace KillrVideo.Cassandra
 {
@@ -42,4 +42,4 @@ public static long ToMicrosecondsSinceEpoch(this DateTimeOffset dateTimeOffset)
             return FromDateTimeOffset(dateTimeOffset);
         }
     }
-}
+}

src/KillrVideo.Cassandra/PreparedStatementCache.cs → KillrVideo.Cassandra/PreparedStatementCache.cs

@@ -1,26 +1,33 @@
-using System;
+using System;
 using System.Collections.Concurrent;
 using System.ComponentModel.Composition;
 using System.Linq;
 using System.Runtime.CompilerServices;
 using System.Threading.Tasks;
-using Cassandra;
+using Dse;
 
 namespace KillrVideo.Cassandra
 {
     /// <summary>
     /// A cache for PreparedStatements based on the CQL string.
     /// </summary>
     [Export]
-    public class PreparedStatementCache
-    {
-        private readonly ISession _session;
+    public class PreparedStatementCache {
+
+        /// <summary>
+        /// Inject Dse Session
+        /// </summary>
+        private readonly IDseSession _session;
+
+        /// <summary>
+        /// Working Cache
+        /// </summary>
         private readonly ConcurrentDictionary<string, Lazy<Task<PreparedStatement>>> _cachedItems;
 
         /// <summary>
         /// Create a new TaskCache using the provided factoryFunc to generate items from keys.
         /// </summary>
-        public PreparedStatementCache(ISession session)
+        public PreparedStatementCache(IDseSession session)
         {
             if (session == null) throw new ArgumentNullException(nameof(session));
             _session = session;

src/KillrVideo.Comments/CommentsService.cs → KillrVideo.Comments/CommentsService.cs

@@ -2,10 +2,12 @@
 using System.ComponentModel.Composition;
 using System.Linq;
 using System.Threading.Tasks;
-using Cassandra;
+//using Cassandra;
+using Dse;
 using Google.Protobuf.Reflection;
 using Google.Protobuf.WellKnownTypes;
 using Grpc.Core;
+using Serilog;
 using KillrVideo.Cassandra;
 using KillrVideo.Comments.Events;
 using KillrVideo.MessageBus;
@@ -20,13 +22,30 @@ namespace KillrVideo.Comments
     [Export(typeof(IGrpcServerService))]
     public class CommentsServiceImpl : CommentsService.CommentsServiceBase, IGrpcServerService
     {
-        private readonly ISession _session;
-        private readonly IBus _bus;
+        /// <summary>
+        /// Logger for the class to write into both files and console
+        /// </summary>
+        private static readonly ILogger Logger = Log.ForContext(typeof(CommentsServiceImpl));
+
+        /// <summary>
+        /// Inject Dse Session.
+        /// </summary>
+        private readonly IDseSession _session;
+
+        /// <summary>
+        /// Cache of results
+        /// </summary>
         private readonly PreparedStatementCache _statementCache;
 
+        /// <summary>
+        /// Exchange messages between services.
+        /// </summary>
+        private readonly IBus _bus;
+
+
         public ServiceDescriptor Descriptor => CommentsService.Descriptor;
 
-        public CommentsServiceImpl(ISession session, PreparedStatementCache statementCache, IBus bus)
+        public CommentsServiceImpl(IDseSession session, PreparedStatementCache statementCache, IBus bus)
         {
             if (session == null) throw new ArgumentNullException(nameof(session));
             if (statementCache == null) throw new ArgumentNullException(nameof(statementCache));

KillrVideo.Comments/KillrVideo.Comments.csproj

@@ -0,0 +1,22 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>netcoreapp2.2</TargetFramework>
+    </PropertyGroup>
+
+    <ItemGroup>
+      <PackageReference Include="Dse" Version="2.8.0" />
+      <PackageReference Include="Google.Protobuf" Version="3.10.0-rc1" />
+      <PackageReference Include="Grpc.Core" Version="2.24.0-pre1" />
+      <PackageReference Include="Serilog" Version="2.8.0" />
+      <PackageReference Include="System.ComponentModel" Version="4.3.0" />
+      <PackageReference Include="System.ComponentModel.Composition" Version="4.5.0" />
+    </ItemGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\KillrVideo.Cassandra\KillrVideo.Cassandra.csproj" />
+      <ProjectReference Include="..\KillrVideo.MessageBus\KillrVideo.MessageBus.csproj" />
+      <ProjectReference Include="..\KillrVideo.Protobuf\KillrVideo.Protobuf.csproj" />
+    </ItemGroup>
+
+</Project>

KillrVideo.GraphDsl/Dsl/Enrichment.cs

@@ -0,0 +1,94 @@
+using System.Collections.Generic;
+using System.Linq;
+using Gremlin.Net.Process.Traversal;
+using static KillrVideo.GraphDsl.__KillrVideo;
+
+namespace KillrVideo.GraphDsl.Dsl
+{
+    /// <summary>
+    /// Provides for pre-built data enrichment options for the <code>enrich(Enrichment...)</code> step. These options will
+    /// include extra information about the <code>Vertex</code>> when output from that step. Note that the enrichment 
+    /// examples presented here are examples to demonstrate this concept. The primary lesson here is to show how one might 
+    /// merge map results as part of a DSL. These enrichment options may not be suitable for traversals in production systems 
+    /// as counting all edges might add an unreasonable amount of time to an otherwise fast traversal.
+    /// </summary>
+    public class Enrichment
+    {
+        private List<string> projectedKeys;
+
+        private List<GraphTraversal<object, object>> traversals;
+
+        private Enrichment(string projectedKeys, GraphTraversal<object, object> traversal) :
+            this(new List<string>() { projectedKeys }, new List<GraphTraversal<object, object>>() { traversal })
+        {
+        }
+
+        private Enrichment(List<string> projectedKeys, List<GraphTraversal<object, object>> traversals)
+        {
+            this.projectedKeys = projectedKeys;
+            this.traversals = traversals;
+        }
+
+        public List<GraphTraversal<object, object>> GetTraversals()
+        {
+            return traversals;
+        }
+
+        public List<string> GetProjectedKeys()
+        {
+            return projectedKeys;
+        }
+
+        /// <summary>
+        /// Include the <code>Vertex</code> itself as a value in the enriched output which might be helpful if additional
+        /// traversing on that element is required.
+        /// </summary>
+        public static Enrichment Vertex()
+        {
+            return new Enrichment(KeyVertex, __.Identity());
+        }
+
+        /// <summary>
+        /// The number of incoming edges on the <code>Vertex</code>.
+        /// </summary>
+        public static Enrichment InDegree()
+        {
+            return new Enrichment(KeyInDegree, __.Map<object>(__.InE().Count()));
+        }
+
+
+        /// <summary>
+        /// The number of outgoing edges on the <code>Vertex</code>.
+        /// </summary>
+        public static Enrichment OutDegree()
+        {
+            return new Enrichment(KeyOutDegree, __.Map<object>(__.OutE().Count()));
+        }
+
+        /// <summary>
+        /// The total number of in and out edges on the <code>Vertex</code>.
+        /// </summary>
+        public static Enrichment Degree()
+        {
+            return new Enrichment(KeyDegree, __.Map<object>(__.BothE().Count()));
+        }
+
+        /// <summary>
+        /// Calculates the edge label distribution for the <code>Vertex</code>>.
+        /// </summary>
+        public static Enrichment Distribution()
+        {
+            return new Enrichment(KeyDistribution, __.Map<object>(__.BothE().GroupCount<string>().By(T.Label)));
+        }
+
+        /// <summary>
+        /// Chooses the keys to include in the output.
+        /// </summary>
+        public static Enrichment Keys(params string[] keys)
+        {
+            var valueTraversals = keys.Select(k => __.Values<object>(k)).ToList();
+            return new Enrichment(keys.ToList(), valueTraversals);
+        }
+    }
+}
+

KillrVideo.GraphDsl/Dsl/KillrVideoGraphTraversalExtensions.cs

@@ -0,0 +1,101 @@
+using System.Collections.Generic;
+using System.Linq;
+using Gremlin.Net.Process.Traversal;
+using Gremlin.Net.Structure;
+using static Gremlin.Net.Process.Traversal.P;
+
+using static KillrVideo.GraphDsl.__KillrVideo;
+
+namespace KillrVideo.GraphDsl.Dsl {
+
+    /// <summary>
+    /// The KillrVideo Traversal class which exposes the available steps of the DSL.
+    /// </summary>
+    public static class KillrVideoGraphTraversalExtensions {
+
+        /// <summary>
+        /// Traverses from a "user" to an "videos" over the "rated" edge.
+        /// </summary>
+        public static GraphTraversal<Vertex, Vertex> RatedVideos(this GraphTraversal<Vertex, Vertex> t, int minRate)
+        {
+            return t.OutE(EdgeRated).Has(PropertyRating, Gt(minRate)).InV();
+        }
+
+        /// <summary>
+        /// what other users rated those videos highly
+        /// </summary>
+        public static GraphTraversal<Vertex, Vertex> OtherUserRatedVideos(this GraphTraversal<Vertex, Vertex> t, int minRate, int sample)
+        {                    
+            return 
+                // what other users rated those videos highly? (this is like saying "what users share my taste")    
+                t.InE(EdgeRated).Has(PropertyRating, Gt(minRate))
+
+                // but don't grab too many, or this won't work OLTP, and "by('rating')" favors the higher ratings
+                 .Sample(sample).By(PropertyRating).OutV();
+        }
+
+        /// <summary>
+        /// what other users rated those videos highly
+        /// </summary>
+        public static GraphTraversal<Vertex, Vertex> VideosRatedByOthers(this GraphTraversal<Vertex, Vertex> t, 
+                                                                         int minRate, long minLocalRate) {
+            return
+                  // Now we're working with "similar users". For those users who share my taste, grab N highly rated 
+                  // videos. Save the rating so we can sum the scores later, and use sack() because it does not require 
+                  // path information. (as()/select() was slow)
+                  t.Local<List<Vertex>>(
+                        __.OutE(EdgeRated)
+                            .Has(PropertyRating, Gt(minRate))
+                            .Limit(minLocalRate))
+                  .Sack(Operator.Assign)
+                        .By(PropertyRating).InV();
+        }
+
+        /// <summary>
+        /// what other users rated those videos highly
+        /// </summary>
+        public static GraphTraversal<Vertex, Vertex> AllRatedVideos(this GraphTraversal<Vertex, Vertex> t)
+        {
+            return t.Map<Vertex>(__.Out(EdgeRated).Dedup().Fold());
+        }
+
+        /// <summary>
+        /// Expects an incoming <code>Vertex</code> and projects it to a <code>Map</code> with the specified <code>Enrichment</code>>
+        /// values passed to it.
+        /// </summary>
+        public static GraphTraversal<Vertex, IDictionary<string, object>> Enrich(this GraphTraversal<Vertex, Vertex> t, bool includeIdLabel, params Enrichment[] enrichments)
+        {
+            var projectTraversals = enrichments.Select(e => e.GetTraversals()).SelectMany(i => i).ToList();
+            if (includeIdLabel)
+            {
+                projectTraversals.Add(__.Id());
+                projectTraversals.Add(__.Map<object>(__.Label()));
+            }
+
+            var keys = enrichments.Select(e => e.GetProjectedKeys()).SelectMany(i => i).ToList();
+            if (includeIdLabel)
+            {
+                keys.Add("id");
+                keys.Add("label");
+            }
+
+            var projectedKeys = keys.GetRange(1, keys.Count() - 1).ToArray();
+            var te = t.Project<object>(keys.First(), projectedKeys);
+            foreach (GraphTraversal<object, object> projectTraversal in projectTraversals)
+            {
+                te = te.By(projectTraversal);
+            }
+
+            return te;
+        }
+
+        /// <summary>
+        /// This step is an alias for the <code>SideEffect()</code> step. As an alias, it makes certain aspects of the DSL more
+        /// readable.
+        /// </summary>
+        public static GraphTraversal<S, E> Ensure<S, S1, E>(
+            this GraphTraversal<S, E> t, GraphTraversal<S1, E> mutationTraversal) {
+            return t.SideEffect(mutationTraversal);
+        }
+    }
+}

KillrVideo.GraphDsl/Dsl/Recommender.cs

@@ -0,0 +1,34 @@
+using System.Collections.Generic;
+using Gremlin.Net.Process.Traversal;
+using Gremlin.Net.Structure;
+
+
+namespace KillrVideo.GraphDsl.Dsl
+{
+
+    /// <summary>
+    /// Provides for pre-built "sampling" settings to the <code>recommend(int, int, Recommender, Traversal)<code>
+    /// step. The sampling options help determine the nature of the initial set of movies to recommend, by limiting the
+    /// number of actors used from highly rated movies of the user who is target for the recommendation.
+    /// </summary>
+    public enum Recommender
+    {
+        SmallSample,
+        LargeSample,
+        Fifty50Sample,
+        TimedSample,
+        All
+    }
+
+    public static class RecommenderLookup
+    {
+        public static readonly Dictionary<Recommender, GraphTraversal<object, IList<Vertex>>> Traversals = new Dictionary<Recommender, GraphTraversal<object, IList<Vertex>>>
+        {
+            {Recommender.SmallSample, __.OutE(KvGraph.EdgeActor).Sample(3).InV().Fold()},
+            {Recommender.LargeSample, __.OutE(KvGraph.EdgeActor).Sample(10).InV().Fold()},
+            {Recommender.Fifty50Sample, __.OutE(KvGraph.EdgeActor).Coin(0.5).InV().Fold()},
+            {Recommender.TimedSample, __.OutE(KvGraph.EdgeActor).TimeLimit(250).InV().Fold()},
+            {Recommender.All, __.OutE(KvGraph.EdgeActor).InV().Fold()}
+        };
+    }
+}