dzoukr / cosmostore Goto Github PK
View Code? Open in Web Editor NEWF# Event store for Azure Cosmos DB, Table Storage, Postgres, LiteDB & ServiceStack
License: MIT License
F# Event store for Azure Cosmos DB, Table Storage, Postgres, LiteDB & ServiceStack
License: MIT License
Yesterday on Build conference 2018 Microsoft announced they dropped the Fixed (up to 10 GB) collections for CosmosDB and lowering the initial throughput from 1000 RU/s to 400 RU/s. From configuration point of view this could be the breaking change so I rather prefer to have it marked as obsolete for a while (few releases) and remove it later, however start to use proper (unlimited) collection setup with partition id.
We should have Couch variance for Cosmo Store.
Hi!
I have to do a lot of back and forth to convert my events to the JToken type. Since Newtonsoft's JsonConvert's default converter serializes fsharp types in a very bloated and unnatural way, I have to use FSharpLu.Json's serializer (see the difference here: https://github.com/Microsoft/fsharplu/wiki/fsharplu.json#option-type).
However, FSharpLu.Json convert object directly to string โ which means I have to take the result of that and parse it with JToken.Parse
to get the desired JToken type. That seems like a lot of unnecessary conversion back and forth. Probably not a huge deal for a modern CPU, but still creates a little unnecessary load โ especially considering that our app will grow and eventual have a lot of events that needs processing.
So... Is there a way to pass it a string directly without parsing it first with JToken.Parse? Also, when I read the event, I have to do a .ToString() to convert it back to raw json, so that FSharpLu.Json can convert it to an object again.
Also, I'm really curious: How do you guys create Json? Do you use JsonConvert from Newtonsoft? How do you deal with the awkward json it makes for SumTypes & Options types etc.? Is there a better library than FSharpLu.Json? Chiron has zero documentation it seems, and all other json projects look pretty dead :( It seems that Json + F# is almost a little unsolved...?
Is it possible to skip the collection/stored procedure check/creation? Sounds like at least 2 roundtrips which are not needed for subsequent calls, or am I missing something?
Even though we are providing more generic types to make things more flexible. In code we are still having JToken
and int64
.
Obviously we are doing logic part on it. We might need to update that, to truly support everything.
As stores are growing we should be having CI/CD and publishing to nuget automated. Just to lower down a burden on maintainer.
Also, we have docs and site deployment as well. Mostly github docs. (I guess we might need more than something to readme. as mentioned in #34 :) )
The WindowsAzure.Storage libraries are pretty old - would you be open to updating this to Azure.Data.Tables? I've got a work in progress, but wanted to make sure @Dzoukr is good with it.
@Dzoukr Loved your work. Is it possible abstract away the implementation to make is more generalize.
I was trying Marten Event Store but it was tied with C# thingy for project part of it. But Marten as store is wonderful.
And then there is lite db for simple testing or simple application. I am noob at ES so can't say how complicated it would be. But it would be good to have it.
Was planning on building an eventstore for CosmosDb like this one.
One requirement would be to support live subscriptions, which could be implemented using the change feed of CosmosDb.
That would break the idea of having 1 interface for both storage implementations (or you would need to implement some kind of polling for Table Store or use the event grid notifications).
I'm totally fine to implement this myself, but I'm in doubt whether to create a PR as improvement for your project or fork/start over with a Cosmos-only implementation.
What would you prefer?
Hi,
In 2.x I used the following when reading from the cosmos change feed;
CosmoStore.CosmosDb.Conversion.documentToEventRead(d)
In 3.x this seems to be private within the EventStore. I'm I missing something, or is it possible to expose this in any way?
Hey, a while ago I was playing around with CosmoStore and I used the in-memory approach.
#load ".paket/load/main.group.fsx"
open System
open System.Collections.Concurrent
open CosmoStore
open CosmoStore.InMemory
open Thoth.Json.Net
open Microsoft.FSharp.Reflection
let config : Configuration =
{ InMemoryStreams = ConcurrentDictionary<string,Stream>()
InMemoryEvents = ConcurrentDictionary<Guid, EventRead>() }
let store = EventStore.getEventStore config
type Event =
| Increase of int
| Decrease of int
let encodeEvent = Encode.Auto.generateEncoder<Event>()
let decodeEvent = Decode.Auto.generateDecoder<Event>()
let getUnionCaseName (x:'a) =
match FSharpValue.GetUnionFields(x, typeof<'a>) with
| case, _ -> case.Name
let createEvent event =
{ Id = (Guid.NewGuid())
CorrelationId = None
CausationId = None
Name = getUnionCaseName event
Data = encodeEvent event
Metadata = None }
let appendEvent event =
store.AppendEvent "CounterStream" ExpectedPosition.Any event
|> Async.AwaitTask
|> Async.RunSynchronously
let increase amount =
Increase(amount)
|> createEvent
|> appendEvent
let decrease amount =
Decrease(amount)
|> createEvent
|> appendEvent
increase 5
decrease 3
increase 198
decrease 203
increase 4
store.GetEvents "CounterStream" EventsReadRange.AllEvents
|> Async.AwaitTask
|> Async.RunSynchronously
|> List.fold (fun acc e ->
match Decode.fromValue "$" decodeEvent e.Data with
| Ok(Increase i) -> acc + i
| Ok(Decrease d) -> acc - d
| Error e ->
failwithf "invalid event from store, %s" e
) 0
|> printfn "Projected state is %d"
When upgrading to v3 I noticed that my configuration isn't correct anymore.
Could a sample perhaps be added to the README for InMemory configuration?
Expected to get success instead.
Didn't want to get too far off-topic in the other issue, so creating a new one to discuss this.
One of the reasons it took me longer to start working on the changefeed, except for me being typical optimistic developer, is the fact a ran into this library: https://github.com/jet/equinox
Seen Jet's presentation long ago and almost no longer expected them to release this as open-source, but they did. Giving that they use this in production at their scale I think it's valuable to do a little comparison between their approach and CosmoStore.
Don't say this should be a reason to stop developing CosmoStore, but I think it's good to agree on the reason why we develop a new CosmosDb-based event-store and be explicit about that.
Hi there!
I am looking for an event store which will be friendly to F#, so the one written in F# should be perfect ;D I tried to use it with postgres + marten, but I am failing to read events if they are represented by discriminated union. Let me just give You the test that I run;
module Tests
open System
open System.Threading.Tasks
open Xunit
open CosmoStore.Marten
open CosmoStore
open FsUnit.Xunit
type DummyEvent = { IncrementedBy: int }
type DummyEvent2 = { Date: DateTime }
type Dummies =
| Dummy1 of DummyEvent
| Dummy2 of DummyEvent2
[<Fact>]
let ``Saving and reading`` () =
// Arrange
let streamId = "MyAmazingStream"
let config: Configuration =
{ Host = "localhost"
Username = "postgres"
Password = "Secret!Passw0rd"
Database = "cativity" }
let es = EventStore.getEventStore config
let eventWrite: EventWrite<Dummies> =
{ Id = Guid.NewGuid()
CorrelationId = None
CausationId = None
Name = nameof DummyEvent
Data = Dummies.Dummy1{ IncrementedBy = 4 }
Metadata = None }
// Act + Assert
[ eventWrite ]
|> es.AppendEvents streamId ExpectedVersion.Any
|> Async.AwaitTask
|> Async.RunSynchronously
|> ignore
let events =
EventsReadRange.AllEvents
|> es.GetEvents "MyAmazingStream"
|> Async.AwaitTask
|> Async.RunSynchronously
[ events.Head.Data ] |> should equal [ eventWrite.Data ]
The test breaks on event read.
System.AggregateException
One or more errors occurred. (unordered field Item)
So it is quite mysterious. Probably I am doing something wrong.
Without DU it works just OK - so if we will use EventWrite<DummyEvent>
Any hints? Thanks guys in advance!
Hi,
I saw the comment that the author moved from Cosmos DB to Azure Table Storage due to high costs with the former. How do you push the data to the Read Model though, I couldnt find Change Feed or similar for Table Storage ... and polling doesnt sound workable ...
Best regards,
Deyan
@Dzoukr I was thinking that we should have read me for all store. How to set up things and all. Not detailed as current one but simpler one? And we can have github pages as well.
Is there a mechanism to observe appended events from different server instances using the libray?
Hi,
Firstly, good stuff so far. This project is looking pretty interesting.
So, I am seriously considering using this, however, I have a couple of scalability concerns which I feel need addressing first. Let me clarify:
Azure table only indexes partition key and row key, which forms a sort of composite primary key. As I understand it there are no secondary indexes at all, so if I were to query a table field which is not partition or row key, I would effectively be running a 'partition scan' (a mini table scan).
Because of this, it is imperative to store (at least a copy) of the data in a way that is read-optimized if it is going to be used as a primary store. Having a dig through the source code it seems that there are basically two common types of query groups:
Get all events in stream
Get all events in stream where offset > myLocalOffset (there are others but this is most prevalent IMO)
Get a specific stream summary by streamId
Get all stream summaries (so I can determine if there are new events to accumulate over)
Get a specific stream summary (as above, probably more important in practice)
A couple of things jumped out at me -
The latter one worries me the most, and I think it can easily be fixed by simply rearranging the ID field with the Position field. As far as I can tell the event Ids are arbitrary and are rarely (if ever) queried against. Perhaps we can exchange the RowKey to actually be the position indexer and the Id as a secondary property? The benefit of this is range queries will be executed against a clustered index, and thus will be massively more efficient.
The summaries issue could be reduced by putting the summaries in a separate partition or similar where the RowKey is the streamId, although this has its own scalability issues if you have a lot of streams. Alternatively, you probably at least want an API that gets just a single stream by partition key and row key (stream id). This would be far more efficient than a cross partition query, and seems like the most common operation to me. I often know what I am projecting from, so to provide that explicitly in a query makes sense (although there is an argument to be made that I could just directly get the events after X, which will be 0 events most of the time).
I would be happy to contribute here or help out btw, just I am aware this is a massive breaking change for existing users so I didnt want to dive right in. Let me know if/how I can help though!
There is a very comprehensive article about all this indexing stuff, along with patterns of how to solve here:
https://docs.microsoft.com/en-gb/azure/cosmos-db/table-storage-design-guide#index-entities-pattern
Triggers
System.AggregateException : One or more errors occurred. (The given key 'Metadata' was not present in the dictionary)
We should have MongoDB implementation for CosmoStore
Would it be possible to return the last stream position (nextPosition) in the ESERROR_POSITION_POSITIONNOTMATCH exception, so that the client code can parse it out and return 409 with ETag = last stream position (in the context of REST API)?
Could you please add a license?
We should have Dapper implementation for CosmoStore.
This issue is created to track discussion about upcoming version 3 of CosmoStore and what should be its main goal. For me it is mainly about making EventStore "JSON payload representation agnostic" (in other words, do not force to use JToken
) since Microsoft itself annonced new JSON API and moving away from Newtonsoft.Json coupling it its libraries. It will also allow to add other stores like LiteDB without unnecessary transformation to JToken and back.
Technically it should be quite easy to make EventWrite
, EventRead
and EventStore
generic like:
type EventWrite<'a> = {
Id : Guid
CorrelationId : Guid option
CausationId : Guid option
Name : string
Data : 'a
Metadata : 'a option
}
Also upgrading existing libraries should be straightforward, so version 3 could be done pretty quickly.
Any other ideas? cc: @kunjee17 @ctaggart @TheAngryByrd @viktorvan
In case of empty list of events passed to Append
, the insert should be skippped
I don't know if that is good idea or now. But CosmoStore (API library) should have agreegrate type as well. (totally inspired by your blog)
It will help for user to have little less code to have. Even I m in favor of any other generic helpers we can provide which can help end users (Not the Store Creators like us) to write ES based systems easily.
Haven't thought of thoroughly but these includes but not limited to Agreegate, ReadDB, Projection Support etc etc.
I don't want to overwhelm user or bound with some way of doing things but if something usual is there we can surely help them.
I tried CosmoStore.LiteDb today and had a weird experience.
I installed it with paket. When running it I got a runtime error about the missing Dll CosmoStore in Version 3.0.0.0
I then opened the CosmoStore.dll which comes with the nuget package in dnspy where it shows it has version 1.0.0.0. I changed this to 3.0.0.0 so that I got rid of the runtime error.
For me it looks like the release 3.0 is faulty but I am absolutely not sure about this, because it is already 4 months old. Am I really the first one who encountered this issue?
It would be nice if we could also create a TableStorage.Configuration
from a storage connection string.
It seems that the Azure Table Store implementation somehow requires a JToken payload. Although if I understand correctly it should allow any type of payload since version 3.
Example script:
#r "nuget: CosmoStore, Version=3.0.1"
#r "nuget: CosmoStore.TableStorage, Version=3.0.1"
open CosmoStore
open System
let storageAccountName = "account"
let storageAuthKey = "key"
let myConfig = TableStorage.Configuration.CreateDefault storageAccountName storageAuthKey
let eventStore = myConfig |> TableStorage.EventStore.getEventStore
let serializedData = """
{
"value": "data",
"somenum": 8
}
"""
let eventToWrite = {
Id = Guid.NewGuid()
CorrelationId = None
CausationId = None
Name = "My custom event"
Data = serializedData
Metadata = Some "Super meta"
}
let streamId = "awesome-stream"
eventToWrite |> eventStore.AppendEvent streamId ExpectedVersion.NoStream |> Async.AwaitTask |> Async.RunSynchronously
When you run the script you get the following error:
C:\repos\cosmo\cosmotest.fsx(29,17): error FS0001: Type mismatch. Expecting a
'EventWrite<string> -> 'a'
but given a
'EventWrite<Newtonsoft.Json.Linq.JToken> -> Threading.Tasks.Task<EventRead<Newtonsoft.Json.Linq.JToken,int64>>'
The type 'string' does not match the type 'Newtonsoft.Json.Linq.JToken'
Been reading the doc and I see CorrelationId but no CausationId, is this a deliberate decision? If so, why? Note that it's very likely I would know this if I were more experienced in ES.
Hi @Dzoukr
It would be great if you can provide some details around the contribution. I ll start with few questions.
createdUtc
is based on event or stream's createUtc.What is exactly the purpose of the document with type=Stream
, including lastPosition
, lastUpdatedUtc
attribute? Shouldn't it be in another collection?
Btw, what is the partition key of the collection?
I really enjoy using CosmoStore. Thanks for creating it!
Apologies if anyone has already shared this or I missed documentation.
I was running into some of the size limitations using AzureTable storage and decided to give the new AzureTable and Serverless/Consumption plan support for CosmosDB.
https://learn.microsoft.com/en-us/azure/cosmos-db/table/introduction
It was easy enough to update the existing TableStorage configuration to make that happen. After creating your CosmosDB AzureTable instance (see the link above for that)...
let sovereignCloud =
CosmoStore.TableStorage.SovereignCloud (acctName, acctKey, "cosmos.azure.com")
let config =
{
TableName = tableName
Account = sovereignCloud
}
config
|> CosmoStore.TableStorage.EventStore.getEventStore
SovereignCloud to the rescue! Someone was forward-thinking there!
Thanks!
Hi,
I am looking into creating a Cosmos DB database with shared throughput across all collections/containers. It seems the database name can be specified when setting up the configuration, but the collection/container name not? Asking because I wanted to create several XxxEvents collections, and have different Cosmos Db Triggers assigned to each one ..
I was trying pull CosmoStore.ServiceStack. It should depend on CosmoStore 3.0 but it is depending on CosmoStore 3.0.1. And that is making paket little crazy.
Can you please cross check. I guess some thing messed up at the time of publishing.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.