Telegram content proxy

Emulsion.ContentProxy/Emulsion.ContentProxy.fsproj

@@ -7,14 +7,19 @@
   <ItemGroup>
     <Compile Include="Proxy.fs" />
     <Compile Include="ContentStorage.fs" />
+    <Compile Include="FileCache.fs" />
   </ItemGroup>
 
   <ItemGroup>
     <ProjectReference Include="..\Emulsion.Database\Emulsion.Database.fsproj" />
+    <ProjectReference Include="..\Emulsion.Settings\Emulsion.Settings.fsproj" />
   </ItemGroup>
 
   <ItemGroup>
     <PackageReference Include="Hashids.net" Version="1.4.1" />
+    <PackageReference Include="Microsoft.Extensions.Http" Version="6.0.0" />
+    <PackageReference Include="Serilog" Version="2.10.0" />
+    <PackageReference Include="SimpleBase" Version="3.1.0" />
   </ItemGroup>
 
 </Project>

Emulsion.ContentProxy/FileCache.fs

@@ -0,0 +1,164 @@
+namespace Emulsion.ContentProxy
+
+open System
+open System.IO
+open System.Net.Http
+open System.Security.Cryptography
+open System.Text
+open System.Threading
+
+open Serilog
+open SimpleBase
+
+open Emulsion.Settings
+
+type DownloadRequest = {
+    Uri: Uri
+    CacheKey: string
+    Size: uint64
+}
+
+module Base58 =
+    /// Suggested by @ttldtor.
+    let M4N71KR = Base58(Base58Alphabet "123456789qwertyuiopasdfghjkzxcvbnmQWERTYUPASDFGHJKLZXCVBNM")
+
+module FileCache =
+    let EncodeFileName(sha256: SHA256, cacheKey: string): string =
+        cacheKey
+        |> Encoding.UTF8.GetBytes
+        |> sha256.ComputeHash
+        |> Base58.M4N71KR.Encode
+
+    let DecodeFileNameToSha256Hash(fileName: string): byte[] =
+        (Base58.M4N71KR.Decode fileName).ToArray()
+
+type FileCache(logger: ILogger,
+               settings: FileCacheSettings,
+               httpClientFactory: IHttpClientFactory,
+               sha256: SHA256) =
+
+    let getFilePath(cacheKey: string) =
+        Path.Combine(settings.Directory, FileCache.EncodeFileName(sha256, cacheKey))
+
+    let getFromCache(cacheKey: string) = async {
+        let path = getFilePath cacheKey
+        return
+            if File.Exists path then
+                Some(new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read|||FileShare.Delete))
+            else
+                None
+    }
+
+    let assertCacheValid() = async {
+        Directory.EnumerateFileSystemEntries settings.Directory
+        |> Seq.iter(fun entry ->
+            let entryName = Path.GetFileName entry
+
+            if not <| File.Exists entry
+            then failwith $"Cache directory invalid: contains a subdirectory: \"{entryName}\"."
+
+            let hash = FileCache.DecodeFileNameToSha256Hash entryName
+            if hash.Length <> sha256.HashSize / 8
+            then failwith (
+                $"Cache directory invalid: contains entry \"{entryName}\" which doesn't correspond to a " +
+                "base58-encoded SHA-256 hash."
+            )
+        )
+    }
+
+    let ensureFreeCache size = async {
+        if size > settings.FileSizeLimitBytes || size > settings.TotalCacheSizeLimitBytes then
+            return false
+        else
+            do! assertCacheValid()
+
+            let allEntries =
+                Directory.EnumerateFileSystemEntries settings.Directory
+                |> Seq.map FileInfo
+
+            // Now, sort the entries from newest to oldest, and start deleting if required at a point when we understand
+            // that there are too much files:
+            let entriesByPriority =
+                allEntries
+                |> Seq.sortByDescending(fun info -> info.LastWriteTimeUtc)
+                |> Seq.toArray
+
+            let mutable currentSize = 0UL
+            for info in entriesByPriority do
+                currentSize <- currentSize + Checked.uint64 info.Length
+                if currentSize + size > settings.TotalCacheSizeLimitBytes then
+                    logger.Information("Deleting a cache item \"{FileName}\" ({Size} bytes)", info.Name, info.Length)
+                    info.Delete()
+
+            return true
+    }
+
+    let download(uri: Uri): Async<Stream> = async {
+        let! ct = Async.CancellationToken
+
+        use client = httpClientFactory.CreateClient()
+        let! response = Async.AwaitTask <| client.GetAsync(uri, ct)
+        return! Async.AwaitTask <| response.EnsureSuccessStatusCode().Content.ReadAsStreamAsync()
+    }
+
+    let downloadIntoCacheAndGet uri cacheKey: Async<Stream> = async {
+        let! ct = Async.CancellationToken
+        let! stream = download uri
+        let path = getFilePath cacheKey
+        logger.Information("Saving {Uri} to path {Path}…", uri, path)
+
+        do! async { // to limit the cachedFile scope
+            use cachedFile = new FileStream(path, FileMode.CreateNew, FileAccess.Write, FileShare.None)
+            do! Async.AwaitTask(stream.CopyToAsync(cachedFile, ct))
+            logger.Information("Download successful: \"{Uri}\" to \"{Path}\".")
+        }
+
+        let! file = getFromCache cacheKey
+        return upcast Option.get file
+    }
+
+    let cancellation = new CancellationTokenSource()
+    let processRequest request: Async<Stream> = async {
+        logger.Information("Cache lookup for content {Uri} (cache key {CacheKey})", request.Uri, request.CacheKey)
+        match! getFromCache request.CacheKey with
+        | Some content ->
+            logger.Information("Cache hit for content {Uri} (cache key {CacheKey})", request.Uri, request.CacheKey)
+            return content
+        | None ->
+            logger.Information("No cache hit for content {Uri} (cache key {CacheKey}), will download", request.Uri, request.CacheKey)
+            let! shouldCache = ensureFreeCache request.Size
+            if shouldCache then
+                logger.Information("Resource {Uri} (cache key {CacheKey}, {Size} bytes) will fit into cache, caching", request.Uri, request.CacheKey, request.Size)
+                let! result = downloadIntoCacheAndGet request.Uri request.CacheKey
+                logger.Information("Resource {Uri} (cache key {CacheKey}, {Size} bytes) downloaded", request.Uri, request.CacheKey, request.Size)
+                return result
+            else
+                logger.Information("Resource {Uri} (cache key {CacheKey}) won't fit into cache, directly downloading", request.Uri, request.CacheKey)
+                let! result = download request.Uri
+                return result
+    }
+
+    let rec processLoop(processor: MailboxProcessor<_ * AsyncReplyChannel<_>>) = async {
+        while true do
+            let! request, replyChannel = processor.Receive()
+            try
+                let! result = processRequest request
+                replyChannel.Reply(Some result)
+            with
+            | ex ->
+                logger.Error(ex, "Exception while processing the file download queue")
+                replyChannel.Reply None
+    }
+    let processor = MailboxProcessor.Start(processLoop, cancellation.Token)
+
+    interface IDisposable with
+        member _.Dispose() =
+            cancellation.Dispose()
+            (processor :> IDisposable).Dispose()
+
+    member _.Download(uri: Uri, cacheKey: string, size: uint64): Async<Stream option> =
+        processor.PostAndAsyncReply(fun chan -> ({
+            Uri = uri
+            CacheKey = cacheKey
+            Size = size
+        }, chan))

Emulsion.Settings/Settings.fs

@@ -34,12 +34,19 @@ type HostingSettings = {
     HashIdSalt: string
 }
 
+type FileCacheSettings = {
+    Directory: string
+    FileSizeLimitBytes: uint64
+    TotalCacheSizeLimitBytes: uint64
+}
+
 type EmulsionSettings = {
     Xmpp: XmppSettings
     Telegram: TelegramSettings
     Log: LogSettings
     Database: DatabaseSettings option
     Hosting: HostingSettings option
+    FileCache: FileCacheSettings option
 }
 
 let defaultConnectionTimeout = TimeSpan.FromMinutes 5.0
@@ -56,6 +63,12 @@ let private readTimeSpan defaultVal key section =
     |> Option.defaultValue defaultVal
 
 let read (config : IConfiguration) : EmulsionSettings =
+    let uint64OrDefault value ``default`` =
+        value
+        |> Option.ofObj
+        |> Option.map uint64
+        |> Option.defaultValue ``default``
+
     let readXmpp (section : IConfigurationSection) = {
         Login = section["login"]
         Password = section["password"]
@@ -91,9 +104,17 @@ let read (config : IConfiguration) : EmulsionSettings =
             }
         | None, None, None -> None
         | other -> failwith $"Parameter pack {other} represents invalid hosting settings."
+    let readFileCache(section: IConfigurationSection) =
+        Option.ofObj section["directory"]
+        |> Option.map(fun directory -> {
+            Directory = directory
+            FileSizeLimitBytes = uint64OrDefault section["fileSizeLimitBytes"] 1024UL * 1024UL
+            TotalCacheSizeLimitBytes = uint64OrDefault section["fileSizeLimitBytes"] 20UL * 1024UL * 1024UL
+        })
 
     { Xmpp = readXmpp <| config.GetSection("xmpp")
       Telegram = readTelegram <| config.GetSection("telegram")
       Log = readLog <| config.GetSection "log"
       Database = readDatabase <| config.GetSection "database"
-      Hosting = readHosting <| config.GetSection "hosting" }
+      Hosting = readHosting <| config.GetSection "hosting"
+      FileCache = readFileCache <| config.GetSection "fileCache" }

Emulsion.Telegram/Client.fs

@@ -1,11 +1,20 @@
 namespace Emulsion.Telegram
 
+open System
 open System.Threading
 
 open Emulsion.Database
 open Emulsion.Messaging.MessageSystem
 open Emulsion.Settings
 
+type FileInfo = {
+    TemporaryLink: Uri
+    Size: uint64
+}
+
+type ITelegramClient =
+    abstract GetFileInfo: fileId: string -> Async<FileInfo option>
+
 type Client(ctx: ServiceContext,
             cancellationToken: CancellationToken,
             telegramSettings: TelegramSettings,
@@ -15,10 +24,27 @@ type Client(ctx: ServiceContext,
 
     let botConfig = { Funogram.Telegram.Bot.Config.defaultConfig with Token = telegramSettings.Token }
 
+    interface ITelegramClient with
+        member this.GetFileInfo(fileId) = async {
+            let logger = ctx.Logger
+            logger.Information("Querying file information for file {FileId}", fileId)
+            let! file = Funogram.sendGetFile botConfig fileId
+            match file.FilePath, file.FileSize with
+            | None, None ->
+                logger.Warning("File {FileId} was not found on server", fileId)
+                return None
+            | Some fp, Some sz ->
+                return Some {
+                    TemporaryLink = Uri $"https://api.telegram.org/file/bot{telegramSettings.Token}/{fp}"
+                    Size = Checked.uint64 sz
+                }
+            | x, y -> return failwith $"Unknown data received from Telegram server: {x}, {y}"
+        }
+
     override _.RunUntilError receiver = async {
         // Run loop of Telegram is in no need of any complicated start, so just return an async that will perform it:
         return Funogram.run ctx.Logger telegramSettings databaseSettings hostingSettings botConfig receiver
     }
 
     override _.Send message =
-        Funogram.send telegramSettings botConfig message
+        Funogram.sendMessage telegramSettings botConfig message

Emulsion.Telegram/Funogram.fs

@@ -298,9 +298,9 @@ module MessageConverter =
                 else extractMessageContent replyTo links.ReplyToContentLinks
             { main = mainMessage; replyTo = Some replyToMessage }
 
-let internal processSendResult(result: Result<'a, ApiResponseError>): unit =
+let internal processSendResult(result: Result<'a, ApiResponseError>): 'a =
     match result with
-    | Ok _ -> ()
+    | Ok x -> x
     | Error e ->
         failwith $"Telegram API Call processing error {e.ErrorCode}: {e.Description}"
 
@@ -347,15 +347,23 @@ let internal prepareHtmlMessage: Message -> string = function
 | Authored {author = author; text = text} -> $"<b>{Html.escape author}</b>\n{Html.escape text}"
 | Event {text = text} -> Html.escape text
 
-let send (settings: TelegramSettings) (botConfig: BotConfig) (OutgoingMessage content): Async<unit> =
+let private send (botConfig: BotConfig) request = api botConfig request
+
+let sendGetFile (botConfig: BotConfig) (fileId: string): Async<File> = async {
+    let! result = send botConfig (Req.GetFile.Make fileId)
+    return processSendResult result
+}
+
+let sendMessage (settings: TelegramSettings) (botConfig: BotConfig) (OutgoingMessage content): Async<unit> =
     let sendHtmlMessage (groupId: ChatId) text =
         Req.SendMessage.Make(groupId, text, ParseMode.HTML)
 
     let groupId = Int(int64 settings.GroupId)
     let message = prepareHtmlMessage content
     async {
-        let! result = api botConfig (sendHtmlMessage groupId message)
-        return processSendResult result
+        let! result = send botConfig (sendHtmlMessage groupId message)
+        processSendResult result |> ignore
+        return ()
     }
 
 let run (logger: ILogger)

Emulsion.TestFramework/Emulsion.TestFramework.fsproj

@@ -0,0 +1,25 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>net6.0</TargetFramework>
+    <GenerateDocumentationFile>true</GenerateDocumentationFile>
+    <OutputType>Library</OutputType>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <Compile Include="LockedBuffer.fs" />
+    <Compile Include="Logging.fs" />
+    <Compile Include="Waiter.fs" />
+    <Compile Include="TestDataStorage.fs" />
+    <Compile Include="Exceptions.fs" />
+    <Compile Include="TelegramClientMock.fs" />
+    <Compile Include="WebFileStorage.fs" />
+    <Compile Include="SimpleHttpClientFactory.fs" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Serilog.Sinks.XUnit" Version="1.0.8" />
+    <ProjectReference Include="..\Emulsion.Database\Emulsion.Database.fsproj" />
+    <ProjectReference Include="..\Emulsion.Telegram\Emulsion.Telegram.fsproj" />
+  </ItemGroup>
+</Project>

Emulsion.Tests/TestUtils/Exceptions.fs → Emulsion.TestFramework/Exceptions.fs

@@ -1,4 +1,4 @@
-module Emulsion.Tests.TestUtils.Exceptions
+module Emulsion.TestFramework.Exceptions
 
 open System
 open Microsoft.EntityFrameworkCore

Emulsion.Tests/TestUtils/LockedBuffer.fs → Emulsion.TestFramework/LockedBuffer.fs

@@ -1,4 +1,4 @@
-namespace Emulsion.Tests.TestUtils
+namespace Emulsion.TestFramework
 
 type LockedBuffer<'T>() =
     let messages = ResizeArray<'T>()

Emulsion.Tests/TestUtils/Logging.fs → Emulsion.TestFramework/Logging.fs

@@ -1,4 +1,4 @@
-module Emulsion.Tests.TestUtils.Logging
+module Emulsion.TestFramework.Logging
 
 open Serilog
 open Xunit.Abstractions

Emulsion.TestFramework/SimpleHttpClientFactory.fs

@@ -0,0 +1,7 @@
+namespace Emulsion.TestFramework
+
+open System.Net.Http
+
+type SimpleHttpClientFactory() =
+    interface IHttpClientFactory with
+        member this.CreateClient _ = new HttpClient()

Emulsion.TestFramework/TelegramClientMock.fs

@@ -0,0 +1,14 @@
+namespace Emulsion.TestFramework
+
+open System.Collections.Generic
+
+open Emulsion.Telegram
+
+type TelegramClientMock() =
+    let responses = Dictionary<string, FileInfo option>()
+
+    interface ITelegramClient with
+        member this.GetFileInfo fileId = async.Return responses[fileId]
+
+    member _.SetResponse(fileId: string, fileInfo: FileInfo option): unit =
+        responses[fileId] <- fileInfo

Emulsion.Tests/TestUtils/TestDataStorage.fs → Emulsion.TestFramework/TestDataStorage.fs

@@ -1,4 +1,4 @@
-module Emulsion.Tests.TestUtils.TestDataStorage
+module Emulsion.TestFramework.TestDataStorage
 
 open System.IO