646fbc7 · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Implement persistent message queue with sequential background flush loop

646fbc7iko6 months ago

client/App.fs+68−23

compare with master

unifiedsplit

@@ -63,7 +63,8 @@ module App =

63 Messages: Map<string, ChatMessage list>

64 Error: string option

65 PollCursor: int64

66− Profile: Profile option }

66+ Profile: Profile option

67+ FlushingOutbox: bool }

68 // ── Msg ──

69 // ── Msg ──

@@ -76,7 +77,6 @@ module App =

76 | AuthErr of string

77 | AuthErr of string

77 | DoDisconnect

78 | DoDisconnect

78 | DoSend

79 | DoSend

79− | SendFailed of string

80 | DoSaveContact

81 | PollResult of events: IncomingEvent list * status: string * cursor: int64

82 | CopyPubKey

@@ -85,12 +85,15 @@ module App =

85 | DoPickPhoto

86 | PhotoPicked of byte[] option

87 | DoSaveProfile

88+ | StartFlush

89+ | FlushResult of sent: bool

89 // ── CmdMsg ──

91 // ── CmdMsg ──

91 type CmdMsg =

93 type CmdMsg =

92 | CmdConnect of url: string * identity: Identity

94 | CmdConnect of url: string * identity: Identity

93− | CmdSend of session: Session * recipientHex: string * envelope: Envelope

95+ | CmdEnqueue of recipientHex: string * envelope: Envelope

96+ | CmdFlushOutbox of session: Session

94 | CmdPoll of session: Session * cursor: int64

97 | CmdPoll of session: Session * cursor: int64

95 | CmdCopyToClipboard of string

98 | CmdCopyToClipboard of string

96 | CmdLoadState

99 | CmdLoadState

@@ -228,7 +231,8 @@ module App =

228 Messages = Map.empty

231 Messages = Map.empty

229 Error = None

232 Error = None

230 PollCursor = 0L

233 PollCursor = 0L

231− Profile = None },

234+ Profile = None

235+ FlushingOutbox = false },

232 [ CmdLoadState ]

236 [ CmdLoadState ]

233

237

234 // ── Update ──

238 // ── Update ──

@@ -267,23 +271,27 @@ module App =

267 Auth = Identified(id, Online token)

271 Auth = Identified(id, Online token)

268 Page = Conversations

272 Page = Conversations

269 Error = None

273 Error = None

270− PollStatus = "polling..." },

274+ PollStatus = "polling..."

275+ FlushingOutbox = true },

271− [ CmdPoll(session, model.PollCursor) ]

276+ [ CmdPoll(session, model.PollCursor)

277+ CmdFlushOutbox session ]

272 | _ -> model, []

278 | _ -> model, []

273

279

274 | AuthErr err ->

280 | AuthErr err ->

275 { setConn Offline model with

281 { setConn Offline model with

276− Error = Some err },

282+ Error = Some err

283+ FlushingOutbox = false },

277 []

284 []

278

285

279 | DoDisconnect ->

286 | DoDisconnect ->

280 { setConn Offline model with

287 { setConn Offline model with

281− PollStatus = "" },

288+ PollStatus = ""

289+ FlushingOutbox = false },

282 []

290 []

283

291

284 | DoSend ->

292 | DoSend ->

285− match model.Page, trySession model with

293+ match model.Page with

286− | Chat(pk, compose), Some session when compose <> "" ->

294+ | Chat(pk, compose) when compose <> "" ->

287 let id = Guid.NewGuid().ToString()

295 let id = Guid.NewGuid().ToString()

288

296

289 let outMsg =

297 let outMsg =

@@ -300,14 +308,30 @@ module App =

300 Messages = model.Messages |> appendMessage pk outMsg }

308 Messages = model.Messages |> appendMessage pk outMsg }

301

309

302 model',

310 model',

303− [ CmdSend(session, pk, Envelope.TextMessage(id, compose))

311+ [ CmdEnqueue(pk, Envelope.TextMessage(id, compose))

304− saveCmdMsg model'

305 match model.Profile with

312 match model.Profile with

306− | Some profile when isFirstInteraction -> CmdSend(session, pk, Envelope.ProfileMessage profile)

313+ | Some profile when isFirstInteraction -> CmdEnqueue(pk, Envelope.ProfileMessage profile)

307− | _ -> () ]

314+ | _ -> ()

315+ saveCmdMsg model' ]

308 | _ -> model, []

316 | _ -> model, []

309

317

310− | SendFailed err -> { model with Error = Some err }, []

318+ | StartFlush ->

319+ if model.FlushingOutbox then

320+ model, []

321+ else

322+ match trySession model with

323+ | Some session ->

324+ { model with FlushingOutbox = true },

325+ [ CmdFlushOutbox session ]

326+ | None -> model, []

327+

328+ | FlushResult sent ->

329+ if sent then

330+ match trySession model with

331+ | Some session -> model, [ CmdFlushOutbox session ]

332+ | None -> { model with FlushingOutbox = false }, []

333+ else

334+ { model with FlushingOutbox = false }, []

311

335

312 | PollResult(events, status, newCursor) ->

336 | PollResult(events, status, newCursor) ->

313 let model' =

337 let model' =

@@ -379,11 +403,8 @@ module App =

379 Profile = Some profile

403 Profile = Some profile

380 Page = Settings },

404 Page = Settings },

381 [ CmdSaveProfile profile

405 [ CmdSaveProfile profile

382− match trySession model with

383− | Some session ->

384− for pk in model.Contacts.Keys do

406+ for pk in model.Contacts.Keys do

385− CmdSend(session, pk, Envelope.ProfileMessage profile)

407+ CmdEnqueue(pk, Envelope.ProfileMessage profile) ]

386− | None -> () ]

387 | _ -> model, []

408 | _ -> model, []

388

409

389 | StateLoaded(Some identity, Some data, profile) ->

410 | StateLoaded(Some identity, Some data, profile) ->

@@ -591,9 +612,33 @@ module App =

591 Cmd.OfAsync.either (fun (url, id) -> authenticate url id) (url, identity) AuthOk (fun ex ->

612 Cmd.OfAsync.either (fun (url, id) -> authenticate url id) (url, identity) AuthOk (fun ex ->

592 AuthErr ex.Message)

613 AuthErr ex.Message)

593

614

594− | CmdSend(session, recipientHex, envelope) ->

615+ | CmdEnqueue(recipientHex, envelope) ->

595− Cmd.OfAsync.attempt (fun (s, r, e) -> sendEnvelope s r e) (session, recipientHex, envelope) (fun ex ->

616+ Cmd.ofEffect (fun dispatch ->

596− SendFailed ex.Message)

617+ let json = serializeEnvelope envelope

618+ Store.enqueueOutbox recipientHex json

619+ dispatch StartFlush)

620+

621+ | CmdFlushOutbox session ->

622+ Cmd.ofEffect (fun dispatch ->

623+ async {

624+ match Store.peekOutbox () with

625+ | Some item ->

626+ try

627+ let blob =

628+ Crypto.encrypt session.Identity.PrivKey (Crypto.fromHex item.RecipientHex) item.EnvelopeJson

629+

630+ let ts = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()

631+ do! sendMessage session.Url session.Token item.RecipientHex (Crypto.toHex blob) ts

632+ Store.dequeueOutbox item.Id

633+ dispatch (FlushResult true)

634+ with ex ->

635+ log $"outbox send error: {ex.Message}"

636+ do! Async.Sleep 3000

637+ dispatch (FlushResult true)

638+ | None ->

639+ dispatch (FlushResult false)

640+ }

641+ |> Async.Start)

597

642

598 | CmdPoll(session, cursor) ->

643 | CmdPoll(session, cursor) ->

599 Cmd.ofEffect (fun dispatch ->

644 Cmd.ofEffect (fun dispatch ->

client/Store.fs+25

@@ -72,6 +72,13 @@ module Store =

72 ServerUrl: string

73 PollCursor: int64 }

75+ [<CLIMutable>]

76+ type OutboxRow =

77+ { Id: int64

78+ RecipientHex: string

79+ EnvelopeJson: string

80+ CreatedAt: int64 }

81+

75 let private profileTable = table'<ProfileRow> "profile"

82 let private profileTable = table'<ProfileRow> "profile"

76 let private contactTable = table'<ContactRow> "contacts"

83 let private contactTable = table'<ContactRow> "contacts"

77 let private messageTable = table'<MessageRow> "messages"

84 let private messageTable = table'<MessageRow> "messages"

@@ -116,6 +123,7 @@ module Store =

116 conn.Execute("CREATE TABLE IF NOT EXISTS contacts (Pubkey TEXT PRIMARY KEY, Nickname TEXT NOT NULL DEFAULT '', DisplayName TEXT NOT NULL DEFAULT '', Bio TEXT NOT NULL DEFAULT '', Photo BLOB)") |> ignore

123 conn.Execute("CREATE TABLE IF NOT EXISTS contacts (Pubkey TEXT PRIMARY KEY, Nickname TEXT NOT NULL DEFAULT '', DisplayName TEXT NOT NULL DEFAULT '', Bio TEXT NOT NULL DEFAULT '', Photo BLOB)") |> ignore

117 conn.Execute("CREATE TABLE IF NOT EXISTS messages (Id TEXT PRIMARY KEY, ConversationId TEXT NOT NULL, Body TEXT NOT NULL DEFAULT '', TimestampUnix INTEGER NOT NULL, IsOutgoing INTEGER NOT NULL DEFAULT 0, Status INTEGER NOT NULL DEFAULT 0)") |> ignore

124 conn.Execute("CREATE TABLE IF NOT EXISTS messages (Id TEXT PRIMARY KEY, ConversationId TEXT NOT NULL, Body TEXT NOT NULL DEFAULT '', TimestampUnix INTEGER NOT NULL, IsOutgoing INTEGER NOT NULL DEFAULT 0, Status INTEGER NOT NULL DEFAULT 0)") |> ignore

118 conn.Execute("CREATE TABLE IF NOT EXISTS settings (Id TEXT PRIMARY KEY, ServerUrl TEXT NOT NULL DEFAULT '', PollCursor INTEGER NOT NULL DEFAULT 0)") |> ignore

125 conn.Execute("CREATE TABLE IF NOT EXISTS settings (Id TEXT PRIMARY KEY, ServerUrl TEXT NOT NULL DEFAULT '', PollCursor INTEGER NOT NULL DEFAULT 0)") |> ignore

126+ conn.Execute("CREATE TABLE IF NOT EXISTS outbox (Id INTEGER PRIMARY KEY AUTOINCREMENT, RecipientHex TEXT NOT NULL, EnvelopeJson TEXT NOT NULL, CreatedAt INTEGER NOT NULL)") |> ignore

119

127

120 let private identityPath = Path.Combine(appDataDir, "identity.key")

128 let private identityPath = Path.Combine(appDataDir, "identity.key")

121

129

@@ -254,3 +262,20 @@ module Store =

254 |> ignore

262 |> ignore

255 with _ ->

263 with _ ->

256 ()

264 ()

265+

266+ let enqueueOutbox (recipientHex: string) (envelopeJson: string) =

267+ use conn = openConn ()

268+ conn.Execute(

269+ "INSERT INTO outbox (RecipientHex, EnvelopeJson, CreatedAt) VALUES (@RecipientHex, @EnvelopeJson, @CreatedAt)",

270+ {| RecipientHex = recipientHex; EnvelopeJson = envelopeJson

271+ CreatedAt = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() |})

272+ |> ignore

273+

274+ let peekOutbox () : OutboxRow option =

275+ use conn = openConn ()

276+ conn.Query<OutboxRow>("SELECT Id, RecipientHex, EnvelopeJson, CreatedAt FROM outbox ORDER BY Id LIMIT 1")

277+ |> Seq.tryHead

278+

279+ let dequeueOutbox (id: int64) =

280+ use conn = openConn ()

281+ conn.Execute("DELETE FROM outbox WHERE Id = @Id", {| Id = id |}) |> ignore