WIP on PostgreSQL data impl
- Rename files - Add cache provider
This commit is contained in:
parent
73c7a686a4
commit
bed08b81ee
@ -6,9 +6,11 @@
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Data.Sqlite" Version="6.0.7" />
|
||||
<PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="6.0.0" />
|
||||
<PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="6.0.0" />
|
||||
<PackageReference Include="Microsoft.FSharpLu.Json" Version="0.11.7" />
|
||||
<PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
|
||||
<PackageReference Include="NodaTime" Version="3.1.2" />
|
||||
<PackageReference Include="Npgsql" Version="6.0.6" />
|
||||
<PackageReference Include="Npgsql.FSharp" Version="5.3.0" />
|
||||
<PackageReference Include="RethinkDb.Driver" Version="2.3.150" />
|
||||
@ -31,16 +33,17 @@
|
||||
<Compile Include="SQLite\SQLiteWebLogData.fs" />
|
||||
<Compile Include="SQLite\SQLiteWebLogUserData.fs" />
|
||||
<Compile Include="SQLiteData.fs" />
|
||||
<Compile Include="PostgreSql\PostgreSqlHelpers.fs" />
|
||||
<Compile Include="PostgreSql\PostgreSqlCategoryData.fs" />
|
||||
<Compile Include="PostgreSql\PostgreSqlPageData.fs" />
|
||||
<Compile Include="PostgreSql\PostgreSqlPostData.fs" />
|
||||
<Compile Include="PostgreSql\PostgreSqlTagMapData.fs" />
|
||||
<Compile Include="PostgreSql\PostgreSqlThemeData.fs" />
|
||||
<Compile Include="PostgreSql\PostgreSqlUploadData.fs" />
|
||||
<Compile Include="PostgreSql\PostgreSqlWebLogData.fs" />
|
||||
<Compile Include="PostgreSql\PostgreSqlWebLogUserData.fs" />
|
||||
<Compile Include="PostgreSqlData.fs" />
|
||||
<Compile Include="Postgres\PostgresHelpers.fs" />
|
||||
<Compile Include="Postgres\PostgresCache.fs" />
|
||||
<Compile Include="Postgres\PostgresCategoryData.fs" />
|
||||
<Compile Include="Postgres\PostgresPageData.fs" />
|
||||
<Compile Include="Postgres\PostgresPostData.fs" />
|
||||
<Compile Include="Postgres\PostgresTagMapData.fs" />
|
||||
<Compile Include="Postgres\PostgresThemeData.fs" />
|
||||
<Compile Include="Postgres\PostgresUploadData.fs" />
|
||||
<Compile Include="Postgres\PostgresWebLogData.fs" />
|
||||
<Compile Include="Postgres\PostgresWebLogUserData.fs" />
|
||||
<Compile Include="PostgresData.fs" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
||||
216
src/MyWebLog.Data/Postgres/PostgresCache.fs
Normal file
216
src/MyWebLog.Data/Postgres/PostgresCache.fs
Normal file
@ -0,0 +1,216 @@
|
||||
namespace MyWebLog.Data.Postgres
|
||||
|
||||
open System.Threading
|
||||
open System.Threading.Tasks
|
||||
open Microsoft.Extensions.Caching.Distributed
|
||||
open NodaTime
|
||||
open Npgsql
|
||||
open Npgsql.FSharp
|
||||
|
||||
/// Helper types and functions for the cache
|
||||
[<AutoOpen>]
|
||||
module private Helpers =
|
||||
|
||||
/// The cache entry
|
||||
type Entry =
|
||||
{ /// The ID of the cache entry
|
||||
Id : string
|
||||
|
||||
/// The value to be cached
|
||||
Payload : byte[]
|
||||
|
||||
/// When this entry will expire
|
||||
ExpireAt : Instant
|
||||
|
||||
/// The duration by which the expiration should be pushed out when being refreshed
|
||||
SlidingExpiration : Duration option
|
||||
|
||||
/// The must-expire-by date/time for the cache entry
|
||||
AbsoluteExpiration : Instant option
|
||||
}
|
||||
|
||||
/// Run a task synchronously
|
||||
let sync<'T> (it : Task<'T>) = it |> (Async.AwaitTask >> Async.RunSynchronously)
|
||||
|
||||
/// Get the current instant
|
||||
let getNow () = SystemClock.Instance.GetCurrentInstant ()
|
||||
|
||||
/// Create a parameter for the expire-at time
|
||||
let expireParam (it : Instant) =
|
||||
"@expireAt", Sql.parameter (NpgsqlParameter ("@expireAt", it))
|
||||
|
||||
/// Create a parameter for a possibly-missing NodaTime type
|
||||
let optParam<'T> name (it : 'T option) =
|
||||
let p = NpgsqlParameter ($"@%s{name}", if Option.isSome it then box it.Value else null)
|
||||
p.ParameterName, Sql.parameter p
|
||||
|
||||
|
||||
/// A distributed cache implementation in PostgreSQL used to handle sessions for myWebLog
|
||||
type DistributedCache (connStr : string) =
|
||||
|
||||
// ~~~ INITIALIZATION ~~~
|
||||
|
||||
do
|
||||
task {
|
||||
let! exists =
|
||||
Sql.connect connStr
|
||||
|> Sql.query $"
|
||||
SELECT EXISTS
|
||||
(SELECT 1 FROM pg_tables WHERE schemaname = 'public' AND tablename = 'session')
|
||||
AS {existsName}"
|
||||
|> Sql.executeRowAsync Map.toExists
|
||||
if not exists then
|
||||
let! _ =
|
||||
Sql.connect connStr
|
||||
|> Sql.query
|
||||
"CREATE TABLE session (
|
||||
id TEXT NOT NULL PRIMARY KEY,
|
||||
payload BYETA NOT NULL,
|
||||
expire_at TIMESTAMPTZ NOT NULL,
|
||||
sliding_expiration INTERVAL,
|
||||
absolute_expiration TIMESTAMPTZ);
|
||||
CREATE INDEX idx_session_expiration ON session (expire_at)"
|
||||
|> Sql.executeNonQueryAsync
|
||||
()
|
||||
} |> sync
|
||||
|
||||
// ~~~ SUPPORT FUNCTIONS ~~~
|
||||
|
||||
/// Get an entry, updating it for sliding expiration
|
||||
let getEntry key = backgroundTask {
|
||||
let idParam = "@id", Sql.string key
|
||||
let! tryEntry =
|
||||
Sql.connect connStr
|
||||
|> Sql.query "SELECT * FROM session WHERE id = @id"
|
||||
|> Sql.parameters [ idParam ]
|
||||
|> Sql.executeAsync (fun row ->
|
||||
{ Id = row.string "id"
|
||||
Payload = row.bytea "payload"
|
||||
ExpireAt = row.fieldValue<Instant> "expire_at"
|
||||
SlidingExpiration = row.fieldValueOrNone<Duration> "sliding_expiration"
|
||||
AbsoluteExpiration = row.fieldValueOrNone<Instant> "absolute_expiration" })
|
||||
|> tryHead
|
||||
match tryEntry with
|
||||
| Some entry ->
|
||||
let now = getNow ()
|
||||
let slideExp = defaultArg entry.SlidingExpiration Duration.MinValue
|
||||
let absExp = defaultArg entry.AbsoluteExpiration Instant.MinValue
|
||||
let needsRefresh, item =
|
||||
if entry.ExpireAt = absExp then false, entry
|
||||
elif slideExp = Duration.MinValue && absExp = Instant.MinValue then false, entry
|
||||
elif absExp > Instant.MinValue && entry.ExpireAt.Plus slideExp > absExp then
|
||||
true, { entry with ExpireAt = absExp }
|
||||
else true, { entry with ExpireAt = now.Plus slideExp }
|
||||
if needsRefresh then
|
||||
let! _ =
|
||||
Sql.connect connStr
|
||||
|> Sql.query "UPDATE session SET expire_at = @expireAt WHERE id = @id"
|
||||
|> Sql.parameters [ expireParam item.ExpireAt; idParam ]
|
||||
|> Sql.executeNonQueryAsync
|
||||
()
|
||||
return if item.ExpireAt > now then Some entry else None
|
||||
| None -> return None
|
||||
}
|
||||
|
||||
/// The last time expired entries were purged (runs every 30 minutes)
|
||||
let mutable lastPurge = Instant.MinValue
|
||||
|
||||
/// Purge expired entries every 30 minutes
|
||||
let purge () = backgroundTask {
|
||||
let now = getNow ()
|
||||
if lastPurge.Plus (Duration.FromMinutes 30L) < now then
|
||||
let! _ =
|
||||
Sql.connect connStr
|
||||
|> Sql.query "DELETE FROM session WHERE expire_at < @expireAt"
|
||||
|> Sql.parameters [ expireParam now ]
|
||||
|> Sql.executeNonQueryAsync
|
||||
lastPurge <- now
|
||||
}
|
||||
|
||||
/// Remove a cache entry
|
||||
let removeEntry key = backgroundTask {
|
||||
let! _ =
|
||||
Sql.connect connStr
|
||||
|> Sql.query "DELETE FROM session WHERE id = @id"
|
||||
|> Sql.parameters [ "@id", Sql.string key ]
|
||||
|> Sql.executeNonQueryAsync
|
||||
()
|
||||
}
|
||||
|
||||
/// Save an entry
|
||||
let saveEntry (opts : DistributedCacheEntryOptions) key payload = backgroundTask {
|
||||
let now = getNow ()
|
||||
let expireAt, slideExp, absExp =
|
||||
if opts.SlidingExpiration.HasValue then
|
||||
let slide = Duration.FromTimeSpan opts.SlidingExpiration.Value
|
||||
now.Plus slide, Some slide, None
|
||||
elif opts.AbsoluteExpiration.HasValue then
|
||||
let exp = Instant.FromDateTimeOffset opts.AbsoluteExpiration.Value
|
||||
exp, None, Some exp
|
||||
elif opts.AbsoluteExpirationRelativeToNow.HasValue then
|
||||
let exp = now.Plus (Duration.FromTimeSpan opts.AbsoluteExpirationRelativeToNow.Value)
|
||||
exp, None, Some exp
|
||||
else
|
||||
// Default to 1 hour sliding expiration
|
||||
let slide = Duration.FromHours 1
|
||||
now.Plus slide, Some slide, None
|
||||
let! _ =
|
||||
Sql.connect connStr
|
||||
|> Sql.query
|
||||
"INSERT INTO session (
|
||||
id, payload, expire_at, sliding_expiration, absolute_expiration
|
||||
) VALUES (
|
||||
@id, @payload, @expireAt, @slideExp, @absExp
|
||||
) ON CONFLICT (id) DO UPDATE
|
||||
SET payload = EXCLUDED.payload,
|
||||
expire_at = EXCLUDED.expire_at,
|
||||
sliding_expiration = EXCLUDED.sliding_expiration,
|
||||
absolute_expiration = EXCLUDED.absolute_expiration"
|
||||
|> Sql.parameters
|
||||
[ "@id", Sql.string key
|
||||
"@payload", Sql.bytea payload
|
||||
expireParam expireAt
|
||||
optParam "slideExp" slideExp
|
||||
optParam "absExp" absExp ]
|
||||
|> Sql.executeNonQueryAsync
|
||||
()
|
||||
}
|
||||
|
||||
// ~~~ IMPLEMENTATION FUNCTIONS ~~~
|
||||
|
||||
/// Retrieve the data for a cache entry
|
||||
let get key (_ : CancellationToken) = backgroundTask {
|
||||
match! getEntry key with
|
||||
| Some entry ->
|
||||
do! purge ()
|
||||
return entry.Payload
|
||||
| None -> return null
|
||||
}
|
||||
|
||||
/// Refresh an entry
|
||||
let refresh key (cancelToken : CancellationToken) = backgroundTask {
|
||||
let! _ = get key cancelToken
|
||||
()
|
||||
}
|
||||
|
||||
/// Remove an entry
|
||||
let remove key (_ : CancellationToken) = backgroundTask {
|
||||
do! removeEntry key
|
||||
do! purge ()
|
||||
}
|
||||
|
||||
/// Set an entry
|
||||
let set key value options (_ : CancellationToken) = backgroundTask {
|
||||
do! saveEntry options key value
|
||||
do! purge ()
|
||||
}
|
||||
|
||||
interface IDistributedCache with
|
||||
member this.Get key = get key CancellationToken.None |> sync
|
||||
member this.GetAsync (key, token) = get key token
|
||||
member this.Refresh key = refresh key CancellationToken.None |> sync
|
||||
member this.RefreshAsync (key, token) = refresh key token
|
||||
member this.Remove key = remove key CancellationToken.None |> sync
|
||||
member this.RemoveAsync (key, token) = remove key token
|
||||
member this.Set (key, value, options) = set key value options CancellationToken.None |> sync
|
||||
member this.SetAsync (key, value, options, token) = set key value options token
|
||||
@ -1,11 +1,12 @@
|
||||
namespace MyWebLog.Data.PostgreSql
|
||||
namespace MyWebLog.Data.Postgres
|
||||
|
||||
open MyWebLog
|
||||
open MyWebLog.Data
|
||||
open Npgsql
|
||||
open Npgsql.FSharp
|
||||
|
||||
type PostgreSqlCategoryData (conn : NpgsqlConnection) =
|
||||
/// PostgreSQL myWebLog category data implementation
|
||||
type PostgresCategoryData (conn : NpgsqlConnection) =
|
||||
|
||||
/// Count all categories for the given web log
|
||||
let countAll webLogId =
|
||||
@ -1,6 +1,6 @@
|
||||
/// Helper functions for the PostgreSQL data implementation
|
||||
[<AutoOpen>]
|
||||
module MyWebLog.Data.PostgreSql.PostgreSqlHelpers
|
||||
module MyWebLog.Data.Postgres.PostgresHelpers
|
||||
|
||||
open System.Threading.Tasks
|
||||
open MyWebLog
|
||||
@ -1,4 +1,4 @@
|
||||
namespace MyWebLog.Data.PostgreSql
|
||||
namespace MyWebLog.Data.Postgres
|
||||
|
||||
open MyWebLog
|
||||
open MyWebLog.Data
|
||||
@ -7,7 +7,7 @@ open Npgsql
|
||||
open Npgsql.FSharp
|
||||
|
||||
/// PostgreSQL myWebLog page data implementation
|
||||
type PostgreSqlPageData (conn : NpgsqlConnection) =
|
||||
type PostgresPageData (conn : NpgsqlConnection) =
|
||||
|
||||
// SUPPORT FUNCTIONS
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
namespace MyWebLog.Data.PostgreSql
|
||||
namespace MyWebLog.Data.Postgres
|
||||
|
||||
open System
|
||||
open MyWebLog
|
||||
@ -8,7 +8,7 @@ open Npgsql
|
||||
open Npgsql.FSharp
|
||||
|
||||
/// PostgreSQL myWebLog post data implementation
|
||||
type PostgreSqlPostData (conn : NpgsqlConnection) =
|
||||
type PostgresPostData (conn : NpgsqlConnection) =
|
||||
|
||||
// SUPPORT FUNCTIONS
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
namespace MyWebLog.Data.PostgreSql
|
||||
namespace MyWebLog.Data.Postgres
|
||||
|
||||
open MyWebLog
|
||||
open MyWebLog.Data
|
||||
@ -6,7 +6,7 @@ open Npgsql
|
||||
open Npgsql.FSharp
|
||||
|
||||
/// PostgreSQL myWebLog tag mapping data implementation
|
||||
type PostgreSqlTagMapData (conn : NpgsqlConnection) =
|
||||
type PostgresTagMapData (conn : NpgsqlConnection) =
|
||||
|
||||
/// Find a tag mapping by its ID for the given web log
|
||||
let findById tagMapId webLogId =
|
||||
@ -1,4 +1,4 @@
|
||||
namespace MyWebLog.Data.PostgreSql
|
||||
namespace MyWebLog.Data.Postgres
|
||||
|
||||
open MyWebLog
|
||||
open MyWebLog.Data
|
||||
@ -6,7 +6,7 @@ open Npgsql
|
||||
open Npgsql.FSharp
|
||||
|
||||
/// PostreSQL myWebLog theme data implementation
|
||||
type PostgreSqlThemeData (conn : NpgsqlConnection) =
|
||||
type PostgresThemeData (conn : NpgsqlConnection) =
|
||||
|
||||
/// Retrieve all themes (except 'admin'; excludes template text)
|
||||
let all () = backgroundTask {
|
||||
@ -135,7 +135,7 @@ type PostgreSqlThemeData (conn : NpgsqlConnection) =
|
||||
|
||||
|
||||
/// PostreSQL myWebLog theme data implementation
|
||||
type PostgreSqlThemeAssetData (conn : NpgsqlConnection) =
|
||||
type PostgresThemeAssetData (conn : NpgsqlConnection) =
|
||||
|
||||
/// Get all theme assets (excludes data)
|
||||
let all () =
|
||||
@ -1,4 +1,4 @@
|
||||
namespace MyWebLog.Data.PostgreSql
|
||||
namespace MyWebLog.Data.Postgres
|
||||
|
||||
open MyWebLog
|
||||
open MyWebLog.Data
|
||||
@ -6,7 +6,7 @@ open Npgsql
|
||||
open Npgsql.FSharp
|
||||
|
||||
/// PostgreSQL myWebLog uploaded file data implementation
|
||||
type PostgreSqlUploadData (conn : NpgsqlConnection) =
|
||||
type PostgresUploadData (conn : NpgsqlConnection) =
|
||||
|
||||
/// The INSERT statement for an uploaded file
|
||||
let upInsert =
|
||||
@ -1,4 +1,4 @@
|
||||
namespace MyWebLog.Data.PostgreSql
|
||||
namespace MyWebLog.Data.Postgres
|
||||
|
||||
open MyWebLog
|
||||
open MyWebLog.Data
|
||||
@ -6,7 +6,7 @@ open Npgsql
|
||||
open Npgsql.FSharp
|
||||
|
||||
/// PostgreSQL myWebLog web log data implementation
|
||||
type PostgreSqlWebLogData (conn : NpgsqlConnection) =
|
||||
type PostgresWebLogData (conn : NpgsqlConnection) =
|
||||
|
||||
// SUPPORT FUNCTIONS
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
namespace MyWebLog.Data.PostgreSql
|
||||
namespace MyWebLog.Data.Postgres
|
||||
|
||||
open MyWebLog
|
||||
open MyWebLog.Data
|
||||
@ -6,7 +6,7 @@ open Npgsql
|
||||
open Npgsql.FSharp
|
||||
|
||||
/// PostgreSQL myWebLog user data implementation
|
||||
type PostgreSqlWebLogUserData (conn : NpgsqlConnection) =
|
||||
type PostgresWebLogUserData (conn : NpgsqlConnection) =
|
||||
|
||||
/// The INSERT statement for a user
|
||||
let userInsert =
|
||||
@ -1,24 +1,24 @@
|
||||
namespace MyWebLog.Data
|
||||
|
||||
open Microsoft.Extensions.Logging
|
||||
open MyWebLog.Data.PostgreSql
|
||||
open MyWebLog.Data.Postgres
|
||||
open Npgsql
|
||||
open Npgsql.FSharp
|
||||
|
||||
/// Data implementation for PostgreSQL
|
||||
type PostgreSqlData (conn : NpgsqlConnection, log : ILogger<PostgreSqlData>) =
|
||||
type PostgresData (conn : NpgsqlConnection, log : ILogger<PostgresData>) =
|
||||
|
||||
interface IData with
|
||||
|
||||
member _.Category = PostgreSqlCategoryData conn
|
||||
member _.Page = PostgreSqlPageData conn
|
||||
member _.Post = PostgreSqlPostData conn
|
||||
member _.TagMap = PostgreSqlTagMapData conn
|
||||
member _.Theme = PostgreSqlThemeData conn
|
||||
member _.ThemeAsset = PostgreSqlThemeAssetData conn
|
||||
member _.Upload = PostgreSqlUploadData conn
|
||||
member _.WebLog = PostgreSqlWebLogData conn
|
||||
member _.WebLogUser = PostgreSqlWebLogUserData conn
|
||||
member _.Category = PostgresCategoryData conn
|
||||
member _.Page = PostgresPageData conn
|
||||
member _.Post = PostgresPostData conn
|
||||
member _.TagMap = PostgresTagMapData conn
|
||||
member _.Theme = PostgresThemeData conn
|
||||
member _.ThemeAsset = PostgresThemeAssetData conn
|
||||
member _.Upload = PostgresUploadData conn
|
||||
member _.WebLog = PostgresWebLogData conn
|
||||
member _.WebLogUser = PostgresWebLogUserData conn
|
||||
|
||||
member _.StartUp () = backgroundTask {
|
||||
|
||||
@ -60,10 +60,10 @@ module DataImplementation =
|
||||
let conn = await (rethinkCfg.CreateConnectionAsync log)
|
||||
upcast RethinkDbData (conn, rethinkCfg, log)
|
||||
elif hasConnStr "PostgreSQL" then
|
||||
let log = sp.GetRequiredService<ILogger<PostgreSqlData>> ()
|
||||
let log = sp.GetRequiredService<ILogger<PostgresData>> ()
|
||||
let conn = new NpgsqlConnection (connStr "PostgreSQL")
|
||||
log.LogInformation $"Using PostgreSQL database {conn.Host}:{conn.Port}/{conn.Database}"
|
||||
PostgreSqlData (conn, log)
|
||||
PostgresData (conn, log)
|
||||
else
|
||||
upcast createSQLite "Data Source=./myweblog.db;Cache=Shared"
|
||||
|
||||
@ -144,13 +144,13 @@ let rec main args =
|
||||
// Use SQLite for caching as well
|
||||
let cachePath = defaultArg (Option.ofObj (cfg.GetConnectionString "SQLiteCachePath")) "./session.db"
|
||||
builder.Services.AddSqliteCache (fun o -> o.CachePath <- cachePath) |> ignore
|
||||
| :? PostgreSqlData ->
|
||||
| :? PostgresData ->
|
||||
// ADO.NET connections are designed to work as per-request instantiation
|
||||
let cfg = sp.GetRequiredService<IConfiguration> ()
|
||||
builder.Services.AddScoped<NpgsqlConnection> (fun sp ->
|
||||
new NpgsqlConnection (cfg.GetConnectionString "PostgreSQL"))
|
||||
|> ignore
|
||||
builder.Services.AddScoped<IData, PostgreSqlData> () |> ignore
|
||||
builder.Services.AddScoped<IData, PostgresData> () |> ignore
|
||||
// Use SQLite for caching (for now)
|
||||
let cachePath = defaultArg (Option.ofObj (cfg.GetConnectionString "SQLiteCachePath")) "./session.db"
|
||||
builder.Services.AddSqliteCache (fun o -> o.CachePath <- cachePath) |> ignore
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user