V2 #36
|
@ -2,6 +2,7 @@ namespace MyWebLog.Data.Postgres
|
||||||
|
|
||||||
open System.Threading
|
open System.Threading
|
||||||
open System.Threading.Tasks
|
open System.Threading.Tasks
|
||||||
|
open BitBadger.Npgsql.FSharp.Documents
|
||||||
open Microsoft.Extensions.Caching.Distributed
|
open Microsoft.Extensions.Caching.Distributed
|
||||||
open NodaTime
|
open NodaTime
|
||||||
open Npgsql.FSharp
|
open Npgsql.FSharp
|
||||||
|
@ -39,35 +40,30 @@ module private Helpers =
|
||||||
typedParam "expireAt"
|
typedParam "expireAt"
|
||||||
|
|
||||||
|
|
||||||
open Npgsql
|
|
||||||
|
|
||||||
/// A distributed cache implementation in PostgreSQL used to handle sessions for myWebLog
|
/// A distributed cache implementation in PostgreSQL used to handle sessions for myWebLog
|
||||||
type DistributedCache (dataSource : NpgsqlDataSource) =
|
type DistributedCache () =
|
||||||
|
|
||||||
// ~~~ INITIALIZATION ~~~
|
// ~~~ INITIALIZATION ~~~
|
||||||
|
|
||||||
do
|
do
|
||||||
task {
|
task {
|
||||||
let! exists =
|
let! exists =
|
||||||
Sql.fromDataSource dataSource
|
Configuration.dataSource ()
|
||||||
|
|> Sql.fromDataSource
|
||||||
|> Sql.query $"
|
|> Sql.query $"
|
||||||
SELECT EXISTS
|
SELECT EXISTS
|
||||||
(SELECT 1 FROM pg_tables WHERE schemaname = 'public' AND tablename = 'session')
|
(SELECT 1 FROM pg_tables WHERE schemaname = 'public' AND tablename = 'session')
|
||||||
AS {existsName}"
|
AS {existsName}"
|
||||||
|> Sql.executeRowAsync Map.toExists
|
|> Sql.executeRowAsync Map.toExists
|
||||||
if not exists then
|
if not exists then
|
||||||
let! _ =
|
do! Custom.nonQuery
|
||||||
Sql.fromDataSource dataSource
|
|
||||||
|> Sql.query
|
|
||||||
"CREATE TABLE session (
|
"CREATE TABLE session (
|
||||||
id TEXT NOT NULL PRIMARY KEY,
|
id TEXT NOT NULL PRIMARY KEY,
|
||||||
payload BYTEA NOT NULL,
|
payload BYTEA NOT NULL,
|
||||||
expire_at TIMESTAMPTZ NOT NULL,
|
expire_at TIMESTAMPTZ NOT NULL,
|
||||||
sliding_expiration INTERVAL,
|
sliding_expiration INTERVAL,
|
||||||
absolute_expiration TIMESTAMPTZ);
|
absolute_expiration TIMESTAMPTZ);
|
||||||
CREATE INDEX idx_session_expiration ON session (expire_at)"
|
CREATE INDEX idx_session_expiration ON session (expire_at)" []
|
||||||
|> Sql.executeNonQueryAsync
|
|
||||||
()
|
|
||||||
} |> sync
|
} |> sync
|
||||||
|
|
||||||
// ~~~ SUPPORT FUNCTIONS ~~~
|
// ~~~ SUPPORT FUNCTIONS ~~~
|
||||||
|
@ -76,16 +72,13 @@ type DistributedCache (dataSource : NpgsqlDataSource) =
|
||||||
let getEntry key = backgroundTask {
|
let getEntry key = backgroundTask {
|
||||||
let idParam = "@id", Sql.string key
|
let idParam = "@id", Sql.string key
|
||||||
let! tryEntry =
|
let! tryEntry =
|
||||||
Sql.fromDataSource dataSource
|
Custom.single "SELECT * FROM session WHERE id = @id" [ idParam ]
|
||||||
|> Sql.query "SELECT * FROM session WHERE id = @id"
|
(fun row ->
|
||||||
|> Sql.parameters [ idParam ]
|
{ Id = row.string "id"
|
||||||
|> Sql.executeAsync (fun row ->
|
Payload = row.bytea "payload"
|
||||||
{ Id = row.string "id"
|
ExpireAt = row.fieldValue<Instant> "expire_at"
|
||||||
Payload = row.bytea "payload"
|
SlidingExpiration = row.fieldValueOrNone<Duration> "sliding_expiration"
|
||||||
ExpireAt = row.fieldValue<Instant> "expire_at"
|
AbsoluteExpiration = row.fieldValueOrNone<Instant> "absolute_expiration" })
|
||||||
SlidingExpiration = row.fieldValueOrNone<Duration> "sliding_expiration"
|
|
||||||
AbsoluteExpiration = row.fieldValueOrNone<Instant> "absolute_expiration" })
|
|
||||||
|> tryHead
|
|
||||||
match tryEntry with
|
match tryEntry with
|
||||||
| Some entry ->
|
| Some entry ->
|
||||||
let now = getNow ()
|
let now = getNow ()
|
||||||
|
@ -98,11 +91,8 @@ type DistributedCache (dataSource : NpgsqlDataSource) =
|
||||||
true, { entry with ExpireAt = absExp }
|
true, { entry with ExpireAt = absExp }
|
||||||
else true, { entry with ExpireAt = now.Plus slideExp }
|
else true, { entry with ExpireAt = now.Plus slideExp }
|
||||||
if needsRefresh then
|
if needsRefresh then
|
||||||
let! _ =
|
do! Custom.nonQuery "UPDATE session SET expire_at = @expireAt WHERE id = @id"
|
||||||
Sql.fromDataSource dataSource
|
[ expireParam item.ExpireAt; idParam ]
|
||||||
|> 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
|
return if item.ExpireAt > now then Some entry else None
|
||||||
| None -> return None
|
| None -> return None
|
||||||
|
@ -115,26 +105,16 @@ type DistributedCache (dataSource : NpgsqlDataSource) =
|
||||||
let purge () = backgroundTask {
|
let purge () = backgroundTask {
|
||||||
let now = getNow ()
|
let now = getNow ()
|
||||||
if lastPurge.Plus (Duration.FromMinutes 30L) < now then
|
if lastPurge.Plus (Duration.FromMinutes 30L) < now then
|
||||||
let! _ =
|
do! Custom.nonQuery "DELETE FROM session WHERE expire_at < @expireAt" [ expireParam now ]
|
||||||
Sql.fromDataSource dataSource
|
|
||||||
|> Sql.query "DELETE FROM session WHERE expire_at < @expireAt"
|
|
||||||
|> Sql.parameters [ expireParam now ]
|
|
||||||
|> Sql.executeNonQueryAsync
|
|
||||||
lastPurge <- now
|
lastPurge <- now
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Remove a cache entry
|
/// Remove a cache entry
|
||||||
let removeEntry key = backgroundTask {
|
let removeEntry key =
|
||||||
let! _ =
|
Delete.byId "session" key
|
||||||
Sql.fromDataSource dataSource
|
|
||||||
|> Sql.query "DELETE FROM session WHERE id = @id"
|
|
||||||
|> Sql.parameters [ "@id", Sql.string key ]
|
|
||||||
|> Sql.executeNonQueryAsync
|
|
||||||
()
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Save an entry
|
/// Save an entry
|
||||||
let saveEntry (opts : DistributedCacheEntryOptions) key payload = backgroundTask {
|
let saveEntry (opts : DistributedCacheEntryOptions) key payload =
|
||||||
let now = getNow ()
|
let now = getNow ()
|
||||||
let expireAt, slideExp, absExp =
|
let expireAt, slideExp, absExp =
|
||||||
if opts.SlidingExpiration.HasValue then
|
if opts.SlidingExpiration.HasValue then
|
||||||
|
@ -150,27 +130,21 @@ type DistributedCache (dataSource : NpgsqlDataSource) =
|
||||||
// Default to 1 hour sliding expiration
|
// Default to 1 hour sliding expiration
|
||||||
let slide = Duration.FromHours 1
|
let slide = Duration.FromHours 1
|
||||||
now.Plus slide, Some slide, None
|
now.Plus slide, Some slide, None
|
||||||
let! _ =
|
Custom.nonQuery
|
||||||
Sql.fromDataSource dataSource
|
"INSERT INTO session (
|
||||||
|> Sql.query
|
id, payload, expire_at, sliding_expiration, absolute_expiration
|
||||||
"INSERT INTO session (
|
) VALUES (
|
||||||
id, payload, expire_at, sliding_expiration, absolute_expiration
|
@id, @payload, @expireAt, @slideExp, @absExp
|
||||||
) VALUES (
|
) ON CONFLICT (id) DO UPDATE
|
||||||
@id, @payload, @expireAt, @slideExp, @absExp
|
SET payload = EXCLUDED.payload,
|
||||||
) ON CONFLICT (id) DO UPDATE
|
expire_at = EXCLUDED.expire_at,
|
||||||
SET payload = EXCLUDED.payload,
|
sliding_expiration = EXCLUDED.sliding_expiration,
|
||||||
expire_at = EXCLUDED.expire_at,
|
absolute_expiration = EXCLUDED.absolute_expiration"
|
||||||
sliding_expiration = EXCLUDED.sliding_expiration,
|
[ "@id", Sql.string key
|
||||||
absolute_expiration = EXCLUDED.absolute_expiration"
|
"@payload", Sql.bytea payload
|
||||||
|> Sql.parameters
|
expireParam expireAt
|
||||||
[ "@id", Sql.string key
|
optParam "slideExp" slideExp
|
||||||
"@payload", Sql.bytea payload
|
optParam "absExp" absExp ]
|
||||||
expireParam expireAt
|
|
||||||
optParam "slideExp" slideExp
|
|
||||||
optParam "absExp" absExp ]
|
|
||||||
|> Sql.executeNonQueryAsync
|
|
||||||
()
|
|
||||||
}
|
|
||||||
|
|
||||||
// ~~~ IMPLEMENTATION FUNCTIONS ~~~
|
// ~~~ IMPLEMENTATION FUNCTIONS ~~~
|
||||||
|
|
||||||
|
|
|
@ -1,8 +1,8 @@
|
||||||
namespace MyWebLog.Data
|
namespace MyWebLog.Data
|
||||||
|
|
||||||
open Microsoft.Extensions.Logging
|
|
||||||
open BitBadger.Npgsql.Documents
|
open BitBadger.Npgsql.Documents
|
||||||
open BitBadger.Npgsql.FSharp.Documents
|
open BitBadger.Npgsql.FSharp.Documents
|
||||||
|
open Microsoft.Extensions.Logging
|
||||||
open MyWebLog
|
open MyWebLog
|
||||||
open MyWebLog.Data.Postgres
|
open MyWebLog.Data.Postgres
|
||||||
open Newtonsoft.Json
|
open Newtonsoft.Json
|
||||||
|
@ -139,6 +139,23 @@ type PostgresData (source : NpgsqlDataSource, log : ILogger<PostgresData>, ser :
|
||||||
let migrate version = backgroundTask {
|
let migrate version = backgroundTask {
|
||||||
match version with
|
match version with
|
||||||
| Some "v2-rc2" -> ()
|
| Some "v2-rc2" -> ()
|
||||||
|
| Some "v2" ->
|
||||||
|
printfn "** MANUAL DATABASE UPGRADE REQUIRED **\n"
|
||||||
|
printfn "The data structure for PostgreSQL changed significantly between v2-rc2 and v2."
|
||||||
|
printfn "To migrate your data:"
|
||||||
|
printfn " - Using a v2-rc2 executable, back up each web log"
|
||||||
|
printfn " - Drop all tables from the database"
|
||||||
|
printfn " - Using this executable, restore each backup"
|
||||||
|
|
||||||
|
let! webLogs =
|
||||||
|
Configuration.dataSource ()
|
||||||
|
|> Sql.fromDataSource
|
||||||
|
|> Sql.query $"SELECT url_base FROM {Table.WebLog}"
|
||||||
|
|> Sql.executeAsync (fun row -> row.string "url_base")
|
||||||
|
|
||||||
|
printfn "\nCommands to back up all web logs:"
|
||||||
|
webLogs |> List.iter (printfn "myWebLog backup %s")
|
||||||
|
exit 1
|
||||||
// Future versions will be inserted here
|
// Future versions will be inserted here
|
||||||
| Some _
|
| Some _
|
||||||
| None ->
|
| None ->
|
||||||
|
|
|
@ -10,7 +10,7 @@ type WebLogMiddleware (next : RequestDelegate, log : ILogger<WebLogMiddleware>)
|
||||||
/// Is the debug level enabled on the logger?
|
/// Is the debug level enabled on the logger?
|
||||||
let isDebug = log.IsEnabled LogLevel.Debug
|
let isDebug = log.IsEnabled LogLevel.Debug
|
||||||
|
|
||||||
member this.InvokeAsync (ctx : HttpContext) = task {
|
member _.InvokeAsync (ctx : HttpContext) = task {
|
||||||
/// Create the full path of the request
|
/// Create the full path of the request
|
||||||
let path = $"{ctx.Request.Scheme}://{ctx.Request.Host.Value}{ctx.Request.Path.Value}"
|
let path = $"{ctx.Request.Scheme}://{ctx.Request.Host.Value}{ctx.Request.Path.Value}"
|
||||||
match WebLogCache.tryGet path with
|
match WebLogCache.tryGet path with
|
||||||
|
@ -165,8 +165,8 @@ let rec main args =
|
||||||
DataImplementation.createNpgsqlDataSource (sp.GetRequiredService<IConfiguration> ()))
|
DataImplementation.createNpgsqlDataSource (sp.GetRequiredService<IConfiguration> ()))
|
||||||
let _ = builder.Services.AddSingleton<IData> postgres
|
let _ = builder.Services.AddSingleton<IData> postgres
|
||||||
let _ =
|
let _ =
|
||||||
builder.Services.AddSingleton<IDistributedCache> (fun sp ->
|
builder.Services.AddSingleton<IDistributedCache> (fun _ ->
|
||||||
Postgres.DistributedCache (sp.GetRequiredService<NpgsqlDataSource> ()) :> IDistributedCache)
|
Postgres.DistributedCache () :> IDistributedCache)
|
||||||
()
|
()
|
||||||
| _ -> ()
|
| _ -> ()
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue
Block a user