BotBuider v4 SQLite Storage

SqliteStorage.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Bot.Builder;
using Microsoft.Data.Sqlite;
using Newtonsoft.Json;

namespace MyBot.Storage
{
    /// <summary>
    ///     A storage layer that uses an sqlite.
    /// </summary>
    public class SqliteStorage : IStorage
    {
        private static readonly JsonSerializer JsonSerializer = JsonSerializer.Create(new JsonSerializerSettings
        {
            // we use All so that we get typed roundtrip out of storage, but we don't use validation because we don't know what types are valid
            TypeNameHandling = TypeNameHandling.All
        });

        private readonly Func<SqliteConnection> _connection;

        /// <summary>
        ///     Initializes a new instance of the <see cref="SqliteStorage" /> class.
        /// </summary>
        /// <param name="dataPath">A database path.</param>
        public SqliteStorage(string dataPath)
        {
            this._connection = () => new SqliteConnection(
                string.Empty + new SqliteConnectionStringBuilder
                {
                    DataSource = string.IsNullOrEmpty(dataPath) ? "storage.db" : dataPath
                }
            );

            if (File.Exists(dataPath))
            {
                return;
            }

            using (var stream = File.Create(dataPath))
            {
            }

            using (var conn = this._connection())
            {
                conn.Open();
                using (var cmd = new SqliteCommand())
                {
                    cmd.Connection = conn;
                    cmd.CommandText =
                        "CREATE TABLE storage (key TEXT PRIMARY KEY, value TEXT NOT NULL, etag TEXT NULL)";
                    cmd.ExecuteNonQuery();
                }
            }
        }

        /// <inheritdoc />
        public async Task DeleteAsync(string[] keys, CancellationToken cancellationToken)
        {
            if (keys == null)
            {
                throw new ArgumentNullException(nameof(keys));
            }

            await Task.WhenAll(keys.Select(RemoveInternal)).ConfigureAwait(false);

            async Task RemoveInternal(string key)
            {
                using (var conn = this._connection())
                {
                    conn.Open();
                    using (var cmd = new SqliteCommand())
                    {
                        cmd.Connection = conn;
                        cmd.CommandText = "DELETE FROM storage WHERE key=@key";
                        cmd.Parameters.AddWithValue("key", key);
                        await cmd.ExecuteNonQueryAsync();
                    }
                }
            }
        }

        /// <inheritdoc />
        public async Task<IDictionary<string, object>> ReadAsync(string[] keys, CancellationToken cancellationToken)
        {
            if (keys == null)
            {
                throw new ArgumentNullException(nameof(keys));
            }

            var readTasks = keys.Select(key => ReadIndividualKey(key));

            await Task.WhenAll(readTasks).ConfigureAwait(false);

            var items = readTasks.Select(readTask => readTask.Result)
                .Where(kvp => kvp.Key != null)
                .ToDictionary(kvp => kvp.Key, kvp => kvp.Value);

            return items;

            async Task<KeyValuePair<string, object>> ReadIndividualKey(string key)
            {
                using (var conn = this._connection())
                {
                    conn.Open();
                    using (var cmd = new SqliteCommand())
                    {
                        cmd.Connection = conn;
                        cmd.CommandText = "SELECT key, value, etag FROM storage WHERE key=@key";
                        cmd.Parameters.AddWithValue("key", key);
                        using (var reader = cmd.ExecuteReader())
                        {
                            if (reader.Read())
                            {
                                using (var textReader = reader.GetTextReader(1))
                                {
                                    using (var jsonReader = new JsonTextReader(textReader))
                                    {
                                        var obj = JsonSerializer.Deserialize(jsonReader);

                                        if (obj is IStoreItem storeItem)
                                        {
                                            storeItem.ETag = reader.GetString(2);
                                        }

                                        return new KeyValuePair<string, object>(key, obj);
                                    }
                                }
                            }
                            return new KeyValuePair<string, object>();
                        }
                    }
                }
            }
        }

        /// <inheritdoc />
        public async Task WriteAsync(IDictionary<string, object> changes, CancellationToken cancellationToken)
        {
            if (changes == null)
            {
                throw new ArgumentNullException(nameof(changes));
            }

            await Task.WhenAll(changes.Select(PersistInternal)).ConfigureAwait(false);


            async Task PersistInternal(KeyValuePair<string, object> keyValuePair)
            {
                var newValue = keyValuePair.Value;
                var storeItem = newValue as IStoreItem;
                var key = keyValuePair.Key;
                var etag = storeItem?.ETag == "*" ? null : storeItem?.ETag;

                using (var conn = this._connection())
                {
                    conn.Open();
                    using (var cmd = new SqliteCommand())
                    {
                        cmd.Connection = conn;
                        if (etag == null)
                        {
                            cmd.CommandText =
                                "INSERT OR REPLACE INTO storage(key, value) VALUES (@key, @value)";
                        }
                        else
                        {
                            cmd.CommandText =
                                "INSERT OR REPLACE INTO storage(key, value, etag) VALUES (@key, @value, @etag)";
                            cmd.Parameters.AddWithValue("etag", etag);
                        }

                        cmd.Parameters.AddWithValue("key", key);
                        cmd.Parameters.Add("value", SqliteType.Text);
                        using (var memoryStream = new MemoryStream())
                        {
                            using (var streamWriter = new StreamWriter(memoryStream, Encoding.Unicode))
                            {
                                using (var streamReader = new StreamReader(memoryStream, Encoding.Unicode))
                                {
                                    JsonSerializer.Serialize(streamWriter, newValue);
                                    streamWriter.Flush();
                                    memoryStream.Seek(0, SeekOrigin.Begin);
                                    var data = await streamReader.ReadToEndAsync();
                                    cmd.Parameters["value"].Value = data;
                                    await cmd.ExecuteNonQueryAsync();
                                }
                            }
                        }
                    }
                }
            }
        }
    }
}