open FsBunny
open FsShelter.Extras.Assemblers

/// DAL helpers
module RabbitMq =
    open System
    open RabbitMQ.Client
    let mkEventStreams (connectionString:string) =
        let uri = Uri(connectionString)
        let cf = ConnectionFactory(Uri = uri)
        RabbitMqEventStreams(cf, "amq.topic", 3us, 1000us)

let rabbit = "amqp://localhost"

/// Create a consumer based on the arguments.
let withQueueConsumer topic queue log cfg = 
    let eventStreams = RabbitMq.mkEventStreams rabbit :> EventStreams
    eventStreams.GetJsonConsumer (Persistent queue) (eventStreams.Routed topic)

/// Create a routed publisher based on the arguments.
let withPublisher publish log cfg = 
    let eventStreams = RabbitMq.mkEventStreams rabbit :> EventStreams
    fun tuple emit -> tuple, publish eventStreams

/// Publish Json
let publishJson exchange (eventStreams:EventStreams) (topic,msg) = 
    msg |-> eventStreams.UsingJsonPublisher (Routed (exchange,topic))

// DAL helpers
module Cassandra =
  open FsCassy
  open Cassandra.Mapping

  let mkInterpreter mkSession (name:string) = 
    let session = mkSession name
    { new Interpreter with 
        member x.Interpret<'t,'r> (statement:Statement<'t,'r>) : 'r= 
          Cassandra.Interpreter.execute (fun _ -> Cassandra.Api.mkTable (MappingConfiguration()) session) statement }

  let mkSession (connectionString:string) =
    Cassandra.Api.mkCluster connectionString
    |> Cassandra.Api.mkSession id

let cassandra = "contact points=localhost;default keyspace=sample"

let withPersisterArgs log cfg =
    let persister = Cassandra.mkInterpreter Cassandra.mkSession cassandra
    fun input emit ->
        (fun f -> f persister), input

let withReaderArgs log cfg =
    let persister = Cassandra.mkInterpreter Cassandra.mkSession cassandra
    fun input emit ->
        (fun f -> f persister), input, emit

open System
open FsCassy

/// Events and Notifications
[<AutoOpen>]
module Messaging =
  type SensorReading = 
    { Id : Guid; Temp : int}

  type SensorAlert = 
    { Id : Guid; Text : string }

/// Cassandra-mapped types
[<AutoOpen>]
module Persistence =
  type Sensor = 
    { Id : Guid; Value : int}

/// Topology streams
type Stream =
    | Update of sensorId:Guid * temp:int
    | Alarm of sensorId:Guid * temp:int
    | NotFound of sensorId:Guid

let sensorWriter (persister:Interpreter) = 
    function 
    | Update(id, temp) -> 
            table<Sensor> 
        >>= where (Quote.X(fun s -> s.Id = id))
        >>= select (Quote.X(fun x -> { x with Value = temp }))
        >>= update >>= execute
            |> persister.Interpret

let thresholdAnalyser (persister:Interpreter) = 
    // realistically we'd use the ^ persister to read the threshold value from the database, but let's keep it simple
    function 
    | Update(id, temp) when temp > 100 -> [Alarm(id,temp)]
    | _ -> []
    >> async.Return

open FsShelter.DSL
open FsShelter.Extras

let sampleTopology = topology "sample" {
    let sensorEvents = 
        createQueueSpout (fun (m:SensorReading) -> Update (m.Id, m.Temp))
        |> runReliableSpout (withQueueConsumer "Topics.Sensor.Started" "sensor_started") ackerOfConsumer

    let sensorWriter = 
        createPersistBolt sensorWriter
        |> runBolt withPersisterArgs 

    let thresholdAnalyser = 
        createTransformBolt thresholdAnalyser
        |> runBolt withReaderArgs 

    let alertNotifier = 
        createNotifierBolt (function Alarm (sensorId,temp) -> "Topics.Alarm", 
                                                              { SensorAlert.Id = sensorId
                                                                Text = sprintf "At %A the reactor went critical" DateTime.Now})
        |> runBolt (withPublisher <| publishJson "amq.topic")

    yield sensorEvents ==> sensorWriter |> shuffle.on Update
    yield sensorEvents ==> thresholdAnalyser |> shuffle.on Update

    yield thresholdAnalyser --> alertNotifier |> shuffle.on Alarm
}