feat!(CosmosStore): Azure.Cosmos 'V4' support / enable mocking Cosmos APIs

samples/Infrastructure/Infrastructure.fsproj

@@ -35,7 +35,6 @@
     <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="2.1.1" />
     <PackageReference Include="Serilog.Sinks.Console" Version="3.1.1" />
     <PackageReference Include="Serilog.Sinks.Seq" Version="4.0.0" />
-    <PackageReference Include="Microsoft.Azure.Cosmos.Direct" Version="3.1.4" />
   </ItemGroup>
 
 </Project>

samples/Infrastructure/Storage.fs

@@ -35,7 +35,7 @@ module Cosmos =
 
     type [<NoEquality; NoComparison>] Arguments =
         | [<AltCommandLine "-V">]       VerboseStore
-        | [<AltCommandLine "-m">]       ConnectionMode of Microsoft.Azure.Cosmos.ConnectionMode
+        | [<AltCommandLine "-m">]       ConnectionMode of Azure.Cosmos.ConnectionMode
         | [<AltCommandLine "-o">]       Timeout of float
         | [<AltCommandLine "-r">]       Retries of int
         | [<AltCommandLine "-rt">]      RetriesWaitTimeS of float
@@ -54,7 +54,7 @@ module Cosmos =
                 | Database _ ->         "specify a database name for store. (optional if environment variable EQUINOX_COSMOS_DATABASE specified)"
                 | Container _ ->        "specify a container name for store. (optional if environment variable EQUINOX_COSMOS_CONTAINER specified)"
     type Info(args : ParseResults<Arguments>) =
-        member __.Mode =                args.GetResult(ConnectionMode,Microsoft.Azure.Cosmos.ConnectionMode.Direct)
+        member __.Mode =                args.GetResult(ConnectionMode,Azure.Cosmos.ConnectionMode.Direct)
         member __.Connection =          args.TryGetResult Connection |> defaultWithEnvVar "EQUINOX_COSMOS_CONNECTION" "Connection"
         member __.Database =            args.TryGetResult Database   |> defaultWithEnvVar "EQUINOX_COSMOS_DATABASE"   "Database"
         member __.Container =           args.TryGetResult Container  |> defaultWithEnvVar "EQUINOX_COSMOS_CONTAINER"  "Container"

src/Equinox.Core/Infrastructure.fs

@@ -6,6 +6,7 @@ open FSharp.Control
 open System
 open System.Diagnostics
 open System.Threading.Tasks
+open System.Threading
 
 type OAttribute = System.Runtime.InteropServices.OptionalAttribute
 type DAttribute = System.Runtime.InteropServices.DefaultParameterValueAttribute

src/Equinox.Cosmos/Equinox.Cosmos.fsproj

@@ -11,6 +11,7 @@
 
   <ItemGroup>
     <Compile Include="..\Equinox.Core\Infrastructure.fs" Link="Infrastructure.fs" />
+    <Compile Include="NewtonsoftJsonSerializer.fs" />
     <Compile Include="Cosmos.fs" />
   </ItemGroup>
 
@@ -19,15 +20,15 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="1.1.0" />
     <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
     <PackageReference Include="MinVer" Version="2.0.0" PrivateAssets="All" />
 
     <PackageReference Include="FSharp.Core" Version="4.3.4" Condition=" '$(TargetFramework)' == 'netstandard2.0' " />
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="1.1.0" />
 
     <PackageReference Include="FsCodec.NewtonsoftJson" Version="2.0.0" />
     <PackageReference Include="FSharp.Control.AsyncSeq" Version="2.0.23" />
-    <PackageReference Include="Microsoft.Azure.Cosmos" Version="3.1.1" />
+    <PackageReference Include="Azure.Cosmos" Version="4.0.0-preview3" />
     <PackageReference Include="System.Runtime.Caching" Version="4.5.0" />
   </ItemGroup>
 

src/Equinox.Cosmos/NewtonsoftJsonSerializer.fs

@@ -0,0 +1,39 @@
+namespace Equinox.Cosmos.Store
+
+open System.IO;
+open System.Text;
+open Azure.Cosmos.Serialization;
+open Newtonsoft.Json;
+open Newtonsoft.Json.Serialization;
+
+type NewtonsoftJsonSerializer () =
+    inherit CosmosSerializer ()
+
+    let encoding = new UTF8Encoding(false, true)
+    let serializer = JsonSerializer.Create()
+
+    override __.FromStream<'T> (stream: Stream): 'T =
+        use stream = stream
+
+        if typeof<Stream>.IsAssignableFrom(typeof<'T>) then
+            stream :> obj :?> 'T
+        else
+            use streamReader = new StreamReader(stream)
+            use jsonReader = new JsonTextReader(streamReader)
+            serializer.Deserialize<'T>(jsonReader)
+
+    override __.ToStream<'T> (input: 'T): Stream =
+        let payload = new MemoryStream()
+
+        (
+            use streamWriter = new StreamWriter(payload, encoding = encoding, bufferSize = 1024, leaveOpen = true)
+            use jsonWriter = new JsonTextWriter(streamWriter)
+
+            jsonWriter.Formatting <- Formatting.None
+            serializer.Serialize(jsonWriter, input)
+            jsonWriter.Flush()
+            streamWriter.Flush()
+        )
+
+        payload.Position <- 0L
+        payload :> Stream

tests/Equinox.Cosmos.Integration/CosmosCoreIntegration.fs

@@ -238,7 +238,7 @@ type Tests(testOutputHelper) =
 
         // 2 items atm
         test <@ [EqxAct.ResponseForward; EqxAct.ResponseForward; EqxAct.QueryForward] = capture.ExternalCalls @>
-        verifyRequestChargesMax 6 } // 5.77
+        verifyRequestChargesMax 9 } // 8.51 // WAS 6 // 5.77
 
     [<AutoData(SkipIfRequestedViaEnvironmentVariable="EQUINOX_INTEGRATION_SKIP_COSMOS")>]
     let ``get Lazy`` (TestStream streamName) = Async.RunSynchronously <| async {
@@ -258,7 +258,7 @@ type Tests(testOutputHelper) =
             | _ -> None
         // validate that, despite only requesting max 1 item, we only needed one trip (which contained only one item)
         [1,1] =! capture.ChooseCalls queryRoundTripsAndItemCounts
-        verifyRequestChargesMax 4 // 3.02 // WAS 3 // 2.97
+        verifyRequestChargesMax 6 // 5.74 // WAS 4 // 3.02 // WAS 3 // 2.97
     }
 
     (* Backward *)
@@ -279,7 +279,7 @@ type Tests(testOutputHelper) =
         verifyCorrectEventsBackward 4L expected res
 
         test <@ [EqxAct.ResponseBackward; EqxAct.QueryBackward] = capture.ExternalCalls @>
-        verifyRequestChargesMax 4 // 3.04 // WAS 3
+        verifyRequestChargesMax 6 // 5.75 // WAS 4 // 3.04 // WAS 3
     }
 
     [<AutoData(SkipIfRequestedViaEnvironmentVariable="EQUINOX_INTEGRATION_SKIP_COSMOS")>]
@@ -323,5 +323,5 @@ type Tests(testOutputHelper) =
             | EqxEvent (Equinox.Cosmos.Store.Log.Event.Query (Equinox.Cosmos.Store.Direction.Backward, responses, { count = c })) -> Some (responses,c)
             | _ -> None
         [1,5] =! capture.ChooseCalls queryRoundTripsAndItemCounts
-        verifyRequestChargesMax 4 // 3.04 // WAS 3 // 2.98
-    }
+        verifyRequestChargesMax 6 // 5.76 // WAS 4 // 3.04 // WAS 3 // 2.98
+    }

tests/Equinox.Cosmos.Integration/Equinox.Cosmos.Integration.fsproj

@@ -27,7 +27,6 @@
     <PackageReference Include="FsCheck.xUnit" Version="2.14.0" />
     <PackageReference Include="JsonDiffPatch.Net" Version="2.1.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.3.0" />
-    <PackageReference Include="Microsoft.Azure.Cosmos.Direct" Version="3.1.4" />
     <PackageReference Include="Serilog.Sinks.Seq" Version="4.0.0" />
     <PackageReference Include="unquote" Version="4.0" />
     <PackageReference Include="xunit" Version="2.4.1" />

tools/Equinox.Tool/Program.fs

@@ -334,7 +334,7 @@ module SqlInit =
         | _ -> failwith "please specify a `ms`,`my` or `pg` endpoint" }
 
 module CosmosStats =
-    type Microsoft.Azure.Cosmos.Container with
+    type Azure.Cosmos.CosmosContainer with
         // NB DO NOT CONSIDER PROMULGATING THIS HACK
         member container.QueryValue<'T>(sqlQuery : string) =
             let query : seq<'T> = failwith "TODO translate" //container.ReadItemAsync(sqlQuery) :> _