<commit_before>﻿namespace FotM.Aether

open System
open System.IO
open System.Net
open System.Text
open System.Threading
open Microsoft.WindowsAzure
open Microsoft.WindowsAzure.Storage
open Microsoft.WindowsAzure.Storage.Blob
open Newtonsoft.Json
open FotM.Data
open FotM.Hephaestus.Async
open FotM.Hephaestus.TraceLogging
open FotM.Hephaestus.CollectionExtensions
open FotM.Utilities

module RepoSync =
    let lockobj = new System.Object()

type Storage (containerName, ?storageConnectionString, ?pathPrefix) =

    let connectionString = defaultArg storageConnectionString (CloudConfigurationManager.GetSetting "Microsoft.Storage.ConnectionString")
    let prefix = defaultArg pathPrefix ""
    let storageAccount = CloudStorageAccount.Parse connectionString

    let blobClient = storageAccount.CreateCloudBlobClient()

    let container = blobClient.GetContainerReference containerName

    do
        logInfo "initializing blob container at %A" container.Uri
        container.CreateIfNotExists(BlobContainerPublicAccessType.Blob) |> ignore

    member this.upload (data, ?path) =
        let relativePath = defaultArg path (sprintf "%A" (Guid.NewGuid()))
        let blob = container.GetBlockBlobReference (Path.Combine (prefix, relativePath))

        try
            lock RepoSync.lockobj (fun () ->
                let json = JsonConvert.SerializeObject data // serialization is not threadsafe here
                let compressed = CompressionUtils.ZipToBase64 json
                blob.UploadText compressed 
            )
        with
            | ex -> 
                logError "%A" ex
                reraise()

        logInfo "uploaded update to %A" blob.Uri

        blob.Uri

    member this.allBlobs(?directory) =
        let directory = defaultArg directory ""

        let allBlobs = 
            container.ListBlobs(prefix = directory, useFlatBlobListing = true)
            |> Seq.toArray

        allBlobs
        |> Array.choose(fun b -> 
            let blob = b :?> CloudBlockBlob
            match blob.Properties.LastModified |> toOption with
            | Some time -> Some(blob, time)
            | None -> None)
        |> Array.sortBy(fun (blob, t) -> t)
        |> Array.map(fun (blob, t) -> blob)

    member this.allFiles(?directory) =
        let directory = defaultArg directory ""

        this.allBlobs(directory) |> Array.map (fun blob -> blob.Uri)

module StorageIO =

    let retry = RetryBuilder(3)

    let downloadAsync (storageLocation: Uri) = async {
            logInfo "Fetching %A" storageLocation
            use webClient = new WebClient()
            webClient.Encoding <- Encoding.UTF8
            let! compressed = webClient.AsyncDownloadString storageLocation
            let json = CompressionUtils.UnzipFromBase64 compressed
            return json
    }

    let download (storageLocation: Uri) = retry {
        let result = downloadAsync storageLocation |> Async.RunSynchronously
        return result
    }<commit_msg>Increased timeout on web requests.<commit_after>﻿namespace FotM.Aether

open System
open System.IO
open System.Net
open System.Text
open System.Threading
open Microsoft.WindowsAzure
open Microsoft.WindowsAzure.Storage
open Microsoft.WindowsAzure.Storage.Blob
open Newtonsoft.Json
open FotM.Data
open FotM.Hephaestus.Async
open FotM.Hephaestus.TraceLogging
open FotM.Hephaestus.CollectionExtensions
open FotM.Utilities

module RepoSync =
    let lockobj = new System.Object()

type Storage (containerName, ?storageConnectionString, ?pathPrefix) =

    let connectionString = defaultArg storageConnectionString (CloudConfigurationManager.GetSetting "Microsoft.Storage.ConnectionString")
    let prefix = defaultArg pathPrefix ""
    let storageAccount = CloudStorageAccount.Parse connectionString

    let blobClient = storageAccount.CreateCloudBlobClient()

    let container = blobClient.GetContainerReference containerName

    do
        logInfo "initializing blob container at %A" container.Uri
        container.CreateIfNotExists(BlobContainerPublicAccessType.Blob) |> ignore

    member this.upload (data, ?path) =
        let relativePath = defaultArg path (sprintf "%A" (Guid.NewGuid()))
        let blob = container.GetBlockBlobReference (Path.Combine (prefix, relativePath))

        try
            lock RepoSync.lockobj (fun () ->
                let json = JsonConvert.SerializeObject data // serialization is not threadsafe here
                let compressed = CompressionUtils.ZipToBase64 json
                blob.UploadText compressed 
            )
        with
            | ex -> 
                logError "%A" ex
                reraise()

        logInfo "uploaded update to %A" blob.Uri

        blob.Uri

    member this.allBlobs(?directory) =
        let directory = defaultArg directory ""

        let allBlobs = 
            container.ListBlobs(prefix = directory, useFlatBlobListing = true)
            |> Seq.toArray

        allBlobs
        |> Array.choose(fun b -> 
            let blob = b :?> CloudBlockBlob
            match blob.Properties.LastModified |> toOption with
            | Some time -> Some(blob, time)
            | None -> None)
        |> Array.sortBy(fun (blob, t) -> t)
        |> Array.map(fun (blob, t) -> blob)

    member this.allFiles(?directory) =
        let directory = defaultArg directory ""

        this.allBlobs(directory) |> Array.map (fun blob -> blob.Uri)

type WebClientWithTimeout(timeout: TimeSpan) =
    inherit WebClient()

    override this.GetWebRequest(uri) =
        let wr = base.GetWebRequest(uri)
        wr.Timeout <- int timeout.TotalMilliseconds
        wr

module StorageIO =

    let retry = RetryBuilder(3)

    let downloadAsync (storageLocation: Uri) = async {
            logInfo "Fetching %A" storageLocation

            use webClient = new WebClientWithTimeout (TimeSpan.FromMinutes 2.0)
            let! compressed = webClient.AsyncDownloadString storageLocation

            let json = CompressionUtils.UnzipFromBase64 compressed
            return json
    }

    let download (storageLocation: Uri) = retry {
        let result = downloadAsync storageLocation |> Async.RunSynchronously
        return result
    }
{'fraction_non_alphanumeric': 0.07799128047795899, 'fraction_numerical': 0.0020991441950589376, 'mean_word_length': 2.6328445747800586, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '198588', 'n_tokens_mistral': 1681, 'n_tokens_neox': 1592, 'n_words': 442}