Telegram content proxy

Emulsion.ContentProxy/ContentStorage.fs

@@ -9,6 +9,8 @@ type MessageContentIdentity = {
     ChatUserName: string
     MessageId: int64
     FileId: string
+    FileName: string
+    MimeType: string
 }
 
 let getOrCreateMessageRecord (context: EmulsionDbContext) (id: MessageContentIdentity): Async<TelegramContent> = async {
@@ -17,7 +19,9 @@ let getOrCreateMessageRecord (context: EmulsionDbContext) (id: MessageContentIde
             for content in context.TelegramContents do
             where (content.ChatUserName = id.ChatUserName
                    && content.MessageId = id.MessageId
-                   && content.FileId = id.FileId)
+                   && content.FileId = id.FileId
+                   && content.FileName = id.FileName
+                   && content.MimeType = id.MimeType)
         } |> tryExactlyOneAsync
     match existingItem with
     | None ->
@@ -26,6 +30,8 @@ let getOrCreateMessageRecord (context: EmulsionDbContext) (id: MessageContentIde
             ChatUserName = id.ChatUserName
             MessageId = id.MessageId
             FileId = id.FileId
+            FileName = id.FileName
+            MimeType = id.MimeType
         }
         do! addAsync context.TelegramContents newItem
         return newItem

Emulsion.ContentProxy/Emulsion.ContentProxy.fsproj

@@ -7,14 +7,21 @@
   <ItemGroup>
     <Compile Include="Proxy.fs" />
     <Compile Include="ContentStorage.fs" />
+    <Compile Include="FileCache.fs" />
+    <Compile Include="SimpleHttpClientFactory.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="JetBrains.Lifetimes" Version="2020.2.2" />
+    <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,218 @@
+namespace Emulsion.ContentProxy
+
+open System
+open System.IO
+open System.Net.Http
+open System.Security.Cryptography
+open System.Text
+open System.Threading
+
+open JetBrains.Collections.Viewable
+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 TryDecodeFileNameToSha256Hash(fileName: string): byte[] option =
+        try
+            Some <| (Base58.M4N71KR.Decode fileName).ToArray()
+        with
+        | :? ArgumentException -> None
+
+    let IsMoveAndDeleteModeEnabled =
+        // NOTE: On older versions of Windows (known to reproduce on windows-2019 GitHub Actions image), the following
+        // scenario may be defunct:
+        //
+        // - open a file with FileShare.Delete (i.e. for download)
+        // - delete a file (i.e. during the cache cleanup)
+        // - try to create a file with the same name again
+        //
+        // According to this article
+        // (https://boostgsoc13.github.io/boost.afio/doc/html/afio/FAQ/deleting_open_files.html), it is impossible to do
+        // since file will occupy its disk name until the last handle is closed.
+        //
+        // In practice, this is allowed (checked at least on Windows 10 20H2 and windows-2022 GitHub Actions image), but
+        // some tests are known to be broken on older versions of Windows (windows-2019).
+        //
+        // As a workaround, let's rename the file to a random name before deleting it.
+        //
+        // This workaround may be removed after these older versions of Windows goes out of support.
+        OperatingSystem.IsWindows()
+
+type FileCache(logger: ILogger,
+               settings: FileCacheSettings,
+               httpClientFactory: IHttpClientFactory,
+               sha256: SHA256) =
+
+    let error = Signal<Exception>()
+
+    let getFilePath(cacheKey: string) =
+        Path.Combine(settings.Directory, FileCache.EncodeFileName(sha256, cacheKey))
+
+    let readFileOptions =
+        FileStreamOptions(Mode = FileMode.Open, Access = FileAccess.Read, Options = FileOptions.Asynchronous, Share = (FileShare.Read ||| FileShare.Delete))
+
+    let writeFileOptions =
+        FileStreamOptions(Mode = FileMode.CreateNew, Access = FileAccess.Write, Options = FileOptions.Asynchronous, Share = FileShare.None)
+
+    let getFromCache(cacheKey: string) = async {
+        let path = getFilePath cacheKey
+        return
+            if File.Exists path then
+                Some(new FileStream(path, readFileOptions))
+            else
+                None
+    }
+
+    let enumerateCacheFiles() =
+        let entries = Directory.EnumerateFileSystemEntries settings.Directory
+        if FileCache.IsMoveAndDeleteModeEnabled then
+            entries |> Seq.filter(fun p -> not(p.EndsWith ".deleted"))
+        else
+            entries
+
+    let deleteFileSafe (fileInfo: FileInfo) = async {
+        if FileCache.IsMoveAndDeleteModeEnabled then
+            fileInfo.MoveTo(Path.Combine(fileInfo.DirectoryName, $"{Guid.NewGuid().ToString()}.deleted"))
+            fileInfo.Delete()
+        else
+            fileInfo.Delete()
+    }
+
+    let assertCacheDirectoryExists() = async {
+        Directory.CreateDirectory settings.Directory |> ignore
+    }
+
+    let assertCacheValid() = async {
+        enumerateCacheFiles()
+        |> Seq.iter(fun entry ->
+            let entryName = Path.GetFileName entry
+
+            if not <| File.Exists entry
+            then failwith $"Cache directory invalid: contains a subdirectory \"{entryName}\"."
+
+            match FileCache.TryDecodeFileNameToSha256Hash entryName with
+            | Some hash when hash.Length = sha256.HashSize / 8 -> ()
+            | _ ->
+                failwith (
+                    $"Cache directory invalid: contains an 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! assertCacheDirectoryExists()
+            do! assertCacheValid()
+
+            let allEntries = enumerateCacheFiles() |> 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)
+                    do! deleteFileSafe info
+
+            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, writeFileOptions)
+            do! Async.AwaitTask(stream.CopyToAsync(cachedFile, ct))
+            logger.Information("Download successful: \"{Uri}\" to \"{Path}\".", uri, 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 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")
+                error.Fire ex
+                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))
+
+    member _.Error: ISource<Exception> = error

Emulsion.ContentProxy/SimpleHttpClientFactory.fs

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

Emulsion.Database/Entities.fs

@@ -8,4 +8,6 @@ type TelegramContent = {
     ChatUserName: string
     MessageId: int64
     FileId: string
+    FileName: string
+    MimeType: string
 }

Emulsion.Database/Migrations/20220828133844_ContentFileNameAndMimeType.fs

@@ -0,0 +1,91 @@
+// <auto-generated />
+namespace Emulsion.Database.Migrations
+
+open System
+open Emulsion.Database
+open Microsoft.EntityFrameworkCore
+open Microsoft.EntityFrameworkCore.Infrastructure
+open Microsoft.EntityFrameworkCore.Migrations
+
+[<DbContext(typeof<EmulsionDbContext>)>]
+[<Migration("20220828133844_ContentFileNameAndMimeType")>]
+type ContentFileNameAndMimeType() =
+    inherit Migration()
+
+    override this.Up(migrationBuilder:MigrationBuilder) =
+        migrationBuilder.AddColumn<string>(
+            name = "FileName"
+            ,table = "TelegramContents"
+            ,``type`` = "TEXT"
+            ,nullable = true
+            ,defaultValue = "file.bin"
+            ) |> ignore
+
+        migrationBuilder.AddColumn<string>(
+            name = "MimeType"
+            ,table = "TelegramContents"
+            ,``type`` = "TEXT"
+            ,nullable = true
+            ,defaultValue = "application/octet-stream"
+            ) |> ignore
+
+        migrationBuilder.Sql @"
+            drop index TelegramContents_Unique;
+
+            create unique index TelegramContents_Unique
+            on TelegramContents(ChatUserName, MessageId, FileId, FileName, MimeType)
+        " |> ignore
+
+
+    override this.Down(migrationBuilder:MigrationBuilder) =
+        migrationBuilder.DropColumn(
+            name = "FileName"
+            ,table = "TelegramContents"
+            ) |> ignore
+
+        migrationBuilder.DropColumn(
+            name = "MimeType"
+            ,table = "TelegramContents"
+            ) |> ignore
+
+        migrationBuilder.Sql @"
+            drop index TelegramContents_Unique;
+
+            create unique index TelegramContents_Unique
+            on TelegramContents(ChatUserName, MessageId, FileId)
+        " |> ignore
+
+
+    override this.BuildTargetModel(modelBuilder: ModelBuilder) =
+        modelBuilder
+            .HasAnnotation("ProductVersion", "5.0.10")
+            |> ignore
+
+        modelBuilder.Entity("Emulsion.Database.Entities.TelegramContent", (fun b ->
+
+            b.Property<Int64>("Id")
+                .IsRequired(true)
+                .ValueGeneratedOnAdd()
+                .HasColumnType("INTEGER") |> ignore
+            b.Property<string>("ChatUserName")
+                .IsRequired(false)
+                .HasColumnType("TEXT") |> ignore
+            b.Property<string>("FileId")
+                .IsRequired(false)
+                .HasColumnType("TEXT") |> ignore
+            b.Property<string>("FileName")
+                .IsRequired(false)
+                .HasColumnType("TEXT") |> ignore
+            b.Property<Int64>("MessageId")
+                .IsRequired(true)
+                .HasColumnType("INTEGER") |> ignore
+            b.Property<string>("MimeType")
+                .IsRequired(false)
+                .HasColumnType("TEXT") |> ignore
+
+            b.HasKey("Id") |> ignore
+
+            b.ToTable("TelegramContents") |> ignore
+
+        )) |> ignore
+