646fbc7 · skrepka · git.iko.soy

git.iko.soy/skrepka

Implement persistent message queue with sequential background flush loop

commit646fbc74f6f305c2eb59031700d5ab00686dffd4 authoriko <mail@iko.soy> dateTue 14 Apr 2026, 05:20 (6 months ago) parent2216c16

2 files changed+93−23

compare with master

unifiedsplit

client/App.fs+68−23 client/Store.fs+25

client/App.fs+68−23

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

6363 Messages: Map<string, ChatMessage list>

6464 Error: string option

6565 PollCursor: int64

66− Profile: Profile option }

66+ Profile: Profile option

67+ FlushingOutbox: bool }

6768

6869 // ── Msg ──

6970

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

7677 | AuthErr of string

7778 | DoDisconnect

7879 | DoSend

79− | SendFailed of string

8080 | DoSaveContact

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

8282 | CopyPubKey

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

8585 | DoPickPhoto

8686 | PhotoPicked of byte[] option

8787 | DoSaveProfile

88+ | StartFlush

89+ | FlushResult of sent: bool

8890

8991 // ── CmdMsg ──

9092

9193 type CmdMsg =

9294 | 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

9497 | CmdPoll of session: Session * cursor: int64

9598 | CmdCopyToClipboard of string

9699 | CmdLoadState

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

228231 Messages = Map.empty

229232 Error = None

230233 PollCursor = 0L

231− Profile = None },

234+ Profile = None

235+ FlushingOutbox = false },

232236 [ CmdLoadState ]

233237

234238 // ── Update ──

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

267271 Auth = Identified(id, Online token)

268272 Page = Conversations

269273 Error = None

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

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

274+ PollStatus = "polling..."

275+ FlushingOutbox = true },

276+ [ CmdPoll(session, model.PollCursor)

277+ CmdFlushOutbox session ]

272278 | _ -> model, []

273279

274280 | AuthErr err ->

275281 { setConn Offline model with

276− Error = Some err },

282+ Error = Some err

283+ FlushingOutbox = false },

277284 []

278285

279286 | DoDisconnect ->

280287 { setConn Offline model with

281− PollStatus = "" },

288+ PollStatus = ""

289+ FlushingOutbox = false },

282290 []

283291

284292 | DoSend ->

285− match model.Page, trySession model with

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

293+ match model.Page with

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

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

288296

289297 let outMsg =

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

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

301309

302310 model',

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

304− saveCmdMsg model'

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

305312 match model.Profile with

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

307− | _ -> () ]

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

314+ | _ -> ()

315+ saveCmdMsg model' ]

308316 | _ -> model, []

309317

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 }, []

311335

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

313337 let model' =

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

379403 Profile = Some profile

380404 Page = Settings },

381405 [ CmdSaveProfile profile

382− match trySession model with

383− | Some session ->

384− for pk in model.Contacts.Keys do

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

386− | None -> () ]

406+ for pk in model.Contacts.Keys do

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

387408 | _ -> model, []

388409

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

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

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

592613 AuthErr ex.Message)

593614

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

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

596− SendFailed ex.Message)

615+ | CmdEnqueue(recipientHex, envelope) ->

616+ Cmd.ofEffect (fun dispatch ->

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)

597642

598643 | CmdPoll(session, cursor) ->

599644 Cmd.ofEffect (fun dispatch ->

client/Store.fs+25

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

7272 ServerUrl: string

7373 PollCursor: int64 }

7474

75+ [<CLIMutable>]

76+ type OutboxRow =

77+ { Id: int64

78+ RecipientHex: string

79+ EnvelopeJson: string

80+ CreatedAt: int64 }

81+

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

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

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

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

116123 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

117124 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

118125 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

119127

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

121129

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

254262 |> ignore

255263 with _ ->

256264 ()

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